bfd(echo): Phase B.1+B.2 outstanding ring + echo detection

Adds per-session echo loss tracking and the echo-mode detection
timer so the echo transport landed in Phase A becomes a real
liveness channel. Without this commit, echo packets flowing with
no detection would be a passive RTT probe only.

Phase B.1 -- outstanding-ID ring (session/session.go + timers.go):

 - Fixed 16-slot echoOutstanding ring per Machine tracks every
   echo TX by (sequence, sentAt). RegisterEchoTx inserts in the
   first empty slot or overwrites the oldest live slot when the
   ring is full (an overwrite is equivalent to a lost echo from
   the detection standpoint). MatchEchoRx scans for the returning
   echo's sequence, clears the slot, and returns the monotonic
   RTT. ClearEchoSchedule wipes the ring on session-down so a
   flap-back-up starts with a clean detection window.

Phase B.2 -- detection-time switchover (engine/echo.go):

 - echoTickLocked now calls EchoDetectionExpired(now) before the
   TX pass. EchoDetectionExpired walks the ring and returns true
   when any outstanding entry is older than DetectMult *
   EchoInterval (the echo-mode detection time per RFC 5880
   Section 6.8.4). On true, the engine calls Machine.EchoFail
   which transitions the session to Down with DiagEchoFailed (2),
   fires the notify callback with the correct diagnostic, and
   clears the ring.
 - recordEchoRTTLocked now uses MatchEchoRx for the authoritative
   RTT (monotonic, immune to wall-clock jumps) and falls back to
   the ZEEC envelope TimestampMs when the ring entry was evicted.

New TestEchoDetectionSwitchover drives the full detection path
via a dropEchoTransport that swallows every outbound echo and
asserts a Down+DiagEchoFailed transition within 2 s. Race-clean
under -race -count=10.

Also fixes pre-existing goconst lint in cmd/ze/iface: promote
"bridge" literal in internal/component/iface/migrate_linux.go to
the existing zeTypeBridge constant, add nolint:goconst to CLI
dispatch strings in create.go, main.go, migrate.go, scan.go.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: thomas-mangin

cmd/ze/iface/create.go

