From cd1b224d20d5a1ef464aaf4e49265c30bfd25e02 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:03:30 +0100 Subject: [PATCH 01/16] fix(registry): make every registry implement its event and audit writes The registry trait gave 9 event and audit methods a do-nothing default, so a registry missing one would drop accepts, ends and their announcements without a word. Only the test stub keeps those defaults now. --- bin/dipper-service/src/registry/agreement.rs | 93 +++++++------------- 1 file changed, 34 insertions(+), 59 deletions(-) diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 7f8618b1..24ebe110 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -359,101 +359,76 @@ pub trait AgreementRegistry { } /// Fetch agreements awaiting a `terminated` event: terminal-cancel status, - /// genuinely accepted on-chain, marker unset. Default returns empty so mocks - /// need not override; the production registry runs the real query. + /// genuinely accepted on-chain, marker unset. async fn get_agreements_pending_terminated_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `terminated` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_terminated_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Fetch agreements awaiting an `accepted` event: `AcceptedOnChain`, accept - /// observed by the new reconcile path, marker unset. Default returns empty so - /// mocks need not override; the production registry runs the real query. + /// observed by the new reconcile path, marker unset. async fn get_agreements_pending_accepted_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `accepted` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_accepted_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Fetch agreements awaiting a `request.expired` event: `Expired`, marker - /// unset. Default returns empty so mocks need not override; the production - /// registry runs the real query. + /// unset. async fn get_agreements_pending_expired_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `request.expired` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_expired_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Record the accepted audit payload out-of-band (rejected-then-accepted /// anomaly), marking the row as genuinely accepted so its eventual - /// `terminated` is sweep-eligible. Default no-op so mocks need not override. + /// `terminated` is sweep-eligible. async fn record_accepted_audit( &self, - _agreement_id: &IndexingAgreementId, - _accepted_at: u64, - _accepted_tx: &str, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, + ) -> RegistryResult<()>; /// Record an agreement's accept and its end together, in 1 write, so nothing reads one - /// without the other. Default no-op so mocks need not override. + /// without the other. #[allow(clippy::too_many_arguments)] async fn record_accept_and_cancel_audit( &self, - _agreement_id: &IndexingAgreementId, - _accepted_at: u64, - _accepted_tx: &str, - _canceled_at: u64, - _canceled_by: &str, - _canceled_tx: Option<&str>, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, + canceled_at: u64, + canceled_by: &str, + canceled_tx: Option<&str>, + ) -> RegistryResult<()>; /// Record the cancel audit payload for a dipper-initiated cancel so the - /// emission sweep can populate the `terminated` event fields. Default no-op - /// so mocks need not override. + /// emission sweep can populate the `terminated` event fields. async fn record_cancel_audit( &self, - _agreement_id: &IndexingAgreementId, - _canceled_at: u64, - _canceled_by: &str, - _canceled_tx: Option<&str>, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + canceled_at: u64, + canceled_by: &str, + canceled_tx: Option<&str>, + ) -> RegistryResult<()>; /// Get `Created` agreements whose deadline has passed (by block timestamp). async fn get_expired_created_agreements( From ef40e474305a5fd89739bbd11703381eec2a4702 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:04:52 +0100 Subject: [PATCH 02/16] fix(k8s): make the example Slack webhook placeholder a valid URL The example config's webhook placeholder wasn't a URL, so a copy that kept it stopped dipper at startup. A test now reads the example's alerts section to keep it valid. --- bin/dipper-service/src/config.rs | 17 +++++++++++++++++ k8s/configmap-example.yaml | 2 +- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 8bc4619c..abab64f5 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -2118,6 +2118,23 @@ mod tests { ); } + #[test] + fn the_example_configmap_alerts_section_parses() { + let example = include_str!("../../../k8s/configmap-example.yaml"); + let json: String = example + .split_once("config.json: |\n") + .expect("the example holds a config.json") + .1 + .lines() + .map(|line| format!("{}\n", line.strip_prefix(" ").unwrap_or(line))) + .collect(); + let config: serde_json::Value = serde_json::from_str(&json).expect("config.json is JSON"); + + let alerts = serde_json::from_value::(config["alerts"].clone()) + .expect("the example's alerts section is a valid alerts config"); + assert!(alerts.slack_webhook_url.is_some()); + } + #[test] fn alerts_config_takes_a_webhook_and_keeps_the_default_events() { let alerts = serde_json::from_str::( diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 88222ea5..ae114cba 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -120,7 +120,7 @@ data: "threshold": 840 }, "alerts": { - "slack_webhook_url": "REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", + "slack_webhook_url": "https://hooks.slack.com/services/REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", "events": ["agreement_cancel_stuck", "rpc_blocks_refused", "nonce_gap_fill_failed"], "throttle": 900 } From 735c06525749cd3da3b9e7832433abffdb08127e Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:06:29 +0100 Subject: [PATCH 03/16] fix(liveness): stop a hung database from stalling the liveness check Marking a stale agreement for cancelling had no time limit, so a hung database held up every other agreement's check. The mark now gives up after the usual database timeout. --- bin/dipper-service/src/cancel_dispatch.rs | 48 ++++++++--- .../src/network/service/liveness_checker.rs | 80 ++++++++++++++----- 2 files changed, 93 insertions(+), 35 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index cedd4326..c6e40059 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -119,32 +119,56 @@ where R: AgreementRegistry + Sync, T: ChainClient, { + mark_cancelling(registry, agreement, reason).await?; + Ok(send_marked_cancel(registry, chain_client, agreement, reason, config).await) +} + +/// The first half of [`start_cancel`]: mark the agreement `Cancelling`, or abandoning. +pub async fn mark_cancelling( + registry: &R, + agreement: &IndexingAgreement, + reason: CancelReason, +) -> RegistryResult<()> { match reason { CancelReason::NotWanted => { registry .mark_indexing_agreement_as_cancelling(&agreement.id) - .await? + .await } CancelReason::Abandoned => { registry .mark_indexing_agreement_as_abandoning(&agreement.id) - .await? + .await } } +} + +/// The second half of [`start_cancel`]: cancel a marked agreement if the chain shows it live. +pub async fn send_marked_cancel( + registry: &R, + chain_client: &T, + agreement: &IndexingAgreement, + reason: CancelReason, + config: &IndexingAgreementConfig, +) -> CancelStarted +where + R: AgreementRegistry + Sync, + T: ChainClient, +{ let tx_hash = match cancel_if_live(chain_client, agreement, config).await { LiveCancel::Ended(tx_hash) => tx_hash, - LiveCancel::NotLive { .. } => return Ok(CancelStarted::NotLive), + LiveCancel::NotLive { .. } => return 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::MayBeLive); + return CancelStarted::MayBeLive; } LiveCancel::Unconfirmed { tx_hash, err } => { log_unconfirmed(agreement, tx_hash, &err); - return Ok(CancelStarted::MayBeLive); + return CancelStarted::MayBeLive; } }; tracing::info!( @@ -154,15 +178,13 @@ where ); // An offer never accepted could still land and be accepted until its deadline. if agreement.status != IndexingAgreementStatus::AcceptedOnChain { - return Ok(CancelStarted::NotLive); + return CancelStarted::NotLive; + } + if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { + CancelStarted::Ended + } else { + CancelStarted::NotLive } - Ok( - if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { - CancelStarted::Ended - } else { - CancelStarted::NotLive - }, - ) } /// Mark an agreement the chain shows dipper ended as ended, recording the cancel when its diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index a95e2682..38cc5102 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -508,7 +508,7 @@ async fn cancel_and_reassess( return; } let Some(started) = - start_abandoned_cancel(agreement, registry, chain_client, agreement_conf).await + start_abandoned_cancel(agreement, registry, chain_client, agreement_conf, db_timeout).await else { return; }; @@ -548,50 +548,64 @@ pub(crate) async fn replace_abandoned( } /// Mark a stale agreement abandoned and send its cancel; `None`, logged, when it isn't marked. +/// Only the mark is bounded by `db_timeout`, so a hung database can't stall the liveness cycle; +/// a mark that lands after the timeout leaves the agreement for the cancel retry. async fn start_abandoned_cancel( agreement: &IndexingAgreement, registry: &R, chain_client: &C, agreement_conf: &crate::config::IndexingAgreementConfig, + db_timeout: Duration, ) -> Option where R: AgreementRegistry + Sync, C: ChainClient, { - match crate::cancel_dispatch::start_cancel( - registry, - chain_client, - agreement, - CancelReason::Abandoned, - agreement_conf, + let marked = tokio::time::timeout( + db_timeout, + crate::cancel_dispatch::mark_cancelling(registry, agreement, CancelReason::Abandoned), ) - .await - { - Ok(started) => { - tracing::info!( - agreement_id = %agreement.id, - ?started, - reason = "indexer_stale", - "Cancelling stale agreement" - ); - Some(started) - } - Err(crate::registry::Error::NoRecordsUpdated) => { + .await; + match marked { + Ok(Ok(())) => {} + Ok(Err(crate::registry::Error::NoRecordsUpdated)) => { tracing::debug!( agreement_id = %agreement.id, "Stale agreement already ended or being cancelled" ); - None + return None; } - Err(err) => { + Ok(Err(err)) => { tracing::error!( agreement_id = %agreement.id, error = %err, "failed to mark stale agreement cancelling, will retry next cycle" ); - None + return None; + } + Err(_) => { + tracing::error!( + agreement_id = %agreement.id, + "timeout marking stale agreement cancelling, will retry next cycle" + ); + return None; } } + let started = crate::cancel_dispatch::send_marked_cancel( + registry, + chain_client, + agreement, + CancelReason::Abandoned, + agreement_conf, + ) + .await; + tracing::info!( + agreement_id = %agreement.id, + ?started, + reason = "indexer_stale", + "Cancelling stale agreement" + ); + Some(started) } /// Drop the pending cancellations an abandoned agreement holds as a replacement, so the @@ -960,6 +974,7 @@ mod tests { struct MockRegistry { calls: MockCalls, already_ending: bool, + mark_hangs: bool, get_request_result: Arc>>>>, } @@ -969,6 +984,7 @@ mod tests { Self { calls, already_ending: false, + mark_hangs: false, get_request_result: Arc::new(Mutex::new(Some(Ok(Some(request))))), } } @@ -1003,6 +1019,9 @@ mod tests { if self.already_ending { return Err(crate::registry::Error::NoRecordsUpdated); } + if self.mark_hangs { + std::future::pending::<()>().await; + } self.calls.abandoning.lock().unwrap().push(*id); Ok(()) } @@ -1610,6 +1629,23 @@ mod tests { assert!(calls.reassessments.lock().unwrap().is_empty()); } + #[tokio::test(start_paused = true)] + async fn gives_up_on_a_stale_agreement_whose_mark_hangs() { + // A hung database must not hold up the rest of the liveness cycle. + let agreement = stale_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry { + mark_hangs: true, + ..MockRegistry::new(calls.clone(), agreement.clone()) + }; + let chain = MockChainClient::success(calls.clone()); + + end_stale(&agreement, ®istry, &chain).await; + + assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert!(calls.reassessments.lock().unwrap().is_empty()); + } + #[tokio::test] async fn test_record_progress_calls_registry() { // Arrange From db50eca7d23eb5ade9e82da9234c6801a0c885fb Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:07:37 +0100 Subject: [PATCH 04/16] fix(cancel): wait longer before retrying a cancel that was just sent A cancel can queue behind another transaction for up to the 7-minute worker job limit, but the retry waited only 2 minutes, so it could send a second cancel. It now waits 8 minutes. --- .../src/network/service/cancel_retry.rs | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 3417b30e..c1673f80 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -41,9 +41,11 @@ 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. -const SETTLE_MINUTES: i32 = 2; +/// when it was marked can be mined first instead of being sent again. That cancel can wait +/// behind another submission for its whole submit deadline, which stays under the 7-minute +/// worker job limit, so this outlasts that limit. One moved back to cancelling had none sent, +/// so it doesn't wait. +const SETTLE_MINUTES: i32 = 8; /// How long the chain listener gets, from when a check first finds an agreement ended, to /// record when and in which transaction it ended, before the retry marks it without them. @@ -433,6 +435,12 @@ mod tests { const DEADLINE: u64 = 1_000; + #[test] + fn a_marked_agreement_waits_out_a_whole_submit_deadline_before_a_retry() { + let settle = Duration::from_secs(u64::from(SETTLE_MINUTES.unsigned_abs()) * 60); + assert!(settle > crate::worker::service::PROCESS_JOB_TIMEOUT); + } + #[derive(Default)] struct MockRegistry { cancelling: Vec, From f642a58ab111259b5e159134a70787e1beb47f6b Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:08:53 +0100 Subject: [PATCH 05/16] fix(listener): record no accept for a rejected agreement already ended When an agreement the indexer rejected shows up accepted on-chain but has ended by the time dipper checks, it stays rejected. Recording its accept announced it with no end to follow. --- .../src/network/service/chain_listener.rs | 38 ++++++++++++++++--- 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index a3ce327c..457d035c 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -845,16 +845,18 @@ where indexer = %snapshot.indexer, "Rejected agreement accepted on-chain, cancelling it" ); - crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; - // This row goes Rejected -> Canceled without ever transiting // AcceptedOnChain, so `apply_reconciliation` never records the accept. // Persist it from the snapshot so the eventual `terminated` (emitted by // the sweep after the cancel) is eligible (`accepted_at IS NOT NULL`) and - // carries accurate accept data. - if let Err(err) = registry - .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) - .await + // carries accurate accept data. One left Rejected, as the chain shows it + // already ended, gets no accept on record, so none is announced. + let reopened = + crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; + if reopened + && let Err(err) = registry + .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) + .await { tracing::warn!( agreement_id = %agreement.id, @@ -2879,6 +2881,30 @@ mod tests { assert!(registry.was_reopened(&agreement_id)); } + #[tokio::test] + async fn test_reconcile_records_no_accept_for_a_rejected_agreement_already_ended() { + // The snapshot shows it accepted, but by the time dipper reads the chain the indexer + // has ended it. It stays Rejected, so an accept on record would be announced with no + // end ever following it. + let registry = MockRegistry::new(); + let chain_client = MockChainClient::default(); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, IndexingAgreementStatus::Rejected); + + let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); + let result = reconcile_agreement( + &snapshot, + ®istry, + &chain_client, + test_agreement_conf().as_ref(), + ) + .await; + + assert!(result.is_ok()); + assert!(!registry.was_reopened(&agreement_id)); + assert!(registry.audit_writes().is_empty()); + } + #[tokio::test] async fn test_reconcile_rejected_already_canceled_skips_queue_and_marks_local_cancel() { // The Rejected agreement was canceled on-chain between polls. Dipper From 90b8fb65fca05bd98aa7066595fdf5c5e031bbcd Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:10:02 +0100 Subject: [PATCH 06/16] fix(cancel): retry accepted cancels even when the chain time is unknown A failed read of the chain's time skipped every cancel in the sweep, though only offers never accepted need it, to tell whether their deadline has passed. Now only those wait. --- .../src/network/service/cancel_retry.rs | 42 +++++++++++++------ 1 file changed, 30 insertions(+), 12 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index c1673f80..322ff341 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -149,7 +149,8 @@ where /// 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. +/// fallen behind doesn't hold that up. Without it, such an offer is still withdrawn, but only +/// accepted agreements can be marked ended. pub async fn retry_cancelling_agreements( registry: &R, chain_client: &T, @@ -171,8 +172,10 @@ pub async fn retry_cancelling_agreements( if cancelling.is_empty() { return; } - let Some(chain_now) = chain_time(chain_client).await else { - return; + let chain_now = if cancelling.iter().any(|row| !row.accepted_on_chain) { + chain_time(chain_client).await + } else { + None }; let started = std::time::Instant::now(); for (done, row) in cancelling.iter().enumerate() { @@ -193,7 +196,7 @@ async fn chain_time(chain_client: &T) -> Option { Err(err) => { tracing::warn!( error = %err, - "Failed to read the chain's time; cancels are retried next sweep" + "Failed to read the chain's time; offers never accepted stay cancelling until a later sweep" ); None } @@ -205,7 +208,7 @@ async fn retry_cancel( chain_client: &T, config: &IndexingAgreementConfig, row: &CancellingAgreement, - chain_now: u64, + chain_now: Option, ) where R: AgreementRegistry + Sync, T: ChainClient, @@ -255,7 +258,7 @@ async fn confirm_if_over( row: &CancellingAgreement, tx_hash: Option, by_indexer: bool, - chain_now: u64, + chain_now: Option, ) -> bool { let agreement = &row.agreement; let past_grace = row @@ -264,7 +267,7 @@ async fn confirm_if_over( let can_confirm = if row.accepted_on_chain { tx_hash.is_some() || past_grace } else { - chain_now > agreement.terms.deadline + chain_now.is_some_and(|now| now > agreement.terms.deadline) }; if !can_confirm { return false; @@ -770,7 +773,8 @@ mod tests { let ended = ended_abandoned(); let registry = MockRegistry { awaiting_replacement: vec![ended.clone()], - ..registry_with_one(true) + // Never accepted, so each sweep reads the chain's time, which counts the sweeps. + ..registry_with_one(false) }; let chain = Arc::new(live_chain()); let queue = MockQueue::default(); @@ -800,17 +804,31 @@ mod tests { } #[tokio::test] - async fn waits_for_the_next_sweep_when_the_chain_time_cannot_be_read() { + async fn cancels_an_accepted_agreement_without_the_chain_time() { + // Only an offer never accepted needs the chain's time, to tell if its deadline passed. let registry = registry_with_one(true); + let chain = live_chain(); + + retry(®istry, &chain, 0).await; + + assert_eq!(chain.clock_reads.load(Ordering::SeqCst), 0); + assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 1); + assert_eq!(registry.marked_cancelled.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn withdraws_an_offer_but_keeps_it_cancelling_when_the_chain_time_cannot_be_read() { + let registry = registry_with_one(false); let chain = MockChain { clock_fails: true, ..live_chain() }; - retry(®istry, &chain, 0).await; + retry(®istry, &chain, DEADLINE + 1).await; - assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 0); - assert_eq!(registry.checks.load(Ordering::SeqCst), 0); + assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 1); + assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + assert_eq!(registry.checks.load(Ordering::SeqCst), 1); } #[tokio::test] From 7aa00d1f9e4ba44f85629057aa0252053ad5a6c2 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:12:53 +0100 Subject: [PATCH 07/16] fix(pacing): count offers still being withdrawn as in flight Offer pacing counted only agreements waiting for acceptance, so an offer dipper was withdrawing didn't count, though the indexer can still accept it until the withdrawal lands or its deadline. --- bin/dipper-service/src/registry/agreement.rs | 6 +- dipper-pgregistry/src/postgres.rs | 11 ++- .../tests/it_registry_postgres.rs | 77 +++++++++++++++++++ 3 files changed, 88 insertions(+), 6 deletions(-) diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 24ebe110..5eb15ee8 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -517,9 +517,9 @@ pub trait AgreementRegistry { &self, ) -> RegistryResult>; - /// Count `Created` (in-flight, not yet accepted) agreements per indexer, - /// plus the global total. Offer pacing reads both to decide whether an - /// indexer or the network has spare acceptance capacity. + /// Count in-flight offers per indexer, plus the global total: `Created` agreements, and + /// `Cancelling` ones never accepted until their offer deadline. Offer pacing reads both to + /// decide whether an indexer or the network has spare acceptance capacity. async fn count_created_agreements_by_indexer( &self, ) -> RegistryResult<(std::collections::HashMap, u64)>; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index e9b7411e..8f02687b 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -2029,9 +2029,10 @@ impl PgRegistry { .collect()) } - /// Count `Created` (in-flight, not yet accepted) agreements per indexer, - /// returning the per-indexer map and global total in one round-trip. Offer - /// pacing reads both to gauge spare acceptance capacity before creating more. + /// Count in-flight offers per indexer, returning the per-indexer map and global total in + /// one round-trip. Offer pacing reads both to gauge spare acceptance capacity before + /// creating more. Besides `Created` agreements, a `Cancelling` one never accepted counts + /// until its offer deadline, since its offer may be accepted before its cancel lands. #[expect( clippy::cast_sign_loss, reason = "predates this lint; fix when next touched" @@ -2046,10 +2047,14 @@ impl PgRegistry { SELECT indexer_id, COUNT(*) as count FROM dipper_reg_indexing_agreements WHERE status = $1 + OR (status = $2 + AND accepted_at IS NULL + AND CAST(terms->>'deadline' AS bigint) > EXTRACT(EPOCH FROM now())) GROUP BY GROUPING SETS ((indexer_id), ()) "#, ) .bind(IndexingAgreementStatus::Created) + .bind(IndexingAgreementStatus::Cancelling) .fetch_all(&self.pool) .await?; diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index d5fd3cef..3b0f8124 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3424,6 +3424,83 @@ async fn count_created_agreements_by_indexer_counts_only_created() { assert_eq!(global, 3, "global counts only the 3 Created rows"); } +#[tokio::test] +async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let requested_by = address!("8f8c426f956876325b1e037c6eae9b189952994c"); + let indexer = indexer_id!("cccccccccccccccccccccccccccccccccccccccc"); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + // 3 agreements being cancelled: an offer still open, one accepted, one past its deadline. + let mut ids = Vec::new(); + for (deployment_id, deadline) in [ + ( + deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"), + now + 3_600, + ), + ( + deployment_id!("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"), + now + 3_600, + ), + ( + deployment_id!("QmYbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"), + now - 60, + ), + ] { + let request_id = match registry + .set_indexing_target_candidates(requested_by, deployment_id, 42161, 3) + .await + .expect("set target candidates") + { + dipper_pgregistry::IndexingRequestSetTargetOutcome::Inserted { id } => id, + other => panic!("unexpected set-target outcome: {other:?}"), + }; + let mut terms = Faker.fake::(); + terms.metadata.subgraph_deployment_id = deployment_id; + terms.deadline = deadline; + let id = registry + .register_new_indexing_agreement(NewAgreementParams { + agreement_id: Faker.fake::(), + nonce_uuid: uuid::Uuid::now_v7(), + request_id, + deployment_id, + indexer_id: indexer, + indexer_url: "http://localhost:8020".parse().expect("Invalid URL"), + terms, + terms_version_hash: None, + }) + .await + .expect("register agreement"); + registry + .mark_indexing_agreement_as_cancelling(&id) + .await + .expect("mark cancelling"); + ids.push(id); + } + registry + .record_accepted_audit(&ids[1], now, "0xacc") + .await + .expect("accept record"); + + //* When + let (per_indexer, global) = registry + .count_created_agreements_by_indexer() + .await + .expect("count in-flight offers"); + + //* Then + assert_eq!( + per_indexer.get(&indexer).copied(), + Some(1), + "only the open offer can still be accepted" + ); + assert_eq!(global, 1); +} + fn fixture_agreement(prefix: u8) -> IndexingAgreementId { let mut bytes = [0u8; 16]; bytes[0] = prefix; From b553c27f0383a999eb8756ae6c40e40f62390c9f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:14:53 +0100 Subject: [PATCH 08/16] fix(listener): re-read a page of chain changes that failed to apply When the database failed while applying a page of on-chain changes, the listener moved past the page anyway, so a change like reopening a live cancelled agreement was lost for good. --- .../src/network/service/chain_listener.rs | 85 +++++++++++++++++-- 1 file changed, 80 insertions(+), 5 deletions(-) diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 457d035c..fbf96add 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -555,6 +555,9 @@ where // another payer's) drop out as `None` below. let snapshot_ids: Vec = result.snapshots.iter().map(|s| s.agreement_id).collect(); + // Agreements whose change this page failed to apply. The cursor stays before the page + // until none fail, as nothing else would look at them again. + let mut unapplied: Vec = Vec::new(); let mut agreements_by_id = match registry.get_indexing_agreements_by_ids(&snapshot_ids).await { Ok(m) => m, @@ -566,6 +569,7 @@ where "Failed to batch-fetch agreements for page; counting snapshots as errors" ); drain_errors += snapshot_ids.len() as u64; + unapplied.extend(&snapshot_ids); std::collections::HashMap::new() } }; @@ -588,12 +592,13 @@ where "Failed to prepare reconciliation for snapshot" ); drain_errors += 1; + unapplied.push(snapshot.agreement_id); } } } - // On batch error the tx rolls back; the next tick re-reads - // these rows via the reorg buffer. The successful path + // On batch error the tx rolls back; the cursor is held so the + // next tick re-reads these rows. The successful path // pre-fills every input id in `outcomes` (with `default()` // for rows whose CAS guard didn't match), so a missing id // is an unambiguous signal that the whole batch failed. @@ -611,6 +616,7 @@ where "Batched apply_reconciliation failed; counting page items as errors" ); drain_errors += items.len() as u64; + unapplied.extend(items.iter().map(|item| item.agreement_id)); std::collections::HashMap::new() } } @@ -672,7 +678,15 @@ where // Persist per-page so a crash mid-drain replays at most // one page. Skip the write when neither cursor nor // ratcheted timestamp moved. - let advance_cursor = *cursor < new_cursor; + let advance_cursor = unapplied.is_empty() && *cursor < new_cursor; + if !unapplied.is_empty() { + tracing::error!( + event = "chain_listener_cursor_held", + cursor_block = cursor.block, + agreement_ids = ?unapplied, + "Failed to apply on-chain changes to agreements; re-reading them next poll" + ); + } let timestamp_changed = ratchet_timestamp != *last_persisted_timestamp; if advance_cursor || timestamp_changed { let cursor_to_persist = if advance_cursor { @@ -701,8 +715,8 @@ where // nothing to do this tick. break 'drain; } else { - // Held-back cursor (parse failure path); reconcile - // already ran on what we did read, retry next tick. + // Held-back cursor (parse failure, or a change that failed to + // apply); reconcile already ran on what we did read, retry next tick. break 'drain; } @@ -4187,6 +4201,67 @@ mod tests { } } + /// A page whose changes fail to apply keeps the cursor before it, so the next poll + /// re-reads it: no later change to those agreements may come to apply it instead. + #[tokio::test] + async fn test_drain_holds_the_cursor_until_a_failed_page_applies() { + let registry = MockRegistry::new(); + let chain_client = MockChainClient::default(); + registry.set_fail_batch(true); + let event_source = super::super::chain_events::mock::MockEventSource::new(); + event_source.set_latest_block(100); + event_source.set_latest_block_timestamp(Some(1_700_000_000)); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, IndexingAgreementStatus::Created); + event_source.add_snapshots(vec![AgreementStateSnapshot { + agreement_id, + indexer: Address::ZERO, + state: super::super::chain_events::AgreementState::Accepted, + canceled_by: Address::ZERO, + last_state_change_block: 10, + accepted_at: 0, + accepted_tx: String::new(), + canceled_at: 0, + canceled_tx: String::new(), + }]); + let mut cursor = Cursor::genesis(); + let mut last_persisted_timestamp: Option = None; + let mut last_chain_ts_persist_wall = std::time::Instant::now(); + let mut last_subgraph_head: u64 = 0; + let mut stall_count: u32 = 0; + let mut consecutive_failures: u32 = 0; + let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + + for fails in [true, false] { + registry.set_fail_batch(fails); + drain_once( + &mut cursor, + &mut last_persisted_timestamp, + &mut last_chain_ts_persist_wall, + &mut last_subgraph_head, + &mut stall_count, + &mut consecutive_failures, + 1337, + 0, + 10, + false, + ®istry, + &chain_client, + &event_source, + &mut rx_stop, + test_agreement_conf().as_ref(), + ) + .await + .expect("subgraph fetch succeeded"); + if fails { + assert_eq!(cursor, Cursor::genesis(), "held before the failed page"); + } + } + + assert!(registry.was_marked_accepted_on_chain(&agreement_id)); + assert!(cursor > Cursor::genesis()); + } + /// Drives a hostile timestamp drift through `drain_once` and asserts /// the persisted chain timestamp is capped near /// `CHAIN_TS_DRIFT_TOLERANCE_SECS` instead of accepting the huge From 6ed2d5b367f0a863452fbfe395b3cb4084e5383a Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:17:17 +0100 Subject: [PATCH 09/16] fix(chain): let 2 RPC endpoints, not 1, move the newest block far ahead Once confirmed, the newest block dipper had seen took any block up to a week ahead from 1 endpoint, so a faulty one could have every correct endpoint refused as behind for hours. --- bin/dipper-service/src/chain_client/client.rs | 131 +++++++++++++++--- 1 file changed, 108 insertions(+), 23 deletions(-) diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 9376c63c..3592ddd1 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -92,12 +92,26 @@ const CROSS_CHECK_DEADLINE: Duration = Duration::from_secs(3); /// Blocks too far ahead of it are refused only once it is confirmed, by the receipt for one of /// dipper's transactions or by 2 endpoints agreeing, so an endpoint stuck far behind, or far /// ahead, that answers first after a restart can't shut out the endpoints that are right. +/// Once confirmed, 1 endpoint alone can't move it far ahead either, for the same reason. #[derive(Debug)] struct SeenBlock { number: u64, moved_at: Instant, confirmed: bool, cross_checked_at: Option, + /// An endpoint reported a block further ahead than 1 endpoint alone may move it, so the + /// endpoints are asked whether they agree. + jump_reported: bool, +} + +/// What became of a block 1 endpoint reported. +#[derive(Debug, PartialEq, Eq)] +enum Noted { + Taken, + /// Too far ahead for 1 endpoint alone; it waits for 2 to agree. + Unagreed, + /// Too far ahead to be real. + TooFarAhead, } impl SeenBlock { @@ -107,6 +121,7 @@ impl SeenBlock { moved_at: Instant::now(), confirmed: false, cross_checked_at: None, + jump_reported: false, } } @@ -130,24 +145,47 @@ impl SeenBlock { (self.number, highest) } - /// Take `block` as seen, unless it is too far ahead of a confirmed one to be real; false - /// when it is. - fn advance(&mut self, block: u64, now: Instant) -> bool { - if self.confirmed && block > self.believable_limit(now) { - return false; + /// The highest block 1 endpoint alone may move a confirmed newest block to: an hour of + /// blocks past it, plus what the chain can have added since. + fn lone_limit(&self, now: Instant) -> u64 { + let since = now.saturating_duration_since(self.moved_at).as_secs(); + self.number + .saturating_add(ALERT_GAP_BLOCKS) + .saturating_add(since.saturating_mul(BLOCKS_PER_SECOND)) + } + + /// Take `block`, reported by 1 endpoint, as seen. Once the newest block is confirmed, one + /// too far ahead to be real is refused, and one past [`Self::lone_limit`] waits for 2 + /// endpoints to agree, unless this is the `only_endpoint` there is to ask. + fn advance(&mut self, block: u64, now: Instant, only_endpoint: bool) -> Noted { + if self.confirmed { + if block > self.believable_limit(now) { + return Noted::TooFarAhead; + } + if !only_endpoint && block > self.lone_limit(now) { + self.jump_reported = true; + return Noted::Unagreed; + } } + self.take(block, now); + Noted::Taken + } + + fn take(&mut self, block: u64, now: Instant) { if block > self.number { self.number = block; self.moved_at = now; } - true } /// Take `block` as one the chain is known to have reached. An unconfirmed newest block more /// than an hour of blocks past it came from a faulty endpoint, so it is replaced. fn confirm(&mut self, block: u64, now: Instant) { + self.jump_reported = false; if self.confirmed { - self.advance(block, now); + if block <= self.believable_limit(now) { + self.take(block, now); + } return; } if block > self.number || self.number > block.saturating_add(ALERT_GAP_BLOCKS) { @@ -157,13 +195,13 @@ impl SeenBlock { self.confirmed = true; } - /// Whether to ask every endpoint for its latest block: only while unconfirmed, and at most - /// once every [`CROSS_CHECK_INTERVAL`]. + /// Whether to ask every endpoint for its latest block: only while unconfirmed or after a + /// jump 1 endpoint reported, and at most once every [`CROSS_CHECK_INTERVAL`]. fn cross_check_due(&mut self, now: Instant) -> bool { let checked_lately = self .cross_checked_at .is_some_and(|at| now.saturating_duration_since(at) < CROSS_CHECK_INTERVAL); - if self.confirmed || checked_lately { + if (self.confirmed && !self.jump_reported) || checked_lately { return false; } self.cross_checked_at = Some(now); @@ -899,24 +937,34 @@ impl AlloyChainClient { Some(head) => self.seen_block().confirm(head, Instant::now()), None => tracing::warn!( "No 2 RPC endpoints agree on the chain's latest block; reads aren't checked \ - against blocks too far ahead until they do" + against blocks too far ahead, nor the newest block seen moved far ahead, until \ + they do" ), } } /// Remember a block dipper has seen, so later reads are never older. One too far ahead to - /// be real is ignored, so it can't refuse every read after it. + /// be real is ignored, so it can't refuse every read after it, and one over an hour ahead + /// waits for 2 endpoints to agree on it. fn note_block(&self, block: u64) { - let (taken, seen) = { + let only_endpoint = self.inner.rpc_pool.endpoint_count() < 2; + let (noted, seen) = { let mut seen = self.seen_block(); - (seen.advance(block, Instant::now()), seen.number) + (seen.advance(block, Instant::now(), only_endpoint), seen.number) }; - if !taken { - tracing::warn!( + match noted { + Noted::Taken => {} + Noted::Unagreed => tracing::warn!( + block, + seen_block = seen, + "Not moving to a block over an hour ahead of the newest block dipper has seen \ + until 2 RPC endpoints agree on it" + ), + Noted::TooFarAhead => tracing::warn!( block, seen_block = seen, "Ignoring a block too far ahead of the newest block dipper has seen" - ); + ), } } @@ -2549,6 +2597,7 @@ mod tests { moved_at: now.checked_sub(Duration::from_secs(600)).expect("instant"), confirmed: true, cross_checked_at: None, + jump_reported: false, }; assert_eq!(seen.bounds(now), (100, 100 + WEEK_OF_BLOCKS + 2_400)); @@ -2590,8 +2639,8 @@ mod tests { // So a read just after dipper's cancel mined can't come from an endpoint behind it. let now = Instant::now(); let mut seen = SeenBlock::new(); - assert!(seen.advance(100, now)); - assert!(seen.advance(90, now)); + assert_eq!(seen.advance(100, now, false), Noted::Taken); + assert_eq!(seen.advance(90, now, false), Noted::Taken); assert_eq!(seen.bounds(now), (100, u64::MAX)); } @@ -2603,21 +2652,57 @@ mod tests { let now = Instant::now(); let mut seen = SeenBlock::new(); let real = 1_000 + WEEK_OF_BLOCKS * 3; - assert!(seen.advance(1_000, now)); - assert!(seen.advance(real, now), "taken while unconfirmed"); + assert_eq!(seen.advance(1_000, now, false), Noted::Taken); + assert_eq!( + seen.advance(real, now, false), + Noted::Taken, + "taken while unconfirmed" + ); seen.confirm(real, now); - assert!(!seen.advance(real + WEEK_OF_BLOCKS * 2, now)); + assert_eq!( + seen.advance(real + WEEK_OF_BLOCKS * 2, now, false), + Noted::TooFarAhead + ); assert_eq!(seen.bounds(now).0, real); } + #[test] + fn waits_for_2_endpoints_to_agree_before_a_jump_past_a_confirmed_block() { + // 1 faulty endpoint moving it far ahead would have every endpoint that is right + // refused as behind, for as long as the real chain takes to catch up. + let now = Instant::now(); + let mut seen = SeenBlock::new(); + let real = 5_000_000; + seen.confirm(real, now); + let jump = real + ALERT_GAP_BLOCKS + 1_000; + + assert_eq!(seen.advance(real + 10, now, false), Noted::Taken); + assert_eq!(seen.advance(jump, now, false), Noted::Unagreed); + assert_eq!(seen.bounds(now).0, real + 10); + assert!(seen.cross_check_due(now), "the endpoints are asked"); + + seen.confirm(jump, now); + + assert_eq!(seen.bounds(now).0, jump, "taken once 2 agree"); + assert!(!seen.cross_check_due(now + CROSS_CHECK_INTERVAL)); + assert_eq!( + seen.advance(jump + ALERT_GAP_BLOCKS + 1, now, true), + Noted::Taken, + "the only endpoint has no other to agree with" + ); + } + #[test] fn a_confirmed_block_replaces_one_far_ahead_that_was_never_confirmed() { let now = Instant::now(); let mut seen = SeenBlock::new(); let real = 5_000_000; - assert!(seen.advance(real + WEEK_OF_BLOCKS * 10, now)); + assert_eq!( + seen.advance(real + WEEK_OF_BLOCKS * 10, now, false), + Noted::Taken + ); seen.confirm(real, now); From ee98b94b8454fa0a8ae347127c5cf974e920cc42 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:19:13 +0100 Subject: [PATCH 10/16] docs(registry): say the test stub alone keeps the no-op audit defaults A comment in the test stub still said its do-nothing event and audit methods copied defaults from the real registry trait, which no longer has any. --- bin/dipper-service/src/registry/agreement_stub.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index dd2950b4..1260add5 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -271,8 +271,8 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("count_accepted_agreements_by_deployment") } - // The emission-sweep and audit methods mirror the real trait's defaults - // (empty batch / no-op) so only emission-focused mocks override them. + // The emission-sweep and audit methods default to an empty batch or a + // no-op, here only, so only emission-focused mocks override them. async fn get_agreements_pending_terminated_emission( &self, _limit: i64, From 15c1e513d14c3d30c361b091f64ca480b670fc41 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:21:53 +0100 Subject: [PATCH 11/16] fix(reassess): keep an abandoned agreement's slot while it may be paid Only the liveness check waited for a stale agreement's cancel before replacing it; any other reassessment saw the request short and filled the slot, paying 2 indexers for it until then. --- .../src/network/service/liveness_checker.rs | 35 ++++++++ bin/dipper-service/src/registry.rs | 10 +++ bin/dipper-service/src/registry/agreement.rs | 7 ++ .../src/registry/agreement_stub.rs | 14 ++++ .../handlers/reassess_indexing_request.rs | 75 +++++++++++++++-- dipper-pgregistry/src/postgres.rs | 24 ++++++ .../tests/it_registry_postgres.rs | 80 +++++++++++++++++++ 7 files changed, 238 insertions(+), 7 deletions(-) diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 38cc5102..e32c6d15 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -520,9 +520,25 @@ async fn cancel_and_reassess( ); return; } + if started == CancelStarted::NotLive { + note_seen_ended(agreement, registry).await; + } replace_abandoned(agreement, registry, worker_queue, db_timeout, queue_timeout).await; } +/// Note that the chain shows a stale agreement still cancelling has ended, so it no longer +/// holds its slot in the request and the reassessment queued next fills it. +async fn note_seen_ended(agreement: &IndexingAgreement, registry: &R) { + match registry.record_cancel_check(&agreement.id, 0, Some(true)).await { + Ok(_) | Err(crate::registry::Error::NoRecordsUpdated) => {} + Err(err) => tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "Failed to note a stale agreement ended; its replacement may wait for a later reassessment" + ), + } +} + /// 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( @@ -969,6 +985,8 @@ mod tests { cancel_audits: Arc>>, /// Ids noted as having their replacement queued. replacements_noted: Arc>>, + /// Ids noted as seen ended on-chain while still cancelling. + seen_ended: Arc>>, } struct MockRegistry { @@ -1031,6 +1049,18 @@ mod tests { Ok(()) } + async fn record_cancel_check( + &self, + id: &IndexingAgreementId, + _failed_attempts: u32, + ended: Option, + ) -> RegistryResult { + if ended == Some(true) { + self.calls.seen_ended.lock().unwrap().push(*id); + } + Ok(0) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -1587,6 +1617,11 @@ mod tests { end_stale(&agreement, ®istry, &chain).await; assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert_eq!( + calls.seen_ended.lock().unwrap().as_slice(), + &[agreement.id], + "so it no longer holds the slot its replacement fills" + ); assert_eq!( calls.reassessments.lock().unwrap().as_slice(), &[agreement.indexing_request_id] diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 71470cda..f51a0ceb 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -733,6 +733,16 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn count_abandoned_agreements_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> RegistryResult { + self.inner + .count_abandoned_agreements_holding_slots(request_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 5eb15ee8..25e23f64 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -497,6 +497,13 @@ pub trait AgreementRegistry { /// Note that an agreement's replacement has been queued. async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()>; + /// How many of a request's slots are held by agreements whose indexer stopped serving them + /// but that may still be paid, so a reassessment fills only the rest. + async fn count_abandoned_agreements_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> 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 1260add5..2e6be4f4 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -236,6 +236,13 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_replacement_queued") } + async fn count_abandoned_agreements_holding_slots( + &self, + _request_id: &IndexingRequestId, + ) -> Result { + unimplemented!("count_abandoned_agreements_holding_slots") + } + async fn update_agreement_sync_progress( &self, _id: &IndexingAgreementId, @@ -568,6 +575,13 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_replacement_queued(self, id).await } + async fn count_abandoned_agreements_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> Result { + StubAgreementRegistry::count_abandoned_agreements_holding_slots(self, request_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 584e08d3..c5c86a7f 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -147,6 +147,15 @@ where .map(|agreement| agreement.indexer.id) .collect(); + // An agreement whose indexer stopped serving it keeps its slot while it may still be paid, + // so it isn't replaced while both are paid; once it can't be, its replacement is queued. + let held_slots = ctx + .registry + .count_abandoned_agreements_holding_slots(indexing_request_id) + .await + .map_err(|err| JobError::Fatal(err.into()))?; + let slots = num_candidates.saturating_sub(held_slots); + // In-flight offer counts, fetched once and reused by both the saturated // short-circuit (global figure only) and the per-indexer cap after IISA. let (in_flight_per_indexer, global_in_flight) = ctx @@ -162,7 +171,7 @@ where if should_defer_saturated( global_in_flight, global_cap, - *num_candidates, + slots, active_agreements.len(), ) { tracing::debug!( @@ -251,7 +260,7 @@ where // stall new requests until IISA recovers — see README "IISA dependency". let target_selected: Vec = match ctx .iisa - .select_indexers(*deployment_id, *num_candidates, &context) + .select_indexers(*deployment_id, slots, &context) .await { Ok(indexers) => indexers, @@ -286,6 +295,7 @@ where indexing_request_id=%indexing_request_id, deployment_id=%deployment_id, requested_num_candidates = num_candidates, + held_slots, iisa_returned_count = target_selected.len(), current_active_count = current_ids.len(), to_add_count = to_add.len(), @@ -302,7 +312,7 @@ where // persistently short request. The atomic latch update makes this retry-safe: // a re-run after a later failure sees the latch already set and does not // re-emit. - let in_shortfall = target_selected.len() < *num_candidates; + let in_shortfall = target_selected.len() < slots; let shortfall_changed = ctx .registry .set_indexing_request_shortfall_active(indexing_request_id, in_shortfall) @@ -321,7 +331,7 @@ where *deployment_id, ctx.signer.chain_id(), proto::SubgraphIndexingAgreementNIndexersUnavailable { - agreements_requested: *num_candidates as i32, + agreements_requested: slots as i32, candidates_returned: target_selected.len() as i32, }, ); @@ -871,7 +881,7 @@ mod lifecycle_event_tests { // ---- Mock: IISA candidate selection ------------------------------------- - /// Returns a fixed list of [`SelectedIndexer`] regardless of arguments. + /// Returns a fixed list of [`SelectedIndexer`], at most as many as asked for. struct MockIisa { selected: Vec, } @@ -881,10 +891,10 @@ mod lifecycle_event_tests { async fn select_indexers( &self, _deployment_id: DeploymentId, - _num_candidates: usize, + num_candidates: usize, _context: &SelectionContext, ) -> std::result::Result, SelectionError> { - Ok(self.selected.clone()) + Ok(self.selected.iter().take(num_candidates).cloned().collect()) } async fn dips_accepting_indexers( @@ -1039,6 +1049,8 @@ mod lifecycle_event_tests { /// Ids marked CanceledByRequester locally. marked_cancelled: Arc>>, marked_cancelling: Arc>>, + /// Slots held by agreements whose indexer stopped serving them. + held_slots: usize, } #[async_trait] @@ -1119,6 +1131,12 @@ mod lifecycle_event_tests { ) -> RegistryResult> { Ok(self.active_agreements.clone()) } + async fn count_abandoned_agreements_holding_slots( + &self, + _request_id: &IndexingRequestId, + ) -> RegistryResult { + Ok(self.held_slots) + } // Populates the `terminated` event's remaining count. async fn count_accepted_agreements_by_deployment( &self, @@ -1500,6 +1518,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(true), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, @@ -1538,6 +1557,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, @@ -1594,6 +1614,43 @@ mod lifecycle_event_tests { assert_eq!(unavailable[0].candidates_returned, 0); } + #[tokio::test] + async fn fills_only_the_slots_an_abandoned_agreement_does_not_hold() { + // The abandoned agreement may still be paid until its cancel lands, so filling its + // slot now would pay 2 indexers for it; its replacement is queued once it can't be. + let candidates = [indexer_id(0x21), indexer_id(0x22), indexer_id(0x23)]; + let mut snapshot = indexer_urls::Snapshot::new(); + for id in candidates { + snapshot.insert(id, indexer_url()); + } + let events = CapturingEventsProducer::new(); + let queue = MockQueue::default(); + let ctx = build_ctx( + MockRegistry { + held_slots: 1, + ..MockRegistry::default() + }, + MockIisa { + selected: candidates.iter().copied().map(selected).collect(), + }, + queue.clone(), + MockChainClient::default(), + events.clone(), + snapshot, + ); + + handle(ctx, &test_message(3)).await.expect("handler ok"); + + assert_eq!(queue.proposals.lock().unwrap().len(), 2); + assert!( + !events + .events() + .iter() + .any(|e| matches!(e, CapturedEvent::NIndexersUnavailable { .. })), + "the 2 slots to fill are filled" + ); + } + #[tokio::test] async fn emits_shortfall_when_coverage_drops_below_target() { // Item 9: an indexer leaves the target group and IISA can't replace it, so @@ -1610,6 +1667,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, @@ -1665,6 +1723,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, @@ -1730,6 +1789,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, @@ -1862,6 +1922,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held_slots: 0, }; let ctx = build_ctx( registry, diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 8f02687b..ff3cdb7b 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1951,6 +1951,30 @@ impl PgRegistry { .map_err(Into::into) } + /// Count a request's agreements whose indexer stopped serving them that may still be paid: + /// being cancelled, not yet seen ended on-chain, and with no replacement queued. Each keeps + /// its slot in the request until then, so it isn't replaced while both are paid. + pub async fn count_abandoned_agreements_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> Result { + let (count,): (i64,) = sqlx::query_as( + r#" + SELECT COUNT(*) + FROM dipper_reg_indexing_agreements + WHERE indexing_request_id = $1 + AND status = $2 + AND replacement_pending + AND ended_seen_at IS NULL + "#, + ) + .bind(request_id) + .bind(IndexingAgreementStatus::Cancelling) + .fetch_one(&self.pool) + .await?; + Ok(usize::try_from(count).unwrap_or_default()) + } + /// Note that an agreement's replacement has been queued. pub async fn mark_replacement_queued( &self, diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index 3b0f8124..e907b216 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3501,6 +3501,86 @@ async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { assert_eq!(global, 1); } +#[tokio::test] +async fn an_abandoned_agreement_holds_its_slot_while_it_may_still_be_paid() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let deployment_id = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"); + let request_id = match registry + .set_indexing_target_candidates( + address!("8f8c426f956876325b1e037c6eae9b189952994c"), + deployment_id, + 42161, + 3, + ) + .await + .expect("set target candidates") + { + dipper_pgregistry::IndexingRequestSetTargetOutcome::Inserted { id } => id, + other => panic!("unexpected set-target outcome: {other:?}"), + }; + let mut ids = Vec::new(); + for indexer_id in [ + indexer_id!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + indexer_id!("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), + ] { + let mut terms = Faker.fake::(); + terms.metadata.subgraph_deployment_id = deployment_id; + let id = registry + .register_new_indexing_agreement(NewAgreementParams { + agreement_id: Faker.fake::(), + nonce_uuid: uuid::Uuid::now_v7(), + request_id, + deployment_id, + indexer_id, + indexer_url: "http://localhost:8020".parse().expect("Invalid URL"), + terms, + terms_version_hash: None, + }) + .await + .expect("register agreement"); + registry + .apply_reconciliation(&id, true, None) + .await + .expect("accept"); + ids.push(id); + } + let (abandoned, unwanted) = (ids[0], ids[1]); + registry + .mark_indexing_agreement_as_abandoning(&abandoned) + .await + .expect("abandon"); + registry + .mark_indexing_agreement_as_cancelling(&unwanted) + .await + .expect("cancel"); + let held = || async { + registry + .count_abandoned_agreements_holding_slots(&request_id) + .await + .expect("count held slots") + }; + + //* Then + assert_eq!(held().await, 1, "only the abandoned one holds its slot"); + registry + .record_cancel_check(&abandoned, 0, Some(true)) + .await + .expect("seen ended"); + assert_eq!(held().await, 0, "not once the chain shows it ended"); + registry + .record_cancel_check(&abandoned, 0, Some(false)) + .await + .expect("seen live again"); + assert_eq!(held().await, 1); + registry + .mark_replacement_queued(&abandoned) + .await + .expect("replacement queued"); + assert_eq!(held().await, 0, "not once its replacement is queued"); +} + fn fixture_agreement(prefix: u8) -> IndexingAgreementId { let mut bytes = [0u8; 16]; bytes[0] = prefix; From 21abb6a51396cbe96e1577540cb9a89491668d0a Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:23:23 +0100 Subject: [PATCH 12/16] refactor(liveness): split marking a stale agreement from cancelling it The function that did both, with the new database time limit, grew past the project's limit on how hard a function may be to follow. --- .../src/network/service/liveness_checker.rs | 65 ++++++++++--------- 1 file changed, 34 insertions(+), 31 deletions(-) diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index e32c6d15..063ff766 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -564,8 +564,6 @@ pub(crate) async fn replace_abandoned( } /// Mark a stale agreement abandoned and send its cancel; `None`, logged, when it isn't marked. -/// Only the mark is bounded by `db_timeout`, so a hung database can't stall the liveness cycle; -/// a mark that lands after the timeout leaves the agreement for the cancel retry. async fn start_abandoned_cancel( agreement: &IndexingAgreement, registry: &R, @@ -577,35 +575,8 @@ where R: AgreementRegistry + Sync, C: ChainClient, { - let marked = tokio::time::timeout( - db_timeout, - crate::cancel_dispatch::mark_cancelling(registry, agreement, CancelReason::Abandoned), - ) - .await; - match marked { - Ok(Ok(())) => {} - Ok(Err(crate::registry::Error::NoRecordsUpdated)) => { - tracing::debug!( - agreement_id = %agreement.id, - "Stale agreement already ended or being cancelled" - ); - return None; - } - Ok(Err(err)) => { - tracing::error!( - agreement_id = %agreement.id, - error = %err, - "failed to mark stale agreement cancelling, will retry next cycle" - ); - return None; - } - Err(_) => { - tracing::error!( - agreement_id = %agreement.id, - "timeout marking stale agreement cancelling, will retry next cycle" - ); - return None; - } + if !mark_abandoning(agreement, registry, db_timeout).await { + return None; } let started = crate::cancel_dispatch::send_marked_cancel( registry, @@ -624,6 +595,38 @@ where Some(started) } +/// Mark a stale agreement cancelling as abandoned; false, logged, when it isn't marked. Bounded +/// by `db_timeout`, so a hung database can't stall the liveness cycle; a mark that lands after +/// the timeout leaves the agreement for the cancel retry. +async fn mark_abandoning( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) -> bool { + let marked = tokio::time::timeout( + db_timeout, + crate::cancel_dispatch::mark_cancelling(registry, agreement, CancelReason::Abandoned), + ) + .await; + match marked { + Ok(Ok(())) => return true, + Ok(Err(crate::registry::Error::NoRecordsUpdated)) => tracing::debug!( + agreement_id = %agreement.id, + "Stale agreement already ended or being cancelled" + ), + Ok(Err(err)) => tracing::error!( + agreement_id = %agreement.id, + error = %err, + "failed to mark stale agreement cancelling, will retry next cycle" + ), + Err(_) => tracing::error!( + agreement_id = %agreement.id, + "timeout marking stale agreement cancelling, will retry next cycle" + ), + } + false +} + /// 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( From b7c1d18fecfd1b9f9c7b413e278ae3d0f76f3eb9 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:23:37 +0100 Subject: [PATCH 13/16] style(fmt): lay out this branch's changes the way rustfmt wants Formatting only, so the format check in CI passes. --- bin/dipper-service/src/chain_client/client.rs | 5 ++++- .../src/network/service/liveness_checker.rs | 15 ++++++++++++--- .../worker/handlers/reassess_indexing_request.rs | 7 +------ 3 files changed, 17 insertions(+), 10 deletions(-) diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 3592ddd1..842d5021 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -950,7 +950,10 @@ impl AlloyChainClient { let only_endpoint = self.inner.rpc_pool.endpoint_count() < 2; let (noted, seen) = { let mut seen = self.seen_block(); - (seen.advance(block, Instant::now(), only_endpoint), seen.number) + ( + seen.advance(block, Instant::now(), only_endpoint), + seen.number, + ) }; match noted { Noted::Taken => {} diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 063ff766..b467ae58 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -507,8 +507,14 @@ async fn cancel_and_reassess( ); return; } - let Some(started) = - start_abandoned_cancel(agreement, registry, chain_client, agreement_conf, db_timeout).await + let Some(started) = start_abandoned_cancel( + agreement, + registry, + chain_client, + agreement_conf, + db_timeout, + ) + .await else { return; }; @@ -529,7 +535,10 @@ async fn cancel_and_reassess( /// Note that the chain shows a stale agreement still cancelling has ended, so it no longer /// holds its slot in the request and the reassessment queued next fills it. async fn note_seen_ended(agreement: &IndexingAgreement, registry: &R) { - match registry.record_cancel_check(&agreement.id, 0, Some(true)).await { + match registry + .record_cancel_check(&agreement.id, 0, Some(true)) + .await + { Ok(_) | Err(crate::registry::Error::NoRecordsUpdated) => {} Err(err) => tracing::warn!( agreement_id = %agreement.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 c5c86a7f..64271414 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -168,12 +168,7 @@ where // When the in-flight window is known-full a top-up can't proceed, so defer // before paying for selection context and an IISA call. Anything else runs: // shrinks free slots and satisfied requests finish instead of polling. - if should_defer_saturated( - global_in_flight, - global_cap, - slots, - active_agreements.len(), - ) { + if should_defer_saturated(global_in_flight, global_cap, slots, active_agreements.len()) { tracing::debug!( event = "offer_pacing_saturated", indexing_request_id = %indexing_request_id, From 6424686d312c3ce7fc1dea0835a73e554a79abac Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:41:47 +0100 Subject: [PATCH 14/16] fix(reassess): never cancel an agreement because another holds a slot Shrinking the selection by the held slots made the kept agreements look surplus, so a target of 1 with 1 held slot cancelled the healthy one. Held slots now only limit new offers. --- bin/dipper-service/src/registry.rs | 6 +- bin/dipper-service/src/registry/agreement.rs | 8 +- .../src/registry/agreement_stub.rs | 12 +- .../handlers/reassess_indexing_request.rs | 121 ++++++++++++------ dipper-pgregistry/src/postgres.rs | 18 +-- .../tests/it_registry_postgres.rs | 5 +- 6 files changed, 110 insertions(+), 60 deletions(-) diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index f51a0ceb..e194517d 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -733,12 +733,12 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } - async fn count_abandoned_agreements_holding_slots( + async fn get_abandoned_indexers_holding_slots( &self, request_id: &IndexingRequestId, - ) -> RegistryResult { + ) -> RegistryResult> { self.inner - .count_abandoned_agreements_holding_slots(request_id) + .get_abandoned_indexers_holding_slots(request_id) .await .map_err(Into::into) } diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 25e23f64..fde09894 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -497,12 +497,12 @@ pub trait AgreementRegistry { /// Note that an agreement's replacement has been queued. async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()>; - /// How many of a request's slots are held by agreements whose indexer stopped serving them - /// but that may still be paid, so a reassessment fills only the rest. - async fn count_abandoned_agreements_holding_slots( + /// The indexers holding a slot in a request with an agreement they stopped serving that may + /// still be paid, so a reassessment fills only the rest and never picks them again. + async fn get_abandoned_indexers_holding_slots( &self, request_id: &IndexingRequestId, - ) -> RegistryResult; + ) -> RegistryResult>; /// Update the sync progress for an agreement. /// diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index 2e6be4f4..1253c754 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -236,11 +236,11 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_replacement_queued") } - async fn count_abandoned_agreements_holding_slots( + async fn get_abandoned_indexers_holding_slots( &self, _request_id: &IndexingRequestId, - ) -> Result { - unimplemented!("count_abandoned_agreements_holding_slots") + ) -> Result> { + unimplemented!("get_abandoned_indexers_holding_slots") } async fn update_agreement_sync_progress( @@ -575,11 +575,11 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_replacement_queued(self, id).await } - async fn count_abandoned_agreements_holding_slots( + async fn get_abandoned_indexers_holding_slots( &self, request_id: &IndexingRequestId, - ) -> Result { - StubAgreementRegistry::count_abandoned_agreements_holding_slots(self, request_id).await + ) -> Result> { + StubAgreementRegistry::get_abandoned_indexers_holding_slots(self, request_id).await } async fn update_agreement_sync_progress( 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 64271414..1d9f19fc 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -149,12 +149,14 @@ where // An agreement whose indexer stopped serving it keeps its slot while it may still be paid, // so it isn't replaced while both are paid; once it can't be, its replacement is queued. - let held_slots = ctx + // Held slots only limit what is added, never what is cancelled. + let held: HashSet = ctx .registry - .count_abandoned_agreements_holding_slots(indexing_request_id) + .get_abandoned_indexers_holding_slots(indexing_request_id) .await - .map_err(|err| JobError::Fatal(err.into()))?; - let slots = num_candidates.saturating_sub(held_slots); + .map_err(|err| JobError::Fatal(err.into()))? + .into_iter() + .collect(); // In-flight offer counts, fetched once and reused by both the saturated // short-circuit (global figure only) and the per-indexer cap after IISA. @@ -168,7 +170,12 @@ where // When the in-flight window is known-full a top-up can't proceed, so defer // before paying for selection context and an IISA call. Anything else runs: // shrinks free slots and satisfied requests finish instead of polling. - if should_defer_saturated(global_in_flight, global_cap, slots, active_agreements.len()) { + if should_defer_saturated( + global_in_flight, + global_cap, + num_candidates.saturating_sub(held.len()), + active_agreements.len(), + ) { tracing::debug!( event = "offer_pacing_saturated", indexing_request_id = %indexing_request_id, @@ -255,7 +262,7 @@ where // stall new requests until IISA recovers — see README "IISA dependency". let target_selected: Vec = match ctx .iisa - .select_indexers(*deployment_id, slots, &context) + .select_indexers(*deployment_id, *num_candidates, &context) .await { Ok(indexers) => indexers, @@ -277,9 +284,18 @@ where let target_pricing: HashMap = target_selected.iter().map(|s| (s.id, s)).collect(); - // Compute the diff + // Compute the diff. New offers fill only what the agreements kept and the held slots leave + // of the target, in IISA's order, and never for an indexer holding a slot. let to_cancel: HashSet<&IndexerId> = current_ids.difference(&target_ids).collect(); - let to_add: HashSet<&IndexerId> = target_ids.difference(¤t_ids).collect(); + let diff_add_count = target_ids.difference(¤t_ids).count(); + let kept = current_ids.intersection(&target_ids).count(); + let room = num_candidates.saturating_sub(held.len() + kept); + let to_add: HashSet<&IndexerId> = target_selected + .iter() + .map(|s| &s.id) + .filter(|id| !current_ids.contains(*id) && !held.contains(*id)) + .take(room) + .collect(); // Surface the IISA outcome alongside the diff so the operator can tell // "everyone is declined" apart from "we're already synced" without @@ -290,7 +306,7 @@ where indexing_request_id=%indexing_request_id, deployment_id=%deployment_id, requested_num_candidates = num_candidates, - held_slots, + held_slots = held.len(), iisa_returned_count = target_selected.len(), current_active_count = current_ids.len(), to_add_count = to_add.len(), @@ -307,7 +323,7 @@ where // persistently short request. The atomic latch update makes this retry-safe: // a re-run after a later failure sees the latch already set and does not // re-emit. - let in_shortfall = target_selected.len() < slots; + let in_shortfall = target_selected.len() < *num_candidates; let shortfall_changed = ctx .registry .set_indexing_request_shortfall_active(indexing_request_id, in_shortfall) @@ -326,7 +342,7 @@ where *deployment_id, ctx.signer.chain_id(), proto::SubgraphIndexingAgreementNIndexersUnavailable { - agreements_requested: slots as i32, + agreements_requested: *num_candidates as i32, candidates_returned: target_selected.len() as i32, }, ); @@ -350,8 +366,11 @@ where // Offer pacing: only create as many agreements as the network can accept // before the deadline, so a burst does not queue offers that expire unaccepted. + // Adds left out for held slots count as withheld, so the old agreements they would have + // replaced stay until a later pass can replace them. let (capped_to_add, offer_pacing_withheld, reserved_cancel_count) = if to_add.is_empty() { - (HashSet::new(), 0usize, 0usize) + let reserved = reserved_cancel_count(old_to_cancel.len(), diff_add_count, 0); + (HashSet::new(), 0usize, reserved) } else { // Reuse the in-flight counts and global cap fetched at the top of the pass. let per_indexer_cap = ctx.agreement_conf.max_in_flight_offers_per_indexer(); @@ -363,7 +382,7 @@ where per_indexer_cap, global_cap, ); - let reserved = reserved_cancel_count(old_to_cancel.len(), to_add.len(), allowed.len()); + let reserved = reserved_cancel_count(old_to_cancel.len(), diff_add_count, allowed.len()); if withheld > 0 { tracing::info!( event = "offer_pacing_capped", @@ -1044,8 +1063,8 @@ mod lifecycle_event_tests { /// Ids marked CanceledByRequester locally. marked_cancelled: Arc>>, marked_cancelling: Arc>>, - /// Slots held by agreements whose indexer stopped serving them. - held_slots: usize, + /// Indexers holding a slot with an agreement they stopped serving. + held: Vec, } #[async_trait] @@ -1126,11 +1145,11 @@ mod lifecycle_event_tests { ) -> RegistryResult> { Ok(self.active_agreements.clone()) } - async fn count_abandoned_agreements_holding_slots( + async fn get_abandoned_indexers_holding_slots( &self, _request_id: &IndexingRequestId, - ) -> RegistryResult { - Ok(self.held_slots) + ) -> RegistryResult> { + Ok(self.held.clone()) } // Populates the `terminated` event's remaining count. async fn count_accepted_agreements_by_deployment( @@ -1513,7 +1532,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(true), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1552,7 +1571,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1613,16 +1632,17 @@ mod lifecycle_event_tests { async fn fills_only_the_slots_an_abandoned_agreement_does_not_hold() { // The abandoned agreement may still be paid until its cancel lands, so filling its // slot now would pay 2 indexers for it; its replacement is queued once it can't be. - let candidates = [indexer_id(0x21), indexer_id(0x22), indexer_id(0x23)]; + let held = indexer_id(0x20); + let candidates = [held, indexer_id(0x21), indexer_id(0x22), indexer_id(0x23)]; + let url = |id: IndexerId| Url::parse(&format!("https://{id}.example")).unwrap(); let mut snapshot = indexer_urls::Snapshot::new(); for id in candidates { - snapshot.insert(id, indexer_url()); + snapshot.insert(id, url(id)); } - let events = CapturingEventsProducer::new(); let queue = MockQueue::default(); let ctx = build_ctx( MockRegistry { - held_slots: 1, + held: vec![held], ..MockRegistry::default() }, MockIisa { @@ -1630,22 +1650,51 @@ mod lifecycle_event_tests { }, queue.clone(), MockChainClient::default(), - events.clone(), + CapturingEventsProducer::new(), snapshot, ); handle(ctx, &test_message(3)).await.expect("handler ok"); - assert_eq!(queue.proposals.lock().unwrap().len(), 2); - assert!( - !events - .events() - .iter() - .any(|e| matches!(e, CapturedEvent::NIndexersUnavailable { .. })), - "the 2 slots to fill are filled" + let proposed: std::collections::HashSet = + queue.proposals.lock().unwrap().iter().cloned().collect(); + assert_eq!( + proposed, + std::collections::HashSet::from([url(candidates[1]), url(candidates[2])]), + "2 offers, the first in IISA's order, none to the indexer holding a slot" ); } + #[tokio::test] + async fn never_cancels_an_agreement_for_a_slot_another_holds() { + // Target of 1: the healthy agreement stays, though an abandoned one holds a slot too. + let healthy = indexer_id(0x11); + let chain_client = MockChainClient::default(); + let queue = MockQueue::default(); + let registry = MockRegistry { + active_agreements: vec![agreement(healthy, IndexingAgreementStatus::AcceptedOnChain)], + held: vec![indexer_id(0x20)], + ..MockRegistry::default() + }; + let marked_cancelling = Arc::clone(®istry.marked_cancelling); + let ctx = build_ctx( + registry, + MockIisa { + selected: vec![selected(healthy)], + }, + queue.clone(), + chain_client.clone(), + CapturingEventsProducer::new(), + indexer_urls::Snapshot::new(), + ); + + handle(ctx, &test_message(1)).await.expect("handler ok"); + + assert!(chain_client.cancelled.lock().unwrap().is_empty()); + assert!(marked_cancelling.lock().unwrap().is_empty()); + assert!(queue.proposals.lock().unwrap().is_empty()); + } + #[tokio::test] async fn emits_shortfall_when_coverage_drops_below_target() { // Item 9: an indexer leaves the target group and IISA can't replace it, so @@ -1662,7 +1711,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1718,7 +1767,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1784,7 +1833,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1917,7 +1966,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), - held_slots: 0, + held: Vec::new(), }; let ctx = build_ctx( registry, diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index ff3cdb7b..254c4b00 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1951,16 +1951,16 @@ impl PgRegistry { .map_err(Into::into) } - /// Count a request's agreements whose indexer stopped serving them that may still be paid: - /// being cancelled, not yet seen ended on-chain, and with no replacement queued. Each keeps - /// its slot in the request until then, so it isn't replaced while both are paid. - pub async fn count_abandoned_agreements_holding_slots( + /// The indexers of a request's agreements whose indexer stopped serving them that may still + /// be paid: being cancelled, not yet seen ended on-chain, and with no replacement queued. + /// Each keeps its slot in the request until then, so it isn't replaced while both are paid. + pub async fn get_abandoned_indexers_holding_slots( &self, request_id: &IndexingRequestId, - ) -> Result { - let (count,): (i64,) = sqlx::query_as( + ) -> Result, Error> { + let rows: Vec<(PgIndexerId,)> = sqlx::query_as( r#" - SELECT COUNT(*) + SELECT indexer_id FROM dipper_reg_indexing_agreements WHERE indexing_request_id = $1 AND status = $2 @@ -1970,9 +1970,9 @@ impl PgRegistry { ) .bind(request_id) .bind(IndexingAgreementStatus::Cancelling) - .fetch_one(&self.pool) + .fetch_all(&self.pool) .await?; - Ok(usize::try_from(count).unwrap_or_default()) + Ok(rows.into_iter().map(|(id,)| id.0).collect()) } /// Note that an agreement's replacement has been queued. diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index e907b216..02b1c05c 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3557,9 +3557,10 @@ async fn an_abandoned_agreement_holds_its_slot_while_it_may_still_be_paid() { .expect("cancel"); let held = || async { registry - .count_abandoned_agreements_holding_slots(&request_id) + .get_abandoned_indexers_holding_slots(&request_id) .await - .expect("count held slots") + .expect("held slots") + .len() }; //* Then From abb43e55150dcdaa7730b1ba65a2a4a69357921c Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:42:19 +0100 Subject: [PATCH 15/16] fix(liveness): stop waiting on a hung database when noting an end Noting that a stale agreement had ended on-chain had no time limit, so a hung database could hold up the liveness check. It now gives up after the usual database timeout. --- .../src/network/service/liveness_checker.rs | 53 +++++++++++++++---- 1 file changed, 44 insertions(+), 9 deletions(-) diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index b467ae58..6f1a87c6 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -527,24 +527,35 @@ async fn cancel_and_reassess( return; } if started == CancelStarted::NotLive { - note_seen_ended(agreement, registry).await; + note_seen_ended(agreement, registry, db_timeout).await; } replace_abandoned(agreement, registry, worker_queue, db_timeout, queue_timeout).await; } /// Note that the chain shows a stale agreement still cancelling has ended, so it no longer -/// holds its slot in the request and the reassessment queued next fills it. -async fn note_seen_ended(agreement: &IndexingAgreement, registry: &R) { - match registry - .record_cancel_check(&agreement.id, 0, Some(true)) - .await - { - Ok(_) | Err(crate::registry::Error::NoRecordsUpdated) => {} - Err(err) => tracing::warn!( +/// holds its slot in the request and the reassessment queued next fills it. Bounded by +/// `db_timeout`, so a hung database can't stall the liveness cycle. +async fn note_seen_ended( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) { + let noted = tokio::time::timeout( + db_timeout, + registry.record_cancel_check(&agreement.id, 0, Some(true)), + ) + .await; + match noted { + Ok(Ok(_) | Err(crate::registry::Error::NoRecordsUpdated)) => {} + Ok(Err(err)) => tracing::warn!( agreement_id = %agreement.id, error = %err, "Failed to note a stale agreement ended; its replacement may wait for a later reassessment" ), + Err(_) => tracing::warn!( + agreement_id = %agreement.id, + "Timeout noting a stale agreement ended; its replacement may wait for a later reassessment" + ), } } @@ -1005,6 +1016,7 @@ mod tests { calls: MockCalls, already_ending: bool, mark_hangs: bool, + check_hangs: bool, get_request_result: Arc>>>>, } @@ -1015,6 +1027,7 @@ mod tests { calls, already_ending: false, mark_hangs: false, + check_hangs: false, get_request_result: Arc::new(Mutex::new(Some(Ok(Some(request))))), } } @@ -1067,6 +1080,9 @@ mod tests { _failed_attempts: u32, ended: Option, ) -> RegistryResult { + if self.check_hangs { + std::future::pending::<()>().await; + } if ended == Some(true) { self.calls.seen_ended.lock().unwrap().push(*id); } @@ -1693,6 +1709,25 @@ mod tests { assert!(calls.reassessments.lock().unwrap().is_empty()); } + #[tokio::test(start_paused = true)] + async fn still_replaces_an_ended_stale_agreement_when_noting_its_end_hangs() { + let agreement = stale_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry { + check_hangs: true, + ..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_eq!( + calls.reassessments.lock().unwrap().as_slice(), + &[agreement.indexing_request_id] + ); + } + #[tokio::test] async fn test_record_progress_calls_registry() { // Arrange From af2b710985e1f6ca8da8b31b1f2230a646ea8469 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 9 Oct 2026 22:43:52 +0100 Subject: [PATCH 16/16] fix(listener): move past a page of changes that keeps failing to apply A change that could never apply, such as a row that can't be read, held the listener's place for good and stalled every later accept and cancel. It now moves on after 10 minutes. --- .../src/network/service/chain_listener.rs | 129 ++++++++++++++++-- 1 file changed, 116 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 fbf96add..2e8ad5c2 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -75,6 +75,39 @@ const STALL_ERROR_THRESHOLD: u32 = 15; /// Sanity bound on a single tick's drain — not an operating point; /// steady-state drains are 0–1 pages. const MAX_PAGES_PER_DRAIN: u32 = 1000; +/// How long the listener keeps re-reading a page whose changes fail to apply before moving past +/// it: long enough to ride out a database failover, short enough that a change that can never +/// apply doesn't hold up every later accept and cancel. +const MAX_CURSOR_HOLD: Duration = Duration::from_secs(600); + +/// Where the cursor has been held, and since when, before a page whose changes failed to apply. +#[derive(Debug)] +struct CursorHold { + cursor: Cursor, + since: std::time::Instant, +} + +/// Whether to move past a page at `cursor` whose changes failed to apply: once the cursor has +/// been held before it for [`MAX_CURSOR_HOLD`]. Until then the hold is noted, so the page is +/// read again. +fn hold_expired(hold: &mut Option, cursor: &Cursor, now: std::time::Instant) -> bool { + match hold { + Some(held) if held.cursor == *cursor => { + if now.saturating_duration_since(held.since) < MAX_CURSOR_HOLD { + return false; + } + *hold = None; + true + } + _ => { + *hold = Some(CursorHold { + cursor: cursor.clone(), + since: now, + }); + false + } + } +} /// Cap on the cancellation-sweep batch so a backlog drains across polls /// instead of blocking one tick. const SWEEP_BATCH_SIZE: i64 = 1000; @@ -213,6 +246,7 @@ where // Track consecutive failures for adaptive backoff let mut consecutive_failures: u32 = 0; + let mut cursor_hold: Option = None; // Observability: heartbeat and stall detection let mut polls_since_last_event: u64 = 0; @@ -289,6 +323,7 @@ where let outcome = match drain_once( &mut cursor, + &mut cursor_hold, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -419,6 +454,7 @@ struct DrainOutcome { )] async fn drain_once( cursor: &mut Cursor, + cursor_hold: &mut Option, last_persisted_timestamp: &mut Option, last_chain_ts_persist_wall: &mut std::time::Instant, last_subgraph_head: &mut u64, @@ -678,8 +714,20 @@ where // Persist per-page so a crash mid-drain replays at most // one page. Skip the write when neither cursor nor // ratcheted timestamp moved. - let advance_cursor = unapplied.is_empty() && *cursor < new_cursor; - if !unapplied.is_empty() { + let skip = !unapplied.is_empty() && hold_expired(cursor_hold, cursor, now_instant); + if unapplied.is_empty() { + *cursor_hold = None; + } + let advance_cursor = (unapplied.is_empty() || skip) && *cursor < new_cursor; + if skip { + tracing::error!( + event = "chain_listener_changes_skipped", + cursor_block = cursor.block, + agreement_ids = ?unapplied, + "Skipped on-chain changes to agreements that kept failing to apply; they won't \ + be read again until their next change" + ); + } else if !unapplied.is_empty() { tracing::error!( event = "chain_listener_cursor_held", cursor_block = cursor.block, @@ -4165,6 +4213,7 @@ mod tests { let outcome = drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4201,13 +4250,15 @@ mod tests { } } - /// A page whose changes fail to apply keeps the cursor before it, so the next poll - /// re-reads it: no later change to those agreements may come to apply it instead. - #[tokio::test] - async fn test_drain_holds_the_cursor_until_a_failed_page_applies() { + /// Drain 1 page holding a change that accepts an agreement, its apply failing or not, with + /// the cursor held before it since `held_since` if given; the cursor it leaves, and whether + /// the agreement was accepted. + async fn drain_a_page( + apply_fails: &[bool], + held_since: Option, + ) -> (Cursor, Option, bool) { let registry = MockRegistry::new(); let chain_client = MockChainClient::default(); - registry.set_fail_batch(true); let event_source = super::super::chain_events::mock::MockEventSource::new(); event_source.set_latest_block(100); event_source.set_latest_block_timestamp(Some(1_700_000_000)); @@ -4225,6 +4276,10 @@ mod tests { canceled_tx: String::new(), }]); let mut cursor = Cursor::genesis(); + let mut cursor_hold = held_since.map(|since| CursorHold { + cursor: Cursor::genesis(), + since, + }); let mut last_persisted_timestamp: Option = None; let mut last_chain_ts_persist_wall = std::time::Instant::now(); let mut last_subgraph_head: u64 = 0; @@ -4232,10 +4287,11 @@ mod tests { let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); - for fails in [true, false] { - registry.set_fail_batch(fails); + for fails in apply_fails { + registry.set_fail_batch(*fails); drain_once( &mut cursor, + &mut cursor_hold, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4253,13 +4309,57 @@ mod tests { ) .await .expect("subgraph fetch succeeded"); - if fails { - assert_eq!(cursor, Cursor::genesis(), "held before the failed page"); - } } + let accepted = registry.was_marked_accepted_on_chain(&agreement_id); + (cursor, cursor_hold, accepted) + } - assert!(registry.was_marked_accepted_on_chain(&agreement_id)); + /// A page whose changes fail to apply keeps the cursor before it, so the next poll + /// re-reads it: no later change to those agreements may come to apply it instead. + #[tokio::test] + async fn test_drain_holds_the_cursor_until_a_failed_page_applies() { + let (cursor, hold, _) = drain_a_page(&[true], None).await; + assert_eq!(cursor, Cursor::genesis(), "held before the failed page"); + assert!(hold.is_some()); + + let (cursor, hold, accepted) = drain_a_page(&[true, false], None).await; + assert!(accepted); + assert!(cursor > Cursor::genesis()); + assert!(hold.is_none()); + } + + /// A page that keeps failing is passed over once the cursor has been held long enough, so + /// it can't hold up every later change. + #[tokio::test] + async fn test_drain_moves_past_a_page_that_keeps_failing() { + let long_ago = std::time::Instant::now() + .checked_sub(MAX_CURSOR_HOLD + Duration::from_secs(1)) + .expect("instant"); + + let (cursor, hold, accepted) = drain_a_page(&[true], Some(long_ago)).await; + + assert!(!accepted); assert!(cursor > Cursor::genesis()); + assert!(hold.is_none()); + } + + #[test] + fn holds_the_cursor_before_a_failing_page_for_a_while() { + let now = std::time::Instant::now(); + let page = Cursor::genesis(); + let mut hold = None; + + assert!( + !hold_expired(&mut hold, &page, now), + "the first failure holds" + ); + assert!(!hold_expired( + &mut hold, + &page, + now + MAX_CURSOR_HOLD - Duration::from_secs(1) + )); + assert!(hold_expired(&mut hold, &page, now + MAX_CURSOR_HOLD)); + assert!(hold.is_none()); } /// Drives a hostile timestamp drift through `drain_once` and asserts @@ -4290,6 +4390,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4317,6 +4418,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4420,6 +4522,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head,