fix(xnet): allow installing a retransmission timer on TCP connections (rebased) (#199)

* fix(xnet): install a retransmission timer on TCP connections

No connection created through x/xnet had one. Neither NewTCPPool nor the
dial path in StackGo set ConnConfig.LossRecovery and ConnConfig.Nanotime,
so tcp.Handler ran with loss recovery disabled: nothing noticed a lost
segment, and the connection simply stopped: the sender waiting for an
ACK that cannot arrive, the receiver for data nobody will resend.

TCPPoolConfig.NanoTime already documents itself as "passed to each
tcp.Conn for retransmission timing (RFC 6298)", including the
time.Now fallback when it is nil; this makes that true. Each connection
gets its own tcp.RTO, which shadows that connection's send sequence space
and so cannot be shared. That is 80 bytes per connection, allocated once
at pool construction, against buffers measured in kilobytes.

The test drops exactly one data segment and requires the byte to arrive
anyway. It depends on the FIN-WAIT-1 retransmission fix, since the server
closes after writing.

* merge main fixes

* claude: fix up test to use ltesto.Sched and enable ltesto.Sched multigoro

* move test and simplify top comment

---------

Co-authored-by: Derek den Haas <i.pestano@easyflor.nl>

internal/ltesto/stackschduler.go

@@ -15,33 +15,57 @@ import (
 // from the test thread, or vice versa.
 func NewSched(t testing.TB) *Sched {
 	return &Sched{
-		t:                  t,
-		goroYieldSignal:    make(chan struct{}),
-		goroContinueSignal: make(chan struct{}),
-		finishChan:         make(chan error, 1),
-		timeout:            time.Second,
+		t:       t,
+		timeout: time.Second,
 	}
 }
 
 // Sched is the shared state behind a [SchedGoro]/[SchedDriver] pair. It exposes no
 // handoff methods directly; obtain a handle with [Sched.Goro] (for the
 // scheduled goroutine) or [Sched.Driver] (for the test thread).
+//
+// A Sched may schedule more than one goroutine: call [Sched.Goro] once per
+// goroutine and drive them as a barrier with [Sched.AwaitAllParked] and
+// [Sched.YieldToAllParked]. The single-goroutine methods ([Sched.AwaitGoroYield],
+// [Sched.AwaitGoroYieldOrDone], [Sched.YieldToGoro] and [Sched.Done]) address the
+// first handle handed out and are the right tool when there is only one.
 type Sched struct {
-	t testing.TB
+	t            testing.TB
+	goros        []*schedGoro
+	finishcalled atomic.Bool
+	timeout      time.Duration
+}
+
+// schedGoro is the per-goroutine handoff state. The channels are shared with the
+// scheduled goroutine; parked, finished and err are driver-side bookkeeping and
+// must only ever be touched from the test thread.
+type schedGoro struct {
 	// when stack backs off it signals here and waits until channel read or timeout.
-	goroYieldSignal chan struct{}
+	yieldSignal chan struct{}
 	// when main goroutine is ready for more information this channel is written to to signal waiting on stack activity.
-	goroContinueSignal chan struct{}
-	finishChan         chan error
-	finishcalled       atomic.Bool
-	coroCalls          atomic.Int32
-	timeout            time.Duration
+	continueSignal chan struct{}
+	finishChan     chan error
+
+	parked   bool // goroutine is suspended inside Yield, awaiting a continue.
+	finished bool // goroutine terminated via FinishWithErr.
+	err      error
+}
+
+// goro0 returns the first handed-out goroutine state, which the single-goroutine
+// driver methods address.
+func (ss *Sched) goro0() *schedGoro {
+	if len(ss.goros) == 0 {
+		panic("Sched.Goro must be called before driving the scheduler")
+	}
+	return ss.goros[0]
 }
 
 // AwaitGoroYield blocks until the coroutine suspends itself via [SchedGoro.Yield].
 func (ss *Sched) AwaitGoroYield() {
+	g := ss.goro0()
 	select {
-	case <-ss.goroYieldSignal:
+	case <-g.yieldSignal:
+		g.parked = true
 	case <-time.After(ss.timeout):
 		ss.t.Fatal("timeout waiting for stack to backoff")
 	}
@@ -54,10 +78,13 @@ func (ss *Sched) AwaitGoroYield() {
 // the same select, avoiding the deadlock of guessing whether the goroutine will yield
 // again. Do not mix with [Sched.Done] on the same scheduler.
 func (ss *Sched) AwaitGoroYieldOrDone() (done bool, err error) {
+	g := ss.goro0()
 	select {
-	case <-ss.goroYieldSignal:
+	case <-g.yieldSignal:
+		g.parked = true
 		return false, nil
-	case err = <-ss.finishChan:
+	case err = <-g.finishChan:
+		g.finished, g.err = true, err
 		return true, err
 	case <-time.After(ss.timeout):
 		ss.t.Fatal("timeout waiting for stack to yield or finish")
@@ -67,34 +94,103 @@ func (ss *Sched) AwaitGoroYieldOrDone() (done bool, err error) {
 
 // YieldToGoro wakes a coroutine parked in [SchedGoro.Yield], letting the goroutine run on.
 func (ss *Sched) YieldToGoro() {
+	g := ss.goro0()
 	select {
-	case ss.goroContinueSignal <- struct{}{}:
+	case g.continueSignal <- struct{}{}:
+		g.parked = false
 	case <-time.After(ss.timeout):
 		ss.t.Fatal("timeout while trying to yield to stack")
 	}
 }
 
+// AwaitAllParked blocks until every scheduled goroutine has either suspended
+// itself in [SchedGoro.Yield] or terminated via [SchedGoro.FinishWithErr]. Once it
+// returns, no scheduled goroutine is runnable, so the driver may touch state they
+// share — pumping frames between stacks, advancing a simulated clock — without
+// racing them. Pair it with [Sched.YieldToAllParked] to step the whole set.
+//
+// allFinished reports that every goroutine has terminated, which is the loop's
+// exit condition; err is the first non-nil terminal error handed over so far.
+func (ss *Sched) AwaitAllParked() (allFinished bool, err error) {
+	ss.goro0() // Panics if the scheduler has no goroutines to drive.
+	for _, g := range ss.goros {
+		if g.parked || g.finished {
+			continue // Already accounted for; waiting again would deadlock.
+		}
+		select {
+		case <-g.yieldSignal:
+			g.parked = true
+		case gerr := <-g.finishChan:
+			g.finished, g.err = true, gerr
+		case <-time.After(ss.timeout):
+			ss.t.Fatal("timeout waiting for scheduled goroutines to park or finish")
+			return true, nil
+		}
+	}
+	allFinished = true
+	for _, g := range ss.goros {
+		if !g.finished {
+			allFinished = false
+		}
+		if err == nil {
+			err = g.err
+		}
+	}
+	return allFinished, err
+}
+
+// YieldToAllParked wakes every goroutine currently parked in [SchedGoro.Yield],
+// letting them all run on until they park again. Goroutines that have already
+// terminated are skipped, so it is safe to call until [Sched.AwaitAllParked]
+// reports every goroutine finished.
+func (ss *Sched) YieldToAllParked() {
+	ss.goro0() // Panics if the scheduler has no goroutines to drive.
+	for _, g := range ss.goros {
+		if !g.parked {
+			continue
+		}
+		select {
+		case g.continueSignal <- struct{}{}:
+			g.parked = false
+		case <-time.After(ss.timeout):
+			ss.t.Fatal("timeout while trying to yield to scheduled goroutine")
+		}
+	}
+}
+
 // Done returns the channel that receives the coroutine's terminal error from
 // [SchedGoro.FinishWithErr]. It may only be called once.
 func (ss *Sched) Done() <-chan error {
+	g := ss.goro0()
 	if ss.finishcalled.CompareAndSwap(false, true) {
-		return ss.finishChan
+		return g.finishChan
 	}
 	panic("Done called twice")
 }
 
 // Goro returns the handle whose methods must be called from inside the
-// scheduled (stack) goroutine.
+// scheduled (stack) goroutine. Call it once per goroutine to be scheduled, from
+// the test thread and before those goroutines start: the handles are handed out
+// unsynchronized. The first handle is the one the single-goroutine driver methods
+// address; drive two or more with [Sched.AwaitAllParked] and [Sched.YieldToAllParked].
 func (ss *Sched) Goro() SchedGoro {
-	if !ss.coroCalls.CompareAndSwap(0, 1) {
-		panic("only one goroutine supported for now")
+	g := &schedGoro{
+		yieldSignal:    make(chan struct{}),
+		continueSignal: make(chan struct{}),
+		finishChan:     make(chan error, 1),
 	}
-	return SchedGoro{ss: ss}
+	ss.goros = append(ss.goros, g)
+	return SchedGoro{ss: ss, g: g}
 }
 
 // SchedGoro is the coroutine-side handle of a [Sched]. Every method MUST be
 // called from inside the scheduled goroutine and never from the test thread.
-type SchedGoro struct{ ss *Sched }
+// It holds its own handoff state directly so the goroutine never reads the
+// scheduler's handle list, which the test thread may still be appending to.
+type SchedGoro struct {
+	ss *Sched
+	g  *schedGoro
+}
 
 // Yield suspends the goroutine at a backoff point and parks until the driver
 // calls [SchedDriver.YieldToGoro]. Its signature satisfies [lneto.BackoffStrategy] so it
@@ -103,12 +199,12 @@ func (c SchedGoro) Yield(consecutiveBackoffs uint) time.Duration {
 	ss := c.ss
 	timeout := time.After(ss.timeout)
 	select {
-	case ss.goroYieldSignal <- struct{}{}:
+	case c.g.yieldSignal <- struct{}{}:
 	case <-timeout:
 		ss.t.Fatal("timeout backing off, possible race condition? Multiple stacks using same backoff is unexpected pattern")
 	}
 	select {
-	case <-ss.goroContinueSignal:
+	case <-c.g.continueSignal:
 	case <-timeout:
 		ss.t.Fatal("timeout waiting for continue")
 	}
@@ -119,10 +215,10 @@ func (c SchedGoro) Yield(consecutiveBackoffs uint) time.Duration {
 // channel. It must be called at most once.
 func (c SchedGoro) FinishWithErr(err error) {
 	ss := c.ss
-	if len(ss.finishChan) != 0 {
+	if len(c.g.finishChan) != 0 {
 		ss.t.Fatal("Coro.FinishWithErr can be called once only")
 	}
-	ss.finishChan <- err
+	c.g.finishChan <- err
 }
 
 // Finish is just shorthand for c.FinishWithErr(nil).

x/xnet/stack-go.go

@@ -176,13 +176,19 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 		if isDial {
 			var conn tcp.Conn
 			// DIAL TCP: active connection a.k.a TCP Client branch.
-			err = conn.Configure(tcp.ConnConfig{
+			conncfg := tcp.ConnConfig{
 				// TODO(pato): Eventually add UDP configuration. we use TCP for now for simplicity's sake.
 				TxBuf:             make([]byte, s.plcfg.TxBufSize),
 				RxBuf:             make([]byte, s.plcfg.RxBufSize),
 				TxPacketQueueSize: s.plcfg.QueueSize,
 				RWBackoff:         s.plcfg.NewBackoff(),
-			})
+			}
+			if s.plcfg.NewPolicy != nil {
+				// A dialed connection needs loss recovery as much as a pooled
+				// one. See [TCPPoolConfig.NewPolicy].
+				conncfg.Policy = s.plcfg.NewPolicy()
+			}
+			err = conn.Configure(conncfg)
 			if err != nil {
 				return nil, err
 			}

x/xnet/tcppool.go

@@ -42,8 +42,9 @@ type TCPPoolConfig struct {
 	ConnLogger *slog.Logger
 
 	// 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().
+	// Used for pool timeout tracking. If nil, defaults to time.Now().UnixNano().
+	// Retransmission timing is not driven by this clock: a [tcp.Policy] carries
+	// its own. See NewPolicy.
 	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.
@@ -56,6 +57,9 @@ type TCPPoolConfig struct {
 	// NewBackoff returns the backoff to use for every newly configured TCP connection. Must be non-nil.
 	// This should always return a static(non-method) function unless you know what you are doing.
 	NewBackoff func() lneto.BackoffStrategy
+	// NewPolicy if non-nil creates a [tcp.Policy] for each [tcp.Conn] used by the configured Listener.
+	// NewPolicy should not return reused policies unless the algorithm is stateless. See [tcp.Policy] for more information.
+	NewPolicy func() tcp.Policy
 }
 
 func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) {
@@ -88,6 +92,11 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) {
 			Logger:            cfg.ConnLogger,
 			RWBackoff:         cfg.NewBackoff(),
 		}
+		if cfg.NewPolicy != nil {
+			// One Policy per connection: it shadows that connection's send
+			// sequence space and so cannot be shared.
+			conncfg.Policy = cfg.NewPolicy()
+		}
 		err := pool.conns[i].Configure(conncfg)
 		if err != nil {
 			return nil, err

x/xnet/xnet_listener_test.go

@@ -1,12 +1,19 @@
 package xnet
 
 import (
+	"context"
+	"fmt"
+	"net"
 	"net/netip"
+	"syscall"
 	"testing"
+	"time"
 
 	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/ethernet"
+	"github.com/soypat/lneto/internal/ltesto"
 	"github.com/soypat/lneto/tcp"
+	"github.com/soypat/lneto/tcp/rto"
 )
 
 func TestStackAsyncListener_SingleConnection(t *testing.T) {
@@ -306,3 +313,191 @@ func TestListener_ResetAfterClose(t *testing.T) {
 		t.Fatalf("expected port %d after re-Reset, got %d", svPort, listener.LocalPort())
 	}
 }
+
+// TestTCPRetransmitsLostSegment drops one data segment and requires bytes to arrive anyway.
+// This in particular tests the RTO [tcp.Policy] since tcp
+// package by itself will not trigger a retransmission unless dupacks are received.
+func TestTCPRetransmitsLostSegment(t *testing.T) {
+	const (
+		MTU     = ethernet.MaxMTU
+		svPort  = 80
+		bufSize = 2 << 10
+		want    = "this segment is lost in transit"
+		// A quiet round means both sides are waiting on the network, which is
+		// what a lost segment looks like: only then does the clock move, so the
+		// RTO expires in a bounded number of rounds instead of in real time.
+		quietStep = 100 * time.Millisecond
+		maxRounds = 600
+		// Headers total 54 bytes, so a larger frame carries payload. Dropping a
+		// bare ACK would exercise the other direction's recovery instead.
+		minDataFrame = 14 + 20 + 20 + 8
+	)
+	client, sv := new(StackAsync), new(StackAsync)
+	if err := client.Reset(StackConfig{
+		Hostname:          "rtx-client",
+		RandSeed:          11,
+		StaticAddress4:    [4]byte{10, 0, 0, 90},
+		MaxActiveTCPPorts: 2,
+		HardwareAddress:   [6]byte{0xbe, 0xef, 0, 0, 0, 90},
+		MTU:               MTU,
+		ICMPQueueLimit:    2,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := sv.Reset(StackConfig{
+		Hostname:          "rtx-server",
+		RandSeed:          ^int64(11),
+		StaticAddress4:    [4]byte{10, 0, 0, 91},
+		MaxActiveTCPPorts: 2,
+		HardwareAddress:   [6]byte{0xbe, 0xef, 0, 0, 0, 91},
+		MTU:               MTU,
+		ICMPQueueLimit:    2,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	client.SetGatewayHardwareAddr(sv.HardwareAddr())
+	sv.SetGatewayHardwareAddr(client.HardwareAddr())
+
+	tsched := ltesto.NewSched(t)
+	svGoro, clGoro := tsched.Goro(), tsched.Goro()
+
+	// Simulated monotonic clock. Only the driver writes it, and only while every
+	// scheduled goroutine is parked, so it needs no synchronization of its own.
+	var now int64
+	nanotime := func() int64 { return now }
+
+	// Each side backs off into its own scheduler handle, so the driver can park
+	// and resume the two independently.
+	newPool := func(yield lneto.BackoffStrategy) TCPPoolConfig {
+		return TCPPoolConfig{
+			PoolSize: 2, QueueSize: 4,
+			TxBufSize: bufSize, RxBufSize: bufSize,
+			// Well past the simulated time this test spends, so the pool never
+			// reaps a connection out from under the retransmission.
+			EstablishedTimeout: 120 * time.Second,
+			ClosingTimeout:     120 * time.Second,
+			NanoTime:           nanotime,
+			NewBackoff:         func() lneto.BackoffStrategy { return yield },
+			NewPolicy: func() tcp.Policy {
+				timer := new(rto.Timer)
+				if err := timer.Configure(nanotime); err != nil {
+					t.Error(err)
+				}
+				return timer
+			},
+		}
+	}
+	svGo := sv.StackBlocking(svGoro.Yield).StackGo(StackGoConfig{
+		ListenerPoolConfig: newPool(svGoro.Yield),
+	})
+	clGo := client.StackBlocking(clGoro.Yield).StackGo(StackGoConfig{
+		ListenerPoolConfig: newPool(clGoro.Yield),
+		TCPDialTimeout:     60 * time.Second,
+		TCPDialRetries:     1,
+	})
+	svGo.blk._nanotime = nanotime
+	clGo.blk._nanotime = nanotime
+
+	lsAny, err := svGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM,
+		netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort), netip.AddrPort{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	listener := lsAny.(net.Listener)
+	defer listener.Close()
+
+	// dropNext arms the driver to swallow the next server→client data frame. It
+	// is handed between the server goroutine and the driver by the scheduler
+	// handoff, which orders every access to it.
+	var dropNext, dropped bool
+
+	go func() {
+		c, err := listener.Accept()
+		if err != nil {
+			svGoro.FinishWithErr(err)
+			return
+		}
+		dropNext = true // The very next data frame is lost in transit.
+		_, err = c.Write([]byte(want))
+		c.Close() // Closing here is what makes #182's FIN-WAIT-1 retransmit matter.
+		svGoro.FinishWithErr(err)
+	}()
+
+	raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort)
+	go func() {
+		cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM,
+			netip.AddrPort{}, raddr)
+		if err != nil {
+			clGoro.FinishWithErr(err)
+			return
+		}
+		conn := cAny.(net.Conn)
+		got := make([]byte, 0, len(want))
+		rb := make([]byte, 64)
+		for len(got) < len(want) {
+			n, err := conn.Read(rb)
+			got = append(got, rb[:n]...)
+			if err != nil {
+				clGoro.FinishWithErr(fmt.Errorf("read %d/%d bytes: %w", len(got), len(want), err))
+				return
+			}
+		}
+		if string(got) != want {
+			clGoro.FinishWithErr(fmt.Errorf("read %q, want %q", got, want))
+			return
+		}
+		// Closed before finishing: a Yield after FinishWithErr would never be
+		// serviced, since the driver stops resuming a goroutine it has reaped.
+		conn.Close()
+		clGoro.Finish()
+	}()
+
+	var buf [MTU + ethernet.MaxOverheadSize]byte
+	// pump moves one frame each way, dropping the armed one. Only ever called
+	// with both goroutines parked.
+	// Ingress errors are not fatal here: once a segment is dropped the frames
+	// behind it arrive past rcv.nxt and are rejected, which is precisely the
+	// stall the retransmission has to break. Egress errors are real faults.
+	pump := func() (moved bool) {
+		n, err := client.EgressEthernet(buf[:])
+		if err != nil {
+			t.Fatal("client egress:", err)
+		} else if n > 0 {
+			sv.IngressEthernet(buf[:n])
+			moved = true
+		}
+		n, err = sv.EgressEthernet(buf[:])
+		if err != nil {
+			t.Fatal("server egress:", err)
+		} else if n > 0 {
+			if dropNext && n > minDataFrame {
+				dropNext, dropped = false, true
+			} else {
+				client.IngressEthernet(buf[:n])
+			}
+			moved = true
+		}
+		return moved
+	}
+
+	for round := 0; ; round++ {
+		if round == maxRounds {
+			t.Fatalf("no retransmission after %d rounds and %v of simulated time (dropped=%v): is a Policy installed?",
+				maxRounds, time.Duration(now), dropped)
+		}
+		allFinished, err := tsched.AwaitAllParked()
+		if err != nil {
+			t.Fatalf("after losing one segment (dropped=%v): %v", dropped, err)
+		}
+		if allFinished {
+			break
+		}
+		if !pump() {
+			now += int64(quietStep) // Both sides idle: let the RTO age.
+		}
+		tsched.YieldToAllParked()
+	}
+	if !dropped {
+		t.Fatal("no frame was dropped, so the test did not exercise retransmission")
+	}
+}