diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 049dc105..17c0eed2 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,17 +23,12 @@ 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, ) -> 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,10 +63,39 @@ pub 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)] +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 +108,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 +153,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 +161,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 +190,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 e56a6c07..ec9e0e63 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -1,12 +1,15 @@ //! 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; 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}, @@ -132,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( @@ -162,7 +165,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 @@ -229,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!( @@ -251,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 { @@ -272,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" @@ -497,6 +507,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..e2dfc5d7 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,47 @@ 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; - } + // 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; } - // 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 +559,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 +570,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 +592,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 +610,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 +864,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 +877,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 +914,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 +1059,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 +1071,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 +1079,7 @@ mod tests { Self { calls, result: Err(ChainClientError::RpcError(anyhow::anyhow!("network error"))), + live: true.into(), } } } @@ -1172,12 +1112,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 +1140,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 +1387,115 @@ 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 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()); - 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_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(); 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..54ac69f0 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, }) } } @@ -742,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/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/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 1d7ebe1c..2a553614 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 no cancel is sent with it, so the retry takes it on its next sweep rather @@ -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 // ========================================================================= @@ -2295,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( @@ -2308,20 +2297,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 // =============================================================================