removed eventuals

Author: taslimmuhammed

Cargo.lock

@@ -18,7 +18,6 @@ prometheus = "0.13.3"
 anyhow = { version = "1.0.72" }
 thiserror = "1.0.49"
 async-trait = "0.1.72"
-eventuals = "0.6.7"
 reqwest = { version = "0.12", features = [
     "charset",
     "h2",

Cargo.toml

@@ -9,7 +9,6 @@ thiserror.workspace = true
 async-trait.workspace = true
 alloy.workspace = true
 anyhow.workspace = true
-eventuals.workspace = true
 reqwest.workspace = true
 sqlx.workspace = true
 tap_core.workspace = true

common/Cargo.toml

@@ -118,7 +118,6 @@ where
         }
 
         let status = match self {
-
             Unauthorized => StatusCode::UNAUTHORIZED,
 
             NoSignerForAllocation(_) | FailedToSignAttestation => StatusCode::INTERNAL_SERVER_ERROR,

common/src/indexer_service/http/indexer_service.rs

@@ -191,23 +191,20 @@ impl DeploymentClient {
             .send()
             .await?)
     }
-    fn check_deployment_status(&self)->Result<(), anyhow::Error>{
+    fn check_deployment_status(&self) -> Result<(), anyhow::Error> {
         if let Some(ref status_rx) = self.status {
             let status_ref = status_rx.borrow();
             let status = status_ref.as_ref();
-            match status{
-                Some(deployment_status)=>{
-                   if !deployment_status.synced || &deployment_status.health != "healthy" {
+            match status {
+                Some(deployment_status) => {
+                    if !deployment_status.synced || &deployment_status.health != "healthy" {
                         return Err(anyhow!(
                             "Deployment `{}` is not ready or healthy to be queried",
                             self.query_url
                         ));
                     }
                 }
-                None => return Err(anyhow!(
-                    "Deployment `{}` is not available",
-                    self.query_url
-                ))
+                None => return Err(anyhow!("Deployment `{}` is not available", self.query_url)),
             }
         }
         Ok(())
@@ -443,7 +440,7 @@ mod test {
 
         // Waiting for status to propegate
         tokio::time::sleep(Duration::from_millis(1000)).await;
-        
+
         // Query the subgraph
         let data = client
             .query::<UserQuery, _>(user_query::Variables {})
@@ -619,4 +616,4 @@ mod test {
 
         assert_eq!(data.user.name, "remote".to_string());
     }
-}
+}

common/src/subgraph_client/client.rs

@@ -1,16 +1,19 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::time::Duration;
 use reqwest::Url;
 use serde::Deserialize;
 use serde_json::json;
+use std::time::Duration;
 use thegraph_core::DeploymentId;
 use thegraph_graphql_http::{
     http::request::IntoRequestParameters,
     http_client::{ReqwestExt, ResponseResult},
 };
-use tokio::{sync::watch::{self, Receiver}, time::{self, sleep}};
+use tokio::{
+    sync::watch::{self, Receiver},
+    time::{self, sleep},
+};
 use tracing::warn;
 
 use super::Query;
@@ -57,29 +60,26 @@ pub fn monitor_deployment_status(
                 [("ids", json!([deployment.to_string()]))],
             );
             let response = query::<DeploymentStatusResponse>(status_url, body)
-            .await
-            .map_err(|e| {
-                format!("Failed to query status of deployment `{deployment}`: {e}")
-            }).unwrap();
+                .await
+                .map_err(|e| format!("Failed to query status of deployment `{deployment}`: {e}"))
+                .unwrap();
             let result = response.map_err(|e| format!("{e}")).and_then(|data| {
                 data.indexing_statuses
                     .and_then(|statuses| statuses.first().cloned())
                     .ok_or_else(|| format!("Deployment `{deployment}` not found"))
             });
-            match result{
+            match result {
                 Ok(deployment_status) => {
-                    dbg!("Sending......");
-                    tx
-                    .send(Some(deployment_status))
-                    .expect("Failed to update deployment_status channel");
-                },
-                Err(err) =>{
+                    tx.send(Some(deployment_status))
+                        .expect("Failed to update deployment_status channel");
+                }
+                Err(err) => {
                     warn!(
                         "Error querying deployment status for `{}`: {}",
                         deployment, err
                     );
                     sleep(Duration::from_secs(15)).await;
-                },
+                }
             }
         }
     });
@@ -122,15 +122,15 @@ mod tests {
             .mount(&mock_server)
             .await;
 
-            let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await.unwrap();
-            let status_ref = status_rx.borrow();
-            assert_eq!(
-                status_ref.as_ref().unwrap().clone(),
-                DeploymentStatus {
-                    synced: true,
-                    health: "healthy".to_string()
-                }
+        let mut status_rx = monitor_deployment_status(deployment, status_url);
+        status_rx.changed().await.unwrap();
+        let status_ref = status_rx.borrow();
+        assert_eq!(
+            status_ref.as_ref().unwrap().clone(),
+            DeploymentStatus {
+                synced: true,
+                health: "healthy".to_string()
+            }
         );
     }
 
@@ -161,15 +161,15 @@ mod tests {
             .mount(&mock_server)
             .await;
 
-            let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await.unwrap();
-            let status_ref = status_rx.borrow();
-            assert_eq!(
-                status_ref.as_ref().unwrap().clone(),
-                DeploymentStatus {
-                    synced: false,
-                    health: "healthy".to_string()
-                }
+        let mut status_rx = monitor_deployment_status(deployment, status_url);
+        status_rx.changed().await.unwrap();
+        let status_ref = status_rx.borrow();
+        assert_eq!(
+            status_ref.as_ref().unwrap().clone(),
+            DeploymentStatus {
+                synced: false,
+                health: "healthy".to_string()
+            }
         );
     }
 
@@ -200,15 +200,15 @@ mod tests {
             .mount(&mock_server)
             .await;
 
-            let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await.unwrap();
-            let status_ref = status_rx.borrow();
-            assert_eq!(
-                status_ref.as_ref().unwrap().clone(),
-                DeploymentStatus {
-                    synced: true,
-                    health: "unhealthy".to_string()
-                }
+        let mut status_rx = monitor_deployment_status(deployment, status_url);
+        status_rx.changed().await.unwrap();
+        let status_ref = status_rx.borrow();
+        assert_eq!(
+            status_ref.as_ref().unwrap().clone(),
+            DeploymentStatus {
+                synced: true,
+                health: "unhealthy".to_string()
+            }
         );
     }
 

common/src/subgraph_client/monitor.rs

@@ -19,7 +19,6 @@ reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
-eventuals.workspace = true
 tracing.workspace = true
 prometheus.workspace = true
 axum.workspace = true

tap-agent/Cargo.toml

@@ -5,19 +5,21 @@ use std::time::Duration;
 
 use alloy::primitives::Address;
 use anyhow::anyhow;
-use eventuals::{Eventual, EventualExt};
 use graphql_client::GraphQLQuery;
 use indexer_common::subgraph_client::SubgraphClient;
 use tap_core::receipt::{
     checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
-use tokio::time::sleep;
+use tokio::{
+    sync::watch::{self, Receiver},
+    time::{self, sleep},
+};
 use tracing::error;
 
 pub struct AllocationId {
-    tap_allocation_redeemed: Eventual<bool>,
+    tap_allocation_redeemed: Receiver<Option<bool>>,
     allocation_id: Address,
 }
 
@@ -29,7 +31,7 @@ impl AllocationId {
         allocation_id: Address,
         escrow_subgraph: &'static SubgraphClient,
     ) -> Self {
-        let tap_allocation_redeemed = tap_allocation_redeemed_eventual(
+        let tap_allocation_redeemed = tap_allocation_redeemed_receiver(
             allocation_id,
             sender_id,
             indexer_address,
@@ -56,46 +58,56 @@ impl Check for AllocationId {
         };
 
         // Check that the allocation ID is not redeemed yet for this consumer
-        match self.tap_allocation_redeemed.value().await {
-            Ok(false) => Ok(()),
-            Ok(true) => Err(CheckError::Failed(anyhow!(
+        match *self.tap_allocation_redeemed.borrow() {
+            Some(false) => Ok(()),
+            Some(true) => Err(CheckError::Failed(anyhow!(
                 "Allocation {} already redeemed",
                 allocation_id
             ))),
-            Err(e) => Err(CheckError::Retryable(anyhow!(
-                "Could not get allocation escrow redemption status from eventual: {:?}",
-                e
+            None => Err(CheckError::Retryable(anyhow!(
+                "Could not get allocation escrow redemption status from channel"
             ))),
         }
     }
 }
 
-fn tap_allocation_redeemed_eventual(
+fn tap_allocation_redeemed_receiver(
     allocation_id: Address,
     sender_address: Address,
     indexer_address: Address,
     escrow_subgraph: &'static SubgraphClient,
     escrow_polling_interval: Duration,
-) -> Eventual<bool> {
-    eventuals::timer(escrow_polling_interval).map_with_retry(
-        move |_| async move {
-            query_escrow_check_transactions(
+) -> Receiver<Option<bool>> {
+    let (tx, rx) = watch::channel(None);
+    tokio::spawn(async move {
+        let mut time_interval = time::interval(escrow_polling_interval);
+        time_interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+        loop {
+            time_interval.tick().await;
+            let result = query_escrow_check_transactions(
                 allocation_id,
                 sender_address,
                 indexer_address,
                 escrow_subgraph,
             )
             .await
-            .map_err(|e| e.to_string())
-        },
-        move |error: String| {
-            error!(
-                "Failed to check the escrow redeem status for allocation {} and sender {}: {}",
-                allocation_id, sender_address, error
-            );
-            sleep(escrow_polling_interval.div_f32(2.))
-        },
-    )
+            .map_err(|e| e.to_string());
+            match result {
+                Ok(tap_allocation_redeemed) => {
+                    tx.send(Some(tap_allocation_redeemed))
+                        .expect("Failed to update tap_allocation_redeemed channel");
+                }
+                Err(err) => {
+                    error!(
+                        "Failed to check the escrow redeem status for allocation {} and sender {}: {}",
+                        allocation_id, sender_address, err
+                    );
+                    sleep(escrow_polling_interval.div_f32(2.)).await;
+                }
+            }
+        }
+    });
+    rx
 }
 
 #[derive(GraphQLQuery)]