Blockchain Chain Integration

Guide for integrating new blockchain networks into Graph Node

This guide explains how to integrate a new blockchain network into Graph Node. Chain integration involves implementing the Blockchain trait and providing chain-specific adapters for data ingestion and trigger processing.

Overview

Graph Node uses a modular architecture that allows different blockchain networks to be integrated by implementing a set of core traits. Each chain integration lives in the chain/ directory and provides:

• Data source definitions
• Trigger types and filtering
• Block stream implementation
• Runtime adapters for contract calls
• Block ingestion from RPC or Firehose

Architecture

The chain integration architecture consists of several key components:

Blockchain → Chain Adapter → Block Stream → Trigger Processing → Runtime → Store

1. Chain Adapter: Connects to blockchain nodes and converts data
2. Block Stream: Provides event-driven streaming of blocks
3. Trigger Processing: Matches blockchain events to subgraph handlers
4. Runtime: Executes subgraph code in WASM sandbox
5. Store: Persists entities with block-level granularity

Project Structure

Each chain integration follows this structure:

chain/
├── <chain-name>/
│   ├── src/
│   │   ├── chain.rs          # Main Chain struct and Blockchain trait impl
│   │   ├── adapter.rs        # Triggers adapter and filtering
│   │   ├── data_source.rs    # Data source types
│   │   ├── trigger.rs        # Trigger data types
│   │   ├── codec.rs          # Protocol buffer codecs
│   │   └── runtime/
│   │       ├── mod.rs        # Runtime adapter
│   │       └── abi.rs        # ABI handling
│   ├── Cargo.toml
│   └── build.rs              # Build script for protobufs
└── common/
    └── src/
        └── lib.rs            # Shared utilities

Core Traits

Blockchain Trait

The Blockchain trait is the primary interface that all chains must implement:

chain/src/chain.rs

use graph::blockchain::{Blockchain, BlockchainKind};
use graph::prelude::*;

#[async_trait]
impl Blockchain for Chain {
    // Unique identifier for this blockchain
    const KIND: BlockchainKind = BlockchainKind::Ethereum;
    
    // Aliases for backward compatibility
    const ALIASES: &'static [&'static str] = &["ethereum/contract"];
    
    // Associated types
    type Client = EthereumNetworkAdapters;
    type Block = BlockFinality;
    type DataSource = DataSource;
    type UnresolvedDataSource = UnresolvedDataSource;
    type DataSourceTemplate = DataSourceTemplate;
    type UnresolvedDataSourceTemplate = UnresolvedDataSourceTemplate;
    type TriggerData = EthereumTrigger;
    type MappingTrigger = MappingTrigger;
    type TriggerFilter = TriggerFilter;
    type NodeCapabilities = NodeCapabilities;
    type DecoderHook = DecoderHook;
    
    // Required methods
    fn triggers_adapter(
        &self,
        loc: &DeploymentLocator,
        capabilities: &Self::NodeCapabilities,
        unified_api_version: UnifiedMappingApiVersion,
    ) -> Result<Arc<dyn TriggersAdapterTrait<Self>>, Error>;
    
    async fn new_block_stream(
        &self,
        deployment: DeploymentLocator,
        store: impl DeploymentCursorTracker,
        start_blocks: Vec<BlockNumber>,
        filter: Arc<TriggerFilterWrapper<Self>>,
        unified_api_version: UnifiedMappingApiVersion,
    ) -> Result<Box<dyn BlockStream<Self>>, Error>;
    
    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error>;
    
    async fn block_pointer_from_number(
        &self,
        logger: &Logger,
        number: BlockNumber,
    ) -> Result<BlockPtr, IngestorError>;
    
    fn is_refetch_block_required(&self) -> bool;
    
    async fn refetch_firehose_block(
        &self,
        logger: &Logger,
        cursor: FirehoseCursor,
    ) -> Result<Self::Block, Error>;
    
    async fn runtime(
        &self,
    ) -> Result<(Arc<dyn RuntimeAdapterTrait<Self>>, Self::DecoderHook)>;
    
    fn chain_client(&self) -> Arc<ChainClient<Self>>;
    
    async fn block_ingestor(&self) -> Result<Box<dyn BlockIngestor>>;
}

Block Trait

Blocks must implement the Block trait:

chain/src/chain.rs

impl Block for BlockFinality {
    fn ptr(&self) -> BlockPtr {
        // Return block pointer (hash + number)
    }
    
    fn parent_ptr(&self) -> Option<BlockPtr> {
        // Return parent block pointer
    }
    
    fn data(&self) -> Result<json::Value, json::Error> {
        // Serialize block data
    }
    
    fn timestamp(&self) -> BlockTime {
        // Return block timestamp
    }
}

Implementation Steps

Extension Points

Custom Node Capabilities

Define chain-specific node capabilities:

src/capabilities.rs

use graph::blockchain::NodeCapabilities as NodeCapabilitiesTrait;

#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
pub struct NodeCapabilities {
    pub archive_mode: bool,
    pub traces: bool,
}

impl NodeCapabilitiesTrait<Chain> for NodeCapabilities {}

