graphprotocol
diff --git a/‎crates/config/default_values.toml‎
Lines changed: 4 additions & 0 deletions b/‎crates/config/default_values.toml‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎crates/config/maximal-config-example.toml‎
Lines changed: 8 additions & 0 deletions b/‎crates/config/maximal-config-example.toml‎
Lines changed: 8 additions & 0 deletions
diff --git a/‎crates/config/minimal-config-example.toml‎
Lines changed: 1 addition & 0 deletions b/‎crates/config/minimal-config-example.toml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/config/src/config.rs‎
Lines changed: 22 additions & 0 deletions b/‎crates/config/src/config.rs‎
Lines changed: 22 additions & 0 deletions
diff --git a/‎crates/monitor/src/escrow_accounts.rs‎
Lines changed: 184 additions & 38 deletions b/‎crates/monitor/src/escrow_accounts.rs‎
Lines changed: 184 additions & 38 deletions
@@ -5,6 +5,8 @@ port = 7300
 syncing_interval_secs = 60
 recently_closed_allocation_buffer_secs = 3600
 max_data_staleness_mins = 30
+escrow_min_balance_grt_wei = "100000000000000000"
+max_signers_per_payer = 0
 
 [subgraphs.escrow]
 syncing_interval_secs = 60
@@ -46,6 +48,8 @@ chain_id = 1337
 receipts_verifier_address_v2 = "0x3333333333333333333333333333333333333333"
 # SubgraphService contract (required).
 subgraph_service_address = "0xcf7ed3acca5a467e9e704c703e8d87f634fb0fc9"
+# GraphTallyCollector contract (required for V2 escrow filtering).
+graph_tally_collector_address = "0x4444444444444444444444444444444444444444"
 
 [horizon]
 # Enable Horizon support and detection
 
@@ -83,6 +83,13 @@ recently_closed_allocation_buffer_secs = 3600
 # This protects against Gateway routing queries to indexers that are significantly behind.
 # Set to 0 to disable staleness checking.
 max_data_staleness_mins = 30
+# Minimum escrow balance (GRT wei) for V2 escrow queries. Filters dust deposits
+# to raise the cost of crowding attacks. Default: 0.1 GRT.
+escrow_min_balance_grt_wei = "100000000000000000"
+# Maximum signers to fetch per payer. 0 = no limit (recommended).
+# A positive value caps pagination but allows attackers to crowd out legitimate
+# signers, orphaning receipts and enabling free queries.
+max_signers_per_payer = 0
 
 [subgraphs.escrow]
 # NOTE: It is heavily recomended to use both `query_url` and `deployment_id`,
@@ -106,6 +113,7 @@ chain_id = 1337
 # Horizon addresses; required when [horizon].enabled = true
 receipts_verifier_address_v2 = "0x3333333333333333333333333333333333333333"
 subgraph_service_address = "0xcf7ed3acca5a467e9e704c703e8d87f634fb0fc9"
+graph_tally_collector_address = "0x4444444444444444444444444444444444444444"
 
 ##############################################
 # Specific configurations to indexer-service #
 
@@ -61,6 +61,7 @@ chain_id = 1337
 # Horizon addresses; required when [horizon].enabled = true
 receipts_verifier_address_v2 = "0x3333333333333333333333333333333333333333"
 subgraph_service_address = "0xcf7ed3acca5a467e9e704c703e8d87f634fb0fc9"
+graph_tally_collector_address = "0x4444444444444444444444444444444444444444"
 
 ########################################
 # Specific configurations to tap-agent #
 
@@ -498,12 +498,32 @@ pub struct NetworkSubgraphConfig {
     /// Default: 30 (minutes)
     #[serde(default = "default_max_data_staleness_mins")]
     pub max_data_staleness_mins: u64,
+
+    /// Minimum escrow balance (GRT wei) for the V2 escrow query. Filters dust
+    /// deposits to raise the cost of crowding attacks. Default: 0.1 GRT.
+    #[serde(default = "default_escrow_min_balance_grt_wei")]
+    pub escrow_min_balance_grt_wei: String,
+
+    /// Maximum signers to fetch per payer. 0 = no limit (recommended).
+    /// Setting a positive value caps pagination but allows attackers to crowd out
+    /// legitimate signers, orphaning receipts and enabling free queries.
+    /// Default: 0.
+    #[serde(default = "default_max_signers_per_payer")]
+    pub max_signers_per_payer: usize,
 }
 
 fn default_max_data_staleness_mins() -> u64 {
     30
 }
 
