fix: bound registry RSS growth (cap pooled save buffer + return freed pages)

server_persist.go

@@ -12,7 +12,9 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"runtime/debug"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pilot-protocol/common/fsutil"
@@ -22,6 +24,14 @@ import (
 	trustpkg "github.com/pilot-protocol/rendezvous/trust"
 )
 
+const (
+	maxPooledSaveBuf    = 128 << 20
+	scavengeIntervalMs  = 180_000
+	scavengeMinSnapshot = 64 << 20
+)
+
+var lastScavengeMs atomic.Int64
+
 // flushSaveBufPool reuses the bytes buffer that backs the snapshot JSON
 // across save ticks. After the first few saves the pool returns a
 // buffer at peak capacity, so subsequent saves do zero allocation in
@@ -40,6 +50,20 @@ var flushSaveBufPool = sync.Pool{
 	},
 }
 
+func putSaveBuf(bp *[]byte) {
+	if cap(*bp) <= maxPooledSaveBuf {
+		flushSaveBufPool.Put(bp)
+	}
+}
+
+func shouldScavenge(dataLen int, nowMs int64) bool {
+	if dataLen < scavengeMinSnapshot {
+		return false
+	}
+	last := lastScavengeMs.Load()
+	return nowMs-last >= scavengeIntervalMs && lastScavengeMs.CompareAndSwap(last, nowMs)
+}
+
 // rawNodeCopy holds raw node fields copied under RLock (no encoding).
 // base64/time.Format happens outside the lock to minimize lock hold time.
 type rawNodeCopy struct {
@@ -502,7 +526,7 @@ func (s *Server) flushSave() (retErr error) {
 	enc.SetEscapeHTML(false)
 	if err := enc.Encode(snap); err != nil {
 		*bp = buf.Bytes()[:0]
-		flushSaveBufPool.Put(bp)
+		putSaveBuf(bp)
 		slog.Error("registry save encode error", "err", err)
 		return fmt.Errorf("encode snapshot: %w", err)
 	}
@@ -517,15 +541,15 @@ func (s *Server) flushSave() (retErr error) {
 	// always produces a JSON object ending with '}' (after newline trim).
 	if len(data) == 0 || data[len(data)-1] != '}' {
 		*bp = buf.Bytes()[:0]
-		flushSaveBufPool.Put(bp)
+		putSaveBuf(bp)
 		return fmt.Errorf("encode snapshot: unexpected JSON format (expected trailing '}')")
 	}
 	data = append(data[:len(data)-1], []byte(`,"checksum":"`+checksum+`"}`)...)
 	defer func() {
 		// Return the (possibly grown) underlying buffer to the pool. AtomicWrite
 		// has copied data to disk by this point, so it is safe to release.
 		*bp = data[:0]
-		flushSaveBufPool.Put(bp)
+		putSaveBuf(bp)
 	}()
 
 	// Persist to disk atomically
@@ -535,6 +559,9 @@ func (s *Server) flushSave() (retErr error) {
 			return fmt.Errorf("write snapshot: %w", err)
 		}
 		s.lastSnapshotSizeB.Store(int64(len(data)))
+		if shouldScavenge(len(data), time.Now().UnixMilli()) {
+			go debug.FreeOSMemory()
+		}
 		// Truncate WAL after successful snapshot (compaction).
 		if w := s.walStore.WAL(); w != nil {
 			if err := w.Truncate(); err != nil {

zz_save_memory_test.go

@@ -0,0 +1,42 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package server
+
+import "testing"
+
+func TestPutSaveBufDropsOversized(t *testing.T) {
+	big := make([]byte, 0, maxPooledSaveBuf+1)
+	putSaveBuf(&big)
+	got := flushSaveBufPool.Get().(*[]byte)
+	if cap(*got) > maxPooledSaveBuf {
+		t.Fatalf("pool retained an oversized buffer: cap=%d", cap(*got))
+	}
+}
+
+func TestPutSaveBufKeepsNormal(t *testing.T) {
+	buf := make([]byte, 0, 1<<20)
+	putSaveBuf(&buf)
+	got := flushSaveBufPool.Get().(*[]byte)
+	if cap(*got) < 1<<20 {
+		t.Fatalf("normal-sized buffer was not pooled: cap=%d", cap(*got))
+	}
+}
+
+func TestShouldScavengeGate(t *testing.T) {
+	lastScavengeMs.Store(0)
+
+	if shouldScavenge(scavengeMinSnapshot-1, scavengeIntervalMs*10) {
+		t.Fatal("scavenge fired for a small snapshot")
+	}
+
+	base := int64(scavengeIntervalMs * 10)
+	if !shouldScavenge(scavengeMinSnapshot, base) {
+		t.Fatal("scavenge did not fire for a large snapshot after the interval")
+	}
+	if shouldScavenge(scavengeMinSnapshot, base+1) {
+		t.Fatal("scavenge fired again inside the rate-limit window")
+	}
+	if !shouldScavenge(scavengeMinSnapshot, base+scavengeIntervalMs) {
+		t.Fatal("scavenge did not fire again after the interval elapsed")
+	}
+}