[Neo Core] Ensure LevelDB Snapshot Only be Used by One Thread

src/Plugins/LevelDBStore/Plugins/Storage/LevelDBStore.cs

@@ -12,7 +12,10 @@
 using Neo.IO.Data.LevelDB;
 using Neo.Persistence;
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 
 namespace Neo.Plugins.Storage
 {
@@ -27,7 +30,7 @@ public LevelDBStore()
 
         public IStore GetStore(string path)
         {
-            if (Environment.CommandLine.Split(' ').Any(p => p == "/repair" || p == "--repair"))
+            if (Environment.CommandLine.Split(' ').Any(p => p is "/repair" or "--repair"))
                 DB.Repair(path, Options.Default);
             return new Store(path);
         }

src/Plugins/LevelDBStore/Plugins/Storage/Snapshot.cs

@@ -11,7 +11,9 @@
 
 using Neo.IO.Data.LevelDB;
 using Neo.Persistence;
+using System;
 using System.Collections.Generic;
+using System.Runtime.CompilerServices;
 using LSnapshot = Neo.IO.Data.LevelDB.Snapshot;
 
 namespace Neo.Plugins.Storage
@@ -22,47 +24,65 @@ internal class Snapshot : ISnapshot
         private readonly LSnapshot snapshot;
         private readonly ReadOptions options;
         private readonly WriteBatch batch;
+        private readonly int _threadId;
 
         public Snapshot(DB db)
         {
             this.db = db;
             snapshot = db.GetSnapshot();
             options = new ReadOptions { FillCache = false, Snapshot = snapshot };
             batch = new WriteBatch();
+            _threadId = Environment.CurrentManagedThreadId;
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void EnsureThreadAccess()
+        {
+            if (_threadId != Environment.CurrentManagedThreadId)
+            {
+                throw new InvalidOperationException("Snapshot cannot be accessed from multiple threads.");
+            }
         }
 
         public void Commit()
         {
+            EnsureThreadAccess();
             db.Write(WriteOptions.Default, batch);
         }
 
         public void Delete(byte[] key)
         {
+            EnsureThreadAccess();
             batch.Delete(key);
         }
 
         public void Dispose()
         {
+            EnsureThreadAccess();
             snapshot.Dispose();
         }
 
         public IEnumerable<(byte[] Key, byte[] Value)> Seek(byte[] prefix, SeekDirection direction = SeekDirection.Forward)
         {
+            EnsureThreadAccess();
             return db.Seek(options, prefix, direction, (k, v) => (k, v));
         }
 
         public void Put(byte[] key, byte[] value)
         {
+            EnsureThreadAccess();
             batch.Put(key, value);
         }
 
         public bool Contains(byte[] key)
         {
+            EnsureThreadAccess();
             return db.Contains(options, key);
         }
 
         public byte[] TryGet(byte[] key)
         {
+            EnsureThreadAccess();
             return db.Get(options, key);
         }
     }

src/Plugins/LevelDBStore/Plugins/Storage/Store.cs

@@ -44,7 +44,7 @@ public ISnapshot GetSnapshot()
             return new Snapshot(db);
         }
 
-        public void Put(byte[] key, byte[] value)
+        public virtual void Put(byte[] key, byte[] value)
         {
             db.Put(WriteOptions.Default, key, value);
         }

tests/Neo.Plugins.Storage.Tests/StoreTest.MultiThread.cs

