feat(afv): hybrid mesh TCP control + UDP voice (PR-10b)

Production multi-node AFV mesh: TCP for Hello/HB/directory/Interest and
UDP for AudioRelay so real-time Opus is not subject to TCP HOL. Wire
NewDefault when AFV_CLUSTER_*=true, fail-closed config with
AFV_CLUSTER_VOICE_LISTEN, generation-scoped peer death, and localhost
hybrid e2e. Update design docs and operator wiki.
This commit is contained in:
Reese Norris
2026-07-28 19:03:24 -04:00
parent 227f20b912
commit bdb136b3b4
16 changed files with 4258 additions and 76 deletions

View File

@@ -5,9 +5,10 @@
| **Title** | AFV multi-node mesh — directory sync + AT frame relay |
| **Author** | openfsd design (implementer-ready) |
| **Date** | 2026-07-28 |
| **Status** | **Memory path Implemented** (PR-10 on `dev`: framing, Interest, AudioRelay, MemoryMesh e2e Cases AF, fail-closed `ENABLED` without TCP). **TCP mesh = PR-10b** (not in production binary). |
| **Status** | **Memory path Implemented** (PR-10: framing, Interest, AudioRelay, MemoryMesh e2e Cases AF). **Production mesh = PR-10b hybrid TCP control + UDP voice** (see `docs/design/afv-mesh-pr10b.md`; TCP-only voice **superseded**). |
| **Parent** | `docs/design/afv-server.md` (Clustering, KD-8/9/16/17, PR-10) |
| **Contract** | `docs/design/afv-mesh-implementer-prompt.md` |
| **Production transport** | `docs/design/afv-mesh-pr10b.md` |
---
@@ -16,9 +17,11 @@
| Item | State |
|------|--------|
| Framing, Interest, remote directory, AudioRelay, MemoryMesh e2e | **Landed** (`internal/afv/mesh_*.go`, Cases AF) |
| Fail-closed `AFV_CLUSTER_ENABLED` without TCP | **Landed** |
| Production `TCPMesh` | **PR-10b** — not in binary; enable fails with clear error |
| Operator wiki for multi-node AFV | Single-node ops documented; multi-host voice deferred until TCP |
| Fail-closed incomplete `AFV_CLUSTER_*` | **Landed** |
| Production mesh transport | **PR-10b hybrid TCP control + UDP voice** (TCP-only voice **superseded**) |
| Operator wiki multi-node AFV | Deliverable of PR-10b stack (Configuration + Deployment) |
> **Callout (PR-10b):** Production **AudioRelay (type 20) is UDP only**. Dual TCP voice+control queues apply to **MemoryMesh** tests only for production semantics. See `afv-mesh-pr10b.md`.
## Overview

File diff suppressed because it is too large Load Diff

View File

