From 95a8b226f21ccf1637aadfd2ea6aea4c654804ad Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:56:33 +0100 Subject: [PATCH 01/13] fix(listener): read the chain for an ended agreement in a worker job For an agreement dipper ended that the subgraph showed accepted, the listener read the chain mid-page, so in an RPC outage every other change on the page waited nearly a minute an endpoint. It now queues a worker job for that read, once per change, and the job records the accept. --- .../handlers/indexing_requests.rs | 9 + bin/dipper-service/src/main.rs | 1 + .../src/network/service/cancel_retry.rs | 8 + .../src/network/service/chain_listener.rs | 479 ++++++++++-------- .../src/network/service/expiration.rs | 8 + .../src/network/service/liveness_checker.rs | 8 + bin/dipper-service/src/worker/handlers.rs | 4 +- .../cancel_rejected_agreement_on_chain.rs | 245 +++++++-- .../handlers/reassess_indexing_request.rs | 9 + .../send_indexing_agreement_proposal.rs | 9 + bin/dipper-service/src/worker/messages.rs | 8 +- bin/dipper-service/src/worker/service.rs | 1 + .../src/worker/service_queue.rs | 31 +- 13 files changed, 565 insertions(+), 255 deletions(-) diff --git a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs index b846dd8e..af36ee22 100644 --- a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs +++ b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs @@ -364,6 +364,15 @@ mod tests { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 17810755..2b4fce79 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -539,6 +539,7 @@ pub async fn main() -> anyhow::Result<()> { chain_listener_notify: chain_listener_notify.clone(), subgraph_indexing_agreements_events_emitter: subgraph_indexing_agreements_events_emitter.clone(), + worker_queue: worker_handle.queue().clone(), }; let (handle, service) = network::service::chain_listener::new(ctx); Some((handle, service)) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 3f3fa6b9..43d148ac 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -758,6 +758,14 @@ mod tests { self.0.lock().unwrap().push(indexing_request_id); Ok(dipper_pgmq::JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 1dd12227..efa3a51c 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -17,7 +17,8 @@ //! | AcceptedOnChain | CanceledBy* | mark canceled (requester vs indexer from the CanceledBy* state) | //! | Expired | Accepted | mark AcceptedOnChain (recovery), run pending cancellations | //! | Expired | CanceledBy* | recover to AcceptedOnChain, run pending cancellations, then cancel | -//! | Rejected | Accepted | queue cancel-on-chain (adversarial path, dead code under proposal-first) | +//! | Rejected | Accepted | queue a chain check, which reopens its cancel if live (adversarial path) | +//! | CanceledByRequester | Accepted | queue a chain check, which reopens its cancel if live | //! | Canceled* | * | no-op (terminal) | //! //! ## Data Source @@ -57,6 +58,7 @@ use crate::{ AgreementRegistry, CancelKind, IndexingAgreement, IndexingAgreementStatus, PendingCancellationRegistry, ReconciliationItem, }, + worker::service::{ChainAccept, JobPriority, WorkerQueue}, }; /// Idle interval used when no `Created` agreements are awaiting acceptance. @@ -135,7 +137,7 @@ impl Handle { } /// Context required by the chain listener service -pub struct Ctx { +pub struct Ctx { /// Registry for querying and updating agreements pub registry: R, /// Chain event source (subgraph) @@ -153,6 +155,8 @@ pub struct Ctx { /// Emits `accepted` and `terminated` lifecycle events as on-chain state is reconciled. pub subgraph_indexing_agreements_events_emitter: Arc, + /// Queues the chain check of an agreement dipper ended that the subgraph shows accepted. + pub worker_queue: W, } /// State persisted in the database @@ -177,11 +181,12 @@ pub struct ChainListenerState { clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] -pub fn new(ctx: Ctx) -> (Handle, impl Future>) +pub fn new(ctx: Ctx) -> (Handle, impl Future>) where R: AgreementRegistry + ChainListenerStateRegistry + PendingCancellationRegistry + Send + Sync, E: ChainEventSource, T: ChainClient + Send + Sync, + W: WorkerQueue + Send + Sync, { let (tx_stop, mut rx_stop) = mpsc::channel(1); @@ -193,6 +198,7 @@ where config, chain_listener_notify, subgraph_indexing_agreements_events_emitter, + worker_queue, } = ctx; let service = async move { @@ -269,6 +275,7 @@ where // prematurely. Persisted in memory only; on restart it resets // and the first poll's bound is strict. let mut last_chain_ts_persist_wall: std::time::Instant = std::time::Instant::now(); + let mut live_checks = LiveChecks::new(worker_queue); loop { tokio::select! { @@ -336,6 +343,7 @@ where bypass_chain_clock_defenses, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, &agreement_conf, @@ -453,7 +461,7 @@ struct DrainOutcome { clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] -async fn drain_once( +async fn drain_once( cursor: &mut Cursor, cursor_hold: &mut Option, last_persisted_timestamp: &mut Option, @@ -467,6 +475,7 @@ async fn drain_once( bypass_chain_clock_defenses: bool, registry: &R, chain_client: &T, + live_checks: &mut LiveChecks, event_source: &E, rx_stop: &mut mpsc::Receiver<()>, config: &crate::config::IndexingAgreementConfig, @@ -475,6 +484,7 @@ where R: AgreementRegistry + ChainListenerStateRegistry + PendingCancellationRegistry + Send + Sync, E: ChainEventSource, T: ChainClient + Send + Sync, + W: WorkerQueue + Sync, { // Read-side cursor backs off by `reorg_buffer_blocks` so a // reorg that moves a state change across the boundary is still @@ -491,6 +501,7 @@ where } else { cursor.clone() }; + live_checks.forget_before(effective_cursor.block); let drain_start = std::time::Instant::now(); let mut pages_per_tick: u32 = 0; @@ -619,7 +630,7 @@ where } let agreement = agreements_by_id.remove(&snapshot.agreement_id); - match prepare_reconciliation(&snapshot, agreement, registry, chain_client).await { + match prepare_reconciliation(&snapshot, agreement, registry, live_checks).await { Ok(Some(prep)) => prepared.push(prep), Ok(None) => {} Err(err) => { @@ -861,15 +872,15 @@ fn apply_chain_ts_drift_cap( clippy::cognitive_complexity, reason = "predates this lint; fix when next touched" )] -async fn prepare_reconciliation( +async fn prepare_reconciliation( snapshot: &AgreementStateSnapshot, agreement: Option, registry: &R, - chain_client: &T, + live_checks: &mut LiveChecks, ) -> anyhow::Result> where R: AgreementRegistry + Sync, - T: ChainClient, + W: WorkerQueue + Sync, { tracing::debug!( agreement_id = %snapshot.agreement_id, @@ -899,31 +910,17 @@ where // (race with gRPC rejection or stale read), not the indexer. // Skipped when remote is already canceled because the cancel tx // would revert against canceled contract state. - if matches!(agreement.status, IndexingAgreementStatus::Rejected) - && snapshot.state.reached_accepted() - && !snapshot.state.is_canceled() - { - tracing::warn!( - agreement_id = %agreement.id, - indexer = %snapshot.indexer, - "Rejected agreement accepted on-chain, cancelling it" - ); - // It never passes AcceptedOnChain, so its accept is recorded here for its end to be - // announced; one left Rejected, as the chain shows it already ended, gets none. A failed - // write gets the snapshot read again, which finds it cancelling and records the accept. - let reopened = - crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; - if reopened { - registry - .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) - .await - .context("failed to record the on-chain accept of a rejected agreement")?; + if ended_but_accepted(snapshot, &agreement) { + if agreement.status == IndexingAgreementStatus::Rejected { + tracing::warn!( + agreement_id = %agreement.id, + indexer = %snapshot.indexer, + "Rejected agreement accepted on-chain, checking whether to cancel it" + ); } + live_checks.queue(snapshot).await?; } - - let reopened = - reopen_if_cancelled_but_accepted(snapshot, &agreement, registry, chain_client).await?; - record_accept_of_cancelling(snapshot, &agreement, reopened, registry).await?; + record_accept_of_cancelling(snapshot, &agreement, registry).await?; // Both transitions are applied atomically downstream so the // Accept-then-Cancel-in-one-snapshot path can't leak an intermediate @@ -1012,42 +1009,75 @@ async fn record_accept_and_cancel_from_chain( .context("failed to record the on-chain accept and cancel of a cancelled agreement") } -/// Safety net for an agreement dipper cancelled whose offer the indexer accepted -/// anyway, such as one that landed after dipper's cancel. Nothing else would end -/// it: reconciliation ignores an accept on a cancelled row. It goes back to -/// `Cancelling` for the cancel retry, unless the chain shows it already ended. Returns whether it -/// moved it back. -async fn reopen_if_cancelled_but_accepted( - snapshot: &AgreementStateSnapshot, - agreement: &IndexingAgreement, - registry: &R, - chain_client: &T, -) -> anyhow::Result -where - R: AgreementRegistry + Sync, - T: ChainClient, -{ - if agreement.status == IndexingAgreementStatus::CanceledByRequester - && snapshot.state.reached_accepted() +/// Whether dipper rejected or cancelled an agreement the subgraph shows accepted and not ended, +/// such as an offer that landed after dipper's cancel. Nothing else would end it, as +/// reconciliation ignores an accept on an ended row, so a job checks the chain for it. +fn ended_but_accepted(snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement) -> bool { + matches!( + agreement.status, + IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester + ) && snapshot.state.reached_accepted() && !snapshot.state.is_canceled() - { - return Ok( - crate::cancel_dispatch::reopen_if_live(registry, chain_client, agreement).await?, - ); +} + +/// Queues the chain check of an agreement dipper ended that the subgraph shows accepted, once +/// per change, as the reorg buffer and a held cursor read a change again. The job runs off the +/// drain, so a slow chain read can't hold up the rest of the page. +struct LiveChecks { + queue: W, + /// The changes checks were queued for, by agreement and block. + queued: std::collections::HashSet<(IndexingAgreementId, u64)>, +} + +impl LiveChecks { + fn new(queue: W) -> Self { + Self { + queue, + queued: std::collections::HashSet::new(), + } + } + + /// Queue the check for this change, unless it is already queued. + async fn queue(&mut self, snapshot: &AgreementStateSnapshot) -> anyhow::Result<()> { + let change = (snapshot.agreement_id, snapshot.last_state_change_block); + if self.queued.contains(&change) { + tracing::debug!( + agreement_id = %snapshot.agreement_id, + "On-chain check already queued for this change" + ); + return Ok(()); + } + let accept = (snapshot.accepted_at != 0).then(|| ChainAccept { + at: snapshot.accepted_at, + tx: snapshot.accepted_tx.clone(), + }); + self.queue + .cancel_rejected_agreement_on_chain( + snapshot.agreement_id, + accept, + JobPriority::Background, + ) + .await + .context("failed to queue the on-chain check of an agreement dipper ended")?; + self.queued.insert(change); + Ok(()) + } + + /// Forget the changes before `block`, which the drain no longer reads. + fn forget_before(&mut self, block: u64) { + self.queued.retain(|&(_, changed_at)| changed_at >= block); } - Ok(false) } -/// An agreement dipper is cancelling, or has just moved back to cancelling, stays `Cancelling` -/// when the chain shows it accepted, so its accept is recorded here, to be announced with its -/// end once the cancel lands. A withdrawn offer reads as cancelled with no accept time. +/// An agreement dipper is cancelling stays `Cancelling` when the chain shows it accepted, so its +/// accept is recorded here, to be announced with its end once the cancel lands. A withdrawn offer +/// reads as cancelled with no accept time. async fn record_accept_of_cancelling( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, - reopened: bool, registry: &R, ) -> anyhow::Result<()> { - if (agreement.status != IndexingAgreementStatus::Cancelling && !reopened) + if agreement.status != IndexingAgreementStatus::Cancelling || !snapshot.state.reached_accepted() || snapshot.accepted_at == 0 { @@ -1129,32 +1159,35 @@ where /// Reconcile a single agreement snapshot against dipper's local DB. /// /// Compares the snapshot's remote state to the local `IndexingAgreementStatus` -/// and applies whatever transitions the diff implies. See the module-level -/// transition table for the full mapping. +/// and applies whatever transitions the diff implies, returning the chain checks it queued. +/// See the module-level transition table for the full mapping. #[cfg(test)] async fn reconcile_agreement( snapshot: &AgreementStateSnapshot, registry: &R, chain_client: &T, config: &crate::config::IndexingAgreementConfig, -) -> anyhow::Result<()> +) -> anyhow::Result> where R: AgreementRegistry + PendingCancellationRegistry + Sync, T: ChainClient, { + let mut live_checks = LiveChecks::new(tests::MockWorkerQueue::default()); let agreement = registry .get_indexing_agreement_by_id(&snapshot.agreement_id) .await?; - let Some(prep) = prepare_reconciliation(snapshot, agreement, registry, chain_client).await? - else { - return Ok(()); + let prep = prepare_reconciliation(snapshot, agreement, registry, &mut live_checks).await?; + let queued = live_checks.queue.queued(); + let Some(prep) = prep else { + return Ok(queued); }; let outcome = registry .apply_reconciliation(&prep.agreement.id, prep.item.apply_accept, prep.item.cancel) .await?; - finalize_reconciliation(&prep, outcome, registry, chain_client, config).await + finalize_reconciliation(&prep, outcome, registry, chain_client, config).await?; + Ok(queued) } /// Execute pending cancellations linked to a newly-accepted agreement. @@ -2705,6 +2738,80 @@ mod tests { } } + /// A chain check the listener queued: the agreement, and the accept it carried. + pub(super) type QueuedCheck = (IndexingAgreementId, Option); + + /// Records the chain checks queued; queuing fails while `fail` is set. + #[derive(Clone, Default)] + pub(super) struct MockWorkerQueue { + checks: Arc>>, + fail: Arc, + } + + impl MockWorkerQueue { + pub(super) fn queued(&self) -> Vec { + self.checks.lock().unwrap().clone() + } + } + + #[async_trait::async_trait] + impl WorkerQueue for MockWorkerQueue { + async fn send_indexing_agreement_proposal( + &self, + _candidate_url: url::Url, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn reassess_indexing_request( + &self, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _num_candidates: usize, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + if self.fail.load(std::sync::atomic::Ordering::SeqCst) { + anyhow::bail!("queue unavailable"); + } + self.checks.lock().unwrap().push((agreement_id, accept)); + Ok(dipper_pgmq::JobId::default()) + } + + async fn submit_offer( + &self, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _indexer_url: url::Url, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + } + + fn snapshot_accept() -> Option { + Some(ChainAccept { + at: TEST_ACCEPTED_AT, + tx: TEST_ACCEPTED_TX.to_string(), + }) + } + // -- reconcile_agreement tests -- #[tokio::test] @@ -2922,52 +3029,42 @@ mod tests { } #[tokio::test] - async fn test_reconcile_reopens_the_cancel_of_a_rejected_agreement_live_on_chain() { - let registry = MockRegistry::new(); - let chain_client = MockChainClient { - live_until_cancelled: true, - ..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_marked_accepted_on_chain(&agreement_id)); - 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); + async fn test_reconcile_queues_a_chain_check_for_an_ended_agreement_accepted_on_chain() { + // Dipper rejected or cancelled it, but the indexer accepted it on-chain anyway, such as an + // offer that landed after dipper's cancel. The chain is read by the queued job, not here, + // so a slow read can't hold up the page, and the job records the accept if it reopens it. + for status in [ + IndexingAgreementStatus::Rejected, + IndexingAgreementStatus::CanceledByRequester, + ] { + let registry = MockRegistry::new(); + let chain_client = live_chain(); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, status); - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - let result = reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await; + let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); + let queued = reconcile_agreement( + &snapshot, + ®istry, + &chain_client, + test_agreement_conf().as_ref(), + ) + .await + .expect("reconcile ok"); - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); - assert!(registry.audit_writes().is_empty()); + assert_eq!(queued, vec![(agreement_id, snapshot_accept())], "{status}"); + assert_eq!( + chain_client.reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "{status}" + ); + assert!(!registry.was_reopened(&agreement_id), "{status}"); + assert!(registry.audit_writes().is_empty(), "{status}"); + assert!( + !registry.was_marked_accepted_on_chain(&agreement_id), + "{status}" + ); + } } #[tokio::test] @@ -2999,8 +3096,7 @@ mod tests { ) .await; - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); + assert!(result.expect("reconcile ok").is_empty(), "no check queued"); assert!(registry.was_marked_canceled_by_requester(&agreement_id)); assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); } @@ -3134,65 +3230,11 @@ mod tests { ) .await; - assert!(result.is_ok()); + // Already canceled on-chain: must not queue a fresh check. + assert!(result.expect("reconcile ok").is_empty()); assert!(registry.was_marked_canceled_by_requester(&agreement_id)); assert!(!registry.was_marked_canceled_by_indexer(&agreement_id)); assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); - // Already canceled on-chain: must not queue a fresh cancel job. - assert!(!registry.was_reopened(&agreement_id)); - } - - #[tokio::test] - async fn test_reconcile_reopens_the_cancel_of_a_cancelled_agreement_live_on_chain() { - // Dipper cancelled the agreement locally, but the indexer accepted its offer - // (for example one that landed after dipper's cancel). Nothing else would end - // it, so it goes back to cancelling for the cancel retry to end. - let registry = MockRegistry::new(); - let chain_client = MockChainClient { - live_until_cancelled: true, - ..MockChainClient::default() - }; - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::CanceledByRequester); - - 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.was_marked_accepted_on_chain(&agreement_id)); - assert!( - registry.audit_writes().contains(&("accept", agreement_id)), - "its accept is recorded, so the cancel retry treats it as paying" - ); - } - - #[tokio::test] - async fn test_reconcile_leaves_a_cancelled_agreement_the_chain_shows_ended() { - // The subgraph can still report an agreement accepted for a few polls after - // dipper's cancel lands; reopening it would only bring it back to cancelling. - let registry = MockRegistry::new(); - let chain_client = MockChainClient::default(); - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::CanceledByRequester); - - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await - .expect("reconcile ok"); - - assert!(!registry.was_reopened(&agreement_id)); } #[tokio::test] @@ -3211,8 +3253,7 @@ mod tests { ) .await; - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); + assert!(result.expect("reconcile ok").is_empty()); } #[tokio::test] @@ -3972,6 +4013,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4055,6 +4097,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4135,6 +4178,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4198,6 +4242,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); let outcome = drain_once( &mut cursor, @@ -4213,6 +4258,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4274,6 +4320,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); for fails in apply_fails { registry.set_fail_batch(*fails); @@ -4291,6 +4338,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4332,13 +4380,18 @@ mod tests { } /// Drain a page holding a change in `state` to an agreement in `status` 3 times: with its - /// audit writes failing, then working, then from the start again, as the reorg buffer - /// re-reads it. The agreement, the registry and the cursor after each drain. + /// audit writes and job queuing failing, then working, then from the start again, as the + /// reorg buffer re-reads it. The agreement, the registry, the queue and each drain's cursor. async fn drain_a_page_whose_audit_write_fails( status: IndexingAgreementStatus, state: AgreementState, - chain_client: &MockChainClient, - ) -> (IndexingAgreementId, MockRegistry, Vec) { + ) -> ( + IndexingAgreementId, + MockRegistry, + MockWorkerQueue, + Vec, + ) { + let chain_client = live_chain(); let registry = MockRegistry::new(); let event_source = super::super::chain_events::mock::MockEventSource::new(); event_source.set_latest_block(100); @@ -4354,6 +4407,8 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let queue = MockWorkerQueue::default(); + let mut live_checks = LiveChecks::new(queue.clone()); let mut cursors = Vec::new(); for (writes_fail, from_start) in [(true, false), (false, false), (false, true)] { @@ -4362,6 +4417,9 @@ mod tests { mock.fail_accept_audit = writes_fail; mock.fail_cancel_audit = writes_fail; } + queue + .fail + .store(writes_fail, std::sync::atomic::Ordering::SeqCst); if from_start { cursor = Cursor::genesis(); } @@ -4378,7 +4436,8 @@ mod tests { 10, false, ®istry, - chain_client, + &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4387,7 +4446,12 @@ mod tests { .expect("subgraph fetch succeeded"); cursors.push(cursor.clone()); } - (agreement_id, registry, cursors) + assert_eq!( + chain_client.reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "the drain reads nothing from the chain" + ); + (agreement_id, registry, queue, cursors) } /// The end of an agreement dipper already ended is its last change on-chain, so if recording @@ -4395,10 +4459,9 @@ mod tests { /// repeats the write, which keeps what is already on record. #[tokio::test] async fn test_drain_rereads_an_accept_and_cancel_that_failed_to_record() { - let (id, registry, cursors) = drain_a_page_whose_audit_write_fails( + let (id, registry, _, cursors) = drain_a_page_whose_audit_write_fails( IndexingAgreementStatus::CanceledByRequester, AgreementState::CanceledByPayer, - &MockChainClient::default(), ) .await; @@ -4417,30 +4480,43 @@ mod tests { assert!(!registry.was_marked_canceled_by_requester(&id)); } - /// The same holds for the accept of an agreement being cancelled, including a rejected or - /// cancelled one moved back to cancelling: read again, it is cancelling, so its accept is - /// recorded without moving it back a second time. + /// The same holds for the accept of an agreement being cancelled. #[tokio::test] async fn test_drain_rereads_the_accept_of_an_agreement_being_cancelled_that_failed_to_record() { - for (status, chain_client) in [ - ( - IndexingAgreementStatus::Cancelling, - MockChainClient::default(), - ), - (IndexingAgreementStatus::CanceledByRequester, live_chain()), - (IndexingAgreementStatus::Rejected, live_chain()), + let (id, registry, _, cursors) = drain_a_page_whose_audit_write_fails( + IndexingAgreementStatus::Cancelling, + AgreementState::Accepted, + ) + .await; + + assert_eq!( + cursors[0], + Cursor::genesis(), + "held before the failed write" + ); + assert!( + cursors[1..] + .iter() + .all(|cursor| *cursor > Cursor::genesis()) + ); + assert_eq!(registry.audit_writes(), vec![("accept", id); 2]); + } + + /// A rejected or cancelled agreement accepted on-chain is read again until its chain check + /// is queued, and only queued once, however often the change is read. + #[tokio::test] + async fn test_drain_queues_the_chain_check_of_an_ended_agreement_once() { + for status in [ + IndexingAgreementStatus::Rejected, + IndexingAgreementStatus::CanceledByRequester, ] { - let (id, registry, cursors) = drain_a_page_whose_audit_write_fails( - status, - AgreementState::Accepted, - &chain_client, - ) - .await; + let (id, registry, queue, cursors) = + drain_a_page_whose_audit_write_fails(status, AgreementState::Accepted).await; assert_eq!( cursors[0], Cursor::genesis(), - "{status}: held before the failed write" + "{status}: held before the failed queuing" ); assert!( cursors[1..] @@ -4448,12 +4524,9 @@ mod tests { .all(|cursor| *cursor > Cursor::genesis()), "{status}" ); - assert_eq!(registry.audit_writes(), vec![("accept", id); 2], "{status}"); - assert_eq!( - registry.state.lock().unwrap().reopened.len(), - usize::from(status != IndexingAgreementStatus::Cancelling), - "{status}" - ); + assert_eq!(queue.queued(), vec![(id, snapshot_accept())], "{status}"); + assert!(!registry.was_reopened(&id), "{status}"); + assert!(registry.audit_writes().is_empty(), "{status}"); } } @@ -4501,6 +4574,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); drain_once( &mut cursor, @@ -4516,6 +4590,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4544,6 +4619,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4633,6 +4709,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); drain_once( &mut cursor, @@ -4648,6 +4725,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4697,6 +4775,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); diff --git a/bin/dipper-service/src/network/service/expiration.rs b/bin/dipper-service/src/network/service/expiration.rs index 12802fcf..c62541b6 100644 --- a/bin/dipper-service/src/network/service/expiration.rs +++ b/bin/dipper-service/src/network/service/expiration.rs @@ -538,6 +538,14 @@ mod tests { ) -> anyhow::Result { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index ac809c90..59bb18d6 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -1328,6 +1328,14 @@ mod tests { self.calls.reassessments.lock().unwrap().push(req_id); Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/handlers.rs b/bin/dipper-service/src/worker/handlers.rs index fe545d82..bfab3b83 100644 --- a/bin/dipper-service/src/worker/handlers.rs +++ b/bin/dipper-service/src/worker/handlers.rs @@ -6,8 +6,8 @@ mod send_indexing_agreement_proposal; mod submit_offer; pub use cancel_rejected_agreement_on_chain::{ - Ctx as CancelRejectedAgreementOnChainCtx, Message as CancelRejectedAgreementOnChain, - handle as cancel_rejected_agreement_on_chain, + ChainAccept, Ctx as CancelRejectedAgreementOnChainCtx, + Message as CancelRejectedAgreementOnChain, handle as cancel_rejected_agreement_on_chain, }; pub use reassess_indexing_request::{ Ctx as ReassessIndexingRequestCtx, Message as ReassessIndexingRequest, diff --git a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs index c6d5735c..c640a0e4 100644 --- a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs +++ b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs @@ -1,6 +1,6 @@ -//! Kept so jobs queued before an upgrade still run. The chain listener now moves an -//! agreement dipper rejected or cancelled that went live on-chain anyway back into -//! `Cancelling` itself, and the cancel retry ends it; this job does the same. +//! Moves an agreement dipper rejected or cancelled that is live on-chain after all back into +//! `Cancelling`, with the accept the chain listener saw on record. The listener queues this +//! rather than reading the chain itself, so a slow read can't hold up the rest of its page. use std::time::Duration; @@ -13,19 +13,38 @@ use crate::{ worker::result::{JobError, JobResult}, }; +/// How long a job whose database read or write failed waits before running again. +const RETRY_DELAY: Duration = Duration::from_secs(30); + pub struct Ctx { pub registry: R, pub chain_client: T, } -/// Cancel on-chain an agreement dipper rejected or cancelled that was accepted anyway. +/// Check on-chain an agreement dipper rejected or cancelled that was accepted anyway. #[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct Message { pub agreement_id: IndexingAgreementId, + /// The accept the chain listener saw; absent from jobs queued before it was carried. + #[serde(default)] + pub accept: Option, +} + +/// When an agreement was accepted on-chain, and in which transaction. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct ChainAccept { + pub at: u64, + pub tx: String, } /// Hand an agreement dipper rejected or cancelled that is live on-chain to the cancel retry. -pub async fn handle(ctx: Ctx, Message { agreement_id }: &Message) -> JobResult<()> +pub async fn handle( + ctx: Ctx, + Message { + agreement_id, + accept, + }: &Message, +) -> JobResult<()> where R: AgreementRegistry + Sync, T: ChainClient, @@ -34,21 +53,41 @@ where .registry .get_indexing_agreement_by_id(agreement_id) .await - .map_err(|err| JobError::Fatal(err.into()))? + .map_err(|err| retry(agreement_id, "read", err.into()))? else { - tracing::warn!(%agreement_id, "Agreement not found for on-chain cancellation"); + tracing::warn!(%agreement_id, "Agreement not found for on-chain check"); return Ok(()); }; - if !matches!( - agreement.status, - IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester - ) { + let cancelling = match agreement.status { + IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester => { + reopen_if_live(&ctx.registry, &ctx.chain_client, &agreement) + .await + .map_err(|err| retry(agreement_id, "reopen the cancel of", err.into()))? + } + // Moved back by an earlier run of this job, or a copy of it, so only its accept is left. + IndexingAgreementStatus::Cancelling => true, + _ => return Ok(()), + }; + if !cancelling { return Ok(()); } - reopen_if_live(&ctx.registry, &ctx.chain_client, &agreement) + let Some(accept) = accept else { + return Ok(()); + }; + ctx.registry + .record_accepted_audit(agreement_id, accept.at, &accept.tx) .await - .map(|_| ()) - .map_err(|err| JobError::Retryable(err.into(), Duration::from_secs(30))) + .map_err(|err| retry(agreement_id, "record the on-chain accept of", err.into())) +} + +/// Log a failed database read or write and run the job again later. +fn retry(agreement_id: &IndexingAgreementId, action: &str, err: anyhow::Error) -> JobError { + tracing::warn!( + %agreement_id, + error = format!("{err:#}"), + "Failed to {action} an agreement dipper ended that was accepted on-chain; will retry" + ); + JobError::Retryable(err, RETRY_DELAY) } #[cfg(test)] @@ -66,9 +105,23 @@ mod tests { registry::{IndexingAgreement, StubAgreementRegistry}, }; - struct MockRegistry { - agreement: IndexingAgreement, - reopened: Arc>>, + #[derive(Default)] + struct State { + agreement: Option, + /// What each reopen was told about the chain. + reopened: Vec, + accepts: Vec, + fail_read: bool, + fail_accept: bool, + } + + #[derive(Clone, Default)] + struct MockRegistry(Arc>); + + fn db_down() -> crate::registry::Error { + crate::registry::Error::BackendError(dipper_pgregistry::Error::DbError( + sqlx::Error::Protocol("simulated outage".into()), + )) } #[async_trait] @@ -77,20 +130,44 @@ mod tests { &self, _id: &IndexingAgreementId, ) -> crate::registry::Result> { - Ok(Some(self.agreement.clone())) + let state = self.0.lock().unwrap(); + if state.fail_read { + return Err(db_down()); + } + Ok(state.agreement.clone()) } async fn reopen_indexing_agreement_cancel( &self, - id: &IndexingAgreementId, - _seen_live: bool, + _id: &IndexingAgreementId, + seen_live: bool, ) -> crate::registry::Result<()> { - self.reopened.lock().unwrap().push(*id); + let mut state = self.0.lock().unwrap(); + let agreement = state.agreement.as_mut().expect("agreement"); + agreement.status = IndexingAgreementStatus::Cancelling; + state.reopened.push(seen_live); + Ok(()) + } + async fn record_accepted_audit( + &self, + _agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, + ) -> crate::registry::Result<()> { + let mut state = self.0.lock().unwrap(); + if state.fail_accept { + return Err(db_down()); + } + state.accepts.push(ChainAccept { + at: accepted_at, + tx: accepted_tx.to_string(), + }); Ok(()) } } + /// Every agreement reads as live or not, or the read fails when `live` is `None`. struct MockChain { - live: bool, + live: Option, } #[async_trait] @@ -128,30 +205,39 @@ mod tests { &self, _agreement_id: &[u8; 16], ) -> Result { - Ok(AgreementOnChain::live_if(self.live)) + self.live + .map(AgreementOnChain::live_if) + .ok_or_else(|| ChainClientError::RpcError(anyhow::anyhow!("every endpoint failed"))) } async fn latest_block_timestamp(&self) -> Result { unimplemented!() } } - async fn run(status: IndexingAgreementStatus, live: bool) -> Vec { - let reopened = Arc::new(Mutex::new(Vec::new())); - let agreement = agreement(status, Some(vec![7u8; 32])); + fn chain_accept() -> ChainAccept { + ChainAccept { + at: 1_700_000_001, + tx: "0x11".to_string(), + } + } + + fn registry(status: IndexingAgreementStatus) -> MockRegistry { + let registry = MockRegistry::default(); + registry.0.lock().unwrap().agreement = Some(agreement(status, Some(vec![7u8; 32]))); + registry + } + + async fn run_job(registry: &MockRegistry, live: Option) -> JobResult<()> { + let agreement_id = registry.0.lock().unwrap().agreement.as_ref().unwrap().id; let message = Message { - agreement_id: agreement.id, + agreement_id, + accept: Some(chain_accept()), }; let ctx = Ctx { - registry: MockRegistry { - agreement, - reopened: Arc::clone(&reopened), - }, + registry: registry.clone(), chain_client: MockChain { live }, }; - - handle(ctx, &message).await.expect("job ok"); - - reopened.lock().unwrap().clone() + handle(ctx, &message).await } #[tokio::test] @@ -160,21 +246,88 @@ mod tests { IndexingAgreementStatus::Rejected, IndexingAgreementStatus::CanceledByRequester, ] { - assert_eq!(run(status, true).await.len(), 1, "{status}"); + let registry = registry(status); + + run_job(®istry, Some(true)).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "{status}"); + assert_eq!( + state.accepts, + vec![chain_accept()], + "{status}: its accept is recorded, so the cancel retry treats it as paying" + ); } } #[tokio::test] async fn leaves_an_agreement_that_already_ended_or_is_still_wanted() { - assert!( - run(IndexingAgreementStatus::CanceledByRequester, false) - .await - .is_empty() - ); - assert!( - run(IndexingAgreementStatus::AcceptedOnChain, true) - .await - .is_empty() - ); + // An accept on record for one left as it is would be announced with no end to follow. + for (status, live) in [ + (IndexingAgreementStatus::Rejected, false), + (IndexingAgreementStatus::CanceledByRequester, false), + (IndexingAgreementStatus::AcceptedOnChain, true), + ] { + let registry = registry(status); + + run_job(®istry, Some(live)).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert!(state.reopened.is_empty(), "{status}"); + assert!(state.accepts.is_empty(), "{status}"); + } + } + + #[tokio::test] + async fn reopens_an_agreement_whose_chain_read_fails() { + // It may have ended, so the end on record stays, and the cancel retry reads it again. + for status in [ + IndexingAgreementStatus::Rejected, + IndexingAgreementStatus::CanceledByRequester, + ] { + let registry = registry(status); + + run_job(®istry, None).await.expect("job ok"); + + assert_eq!(registry.0.lock().unwrap().reopened, vec![false], "{status}"); + } + } + + #[tokio::test] + async fn retries_an_accept_that_failed_to_record_without_reopening_again() { + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + registry.0.lock().unwrap().fail_accept = true; + + let first = run_job(®istry, Some(true)).await; + registry.0.lock().unwrap().fail_accept = false; + run_job(®istry, Some(true)).await.expect("retry ok"); + + assert!(matches!(first, Err(JobError::Retryable(..)))); + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true]); + assert_eq!(state.accepts, vec![chain_accept()]); + } + + #[tokio::test] + async fn retries_when_the_agreement_cannot_be_read() { + let registry = registry(IndexingAgreementStatus::Rejected); + registry.0.lock().unwrap().fail_read = true; + + let result = run_job(®istry, Some(true)).await; + + assert!(matches!(result, Err(JobError::Retryable(..)))); + assert!(registry.0.lock().unwrap().reopened.is_empty()); + } + + /// A job queued before the listener carried the accept still runs. + #[test] + fn reads_a_job_queued_without_an_accept() { + let id = IndexingAgreementId::from_bytes([1; 16]); + let queued = serde_json::json!({ "agreement_id": id }); + + let message: Message = serde_json::from_value(queued).expect("old message"); + + assert_eq!(message.agreement_id, id); + assert_eq!(message.accept, None); } } 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 1d9f19fc..4c1ba5c2 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -958,6 +958,15 @@ mod lifecycle_event_tests { unimplemented!("not exercised by reassess handler") } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs index e3e6fbc4..10a1da7f 100644 --- a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs +++ b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs @@ -729,6 +729,15 @@ mod tests { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/messages.rs b/bin/dipper-service/src/worker/messages.rs index 479d25b4..f279f27d 100644 --- a/bin/dipper-service/src/worker/messages.rs +++ b/bin/dipper-service/src/worker/messages.rs @@ -22,12 +22,8 @@ pub enum Message { /// See [`SendIndexingAgreementProposal`] for more details. SendIndexingAgreementProposal(SendIndexingAgreementProposal), - /// Cancel a rejected agreement on-chain. - /// - /// When an indexer rejects an agreement off-chain but later accepts it on-chain, - /// this message triggers on-chain cancellation via `cancelIndexingAgreementByPayer`. - /// - /// See [`CancelRejectedAgreementOnChain`] for more details. + /// Check on-chain an agreement dipper rejected or cancelled that the chain listener saw + /// accepted. See [`CancelRejectedAgreementOnChain`] for more details. CancelRejectedAgreementOnChain(CancelRejectedAgreementOnChain), /// Submit an RCA offer on-chain after the indexer has accepted the diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 659ca90c..1d97edfd 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -20,6 +20,7 @@ use super::{ result::{JobError, JobResult, calculate_backoff_delay, retries_within_window}, }; pub use super::{ + handlers::ChainAccept, queue::JobPriority, service_queue::{WorkerQueue, WorkerQueueHandle}, }; diff --git a/bin/dipper-service/src/worker/service_queue.rs b/bin/dipper-service/src/worker/service_queue.rs index 49ad37e1..d86b36b5 100644 --- a/bin/dipper-service/src/worker/service_queue.rs +++ b/bin/dipper-service/src/worker/service_queue.rs @@ -4,7 +4,10 @@ use thegraph_core::{DeploymentId, alloy::primitives::ChainId}; use url::Url; use super::{ - handlers::{ReassessIndexingRequest, SendIndexingAgreementProposal, SubmitOffer}, + handlers::{ + CancelRejectedAgreementOnChain, ChainAccept, ReassessIndexingRequest, + SendIndexingAgreementProposal, SubmitOffer, + }, messages::Message, queue::{JobId, JobPriority, Queue}, }; @@ -30,6 +33,15 @@ pub trait WorkerQueue { priority: JobPriority, ) -> anyhow::Result; + /// Check on-chain an agreement dipper rejected or cancelled that was accepted anyway, and + /// move it back into `Cancelling` if it is live, recording `accept`. + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + priority: JobPriority, + ) -> anyhow::Result; + /// Submit an RCA offer on-chain as the first step of a new proposal. The /// job retries until the indexer's window to accept has closed, so a /// provider outage costs one offer only if it outlasts that window. @@ -112,6 +124,23 @@ where .await } + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + priority: JobPriority, + ) -> anyhow::Result { + self.queue + .push( + Message::CancelRejectedAgreementOnChain(CancelRejectedAgreementOnChain { + agreement_id, + accept, + }), + priority, + ) + .await + } + async fn submit_offer( &self, agreement_id: IndexingAgreementId, From 2e61734a69c500ce62a8c4295c12dbf9fafa673d Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:59:29 +0100 Subject: [PATCH 02/13] fix(cancel): keep reading a cancelled agreement whose chain read fails A cancelled agreement the subgraph still showed accepted went back to cancelling on a failed chain read, though the subgraph lagging dipper's own cancel is the likelier cause. Its check now reads again every 5 minutes until the chain answers; a rejected one is still reopened at once. --- bin/dipper-service/src/cancel_dispatch.rs | 65 ++++++++++++---- .../cancel_rejected_agreement_on_chain.rs | 76 +++++++++++++++---- 2 files changed, 109 insertions(+), 32 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 1bb03b3c..1dcfe815 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -288,15 +288,25 @@ async fn record_cancel( } } -/// Move an agreement dipper had already rejected or cancelled back into `Cancelling` when the -/// chain shows it live after all, so the cancel retry ends it. The chain is read first, so a -/// subgraph report from before dipper's cancel landed reopens nothing; an unreadable chain -/// reopens it anyway, as the retry reads again before sending. True if it was reopened. +/// What [`reopen_if_live`] did. +#[derive(Debug)] +pub enum Reopen { + /// Moved back into `Cancelling`. + Reopened, + /// Left as it was: the chain shows it ended, or it is no longer rejected or cancelled. + Left, + /// Left as it was, as the chain couldn't be read, so it needs reading again. + Unread(ChainClientError), +} + +/// Move an agreement dipper rejected or cancelled back into `Cancelling` if the chain shows it +/// live, for the cancel retry to end. An unreadable chain reopens only a rejected one, whose +/// indexer may be paid; a cancelled one likely shows accepted only as the subgraph lags its end. pub async fn reopen_if_live( registry: &R, chain_client: &T, agreement: &IndexingAgreement, -) -> RegistryResult +) -> RegistryResult where R: AgreementRegistry + Sync, T: ChainClient, @@ -306,12 +316,15 @@ where .await { Ok(AgreementOnChain::Live) => true, - Ok(_) => return Ok(false), + Ok(_) => return Ok(Reopen::Left), + Err(err) if agreement.status != IndexingAgreementStatus::Rejected => { + return Ok(Reopen::Unread(err)); + } Err(err) => { tracing::warn!( agreement_id = %agreement.id, error = %err, - "Failed to read an ended agreement reported live; the cancel retry checks it" + "Failed to read a rejected agreement reported live; the cancel retry checks it" ); false } @@ -321,7 +334,7 @@ where .await { Ok(()) => {} - Err(crate::registry::Error::NoRecordsUpdated) => return Ok(false), + Err(crate::registry::Error::NoRecordsUpdated) => return Ok(Reopen::Left), Err(err) => return Err(err), } tracing::warn!( @@ -333,7 +346,7 @@ where reason = "live_on_chain_after_end", "agreement state transition" ); - Ok(true) + Ok(Reopen::Reopened) } /// Log a cancel that mined but could not be read back, naming its transaction so it isn't lost; @@ -724,11 +737,8 @@ pub(crate) mod tests { #[tokio::test] async fn a_reopen_clears_the_end_on_record_only_when_the_chain_shows_it_live() { - // An unread chain reopens it all the same, but it may have ended, so its record stays. - let ag = agreement( - IndexingAgreementStatus::CanceledByRequester, - Some(vec![7u8; 32]), - ); + // An unread chain reopens a rejected one all the same, but it may have ended, so its + // record stays. let live = RecordingChainClient { still_active_after_cancel: true, ..Default::default() @@ -739,14 +749,37 @@ pub(crate) mod tests { }; for (client, seen_live) in [(live, true), (unread, false)] { + let ag = agreement(IndexingAgreementStatus::Rejected, Some(vec![7u8; 32])); let registry = ReopenRegistry::default(); - let reopened = super::reopen_if_live(®istry, &client, &ag) + let reopen = super::reopen_if_live(®istry, &client, &ag) .await .expect("reopen"); - assert!(reopened); + assert!(matches!(reopen, super::Reopen::Reopened)); assert_eq!(*registry.seen_live.lock().unwrap(), vec![seen_live]); } } + + #[tokio::test] + async fn an_unread_chain_leaves_a_cancelled_agreement_to_be_read_again() { + // The subgraph lagging behind dipper's own cancel is the likelier reason it shows + // accepted, and reopening an ended one counts it live until the cancel retry reads it. + let ag = agreement( + IndexingAgreementStatus::CanceledByRequester, + Some(vec![7u8; 32]), + ); + let unread = RecordingChainClient { + read_back_fails: true, + ..Default::default() + }; + let registry = ReopenRegistry::default(); + + let reopen = super::reopen_if_live(®istry, &unread, &ag) + .await + .expect("reopen"); + + assert!(matches!(reopen, super::Reopen::Unread(_))); + assert!(registry.seen_live.lock().unwrap().is_empty()); + } } diff --git a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs index c640a0e4..0c9e480d 100644 --- a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs +++ b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs @@ -7,15 +7,19 @@ use std::time::Duration; use dipper_core::ids::IndexingAgreementId; use crate::{ - cancel_dispatch::reopen_if_live, + cancel_dispatch::{Reopen, reopen_if_live}, chain_client::ChainClient, - registry::{AgreementRegistry, IndexingAgreementStatus}, + registry::{AgreementRegistry, IndexingAgreement, IndexingAgreementStatus}, worker::result::{JobError, JobResult}, }; /// How long a job whose database read or write failed waits before running again. const RETRY_DELAY: Duration = Duration::from_secs(30); +/// How long a job that couldn't read a cancelled agreement on-chain waits to read it again, at +/// the cancel retry's own pace. It keeps reading until the chain answers. +const CHAIN_READ_RETRY_DELAY: Duration = Duration::from_secs(300); + pub struct Ctx { pub registry: R, pub chain_client: T, @@ -60,9 +64,7 @@ where }; let cancelling = match agreement.status { IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester => { - reopen_if_live(&ctx.registry, &ctx.chain_client, &agreement) - .await - .map_err(|err| retry(agreement_id, "reopen the cancel of", err.into()))? + reopen(&ctx, &agreement).await? } // Moved back by an earlier run of this job, or a copy of it, so only its accept is left. IndexingAgreementStatus::Cancelling => true, @@ -80,6 +82,31 @@ where .map_err(|err| retry(agreement_id, "record the on-chain accept of", err.into())) } +/// Move an agreement back into `Cancelling` if the chain shows it live; whether it moved. +async fn reopen(ctx: &Ctx, agreement: &IndexingAgreement) -> JobResult +where + R: AgreementRegistry + Sync, + T: ChainClient, +{ + let agreement_id = &agreement.id; + match reopen_if_live(&ctx.registry, &ctx.chain_client, agreement) + .await + .map_err(|err| retry(agreement_id, "reopen the cancel of", err.into()))? + { + Reopen::Reopened => Ok(true), + Reopen::Left => Ok(false), + Reopen::Unread(err) => { + tracing::warn!( + %agreement_id, + error = %err, + retry_secs = CHAIN_READ_RETRY_DELAY.as_secs(), + "Failed to read whether a cancelled agreement reported live is live; will read again" + ); + Err(JobError::Deferred(CHAIN_READ_RETRY_DELAY)) + } + } +} + /// Log a failed database read or write and run the job again later. fn retry(agreement_id: &IndexingAgreementId, action: &str, err: anyhow::Error) -> JobError { tracing::warn!( @@ -102,7 +129,7 @@ mod tests { use crate::{ cancel_dispatch::tests::agreement, chain_client::{AgreementOnChain, ChainClientError}, - registry::{IndexingAgreement, StubAgreementRegistry}, + registry::StubAgreementRegistry, }; #[derive(Default)] @@ -279,18 +306,35 @@ mod tests { } #[tokio::test] - async fn reopens_an_agreement_whose_chain_read_fails() { - // It may have ended, so the end on record stays, and the cancel retry reads it again. - for status in [ - IndexingAgreementStatus::Rejected, - IndexingAgreementStatus::CanceledByRequester, - ] { - let registry = registry(status); + async fn reopens_a_rejected_agreement_whose_chain_read_fails() { + // Its indexer may be paid, so it is reopened; it may have ended, so the end on record + // stays, and the cancel retry reads it again. + let registry = registry(IndexingAgreementStatus::Rejected); - run_job(®istry, None).await.expect("job ok"); + run_job(®istry, None).await.expect("job ok"); - assert_eq!(registry.0.lock().unwrap().reopened, vec![false], "{status}"); - } + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![false]); + assert_eq!(state.accepts, vec![chain_accept()]); + } + + #[tokio::test] + async fn reads_a_cancelled_agreement_again_when_its_chain_read_fails() { + // Most likely dipper's cancel landed and the subgraph is behind, so it isn't reopened on + // a guess, but it may be an offer accepted after dipper withdrew it, so it is read again. + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + + let unread = run_job(®istry, None).await; + let read = run_job(®istry, Some(true)).await; + + assert!(matches!( + unread, + Err(JobError::Deferred(CHAIN_READ_RETRY_DELAY)) + )); + read.expect("job ok"); + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "reopened only once read live"); + assert_eq!(state.accepts, vec![chain_accept()]); } #[tokio::test] From 4f09441ece01a76da33ee66a130d8cc15a30da22 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:37:44 +0100 Subject: [PATCH 03/13] fix(rpc): stop an endpoint far ahead from locking out healthy ones Dipper refuses endpoints behind the newest block it has seen, and took endpoints up to an hour of blocks apart as agreeing on it, so 1 faulty endpoint ahead could set it and have healthy ones refused for most of an hour. Agreeing, or 1 endpoint moving it alone, now allows only a minute. --- bin/dipper-service/src/chain_client/client.rs | 121 +++++++++++++++--- 1 file changed, 105 insertions(+), 16 deletions(-) diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 842d5021..7c64c580 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -77,6 +77,11 @@ const BLOCKS_PER_SECOND: u64 = 4; /// the newest block dipper has seen raises an ERROR; nearer is lag that clears by itself. const ALERT_GAP_BLOCKS: u64 = 14_400; +/// A minute of blocks at the chain's fastest: how far apart endpoints may be and still agree. +/// Ones that are right differ by a few blocks, and the chain's progress is allowed for apart from +/// this, so it only bounds how far ahead a faulty endpoint can move the newest block seen. +const AGREE_WITHIN_BLOCKS: u64 = 60 * BLOCKS_PER_SECOND; + /// How a read refused by an endpoint whose latest block is too far ahead to be real says so. const FAR_AHEAD_OF_A_SEEN_BLOCK: &str = "too far ahead of the newest block seen"; @@ -87,12 +92,9 @@ const CROSS_CHECK_INTERVAL: Duration = Duration::from_secs(60); /// An answer takes well under a second, so this only bounds how long a hung endpoint holds a read. const CROSS_CHECK_DEADLINE: Duration = Duration::from_secs(3); -/// The newest block dipper has seen, from reads, receipts and latest-block lookups, and when it -/// last moved. It never goes backwards, so a lagging endpoint can't show state from before it. -/// 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. +/// The newest block dipper has seen and when it last moved; it never goes back, so a lagging +/// endpoint can't show state from before it. Nothing ahead is refused until a receipt or 2 agreeing +/// endpoints confirm it, so a wrong one answering first after a restart can't shut out the rest. #[derive(Debug)] struct SeenBlock { number: u64, @@ -145,12 +147,12 @@ impl SeenBlock { (self.number, highest) } - /// The highest block 1 endpoint alone may move a confirmed newest block to: an hour of + /// The highest block 1 endpoint alone may move a confirmed newest block to: a minute 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(AGREE_WITHIN_BLOCKS) .saturating_add(since.saturating_mul(BLOCKS_PER_SECOND)) } @@ -179,7 +181,7 @@ impl SeenBlock { } /// 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. + /// than a minute 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 { @@ -188,7 +190,7 @@ impl SeenBlock { } return; } - if block > self.number || self.number > block.saturating_add(ALERT_GAP_BLOCKS) { + if block > self.number || self.number > block.saturating_add(AGREE_WITHIN_BLOCKS) { self.number = block; self.moved_at = now; } @@ -210,13 +212,13 @@ impl SeenBlock { } /// The newest of the latest blocks that the most endpoints agree on, at least 2 of them within -/// an hour of blocks of each other, or `None` when no 2 agree. +/// a minute of blocks of each other, or `None` when no 2 agree. fn agreed_head(mut heads: Vec) -> Option { heads.sort_unstable(); let mut agreed: Option<(usize, u64)> = None; let mut start = 0; for (end, &head) in heads.iter().enumerate() { - while head - heads[start] > ALERT_GAP_BLOCKS { + while head - heads[start] > AGREE_WITHIN_BLOCKS { start += 1; } let agreeing = end - start + 1; @@ -944,7 +946,7 @@ impl AlloyChainClient { } /// 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, and one over an hour ahead + /// be real is ignored, so it can't refuse every read after it, and one over a minute ahead /// waits for 2 endpoints to agree on it. fn note_block(&self, block: u64) { let only_endpoint = self.inner.rpc_pool.endpoint_count() < 2; @@ -960,7 +962,7 @@ impl AlloyChainClient { Noted::Unagreed => tracing::warn!( block, seen_block = seen, - "Not moving to a block over an hour ahead of the newest block dipper has seen \ + "Not moving to a block over a minute ahead of the newest block dipper has seen \ until 2 RPC endpoints agree on it" ), Noted::TooFarAhead => tracing::warn!( @@ -2679,7 +2681,7 @@ mod tests { let mut seen = SeenBlock::new(); let real = 5_000_000; seen.confirm(real, now); - let jump = real + ALERT_GAP_BLOCKS + 1_000; + let jump = real + AGREE_WITHIN_BLOCKS + 1_000; assert_eq!(seen.advance(real + 10, now, false), Noted::Taken); assert_eq!(seen.advance(jump, now, false), Noted::Unagreed); @@ -2691,7 +2693,7 @@ mod tests { 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), + seen.advance(jump + AGREE_WITHIN_BLOCKS + 1, now, true), Noted::Taken, "the only endpoint has no other to agree with" ); @@ -2724,6 +2726,93 @@ mod tests { assert_eq!(agreed_head(vec![real]), None, "1 can't agree with itself"); } + #[test] + fn a_faulty_endpoint_well_ahead_neither_agrees_nor_moves_a_confirmed_block() { + // Its block taken as seen would have the endpoints that are right refused as behind + // until the chain caught up with it: 10,000 blocks is about 42 minutes. + let now = Instant::now(); + let real = 5_000_000; + let faulty = real + 10_000; + assert_eq!(agreed_head(vec![real, real + 2, faulty]), Some(real + 2)); + + let mut seen = SeenBlock::new(); + seen.confirm(real, now); + assert_eq!(seen.advance(faulty, now, false), Noted::Unagreed); + assert_eq!(seen.bounds(now).0, real); + } + + #[test] + fn a_block_from_a_faulty_endpoint_well_ahead_is_replaced_once_confirmed() { + // After a restart the faulty endpoint may answer first, before anything is confirmed. + let now = Instant::now(); + let real = 5_000_000; + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(real + 10_000, now, false), Noted::Taken); + + seen.confirm(real, now); + + assert_eq!(seen.bounds(now).0, real); + } + + #[test] + fn takes_the_few_blocks_endpoints_that_are_right_differ_by() { + let now = Instant::now(); + let real = 5_000_000; + let minute_ahead = real + AGREE_WITHIN_BLOCKS; + assert_eq!(agreed_head(vec![real, real + 3]), Some(real + 3)); + assert_eq!(agreed_head(vec![real, minute_ahead]), Some(minute_ahead)); + + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(minute_ahead, now, false), Noted::Taken); + seen.confirm(real, now); + assert_eq!( + seen.bounds(now).0, + minute_ahead, + "a lead of a minute is kept" + ); + + let mut seen = SeenBlock::new(); + seen.confirm(real, now); + assert_eq!(seen.advance(real + 40, now, false), Noted::Taken); + assert_eq!(seen.advance(minute_ahead + 40, now, false), Noted::Taken); + + // 10 quiet minutes, in which the chain can add 2,400 blocks. + let mut quiet = SeenBlock { + number: real, + moved_at: now.checked_sub(Duration::from_secs(600)).expect("instant"), + confirmed: true, + cross_checked_at: None, + jump_reported: false, + }; + assert_eq!(quiet.advance(real + 2_400 + 40, now, false), Noted::Taken); + } + + #[tokio::test] + async fn a_faulty_endpoint_well_ahead_does_not_shut_out_the_ones_that_are_right() { + let real = 5_000_000; + let ended = STATE_REGISTERED | STATE_ACCEPTED | STATE_NOTICE_GIVEN; + let right = server_at_block(real, ended).await; + let also_right = server_at_block(real, ended).await; + let faulty = server_at_block(real + 10_000, STATE_REGISTERED).await; + let client = client_over(vec![ + right.uri().parse().expect("provider URL"), + also_right.uri().parse().expect("provider URL"), + faulty.uri().parse().expect("provider URL"), + ]); + // The faulty endpoint answered first after a restart. + client.note_block(real + 10_000); + + let live = client + .agreement_on_chain(&[0xab; 16]) + .await + .expect("read") + .is_live(); + + assert!(!live, "read from an endpoint that is right"); + assert!(client.seen_block().confirmed); + assert_eq!(client.seen_block().number, real); + } + #[tokio::test] async fn confirms_the_newest_block_from_endpoints_that_agree() { // The first endpoint is stuck far behind; the other 2 agree, so it is passed over From 8cf4eefc45b416f3705bfb64b75367c1f1c09f4e Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:45:51 +0100 Subject: [PATCH 04/13] perf(rpc): reuse each endpoint's latest block for a second of reads Every agreement read asked an endpoint for its latest block before reading at that block, so a run of reads made twice the calls. A block given under a second ago is now reused, unless it is behind the newest block dipper has seen or a read at it failed, in which case it is asked again. --- bin/dipper-service/src/chain_client/client.rs | 172 ++++++++++++++++-- .../src/chain_client/rpc_provider.rs | 55 +++++- 2 files changed, 210 insertions(+), 17 deletions(-) diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 7c64c580..2b1d300a 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -28,7 +28,7 @@ use tokio::sync::Mutex; use super::{ abi::{IRecurringAgreementManager, IRecurringCollector}, gas::{GasEstimator, calculate_max_fee, exceeds_max_gas_price, get_gas_prices}, - rpc_provider::{BEHIND_A_SEEN_BLOCK, RpcProviderPool}, + rpc_provider::{BEHIND_A_SEEN_BLOCK, HttpProvider, RpcProviderPool}, }; use crate::{ chain_client::{ @@ -301,6 +301,61 @@ impl Refusals { } } +/// How long a read reuses an endpoint's latest block before asking it again: about 4 blocks. +/// A read at it is still checked against the newest block seen, and asking before every read +/// doubled the calls a run of reads makes. +const HEAD_REUSE: Duration = Duration::from_secs(1); + +/// Each endpoint's latest block and when it was asked for, by the endpoint's place in the list. +struct EndpointHeads(std::sync::Mutex>>); + +impl EndpointHeads { + fn new(endpoints: usize) -> Self { + Self(std::sync::Mutex::new(vec![None; endpoints])) + } + + fn slots(&self) -> std::sync::MutexGuard<'_, Vec>> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// `endpoint`'s latest block, if it was asked for under [`HEAD_REUSE`] before `now`. + fn fresh(&self, endpoint: usize, now: Instant) -> Option { + let (head, asked_at) = self.slots()[endpoint]?; + (now.saturating_duration_since(asked_at) < HEAD_REUSE).then_some(head) + } + + fn remember(&self, endpoint: usize, head: u64, asked_at: Instant) { + self.slots()[endpoint] = Some((head, asked_at)); + } + + fn forget(&self, endpoint: usize) { + self.slots()[endpoint] = None; + } + + /// `endpoint`'s latest block, reused if fresh and not behind `seen`, the newest block seen, + /// otherwise asked for and remembered. + async fn latest( + &self, + provider: &HttpProvider, + endpoint: usize, + seen: u64, + refusals: &Refusals, + ) -> Result { + let reusable = self + .fresh(endpoint, Instant::now()) + .filter(|&head| head >= seen); + if let Some(head) = reusable { + return Ok(head); + } + let asked_at = Instant::now(); + let head = refusals.other(provider.get_block_number().await)?; + self.remember(endpoint, head, asked_at); + Ok(head) + } +} + /// `AgreementDetails.state` flags from `IAgreementCollector.sol` (REGISTERED=1, /// ACCEPTED=2, NOTICE_GIVEN=4, SETTLED=8, BY_PROVIDER=32). `getAgreementDetails` keeps /// ACCEPTED set on a canceled agreement and ORs in NOTICE_GIVEN, so a cancel must clear it, @@ -498,6 +553,8 @@ struct AlloyChainClientInner { /// behind it, so a lagging endpoint can't undo what dipper saw, nor from one too far /// ahead of it to be real, so a faulty endpoint can't refuse every read after it. seen_block: std::sync::Mutex, + /// Each endpoint's latest block, so a run of reads asks for it once a second, not every read. + endpoint_heads: EndpointHeads, } impl AlloyChainClient { @@ -534,6 +591,7 @@ impl AlloyChainClient { ); let submit_deadline = derive_submit_deadline(&rpc_pool); + let endpoint_heads = EndpointHeads::new(rpc_pool.endpoint_count()); Ok(Self { inner: Arc::new(AlloyChainClientInner { @@ -549,6 +607,7 @@ impl AlloyChainClient { submit_lock: Mutex::new(()), submit_deadline, seen_block: std::sync::Mutex::new(SeenBlock::new()), + endpoint_heads, }), }) } @@ -879,9 +938,9 @@ impl AlloyChainClient { .state) } - /// Run a read-only contract call at the endpoint's latest block, refusing an endpoint - /// whose latest block is older than one dipper has already seen, or too far ahead of it - /// to be real; the pool moves on to the next endpoint instead. + /// Run a read-only contract call at the endpoint's latest block, as of [`HEAD_REUSE`] ago at + /// most, refusing an endpoint whose latest block is older than one dipper has already seen, + /// or too far ahead of it to be real; the pool moves on to the next endpoint instead. async fn view_at_seen_block( &self, to: Address, @@ -892,19 +951,23 @@ impl AlloyChainClient { self.cross_check_seen_block().await; let (seen, highest) = self.block_bounds(); let refusals = Refusals::new(); + let heads = &self.inner.endpoint_heads; let read = self .inner .rpc_pool - .execute(operation, |provider| { + .execute_with_endpoint(operation, |provider, endpoint| { let calldata = calldata.clone(); let refusals = &refusals; async move { - let head = refusals.other(provider.get_block_number().await)?; - refusals.behind(head, seen)?; - refusals.far_ahead(head, seen, highest)?; - let tx = TransactionRequest::default().to(to).input(calldata.into()); - let output = - refusals.other(provider.call(tx).block(BlockId::number(head)).await)?; + let head = heads.latest(&provider, endpoint, seen, refusals).await?; + let read = async { + refusals.behind(head, seen)?; + refusals.far_ahead(head, seen, highest)?; + let tx = TransactionRequest::default().to(to).input(calldata.into()); + refusals.other(provider.call(tx).block(BlockId::number(head)).await) + }; + // An endpoint refused or failing may have moved on by the next attempt. + let output = read.await.inspect_err(|_| heads.forget(endpoint))?; Ok((head, output)) } }) @@ -1383,7 +1446,10 @@ impl ManagerEscrowReader for AlloyChainClient { #[cfg(test)] mod tests { use url::Url; - use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate, matchers::method}; + use wiremock::{ + Mock, MockServer, Request, Respond, ResponseTemplate, + matchers::{body_partial_json, method}, + }; use super::*; @@ -2505,6 +2571,88 @@ mod tests { assert!(live); } + /// How many of `requests` called `method`. + fn calls_to(requests: &[wiremock::Request], method: &str) -> usize { + requests + .iter() + .filter(|r| { + let body: serde_json::Value = + serde_json::from_slice(&r.body).expect("JSON-RPC request body"); + body["method"] == method + }) + .count() + } + + #[tokio::test] + async fn reads_in_quick_succession_ask_for_the_latest_block_once() { + let endpoint = server_at_block(100, STATE_REGISTERED).await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + + for _ in 0..3 { + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + } + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 1); + assert_eq!(calls_to(&requests, "eth_call"), 3); + } + + #[test] + fn reuses_an_endpoints_latest_block_for_a_second_unless_forgotten() { + let asked_at = Instant::now(); + let heads = EndpointHeads::new(2); + heads.remember(0, 100, asked_at); + + let soon = asked_at + Duration::from_millis(900); + assert_eq!(heads.fresh(0, soon), Some(100)); + assert_eq!(heads.fresh(0, asked_at + HEAD_REUSE), None, "asked again"); + assert_eq!(heads.fresh(1, soon), None, "each endpoint gives its own"); + + heads.forget(0); + assert_eq!(heads.fresh(0, soon), None, "asked again after a failure"); + } + + #[tokio::test] + async fn asks_for_the_latest_block_again_once_a_newer_one_is_seen() { + // As when dipper's own transaction has just mined: the read must not come from before it. + let endpoint = server_catching_up(100, 1, STATE_REGISTERED).await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + trust_block(&client, 101); + + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 2); + } + + #[tokio::test] + async fn asks_for_the_latest_block_again_after_a_failed_read() { + // The endpoint may have answered from a node that lacks the block it gave. + let endpoint = server_at_block(100, STATE_REGISTERED).await; + Mock::given(method("POST")) + .and(body_partial_json( + serde_json::json!({ "method": "eth_call" }), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "error": { "code": -32000, "message": "missing trie node" }, + }))) + .up_to_n_times(1) + .with_priority(1) + .mount(&endpoint) + .await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + let failed = client.agreement_on_chain(&[0xab; 16]).await; + assert!(failed.is_err(), "got {failed:?}"); + + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 2); + } + #[tokio::test] async fn never_reads_an_agreement_from_an_endpoint_behind_a_block_already_seen() { // The lagging endpoint still shows the offer it hasn't seen accepted and cancelled. diff --git a/bin/dipper-service/src/chain_client/rpc_provider.rs b/bin/dipper-service/src/chain_client/rpc_provider.rs index f05bd1f7..7cdbc88d 100644 --- a/bin/dipper-service/src/chain_client/rpc_provider.rs +++ b/bin/dipper-service/src/chain_client/rpc_provider.rs @@ -223,6 +223,21 @@ impl RpcProviderPool { where F: Fn(HttpProvider) -> Fut, Fut: Future>, + { + self.execute_with_endpoint(operation, |provider, _| f(provider)) + .await + } + + /// As [`execute`](Self::execute), also telling `f` which endpoint it is asking, by its place + /// in the list, so a caller can keep something for each endpoint. + pub async fn execute_with_endpoint( + &self, + operation: &str, + f: F, + ) -> Result + where + F: Fn(HttpProvider, usize) -> Fut, + Fut: Future>, { self.execute_with_retries(operation, self.max_retries, f) .await @@ -235,7 +250,7 @@ impl RpcProviderPool { f: F, ) -> Result where - F: Fn(HttpProvider) -> Fut, + F: Fn(HttpProvider, usize) -> Fut, Fut: Future>, { // What each endpoint said, in the order they were tried. `sign_and_send` matches @@ -254,11 +269,12 @@ impl RpcProviderPool { let start = self.current_index.load(Ordering::Relaxed); loop { - let current_url = self.url_at(start + providers_tried).clone(); + let position = (start + providers_tried) % self.providers.len(); + let current_url = self.url_at(position).clone(); let endpoint = endpoint_name(¤t_url); let endpoint_error = match self - .ask_endpoint(operation, ¤t_url, max_retries, &f) + .ask_endpoint(operation, position, ¤t_url, max_retries, &f) .await { Ok(result) => return Ok(result), @@ -320,12 +336,13 @@ impl RpcProviderPool { async fn ask_endpoint( &self, operation: &str, + position: usize, url: &Url, max_retries: u32, f: &F, ) -> Result where - F: Fn(HttpProvider) -> Fut, + F: Fn(HttpProvider, usize) -> Fut, Fut: Future>, { let endpoint = endpoint_name(url); @@ -335,7 +352,7 @@ impl RpcProviderPool { let mut endpoint_error: Option = None; let mut lag_retries = 0; for attempt in 0..=max_retries { - match f(provider.clone()).await { + match f(provider.clone(), position).await { Ok(result) => return Ok(result), Err(e) if Self::is_behind(&e) @@ -537,6 +554,34 @@ mod tests { } } + /// A caller keeping something for each endpoint must be told the one it is really asking. + #[tokio::test] + async fn a_call_is_told_which_endpoint_it_is_asking() { + let servers = [server_answering_block(7).await, server_refusing().await]; + let pool = RpcProviderPool::new( + servers + .iter() + .map(|s| s.uri().parse().expect("server URL")) + .collect(), + Duration::from_secs(5), + 0, + ) + .expect("pool"); + pool.rotate(); + + let asked = std::sync::Mutex::new(Vec::new()); + let block = pool + .execute_with_endpoint("probe", |provider, endpoint| { + asked.lock().expect("asked").push(endpoint); + async move { provider.get_block_number().await } + }) + .await + .expect("the first endpoint answers"); + + assert_eq!(block, 7); + assert_eq!(*asked.lock().expect("asked"), vec![1, 0], "wrapped round"); + } + /// Hosted RPC endpoints carry their API key in the path, and this failure text reaches the /// logs and every error built from it, so it has to say which endpoint refused without /// repeating the key that gets it in. From a2014cb3a9da40ffb896343410233bcf66c46fa1 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:41:10 +0100 Subject: [PATCH 05/13] fix(registry): replace an estimated end whole with the chain's end When the chain listener is over an hour late, dipper records an indexer's end itself, at the time it noticed and with no transaction. The listener's later record only filled in the transaction, mixing 2 ends; the chain's end now replaces dipper's whole and is announced again. --- .../src/network/service/cancel_retry.rs | 4 +- .../src/network/service/chain_listener.rs | 4 +- dipper-pgregistry/src/postgres.rs | 21 ++-- .../tests/it_registry_postgres.rs | 102 ++++++++++++++++++ 4 files changed, 116 insertions(+), 15 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 43d148ac..05863eb1 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -288,8 +288,8 @@ async fn confirm_if_over( } /// Mark an agreement the indexer ended `CanceledByIndexer` when the chain listener hasn't in -/// time, so it doesn't stay cancelling for good. The indexer is recorded as ending it first, -/// so its announcement names them; the time recorded is when dipper noticed. +/// time, so it doesn't stay cancelling for good. The indexer is recorded as ending it first, at +/// the time dipper noticed: an estimate the chain's end, once the listener reads it, replaces. async fn record_end_by_indexer( registry: &R, agreement: &IndexingAgreement, diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index efa3a51c..d0e4b09b 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -986,8 +986,8 @@ where } /// Record the accept and end of an agreement dipper already ended that went live on-chain, in 1 -/// write, as the terminated sweep waits only for the accept; what is on record wins, but an end -/// before the accept gives way. The end is its last change on-chain, so a failure is returned. +/// write, as the terminated sweep waits only for the accept. What is on record wins, unless its end +/// predates the accept or has no transaction. The end is its last change, so a failure is returned. async fn record_accept_and_cancel_from_chain( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 679ea805..529e01ed 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1558,13 +1558,9 @@ impl PgRegistry { Ok(()) } - /// Record the cancel audit payload for a dipper-initiated cancel so the - /// emission sweep can populate the `terminated` event's tx/by/at fields. - /// `COALESCE` keeps any value already observed on-chain. Best-effort - /// enrichment: the event still emits (with fallbacks) if never recorded. - /// An end recorded before the agreement's accept, such as its offer's withdrawal before the - /// offer landed after all, can't be its end, so a later one replaces it and is announced. - /// That is decided once per row, so the end's 4 fields always come from the same end. + /// Record an agreement's end for its `terminated` event. A new end from the accept on replaces + /// the end on record whole, to be announced again, if that came before the accept, so can't + /// be the end, or is dipper's estimate, with no transaction, and the new one has it. #[expect( clippy::cast_possible_wrap, reason = "predates this lint; fix when next touched" @@ -1586,8 +1582,9 @@ impl PgRegistry { CASE WHEN replace_end THEN $4 ELSE COALESCE(agreement.canceled_tx, $4) END, CASE WHEN replace_end THEN NULL ELSE agreement.terminated_event_emitted_at END FROM ( - SELECT agreement.canceled_at < agreement.accepted_at - AND $2 >= agreement.accepted_at AS replace_end + SELECT $2 >= agreement.accepted_at + AND (agreement.canceled_at < agreement.accepted_at + OR (agreement.canceled_tx IS NULL AND $4 IS NOT NULL)) AS replace_end ) AS end_rule ) WHERE agreement.id = $1 @@ -1631,8 +1628,10 @@ impl PgRegistry { CASE WHEN replace_end THEN NULL ELSE agreement.terminated_event_emitted_at END FROM ( - SELECT agreement.canceled_at < COALESCE(agreement.accepted_at, $2) - AND $4 >= COALESCE(agreement.accepted_at, $2) AS replace_end + SELECT $4 >= COALESCE(agreement.accepted_at, $2) + AND (agreement.canceled_at < COALESCE(agreement.accepted_at, $2) + OR (agreement.canceled_tx IS NULL AND $6 IS NOT NULL)) + AS replace_end ) AS end_rule ) WHERE agreement.id = $1 diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index 17ff4501..90645428 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -4170,6 +4170,108 @@ async fn an_accept_and_end_from_the_chain_are_recorded_in_1_write() { ); } +/// An end dipper records itself when the chain listener is slow, with no transaction, is an +/// estimate: the chain's end replaces it whole, never mixed with it, and is announced again. +#[tokio::test] +async fn an_estimated_end_gives_way_whole_to_the_chains() { + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + let registry = PgRegistry::new(db.clone()); + // The end on record, and whether it is yet to be announced. + let end = async |id: IndexingAgreementId| { + sqlx::query_as::<_, (Option, Option, Option, bool)>( + "SELECT canceled_at, canceled_by, canceled_tx, terminated_event_emitted_at IS NULL \ + FROM dipper_reg_indexing_agreements WHERE id = $1", + ) + .bind(id) + .fetch_one(&db) + .await + .expect("cancel record query") + }; + let chains = |unannounced: bool| { + ( + Some(5_000), + Some("0xprovider".to_owned()), + Some("0xend".to_owned()), + unannounced, + ) + }; + + let id = fixture_agreement(0xaa); + registry + .mark_indexing_agreement_as_cancelling(&id) + .await + .expect("mark cancelling"); + registry + .record_accepted_audit(&id, 1_000, "0xacc") + .await + .expect("accept record"); + registry + .record_cancel_audit(&id, 9_000, "0xindexer", None) + .await + .expect("estimated end"); + let marked = registry + .apply_reconciliation(&id, false, Some(CancelKind::ByIndexer)) + .await + .expect("mark ended by the indexer"); + assert!(marked.did_cancel); + // Its terminated event waited an hour for a transaction, then went out with the estimate. + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce"); + + let read_from_chain = async || { + registry + .record_accept_and_cancel_audit(&id, 1_000, "0xacc", 5_000, "0xprovider", Some("0xend")) + .await + .expect("the chain's end"); + }; + read_from_chain().await; + assert_eq!( + end(id).await, + chains(true), + "the chain's end replaces the estimate whole, to be announced again" + ); + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce again"); + read_from_chain().await; + assert_eq!(end(id).await, chains(false), "a replay changes nothing"); + registry + .record_cancel_audit(&id, 9_500, "0xindexer", None) + .await + .expect("another estimate"); + assert_eq!( + end(id).await, + chains(false), + "an estimate never replaces the chain's end" + ); + + // The other writer of an end follows the same rule. + let other = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]); + registry + .record_accepted_audit(&other, 1_000, "0xacc") + .await + .expect("accept record"); + registry + .record_cancel_audit(&other, 9_000, "0xindexer", None) + .await + .expect("estimated end"); + registry + .record_cancel_audit(&other, 5_000, "0xprovider", Some("0xend")) + .await + .expect("end with its transaction"); + assert_eq!(end(other).await, chains(true)); +} + /// An ended agreement's `terminated` event waits for the transaction that ended it, which the /// chain listener can record after dipper marks it ended, but not for ever. #[tokio::test] From ae2170d9d68fc09e6d40f58ece2004caf3699528 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:52:11 +0100 Subject: [PATCH 06/13] fix(registry): warn when an announced agreement end changes Correcting an agreement's end after its terminated event went out sends the event again, but consumers drop repeats of the same agreement and event type, so they may never see it. Dipper now logs a tagged WARN with the old and new end, and a sweep comment names the right key. --- .../src/network/service/cancel_retry.rs | 25 +--- .../src/network/service/chain_listener.rs | 2 +- bin/dipper-service/src/registry.rs | 90 +++++++++++- bin/dipper-service/src/test_support.rs | 27 +++- dipper-pgregistry/src/lib.rs | 4 +- dipper-pgregistry/src/postgres.rs | 130 +++++++++++++++--- .../tests/it_registry_postgres.rs | 116 +++++++++++++++- 7 files changed, 343 insertions(+), 51 deletions(-) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 05863eb1..cdfa7b53 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -445,6 +445,7 @@ mod tests { cancel_dispatch::tests::agreement, chain_client::AgreementOnChain, registry::{IndexingAgreementStatus, StubAgreementRegistry}, + test_support::logged, worker::service::JobPriority, }; @@ -1004,30 +1005,6 @@ mod tests { } } - /// What dipper logs, at every level, while `run` runs. - async fn logged(run: impl Future) -> String { - struct Buffer(Arc>>); - impl std::io::Write for Buffer { - fn write(&mut self, bytes: &[u8]) -> std::io::Result { - self.0.lock().unwrap().extend_from_slice(bytes); - Ok(bytes.len()) - } - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - let lines = Arc::new(Mutex::new(Vec::new())); - let writer = Arc::clone(&lines); - let subscriber = tracing_subscriber::fmt() - .with_ansi(false) - .with_max_level(tracing::Level::DEBUG) - .with_writer(move || Buffer(Arc::clone(&writer))) - .finish(); - let _logging = tracing::subscriber::set_default(subscriber); - run.await; - String::from_utf8(lines.lock().unwrap().clone()).unwrap() - } - #[tokio::test] async fn logs_an_end_by_the_indexer_as_a_status_change_only_when_it_made_one() { for listener_ended_it in [false, true] { diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index d0e4b09b..73ea8b10 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -1608,7 +1608,7 @@ where .await { // Sent but not stamped: the next sweep re-sends (deduped by - // consumers on agreement_id). Better than dropping. + // consumers on agreement_id and event type). Better than dropping. tracing::warn!( agreement_id = %p.agreement_id, error = %err, diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index e194517d..d451f4ae 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -99,6 +99,30 @@ fn filter_map_with_logging( } } +/// Warn when a write changed an agreement's end after its `terminated` event went out. The event +/// goes out again with the new end, but consumers dedupe on the agreement id and event type, so +/// they may drop it and keep the old end. +fn warn_if_end_corrected( + agreement_id: &IndexingAgreementId, + correction: Option, +) { + let Some(dipper_pgregistry::EndCorrection { old, new }) = correction else { + return; + }; + tracing::warn!( + event = "agreement_end_corrected_after_announcement", + %agreement_id, + old_canceled_at = ?old.canceled_at, + old_canceled_by = ?old.canceled_by, + old_canceled_tx = ?old.canceled_tx, + new_canceled_at = ?new.canceled_at, + new_canceled_by = ?new.canceled_by, + new_canceled_tx = ?new.canceled_tx, + "An agreement's end changed after its terminated event went out; the event goes out \ + again, but consumers may drop it as a repeat" + ); +} + /// A service for interacting with the registry, including registering new /// indexing requests and indexing agreements. #[derive(Clone)] @@ -416,10 +440,12 @@ impl AgreementRegistry for RegistryProvider { id: &IndexingAgreementId, seen_live: bool, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .reopen_indexing_agreement_cancel(id, seen_live) - .await - .map_err(Into::into) + .await?; + warn_if_end_corrected(id, correction); + Ok(()) } async fn get_cancelling_agreements( @@ -620,9 +646,11 @@ impl AgreementRegistry for RegistryProvider { canceled_by: &str, canceled_tx: Option<&str>, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .record_cancel_audit(agreement_id, canceled_at, canceled_by, canceled_tx) .await?; + warn_if_end_corrected(agreement_id, correction); Ok(()) } @@ -635,7 +663,8 @@ impl AgreementRegistry for RegistryProvider { canceled_by: &str, canceled_tx: Option<&str>, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .record_accept_and_cancel_audit( agreement_id, accepted_at, @@ -645,6 +674,7 @@ impl AgreementRegistry for RegistryProvider { canceled_tx, ) .await?; + warn_if_end_corrected(agreement_id, correction); Ok(()) } @@ -894,3 +924,53 @@ impl crate::network::service::chain_listener::ChainListenerStateRegistry for Reg .map_err(Into::into) } } + +#[cfg(test)] +mod tests { + use dipper_core::ids::IndexingAgreementId; + use dipper_pgregistry::{EndCorrection, RecordedEnd}; + + use super::warn_if_end_corrected; + use crate::test_support::logged; + + #[tokio::test] + async fn warns_with_both_ends_when_an_announced_end_changes() { + let id = IndexingAgreementId::from_bytes([0xaa; 16]); + let correction = EndCorrection { + old: RecordedEnd { + canceled_at: Some(9_000), + canceled_by: Some("0xindexer".to_owned()), + canceled_tx: None, + }, + new: RecordedEnd { + canceled_at: Some(5_000), + canceled_by: Some("0xprovider".to_owned()), + canceled_tx: Some("0xend".to_owned()), + }, + }; + + let logs = logged(async { warn_if_end_corrected(&id, Some(correction)) }).await; + + for expected in [ + "WARN", + "event=\"agreement_end_corrected_after_announcement\"", + &format!("agreement_id={id}"), + "old_canceled_at=Some(9000)", + "old_canceled_tx=None", + "new_canceled_at=Some(5000)", + "new_canceled_tx=Some(\"0xend\")", + "consumers may drop it", + ] { + assert!(logs.contains(expected), "missing {expected} in {logs}"); + } + } + + #[tokio::test] + async fn stays_quiet_when_no_announced_end_changes() { + let id = IndexingAgreementId::from_bytes([0xaa; 16]); + + let logs = logged(async { warn_if_end_corrected(&id, None) }).await; + + assert!(logs.is_empty(), "{logs}"); + } +} diff --git a/bin/dipper-service/src/test_support.rs b/bin/dipper-service/src/test_support.rs index 4ef6eb37..d3c4112e 100644 --- a/bin/dipper-service/src/test_support.rs +++ b/bin/dipper-service/src/test_support.rs @@ -3,7 +3,8 @@ //! Currently provides [`CapturingEventsProducer`], a test double for //! [`SubgraphIndexingAgreementEventsProducer`] that records every produced event //! so handler/service tests can assert exactly which lifecycle events were emitted -//! (and with which payloads) without a Kafka backend. +//! (and with which payloads) without a Kafka backend, and [`logged`], which captures what +//! dipper logs. use std::sync::{Arc, Mutex}; @@ -157,3 +158,27 @@ impl SubgraphIndexingAgreementEventsProducer for CapturingEventsProducer { Ok(()) } } + +/// What dipper logs, at every level, while `run` runs. +pub async fn logged(run: impl std::future::Future) -> String { + struct Buffer(Arc>>); + impl std::io::Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let lines = Arc::new(Mutex::new(Vec::new())); + let writer = Arc::clone(&lines); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing::Level::DEBUG) + .with_writer(move || Buffer(Arc::clone(&writer))) + .finish(); + let _logging = tracing::subscriber::set_default(subscriber); + run.await; + String::from_utf8(lines.lock().unwrap().clone()).unwrap() +} diff --git a/dipper-pgregistry/src/lib.rs b/dipper-pgregistry/src/lib.rs index 8e0ea869..c78ed6c0 100644 --- a/dipper-pgregistry/src/lib.rs +++ b/dipper-pgregistry/src/lib.rs @@ -20,9 +20,9 @@ pub use indexing_request::{ Status as IndexingRequestStatus, }; pub use postgres::{ - CancelKind, CancellingAgreement, ChainListenerStateRow, NewAgreementParams, + CancelKind, CancellingAgreement, ChainListenerStateRow, EndCorrection, NewAgreementParams, PendingAcceptedEvent, PendingExpiredEvent, PendingTerminatedEvent, PgRegistry, - ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, + ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, RecordedEnd, }; pub use result::{Error, Result}; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 529e01ed..75e0c144 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -95,6 +95,46 @@ pub struct ReconciliationAudit { pub canceled_tx: Option, } +/// An agreement's end as recorded: when, in chain seconds, who ended it, and in which transaction. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RecordedEnd { + pub canceled_at: Option, + pub canceled_by: Option, + pub canceled_tx: Option, +} + +/// An announced end that a write replaced or cleared, so its `terminated` event goes out again. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EndCorrection { + pub old: RecordedEnd, + pub new: RecordedEnd, +} + +/// What a write that can replace an end returns: whether it cleared that end's announcement, then +/// the end before and after it. The end before is read under a row lock, so it is the one replaced. +type EndChangeRow = ( + bool, + Option, + Option, + Option, + Option, + Option, + Option, +); + +fn end_correction(row: EndChangeRow) -> Option { + let (announcement_cleared, old_at, old_by, old_tx, new_at, new_by, new_tx) = row; + let end = |at: Option, canceled_by, canceled_tx| RecordedEnd { + canceled_at: at.and_then(|at| u64::try_from(at).ok()), + canceled_by, + canceled_tx, + }; + announcement_cleared.then(|| EndCorrection { + old: end(old_at, old_by, old_tx), + new: end(new_at, new_by, new_tx), + }) +} + /// A row that needs a `terminated` lifecycle event emitted. Sourced entirely /// from the agreement row so the emission sweep is a pure DB -> broker drain. #[derive(Debug, Clone)] @@ -1015,13 +1055,14 @@ impl PgRegistry { /// as checked, since no cancel is sent with it, so the retry takes it on its next sweep rather /// than waiting for one to be mined. When the chain was read and showed it live (`seen_live`), /// the end on record, and any announcement of it, no longer stands, so both are cleared for - /// the end still to come; an unread chain leaves them, as the agreement may have ended. + /// the end still to come, and the end returned if it was announced; an unread chain leaves + /// them, as the agreement may have ended. pub async fn reopen_indexing_agreement_cancel( &self, agreement_id: &IndexingAgreementId, seen_live: bool, - ) -> Result<(), Error> { - let updated = sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements SET @@ -1035,7 +1076,14 @@ impl PgRegistry { terminated_event_emitted_at = CASE WHEN $5 THEN NULL ELSE terminated_event_emitted_at END, updated_at = timezone('UTC', now()) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $2 FOR UPDATE + ) AS before WHERE id = $2 AND status IN ($3, $4) + RETURNING was_announced AND terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, canceled_at, canceled_by, canceled_tx "#, ) .bind(IndexingAgreementStatus::Cancelling) @@ -1043,12 +1091,9 @@ impl PgRegistry { .bind(IndexingAgreementStatus::CanceledByRequester) .bind(IndexingAgreementStatus::Rejected) .bind(seen_live) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - if updated.rows_affected() == 0 { - return Err(Error::NoRecordsUpdated); - } - Ok(()) + row.map(end_correction).ok_or(Error::NoRecordsUpdated) } /// `Cancelling` agreements, those checked longest ago first, leaving out any never checked @@ -1571,8 +1616,8 @@ impl PgRegistry { canceled_at: u64, canceled_by: &str, canceled_tx: Option<&str>, - ) -> Result<(), Error> { - sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements AS agreement SET (canceled_at, canceled_by, canceled_tx, terminated_event_emitted_at) = ( @@ -1587,16 +1632,24 @@ impl PgRegistry { OR (agreement.canceled_tx IS NULL AND $4 IS NOT NULL)) AS replace_end ) AS end_rule ) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $1 FOR UPDATE + ) AS before WHERE agreement.id = $1 + RETURNING was_announced AND agreement.terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, + agreement.canceled_at, agreement.canceled_by, agreement.canceled_tx "#, ) .bind(agreement_id) .bind(canceled_at as i64) .bind(canceled_by) .bind(canceled_tx) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - Ok(()) + Ok(row.and_then(end_correction)) } /// Record an agreement's accept and its end together, as the chain shows them, in 1 write, @@ -1614,8 +1667,8 @@ impl PgRegistry { canceled_at: u64, canceled_by: &str, canceled_tx: Option<&str>, - ) -> Result<(), Error> { - sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements AS agreement SET accepted_at = COALESCE(agreement.accepted_at, $2), @@ -1634,7 +1687,15 @@ impl PgRegistry { AS replace_end ) AS end_rule ) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $1 FOR UPDATE + ) AS before WHERE agreement.id = $1 + RETURNING was_announced AND agreement.terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, + agreement.canceled_at, agreement.canceled_by, agreement.canceled_tx "#, ) .bind(agreement_id) @@ -1643,9 +1704,9 @@ impl PgRegistry { .bind(canceled_at as i64) .bind(canceled_by) .bind(canceled_tx) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - Ok(()) + Ok(row.and_then(end_correction)) } // ========================================================================= @@ -2556,3 +2617,40 @@ async fn update_status_from( .await? .is_empty()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_end_counts_as_corrected_only_when_its_announcement_was_cleared() { + let row = |cleared| { + ( + cleared, + Some(9_000), + Some("0xindexer".to_owned()), + None, + Some(5_000), + Some("0xprovider".to_owned()), + Some("0xend".to_owned()), + ) + }; + + assert_eq!(end_correction(row(false)), None); + assert_eq!( + end_correction(row(true)), + Some(EndCorrection { + old: RecordedEnd { + canceled_at: Some(9_000), + canceled_by: Some("0xindexer".to_owned()), + canceled_tx: None, + }, + new: RecordedEnd { + canceled_at: Some(5_000), + canceled_by: Some("0xprovider".to_owned()), + canceled_tx: Some("0xend".to_owned()), + }, + }) + ); + } +} diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index 90645428..f3e9d974 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -4,8 +4,9 @@ use std::collections::HashSet; use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; use dipper_pgregistry::{ - CancelKind, Error, IndexingAgreementStatus, IndexingAgreementTerms, IndexingRequestStatus, - NewAgreementParams, PgRegistry, ReconciliationAudit, ReconciliationItem, + CancelKind, EndCorrection, Error, IndexingAgreementStatus, IndexingAgreementTerms, + IndexingRequestStatus, NewAgreementParams, PgRegistry, ReconciliationAudit, ReconciliationItem, + RecordedEnd, }; use fake::{Fake, Faker}; use pgtemp::PgTempDB; @@ -3927,6 +3928,117 @@ async fn an_end_recorded_before_the_accept_gives_way_to_the_real_one() { ); } +/// Each write that can change an agreement's end reports a change to an end already announced, +/// whose `terminated` event then goes out again, and nothing otherwise. +#[tokio::test] +async fn a_change_to_an_announced_end_is_reported() { + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + let registry = PgRegistry::new(db.clone()); + let announce = async |id: IndexingAgreementId| { + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce"); + }; + let end = |at: u64, by: &str, tx: &str| RecordedEnd { + canceled_at: Some(at), + canceled_by: Some(by.to_owned()), + canceled_tx: Some(tx.to_owned()), + }; + + // An offer withdrawn and announced, then accepted after all, so its real end replaces it. + let id = fixture_agreement(0xaa); + let first = registry + .record_cancel_audit(&id, 1_000, "0xwithdrawer", Some("0xwithdrawal")) + .await + .expect("cancel record"); + assert_eq!(first, None, "nothing was announced yet"); + announce(id).await; + registry + .record_accepted_audit(&id, 2_000, "0xaccept") + .await + .expect("accept record"); + let replaced = registry + .record_cancel_audit(&id, 3_000, "0xpayer", Some("0xcancel")) + .await + .expect("cancel record"); + assert_eq!( + replaced, + Some(EndCorrection { + old: end(1_000, "0xwithdrawer", "0xwithdrawal"), + new: end(3_000, "0xpayer", "0xcancel"), + }) + ); + announce(id).await; + let kept = registry + .record_cancel_audit(&id, 4_000, "0xlater", Some("0xlater")) + .await + .expect("cancel record"); + assert_eq!(kept, None, "an end kept is no change"); + + // Found live on-chain after its announced end, which no longer stands. + registry + .mark_indexing_agreement_as_canceled_by_requester(&id) + .await + .expect("mark ended"); + let unread = registry + .reopen_indexing_agreement_cancel(&id, false) + .await + .expect("reopen"); + assert_eq!( + unread, None, + "an unread chain keeps the end and its announcement" + ); + registry + .mark_indexing_agreement_as_canceled_by_requester(&id) + .await + .expect("mark ended"); + let reopened = registry + .reopen_indexing_agreement_cancel(&id, true) + .await + .expect("reopen"); + assert_eq!( + reopened, + Some(EndCorrection { + old: end(3_000, "0xpayer", "0xcancel"), + new: RecordedEnd::default(), + }) + ); + + // The chain's accept and end, read together, replace an announced withdrawal the same way. + let other = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]); + sqlx::query( + "UPDATE dipper_reg_indexing_agreements SET canceled_at = 50, canceled_by = '0xwd', \ + canceled_tx = '0xwd', terminated_event_emitted_at = now() WHERE id = $1", + ) + .bind(other) + .execute(&db) + .await + .expect("withdraw"); + let read = async || { + registry + .record_accept_and_cancel_audit(&other, 100, "0xacc", 200, "0xpayer", Some("0xend")) + .await + .expect("record") + }; + assert_eq!( + read().await, + Some(EndCorrection { + old: end(50, "0xwd", "0xwd"), + new: end(200, "0xpayer", "0xend"), + }) + ); + announce(other).await; + assert_eq!(read().await, None, "a replay changes nothing"); +} + #[tokio::test] async fn a_cancelling_agreement_stays_live_and_unannounced_until_it_ends() { let (db, _temp_db) = temp_registry_db().await; From 04e21c6aa16ee88606a866ef7bf8796e8d04c32f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:38:09 +0100 Subject: [PATCH 07/13] fix(retry): stop warning when an agreement ends mid-check Dipper's cancel retry records each check of an agreement it is still cancelling. When the chain listener has already ended that agreement nothing is updated, which is expected, yet it logged a warning. That case now logs at debug level; failed database writes still warn. --- .../src/network/service/cancel_retry.rs | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index cdfa7b53..3f8682d2 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -366,6 +366,10 @@ async fn note_check( log_failed_cancel(row, attempts, failed_attempts, err); } } + Err(crate::registry::Error::NoRecordsUpdated) => tracing::debug!( + agreement_id = %agreement.id, + "Agreement no longer cancelling when its check was recorded, as the chain listener can end it first" + ), Err(err) => tracing::warn!( agreement_id = %agreement.id, error = %err, @@ -469,6 +473,8 @@ mod tests { found_ended: Mutex>>, /// The chain listener marks it ended before the retry's own mark lands. listener_ended_it: bool, + /// Recording a check fails, as with the database down. + check_fails: bool, writes: Mutex>, /// Ended agreements awaiting replacement, until noted as replaced. awaiting_replacement: Vec, @@ -549,6 +555,14 @@ mod tests { failed_attempts: u32, ended: Option, ) -> crate::registry::Result { + if self.listener_ended_it { + return Err(crate::registry::Error::NoRecordsUpdated); + } + if self.check_fails { + return Err(crate::registry::Error::BackendError( + dipper_pgregistry::Error::DbError(sqlx::Error::PoolTimedOut), + )); + } self.checks.fetch_add(1, Ordering::SeqCst); self.found_ended.lock().unwrap().push(ended); Ok(self.attempts.fetch_add(failed_attempts, Ordering::SeqCst) + failed_attempts) @@ -1040,6 +1054,38 @@ mod tests { } } + #[tokio::test] + async fn logs_a_check_the_listener_beat_at_debug_and_a_failed_one_as_a_warning() { + // The listener ending the agreement mid-sweep is expected; only a failed write needs a look. + let check = async |listener_ended_it, check_fails| { + let registry = MockRegistry { + listener_ended_it, + check_fails, + ..registry_with_one(false) + }; + // Withdrawn before its deadline, so the check is recorded and it stays cancelling. + logged(retry(®istry, &MockChain::default(), DEADLINE)).await + }; + + let beaten = check(true, false).await; + let failed = check(false, true).await; + + let line = |logs: &str, message: &str| { + let line = logs.lines().find(|line| line.contains(message)); + line.unwrap_or_else(|| panic!("no {message:?} in {logs}")) + .to_owned() + }; + assert!( + line(&beaten, "no longer cancelling").contains("DEBUG"), + "{beaten}" + ); + assert!(!beaten.contains("WARN"), "{beaten}"); + assert!( + line(&failed, "Failed to record a check").contains("WARN"), + "{failed}" + ); + } + #[tokio::test] async fn reads_an_ended_agreement_once_to_learn_who_ended_it() { let mut registry = registry_with_one(true); From c5c7474ab9a484b6d4b595791cf8c75639e91e29 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:41:56 +0100 Subject: [PATCH 08/13] fix(logs): name agreement statuses the same way in every status log Logs of an agreement's status change typed the status names by hand, so they could drift from the names dipper prints elsewhere, as the expiry log had ("Created" for CREATED). Each now prints the status itself, so the names always match, without the quotes plain-text logs added. --- bin/dipper-service/src/cancel_dispatch.rs | 26 ++++++++++++++----- .../src/network/service/cancel_retry.rs | 18 +++++++------ .../src/network/service/expiration.rs | 9 ++++--- .../handlers/reassess_indexing_request.rs | 22 ++++++++-------- .../send_indexing_agreement_proposal.rs | 8 +++--- 5 files changed, 50 insertions(+), 33 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 1dcfe815..31701c0f 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -85,10 +85,10 @@ pub enum CancelReason { } impl CancelReason { - fn ended_status(self) -> &'static str { + fn ended_status(self) -> IndexingAgreementStatus { match self { - Self::NotWanted => "CANCELED_BY_REQUESTER", - Self::Abandoned => "ABANDONED_BY_INDEXER", + Self::NotWanted => IndexingAgreementStatus::CanceledByRequester, + Self::Abandoned => IndexingAgreementStatus::AbandonedByIndexer, } } } @@ -258,8 +258,8 @@ pub async fn confirm_cancelled( tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "CANCELLING", - new_status = reason.ended_status(), + old_status = %IndexingAgreementStatus::Cancelling, + new_status = %reason.ended_status(), reason = "cancel_confirmed_on_chain", "agreement state transition" ); @@ -342,7 +342,7 @@ where indexer_id = %agreement.indexer.id, indexing_request_id = %agreement.indexing_request_id, old_status = %agreement.status, - new_status = "CANCELLING", + new_status = %IndexingAgreementStatus::Cancelling, reason = "live_on_chain_after_end", "agreement state transition" ); @@ -420,7 +420,7 @@ pub(crate) mod tests { use time::OffsetDateTime; use url::Url; - use super::{LiveCancel, SCOPE_BOTH, cancel_agreement_on_chain}; + use super::{CancelReason, LiveCancel, SCOPE_BOTH, cancel_agreement_on_chain}; use crate::{ chain_client::{AgreementOnChain, ChainClient, ChainClientError}, config::IndexingAgreementConfig, @@ -560,6 +560,18 @@ pub(crate) mod tests { } } + #[test] + fn each_reason_ends_an_agreement_in_its_own_status() { + assert_eq!( + CancelReason::NotWanted.ended_status(), + IndexingAgreementStatus::CanceledByRequester + ); + assert_eq!( + CancelReason::Abandoned.ended_status(), + IndexingAgreementStatus::AbandonedByIndexer + ); + } + #[tokio::test] async fn manager_cancel_uses_both_scopes_for_accepted() { // comp-1: a manager cancel passes BOTH scope bits so the collector diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 3f8682d2..92ec79e2 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -16,7 +16,7 @@ use crate::{ config::IndexingAgreementConfig, registry::{ AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement, - IndexingRequestRegistry, + IndexingAgreementStatus, IndexingRequestRegistry, }, worker::service::WorkerQueue, }; @@ -329,8 +329,8 @@ fn log_end_by_indexer(agreement: &IndexingAgreement, changed: bool) { tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "CANCELLING", - new_status = "CANCELED_BY_INDEXER", + old_status = %IndexingAgreementStatus::Cancelling, + new_status = %IndexingAgreementStatus::CanceledByIndexer, reason = "indexer_ended_it_before_dipper_cancelled", "agreement state transition" ); @@ -446,11 +446,8 @@ mod tests { use super::*; use crate::{ - cancel_dispatch::tests::agreement, - chain_client::AgreementOnChain, - registry::{IndexingAgreementStatus, StubAgreementRegistry}, - test_support::logged, - worker::service::JobPriority, + cancel_dispatch::tests::agreement, chain_client::AgreementOnChain, + registry::StubAgreementRegistry, test_support::logged, worker::service::JobPriority, }; const DEADLINE: u64 = 1_000; @@ -1046,6 +1043,11 @@ mod tests { changed, "{logs}" ); + assert_eq!( + logs.contains("old_status=CANCELLING new_status=CANCELED_BY_INDEXER"), + changed, + "{logs}" + ); assert_eq!( logs.contains("already recorded the indexer's end"), !changed, diff --git a/bin/dipper-service/src/network/service/expiration.rs b/bin/dipper-service/src/network/service/expiration.rs index c62541b6..85162713 100644 --- a/bin/dipper-service/src/network/service/expiration.rs +++ b/bin/dipper-service/src/network/service/expiration.rs @@ -10,7 +10,10 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use crate::{ config::ExpirationConfig, network::service::chain_listener::ChainListenerStateRegistry, - registry::{AgreementRegistry, IndexingRequestRegistry, PendingCancellationRegistry}, + registry::{ + AgreementRegistry, IndexingAgreementStatus, IndexingRequestRegistry, + PendingCancellationRegistry, + }, worker::service::{JobPriority, WorkerQueue}, }; @@ -185,8 +188,8 @@ where tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "Created", - new_status = "Expired", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Expired, "agreement state transition" ); 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 4c1ba5c2..80a1df0c 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -27,9 +27,9 @@ use crate::{ }, }, registry::{ - AgreementRegistry, IndexerDenylistRegistry, IndexingAgreementTerms, - IndexingAgreementTermsMetadata, IndexingRequestRegistry, NewAgreementParams, - PendingCancellationRegistry, + AgreementRegistry, IndexerDenylistRegistry, IndexingAgreementStatus, + IndexingAgreementTerms, IndexingAgreementTermsMetadata, IndexingRequestRegistry, + NewAgreementParams, PendingCancellationRegistry, }, signing::eip712::Eip712Signer, worker::{ @@ -556,7 +556,7 @@ where agreement_id = %id, indexing_request_id = %indexing_request_id, old_status = "none", - new_status = "CREATED", + new_status = %IndexingAgreementStatus::Created, reason = "reassessment_replacement", "agreement state transition" ); @@ -594,7 +594,7 @@ where agreement_id = %id, indexing_request_id = %indexing_request_id, old_status = "none", - new_status = "CREATED", + new_status = %IndexingAgreementStatus::Created, reason = "reassessment_addition", "agreement state transition" ); @@ -674,7 +674,7 @@ where agreement_id = %old_agreement.id, indexing_request_id = %indexing_request_id, old_status = %old_agreement.status, - new_status, + new_status = %new_status, reason = "reassessment_not_in_target_group", "agreement state transition" ); @@ -2015,8 +2015,8 @@ mod lifecycle_event_tests { /// What became of an old agreement taken out of the target group. #[derive(Debug, PartialEq, Eq)] enum Unpaired { - /// Moved to the status named. - Moved(&'static str), + /// Moved to this status. + Moved(IndexingAgreementStatus), /// Already ended or being cancelled, as a race with another path can leave it. AlreadyEnding, /// Couldn't be marked; logged. @@ -2042,7 +2042,7 @@ where .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) .await { - Ok(()) => Unpaired::Moved("CANCELED_BY_REQUESTER"), + Ok(()) => Unpaired::Moved(IndexingAgreementStatus::CanceledByRequester), Err(err) => unmarked(agreement, &err), }; } @@ -2056,12 +2056,12 @@ where .await { Ok(crate::cancel_dispatch::CancelStarted::Ended) => { - Unpaired::Moved("CANCELED_BY_REQUESTER") + Unpaired::Moved(IndexingAgreementStatus::CanceledByRequester) } Ok( crate::cancel_dispatch::CancelStarted::NotLive | crate::cancel_dispatch::CancelStarted::MayBeLive, - ) => Unpaired::Moved("CANCELLING"), + ) => Unpaired::Moved(IndexingAgreementStatus::Cancelling), Err(err) => unmarked(agreement, &err), } } diff --git a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs index 10a1da7f..3825f91a 100644 --- a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs +++ b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs @@ -190,8 +190,8 @@ where tracing::info!( agreement_id = %agreement_id, indexing_request_id = %indexing_request_id, - old_status = "CREATED", - new_status = "REJECTED", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Rejected, reason = %format_args!("rejected_{rejection_reason_str}"), detail = %rejected.detail, "agreement state transition" @@ -232,8 +232,8 @@ where tracing::info!( agreement_id = %agreement_id, indexing_request_id = %indexing_request_id, - old_status = "CREATED", - new_status = "UNRESPONSIVE", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Unresponsive, reason = "unresponsive", "agreement state transition" ); From 13a5bf2c36cdd127dd94756ce06caeda7fb1258d Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:49:44 +0100 Subject: [PATCH 09/13] refactor(registry): choose end statuses where the reason is known Saving an agreement as cancelled by dipper quietly saved it as abandoned if dipper had noted giving up on its indexer, so a caller could get a status it never asked for. Callers that know why dipper cancelled now name the status; a cancel read off the chain still uses the note. --- bin/dipper-service/src/cancel_dispatch.rs | 33 +++-- .../src/network/service/cancel_retry.rs | 24 ++++ bin/dipper-service/src/registry.rs | 10 ++ bin/dipper-service/src/registry/agreement.rs | 14 +- .../src/registry/agreement_stub.rs | 17 +++ dipper-pgregistry/src/postgres.rs | 133 ++++++++++-------- .../tests/it_registry_postgres.rs | 40 +++++- 7 files changed, 201 insertions(+), 70 deletions(-) diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 31701c0f..a32029bd 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -162,6 +162,26 @@ pub async fn mark_cancelling( } } +/// Mark an agreement dipper's cancel ended in the status its `reason` ends it in. +async fn mark_ended( + registry: &R, + agreement: &IndexingAgreement, + reason: CancelReason, +) -> RegistryResult<()> { + match reason { + CancelReason::NotWanted => { + registry + .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) + .await + } + CancelReason::Abandoned => { + registry + .mark_indexing_agreement_as_abandoned_by_indexer(&agreement.id) + .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, @@ -219,10 +239,10 @@ async fn finish_marked_cancel( } } -/// Mark an agreement the chain shows dipper ended as ended, recording the cancel when its -/// transaction is known, so the `terminated` sweep announces it. False, logged, when the mark -/// fails; it stays `Cancelling` for the cancel retry. One the chain listener already marked -/// ended counts as ended. +/// Mark an agreement the chain shows dipper ended as ended, as its `reason` says, recording the +/// cancel when its transaction is known, so the `terminated` sweep announces it. False, logged, +/// when the mark fails; it stays `Cancelling` for the cancel retry. One the chain listener +/// already marked ended counts as ended. pub async fn confirm_cancelled( registry: &R, agreement: &IndexingAgreement, @@ -234,10 +254,7 @@ pub async fn confirm_cancelled( if tx_hash.is_some() { record_cancel(registry, agreement, tx_hash, config).await; } - match registry - .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) - .await - { + match mark_ended(registry, agreement, reason).await { Ok(()) => {} Err(crate::registry::Error::NoRecordsUpdated) => { tracing::debug!( diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 92ec79e2..bafff371 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -462,6 +462,7 @@ mod tests { struct MockRegistry { cancelling: Vec, marked_cancelled: Mutex>, + marked_abandoned: Mutex>, marked_by_indexer: Mutex>, audits: Mutex>>, audited_by: Mutex>, @@ -518,6 +519,16 @@ mod tests { self.writes.lock().unwrap().push("ended"); Ok(()) } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> crate::registry::Result<()> { + if self.listener_ended_it { + return Err(crate::registry::Error::NoRecordsUpdated); + } + self.marked_abandoned.lock().unwrap().push(*id); + Ok(()) + } async fn record_cancel_audit( &self, _id: &IndexingAgreementId, @@ -886,6 +897,19 @@ mod tests { assert_eq!(*registry.audits.lock().unwrap(), vec![Some(tx)]); } + #[tokio::test] + async fn ends_an_agreement_its_indexer_stopped_serving_as_abandoned() { + let mut registry = registry_with_one(true); + registry.cancelling[0].abandoned = true; + + let logs = logged(retry(®istry, &live_chain(), 0)).await; + + let id = registry.cancelling[0].agreement.id; + assert_eq!(*registry.marked_abandoned.lock().unwrap(), vec![id]); + assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + assert!(logs.contains("ABANDONED_BY_INDEXER"), "{logs}"); + } + #[tokio::test] async fn records_the_cancel_before_marking_the_agreement_ended() { // The end is announced once the mark lands; recorded after, the announcement could diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index d451f4ae..bb07b366 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -415,6 +415,16 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()> { + self.inner + .mark_indexing_agreement_as_abandoned_by_indexer(id) + .await + .map_err(Into::into) + } + async fn mark_indexing_agreement_as_cancelling( &self, id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index fde09894..33d05ff5 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -260,13 +260,20 @@ pub trait AgreementRegistry { /// /// If there is no indexing agreement with the given ID, or if the agreement is not in the /// `CREATED`, `ACCEPTED_ON_CHAIN`, `REJECTED` or `CANCELLING` state, this method returns a - /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. One dipper was cancelling because its - /// indexer stopped serving it becomes `ABANDONED_BY_INDEXER` instead. + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, ) -> RegistryResult<()>; + /// Mark a `CANCELLING` agreement dipper was ending because its indexer stopped serving it + /// `ABANDONED_BY_INDEXER`, once the chain confirms dipper's cancel; + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) otherwise. + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()>; + /// Mark a `CREATED`, `ACCEPTED_ON_CHAIN`, `REJECTED` or `EXPIRED` agreement `CANCELLING`, /// before its cancel is sent; [`NoRecordUpdated`](Error::NoRecordsUpdated) otherwise. async fn mark_indexing_agreement_as_cancelling( @@ -324,7 +331,8 @@ pub trait AgreementRegistry { /// - `apply_accept`: attempt `Created | Expired → AcceptedOnChain`. /// May be a no-op if the row is already in another status. /// - `cancel`: optionally apply `CanceledByRequester` or - /// `CanceledByIndexer` from an accepting/pre-accept state. + /// `CanceledByIndexer` from an accepting/pre-accept state. Dipper's cancel of one it was + /// ending because its indexer stopped serving it applies `AbandonedByIndexer` instead. /// /// Returns which transitions actually affected a row so callers can /// gate post-commit side effects (e.g. running diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index 1253c754..ef5bc214 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -133,6 +133,16 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_indexing_agreement_as_canceled_by_requester") } + /// Defaults to the requester mark, so a mock that records only that dipper ended an + /// agreement sees both kinds of end; override it to tell them apart. + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> Result<()> { + self.mark_indexing_agreement_as_canceled_by_requester(id) + .await + } + async fn mark_indexing_agreement_as_cancelling(&self, _id: &IndexingAgreementId) -> Result<()> { unimplemented!("mark_indexing_agreement_as_cancelling") } @@ -473,6 +483,13 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_indexing_agreement_as_canceled_by_requester(self, id).await } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> Result<()> { + StubAgreementRegistry::mark_indexing_agreement_as_abandoned_by_indexer(self, id).await + } + async fn mark_indexing_agreement_as_cancelling(&self, id: &IndexingAgreementId) -> Result<()> { StubAgreementRegistry::mark_indexing_agreement_as_cancelling(self, id).await } diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 75e0c144..013225b0 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -990,7 +990,7 @@ impl PgRegistry { Ok(()) } - /// One being cancelled because its indexer stopped serving it ends `AbandonedByIndexer`. + /// End an agreement dipper no longer wants `CanceledByRequester`. pub async fn mark_indexing_agreement_as_canceled_by_requester( &self, agreement_id: &IndexingAgreementId, @@ -1003,6 +1003,20 @@ impl PgRegistry { .await } + /// End an agreement dipper was cancelling because its indexer stopped serving it, once the + /// chain confirms dipper's cancel. + pub async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + agreement_id: &IndexingAgreementId, + ) -> Result<(), Error> { + self.set_status_from( + agreement_id, + IndexingAgreementStatus::AbandonedByIndexer, + &[IndexingAgreementStatus::Cancelling], + ) + .await + } + /// Mark an agreement that may be live on-chain `Cancelling`, before dipper sends its /// on-chain cancel. One marked `Expired` may have been accepted unseen by a lagging listener. pub async fn mark_indexing_agreement_as_cancelling( @@ -1257,18 +1271,7 @@ impl PgRegistry { let mut did_cancel = false; if let Some(kind) = cancel { - let (new_status, allowed_from): (_, &[IndexingAgreementStatus]) = match kind { - CancelKind::ByRequester => ( - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ), - CancelKind::ByIndexer => ( - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ), - }; - did_cancel = - update_status_from(&mut tx, agreement_id, new_status, allowed_from).await?; + did_cancel = apply_cancel(&mut tx, agreement_id, kind).await?; if did_accept && !did_cancel { return Err(Error::NoRecordsUpdated); } @@ -1293,7 +1296,6 @@ impl PgRegistry { /// stale outcome flags from the first iteration; debug_asserted. #[expect( clippy::expect_used, - clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] pub async fn apply_reconciliation_batch( @@ -1351,18 +1353,7 @@ impl PgRegistry { ], ) .await?; - let (new_status, allowed_from): (_, &[IndexingAgreementStatus]) = match cancel_kind { - CancelKind::ByRequester => ( - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ), - CancelKind::ByIndexer => ( - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ), - }; - let did_cancel = - update_status_from(&mut tx, &item.agreement_id, new_status, allowed_from).await?; + let did_cancel = apply_cancel(&mut tx, &item.agreement_id, cancel_kind).await?; if did_accept && !did_cancel { return Err(Error::NoRecordsUpdated); } @@ -1389,25 +1380,12 @@ impl PgRegistry { outcomes.entry(id).or_default().did_accept = true; } - for id in batch_update_status_from( - &mut tx, - &cancel_by_requester, - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ) - .await? + for id in batch_apply_cancel(&mut tx, &cancel_by_requester, CancelKind::ByRequester).await? { outcomes.entry(id).or_default().did_cancel = true; } - for id in batch_update_status_from( - &mut tx, - &cancel_by_indexer, - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ) - .await? - { + for id in batch_apply_cancel(&mut tx, &cancel_by_indexer, CancelKind::ByIndexer).await? { outcomes.entry(id).or_default().did_cancel = true; } @@ -2496,41 +2474,84 @@ impl PgRegistry { } } -/// Batched form of `update_status_from`: transitions all rows whose `id` -/// is in `agreement_ids` and whose current status is in `allowed_from` to -/// `new_status`, in one statement; a row noted abandoned that `new_status` would make -/// `CanceledByRequester` becomes `AbandonedByIndexer` instead. Returns the ids of the rows that -/// actually flipped (matched the CAS guard) so callers can build per-id -/// outcome maps. Empty input is a fast-path no-op. +/// Batched form of `update_status_from`: moves every row in `agreement_ids` whose status is in +/// `allowed_from` to `new_status`, in 1 statement. Returns the ids of the rows that moved, so +/// callers can build per-id outcome maps; empty input is a no-op. async fn batch_update_status_from( tx: &mut sqlx::Transaction<'_, Postgres>, agreement_ids: &[IndexingAgreementId], new_status: IndexingAgreementStatus, allowed_from: &[IndexingAgreementStatus], +) -> Result, Error> { + batch_set_status(tx, agreement_ids, "$1", &[new_status], allowed_from).await +} + +/// Record a cancel by `kind`, seen on the chain, in 1 row; true if it ended the row. +async fn apply_cancel( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_id: &IndexingAgreementId, + kind: CancelKind, +) -> Result { + let ended = batch_apply_cancel(tx, std::slice::from_ref(agreement_id), kind).await?; + Ok(!ended.is_empty()) +} + +/// Record a cancel by `kind`, seen on the chain, in each row it ends, returning their ids. The +/// chain doesn't say why dipper cancelled, so its cancel ends a row noted abandoned (its indexer +/// stopped serving it) `AbandonedByIndexer`, and the rest `CanceledByRequester`. +async fn batch_apply_cancel( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_ids: &[IndexingAgreementId], + kind: CancelKind, +) -> Result, Error> { + match kind { + CancelKind::ByRequester => { + let ended = [ + IndexingAgreementStatus::AbandonedByIndexer, + IndexingAgreementStatus::CanceledByRequester, + ]; + let status = "CASE WHEN abandoned THEN $1 ELSE $2 END"; + batch_set_status(tx, agreement_ids, status, &ended, CANCEL_BY_REQUESTER_FROM).await + } + CancelKind::ByIndexer => { + let ended = IndexingAgreementStatus::CanceledByIndexer; + batch_update_status_from(tx, agreement_ids, ended, CANCEL_BY_INDEXER_FROM).await + } + } +} + +/// Set the status of the rows in `agreement_ids` whose status is in `allowed_from` to +/// `status_sql`, which reads `new_statuses` as `$1`, `$2` and so on, returning their ids. +async fn batch_set_status( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_ids: &[IndexingAgreementId], + status_sql: &str, + new_statuses: &[IndexingAgreementStatus], + allowed_from: &[IndexingAgreementStatus], ) -> Result, Error> { if agreement_ids.is_empty() { return Ok(Vec::new()); } + let ids_at = new_statuses.len() + 1; let placeholders = (0..allowed_from.len()) - .map(|i| format!("${}", i + 5)) + .map(|i| format!("${}", ids_at + 1 + i)) .collect::>() .join(", "); - // An agreement dipper ended because its indexer stopped serving it ends as abandoned. let sql = format!( r#" UPDATE dipper_reg_indexing_agreements SET - status = CASE WHEN abandoned AND $1 = $3 THEN $4 ELSE $1 END, + status = {status_sql}, updated_at = timezone('UTC', now()) - WHERE id = ANY($2) AND status IN ({placeholders}) + WHERE id = ANY(${ids_at}) AND status IN ({placeholders}) RETURNING id "# ); - let mut query = sqlx::query_as::<_, (IndexingAgreementId,)>(&sql) - .bind(new_status) - .bind(agreement_ids) - .bind(IndexingAgreementStatus::CanceledByRequester) - .bind(IndexingAgreementStatus::AbandonedByIndexer); + let mut query = sqlx::query_as::<_, (IndexingAgreementId,)>(&sql); + for status in new_statuses { + query = query.bind(*status); + } + query = query.bind(agreement_ids); for status in allowed_from { query = query.bind(*status); } diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index f3e9d974..94ee111d 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -698,7 +698,7 @@ async fn get_declined_indexers_by_deployment_includes_an_indexer_that_abandoned_ .await .expect("abandon"); registry - .mark_indexing_agreement_as_canceled_by_requester(&agreement_id) + .mark_indexing_agreement_as_abandoned_by_indexer(&agreement_id) .await .expect("ended abandoned"); @@ -1858,7 +1858,7 @@ where async fn an_abandoned_agreement_dipper_cancels_ends_abandoned() { let (again, status) = abandon_then_end(async |registry, id| { registry - .mark_indexing_agreement_as_canceled_by_requester(id) + .mark_indexing_agreement_as_abandoned_by_indexer(id) .await .expect("dipper's cancel confirmed"); }) @@ -1871,6 +1871,40 @@ async fn an_abandoned_agreement_dipper_cancels_ends_abandoned() { ); } +#[tokio::test] +async fn the_requester_mark_writes_its_own_status_whatever_the_abandoned_flag() { + // The caller picks the end status, so it logs the one written. + let (_, status) = abandon_then_end(async |registry, id| { + registry + .mark_indexing_agreement_as_canceled_by_requester(id) + .await + .expect("marked"); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::CanceledByRequester); +} + +#[tokio::test] +async fn an_abandoned_agreement_the_listener_sees_cancelled_in_a_batch_ends_abandoned() { + let (_, status) = abandon_then_end(async |registry, id| { + let items = [ReconciliationItem { + agreement_id: *id, + apply_accept: false, + cancel: Some(CancelKind::ByRequester), + audit: ReconciliationAudit::default(), + }]; + let outcomes = registry + .apply_reconciliation_batch(&items) + .await + .expect("reconciliation"); + assert!(outcomes[id].did_cancel); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::AbandonedByIndexer); +} + #[tokio::test] async fn an_abandoned_agreement_the_listener_sees_cancelled_ends_abandoned() { let (_, status) = abandon_then_end(async |registry, id| { @@ -4198,7 +4232,7 @@ async fn an_abandoned_agreement_awaits_replacement_once_ended_until_queued() { ); registry - .mark_indexing_agreement_as_canceled_by_requester(&id) + .mark_indexing_agreement_as_abandoned_by_indexer(&id) .await .expect("end it"); assert_eq!(awaiting().await, vec![id]); From 090c5996dc269a2e47dfd204d79b3eb4750e3679 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 15:49:14 +0100 Subject: [PATCH 10/13] docs(registry): move a status update's description onto that function The note describing the single-row status update sat on the function that stores an agreement's accept details instead, so it described the wrong code. --- dipper-pgregistry/src/postgres.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 013225b0..e47ebd12 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -2559,8 +2559,6 @@ async fn batch_set_status( Ok(rows.into_iter().map(|(id,)| id).collect()) } -/// Transition an agreement's status inside a transaction. Thin wrapper -/// over `batch_update_status_from` for the single-row case. /// Persist the accepted-transition audit payload for a single row. `COALESCE` /// keeps any already-stored value rather than clobbering it with `NULL`. #[expect( @@ -2623,6 +2621,8 @@ async fn persist_cancel_audit( Ok(()) } +/// Move an agreement whose status is in `allowed_from` to `new_status` inside a transaction, +/// returning whether it moved. Thin wrapper over `batch_update_status_from` for 1 row. async fn update_status_from( tx: &mut sqlx::Transaction<'_, Postgres>, agreement_id: &IndexingAgreementId, From e6a0da241004a038605e6681c0e22e9eb4d73add Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 16:05:25 +0100 Subject: [PATCH 11/13] docs(comments): write counts as numerals in 4 comments House style writes a count in a code comment as a numeral ("1 endpoint", not "one endpoint"), and 4 comments on agreement cancels spelled theirs out. --- bin/dipper-service/src/chain_client/rpc_provider.rs | 2 +- bin/dipper-service/src/network/service/cancel_retry.rs | 2 +- bin/dipper-service/src/network/service/chain_listener.rs | 2 +- dipper-pgregistry/tests/it_registry_postgres.rs | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/bin/dipper-service/src/chain_client/rpc_provider.rs b/bin/dipper-service/src/chain_client/rpc_provider.rs index 7cdbc88d..3db14d3e 100644 --- a/bin/dipper-service/src/chain_client/rpc_provider.rs +++ b/bin/dipper-service/src/chain_client/rpc_provider.rs @@ -331,7 +331,7 @@ impl RpcProviderPool { } } - /// Run a call against one endpoint, retrying a fault worth another go, and return what it + /// Run a call against 1 endpoint, retrying a fault worth another go, and return what it /// answered or why it last failed. async fn ask_endpoint( &self, diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index bafff371..11a786f2 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -386,7 +386,7 @@ fn log_uncounted_failure(agreement: &IndexingAgreement, err: &ChainClientError) ); } -/// One ERROR as an agreement reaches the limit, for an operator to look into; a WARN for +/// Log 1 ERROR as an agreement reaches the limit, for an operator to look into, and a WARN for /// every other failed cancel. fn log_failed_cancel( row: &CancellingAgreement, diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 73ea8b10..087af90c 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -3553,7 +3553,7 @@ mod tests { assert!(!registry.was_cancel_audit_recorded(&old_id)); } - /// Runs the pending cancellation of one old agreement in `status`, returning + /// Runs the pending cancellation of 1 old agreement in `status`, returning /// whether it was already marked cancelling when its on-chain cancel went out. async fn marked_at_pending_cancel(status: IndexingAgreementStatus) -> Vec { let registry = MockRegistry::new(); diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index 94ee111d..d613288c 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3470,7 +3470,7 @@ async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs(); - // 3 agreements being cancelled: an offer still open, one accepted, one past its deadline. + // 3 agreements being cancelled: an offer still open, 1 accepted, 1 past its deadline. let mut ids = Vec::new(); for (deployment_id, deadline) in [ ( From 7c4ddf97c62f5e8b41afc2577f6fb6c2571127b7 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 16:27:18 +0100 Subject: [PATCH 12/13] fix(cancel): retry a chain check until the database answers A job checks whether an agreement dipper ended was accepted on-chain anyway, and if so ends it. On database errors it gave up after 3 tries in about 90 seconds, and nothing queued it again, so the indexer kept being paid. It now tries again every 30 seconds until the database answers. --- .../cancel_rejected_agreement_on_chain.rs | 54 +++++++++++++++---- 1 file changed, 43 insertions(+), 11 deletions(-) diff --git a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs index 0c9e480d..aadb9f8d 100644 --- a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs +++ b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs @@ -13,7 +13,9 @@ use crate::{ worker::result::{JobError, JobResult}, }; -/// How long a job whose database read or write failed waits before running again. +/// How long a job whose database read or write failed waits before running again. It keeps +/// running until the database answers: the listener queues it once per change, and nothing else +/// checks an ended agreement for a live one, so a job given up on would leave its indexer paid. const RETRY_DELAY: Duration = Duration::from_secs(30); /// How long a job that couldn't read a cancelled agreement on-chain waits to read it again, at @@ -107,14 +109,16 @@ where } } -/// Log a failed database read or write and run the job again later. +/// Log a failed database read or write and run the job again later, without using up its +/// attempts. fn retry(agreement_id: &IndexingAgreementId, action: &str, err: anyhow::Error) -> JobError { tracing::warn!( %agreement_id, error = format!("{err:#}"), + retry_secs = RETRY_DELAY.as_secs(), "Failed to {action} an agreement dipper ended that was accepted on-chain; will retry" ); - JobError::Retryable(err, RETRY_DELAY) + JobError::Deferred(RETRY_DELAY) } #[cfg(test)] @@ -139,6 +143,7 @@ mod tests { reopened: Vec, accepts: Vec, fail_read: bool, + fail_reopen: bool, fail_accept: bool, } @@ -169,6 +174,9 @@ mod tests { seen_live: bool, ) -> crate::registry::Result<()> { let mut state = self.0.lock().unwrap(); + if state.fail_reopen { + return Err(db_down()); + } let agreement = state.agreement.as_mut().expect("agreement"); agreement.status = IndexingAgreementStatus::Cancelling; state.reopened.push(seen_live); @@ -346,21 +354,45 @@ mod tests { registry.0.lock().unwrap().fail_accept = false; run_job(®istry, Some(true)).await.expect("retry ok"); - assert!(matches!(first, Err(JobError::Retryable(..)))); + assert!(matches!(first, Err(JobError::Deferred(RETRY_DELAY)))); let state = registry.0.lock().unwrap(); assert_eq!(state.reopened, vec![true]); assert_eq!(state.accepts, vec![chain_accept()]); } #[tokio::test] - async fn retries_when_the_agreement_cannot_be_read() { - let registry = registry(IndexingAgreementStatus::Rejected); - registry.0.lock().unwrap().fail_read = true; - - let result = run_job(®istry, Some(true)).await; + async fn keeps_a_job_whose_database_keeps_failing() { + // The listener queues this check once per change, so a job given up on would leave an + // agreement live on-chain marked ended; deferring reruns it without using up attempts. + type Failure = (&'static str, fn(&mut State)); + let failures: [Failure; 3] = [ + ("read", |state| state.fail_read = true), + ("reopen", |state| state.fail_reopen = true), + ("accept", |state| state.fail_accept = true), + ]; + for (failing, fail) in failures { + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + fail(&mut registry.0.lock().unwrap()); + + for _ in 0..5 { + let result = run_job(®istry, Some(true)).await; + assert!( + matches!(result, Err(JobError::Deferred(RETRY_DELAY))), + "{failing}: {result:?}" + ); + } + { + let mut state = registry.0.lock().unwrap(); + state.fail_read = false; + state.fail_reopen = false; + state.fail_accept = false; + } + run_job(®istry, Some(true)).await.expect("job ok"); - assert!(matches!(result, Err(JobError::Retryable(..)))); - assert!(registry.0.lock().unwrap().reopened.is_empty()); + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "{failing}"); + assert_eq!(state.accepts, vec![chain_accept()], "{failing}"); + } } /// A job queued before the listener carried the accept still runs. From 018b337e129eac39486166444053bf326fd1e9db Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 12 Oct 2026 16:28:47 +0100 Subject: [PATCH 13/13] docs(rpc): say 2 endpoints that disagree can still refuse the right one Dipper refuses RPC endpoints behind the newest block it has seen. With only 2 that disagree it can't tell which is right, so 1 faulty far ahead still has the other refused until a receipt. The code comment and warning now say so, and the warning lists each endpoint's latest block. --- bin/dipper-service/src/chain_client/client.rs | 39 ++++++++++++++++--- 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 2b1d300a..beaf922f 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -79,7 +79,7 @@ const ALERT_GAP_BLOCKS: u64 = 14_400; /// A minute of blocks at the chain's fastest: how far apart endpoints may be and still agree. /// Ones that are right differ by a few blocks, and the chain's progress is allowed for apart from -/// this, so it only bounds how far ahead a faulty endpoint can move the newest block seen. +/// this, so it only bounds how far ahead a faulty endpoint can move a confirmed newest block seen. const AGREE_WITHIN_BLOCKS: u64 = 60 * BLOCKS_PER_SECOND; /// How a read refused by an endpoint whose latest block is too far ahead to be real says so. @@ -94,7 +94,10 @@ const CROSS_CHECK_DEADLINE: Duration = Duration::from_secs(3); /// The newest block dipper has seen and when it last moved; it never goes back, so a lagging /// endpoint can't show state from before it. Nothing ahead is refused until a receipt or 2 agreeing -/// endpoints confirm it, so a wrong one answering first after a restart can't shut out the rest. +/// endpoints confirm it, which replaces a block far ahead set by a faulty endpoint that answered +/// first after a restart. Until then any endpoint's block is taken: with only 2 that disagree, +/// which is right can't be told, so 1 far ahead has the other refused as behind until a receipt +/// or the chain catches up. Refusing it instead would let 1 stuck far behind serve stale reads. #[derive(Debug)] struct SeenBlock { number: u64, @@ -998,12 +1001,14 @@ impl AlloyChainClient { if pool.endpoint_count() < 2 || !self.seen_block().cross_check_due(Instant::now()) { return; } - match agreed_head(pool.latest_blocks(CROSS_CHECK_DEADLINE).await) { + let heads = pool.latest_blocks(CROSS_CHECK_DEADLINE).await; + match agreed_head(heads.clone()) { 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, nor the newest block seen moved far ahead, until \ - they do" + latest_blocks = ?heads, + "No 2 RPC endpoints agree on the chain's latest block; until they do, or a \ + receipt confirms it, any endpoint's block is taken as the newest seen and \ + endpoints behind it are refused" ), } } @@ -2902,6 +2907,28 @@ mod tests { assert_eq!(seen.bounds(now).0, real); } + #[test] + fn takes_the_higher_block_from_2_endpoints_that_disagree() { + // Which is right can't be told, so 1 stuck far behind is refused, and so is a right one + // behind 1 faulty far ahead, until a receipt replaces its block. + let now = Instant::now(); + let real = 5_000_000; + let stuck = real - 10_000; + let faulty = real + 10_000; + assert_eq!(agreed_head(vec![stuck, real]), None); + assert_eq!(agreed_head(vec![real, faulty]), None); + + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(stuck, now, false), Noted::Taken); + assert_eq!(seen.advance(real, now, false), Noted::Taken); + assert_eq!(seen.bounds(now).0, real, "the stuck one is refused"); + assert_eq!(seen.advance(faulty, now, false), Noted::Taken); + assert_eq!(seen.bounds(now).0, faulty, "the right one is refused"); + + seen.confirm(real + 4, now); + assert_eq!(seen.bounds(now).0, real + 4); + } + #[test] fn takes_the_few_blocks_endpoints_that_are_right_differ_by() { let now = Instant::now();