fix: match Python Channel RX order, send window/MDU, and HMU segment bounds

pkg/buffer/buffer.go

@@ -6,6 +6,7 @@ package buffer
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"encoding/binary"
 	"io"
 	"sync"
@@ -187,6 +188,9 @@ func (w *RawChannelWriter) Write(p []byte) (n int, err error) {
 		processed = len(p)
 	}
 
+	if err := w.channel.WaitReady(context.Background()); err != nil {
+		return 0, err
+	}
 	if err := w.channel.Send(msg); err != nil {
 		return 0, err
 	}

pkg/buffer/buffer_test.go

@@ -223,8 +223,12 @@ func (m *mockLink) GetLinkID() []byte                                     { retu
 func (m *mockLink) Send(data []byte) any                                  { return &packet.Packet{Raw: data} }
 func (m *mockLink) Resend(p any) error                                    { return nil }
 func (m *mockLink) SetPacketTimeout(p any, cb func(any), t time.Duration) {}
-func (m *mockLink) SetPacketDelivered(p any, cb func(any))                {}
-func (m *mockLink) HandleInbound(pkt *packet.Packet) error                { return nil }
+func (m *mockLink) SetPacketDelivered(p any, cb func(any)) {
+	if cb != nil {
+		cb(p)
+	}
+}
+func (m *mockLink) HandleInbound(pkt *packet.Packet) error { return nil }
 func (m *mockLink) ValidateLinkProof(pkt *packet.Packet, networkIface common.NetworkInterface) error {
 	return nil
 }

pkg/buffer/regression_test.go

@@ -31,8 +31,12 @@ func (m *captureLink) Send(data []byte) any {
 }
 func (m *captureLink) Resend(p any) error                                    { return nil }
 func (m *captureLink) SetPacketTimeout(p any, cb func(any), t time.Duration) {}
-func (m *captureLink) SetPacketDelivered(p any, cb func(any))                {}
-func (m *captureLink) HandleInbound(pkt *packet.Packet) error                { return nil }
+func (m *captureLink) SetPacketDelivered(p any, cb func(any)) {
+	if cb != nil {
+		cb(p)
+	}
+}
+func (m *captureLink) HandleInbound(pkt *packet.Packet) error { return nil }
 func (m *captureLink) ValidateLinkProof(pkt *packet.Packet, networkIface common.NetworkInterface) error {
 	return nil
 }

pkg/channel/channel.go

@@ -17,9 +17,14 @@ import (
 	"quad4/reticulum-go/pkg/transport"
 )
 
-// ErrLinkNotReady is returned when a send is attempted on a non-ready outlet.
+// ErrLinkNotReady is returned when a send is attempted on a non-ready outlet
+// or when the TX window is full, matching Python ChannelException ME_LINK_NOT_READY.
 var ErrLinkNotReady = errors.New("link not ready")
 
+// ErrTooBig is returned when the packed envelope exceeds the outlet MDU,
+// matching Python ChannelException ME_TOO_BIG.
+var ErrTooBig = errors.New("channel message too big")
+
 // SystemMessageTypeMin is the lower bound for system-reserved MSGTYPE values.
 // Matches Python RNS Channel (MSGTYPE >= 0xf000).
 const SystemMessageTypeMin uint16 = 0xf000
@@ -56,6 +61,7 @@ type Channel struct {
 	sendMu            sync.Mutex
 	mutex             sync.RWMutex
 	txRing            []*Envelope
+	rxRing            []rxEnvelope
 	window            int
 	windowMax         int
 	windowMin         int
@@ -63,12 +69,18 @@ type Channel struct {
 	fastRateRounds    int
 	mediumRateRounds  int
 	nextSequence      uint16
+	nextRxSequence    uint16
 	maxTries          int
 	messageHandlers   []messageHandlerEntry
 	nextHandlerID     int
 	factories         map[uint16]MessageConstructor
 }
 
+type rxEnvelope struct {
+	sequence uint16
+	message  MessageBase
+}
+
 type messageHandlerEntry struct {
 	id      int
 	handler func(MessageBase) bool
@@ -161,11 +173,13 @@ func packEnvelope(msgType, sequence uint16, body []byte) ([]byte, error) {
 // Send transmits a message over the channel.
 // Sequence allocation and tx-ring emplace happen only after a successful
 // outlet send so a failing link cannot leave ghost envelopes or sequence holes.
+// A full TX window or packed envelope larger than the outlet MDU is refused,
+// matching Python Channel.send.
 func (c *Channel) Send(msg MessageBase) error {
 	c.sendMu.Lock()
 	defer c.sendMu.Unlock()
 
-	if !outletReady(c.link.GetStatus()) {
+	if !c.IsReadyToSend() {
 		return ErrLinkNotReady
 	}
 
@@ -176,16 +190,23 @@ func (c *Channel) Send(msg MessageBase) error {
 
 	c.mutex.Lock()
 	reserved := c.nextSequence
-	c.nextSequence = uint16((uint32(reserved) + 1) % SeqModulus)
 	c.mutex.Unlock()
 
 	raw, err := packEnvelope(msg.GetType(), reserved, body)
 	if err != nil {
-		c.mutex.Lock()
-		c.nextSequence = reserved
-		c.mutex.Unlock()
 		return err
 	}
+	if len(raw) > c.outletMDU() {
+		return ErrTooBig
+	}
+
+	c.mutex.Lock()
+	if c.nextSequence != reserved {
+		c.mutex.Unlock()
+		return ErrLinkNotReady
+	}
+	c.nextSequence = uint16((uint32(reserved) + 1) % SeqModulus)
+	c.mutex.Unlock()
 
 	packet := c.link.Send(raw)
 	if !packetTransmitted(packet) {
@@ -305,7 +326,8 @@ func (c *Channel) RemoveMessageHandler(id int) {
 }
 
 // HandleInbound processes an inbound channel packet and dispatches to registered handlers.
-// Registered factories unpack into typed messages. Unknown types become GenericMessage.
+// Sequences are buffered on the RX ring and delivered in order, duplicates are
+// dropped, matching Python Channel._receive.
 func (c *Channel) HandleInbound(data []byte) error {
 	if len(data) < ChannelHeaderSize {
 		return errors.New("channel packet too short")
@@ -319,15 +341,17 @@ func (c *Channel) HandleInbound(data []byte) error {
 		return errors.New("channel packet incomplete")
 	}
 
+	c.mutex.RLock()
+	stale := staleRXSequence(sequence, c.nextRxSequence)
+	ctor := c.factories[msgType]
+	c.mutex.RUnlock()
+	if stale {
+		return nil
+	}
+
 	msgData := make([]byte, length)
 	copy(msgData, data[ChannelHeaderSize:ChannelHeaderSize+int(length)])
 
-	c.mutex.RLock()
-	ctor := c.factories[msgType]
-	handlers := make([]messageHandlerEntry, len(c.messageHandlers))
-	copy(handlers, c.messageHandlers)
-	c.mutex.RUnlock()
-
 	var msg MessageBase
 	if ctor != nil {
 		msg = ctor()
@@ -342,9 +366,23 @@ func (c *Channel) HandleInbound(data []byte) error {
 		}
 	}
 
-	for _, entry := range handlers {
-		if entry.handler != nil {
-			if entry.handler(msg) {
+	c.mutex.Lock()
+	if staleRXSequence(sequence, c.nextRxSequence) {
+		c.mutex.Unlock()
+		return nil
+	}
+	if !c.emplaceRXLocked(rxEnvelope{sequence: sequence, message: msg}) {
+		c.mutex.Unlock()
+		return nil
+	}
+	delivered := c.drainRXLocked()
+	handlers := make([]messageHandlerEntry, len(c.messageHandlers))
+	copy(handlers, c.messageHandlers)
+	c.mutex.Unlock()
+
+	for _, m := range delivered {
+		for _, entry := range handlers {
+			if entry.handler != nil && entry.handler(m) {
 				break
 			}
 		}
@@ -353,6 +391,56 @@ func (c *Channel) HandleInbound(data []byte) error {
 	return nil
 }
 
+// staleRXSequence reports whether seq is behind nextRx and outside the wrap
+// window, matching Python Channel._receive WINDOW_MAX overflow logic.
+func staleRXSequence(seq, next uint16) bool {
+	if seq >= next {
+		return false
+	}
+	windowOverflow := uint16((uint32(next) + uint32(WindowMax)) % SeqModulus)
+	if windowOverflow < next {
+		return seq > windowOverflow
+	}
+	return true
+}
+
+func (c *Channel) emplaceRXLocked(env rxEnvelope) bool {
+	for i, existing := range c.rxRing {
+		if env.sequence == existing.sequence {
+			return false
+		}
+		dist := int32(c.nextRxSequence) - int32(env.sequence)
+		if env.sequence < existing.sequence && dist <= int32(SeqMax)/2 {
+			c.rxRing = append(c.rxRing, rxEnvelope{})
+			copy(c.rxRing[i+1:], c.rxRing[i:])
+			c.rxRing[i] = env
+			return true
+		}
+	}
+	c.rxRing = append(c.rxRing, env)
+	return true
+}
+
+func (c *Channel) drainRXLocked() []MessageBase {
+	var out []MessageBase
+	for {
+		found := -1
+		for i, env := range c.rxRing {
+			if env.sequence == c.nextRxSequence {
+				found = i
+				break
+			}
+		}
+		if found < 0 {
+			return out
+		}
+		env := c.rxRing[found]
+		c.rxRing = append(c.rxRing[:found], c.rxRing[found+1:]...)
+		out = append(out, env.message)
+		c.nextRxSequence = uint16((uint32(c.nextRxSequence) + 1) % SeqModulus)
+	}
+}
+
 // GenericMessage is a default message implementation with type, data, and sequence.
 type GenericMessage struct {
 	Type uint16
@@ -411,16 +499,22 @@ type mduOutlet interface {
 	GetMDU() int
 }
 
+func (c *Channel) outletMDU() int {
+	mdu := DefaultOutletMDU
+	if c.link != nil {
+		if g, ok := c.link.(mduOutlet); ok {
+			if n := g.GetMDU(); n > 0 {
+				mdu = n
+			}
+		}
+	}
+	return mdu
+}
+
 // MDU is bytes available for a channel message body, matching Python
 // Channel.mdu (outlet MDU minus 6-byte envelope header).
 func (c *Channel) MDU() int {
-	mdu := DefaultOutletMDU
-	if g, ok := c.link.(mduOutlet); ok {
-		if n := g.GetMDU(); n > 0 {
-			mdu = n
-		}
-	}
-	mdu -= ChannelHeaderSize
+	mdu := c.outletMDU() - ChannelHeaderSize
 	if mdu > 0xFFFF {
 		mdu = 0xFFFF
 	}
@@ -522,5 +616,6 @@ func (c *Channel) Close() error {
 		releaseEnvelope(env)
 	}
 	c.txRing = nil
+	c.rxRing = nil
 	return nil
 }

pkg/channel/channel_test.go

@@ -119,7 +119,7 @@ func TestHandleInboundTypedFactory(t *testing.T) {
 		return true
 	})
 
-	raw, err := packEnvelope(1, 7, []byte("abcd"))
+	raw, err := packEnvelope(1, 0, []byte("abcd"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -142,7 +142,7 @@ func TestHandleInbound(t *testing.T) {
 	})
 
 	// Packet format: [type 2][seq 2][len 2][data]
-	data := []byte{0, 1, 0, 1, 0, 4, 't', 'e', 's', 't'}
+	data := []byte{0, 1, 0, 0, 0, 4, 't', 'e', 's', 't'}
 	err := c.HandleInbound(data)
 	if err != nil {
 		t.Fatalf("HandleInbound failed: %v", err)
@@ -190,6 +190,7 @@ type scaleMockLink struct {
 }
 
 func (m *scaleMockLink) GetStatus() byte                                       { return m.status }
+func (m *scaleMockLink) GetMDU() int                                           { return DefaultOutletMDU }
 func (m *scaleMockLink) GetRTT() float64                                       { return 0.1 }
 func (m *scaleMockLink) RTT() float64                                          { return 0.1 }
 func (m *scaleMockLink) GetLinkID() []byte                                     { return []byte("mocklink") }
@@ -243,14 +244,14 @@ func BenchmarkChannelSendScale(b *testing.B) {
 	b.ReportAllocs()
 
 	for i := 0; i < b.N; i++ {
-		_ = ch.Send(msg)
-		if i%100 == 0 {
+		if err := ch.Send(msg); err != nil {
 			ch.mutex.Lock()
 			for _, env := range ch.txRing {
 				releaseEnvelope(env)
 			}
 			ch.txRing = nil
 			ch.mutex.Unlock()
+			_ = ch.Send(msg)
 		}
 	}
 	ch.mutex.Lock()

pkg/channel/fuzz_test.go

@@ -45,6 +45,13 @@ func FuzzHandleInboundEnvelopeExploratory(f *testing.F) {
 		if len(data) < ChannelHeaderSize+int(length) {
 			t.Fatal("HandleInbound succeeded when declared length exceeds buffer")
 		}
+		seq := binary.BigEndian.Uint16(data[2:4])
+		if seq != 0 {
+			if got != nil {
+				t.Fatal("nonzero first sequence must not dispatch")
+			}
+			return
+		}
 		gm, ok := got.(*GenericMessage)
 		if !ok {
 			return
@@ -85,6 +92,12 @@ func FuzzPackHandleInboundRoundTrip(f *testing.F) {
 		if err := ch.HandleInbound(raw); err != nil {
 			t.Fatalf("HandleInbound: %v", err)
 		}
+		if seq != 0 {
+			if got != nil {
+				t.Fatal("nonzero first sequence must not dispatch")
+			}
+			return
+		}
 		if got == nil {
 			t.Fatal("handler did not receive GenericMessage")
 		}

pkg/channel/oracle_rx_sequence_test.go

@@ -0,0 +1,113 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2024-2026 Quad4.io
+
+package channel
+
+import (
+	"errors"
+	"testing"
+
+	"quad4/reticulum-go/pkg/transport"
+)
+
+func TestHandleInboundReordersAndDropsDuplicates(t *testing.T) {
+	link := &mockLink{status: transport.StatusActive}
+	c := NewChannel(link)
+	defer func() { _ = c.Close() }()
+
+	if err := c.RegisterMessageType(1, func() MessageBase { return &testMessage{} }); err != nil {
+		t.Fatal(err)
+	}
+
+	var got []string
+	c.AddMessageHandler(func(msg MessageBase) bool {
+		if m, ok := msg.(*testMessage); ok {
+			got = append(got, string(m.data))
+		}
+		return true
+	})
+
+	raw1, err := packEnvelope(1, 1, []byte("B"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw0, err := packEnvelope(1, 0, []byte("A"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := c.HandleInbound(raw1); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("seq 1 before 0 dispatched %v", got)
+	}
+	if err := c.HandleInbound(raw0); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 2 || got[0] != "A" || got[1] != "B" {
+		t.Fatalf("order=%v want [A B]", got)
+	}
+
+	if err := c.HandleInbound(raw0); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("duplicate seq 0 dispatched, got %v", got)
+	}
+}
+
+func TestSendRefusesFullWindow(t *testing.T) {
+	link := &mockLink{status: transport.StatusActive}
+	c := NewChannel(link)
+	defer func() { _ = c.Close() }()
+
+	for i := 0; i < WindowInitial; i++ {
+		if err := c.Send(&testMessage{data: []byte{byte(i)}}); err != nil {
+			t.Fatalf("setup send %d: %v", i, err)
+		}
+	}
+	if c.IsReadyToSend() {
+		t.Fatal("window should be full")
+	}
+	err := c.Send(&testMessage{data: []byte("overflow")})
+	if !errors.Is(err, ErrLinkNotReady) {
+		t.Fatalf("Send: got %v want ErrLinkNotReady", err)
+	}
+	if c.TxRingLen() != WindowInitial {
+		t.Fatalf("tx ring len=%d want %d", c.TxRingLen(), WindowInitial)
+	}
+}
+
+func TestSendRefusesOversizeEnvelope(t *testing.T) {
+	link := &mockLink{status: transport.StatusActive}
+	c := NewChannel(link)
+	defer func() { _ = c.Close() }()
+
+	body := make([]byte, c.MDU()+64)
+	err := c.Send(&testMessage{data: body})
+	if !errors.Is(err, ErrTooBig) {
+		t.Fatalf("Send: got %v want ErrTooBig", err)
+	}
+	if c.TxRingLen() != 0 {
+		t.Fatalf("tx ring len=%d want 0", c.TxRingLen())
+	}
+	if c.NextSequence() != 0 {
+		t.Fatalf("sequence consumed on oversized send: %d", c.NextSequence())
+	}
+}
+
+func TestStaleRXSequenceWrapWindow(t *testing.T) {
+	if staleRXSequence(5, 10) != true {
+		t.Fatal("seq 5 must be stale when next is 10")
+	}
+	if staleRXSequence(10, 10) != false {
+		t.Fatal("current next sequence is not stale")
+	}
+	if staleRXSequence(0, 65530) != false {
+		t.Fatal("seq 0 after wrap must be accepted while next is 65530")
+	}
+	if staleRXSequence(100, 65530) != true {
+		t.Fatal("seq 100 must be stale when next is 65530")
+	}
+}

pkg/channel/regression_test.go

@@ -73,7 +73,7 @@ func TestRegression_HandleInboundUsesRegisteredFactory(t *testing.T) {
 		return true
 	})
 
-	raw, err := packEnvelope(1, 3, []byte("typed"))
+	raw, err := packEnvelope(1, 0, []byte("typed"))
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/link/hmu_negative_segment_oracle_test.go

@@ -50,4 +50,14 @@ func TestOracleHMUOversizeSegmentDoesNotPanic(t *testing.T) {
 	if added != 0 {
 		t.Fatalf("expected 0 entries applied for oversize segment, got %d", added)
 	}
+
+	added = rx.applyHashmapSegment(1<<61, hashmapBytes)
+	if added != 0 {
+		t.Fatalf("expected 0 entries applied for overflowing segment, got %d", added)
+	}
+	for i, mh := range rx.mapHashes {
+		if mh != nil {
+			t.Fatalf("mapHashes[%d] was populated by an overflowing-segment HMU", i)
+		}
+	}
 }

pkg/link/incoming_resource.go

@@ -129,9 +129,8 @@ type incomingResourceAsm struct {
 func (rx *incomingResourceAsm) applyHashmapSegment(segment int, hashmapBytes []byte) int {
 	// segment is parsed straight off the wire in handleResourceHashmapUpdate
 	// (wireInt(update[0])) and can be any attacker-chosen int, including
-	// negative. Reject it here rather than trusting callers, since a
-	// negative segment*segLen would otherwise produce a negative slice
-	// index below and panic the process (remote DoS).
+	// negative or so large that segment*segLen overflows int. Reject
+	// those here. A wrapping multiply would otherwise write hashmap[0].
 	if segment < 0 {
 		return 0
 	}
@@ -139,10 +138,17 @@ func (rx *incomingResourceAsm) applyHashmapSegment(segment int, hashmapBytes []b
 	if segLen <= 0 {
 		segLen = 1
 	}
+	if segment > math.MaxInt/segLen {
+		return 0
+	}
+	base := segment * segLen
 	added := 0
 	hashes := len(hashmapBytes) / resource.MapHashLen
 	for i := range hashes {
-		idx := i + segment*segLen
+		if i > math.MaxInt-base {
+			return added
+		}
+		idx := base + i
 		if idx < 0 || idx >= rx.totalParts {
 			return added
 		}

pkg/link/oracle_hmu_wrap_test.go

@@ -0,0 +1,32 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2024-2026 Quad4.io
+
+package link
+
+import (
+	"bytes"
+	"testing"
+
+	"quad4/reticulum-go/pkg/resource"
+)
+
+func TestHMUSegmentIndexWrapDoesNotOverwriteSlotZero(t *testing.T) {
+	rx := &incomingResourceAsm{
+		hashmapSegLen: 8,
+		totalParts:    16,
+		partSlots:     make([][]byte, 16),
+		mapHashes:     make([][]byte, 16),
+	}
+
+	marker := bytes.Repeat([]byte{0xAA}, resource.MapHashLen)
+	hashmapBytes := make([]byte, resource.MapHashLen*4)
+	copy(hashmapBytes, marker)
+
+	added := rx.applyHashmapSegment(1<<61, hashmapBytes)
+	if added != 0 {
+		t.Fatalf("wrapped segment applied %d entries", added)
+	}
+	if rx.mapHashes[0] != nil {
+		t.Fatalf("slot 0 overwritten by wrapping HMU segment: %x", rx.mapHashes[0])
+	}
+}