@@ -5,7 +5,7 @@
| **Document** | AFV (Audio for VATSIM) server — full feature design |
| **Author** | _(design author / implementer)_ |
| **Date** | 2026-07-28 |
| **Status** | **P0 Implemented** (single-node REST + UDP on `dev`; design history retained). Mesh: MemoryMesh path landed (PR-10); production TCP = PR-10b. |
| **Status** | **P0 Implemented** (single-node REST + UDP on `dev`; design history retained). Mesh: MemoryMesh path landed (PR-10); production hybrid mesh = PR-10b (`docs/design/afv-mesh-pr10b.md`). |
| **Project** | openfsd |
| **Target land path** | `docs/design/afv-server.md` |
| **Related** | `Agents.md`, `docs/design/distributed-openfsd.md`, `internal/auth`, `internal/geo`, `internal/cluster`, `internal/db`, `cmd/openfsd/main.go`, [AFV-Native](https://github.com/xsquawkbox/AFV-Native) (BSD-3) |
@@ -20,8 +20,8 @@
|-------|--------|
| P0 single-node (`pkg/afvprotocol`, `internal/afv`, `-afv`) | **Landed** — REST auth/callsign/transceivers; UDP H/HA + AT→AR; range model; reaper; empty stations route |
| PR-10 mesh (MemoryMesh + framing + Interest + AudioRelay) | **Landed** (in-process / test mesh; e2e Cases AF) |
| PR-10b production TCP mesh | **Not landed**`AFV_CLUSTER_ENABLED=true` fails closed without TCP in binary |
| Operator docs | Wiki Configuration/Deployment AFV sections; root README |
| PR-10b production hybrid mesh | **Landed**TCP control + UDP AudioRelay (`HybridMesh`); see `docs/design/afv-mesh-pr10b.md` |
| Operator docs | Wiki Configuration/Deployment AFV sections (incl. multi-node hybrid mesh); root README |
Design body below is the engineering record (decisions, protocol, PR plan). Prefer wiki + `internal/afv` for day-to-day ops.
@@ -891,26 +891,27 @@ Why separate from `internal/cluster`: import allowlist, rate, interest model.
**Goals:** in-process / localhost **memory mesh** or TCP between two nodes; directory sync; AT relay; e2e two clients on different nodes hear each other in range.
##### Normative mesh framing (KD-17) — first cut
##### Normative mesh framing (KD-17) — production transport (PR-10b)
| Item | Spec |
|------|------|
| Transport | TCP between static peers (`AFV_CLUSTER_PEERS=id=host:port,...`) |
| Frame | `[u32 BE length][u8 type][payload]` where length = 1+len(payload); max payload **1 MiB** |
| Auth | First message must be Hello with HMAC-SHA256(PSK, nodeID\|\|nonce) or simple constant-time compare of shared `AFV_CLUSTER_PSK` on Hello (document chosen scheme in PR); reject peers failing auth |
| Types | `1=Hello`, `2=Heartbeat`, `10=TrxSnapshot`, `11=TrxDelta`, `12=SessionLeave`, `20=AudioRelay`, `30=Interest` |
| Heartbeat | every 2s; peer-death after 15s → drop remote directory entries for that node; local sessions unaffected |
| **Control transport** | **TCP** between static peers (`AFV_CLUSTER_LISTEN`, `AFV_CLUSTER_PEERS=id=host:tcpPort[/voicePort],...`) |
| **AudioRelay (type 20)** | **UDP mesh voice only** (`AFV_CLUSTER_VOICE_LISTEN` + peer voice addrs); **never on TCP** |
| Frame | `[u32 BE length][u8 type][payload]` where length = 1+len(payload); TCP max payload **1 MiB**; UDP voice max **16 KiB** datagram |
| Auth | First **TCP** frame must be Hello; constant-time compare of shared `AFV_CLUSTER_PSK`; reject peers failing auth |
| Types | `1=Hello`, `2=Heartbeat`, `10=TrxSnapshot`, `11=TrxDelta`, `12=SessionLeave`, `20=AudioRelay` (**UDP only**), `30=Interest` |
| Heartbeat | every 2s on TCP; peer-death after 15s (or TCP read/write error) → drop remote directory entries for that node; local sessions unaffected |
| Interest | each node advertises set of `(freqHz, cellKey)` for local RX; rate ≤ 2 Hz; cap **4096** entries (drop coarsest cells if over) |
| AudioRelay payload | msgpack/json binary: originNode, callsign, seq, audio, last, isXC, []{freq, lat, lon, alt, txID} — **never client AEAD keys** |
| Relay queues | per-peer outbound channel depth **256**; **drop oldest AudioRelay** under pressure; never block UDP hot path on full mesh queue |
| Deadlock | never hold registry lock while sending on mesh TCP |
| Max peers first cut | **≤ 4** (document; like FSD mesh caution) |
| Sticky LB | required: REST sticky by CID or per-node API only advertises local UDP |
| Dual-login | per-region uniqueness only in first cut; no cluster-wide claim |
| AudioRelay payload | originNode, callsign, seq, audio, last, isATC, isXC, []{freq, lat, lon, alt, txID} — **never client AEAD keys** |
| Relay queues | voice depth **256** drop-oldest (UDP); control depth **64** drop-oldest (TCP); never block client UDP hot path |
| Deadlock | never hold registry lock while sending on mesh TCP/UDP |
| Max peers | **≤ 4** remote |
| Sticky LB | required: REST sticky by CID or per-node API only advertises local client UDP (`AFV_UDP_ADVERTISE_IPV4`) |
| Dual-login | per-region uniqueness only; no cluster-wide claim |
Keys never leave the home node. Local node re-runs range model for **local** RX only and encrypts AR with local keys.
**PR plan:** PR-10 is **framing + memory mesh two-node e2e only**. Production TCP polish / interest caps / ops docs may split to PR-10b after a short design addendum if scope grows. Full FSD-mesh-scale work is **out of band** relative to P0.
**Normative production design:** `docs/design/afv-mesh-pr10b.md` (hybrid TCP control + UDP voice). MemoryMesh remains tests-only.
---

View File

@@ -77,13 +77,23 @@ func NewDefault(ctx context.Context) (*Server, error) {
if err := cfg.ValidateCluster(); err != nil {
return nil, err
}
// M-11: ENABLED=true without TCP mesh in this binary fails closed.
// Never wire production NewDefault to MemoryMesh.
s := New(cfg, repos.UserRepo, repos.ConfigRepo, jwtSecret)
// Production hybrid mesh (PR-10b): TCP control + UDP voice. Never MemoryMesh.
if cfg.ClusterEnabled {
return nil, errClusterTCPNotBuilt
hmCfg, err := hybridConfigFrom(cfg)
if err != nil {
return nil, fmt.Errorf("afv hybrid mesh: %w", err)
}
hm, err := NewHybridMesh(hmCfg)
if err != nil {
return nil, fmt.Errorf("afv hybrid mesh failed to construct: %w", err)
}
s.SetMesh(hm)
}
return New(cfg, repos.UserRepo, repos.ConfigRepo, jwtSecret), nil
return s, nil
}
func resolveJWTSecret(ctx context.Context, cfg *Config, kv db.ConfigRepository) ([]byte, error) {

View File

@@ -37,6 +37,7 @@ type Config struct {
ClusterEnabled bool `env:"AFV_CLUSTER_ENABLED, default=false"`
ClusterNodeID string `env:"AFV_CLUSTER_NODE_ID"`
ClusterListen string `env:"AFV_CLUSTER_LISTEN"`
ClusterVoiceListen string `env:"AFV_CLUSTER_VOICE_LISTEN"`
ClusterPeers string `env:"AFV_CLUSTER_PEERS"`
ClusterPSK string `env:"AFV_CLUSTER_PSK"`

View File

@@ -1,6 +1,6 @@
package afv
// AFV multi-node mesh (PR-10 / KD-17).
// AFV multi-node mesh (PR-10 / PR-10b / KD-17).
//
// Ops: sticky LB required — each node mints unique ChannelTag + AEAD keys and
// advertises its own AFV_UDP_ADVERTISE_IPV4. Clients must use the same node for
@@ -8,14 +8,15 @@ package afv
// replace sticky affinity. AEAD keys never leave the home node; cross-node
// audio is re-encrypted at the listener's home with local ClientRxKey only.
//
// Production: AFV_CLUSTER_ENABLED=true without TCP mesh in this binary fails
// closed at startup (MemoryMesh is tests-only via SetMesh).
// Production: HybridMesh (TCP control + UDP voice) when AFV_CLUSTER_ENABLED=true
// with valid VOICE_LISTEN + peers. MemoryMesh is tests-only via SetMesh.
import (
"context"
"fmt"
"log/slog"
"net"
"strconv"
"strings"
"time"
@@ -24,7 +25,7 @@ import (
)
// Mesh is the AFV inter-node fabric. Implementations: MemoryMesh (tests);
// TCPMesh is PR-10b / optional.
// HybridMesh (production: TCP control + UDP AudioRelay).
//
// All publish methods must be non-blocking w.r.t. the UDP hot path (enqueue or
// drop). Callers must not hold registry locks across Publish*/Enqueue*.
@@ -103,7 +104,7 @@ type RelayTxRadio struct {
HeightM float64
}
// MeshConfig is shared construction config for MemoryMesh / future TCPMesh.
// MeshConfig is shared construction config for MemoryMesh.
type MeshConfig struct {
NodeID string
PSK string
@@ -115,17 +116,19 @@ type MeshConfig struct {
const (
interestMinInterval = 500 * time.Millisecond // ≤ 2 Hz
defaultPeerDeathGrace = 15 * time.Second
defaultMeshHeartbeat = 2 * time.Second
)
// interestMaxEntries is the Interest set cap (M-5). Var so tests can lower it.
var interestMaxEntries = 4096
// --- Cluster config validation (M-11) ---
// --- Cluster config validation (M-11 / H-20) ---
// ClusterPeer is one static remote peer (id + host:port).
// ClusterPeer is one static remote peer (TCP control + UDP voice).
type ClusterPeer struct {
ID string
Addr string
ID string
Addr string // TCP host:port (control) — SplitHostPort-valid
VoiceAddr string // UDP host:port (voice) — derived or explicit
}
// ValidateCluster fails closed when cluster is enabled with incomplete config.
@@ -140,9 +143,32 @@ func (c *Config) ValidateCluster() error {
if listen == "" {
return fmt.Errorf("AFV_CLUSTER_LISTEN required when AFV_CLUSTER_ENABLED=true")
}
if _, _, err := net.SplitHostPort(listen); err != nil {
tcpHost, tcpPortStr, err := net.SplitHostPort(listen)
if err != nil {
return fmt.Errorf("AFV_CLUSTER_LISTEN: want host:port: %w", err)
}
tcpPort, err := parseUint16Port(tcpPortStr)
if err != nil || tcpPort == 0 {
return fmt.Errorf("AFV_CLUSTER_LISTEN: invalid port")
}
_ = tcpHost
voiceListen := strings.TrimSpace(c.ClusterVoiceListen)
if voiceListen == "" {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN required when AFV_CLUSTER_ENABLED=true")
}
voiceHost, voicePortStr, err := net.SplitHostPort(voiceListen)
if err != nil {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN: want host:port: %w", err)
}
voicePort, err := parseUint16Port(voicePortStr)
if err != nil || voicePort == 0 {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN: invalid port (port 0 rejected)")
}
if voicePortStr == tcpPortStr {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN: must differ from AFV_CLUSTER_LISTEN port")
}
if strings.TrimSpace(c.ClusterPSK) == "" {
return fmt.Errorf("AFV_CLUSTER_PSK required when AFV_CLUSTER_ENABLED=true")
}
@@ -157,7 +183,11 @@ func (c *Config) ValidateCluster() error {
return fmt.Errorf("AFV_CLUSTER_PEERS: max 4 remote peers, got %d", len(peers))
}
self := strings.TrimSpace(c.ClusterNodeID)
seen := make(map[string]struct{}, len(peers))
seenID := make(map[string]struct{}, len(peers))
seenVoice := make(map[string]struct{}, len(peers))
localVoiceNorm := net.JoinHostPort(voiceHost, voicePortStr)
localTCPNorm := net.JoinHostPort(tcpHost, tcpPortStr)
for _, p := range peers {
if p.ID == self {
return fmt.Errorf("AFV_CLUSTER_PEERS: must not include self node id %q", self)
@@ -168,15 +198,97 @@ func (c *Config) ValidateCluster() error {
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 {
if _, ok := seenID[p.ID]; ok {
return fmt.Errorf("AFV_CLUSTER_PEERS: duplicate peer id %q", p.ID)
}
seen[p.ID] = struct{}{}
seenID[p.ID] = struct{}{}
if _, ok := seenVoice[p.VoiceAddr]; ok {
return fmt.Errorf("AFV_CLUSTER_PEERS: duplicate peer VoiceAddr %q", p.VoiceAddr)
}
seenVoice[p.VoiceAddr] = struct{}{}
// H-20: full host:port equality only for local voice vs peer voice.
if p.VoiceAddr == localVoiceNorm {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN equals peer %q VoiceAddr %q", p.ID, p.VoiceAddr)
}
// Optional same-host loopback tightening.
if sameLoopbackOrWildcardVoiceCollision(voiceHost, voicePortStr, p.VoiceAddr) {
return fmt.Errorf("AFV_CLUSTER_VOICE_LISTEN collides with peer %q loopback VoiceAddr %q", p.ID, p.VoiceAddr)
}
if p.Addr == localTCPNorm {
return fmt.Errorf("AFV_CLUSTER_LISTEN equals peer %q Addr %q", p.ID, p.Addr)
}
}
// H-20: ClusterVoiceListen ≠ UDPListen (client CryptoDTO socket).
if err := rejectSameSocket(c.ClusterVoiceListen, c.UDPListen, "AFV_CLUSTER_VOICE_LISTEN", "AFV_UDP_LISTEN"); err != nil {
return err
}
return nil
}
// parseClusterPeers parses "id=host:port,id2=host:port2".
// sameLoopbackOrWildcardVoiceCollision catches localhost multi-process when local
// binds loopback/wildcard and peer is loopback with the same voice port.
func sameLoopbackOrWildcardVoiceCollision(localHost, localPort, peerVoice string) bool {
ph, pp, err := net.SplitHostPort(peerVoice)
if err != nil || pp != localPort {
return false
}
if !isLoopbackHost(ph) {
return false
}
return isLoopbackHost(localHost) || isWildcardHost(localHost)
}
func isLoopbackHost(h string) bool {
h = strings.TrimSpace(h)
if h == "localhost" || h == "127.0.0.1" || h == "::1" {
return true
}
ip := net.ParseIP(h)
return ip != nil && ip.IsLoopback()
}
func isWildcardHost(h string) bool {
h = strings.TrimSpace(h)
return h == "" || h == "0.0.0.0" || h == "::" || h == "*"
}
// rejectSameSocket fails if two listen specs collide (wildcard-equal hosts if ports equal).
func rejectSameSocket(a, b, nameA, nameB string) error {
a = strings.TrimSpace(a)
b = strings.TrimSpace(b)
if a == "" || b == "" {
return nil
}
ah, ap, errA := net.SplitHostPort(a)
bh, bp, errB := net.SplitHostPort(b)
if errA != nil || errB != nil {
return nil // other validators handle format
}
if ap != bp {
return nil
}
if ah == bh || isWildcardHost(ah) || isWildcardHost(bh) {
return fmt.Errorf("%s must differ from %s (same host:port / wildcard port collision)", nameA, nameB)
}
// concrete equal already covered by ah==bh
return nil
}
func parseUint16Port(s string) (uint16, error) {
s = strings.TrimSpace(s)
if s == "" {
return 0, fmt.Errorf("empty port")
}
n, err := strconv.ParseUint(s, 10, 16)
if err != nil {
return 0, err
}
return uint16(n), nil
}
// parseClusterPeers parses AFV_CLUSTER_PEERS with optional /voicePort (H-4).
// Grammar: id=host:tcpPort[/voicePort],... — strip /voice before SplitHostPort.
func parseClusterPeers(s string) ([]ClusterPeer, error) {
s = strings.TrimSpace(s)
if s == "" {
@@ -194,19 +306,65 @@ func parseClusterPeers(s string) ([]ClusterPeer, error) {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: invalid entry %q (want id=host:port)", p)
}
id := strings.TrimSpace(p[:eq])
addr := strings.TrimSpace(p[eq+1:])
if id == "" || addr == "" {
rest := strings.TrimSpace(p[eq+1:])
if id == "" || rest == "" {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: invalid entry %q", p)
}
out = append(out, ClusterPeer{ID: id, Addr: addr})
hostPort := rest
var voicePort uint16
voiceSet := false
if slash := strings.LastIndexByte(rest, '/'); slash >= 0 {
// Extra '/' in the hostPort side is rejected by SplitHostPort or empty checks.
if strings.Count(rest, "/") > 1 {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: invalid entry %q (extra /)", p)
}
hostPort = strings.TrimSpace(rest[:slash])
voicePortStr := strings.TrimSpace(rest[slash+1:])
if hostPort == "" || voicePortStr == "" {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: empty voice port in %q", p)
}
vp, err := parseUint16Port(voicePortStr)
if err != nil {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: non-numeric voice port in %q", p)
}
if vp == 0 {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: port 0 rejected in %q", p)
}
voicePort = vp
voiceSet = true
}
host, tcpPortStr, err := net.SplitHostPort(hostPort)
if err != nil {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: peer %q addr want host:port: %w", id, err)
}
if strings.TrimSpace(host) == "" {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: peer %q empty host", id)
}
tcpPort, err := parseUint16Port(tcpPortStr)
if err != nil {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: peer %q non-numeric tcp port", id)
}
if tcpPort == 0 {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: port 0 rejected for peer %q", id)
}
if !voiceSet {
if tcpPort == 65535 {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: voice port overflow for peer %q", id)
}
voicePort = tcpPort + 1
}
if voicePort == 0 {
return nil, fmt.Errorf("AFV_CLUSTER_PEERS: port 0 rejected for peer %q voice", id)
}
voiceAddr := net.JoinHostPort(host, strconv.Itoa(int(voicePort)))
// Preserve SplitHostPort-valid form of hostPort (brackets for IPv6).
out = append(out, ClusterPeer{ID: id, Addr: hostPort, VoiceAddr: voiceAddr})
}
return out, nil
}
// errClusterTCPNotBuilt is returned when ENABLED=true but TCP mesh is not available.
var errClusterTCPNotBuilt = fmt.Errorf(
"AFV_CLUSTER_ENABLED=true but AFV mesh TCP is not built in this binary (use tests with SetMesh(MemoryMesh) or enable PR-10b TCP)")
// --- Server mesh hooks ---
// SetMesh injects a Mesh (tests: MemoryMesh). Must be called before Run.

View File

@@ -13,11 +13,13 @@ func TestValidateCluster(t *testing.T) {
t.Fatal(err)
}
base := &Config{
ClusterEnabled: true,
ClusterNodeID: "n1",
ClusterListen: "127.0.0.1:9100",
ClusterPeers: "n2=127.0.0.1:9101",
ClusterPSK: "psk",
ClusterEnabled: true,
ClusterNodeID: "n1",
ClusterListen: "127.0.0.1:9100",
ClusterVoiceListen: "127.0.0.1:9101",
ClusterPeers: "n2=127.0.0.1:9110/9111",
ClusterPSK: "psk",
UDPListen: "127.0.0.1:50000",
}
if err := base.ValidateCluster(); err != nil {
t.Fatal(err)
@@ -29,16 +31,25 @@ func TestValidateCluster(t *testing.T) {
}{
{"no node", func(c *Config) { c.ClusterNodeID = "" }, "NODE_ID"},
{"no listen", func(c *Config) { c.ClusterListen = "" }, "LISTEN"},
{"no voice", func(c *Config) { c.ClusterVoiceListen = "" }, "VOICE_LISTEN"},
{"no psk", func(c *Config) { c.ClusterPSK = "" }, "PSK"},
{"no peers", func(c *Config) { c.ClusterPeers = "" }, "PEERS"},
{"self peer", func(c *Config) { c.ClusterPeers = "n1=127.0.0.1:1" }, "self"},
{"self peer", func(c *Config) { c.ClusterPeers = "n1=127.0.0.1:9110/9111" }, "self"},
{"too many", func(c *Config) {
c.ClusterPeers = "a=1:1,b=1:2,c=1:3,d=1:4,e=1:5"
c.ClusterPeers = "a=1.1.1.1:1/2,b=1.1.1.1:3/4,c=1.1.1.1:5/6,d=1.1.1.1:7/8,e=1.1.1.1:9/10"
}, "max 4"},
{"bad peer", func(c *Config) { c.ClusterPeers = "notvalid" }, "invalid"},
{"dup", func(c *Config) { c.ClusterPeers = "n2=1:1,n2=1:2" }, "duplicate"},
{"dup", func(c *Config) { c.ClusterPeers = "n2=1.1.1.1:1/2,n2=1.1.1.1:3/4" }, "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"},
{"voice same port as tcp", func(c *Config) { c.ClusterVoiceListen = "127.0.0.1:9100" }, "differ"},
{"voice equals peer", func(c *Config) {
c.ClusterVoiceListen = "127.0.0.1:9111"
}, "VoiceAddr"},
{"voice equals udp", func(c *Config) {
c.ClusterVoiceListen = "127.0.0.1:50000"
}, "UDP_LISTEN"},
{"voice port 0", func(c *Config) { c.ClusterVoiceListen = "127.0.0.1:0" }, "port"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
@@ -52,21 +63,107 @@ func TestValidateCluster(t *testing.T) {
}
}
func TestNewDefault_ClusterEnabledFailsClosed(t *testing.T) {
func TestValidateCluster_MultiHostSameVoicePortOK(t *testing.T) {
// H-20 rev3: local 0.0.0.0:17001 + peer 10.0.0.2:17001 is OK
c := &Config{
ClusterEnabled: true,
ClusterNodeID: "n1",
ClusterListen: "0.0.0.0:17000",
ClusterVoiceListen: "0.0.0.0:17001",
ClusterPeers: "n2=10.0.0.2:17000", // voice derives 10.0.0.2:17001
ClusterPSK: "psk",
UDPListen: "0.0.0.0:50000",
}
if err := c.ValidateCluster(); err != nil {
t.Fatal(err)
}
}
func TestValidateCluster_DupVoiceAddr(t *testing.T) {
c := &Config{
ClusterEnabled: true,
ClusterNodeID: "n1",
ClusterListen: "0.0.0.0:17000",
ClusterVoiceListen: "0.0.0.0:17001",
// Same VoiceAddr host:port on two peers (H-20 unique VoiceAddr)
ClusterPeers: "n2=10.0.0.2:17000/17001,n3=10.0.0.2:17010/17001",
ClusterPSK: "psk",
UDPListen: "0.0.0.0:50000",
}
err := c.ValidateCluster()
if err == nil || !strings.Contains(err.Error(), "duplicate peer VoiceAddr") {
t.Fatalf("err=%v", err)
}
}
func TestParseClusterPeers(t *testing.T) {
fixtures := []struct {
in string
wantID string
wantAddr string
wantVoice string
wantErr string
}{
{"n2=10.0.0.2:17000", "n2", "10.0.0.2:17000", "10.0.0.2:17001", ""},
{"n2=10.0.0.2:17000/17100", "n2", "10.0.0.2:17000", "10.0.0.2:17100", ""},
{"n2=[::1]:17000", "n2", "[::1]:17000", "[::1]:17001", ""},
{"n2=[::1]:17000/17050", "n2", "[::1]:17000", "[::1]:17050", ""},
{" n2 = 10.0.0.2:17000/ 17100 ", "n2", "10.0.0.2:17000", "10.0.0.2:17100", ""},
{"n2=10.0.0.2:65535", "", "", "", "overflow"},
{"n2=10.0.0.2:17000/", "", "", "", "empty voice"},
{"n2=10.0.0.2:17000/abc", "", "", "", "non-numeric"},
{"n2=10.0.0.2:17000/0", "", "", "", "port 0"},
{"n2=10.0.0.2:0", "", "", "", "port 0"},
{"n2=::1:17000", "", "", "", "host:port"},
}
for _, tc := range fixtures {
t.Run(tc.in, func(t *testing.T) {
p, err := parseClusterPeers(tc.in)
if tc.wantErr != "" {
if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
t.Fatalf("err=%v want %q", err, tc.wantErr)
}
return
}
if err != nil || len(p) != 1 {
t.Fatalf("p=%+v err=%v", p, err)
}
if p[0].ID != tc.wantID || p[0].Addr != tc.wantAddr || p[0].VoiceAddr != tc.wantVoice {
t.Fatalf("got %+v want id=%s addr=%s voice=%s", p[0], tc.wantID, tc.wantAddr, tc.wantVoice)
}
})
}
// multi
p, err := parseClusterPeers("a=1.2.3.4:5, b=6.7.8.9:10")
if err != nil || len(p) != 2 || p[0].ID != "a" || p[0].VoiceAddr != "1.2.3.4:6" {
t.Fatalf("%+v %v", p, err)
}
}
func TestNewDefault_ClusterEnabledWiresHybrid(t *testing.T) {
dir := t.TempDir()
t.Setenv("DATABASE_DRIVER", "sqlite")
t.Setenv("DATABASE_SOURCE_NAME", filepath.Join(dir, "x.db"))
t.Setenv("DATABASE_AUTO_MIGRATE", "true")
t.Setenv("AFV_UDP_ADVERTISE_IPV4", "127.0.0.1:50000")
t.Setenv("AFV_UDP_LISTEN", "127.0.0.1:50000")
t.Setenv("AFV_JWT_SECRET", "bootstrap-test-secret-key-material")
t.Setenv("AFV_CLUSTER_ENABLED", "true")
t.Setenv("AFV_CLUSTER_NODE_ID", "n1")
t.Setenv("AFV_CLUSTER_LISTEN", "127.0.0.1:9100")
t.Setenv("AFV_CLUSTER_PEERS", "n2=127.0.0.1:9101")
t.Setenv("AFV_CLUSTER_LISTEN", "127.0.0.1:19100")
t.Setenv("AFV_CLUSTER_VOICE_LISTEN", "127.0.0.1:19101")
t.Setenv("AFV_CLUSTER_PEERS", "n2=127.0.0.1:19110/19111")
t.Setenv("AFV_CLUSTER_PSK", "psk")
_, err := NewDefault(context.Background())
if err == nil || !strings.Contains(err.Error(), "TCP") {
t.Fatalf("err=%v", err)
s, err := NewDefault(context.Background())
if err != nil {
t.Fatal(err)
}
if s.Mesh() == nil {
t.Fatal("expected non-nil Mesh")
}
if _, ok := s.Mesh().(*HybridMesh); !ok {
t.Fatalf("want *HybridMesh got %T", s.Mesh())
}
}
@@ -85,9 +182,19 @@ func TestNewDefault_ClusterIncompleteConfig(t *testing.T) {
}
}
func TestParseClusterPeers(t *testing.T) {
p, err := parseClusterPeers("a=1.2.3.4:5, b=6.7.8.9:10")
if err != nil || len(p) != 2 || p[0].ID != "a" {
t.Fatalf("%+v %v", p, err)
func TestNewDefault_ClusterDisabledNilMesh(t *testing.T) {
dir := t.TempDir()
t.Setenv("DATABASE_DRIVER", "sqlite")
t.Setenv("DATABASE_SOURCE_NAME", filepath.Join(dir, "z.db"))
t.Setenv("DATABASE_AUTO_MIGRATE", "true")
t.Setenv("AFV_UDP_ADVERTISE_IPV4", "127.0.0.1:50000")
t.Setenv("AFV_JWT_SECRET", "bootstrap-test-secret-key-material")
t.Setenv("AFV_CLUSTER_ENABLED", "false")
s, err := NewDefault(context.Background())
if err != nil {
t.Fatal(err)
}
if s.Mesh() != nil {
t.Fatal("expected nil mesh when disabled")
}
}

View File

@@ -1,16 +1,15 @@
package afv
// AFV multi-node mesh framing (PR-10 / KD-17).
// AFV multi-node mesh framing (PR-10 / PR-10b / KD-17).
//
// Hello auth (M-3): constant-time compare of shared PSK. Trust root and payload
// 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.
// Hello first-frame rule (TCP control plane, HybridMesh): 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.
// AudioRelay (type 20) is forbidden on TCP (H-2); production voice is UDP only.
//
// Local length-prefix framing — do not import internal/cluster.
@@ -103,6 +102,46 @@ func DecodeMeshFrame(r io.Reader) (MeshFrame, error) {
return MeshFrame{Type: typ, Payload: payload}, nil
}
// EncodeMeshFrameBytes returns the full length-prefixed mesh frame as a byte slice.
func EncodeMeshFrameBytes(typ byte, payload []byte) ([]byte, error) {
if len(payload) > MaxMeshPayload {
return nil, errMeshFrameTooLarge
}
n := uint32(1 + len(payload))
out := make([]byte, 5+len(payload))
binary.BigEndian.PutUint32(out[0:4], n)
out[4] = typ
copy(out[5:], payload)
return out, nil
}
// DecodeMeshFrameExact decodes exactly one frame from b with zero trailing bytes
// (H-3: decoded frame length == len(b)). Used for UDP voice datagrams.
func DecodeMeshFrameExact(b []byte) (MeshFrame, error) {
if len(b) < 5 {
return MeshFrame{}, errMeshShortFrame
}
n := binary.BigEndian.Uint32(b[0:4])
if n == 0 {
return MeshFrame{}, errMeshShortFrame
}
if n > maxMeshFrameTotal {
return MeshFrame{}, errMeshFrameTooLarge
}
// 4-byte length prefix + n body bytes must consume the whole datagram.
if 4+int(n) != len(b) {
return MeshFrame{}, errMeshBadPayload
}
typ := b[4]
payLen := int(n) - 1
var payload []byte
if payLen > 0 {
payload = make([]byte, payLen)
copy(payload, b[5:5+payLen])
}
return MeshFrame{Type: typ, Payload: payload}, nil
}
// --- binary helpers (local copies; no cluster import) ---
func meshEncodeString(b []byte, s string) []byte {

752
internal/afv/mesh_hybrid.go Normal file
View File

@@ -0,0 +1,752 @@
package afv
// HybridMesh is the production AFV multi-node fabric (PR-10b):
// TCP control plane (Hello, HB, Trx*, Leave, Interest) + UDP voice (AudioRelay only).
// MemoryMesh remains tests-only.
//
// Lifecycle: single Start/Stop per instance (not restartable after Stop).
import (
"context"
"fmt"
"log/slog"
"net"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/renorris/openfsd/internal/geo"
)
// HybridMeshConfig configures production hybrid mesh.
type HybridMeshConfig struct {
NodeID string
ListenTCP string // AFV_CLUSTER_LISTEN
ListenVoice string // AFV_CLUSTER_VOICE_LISTEN
Peers []ClusterPeer
PSK string
HeartbeatInterval time.Duration // default 2s
PeerDeathAfter time.Duration // default 15s
CtrlQueueDepth int // default 64
VoiceQueueDepth int // default 256
Logger *slog.Logger
}
// HybridMesh implements Mesh over TCP control + UDP AudioRelay voice.
type HybridMesh struct {
cfg HybridMeshConfig
peerDeathGrace time.Duration
hbInterval time.Duration
ctrlDepth int
voiceDepth int
logger *slog.Logger
// peerID → static config
peerCfg map[string]ClusterPeer
mu sync.RWMutex
// peerInterest filled only from inbound TCP Interest (H-14)
peerInterest map[string]map[FreqCell]struct{}
peers map[string]*hybridPeer
// UDP allowlist: canonical key → peerID
allowlist map[string]string
// resolved send targets
sendVoice map[string]*net.UDPAddr
ln net.Listener
voicePC net.PacketConn
onAudio func(fromNode string, r AudioRelay)
onPeerDead func(nodeID string)
onDir MeshDirectoryHandler
snapFn func() []MeshSessionBlock
interestFn func() []InterestEntry
// counters
voiceTx atomic.Uint64
voiceRx atomic.Uint64
udpAllowDrops atomic.Uint64
udpOversizeDrops atomic.Uint64
inboundRateDrops atomic.Uint64
voiceQueueDrops atomic.Uint64
ctrlQueueDrops atomic.Uint64
udpSendErrs atomic.Uint64
lastMuteCheck atomic.Int64 // unix for silent-mute debug
stopCh chan struct{}
stopOnce sync.Once
wg sync.WaitGroup
started atomic.Bool
}
// hybridPeer is per-remote-peer control + voice state.
type hybridPeer struct {
id string
tcpAddr string
voiceAddr string
mu sync.Mutex
conn net.Conn
ctrlQ *dropOldestQueue[meshCtrlJob]
voiceQ *dropOldestQueue[voiceJob]
lastHB time.Time
authed bool
stop chan struct{} // closed to stop writer/drainer for this conn gen
// inbound rate (H-19)
rateMu sync.Mutex
rateWindow time.Time
rateCount int
voiceTxCount atomic.Uint64
voiceRxCount atomic.Uint64
// lastVoiceUnixNano: last successful voice tx or rx (mute watch Issue 15)
lastVoiceUnixNano atomic.Int64
}
type voiceJob struct {
packet []byte // exclusive copy of full mesh frame
}
// NewHybridMesh constructs a HybridMesh (not started).
func NewHybridMesh(cfg HybridMeshConfig) (*HybridMesh, error) {
if strings.TrimSpace(cfg.NodeID) == "" {
return nil, fmt.Errorf("afv hybrid mesh: empty NodeID")
}
if strings.TrimSpace(cfg.ListenTCP) == "" || strings.TrimSpace(cfg.ListenVoice) == "" {
return nil, fmt.Errorf("afv hybrid mesh: ListenTCP and ListenVoice required")
}
if strings.TrimSpace(cfg.PSK) == "" {
return nil, fmt.Errorf("afv hybrid mesh: PSK required")
}
if len(cfg.Peers) == 0 {
return nil, fmt.Errorf("afv hybrid mesh: at least one peer required")
}
if len(cfg.Peers) > 4 {
return nil, fmt.Errorf("afv hybrid mesh: max 4 peers")
}
pg := cfg.PeerDeathAfter
if pg <= 0 {
pg = defaultPeerDeathGrace
}
hb := cfg.HeartbeatInterval
if hb <= 0 {
hb = defaultMeshHeartbeat
}
cd := cfg.CtrlQueueDepth
if cd <= 0 {
cd = meshControlQueueDepth
}
vd := cfg.VoiceQueueDepth
if vd <= 0 {
vd = meshVoiceQueueDepth
}
log := cfg.Logger
if log == nil {
log = slog.Default()
}
peerCfg := make(map[string]ClusterPeer, len(cfg.Peers))
for _, p := range cfg.Peers {
if p.ID == "" || p.ID == cfg.NodeID {
return nil, fmt.Errorf("afv hybrid mesh: invalid peer id %q", p.ID)
}
if _, ok := peerCfg[p.ID]; ok {
return nil, fmt.Errorf("afv hybrid mesh: duplicate peer %q", p.ID)
}
peerCfg[p.ID] = p
}
return &HybridMesh{
cfg: cfg,
peerDeathGrace: pg,
hbInterval: hb,
ctrlDepth: cd,
voiceDepth: vd,
logger: log,
peerCfg: peerCfg,
peerInterest: make(map[string]map[FreqCell]struct{}),
peers: make(map[string]*hybridPeer),
allowlist: make(map[string]string),
sendVoice: make(map[string]*net.UDPAddr),
stopCh: make(chan struct{}),
}, nil
}
// hybridConfigFrom builds HybridMeshConfig from AFV Config (bootstrap).
func hybridConfigFrom(cfg *Config) (HybridMeshConfig, error) {
peers, err := parseClusterPeers(cfg.ClusterPeers)
if err != nil {
return HybridMeshConfig{}, err
}
return HybridMeshConfig{
NodeID: cfg.ClusterNodeID,
ListenTCP: cfg.ClusterListen,
ListenVoice: cfg.ClusterVoiceListen,
Peers: peers,
PSK: cfg.ClusterPSK,
}, nil
}
func (m *HybridMesh) NodeID() string {
if m == nil {
return ""
}
return m.cfg.NodeID
}
func (m *HybridMesh) OnAudioRelay(fn func(fromNode string, r AudioRelay)) {
m.mu.Lock()
m.onAudio = fn
m.mu.Unlock()
}
func (m *HybridMesh) OnPeerDead(fn func(nodeID string)) {
m.mu.Lock()
m.onPeerDead = fn
m.mu.Unlock()
}
func (m *HybridMesh) OnDirectory(fn MeshDirectoryHandler) {
m.mu.Lock()
m.onDir = fn
m.mu.Unlock()
}
func (m *HybridMesh) SetSnapshotProvider(fn func() []MeshSessionBlock) {
m.mu.Lock()
m.snapFn = fn
m.mu.Unlock()
}
func (m *HybridMesh) SetInterestProvider(fn func() []InterestEntry) {
m.mu.Lock()
m.interestFn = fn
m.mu.Unlock()
}
// PeerWants reports whether peer advertised Interest for freq/cell (inbound TCP only).
func (m *HybridMesh) PeerWants(peerID string, freqHz uint32, cell geo.CellKey) bool {
if m == nil {
return false
}
m.mu.RLock()
defer m.mu.RUnlock()
set := m.peerInterest[peerID]
if len(set) == 0 {
return false
}
_, ok := set[FreqCell{FreqHz: freqHz, Cell: cell}]
return ok
}
// InterestedPeers returns peer IDs that want any of the given keys (authed only).
func (m *HybridMesh) InterestedPeers(keys []FreqCell) []string {
if m == nil {
return nil
}
// Snapshot under m.mu only — never nest p.mu.
type snap struct {
id string
p *hybridPeer
set map[FreqCell]struct{}
}
m.mu.RLock()
snaps := make([]snap, 0, len(m.peerInterest))
for pid, set := range m.peerInterest {
if len(set) == 0 {
continue
}
p := m.peers[pid]
if p == nil {
continue
}
// shallow ref to set (read-only after unlock for membership checks)
snaps = append(snaps, snap{id: pid, p: p, set: set})
}
m.mu.RUnlock()
var out []string
for _, s := range snaps {
s.p.mu.Lock()
authed := s.p.authed
s.p.mu.Unlock()
if !authed {
continue
}
for _, k := range keys {
if _, ok := s.set[k]; ok {
out = append(out, s.id)
break
}
}
}
return out
}
// Start binds TCP+UDP, resolves allowlist, accept/dial, heartbeat.
// Single Start/Stop lifecycle — not safe to Start again after Stop.
func (m *HybridMesh) Start(ctx context.Context) error {
if m == nil {
return fmt.Errorf("afv hybrid mesh: nil")
}
if !m.started.CompareAndSwap(false, true) {
return nil // already started (or concurrent Start lost the race after first)
}
if err := m.resolveVoiceAllowlist(); err != nil {
m.started.Store(false)
return err
}
ln, err := net.Listen("tcp", m.cfg.ListenTCP)
if err != nil {
m.started.Store(false)
return fmt.Errorf("afv hybrid mesh tcp listen: %w", err)
}
m.ln = ln
m.logger.Info("AFV hybrid mesh TCP control listening", "addr", ln.Addr().String())
pc, err := net.ListenPacket("udp", m.cfg.ListenVoice)
if err != nil {
_ = ln.Close()
m.started.Store(false)
return fmt.Errorf("afv hybrid mesh udp voice listen: %w", err)
}
m.voicePC = pc
m.logger.Info("AFV hybrid mesh UDP voice listening", "addr", pc.LocalAddr().String())
// Pre-create peer shells (queues re-init on Hello)
for id, p := range m.peerCfg {
m.mu.Lock()
m.peers[id] = &hybridPeer{
id: id,
tcpAddr: p.Addr,
voiceAddr: p.VoiceAddr,
ctrlQ: newDropOldestQueue[meshCtrlJob](m.ctrlDepth),
voiceQ: newDropOldestQueue[voiceJob](m.voiceDepth),
stop: make(chan struct{}),
}
m.peerInterest[id] = make(map[FreqCell]struct{})
m.mu.Unlock()
}
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.acceptLoop(ctx)
}()
for id, p := range m.peerCfg {
if id <= m.cfg.NodeID {
continue // dial only if peer ID > self (H-9)
}
peer := p
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.dialLoop(ctx, peer)
}()
}
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.heartbeatLoop(ctx)
}()
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.voiceReadLoop(ctx)
}()
// Per-peer voice drainers
for id := range m.peerCfg {
pid := id
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.voiceDrainLoop(ctx, pid)
}()
}
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.muteWatchLoop(ctx)
}()
return nil
}
// Stop closes listeners/conns and waits for workers.
func (m *HybridMesh) Stop() error {
if m == nil {
return nil
}
m.stopOnce.Do(func() {
close(m.stopCh)
if m.ln != nil {
_ = m.ln.Close()
}
if m.voicePC != nil {
_ = m.voicePC.Close()
}
m.mu.Lock()
for _, p := range m.peers {
p.mu.Lock()
if p.conn != nil {
_ = p.conn.Close()
}
select {
case <-p.stop:
default:
close(p.stop)
}
p.authed = false
p.mu.Unlock()
}
m.mu.Unlock()
m.started.Store(false)
})
m.wg.Wait()
return nil
}
// PublishTrxSnapshot encodes and enqueues control to authed peers.
func (m *HybridMesh) PublishTrxSnapshot() {
if m == nil {
return
}
m.mu.RLock()
snapFn := m.snapFn
m.mu.RUnlock()
var sessions []MeshSessionBlock
if snapFn != nil {
sessions = snapFn()
}
payload := EncodeTrxSnapshot(TrxSnapshotPayload{
OriginNodeID: m.cfg.NodeID,
Sessions: sessions,
})
m.broadcastCtrl(MeshTypeTrxSnapshot, payload)
}
// PublishTrxDelta enqueues a directory delta.
func (m *HybridMesh) PublishTrxDelta(callsign string, isATC bool, trxs []Transceiver) {
if m == nil {
return
}
meshTrxs := make([]MeshTrx, 0, len(trxs))
for _, t := range trxs {
meshTrxs = append(meshTrxs, MeshTrx{
ID: t.ID, FreqHz: t.Frequency,
LatDeg: t.LatDeg, LonDeg: t.LonDeg, AltM: t.HeightMslM,
})
}
payload := EncodeTrxDelta(TrxDeltaPayload{
OriginNodeID: m.cfg.NodeID,
Callsign: callsign,
IsATC: isATC,
Trxs: meshTrxs,
})
m.broadcastCtrl(MeshTypeTrxDelta, payload)
}
// PublishSessionLeave enqueues leave.
func (m *HybridMesh) PublishSessionLeave(callsign string) {
if m == nil {
return
}
payload := EncodeSessionLeave(SessionLeavePayload{
OriginNodeID: m.cfg.NodeID,
Callsign: callsign,
})
m.broadcastCtrl(MeshTypeSessionLeave, payload)
}
// PublishInterest encodes + enqueues control only (does not mutate local peerInterest).
func (m *HybridMesh) PublishInterest(entries []InterestEntry) {
if m == nil {
return
}
payload := EncodeInterest(InterestPayload{NodeID: m.cfg.NodeID, Entries: entries})
m.broadcastCtrl(MeshTypeInterest, payload)
}
// EnqueueAudioRelay Interest-filters and enqueues UDP voice frames (MemoryMesh-compatible).
// Lock order: snapshot under m.mu, then p.mu alone (never nested).
func (m *HybridMesh) EnqueueAudioRelay(relay AudioRelay) {
if m == nil {
return
}
if relay.Audio != nil {
relay.Audio = append([]byte(nil), relay.Audio...)
}
if relay.TxRadios != nil {
relay.TxRadios = append([]RelayTxRadio(nil), relay.TxRadios...)
}
if relay.OriginNode == "" {
relay.OriginNode = m.cfg.NodeID
}
if len(relay.TxRadios) > maxMeshVoiceTxRadios {
return
}
// Snapshot peer pointers under m.mu only.
m.mu.RLock()
type snap struct {
id string
p *hybridPeer
}
snaps := make([]snap, 0, len(m.peers))
for id, p := range m.peers {
snaps = append(snaps, snap{id: id, p: p})
}
m.mu.RUnlock()
// Encode once; per-peer packet copy on enqueue.
var packet []byte
encoded := false
for _, s := range snaps {
s.p.mu.Lock()
authed := s.p.authed
q := s.p.voiceQ
s.p.mu.Unlock()
if !authed || q == nil {
continue
}
want := false
for _, tx := range relay.TxRadios {
ck := geo.CellKey{
ILat: geo.CellIndex(tx.LatDeg, geo.DefaultGridCellDeg),
ILon: geo.CellIndex(tx.LonDeg, geo.DefaultGridCellDeg),
}
if m.PeerWants(s.id, tx.FreqHz, ck) {
want = true
break
}
}
if !want {
continue
}
if !encoded {
payload, err := EncodeAudioRelay(relay)
if err != nil {
return
}
pkt, err := EncodeMeshFrameBytes(MeshTypeAudioRelay, payload)
if err != nil || len(pkt) > maxMeshVoiceDatagram {
return
}
packet = pkt
encoded = true
}
job := voiceJob{packet: append([]byte(nil), packet...)}
if q.Enqueue(job) {
m.voiceQueueDrops.Add(1)
}
}
}
func (m *HybridMesh) broadcastCtrl(typ byte, payload []byte) {
m.mu.RLock()
peers := make([]*hybridPeer, 0, len(m.peers))
for _, p := range m.peers {
peers = append(peers, p)
}
m.mu.RUnlock()
for _, p := range peers {
p.mu.Lock()
authed := p.authed
q := p.ctrlQ
p.mu.Unlock()
if !authed || q == nil {
continue
}
job := meshCtrlJob{typ: typ, payload: append([]byte(nil), payload...)}
if q.Enqueue(job) {
m.ctrlQueueDrops.Add(1)
m.logger.Debug("AFV hybrid mesh control drop", "peer", p.id, "type", typ)
}
}
}
// PeerAuthedForTest reports whether peer is control-authed (tests).
func (m *HybridMesh) PeerAuthedForTest(peerID string) bool {
if m == nil {
return false
}
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return false
}
p.mu.Lock()
defer p.mu.Unlock()
return p.authed
}
// WaitPeerAuthed waits until peer is authed or timeout.
func (m *HybridMesh) WaitPeerAuthed(peerID string, d time.Duration) bool {
deadline := time.Now().Add(d)
for time.Now().Before(deadline) {
if m.PeerAuthedForTest(peerID) {
return true
}
time.Sleep(10 * time.Millisecond)
}
return false
}
// ForcePeerDeathForTest closes the peer TCP conn to trigger death path (tests).
func (m *HybridMesh) ForcePeerDeathForTest(peerID string) {
if m == nil {
return
}
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return
}
p.mu.Lock()
c := p.conn
p.mu.Unlock()
if c != nil {
_ = c.Close()
}
}
// UDPAllowDrops / InboundRateDrops / VoiceQueueDrops expose counters for tests.
func (m *HybridMesh) UDPAllowDrops() uint64 { return m.udpAllowDrops.Load() }
func (m *HybridMesh) UDPOversizeDrops() uint64 { return m.udpOversizeDrops.Load() }
func (m *HybridMesh) InboundRateDrops() uint64 { return m.inboundRateDrops.Load() }
func (m *HybridMesh) VoiceQueueDrops() uint64 { return m.voiceQueueDrops.Load() }
func (m *HybridMesh) CtrlQueueDrops() uint64 { return m.ctrlQueueDrops.Load() }
func (m *HybridMesh) VoiceTx() uint64 { return m.voiceTx.Load() }
func (m *HybridMesh) VoiceRx() uint64 { return m.voiceRx.Load() }
// CurrentConnForTest returns the current TCP conn for peer (tests / dual-conn).
func (m *HybridMesh) CurrentConnForTest(peerID string) net.Conn {
if m == nil {
return nil
}
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return nil
}
p.mu.Lock()
defer p.mu.Unlock()
return p.conn
}
// ApplyInterestDirect is test helper for Interest maps without TCP.
func (m *HybridMesh) ApplyInterestDirect(fromPeer string, entries []InterestEntry) {
set := make(map[FreqCell]struct{}, len(entries))
for _, e := range entries {
set[FreqCell{FreqHz: e.FreqHz, Cell: geo.CellKey{ILat: e.ILat, ILon: e.ILon}}] = struct{}{}
}
m.mu.Lock()
m.peerInterest[fromPeer] = set
m.mu.Unlock()
}
// ForceEnqueueCtrlForTest fills control queue (drop-oldest tests).
func (m *HybridMesh) ForceEnqueueCtrlForTest(peerID string, n int) {
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return
}
p.mu.Lock()
q := p.ctrlQ
p.mu.Unlock()
if q == nil {
return
}
for i := 0; i < n; i++ {
job := meshCtrlJob{typ: MeshTypeHeartbeat, payload: EncodeHeartbeatPayload(uint64(i))}
if q.Enqueue(job) {
m.ctrlQueueDrops.Add(1)
}
}
}
// ForceEnqueueVoiceForTest fills voice queue (drop-oldest tests).
func (m *HybridMesh) ForceEnqueueVoiceForTest(peerID string, n int) {
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return
}
p.mu.Lock()
q := p.voiceQ
p.mu.Unlock()
if q == nil {
return
}
pkt := make([]byte, 32)
for i := 0; i < n; i++ {
job := voiceJob{packet: append([]byte(nil), pkt...)}
if q.Enqueue(job) {
m.voiceQueueDrops.Add(1)
}
}
}
// PeerCtrlQueueLen / PeerVoiceQueueLen for tests.
func (m *HybridMesh) PeerCtrlQueueLen(peerID string) int {
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return 0
}
p.mu.Lock()
q := p.ctrlQ
p.mu.Unlock()
if q == nil {
return 0
}
return q.Len()
}
func (m *HybridMesh) PeerVoiceQueueLen(peerID string) int {
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return 0
}
p.mu.Lock()
q := p.voiceQ
p.mu.Unlock()
if q == nil {
return 0
}
return q.Len()
}
// ListenTCPAddr / ListenVoiceAddr return bound addresses after Start (tests, :0).
func (m *HybridMesh) ListenTCPAddr() string {
if m == nil || m.ln == nil {
return m.cfg.ListenTCP
}
return m.ln.Addr().String()
}
func (m *HybridMesh) ListenVoiceAddr() string {
if m == nil || m.voicePC == nil {
return m.cfg.ListenVoice
}
return m.voicePC.LocalAddr().String()
}
var _ Mesh = (*HybridMesh)(nil)

View File

@@ -0,0 +1,249 @@
package afv_test
import (
"context"
"database/sql"
"net"
"strconv"
"testing"
"time"
"github.com/renorris/openfsd/internal/afv"
"github.com/renorris/openfsd/internal/db"
"github.com/renorris/openfsd/internal/geo"
"github.com/renorris/openfsd/pkg/afvprotocol"
"github.com/renorris/openfsd/pkg/protocol"
)
func freePortTCP(t *testing.T) int {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
p := ln.Addr().(*net.TCPAddr).Port
_ = ln.Close()
return p
}
func freePortUDP(t *testing.T) int {
t.Helper()
pc, err := net.ListenPacket("udp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
p := pc.LocalAddr().(*net.UDPAddr).Port
_ = pc.Close()
return p
}
type hybridNode struct {
srv *afv.Server
mesh *afv.HybridMesh
api string
udp string
cancel context.CancelFunc
}
func setupHybridPair(t *testing.T) (n1, n2 *hybridNode, cidA, cidB int) {
t.Helper()
tcp1, voice1 := freePortTCP(t), freePortUDP(t)
tcp2, voice2 := freePortTCP(t), freePortUDP(t)
open := func(name, nodeID, peerID string, tcp, voice, peerTCP, peerVoice int) (*hybridNode, *db.Repositories) {
sqlDB, err := sql.Open("sqlite", "file:hybride2e_"+t.Name()+"_"+name+"?mode=memory&cache=shared")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sqlDB.Close() })
if err := db.Migrate(sqlDB); err != nil {
t.Fatal(err)
}
repos, err := db.NewRepositories(sqlDB)
if err != nil {
t.Fatal(err)
}
hm, err := afv.NewHybridMesh(afv.HybridMeshConfig{
NodeID: nodeID,
ListenTCP: "127.0.0.1:" + strconv.Itoa(tcp),
ListenVoice: "127.0.0.1:" + strconv.Itoa(voice),
Peers: []afv.ClusterPeer{{
ID: peerID, Addr: "127.0.0.1:" + strconv.Itoa(peerTCP),
VoiceAddr: "127.0.0.1:" + strconv.Itoa(peerVoice),
}},
PSK: "hybrid-e2e-psk",
})
if err != nil {
t.Fatal(err)
}
cfg := &afv.Config{
APIListen: "127.0.0.1:0", UDPListen: "127.0.0.1:0",
UDPAdvertiseIPv4: "127.0.0.1:0", JWTTTL: time.Hour,
MaxSessions: 100, MaxSessionsPerCID: 5,
AuthFailMax: 50, AuthFailWindow: time.Minute, MaxDatagram: 8192,
RangeDefaultNM: 40, RangeATCNM: 150, RangeUnicomNM: 15, RangeEdgeRatio: 0.1,
HeartbeatTimeout: 30 * time.Second, SessionIdleTimeout: 60 * time.Second,
}
s := afv.New(cfg, repos.UserRepo, repos.ConfigRepo, []byte("hybrid-e2e-jwt-secret-key-32b!!"))
s.SetMesh(hm)
api, udp, cancel := startAFV(t, s)
cfg.UDPAdvertiseIPv4 = udp
return &hybridNode{srv: s, mesh: hm, api: api, udp: udp, cancel: cancel}, repos
}
var r1, r2 *db.Repositories
n1, r1 = open("n1", "n1", "n2", tcp1, voice1, tcp2, voice2)
n2, r2 = open("n2", "n2", "n1", tcp2, voice2, tcp1, voice1)
t.Cleanup(func() { n1.cancel(); n2.cancel() })
// Wait mesh auth both ways
if !n1.mesh.WaitPeerAuthed("n2", 5*time.Second) {
t.Fatal("n1↔n2 auth timeout")
}
if !n2.mesh.WaitPeerAuthed("n1", 5*time.Second) {
t.Fatal("n2↔n1 auth timeout")
}
uA := &db.User{Password: "pw", NetworkRating: int(protocol.NetworkRatingObserver), PilotRating: 1}
uB := &db.User{Password: "pw", NetworkRating: int(protocol.NetworkRatingObserver), PilotRating: 1}
if err := r1.UserRepo.CreateUser(context.Background(), uA); err != nil {
t.Fatal(err)
}
if err := r2.UserRepo.CreateUser(context.Background(), uB); err != nil {
t.Fatal(err)
}
return n1, n2, uA.CID, uB.CID
}
func waitHybridPeerWants(t *testing.T, m *afv.HybridMesh, peer string, freq uint32, txLat, txLon float64) {
t.Helper()
ck := geo.CellKey{
ILat: geo.CellIndex(txLat, geo.DefaultGridCellDeg),
ILon: geo.CellIndex(txLon, geo.DefaultGridCellDeg),
}
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
if m.PeerWants(peer, freq, ck) {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("timeout PeerWants peer=%s", peer)
}
func TestHybridMeshE2E_A2A(t *testing.T) {
n1, n2, cidA, cidB := setupHybridPair(t)
tokA := meshAuth(t, n1.api, cidA)
tokB := meshAuth(t, n2.api, cidB)
pcA := meshPostCS(t, n1.api, tokA, cidA, "N1A", n1.udp)
pcB := meshPostCS(t, n2.api, tokB, cidB, "N2B", n2.udp)
// Same frequency, nearby positions (in range)
freq := uint32(122800000)
latA, lonA := 51.4700, -0.4543
latB, lonB := 51.4710, -0.4530
meshPostTrx(t, n1.api, tokA, cidA, "N1A", latA, lonA, freq)
meshPostTrx(t, n2.api, tokB, cidB, "N2B", latB, lonB, freq)
// Bind UDP + HA first — Interest only includes Bound sessions
srvA, err := net.ResolveUDPAddr("udp", n1.udp)
if err != nil {
t.Fatal(err)
}
srvB, err := net.ResolveUDPAddr("udp", n2.udp)
if err != nil {
t.Fatal(err)
}
cliA, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
if err != nil {
t.Fatal(err)
}
defer cliA.Close()
cliB, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
if err != nil {
t.Fatal(err)
}
defer cliB.Close()
chA, err := afvprotocol.ClientChannel(pcA.VoiceServer.ChannelConfig.ChannelTag, pcA.VoiceServer.ChannelConfig.AeadReceiveKey, pcA.VoiceServer.ChannelConfig.AeadTransmitKey)
if err != nil {
t.Fatal(err)
}
chB, err := afvprotocol.ClientChannel(pcB.VoiceServer.ChannelConfig.ChannelTag, pcB.VoiceServer.ChannelConfig.AeadReceiveKey, pcB.VoiceServer.ChannelConfig.AeadTransmitKey)
if err != nil {
t.Fatal(err)
}
meshBindHB(t, chA, cliA, srvA, "N1A", 0)
meshBindHB(t, chB, cliB, srvB, "N2B", 0)
// Interest after bind; barrier on PeerWants (async TCP apply)
n1.srv.PublishInterestNowForTest()
n2.srv.PublishInterestNowForTest()
waitHybridPeerWants(t, n1.mesh, "n2", freq, latA, lonA)
// A transmits AT
at := afvprotocol.AudioTx{
Callsign: "N1A",
SequenceCounter: 1,
Audio: []byte("hybrid-opus"),
LastPacket: false,
Transceivers: []afvprotocol.TxTransceiver{{ID: 0}},
}
pkt, err := chA.Encapsulate(2, afvprotocol.DTONameAudioTx, at.EncodeMsgpack(), nil)
if err != nil {
t.Fatal(err)
}
if _, err := cliA.WriteTo(pkt, srvA); err != nil {
t.Fatal(err)
}
ar, ok := readAR(t, chB, cliB, 3*time.Second)
if !ok {
t.Fatal("B did not receive AR via hybrid mesh")
}
if ar.Callsign != "N1A" {
t.Fatalf("callsign %q", ar.Callsign)
}
}
func TestHybridMeshE2E_WrongPSK(t *testing.T) {
tcp1, voice1 := freePortTCP(t), freePortUDP(t)
tcp2, voice2 := freePortTCP(t), freePortUDP(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
m1, err := afv.NewHybridMesh(afv.HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + strconv.Itoa(tcp1), ListenVoice: "127.0.0.1:" + strconv.Itoa(voice1),
Peers: []afv.ClusterPeer{{ID: "n2", Addr: "127.0.0.1:" + strconv.Itoa(tcp2), VoiceAddr: "127.0.0.1:" + strconv.Itoa(voice2)}},
PSK: "psk-a",
})
if err != nil {
t.Fatal(err)
}
m2, err := afv.NewHybridMesh(afv.HybridMeshConfig{
NodeID: "n2", ListenTCP: "127.0.0.1:" + strconv.Itoa(tcp2), ListenVoice: "127.0.0.1:" + strconv.Itoa(voice2),
Peers: []afv.ClusterPeer{{ID: "n1", Addr: "127.0.0.1:" + strconv.Itoa(tcp1), VoiceAddr: "127.0.0.1:" + strconv.Itoa(voice1)}},
PSK: "psk-b",
})
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
if err := m2.Start(ctx); err != nil {
t.Fatal(err)
}
defer m1.Stop()
defer m2.Stop()
// Poll window: must remain unauthenticated under wrong PSK.
deadline := time.Now().Add(1500 * time.Millisecond)
for time.Now().Before(deadline) {
if m1.PeerAuthedForTest("n2") || m2.PeerAuthedForTest("n1") {
t.Fatal("wrong PSK must not auth")
}
time.Sleep(20 * time.Millisecond)
}
}

493
internal/afv/mesh_tcp.go Normal file
View File

@@ -0,0 +1,493 @@
package afv
// HybridMesh TCP control plane: listen/dial, Hello, HB, directory, Interest, peer death.
// AudioRelay (type 20) is forbidden on TCP (H-2).
import (
"context"
"errors"
"fmt"
"net"
"strings"
"time"
"github.com/renorris/openfsd/internal/geo"
)
func (m *HybridMesh) acceptLoop(ctx context.Context) {
for {
conn, err := m.ln.Accept()
if err != nil {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
default:
}
// Permanent close of listener → exit; transient errors → backoff + continue.
if errors.Is(err, net.ErrClosed) {
return
}
m.logger.Debug("AFV hybrid mesh accept error", "err", err)
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(100 * time.Millisecond):
}
continue
}
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.serveConn(conn)
}()
}
}
func (m *HybridMesh) dialLoop(ctx context.Context, peer ClusterPeer) {
backoff := 200 * time.Millisecond
for {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
default:
}
// Skip dial if already authed with live conn.
m.mu.RLock()
p := m.peers[peer.ID]
m.mu.RUnlock()
if p != nil {
p.mu.Lock()
live := p.authed && p.conn != nil
p.mu.Unlock()
if live {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(500 * time.Millisecond):
}
continue
}
}
d := net.Dialer{Timeout: 5 * time.Second}
conn, err := d.DialContext(ctx, "tcp", peer.Addr)
if err != nil {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff < 5*time.Second {
backoff *= 2
}
continue
}
backoff = 200 * time.Millisecond
m.serveConn(conn)
// reconnect after disconnect
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(200 * time.Millisecond):
}
}
}
func (m *HybridMesh) serveConn(conn net.Conn) {
defer func() { _ = conn.Close() }()
// Send Hello immediately
helloPay := EncodeHelloPayload(HelloPayload{NodeID: m.cfg.NodeID, PSK: m.cfg.PSK})
if err := EncodeMeshFrame(conn, MeshTypeHello, helloPay); err != nil {
return
}
_ = conn.SetReadDeadline(time.Now().Add(5 * time.Second))
fr, err := DecodeMeshFrame(conn)
if err != nil || fr.Type != MeshTypeHello {
m.logger.Warn("AFV hybrid mesh non-Hello first frame or read error", "err", err)
return
}
hp, err := DecodeHelloPayload(fr.Payload)
if err != nil {
return
}
if err := VerifyHelloPSK(m.cfg.PSK, hp.PSK); err != nil {
m.logger.Warn("AFV hybrid mesh hello auth failed", "peer", hp.NodeID)
return
}
peerID := strings.TrimSpace(hp.NodeID)
if peerID == "" || peerID == m.cfg.NodeID {
m.logger.Warn("AFV hybrid mesh reject self or empty peer")
return
}
if _, ok := m.peerCfg[peerID]; !ok {
m.logger.Warn("AFV hybrid mesh reject unknown peer", "peer", peerID)
return
}
// H-15: log TCP RemoteAddr host ≠ configured peer TCP host as warn only
if expected, ok := m.peerCfg[peerID]; ok && expected.Addr != "" {
if ra := conn.RemoteAddr(); ra != nil {
host, _, _ := net.SplitHostPort(ra.String())
expHost, _, _ := net.SplitHostPort(expected.Addr)
if host != "" && expHost != "" && host != expHost && expHost != "0.0.0.0" {
m.logger.Warn("AFV hybrid mesh Hello RemoteAddr host differs from config",
"peer", peerID, "remote", host, "configured", expHost)
}
}
}
_ = conn.SetReadDeadline(time.Time{})
// Install peer (H-9 dual-conn replace); capture this generation's stop+queues.
// Soft replace does not run peerDeath / OnPeerDead.
p, stop, ctrlQ := m.installPeerConn(peerID, conn)
if p == nil {
return
}
// Per-peer TCP writer for this generation only (bound to ctrlQ + stop)
m.wg.Add(1)
go func() {
defer m.wg.Done()
m.peerCtrlWriter(p, conn, ctrlQ, stop)
}()
// Post-Hello sequence: Snapshot then Interest (M-16) on TCP only
m.sendPostHello(ctrlQ)
// Read loop
for {
select {
case <-m.stopCh:
return
case <-stop:
// Superseded by dual-conn replace or death of this generation — exit quietly.
return
default:
}
_ = conn.SetReadDeadline(time.Now().Add(m.peerDeathGrace + 5*time.Second))
fr, err := DecodeMeshFrame(conn)
if err != nil {
// Generation-scoped: only tear down if this conn is still current.
m.peerDeath(peerID, "read", conn)
return
}
// Generation gate (same as lastHB): superseded dual-conn readers must not
// apply Interest/directory or peerDeath after a successful late decode.
p.mu.Lock()
stillMine := p.conn == conn
if stillMine {
p.lastHB = time.Now() // H-8: every post-Hello control frame
}
p.mu.Unlock()
if !stillMine {
return
}
if !m.handleCtrlFrame(peerID, fr, conn) {
return // closed (type 20, bad, unexpected Hello)
}
}
}
// installPeerConn replaces any existing conn for peerID (H-9).
// Soft replace: closes previous generation stop+conn without OnPeerDead / Interest wipe.
// Returns the peer, this generation's stop channel, and ctrl queue.
func (m *HybridMesh) installPeerConn(peerID string, conn net.Conn) (*hybridPeer, chan struct{}, *dropOldestQueue[meshCtrlJob]) {
m.mu.RLock()
p, ok := m.peers[peerID]
m.mu.RUnlock()
if !ok {
return nil, nil, nil
}
p.mu.Lock()
// Close previous live generation (soft replace — no peerDeath)
if p.conn != nil && p.conn != conn {
select {
case <-p.stop:
default:
close(p.stop)
}
_ = p.conn.Close()
}
// Reset queues + stop channel for new generation
ctrlQ := newDropOldestQueue[meshCtrlJob](m.ctrlDepth)
voiceQ := newDropOldestQueue[voiceJob](m.voiceDepth)
stop := make(chan struct{})
p.ctrlQ = ctrlQ
p.voiceQ = voiceQ
p.stop = stop
p.conn = conn
p.authed = true
p.lastHB = time.Now()
p.mu.Unlock()
m.logger.Info("AFV hybrid mesh peer authenticated", "peer", peerID)
return p, stop, ctrlQ
}
func (m *HybridMesh) sendPostHello(ctrlQ *dropOldestQueue[meshCtrlJob]) {
if ctrlQ == nil {
return
}
// Snapshot
m.mu.RLock()
snapFn := m.snapFn
m.mu.RUnlock()
var sessions []MeshSessionBlock
if snapFn != nil {
sessions = snapFn()
}
snap := EncodeTrxSnapshot(TrxSnapshotPayload{
OriginNodeID: m.cfg.NodeID,
Sessions: sessions,
})
_ = ctrlQ.Enqueue(meshCtrlJob{typ: MeshTypeTrxSnapshot, payload: snap})
// Interest
m.mu.RLock()
ifn := m.interestFn
m.mu.RUnlock()
var entries []InterestEntry
if ifn != nil {
entries = ifn()
}
ipay := EncodeInterest(InterestPayload{NodeID: m.cfg.NodeID, Entries: entries})
_ = ctrlQ.Enqueue(meshCtrlJob{typ: MeshTypeInterest, payload: ipay})
}
func (m *HybridMesh) peerCtrlWriter(p *hybridPeer, conn net.Conn, q *dropOldestQueue[meshCtrlJob], stop chan struct{}) {
if q == nil {
return
}
for {
select {
case <-m.stopCh:
return
case <-stop:
return
case job, ok := <-q.Chan():
if !ok {
return
}
if err := EncodeMeshFrame(conn, job.typ, job.payload); err != nil {
m.peerDeath(p.id, "write", conn)
return
}
}
}
}
// handleCtrlFrame returns false if the peer should be closed.
// deadConn is the TCP conn of this reader generation (for generation-scoped death/apply).
// Superseded readers (p.conn != deadConn) are no-ops for apply and death.
func (m *HybridMesh) handleCtrlFrame(peerID string, fr MeshFrame, deadConn net.Conn) bool {
// Defense in depth: re-check generation before any side effects.
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return false
}
p.mu.Lock()
stillMine := p.conn == deadConn
p.mu.Unlock()
if !stillMine {
return true // no apply / no death; caller may already be exiting
}
switch fr.Type {
case MeshTypeHello:
m.logger.Warn("AFV hybrid mesh unexpected Hello after auth", "peer", peerID)
m.peerDeath(peerID, "unexpected Hello", deadConn)
return false
case MeshTypeHeartbeat:
return true
case MeshTypeTrxSnapshot:
pl, err := DecodeTrxSnapshot(fr.Payload)
if err != nil {
m.peerDeath(peerID, "bad TrxSnapshot", deadConn)
return false
}
if pl.OriginNodeID != peerID {
m.peerDeath(peerID, "snapshot origin mismatch", deadConn)
return false
}
m.mu.RLock()
dir := m.onDir
m.mu.RUnlock()
if dir != nil {
dir.ApplySnapshot(pl.OriginNodeID, meshSessionsToRemote(pl.Sessions))
}
return true
case MeshTypeTrxDelta:
pl, err := DecodeTrxDelta(fr.Payload)
if err != nil {
m.peerDeath(peerID, "bad TrxDelta", deadConn)
return false
}
if pl.OriginNodeID != peerID {
m.peerDeath(peerID, "delta origin mismatch", deadConn)
return false
}
m.mu.RLock()
dir := m.onDir
m.mu.RUnlock()
if dir != nil {
dir.ApplyDelta(pl.OriginNodeID, pl.Callsign, pl.IsATC, meshTrxToLocal(pl.Trxs))
}
return true
case MeshTypeSessionLeave:
pl, err := DecodeSessionLeave(fr.Payload)
if err != nil {
m.peerDeath(peerID, "bad SessionLeave", deadConn)
return false
}
if pl.OriginNodeID != peerID {
m.peerDeath(peerID, "leave origin mismatch", deadConn)
return false
}
m.mu.RLock()
dir := m.onDir
m.mu.RUnlock()
if dir != nil {
dir.ApplyLeave(pl.OriginNodeID, pl.Callsign)
}
return true
case MeshTypeAudioRelay:
// H-2: type 20 forbidden on TCP
m.logger.Warn("AFV hybrid mesh type 20 AudioRelay on TCP — closing peer", "peer", peerID)
m.peerDeath(peerID, "audio on TCP", deadConn)
return false
case MeshTypeInterest:
// H-14: full replace for origin
pl, err := DecodeInterest(fr.Payload)
if err != nil {
m.peerDeath(peerID, "bad Interest", deadConn)
return false
}
if pl.NodeID != peerID {
m.peerDeath(peerID, "interest node mismatch", deadConn)
return false
}
set := make(map[FreqCell]struct{}, len(pl.Entries))
for _, e := range pl.Entries {
set[FreqCell{FreqHz: e.FreqHz, Cell: geo.CellKey{ILat: e.ILat, ILon: e.ILon}}] = struct{}{}
}
m.mu.Lock()
m.peerInterest[peerID] = set
m.mu.Unlock()
return true
default:
m.logger.Warn("AFV hybrid mesh unknown control type", "peer", peerID, "type", fr.Type)
m.peerDeath(peerID, fmt.Sprintf("unknown type %d", fr.Type), deadConn)
return false
}
}
// peerDeath is generation-scoped death procedure (H-8 / H-9).
// Only tears down if deadConn is still the peer's current conn (or deadConn is nil
// and the peer is authed — used for heartbeat timeout with captured conn).
// Lock order: p.mu alone, then m.mu alone — never nested.
func (m *HybridMesh) peerDeath(peerID, reason string, deadConn net.Conn) {
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
return
}
p.mu.Lock()
// Generation guard: stale reader/writer from dual-conn replace must not kill successor.
if deadConn != nil && p.conn != deadConn {
p.mu.Unlock()
return
}
if deadConn == nil && (!p.authed || p.conn == nil) {
p.mu.Unlock()
return
}
wasAuthed := p.authed
p.authed = false
if p.conn != nil {
_ = p.conn.Close()
p.conn = nil
}
select {
case <-p.stop:
default:
close(p.stop)
}
// re-init empty queues
p.ctrlQ = newDropOldestQueue[meshCtrlJob](m.ctrlDepth)
p.voiceQ = newDropOldestQueue[voiceJob](m.voiceDepth)
p.mu.Unlock()
// clear Interest + load callback under m.mu only (no p.mu)
m.mu.Lock()
m.peerInterest[peerID] = make(map[FreqCell]struct{})
cb := m.onPeerDead
m.mu.Unlock()
if wasAuthed {
m.logger.Warn("AFV hybrid mesh peer dead", "peer", peerID, "reason", reason)
if cb != nil {
cb(peerID)
}
}
}
func (m *HybridMesh) heartbeatLoop(ctx context.Context) {
ticker := time.NewTicker(m.hbInterval)
defer ticker.Stop()
for {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-ticker.C:
now := time.Now()
ms := uint64(now.UnixMilli())
pay := EncodeHeartbeatPayload(ms)
// Snapshot peer pointers under m.mu only — never nest p.mu.
m.mu.RLock()
peers := make([]*hybridPeer, 0, len(m.peers))
for _, p := range m.peers {
peers = append(peers, p)
}
m.mu.RUnlock()
for _, p := range peers {
p.mu.Lock()
authed := p.authed
last := p.lastHB
conn := p.conn
q := p.ctrlQ
p.mu.Unlock()
if !authed {
continue
}
if !last.IsZero() && now.Sub(last) > m.peerDeathGrace {
m.peerDeath(p.id, "heartbeat timeout", conn)
continue
}
if q != nil {
_ = q.Enqueue(meshCtrlJob{typ: MeshTypeHeartbeat, payload: pay})
}
}
}
}
}

