Skip to content

fix(tap-agent): accept receipts newer than the last redemption #1086

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Maikol wants to merge 2 commits into
base: main-dips-rebased
Choose a base branch
from
Open

fix(tap-agent): accept receipts newer than the last redemption #1086

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a2937e2
fix(tap-agent): accept receipts newer than the last redemption
Maikol Aug 11, 2026
eadc7ec
fix(tap-agent): drop the redeemed-allocation check instead of retimin…
Maikol Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
161 changes: 136 additions & 25 deletions crates/tap-agent/src/tap/context/checks/allocation_id.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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<bool>,
last_redeemed_at_secs: Receiver<Option<u64>>,
allocation_id: Address,
collection_id: CollectionId,
}
Expand All @@ -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,
Expand All @@ -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,
}
Expand Down Expand Up @@ -79,14 +86,26 @@ impl Check<TapReceipt> 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 {

@MoonBoi9001 MoonBoi9001 Aug 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here we're comparing the receipt_timestamp_ns to last_redeemed_at_ns, which comes from the block timestamp of the redeem transaction. When tap-agent collects, it excludes any receipts newer than timestamp_buffer_secs. So I think there's a gap here where some receipts will be lost.

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(())
}
}

Expand All @@ -96,9 +115,9 @@ async fn tap_allocation_redeemed_watcher(
indexer_address: Address,
network_subgraph: &'static SubgraphClient,
escrow_polling_interval: Duration,
) -> anyhow::Result<Receiver<bool>> {
) -> anyhow::Result<Receiver<Option<u64>>> {
new_watcher(escrow_polling_interval, move || async move {
query_network_redeem_transactions(
query_latest_redeem_timestamp_secs(
collection_id,
sender_address,
indexer_address,
Expand All @@ -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<bool> {
) -> anyhow::Result<Option<u64>> {
// 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()];
Expand All @@ -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<u64> = None;

@MoonBoi9001 MoonBoi9001 Aug 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With no first, orderBy or orderDirection, graph-node will return 100 rows ordered by id.

Adding first: 1 would break sender_account.rs. So this check could use it's own query:

query LatestRedeemTransactionQuery($payer: Bytes!, $receiver: Bytes!, $allocationId: Bytes!) {
  paymentsEscrowTransactions(
    first: 1
    orderBy: timestamp
    orderDirection: desc
    where: { type: "redeem", payer_: { id: $payer }, receiver_: { id: $receiver }, allocationId: $allocationId }
  ) { timestamp }
}

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<u64>) -> 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";
Expand All @@ -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" }
]
}
}))),
Expand All @@ -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(),
Expand All @@ -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";
Expand Down Expand Up @@ -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(),
Expand All @@ -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";
Expand Down Expand Up @@ -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(),
Expand All @@ -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
);
}
}
Loading