From 8e44b3b1ab4c6a816f7301c84aa2c5aa391df7a6 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 5 Oct 2026 10:56:37 +0100 Subject: [PATCH 1/4] fix(liveness): end stale agreements through the cancelling status The check for indexers that stopped serving cancelled first and marked afterwards, missing the read before sending, the retry limit and the alert. It now marks the agreement cancelling, noted as abandoned so it still ends that way, and the code that only it used is gone. --- bin/dipper-service/src/cancel_dispatch.rs | 54 ++- .../src/network/service/cancel_retry.rs | 12 +- .../src/network/service/chain_listener.rs | 29 +- .../src/network/service/liveness_checker.rs | 379 +++++------------- bin/dipper-service/src/registry.rs | 21 +- bin/dipper-service/src/registry/agreement.rs | 26 +- .../src/registry/agreement_stub.rs | 22 +- .../handlers/reassess_indexing_request.rs | 1 + .../send_indexing_agreement_proposal.rs | 7 - .../20261005000001_add_abandoned_flag.sql | 5 + dipper-pgregistry/src/postgres.rs | 89 ++-- .../tests/it_registry_postgres.rs | 77 +++- 12 files changed, 314 insertions(+), 408 deletions(-) create mode 100644 dipper-pgregistry/migrations/20261005000001_add_abandoned_flag.sql diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 049dc105..e3be8cc1 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -1,5 +1,5 @@ -//! On-chain cancel dispatch. Every cancel goes through -//! [`cancel_agreement_on_chain`] so the manager-routed path lives in one place. +//! On-chain cancel dispatch. Every cancel starts with [`start_cancel`] and goes out through +//! `cancel_agreement_on_chain`, so the manager-routed path lives in one place. use dipper_core::time::now_secs; use thegraph_core::alloy::primitives::B256; @@ -23,7 +23,7 @@ const SCOPE_BOTH: u16 = SCOPE_ACTIVE | SCOPE_PENDING; /// Cancel an agreement on-chain through the RecurringAgreementManager. Passes /// both scope bits so the collector cancels whichever scope the agreement is in, /// and treats a missing or short stored hash as `MissingTermsVersionHash`. -pub async fn cancel_agreement_on_chain( +async fn cancel_agreement_on_chain( chain_client: &T, agreement: &IndexingAgreement, config: &IndexingAgreementConfig, @@ -68,10 +68,29 @@ pub async fn cancel_agreement_on_chain( } } +/// Why dipper is ending an agreement, which decides the status it ends in once the chain +/// confirms dipper's cancel. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CancelReason { + /// Dipper no longer wants it: it ends `CanceledByRequester`. + NotWanted, + /// Its indexer stopped serving it: it ends `AbandonedByIndexer`. + Abandoned, +} + +impl CancelReason { + fn ended_status(self) -> &'static str { + match self { + Self::NotWanted => "CANCELED_BY_REQUESTER", + Self::Abandoned => "ABANDONED_BY_INDEXER", + } + } +} + /// What [`start_cancel`] left an agreement as. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CancelStarted { - /// It was accepted and its cancel landed: now `CanceledByRequester`. + /// It was accepted and its cancel landed: now ended, as its [`CancelReason`] says. Ended, /// Still `Cancelling`; the chain listener finishes it once it can't go live. Cancelling, @@ -84,15 +103,25 @@ pub async fn start_cancel( registry: &R, chain_client: &T, agreement: &IndexingAgreement, + reason: CancelReason, config: &IndexingAgreementConfig, ) -> RegistryResult where R: AgreementRegistry + Sync, T: ChainClient, { - registry - .mark_indexing_agreement_as_cancelling(&agreement.id) - .await?; + match reason { + CancelReason::NotWanted => { + registry + .mark_indexing_agreement_as_cancelling(&agreement.id) + .await? + } + CancelReason::Abandoned => { + registry + .mark_indexing_agreement_as_abandoning(&agreement.id) + .await? + } + } let tx_hash = match cancel_if_live(chain_client, agreement, config).await { LiveCancel::Ended(tx_hash) => tx_hash, LiveCancel::NotLive { .. } => return Ok(CancelStarted::Cancelling), @@ -119,7 +148,7 @@ where return Ok(CancelStarted::Cancelling); } Ok( - if confirm_cancelled(registry, agreement, tx_hash, config).await { + if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { CancelStarted::Ended } else { CancelStarted::Cancelling @@ -127,12 +156,13 @@ where ) } -/// Mark an agreement the chain shows dipper ended `CanceledByRequester`, recording the cancel -/// when its transaction is known, so the `terminated` sweep announces it. False, logged, when -/// the mark fails; it stays `Cancelling` for the cancel retry. +/// Mark an agreement the chain shows dipper ended as ended, recording the cancel when its +/// transaction is known, so the `terminated` sweep announces it. False, logged, when the mark +/// fails; it stays `Cancelling` for the cancel retry. pub async fn confirm_cancelled( registry: &R, agreement: &IndexingAgreement, + reason: CancelReason, tx_hash: Option, config: &IndexingAgreementConfig, ) -> bool { @@ -155,7 +185,7 @@ pub async fn confirm_cancelled( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, old_status = "CANCELLING", - new_status = "CANCELED_BY_REQUESTER", + new_status = reason.ended_status(), reason = "cancel_confirmed_on_chain", "agreement state transition" ); diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 5a9025e2..1ba0fa73 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -6,7 +6,9 @@ use dipper_core::time::now_secs; use thegraph_core::alloy::primitives::B256; use crate::{ - cancel_dispatch::{LiveCancel, cancel_if_live, confirm_cancelled, log_unconfirmed}, + cancel_dispatch::{ + CancelReason, LiveCancel, cancel_if_live, confirm_cancelled, log_unconfirmed, + }, chain_client::{ChainClient, ChainClientError}, config::IndexingAgreementConfig, registry::{AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement}, @@ -162,7 +164,12 @@ async fn confirm_if_over( ); return past_grace && record_end_by_indexer(registry, agreement).await; } - confirm_cancelled(registry, agreement, tx_hash, config).await + let reason = if row.abandoned { + CancelReason::Abandoned + } else { + CancelReason::NotWanted + }; + confirm_cancelled(registry, agreement, reason, tx_hash, config).await } /// Mark an agreement the indexer ended `CanceledByIndexer` when the chain listener hasn't in @@ -492,6 +499,7 @@ mod tests { agreement: cancelling, accepted_on_chain, ended_seen_at: None, + abandoned: false, }], ..MockRegistry::default() } diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index e7e5c76f..4da1555c 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -1086,7 +1086,7 @@ where match prep.item.cancel { Some(CancelKind::ByRequester) => tracing::info!( agreement_id = %prep.agreement.id, - "Agreement marked as CanceledByRequester (on-chain confirmation)" + "Agreement marked as ended by dipper (on-chain confirmation)" ), Some(CancelKind::ByIndexer) => tracing::info!( agreement_id = %prep.agreement.id, @@ -1224,8 +1224,14 @@ where None => return Ok(false), } } - let started = - crate::cancel_dispatch::start_cancel(registry, chain_client, &old_agreement, config).await; + let started = crate::cancel_dispatch::start_cancel( + registry, + chain_client, + &old_agreement, + crate::cancel_dispatch::CancelReason::NotWanted, + config, + ) + .await; Ok(note_replaced_cancel( new_agreement_id, &old_agreement, @@ -1314,8 +1320,14 @@ async fn sweep_orphan_canceled_agreements( }; for agreement in orphans { - let started = - crate::cancel_dispatch::start_cancel(registry, chain_client, &agreement, config).await; + let started = crate::cancel_dispatch::start_cancel( + registry, + chain_client, + &agreement, + crate::cancel_dispatch::CancelReason::NotWanted, + config, + ) + .await; log_orphan_cancel(&agreement, started); } } @@ -2439,13 +2451,6 @@ mod tests { Ok((per_indexer, global)) } - async fn mark_indexing_agreement_as_abandoned( - &self, - _id: &IndexingAgreementId, - ) -> RegistryResult { - Err(crate::registry::Error::NoRecordsUpdated) - } - async fn get_agreement_fee_rates(&self) -> RegistryResult> { Ok(vec![]) } diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 4c2733a4..ebef68ce 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -43,8 +43,8 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use url::Url; use crate::{ - cancel_dispatch::LiveCancel, - chain_client::{ChainClient, ChainClientError}, + cancel_dispatch::CancelReason, + chain_client::ChainClient, config::LivenessCheckerConfig, network::provider::NetworkProviderService, registry::{ @@ -481,10 +481,9 @@ async fn record_progress( } } -/// Cancel a stale agreement on-chain and queue reassessment. -/// -/// If the on-chain cancel fails, the DB is not updated and reassessment is not -/// queued, leaving the agreement in `AcceptedOnChain` for the next cycle to retry. +/// End a stale agreement and queue a reassessment to replace it. It is marked `Cancelling` +/// before any cancel is sent, so the cancel retry finishes one that fails here, and it ends +/// `AbandonedByIndexer` once the chain confirms dipper's cancel. #[allow(clippy::too_many_arguments)] #[expect( clippy::cognitive_complexity, @@ -504,102 +503,37 @@ async fn cancel_and_reassess( W: WorkerQueue + Send + Sync, C: ChainClient + Send + Sync, { - // 1. Cancel on-chain (mode-aware dispatch) - let mut on_chain_cancel_tx: Option = None; - match crate::cancel_dispatch::cancel_agreement_on_chain(chain_client, agreement, agreement_conf) - .await - { - LiveCancel::Ended(tx_hash) => { - tracing::info!( - agreement_id = %agreement.id, - tx_hash = ?tx_hash, - "canceled stale agreement on-chain" - ); - on_chain_cancel_tx = tx_hash.map(|hash| hash.to_string()); - } - LiveCancel::NotLive { .. } => { - tracing::info!( - agreement_id = %agreement.id, - "stale agreement already canceled on-chain; proceeding to mark abandoned" - ); - } - LiveCancel::CancelFailed(err @ ChainClientError::MissingTermsVersionHash { .. }) => { - // Permanent per-agreement condition: the on-chain agreement is - // still live, so do NOT mark abandoned (that would hide a - // money-draining agreement). Surface for operator action. - tracing::error!( - agreement_id = %agreement.id, - error = %err, - "cannot cancel stale agreement: missing terms_version_hash; leaving active for operator action" - ); - return; - } - LiveCancel::CancelFailed(ChainClientError::ConfigError(_)) => { - // Chain client disabled: still proceed to mark and reassess so the - // DB reflects the detected abandonment even without an on-chain tx. - tracing::warn!( - agreement_id = %agreement.id, - "chain client not configured, skipping on-chain cancellation" - ); - } - LiveCancel::Unconfirmed { tx_hash, err } => { - crate::cancel_dispatch::log_unconfirmed(agreement, tx_hash, &err); - return; - } - LiveCancel::ReadFailed(err) | LiveCancel::CancelFailed(err) => { - tracing::error!( - agreement_id = %agreement.id, - error = %err, - "failed to cancel stale agreement on-chain, will retry next cycle" - ); - return; - } - } - - // 2. Mark as abandoned in DB - let abandoned = match tokio::time::timeout( - db_timeout, - registry.mark_indexing_agreement_as_abandoned(&agreement.id), + // 1. Start the cancel + match crate::cancel_dispatch::start_cancel( + registry, + chain_client, + agreement, + CancelReason::Abandoned, + agreement_conf, ) .await { - Ok(Ok(a)) => a, - Ok(Err(err)) => { - tracing::error!( + Ok(started) => tracing::info!( + agreement_id = %agreement.id, + ?started, + reason = "indexer_stale", + "Cancelling stale agreement" + ), + Err(crate::registry::Error::NoRecordsUpdated) => { + tracing::debug!( agreement_id = %agreement.id, - error = %err, - "failed to mark agreement as abandoned" + "Stale agreement already ended or being cancelled" ); return; } - Err(_) => { + Err(err) => { tracing::error!( agreement_id = %agreement.id, - "timeout marking agreement as abandoned" + error = %err, + "failed to mark stale agreement cancelling, will retry next cycle" ); return; } - }; - - // The accepted agreement was canceled on-chain because the indexer went - // stale. Record the cancel audit so the chain_listener's `terminated` sweep - // announces it durably: this row is marked `AbandonedByIndexer` (terminal) - // and was accepted on-chain, so it is sweep-eligible. - let manager = agreement_conf.recurring_agreement_manager().to_string(); - if let Err(err) = registry - .record_cancel_audit( - &agreement.id, - dipper_core::time::now_secs(), - &manager, - on_chain_cancel_tx.as_deref(), - ) - .await - { - tracing::warn!( - agreement_id = %agreement.id, - error = %err, - "failed to record cancel audit; terminated event may emit with fallback fields" - ); } // Clean up pending cancellations: if this abandoned agreement was a @@ -615,10 +549,10 @@ async fn cancel_and_reassess( ); } - // 3. Fetch the indexing request for num_candidates + // 2. Fetch the indexing request for num_candidates let request = match tokio::time::timeout( db_timeout, - registry.get_indexing_request_by_id(&abandoned.indexing_request_id), + registry.get_indexing_request_by_id(&agreement.indexing_request_id), ) .await { @@ -626,7 +560,7 @@ async fn cancel_and_reassess( Ok(Ok(None)) => { tracing::warn!( agreement_id = %agreement.id, - indexing_request_id = %abandoned.indexing_request_id, + indexing_request_id = %agreement.indexing_request_id, "indexing request not found for abandoned agreement" ); return; @@ -648,13 +582,13 @@ async fn cancel_and_reassess( } }; - // 4. Queue reassessment + // 3. Queue reassessment let push_result = tokio::time::timeout( queue_timeout, worker_queue.reassess_indexing_request( - abandoned.indexing_request_id, - abandoned.terms.metadata.subgraph_deployment_id, - abandoned.terms.metadata.chain_id, + agreement.indexing_request_id, + agreement.terms.metadata.subgraph_deployment_id, + agreement.terms.metadata.chain_id, request.num_candidates, // Background: abandonment remediation yields to interactive work. JobPriority::Background, @@ -666,7 +600,7 @@ async fn cancel_and_reassess( Ok(Ok(_job_id)) => { tracing::info!( agreement_id = %agreement.id, - indexing_request_id = %abandoned.indexing_request_id, + indexing_request_id = %agreement.indexing_request_id, "queued reassessment for abandoned agreement" ); } @@ -920,7 +854,10 @@ mod tests { #[derive(Clone, Default)] struct MockCalls { progress_updates: Arc>>, - abandoned: Arc>>, + /// Ids marked cancelling as abandoned, before any cancel is sent. + abandoning: Arc>>, + /// Ids marked ended once the chain confirmed dipper's cancel. + ended: Arc>>, reassessments: Arc>>, chain_cancels: Arc>>, /// Ids passed to `record_cancel_audit` -- the signal the handler drives @@ -930,30 +867,19 @@ mod tests { struct MockRegistry { calls: MockCalls, - mark_abandoned_result: Arc>>>, + already_ending: bool, get_request_result: Arc>>>>, } impl MockRegistry { fn new(calls: MockCalls, agreement: IndexingAgreement) -> Self { - let abandoned_agreement = { - let mut a = agreement.clone(); - a.status = IndexingAgreementStatus::AbandonedByIndexer; - a - }; let request = make_request(agreement.indexing_request_id, 2); Self { calls, - mark_abandoned_result: Arc::new(Mutex::new(Some(Ok(abandoned_agreement)))), + already_ending: false, get_request_result: Arc::new(Mutex::new(Some(Ok(Some(request))))), } } - - fn with_chain_error(calls: MockCalls, agreement: IndexingAgreement) -> Self { - let mut mock = Self::new(calls, agreement); - mock.mark_abandoned_result = Arc::new(Mutex::new(None)); - mock - } } #[async_trait] @@ -978,16 +904,23 @@ mod tests { .push((*id, block_height)); Ok(()) } - async fn mark_indexing_agreement_as_abandoned( + async fn mark_indexing_agreement_as_abandoning( &self, id: &IndexingAgreementId, - ) -> RegistryResult { - self.calls.abandoned.lock().unwrap().push(*id); - self.mark_abandoned_result - .lock() - .unwrap() - .take() - .expect("mark_abandoned called more than once") + ) -> RegistryResult<()> { + if self.already_ending { + return Err(crate::registry::Error::NoRecordsUpdated); + } + self.calls.abandoning.lock().unwrap().push(*id); + Ok(()) + } + + async fn mark_indexing_agreement_as_canceled_by_requester( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()> { + self.calls.ended.lock().unwrap().push(*id); + Ok(()) } async fn record_cancel_audit( @@ -1116,9 +1049,11 @@ mod tests { } } + /// An agreement live on-chain until a cancel that succeeds ends it. struct MockChainClient { calls: MockCalls, result: Result, + live: std::sync::atomic::AtomicBool, } impl MockChainClient { @@ -1126,13 +1061,7 @@ mod tests { Self { calls, result: Ok(B256::ZERO), - } - } - - fn config_error(calls: MockCalls) -> Self { - Self { - calls, - result: Err(ChainClientError::ConfigError("disabled".into())), + live: true.into(), } } @@ -1140,6 +1069,7 @@ mod tests { Self { calls, result: Err(ChainClientError::RpcError(anyhow::anyhow!("network error"))), + live: true.into(), } } } @@ -1172,12 +1102,9 @@ mod tests { // existing cancel-path assertions hold. self.calls.chain_cancels.lock().unwrap().push(*agreement_id); match &self.result { - Ok(hash) => Ok(Some(*hash)), - Err(ChainClientError::ConfigError(s)) => { - Err(ChainClientError::ConfigError(s.clone())) - } - Err(ChainClientError::RpcError(e)) => { - Err(ChainClientError::RpcError(anyhow::anyhow!("{e}"))) + Ok(hash) => { + self.live.store(false, std::sync::atomic::Ordering::SeqCst); + Ok(Some(*hash)) } Err(e) => Err(ChainClientError::RpcError(anyhow::anyhow!("{e}"))), } @@ -1203,9 +1130,9 @@ mod tests { &self, _agreement_id: &[u8; 16], ) -> Result { - // Cancel dispatch reads back after a mined cancel; reporting - // not-active means "cancel confirmed", which these tests expect. - Ok(AgreementOnChain::NotLive) + Ok(AgreementOnChain::live_if( + self.live.load(std::sync::atomic::Ordering::SeqCst), + )) } } @@ -1450,183 +1377,99 @@ mod tests { // ---- cancel_and_reassess behavior tests ---- - #[tokio::test] - async fn test_cancel_and_reassess_success() { - // Arrange + fn stale_agreement() -> IndexingAgreement { let dep: DeploymentId = "QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9" .parse() .unwrap(); - let indexer_id = IndexerId::from(Address::ZERO); let url: Url = "http://indexer.example.com/".parse().unwrap(); - let agreement = make_agreement( - indexer_id, + make_agreement( + IndexerId::from(Address::ZERO), url, dep, Some(100), Some(OffsetDateTime::now_utc()), - ); - let req_id = agreement.indexing_request_id; - let agr_id = agreement.id; + ) + } - let calls = MockCalls::default(); - let registry = MockRegistry::new(calls.clone(), agreement.clone()); + async fn end_stale( + agreement: &IndexingAgreement, + registry: &MockRegistry, + chain: &MockChainClient, + ) { let queue = MockWorkerQueue { - calls: calls.clone(), + calls: registry.calls.clone(), }; - let chain = MockChainClient::success(calls.clone()); - - // Act cancel_and_reassess( - &agreement, - ®istry, + agreement, + registry, &queue, - &chain, + chain, &test_agreement_conf(), DB_TIMEOUT, QUEUE_TIMEOUT, ) .await; - - // Assert - assert_eq!( - calls.chain_cancels.lock().unwrap().as_slice(), - &[agreement.id.into_bytes()] - ); - assert_eq!(calls.abandoned.lock().unwrap().as_slice(), &[agr_id]); - assert_eq!(calls.reassessments.lock().unwrap().as_slice(), &[req_id]); } #[tokio::test] - async fn cancel_and_reassess_records_cancel_audit() { - // The stale-agreement cancel no longer emits `terminated` directly: it - // records the cancel audit and the chain_listener sweep announces it. - let dep: DeploymentId = "QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9" - .parse() - .unwrap(); - let indexer_id = IndexerId::from(Address::ZERO); - let url: Url = "http://indexer.example.com/".parse().unwrap(); - let agreement = make_agreement( - indexer_id, - url, - dep, - Some(100), - Some(OffsetDateTime::now_utc()), - ); - let agr_id = agreement.id; - + async fn ends_a_stale_agreement_through_cancelling_and_reassesses() { + let agreement = stale_agreement(); let calls = MockCalls::default(); let registry = MockRegistry::new(calls.clone(), agreement.clone()); - let queue = MockWorkerQueue { - calls: calls.clone(), - }; let chain = MockChainClient::success(calls.clone()); - cancel_and_reassess( - &agreement, - ®istry, - &queue, - &chain, - &test_agreement_conf(), - DB_TIMEOUT, - QUEUE_TIMEOUT, - ) - .await; + end_stale(&agreement, ®istry, &chain).await; + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + assert_eq!( + calls.chain_cancels.lock().unwrap().as_slice(), + &[agreement.id.into_bytes()] + ); + assert_eq!(calls.ended.lock().unwrap().as_slice(), &[agreement.id]); assert_eq!( calls.cancel_audits.lock().unwrap().as_slice(), - &[agr_id], - "exactly one cancel audit recorded for the stale agreement" + &[agreement.id], + "its cancel is recorded, so the terminated sweep announces it" + ); + assert_eq!( + calls.reassessments.lock().unwrap().as_slice(), + &[agreement.indexing_request_id] ); } #[tokio::test] - async fn test_cancel_and_reassess_config_error_proceeds() { - // Arrange: chain client disabled (ConfigError) → still mark abandoned and reassess - let dep: DeploymentId = "QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9" - .parse() - .unwrap(); - let indexer_id = IndexerId::from(Address::ZERO); - let url: Url = "http://indexer.example.com/".parse().unwrap(); - let agreement = make_agreement( - indexer_id, - url, - dep, - Some(100), - Some(OffsetDateTime::now_utc()), - ); - let req_id = agreement.indexing_request_id; - let agr_id = agreement.id; - + async fn leaves_a_failed_cancel_to_the_retry_and_still_reassesses() { + // Marked cancelling, the agreement is out of the checker's sight and its indexer out + // of selection, so it is replaced now and the cancel retry finishes the cancel. + let agreement = stale_agreement(); let calls = MockCalls::default(); let registry = MockRegistry::new(calls.clone(), agreement.clone()); - let queue = MockWorkerQueue { - calls: calls.clone(), - }; - let chain = MockChainClient::config_error(calls.clone()); + let chain = MockChainClient::rpc_error(calls.clone()); - // Act - cancel_and_reassess( - &agreement, - ®istry, - &queue, - &chain, - &test_agreement_conf(), - DB_TIMEOUT, - QUEUE_TIMEOUT, - ) - .await; + end_stale(&agreement, ®istry, &chain).await; - // Assert: no on-chain cancel (ConfigError is treated as disabled, not a real error) - // but DB mark and reassessment still happen + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + assert!(calls.ended.lock().unwrap().is_empty()); + assert!(calls.cancel_audits.lock().unwrap().is_empty()); assert_eq!( - calls.chain_cancels.lock().unwrap().as_slice(), - &[agreement.id.into_bytes()] + calls.reassessments.lock().unwrap().as_slice(), + &[agreement.indexing_request_id] ); - assert_eq!(calls.abandoned.lock().unwrap().as_slice(), &[agr_id]); - assert_eq!(calls.reassessments.lock().unwrap().as_slice(), &[req_id]); } #[tokio::test] - async fn test_cancel_and_reassess_chain_error_skips() { - // Arrange: transient RPC error → do nothing (retry next cycle) - let dep: DeploymentId = "QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9" - .parse() - .unwrap(); - let indexer_id = IndexerId::from(Address::ZERO); - let url: Url = "http://indexer.example.com/".parse().unwrap(); - let agreement = make_agreement( - indexer_id, - url, - dep, - Some(100), - Some(OffsetDateTime::now_utc()), - ); - + async fn leaves_an_agreement_already_ending_alone() { + let agreement = stale_agreement(); let calls = MockCalls::default(); - let registry = MockRegistry::with_chain_error(calls.clone(), agreement.clone()); - let queue = MockWorkerQueue { - calls: calls.clone(), + let registry = MockRegistry { + already_ending: true, + ..MockRegistry::new(calls.clone(), agreement.clone()) }; - let chain = MockChainClient::rpc_error(calls.clone()); + let chain = MockChainClient::success(calls.clone()); - // Act - cancel_and_reassess( - &agreement, - ®istry, - &queue, - &chain, - &test_agreement_conf(), - DB_TIMEOUT, - QUEUE_TIMEOUT, - ) - .await; + end_stale(&agreement, ®istry, &chain).await; - // Assert: chain cancel attempted, but DB and queue untouched - assert_eq!( - calls.chain_cancels.lock().unwrap().as_slice(), - &[agreement.id.into_bytes()] - ); - assert!(calls.abandoned.lock().unwrap().is_empty()); + assert!(calls.chain_cancels.lock().unwrap().is_empty()); assert!(calls.reassessments.lock().unwrap().is_empty()); } diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 89c96724..239ff563 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -401,6 +401,16 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn mark_indexing_agreement_as_abandoning( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()> { + self.inner + .mark_indexing_agreement_as_abandoning(id) + .await + .map_err(Into::into) + } + async fn reopen_indexing_agreement_cancel( &self, id: &IndexingAgreementId, @@ -716,17 +726,6 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } - async fn mark_indexing_agreement_as_abandoned( - &self, - id: &IndexingAgreementId, - ) -> RegistryResult { - let raw = self.inner.mark_indexing_agreement_as_abandoned(id).await?; - // The conversion only fails for Unknown status; since we just wrote - // AbandonedByIndexer, this cannot fail in practice. - IndexingAgreement::try_from(raw) - .map_err(|_| dipper_pgregistry::Error::NoRecordsUpdated.into()) - } - async fn get_agreement_fee_rates(&self) -> RegistryResult> { self.inner .get_agreement_fee_rates() diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 3aeb452e..0b94535a 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -260,7 +260,8 @@ pub trait AgreementRegistry { /// /// If there is no indexing agreement with the given ID, or if the agreement is not in the /// `CREATED`, `ACCEPTED_ON_CHAIN`, `REJECTED` or `CANCELLING` state, this method returns a - /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. One dipper was cancelling because its + /// indexer stopped serving it becomes `ABANDONED_BY_INDEXER` instead. async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -273,6 +274,14 @@ pub trait AgreementRegistry { id: &IndexingAgreementId, ) -> RegistryResult<()>; + /// Mark an `ACCEPTED_ON_CHAIN` agreement whose indexer stopped serving it `CANCELLING`, + /// before its cancel is sent, so it ends `ABANDONED_BY_INDEXER` once the chain confirms it; + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) otherwise. + async fn mark_indexing_agreement_as_abandoning( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()>; + /// Move a `CANCELED_BY_REQUESTER` or `REJECTED` agreement the chain shows live back to /// `CANCELLING`, its cancel attempts reset; [`NoRecordUpdated`](Error::NoRecordsUpdated) /// otherwise. @@ -527,17 +536,6 @@ pub trait AgreementRegistry { .map(|m| !m.is_empty()) } - /// Mark an indexing agreement as `ABANDONED_BY_INDEXER`. - /// - /// Transitions `AcceptedOnChain → AbandonedByIndexer`. Returns the full agreement - /// for use in the subsequent reassessment call. - /// Returns [`NoRecordsUpdated`](Error::NoRecordsUpdated) if the agreement doesn't - /// exist or isn't in `AcceptedOnChain` status. - async fn mark_indexing_agreement_as_abandoned( - &self, - id: &IndexingAgreementId, - ) -> RegistryResult; - /// Get per-agreement rate fields from active agreements. /// /// Returns base rate, entity rate, and deployment ID for each active @@ -566,6 +564,9 @@ pub struct CancellingAgreement { pub accepted_on_chain: bool, /// When a check first found it no longer live on-chain, if one has. pub ended_seen_at: Option, + /// Whether it is being cancelled because its indexer stopped serving it, so it ends + /// `ABANDONED_BY_INDEXER`. + pub abandoned: bool, } impl TryFrom for CancellingAgreement { @@ -576,6 +577,7 @@ impl TryFrom for CancellingAgreement { agreement: value.agreement.try_into()?, accepted_on_chain: value.accepted_on_chain, ended_seen_at: value.ended_seen_at, + abandoned: value.abandoned, }) } } diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index aa91b431..cbb62dd6 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -137,6 +137,10 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_indexing_agreement_as_cancelling") } + async fn mark_indexing_agreement_as_abandoning(&self, _id: &IndexingAgreementId) -> Result<()> { + unimplemented!("mark_indexing_agreement_as_abandoning") + } + async fn reopen_indexing_agreement_cancel(&self, _id: &IndexingAgreementId) -> Result<()> { unimplemented!("reopen_indexing_agreement_cancel") } @@ -241,13 +245,6 @@ pub trait StubAgreementRegistry: Send + Sync { .map(|m| !m.is_empty()) } - async fn mark_indexing_agreement_as_abandoned( - &self, - _id: &IndexingAgreementId, - ) -> Result { - unimplemented!("mark_indexing_agreement_as_abandoned") - } - async fn get_agreement_fee_rates(&self) -> Result> { unimplemented!("get_agreement_fee_rates") } @@ -445,6 +442,10 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_indexing_agreement_as_cancelling(self, id).await } + async fn mark_indexing_agreement_as_abandoning(&self, id: &IndexingAgreementId) -> Result<()> { + StubAgreementRegistry::mark_indexing_agreement_as_abandoning(self, id).await + } + async fn reopen_indexing_agreement_cancel(&self, id: &IndexingAgreementId) -> Result<()> { StubAgreementRegistry::reopen_indexing_agreement_cancel(self, id).await } @@ -546,13 +547,6 @@ impl AgreementRegistry for T { StubAgreementRegistry::exists_active_agreements(self).await } - async fn mark_indexing_agreement_as_abandoned( - &self, - id: &IndexingAgreementId, - ) -> Result { - StubAgreementRegistry::mark_indexing_agreement_as_abandoned(self, id).await - } - async fn get_agreement_fee_rates(&self) -> Result> { StubAgreementRegistry::get_agreement_fee_rates(self).await } diff --git a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs index 1e486ff1..9fcc916b 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -1901,6 +1901,7 @@ where &ctx.registry, &ctx.chain_client, agreement, + crate::cancel_dispatch::CancelReason::NotWanted, &ctx.agreement_conf, ) .await diff --git a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs index 1022faff..e3e6fbc4 100644 --- a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs +++ b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs @@ -607,13 +607,6 @@ mod tests { Ok((std::collections::HashMap::new(), 0)) } - async fn mark_indexing_agreement_as_abandoned( - &self, - _id: &IndexingAgreementId, - ) -> crate::registry::Result { - Err(crate::registry::Error::NoRecordsUpdated) - } - async fn get_agreement_fee_rates(&self) -> crate::registry::Result> { Ok(vec![]) } diff --git a/dipper-pgregistry/migrations/20261005000001_add_abandoned_flag.sql b/dipper-pgregistry/migrations/20261005000001_add_abandoned_flag.sql new file mode 100644 index 00000000..87fd09c3 --- /dev/null +++ b/dipper-pgregistry/migrations/20261005000001_add_abandoned_flag.sql @@ -0,0 +1,5 @@ +-- abandoned: dipper is cancelling the agreement because its indexer stopped serving it, so once +-- the chain confirms dipper's cancel it ends AbandonedByIndexer (8) rather than +-- CanceledByRequester (3). +ALTER TABLE dipper_reg_indexing_agreements + ADD COLUMN abandoned BOOLEAN NOT NULL DEFAULT false; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 14915221..e1fee4f1 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -195,6 +195,8 @@ pub struct CancellingAgreement { pub accepted_on_chain: bool, /// When a check first found it no longer live on-chain, if one has. pub ended_seen_at: Option, + /// Whether it is being cancelled because its indexer stopped serving it. + pub abandoned: bool, } impl sqlx::FromRow<'_, sqlx::postgres::PgRow> for CancellingAgreement { @@ -205,6 +207,7 @@ impl sqlx::FromRow<'_, sqlx::postgres::PgRow> for CancellingAgreement { agreement: IndexingAgreement::from_row(row)?, accepted_on_chain: accepted_at.is_some(), ended_seen_at: row.try_get("ended_seen_at")?, + abandoned: row.try_get("abandoned")?, }) } } @@ -943,6 +946,7 @@ impl PgRegistry { Ok(()) } + /// One being cancelled because its indexer stopped serving it ends `AbandonedByIndexer`. pub async fn mark_indexing_agreement_as_canceled_by_requester( &self, agreement_id: &IndexingAgreementId, @@ -974,6 +978,33 @@ impl PgRegistry { .await } + /// Start ending an accepted agreement whose indexer stopped serving it: `Cancelling`, and + /// noted as abandoned, so the chain confirming dipper's cancel ends it `AbandonedByIndexer`. + pub async fn mark_indexing_agreement_as_abandoning( + &self, + agreement_id: &IndexingAgreementId, + ) -> Result<(), Error> { + let updated = sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET + status = $1, + abandoned = true, + updated_at = timezone('UTC', now()) + WHERE id = $2 AND status = $3 + "#, + ) + .bind(IndexingAgreementStatus::Cancelling) + .bind(agreement_id) + .bind(IndexingAgreementStatus::AcceptedOnChain) + .execute(&self.pool) + .await?; + if updated.rows_affected() == 0 { + return Err(Error::NoRecordsUpdated); + } + Ok(()) + } + /// Move an agreement dipper had already ended, cancelled or rejected, back to `Cancelling` /// once the chain shows it live after all, with its cancel attempts started afresh. It counts /// as checked, since the chain was just read and no cancel sent, so the retry takes it on its @@ -1036,7 +1067,8 @@ impl PgRegistry { rejection_reason, terms_version_hash, accepted_at, - ended_seen_at + ended_seen_at, + abandoned FROM dipper_reg_indexing_agreements WHERE status = $1 AND ( @@ -1933,50 +1965,6 @@ impl PgRegistry { Ok(exists) } - /// Mark an agreement as `AbandonedByIndexer`. - /// - /// Transitions `AcceptedOnChain → AbandonedByIndexer`. Returns the full - /// agreement for use in the subsequent reassessment call. - /// - /// Returns [`NoRecordsUpdated`](Error::NoRecordsUpdated) if the agreement - /// doesn't exist or isn't in `AcceptedOnChain` status. - pub async fn mark_indexing_agreement_as_abandoned( - &self, - agreement_id: &IndexingAgreementId, - ) -> Result { - let record: Option = sqlx::query_as( - r#" - UPDATE dipper_reg_indexing_agreements - SET - status = $1, - updated_at = timezone('UTC', now()) - WHERE id = $2 AND status = $3 - RETURNING - id, - nonce_uuid, - created_at, - updated_at, - status, - indexing_request_id, - deployment_id, - indexer_id, - indexer_url, - terms, - last_block_height, - last_progress_at, - rejection_reason, - terms_version_hash - "#, - ) - .bind(IndexingAgreementStatus::AbandonedByIndexer) - .bind(agreement_id) - .bind(IndexingAgreementStatus::AcceptedOnChain) - .fetch_optional(&self.pool) - .await?; - - record.ok_or(Error::NoRecordsUpdated) - } - // ========================================================================= // Indexer denylist operations // ========================================================================= @@ -2308,20 +2296,25 @@ async fn batch_update_status_from( return Ok(Vec::new()); } let placeholders = (0..allowed_from.len()) - .map(|i| format!("${}", i + 3)) + .map(|i| format!("${}", i + 5)) .collect::>() .join(", "); + // An agreement dipper ended because its indexer stopped serving it ends as abandoned. let sql = format!( r#" UPDATE dipper_reg_indexing_agreements - SET status = $1, updated_at = timezone('UTC', now()) + SET + status = CASE WHEN abandoned AND $1 = $3 THEN $4 ELSE $1 END, + updated_at = timezone('UTC', now()) WHERE id = ANY($2) AND status IN ({placeholders}) RETURNING id "# ); let mut query = sqlx::query_as::<_, (IndexingAgreementId,)>(&sql) .bind(new_status) - .bind(agreement_ids); + .bind(agreement_ids) + .bind(IndexingAgreementStatus::CanceledByRequester) + .bind(IndexingAgreementStatus::AbandonedByIndexer); for status in allowed_from { query = query.bind(*status); } diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index e9de6789..c0b5de0f 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -1768,43 +1768,76 @@ async fn test_count_active_agreements_by_deployment() { ); } -#[tokio::test] -async fn test_mark_as_abandoned_transitions_status() { - //* Given +/// Start abandoning fixture 0002's accepted agreement, then end it the way `end` does. +#[expect( + clippy::expect_used, + reason = "a test helper fails its test on any error" +)] +async fn abandon_then_end(end: F) -> (Result<(), Error>, IndexingAgreementStatus) +where + F: AsyncFnOnce(&PgRegistry, &IndexingAgreementId), +{ let (db, _temp_db) = temp_registry_db().await; run_fixture(&db, include_str!("fixtures/0002_indexing_agreements.sql")) .await .expect("Failed to run fixture"); let registry = PgRegistry::new(db); - - // AcceptedOnChain agreement from fixture 0002 let agreement_id = IndexingAgreementId::from_bytes([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 3]); - - //* When - let abandoned = registry - .mark_indexing_agreement_as_abandoned(&agreement_id) + registry + .mark_indexing_agreement_as_abandoning(&agreement_id) + .await + .expect("an accepted agreement can be abandoned"); + let listed = registry + .get_cancelling_agreements(100, 10, 0) .await - .expect("Failed to mark agreement as abandoned"); + .expect("cancelling query"); + assert!(listed[0].abandoned); - //* Then - assert_eq!( - abandoned.status, - IndexingAgreementStatus::AbandonedByIndexer, - "Status should be AbandonedByIndexer" - ); + end(®istry, &agreement_id).await; - // Second call must fail — agreement is no longer AcceptedOnChain - let err = registry - .mark_indexing_agreement_as_abandoned(&agreement_id) + let again = registry + .mark_indexing_agreement_as_abandoning(&agreement_id) + .await; + let ended = registry + .get_indexing_agreement_by_id(&agreement_id) .await - .expect_err("Expected error on second mark_as_abandoned call"); + .expect("agreement query") + .expect("agreement"); + (again, ended.status) +} + +#[tokio::test] +async fn an_abandoned_agreement_dipper_cancels_ends_abandoned() { + let (again, status) = abandon_then_end(async |registry, id| { + registry + .mark_indexing_agreement_as_canceled_by_requester(id) + .await + .expect("dipper's cancel confirmed"); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::AbandonedByIndexer); assert!( - matches!(err, Error::NoRecordsUpdated), - "Expected NoRecordsUpdated, got: {err:?}" + matches!(again, Err(Error::NoRecordsUpdated)), + "got {again:?}" ); } +#[tokio::test] +async fn an_abandoned_agreement_the_listener_sees_cancelled_ends_abandoned() { + let (_, status) = abandon_then_end(async |registry, id| { + let outcome = registry + .apply_reconciliation(id, false, Some(CancelKind::ByRequester)) + .await + .expect("reconciliation"); + assert!(outcome.did_cancel); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::AbandonedByIndexer); +} + // ============================================================================= // Rejection reason storage tests // ============================================================================= From 9c52bd626949f130dbe87da5c99a959d16066101 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 5 Oct 2026 11:09:24 +0100 Subject: [PATCH 2/4] docs(cancel): describe how an abandoned agreement now ends Several comments still said a cancelled agreement always ends cancelled by dipper and that the liveness check cancels directly. They now say an agreement dipper ends because its indexer stopped serving it passes through cancelling and ends abandoned by the indexer. --- bin/dipper-service/src/network/service/cancel_retry.rs | 5 +++-- .../src/network/service/liveness_checker.rs | 4 ++-- bin/dipper-service/src/registry/agreement.rs | 8 ++++---- dipper-pgregistry/src/indexing_agreement.rs | 8 ++++---- dipper-pgregistry/src/postgres.rs | 3 ++- 5 files changed, 15 insertions(+), 13 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 1ba0fa73..e0634898 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -1,6 +1,7 @@ //! Finishes the cancels dipper starts. An agreement dipper wants ended is marked //! `Cancelling` before its on-chain cancel goes out; this sweep re-sends the cancel while -//! the chain shows it live, and marks it `CanceledByRequester` once it can no longer be. +//! the chain shows it live, and marks it ended once it can no longer be: `CanceledByRequester`, +//! or `AbandonedByIndexer` for one dipper ended because its indexer stopped serving it. use dipper_core::time::now_secs; use thegraph_core::alloy::primitives::B256; @@ -134,7 +135,7 @@ async fn retry_cancel( note_check(registry, row, failure.as_ref(), Some(failure.is_none())).await; } -/// Mark the agreement `CanceledByRequester` once it can't go live again: this sweep's cancel +/// Mark the agreement ended by dipper once it can't go live again: this sweep's cancel /// ended it, or nobody accepted its offer before the deadline to. One ended otherwise is left /// to the chain listener for a while; one the indexer ended then becomes `CanceledByIndexer`. async fn confirm_if_over( diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index ebef68ce..bf3eedf2 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -1439,8 +1439,8 @@ mod tests { #[tokio::test] async fn leaves_a_failed_cancel_to_the_retry_and_still_reassesses() { - // Marked cancelling, the agreement is out of the checker's sight and its indexer out - // of selection, so it is replaced now and the cancel retry finishes the cancel. + // Marked cancelling, the agreement is out of the checker's sight, and its indexer out + // of selection until the cancel lands, so it is replaced now and the retry finishes it. let agreement = stale_agreement(); let calls = MockCalls::default(); let registry = MockRegistry::new(calls.clone(), agreement.clone()); diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 0b94535a..54ac69f0 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -744,15 +744,15 @@ pub enum Status { /// The liveness checker detected no indexing progress within the tolerance window. /// - /// Dipper canceled the agreement via `cancelIndexingAgreementByPayer` and will - /// trigger reassignment to find a replacement indexer. + /// Dipper cancelled the agreement on-chain, passing through `Cancelling` until the chain + /// confirmed it, and triggered reassignment to find a replacement indexer. /// /// This is a terminal state. AbandonedByIndexer, /// Dipper decided to end the agreement and is cancelling it on-chain, where it may - /// still be live. It becomes `CanceledByRequester`, announced as ended, only once - /// the chain confirms the end. + /// still be live. It becomes `CanceledByRequester`, or `AbandonedByIndexer` when its + /// indexer stopped serving it, announced as ended, only once the chain confirms the end. Cancelling, } diff --git a/dipper-pgregistry/src/indexing_agreement.rs b/dipper-pgregistry/src/indexing_agreement.rs index 4a3b24fe..241f8419 100644 --- a/dipper-pgregistry/src/indexing_agreement.rs +++ b/dipper-pgregistry/src/indexing_agreement.rs @@ -200,15 +200,15 @@ pub enum Status { /// The liveness checker detected no indexing progress within the tolerance window. /// - /// Dipper canceled the agreement via `cancelIndexingAgreementByPayer` and will - /// trigger reassignment to find a replacement indexer. + /// Dipper cancelled the agreement on-chain, passing through `Cancelling` until the chain + /// confirmed it, and triggered reassignment to find a replacement indexer. /// /// This is a terminal state. AbandonedByIndexer = 8, /// Dipper decided to end the agreement and is cancelling it on-chain, where it may - /// still be live. It becomes `CanceledByRequester`, announced as ended, only once - /// the chain confirms the end. + /// still be live. It becomes `CanceledByRequester`, or `AbandonedByIndexer` when its + /// indexer stopped serving it, announced as ended, only once the chain confirms the end. Cancelling = 9, /// A fallback for unknown status values. diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index a6564e53..2a553614 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -2283,7 +2283,8 @@ impl PgRegistry { /// Batched form of `update_status_from`: transitions all rows whose `id` /// is in `agreement_ids` and whose current status is in `allowed_from` to -/// `new_status`, in one statement. Returns the ids of the rows that +/// `new_status`, in one statement; a row noted abandoned that `new_status` would make +/// `CanceledByRequester` becomes `AbandonedByIndexer` instead. Returns the ids of the rows that /// actually flipped (matched the CAS guard) so callers can build per-id /// outcome maps. Empty input is a fast-path no-op. async fn batch_update_status_from( From b730812aa7e52edfaa308884f38fb18fe7b03063 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 5 Oct 2026 11:10:01 +0100 Subject: [PATCH 3/4] fix(cancel): say in the stuck-cancel alert why dipper is cancelling The alert for a cancel that keeps failing didn't say whether dipper was ending the agreement because its indexer stopped serving it, which an operator needs to decide what to do. It now carries that. --- bin/dipper-service/src/network/service/cancel_retry.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index e0634898..6d5932db 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -237,7 +237,7 @@ async fn note_check( { Ok(attempts) => { if let Some(err) = failure.filter(|_| failed_attempts > 0) { - log_failed_cancel(agreement, attempts, failed_attempts, err); + log_failed_cancel(row, attempts, failed_attempts, err); } } Err(err) => tracing::warn!( @@ -259,11 +259,12 @@ fn log_uncounted_failure(agreement: &IndexingAgreement, err: &ChainClientError) /// One ERROR as an agreement reaches the limit, for an operator to look into; a WARN for /// every other failed cancel. fn log_failed_cancel( - agreement: &IndexingAgreement, + row: &CancellingAgreement, attempts: u32, failed: u32, err: &ChainClientError, ) { + let agreement = &row.agreement; let reached_limit = attempts >= MAX_CANCEL_ATTEMPTS && attempts.saturating_sub(failed) < MAX_CANCEL_ATTEMPTS; if !reached_limit { @@ -280,6 +281,7 @@ fn log_failed_cancel( agreement_id = %agreement.id, indexer_id = %agreement.indexer.id, indexing_request_id = %agreement.indexing_request_id, + abandoned = row.abandoned, attempts, error = %err, "Cancelling an agreement keeps failing; it may still be live. Dipper now retries it hourly" From 16a59585f7002ff84d74721c341cc7f446bee987 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 5 Oct 2026 12:05:30 +0100 Subject: [PATCH 4/4] fix(liveness): leave a stale agreement that can't be cancelled alone A stale agreement with no stored terms hash was marked cancelling and replaced, but its cancel can never be sent, so both indexers would be paid. It is left active with an ERROR for an operator, as before. --- bin/dipper-service/src/cancel_dispatch.rs | 17 +++++++----- .../src/network/service/liveness_checker.rs | 26 +++++++++++++++++++ 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index e3be8cc1..17c0eed2 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -28,12 +28,7 @@ async fn cancel_agreement_on_chain( agreement: &IndexingAgreement, config: &IndexingAgreementConfig, ) -> LiveCancel { - let Some(version_hash) = agreement - .terms_version_hash - .as_deref() - .filter(|h| h.len() == 32) - .map(B256::from_slice) - else { + let Some(version_hash) = cancel_hash(agreement) else { return LiveCancel::CancelFailed(ChainClientError::MissingTermsVersionHash { agreement_id: agreement.id.to_string(), }); @@ -68,6 +63,16 @@ async fn cancel_agreement_on_chain( } } +/// The stored terms hash an on-chain cancel needs, or `None` when the agreement has no 32-byte +/// one, so no cancel can ever be sent for it. +pub fn cancel_hash(agreement: &IndexingAgreement) -> Option { + agreement + .terms_version_hash + .as_deref() + .filter(|h| h.len() == 32) + .map(B256::from_slice) +} + /// Why dipper is ending an agreement, which decides the status it ends in once the chain /// confirms dipper's cancel. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index bf3eedf2..e2dfc5d7 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -503,6 +503,16 @@ async fn cancel_and_reassess( W: WorkerQueue + Send + Sync, C: ChainClient + Send + Sync, { + // Without a cancel that can ever be sent, replacing it would pay 2 indexers until an + // operator ends it, so it stays as it is, active, for one to deal with. + if crate::cancel_dispatch::cancel_hash(agreement).is_none() { + tracing::error!( + agreement_id = %agreement.id, + "cannot cancel stale agreement: missing terms_version_hash; leaving active for operator action" + ); + return; + } + // 1. Start the cancel match crate::cancel_dispatch::start_cancel( registry, @@ -1457,6 +1467,22 @@ mod tests { ); } + #[tokio::test] + async fn leaves_a_stale_agreement_it_can_never_cancel_for_an_operator() { + // Replacing it would pay both indexers, since its cancel can never be sent. + let mut agreement = stale_agreement(); + agreement.terms_version_hash = None; + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + + end_stale(&agreement, ®istry, &chain).await; + + assert!(calls.abandoning.lock().unwrap().is_empty()); + assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert!(calls.reassessments.lock().unwrap().is_empty()); + } + #[tokio::test] async fn leaves_an_agreement_already_ending_alone() { let agreement = stale_agreement();