Log index

src/Nethermind/Nethermind.Api/IApiWithStores.cs

@@ -6,6 +6,7 @@
 using Nethermind.Blockchain.Receipts;
 using Nethermind.Consensus;
 using Nethermind.Crypto;
+using Nethermind.Db;
 using Nethermind.Db.Blooms;
 using Nethermind.Facade.Find;
 using Nethermind.State.Repositories;
@@ -21,6 +22,7 @@ public interface IApiWithStores : IBasicApi
         IBloomStorage? BloomStorage { get; set; }
         IChainLevelInfoRepository? ChainLevelInfoRepository { get; set; }
         ILogFinder? LogFinder { get; set; }
+        ILogIndexStorage? LogIndexStorage { get; set; }
         ISigner? EngineSigner { get; set; }
         ISignerStore? EngineSignerStore { get; set; }
         IProtectedPrivateKey? NodeKey { get; set; }

src/Nethermind/Nethermind.Api/NethermindApi.cs

@@ -157,6 +157,7 @@ public IBlockchainBridge CreateBlockchainBridge()
         public IKeyStore? KeyStore { get; set; }
         public IPasswordProvider? PasswordProvider { get; set; }
         public ILogFinder? LogFinder { get; set; }
+        public ILogIndexStorage? LogIndexStorage { get; set; }
         public ILogManager LogManager => _dependencies.LogManager;
         public IMessageSerializationService MessageSerializationService { get; } = new MessageSerializationService();
         public IGossipPolicy GossipPolicy { get; set; } = Policy.FullGossip;

src/Nethermind/Nethermind.Blockchain.Test/Find/LogFinderTests.cs

@@ -35,6 +35,7 @@ public class LogFinderTests
     private IBloomStorage _bloomStorage = null!;
     private IReceiptsRecovery _receiptsRecovery = null!;
     private Block _headTestBlock = null!;
+    private ILogIndexStorage _logIndexStorage = null!;
 
     [SetUp]
     public void SetUp()
@@ -43,7 +44,11 @@ public void SetUp()
     }
 
     [TearDown]
-    public void TearDown() => _bloomStorage?.Dispose();
+    public async Task TearDown()
+    {
+        _bloomStorage?.Dispose();
+        await _logIndexStorage.DisposeAsync();
+    }
 
     private void SetUp(bool allowReceiptIterator)
     {
@@ -57,7 +62,8 @@ private void SetUp(bool allowReceiptIterator)
         _blockTree = _rawBlockTree;
         _bloomStorage = new BloomStorage(new BloomConfig(), new MemDb(), new InMemoryDictionaryFileStoreFactory());
         _receiptsRecovery = Substitute.For<IReceiptsRecovery>();
-        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery);
+        _logIndexStorage = Substitute.For<ILogIndexStorage>();
+        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
     }
 
     private void SetupHeadWithNoTransaction()
