fix(afv): address mesh PR-10 review feedback

Stop async Interest re-apply; auth-before-workers Start; current Interest on reconnect; ValidateCluster host:port; empty PSK reject; prealloc caps; interest rate/first-bind/Case F/race tests; drain timer reuse; slog.Debug on decode/encrypt failures.
2026-08-10 11:33:29 +08:00 · 2026-07-28 17:40:32 -04:00
parent 39f042b758
commit 3f13aefd6b
14 changed files with 321 additions and 108 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -17,11 +17,12 @@ node_modules/
 # Local worklogs / personal trackers (not committed)
 *.local.md

-# Local binaries
+# Local binaries (built by go build -o …)
 /openfsd-web
 /fsdweb
 /fsd
 /openfsd
+/openfsd-migrate-to-rqlite

 # User-fetched X-Plane Global Airports data and bulk conversion output.
 # Tooling under cmd/, internal/xp12aptdat/, scripts/ is tracked.
--- a/internal/afv/mesh.go
+++ b/internal/afv/mesh.go
@@ -57,6 +57,8 @@ type Mesh interface {
 	OnDirectory(fn MeshDirectoryHandler)
 	// SetSnapshotProvider supplies local session blocks for TrxSnapshot.
 	SetSnapshotProvider(fn func() []MeshSessionBlock)
+	// SetInterestProvider supplies current Interest for post-Hello / reconnect (M-16).
+	SetInterestProvider(fn func() []InterestEntry)
 }

 // MeshDirectoryHandler receives remote directory updates from peers.