View File

@@ -0,0 +1,469 @@
package afv
import (
"context"
"io"
"net"
"strconv"
"sync/atomic"
"testing"
"time"
"github.com/renorris/openfsd/internal/geo"
)
func freeTCPPort(t *testing.T) int {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer ln.Close()
return ln.Addr().(*net.TCPAddr).Port
}
func freeUDPPort(t *testing.T) int {
t.Helper()
pc, err := net.ListenPacket("udp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer pc.Close()
return pc.LocalAddr().(*net.UDPAddr).Port
}
func hybridPairPorts(t *testing.T) (tcp1, voice1, tcp2, voice2 int) {
t.Helper()
tcp1 = freeTCPPort(t)
voice1 = freeUDPPort(t)
tcp2 = freeTCPPort(t)
voice2 = freeUDPPort(t)
return
}
func startHybridPair(t *testing.T, psk string) (m1, m2 *HybridMesh, cancel context.CancelFunc) {
t.Helper()
tcp1, voice1, tcp2, voice2 := hybridPairPorts(t)
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
cfg1 := HybridMeshConfig{
NodeID: "n1",
ListenTCP: "127.0.0.1:" + itoa(tcp1),
ListenVoice: "127.0.0.1:" + itoa(voice1),
Peers: []ClusterPeer{{
ID: "n2", Addr: "127.0.0.1:" + itoa(tcp2), VoiceAddr: "127.0.0.1:" + itoa(voice2),
}},
PSK: psk,
}
cfg2 := HybridMeshConfig{
NodeID: "n2",
ListenTCP: "127.0.0.1:" + itoa(tcp2),
ListenVoice: "127.0.0.1:" + itoa(voice2),
Peers: []ClusterPeer{{
ID: "n1", Addr: "127.0.0.1:" + itoa(tcp1), VoiceAddr: "127.0.0.1:" + itoa(voice1),
}},
PSK: psk,
}
var err error
m1, err = NewHybridMesh(cfg1)
if err != nil {
t.Fatal(err)
}
m2, err = NewHybridMesh(cfg2)
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
if err := m2.Start(ctx); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = m1.Stop(); _ = m2.Stop() })
if !m1.WaitPeerAuthed("n2", 3*time.Second) {
t.Fatal("n1 did not auth n2")
}
if !m2.WaitPeerAuthed("n1", 3*time.Second) {
t.Fatal("n2 did not auth n1")
}
return m1, m2, cancel
}
func itoa(n int) string {
return strconv.Itoa(n)
}
func geoCell(ilat, ilon int32) geo.CellKey {
return geo.CellKey{ILat: ilat, ILon: ilon}
}
func TestHybridMesh_HelloWrongPSK(t *testing.T) {
tcp1, voice1, tcp2, voice2 := hybridPairPorts(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
m1, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + itoa(tcp1), ListenVoice: "127.0.0.1:" + itoa(voice1),
Peers: []ClusterPeer{{ID: "n2", Addr: "127.0.0.1:" + itoa(tcp2), VoiceAddr: "127.0.0.1:" + itoa(voice2)}},
PSK: "secret-a",
})
if err != nil {
t.Fatal(err)
}
m2, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n2", ListenTCP: "127.0.0.1:" + itoa(tcp2), ListenVoice: "127.0.0.1:" + itoa(voice2),
Peers: []ClusterPeer{{ID: "n1", Addr: "127.0.0.1:" + itoa(tcp1), VoiceAddr: "127.0.0.1:" + itoa(voice1)}},
PSK: "secret-b",
})
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
if err := m2.Start(ctx); err != nil {
t.Fatal(err)
}
defer m1.Stop()
defer m2.Stop()
// Poll a window: must never become authed under wrong PSK.
deadline := time.Now().Add(1500 * time.Millisecond)
for time.Now().Before(deadline) {
if m1.PeerAuthedForTest("n2") || m2.PeerAuthedForTest("n1") {
t.Fatal("wrong PSK must not authenticate")
}
time.Sleep(20 * time.Millisecond)
}
}
func TestHybridMesh_NonHelloFirstFrame(t *testing.T) {
tcpPort := freeTCPPort(t)
voicePort := freeUDPPort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Peer that will dial us (higher ID) — but we inject a raw non-Hello first frame
// by connecting as an attacker to the listen port.
m, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + itoa(tcpPort), ListenVoice: "127.0.0.1:" + itoa(voicePort),
Peers: []ClusterPeer{{ID: "n2", Addr: "127.0.0.1:1", VoiceAddr: "127.0.0.1:2"}},
PSK: "psk",
})
if err != nil {
t.Fatal(err)
}
if err := m.Start(ctx); err != nil {
t.Fatal(err)
}
defer m.Stop()
conn, err := net.Dial("tcp", "127.0.0.1:"+itoa(tcpPort))
if err != nil {
t.Fatal(err)
}
defer conn.Close()
// Send Heartbeat as first frame
if err := EncodeMeshFrame(conn, MeshTypeHeartbeat, EncodeHeartbeatPayload(1)); err != nil {
t.Fatal(err)
}
// Server should close after rejecting non-Hello
_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
buf := make([]byte, 8)
_, err = conn.Read(buf)
if err == nil {
// may get Hello from server first, then close
_, err = io.ReadAll(conn)
}
// peer must not be authed
if m.PeerAuthedForTest("n2") {
t.Fatal("non-Hello first frame must not auth")
}
}
func TestHybridMesh_Type20OnTCPCloses(t *testing.T) {
m1, m2, _ := startHybridPair(t, "psk")
_ = m2
// Inject type 20 on the TCP control conn from m1's peer n2
m1.mu.RLock()
p := m1.peers["n2"]
m1.mu.RUnlock()
if p == nil {
t.Fatal("no peer")
}
p.mu.Lock()
conn := p.conn
p.mu.Unlock()
if conn == nil {
t.Fatal("no conn")
}
// Write AudioRelay frame on TCP
pay, err := EncodeAudioRelay(AudioRelay{OriginNode: "n1", Callsign: "X", Audio: []byte{1}})
if err != nil {
t.Fatal(err)
}
if err := EncodeMeshFrame(conn, MeshTypeAudioRelay, pay); err != nil {
t.Fatal(err)
}
// Wait for peer death on m2 (received type 20)
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if !m2.PeerAuthedForTest("n1") {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatal("type 20 on TCP should close peer on receiver")
}
func TestHybridMesh_InterestReplace(t *testing.T) {
m1, m2, _ := startHybridPair(t, "psk")
// m2 publishes Interest → applied on m1's TCP reader
entries := []InterestEntry{{FreqHz: 122800000, ILat: 1, ILon: 2}}
m2.PublishInterest(entries)
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m1.PeerWants("n2", 122800000, geoCell(1, 2)) {
break
}
time.Sleep(10 * time.Millisecond)
}
if !m1.PeerWants("n2", 122800000, geoCell(1, 2)) {
t.Fatal("interest not applied")
}
// Empty Interest clears (replace, not merge)
m2.PublishInterest(nil)
deadline = time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if !m1.PeerWants("n2", 122800000, geoCell(1, 2)) {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatal("empty Interest should clear prior set")
}
func TestHybridMesh_LastHBAdvancesOnTrxDelta(t *testing.T) {
// Short death grace for test speed — use normal and just verify lastHB moves
m1, m2, _ := startHybridPair(t, "psk")
m1.mu.RLock()
p := m1.peers["n2"]
m1.mu.RUnlock()
p.mu.Lock()
before := p.lastHB
p.mu.Unlock()
time.Sleep(20 * time.Millisecond)
m2.PublishTrxDelta("N2PILOT", false, []Transceiver{{ID: 0, Frequency: 1}})
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
p.mu.Lock()
after := p.lastHB
p.mu.Unlock()
if after.After(before) {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatal("lastHB should advance on TrxDelta")
}
func TestHybridMesh_PeerDeathReadError(t *testing.T) {
var dead atomic.Bool
m1, m2, _ := startHybridPair(t, "psk")
m1.OnPeerDead(func(id string) {
if id == "n2" {
dead.Store(true)
}
})
// Arm Interest then kill
m2.PublishInterest([]InterestEntry{{FreqHz: 1, ILat: 0, ILon: 0}})
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m1.PeerWants("n2", 1, geo.CellKey{}) {
break
}
time.Sleep(10 * time.Millisecond)
}
m1.ForcePeerDeathForTest("n2")
deadline = time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if dead.Load() && !m1.PeerAuthedForTest("n2") && !m1.PeerWants("n2", 1, geo.CellKey{}) {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("peer death: dead=%v authed=%v wants=%v", dead.Load(), m1.PeerAuthedForTest("n2"), m1.PeerWants("n2", 1, geo.CellKey{}))
}
func TestHybridMesh_CtrlQueueDropOldest(t *testing.T) {
// No live peer writer: start only n1 so n2 never Hello — force drops.
tcp1, voice1, tcp2, voice2 := hybridPairPorts(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
m1, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + itoa(tcp1), ListenVoice: "127.0.0.1:" + itoa(voice1),
Peers: []ClusterPeer{{ID: "n2", Addr: "127.0.0.1:" + itoa(tcp2), VoiceAddr: "127.0.0.1:" + itoa(voice2)}},
PSK: "psk",
})
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
defer m1.Stop()
before := m1.CtrlQueueDrops()
m1.ForceEnqueueCtrlForTest("n2", meshControlQueueDepth+50)
if m1.CtrlQueueDrops() <= before {
t.Fatalf("want ctrl drops, got %d→%d", before, m1.CtrlQueueDrops())
}
if m1.PeerCtrlQueueLen("n2") > meshControlQueueDepth {
t.Fatal("queue over depth")
}
}
func TestHybridMesh_DualConnReplace(t *testing.T) {
// T10b: second live Hello replaces first without OnPeerDead (soft replace).
m1, m2, _ := startHybridPair(t, "psk")
var deadN1 atomic.Int32
m2.OnPeerDead(func(id string) {
if id == "n1" {
deadN1.Add(1)
}
})
// Seed Interest on m2's view of n1 (m1 publishes → m2 applies).
m1.PublishInterest([]InterestEntry{{FreqHz: 42, ILat: 1, ILon: 2}})
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m2.PeerWants("n1", 42, geoCell(1, 2)) {
break
}
time.Sleep(10 * time.Millisecond)
}
if !m2.PeerWants("n1", 42, geoCell(1, 2)) {
t.Fatal("interest not applied before replace")
}
oldConn := m2.CurrentConnForTest("n1")
if oldConn == nil {
t.Fatal("expected live conn")
}
// Second TCP session claiming n1 → soft replace on m2.
conn, err := net.Dial("tcp", m2.ListenTCPAddr())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
// m2 sends Hello first; read it, then send ours.
_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
fr, err := DecodeMeshFrame(conn)
if err != nil || fr.Type != MeshTypeHello {
t.Fatalf("expected Hello from m2: %v %+v", err, fr)
}
hello := EncodeHelloPayload(HelloPayload{NodeID: "n1", PSK: "psk"})
if err := EncodeMeshFrame(conn, MeshTypeHello, hello); err != nil {
t.Fatal(err)
}
// Wait until current conn is replaced and still authed.
deadline = time.Now().Add(3 * time.Second)
replaced := false
for time.Now().Before(deadline) {
cur := m2.CurrentConnForTest("n1")
if cur != nil && cur != oldConn && m2.PeerAuthedForTest("n1") {
replaced = true
break
}
time.Sleep(10 * time.Millisecond)
}
if !replaced {
t.Fatal("second Hello did not replace peer conn")
}
if deadN1.Load() != 0 {
t.Fatalf("OnPeerDead must not fire on soft replace, got %d", deadN1.Load())
}
// Soft replace retains Interest (no purge)
if !m2.PeerWants("n1", 42, geoCell(1, 2)) {
t.Fatal("Interest should survive soft replace")
}
// Stale reader apply gate: Interest clear framed as oldConn must not clobber state.
emptyPay := EncodeInterest(InterestPayload{NodeID: "n1", Entries: nil})
ok := m2.handleCtrlFrame("n1", MeshFrame{Type: MeshTypeInterest, Payload: emptyPay}, oldConn)
if !ok {
t.Fatal("stale handleCtrlFrame should return true (no death)")
}
if !m2.PeerWants("n1", 42, geoCell(1, 2)) {
t.Fatal("stale reader must not clear Interest after soft replace")
}
// Stale directory delta must also be ignored
deltaPay := EncodeTrxDelta(TrxDeltaPayload{
OriginNodeID: "n1", Callsign: "STALE", IsATC: false,
})
_ = m2.handleCtrlFrame("n1", MeshFrame{Type: MeshTypeTrxDelta, Payload: deltaPay}, oldConn)
// live generation apply still works
live := m2.CurrentConnForTest("n1")
if live == nil {
t.Fatal("expected live conn after replace")
}
newPay := EncodeInterest(InterestPayload{NodeID: "n1", Entries: []InterestEntry{{FreqHz: 99, ILat: 3, ILon: 4}}})
if !m2.handleCtrlFrame("n1", MeshFrame{Type: MeshTypeInterest, Payload: newPay}, live) {
t.Fatal("live handleCtrlFrame failed")
}
if !m2.PeerWants("n1", 99, geoCell(3, 4)) {
t.Fatal("live generation must still apply Interest")
}
// True death: close the live generation — OnPeerDead must fire, Interest cleared.
m2.ForcePeerDeathForTest("n1")
deadline = time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if deadN1.Load() > 0 && !m2.PeerAuthedForTest("n1") && !m2.PeerWants("n1", 99, geoCell(3, 4)) {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("true death: dead=%d authed=%v wants=%v",
deadN1.Load(), m2.PeerAuthedForTest("n1"), m2.PeerWants("n1", 99, geoCell(3, 4)))
}
func TestHybridMesh_EnqueueDuringDeathNoDeadlock(t *testing.T) {
// Stress: EnqueueAudioRelay while ForcePeerDeath (lock-order regression).
m1, m2, _ := startHybridPair(t, "psk")
m1.ApplyInterestDirect("n2", []InterestEntry{{
FreqHz: 122800000,
ILat: geo.CellIndex(51.5, geo.DefaultGridCellDeg),
ILon: geo.CellIndex(-0.1, geo.DefaultGridCellDeg),
}})
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; i < 2000; i++ {
m1.EnqueueAudioRelay(AudioRelay{
OriginNode: "n1", Callsign: "X", SequenceCounter: uint32(i),
Audio: []byte{1, 2, 3},
TxRadios: []RelayTxRadio{{FreqHz: 122800000, LatDeg: 51.5, LonDeg: -0.1}},
})
}
}()
for i := 0; i < 30; i++ {
m1.ForcePeerDeathForTest("n2")
_ = m1.WaitPeerAuthed("n2", 500*time.Millisecond)
}
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("enqueue+death deadlock or hang")
}
_ = m2
}

