fix: keep retrying an offer while the indexer can accept it

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,
+};

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<R, T> {
     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));
         }
     }
 

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<JobId>;
 
+    /// 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<JobId>;
+
     /// Pulls a job from the queue
     async fn pop(&self) -> anyhow::Result<Option<JobGuard<'_, M>>>;
 
@@ -51,6 +61,21 @@ where
             .await
     }
 
+    async fn push_with_max_retries(
+        &self,
+        msg: M,
+        priority: JobPriority,
+        max_retries: u32,
+    ) -> anyhow::Result<JobId> {
+        self.inner
+            .push(
+                JobBuilder::new(msg)
+                    .priority(priority)
+                    .max_retries(max_retries),
+            )
+            .await
+    }
+
     async fn pop(&self) -> anyhow::Result<Option<JobGuard<'_, M>>> {
         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.

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

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<JobId> {
+            anyhow::bail!("these tests never push")
+        }
+
         async fn pop(&self) -> anyhow::Result<Option<JobGuard<'_, Message>>> {
             Ok(None)
         }