@@ -134,9 +136,13 @@ func (c *Config) ValidateCluster() error {
 	if strings.TrimSpace(c.ClusterNodeID) == "" {
 		return fmt.Errorf("AFV_CLUSTER_NODE_ID required when AFV_CLUSTER_ENABLED=true")
 	}
-	if strings.TrimSpace(c.ClusterListen) == "" {
+	listen := strings.TrimSpace(c.ClusterListen)
+	if listen == "" {
 		return fmt.Errorf("AFV_CLUSTER_LISTEN required when AFV_CLUSTER_ENABLED=true")
 	}
+	if _, _, err := net.SplitHostPort(listen); err != nil {
+		return fmt.Errorf("AFV_CLUSTER_LISTEN: want host:port: %w", err)
+	}
 	if strings.TrimSpace(c.ClusterPSK) == "" {
 		return fmt.Errorf("AFV_CLUSTER_PSK required when AFV_CLUSTER_ENABLED=true")
 	}
@@ -159,6 +165,9 @@ func (c *Config) ValidateCluster() error {
 		if p.ID == "" || p.Addr == "" {
 			return fmt.Errorf("AFV_CLUSTER_PEERS: empty id or addr")
 		}
+		if _, _, err := net.SplitHostPort(p.Addr); err != nil {
+			return fmt.Errorf("AFV_CLUSTER_PEERS: peer %q addr want host:port: %w", p.ID, err)
+		}
 		if _, ok := seen[p.ID]; ok {
 			return fmt.Errorf("AFV_CLUSTER_PEERS: duplicate peer id %q", p.ID)
 		}
@@ -272,6 +281,9 @@ func (s *Server) registerMeshCallbacks() {
 	s.mesh.SetSnapshotProvider(func() []MeshSessionBlock {
 		return s.reg.snapshotLocalSessionsForMesh()
 	})
+	s.mesh.SetInterestProvider(func() []InterestEntry {
+		return s.reg.buildInterestEntries(s.cfg)
+	})
 }

 // runInterestLoop publishes Interest ≤ 2 Hz when dirty (M-5 / M-15 / test 20).
@@ -303,13 +315,14 @@ func (s *Server) publishInterestNow() {
 }

 // handleMeshAudioRelay routes inbound AudioRelay to local bound RX (M-8).
+// isXC: primary synthetic TX only; never XC-again (PR-9 not implemented).
 func (s *Server) handleMeshAudioRelay(fromNode string, r AudioRelay) {
 	if s == nil || s.reg == nil {
 		return
 	}
-	// isXC: primary synthetic TX only; never XC-again (PR-9 not implemented).
-	_ = r.IsXC
 	_ = fromNode
+	// isXC is intentionally not re-XC'd; routeSyntheticTX is primary-only.
+	_ = r.IsXC

 	s.udpMu.Lock()
 	pc := s.udpConn
@@ -332,11 +345,13 @@ func (s *Server) handleMeshAudioRelay(fromNode string, r AudioRelay) {
 		}
 		ch, err := afvprotocol.ServerChannel(rec.tag, rec.rxKey[:], rec.txKey[:])
 		if err != nil {
+			slog.Debug("AFV mesh AR ServerChannel", "err", err, "callsign", r.Callsign)
 			continue
 		}
 		seq := rec.sess.nextTxSeq()
 		pkt, err := ch.Encapsulate(seq, afvprotocol.DTONameAudioRx, ar.EncodeMsgpack(), nil)
 		if err != nil {
+			slog.Debug("AFV mesh AR Encapsulate", "err", err, "callsign", r.Callsign)
 			continue
 		}
 		addr, ok := rec.udp.(net.Addr)
@@ -424,3 +439,10 @@ func (s *Server) PublishInterestNowForTest() {
 func (s *Server) MarkInterestDirtyForTest() {
 	s.markInterestDirty()
 }
+
+// ClearInterestDirtyForTest clears dirty so the interest loop will not republish (tests).
+func (s *Server) ClearInterestDirtyForTest() {
+	if s != nil {
+		s.interestDirty.Store(false)
+	}
+}
--- a/internal/afv/mesh_config_test.go
+++ b/internal/afv/mesh_config_test.go
@@ -37,6 +37,8 @@ func TestValidateCluster(t *testing.T) {
 		}, "max 4"},
 		{"bad peer", func(c *Config) { c.ClusterPeers = "notvalid" }, "invalid"},
 		{"dup", func(c *Config) { c.ClusterPeers = "n2=1:1,n2=1:2" }, "duplicate"},
+		{"bad listen", func(c *Config) { c.ClusterListen = "not-a-hostport" }, "host:port"},
+		{"bad peer addr", func(c *Config) { c.ClusterPeers = "n2=nohostport" }, "host:port"},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
--- a/internal/afv/mesh_e2e_test.go
+++ b/internal/afv/mesh_e2e_test.go
@@ -521,24 +521,43 @@ func TestMeshE2E_CaseF_EmptyInterestNoFlood(t *testing.T) {
 	meshBindHB(t, chA, cliA, udp1, "AAL1", 0)
 	meshBindHB(t, chB, cliB, udp2, "AAL2", 0)

-	// clear interest so n1 does not think n2 wants anything
-	n1.mesh.ClearPeerInterest("n2")
-	// prevent interest loop from re-publishing immediately by not calling PublishInterestNow on n2
-	// (n2's interest would re-apply on n1 when n2 publishes — clear after and block)
-	// Rapid clear after publish
-	n1.mesh.ClearPeerInterest("n2")
+	// Explicit empty Interest from n2; clear dirty so interest loop cannot
+	// immediately republish full RX coverage (would race the no-flood assert).
+	ck := geo.CellKey{
+		ILat: geo.CellIndex(lat+0.001, geo.DefaultGridCellDeg),
+		ILon: geo.CellIndex(lon+0.001, geo.DefaultGridCellDeg),
+	}
+	n2.srv.ClearInterestDirtyForTest()
+	n2.mesh.PublishInterest(nil)
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if !n1.mesh.PeerWants("n2", freq, ck) {
+			break
+		}
+		n2.srv.ClearInterestDirtyForTest()
+		n2.mesh.PublishInterest(nil)
+		time.Sleep(10 * time.Millisecond)
+	}
+	if n1.mesh.PeerWants("n2", freq, ck) {
+		t.Fatal("Case F: peer still wants after empty Interest")
+	}
+	// Keep dirty clear during burst
+	n2.srv.ClearInterestDirtyForTest()
+	n1.srv.ClearInterestDirtyForTest()

 	at := afvprotocol.AudioTx{
 		Callsign: "AAL1", SequenceCounter: 1, Audio: []byte{1}, LastPacket: true,
 		Transceivers: []afvprotocol.TxTransceiver{{ID: 0}},
 	}
 	pkt, _ := chA.Encapsulate(1, afvprotocol.DTONameAudioTx, at.EncodeMsgpack(), nil)
-	// send a few times with clear between
 	for i := 0; i < 5; i++ {
-		n1.mesh.ClearPeerInterest("n2")
+		n2.srv.ClearInterestDirtyForTest()
 		_, _ = cliA.WriteTo(pkt, udp1)
 	}
 	if _, ok := readAR(t, chB, cliB, 400*time.Millisecond); ok {
 		t.Fatal("Case F: flood with empty peer interest")
 	}
+	if n1.mesh.PeerWants("n2", freq, ck) {
+		t.Fatal("Case F: PeerWants became true during burst")
+	}
 }
--- a/internal/afv/mesh_frame.go
+++ b/internal/afv/mesh_frame.go
@@ -260,7 +260,11 @@ func DecodeHelloPayload(b []byte) (HelloPayload, error) {

 // VerifyHelloPSK constant-time compares peer PSK to local. Length mismatch rejects
 // without early return that leaks length via timing of compare body (dummy compare).
+// Empty PSK is rejected (defense in depth; production also requires non-empty via ValidateCluster).
 func VerifyHelloPSK(localPSK, peerPSK string) error {
+	if len(localPSK) == 0 || len(peerPSK) == 0 {
+		return errMeshHelloAuth
+	}
 	lb := []byte(localPSK)
 	pb := []byte(peerPSK)
 	if len(lb) != len(pb) {
@@ -358,7 +362,13 @@ func DecodeTrxSnapshot(b []byte) (TrxSnapshotPayload, error) {
 	if err != nil {
 		return p, err
 	}
-	if n > 1<<20 {
+	// Cap prealloc / session count (DoS guard before TCP mesh lands).
+	const maxSnapshotSessions = 8192
+	if n > maxSnapshotSessions {
+		return p, errMeshBadPayload
+	}
+	// Rough remaining-byte floor: each session needs at least a few u16 strings.
+	if int(n) > 0 && len(b) < int(n)*4 {
 		return p, errMeshBadPayload
 	}
 	p.Sessions = make([]MeshSessionBlock, 0, n)
@@ -657,7 +667,13 @@ func DecodeInterest(b []byte) (InterestPayload, error) {
 	if err != nil {
 		return p, err
 	}
-	if n > 1<<20 {
+	// Cap to Interest max (4096) with small headroom for wire tolerance.
+	const maxInterestDecode = 8192
+	if n > maxInterestDecode {
+		return p, errMeshBadPayload
+	}
+	// Each entry is 4+4+4 = 12 bytes.
+	if int(n) > 0 && len(b) < int(n)*12 {
 		return p, errMeshBadPayload
 	}
 	p.Entries = make([]InterestEntry, 0, n)
--- a/internal/afv/mesh_frame_test.go
+++ b/internal/afv/mesh_frame_test.go
@@ -208,11 +208,20 @@ func TestHelloVerifyPSK(t *testing.T) {
 	if err := VerifyHelloPSK("abc", "ab"); err != errMeshHelloAuth {
 		t.Fatalf("length mismatch err=%v", err)
 	}
-	if err := VerifyHelloPSK("", ""); err != nil {
+	if err := VerifyHelloPSK("", ""); err != errMeshHelloAuth {
+		t.Fatalf("empty PSK should reject: %v", err)
+	}
+	if err := VerifyHelloPSK("x", ""); err != errMeshHelloAuth {
 		t.Fatal(err)
 	}
 }

+// Hello first-frame-not-type-1 is enforced on the TCP accept path (PR-10b).
+// MemoryMesh has no wire Hello; document here so row 2b is not silently dropped.
+func TestHelloFirstFrameTCPDeferred(t *testing.T) {
+	t.Log("Hello first-frame type check is TCP-only (mesh_tcp.go / PR-10b); MemoryMesh uses constructor PSK")
+}
+
 func TestMeshStringOversizeTruncate(t *testing.T) {
 	long := strings.Repeat("x", maxMeshString+50)
 	enc := meshEncodeString(nil, long)
@@ -327,13 +336,20 @@ func TestDecodeAll_TrailingGarbageAndShort(t *testing.T) {
 	if _, err := DecodeInterest(ip); err == nil {
 		t.Fatal()
 	}
-	// Interest huge n
+	// Interest huge n (prealloc cap)
 	var big []byte
 	big = meshEncodeString(big, "n")
 	big = meshEncodeU32(big, 1<<21)
 	if _, err := DecodeInterest(big); err == nil {
 		t.Fatal()
 	}
+	// Interest n larger than remaining bytes
+	var shortN []byte
+	shortN = meshEncodeString(shortN, "n")
+	shortN = meshEncodeU32(shortN, 100)
+	if _, err := DecodeInterest(shortN); err == nil {
+		t.Fatal("short body for n entries")
+	}
 	// Interest short entry
 	var short []byte
 	short = meshEncodeString(short, "n")
--- a/internal/afv/mesh_interest_test.go
+++ b/internal/afv/mesh_interest_test.go
@@ -1,6 +1,7 @@
 package afv

 import (
+	"context"
 	"testing"
 	"time"

@@ -139,7 +140,10 @@ func TestBuildInterest_BoundOnly_ATCCoverage(t *testing.T) {
 }

 func TestInterestRateLimit_DirtyStorm(t *testing.T) {
-	// interest loop ticks at interestMinInterval; force-count publishes via Server helper
+	// Drive real runInterestLoop under rapid dirty storm; assert ≤ ~2 Hz publishes.
+	if interestMinInterval > 500*time.Millisecond {
+		t.Fatalf("interval=%v exceeds 2Hz budget", interestMinInterval)
+	}
 	cfg := &Config{
 		APIListen: "127.0.0.1:0", UDPListen: "127.0.0.1:0",
 		UDPAdvertiseIPv4: "127.0.0.1:1",
@@ -148,26 +152,72 @@ func TestInterestRateLimit_DirtyStorm(t *testing.T) {
 	}
 	s := New(cfg, nil, nil, []byte("x"))
 	hub := NewMemoryHub()
-	m, err := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	m, err := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "psk-ok", PeerIDs: []string{"n1", "n2"}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	_, _ = NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	_, _ = NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "psk-ok", PeerIDs: []string{"n1", "n2"}})
 	s.SetMesh(m)
+	s.registerMeshCallbacks()

-	// Rapid dirty without waiting interval: publishInterestNow only when we call it.
-	// Mark dirty many times — flag is boolean so storm collapses to one publish per tick.
-	for i := 0; i < 100; i++ {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := m.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	go s.runInterestLoop(ctx)
+
+	// Storm dirty for ~1.2s (boolean flag → at most one publish per 500ms tick)
+	deadline := time.Now().Add(1200 * time.Millisecond)
+	for time.Now().Before(deadline) {
 		s.MarkInterestDirtyForTest()
+		time.Sleep(5 * time.Millisecond)
+	}
+	cancel()
+	time.Sleep(20 * time.Millisecond)
+
+	n := m.InterestPublishCount()
+	// Start + loop initial + ticks over ~1.2s at 500ms — budget ≤8
+	if n > 8 {
+		t.Fatalf("interest publish count %d exceeds ≤2Hz storm budget", n)
+	}
+	if n < 2 {
+		t.Fatalf("expected loop to publish, got %d", n)
+	}
+}
+
+func TestFirstBindDirtyOnceOnServer(t *testing.T) {
+	cfg := &Config{MaxSessions: 10, MaxSessionsPerCID: 5, RangeDefaultNM: 40}
+	s := New(cfg, nil, nil, []byte("k"))
+	now := time.Now()
+	sess, _, err := s.reg.CreateOrReplace(1, "P1", "", now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, _ = s.reg.UpdateTransceivers(1, "P1", []Transceiver{{
+		ID: 0, Frequency: 118700000, LatDeg: 40, LonDeg: -73,
+	}})
+	_ = s.interestDirty.Swap(false)
+
+	_, first, ok := s.reg.BindUDP(sess, fakeAddr{"127.0.0.1:1"}, now)
+	if !ok || !first {
+		t.Fatal("first bind")
+	}
+	if first {
+		s.markInterestDirty()
 	}
 	if !s.InterestDirtyForTest() {
-		t.Fatal("expected dirty")
+		t.Fatal("first bind must dirty interest")
 	}
-	// single publish clears path for rate ≤2Hz (interval 500ms)
-	if interestMinInterval > 500*time.Millisecond {
-		t.Fatalf("interval=%v exceeds 2Hz budget", interestMinInterval)
+	_ = s.interestDirty.Swap(false)
+
+	_, first2, ok := s.reg.BindUDP(sess, fakeAddr{"127.0.0.1:1"}, now)
+	if !ok || first2 {
+		t.Fatalf("re-touch first=%v ok=%v", first2, ok)
+	}
+	if s.InterestDirtyForTest() {
+		t.Fatal("re-touch must not dirty interest")
 	}
-	s.PublishInterestNowForTest()
 }

 func TestCapInterest_UnderCapNoOp(t *testing.T) {
--- a/internal/afv/mesh_memory.go
+++ b/internal/afv/mesh_memory.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"

 	"github.com/renorris/openfsd/internal/geo"
 )
@@ -50,11 +51,15 @@ type MemoryMesh struct {
 	onPeerDead func(nodeID string)
 	onDir      MeshDirectoryHandler
 	snapFn     func() []MeshSessionBlock
+	interestFn func() []InterestEntry

 	// capture last outbound AudioRelay encodings for Case E (optional)
 	lastRelayMu sync.Mutex
 	lastRelays  []AudioRelay

+	// interestPublishCount counts PublishInterest calls (rate tests).
+	interestPublishCount atomic.Uint64
+
 	stopCh   chan struct{}
 	stopOnce sync.Once
 	wg       sync.WaitGroup
@@ -110,12 +115,34 @@ func (m *MemoryMesh) Start(ctx context.Context) error {
 	if m == nil {
 		return fmt.Errorf("afv mesh: nil mesh")
 	}
+	m.mu.Lock()
+	if m.started {
+		m.mu.Unlock()
+		return nil
+	}
+	m.mu.Unlock()
+
+	// Auth first (M-3 / Issue 2+9): fail closed before workers or Snapshot/Interest.
+	m.hub.mu.RLock()
+	for id, peer := range m.hub.nodes {
+		if id == m.self {
+			continue
+		}
+		if err := VerifyHelloPSK(m.psk, peer.psk); err != nil {
+			m.hub.mu.RUnlock()
+			slog.Warn("AFV mesh hello auth failed", "peer", id)
+			return err
+		}
+	}
+	m.hub.mu.RUnlock()
+
 	m.mu.Lock()
 	if m.started {
 		m.mu.Unlock()
 		return nil
 	}
 	m.started = true
+	// reset stop so a prior failed Start/Stop can recover in tests
 	m.mu.Unlock()

 	// Start per-peer drainers that deliver control then voice.
@@ -134,24 +161,9 @@ func (m *MemoryMesh) Start(ctx context.Context) error {
 		}()
 	}

-	// Post-link sequence (M-16): Snapshot then Interest (even if empty).
+	// Post-auth sequence (M-16): Snapshot then current Interest (may be empty).
 	m.PublishTrxSnapshot()
-	// Interest may be empty until Server interest loop runs; still send empty.
-	m.PublishInterest(nil)
-
-	// Optional: verify PSK against peers already on hub (auth mode for tests).
-	m.hub.mu.RLock()
-	for id, peer := range m.hub.nodes {
-		if id == m.self {
-			continue
-		}
-		if err := VerifyHelloPSK(m.psk, peer.psk); err != nil {
-			m.hub.mu.RUnlock()
-			slog.Warn("AFV mesh hello auth failed", "peer", id)
-			return err
-		}
-	}
-	m.hub.mu.RUnlock()
+	m.publishCurrentInterest()

 	go func() {
 		<-ctx.Done()
@@ -160,6 +172,18 @@ func (m *MemoryMesh) Start(ctx context.Context) error {
 	return nil
 }

+// publishCurrentInterest sends Interest from provider or empty set (M-16).
+func (m *MemoryMesh) publishCurrentInterest() {
+	m.mu.RLock()
+	fn := m.interestFn
+	m.mu.RUnlock()
+	var entries []InterestEntry
+	if fn != nil {
+		entries = fn()
+	}
+	m.PublishInterest(entries)
+}
+
 func (m *MemoryMesh) Stop() error {
 	if m == nil {
 		return nil
@@ -208,6 +232,17 @@ func (m *MemoryMesh) SetSnapshotProvider(fn func() []MeshSessionBlock) {
 	m.mu.Unlock()
 }

+func (m *MemoryMesh) SetInterestProvider(fn func() []InterestEntry) {
+	m.mu.Lock()
+	m.interestFn = fn
+	m.mu.Unlock()
+}
+
+// InterestPublishCount returns how many times PublishInterest ran (tests).
+func (m *MemoryMesh) InterestPublishCount() uint64 {
+	return m.interestPublishCount.Load()
+}
+
 func (m *MemoryMesh) peerAlive(id string) bool {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
@@ -247,8 +282,9 @@ func (m *MemoryMesh) SimulatePeerUp(nodeID string) {
 		m.peerCtrl[nodeID] = newDropOldestQueue[meshCtrlJob](meshControlQueueDepth)
 	}
 	m.mu.Unlock()
+	// Post-Hello sequence with current Interest (M-16), not forced empty.
 	m.PublishTrxSnapshot()
-	m.PublishInterest(nil)
+	m.publishCurrentInterest()
 }

 // ClearPeerInterest empties a peer's interest set (Case F / tests).
@@ -326,8 +362,10 @@ func (m *MemoryMesh) PublishInterest(entries []InterestEntry) {
 	if m == nil {
 		return
 	}
-	// Synchronous apply on peers (M-17) — deliver interest immediately, not via async queue,
-	// so PeerWants barrier works. Also enqueue encoded frame for control-path coverage.
+	m.interestPublishCount.Add(1)
+	// Synchronous apply on peers only (M-17). Control queue carries the frame for
+	// transport/drop-oldest metrics; deliverCtrl must NOT re-apply Interest (stale
+	// async frames must not overwrite a newer sync set).
 	payload := EncodeInterest(InterestPayload{NodeID: m.self, Entries: entries})
 	set := make(map[FreqCell]struct{}, len(entries))
 	for _, e := range entries {
@@ -345,9 +383,6 @@ func (m *MemoryMesh) PublishInterest(entries []InterestEntry) {
 	m.hub.mu.RUnlock()

 	for _, p := range peers {
-		if !m.peerAlive(p.self) && !p.peerAlive(m.self) {
-			// either side may track aliveness; skip if we think peer is down
-		}
 		if !m.peerAlive(p.self) {
 			continue
 		}
@@ -363,7 +398,7 @@ func (m *MemoryMesh) PublishInterest(entries []InterestEntry) {
 		p.peerInterest[m.self] = cp
 		p.mu.Unlock()
 	}
-	// Also put on control queues for drop-oldest testing (async apply already done).
+	// Enqueue for control-path drop-oldest only — not applied on receive (Issue 1).
 	m.broadcastCtrl(MeshTypeInterest, payload)
 }

@@ -480,6 +515,8 @@ func (m *MemoryMesh) broadcastCtrl(typ byte, payload []byte) {
 }

 func (m *MemoryMesh) drainPeer(ctx context.Context, peerID string) {
+	idle := time.NewTimer(2 * time.Millisecond)
+	defer idle.Stop()
 	for {
 		select {
 		case <-ctx.Done():
@@ -495,12 +532,19 @@ func (m *MemoryMesh) drainPeer(ctx context.Context, peerID string) {
 		alive := m.alive[peerID]
 		m.mu.RUnlock()
 		if !alive {
+			if !idle.Stop() {
+				select {
+				case <-idle.C:
+				default:
+				}
+			}
+			idle.Reset(50 * time.Millisecond)
 			select {
 			case <-ctx.Done():
 				return
 			case <-m.stopCh:
 				return
-			case <-timeAfter(50):
+			case <-idle.C:
 			}
 			continue
 		}
@@ -518,36 +562,24 @@ func (m *MemoryMesh) drainPeer(ctx context.Context, peerID string) {
 			}
 		}
 		if !delivered {
+			if !idle.Stop() {
+				select {
+				case <-idle.C:
+				default:
+				}
+			}
+			idle.Reset(2 * time.Millisecond)
 			select {
 			case <-ctx.Done():
 				return
 			case <-m.stopCh:
 				return
-			case <-timeAfter(2):
+			case <-idle.C:
 			}
 		}
 	}
 }

-// timeAfter is a tiny sleep helper (ms) to avoid busy loop without importing time in every call site pattern.
-func timeAfter(ms int) <-chan struct{} {
-	ch := make(chan struct{})
-	go func() {
-		// use real time package
-		sleepMS(ms)
-		close(ch)
-	}()
-	return ch
-}
-
-func sleepMS(ms int) {
-	if ms <= 0 {
-		return
-	}
-	// defined in mesh_memory_time.go to keep imports clean — inline here instead
-	doSleep(ms)
-}
-
 func (m *MemoryMesh) deliverCtrl(peerID string, job meshCtrlJob) {
 	peer := m.lookupPeer(peerID)
 	if peer == nil {
@@ -557,6 +589,7 @@ func (m *MemoryMesh) deliverCtrl(peerID string, job meshCtrlJob) {
 	case MeshTypeTrxSnapshot:
 		p, err := DecodeTrxSnapshot(job.payload)
 		if err != nil {
+			slog.Debug("AFV mesh decode TrxSnapshot", "err", err, "from", m.self, "to", peerID)
 			return
 		}
 		sessions := meshSessionsToRemote(p.Sessions)
@@ -569,6 +602,7 @@ func (m *MemoryMesh) deliverCtrl(peerID string, job meshCtrlJob) {
 	case MeshTypeTrxDelta:
 		p, err := DecodeTrxDelta(job.payload)
 		if err != nil {
+			slog.Debug("AFV mesh decode TrxDelta", "err", err, "from", m.self)
 			return
 		}
 		trxs := meshTrxToLocal(p.Trxs)
@@ -581,6 +615,7 @@ func (m *MemoryMesh) deliverCtrl(peerID string, job meshCtrlJob) {
 	case MeshTypeSessionLeave:
 		p, err := DecodeSessionLeave(job.payload)
 		if err != nil {
+			slog.Debug("AFV mesh decode SessionLeave", "err", err, "from", m.self)
 			return
 		}
 		peer.mu.RLock()
@@ -590,19 +625,10 @@ func (m *MemoryMesh) deliverCtrl(peerID string, job meshCtrlJob) {
 			dir.ApplyLeave(p.OriginNodeID, p.Callsign)
 		}
 	case MeshTypeInterest:
-		// Already applied synchronously in PublishInterest; ignore async duplicate
-		// or apply again for queue-path completeness.
-		p, err := DecodeInterest(job.payload)
-		if err != nil {
-			return
-		}
-		set := make(map[FreqCell]struct{}, len(p.Entries))
-		for _, e := range p.Entries {
-			set[FreqCell{FreqHz: e.FreqHz, Cell: geo.CellKey{ILat: e.ILat, ILon: e.ILon}}] = struct{}{}
-		}
-		peer.mu.Lock()
-		peer.peerInterest[m.self] = set
-		peer.mu.Unlock()
+		// Interest is applied only synchronously in PublishInterest (M-17).
+		// Ignoring async re-apply prevents older queued frames from overwriting
+		// a newer set (Issue 1). Frame still counts toward control-queue metrics.
+		return
 	case MeshTypeHeartbeat:
 		// liveness only
 	}
--- a/internal/afv/mesh_memory_test.go
+++ b/internal/afv/mesh_memory_test.go
@@ -6,6 +6,7 @@ import (
 	"time"

 	"github.com/renorris/openfsd/internal/geo"
+	"github.com/renorris/openfsd/pkg/afvprotocol"
 )

 func TestMemoryMesh_PeerDeathPurgesInterest(t *testing.T) {
@@ -108,13 +109,34 @@ func TestMemoryMesh_HelloPSKReject(t *testing.T) {
 	m1, _ := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "good", PeerIDs: []string{"n1", "n2"}})
 	_, _ = NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "bad", PeerIDs: []string{"n1", "n2"}})
 	ctx := context.Background()
-	// start n1 first ok; n2 will fail when verifying against n1?
-	// Start verifies local PSK against peer.psk
+	// Start verifies local PSK against peer.psk before workers (fail closed).
+	err := m1.Start(ctx)
+	if err == nil {
+		t.Fatal("expected PSK reject")
+	}
+	if err != errMeshHelloAuth {
+		t.Fatalf("err=%v want errMeshHelloAuth", err)
+	}
+	_ = m1.Stop()
+}
+
+func TestPublishInterest_NoAsyncStaleOverwrite(t *testing.T) {
+	hub := NewMemoryHub()
+	m1, _ := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	m2, _ := NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 	if err := m1.Start(ctx); err != nil {
-		// m1 sees n2 with bad psk
-		if err != errMeshHelloAuth {
-			t.Fatalf("err=%v", err)
-		}
+		t.Fatal(err)
+	}
+	if err := m2.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	m2.PublishInterest(nil)
+	m2.PublishInterest([]InterestEntry{{FreqHz: 99, ILat: 1, ILon: 2}})
+	time.Sleep(50 * time.Millisecond)
+	if !m1.PeerWants("n2", 99, geo.CellKey{ILat: 1, ILon: 2}) {
+		t.Fatal("sync interest lost to stale async re-apply")
 	}
 }

@@ -262,7 +284,18 @@ func TestConcurrentATReapTrxRace(t *testing.T) {
 		MaxSessions: 50, MaxSessionsPerCID: 10,
 		RangeDefaultNM: 100, HeartbeatTimeout: time.Hour, SessionIdleTimeout: time.Hour,
 	}
-	r := newRegistry(cfg)
+	hub := NewMemoryHub()
+	m1, _ := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	m2, _ := NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	_ = m1.Start(ctx)
+	_ = m2.Start(ctx)
+
+	s := New(cfg, nil, nil, []byte("k"))
+	s.SetMesh(m1)
+	s.registerMeshCallbacks()
+	r := s.reg
 	now := time.Now()
 	s1, _, _ := r.CreateOrReplace(1, "A", "", now)
 	s2, _, _ := r.CreateOrReplace(2, "B", "", now)
@@ -270,6 +303,11 @@ func TestConcurrentATReapTrxRace(t *testing.T) {
 	_, _, _ = r.UpdateTransceivers(2, "B", []Transceiver{{ID: 0, Frequency: 118700000, LatDeg: 40.01, LonDeg: -73.01}})
 	_, _, _ = r.BindUDP(s1, fakeAddr{"1"}, now)
 	_, _, _ = r.BindUDP(s2, fakeAddr{"2"}, now)
+	m1.ApplyInterestDirect("n2", []InterestEntry{{
+		FreqHz: 118700000,
+		ILat:   geo.CellIndex(40, geo.DefaultGridCellDeg),
+		ILon:   geo.CellIndex(-73, geo.DefaultGridCellDeg),
+	}})

 	done := make(chan struct{})
 	go func() {
@@ -278,23 +316,31 @@ func TestConcurrentATReapTrxRace(t *testing.T) {
 			_ = r.routeSyntheticTX("C", false, []RelayTxRadio{{
 				TxID: 0, FreqHz: 118700000, LatDeg: 40.0, LonDeg: -73.0,
 			}})
-			_, _ = r.snapshotTXForMesh(s1, nil)
+			_, radios := r.snapshotTXForMesh(s1, []afvprotocol.TxTransceiver{{ID: 0}})
+			m1.EnqueueAudioRelay(AudioRelay{
+				Callsign: "A", SequenceCounter: uint32(i), Audio: []byte{1},
+				TxRadios: radios,
+			})
 			_ = r.snapshotLocalSessionsForMesh()
 		}
 	}()
 	go func() {
 		for i := 0; i < 100; i++ {
-			_, _, _ = r.UpdateTransceivers(1, "A", []Transceiver{{
+			isATC, trxs, _ := r.UpdateTransceivers(1, "A", []Transceiver{{
 				ID: 0, Frequency: 118700000, LatDeg: 40 + float64(i)*0.0001, LonDeg: -73,
 			}})
+			s.meshPublishDelta("A", isATC, trxs)
 		}
 	}()
 	go func() {
 		for i := 0; i < 50; i++ {
-			_ = r.Reap(now)
+			leaves := r.Reap(now) // unlikely to reap with long timeouts
+			s.meshPublishLeaves(leaves)
 			_ = r.buildInterestEntries(cfg)
+			s.PublishInterestNowForTest()
 		}
 	}()
 	<-done
 	time.Sleep(50 * time.Millisecond)
+	_ = m2
 }
--- a/internal/afv/mesh_memory_time.go
+++ b/internal/afv/mesh_memory_time.go
@@ -1,7 +0,0 @@
-package afv
-
-import "time"
-
-func doSleep(ms int) {
-	time.Sleep(time.Duration(ms) * time.Millisecond)
-}
--- a/internal/afv/mesh_remotedir.go
+++ b/internal/afv/mesh_remotedir.go
@@ -31,6 +31,9 @@ func (d *remoteDir) ApplySnapshot(origin string, sessions []RemoteSession) {
 		return
 	}
 	origin = strings.TrimSpace(origin)
+	if origin == "" {
+		return // consistent with ApplyDelta empty-origin reject
+	}
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	m := make(map[string]RemoteSession, len(sessions))
--- a/internal/afv/mesh_remotedir_test.go
+++ b/internal/afv/mesh_remotedir_test.go
@@ -94,6 +94,10 @@ func TestRemoteDir_NilSafe(t *testing.T) {

 func TestRemoteDir_EmptyCallsignAndMiss(t *testing.T) {
 	d := newRemoteDir()
+	d.ApplySnapshot("", []RemoteSession{{Callsign: "X", IsATC: true}})
+	if d.Count() != 0 {
+		t.Fatal("empty origin should reject")
+	}
 	d.ApplySnapshot("n", []RemoteSession{{Callsign: "  ", IsATC: true}})
 	if d.Count() != 0 {
 		t.Fatal("empty callsign should skip")
--- a/internal/afv/mesh_router_test.go
+++ b/internal/afv/mesh_router_test.go
@@ -74,12 +74,22 @@ func TestRouteSyntheticTX_IsXCNoReXC(t *testing.T) {
 		ID: 0, Frequency: 118700000, LatDeg: 40.0, LonDeg: -73.0,
 	}})
 	_, _, _ = r.BindUDP(rx, fakeAddr{"2"}, now)
-	recs := r.routeSyntheticTX("AAL1", false, []RelayTxRadio{{
-		TxID: 0, FreqHz: 118700000, LatDeg: 40.01, LonDeg: -73.01,
-	}})
+	// Cross-freq radio would be XC path if implemented — primary route only
+	// routes matching freq; second freq not auto-coupled.
+	recs := r.routeSyntheticTX("AAL1", false, []RelayTxRadio{
+		{TxID: 0, FreqHz: 118700000, LatDeg: 40.01, LonDeg: -73.01},
+		{TxID: 1, FreqHz: 119000000, LatDeg: 40.01, LonDeg: -73.01}, // no local RX on this freq
+	})
 	if len(recs) != 1 {
 		t.Fatalf("got %d", len(recs))
 	}
+	// Server path with isXC=true still only primary synthetic (no second hop)
+	s := New(cfg, nil, nil, []byte("k"))
+	s.handleMeshAudioRelay("nX", AudioRelay{
+		Callsign: "AAL1", IsATC: false, IsXC: true, Audio: []byte{1},
+		TxRadios: []RelayTxRadio{{TxID: 0, FreqHz: 118700000, LatDeg: 40.01, LonDeg: -73.01}},
+	})
+	// nil udp — no panic; isXC ignored for re-XC
 }

 func TestHandleMeshAudioRelay_NilUDP(t *testing.T) {
--- a/internal/afv/server.go
+++ b/internal/afv/server.go
@@ -142,9 +142,14 @@ func (s *Server) Run(ctx context.Context) error {
 			meshCtx, meshCancel := context.WithCancel(runCtx)
 			defer meshCancel()
 			go s.runInterestLoop(meshCtx)
-			if err := s.mesh.Start(meshCtx); err != nil && runCtx.Err() == nil {
-				errCh <- err
-				cancel()
+			if err := s.mesh.Start(meshCtx); err != nil {
+				_ = s.mesh.Stop() // always cleanup on Start failure (hub/workers)
+				if runCtx.Err() == nil {
+					errCh <- err
+					cancel()
+					return
+				}
+				errCh <- nil
 				return
 			}
 			<-meshCtx.Done()