From a2937e280050daa55f17a83cc8a089b8858f115c Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Tue, 11 Aug 2026 11:37:27 -0300 Subject: [PATCH 1/2] fix(tap-agent): accept receipts newer than the last redemption --- .../src/tap/context/checks/allocation_id.rs | 161 +++++++++++++++--- 1 file changed, 136 insertions(+), 25 deletions(-) diff --git a/crates/tap-agent/src/tap/context/checks/allocation_id.rs b/crates/tap-agent/src/tap/context/checks/allocation_id.rs index 6dfea06e1..b261a47c8 100644 --- a/crates/tap-agent/src/tap/context/checks/allocation_id.rs +++ b/crates/tap-agent/src/tap/context/checks/allocation_id.rs @@ -7,7 +7,10 @@ use anyhow::anyhow; use indexer_monitor::SubgraphClient; use indexer_query::payments_escrow_transactions_redeem; use indexer_watcher::new_watcher; -use tap_core::receipt::checks::{Check, CheckError, CheckResult}; +use tap_core::receipt::{ + checks::{Check, CheckError, CheckResult}, + WithValueAndTimestamp, +}; use thegraph_core::{ alloy::{hex::ToHexExt, primitives::Address}, CollectionId, @@ -16,11 +19,15 @@ use tokio::sync::watch::Receiver; use crate::tap::{CheckingReceipt, TapReceipt}; +const NANOS_PER_SECOND: u64 = 1_000_000_000; + /// AllocationId check /// -/// Verifies if the allocation is already redeemed. +/// Verifies that a receipt is newer than the allocation's most recent on-chain redemption. +/// Redemption no longer implies the allocation is closed — indexer-agent redeems RAVs on open +/// allocations on a timer — so only receipts at or older than the last redemption are replays. pub struct AllocationId { - tap_allocation_redeemed: Receiver, + last_redeemed_at_secs: Receiver>, allocation_id: Address, collection_id: CollectionId, } @@ -35,7 +42,7 @@ impl AllocationId { collection_id: CollectionId, network_subgraph: &'static SubgraphClient, ) -> Self { - let tap_allocation_redeemed = tap_allocation_redeemed_watcher( + let last_redeemed_at_secs = tap_allocation_redeemed_watcher( collection_id, sender_id, indexer_address, @@ -46,7 +53,7 @@ impl AllocationId { .expect("Failed to initialize tap_allocation_redeemed_watcher"); Self { - tap_allocation_redeemed, + last_redeemed_at_secs, allocation_id, collection_id, } @@ -79,14 +86,26 @@ impl Check for AllocationId { 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.borrow() { - false => Ok(()), - true => Err(CheckError::Failed(anyhow!( - "Allocation {:?} already redeemed", + let Some(last_redeemed_at_secs) = *self.last_redeemed_at_secs.borrow() else { + return Ok(()); + }; + let last_redeemed_at_ns = last_redeemed_at_secs + .checked_mul(NANOS_PER_SECOND) + .ok_or_else(|| { + CheckError::Failed(anyhow!( + "Last redeemed timestamp {last_redeemed_at_secs}s overflows when converted to nanoseconds" + )) + })?; + let receipt_timestamp_ns = receipt.signed_receipt().timestamp_ns(); + + if receipt_timestamp_ns <= last_redeemed_at_ns { + return Err(CheckError::Failed(anyhow!( + "Receipt timestamp {receipt_timestamp_ns}ns for allocation {:?} is not newer than the last redemption at {last_redeemed_at_secs}s ({last_redeemed_at_ns}ns)", self.collection_id.encode_hex() - ))), + ))); } + + Ok(()) } } @@ -96,9 +115,9 @@ async fn tap_allocation_redeemed_watcher( indexer_address: Address, network_subgraph: &'static SubgraphClient, escrow_polling_interval: Duration, -) -> anyhow::Result> { +) -> anyhow::Result>> { new_watcher(escrow_polling_interval, move || async move { - query_network_redeem_transactions( + query_latest_redeem_timestamp_secs( collection_id, sender_address, indexer_address, @@ -109,12 +128,13 @@ async fn tap_allocation_redeemed_watcher( .await } -async fn query_network_redeem_transactions( +/// Returns `None` if the allocation has never been redeemed. +async fn query_latest_redeem_timestamp_secs( collection_id: CollectionId, sender_address: Address, indexer_address: Address, network_subgraph: &'static SubgraphClient, -) -> anyhow::Result { +) -> anyhow::Result> { // Horizon network subgraph stores allocationId as the 20-byte address derived // from the 32-byte collection_id (rightmost 20 bytes). let allocation_ids = vec![collection_id.as_address().encode_hex()]; @@ -128,18 +148,44 @@ async fn query_network_redeem_transactions( ) .await?; - Ok(!data.payments_escrow_transactions.is_empty()) + let mut latest_redeemed_at_secs: Option = None; + for transaction in &data.payments_escrow_transactions { + let timestamp: u64 = transaction.timestamp.parse().map_err(|e| { + anyhow!( + "Invalid redeem transaction timestamp {:?}: {e}", + transaction.timestamp + ) + })?; + latest_redeemed_at_secs = latest_redeemed_at_secs.max(Some(timestamp)); + } + + Ok(latest_redeemed_at_secs) } #[cfg(test)] mod tests { use indexer_monitor::{DeploymentDetails, SubgraphClient}; use serde_json::json; + use tap_core::receipt::{checks::Check, Context}; + use test_assets::{ALLOCATION_ID_0, COLLECTION_ID_0, TAP_SIGNER as SIGNER}; use thegraph_core::{alloy::hex::ToHexExt, CollectionId}; + use tokio::sync::watch; use wiremock::{matchers::body_string_contains, Mock, MockServer, ResponseTemplate}; + use crate::test::create_received_receipt_v2; + + /// Builds the check directly, bypassing `new`'s network watcher, with a fixed redemption time. + fn allocation_id_check(last_redeemed_at_secs: Option) -> super::AllocationId { + let (_tx, rx) = watch::channel(last_redeemed_at_secs); + super::AllocationId { + last_redeemed_at_secs: rx, + allocation_id: ALLOCATION_ID_0, + collection_id: COLLECTION_ID_0, + } + } + #[tokio::test] - async fn test_network_redeem_transactions_true_when_present() { + async fn test_latest_redeem_timestamp_returns_latest_timestamp() { let mock_server: MockServer = MockServer::start().await; let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; @@ -156,7 +202,9 @@ mod tests { .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "data": { "paymentsEscrowTransactions": [ - { "id": "0x01", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "1" } + { "id": "0x01", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "5" }, + { "id": "0x02", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "9" }, + { "id": "0x03", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "3" } ] } }))), @@ -172,7 +220,7 @@ mod tests { .await, )); - let result = super::query_network_redeem_transactions( + let result = super::query_latest_redeem_timestamp_secs( collection_id, sender_address.parse().unwrap(), indexer_address.parse().unwrap(), @@ -181,11 +229,14 @@ mod tests { .await .unwrap(); - assert!(result); + // The largest timestamp is deliberately not the last row: an allocation redeemed + // more than once must yield the most recent redemption, whatever order the + // subgraph returns the transactions in. + assert_eq!(result, Some(9)); } #[tokio::test] - async fn test_network_redeem_transactions_false_when_empty() { + async fn test_latest_redeem_timestamp_returns_none_when_empty() { let mock_server: MockServer = MockServer::start().await; let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; @@ -213,7 +264,7 @@ mod tests { .await, )); - let result = super::query_network_redeem_transactions( + let result = super::query_latest_redeem_timestamp_secs( collection_id, sender_address.parse().unwrap(), indexer_address.parse().unwrap(), @@ -222,11 +273,11 @@ mod tests { .await .unwrap(); - assert!(!result); + assert_eq!(result, None); } #[tokio::test] - async fn test_network_redeem_transactions_error_when_subgraph_fails() { + async fn test_latest_redeem_timestamp_error_when_subgraph_fails() { let mock_server: MockServer = MockServer::start().await; let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; @@ -254,7 +305,7 @@ mod tests { .await, )); - let result = super::query_network_redeem_transactions( + let result = super::query_latest_redeem_timestamp_secs( collection_id, sender_address.parse().unwrap(), indexer_address.parse().unwrap(), @@ -264,4 +315,64 @@ mod tests { assert!(result.is_err()); } + + #[tokio::test] + async fn test_receipt_newer_than_redemption_is_accepted() { + let redeemed_at_secs = 1_700_000_000u64; + let check = allocation_id_check(Some(redeemed_at_secs)); + + let receipt_timestamp_ns = (redeemed_at_secs + 10) * 1_000_000_000; + let receipt = + create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, receipt_timestamp_ns, 1); + + let result = check.check(&Context::new(), &receipt).await; + + assert!( + result.is_ok(), + "expected a receipt newer than the last redemption to be accepted, got {:?}", + result + ); + } + + #[tokio::test] + async fn test_receipt_at_or_older_than_redemption_is_rejected() { + let redeemed_at_secs = 1_700_000_000u64; + let check = allocation_id_check(Some(redeemed_at_secs)); + + let receipt_timestamp_ns = redeemed_at_secs * 1_000_000_000; + let receipt = + create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, receipt_timestamp_ns, 1); + + let result = check.check(&Context::new(), &receipt).await; + + assert!( + result.is_err(), + "expected a receipt at the last redemption to still be rejected (anti-replay)" + ); + + let receipt_timestamp_ns = (redeemed_at_secs - 10) * 1_000_000_000; + let receipt = + create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 2, receipt_timestamp_ns, 1); + + let result = check.check(&Context::new(), &receipt).await; + + assert!( + result.is_err(), + "expected a receipt older than the last redemption to be rejected (anti-replay)" + ); + } + + #[tokio::test] + async fn test_no_redemption_accepts_receipt() { + let check = allocation_id_check(None); + let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, 1, 1); + + let result = check.check(&Context::new(), &receipt).await; + + assert!( + result.is_ok(), + "expected a receipt to be accepted when the allocation was never redeemed, got {:?}", + result + ); + } } From eadc7ec1a0feccd82115383b73351959c29f79b7 Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Thu, 20 Aug 2026 12:01:51 -0300 Subject: [PATCH 2/2] fix(tap-agent): drop the redeemed-allocation check instead of retiming it --- crates/tap-agent/src/agent/sender_account.rs | 1 - .../tap-agent/src/agent/sender_allocation.rs | 234 +++---------- .../src/tap/context/checks/allocation_id.rs | 323 ++---------------- 3 files changed, 76 insertions(+), 482 deletions(-) diff --git a/crates/tap-agent/src/agent/sender_account.rs b/crates/tap-agent/src/agent/sender_account.rs index 8b3857515..8ca74521f 100644 --- a/crates/tap-agent/src/agent/sender_account.rs +++ b/crates/tap-agent/src/agent/sender_account.rs @@ -472,7 +472,6 @@ impl State { .sender(self.sender) .escrow_accounts(self.escrow_accounts.clone()) .escrow_accounts_strict(self.escrow_accounts_strict.clone()) - .network_subgraph(self.network_subgraph) .domain_separator(self.domain_separator_v2.clone()) .sender_account_ref(sender_account_ref.clone()) .sender_aggregator(self.aggregator_v2.clone()) diff --git a/crates/tap-agent/src/agent/sender_allocation.rs b/crates/tap-agent/src/agent/sender_allocation.rs index f27dc36dd..c3a604d1d 100644 --- a/crates/tap-agent/src/agent/sender_allocation.rs +++ b/crates/tap-agent/src/agent/sender_allocation.rs @@ -10,7 +10,7 @@ use std::{ use anyhow::{anyhow, ensure}; use bigdecimal::{num_bigint::BigInt, ToPrimitive}; -use indexer_monitor::{EscrowAccounts, SubgraphClient}; +use indexer_monitor::EscrowAccounts; use prometheus::{register_counter_vec, register_histogram_vec, CounterVec, HistogramVec}; use ractor::{Actor, ActorProcessingErr, ActorRef}; use sqlx::{types::BigDecimal, PgPool, Row}; @@ -179,8 +179,6 @@ pub struct AllocationConfig { pub rav_request_receipt_limit: u64, /// Current indexer address pub indexer_address: Address, - /// Polling interval for escrow subgraph - pub escrow_polling_interval: Duration, /// SubgraphService contract address pub subgraph_service_address: Address, } @@ -192,7 +190,6 @@ impl AllocationConfig { timestamp_buffer_ns: config.rav_request_buffer.as_nanos() as u64, rav_request_receipt_limit: config.rav_request_receipt_limit, indexer_address: config.indexer_address, - escrow_polling_interval: config.escrow_polling_interval, subgraph_service_address: config.subgraph_service_address, } } @@ -212,8 +209,6 @@ pub struct SenderAllocationArgs { /// Watcher containing escrow accounts with strict signer filtering /// (excludes thawing signers, used for RAV signature verification) pub escrow_accounts_strict: Receiver, - /// SubgraphClient of the network subgraph - pub network_subgraph: &'static SubgraphClient, /// Domain separator used for tap pub domain_separator: Eip712Domain, /// Reference to [super::sender_account::SenderAccount] actor @@ -461,7 +456,6 @@ where sender, escrow_accounts, escrow_accounts_strict, - network_subgraph, domain_separator, sender_account_ref, sender_aggregator, @@ -470,17 +464,10 @@ where ) -> anyhow::Result { let collection_id = T::to_allocation_id_enum(&allocation_id).0; let required_checks: Vec + Send + Sync>> = vec![ - Arc::new( - AllocationId::new( - config.indexer_address, - config.escrow_polling_interval, - sender, - T::allocation_id_to_address(&allocation_id), - collection_id, - network_subgraph, - ) - .await, - ), + Arc::new(AllocationId::new( + T::allocation_id_to_address(&allocation_id), + collection_id, + )), Arc::new(Signature::new( domain_separator.clone(), escrow_accounts.clone(), @@ -1211,14 +1198,12 @@ mod tests { use std::collections::HashMap; use futures::future::join_all; - use indexer_monitor::{DeploymentDetails, EscrowAccounts, SubgraphClient}; + use indexer_monitor::EscrowAccounts; use ractor::{call, ActorStatus}; - use serde_json::json; use tap_aggregator::grpc::v2::tap_aggregator_client::TapAggregatorClient; use test_assets::{flush_messages, TAP_SENDER as SENDER, TAP_SIGNER as SIGNER}; use thegraph_core::{alloy::primitives::U256, CollectionId}; use tokio::sync::watch; - use wiremock::{matchers::body_string_contains, Mock, MockServer, ResponseTemplate}; use super::*; use crate::tap::CheckingReceipt; @@ -1228,39 +1213,24 @@ mod tests { ALLOCATION_ID_0, ESCROW_VALUE, TAP_EIP712_DOMAIN_SEPARATOR_V2, }; - async fn setup_network_subgraph(redeemed: bool) -> MockServer { - let mock_server = MockServer::start().await; - let response = if redeemed { - json!({ - "data": { - "paymentsEscrowTransactions": [ - { "id": "0x01", "allocationId": ALLOCATION_ID_0.encode_hex(), "timestamp": "1" } - ] - } - }) - } else { - json!({ "data": { "paymentsEscrowTransactions": [] } }) - }; - - mock_server - .register( - Mock::given(body_string_contains("paymentsEscrowTransactions")) - .respond_with(ResponseTemplate::new(200).set_body_json(response)), - ) - .await; - - mock_server + async fn build_sender_allocation_args( + pgpool: PgPool, + sender_account_ref: ActorRef, + ) -> SenderAllocationArgs { + build_sender_allocation_args_with_balance(pgpool, sender_account_ref, ESCROW_VALUE).await } - async fn build_sender_allocation_args( + /// SIGNER stays an authorized signer whatever the balance, so receipts are still retrieved + /// for a RAV request -- a zero balance fails them in the signature check instead. + async fn build_sender_allocation_args_with_balance( pgpool: PgPool, - network_subgraph: &'static SubgraphClient, sender_account_ref: ActorRef, + escrow_balance: u128, ) -> SenderAllocationArgs { let (escrow_accounts_tx, escrow_accounts_rx) = watch::channel(EscrowAccounts::default()); escrow_accounts_tx .send(EscrowAccounts::new( - HashMap::from([(SENDER.1, U256::from(ESCROW_VALUE))]), + HashMap::from([(SENDER.1, U256::from(escrow_balance))]), HashMap::from([(SENDER.1, vec![SIGNER.1])]), )) .unwrap(); @@ -1276,7 +1246,6 @@ mod tests { .sender(SENDER.1) .escrow_accounts(escrow_accounts_rx.clone()) .escrow_accounts_strict(escrow_accounts_rx) - .network_subgraph(network_subgraph) .domain_separator(TAP_EIP712_DOMAIN_SEPARATOR_V2.clone()) .sender_account_ref(sender_account_ref) .sender_aggregator(sender_aggregator) @@ -1286,24 +1255,32 @@ mod tests { .build() } - async fn build_state( - pgpool: PgPool, - network_subgraph: &'static SubgraphClient, - ) -> SenderAllocationState { + async fn build_state(pgpool: PgPool) -> SenderAllocationState { let (_receiver, sender_account_ref) = create_mock_sender_account().await; - let args = build_sender_allocation_args(pgpool, network_subgraph, sender_account_ref).await; + let args = build_sender_allocation_args(pgpool, sender_account_ref).await; SenderAllocationState::new(args).await.unwrap() } async fn spawn_sender_allocation( pgpool: PgPool, - network_subgraph: &'static SubgraphClient, + ) -> ( + ActorRef, + tokio::sync::mpsc::Receiver, + ) { + spawn_sender_allocation_with_balance(pgpool, ESCROW_VALUE).await + } + + async fn spawn_sender_allocation_with_balance( + pgpool: PgPool, + escrow_balance: u128, ) -> ( ActorRef, tokio::sync::mpsc::Receiver, ) { let (mut receiver, sender_account_ref) = create_mock_sender_account().await; - let args = build_sender_allocation_args(pgpool, network_subgraph, sender_account_ref).await; + let args = + build_sender_allocation_args_with_balance(pgpool, sender_account_ref, escrow_balance) + .await; let (sender_allocation, _) = SenderAllocation::::spawn(None, SenderAllocation::default(), args) .await @@ -1347,18 +1324,8 @@ mod tests { #[tokio::test] async fn test_several_receipts_rav_request() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let (sender_allocation, mut receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + let (sender_allocation, mut receiver) = spawn_sender_allocation(test_db.pool.clone()).await; const AMOUNT_OF_RECEIPTS: u64 = 1000; for i in 0..AMOUNT_OF_RECEIPTS { @@ -1384,18 +1351,8 @@ mod tests { #[tokio::test] async fn test_several_receipts_batch_insert_rav_request() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let (sender_allocation, mut receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + let (sender_allocation, mut receiver) = spawn_sender_allocation(test_db.pool.clone()).await; const AMOUNT_OF_RECEIPTS: u64 = 1000; for i in 0..AMOUNT_OF_RECEIPTS { @@ -1421,18 +1378,8 @@ mod tests { #[tokio::test] async fn test_close_allocation_no_pending_fees() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let (sender_allocation, _receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + let (sender_allocation, _receiver) = spawn_sender_allocation(test_db.pool.clone()).await; sender_allocation.stop_and_wait(None, None).await.unwrap(); assert_eq!(sender_allocation.get_status(), ActorStatus::Stopped); @@ -1448,15 +1395,6 @@ mod tests { #[tokio::test] async fn test_close_allocation_with_pending_fees() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); for i in 0..10 { let receipt = @@ -1465,8 +1403,7 @@ mod tests { store_receipt(&test_db.pool, &signed).await.unwrap(); } - let (sender_allocation, _receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + let (sender_allocation, _receiver) = spawn_sender_allocation(test_db.pool.clone()).await; sender_allocation.stop_and_wait(None, None).await.unwrap(); assert_eq!(sender_allocation.get_status(), ActorStatus::Stopped); @@ -1482,16 +1419,7 @@ mod tests { #[tokio::test] async fn should_return_unaggregated_fees_without_rav() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let state = build_state(test_db.pool.clone(), network_subgraph).await; + let state = build_state(test_db.pool.clone()).await; for i in 1..10 { let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, i, i, i.into()); @@ -1506,16 +1434,7 @@ mod tests { #[tokio::test] async fn should_calculate_invalid_receipts_fee() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let mut state = build_state(test_db.pool.clone(), network_subgraph).await; + let mut state = build_state(test_db.pool.clone()).await; let failing_receipts = make_failing_receipts().await; @@ -1533,15 +1452,6 @@ mod tests { #[tokio::test] async fn should_return_unaggregated_fees_with_rav() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); let signed_rav = create_rav_v2( *CollectionId::from(ALLOCATION_ID_0), @@ -1553,7 +1463,7 @@ mod tests { .await .unwrap(); - let state = build_state(test_db.pool.clone(), network_subgraph).await; + let state = build_state(test_db.pool.clone()).await; for i in 1..10 { let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, i, i, i.into()); @@ -1568,16 +1478,7 @@ mod tests { #[tokio::test] async fn test_store_failed_rav() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let state = build_state(test_db.pool.clone(), network_subgraph).await; + let state = build_state(test_db.pool.clone()).await; let signed_rav = create_rav_v2( *CollectionId::from(ALLOCATION_ID_0), @@ -1594,16 +1495,7 @@ mod tests { #[tokio::test] async fn test_store_invalid_receipts() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let mut state = build_state(test_db.pool.clone(), network_subgraph).await; + let mut state = build_state(test_db.pool.clone()).await; let failing_receipts = make_failing_receipts().await; @@ -1619,16 +1511,7 @@ mod tests { #[tokio::test] async fn test_store_invalid_receipts_rolls_back_with_transaction() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let mut state = build_state(test_db.pool.clone(), network_subgraph).await; + let mut state = build_state(test_db.pool.clone()).await; let failing_receipts = make_failing_receipts().await; @@ -1655,16 +1538,7 @@ mod tests { #[tokio::test] async fn test_mark_rav_last() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); - let state = build_state(test_db.pool.clone(), network_subgraph).await; + let state = build_state(test_db.pool.clone()).await; let signed_rav = create_rav_v2( *CollectionId::from(ALLOCATION_ID_0), @@ -1683,15 +1557,6 @@ mod tests { #[tokio::test] async fn test_failed_rav_request() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(false).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); // Use receipts signed by a wallet not in escrow_accounts to force invalid receipts. for i in 0..10 { @@ -1706,8 +1571,7 @@ mod tests { store_receipt(&test_db.pool, &signed).await.unwrap(); } - let (sender_allocation, mut receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + let (sender_allocation, mut receiver) = spawn_sender_allocation(test_db.pool.clone()).await; sender_allocation .cast(SenderAllocationMessage::TriggerRavRequest) @@ -1720,15 +1584,6 @@ mod tests { #[tokio::test] async fn test_rav_request_when_all_receipts_invalid() { let test_db = test_assets::setup_shared_test_db().await; - let network_mock = setup_network_subgraph(true).await; - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&network_mock.uri()).unwrap(), - ) - .await, - )); let timestamp = 1u64; const RECEIPT_VALUE: u128 = 10; @@ -1745,8 +1600,9 @@ mod tests { store_receipt(&test_db.pool, &signed).await.unwrap(); } + // Zero escrow balance fails every receipt in the signature check. let (sender_allocation, mut receiver) = - spawn_sender_allocation(test_db.pool.clone(), network_subgraph).await; + spawn_sender_allocation_with_balance(test_db.pool.clone(), 0).await; sender_allocation .cast(SenderAllocationMessage::TriggerRavRequest) diff --git a/crates/tap-agent/src/tap/context/checks/allocation_id.rs b/crates/tap-agent/src/tap/context/checks/allocation_id.rs index b261a47c8..6d03ceb27 100644 --- a/crates/tap-agent/src/tap/context/checks/allocation_id.rs +++ b/crates/tap-agent/src/tap/context/checks/allocation_id.rs @@ -1,59 +1,29 @@ // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs. // SPDX-License-Identifier: Apache-2.0 -use std::time::Duration; - use anyhow::anyhow; -use indexer_monitor::SubgraphClient; -use indexer_query::payments_escrow_transactions_redeem; -use indexer_watcher::new_watcher; -use tap_core::receipt::{ - checks::{Check, CheckError, CheckResult}, - WithValueAndTimestamp, -}; -use thegraph_core::{ - alloy::{hex::ToHexExt, primitives::Address}, - CollectionId, -}; -use tokio::sync::watch::Receiver; +use tap_core::receipt::checks::{Check, CheckError, CheckResult}; +use thegraph_core::{alloy::primitives::Address, CollectionId}; use crate::tap::{CheckingReceipt, TapReceipt}; -const NANOS_PER_SECOND: u64 = 1_000_000_000; - /// AllocationId check /// -/// Verifies that a receipt is newer than the allocation's most recent on-chain redemption. -/// Redemption no longer implies the allocation is closed — indexer-agent redeems RAVs on open -/// allocations on a timer — so only receipts at or older than the last redemption are replays. +/// Verifies that a receipt is addressed to the allocation this actor serves. +/// +/// Replay protection is deliberately not done here. `tap_core` only collects receipts newer than +/// the last RAV, the sender's aggregator refuses to sign a RAV containing a receipt at or below +/// the previous RAV's timestamp, and `GraphTallyCollector` pays out only the delta over what it +/// has already collected for the collection. pub struct AllocationId { - last_redeemed_at_secs: Receiver>, allocation_id: Address, collection_id: CollectionId, } impl AllocationId { /// Creates a new allocation id check - pub async fn new( - indexer_address: Address, - escrow_polling_interval: Duration, - sender_id: Address, - allocation_id: Address, - collection_id: CollectionId, - network_subgraph: &'static SubgraphClient, - ) -> Self { - let last_redeemed_at_secs = tap_allocation_redeemed_watcher( - collection_id, - sender_id, - indexer_address, - network_subgraph, - escrow_polling_interval, - ) - .await - .expect("Failed to initialize tap_allocation_redeemed_watcher"); - + pub fn new(allocation_id: Address, collection_id: CollectionId) -> Self { Self { - last_redeemed_at_secs, allocation_id, collection_id, } @@ -86,293 +56,62 @@ impl Check for AllocationId { return Err(CheckError::Failed(anyhow!("Receipt allocation_id different from expected: allocation_id: {:?}, expected_allocation_id: {}", allocation_id, self.allocation_id))); }; - let Some(last_redeemed_at_secs) = *self.last_redeemed_at_secs.borrow() else { - return Ok(()); - }; - let last_redeemed_at_ns = last_redeemed_at_secs - .checked_mul(NANOS_PER_SECOND) - .ok_or_else(|| { - CheckError::Failed(anyhow!( - "Last redeemed timestamp {last_redeemed_at_secs}s overflows when converted to nanoseconds" - )) - })?; - let receipt_timestamp_ns = receipt.signed_receipt().timestamp_ns(); - - if receipt_timestamp_ns <= last_redeemed_at_ns { - return Err(CheckError::Failed(anyhow!( - "Receipt timestamp {receipt_timestamp_ns}ns for allocation {:?} is not newer than the last redemption at {last_redeemed_at_secs}s ({last_redeemed_at_ns}ns)", - self.collection_id.encode_hex() - ))); - } - Ok(()) } } -async fn tap_allocation_redeemed_watcher( - collection_id: CollectionId, - sender_address: Address, - indexer_address: Address, - network_subgraph: &'static SubgraphClient, - escrow_polling_interval: Duration, -) -> anyhow::Result>> { - new_watcher(escrow_polling_interval, move || async move { - query_latest_redeem_timestamp_secs( - collection_id, - sender_address, - indexer_address, - network_subgraph, - ) - .await - }) - .await -} - -/// Returns `None` if the allocation has never been redeemed. -async fn query_latest_redeem_timestamp_secs( - collection_id: CollectionId, - sender_address: Address, - indexer_address: Address, - network_subgraph: &'static SubgraphClient, -) -> anyhow::Result> { - // Horizon network subgraph stores allocationId as the 20-byte address derived - // from the 32-byte collection_id (rightmost 20 bytes). - let allocation_ids = vec![collection_id.as_address().encode_hex()]; - let data = network_subgraph - .query::( - payments_escrow_transactions_redeem::Variables { - payer: sender_address.encode_hex(), - receiver: indexer_address.encode_hex(), - allocation_ids: Some(allocation_ids), - }, - ) - .await?; - - let mut latest_redeemed_at_secs: Option = None; - for transaction in &data.payments_escrow_transactions { - let timestamp: u64 = transaction.timestamp.parse().map_err(|e| { - anyhow!( - "Invalid redeem transaction timestamp {:?}: {e}", - transaction.timestamp - ) - })?; - latest_redeemed_at_secs = latest_redeemed_at_secs.max(Some(timestamp)); - } - - Ok(latest_redeemed_at_secs) -} - #[cfg(test)] mod tests { - use indexer_monitor::{DeploymentDetails, SubgraphClient}; - use serde_json::json; use tap_core::receipt::{checks::Check, Context}; - use test_assets::{ALLOCATION_ID_0, COLLECTION_ID_0, TAP_SIGNER as SIGNER}; - use thegraph_core::{alloy::hex::ToHexExt, CollectionId}; - use tokio::sync::watch; - use wiremock::{matchers::body_string_contains, Mock, MockServer, ResponseTemplate}; + use test_assets::{ALLOCATION_ID_0, ALLOCATION_ID_1, COLLECTION_ID_0, TAP_SIGNER as SIGNER}; + use thegraph_core::CollectionId; use crate::test::create_received_receipt_v2; - /// Builds the check directly, bypassing `new`'s network watcher, with a fixed redemption time. - fn allocation_id_check(last_redeemed_at_secs: Option) -> super::AllocationId { - let (_tx, rx) = watch::channel(last_redeemed_at_secs); - super::AllocationId { - last_redeemed_at_secs: rx, - allocation_id: ALLOCATION_ID_0, - collection_id: COLLECTION_ID_0, - } - } - - #[tokio::test] - async fn test_latest_redeem_timestamp_returns_latest_timestamp() { - let mock_server: MockServer = MockServer::start().await; - let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; - let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; - let collection_id = CollectionId::from( - sender_address - .parse::() - .unwrap(), - ); - - mock_server - .register( - Mock::given(body_string_contains("paymentsEscrowTransactions")) - .and(body_string_contains(collection_id.as_address().encode_hex())) - .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "data": { - "paymentsEscrowTransactions": [ - { "id": "0x01", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "5" }, - { "id": "0x02", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "9" }, - { "id": "0x03", "allocationId": collection_id.as_address().encode_hex(), "timestamp": "3" } - ] - } - }))), - ) - .await; - - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&mock_server.uri()).unwrap(), - ) - .await, - )); - - let result = super::query_latest_redeem_timestamp_secs( - collection_id, - sender_address.parse().unwrap(), - indexer_address.parse().unwrap(), - network_subgraph, - ) - .await - .unwrap(); - - // The largest timestamp is deliberately not the last row: an allocation redeemed - // more than once must yield the most recent redemption, whatever order the - // subgraph returns the transactions in. - assert_eq!(result, Some(9)); - } - - #[tokio::test] - async fn test_latest_redeem_timestamp_returns_none_when_empty() { - let mock_server: MockServer = MockServer::start().await; - let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; - let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; - let collection_id = CollectionId::from( - sender_address - .parse::() - .unwrap(), - ); - - mock_server - .register( - Mock::given(body_string_contains("paymentsEscrowTransactions")).respond_with( - ResponseTemplate::new(200) - .set_body_json(json!({ "data": { "paymentsEscrowTransactions": [] } })), - ), - ) - .await; - - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&mock_server.uri()).unwrap(), - ) - .await, - )); - - let result = super::query_latest_redeem_timestamp_secs( - collection_id, - sender_address.parse().unwrap(), - indexer_address.parse().unwrap(), - network_subgraph, - ) - .await - .unwrap(); - - assert_eq!(result, None); - } - - #[tokio::test] - async fn test_latest_redeem_timestamp_error_when_subgraph_fails() { - let mock_server: MockServer = MockServer::start().await; - let sender_address = "0x21fed3c4340f67dbf2b78c670ebd1940668ca03e"; - let indexer_address = "0x54d7db28ce0d0e2e87764cd09298f9e4e913e567"; - let collection_id = CollectionId::from( - sender_address - .parse::() - .unwrap(), - ); - - mock_server - .register( - Mock::given(body_string_contains("paymentsEscrowTransactions")).respond_with( - ResponseTemplate::new(200) - .set_body_json(json!({ "errors": [{ "message": "boom" }] })), - ), - ) - .await; - - let network_subgraph = Box::leak(Box::new( - SubgraphClient::new( - reqwest::Client::new(), - None, - DeploymentDetails::for_query_url(&mock_server.uri()).unwrap(), - ) - .await, - )); - - let result = super::query_latest_redeem_timestamp_secs( - collection_id, - sender_address.parse().unwrap(), - indexer_address.parse().unwrap(), - network_subgraph, - ) - .await; - - assert!(result.is_err()); - } - #[tokio::test] - async fn test_receipt_newer_than_redemption_is_accepted() { - let redeemed_at_secs = 1_700_000_000u64; - let check = allocation_id_check(Some(redeemed_at_secs)); - - let receipt_timestamp_ns = (redeemed_at_secs + 10) * 1_000_000_000; - let receipt = - create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, receipt_timestamp_ns, 1); + async fn test_receipt_for_this_allocation_is_accepted() { + let check = super::AllocationId::new(ALLOCATION_ID_0, COLLECTION_ID_0); + let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, 1, 1); let result = check.check(&Context::new(), &receipt).await; assert!( result.is_ok(), - "expected a receipt newer than the last redemption to be accepted, got {:?}", - result + "expected a receipt for this allocation to be accepted, got {result:?}" ); } #[tokio::test] - async fn test_receipt_at_or_older_than_redemption_is_rejected() { - let redeemed_at_secs = 1_700_000_000u64; - let check = allocation_id_check(Some(redeemed_at_secs)); - - let receipt_timestamp_ns = redeemed_at_secs * 1_000_000_000; - let receipt = - create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, receipt_timestamp_ns, 1); - - let result = check.check(&Context::new(), &receipt).await; - - assert!( - result.is_err(), - "expected a receipt at the last redemption to still be rejected (anti-replay)" - ); - - let receipt_timestamp_ns = (redeemed_at_secs - 10) * 1_000_000_000; - let receipt = - create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 2, receipt_timestamp_ns, 1); + async fn test_receipt_for_another_allocation_is_rejected() { + let check = super::AllocationId::new(ALLOCATION_ID_0, COLLECTION_ID_0); + let receipt = create_received_receipt_v2(&ALLOCATION_ID_1, &SIGNER.0, 1, 1, 1); let result = check.check(&Context::new(), &receipt).await; assert!( result.is_err(), - "expected a receipt older than the last redemption to be rejected (anti-replay)" + "expected a receipt carrying another allocation's collection_id to be rejected" ); } #[tokio::test] - async fn test_no_redemption_accepts_receipt() { - let check = allocation_id_check(None); - let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, 1, 1); + async fn test_receipt_is_accepted_regardless_of_timestamp() { + // Anti-replay is enforced by tap_core's min_timestamp, the sender's aggregator and the + // collector contract -- never by this check. An old receipt is not this check's problem. + let check = super::AllocationId::new(ALLOCATION_ID_0, COLLECTION_ID_0); + let receipt = create_received_receipt_v2(&ALLOCATION_ID_0, &SIGNER.0, 1, 0, 1); let result = check.check(&Context::new(), &receipt).await; assert!( result.is_ok(), - "expected a receipt to be accepted when the allocation was never redeemed, got {:?}", - result + "expected timestamp to be irrelevant to the allocation id check, got {result:?}" ); } + + #[test] + fn test_collection_id_0_matches_allocation_id_0() { + // The two tests above only mean anything if these agree. + assert_eq!(COLLECTION_ID_0, CollectionId::from(ALLOCATION_ID_0)); + } }