Nethereum.BlockchainProcessing

Nethereum.BlockchainProcessing provides a comprehensive framework for crawling, processing, and storing Ethereum blockchain data including blocks, transactions, receipts, and event logs with flexible filtering, progress tracking, and storage capabilities.

Overview

This package enables applications to:

Crawl blockchain data - Process blocks, transactions, receipts, and logs sequentially
Track progress - Resume processing from last processed block
Store data - Persist blockchain data to custom storage backends
Filter events - Process specific event types or contracts
Handle confirmations - Wait for block confirmations before processing
Batch process logs - Efficiently retrieve and process event logs in batches
Process ERC20/ERC721/ERC1155 - Built-in support for token transfer indexing and balance aggregation
Detect reorgs - Chain consistency validation with automatic rewind and non-canonical marking
Index internal transactions - Trace-based call tree extraction via configurable trace provider
Emit metrics - OpenTelemetry-compatible instrumentation via ILogProcessingObserver

Installation

dotnet add package Nethereum.BlockchainProcessing

Core Architecture

The package follows a modular pipeline architecture with clear separation of concerns:

BlockchainProcessor (executor)
  ├── IBlockchainProcessingOrchestrator (strategy)
  │   ├── BlockCrawlOrchestrator (block-by-block crawling)
  │   ├── LogOrchestrator (batch log retrieval)
  │   └── InternalTransactionOrchestrator (trace-based indexing)
  ├── BlockProcessingSteps (pipeline stages)
  ├── IBlockProgressRepository (progress tracking)
  │   └── ReorgBufferedBlockProgressRepository (decorator)
  ├── IChainStateRepository (reorg detection state)
  ├── ILastConfirmedBlockNumberService (confirmation management)
  └── ILogProcessingObserver (metrics/telemetry)

Core Components

BlockchainProcessor

Main processor that manages continuous blockchain processing. Located in BlockchainProcessor.cs.

Key Methods:

ExecuteAsync(CancellationToken, BigInteger?) - Process until cancelled
ExecuteAsync(BigInteger toBlockNumber, ...) - Process to specific block

Features:

Progress tracking via IBlockProgressRepository
Automatic block confirmation handling
Cancellation token support
Resume from last processed block

BlockCrawlOrchestrator

Orchestrates crawling of blocks, transactions, receipts, and logs. Located in BlockProcessing/BlockCrawlOrchestrator.cs.

Processing Flow:

Fetch Block → BlockCrawlerStep
For Each Transaction:
- Process Transaction → TransactionCrawlerStep
- Fetch Receipt → TransactionReceiptCrawlerStep
- Extract Contract Creation → ContractCreatedCrawlerStep (if applicable)
For Each Log:
- Process Log → FilterLogCrawlerStep

BlockProcessingSteps

Defines processing pipeline stages. Located in BlockProcessing/BlockProcessingSteps.cs.

Steps:

BlockStep - Processes BlockWithTransactions
TransactionStep - Processes TransactionVO
TransactionReceiptStep - Processes TransactionReceiptVO
FilterLogStep - Processes FilterLogVO
ContractCreationStep - Processes ContractCreationVO

Each step is a Processor<T> that can have multiple handlers.

Processor

Generic processor that executes multiple handlers. Located in Processor/Processor.cs.

Key Features:

Multiple handlers per processor
Optional match criteria for filtering
Sequential handler execution
Synchronous and asynchronous handler support

Usage Examples

Example 1: Basic Block Processing

Process all blocks, transactions, and logs:

using Nethereum.BlockchainProcessing;
using Nethereum.Web3;
using System.Numerics;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");

var processedData = new ProcessedData();

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(steps =>
{
    // Process each block
    steps.BlockStep.AddSynchronousProcessorHandler(block =>
    {
        Console.WriteLine($"Block: {block.Number}");
        processedData.Blocks.Add(block);
    });

    // Process each transaction
    steps.TransactionStep.AddSynchronousProcessorHandler(tx =>
    {
        Console.WriteLine($"  Transaction: {tx.Transaction.TransactionHash}");
        processedData.Transactions.Add(tx);
    });

    // Process transaction receipts
    steps.TransactionReceiptStep.AddSynchronousProcessorHandler(tx =>
    {
        Console.WriteLine($"  Receipt - Gas Used: {tx.TransactionReceipt.GasUsed}");
        processedData.TransactionsWithReceipt.Add(tx);
    });

    // Process event logs
    steps.FilterLogStep.AddSynchronousProcessorHandler(filterLog =>
    {
        Console.WriteLine($"    Log: {filterLog.Log.Address}");
        processedData.FilterLogs.Add(filterLog);
    });
});

// Process blocks 100-110
await blockProcessor.ExecuteAsync(
    toBlockNumber: new BigInteger(110),
    cancellationToken: CancellationToken.None,
    startAtBlockNumberIfNotProcessed: new BigInteger(100)
);

From test: BlockProcessing/BlockProcessingTests.cs:17-44

Example 2: Processing with Progress Tracking

Resume processing from last processed block:

using Nethereum.BlockchainProcessing.ProgressRepositories;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");

// Track progress (persists across runs)
var progressRepository = new InMemoryBlockchainProgressRepository(
    lastBlockProcessed: new BigInteger(1000)
);

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(
    progressRepository,
    steps =>
    {
        steps.BlockStep.AddSynchronousProcessorHandler(block =>
        {
            Console.WriteLine($"Processing block: {block.Number}");
        });
    }
);

