diff --git a/crates/manager-core/src/outbox_repo.rs b/crates/manager-core/src/outbox_repo.rs index 8ebd81e..a7652fc 100644 --- a/crates/manager-core/src/outbox_repo.rs +++ b/crates/manager-core/src/outbox_repo.rs @@ -133,6 +133,11 @@ pub trait OutboxRepo: Send + Sync { /// Returns how many rows were reclaimed. `attempt_count` is deliberately NOT /// bumped: the handler never ran, so a reclaim must not consume the row's /// retry budget (same reasoning as the transient queue `release`). + /// + /// SYNCHRONOUS rows (`reply_to IS NOT NULL`) are NOT reclaimed — the schema + /// makes `reply_to.is_some()` the "never retry" signal, and their caller is + /// long gone. They are deleted instead: nobody is waiting, they can never be + /// dispatched, and leaving them would leak a row per crashed sync dispatch. async fn reclaim_stale_claims(&self, timeout_secs: u32) -> Result; /// Remove a row after a terminal outcome (success or dead-letter). @@ -231,14 +236,34 @@ impl OutboxRepo for PostgresOutboxRepo { } async fn reclaim_stale_claims(&self, timeout_secs: u32) -> Result { + // `reply_to IS NULL` is load-bearing: a row with a reply_to is a SYNCHRONOUS + // HTTP dispatch, and the schema (0009_outbox.sql) makes reply_to.is_some() + // the "never retry" signal. Returning one to the queue would re-execute a + // handler whose caller left minutes ago and whose inbox oneshot no longer + // exists — running its side effects a second time for nobody. let res = sqlx::query( "UPDATE outbox SET claimed_at = NULL, claimed_by = NULL \ WHERE claimed_at IS NOT NULL \ + AND reply_to IS NULL \ AND claimed_at < NOW() - ($1 || ' seconds')::INTERVAL", ) .bind(i64::from(timeout_secs)) .execute(&self.pool) .await?; + + // A stale SYNC row can never be dispatched again (never-retry) and nobody + // is waiting on it, so it is pure garbage — drop it rather than leak a row + // per crashed sync dispatch forever. + sqlx::query( + "DELETE FROM outbox \ + WHERE claimed_at IS NOT NULL \ + AND reply_to IS NOT NULL \ + AND claimed_at < NOW() - ($1 || ' seconds')::INTERVAL", + ) + .bind(i64::from(timeout_secs)) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) }