fix(flat): isolate warmer misses and safely publish verified reads

src/Nethermind/Nethermind.State.Flat.Test/FlatTestHelpers.cs

@@ -34,7 +34,15 @@ public static SnapshotPooledList SnapshotList(params Snapshot[] snapshots)
     /// optionally pre-populating the snapshot content via <paramref name="populate"/>.
     /// </summary>
     public static ReadOnlySnapshotBundle MakeBundle(ResourcePool pool, Action<SnapshotContent>? populate = null) =>
-        new(SnapshotList(MakeSnapshot(pool, populate)), Substitute.For<IPersistence.IPersistenceReader>(),
+        MakeBundle(pool, Substitute.For<IPersistence.IPersistenceReader>(), populate);
+
+    /// <inheritdoc cref="MakeBundle(ResourcePool, Action{SnapshotContent})"/>
+    /// <param name="reader">Persistence reader to back the bundle with, for tests that assert on its reads.</param>
+    public static ReadOnlySnapshotBundle MakeBundle(
+        ResourcePool pool,
+        IPersistence.IPersistenceReader reader,
+        Action<SnapshotContent>? populate = null) =>
+        new(SnapshotList(MakeSnapshot(pool, populate)), reader,
             recordDetailedMetrics: false, PersistedSnapshotStack.Empty());
 }
 

src/Nethermind/Nethermind.State.Flat/SnapshotBundle.cs

@@ -166,7 +166,8 @@ public TrieNode FindStateNodeOrUnknown(in TreePath path, Hash256 hash)
         {
             Nethermind.Trie.Pruning.Metrics.IncrementLoadedFromCacheNodesCount();
         }
-        else if (_transientResource.TryGetStateNode(path, hash, out node) && !TrieNodeCache.IsPlaceholder(node))
+        else if (_transientResource.TryGetStateNode(path, hash, out node)
+                 && (!node.IsWarmerOwned || node.IsWarmerResolved))
         {
             Nethermind.Trie.Pruning.Metrics.IncrementLoadedFromCacheNodesCount();
         }
@@ -189,7 +190,7 @@ public TrieNode FindStateNodeOrUnknownForTrieWarmer(in TreePath path, Hash256 ha
         TransientResource? transientResource = TryLeaseTransientResource();
         if (transientResource is null)
         {
-            return TryFindStateNodeInPersistence(path, hash, out TrieNode? node) ? node : new TrieNode(NodeType.Unknown, hash);
+            return TryFindStateNodeInPersistence(path, hash, out TrieNode? node) ? node : CreateWarmerUnknownNode(hash);
         }
 
         try
@@ -210,8 +211,16 @@ private TrieNode WarmUpStateNode(TransientResource transientResource, in TreePat
             return node;
         }
 
-        return transientResource.GetOrAddStateNode(path,
-            TryFindStateNodeInPersistence(path, hash, out node) ? node : new TrieNode(NodeType.Unknown, hash));
+        return TryFindStateNodeInPersistence(path, hash, out node)
+            ? transientResource.GetOrAddStateNode(path, node)
+            : transientResource.GetOrAddStateNode(path, CreateWarmerUnknownNode(hash));
+    }
+
+    private static TrieNode CreateWarmerUnknownNode(Hash256 hash)
+    {
+        TrieNode node = new(NodeType.Unknown, hash);
+        node.MarkWarmerOwned();
+        return node;
     }
 
     // Returns a leased transient, or null once the bundle is being torn down. A stale read can acquire a
