From abe6780ec74118a3e3e250bd6fe79c33d8b4cdcf Mon Sep 17 00:00:00 2001
From: smcio <shaun@darkwaves.io>
Date: Fri, 16 Oct 2026 11:48:24 +0000
Subject: [PATCH] perf: bound replay memory and improve account-storage reclaim

---
 cmd/mithril/node/memory.go                  | 112 +++++++
 cmd/mithril/node/memory_test.go             |  70 +++++
 cmd/mithril/node/node.go                    |  47 +--
 cmd/mithril/node/storage_compaction.go      |  37 +++
 cmd/mithril/node/storage_compaction_test.go |  43 +++
 cmd/mithril/node/storage_space.go           |  61 ++++
 cmd/mithril/node/storage_space_test.go      |  54 ++++
 config.example.toml                         |  32 +-
 pkg/accounts/working_set.go                 |  72 ++++-
 pkg/accounts/working_set_memory_test.go     |  54 ++++
 pkg/accountsdb/accountsdb.go                |  21 +-
 pkg/accountsdb/compact.go                   | 262 +++++++++++++----
 pkg/accountsdb/disk_space.go                |  24 ++
 pkg/accountsdb/disk_space_test.go           |  42 +++
 pkg/accountsdb/fold.go                      |  16 +
 pkg/accountsdb/rewind.go                    |   6 +-
 pkg/accountsdb/segment.go                   |   2 +-
 pkg/accountsdb/sparse_compact.go            | 152 ++++++++++
 pkg/accountsdb/sparse_compact_test.go       | 306 ++++++++++++++++++++
 pkg/accountsdb/sparse_linux.go              |  26 ++
 pkg/accountsdb/sparse_other.go              |  24 ++
 pkg/replay/async_promotion_test.go          |  61 +++-
 pkg/replay/block.go                         |  82 ++++--
 pkg/replay/promotion.go                     |  89 ++++--
 pkg/replay/promotion_memory_test.go         | 198 +++++++++++++
 25 files changed, 1741 insertions(+), 152 deletions(-)
 create mode 100644 cmd/mithril/node/memory.go
 create mode 100644 cmd/mithril/node/memory_test.go
 create mode 100644 cmd/mithril/node/storage_compaction.go
 create mode 100644 cmd/mithril/node/storage_compaction_test.go
 create mode 100644 cmd/mithril/node/storage_space.go
 create mode 100644 cmd/mithril/node/storage_space_test.go
 create mode 100644 pkg/accounts/working_set_memory_test.go
 create mode 100644 pkg/accountsdb/disk_space.go
 create mode 100644 pkg/accountsdb/disk_space_test.go
 create mode 100644 pkg/accountsdb/sparse_compact.go
 create mode 100644 pkg/accountsdb/sparse_compact_test.go
 create mode 100644 pkg/accountsdb/sparse_linux.go
 create mode 100644 pkg/accountsdb/sparse_other.go
 create mode 100644 pkg/replay/promotion_memory_test.go

