(mynewt-newtmgr) 04/20: nmxact/nmcoap: Fix loss of back-to-back CoAP-TCP messages

[ja...@apache.org]

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 d4a9509200d4709968c00eb1b261ddfc77510293
Author: Szymon Janc <sz...@codecoup.pl>
AuthorDate: Thu Oct 22 00:24:39 2026 +0200

    nmxact/nmcoap: Fix loss of back-to-back CoAP-TCP messages
    
    Reassembler.RxFrag stored the unread remainder returned by
    coap.PullTcp but then unconditionally cleared it after extracting a
    message. If a received chunk contained the end of one message and the
    start (or all) of the next one, the second message was silently
    dropped, and with it the response a caller was waiting for, which then
    timed out.
    
    Keep the remainder after a successful pull. Since RxFrag returns only
    one message per call, also add Receiver.RxBuffered and make
    Dispatcher.Dispatch keep dispatching until no complete message is left
    in the buffer, so a second complete message is delivered right away
    instead of waiting for the next fragment. Only CoAP-TCP (BLE) buffers
    data; UDP behaviour is unchanged.
    
    Assisted-by: Claude:claude-opus-5-5
---
 nmxact/nmcoap/dispatch.go      | 16 ++++++++---
 nmxact/nmcoap/dispatch_test.go | 65 ++++++++++++++++++++++++++++++++++++++++++
 nmxact/nmcoap/frag.go          |  5 +++-
 nmxact/nmcoap/frag_test.go     | 50 ++++++++++++++++++++++++++++++++
 nmxact/nmcoap/receiver.go      | 15 ++++++++++
 5 files changed, 146 insertions(+), 5 deletions(-)