// Process continuously until cancelled
// Will start from block 1001 (last processed + 1)
var cancellationTokenSource = new CancellationTokenSource();

await blockProcessor.ExecuteAsync(cancellationTokenSource.Token);

// Progress is automatically saved after each block

From test: BlockProcessing/BlockProcessingTests.cs:164-195

Example 3: Block Confirmations

Wait for block confirmations before processing:

const uint MIN_CONFIRMATIONS = 12;

var progressRepository = new InMemoryBlockchainProgressRepository(
    lastBlockProcessed: new BigInteger(100)
);

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(
    progressRepository,
    steps =>
    {
        steps.BlockStep.AddSynchronousProcessorHandler(block =>
        {
            Console.WriteLine($"Processing confirmed block: {block.Number}");
        });
    },
    minimumBlockConfirmations: MIN_CONFIRMATIONS  // Wait for 12 confirmations
);

await blockProcessor.ExecuteAsync(CancellationToken.None);

// Only processes blocks with at least 12 confirmations
// If latest block is 1000, will process up to block 988

From test: BlockProcessing/BlockProcessingTests.cs:231-267

Example 4: Filtering with Criteria

Process only specific transactions:

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(steps =>
{
    // Only process transactions with non-zero value
    steps.TransactionStep.SetMatchCriteria(tx =>
        tx.Transaction.Value?.Value > 0);

    // Only process receipts for transaction index 0
    steps.TransactionReceiptStep.SetMatchCriteria(tx =>
        tx.Transaction.TransactionIndex.Value == 0);

    steps.TransactionReceiptStep.AddSynchronousProcessorHandler(tx =>
    {
        Console.WriteLine($"High-value transaction at index 0: {tx.TransactionHash}");
    });
});

await blockProcessor.ExecuteAsync(new BigInteger(110), CancellationToken.None, new BigInteger(100));

// Only transactions matching ALL criteria will be processed

From test: BlockProcessing/BlockProcessingTests.cs:134-161

Example 5: Disabling Processing Steps

Optimize by skipping unnecessary steps:

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(steps =>
{
    // Only interested in blocks, not transactions
    steps.BlockStep.AddSynchronousProcessorHandler(block =>
    {
        Console.WriteLine($"Block {block.Number}: {block.TransactionCount} transactions");
    });
});

// Disable receipt and log processing for performance
blockProcessor.Orchestrator.TransactionWithReceiptCrawlerStep.Enabled = false;
blockProcessor.Orchestrator.FilterLogCrawlerStep.Enabled = false;

await blockProcessor.ExecuteAsync(new BigInteger(110), CancellationToken.None, new BigInteger(100));

// Receipts and logs won't be fetched or processed

From test: BlockProcessing/BlockProcessingTests.cs:76-106

Example 6: Block Storage Processor

Automatically store blockchain data to repositories:

using Nethereum.BlockchainProcessing.BlockStorage.Repositories;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");

// In-memory storage (replace with your database implementation)
var context = new InMemoryBlockchainStorageRepositoryContext();
var repositoryFactory = new InMemoryBlockchainStorageRepositoryFactory(context);

var processor = web3.Processing.Blocks.CreateBlockStorageProcessor(
    repositoryFactory,
    minimumBlockConfirmations: 6
);

// Process and automatically store blocks, transactions, and logs
await processor.ExecuteAsync(
    toBlockNumber: new BigInteger(110),
    cancellationToken: CancellationToken.None,
    startAtBlockNumberIfNotProcessed: new BigInteger(100)
);

// Data is automatically persisted
Console.WriteLine($"Blocks stored: {context.Blocks.Count}");
Console.WriteLine($"Transactions stored: {context.Transactions.Count}");
Console.WriteLine($"Logs stored: {context.TransactionLogs.Count}");

From test: BlockStorage/BlockStorageProcessorTests.cs:16-39

Example 7: Custom Storage Configuration

Add custom processing alongside storage:

var repositoryFactory = new InMemoryBlockchainStorageRepositoryFactory(context);

var processor = web3.Processing.Blocks.CreateBlockStorageProcessor(
    repositoryFactory,
    minimumBlockConfirmations: 6,
    configureSteps: steps =>
    {
        // Add custom handler alongside automatic storage
        steps.BlockStep.AddSynchronousProcessorHandler(block =>
        {
            // Send notification, update cache, etc.
            Console.WriteLine($"New block stored: {block.Number}");
        });

        // Add custom filtering
        steps.TransactionStep.SetMatchCriteria(tx =>
            tx.Transaction.Value?.Value > Web3.Convert.ToWei(1));
    }
);

await processor.ExecuteAsync(new BigInteger(110), CancellationToken.None, new BigInteger(100));

Log Processing

For event-focused processing, use LogOrchestrator for efficient batch retrieval. Located in LogProcessing/LogOrchestrator.cs.

Example 8: Process All Logs

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");

var logsProcessed = new List<FilterLog>();

var logProcessor = web3.Processing.Logs.CreateProcessor(
    filterLog => logsProcessed.Add(filterLog)
);

// Batch retrieval of logs (more efficient than block-by-block)
await logProcessor.ExecuteAsync(
    toBlockNumber: new BigInteger(110),
    cancellationToken: CancellationToken.None,
    startAtBlockNumberIfNotProcessed: new BigInteger(100)
);

Console.WriteLine($"Processed {logsProcessed.Count} logs");

From test: LogProcessing/LogProcessingTests.cs:15-60

Example 9: Process Specific Event Type