@@ -288,7 +297,8 @@ public TrieNode FindStorageNodeOrUnknown(Hash256 address, in TreePath path, Hash
         {
             Nethermind.Trie.Pruning.Metrics.IncrementLoadedFromCacheNodesCount();
         }
-        else if (_transientResource.TryGetStorageNode((Hash256AsKey)address, path, hash, out node) && !TrieNodeCache.IsPlaceholder(node))
+        else if (_transientResource.TryGetStorageNode((Hash256AsKey)address, path, hash, out node)
+                 && (!node.IsWarmerOwned || node.IsWarmerResolved))
         {
             Nethermind.Trie.Pruning.Metrics.IncrementLoadedFromCacheNodesCount();
         }
@@ -314,7 +324,7 @@ public TrieNode FindStorageNodeOrUnknownTrieWarmer(Hash256 address, in TreePath
         {
             return TryFindStorageNodeInPersistence(address, path, hash, out TrieNode? node)
                 ? node
-                : new TrieNode(NodeType.Unknown, hash);
+                : CreateWarmerUnknownNode(hash);
         }
 
         try
@@ -335,8 +345,9 @@ private TrieNode WarmUpStorageNode(TransientResource transientResource, Hash256
             return node;
         }
 
-        return transientResource.GetOrAddStorageNode((Hash256AsKey)address, path,
-            TryFindStorageNodeInPersistence(address, path, hash, out node) ? node : new TrieNode(NodeType.Unknown, hash));
+        return TryFindStorageNodeInPersistence(address, path, hash, out node)
+            ? transientResource.GetOrAddStorageNode((Hash256AsKey)address, path, node)
+            : transientResource.GetOrAddStorageNode((Hash256AsKey)address, path, CreateWarmerUnknownNode(hash));
     }
 
     private bool TryFindStorageNodeInPersistence(Hash256 address, in TreePath path, Hash256 hash, [NotNullWhen(true)] out TrieNode? node)

src/Nethermind/Nethermind.State.Flat/TransientResource.cs

@@ -43,6 +43,19 @@ internal void OnRented(IResourcePool pool, ResourcePool.Usage usage)
 
     internal bool TryAcquireLease() => RefCountingLease.TryAcquire(ref _leases);
 
+    /// <summary>
+    /// Waits until this retired resource is held only by its owner, so in-flight warmer reads have drained before
+    /// retirement scans its caches.
+    /// </summary>
+    internal void WaitForExclusiveLease()
+    {
+        SpinWait spinWait = default;
+        while (Volatile.Read(ref _leases) != RefCountingLease.Single)
+        {
+            spinWait.SpinOnce();
+        }
+    }
+
     /// <summary>
     /// Releases one lease; the final release returns the resource to the pool it was checked out from.
     /// </summary>

src/Nethermind/Nethermind.State.Flat/TrieNodeCache.cs

@@ -4,10 +4,12 @@
 using System.Diagnostics.CodeAnalysis;
 using System.Numerics;
 using System.Runtime.CompilerServices;
+using Nethermind.Core.Buffers;
 using Nethermind.Core.Crypto;
 using Nethermind.Db;
 using Nethermind.Logging;
 using Nethermind.Trie;
+using Nethermind.Trie.Pruning;
 
 namespace Nethermind.State.Flat;
 
@@ -98,14 +100,16 @@ public bool TryGet(Hash256? address, in TreePath path, Hash256 hash, [NotNullWhe
 
     public void Add(TransientResource transientResource)
     {
+        transientResource.WaitForExclusiveLease();
+
         if (_maxCacheMemoryThreshold == 0)
         {
             for (int i = 0; i < ShardCount; i++)
             {
                 (int hashCode, TrieNode? node)[] shard = transientResource.Nodes.Shards[i];
                 for (int j = 0; j < shard.Length; j++)
                 {
-                    if (shard[j].node is { } newNode) newNode.PrunePersistedRecursively(1);
+                    if (shard[j].node is { } newNode && !newNode.IsWarmerOwned) newNode.PrunePersistedRecursively(1);
 
                 }
             }
@@ -128,12 +132,49 @@ void AddToCacheWithHashCode(int shardIdx, int hashCode, TrieNode newNode)
             }
         }
 
+        static TrieNode? TryMaterializeResolvedWarmerNode(TrieNode source)
+        {
+            if (!source.IsWarmerResolved) return null;
+
+            CappedArray<byte> fullRlp = source.FullRlp;
+            if (fullRlp.IsNull) return null;
+
+            Hash256? keccak = source.Keccak;
+            if (keccak is not null && ValueKeccak.Compute(fullRlp.AsSpan()) != keccak) return null;
+
+            TrieNode detached = keccak is null
+                ? new TrieNode(NodeType.Unknown, fullRlp)
+                : new TrieNode(NodeType.Unknown, keccak, fullRlp);
+            TreePath path = TreePath.Empty;
+
+            try
+            {
+                return detached.TryResolveNode(NullTrieNodeResolver.Instance, ref path) ? detached : null;
+            }
+            catch (IndexOutOfRangeException)
+            {
+                return null;
+            }
+            catch (ArgumentOutOfRangeException)
+            {
+                return null;
+            }
+        }
+
         Parallel.For(0, ShardCount, (i) =>
         {
             (int hashCode, TrieNode? node)[] shard = transientResource.Nodes.Shards[i];
             for (int j = 0; j < shard.Length; j++)
             {
-                if (shard[j].node is { } newNode && !IsPlaceholder(newNode)) AddToCacheWithHashCode(i, shard[j].hashCode, newNode);
+                if (shard[j].node is not { } source) continue;
+
+                TrieNode? newNode = source.IsWarmerOwned
+                    ? TryMaterializeResolvedWarmerNode(source)
+                    : source;
+                if (newNode is not null)
+                {
+                    AddToCacheWithHashCode(i, shard[j].hashCode, newNode);
+                }
             }
         });
 
@@ -169,14 +210,6 @@ void AddToCacheWithHashCode(int shardIdx, int hashCode, TrieNode newNode)
         Nethermind.Trie.Pruning.Metrics.MemoryUsedByCache = currentTotalMemory;
     }
 
-    /// <summary>
-    /// Identifies a placeholder trie node: <see cref="NodeType.Unknown"/> with empty RLP, carrying only a hash. The
-    /// trie warmer's negative cache and the trie commit path both produce it; it is not an authoritative node, so it
-    /// must neither enter this shared cache nor satisfy a live read - callers fall through to the snapshots or
-    /// persistence lookup instead.
-    /// </summary>
-    internal static bool IsPlaceholder(TrieNode node) => node.NodeType == NodeType.Unknown && node.FullRlp.Length == 0;
-
     /// <summary>
     /// Clears all cached trie nodes.
     /// </summary>

src/Nethermind/Nethermind.Trie.Test/TrieNodeTests.cs

@@ -99,6 +99,118 @@ public void When_resolving_an_unknown_node_without_rlp_trie_exception_should_be_
         Assert.Throws<TrieException>(() => trieNode.ResolveNode(NullTrieNodeResolver.Instance, TreePath.Empty));
     }
 
+    [Test]
+    public void Warmer_owned_resolution_preserves_unrelated_flags()
+    {
+        (byte[] rlp, _) = EncodedLeaf();
+        TrieNode trieNode = new(NodeType.Unknown, rlp);
+        trieNode.MarkWarmerOwned();
+        trieNode.IsBoundaryProofNode = true;
+        trieNode.IsPersisted = false;
+
+        TreePath path = TreePath.Empty;
+        Assert.That(trieNode.TryResolveNode(NullTrieNodeResolver.Instance, ref path), Is.True);
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(trieNode.IsWarmerOwned, Is.True);
+            Assert.That(trieNode.IsWarmerResolved, Is.True);
+            Assert.That(trieNode.IsBoundaryProofNode, Is.True);
+            Assert.That(trieNode.IsPersisted, Is.False);
+        }
+    }
+
+    [Test]
+    public void Inline_child_of_warmer_owned_node_uses_owned_resolution()
+    {
+        TrieNode inlineLeaf = TrieNodeFactory.CreateLeaf([0x3, 0x4], new CappedArray<byte>(new byte[] { 0x5 }));
+        TrieNode branch = new(NodeType.Branch);
+        branch.SetChild(0, inlineLeaf);
+        TreePath path = TreePath.Empty;
+        branch.ResolveKey(NullTrieNodeResolver.Instance, ref path);
+
+        TrieNode owned = new(NodeType.Unknown, branch.Keccak!);
+        owned.MarkWarmerOwned();
+        ITrieNodeResolver resolver = Substitute.For<ITrieNodeResolver>();
+        resolver.TryLoadRlp(TreePath.Empty, branch.Keccak!, ReadFlags.None).Returns(branch.FullRlp.ToArray());
+
+        Assert.That(owned.TryResolveNode(resolver, ref path), Is.True);
+
+        owned.AppendChildPath(ref path, 0);
+        TrieNode child = owned.GetChildWithChildPath(NullTrieNodeResolver.Instance, ref path, 0, keepChildRef: true)!;
+        Assert.That(child.TryResolveNode(NullTrieNodeResolver.Instance, ref path), Is.True);
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(child.IsWarmerOwned, Is.True);
+            Assert.That(child.IsWarmerResolved, Is.True);
+        }
+    }
+
+    [Test]
+    public void Concurrent_warmer_owned_try_resolve_loads_once()
+    {
+        (byte[] rlp, Hash256 hash) = EncodedLeaf();
+        TrieNode trieNode = new(NodeType.Unknown, hash);
+        trieNode.MarkWarmerOwned();
+
+        int loads = 0;
+        using ManualResetEventSlim loadStarted = new(false);
+        using ManualResetEventSlim allowLoad = new(false);
+        ITrieNodeResolver resolver = Substitute.For<ITrieNodeResolver>();
+        resolver.TryLoadRlp(TreePath.Empty, hash, ReadFlags.None).Returns(_ =>
+        {
+            Interlocked.Increment(ref loads);
+            loadStarted.Set();
+            if (!allowLoad.Wait(TimeSpan.FromSeconds(30))) throw new TimeoutException("owned resolver was not released");
+            return rlp;
+        });
+
+        using ManualResetEventSlim start = new(false);
+        Task[] tasks = new Task[4];
+        for (int i = 0; i < tasks.Length; i++)
+        {
+            tasks[i] = Task.Run(() =>
+            {
+                start.Wait();
+                TreePath path = TreePath.Empty;
+                Assert.That(trieNode.TryResolveNode(resolver, ref path), Is.True);
+            });
+        }
+
+        start.Set();
+        bool firstLoadStarted = loadStarted.Wait(TimeSpan.FromSeconds(30));
+        allowLoad.Set();
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(firstLoadStarted, Is.True);
+            Assert.That(Task.WaitAll(tasks, TimeSpan.FromSeconds(30)), Is.True);
+            Assert.That(Volatile.Read(ref loads), Is.EqualTo(1));
+        }
+    }
+
+    [Test]
+    public void Warmer_owned_try_resolve_rejects_rlp_of_another_node()
+    {
+        (byte[] unrelatedRlp, _) = EncodedLeaf();
+        Hash256 requestedHash = Keccak.Compute("requested node");
+        TrieNode trieNode = new(NodeType.Unknown, requestedHash);
+        trieNode.MarkWarmerOwned();
+
+        ITrieNodeResolver resolver = Substitute.For<ITrieNodeResolver>();
+        resolver.TryLoadRlp(TreePath.Empty, requestedHash, ReadFlags.None).Returns(unrelatedRlp);
+
+        TreePath path = TreePath.Empty;
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(trieNode.TryResolveNode(resolver, ref path), Is.False);
+            Assert.That(trieNode.NodeType, Is.EqualTo(NodeType.Unknown));
+            Assert.That(trieNode.IsWarmerResolved, Is.False);
+            Assert.That(trieNode.FullRlp.IsNotNull, Is.False);
+        }
+    }
+
     [Test]
     public void Encoding_leaf_without_key_throws_trie_exception()
     {
@@ -1056,6 +1168,14 @@ public void Do_Not_MarkUnpersistedChildAsPersisted()
         Assert.That(child.IsPersisted, Is.False);
     }
 
