refactor(flat): replace persisted-snapshot HSST with single-level sorted table

src/Nethermind/Nethermind.Benchmark/State/ReadOnlySnapshotBundleBenchmark.cs

@@ -14,6 +14,7 @@
 using Nethermind.Logging;
 using Nethermind.State.Flat;
 using Nethermind.State.Flat.Persistence;
+using Nethermind.State.Flat.PersistedSnapshots;
 using Nethermind.State.Flat.ScopeProvider;
 using Nethermind.Trie;
 using FlatSnapshot = Nethermind.State.Flat.Snapshot;
@@ -73,16 +74,17 @@ public void Setup()
             int storageAccountCount = 20 * multiplier;
             int slotsPerStorageAccount = 100 * multiplier;
 
-            // Build ReadOnlySnapshotBundle from previously captured snapshots
             SnapshotPooledList prevSnapshots = new(allSnapshots.Count);
             foreach (FlatSnapshot s in allSnapshots)
             {
                 s.TryAcquire();
                 prevSnapshots.Add(s);
             }
 
+            // Build ReadOnlySnapshotBundle from previously captured snapshots
             ReadOnlySnapshotBundle readOnly = new(
-                prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false);
+                prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false,
+                PersistedSnapshotStack.Empty());
             NullTrieNodeCache cache = new();
             SnapshotBundle bundle = new(
                 readOnly, cache, resourcePool, ResourcePool.Usage.MainBlockProcessing);
@@ -154,16 +156,17 @@ public void Setup()
                 maxSlotsPerStorageAccount = slotsPerStorageAccount;
         }
 
-        // Build final ReadOnlySnapshotBundle with all 8 snapshots
         SnapshotPooledList finalSnapshots = new(allSnapshots.Count);
         foreach (FlatSnapshot s in allSnapshots)
         {
             s.TryAcquire();
             finalSnapshots.Add(s);
         }
 
+        // Build final ReadOnlySnapshotBundle with all 8 snapshots
         _bundle = new ReadOnlySnapshotBundle(
-            finalSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false);
+            finalSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false,
+            PersistedSnapshotStack.Empty());
 
         // --- Hit arrays ---
         _hitAccounts = new Address[ArraySize];

src/Nethermind/Nethermind.Benchmark/State/WriteBatchBenchmark.cs

@@ -13,6 +13,7 @@
 using Nethermind.Logging;
 using Nethermind.State.Flat;
 using Nethermind.State.Flat.Persistence;
+using Nethermind.State.Flat.PersistedSnapshots;
 using Nethermind.State.Flat.ScopeProvider;
 using Nethermind.Trie;
 using FlatSnapshot = Nethermind.State.Flat.Snapshot;
@@ -65,7 +66,8 @@ public void GlobalSetup()
             }
 
             ReadOnlySnapshotBundle readOnly = new(
-                prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false);
+                prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false,
+                PersistedSnapshotStack.Empty());
             NullTrieNodeCache cache = new();
             SnapshotBundle bundle = new(
                 readOnly, cache, _resourcePool, ResourcePool.Usage.MainBlockProcessing);
@@ -147,7 +149,8 @@ public void IterationSetup()
         }
 
         ReadOnlySnapshotBundle readOnly = new(
-            prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false);
+            prevSnapshots, new NoopPersistenceReader(), recordDetailedMetrics: false,
+            PersistedSnapshotStack.Empty());
         NullTrieNodeCache cache = new();
         SnapshotBundle bundle = new(
             readOnly, cache, _resourcePool, ResourcePool.Usage.MainBlockProcessing);

src/Nethermind/Nethermind.Core/Utils/SmallRefCountingDisposable.cs

