fix(internal): keep the write position when a ring is read empty (#181)

Ring.onReadEnd and Ring.ReadDiscard call Reset() when the last readable
byte is consumed, which rewinds Off and End to 0. That is a valid empty
state, but it also moves the write position, and bytes staged past that
position with PeekWrite are addressed relative to it. After the rewind a
Commit hands back whatever bytes happen to live at the start of the
buffer instead of the staged ones.

tcp stages out-of-order segments exactly this way (see reassembly.store,
"the ring write pointer advances in lockstep with rcv.NXT, so the staged
bytes are always where seq implies"). Reading is driven by the
application, so a receiver that drains its stream while a gap is open
loses the staged tail silently: the stream arrives with the correct
length and the wrong contents.

Observed on a Sophgo SG2002 (100Mbit DWMAC, receive ring dropping frames
under load): a 16 MiB download arrived complete with a different
SHA-256, and TLS over the same path failed with "bad record MAC", while
the transmit path was bit-perfect.

Mark the ring empty without rewinding instead: End=0 is what empty
means, and the write position is then Off, which Ring.Write already
supports explicitly.

Two tests, both failing before the change:

  - internal: staged bytes survive the ring being read empty.
  - tcp: a reassembled stream is byte-identical when segments arrive
    reordered (segment K arrived as the contents of an earlier segment).

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

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/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 := 0; round < rounds; round++ {
+		// Capture this round's segments on the wire, one segment per write.
+		segs := make([][]byte, 0, nsegs)
+		for i := 0; i < nsegs; i++ {
+			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)
+}