fix(db): cross-column snapshot atomicity in SnapshotableMemColumnsDb

SnapshotableMemColumnsDb.CreateSnapshot iterates columns in a loop and
captures each column's snapshot independently. InMemoryColumnWriteBatch
.Dispose iterates the per-column batches and disposes each
independently. Without a global guard, a CreateSnapshot call concurrent
with an in-flight writeBatch.Dispose can capture some columns AFTER the
new writes and others BEFORE — producing a cross-column-inconsistent
reader view.

This race only matters in the test backend (RocksDB has atomic
cross-CF snapshots), but it manifested concretely in
E2ESyncTests.SnapSync on Windows: post-snap-sync block processing leases
a fresh persistenceReader while the persistence pipeline is committing,
gets a snapshot whose Accounts column is updated but whose StateNodes
column is not, then walks the trie at the new state root and throws
TrieNodeException for a node that "should" be there.

Add a ReaderWriterLockSlim around CreateSnapshot (read) and
writeBatch.Dispose (write) so multi-column commits are atomic w.r.t.
snapshot creation. Multiple snapshots can still proceed concurrently —
the only contention is the rare overlap of a snapshot creation with a
multi-column commit.

This eliminates the TrieNodeException failure mode in the stress
reproducer (`SnapSync_StressRepro`). The "missing in flat" mode is a
separate, deterministic bug (the same ~9 addresses miss reliably) that
appears unaffected by this fix and needs follow-up investigation.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: LukaszRozmej

src/Nethermind/Nethermind.Db/SnapshotableMemColumnsDb.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using Nethermind.Core;
 
 namespace Nethermind.Db
@@ -16,6 +17,17 @@ public class SnapshotableMemColumnsDb<TKey> : IColumnsDb<TKey> where TKey : stru
         private readonly Dictionary<TKey, SnapshotableMemDb> _columnDbs = new();
         private readonly bool _neverPrune;
 
+        // Cross-column atomicity guard. Each per-column SnapshotableMemDb has its own version
+        // counter and lock, so per-column reads/writes are individually consistent. But a
+        // multi-column writeBatch dispose applies columns one-by-one, and CreateSnapshot
+        // captures column snapshots one-by-one. Without this lock a snapshot taken concurrently
+        // with an in-flight writeBatch dispose can capture some columns AFTER the new writes
+        // and others BEFORE, producing a cross-column-inconsistent reader view. RocksDB does
+        // not have this problem (its snapshots are atomic across CFs); this lock makes the
+        // in-memory test backend match. The lock is only contended when a snapshot is taken
+        // simultaneously with a multi-column writeBatch.Dispose, which is rare.
+        private readonly ReaderWriterLockSlim _atomicityLock = new();
+
         private SnapshotableMemColumnsDb(TKey[] keys, bool neverPrune)
         {
             _neverPrune = neverPrune;
@@ -55,16 +67,52 @@ public IDb GetColumnDb(TKey key)
 
         public IReadOnlyColumnDb<TKey> CreateReadOnly(bool createInMemWriteStore) => new ReadOnlyColumnsDb<TKey>(this, createInMemWriteStore);
 
-        public IColumnsWriteBatch<TKey> StartWriteBatch() => new InMemoryColumnWriteBatch<TKey>(this);
+        public IColumnsWriteBatch<TKey> StartWriteBatch() => new AtomicColumnsWriteBatch(this);
 
         public IColumnDbSnapshot<TKey> CreateSnapshot()
         {
-            Dictionary<TKey, IKeyValueStoreSnapshot> snapshots = new();
-            foreach (KeyValuePair<TKey, SnapshotableMemDb> kvp in _columnDbs)
+            // Read lock: many concurrent snapshots are fine; writeBatch dispose blocks them.
+            _atomicityLock.EnterReadLock();
+            try
+            {
+                Dictionary<TKey, IKeyValueStoreSnapshot> snapshots = new();
+                foreach (KeyValuePair<TKey, SnapshotableMemDb> kvp in _columnDbs)
+                {
+                    snapshots[kvp.Key] = kvp.Value.CreateSnapshot();
+                }
+                return new ColumnSnapshot(snapshots);
+            }
+            finally
+            {
+                _atomicityLock.ExitReadLock();
+            }
+        }
+
+        /// <summary>
+        /// Wraps <see cref="InMemoryColumnWriteBatch{TKey}"/> so the per-column commit phase
+        /// happens under the columns DB's write lock, making the multi-column commit atomic
+        /// w.r.t. <see cref="CreateSnapshot"/>.
+        /// </summary>
+        private sealed class AtomicColumnsWriteBatch(SnapshotableMemColumnsDb<TKey> db) : IColumnsWriteBatch<TKey>
+        {
+            private readonly InMemoryColumnWriteBatch<TKey> _inner = new(db);
+
+            public IWriteBatch GetColumnBatch(TKey key) => _inner.GetColumnBatch(key);
+
+            public void Clear() => _inner.Clear();
+
+            public void Dispose()
             {
-                snapshots[kvp.Key] = kvp.Value.CreateSnapshot();
+                db._atomicityLock.EnterWriteLock();
+                try
+                {
+                    _inner.Dispose();
+                }
+                finally
+                {
+                    db._atomicityLock.ExitWriteLock();
+                }
             }
-            return new ColumnSnapshot(snapshots);
         }
 
         public void Dispose()
@@ -73,6 +121,7 @@ public void Dispose()
             {
                 db.Dispose();
             }
+            _atomicityLock.Dispose();
         }
 
         public void Flush(bool onlyWal = false)