Process typed events with automatic decoding:

using Nethereum.Contracts.Standards.ERC20.ContractDefinition;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");

var transferEvents = new List<EventLog<TransferEventDTO>>();

var logProcessor = web3.Processing.Logs.CreateProcessor<TransferEventDTO>(
    transferEvent =>
    {
        Console.WriteLine($"Transfer: {transferEvent.Event.Value} from {transferEvent.Event.From} to {transferEvent.Event.To}");
        transferEvents.Add(transferEvent);
    }
);

await logProcessor.ExecuteAsync(
    toBlockNumber: new BigInteger(110),
    startAtBlockNumberIfNotProcessed: new BigInteger(100)
);

Console.WriteLine($"Total transfers: {transferEvents.Count}");

From test: LogProcessing/LogProcessingForEventTests.cs:15-41

Example 10: Event Processing with Criteria

Filter events during processing:

var largeTransfers = new List<EventLog<TransferEventDTO>>();

var logProcessor = web3.Processing.Logs.CreateProcessor<TransferEventDTO>(
    // Action
    action: transferEventLog =>
    {
        largeTransfers.Add(transferEventLog);
        return Task.CompletedTask;
    },
    // Criteria - only transfers over 1 ETH equivalent
    criteria: transferEventLog =>
    {
        var match = transferEventLog.Event.Value > Web3.Convert.ToWei(1);
        return Task.FromResult(match);
    }
);

await logProcessor.ExecuteAsync(
    toBlockNumber: new BigInteger(110),
    startAtBlockNumberIfNotProcessed: new BigInteger(100)
);

Console.WriteLine($"Large transfers: {largeTransfers.Count}");

From test: LogProcessing/LogProcessingForEventTests.cs:68-99

Example 11: Process Contract-Specific Events

Process events from specific contract:

var usdcAddress = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48";

var logProcessor = web3.Processing.Logs.CreateProcessorForContract<TransferEventDTO>(
    usdcAddress,
    transferEvent =>
    {
        Console.WriteLine($"USDC Transfer: {transferEvent.Event.Value}");
    }
);

await logProcessor.ExecuteAsync(new BigInteger(110), startAtBlockNumberIfNotProcessed: new BigInteger(100));

From: Services/BlockchainLogProcessingService.cs:185-218

Example 12: Process Multiple Contracts

var tokenAddresses = new[]
{
    "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", // USDC
    "0xdAC17F958D2ee523a2206206994597C13D831ec7"  // USDT
};

var logProcessor = web3.Processing.Logs.CreateProcessorForContracts<TransferEventDTO>(
    tokenAddresses,
    transferEvent =>
    {
        Console.WriteLine($"Stablecoin Transfer at {transferEvent.Log.Address}: {transferEvent.Event.Value}");
    }
);

await logProcessor.ExecuteAsync(new BigInteger(110), startAtBlockNumberIfNotProcessed: new BigInteger(100));

From: Services/BlockchainLogProcessingService.cs:220-261

ERC20/ERC721 Processing

Example 13: ERC20 Transfer Processing

Built-in support for ERC20 token tracking:

using Nethereum.BlockchainProcessing.Services.SmartContracts;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");
var erc20Service = new ERC20LogProcessingService(web3.Eth);

var usdcAddress = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48";

// Get all USDC transfers in block range
var transfers = await erc20Service.GetAllTransferEventsForContract(
    contractAddress: usdcAddress,
    fromBlockNumber: new BigInteger(100),
    toBlockNumber: new BigInteger(110),
    cancellationToken: CancellationToken.None
);

foreach (var transfer in transfers)
{
    Console.WriteLine($"Transfer: {transfer.Event.Value} from {transfer.Event.From} to {transfer.Event.To}");
    Console.WriteLine($"  Block: {transfer.Log.BlockNumber}, Tx: {transfer.Log.TransactionHash}");
}

From: Services/SmartContracts/ERC20LogProcessingService.cs:23-29

Example 14: Track Account Token Activity

Get all transfers involving specific account:

var tokenAddresses = new[]
{
    "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", // USDC
    "0xdAC17F958D2ee523a2206206994597C13D831ec7"  // USDT
};

var accountAddress = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb";

// Get transfers TO and FROM the account
var accountTransfers = await erc20Service.GetAllTransferEventsFromAndToAccount(
    contractAddresses: tokenAddresses,
    account: accountAddress,
    fromBlockNumber: new BigInteger(100),
    toBlockNumber: new BigInteger(110)
);

var received = accountTransfers.Count(t => t.Event.To.Equals(accountAddress, StringComparison.OrdinalIgnoreCase));
var sent = accountTransfers.Count(t => t.Event.From.Equals(accountAddress, StringComparison.OrdinalIgnoreCase));

Console.WriteLine($"Received: {received}, Sent: {sent}");

From: Services/SmartContracts/ERC20LogProcessingService.cs:47-62

Example 15: ERC721 Ownership Tracking

Track NFT ownership from transfer events:

using Nethereum.BlockchainProcessing.Services.SmartContracts;

var web3 = new Web3("https://mainnet.infura.io/v3/YOUR_KEY");
var erc721Service = new ERC721LogProcessingService(web3.Eth);

var nftContract = "0xYourNFTContract";

// Get current owners by processing all transfer events
var owners = await erc721Service.GetAllCurrentOwnersProcessingAllTransferEvents(
    contractAddress: nftContract,
    fromBlockNumber: new BigInteger(0),  // Process all history
    toBlockNumber: null  // Up to latest
);

