diff --git a/pkg/router/reorder.go b/pkg/router/reorder.go index 5c81bb729e..a55c5f51e4 100644 --- a/pkg/router/reorder.go +++ b/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() diff --git a/pkg/router/reorder_test.go b/pkg/router/reorder_test.go index 259e1bc5d2..395c2b927f 100644 --- a/pkg/router/reorder_test.go +++ b/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()) } diff --git a/pkg/router/route_mux.go b/pkg/router/route_mux.go index 30496b7d48..cdca36edf8 100644 --- a/pkg/router/route_mux.go +++ b/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 {