+    private static (byte[] Rlp, Hash256 Hash) EncodedLeaf()
+    {
+        TrieNode leaf = TrieNodeFactory.CreateLeaf([0x3, 0x4], new byte[32]);
+        TreePath path = TreePath.Empty;
+        leaf.ResolveKey(NullTrieNodeResolver.Instance, ref path);
+        return (leaf.FullRlp.ToArray()!, leaf.Keccak!);
+    }
+
     private class InMemoryScopedTrieStore : IScopedTrieStore
     {
         private readonly ConcurrentDictionary<TreePath, TrieNode> _nodes = new();

src/Nethermind/Nethermind.Trie.Test/TrieTests.cs

@@ -1319,6 +1319,16 @@ public void WarmUpPath_DoesNotThrow()
             Assert.That(() => patriciaTree.WarmUpPath(Bytes.FromHexString("fffffffffffff")), Throws.Nothing);  // Completely different path
         }
 
+        [Test]
+        public void WarmUpPath_DoesNotThrow_WhenPersistenceServesAnotherVersionOfTheNode()
+        {
+            StaleWarmerTrieStore trieStore = new();
+            PatriciaTree patriciaTree = new(trieStore, _logManager) { RootHash = StaleWarmerTrieStore.RootHashToWarm };
+
+            Assert.That(() => patriciaTree.WarmUpPath(_keyA), Throws.Nothing);
+            Assert.That(patriciaTree.RootRef!.NodeType, Is.EqualTo(NodeType.Unknown));
+        }
+
         [Test]
         public void Commit_DoesNotDeadlock_WhenRunOnBoundedScheduler()
         {
@@ -1347,5 +1357,42 @@ public void Commit_DoesNotDeadlock_WhenRunOnBoundedScheduler()
 
             Assert.That(task.Wait(TimeSpan.FromSeconds(10)), Is.True, "Commit deadlocked on bounded scheduler");
         }
