feat(FlatDB): long finality support with persisted snapshots

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.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,14 @@ public class FlatDbConfig : IFlatDbConfig
     public ulong BlockCacheSizeBudget { get; set; } = 1UL.GiB;
     public long CompactionOffset { get; set; } = -1;
     public ulong TrieCacheMemoryBudget { get; set; } = 512UL.MiB;
+    public bool EnableLongFinality { get; set; } = true;
+    public ulong LongFinalityMaxReorgDepth { get; set; } = 90000;
+    public int MaxInMemoryBaseSnapshotCount { get; set; } = 160;
+    public long ArenaFileSizeBytes { get; set; } = 1.GiB;
+    public long PersistedSnapshotDedicatedArenaThresholdBytes { get; set; } = 1.GiB;
+    public long PersistedSnapshotArenaPageCacheBytes { get; set; } = 4.GiB;
+    public bool PersistedSnapshotPunchHoleOnReclaim { get; set; } = true;
+    public ulong 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")]
     ulong MaxReorgDepth { get; set; }
 
     [ConfigItem(Description = "Minimum reorg depth", DefaultValue = "128")]
@@ -55,6 +55,36 @@ 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 = "true")]
+    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")]
+    ulong 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. Sized as a ~128 target plus one CompactSize of headroom, since a bulk (CompactSize-wide) conversion drops the in-memory count by up to CompactSize at a boundary — so the tier still retains ~128 base snapshots after each conversion.", DefaultValue = "160")]
+    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 = "Page-cache budget (bytes) for the persisted-snapshot arena. Backs the PageResidencyTracker that drives madvise(DONTNEED) eviction on mmap'd arena files. 0 disables the tracker.", DefaultValue = "4294967296")]
+    long PersistedSnapshotArenaPageCacheBytes { 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")]
+    ulong 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;
@@ -16,9 +17,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,21 @@ protected override void Load(ContainerBuilder builder)
             .AddSingleton<ICompactionSchedule, CompactionSchedule>()
             .AddSingleton<ISnapshotCompactor, SnapshotCompactor>()
             .AddSingleton<IPersistenceManager, PersistenceManager>()
+            .AddSingleton<IArenaManager, IFlatDbConfig, IInitConfig, ILogManager>((cfg, initConfig, logManager) =>
+            {
+                string basePath = Path.Combine(initConfig.BaseDbPath, "persistedSnapshot");
+                return new ArenaManager(Path.Combine(basePath, "arena"), cfg, logManager);
+            })
+            .AddSingleton<BlobArenaManager, IFlatDbConfig, IInitConfig>((cfg, initConfig) =>
+            {
+                string basePath = Path.Combine(initConfig.BaseDbPath, "persistedSnapshot");
+                return new BlobArenaManager(
+                    Path.Combine(basePath, "blob"),
+                    cfg.ArenaFileSizeBytes);
+            })
+            .AddSingleton<IPersistedSnapshotCompactor, PersistedSnapshotCompactor>()
             .AddSingleton<ISnapshotRepository, SnapshotRepository>()
+            .AddSingleton<IPersistedSnapshotLoader, PersistedSnapshotLoader>()
             .AddSingleton<ITrieWarmer>(flatDbConfig.TrieWarmerWorkerCount == 0
                 ? _ => new NoopTrieWarmer()
                 : ctx => ctx.Resolve<TrieWarmer>())
@@ -72,6 +91,13 @@ protected override void Load(ContainerBuilder builder)
 
             // Persistences
             .AddColumnDatabase<FlatDbColumns>(DbNames.Flat)
+            .AddKeyedSingleton<IDb>(DbNames.PersistedSnapshotCatalog, ctx => ctx
+                .Resolve<IDbFactory>()
+                .CreateDb(new DbSettings(
+                    nameof(DbNames.PersistedSnapshotCatalog),
+                    Path.Combine("persistedSnapshot", "catalog"))))
+            .AddSingleton<SnapshotCatalog>()
+            .AddSingleton<ISnapshotCatalog>(ctx => ctx.Resolve<SnapshotCatalog>())
             .AddSingleton<RocksDbPersistence>()
             .AddSingleton<FlatInTriePersistence>()
             .AddDecorator<IRocksDbConfigFactory, FlatRocksDbConfigAdjuster>()
@@ -100,6 +126,14 @@ protected override void Load(ContainerBuilder builder)
             })
             ;
 
+        if (!flatDbConfig.EnableLongFinality)
+        {
+            builder
+                .AddSingleton<ISnapshotCatalog>(NullSnapshotCatalog.Instance)
+                .AddSingleton<IPersistedSnapshotLoader>(NullPersistedSnapshotLoader.Instance)
+                .AddSingleton<IPersistedSnapshotCompactor>(NullPersistedSnapshotCompactor.Instance);
+        }
+
         if (flatDbConfig.ImportFromPruningTrieState)
         {
             builder