Registry hardening: panic recovery, WAL replay/cap, 3-phase patterns

Panic recovery wrapper around registry handlers — a panicking handler
no longer takes the server down; stack trace logged, connection closed.

WAL replay on startup. Size cap (configurable) so a stuck flushSave
cannot let the WAL grow until disk fills.

3-phase pattern (RLock-verify-unlock-Lock-mutate) applied to
handleRotateKey + handleSetKeyExpiry — eliminates lock-hold during
signature checks.

Plus: heartbeat-ack pre-build to keep the hot path off json.Marshal,
list_nodes cache-invalidation hooks at every membership mutation site,
apply_snapshot deep-copy moved outside s.mu (~350x lock-hold drop),
snapshot lock-discipline tests, stale-threshold tests.

RelayOnly registration mode: registry substitutes beacon_addr for
real_addr in resolve responses when the peer registered with
RelayOnly=true.

Author: teovl

pkg/registry/apply_snapshot_test.go

@@ -0,0 +1,164 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package registry
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// buildSnapshot constructs a synthetic snapshot JSON blob for tests. It
+// produces the exact wire shape applySnapshot expects.
+func buildSnapshot(t *testing.T, nodes int, networks int) []byte {
+	t.Helper()
+	now := time.Now().UTC().Format(time.RFC3339)
+
+	snap := snapshot{
+		NextNode: uint32(nodes + 100),
+		NextNet:  uint16(networks + 10),
+		Nodes:    make(map[string]*snapshotNode, nodes),
+		Networks: make(map[string]*snapshotNet, networks),
+	}
+
+	for i := 1; i <= nodes; i++ {
+		key := fmt.Sprintf("%d", i)
+		snap.Nodes[key] = &snapshotNode{
+			ID:        uint32(i),
+			Owner:     fmt.Sprintf("owner-%d", i),
+			PublicKey: "AAAA" + fmt.Sprintf("%032d", i),
+			RealAddr:  fmt.Sprintf("10.0.0.%d:9000", i%255),
+			Networks:  []uint16{1},
+			LastSeen:  now,
+			Hostname:  fmt.Sprintf("host-%d", i),
+		}
+	}
+
+	for i := 1; i <= networks; i++ {
+		key := fmt.Sprintf("%d", i)
+		members := make([]uint32, 0, 5)
+		for j := i; j <= nodes && j < i+5; j++ {
+			members = append(members, uint32(j))
+		}
+		snap.Networks[key] = &snapshotNet{
+			ID:       uint16(i),
+			Name:     fmt.Sprintf("net-%d", i),
+			JoinRule: "open",
+			Members:  members,
+			Created:  now,
+		}
+	}
+
+	data, err := json.Marshal(snap)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	return data
+}
+
+// TestApplySnapshotBasic locks down the happy path: applySnapshot replaces
+// the registry's nodes, networks, and indices wholesale.
+func TestApplySnapshotBasic(t *testing.T) {
+	s := newTestServer(t, "")
+	data := buildSnapshot(t, 5, 2)
+
+	if err := s.applySnapshot(data); err != nil {
+		t.Fatalf("applySnapshot: %v", err)
+	}
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if got := len(s.nodes); got != 5 {
+		t.Fatalf("nodes count = %d, want 5", got)
+	}
+	if got := len(s.networks); got != 2 {
+		t.Fatalf("networks count = %d, want 2", got)
+	}
+	if s.nextNode != 105 {
+		t.Fatalf("nextNode = %d, want 105", s.nextNode)
+	}
+	if s.nextNet != 12 {
+		t.Fatalf("nextNet = %d, want 12", s.nextNet)
+	}
+	// Hostname index populated
+	if got := s.hostnameIdx["host-1"]; got != 1 {
+		t.Fatalf("hostnameIdx[host-1] = %d, want 1", got)
+	}
+}
+
+// TestApplySnapshotReplacesPreviousState validates that a second apply
+// fully replaces the prior state — the legacy contract.
+func TestApplySnapshotReplacesPreviousState(t *testing.T) {
+	s := newTestServer(t, "")
+	if err := s.applySnapshot(buildSnapshot(t, 3, 1)); err != nil {
+		t.Fatalf("first apply: %v", err)
+	}
+	if err := s.applySnapshot(buildSnapshot(t, 7, 3)); err != nil {
+		t.Fatalf("second apply: %v", err)
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if got := len(s.nodes); got != 7 {
+		t.Fatalf("nodes count after second apply = %d, want 7", got)
+	}
+	if got := len(s.networks); got != 3 {
+		t.Fatalf("networks count after second apply = %d, want 3", got)
+	}
+}
+
+// TestApplySnapshotRejectsTooLarge enforces the size cap.
+func TestApplySnapshotRejectsTooLarge(t *testing.T) {
+	s := newTestServer(t, "")
+	huge := make([]byte, maxSnapshotSize+1)
+	if err := s.applySnapshot(huge); err == nil {
+		t.Fatalf("expected error for oversized snapshot")
+	}
+}
+
+// TestApplySnapshotConcurrentReadsStaySane stresses the new pattern: while
+// applySnapshot runs, concurrent readers (sampleStats) must see either the
+// fully-old or fully-new state — not a partial mix. Run with -race.
+func TestApplySnapshotConcurrentReadsStaySane(t *testing.T) {
+	s := newTestServer(t, "")
+	if err := s.applySnapshot(buildSnapshot(t, 50, 5)); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	const iterations = 50
+	var stop atomic.Bool
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations && !stop.Load(); i++ {
+			result := s.sampleStats()
+			// State count is either the prior state size or new state size,
+			// not negative or absurd. We don't pin a specific count because
+			// it's racing with applySnapshot; we just want the read path to
+			// not panic and to return a non-empty global sample.
+			if result.global.TotalNodes < 0 {
+				t.Errorf("negative TotalNodes")
+				return
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations/5 && !stop.Load(); i++ {
+			data := buildSnapshot(t, 50+i, 5)
+			if err := s.applySnapshot(data); err != nil {
+				t.Errorf("applySnapshot at iter %d: %v", i, err)
+				return
+			}
+		}
+	}()
+
+	wg.Wait()
+	stop.Store(true)
+}

pkg/registry/cache_invalidation_test.go

@@ -0,0 +1,204 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package registry
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+// stubCacheEntry installs a sentinel cache entry under the given netID so
+// tests can assert whether handlers invalidate it.
+func stubCacheEntry(t *testing.T, s *Server, netID uint16) {
+	t.Helper()
+	c := &listNodesCacheState{
+		fullBody: []byte(`{"sentinel":true}`),
+		builtAt:  time.Now(),
+	}
+	c.cond = sync.NewCond(&c.mu)
+	s.listNodesPerNetMu.Lock()
+	s.listNodesPerNet[netID] = c
+	s.listNodesPerNetMu.Unlock()
+}
+
+// hasCacheEntry returns true if a per-network cache state still exists.
+func hasCacheEntry(s *Server, netID uint16) bool {
+	s.listNodesPerNetMu.Lock()
+	defer s.listNodesPerNetMu.Unlock()
+	_, ok := s.listNodesPerNet[netID]
+	return ok
+}
+
+// seedNetworkAndNode installs a network with a single node directly in
+// server state. Bypasses the create/register handlers so tests don't have
+// to deal with crypto identity. Returns the netID + nodeID.
+func seedNetworkAndNode(t *testing.T, s *Server) (uint16, uint32) {
+	t.Helper()
+	const netID uint16 = 42
+	const nodeID uint32 = 1001
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.networks[netID] = &NetworkInfo{
+		ID:       netID,
+		Name:     "test-net",
+		JoinRule: "open",
+		Members:  []uint32{nodeID},
+		MemberRoles: map[uint32]Role{
+			nodeID: RoleOwner,
+		},
+		Created: time.Now(),
+	}
+	n := &NodeInfo{
+		ID:        nodeID,
+		Owner:     "test",
+		PublicKey: []byte("fake-pk"),
+		Networks:  []uint16{netID},
+	}
+	n.setLastSeen(time.Now())
+	s.nodes[nodeID] = n
+	if s.nextNet <= netID {
+		s.nextNet = netID + 1
+	}
+	if s.nextNode <= nodeID {
+		s.nextNode = nodeID + 1
+	}
+	return netID, nodeID
+}
+
+// TestInvalidateListNodesCacheForNetworkDropsEntry validates the helper
+// itself (used by all mutation paths). Smoke test for the contract.
+func TestInvalidateListNodesCacheForNetworkDropsEntry(t *testing.T) {
+	s := newTestServer(t, "")
+	stubCacheEntry(t, s, 42)
+	if !hasCacheEntry(s, 42) {
+		t.Fatalf("setup: expected cache entry for net 42")
+	}
+	s.invalidateListNodesCacheForNetwork(42)
+	if hasCacheEntry(s, 42) {
+		t.Fatalf("cache entry for net 42 should be gone after invalidate")
+	}
+}
+
+// TestHandleDeleteNetworkInvalidatesCache locks down the rc3 fix: when a
+// network is deleted, its cached list_nodes response must be dropped so we
+// don't leak orphaned cache memory and don't return zombie data if the same
+// netID is reused.
+func TestHandleDeleteNetworkInvalidatesCache(t *testing.T) {
+	s := newTestServer(t, "ADM")
+	netID, _ := seedNetworkAndNode(t, s)
+
+	stubCacheEntry(t, s, netID)
+	if !hasCacheEntry(s, netID) {
+		t.Fatalf("setup: cache entry should exist for net %d", netID)
+	}
+
+	if _, err := s.handleDeleteNetwork(map[string]interface{}{
+		"network_id":  float64(netID),
+		"admin_token": "ADM",
+	}); err != nil {
+		t.Fatalf("delete_network: %v", err)
+	}
+
+	if hasCacheEntry(s, netID) {
+		t.Fatalf("cache entry for net %d should be invalidated after delete_network", netID)
+	}
+}
+
+// TestHandleJoinNetworkInvalidatesCache pins the existing wiring at
+// server.go:3403 so a future refactor can't accidentally drop it.
+func TestHandleJoinNetworkInvalidatesCache(t *testing.T) {
+	s := newTestServer(t, "ADM")
+
+	// Pre-existing network with no members; we'll join it.
+	const netID uint16 = 50
+	const nodeID uint32 = 2001
+	s.mu.Lock()
+	s.networks[netID] = &NetworkInfo{
+		ID:       netID,
+		Name:     "join-test",
+		JoinRule: "open",
+		Members:  []uint32{},
+		Created:  time.Now(),
+	}
+	s.nodes[nodeID] = &NodeInfo{
+		ID:        nodeID,
+		Owner:     "alice",
+		PublicKey: []byte("pk"),
+	}
+	s.nodes[nodeID].setLastSeen(time.Now())
+	if s.nextNet <= netID {
+		s.nextNet = netID + 1
+	}
+	if s.nextNode <= nodeID {
+		s.nextNode = nodeID + 1
+	}
+	s.mu.Unlock()
+
+	stubCacheEntry(t, s, netID)
+	if !hasCacheEntry(s, netID) {
+		t.Fatalf("setup: cache should exist for net %d", netID)
+	}
+
+	if _, err := s.handleJoinNetwork(map[string]interface{}{
+		"node_id":     float64(nodeID),
+		"network_id":  float64(netID),
+		"admin_token": "ADM",
+	}); err != nil {
+		t.Fatalf("join_network: %v", err)
+	}
+
+	if hasCacheEntry(s, netID) {
+		t.Fatalf("cache entry for net %d should be invalidated after join_network", netID)
+	}
+}
+
+// TestHandleLeaveNetworkInvalidatesCache pins the wiring at server.go:3493.
+func TestHandleLeaveNetworkInvalidatesCache(t *testing.T) {
+	s := newTestServer(t, "ADM")
+	netID, nodeID := seedNetworkAndNode(t, s)
+
+	stubCacheEntry(t, s, netID)
+	if !hasCacheEntry(s, netID) {
+		t.Fatalf("setup: cache should exist for net %d", netID)
+	}
+
+	if _, err := s.handleLeaveNetwork(map[string]interface{}{
+		"node_id":     float64(nodeID),
+		"network_id":  float64(netID),
+		"admin_token": "ADM",
+	}); err != nil {
+		t.Fatalf("leave_network: %v", err)
+	}
+
+	if hasCacheEntry(s, netID) {
+		t.Fatalf("cache entry for net %d should be invalidated after leave_network", netID)
+	}
+}
+
+// TestConcurrentInvalidateAndPopulateRace exercises the cache map under
+// `-race` with simultaneous invalidate + manual populate to catch any
+// missing locking on the listNodesPerNet map. Run with `go test -race`.
+func TestConcurrentInvalidateAndPopulateRace(t *testing.T) {
+	s := newTestServer(t, "")
+
+	const netID uint16 = 7
+	const iterations = 500
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			stubCacheEntry(t, s, netID)
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := 0; i < iterations; i++ {
+			s.invalidateListNodesCacheForNetwork(netID)
+		}
+	}()
+	wg.Wait()
+}

pkg/registry/delta.go

@@ -24,6 +24,10 @@ const (
 	DeltaNetworkLeave  DeltaType = 11
 	DeltaKeyRotation   DeltaType = 12
 	DeltaTaskExec      DeltaType = 13
+	// DeltaNetworkDelete added for WAL wiring (2026-04-29). On older
+	// binaries that don't recognize this type, replay logs and skips —
+	// safe forward-compat.
+	DeltaNetworkDelete DeltaType = 14
 )
 
 // DeltaEntry records a single state mutation for incremental replication.