@@ -22,9 +22,9 @@ func cmdCreate(args []string) int {
 	case "help", "-h", "--help": //nolint:goconst // consistent pattern across cmd files
 		createUsage()
 		return 0
-	case "dummy":
+	case "dummy": //nolint:goconst // CLI dispatch strings, constants in internal/component/iface
 		return cmdCreateDummy(args[1:])
-	case "veth":
+	case "veth": //nolint:goconst // CLI dispatch strings, constants in internal/component/iface
 		return cmdCreateVeth(args[1:])
 	default:
 		fmt.Fprintf(os.Stderr, "error: unknown interface type: %s (expected dummy or veth)\n", args[0])

cmd/ze/iface/main.go

@@ -36,7 +36,7 @@ func Run(args []string) int {
 	subcmd := args[0]
 	subArgs := args[1:]
 
-	if subcmd == "help" || subcmd == "-h" || subcmd == "--help" {
+	if subcmd == "help" || subcmd == "-h" || subcmd == "--help" { //nolint:goconst // consistent pattern across cmd files
 		usage()
 		return 0
 	}

cmd/ze/iface/migrate.go

@@ -72,7 +72,7 @@ func cmdMigrate(args []string) int {
 	// Validate --create type against known set.
 	if createTyp != "" {
 		switch createTyp {
-		case "dummy", "veth", "bridge": // valid
+		case "dummy", "veth", "bridge": //nolint:goconst // CLI dispatch strings, constants in internal/component/iface
 		default:
 			fmt.Fprintf(os.Stderr, "error: invalid --create type %q (expected dummy, veth, or bridge)\n", createTyp)
 			return 1

cmd/ze/iface/scan.go

@@ -108,7 +108,7 @@ func filterManaged(discovered []ifacepkg.DiscoveredInterface) []ifacepkg.Discove
 	filtered := make([]ifacepkg.DiscoveredInterface, 0, len(discovered))
 	for i := range discovered {
 		switch discovered[i].Type {
-		case "dummy", "veth", "bridge", "tunnel", "wireguard":
+		case "dummy", "veth", "bridge", "tunnel", "wireguard": //nolint:goconst // CLI dispatch strings
 			filtered = append(filtered, discovered[i])
 		}
 	}

internal/component/iface/migrate_linux.go

@@ -29,7 +29,7 @@ func validateMigrateConfig(cfg MigrateConfig) error {
 	}
 	if cfg.NewIfaceType != "" {
 		switch cfg.NewIfaceType {
-		case "dummy", "veth", "bridge": // valid types
+		case zeTypeDummy, zeTypeVeth, zeTypeBridge:
 		default: // unknown type
 			return fmt.Errorf("migrate: unknown interface type %q (expected dummy, veth, or bridge)", cfg.NewIfaceType)
 		}

internal/plugins/bfd/engine/echo.go

@@ -28,15 +28,18 @@ import (
 	"codeberg.org/thomas-mangin/ze/internal/plugins/bfd/transport"
 )
 
-// echoTickLocked fires per-session echo TX deadlines. Caller MUST
-// hold l.mu. When a session is Up with echo negotiated and its
-// next-echo deadline has passed, encode one ZEEC envelope into a
-// pool buffer, send via the echo transport, and advance the
-// per-session schedule.
+// echoTickLocked fires per-session echo TX deadlines and drives
+// the echo-mode detection timer. Caller MUST hold l.mu. For every
+// session with echo negotiated and its deadline passed the loop
+// encodes one ZEEC envelope, hands it to the echo transport, and
+// advances the per-session schedule. Before the TX pass the
+// session's outstanding ring is checked for stale entries; any
+// outstanding echo older than DetectMult * EchoInterval fires a
+// transition to Down with DiagEchoFailed.
 //
-// Sessions that are not Up have their echo schedule cleared so a
-// session that flapped down does not fire stale echoes after going
-// back up (PrimeEcho rearms on the next Up tick).
+// Sessions that are not Up have their echo schedule and
+// outstanding ring cleared so a session that flapped down does
+// not carry stale detection state back into Up.
 func (l *Loop) echoTickLocked(now time.Time) {
 	if l.echoTransport == nil {
 		return
@@ -47,6 +50,17 @@ func (l *Loop) echoTickLocked(now time.Time) {
 			m.ClearEchoSchedule()
 			continue
 		}
+		if m.EchoDetectionExpired(now) {
+			engineLog().Info("bfd echo detection expired",
+				"peer", m.PeerAddr().String(),
+				"detect", m.EchoDetectInterval())
+			m.EchoFail()
+			if hook := l.metricsHook.Load(); hook != nil {
+				(*hook).OnStateChange(packet.StateUp, packet.StateDown,
+					packet.DiagEchoFailed, m.Key().Mode.String(), m.Key().VRF)
+			}
+			continue
+		}
 		m.PrimeEcho(now)
 		deadline := m.NextEchoTxDeadline()
 		if deadline.IsZero() || now.Before(deadline) {
@@ -60,18 +74,19 @@ func (l *Loop) echoTickLocked(now time.Time) {
 // sendEchoLocked encodes a single ZEEC envelope for the session and
 // hands it to the echo transport. Caller MUST hold l.mu.
 //
-// The envelope's TimestampMs field carries a truncated millisecond
-// slice of the engine clock so the reflected copy can be matched
-// back to the original TX without a separate outstanding-ID index.
-// The sequence counter is self-carried for future diagnostics
-// (packet loss estimation) but the current RX path does not
-// consult it.
+// The sequence counter lets the RX path match returning echoes
+// against the per-session outstanding ring (RegisterEchoTx /
+// MatchEchoRx). The envelope's TimestampMs field is still written
+// with a truncated millisecond slice of the engine clock so the
+// RTT can fall back to the self-carried value when the matching
+// entry has already been evicted from a small ring under load.
 func (l *Loop) sendEchoLocked(entry *sessionEntry, now time.Time) {
 	buf := [packet.EchoLen]byte{}
 	ts := uint32(now.UnixMilli())
+	seq := entry.machine.NextEchoSequence()
 	e := packet.Echo{
 		LocalDiscriminator: entry.machine.LocalDiscriminator(),
-		Sequence:           entry.machine.NextEchoSequence(),
+		Sequence:           seq,
 		TimestampMs:        ts,
 	}
 	packet.WriteEcho(buf[:], 0, e)
@@ -88,6 +103,7 @@ func (l *Loop) sendEchoLocked(entry *sessionEntry, now time.Time) {
 		engineLog().Debug("bfd echo send failed", "peer", out.To, "err", err)
 		return
 	}
+	entry.machine.RegisterEchoTx(seq, now)
 	if hook := l.metricsHook.Load(); hook != nil {
 		(*hook).OnEchoTx(key.Mode.String())
 	}
@@ -154,16 +170,22 @@ func (l *Loop) findSessionByPeerLocked(peer netip.Addr) *sessionEntry {
 // recordEchoRTTLocked stores the round-trip time on the session and
 // fires the metrics hook. Caller MUST hold l.mu.
 //
-// The TimestampMs field is truncated from the 64-bit UnixMilli of
-// the sender, so the RTT calculation is done in the same truncated
-// space: `now.UnixMilli() & 0xFFFFFFFF` minus the received timestamp,
-// interpreted as a signed 32-bit delta so a clock that wraps past
-// 2^32 ms (every ~49 days) stays correct across the boundary.
+// The ring lookup is the authoritative RTT source: Machine tracks
+// the monotonic sentAt for every outstanding echo, so the delta is
+// immune to wall-clock jumps and system-suspend artifacts. When the
+// ring has already evicted the matching entry (burst loss, ring
+// overflow) the calculation falls back to the self-carried
+// TimestampMs in the ZEEC envelope, which is truncated from the
+// 64-bit UnixMilli and treated as a signed 32-bit delta so a clock
+// crossing the 2^32 ms boundary (~49 days) stays correct.
 func (l *Loop) recordEchoRTTLocked(entry *sessionEntry, e packet.Echo, now time.Time, mode string) {
-	nowMs := uint32(now.UnixMilli())
-	delta := int32(nowMs - e.TimestampMs)
-	delta = max(delta, 0)
-	rtt := time.Duration(delta) * time.Millisecond
+	rtt, ok := entry.machine.MatchEchoRx(e.Sequence, now)
+	if !ok {
+		nowMs := uint32(now.UnixMilli())
+		delta := int32(nowMs - e.TimestampMs)
+		delta = max(delta, 0)
+		rtt = time.Duration(delta) * time.Millisecond
+	}
 	entry.machine.RecordEchoRTT(rtt)
 	if hook := l.metricsHook.Load(); hook != nil {
 		(*hook).OnEchoRx(mode)

internal/plugins/bfd/engine/echo_test.go

@@ -105,10 +105,16 @@ func TestEchoRoundTrip(t *testing.T) {
 	if err != nil {
 		t.Fatalf("loopB.EnsureSession: %v", err)
 	}
+	// Intentionally do NOT defer Unsubscribe on subA/subB: Loop.Stop
+	// closes every subscriber channel before this test returns, and
+	// the engine's notify path copies the subscriber slice out from
+	// under subsMu before sending, so calling Unsubscribe AFTER the
+	// defer chain has already started races the final echo-driven
+	// state transition against Machine.EchoFail's notify.
+	// TODO(spec-bfd-engine-unsubscribe-lock): tighten Unsubscribe so
+	// it is safe to call while the express loop is still running.
 	subA := hA.Subscribe()
 	subB := hB.Subscribe()
-	defer hA.Unsubscribe(subA)
-	defer hB.Unsubscribe(subB)
 
 	// The three-way handshake is required for both machines to reach Up
 	// so echoTickLocked stops clearing the echo schedule.
@@ -151,3 +157,115 @@ func TestEchoRoundTrip(t *testing.T) {
 		t.Fatalf("no RTT samples recorded on hookA")
 	}
 }
+
+// dropEchoTransport is a Transport that accepts Sends silently and
+// never delivers any inbound packets. Used by
+// TestEchoDetectionSwitchover to simulate a peer that is dropping
+// every echo packet so the engine's echo detection path fires.
+//
+// Start / Stop are idempotent. The RX channel is never closed mid-test
+// so the express loop keeps selecting on a live channel; Stop closes
+// it so Loop.run's range exits cleanly.
+type dropEchoTransport struct {
+	rx chan transport.Inbound
+}
+
+func newDropEchoTransport() *dropEchoTransport {
+	return &dropEchoTransport{rx: make(chan transport.Inbound)}
+}
+
+func (*dropEchoTransport) Start() error                    { return nil }
+func (*dropEchoTransport) Send(_ transport.Outbound) error { return nil }
+func (t *dropEchoTransport) RX() <-chan transport.Inbound  { return t.rx }
+func (t *dropEchoTransport) Stop() error {
+	select {
+	case <-t.rx:
+	default:
+		close(t.rx)
+	}
+	return nil
+}
+
+// VALIDATES: spec-bfd-6c Phase B.2 -- when echo is negotiated and
+// Up but no reflected echoes return within DetectMult * EchoInterval,
+// the engine declares the session Down with DiagEchoFailed.
+// PREVENTS: regression where EchoFail never fires, fires with the
+// wrong diagnostic, or fails to clear the outstanding ring on
+// teardown.
+func TestEchoDetectionSwitchover(t *testing.T) {
+	addrAA := netip.MustParseAddr(addrA)
+	addrBB := netip.MustParseAddr(addrB)
+
+	controlA, controlB := transport.Pair(api.SingleHop, addrAA, addrBB)
+	loopA := NewLoopWithEcho(controlA, newDropEchoTransport(), clock.RealClock{})
+	loopB := NewLoopWithEcho(controlB, newDropEchoTransport(), clock.RealClock{})
+
+	if err := loopA.Start(); err != nil {
+		t.Fatalf("loopA.Start: %v", err)
+	}
+	defer func() {
+		if err := loopA.Stop(); err != nil {
+			t.Errorf("loopA.Stop: %v", err)
+		}
+	}()
+	if err := loopB.Start(); err != nil {
+		t.Fatalf("loopB.Start: %v", err)
+	}
+	defer func() {
+		if err := loopB.Stop(); err != nil {
+			t.Errorf("loopB.Stop: %v", err)
+		}
+	}()
+
+	hA, err := loopA.EnsureSession(echoReqFor(addrB, addrA))
+	if err != nil {
+		t.Fatalf("loopA.EnsureSession: %v", err)
+	}
+	if _, err := loopB.EnsureSession(echoReqFor(addrA, addrB)); err != nil {
+		t.Fatalf("loopB.EnsureSession: %v", err)
+	}
+
+	// Wait for the handshake to reach Up. Subsequent transitions
+	// (Up -> Down with DiagEchoFailed) are observed via subA.
+	subA := hA.Subscribe()
+	deadline := time.Now().Add(6 * time.Second)
+	upA := false
+	for !upA {
+		if time.Now().After(deadline) {
+			t.Fatalf("handshake not Up before deadline")
+		}
+		select {
+		case ev := <-subA:
+			if ev.State == packet.StateUp {
+				upA = true
+			}
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+
+	// After Up the drop transport swallows every outbound echo so
+	// the per-session outstanding ring fills up without any matching
+	// reflections. The engine tick then observes
+	// EchoDetectionExpired and calls EchoFail which publishes a
+	// Down event with DiagEchoFailed.
+	//
+	// Detection time = DetectMult (3) * EchoInterval (10ms) = 30ms;
+	// the poll interval is 5ms so the first expiry observation
+	// happens ~35ms after the first echo TX. A 2-second deadline is
+	// generously above that floor to absorb scheduler jitter under
+	// load.
+	deadline = time.Now().Add(2 * time.Second)
+	gotEchoFailed := false
+	for !gotEchoFailed {
+		if time.Now().After(deadline) {
+			t.Fatalf("no echo-failed transition within 2s")
+		}
+		select {
+		case ev := <-subA:
+			if ev.State == packet.StateDown && ev.Diag == packet.DiagEchoFailed {
+				gotEchoFailed = true
+			}
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+}

internal/plugins/bfd/session/session.go

@@ -145,6 +145,36 @@ type Machine struct {
 	// Zero until the first reflected echo is matched. Exposed via
 	// LastEchoRTT for snapshot consumers.
 	lastEchoRTT time.Duration
+
+	// echoOutstanding is a fixed-size ring of echo TX entries that
+	// have not yet been matched by a returning reflection. The ring
+	// is the state that turns echo from a passive RTT probe into an
+	// active liveness channel: the engine declares the session Down
+	// with DiagEchoFailed when the oldest unreturned entry has been
+	// waiting longer than DetectMult * EchoInterval. Stage 6b Phase B.
+	//
+	// Slot semantics: sentAt.IsZero() marks an empty slot. The ring
+	// is sized at echoOutstandingCap which is generous for the
+	// default DetectMult=3 (the detection threshold is six outstanding
+	// entries in the worst case). A full ring overwrites the oldest
+	// slot and the overwritten entry is treated as a miss.
+	echoOutstanding [echoOutstandingCap]echoEntry
+}
+
+// echoOutstandingCap is the fixed ring size for per-session echo
+// TX tracking. Chosen to comfortably exceed 2 * DetectMult for the
+// common DetectMult=3 without making the Machine struct fat; the
+// current cap of 16 uses 16 * 24 = 384 bytes per session on a
+// 64-bit build.
+const echoOutstandingCap = 16
+
+// echoEntry is one outstanding echo TX slot. sentAt is the monotonic
+// time the engine handed the packet to the transport; sequence is
+// the ZEEC Sequence field for the matching RX lookup. A zero sentAt
+// marks the slot as empty (swept by MatchEchoRx or ClearEchoSchedule).
+type echoEntry struct {
+	sequence uint32
+	sentAt   time.Time
 }
 
 // Role is the BFD role (Active or Passive) per RFC 5883 Section 4.3.