diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index bd3674db..cedd4326 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -92,13 +92,17 @@ impl CancelReason { } } -/// What [`start_cancel`] left an agreement as. +/// What [`start_cancel`] left an agreement as. Unless `Ended`, it stays `Cancelling` for the +/// cancel retry to finish. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CancelStarted { /// 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, + /// The chain shows nothing live, so nobody is being paid for it. + NotLive, + /// The chain couldn't be read, or its cancel failed or couldn't be confirmed, so it may + /// still be live and paid. + MayBeLive, } /// Start ending an agreement that may be live on-chain. It is marked `Cancelling` first, so an @@ -129,18 +133,18 @@ where } let tx_hash = match cancel_if_live(chain_client, agreement, config).await { LiveCancel::Ended(tx_hash) => tx_hash, - LiveCancel::NotLive { .. } => return Ok(CancelStarted::Cancelling), + LiveCancel::NotLive { .. } => return Ok(CancelStarted::NotLive), LiveCancel::ReadFailed(err) | LiveCancel::CancelFailed(err) => { tracing::warn!( agreement_id = %agreement.id, error = %err, - "On-chain cancel failed; the chain listener retries it" + "On-chain cancel failed; the cancel retry sends it again" ); - return Ok(CancelStarted::Cancelling); + return Ok(CancelStarted::MayBeLive); } LiveCancel::Unconfirmed { tx_hash, err } => { log_unconfirmed(agreement, tx_hash, &err); - return Ok(CancelStarted::Cancelling); + return Ok(CancelStarted::MayBeLive); } }; tracing::info!( @@ -150,13 +154,13 @@ where ); // An offer never accepted could still land and be accepted until its deadline. if agreement.status != IndexingAgreementStatus::AcceptedOnChain { - return Ok(CancelStarted::Cancelling); + return Ok(CancelStarted::NotLive); } Ok( if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { CancelStarted::Ended } else { - CancelStarted::Cancelling + CancelStarted::NotLive }, ) } diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 4b52b7c2..17810755 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -107,6 +107,7 @@ pub async fn main() -> anyhow::Result<()> { let chain_listener_agreement_conf = agreement_conf.clone(); let liveness_agreement_conf = agreement_conf.clone(); let escrow_reconciler_agreement_conf = agreement_conf.clone(); + let cancel_retry_agreement_conf = agreement_conf.clone(); // Canonical chain id and RecurringCollector address, read once and shared by the // admin signer, the gRPC proposal signer, and the on-chain chain client so their @@ -545,6 +546,15 @@ pub async fn main() -> anyhow::Result<()> { _ => None, }; + //- The cancel retry, always on: it alone finishes the cancels dipper starts + let (cancel_retry_handle, cancel_retry_service) = + network::service::cancel_retry::new(network::service::cancel_retry::Ctx { + registry: registry.clone(), + chain_client: chain_client.clone(), + agreement_conf: cancel_retry_agreement_conf, + worker_queue: worker_handle.queue().clone(), + }); + //- The liveness checker service (optional, enabled by config) // Detects indexers who silently stop indexing active AcceptedOnChain agreements let liveness_checker_handle = match conf.liveness_checker { @@ -683,6 +693,9 @@ pub async fn main() -> anyhow::Result<()> { None }; + let cancel_retry_task_handle = task_tree.spawn(cancel_retry_service); + tracing::debug!(task_id=%cancel_retry_task_handle.id(), "Cancel retry service started"); + // Spawn the escrow reconciler service if enabled let escrow_reconciler_stop_handle = if let Some((handle, service)) = escrow_reconciler_handle { let task_handle = task_tree.spawn(service); @@ -763,6 +776,9 @@ pub async fn main() -> anyhow::Result<()> { all_stopped &= stop_service("Chain listener", handle.stop()).await; } + // Stop the cancel retry before worker (it queues replacements) + all_stopped &= stop_service("Cancel retry", cancel_retry_handle.stop()).await; + // Stop escrow reconciler service before the DB pool closes if let Some(handle) = escrow_reconciler_stop_handle { all_stopped &= stop_service("Escrow reconciler", handle.stop()).await; diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 63cf5d50..3417b30e 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -1,10 +1,12 @@ -//! 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 ended once it can no longer be: `CanceledByRequester`, -//! or `AbandonedByIndexer` for one dipper ended because its indexer stopped serving it. +//! Finishes the cancels dipper starts, marked `Cancelling` before they go out: re-sends each +//! while the chain shows it live, then marks it ended, `AbandonedByIndexer` if its indexer +//! stopped serving it. Runs on its own, as nothing else finishes them. + +use std::{future::Future, sync::Arc, time::Duration}; use dipper_core::time::now_secs; use thegraph_core::alloy::primitives::B256; +use tokio::{sync::mpsc, time::MissedTickBehavior}; use crate::{ cancel_dispatch::{ @@ -12,7 +14,11 @@ use crate::{ }, chain_client::{ChainClient, ChainClientError}, config::IndexingAgreementConfig, - registry::{AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement}, + registry::{ + AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement, + IndexingRequestRegistry, + }, + worker::service::WorkerQueue, }; /// Failed cancels before dipper alerts an operator and retries the agreement only hourly, so a @@ -23,9 +29,16 @@ pub const MAX_CANCEL_ATTEMPTS: u32 = 10; /// below decides how many it gets through. const BATCH_SIZE: i64 = 50; -/// Time a sweep may take before leaving the rest to the next one: it holds up the chain -/// listener while it runs, and each cancel can wait up to 15 s to be mined. -const SWEEP_BUDGET: std::time::Duration = std::time::Duration::from_secs(30); +/// How often agreements still being cancelled get their cancel retried. +const SWEEP_INTERVAL: Duration = Duration::from_secs(300); + +/// Time a sweep may take before leaving the rest to the next one, as each cancel can wait up +/// to 15 s to be mined. +const SWEEP_BUDGET: Duration = Duration::from_secs(30); + +/// Time allowed to read an ended agreement's request, and to queue its replacement. +const DB_TIMEOUT: Duration = Duration::from_secs(30); +const QUEUE_TIMEOUT: Duration = Duration::from_secs(10); /// Minutes an agreement stays out of the retry after it is marked, so the cancel sent /// when it was marked can be mined first instead of being sent again. One moved back to @@ -36,6 +49,102 @@ const SETTLE_MINUTES: i32 = 2; /// record when and in which transaction it ended, before the retry marks it without them. const LISTENER_GRACE: time::Duration = time::Duration::HOUR; +/// Handle for stopping the cancel retry. +#[derive(Clone)] +pub struct Handle { + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Stop the cancel retry, cutting short a sweep in progress. + pub async fn stop(&self) { + if self.tx_stop.is_closed() { + return; + } + let _ = self.tx_stop.send(()).await; + self.tx_stop.closed().await; + } +} + +/// What the cancel retry needs. +pub struct Ctx { + pub registry: R, + pub chain_client: T, + pub agreement_conf: Arc, + /// Queues the replacement of an agreement ended because its indexer stopped serving it. + pub worker_queue: W, +} + +/// Create the cancel retry. Returns a handle plus a future to spawn, which sweeps at once and +/// then every [`SWEEP_INTERVAL`]. +pub fn new(ctx: Ctx) -> (Handle, impl Future>) +where + R: AgreementRegistry + IndexingRequestRegistry + Send + Sync, + T: ChainClient + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let (tx_stop, mut rx_stop) = mpsc::channel(1); + let Ctx { + registry, + chain_client, + agreement_conf, + worker_queue, + } = ctx; + let service = async move { + tracing::info!( + interval_secs = SWEEP_INTERVAL.as_secs(), + "cancel retry service started" + ); + let mut timer = tokio::time::interval(SWEEP_INTERVAL); + timer.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = rx_stop.recv() => break, + _ = timer.tick() => {}, + } + tokio::select! { + _ = rx_stop.recv() => break, + () = async { + retry_cancelling_agreements(®istry, &chain_client, &agreement_conf).await; + replace_ended_abandoned(®istry, &worker_queue).await; + } => {}, + } + } + tracing::debug!("cancel retry service stopped"); + Ok(()) + }; + (Handle { tx_stop }, service) +} + +/// Queue the replacement of each agreement whose indexer stopped serving it once it has ended, +/// whatever ended it: this retry, the chain listener or the indexer. +async fn replace_ended_abandoned(registry: &R, worker_queue: &W) +where + R: AgreementRegistry + IndexingRequestRegistry + Sync, + W: WorkerQueue + Sync, +{ + let ended = match registry + .get_ended_agreements_awaiting_replacement(BATCH_SIZE) + .await + { + Ok(ended) => ended, + Err(err) => { + tracing::warn!(error = %err, "Failed to list ended agreements awaiting replacement"); + return; + } + }; + for agreement in &ended { + super::liveness_checker::replace_abandoned( + agreement, + registry, + worker_queue, + DB_TIMEOUT, + QUEUE_TIMEOUT, + ) + .await; + } +} + /// Retry the cancel of agreements still `Cancelling`. The chain's own latest block time /// decides when an offer that was never accepted no longer can be, so a subgraph that has /// fallen behind doesn't hold that up. @@ -288,21 +397,17 @@ fn log_failed_cancel( ); } -/// How many of an agreement's cancel attempts a failure uses up. A cancel the contract -/// refused, before sending or once mined, that mined without ending the agreement, or that -/// endpoints answered had no receipt counts, and one that can never be sent uses them all. An -/// unreachable chain, including one whose receipt checks all failed, is retried freely. +/// How many of an agreement's cancel attempts a failed cancel uses up. The chain was read just +/// before, so any failure counts, a refusal to send (gas over the cap, signer out of funds) +/// included, except a cancel whose receipt checks all failed, which may have mined unseen. fn failed_attempts(err: &ChainClientError) -> u32 { match err { - ChainClientError::CancelNotConfirmed { .. } - | ChainClientError::TxReverted { .. } - | ChainClientError::TxDropped { - receipt_checked: true, + ChainClientError::TxDropped { + receipt_checked: false, .. - } - | ChainClientError::ContractRevert { .. } => 1, + } => 0, ChainClientError::MissingTermsVersionHash { .. } => MAX_CANCEL_ATTEMPTS, - _ => 0, + _ => 1, } } @@ -314,7 +419,7 @@ mod tests { }; use async_trait::async_trait; - use dipper_core::ids::IndexingAgreementId; + use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; use dipper_rpc::indexer::indexer_client::sol::RecurringCollectionAgreement; use thegraph_core::alloy::primitives::Address; @@ -323,6 +428,7 @@ mod tests { cancel_dispatch::tests::agreement, chain_client::AgreementOnChain, registry::{IndexingAgreementStatus, StubAgreementRegistry}, + worker::service::JobPriority, }; const DEADLINE: u64 = 1_000; @@ -340,6 +446,9 @@ mod tests { /// The chain listener marks it ended before the retry's own mark lands. listener_ended_it: bool, writes: Mutex>, + /// Ended agreements awaiting replacement, until noted as replaced. + awaiting_replacement: Vec, + replacements_noted: Mutex>, } #[async_trait] @@ -352,6 +461,25 @@ mod tests { ) -> crate::registry::Result> { Ok(self.cancelling.clone()) } + async fn get_ended_agreements_awaiting_replacement( + &self, + _batch_size: i64, + ) -> crate::registry::Result> { + let noted = self.replacements_noted.lock().unwrap(); + Ok(self + .awaiting_replacement + .iter() + .filter(|agreement| !noted.contains(&agreement.id)) + .cloned() + .collect()) + } + async fn mark_replacement_queued( + &self, + id: &IndexingAgreementId, + ) -> crate::registry::Result<()> { + self.replacements_noted.lock().unwrap().push(*id); + Ok(()) + } async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -408,7 +536,8 @@ mod tests { struct MockChain { live: AtomicBool, read_fails: bool, - send_fails: bool, + /// What every send fails with, when set. + send_error: Option ChainClientError>, mined_cancel_reverts: bool, never_mines: bool, receipt_unreadable: bool, @@ -439,8 +568,8 @@ mod tests { _version_hash: B256, _options: u16, ) -> Result, ChainClientError> { - if self.send_fails { - return Err(ChainClientError::RpcError(anyhow::anyhow!("rpc down"))); + if let Some(send_error) = self.send_error { + return Err(send_error()); } if self.reverts_before_sending { return Err(ChainClientError::ContractRevert { @@ -531,6 +660,137 @@ mod tests { retry_cancelling_agreements(registry, chain, &config).await; } + #[async_trait] + impl IndexingRequestRegistry for MockRegistry { + async fn set_indexing_target_candidates( + &self, + _requested_by: Address, + _deployment_id: thegraph_core::DeploymentId, + _deployment_chain_id: u64, + _num_candidates: usize, + ) -> crate::registry::Result { + unimplemented!() + } + async fn get_all_indexing_requests( + &self, + ) -> crate::registry::Result> { + unimplemented!() + } + async fn get_indexing_request_by_id( + &self, + id: &IndexingRequestId, + ) -> crate::registry::Result> { + let agreement = ended_abandoned(); + Ok(Some(crate::registry::IndexingRequest { + id: *id, + created_at: time::OffsetDateTime::now_utc(), + updated_at: time::OffsetDateTime::now_utc(), + status: crate::registry::IndexingRequestStatus::Open, + requested_by: Address::ZERO, + deployment_id: agreement.terms.metadata.subgraph_deployment_id, + deployment_chain_id: agreement.terms.metadata.chain_id, + num_candidates: 3, + })) + } + async fn get_indexing_requests_by_deployment_id( + &self, + _deployment_id: &thegraph_core::DeploymentId, + ) -> crate::registry::Result> { + unimplemented!() + } + async fn get_open_indexing_requests_for_reassessment( + &self, + _min_age_seconds: i64, + _batch_size: i64, + ) -> crate::registry::Result> { + unimplemented!() + } + } + + /// Records the requests it is asked to reassess. + #[derive(Default)] + struct MockQueue(Arc>>); + + #[async_trait] + impl WorkerQueue for MockQueue { + async fn send_indexing_agreement_proposal( + &self, + _candidate_url: url::Url, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _deployment_id: thegraph_core::DeploymentId, + _deployment_chain_id: u64, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn reassess_indexing_request( + &self, + indexing_request_id: IndexingRequestId, + _deployment_id: thegraph_core::DeploymentId, + _deployment_chain_id: u64, + _num_candidates: usize, + _priority: JobPriority, + ) -> anyhow::Result { + self.0.lock().unwrap().push(indexing_request_id); + Ok(dipper_pgmq::JobId::default()) + } + async fn submit_offer( + &self, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _indexer_url: url::Url, + _deployment_id: thegraph_core::DeploymentId, + _deployment_chain_id: u64, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + } + + fn ended_abandoned() -> IndexingAgreement { + agreement( + IndexingAgreementStatus::AbandonedByIndexer, + Some(vec![7u8; 32]), + ) + } + + /// Nothing else finishes a cancel, so the retry can't wait on the chain listener, which + /// config can turn off. + #[tokio::test(start_paused = true)] + async fn sweeps_on_its_own_and_replaces_an_abandoned_agreement_once_it_has_ended() { + let ended = ended_abandoned(); + let registry = MockRegistry { + awaiting_replacement: vec![ended.clone()], + ..registry_with_one(true) + }; + let chain = Arc::new(live_chain()); + let queue = MockQueue::default(); + let reassessed = Arc::clone(&queue.0); + let (handle, service) = new(Ctx { + registry, + chain_client: Arc::clone(&chain), + agreement_conf: Arc::new(IndexingAgreementConfig::for_tests()), + worker_queue: queue, + }); + let service = tokio::spawn(service); + + tokio::time::sleep(SWEEP_INTERVAL + Duration::from_secs(1)).await; + handle.stop().await; + + service.await.unwrap().unwrap(); + assert_eq!( + chain.clock_reads.load(Ordering::SeqCst), + 2, + "1 sweep at start, 1 later" + ); + assert_eq!( + *reassessed.lock().unwrap(), + vec![ended.indexing_request_id], + "queued once, then noted as replaced" + ); + } + #[tokio::test] async fn waits_for_the_next_sweep_when_the_chain_time_cannot_be_read() { let registry = registry_with_one(true); @@ -804,23 +1064,42 @@ mod tests { #[tokio::test] async fn an_unreachable_chain_neither_sends_nor_counts_an_attempt() { - for chain in [ - MockChain { - read_fails: true, - ..live_chain() - }, - MockChain { - send_fails: true, - ..live_chain() + let registry = registry_with_one(true); + let chain = MockChain { + read_fails: true, + ..live_chain() + }; + + retry(®istry, &chain, 0).await; + + assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 0); + assert_eq!(registry.attempts.load(Ordering::SeqCst), 0); + assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn counts_a_cancel_that_could_not_be_sent() { + // The chain was just read, so a send that fails every time (gas over the cap, signer + // out of funds) is no outage, and only counting it reaches the alert. + let refusals: [fn() -> ChainClientError; 2] = [ + || ChainClientError::SubmitFailed(anyhow::anyhow!("Gas price exceeds maximum")), + || { + ChainClientError::RpcError(anyhow::anyhow!( + "Gas estimation failed: insufficient funds" + )) }, - ] { + ]; + for err in refusals { let registry = registry_with_one(true); + let chain = MockChain { + send_error: Some(err), + ..live_chain() + }; retry(®istry, &chain, 0).await; assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 0); - assert_eq!(registry.attempts.load(Ordering::SeqCst), 0); - assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + assert_eq!(registry.attempts.load(Ordering::SeqCst), 1); } } diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 15c00d57..aa8b761e 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -82,10 +82,6 @@ const SWEEP_BATCH_SIZE: i64 = 1000; /// crash-recovery; the steady-state fan-out fires from finalize on a /// fresh accept, so per-poll execution is wasted DB work. const SWEEP_POLLS: u64 = 60; -/// How often agreements still being cancelled get their cancel retried, whichever rate the -/// listener polls at: just under the slow poll interval, so every slow poll retries. -const CANCEL_RETRY_INTERVAL: Duration = Duration::from_secs(290); - /// Handle for controlling the chain listener service lifecycle #[derive(Clone)] pub struct Handle { @@ -226,7 +222,6 @@ where // Starts at SWEEP_POLLS so the first poll runs the sweep, // recovering any pre-startup orphans. let mut polls_since_sweep: u64 = SWEEP_POLLS; - let mut last_cancel_retry: Option = None; // Pause the event sweeps after a Kafka send failure, backing off from one // poll interval up to the idle interval, so a hung broker cannot stall the // poll loop on every iteration. @@ -292,18 +287,6 @@ where tokio::time::sleep(backoff).await; } - // Ahead of the drain, which ends the poll early while the subgraph is down: - // finishing a cancel needs only the chain. - if last_cancel_retry.is_none_or(|at| at.elapsed() >= CANCEL_RETRY_INTERVAL) { - last_cancel_retry = Some(Instant::now()); - super::cancel_retry::retry_cancelling_agreements( - ®istry, - &chain_client, - &agreement_conf, - ) - .await; - } - let outcome = match drain_once( &mut cursor, &mut last_persisted_timestamp, @@ -1156,7 +1139,7 @@ where /// /// Called from the Created -> AcceptedOnChain and Expired -> AcceptedOnChain /// transitions. Each replaced agreement is marked `Cancelling` and its on-chain cancel -/// sent once; the cancel retry in this listener finishes any that don't end at once. +/// sent once; the cancel retry finishes any that don't end at once. /// A pending row is deleted once its agreement is marked; a failed mark keeps it for retry. async fn execute_pending_cancellations( agreement_id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index e2dfc5d7..a95e2682 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -43,7 +43,7 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use url::Url; use crate::{ - cancel_dispatch::CancelReason, + cancel_dispatch::{CancelReason, CancelStarted}, chain_client::ChainClient, config::LivenessCheckerConfig, network::provider::NetworkProviderService, @@ -481,15 +481,10 @@ async fn record_progress( } } -/// 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. +/// End a stale agreement and queue a reassessment to replace it once it can't be paid. It is +/// marked `Cancelling`, and as awaiting replacement, before any cancel is sent, so the cancel +/// retry finishes one that fails here and queues its replacement once it has ended. #[allow(clippy::too_many_arguments)] -#[expect( - clippy::cognitive_complexity, - clippy::too_many_lines, - reason = "predates this lint; fix when next touched" -)] async fn cancel_and_reassess( agreement: &IndexingAgreement, registry: &R, @@ -512,8 +507,57 @@ async fn cancel_and_reassess( ); return; } + let Some(started) = + start_abandoned_cancel(agreement, registry, chain_client, agreement_conf).await + else { + return; + }; + forget_replaced_agreement(agreement, registry).await; + if started == CancelStarted::MayBeLive { + tracing::info!( + agreement_id = %agreement.id, + "Stale agreement may still be paid; the cancel retry replaces it once it has ended" + ); + return; + } + replace_abandoned(agreement, registry, worker_queue, db_timeout, queue_timeout).await; +} + +/// Queue the replacement of an agreement whose indexer stopped serving it, and note it queued +/// so it isn't queued again. +pub(crate) async fn replace_abandoned( + agreement: &IndexingAgreement, + registry: &R, + worker_queue: &W, + db_timeout: Duration, + queue_timeout: Duration, +) where + R: AgreementRegistry + IndexingRequestRegistry + Sync, + W: WorkerQueue + Sync, +{ + if !queue_replacement(agreement, registry, worker_queue, db_timeout, queue_timeout).await { + return; + } + if let Err(err) = registry.mark_replacement_queued(&agreement.id).await { + tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "Failed to note an abandoned agreement's replacement queued; it may be queued again" + ); + } +} - // 1. Start the cancel +/// Mark a stale agreement abandoned and send its cancel; `None`, logged, when it isn't marked. +async fn start_abandoned_cancel( + agreement: &IndexingAgreement, + registry: &R, + chain_client: &C, + agreement_conf: &crate::config::IndexingAgreementConfig, +) -> Option +where + R: AgreementRegistry + Sync, + C: ChainClient, +{ match crate::cancel_dispatch::start_cancel( registry, chain_client, @@ -523,18 +567,21 @@ async fn cancel_and_reassess( ) .await { - Ok(started) => tracing::info!( - agreement_id = %agreement.id, - ?started, - reason = "indexer_stale", - "Cancelling stale agreement" - ), + Ok(started) => { + tracing::info!( + agreement_id = %agreement.id, + ?started, + reason = "indexer_stale", + "Cancelling stale agreement" + ); + Some(started) + } Err(crate::registry::Error::NoRecordsUpdated) => { tracing::debug!( agreement_id = %agreement.id, "Stale agreement already ended or being cancelled" ); - return; + None } Err(err) => { tracing::error!( @@ -542,12 +589,17 @@ async fn cancel_and_reassess( error = %err, "failed to mark stale agreement cancelling, will retry next cycle" ); - return; + None } } +} - // Clean up pending cancellations: if this abandoned agreement was a - // replacement, the old agreement it was replacing should stay active. +/// Drop the pending cancellations an abandoned agreement holds as a replacement, so the +/// agreement it was to replace stays active. +async fn forget_replaced_agreement( + agreement: &IndexingAgreement, + registry: &R, +) { if let Err(err) = registry .delete_pending_cancellations_by_new_agreement(agreement.id) .await @@ -558,41 +610,26 @@ async fn cancel_and_reassess( "failed to clean up pending cancellations for abandoned agreement" ); } +} - // 2. Fetch the indexing request for num_candidates - let request = match tokio::time::timeout( - db_timeout, - registry.get_indexing_request_by_id(&agreement.indexing_request_id), - ) - .await - { - Ok(Ok(Some(r))) => r, - Ok(Ok(None)) => { - tracing::warn!( - agreement_id = %agreement.id, - indexing_request_id = %agreement.indexing_request_id, - "indexing request not found for abandoned agreement" - ); - return; - } - Ok(Err(err)) => { - tracing::warn!( - agreement_id = %agreement.id, - error = %err, - "failed to fetch indexing request for abandoned agreement" - ); - return; - } - Err(_) => { - tracing::warn!( - agreement_id = %agreement.id, - "timeout fetching indexing request for abandoned agreement" - ); - return; - } +/// Queue a reassessment of an abandoned agreement's request, which replaces its indexer. True +/// once nothing is left to do: queued, or its request is gone. +async fn queue_replacement( + agreement: &IndexingAgreement, + registry: &R, + worker_queue: &W, + db_timeout: Duration, + queue_timeout: Duration, +) -> bool +where + R: IndexingRequestRegistry + Sync, + W: WorkerQueue + Sync, +{ + let request = match abandoned_request(agreement, registry, db_timeout).await { + Ok(Some(request)) => request, + Ok(None) => return true, + Err(()) => return false, }; - - // 3. Queue reassessment let push_result = tokio::time::timeout( queue_timeout, worker_queue.reassess_indexing_request( @@ -613,6 +650,7 @@ async fn cancel_and_reassess( indexing_request_id = %agreement.indexing_request_id, "queued reassessment for abandoned agreement" ); + true } Ok(Err(err)) => { tracing::warn!( @@ -620,12 +658,54 @@ async fn cancel_and_reassess( error = %err, "failed to queue reassessment for abandoned agreement" ); + false } Err(_) => { tracing::warn!( agreement_id = %agreement.id, "timeout queuing reassessment for abandoned agreement" ); + false + } + } +} + +/// The request an abandoned agreement served, `None` when it is gone; `Err`, logged, when it +/// can't be read. +async fn abandoned_request( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) -> Result, ()> { + match tokio::time::timeout( + db_timeout, + registry.get_indexing_request_by_id(&agreement.indexing_request_id), + ) + .await + { + Ok(Ok(Some(r))) => Ok(Some(r)), + Ok(Ok(None)) => { + tracing::warn!( + agreement_id = %agreement.id, + indexing_request_id = %agreement.indexing_request_id, + "indexing request not found for abandoned agreement" + ); + Ok(None) + } + Ok(Err(err)) => { + tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "failed to fetch indexing request for abandoned agreement" + ); + Err(()) + } + Err(_) => { + tracing::warn!( + agreement_id = %agreement.id, + "timeout fetching indexing request for abandoned agreement" + ); + Err(()) } } } @@ -873,6 +953,8 @@ mod tests { /// Ids passed to `record_cancel_audit` -- the signal the handler drives /// the terminated event (the chain_listener sweep emits from this audit). cancel_audits: Arc>>, + /// Ids noted as having their replacement queued. + replacements_noted: Arc>>, } struct MockRegistry { @@ -925,6 +1007,11 @@ mod tests { Ok(()) } + async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()> { + self.calls.replacements_noted.lock().unwrap().push(*id); + Ok(()) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -1445,12 +1532,17 @@ mod tests { calls.reassessments.lock().unwrap().as_slice(), &[agreement.indexing_request_id] ); + assert_eq!( + calls.replacements_noted.lock().unwrap().as_slice(), + &[agreement.id], + "so the cancel retry doesn't queue it again" + ); } #[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 until the cancel lands, so it is replaced now and the retry finishes it. + async fn holds_back_the_replacement_while_a_failed_cancel_leaves_it_paid() { + // Replacing it now would pay both indexers until the retry lands the cancel, which then + // queues the replacement. let agreement = stale_agreement(); let calls = MockCalls::default(); let registry = MockRegistry::new(calls.clone(), agreement.clone()); @@ -1461,10 +1553,29 @@ mod tests { 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!(calls.reassessments.lock().unwrap().is_empty()); + assert!(calls.replacements_noted.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn replaces_at_once_a_stale_agreement_the_chain_shows_ended() { + let agreement = stale_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::rpc_error(calls.clone()); + chain.live.store(false, std::sync::atomic::Ordering::SeqCst); + + end_stale(&agreement, ®istry, &chain).await; + + assert!(calls.chain_cancels.lock().unwrap().is_empty()); assert_eq!( calls.reassessments.lock().unwrap().as_slice(), &[agreement.indexing_request_id] ); + assert_eq!( + calls.replacements_noted.lock().unwrap().as_slice(), + &[agreement.id] + ); } #[tokio::test] diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index e49d8c4c..86ac2a7f 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -690,6 +690,27 @@ impl AgreementRegistry for RegistryProvider { .collect()) } + async fn get_ended_agreements_awaiting_replacement( + &self, + batch_size: i64, + ) -> RegistryResult> { + Ok(self + .inner + .get_ended_agreements_awaiting_replacement(batch_size) + .await? + .into_iter() + .map(IndexingAgreement::try_from) + .filter_map(filter_map_with_logging) + .collect()) + } + + async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()> { + self.inner + .mark_replacement_queued(id) + .await + .map_err(Into::into) + } + async fn update_agreement_sync_progress( &self, id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 95da2b73..10cc02f8 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -497,6 +497,16 @@ pub trait AgreementRegistry { batch_size: i64, ) -> RegistryResult>; + /// Agreements whose indexer stopped serving them that have ended, longest ended first, + /// whose replacement is yet to be queued. + async fn get_ended_agreements_awaiting_replacement( + &self, + batch_size: i64, + ) -> RegistryResult>; + + /// Note that an agreement's replacement has been queued. + async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()>; + /// Update the sync progress for an agreement. /// /// Called when the liveness checker observes the block height has changed diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index d9092a6d..cc3c717f 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -225,6 +225,17 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("get_agreements_pending_chain_cancel") } + async fn get_ended_agreements_awaiting_replacement( + &self, + _batch_size: i64, + ) -> Result> { + unimplemented!("get_ended_agreements_awaiting_replacement") + } + + async fn mark_replacement_queued(&self, _id: &IndexingAgreementId) -> Result<()> { + unimplemented!("mark_replacement_queued") + } + async fn update_agreement_sync_progress( &self, _id: &IndexingAgreementId, @@ -533,6 +544,17 @@ impl AgreementRegistry for T { StubAgreementRegistry::get_agreements_pending_chain_cancel(self, batch_size).await } + async fn get_ended_agreements_awaiting_replacement( + &self, + batch_size: i64, + ) -> Result> { + StubAgreementRegistry::get_ended_agreements_awaiting_replacement(self, batch_size).await + } + + async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> Result<()> { + StubAgreementRegistry::mark_replacement_queued(self, id).await + } + async fn update_agreement_sync_progress( &self, id: &IndexingAgreementId, 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 e07f5ba5..584e08d3 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -1827,8 +1827,8 @@ mod lifecycle_event_tests { } #[tokio::test] - async fn an_agreement_whose_cancel_fails_is_left_cancelling_for_the_listener() { - // The chain listener retries the cancel of every cancelling agreement until + async fn an_agreement_whose_cancel_fails_is_left_cancelling_for_the_retry() { + // The cancel retry re-sends the cancel of every cancelling agreement until // the chain shows it ended, so nothing is queued here. for status in [ IndexingAgreementStatus::Created, @@ -1910,7 +1910,7 @@ enum Unpaired { } /// Take an old agreement out of the target group. One that may be live on-chain is cancelled -/// there too, which the chain listener retries until it ends. +/// there too, which the cancel retry re-sends until it ends. async fn cancel_unpaired( ctx: &Ctx, agreement: &crate::registry::IndexingAgreement, @@ -1944,7 +1944,10 @@ where Ok(crate::cancel_dispatch::CancelStarted::Ended) => { Unpaired::Moved("CANCELED_BY_REQUESTER") } - Ok(crate::cancel_dispatch::CancelStarted::Cancelling) => Unpaired::Moved("CANCELLING"), + Ok( + crate::cancel_dispatch::CancelStarted::NotLive + | crate::cancel_dispatch::CancelStarted::MayBeLive, + ) => Unpaired::Moved("CANCELLING"), Err(err) => unmarked(agreement, &err), } } diff --git a/bin/dipper-service/src/worker/handlers/submit_offer.rs b/bin/dipper-service/src/worker/handlers/submit_offer.rs index f5ee5088..83867064 100644 --- a/bin/dipper-service/src/worker/handlers/submit_offer.rs +++ b/bin/dipper-service/src/worker/handlers/submit_offer.rs @@ -283,8 +283,8 @@ fn dipper_cancelled(status: IndexingAgreementStatus) -> bool { } /// The agreement, if it was cancelled after this job's status check. After a failed read -/// the job still finishes: retrying would send the offer again, and the chain listener's -/// cancel retry withdraws the offer of an agreement left cancelling. +/// the job still finishes: retrying would send the offer again, and the cancel retry +/// withdraws the offer of an agreement left cancelling. async fn cancelled_meanwhile( registry: &R, agreement_id: &IndexingAgreementId, diff --git a/dipper-pgregistry/migrations/20261008000000_add_replacement_pending.sql b/dipper-pgregistry/migrations/20261008000000_add_replacement_pending.sql new file mode 100644 index 00000000..bf58fbb4 --- /dev/null +++ b/dipper-pgregistry/migrations/20261008000000_add_replacement_pending.sql @@ -0,0 +1,8 @@ +-- replacement_pending: the agreement's indexer stopped serving it and dipper has yet to queue a +-- reassessment to replace it, which waits until the agreement can no longer be paid. +ALTER TABLE dipper_reg_indexing_agreements + ADD COLUMN replacement_pending BOOLEAN NOT NULL DEFAULT false; + +CREATE INDEX idx_indexing_agreements_replacement_pending + ON dipper_reg_indexing_agreements (updated_at) + WHERE replacement_pending; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index d1bc8194..d2bce6bd 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -989,6 +989,7 @@ impl PgRegistry { SET status = $1, abandoned = true, + replacement_pending = true, updated_at = timezone('UTC', now()) WHERE id = $2 AND status = $3 "#, @@ -1858,6 +1859,56 @@ impl PgRegistry { .map_err(Into::into) } + /// Agreements whose indexer stopped serving them that have ended, longest ended first, + /// whose replacement is yet to be queued. + pub async fn get_ended_agreements_awaiting_replacement( + &self, + batch_size: i64, + ) -> Result, Error> { + sqlx::query_as( + r#" + SELECT + 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 + FROM dipper_reg_indexing_agreements + WHERE replacement_pending AND status <> $1 + ORDER BY updated_at ASC + LIMIT $2 + "#, + ) + .bind(IndexingAgreementStatus::Cancelling) + .bind(batch_size) + .fetch_all(&self.pool) + .await + .map_err(Into::into) + } + + /// Note that an agreement's replacement has been queued. + pub async fn mark_replacement_queued( + &self, + agreement_id: &IndexingAgreementId, + ) -> Result<(), Error> { + sqlx::query( + "UPDATE dipper_reg_indexing_agreements SET replacement_pending = false WHERE id = $1", + ) + .bind(agreement_id) + .execute(&self.pool) + .await?; + Ok(()) + } + /// Update the sync progress for an agreement. /// /// Called when the liveness checker observes the block height has changed diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index e5bd0f3a..88b993ef 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3884,3 +3884,53 @@ async fn an_offer_mined_after_the_cancel_began_keeps_its_hash() { .expect_err("an ended agreement doesn't take the hash"); assert!(matches!(err, Error::NoRecordsUpdated)); } + +/// An agreement whose indexer stopped serving it is replaced only once it can't be paid, and +/// only once, whatever ended it. +#[tokio::test] +async fn an_abandoned_agreement_awaits_replacement_once_ended_until_queued() { + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + let id = fixture_agreement(0xaa); + sqlx::query("UPDATE dipper_reg_indexing_agreements SET status = $1 WHERE id = $2") + .bind(IndexingAgreementStatus::AcceptedOnChain) + .bind(id) + .execute(&db) + .await + .expect("accept the agreement"); + let registry = PgRegistry::new(db); + registry + .mark_indexing_agreement_as_abandoning(&id) + .await + .expect("mark abandoning"); + let awaiting = async || { + registry + .get_ended_agreements_awaiting_replacement(10) + .await + .expect("awaiting query") + .into_iter() + .map(|agreement| agreement.id) + .collect::>() + }; + assert!( + awaiting().await.is_empty(), + "still cancelling, so maybe paid" + ); + + registry + .mark_indexing_agreement_as_canceled_by_requester(&id) + .await + .expect("end it"); + assert_eq!(awaiting().await, vec![id]); + + registry + .mark_replacement_queued(&id) + .await + .expect("note it queued"); + assert!(awaiting().await.is_empty()); +}