diff --git a/bin/dipper-service/src/worker/handlers.rs b/bin/dipper-service/src/worker/handlers.rs index e5cb76ef..fe545d82 100644 --- a/bin/dipper-service/src/worker/handlers.rs +++ b/bin/dipper-service/src/worker/handlers.rs @@ -17,4 +17,8 @@ pub use send_indexing_agreement_proposal::{ Ctx as SendIndexingAgreementProposalCtx, Message as SendIndexingAgreementProposal, handle as send_indexing_agreement_proposal, }; -pub use submit_offer::{Ctx as SubmitOfferCtx, Message as SubmitOffer, handle as submit_offer}; +pub use submit_offer::{ + Ctx as SubmitOfferCtx, DROPPED_TX_RETRY_BASE as SUBMIT_OFFER_DROPPED_TX_RETRY_BASE, + Message as SubmitOffer, TRANSIENT_RETRY_BASE as SUBMIT_OFFER_TRANSIENT_RETRY_BASE, + handle as submit_offer, +}; diff --git a/bin/dipper-service/src/worker/handlers/submit_offer.rs b/bin/dipper-service/src/worker/handlers/submit_offer.rs index f1966174..21511f8c 100644 --- a/bin/dipper-service/src/worker/handlers/submit_offer.rs +++ b/bin/dipper-service/src/worker/handlers/submit_offer.rs @@ -29,6 +29,12 @@ use crate::{ worker::result::{JobError, JobResult}, }; +/// Backoff base for a tx the RPC accepted and then dropped from the mempool. +pub const DROPPED_TX_RETRY_BASE: Duration = Duration::from_secs(5); + +/// Backoff base for a transient submission failure: RPC, gas or nonce. +pub const TRANSIENT_RETRY_BASE: Duration = Duration::from_secs(30); + pub struct Ctx { pub registry: R, pub chain_client: T, @@ -150,7 +156,7 @@ where error = %err, "Offer tx dropped from mempool, will retry with fresh nonce" ); - return Err(JobError::Retryable(err.into(), Duration::from_secs(5))); + return Err(JobError::Retryable(err.into(), DROPPED_TX_RETRY_BASE)); } Err(ChainClientError::ContractRevert { selector, data }) => { // A gas-estimation revert won't clear on a quick retry: bad terms @@ -175,7 +181,7 @@ where error = %err, "Failed to submit offer on-chain, will retry" ); - return Err(JobError::Retryable(err.into(), Duration::from_secs(30))); + return Err(JobError::Retryable(err.into(), TRANSIENT_RETRY_BASE)); } } diff --git a/bin/dipper-service/src/worker/queue.rs b/bin/dipper-service/src/worker/queue.rs index 3d53bb4e..b20951c1 100644 --- a/bin/dipper-service/src/worker/queue.rs +++ b/bin/dipper-service/src/worker/queue.rs @@ -18,6 +18,16 @@ where /// Pushes a message to the queue for immediate processing at `priority`. async fn push(&self, msg: M, priority: JobPriority) -> anyhow::Result; + /// Same as [`Queue::push`], but with a retry budget of this job's own + /// rather than the queue-wide default, for work whose deadline decides how + /// many attempts are worth making. + async fn push_with_max_retries( + &self, + msg: M, + priority: JobPriority, + max_retries: u32, + ) -> anyhow::Result; + /// Pulls a job from the queue async fn pop(&self) -> anyhow::Result>>; @@ -51,6 +61,21 @@ where .await } + async fn push_with_max_retries( + &self, + msg: M, + priority: JobPriority, + max_retries: u32, + ) -> anyhow::Result { + self.inner + .push( + JobBuilder::new(msg) + .priority(priority) + .max_retries(max_retries), + ) + .await + } + async fn pop(&self) -> anyhow::Result>> { self.inner.pop().await } @@ -63,10 +88,9 @@ where /// A listener for the queue job available notification pub struct QueueImplListener(PgQueueListener); -/// A source of "a job may be available" notifications. -/// -/// Abstracted behind a trait so the worker loop's degrade-to-polling behaviour -/// can be unit tested without a live Postgres `LISTEN`/`NOTIFY` connection. +/// A source of "a job may be available" notifications. Abstracted behind a +/// trait so the worker loop's degrade-to-polling behaviour can be unit tested +/// without a live Postgres `LISTEN`/`NOTIFY` connection. #[async_trait] pub trait JobNotifications: Send { /// Waits for the next job-available notification. diff --git a/bin/dipper-service/src/worker/result.rs b/bin/dipper-service/src/worker/result.rs index 24a02b35..1f40cfc0 100644 --- a/bin/dipper-service/src/worker/result.rs +++ b/bin/dipper-service/src/worker/result.rs @@ -14,6 +14,28 @@ pub fn calculate_backoff_delay(base_delay: Duration, attempt: u32) -> Duration { } } +/// Upper bound on what [`retries_within_window`] will return, so a window set +/// absurdly long cannot hand a job an unbounded budget. Attempts are 5 minutes +/// apart by then, so the ones given up buy little. +const MAX_RETRIES_PER_WINDOW: u32 = 64; + +/// How many retries fit inside `window`, each waiting +/// [`calculate_backoff_delay`] longer than the last. Ties a job's budget to the +/// deadline its work has to meet, not a fixed count that can run out early. +pub fn retries_within_window(window: Duration, base_delay: Duration) -> u32 { + let mut elapsed = Duration::ZERO; + + for retries in 0..MAX_RETRIES_PER_WINDOW { + let next = elapsed.saturating_add(calculate_backoff_delay(base_delay, retries)); + if next > window { + return retries; + } + elapsed = next; + } + + MAX_RETRIES_PER_WINDOW +} + /// The error type for job processing. #[derive(Debug, thiserror::Error)] pub enum JobError { @@ -70,6 +92,56 @@ mod tests { assert_eq!(calculate_backoff_delay(base, 100), Duration::from_secs(300)); } + /// The live case: a 600 second acceptance window and the 30 second base + /// delay an offer submission retries on. The 4 retries fall at 30, 90, 210 + /// and 450 seconds, and a 5th would wait until 930, past the deadline. + #[test] + fn test_retries_fill_the_default_acceptance_window() { + let window = Duration::from_secs(600); + let base = Duration::from_secs(30); + + assert_eq!(retries_within_window(window, base), 4); + + let mut elapsed = Duration::ZERO; + for retry in 0..4 { + elapsed += calculate_backoff_delay(base, retry); + } + assert_eq!(elapsed, Duration::from_secs(450)); + assert!(elapsed + calculate_backoff_delay(base, 4) > window); + + // The budget is sized on the shorter 5 second base, which fits 6. + assert_eq!(retries_within_window(window, Duration::from_secs(5)), 6); + } + + #[test] + fn test_no_retries_fit_a_window_shorter_than_the_first_delay() { + let window = Duration::from_secs(10); + let base = Duration::from_secs(30); + + assert_eq!(retries_within_window(window, base), 0); + } + + /// A retry landing exactly on the deadline still has time to be made. + #[test] + fn test_a_retry_landing_on_the_boundary_counts() { + let base = Duration::from_secs(30); + + assert_eq!(retries_within_window(Duration::from_secs(30), base), 1); + assert_eq!(retries_within_window(Duration::from_secs(29), base), 0); + } + + /// Past the exponential phase the delay is flat, so an unbounded window + /// would otherwise keep counting; the cap stops it. + #[test] + fn test_retries_within_window_is_capped() { + let huge = Duration::from_secs(u64::MAX / 2); + + assert_eq!( + retries_within_window(huge, Duration::from_secs(30)), + MAX_RETRIES_PER_WINDOW + ); + } + #[test] fn test_backoff_handles_overflow() { // Very large base delay should saturate rather than overflow diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 0fd22880..68c91448 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -12,11 +12,12 @@ use super::{ context::{Ctx, InnerCtx}, handlers::{ self, CancelRejectedAgreementOnChainCtx, ReassessIndexingRequestCtx, + SUBMIT_OFFER_DROPPED_TX_RETRY_BASE, SUBMIT_OFFER_TRANSIENT_RETRY_BASE, SendIndexingAgreementProposalCtx, SubmitOfferCtx, }, messages::Message, queue::{JobNotifications, Queue}, - result::{JobError, JobResult, calculate_backoff_delay}, + result::{JobError, JobResult, calculate_backoff_delay, retries_within_window}, }; pub use super::{ queue::JobPriority, @@ -447,9 +448,17 @@ where // receiver could only wake one of them. let (stop_tx, stop_rx) = watch::channel(false); + // An offer is worth resubmitting for as long as the indexer can still accept + // it, so the budget comes from that window. Sized on the shorter of the two + // backoffs, it cannot run out early whichever failure the job keeps hitting. + let submit_offer_max_retries = retries_within_window( + Duration::from_secs(agreement_conf.deadline_seconds()), + SUBMIT_OFFER_DROPPED_TX_RETRY_BASE.min(SUBMIT_OFFER_TRANSIENT_RETRY_BASE), + ); + let handle = Handle { stop_tx: stop_tx.clone(), - worker_queue_handle: WorkerQueueHandle::new(queue.clone()), + worker_queue_handle: WorkerQueueHandle::new(queue.clone(), submit_offer_max_retries), }; let fut = async move { // Built once, cloned per loop. Every field is Arc/Clone, so the clones @@ -468,7 +477,7 @@ where additional_networks, entity_count_cache, chain_listener_notify, - worker: WorkerQueueHandle::new(queue.clone()), + worker: WorkerQueueHandle::new(queue.clone(), submit_offer_max_retries), bypass_chain_clock_defenses, chain_listener_chain_id, reassess_lock, @@ -765,6 +774,15 @@ mod tests { anyhow::bail!("these tests never push") } + async fn push_with_max_retries( + &self, + _msg: Message, + _priority: JobPriority, + _max_retries: u32, + ) -> anyhow::Result { + anyhow::bail!("these tests never push") + } + async fn pop(&self) -> anyhow::Result>> { Ok(None) } diff --git a/bin/dipper-service/src/worker/service_queue.rs b/bin/dipper-service/src/worker/service_queue.rs index e847e502..064928d8 100644 --- a/bin/dipper-service/src/worker/service_queue.rs +++ b/bin/dipper-service/src/worker/service_queue.rs @@ -33,17 +33,18 @@ pub trait WorkerQueue { priority: JobPriority, ) -> anyhow::Result; - /// Cancel a rejected agreement on-chain. - /// - /// When an indexer rejected off-chain but accepted on-chain, this cancels - /// the agreement via `cancelIndexingAgreementByPayer`. + /// Cancel a rejected agreement on-chain. When an indexer rejected off-chain + /// but accepted on-chain, this cancels the agreement via + /// `cancelIndexingAgreementByPayer`. async fn cancel_rejected_agreement_on_chain( &self, agreement_id: IndexingAgreementId, priority: JobPriority, ) -> anyhow::Result; - /// Submit an RCA offer on-chain as the first step of a new proposal. + /// 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. async fn submit_offer( &self, agreement_id: IndexingAgreementId, @@ -59,12 +60,18 @@ pub trait WorkerQueue { #[derive(Clone)] pub struct WorkerQueueHandle { queue: Q, + /// Retries allowed on an offer submission, sized so the budget lasts as + /// long as the indexer's window to accept. See [`WorkerQueue::submit_offer`]. + submit_offer_max_retries: u32, } impl WorkerQueueHandle { /// Create a new instance of the worker queue handle - pub(super) fn new(queue: Q) -> Self { - Self { queue } + pub(super) fn new(queue: Q, submit_offer_max_retries: u32) -> Self { + Self { + queue, + submit_offer_max_retries, + } } } @@ -142,7 +149,7 @@ where priority: JobPriority, ) -> anyhow::Result { self.queue - .push( + .push_with_max_retries( Message::SubmitOffer(SubmitOffer { agreement_id, indexing_request_id, @@ -151,7 +158,122 @@ where deployment_chain_id, }), priority, + self.submit_offer_max_retries, + ) + .await + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use thegraph_core::deployment_id; + + use super::*; + use crate::worker::queue::{JobGuard, JobNotifications}; + + /// Records the retry budget each push carried, `None` for a push that took + /// the queue-wide default. + #[derive(Default)] + struct RecordingQueue { + pushes: Mutex>>, + } + + struct NeverNotifies; + + #[async_trait] + impl JobNotifications for NeverNotifies { + async fn wait_for_notification(&mut self) -> anyhow::Result<()> { + anyhow::bail!("these tests never subscribe") + } + } + + #[async_trait] + impl Queue for RecordingQueue { + type Listener = NeverNotifies; + + async fn push(&self, _msg: Message, _priority: JobPriority) -> anyhow::Result { + self.pushes.lock().unwrap().push(None); + Ok(JobId::default()) + } + + async fn push_with_max_retries( + &self, + _msg: Message, + _priority: JobPriority, + max_retries: u32, + ) -> anyhow::Result { + self.pushes.lock().unwrap().push(Some(max_retries)); + Ok(JobId::default()) + } + + async fn pop(&self) -> anyhow::Result>> { + Ok(None) + } + + async fn subscribe(&self) -> anyhow::Result { + anyhow::bail!("these tests never subscribe") + } + } + + fn handle(max_retries: u32) -> WorkerQueueHandle { + WorkerQueueHandle::new(RecordingQueue::default(), max_retries) + } + + /// The budget the worker sized from the acceptance window has to reach the + /// job. Falling back to the queue default is the bug this guards: 3 attempts + /// inside 95 seconds, then nothing for the rest of a 600 second window. + #[tokio::test] + async fn an_offer_submission_carries_the_window_sized_retry_budget() { + //* Arrange + let queue = handle(4); + + //* Act + queue + .submit_offer( + IndexingAgreementId::from_bytes([0; 16]), + IndexingRequestId::new(), + "https://indexer.example.com".parse().unwrap(), + deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"), + 1, + JobPriority::Background, + ) + .await + .unwrap(); + + //* Assert + assert_eq!(*queue.queue.pushes.lock().unwrap(), vec![Some(4)]); + } + + /// Only the offer submission has a deadline to spend its retries against, + /// so every other job keeps the queue-wide budget. + #[tokio::test] + async fn other_jobs_keep_the_queue_default_retry_budget() { + //* Arrange + let queue = handle(4); + + //* Act + queue + .send_indexing_agreement_proposal( + "https://indexer.example.com".parse().unwrap(), + IndexingAgreementId::from_bytes([0; 16]), + IndexingRequestId::new(), + deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"), + 1, + JobPriority::Background, + ) + .await + .unwrap(); + queue + .cancel_rejected_agreement_on_chain( + IndexingAgreementId::from_bytes([0; 16]), + JobPriority::Background, ) .await + .unwrap(); + + //* Assert + assert_eq!(*queue.queue.pushes.lock().unwrap(), vec![None, None]); } }