reversed tokio watch for all except disputes

Author: taslimmuhammed

.sqlx/query-23cabe6cd63488809ca71db63f59e194653d2acaa6a74404c8675614759db032.json renamed to .sqlx/query-b1d4dfcd202af310df032edc34d83dabb56d7b947b023ee3a8b32b24b07bcd18.json

@@ -10,12 +10,10 @@ use std::{
 use super::Allocation;
 use crate::prelude::SubgraphClient;
 use alloy::primitives::{TxHash, B256, U256};
+use eventuals::{timer, Eventual, EventualExt};
 use graphql_client::GraphQLQuery;
 use thegraph_core::{Address, DeploymentId};
-use tokio::{
-    sync::watch::{self, Receiver},
-    time::{self, sleep},
-};
+use tokio::time::sleep;
 use tracing::warn;
 
 type BigInt = U256;
@@ -63,43 +61,30 @@ pub fn indexer_allocations(
     indexer_address: Address,
     interval: Duration,
     recently_closed_allocation_buffer: Duration,
-) -> Receiver<HashMap<Address, Allocation>> {
-    let (tx, rx) = watch::channel(HashMap::new());
-    tokio::spawn(async move {
-        let mut time_interval = time::interval(interval);
-        // Refresh indexer allocations every now and then
-        loop {
-            time_interval.tick().await;
-            let result = async {
-                get_allocations(
-                    network_subgraph,
-                    indexer_address,
-                    recently_closed_allocation_buffer,
-                )
-                .await
-                .map_err(|e| e.to_string())
-            }
-            .await;
-            match result {
-                Ok(res) => {
-                    if tx.send(res).is_err() {
-                        //stopping[something gone wrong with channel]
-                        break;
-                    }
-                }
-                Err(err) => {
-                    warn!(
-                        "Failed to fetch active or recently closed allocations for indexer {:?}: {}",
-                        indexer_address, err
-                    );
-
-                    // Sleep for a bit before we retry
-                    sleep(interval.div_f32(2.0)).await;
-                }
-            }
-        }
-    });
-    rx
+) -> Eventual<HashMap<Address, Allocation>> {
+    // Refresh indexer allocations every now and then
+    timer(interval).map_with_retry(
+        move |_| async move {
+            get_allocations(
+                network_subgraph,
+                indexer_address,
+                recently_closed_allocation_buffer,
+            )
+            .await
+            .map_err(|e| e.to_string())
+        },
+        // Need to use string errors here because eventuals `map_with_retry` retries
+        // errors that can be cloned
+        move |err: String| {
+            warn!(
+                "Failed to fetch active or recently closed allocations for indexer {:?}: {}",
+                indexer_address, err
+            );
+
+            // Sleep for a bit before we retry
+            sleep(interval.div_f32(2.0))
+        },
+    )
 }
 
 pub async fn get_allocations(

common/src/allocations/monitor.rs

@@ -24,11 +24,11 @@ struct DisputeManager;
 pub fn dispute_manager(
     network_subgraph: &'static SubgraphClient,
     interval: Duration,
-) -> Receiver<Address> {
-    let (tx, rx) = watch::channel(Address::default());
+) -> Receiver<Option<Address>> {
+    let (tx, rx) = watch::channel(None);
     tokio::spawn(async move {
         let mut time_interval = time::interval(interval);
-
+        time_interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
         loop {
             time_interval.tick().await;
 
@@ -48,7 +48,7 @@ pub fn dispute_manager(
 
             match result {
                 Ok(address) => {
-                    if tx.send(address).is_err() {
+                    if tx.send(Some(address)).is_err() {
                         // stopping
                         break;
                     }
@@ -118,6 +118,6 @@ mod test {
         let dispute_manager = dispute_manager(network_subgraph, Duration::from_secs(60));
         sleep(Duration::from_millis(50)).await;
         let result = *dispute_manager.borrow();
-        assert_eq!(result, *DISPUTE_MANAGER_ADDRESS);
+        assert_eq!(result.unwrap(), *DISPUTE_MANAGER_ADDRESS);
     }
 }

common/src/attestations/dispute_manager.rs

@@ -1,41 +1,58 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
+use eventuals::{Eventual, EventualExt};
 use std::collections::HashMap;
 use std::sync::Arc;
 use thegraph_core::{Address, ChainId};
-use tokio::sync::{watch, watch::Receiver, Mutex};
+use tokio::sync::watch;
+use tokio::{
+    select,
+    sync::{watch::Receiver, Mutex},
+};
 use tracing::warn;
 
 use crate::prelude::{Allocation, AttestationSigner};
 
 /// An always up-to-date list of attestation signers, one for each of the indexer's allocations.
 pub fn attestation_signers(
-    mut indexer_allocations: Receiver<HashMap<Address, Allocation>>,
+    indexer_allocations: Eventual<HashMap<Address, Allocation>>,
     indexer_mnemonic: String,
     chain_id: ChainId,
-    mut dispute_manager: Receiver<Address>,
-) -> Receiver<HashMap<Address, AttestationSigner>> {
+    mut dispute_manager_rx: Receiver<Option<Address>>,
+) -> Eventual<HashMap<Address, AttestationSigner>> {
     let attestation_signers_map: &'static Mutex<HashMap<Address, AttestationSigner>> =
         Box::leak(Box::new(Mutex::new(HashMap::new())));
+
     let indexer_mnemonic = Arc::new(indexer_mnemonic);
 
-    let (tx, rx) = watch::channel(HashMap::new());
+    // Whenever the indexer's active or recently closed allocations change, make sure
+    // we have attestation signers for all of them
+    let (mut signers_writer, signers_reader) =
+        Eventual::<HashMap<Address, AttestationSigner>>::new();
 
     tokio::spawn(async move {
+        //listening to the allocation eventual and converting them to reciever
+        //using pipe for updation
+        //for temporary pupose only
+        let (allocations_tx, mut allocations_rx) =
+            watch::channel(indexer_allocations.value().await.unwrap());
+        let _p1 = indexer_allocations.pipe(move |allocatons| {
+            let _ = allocations_tx.send(allocatons);
+        });
         loop {
-            tokio::select! {
-                Ok(_) = indexer_allocations.changed() => {},
-                Ok(_) = dispute_manager.changed() => {},
-                else => break,
+            select! {
+                Ok(_)= allocations_rx.changed() =>{},
+                Ok(_)= dispute_manager_rx.changed() =>{},
             }
-
-            let allocations = indexer_allocations.borrow().clone();
-            let dispute_manager = *dispute_manager.borrow();
-            let indexer_mnemonic = indexer_mnemonic.clone();
-
             let mut signers = attestation_signers_map.lock().await;
 
+            let allocations = allocations_rx.borrow().clone();
+            let dispute_manager = *dispute_manager_rx.borrow();
+            if dispute_manager.is_none() {
+                continue;
+            }
+            let dispute_manager = dispute_manager.unwrap();
             // Remove signers for allocations that are no longer active or recently closed
             signers.retain(|id, _| allocations.contains_key(id));
 
@@ -60,18 +77,15 @@ pub fn attestation_signers(
                 }
             }
 
-            // sending updated signers map
-            tx.send(signers.clone()).unwrap();
+            signers_writer.write(signers.clone());
         }
     });
 
-    rx
+    signers_reader
 }
 
 #[cfg(test)]
 mod tests {
-    use tokio::time::sleep;
-
     use crate::test_vectors::{
         DISPUTE_MANAGER_ADDRESS, INDEXER_ALLOCATIONS, INDEXER_OPERATOR_MNEMONIC,
     };
@@ -80,29 +94,29 @@ mod tests {
 
     #[tokio::test]
     async fn test_attestation_signers_update_with_allocations() {
-        let (allocations_tx, allocations_rx) = watch::channel(HashMap::new());
-        let (dispute_manager_tx, dispute_manager_rx) = watch::channel(Address::default());
+        let (mut allocations_writer, allocations) = Eventual::<HashMap<Address, Allocation>>::new();
+        let (dispute_manager_writer, dispute_manager) = watch::channel(None);
 
-        dispute_manager_tx.send(*DISPUTE_MANAGER_ADDRESS).unwrap();
+        dispute_manager_writer
+            .send(Some(*DISPUTE_MANAGER_ADDRESS))
+            .unwrap();
 
-        let signers_rx = attestation_signers(
-            allocations_rx,
+        let signers = attestation_signers(
+            allocations,
             (*INDEXER_OPERATOR_MNEMONIC).to_string(),
             1,
-            dispute_manager_rx,
+            dispute_manager,
         );
+        let mut signers = signers.subscribe();
 
         // Test that an empty set of allocations leads to an empty set of signers
-        allocations_tx.send(HashMap::new()).unwrap();
-        //change wait if required
-        sleep(std::time::Duration::from_millis(50)).await; // waiting for propegation
-        let latest_signers = signers_rx.borrow().clone();
+        allocations_writer.write(HashMap::new());
+        let latest_signers = signers.next().await.unwrap();
         assert_eq!(latest_signers, HashMap::new());
 
         // Test that writing our set of test allocations results in corresponding signers for all of them
-        allocations_tx.send((*INDEXER_ALLOCATIONS).clone()).unwrap();
-        sleep(std::time::Duration::from_millis(50)).await; // waiting for propegation
-        let latest_signers = signers_rx.borrow().clone();
+        allocations_writer.write((*INDEXER_ALLOCATIONS).clone());
+        let latest_signers = signers.next().await.unwrap();
         assert_eq!(latest_signers.len(), INDEXER_ALLOCATIONS.len());
         for signer_allocation_id in latest_signers.keys() {
             assert!(INDEXER_ALLOCATIONS

common/src/attestations/signers.rs

@@ -9,13 +9,11 @@ use std::{
 
 use alloy::primitives::U256;
 use anyhow::{anyhow, Result};
+use eventuals::{timer, Eventual, EventualExt};
 use graphql_client::GraphQLQuery;
 use thegraph_core::Address;
 use thiserror::Error;
-use tokio::{
-    sync::watch::{self, Receiver},
-    time::{self, sleep},
-};
+use tokio::time::sleep;
 use tracing::{error, warn};
 
 use crate::prelude::SubgraphClient;
@@ -107,34 +105,22 @@ pub fn escrow_accounts(
     indexer_address: Address,
     interval: Duration,
     reject_thawing_signers: bool,
-) -> Receiver<EscrowAccounts> {
-    let (tx, rx) = watch::channel(EscrowAccounts::default());
-    tokio::spawn(async move {
-        let mut time_interval = time::interval(interval);
+) -> Eventual<EscrowAccounts> {
+    timer(interval).map_with_retry(
+        move |_| async move {
+            get_escrow_accounts(escrow_subgraph, indexer_address, reject_thawing_signers)
+                .await
+                .map_err(|e| e.to_string())
+        },
+        move |err: String| {
+            error!(
+                "Failed to fetch escrow accounts for indexer {:?}: {}",
+                indexer_address, err
+            );
 
-        loop {
-            time_interval.tick().await;
-            let result =
-                get_escrow_accounts(escrow_subgraph, indexer_address, reject_thawing_signers)
-                    .await
-                    .map_err(|e| e.to_string());
-            match result {
-                Ok(escrow_accounts) => {
-                    if tx.send(escrow_accounts).is_err() {
-                        break; // something wrong with channel
-                    }
-                }
-                Err(err) => {
-                    error!(
-                        "Failed to fetch escrow accounts for indexer {:?}: {}",
-                        indexer_address, err
-                    );
-                    sleep(interval.div_f32(2.0)).await;
-                }
-            }
-        }
-    });
-    rx
+            sleep(interval.div_f32(2.0))
+        },
+    )
 }
 
 async fn get_escrow_accounts(
@@ -250,15 +236,15 @@ mod tests {
             );
         mock_server.register(mock).await;
 
-        let accounts_rx = escrow_accounts(
+        let accounts = escrow_accounts(
             escrow_subgraph,
             *test_vectors::INDEXER_ADDRESS,
             Duration::from_secs(60),
             true,
         );
-        sleep(Duration::from_millis(50)).await;
+
         assert_eq!(
-            accounts_rx.borrow().clone(),
+            accounts.value().await.unwrap(),
             EscrowAccounts::new(
                 test_vectors::ESCROW_ACCOUNTS_BALANCES.to_owned(),
                 test_vectors::ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),

common/src/escrow_accounts.rs

@@ -20,6 +20,7 @@ use axum::{
 };
 use axum::{serve, ServiceExt};
 use build_info::BuildInfo;
+use eventuals::Eventual;
 use prometheus::TextEncoder;
 use reqwest::StatusCode;
 use serde::{de::DeserializeOwned, Serialize};
@@ -29,7 +30,6 @@ use thegraph_core::{Address, Attestation, DeploymentId};
 use thiserror::Error;
 use tokio::net::TcpListener;
 use tokio::signal;
-use tokio::sync::watch::Receiver;
 use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
 use tower_http::{cors, cors::CorsLayer, normalize_path::NormalizePath, trace::TraceLayer};
 use tracing::error;
@@ -184,12 +184,12 @@ where
     I: IndexerServiceImpl + Sync + Send + 'static,
 {
     pub config: IndexerServiceConfig,
-    pub attestation_signers: Receiver<HashMap<Address, AttestationSigner>>,
+    pub attestation_signers: Eventual<HashMap<Address, AttestationSigner>>,
     pub tap_manager: Manager<IndexerTapContext>,
     pub service_impl: Arc<I>,
 
     // tap
-    pub escrow_accounts: Receiver<EscrowAccounts>,
+    pub escrow_accounts: Eventual<EscrowAccounts>,
     pub domain_separator: Eip712Domain,
 }