perf: bound replay memory and improve account-storage reclaim

cmd/mithril/node/memory.go

@@ -0,0 +1,112 @@
+package node
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"runtime"
+	"runtime/debug"
+	"runtime/pprof"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/config"
+	"github.com/Overclock-Validator/mithril/pkg/mlog"
+	"github.com/Overclock-Validator/mithril/pkg/replay"
+)
+
+var nodeMemoryLimit int64
+var heapProfileThreshold uint64
+
+func memoryMiB(key string, fallback uint64, allowZero bool) (uint64, error) {
+	if !config.IsSet(key) {
+		return fallback, nil
+	}
+	value := config.GetInt64(key)
+	if value < 0 || (!allowZero && value == 0) || value > (1<<63-1)>>20 {
+		return 0, fmt.Errorf("%s must be a valid %sMiB count", key, map[bool]string{true: "nonnegative ", false: "positive "}[allowZero])
+	}
+	return uint64(value) << 20, nil
+}
+
+func configureMemoryBudgets() error {
+	fold, err := memoryMiB("storage.fold_max_mb", replay.DefaultFoldMaxBytes, false)
+	if err != nil {
+		return err
+	}
+	retained, err := memoryMiB("storage.working_set_max_mb", replay.DefaultWorkingSetMaxBytes, false)
+	if err != nil {
+		return err
+	}
+	if fold > retained {
+		return fmt.Errorf("storage.fold_max_mb must not exceed storage.working_set_max_mb")
+	}
+	limit, err := memoryMiB("tuning.memory_limit_mb", 0, true)
+	if err != nil {
+		return err
+	}
+	threshold, err := memoryMiB("tuning.pprof.heap_profile_threshold_mb", 0, true)
+	if err != nil {
+		return err
+	}
+	replay.FoldMaxBytes, replay.WorkingSetMaxBytes = fold, retained
+	nodeMemoryLimit, heapProfileThreshold = int64(limit), threshold
+	return nil
+}
+
+// Sample independently of replay, so a slow fold or stalled slot cannot hide
+// memory growth. Profiles are opt-in, rate-limited, and rotate over three files
+// in the run directory. Do not force GC: that would perturb the workload being
+// diagnosed. Heap profiles describe allocations as of the latest completed GC.
+func startMemoryDiagnostics(parent context.Context, dir string, threshold uint64) func() {
+	ctx, cancel := context.WithCancel(parent)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+		var lastProfile time.Time
+		profileIndex := 0
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case now := <-ticker.C:
+				var mem runtime.MemStats
+				runtime.ReadMemStats(&mem)
+				used := mem.Sys - mem.HeapReleased
+				mlog.Log.FileOnlyf("runtime memory: heap_bytes=%d go_bytes=%d next_gc_bytes=%d gc_total=%d memory_limit_bytes=%d", mem.HeapAlloc, used, mem.NextGC, mem.NumGC, debug.SetMemoryLimit(-1))
+				if threshold == 0 || used < threshold || dir == "" || now.Sub(lastProfile) < time.Minute {
+					continue
+				}
+				lastProfile = now // rate-limit failed writes too
+				path := filepath.Join(dir, fmt.Sprintf("heap-pressure-%d.pprof", profileIndex%3))
+				if err := writeHeapProfile(path); err != nil {
+					mlog.Log.Warnf("memory diagnostics: %v", err)
+					continue
+				}
+				profileIndex++
+				mlog.Log.Infof("memory diagnostics: heap profile saved to %s (heap=%d MiB Go=%d MiB)", path, mem.HeapAlloc>>20, used>>20)
+			}
+		}
+	}()
+	return func() { cancel(); <-done }
+}
+
+func writeHeapProfile(path string) error {
+	tmp := path + ".tmp"
+	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
+	if err != nil {
+		return err
+	}
+	defer os.Remove(tmp)
+	err = pprof.WriteHeapProfile(f)
+	closeErr := f.Close()
+	if err != nil {
+		return err
+	}
+	if closeErr != nil {
+		return closeErr
+	}
+	return os.Rename(tmp, path)
+}

cmd/mithril/node/memory_test.go