foreach (var owner in owners)
{
    Console.WriteLine($"Token #{owner.TokenId}: Owned by {owner.Owner}");
}

From: Services/SmartContracts/ERC721LogProcessingService.cs:51-58

Example 16: Account NFT Portfolio

Get all NFTs owned by an account:

var accountAddress = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb";
var nftContract = "0xYourNFTContract";

var ownedNFTs = await erc721Service.GetErc721OwnedByAccountUsingAllTransfers(
    contractAddress: nftContract,
    account: accountAddress,
    fromBlockNumber: new BigInteger(0),
    toBlockNumber: null
);

Console.WriteLine($"Account owns {ownedNFTs.Count} NFTs:");
foreach (var nft in ownedNFTs)
{
    Console.WriteLine($"  Token #{nft.TokenId}");
}

From: Services/SmartContracts/ERC721LogProcessingService.cs:26-36

Progress Tracking

JSON File Progress Repository

Persist progress to JSON file:

using Nethereum.BlockchainProcessing.ProgressRepositories;
using System.IO;

var progressFile = "blockchain-progress.json";

var progressRepository = new JsonBlockProgressRepository(
    jsonSourceExists: async () => File.Exists(progressFile),
    jsonWriter: async (json) => await File.WriteAllTextAsync(progressFile, json),
    jsonRetriever: async () => await File.ReadAllTextAsync(progressFile),
    lastBlockProcessed: new BigInteger(0)  // Starting block if file doesn't exist
);

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(
    progressRepository,
    steps => { /* configure steps */ }
);

// Progress automatically persists to JSON file
await blockProcessor.ExecuteAsync(CancellationToken.None);

// Resume from same file on next run

From: ProgressRepositories/JsonBlockProgressRepository.cs:13-82

Custom Progress Repository

Implement IBlockProgressRepository for custom storage:

public class DatabaseProgressRepository : IBlockProgressRepository
{
    private readonly IDbConnection _connection;

    public DatabaseProgressRepository(IDbConnection connection)
    {
        _connection = connection;
    }

    public async Task UpsertProgressAsync(BigInteger blockNumber)
    {
        await _connection.ExecuteAsync(
            "UPDATE BlockchainProgress SET LastBlock = @blockNumber, UpdatedAt = @now",
            new { blockNumber = (long)blockNumber, now = DateTime.UtcNow }
        );
    }

    public async Task<BigInteger?> GetLastBlockNumberProcessedAsync()
    {
        var result = await _connection.QuerySingleOrDefaultAsync<long?>(
            "SELECT LastBlock FROM BlockchainProgress"
        );
        return result.HasValue ? new BigInteger(result.Value) : null;
    }
}

From interface: ProgressRepositories/IBlockProgressRepository.cs:5-10

Reorg Handling

The processor supports chain reorganisation detection and recovery via ChainConsistencyValidationService and IChainStateRepository.

Chain State Tracking

ChainState tracks the last known canonical block number and hash. After each block is processed, the state is updated. On startup, the stored hash is compared against the RPC node — if they differ, a reorg is detected.

using Nethereum.BlockchainProcessing.BlockStorage.Entities;
using Nethereum.BlockchainProcessing.ProgressRepositories;
using Nethereum.BlockchainProcessing.Services;

var chainStateRepo = new MyChainStateRepository(); // implements IChainStateRepository
var validator = new ChainConsistencyValidationService(web3.Eth, chainStateRepo);
validator.ReorgBuffer = 10; // rewind 10 blocks on reorg detection

try
{
    await validator.ValidateAsync(cancellationToken);
}
catch (ReorgDetectedException ex)
{
    Console.WriteLine($"Reorg detected at block {ex.LastCanonicalBlockNumber}");
    Console.WriteLine($"Rewinding to block {ex.RewindToBlockNumber}");
    // Mark non-canonical records, rewind progress, restart processing
}

From: Services/ChainConsistencyValidationService.cs

Chain ID Validation

Prevents accidental indexing of the wrong chain by comparing the RPC chain ID against the stored value:

await ChainStateValidationService.EnsureChainIdMatchesAsync(
    web3.Eth, chainStateRepositoryFactory);
// Throws InvalidOperationException if chain IDs don't match

From: Services/ChainStateValidationService.cs

ReorgBufferedBlockProgressRepository

Wraps any IBlockProgressRepository to subtract a reorg buffer from the reported last block, ensuring blocks within the buffer are always re-processed:

var innerProgress = new InMemoryBlockchainProgressRepository();
var bufferedProgress = new ReorgBufferedBlockProgressRepository(innerProgress, reorgBuffer: 12);

// If inner reports block 100 as last processed, buffered returns 88
var lastBlock = await bufferedProgress.GetLastBlockNumberProcessedAsync();

From: ProgressRepositories/ReorgBufferedBlockProgressRepository.cs

Storage System

Implementing Custom Storage

Implement IBlockchainStoreRepositoryFactory for your database:

public class MyDatabaseRepositoryFactory : IBlockchainStoreRepositoryFactory
{
    private readonly IDbConnection _connection;

    public MyDatabaseRepositoryFactory(IDbConnection connection)
    {
        _connection = connection;
    }

    public IBlockRepository CreateBlockRepository()
    {
        return new MyBlockRepository(_connection);
    }

    public ITransactionRepository CreateTransactionRepository()
    {
        return new MyTransactionRepository(_connection);
    }

    public ITransactionLogRepository CreateTransactionLogRepository()
    {
        return new MyTransactionLogRepository(_connection);
    }

