feat: Add conditional indexing to filter Safes by initiator address (#2703)

Implement selective Safe indexing on L2 networks to reduce database size and improve indexing speed by filtering out Safes created by specific initiators.

Changes:
- Add ETH_EVENTS_IGNORED_INITIATORS setting to define blocklisted initiators
- Override process_elements in SafeEventsIndexer to filter by tx._from before storing EthereumTx records
- Optimize RPC calls: fetch transactions first, filter, then fetch receipts only for allowed transactions
- Add comprehensive tests for conditional indexing

Closes #2703

Author: Uxio0

config/settings/base.py

@@ -609,6 +609,11 @@
 ETH_ERC20_LOAD_ADDRESSES_CHUNK_SIZE = env.int(
     "ETH_ERC20_LOAD_ADDRESSES_CHUNK_SIZE", default=500_000
 )  # Load Safe addresses for the ERC20 indexer with a database iterator with the defined `chunk_size`
+ETH_EVENTS_IGNORED_INITIATORS: set[ChecksumAddress] = {
+    ChecksumAddress(HexAddress(HexStr(address)))
+    for address in env.list("ETH_EVENTS_IGNORED_INITIATORS", default=[])
+}  # Initiator addresses whose created Safes should be ignored during L2 indexing
+
 
 # ENABLE/DISABLE COLLECTIBLES DOWNLOAD METADATA, enable=True, disabled by default
 COLLECTIBLES_ENABLE_DOWNLOAD_METADATA = env.bool(

safe_transaction_service/history/indexers/safe_events_indexer.py

@@ -1,7 +1,8 @@
 import datetime
+from collections import OrderedDict
+from collections.abc import Sequence
 from functools import cached_property
 from logging import getLogger
-from typing import Any
 
 from django.conf import settings
 
@@ -20,14 +21,16 @@
 )
 from safe_eth.util.util import to_0x_hex_str
 from web3.contract.contract import ContractEvent
-from web3.types import EventData
+from web3.types import EventData, LogReceipt, TxData, TxReceipt
 
 from ..models import (
     EthereumBlock,
+    EthereumTx,
     EthereumTxCallType,
     InternalTx,
     InternalTxDecoded,
     InternalTxType,
+    SafeContract,
     SafeMasterCopy,
     SafeRelevantTransaction,
 )
@@ -65,11 +68,133 @@ def __init__(self, *args, **kwargs):
         kwargs.setdefault(
             "eth_zksync_compatible_network", settings.ETH_ZKSYNC_COMPATIBLE_NETWORK
         )
+        kwargs.setdefault("ignored_initiators", settings.ETH_EVENTS_IGNORED_INITIATORS)
+
         self.eth_zksync_compatible_network = kwargs["eth_zksync_compatible_network"]
+        self.ignored_initiators = kwargs["ignored_initiators"]
+        self.conditional_indexing_enabled = bool(self.ignored_initiators)
         # Cache timestamp for block hashes
         self.block_hashes_with_timestamp: dict[bytes, datetime.datetime] = {}
         super().__init__(*args, **kwargs)
 
+    def process_elements(self, log_receipts: Sequence[LogReceipt]) -> list[InternalTx]:
+        """
+        Override to filter events by tx._from when conditional indexing is enabled.
+        This avoids storing EthereumTx in database for blocklisted initiators.
+        """
+        if not log_receipts:
+            return []
+
+        if not self.ignored_initiators:
+            # No blocklist configured, use standard flow
+            return super().process_elements(log_receipts)
+
+        # --- Conditional indexing enabled ---
+        logger.debug("Conditional indexing: filtering events by tx._from")
+
+        # 1. Filter already processed log receipts
+        not_processed_log_receipts = [
+            log_receipt
+            for log_receipt in log_receipts
+            if not self.element_already_processed_checker.is_processed(
+                log_receipt["transactionHash"],
+                log_receipt["blockHash"],
+                log_receipt["logIndex"],
+            )
+        ]
+
+        if not not_processed_log_receipts:
+            return []
+
+        # 2. Get unique tx_hashes preserving order
+        tx_hashes = list(
+            OrderedDict.fromkeys(
+                HexBytes(r["transactionHash"]) for r in not_processed_log_receipts
+            ).keys()
+        )
+
+        # 3. Check DB for existing txs
+        db_txs: dict[bytes, EthereumTx] = {
+            HexBytes(tx.tx_hash): tx
+            for tx in EthereumTx.objects.filter(tx_hash__in=tx_hashes).exclude(
+                block=None
+            )
+        }
+        logger.debug("Conditional indexing: found %d existing txs in DB", len(db_txs))
+
+        # 4. Fetch missing txs from RPC (without receipts - we'll fetch those only for allowed txs)
+        missing_hashes = [tx_hash for tx_hash in tx_hashes if tx_hash not in db_txs]
+        logger.debug(
+            "Conditional indexing: fetching %d missing txs from RPC",
+            len(missing_hashes),
+        )
+        fetched_txs = self._fetch_txs(missing_hashes)
+
+        # 5. Filter by _from (blocklist check)
+        allowed_tx_hashes: set[bytes] = set()
+
+        # Check existing DB txs
+        for tx_hash, db_tx in db_txs.items():
+            if db_tx._from not in self.ignored_initiators:
+                allowed_tx_hashes.add(tx_hash)
+            else:
+                logger.debug(
+                    "Conditional indexing: filtering existing tx %s from blocklisted initiator %s",
+                    to_0x_hex_str(tx_hash),
+                    db_tx._from,
+                )
+
+        # Check fetched txs, filter allowed ones
+        allowed_fetched_txs: list[TxData] = []
+        for tx in fetched_txs:
+            tx_from = tx["from"]
+            if tx_from not in self.ignored_initiators:
+                allowed_fetched_txs.append(tx)
+                allowed_tx_hashes.add(HexBytes(tx["hash"]))
+            else:
+                logger.debug(
+                    "Conditional indexing: filtering tx %s from blocklisted initiator %s",
+                    to_0x_hex_str(tx["hash"]),
+                    tx_from,
+                )
+
+        logger.debug(
+            "Conditional indexing: %d/%d txs allowed after filtering",
+            len(allowed_tx_hashes),
+            len(tx_hashes),
+        )
+
+        # 6. Fetch receipts only for allowed txs and store them
+        if allowed_fetched_txs:
+            number_allowed_txs_inserted = self._fetch_receipts_and_store(
+                allowed_fetched_txs
+            )
+            logger.debug(
+                "Conditional indexing: %d allowed txs inserted",
+                number_allowed_txs_inserted,
+            )
+
+        # 7. Filter log_receipts to only allowed txs
+        filtered_log_receipts = [
+            log_receipt
+            for log_receipt in not_processed_log_receipts
+            if HexBytes(log_receipt["transactionHash"]) in allowed_tx_hashes
+        ]
+
+        # 8. Decode and process
+        decoded_elements = self.decode_elements(filtered_log_receipts)
+        processed_elements = self._process_decoded_elements(decoded_elements)
+
+        # 9. Mark ALL original receipts as processed (so we don't re-fetch blocked ones)
+        for log_receipt in not_processed_log_receipts:
+            self.element_already_processed_checker.mark_as_processed(
+                log_receipt["transactionHash"],
+                log_receipt["blockHash"],
+                log_receipt["logIndex"],
+            )
+
+        return processed_elements
+
     @cached_property
     def contract_events(self) -> list[ContractEvent]:
         """
@@ -545,6 +670,9 @@ def _process_safe_creation_events(
         logger.debug(
             "InternalTx and InternalTxDecoded objects for creation will be built"
         )
+        # Track Safe addresses and their creation tx hashes for SafeContract creation
+        safe_address_to_tx_hash: dict[ChecksumAddress, bytes] = {}
+
         for safe_address in addresses_to_index:
             events = safe_addresses_with_creation_events[safe_address]
 
@@ -617,11 +745,33 @@ def _process_safe_creation_events(
                 internal_txs.append(internal_tx)
                 internal_txs_decoded.append(internal_tx_decoded)
 
+                # Track for SafeContract creation
+                safe_address_to_tx_hash[safe_address] = setup_event["transactionHash"]
+
         logger.debug("InternalTx and InternalTxDecoded objects for creation were built")
-        return InternalTx.objects.store_internal_txs_and_decoded_in_db(
+
+        stored_internal_txs = InternalTx.objects.store_internal_txs_and_decoded_in_db(
             internal_txs, internal_txs_decoded
         )
 
+        # Create SafeContract entries for newly created Safes
+        # This ensures SafeContract exists before non-creation events are filtered
+        # (when conditional indexing is enabled)
+        if safe_address_to_tx_hash:
+            logger.debug(
+                "Creating %d SafeContract entries for new Safes",
+                len(safe_address_to_tx_hash),
+            )
+            SafeContract.objects.bulk_create(
+                [
+                    SafeContract(address=safe_address, ethereum_tx_id=tx_hash)
+                    for safe_address, tx_hash in safe_address_to_tx_hash.items()
+                ],
+                ignore_conflicts=True,  # Safe may already exist from previous indexing
+            )
+
+        return stored_internal_txs
+
     def _prefetch_timestamp_for_blocks(
         self, decoded_elements: list[EventData]
     ) -> dict[bytes, datetime.datetime]:
@@ -646,8 +796,92 @@ def _prefetch_timestamp_for_blocks(
         logger.debug("Ended prefetching timestamp for every block hash")
         return block_hashes_with_timestamp
 
-    def _process_decoded_elements(self, decoded_elements: list[EventData]) -> list[Any]:
-        processed_elements = []
+    def _fetch_txs(self, tx_hashes: list[bytes]) -> list[TxData]:
+        """
+        Fetch transactions from RPC without receipts.
+        Used for conditional indexing to check tx._from before deciding to fetch receipts.
+
+        :param tx_hashes: List of transaction hashes to fetch
+        :return: List of transactions
+        """
+        if not tx_hashes:
+            return []
+
+        txs: list[TxData] = []
+        for tx_hash, tx in zip(
+            tx_hashes,
+            self.ethereum_client.get_transactions(tx_hashes),
+            strict=False,
+        ):
+            tx = tx or self.ethereum_client.get_transaction(tx_hash)  # Retry if failed
+            if tx:
+                txs.append(tx)
+
+        return txs
+
+    def _fetch_receipts_and_store(self, txs: list[TxData]) -> int:
+        """
+        Fetch receipts for allowed transactions and store them in the database.
+        Called after filtering by tx._from to avoid fetching receipts for blocklisted txs.
+
+        :param txs: List of allowed transactions to fetch receipts for and store
+        :return: Number of transactions inserted
+        """
+        if not txs:
+            return 0
+
+        tx_hashes = [tx["hash"] for tx in txs]
+
+        # Fetch receipts for allowed transactions
+        logger.debug(
+            "Conditional indexing: fetching %d receipts for allowed txs",
+            len(tx_hashes),
+        )
+
+        # Build list of (tx, receipt) pairs, only including successful receipt fetches
+        txs_with_receipts: list[tuple[TxData, TxReceipt]] = []
+        for tx, tx_receipt in zip(
+            txs,
+            self.ethereum_client.get_transaction_receipts(tx_hashes),
+            strict=False,
+        ):
+            tx_receipt = tx_receipt or self.ethereum_client.get_transaction_receipt(
+                tx["hash"]
+            )  # Retry if failed
+            if tx_receipt:
+                txs_with_receipts.append((tx, tx_receipt))
+            else:
+                logger.warning(
+                    "Conditional indexing: failed to fetch receipt for tx %s",
+                    to_0x_hex_str(tx["hash"]),
+                )
+
+        if not txs_with_receipts:
+            return 0
+
+        # Collect block hashes only from txs with successful receipts
+        block_hashes = {to_0x_hex_str(tx["blockHash"]) for tx, _ in txs_with_receipts}
+
+        # Create blocks
+        logger.debug("Conditional indexing: inserting %d blocks", len(block_hashes))
+        self.index_service.txs_create_or_update_from_block_hashes(block_hashes)
+
+        # Create EthereumTx records
+        logger.debug(
+            "Conditional indexing: inserting %d transactions", len(txs_with_receipts)
+        )
+        ethereum_txs_to_insert = [
+            EthereumTx.objects.from_tx_dict(tx, receipt)
+            for tx, receipt in txs_with_receipts
+        ]
+        return EthereumTx.objects.bulk_create_from_generator(
+            iter(ethereum_txs_to_insert), ignore_conflicts=True
+        )
+
+    def _process_decoded_elements(
+        self, decoded_elements: list[EventData]
+    ) -> list[InternalTx]:
+        processed_elements: list[InternalTx] = []
 
         self.block_hashes_with_timestamp = self._prefetch_timestamp_for_blocks(
             decoded_elements
@@ -659,18 +893,52 @@ def _process_decoded_elements(self, decoded_elements: list[EventData]) -> list[A
         )
         if safe_addresses_creation_events:
             # Process safe creation events
+            # Note: When conditional indexing is enabled, events are already filtered
+            # by tx._from in process_elements() before reaching this point
             creation_events_processed = self._process_safe_creation_events(
                 safe_addresses_creation_events
             )
             processed_elements.extend(creation_events_processed)
 
+        elements_to_process = [
+            element
+            for element in decoded_elements
+            if element["event"] not in ("SafeSetup", "ProxyCreation")
+        ]
+
+        # When conditional indexing is enabled, only process events for Safes
+        # that exist in SafeContract table
+        if self.conditional_indexing_enabled:
+            # Get all unique Safe addresses from non-creation events
+            non_creation_addresses = {
+                element["address"] for element in elements_to_process
+            }
+            if non_creation_addresses:
+                # Check which addresses exist in SafeContract
+                existing_addresses = SafeContract.objects.get_existing_addresses(
+                    non_creation_addresses
+                )
+                len_non_creation_addresses = len(non_creation_addresses)
+                len_existing_addresses = len(existing_addresses)
+                logger.debug(
+                    "Conditional indexing: %d/%d Safes will be processed",
+                    len_existing_addresses,
+                    len_non_creation_addresses,
+                )
+                # Filter elements to only those with existing SafeContract
+                elements_to_process = [
+                    element
+                    for element in elements_to_process
+                    if element["address"] in existing_addresses
+                ]
+
         # Store everything together in the database if possible
         logger.debug("InternalTx and InternalTx for non creation events will be built")
         internal_txs_to_insert: list[InternalTx] = []
         internal_txs_decoded_to_insert: list[InternalTxDecoded] = []
         safe_relevant_txs: list[SafeRelevantTransaction] = []
         # Process the rest of Safe events. Store all together
-        for decoded_element in decoded_elements:
+        for decoded_element in elements_to_process:
             elements_to_insert = self._process_decoded_element(decoded_element)
             for element_to_insert in elements_to_insert:
                 if isinstance(element_to_insert, InternalTx):
@@ -681,14 +949,16 @@ def _process_decoded_elements(self, decoded_elements: list[EventData]) -> list[A
                     safe_relevant_txs.append(element_to_insert)
         logger.debug("InternalTx and InternalTx for non creation events were built")
 
-        stored_internal_txs = InternalTx.objects.store_internal_txs_and_decoded_in_db(
-            internal_txs_to_insert, internal_txs_decoded_to_insert
+        stored_internal_txs: list[InternalTx] = (
+            InternalTx.objects.store_internal_txs_and_decoded_in_db(
+                internal_txs_to_insert, internal_txs_decoded_to_insert
+            )
         )
         logger.debug("Inserting %d SafeRelevantTransaction", len(safe_relevant_txs))
         SafeRelevantTransaction.objects.bulk_create(
             safe_relevant_txs, ignore_conflicts=True
         )
-        logger.debug("Inserted SafeRelevantTransaction")
+        logger.debug("Inserted %d SafeRelevantTransaction", len(safe_relevant_txs))
 
         processed_elements.extend(stored_internal_txs)
         return processed_elements