fix(webhooks): claim a repair attempt before requesting redelivery

`repair_delivery` read the repair row, checked the attempt cap and the
backoff, sent `POST /app/hook/deliveries/{id}/attempts`, and only then
wrote the attempt. Two watchdogs reading the same open row — an engine
overlapping its own restart, or two nodes on one store — could both pass
those checks and both ask GitHub to redeliver the same delivery, charge
it two attempts, and so reach the cap in half the real retries. The
retry pass added in 4d670b3 drives repairs from the table on every
poll, which is what made that window routine rather than incidental.

The attempt is now taken in one conditional store write before the
request goes out: `claim_webhook_redelivery` upserts with
`attempts = attempts + 1, last_attempt_at = .., resolved_at = NULL` but
only where the stored row still has the attempt count and the resolution
state the caller read. Exactly one claimant matches; the others get
`None` and send nothing. A close committed since the read also fails the
compare, so a pass can no longer resurrect a repair whose delivery has
landed. The per-GUID backoff ladder and the cap stay in the watchdog,
where the scan and the retry pass share them.

Both backends implement the claim as a single statement (`ON CONFLICT ..
DO UPDATE .. WHERE .. RETURNING`), which is the shape the webhook inbox
already uses to lease deliveries. New tests: the claim's compare on both
backends (attempt charged once, loser charges nothing, close stands,
resolved row reopens under the read resolution), and a two-node race on
one Postgres database proving one winner. Verified the SQL directly
against Postgres 18 and SQLite as well as through the crate's suites.

Author: Bnjoroge1

CHANGELOG.md

@@ -47,14 +47,14 @@ Releases before v0.27.0 predate the changelog.
   delivery watchdog treated every delivery in GitHub's history as a phantom
   ack when the store had no watchdog cursor, and asked GitHub to redeliver
   all of them. After a store cutover this re-ran CI for merged PRs and
-  superseded pushes — 2,435 redeliveries and hundreds of runs on one host —
-  and the replays queued ahead of live webhooks in the inbox. A store's
-  first poll now adopts the history before it (the watermark starts at the
-  grace boundary, drawn and persisted at the first attempt, so a history
-  outage right after startup cannot swallow the deliveries that failed while
-  polling was down); later passes repair phantom acks and edge failures as
-  before, and a restored snapshot keeps its cursor and still repairs
-  everything since. Recover a known gap by redelivering it from GitHub.
+  superseded pushes, and the replays queued ahead of live webhooks in the
+  inbox. A store's first poll now adopts the history before it (the
+  watermark starts at the grace boundary, drawn and persisted at the first
+  attempt, so a history outage right after startup cannot swallow the
+  deliveries that failed while polling was down); later passes repair
+  phantom acks and edge failures as before, and a restored snapshot keeps
+  its cursor and still repairs everything since. Recover a known gap by
+  redelivering it from GitHub.
 
 - **Judged-missing webhooks are no longer stranded below the watchdog
   watermark**: the watermark advances past everything a pass examines, so a
@@ -67,7 +67,10 @@ Releases before v0.27.0 predate the changelog.
   backoff and attempt cap as the scan, so untracked history is still never
   replayed. The retry scan reads this App's rows that are still below the
   attempt cap, so another App's backlog — or rows at the cap — cannot fill
-  the bounded window and starve newer repairs out of it.
+  the bounded window and starve newer repairs out of it. An attempt is
+  claimed before the request, in one conditional store write, so two
+  watchdogs overlapping a restart cannot both request or both charge the
+  same delivery.
 
 - **The conformance campaign survives a second run against the same
   Postgres database**: `conformance-5repos.sh` wiped the campaign home,

crates/preloop-runner-server/src/control/backend.rs

@@ -1024,6 +1024,20 @@ pub(crate) trait ControlBackend: Send + Sync {
         attempt_cap: u32,
         limit: usize,
     ) -> Result<Vec<WebhookRedeliveryRecord>, ControlError>;
+    /// Atomically take one redelivery attempt for a repair row.
+    ///
+    /// Succeeds only when the stored row still looks exactly like `observed`
+    /// — same attempt count, same resolution state — and charges the attempt
+    /// and the last-attempt clock in the same statement, returning the
+    /// claimed row. A second claimant (a watchdog overlapping a restart, a
+    /// second server on the store) reads a row that no longer matches what it
+    /// read, gets `None`, and must not send the request: the winner's attempt
+    /// already covers the delivery.
+    async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError>;
     async fn resolve_webhook_redelivery(
         &self,
         delivery_guid: &str,
@@ -2874,6 +2888,16 @@ impl ControlBackend for Backend {
             }
         }
     }
+    async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError> {
+        match self {
+            Self::Sqlite(b) => b.claim_webhook_redelivery(observed, claimed_at_us).await,
+            Self::Postgres(b) => b.claim_webhook_redelivery(observed, claimed_at_us).await,
+        }
+    }
     async fn resolve_webhook_redelivery(
         &self,
         delivery_guid: &str,

crates/preloop-runner-server/src/control/lite/impls.rs

@@ -732,6 +732,13 @@ impl ControlBackend for LiteBackend {
         self.retryable_webhook_redeliveries(app_id, attempt_cap, limit)
             .await
     }
+    async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError> {
+        self.claim_webhook_redelivery(observed, claimed_at_us).await
+    }
     async fn resolve_webhook_redelivery(
         &self,
         delivery_guid: &str,

crates/preloop-runner-server/src/control/lite/webhooks.rs

@@ -760,6 +760,58 @@ impl LiteBackend {
         })
     }
 