impl fmt::Display for NodeCapabilities {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "archive: {}, traces: {}", self.archive_mode, self.traces)
    }
}

Custom Decoder Hooks

Implement decoder hooks for preprocessing data:

src/data_source.rs

pub struct DecoderHook {
    client: Arc<ChainClient>,
}

impl DecoderHook {
    pub async fn decode_transaction(
        &self,
        tx_hash: &str,
    ) -> Result<Transaction, Error> {
        // Custom decoding logic
    }
}

Subgraph Manifest Schema

Define the YAML schema for your chain:

Example subgraph.yaml

specVersion: 0.0.8
schema:
  file: ./schema.graphql
dataSources:
  - kind: mychain
    name: MyContract
    network: mainnet
    source:
      address: "0x123..."
      startBlock: 1000000
    mapping:
      kind: mychain/events
      apiVersion: 0.0.8
      language: wasm/assemblyscript
      entities:
        - Transfer
      eventHandlers:
        - event: Transfer(address,address,uint256)
          handler: handleTransfer
      file: ./src/mapping.ts

Testing

Unit Tests

Write unit tests for core components:

src/adapter.rs

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_trigger_filter() {
        let filter = TriggerFilter::default();
        // Test filter logic
    }
    
    #[graph::test]
    async fn test_triggers_adapter() {
        let adapter = MyChainTriggersAdapter::new(...);
        let triggers = adapter.scan_triggers(0, 100, &filter).await.unwrap();
        assert!(!triggers.is_empty());
    }
}

Integration Tests

Create integration tests in tests/integration-tests/:

# Run integration tests for your chain
TEST_CASE=mychain_basic just test-integration

Runner Tests

Create runner tests in store/test-store/tests/chain/mychain/:

# Run runner tests
just test-runner mychain

Examples

Ethereum Implementation

See chain/ethereum/ for a full-featured reference implementation with:

• RPC and Firehose support
• Event, call, and block handlers
• Call caching
• Reorg handling
• Trace support

NEAR Implementation

See chain/near/ for a simpler Firehose-only implementation:

• Receipt-based triggers
• No dynamic data sources
• Account filtering (exact and partial matches)

Common Patterns

Handling Reorgs

impl TriggersAdapter<Chain> for MyChainAdapter {
    async fn is_on_main_chain(&self, ptr: BlockPtr) -> Result<bool, Error> {
        // Fetch block at same height from current chain head
        let current = self.client.block_by_number(ptr.number).await?;
        
        // Compare hashes
        Ok(current.hash == ptr.hash)
    }
}

Block Caching

impl TriggersAdapter<Chain> for MyChainAdapter {
    async fn ancestor_block(
        &self,
        ptr: BlockPtr,
        offset: BlockNumber,
    ) -> Result<Option<Block>, Error> {
        // Try cache first
        if let Some(block) = self.chain_store.ancestor_block(ptr, offset).await? {
            return Ok(Some(block));
        }
        
        // Fall back to fetching from node
        self.fetch_ancestor(ptr, offset).await
    }
}

Configuration

Add chain configuration to config.toml:

config.toml

[chains.mychain]
provider = [
  { label = "mychain-mainnet", url = "https://rpc.mychain.io", features = ["archive", "traces"] }
]

[chains.mychain.firehose]
endpoint = "mychain-firehose.example.com:443"

Best Practices

• Error Handling: Use Result types and provide context with anyhow
• Async/Await: Use async_trait for trait implementations
• Logging: Use structured logging with slog
• Caching: Cache frequently accessed data (blocks, receipts)
• Metrics: Add Prometheus metrics for monitoring
• Testing: Write comprehensive unit and integration tests
• Documentation: Document chain-specific behavior and limitations

Performance Optimization

Batch Loading

impl TriggersAdapter<Chain> for MyChainAdapter {
    async fn scan_triggers(
        &self,
        from: BlockNumber,
        to: BlockNumber,
        filter: &TriggerFilter,
    ) -> Result<(Vec<BlockWithTriggers<Chain>>, BlockNumber), Error> {
        // Batch load blocks
        let blocks = self.client.load_blocks_batch(from, to).await?;
        
        // Process in parallel
        let results = blocks
            .into_iter()
            .map(|block| self.extract_triggers(&block, filter))
            .collect();
        
        Ok((results, to))
    }
}

Parallel Processing

Use FuturesOrdered for parallel processing:

use futures03::stream::FuturesOrdered;
use futures03::TryStreamExt;

let futures: FuturesOrdered<_> = block_numbers
    .into_iter()
    .map(|num| self.client.load_block(num))
    .collect();

let blocks = futures.try_collect::<Vec<_>>().await?;

Debugging

Enable Debug Logging

RUST_LOG=graph_chain_mychain=debug graph-node ...

Common Issues

Block not found: Ensure block ingestion is running and synced

Trigger mismatch: Check filter implementation and data source matching logic

Reorg issues: Verify is_on_main_chain and ancestor_block implementations

See Also

• Adding API Versions
• Graph Node Architecture
• Runtime Development
• Firehose Integration