From 64b2647a5e0e9907c4b5ccada71f6219c3660301 Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Thu, 26 Mar 2026 11:48:02 -0300 Subject: [PATCH] tcp: retransmit logic rollback (#61) * tcp: remove retransmit logic entirely; add duplicate ack counting to ControlBlock * retransmit implemented in nice simple straightforward way * narrow down retransmission cases * remove old timing tests * add ControlBlock retransmit test * add failing handler test * reworking payload length semantic meaning in code * fix establish conn logic * clean up tests and add TCB dupack generation and test it * catch pending retransmit satisfy in test * add fuzz test for control block * bugfix: be more strict in what is considered dupack * add IncomingIsDupACK docs * limit queue of retransmits * protect retransmit overflow from incorrectly updating nxt --- internal/prand.go | 8 + tcp/conn.go | 22 -- tcp/control.go | 100 +++-- tcp/control_test.go | 82 ++++ tcp/definitions.go | 2 + tcp/handler.go | 136 ++----- tcp/handler_retransmit_test.go | 674 --------------------------------- tcp/handler_test.go | 191 +++++++++- tcp/tcp_fuzz_test.go | 98 +++++ tcp/txqueue.go | 61 +-- tcp/txqueue_test.go | 22 +- x/xnet/tcppool.go | 1 - 12 files changed, 508 insertions(+), 889 deletions(-) delete mode 100644 tcp/handler_retransmit_test.go create mode 100644 tcp/tcp_fuzz_test.go diff --git a/internal/prand.go b/internal/prand.go index a5544c0..b07c514 100644 --- a/internal/prand.go +++ b/internal/prand.go @@ -17,3 +17,11 @@ func Prand32[T ~uint32](seed T) T { seed ^= seed << 5 return seed } + +// Prand32 generates a pseudo random number from a seed. +func Prand64[T ~uint64](seed T) T { + seed ^= seed << 13 + seed ^= seed >> 7 + seed ^= seed << 17 + return seed +} diff --git a/tcp/conn.go b/tcp/conn.go index 6ae16e6..164ed72 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -28,7 +28,6 @@ type Conn struct { mu sync.Mutex h Handler remoteAddr []byte - nanoTime func() int64 // monotonic clock source; set by Configure. rdead time.Time wdead time.Time @@ -56,10 +55,6 @@ type ConnConfig struct { TxBuf []byte TxPacketQueueSize int Logger *slog.Logger - // NanoTime returns the current monotonic time in nanoseconds. - // Used for retransmission timing (RFC 6298). - // If nil, defaults to a function that calls time.Now().UnixNano(). - NanoTime func() int64 } func (conn *Conn) Configure(config ConnConfig) (err error) { @@ -70,19 +65,9 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { return err } conn.logger.log = config.Logger - conn.nanoTime = config.NanoTime // nil is fine; conn.now() falls back to time.Now(). return nil } -// now returns the current monotonic time in nanoseconds. -// Uses the configured NanoTime function or falls back to time.Now().UnixNano(). -func (conn *Conn) now() int64 { - if conn.nanoTime != nil { - return conn.nanoTime() - } - return time.Now().UnixNano() -} - // LocalPort returns the local port on which the socket is listening or connected to. func (conn *Conn) LocalPort() uint16 { conn.mu.Lock() @@ -354,7 +339,6 @@ func (conn *Conn) Demux(buf []byte, off int) (err error) { return lneto.ErrMismatch } conn.trace("tcpconn.Recv", slog.Uint64("lport", uint64(conn.h.LocalPort())), slog.Uint64("rport", uint64(conn.h.remotePort))) - conn.h.SetNow(uint32(conn.now() / 1e6)) // ns → ms for accurate ACK timestamps. err = conn.h.Recv(buf[off:]) if err != nil { return err @@ -382,12 +366,6 @@ func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) } else if len(raddr) != len(conn.remoteAddr) { return 0, lneto.ErrMismatchLen } - conn.h.SetNow(uint32(conn.now() / 1e6)) // ns → ms. - // RFC 6298 §5.1: check RTO before sending new data. - if conn.h.ShouldRetransmit() { - conn.h.triggerRetransmit() - conn.h.dupACKs = 0 // RTO is a new loss event; reset dup-ACK counter. - } n, err = conn.h.Send(carrierData[offsetToFrame:]) if err != nil || n == 0 { return 0, err diff --git a/tcp/control.go b/tcp/control.go index 5a6cc9c..9c6d6a8 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -9,6 +9,13 @@ import ( "github.com/soypat/lneto/internal" ) +const ( + // signals to create a retransmit packet after receiving this number of duplicate acks, not including the ack that set UNA. + retransmitAfterDupacks = 3 + // retransmitMaxQueued sets maximum amount of retransmits to queue while receiving dupacks. + retransmitMaxQueued = 2 +) + // ControlBlock is a partial Transmission Control Block (TCB) implementation as // per RFC 9293 in section 3.3.1. In contrast with the description in RFC9293, // this implementation is limited to receiving only sequential segments. @@ -59,6 +66,10 @@ type ControlBlock struct { pending [2]Flags _state State // leading underscore so field not suggested on top of exported State method when developing. challengeAck bool + // dupack counts received ACK==snd.UNA && ACK= retransmitAfterDupacks && tcb.nRetransmit <= tcb.dupack-retransmitAfterDupacks +} // PendingSegment calculates a suitable next segment to send from a payload length. // It does not modify the ControlBlock state or pending segment queue. func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { + pending := tcb.pending[0] if tcb.challengeAck { // Do not clear challengeAck here: PendingSegment is documented as read-only. // The flag is consumed in Send when the ACK segment is actually transmitted. return Segment{SEQ: tcb.snd.NXT, ACK: tcb.rcv.NXT, Flags: FlagACK, WND: tcb.rcv.WND}, true + } else if !pending.HasAny(flagctl) && tcb.HasPendingRetransmit() { + // Optimist Strategy: retransmit oldest data once. + return Segment{SEQ: tcb.snd.UNA, DATALEN: Size(payloadLen), ACK: tcb.rcv.NXT, WND: tcb.rcv.WND, Flags: FlagACK}, true } - pending := tcb.pending[0] established := tcb._state == StateEstablished canSendData := established || tcb._state == StateCloseWait if !canSendData { @@ -217,9 +262,6 @@ func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { if tcb.snd.MSS > 0 && payloadLen > int(tcb.snd.MSS) { payloadLen = int(tcb.snd.MSS) } - if payloadLen > 0 { - pending |= FlagPSH // By default ensure all data flushed to destination application immediately on receive. - } if canSendData { pending |= FlagACK // ACK is always set in established state. Not in RFC9293 but somehow expected? @@ -321,10 +363,19 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) { tcb.snd.WL1 = seg.SEQ tcb.snd.WL2 = seg.ACK } - if seg.Flags.HasAny(FlagACK) && tcb.snd.UNA.LessThan(seg.ACK) && seg.ACK.LessThanEq(tcb.snd.NXT) { - // Only update ACK if it advances UNA and is not in the future. - tcb.snd.UNA = seg.ACK + + if seg.Flags.HasAny(FlagACK) && seg.ACK.LessThanEq(tcb.snd.NXT) { + if tcb.IncomingIsDupACK(seg.ACK) && tcb.State().TxDataOpen() && !seg.Flags.HasAny(flagctl) && tcb.dupack < tcb.nRetransmit+retransmitMaxQueued+retransmitMaxQueued { + // Duplicate ack. Don't advance dupack counter past scb.nRetransmit+retransmitAfterDupacks + tcb.dupack++ + } else if tcb.snd.UNA.LessThan(seg.ACK) { + // Only update ACK if it advances UNA and is not in the future. + tcb.snd.UNA = seg.ACK + tcb.dupack = 0 + tcb.nRetransmit = 0 + } } + seglen := seg.LEN() tcb.rcv.NXT.UpdateForward(seglen) @@ -385,9 +436,16 @@ func (tcb *ControlBlock) Send(seg Segment) error { // The segment is valid, we can update TCB state. seglen := seg.LEN() - tcb.snd.NXT.UpdateForward(seglen) - tcb.rcv.WND = seg.WND + retransmit := seg.SEQ.LessThan(tcb.snd.NXT) + if retransmit { + if tcb.nRetransmit < 255-retransmitMaxQueued-retransmitAfterDupacks { + tcb.nRetransmit++ + } + } else { + tcb.snd.NXT.UpdateForward(seglen) + } + tcb.rcv.WND = seg.WND if tcb.logenabled(internal.LevelTrace) { tcb.traceSnd("tcb:snd") tcb.traceSeg("tcb:snd", seg) @@ -405,6 +463,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { zeroWindowOK := tcb.snd.WND == 0 && seg.DATALEN == 0 && seg.SEQ == tcb.snd.NXT outOfWindow := checkSeq && !seg.SEQ.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK + isRetransmit := checkSeq && seg.SEQ.InRange(tcb.snd.UNA, tcb.snd.NXT) switch { case tcb._state == StateClosed && !isFirst: err = io.ErrClosedPipe @@ -413,7 +472,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { case hasAck && seg.ACK != tcb.rcv.NXT: err = errAckNotNext - case outOfWindow: + case outOfWindow && !isRetransmit: if tcb.snd.WND == 0 { err = errZeroWindow } else { @@ -426,7 +485,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { case checkSeq && tcb.snd.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.snd.NXT: err = errZeroWindow - case checkSeq && !seglast.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK: + case checkSeq && !seglast.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK && !isRetransmit: err = errLastNotInWindow } return err @@ -489,7 +548,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { case established && acksOld && !ctlOrDataSegment: // We don't drop packet. if isDebug { - tcb.debug("rcv:ACK-dup", slog.String("state", tcb._state.String()), + tcb.debug("rcv:ACK-old", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.una", uint64(tcb.snd.UNA))) } @@ -571,20 +630,7 @@ func (tcb *ControlBlock) rstJump() Value { // and Send calls to retransmit unacknowledged data. Must be paired with // ringTx.RetransmitFromUNA to rewind the transmit buffer. // Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT). -func (tcb *ControlBlock) Retransmit() { tcb.snd.NXT = tcb.snd.UNA } - -// RecoveryACK accepts a cumulative ACK that covers data sent before a retransmit -// rewind. After Retransmit() rewinds snd.NXT, the remote may ACK data it received -// pre-rewind — a valid cumulative ACK that exceeds the rewound snd.NXT. This method -// advances snd.UNA, snd.NXT and updates the send window from the segment. -// The caller must verify that seg.ACK is within the pre-rewind NXT range. -func (tcb *ControlBlock) RecoveryACK(seg Segment) { - tcb.snd.UNA = seg.ACK - tcb.snd.NXT = seg.ACK - tcb.snd.WND = seg.WND - // Clear any pending ACK that validateIncomingSegment queued on rejection. - tcb.pending[0] &^= FlagACK -} +// func (tcb *ControlBlock) Retransmit() { tcb.snd.NXT = tcb.snd.UNA } // Abort sets ControlBlock state to Closed and resets all sequence numbers and pending flag. // No more data can be sent nor received after the connection is aborted until opened again. diff --git a/tcp/control_test.go b/tcp/control_test.go index dc2c1b2..a8c0813 100644 --- a/tcp/control_test.go +++ b/tcp/control_test.go @@ -293,3 +293,85 @@ func TestCloseWait_NoAutoFINBeforeUserClose(t *testing.T) { "(control.go:353-354 queues finack on any ACK sent in CLOSE-WAIT)", seg.Flags) } } + +func TestPendingSegment_RetransmitAfter3DupACKs(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + inFlight = 10 + wnd Size = 1024 + ) + + var tcb ControlBlock + tcb.HelperInitState(StateEstablished, iss, iss+inFlight, wnd) + tcb.HelperInitRcv(remoteISS, remoteISS+1, wnd) + + // Three duplicate ACKs against UNA must trigger retransmit state + for i := 0; i < 3; i++ { + dup := Segment{ + SEQ: remoteISS + 1, + ACK: iss, // UNA (duplicate, no progress) + Flags: FlagACK, + WND: wnd, + } + if !tcb.IncomingIsDupACK(dup.ACK) { + t.Fatal("supposed duplicate ack segment not considered dupack") + } + if err := tcb.Recv(dup); err != nil { + t.Fatalf("dup ACK %d: unexpected error: %v", i+1, err) + } + } + + if tcb.dupack != 3 { + t.Fatalf("dupack = %d; want 3", tcb.dupack) + } + if !tcb.HasPendingRetransmit() { + t.Fatal("expected HasPendingRetransmit() == true after 3 dupacks") + } + + seg, ok := tcb.PendingSegment(4) + if !ok { + t.Fatal("PendingSegment(false) returned no segment; expected retransmit segment") + } + if seg.SEQ != tcb.snd.UNA { + t.Fatalf("retransmit SEQ = %d; want UNA(%d)", seg.SEQ, tcb.snd.UNA) + } + if seg.ACK != tcb.rcv.NXT { + t.Fatalf("retransmit ACK = %d; want RCV.NXT(%d)", seg.ACK, tcb.rcv.NXT) + } + if !seg.Flags.HasAny(FlagACK) { + t.Errorf("retransmit segment must include ACK") + } + + // Send retransmit, expect nRetransmit to be incremented and NXT not moved + prevNXT := tcb.snd.NXT + if err := tcb.Send(seg); err != nil { + t.Fatalf("Send(retransmit) unexpected error: %v", err) + } + if tcb.nRetransmit != 1 { + t.Fatalf("nRetransmit = %d; want 1", tcb.nRetransmit) + } + if tcb.snd.NXT != prevNXT { + t.Fatalf("snd.NXT advanced on retransmit: got %d, want %d", tcb.snd.NXT, prevNXT) + } + if tcb.HasPendingRetransmit() { + t.Fatal("expected retransmit reservation gone after retransmit send") + } + + // Deliver cumulative ACK for all in-flight data => reset dupack + nRetransmit + successACK := Segment{ + SEQ: remoteISS + 1, + ACK: iss + inFlight, + Flags: FlagACK, + WND: wnd, + } + if err := tcb.Recv(successACK); err != nil { + t.Fatalf("successful ACK unexpected err: %v", err) + } + if tcb.dupack != 0 { + t.Fatalf("dupack after progress = %d; want 0", tcb.dupack) + } + if tcb.nRetransmit != 0 { + t.Fatalf("nRetransmit after progress = %d; want 0", tcb.nRetransmit) + } +} diff --git a/tcp/definitions.go b/tcp/definitions.go index 8346800..eff90e2 100644 --- a/tcp/definitions.go +++ b/tcp/definitions.go @@ -172,6 +172,8 @@ const ( synack = FlagSYN | FlagACK finack = FlagFIN | FlagACK pshack = FlagPSH | FlagACK + + flagctl = FlagSYN | FlagFIN | FlagRST ) // HasAll checks if mask bits are all set in the receiver flags. diff --git a/tcp/handler.go b/tcp/handler.go index a031b04..2e8d3e7 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -30,22 +30,8 @@ type Handler struct { optcodec OptionCodec closing bool - // dupACKs counts consecutive duplicate ACKs for fast retransmit (RFC 5681 §3.2). - dupACKs uint8 - // nRetx counts consecutive retransmissions for exponential backoff (RFC 6298 §5.5). - nRetx uint8 - // Retransmission timer state — all uint32 milliseconds, no time package needed. - // rto is the current retransmission timeout in ms; starts at 1000 per RFC 6298 §2.1. - rto uint32 - // now is the current time in ms, set by Conn before Send/Recv via SetNow. - now uint32 - // lastACK is the last ACK value seen, for duplicate ACK detection (RFC 5681 §3.2). - lastACK Value - - // retransmitNXT is the pre-rewind value of snd.NXT, saved when fast retransmit - // fires. A cumulative ACK with seg.ACK <= retransmitNXT is valid even if it - // exceeds the rewound snd.NXT. Zero means not in recovery. - retransmitNXT Value + // nRetransmit stores the number of times the oldest packet was retransmit. + nRetransmit uint8 } func (h *Handler) SetLoggers(handler, scb *slog.Logger) { @@ -142,19 +128,11 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { validator: h.validator, logger: h.logger, closing: false, - rto: rtoInitial, // RFC 6298 §2.1: initial RTO = 1s. } h.bufTx.ResetOrReuse(nil, 0, iss) h.bufRx.Reset() } -const ( - // rtoInitial is the initial RTO per RFC 6298 §2.1: "the sender SHOULD set RTO <- 1 second". - rtoInitial uint32 = 1000 - // rtoMax caps exponential backoff per RFC 6298 §2.5. - rtoMax uint32 = 60_000 -) - // Recv receives an incoming TCP packet frame with the first byte being the first octet of the TCP frame. // The [Handler]'s internal state is updated if the packet is admitted successfully. func (h *Handler) Recv(incomingPacket []byte) error { @@ -188,34 +166,16 @@ func (h *Handler) Recv(incomingPacket []byte) error { h.info("tcp.Handler:rx-keepalive", slog.Uint64("port", uint64(h.localPort))) return nil } + prevState := h.scb.State() prevUNA := h.scb.snd.UNA // Capture before Recv updates snd.UNA (RFC 6298 §5.3). err = h.scb.Recv(segIncoming) if err != nil { - // Recovery path: after fast retransmit rewinds snd.NXT, a cumulative ACK - // for data sent pre-rewind exceeds the rewound NXT. The ControlBlock rejects - // it, but we know it's valid if ACK <= retransmitNXT (pre-rewind high water mark). - if h.retransmitNXT != 0 && segIncoming.Flags.HasAny(FlagACK) && - h.scb.snd.NXT.LessThan(segIncoming.ACK) && - segIncoming.ACK.LessThanEq(h.retransmitNXT) { - // TODO: This is a very hacky workaround. It'd be great - // to detect recover acks in Handler before calling ControlBlock.Recv - // and handle it cleanly instead of with an error. - h.scb.RecoveryACK(segIncoming) - h.bufTx.RecoveryACK(segIncoming.ACK) - h.retransmitNXT = 0 - h.rto = rtoInitial - h.nRetx = 0 - h.dupACKs = 0 - h.lastACK = segIncoming.ACK - err = nil // Accept the segment. - } else { - if h.scb.State() == StateClosed { - // TODO(soypat): Should return EOF/ErrClosed? - err = net.ErrClosed //err // Connection closed by reset. - } - return err + if h.scb.State() == StateClosed { + // TODO(soypat): Should return EOF/ErrClosed? + err = net.ErrClosed //err // Connection closed by reset. } + return err } if h.scb.State() == StateClosed { // TCB aborted, likely because it received an ACK in LastAck state. @@ -232,27 +192,12 @@ func (h *Handler) Recv(incomingPacket []byte) error { } } if segIncoming.Flags.HasAny(FlagACK) { - // Update TX ring buffer to free up acked data. - h.bufTx.RecvACK(segIncoming.ACK) - // Dup-ACK tracking per RFC 5681 §3.2 and RTO reset per RFC 6298 §5.3. - if segIncoming.ACK != prevUNA && prevUNA.LessThan(segIncoming.ACK) { - // New data acknowledged — reset RTO and dup-ACK counter. - h.rto = rtoInitial // RFC 6298 §5.3. - h.nRetx = 0 - h.dupACKs = 0 - h.lastACK = segIncoming.ACK - } else if segIncoming.ACK == h.lastACK && segIncoming.DATALEN == 0 && - !segIncoming.Flags.HasAny(FlagSYN|FlagFIN) && h.bufTx.BufferedSent() > 0 { - // Duplicate ACK per RFC 5681 §2: same ACK, no data, no SYN/FIN, - // and receiver has outstanding data. - h.dupACKs++ - if h.dupACKs == 3 { - // RFC 5681 §3.2: "After receiving 3 duplicate ACKs [...] - // TCP performs a retransmission of what appears to be the - // missing segment, without waiting for the retransmission - // timer to expire." - h.triggerRetransmit() - } + if segIncoming.ACK == prevUNA { + // scb keeping track of duplicate acks. + h.info("tcp.Handler:dupack", slog.Uint64("ndupack", uint64(h.scb.dupack)), slog.Uint64("ack", uint64(segIncoming.ACK)), slog.Uint64("lport", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort))) + } else { + // Update TX ring buffer to free up acked data. + h.bufTx.RecvACK(segIncoming.ACK) } } if segIncoming.Flags.HasAny(FlagSYN) { @@ -340,23 +285,24 @@ func (h *Handler) Send(b []byte) (int, error) { offset++ } else { var ok bool - available := min(buffered, len(b)-sizeHeaderTCP) - segment, ok = h.scb.PendingSegment(available) + maxPayload := len(b) - sizeHeaderTCP + segment, ok = h.scb.PendingSegment(maxPayload) segment.WND = Size(h.bufRx.Free()) if !ok { // No pending control segment or data to send. Yield. return 0, nil - } - if segment.DATALEN > 0 { - n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ, h.now) - if err != nil { - return 0, err - } else if n != int(segment.DATALEN) { - panic("expected n == available") - } } else if segment.Flags == synack { h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss) offset++ + } else if segment.DATALEN > 0 { + n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ) + if err != nil { + return 0, err + } + segment.DATALEN = Size(n) + if n > 0 { + segment.Flags |= FlagPSH + } } } prevState := h.scb.State() @@ -492,40 +438,6 @@ func min(a, b int) int { return b } -// SetNow sets the current time in milliseconds for retransmission timing. -// Must be called by Conn before Send/Recv operations. -func (h *Handler) SetNow(ms uint32) { h.now = ms } - -// ShouldRetransmit returns true if the retransmission timeout has expired -// on the oldest unacknowledged segment. Per RFC 6298 §5.1 and §5.4. -func (h *Handler) ShouldRetransmit() bool { - oldest := h.bufTx.slist.Oldest() - if oldest == nil { - return false - } - return h.now-oldest.sentAt >= h.rto -} - -// triggerRetransmit rewinds the transmit queue and control block so the next -// Send call retransmits from snd.UNA. Per RFC 9293 §3.10.8, RFC 6298 §5.4–5.5. -func (h *Handler) triggerRetransmit() { - // Save the high-water mark of NXT before rewinding so that cumulative ACKs - // for data sent pre-rewind can still be accepted (see Recv recovery path). - if h.retransmitNXT == 0 || h.retransmitNXT.LessThan(h.scb.snd.NXT) { - h.retransmitNXT = h.scb.snd.NXT - } - h.scb.Retransmit() - h.bufTx.RetransmitFromUNA() - // RFC 6298 §5.5: "The host MUST set RTO <- RTO * 2 ('back off the timer')." - h.nRetx++ - h.rto *= 2 - if h.rto > rtoMax { - h.rto = rtoMax - } - h.debug("tcp.Handler:retransmit", slog.Uint64("port", uint64(h.localPort)), - slog.Uint64("rto", uint64(h.rto)), slog.Uint64("nRetx", uint64(h.nRetx))) -} - func errstr(err error) string { if err == nil { return "" diff --git a/tcp/handler_retransmit_test.go b/tcp/handler_retransmit_test.go deleted file mode 100644 index 127b7ad..0000000 --- a/tcp/handler_retransmit_test.go +++ /dev/null @@ -1,674 +0,0 @@ -package tcp - -import ( - "math/rand" - "testing" -) - -// TestRTOResetsOnNewACK is a regression test for a bug where prevUNA was -// captured AFTER ControlBlock.Recv updated snd.UNA, making the "new ACK" -// condition (seg.ACK != prevUNA) always false. This caused the RTO to never -// reset per RFC 6298 §5.3, leading to exponential backoff escalation even -// when the network was healthy. -// -// The fix: capture prevUNA before calling scb.Recv in Handler.Recv. -func TestRTOResetsOnNewACK(t *testing.T) { - const mtu = 1500 - const maxpackets = 3 - rng := rand.New(rand.NewSource(100)) - client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) - setupClientServer(t, rng, client, server) - var rawbuf [mtu]byte - establish(t, client, server, rawbuf[:]) - - // Write and send data from client. - data := []byte("hello retransmit") - n, err := client.Write(data) - if err != nil { - t.Fatal("client write:", err) - } else if n != len(data) { - t.Fatal("short write") - } - - clear(rawbuf[:]) - n, err = client.Send(rawbuf[:]) - if err != nil { - t.Fatal("client send:", err) - } - - // Simulate prior retransmissions: RTO has been backed off and nRetx > 0. - client.rto = rtoInitial * 4 - client.nRetx = 2 - - // Server receives data and sends ACK. - err = server.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("server recv:", err) - } - clear(rawbuf[:]) - n, err = server.Send(rawbuf[:]) - if err != nil { - t.Fatal("server send ACK:", err) - } - if n == 0 { - t.Fatal("expected server to send ACK") - } - - // Client receives ACK — RTO and nRetx should reset. - err = client.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("client recv ACK:", err) - } - - if client.rto != rtoInitial { - t.Fatalf("BUG: RTO not reset on new ACK: got %d, want %d (RFC 6298 §5.3)", client.rto, rtoInitial) - } - if client.nRetx != 0 { - t.Fatalf("BUG: nRetx not reset on new ACK: got %d, want 0", client.nRetx) - } - if client.dupACKs != 0 { - t.Fatalf("dupACKs not reset on new ACK: got %d, want 0", client.dupACKs) - } -} - -// TestPostRetransmitACKAccepted is a regression test for a bug where after -// Retransmit() rewound snd.NXT to snd.UNA, a valid cumulative ACK from the -// remote (acknowledging data sent pre-rewind) was rejected as "acks unsent -// data" because seg.ACK > snd.NXT. -// -// The fix: in validateIncomingSegment, when snd.NXT == snd.UNA (retransmit -// active) and seg.ACK is within the send window, accept the ACK and advance -// snd.NXT to seg.ACK. -func TestPostRetransmitACKAccepted(t *testing.T) { - const mtu = 1500 - const maxpackets = 3 - rng := rand.New(rand.NewSource(200)) - client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) - setupClientServer(t, rng, client, server) - var rawbuf [mtu]byte - establish(t, client, server, rawbuf[:]) - - // Client writes and sends data. - data := []byte("data before rewind") - n, err := client.Write(data) - if err != nil { - t.Fatal("client write:", err) - } else if n != len(data) { - t.Fatal("short write") - } - clear(rawbuf[:]) - n, err = client.Send(rawbuf[:]) - if err != nil { - t.Fatal("client send:", err) - } - - // Server receives data — its next ACK will acknowledge up to the - // original snd.NXT. - err = server.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("server recv:", err) - } - - // Client triggers retransmit: snd.NXT rewound to snd.UNA. - preRewindNXT := client.scb.snd.NXT - client.triggerRetransmit() - if client.scb.snd.NXT != client.scb.snd.UNA { - t.Fatal("retransmit did not rewind snd.NXT to snd.UNA") - } - - // Server sends ACK for the data it already received. seg.ACK = preRewindNXT, - // which is > client.snd.NXT (now rewound to snd.UNA). - clear(rawbuf[:]) - n, err = server.Send(rawbuf[:]) - if err != nil { - t.Fatal("server send ACK:", err) - } - if n == 0 { - t.Fatal("expected server to send ACK") - } - - // Client receives ACK — should NOT be rejected. - err = client.Recv(rawbuf[:n]) - if err != nil { - t.Fatalf("BUG: post-retransmit ACK rejected: %v\n"+ - "After Retransmit() rewound snd.NXT to snd.UNA, the remote's cumulative\n"+ - "ACK (for data sent pre-rewind) exceeds the rewound snd.NXT and was\n"+ - "incorrectly rejected as 'acks unsent data'.", err) - } - - // snd.NXT should have advanced back to where it was before the rewind. - if client.scb.snd.NXT != preRewindNXT { - t.Fatalf("snd.NXT not restored: got %d, want %d", client.scb.snd.NXT, preRewindNXT) - } - // snd.UNA should have advanced to acknowledge the data. - if client.scb.snd.UNA != preRewindNXT { - t.Fatalf("snd.UNA not advanced: got %d, want %d", client.scb.snd.UNA, preRewindNXT) - } -} - -// TestRecoveryACKSkipsSpuriousRetransmit verifies that after fast retransmit -// rewinds snd.NXT and the client re-sends the lost segment, a cumulative ACK -// from the remote (acknowledging all data received before and after the hole) -// is accepted — even though it exceeds the rewound snd.NXT. -// -// Without this fix, lneto rejects the cumulative ACK as "acks unsent data" -// and then spuriously retransmits data that was already received by the remote. -// -// Timeline: -// 1. Client sends packets 0..N; packet 1 is lost (the "hole") -// 2. Server ACKs packet 0; sends 3 dup ACKs → fast retransmit fires -// 3. Client rewinds to snd.UNA, re-sends lost segment → snd.NXT advances by 1 MSS -// 4. Server (having received all other packets) sends cumulative ACK for ALL data -// 5. Client should accept this ACK (not reject it) and NOT send spurious retransmissions -func TestRecoveryACKSkipsSpuriousRetransmit(t *testing.T) { - const mtu = 60 // 20-byte header + 40-byte payload per packet. - const txBuf = 2048 - const maxpackets = 10 - rng := rand.New(rand.NewSource(59)) - - client := new(Handler) - server := new(Handler) - err := client.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - client.rto = rtoInitial - err = server.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - server.rto = rtoInitial - - err = server.OpenListen(uint16(rng.Uint32()), 0) - if err != nil { - t.Fatal(err) - } - err = client.OpenActive(uint16(rng.Uint32()), server.LocalPort(), 0) - if err != nil { - t.Fatal(err) - } - - var rawbuf [mtu]byte - establish(t, client, server, rawbuf[:]) - - // Send enough data to fill several packets (MSS=40). - data := make([]byte, 40*6) // 6 packets worth of data. - for i := range data { - data[i] = byte(i) - } - written := 0 - var packets [][]byte - for written < len(data) { - n, werr := client.Write(data[written:]) - if werr != nil { - t.Fatal("client write:", werr) - } - written += n - for { - clear(rawbuf[:]) - ns, serr := client.Send(rawbuf[:]) - if serr != nil { - t.Fatal("client send:", serr) - } - if ns == 0 { - break - } - packets = append(packets, append([]byte(nil), rawbuf[:ns]...)) - } - } - if len(packets) < 4 { - t.Fatalf("need at least 4 data packets, got %d", len(packets)) - } - t.Logf("sent %d data packets", len(packets)) - - // Record the sequence endpoint: this is the ACK value the server will - // send once it receives all data (including the "lost" packet). - preRewindNXT := client.scb.snd.NXT - t.Logf("pre-rewind snd.NXT=%d, snd.UNA=%d", preRewindNXT, client.scb.snd.UNA) - - // Server receives only packet 0 → ACKs it. This establishes lastACK on client. - err = server.Recv(packets[0]) - if err != nil { - t.Fatal("server recv pkt0:", err) - } - clear(rawbuf[:]) - n, err := server.Send(rawbuf[:]) - if err != nil { - t.Fatal("server send ACK:", err) - } - if n == 0 { - t.Fatal("expected server to send ACK") - } - err = client.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("client recv ACK:", err) - } - dupACKValue := client.lastACK - t.Logf("lastACK=%d after first ACK", dupACKValue) - - // Craft 3 dup ACKs (packet 1 is "lost", server keeps acking dupACKValue). - for i := 0; i < 3; i++ { - var buf [mtu]byte - frm, ferr := NewFrame(buf[:]) - if ferr != nil { - t.Fatal(ferr) - } - frm.SetSourcePort(server.LocalPort()) - frm.SetDestinationPort(client.LocalPort()) - frm.SetSegment(Segment{ - SEQ: server.scb.snd.NXT, - ACK: dupACKValue, - Flags: FlagACK, - WND: 65535, - }, 5) - rerr := client.Recv(buf[:sizeHeaderTCP]) - if rerr != nil { - t.Logf("dup ACK %d recv err (expected): %v", i+1, rerr) - } - } - if client.nRetx == 0 { - t.Fatal("fast retransmit did not fire after 3 dup ACKs") - } - t.Logf("fast retransmit fired: snd.NXT=%d, snd.UNA=%d", client.scb.snd.NXT, client.scb.snd.UNA) - - // Client re-sends the lost segment. After this, snd.NXT > snd.UNA - // (advanced by one MSS), but still < preRewindNXT. - clear(rawbuf[:]) - n, err = client.Send(rawbuf[:]) - if err != nil { - t.Fatal("client retransmit send:", err) - } - if n == 0 { - t.Fatal("expected client to send retransmit packet") - } - if client.scb.snd.NXT == client.scb.snd.UNA { - t.Fatal("expected snd.NXT to advance past snd.UNA after re-send") - } - t.Logf("after retransmit send: snd.NXT=%d, snd.UNA=%d (preRewind=%d)", - client.scb.snd.NXT, client.scb.snd.UNA, preRewindNXT) - - // Craft cumulative ACK from server for ALL data (as if server had received - // everything and the lost packet just arrived, filling the hole). - { - var buf [mtu]byte - frm, ferr := NewFrame(buf[:]) - if ferr != nil { - t.Fatal(ferr) - } - frm.SetSourcePort(server.LocalPort()) - frm.SetDestinationPort(client.LocalPort()) - frm.SetSegment(Segment{ - SEQ: server.scb.snd.NXT, - ACK: preRewindNXT, // ACKs all data sent before the rewind. - Flags: FlagACK, - WND: 65535, - }, 5) - - err = client.Recv(buf[:sizeHeaderTCP]) - if err != nil { - t.Fatalf("BUG: cumulative recovery ACK rejected: %v\n"+ - "After fast retransmit rewound snd.NXT and client re-sent one packet,\n"+ - "the remote's cumulative ACK (seg.ACK=%d) exceeds the current snd.NXT=%d\n"+ - "and is incorrectly rejected as 'acks unsent data'.\n"+ - "This causes spurious retransmissions of already-received data.", - err, preRewindNXT, client.scb.snd.NXT) - } - } - - // snd.UNA should have advanced to cover all original data. - if client.scb.snd.UNA != preRewindNXT { - t.Fatalf("snd.UNA not advanced: got %d, want %d", client.scb.snd.UNA, preRewindNXT) - } - // snd.NXT should be at least preRewindNXT. - if client.scb.snd.NXT.LessThan(preRewindNXT) { - t.Fatalf("snd.NXT behind preRewindNXT: got %d, want >= %d", client.scb.snd.NXT, preRewindNXT) - } - - // No more data should be sent — any Send() output here is a spurious retransmission. - clear(rawbuf[:]) - n, err = client.Send(rawbuf[:]) - if err != nil { - t.Fatal("client send after recovery:", err) - } - if n != 0 { - t.Fatalf("BUG: spurious retransmission after recovery ACK: sent %d bytes.\n"+ - "All data was already acknowledged by the cumulative ACK, but the client\n"+ - "still has 'unsent' data in the TX buffer that was actually received.", n) - } -} - -// TestFastRetransmitOncePerLoss is a regression test for -// https://github.com/soypat/lneto/issues/58 -// where fast retransmit was triggered multiple times for the same lost segment. -// -// When N packets are in flight and one is lost, up to N dup ACKs arrive. -// The bug: triggerRetransmit() reset dupACKs to 0, so every 3 dup ACKs -// triggered another fast retransmit of the same sequence. With 10 packets -// in flight, a single loss caused 3 retransmissions instead of 1. -// -// Per RFC 5681 §3.2, fast retransmit should fire once per loss event. -// Subsequent dup ACKs (beyond the 3rd) should NOT re-trigger it. -func TestFastRetransmitOncePerLoss(t *testing.T) { - const mtu = 60 // Small MTU: 20 byte header + 40 bytes payload per packet. - const txBuf = 2048 - const maxpackets = 10 - rng := rand.New(rand.NewSource(58)) - - client := new(Handler) - server := new(Handler) - err := client.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - client.rto = rtoInitial - err = server.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - server.rto = rtoInitial - - err = server.OpenListen(uint16(rng.Uint32()), 0) - if err != nil { - t.Fatal(err) - } - err = client.OpenActive(uint16(rng.Uint32()), server.LocalPort(), 0) - if err != nil { - t.Fatal(err) - } - - var rawbuf [mtu]byte - establish(t, client, server, rawbuf[:]) - - // With MSS=40 (mtu-20), write enough to fill several packets. - data := make([]byte, 40*maxpackets) - for i := range data { - data[i] = byte(i) - } - // Write in chunks since TX buffer may limit us. - written := 0 - var packets [][]byte - for written < len(data) { - n, werr := client.Write(data[written:]) - if werr != nil { - t.Fatal("client write:", werr) - } - written += n - // Send as many packets as possible. - for { - clear(rawbuf[:]) - ns, serr := client.Send(rawbuf[:]) - if serr != nil { - t.Fatal("client send:", serr) - } - if ns == 0 { - break - } - packets = append(packets, append([]byte(nil), rawbuf[:ns]...)) - } - } - if len(packets) < 6 { - t.Fatal("need at least 6 data packets, got", len(packets)) - } - t.Logf("sent %d data packets", len(packets)) - - // Server receives only the first packet so it ACKs it, advancing client's lastACK. - err = server.Recv(packets[0]) - if err != nil { - t.Fatal("server recv first packet:", err) - } - clear(rawbuf[:]) - n, err := server.Send(rawbuf[:]) - if err != nil { - t.Fatal("server send ACK:", err) - } - if n == 0 { - t.Fatal("expected server to send ACK") - } - // Client receives the ACK for the first packet, establishing lastACK. - err = client.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("client recv ACK:", err) - } - dupACKValue := client.lastACK - t.Logf("lastACK established at %d, client.dupACKs=%d", dupACKValue, client.dupACKs) - - // Craft 9 identical dup ACKs: same ACK value, no data, no SYN/FIN. - // These simulate what the server would send on receiving out-of-order packets. - const numDupAcks = 9 - var dupAcks [numDupAcks][]byte - for i := range dupAcks { - var buf [mtu]byte - frm, ferr := NewFrame(buf[:]) - if ferr != nil { - t.Fatal("new frame:", ferr) - } - frm.SetSourcePort(server.LocalPort()) - frm.SetDestinationPort(client.LocalPort()) - frm.SetSegment(Segment{ - SEQ: server.scb.snd.NXT, - ACK: dupACKValue, - Flags: FlagACK, - WND: 65535, - }, 5) - dupAcks[i] = append([]byte(nil), buf[:sizeHeaderTCP]...) - } - t.Logf("crafted %d dup ACKs", numDupAcks) - - // Feed all dup ACKs to client, counting how many times fast retransmit fires. - // Between dup ACKs, call Send() to transmit retransmitted packets (as a real - // stack would do). This makes BufferedSent > 0 again, which is required for - // the dup ACK condition to be met. - retransmitCount := 0 - prevNRetx := client.nRetx - for i, ack := range dupAcks[:] { - rerr := client.Recv(ack) - if rerr != nil { - continue - } - if client.nRetx > prevNRetx { - retransmitCount++ - t.Logf("fast retransmit #%d triggered at dup ACK %d (nRetx=%d)", retransmitCount, i+1, client.nRetx) - prevNRetx = client.nRetx - } - // Simulate real behavior: Send() is called between receives, - // which re-sends retransmitted data and makes BufferedSent > 0. - clear(rawbuf[:]) - client.Send(rawbuf[:]) - } - - if retransmitCount == 0 { - t.Fatal("fast retransmit never triggered (expected exactly 1)") - } - if retransmitCount > 1 { - t.Fatalf("BUG (issue #58): fast retransmit triggered %d times for a single loss event, want 1.\n"+ - "triggerRetransmit() resets dupACKs to 0, causing every 3rd dup ACK to\n"+ - "re-trigger fast retransmit for the same lost sequence.", retransmitCount) - } -} - -// TestFastRetransmitResetsOnNewACK verifies that after recovering from a loss -// event (new data ACKed), the dup-ACK counter resets so that a subsequent loss -// can trigger fast retransmit again. -// -// Without the dupACKs=0 reset on new ACK (handler.go line 219), the counter -// would stay above 3 after the first loss event and never reach ==3 again, -// disabling fast retransmit for all subsequent losses. -func TestFastRetransmitResetsOnNewACK(t *testing.T) { - const mtu = 60 // Small MTU: 20 byte header + 40 bytes payload per packet. - const txBuf = 2048 - const maxpackets = 10 - rng := rand.New(rand.NewSource(59)) - - client := new(Handler) - server := new(Handler) - err := client.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - client.rto = rtoInitial - err = server.SetBuffers(make([]byte, txBuf), make([]byte, txBuf), maxpackets) - if err != nil { - t.Fatal(err) - } - server.rto = rtoInitial - - err = server.OpenListen(uint16(rng.Uint32()), 0) - if err != nil { - t.Fatal(err) - } - err = client.OpenActive(uint16(rng.Uint32()), server.LocalPort(), 0) - if err != nil { - t.Fatal(err) - } - - var rawbuf [mtu]byte - establish(t, client, server, rawbuf[:]) - - // Helper: craft a dup ACK packet from server to client for the given ACK value. - craftDupACK := func(ackVal Value) []byte { - var buf [mtu]byte - frm, ferr := NewFrame(buf[:]) - if ferr != nil { - t.Fatal("new frame:", ferr) - } - frm.SetSourcePort(server.LocalPort()) - frm.SetDestinationPort(client.LocalPort()) - frm.SetSegment(Segment{ - SEQ: server.scb.snd.NXT, - ACK: ackVal, - Flags: FlagACK, - WND: 65535, - }, 5) - return append([]byte(nil), buf[:sizeHeaderTCP]...) - } - - // Helper: write data, send packets, return them. - sendPackets := func(payload []byte) [][]byte { - written := 0 - var pkts [][]byte - for written < len(payload) { - n, werr := client.Write(payload[written:]) - if werr != nil { - t.Fatal("client write:", werr) - } - written += n - for { - clear(rawbuf[:]) - ns, serr := client.Send(rawbuf[:]) - if serr != nil { - t.Fatal("client send:", serr) - } - if ns == 0 { - break - } - pkts = append(pkts, append([]byte(nil), rawbuf[:ns]...)) - } - } - return pkts - } - - // Helper: deliver packet to server, get ACK, deliver ACK to client. - deliverAndACK := func(pkt []byte) { - rerr := server.Recv(pkt) - if rerr != nil { - t.Fatal("server recv:", rerr) - } - clear(rawbuf[:]) - n, serr := server.Send(rawbuf[:]) - if serr != nil { - t.Fatal("server send:", serr) - } - if n == 0 { - t.Fatal("expected server to send ACK") - } - rerr = client.Recv(rawbuf[:n]) - if rerr != nil { - t.Fatal("client recv ACK:", rerr) - } - } - - // === Loss event #1 === - packets1 := sendPackets(make([]byte, 40*4)) - if len(packets1) < 4 { - t.Fatal("need at least 4 packets for loss event #1, got", len(packets1)) - } - // Deliver the first packet to establish lastACK. - deliverAndACK(packets1[0]) - ackVal1 := client.lastACK - - // Send 3 dup ACKs to trigger fast retransmit. - for i := 0; i < 3; i++ { - err = client.Recv(craftDupACK(ackVal1)) - if err != nil { - t.Fatalf("loss #1: dup ACK %d recv: %v", i+1, err) - } - clear(rawbuf[:]) - client.Send(rawbuf[:]) // Keep BufferedSent > 0. - } - if client.nRetx != 1 { - t.Fatalf("loss #1: expected nRetx=1 after 3 dup ACKs, got %d", client.nRetx) - } - t.Logf("loss #1: fast retransmit triggered (nRetx=%d)", client.nRetx) - - // === Recovery: retransmit the lost packet, server ACKs all data === - // Re-send all the packets that server missed (client retransmits from UNA). - for { - clear(rawbuf[:]) - n, serr := client.Send(rawbuf[:]) - if serr != nil { - t.Fatal("client retransmit send:", serr) - } - if n == 0 { - break - } - _ = server.Recv(rawbuf[:n]) // Deliver retransmitted + new data. - } - // Deliver remaining original packets too. - for i := 1; i < len(packets1); i++ { - _ = server.Recv(packets1[i]) - } - // Server sends cumulative ACK for all received data. - clear(rawbuf[:]) - n, serr := server.Send(rawbuf[:]) - if serr != nil { - t.Fatal("server send recovery ACK:", serr) - } - if n == 0 { - t.Fatal("expected server to send recovery ACK") - } - err = client.Recv(rawbuf[:n]) - if err != nil { - t.Fatal("client recv recovery ACK:", err) - } - t.Logf("recovery: nRetx=%d, lastACK=%d", client.nRetx, client.lastACK) - - // === Loss event #2 === - packets2 := sendPackets(make([]byte, 40*4)) - if len(packets2) < 4 { - t.Fatal("need at least 4 packets for loss event #2, got", len(packets2)) - } - // Deliver the first packet to advance lastACK. - deliverAndACK(packets2[0]) - ackVal2 := client.lastACK - if ackVal2 == ackVal1 { - t.Fatal("lastACK did not advance between loss events") - } - - // Send 3 dup ACKs — fast retransmit should trigger again. - prevNRetx := client.nRetx - for i := 0; i < 3; i++ { - err = client.Recv(craftDupACK(ackVal2)) - if err != nil { - t.Fatalf("loss #2: dup ACK %d recv: %v", i+1, err) - } - clear(rawbuf[:]) - client.Send(rawbuf[:]) // Keep BufferedSent > 0. - } - if client.nRetx <= prevNRetx { - t.Fatalf("BUG: fast retransmit did NOT trigger for loss event #2.\n"+ - "nRetx=%d (was %d).\n"+ - "The dup-ACK counter was not reset after recovery from loss event #1,\n"+ - "so it never reached the threshold again.", client.nRetx, prevNRetx) - } - t.Logf("loss #2: fast retransmit triggered (nRetx=%d)", client.nRetx) -} diff --git a/tcp/handler_test.go b/tcp/handler_test.go index 36c92f0..1ef927f 100644 --- a/tcp/handler_test.go +++ b/tcp/handler_test.go @@ -2,6 +2,7 @@ package tcp import ( "bytes" + "fmt" "math/rand" "testing" ) @@ -53,7 +54,6 @@ func newHandler(t *testing.T, mtu, mintaxpackets int) *Handler { if err != nil { t.Fatal(err) } - h.rto = rtoInitial // Fake time: now=0 and sentAt=0, so RTO never fires in tests. return h } @@ -963,3 +963,192 @@ func TestChallengeACKWithBufferedData(t *testing.T) { t.Fatal("expected data packet, got header-only") } } + +func TestHandler_RetransmitAfter3DupACKs(t *testing.T) { + const ( + mtu = 1500 + maxpackets = 3 + ) + rng := rand.New(rand.NewSource(42)) + + client := newHandler(t, mtu, maxpackets) + server := newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var pkt [mtu]byte + establish(t, client, server, pkt[:]) + + // Client sends some data in flight. + payload := []byte("0123456789") + + written, err := client.Write(payload) + if err != nil || written != len(payload) { + t.Fatalf("client.Write failed: %v len=%d", err, written) + } + + n, err := client.Send(pkt[:]) + if err != nil { + t.Fatalf("client.Send initial data: %v", err) + } + if n <= sizeHeaderTCP { + t.Fatalf("expected non-empty data packet; got %d", n) + } + // Server does NOT receive the intended packet, but rather the retransmission later on. + // no server.Recv(pkt[:n]) -> Packet loss. + + // Simulate 3 duplicate ACKs (ACK == UNA, no progress). + dup := server.scb.MakeDupACK() + if !client.scb.IncomingIsDupACK(dup.ACK) { + t.Fatal("MakeRetransmitDupACK return should be considered a duplicate ACK by remote") + } + for i := 0; i < 3; i++ { + fb, _ := NewFrame(pkt[:]) + fb.SetSourcePort(server.LocalPort()) + fb.SetDestinationPort(client.LocalPort()) + fb.SetSegment(dup, 5) + + if err := client.Recv(pkt[:sizeHeaderTCP]); err != nil { + t.Fatalf("client.Recv dupACK #%d failed: %v", i+1, err) + } + } + + if client.scb.dupack != 3 { + t.Fatalf("expected dupack=3; got %d", client.scb.dupack) + } + if !client.scb.HasPendingRetransmit() { + t.Fatal("expected HasPendingRetransmit() true after 3 dupACKs") + } + + oldUNA := client.scb.snd.UNA + n, err = client.Send(pkt[:]) + if err != nil { + t.Fatalf("client.Send retransmit failed: %v", err) + } + if n <= sizeHeaderTCP { + t.Fatalf("expected retransmit segment (>=20 bytes); got %d", n) + } else if client.scb.HasPendingRetransmit() { + t.Fatal("expected client to satisfy pending retransmit after single Send call") + } + + retransmitFrame, _ := NewFrame(pkt[:n]) + rtSeg := retransmitFrame.Segment(0) + if rtSeg.SEQ != oldUNA { + t.Fatalf("retransmit SEQ = %d; expected UNA=%d", rtSeg.SEQ, oldUNA) + } + if !rtSeg.Flags.HasAny(FlagACK) { + t.Fatalf("retransmit missing ACK flag: %#v", rtSeg.Flags) + } + if client.scb.nRetransmit != 1 { + t.Fatalf("expected scb.nRetransmit = 1; got %d", client.scb.nRetransmit) + } + + // Ensure remote side can receive the retransmit frame. + if err := server.Recv(pkt[:n]); err != nil { + t.Fatalf("server.Recv retransmit packet failed: %v", err) + } +} + +func TestHandler_RetransmitAfterMultipleLossesBothDirections(t *testing.T) { + const ( + mtu = 1500 + maxpackets = 3 + loops = 3 + ) + + rng := rand.New(rand.NewSource(1)) + client := newHandler(t, mtu, maxpackets) + server := newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var pkt [mtu]byte + establish(t, client, server, pkt[:]) + sendWithLoss := func(sender, receiver *Handler, pay []byte) { + n, err := sender.Write(pay) + if err != nil || n != len(pay) { + t.Fatalf("write failed: %v len=%d", err, n) + } + n, err = sender.Send(pkt[:]) + if err != nil { + t.Fatalf("Send initial data: %v", err) + } else if n <= sizeHeaderTCP { + t.Fatalf("expected non-empty data packet; got %d", n) + } else if sender.BufferedUnsent() > 0 { + t.Fatal("buffer too small to send all data") + } + + // Drop packet (simulate loss): NO receiver.Recv(pkt[:n]). + + // Three dupACKs from receiver side (its rcv state has not advanced). + dup := receiver.scb.MakeDupACK() + if !sender.scb.IncomingIsDupACK(dup.ACK) { + t.Fatal("dup ACK not recognized as dupack by sender") + } + for i := 0; i < 3; i++ { + clear(pkt[:]) + fb, _ := NewFrame(pkt[:]) + fb.SetSourcePort(receiver.LocalPort()) + fb.SetDestinationPort(sender.LocalPort()) + fb.SetSegment(dup, 5) + if !sender.scb.IncomingIsDupACK(dup.ACK) { + t.Fatal("expected incoming segment to be dupack") + } + if err := sender.Recv(pkt[:sizeHeaderTCP]); err != nil { + t.Fatalf("sender.Recv dupACK #%d failed: %v", i+1, err) + } + } + t.Log("dupack", sender.scb.dupack) + if sender.scb.dupack != 3 { + t.Fatalf("expected dupack=3; got=%d", sender.scb.dupack) + } else if !sender.scb.HasPendingRetransmit() { + t.Fatal("expected pending retransmit after 3 dupacks") + } + + // Now expect retransmission packet. + oldUNA := sender.scb.snd.UNA + clear(pkt[:]) + n, err = sender.Send(pkt[:]) + if err != nil { + t.Fatalf("sender.Send retransmit failed: %v", err) + } else if n <= sizeHeaderTCP { + t.Fatalf("expected retransmit packet; got %d", n) + } else if sender.scb.HasPendingRetransmit() { + t.Error("after one retransmit should be satisfied") + } + retrFrm, _ := NewFrame(pkt[:n]) + seg := retrFrm.Segment(0) + if seg.SEQ != oldUNA { + t.Fatalf("retransmit SEQ=%d; want=%d", seg.SEQ, oldUNA) + } + // Receiver consumes retransmit + if err := receiver.Recv(pkt[:n]); err != nil { + t.Fatalf("receiver.Recv retransmit failed: %v", err) + } + if receiver.scb.dupack > 0 { + t.Fatal("receiver has dupack", receiver.scb.dupack) + } + + // Receiver ACKs, so sender progresses and dupack should reset. + clear(pkt[:]) + n, err = receiver.Send(pkt[:]) + if err != nil { + t.Fatalf("receiver.Send ACK after retransmit: %v", err) + } + if n > 0 { + if err := sender.Recv(pkt[:n]); err != nil { + t.Fatalf("sender.Recv ACK after retransmit: %v", err) + } + } + if sender.scb.dupack != 0 { + t.Fatalf("expected sender.dupack reset, got %d", sender.scb.dupack) + } + } + + // Do several losses in client->server direction + for i := 0; i < loops; i++ { + payload := []byte(fmt.Sprintf("C->S loss %d", i)) + sendWithLoss(client, server, payload) + sendWithLoss(client, server, payload) + sendWithLoss(server, client, payload) + sendWithLoss(client, server, payload) + sendWithLoss(server, client, payload) + sendWithLoss(server, client, payload) + } +} diff --git a/tcp/tcp_fuzz_test.go b/tcp/tcp_fuzz_test.go new file mode 100644 index 0000000..4f52f3d --- /dev/null +++ b/tcp/tcp_fuzz_test.go @@ -0,0 +1,98 @@ +package tcp + +import ( + "testing" + + "github.com/soypat/lneto/internal" +) + +func FuzzTCPControlBlock(f *testing.F) { + const ( + mutopFlags = 1 << iota + mutopSeq + mutopAck + mutopMaxBit + ) + const mutopBits = mutopMaxBit - 1 + const ( + mutPacketBits = 4 + + mutFlags = 1 << iota + ) + const wnd = 1500 + var seed uint64 + editPkt := func(rngSeed, ops uint64, seg *Segment) { + ops &= mutopBits + if ops&mutopFlags != 0 { + seg.Flags ^= Flags(seed & uint64(FlagACK|FlagFIN|FlagRST|FlagSYN)) + } + if ops&mutopSeq != 0 { + seg.SEQ = Value(int32(seg.SEQ) + int32(int8(uint8(rngSeed>>32)))) + } + if ops&mutopAck != 0 { + seg.ACK = Value(int32(seg.ACK) + int32(int8(uint8(rngSeed>>48)))) + } + } + f.Add(seed, uint64(mutopFlags)) + f.Fuzz(func(t *testing.T, seed, op uint64) { + var tcb0, tcb1 ControlBlock + iss0 := Value(seed) + iss1 := Value(seed >> 32) + err := tcb0.Open(iss0, wnd) + if err != nil { + t.Fatal(err) + } + synseg := ClientSynSegment(iss1, wnd) + err = tcb1.Send(synseg) + if err != nil { + t.Fatal(err) + } + err = tcb0.Recv(synseg) + if err != nil { + t.Fatal(err) + } + sent := 0 + const maxpkts = 30 + pktEdits := internal.Prand64(seed) + nextOp := internal.Prand64(op) + for range maxpkts { + nextOp = internal.Prand64(nextOp) + seg0, ok := tcb0.PendingSegment(10) + if ok { + edit := pktEdits&1 != 0 + pktEdits >>= 1 + if edit { + editPkt(seed, nextOp, &seg0) + } + err = tcb0.Send(seg0) + if err == nil { + sent++ + err = tcb1.Recv(seg0) + if err != nil { + t.Fatal("packet sent from TCB0 to TCB1 failed:\n", StringExchange(seg0, tcb0.State(), tcb1.State(), false)) + } + } + } + nextOp = internal.Prand64(nextOp) + seg1, ok := tcb1.PendingSegment(10) + if ok { + edit := pktEdits&1 != 0 + pktEdits >>= 1 + if edit { + editPkt(seed, nextOp, &seg1) + } + err = tcb1.Send(seg1) + if err == nil { + sent++ + err = tcb0.Recv(seg1) + if err != nil { + t.Fatal("packet sent from TCB1 to TCB0 failed:\n", StringExchange(seg1, tcb0.State(), tcb1.State(), true)) + } + } + } + } + if sent == 0 { + t.Fatal("no packets sent") + } + }) +} diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 9f96192..386b911 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -45,9 +45,6 @@ type ringidx struct { seq Value // size is the size of the packet in bytes. size Size - // sentAt is the time in milliseconds when this packet was first sent. - // Used for RTO detection per RFC 6298 §5. - sentAt uint32 } // Reset resets the RingTx's internal state to use buf as the main ring buffer and creates or reuses @@ -118,20 +115,31 @@ func (rtx *ringTx) Write(b []byte) (n int, err error) { } // MakePacket reads from the unsent data ring buffer and generates a new packet segment. -// It fails if the sent packet queue is full. sentAt is the current time in milliseconds, -// stamped on the packet for RTO detection per RFC 6298 §5.1. -func (rtx *ringTx) MakePacket(b []byte, currentSeq Value, sentAt uint32) (int, error) { +// It fails if the sent packet queue is full. +func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { free := rtx.slist.Free() if free == 0 { return 0, lneto.ErrBufferFull } endSeq, ok := rtx.sentEndSeq() if ok && currentSeq.LessThan(endSeq) { + // maybe retransmit. Look for exact match. + for i := range rtx.slist.pkts { + pkt := &rtx.slist.pkts[i] + if pkt.seq == currentSeq { + // This packet to be retransmit. + data := rtx.ring(pkt.off, pkt.end) + return data.Read(b) + } + } internal.LogAttrs(nil, slog.LevelError, "txqueue:seq 0 && rtx.unsentend != 0 { - rtx.unsentoff = addOff(rtx.unsentoff, int(excess), size) - if rtx.unsentoff == rtx.unsentend { - rtx.unsentoff = 0 - rtx.unsentend = 0 - } - } - rtx.slist.Reset(cap(rtx.slist.pkts), ack) - rtx.consolidateBufs() -} - func (rtx *ringTx) consolidateBufs() { unsentEmpty := rtx.unsentend == 0 sentEmpty := rtx.sentend == 0 @@ -351,7 +331,7 @@ func (sl *sentlist) Free() int { return cap(sl.pkts) - len(sl.pkts) } -func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value, sentAt uint32) *ringidx { +func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx { free := sl.Free() if free == 0 { panic("pkt buffer full") @@ -361,11 +341,10 @@ func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value, sentAt uint3 panic("new sent packet offset must match last sent packet end") } sl.pkts = append(sl.pkts, ringidx{ - off: off, - end: addEnd(off, datalen, bufsize), - seq: seq, - size: Size(datalen), - sentAt: sentAt, + off: off, + end: addEnd(off, datalen, bufsize), + seq: seq, + size: Size(datalen), }) return &sl.pkts[len(sl.pkts)-1] } diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 6c90c50..f10d02c 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -84,7 +84,7 @@ func TestRingTx_op(t *testing.T) { clear(opWriteData) case opSend: // oplen=num bytes to send in this operation. - nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq, 0) + nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq) megafail := nsgot > nunsent if err != nil && oplen <= nunsent && availPkt > 0 { t.Fatal(itest, iop, err) @@ -138,17 +138,17 @@ func TestSentlist_multi(t *testing.T) { sl.Reset(3, 0) // Test multi packet x2. - p1 := sl.AddPacket(5, 0, bufsize, 0, 0) - p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq(), 0) + p1 := sl.AddPacket(5, 0, bufsize, 0) + p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq()) sl.RecvAck(Value(p2.size+p1.size), bufsize) if sl.Oldest() != nil { t.Fatal("expected full ack") } // multi packet x3. sl.Reset(3, 0) - p1 = sl.AddPacket(3, 0, bufsize, 0, 0) - p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq(), 0) - p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq(), 0) + p1 = sl.AddPacket(3, 0, bufsize, 0) + p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq()) + p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq()) sl.RecvAck(2, bufsize) oldest := sl.Oldest() if oldest != p1 { @@ -167,7 +167,7 @@ func TestSentlist_simple(t *testing.T) { // Test full ack. const bufsize = 16 const pkt = 10 - sl.AddPacket(pkt, 0, bufsize, 0, 0) + sl.AddPacket(pkt, 0, bufsize, 0) if sl.Oldest() == nil || sl.Newest() != sl.Oldest() { t.Error("expected same oldest/newest non-nil packet") } @@ -179,7 +179,7 @@ func TestSentlist_simple(t *testing.T) { } // Test partial ack. - sl.AddPacket(pkt, 0, bufsize, sl.ssn, 0) + sl.AddPacket(pkt, 0, bufsize, sl.ssn) for i := Value(0); i < pkt-1; i++ { ack++ sl.RecvAck(ack, bufsize) @@ -239,7 +239,7 @@ func TestTxQueue_multipacket(t *testing.T) { pktlen := rng.Intn(maxToPacket) + 1 pkt := rbuf[roff : roff+pktlen] expectPkt := wbuf[roff : roff+pktlen] - ngot, err := rtx.MakePacket(pkt, seq, 0) + ngot, err := rtx.MakePacket(pkt, seq) testQueueSanity(t, &rtx) roff += ngot seq += Value(ngot) @@ -374,7 +374,7 @@ func TestTxQueue(t *testing.T) { datalens = datalens[:0] for rtx.BufferedUnsent() != 0 { nbytes := rng.Intn(maxPacketSize-minBufferSize) + minBufferSize - n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq, 0) + n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq) if err != nil { t.Fatal(err) } else if n == 0 { @@ -572,7 +572,7 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new if wantRead != len(wantBufRead) { t.Fatalf("miscalculated expect read %d != %d", wantRead, len(wantBufRead)) } - n, err := rtx.MakePacket(readPacket, newPacketSeq, 0) + n, err := rtx.MakePacket(readPacket, newPacketSeq) if err != nil && wantRead != 0 { t.Errorf("error reading: %s", err) } else if n != wantRead { diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index d67ba43..f6f039a 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -80,7 +80,6 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { TxBuf: bufSpace[txOff : txOff+cfg.TxBufSize], TxPacketQueueSize: cfg.QueueSize, Logger: cfg.ConnLogger, - NanoTime: cfg.NanoTime, }) if err != nil { return nil, err