fix: no need for collection in tap-agent after all

Author: pcarranzav

Cargo.lock

@@ -153,4 +153,3 @@ max_receipts_per_request = 10000
 host = "0.0.0.0"
 port = "7601"
 allowed_payers = ["0x3333333333333333333333333333333333333333"]
-dipper_grpc_url = "https://dipper.thegraph.com/"

crates/config/maximal-config-example.toml

@@ -390,7 +390,6 @@ pub struct DipsConfig {
     pub host: String,
     pub port: String,
     pub allowed_payers: Vec<Address>,
-    pub dipper_grpc_url: String,
 }
 
 impl Default for DipsConfig {
@@ -399,7 +398,6 @@ impl Default for DipsConfig {
             host: "0.0.0.0".to_string(),
             port: "7601".to_string(),
             allowed_payers: vec![],
-            dipper_grpc_url: "".to_string(),
         }
     }
 }
@@ -464,7 +462,6 @@ mod tests {
             allowed_payers: vec![Address(
                 FixedBytes::<20>::from_str("0x3333333333333333333333333333333333333333").unwrap(),
             )],
-            dipper_grpc_url: "https://dipper.thegraph.com/".to_string(),
             ..Default::default()
         });
 

crates/config/src/config.rs

@@ -50,49 +50,9 @@ impl AgreementStore for PsqlAgreementStore {
             cancelled,
             current_allocation_id: item.current_allocation_id,
             last_allocation_id: item.last_allocation_id,
+            last_payment_collected_at: item.last_payment_collected_at,
         }))
     }
