xnet: debug packet before ingressing (#192)

* ci: laxify codecoverage

* reread coverage docs

* try this yaml out

* add project level codecov informational

* add a project wide coverage

* test project wide coverage status fail

* remove test code

.codecov.yml

@@ -1,3 +1,12 @@
+coverage:
+  status:
+    project:
+      default:
+        target: 62% # Ensure we don't accumulate too much debt.
+    patch:
+      default:
+        informational: true # Shows the patch metric but never turns red/fails the PR
+
 ignore:
   - "examples/**"
   - "**/stringers.go"

internal/ring.go

@@ -157,7 +157,7 @@ func (r *Ring) ReadDiscard(n int) error {
 	case n > buffered:
 		return errDiscardExceeds
 	case n == buffered:
-		r.Reset()
+		r.emptied()
 	case n+r.Off > len(r.Buf):
 		r.Off = n - (len(r.Buf) - r.Off)
 	default:
@@ -224,6 +224,18 @@ func (r *Ring) Reset() {
 	r.End = 0
 }
 
+// emptied marks the ring empty keeping the write position where it is, unlike
+// [Ring.Reset] which rewinds it to index 0. Bytes staged past that position with
+// [Ring.PeekWrite] are addressed relative to it, so moving it makes a later
+// [Ring.Commit] hand back the wrong bytes.
+func (r *Ring) emptied() {
+	off := r.End
+	if off == len(r.Buf) {
+		off = 0 // Tail exhausted, next write wraps.
+	}
+	r.Off, r.End = off, 0
+}
+
 // Size returns the capacity of the ring buffer.
 func (r *Ring) Size() int {
 	return len(r.Buf)
@@ -306,7 +318,7 @@ func (r *Ring) onReadEnd(totalRead int) {
 	}
 	newOff := r.addOff(r.Off, totalRead)
 	if newOff == r.End {
-		r.Reset()
+		r.emptied()
 	} else if newOff == len(r.Buf) {
 		r.Off = 0 // Optimization case.
 	} else {

internal/ring_test.go

@@ -661,3 +661,40 @@ func TestRingPeekWriteRejects(t *testing.T) {
 		t.Error("Commit beyond free must error")
 	}
 }
+
+// TestRingPeekWriteSurvivesEmptyRead checks bytes staged with [Ring.PeekWrite]
+// survive the ring being read empty, an event the stager does not control.
+func TestRingPeekWriteSurvivesEmptyRead(t *testing.T) {
+	r := &Ring{Buf: make([]byte, 16)}
+	if _, err := r.Write([]byte("AAAA")); err != nil {
+		t.Fatal("first write:", err)
+	}
+	// Stage "CCCC" one 4-byte gap past the write position.
+	if !r.PeekWrite([]byte("CCCC"), 4) {
+		t.Fatal("PeekWrite should fit")
+	}
+	// Drain everything readable: ring goes empty, staged bytes still pending.
+	got := make([]byte, 16)
+	n, err := r.Read(got)
+	if err != nil || string(got[:n]) != "AAAA" {
+		t.Fatalf("drain read %q (%v), want AAAA", got[:n], err)
+	}
+	if !r.IsEmpty() {
+		t.Fatal("ring should be empty after draining")
+	}
+	// Fill the gap and commit the staged tail.
+	if _, err := r.Write([]byte("BBBB")); err != nil {
+		t.Fatal("gap write:", err)
+	}
+	if err := r.Commit(4); err != nil {
+		t.Fatal("commit:", err)
+	}
+	n, err = r.Read(got)
+	if err != nil {
+		t.Fatal("read:", err)
+	}
+	if string(got[:n]) != "BBBBCCCC" {
+		t.Fatalf("read %q, want BBBBCCCC: the staged bytes were committed from the wrong offset", got[:n])
+	}
+	testRingSanity(t, r)
+}

tcp/control.go

@@ -279,10 +279,9 @@ func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) {
 		// 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
 	}
-	established := tcb._state == StateEstablished
+	canSendData := tcb._state.txQueuedDataOpen()
-	canSendData := established || tcb._state == StateCloseWait
 	if !canSendData {
-		payloadLen = 0 // Can't send data if not established or close-wait.
+		payloadLen = 0 // No send-buffer data may go out in this state.
 	}
 	if pending == 0 && payloadLen == 0 {
 		return Segment{}, false // No pending segment.
@@ -522,8 +521,12 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) {
 			err = errSeqNotInWindow
 		}
 
-	case seg.DATALEN > 0 && (tcb._state == StateFinWait1 || tcb._state == StateFinWait2):
+	case seg.DATALEN > 0 && tcb._state == StateFinWait2:
-		err = errConnectionClosing // Case 1: No further SENDs from the user will be accepted by the TCP implementation.
+		// FIN-WAIT-2 means our FIN was acknowledged, so no data below it can be
+		// unacknowledged and data here is a caller error. FIN-WAIT-1 is excluded:
+		// its FIN sits above data the peer may still be missing, which must go out
+		// for either side to make progress (RFC 9293 §3.10.8).
+		err = errConnectionClosing
 
 	case checkSeq && tcb.snd.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.snd.NXT:
 		err = errZeroWindow

tcp/definitions.go

@@ -329,6 +329,14 @@ func (s State) TxDataOpen() bool {
 	return s == StateEstablished || s == StateCloseWait
 }
 
+// txQueuedDataOpen returns true if already-queued send-buffer data may still be
+// put on the wire. It stays true after a local close, where the FIN occupies a
+// sequence above data the peer has not acknowledged: until that data is
+// (re)transmitted the peer cannot reach the FIN. RFC 9293 §3.10.8.
+func (s State) txQueuedDataOpen() bool {
+	return s.TxDataOpen() || s == StateFinWait1 || s == StateClosing || s == StateLastAck
+}
+
 // RxDataOpen returns true if the state allows the receiving of incoming data segments.
 // Combine with [State.IsPreestablished] to know whether there is no more data to be received over the network.
 func (s State) RxDataOpen() bool {

tcp/reorderstream_test.go

@@ -0,0 +1,122 @@
+package tcp
+
+import (
+	"math/rand"
+	"strconv"
+	"testing"
+
+	"github.com/soypat/lneto/ethernet"
+)
+
+// TestHandlerStreamIntegrityUnderReorder asserts byte identity of a reassembled
+// stream whose segments arrive out of order: arrival order is randomised within
+// each block of shuffleWindow segments, and nothing is lost or retransmitted, so
+// several segments sit staged in the receive ring at once. Reordering may cost
+// throughput; it may not change the bytes.
+func TestHandlerStreamIntegrityUnderReorder(t *testing.T) {
+	const (
+		mtu           = ethernet.MaxMTU
+		maxpackets    = 8
+		segSize       = 100
+		nsegs         = 8  // per round; 800 bytes through a 1500-byte ring
+		rounds        = 40 // enough for the ring to wrap many times
+		shuffleWindow = 4  // segments that may arrive in any order among themselves
+	)
+	rng := rand.New(rand.NewSource(3))
+	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
+	setupClientServer(t, rng, client, server)
+	var rawbuf [mtu]byte
+	establish(t, client, server, rawbuf[:])
+
+	var want, got []byte
+	rb := make([]byte, mtu)
+	letter := byte('A')
+	for round := range rounds {
+		// Capture this round's segments on the wire, one segment per write.
+		segs := make([][]byte, 0, nsegs)
+		for range nsegs {
+			payload := make([]byte, segSize)
+			for j := range payload {
+				payload[j] = letter
+			}
+			letter++
+			if letter > 'Z' {
+				letter = 'A'
+			}
+			if n, err := client.Write(payload); err != nil || n != segSize {
+				t.Fatalf("round %d: client write: %d %v", round, n, err)
+			}
+			clear(rawbuf[:])
+			n, err := client.Send(rawbuf[:])
+			if err != nil {
+				t.Fatalf("round %d: client send: %v", round, err)
+			}
+			segs = append(segs, append([]byte(nil), rawbuf[:n]...))
+			want = append(want, payload...)
+		}
+
+		order := make([]int, 0, nsegs)
+		for i := 0; i < nsegs; i += shuffleWindow {
+			block := make([]int, 0, shuffleWindow)
+			for j := i; j < min(i+shuffleWindow, nsegs); j++ {
+				block = append(block, j)
+			}
+			rng.Shuffle(len(block), func(a, b int) { block[a], block[b] = block[b], block[a] })
+			order = append(order, block...)
+		}
+
+		for _, idx := range order {
+			if err := server.Recv(append([]byte(nil), segs[idx]...)); err != nil {
+				t.Logf("round %d segment %d refused: %v", round, idx, err)
+			}
+			// Drain as an application would, keeping the ring from filling.
+			for {
+				n, err := server.Read(rb)
+				if n > 0 {
+					got = append(got, rb[:n]...)
+				}
+				if n == 0 || err != nil {
+					break
+				}
+			}
+			// Feed ACKs back so the sender's window keeps opening; without this
+			// the test stalls on flow control instead of exercising reassembly.
+			clear(rawbuf[:])
+			if n, err := server.Send(rawbuf[:]); err == nil && n > 0 {
+				client.Recv(rawbuf[:n])
+			}
+		}
+
+		if string(got) != string(want) {
+			// Report the first divergence; later rounds only add noise.
+			i := 0
+			for i < len(got) && i < len(want) && got[i] == want[i] {
+				i++
+			}
+			t.Errorf("stream diverges in round %d at byte %d of %d; arrival order %v",
+				round, i, len(want), order)
+			lo := max(0, i-200)
+			t.Errorf("got  %s", summarizeRuns(got[lo:min(len(got), i+200)]))
+			t.Errorf("want %s", summarizeRuns(want[lo:min(len(want), i+200)]))
+			t.FailNow()
+		}
+	}
+	t.Logf("%d bytes intact across %d rounds of reordering (window %d)", len(got), rounds, shuffleWindow)
+}
+
+// summarizeRuns renders a byte stream as run-length pairs ("A*100 B*100") so a
+// duplicated or missing segment is visible at a glance.
+func summarizeRuns(b []byte) string {
+	out := make([]byte, 0, 64)
+	for i := 0; i < len(b); {
+		j := i
+		for j < len(b) && b[j] == b[i] {
+			j++
+		}
+		out = append(out, b[i], '*')
+		out = append(out, strconv.Itoa(j-i)...)
+		out = append(out, ' ')
+		i = j
+	}
+	return string(out)
+}

tcp/rtointegration_test.go

@@ -0,0 +1,120 @@
+package tcp
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/soypat/lneto/ethernet"
+)
+
+// TestHandlerRetransmitsAfterRTO covers the seam between a Handler and its
+// LossRecovery, which the RTO unit tests do not: a lost data segment must be
+// resent once the timer expires, with nothing arriving to prompt it.
+func TestHandlerRetransmitsAfterRTO(t *testing.T) {
+	const mtu = ethernet.MaxMTU
+	const maxpackets = 4
+	rng := rand.New(rand.NewSource(5))
+	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
+
+	var now int64 // injected monotonic clock, in nanoseconds
+	client.SetLossRecovery(new(RTO), func() int64 { return now })
+
+	setupClientServer(t, rng, client, server)
+	var rawbuf [mtu]byte
+	establish(t, client, server, rawbuf[:])
+
+	data := []byte("hello")
+	if n, err := client.Write(data); err != nil || n != len(data) {
+		t.Fatal("client write:", n, err)
+	}
+	clear(rawbuf[:])
+	n, err := client.Send(rawbuf[:])
+	if err != nil || n == 0 {
+		t.Fatal("client send:", n, err)
+	}
+	// That frame is lost: it is never handed to the server.
+
+	// Nothing may come back before the timer expires.
+	var probe [mtu]byte
+	if n, err := client.Send(probe[:]); err != nil || n != 0 {
+		t.Fatalf("client sent %d bytes before the RTO expired (err %v)", n, err)
+	}
+
+	now += int64(3 * time.Second) // past the initial RTO and one backoff
+
+	clear(probe[:])
+	n, err = client.Send(probe[:])
+	if err != nil {
+		t.Fatal("client send after RTO:", err)
+	}
+	if n == 0 {
+		t.Fatal("no retransmission after the RTO expired: the loss-recovery directive is never applied")
+	}
+	if err := server.Recv(probe[:n]); err != nil {
+		t.Fatal("server refused the retransmission:", err)
+	}
+	got := make([]byte, 16)
+	nr, err := server.Read(got)
+	if err != nil || string(got[:nr]) != string(data) {
+		t.Fatalf("server read %q (%v), want %q", got[:nr], err, data)
+	}
+}
+
+// TestHandlerRetransmitsAfterCloseWithUnackedData is the write-then-close case
+// every server performs. With the last data segment lost, the FIN behind it sits
+// above a gap the peer cannot cross, so FIN-WAIT-1 must still retransmit that
+// data or both sides wait forever.
+func TestHandlerRetransmitsAfterCloseWithUnackedData(t *testing.T) {
+	const mtu = ethernet.MaxMTU
+	const maxpackets = 4
+	rng := rand.New(rand.NewSource(9))
+	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
+
+	var now int64
+	client.SetLossRecovery(new(RTO), func() int64 { return now })
+
+	setupClientServer(t, rng, client, server)
+	var rawbuf [mtu]byte
+	establish(t, client, server, rawbuf[:])
+
+	data := []byte("last response bytes")
+	if n, err := client.Write(data); err != nil || n != len(data) {
+		t.Fatal("client write:", n, err)
+	}
+	clear(rawbuf[:])
+	n, err := client.Send(rawbuf[:]) // this frame is lost in transit
+	if err != nil || n == 0 {
+		t.Fatal("client send:", n, err)
+	}
+
+	// The application closes right after writing.
+	if err := client.Close(); err != nil {
+		t.Fatal("client close:", err)
+	}
+	var finbuf [mtu]byte
+	nfin, err := client.Send(finbuf[:]) // FIN (also lost, or simply unacked)
+	if err != nil {
+		t.Fatal("client send FIN:", err)
+	}
+	t.Logf("state after close: %s (FIN frame %d bytes)", client.State(), nfin)
+
+	now += int64(3 * time.Second) // past the RTO
+
+	var probe [mtu]byte
+	n, err = client.Send(probe[:])
+	if err != nil {
+		t.Fatal("client send after RTO:", err)
+	}
+	if n == 0 {
+		t.Fatalf("no retransmission in %s: unacknowledged data is stranded by the close", client.State())
+	}
+	if err := server.Recv(probe[:n]); err != nil {
+		t.Fatal("server refused the retransmission:", err)
+	}
+	got := make([]byte, 32)
+	nr, err := server.Read(got)
+	if err != nil || string(got[:nr]) != string(data) {
+		t.Fatalf("server read %q (%v), want %q", got[:nr], err, data)
+	}
+}

x/xnet/stack-async.go

@@ -53,6 +53,10 @@ type StackAsync struct {
 	lookup  dns.Message
 	dnssv   netip.Addr
 
+	// ephPort drives sequential ephemeral-port allocation (see
+	// [StackAsync.ephemeralPort]); zero means not yet seeded.
+	ephPort uint32
+
 	ntpUDP internet.StackUDPPort
 	ntp    ntp.Client
 
@@ -125,8 +129,8 @@ func (s *StackAsync) IngressEthernet(ethernetFrame []byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.stats.TotalReceived += uint64(len(ethernetFrame))
-	err := s.link.Demux(ethernetFrame, 0)
 	debugPacket("IN ", ethernetFrame)
+	err := s.link.Demux(ethernetFrame, 0)
 	if err == nil {
 		s.arpt.learnFromIngressEthernet(ethernetFrame)
 	}
@@ -352,6 +356,23 @@ func (s *StackAsync) Prand32() (randval uint32) {
 	return randval
 }
 
+// ephemeralPort returns the next port of the IANA dynamic range (49152-65535,
+// RFC 6335 §6), allocated sequentially from a random per-stack start so a port is
+// revisited only after the full 16384-port cycle. Random selection instead reuses
+// a recent port at birthday-paradox rates, and a reused 4-tuple can collide with
+// state the previous conversation left behind (a TIME-WAIT, a NAT flow entry)
+// which swallows the new SYN.
+func (s *StackAsync) ephemeralPort() uint16 {
+	s.mu.Lock()
+	if s.ephPort == 0 {
+		s.ephPort = s.prand32()%16384 | 1
+	}
+	port := 49152 + s.ephPort%16384
+	s.ephPort++
+	s.mu.Unlock()
+	return uint16(port)
+}
+
 func (s *StackAsync) prand32() uint32 {
 	/* Algorithm "xor" from p. 4 of Marsaglia, "Xorshift RNGs" */
 	seed := internal.Prand32(s.prng)

x/xnet/stack-go.go

@@ -102,8 +102,9 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 	isDial := raddr.IsValid() && !raddr.Addr().IsUnspecified()
 	if laddr.Port() == 0 {
 		// Auto-assign an ephemeral port for both outbound dials and for listeners
-		// that did not request a fixed port.
+		// that did not request a fixed port. Sequential, not random: see
-		laddr = netip.AddrPortFrom(laddr.Addr(), uint16(49152+s.blk.async.Prand32()%16384))
+		// [StackAsync.ephemeralPort] for why random selection breaks dial churn.
+		laddr = netip.AddrPortFrom(laddr.Addr(), s.blk.async.ephemeralPort())
 	}
 	if laddr.Addr().IsUnspecified() {
 		// Fill in the stack's configured address for the requested family.

x/xnet/xnet_test.go

@@ -1196,3 +1196,39 @@ func TestEgressIP_TCPMSSAdvertisesMTU(t *testing.T) {
 		t.Errorf("advertised MSS = %d, want %d (MTU %d - 40)", gotMSS, wantMSS, mtu)
 	}
 }
+
+// TestEphemeralPortSequence checks no ephemeral port is reused before the whole
+// 16384-port dynamic range has cycled, which is what keeps a redial off the
+// teardown state (TIME-WAIT, NAT flow entries) of the conversation before it.
+func TestEphemeralPortSequence(t *testing.T) {
+	s := new(StackAsync)
+	err := s.Reset(StackConfig{
+		Hostname:          "eph",
+		RandSeed:          42,
+		StaticAddress4:    [4]byte{10, 0, 0, 50},
+		MaxActiveTCPPorts: 1,
+		HardwareAddress:   [6]byte{0xbe, 0xef, 0, 0, 0, 50},
+		MTU:               ethernet.MaxMTU,
+		ICMPQueueLimit:    2,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	const cycle = 16384
+	var seen [cycle]bool
+	for i := range cycle {
+		port := s.ephemeralPort()
+		if port < 49152 {
+			t.Fatalf("port %d below dynamic range (RFC 6335)", port)
+		}
+		idx := port - 49152
+		if seen[idx] {
+			t.Fatalf("port %d reused after only %d allocations (want full %d cycle)", port, i, cycle)
+		}
+		seen[idx] = true
+	}
+	// The cycle is exhausted: the next allocation may legitimately reuse.
+	if got := s.ephemeralPort(); got < 49152 {
+		t.Fatalf("post-cycle port %d below dynamic range", got)
+	}
+}