    public IContractRepository CreateContractRepository()
    {
        return new MyContractRepository(_connection);
    }

    public IAddressTransactionRepository CreateAddressTransactionRepository()
    {
        return new MyAddressTransactionRepository(_connection);
    }

    public ITransactionVMStackRepository CreateTransactionVMStackRepository()
    {
        return new MyTransactionVMStackRepository(_connection);
    }
}

From: BlockStorage/Repositories/IBlockchainStoreRepositoryFactory.cs:5-11

Storage Entities

The package provides ready-to-use entity models:

Block Entity (BlockStorage/Entities/Block.cs):

BlockNumber (long), Hash, ParentHash, Nonce, Difficulty
Miner, GasUsed, GasLimit, Timestamp (long)
TransactionCount (long), BaseFeePerGas
StateRoot, ReceiptsRoot, LogsBloom, WithdrawalsRoot
BlobGasUsed, ExcessBlobGas, ParentBeaconBlockRoot (EIP-4844/4788)
RequestsHash (EIP-7685), TransactionsRoot, MixHash, Sha3Uncles
IsCanonical, IsFinalized, ChainId

Transaction Entity (BlockStorage/Entities/TransactionBase.cs):

Hash, BlockNumber (long), TransactionIndex (long)
AddressFrom, AddressTo, Value, Gas, GasPrice, GasUsed
Nonce (long), TransactionType (long), TimeStamp (long)
NewContractAddress, Failed, ReceiptHash
MaxFeePerGas, MaxPriorityFeePerGas, EffectiveGasPrice (EIP-1559)
MaxFeePerBlobGas, BlobGasUsed, BlobGasPrice (EIP-4844)
IsCanonical

TransactionLog Entity (BlockStorage/Entities/TransactionLog.cs):

TransactionHash, LogIndex (long), BlockNumber (long), Address
EventHash (Topics[0])
IndexVal1, IndexVal2, IndexVal3 (Indexed parameters)
Data (Non-indexed parameters)
IsCanonical

Contract Entity (BlockStorage/Entities/Contract.cs):

Address, Name, ABI, Code
Creator, TransactionHash

InternalTransaction Entity (BlockStorage/Entities/InternalTransaction.cs):

TransactionHash, BlockNumber, BlockHash
TraceIndex, Depth, Type (CALL, DELEGATECALL, CREATE, etc.)
AddressFrom, AddressTo, Value, Gas, GasUsed
Input, Output, Error, RevertReason
IsCanonical

TokenTransferLog Entity (BlockStorage/Entities/TokenTransferLog.cs):

TransactionHash, LogIndex, BlockNumber, BlockHash
ContractAddress, EventHash, FromAddress, ToAddress
Amount, TokenId, OperatorAddress, TokenType
IsCanonical

TokenBalance Entity (BlockStorage/Entities/TokenBalance.cs):

Address, ContractAddress, Balance, TokenType, LastUpdatedBlockNumber

TokenMetadata Entity (BlockStorage/Entities/TokenMetadata.cs):

ContractAddress, Name, Symbol, Decimals, TokenType

NFTInventory Entity (BlockStorage/Entities/NFTInventory.cs):

Address, ContractAddress, TokenId, Amount, TokenType, LastUpdatedBlockNumber

AccountState Entity (BlockStorage/Entities/AccountState.cs):

Address, Balance, Nonce, IsContract, LastUpdatedBlock

ChainState Entity (BlockStorage/Entities/ChainState.cs):

LastCanonicalBlockNumber (long?), LastCanonicalBlockHash
FinalizedBlockNumber (long?), ChainId (int?)

All numeric indexing fields (BlockNumber, LogIndex, TransactionIndex, Nonce, Timestamp, TransactionType, TransactionCount) use long. Gas and value fields (Value, Gas, GasPrice, GasUsed, Balance, Amount) remain string to preserve full uint256 precision.

Advanced Configuration

Log Processing Batch Size

Configure batch size for log retrieval:

using Nethereum.BlockchainProcessing.LogProcessing;

var logProcessor = web3.Processing.Logs.CreateProcessor(filterLog => { /* ... */ });

// Customize batch size (default: 1,000,000 blocks)
logProcessor.Orchestrator.BlockRangeRequestStrategy = new BlockRangeRequestStrategy(
    defaultNumberOfBlocksPerRequest: 10000,  // 10k blocks per batch
    retryWeight: 50  // Reduce batch size on failures
);

await logProcessor.ExecuteAsync(new BigInteger(110), startAtBlockNumberIfNotProcessed: new BigInteger(100));

From: Services/BlockchainLogProcessingService.cs:24-25

Retry Configuration

Configure retry behavior for log retrieval:

var logProcessor = web3.Processing.Logs.CreateProcessor(filterLog => { /* ... */ });

// Configure retries (default: 10 retries)
logProcessor.Orchestrator.MaxGetLogsRetries = 5;
logProcessor.Orchestrator.MaxGetLogsNullRetries = 2;

await logProcessor.ExecuteAsync(new BigInteger(110), startAtBlockNumberIfNotProcessed: new BigInteger(100));

From: LogProcessing/LogOrchestrator.cs:57-58

Parallel vs Sequential Log Processing

Choose processing strategy:

using Nethereum.BlockchainProcessing.LogProcessing;

var logProcessor = web3.Processing.Logs.CreateProcessor(filterLog => { /* ... */ });

