From d5c4fa53c6e96be5562c5835c47b0ac0202add3f Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Wed, 4 Mar 2026 16:59:30 +0100 Subject: [PATCH] tcp: bugfix for seq not in snd/rcv.wnd error (#49) * tcp: bugfix for seq not in snd/rcv.wnd error * accept dupe ACKs for window updates and prevent decrement of UNA * add fixes for incorrect tcp functioning * remove Conn.Available* methods in favor of Conn.Free* methods due to ambiguous name --- tcp/conn.go | 15 +- tcp/control.go | 59 ++++++-- tcp/control_rcvhandlers.go | 8 +- tcp/control_test.go | 295 +++++++++++++++++++++++++++++++++++++ tcp/handler.go | 53 ++++--- tcp/handler_test.go | 20 +-- tcp/seqsbug_test.go | 173 ++++++++++++++++++++++ tcp/tcp_legacy_test.go | 28 ++-- tcp/tcp_test.go | 30 ++-- x/xnet/xnet_test.go | 4 +- 10 files changed, 600 insertions(+), 85 deletions(-) create mode 100644 tcp/control_test.go diff --git a/tcp/conn.go b/tcp/conn.go index 021ae34..164ed72 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -111,18 +111,21 @@ func (conn *Conn) BufferedUnsent() int { return conn.h.BufferedUnsent() } -func (conn *Conn) AvailableInput() int { +// FreeInput returns the number of free bytes in the socket's receive(input) buffer. +// The TCP window mechanism advertises this space to the remote peer, +// preventing the sender from transmitting more data than the buffer can hold. +func (conn *Conn) FreeInput() int { conn.mu.Lock() defer conn.mu.Unlock() - return conn.h.FreeRx() + return conn.h.FreeInput() } -// AvailableOutput returns amount of bytes available to write to output -// before [Conn.Write] returns an error due to insufficient space to store outgoing data. -func (conn *Conn) AvailableOutput() int { +// FreeOutput returns the number of free bytes in the socket's transmit(output) buffer. +// This is the amount of data that can be written via [Conn.Write] before it blocks. +func (conn *Conn) FreeOutput() int { conn.mu.Lock() defer conn.mu.Unlock() - return conn.h.AvailableOutput() + return conn.h.FreeOutput() } // OpenActive opens a connection to a remote peer with a known IP address and port combination. diff --git a/tcp/control.go b/tcp/control.go index bc2330b..4cdb09c 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -82,7 +82,10 @@ func (tcb *ControlBlock) MaxInFlightData() Size { return 0 // SYN not yet received. } unacked := Sizeof(tcb.snd.UNA, tcb.snd.NXT) - return tcb.snd.WND - unacked - 1 // TODO: is this -1 supposed to be here? + if unacked >= tcb.snd.WND { + return 0 + } + return tcb.snd.WND - unacked } // SetWindow sets the local receive window size. This represents the maximum amount of data @@ -123,8 +126,8 @@ type sendSpace struct { NXT Value // send next. This seq and up to UNA+WND-1 are allowed to be sent. Corresponds to local data. WND Size // send window defined by remote. Permitted number of local unacked octets in flight. MSS Size // maximum segment size advertised by remote peer. 0 means not set. - // WL1 Value // segment sequence number used for last window update - // WL2 Value // segment acknowledgment number used for last window update + WL1 Value // segment SEQ number of the last send-window update (RFC 9293 §3.10.7.4) + WL2 Value // segment ACK number of the last send-window update (RFC 9293 §3.10.7.4) } // inFlight returns amount of unacked bytes sent out. @@ -184,7 +187,8 @@ func (tcb *ControlBlock) HasPending() bool { return tcb.pending[0] != 0 } // It does not modify the ControlBlock state or pending segment queue. func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { if tcb.challengeAck { - tcb.challengeAck = false + // 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 } pending := tcb.pending[0] @@ -202,7 +206,7 @@ func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { _ = inFlight maxPayload := tcb.snd.maxSend() if payloadLen > int(maxPayload) { - if maxPayload == 0 && !pending.HasAny(FlagFIN|FlagRST|FlagSYN) { + if maxPayload == 0 && pending == 0 { return Segment{}, false } else if maxPayload > tcb.snd.WND { panic("seqs: bad calculation") @@ -304,8 +308,21 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) { } // We accept the segment and update TCB state. - tcb.snd.WND = seg.WND - if seg.Flags.HasAny(FlagACK) { + // RFC 9293 §3.10.7.4 step 5: update send window only when WL1/WL2 conditions allow it. + // WL1==WL2==0 is the uninitialized sentinel; the first update is always allowed so that + // connections with a remote ISS in the upper half of the uint32 space still work + // (modular LessThan would otherwise return false for 0.LessThan(largeISS)). + // Within that, duplicate ACKs (non-advancing) may only open the window, never shrink it. + wlUnset := tcb.snd.WL1 == 0 && tcb.snd.WL2 == 0 + if wlUnset || tcb.snd.WL1.LessThan(seg.SEQ) || (tcb.snd.WL1 == seg.SEQ && tcb.snd.WL2.LessThanEq(seg.ACK)) { + if tcb.snd.UNA.LessThan(seg.ACK) || seg.WND > tcb.snd.WND { + tcb.snd.WND = seg.WND + } + 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 } seglen := seg.LEN() @@ -349,9 +366,8 @@ func (tcb *ControlBlock) Send(seg Segment) error { case StateCloseWait: if hasFIN { tcb._state = StateLastAck - } else if hasACK { - newPending = finack // Queue finack. } + // No auto-queue of FIN on ACK: user must call Close() to initiate local FIN. } // Advance pending flags queue. @@ -362,6 +378,11 @@ func (tcb *ControlBlock) Send(seg Segment) error { } tcb.pending[0] |= newPending + // Sending an ACK satisfies any outstanding challenge-ACK obligation. + if tcb.challengeAck && seg.Flags.HasAny(FlagACK) { + tcb.challengeAck = false + } + // The segment is valid, we can update TCB state. seglen := seg.LEN() tcb.snd.NXT.UpdateForward(seglen) @@ -465,8 +486,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { // Special treatment of duplicate ACKs on established connection and of ACKs of unsent data. // https://www.rfc-editor.org/rfc/rfc9293.html#section-3.10.7.4-2.5.2.2.2.3.2.1 case established && acksOld && !ctlOrDataSegment: - err = errDropSegment - tcb.pending[0] &= FlagFIN // Completely ignore duplicate ACKs but do not erase fin bit. + // We don't drop packet. if isDebug { tcb.debug("rcv:ACK-dup", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.una", uint64(tcb.snd.UNA))) @@ -474,7 +494,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { case established && acksUnsentData: err = errDropSegment - tcb.pending[0] = FlagACK // Send ACK for unsent data. + tcb.pending[0] |= FlagACK // Send ACK for unsent data; |= preserves any pending FIN. if isDebug { tcb.debug("rcv:ACK-unsent", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT))) @@ -512,16 +532,23 @@ func (tcb *ControlBlock) resetRcv(localWND Size, remoteISS Value) { func (tcb *ControlBlock) handleRST(seq Value) error { tcb.debug("rcv:RST", slog.String("state", tcb._state.String())) - if tcb._state.IsPreestablished() { - // RFC 9293 §3.5.3: non-synchronized states accept RST if SEQ is in window. - // No challenge ACK for non-synchronized states. Return to LISTEN. + switch tcb._state { + case StateSynSent: + // RFC 9293 §3.10.7.2: RST in SYN-SENT aborts the active open. + tcb.Abort() + return net.ErrClosed + case StateListen: + // RFC 9293 §3.5.3: RST in LISTEN state is ignored. + return errDropSegment + case StateSynRcvd: + // RFC 9293 §3.5.3: SYN-RCVD (passive open) returns to LISTEN on RST. tcb.pending[0] = 0 tcb._state = StateListen tcb.resetSnd(tcb.snd.ISS+tcb.rstJump(), tcb.snd.WND) tcb.resetRcv(tcb.rcv.WND, 3_14159_2653^tcb.rcv.IRS) return errDropSegment } - // Synchronized states: exact match required, challenge ACK for in-window non-exact. + // Synchronized states: exact SEQ match required; challenge ACK for in-window non-exact. if seq != tcb.rcv.NXT { tcb.challengeAck = true tcb.pending[0] |= FlagACK diff --git a/tcp/control_rcvhandlers.go b/tcp/control_rcvhandlers.go index 9e7560b..a69908f 100644 --- a/tcp/control_rcvhandlers.go +++ b/tcp/control_rcvhandlers.go @@ -48,8 +48,9 @@ func (tcb *ControlBlock) rcvSynSent(seg Segment) (pending Flags, err error) { func (tcb *ControlBlock) rcvSynRcvd(seg Segment) (pending Flags, err error) { switch { - // case !seg.Flags.HasAll(FlagACK): - // err = errors.New("rcvSynRcvd: expected ACK") + case !seg.Flags.HasAll(FlagACK): + // RFC 9293 §3.10.7.4 step 5: "If the ACK bit is off, drop the segment and return." + err = errBadSegack case seg.ACK != tcb.snd.UNA+1: err = errBadSegack } @@ -69,7 +70,8 @@ func (tcb *ControlBlock) rcvEstablished(seg Segment) (pending Flags, err error) if hasFin { // See Figure 5: TCP Connection State Diagram of RFC 9293. tcb._state = StateCloseWait - tcb.pending[1] = FlagFIN // Queue FIN for after the CloseWait ACK. + // RFC 9293 §3.5: CLOSE-WAIT allows local side to continue sending. + // Do NOT auto-queue FIN here; user must call Close() explicitly. } } diff --git a/tcp/control_test.go b/tcp/control_test.go new file mode 100644 index 0000000..dc2c1b2 --- /dev/null +++ b/tcp/control_test.go @@ -0,0 +1,295 @@ +package tcp + +import ( + "testing" +) + +// TestMaxInFlightData_Underflow verifies that MaxInFlightData never underflows +// when unacked data equals or exceeds the send window. +func TestMaxInFlightData_Underflow(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + localWND Size = 1024 + ) + setup := func(wnd Size, unacked Size) ControlBlock { + var tcb ControlBlock + // snd.UNA = iss, snd.NXT = iss + unacked + tcb.HelperInitState(StateEstablished, iss, iss+Value(unacked), localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, wnd) // snd.WND = wnd + // HelperInitState sets snd.UNA = iss; unacked = snd.NXT - snd.UNA = unacked. + return tcb + } + + // unacked == snd.WND: usable = 0, but code returns 0 - 1 = 2^32-1. + t.Run("UnackedEqualsWindow", func(t *testing.T) { + const wnd Size = 200 + tcb := setup(wnd, wnd) // unacked == wnd + got := tcb.MaxInFlightData() + if got != 0 { + t.Errorf("MaxInFlightData() = %d; want 0 (unacked=%d == WND=%d, uint32 underflow via -1)", + got, Sizeof(tcb.snd.UNA, tcb.snd.NXT), tcb.snd.WND) + } + }) + + // unacked > snd.WND: window shrank; usable = 0, but code underflows. + t.Run("UnackedExceedsWindow", func(t *testing.T) { + const wnd Size = 100 + tcb := setup(wnd, 250) // unacked 250 > wnd 100 + got := tcb.MaxInFlightData() + if got != 0 { + t.Errorf("MaxInFlightData() = %d; want 0 (unacked=%d > WND=%d, must not underflow)", + got, Sizeof(tcb.snd.UNA, tcb.snd.NXT), tcb.snd.WND) + } + }) +} + +// TestMaxInFlightData_BogusMinusOne verifies that MaxInFlightData returns +// snd.WND - unacked (RFC value), not snd.WND - unacked - 1. +func TestMaxInFlightData_BogusMinusOne(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + localWND Size = 1024 + wnd Size = 100 + ) + setup := func(unacked Size) ControlBlock { + var tcb ControlBlock + tcb.HelperInitState(StateEstablished, iss, iss+Value(unacked), localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, wnd) + return tcb + } + + // Zero unacked: RFC says wnd, code returns wnd-1. + t.Run("ZeroUnacked_WantWND", func(t *testing.T) { + tcb := setup(0) + want := wnd + got := tcb.MaxInFlightData() + if got != want { + t.Errorf("MaxInFlightData() = %d; want %d (RFC: WND - unacked, no bogus -1)", got, want) + } + }) + + // unacked = 50: RFC says 50, code returns 49. + t.Run("PartialUnacked_WantWNDMinusUnacked", func(t *testing.T) { + tcb := setup(50) + want := wnd - 50 + got := tcb.MaxInFlightData() + if got != want { + t.Errorf("MaxInFlightData() = %d; want %d (RFC: WND - unacked, no bogus -1)", got, want) + } + }) +} + +// TestRecvAcksUnsent_PreservesPendingFIN verifies that receiving a segment whose +// ACK field acknowledges unsent data does not clobber other pending flags (FIN). +func TestRecvAcksUnsent_PreservesPendingFIN(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + localWND Size = 2048 + remoteWND Size = 2048 + ) + var tcb ControlBlock + // 50 bytes in flight (snd.NXT = iss+50), snd.UNA = iss. + tcb.HelperInitState(StateEstablished, iss, iss+50, localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, remoteWND) + + // Simulate a pending FIN+ACK that should not be clobbered. + tcb.pending[0] = FlagFIN | FlagACK + + // Bogus ACK: seg.ACK > snd.NXT (acks data not yet sent). + bogusACK := Segment{ + SEQ: remoteISS + 1, // == rcv.NXT + ACK: iss + 51, // > snd.NXT=iss+50 → acksUnsentData + Flags: FlagACK, + WND: remoteWND, + } + + err := tcb.Recv(bogusACK) + if !IsDroppedErr(err) { + t.Errorf("expected errDropSegment for ACK of unsent data, got: %v", err) + } + + // FIN must be preserved (|= not =). Currently FAILS: pending[0] = FlagACK only. + if !tcb.pending[0].HasAll(FlagFIN) { + t.Errorf("FIN cleared from pending[0] by ACK-unsent handler: got %s, want FlagFIN|FlagACK", + tcb.pending[0]) + } +} + +// TestHandleRST_SynSent_GoesToClosed verifies that receiving RST in SYN-SENT +// transitions to CLOSED (not LISTEN), per RFC 9293 §3.10.7.2. +func TestHandleRST_SynSent_GoesToClosed(t *testing.T) { + const ( + iss Value = 1000 + localWND Size = 2048 + ) + var tcb ControlBlock + // Active open: SYN-SENT. rcv.NXT=0, rcv.WND=localWND (SYN not yet received). + tcb.HelperInitState(StateSynSent, iss, iss+1, localWND) + + // RST with SEQ in receive window [rcv.NXT=0, 0+localWND=2048). + rst := Segment{ + SEQ: 0, // in [0, localWND) + Flags: FlagRST, + } + + err := tcb.Recv(rst) + if err == nil { + t.Fatal("RST in SYN-SENT must return an error (connection reset)") + } + + // RFC 9293 §3.10.7.2: "enter CLOSED state, delete TCB, and return." + // Bug: code sets state to StateListen instead. + if tcb.State() != StateClosed { + t.Errorf("state = %s after RST in SYN-SENT; want CLOSED (RFC 9293 §3.10.7.2)", tcb.State()) + } +} + +// TestRecvDuplicateACK_DoesNotShrinkWindow verifies that a duplicate ACK +// (ACK == snd.UNA) carrying a smaller window does not reduce snd.WND. +func TestRecvDuplicateACK_DoesNotShrinkWindow(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + localWND Size = 2048 + remoteWND Size = 1000 // initial send window + ) + var tcb ControlBlock + tcb.HelperInitState(StateEstablished, iss, iss+10, localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, remoteWND) // snd.WND = 1000 + tcb.snd.UNA = iss + 10 // all sent data acked + + // Duplicate ACK: ACK == snd.UNA (no new data acked), WND reduced to 100. + dupACK := Segment{ + SEQ: remoteISS + 1, // == rcv.NXT + ACK: iss + 10, // == snd.UNA (duplicate) + Flags: FlagACK, + WND: 100, // smaller than current snd.WND=1000 + } + + err := tcb.Recv(dupACK) + if err != nil { + t.Fatalf("duplicate ACK must be silently accepted: %v", err) + } + + // snd.WND must not shrink. Currently FAILS: snd.WND gets set to 100. + if tcb.snd.WND != remoteWND { + t.Errorf("snd.WND = %d after duplicate ACK; want %d (missing WL1/WL2 guard per RFC 9293 §3.10.7.4)", + tcb.snd.WND, remoteWND) + } +} + +// TestPendingSegment_ChallengeACK_Idempotent verifies that calling PendingSegment +// does not consume the challengeAck flag, honouring its read-only contract. +func TestPendingSegment_ChallengeACK_Idempotent(t *testing.T) { + var tcb ControlBlock + tcb.HelperInitState(StateEstablished, 100, 101, 1024) + tcb.HelperInitRcv(500, 501, 1024) + tcb.challengeAck = true + + seg1, ok1 := tcb.PendingSegment(0) + if !ok1 { + t.Fatal("PendingSegment returned !ok when challengeAck=true") + } + + // PendingSegment must not consume challengeAck (read-only contract). + // Currently FAILS: challengeAck is set to false on the first call. + if !tcb.challengeAck { + t.Error("PendingSegment cleared challengeAck flag; violates documented read-only contract") + } + + // A second call (e.g., before the segment is actually transmitted) must still succeed. + seg2, ok2 := tcb.PendingSegment(0) + if !ok2 { + t.Fatal("second PendingSegment call returned !ok; challengeAck was consumed on first call") + } + if seg1 != seg2 { + t.Errorf("PendingSegment not idempotent:\n first=%+v\nsecond=%+v", seg1, seg2) + } +} + +// TestRcvSynRcvd_NoACKFlag_DoesNotCompleteHandshake verifies that a segment +// lacking the ACK flag cannot complete the 3-way handshake even if its ACK +// field value coincidentally matches snd.UNA+1. +func TestRcvSynRcvd_NoACKFlag_DoesNotCompleteHandshake(t *testing.T) { + const ( + iss Value = 1000 + remoteISS Value = 5000 + localWND Size = 2048 + remoteWND Size = 2048 + ) + var tcb ControlBlock + // SYN-RCVD: server received SYN, sent SYN-ACK. Waiting for client ACK. + // snd.UNA=iss, snd.NXT=iss+1 (SYN-ACK consumed one seq). + // rcv.NXT=remoteISS+1 (client SYN consumed one seq). + tcb.HelperInitState(StateSynRcvd, iss, iss+1, localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, remoteWND) + + // Segment with NO ACK flag but ACK value == snd.UNA+1 (coincidentally correct). + noACKSeg := Segment{ + SEQ: remoteISS + 1, // == rcv.NXT + ACK: iss + 1, // == snd.UNA+1 (right value, wrong flag) + Flags: 0, // ACK bit NOT set + WND: remoteWND, + } + + tcb.Recv(noACKSeg) //nolint:errcheck // error value not the focus here + + // Handshake must not complete without the ACK flag. + // Currently FAILS: state becomes ESTABLISHED because the flag check is commented out. + if tcb.State() == StateEstablished { + t.Errorf("3-way handshake completed without FlagACK in SYN-RCVD; " + + "ACK flag validation is commented out (control_rcvhandlers.go:51-52)") + } +} + +// TestCloseWait_NoAutoFINBeforeUserClose verifies that entering CLOSE-WAIT after +// a remote FIN does not auto-queue a local FIN until the user calls Close(). +func TestCloseWait_NoAutoFINBeforeUserClose(t *testing.T) { + const ( + iss Value = 100 + remoteISS Value = 500 + localWND Size = 2048 + remoteWND Size = 2048 + ) + var tcb ControlBlock + tcb.HelperInitState(StateEstablished, iss, iss+1, localWND) + tcb.HelperInitRcv(remoteISS, remoteISS+1, remoteWND) + + // Remote sends FIN-ACK → we should enter CLOSE-WAIT. + finAck := Segment{ + SEQ: remoteISS + 1, // == rcv.NXT + ACK: iss + 1, // == snd.NXT + Flags: FlagFIN | FlagACK, + WND: remoteWND, + } + if err := tcb.Recv(finAck); err != nil { + t.Fatalf("recv FIN-ACK: %v", err) + } + if tcb.State() != StateCloseWait { + t.Fatalf("state = %s; want CLOSE-WAIT after receiving FIN", tcb.State()) + } + + // Retrieve and send the pending ACK for the FIN. + pendSeg, ok := tcb.PendingSegment(0) + if !ok { + t.Fatal("no pending ACK after receiving FIN") + } + if err := tcb.Send(pendSeg); err != nil { + t.Fatalf("send ACK in CLOSE-WAIT: %v", err) + } + if tcb.State() != StateCloseWait { + t.Fatalf("state = %s after sending ACK; want CLOSE-WAIT (user has not called Close())", tcb.State()) + } + + // RFC 9293 §3.5: user may still send data in CLOSE-WAIT. + // FIN must NOT be pending until the user calls Close(). + // Currently FAILS: Send(ACK) in CLOSE-WAIT auto-queues FINACK into pending[0]. + seg, hasPending := tcb.PendingSegment(0) + if hasPending && seg.Flags.HasAny(FlagFIN) { + t.Errorf("FIN auto-queued in CLOSE-WAIT before user calls Close(): pending flags=%s "+ + "(control.go:353-354 queues finack on any ACK sent in CLOSE-WAIT)", seg.Flags) + } +} diff --git a/tcp/handler.go b/tcp/handler.go index 9815ea4..31966ef 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -241,6 +241,12 @@ func (h *Handler) Send(b []byte) (int, error) { } awaitingSyn := h.AwaitingSynSend() buffered := h.bufTx.BufferedUnsent() + if h.scb.State() == StateCloseWait && !h.closing && buffered == 0 && !h.scb.HasPending() { + // Remote closed with no application data left to send: initiate our own close. + // Checked here (not in Recv) so the application can still write in CLOSE-WAIT + // before Send is called, implementing the half-close per RFC 9293 §3.5. + h.closing = true + } if !awaitingSyn && buffered == 0 && !h.closing && !h.scb.HasPending() { // Early nop short circuit. return 0, nil @@ -249,8 +255,10 @@ func (h *Handler) Send(b []byte) (int, error) { if err != nil { return 0, err } - if buffered == 0 && h.closing { - // If Close called and no more data to be sent, terminate connection! + if buffered == 0 && h.closing && (h.scb.State() != StateCloseWait || !h.scb.HasPending()) { + // If Close called and no more data to be sent, terminate connection. + // In CLOSE-WAIT: wait until the pending ACK is sent first, since scb.Close() + // overwrites pending with [FIN|ACK] (unlike ESTABLISHED which merges via bitmask). h.closing = false err = h.scb.Close() if err != nil { @@ -307,21 +315,6 @@ func (h *Handler) Send(b []byte) (int, error) { return datalen, nil } -// FreeTx returns the amount of space free in the transmit buffer. A call to [Handler.Write] with a larger buffer will fail. -func (h *Handler) FreeTx() int { - return h.bufTx.Free() -} - -// FreeRx returns the amount of space free in the receive buffer. -func (h *Handler) FreeRx() int { - return h.bufRx.Free() -} - -// SizeRx returns the size of the TCP receive ring buffer. -func (h *Handler) SizeRx() int { - return h.bufRx.Size() -} - // Write implements [io.Writer] by copying b to a internal buffer to be sent over the network on the next // [Handler.Send] call that can send data to remote peer. Use [Handler.Free] to know the maximum length the argument slice can be before erroring. func (h *Handler) Write(b []byte) (int, error) { @@ -376,24 +369,36 @@ func (h *Handler) maybeQueueWindowUpdate() { } } -// BufferedInput returns amount of bytes buffered in receive(input) buffer and ready to read -// with a [Handler.Read] call. +// SizeOutput returns the total size of the transmit ring buffer. +func (h *Handler) SizeOutput() int { + return h.bufTx.Size() +} + +// SizeInput returns the total size of the receive ring buffer. +func (h *Handler) SizeInput() int { + return h.bufRx.Size() +} + +// BufferedInput returns the number of unread bytes in the receive buffer. func (h *Handler) BufferedInput() int { return h.bufRx.Buffered() } -// BufferedUnsent returns the number of bytes in the socket's transmit(output) buffer -// that has yet to be sent. +// BufferedUnsent returns the number of written but unsent bytes in the transmit buffer. func (h *Handler) BufferedUnsent() int { return h.bufTx.BufferedUnsent() } -// AvailableOutput returns amount of bytes available to write to output -// before [Handler.Write] returns an error. -func (h *Handler) AvailableOutput() int { +// FreeOutput returns the number of free bytes in the transmit buffer. +func (h *Handler) FreeOutput() int { return h.bufTx.Free() } +// FreeInput returns the number of free bytes in the receive buffer. +func (h *Handler) FreeInput() int { + return h.bufRx.Free() +} + // 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. func (h *Handler) AwaitingSynResponse() bool { return h.remotePort != 0 && h.scb.State() == StateSynSent diff --git a/tcp/handler_test.go b/tcp/handler_test.go index c9ca46e..2be514b 100644 --- a/tcp/handler_test.go +++ b/tcp/handler_test.go @@ -266,7 +266,7 @@ func TestTxBufferFreedOnACK(t *testing.T) { establish(t, client, server, rawbuf[:]) // Record initial available space. - initialAvailable := client.AvailableOutput() + initialAvailable := client.FreeOutput() if initialAvailable == 0 { t.Fatal("expected non-zero initial available output") } @@ -284,7 +284,7 @@ func TestTxBufferFreedOnACK(t *testing.T) { } // Available space should have decreased. - afterWriteAvailable := client.AvailableOutput() + afterWriteAvailable := client.FreeOutput() if afterWriteAvailable >= initialAvailable { t.Fatalf("expected available to decrease after write: before=%d, after=%d", initialAvailable, afterWriteAvailable) @@ -303,7 +303,7 @@ func TestTxBufferFreedOnACK(t *testing.T) { // After sending, data moves from "unsent" to "sent" - available should still be reduced // until we receive an ACK. - afterSendAvailable := client.AvailableOutput() + afterSendAvailable := client.FreeOutput() // Server receives DATA. err = server.Recv(dataPacket) @@ -329,7 +329,7 @@ func TestTxBufferFreedOnACK(t *testing.T) { // THE BUG: After receiving ACK, the TX buffer should be freed. // Without the fix, AvailableOutput() stays at the post-send value. - afterAckAvailable := client.AvailableOutput() + afterAckAvailable := client.FreeOutput() if afterAckAvailable <= afterSendAvailable { t.Fatalf("BUG (issue #22): TX buffer not freed after receiving ACK\n"+ @@ -384,7 +384,7 @@ func TestWindowUpdateAfterRead(t *testing.T) { establish(t, client, server, rawbuf[:]) // Fill the server's RX buffer completely (without reading). - fillData := make([]byte, server.FreeRx()) + fillData := make([]byte, server.FreeInput()) n, err := client.Write(fillData) if err != nil { t.Fatal("client write:", err) @@ -400,8 +400,8 @@ func TestWindowUpdateAfterRead(t *testing.T) { if err != nil { t.Fatal("server recv:", err) } - if server.FreeRx() != 0 { - t.Fatalf("expected server RX buffer full, got %d free", server.FreeRx()) + if server.FreeInput() != 0 { + t.Fatalf("expected server RX buffer full, got %d free", server.FreeInput()) } // Server sends ACK — should advertise Window=0. @@ -451,7 +451,7 @@ func TestWindowUpdateAfterRead(t *testing.T) { if wnd := wndFrm.WindowSize(); wnd == 0 { t.Fatal("BUG: window update ACK still has Window=0") } - t.Logf("window update sent: Window=%d (buffer free=%d)", wndFrm.WindowSize(), server.FreeRx()) + t.Logf("window update sent: Window=%d (buffer free=%d)", wndFrm.WindowSize(), server.FreeInput()) } // TestWindowUpdateSWSAvoidance verifies that small reads that free less than @@ -486,7 +486,7 @@ func TestWindowUpdateSWSAvoidance(t *testing.T) { establish(t, client, server, rawbuf[:]) // Fill most of the server's RX buffer (leave a tiny amount free). - fillSize := server.FreeRx() - 10 + fillSize := server.FreeInput() - 10 fillData := make([]byte, fillSize) for i := range fillData { fillData[i] = byte(i) @@ -542,7 +542,7 @@ func TestWindowUpdateSWSAvoidance(t *testing.T) { t.Logf("NOTE: window update sent after small read (freed %d of %d buffer)", len(smallRead), rxBufSize) // This is acceptable if the threshold is met, but for SWS avoidance // we expect no update when the freed increment is < bufSize/2. - freeAfterRead := Size(server.FreeRx()) + freeAfterRead := Size(server.FreeInput()) if freeAfterRead < Size(rxBufSize/2) { t.Fatalf("SWS violation: window update sent when free=%d < bufSize/2=%d", freeAfterRead, rxBufSize/2) } diff --git a/tcp/seqsbug_test.go b/tcp/seqsbug_test.go index 2d7ec94..f0dfd26 100644 --- a/tcp/seqsbug_test.go +++ b/tcp/seqsbug_test.go @@ -342,6 +342,102 @@ func TestWindowReject_ChallengeACK(t *testing.T) { }) } +// TestPendingSegment_ACKSuppressedWhenWindowFull replicates the root cause of the +// "reject in/out seg: seq not in snd/rcv.wnd" regression reported after eab43c4. +// +// When inFlight >= snd.WND (send window full) and the caller has buffered TX data +// (payloadLen > 0), PendingSegment suppresses ALL segments—including pending ACKs— +// because the maxPayload==0 guard only allows FIN/RST/SYN through. +// +// Before eab43c4, maxSend() underflowed to ~4 billion when inFlight > WND, so the +// guard on line 204 never triggered and ACKs always piggybacked on data segments. +// After the underflow fix, maxSend() correctly returns 0, but this exposes the +// latent bug: ACKs are starved when the send window is full and there's TX data. +// +// The deadlock chain in production: +// 1. Publisher keeps TX buffer non-empty (payloadLen > 0 always) +// 2. Send window fills → maxSend() returns 0 +// 3. PendingSegment drops the pending ACK → remote never learns data was received +// 4. Remote retransmits with stale SEQ → "seq not in snd/rcv.wnd" rejection +// 5. (Before 3d0bc93: no challenge ACK → permanent deadlock → i/o timeout) +// +// The challenge ACK fix at 3d0bc93 masks this by allowing recovery, but the +// underlying ACK suppression still causes unnecessary retransmission delays. +func TestPendingSegment_ACKSuppressedWhenWindowFull(t *testing.T) { + const ( + localISS Value = 1000 + remoteISS Value = 5000 + dataInFlight Size = 500 + remoteWND Size = 500 // == dataInFlight, so inFlight >= WND → maxSend()=0 + localWND Size = 1024 + ) + + setup := func() ControlBlock { + var tcb ControlBlock + // snd.NXT = localISS + 1 + dataInFlight (SYN consumed 1 seq, then 500 bytes sent) + tcb.HelperInitState(StateEstablished, localISS, localISS+1+Value(dataInFlight), localWND) + // snd.WND = remoteWND (500), so inFlight(500) >= WND(500) → maxSend()=0 + tcb.HelperInitRcv(remoteISS, remoteISS+1, remoteWND) + tcb.snd.UNA = localISS + 1 // SYN acked, 500 bytes unacked + return tcb + } + + // Sanity check: maxSend() is actually 0 in our setup. + t.Run("precondition_maxSend_zero", func(t *testing.T) { + tcb := setup() + if ms := tcb.snd.maxSend(); ms != 0 { + t.Fatalf("maxSend()=%d; want 0 (test precondition broken)", ms) + } + }) + + // Control: PendingSegment(0) correctly returns the ACK when no payload requested. + t.Run("payloadLen_0_ACK_sent", func(t *testing.T) { + tcb := setup() + tcb.pending[0] |= FlagACK // Simulate pending ACK from receiving data. + seg, ok := tcb.PendingSegment(0) + if !ok { + t.Fatal("PendingSegment(0) returned !ok; pending ACK should be sent when no payload requested") + } + if !seg.Flags.HasAll(FlagACK) { + t.Errorf("segment flags = %s; want ACK", seg.Flags) + } + }) + + // THE BUG: PendingSegment(>0) suppresses the ACK when window is full. + // This is the exact scenario of a publisher with buffered TX data. + t.Run("payloadLen_gt0_ACK_must_not_be_suppressed", func(t *testing.T) { + tcb := setup() + tcb.pending[0] |= FlagACK // Simulate pending ACK from receiving data. + + // Caller has 100 bytes buffered to send, but window is full. + // PendingSegment should return the ACK with DATALEN=0 (no data, window full) + // instead of returning false and dropping the ACK entirely. + seg, ok := tcb.PendingSegment(100) + if !ok { + t.Fatal("PendingSegment(100) returned !ok; pending ACK was suppressed because " + + "maxPayload==0 guard only allows FIN/RST/SYN, not ACK. " + + "This causes the remote to never learn data was received, " + + "leading to retransmissions and eventual 'seq not in snd/rcv.wnd' rejection") + } + if !seg.Flags.HasAll(FlagACK) { + t.Errorf("segment flags = %s; want ACK", seg.Flags) + } + if seg.DATALEN != 0 { + t.Errorf("segment DATALEN = %d; want 0 (window is full, no data should be sent)", seg.DATALEN) + } + }) + + // Verify FIN is still allowed through when window is full (existing behavior). + t.Run("payloadLen_gt0_FIN_allowed", func(t *testing.T) { + tcb := setup() + tcb.pending[0] |= FlagFIN | FlagACK + _, ok := tcb.PendingSegment(100) + if !ok { + t.Fatal("PendingSegment suppressed FIN+ACK when window full; FIN must always go through") + } + }) +} + // TestSYNPreestablished_StillAllowed ensures the fix doesn't break normal SYN // processing in pre-established states (LISTEN, SYN-SENT, SYN-RCVD). func TestSYNPreestablished_StillAllowed(t *testing.T) { @@ -374,3 +470,80 @@ func TestSYNPreestablished_StillAllowed(t *testing.T) { } }) } + +func TestRecvAckUpdatesUnaCorrectly(t *testing.T) { + // Create a TCB in ESTABLISHED state with some data sent but not yet acknowledged. + tcb := &ControlBlock{ + _state: StateEstablished, + snd: sendSpace{ + UNA: 1000, // oldest unacknowledged sequence number + NXT: 2000, // next sequence number to send (1000 bytes outstanding) + ISS: 500, // initial send sequence number (not critical here) + WND: 65535, // large window to avoid window issues + }, + rcv: recvSpace{ + NXT: 3000, // any value, not used in these tests + WND: 65535, + }, + // logger can be nil or a no-op for tests + } + + // Helper to check that UNA stays at expected value after processing a segment. + checkUna := func(want Value, msg string) { + if got := tcb.snd.UNA; got != want { + t.Errorf("%s: UNA = %d, want %d", msg, got, want) + } + } + + // 1. Send an old ACK (below current UNA) – should be silently accepted but not advance UNA. + oldAckSeg := Segment{ + Flags: FlagACK, + ACK: 500, // less than UNA=1000 + WND: 65535, + SEQ: 3000, // any acceptable sequence (within rcv window) + } + err := tcb.Recv(oldAckSeg) + if err != nil { + t.Errorf("old ACK returned error: %v, want nil (silent accept)", err) + } + checkUna(1000, "after old ACK") + + // 2. Send an ACK for unsent data (beyond NXT) – should be rejected (error) and UNA unchanged. + futureAckSeg := Segment{ + Flags: FlagACK, + ACK: 2500, // > NXT=2000 + WND: 65535, + SEQ: 3000, + } + err = tcb.Recv(futureAckSeg) + if err == nil { + t.Error("ACK for unsent data returned nil, want error") + } + checkUna(1000, "after future ACK") + + // 3. Send a valid ACK that acknowledges some, but not all, outstanding data. + validAckSeg := Segment{ + Flags: FlagACK, + ACK: 1500, // between UNA and NXT + WND: 65535, + SEQ: 3000, + } + err = tcb.Recv(validAckSeg) + if err != nil { + t.Errorf("valid ACK returned error: %v, want nil", err) + } + checkUna(1500, "after valid ACK") + + // 4. Send an ACK that acknowledges exactly all outstanding data (ACK == NXT). + allAckSeg := Segment{ + Flags: FlagACK, + ACK: 2000, // == NXT + WND: 65535, + SEQ: 3000, + } + err = tcb.Recv(allAckSeg) + if err != nil { + t.Errorf("ACK == NXT returned error: %v, want nil", err) + } + checkUna(2000, "after ACK == NXT") +} diff --git a/tcp/tcp_legacy_test.go b/tcp/tcp_legacy_test.go index ec5bf71..28b1345 100644 --- a/tcp/tcp_legacy_test.go +++ b/tcp/tcp_legacy_test.go @@ -285,24 +285,26 @@ a FIN from A, acknowledges it, then later closes and sends its own FIN. */ func TestExchange_rfc9293_figure12_peerB(t *testing.T) { const issA, issB, windowA, windowB = 100, 300, 1000, 1000 - // Note: After B sends an ACK in CLOSE-WAIT, the implementation auto-queues FIN|ACK. - // This is an optimization that combines steps 3 and 4 of RFC 9293 Figure 12. - exchangeB := []tcp.Exchange{ - 0: { // B receives FIN|ACK from A, goes to CLOSE-WAIT with pending ACK. + // RFC 9293 Figure 12 steps 2-3: B receives FIN|ACK, then sends back ACK. + // B remains in CLOSE-WAIT, able to keep sending (RFC 9293 §3.5). + exchangeBeforeClose := []tcp.Exchange{ + 0: { // Step 2: B receives FIN|ACK from A, goes to CLOSE-WAIT with pending ACK. Incoming: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, WantState: tcp.StateCloseWait, WantPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, }, - 1: { // B sends ACK to A. Implementation auto-queues FIN|ACK for close. - Outgoing: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateCloseWait, - WantPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + 1: { // Step 3: B sends ACK to A. B remains in CLOSE-WAIT. + Outgoing: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateCloseWait, }, - 2: { // B sends FIN|ACK to A, goes to LAST-ACK. + } + // RFC 9293 Figure 12 step 4: B calls Close(). Queues FIN|ACK, goes to LAST-ACK. + exchangeAfterClose := []tcp.Exchange{ + 0: { // Step 4: B sends FIN|ACK to A, goes to LAST-ACK. Outgoing: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, WantState: tcp.StateLastAck, }, - 3: { // B receives final ACK from A, goes to CLOSED. + 1: { // Step 5: B receives final ACK from A, goes to CLOSED. Incoming: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, WantState: tcp.StateClosed, }, @@ -310,7 +312,11 @@ func TestExchange_rfc9293_figure12_peerB(t *testing.T) { var tcbB tcp.ControlBlock tcbB.HelperInitState(tcp.StateEstablished, issB, issB, windowB) tcbB.HelperInitRcv(issA, issA, windowA) - tcbB.HelperExchange(t, exchangeB) + tcbB.HelperExchange(t, exchangeBeforeClose) + if err := tcbB.Close(); err != nil { // Step 4: (Close) from RFC Figure 12. + t.Fatal("close:", err) + } + tcbB.HelperExchange(t, exchangeAfterClose) } /* diff --git a/tcp/tcp_test.go b/tcp/tcp_test.go index 8452573..7d583f5 100644 --- a/tcp/tcp_test.go +++ b/tcp/tcp_test.go @@ -34,16 +34,15 @@ func TestExchangeTest_PassiveClose_FINACKRegression(t *testing.T) { BState: tcp.StateCloseWait, BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, }, - 1: { // B sends ACK to A. Auto-queues FIN|ACK in CLOSE-WAIT. - Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, - Action: tcp.StepBSends, - AState: tcp.StateFinWait2, - BState: tcp.StateCloseWait, - BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + 1: { // B sends ACK to A (RFC 9293 Figure 12 step 3). B stays in CLOSE-WAIT. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateFinWait2, + BState: tcp.StateCloseWait, }, - 2: { // B calls Close(). Goes to LAST-ACK. Pending must be FIN|ACK combined. - // This is the regression check: Close() must NOT overwrite the auto-queued - // FIN|ACK with [FlagFIN, FlagACK] in separate pending slots. + 2: { // B calls Close() (RFC 9293 Figure 12 step 4). Goes to LAST-ACK. + // Regression check: Close() must queue FIN|ACK as a single combined flag + // in pending[0], not as [FlagFIN, FlagACK] in separate pending slots. Action: tcp.StepBCloses, AState: tcp.StateFinWait2, // A unchanged. BState: tcp.StateLastAck, @@ -79,7 +78,7 @@ func TestExchangeTest_figure12(t *testing.T) { InitStateA: tcp.StateEstablished, InitStateB: tcp.StateEstablished, Steps: []tcp.SegmentStep{ - 0: { // A sends FIN|ACK to B. + 0: { // A sends FIN|ACK to B (RFC 9293 Figure 12 step 2). Seg: tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, Action: tcp.StepASends, AState: tcp.StateFinWait1, @@ -87,15 +86,20 @@ func TestExchangeTest_figure12(t *testing.T) { APending: nil, BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, }, - 1: { // B sends ACK to A. (Auto-queues FIN|ACK in CLOSE-WAIT) + 1: { // B sends ACK to A (RFC 9293 Figure 12 step 3). B stays in CLOSE-WAIT. Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, Action: tcp.StepBSends, AState: tcp.StateFinWait2, BState: tcp.StateCloseWait, APending: &tcp.Segment{SEQ: issA + 1, ACK: issB, Flags: tcp.FlagACK, WND: windowA}, // TODO: should be nil? + }, + 2: { // B calls Close() (RFC 9293 Figure 12 step 4). B goes to LAST-ACK. + Action: tcp.StepBCloses, + AState: tcp.StateFinWait2, + BState: tcp.StateLastAck, BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, }, - 2: { // B sends FIN|ACK to A. + 3: { // B sends FIN|ACK to A (RFC 9293 Figure 12 step 4 cont). Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, Action: tcp.StepBSends, AState: tcp.StateTimeWait, @@ -103,7 +107,7 @@ func TestExchangeTest_figure12(t *testing.T) { APending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, BPending: nil, }, - 3: { // A sends final ACK to B. + 4: { // A sends final ACK to B (RFC 9293 Figure 12 step 5). Seg: tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, Action: tcp.StepASends, AState: tcp.StateTimeWait, diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index 1368e19..d4477a1 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -287,8 +287,8 @@ func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) { func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync, srcConn, dstConn *tcp.Conn, sendData []byte) { t := tst.t t.Helper() - availTx := srcConn.AvailableOutput() - availRx := dstConn.AvailableInput() + availTx := srcConn.FreeOutput() + availRx := dstConn.FreeInput() if availTx < len(sendData) { t.Fatal("insufficient space for write call", availTx, len(sendData)) } else if len(sendData) <= 0 {