refactor: use metrics middleware in request handler

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

crates/service/src/error.rs

@@ -30,9 +30,6 @@ pub enum IndexerServiceError {
     #[error("Failed to sign attestation")]
     FailedToSignAttestation,
 
-    #[error("Could not decode signer: {0}")]
-    CouldNotDecodeSigner(tap_core::Error),
-
     #[error("There was an error while accessing escrow account: {0}")]
     EscrowAccount(#[from] EscrowAccountsError),
 }
@@ -54,7 +51,6 @@ impl IntoResponse for IndexerServiceError {
             ReceiptError(_)
             | InvalidRequest(_)
             | InvalidFreeQueryAuthToken
-            | CouldNotDecodeSigner(_)
             | EscrowAccount(_)
             | ProcessingError(_) => StatusCode::BAD_REQUEST,
         };

crates/service/src/middleware.rs

@@ -7,3 +7,10 @@ mod inject_labels;
 mod inject_receipt;
 mod inject_sender;
 mod metrics;
+
+pub use inject_allocation::{allocation_middleware, AllocationState};
+pub use inject_deployment::deployment_middleware;
+pub use inject_labels::labels_middleware;
+pub use inject_receipt::receipt_middleware;
+pub use inject_sender::{sender_middleware, SenderState, Sender};
+pub use metrics::MetricsMiddlewareLayer;

crates/service/src/routes/request_handler.rs

@@ -3,11 +3,14 @@
 
 use std::sync::Arc;
 
-use crate::{error::IndexerServiceError, metrics::{FAILED_RECEIPT, HANDLER_FAILURE, HANDLER_HISTOGRAM}, tap::AgoraQuery};
+use crate::{
+    error::IndexerServiceError, metrics::FAILED_RECEIPT, middleware::Sender, tap::AgoraQuery,
+};
 use axum::{
     extract::{Path, State},
     http::HeaderMap,
     response::IntoResponse,
+    Extension,
 };
 use axum_extra::TypedHeader;
 use reqwest::StatusCode;
@@ -20,23 +23,8 @@ use crate::service::{AttestationOutput, IndexerServiceResponse, IndexerServiceSt
 
 pub async fn request_handler(
     Path(manifest_id): Path<DeploymentId>,
-    typed_header: TypedHeader<TapReceipt>,
-    state: State<Arc<IndexerServiceState>>,
-    headers: HeaderMap,
-    body: String,
-) -> Result<impl IntoResponse, IndexerServiceError> {
-    _request_handler(manifest_id, typed_header, state, headers, body)
-        .await
-        .inspect_err(|_| {
-            HANDLER_FAILURE
-                .with_label_values(&[&manifest_id.to_string()])
-                .inc()
-        })
-}
-
-async fn _request_handler(
-    manifest_id: DeploymentId,
     TypedHeader(receipt): TypedHeader<TapReceipt>,
+    Extension(sender): Extension<Sender>,
     State(state): State<Arc<IndexerServiceState>>,
     headers: HeaderMap,
     req: String,
@@ -87,30 +75,6 @@ async fn _request_handler(
         variables,
     });
 
-    // recover the signer address
-    // get escrow accounts from channel
-    // return sender from signer
-    //
-    // TODO: We are currently doing this process twice.
-    // One here and other on common/src/tap/checks/sender_balance_check.rs
-    // We'll get back to normal once we have attachable context to `verify_and_store_receipt`
-    let signer = receipt
-        .recover_signer(&state.domain_separator)
-        .map_err(IndexerServiceError::CouldNotDecodeSigner)?;
-    let sender = state
-        .escrow_accounts
-        .borrow()
-        .get_sender_for_signer(&signer)
-        .map_err(IndexerServiceError::EscrowAccount)?;
-
-    let _metric = HANDLER_HISTOGRAM
-        .with_label_values(&[
-            &manifest_id.to_string(),
-            &allocation_id.to_string(),
-            &sender.to_string(),
-        ])
-        .start_timer();
-
     // Verify the receipt and store it in the database
     state
         .tap_manager
@@ -121,7 +85,7 @@ async fn _request_handler(
                 .with_label_values(&[
                     &manifest_id.to_string(),
                     &allocation_id.to_string(),
-                    &sender.to_string(),
+                    &sender.0.to_string(),
                 ])
                 .inc()
         })

crates/service/src/service/indexer_service.rs

@@ -5,18 +5,18 @@ use alloy::dyn_abi::Eip712Domain;
 use anyhow;
 use axum::extract::MatchedPath;
 use axum::extract::Request as ExtractRequest;
-use axum::http::{Method, Request};
 use axum::{
+    http::{Method, Request},
+    middleware::{from_fn, from_fn_with_state},
     response::IntoResponse,
     routing::{get, post},
-    Json, Router,
+    serve, Json, Router, ServiceExt,
 };
-use axum::{serve, ServiceExt};
 use build_info::BuildInfo;
 use indexer_attestation::AttestationSigner;
 use indexer_monitor::{
-    attestation_signers, dispute_manager, escrow_accounts, indexer_allocations, DeploymentDetails,
-    EscrowAccounts, SubgraphClient,
+    attestation_signers, deployment_to_allocation, dispute_manager, escrow_accounts,
+    indexer_allocations, DeploymentDetails, EscrowAccounts, SubgraphClient,
 };
 use prometheus::TextEncoder;
 use reqwest::StatusCode;
@@ -30,12 +30,23 @@ use thegraph_core::{Address, Attestation};
 use tokio::net::TcpListener;
 use tokio::signal;
 use tokio::sync::watch::Receiver;
+use tower::ServiceBuilder;
 use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
 use tower_http::validate_request::ValidateRequestHeaderLayer;
 use tower_http::{cors, cors::CorsLayer, normalize_path::NormalizePath, trace::TraceLayer};
 use tracing::warn;
 use tracing::{error, info, info_span};
 
+use crate::metrics::HANDLER_FAILURE;
+use crate::metrics::HANDLER_HISTOGRAM;
+use crate::middleware::allocation_middleware;
+use crate::middleware::deployment_middleware;
+use crate::middleware::labels_middleware;
+use crate::middleware::receipt_middleware;
+use crate::middleware::sender_middleware;
+use crate::middleware::AllocationState;
+use crate::middleware::MetricsMiddlewareLayer;
+use crate::middleware::SenderState;
 use crate::routes::health;
 use crate::routes::request_handler;
 use crate::routes::static_subgraph_request_handler;
@@ -257,7 +268,7 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
 
     let checks = IndexerTapContext::get_checks(
         database,
-        allocations,
+        allocations.clone(),
         escrow_accounts.clone(),
         domain_separator.clone(),
         timestamp_error_tolerance,
@@ -276,8 +287,8 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
         attestation_signers,
         tap_manager,
         service_impl: options.service_impl,
-        escrow_accounts,
-        domain_separator,
+        escrow_accounts: escrow_accounts.clone(),
+        domain_separator: domain_separator.clone(),
     });
 
     // Rate limits by allowing bursts of 10 requests and requiring 100ms of