-    async fn get_by_last_allocation_id(
-        &self,
-        allocation_id: String,
-    ) -> Result<Vec<StoredIndexingAgreement>, DipsError> {
-        let items = sqlx::query!(
-            "SELECT * FROM indexing_agreements WHERE last_allocation_id=$1",
-            allocation_id,
-        )
-        .fetch_all(&self.pool)
-        .await;
-
-        let items = match items {
-            Ok(items) => items,
-            Err(sqlx::Error::RowNotFound) => return Ok(vec![]),
-            Err(err) => return Err(DipsError::UnknownError(err.into())),
-        };
-
-        // Note: we discard any agreements that fail to decode
-        let agreements = items
-            .into_iter()
-            .map(|item| {
-                let signed =
-                    SignedIndexingAgreementVoucher::abi_decode(item.signed_payload.as_ref(), true)
-                        .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
-                let metadata = SubgraphIndexingVoucherMetadata::abi_decode(
-                    signed.voucher.metadata.as_ref(),
-                    true,
-                )
-                .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
-                Ok(StoredIndexingAgreement {
-                    voucher: signed,
-                    metadata,
-                    cancelled: item.cancelled_at.is_some(),
-                    current_allocation_id: item.current_allocation_id,
-                    last_allocation_id: item.last_allocation_id,
-                })
-            })
-            .filter_map(|agreement: Result<StoredIndexingAgreement, DipsError>| agreement.ok())
-            .collect();
-        Ok(agreements)
-    }
     async fn create_agreement(
         &self,
         agreement: SignedIndexingAgreementVoucher,
@@ -121,7 +81,7 @@ impl AgreementStore for PsqlAgreementStore {
         let min_epochs_per_collection: i64 = agreement.voucher.minEpochsPerCollection.into();
         let max_epochs_per_collection: i64 = agreement.voucher.maxEpochsPerCollection.into();
         sqlx::query!(
-            "INSERT INTO indexing_agreements VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,null,null,null,null)",
+            "INSERT INTO indexing_agreements VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,null,null,null,null,null)",
             id,
             agreement.signature.as_ref(),
             bs,

crates/dips/src/database.rs

@@ -20,7 +20,6 @@ pub mod store;
 
 use store::AgreementStore;
 use thiserror::Error;
-use tonic::transport::Channel;
 use uuid::Uuid;
 
 /// The Arbitrum One (mainnet) chain ID (eip155).
@@ -153,8 +152,6 @@ pub enum DipsError {
     AgreementCancelled,
     #[error("invalid voucher: {0}")]
     InvalidVoucher(String),
-    #[error("connection error: {0}")]
-    ConnectionError(#[from] tonic::transport::Error),
 }
 
 // TODO: send back messages
@@ -329,25 +326,6 @@ pub async fn validate_and_cancel_agreement(
     Ok(id)
 }
 
-pub async fn collect_indexing_agreements(
-    store: Arc<dyn AgreementStore>,
-    allocation_id: Address,
-    _dipper_client: &DipperServiceClient<Channel>,
-) -> Result<(), DipsError> {
-    let agreements = store
-        .get_by_last_allocation_id(allocation_id.encode_hex())
-        .await?;
-
-    for _agreement in agreements {
-        // TODO get the entities count for the deployment from graph node
-        // TODO create collection request
-        // TODO sign collection request
-        // TODO send collection request to the dipper via grpc
-    }
-
-    Ok(())
-}
-
 #[cfg(test)]
 mod test {
     use std::{

crates/dips/src/lib.rs

@@ -4,6 +4,7 @@
 use std::collections::HashMap;
 
 use async_trait::async_trait;
+use build_info::chrono::{DateTime, Utc};
 use uuid::Uuid;
 
 use crate::{
@@ -18,15 +19,12 @@ pub struct StoredIndexingAgreement {
     pub cancelled: bool,
     pub current_allocation_id: Option<String>,
     pub last_allocation_id: Option<String>,
+    pub last_payment_collected_at: Option<DateTime<Utc>>,
 }
 
 #[async_trait]
 pub trait AgreementStore: Sync + Send + std::fmt::Debug {
     async fn get_by_id(&self, id: Uuid) -> Result<Option<StoredIndexingAgreement>, DipsError>;
-    async fn get_by_last_allocation_id(
-        &self,
-        allocation_id: String,
-    ) -> Result<Vec<StoredIndexingAgreement>, DipsError>;
     async fn create_agreement(
         &self,
         agreement: SignedIndexingAgreementVoucher,
@@ -53,19 +51,6 @@ impl AgreementStore for InMemoryAgreementStore {
             .get(&id)
             .cloned())
     }
-    async fn get_by_last_allocation_id(
-        &self,
-        allocation_id: String,
-    ) -> Result<Vec<StoredIndexingAgreement>, DipsError> {
-        Ok(self
-            .data
-            .try_read()
-            .map_err(|e| DipsError::UnknownError(e.into()))?
-            .values()
-            .filter(|agreement| agreement.last_allocation_id == Some(allocation_id.clone()))
-            .cloned()
-            .collect())
-    }
     async fn create_agreement(
         &self,
         agreement: SignedIndexingAgreementVoucher,
@@ -78,6 +63,7 @@ impl AgreementStore for InMemoryAgreementStore {
             cancelled: false,
             current_allocation_id: None,
             last_allocation_id: None,
+            last_payment_collected_at: None,
         };
         self.data
             .try_write()

crates/dips/src/store.rs

@@ -19,5 +19,6 @@ query ClosedAllocations(
         }
     ) {
       id
+      poi
     }
 }

crates/query/graphql/closed_allocations.query.graphql

@@ -124,7 +124,6 @@ pub async fn run() -> anyhow::Result<()> {
             host,
             port,
             allowed_payers,
-            dipper_grpc_url: _,
         } = dips;
 
         let addr = format!("{}:{}", host, port)

crates/service/src/service.rs

@@ -16,7 +16,6 @@ indexer-monitor = { path = "../monitor" }
 indexer-watcher = { path = "../watcher" }
 indexer-allocation = { path = "../allocation" }
 indexer-config = { path = "../config" }
-indexer-dips = { path = "../dips" }
 indexer-query = { path = "../query" }
 indexer-receipt = { path = "../indexer-receipt" }
 anyhow.workspace = true

crates/tap-agent/Cargo.toml

@@ -4,18 +4,12 @@
 use std::{
     collections::{HashMap, HashSet},
     str::FromStr,
-    sync::Arc,
     time::Duration,
 };
 
 use anyhow::Context;
 use bigdecimal::{num_bigint::ToBigInt, ToPrimitive};
 use futures::{stream, StreamExt};
-use indexer_dips::{
-    collect_indexing_agreements, database::PsqlAgreementStore,
-    proto::gateway::graphprotocol::gateway::dips::dipper_service_client::DipperServiceClient,
-    store::AgreementStore, DipsError,
-};
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use indexer_query::{
     closed_allocations::{self, ClosedAllocations},
@@ -53,6 +47,7 @@ use crate::{
     tap::context::{Horizon, Legacy},
     tracker::{SenderFeeTracker, SimpleFeeTracker},
 };
+
 lazy_static! {
     static ref SENDER_DENIED: IntGaugeVec =
         register_int_gauge_vec!("tap_sender_denied", "Sender is denied", &["sender"]).unwrap();
@@ -324,10 +319,6 @@ pub struct State {
 
     // Config forwarded to [SenderAllocation]
     config: &'static SenderAccountConfig,
-    /// Indexing agreement store
-    agreement_store: Arc<dyn AgreementStore>,
-    /// Client for the dipper grpc server
-    dipper_client: Option<DipperServiceClient<Channel>>,
 }
 
 /// Configuration derived from config.toml
@@ -352,8 +343,6 @@ pub struct SenderAccountConfig {
     ///
     /// This is reached if the database is too slow
     pub tap_sender_timeout: Duration,
-    /// URL of the dipper grpc server
-    pub dipper_grpc_url: Option<String>,
 }
 
 impl SenderAccountConfig {
@@ -368,10 +357,6 @@ impl SenderAccountConfig {
             trigger_value: config.tap.get_trigger_value(),
             rav_request_timeout: config.tap.rav_request.request_timeout_secs,
             tap_sender_timeout: config.tap.sender_timeout_secs,
-            dipper_grpc_url: config
-                .dips
-                .as_ref()
-                .map(|dips_config| dips_config.dipper_grpc_url.clone()),
         }
     }
 }
@@ -836,19 +821,6 @@ impl Actor for SenderAccount {
         // wiremock_grpc used for tests doesn't support Zstd compression
         #[cfg(not(test))]
         let aggregator_v2 = aggregator_v2.send_compressed(tonic::codec::CompressionEncoding::Zstd);
-
-        let agreement_store = Arc::new(PsqlAgreementStore {
-            pool: pgpool.clone(),
-        });
-
-        let dipper_client = if let Some(ref url) = config.dipper_grpc_url {
-            let endpoint = tonic::transport::Endpoint::from_str(url)
-                .map_err(DipsError::ConnectionError)?
-                .connect_lazy();
-            Some(DipperServiceClient::new(endpoint))
-        } else {
-            None
-        };
         let state = State {
             prefix,
             sender_fee_tracker: SenderFeeTracker::new(config.rav_request_buffer),
@@ -870,8 +842,6 @@ impl Actor for SenderAccount {
             aggregator_v2,
             backoff_info: BackoffInfo::default(),
             config,
-            agreement_store,
-            dipper_client,
         };
 
         stream::iter(allocation_ids)
@@ -1083,24 +1053,6 @@ impl Actor for SenderAccount {
                         if let Some(sender_handle) = ActorRef::<SenderAllocationMessage>::where_is(
                             state.format_sender_allocation(&allocation_id.address()),
                         ) {
-                            if let Some(ref dipper_client) = state.dipper_client {
-                                tracing::trace!(%allocation_id, "SenderAccount checking for indexing agreements to collect");
-                                match collect_indexing_agreements(
-                                    state.agreement_store.clone(),
-                                    allocation_id.address(),
-                                    dipper_client,
-                                )
-                                .await
-                                {
-                                    Ok(_) => {
-                                        tracing::info!(%allocation_id, "Indexing agreements collected")
-                                    }
-                                    Err(e) => {
-                                        tracing::warn!(%allocation_id, error = %e, "Error collecting indexing agreements")
-                                    }
-                                }
-                            }
-
                             tracing::trace!(%allocation_id, "SenderAccount shutting down SenderAllocation");
                             // we can not send a rav request to this allocation
                             // because it's gonna trigger the last rav

crates/tap-agent/src/agent/sender_account.rs

@@ -81,7 +81,6 @@ pub fn get_sender_account_config() -> &'static SenderAccountConfig {
         indexer_address: INDEXER.1,
         escrow_polling_interval: ESCROW_POLLING_INTERVAL,
         tap_sender_timeout: Duration::from_secs(63),
-        dipper_grpc_url: None,
     }))
 }
 
@@ -111,7 +110,6 @@ pub async fn create_sender_account(
         indexer_address: INDEXER.1,
         escrow_polling_interval: Duration::default(),
         tap_sender_timeout: TAP_SENDER_TIMEOUT,
-        dipper_grpc_url: None,
     }));
 
     let network_subgraph = Box::leak(Box::new(

crates/tap-agent/src/test.rs

@@ -87,7 +87,6 @@ pub async fn start_agent(
         indexer_address: INDEXER_ADDRESS,
         escrow_polling_interval: Duration::from_secs(10),
         tap_sender_timeout: Duration::from_secs(30),
-        dipper_grpc_url: None,
     }));
 
     let args = SenderAccountsManagerArgs {

crates/tap-agent/tests/tap_agent_test.rs

@@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS indexing_agreements (
     signed_cancellation_payload BYTEA,
 
     current_allocation_id CHAR(40),
-    last_allocation_id CHAR(40)
+    last_allocation_id CHAR(40),
+    last_payment_collected_at TIMESTAMP WITH TIME ZONE
 );
 
 CREATE UNIQUE INDEX IX_UNIQ_SIGNATURE_AGREEMENT on indexing_agreements(signature);