This is an automated email from the ASF dual-hosted git repository. sjanc pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git
commit 0ab6e91128285d5b235a09433647e0f0cb02e74b Author: Szymon Janc <[email protected]> AuthorDate: Thu Oct 22 00:26:24 2026 +0200 nmxact/omp: Fix deadlock on duplicate OMP response or error The per-listener goroutine forwarded decoded responses and errors to the 1-deep NMP listener channels with blocking sends. A second response for the same sequence number (e.g. a CoAP retransmission), or a second error such as repeated ErrorAll() calls, blocked the goroutine forever: - it stopped draining the CoAP listener, so the next message blocked nmcoap.Dispatcher.Dispatch() with the CoAP dispatcher lock held, stalling the whole receive path; - it no longer watched stopCh, so Dispatcher.Stop() waited forever on the WaitGroup while holding the OMP dispatcher lock. The client consumes a single response or error per sequence number, so forward with non-blocking sends and drop the extras. Assisted-by: Claude:claude-opus-5-5 --- nmxact/omp/dispatch.go | 22 ++++++++++++--- nmxact/omp/dispatch_test.go | 66 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 3 deletions(-) diff --git a/nmxact/omp/dispatch.go b/nmxact/omp/dispatch.go index 9041bd2..1bfc6c0 100644 --- a/nmxact/omp/dispatch.go +++ b/nmxact/omp/dispatch.go @@ -93,6 +93,22 @@ func (d *Dispatcher) addOmpListener(seq uint8) (*Listener, error) { defer d.wg.Done() defer d.RemoveCoapListener(mc) + // The client expects a single response or error. Never block + // forwarding extras (e.g. duplicate responses); this goroutine must + // keep draining the CoAP listener and remain responsive to stopCh. + sendRsp := func(rsp nmp.NmpRsp) { + select { + case ompl.nmpl.RspChan <- rsp: + default: + } + } + sendErr := func(err error) { + select { + case ompl.nmpl.ErrChan <- err: + default: + } + } + // Listen for events. All feedback is sent to the client via the NMP // listener channels. It is done this way so that client code can be // OMP/NMP agnostic. @@ -101,16 +117,16 @@ func (d *Dispatcher) addOmpListener(seq uint8) (*Listener, error) { case m := <-ompl.coapl.RspChan: rsp, err := DecodeOmp(m, d.rxFilter) if err != nil { - ompl.nmpl.ErrChan <- err + sendErr(err) } else if rsp != nil { - ompl.nmpl.RspChan <- rsp + sendRsp(rsp) } else { /* no error, no response */ } case err := <-ompl.coapl.ErrChan: if err != nil { - ompl.nmpl.ErrChan <- err + sendErr(err) } case <-ompl.stopCh: diff --git a/nmxact/omp/dispatch_test.go b/nmxact/omp/dispatch_test.go index 97e7e75..956974f 100644 --- a/nmxact/omp/dispatch_test.go +++ b/nmxact/omp/dispatch_test.go @@ -390,3 +390,69 @@ func TestDispatcherStopRemovesCoapListeners(t *testing.T) { } } } + +// A device may send more than one response with the same sequence number +// (e.g. a CoAP retransmission). The extra responses must not block the RX +// path or the listener goroutine, otherwise Stop() never returns. +func TestDispatcherDuplicateRspDoesNotBlock(t *testing.T) { + d, err := NewDispatcher(nil, false, 0) + if err != nil { + t.Fatal(err) + } + + nl, err := d.AddNmpListener(50) + if err != nil { + t.Fatal(err) + } + + b := encodeTestOmpRsp(t, coap.Content, 50, "dup") + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 5; i++ { + d.Dispatch(b) + } + }() + + select { + case <-done: + case <-time.After(testTimeout): + t.Fatalf("Dispatch of duplicate response blocked") + } + + if echo := recvRsp(t, nl).(*nmp.EchoRsp); echo.Payload != "dup" { + t.Errorf("payload = %q", echo.Payload) + } + + d.RemoveNmpListener(50) + stopWithTimeout(t, d) +} + +// Repeated errors must not block the listener goroutine either. +func TestDispatcherRepeatedErrorDoesNotBlock(t *testing.T) { + d, err := NewDispatcher(nil, false, 0) + if err != nil { + t.Fatal(err) + } + + if _, err := d.AddNmpListener(51); err != nil { + t.Fatal(err) + } + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 5; i++ { + d.ErrorAll(errors.New("err")) + } + }() + + select { + case <-done: + case <-time.After(testTimeout): + t.Fatalf("ErrorAll blocked") + } + + stopWithTimeout(t, d) +}
