Harden durable replay and lock recovery

crates/remora-durable/src/ledger.rs

@@ -39,6 +39,8 @@ impl MutationStatus {
     }
 }
 
+/// On conflict, `response` is `Null`; higher layers must map `disposition`
+/// instead of serializing it as a stored API response.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct MutationOutcome {
     pub disposition: MutationDisposition,
@@ -165,6 +167,9 @@ struct MutationInsert<'a> {
 
 fn identity_lock_key(scope: MutationScope<'_>) -> i64 {
     let mut hasher = Sha256::new();
+    hasher.update(b"remora.advisory.mutation.v1");
+    hasher.update(tx::ADVISORY_CLASS_MUTATION.to_be_bytes());
+    hasher.update([0xff]);
     hasher.update(scope.subject.as_str().as_bytes());
     hasher.update([0xff]);
     hasher.update(scope.operation.as_bytes());

crates/remora-durable/src/lib.rs

@@ -26,4 +26,7 @@ pub use outbox::{
     enqueue_after, mark_projected, mark_projected_pool, release_with_error, unprojected_count,
 };
 pub use pool::{PoolConfig, connect};
-pub use tx::{BoxFuture, PgTx, advisory_xact_lock, set_lock_timeout, with_transaction};
+pub use tx::{
+    ADVISORY_CLASS_APP, ADVISORY_CLASS_MUTATION, BoxFuture, PgTx, advisory_key, advisory_xact_lock,
+    set_lock_timeout, with_transaction,
+};

crates/remora-durable/src/outbox.rs

@@ -102,6 +102,11 @@ pub async fn claim_batch_in_tx(
     limit: i64,
     lease: time::Duration,
 ) -> Result<Vec<OutboxMessage>, DurableError> {
+    if limit <= 0 {
+        return Err(DurableError::Config(
+            "outbox claim limit must be positive".into(),
+        ));
+    }
     let lease_secs = lease.as_seconds_f64().max(0.0);
     let topic = match filter {
         OutboxClaimFilter::Any => None,
@@ -237,9 +242,8 @@ pub async fn mark_projected_pool(pool: &PgPool, claim: &OutboxClaim) -> Result<(
     .await
 }
 
-/// Release a claimed message for later retry. Fenced to the claim token so a
+/// Fenced release is idempotent for ambiguous commits; missing messages and
-/// stale claimant cannot clear a reclaimed lease. Retry availability uses the
+/// foreign claims fail distinctly, and retry timing uses the database clock.
-/// database clock.
 pub async fn release_with_error(
     tx: &mut PgTx<'_>,
     claim: &OutboxClaim,
@@ -265,13 +269,39 @@ pub async fn release_with_error(
     .bind(error)
     .execute(tx.as_mut())
     .await?;
-    if result.rows_affected() != 1 {
+    if result.rows_affected() == 1 {
-        return Err(DurableError::OutboxClaimMismatch {
+        return Ok(());
-            message_id: claim.message_id,
-        });
     }
+
+    let row: Option<(bool, Option<String>, i64, Option<String>)> = sqlx::query_as(
+        "SELECT
+            projected_at IS NOT NULL,
+            claimed_by,
+            claim_generation,
+            last_error
+         FROM outbox_messages
+         WHERE id = $1",
+    )
+    .bind(claim.message_id)
+    .fetch_optional(tx.as_mut())
+    .await?;
+
+    match row {
+        None => Err(DurableError::OutboxMissing {
+            message_id: claim.message_id,
+        }),
+        // Already released under this claim generation: unclaimed, unprojected,
+        // same generation, same last_error as this release attempt.
+        Some((false, None, generation, last_error))
+            if generation == claim.generation && last_error.as_deref() == Some(error) =>
+        {
             Ok(())
         }
+        Some(_) => Err(DurableError::OutboxClaimMismatch {
+            message_id: claim.message_id,
+        }),
+    }
+}
 
 pub async fn unprojected_count(
     pool: &PgPool,

crates/remora-durable/src/tx.rs

@@ -1,6 +1,7 @@
 use std::future::Future;
 use std::pin::Pin;
 
+use sha2::{Digest, Sha256};
 use sqlx::{PgPool, Postgres, Transaction};
 
 use crate::error::DurableError;
@@ -33,6 +34,29 @@ pub async fn set_lock_timeout(tx: &mut PgTx<'_>) -> Result<(), DurableError> {
     Ok(())
 }
 
+/// Advisory lock class reserved for mutation ledger identity locks.
+/// Feature packages must not use this class.
+pub const ADVISORY_CLASS_MUTATION: i32 = 1;
+
+/// Advisory lock class for application and feature-package locks.
+/// Prefer this (or higher package-owned classes) over the mutation class.
+pub const ADVISORY_CLASS_APP: i32 = 2;
+
+/// Mixes the class into a 64-bit key to separate lock domains without reducing
+/// the PostgreSQL `bigint` identity space.
+#[must_use]
+pub fn advisory_key(class: i32, key: i32) -> i64 {
+    let mut hasher = Sha256::new();
+    hasher.update(b"remora.advisory.v1");
+    hasher.update(class.to_be_bytes());
+    hasher.update(key.to_be_bytes());
+    let digest = hasher.finalize();
+    i64::from_be_bytes([
+        digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7],
+    ])
+}
+
+/// Transaction-scoped advisory lock on a domain-separated 64-bit key.
 pub async fn advisory_xact_lock(tx: &mut PgTx<'_>, key: i64) -> Result<(), DurableError> {
     sqlx::query("SELECT pg_advisory_xact_lock($1)")
         .bind(key)

crates/remora-durable/tests/substrate.rs

@@ -4,11 +4,11 @@ use std::time::Duration;
 use remora_domain::ports::MutationScope;
 use remora_domain::{OpaqueId, fingerprint};
 use remora_durable::{
-    CursorCodec, DurableError, KeysetPosition, MutationDisposition, MutationStatus, OutboxClaim,
+    ADVISORY_CLASS_APP, CursorCodec, DurableError, KeysetPosition, MutationDisposition,
-    OutboxClaimFilter, PoolConfig, advisory_xact_lock, cas_set_text, claim_batch,
+    MutationStatus, OutboxClaim, OutboxClaimFilter, PoolConfig, advisory_key, advisory_xact_lock,
-    claim_batch_in_tx, connect, enqueue, enqueue_after, execute_mutation, first_page_cursor,
+    cas_set_text, claim_batch, claim_batch_in_tx, connect, enqueue, enqueue_after,
-    mark_projected, mark_projected_pool, migrate, next_page_cursor, release_with_error,
+    execute_mutation, first_page_cursor, mark_projected, mark_projected_pool, migrate,
-    unprojected_count, with_transaction,
+    next_page_cursor, release_with_error, unprojected_count, with_transaction,
 };
 use serde_json::json;
 use sqlx::PgPool;
@@ -60,6 +60,26 @@ fn eligibility(attempt: &str, fence: u64, worker: &str) -> Vec<u8> {
     format!("attempt={attempt}|fence={fence}|worker={worker}").into_bytes()
 }
 
+async fn ensure_effect_probe_table(pool: &PgPool) {
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS substrate_effect_probe (
+            id TEXT PRIMARY KEY,
+            hits BIGINT NOT NULL
+         )",
+    )
+    .execute(pool)
+    .await
+    .expect("create effect probe table");
+}
+
+async fn clear_mutation_subject(pool: &PgPool, owner: &OpaqueId) {
+    sqlx::query("DELETE FROM mutation_records WHERE scope_subject = $1")
+        .bind(owner.as_str())
+        .execute(pool)
+        .await
+        .expect("clear mutation rows for subject");
+}
+
 #[tokio::test]
 async fn migration_round_trip_is_idempotent() {
     let Some(pool) = test_pool().await else {
@@ -131,13 +151,20 @@ async fn mutation_ledger_exact_replay_conflict_and_rejection() {
     let Some(pool) = test_pool().await else {
         return;
     };
+    ensure_effect_probe_table(&pool).await;
     let owner = subject("replay");
+    clear_mutation_subject(&pool, &owner).await;
     let operation = "submit";
     let key = "op-replay-0001";
     let request = json!({"source": "http", "bucket": "demo"});
     let digest = fingerprint(&request).expect("fingerprint");
     let claim = eligibility("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
     let probe = format!("effect-{key}");
+    sqlx::query("DELETE FROM substrate_effect_probe WHERE id = $1")
+        .bind(&probe)
+        .execute(&pool)
+        .await
+        .expect("clear probe");
 
     let first = execute_mutation(
         &pool,
@@ -207,6 +234,7 @@ async fn mutation_ledger_exact_replay_conflict_and_rejection() {
     .expect("conflict observation");
     assert_eq!(conflict.disposition, MutationDisposition::Conflict);
     assert_eq!(conflict.status, MutationStatus::Rejected);
+    assert!(conflict.response.is_null());
 
     let kind_conflict = execute_mutation(
         &pool,
@@ -309,12 +337,19 @@ async fn concurrent_first_observation_serializes_to_one_effect() {
     let Some(pool) = test_pool().await else {
         return;
     };
+    ensure_effect_probe_table(&pool).await;
     let owner = subject("race");
+    clear_mutation_subject(&pool, &owner).await;
     let operation = "submit";
     let key = "op-race-0001";
     let digest = fingerprint(&json!({"race": true})).expect("fingerprint");
     let claim = eligibility("attempt-raceaaaa", 1, "worker-raceaaaa");
     let probe = format!("effect-{key}");
+    sqlx::query("DELETE FROM substrate_effect_probe WHERE id = $1")
+        .bind(&probe)
+        .execute(&pool)
+        .await
+        .expect("clear probe");
     let workers = 8_usize;
     let barrier = Arc::new(Barrier::new(workers));
     let mut handles = Vec::with_capacity(workers);
@@ -382,14 +417,6 @@ async fn apply_effect(
     tx: &mut remora_durable::PgTx<'_>,
     probe_id: &str,
 ) -> Result<(), DurableError> {
-    sqlx::query(
-        "CREATE TABLE IF NOT EXISTS substrate_effect_probe (
-            id TEXT PRIMARY KEY,
-            hits BIGINT NOT NULL
-         )",
-    )
-    .execute(tx.as_mut())
-    .await?;
     sqlx::query(
         "INSERT INTO substrate_effect_probe (id, hits) VALUES ($1, 1)
          ON CONFLICT (id) DO UPDATE SET hits = substrate_effect_probe.hits + 1",
@@ -679,6 +706,38 @@ async fn outbox_claim_projection_recovery_and_claim_fencing() {
     .await
     .expect("release with error");
 
+    // Ambiguous-commit: same claim token may retry release successfully.
+    with_transaction(&pool, |tx| {
+        let claim = failed_claim[0].claim.clone();
+        Box::pin(async move {
+            release_with_error(tx, &claim, "projection failed", TimeDuration::seconds(60)).await
+        })
+    })
+    .await
+    .expect("idempotent release_with_error");
+
+    // Missing message id is distinct from claim mismatch.
+    let missing = with_transaction(&pool, |tx| {
+        Box::pin(async move {
+            release_with_error(
+                tx,
+                &OutboxClaim {
+                    message_id: i64::MAX,
+                    worker_id: "worker-fail".into(),
+                    generation: 1,
+                },
+                "missing",
+                TimeDuration::seconds(1),
+            )
+            .await
+        })
+    })
+    .await;
+    assert!(
+        matches!(missing, Err(DurableError::OutboxMissing { .. })),
+        "expected OutboxMissing, got {missing:?}"
+    );
+
     let early = claim_batch(
         &pool,
         "worker-early",
@@ -739,6 +798,19 @@ async fn outbox_claim_projection_recovery_and_claim_fencing() {
         unprojected_count(&pool, retry_filter).await.expect("count"),
         0
     );
+
+    let bad_limit = claim_batch(
+        &pool,
+        "worker-bad-limit",
+        batch_filter,
+        0,
+        TimeDuration::seconds(30),
+    )
+    .await;
+    assert!(
+        matches!(bad_limit, Err(DurableError::Config(_))),
+        "non-positive claim limit must fail closed: {bad_limit:?}"
+    );
 }
 
 #[tokio::test]
@@ -947,7 +1019,7 @@ async fn advisory_lock_serializes_critical_section() {
             barrier.wait().await;
             with_transaction(&pool, |tx| {
                 Box::pin(async move {
-                    advisory_xact_lock(tx, 42).await?;
+                    advisory_xact_lock(tx, advisory_key(ADVISORY_CLASS_APP, 42)).await?;
                     let (value,): (i64,) = sqlx::query_as(
                         "SELECT value FROM substrate_lock_probe WHERE id = 'lock-row'",
                     )