diff --git a/cmd/mithril/node/memory.go b/cmd/mithril/node/memory.go
new file mode 100644
index 000000000..a1efe6baf
--- /dev/null
+++ b/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)
+}
diff --git a/cmd/mithril/node/memory_test.go b/cmd/mithril/node/memory_test.go
new file mode 100644
index 000000000..447646b6e
--- /dev/null
+++ b/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))
+	}
+}
diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go
index eab065b68..332dc0e17 100644
--- a/cmd/mithril/node/node.go
+++ b/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 {
diff --git a/cmd/mithril/node/storage_compaction.go b/cmd/mithril/node/storage_compaction.go
new file mode 100644
index 000000000..b33450bf6
--- /dev/null
+++ b/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 }
+}
diff --git a/cmd/mithril/node/storage_compaction_test.go b/cmd/mithril/node/storage_compaction_test.go
new file mode 100644
index 000000000..5e87c1e3a
--- /dev/null
+++ b/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")
+	}
+}
diff --git a/cmd/mithril/node/storage_space.go b/cmd/mithril/node/storage_space.go
new file mode 100644
index 000000000..699eb0edf
--- /dev/null
+++ b/cmd/mithril/node/storage_space.go
@@ -0,0 +1,61 @@
+package node
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/mlog"
+)
+
+var storageMinFreeBytes uint64
+
+func configureDiskBudget() error {
+	value, err := memoryMiB("storage.min_free_mb", 32<<30, false)
+	if err != nil {
+		return err
+	}
+	storageMinFreeBytes = value
+	return nil
+}
+
+func checkStorageSpace(paths []string, reserve uint64, available func(string) (uint64, error)) error {
+	for _, path := range paths {
+		if path == "" {
+			continue
+		}
+		free, err := available(path)
+		if err != nil {
+			return err
+		}
+		if free < reserve {
+			return fmt.Errorf("storage space guard: %s has %d MiB available, below %d MiB reserve; stopping replay to preserve room for vote history and recovery", path, free>>20, reserve>>20)
+		}
+	}
+	return nil
+}
+
+// Check all writing volumes independently of fold progress (including while
+// waiting for shreds). Cancellation uses the normal replay shutdown path.
+func startStorageSpaceGuard(parent context.Context, check func() error, cancelRun context.CancelFunc) func() {
+	ctx, cancel := context.WithCancel(parent)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
+		for {
+			if err := check(); err != nil {
+				mlog.Log.Errorf("%v", err)
+				cancelRun()
+				return
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+	return func() { cancel(); <-done }
+}
diff --git a/cmd/mithril/node/storage_space_test.go b/cmd/mithril/node/storage_space_test.go
new file mode 100644
index 000000000..5c165abde
--- /dev/null
+++ b/cmd/mithril/node/storage_space_test.go
@@ -0,0 +1,54 @@
+package node
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStorageGuardChecksEveryWritingVolume(t *testing.T) {
+	paths := []string{"accounts", "shredstore", "logs"}
+	for _, low := range paths {
+		err := checkStorageSpace(paths, 32<<30, func(path string) (uint64, error) {
+			if path == low {
+				return 31 << 30, nil
+			}
+			return 100 << 30, nil
+		})
+		require.ErrorContains(t, err, low)
+	}
+	require.NoError(t, checkStorageSpace(paths, 32<<30, func(string) (uint64, error) { return 32 << 30, nil }))
+	require.Error(t, checkStorageSpace(paths, 32<<30, func(string) (uint64, error) { return 0, errors.New("stat failed") }))
+}
+
+func TestStorageGuardCancelsRunAndJoins(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	stop := startStorageSpaceGuard(ctx, func() error { return errors.New("low disk") }, cancel)
+	defer stop()
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("storage guard did not stop replay")
+	}
+	stop()
+}
+
+func TestConfigureDiskBudget(t *testing.T) {
+	old := storageMinFreeBytes
+	t.Cleanup(func() { viper.Reset(); storageMinFreeBytes = old })
+	viper.Reset()
+	require.NoError(t, configureDiskBudget())
+	require.Equal(t, uint64(32<<30), storageMinFreeBytes)
+	viper.Set("storage.min_free_mb", 65536)
+	require.NoError(t, configureDiskBudget())
+	require.Equal(t, uint64(64<<30), storageMinFreeBytes)
+	for _, invalid := range []int64{0, -1, 1 << 43} {
+		viper.Set("storage.min_free_mb", invalid)
+		require.Error(t, configureDiskBudget())
+	}
+}
diff --git a/config.example.toml b/config.example.toml
index 4211c3871..c6212a866 100644
--- a/config.example.toml
+++ b/config.example.toml
@@ -101,6 +101,19 @@ name = "mithril"
     # longer bounded re-execution window after a hard crash. Range 32..512.
     fold_batch_slots = 128
 
+    # Fold sooner when retained account versions reach this byte target (MiB).
+    # Slot versions are indivisible, so the final slot may overshoot the target.
+    fold_max_mb = 4096
+    # Pause replay at slot boundaries to drain safe folds at this retained-state
+    # budget. Halt if finality/verification prevents progress. Includes account
+    # backing-array capacities and estimated bookkeeping, not the entire heap.
+    # Leave RAM headroom for the executing slot, caches, runtime and GC.
+    working_set_max_mb = 16384
+    # Stop admitting folds before they consume this free-space reserve (MiB).
+    # Also monitor accounts, shredstore and log volumes during replay. Leaves
+    # room for vote history, in-flight I/O and graceful recovery/shutdown writes.
+    min_free_mb = 32768
+
     # Account-index Pebble WAL. Set false to remove the WAL write stream
     # entirely: fold manifests then serve as the index redo log and recovery
     # replays them. Keep true until the deployment has soaked.
@@ -112,14 +125,16 @@ name = "mithril"
     # re-bootstrapping from a snapshot. Also pins those batches' files against
     # compaction. At K=128 slots/batch, 64 batches =~ 55 minutes of chain.
     rewind_horizon_batches = 64
+    # Byte-triggered smaller folds shorten the corresponding slot/time span.
 
     # Background compaction: reclaims dead bytes from out-of-horizon segments
     # and bootstrap appendvecs (folds never overwrite, so superseded account
-    # versions accumulate). OFF by default until soaked — each cycle holds the
-    # store's fold lock while it works, so on a latency-sensitive validator keep
-    # the per-cycle budgets small (defaults below) to avoid stalling folds.
+    # versions accumulate). Enabled by default in rooted-durable mode. Large
+    # segments release obsolete payload pages while retaining exact live/undo
+    # records and their offsets. Cycles have bounded scan/file/time budgets.
     #[storage.compact]
-    #    enabled = false
+    #    enabled = true
+    #    interval_seconds = 1
     #    # Move a file's live records only once at least this fraction is dead.
     #    min_dead_fraction = 0.7
     #    # Per-cycle caps on live bytes moved / bytes scanned (MiB). Smaller =
@@ -503,6 +518,10 @@ name = "mithril"
 # The defaults work well for most deployments.
 
 [tuning]
+    # Soft Go runtime memory limit in MiB. 0 preserves GOMEMLIMIT/runtime default.
+    # For a dedicated 128 GiB host, 81920 leaves room for the OS and native memory.
+    # This is not a hard RSS ceiling and cannot reclaim live retained accounts.
+    memory_limit_mb = 0
     # Transaction parallelism for block execution. Validator mode defaults to
     # 2x CPU cores when this key and --txpar are both unset. Verifying mode
     # stays sequential by default; explicit 0 always requests sequential mode.
@@ -596,6 +615,11 @@ name = "mithril"
         # Port to serve HTTP pprof endpoint (-1 to disable)
         # port = 6060
 
+        # Capture a heap profile when Go runtime memory exceeds this MiB value.
+        # 0 disables capture. At most once/minute, rotating three files in the
+        # run log directory; no forced GC. Runtime memory is logged every 10s.
+        heap_profile_threshold_mb = 0
+
         # Filename to write CPU profile (for offline analysis with go tool pprof)
         # cpu_profile_path = "/mnt/mithril-data/profiling/cpu.pprof"
 
diff --git a/pkg/accounts/working_set.go b/pkg/accounts/working_set.go
index 083a4405d..51b855bfa 100644
--- a/pkg/accounts/working_set.go
+++ b/pkg/accounts/working_set.go
@@ -13,10 +13,11 @@ import (
 // writes) instead of a full rescan. Siblings are never materialized as state:
 // a switch unwinds this suffix and re-executes the certified block.
 type WorkingSet struct {
-	mu     sync.RWMutex
-	bySlot map[uint64]*slotLayer
-	order  []uint64               // held slots, ascending
-	flat   map[[32]byte]flatEntry // newest unrooted value per key
+	mu            sync.RWMutex
+	bySlot        map[uint64]*slotLayer
+	order         []uint64               // held slots, ascending
+	flat          map[[32]byte]flatEntry // newest unrooted value per key
+	retainedBytes uint64                 // account capacities plus estimated per-version bookkeeping
 }
 
 type flatEntry struct {
@@ -25,14 +26,25 @@ type flatEntry struct {
 }
 
 type slotLayer struct {
-	slot   uint64
-	writes map[[32]byte]*Account
+	slot          uint64
+	writes        map[[32]byte]*Account
+	retainedBytes uint64
 	// undo records, one per FIRST write of a key in this slot: what the flat
 	// entry pointed at before this slot overwrote it. Applying a suffix's
 	// undos newest-layer-first restores the flat map exactly.
 	undo []undoPtr
 }
 
+// Include account structs, map entries and undo records as well as the backing
+// array capacity. This is a conservative budget, not a measurement of the heap;
+// shared account values in different slots are deliberately counted per version.
+const workingSetVersionOverhead = 512
+
+// RetainedAccountBytes estimates the cost of one immutable working-set version.
+func RetainedAccountBytes(a *Account) uint64 {
+	return uint64(cap(a.Data)) + workingSetVersionOverhead
+}
+
 type undoPtr struct {
 	key      [32]byte
 	prevSlot uint64 // meaningful when existed
@@ -66,6 +78,14 @@ func (w *WorkingSet) Add(slot uint64, delta []*Account) {
 			continue
 		}
 		key := [32]byte(a.Key)
+		if previous, exists := layer.writes[key]; exists {
+			cost := RetainedAccountBytes(previous)
+			layer.retainedBytes -= cost
+			w.retainedBytes -= cost
+		}
+		cost := RetainedAccountBytes(a)
+		layer.retainedBytes += cost
+		w.retainedBytes += cost
 		if _, again := layer.writes[key]; !again {
 			// First write of this key in this slot: journal what flat held.
 			if e, exists := w.flat[key]; exists && e.slot != slot {
@@ -90,6 +110,13 @@ func (w *WorkingSet) HeldSlots() int {
 	return len(w.order)
 }
 
+// RetainedBytes includes every buffered slot version, not only newest keys.
+func (w *WorkingSet) RetainedBytes() uint64 {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	return w.retainedBytes
+}
+
 // Lookup returns the newest unrooted value for pubkey (nil, false if none held).
 // No fall-through to durable; the newest held value is the correct pre-root value.
 func (w *WorkingSet) Lookup(pubkey [32]byte) (*Account, bool) {
@@ -134,7 +161,7 @@ func (w *WorkingSet) PromotionPrefix(through uint64) []SlotDelta {
 	w.mu.RLock()
 	defer w.mu.RUnlock()
 
-	return w.promotionChunkLocked(through, len(w.order), true)
+	return w.promotionChunkLocked(through, len(w.order), 0, true)
 }
 
 // PromotionChunk returns at most maxSlots oldest held slots through the caller's
@@ -144,17 +171,30 @@ func (w *WorkingSet) PromotionPrefix(through uint64) []SlotDelta {
 // This only prepares borrowed account pointers; it does not commit, prune, or
 // advance durability. Callers still own finality checks and durable commit order.
 func (w *WorkingSet) PromotionChunk(through uint64, maxSlots int, allowPartial bool) []SlotDelta {
+	return w.PromotionChunkWithByteLimit(through, maxSlots, 0, allowPartial)
+}
+
+// PromotionChunkWithByteLimit admits a shorter chunk once maxBytes is reached.
+// Slots are atomic: the last slot may overshoot the byte target. The caller's
+// verified through bound is always respected, including under memory pressure.
+// Zero maxBytes preserves slot-only batching.
+func (w *WorkingSet) PromotionChunkWithByteLimit(through uint64, maxSlots int, maxBytes uint64, allowPartial bool) []SlotDelta {
 	w.mu.RLock()
 	defer w.mu.RUnlock()
-	return w.promotionChunkLocked(through, maxSlots, allowPartial)
+	return w.promotionChunkLocked(through, maxSlots, maxBytes, allowPartial)
 }
 
-func (w *WorkingSet) promotionChunkLocked(through uint64, maxSlots int, allowPartial bool) []SlotDelta {
+func (w *WorkingSet) promotionChunkLocked(through uint64, maxSlots int, maxBytes uint64, allowPartial bool) []SlotDelta {
 	count := 0
+	var bytes uint64
 	for count < len(w.order) && count < maxSlots && w.order[count] <= through {
+		bytes += w.bySlot[w.order[count]].retainedBytes
 		count++
+		if maxBytes > 0 && bytes >= maxBytes {
+			break
+		}
 	}
-	if count == 0 || (!allowPartial && count < maxSlots) {
+	if count == 0 || (!allowPartial && count < maxSlots && (maxBytes == 0 || bytes < maxBytes)) {
 		return nil
 	}
 	batch := make([]SlotDelta, count)
@@ -187,6 +227,7 @@ func (w *WorkingSet) PromotePrefix(through uint64) {
 				delete(w.flat, key) // no surviving held writer -> durable
 			}
 		}
+		w.retainedBytes -= w.bySlot[slot].retainedBytes
 		delete(w.bySlot, slot)
 	}
 	w.order = kept
@@ -245,6 +286,7 @@ func (w *WorkingSet) EvictFrom(slot uint64) {
 		}
 	}
 	for _, s := range suffix {
+		w.retainedBytes -= w.bySlot[s].retainedBytes
 		delete(w.bySlot, s)
 	}
 	w.order = w.order[:cut]
@@ -256,10 +298,20 @@ func (w *WorkingSet) CheckInvariants() error {
 	defer w.mu.RUnlock()
 
 	want := make(map[[32]byte]flatEntry)
+	var retainedBytes uint64
 	for _, slot := range w.order {
+		var layerBytes uint64
 		for key, acct := range w.bySlot[slot].writes {
 			want[key] = flatEntry{slot: slot, acct: acct}
+			layerBytes += RetainedAccountBytes(acct)
+		}
+		if layerBytes != w.bySlot[slot].retainedBytes {
+			return errInvariant("slot retained bytes", w.bySlot[slot].retainedBytes, layerBytes)
 		}
+		retainedBytes += layerBytes
+	}
+	if retainedBytes != w.retainedBytes {
+		return errInvariant("retained bytes", w.retainedBytes, retainedBytes)
 	}
 	if len(want) != len(w.flat) {
 		return errInvariant("flat size", len(w.flat), len(want))
diff --git a/pkg/accounts/working_set_memory_test.go b/pkg/accounts/working_set_memory_test.go
new file mode 100644
index 000000000..6b2425902
--- /dev/null
+++ b/pkg/accounts/working_set_memory_test.go
@@ -0,0 +1,54 @@
+package accounts
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkingSetRetainedVersionsReleaseOnPromotionAndUnwind(t *testing.T) {
+	w := NewWorkingSet()
+	a := wsAcct(1, 1)
+	a.Data = make([]byte, 1, 8192) // charge capacity, not just visible length
+	w.Add(1, []*Account{a, a, nil})
+	first := w.RetainedBytes()
+	require.GreaterOrEqual(t, first, uint64(8192))
+	w.Add(1, []*Account{a})
+	require.Equal(t, first, w.RetainedBytes(), "same-slot replacement must not double-charge")
+	b := wsAcct(1, 2)
+	b.Data = make([]byte, 16384)
+	w.Add(2, []*Account{b})
+	require.Greater(t, w.RetainedBytes(), first+16384, "older versions still occupy RAM")
+	require.NoError(t, w.CheckInvariants())
+	w.EvictFrom(2)
+	require.Equal(t, first, w.RetainedBytes())
+	got, ok := w.Lookup(wsKey(1))
+	require.True(t, ok)
+	require.Same(t, a, got)
+	w.Add(2, []*Account{b})
+	w.PromotePrefix(1)
+	require.Equal(t, RetainedAccountBytes(b), w.RetainedBytes())
+	require.NoError(t, w.CheckInvariants())
+	w.PromotePrefix(2)
+	require.Zero(t, w.RetainedBytes())
+	require.NoError(t, w.CheckInvariants())
+}
+
+func TestWorkingSetByteBoundedChunksRespectSafePrefix(t *testing.T) {
+	w := NewWorkingSet()
+	for slot := uint64(1); slot <= 5; slot++ {
+		a := wsAcct(1, slot)
+		a.Data = make([]byte, 8192)
+		w.Add(slot, []*Account{a})
+	}
+	require.Nil(t, w.PromotionChunkWithByteLimit(1, 128, 16384, false), "unverified suffix cannot satisfy the byte target")
+	chunk := w.PromotionChunkWithByteLimit(5, 128, 16384, false)
+	require.Len(t, chunk, 2)
+	require.Equal(t, uint64(2), chunk[1].Slot)
+	partial := w.PromotionChunkWithByteLimit(1, 128, 16384, true)
+	require.Len(t, partial, 1)
+	require.Equal(t, uint64(1), partial[0].Slot)
+	require.Nil(t, w.PromotionChunkWithByteLimit(0, 128, 1, true))
+	require.Len(t, w.PromotionChunkWithByteLimit(5, 128, 1, false), 1, "an oversized atomic slot must make progress")
+	require.Equal(t, 5, w.HeldSlots(), "selection must not mutate retained state")
+}
diff --git a/pkg/accountsdb/accountsdb.go b/pkg/accountsdb/accountsdb.go
index acc1666e5..121eb5aec 100644
--- a/pkg/accountsdb/accountsdb.go
+++ b/pkg/accountsdb/accountsdb.go
@@ -61,6 +61,9 @@ type AccountsDb struct {
 	// in an in-RAM working set (pkg/accounts) and fold to disk via CommitBatch
 	// once finalized+verified. The Alpenglow node forces this on at startup.
 	RootedDurable bool
+	// MinFreeBytes leaves operational headroom before admitting a fold. Set
+	// before replay starts; zero disables the admission check for offline tools.
+	MinFreeBytes uint64
 
 	// IndexWALDisabled runs the Pebble index without a WAL; the fold manifests
 	// are the index redo log (recovery replays the contiguous manifest run
@@ -69,11 +72,19 @@ type AccountsDb struct {
 
 	// Batch-fold state (segment.go/fold.go/recovery.go). foldMu serializes
 	// CommitBatch, recovery, rewind, and compaction.
-	foldMu         sync.Mutex
-	lastBatchSeq   uint64        // guarded by foldMu; seeded by RecoverFoldState
-	durableThrough atomic.Uint64 // observability: highest durably folded slot
-	foldHooks      foldTestHooks // test-only crash injection
-	compactCursor  string        // guarded by foldMu; scan resume point across CompactOnce cycles
+	foldMu           sync.Mutex
+	lastBatchSeq     uint64        // guarded by foldMu; seeded by RecoverFoldState
+	durableThrough   atomic.Uint64 // observability: highest durably folded slot
+	foldHooks        foldTestHooks // test-only crash injection
+	compactCursor    string        // guarded by foldMu; scan resume point across CompactOnce cycles
+	sparseCursor     string        // independent cursor so bootstrap scanning cannot starve large segments
+	resparseNext     bool          // alternate new and previously sparse segments, even with one-file cycles
+	compactScanSeq   uint64
+	compactScanned   map[uint64]struct{} // successful sparse scans at compactScanSeq; reset when the head changes
+	compactScanClock uint64
+	compactLastScan  map[uint64]uint64 // oldest scan first; a growing directory must not postpone cursor wrap forever
+	compactCycles    uint64            // periodically serve bootstrap and mostly-empty sparse files too
+	compactHooks     compactionTestHooks
 
 	// A list of store requests. They are added to the back as they arrive and
 	// removed from the front as they are persisted.
diff --git a/pkg/accountsdb/compact.go b/pkg/accountsdb/compact.go
index 6b167f7ed..a007c9ec3 100644
--- a/pkg/accountsdb/compact.go
+++ b/pkg/accountsdb/compact.go
@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/Overclock-Validator/mithril/pkg/mlog"
 	"github.com/Overclock-Validator/mithril/pkg/util"
@@ -18,15 +19,15 @@ import (
 // overwrite: every re-written account leaves its previous version behind as
 // dead bytes in an older segment or bootstrap appendvec. CompactOnce scans
 // mostly-dead files and moves the still-live records into a fresh output file
-// (1 source -> 1 output), then deletes the source.
+// (1 source -> 1 output), then deletes the source. Large manifest-backed files
+// instead release obsolete payload pages in place; see sparsifyFile.
 //
 // The pin rule (invariant I5) is what keeps rewind sound: a file is untouchable
-// while it is inside the rewind horizon — either it IS an in-horizon fold
-// segment, or an in-horizon fold's undo pointer (ManifestRecord.Prev) names it.
-// Once the horizon ages past a batch, its undo targets stop being needed and
-// the files become ordinary candidates. Bootstrap appendvecs are never pinned
-// by seq (they predate all folds) and compact like any other file once their
-// live fraction decays — that is where the bulk of long-run reclaim comes from.
+// while it is inside the rewind horizon, including the target boundary itself.
+// An older file named by an in-horizon undo pointer cannot be relocated, but
+// sparse reclamation may free other obsolete payloads in it while preserving
+// every referenced record at its exact location. Bootstrap/small files retain
+// the conservative whole-file pin for undo references.
 //
 // Liveness is exact, not heuristic: a record is live iff the index still maps
 // its pubkey to this exact (fileId, offset). FileIds are never reused (I7), so
@@ -49,8 +50,8 @@ import (
 
 // CompactionConfig bounds one CompactOnce cycle.
 type CompactionConfig struct {
-	// RewindHorizonBatches pins the newest N fold batches and every file their
-	// undo pointers name. Must match (or exceed) the operational rewind
+	// RewindHorizonBatches pins the newest N fold batches, their target boundary
+	// and every record their undo pointers name. Must match the operational rewind
 	// horizon. Clamped to >= 1 so the committed head fold is always pinned.
 	RewindHorizonBatches uint64
 	// MinDeadFraction is the dead-byte fraction a file must reach before its
@@ -63,6 +64,11 @@ type CompactionConfig struct {
 	// churn bound; mostly-live files cost a scan but yield no move). Progress
 	// across cycles is kept by a directory cursor. Default 1GB.
 	MaxScanBytesPerCycle int64
+	// MaxFilesPerCycle and MaxDuration bound metadata-only reclamation work.
+	// A single candidate can exceed the duration; it is never interrupted in
+	// the middle of publishing its durable reclamation marker.
+	MaxFilesPerCycle int
+	MaxDuration      time.Duration
 }
 
 // CompactStats reports one CompactOnce cycle.
@@ -70,6 +76,7 @@ type CompactStats struct {
 	CandidatesScanned int
 	FilesCompacted    int
 	FilesDeleted      int // fully-dead fast path (no bytes moved)
+	FilesSparsified   int // dead payload pages released without moving live/undo records
 	LiveBytesMoved    int64
 	BytesReclaimed    int64 // source bytes freed (compacted + deleted)
 }
@@ -81,11 +88,14 @@ const (
 )
 
 type compactCandidate struct {
-	name         string
-	slot         uint64
-	fileId       uint64
-	size         int64
-	manifestPath string // the source's own manifest; "" for bootstrap appendvecs
+	name          string
+	slot          uint64
+	fileId        uint64
+	size          int64
+	allocated     int64
+	manifestPath  string // the source's own manifest; "" for bootstrap appendvecs
+	alreadySparse bool
+	scanOrder     uint64
 }
 
 // CompactOnce runs one bounded compaction cycle and returns what it did.
@@ -106,6 +116,9 @@ func (db *AccountsDb) CompactOnce(cfg CompactionConfig) (CompactStats, error) {
 	if cfg.MaxScanBytesPerCycle <= 0 {
 		cfg.MaxScanBytesPerCycle = defaultMaxScanBytesPerCycle
 	}
+	if cfg.MaxFilesPerCycle <= 0 {
+		cfg.MaxFilesPerCycle = 16
+	}
 
 	db.foldMu.Lock()
 	defer db.foldMu.Unlock()
@@ -114,12 +127,27 @@ func (db *AccountsDb) CompactOnce(cfg CompactionConfig) (CompactStats, error) {
 	if err != nil {
 		return stats, err
 	}
+	if db.compactScanned == nil || db.compactScanSeq != meta.BatchSeq {
+		db.compactScanSeq = meta.BatchSeq
+		db.compactScanned = make(map[uint64]struct{})
+	}
+	if db.compactLastScan == nil {
+		db.compactLastScan = make(map[uint64]uint64)
+	}
 
 	pinned, manifestByFileId, err := db.compactionPinSet(meta.BatchSeq, cfg.RewindHorizonBatches)
 	if err != nil {
 		return stats, err
 	}
 
+	// Retiring a fold manifest (even a fully-dead file) requires its index
+	// frontier to survive a crash. WAL-off folds otherwise rely on that redo log.
+	if db.IndexWALDisabled {
+		if err := db.Index.Flush(); err != nil {
+			return stats, err
+		}
+	}
+
 	bootstrapHigh := db.bootstrapHighFileId()
 	entries, err := os.ReadDir(db.AcctsDir)
 	if err != nil {
@@ -135,10 +163,13 @@ func (db *AccountsDb) CompactOnce(cfg CompactionConfig) (CompactStats, error) {
 		if !ok {
 			continue
 		}
-		if _, isPinned := pinned[fileId]; isPinned {
+		if _, isPinned := pinned.whole[fileId]; isPinned {
 			continue
 		}
-		mpath, hasManifest := manifestByFileId[fileId]
+		if _, scanned := db.compactScanned[fileId]; scanned {
+			continue // no liveness or rewind-pin change since the successful scan
+		}
+		manifest, hasManifest := manifestByFileId[fileId]
 		if !hasManifest && fileId > bootstrapHigh {
 			continue // undecided orphan — recovery's to delete, not ours
 		}
@@ -146,34 +177,106 @@ func (db *AccountsDb) CompactOnce(cfg CompactionConfig) (CompactStats, error) {
 		if ierr != nil {
 			continue
 		}
+		allocated := allocatedBytesFromInfo(info)
+		if allocated < sparseCompactionMinSize {
+			delete(db.compactLastScan, fileId)
+		}
+		// Small files and bootstrap appendvecs still use relocation compaction.
+		// Only manifest-backed large files support reclaiming individual dead
+		// payloads while keeping all undo pointers at their original offsets.
+		if len(pinned.records[fileId]) > 0 && (!hasManifest || info.Size() < sparseCompactionMinSize || !sparseCompactionSupported) {
+			continue
+		}
 		candidates = append(candidates, compactCandidate{
-			name: name, slot: slot, fileId: fileId, size: info.Size(), manifestPath: mpath,
+			name: name, slot: slot, fileId: fileId, size: info.Size(), allocated: allocated, manifestPath: manifest.Path,
+			alreadySparse: (manifest.Kind == ManifestKindCompact && manifest.DataCRC == 0) || db.compactLastScan[fileId] != 0,
+			scanOrder:     db.compactLastScan[fileId],
 		})
 	}
+	// Prefer large manifest-backed segments over the bootstrap directory. They
+	// can grow by gigabytes per fold while bootstrap files remain unchanged.
 	// Deterministic order + resume after the previous cycle's cursor so large
 	// directories make steady progress instead of rescanning the same prefix.
-	sort.Slice(candidates, func(i, j int) bool { return candidates[i].name < candidates[j].name })
-	if db.compactCursor != "" {
-		rotated := make([]compactCandidate, 0, len(candidates))
-		var before []compactCandidate
-		for _, c := range candidates {
-			if c.name > db.compactCursor {
-				rotated = append(rotated, c)
-			} else {
-				before = append(before, c)
-			}
+	sort.Slice(candidates, func(i, j int) bool {
+		a, b := candidates[i], candidates[j]
+		ak, bk := compactCandidateKey(a), compactCandidateKey(b)
+		if ak[0] == '1' && bk[0] == '1' && a.scanOrder != b.scanOrder {
+			return a.scanOrder < b.scanOrder
 		}
-		candidates = append(rotated, before...)
+		return ak < bk
+	})
+	freshEnd := sort.Search(len(candidates), func(i int) bool { return compactCandidateKey(candidates[i])[0] >= '1' })
+	sparseEnd := sort.Search(len(candidates), func(i int) bool { return compactCandidateKey(candidates[i])[0] >= '2' })
+	fresh := rotateCompactCandidates(candidates[:freshEnd], db.sparseCursor)
+	// A name cursor can chase newly arriving files forever without wrapping.
+	// Serve the least recently scanned retry first instead. Copy before reusing
+	// candidates' backing array to assemble the interleaved queues below.
+	resparse := append([]compactCandidate(nil), candidates[freshEnd:sparseEnd]...)
+	slow := rotateCompactCandidates(candidates[sparseEnd:], db.compactCursor)
+	// Partial reclamation retains current and undo versions. Those versions
+	// subsequently die too: never put their second pass behind the bootstrap
+	// directory. Alternate the two large-file queues, carrying the turn across
+	// cycle boundaries so a one-file/time-limited cycle cannot starve either.
+	candidates = candidates[:0]
+	next := db.resparseNext
+	for len(fresh) > 0 || len(resparse) > 0 {
+		if len(resparse) > 0 && (next || len(fresh) == 0) {
+			candidates = append(candidates, resparse[0])
+			resparse = resparse[1:]
+			next = false
+		} else {
+			candidates = append(candidates, fresh[0])
+			fresh = fresh[1:]
+			next = true
+		}
+	}
+	db.compactCycles++
+	if db.compactCycles%8 == 0 {
+		// Busy large-file queues must not permanently exclude bootstrap files
+		// or the small residual allocations in mostly-empty sparse files.
+		candidates = append(slow, candidates...)
+	} else {
+		candidates = append(candidates, slow...)
 	}
 
 	var scannedBytes int64
+	started := time.Now()
 	for _, c := range candidates {
-		if stats.LiveBytesMoved >= cfg.MaxMoveBytesPerCycle || scannedBytes >= cfg.MaxScanBytesPerCycle {
+		if stats.LiveBytesMoved >= cfg.MaxMoveBytesPerCycle || scannedBytes >= cfg.MaxScanBytesPerCycle || stats.CandidatesScanned >= cfg.MaxFilesPerCycle ||
+			(cfg.MaxDuration > 0 && stats.CandidatesScanned > 0 && time.Since(started) >= cfg.MaxDuration) {
 			break
 		}
-		db.compactCursor = c.name
-		scannedBytes += c.size
+		switch compactCandidateKey(c)[0] {
+		case '0':
+			db.sparseCursor = compactCandidateKey(c)
+			db.resparseNext = true
+		case '1':
+			db.resparseNext = false
+		default:
+			db.compactCursor = compactCandidateKey(c)
+		}
 		stats.CandidatesScanned++
+		if c.manifestPath != "" && c.size >= sparseCompactionMinSize && sparseCompactionSupported {
+			reclaimed, scanned, deleted, err := db.sparsifyFile(c, pinned.records[c.fileId])
+			scannedBytes += scanned
+			if err != nil {
+				return stats, fmt.Errorf("accountsdb: sparsify %s: %w", c.name, err)
+			}
+			db.compactScanned[c.fileId] = struct{}{}
+			db.compactScanClock++
+			db.compactLastScan[c.fileId] = db.compactScanClock
+			if deleted {
+				delete(db.compactLastScan, c.fileId)
+			}
+			stats.BytesReclaimed += reclaimed
+			if deleted {
+				stats.FilesDeleted++
+			} else if reclaimed > 0 {
+				stats.FilesSparsified++
+			}
+			continue
+		}
+		scannedBytes += c.size
 
 		acted, moved, err := db.compactFile(c, cfg.MinDeadFraction)
 		if err != nil {
@@ -191,28 +294,59 @@ func (db *AccountsDb) CompactOnce(cfg CompactionConfig) (CompactStats, error) {
 			stats.BytesReclaimed += c.size - moved
 		}
 	}
-	if stats.FilesCompacted > 0 || stats.FilesDeleted > 0 {
-		mlog.Log.Infof("accountsdb: compaction cycle — %d scanned, %d compacted, %d deleted, %s live moved, %s reclaimed",
-			stats.CandidatesScanned, stats.FilesCompacted, stats.FilesDeleted,
+	if stats.FilesCompacted > 0 || stats.FilesDeleted > 0 || stats.FilesSparsified > 0 {
+		mlog.Log.Infof("accountsdb: compaction cycle — %d scanned, %d compacted, %d sparsified, %d deleted, %s live moved, %s reclaimed",
+			stats.CandidatesScanned, stats.FilesCompacted, stats.FilesSparsified, stats.FilesDeleted,
 			humanBytes(stats.LiveBytesMoved), humanBytes(stats.BytesReclaimed))
 	}
 	return stats, nil
 }
 
-// compactionPinSet returns the fileIds compaction must not touch (I5) plus a
-// fileId -> manifest-path map for every current (non-parked) manifest.
+func compactCandidateKey(c compactCandidate) string {
+	if c.manifestPath != "" && c.allocated >= sparseCompactionMinSize {
+		if c.alreadySparse {
+			// Hole punching retains record headers and edge pages. Repeatedly
+			// walking a multi-GB logical file with only a few percent allocated
+			// is expensive and seldom releases much. Keep it in the periodic
+			// slow queue; reserve fast retries for substantial payload remains.
+			if c.allocated >= c.size/10 {
+				return "1:" + c.name
+			}
+			return "2:" + c.name
+		}
+		return "0:" + c.name
+	}
+	return "2:" + c.name
+}
+
+func rotateCompactCandidates(candidates []compactCandidate, cursor string) []compactCandidate {
+	i := sort.Search(len(candidates), func(i int) bool { return compactCandidateKey(candidates[i]) > cursor })
+	rotated := make([]compactCandidate, 0, len(candidates))
+	rotated = append(rotated, candidates[i:]...)
+	return append(rotated, candidates[:i]...)
+}
+
+type compactionPins struct {
+	whole map[uint64]struct{}
+	// fileId -> offset -> expected account key. Undo references retain exact
+	// records, not every obsolete account that happens to share their file.
+	records map[uint64]map[uint64][32]byte
+}
+
+// compactionPinSet returns whole-file and exact undo-record pins (I5) plus a
+// fileId -> header map for every current (non-parked) manifest.
 //
-// Pinned: in-horizon fold segments (BatchSeq > head-horizon), every file an
-// in-horizon undo pointer names, and — unconditionally — parked ".rewound"
-// manifests' segments and undo targets (an interrupted rewind resumes through
-// them at the next startup; compaction must not pull files out from under it).
-func (db *AccountsDb) compactionPinSet(headSeq, horizon uint64) (map[uint64]struct{}, map[uint64]string, error) {
+// Whole-file pins: in-horizon segments including the target (BatchSeq >=
+// head-horizon), and parked ".rewound" segments and undo targets. Other undo
+// targets pin individual records, allowing sparse reclamation without relocation.
+func (db *AccountsDb) compactionPinSet(headSeq, horizon uint64) (*compactionPins, map[uint64]ManifestHeader, error) {
 	horizonFloor := uint64(0)
 	if headSeq > horizon {
 		horizonFloor = headSeq - horizon
 	}
-	pinned := make(map[uint64]struct{})
-	manifestByFileId := make(map[uint64]string)
+	pinned := &compactionPins{whole: make(map[uint64]struct{}), records: make(map[uint64]map[uint64][32]byte)}
+	manifestByFileId := make(map[uint64]ManifestHeader)
+	retainedSeqs := make(map[uint64]bool)
 
 	entries, err := os.ReadDir(db.AcctsDir)
 	if err != nil {
@@ -230,27 +364,57 @@ func (db *AccountsDb) compactionPinSet(headSeq, horizon uint64) (map[uint64]stru
 		path := filepath.Join(db.AcctsDir, name)
 		hdr, herr := readManifestHeader(path)
 		if herr != nil {
-			continue // torn — recovery quarantines; its data file stays untouched (no manifest entry -> orphan rule)
+			return nil, nil, fmt.Errorf("cannot establish rewind pins from %s: %w", name, herr)
+		}
+		dataName := strings.TrimSuffix(strings.TrimSuffix(name, ".rewound"), segManifestSuffix)
+		slot, fileID, validName := parseDataFileName(dataName)
+		if !validName || hdr.FileId != fileID || hdr.ThroughSlot != slot ||
+			(hdr.Kind != ManifestKindFold && hdr.Kind != ManifestKindCompact) ||
+			(hdr.Kind == ManifestKindCompact && hdr.BatchSeq != 0) ||
+			(hdr.Kind == ManifestKindFold && hdr.BatchSeq == 0) {
+			return nil, nil, fmt.Errorf("cannot establish rewind pins: invalid manifest header %s", name)
 		}
 		if !parked {
-			manifestByFileId[hdr.FileId] = path
+			manifestByFileId[hdr.FileId] = hdr
 		}
 		if hdr.Kind != ManifestKindFold {
 			continue
 		}
-		if parked || hdr.BatchSeq > horizonFloor {
-			pinned[hdr.FileId] = struct{}{}
+		// Keep the target boundary itself as well as all undoable batches.
+		if parked || hdr.BatchSeq >= horizonFloor {
+			retainedSeqs[hdr.BatchSeq] = true
+			pinned.whole[hdr.FileId] = struct{}{}
 			m, merr := ReadSegmentManifest(path)
 			if merr != nil {
-				continue // segment itself stays pinned; undo targets unknown but recovery will quarantine this manifest
+				return nil, nil, fmt.Errorf("cannot establish rewind pins from %s: %w", name, merr)
+			}
+			if !parked && hdr.BatchSeq == horizonFloor {
+				continue // boundary's own undo pointers are outside the horizon
 			}
 			for i := range m.Records {
 				if m.Records[i].PrevValid {
-					pinned[m.Records[i].Prev.FileId] = struct{}{}
+					r := m.Records[i]
+					if parked {
+						pinned.whole[r.Prev.FileId] = struct{}{}
+					}
+					if pinned.records[r.Prev.FileId] == nil {
+						pinned.records[r.Prev.FileId] = make(map[uint64][32]byte)
+					}
+					if key, exists := pinned.records[r.Prev.FileId][r.Prev.Offset]; exists && key != r.Pubkey {
+						return nil, nil, fmt.Errorf("conflicting undo pointers for file %d offset %d", r.Prev.FileId, r.Prev.Offset)
+					}
+					pinned.records[r.Prev.FileId][r.Prev.Offset] = r.Pubkey
 				}
 			}
 		}
 	}
+	// Never interpret a missing/corrupted in-horizon fold as an empty undo log.
+	// In particular a damaged header must not silently age itself out of pins.
+	for seq := max(uint64(1), horizonFloor); seq <= headSeq && seq != 0; seq++ {
+		if !retainedSeqs[seq] {
+			return nil, nil, fmt.Errorf("cannot establish rewind pins: missing fold manifest seq %d", seq)
+		}
+	}
 	return pinned, manifestByFileId, nil
 }
 
diff --git a/pkg/accountsdb/disk_space.go b/pkg/accountsdb/disk_space.go
new file mode 100644
index 000000000..95d79f32a
--- /dev/null
+++ b/pkg/accountsdb/disk_space.go
@@ -0,0 +1,24 @@
+package accountsdb
+
+import (
+	"fmt"
+	"syscall"
+)
+
+// AvailableDiskBytes excludes filesystem-reserved blocks, even when running
+// as root. The reserve belongs to recovery, vote history and other writers.
+func AvailableDiskBytes(path string) (uint64, error) {
+	var stat syscall.Statfs_t
+	if err := syscall.Statfs(path, &stat); err != nil {
+		return 0, fmt.Errorf("disk space for %s: %w", path, err)
+	}
+	return stat.Bavail * uint64(stat.Bsize), nil
+}
+
+func checkDiskBudget(available, required, reserve uint64) error {
+	// Subtract only after checking the reserve, to avoid unsigned overflow.
+	if available < reserve || required > available-reserve {
+		return fmt.Errorf("accountsdb: insufficient disk space: available_bytes=%d required_bytes=%d reserve_bytes=%d; reclaim space before resuming", available, required, reserve)
+	}
+	return nil
+}
diff --git a/pkg/accountsdb/disk_space_test.go b/pkg/accountsdb/disk_space_test.go
new file mode 100644
index 000000000..84898f5af
--- /dev/null
+++ b/pkg/accountsdb/disk_space_test.go
@@ -0,0 +1,42 @@
+package accountsdb
+
+import (
+	"os"
+	"testing"
+
+	"github.com/Overclock-Validator/mithril/pkg/accounts"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDiskBudgetBoundaryAndOverflow(t *testing.T) {
+	require.NoError(t, checkDiskBudget(100, 60, 40))
+	require.Error(t, checkDiskBudget(99, 60, 40))
+	require.Error(t, checkDiskBudget(39, 0, 40))
+	require.Error(t, checkDiskBudget(100, ^uint64(0), 40))
+}
+
+func TestLowDiskRejectsFoldBeforeAnyStoreMutation(t *testing.T) {
+	db, dir := newFoldTestDb(t)
+	defer func() { db.CloseDb() }()
+	commitTestBatch(t, db, 110, foldAcct(1, 10, []byte("durable")))
+	before, err := os.ReadDir(db.AcctsDir)
+	require.NoError(t, err)
+	index := snapshotIndex(t, db)
+	id := db.LargestFileId.Load()
+	// Larger than any possible available filesystem capacity.
+	db.MinFreeBytes = ^uint64(0)
+	_, err = db.CommitBatch([]accounts.SlotDelta{{Slot: 120, Delta: []*accounts.Account{foldAcct(1, 11, []byte("rejected"))}}}, 120, nil, nil)
+	require.ErrorContains(t, err, "insufficient disk space")
+	require.Equal(t, id, db.LargestFileId.Load())
+	require.Equal(t, index, snapshotIndex(t, db))
+	after, err := os.ReadDir(db.AcctsDir)
+	require.NoError(t, err)
+	require.Len(t, after, len(before))
+	db = reopenFoldTestDb(t, db, dir)
+	recovered, err := db.RecoverFoldState()
+	require.NoError(t, err)
+	require.Equal(t, uint64(110), recovered.DurableThrough)
+	require.Empty(t, recovered.OrphansRemoved)
+	db.MinFreeBytes = 1
+	commitTestBatch(t, db, 120, foldAcct(1, 11, []byte("resumed")))
+}
diff --git a/pkg/accountsdb/fold.go b/pkg/accountsdb/fold.go
index 096a59a16..d4b8a1f98 100644
--- a/pkg/accountsdb/fold.go
+++ b/pkg/accountsdb/fold.go
@@ -150,6 +150,22 @@ func (db *AccountsDb) CommitBatch(
 	// An all-empty batch (empty blocks) still folds: the manifest records the
 	// batch's bankhashes + resume context and advances the watermark; the data
 	// file is empty and no index entries are written.
+	if db.MinFreeBytes > 0 {
+		// Account for the deduplicated segment, manifest and index writes before
+		// allocating a file ID or writing any data. The reserve covers concurrent
+		// writers and shutdown; this is admission control, not a space reservation.
+		required := uint64(1<<20) + uint64(len(resumeCtx)) + uint64(len(bankhashes))*40
+		for _, v := range union {
+			required += hdrLen + ((uint64(len(v.acct.Data)) + 7) &^ uint64(7)) + manifestRecordSize + 128
+		}
+		available, err := AvailableDiskBytes(db.AcctsDir)
+		if err != nil {
+			return BatchCommitResult{}, err
+		}
+		if err := checkDiskBudget(available, required, db.MinFreeBytes); err != nil {
+			return BatchCommitResult{}, err
+		}
+	}
 
 	// Deterministic segment layout (test reproducibility + stable offsets).
 	keys := make([][32]byte, 0, len(union))
diff --git a/pkg/accountsdb/rewind.go b/pkg/accountsdb/rewind.go
index 7dd5cce10..6ad48b5bc 100644
--- a/pkg/accountsdb/rewind.go
+++ b/pkg/accountsdb/rewind.go
@@ -278,7 +278,11 @@ func (db *AccountsDb) RewindToBatchBoundary(throughSlot uint64) (RewindResult, e
 		return res, err
 	}
 	db.readCacheEpochMu.Lock()
-	commitErr := batch.Commit(pebble.Sync)
+	writeOpts := pebble.Sync
+	if db.IndexWALDisabled {
+		writeOpts = pebble.NoSync // the explicit Flush below provides durability
+	}
+	commitErr := batch.Commit(writeOpts)
 	if commitErr == nil {
 		db.readCacheEpoch++
 		db.resetReadCachesLocked()
diff --git a/pkg/accountsdb/segment.go b/pkg/accountsdb/segment.go
index 9359f7841..c3367df0c 100644
--- a/pkg/accountsdb/segment.go
+++ b/pkg/accountsdb/segment.go
@@ -74,7 +74,7 @@ type SegmentManifest struct {
 	ThroughSlot uint64 // inclusive; equals the data file's name slot
 	FileId      uint64
 	DataLen     uint64 // exact data file length
-	DataCRC     uint32 // crc32 (IEEE) of the data file bytes
+	DataCRC     uint32 // crc32 (IEEE) of data; 0 for compact-kind in-place sparse segments
 	Bankhashes  []SlotBankhash
 	Records     []ManifestRecord
 	ResumeCtx   []byte // opaque serialized state.ResumeContext at ThroughSlot
diff --git a/pkg/accountsdb/sparse_compact.go b/pkg/accountsdb/sparse_compact.go
new file mode 100644
index 000000000..e6ec06e88
--- /dev/null
+++ b/pkg/accountsdb/sparse_compact.go
@@ -0,0 +1,152 @@
+package accountsdb
+
+import (
+	"encoding/binary"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/cockroachdb/pebble"
+)
+
+const sparseCompactionMinSize = int64(8 << 20)
+
+type deadPayload struct{ offset, length int64 }
+
+// sparsifyFile releases only obsolete payload pages in an out-of-horizon
+// segment. File IDs, record headers, offsets, live bytes and undo bytes never
+// change. This prevents one tiny undo record from pinning gigabytes of dead
+// account data (testnet slots 452462795..452482738).
+//
+// Before the first hole is punched, atomically replace the retired fold's
+// manifest with a compact-kind marker. Like other compact manifests it is NOT
+// replayed as an index redo log. DataCRC=0 denotes an in-place sparse segment;
+// its original whole-file checksum no longer describes obsolete payloads.
+// The marker retains all record offsets and has its own normal manifest CRC.
+// A crash before the marker leaves the original fold intact; a crash after it
+// leaves some or all dead pages allocated. Either state is safe and retryable.
+// Caller holds foldMu and has excluded every in-horizon/parked whole-file pin.
+func (db *AccountsDb) sparsifyFile(c compactCandidate, undo map[uint64][32]byte) (reclaimed, scanned int64, deleted bool, err error) {
+	m, err := ReadSegmentManifest(c.manifestPath)
+	if err != nil {
+		return 0, 0, false, err
+	}
+	info, err := os.Stat(c.manifestPath)
+	if err != nil {
+		return 0, 0, false, err
+	}
+	scanned = info.Size()
+	if m.FileId != c.fileId || m.ThroughSlot != c.slot || m.DataLen != uint64(c.size) ||
+		(m.Kind != ManifestKindFold && m.Kind != ManifestKindCompact) {
+		return 0, scanned, false, fmt.Errorf("manifest does not describe source %s", c.name)
+	}
+	path := filepath.Join(db.AcctsDir, c.name)
+	f, err := os.OpenFile(path, os.O_RDWR, 0)
+	if err != nil {
+		return 0, scanned, false, err
+	}
+	defer f.Close()
+	before, err := allocatedFileBytes(f)
+	if err != nil {
+		return 0, scanned, false, err
+	}
+	var dead []deadPayload
+	retained, foundUndo := 0, 0
+	var next uint64
+	for _, r := range m.Records {
+		if r.Offset != next || r.Offset > m.DataLen || m.DataLen-r.Offset < hdrLen {
+			return 0, scanned, false, fmt.Errorf("invalid record offset %d in %s", r.Offset, c.name)
+		}
+		var h [hdrLen]byte
+		if _, err := f.ReadAt(h[:], int64(r.Offset)); err != nil {
+			return 0, scanned, false, err
+		}
+		scanned += hdrLen
+		if [32]byte(h[16:48]) != r.Pubkey {
+			return 0, scanned, false, fmt.Errorf("record key mismatch at offset %d", r.Offset)
+		}
+		length := binary.LittleEndian.Uint64(h[dataLenOffset : dataLenOffset+8])
+		if length > m.DataLen-r.Offset-hdrLen {
+			return 0, scanned, false, fmt.Errorf("record payload exceeds source at offset %d", r.Offset)
+		}
+		next = r.Offset + hdrLen + ((length + 7) &^ uint64(7))
+		keep := false
+		if key, pinned := undo[r.Offset]; pinned {
+			if key != r.Pubkey {
+				return 0, scanned, false, fmt.Errorf("undo key mismatch at offset %d", r.Offset)
+			}
+			foundUndo++
+			keep = true
+		}
+		entry, closer, getErr := db.Index.Get(r.Pubkey[:])
+		if getErr == nil {
+			current, decodeErr := UnmarshalAcctIdxEntryValue(entry)
+			closer.Close()
+			if decodeErr != nil {
+				return 0, scanned, false, decodeErr
+			}
+			if current.FileId == c.fileId {
+				if current.Offset != r.Offset || current.Slot != c.slot {
+					return 0, scanned, false, fmt.Errorf("index location disagrees with source for %x", r.Pubkey)
+				}
+				keep = true
+			}
+		} else if getErr != pebble.ErrNotFound {
+			return 0, scanned, false, getErr
+		}
+		if keep {
+			retained++
+			continue
+		}
+		// Retain headers and edge pages so old appendvec scanners remain valid.
+		// Only complete 4 KiB pages wholly inside dead data may be discarded.
+		start := (int64(r.Offset) + hdrLen + 4095) &^ int64(4095)
+		end := (int64(r.Offset) + hdrLen + int64(length)) &^ int64(4095)
+		if end > start {
+			dead = append(dead, deadPayload{start, end - start})
+		}
+	}
+	if next != m.DataLen || foundUndo != len(undo) {
+		return 0, scanned, false, fmt.Errorf("manifest does not cover source or all undo records")
+	}
+	if retained > 0 && len(dead) == 0 {
+		return 0, scanned, false, nil
+	}
+	if retained == 0 {
+		db.appendVecReadMu.Lock()
+		err := removeSourceFiles(db.AcctsDir, path, c.manifestPath)
+		db.appendVecReadMu.Unlock()
+		return before, scanned, err == nil, err
+	}
+	if m.Kind != ManifestKindCompact || m.DataCRC != 0 {
+		m.Kind, m.BatchSeq, m.DataCRC = ManifestKindCompact, 0, 0
+		if err := WriteSegmentManifest(db.AcctsDir, m); err != nil {
+			return 0, scanned, false, err
+		}
+	}
+	if db.compactHooks.afterSparseMarker != nil {
+		db.compactHooks.afterSparseMarker()
+	}
+	// Drain readers holding older index snapshots before punching any bytes.
+	// foldMu prevents new index generations while the plan is applied.
+	db.appendVecReadMu.Lock()
+	defer db.appendVecReadMu.Unlock()
+	for _, span := range dead {
+		if err := punchDeadPayload(f, span.offset, span.length); err != nil {
+			return 0, scanned, false, err
+		}
+		if db.compactHooks.afterSparsePunch != nil {
+			db.compactHooks.afterSparsePunch()
+		}
+	}
+	if err := f.Sync(); err != nil {
+		return 0, scanned, false, err
+	}
+	after, err := allocatedFileBytes(f)
+	return max(int64(0), before-after), scanned, false, err
+}
+
+type compactionTestHooks struct {
+	afterSparseMarker func()
+	afterSparsePunch  func()
+}
diff --git a/pkg/accountsdb/sparse_compact_test.go b/pkg/accountsdb/sparse_compact_test.go
new file mode 100644
index 000000000..6c4a2cae5
--- /dev/null
+++ b/pkg/accountsdb/sparse_compact_test.go
@@ -0,0 +1,306 @@
+//go:build linux
+
+package accountsdb
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSparseCompactionRetainsExactLiveAndUndoRecords(t *testing.T) {
+	for _, walOff := range []bool{false, true} {
+		for _, crash := range []string{"none", "marker", "punch"} {
+			t.Run(fmt.Sprintf("wal_off=%t/crash=%s", walOff, crash), func(t *testing.T) {
+				previous := DisableIndexWAL
+				DisableIndexWAL = walOff
+				t.Cleanup(func() { DisableIndexWAL = previous })
+				db, dir := newFoldTestDb(t)
+				t.Cleanup(func() { db.CloseDb() })
+				big := bytes.Repeat([]byte{0xa5}, 16<<20)
+				undo := bytes.Repeat([]byte{0x71}, 12<<10)
+				r1 := commitTestBatch(t, db, 110, foldAcct(1, 10, big), foldAcct(2, 20, undo), foldAcct(3, 30, []byte("live")))
+				commitTestBatch(t, db, 120, foldAcct(1, 11, []byte("newer")))
+				commitTestBatch(t, db, 130, foldAcct(1, 12, []byte("boundary")))
+				commitTestBatch(t, db, 140, foldAcct(1, 13, []byte("head")), foldAcct(2, 21, []byte("updated")))
+				indexBefore := snapshotIndex(t, db)
+				source := filepath.Join(db.AcctsDir, SegmentDataName(110, r1.FileId))
+				before, err := os.Stat(source)
+				require.NoError(t, err)
+				if crash == "marker" {
+					db.compactHooks.afterSparseMarker = func() { panic("simulated crash") }
+				} else if crash == "punch" {
+					db.compactHooks.afterSparsePunch = func() { panic("simulated crash") }
+				}
+				cfg := CompactionConfig{RewindHorizonBatches: 1}
+				if crash == "none" {
+					stats, err := db.CompactOnce(cfg)
+					require.NoError(t, err)
+					require.Equal(t, 1, stats.FilesSparsified)
+					require.Greater(t, stats.BytesReclaimed, int64(15<<20))
+				} else {
+					require.Panics(t, func() { _, _ = db.CompactOnce(cfg) })
+				}
+				db = reopenFoldTestDb(t, db, dir)
+				recovered, err := db.RecoverFoldState()
+				require.NoError(t, err)
+				require.Equal(t, uint64(140), recovered.DurableThrough)
+				require.Equal(t, bh(140), recovered.RootedBankhash)
+				_, err = db.CompactOnce(cfg)
+				require.NoError(t, err, "partial reclamation must be retryable")
+				require.Equal(t, indexBefore, snapshotIndex(t, db), "sparse reclamation must not move account locations")
+				after, err := os.Stat(source)
+				require.NoError(t, err)
+				require.Equal(t, before.Size(), after.Size(), "logical offsets and file length remain stable")
+				require.Less(t, allocatedBytesFromInfo(after), allocatedBytesFromInfo(before)-(15<<20))
+				require.Equal(t, []byte("live"), mustColdRead(t, db, 140, solana.PublicKey{3}).Data)
+				require.Equal(t, []byte("head"), mustColdRead(t, db, 140, solana.PublicKey{1}).Data)
+				marker, err := ReadSegmentManifest(segmentManifestPath(db.AcctsDir, 110, r1.FileId))
+				require.NoError(t, err)
+				require.Equal(t, ManifestKindCompact, marker.Kind)
+				require.Zero(t, marker.DataCRC)
+				// Header-only parsing remains compatible with the old appendvec
+				// reader; only obsolete payload bytes can differ.
+				data, err := os.ReadFile(source)
+				require.NoError(t, err)
+				keys, _, _, err := BuildIndexEntriesFromAppendVecs(data, uint64(len(data)), 110, r1.FileId)
+				require.NoError(t, err)
+				require.Len(t, keys, 3)
+				_, err = db.RewindToBatchBoundary(130)
+				require.NoError(t, err)
+				require.Equal(t, undo, mustColdRead(t, db, 130, solana.PublicKey{2}).Data, "undo record sharing the reclaimed file must be byte-exact")
+				require.Equal(t, []byte("boundary"), mustColdRead(t, db, 130, solana.PublicKey{1}).Data)
+			})
+		}
+	}
+}
+
+func TestSparseCompactionRefusesUnknownRewindPins(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	r1 := commitTestBatch(t, db, 110, foldAcct(1, 10, bytes.Repeat([]byte{1}, 16<<20)), foldAcct(2, 20, []byte("old")))
+	commitTestBatch(t, db, 120, foldAcct(1, 11, nil))
+	commitTestBatch(t, db, 130, foldAcct(1, 12, nil))
+	r4 := commitTestBatch(t, db, 140, foldAcct(2, 21, []byte("new")))
+	source := filepath.Join(db.AcctsDir, SegmentDataName(110, r1.FileId))
+	before, err := os.Stat(source)
+	require.NoError(t, err)
+	manifestPath := segmentManifestPath(db.AcctsDir, 140, r4.FileId)
+	m, err := ReadSegmentManifest(manifestPath)
+	require.NoError(t, err)
+	m.Records[0].Prev.Offset++ // valid manifest CRC, invalid undo pointer
+	require.NoError(t, WriteSegmentManifest(db.AcctsDir, m))
+	_, err = db.CompactOnce(CompactionConfig{RewindHorizonBatches: 1})
+	require.Error(t, err)
+	after, err := os.Stat(source)
+	require.NoError(t, err)
+	require.Equal(t, allocatedBytesFromInfo(before), allocatedBytesFromInfo(after))
+	// A header that makes a retained fold look retired must not hide its undo
+	// pointers. The missing in-horizon sequence makes reclamation fail closed.
+	m.Records[0].Prev.Offset--
+	m.Kind, m.BatchSeq = ManifestKindCompact, 0
+	require.NoError(t, WriteSegmentManifest(db.AcctsDir, m))
+	_, err = db.CompactOnce(CompactionConfig{RewindHorizonBatches: 1})
+	require.ErrorContains(t, err, "missing fold manifest seq 4")
+	// An unreadable in-horizon manifest must also abort, never guess at pins.
+	require.NoError(t, os.WriteFile(manifestPath, []byte("torn"), 0600))
+	_, err = db.CompactOnce(CompactionConfig{RewindHorizonBatches: 1})
+	require.ErrorContains(t, err, "cannot establish rewind pins")
+}
+
+func TestSparseCompactionWaitsForOlderIndexReaders(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	big := bytes.Repeat([]byte{0x5a}, 16<<20)
+	r1 := commitTestBatch(t, db, 110, foldAcct(1, 10, big), foldAcct(2, 20, []byte("keep")))
+	// Pin a reader from before the overwrite, as GetAccountsBatch does.
+	db.appendVecReadMu.RLock()
+	locked := true
+	defer func() {
+		if locked {
+			db.appendVecReadMu.RUnlock()
+		}
+	}()
+	old := db.Index.NewSnapshot()
+	defer old.Close()
+	commitTestBatch(t, db, 120, foldAcct(1, 11, nil))
+	commitTestBatch(t, db, 130, foldAcct(1, 12, nil))
+	commitTestBatch(t, db, 140, foldAcct(1, 13, nil))
+	marker := make(chan struct{})
+	db.compactHooks.afterSparseMarker = func() { close(marker) }
+	done := make(chan error, 1)
+	go func() { _, err := db.CompactOnce(CompactionConfig{RewindHorizonBatches: 1}); done <- err }()
+	select {
+	case <-marker:
+	case <-time.After(10 * time.Second):
+		t.Fatal("compactor did not reach marker")
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("compaction passed a pinned reader: %v", err)
+	default:
+	}
+	entry, closer, err := old.Get(solana.PublicKey{1}.Bytes())
+	require.NoError(t, err)
+	location, err := UnmarshalAcctIdxEntryValue(entry)
+	closer.Close()
+	require.NoError(t, err)
+	require.Equal(t, r1.FileId, location.FileId)
+	f, err := os.Open(filepath.Join(db.AcctsDir, SegmentDataName(110, r1.FileId)))
+	require.NoError(t, err)
+	read := make([]byte, len(big))
+	_, err = f.ReadAt(read, int64(location.Offset)+hdrLen)
+	f.Close()
+	require.NoError(t, err)
+	require.Equal(t, big, read, "older reader must finish before obsolete pages are freed")
+	db.appendVecReadMu.RUnlock()
+	locked = false
+	require.NoError(t, <-done)
+}
+
+func TestSparseCompactionPrioritizesUnreclaimedSegments(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	big := bytes.Repeat([]byte{0x55}, 16<<20)
+	r1 := commitTestBatch(t, db, 110, foldAcct(1, 10, big), foldAcct(2, 20, big))
+	r2 := commitTestBatch(t, db, 120, foldAcct(1, 11, big), foldAcct(3, 30, []byte("keep")))
+	commitTestBatch(t, db, 130, foldAcct(1, 12, nil))
+	commitTestBatch(t, db, 140, foldAcct(1, 13, nil))
+	cfg := CompactionConfig{RewindHorizonBatches: 1, MaxFilesPerCycle: 1}
+	first, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Equal(t, 1, first.FilesSparsified)
+	retained, err := os.Stat(filepath.Join(db.AcctsDir, SegmentDataName(110, r1.FileId)))
+	require.NoError(t, err)
+	require.Greater(t, allocatedBytesFromInfo(retained), sparseCompactionMinSize)
+	// Even after restart resets the cursor, a partly reclaimed segment must
+	// not consume the next cycle ahead of newly eligible high-volume data.
+	db.sparseCursor = ""
+	second, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Equal(t, 1, second.FilesSparsified)
+	marker, err := ReadSegmentManifest(segmentManifestPath(db.AcctsDir, 120, r2.FileId))
+	require.NoError(t, err)
+	require.Equal(t, ManifestKindCompact, marker.Kind)
+	require.Equal(t, big, mustColdRead(t, db, 140, solana.PublicKey{2}).Data)
+}
+
+func TestSparseCompactionRevisitsExpiredUndoAheadOfNewSegments(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	big := bytes.Repeat([]byte{0x55}, 16<<20)
+	r1 := commitTestBatch(t, db, 110, foldAcct(1, 10, big), foldAcct(2, 20, big), foldAcct(3, 30, []byte("live")))
+	commitTestBatch(t, db, 120, foldAcct(1, 11, nil))
+	commitTestBatch(t, db, 130, foldAcct(1, 12, nil))
+	commitTestBatch(t, db, 140, foldAcct(2, 21, nil), foldAcct(9, 90, big))
+	cfg := CompactionConfig{RewindHorizonBatches: 1, MaxFilesPerCycle: 1}
+	first, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Greater(t, first.BytesReclaimed, int64(15<<20))
+	source := filepath.Join(db.AcctsDir, SegmentDataName(110, r1.FileId))
+	pinned, err := os.Stat(source)
+	require.NoError(t, err)
+	require.Greater(t, allocatedBytesFromInfo(pinned), int64(16<<20), "undo version is still pinned")
+
+	// New large segments arrive continuously, while the old undo record ages
+	// out. Even with one-file cycles, its second pass must get a turn.
+	commitTestBatch(t, db, 150, foldAcct(9, 91, big))
+	commitTestBatch(t, db, 160, foldAcct(9, 92, big))
+	commitTestBatch(t, db, 170, foldAcct(9, 93, big))
+	second, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Greater(t, second.BytesReclaimed, int64(15<<20))
+	after, err := os.Stat(source)
+	require.NoError(t, err)
+	require.Less(t, allocatedBytesFromInfo(after), int64(1<<20), "expired undo payload must not be stranded behind fresh/bootstrap files")
+	require.Equal(t, []byte("live"), mustColdRead(t, db, 170, solana.PublicKey{3}).Data)
+	third, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Equal(t, 1, third.FilesSparsified, "newly eligible segments also get a turn")
+	require.Greater(t, third.BytesReclaimed, int64(15<<20))
+	_, err = db.RewindToBatchBoundary(160)
+	require.NoError(t, err)
+	require.Equal(t, big, mustColdRead(t, db, 160, solana.PublicKey{9}).Data)
+	require.Equal(t, uint64(92), mustColdRead(t, db, 160, solana.PublicKey{9}).Lamports)
+}
+
+func TestSparseCompactionSkipsUnchangedHeadAndRechecksAfterFold(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	big := bytes.Repeat([]byte{0x55}, 16<<20)
+	commitTestBatch(t, db, 110, foldAcct(1, 10, big), foldAcct(2, 20, big))
+	commitTestBatch(t, db, 120, foldAcct(1, 11, nil))
+	commitTestBatch(t, db, 130, foldAcct(1, 12, nil))
+	cfg := CompactionConfig{RewindHorizonBatches: 1, MaxFilesPerCycle: 1}
+	first, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Equal(t, 1, first.FilesSparsified)
+	second, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.Zero(t, second.CandidatesScanned, "a retained large live account does not require repeated idle scans")
+	commitTestBatch(t, db, 140, foldAcct(2, 21, nil))
+	commitTestBatch(t, db, 150, foldAcct(1, 13, nil))
+	_, err = db.CompactOnce(cfg)
+	require.NoError(t, err)
+	commitTestBatch(t, db, 160, foldAcct(1, 14, nil))
+	third, err := db.CompactOnce(cfg)
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, third.FilesDeleted, 1)
+}
+
+func TestSparseCompactionRecurringChurnStaysBounded(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	big := bytes.Repeat([]byte{0x37}, 8<<20)
+	cfg := CompactionConfig{RewindHorizonBatches: 2, MaxFilesPerCycle: 1}
+	for step := uint64(1); step <= 64; step++ {
+		// One hot account and four less frequently updated accounts leave
+		// partial files whose live/undo versions expire on later cycles.
+		commitTestBatch(t, db, 100+step, foldAcct(1, step, big), foldAcct(byte(2+step%4), step, big))
+		for range 3 {
+			_, err := db.CompactOnce(cfg)
+			require.NoError(t, err)
+		}
+		var allocated int64
+		entries, err := os.ReadDir(db.AcctsDir)
+		require.NoError(t, err)
+		for _, e := range entries {
+			if !isDataFileName(e.Name()) {
+				continue
+			}
+			info, err := e.Info()
+			require.NoError(t, err)
+			allocated += allocatedBytesFromInfo(info)
+		}
+		require.Less(t, allocated, int64(256<<20), "retained disk bytes must stay bounded through repeated pin expiry (step %d)", step)
+	}
+	_, err := db.RewindToBatchBoundary(162)
+	require.NoError(t, err)
+	require.Equal(t, uint64(62), mustColdRead(t, db, 162, solana.PublicKey{1}).Lamports)
+	require.Equal(t, big, mustColdRead(t, db, 162, solana.PublicKey{1}).Data)
+}
+
+func TestSparseCompactionAlsoServicesSmallFilesDuringLargeChurn(t *testing.T) {
+	db, _ := newFoldTestDb(t)
+	defer db.CloseDb()
+	old := commitTestBatch(t, db, 110, foldAcct(200, 10, []byte("obsolete")))
+	big := bytes.Repeat([]byte{0x37}, 8<<20)
+	commitTestBatch(t, db, 120, foldAcct(200, 11, nil), foldAcct(1, 1, big))
+	commitTestBatch(t, db, 130, foldAcct(1, 2, big))
+	commitTestBatch(t, db, 140, foldAcct(1, 3, big))
+	source := filepath.Join(db.AcctsDir, SegmentDataName(110, old.FileId))
+	for step := uint64(1); step <= 8; step++ {
+		commitTestBatch(t, db, 140+step, foldAcct(1, 3+step, big))
+		_, err := db.CompactOnce(CompactionConfig{RewindHorizonBatches: 1, MaxFilesPerCycle: 1})
+		require.NoError(t, err)
+	}
+	_, err := os.Stat(source)
+	require.True(t, os.IsNotExist(err), "continuous large segments must not starve the small-file queue")
+	require.Equal(t, big, mustColdRead(t, db, 148, solana.PublicKey{1}).Data)
+}
diff --git a/pkg/accountsdb/sparse_linux.go b/pkg/accountsdb/sparse_linux.go
new file mode 100644
index 000000000..879b79b93
--- /dev/null
+++ b/pkg/accountsdb/sparse_linux.go
@@ -0,0 +1,26 @@
+//go:build linux
+
+package accountsdb
+
+import (
+	"os"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+const sparseCompactionSupported = true
+
+func allocatedBytesFromInfo(info os.FileInfo) int64 { return info.Sys().(*syscall.Stat_t).Blocks * 512 }
+
+func allocatedFileBytes(f *os.File) (int64, error) {
+	var stat unix.Stat_t
+	if err := unix.Fstat(int(f.Fd()), &stat); err != nil {
+		return 0, err
+	}
+	return stat.Blocks * 512, nil
+}
+
+func punchDeadPayload(f *os.File, offset, length int64) error {
+	return unix.Fallocate(int(f.Fd()), unix.FALLOC_FL_PUNCH_HOLE|unix.FALLOC_FL_KEEP_SIZE, offset, length)
+}
diff --git a/pkg/accountsdb/sparse_other.go b/pkg/accountsdb/sparse_other.go
new file mode 100644
index 000000000..79d79aee1
--- /dev/null
+++ b/pkg/accountsdb/sparse_other.go
@@ -0,0 +1,24 @@
+//go:build !linux
+
+package accountsdb
+
+import (
+	"fmt"
+	"os"
+)
+
+const sparseCompactionSupported = false
+
+func allocatedBytesFromInfo(info os.FileInfo) int64 { return info.Size() }
+
+func allocatedFileBytes(f *os.File) (int64, error) {
+	info, err := f.Stat()
+	if err != nil {
+		return 0, err
+	}
+	return info.Size(), nil
+}
+
+func punchDeadPayload(_ *os.File, _, _ int64) error {
+	return fmt.Errorf("sparse compaction is not supported on this platform")
+}
diff --git a/pkg/replay/async_promotion_test.go b/pkg/replay/async_promotion_test.go
index 2780e425d..2e14cc9fc 100644
--- a/pkg/replay/async_promotion_test.go
+++ b/pkg/replay/async_promotion_test.go
@@ -3,7 +3,9 @@ package replay
 import (
 	"encoding/json"
 	"errors"
+	"os"
 	"sync"
+	"syscall"
 	"testing"
 	"time"
 
@@ -311,11 +313,11 @@ func TestFoldCapacityBackpressureDrainsOneAdmittedFold(t *testing.T) {
 	tail.Add(3, []*accounts.Account{testAccount(3, 3)}, testHashBytes(3))
 	tail.SetContext(3, &state.ResumeContext{Slot: 3})
 	require.True(t, tail.OverCap())
-	stillOver := settleInFlightFoldAtCapacity(tail, p, 3, func(res *foldResult) {
+	stillOver := settleFoldsAtCapacity(tail, p, 3, func(res *foldResult) {
 		if res != nil && res.err == nil {
 			tail.applyFoldJob(res.job)
 		}
-	})
+	}, nil)
 	require.False(t, stillOver)
 	require.False(t, p.inFlight)
 	require.Equal(t, 1, tail.overlay.HeldSlots())
@@ -344,11 +346,11 @@ func TestFoldCapacityBackpressureFailsClosed(t *testing.T) {
 		defer p.stop()
 		p.enqueue(job)
 
-		stillOver := settleInFlightFoldAtCapacity(tail, p, 3, func(res *foldResult) {
+		stillOver := settleFoldsAtCapacity(tail, p, 3, func(res *foldResult) {
 			if res != nil && res.err == nil {
 				tail.applyFoldJob(res.job)
 			}
-		})
+		}, nil)
 		require.True(t, stillOver, "failed fold must leave the cap halt armed")
 		require.False(t, p.inFlight)
 		require.Equal(t, 3, tail.overlay.HeldSlots())
@@ -365,7 +367,7 @@ func TestFoldCapacityBackpressureFailsClosed(t *testing.T) {
 		p := newAsyncPromoter(fc)
 		defer p.stop()
 
-		require.True(t, settleInFlightFoldAtCapacity(tail, p, 3, nil))
+		require.True(t, settleFoldsAtCapacity(tail, p, 3, nil, nil))
 		require.Empty(t, fc.committed, "capacity handler admitted new fold work")
 	})
 
@@ -383,19 +385,19 @@ func TestFoldCapacityBackpressureFailsClosed(t *testing.T) {
 		p.enqueue(job)
 		tail.Add(3, nil, testHashBytes(3)) // deliberately omit SetContext(3)
 
-		require.True(t, settleInFlightFoldAtCapacity(tail, p, 3, func(res *foldResult) {
+		require.True(t, settleFoldsAtCapacity(tail, p, 3, func(res *foldResult) {
 			if res != nil && res.err == nil {
 				tail.applyFoldJob(res.job)
 			}
-		}))
+		}, nil))
 		require.True(t, p.inFlight, "context-less tip must be rejected before draining hides it")
 		require.Equal(t, 3, tail.overlay.HeldSlots())
 	})
 }
 
-// A failed fold leaves the tail untouched (natural retry: the same chunk
-// rebuilds because nothing advanced) and reports the error via the result.
-func TestAsyncPromoterFailedFoldRetries(t *testing.T) {
+// A failed durable fold leaves the tail untouched and refuses further writes
+// until restart recovery has reconciled any already-published manifest.
+func TestAsyncPromoterFailedFoldRequiresRecovery(t *testing.T) {
 	fc := &fakeCommitter{durable: accounts.NewMemAccounts(), failOn: 5}
 	tail := asyncTestTail(fc, 5, 6, 7)
 	p := newAsyncPromoter(fc)
@@ -409,16 +411,17 @@ func TestAsyncPromoterFailedFoldRetries(t *testing.T) {
 	require.Error(t, res.err)
 	assert.Equal(t, 3, tail.overlay.HeldSlots(), "failed fold must not touch the overlay")
 
-	// Retry after the fault clears: identical chunk rebuilds and succeeds.
+	// Even if the apparent fault clears, in-process retry is unsafe: the
+	// failed call could have durably published a manifest before returning.
 	fc.failOn = 0
 	job2, err := tail.buildFoldJob(7, false)
 	require.NoError(t, err)
 	require.Equal(t, job.through, job2.through, "same chunk rebuilds after a failed fold")
-	p.enqueue(job2)
-	res = p.drain()
-	require.NoError(t, res.err)
-	tail.applyFoldJob(res.job)
-	assert.Equal(t, 1, tail.overlay.HeldSlots())
+	require.ErrorIs(t, p.enqueue(job2), res.err)
+	require.Nil(t, p.drain())
+	require.False(t, p.inFlight)
+	require.Empty(t, fc.committed)
+	assert.Equal(t, 3, tail.overlay.HeldSlots())
 }
 
 // Codex item 5 — the shutdown gate proof: the flush path folds THROUGH THE
@@ -467,3 +470,29 @@ func BenchmarkBuildFoldJobWaitingForBatch(b *testing.B) {
 		}
 	}
 }
+
+func TestAsyncPromoterDiskFullLatchedThroughPoll(t *testing.T) {
+	fc := &fakeCommitter{durable: accounts.NewMemAccounts()}
+	tail := asyncTestTail(fc, 5, 6)
+	writes := 0
+	require.NoError(t, tail.SetTransactionStatusCheckpointHooks(TransactionStatusCheckpointHooks{
+		Capture: func(uint64) (TransactionStatusSnapshot, error) { return testCheckpointBytes([]byte("captured")), nil },
+		Install: func(uint64, []byte) (*state.TransactionStatusCheckpointRef, error) {
+			writes++
+			return nil, &os.PathError{Op: "write", Path: "checkpoint.tmp", Err: syscall.ENOSPC}
+		},
+	}))
+	job, err := tail.buildFoldJob(6, false)
+	require.NoError(t, err)
+	p := newAsyncPromoter(fc)
+	defer p.stop()
+	require.NoError(t, p.enqueue(job))
+	var result *foldResult
+	require.Eventually(t, func() bool { result = p.poll(); return result != nil }, time.Second, time.Millisecond)
+	require.ErrorIs(t, result.err, syscall.ENOSPC)
+	require.ErrorIs(t, p.enqueue(job), syscall.ENOSPC, "a full disk must not trigger another write")
+	require.False(t, p.inFlight)
+	require.Equal(t, 1, writes)
+	require.Empty(t, fc.throughs)
+	require.Equal(t, 2, tail.overlay.HeldSlots())
+}
diff --git a/pkg/replay/block.go b/pkg/replay/block.go
index 9b7bf6510..ad510c4a3 100644
--- a/pkg/replay/block.go
+++ b/pkg/replay/block.go
@@ -1644,18 +1644,16 @@ func (t *persistedTracker) Get() (uint64, []byte) {
 	return slot, out
 }
 
-// settleInFlightFoldAtCapacity turns the async fold worker into bounded
-// backpressure only when the speculative tail crosses its hard cap. A fold
-// already in flight was admitted through every finality/verification gate, so
-// waiting for and applying it is safe. We deliberately do not enqueue work
-// here: no in-flight fold at capacity means the normal loop could not admit
-// one (or its previous fold failed), which remains a fail-closed halt.
+// settleFoldsAtCapacity stops replay while durable writes relieve pressure.
+// admit must use the normal finality/verification gate; no ungated partial fold
+// is allowed here. With no safe progress available, the caller halts closed.
 // It returns whether the tail is still over capacity after settlement.
-func settleInFlightFoldAtCapacity(
+func settleFoldsAtCapacity(
 	tail *unrootedTail,
 	promoter *asyncPromoter,
 	tipSlot uint64,
 	applyFoldOutcome func(*foldResult),
+	admit func() bool, // true means a safety or storage failure requires a halt
 ) bool {
 	if tail == nil || !tail.OverCap() {
 		return false
@@ -1667,11 +1665,21 @@ func settleInFlightFoldAtCapacity(
 	if tail.contexts[tipSlot] == nil {
 		return true
 	}
-	if promoter != nil && promoter.inFlight {
+	for tail.OverCap() {
+		before := tail.overlay.HeldSlots()
+		if admit != nil && admit() {
+			return true
+		}
+		if promoter == nil || !promoter.inFlight {
+			break
+		}
 		res := promoter.drain()
 		if applyFoldOutcome != nil {
 			applyFoldOutcome(res)
 		}
+		if promoter.failure != nil || tail.overlay.HeldSlots() >= before {
+			break
+		}
 	}
 	return tail.OverCap()
 }
@@ -2043,7 +2051,7 @@ func ReplayBlocks(
 			}
 			promoter.stop()
 		}()
-		mlog.Log.FileOnlyf("rooted-durable mode: canonical store stays rooted-only; replayed slots buffer in RAM until rooted (halt cap %d slots); folds run async off the replay loop", unrootedTailHaltCap)
+		mlog.Log.Infof("rooted-durable: fold targets %d slots / %d MiB; retained-state budget %d MiB / %d slots", FoldBatchSlots, FoldMaxBytes>>20, WorkingSetMaxBytes>>20, unrootedTailHaltCap)
 	}
 	// Any stake-index entries still pending from a previous in-process replay
 	// attempt (rooted-checkpoint re-replay after a fork switch or finality
@@ -2130,14 +2138,18 @@ func ReplayBlocks(
 	}
 
 	// applyFoldOutcome applies a completed async fold on the loop thread. A
-	// failed fold only logs: LastRootedSlot did not advance, so the next
-	// iteration rebuilds the same chunk (natural retry); a permanently broken
-	// store surfaces as the OverCap halt (fail-closed).
+	// failed durable write stops replay. Its manifest may already have committed,
+	// so restart recovery must reconcile it before any retry. In particular an
+	// ENOSPC must not retry every slot until vote-history persistence also fails.
 	applyFoldOutcome = func(res *foldResult) {
 		if res == nil {
 			return
 		}
 		if res.err != nil {
+			promoter.failure = res.err
+			if result.Error == nil {
+				result.Error = fmt.Errorf("rooted-durable: storage write failed (last confirmed durable slot %d): %w; restart recovery required", mithrilState.LastRootedSlot, res.err)
+			}
 			mlog.Log.Errorf("rooted-durable: async fold failed: %v", res.err)
 			return
 		}
@@ -2164,6 +2176,9 @@ func ReplayBlocks(
 		if unrootedTailState == nil {
 			return false
 		}
+		if promoter.failure != nil {
+			return true
+		}
 		if safety, ok := consensusEngine.(consensusengine.AlpenglowSafetyStatus); ok {
 			if err := safety.AlpenglowSafetyError(); err != nil {
 				if result.Error == nil {
@@ -2176,6 +2191,9 @@ func ReplayBlocks(
 		// Apply any completed async fold first so the gates below see the
 		// current durable frontier.
 		applyFoldOutcome(promoter.poll())
+		if promoter.failure != nil {
+			return true
+		}
 		if lastRootedWatermark == 0 {
 			return false
 		}
@@ -2252,6 +2270,9 @@ func ReplayBlocks(
 			if res := promoter.drain(); res != nil {
 				applyFoldOutcome(res)
 			}
+			if promoter.failure != nil {
+				return true
+			}
 			if rewardsHoldBelowSlot > 0 && partitionedRewardsInfo != nil && promoteThrough >= rewardsHoldBelowSlot {
 				job, jerr := unrootedTailState.buildRewardsCompletionFoldJob(rewardsCompletion.slot)
 				if jerr != nil {
@@ -2259,19 +2280,19 @@ func ReplayBlocks(
 					return false
 				}
 				if err := runFoldJob(unrootedTailState.committer, job); err != nil {
-					mlog.Log.Errorf("rooted-durable: rewards completion fold: %v", err)
-					return false
+					applyFoldOutcome(&foldResult{job: job, err: err})
+					return true
 				}
 				applyFoldOutcome(&foldResult{job: job})
 			}
 			promotedThrough, rootedCtx, perr := unrootedTailState.flush(promoteThrough)
 			if perr != nil {
-				mlog.Log.Errorf("rooted-durable: forced fold stopped at slot %d: %v", promotedThrough, perr)
+				applyFoldOutcome(&foldResult{err: perr})
 			}
 			if promotedThrough > mithrilState.LastRootedSlot && rootedCtx != nil {
 				applyPromotionBookkeeping(promotedThrough, rootedCtx)
 			}
-			return false
+			return perr != nil
 		}
 		// Async: one chunk in flight at a time. Enqueue the next chunk only
 		// when idle; completions are applied at the top of this function on a
@@ -2289,7 +2310,10 @@ func ReplayBlocks(
 				return false
 			}
 			if job != nil {
-				promoter.enqueue(job)
+				if err := promoter.enqueue(job); err != nil {
+					applyFoldOutcome(&foldResult{job: job, err: err})
+					return true
+				}
 			}
 		}
 		return false
@@ -2475,6 +2499,9 @@ func ReplayBlocks(
 		// durable frontier past the switch slot, mutating source ancestry first
 		// would leave the source on a branch AccountsDB can no longer adopt.
 		applyFoldOutcome(promoter.drain())
+		if promoter.failure != nil {
+			return false
+		}
 		if sw.Slot <= mithrilState.LastRootedSlot {
 			windowSwitchFallback++
 			switchFallbackReasons["durable-overlap"]++
@@ -2718,6 +2745,9 @@ func ReplayBlocks(
 				// afterwards, and halted with no retained rewind boundary.
 				if promoter != nil {
 					applyFoldOutcome(promoter.drain())
+					if promoter.failure != nil {
+						break
+					}
 				}
 				if parentSwitch.SwitchSlot <= mithrilState.LastRootedSlot {
 					windowSwitches++
@@ -3366,12 +3396,17 @@ func ReplayBlocks(
 
 		// Rooted-durable backpressure runs only after this slot is complete and its
 		// resume context is attached. If the async writer is merely slower than
-		// replay, wait for its already-admitted fold and apply it before deciding
-		// the tail is truly stalled. No new fold is admitted here: a still-full
-		// tail remains a fail-closed error.
+		// replay, wait for safe folds before admitting any more account data. Each
+		// additional fold must pass the same gates as normal promotion.
 		if unrootedTailState != nil && unrootedTailState.OverCap() {
 			hadInFlightFold := promoter != nil && promoter.inFlight
-			stillOverCap := settleInFlightFoldAtCapacity(unrootedTailState, promoter, block.Slot, applyFoldOutcome)
+			pressureStart := time.Now()
+			mlog.Log.Infof("rooted-durable: memory/slot backpressure at slot %d (held_slots=%d retained_bytes=%d)", block.Slot, unrootedTailState.overlay.HeldSlots(), unrootedTailState.overlay.RetainedBytes())
+			stillOverCap := settleFoldsAtCapacity(unrootedTailState, promoter, block.Slot, applyFoldOutcome, func() bool { return foldRootedPrefix(false) })
+			mlog.Log.Infof("rooted-durable: backpressure settled in %s (held_slots=%d retained_bytes=%d)", time.Since(pressureStart).Round(time.Millisecond), unrootedTailState.overlay.HeldSlots(), unrootedTailState.overlay.RetainedBytes())
+			if result.Error != nil || promoter.failure != nil {
+				break
+			}
 			if stillOverCap {
 				// Diagnose WHY rooting stalled. Validator mode verifies the footer
 				// bank hash inline and has no trailing-verifier watermark; do not
@@ -3390,7 +3425,7 @@ func ReplayBlocks(
 				if trailingVerifier != nil && TrailingVerifierCfg.Required && lastRootedWatermark > verifiedWM+uint64(FoldBatchSlots) {
 					hint = " — the trailing verifier cannot keep pace with replay (finality is ahead; the verifier fetches every block's execution metas via RPC, which is slow for very large blocks). Options: verifier.required=false to gate folds on certificate finality only, a faster/archival RPC for the verifier, or wait for peer bankhash cross-checking to replace the RPC oracle"
 				}
-				result.Error = fmt.Errorf("rooted-durable: speculative state exceeded %d held slots at slot %d; rooting stalled (%s)%s; halting", unrootedTailHaltCap, block.Slot, diag, hint)
+				result.Error = fmt.Errorf("rooted-durable: speculative state reached capacity at slot %d (held_slots=%d/%d retained_bytes=%d/%d); rooting stalled (%s)%s; halting", block.Slot, unrootedTailState.overlay.HeldSlots(), unrootedTailHaltCap, unrootedTailState.overlay.RetainedBytes(), unrootedTailState.maxBytes, diag, hint)
 				mlog.Log.Errorf("%v", result.Error)
 				break
 			}
@@ -3601,6 +3636,9 @@ func ReplayBlocks(
 				mlog.Log.FileOnlyf("memory detail: alloc %.1fGiB | inuse %.1fGiB | idle %.1fGiB | released %.1fGiB | next_gc %.1fGiB | objs %d | gc_total %d | store_queue %d",
 					float64(mem.HeapAlloc)/gib, float64(mem.HeapInuse)/gib, float64(mem.HeapIdle)/gib,
 					float64(mem.HeapReleased)/gib, float64(mem.NextGC)/gib, mem.HeapObjects, mem.NumGC, acctsDb.StoreQueueLen())
+				if unrootedTailState != nil {
+					mlog.Log.FileOnlyf("working set: held_slots=%d retained_bytes=%d budget_bytes=%d fold_in_flight=%t durable=%d", unrootedTailState.overlay.HeldSlots(), unrootedTailState.overlay.RetainedBytes(), unrootedTailState.maxBytes, promoter.inFlight, mithrilState.LastRootedSlot)
+				}
 				if fetchStats.Attempts > 0 {
 					retryRate := float64(fetchStats.Retries) / float64(fetchStats.Attempts) * 100
 					mlog.Log.FileOnlyf("getBlock fetch: %.1f rps (%d calls) | avg %.0fms | %.0f%% success | retries %.1f%% | errs: na:%d rl:%d bt:%d tr:%d",
diff --git a/pkg/replay/promotion.go b/pkg/replay/promotion.go
index 5debf2ffa..2df9b030f 100644
--- a/pkg/replay/promotion.go
+++ b/pkg/replay/promotion.go
@@ -50,6 +50,17 @@ const defaultFoldBatchSlots = 128
 // better union dedupe and less NVMe wear, more RAM, longer crash re-execution.
 var FoldBatchSlots = defaultFoldBatchSlots
 
+const (
+	DefaultFoldMaxBytes       = uint64(4 << 30)
+	DefaultWorkingSetMaxBytes = uint64(16 << 30)
+)
+
+// These budgets count retained account capacities and bookkeeping, not total
+// process RAM. Leave headroom for the executing bank, caches and GC. A slot is
+// indivisible and can overshoot either budget; GOMEMLIMIT is a separate guard.
+var FoldMaxBytes = DefaultFoldMaxBytes
+var WorkingSetMaxBytes = DefaultWorkingSetMaxBytes
+
 // blockAccountSource is the slot-scoped read API the block loader needs; both
 // AccountsDb and unrootedTail satisfy it, so the loader is mode-agnostic.
 type blockAccountSource interface {
@@ -120,12 +131,14 @@ type unrootedState interface {
 // unrootedTail layers an in-RAM UnrootedOverlay over the durable store: reads
 // resolve overlay→durable, commits buffer until rooted slots promote out.
 type unrootedTail struct {
-	overlay     *accounts.WorkingSet
-	durable     blockAccountSource // the canonical rooted store (for read fall-through)
-	committer   batchCommitter     // durable promotion of rooted slot batches
-	bankhashes  map[uint64][32]byte
-	batchSlots  int    // fold chunk size K
-	stakeIdxDir string // directory of stake_pubkeys.idx; pending stake entries flush here at fold time
+	overlay      *accounts.WorkingSet
+	durable      blockAccountSource // the canonical rooted store (for read fall-through)
+	committer    batchCommitter     // durable promotion of rooted slot batches
+	bankhashes   map[uint64][32]byte
+	batchSlots   int    // fold chunk size K
+	foldMaxBytes uint64 // start a fold before K slots when this byte target is met
+	maxBytes     uint64 // stop admitting slots while retained state exceeds this budget
+	stakeIdxDir  string // directory of stake_pubkeys.idx; pending stake entries flush here at fold time
 	// contexts holds the deep-copied end-of-slot resume context per held slot,
 	// retained until promotion so the context as of the last rooted slot survives for resume.
 	contexts map[uint64]*state.ResumeContext
@@ -146,15 +159,17 @@ func newUnrootedTail(durable blockAccountSource, committer batchCommitter, haltC
 		batchSlots = defaultFoldBatchSlots
 	}
 	return &unrootedTail{
-		overlay:     accounts.NewWorkingSet(),
-		durable:     durable,
-		committer:   committer,
-		bankhashes:  make(map[uint64][32]byte),
-		batchSlots:  batchSlots,
-		stakeIdxDir: stakeIdxDir,
-		contexts:    make(map[uint64]*state.ResumeContext),
-		bankSysvars: make(map[uint64]*sealevel.BankSysvars),
-		haltCap:     haltCap,
+		overlay:      accounts.NewWorkingSet(),
+		durable:      durable,
+		committer:    committer,
+		bankhashes:   make(map[uint64][32]byte),
+		batchSlots:   batchSlots,
+		foldMaxBytes: FoldMaxBytes,
+		maxBytes:     WorkingSetMaxBytes,
+		stakeIdxDir:  stakeIdxDir,
+		contexts:     make(map[uint64]*state.ResumeContext),
+		bankSysvars:  make(map[uint64]*sealevel.BankSysvars),
+		haltCap:      haltCap,
 	}
 }
 
@@ -363,15 +378,15 @@ func (t *unrootedTail) promoteChunked(through uint64, force bool) (uint64, *stat
 // - A completed-but-unapplied fold on exit is identical to the supported
 //   "crash after commit, before state-file update" case: RecoverFoldState
 //   reconciles the store frontier forward on the next start.
-// - A failed fold is retried naturally: LastRootedSlot did not advance, so the
-//   next iteration rebuilds the same chunk. The stake-index flush that already
-//   landed is a harmless superset (second flush is a no-op).
+// - A failed durable write halts replay until recovery reconciles its manifest.
 
 // foldJob is an immutable snapshot of one K-slot fold chunk. Account values
 // reference immutable retained WorkingSet layers; the layers are not removed
 // until this job completes and applies.
 type foldJob struct {
 	chunk                                  []accounts.SlotDelta
+	slotCount                              int
+	retainedBytes                          uint64
 	through                                uint64
 	bankhashes                             map[uint64][32]byte
 	ctx                                    *state.ResumeContext
@@ -397,6 +412,7 @@ type foldResult struct {
 func (t *unrootedTail) buildRewardsCompletionFoldJob(through uint64) (*foldJob, error) {
 	whole := *t
 	whole.batchSlots = t.overlay.HeldSlots()
+	whole.foldMaxBytes = 0 // the rewards completion checkpoint must stay atomic
 	job, err := whole.buildFoldJob(through, true)
 	if err != nil {
 		return nil, err
@@ -420,7 +436,7 @@ func (t *unrootedTail) buildFoldJob(through uint64, force bool, hookOverrides ..
 	// Check chunk eligibility before materializing account-write lists. Replay
 	// calls this on every iteration, including skipped slots; a partial batch
 	// remains in RAM without rescanning all of its accounts each time.
-	chunk := t.overlay.PromotionChunk(through, t.batchSlots, force)
+	chunk := t.overlay.PromotionChunkWithByteLimit(through, t.batchSlots, t.foldMaxBytes, force || t.overMemoryLimit())
 	if len(chunk) == 0 {
 		return nil, nil
 	}
@@ -445,13 +461,19 @@ func (t *unrootedTail) buildFoldJob(through uint64, force bool, hookOverrides ..
 		}
 	}
 	bankhashes := make(map[uint64][32]byte, len(chunk))
+	var retainedBytes uint64
 	for _, sd := range chunk {
+		for _, a := range sd.Delta {
+			retainedBytes += accounts.RetainedAccountBytes(a)
+		}
 		if bh, ok := t.bankhashes[sd.Slot]; ok {
 			bankhashes[sd.Slot] = bh
 		}
 	}
 	return &foldJob{
 		chunk:                                  append([]accounts.SlotDelta(nil), chunk...),
+		slotCount:                              len(chunk),
+		retainedBytes:                          retainedBytes,
 		through:                                through,
 		bankhashes:                             bankhashes,
 		ctx:                                    ctx,
@@ -470,9 +492,13 @@ func runFoldJob(committer batchCommitter, job *foldJob) error {
 	if job == nil {
 		return errors.New("fold job has no resume context")
 	}
-	// Failed folds are rebuilt from the retained tail. Neither a failed result
-	// nor a completed-but-unapplied job should keep checkpoint deltas alive.
-	defer func() { job.transactionStatusSnapshot = nil }()
+	// The overlay owns the versions until apply. Do not also retain an entire
+	// completed batch on the idle worker's stack or its result channel.
+	defer func() {
+		job.transactionStatusSnapshot = nil
+		job.chunk = nil
+		job.ctxJSON = nil
+	}()
 	if job.ctx == nil {
 		return errors.New("fold job has no resume context")
 	}
@@ -550,6 +576,7 @@ type asyncPromoter struct {
 	jobs      chan *foldJob
 	results   chan foldResult
 	inFlight  bool
+	failure   error // latched until reopen/recovery; never retry a failed durable write in place
 	done      chan struct{}
 }
 
@@ -570,8 +597,8 @@ func (p *asyncPromoter) run() {
 		start := time.Now()
 		err := runFoldJob(p.committer, job)
 		if err == nil {
-			mlog.Log.FileOnlyf("async fold: committed %d slots through %d in %s checkpoint_capture=%s checkpoint_encode=%s checkpoint_bytes=%d",
-				len(job.chunk), job.through, time.Since(start).Round(time.Millisecond),
+			mlog.Log.FileOnlyf("async fold: committed %d slots through %d in %s retained_bytes=%d checkpoint_capture=%s checkpoint_encode=%s checkpoint_bytes=%d",
+				job.slotCount, job.through, time.Since(start).Round(time.Millisecond), job.retainedBytes,
 				job.checkpointCaptureTime, job.checkpointEncodeTime, job.checkpointBytes)
 		}
 		p.results <- foldResult{job: job, err: err}
@@ -579,9 +606,13 @@ func (p *asyncPromoter) run() {
 }
 
 // enqueue hands a job to the worker (loop thread; requires !inFlight).
-func (p *asyncPromoter) enqueue(job *foldJob) {
+func (p *asyncPromoter) enqueue(job *foldJob) error {
+	if p.failure != nil {
+		return p.failure
+	}
 	p.jobs <- job
 	p.inFlight = true
+	return nil
 }
 
 // poll returns a completed result without blocking (nil when none / none in
@@ -593,6 +624,7 @@ func (p *asyncPromoter) poll() *foldResult {
 	select {
 	case res := <-p.results:
 		p.inFlight = false
+		p.failure = res.err
 		return &res
 	default:
 		return nil
@@ -608,6 +640,7 @@ func (p *asyncPromoter) drain() *foldResult {
 	}
 	res := <-p.results
 	p.inFlight = false
+	p.failure = res.err
 	return &res
 }
 
@@ -667,7 +700,11 @@ func (t *unrootedTail) unwind(fromSlot uint64) (*state.ResumeContext, *sealevel.
 // OverCap reports whether the unrooted tail has grown past the halt cap, i.e.
 // rooting has stalled and we must stop replay rather than grow RAM unbounded.
 func (t *unrootedTail) OverCap() bool {
-	return t.haltCap > 0 && t.overlay.HeldSlots() > t.haltCap
+	return (t.haltCap > 0 && t.overlay.HeldSlots() > t.haltCap) || t.overMemoryLimit()
+}
+
+func (t *unrootedTail) overMemoryLimit() bool {
+	return t.maxBytes > 0 && t.overlay.RetainedBytes() >= t.maxBytes
 }
 
 // promoteRootedBatched folds held slots <= through in chunks of batchSlots.
diff --git a/pkg/replay/promotion_memory_test.go b/pkg/replay/promotion_memory_test.go
new file mode 100644
index 000000000..2847084e1
--- /dev/null
+++ b/pkg/replay/promotion_memory_test.go
@@ -0,0 +1,198 @@
+package replay
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/accounts"
+	"github.com/Overclock-Validator/mithril/pkg/accountsdb"
+	"github.com/Overclock-Validator/mithril/pkg/state"
+	"github.com/stretchr/testify/require"
+)
+
+func TestByteTriggeredFoldsRecoverAndRewind(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, os.Mkdir(filepath.Join(dir, "accounts"), 0o755))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "largest_file_id"), make([]byte, 8), 0o644))
+	db, err := accountsdb.OpenDb(dir)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if db != nil {
+			db.CloseDb()
+		}
+	})
+	db.InitCaches()
+	db.RootedDurable = true
+	tail := newUnrootedTail(db, db, 512, 128, "")
+	tail.foldMaxBytes = 16 << 10
+	for s := uint64(1); s <= 5; s++ {
+		addMemorySlot(tail, s, 8<<10)
+	}
+	for _, through := range []uint64{2, 4} {
+		job, err := tail.buildFoldJob(5, false)
+		require.NoError(t, err)
+		require.Equal(t, through, job.through)
+		require.NoError(t, runFoldJob(db, job))
+		if through == 2 {
+			tail.applyFoldJob(job)
+		}
+		// Deliberately leave the second result unapplied, as at a hard exit.
+	}
+	db.CloseDb()
+	db = nil
+	db, err = accountsdb.OpenDb(dir)
+	require.NoError(t, err)
+	db.InitCaches()
+	db.RootedDurable = true
+	recovered, err := db.RecoverFoldState()
+	require.NoError(t, err)
+	require.Equal(t, uint64(4), recovered.DurableThrough)
+	key := testAccount(1, 1).Key
+	a, err := db.GetAccount(4, key)
+	require.NoError(t, err)
+	require.Equal(t, byte(4), a.Data[0])
+	_, err = db.RewindToBatchBoundary(2)
+	require.NoError(t, err)
+	a, err = db.GetAccount(2, key)
+	require.NoError(t, err)
+	require.Equal(t, byte(2), a.Data[0], "early folds must retain exact undo history")
+}
+
+func addMemorySlot(tail *unrootedTail, slot uint64, bytes int) {
+	a := testAccount(1, slot)
+	a.Data = make([]byte, bytes)
+	a.Data[0] = byte(slot)
+	tail.Add(slot, []*accounts.Account{a}, testHashBytes(byte(slot)))
+	tail.SetContext(slot, &state.ResumeContext{Slot: slot})
+}
+
+func TestFoldMemoryBudgetAndCompletedJobRelease(t *testing.T) {
+	fc := &fakeCommitter{durable: accounts.NewMemAccounts()}
+	tail := newUnrootedTail(&fakeDurable{}, fc, 512, 128, "")
+	tail.foldMaxBytes = 16 << 10
+	for s := uint64(1); s <= 5; s++ {
+		addMemorySlot(tail, s, 8<<10)
+	}
+	job, err := tail.buildFoldJob(1, false)
+	require.NoError(t, err)
+	require.Nil(t, job)
+	job, err = tail.buildFoldJob(5, false)
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), job.through)
+	require.Equal(t, 2, job.slotCount)
+	require.NoError(t, runFoldJob(fc, job))
+	require.Nil(t, job.chunk, "idle worker/result must not pin completed account versions")
+	require.Equal(t, 5, tail.overlay.HeldSlots(), "durability does not prune before loop apply")
+	tail.applyFoldJob(job)
+	require.Equal(t, 3, tail.overlay.HeldSlots())
+	require.NoError(t, tail.overlay.CheckInvariants())
+}
+
+func TestFoldMemoryPressureDrainsMultipleSafeChunks(t *testing.T) {
+	fc := &slowCommitter{fakeCommitter: fakeCommitter{durable: accounts.NewMemAccounts()}, delay: 20 * time.Millisecond}
+	tail := newUnrootedTail(&fakeDurable{}, fc, 512, 128, "")
+	tail.foldMaxBytes, tail.maxBytes = 16<<10, 24<<10
+	for s := uint64(1); s <= 6; s++ {
+		addMemorySlot(tail, s, 8<<10)
+	}
+	p := newAsyncPromoter(fc)
+	defer p.stop()
+	admit := func() bool {
+		if p.inFlight {
+			return false
+		}
+		job, err := tail.buildFoldJob(6, false)
+		require.NoError(t, err)
+		if job != nil {
+			require.NoError(t, p.enqueue(job))
+		}
+		return false
+	}
+	start := time.Now()
+	stillOver := settleFoldsAtCapacity(tail, p, 6, func(res *foldResult) {
+		require.NoError(t, res.err)
+		tail.applyFoldJob(res.job)
+	}, admit)
+	require.False(t, stillOver)
+	require.GreaterOrEqual(t, time.Since(start), 40*time.Millisecond)
+	require.Equal(t, []uint64{1, 2, 3, 4}, fc.committed)
+	require.Equal(t, 2, tail.overlay.HeldSlots())
+	require.Less(t, tail.overlay.RetainedBytes(), tail.maxBytes)
+}
+
+func TestFoldMemoryPressureNeverCrossesVerificationBound(t *testing.T) {
+	fc := &fakeCommitter{durable: accounts.NewMemAccounts()}
+	tail := newUnrootedTail(&fakeDurable{}, fc, 512, 128, "")
+	tail.foldMaxBytes, tail.maxBytes = 16<<10, 24<<10
+	for s := uint64(1); s <= 5; s++ {
+		addMemorySlot(tail, s, 8<<10)
+	}
+	p := newAsyncPromoter(fc)
+	defer p.stop()
+	admit := func() bool {
+		job, err := tail.buildFoldJob(safePromoteTarget(5, true, 1, 0), false)
+		require.NoError(t, err)
+		if job != nil {
+			require.NoError(t, p.enqueue(job))
+		}
+		return false
+	}
+	require.True(t, settleFoldsAtCapacity(tail, p, 5, func(res *foldResult) {
+		require.NoError(t, res.err)
+		tail.applyFoldJob(res.job)
+	}, admit), "must halt if the remaining bytes cannot be safely folded")
+	require.Equal(t, []uint64{1}, fc.committed)
+	require.Equal(t, 4, tail.overlay.HeldSlots())
+}
+
+func TestRewardsCompletionIgnoresNormalFoldByteTarget(t *testing.T) {
+	fc := &fakeCommitter{durable: accounts.NewMemAccounts()}
+	tail := newUnrootedTail(&fakeDurable{}, fc, 512, 128, "")
+	tail.foldMaxBytes = 16 << 10
+	for s := uint64(1); s <= 5; s++ {
+		addMemorySlot(tail, s, 8<<10)
+	}
+	job, err := tail.buildRewardsCompletionFoldJob(5)
+	require.NoError(t, err)
+	require.Len(t, job.chunk, 5, "never persist a partial rewards distribution")
+}
+
+// Repeated large rewrites of one key used to fill all 128 slot layers before
+// folding. Scale bytes down so CI can exercise the same accumulation pattern.
+func TestFoldMemoryPressureSoak(t *testing.T) {
+	fc := &fakeCommitter{durable: accounts.NewMemAccounts()}
+	tail := newUnrootedTail(&fakeDurable{}, fc, 512, 128, "")
+	tail.foldMaxBytes, tail.maxBytes = 2<<20, 8<<20
+	p := newAsyncPromoter(fc)
+	defer p.stop()
+	apply := func(res *foldResult) {
+		if res != nil {
+			require.NoError(t, res.err)
+			tail.applyFoldJob(res.job)
+		}
+	}
+	var peak uint64
+	for s := uint64(1); s <= 512; s++ {
+		addMemorySlot(tail, s, 256<<10)
+		peak = max(peak, tail.overlay.RetainedBytes())
+		admit := func() bool {
+			apply(p.poll())
+			if !p.inFlight {
+				job, err := tail.buildFoldJob(s, false)
+				require.NoError(t, err)
+				if job != nil {
+					require.NoError(t, p.enqueue(job))
+				}
+			}
+			return false
+		}
+		admit()
+		require.False(t, settleFoldsAtCapacity(tail, p, s, apply, admit))
+		require.Less(t, tail.overlay.RetainedBytes(), tail.maxBytes)
+	}
+	apply(p.drain())
+	require.Less(t, peak, tail.maxBytes+(512<<10))
+	t.Logf("512 rewrites, peak retained bytes %d, budget %d, durable through %d", peak, tail.maxBytes, fc.throughs[len(fc.throughs)-1])
+}