@@ -362,13 +373,39 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
 
     misc_routes = misc_routes.with_state(state.clone());
 
+    let deployment_to_allocation = deployment_to_allocation(allocations);
+    let allocation_state = AllocationState {
+        deployment_to_allocation,
+    };
+    let sender_state = SenderState {
+        escrow_accounts,
+        domain_separator,
+    };
+
+    let service_builder = ServiceBuilder::new()
+        // inject deployment id
+        .layer(from_fn(deployment_middleware))
+        // inject receipt
+        .layer(from_fn(receipt_middleware))
+        // inject deployment id
+        .layer(from_fn_with_state(allocation_state, allocation_middleware))
+        // inject sender
+        .layer(from_fn_with_state(sender_state, sender_middleware))
+        // inject metrics labels
+        .layer(from_fn(labels_middleware))
+        // metrics for histogram and failure
+        .layer(MetricsMiddlewareLayer::new(
+            HANDLER_HISTOGRAM.clone(),
+            HANDLER_FAILURE.clone(),
+        ));
+
     let data_routes = Router::new()
         .route(
             PathBuf::from(&options.config.service.url_prefix)
                 .join(format!("{}/id/:id", options.url_namespace))
                 .to_str()
                 .expect("Failed to set up `/{url_namespace}/id/:id` route"),
-            post(request_handler),
+            post(request_handler).route_layer(service_builder),
         )
         .with_state(state.clone());