@@ -0,0 +1,70 @@
+package node
+
+import (
+	"compress/gzip"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/Overclock-Validator/mithril/pkg/replay"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigureMemoryBudgets(t *testing.T) {
+	oldFold, oldWorking, oldLimit, oldProfile := replay.FoldMaxBytes, replay.WorkingSetMaxBytes, nodeMemoryLimit, heapProfileThreshold
+	t.Cleanup(func() {
+		viper.Reset()
+		replay.FoldMaxBytes, replay.WorkingSetMaxBytes, nodeMemoryLimit, heapProfileThreshold = oldFold, oldWorking, oldLimit, oldProfile
+	})
+	viper.Reset()
+	require.NoError(t, configureMemoryBudgets())
+	require.Equal(t, replay.DefaultFoldMaxBytes, replay.FoldMaxBytes)
+	require.Equal(t, replay.DefaultWorkingSetMaxBytes, replay.WorkingSetMaxBytes)
+	require.Zero(t, nodeMemoryLimit, "omitted setting must preserve GOMEMLIMIT")
+	viper.Set("storage.fold_max_mb", 1024)
+	viper.Set("storage.working_set_max_mb", 4096)
+	viper.Set("tuning.memory_limit_mb", 81920)
+	viper.Set("tuning.pprof.heap_profile_threshold_mb", 16384)
+	require.NoError(t, configureMemoryBudgets())
+	require.Equal(t, uint64(1<<30), replay.FoldMaxBytes)
+	require.Equal(t, uint64(4<<30), replay.WorkingSetMaxBytes)
+	require.Equal(t, int64(80<<30), nodeMemoryLimit)
+	require.Equal(t, uint64(16<<30), heapProfileThreshold)
+	for _, tc := range []struct {
+		key   string
+		value int64
+	}{
+		{"storage.fold_max_mb", 0},
+		{"storage.fold_max_mb", 8192},
+		{"storage.working_set_max_mb", -1},
+		{"tuning.memory_limit_mb", -1},
+		{"tuning.memory_limit_mb", 1 << 43},
+		{"tuning.pprof.heap_profile_threshold_mb", -1},
+	} {
+		old := viper.Get(tc.key)
+		viper.Set(tc.key, tc.value)
+		require.Error(t, configureMemoryBudgets(), tc.key)
+		require.Equal(t, int64(80<<30), nodeMemoryLimit, "invalid configuration must not partially apply")
+		viper.Set(tc.key, old)
+	}
+}
+
+func TestHeapProfileIsReadableAndAtomicallyReplaced(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "heap-pressure-0.pprof")
+	for range 2 {
+		require.NoError(t, writeHeapProfile(path))
+		f, err := os.Open(path)
+		require.NoError(t, err)
+		r, err := gzip.NewReader(f)
+		require.NoError(t, err)
+		data, err := io.ReadAll(r)
+		require.NoError(t, err)
+		require.NotEmpty(t, data)
+		require.NoError(t, r.Close())
+		require.NoError(t, f.Close())
+		_, err = os.Stat(path + ".tmp")
+		require.True(t, os.IsNotExist(err))
+	}
+}

cmd/mithril/node/node.go

@@ -1202,7 +1202,10 @@ func initConfigAndBindFlags(cmd *cobra.Command) error {
 		return fmt.Errorf("tuning.common_account_cache_max_mb must be > 0")
 	}
 
-	return nil
+	if err := configureMemoryBudgets(); err != nil {
+		return err
+	}
+	return configureDiskBudget()
 }
 
 // buildSnapshotConfig creates a snapshotdl.SnapshotConfig from the mithril config.
@@ -1365,6 +1368,12 @@ func runLive(c *cobra.Command, args []string) {
 		fmt.Fprintf(os.Stderr, "warning: failed to initialize file logging: %v\n", err)
 	}
 	defer mlog.Shutdown()
+	if nodeMemoryLimit > 0 {
+		debug.SetMemoryLimit(nodeMemoryLimit)
+	}
+	mlog.Log.Infof("Go runtime soft memory limit: %d MiB (tuning.memory_limit_mb overrides GOMEMLIMIT when set)", debug.SetMemoryLimit(-1)>>20)
+	stopMemoryDiagnostics := startMemoryDiagnostics(ctx, mlog.GetLogDir(), heapProfileThreshold)
+	defer stopMemoryDiagnostics()
 
 	// Kill any existing mithril processes to prevent zombie accumulation
 	if killed := killExistingMithrilProcesses(); killed > 0 {
@@ -2357,6 +2366,8 @@ postBootstrap:
 		RewindHorizonBatches: 64,
 		MaxMoveBytesPerCycle: 64 << 20,  // 64 MiB moved per cycle
 		MaxScanBytesPerCycle: 256 << 20, // 256 MiB scanned per cycle
+		MaxFilesPerCycle:     16,
+		MaxDuration:          time.Second,
 	}
 	if v := config.GetInt("storage.rewind_horizon_batches"); v > 0 {
 		compactCfg.RewindHorizonBatches = uint64(v)
@@ -2370,12 +2381,17 @@ postBootstrap:
 	if v := config.GetInt("storage.compact.max_scan_mb"); v > 0 {
 		compactCfg.MaxScanBytesPerCycle = int64(v) << 20
 	}
-	// Background compaction is OFF by default until soaked (it holds the store's
-	// fold lock during each cycle); operators opt in via storage.compact.enabled.
-	compactEnabled := false
+	accountsDb.MinFreeBytes = storageMinFreeBytes
+	// Rooted storage is append-only: reclamation must run during normal replay.
+	// Large segments reclaim dead payload pages without relocating undo records.
+	compactEnabled := alpenglowMode
 	if alpenglowMode && config.IsSet("storage.compact.enabled") {
 		compactEnabled = config.GetBool("storage.compact.enabled")
 	}
+	compactInterval := time.Second
+	if seconds := config.GetInt("storage.compact.interval_seconds"); seconds > 0 {
+		compactInterval = time.Duration(seconds) * time.Second
+	}
 
 	// Crash recovery reconcile: the durable fold frontier (recovered above from
 	// the store's manifests + index meta — the state file is written only on
@@ -2949,30 +2965,25 @@ postBootstrap:
 	// out-of-horizon segments and bootstrap appendvecs. Each cycle is bounded
 	// (move + scan budgets) and serialized against folds via the store's
 	// internal lock; the rewind horizon pins are what it must never touch.
+	stopCompactor := func() {}
 	if compactEnabled {
-		go func() {
-			ticker := time.NewTicker(10 * time.Minute)
-			defer ticker.Stop()
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case <-ticker.C:
-					if _, cerr := accountsDb.CompactOnce(compactCfg); cerr != nil {
-						mlog.Log.Warnf("compaction cycle failed: %v", cerr)
-					}
-				}
-			}
-		}()
+		stopCompactor = startAccountsCompactor(ctx, accountsDb, compactCfg, compactInterval)
+		defer stopCompactor()
 	}
 
 	startSigverifyReporter(ctx)
