From 39a5c8d16d66f6133c49764a3a1e9947be6c2d86 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sat, 25 Jul 2026 01:01:04 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20bound=20registry=20RSS=20growth=20?= =?UTF-8?q?=E2=80=94=20cap=20pooled=20save=20buffer=20+=20return=20freed?= =?UTF-8?q?=20pages?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-caused from a live heap profile during the recurring registry wedge: live Go heap was ~4GB but RSS ~22.7GB and climbing ~3.3GB/hr with a stable connection count. Two compounding causes, both fixed here without touching the on-disk snapshot format: - flushSave encodes the whole snapshot (~2.7GB at 218k nodes) into a sync.Pool'd bytes.Buffer that grows with node count and is retained forever. putSaveBuf now drops buffers larger than maxPooledSaveBuf (128MB) instead of returning them to the pool, so the giant buffer is GC-eligible after each save. - With GOMEMLIMIT=48GiB the runtime holds freed pages rather than returning them, so RSS ratchets toward the ceiling and eventually GC-thrashes. After a large snapshot, shouldScavenge triggers an async debug.FreeOSMemory at most once per 3 minutes to return the freed pages to the OS. Neither change touches the snapshot bytes, checksum, WAL, or load path — the worst case is an extra allocation or GC, never data corruption. Co-Authored-By: Claude Opus 4.8 --- server_persist.go | 33 ++++++++++++++++++++++++++++++--- zz_save_memory_test.go | 42 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 3 deletions(-) create mode 100644 zz_save_memory_test.go diff --git a/server_persist.go b/server_persist.go index 64f2282..61b9f5b 100644 --- a/server_persist.go +++ b/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,7 +541,7 @@ 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+`"}`)...) @@ -525,7 +549,7 @@ func (s *Server) flushSave() (retErr error) { // 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 { diff --git a/zz_save_memory_test.go b/zz_save_memory_test.go new file mode 100644 index 0000000..3e85d16 --- /dev/null +++ b/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") + } +}