diff --git a/nmxact/nmcoap/dispatch.go b/nmxact/nmcoap/dispatch.go
index 4048cd9..c8a3e71 100644
--- a/nmxact/nmcoap/dispatch.go
+++ b/nmxact/nmcoap/dispatch.go
@@ -104,12 +104,20 @@ func (d *Dispatcher) RemoveListener(mc MsgCriteria) *Listener {
 
 // Returns true if the response was dispatched.
 func (d *Dispatcher) Dispatch(data []byte) bool {
-	// See if this fragment completes a packet.
-	msg := d.rxer.Rx(data)
-	if msg == nil {
-		return false
+	dispatched := false
+
+	// See if this fragment completes a packet.  With TCP, a fragment may
+	// complete more than one packet; dispatch all of them.
+	for msg := d.rxer.Rx(data); msg != nil; msg = d.rxer.RxBuffered() {
+		if d.dispatchMsg(msg) {
+			dispatched = true
+		}
 	}
 
+	return dispatched
+}
+
+func (d *Dispatcher) dispatchMsg(msg coap.Message) bool {
 	d.mtx.Lock()
 	defer d.mtx.Unlock()
 
diff --git a/nmxact/nmcoap/dispatch_test.go b/nmxact/nmcoap/dispatch_test.go
new file mode 100644
index 0000000..a8df620
--- /dev/null
+++ b/nmxact/nmcoap/dispatch_test.go
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package nmcoap
+
+import (
+	"testing"
+)
+
+// A single TCP chunk containing two complete messages must dispatch both
+// of them immediately, not just the first one.
+func TestDispatcherTcpBackToBack(t *testing.T) {
+	d := NewDispatcher(true, 0)
+
+	l1, err := d.AddListener(MsgCriteria{Token: []byte{1}})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	l2, err := d.AddListener(MsgCriteria{Token: []byte{2}})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	defer d.RemoveListener(l1.Criteria)
+	defer d.RemoveListener(l2.Criteria)
+
+	b1 := tcpMsg(t, []byte{1}, "", []byte("one"))
+	b2 := tcpMsg(t, []byte{2}, "", []byte("two"))
+	if !d.Dispatch(append(append([]byte{}, b1...), b2...)) {
+		t.Fatalf("Dispatch() = false, want true")
+	}
+
+	for i, l := range []*Listener{l1, l2} {
+		select {
+		case m := <-l.RspChan:
+			if m.Token()[0] != byte(i+1) {
+				t.Fatalf("listener %d got token % x", i+1, m.Token())
+			}
+		default:
+			t.Fatalf("listener %d did not receive its message", i+1)
+		}
+	}
+}
+
+func TestDispatcherNoListener(t *testing.T) {
+	d := NewDispatcher(true, 0)
+	if d.Dispatch(tcpMsg(t, []byte{1}, "", nil)) {
+		t.Fatalf("Dispatch() = true without a listener")
+	}
+}
diff --git a/nmxact/nmcoap/frag.go b/nmxact/nmcoap/frag.go
index a7371d7..673da9d 100644
--- a/nmxact/nmcoap/frag.go
+++ b/nmxact/nmcoap/frag.go
@@ -48,6 +48,9 @@ func (r *Reassembler) RxFrag(frag []byte) *coap.TcpMessage {
 		return nil
 	}
 
-	r.cur = nil
+	// Keep any trailing bytes; they are the start of the next message.
+	if len(r.cur) == 0 {
+		r.cur = nil
+	}
 	return tm
 }
diff --git a/nmxact/nmcoap/frag_test.go b/nmxact/nmcoap/frag_test.go
index 7a3a3ff..7aefb98 100644
--- a/nmxact/nmcoap/frag_test.go
+++ b/nmxact/nmcoap/frag_test.go
@@ -159,3 +159,53 @@ func TestReceiverTcp(t *testing.T) {
 		t.Fatalf("unexpected message %v", got)
 	}
 }
+
+// Two messages received in a single buffer must both be returned, one per
+// call.  Previously the second message was silently dropped.
+func TestReassemblerBackToBack(t *testing.T) {
+	b1 := tcpMsg(t, []byte{1}, "", []byte("first"))
+	b2 := tcpMsg(t, []byte{2}, "", []byte("second"))
+	b3 := tcpMsg(t, []byte{3}, "", []byte("third"))
+
+	r := NewReassembler()
+	buf := append(append([]byte{}, b1...), b2...)
+
+	checkTcpMsg(t, r.RxFrag(buf), []byte{1}, []byte("first"))
+	checkTcpMsg(t, r.RxFrag(nil), []byte{2}, []byte("second"))
+	if m := r.RxFrag(nil); m != nil {
+		t.Fatalf("unexpected third message: %v", m)
+	}
+
+	// First message complete plus the start of the next one.
+	buf = append(append([]byte{}, b1...), b3[:3]...)
+	checkTcpMsg(t, r.RxFrag(buf), []byte{1}, []byte("first"))
+	checkTcpMsg(t, r.RxFrag(b3[3:]), []byte{3}, []byte("third"))
+	if m := r.RxFrag(nil); m != nil {
+		t.Fatalf("unexpected extra message: %v", m)
+	}
+}
+
+func TestReceiverRxBuffered(t *testing.T) {
+	udp := NewReceiver(false)
+	if m := udp.RxBuffered(); m != nil {
+		t.Fatalf("UDP receiver returned buffered message %v", m)
+	}
+
+	tcp := NewReceiver(true)
+	if m := tcp.RxBuffered(); m != nil {
+		t.Fatalf("empty TCP receiver returned message %v", m)
+	}
+
+	b1 := tcpMsg(t, []byte{1}, "", nil)
+	b2 := tcpMsg(t, []byte{2}, "", nil)
+	if m := tcp.Rx(append(append([]byte{}, b1...), b2...)); m == nil {
+		t.Fatalf("expected first message")
+	}
+	m := tcp.RxBuffered()
+	if m == nil || m.Token()[0] != 2 {
+		t.Fatalf("expected second message, got %v", m)
+	}
+	if m := tcp.RxBuffered(); m != nil {
+		t.Fatalf("unexpected message %v", m)
+	}
+}
diff --git a/nmxact/nmcoap/receiver.go b/nmxact/nmcoap/receiver.go
index c8587ee..0835535 100644
--- a/nmxact/nmcoap/receiver.go
+++ b/nmxact/nmcoap/receiver.go
@@ -57,3 +57,18 @@ func (r *Receiver) Rx(data []byte) coap.Message {
 		return m
 	}
 }
+
+// RxBuffered returns the next complete message that is already buffered
+// from previously received data, or nil if there is none.  Only TCP can
+// buffer messages.
+func (r *Receiver) RxBuffered() coap.Message {
+	if r.reassembler == nil {
+		return nil
+	}
+
+	tm := r.reassembler.RxFrag(nil)
+	if tm == nil {
+		return nil
+	}
+	return tm
+}