@@ -0,0 +1,127 @@
+// SPDX-FileCopyrightText: 2025 Demerzel Solutions Limited
+// SPDX-License-Identifier: LGPL-3.0-only
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
+using System.Threading;
+
+namespace Nethermind.Core.Utils;
+
+/// <summary>
+/// Variant of <see cref="RefCountingDisposable"/> that stores its lease counter inline as a single
+/// <see cref="long"/> instead of a cache-line-padded one, trading false-sharing protection for a much
+/// smaller per-instance footprint. Prefer it for types that exist in large numbers and whose lease
+/// counts are rarely contended across cores.
+/// </summary>
+public abstract class SmallRefCountingDisposable(int initialCount = 1) : IDisposable
+{
+    private const int Single = 1;
+    private const int NoAccessors = 0;
+    private const int Disposing = -1;
+
+    private long _leases = initialCount;
+
+    public void AcquireLease()
+    {
+        if (!TryAcquireLease())
+        {
+            ThrowCouldNotAcquire();
+        }
+
+        [DoesNotReturn]
+        [StackTraceHidden]
+        static void ThrowCouldNotAcquire() => throw new InvalidOperationException("The lease cannot be acquired");
+    }
+
+    protected bool TryAcquireLease()
+    {
+        // Volatile read for starting value
+        long current = Volatile.Read(ref _leases);
+
+        while (true)
+        {
+            // Reject once the count has reached zero (NoAccessors) or gone to Disposing: the object is
+            // being torn down. Acquiring at NoAccessors would resurrect an object whose owner has
+            // already observed the zero count and begun teardown — the release path moves the count
+            // 1 → 0 and only then CASes 0 → Disposing, so a concurrent acquirer can briefly see 0.
+            // Checking inside the loop (not just on the initial read) also closes the window where a
+            // failed CAS hands back a now-zero count.
+            if (current <= NoAccessors)
+            {
+                return false;
+            }
+
+            long prev = Interlocked.CompareExchange(ref _leases, current + Single, current);
+            if (prev == current)
+            {
+                // Successfully acquired
+                return true;
+            }
+
+            // Try again with the observed value
+            current = prev;
+            // Add PAUSE instruction to reduce shared core contention
+            Thread.SpinWait(1);
+        }
+    }
+
+    /// <summary>
+    /// Disposes it once, decreasing the lease count by 1.
+    /// </summary>
+    public void Dispose() => ReleaseLeaseOnce();
+
+    private void ReleaseLeaseOnce()
+    {
+        // Volatile read for starting value
+        long current = Volatile.Read(ref _leases);
+        if (current <= NoAccessors)
+        {
+            // Mismatched Acquire/Release
+            ThrowOverDisposed();
+        }
+
+        while (true)
+        {
+            long prev = Interlocked.CompareExchange(ref _leases, current - Single, current);
+            if (prev != current)
+            {
+                current = prev;
+                // Add PAUSE instruction to reduce shared core contention
+                Thread.SpinWait(1);
+                continue;
+            }
+            if (prev == Single)
+            {
+                // Last use, try to dispose underlying
+                break;
+            }
+            if (prev <= NoAccessors)
+            {
+                // Mismatched Acquire/Release
+                ThrowOverDisposed();
+            }
+
+            // Successfully released
+            return;
+        }
+
+        if (Interlocked.CompareExchange(ref _leases, Disposing, NoAccessors) == NoAccessors)
+        {
+            // set to disposed by this Release
+            CleanUp();
+        }
+
+        [DoesNotReturn]
+        [StackTraceHidden]
+        static void ThrowOverDisposed() => throw new ObjectDisposedException("The lease has already been disposed");
+    }
+
+    protected abstract void CleanUp();
+
+    public override string ToString()
+    {
+        long leases = Volatile.Read(ref _leases);
+        return leases == Disposing ? "Disposed" : $"Leases: {leases}";
+    }
+}

src/Nethermind/Nethermind.Db/DbNames.cs

@@ -23,5 +23,6 @@ public static class DbNames
         public const string PeersDb = "peers";
         public const string LogIndex = "logIndex";
         public const string Preimage = "preimage";
+        public const string PersistedSnapshotCatalog = "persistedSnapshotCatalog";
     }
 }

src/Nethermind/Nethermind.Db/FlatDbConfig.cs

@@ -24,4 +24,13 @@ public class FlatDbConfig : IFlatDbConfig
     public long BlockCacheSizeBudget { get; set; } = 1.GiB;
     public long CompactionOffset { get; set; } = -1;
     public long TrieCacheMemoryBudget { get; set; } = 512.MiB;
+    public bool EnableLongFinality { get; set; } = false;
+    public int LongFinalityMaxReorgDepth { get; set; } = 90000;
+    public int MaxInMemoryBaseSnapshotCount { get; set; } = 128;
+    public long ArenaFileSizeBytes { get; set; } = 1.GiB;
+    public long PersistedSnapshotDedicatedArenaThresholdBytes { get; set; } = 1.GiB;
+    public bool PersistedSnapshotPunchHoleOnReclaim { get; set; } = true;
+    public int PersistedSnapshotMaxCompactSize { get; set; } = 1024 * 1024;
+    public bool ValidatePersistedSnapshot { get; set; } = false;
+    public double PersistedSnapshotBloomBitsPerKey { get; set; } = 14.0;
 }

src/Nethermind/Nethermind.Db/IFlatDbConfig.cs

@@ -34,7 +34,7 @@ public interface IFlatDbConfig : IConfig
     [ConfigItem(Description = "Max in flight compact job", DefaultValue = "32")]
     int MaxInFlightCompactJob { get; set; }
 
