diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index ccee80c5..d9e0bad9 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -794,9 +794,9 @@ pub struct DipsAgreementConfig { #[serde(default = "default_max_grt_per_billion_entities_per_30_days")] pub max_grt_per_billion_entities_per_30_days: f64, - /// Number of days to look back for declined indexers (standard exclusion). - /// Covers CanceledByIndexer/Expired agreements and structurally persistent - /// rejections (UNSUPPORTED_NETWORK, MANIFEST_TOO_LARGE). Default: 30 days. + /// Number of days to look back for declined indexers (standard exclusion). Covers + /// CanceledByIndexer, expiries whose offer reached the chain, and structurally + /// persistent rejections (UNSUPPORTED_NETWORK, MANIFEST_TOO_LARGE). Default: 30 days. #[serde(default = "default_declined_indexer_lookback_days")] pub declined_indexer_lookback_days: i32, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 8248203f..fac5d5fe 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -173,8 +173,8 @@ pub trait AgreementRegistry { ) -> RegistryResult>>; /// Get declined `CanceledByIndexer`/`Expired`/`Rejected` indexers grouped by - /// deployment. Each rejection reason gets its own exclusion window (price, - /// transient, uncertain, or default); see the `rejection_reason` constants. + /// deployment. Each rejection reason gets its own exclusion window, as does an + /// expiry that never had an offer transaction; see the query for the details. async fn get_declined_indexers_by_deployment( &self, default_lookback_days: i32, diff --git a/bin/dipper-service/src/worker/handlers/submit_offer.rs b/bin/dipper-service/src/worker/handlers/submit_offer.rs index 022dd07c..f1966174 100644 --- a/bin/dipper-service/src/worker/handlers/submit_offer.rs +++ b/bin/dipper-service/src/worker/handlers/submit_offer.rs @@ -12,14 +12,9 @@ //! via `RecurringCollector.offer()`. The indexer-agent then calls //! `acceptIndexingAgreement` — the contract checks `rcaOffers`. //! -//! Idempotency is gated on the indexing-payments-subgraph's `Offer` entity, -//! not an RPC call. The `rcaOffers` mapping on `RecurringCollector` lives -//! inside an ERC-7201 namespaced storage struct with no public getter, so -//! dipper reuses the same subgraph indexer-rs queries to check whether a -//! prior submission already landed. The subgraph handler is idempotent on -//! duplicate `OfferStored` events, so a crashed restart that races the -//! subgraph's indexing lag and re-submits will end up as a no-op at the -//! entity level even if it costs a second on-chain transaction. +//! Nothing here is idempotent across a crash: a re-run submits the offer again. The +//! `rcaOffers` mapping on `RecurringCollector` sits in an ERC-7201 namespaced storage +//! struct with no public getter, so the chain cannot cheaply be asked what already landed. use std::time::Duration; @@ -118,9 +113,11 @@ where // through it rather than posting directly. match ctx.chain_client.offer_via_manager(&rca).await { Ok(None) => { + // The chain client reports nothing was submitted. The live client always + // submits, so this only fires for a client that skips the offer itself. tracing::info!( agreement_id = %agreement_id, - "Offer already stored on-chain with matching hash, proceeding to dispatch" + "Offer needed no transaction, proceeding to dispatch" ); } Ok(Some(tx_hash)) => { diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 1463989e..8f2254fc 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -689,7 +689,7 @@ impl PgRegistry { /// Get declined `CanceledByIndexer`/`Expired`/`Rejected` indexers grouped by /// deployment (deployment id -> indexer ids). Each rejection reason gets its own - /// exclusion window (price, transient, uncertain, default); see the constants. + /// exclusion window, as does an expiry that never had an offer transaction. pub async fn get_declined_indexers_by_deployment( &self, default_lookback_days: i32, @@ -726,8 +726,21 @@ impl PgRegistry { (rejection_reason IN ($18, $19) AND updated_at >= timezone('UTC', now()) - make_interval(days => $17)) OR + -- An expiry with no offer transaction and no rejection reason means we never + -- landed the offer, so the indexer never had one to accept: our fault, so it + -- gets the short dipper-side lookback. An expiry that does carry a reason keeps + -- the window that reason earns. The catch-all below excludes exactly this set. + -- Read the empty hash as strong evidence, not proof. It is also empty when the + -- write recording it failed, and when the transaction confirmed after this row + -- had already expired, which both let off an indexer that could have accepted: + -- the harmless direction. The column's own migration calls it observability + -- only, so note that this query is what makes it load-bearing. + (status = $2 AND offer_tx_hash IS NULL AND rejection_reason IS NULL + AND updated_at >= timezone('UTC', now()) - make_interval(mins => $7)) + OR -- All other rejections/expirations/cancellations: standard lookback (COALESCE(rejection_reason, '') NOT IN ($6, $8, $9, $10, $11, $12, $13, $14, $15, $16, $18, $19) + AND NOT (status = $2 AND offer_tx_hash IS NULL AND rejection_reason IS NULL) AND updated_at >= timezone('UTC', now()) - make_interval(days => $5)) ) GROUP BY deployment_id diff --git a/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql b/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql index 4c68bfcc..f090f8b3 100644 --- a/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql +++ b/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql @@ -61,12 +61,16 @@ VALUES ('\xbb000000000000000000000000000001'::bytea, '01930100-0002-7000-8000-00 '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "service_provider": "0x2222222222222222222222222222222222222222", "data_service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "deadline": 1700000300, "ends_at": 1700086400, "max_initial_tokens": "1000", "max_ongoing_tokens_per_second": "100", "min_seconds_per_collection": 86400, "max_seconds_per_collection": 864000, "metadata": {"tokens_per_second": "10", "tokens_per_entity_per_second": "1", "subgraph_deployment_id": "QmDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD4d", "protocol_network": 1, "chain_id": 1}}'::json); -- Indexer C: 0x3333333333333333333333333333333333333333 --- Has NO active agreements (only expired) -INSERT INTO dipper_reg_indexing_agreements (id, nonce_uuid, created_at, updated_at, status, indexing_request_id, deployment_id, indexer_id, indexer_url, terms) +-- Has NO active agreements (only expired). The expiry carries an offer transaction, so it is an +-- indexer who had an offer on chain and let the window close, which earns the standard exclusion. +-- Leave the transaction set: without it the row reads as dipper failing to submit, and every test +-- that expects this row on the standard window would silently move to the short one instead. +INSERT INTO dipper_reg_indexing_agreements (id, nonce_uuid, created_at, updated_at, status, indexing_request_id, deployment_id, indexer_id, indexer_url, terms, offer_tx_hash) VALUES ('\xcc000000000000000000000000000001'::bytea, '01930100-0003-7000-8000-000000000001'::uuid, timezone('UTC', now()), timezone('UTC', now()), 5, -- Expired '01930100-0000-7000-8000-000000000001'::uuid, 'QmEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE5e', '\x3333333333333333333333333333333333333333'::bytea, 'https://indexer-c.com', - '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "service_provider": "0x3333333333333333333333333333333333333333", "data_service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "deadline": 1700000300, "ends_at": 1700086400, "max_initial_tokens": "1000", "max_ongoing_tokens_per_second": "100", "min_seconds_per_collection": 86400, "max_seconds_per_collection": 864000, "metadata": {"tokens_per_second": "10", "tokens_per_entity_per_second": "1", "subgraph_deployment_id": "QmEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE5e", "protocol_network": 1, "chain_id": 1}}'::json); + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "service_provider": "0x3333333333333333333333333333333333333333", "data_service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "deadline": 1700000300, "ends_at": 1700086400, "max_initial_tokens": "1000", "max_ongoing_tokens_per_second": "100", "min_seconds_per_collection": 86400, "max_seconds_per_collection": 864000, "metadata": {"tokens_per_second": "10", "tokens_per_entity_per_second": "1", "subgraph_deployment_id": "QmEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE5e", "protocol_network": 1, "chain_id": 1}}'::json, + decode(repeat('cc', 32), 'hex')); diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index 1cd628ed..84a63bcb 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -2095,6 +2095,186 @@ async fn get_declined_indexers_other_reason_excluded_after_30_days() { ); } +/// A dipper-caused expiry moves to the 5 minute window rather than out of the query entirely, +/// so inside that window the indexer still counts as declined and a reassessment seconds later +/// does not hand them the same deployment while our submission path is still failing. +#[tokio::test] +async fn get_declined_indexers_expiry_without_offer_tx_included_within_5_minutes() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + + // Age everything the fixture provides out of every window, so only the row under test + // can appear in the result. + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET updated_at = timezone('UTC', now()) - interval '31 days' + "#, + ) + .execute(&db) + .await + .expect("Failed to age the fixture agreements"); + + let agreement_id = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]); + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET status = 5, rejection_reason = NULL, offer_tx_hash = NULL, + updated_at = timezone('UTC', now()) - interval '1 minute' + WHERE id = $1 + "#, + ) + .bind(agreement_id) + .execute(&db) + .await + .expect("Failed to update agreement"); + + let registry = PgRegistry::new(db); + + //* When + let result = registry + .get_declined_indexers_by_deployment(30, 1, 5, 1) + .await + .expect("Failed to get declined indexers"); + + //* Then + let deployment_1a: DeploymentId = "QmAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA1a" + .parse() + .unwrap(); + let indexer_a = indexer_id!("1111111111111111111111111111111111111111"); + let declined = result + .get(&deployment_1a) + .expect("Deployment 1a should be in the declined list"); + assert!( + declined.contains(&indexer_a), + "an expiry we caused should still count within the 5 minute window, got {declined:?}" + ); +} + +/// An expiry with no offer transaction means we never landed the offer, so the indexer never +/// had one to accept and must not be benched for a month over it. On 2026-07-29 exactly this +/// benched a willing indexer for 30 days because our RPC provider answered HTTP 500. +#[tokio::test] +async fn get_declined_indexers_expiry_without_offer_tx_excluded_after_5_minutes() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + + // Age everything the fixture provides out of every window, so only the row under test + // can appear in the result. + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET updated_at = timezone('UTC', now()) - interval '31 days' + "#, + ) + .execute(&db) + .await + .expect("Failed to age the fixture agreements"); + + let agreement_id = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]); + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET status = 5, rejection_reason = NULL, offer_tx_hash = NULL, + updated_at = timezone('UTC', now()) - interval '1 hour' + WHERE id = $1 + "#, + ) + .bind(agreement_id) + .execute(&db) + .await + .expect("Failed to update agreement"); + + let registry = PgRegistry::new(db); + + //* When + let result = registry + .get_declined_indexers_by_deployment(30, 1, 5, 1) + .await + .expect("Failed to get declined indexers"); + + //* Then + assert!( + result.is_empty(), + "an expiry we caused must not bench the indexer, got {result:?}" + ); +} + +/// The mirror case: the offer did land, so the indexer had one on chain and let the window +/// close, which is their decision and does earn the standard bench. +#[tokio::test] +async fn get_declined_indexers_expiry_with_offer_tx_included_within_30_days() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET updated_at = timezone('UTC', now()) - interval '31 days' + "#, + ) + .execute(&db) + .await + .expect("Failed to age the fixture agreements"); + + let agreement_id = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]); + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET status = 5, rejection_reason = NULL, + offer_tx_hash = decode(repeat('11', 32), 'hex'), + updated_at = timezone('UTC', now()) - interval '1 hour' + WHERE id = $1 + "#, + ) + .bind(agreement_id) + .execute(&db) + .await + .expect("Failed to update agreement"); + + let registry = PgRegistry::new(db); + + //* When + let result = registry + .get_declined_indexers_by_deployment(30, 1, 5, 1) + .await + .expect("Failed to get declined indexers"); + + //* Then + let deployment_1a: DeploymentId = "QmAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA1a" + .parse() + .unwrap(); + let indexer_a = indexer_id!("1111111111111111111111111111111111111111"); + let declined = result + .get(&deployment_1a) + .expect("Deployment 1a should be in the declined list"); + assert!( + declined.contains(&indexer_a), + "an expiry after the offer landed should still bench the indexer, got {declined:?}" + ); +} + #[tokio::test] async fn get_declined_indexers_capacity_exceeded_included_within_5_minutes() { //* Given