Project: EVM Blockchain Indexer
Meet XQuery, the generic and high-performance EVM-chain indexer I developed a few months back. Engineered to be multi-threaded and resource-efficient, this indexer specializes in the intricate task of indexing EVM smart contract events.
Inspired by The Graph project, XQuery empowers users with the flexibility to implement adaptable indexing strategies. These strategies enable the querying of data not directly available on the blockchain. Indexers like XQuery play a vital role in the web3 ecosystem, enabling developers to seamlessly build decentralized applications (dApps) that cater to the specific needs of users.
Key Features of XQuery:
-
Written in Python: XQuery leverages the Python programming language, known for its readability and versatility, making it accessible to a broad user base.
-
Scalable: XQuery adopts a multi-process architecture, enabling parallel indexing and processing of data. This scalability ensures optimal performance, especially when dealing with large datasets.
-
Flexible: XQuery offers support for any EVM (Ethereum Virtual Machine) based blockchain, providing users with the flexibility to apply its capabilities across various blockchain networks.
-
Customizable: XQuery prides itself on being dynamic and versatile, allowing users to configure and customize its functionalities according to their specific requirements.
What is Blockchain Indexing?
Blockchain's decentralized and distributed storage presents a challenge in swiftly accessing and retrieving data. Unlike traditional databases, on-chain data is not readily searchable by default. This becomes more evident as the volume of blockchain data continues to expand, leading to noticeable performance issues.
To overcome these limitations, indexers play a vital role in the blockchain ecosystem. These tools extract transaction data from a blockchain node and efficiently load it into a database or another service. This process enables seamless and easy querying of stored information, simplifying the retrieval of data that is inherently difficult to access directly from the blockchain.
In essence, blockchain indexing is a crucial tool that not only simplifies data access but also tackles performance challenges, making blockchain data more accessible and user-friendly for a diverse range of applications and users.
How does XQuery work?
XQuery processes blockchain data in three customizable steps:
1. Data Acquisition - "Filter"
The initial stage involves the acquisition of relevant data from any blockchain source, be it a local file, RPC (Remote Procedure Call), or a data stream. The emphasis here is on optimizing fetching performance by empowering users to filter events based on specific topics, ensuring the extraction of precisely the data needed.
2. Data Indexing - "Indexer"
In the second stage, XQuery performs actions on the fetched blockchain data (events) to extract pertinent information. The primary objective of this indexing step is to gather all "external" data and efficiently move it to the database or memory for further processing.
3. Data Processing - "Processor"
The final stage focuses on post-processing and aggregating the indexed data. Its objective is to calculate and aggregate supplementary information derived from the indexed data, providing users with a comprehensive and refined dataset.
Example Strategy
In this simple example, we'll demonstrate how to implement a strategy using the XQuery framework to index the Transfer events of an ERC-20 token contract.
For a practical demonstration, let's choose a low-traffic token contract from the Avalanche (AVAX) blockchain:
- Address: 0x08287930ca952673B02D3B70eEc2893Dd7846743
- Decimals: 8
- Deployment Block: 18501919
The next sections will guide you through the implementation of the necessary components: a database model, event filter, indexer, and processor.
Database Model
The database table serves as a repository for storing indexed blockchain data.
Creating a new database table involves inheriting from the Base class.
In this demonstration, we'll focus on key attributes: transaction hash, sender and receiver addresses, and the transferred amount.
from sqlalchemy import (
Column,
String,
Numeric,
)
from .base import (
Base,
BaseModel,
)
class Transfer(BaseModel, Base):
__tablename__ = "transfer"
tx_hash = Column(String(length=66), nullable=False)
sender = Column(String(length=42), nullable=False)
receiver = Column(String(length=42), nullable=False)
amount = Column(Numeric(precision=78, scale=8), nullable=False)
Event Filter
The event filter is responsible for gathering relevant data from the blockchain.
It's crucial to emphasize that any filter implementation must inherit from the EventFilter base class.
In the provided example, the filter utilizes the web3 eth_getLogs RPC method to fetch events from a specific contract.
To refine the data retrieval process, a topic filter is created and applied, ensuring that only events related to transfers are returned.
from typing import List
import operator
from web3 import Web3
from web3.datastructures import AttributeDict
from web3.types import LogReceipt
from web3._utils.filters import construct_event_topic_set
from xquery.types import ExtendedLogReceipt
from xquery.util import convert
from .filter import EventFilter
# Partial ERC-20 ABI
ABI = """[
{
"anonymous": false,
"inputs": [
{
"indexed": true,
"name": "from",
"type": "address"
},
{
"indexed": true,
"name": "to",
"type": "address"
},
{
"indexed": false,
"name": "value",
"type": "uint256"
}
],
"name": "Transfer",
"type": "event"
}
]
"""
class EventFilterExample(EventFilter):
def __init__(self, w3: Web3) -> None:
self._contract = w3.eth.contract(
address="0x08287930ca952673B02D3B70eEc2893Dd7846743",
abi=ABI
)
super().__init__(w3, [self._contract.events.Transfer])
abi_transfer = self._contract.events.Transfer._get_event_abi()
self._topic = construct_event_topic_set(
event_abi=abi_transfer,
abi_codec=self.w3.codec,
)
@staticmethod
def _fix_attrdict(attrdicts: List[AttributeDict]) -> List[AttributeDict]:
for a in attrdicts:
a.__dict__ = convert(a.__dict__)
return attrdicts
def _get_logs(self, from_block: int, to_block: int) -> List[LogReceipt]:
assert from_block <= to_block
logs = set()
entries = self.w3.eth.get_logs({
"fromBlock": hex(from_block),
"toBlock": hex(to_block),
"address": self._contract.address,
"topics": [
self._topic,
],
})
logs.update(self.__class__._fix_attrdict(entries))
return sorted(logs, key=operator.itemgetter("blockNumber", "logIndex"))
def get_logs(self, from_block: int, chunk_size: int) -> List[ExtendedLogReceipt]:
assert chunk_size > 0
logs = self._get_logs(from_block, from_block + chunk_size - 1)
self._decode_event_data(logs)
self._add_event_name(logs)
return logs # type: ignore
Event Indexer
The indexer is responsible for parsing the filtered events and extracting meaningful information.
This basic indexer is designed to simply map Transfer events to Transfer database objects.
An indexer implementation should always inherit from the EventIndexer base class.
from typing import List
import logging
from web3 import Web3
import xquery.cache
import xquery.db
import xquery.db.orm as orm
from xquery.types import ExtendedLogReceipt
from xquery.util import token_to_decimal
from .indexer import EventIndexer
log = logging.getLogger(__name__)
class EventIndexerExample(EventIndexer):
def __init__(
self,
w3: Web3,
db: xquery.db.FusionSQL,
cache: xquery.cache.Cache,
) -> None:
super().__init__(w3, db, cache)
def _handle_transfer(self, entry: ExtendedLogReceipt) -> List[orm.Base]:
return [orm.Transfer(
tx_hash=entry.transactionHash.hex(),
sender=Web3.to_checksum_address(entry.dataDecoded["from"]),
receiver=Web3.to_checksum_address(entry.dataDecoded.to),
amount=token_to_decimal(entry.dataDecoded.value, 8),
)]
@classmethod
def setup(cls, w3: Web3, db: xquery.db.FusionSQL, start_block: int) -> List[orm.Base]:
return []
def process(self, entry: ExtendedLogReceipt) -> List[orm.Base]:
objects = []
if entry.name == "Transfer":
result = self._handle_transfer(entry)
objects.extend(result)
else:
log.warning(f"Encountered unknown event '{entry.name}'")
return objects
Runner
Now, let's assemble all the components and set XQuery into motion.
The run script takes care of initializing connections to the web3 node, PostgreSQL database and Redis cache.
Subsequently, the XQuery controller is executed with the EventFilterExample filter and EventIndexerExample indexer that we implemented earlier.
For the sake of simplicity, we are exclusively mapping events without delving into any post-processing.
Hence, the dummy EventProcessorDummy processor is employed.
from web3 import Web3
from web3.middleware import geth_poa_middleware
import xquery.cache
import xquery.controller
import xquery.db
from xquery.config import CONFIG as C
from xquery.event import (
EventFilterExample,
EventIndexerExample,
EventProcessorDummy,
)
from xquery.provider import BatchHTTPProvider
def main() -> int:
w3 = Web3(BatchHTTPProvider(endpoint_uri=C["API_URL"], request_kwargs={"timeout": 120}))
w3.middleware_onion.remove("gas_price_strategy")
w3.middleware_onion.remove("gas_estimate")
w3.middleware_onion.inject(geth_poa_middleware, "geth_poa", layer=0)
db = xquery.db.FusionSQL(
conn=xquery.db.build_url(
driver=C["DB_DRIVER"],
host=C["DB_HOST"],
port=C["DB_PORT"],
username=C["DB_USERNAME"],
password=C["DB_PASSWORD"],
database=C["DB_DATABASE"],
),
verbose=C["DB_DEBUG"],
)
cache = xquery.cache.Cache_Redis(
host=C["REDIS_HOST"],
port=C["REDIS_PORT"],
password=C["REDIS_PASSWORD"],
db=C["REDIS_DATABASE"],
)
# select the event indexer class/type
# Note: will be instantiated in the worker process and therefore needs to be passed as type
indexer_cls = EventIndexerExample
# create an event filter
event_filter = EventFilterExample(w3=w3)
# create an event processor
# Note: the actual processor stages will be instantiated in the worker process
event_processor = EventProcessorDummy()
with xquery.controller.Controller(w3=w3, db=db, cache=cache, indexer_cls=indexer_cls, num_workers=8) as c:
c.run(
start_block=18501919,
end_block="latest",
num_safety_blocks=10,
filter_=event_filter,
processor=event_processor,
chunk_size=2048,
target_sleep_time=30,
)
return 0
if __name__ == "__main__":
main()
Results
On public infrastructure with substantial rate limiting, it takes approximately 5-10 minutes to index all Transfer events of the example contract.
However, on a private setup, this process is completed much faster, usually in less than 30 seconds.
It's important to note that contracts with a higher volume of events would naturally increase both indexing and processing times.