Checksum the header and refuse sequence numbers the sender never sent

frame.go

@@ -13,6 +13,8 @@ const (
 	hdrMagic  = 0x43424c54
 	minFrame  = ethHdrLen + hdrLen
 	maxFrame  = 9216
+
+	hdrCheckOff = 18
 )
 
 var crcTable = crc32.MakeTable(crc32.Castagnoli)
@@ -113,6 +115,26 @@ func (f *frameSpec) expectedCRC(patIdx, payLen int) (uint32, bool) {
 	return crc, ok
 }
 
+// A one's complement sum over the header, skipping the two bytes it is written
+// into. The payload checksum covers the payload alone, so without this the
+// sequence number is the one field on the wire that nothing checks, and a bit
+// flipped in it arrives looking exactly like a legitimate frame from far in the
+// future. Every single bit error changes the folded sum, which is the error a
+// cable going marginal actually produces.
+func headerCheck(h []byte) uint16 {
+	var sum uint32
+	for i := 0; i+1 < hdrLen; i += 2 {
+		if i == hdrCheckOff {
+			continue
+		}
+		sum += uint32(binary.BigEndian.Uint16(h[i:]))
+	}
+	for sum>>16 != 0 {
+		sum = sum&0xffff + sum>>16
+	}
+	return ^uint16(sum)
+}
+
 func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int) {
 	h := buf[ethHdrLen:]
 	binary.BigEndian.PutUint32(h[0:4], hdrMagic)
@@ -121,7 +143,7 @@ func putHeader(buf []byte, patIdx int, stream uint16, seq uint64, payLen int) {
 	binary.BigEndian.PutUint16(h[6:8], stream)
 	binary.BigEndian.PutUint64(h[8:16], seq)
 	binary.BigEndian.PutUint16(h[16:18], uint16(payLen))
-	binary.BigEndian.PutUint16(h[18:20], 0)
+	binary.BigEndian.PutUint16(h[hdrCheckOff:], headerCheck(h))
 }
 
 type parsed struct {
@@ -131,21 +153,35 @@ type parsed struct {
 	payLen int
 }
 
-func parseHeader(buf []byte) (parsed, bool) {
+// Why a frame was turned away. A header that is not ours and a header of ours
+// that arrived damaged are different things to report: the first is somebody
+// else on the wire, the second is the fault being tested for.
+type hdrStatus int
+
+const (
+	hdrOK hdrStatus = iota
+	hdrForeign
+	hdrDamaged
+)
+
+func parseHeader(buf []byte) (parsed, hdrStatus) {
 	var p parsed
 	if len(buf) < minFrame {
-		return p, false
+		return p, hdrForeign
 	}
 	h := buf[ethHdrLen:]
 	if binary.BigEndian.Uint32(h[0:4]) != hdrMagic {
-		return p, false
+		return p, hdrForeign
+	}
+	if binary.BigEndian.Uint16(h[hdrCheckOff:]) != headerCheck(h) {
+		return p, hdrDamaged
 	}
 	p.patIdx = int(h[5])
 	p.stream = binary.BigEndian.Uint16(h[6:8])
 	p.seq = binary.BigEndian.Uint64(h[8:16])
 	p.payLen = int(binary.BigEndian.Uint16(h[16:18]))
 	if minFrame+p.payLen > len(buf) {
-		return p, false
+		return p, hdrDamaged
 	}
-	return p, true
+	return p, hdrOK
 }

frame_test.go

@@ -0,0 +1,66 @@
+package main
+
+import "testing"
+
+func testFrame() []byte {
+	buf := make([]byte, 512)
+	spec := newFrameSpec([6]byte{1, 2, 3, 4, 5, 6}, [6]byte{7, 8, 9, 10, 11, 12},
+		etherBase, []int{512})
+	spec.prefill(buf, 3)
+	putHeader(buf, 3, 5, 0x0123456789ab, 512-minFrame)
+	return buf
+}
+
+func TestHeaderRoundTrip(t *testing.T) {
+	p, st := parseHeader(testFrame())
+	if st != hdrOK {
+		t.Fatalf("status = %v, want hdrOK", st)
+	}
+	want := parsed{patIdx: 3, stream: 5, seq: 0x0123456789ab, payLen: 512 - minFrame}
+	if p != want {
+		t.Errorf("parsed = %+v, want %+v", p, want)
+	}
+}
+
+// The sequence number is the one field on the wire that nothing else checks, and
+// a single flipped bit in it is exactly what a marginal cable produces. Every
+// bit of the header has to be covered, including the checksum's own.
+func TestHeaderChecksumCatchesEverySingleBitFlip(t *testing.T) {
+	for byteIdx := 0; byteIdx < hdrLen; byteIdx++ {
+		for bit := 0; bit < 8; bit++ {
+			buf := testFrame()
+			buf[ethHdrLen+byteIdx] ^= 1 << bit
+
+			_, st := parseHeader(buf)
+			if st == hdrOK {
+				t.Errorf("header byte %d bit %d flipped and the frame was accepted",
+					byteIdx, bit)
+			}
+			// The magic is what says the frame is ours at all, so damage there is
+			// somebody else's traffic as far as we can tell.
+			wantForeign := byteIdx < 4
+			if wantForeign && st != hdrForeign {
+				t.Errorf("magic byte %d bit %d = %v, want hdrForeign", byteIdx, bit, st)
+			}
+			if !wantForeign && st != hdrDamaged {
+				t.Errorf("header byte %d bit %d = %v, want hdrDamaged", byteIdx, bit, st)
+			}
+		}
+	}
+}
+
+func TestHeaderShortFrameIsForeign(t *testing.T) {
+	if _, st := parseHeader(testFrame()[:minFrame-1]); st != hdrForeign {
+		t.Errorf("status = %v, want hdrForeign", st)
+	}
+}
+
+// A length that runs past the frame is damage rather than a foreign header: the
+// magic and the checksum both said the header was ours.
+func TestHeaderLengthPastFrameIsDamaged(t *testing.T) {
+	buf := testFrame()
+	putHeader(buf, 3, 5, 1, len(buf))
+	if _, st := parseHeader(buf); st != hdrDamaged {
+		t.Errorf("status = %v, want hdrDamaged", st)
+	}
+}

loss.go

@@ -18,19 +18,39 @@ type lossWindow struct {
 	bits   []uint64
 	lost   atomic.Uint64
 	late   atomic.Uint64
+
+	// The sending half of this stream. Both halves are in this process and the
+	// receiving socket ignores its own outgoing frames, so this window is fed by
+	// that sender alone and by nothing else on the wire.
+	sent *atomic.Uint64
 }
 
-func newLossWindows(n int) []lossWindow {
+// Takes the sending halves rather than a count, so a window cannot be built
-	w := make([]lossWindow, n)
+// without the bound it judges sequence numbers against.
+func newLossWindows(tx []*txStats) []lossWindow {
+	w := make([]lossWindow, len(tx))
 	for i := range w {
 		w[i].bits = make([]uint64, lossWords)
+		w[i].sent = &tx[i].sent
 	}
 	return w
 }
 
 // A sequence number is only judged once it falls out of the window, so frames
 // still queued in another rx worker are never miscounted as lost.
-func (w *lossWindow) observe(seq uint64) {
+//
+// Reports whether the sequence number could have come off the wire at all. One
+// above what the sender has reached is a damaged header rather than a gap, and
+// is refused before it can touch the base. Believing one would drag the window
+// up past every real sequence, write off the span in between as lost, and leave
+// every frame after it arriving below the base and counted late for the rest of
+// the run. Nothing that arrives later is evidence enough to undo that, which is
+// why the sender's own frontier is the guard rather than a limit on how far a
+// single step may move.
+func (w *lossWindow) observe(seq uint64) bool {
+	if seq >= w.sent.Load() {
+		return false
+	}
 	w.mu.Lock()
 	if !w.inited {
 		// Start half a window below the first sequence seen, so frames another
@@ -43,7 +63,7 @@ func (w *lossWindow) observe(seq uint64) {
 	if seq < w.base {
 		w.mu.Unlock()
 		w.late.Add(1)
-		return
+		return true
 	}
 	if seq >= w.base+lossSlots {
 		w.evict(seq - lossSlots + 1)
@@ -51,6 +71,7 @@ func (w *lossWindow) observe(seq uint64) {
 	idx := seq & (lossSlots - 1)
 	w.bits[idx>>6] |= 1 << (idx & 63)
 	w.mu.Unlock()
+	return true
 }
 
 func (w *lossWindow) evict(newBase uint64) {

loss_test.go

@@ -2,13 +2,17 @@ package main
 
 import "testing"
 
-func newWindow() *lossWindow {
+// sent is the sender's frontier: sequence numbers at or above it were never put
-	w := newLossWindows(1)
+// on the wire.
+func newWindow(sent uint64) *lossWindow {
+	tx := &txStats{}
+	tx.sent.Store(sent)
+	w := newLossWindows([]*txStats{tx})
 	return &w[0]
 }
 
 func TestLossWindowContiguousLosesNothing(t *testing.T) {
-	w := newWindow()
+	w := newWindow(lossSlots + 1000)
 	for seq := uint64(0); seq < lossSlots+1000; seq++ {
 		w.observe(seq)
 	}
@@ -21,7 +25,7 @@ func TestLossWindowContiguousLosesNothing(t *testing.T) {
 }
 
 func TestLossWindowCountsGapOnceEvicted(t *testing.T) {
-	w := newWindow()
+	w := newWindow(lossSlots + 1000)
 	for seq := uint64(0); seq < lossSlots+1000; seq++ {
 		if seq == 100 {
 			continue
@@ -36,7 +40,7 @@ func TestLossWindowCountsGapOnceEvicted(t *testing.T) {
 // Arriving out of order inside the window is not loss: a sequence is only
 // judged once it falls out the far end.
 func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) {
-	w := newWindow()
+	w := newWindow(lossSlots + 1000)
 	for seq := uint64(99); ; seq-- {
 		w.observe(seq)
 		if seq == 0 {
@@ -54,7 +58,7 @@ func TestLossWindowOutOfOrderIsNotLoss(t *testing.T) {
 // The other branch of evict: a jump past a whole window writes off everything
 // the window held plus the sequences that never landed in it at all.
 func TestLossWindowJumpBeyondWindow(t *testing.T) {
-	w := newWindow()
+	w := newWindow(200001)
 	w.observe(0)
 	w.observe(200000)
 
@@ -69,7 +73,7 @@ func TestLossWindowJumpBeyondWindow(t *testing.T) {
 }
 
 func TestLossWindowBelowBaseIsLate(t *testing.T) {
-	w := newWindow()
+	w := newWindow(100001)
 	w.observe(100000)
 	w.observe(1000)
 	if got := w.late.Load(); got != 1 {
@@ -83,9 +87,58 @@ func TestLossWindowBelowBaseIsLate(t *testing.T) {
 // The first sequence seen starts the window half a span below it, so frames
 // another worker is still holding land inside rather than arriving late.
 func TestLossWindowStartsHalfAWindowBack(t *testing.T) {
-	w := newWindow()
+	w := newWindow(100001)
 	w.observe(100000)
 	if w.base != 100000-lossSlots/2 {
 		t.Errorf("base = %d, want %d", w.base, 100000-lossSlots/2)
 	}
 }
+
+// The failure this guard exists for. A sequence number the sender never reached
+// used to drag the base up to it, write off the span in between as lost, and
+// leave every real frame after it below the base and counted late for the rest
+// of the run. It has to be refused outright, and the stream has to go on
+// counting as if it had never arrived.
+func TestLossWindowRefusesUnsentSeq(t *testing.T) {
+	const first = 100000
+	w := newWindow(first + 2000)
+	for seq := uint64(first); seq < first+1000; seq++ {
+		w.observe(seq)
+	}
+	base := w.base
+
+	if w.observe(1 << 62) {
+		t.Error("a sequence number far past the sender's frontier was accepted")
+	}
+	if got := w.lost.Load(); got != 0 {
+		t.Errorf("lost = %d after one impossible sequence number, want 0", got)
+	}
+	if w.base != base {
+		t.Errorf("base moved to %d, want it left at %d where the real traffic put it",
+			w.base, base)
+	}
+
+	// Still tracking the real traffic, rather than reporting every frame late
+	// against a base that ran away.
+	for seq := uint64(first + 1000); seq < first+2000; seq++ {
+		w.observe(seq)
+	}
+	if got := w.late.Load(); got != 0 {
+		t.Errorf("late = %d, want 0", got)
+	}
+	if got := w.lost.Load(); got != 0 {
+		t.Errorf("lost = %d, want 0", got)
+	}
+}
+
+// The sender's own frontier is the bound, so the sequence one past it is
+// refused while the one below it is not.
+func TestLossWindowBoundIsExclusive(t *testing.T) {
+	w := newWindow(500)
+	if !w.observe(499) {
+		t.Error("the last sequence the sender put on the wire was refused")
+	}
+	if w.observe(500) {
+		t.Error("a sequence the sender had not reached was accepted")
+	}
+}

main.go

@@ -169,6 +169,7 @@ type sample struct {
 	rxFrames, rxBytes uint64
 	lost, late        uint64
 	crcErr, badMagic  uint64
+	badHdr            uint64
 	badLen            uint64
 	txErrs            uint64
 	rxErrs            uint64
@@ -204,6 +205,7 @@ func (d *direction) snapshot() sample {
 		s.rxBytes += r.bytes.Load()
 		s.crcErr += r.crcErr.Load()
 		s.badMagic += r.badMagic.Load()
+		s.badHdr += r.badHdr.Load()
 		s.badLen += r.badLen.Load()
 		s.rxErrs += r.rxErrs.Load()
 	}
@@ -279,10 +281,11 @@ type view struct {
 func errsBetween(b, n counterSet) errs {
 	return errs{
 		lost: n.s.lost - b.s.lost,
-		// Three ways of noticing one thing: a payload that does not match its
+		// Four ways of noticing one thing: a payload that does not match its
-		// checksum, a header that is not ours, and a length that cannot be.
+		// checksum, a header that does not match its own, a header that is not
-		corrupt: (n.s.crcErr - b.s.crcErr) + (n.s.badMagic - b.s.badMagic) +
+		// ours, and a length that cannot be.
-			(n.s.badLen - b.s.badLen),
+		corrupt: (n.s.crcErr - b.s.crcErr) + (n.s.badHdr - b.s.badHdr) +
+			(n.s.badMagic - b.s.badMagic) + (n.s.badLen - b.s.badLen),
 		// What the hardware reported. Nothing the host declined to send is here,
 		// so this one going red means the cable.
 		link: (n.nic - b.nic) + (n.s.rxErrs - b.s.rxErrs),
@@ -372,8 +375,15 @@ func (d *direction) primeCounters() {
 }
 
 func buildDirection(label string, tx, rx endpoint) (*direction, error) {
+	// Built before the windows, since each window judges sequence numbers against
+	// the frontier its own sender publishes.
+	txs := make([]*txStats, numStreams)
+	for i := range txs {
+		txs[i] = &txStats{}
+	}
 	d := &direction{
-		streams: newLossWindows(numStreams),
+		txStats: txs,
+		streams: newLossWindows(txs),
 		cable:   newCableStats(),
 	}
 	// Held open for the life of the run: the stats ioctl is issued five times a
@@ -399,7 +409,6 @@ func buildDirection(label string, tx, rx endpoint) (*direction, error) {
 			return nil, fmt.Errorf("%s tx socket: %w", label, err)
 		}
 		d.txFDs = append(d.txFDs, fd)
-		d.txStats = append(d.txStats, &txStats{})
 
 		fd, err = openRxSocket(rx.idx, et)
 		if err != nil {

main_test.go

@@ -95,19 +95,19 @@ func TestErrsBetweenBuckets(t *testing.T) {
 	n := counterSet{
 		s: sample{
 			lost: 1, late: 7,
-			crcErr: 2, badMagic: 3, badLen: 4,
+			crcErr: 2, badMagic: 3, badLen: 4, badHdr: 10,
 			txErrs: 6, rxErrs: 5,
 		},
 		drops: 9,
 		nic:   8,
 	}
 	got := errsBetween(counterSet{}, n)
-	want := errs{lost: 1, corrupt: 2 + 3 + 4, link: 8 + 5, internal: 9 + 7 + 6}
+	want := errs{lost: 1, corrupt: 2 + 3 + 4 + 10, link: 8 + 5, internal: 9 + 7 + 6}
 	if got != want {
 		t.Errorf("errsBetween = %+v, want %+v", got, want)
 	}
-	if got.total() != 45 {
+	if got.total() != 55 {
-		t.Errorf("total = %d, want 45", got.total())
+		t.Errorf("total = %d, want 55", got.total())
 	}
 }
 

probe.go

@@ -210,8 +210,8 @@ func (r *probeReceiver) run(done *atomic.Bool) {
 		if err != nil {
 			continue
 		}
-		h, ok := parseHeader(buf[:n])
+		h, st := parseHeader(buf[:n])
-		if !ok || h.stream != probeStream {
+		if st != hdrOK || h.stream != probeStream {
 			continue
 		}
 		ts, ok := hwTimestamp(oob[:oobn])

rx.go

@@ -12,6 +12,7 @@ type rxStats struct {
 	frames   atomic.Uint64
 	bytes    atomic.Uint64
 	badMagic atomic.Uint64
+	badHdr   atomic.Uint64
 	badLen   atomic.Uint64
 	crcErr   atomic.Uint64
 	rxErrs   atomic.Uint64
@@ -99,9 +100,13 @@ func (w *rxWorker) run(done *atomic.Bool) {
 		}
 		for i := 0; i < n; i++ {
 			buf := bufs[i][:int(hdrs[i].len)]
-			p, ok := parseHeader(buf)
+			p, st := parseHeader(buf)
-			if !ok {
+			if st != hdrOK {
+				if st == hdrForeign {
 					w.stats.badMagic.Add(1)
+				} else {
+					w.stats.badHdr.Add(1)
+				}
 				continue
 			}
 			w.stats.frames.Add(1)
@@ -110,8 +115,13 @@ func (w *rxWorker) run(done *atomic.Bool) {
 				w.stats.observe(ts, uint64(len(buf)))
 			}
 
-			if int(p.stream) < len(w.streams) {
+			// A sequence number the sender never reached got past the header
-				w.streams[p.stream].observe(p.seq)
+			// checksum, so the frame is damaged whatever its payload says. Counted
+			// here rather than left to the payload check, which would report the
+			// same frame twice or, if only the header was hit, not at all.
+			if int(p.stream) < len(w.streams) && !w.streams[p.stream].observe(p.seq) {
+				w.stats.badHdr.Add(1)
+				continue
 			}
 
 			want, ok := w.spec.expectedCRC(p.patIdx, p.payLen)

tx.go

@@ -8,6 +8,12 @@ import (
 
 type txStats struct {
 	errs atomic.Uint64
+
+	// The exclusive upper bound on any sequence number this stream can have put
+	// on the wire. Read by the receiving half of the same stream, which is the
+	// only thing that can tell a gap it must write off from a sequence number
+	// that was never sent at all.
+	sent atomic.Uint64
 }
 
 type txWorker struct {
@@ -46,6 +52,11 @@ func (w *txWorker) run(done *atomic.Bool) {
 			iovs[i].Len = uint64(size)
 		}
 
+		// Published before the send rather than after it, so it is never behind a
+		// frame already in flight. Overshooting a partial send is harmless: it is
+		// a bound, and one the next pass raises again.
+		w.stats.sent.Store(seq + uint64(w.batch))
+
 		n, err := sendmmsg(w.fd, hdrs)
 		if n > 0 {
 			seq += uint64(n)