+
+        /// <summary>
+        /// A path-keyed store that answers a warmer read with the RLP of another version of the node at that path,
+        /// which is what the flat DB does when the warmer runs ahead of, or behind, the live reads.
+        /// </summary>
+        private class StaleWarmerTrieStore : IScopedTrieStore
+        {
+            public static readonly Hash256 RootHashToWarm = Keccak.Compute("root to warm");
+
+            private readonly byte[] _rlpOfAnotherNode;
+
+            public StaleWarmerTrieStore()
+            {
+                TrieNode leaf = TrieNodeFactory.CreateLeaf([0x1, 0x2], new byte[32]);
+                TreePath path = TreePath.Empty;
+                leaf.ResolveKey(NullTrieNodeResolver.Instance, ref path);
+                _rlpOfAnotherNode = leaf.FullRlp.ToArray()!;
+            }
+
+            public TrieNode FindCachedOrUnknown(in TreePath path, Hash256 hash)
+            {
+                TrieNode node = new(NodeType.Unknown, hash);
+                node.MarkWarmerOwned();
+                return node;
+            }
+
+            public byte[]? LoadRlp(in TreePath path, Hash256 hash, ReadFlags flags = ReadFlags.None) => _rlpOfAnotherNode;
+
+            public byte[]? TryLoadRlp(in TreePath path, Hash256 hash, ReadFlags flags = ReadFlags.None) => _rlpOfAnotherNode;
+
+            public ITrieNodeResolver GetStorageTrieNodeResolver(Hash256? address) => this;
+
+            public INodeStorage.KeyScheme Scheme => INodeStorage.KeyScheme.HalfPath;
+
+            public ICommitter BeginCommit(TrieNode? root, WriteFlags writeFlags = WriteFlags.None) =>
+                throw new NotSupportedException();
+        }
     }
 }