fix: configure escrow watchers based on horizon settings

Author: suchapalaver

crates/service/src/middleware/sender.rs

@@ -6,7 +6,7 @@ use axum::{
     middleware::Next,
     response::Response,
 };
-use indexer_monitor::EscrowAccounts;
+use indexer_monitor::{EscrowAccounts, EscrowAccountsError};
 use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain};
 use tokio::sync::watch;
 
@@ -18,9 +18,9 @@ pub struct SenderState {
     /// Used to recover the signer address
     pub domain_separator: Eip712Domain,
     /// Used to get the sender address given the signer address if v1 receipt
-    pub escrow_accounts_v1: watch::Receiver<EscrowAccounts>,
+    pub escrow_accounts_v1: Option<watch::Receiver<EscrowAccounts>>,
     /// Used to get the sender address given the signer address if v2 receipt
-    pub escrow_accounts_v2: watch::Receiver<EscrowAccounts>,
+    pub escrow_accounts_v2: Option<watch::Receiver<EscrowAccounts>>,
 }
 
 /// The current query Sender address
@@ -48,14 +48,24 @@ pub async fn sender_middleware(
     if let Some(receipt) = request.extensions().get::<TapReceipt>() {
         let signer = receipt.recover_signer(&state.domain_separator)?;
         let sender = match receipt {
-            TapReceipt::V1(_) => state
-                .escrow_accounts_v1
-                .borrow()
-                .get_sender_for_signer(&signer)?,
-            TapReceipt::V2(_) => state
-                .escrow_accounts_v2
-                .borrow()
-                .get_sender_for_signer(&signer)?,
+            TapReceipt::V1(_) => {
+                if let Some(ref escrow_accounts_v1) = state.escrow_accounts_v1 {
+                    escrow_accounts_v1.borrow().get_sender_for_signer(&signer)?
+                } else {
+                    return Err(IndexerServiceError::EscrowAccount(
+                        EscrowAccountsError::NoSenderFound { signer },
+                    ));
+                }
+            }
+            TapReceipt::V2(_) => {
+                if let Some(ref escrow_accounts_v2) = state.escrow_accounts_v2 {
+                    escrow_accounts_v2.borrow().get_sender_for_signer(&signer)?
+                } else {
+                    return Err(IndexerServiceError::EscrowAccount(
+                        EscrowAccountsError::NoSenderFound { signer },
+                    ));
+                }
+            }
         };
         request.extensions_mut().insert(Sender(sender));
     }
@@ -100,8 +110,8 @@ mod tests {
 
         let state = SenderState {
             domain_separator: test_assets::TAP_EIP712_DOMAIN.clone(),
-            escrow_accounts_v1,
-            escrow_accounts_v2,
+            escrow_accounts_v1: Some(escrow_accounts_v1),
+            escrow_accounts_v2: Some(escrow_accounts_v2),
         };
 
         let middleware = from_fn_with_state(state, sender_middleware);

crates/service/src/service.rs

@@ -105,19 +105,58 @@ pub async fn run() -> anyhow::Result<()> {
     let indexer_address = config.indexer.indexer_address;
     let ipfs_url = config.service.ipfs_url.clone();
 
-    let router = ServiceRouter::builder()
-        .database(database.clone())
-        .domain_separator(domain_separator.clone())
-        .graph_node(config.graph_node)
-        .http_client(http_client)
-        .release(release)
-        .indexer(config.indexer)
-        .service(config.service)
-        .blockchain(config.blockchain)
-        .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
-        .network_subgraph(network_subgraph, config.subgraphs.network)
-        .escrow_subgraph(escrow_subgraph, config.subgraphs.escrow)
-        .build();
+    // Configure router with escrow watchers based on Horizon setting
+    let router = if config.horizon.enabled {
+        tracing::info!("Horizon support enabled - using escrow accounts v2");
+        // Only create v2 watcher when Horizon is enabled
+        let v2_watcher = indexer_monitor::escrow_accounts_v2(
+            escrow_subgraph,
+            indexer_address,
+            config.subgraphs.escrow.config.syncing_interval_secs,
+            true, // Reject thawing signers eagerly
+        )
+        .await
+        .expect("Error creating escrow_accounts_v2 channel");
+
+        ServiceRouter::builder()
+            .database(database.clone())
+            .domain_separator(domain_separator.clone())
+            .graph_node(config.graph_node)
+            .http_client(http_client)
+            .release(release)
+            .indexer(config.indexer)
+            .service(config.service)
+            .blockchain(config.blockchain)
+            .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
+            .network_subgraph(network_subgraph, config.subgraphs.network)
+            .escrow_accounts_v2(v2_watcher)
+            .build()
+    } else {
+        tracing::info!("Horizon support disabled - using escrow accounts v1");
+        // Only create v1 watcher when Horizon is disabled
+        let v1_watcher = indexer_monitor::escrow_accounts_v1(
+            escrow_subgraph,
+            indexer_address,
+            config.subgraphs.escrow.config.syncing_interval_secs,
+            true, // Reject thawing signers eagerly
+        )
+        .await
+        .expect("Error creating escrow_accounts_v1 channel");
+
+        ServiceRouter::builder()
+            .database(database.clone())
+            .domain_separator(domain_separator.clone())
+            .graph_node(config.graph_node)
+            .http_client(http_client)
+            .release(release)
+            .indexer(config.indexer)
+            .service(config.service)
+            .blockchain(config.blockchain)
+            .timestamp_buffer_secs(config.tap.rav_request.timestamp_buffer_secs)
+            .network_subgraph(network_subgraph, config.subgraphs.network)
+            .escrow_accounts_v1(v1_watcher)
+            .build()
+    };
 
     serve_metrics(config.metrics.get_socket_addr());
 

crates/service/src/service/router.rs

@@ -140,33 +140,42 @@ impl ServiceRouter {
         // Monitor escrow accounts v1
         // if not provided, create monitor from subgraph
         let escrow_accounts_v1 = match (self.escrow_accounts_v1, self.escrow_subgraph.as_ref()) {
-            (Some(escrow_account), _) => escrow_account,
-            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v1(
-                escrow_subgraph,
-                indexer_address,
-                escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts channel"),
-            (None, None) => panic!("No escrow accounts or escrow subgraph was provided"),
+            (Some(escrow_account), _) => Some(escrow_account),
+            (_, Some((escrow_subgraph, escrow))) => Some(
+                escrow_accounts_v1(
+                    escrow_subgraph,
+                    indexer_address,
+                    escrow.config.syncing_interval_secs,
+                    true, // Reject thawing signers eagerly
+                )
+                .await
+                .expect("Error creating escrow_accounts_v1 channel"),
+            ),
+            (None, None) => None,
         };
 
         // Monitor escrow accounts v2
         // if not provided, create monitor from subgraph
         let escrow_accounts_v2 = match (self.escrow_accounts_v2, self.escrow_subgraph.as_ref()) {
-            (Some(escrow_account), _) => escrow_account,
-            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v2(
-                escrow_subgraph,
-                indexer_address,
-                escrow.config.syncing_interval_secs,
-                true, // Reject thawing signers eagerly
-            )
-            .await
-            .expect("Error creating escrow_accounts channel"),
-            (None, None) => panic!("No escrow accounts or escrow subgraph was provided"),
+            (Some(escrow_account), _) => Some(escrow_account),
+            (_, Some((escrow_subgraph, escrow))) => Some(
+                escrow_accounts_v2(
+                    escrow_subgraph,
+                    indexer_address,
+                    escrow.config.syncing_interval_secs,
+                    true, // Reject thawing signers eagerly
+                )
+                .await
+                .expect("Error creating escrow_accounts_v2 channel"),
+            ),
+            (None, None) => None,
         };
 
+        // Ensure at least one escrow accounts watcher is available
+        if escrow_accounts_v1.is_none() && escrow_accounts_v2.is_none() {
+            panic!("At least one escrow accounts watcher (v1 or v2) must be provided");
+        }
+
         // Monitor dispute manager address
         // if not provided, create monitor from subgraph
         let dispute_manager = match (self.dispute_manager, self.network_subgraph.as_ref()) {

crates/service/src/tap.rs

@@ -51,8 +51,8 @@ impl IndexerTapContext {
     pub async fn get_checks(
         pgpool: PgPool,
         indexer_allocations: Receiver<HashMap<Address, Allocation>>,
-        escrow_accounts_v1: Receiver<EscrowAccounts>,
-        escrow_accounts_v2: Receiver<EscrowAccounts>,
+        escrow_accounts_v1: Option<Receiver<EscrowAccounts>>,
+        escrow_accounts_v2: Option<Receiver<EscrowAccounts>>,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
     ) -> Vec<ReceiptCheck<TapReceipt>> {

crates/service/src/tap/checks/sender_balance_check.rs

@@ -13,14 +13,14 @@ use crate::{
 };
 
 pub struct SenderBalanceCheck {
-    escrow_accounts_v1: Receiver<EscrowAccounts>,
-    escrow_accounts_v2: Receiver<EscrowAccounts>,
+    escrow_accounts_v1: Option<Receiver<EscrowAccounts>>,
+    escrow_accounts_v2: Option<Receiver<EscrowAccounts>>,
 }
 
 impl SenderBalanceCheck {
     pub fn new(
-        escrow_accounts_v1: Receiver<EscrowAccounts>,
-        escrow_accounts_v2: Receiver<EscrowAccounts>,
+        escrow_accounts_v1: Option<Receiver<EscrowAccounts>>,
+        escrow_accounts_v2: Option<Receiver<EscrowAccounts>>,
     ) -> Self {
         Self {
             escrow_accounts_v1,
@@ -36,17 +36,32 @@ impl Check<TapReceipt> for SenderBalanceCheck {
         ctx: &tap_core::receipt::Context,
         receipt: &CheckingReceipt,
     ) -> CheckResult {
-        let escrow_accounts_snapshot_v1 = self.escrow_accounts_v1.borrow();
-        let escrow_accounts_snapshot_v2 = self.escrow_accounts_v2.borrow();
-
         let Sender(receipt_sender) = ctx
             .get::<Sender>()
             .ok_or(CheckError::Failed(anyhow::anyhow!("Could not find sender")))?;
 
         // get balance for escrow account given receipt type
         let balance_result = match receipt.signed_receipt() {
-            TapReceipt::V1(_) => escrow_accounts_snapshot_v1.get_balance_for_sender(receipt_sender),
-            TapReceipt::V2(_) => escrow_accounts_snapshot_v2.get_balance_for_sender(receipt_sender),
+            TapReceipt::V1(_) => {
+                if let Some(ref escrow_accounts_v1) = self.escrow_accounts_v1 {
+                    let escrow_accounts_snapshot_v1 = escrow_accounts_v1.borrow();
+                    escrow_accounts_snapshot_v1.get_balance_for_sender(receipt_sender)
+                } else {
+                    return Err(CheckError::Failed(anyhow!(
+                        "Receipt v1 received but no escrow accounts v1 watcher is available"
+                    )));
+                }
+            }
+            TapReceipt::V2(_) => {
+                if let Some(ref escrow_accounts_v2) = self.escrow_accounts_v2 {
+                    let escrow_accounts_snapshot_v2 = escrow_accounts_v2.borrow();
+                    escrow_accounts_snapshot_v2.get_balance_for_sender(receipt_sender)
+                } else {
+                    return Err(CheckError::Failed(anyhow!(
+                        "Receipt v2 received but no escrow accounts v2 watcher is available"
+                    )));
+                }
+            }
         };
 
         // Check that the sender has a non-zero balance -- more advanced accounting is done in

crates/tap-agent/src/agent.rs

@@ -176,14 +176,12 @@ pub async fn start_agent() -> (ActorRef<SenderAccountsManagerMessage>, JoinHandl
 
     let config = Box::leak(Box::new({
         let mut config = SenderAccountConfig::from_config(&CONFIG);
-        // Enable Horizon support
-        config.horizon_enabled = true;
-        // Add a warning log so operators know their setting was ignore
+        // Use the configuration setting for Horizon support
+        config.horizon_enabled = CONFIG.horizon.enabled;
         if CONFIG.horizon.enabled {
-            tracing::warn!(
-            "Horizon support is configured as enabled but has been forcibly disabled as it's not fully supported yet. \
-            This is a temporary measure until Horizon support is stable."
-        );
+            tracing::info!("Horizon support is enabled");
+        } else {
+            tracing::info!("Horizon support is disabled");
         }
         config
     }));