fix: stop benching indexers for offers we never landed

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,
 

bin/dipper-service/src/registry/agreement.rs

@@ -173,8 +173,8 @@ pub trait AgreementRegistry {
     ) -> RegistryResult<std::collections::HashMap<DeploymentId, Vec<IndexerId>>>;
 
     /// 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,

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)) => {

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

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'));

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