diff --git a/errors.go b/errors.go index 284d15c..22ec0f1 100644 --- a/errors.go +++ b/errors.go @@ -24,6 +24,7 @@ const ( ErrAlreadyRegistered // protocol already registered ErrTruncatedFrame // truncated frame ErrMissingHALConfig // missing HAL configuration + ErrBadState // operation invalid in current state // Below are potentially good future error additions // based on one or two encountered use cases, example use case included. /* diff --git a/internal/ring.go b/internal/ring.go index 94b2ec9..b19b54b 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -13,6 +13,7 @@ import ( var ( ErrRingBufferFull = lneto.ErrBufferFull errRingNoData = errors.New("lneto/ring: empty write") + errInvalidCommit = errors.New("lneto/ring: invalid commit amount") errInvalidDiscard = errors.New("lneto/ring: invalid discard amount") errDiscardExceeds = errors.New("lneto/ring: discard exceeds length") errOffsetOverflow = errors.New("lneto/ring: offset too large (32 bit overflow)") @@ -92,6 +93,58 @@ func (r *Ring) Write(b []byte) (int, error) { return n, nil } +// writeStart returns the buffer index where the next [Ring.Write] or +// [Ring.Commit] begins, matching [Ring.Write]'s placement (including wrap). +func (r *Ring) writeStart() int { + if r.End == 0 { + return r.Off // Empty: writing begins at Off. + } + if r.End == len(r.Buf) { + return 0 // Tail full: next byte wraps to the start. + } + return r.End +} + +// PeekWrite stages b offset bytes past the write position (see +// [Ring.writeStart]) without advancing it, so the bytes are not yet readable; a +// later [Ring.Commit] reveals them. It reports false, writing nothing, when +// offset is negative or offset+len(b) exceeds [Ring.Free]. Used to place +// out-of-order data ahead of a gap that a normal Write later fills. +func (r *Ring) PeekWrite(b []byte, offset int) bool { + if offset < 0 || offset+len(b) > r.Free() { + return false + } + off := r.writeStart() + offset + if off >= len(r.Buf) { + off -= len(r.Buf) + } + n := copy(r.Buf[off:], b) + if n < len(b) { + copy(r.Buf, b[n:]) + } + return true +} + +// Commit advances the write pointer by n bytes, making readable any bytes +// previously staged with [Ring.PeekWrite]. It copies nothing and errors if n is +// not positive or exceeds [Ring.Free]. +func (r *Ring) Commit(n int) error { + if n <= 0 { + return errInvalidCommit + } else if n > r.Free() { + return ErrRingBufferFull + } + if r.End == 0 { + r.End = r.Off // Match Write: commit begins at Off when empty. + } + end := r.End + n + if end > len(r.Buf) { + end -= len(r.Buf) + } + r.End = end // Never 0 here: end==len(Buf) is kept. + return nil +} + // ReadDiscard is a performance auxiliary method that performs a dummy read or no-op read // for advancing the read pointer n bytes without actually copying data. // This method panics if amount of bytes is more than buffered (see [Ring.Buffered]). diff --git a/internal/ring_test.go b/internal/ring_test.go index 0169747..fbd32b9 100644 --- a/internal/ring_test.go +++ b/internal/ring_test.go @@ -580,3 +580,84 @@ func canonRing(r *Ring) { r.onReadEnd(1) } } + +// TestRingPeekWriteCommit verifies that bytes staged ahead of a gap with +// PeekWrite become readable in order once the gap is filled and committed. +func TestRingPeekWriteCommit(t *testing.T) { + r := &Ring{Buf: make([]byte, 16)} + // Stage "BBBB" 4 bytes ahead of the write position (the gap). + if !r.PeekWrite([]byte("BBBB"), 4) { + t.Fatal("PeekWrite should fit") + } + // Staged bytes are not yet readable. + if r.Buffered() != 0 { + t.Fatalf("staged bytes must not be readable, buffered=%d", r.Buffered()) + } + // Fill the gap with a normal write, then commit the staged tail. + if _, err := r.Write([]byte("AAAA")); err != nil { + t.Fatal("gap write:", err) + } + if err := r.Commit(4); err != nil { + t.Fatal("commit:", err) + } + got := make([]byte, 8) + n, err := r.Read(got) + if err != nil { + t.Fatal("read:", err) + } + if string(got[:n]) != "AAAABBBB" { + t.Fatalf("read %q, want AAAABBBB", got[:n]) + } + testRingSanity(t, r) +} + +// TestRingPeekWriteWrap exercises PeekWrite/Commit when the staged region wraps +// across the end of the backing buffer. Existing data at Off=2,End=6 puts the +// write position at index 6, so a 2-byte gap fills indices 6,7 and the staged +// tail wraps to indices 0,1. +func TestRingPeekWriteWrap(t *testing.T) { + r := &Ring{Buf: make([]byte, 8)} + setRingData(t, r, 2, []byte("WXYZ")) // Off=2, End=6, 4 bytes buffered. + if r.writeStart() != 6 { + t.Fatalf("writeStart=%d, want 6", r.writeStart()) + } + // Stage "CD" 2 bytes ahead of the write position (6) → wraps to indices 0,1. + if !r.PeekWrite([]byte("CD"), 2) { + t.Fatal("PeekWrite (wrap) should fit") + } + // Fill the 2-byte gap at indices 6,7, then commit the wrapped tail. + if _, err := r.Write([]byte("AB")); err != nil { + t.Fatal("gap write:", err) + } + if err := r.Commit(2); err != nil { + t.Fatal("commit:", err) + } + got := make([]byte, 8) + n, err := r.Read(got) + if err != nil { + t.Fatal("read:", err) + } + if string(got[:n]) != "WXYZABCD" { + t.Fatalf("read %q, want WXYZABCD", got[:n]) + } + testRingSanity(t, r) +} + +func TestRingPeekWriteRejects(t *testing.T) { + r := &Ring{Buf: make([]byte, 8)} + if r.PeekWrite([]byte("toolong!!"), 0) { + t.Error("PeekWrite must reject data larger than the buffer") + } + if r.PeekWrite([]byte("data"), 5) { // 5+4 > 8 free. + t.Error("PeekWrite must reject offset+len beyond free space") + } + if r.PeekWrite([]byte("x"), -1) { + t.Error("PeekWrite must reject negative offset") + } + if err := r.Commit(0); err == nil { + t.Error("Commit(0) must error") + } + if err := r.Commit(9); err == nil { + t.Error("Commit beyond free must error") + } +} diff --git a/tcp/handler.go b/tcp/handler.go index 822a8db..369a1ef 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -28,7 +28,10 @@ type Handler struct { // connection is established via Open calls. This disambiguates whether // Read and Write calls belong to the current connection. - optcodec OptionCodec + optcodec OptionCodec + // reasm tracks out-of-order segments staged in bufRx's free region. Always + // enabled once buffers are set (see [Handler.SetBuffers]). + reasm reassembly closing bool shutdownRx bool // nRetransmit stores the number of times the oldest packet was retransmit. @@ -64,6 +67,7 @@ func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { } h.scb.SetRecvWindow(Size(h.bufRx.Size())) h.bufRx.Reset() + h.reasm.reset(maxReasmSegments) return h.bufTx.ResetOrReuse(txbuf, packets, 0) } @@ -131,9 +135,11 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { remotePort: remotePort, validator: h.validator, logger: h.logger, + reasm: h.reasm, closing: false, shutdownRx: false, } + h.reasm.clear() // preserve metadata capacity across reopen, drop held segments. h.bufTx.ResetOrReuse(nil, 0, iss) h.bufRx.Reset() } @@ -163,15 +169,22 @@ func (h *Handler) Recv(incomingPacket []byte) error { return lneto.ErrMismatch } payload := tfrm.Payload() - if !h.shutdownRx && len(payload) > h.bufRx.Free() { - return lneto.ErrBufferFull - } segIncoming := tfrm.Segment(len(payload)) if h.scb.IncomingIsKeepalive(segIncoming) { h.info("tcp.Handler:rx-keepalive", slog.Uint64("port", uint64(h.localPort))) return nil } + // Out-of-order reassembly: buffer in-window data that arrived ahead of the + // next expected sequence number before the ControlBlock (sequential-only) + // would reject it. Buffered segments live in bufRx's free region. + if h.reasm.enabled() && h.handleOutOfOrder(segIncoming, payload) { + return nil + } + if !h.shutdownRx && len(payload) > h.bufRx.Free() { + return lneto.ErrBufferFull + } + prevState := h.scb.State() prevUNA := h.scb.snd.UNA // Capture before Recv updates snd.UNA (RFC 6298 §5.3). err = h.scb.Recv(segIncoming) @@ -204,6 +217,11 @@ func (h *Handler) Recv(incomingPacket []byte) error { return err } } + if segIncoming.DATALEN != 0 { + // The just-accepted in-order segment may have filled a gap; deliver any + // now-contiguous buffered segments. + h.deliverReassembled() + } if segIncoming.Flags.HasAny(FlagACK) { if segIncoming.ACK == prevUNA { // scb keeping track of duplicate acks. @@ -242,6 +260,54 @@ func (h *Handler) Recv(incomingPacket []byte) error { return nil } +// handleOutOfOrder buffers an in-window data segment that arrived ahead of the +// next expected sequence number and queues a duplicate ACK so the sender fast- +// retransmits the gap. It returns true when it has consumed the segment; false +// leaves the segment to the ControlBlock (in-order data, control segments, old +// or out-of-window segments, or when the reassembly buffer cannot hold it). +func (h *Handler) handleOutOfOrder(seg Segment, payload []byte) bool { + if h.shutdownRx { + // Discard mode drops payloads, which would break the ring/rcv.NXT + // lockstep reassemble relies on; do not buffer. + return false + } + if seg.DATALEN == 0 || seg.Flags.HasAny(flagctl) { + return false // only pure data segments are buffered out of order. + } + rcvNxt := h.scb.RecvNext() + if seg.SEQ == rcvNxt { + return false // in order: the ControlBlock handles it normally. + } + rcvWnd := h.scb.RecvWindow() + if !seg.SEQ.InWindow(rcvNxt, rcvWnd) || !seg.Last().InWindow(rcvNxt, rcvWnd) { + return false // old or out of window: let the ControlBlock decide. + } + if !h.reasm.store(&h.bufRx, rcvNxt, seg.SEQ, payload) { + return false // no room: fall back to ControlBlock (challenge ACK). + } + h.scb.pending[0] |= FlagACK // duplicate ACK advertises the gap at rcv.NXT. + h.trace("tcp.Handler:rx-ooo", slog.Uint64("seg.seq", uint64(seg.SEQ)), slog.Uint64("rcv.nxt", uint64(rcvNxt))) + return true +} + +// deliverReassembled hands any now-contiguous out-of-order segments to the +// receive stream, advancing rcv.NXT and queuing an ACK for what was delivered. +func (h *Handler) deliverReassembled() { + if h.reasm.buffered() == 0 { + return + } + if h.shutdownRx { + // Discard mode skips the gap-filling write, so staged bytes can no + // longer be committed coherently; drop them (the peer retransmits). + h.reasm.clear() + return + } + if delivered := h.reasm.reassemble(&h.bufRx, h.scb.RecvNext()); delivered > 0 { + h.scb.rcv.NXT.UpdateForward(delivered) + h.scb.pending[0] |= FlagACK + } +} + // ShutdownRead activates local discard mode: incoming payload bytes are dropped // (ACK/SEQ still advance normally) and Read returns [io.EOF] immediately. // Not reversible within the lifetime of a connection. @@ -328,7 +394,7 @@ func (h *Handler) Send(b []byte) (int, error) { var ok bool maxPayload := len(b) - sizeHeaderTCP segment, ok = h.scb.PendingSegment(maxPayload) - segment.WND = Size(h.bufRx.Free()) + segment.WND = h.recvWindow() if !ok { // No pending control segment or data to send. Yield. return 0, nil @@ -391,6 +457,9 @@ func (h *Handler) Read(b []byte) (n int, err error) { n, err = h.bufRx.Read(b) } if n > 0 { + // Reading freed receive-buffer space; deliver any contiguous + // out-of-order data that was waiting for room. + h.deliverReassembled() h.maybeQueueWindowUpdate() } if n == 0 && err == nil { @@ -413,7 +482,7 @@ func (h *Handler) Read(b []byte) (n int, err error) { // space >= min(bufferSize/2, MSS). This applies uniformly including zero-window // recovery — the remote uses zero-window probes until enough space opens. func (h *Handler) maybeQueueWindowUpdate() { - currentFree := Size(h.bufRx.Free()) + currentFree := h.recvWindow() lastAdvertised := h.scb.RecvWindow() if currentFree <= lastAdvertised { return // Window hasn't grown. @@ -454,7 +523,21 @@ func (h *Handler) FreeOutput() int { // FreeInput returns the number of free bytes in the receive buffer. func (h *Handler) FreeInput() int { - return h.bufRx.Free() + return int(h.recvWindow()) +} + +// recvWindow returns the receive window to advertise: free receive-buffer space +// minus the bytes already held out of order. Subtracting them prevents the +// sender from overrunning the receiver while a gap is open. +func (h *Handler) recvWindow() Size { + free := Size(h.bufRx.Free()) + if !h.reasm.enabled() { + return free + } + if ooo := Size(h.reasm.bufferedBytes()); ooo < free { + return free - ooo + } + return 0 } // AwaitingSynResponse returns true if the Handler is an active client opened with [Handler.OpenActive] and has already sent out the first SYN packet to the remote client. diff --git a/tcp/handler_test.go b/tcp/handler_test.go index 8722106..2700000 100644 --- a/tcp/handler_test.go +++ b/tcp/handler_test.go @@ -3,6 +3,7 @@ package tcp import ( "bytes" "fmt" + "io" "math/rand" "testing" @@ -1487,3 +1488,103 @@ func TestRetransmit_CumulativeACK_NoSpurious(t *testing.T) { t.Fatalf("spurious retransmission after cumulative ACK: SEQ=%d len=%d (issue #57)", seg.SEQ, seg.DATALEN) } } + +// emitClientData writes payload to the client and emits it as one data packet, +// returning a copy of the wire bytes (the caller controls delivery order). +func emitClientData(t *testing.T, client *Handler, buf []byte, payload string) []byte { + t.Helper() + if _, err := client.Write([]byte(payload)); err != nil { + t.Fatal("client write:", err) + } + clear(buf) + n, err := client.Send(buf) + if err != nil { + t.Fatal("client send:", err) + } + if n <= sizeHeaderTCP { + t.Fatal("expected a data segment, got header-only") + } + return append([]byte(nil), buf[:n]...) +} + +// TestHandler_OutOfOrderReassembly drives the full out-of-order path: a later +// segment delivered before the gap-filling one is staged, then delivered +// contiguously once the gap arrives, without go-back-N. +func TestHandler_OutOfOrderReassembly(t *testing.T) { + const mtu = ethernet.MaxMTU + const maxpackets = 4 + rng := rand.New(rand.NewSource(99)) + client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + pkt1 := emitClientData(t, client, buf[:], "AAAA") // seq S, covers S..S+4. + pkt2 := emitClientData(t, client, buf[:], "BBBB") // seq S+4, covers S+4..S+8. + + full := server.SizeInput() + + // Deliver the second segment first: accepted, buffered, not yet readable. + if err := server.Recv(pkt2); err != nil { + t.Fatalf("out-of-order segment must be accepted, got: %v", err) + } + if server.BufferedInput() != 0 { + t.Fatalf("OOO data must not be readable yet, buffered=%d", server.BufferedInput()) + } + // Advertised window shrinks by the held bytes so the peer cannot overrun. + if got := server.FreeInput(); got != full-4 { + t.Fatalf("FreeInput=%d, want %d (window reduced by held OOO bytes)", got, full-4) + } + + // Deliver the gap-filling first segment: both become contiguous. + if err := server.Recv(pkt1); err != nil { + t.Fatalf("gap-filling segment: %v", err) + } + if server.BufferedInput() != 8 { + t.Fatalf("buffered=%d, want 8 after gap fill", server.BufferedInput()) + } + if got := server.FreeInput(); got != full-8 { + t.Fatalf("FreeInput=%d, want %d after delivery", got, full-8) + } + + var rd [16]byte + n, err := server.Read(rd[:]) + if err != nil { + t.Fatal("server read:", err) + } + if string(rd[:n]) != "AAAABBBB" { + t.Fatalf("reassembled %q, want AAAABBBB", rd[:n]) + } +} + +// TestHandler_OutOfOrderDiscardedAfterShutdownRead verifies staged segments are +// dropped, not delivered, when the read side is shut down before the gap fills. +func TestHandler_OutOfOrderDiscardedAfterShutdownRead(t *testing.T) { + const mtu = ethernet.MaxMTU + const maxpackets = 4 + rng := rand.New(rand.NewSource(100)) + client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + pkt1 := emitClientData(t, client, buf[:], "AAAA") + pkt2 := emitClientData(t, client, buf[:], "BBBB") + + if err := server.Recv(pkt2); err != nil { // buffer out of order. + t.Fatalf("OOO segment: %v", err) + } + server.ShutdownRead() // application done reading; staged data must be dropped. + + if err := server.Recv(pkt1); err != nil && !IsDroppedErr(err) { + t.Fatalf("gap-filling segment after shutdown: %v", err) + } + var rd [16]byte + n, err := server.Read(rd[:]) + if n != 0 || err != io.EOF { + t.Fatalf("read after ShutdownRead = %d,%v want 0,EOF", n, err) + } + if server.BufferedInput() != 0 { + t.Fatalf("discard mode must hold no data, buffered=%d", server.BufferedInput()) + } +} diff --git a/tcp/reassembly.go b/tcp/reassembly.go new file mode 100644 index 0000000..1f7951f --- /dev/null +++ b/tcp/reassembly.go @@ -0,0 +1,135 @@ +package tcp + +import "github.com/soypat/lneto/internal" + +// maxReasmSegments bounds how many distinct out-of-order segments may be held. +// It caps only fixed metadata; payload bytes live in the receive ring, bounded +// by its free space. Independent of the transmit queue depth. +const maxReasmSegments = 8 + +// reassembly holds in-window TCP segments that arrived ahead of the next +// expected sequence number, so that once the gap is filled the buffered tail is +// delivered without go-back-N. Payloads are staged in the free region of the +// Handler receive ring (see [internal.Ring.PeekWrite]); only fixed, reused +// metadata lives here, so the data path allocates nothing. +// +// held is kept ordered by ascending sequence number (oldest to newest), which +// lets [reassembly.store] locate insertions and overlaps by neighbour and lets +// [reassembly.reassemble] deliver a contiguous prefix and truncate it in one +// pass. +type reassembly struct { + held []reasmSeg +} + +// reasmSeg records a held segment by sequence number and payload length. No +// buffer offset is kept: the ring write pointer advances in lockstep with +// rcv.NXT, so the staged bytes are always where seq implies (see +// [reassembly.reassemble]). +type reasmSeg struct { + seq Value + n int +} + +// reset (re)configures bounded metadata for up to maxSegs held segments, or +// disables reassembly when maxSegs is not positive. Held state is cleared; +// metadata capacity persists across connection reopens. +func (r *reassembly) reset(maxSegs int) { + if maxSegs <= 0 { + r.held = nil + return + } + internal.SliceReuse(&r.held, maxSegs) +} + +// clear drops all held segments without changing configuration. +func (r *reassembly) clear() { r.held = r.held[:0] } + +// enabled reports whether out-of-order buffering is configured. +func (r *reassembly) enabled() bool { return cap(r.held) > 0 } + +// buffered reports the number of out-of-order segments currently held. +func (r *reassembly) buffered() int { return len(r.held) } + +// bufferedBytes reports the total payload bytes currently held out of order. +// The receiver subtracts these from its advertised window so the sender cannot +// overrun the space the held segments already consume. +func (r *reassembly) bufferedBytes() int { + n := 0 + for i := range r.held { + n += r.held[i].n + } + return n +} + +// store stages payload at the offset it will occupy in rx once the gap from +// rcvNxt fills, keeping held ordered by seq. It returns true when held, +// including when already held (storing is idempotent), and false when disabled, +// the payload is empty, metadata is full, it does not fit rx's free region, or +// it overlaps a held segment. +func (r *reassembly) store(rx *internal.Ring, rcvNxt, seq Value, payload []byte) bool { + if !r.enabled() || len(payload) == 0 || len(r.held) >= cap(r.held) { + return false + } + gap := int(Sizeof(rcvNxt, seq)) + // Early free-space bail before the ordered-insert scan; strictly cautious, + // as PeekWrite re-checks this below. + if gap+len(payload) > rx.Free() { + return false + } + // Find the insertion point that keeps held ordered by ascending seq. + end := Add(seq, Size(len(payload))) + i := 0 + for i < len(r.held) && r.held[i].seq.LessThan(seq) { + i++ + } + if i > 0 { // Overlaps the predecessor? + if prev := r.held[i-1]; seq.LessThan(Add(prev.seq, Size(prev.n))) { + return false + } + } + if i < len(r.held) { // Duplicate, or overlaps the successor? + if next := r.held[i]; next.seq == seq { + return true // already buffered; idempotent. + } else if next.seq.LessThan(end) { + return false + } + } + if !rx.PeekWrite(payload, gap) { + return false + } + r.held = append(r.held, reasmSeg{}) + copy(r.held[i+1:], r.held[i:]) + r.held[i] = reasmSeg{seq: seq, n: len(payload)} + return true +} + +// reassemble delivers held segments contiguous with nxt by committing their +// staged bytes to rx, and drops any beginning before nxt (stale, or overwritten +// by the in-order write that advanced nxt). Because held is ordered, it walks a +// leading prefix and truncates once. It returns the bytes delivered; the caller +// advances rcv.NXT and ACKs. Delivery stops at the first gap, or if rx is full +// (the remainder is delivered on a later call). +func (r *reassembly) reassemble(rx *internal.Ring, nxt Value) Size { + var delivered Size + i := 0 + for i < len(r.held) { + seg := r.held[i] + switch { + case seg.seq == nxt: + if rx.Commit(seg.n) != nil { + r.held = append(r.held[:0], r.held[i:]...) + return delivered + } + nxt = Add(nxt, Size(seg.n)) + delivered += Size(seg.n) + i++ + case seg.seq.LessThan(nxt): + i++ // stale/overwritten: drop. + default: + r.held = append(r.held[:0], r.held[i:]...) + return delivered // gap before the next segment. + } + } + r.held = r.held[:0] + return delivered +} diff --git a/tcp/reassembly_test.go b/tcp/reassembly_test.go new file mode 100644 index 0000000..3753bfa --- /dev/null +++ b/tcp/reassembly_test.go @@ -0,0 +1,168 @@ +package tcp + +import ( + "bytes" + "testing" + + "github.com/soypat/lneto/internal" +) + +func TestReassemblyDisabledByDefault(t *testing.T) { + var r reassembly + if r.enabled() { + t.Fatal("zero-value reassembly must be disabled") + } + var rx internal.Ring + if r.store(&rx, 100, 100, []byte("x")) { + t.Error("store must fail when disabled") + } +} + +func TestReassemblyStoreAndReassemble(t *testing.T) { + var r reassembly + r.reset(4) + rx := internal.Ring{Buf: make([]byte, 32)} + if !r.enabled() { + t.Fatal("reassembly should be enabled after reset") + } + // Buffer two out-of-order segments (gap at seq 100), stored newest-first to + // prove store keeps held ordered by seq. + if !r.store(&rx, 100, 108, []byte("CCC")) { // covers 108..111 + t.Fatal("store 108 failed") + } + if !r.store(&rx, 100, 104, []byte("BBBB")) { // covers 104..108 + t.Fatal("store 104 failed") + } + if r.buffered() != 2 { + t.Fatalf("buffered=%d, want 2", r.buffered()) + } + if r.held[0].seq != 104 || r.held[1].seq != 108 { + t.Fatalf("held not ordered by seq: %+v", r.held) + } + // A gap at 100 blocks delivery entirely. + if got := r.reassemble(&rx, 100); got != 0 { + t.Fatalf("reassemble(100)=%d, want 0 with a gap at 100", got) + } + // Once 100..104 is delivered in order, both held segments become contiguous. + if _, err := rx.Write([]byte("AAAA")); err != nil { + t.Fatal("gap write:", err) + } + if got := r.reassemble(&rx, 104); got != 7 { + t.Fatalf("reassemble(104)=%d, want 7", got) + } + if r.buffered() != 0 { + t.Errorf("buffered=%d, want 0 after full delivery", r.buffered()) + } + out := make([]byte, 16) + n, _ := rx.Read(out) + if !bytes.Equal(out[:n], []byte("AAAABBBBCCC")) { + t.Fatalf("reassembled %q, want AAAABBBBCCC", out[:n]) + } +} + +func TestReassemblyDedup(t *testing.T) { + var r reassembly + r.reset(4) + rx := internal.Ring{Buf: make([]byte, 32)} + if !r.store(&rx, 100, 104, []byte("AAA")) { + t.Fatal("first store failed") + } + if !r.store(&rx, 100, 104, []byte("AAA")) { + t.Error("duplicate store of same seq should be idempotent true") + } + if r.buffered() != 1 { + t.Errorf("buffered=%d, want 1 (duplicate must not add a segment)", r.buffered()) + } +} + +func TestReassemblyFull(t *testing.T) { + var r reassembly + r.reset(2) // only 2 slots. + rx := internal.Ring{Buf: make([]byte, 32)} + if !r.store(&rx, 100, 104, []byte("a")) || !r.store(&rx, 100, 108, []byte("b")) { + t.Fatal("filling slots failed") + } + if r.store(&rx, 100, 116, []byte("c")) { + t.Error("store must fail when all slots are occupied") + } +} + +func TestReassemblyOversizedRejected(t *testing.T) { + var r reassembly + r.reset(2) + rx := internal.Ring{Buf: make([]byte, 4)} + if r.store(&rx, 100, 104, []byte("toolong")) { + t.Error("payload larger than free receive space must be rejected") + } +} + +func TestReassemblyOverlapRejected(t *testing.T) { + var r reassembly + r.reset(4) + rx := internal.Ring{Buf: make([]byte, 32)} + if !r.store(&rx, 100, 104, []byte("BBBB")) { // covers 104..108 + t.Fatal("store 104 failed") + } + // Segments overlapping the held 104..108 region must be rejected. + if r.store(&rx, 100, 106, []byte("XX")) { // 106..108 overlaps its successor + t.Error("overlapping store must be rejected") + } + if r.store(&rx, 100, 102, []byte("YYYY")) { // 102..106 overlaps its predecessor + t.Error("overlapping store must be rejected") + } + if r.buffered() != 1 { + t.Errorf("buffered=%d, want 1 (overlaps must not be stored)", r.buffered()) + } +} + +// TestReassembleDropsStale checks segments beginning before nxt are dropped, not +// delivered (their staged bytes may have been overwritten by the in-order write). +func TestReassembleDropsStale(t *testing.T) { + var r reassembly + r.reset(4) + rx := internal.Ring{Buf: make([]byte, 32)} + r.store(&rx, 100, 104, []byte("BBBB")) // covers 104..108 + r.store(&rx, 100, 112, []byte("DDDD")) // covers 112..116 + // rcv.NXT advanced to 106, partway into the first held segment, with a gap + // before the second: the stale segment is dropped and nothing is delivered. + if got := r.reassemble(&rx, 106); got != 0 { + t.Errorf("reassemble(106)=%d, want 0", got) + } + if r.buffered() != 1 || r.held[0].seq != 112 { + t.Errorf("held=%+v, want only seq 112", r.held) + } +} + +func TestReassemblyResetDisables(t *testing.T) { + var r reassembly + rx := internal.Ring{Buf: make([]byte, 16)} + r.reset(4) + r.store(&rx, 100, 104, []byte("a")) + r.reset(0) + if r.enabled() { + t.Error("reset(0) must disable reassembly") + } + if r.buffered() != 0 { + t.Error("reset must clear held segments") + } +} + +// TestReassembly_noAllocs verifies the data-path operations allocate nothing +// once configured (metadata is bounded at reset, payloads reuse the ring). +func TestReassembly_noAllocs(t *testing.T) { + var r reassembly + r.reset(4) + rx := internal.Ring{Buf: make([]byte, 32)} + seg := []byte("DATA") + allocs := testing.AllocsPerRun(100, func() { + r.clear() + rx.Reset() + r.store(&rx, 100, 108, seg) // buffer out of order. + r.store(&rx, 100, 104, seg) // buffer out of order. + _ = r.bufferedBytes() + _ = r.reassemble(&rx, 112) + }) + if allocs != 0 { + t.Errorf("reassembly data path must not allocate, got %v allocs/op", allocs) + } +}