diff --git a/internal/afv/mesh_e2e_test.go b/internal/afv/mesh_e2e_test.go index 8181840..e6ce498 100644 --- 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") + } +} diff --git a/internal/afv/mesh_frame.go b/internal/afv/mesh_frame.go index 34e51f9..df96551 100644 --- 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 ( diff --git a/internal/afv/mesh_interest_test.go b/internal/afv/mesh_interest_test.go index 8114259..79db825 100644 --- 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") } diff --git a/internal/afv/mesh_memory_test.go b/internal/afv/mesh_memory_test.go index 1d5f77f..bad73ae 100644 --- 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") } } diff --git a/internal/afv/mesh_router_test.go b/internal/afv/mesh_router_test.go index 5b9349d..9bd3c4e 100644 --- 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) {