fix(service): 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

@@ -18,7 +18,7 @@ use indexer_dips::{
     server::{DipsServer, DipsServerContext},
     signers::EscrowSignerValidator,
 };
-use indexer_monitor::{escrow_accounts_v1, DeploymentDetails, SubgraphClient};
+use indexer_monitor::{escrow_accounts_v1, escrow_accounts_v2, DeploymentDetails, SubgraphClient};
 use release::IndexerServiceRelease;
 use reqwest::Url;
 use tap_core::tap_eip712_domain;
@@ -78,12 +78,18 @@ pub async fn run() -> anyhow::Result<()> {
     )
     .await;
 
-    let escrow_subgraph = create_subgraph_client(
-        http_client.clone(),
-        &config.graph_node,
-        &config.subgraphs.escrow.config,
-    )
-    .await;
+    let escrow_subgraph_v2 = if let Some(ref escrow_v2_config) = config.subgraphs.escrow_v2 {
+        Some(
+            create_subgraph_client(
+                http_client.clone(),
+                &config.graph_node,
+                &escrow_v2_config.config,
+            )
+            .await,
+        )
+    } else {
+        None
+    };
 
     // Establish Database connection necessary for serving indexer management
     // requests with defined schema
@@ -105,19 +111,161 @@ 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();
+    // Capture individual fields needed for DIPS before they get moved
+    let escrow_v1_query_url_for_dips = config.subgraphs.escrow.config.query_url.clone();
+    let escrow_v2_query_url_for_dips = config
+        .subgraphs
+        .escrow_v2
+        .as_ref()
+        .map(|c| c.config.query_url.clone());
+
+    // Configure router with escrow watchers based on Horizon mode
+    use indexer_config::HorizonMode;
+    let router = match config.horizon.mode {
+        HorizonMode::Legacy => {
+            tracing::info!("Horizon mode: Legacy - using escrow accounts v1 only");
+            // Only create v1 watcher for legacy mode
+            let escrow_subgraph_v1 = create_subgraph_client(
+                http_client.clone(),
+                &config.graph_node,
+                &config.subgraphs.escrow.config,
+            )
+            .await;
+
+            let v1_watcher = indexer_monitor::escrow_accounts_v1(
+                escrow_subgraph_v1,
+                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()
+        }
+        HorizonMode::Transition => {
+            tracing::info!("Horizon mode: Transition - using both escrow accounts v1 and v2");
+            // Create both watchers for transition mode using separate subgraph clients
+            let escrow_subgraph_v1 = create_subgraph_client(
+                http_client.clone(),
+                &config.graph_node,
+                &config.subgraphs.escrow.config,
+            )
+            .await;
+
+            let v1_watcher = indexer_monitor::escrow_accounts_v1(
+                escrow_subgraph_v1,
+                indexer_address,
+                config.subgraphs.escrow.config.syncing_interval_secs,
+                true, // Reject thawing signers eagerly
+            )
+            .await
+            .expect("Error creating escrow_accounts_v1 channel");
+
+            if let Some(escrow_v2_subgraph) = escrow_subgraph_v2 {
+                let v2_watcher = indexer_monitor::escrow_accounts_v2(
+                    escrow_v2_subgraph,
+                    indexer_address,
+                    config
+                        .subgraphs
+                        .escrow_v2
+                        .as_ref()
+                        .unwrap()
+                        .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_v1(v1_watcher)
+                    .escrow_accounts_v2(v2_watcher)
+                    .build()
+            } else {
+                tracing::warn!("Horizon mode is Transition but no escrow_v2 configuration provided, falling back to v1 only");
+                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()
+            }
+        }
+        HorizonMode::Full => {
+            tracing::info!("Horizon mode: Full - using escrow accounts v2 only");
+            // Only create v2 watcher for full Horizon mode
+            let v2_subgraph = if let Some(escrow_v2_subgraph) = escrow_subgraph_v2 {
+                escrow_v2_subgraph
+            } else {
+                tracing::warn!("Horizon mode is Full but no escrow_v2 configuration provided, falling back to escrow v1 endpoint for v2 queries");
+                create_subgraph_client(
+                    http_client.clone(),
+                    &config.graph_node,
+                    &config.subgraphs.escrow.config,
+                )
+                .await
+            };
+            let v2_config = config
+                .subgraphs
+                .escrow_v2
+                .as_ref()
+                .map(|c| &c.config)
+                .unwrap_or(&config.subgraphs.escrow.config);
+
+            let v2_watcher = indexer_monitor::escrow_accounts_v2(
+                v2_subgraph,
+                indexer_address,
+                v2_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()
+        }
+    };
 
     serve_metrics(config.metrics.get_socket_addr());
 
@@ -143,14 +291,64 @@ pub async fn run() -> anyhow::Result<()> {
             Arc::new(IpfsClient::new(ipfs_url.as_str()).unwrap());
 
         // TODO: Try to re-use the same watcher for both DIPS and TAP
-        let watcher = escrow_accounts_v1(
-            escrow_subgraph,
-            indexer_address,
-            Duration::from_secs(500),
-            true,
-        )
-        .await
-        .expect("Failed to create escrow accounts watcher");
+        // DIPS is part of Horizon/v2, so use v2 escrow watcher when available
+        let dips_http_client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(60))
+            .build()
+            .expect("Failed to init HTTP client");
+
+        let escrow_subgraph_for_dips = if let Some(ref escrow_v2_url) = escrow_v2_query_url_for_dips
+        {
+            tracing::info!("DIPS using v2 escrow subgraph");
+            // Create subgraph client for v2
+            Box::leak(Box::new(
+                SubgraphClient::new(
+                    dips_http_client,
+                    None, // No local deployment
+                    DeploymentDetails::for_query_url_with_token(
+                        escrow_v2_url.clone(),
+                        None, // No auth token
+                    ),
+                )
+                .await,
+            ))
+        } else {
+            tracing::info!("DIPS falling back to v1 escrow subgraph");
+            // Create subgraph client for v1
+            Box::leak(Box::new(
+                SubgraphClient::new(
+                    dips_http_client,
+                    None, // No local deployment
+                    DeploymentDetails::for_query_url_with_token(
+                        escrow_v1_query_url_for_dips,
+                        None, // No auth token
+                    ),
+                )
+                .await,
+            ))
+        };
+
+        let watcher = if escrow_v2_query_url_for_dips.is_some() {
+            // Use v2 watcher for DIPS when v2 is available
+            escrow_accounts_v2(
+                escrow_subgraph_for_dips,
+                indexer_address,
+                Duration::from_secs(500),
+                true,
+            )
+            .await
+            .expect("Failed to create escrow accounts v2 watcher for DIPS")
+        } else {
+            // Fall back to v1 watcher
+            escrow_accounts_v1(
+                escrow_subgraph_for_dips,
+                indexer_address,
+                Duration::from_secs(500),
+                true,
+            )
+            .await
+            .expect("Failed to create escrow accounts v1 watcher for DIPS")
+        };
 
         let registry = NetworksRegistry::from_latest_version().await.unwrap();