+fn default_escrow_min_balance_grt_wei() -> String {
+    "100000000000000000".to_string() // 0.1 GRT
+}
+
+fn default_max_signers_per_payer() -> usize {
+    0
+}
+
 #[derive(Debug, Deserialize)]
 #[cfg_attr(test, derive(PartialEq))]
 pub struct EscrowSubgraphConfig {
@@ -546,6 +566,8 @@ pub struct BlockchainConfig {
     pub receipts_verifier_address_v2: Address,
     /// Address of the SubgraphService contract used for Horizon operations
     pub subgraph_service_address: Option<Address>,
+    /// Address of the GraphTallyCollector contract used to filter V2 escrow account queries.
+    pub graph_tally_collector_address: Address,
 }
 
 impl BlockchainConfig {
 
@@ -101,13 +101,6 @@ impl EscrowAccounts {
     pub fn contains_signer(&self, signer: &Address) -> bool {
         self.signers_to_senders.contains_key(signer)
     }
-
-    pub fn get_all_mappings(&self) -> Vec<(Address, Address)> {
-        self.signers_to_senders
-            .iter()
-            .map(|(signer, sender)| (*signer, *sender))
-            .collect()
-    }
 }
 
 pub type EscrowAccountsWatcher = Receiver<EscrowAccounts>;
@@ -135,9 +128,19 @@ pub async fn escrow_accounts_v2(
     indexer_address: Address,
     interval: Duration,
     reject_thawing_signers: bool,
+    graph_tally_collector_address: Address,
+    min_balance_grt_wei: String,
+    max_signers_per_payer: usize,
 ) -> Result<EscrowAccountsWatcher, anyhow::Error> {
     indexer_watcher::new_watcher(interval, move || {
-        get_escrow_accounts_v2(escrow_subgraph, indexer_address, reject_thawing_signers)
+        get_escrow_accounts_v2(
+            escrow_subgraph,
+            indexer_address,
+            reject_thawing_signers,
+            graph_tally_collector_address,
+            min_balance_grt_wei.clone(),
+            max_signers_per_payer,
+        )
     })
     .await
 }
@@ -146,45 +149,180 @@ async fn get_escrow_accounts_v2(
     escrow_subgraph: &'static SubgraphClient,
     indexer_address: Address,
     reject_thawing_signers: bool,
+    graph_tally_collector_address: Address,
+    min_balance_grt_wei: String,
+    max_signers_per_payer: usize,
 ) -> anyhow::Result<EscrowAccounts> {
     tracing::trace!(
         indexer_address = ?indexer_address,
         reject_thawing_signers,
         "Loading V2 escrow accounts for indexer"
     );
-    // Query V2 escrow accounts from the network subgraph which tracks PaymentsEscrow
-    // and GraphTallyCollector contract events.
 
     use indexer_query::network_escrow_account_v2::{
-        self as network_escrow_account_v2, NetworkEscrowAccountQueryV2,
+        self as network_escrow_account_v2, Block_height, NetworkEscrowAccountQueryV2,
     };
+    use indexer_query::signers_by_payer::{self as signers_by_payer, SignersByPayerQuery};
 
-    let response = escrow_subgraph
-        .query::<NetworkEscrowAccountQueryV2, _>(network_escrow_account_v2::Variables {
-            receiver: format!("{indexer_address:x?}"),
-            thaw_end_timestamp: if reject_thawing_signers {
-                U256::ZERO.to_string()
-            } else {
-                U256::MAX.to_string()
-            },
-        })
-        .await?;
+    let page_size: i64 = 200;
+    let mut last: Option<String> = None;
+    let mut block_hash: Option<String> = None;
+    let mut all_accounts = vec![];
 
-    // V2 TAP receipts use different field names (payer/service_provider) but the underlying
-    // escrow account model is identical to V1. Both V1 and V2 receipts reference the same
-    // sender addresses and the same escrow relationships.
-    //
-    // V1 queries the TAP subgraph while V2 queries the network subgraph, but both return
-    // the same escrow account structure for processing.
-    //
-    // V2 receipt flow:
-    // 1. V2 receipt contains payer address (equivalent to V1 sender)
-    // 2. Receipt is signed by a signer authorized by the payer
-    // 3. Escrow accounts map: signer -> payer (sender) -> balance
-    // 4. Service provider (indexer) receives payments from payer's escrow
+    let thaw_end_timestamp = if reject_thawing_signers {
+        U256::ZERO.to_string()
+    } else {
+        U256::MAX.to_string()
+    };
 
-    let senders_balances: HashMap<Address, U256> = response
-        .payments_escrow_accounts
+    let mut pages_fetched: u64 = 0;
+
+    loop {
+        let response = escrow_subgraph
+            .query::<NetworkEscrowAccountQueryV2, _>(network_escrow_account_v2::Variables {
+                receiver: format!("{indexer_address:x?}"),
+                collector: format!("{graph_tally_collector_address:x?}"),
+                thaw_end_timestamp: thaw_end_timestamp.clone(),
+                first: page_size,
+                last: last.unwrap_or_default(),
+                min_balance: min_balance_grt_wei.clone(),
+                block: block_hash.as_ref().map(|h| Block_height {
+                    hash: Some(h.clone()),
+                    number: None,
+                    number_gte: None,
+                }),
+            })
+            .await?;
+
+        let page_len = response.payments_escrow_accounts.len();
+        pages_fetched += 1;
+
+        // Pin to the block hash from the first page for consistent reads across pages
+        if block_hash.is_none() {
+            block_hash = response.meta.as_ref().and_then(|m| m.block.hash.clone());
+        }
+
+        // Paginate additional signers for any payer that hit the 1000-per-payer nested cap
+        let mut response = response;
+        for account in &mut response.payments_escrow_accounts {
+            if max_signers_per_payer > 0 && account.payer.signers.len() >= max_signers_per_payer {
+                tracing::warn!(
+                    payer = %account.payer.id,
+                    signers = account.payer.signers.len(),
+                    max = max_signers_per_payer,
+                    "Payer signers already at or above max_signers_per_payer cap; skipping follow-up pagination"
+                );
+                account.payer.signers.truncate(max_signers_per_payer);
+                continue;
+            }
+
+            if account.payer.signers.len() < 1000 {
+                continue;
+            }
+
+            let mut signer_cursor = account
+                .payer
+                .signers
+                .last()
+                .map(|s| s.id.clone())
+                .unwrap_or_default();
+
+            tracing::info!(
+                payer = %account.payer.id,
+                initial_signers = account.payer.signers.len(),
+                "Payer hit nested signer cap; starting follow-up pagination"
+            );
+
+            loop {
+                let signer_response = escrow_subgraph
+                    .query::<SignersByPayerQuery, _>(signers_by_payer::Variables {
+                        payer: account.payer.id.clone(),
+                        first: 1000,
+                        last: signer_cursor.clone(),
+                        thaw_end_timestamp: thaw_end_timestamp.clone(),
+                        block: block_hash.as_ref().map(|h| signers_by_payer::Block_height {
+                            hash: Some(h.clone()),
+                            number: None,
+                            number_gte: None,
+                        }),
+                    })
+                    .await?;
+
+                let page_len = signer_response.signers.len();
+
+                for signer in &signer_response.signers {
+                    account.payer.signers.push(
+                        network_escrow_account_v2::NetworkEscrowAccountQueryV2PaymentsEscrowAccountsPayerSigners {
+                            id: signer.id.clone(),
+                        },
+                    );
+                }
+
+                if page_len < 1000
+                    || (max_signers_per_payer > 0
+                        && account.payer.signers.len() >= max_signers_per_payer)
+                {
+                    break;
+                }
+
+                signer_cursor = signer_response
+                    .signers
+                    .last()
+                    .map(|s| s.id.clone())
+                    .unwrap_or_default();
+            }
+
+            if max_signers_per_payer > 0 && account.payer.signers.len() >= max_signers_per_payer {
+                tracing::warn!(
+                    payer = %account.payer.id,
+                    signers = account.payer.signers.len(),
+                    max = max_signers_per_payer,
+                    "Payer signers capped at max_signers_per_payer"
+                );
+                account.payer.signers.truncate(max_signers_per_payer);
+            }
+
+            if account.payer.signers.len() > 1000 {
+                tracing::warn!(
+                    payer = %account.payer.id,
+                    signers = account.payer.signers.len(),
+                    "Payer has an unusual number of signers which may indicate a signer-flooding attack; consider adding to deny list"
+                );
+            }
+
+            tracing::info!(
+                payer = %account.payer.id,
+                total_signers = account.payer.signers.len(),
+                "Signer follow-up pagination complete"
+            );
+        }
+
+        last = response
+            .payments_escrow_accounts
+            .last()
+            .map(|a| a.id.to_string());
+        all_accounts.extend(response.payments_escrow_accounts);
+
+        if (page_len as i64) < page_size {
+            break;
+        }
+    }
+
+    tracing::info!(
+        pages = pages_fetched,
+        accounts = all_accounts.len(),
+        "V2 escrow account pagination complete"
+    );
+
+    if pages_fetched > 5 {
+        tracing::warn!(
+            pages = pages_fetched,
+            accounts = all_accounts.len(),
+            "Unusually high number of V2 escrow accounts; this may indicate a dust-deposit crowding attack"
+        );
+    }
+
+    let senders_balances: HashMap<Address, U256> = all_accounts
         .iter()
         .map(|account| {
             let balance = U256::checked_sub(
@@ -203,8 +341,7 @@ async fn get_escrow_accounts_v2(
         })
         .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
 
-    let senders_to_signers = response
-        .payments_escrow_accounts
+    let senders_to_signers = all_accounts
         .into_iter()
         .map(|account| {
             let payer = Address::from_str(&account.payer.id)?;
@@ -218,7 +355,13 @@ async fn get_escrow_accounts_v2(
         })
         .collect::<Result<HashMap<_, _>, anyhow::Error>>()?;
 
-    let escrow_accounts = EscrowAccounts::new(senders_balances.clone(), senders_to_signers.clone());
+    let escrow_accounts = EscrowAccounts::new(senders_balances, senders_to_signers);
+
+    tracing::info!(
+        senders = escrow_accounts.get_senders().len(),
+        signers = escrow_accounts.signer_count(),
+        "V2 escrow accounts loaded"
+    );
 
     Ok(escrow_accounts)
 }
@@ -405,6 +548,9 @@ mod tests {
             test_assets::INDEXER_ADDRESS,
             Duration::from_secs(60),
             true,
+            Address::ZERO, // collector address; mock ignores query variables
+            "100000000000000000".to_string(),
+            0,
         )
         .await
         .unwrap();