refactor: move tap manager into indexer-common

Author: Jannis

Cargo.lock

@@ -21,11 +21,19 @@ reqwest = "0.11.20"
 secp256k1 = { version = "0.28.0", features = ["recovery"] }
 serde = { version = "1.0.188", features = ["derive"] }
 serde_json = "1.0.107"
+sqlx = { version = "0.7.1", features = [
+  "postgres",
+  "runtime-tokio",
+  "bigdecimal",
+  "rust_decimal",
+  "time",
+] }
 tokio = { version = "1.32.0", features = ["full", "macros", "rt"] }
 toolshed = { git = "https://github.com/edgeandnode/toolshed", branch = "main", features = [
   "graphql",
 ] }
 graphql = { git = "https://github.com/edgeandnode/toolshed", branch = "main" }
+tap_core = "0.6.0"
 
 [dev-dependencies]
 env_logger = "0.9.0"

common/Cargo.toml

@@ -7,6 +7,7 @@ pub mod escrow_accounts;
 pub mod graphql;
 pub mod signature_verification;
 pub mod subgraph_client;
+pub mod tap_manager;
 
 #[cfg(test)]
 mod test_vectors;
@@ -20,4 +21,5 @@ pub mod prelude {
     };
     pub use super::escrow_accounts::escrow_accounts;
     pub use super::subgraph_client::SubgraphClient;
+    pub use super::tap_manager::TapManager;
 }

common/src/lib.rs

@@ -3,15 +3,15 @@
 
 use alloy_primitives::Address;
 use alloy_sol_types::Eip712Domain;
+use anyhow::anyhow;
 use ethers_core::types::U256;
 use eventuals::Eventual;
-use indexer_common::prelude::Allocation;
 use log::error;
 use sqlx::{types::BigDecimal, PgPool};
 use std::{collections::HashMap, sync::Arc};
 use tap_core::tap_manager::SignedReceipt;
 
-use crate::query_processor::QueryError;
+use crate::prelude::Allocation;
 
 #[derive(Clone)]
 pub struct TapManager {
@@ -42,7 +42,10 @@ impl TapManager {
     ///
     /// The rest of the TAP receipt checks are expected to be performed out-of-band by the receipt aggregate requester
     /// service.
-    pub async fn verify_and_store_receipt(&self, receipt: SignedReceipt) -> Result<(), QueryError> {
+    pub async fn verify_and_store_receipt(
+        &self,
+        receipt: SignedReceipt,
+    ) -> Result<(), anyhow::Error> {
         let allocation_id = &receipt.message.allocation_id;
         if !self
             .indexer_allocations
@@ -51,17 +54,17 @@ impl TapManager {
             .map(|allocations| allocations.contains_key(allocation_id))
             .unwrap_or(false)
         {
-            return Err(QueryError::Other(anyhow::Error::msg(format!(
+            return Err(anyhow!(
                 "Receipt allocation ID `{}` is not eligible for this indexer",
                 allocation_id
-            ))));
+            ));
         }
 
         let receipt_signer = receipt
             .recover_signer(self.domain_separator.as_ref())
             .map_err(|e| {
                 error!("Failed to recover receipt signer: {}", e);
-                QueryError::Other(anyhow::Error::from(e))
+                anyhow!(e)
             })?;
         if !self
             .escrow_accounts
@@ -74,10 +77,10 @@ impl TapManager {
             })
             .unwrap_or(false)
         {
-            return Err(QueryError::Other(anyhow::Error::msg(format!(
+            return Err(anyhow!(
                 "Receipt sender `{}` is not eligible for this indexer",
                 receipt_signer
-            ))));
+            ));
         }
 
         // TODO: consider doing this in another async task to avoid slowing down the paid query flow.
@@ -91,13 +94,13 @@ impl TapManager {
                 .unwrap()
                 .to_owned(),
             BigDecimal::from(receipt.message.timestamp_ns),
-            serde_json::to_value(receipt).map_err(|e| QueryError::Other(anyhow::Error::from(e)))?
+            serde_json::to_value(receipt).map_err(|e| anyhow!(e))?
         )
         .execute(&self.pgpool)
         .await
         .map_err(|e| {
             error!("Failed to store receipt: {}", e);
-            QueryError::Other(anyhow::Error::from(e))
+            anyhow!(e)
         })?;
 
         Ok(())
@@ -108,11 +111,11 @@ impl TapManager {
 mod test {
     use std::str::FromStr;
 
+    use crate::prelude::{AllocationStatus, SubgraphDeployment};
     use alloy_primitives::Address;
     use alloy_sol_types::{eip712_domain, Eip712Domain};
-    use ethereum_types::{H256, U256};
     use ethers::signers::{coins_bip39::English, LocalWallet, MnemonicBuilder, Signer};
-    use indexer_common::prelude::{AllocationStatus, SubgraphDeployment};
+    use keccak_hash::H256;
     use sqlx::postgres::PgListener;
 
     use tap_core::tap_manager::SignedReceipt;

service/src/tap_manager.rs renamed to common/src/tap_manager.rs

@@ -8,8 +8,11 @@ use ethereum_types::U256;
 use std::{net::SocketAddr, str::FromStr, time::Duration};
 use tracing::info;
 
-use indexer_common::prelude::{
-    attestation_signers, dispute_manager, escrow_accounts, indexer_allocations, SubgraphClient,
+use indexer_common::{
+    prelude::{
+        attestation_signers, dispute_manager, escrow_accounts, indexer_allocations, SubgraphClient,
+    },
+    tap_manager::TapManager,
 };
 
 use util::{package_version, shutdown_signal};
@@ -27,7 +30,6 @@ mod graph_node;
 mod metrics;
 mod query_processor;
 mod server;
-mod tap_manager;
 mod util;
 
 #[cfg(test)]
@@ -119,7 +121,7 @@ async fn main() -> Result<(), std::io::Error> {
         Duration::from_secs(config.escrow_subgraph.escrow_syncing_interval),
     );
 
-    let tap_manager = tap_manager::TapManager::new(
+    let tap_manager = TapManager::new(
         indexer_management_db.clone(),
         indexer_allocations,
         escrow_accounts,

service/src/main.rs

@@ -5,6 +5,7 @@ use std::collections::HashMap;
 
 use alloy_primitives::Address;
 use eventuals::Eventual;
+use indexer_common::tap_manager::TapManager;
 use log::error;
 use serde::{Deserialize, Serialize};
 use tap_core::tap_manager::SignedReceipt;
@@ -14,7 +15,6 @@ use toolshed::thegraph::DeploymentId;
 use indexer_common::prelude::AttestationSigner;
 
 use crate::graph_node::GraphNodeInstance;
-use crate::tap_manager::TapManager;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct QueryResult {
@@ -117,7 +117,8 @@ impl QueryProcessor {
 
         self.tap_manager
             .verify_and_store_receipt(parsed_receipt)
-            .await?;
+            .await
+            .map_err(QueryError::Other)?;
 
         let signers = self
             .attestation_signers