diff --git a/connection_state.go b/connection_state.go index 60e1b014..789e2c20 100644 --- a/connection_state.go +++ b/connection_state.go @@ -14,6 +14,16 @@ import ( const ReplayWindow = 8192 +// sessionEpoch hands out a receiver-local ordinal to every ConnectionState as +// it is created. The RX staging sort (overlay/batch) orders packets by +// (epoch, message counter). A re-handshake never rekeys an existing tunnel; it +// brings up a whole new hostinfo (and ConnectionState) with its own counter +// space starting near zero, while the old tunnel keeps decrypting in-flight +// packets until it is torn down. During that cutover one flush batch can hold +// packets from both tunnels, and the epoch is what keeps the old tunnel's +// packets sorted ahead of the new tunnel's. +var sessionEpoch atomic.Uint64 + type ConnectionState struct { eKey noiseutil.CipherState dKey noiseutil.CipherState @@ -24,6 +34,8 @@ type ConnectionState struct { window *Bits decryptLock sync.Mutex writeLock sync.Mutex + // epoch is this session's sessionEpoch ordinal. Immutable after creation. + epoch uint64 } // newConnectionStateFromResult builds a fully-populated ConnectionState from a @@ -38,6 +50,7 @@ func newConnectionStateFromResult(r *handshake.Result) *ConnectionState { eKey: noiseutil.NewCipherState(r.EKey, r.Cipher), dKey: noiseutil.NewCipherState(r.DKey, r.Cipher), window: NewBits(ReplayWindow), + epoch: sessionEpoch.Add(1), } ci.messageCounter.Add(r.MessageIndex) for i := uint64(1); i <= r.MessageIndex; i++ { diff --git a/outside.go b/outside.go index a19f2bdb..f6f0c6d1 100644 --- a/outside.go +++ b/outside.go @@ -13,6 +13,7 @@ import ( "github.com/slackhq/nebula/firewall" "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/overlay/batch" "golang.org/x/net/ipv4" ) @@ -143,7 +144,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, scratch []byte, packet []b case header.Message: switch h.Subtype { case header.MessageNone: - f.handleOutsideMessagePacket(hostinfo, out, scratch, fwPacket, nb, q, localCache) + f.handleOutsideMessagePacket(hostinfo, h.MessageCounter, out, scratch, fwPacket, nb, q, localCache) default: hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected message subtype seen", "from", via, "header", h) return @@ -500,7 +501,7 @@ func parseV4(data []byte, incoming bool, fp *firewall.Packet) error { return nil } -func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, scratch []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache) { +func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounter uint64, out []byte, scratch []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache) { err := newPacket(out, true, fwPacket) if err != nil { hostinfo.logger(f.l).Warn("Error while validating inbound packet", @@ -522,7 +523,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, s return } - err = f.batchers[q].Commit(out) + err = f.batchers[q].Commit(out, batch.SortKey{Epoch: hostinfo.ConnectionState.epoch, Counter: messageCounter}) if err != nil { f.l.Error("Failed to write to tun", "error", err) } diff --git a/overlay/batch/batch.go b/overlay/batch/batch.go index 7eda8c97..eadb8573 100644 --- a/overlay/batch/batch.go +++ b/overlay/batch/batch.go @@ -1,14 +1,31 @@ package batch +// SortKey identifies a packet's position in its sender's transmission order. +// Epoch is a receiver-local ordinal for the tunnel (ConnectionState) that +// decrypted the packet. A re-handshake replaces the tunnel outright — new +// hostinfo, new keys, a fresh counter space — and the replacement's epoch is +// higher, so during the cutover overlap the old tunnel's packets sort first. +// Counter is the packet's AEAD message counter within that tunnel. The replay +// window has already rejected duplicates by Commit time, so keys are unique +// per tunnel and (Epoch, Counter) is a total order with no ties. +type SortKey struct { + Epoch uint64 + Counter uint64 +} + type RxBatcher interface { - // Commit commits pkt to be flushed by the batch. The caller must keep pkt valid until the next Flush, and not re-use it. - Commit(pkt []byte) error - // Flush emits every queued packet. The guarantee is per-flow DATA order: - // a flow's payload-bearing packets are never reordered relative to each - // other. Cross-flow and cross-lane order is not preserved, and two shapes - // may legally be overtaken by later same-flow data: pure ACKs (by design, - // stale ACKs are ignored) and unparseable shapes such as fragments (an accepted tradeoff; see MultiCoalescer). - // Returns the first error observed; keeps draining so one bad packet doesn't hold up the rest. + // Commit stages pkt to be flushed by the batch. key must carry the + // packet's session epoch and message counter. The caller must keep pkt + // valid until the next Flush, and not re-use it. + Commit(pkt []byte, key SortKey) error + // Flush emits every staged packet. Packets are first sorted by key, so + // within each protocol lane emission follows the sender's transmission + // order regardless of arrival order. One shape may legally be overtaken + // by later same-flow data: a pure TCP ACK, which does not close its + // flow's open coalesce chain (a late ACK is just a stale ACK). Cross-lane + // order (TCP vs UDP vs everything else) is not preserved. + // Returns the first error observed; keeps draining so one bad packet + // doesn't hold up the rest. // After Flush returns, committed payload slices may be recycled. Flush() error } diff --git a/overlay/batch/multi_coalesce.go b/overlay/batch/multi_coalesce.go index 8a4fceab..3ee4cbe2 100644 --- a/overlay/batch/multi_coalesce.go +++ b/overlay/batch/multi_coalesce.go @@ -4,50 +4,64 @@ import ( "errors" "io" "log/slog" + "slices" "github.com/slackhq/nebula/iputil" ) -// MultiCoalescer fans plaintext packets out to lane-specific batchers based -// on the IP/L4 protocol of the packet. +// MultiCoalescer stages plaintext packets with their (epoch, counter) sort +// keys, and at Flush replays them in sender-transmission order into +// lane-specific batchers selected by the IP/L4 protocol of the packet. // -// Lanes are processed independently: the TCP coalescer only sees TCP, the -// UDP coalescer only sees UDP, and the verbatim lane handles everything else. -// The ordering contract is per-flow DATA order: a flow's payload-bearing -// packets are never reordered relative to each other, because a single -// 5-tuple only ever lands in one lane and each lane emits its slots in -// creation order. Two shapes are deliberately allowed to be overtaken by -// later same-flow data: -// - pure ACKs, which pass through without sealing the flow's open slot -// (a late ACK is just a stale ACK; see TCPCoalescer.commitParsed); -// - unparseable in-flow shapes (fragments, IP options), whose lane-level -// addVerbatim does not close the flow's open slot either. Closing it -// would need a full open-slot barrier (the flow key is unknown when the -// parse fails) — an accepted tradeoff: mid-flow fragments are rare and -// receivers reassemble regardless of arrival order. +// Sorting *before* the lanes see anything is what makes the ordering story +// simple: each lane consumes packets in transmission order, builds its slots +// in that order, and emits them in creation order. Wire reorder inside a +// flush batch is repaired here, before it can fragment a lane's coalesce +// chains, so the lanes carry no reorder-repair machinery of their own. // -// Routing still follows the flow, not the coalesceability: IPv4 fragments -// keep their L4 proto visible and IPv6 extension chains are walked to the -// terminal proto, so a flow's non-coalesceable shapes ride its lane as -// in-lane passthroughs rather than falling to the later-flushed pt lane. +// The ordering contract is per-tunnel transmission order within each lane: +// a sender's packets are emitted in the order it encrypted them. Two +// qualifications: +// - a pure TCP ACK may be overtaken by later same-flow data, because it +// does not close the flow's open coalesce chain (a late ACK is just a +// stale ACK; see TCPCoalescer.commitParsed); +// - an unparseable shape (fragment, IP options) seals every open chain in +// its lane — its flow is unknowable, so this is the only way to keep +// later data from extending a chain that would emit ahead of it. The +// packet then rides its lane as an in-lane passthrough, still in +// transmission order. +// +// Routing follows the flow, not the coalesceability: IPv4 fragments keep +// their L4 proto visible and IPv6 extension chains are walked to the +// terminal proto, so a flow's non-coalesceable shapes ride its lane rather +// than falling to the later-flushed pt lane. // // Cross-lane order is intentionally NOT preserved across the TCP/UDP/verbatim split. type MultiCoalescer struct { tcp *TCPCoalescer udp *UDPCoalescer pt *Passthrough + + // staged holds this batch's packets and sort keys until Flush. Borrowed: + // the caller keeps each pkt alive until Flush returns. + staged []stagedPacket } -// NewMultiCoalescer builds a multi-lane batcher over w, based on available protocol support. +type stagedPacket struct { + pkt []byte + key SortKey +} + +// NewMultiCoalescer builds a multi-lane batcher over w, based on available +// protocol support. The staging sort applies even when no GSO lane is +// available: passthrough-only platforms still get transmission-order repair. func NewMultiCoalescer(w io.Writer, l *slog.Logger) RxBatcher { m := &MultiCoalescer{ - pt: NewPassthrough(w), + pt: NewPassthrough(w), + staged: make([]stagedPacket, 0, initialSlots), } m.tcp = NewTCPCoalescer(w, l) m.udp = NewUDPCoalescer(w) - if m.tcp == nil && m.udp == nil { - return m.pt //no offloads? Use verbatim directly. - } return m } @@ -73,12 +87,40 @@ func isIPv6ExtHeader(nh byte) bool { return false } -// Commit dispatches pkt to the appropriate lane based on IP version + L4 proto. -// On the success path the IP/TCP-or-UDP parse happens here once and the -// parsed struct is handed to the lane via commitParsed so the lane doesn't re-walk the header. -func (m *MultiCoalescer) Commit(pkt []byte) error { +// Commit stages pkt for the next Flush. All parsing and lane dispatch is +// deferred to Flush so it runs on packets already in transmission order. +func (m *MultiCoalescer) Commit(pkt []byte, key SortKey) error { + m.staged = append(m.staged, stagedPacket{pkt: pkt, key: key}) + return nil +} + +// compareStaged orders staged packets by (epoch, counter): sender +// transmission order within a tunnel, tunnel-creation order across a +// re-handshake cutover. Keys are unique (see SortKey), so this is a total +// order and sort stability doesn't matter. +func compareStaged(a, b stagedPacket) int { + if a.key.Epoch != b.key.Epoch { + if a.key.Epoch < b.key.Epoch { + return -1 + } + return 1 + } + if a.key.Counter == b.key.Counter { + return 0 + } + if a.key.Counter < b.key.Counter { + return -1 + } + return 1 +} + +// dispatch routes one packet to the appropriate lane based on IP version + +// L4 proto. On the success path the IP/TCP-or-UDP parse happens here once +// and the parsed struct is handed to the lane via commitParsed so the lane +// doesn't re-walk the header. +func (m *MultiCoalescer) dispatch(pkt []byte) error { if len(pkt) < 20 { - return m.pt.Commit(pkt) + return m.pt.enqueue(pkt) } v := pkt[0] >> 4 var proto byte @@ -87,7 +129,7 @@ func (m *MultiCoalescer) Commit(pkt []byte) error { proto = pkt[9] case 6: if len(pkt) < 40 { - return m.pt.Commit(pkt) + return m.pt.enqueue(pkt) } proto = pkt[6] if isIPv6ExtHeader(proto) { @@ -96,15 +138,18 @@ func (m *MultiCoalescer) Commit(pkt []byte) error { proto, _, _ = iputil.IPv6FindUpperProtocol(pkt) } default: - return m.pt.Commit(pkt) + return m.pt.enqueue(pkt) } switch proto { case ipProtoTCP: if m.tcp != nil { info, ok := parseTCPBase(pkt) if !ok { - // Malformed/unsupported TCP shape (IP options, fragments, ...). - // Handle this via verbatim support in the TCP coalescer, to attempt to preserve flow order. + // Unsupported TCP shape (IP options, fragments, ...). Its flow + // key is unknowable, so seal every open chain: dispatch runs in + // transmission order, and sealing is what keeps later data from + // extending a chain that would emit ahead of this packet. + m.tcp.sealAllOpen() m.tcp.addVerbatim(pkt) return nil } @@ -114,17 +159,32 @@ func (m *MultiCoalescer) Commit(pkt []byte) error { if m.udp != nil { info, ok := parseUDP(pkt) if !ok { + m.udp.sealAllOpen() m.udp.addVerbatim(pkt) return nil } return m.udp.commitParsed(pkt, info) } } - return m.pt.Commit(pkt) + return m.pt.enqueue(pkt) } +// Flush sorts the staged batch into transmission order, replays it into the +// lanes, then flushes each lane. func (m *MultiCoalescer) Flush() error { + // Arrival order is already almost sorted (reorder is the exception, not + // the rule), which pdqsort detects and handles in near-linear time. + slices.SortFunc(m.staged, compareStaged) + var errs []error + for _, sp := range m.staged { + if err := m.dispatch(sp.pkt); err != nil { + errs = append(errs, err) + } + } + clear(m.staged) // drop borrowed pkt refs + m.staged = m.staged[:0] + if m.tcp != nil { if err := m.tcp.Flush(); err != nil { errs = append(errs, err) diff --git a/overlay/batch/multi_coalesce_test.go b/overlay/batch/multi_coalesce_test.go index 97fc2881..ea138d6c 100644 --- a/overlay/batch/multi_coalesce_test.go +++ b/overlay/batch/multi_coalesce_test.go @@ -9,10 +9,19 @@ import ( "github.com/slackhq/nebula/test" ) -// newTestMultiCoalescer builds a batcher over w and asserts it really is -// multi-lane. NewMultiCoalescer collapses to a bare Passthrough when w can -// offload neither protocol, and a test that meant to exercise a lane would -// otherwise pass vacuously. +// keySeq hands out SortKeys with ascending counters in a fixed epoch, for +// tests where commit order IS transmission order. +type keySeq struct { + epoch, counter uint64 +} + +func (k *keySeq) next() SortKey { + k.counter++ + return SortKey{Epoch: k.epoch, Counter: k.counter} +} + +// newTestMultiCoalescer builds a batcher over w and asserts the concrete +// type so tests can reach into the lanes. func newTestMultiCoalescer(tb testing.TB, w io.Writer) *MultiCoalescer { tb.Helper() b := NewMultiCoalescer(w, test.NewLogger()) @@ -29,6 +38,7 @@ func newTestMultiCoalescer(tb testing.TB, w io.Writer) *MultiCoalescer { func TestMultiCoalescerRoutesByProto(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} m := newTestMultiCoalescer(t, w) + k := &keySeq{epoch: 1} tcpPay := make([]byte, 1200) udpPay := make([]byte, 1200) @@ -38,19 +48,19 @@ func TestMultiCoalescerRoutesByProto(t *testing.T) { icmp[3] = 28 icmp[9] = 1 - if err := m.Commit(buildTCPv4(1000, tcpAck, tcpPay)); err != nil { + if err := m.Commit(buildTCPv4(1000, tcpAck, tcpPay), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildTCPv4(2200, tcpAck, tcpPay)); err != nil { + if err := m.Commit(buildTCPv4(2200, tcpAck, tcpPay), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildUDPv4(2000, 53, udpPay)); err != nil { + if err := m.Commit(buildUDPv4(2000, 53, udpPay), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildUDPv4(2000, 53, udpPay)); err != nil { + if err := m.Commit(buildUDPv4(2000, 53, udpPay), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(icmp); err != nil { + if err := m.Commit(icmp, k.next()); err != nil { t.Fatal(err) } if err := m.Flush(); err != nil { @@ -65,20 +75,162 @@ func TestMultiCoalescerRoutesByProto(t *testing.T) { } } +// TestMultiCoalescerRestoresTransmissionOrder is the core staging-sort +// property: packets committed out of counter order (wire reorder inside one +// flush batch) are replayed into the lanes in transmission order, so the +// reorder never fragments the coalesce chain — one superpacket, in seq +// order, exactly as if the wire had never reordered. The retransmit shape +// falls out of the same key: a retransmit carries a lower seq but a HIGHER +// counter (it was encrypted later), so it emits after the data it trails. +func TestMultiCoalescerRestoresTransmissionOrder(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + m := newTestMultiCoalescer(t, w) + pay := make([]byte, 1200) + + // Transmission order: seq 1000 (c1), 2200 (c2), 3400 (c3). + // Arrival order: 3400, 1000, 2200. + if err := m.Commit(buildTCPv4(3400, tcpAck, pay), SortKey{Epoch: 1, Counter: 3}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4(1000, tcpAck, pay), SortKey{Epoch: 1, Counter: 1}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4(2200, tcpAck, pay), SortKey{Epoch: 1, Counter: 2}); err != nil { + t.Fatal(err) + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 1 || len(w.writes) != 0 { + t.Fatalf("want 1 gso write (unfragmented chain), got gso=%d plain=%d", len(w.gsoWrites), len(w.writes)) + } + g := w.gsoWrites[0] + if len(g.pays) != 3 { + t.Fatalf("segs=%d want 3", len(g.pays)) + } + const ipHdrLen = 20 + if seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]); seedSeq != 1000 { + t.Errorf("seed seq=%d want 1000", seedSeq) + } + + // Retransmit: seq 1000 again but counter 4 — sorts after seq 4600 (c3). + w.writes, w.gsoWrites, w.order = nil, nil, nil + if err := m.Commit(buildTCPv4(1000, tcpAck, pay), SortKey{Epoch: 1, Counter: 4}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4(4600, tcpAck, pay), SortKey{Epoch: 1, Counter: 3}); err != nil { + t.Fatal(err) + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + if len(w.writes) != 2 { + t.Fatalf("want 2 plain writes, got %d (gso=%d)", len(w.writes), len(w.gsoWrites)) + } + first := binary.BigEndian.Uint32(w.writes[0][24:28]) + second := binary.BigEndian.Uint32(w.writes[1][24:28]) + if first != 4600 || second != 1000 { + t.Fatalf("emission (%d, %d), want (4600, 1000): retransmit must not overtake in-flight data", first, second) + } +} + +// TestMultiCoalescerRestoresOrderAcrossFlows scrambles two interleaved flows; +// the staging sort must repair each flow into one superpacket without any +// cross-flow contamination. +func TestMultiCoalescerRestoresOrderAcrossFlows(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + m := newTestMultiCoalescer(t, w) + pay := make([]byte, 1200) + + // Transmission: A.100 (c1), B.500 (c2), A.1300 (c3), B.1700 (c4). + // Arrival: A.1300, B.1700, A.100, B.500. + if err := m.Commit(buildTCPv4Ports(1000, 2000, 1300, tcpAck, pay), SortKey{Epoch: 1, Counter: 3}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4Ports(3000, 2000, 1700, tcpAck, pay), SortKey{Epoch: 1, Counter: 4}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay), SortKey{Epoch: 1, Counter: 1}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay), SortKey{Epoch: 1, Counter: 2}); err != nil { + t.Fatal(err) + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 2 { + t.Fatalf("want 2 gso writes (one per flow), got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + } + for i, g := range w.gsoWrites { + if len(g.pays) != 2 { + t.Errorf("gso[%d] segs=%d want 2", i, len(g.pays)) + } + const ipHdrLen = 20 + seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]) + sport := binary.BigEndian.Uint16(g.hdr[ipHdrLen : ipHdrLen+2]) + switch sport { + case 1000: + if seedSeq != 100 { + t.Errorf("flow A seed seq=%d want 100", seedSeq) + } + case 3000: + if seedSeq != 500 { + t.Errorf("flow B seed seq=%d want 500", seedSeq) + } + default: + t.Errorf("unexpected sport %d", sport) + } + } +} + +// TestMultiCoalescerEpochOrdersAcrossRehandshake: a re-handshake replaces +// the tunnel, and the replacement's counter space starts near zero — raw +// counter order would emit the new tunnel's packets first while the old +// tunnel's backlog is still arriving. The epoch key must dominate: +// everything from the old tunnel emits before anything from the new one. +func TestMultiCoalescerEpochOrdersAcrossRehandshake(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + m := newTestMultiCoalescer(t, w) + pay := make([]byte, 1200) + + // New session's first data arrives before the old session's last data. + if err := m.Commit(buildTCPv4(2200, tcpAck, pay), SortKey{Epoch: 8, Counter: 1}); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildTCPv4(1000, tcpAck, pay), SortKey{Epoch: 7, Counter: 9_000_000}); err != nil { + t.Fatal(err) + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + // Same flow, contiguous seq, identical headers: after the epoch sort the + // two segments append into one superpacket seeded by the OLD session's + // packet. + if len(w.gsoWrites) != 1 { + t.Fatalf("want 1 gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + } + const ipHdrLen = 20 + if seedSeq := binary.BigEndian.Uint32(w.gsoWrites[0].hdr[ipHdrLen+4 : ipHdrLen+8]); seedSeq != 1000 { + t.Errorf("seed seq=%d want 1000 (old session first)", seedSeq) + } +} + // TestMultiCoalescerNoUSOFallsThrough verifies that on a queue without USO // (older kernel: TSO but no GSO_UDP_L4) the UDP lane never comes up and UDP // packets still reach the kernel via verbatim rather than being lost. func TestMultiCoalescerNoUSOFallsThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true, noUSO: true} m := newTestMultiCoalescer(t, w) + k := &keySeq{epoch: 1} if m.udp != nil { t.Fatal("UDP lane must not come up without USO") } - if err := m.Commit(buildUDPv4(1000, 53, make([]byte, 800))); err != nil { + if err := m.Commit(buildUDPv4(1000, 53, make([]byte, 800)), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildUDPv4(1000, 53, make([]byte, 800))); err != nil { + if err := m.Commit(buildUDPv4(1000, 53, make([]byte, 800)), k.next()); err != nil { t.Fatal(err) } if err := m.Flush(); err != nil { @@ -92,26 +244,24 @@ func TestMultiCoalescerNoUSOFallsThrough(t *testing.T) { } } -// TestMultiCoalescerNoOffloadsIsPassthrough covers a queue that can't offload -// anything. Both lane constructors refuse, so there's nothing left to -// dispatch between and NewMultiCoalescer hands back the verbatim lane -// itself — no wrapper, no per-packet protocol demux, and every packet reaches -// the kernel in arrival order. This is the case Interface.activate used to -// special-case with a bare Passthrough. -func TestMultiCoalescerNoOffloadsIsPassthrough(t *testing.T) { +// TestMultiCoalescerNoOffloadsStillSorts covers a queue that can't offload +// anything. Both lane constructors refuse, so every packet rides the +// verbatim lane — but the staging sort still applies, so emission follows +// transmission order even without GSO. +func TestMultiCoalescerNoOffloadsStillSorts(t *testing.T) { w := &fakeTunWriter{gsoEnabled: false} - m := NewMultiCoalescer(w, test.NewLogger()) - - if _, ok := m.(*Passthrough); !ok { - t.Fatalf("want a bare *Passthrough when neither offload is available, got %T", m) + m := newTestMultiCoalescer(t, w) + if m.tcp != nil || m.udp != nil { + t.Fatal("no lane may come up without offloads") } pkts := [][]byte{ buildTCPv4(1000, tcpAck, make([]byte, 1200)), buildUDPv4(1000, 53, make([]byte, 800)), buildTCPv4(2200, tcpAck, make([]byte, 1200)), } - for _, p := range pkts { - if err := m.Commit(p); err != nil { + // Committed in reverse transmission order; keys carry the truth. + for i := len(pkts) - 1; i >= 0; i-- { + if err := m.Commit(pkts[i], SortKey{Epoch: 1, Counter: uint64(i + 1)}); err != nil { t.Fatal(err) } } @@ -124,7 +274,7 @@ func TestMultiCoalescerNoOffloadsIsPassthrough(t *testing.T) { if len(w.writes) != len(pkts) { t.Fatalf("want %d plain writes, got %d", len(pkts), len(w.writes)) } - // One lane for everything means arrival order survives end to end. + // One lane for everything means the sorted order survives end to end. for i, want := range pkts { if !bytes.Equal(w.writes[i], want) { t.Errorf("write %d out of order or corrupt", i) @@ -173,14 +323,15 @@ func buildUDPv6Fragment(sport, dport uint16, payload []byte) []byte { func TestMultiCoalescerIPv6FragmentStaysInLane(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} m := newTestMultiCoalescer(t, w) + k := &keySeq{epoch: 1} - if err := m.Commit(buildUDPv6Fragment(2000, 53, make([]byte, 512))); err != nil { + if err := m.Commit(buildUDPv6Fragment(2000, 53, make([]byte, 512)), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800))); err != nil { + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800))); err != nil { + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { t.Fatal(err) } if err := m.Flush(); err != nil { @@ -192,25 +343,65 @@ func TestMultiCoalescerIPv6FragmentStaysInLane(t *testing.T) { if len(w.gsoWrites) != 1 { t.Fatalf("want the two whole datagrams coalesced into 1 gso write, got %d", len(w.gsoWrites)) } - // Arrival order was fragment-then-data; same-lane routing must keep it. + // Transmission order was fragment-then-data; same-lane routing must keep it. if w.order[0] != "write" { t.Fatalf("fragment must be emitted before later data (in-lane verbatim), order=%v", w.order) } } +// TestMultiCoalescerFragmentSealsUDPChains: an unparseable datagram +// (fragment) seals every open UDP chain, so datagrams from before and after +// it land in separate superpackets and the fragment holds its transmission- +// order position between them. +func TestMultiCoalescerFragmentSealsUDPChains(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + m := newTestMultiCoalescer(t, w) + k := &keySeq{epoch: 1} + + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildUDPv6Fragment(2000, 53, make([]byte, 512)), k.next()); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { + t.Fatal(err) + } + if err := m.Commit(buildUDPv6(2000, 53, make([]byte, 800)), k.next()); err != nil { + t.Fatal(err) + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 2 { + t.Fatalf("want 2 gso writes (chains sealed around the fragment), got %d", len(w.gsoWrites)) + } + if len(w.writes) != 1 { + t.Fatalf("want the fragment as 1 plain write, got %d", len(w.writes)) + } + want := []string{"gso", "write", "gso"} + if len(w.order) != 3 || w.order[0] != want[0] || w.order[1] != want[1] || w.order[2] != want[2] { + t.Fatalf("emission order = %v, want %v", w.order, want) + } +} + // TestMultiCoalescerNoTSOFallsThrough mirrors the no-TSO case. func TestMultiCoalescerNoTSOFallsThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true, noTSO: true} m := newTestMultiCoalescer(t, w) + k := &keySeq{epoch: 1} if m.tcp != nil { t.Fatal("TCP lane must not come up without TSO") } pay := make([]byte, 1200) - if err := m.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { + if err := m.Commit(buildTCPv4(1000, tcpAck, pay), k.next()); err != nil { t.Fatal(err) } - if err := m.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { + if err := m.Commit(buildTCPv4(2200, tcpAck, pay), k.next()); err != nil { t.Fatal(err) } if err := m.Flush(); err != nil { diff --git a/overlay/batch/passthrough.go b/overlay/batch/passthrough.go index 7eb77231..9d1c9308 100644 --- a/overlay/batch/passthrough.go +++ b/overlay/batch/passthrough.go @@ -17,7 +17,15 @@ func NewPassthrough(w io.Writer) *Passthrough { } } -func (p *Passthrough) Commit(pkt []byte) error { +// Commit ignores the sort key: a bare Passthrough (no MultiCoalescer in +// front) emits in arrival order, exactly as before keys existed. +func (p *Passthrough) Commit(pkt []byte, _ SortKey) error { + return p.enqueue(pkt) +} + +// enqueue is the lane-facing half of Commit: MultiCoalescer.dispatch hands +// packets here already sorted into transmission order. +func (p *Passthrough) enqueue(pkt []byte) error { p.slots = append(p.slots, pkt) return nil } diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index 2d277c5d..27a384df 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -7,7 +7,6 @@ import ( "io" "log/slog" "net/netip" - "slices" "github.com/slackhq/nebula/overlay/tio" ) @@ -28,24 +27,15 @@ const tcpCoalesceMaxSegs = 64 const tcpCoalesceHdrCap = 100 // coalesceSlot is one entry in the coalescer's ordered event queue. -// When verbatim is true the slot holds a single borrowed packet that must be -// emitted verbatim (non-TCP, non-admissible TCP, or oversize seed). +// When verbatim is true the slot holds a single borrowed packet that is +// emitted as-is (pure ACK, non-admissible TCP, unparseable, or oversize seed). // When verbatim is false the slot is an in-progress coalesced superpacket. // hdrBuf is a mutable copy of the seed's IP+TCP header // (we patch total length and pseudo-header partial at flush) // payIovs are *borrowed* slices from the caller's plaintext buffers. // The caller (listenOut) must keep those buffers alive until Flush. - -const ( - verbatimFalse = iota - // verbatimTrue means a sync-point packet, that may not be re-ordered - verbatimTrue - // verbatimACK packets are "passed through" without coalescing, but traffic "after" them may be pulled forward to facilitate coalescing. - verbatimACK -) - type coalesceSlot struct { - verbatim uint8 + verbatim bool // rawPkt is borrowed: the whole packet for verbatim slots, the seed // packet for coalesce slots. A coalesce slot that never grows past one // segment is emitted from rawPkt so its original (already valid) L4 @@ -61,26 +51,20 @@ type coalesceSlot struct { numSeg int totalPay int nextSeq uint32 - // tsVal is the TCP timestamp of the slot's seed segment (uniform across - // the slot: headersMatch requires byte-equal options for every append and - // merge). Sort key only, see compareCoalesceSlots. - tsVal uint32 - hasTS bool - // sealed marks the chain permanently closed: the last-accepted segment had PSH or was sub-gsoSize, - // so no append or flush-time merge may follow. - // Distinct from eviction out of openSlots (e.g. on seq mismatch), - // which leaves sealed=false so reorderForFlush can still merge the slot. + // sealed marks the chain permanently closed: the last-accepted segment + // had PSH or was sub-gsoSize, so no append may follow. Belt-and- + // suspenders with removal from openSlots, which is what actually stops + // the append paths from finding the slot. sealed bool payIovs [][]byte } -func (c *coalesceSlot) isVerbatim() bool { - return c.verbatim != verbatimFalse -} - // TCPCoalescer accumulates adjacent in-flow TCP data segments across multiple concurrent flows // and emits each flow's run as a single TSO superpacket via tio.GSOWriter. -// All output, coalesced or not, is deferred until Flush so arrival order is preserved on the wire. +// It expects its input in sender-transmission order (MultiCoalescer sorts the +// staged batch by (epoch, counter) before dispatching here) and emits slots in +// creation order, which therefore reproduces transmission order — modulo the +// pure-ACK allowance in commitParsed. // Owns no locks; one coalescer per TUN write queue. type TCPCoalescer struct { w tio.GSOWriter @@ -130,7 +114,6 @@ type parsedTCP struct { payLen int seq uint32 flags byte - options []byte } // parseTCPBase extracts the flow key and IP/TCP offsets for any TCP packet, @@ -163,10 +146,6 @@ func parseTCPBase(pkt []byte) (parsedTCP, bool) { p.fk.dport = binary.BigEndian.Uint16(pkt[p.ipHdrLen+2 : p.ipHdrLen+4]) p.seq = binary.BigEndian.Uint32(pkt[p.ipHdrLen+4 : p.ipHdrLen+8]) p.flags = pkt[p.ipHdrLen+13] - //window: 14, 15 - //csum: 16, 17 - //urg: 18, 19 - p.options = pkt[p.ipHdrLen+20 : p.ipHdrLen+p.tcpHdrLen : p.ipHdrLen+p.tcpHdrLen] return p, true } @@ -208,12 +187,24 @@ func (p parsedTCP) pureAck() bool { func (c *TCPCoalescer) Commit(pkt []byte) error { info, ok := parseTCPBase(pkt) if !ok { + // Unparseable shape: flow key unknowable, so seal every open chain to + // keep later data from extending a chain that would emit ahead of it. + c.sealAllOpen() c.addVerbatim(pkt) return nil } return c.commitParsed(pkt, info) } +// sealAllOpen closes every open coalesce chain: nothing committed after this +// call can extend a slot created before it. Called when an unparseable packet +// arrives — its flow is unknown, so any open chain might be the one whose +// later data would otherwise leapfrog it. +func (c *TCPCoalescer) sealAllOpen() { + clear(c.openSlots) + c.lastSlot = nil +} + // commitParsed is the post-parse half of Commit. The caller must have // already verified parseTCPBase succeeded (info is a valid TCP parse). // Used by MultiCoalescer.Commit to avoid re-walking the IP/TCP header @@ -222,18 +213,20 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { if !info.coalesceable() { if info.pureAck() { // A bare window/ack update carries no ordering obligation toward - // the flow's data: delivering it after later-arriving data only + // the flow's data: delivering it after later-transmitted data only // makes it a stale ACK, which receivers ignore. Skipping the // evict keeps a bidirectional flow's inbound data run coalescing // across the peer ACKs interleaved into it — kernel GRO likewise - // doesn't flush held data on a pure ACK. - c.addVerbatimACK(pkt, info) + // doesn't flush held data on a pure ACK. This is the one place + // emission can deviate from transmission order. + c.addVerbatim(pkt) return nil } // TCP but not admissible (SYN/FIN/RST/URG/CWR or a shape the flow // must observe in sequence). Seal this flow's open slot so later - // in-flow packets don't extend it and accidentally reorder past this - // verbatim. The len guard skips hashing the 38-byte key on + // in-flow packets don't extend it and emit ahead of this verbatim; + // with input in transmission order that pins the verbatim's exact + // in-flow position. The len guard skips hashing the 38-byte key on // ack-dominant queues, where the map is almost always empty. if len(c.openSlots) != 0 { if last := c.lastSlot; last != nil && last.fk == info.fk { @@ -267,8 +260,8 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { } return nil } - // Can't extend: evict it from openSlots and fall through to seed a fresh slot. - // The slot stays unsealed so reorderForFlush may still merge it. + // Can't extend (seq gap from upstream loss, header change, or a full + // chain): evict it from openSlots and fall through to seed a fresh slot. delete(c.openSlots, info.fk) if c.lastSlot == open { c.lastSlot = nil @@ -279,16 +272,18 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { } func (c *TCPCoalescer) Flush() error { - c.reorderForFlush() + if c.l.Enabled(context.Background(), slog.LevelDebug) { + c.logSeqGaps() + } var first error for _, s := range c.slots { var err error - if s.isVerbatim() || s.numSeg == 1 { + if s.verbatim || s.numSeg == 1 { // A slot that never grew (nor absorbed a merge) is byte-identical // to the packet it was seeded from; ship the original so its valid // checksum rides the DATA_VALID path instead of paying a kernel - // software csum. appendPayload and mergeSlots only touch hdrBuf - // once numSeg >= 2, so rawPkt is still pristine here. + // software csum. appendPayload only touches hdrBuf once + // numSeg >= 2, so rawPkt is still pristine here. _, err = c.w.Write(s.rawPkt) } else { err = c.flushSlot(s) @@ -308,27 +303,11 @@ func (c *TCPCoalescer) Flush() error { func (c *TCPCoalescer) addVerbatim(pkt []byte) { s := c.take() - s.verbatim = verbatimTrue + s.verbatim = true s.rawPkt = pkt c.slots = append(c.slots, s) } -// addVerbatimACK commits a pure ACK as a verbatim slot that keeps its -// flow identity and sort keys. Unlike addVerbatim slots it does not split -// sort runs, so reorderForFlush may sort same-flow data across it (the -// contract allows data to overtake a bare ACK). A pure ACK's seq is the -// sender's snd_nxt, which orders it after all data the peer sent before it, -// and the TSval-first comparator keeps it behind any older-timestamp data. -func (c *TCPCoalescer) addVerbatimACK(pkt []byte, info parsedTCP) { - s := c.take() - s.verbatim = verbatimACK - s.rawPkt = pkt - s.fk = info.fk - s.nextSeq = info.seq // totalPay stays 0, so slotSeedSeq yields info.seq - s.tsVal, _, s.hasTS = parseTCPOptions(info.options) - c.slots = append(c.slots, s) -} - func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { if info.hdrLen > tcpCoalesceHdrCap || info.hdrLen+info.payLen > tcpCoalesceBufSize { // Pathological shape. Can't fit our scratch, emit as-is. @@ -336,7 +315,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { return } s := c.take() - s.verbatim = verbatimFalse + s.verbatim = false s.rawPkt = pkt // kept for the numSeg==1 fast path in Flush copy(s.hdrBuf[:], pkt[:info.hdrLen]) s.hdrLen = info.hdrLen @@ -347,7 +326,6 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { s.numSeg = 1 s.totalPay = info.payLen s.nextSeq = info.seq + uint32(info.payLen) - s.tsVal, _, s.hasTS = parseTCPOptions(info.options) s.sealed = info.flags&tcpFlagPsh != 0 s.payIovs = append(s.payIovs[:0], pkt[info.hdrLen:info.hdrLen+info.payLen]) c.slots = append(c.slots, s) @@ -424,7 +402,7 @@ func (c *TCPCoalescer) take() *coalesceSlot { } func (c *TCPCoalescer) release(s *coalesceSlot) { - s.verbatim = verbatimFalse + s.verbatim = false s.rawPkt = nil clear(s.payIovs) s.payIovs = s.payIovs[:0] @@ -440,8 +418,6 @@ func (c *TCPCoalescer) release(s *coalesceSlot) { s.isV6 = false s.gsoSize = 0 s.nextSeq = 0 - s.tsVal = 0 - s.hasTS = false c.pool = append(c.pool, s) } @@ -500,79 +476,36 @@ func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { return true } -// reorderForFlush neutralizes wire-side reorder that the rxOrder buffer -// couldn't catch (anything crossing a recvmmsg batch boundary). -// Without this pass a small wire reorder, counter 250 arriving in batch K when -// 200..249 are coming in batch K+1, would seed an out-of-seq slot first -// and emit it ahead of the lower-seq slot, manifesting at the inner TCP -// receiver as a much larger reorder than the wire actually had. -// -// Two phases: -// 1. Sort each verbatim-bounded segment of c.slots by (flow, seq). -// Cross-flow ordering inside a segment isn't preserved (it never was -// and doesn't matter for any single flow's TCP correctness). -// 2. Sweep once and merge adjacent same-flow slots whose ranges are now -// contiguous AND whose tail is gsoSize-aligned. The tail constraint -// matters because the kernel TSO splitter chops at gsoSize from the -// start of the merged payload. A short segment in the middle would -// desynchronize every later segment. -// -// Verbatim slots act as barriers: the merge check skips them on either -// side, so a SYN/FIN/RST/CWR is never reordered relative to its flow's -// data. -func (c *TCPCoalescer) reorderForFlush() { - if len(c.slots) <= 1 { - return - } - runStart := 0 - for i := 0; i <= len(c.slots); i++ { - // Only hard verbatims (unparseable, SYN/FIN/RST/CWR, oversized) - // split sort runs. Pure-ACK verbatims stay inside the run so - // same-flow data separated by an interleaved ACK can still sort - // adjacent and merge; their own sort keys keep them ordered. - if i < len(c.slots) && c.slots[i].verbatim != verbatimTrue { +// logSeqGaps reports same-flow seq discontinuities between consecutively +// created data slots. Input arrives in transmission order (MultiCoalescer +// sorts by (epoch, counter) before dispatch), so a gap here is traffic this +// batch never contained: loss upstream of nebula, a reorder spanning a flush +// boundary (which no intra-batch mechanism can repair), or a retransmit +// (negative gap). Logged so the operator can quantify how often that happens. +// The caller gates on debug level, so the map only allocates when asked for. +func (c *TCPCoalescer) logSeqGaps() { + prevByFlow := make(map[flowKey]*coalesceSlot, len(c.slots)) + for _, s := range c.slots { + if s.verbatim { continue } - c.sortRun(c.slots[runStart:i]) - runStart = i + 1 - } - out := c.slots[:0] - for _, s := range c.slots { - if n := len(out); n > 0 { - prev := out[n-1] - if !prev.isVerbatim() && !s.isVerbatim() && prev.fk == s.fk { - // Same-flow neighbors after sort. If they aren't seq- - // contiguous it's a real gap: packets the wire reordered - // across batches, or actual loss before nebula. Log it so - // the operator can quantify how often it happens - if c.l.Enabled(context.Background(), slog.LevelDebug) { - if prev.nextSeq != slotSeedSeq(s) { - gap := int64(slotSeedSeq(s)) - int64(prev.nextSeq) - c.l.Debug("tcp coalesce: cross-slot seq gap", - "src", flowKeyAddr(s.fk, false), - "dst", flowKeyAddr(s.fk, true), - "sport", s.fk.sport, - "dport", s.fk.dport, - "prev_seed_seq", slotSeedSeq(prev), - "prev_next_seq", prev.nextSeq, - "this_seed_seq", slotSeedSeq(s), - "gap_bytes", gap, - "prev_seg_count", prev.numSeg, - "prev_total_pay", prev.totalPay, - ) - } - } - - if canMergeSlots(prev, s) { - mergeSlots(prev, s) - c.release(s) - continue - } - } + if prev, ok := prevByFlow[s.fk]; ok && prev.nextSeq != slotSeedSeq(s) { + gap := int64(slotSeedSeq(s)) - int64(prev.nextSeq) + c.l.Debug("tcp coalesce: cross-slot seq gap", + "src", flowKeyAddr(s.fk, false), + "dst", flowKeyAddr(s.fk, true), + "sport", s.fk.sport, + "dport", s.fk.dport, + "prev_seed_seq", slotSeedSeq(prev), + "prev_next_seq", prev.nextSeq, + "this_seed_seq", slotSeedSeq(s), + "gap_bytes", gap, + "prev_seg_count", prev.numSeg, + "prev_total_pay", prev.totalPay, + ) } - out = append(out, s) + prevByFlow[s.fk] = s } - c.slots = out } // flowKeyAddr returns the src or dst address from fk as a netip.Addr for @@ -591,87 +524,6 @@ func flowKeyAddr(fk flowKey, dst bool) netip.Addr { return netip.AddrFrom4(v4) } -// sortRun stable-sorts run by (flowKey, seedSeq) so each flow's slots -// cluster together in seq order, ready for the merge sweep. Stable so -// equal-key slots keep their original relative position (defensive — a -// duplicate seedSeq would already mean something's wrong upstream). -func (c *TCPCoalescer) sortRun(run []*coalesceSlot) { - if len(run) <= 1 { - return - } - // slices.SortStableFunc with a free, non-capturing comparator avoids the - // reflection + closure-escape allocations that sort.SliceStable forces. - slices.SortStableFunc(run, compareCoalesceSlots) -} - -func compareCoalesceSlots(a, b *coalesceSlot) int { - if cmp := flowKeyCompare(a.fk, b.fk); cmp != 0 { - return cmp - } - // A retransmit carries a lower seq but a newer TCP timestamp than - // in-flight original data. Emitting it first would advance the - // receiver's ts_recent past the original's TSval, and PAWS would then - // drop the original as an old duplicate. So order by TSval before seq: - // TSval order approximates transmission order (which wire reordering - // never changed), and slots whose TSvals tie still get seq-repaired below. - // Flows without timestamps fall through to pure seq order, where PAWS cannot apply. - // tcpSeqLess is reused for the TSval compare: RFC 7323 defines TSval - // comparison in the same serial-number arithmetic. - if a.hasTS && b.hasTS && a.tsVal != b.tsVal { - if tcpSeqLess(a.tsVal, b.tsVal) { - return -1 - } - return 1 - } - - aSeq, bSeq := slotSeedSeq(a), slotSeedSeq(b) - if aSeq == bSeq { - return 0 - } - if tcpSeqLess(aSeq, bSeq) { - return -1 - } - return 1 -} - -// parseTCPOptions attempts to locate timestamps. If it finds them, it returns tsval, secr, true. 0,0,false otherwise. -func parseTCPOptions(opts []byte) (uint32, uint32, bool) { - const timeStampOptionSize = 1 + 1 + 4 + 4 - const timeStampOptionCode = 0x8 - const nopOptionCode = 0x1 - const eolOptionCode = 0x0 - // Inclusive bound: a timestamp ending exactly at len(opts) is the common - // case (Linux emits NOP,NOP,TS as the whole option block). It also - // guards opts[i+1] in every arm, since timeStampOptionSize >= 2. - for i := 0; i+timeStampOptionSize <= len(opts); /* no increment */ { - switch opts[i] { - case eolOptionCode: - // End-of-option-list: everything after is padding. - return 0, 0, false - case nopOptionCode: - i++ - case timeStampOptionCode: - // we found it! - length := opts[i+1] - if length != timeStampOptionSize { - return 0, 0, false //weird, wrong option? - } - tsval := binary.BigEndian.Uint32(opts[i+2 : i+2+4]) - secr := binary.BigEndian.Uint32(opts[i+2+4 : i+2+4+4]) - return tsval, secr, true - default: - length := int(opts[i+1]) - if length < 2 { - // Malformed: a non-NOP option shorter than its own - // kind+length bytes would loop forever. - return 0, 0, false - } - i += length - } - } - return 0, 0, false -} - // slotSeedSeq returns the TCP seq of the slot's seed (first segment). // nextSeq tracks the seq just past the last appended byte; subtracting // totalPay walks back to the seed. uint32 wraparound is the right TCP @@ -680,114 +532,6 @@ func slotSeedSeq(s *coalesceSlot) uint32 { return s.nextSeq - uint32(s.totalPay) } -// tcpSeqLess reports whether a precedes b in TCP serial-number arithmetic -// (RFC 1323 §2.3). The signed int32 cast turns the modular subtraction -// into the right comparison even across the 2^32 wrap. -func tcpSeqLess(a, b uint32) bool { - return int32(a-b) < 0 -} - -// flowKeyCompare orders flowKeys deterministically. The exact ordering -// is irrelevant — only that same-flow slots cluster together so the -// post-sort sweep can merge contiguous pairs. -func flowKeyCompare(a, b flowKey) int { - // Cheap scalar fields first so most non-matching keys short-circuit - // without ever calling bytes.Compare. sport is the ephemeral port on - // egress flows and discriminates fastest. For matching keys (same - // flow), array equality on src/dst inlines to word-sized compares, - // so we only pay bytes.Compare when the arrays actually differ. - if a.sport != b.sport { - if a.sport < b.sport { - return -1 - } - return 1 - } - if a.dport != b.dport { - if a.dport < b.dport { - return -1 - } - return 1 - } - if a.dst != b.dst { - return bytes.Compare(a.dst[:], b.dst[:]) - } - if a.src != b.src { - return bytes.Compare(a.src[:], b.src[:]) - } - if a.isV6 != b.isV6 { - if !a.isV6 { - return -1 - } - return 1 - } - return 0 -} - -// canMergeSlots reports whether s can fold into prev as one merged TSO -// superpacket. Same flow, contiguous TCP byte range, equal gsoSize, and -// fits within the kernel TSO limits. The tail-of-prev check rejects any -// merge whose first slot ended on a sub-gsoSize segment — kernel TSO -// would split the merged skb at gsoSize boundaries from the start, so a -// short segment in the middle would corrupt every later segment. PSH and -// ECE state must agree across both slots: PSH is a semantic delimiter -// (preserving the sender's push boundary) and ECE state must be uniform -// across a window (the same rule canAppend enforces for in-flow appends). -// The IP-level ECN codepoint must also match: this check calls headersMatch -// → ipHeadersMatch, which compares the full DSCP/ECN byte, so two slots with -// differing ECN marks stay separate superpackets, each keeping its own mark. -// -// Note: a slot evicted on reorder (canAppend returned false on seq mismatch) -// stays sealed=false, so this restriction does not block the reorder-fix merge, -// only chains ended by PSH or a short tail. -func canMergeSlots(prev, s *coalesceSlot) bool { - if prev.sealed { - return false - } - if prev.fk != s.fk { - return false - } - if prev.gsoSize != s.gsoSize { - return false - } - if prev.nextSeq != slotSeedSeq(s) { - return false - } - if prev.numSeg+s.numSeg > tcpCoalesceMaxSegs { - return false - } - if prev.hdrLen+prev.totalPay+s.totalPay > tcpCoalesceBufSize { - return false - } - if len(prev.payIovs[len(prev.payIovs)-1]) != prev.gsoSize { - return false - } - prevFlags := prev.hdrBuf[prev.ipHdrLen+13] - sFlags := s.hdrBuf[s.ipHdrLen+13] - if (prevFlags^sFlags)&tcpFlagEce != 0 { - return false - } - if !prev.isV6 && !ipv4CanCoalesceID(prev.hdrBuf[:], s.hdrBuf[:], prev.numSeg) { - return false - } - if !headersMatch(prev.hdrBuf[:prev.hdrLen], s.hdrBuf[:s.hdrLen], prev.isV6, prev.ipHdrLen) { - return false - } - return true -} - -// mergeSlots folds src into dst in place: payIovs concatenated, counters -// and totals updated. The seed header's seq, gsoSize, and fk are unchanged. -// The caller must release src (it's no longer in c.slots after this call). -func mergeSlots(dst, src *coalesceSlot) { - dst.payIovs = append(dst.payIovs, src.payIovs...) - dst.numSeg += src.numSeg - dst.totalPay += src.totalPay - dst.nextSeq = src.nextSeq - dst.sealed = src.sealed // dst is open — canMergeSlots rejects a sealed prev - // carry PSH through - dst.hdrBuf[dst.ipHdrLen+13] |= src.hdrBuf[src.ipHdrLen+13] & tcpFlagPsh -} - // ipv4HdrChecksum computes the IPv4 header checksum over hdr (which must // already have its checksum field zeroed) and returns the folded/inverted // 16-bit value to store. diff --git a/overlay/batch/tcp_coalesce_bench_test.go b/overlay/batch/tcp_coalesce_bench_test.go index 9979566b..da78fbf6 100644 --- a/overlay/batch/tcp_coalesce_bench_test.go +++ b/overlay/batch/tcp_coalesce_bench_test.go @@ -168,9 +168,9 @@ func BenchmarkCommitNonCoalesceableTCP(b *testing.B) { runCommitBench(b, pkts, 64) } -// runMultiCommitBench drives MultiCoalescer.Commit. The dispatcher does -// the IP/L4 parse once and passes the parsed struct to the lane, so this -// is the bench that shows the savings of skipping the lane's re-parse. +// runMultiCommitBench drives MultiCoalescer.Commit with in-order keys, so +// it includes the staging sort's already-sorted fast path plus the +// dispatch-time parse — the full steady-state cost of the batcher. func runMultiCommitBench(b *testing.B, pkts [][]byte, batchSize int) { b.Helper() m := NewMultiCoalescer(nopTunWriter{}, test.NewLogger()) @@ -179,7 +179,7 @@ func runMultiCommitBench(b *testing.B, pkts [][]byte, batchSize int) { b.ResetTimer() for i := 0; i < b.N; i++ { pkt := pkts[i%len(pkts)] - if err := m.Commit(pkt); err != nil { + if err := m.Commit(pkt, SortKey{Epoch: 1, Counter: uint64(i + 1)}); err != nil { b.Fatal(err) } if (i+1)%batchSize == 0 { diff --git a/overlay/batch/tcp_coalesce_test.go b/overlay/batch/tcp_coalesce_test.go index c77330f7..ff8559b6 100644 --- a/overlay/batch/tcp_coalesce_test.go +++ b/overlay/batch/tcp_coalesce_test.go @@ -994,108 +994,17 @@ func TestCoalescerIPv6CoalescesEceFlow(t *testing.T) { } } -// TestCoalescerSortsReorderedSeedsAndMerges feeds three same-flow MSS -// segments out of TCP-seq order (mimicking a wire reorder that escaped -// the rxOrder per-batch sort). Without the reorderForFlush sort+merge, -// each out-of-seq arrival would seed its own slot and the slots would -// emit in arrival order, producing a kernel-visible TCP reorder. With -// the sort+merge, the three slots are sorted by seq and folded back into -// one in-order TSO superpacket — same shape the receiver TCP would have -// seen had the wire never reordered. -func TestCoalescerSortsReorderedSeedsAndMerges(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - pay := make([]byte, 1200) - // Arrival order: seq 1000, 3400, 2200. The 3400 seeds a separate slot - // because 3400 != nextSeq=2200, then 2200 fails to extend the 3400 slot - // and seeds its own. Three slots end up in c.slots; reorderForFlush - // should sort them into [1000,2200,3400] and merge them back into one. - if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write got %d", len(w.gsoWrites)) - } - g := w.gsoWrites[0] - if len(g.pays) != 3 { - t.Fatalf("merged segs=%d want 3", len(g.pays)) - } - const ipHdrLen = 20 - if seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]); seedSeq != 1000 { - t.Errorf("merged seed seq=%d want 1000 (lowest)", seedSeq) - } -} - -// TestCoalescerSortAcrossFlowsMergesEachIndependently checks that two -// flows interleaved with reorder are each sorted-and-merged in isolation -// without any cross-flow contamination. -func TestCoalescerSortAcrossFlowsMergesEachIndependently(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - pay := make([]byte, 1200) - // Flow A (sport 1000) seq 100, 1300; flow B (sport 3000) seq 500, 1700. - // Arrival: A.1300, B.1700, A.100, B.500 — every flow reordered. - if err := c.Commit(buildTCPv4Ports(1000, 2000, 1300, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4Ports(3000, 2000, 1700, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.gsoWrites) != 2 { - t.Fatalf("want 2 gso writes (one per flow merged), got %d", len(w.gsoWrites)) - } - for i, g := range w.gsoWrites { - if len(g.pays) != 2 { - t.Errorf("gso[%d] segs=%d want 2", i, len(g.pays)) - } - const ipHdrLen = 20 - seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]) - sport := binary.BigEndian.Uint16(g.hdr[ipHdrLen : ipHdrLen+2]) - // Each flow's merged seed should be the LOWER of its two seqs. - switch sport { - case 1000: - if seedSeq != 100 { - t.Errorf("flow A seed seq=%d want 100", seedSeq) - } - case 3000: - if seedSeq != 500 { - t.Errorf("flow B seed seq=%d want 500", seedSeq) - } - default: - t.Errorf("unexpected sport %d", sport) - } - } -} - -// TestCoalescerSortKeepsPSHBoundary verifies that a PSH-sealed slot is -// not folded into a later seq-contiguous slot — PSH placement is part of -// the wire signal and merging across it would shift the receiver's push -// boundary by an arbitrary number of segments. -func TestCoalescerSortKeepsPSHBoundary(t *testing.T) { +// TestCoalescerPSHKeepsChainBoundary verifies that a PSH-sealed chain is +// not extended by a later seq-contiguous segment — PSH placement is part of +// the wire signal and growing the superpacket past it would shift the +// receiver's push boundary by an arbitrary number of segments. +func TestCoalescerPSHKeepsChainBoundary(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Seq 1000 (no PSH) + 2200 (PSH) → seal one slot with PSH set. - // Seq 3400 (no PSH) is contiguous to 3400 from seq 2200+1200; without - // the PSH check it would merge in. + // Seq 3400 is contiguous to the sealed chain's nextSeq; without the + // seal check it would append in. if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) } @@ -1115,38 +1024,36 @@ func TestCoalescerSortKeepsPSHBoundary(t *testing.T) { } } -// TestCoalescerSortKeepsPassthroughBarrier confirms a verbatim slot in -// the middle of the queue prevents the post-sort merge from folding -// across it. Reordered same-flow data on either side of the verbatim -// is sorted/merged independently. -func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) { +// TestCoalescerSynSealsFlowChain confirms a non-admissible in-flow packet +// (SYN+ACK here) seals its flow's open chain and holds its emission +// position: data committed after it seeds a fresh slot and emits after it, +// never extending a chain created before it. +func TestCoalescerSynSealsFlowChain(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) - // First two segments seed S1 (then a 3400 reorder seeds S2). if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) } + // Discontiguous seq: evicts the 1000 slot and seeds its own. if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { t.Fatal(err) } - // Non-coalesceable packet (SYN+ACK) flushes S1's openSlots entry and - // becomes a verbatim barrier in c.slots. + // Non-coalesceable packet (SYN+ACK) seals the flow's open slot and + // becomes a verbatim slot in c.slots. if err := c.Commit(buildTCPv4(9999, tcpSyn|tcpAck, pay)); err != nil { t.Fatal(err) } - // Post-barrier same-flow data: should never end up before the SYN. + // Post-SYN data: must emit after the SYN, in its own slot. if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { t.Fatal(err) } if err := c.Flush(); err != nil { t.Fatal(err) } - // All four packets emit as plain writes: 1000 and 3400 are separate - // single-segment slots (not contiguous, so the post-sort merge can't - // fold them), the SYN is verbatim, and the post-barrier 2200 stays - // a single-segment slot after the SYN. The pre-barrier sort must land - // 1000 before 3400, and 2200 must never move before the SYN. + // All four packets emit as plain writes in creation order: 1000 and + // 3400 are separate single-segment slots, the SYN is verbatim, and the + // post-SYN 2200 is a fresh single-segment slot after it. if len(w.writes) != 4 || len(w.gsoWrites) != 0 { t.Fatalf("want 4 plain writes, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) } @@ -1204,150 +1111,6 @@ func TestCoalescerIPv6DifferingECNReseeds(t *testing.T) { } } -func TestSortRunZeroAllocs(t *testing.T) { - c := &TCPCoalescer{} - mk := func(srcByte byte, seq uint32, pay int) *coalesceSlot { - s := &coalesceSlot{nextSeq: seq + uint32(pay), totalPay: pay} - s.fk.src[0] = srcByte - return s - } - run := []*coalesceSlot{ - mk(3, 5000, 100), - mk(1, 1000, 50), - mk(2, 2000, 75), - mk(1, 900, 50), - mk(3, 4900, 100), - mk(2, 1925, 75), - mk(1, 1050, 50), - mk(3, 5100, 100), - } - - allocs := testing.AllocsPerRun(100, func() { - // Re-shuffle so each run actually does sorting work. - run[0], run[1], run[2], run[3] = run[3], run[2], run[1], run[0] - c.sortRun(run) - }) - if allocs != 0 { - t.Fatalf("sortRun allocates %v times per run; want 0", allocs) - } -} - -// TestCoalescerMergeShortTailDoesNotFabricatePSH: a slot sealed by a -// sub-gsoSize tail segment has psh=true in the chain-closed sense but no -// PSH flag on any of its packets. When reorderForFlush folds it into the -// preceding slot, the merged header must not grow a PSH the sender never -// sent — mergeSlots must copy the wire flag from the source header, not -// synthesize it from the seal bool. -func TestCoalescerMergeShortTailDoesNotFabricatePSH(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - pay := make([]byte, 1200) - short := make([]byte, 600) - // Arrival: seq 3400 (full), 4600 (short, seals the slot), then the - // reordered front of the window: 1000, 2200. Flush sorts the two slots - // into [1000..3400) + [3400..5200) and merges them. - if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(4600, tcpAck, short)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) - } - g := w.gsoWrites[0] - if len(g.pays) != 4 { - t.Fatalf("merged segs=%d want 4", len(g.pays)) - } - const ipHdrLen = 20 - if flags := g.hdr[ipHdrLen+13]; flags&tcpPsh != 0 { - t.Errorf("merged header flags=%#x: PSH fabricated by short-tail merge", flags) - } -} - -// TestCoalescerMergePreservesRealPSH is the positive companion: when the -// source slot's tail really carried PSH, the merged header must keep it. -func TestCoalescerMergePreservesRealPSH(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - pay := make([]byte, 1200) - short := make([]byte, 600) - if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(4600, tcpAckPsh, short)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { - t.Fatal(err) - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) - } - g := w.gsoWrites[0] - if len(g.pays) != 4 { - t.Fatalf("merged segs=%d want 4", len(g.pays)) - } - const ipHdrLen = 20 - if flags := g.hdr[ipHdrLen+13]; flags&tcpPsh == 0 { - t.Errorf("merged header flags=%#x: real PSH lost in merge", flags) - } -} - -// TestCoalescerSeqWrapAroundSortsAndMerges pins the serial-number -// arithmetic through the sort-and-merge path: a chain that crosses the -// 2^32 seq wrap must still sort pre-wrap before post-wrap and merge into -// one superpacket when contiguous. -func TestCoalescerSeqWrapAroundSortsAndMerges(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - - payA := bytes.Repeat([]byte{'A'}, 32) - payB := bytes.Repeat([]byte{'B'}, 32) - seqA := uint32(0xffffffe0) // 32 before the wrap: nextSeq lands exactly on 0 - - // The post-wrap segment arrives first — wire reorder across a batch - // boundary, the case reorderForFlush exists for. - if err := c.Commit(buildTCPv4(0, tcpAck, payB)); err != nil { - t.Fatal(err) - } - if err := c.Commit(buildTCPv4(seqA, tcpAck, payA)); err != nil { - t.Fatal(err) - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write across the wrap, got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) - } - g := w.gsoWrites[0] - const ipHdrLen = 20 - if seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]); seedSeq != seqA { - t.Errorf("merged seed seq=%#x want %#x (pre-wrap segment first)", seedSeq, seqA) - } - if len(g.pays) != 2 { - t.Fatalf("merged segs=%d want 2", len(g.pays)) - } - if !bytes.Equal(g.pays[0], payA) || !bytes.Equal(g.pays[1], payB) { - t.Errorf("payload order wrong across the wrap: got %q then %q", g.pays[0][:1], g.pays[1][:1]) - } -} - // TestCoalescerNonAtomicSequentialIDsCoalesce: with DF clear, coalescing // is allowed when the IPv4 IDs already run seed+1 per segment — kernel // TSO's re-stamp then reproduces the originals exactly (the kernel GRO @@ -1438,178 +1201,87 @@ func TestCoalescerAtomicRandomIDsCoalesce(t *testing.T) { } } -// buildTCPv4TS is buildTCPv4 with a TCP timestamp option in the standard -// Linux layout (NOP,NOP,TS — a 32-byte TCP header). -func buildTCPv4TS(seq uint32, flags byte, tsVal, tsEcr uint32, payload []byte) []byte { - const ipHdrLen = 20 - const tcpHdrLen = 32 - total := ipHdrLen + tcpHdrLen + len(payload) - pkt := make([]byte, total) - - pkt[0] = 0x45 - pkt[1] = 0x00 - binary.BigEndian.PutUint16(pkt[2:4], uint16(total)) - binary.BigEndian.PutUint16(pkt[4:6], 0) - binary.BigEndian.PutUint16(pkt[6:8], 0x4000) - pkt[8] = 64 - pkt[9] = ipProtoTCP - copy(pkt[12:16], []byte{10, 0, 0, 1}) - copy(pkt[16:20], []byte{10, 0, 0, 2}) - - binary.BigEndian.PutUint16(pkt[20:22], 1000) - binary.BigEndian.PutUint16(pkt[22:24], 2000) - binary.BigEndian.PutUint32(pkt[24:28], seq) - binary.BigEndian.PutUint32(pkt[28:32], 12345) - pkt[32] = 0x80 // doff=8: 32-byte TCP header - pkt[33] = flags - binary.BigEndian.PutUint16(pkt[34:36], 0xffff) - pkt[40] = 0x01 // NOP - pkt[41] = 0x01 // NOP - pkt[42] = 0x08 // TS kind - pkt[43] = 10 // TS length - binary.BigEndian.PutUint32(pkt[44:48], tsVal) - binary.BigEndian.PutUint32(pkt[48:52], tsEcr) - - copy(pkt[52:], payload) - return pkt -} - -func TestParseTCPOptions(t *testing.T) { - ts := func(val, ecr uint32) []byte { - b := make([]byte, 10) - b[0], b[1] = 0x08, 10 - binary.BigEndian.PutUint32(b[2:6], val) - binary.BigEndian.PutUint32(b[6:10], ecr) - return b - } - cases := []struct { - name string - opts []byte - wantVal uint32 - wantEcr uint32 - wantOK bool - }{ - {"empty", nil, 0, 0, false}, - {"bare TS filling the block exactly", ts(100, 200), 100, 200, true}, - {"standard linux NOP,NOP,TS", append([]byte{1, 1}, ts(7, 9)...), 7, 9, true}, - {"unknown option then TS", append([]byte{254, 4, 0, 0}, ts(3, 4)...), 3, 4, true}, - {"EOL terminates before garbage", append([]byte{0, 0}, ts(1, 2)...), 0, 0, false}, - {"zero-length option must not hang", []byte{254, 0, 8, 10, 0, 0, 0, 1, 0, 0, 0, 2}, 0, 0, false}, - {"TS with wrong length", []byte{8, 4, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1}, 0, 0, false}, - {"truncated TS", append([]byte{1, 1, 1}, ts(5, 6)[:9]...), 0, 0, false}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - val, ecr, ok := parseTCPOptions(tc.opts) - if val != tc.wantVal || ecr != tc.wantEcr || ok != tc.wantOK { - t.Fatalf("parseTCPOptions(%v) = (%d, %d, %v), want (%d, %d, %v)", - tc.opts, val, ecr, ok, tc.wantVal, tc.wantEcr, tc.wantOK) - } - }) - } -} - -// TestCompareCoalesceSlotsAntisymmetric pins the comparator contract for the -// retransmit shape: a lower seq with a newer TSval (retransmit) versus a -// higher seq with an older TSval (delayed original). The TSval must win in -// BOTH directions — an asymmetric comparator gives SortStableFunc an -// inconsistent order and unspecified output. -func TestCompareCoalesceSlotsAntisymmetric(t *testing.T) { - mk := func(seq, tsVal uint32, hasTS bool) *coalesceSlot { - return &coalesceSlot{nextSeq: seq, tsVal: tsVal, hasTS: hasTS} - } - original := mk(5000, 100, true) // sent first, delayed in flight - retransmit := mk(1000, 105, true) // sent later, lower seq - - if got := compareCoalesceSlots(original, retransmit); got != -1 { - t.Fatalf("compare(original, retransmit) = %d, want -1 (older TSval first)", got) - } - if got := compareCoalesceSlots(retransmit, original); got != 1 { - t.Fatalf("compare(retransmit, original) = %d, want 1", got) - } - - // Equal TSvals (a burst within one tick) fall back to seq order, - // still antisymmetrically. - a, b := mk(1000, 50, true), mk(2000, 50, true) - if compareCoalesceSlots(a, b) != -1 || compareCoalesceSlots(b, a) != 1 { - t.Fatal("equal-TSval slots must order by seq in both directions") - } - - // Timestamp-less flows keep pure seq order. - c, d := mk(2000, 0, false), mk(1000, 99, true) - if compareCoalesceSlots(c, d) != 1 || compareCoalesceSlots(d, c) != -1 { - t.Fatal("mixed/absent timestamps must fall back to seq in both directions") - } -} - -// TestCoalescerRetransmitEmitsAfterDelayedOriginal: a retransmit (lower seq, -// newer TSval) and a delayed original (higher seq, older TSval) land in one -// flush window. Seq-only sorting would emit the retransmit first; the -// receiver would advance ts_recent past the original's TSval and PAWS would -// drop the original. TSval-first ordering must emit the original first. -func TestCoalescerRetransmitEmitsAfterDelayedOriginal(t *testing.T) { +// TestCoalescerSeqWrapAroundAppends pins the serial-number arithmetic on the +// append path: a chain crossing the 2^32 seq wrap must keep extending when +// contiguous. +func TestCoalescerSeqWrapAroundAppends(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := newTestTCPCoalescer(t, w) - pay := make([]byte, 100) - original := buildTCPv4TS(5000, tcpAck, 100, 1, pay) - retransmit := buildTCPv4TS(1000, tcpAck, 105, 1, pay) + payA := bytes.Repeat([]byte{'A'}, 32) + payB := bytes.Repeat([]byte{'B'}, 32) + seqA := uint32(0xffffffe0) // 32 before the wrap: nextSeq lands exactly on 0 - if err := c.Commit(original); err != nil { + if err := c.Commit(buildTCPv4(seqA, tcpAck, payA)); err != nil { t.Fatal(err) } - if err := c.Commit(retransmit); err != nil { + if err := c.Commit(buildTCPv4(0, tcpAck, payB)); err != nil { t.Fatal(err) } if err := c.Flush(); err != nil { t.Fatal(err) } - if len(w.writes) != 2 { - t.Fatalf("want 2 plain writes (non-contiguous single-segment slots), got %d writes, %d gso", len(w.writes), len(w.gsoWrites)) - } - firstSeq := binary.BigEndian.Uint32(w.writes[0][24:28]) - secondSeq := binary.BigEndian.Uint32(w.writes[1][24:28]) - if firstSeq != 5000 || secondSeq != 1000 { - t.Fatalf("emission order (%d, %d), want (5000, 1000): retransmit must not overtake the older-TSval original", firstSeq, secondSeq) - } -} - -// TestCoalescerACKDoesNotSplitSortRun: an interleaved pure ACK must not stop -// wire-reordered same-flow data on either side of it from sorting adjacent -// and merging — the contract explicitly allows data to overtake a bare ACK. -// Arrival is D2, ACK, D1; the two data slots must still merge into one -// superpacket, with the ACK emitted after (its seq is the peer's snd_nxt, -// which orders it behind the data it followed). -func TestCoalescerACKDoesNotSplitSortRun(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - c := newTestTCPCoalescer(t, w) - pay := make([]byte, 1200) - - d2 := buildTCPv4(2200, tcpAck, pay) - ack := buildTCPv4(3400, tcpAck, nil) - d1 := buildTCPv4(1000, tcpAck, pay) - - for _, pkt := range [][]byte{d2, ack, d1} { - if err := c.Commit(pkt); err != nil { - t.Fatal(err) - } - } - if err := c.Flush(); err != nil { - t.Fatal(err) - } if len(w.gsoWrites) != 1 { - t.Fatalf("want the two data slots merged into 1 gso write across the ACK, got %d gso + %d plain", len(w.gsoWrites), len(w.writes)) + t.Fatalf("want 1 gso write across the wrap, got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) } - if got := w.gsoWrites[0].payLen(); got != 2400 { - t.Fatalf("merged payload = %d, want 2400", got) + g := w.gsoWrites[0] + const ipHdrLen = 20 + if seedSeq := binary.BigEndian.Uint32(g.hdr[ipHdrLen+4 : ipHdrLen+8]); seedSeq != seqA { + t.Errorf("seed seq=%#x want %#x", seedSeq, seqA) } - if len(w.writes) != 1 { - t.Fatalf("want the ACK as 1 plain write, got %d", len(w.writes)) + if len(g.pays) != 2 { + t.Fatalf("segs=%d want 2", len(g.pays)) } - if got := binary.BigEndian.Uint32(w.writes[0][24:28]); got != 3400 { - t.Fatalf("plain write seq = %d, want the ACK (3400)", got) - } - if len(w.order) != 2 || w.order[0] != "gso" || w.order[1] != "write" { - t.Fatalf("emission order = %v, want [gso write]", w.order) + if !bytes.Equal(g.pays[0], payA) || !bytes.Equal(g.pays[1], payB) { + t.Errorf("payload order wrong across the wrap: got %q then %q", g.pays[0][:1], g.pays[1][:1]) + } +} + +// TestCoalescerUnparseableSealsAllChains: an unparseable packet's flow is +// unknowable, so it must close every open chain. Later data — even data +// seq-contiguous with a pre-existing chain — seeds a fresh slot and emits +// after the unparseable packet, exactly as transmitted. +func TestCoalescerUnparseableSealsAllChains(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := newTestTCPCoalescer(t, w) + pay := make([]byte, 1200) + + if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { + t.Fatal(err) + } + // IHL=6 fakes IP options: parseTCPBase bails, flow key unknown. + opts := buildTCPv4(5000, tcpAck, make([]byte, 500)) + opts[0] = 0x46 + if err := c.Commit(opts); err != nil { + t.Fatal(err) + } + // Contiguous with the first chain (nextSeq 3400), but that chain is + // sealed now: must not append, must not emit before the unparseable. + if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 1 { + t.Fatalf("want 1 gso write (pre-fragment pair), got %d", len(w.gsoWrites)) + } + if len(w.gsoWrites[0].pays) != 2 { + t.Fatalf("pre-fragment chain segs=%d want 2", len(w.gsoWrites[0].pays)) + } + if len(w.writes) != 2 { + t.Fatalf("want 2 plain writes (unparseable + post-fragment seed), got %d", len(w.writes)) + } + if w.writes[0][0] != 0x46 { + t.Errorf("first plain write must be the unparseable packet") + } + if seq := binary.BigEndian.Uint32(w.writes[1][24:28]); seq != 3400 { + t.Errorf("post-fragment data seq=%d want 3400", seq) + } + if len(w.order) != 3 || w.order[0] != "gso" || w.order[1] != "write" || w.order[2] != "write" { + t.Fatalf("emission order = %v, want [gso write write]", w.order) } } diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index ba347807..de5cbf58 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -195,6 +195,15 @@ func (c *UDPCoalescer) Flush() error { return first } +// sealAllOpen closes every open coalesce chain: nothing committed after this +// call can extend a slot created before it. Called when an unparseable packet +// arrives — its flow is unknown, so any open chain might be the one whose +// later data would otherwise leapfrog it. +func (c *UDPCoalescer) sealAllOpen() { + clear(c.openSlots) + c.lastSlot = nil +} + func (c *UDPCoalescer) addVerbatim(pkt []byte) { s := c.take() s.verbatim = true