From 15ed860aa90a6b954919b86dcc6e9eab91709358 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:18:24 +0100 Subject: [PATCH 1/7] fix(cancel): count cancels that can never be sent toward the stuck alert A cancel the chain client refused to send, such as gas over the cap or a signer out of funds, counted as an outage, so it was retried forever without raising agreement_cancel_stuck. The chain is read just before each send, so every failed send now counts except an unchecked receipt. --- .../src/network/service/cancel_retry.rs | 66 ++++++++++++------- 1 file changed, 41 insertions(+), 25 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 63cf5d50..0a8e3661 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -288,21 +288,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, } } @@ -408,7 +404,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 +436,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 { @@ -804,23 +801,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); } } From ef53812bcedb0a4a32afb90dc080f04a1f34afef Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:20:27 +0100 Subject: [PATCH 2/7] fix(cancel): run the cancel retry as its own service Agreements marked Cancelling were only ever finished by a sweep inside the chain listener, which config can turn off, leaving them cancelling for good. The sweep now runs on its own every 5 minutes whatever the listener does, and a stop request cuts a slow sweep short. --- bin/dipper-service/src/cancel_dispatch.rs | 2 +- bin/dipper-service/src/main.rs | 14 +++ .../src/network/service/cancel_retry.rs | 114 ++++++++++++++++-- .../src/network/service/chain_listener.rs | 19 +-- .../handlers/reassess_indexing_request.rs | 6 +- .../src/worker/handlers/submit_offer.rs | 4 +- 6 files changed, 128 insertions(+), 31 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index bd3674db..84638db5 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -134,7 +134,7 @@ where 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); } diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 4b52b7c2..47ff1c70 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,14 @@ 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, + }); + //- 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 +692,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 +775,8 @@ pub async fn main() -> anyhow::Result<()> { all_stopped &= stop_service("Chain listener", handle.stop()).await; } + 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 0a8e3661..cce7016e 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::{ @@ -23,9 +25,12 @@ 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); /// 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 +41,66 @@ 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, +} + +/// 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 + Send + Sync, + T: ChainClient + Send + Sync, +{ + let (tx_stop, mut rx_stop) = mpsc::channel(1); + let Ctx { + registry, + chain_client, + agreement_conf, + } = 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, + () = retry_cancelling_agreements(®istry, &chain_client, &agreement_conf) => {}, + } + } + tracing::debug!("cancel retry service stopped"); + Ok(()) + }; + (Handle { tx_stop }, service) +} + /// 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. @@ -528,6 +593,41 @@ mod tests { retry_cancelling_agreements(registry, chain, &config).await; } + /// Counts the sweeps that ask for agreements being cancelled, and has none. + struct CountsSweeps(Arc); + + #[async_trait] + impl StubAgreementRegistry for CountsSweeps { + async fn get_cancelling_agreements( + &self, + _batch_size: i64, + _max_attempts: u32, + _min_age_minutes: i32, + ) -> crate::registry::Result> { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(Vec::new()) + } + } + + /// 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_at_start_and_then_every_interval_until_stopped() { + let sweeps = Arc::new(AtomicU32::new(0)); + let (handle, service) = new(Ctx { + registry: CountsSweeps(Arc::clone(&sweeps)), + chain_client: MockChain::default(), + agreement_conf: Arc::new(IndexingAgreementConfig::for_tests()), + }); + let service = tokio::spawn(service); + + tokio::time::sleep(SWEEP_INTERVAL + Duration::from_secs(1)).await; + handle.stop().await; + + service.await.unwrap().unwrap(); + assert_eq!(sweeps.load(Ordering::SeqCst), 2); + } + #[tokio::test] async fn waits_for_the_next_sweep_when_the_chain_time_cannot_be_read() { let registry = registry_with_one(true); 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/worker/handlers/reassess_indexing_request.rs b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs index e07f5ba5..9f16c526 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, 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, From 653efc6b5230819d4a8fc79d4cd8b99a8f1e24f7 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:25:07 +0100 Subject: [PATCH 3/7] fix(liveness): replace a stale agreement only once it can't be paid When the cancel of an agreement whose indexer stopped serving it failed, a replacement was queued anyway, so dipper paid both indexers until a retry landed the cancel. The replacement now waits until the chain shows the old one ended, and the cancel retry queues it once it ends it. --- bin/dipper-service/src/cancel_dispatch.rs | 20 +- bin/dipper-service/src/main.rs | 2 + .../src/network/service/cancel_retry.rs | 218 +++++++++++++++--- .../src/network/service/liveness_checker.rs | 174 +++++++++----- .../handlers/reassess_indexing_request.rs | 5 +- 5 files changed, 320 insertions(+), 99 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 84638db5..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 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 47ff1c70..17810755 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -552,6 +552,7 @@ pub async fn main() -> anyhow::Result<()> { 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) @@ -775,6 +776,7 @@ 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 diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index cce7016e..70fe4394 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -14,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 @@ -32,6 +36,10 @@ const SWEEP_INTERVAL: Duration = Duration::from_secs(300); /// 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 /// cancelling had none sent, so it doesn't wait. @@ -59,24 +67,28 @@ impl Handle { } /// What the cancel retry needs. -pub struct Ctx { +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>) +pub fn new(ctx: Ctx) -> (Handle, impl Future>) where - R: AgreementRegistry + Send + Sync, + 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!( @@ -90,9 +102,19 @@ where _ = rx_stop.recv() => break, _ = timer.tick() => {}, } - tokio::select! { + let abandoned = tokio::select! { _ = rx_stop.recv() => break, - () = retry_cancelling_agreements(®istry, &chain_client, &agreement_conf) => {}, + ended = retry_cancelling_agreements(®istry, &chain_client, &agreement_conf) => ended, + }; + for agreement in &abandoned { + super::liveness_checker::queue_replacement( + agreement, + ®istry, + &worker_queue, + DB_TIMEOUT, + QUEUE_TIMEOUT, + ) + .await; } } tracing::debug!("cancel retry service stopped"); @@ -101,14 +123,15 @@ where (Handle { tx_stop }, service) } -/// 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. +/// Retry the cancel of agreements still `Cancelling`, returning those it ended that were +/// abandoned by their indexer, to be replaced. The chain's own latest block time decides when +/// an offer that was never accepted no longer can be, so a lagging subgraph doesn't hold it up. pub async fn retry_cancelling_agreements( registry: &R, chain_client: &T, config: &IndexingAgreementConfig, -) where +) -> Vec +where R: AgreementRegistry + Sync, T: ChainClient, { @@ -119,16 +142,17 @@ pub async fn retry_cancelling_agreements( Ok(cancelling) => cancelling, Err(err) => { tracing::warn!(error = %err, "Failed to list agreements still being cancelled"); - return; + return Vec::new(); } }; if cancelling.is_empty() { - return; + return Vec::new(); } let Some(chain_now) = chain_time(chain_client).await else { - return; + return Vec::new(); }; let started = std::time::Instant::now(); + let mut abandoned = Vec::new(); for (done, row) in cancelling.iter().enumerate() { if started.elapsed() >= SWEEP_BUDGET { tracing::info!( @@ -137,8 +161,11 @@ pub async fn retry_cancelling_agreements( ); break; } - retry_cancel(registry, chain_client, config, row, chain_now).await; + if retry_cancel(registry, chain_client, config, row, chain_now).await && row.abandoned { + abandoned.push(row.agreement.clone()); + } } + abandoned } async fn chain_time(chain_client: &T) -> Option { @@ -154,13 +181,15 @@ async fn chain_time(chain_client: &T) -> Option { } } +/// Retry 1 agreement's cancel; true once it is marked ended. async fn retry_cancel( registry: &R, chain_client: &T, config: &IndexingAgreementConfig, row: &CancellingAgreement, chain_now: u64, -) where +) -> bool +where R: AgreementRegistry + Sync, T: ChainClient, { @@ -174,7 +203,8 @@ async fn retry_cancel( "Failed to read a cancelling agreement on-chain, will retry" ); // Unread, it may still be live, so it can't be confirmed ended. - return note_check(registry, row, None, None).await; + note_check(registry, row, None, None).await; + return false; } LiveCancel::NotLive { by_indexer } => (None, by_indexer, None), LiveCancel::Ended(tx_hash) => { @@ -188,16 +218,18 @@ async fn retry_cancel( LiveCancel::CancelFailed(err) => (None, false, Some(err)), LiveCancel::Unconfirmed { tx_hash, err } => { log_unconfirmed(&row.agreement, tx_hash, &err); - return note_check(registry, row, None, None).await; + note_check(registry, row, None, None).await; + return false; } }; if failure.is_none() && confirm_if_over(registry, config, row, tx_hash, by_indexer, chain_now).await { - return; + return true; } // A cancel that failed found it live; otherwise it is over, or withdrawn until its deadline. note_check(registry, row, failure.as_ref(), Some(failure.is_none())).await; + false } /// Mark the agreement ended by dipper once it can't go live again: this sweep's cancel @@ -375,7 +407,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; @@ -384,6 +416,7 @@ mod tests { cancel_dispatch::tests::agreement, chain_client::AgreementOnChain, registry::{IndexingAgreementStatus, StubAgreementRegistry}, + worker::service::JobPriority, }; const DEADLINE: u64 = 1_000; @@ -593,31 +626,114 @@ mod tests { retry_cancelling_agreements(registry, chain, &config).await; } - /// Counts the sweeps that ask for agreements being cancelled, and has none. - struct CountsSweeps(Arc); - #[async_trait] - impl StubAgreementRegistry for CountsSweeps { - async fn get_cancelling_agreements( + 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 = &self.cancelling[0].agreement; + 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, - _max_attempts: u32, - _min_age_minutes: i32, - ) -> crate::registry::Result> { - self.0.fetch_add(1, Ordering::SeqCst); - Ok(Vec::new()) + ) -> 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 registry_with_one_abandoned() -> MockRegistry { + let mut registry = registry_with_one(true); + registry.cancelling[0].abandoned = true; + registry + } + /// 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_at_start_and_then_every_interval_until_stopped() { - let sweeps = Arc::new(AtomicU32::new(0)); + async fn sweeps_on_its_own_and_queues_the_replacement_of_an_abandoned_agreement_it_ends() { + let registry = registry_with_one_abandoned(); + let request = registry.cancelling[0].agreement.indexing_request_id; + let chain = Arc::new(live_chain()); + let queue = MockQueue::default(); + let reassessed = Arc::clone(&queue.0); let (handle, service) = new(Ctx { - registry: CountsSweeps(Arc::clone(&sweeps)), - chain_client: MockChain::default(), + registry, + chain_client: Arc::clone(&chain), agreement_conf: Arc::new(IndexingAgreementConfig::for_tests()), + worker_queue: queue, }); let service = tokio::spawn(service); @@ -625,7 +741,41 @@ mod tests { handle.stop().await; service.await.unwrap().unwrap(); - assert_eq!(sweeps.load(Ordering::SeqCst), 2); + assert_eq!( + chain.clock_reads.load(Ordering::SeqCst), + 2, + "1 sweep at start, 1 later" + ); + assert_eq!(*reassessed.lock().unwrap(), vec![request]); + } + + /// The liveness checker leaves replacing an abandoned agreement to whoever ends it, when + /// its own cancel failed and the agreement may still be paid. + #[tokio::test] + async fn reports_only_the_abandoned_agreements_it_ends() { + let config = IndexingAgreementConfig::for_tests(); + for (registry, chain, replaced) in [ + (registry_with_one_abandoned(), live_chain(), true), + (registry_with_one(true), live_chain(), false), + ( + registry_with_one_abandoned(), + MockChain { + read_fails: true, + ..live_chain() + }, + false, + ), + ] { + let ended = retry_cancelling_agreements(®istry, &chain, &config).await; + + let ids: Vec<_> = ended.iter().map(|agreement| agreement.id).collect(); + let expected = if replaced { + vec![registry.cancelling[0].agreement.id] + } else { + Vec::new() + }; + assert_eq!(ids, expected); + } } #[tokio::test] diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index e2dfc5d7..8aeb1e33 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` before any cancel is sent, so the cancel retry finishes one that fails +/// here, queueing the replacement then; it ends `AbandonedByIndexer`. #[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,33 @@ 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; it is replaced once the cancel retry ends it" + ); + return; + } + queue_replacement(agreement, registry, worker_queue, db_timeout, queue_timeout).await; +} - // 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 +543,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 +565,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 +586,22 @@ 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. +pub(crate) async fn queue_replacement( + agreement: &IndexingAgreement, + registry: &R, + worker_queue: &W, + db_timeout: Duration, + queue_timeout: Duration, +) where + R: IndexingRequestRegistry + Sync, + W: WorkerQueue + Sync, +{ + let Some(request) = abandoned_request(agreement, registry, db_timeout).await else { + return; }; - - // 3. Queue reassessment let push_result = tokio::time::timeout( queue_timeout, worker_queue.reassess_indexing_request( @@ -630,6 +639,45 @@ async fn cancel_and_reassess( } } +/// The request an abandoned agreement served; `None`, logged, when it can't be read. +async fn abandoned_request( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) -> Option { + match tokio::time::timeout( + db_timeout, + registry.get_indexing_request_by_id(&agreement.indexing_request_id), + ) + .await + { + Ok(Ok(Some(r))) => 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" + ); + None + } + Ok(Err(err)) => { + tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "failed to fetch indexing request for abandoned agreement" + ); + None + } + Err(_) => { + tracing::warn!( + agreement_id = %agreement.id, + "timeout fetching indexing request for abandoned agreement" + ); + None + } + } +} + /// The action the liveness checker should take for a single agreement after evaluating its sync state. #[derive(Debug, PartialEq)] pub(crate) enum LivenessAction { @@ -1448,9 +1496,9 @@ 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 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,6 +1509,20 @@ 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()); + } + + #[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] 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 9f16c526..584e08d3 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -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), } } From aa13450333bf6758cc988d9e59c52a62ea162513 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:28:33 +0100 Subject: [PATCH 4/7] fix(listener): cancel a replaced agreement dipper can't mark cancelling When a replacement was accepted, an old agreement in a status dipper can't mark Cancelling, such as Unresponsive with its offer still open, got no on-chain cancel, so the indexer could accept it. It now gets one if the chain shows it live, and its pending row stays for retry if that fails. --- .../src/network/service/chain_listener.rs | 144 +++++++++++++++++- 1 file changed, 143 insertions(+), 1 deletion(-) diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index aa8b761e..aa76a007 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -50,7 +50,8 @@ use tokio::{ use super::chain_events::{AgreementState, AgreementStateSnapshot, ChainEventSource, Cursor}; use crate::{ - chain_client::ChainClient, + cancel_dispatch::LiveCancel, + chain_client::{ChainClient, ChainClientError}, config::ChainListenerConfig, registry::{ AgreementRegistry, CancelKind, IndexingAgreement, IndexingAgreementStatus, @@ -1224,6 +1225,13 @@ where config, ) .await; + // One already cancelling is the cancel retry's; any other status that can't be marked + // gets no retry, so its cancel goes out now. + if matches!(started, Err(crate::registry::Error::NoRecordsUpdated)) + && old_agreement.status != IndexingAgreementStatus::Cancelling + { + return Ok(cancel_unmarked(new_agreement_id, &old_agreement, chain_client, config).await); + } Ok(note_replaced_cancel( new_agreement_id, &old_agreement, @@ -1231,6 +1239,60 @@ where )) } +/// Cancel a replaced agreement that couldn't be marked `Cancelling` if the chain shows it live, +/// such as an `Unresponsive` one whose offer is still open. False, keeping its pending +/// cancellation for the next sweep, when the chain can't settle it yet. +async fn cancel_unmarked( + new_agreement_id: &IndexingAgreementId, + old_agreement: &IndexingAgreement, + chain_client: &T, + config: &crate::config::IndexingAgreementConfig, +) -> bool { + let outcome = crate::cancel_dispatch::cancel_if_live(chain_client, old_agreement, config).await; + log_unmarked_cancel(new_agreement_id, old_agreement, &outcome); + match outcome { + LiveCancel::NotLive { .. } | LiveCancel::Ended(_) => true, + // No cancel can ever be sent for it, so trying again would only repeat the error. + LiveCancel::CancelFailed(ChainClientError::MissingTermsVersionHash { .. }) => true, + LiveCancel::ReadFailed(_) + | LiveCancel::CancelFailed(_) + | LiveCancel::Unconfirmed { .. } => false, + } +} + +fn log_unmarked_cancel( + new_agreement_id: &IndexingAgreementId, + old_agreement: &IndexingAgreement, + outcome: &LiveCancel, +) { + let old_agreement_id = old_agreement.id; + match outcome { + LiveCancel::NotLive { .. } => {} + LiveCancel::Ended(tx_hash) => tracing::info!( + %new_agreement_id, + %old_agreement_id, + old_status = %old_agreement.status, + tx_hash = ?tx_hash, + reason = "replacement_accepted", + "Cancelled a replaced agreement still live on-chain" + ), + LiveCancel::CancelFailed(err @ ChainClientError::MissingTermsVersionHash { .. }) => { + tracing::error!( + %old_agreement_id, + error = %err, + "Replaced agreement is live on-chain but can never be cancelled" + ); + } + LiveCancel::ReadFailed(err) + | LiveCancel::CancelFailed(err) + | LiveCancel::Unconfirmed { err, .. } => tracing::warn!( + %old_agreement_id, + error = %err, + "Failed to cancel a replaced agreement that may be live, retaining pending row" + ), + } +} + /// Whether an agreement marked `Expired` is live on-chain after all, accepted unseen by a /// lagging listener; `None`, logged, when the chain can't be read. One that really expired /// stays `Expired`. @@ -2190,6 +2252,19 @@ mod tests { )), )); } + // As the database does, refuse an agreement it couldn't be cancelling from. + let markable = state.agreements.get(id).is_none_or(|agreement| { + matches!( + agreement.status, + IndexingAgreementStatus::Created + | IndexingAgreementStatus::AcceptedOnChain + | IndexingAgreementStatus::Rejected + | IndexingAgreementStatus::Expired + ) + }); + if !markable { + return Err(crate::registry::Error::NoRecordsUpdated); + } state.marked_cancelling.push(*id); Ok(()) } @@ -3675,6 +3750,73 @@ mod tests { assert!(registry.was_pending_cancellation_deleted(&new_id, &old_id)); } + /// Pending cancellations for an old agreement in `status`, run against `chain_client`; + /// returns the result and whether its pending row was deleted. + async fn replace_agreement_in( + status: IndexingAgreementStatus, + chain_client: &MockChainClient, + ) -> (anyhow::Result<()>, IndexingAgreementId, bool) { + let registry = MockRegistry::new(); + let new_id = IndexingAgreementId::from_bytes(rand::random()); + let old_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(new_id, IndexingAgreementStatus::AcceptedOnChain); + registry.add_agreement(old_id, status); + registry.add_pending_cancellation(new_id, old_id); + + let result = execute_pending_cancellations( + &new_id, + ®istry, + chain_client, + test_agreement_conf().as_ref(), + ) + .await; + + assert!(!registry.was_marked_cancelling(&old_id)); + let deleted = registry.was_pending_cancellation_deleted(&new_id, &old_id); + (result, old_id, deleted) + } + + #[tokio::test] + async fn test_pending_cancellations_cancel_an_unmarkable_agreement_still_live() { + // An `Unresponsive` agreement can't be marked cancelling, yet its offer may still be + // open on-chain for the indexer to accept alongside its replacement. + let chain_client = live_chain(); + + let (result, old_id, deleted) = + replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; + + assert!(result.is_ok(), "got {result:?}"); + assert!(chain_client.was_on_chain_cancel_attempted(&old_id)); + assert!(deleted); + } + + #[tokio::test] + async fn test_pending_cancellations_retain_an_unmarkable_agreement_whose_cancel_fails() { + // No cancel retry tracks it, so the next sweep tries again. + let chain_client = MockChainClient { + fail_cancels: true, + ..live_chain() + }; + + let (result, _, deleted) = + replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; + + assert!(result.is_err()); + assert!(!deleted); + } + + #[tokio::test] + async fn test_pending_cancellations_leave_a_cancelling_agreement_to_the_cancel_retry() { + let chain_client = live_chain(); + + let (result, old_id, deleted) = + replace_agreement_in(IndexingAgreementStatus::Cancelling, &chain_client).await; + + assert!(result.is_ok(), "got {result:?}"); + assert!(!chain_client.was_on_chain_cancel_attempted(&old_id)); + assert!(deleted); + } + #[tokio::test] async fn test_sweep_clears_already_canceled_pending_rows() { // The sweep is the long-lived recovery path. If the on-chain cancel From 76d4c643597eb51a42906daaac756475d9ae78ef Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:36:21 +0100 Subject: [PATCH 5/7] fix(liveness): replace an abandoned agreement however it ends A stale agreement whose cancel failed was replaced only if the cancel retry ended it, so one the chain listener ended first was never replaced. The agreement now records that a replacement is due, and the cancel retry queues it once the agreement has ended, whatever ended it, and only once. --- .../src/network/service/cancel_retry.rs | 157 ++++++++++-------- .../src/network/service/liveness_checker.rs | 79 +++++++-- bin/dipper-service/src/registry.rs | 21 +++ bin/dipper-service/src/registry/agreement.rs | 10 ++ .../src/registry/agreement_stub.rs | 22 +++ ...20261008000000_add_replacement_pending.sql | 8 + dipper-pgregistry/src/postgres.rs | 51 ++++++ .../tests/it_registry_postgres.rs | 50 ++++++ 8 files changed, 311 insertions(+), 87 deletions(-) create mode 100644 dipper-pgregistry/migrations/20261008000000_add_replacement_pending.sql diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 70fe4394..3417b30e 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -102,19 +102,12 @@ where _ = rx_stop.recv() => break, _ = timer.tick() => {}, } - let abandoned = tokio::select! { + tokio::select! { _ = rx_stop.recv() => break, - ended = retry_cancelling_agreements(®istry, &chain_client, &agreement_conf) => ended, - }; - for agreement in &abandoned { - super::liveness_checker::queue_replacement( - agreement, - ®istry, - &worker_queue, - DB_TIMEOUT, - QUEUE_TIMEOUT, - ) - .await; + () = async { + retry_cancelling_agreements(®istry, &chain_client, &agreement_conf).await; + replace_ended_abandoned(®istry, &worker_queue).await; + } => {}, } } tracing::debug!("cancel retry service stopped"); @@ -123,15 +116,43 @@ where (Handle { tx_stop }, service) } -/// Retry the cancel of agreements still `Cancelling`, returning those it ended that were -/// abandoned by their indexer, to be replaced. The chain's own latest block time decides when -/// an offer that was never accepted no longer can be, so a lagging subgraph doesn't hold it up. +/// 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. pub async fn retry_cancelling_agreements( registry: &R, chain_client: &T, config: &IndexingAgreementConfig, -) -> Vec -where +) where R: AgreementRegistry + Sync, T: ChainClient, { @@ -142,17 +163,16 @@ where Ok(cancelling) => cancelling, Err(err) => { tracing::warn!(error = %err, "Failed to list agreements still being cancelled"); - return Vec::new(); + return; } }; if cancelling.is_empty() { - return Vec::new(); + return; } let Some(chain_now) = chain_time(chain_client).await else { - return Vec::new(); + return; }; let started = std::time::Instant::now(); - let mut abandoned = Vec::new(); for (done, row) in cancelling.iter().enumerate() { if started.elapsed() >= SWEEP_BUDGET { tracing::info!( @@ -161,11 +181,8 @@ where ); break; } - if retry_cancel(registry, chain_client, config, row, chain_now).await && row.abandoned { - abandoned.push(row.agreement.clone()); - } + retry_cancel(registry, chain_client, config, row, chain_now).await; } - abandoned } async fn chain_time(chain_client: &T) -> Option { @@ -181,15 +198,13 @@ async fn chain_time(chain_client: &T) -> Option { } } -/// Retry 1 agreement's cancel; true once it is marked ended. async fn retry_cancel( registry: &R, chain_client: &T, config: &IndexingAgreementConfig, row: &CancellingAgreement, chain_now: u64, -) -> bool -where +) where R: AgreementRegistry + Sync, T: ChainClient, { @@ -203,8 +218,7 @@ where "Failed to read a cancelling agreement on-chain, will retry" ); // Unread, it may still be live, so it can't be confirmed ended. - note_check(registry, row, None, None).await; - return false; + return note_check(registry, row, None, None).await; } LiveCancel::NotLive { by_indexer } => (None, by_indexer, None), LiveCancel::Ended(tx_hash) => { @@ -218,18 +232,16 @@ where LiveCancel::CancelFailed(err) => (None, false, Some(err)), LiveCancel::Unconfirmed { tx_hash, err } => { log_unconfirmed(&row.agreement, tx_hash, &err); - note_check(registry, row, None, None).await; - return false; + return note_check(registry, row, None, None).await; } }; if failure.is_none() && confirm_if_over(registry, config, row, tx_hash, by_indexer, chain_now).await { - return true; + return; } // A cancel that failed found it live; otherwise it is over, or withdrawn until its deadline. note_check(registry, row, failure.as_ref(), Some(failure.is_none())).await; - false } /// Mark the agreement ended by dipper once it can't go live again: this sweep's cancel @@ -434,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] @@ -446,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, @@ -646,7 +680,7 @@ mod tests { &self, id: &IndexingRequestId, ) -> crate::registry::Result> { - let agreement = &self.cancelling[0].agreement; + let agreement = ended_abandoned(); Ok(Some(crate::registry::IndexingRequest { id: *id, created_at: time::OffsetDateTime::now_utc(), @@ -714,18 +748,22 @@ mod tests { } } - fn registry_with_one_abandoned() -> MockRegistry { - let mut registry = registry_with_one(true); - registry.cancelling[0].abandoned = true; - registry + 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_queues_the_replacement_of_an_abandoned_agreement_it_ends() { - let registry = registry_with_one_abandoned(); - let request = registry.cancelling[0].agreement.indexing_request_id; + 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); @@ -746,36 +784,11 @@ mod tests { 2, "1 sweep at start, 1 later" ); - assert_eq!(*reassessed.lock().unwrap(), vec![request]); - } - - /// The liveness checker leaves replacing an abandoned agreement to whoever ends it, when - /// its own cancel failed and the agreement may still be paid. - #[tokio::test] - async fn reports_only_the_abandoned_agreements_it_ends() { - let config = IndexingAgreementConfig::for_tests(); - for (registry, chain, replaced) in [ - (registry_with_one_abandoned(), live_chain(), true), - (registry_with_one(true), live_chain(), false), - ( - registry_with_one_abandoned(), - MockChain { - read_fails: true, - ..live_chain() - }, - false, - ), - ] { - let ended = retry_cancelling_agreements(®istry, &chain, &config).await; - - let ids: Vec<_> = ended.iter().map(|agreement| agreement.id).collect(); - let expected = if replaced { - vec![registry.cancelling[0].agreement.id] - } else { - Vec::new() - }; - assert_eq!(ids, expected); - } + assert_eq!( + *reassessed.lock().unwrap(), + vec![ended.indexing_request_id], + "queued once, then noted as replaced" + ); } #[tokio::test] diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 8aeb1e33..a95e2682 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -482,8 +482,8 @@ async fn record_progress( } /// End a stale agreement and queue a reassessment to replace it once it can't be paid. It is -/// marked `Cancelling` before any cancel is sent, so the cancel retry finishes one that fails -/// here, queueing the replacement then; it ends `AbandonedByIndexer`. +/// 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)] async fn cancel_and_reassess( agreement: &IndexingAgreement, @@ -516,11 +516,35 @@ async fn cancel_and_reassess( if started == CancelStarted::MayBeLive { tracing::info!( agreement_id = %agreement.id, - "Stale agreement may still be paid; it is replaced once the cancel retry ends it" + "Stale agreement may still be paid; the cancel retry replaces it once it has ended" ); return; } - queue_replacement(agreement, registry, worker_queue, db_timeout, queue_timeout).await; + 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" + ); + } } /// Mark a stale agreement abandoned and send its cancel; `None`, logged, when it isn't marked. @@ -588,19 +612,23 @@ async fn forget_replaced_agreement( } } -/// Queue a reassessment of an abandoned agreement's request, which replaces its indexer. -pub(crate) async fn queue_replacement( +/// 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, -) where +) -> bool +where R: IndexingRequestRegistry + Sync, W: WorkerQueue + Sync, { - let Some(request) = abandoned_request(agreement, registry, db_timeout).await else { - return; + let request = match abandoned_request(agreement, registry, db_timeout).await { + Ok(Some(request)) => request, + Ok(None) => return true, + Err(()) => return false, }; let push_result = tokio::time::timeout( queue_timeout, @@ -622,6 +650,7 @@ pub(crate) async fn queue_replacement( indexing_request_id = %agreement.indexing_request_id, "queued reassessment for abandoned agreement" ); + true } Ok(Err(err)) => { tracing::warn!( @@ -629,36 +658,39 @@ pub(crate) async fn queue_replacement( 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`, logged, when it can't be read. +/// 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, -) -> Option { +) -> Result, ()> { match tokio::time::timeout( db_timeout, registry.get_indexing_request_by_id(&agreement.indexing_request_id), ) .await { - Ok(Ok(Some(r))) => Some(r), + 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" ); - None + Ok(None) } Ok(Err(err)) => { tracing::warn!( @@ -666,14 +698,14 @@ async fn abandoned_request( error = %err, "failed to fetch indexing request for abandoned agreement" ); - None + Err(()) } Err(_) => { tracing::warn!( agreement_id = %agreement.id, "timeout fetching indexing request for abandoned agreement" ); - None + Err(()) } } } @@ -921,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 { @@ -973,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, @@ -1493,6 +1532,11 @@ 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] @@ -1510,6 +1554,7 @@ mod tests { 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] @@ -1527,6 +1572,10 @@ mod tests { 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/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()); +} From 1e15d05fe016c6ec0b146b9cb7cac24528db0619 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 21:37:55 +0100 Subject: [PATCH 6/7] fix(listener): alert instead of resending a cancel the contract refuses A replaced agreement that can't be marked cancelling had its cancel resent every sweep even when the contract refused it or it mined without ending the agreement, costing gas each time with no alert. Those now raise agreement_cancel_stuck once and stop; a failed read or send still retries. --- .../src/network/service/chain_listener.rs | 58 ++++++++++++++----- 1 file changed, 45 insertions(+), 13 deletions(-) diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index aa76a007..4a5242a5 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -1252,14 +1252,24 @@ async fn cancel_unmarked( log_unmarked_cancel(new_agreement_id, old_agreement, &outcome); match outcome { LiveCancel::NotLive { .. } | LiveCancel::Ended(_) => true, - // No cancel can ever be sent for it, so trying again would only repeat the error. - LiveCancel::CancelFailed(ChainClientError::MissingTermsVersionHash { .. }) => true, - LiveCancel::ReadFailed(_) - | LiveCancel::CancelFailed(_) - | LiveCancel::Unconfirmed { .. } => false, + LiveCancel::CancelFailed(err) => is_refused(&err), + LiveCancel::ReadFailed(_) | LiveCancel::Unconfirmed { .. } => false, } } +/// Whether a cancel failed in a way sending it again won't change: no terms hash to send it +/// with, the contract refused it, or it mined without ending the agreement. Resending one that +/// mined costs gas each time. +fn is_refused(err: &ChainClientError) -> bool { + matches!( + err, + ChainClientError::MissingTermsVersionHash { .. } + | ChainClientError::ContractRevert { .. } + | ChainClientError::TxReverted { .. } + | ChainClientError::CancelNotConfirmed { .. } + ) +} + fn log_unmarked_cancel( new_agreement_id: &IndexingAgreementId, old_agreement: &IndexingAgreement, @@ -1276,13 +1286,14 @@ fn log_unmarked_cancel( reason = "replacement_accepted", "Cancelled a replaced agreement still live on-chain" ), - LiveCancel::CancelFailed(err @ ChainClientError::MissingTermsVersionHash { .. }) => { - tracing::error!( - %old_agreement_id, - error = %err, - "Replaced agreement is live on-chain but can never be cancelled" - ); - } + LiveCancel::CancelFailed(err) if is_refused(err) => tracing::error!( + event = "agreement_cancel_stuck", + agreement_id = %old_agreement_id, + indexer_id = %old_agreement.indexer.id, + indexing_request_id = %old_agreement.indexing_request_id, + error = %err, + "A replaced agreement live on-chain refuses its cancel; an operator must end it" + ), LiveCancel::ReadFailed(err) | LiveCancel::CancelFailed(err) | LiveCancel::Unconfirmed { err, .. } => tracing::warn!( @@ -2612,6 +2623,8 @@ mod tests { fail_cancels: bool, /// When set, every agreement reads as live until a cancel is sent for it. live_until_cancelled: bool, + /// When set, a cancel mines but leaves the agreement live. + cancels_do_nothing: bool, } /// A chain on which every agreement is live until a cancel is sent for it. @@ -2708,7 +2721,9 @@ mod tests { // Cancel dispatch always reads back after a mined cancel; reporting // not-active here means "cancel confirmed", which these tests expect. Ok(crate::chain_client::AgreementOnChain::live_if( - self.live_until_cancelled && !self.cancels.lock().unwrap().contains(agreement_id), + self.live_until_cancelled + && (self.cancels_do_nothing + || !self.cancels.lock().unwrap().contains(agreement_id)), )) } } @@ -3805,6 +3820,23 @@ mod tests { assert!(!deleted); } + #[tokio::test] + async fn test_pending_cancellations_give_up_on_an_unmarkable_agreement_refusing_its_cancel() { + // Each resend of a cancel that mines without ending it costs gas, so it goes to an + // operator instead. + let chain_client = MockChainClient { + cancels_do_nothing: true, + ..live_chain() + }; + + let (result, old_id, deleted) = + replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; + + assert!(result.is_ok(), "got {result:?}"); + assert!(chain_client.was_on_chain_cancel_attempted(&old_id)); + assert!(deleted); + } + #[tokio::test] async fn test_pending_cancellations_leave_a_cancelling_agreement_to_the_cancel_retry() { let chain_client = live_chain(); From 89b8c59f2507deb993818e070d99144d8090227d Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 13:38:51 +0100 Subject: [PATCH 7/7] revert(listener): drop the cancel for replaced unmarkable agreements That change cancelled a replaced Unresponsive agreement on-chain, assuming its offer could still be open. An agreement is only marked Unresponsive when the gRPC proposal fails, and dipper puts an offer on-chain only after the indexer accepts that proposal, so no such offer exists. --- .../src/network/service/chain_listener.rs | 178 +----------------- 1 file changed, 2 insertions(+), 176 deletions(-) diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 4a5242a5..aa8b761e 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -50,8 +50,7 @@ use tokio::{ use super::chain_events::{AgreementState, AgreementStateSnapshot, ChainEventSource, Cursor}; use crate::{ - cancel_dispatch::LiveCancel, - chain_client::{ChainClient, ChainClientError}, + chain_client::ChainClient, config::ChainListenerConfig, registry::{ AgreementRegistry, CancelKind, IndexingAgreement, IndexingAgreementStatus, @@ -1225,13 +1224,6 @@ where config, ) .await; - // One already cancelling is the cancel retry's; any other status that can't be marked - // gets no retry, so its cancel goes out now. - if matches!(started, Err(crate::registry::Error::NoRecordsUpdated)) - && old_agreement.status != IndexingAgreementStatus::Cancelling - { - return Ok(cancel_unmarked(new_agreement_id, &old_agreement, chain_client, config).await); - } Ok(note_replaced_cancel( new_agreement_id, &old_agreement, @@ -1239,71 +1231,6 @@ where )) } -/// Cancel a replaced agreement that couldn't be marked `Cancelling` if the chain shows it live, -/// such as an `Unresponsive` one whose offer is still open. False, keeping its pending -/// cancellation for the next sweep, when the chain can't settle it yet. -async fn cancel_unmarked( - new_agreement_id: &IndexingAgreementId, - old_agreement: &IndexingAgreement, - chain_client: &T, - config: &crate::config::IndexingAgreementConfig, -) -> bool { - let outcome = crate::cancel_dispatch::cancel_if_live(chain_client, old_agreement, config).await; - log_unmarked_cancel(new_agreement_id, old_agreement, &outcome); - match outcome { - LiveCancel::NotLive { .. } | LiveCancel::Ended(_) => true, - LiveCancel::CancelFailed(err) => is_refused(&err), - LiveCancel::ReadFailed(_) | LiveCancel::Unconfirmed { .. } => false, - } -} - -/// Whether a cancel failed in a way sending it again won't change: no terms hash to send it -/// with, the contract refused it, or it mined without ending the agreement. Resending one that -/// mined costs gas each time. -fn is_refused(err: &ChainClientError) -> bool { - matches!( - err, - ChainClientError::MissingTermsVersionHash { .. } - | ChainClientError::ContractRevert { .. } - | ChainClientError::TxReverted { .. } - | ChainClientError::CancelNotConfirmed { .. } - ) -} - -fn log_unmarked_cancel( - new_agreement_id: &IndexingAgreementId, - old_agreement: &IndexingAgreement, - outcome: &LiveCancel, -) { - let old_agreement_id = old_agreement.id; - match outcome { - LiveCancel::NotLive { .. } => {} - LiveCancel::Ended(tx_hash) => tracing::info!( - %new_agreement_id, - %old_agreement_id, - old_status = %old_agreement.status, - tx_hash = ?tx_hash, - reason = "replacement_accepted", - "Cancelled a replaced agreement still live on-chain" - ), - LiveCancel::CancelFailed(err) if is_refused(err) => tracing::error!( - event = "agreement_cancel_stuck", - agreement_id = %old_agreement_id, - indexer_id = %old_agreement.indexer.id, - indexing_request_id = %old_agreement.indexing_request_id, - error = %err, - "A replaced agreement live on-chain refuses its cancel; an operator must end it" - ), - LiveCancel::ReadFailed(err) - | LiveCancel::CancelFailed(err) - | LiveCancel::Unconfirmed { err, .. } => tracing::warn!( - %old_agreement_id, - error = %err, - "Failed to cancel a replaced agreement that may be live, retaining pending row" - ), - } -} - /// Whether an agreement marked `Expired` is live on-chain after all, accepted unseen by a /// lagging listener; `None`, logged, when the chain can't be read. One that really expired /// stays `Expired`. @@ -2263,19 +2190,6 @@ mod tests { )), )); } - // As the database does, refuse an agreement it couldn't be cancelling from. - let markable = state.agreements.get(id).is_none_or(|agreement| { - matches!( - agreement.status, - IndexingAgreementStatus::Created - | IndexingAgreementStatus::AcceptedOnChain - | IndexingAgreementStatus::Rejected - | IndexingAgreementStatus::Expired - ) - }); - if !markable { - return Err(crate::registry::Error::NoRecordsUpdated); - } state.marked_cancelling.push(*id); Ok(()) } @@ -2623,8 +2537,6 @@ mod tests { fail_cancels: bool, /// When set, every agreement reads as live until a cancel is sent for it. live_until_cancelled: bool, - /// When set, a cancel mines but leaves the agreement live. - cancels_do_nothing: bool, } /// A chain on which every agreement is live until a cancel is sent for it. @@ -2721,9 +2633,7 @@ mod tests { // Cancel dispatch always reads back after a mined cancel; reporting // not-active here means "cancel confirmed", which these tests expect. Ok(crate::chain_client::AgreementOnChain::live_if( - self.live_until_cancelled - && (self.cancels_do_nothing - || !self.cancels.lock().unwrap().contains(agreement_id)), + self.live_until_cancelled && !self.cancels.lock().unwrap().contains(agreement_id), )) } } @@ -3765,90 +3675,6 @@ mod tests { assert!(registry.was_pending_cancellation_deleted(&new_id, &old_id)); } - /// Pending cancellations for an old agreement in `status`, run against `chain_client`; - /// returns the result and whether its pending row was deleted. - async fn replace_agreement_in( - status: IndexingAgreementStatus, - chain_client: &MockChainClient, - ) -> (anyhow::Result<()>, IndexingAgreementId, bool) { - let registry = MockRegistry::new(); - let new_id = IndexingAgreementId::from_bytes(rand::random()); - let old_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(new_id, IndexingAgreementStatus::AcceptedOnChain); - registry.add_agreement(old_id, status); - registry.add_pending_cancellation(new_id, old_id); - - let result = execute_pending_cancellations( - &new_id, - ®istry, - chain_client, - test_agreement_conf().as_ref(), - ) - .await; - - assert!(!registry.was_marked_cancelling(&old_id)); - let deleted = registry.was_pending_cancellation_deleted(&new_id, &old_id); - (result, old_id, deleted) - } - - #[tokio::test] - async fn test_pending_cancellations_cancel_an_unmarkable_agreement_still_live() { - // An `Unresponsive` agreement can't be marked cancelling, yet its offer may still be - // open on-chain for the indexer to accept alongside its replacement. - let chain_client = live_chain(); - - let (result, old_id, deleted) = - replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; - - assert!(result.is_ok(), "got {result:?}"); - assert!(chain_client.was_on_chain_cancel_attempted(&old_id)); - assert!(deleted); - } - - #[tokio::test] - async fn test_pending_cancellations_retain_an_unmarkable_agreement_whose_cancel_fails() { - // No cancel retry tracks it, so the next sweep tries again. - let chain_client = MockChainClient { - fail_cancels: true, - ..live_chain() - }; - - let (result, _, deleted) = - replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; - - assert!(result.is_err()); - assert!(!deleted); - } - - #[tokio::test] - async fn test_pending_cancellations_give_up_on_an_unmarkable_agreement_refusing_its_cancel() { - // Each resend of a cancel that mines without ending it costs gas, so it goes to an - // operator instead. - let chain_client = MockChainClient { - cancels_do_nothing: true, - ..live_chain() - }; - - let (result, old_id, deleted) = - replace_agreement_in(IndexingAgreementStatus::Unresponsive, &chain_client).await; - - assert!(result.is_ok(), "got {result:?}"); - assert!(chain_client.was_on_chain_cancel_attempted(&old_id)); - assert!(deleted); - } - - #[tokio::test] - async fn test_pending_cancellations_leave_a_cancelling_agreement_to_the_cancel_retry() { - let chain_client = live_chain(); - - let (result, old_id, deleted) = - replace_agreement_in(IndexingAgreementStatus::Cancelling, &chain_client).await; - - assert!(result.is_ok(), "got {result:?}"); - assert!(!chain_client.was_on_chain_cancel_attempted(&old_id)); - assert!(deleted); - } - #[tokio::test] async fn test_sweep_clears_already_canceled_pending_rows() { // The sweep is the long-lived recovery path. If the on-chain cancel