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
- Add ETH_EVENTS_CONDITIONAL_INDEXING flag (auto-enabled when blocklist is set)
- Create SafeToIndex model for manual allowlist management
- 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 safe_allowlist management command (add/remove/list/count)
- Add Django admin for SafeToIndex model
- Add comprehensive tests for conditional indexing and management command

Closes #2703

Author: Uxio0

config/settings/base.py

@@ -609,6 +609,13 @@
 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
+
+# Auto-enabled when any conditional indexing selector is configured (e.g., ETH_EVENTS_IGNORED_INITIATORS)
+ETH_EVENTS_CONDITIONAL_INDEXING: bool = bool(ETH_EVENTS_IGNORED_INITIATORS)
 
 # ENABLE/DISABLE COLLECTIBLES DOWNLOAD METADATA, enable=True, disabled by default
 COLLECTIBLES_ENABLE_DOWNLOAD_METADATA = env.bool(

safe_transaction_service/history/admin.py

@@ -32,6 +32,7 @@
     SafeLastStatus,
     SafeMasterCopy,
     SafeStatus,
+    SafeToIndex,
 )
 from .services import IndexServiceProvider
 from .utils import HexField
@@ -555,6 +556,13 @@ class SafeContractDelegateAdmin(AdvancedAdminSearchMixin, admin.ModelAdmin):
     search_fields = ["==safe_contract__address", "==delegate", "==delegator"]
 
 
+@admin.register(SafeToIndex)
+class SafeToIndexAdmin(AdvancedAdminSearchMixin, admin.ModelAdmin):
+    list_display = ("address",)
+    ordering = ["address"]
+    search_fields = ["==address"]
+
+
 class SafeStatusModulesListFilter(admin.SimpleListFilter):
     title = "Modules enabled in Safe"
     parameter_name = "enabled_modules"

safe_transaction_service/history/indexers/safe_events_indexer.py

@@ -1,4 +1,6 @@
 import datetime
+from collections import OrderedDict
+from collections.abc import Sequence
 from functools import cached_property
 from logging import getLogger
 from typing import Any
@@ -20,10 +22,11 @@
 )
 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,
@@ -70,6 +73,122 @@ def __init__(self, *args, **kwargs):
         self.block_hashes_with_timestamp: dict[bytes, datetime.datetime] = {}
         super().__init__(*args, **kwargs)
 
+    def process_elements(self, log_receipts: Sequence[LogReceipt]) -> list[Any]:
+        """
+        Override to filter events by tx._from when conditional indexing is enabled.
+        This avoids storing EthereumTx records for blocklisted initiators.
+        """
+        if not settings.ETH_EVENTS_CONDITIONAL_INDEXING:
+            return super().process_elements(log_receipts)
+
+        # --- Conditional indexing enabled ---
+        if not log_receipts:
+            return []
+
+        ignored_initiators = settings.ETH_EVENTS_IGNORED_INITIATORS
+        if not ignored_initiators:
+            # No blocklist configured, use standard flow
+            return super().process_elements(log_receipts)
+
+        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 = [h for h in tx_hashes if h 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 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 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:
+            self._fetch_receipts_and_store(allowed_fetched_txs)
+
+        # 7. Filter log_receipts to only allowed txs
+        filtered_log_receipts = [
+            r
+            for r in not_processed_log_receipts
+            if HexBytes(r["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]:
         """
@@ -516,6 +635,9 @@ def _process_safe_creation_events(
         - SafeSetup
         - ProxyCreation
 
+        Note: When conditional indexing is enabled, events are filtered by tx._from
+        in process_elements() before this method is called.
+
         :param safe_addresses_with_creation_events:
         :return: Generated InternalTxs for safe creation
         """
@@ -618,9 +740,11 @@ def _process_safe_creation_events(
                 internal_txs_decoded.append(internal_tx_decoded)
 
         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
         )
+        return stored_internal_txs
 
     def _prefetch_timestamp_for_blocks(
         self, decoded_elements: list[EventData]
@@ -646,6 +770,81 @@ def _prefetch_timestamp_for_blocks(
         logger.debug("Ended prefetching timestamp for every block hash")
         return block_hashes_with_timestamp
 
+    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]) -> None:
+        """
+        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
+        """
+        if not txs:
+            return
+
+        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),
+        )
+        receipts: list[TxReceipt] = []
+        for tx_hash, tx_receipt in zip(
+            tx_hashes,
+            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:
+                receipts.append(tx_receipt)
+
+        # Collect block hashes
+        block_hashes = {to_0x_hex_str(tx["blockHash"]) for tx in txs}
+
+        # Create blocks
+        logger.debug("Conditional indexing: inserting %d blocks", len(block_hashes))
+        _, blocks = self.index_service.txs_create_or_update_from_block_hashes(
+            block_hashes
+        )
+
+        # Create EthereumTx records
+        logger.debug("Conditional indexing: inserting %d transactions", len(txs))
+        ethereum_txs_to_insert = [
+            EthereumTx.objects.from_tx_dict(tx, receipt)
+            for tx, receipt in zip(txs, receipts, strict=False)
+        ]
+        EthereumTx.objects.bulk_create_from_generator(
+            iter(ethereum_txs_to_insert), ignore_conflicts=True
+        )
+
+        # Set block references
+        for ethereum_tx, tx in zip(ethereum_txs_to_insert, txs, strict=False):
+            ethereum_tx.block = blocks[tx["blockHash"]]
+
     def _process_decoded_elements(self, decoded_elements: list[EventData]) -> list[Any]:
         processed_elements = []
 
@@ -659,6 +858,8 @@ 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
             )