feat(all): add retryable checks (#293)

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

Author: gusinacio

Cargo.lock

@@ -38,7 +38,7 @@ sqlx = { version = "0.7.2", features = [
 tracing = { version = "0.1.40", default-features = false }
 bigdecimal = "0.4.3"
 build-info = "0.0.38"
-tap_core = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "9e1915b", default-features = false }
+tap_core = { git = "https://github.com/semiotic-ai/timeline-aggregation-protocol", rev = "ff856d9", default-features = false }
 tracing-subscriber = { version = "0.3", features = [
     "json",
     "env-filter",

Cargo.toml

@@ -8,7 +8,7 @@ use anyhow::anyhow;
 use eventuals::Eventual;
 
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -36,10 +36,10 @@ impl Check for AllocationEligible {
             .map(|allocations| allocations.contains_key(&allocation_id))
             .unwrap_or(false)
         {
-            return Err(anyhow!(
+            return Err(CheckError::Failed(anyhow!(
                 "Receipt allocation ID `{}` is not eligible for this indexer",
                 allocation_id
-            ));
+            )));
         }
         Ok(())
     }

common/src/tap/checks/allocation_eligible.rs

@@ -9,6 +9,7 @@ use sqlx::PgPool;
 use std::collections::HashSet;
 use std::sync::RwLock;
 use std::{str::FromStr, sync::Arc};
+use tap_core::receipt::checks::CheckError;
 use tap_core::receipt::{
     checks::{Check, CheckResult},
     state::Checking,
@@ -153,12 +154,16 @@ impl Check for DenyListCheck {
         let receipt_signer = receipt
             .signed_receipt()
             .recover_signer(&self.domain_separator)
-            .inspect_err(|e| {
+            .map_err(|e| {
                 error!("Failed to recover receipt signer: {}", e);
-            })?;
+                anyhow::anyhow!(e)
+            })
+            .map_err(CheckError::Failed)?;
         let escrow_accounts_snapshot = self.escrow_accounts.value_immediate().unwrap_or_default();
 
-        let receipt_sender = escrow_accounts_snapshot.get_sender_for_signer(&receipt_signer)?;
+        let receipt_sender = escrow_accounts_snapshot
+            .get_sender_for_signer(&receipt_signer)
+            .map_err(|e| CheckError::Failed(e.into()))?;
 
         // Check that the sender is not denylisted
         if self
@@ -167,10 +172,10 @@ impl Check for DenyListCheck {
             .unwrap()
             .contains(&receipt_sender)
         {
-            return Err(anyhow::anyhow!(
+            return Err(CheckError::Failed(anyhow::anyhow!(
                 "Received a receipt from a denylisted sender: {}",
                 receipt_sender
-            ));
+            )));
         }
 
         Ok(())

common/src/tap/checks/deny_list_check.rs

@@ -7,7 +7,7 @@ pub struct ReceiptMaxValueCheck {
 }
 
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -26,10 +26,10 @@ impl Check for ReceiptMaxValueCheck {
         if receipt_value < self.receipt_max_value {
             Ok(())
         } else {
-            Err(anyhow!(
+            Err(CheckError::Failed(anyhow!(
                 "Receipt value `{}` is higher than the limit set by the user",
                 receipt_value
-            ))
+            )))
         }
     }
 }

common/src/tap/checks/receipt_max_val_check.rs

@@ -7,7 +7,7 @@ use alloy::primitives::U256;
 use anyhow::anyhow;
 use eventuals::Eventual;
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -38,22 +38,25 @@ impl Check for SenderBalanceCheck {
             .recover_signer(&self.domain_separator)
             .inspect_err(|e| {
                 error!("Failed to recover receipt signer: {}", e);
-            })?;
+            })
+            .map_err(|e| CheckError::Failed(e.into()))?;
 
         // We bail if the receipt signer does not have a corresponding sender in the escrow
         // accounts.
-        let receipt_sender = escrow_accounts_snapshot.get_sender_for_signer(&receipt_signer)?;
+        let receipt_sender = escrow_accounts_snapshot
+            .get_sender_for_signer(&receipt_signer)
+            .map_err(|e| CheckError::Failed(e.into()))?;
 
         // Check that the sender has a non-zero balance -- more advanced accounting is done in
         // `tap-agent`.
         if !escrow_accounts_snapshot
             .get_balance_for_sender(&receipt_sender)
             .map_or(false, |balance| balance > U256::ZERO)
         {
-            return Err(anyhow!(
+            return Err(CheckError::Failed(anyhow!(
                 "Receipt sender `{}` does not have a sufficient balance",
                 receipt_signer,
-            ));
+            )));
         }
         Ok(())
     }

common/src/tap/checks/sender_balance_check.rs

@@ -8,7 +8,7 @@ pub struct TimestampCheck {
 }
 
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -24,7 +24,9 @@ impl TimestampCheck {
 #[async_trait::async_trait]
 impl Check for TimestampCheck {
     async fn check(&self, receipt: &ReceiptWithState<Checking>) -> CheckResult {
-        let timestamp_now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
+        let timestamp_now = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .map_err(|e| CheckError::Failed(e.into()))?;
         let min_timestamp = timestamp_now - self.timestamp_error_tolerance;
         let max_timestamp = timestamp_now + self.timestamp_error_tolerance;
 
@@ -33,10 +35,10 @@ impl Check for TimestampCheck {
         if receipt_timestamp < max_timestamp && receipt_timestamp > min_timestamp {
             Ok(())
         } else {
-            Err(anyhow!(
+            Err(CheckError::Failed(anyhow!(
                 "Receipt timestamp `{}` is outside of current system time +/- timestamp_error_tolerance",
                 receipt_timestamp.as_secs()
-            ))
+            )))
         }
     }
 }

common/src/tap/checks/timestamp_check.rs

@@ -810,7 +810,7 @@ pub mod tests {
     };
     use tap_aggregator::{jsonrpsee_helpers::JsonRpcResponse, server::run_server};
     use tap_core::receipt::{
-        checks::{Check, CheckList},
+        checks::{Check, CheckError, CheckList, CheckResult},
         state::Checking,
         ReceiptWithState,
     };
@@ -1386,8 +1386,8 @@ pub mod tests {
 
         #[async_trait::async_trait]
         impl Check for FailingCheck {
-            async fn check(&self, _receipt: &ReceiptWithState<Checking>) -> anyhow::Result<()> {
-                Err(anyhow::anyhow!("Failing check"))
+            async fn check(&self, _receipt: &ReceiptWithState<Checking>) -> CheckResult {
+                Err(CheckError::Failed(anyhow::anyhow!("Failing check")))
             }
         }
 
@@ -1406,7 +1406,13 @@ pub mod tests {
         // make sure to fail them
         let failing_receipts = checking_receipts
             .into_iter()
-            .map(|receipt| async { receipt.finalize_receipt_checks(&checks).await.unwrap_err() })
+            .map(|receipt| async {
+                receipt
+                    .finalize_receipt_checks(&checks)
+                    .await
+                    .unwrap()
+                    .unwrap_err()
+            })
             .collect::<Vec<_>>();
         let failing_receipts: Vec<_> = join_all(failing_receipts).await;
 

tap-agent/src/agent/sender_allocation.rs

@@ -8,7 +8,7 @@ use anyhow::anyhow;
 use eventuals::{Eventual, EventualExt};
 use indexer_common::subgraph_client::{Query, SubgraphClient};
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -52,17 +52,20 @@ impl Check for AllocationId {
         // ID. So the receipts that are received here should already have been filtered by
         // allocation ID.
         if allocation_id != self.allocation_id {
-            return Err(anyhow!("Receipt allocation_id different from expected: allocation_id: {}, expected_allocation_id: {}", allocation_id, self.allocation_id));
+            return Err(CheckError::Failed(anyhow!("Receipt allocation_id different from expected: allocation_id: {}, expected_allocation_id: {}", allocation_id, self.allocation_id)));
         };
 
         // 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(anyhow!("Allocation {} already redeemed", allocation_id)),
-            Err(e) => Err(anyhow!(
+            Ok(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
-            )),
+            ))),
         }
     }
 }

tap-agent/src/tap/context/checks/allocation_id.rs

@@ -6,7 +6,7 @@ use anyhow::anyhow;
 use eventuals::Eventual;
 use indexer_common::escrow_accounts::EscrowAccounts;
 use tap_core::receipt::{
-    checks::{Check, CheckResult},
+    checks::{Check, CheckError, CheckResult},
     state::Checking,
     ReceiptWithState,
 };
@@ -32,25 +32,31 @@ impl Check for Signature {
     async fn check(&self, receipt: &ReceiptWithState<Checking>) -> CheckResult {
         let signer = receipt
             .signed_receipt()
-            .recover_signer(&self.domain_separator)?;
-        let escrow_accounts =
-            self.escrow_accounts
-                .value()
-                .await
-                .map_err(|e| AdapterError::ValidationError {
-                    error: format!("Could not get escrow accounts from eventual: {:?}", e),
-                })?;
+            .recover_signer(&self.domain_separator)
+            .map_err(|e| CheckError::Failed(e.into()))?;
+        let escrow_accounts = self
+            .escrow_accounts
+            .value()
+            .await
+            .map_err(|e| AdapterError::ValidationError {
+                error: format!("Could not get escrow accounts from eventual: {:?}", e),
+            })
+            .map_err(|e| CheckError::Retryable(e.into()))?;
 
-        let sender = escrow_accounts.get_sender_for_signer(&signer)?;
+        let sender = escrow_accounts
+            .get_sender_for_signer(&signer)
+            .map_err(|e| CheckError::Failed(e.into()))?;
 
-        let balance = escrow_accounts.get_balance_for_sender(&sender)?;
+        let balance = escrow_accounts
+            .get_balance_for_sender(&sender)
+            .map_err(|e| CheckError::Failed(e.into()))?;
 
         if balance == U256::ZERO {
-            Err(anyhow!(
+            Err(CheckError::Failed(anyhow!(
                 "Balance for sender {}, signer {} is not positive",
                 sender,
                 signer
-            ))
+            )))
         } else {
             Ok(())
         }