Implement basic retransmit queue reset method (#52)

* implement basic retransmit queue reset method

* tcp: rto reset on new ack; accept ack after retransmit bugfix

tcp/conn.go

@@ -28,6 +28,7 @@ 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
@@ -55,6 +56,10 @@ 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) {
@@ -65,9 +70,19 @@ 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()
@@ -339,6 +354,7 @@ 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
@@ -366,6 +382,11 @@ 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()
+	}
 	n, err = conn.h.Send(carrierData[offsetToFrame:])
 	if err != nil || n == 0 {
 		return 0, err

tcp/control.go

@@ -493,11 +493,24 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) {
 		}
 
 	case established && acksUnsentData:
-		err = errDropSegment
-		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)))
+		// After Retransmit() rewinds snd.NXT to snd.UNA, the remote may ACK
+		// data it received pre-rewind — a valid cumulative ACK that exceeds
+		// the rewound snd.NXT. Detect this case (NXT==UNA means rewind active)
+		// and accept the ACK if within the send window.
+		retransmitActive := tcb.snd.NXT == tcb.snd.UNA
+		if retransmitActive && seg.ACK.InWindow(tcb.snd.UNA, tcb.snd.WND) {
+			tcb.snd.NXT = seg.ACK
+			if isDebug {
+				tcb.debug("rcv:ACK-advance-nxt", slog.String("state", tcb._state.String()),
+					slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT)))
+			}
+		} else {
+			err = errDropSegment
+			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)))
+			}
 		}
 
 	case preestablished && (acksOld || acksUnsentData):
@@ -562,6 +575,12 @@ func (tcb *ControlBlock) rstJump() Value {
 	return 100
 }
 
+// Retransmit resets snd.NXT back to snd.UNA, allowing the next PendingSegment
+// 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 }
+
 // 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.
 // An abort call prepares the connection for opening an active connection via a

tcp/handler.go