+	mlog.Log.Infof("storage space guard enabled (reserve=%d MiB; checked before folds and every second)", storageMinFreeBytes>>20)
+	stopSpaceGuard := startStorageSpaceGuard(ctx, func() error {
+		return checkStorageSpace([]string{accountsPath, blockstorePath, mlog.GetLogDir()}, storageMinFreeBytes, accountsdb.AvailableDiskBytes)
+	}, cancelRun)
+	defer stopSpaceGuard()
 
 	turbineAlpenglowAddr := ""
 	if alpenglowMode {
 		turbineAlpenglowAddr = alpenglowAddrForGossip(alpenglowObserverBindAddr)
 	}
 	result := runReplayWithRecovery(ctx, accountsDb, accountsPath, manifest, resumeState, uint64(startSlot), liveEndSlot, rpcEndpoints, lightbringerEndpoint, turbineBindAddr, turbineGossipEntrypoint, turbineGossipBindAddr, turbineAdvertisedIP, uint16(turbineShredVersion), turbineAlpenglowAddr, validatorIdentity, blockstorePath, int(txParallelism), true, useLightbringer, useTurbine, dbgOpts, metricsWriter, slotCtxSetter, mithrilState, blockFetchOpts, consensusOpts, compactCfg.RewindHorizonBatches, replayStartTime)
+	stopCompactor() // join before shutdown closes AccountsDB
 
 	if result.Error != nil {
 		if result.LastPersistedSlot == 0 {

cmd/mithril/node/storage_compaction.go

@@ -0,0 +1,37 @@
+package node
+
+import (
+	"context"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/accountsdb"
+	"github.com/Overclock-Validator/mithril/pkg/mlog"
+)
+
+type accountsCompactor interface {
+	CompactOnce(accountsdb.CompactionConfig) (accountsdb.CompactStats, error)
+}
+
+// stop joins the worker before AccountsDB closes. A timer after each completed
+// cycle avoids queued back-to-back work if storage is temporarily slow.
+func startAccountsCompactor(parent context.Context, db accountsCompactor, cfg accountsdb.CompactionConfig, interval time.Duration) func() {
+	ctx, cancel := context.WithCancel(parent)
+	done := make(chan struct{})
+	mlog.Log.Infof("accountsdb: background compaction enabled (interval=%s rewind_horizon=%d batches)", interval, cfg.RewindHorizonBatches)
+	go func() {
+		defer close(done)
+		for ctx.Err() == nil {
+			if _, err := db.CompactOnce(cfg); err != nil {
+				mlog.Log.Warnf("compaction cycle failed: %v", err)
+			}
+			timer := time.NewTimer(interval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
+		}
+	}()
+	return func() { cancel(); <-done }
+}

cmd/mithril/node/storage_compaction_test.go

@@ -0,0 +1,43 @@
+package node
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/accountsdb"
+)
+
+type blockedCompactor struct{ started, release chan struct{} }
+
+func (c *blockedCompactor) CompactOnce(accountsdb.CompactionConfig) (accountsdb.CompactStats, error) {
+	close(c.started)
+	<-c.release
+	return accountsdb.CompactStats{}, nil
+}
+
+func TestAccountsCompactorShutdownJoinsInFlightCycle(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := &blockedCompactor{started: make(chan struct{}), release: make(chan struct{})}
+	stop := startAccountsCompactor(ctx, c, accountsdb.CompactionConfig{}, time.Hour)
+	select {
+	case <-c.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("first compaction never started")
+	}
+	cancel()
+	done := make(chan struct{})
+	go func() { stop(); close(done) }()
+	select {
+	case <-done:
+		t.Fatal("shutdown returned with compactor still using the DB")
+	case <-time.After(20 * time.Millisecond):
+	}
+	close(c.release)
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("shutdown did not join completed compaction")
+	}
+}