rendezvous: faster snapshot saves (hand-written encoder + trust-pairs cache)

replication.go

@@ -155,8 +155,12 @@ func (s *Server) snapshotJSON() []byte {
 		snap.Networks[fmt.Sprintf("%d", id)] = sn
 	}
 
-	// Include trust pairs via the trust sub-package (R2.1).
-	snap.TrustPairs = s.trust.Pairs()
+	// Include trust pairs via the trust sub-package (R2.1), served from the
+	// same revision-keyed cache flushSave uses so replication doesn't pay the
+	// full re-encode either.
+	if tp, err := s.trustPairsCache.encode(s.trust.Revision(), s.trust.Pairs, trustPairsEncodeCacheTTL); err == nil {
+		snap.TrustPairs = tp
+	}
 
 	// Include handshake inboxes via the trust sub-package (R2.1).
 	inbox, responses := s.trust.InboxSnapshot()
@@ -476,8 +480,14 @@ func (s *Server) applySnapshot(data []byte) error {
 
 	// --- Phase 3: side state under their own locks ---
 
-	// Trust pairs and handshake state are owned by s.trust (R2.1).
-	s.trust.RestorePairs(snap.TrustPairs)
+	// Trust pairs and handshake state are owned by s.trust (R2.1). TrustPairs
+	// arrives as a pre-encoded JSON array (RawMessage).
+	if len(snap.TrustPairs) > 0 {
+		var pairs []string
+		if err := json.Unmarshal(snap.TrustPairs, &pairs); err == nil {
+			s.trust.RestorePairs(pairs)
+		}
+	}
 	s.trust.RestoreInbox(newHandshakeInbox, newHandshakeResponses)
 
 	if len(snap.AuditLog) > 0 {

server.go

@@ -97,6 +97,17 @@ type Server struct {
 	lastSnapshotRLockMs atomic.Int64 // s.mu.RLock hold time in phase 1
 	maxSnapshotDurMs    atomic.Int64 // worst-ever save duration
 
+	// trustPairsCache memoizes the JSON encoding of the trust-pair list
+	// across saves (see server_persist.go). The set is near-static at fleet
+	// scale, so this avoids re-encoding millions of strings every save.
+	trustPairsCache trustPairsEncodeCache
+
+	// snapshotBuf is the reused buffer for the hand-written snapshot encoder
+	// (wal.AppendSnapshot). snapshotEncodeMu guards it against concurrent
+	// flushSave callers (saveLoop and a rare replication-triggered flush).
+	snapshotEncodeMu sync.Mutex
+	snapshotBuf      []byte
+
 	// buildInfo carries the build-time identity surfaced on
 	// /api/public-stats for code-verification (version, git commit, ISO
 	// build time, SHA256 of /proc/self/exe, Go runtime version). Set

server_persist.go

@@ -14,13 +14,15 @@ import (
 	"os"
 	"path/filepath"
 	"runtime/debug"
+	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/pilot-protocol/common/registry/wire"
 	"github.com/pilot-protocol/common/urlvalidate"
 	dashpkg "github.com/pilot-protocol/rendezvous/dashboard"
 	trustpkg "github.com/pilot-protocol/rendezvous/trust"
+	walpkg "github.com/pilot-protocol/rendezvous/wal"
 )
 
 const (
@@ -30,37 +32,6 @@ const (
 
 var lastScavengeMs atomic.Int64
 
-// trimTrailingNewlineWriter forwards every byte it is given except the final
-// one. json.Encoder.Encode terminates its output with '\n', but the snapshot
-// checksum is defined over the encoded value WITHOUT that newline (load()
-// strips it before recomputing too). Holding back the last byte yields a file
-// and hash of `{...}` while writing straight to disk — no full-size buffer.
-type trimTrailingNewlineWriter struct {
-	w       io.Writer
-	held    byte
-	hasHeld bool
-}
-
-func (t *trimTrailingNewlineWriter) Write(p []byte) (int, error) {
-	if len(p) == 0 {
-		return 0, nil
-	}
-	if t.hasHeld {
-		if _, err := t.w.Write([]byte{t.held}); err != nil {
-			return 0, err
-		}
-		t.hasHeld = false
-	}
-	if len(p) > 1 {
-		if _, err := t.w.Write(p[:len(p)-1]); err != nil {
-			return 0, err
-		}
-	}
-	t.held = p[len(p)-1]
-	t.hasHeld = true
-	return len(p), nil
-}
-
 func shouldScavenge(dataLen int, nowMs int64) bool {
 	if dataLen < scavengeMinSnapshot {
 		return false
@@ -69,6 +40,56 @@ func shouldScavenge(dataLen int, nowMs int64) bool {
 	return nowMs-last >= scavengeIntervalMs && lastScavengeMs.CompareAndSwap(last, nowMs)
 }
 
+// trustPairsEncodeCacheTTL bounds how long a cached trust-pairs encoding is
+// served without re-encoding, independent of the revision. The revision
+// invalidation is exact; this is a belt-and-braces safety net so a missed
+// mutation can't wedge a stale list indefinitely.
+const trustPairsEncodeCacheTTL = 10 * time.Minute
+
+// trustPairsEncodeCache memoizes the JSON array of trust-pair keys. The set
+// (millions of entries at fleet scale) is near-static, so reusing the bytes
+// across saves removes the dominant cost of snapshot encoding while
+// producing output byte-identical to encoding the slice directly.
+type trustPairsEncodeCache struct {
+	mu  sync.Mutex
+	rev uint64
+	at  time.Time
+	b   json.RawMessage
+}
+
+// encode returns the JSON array for pairs, reusing the cached bytes while
+// rev is unchanged and the TTL hasn't elapsed. pairs is only invoked on a
+// cache miss. An empty list returns nil so the omitempty tag still drops the
+// field, matching the previous []string behaviour.
+func (c *trustPairsEncodeCache) encode(rev uint64, pairs func() []string, ttl time.Duration) (json.RawMessage, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.b != nil && c.rev == rev && time.Since(c.at) < ttl {
+		return c.b, nil
+	}
+	p := pairs()
+	if len(p) == 0 {
+		c.b = nil
+		c.rev = rev
+		c.at = time.Now()
+		return nil, nil
+	}
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(p); err != nil {
+		return nil, err
+	}
+	b := buf.Bytes()
+	if len(b) > 0 && b[len(b)-1] == '\n' {
+		b = b[:len(b)-1]
+	}
+	c.b = append(c.b[:0], b...)
+	c.rev = rev
+	c.at = time.Now()
+	return c.b, nil
+}
+
 // atomicWriteStream writes path atomically via a temp file + rename, but
 // streams the content: fn receives the open *os.File and may write
 // incrementally (and Seek/WriteAt to backfill a trailer) instead of the
@@ -307,8 +328,9 @@ func (s *Server) flushSave() (retErr error) {
 		}
 	}
 
-	// Copy trust pairs and handshake inboxes from the trust sub-package.
-	trustPairs := s.trust.Pairs()
+	// Handshake inboxes are copied here; the trust-pair list is encoded in
+	// Phase 2 through the revision-keyed cache (it is near-static and by far
+	// the most expensive part to serialize).
 	handshakeInbox, handshakeResponses := s.trust.InboxSnapshot()
 	var inviteInbox map[uint32][]*NetworkInvite
 	if len(s.inviteInbox) > 0 {
@@ -457,7 +479,11 @@ func (s *Server) flushSave() (retErr error) {
 	}
 
 	snap.PubKeyIdx = pubKeyIdx
-	snap.TrustPairs = trustPairs
+	tp, tpErr := s.trustPairsCache.encode(s.trust.Revision(), s.trust.Pairs, trustPairsEncodeCacheTTL)
+	if tpErr != nil {
+		return fmt.Errorf("encode trust pairs: %w", tpErr)
+	}
+	snap.TrustPairs = tp
 
 	// Handshake inboxes
 	if len(handshakeInbox) > 0 {
@@ -551,52 +577,35 @@ func (s *Server) flushSave() (retErr error) {
 	}
 	s.auditMu.Unlock()
 
-	// Stream-encode straight to the temp file: no full-size in-memory buffer
-	// and no second copy through AtomicWrite. The body is hashed as it is
-	// written, then the checksum is backfilled as a trailing JSON member.
-	//
-	// At 170k+ nodes the old path materialized a ~50-100 MB pooled buffer and
-	// was the dominant heap allocator (~1 GB live, GC STW pauses causing UDP
-	// drops). Streaming writes the same bytes directly to disk instead.
-	//
-	// Checksum invariant: load() re-encodes the struct with Checksum omitted
-	// and hashes those bytes (newline stripped). So the hash here must cover
-	// exactly the encoder output minus its trailing '\n' — which
-	// trimTrailingNewlineWriter guarantees by withholding the final byte.
+	// Encode with the hand-written, reflection-free encoder (wal.AppendSnapshot)
+	// into a buffer reused across saves. It is byte-identical to encoding/json
+	// (differential-tested), so the checksum and load() parity are unchanged,
+	// but it skips the reflection + per-node allocation that dominated saves at
+	// fleet scale. The body is hashed, then the checksum is written as a
+	// trailing JSON member by replacing the final '}'.
 	snap.Checksum = ""
 	var written int64
 	if s.storePath != "" {
+		s.snapshotEncodeMu.Lock()
+		body := walpkg.AppendSnapshot(s.snapshotBuf[:0], &snap)
+		s.snapshotBuf = body
+		if len(body) == 0 || body[len(body)-1] != '}' {
+			s.snapshotEncodeMu.Unlock()
+			return fmt.Errorf("encode snapshot: unexpected JSON format (expected trailing '}')")
+		}
+		sum := sha256.Sum256(body)
+		trailer := `,"checksum":"` + hex.EncodeToString(sum[:]) + `"}`
 		err := atomicWriteStream(s.storePath, func(f *os.File) error {
-			h := sha256.New()
-			enc := json.NewEncoder(&trimTrailingNewlineWriter{w: io.MultiWriter(f, h)})
-			enc.SetEscapeHTML(false)
-			if err := enc.Encode(snap); err != nil {
-				return fmt.Errorf("encode snapshot: %w", err)
-			}
-			// The encoded object ends with '}' (the trailing newline was
-			// withheld). Replace that '}' with the checksum trailer so the
-			// on-disk file is `{...` + `,"checksum":"<hex>"}`.
-			end, err := f.Seek(0, io.SeekEnd)
-			if err != nil {
-				return fmt.Errorf("seek snapshot temp: %w", err)
-			}
-			var last [1]byte
-			if end == 0 {
-				return fmt.Errorf("encode snapshot: empty output")
+			if _, err := f.Write(body[:len(body)-1]); err != nil {
+				return fmt.Errorf("write snapshot body: %w", err)
 			}
-			if _, err := f.ReadAt(last[:], end-1); err != nil {
-				return fmt.Errorf("read snapshot tail: %w", err)
-			}
-			if last[0] != '}' {
-				return fmt.Errorf("encode snapshot: unexpected JSON format (expected trailing '}')")
-			}
-			trailer := `,"checksum":"` + hex.EncodeToString(h.Sum(nil)) + `"}`
-			if _, err := f.WriteAt([]byte(trailer), end-1); err != nil {
+			if _, err := io.WriteString(f, trailer); err != nil {
 				return fmt.Errorf("write checksum trailer: %w", err)
 			}
-			written = end - 1 + int64(len(trailer))
+			written = int64(len(body) - 1 + len(trailer))
 			return nil
 		})
+		s.snapshotEncodeMu.Unlock()
 		if err != nil {
 			slog.Error("registry save error", "err", err)
 			return fmt.Errorf("write snapshot: %w", err)
@@ -893,10 +902,15 @@ func (s *Server) load() error {
 		s.networks[n.ID] = net
 	}
 
-	// Restore trust pairs (delegated to the trust sub-package).
-	s.trust.RestorePairs(snap.TrustPairs)
+	// Restore trust pairs (delegated to the trust sub-package). TrustPairs is
+	// a pre-encoded JSON array (RawMessage) so flushSave can cache it.
 	if len(snap.TrustPairs) > 0 {
-		slog.Info("loaded trust pairs", "count", len(snap.TrustPairs))
+		var pairs []string
+		if err := json.Unmarshal(snap.TrustPairs, &pairs); err != nil {
+			return fmt.Errorf("decode trust_pairs: %w", err)
+		}
+		s.trust.RestorePairs(pairs)
+		slog.Info("loaded trust pairs", "count", len(pairs))
 	}
 
 	// Restore persisted pubKeyIdx (entries for reaped nodes that aren't in snap.Nodes)

trust/pairset.go

@@ -2,7 +2,10 @@
 
 package trust
 
-import "sync"
+import (
+	"sync"
+	"sync/atomic"
+)
 
 // trustPairShards is the shard count for the trust-pair set. A power of two
 // so the index is a mask. Sized well above the registry's request
@@ -27,8 +30,15 @@ type trustPairSet struct {
 		mu sync.RWMutex
 		m  map[string]bool
 	}
+	// rev ticks on every mutation that actually changes the set (new add or
+	// real remove). Consumers can use it as a cache key: an unchanged rev
+	// means the serialized pair list is unchanged.
+	rev atomic.Uint64
 }
 
+// revision returns the set's mutation counter.
+func (s *trustPairSet) revision() uint64 { return s.rev.Load() }
+
 func newTrustPairSet() *trustPairSet {
 	s := &trustPairSet{}
 	for i := range s.shards {
@@ -47,7 +57,10 @@ func (s *trustPairSet) has(key string) bool {
 func (s *trustPairSet) add(key string) {
 	sh := &s.shards[trustShardIdx(key)]
 	sh.mu.Lock()
-	sh.m[key] = true
+	if !sh.m[key] {
+		sh.m[key] = true
+		s.rev.Add(1)
+	}
 	sh.mu.Unlock()
 }
 
@@ -60,6 +73,7 @@ func (s *trustPairSet) remove(key string) bool {
 		return false
 	}
 	delete(sh.m, key)
+	s.rev.Add(1)
 	return true
 }
 
@@ -89,5 +103,9 @@ func (s *trustPairSet) keys() []string {
 
 // addUnlocked inserts without locking — for startup restore before serving.
 func (s *trustPairSet) addUnlocked(key string) {
-	s.shards[trustShardIdx(key)].m[key] = true
+	m := s.shards[trustShardIdx(key)].m
+	if !m[key] {
+		m[key] = true
+		s.rev.Add(1)
+	}
 }

trust/trust.go

@@ -14,8 +14,8 @@ import (
 	"log/slog"
 	"time"
 
-	"github.com/pilot-protocol/common/protocol"
 	"github.com/pilot-protocol/common/crypto"
+	"github.com/pilot-protocol/common/protocol"
 )
 
 // maxHandshakeInbox limits the number of pending handshake requests per node.
@@ -92,8 +92,8 @@ type Store struct {
 // NewStore creates an empty, ready-to-use Store.
 func NewStore(nodes NodeView, cb Callbacks) *Store {
 	return &Store{
-		nodes:              nodes,
-		cb:                 cb,
+		nodes:      nodes,
+		cb:         cb,
 		trustPairs: newTrustPairSet(),
 		handshakes: newHandshakeState(),
 	}
@@ -106,6 +106,13 @@ func (st *Store) Count() int {
 	return st.trustPairs.count()
 }
 
+// Revision returns a counter that increments whenever the trust-pair set
+// actually changes (a new pair is added or one is removed). It is a cheap
+// cache key for consumers that serialize Pairs(): same revision ⇒ same set.
+func (st *Store) Revision() uint64 {
+	return st.trustPairs.revision()
+}
+
 // IsTrusted reports whether nodes a and b have an established trust pair.
 // The relation is symmetric: IsTrusted(a, b) == IsTrusted(b, a).
 func (st *Store) IsTrusted(a, b uint32) bool {