+    /// Take one redelivery attempt for a repair row, or report that someone
+    /// else already did.
+    ///
+    /// The `WHERE` on the conflict path is the compare of a compare-and-swap
+    /// over the row the caller read: an attempt committed since that read
+    /// (another watchdog, a restart overlap, a second server on the store)
+    /// no longer matches, so the update is skipped, nothing is returned, and
+    /// the caller sends no request. The charge and the last-attempt clock
+    /// move in the same statement, so a claimed attempt is exactly one
+    /// request. `IS` is null-safe equality, so an open row (`resolved_at`
+    /// NULL) matches a NULL `resolved_at` and a closed row only reopens
+    /// under the resolution the caller read.
+    pub(crate) async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError> {
+        self.write(|tx| {
+            tx.query_row(
+                &format!(
+                    "INSERT INTO webhook_redeliveries (delivery_guid, github_delivery_id, app_id, \
+                     reason, attempts, first_seen_at, last_attempt_at, resolved_at, last_error) \
+                     VALUES (?1, ?2, ?3, ?4, 1, ?5, ?6, NULL, NULL) \
+                     ON CONFLICT (delivery_guid) DO UPDATE SET \
+                     github_delivery_id = excluded.github_delivery_id, \
+                     app_id = excluded.app_id, reason = excluded.reason, \
+                     attempts = webhook_redeliveries.attempts + 1, \
+                     last_attempt_at = excluded.last_attempt_at, \
+                     resolved_at = NULL, last_error = NULL \
+                     WHERE webhook_redeliveries.attempts = ?7 \
+                     AND webhook_redeliveries.resolved_at IS ?8 \
+                     RETURNING {REDELIVERY_COLUMNS}"
+                ),
+                params![
+                    observed.delivery_guid,
+                    observed.github_delivery_id,
+                    observed.app_id,
+                    observed.reason.as_str(),
+                    observed.first_seen_at_us,
+                    claimed_at_us,
+                    observed.attempts as i64,
+                    observed.resolved_at_us,
+                ],
+                redelivery_row,
+            )
+            .optional()
+            .map_err(db)?
+            .map(finish_redelivery)
+            .transpose()
+        })
+    }
+
     /// Close an open repair; `false` when unknown or already resolved.
     pub(crate) async fn resolve_webhook_redelivery(
         &self,

crates/preloop-runner-server/src/control/pg/trait_impl.rs

@@ -734,6 +734,13 @@ impl ControlBackend for PgBackend {
         self.retryable_webhook_redeliveries(app_id, attempt_cap, limit)
             .await
     }
+    async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError> {
+        self.claim_webhook_redelivery(observed, claimed_at_us).await
+    }
     async fn resolve_webhook_redelivery(
         &self,
         delivery_guid: &str,

crates/preloop-runner-server/src/control/pg/webhooks.rs

@@ -759,6 +759,66 @@ impl PgBackend {
             .collect()
     }
 
+    /// Take one redelivery attempt for a repair row, or report that someone
+    /// else already did.
+    ///
+    /// Statement: `INSERT .. ON CONFLICT (delivery_guid) DO UPDATE SET
+    /// attempts = r.attempts + 1, .. WHERE r.attempts = $7 AND r.resolved_at
+    /// IS NOT DISTINCT FROM $8 RETURNING ..`.
+    ///
+    /// The `WHERE` is the compare of a compare-and-swap over the row the
+    /// caller read: an attempt committed since that read (another watchdog,
+    /// a restart overlap, a second server) no longer matches, so the update
+    /// is skipped, nothing is returned, and the caller sends no request. The
+    /// charge and the last-attempt clock move in the same statement, so a
+    /// claimed attempt is exactly one request. Under `READ COMMITTED` the
+    /// conflict path re-checks the condition against the row's latest
+    /// committed version, which is what makes the race safe.
+    pub(super) async fn claim_webhook_redelivery(
+        &self,
+        observed: &WebhookRedeliveryRecord,
+        claimed_at_us: i64,
+    ) -> Result<Option<WebhookRedeliveryRecord>, ControlError> {
+        let client = self.writer().await?;
+        let sql = format!(
+            "INSERT INTO webhook_redeliveries AS r (delivery_guid, github_delivery_id, app_id, \
+             reason, attempts, first_seen_at, last_attempt_at, resolved_at, last_error) \
+             VALUES ($1, $2, $3, $4, 1, {}, {}, NULL, NULL) \
+             ON CONFLICT (delivery_guid) DO UPDATE SET \
+             github_delivery_id = EXCLUDED.github_delivery_id, \
+             app_id = EXCLUDED.app_id, reason = EXCLUDED.reason, \
+             attempts = r.attempts + 1, last_attempt_at = EXCLUDED.last_attempt_at, \
+             resolved_at = NULL, last_error = NULL \
+             WHERE r.attempts = $7 AND r.resolved_at IS NOT DISTINCT FROM {} \
+             RETURNING {REDELIVERY_COLUMNS}",
+            // The two timestamps are microseconds on the wire, so the
+            // comparison against `resolved_at` needs the same epoch
+            // conversion the other timestamp columns use.
+            ts!("$5"),
+            ts!("$6"),
+            ts!("$8")
+        );
+        client
+            .query_opt(
+                &sql,
+                &[
+                    &observed.delivery_guid,
+                    &observed.github_delivery_id,
+                    &observed.app_id,
+                    &observed.reason.as_str(),
+                    &observed.first_seen_at_us,
+                    &claimed_at_us,
+                    &(observed.attempts.min(i32::MAX as u32) as i32),
+                    &observed.resolved_at_us,
+                ],
+            )
+            .await
+            .map_err(db)?
+            .as_ref()
+            .map(redelivery_from_row)
+            .transpose()
+    }
+
     /// Mark a repair row resolved (first resolution wins).
     ///
     /// Statement: `UPDATE webhook_redeliveries SET resolved_at = $2 WHERE