View File

@@ -0,0 +1,333 @@
package afv
// HybridMesh UDP voice plane: AudioRelay (type 20) only.
// Allowlist H-16, maxMeshVoiceDatagram H-12, inbound rate H-19.
import (
"context"
"fmt"
"net"
"strconv"
"time"
)
// maxMeshVoiceDatagram is the only accepted UDP voice size (bytes). H-12.
// Do NOT use MaxMeshPayload (1 MiB) for UDP reads.
// Var so tests can lower the cap without exceeding OS UDP send limits.
var maxMeshVoiceDatagram = 16 << 10 // 16384
// maxMeshVoiceTxRadios caps TxRadios on mesh AudioRelay for voice path.
const maxMeshVoiceTxRadios = 64
// meshVoiceInboundRatePerSec soft cap per authed peer (datagrams/s). H-19.
const meshVoiceInboundRatePerSec = 5000
// resolveVoiceAllowlist resolves peer VoiceAddrs at Start (H-15 / H-16).
func (m *HybridMesh) resolveVoiceAllowlist() error {
allow := make(map[string]string)
send := make(map[string]*net.UDPAddr)
for id, p := range m.peerCfg {
host, portStr, err := net.SplitHostPort(p.VoiceAddr)
if err != nil {
return fmt.Errorf("afv hybrid mesh peer %q VoiceAddr: %w", id, err)
}
port, err := strconv.Atoi(portStr)
if err != nil || port <= 0 || port > 65535 {
return fmt.Errorf("afv hybrid mesh peer %q voice port invalid", id)
}
var ips []net.IP
if ip := net.ParseIP(host); ip != nil {
ips = []net.IP{ip}
} else {
resolved, err := net.LookupIP(host)
if err != nil || len(resolved) == 0 {
return fmt.Errorf("afv hybrid mesh resolve peer %q voice host %q: %w", id, host, err)
}
ips = resolved
}
// Prefer IPv4 for send target
var sendIP net.IP
for _, ip := range ips {
if v4 := ip.To4(); v4 != nil {
sendIP = v4
break
}
}
if sendIP == nil {
sendIP = ips[0]
}
send[id] = &net.UDPAddr{IP: sendIP, Port: port}
for _, ip := range ips {
for _, key := range canonicalUDPKeys(ip, port) {
if other, ok := allow[key]; ok && other != id {
return fmt.Errorf("afv hybrid mesh allowlist collision key %q peers %q and %q", key, other, id)
}
allow[key] = id
}
}
}
m.allowlist = allow
m.sendVoice = send
return nil
}
// canonicalUDPKeys returns H-16 dual keys for IPv4 / IPv4-mapped IPv6.
func canonicalUDPKeys(ip net.IP, port int) []string {
var keys []string
ip16 := ip.To16()
if ip16 != nil {
keys = append(keys, net.JoinHostPort(ip16.String(), strconv.Itoa(port)))
}
if v4 := ip.To4(); v4 != nil {
keys = append(keys, net.JoinHostPort(v4.String(), strconv.Itoa(port)))
// IPv4-mapped form explicitly
mapped := net.IP{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, v4[0], v4[1], v4[2], v4[3]}
keys = append(keys, net.JoinHostPort(mapped.String(), strconv.Itoa(port)))
}
return keys
}
func (m *HybridMesh) lookupAllowlist(src net.Addr) string {
ua, ok := src.(*net.UDPAddr)
if !ok {
host, portStr, err := net.SplitHostPort(src.String())
if err != nil {
return ""
}
ip := net.ParseIP(host)
if ip == nil {
return ""
}
port, _ := strconv.Atoi(portStr)
ua = &net.UDPAddr{IP: ip, Port: port}
}
for _, key := range canonicalUDPKeys(ua.IP, ua.Port) {
if id, ok := m.allowlist[key]; ok {
return id
}
}
// also try raw JoinHostPort of To16
if ua.IP != nil {
k := net.JoinHostPort(ua.IP.String(), strconv.Itoa(ua.Port))
if id, ok := m.allowlist[k]; ok {
return id
}
}
return ""
}
func (m *HybridMesh) voiceReadLoop(ctx context.Context) {
buf := make([]byte, maxMeshVoiceDatagram+1)
for {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
default:
}
if m.voicePC == nil {
return
}
_ = m.voicePC.SetReadDeadline(time.Now().Add(1 * time.Second))
n, src, err := m.voicePC.ReadFrom(buf)
if err != nil {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
default:
// timeout or transient
continue
}
}
if n > maxMeshVoiceDatagram {
m.udpOversizeDrops.Add(1)
continue
}
peerID := m.lookupAllowlist(src)
if peerID == "" {
m.udpAllowDrops.Add(1)
continue
}
m.mu.RLock()
p := m.peers[peerID]
m.mu.RUnlock()
if p == nil {
m.udpAllowDrops.Add(1)
continue
}
p.mu.Lock()
authed := p.authed
p.mu.Unlock()
if !authed {
m.udpAllowDrops.Add(1)
continue
}
if !m.allowInboundRate(p) {
m.inboundRateDrops.Add(1)
continue
}
frame, err := DecodeMeshFrameExact(buf[:n])
if err != nil || frame.Type != MeshTypeAudioRelay {
continue
}
relay, err := DecodeAudioRelay(frame.Payload)
if err != nil || len(relay.TxRadios) > maxMeshVoiceTxRadios {
continue
}
if relay.OriginNode != peerID {
continue
}
m.voiceRx.Add(1)
p.voiceRxCount.Add(1)
p.lastVoiceUnixNano.Store(time.Now().UnixNano())
m.mu.RLock()
fn := m.onAudio
m.mu.RUnlock()
if fn != nil {
fn(peerID, relay)
}
}
}
func (m *HybridMesh) allowInboundRate(p *hybridPeer) bool {
p.rateMu.Lock()
defer p.rateMu.Unlock()
now := time.Now()
if p.rateWindow.IsZero() || now.Sub(p.rateWindow) >= time.Second {
p.rateWindow = now
p.rateCount = 1
return true
}
if p.rateCount >= meshVoiceInboundRatePerSec {
return false
}
p.rateCount++
return true
}
func (m *HybridMesh) voiceDrainLoop(ctx context.Context, peerID string) {
// TryRecv + short sleep so queue replace on reconnect never blocks forever
// on an abandoned channel (H-9 dual-conn replace).
for {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
default:
}
m.mu.RLock()
p := m.peers[peerID]
sendAddr := m.sendVoice[peerID]
pc := m.voicePC
m.mu.RUnlock()
if p == nil || pc == nil {
return
}
p.mu.Lock()
q := p.voiceQ
authed := p.authed
p.mu.Unlock()
if !authed || q == nil {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(20 * time.Millisecond):
}
continue
}
job, ok := q.TryRecv()
if !ok {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-time.After(2 * time.Millisecond):
}
continue
}
if sendAddr == nil {
continue
}
_, err := pc.WriteTo(job.packet, sendAddr)
if err != nil {
m.udpSendErrs.Add(1)
m.logger.Debug("AFV hybrid mesh UDP WriteTo", "peer", peerID, "err", err)
// UDP errors never alone mark death (H-8)
continue
}
m.voiceTx.Add(1)
p.voiceTxCount.Add(1)
p.lastVoiceUnixNano.Store(time.Now().UnixNano())
}
}
func (m *HybridMesh) muteWatchLoop(ctx context.Context) {
// Design Issue 15: ≤1/min Debug if Interest non-empty and no voice for >30s.
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
// per-peer last mute log time (unix nano)
lastLog := make(map[string]int64)
const muteSilence = 30 * time.Second
const muteLogMin = time.Minute
for {
select {
case <-m.stopCh:
return
case <-ctx.Done():
return
case <-ticker.C:
// Snapshot under m.mu only — never nest p.mu.
type snap struct {
id string
p *hybridPeer
has bool
}
m.mu.RLock()
snaps := make([]snap, 0, len(m.peers))
for id, p := range m.peers {
snaps = append(snaps, snap{id: id, p: p, has: len(m.peerInterest[id]) > 0})
}
m.mu.RUnlock()
now := time.Now()
nowN := now.UnixNano()
for _, s := range snaps {
if !s.has {
continue
}
s.p.mu.Lock()
authed := s.p.authed
s.p.mu.Unlock()
if !authed {
continue
}
lastV := s.p.lastVoiceUnixNano.Load()
// No activity yet since connect, or silence > 30s
silent := lastV == 0 || nowN-lastV > int64(muteSilence)
if !silent {
continue
}
if prev, ok := lastLog[s.id]; ok && nowN-prev < int64(muteLogMin) {
continue
}
lastLog[s.id] = nowN
m.logger.Debug("AFV hybrid mesh silent mute watch: Interest non-empty but no voice tx/rx for >30s",
"peer", s.id)
}
}
}
}