@@ -13,7 +13,6 @@ import (
 // Handler is a low level TCP handling data structure. It implements logic
 // related to data buffering, frame sequencing and connection state handling.
 // Does NOT implement IP related logic, so no CRC calculation/validation or pseudo header logic.
-// Does NOT implement connection lifetime handling, so NO deadlines, keepalives, backoffs or anything that requires use of time package.
 //
 // See [Conn] for a higher level abstraction of a TCP connection, and see [ControlBlock] for the lower level bits of a TCP connection.
 type Handler struct {
@@ -25,12 +24,24 @@ type Handler struct {
 	validator  lneto.Validator
 	localPort  uint16
 	remotePort uint16
-	// connid is a conenction counter that is incremented each time a new
-	// connection is established via Open calls. This disambiguate's whether
+	// connid is a connection counter that is incremented each time a new
+	// connection is established via Open calls. This disambiguates whether
 	// Read and Write calls belong to the current connection.
 
 	optcodec OptionCodec
 	closing  bool
+
+	// 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
+	// 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
 }
 
 func (h *Handler) SetLoggers(handler, scb *slog.Logger) {
@@ -127,11 +138,19 @@ 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 {
@@ -166,6 +185,7 @@ func (h *Handler) Recv(incomingPacket []byte) error {
 		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 {
 		if h.scb.State() == StateClosed {
@@ -191,6 +211,26 @@ 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.Flags.HasAny(FlagSYN) {
 		// Parse remote MSS from TCP options.
@@ -285,7 +325,7 @@ func (h *Handler) Send(b []byte) (int, error) {
 			return 0, nil
 		}
 		if segment.DATALEN > 0 {
-			n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ)
+			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) {
@@ -429,6 +469,36 @@ 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() {
+	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.dupACKs = 0
+	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 "<nil>"

tcp/handler_retransmit_test.go

@@ -0,0 +1,147 @@
+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)
+	}
+}

tcp/handler_test.go

@@ -53,6 +53,7 @@ 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
 }
 

tcp/txqueue.go

@@ -34,7 +34,8 @@ type ringTx struct {
 	iss       Value
 }
 
-// ringidx represents packet data inside RingTx
+// ringidx represents packet data inside RingTx.
+// Part of the retransmission queue required by RFC 9293 §3.4, §3.10.8.
 type ringidx struct {
 	// off is data start offset of packet data inside buf. Follows [internal.Ring] semantics.
 	off int
@@ -44,7 +45,9 @@ type ringidx struct {
 	seq Value
 	// size is the size of the packet in bytes.
 	size Size
-	// time is a measure of the instant of time message was sent at.
+	// 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
@@ -115,8 +118,9 @@ 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.
-func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) {
+// 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) {
 	free := rtx.slist.Free()
 	if free == 0 {
 		return 0, lneto.ErrBufferFull
@@ -137,7 +141,7 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) {
 	// Start of buffer will be SENT, end of buffer will be UNSENT(or empty).
 	// Packet generated has offset at old unsentOff.
 	size := rtx.Size()
-	pkt := rtx.slist.AddPacket(n, oldUnsentOff, size, currentSeq)
+	pkt := rtx.slist.AddPacket(n, oldUnsentOff, size, currentSeq, sentAt)
 	if pkt.off != oldUnsentOff || pkt.end != addEnd(pkt.off, n, size) {
 		panic("invalid generated packet")
 	}
@@ -203,6 +207,32 @@ func (rtx *ringTx) ring(off, end int) internal.Ring {
 // Result of addEnd will never be 0 unless arguments are (0,0).
 func (rtx *ringTx) addEnd(a, b int) int { return addEnd(a, b, len(rtx.rawbuf)) }
 
+// RetransmitFromUNA rewinds the transmit queue so that all sent-but-unacked
+// data becomes unsent again. The next MakePacket call will re-send starting
+// from snd.UNA. This is the smoltcp-style pointer-rewind approach: no extra
+// mode flag, Send() has a single code path.
+//
+// Implements "send the segment at the front of the retransmission queue"
+// per RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT).
+func (rtx *ringTx) RetransmitFromUNA() {
+	oldest := rtx.slist.Oldest()
+	if oldest == nil {
+		return // Nothing in the retransmission queue.
+	}
+	unaSeq := oldest.seq
+	if rtx.sentend != 0 {
+		// Merge sent region [sentoff, sentend) back into unsent.
+		rtx.unsentoff = rtx.sentoff
+		if rtx.unsentend == 0 {
+			rtx.unsentend = rtx.sentend
+		}
+		rtx.sentoff = 0
+		rtx.sentend = 0
+	}
+	// Clear packet metadata; sequence tracking restarts from UNA.
+	rtx.slist.Reset(cap(rtx.slist.pkts), unaSeq)
+}
+
 func (rtx *ringTx) consolidateBufs() {
 	unsentEmpty := rtx.unsentend == 0
 	sentEmpty := rtx.sentend == 0
@@ -293,7 +323,7 @@ func (sl *sentlist) Free() int {
 	return cap(sl.pkts) - len(sl.pkts)
 }
 
-func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx {
+func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value, sentAt uint32) *ringidx {
 	free := sl.Free()
 	if free == 0 {
 		panic("pkt buffer full")
@@ -303,10 +333,11 @@ func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx {
 		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),
+		off:    off,
+		end:    addEnd(off, datalen, bufsize),
+		seq:    seq,
+		size:   Size(datalen),
+		sentAt: sentAt,
 	})
 	return &sl.pkts[len(sl.pkts)-1]
 }

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)
+					nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq, 0)
 					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)
-	p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq())
+	p1 := sl.AddPacket(5, 0, bufsize, 0, 0)
+	p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq(), 0)
 	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)
-	p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq())
-	p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq())
+	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)
 	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)
+	sl.AddPacket(pkt, 0, bufsize, 0, 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)
+	sl.AddPacket(pkt, 0, bufsize, sl.ssn, 0)
 	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)
+			ngot, err := rtx.MakePacket(pkt, seq, 0)
 			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)
+						n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq, 0)
 						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)
