fix(router): size reorder buffer to the BDP + drop-not-skip at the OOM cap

pkg/router/reorder.go

@@ -2,7 +2,6 @@
 package router
 
 import (
-	"sort"
 	"sync"
 	"time"
 )
@@ -60,6 +59,18 @@ func (rb *reorderBuffer) Insert(seq uint32, data []byte) [][]byte {
 
 	// Buffer out-of-order packet
 	if seq != rb.nextSeq {
+		// Emergency OOM backstop: if the buffer is already at capacity, DROP this
+		// out-of-order packet rather than skip the gap (delivering past the missing
+		// seq corrupts the reliable stream — the bad-record-mac failure). The buffer
+		// stays bounded at maxGap; the dropped seq is re-requested by SACK and
+		// retransmitted, and the real recovery is the leg-dataprogress prune removing
+		// the dead leg so its seqs retransmit on a live one and the frontier drains
+		// IN ORDER. If the gap never fills, keepalive/liveness closes the group
+		// cleanly — a stall, never corruption. maxGap (reorderWindow) is sized to the
+		// aggregate BDP so this is only reached on a genuine mid-stream leg death.
+		if len(rb.buf) >= rb.maxGap {
+			return nil
+		}
 		// Make a copy since the underlying packet buffer may be reused
 		cp := make([]byte, len(data))
 		copy(cp, data)
@@ -96,17 +107,6 @@ func (rb *reorderBuffer) Insert(seq uint32, data []byte) [][]byte {
 		// only ever closed by the missing seq actually arriving (skew or retransmit),
 		// never by skipping it.
 
-		// Emergency OOM backstop only: with reliable legs the missing seq arrives
-		// (via skew or SACK retransmit) and drains the buffer, so this never trips
-		// in normal operation. Reaching it means a leg has gone dead mid-stream AND
-		// retransmit has not refilled within maxGap packets. Force-flushing here
-		// still skips the gap (and so corrupts a noise stream) — it is a
-		// last-resort guard against unbounded memory, not a recovery path; the
-		// route group's liveness prune + teardown is the real handler. Do NOT lower
-		// maxGap to "save memory": that reintroduces the mux>1 corruption bug.
-		if len(rb.buf) >= rb.maxGap {
-			return rb.flushAll()
-		}
 		return nil
 	}
 
@@ -150,37 +150,6 @@ func (rb *reorderBuffer) GapAge() time.Duration {
 	return time.Since(rb.gapSince)
 }
 
-// flushAll delivers all buffered packets in sequence order and resets.
-// Called when the gap exceeds maxGap to prevent unbounded memory growth.
-// The caller must hold rb.mu.
-func (rb *reorderBuffer) flushAll() [][]byte {
-	if len(rb.buf) == 0 {
-		return nil
-	}
-
-	seqs := make([]uint32, 0, len(rb.buf))
-	for seq := range rb.buf {
-		seqs = append(seqs, seq)
-	}
-	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
-
-	delivered := make([][]byte, 0, len(seqs))
-	for _, seq := range seqs {
-		delivered = append(delivered, rb.buf[seq])
-		delete(rb.buf, seq)
-	}
-
-	// Advance nextSeq past all delivered
-	if len(seqs) > 0 {
-		rb.nextSeq = seqs[len(seqs)-1] + 1
-	}
-
-	// Buffer drained — the gap is closed (skipped).
-	rb.gapSince = time.Time{}
-
-	return delivered
-}
-
 // Pending returns the number of packets currently buffered out-of-order.
 func (rb *reorderBuffer) Pending() int {
 	rb.mu.Lock()

pkg/router/reorder_test.go

@@ -56,17 +56,22 @@ func TestReorderBuffer_Duplicate(t *testing.T) {
 	assert.Nil(t, d)
 }
 
-func TestReorderBuffer_ForceFlush(t *testing.T) {
+func TestReorderBuffer_MaxGapDropsExcessNeverSkips(t *testing.T) {
+	// At the OOM backstop (maxGap) the buffer DROPS excess out-of-order packets —
+	// it never skips the frontier gap (skipping would corrupt the reliable stream).
 	rb := newReorderBuffer(3)
-	// Skip seq 0, send 1, 2, 3 — triggers flush at maxGap=3
-	rb.Insert(1, []byte("b"))
-	rb.Insert(2, []byte("c"))
-	d := rb.Insert(3, []byte("d"))
-	// Should flush all buffered in order
-	assert.Equal(t, 3, len(d))
-	assert.Equal(t, []byte("b"), d[0])
-	assert.Equal(t, []byte("c"), d[1])
-	assert.Equal(t, []byte("d"), d[2])
+	// Gap at seq 0; buffer 1,2,3 fills to maxGap — none delivered (frontier held).
+	assert.Nil(t, rb.Insert(1, []byte("b")))
+	assert.Nil(t, rb.Insert(2, []byte("c")))
+	assert.Nil(t, rb.Insert(3, []byte("d")))
+	assert.Equal(t, 3, rb.Pending())
+	// At capacity, a further out-of-order packet is DROPPED (not buffered), and the
+	// gap is still NOT skipped — seq 4 will be re-requested via SACK.
+	assert.Nil(t, rb.Insert(4, []byte("e")))
+	assert.Equal(t, 3, rb.Pending(), "buffer bounded at maxGap; excess dropped, gap never skipped")
+	// The missing seq 0 finally arrives -> the frontier drains IN ORDER (0..3).
+	d := rb.Insert(0, []byte("a"))
+	assert.Equal(t, [][]byte{[]byte("a"), []byte("b"), []byte("c"), []byte("d")}, d)
 	assert.Equal(t, 0, rb.Pending())
 }
 

pkg/router/route_mux.go

@@ -190,7 +190,16 @@ type routeMux struct {
 // flushes. A flush at this cap is a last-resort OOM guard for a genuinely
 // stalled/dead leg — which per-leg liveness prunes, after which the peer
 // retransmits that leg's unacked sequences on the surviving legs.
-const reorderWindow = 2048
+// Sized to the aggregate bandwidth-delay-product per the MPTCP receive-buffer
+// requirement B >= 2*sum(BW)*RTT_max: at the ~500 Mbps gigabit target with a
+// ~350 ms slowest-active-leg RTT that is ~46 MB ~= 32Ki packets, so the old 2Ki
+// (~2.8 MB) window was ~16x too small — it collapsed throughput under wide-mux
+// skew and hit the OOM backstop. This is the CAP, not steady occupancy: normal
+// skew buffers only a handful; only a very lagged/dead leg approaches it, and at
+// the cap the buffer now DROPS excess (never skips) while the leg-dataprogress
+// prune + SACK retransmit refill the frontier in order. TODO: make adaptive to
+// the measured RTT_max of the active set instead of a flat gigabit-sized cap.
+const reorderWindow = 32768
 
 // newRouteMux creates a new routeMux instance with all sub-components initialized.
 func newRouteMux(logger *logging.Logger, sackEnabled bool) *routeMux {