fix: support sender signers

common/src/escrow_accounts.rs

@@ -13,18 +13,43 @@ use tracing::{error, warn};
 
 use crate::prelude::{Query, SubgraphClient};
 
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct EscrowAccounts {
+    pub senders_balances: HashMap<Address, U256>,
+    pub signers_to_senders: HashMap<Address, Address>,
+    pub senders_to_signers: HashMap<Address, Vec<Address>>,
+}
+
+impl EscrowAccounts {
+    pub fn new(
+        senders_balances: HashMap<Address, U256>,
+        senders_to_signers: HashMap<Address, Vec<Address>>,
+    ) -> Self {
+        let signers_to_senders = senders_to_signers
+            .iter()
+            .flat_map(|(sender, signers)| signers.iter().map(move |signer| (*signer, *sender)))
+            .collect();
+
+        Self {
+            senders_balances,
+            signers_to_senders,
+            senders_to_signers,
+        }
+    }
+}
+
 pub fn escrow_accounts(
     escrow_subgraph: &'static SubgraphClient,
     indexer_address: Address,
     interval: Duration,
-) -> Eventual<HashMap<Address, U256>> {
+    reject_thawing_signers: bool,
+) -> Eventual<EscrowAccounts> {
     // Types for deserializing the network subgraph response
     #[derive(Deserialize)]
     #[serde(rename_all = "camelCase")]
     struct EscrowAccountsResponse {
         escrow_accounts: Vec<EscrowAccount>,
     }
-    // These 2 structs are used to deserialize the response from the escrow subgraph.
     // Note that U256's serde implementation is based on serializing the internal bytes, not the string decimal
     // representation. This is why we deserialize them as strings below.
     #[derive(Deserialize)]
@@ -38,50 +63,105 @@ pub fn escrow_accounts(
     #[serde(rename_all = "camelCase")]
     struct Sender {
         id: Address,
+        authorized_signers: Vec<AuthorizedSigner>,
+    }
+    #[derive(Deserialize)]
+    #[serde(rename_all = "camelCase")]
+    struct AuthorizedSigner {
+        id: Address,
     }
 
+    // thawEndTimestamp == 0 means that the signer is not thawing. This also means
+    // that we don't wait for the thawing period to end before stopping serving
+    // queries for this signer.
+    // isAuthorized == true means that the signer is still authorized to sign
+    // payments in the name of the sender.
+    let query = if reject_thawing_signers {
+        r#"
+        query ($indexer: ID!) {
+            escrowAccounts(where: {receiver_: {id: $indexer}}) {
+                balance
+                totalAmountThawing
+                sender {
+                    id
+                    authorizedSigners(
+                        where: {thawEndTimestamp: "0", isAuthorized: true}
+                    ) {
+                        id
+                    }
+                }
+            }
+        }
+    "#
+    } else {
+        r#"
+        query ($indexer: ID!) {
+            escrowAccounts(where: {receiver_: {id: $indexer}}) {
+                balance
+                totalAmountThawing
+                sender {
+                    id
+                    authorizedSigners(
+                        where: {isAuthorized: true}
+                    ) {
+                        id
+                    }
+                }
+            }
+        }
+    "#
+    };
+
     timer(interval).map_with_retry(
         move |_| async move {
             let response = escrow_subgraph
                 .query::<EscrowAccountsResponse>(Query::new_with_variables(
-                    r#"
-                        query ($indexer: ID!) {
-                            escrowAccounts(where: {receiver_: {id: $indexer}}) {
-                                balance
-                                totalAmountThawing
-                                sender {
-                                    id
-                                }
-                            }
-                        }
-                    "#,
+                    query,
                     [("indexer", format!("{:x?}", indexer_address).into())],
                 ))
                 .await
                 .map_err(|e| e.to_string())?;
 
-            response.map_err(|e| e.to_string()).and_then(|data| {
-                data.escrow_accounts
-                    .iter()
-                    .map(|account| {
-                        let balance = U256::checked_sub(
-                            U256::from_dec_str(&account.balance)?,
-                            U256::from_dec_str(&account.total_amount_thawing)?,
-                        )
-                        .unwrap_or_else(|| {
-                            warn!(
-                                "Balance minus total amount thawing underflowed for account {}. \
+            let response = response.map_err(|e| e.to_string())?;
+
+            let senders_balances = response
+                .escrow_accounts
+                .iter()
+                .map(|account| {
+                    let balance = U256::checked_sub(
+                        U256::from_dec_str(&account.balance)?,
+                        U256::from_dec_str(&account.total_amount_thawing)?,
+                    )
+                    .unwrap_or_else(|| {
+                        warn!(
+                            "Balance minus total amount thawing underflowed for account {}. \
                                  Setting balance to 0, no queries will be served for this sender.",
-                                account.sender.id
-                            );
-                            U256::from(0)
-                        });
-
-                        Ok((account.sender.id, balance))
-                    })
-                    .collect::<Result<HashMap<_, _>, anyhow::Error>>()
-                    .map_err(|e| format!("{}", e))
-            })
+                            account.sender.id
+                        );
+                        U256::from(0)
+                    });
+
+                    Ok((account.sender.id, balance))
+                })
+                .collect::<Result<HashMap<_, _>, anyhow::Error>>()
+                .map_err(|e| format!("{}", e))?;
+
+            let senders_to_signers = response
+                .escrow_accounts
+                .iter()
+                .map(|account| {
+                    let sender = account.sender.id;
+                    let signers = account
+                        .sender
+                        .authorized_signers
+                        .iter()
+                        .map(|signer| signer.id)
+                        .collect();
+                    (sender, signers)
+                })
+                .collect();
+
+            Ok(EscrowAccounts::new(senders_balances, senders_to_signers))
         },
         move |err: String| {
             error!(
@@ -96,6 +176,7 @@ pub fn escrow_accounts(
 
 #[cfg(test)]
 mod tests {
+    use test_log::test;
     use wiremock::matchers::{method, path};
     use wiremock::{Mock, MockServer, ResponseTemplate};
 
@@ -104,7 +185,20 @@ mod tests {
 
     use super::*;
 
-    #[tokio::test]
+    #[test]
+    fn test_new_escrow_accounts() {
+        let escrow_accounts = EscrowAccounts::new(
+            test_vectors::ESCROW_ACCOUNTS_BALANCES.to_owned(),
+            test_vectors::ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),
+        );
+
+        assert_eq!(
+            escrow_accounts.signers_to_senders,
+            test_vectors::ESCROW_ACCOUNTS_SIGNERS_TO_SENDERS.to_owned()
+        )
+    }
+
+    #[test(tokio::test)]
     async fn test_current_accounts() {
         // Set up a mock escrow subgraph
         let mock_server = MockServer::start().await;
@@ -134,11 +228,15 @@ mod tests {
             escrow_subgraph,
             *test_vectors::INDEXER_ADDRESS,
             Duration::from_secs(60),
+            true,
         );
 
         assert_eq!(
             accounts.value().await.unwrap(),
-            *test_vectors::ESCROW_ACCOUNTS
+            EscrowAccounts::new(
+                test_vectors::ESCROW_ACCOUNTS_BALANCES.to_owned(),
+                test_vectors::ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),
+            )
         );
     }
 }

common/src/indexer_service/http/indexer_service.rs

@@ -262,6 +262,7 @@ impl IndexerService {
             escrow_subgraph,
             options.config.indexer.indexer_address,
             Duration::from_secs(options.config.escrow_subgraph.syncing_interval),
+            true, // Reject thawing signers eagerly
         );
 
         // Establish Database connection necessary for serving indexer management