// Sequential processing (default for most use cases)
logProcessor.Orchestrator.LogProcessStrategy = new LogProcessSequentialStrategy();

// Parallel processing (faster but uses more resources)
logProcessor.Orchestrator.LogProcessStrategy = new LogProcessParallelStrategy();

await logProcessor.ExecuteAsync(new BigInteger(110), startAtBlockNumberIfNotProcessed: new BigInteger(100));

From: LogProcessing/LogOrchestrator.cs:64, LogProcessing/ILogProcessStrategy.cs

Contract Creation Code Retrieval

Enable code retrieval for deployed contracts:

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(steps =>
{
    steps.ContractCreationStep.AddSynchronousProcessorHandler(contractCreation =>
    {
        Console.WriteLine($"Contract deployed at: {contractCreation.ContractAddress}");
        Console.WriteLine($"Code length: {contractCreation.Code?.Length ?? 0}");
    });
});

// Enable code retrieval (requires extra RPC call per contract)
blockProcessor.Orchestrator.ContractCreatedCrawlerStep.RetrieveCode = true;

await blockProcessor.ExecuteAsync(new BigInteger(110), CancellationToken.None, new BigInteger(100));

From: BlockProcessing/CrawlerSteps/ContractCreatedCrawlerStep.cs:9, 21

Performance Considerations

Block Processing Performance

Block-by-block processing:

Fetches full block data with transactions
For each transaction, fetches receipt (separate RPC call)
RPC calls: 1 (block) + N (receipts) per block
Best for: Complete blockchain indexing

Optimization tips:

Disable unused steps
Use criteria to filter early
Increase minimum confirmations to avoid reorgs
Process in batches with multiple processors

Log Processing Performance

Batch log retrieval:

Fetches all logs in block range (single RPC call)
RPC calls: Blocks / BatchSize
Best for: Event-focused applications

Batch size considerations:

Default: 1,000,000 blocks per batch
Reduce for nodes with rate limits
Increase for archive nodes
Auto-reduces on errors via BlockRangeRequestStrategy

From: LogProcessing/LogOrchestrator.cs:153-194

RPC Call Comparison

Process 1000 blocks with 20 transactions each:

Block Processing:

1000 block fetches
20,000 receipt fetches
Total: 21,000 RPC calls

Log Processing (batch):

1 log fetch (if 1000 blocks < batch size)
Total: 1 RPC call

Recommendation: Use LogOrchestrator for event tracking, BlockCrawlOrchestrator for complete data.

Error Handling

Orchestrator Error Handling

Orchestrators return error status:

var progress = await blockProcessor.Orchestrator.ProcessAsync(
    fromNumber: new BigInteger(100),
    toNumber: new BigInteger(110),
    cancellationToken: CancellationToken.None
);

if (progress.HasErrored)
{
    Console.WriteLine($"Error processing block {progress.BlockNumberProcessTo}:");
    Console.WriteLine(progress.Exception.Message);

    // Can resume from failed block
    await blockProcessor.ExecuteAsync(
        toBlockNumber: new BigInteger(110),
        cancellationToken: CancellationToken.None,
        startAtBlockNumberIfNotProcessed: progress.BlockNumberProcessTo
    );
}

From: Orchestrator/OrchestrationProgress.cs:6-11

Handler Error Handling

Wrap handlers in try-catch for graceful error handling:

var blockProcessor = web3.Processing.Blocks.CreateBlockProcessor(steps =>
{
    steps.TransactionStep.AddProcessorHandler(async tx =>
    {
        try
        {
            await ProcessTransactionAsync(tx);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Error processing transaction {tx.Transaction.TransactionHash}: {ex.Message}");
            // Log error, send alert, etc.
        }
    });
});

Log Processing Retry

Log processing has built-in retry with exponential backoff:

// Automatically retries on RPC errors
// Reduces batch size on repeated failures
// Configured via MaxGetLogsRetries and BlockRangeRequestStrategy

From: LogProcessing/LogOrchestrator.cs:153-194

Use Cases

Blockchain Explorer

Build complete blockchain indexer:

var repositoryFactory = CreateDatabaseRepositoryFactory();

var processor = web3.Processing.Blocks.CreateBlockStorageProcessor(
    repositoryFactory,
    minimumBlockConfirmations: 12,
    configureSteps: steps =>
    {
        // Index all data
        // Storage handlers are automatically added

        // Add custom indexing
        steps.TransactionStep.AddProcessorHandler(async tx =>
        {
            await UpdateAddressBalanceCache(tx);
        });
    }
);

// Process continuously
await processor.ExecuteAsync(cancellationToken);

DEX Event Tracker

Track Uniswap swaps:

var swapEvents = new List<EventLog<SwapEventDTO>>();

var logProcessor = web3.Processing.Logs.CreateProcessor<SwapEventDTO>(
    swapEvent =>
    {
        var pool = swapEvent.Log.Address;
        var swap = swapEvent.Event;

        Console.WriteLine($"Swap on {pool}:");
        Console.WriteLine($"  Amount0In: {swap.Amount0In}");
        Console.WriteLine($"  Amount1Out: {swap.Amount1Out}");

        swapEvents.Add(swapEvent);
    }
);

await logProcessor.ExecuteAsync(toBlockNumber, startAtBlockNumberIfNotProcessed: fromBlockNumber);

Token Balance Tracker

Track token balances for addresses:

var balances = new Dictionary<string, BigInteger>();

var erc20Service = new ERC20LogProcessingService(web3.Eth);