-    [ConfigItem(Description = "Max reorg depth", DefaultValue = "256")]
+    [ConfigItem(Description = "Max reorg depth — the force-persist backstop used when EnableLongFinality is off: once the in-memory depth exceeds it while finality is stalled, persistence is forced to bound memory.", DefaultValue = "256")]
     int MaxReorgDepth { get; set; }
 
     [ConfigItem(Description = "Minimum reorg depth", DefaultValue = "128")]
@@ -55,6 +55,33 @@ public interface IFlatDbConfig : IConfig
     [ConfigItem(Description = "Verify with trie", DefaultValue = "false")]
     bool VerifyWithTrie { get; set; }
 
+    [ConfigItem(Description = "Enable long finality support with persisted snapshots", DefaultValue = "false")]
+    bool EnableLongFinality { get; set; }
+
+    [ConfigItem(Description = "Force-persist backstop used when EnableLongFinality is on, in place of MaxReorgDepth. The persisted-snapshot tier serves deep reorgs, so this is much larger than the non-long-finality backstop.", DefaultValue = "90000")]
+    int LongFinalityMaxReorgDepth { get; set; }
+
+    [ConfigItem(Description = "Maximum number of in-memory base snapshots before conversion to the persisted-snapshot tier kicks in. Counted as `SnapshotCount` of the in-memory repository, not a block-distance depth.", DefaultValue = "128")]
+    int MaxInMemoryBaseSnapshotCount { get; set; }
+
+    [ConfigItem(Description = "Maximum size in bytes for a single arena file before a new one is started.", DefaultValue = "1073741824")]
+    long ArenaFileSizeBytes { get; set; }
+
+    [ConfigItem(Description = "Estimated-size threshold (bytes) at or above which a persisted-snapshot arena write goes to its own dedicated file instead of being packed into a shared arena.", DefaultValue = "1073741824")]
+    long PersistedSnapshotDedicatedArenaThresholdBytes { get; set; }
+
+    [ConfigItem(Description = "When reclaiming dead persisted-snapshot arena ranges — metadata reservation cleanup and blob-file frontier reset — call fallocate(FALLOC_FL_PUNCH_HOLE) to free the underlying disk blocks. Linux-only; automatically and permanently disabled per arena pool if the filesystem reports the operation unsupported. Set false to skip hole-punching entirely (the page-cache posix_fadvise still runs).", DefaultValue = "true")]
+    bool PersistedSnapshotPunchHoleOnReclaim { get; set; }
+
+    [ConfigItem(Description = "Max persisted snapshot compaction size (hierarchical compaction ceiling for persisted layer), in blocks", DefaultValue = "1048576")]
+    int PersistedSnapshotMaxCompactSize { get; set; }
+
+    [ConfigItem(Description = "Validate persisted snapshots against in-memory snapshots after conversion (debug/diagnostic only)", DefaultValue = "false")]
+    bool ValidatePersistedSnapshot { get; set; }
+
+    [ConfigItem(Description = "Bits per key for the per-snapshot in-memory bloom filter. One unified filter covers address/slot/self-destruct keys plus state-trie and storage-trie node paths. Higher = lower false-positive rate but more RAM. 0 disables the filter (lookups behave as full sweeps).", DefaultValue = "14.0")]
+    double PersistedSnapshotBloomBitsPerKey { get; set; }
+
     [ConfigItem(Description = "Persistent dedicated reader threads used to resolve hinted BAL read sets into the pre-block cache. -1 for 4x logical processor count capped at 64. Values below 1 are clamped to 1. Use --Blocks.ParallelExecutionBatchRead=false to disable BAL warming entirely.", DefaultValue = "-1")]
     int WarmReadConcurrency { get; set; }
 }

src/Nethermind/Nethermind.Init/Modules/FlatWorldStateModule.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 using System;
+using System.IO;
 using Autofac;
 using Nethermind.Api.Steps;
 using Nethermind.Blockchain;
@@ -17,9 +18,12 @@
 using Nethermind.JsonRpc.Modules.Admin;
 using Nethermind.Logging;
 using Nethermind.Monitoring.Config;
+using Nethermind.Api;
 using Nethermind.State.Flat;
 using Nethermind.State.Flat.Persistence;
+using Nethermind.State.Flat.PersistedSnapshots;
 using Nethermind.State.Flat.ScopeProvider;
+using Nethermind.State.Flat.PersistedSnapshots.Storage;
 using Nethermind.State.Flat.Sync;
 using Nethermind.State.Flat.Sync.Snap;
 
@@ -46,6 +50,7 @@ protected override void Load(ContainerBuilder builder)
                 ctx.Resolve<ISnapshotCompactor>(),
                 ctx.Resolve<ISnapshotRepository>(),
                 ctx.Resolve<IPersistenceManager>(),
