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") + } +}