@@ -145,7 +151,7 @@ public void throw_exception_when_receipts_are_missing([ValueSource(nameof(WithBl
     {
         StoreTreeBlooms(withBloomDb);
         _receiptStorage = NullReceiptStorage.Instance;
-        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery);
+        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
 
         var logFilter = AllBlockFilter().Build();
 
@@ -158,7 +164,7 @@ public void throw_exception_when_receipts_are_missing([ValueSource(nameof(WithBl
     public void when_receipts_are_missing_and_header_has_no_receipt_root_do_not_throw_exception_()
     {
         _receiptStorage = NullReceiptStorage.Instance;
-        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery);
+        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
 
         SetupHeadWithNoTransaction();
 
@@ -174,7 +180,7 @@ public void filter_all_logs_should_throw_when_to_block_is_not_found([ValueSource
     {
         StoreTreeBlooms(withBloomDb);
         var blockFinder = Substitute.For<IBlockFinder>();
-        _logFinder = new LogFinder(blockFinder, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery);
+        _logFinder = new LogFinder(blockFinder, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
         var logFilter = AllBlockFilter().Build();
         var action = new Func<IEnumerable<FilterLog>>(() => _logFinder.FindLogs(logFilter));
         action.Should().Throw<ResourceNotFoundException>();
@@ -277,7 +283,7 @@ public void filter_by_blocks(LogFilter filter, int expectedCount, bool withBloom
     public void filter_by_blocks_with_limit([ValueSource(nameof(WithBloomValues))] bool withBloomDb)
     {
         StoreTreeBlooms(withBloomDb);
-        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, 2);
+        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
         var filter = FilterBuilder.New().FromLatestBlock().ToLatestBlock().Build();
         var logs = _logFinder.FindLogs(filter).ToArray();
 
@@ -323,7 +329,7 @@ public async Task Throw_log_finder_operation_canceled_after_given_timeout([Value
         CancellationToken cancellationToken = cancellationTokenSource.Token;
 
         StoreTreeBlooms(true);
-        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery);
+        _logFinder = new LogFinder(_blockTree, _receiptStorage, _receiptStorage, _bloomStorage, LimboLogs.Instance, _receiptsRecovery, _logIndexStorage);
         var logFilter = AllBlockFilter().Build();
         var logs = _logFinder.FindLogs(logFilter, cancellationToken);
 

src/Nethermind/Nethermind.Blockchain/Receipts/IReceiptConfig.cs

@@ -13,8 +13,8 @@ public interface IReceiptConfig : IConfig
     [ConfigItem(Description = "Whether to migrate the receipts database to the new schema.", DefaultValue = "false")]
     bool ReceiptsMigration { get; set; }
 
-    [ConfigItem(Description = "The degree of parallelism during receipt migration.", DefaultValue = "0", HiddenFromDocs = true)]
-    int ReceiptsMigrationDegreeOfParallelism { get; set; }
+    [ConfigItem(Description = "The degree of IO operations parallelism during receipts migration.", DefaultValue = "32", HiddenFromDocs = true)]
+    int ReceiptsMigrationIOParallelism { get; set; }
 
     [ConfigItem(Description = "Force receipt recovery if its not able to detect it.", DefaultValue = "false", HiddenFromDocs = true)]
     bool ForceReceiptsMigration { get; set; }

src/Nethermind/Nethermind.Blockchain/Receipts/ReceiptConfig.cs

@@ -7,7 +7,7 @@ public class ReceiptConfig : IReceiptConfig
 {
     public bool StoreReceipts { get; set; } = true;
     public bool ReceiptsMigration { get; set; } = false;
-    public int ReceiptsMigrationDegreeOfParallelism { get; set; } = 0;
+    public int ReceiptsMigrationIOParallelism { get; set; } = 16; // TODO: add validation
     public bool ForceReceiptsMigration { get; set; } = false;
     public bool CompactReceiptStore { get; set; } = true;
     public bool CompactTxIndex { get; set; } = true;

src/Nethermind/Nethermind.Core.Test/Blockchain/TestBlockchain.cs

@@ -219,7 +219,8 @@ protected virtual async Task<TestBlockchain> Build(ISpecProvider? specProvider =
 
         BloomStorage bloomStorage = new(new BloomConfig(), new MemDb(), new InMemoryDictionaryFileStoreFactory());
         ReceiptsRecovery receiptsRecovery = new(new EthereumEcdsa(SpecProvider.ChainId), SpecProvider);
-        LogFinder = new LogFinder(BlockTree, ReceiptStorage, ReceiptStorage, bloomStorage, LimboLogs.Instance, receiptsRecovery);
+        LogIndexStorage logIndexStorage = new(DbProvider.LogIndexDb, LogManager.GetClassLogger<LogIndexStorage>(), "test-db", ioParallelism: 8);
+        LogFinder = new LogFinder(BlockTree, ReceiptStorage, ReceiptStorage, bloomStorage, LimboLogs.Instance, receiptsRecovery, logIndexStorage);
         BlockProcessor = CreateBlockProcessor(WorldStateManager.GlobalWorldState);
 
         BlockchainProcessor chainProcessor = new(BlockTree, BlockProcessor, BlockPreprocessorStep, StateReader, LogManager, Consensus.Processing.BlockchainProcessor.Options.Default);

src/Nethermind/Nethermind.Core/Collections/CollectionExtensions.cs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 using System;
+using System.Buffers;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq.Expressions;
@@ -87,6 +88,23 @@ public static bool NoResizeClear<TKey, TValue>(this ConcurrentDictionary<TKey, T
             return true;
         }
 
+        public static TValue GetOrAdd<TKey, TValue>(this Dictionary<TKey, TValue> dictionary, TKey key, Func<TKey, TValue> factory)
+            where TKey : notnull
+        {
+            if (dictionary.TryGetValue(key, out TValue? value))
+            {
+                return value;
+            }
+            else
+            {
+                value = factory(key);
+                dictionary.Add(key, value);
+                return value;
+            }
+        }
+
+        public static Span<T> RentSpan<T>(this ArrayPool<T> pool, int length) => pool.Rent(length).AsSpan(length);
+
         private static class ClearCache<TKey, TValue> where TKey : notnull
         {
             public static readonly Action<ConcurrentDictionary<TKey, TValue>> Clear = CreateNoResizeClearExpression();

src/Nethermind/Nethermind.Core/Extensions/IntExtensions.cs

@@ -29,10 +29,14 @@ public static UInt256 GWei(this int @this)
         return (uint)@this * Unit.GWei;
     }
 
-    public static byte[] ToByteArray(this int value)
+    public static byte[] ToLittleEndianByteArray(this int value)
     {
-        byte[] bytes = new byte[sizeof(int)];
-        BinaryPrimitives.WriteInt32BigEndian(bytes, value);
+        byte[] bytes = BitConverter.GetBytes(value);
+        if (!BitConverter.IsLittleEndian)
+        {
+            Array.Reverse(bytes);
+        }
+
         return bytes;
     }
 

src/Nethermind/Nethermind.Core/Extensions/TaskExtensions.cs

@@ -40,4 +40,13 @@ public static bool HasTimeoutException(this Task? task)
         return exception.InnerException is TimeoutException
             || exception.InnerExceptions.Any(static ex => ex is TimeoutException);
     }
+
+    public static Task IgnoreException(this Task task)
+    {
+        return task.ContinueWith(_ => { });
+    }
+
+    public static T WaitResult<T>(this ValueTask<T> task) => task.Preserve().GetAwaiter().GetResult();
+
+    public static void Wait(this ValueTask task) => task.Preserve().GetAwaiter().GetResult();
 }

src/Nethermind/Nethermind.Db.Rocks/ColumnDb.cs

@@ -57,6 +57,11 @@ public void PutSpan(ReadOnlySpan<byte> key, ReadOnlySpan<byte> value, WriteFlags
         _mainDb.SetWithColumnFamily(key, _columnFamily, value, writeFlags);
     }
 
+    public void Merge(ReadOnlySpan<byte> key, ReadOnlySpan<byte> value, WriteFlags writeFlags = WriteFlags.None)
+    {
+        _mainDb.MergeWithColumnFamily(key, _columnFamily, value, writeFlags);
+    }
+
     public KeyValuePair<byte[], byte[]?>[] this[byte[][] keys] =>
         _rocksDb.MultiGet(keys, keys.Select(k => _columnFamily).ToArray());
 
@@ -115,6 +120,11 @@ public void PutSpan(ReadOnlySpan<byte> key, ReadOnlySpan<byte> value, WriteFlags
         {
             _underlyingWriteBatch.Set(key, value, _columnDb._columnFamily, flags);
         }
+
+        public void Merge(ReadOnlySpan<byte> key, ReadOnlySpan<byte> value, WriteFlags flags = WriteFlags.None)
+        {
+            _underlyingWriteBatch.Merge(key, value, _columnDb._columnFamily, flags);
+        }
     }
 
     public void Remove(ReadOnlySpan<byte> key)
@@ -130,7 +140,7 @@ public bool KeyExists(ReadOnlySpan<byte> key)
 
     public void Flush(bool onlyWal)
     {
-        _mainDb.Flush(onlyWal);
+        _mainDb.FlushWithColumnFamily(_columnFamily);
     }
 
     public void Compact()
@@ -151,4 +161,14 @@ public void DangerousReleaseMemory(in ReadOnlySpan<byte> span)
     {
         _mainDb.DangerousReleaseMemory(span);
     }
+
+    public IIterator<byte[], byte[]> GetIterator(bool isTailing = false)
+    {
+        return _mainDb.GetIterator(isTailing, _columnFamily);
+    }
+
+    public IIterator<byte[], byte[]> GetIterator(ref IteratorOptions options)
+    {
+        return _mainDb.GetIterator(ref options, _columnFamily);
+    }
 }

src/Nethermind/Nethermind.Db.Rocks/Config/DbConfig.cs

@@ -239,4 +239,15 @@ public class DbConfig : IDbConfig
     public string L1OriginDbRocksDbOptions { get; set; } = "";
 
     public string? L1OriginDbAdditionalRocksDbOptions { get; set; }
+
+    // TODO: cleanup & optimize settings
+    public string? LogIndexStorageDbRocksDbOptions { get; set; } =
+        "merge_operator=LogIndexStorage.Merge;";
+    public string? LogIndexStorageDbAdditionalRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageDefaultDbRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageDefaultDbAdditionalRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageAddressesDbRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageAddressesDbAdditionalRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageTopicsDbRocksDbOptions { get; set; } = "";
+    public string? LogIndexStorageTopicsDbAdditionalRocksDbOptions { get; set; } = "";
 }

src/Nethermind/Nethermind.Db.Rocks/Config/IDbConfig.cs

@@ -97,4 +97,13 @@ public interface IDbConfig : IConfig
 
     string L1OriginDbRocksDbOptions { get; set; }
     string? L1OriginDbAdditionalRocksDbOptions { get; set; }
+
+    string? LogIndexStorageDbRocksDbOptions { get; set; }
+    string? LogIndexStorageDbAdditionalRocksDbOptions { get; set; }
+    string? LogIndexStorageDefaultDbRocksDbOptions { get; set; }
+    string? LogIndexStorageDefaultDbAdditionalRocksDbOptions { get; set; }
+    string? LogIndexStorageAddressesDbRocksDbOptions { get; set; }
+    string? LogIndexStorageAddressesDbAdditionalRocksDbOptions { get; set; }
+    string? LogIndexStorageTopicsDbRocksDbOptions { get; set; }
+    string? LogIndexStorageTopicsDbAdditionalRocksDbOptions { get; set; }
 }

src/Nethermind/Nethermind.Db.Rocks/Config/PerTableDbConfig.cs

@@ -40,7 +40,7 @@ private void EnsureConfigIsAvailable(string propertyName)
         foreach (var prefix in _prefixes)
         {
             string prefixed = string.Concat(prefix, propertyName);
-            if (type.GetProperty(prefixed, BindingFlags.Public | BindingFlags.Instance) is null)
+            if (GetPrefixedConfigProperty(type, prefixed) is null)
             {
                 throw new InvalidConfigurationException($"Configuration {propertyName} not available with prefix {prefix}. Add {prefix}{propertyName} to {nameof(IDbConfig)}.", -1);
             }
@@ -92,13 +92,14 @@ private static string ReadRocksdbOptions(IDbConfig dbConfig, string propertyName
         Type type = dbConfig.GetType();
         PropertyInfo? propertyInfo;
 
+        // TODO: clarify if this can be case-insensitive
         string val = (string)type.GetProperty(propertyName, BindingFlags.Public | BindingFlags.Instance)!.GetValue(dbConfig)!;
 
         foreach (var prefix in prefixes)
         {
             string prefixed = string.Concat(prefix, propertyName);
 
-            propertyInfo = type.GetProperty(prefixed, BindingFlags.IgnoreCase | BindingFlags.Public | BindingFlags.Instance);
+            propertyInfo = GetPrefixedConfigProperty(type, prefixed);
             if (propertyInfo is not null)
             {
                 string? valObj = (string?)propertyInfo.GetValue(dbConfig);
@@ -124,7 +125,7 @@ private static string ReadRocksdbOptions(IDbConfig dbConfig, string propertyName
             {
                 string prefixed = string.Concat(prefix, propertyName);
 
-                propertyInfo = type.GetProperty(prefixed, BindingFlags.IgnoreCase | BindingFlags.Public | BindingFlags.Instance);
+                propertyInfo = GetPrefixedConfigProperty(type, prefixed);
                 if (propertyInfo is not null)
                 {
                     if (propertyInfo.PropertyType.CanBeAssignedNull())
@@ -149,7 +150,7 @@ private static string ReadRocksdbOptions(IDbConfig dbConfig, string propertyName
             }
 
             // Use generic one even if its available
-            propertyInfo = type.GetProperty(propertyName, BindingFlags.IgnoreCase | BindingFlags.Public | BindingFlags.Instance);
+            propertyInfo = GetPrefixedConfigProperty(type, propertyName);
             return (T?)propertyInfo?.GetValue(dbConfig);
         }
         catch (Exception e)
@@ -158,4 +159,8 @@ private static string ReadRocksdbOptions(IDbConfig dbConfig, string propertyName
         }
     }
 
+    private static PropertyInfo? GetPrefixedConfigProperty(Type type, string name)
+    {
+        return type.GetProperty(name, BindingFlags.IgnoreCase | BindingFlags.Public | BindingFlags.Instance);
+    }
 }