+                ctx.Resolve<IPersistedSnapshotLoader>(),
                 ctx.Resolve<IFlatDbConfig>(),
                 ctx.Resolve<IBlocksConfig>(),
                 ctx.Resolve<ILogManager>(),
@@ -55,7 +60,26 @@ protected override void Load(ContainerBuilder builder)
             .AddSingleton<ICompactionSchedule, CompactionSchedule>()
             .AddSingleton<ISnapshotCompactor, SnapshotCompactor>()
             .AddSingleton<IPersistenceManager, PersistenceManager>()
+            // Shared ArenaManager + BlobArenaManager singletons: the persisted-snapshot repo and
+            // the compactor MUST resolve the same instances, otherwise compaction would write
+            // through a different mmap than the repository reads from.
+            .AddSingleton<ArenaManager, IFlatDbConfig, IInitConfig, ILogManager>((cfg, initConfig, logManager) =>
+            {
+                string basePath = Path.Combine(initConfig.BaseDbPath, "persisted_snapshot");
+                return new ArenaManager(Path.Combine(basePath, "arena"), cfg, logManager);
+            })
+            .AddSingleton<IArenaManager>(ctx => ctx.Resolve<ArenaManager>())
+            .AddSingleton<BlobArenaManager, IFlatDbConfig, IInitConfig>((cfg, initConfig) =>
+            {
+                string basePath = Path.Combine(initConfig.BaseDbPath, "persisted_snapshot");
+                return new BlobArenaManager(
+                    Path.Combine(basePath, "blob"),
+                    cfg.ArenaFileSizeBytes);
+            })
+            .AddSingleton<IPersistedSnapshotCompactor, PersistedSnapshotCompactor>()
             .AddSingleton<ISnapshotRepository, SnapshotRepository>()
+            // Registered after ISnapshotRepository so DI disposes it first.
+            .AddSingleton<IPersistedSnapshotLoader, PersistedSnapshotLoader>()
             .AddSingleton<ITrieWarmer>(flatDbConfig.TrieWarmerWorkerCount == 0
                 ? _ => new NoopTrieWarmer()
                 : ctx => ctx.Resolve<TrieWarmer>())
@@ -72,6 +96,17 @@ protected override void Load(ContainerBuilder builder)
 
             // Persistences
             .AddColumnDatabase<FlatDbColumns>(DbNames.Flat)
+            // Persisted snapshot catalog: dedicated RocksDB co-located with the arena/blob files it
+            // indexes under <BaseDbPath>/persisted_snapshot/catalog/. Wiping persisted_snapshot/
+            // therefore wipes the catalog alongside the data.
+            .AddKeyedSingleton<IDb>(DbNames.PersistedSnapshotCatalog, ctx => ctx
+                .Resolve<IDbFactory>()
+                .CreateDb(new DbSettings(
+                    nameof(DbNames.PersistedSnapshotCatalog),
+                    Path.Combine("persisted_snapshot", "catalog"))))
+            .AddSingleton<SnapshotCatalog>(ctx =>
+                new SnapshotCatalog(ctx.ResolveKeyed<IDb>(DbNames.PersistedSnapshotCatalog)))
+            .AddSingleton<ISnapshotCatalog>(ctx => ctx.Resolve<SnapshotCatalog>())
             .AddSingleton<RocksDbPersistence>()
             .AddSingleton<FlatInTriePersistence>()
             .AddDecorator<IRocksDbConfigFactory, FlatRocksDbConfigAdjuster>()
@@ -99,6 +134,20 @@ protected override void Load(ContainerBuilder builder)
             })
             ;
 
+        // EnableLongFinality off: inert the whole persisted tier. The Null loader skips loading any
+        // on-disk tier at startup and never converts in-memory snapshots into it; the Null catalog keeps
+        // it empty (nothing recorded or loaded); the Null compactor runs no background compaction. The
+        // conversion paths in PersistenceManager.DetermineSnapshotAction are also gated on this flag.
+        // SnapshotRepository still constructs its arena/blob/catalog stores under
+        // `<data-dir>/persisted_snapshot/`, but they stay empty and unread.
+        if (!flatDbConfig.EnableLongFinality)
+        {
+            builder
+                .AddSingleton<ISnapshotCatalog>(NullSnapshotCatalog.Instance)
+                .AddSingleton<IPersistedSnapshotLoader>(NullPersistedSnapshotLoader.Instance)
+                .AddSingleton<IPersistedSnapshotCompactor>(NullPersistedSnapshotCompactor.Instance);
+        }
+
         if (flatDbConfig.ImportFromPruningTrieState)
         {
             builder