diff --git a/.gitignore b/.gitignore index 6e78a6e..ec6bb11 100644 --- 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. diff --git a/internal/afv/mesh.go b/internal/afv/mesh.go index c2013d1..7db813a 100644 --- 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) + } +} diff --git a/internal/afv/mesh_config_test.go b/internal/afv/mesh_config_test.go index df577d5..6239987 100644 --- 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) { diff --git a/internal/afv/mesh_e2e_test.go b/internal/afv/mesh_e2e_test.go index 3bf3b5d..8181840 100644 --- 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") + } } diff --git a/internal/afv/mesh_frame.go b/internal/afv/mesh_frame.go index df3beff..34e51f9 100644 --- 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) diff --git a/internal/afv/mesh_frame_test.go b/internal/afv/mesh_frame_test.go index 14614ac..0e05068 100644 --- 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") diff --git a/internal/afv/mesh_interest_test.go b/internal/afv/mesh_interest_test.go index c38b4e8..8114259 100644 --- 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) { diff --git a/internal/afv/mesh_memory.go b/internal/afv/mesh_memory.go index d442683..30fc631 100644 --- 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 } diff --git a/internal/afv/mesh_memory_test.go b/internal/afv/mesh_memory_test.go index 2f3fccd..1d5f77f 100644 --- 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 } diff --git a/internal/afv/mesh_memory_time.go b/internal/afv/mesh_memory_time.go deleted file mode 100644 index b31ed14..0000000 --- a/internal/afv/mesh_memory_time.go +++ /dev/null @@ -1,7 +0,0 @@ -package afv - -import "time" - -func doSleep(ms int) { - time.Sleep(time.Duration(ms) * time.Millisecond) -} diff --git a/internal/afv/mesh_remotedir.go b/internal/afv/mesh_remotedir.go index 490ca4d..e78a43d 100644 --- 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)) diff --git a/internal/afv/mesh_remotedir_test.go b/internal/afv/mesh_remotedir_test.go index 48ceae4..b62bd10 100644 --- 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") diff --git a/internal/afv/mesh_router_test.go b/internal/afv/mesh_router_test.go index 7db9f32..5b9349d 100644 --- 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) { diff --git a/internal/afv/server.go b/internal/afv/server.go index a39bff5..eaadc74 100644 --- 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()