+		n, err := rtx.MakePacket(readPacket, newPacketSeq, 0)
 		if err != nil && wantRead != 0 {
 			t.Errorf("error reading: %s", err)
 		} else if n != wantRead {

x/xnet/tcppool.go

@@ -16,11 +16,11 @@ type TCPPool struct {
 	naqcuired      int
 	conns          []tcp.Conn
 	userData       []any
-	acquiredAt     []time.Time
-	closingAt      []time.Time
-	abortedAt      []time.Time
+	acquiredAt     []int64
+	closingAt      []int64
+	abortedAt      []int64
 	nextISS        tcp.Value
-	_now           func() time.Time
+	_now           func() int64
 	estbTimeout    time.Duration
 	closingTimeout time.Duration
 	logger         *slog.Logger
@@ -39,7 +39,11 @@ type TCPPoolConfig struct {
 
 	Logger     *slog.Logger
 	ConnLogger *slog.Logger
-	Now        func() time.Time
+
+	// NanoTime returns the current monotonic time in nanoseconds.
+	// Used for pool timeout tracking and passed to each [tcp.Conn] for
+	// retransmission timing (RFC 6298). If nil, defaults to time.Now().UnixNano().
+	NanoTime func() int64
 	// EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established.
 	// If the connection does not establish in this time it will be closed by the pool.
 	EstablishedTimeout time.Duration
@@ -56,12 +60,12 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) {
 	}
 	n := cfg.PoolSize
 	pool := &TCPPool{
-		acquiredAt:     make([]time.Time, n),
-		closingAt:      make([]time.Time, n),
-		abortedAt:      make([]time.Time, n),
+		acquiredAt:     make([]int64, n),
+		closingAt:      make([]int64, n),
+		abortedAt:      make([]int64, n),
 		conns:          make([]tcp.Conn, n),
 		userData:       make([]any, n),
-		_now:           cfg.Now,
+		_now:           cfg.NanoTime,
 		estbTimeout:    cfg.EstablishedTimeout,
 		closingTimeout: cfg.ClosingTimeout,
 		logger:         cfg.Logger,
@@ -76,6 +80,7 @@ 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
@@ -98,7 +103,7 @@ func (p *TCPPool) GetTCP() (conn *tcp.Conn, userData any, SuggestedISS tcp.Value
 	defer p.mu.Unlock()
 	p.debug("TCPPool:get")
 	for i := range p.conns {
-		if p.acquiredAt[i].IsZero() {
+		if p.acquiredAt[i] == 0 {
 			p.acquiredAt[i] = p.now()
 			p.nextISS += 1000
 			p.naqcuired++
@@ -116,9 +121,9 @@ func (p *TCPPool) PutTCP(conn *tcp.Conn) {
 		if &p.conns[i] == conn {
 			// p.mu.Lock()
 			p.conns[i].Abort()
-			p.acquiredAt[i] = time.Time{}
-			p.abortedAt[i] = time.Time{}
-			p.closingAt[i] = time.Time{}
+			p.acquiredAt[i] = 0
+			p.abortedAt[i] = 0
+			p.closingAt[i] = 0
 			p.naqcuired--
 			// p.mu.Unlock()
 			return
@@ -140,7 +145,7 @@ func (p *TCPPool) CheckTimeouts() {
 		// p.mu.Lock()
 		acq := p.acquiredAt[i]
 		// p.mu.Unlock()
-		if acq.IsZero() {
+		if acq == 0 {
 			continue
 		} else if st.IsPreestablished() && p.since(acq) > p.estbTimeout {
 			// Was acquired and did not reach establishment state so we close.
@@ -148,12 +153,12 @@ func (p *TCPPool) CheckTimeouts() {
 			conn.Close()
 		} else if st.IsClosed() || st.IsClosing() {
 			// p.mu.Lock()
-			if p.closingAt[i].IsZero() {
+			if p.closingAt[i] == 0 {
 				p.closingAt[i] = p.now()
-			} else if p.abortedAt[i].IsZero() && p.since(p.closingAt[i]) > p.closingTimeout {
+			} else if p.abortedAt[i] == 0 && p.since(p.closingAt[i]) > p.closingTimeout {
 				p.abortedAt[i] = p.now()
 				conn.Abort()
-			} else if !p.abortedAt[i].IsZero() && p.since(p.abortedAt[i]) > 10*time.Second {
+			} else if p.abortedAt[i] != 0 && p.since(p.abortedAt[i]) > 10*time.Second {
 				println("connection aborted and still not returned to TCPPool")
 				println("source", conn.LocalPort(), "remote", conn.RemotePort(), "state", conn.State().String())
 			}
@@ -161,16 +166,13 @@ func (p *TCPPool) CheckTimeouts() {
 	}
 }
 
-func (p *TCPPool) since(t time.Time) time.Duration {
-	if p._now == nil {
-		return time.Since(t)
-	}
-	return p._now().Sub(t)
+func (p *TCPPool) since(t int64) time.Duration {
+	return time.Duration(p.now() - t)
 }
 
-func (p *TCPPool) now() time.Time {
+func (p *TCPPool) now() int64 {
 	if p._now == nil {
-		return time.Now()
+		return time.Now().UnixNano()
 	}
 	return p._now()
 }