fix(afv): mesh e2e and test residual polish

PeerWants barriers use speaker TX cells; Case A LastPacket both ways; Case D relies on Interest provider after reconnect; first-bind via real UDP HB; harden reconnect Snapshot; document Hello first-frame as TCP-only.
2026-08-13 04:55:42 +08:00 · 2026-07-28 17:46:34 -04:00
parent 3f13aefd6b
commit 6faa09aebb
5 changed files with 180 additions and 60 deletions
--- a/internal/afv/mesh_e2e_test.go
+++ b/internal/afv/mesh_e2e_test.go
@@ -105,11 +105,13 @@ func meshBindHB(t *testing.T, ch *afvprotocol.Channel, cli *net.UDPConn, server
 	}
 }

-func waitPeerWants(t *testing.T, m *afv.MemoryMesh, peer string, freq uint32, lat, lon float64) {
+// waitPeerWantsTX waits until local mesh believes peer wants the TX radio cell
+// used by EnqueueAudioRelay fan-out (CellKey of speaker TX lat/lon), not RX pos.
+func waitPeerWantsTX(t *testing.T, m *afv.MemoryMesh, peer string, freq uint32, txLat, txLon float64) {
 	t.Helper()
 	ck := geo.CellKey{
-		ILat: geo.CellIndex(lat, geo.DefaultGridCellDeg),
-		ILon: geo.CellIndex(lon, geo.DefaultGridCellDeg),
+		ILat: geo.CellIndex(txLat, geo.DefaultGridCellDeg),
+		ILon: geo.CellIndex(txLon, geo.DefaultGridCellDeg),
 	}
 	deadline := time.Now().Add(3 * time.Second)
 	for time.Now().Before(deadline) {
@@ -118,7 +120,7 @@ func waitPeerWants(t *testing.T, m *afv.MemoryMesh, peer string, freq uint32, la
 		}
 		time.Sleep(20 * time.Millisecond)
 	}
-	t.Fatalf("timeout PeerWants peer=%s freq=%d cell=%+v", peer, freq, ck)
+	t.Fatalf("timeout PeerWants peer=%s freq=%d txCell=%+v (tx=%.4f,%.4f)", peer, freq, ck, txLat, txLon)
 }

 func setupMeshPairWithUsers(t *testing.T) (n1, n2 *meshNode, cidA, cidB int) {
@@ -235,12 +237,13 @@ func TestMeshE2E_CaseA_BidirectionalA2A(t *testing.T) {
 	meshBindHB(t, chA, cliA, udp1, "AAL1", 0)
 	meshBindHB(t, chB, cliB, udp2, "AAL2", 0)

-	// force interest now and wait PeerWants barriers
+	// force interest now; barriers use speaker TX cells (fan-out filter keys)
 	n1.srv.PublishInterestNowForTest()
 	n2.srv.PublishInterestNowForTest()
-	// n1 needs to know n2 wants TX cells near A; n2 wants n1's TX cells near B
-	waitPeerWants(t, n1.mesh, "n2", freq, lat+0.001, lon+0.001)
-	waitPeerWants(t, n2.mesh, "n1", freq, lat, lon)
+	// A→B: n1 fans out if n2 wants A's TX cell
+	waitPeerWantsTX(t, n1.mesh, "n2", freq, lat, lon)
+	// B→A: n2 fans out if n1 wants B's TX cell
+	waitPeerWantsTX(t, n2.mesh, "n1", freq, lat+0.001, lon+0.001)

 	// A → B
 	at := afvprotocol.AudioTx{
@@ -278,8 +281,8 @@ func TestMeshE2E_CaseA_BidirectionalA2A(t *testing.T) {
 	if !ok {
 		t.Fatal("Case A: A did not receive AR from B")
 	}
-	if ar2.Callsign != "AAL2" || ar2.SequenceCounter != 9 {
-		t.Fatalf("AR2=%+v", ar2)
+	if ar2.Callsign != "AAL2" || ar2.SequenceCounter != 9 || ar2.LastPacket {
+		t.Fatalf("AR2=%+v want LastPacket=false", ar2)
 	}
 }

@@ -352,7 +355,8 @@ func TestMeshE2E_CaseC_ATCRadius(t *testing.T) {
 	meshBindHB(t, chB, cliB, udp2, "AAL2", 0)
 	n1.srv.PublishInterestNowForTest()
 	n2.srv.PublishInterestNowForTest()
-	waitPeerWants(t, n1.mesh, "n2", freq, 41.5, -73.0)
+	// ATC TX at 40.0: n1 fans out if n2 wants that TX cell (ATC-range interest)
+	waitPeerWantsTX(t, n1.mesh, "n2", freq, 40.0, -73.0)

 	// ATC TX → pilot AR
 	at := afvprotocol.AudioTx{
@@ -369,7 +373,8 @@ func TestMeshE2E_CaseC_ATCRadius(t *testing.T) {
 	// drain any leftover
 	_, _ = readAR(t, chA, cliA, 50*time.Millisecond)
 	n2.srv.PublishInterestNowForTest()
-	waitPeerWants(t, n2.mesh, "n1", freq, 40.0, -73.0)
+	// pilot TX at 41.5: n2 fans out if n1 wants pilot TX cell
+	waitPeerWantsTX(t, n2.mesh, "n1", freq, 41.5, -73.0)
 	atP := afvprotocol.AudioTx{
 		Callsign: "AAL2", SequenceCounter: 2, Audio: []byte{6}, LastPacket: true,
 		Transceivers: []afvprotocol.TxTransceiver{{ID: 0}},
@@ -404,7 +409,7 @@ func TestMeshE2E_CaseD_PeerDeathReconnect(t *testing.T) {
 	meshBindHB(t, chB, cliB, udp2, "AAL2", 0)
 	n1.srv.PublishInterestNowForTest()
 	n2.srv.PublishInterestNowForTest()
-	waitPeerWants(t, n1.mesh, "n2", freq, lat+0.001, lon+0.001)
+	waitPeerWantsTX(t, n1.mesh, "n2", freq, lat, lon)

 	// peer death both directions
 	n1.mesh.SimulatePeerDown("n2")
@@ -426,13 +431,12 @@ func TestMeshE2E_CaseD_PeerDeathReconnect(t *testing.T) {
 		t.Fatal("AR during peer death")
 	}

-	// reconnect
+	// reconnect: SimulatePeerUp re-runs Snapshot + current Interest via provider (M-16)
 	n1.mesh.SimulatePeerUp("n2")
 	n2.mesh.SimulatePeerUp("n1")
-	n1.srv.PublishInterestNowForTest()
-	n2.srv.PublishInterestNowForTest()
-	waitPeerWants(t, n1.mesh, "n2", freq, lat+0.001, lon+0.001)
-	waitPeerWants(t, n2.mesh, "n1", freq, lat, lon)
+	// No PublishInterestNowForTest — rely on SetInterestProvider path from Server.
+	waitPeerWantsTX(t, n1.mesh, "n2", freq, lat, lon)
+	waitPeerWantsTX(t, n2.mesh, "n1", freq, lat+0.001, lon+0.001)

 	at.SequenceCounter = 4
 	pkt, _ = chA.Encapsulate(12, afvprotocol.DTONameAudioTx, at.EncodeMsgpack(), nil)
@@ -465,7 +469,7 @@ func TestMeshE2E_CaseE_KeysNeverOnMesh(t *testing.T) {
 	meshBindHB(t, chB, cliB, udp2, "AAL2", 0)
 	n1.srv.PublishInterestNowForTest()
 	n2.srv.PublishInterestNowForTest()
-	waitPeerWants(t, n1.mesh, "n2", freq, lat+0.001, lon+0.001)
+	waitPeerWantsTX(t, n1.mesh, "n2", freq, lat, lon)

 	rxKey := append([]byte(nil), pcA.VoiceServer.ChannelConfig.AeadReceiveKey...)
 	txKey := append([]byte(nil), pcA.VoiceServer.ChannelConfig.AeadTransmitKey...)
@@ -561,3 +565,58 @@ func TestMeshE2E_CaseF_EmptyInterestNoFlood(t *testing.T) {
 		t.Fatal("Case F: PeerWants became true during burst")
 	}
 }
+
+// TestFirstBindDirtyOnceViaUDPHB exercises production BindUDP→markInterestDirty
+// via real UDP heartbeat (not a reimplemented mark path).
+func TestFirstBindDirtyOnceViaUDPHB(t *testing.T) {
+	n1, _, cidA, _ := setupMeshPairWithUsers(t)
+	tok := meshAuth(t, n1.api, cidA)
+	pc := meshPostCS(t, n1.api, tok, cidA, "P1", n1.udp)
+	meshPostTrx(t, n1.api, tok, cidA, "P1", 40.0, -73.0, 118700000)
+
+	// clear dirty from trx post
+	n1.srv.ClearInterestDirtyForTest()
+	if n1.srv.InterestDirtyForTest() {
+		t.Fatal("dirty should be clear before first HB")
+	}
+
+	udp, err := net.ResolveUDPAddr("udp", n1.udp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cli, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cli.Close()
+	ch, err := afvprotocol.ClientChannel(
+		pc.VoiceServer.ChannelConfig.ChannelTag,
+		pc.VoiceServer.ChannelConfig.AeadReceiveKey,
+		pc.VoiceServer.ChannelConfig.AeadTransmitKey,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// First HB binds and dirties interest (production udp.go path)
+	meshBindHB(t, ch, cli, udp, "P1", 0)
+	// allow handleUDP to finish
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if n1.srv.InterestDirtyForTest() {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if !n1.srv.InterestDirtyForTest() {
+		t.Fatal("first UDP HB must mark interest dirty")
+	}
+
+	n1.srv.ClearInterestDirtyForTest()
+	// Second HB same addr: re-touch, must not dirty again
+	meshBindHB(t, ch, cli, udp, "P1", 1)
+	time.Sleep(50 * time.Millisecond)
+	if n1.srv.InterestDirtyForTest() {
+		t.Fatal("second HB re-touch must not dirty interest")
+	}
+}
--- a/internal/afv/mesh_frame.go
+++ b/internal/afv/mesh_frame.go
@@ -6,6 +6,12 @@ package afv
 // shape match FSD mesh Hello (nodeID + PSK strings); comparison is CT-upgraded
 // vs FSD's non-constant-time !=. Do not log PSK contents on failure.
 //
+// Hello first-frame rule (TCP only, PR-10b / M-11): on a TCP mesh connection the
+// first frame must be type Hello; any other type closes the conn. MemoryMesh has
+// no wire Hello — auth is constructor PSK verify on Start. When mesh_tcp.go
+// lands, enforce first-frame type == MeshTypeHello on accept/dial before any
+// Snapshot/Interest/AudioRelay.
+//
 // Local length-prefix framing — do not import internal/cluster.

 import (
--- a/internal/afv/mesh_interest_test.go
+++ b/internal/afv/mesh_interest_test.go
@@ -186,6 +186,8 @@ func TestInterestRateLimit_DirtyStorm(t *testing.T) {
 	}
 }

+// TestFirstBindDirtyOnceOnServer covers registry firstBind flag (M-15).
+// UDP production path is exercised in TestFirstBindDirtyOnceViaUDPHB (afv_test).
 func TestFirstBindDirtyOnceOnServer(t *testing.T) {
 	cfg := &Config{MaxSessions: 10, MaxSessionsPerCID: 5, RangeDefaultNM: 40}
 	s := New(cfg, nil, nil, []byte("k"))
@@ -203,6 +205,7 @@ func TestFirstBindDirtyOnceOnServer(t *testing.T) {
 	if !ok || !first {
 		t.Fatal("first bind")
 	}
+	// Production UDP path: mark only when firstBind (mirrored here for unit isolation)
 	if first {
 		s.markInterestDirty()
 	}
@@ -215,6 +218,9 @@ func TestFirstBindDirtyOnceOnServer(t *testing.T) {
 	if !ok || first2 {
 		t.Fatalf("re-touch first=%v ok=%v", first2, ok)
 	}
+	if first2 {
+		s.markInterestDirty()
+	}
 	if s.InterestDirtyForTest() {
 		t.Fatal("re-touch must not dirty interest")
 	}
--- a/internal/afv/mesh_memory_test.go
+++ b/internal/afv/mesh_memory_test.go
@@ -178,54 +178,103 @@ func TestMemoryMesh_KeysNeverOnCapturedRelay(t *testing.T) {
 	}
 }

+// TestMemoryMesh_ReconnectSnapshot hardens Snapshot apply after death+up (M-16).
 func TestMemoryMesh_ReconnectSnapshot(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"}})
-	remote := newRemoteDir()
-	m1.OnDirectory(remote)
+	m1, err := NewMemoryMesh(hub, MeshConfig{NodeID: "n1", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m2, err := NewMemoryMesh(hub, MeshConfig{NodeID: "n2", PSK: "p", PeerIDs: []string{"n1", "n2"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	remote1 := newRemoteDir()
+	remote2 := newRemoteDir()
+	m1.OnDirectory(remote1)
+	m2.OnDirectory(remote2)
+	m1.OnPeerDead(func(id string) { remote1.RemoveNode(id) })
+	m2.OnPeerDead(func(id string) { remote2.RemoveNode(id) })
+
 	m1.SetSnapshotProvider(func() []MeshSessionBlock {
-		return []MeshSessionBlock{{Callsign: "LOC", IsATC: false}}
+		return []MeshSessionBlock{{Callsign: "LOC", IsATC: false, Trxs: []MeshTrx{{ID: 0, FreqHz: 118700000}}}}
 	})
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	_ = m1.Start(ctx)
-	_ = m2.Start(ctx)
-	// allow ctrl drain for snapshot to m2... actually m1 PublishTrxSnapshot enqueues to n2
-	// m2 delivers to m2.onDir which is nil — set m2's dir to receive n1
-	// reverse: m2 publishes so m1 remote sees it
 	m2.SetSnapshotProvider(func() []MeshSessionBlock {
 		return []MeshSessionBlock{{Callsign: "REM", IsATC: true, Trxs: []MeshTrx{{ID: 0, FreqHz: 1}}}}
 	})
-	m2.OnDirectory(newRemoteDir())
-	m2.PublishTrxSnapshot()
-	// wait for drain
+	m1.SetInterestProvider(func() []InterestEntry {
+		return []InterestEntry{{FreqHz: 1, ILat: 0, ILon: 0}}
+	})
+	m2.SetInterestProvider(func() []InterestEntry {
+		return []InterestEntry{{FreqHz: 1, ILat: 0, ILon: 0}}
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := m1.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := m2.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	// Wait for cross Snapshot delivery (control drain)
 	deadline := time.Now().Add(2 * time.Second)
 	for time.Now().Before(deadline) {
-		if remote.CountOrigin("n2") > 0 {
+		if remote1.CountOrigin("n2") > 0 && remote2.CountOrigin("n1") > 0 {
 			break
 		}
-		// force delivery by also applying directly if drain slow
 		time.Sleep(20 * time.Millisecond)
 	}
-	// death + up
-	m1.SimulatePeerDown("n2")
-	if remote.CountOrigin("n2") != 0 {
-		// OnPeerDead removes — we didn't wire RemoveNode to remote in this test path
-		// SimulatePeerDown calls onPeerDead which we didn't set — set it
+	if remote1.CountOrigin("n2") == 0 {
+		// Start already published; force one more snapshot exchange
+		m2.PublishTrxSnapshot()
+		deadline = time.Now().Add(2 * time.Second)
+		for time.Now().Before(deadline) {
+			if remote1.CountOrigin("n2") > 0 {
+				break
+			}
+			time.Sleep(20 * time.Millisecond)
+		}
 	}
-	// rewire death
-	m1.OnPeerDead(func(id string) { remote.RemoveNode(id) })
-	remote.ApplySnapshot("n2", []RemoteSession{{Callsign: "REM"}})
-	m1.SimulatePeerDown("n2")
-	if remote.CountOrigin("n2") != 0 {
-		t.Fatal("expected purge")
+	if remote1.CountOrigin("n2") == 0 {
+		t.Fatal("expected remote snapshot of n2 on n1 before death")
 	}
+	s, ok := remote1.Session("n2", "REM")
+	if !ok || !s.IsATC {
+		t.Fatalf("REM snapshot missing or wrong: %+v ok=%v", s, ok)
+	}
+
+	// Death purges remote dir
+	m1.SimulatePeerDown("n2")
+	if remote1.CountOrigin("n2") != 0 {
+		t.Fatal("expected purge after peer death")
+	}
+
+	// Reconnect: Snapshot + Interest via provider (no test-only force)
 	m1.SimulatePeerUp("n2")
-	// interest barrier
-	m2.PublishInterest([]InterestEntry{{FreqHz: 1, ILat: 0, ILon: 0}})
+	m2.SimulatePeerUp("n1")
+	// PeerUp re-publishes Snapshot from provider; wait for apply
+	deadline = time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if remote1.CountOrigin("n2") > 0 {
+			break
+		}
+		// m2 must also re-snapshot toward n1 after m1 is up
+		m2.PublishTrxSnapshot()
+		time.Sleep(20 * time.Millisecond)
+	}
+	if remote1.CountOrigin("n2") == 0 {
+		t.Fatal("expected Snapshot re-apply after reconnect")
+	}
+	s, ok = remote1.Session("n2", "REM")
+	if !ok || !s.IsATC || len(s.Trxs) != 1 {
+		t.Fatalf("post-reconnect snapshot %+v ok=%v", s, ok)
+	}
+	// Interest provider path on PeerUp
 	if !m1.PeerWants("n2", 1, geo.CellKey{ILat: 0, ILon: 0}) {
-		t.Fatal("reconnect interest")
+		// PeerUp publishes n2's interest onto n1 when n2.SimulatePeerUp runs
+		// n2.SimulatePeerUp publishes n2's current interest to peers including n1
+		t.Fatal("reconnect interest from provider")
 	}
 }

--- a/internal/afv/mesh_router_test.go
+++ b/internal/afv/mesh_router_test.go
@@ -63,9 +63,8 @@ func TestRouteSyntheticTX_DualLoginSkip(t *testing.T) {
 }

 func TestRouteSyntheticTX_IsXCNoReXC(t *testing.T) {
-	// isXC is handled in handleMeshAudioRelay (primary only); routeSyntheticTX
-	// itself is frequency-primary. Ensure basic route still works with XC flag
-	// ignored at router layer.
+	// isXC: primary synthetic TX only — never XC-again (PR-9 not implemented).
+	// routeSyntheticTX is frequency-primary; second freq is not auto-coupled.
 	cfg := &Config{MaxSessions: 10, MaxSessionsPerCID: 5, RangeDefaultNM: 100}
 	r := newRegistry(cfg)
 	now := time.Now()
@@ -74,22 +73,23 @@ func TestRouteSyntheticTX_IsXCNoReXC(t *testing.T) {
 		ID: 0, Frequency: 118700000, LatDeg: 40.0, LonDeg: -73.0,
 	}})
 	_, _, _ = r.BindUDP(rx, fakeAddr{"2"}, now)
-	// 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
+		{TxID: 1, FreqHz: 119000000, LatDeg: 40.01, LonDeg: -73.01}, // no local RX
 	})
 	if len(recs) != 1 {
-		t.Fatalf("got %d", len(recs))
+		t.Fatalf("primary-only recipients=%d", len(recs))
 	}
-	// Server path with isXC=true still only primary synthetic (no second hop)
+	// handleMeshAudioRelay with IsXC=true: no panic, primary route only (nil udp drops)
 	s := New(cfg, nil, nil, []byte("k"))
+	s.reg = r
 	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}},
+		TxRadios: []RelayTxRadio{
+			{TxID: 0, FreqHz: 118700000, LatDeg: 40.01, LonDeg: -73.01},
+			{TxID: 1, FreqHz: 119000000, LatDeg: 40.01, LonDeg: -73.01},
+		},
 	})
-	// nil udp — no panic; isXC ignored for re-XC
 }

 func TestHandleMeshAudioRelay_NilUDP(t *testing.T) {