View File

@@ -0,0 +1,305 @@
package afv
import (
"bytes"
"context"
"net"
"sync/atomic"
"testing"
"time"
"github.com/renorris/openfsd/internal/geo"
)
func TestDecodeMeshFrameExact(t *testing.T) {
pay := []byte("hello")
raw, err := EncodeMeshFrameBytes(MeshTypeHeartbeat, pay)
if err != nil {
t.Fatal(err)
}
fr, err := DecodeMeshFrameExact(raw)
if err != nil || fr.Type != MeshTypeHeartbeat || !bytes.Equal(fr.Payload, pay) {
t.Fatalf("%+v %v", fr, err)
}
// trailing bytes
bad := append(append([]byte{}, raw...), 0x00)
if _, err := DecodeMeshFrameExact(bad); err == nil {
t.Fatal("trailing should fail")
}
// short
if _, err := DecodeMeshFrameExact(raw[:3]); err == nil {
t.Fatal("short should fail")
}
}
func TestHybridMesh_UDPUnknownSource(t *testing.T) {
m1, m2, _ := startHybridPair(t, "psk")
var got atomic.Int32
m1.OnAudioRelay(func(from string, r AudioRelay) {
got.Add(1)
})
// Send spoofed UDP to m1 voice from random port (not peer voice source port)
// Write to m1 voice listen with frame claiming origin n2
addr, err := net.ResolveUDPAddr("udp", m1.ListenVoiceAddr())
if err != nil {
t.Fatal(err)
}
cli, err := net.ListenPacket("udp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer cli.Close()
relay := AudioRelay{OriginNode: "n2", Callsign: "X", Audio: []byte{1, 2, 3},
TxRadios: []RelayTxRadio{{FreqHz: 1, LatDeg: 0, LonDeg: 0}}}
pay, _ := EncodeAudioRelay(relay)
pkt, _ := EncodeMeshFrameBytes(MeshTypeAudioRelay, pay)
before := m1.UDPAllowDrops()
_, _ = cli.WriteTo(pkt, addr)
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m1.UDPAllowDrops() > before {
break
}
time.Sleep(10 * time.Millisecond)
}
if got.Load() != 0 {
t.Fatal("OnAudioRelay must not fire for unknown source")
}
if m1.UDPAllowDrops() <= before {
t.Fatalf("want UDPAllowDrops increase, got %d→%d", before, m1.UDPAllowDrops())
}
_ = m2
}
func TestHybridMesh_UDPNotAuthed(t *testing.T) {
tcp1, voice1, tcp2, voice2 := hybridPairPorts(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Only start m1; peer n2 never authenticates
m1, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + itoa(tcp1), ListenVoice: "127.0.0.1:" + itoa(voice1),
Peers: []ClusterPeer{{ID: "n2", Addr: "127.0.0.1:" + itoa(tcp2), VoiceAddr: "127.0.0.1:" + itoa(voice2)}},
PSK: "psk",
})
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
defer m1.Stop()
var got atomic.Int32
m1.OnAudioRelay(func(from string, r AudioRelay) { got.Add(1) })
// Send from the configured peer voice port (allowlisted) but peer not authed
cli, err := net.ListenPacket("udp", "127.0.0.1:"+itoa(voice2))
if err != nil {
// port may be free since m2 not started
t.Fatal(err)
}
defer cli.Close()
addr, _ := net.ResolveUDPAddr("udp", m1.ListenVoiceAddr())
pay, _ := EncodeAudioRelay(AudioRelay{OriginNode: "n2", Callsign: "X", Audio: []byte{9}})
pkt, _ := EncodeMeshFrameBytes(MeshTypeAudioRelay, pay)
_, _ = cli.WriteTo(pkt, addr)
time.Sleep(100 * time.Millisecond)
if got.Load() != 0 {
t.Fatal("not-authed peer must not deliver")
}
}
func TestHybridMesh_IPv4MappedAllowlist(t *testing.T) {
// Unit: canonicalUDPKeys dual-key
ip := net.ParseIP("127.0.0.1")
keys := canonicalUDPKeys(ip, 17001)
foundV4, foundMapped := false, false
for _, k := range keys {
if k == "127.0.0.1:17001" {
foundV4 = true
}
if k == "[::ffff:127.0.0.1]:17001" || k == "::ffff:127.0.0.1:17001" {
foundMapped = true
}
}
if !foundV4 {
t.Fatalf("missing v4 key in %v", keys)
}
if !foundMapped {
// To16 string form of mapped
mapped := net.IP{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 127, 0, 0, 1}
mk := net.JoinHostPort(mapped.String(), "17001")
ok := false
for _, k := range keys {
if k == mk {
ok = true
}
}
if !ok {
t.Fatalf("missing mapped key; keys=%v want %s", keys, mk)
}
}
}
func TestHybridMesh_InterestEmptyNoFlood(t *testing.T) {
m1, m2, _ := startHybridPair(t, "psk")
// Empty Interest: Enqueue should not send UDP
before := m1.VoiceTx()
for i := 0; i < 20; i++ {
m1.EnqueueAudioRelay(AudioRelay{
OriginNode: "n1", Callsign: "A", SequenceCounter: uint32(i),
Audio: []byte{1, 2, 3},
TxRadios: []RelayTxRadio{{FreqHz: 122800000, LatDeg: 51.5, LonDeg: -0.1}},
})
}
time.Sleep(100 * time.Millisecond)
if m1.VoiceTx() != before {
t.Fatalf("empty Interest must not flood voice tx: %d→%d", before, m1.VoiceTx())
}
_ = m2
}
func TestHybridMesh_VoiceFanoutWithInterest(t *testing.T) {
m1, m2, _ := startHybridPair(t, "psk")
var got atomic.Int32
m2.OnAudioRelay(func(from string, r AudioRelay) {
if from == "n1" && r.Callsign == "TX1" {
got.Add(1)
}
})
// m2 advertises Interest for TX cell
lat, lon := 51.5, -0.12
freq := uint32(122800000)
ck := geo.CellKey{
ILat: geo.CellIndex(lat, geo.DefaultGridCellDeg),
ILon: geo.CellIndex(lon, geo.DefaultGridCellDeg),
}
m2.PublishInterest([]InterestEntry{{FreqHz: freq, ILat: ck.ILat, ILon: ck.ILon}})
// Wait until m1 sees PeerWants
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m1.PeerWants("n2", freq, ck) {
break
}
time.Sleep(10 * time.Millisecond)
}
if !m1.PeerWants("n2", freq, ck) {
t.Fatal("PeerWants not ready")
}
m1.EnqueueAudioRelay(AudioRelay{
OriginNode: "n1", Callsign: "TX1", SequenceCounter: 1,
Audio: []byte("opus"),
TxRadios: []RelayTxRadio{{TxID: 0, FreqHz: freq, LatDeg: lat, LonDeg: lon, HeightM: 100}},
})
deadline = time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if got.Load() > 0 {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatal("expected AudioRelay delivery over UDP")
}
func TestHybridMesh_UDPOversizeDrop(t *testing.T) {
// Lower cap so OS can send "oversize" datagrams on localhost.
old := maxMeshVoiceDatagram
maxMeshVoiceDatagram = 256
t.Cleanup(func() { maxMeshVoiceDatagram = old })
m1, m2, _ := startHybridPair(t, "psk")
var got atomic.Int32
m1.OnAudioRelay(func(from string, r AudioRelay) { got.Add(1) })
addr, err := net.ResolveUDPAddr("udp", m1.ListenVoiceAddr())
if err != nil {
t.Fatal(err)
}
before := m1.UDPOversizeDrops()
// Source from m2's mesh voice socket (allowlisted).
m2.mu.RLock()
pc := m2.voicePC
m2.mu.RUnlock()
if pc == nil {
t.Fatal("m2 voicePC nil")
}
big := make([]byte, maxMeshVoiceDatagram+64) // 320 > 256
if _, err := pc.WriteTo(big, addr); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if m1.UDPOversizeDrops() > before {
break
}
time.Sleep(10 * time.Millisecond)
}
if m1.UDPOversizeDrops() <= before {
t.Fatalf("want oversize drops, got %d→%d", before, m1.UDPOversizeDrops())
}
if got.Load() != 0 {
t.Fatal("OnAudioRelay must not fire for oversize")
}
}
func TestHybridMesh_VoiceQueueDropOldest(t *testing.T) {
// No voice drainer consuming: unauthed peer shell only.
tcp1, voice1, tcp2, voice2 := hybridPairPorts(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
m1, err := NewHybridMesh(HybridMeshConfig{
NodeID: "n1", ListenTCP: "127.0.0.1:" + itoa(tcp1), ListenVoice: "127.0.0.1:" + itoa(voice1),
Peers: []ClusterPeer{{ID: "n2", Addr: "127.0.0.1:" + itoa(tcp2), VoiceAddr: "127.0.0.1:" + itoa(voice2)}},
PSK: "psk",
})
if err != nil {
t.Fatal(err)
}
if err := m1.Start(ctx); err != nil {
t.Fatal(err)
}
defer m1.Stop()
// Voice drainers run but skip when !authed — TryRecv only when authed, so queue fills.
// Actually drain loop only TryRecv when authed; unauthed never drains → drops work.
before := m1.VoiceQueueDrops()
m1.ForceEnqueueVoiceForTest("n2", meshVoiceQueueDepth+30)
if m1.VoiceQueueDrops() <= before {
t.Fatalf("want voice drops, got %d→%d", before, m1.VoiceQueueDrops())
}
if m1.PeerVoiceQueueLen("n2") > meshVoiceQueueDepth {
t.Fatal("voice queue over depth")
}
}
func TestHybridMesh_InboundRateLimit(t *testing.T) {
// Lower rate isn't configurable; test that under 5000/s delivers and counter type exists.
// Send a burst of 50 — all should pass at 5000/s.
m1, m2, _ := startHybridPair(t, "psk")
var got atomic.Int32
m1.OnAudioRelay(func(from string, r AudioRelay) { got.Add(1) })
// Need authed + allowlist source = m2 voice listen
// Use internal path: mark as if from peer by calling handle after allowlist
// Direct unit: allowInboundRate
m1.mu.RLock()
p := m1.peers["n2"]
m1.mu.RUnlock()
// simulate 5000 allows then drop
p.rateMu.Lock()
p.rateWindow = time.Now()
p.rateCount = meshVoiceInboundRatePerSec
p.rateMu.Unlock()
if m1.allowInboundRate(p) {
t.Fatal("should rate-limit when at cap")
}
// reset window
p.rateMu.Lock()
p.rateWindow = time.Time{}
p.rateCount = 0
p.rateMu.Unlock()
if !m1.allowInboundRate(p) {
t.Fatal("should allow under cap")
}
_ = m2
_ = got
}

View File

@@ -99,13 +99,20 @@ Enabled only when the process is started with **`-afv`**. Shares `DATABASE_*` wi
| `AFV_RANGE_EDGE_RATIO` | Range edge ratio for volume falloff | `0.1` |
| `AFV_STATIONS_FILE` | Optional station alias file (empty → `[]` aliases) | *(empty)* |
| `AFV_CROSS_COUPLE` | Multi-freq ATC cross-coupling | `true` |
| `AFV_CLUSTER_ENABLED` | AFV multi-node mesh | `false` |
| `AFV_CLUSTER_ENABLED` | AFV multi-node mesh (hybrid TCP control + UDP voice) | `false` |
| `AFV_CLUSTER_NODE_ID` | AFV mesh node id | *(required when enabled)* |
| `AFV_CLUSTER_LISTEN` | AFV mesh listen | *(required when enabled)* |
| `AFV_CLUSTER_PEERS` | `id=host:port,...` (max 4 remote) | *(required when enabled)* |
| `AFV_CLUSTER_PSK` | AFV mesh shared secret | *(required when enabled)* |
| `AFV_CLUSTER_LISTEN` | **TCP control** listen `host:port` | *(required when enabled)* |
| `AFV_CLUSTER_VOICE_LISTEN` | **UDP mesh voice** listen `host:port` (separate from client `AFV_UDP_LISTEN`) | *(required when enabled)* |
| `AFV_CLUSTER_PEERS` | Remote peers: `id=host:tcpPort[/voicePort],...` (max 4). Default voice port = TCP+1 | *(required when enabled)* |
| `AFV_CLUSTER_PSK` | Shared Hello secret (constant-time compare) | *(required when enabled)* |
**Note:** `AFV_CLUSTER_ENABLED=true` currently **fails closed** without production TCP mesh in the binary (MemoryMesh is tests-only). See [Deployment](Deployment.md#optional-afv-voice).
**Multi-node notes:**
- Mesh is **inter-node only**. Clients stay sticky to one home node for REST + client UDP (`AFV_UDP_ADVERTISE_IPV4`).
- Open **both** control TCP and voice UDP between node security groups.
- Peer grammar examples: `n2=10.0.0.2:17000` (voice `10.0.0.2:17001`), `n2=10.0.0.2:17000/17100`, `n2=[2001:db8::1]:17000`.
- `AFV_CLUSTER_VOICE_LISTEN` must differ from `AFV_UDP_LISTEN` and from local control port. Recommended multi-host: every node `LISTEN=0.0.0.0:N`, `VOICE_LISTEN=0.0.0.0:N+1`, peers `id=otherHost:N`.
- Design: `docs/design/afv-mesh-pr10b.md`. Ops: [Deployment](Deployment.md#optional-afv-voice).
Minimal single-node AFV example:
@@ -117,6 +124,32 @@ AFV_UDP_ADVERTISE_IPV4=voice.example.com:50000
DATABASE_SOURCE_NAME=/db/openfsd.db?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)
```
Two-node localhost hybrid mesh example:
```text
# Node n1
AFV_CLUSTER_ENABLED=true
AFV_CLUSTER_NODE_ID=n1
AFV_CLUSTER_LISTEN=127.0.0.1:17000
AFV_CLUSTER_VOICE_LISTEN=127.0.0.1:17001
AFV_CLUSTER_PEERS=n2=127.0.0.1:17010/17011
AFV_CLUSTER_PSK=dev-shared-secret
AFV_UDP_LISTEN=127.0.0.1:50000
AFV_UDP_ADVERTISE_IPV4=127.0.0.1:50000
AFV_API_LISTEN=127.0.0.1:8080
# Node n2
AFV_CLUSTER_ENABLED=true
AFV_CLUSTER_NODE_ID=n2
AFV_CLUSTER_LISTEN=127.0.0.1:17010
AFV_CLUSTER_VOICE_LISTEN=127.0.0.1:17011
AFV_CLUSTER_PEERS=n1=127.0.0.1:17000/17001
AFV_CLUSTER_PSK=dev-shared-secret
AFV_UDP_LISTEN=127.0.0.1:50001
AFV_UDP_ADVERTISE_IPV4=127.0.0.1:50001
AFV_API_LISTEN=127.0.0.1:8081
```
### Logging
| Name | Description | Default |

View File

@@ -182,6 +182,29 @@ AFV is **opt-in** (`-afv`). It is not started by the default Docker CMD.
Share the same `DATABASE_*` as FSD so AFV authenticates against the same certificates. Optional `AFV_REQUIRE_FSD_ONLINE=true` gates voice sessions on the FSD online_users service.
Multi-node AFV mesh (`AFV_CLUSTER_*`) directory + AT relay is implemented for in-process/MemoryMesh testing; **production TCP mesh is not enabled** in the binary yet (`AFV_CLUSTER_ENABLED=true` fails closed until PR-10b). See `docs/design/afv-server.md` and `docs/design/afv-mesh-pr10.md`.
### Multi-node AFV (hybrid mesh)
Full env list: [Configuration](Configuration.md#afv-voice-optional).
Production multi-node voice uses a **hybrid mesh** (PR-10b): **TCP control** (Hello, Heartbeat, directory, Interest) + **UDP AudioRelay** between AFV processes. Clients remain sticky to one home node; mesh is inter-node only. AEAD keys never leave the home node.
| Plane | Env | Notes |
|-------|-----|--------|
| Control TCP | `AFV_CLUSTER_LISTEN` | Hello first-frame + PSK; dial if peer id > self |
| Voice UDP | `AFV_CLUSTER_VOICE_LISTEN` | Type-20 AudioRelay only; allowlist of peer voice addrs |
| Peers | `AFV_CLUSTER_PEERS` | `id=host:tcpPort[/voicePort]` (default voice = TCP+1); max 4 remote |
| PSK | `AFV_CLUSTER_PSK` | Shared secret; required when enabled |
**Firewall / security groups:** open **both** control TCP and mesh voice UDP between nodes. Client CryptoDTO UDP (`AFV_UDP_LISTEN`, often 50000) is **independent** of mesh voice (often N+1).
**Sticky LB:** REST must stay on the node that minted channel keys; advertise per-node `AFV_UDP_ADVERTISE_IPV4`. Mesh does not replace client affinity.
**Ops checklist:**
1. Both nodes log control TCP listen + voice UDP listen at Start
2. Hello success both directions
3. Interest non-empty after clients bind
4. Cross-node AR heard (in-range clients on different homes)
5. Kill one process → peer death warn; survivors keep local A2A
6. Wrong PSK → auth fail; no voice
7. If Interest is non-empty but cross-node mute: check UDP firewall / path MTU (large clamps can IP-fragment on hostile paths)
Full env list + two-node example: [Configuration](Configuration.md#afv-voice-optional). Design: `docs/design/afv-mesh-pr10b.md`.