@@ -0,0 +1,148 @@
+// Copyright (C) 2015-2024 The Neo Project.
+//
+// StoreSnapshotTest.cs file belongs to the neo project and is free
+// software distributed under the MIT software license, see the
+// accompanying file LICENSE in the main directory of the
+// repository or http://www.opensource.org/licenses/mit-license.php
+// for more details.
+//
+// Redistribution and use in source and binary forms with or without
+// modifications are permitted.
+
+using Akka.Util.Internal;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Neo.Ledger;
+using Neo.Network.P2P.Payloads;
+using System;
+using System.IO;
+using System.Threading.Tasks;
+
+namespace Neo.Plugins.Storage.Tests;
+
+
+partial class StoreTest
+{
+
+    #region Async Tests for OnCommitting and OnCommitted
+
+    // Tests bellow are tests for the async delegate issue
+    // first reported by Vitor (vncoelho) in https://github.com/neo-project/neo/issues/3356
+
+    [TestMethod]
+    public void TestOneThreadLevelDbSnapshotPut()
+    {
+        using var store = levelDbStore.GetStore(path_leveldb);
+        var snapshot = store.GetSnapshot();
+
+        var testKey = new byte[] { 0x01, 0x02, 0x03 };
+
+        for (var i = 0; i < 2; i++)
+        {
+            var value = new byte[] { 0x04, 0x05, 0x06, (byte)i };
+            snapshot.Put(testKey, value);
+            snapshot.Commit();
+        }
+        snapshot.Dispose();
+    }
+
+    [TestMethod]
+    public void TestSingleExtraThreadLevelDbSnapshotPut()
+    {
+        using var store = levelDbStore.GetStore(path_leveldb);
+
+        object locker = new();
+        var snapshot = store.GetSnapshot();
+
+        var testKey = new byte[] { 0x01, 0x02, 0x03 };
+
+        var tasks = new Task[1];
+        for (var i = 0; i < tasks.Length; i++)
+        {
+            var value = new byte[] { 0x04, 0x05, 0x06, (byte)i };
+            tasks[i] = Task.Run(() =>
+            {
+                try
+                {
+                    lock (locker)
+                    {
+                        snapshot.Put(testKey, value);
+                        snapshot.Commit();
+                    }
+                }
+                catch (Exception ex)
+                {
+                    Console.WriteLine($"Exception in task: {ex.Message}");
+                    throw;
+                }
+            });
+        }
+
+        try
+        {
+            Task.WaitAll(tasks);
+        }
+        catch (AggregateException ae)
+        {
+            foreach (var ex in ae.InnerExceptions)
+            {
+                Console.WriteLine($"AggregateException: {ex.Message}");
+            }
+            throw;
+        }
+        finally
+        {
+            snapshot.Dispose();
+        }
+    }
+
+    [TestMethod]
+    [ExpectedException(typeof(AggregateException))]
+    public void TestMultiThreadLevelDbSnapshotPut()
+    {
+        using var store = levelDbStore.GetStore(path_leveldb);
+
+        object locker = new();
+        var snapshot = store.GetSnapshot();
+
+        var testKey = new byte[] { 0x01, 0x02, 0x03 };
+
+        var tasks = new Task[2];
+        for (var i = 0; i < tasks.Length; i++)
+        {
+            var value = new byte[] { 0x04, 0x05, 0x06, (byte)i };
+            tasks[i] = Task.Run(() =>
+            {
+                lock (locker)
+                {
+                    snapshot.Put(testKey, value);
+                    snapshot.Commit();
+                }
+            });
+        }
+        Task.WaitAll(tasks);
+        snapshot.Dispose();
+    }
+
+    [TestMethod]
+    public void TestOneSnapshotPerThreadLevelDbSnapshotPut()
+    {
+        using var store = levelDbStore.GetStore(path_leveldb);
+        var testKey = new byte[] { 0x01, 0x02, 0x03 };
+
+        var tasks = new Task[100];
+        for (var i = 0; i < tasks.Length; i++)
+        {
+            var value = new byte[] { 0x04, 0x05, 0x06, (byte)i };
+            tasks[i] = Task.Run(() =>
+            {
+                var snapshot = store.GetSnapshot();
+                snapshot.Put(testKey, value);
+                snapshot.Commit();
+                snapshot.Dispose();
+            });
+        }
+        Task.WaitAll(tasks);
+    }
+
+    #endregion
+}

tests/Neo.Plugins.Storage.Tests/StoreTest.cs

@@ -17,7 +17,7 @@
 namespace Neo.Plugins.Storage.Tests
 {
     [TestClass]
-    public class StoreTest
+    public partial class StoreTest
     {
         private const string path_leveldb = "Data_LevelDB_UT";
         private const string path_rocksdb = "Data_RocksDB_UT";