var transfers = await erc20Service.GetAllTransferEventsForContract(
    usdcAddress,
    fromBlockNumber,
    toBlockNumber,
    CancellationToken.None
);

foreach (var transfer in transfers)
{
    var from = transfer.Event.From;
    var to = transfer.Event.To;
    var value = transfer.Event.Value;

    balances[from] = (balances.GetValueOrDefault(from)) - value;
    balances[to] = (balances.GetValueOrDefault(to)) + value;
}

foreach (var (address, balance) in balances)
{
    Console.WriteLine($"{address}: {balance}");
}

NFT Transfer Monitor

Monitor NFT transfers in real-time:

var progressRepo = new JsonBlockProgressRepository(/* ... */);

var logProcessor = web3.Processing.Logs.CreateProcessor<TransferEventDTO>(
    progressRepo,
    transferEvent =>
    {
        Console.WriteLine($"NFT Transfer:");
        Console.WriteLine($"  Token: {transferEvent.Event.TokenId}");
        Console.WriteLine($"  From: {transferEvent.Event.From}");
        Console.WriteLine($"  To: {transferEvent.Event.To}");
        Console.WriteLine($"  Tx: {transferEvent.Log.TransactionHash}");

        // Send notification, update database, etc.
    }
);

// Run continuously
await logProcessor.ExecuteAsync(CancellationToken.None);

Token Transfer Processing

The TokenTransferLogProcessingService indexes ERC-20, ERC-721, and ERC-1155 transfer events into ITokenTransferLogRepository with a unified filter that matches all three token standards in a single log query.

using Nethereum.BlockchainProcessing.Services.SmartContracts;

var tokenService = new TokenTransferLogProcessingService(
    web3.Processing.Logs, web3.Eth);

var processor = tokenService.CreateProcessor(
    transferLogRepository, blockProgressRepository,
    numberOfBlocksPerRequest: 1000);

await processor.ExecuteAsync(cancellationToken);

From: Services/SmartContracts/TokenTransferLogProcessingService.cs

Token Balance Aggregation

The TokenBalanceAggregationService reads stored TokenTransferLog records and maintains running TokenBalance and NFTInventory tables:

var aggregationService = new TokenBalanceAggregationService(
    transferLogRepository, balanceRepository, nftRepository, progressRepository);

await aggregationService.AggregateAsync(fromBlock, toBlock, cancellationToken);

From: Services/SmartContracts/TokenBalanceAggregationService.cs

Internal Transaction Processing

The InternalTransactionPostProcessor orchestrates trace-based internal transaction indexing. It accepts a trace provider function (e.g., debug_traceTransaction) and stores results via IInternalTransactionRepository:

var postProcessor = new InternalTransactionPostProcessor(
    internalTransactionRepository,
    traceProvider: async txHash => await GetTracesFromRpc(txHash),
    getContractTransactionsInRange: async (from, to) => await GetContractTxs(from, to),
    progressRepository, lastConfirmedBlockService);

await postProcessor.ExecuteAsync(cancellationToken);

From: Services/InternalTransactionPostProcessor.cs

Metrics and Observability

The ILogProcessingObserver interface enables telemetry integration. The built-in LogProcessingMetrics implementation uses System.Diagnostics.Metrics (net8.0+) for OpenTelemetry-compatible instrumentation:

using Nethereum.BlockchainProcessing.Metrics;

var metrics = new LogProcessingMetrics(
    chainId: "1", processorType: "TokenTransfers", name: "MyApp");

// Pass to log processing service
var processor = logProcessingService.CreateProcessor(
    transferLogRepository, blockProgressRepository,
    observer: metrics);

// Emitted metrics:
// logprocessing.blocks.processed   - counter
// logprocessing.logs.processed     - counter
// logprocessing.errors             - counter
// logprocessing.reorgs             - counter
// logprocessing.getlogs.retries    - counter
// logprocessing.batch.duration     - histogram
// logprocessing.last_block         - gauge
// logprocessing.lag                - gauge (blocks behind chain head)

From: Metrics/LogProcessingMetrics.cs

RetryRunner

RetryRunner.RunWithExponentialBackoffAsync provides resilient execution with exponential backoff for long-running processing loops:

using Nethereum.BlockchainProcessing;

await RetryRunner.RunWithExponentialBackoffAsync(
    async ct =>
    {
        await processor.ExecuteAsync(ct);
    },
    cancellationToken,
    onRetry: (ex, attempt, delay) =>
        logger.LogError(ex, "Processing failed (attempt {Attempt}), retrying in {Delay}s", attempt, delay),
    initialDelaySeconds: 5,
    maxDelaySeconds: 300);

From: RetryRunner.cs

Non-Canonical Record Management

Repository interfaces for marking records as non-canonical during reorg recovery:

INonCanonicalBlockRepository — MarkNonCanonicalAsync(BigInteger blockNumber)
INonCanonicalTransactionRepository — MarkNonCanonicalAsync(BigInteger blockNumber)
INonCanonicalTransactionLogRepository — MarkNonCanonicalAsync(BigInteger blockNumber)
INonCanonicalTokenTransferLogRepository — MarkNonCanonicalAsync(BigInteger blockNumber)
IReorgHandler — composite interface combining all non-canonical operations with HandleReorgAsync(BigInteger fromBlock)

From: BlockStorage/Repositories/INonCanonical*.cs, BlockStorage/Repositories/IReorgHandler.cs

Dependencies

Required packages:

Nethereum.Hex - Hex conversions
Nethereum.JsonRpc.RpcClient - RPC client
Nethereum.RPC - RPC DTOs and services
Nethereum.Util - Utility functions
Nethereum.Contracts - Contract interaction and event decoding

Source Files Reference

Core Processing:

BlockchainProcessor.cs - Main processor
BlockchainCrawlingProcessor.cs - Block crawling processor
Orchestrator/IBlockchainProcessingOrchestrator.cs - Orchestrator interface

Block Processing:

BlockProcessing/BlockProcessingSteps.cs - Pipeline steps
BlockProcessing/BlockCrawlOrchestrator.cs - Block crawling orchestrator
BlockProcessing/CrawlerSteps/*.cs - Data fetchers

Log Processing:

LogProcessing/LogOrchestrator.cs - Log batch processor
LogProcessing/BlockRangeRequestStrategy.cs - Batch sizing strategy

Processor Framework:

Processor/IProcessor.cs - Processor interface
Processor/Processor.cs - Generic processor
Processor/ProcessorHandler.cs - Handler wrapper

Storage:

BlockStorage/Repositories/IBlockchainStoreRepositoryFactory.cs - Repository factory
BlockStorage/Entities/*.cs - Storage entities
BlockStorage/BlockStorageProcessingSteps.cs - Storage handlers

Progress:

ProgressRepositories/IBlockProgressRepository.cs - Progress interface
ProgressRepositories/JsonBlockProgressRepository.cs - JSON persistence
ProgressRepositories/IChainStateRepository.cs - Chain state tracking
ProgressRepositories/ReorgBufferedBlockProgressRepository.cs - Reorg-buffered progress

Services:

Services/BlockchainProcessingService.cs - Service entry point
Services/BlockchainLogProcessingService.cs - Log processing service
Services/ChainConsistencyValidationService.cs - Reorg detection
Services/ChainStateValidationService.cs - Chain ID validation
Services/InternalTransactionPostProcessor.cs - Trace-based internal transaction indexing
Services/SmartContracts/ERC20LogProcessingService.cs - ERC20 utilities
Services/SmartContracts/ERC721LogProcessingService.cs - ERC721 utilities
Services/SmartContracts/TokenTransferLogProcessingService.cs - Unified token transfer indexing
Services/SmartContracts/TokenBalanceAggregationService.cs - Balance aggregation from transfers

Metrics:

Metrics/ILogProcessingObserver.cs - Observer interface
Metrics/LogProcessingMetrics.cs - OpenTelemetry metrics implementation

Infrastructure:

RetryRunner.cs - Exponential backoff retry runner

Related Packages

Nethereum.Web3 - Ethereum client library
Nethereum.Contracts - Smart contract interaction
Nethereum.RPC - RPC infrastructure

Support

GitHub: https://github.com/Nethereum/Nethereum
Documentation: https://docs.nethereum.com
Discord: https://discord.gg/jQPrR58FxX

Name		Name	Last commit message	Last commit date
parent directory ..
BlockProcessing		BlockProcessing
BlockStorage		BlockStorage
LogProcessing		LogProcessing
Metrics		Metrics
Orchestrator		Orchestrator
Processor		Processor
ProgressRepositories		ProgressRepositories
Properties		Properties
Services		Services
BlockchainCrawlingProcessor.cs		BlockchainCrawlingProcessor.cs
BlockchainProcessor.cs		BlockchainProcessor.cs
Nethereum.BlockchainProcessing.csproj		Nethereum.BlockchainProcessing.csproj
README.md		README.md
RetryRunner.cs		RetryRunner.cs

FilesExpand file tree

Nethereum.BlockchainProcessing

Directory actions

More options

Directory actions

More options

Latest commit

History

Nethereum.BlockchainProcessing

Folders and files

parent directory

README.md

Nethereum.BlockchainProcessing

Overview

Installation

Core Architecture

Core Components

BlockchainProcessor

BlockCrawlOrchestrator

BlockProcessingSteps

Processor

Usage Examples

Example 1: Basic Block Processing

Example 2: Processing with Progress Tracking

Example 3: Block Confirmations

Example 4: Filtering with Criteria

Example 5: Disabling Processing Steps

Example 6: Block Storage Processor

Example 7: Custom Storage Configuration

Log Processing

Example 8: Process All Logs

Example 9: Process Specific Event Type

Example 10: Event Processing with Criteria

Example 11: Process Contract-Specific Events

Example 12: Process Multiple Contracts

ERC20/ERC721 Processing

Example 13: ERC20 Transfer Processing

Example 14: Track Account Token Activity

Example 15: ERC721 Ownership Tracking

Example 16: Account NFT Portfolio

Progress Tracking

JSON File Progress Repository

Custom Progress Repository

Reorg Handling

Chain State Tracking

Chain ID Validation

ReorgBufferedBlockProgressRepository

Storage System

Implementing Custom Storage

Storage Entities

Advanced Configuration

Log Processing Batch Size

Retry Configuration

Parallel vs Sequential Log Processing

Contract Creation Code Retrieval

Performance Considerations

Block Processing Performance

Log Processing Performance

RPC Call Comparison

Error Handling

Orchestrator Error Handling

Handler Error Handling

Log Processing Retry

Use Cases

Blockchain Explorer

DEX Event Tracker

Token Balance Tracker

NFT Transfer Monitor

Token Transfer Processing

Token Balance Aggregation

Internal Transaction Processing

Metrics and Observability

RetryRunner

Non-Canonical Record Management

Dependencies

Source Files Reference

Related Packages

Support