Add PostgreSQL DurableStore substrate for WP-03

Introduce remora-durable with pool/migrations, a RemoraApiMutation-shaped
ledger, claim-fenced outbox projection, CAS helpers, and keyset cursors,
plus a kind Job acceptance suite against the hermetic fixture.

crates/remora-durable/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "remora-durable"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+base64.workspace = true
+hmac.workspace = true
+remora-domain.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+sha2.workspace = true
+sqlx.workspace = true
+thiserror.workspace = true
+time.workspace = true
+tokio.workspace = true
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }

crates/remora-durable/migrations/0001_substrate.sql

@@ -0,0 +1,38 @@
+-- DurableStore substrate tables owned by remora-durable.
+-- Feature packages add their own migrations; they do not alter these tables.
+
+CREATE TABLE mutation_records (
+    scope_subject TEXT NOT NULL,
+    scope_operation TEXT NOT NULL,
+    operation_key TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    -- Opaque RemoraApiMutation eligibility / authority binding (attempt, fence,
+    -- worker incarnation, or other claim material). Compared for exact replay.
+    eligibility_claim BYTEA NOT NULL,
+    request_fingerprint BYTEA NOT NULL CHECK (octet_length(request_fingerprint) = 32),
+    response_status TEXT NOT NULL CHECK (response_status IN ('accepted', 'rejected')),
+    response_body JSONB NOT NULL,
+    effect_applied BOOLEAN NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (scope_subject, scope_operation, operation_key),
+    CHECK (effect_applied = (response_status = 'accepted'))
+);
+
+CREATE TABLE outbox_messages (
+    id BIGSERIAL PRIMARY KEY,
+    topic TEXT NOT NULL,
+    payload JSONB NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    available_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    claimed_by TEXT,
+    claimed_until TIMESTAMPTZ,
+    -- Monotonic per-row claim epoch; with claimed_by forms the ownership token.
+    claim_generation BIGINT NOT NULL DEFAULT 0,
+    projected_at TIMESTAMPTZ,
+    attempt_count INTEGER NOT NULL DEFAULT 0,
+    last_error TEXT
+);
+
+CREATE INDEX outbox_messages_claimable_idx
+    ON outbox_messages (available_at, id)
+    WHERE projected_at IS NULL;

crates/remora-durable/src/cas.rs

@@ -0,0 +1,90 @@
+use sqlx::{AssertSqlSafe, PgPool};
+
+use crate::error::DurableError;
+use crate::tx::{self, PgTx};
+
+/// Optimistic compare-and-swap on a caller-owned table.
+///
+/// The table must expose `revision BIGINT NOT NULL` and a primary key column
+/// bound as `$1`. Feature packages own the surrounding schema; this helper only
+/// serializes the revision bump. Identifiers are restricted to lowercase SQL
+/// identifiers before interpolation.
+pub async fn cas_set_text(
+    tx: &mut PgTx<'_>,
+    table: &str,
+    id_column: &str,
+    id: &str,
+    value_column: &str,
+    value: &str,
+    expected_revision: i64,
+) -> Result<i64, DurableError> {
+    validate_ident(table)?;
+    validate_ident(id_column)?;
+    validate_ident(value_column)?;
+
+    let sql = format!(
+        "UPDATE {table}
+         SET {value_column} = $2,
+             revision = revision + 1
+         WHERE {id_column} = $1
+           AND revision = $3
+         RETURNING revision"
+    );
+    let next: Option<(i64,)> = sqlx::query_as(AssertSqlSafe(sql))
+        .bind(id)
+        .bind(value)
+        .bind(expected_revision)
+        .fetch_optional(tx.as_mut())
+        .await?;
+    next.map(|(revision,)| revision)
+        .ok_or(DurableError::CasConflict)
+}
+
+pub async fn cas_set_text_pool(
+    pool: &PgPool,
+    table: &str,
+    id_column: &str,
+    id: &str,
+    value_column: &str,
+    value: &str,
+    expected_revision: i64,
+) -> Result<i64, DurableError> {
+    let table = table.to_owned();
+    let id_column = id_column.to_owned();
+    let id = id.to_owned();
+    let value_column = value_column.to_owned();
+    let value = value.to_owned();
+    tx::with_transaction(pool, move |tx| {
+        Box::pin(async move {
+            cas_set_text(
+                tx,
+                &table,
+                &id_column,
+                &id,
+                &value_column,
+                &value,
+                expected_revision,
+            )
+            .await
+        })
+    })
+    .await
+}
+
+fn validate_ident(value: &str) -> Result<(), DurableError> {
+    if value.is_empty()
+        || value.len() > 63
+        || !value.chars().all(|character| {
+            character.is_ascii_lowercase() || character.is_ascii_digit() || character == '_'
+        })
+        || value
+            .chars()
+            .next()
+            .is_none_or(|character| !(character.is_ascii_lowercase() || character == '_'))
+    {
+        return Err(DurableError::Config(format!(
+            "invalid SQL identifier '{value}'"
+        )));
+    }
+    Ok(())
+}

crates/remora-durable/src/cursor.rs

@@ -0,0 +1,231 @@
+use base64::Engine;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD;
+use hmac::{Hmac, KeyInit, Mac};
+use remora_domain::fingerprint;
+use serde::{Deserialize, Serialize};
+use sha2::Sha256;
+use time::OffsetDateTime;
+
+use crate::error::DurableError;
+
+type HmacSha256 = Hmac<Sha256>;
+
+/// Stable keyset order for job-style listings: `(created_at DESC, id DESC)`.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct KeysetPosition {
+    #[serde(with = "time::serde::rfc3339")]
+    pub created_at: OffsetDateTime,
+    pub id: String,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+struct CursorPayload {
+    subject: String,
+    filter_fingerprint: String,
+    #[serde(with = "time::serde::rfc3339")]
+    high_water_created_at: OffsetDateTime,
+    high_water_id: String,
+    after: Option<KeysetPosition>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct KeysetCursor {
+    pub subject: String,
+    pub filter_fingerprint: [u8; 32],
+    pub high_water: KeysetPosition,
+    pub after: Option<KeysetPosition>,
+}
+
+#[derive(Clone, Debug)]
+pub struct CursorCodec {
+    secret: Vec<u8>,
+}
+
+impl CursorCodec {
+    pub fn new(secret: impl Into<Vec<u8>>) -> Result<Self, DurableError> {
+        let secret = secret.into();
+        if secret.len() < 16 {
+            return Err(DurableError::Config(
+                "cursor integrity secret must be at least 16 bytes".into(),
+            ));
+        }
+        Ok(Self { secret })
+    }
+
+    pub fn filter_fingerprint(filters: &impl Serialize) -> Result<[u8; 32], DurableError> {
+        let digest = fingerprint(filters).map_err(|_| {
+            DurableError::Config("filters cannot be represented as canonical JSON".into())
+        })?;
+        Ok(*digest.as_bytes())
+    }
+
+    pub fn encode(&self, cursor: &KeysetCursor) -> Result<String, DurableError> {
+        let payload = CursorPayload {
+            subject: cursor.subject.clone(),
+            filter_fingerprint: hex::encode(cursor.filter_fingerprint),
+            high_water_created_at: cursor.high_water.created_at,
+            high_water_id: cursor.high_water.id.clone(),
+            after: cursor.after.clone(),
+        };
+        let body = serde_json::to_vec(&payload).map_err(|error| {
+            DurableError::Config(format!("cursor payload is not JSON-serializable: {error}"))
+        })?;
+        let mut mac = HmacSha256::new_from_slice(&self.secret)
+            .map_err(|error| DurableError::Config(format!("invalid cursor secret: {error}")))?;
+        mac.update(&body);
+        let tag = mac.finalize().into_bytes();
+
+        let mut framed = Vec::with_capacity(body.len() + tag.len());
+        framed.extend_from_slice(&body);
+        framed.extend_from_slice(&tag);
+        Ok(URL_SAFE_NO_PAD.encode(framed))
+    }
+
+    pub fn decode(&self, token: &str) -> Result<KeysetCursor, DurableError> {
+        let framed = URL_SAFE_NO_PAD
+            .decode(token.as_bytes())
+            .map_err(|_| DurableError::InvalidCursor)?;
+        if framed.len() <= 32 {
+            return Err(DurableError::InvalidCursor);
+        }
+        let (body, tag) = framed.split_at(framed.len() - 32);
+        let mut mac = HmacSha256::new_from_slice(&self.secret)
+            .map_err(|error| DurableError::Config(format!("invalid cursor secret: {error}")))?;
+        mac.update(body);
+        mac.verify_slice(tag)
+            .map_err(|_| DurableError::InvalidCursor)?;
+
+        let payload: CursorPayload =
+            serde_json::from_slice(body).map_err(|_| DurableError::InvalidCursor)?;
+        let filter_fingerprint = parse_hex32(&payload.filter_fingerprint)?;
+        Ok(KeysetCursor {
+            subject: payload.subject,
+            filter_fingerprint,
+            high_water: KeysetPosition {
+                created_at: payload.high_water_created_at,
+                id: payload.high_water_id,
+            },
+            after: payload.after,
+        })
+    }
+
+    pub fn bind(
+        &self,
+        token: &str,
+        subject: &str,
+        filter_fingerprint: [u8; 32],
+    ) -> Result<KeysetCursor, DurableError> {
+        let cursor = self.decode(token)?;
+        if cursor.subject != subject || cursor.filter_fingerprint != filter_fingerprint {
+            return Err(DurableError::CursorBindingMismatch);
+        }
+        Ok(cursor)
+    }
+}
+
+/// Build a first-page cursor after returning one page.
+///
+/// `high_water` is the newest visible row at snapshot time (membership bound).
+/// `last_on_page` is the last row of the returned page and becomes the exclusive
+/// keyset lower bound for the next page. For a single-item page the two
+/// positions are equal.
+pub fn first_page_cursor(
+    subject: impl Into<String>,
+    filter_fingerprint: [u8; 32],
+    high_water: KeysetPosition,
+    last_on_page: KeysetPosition,
+) -> KeysetCursor {
+    KeysetCursor {
+        subject: subject.into(),
+        filter_fingerprint,
+        high_water,
+        after: Some(last_on_page),
+    }
+}
+
+/// Advance a keyset cursor after returning `last` on the current page.
+pub fn next_page_cursor(current: &KeysetCursor, last: KeysetPosition) -> KeysetCursor {
+    KeysetCursor {
+        subject: current.subject.clone(),
+        filter_fingerprint: current.filter_fingerprint,
+        high_water: current.high_water.clone(),
+        after: Some(last),
+    }
+}
+
+fn parse_hex32(value: &str) -> Result<[u8; 32], DurableError> {
+    if value.len() != 64 {
+        return Err(DurableError::InvalidCursor);
+    }
+    let mut out = [0_u8; 32];
+    for (index, slot) in out.iter_mut().enumerate() {
+        *slot = u8::from_str_radix(&value[index * 2..index * 2 + 2], 16)
+            .map_err(|_| DurableError::InvalidCursor)?;
+    }
+    Ok(out)
+}
+
+// Minimal hex encode without an extra dependency.
+mod hex {
+    pub fn encode(bytes: [u8; 32]) -> String {
+        const HEX: &[u8; 16] = b"0123456789abcdef";
+        let mut output = String::with_capacity(64);
+        for byte in bytes {
+            output.push(char::from(HEX[usize::from(byte >> 4)]));
+            output.push(char::from(HEX[usize::from(byte & 0x0f)]));
+        }
+        output
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+    use time::macros::datetime;
+
+    #[test]
+    fn round_trips_and_binds_subject_and_filters() {
+        let codec = CursorCodec::new(b"test-cursor-secret").unwrap();
+        let filters = json!({"bucket": "alpha", "lifecycle": "visible"});
+        let fingerprint = CursorCodec::filter_fingerprint(&filters).unwrap();
+        let newest = KeysetPosition {
+            created_at: datetime!(2026-07-16 12:00:00 UTC),
+            id: "job-zzzzzzzz".into(),
+        };
+        let last = KeysetPosition {
+            created_at: datetime!(2026-07-16 11:00:00 UTC),
+            id: "job-aaaaaaaa".into(),
+        };
+        let cursor = first_page_cursor("subject-001", fingerprint, newest, last);
+        let token = codec.encode(&cursor).unwrap();
+        let decoded = codec.bind(&token, "subject-001", fingerprint).unwrap();
+        assert_eq!(decoded, cursor);
+        assert!(matches!(
+            codec.bind(&token, "other-subject", fingerprint),
+            Err(DurableError::CursorBindingMismatch)
+        ));
+        let other = CursorCodec::filter_fingerprint(&json!({"bucket": "beta"})).unwrap();
+        assert!(matches!(
+            codec.bind(&token, "subject-001", other),
+            Err(DurableError::CursorBindingMismatch)
+        ));
+    }
+
+    #[test]
+    fn rejects_tampered_tokens() {
+        let codec = CursorCodec::new(b"test-cursor-secret").unwrap();
+        let fingerprint = [7_u8; 32];
+        let position = KeysetPosition {
+            created_at: datetime!(2026-07-16 12:00:00 UTC),
+            id: "job-aaaaaaaa".into(),
+        };
+        let cursor = first_page_cursor("subject-001", fingerprint, position.clone(), position);
+        let mut token = codec.encode(&cursor).unwrap();
+        token.replace_range(0..1, if token.starts_with('A') { "B" } else { "A" });
+        assert!(matches!(
+            codec.decode(&token),
+            Err(DurableError::InvalidCursor)
+        ));
+    }
+}

crates/remora-durable/src/error.rs

@@ -0,0 +1,41 @@
+use std::fmt;
+
+#[derive(Debug, thiserror::Error)]
+pub enum DurableError {
+    #[error("database error: {0}")]
+    Database(#[from] sqlx::Error),
+    #[error("migration error: {0}")]
+    Migrate(#[from] sqlx::migrate::MigrateError),
+    /// Transport-layer mapping of a ledger conflict disposition.
+    ///
+    /// The substrate ledger returns
+    /// `MutationOutcome { disposition: Conflict, … }` for RemoraApiMutation
+    /// conflict observations. Higher layers may convert that outcome into this
+    /// error when an HTTP or worker adapter wants an error path.
+    #[error("mutation conflict for subject={subject} operation={operation} key={key}")]
+    MutationConflict {
+        subject: String,
+        operation: String,
+        key: String,
+    },
+    #[error("mutation effect failed: {0}")]
+    Effect(String),
+    #[error("outbox claim does not own message {message_id}")]
+    OutboxClaimMismatch { message_id: i64 },
+    #[error("outbox message {message_id} is missing")]
+    OutboxMissing { message_id: i64 },
+    #[error("invalid keyset cursor")]
+    InvalidCursor,
+    #[error("keyset cursor binding mismatch")]
+    CursorBindingMismatch,
+    #[error("compare-and-swap precondition failed")]
+    CasConflict,
+    #[error("configuration error: {0}")]
+    Config(String),
+}
+
+impl DurableError {
+    pub fn effect(error: impl fmt::Display) -> Self {
+        Self::Effect(error.to_string())
+    }
+}

crates/remora-durable/src/ledger.rs

@@ -0,0 +1,260 @@
+use remora_domain::RequestFingerprint;
+use remora_domain::ports::MutationScope;
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+use sqlx::PgPool;
+
+use crate::error::DurableError;
+use crate::tx::{self, BoxFuture, PgTx};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum MutationDisposition {
+    First,
+    Replay,
+    Conflict,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum MutationStatus {
+    Accepted,
+    Rejected,
+}
+
+impl MutationStatus {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Accepted => "accepted",
+            Self::Rejected => "rejected",
+        }
+    }
+
+    fn parse(value: &str) -> Result<Self, DurableError> {
+        match value {
+            "accepted" => Ok(Self::Accepted),
+            "rejected" => Ok(Self::Rejected),
+            other => Err(DurableError::Config(format!(
+                "unknown mutation status '{other}'"
+            ))),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct MutationOutcome {
+    pub disposition: MutationDisposition,
+    pub status: MutationStatus,
+    pub response: Value,
+}
+
+#[derive(Clone, Debug)]
+struct StoredRecord {
+    kind: String,
+    eligibility_claim: Vec<u8>,
+    request_fingerprint: Vec<u8>,
+    response_status: String,
+    response_body: Value,
+}
+
+/// Execute a mutation refining `RemoraApiMutation`.
+///
+/// Identity is `(subject, operation, key)`. Exact kind, eligibility claim, and
+/// request fingerprint replay the stored response. A change to any of those
+/// fields is a conflict and leaves the record unchanged. The eligibility claim
+/// binds attempt, fence, worker incarnation, or other authority material
+/// outside the semantic payload fingerprint.
+///
+/// Concurrent first observations for the same identity serialize on a
+/// transaction-scoped advisory lock before eligibility or effects run.
+pub async fn execute_mutation<D>(
+    pool: &PgPool,
+    scope: MutationScope<'_>,
+    kind: &str,
+    eligibility_claim: &[u8],
+    request_fingerprint: RequestFingerprint,
+    decide: D,
+) -> Result<MutationOutcome, DurableError>
+where
+    D: for<'c> FnOnce(
+            &'c mut PgTx<'_>,
+        ) -> BoxFuture<'c, Result<(MutationStatus, Value), DurableError>>
+        + Send
+        + 'static,
+{
+    let subject = scope.subject.clone();
+    let operation = scope.operation.to_owned();
+    let key = scope.key.to_owned();
+    let kind = kind.to_owned();
+    let eligibility_claim = eligibility_claim.to_owned();
+
+    tx::with_transaction(pool, move |tx| {
+        Box::pin(async move {
+            let scope = MutationScope {
+                subject: &subject,
+                operation: &operation,
+                key: &key,
+            };
+            execute_mutation_in_tx(
+                tx,
+                scope,
+                &kind,
+                &eligibility_claim,
+                request_fingerprint,
+                decide,
+            )
+            .await
+        })
+    })
+    .await
+}
+
+pub async fn execute_mutation_in_tx<D>(
+    tx: &mut PgTx<'_>,
+    scope: MutationScope<'_>,
+    kind: &str,
+    eligibility_claim: &[u8],
+    request_fingerprint: RequestFingerprint,
+    decide: D,
+) -> Result<MutationOutcome, DurableError>
+where
+    D: for<'c> FnOnce(
+            &'c mut PgTx<'_>,
+        ) -> BoxFuture<'c, Result<(MutationStatus, Value), DurableError>>
+        + Send,
+{
+    tx::set_lock_timeout(tx).await?;
+    // Serialize empty-row first observations: FOR UPDATE only locks existing
+    // rows, so concurrent first writers need an exclusive identity claim first.
+    tx::advisory_xact_lock(tx, identity_lock_key(scope)).await?;
+    let fingerprint = request_fingerprint.as_bytes().as_slice();
+
+    if let Some(existing) = load_record(tx, scope).await? {
+        return observe_existing(kind, eligibility_claim, fingerprint, existing);
+    }
+
+    let (status, response) = decide(tx).await?;
+    let effect_applied = status == MutationStatus::Accepted;
+    insert_record(
+        tx,
+        &MutationInsert {
+            scope,
+            kind,
+            eligibility_claim,
+            fingerprint,
+            status,
+            response_body: &response,
+            effect_applied,
+        },
+    )
+    .await?;
+    Ok(MutationOutcome {
+        disposition: MutationDisposition::First,
+        status,
+        response,
+    })
+}
+
+struct MutationInsert<'a> {
+    scope: MutationScope<'a>,
+    kind: &'a str,
+    eligibility_claim: &'a [u8],
+    fingerprint: &'a [u8],
+    status: MutationStatus,
+    response_body: &'a Value,
+    effect_applied: bool,
+}
+
+fn identity_lock_key(scope: MutationScope<'_>) -> i64 {
+    let mut hasher = Sha256::new();
+    hasher.update(scope.subject.as_str().as_bytes());
+    hasher.update([0xff]);
+    hasher.update(scope.operation.as_bytes());
+    hasher.update([0xff]);
+    hasher.update(scope.key.as_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],
+    ])
+}
+
+fn observe_existing(
+    kind: &str,
+    eligibility_claim: &[u8],
+    fingerprint: &[u8],
+    existing: StoredRecord,
+) -> Result<MutationOutcome, DurableError> {
+    if existing.kind != kind
+        || existing.eligibility_claim != eligibility_claim
+        || existing.request_fingerprint != fingerprint
+    {
+        return Ok(MutationOutcome {
+            disposition: MutationDisposition::Conflict,
+            status: MutationStatus::Rejected,
+            response: Value::Null,
+        });
+    }
+    let status = MutationStatus::parse(&existing.response_status)?;
+    Ok(MutationOutcome {
+        disposition: MutationDisposition::Replay,
+        status,
+        response: existing.response_body,
+    })
+}
+
+async fn load_record(
+    tx: &mut PgTx<'_>,
+    scope: MutationScope<'_>,
+) -> Result<Option<StoredRecord>, DurableError> {
+    let row = sqlx::query_as::<_, (String, Vec<u8>, Vec<u8>, String, Value)>(
+        "SELECT kind, eligibility_claim, request_fingerprint, response_status, response_body
+         FROM mutation_records
+         WHERE scope_subject = $1
+           AND scope_operation = $2
+           AND operation_key = $3
+         FOR UPDATE",
+    )
+    .bind(scope.subject.as_str())
+    .bind(scope.operation)
+    .bind(scope.key)
+    .fetch_optional(tx.as_mut())
+    .await?;
+
+    Ok(row.map(
+        |(kind, eligibility_claim, request_fingerprint, response_status, response_body)| {
+            StoredRecord {
+                kind,
+                eligibility_claim,
+                request_fingerprint,
+                response_status,
+                response_body,
+            }
+        },
+    ))
+}
+
+async fn insert_record(tx: &mut PgTx<'_>, record: &MutationInsert<'_>) -> Result<(), DurableError> {
+    sqlx::query(
+        "INSERT INTO mutation_records (
+            scope_subject,
+            scope_operation,
+            operation_key,
+            kind,
+            eligibility_claim,
+            request_fingerprint,
+            response_status,
+            response_body,
+            effect_applied
+         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
+    )
+    .bind(record.scope.subject.as_str())
+    .bind(record.scope.operation)
+    .bind(record.scope.key)
+    .bind(record.kind)
+    .bind(record.eligibility_claim)
+    .bind(record.fingerprint)
+    .bind(record.status.as_str())
+    .bind(record.response_body)
+    .bind(record.effect_applied)
+    .execute(tx.as_mut())
+    .await?;
+    Ok(())
+}

crates/remora-durable/src/lib.rs

@@ -0,0 +1,29 @@
+//! PostgreSQL DurableStore substrate.
+//!
+//! Provides the pool, migration runner, transaction helpers, generic mutation
+//! ledger refining `RemoraApiMutation`, durable outbox claim/projection, and
+//! integrity-protected keyset cursor primitives. Feature packages own their
+//! domain schema additions.
+
+mod cas;
+mod cursor;
+mod error;
+mod ledger;
+mod migrate;
+mod outbox;
+mod pool;
+mod tx;
+
+pub use cas::{cas_set_text, cas_set_text_pool};
+pub use cursor::{CursorCodec, KeysetCursor, KeysetPosition, first_page_cursor, next_page_cursor};
+pub use error::DurableError;
+pub use ledger::{
+    MutationDisposition, MutationOutcome, MutationStatus, execute_mutation, execute_mutation_in_tx,
+};
+pub use migrate::{migrate, migrator};
+pub use outbox::{
+    OutboxClaim, OutboxClaimFilter, OutboxMessage, claim_batch, claim_batch_in_tx, enqueue,
+    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};

crates/remora-durable/src/migrate.rs

@@ -0,0 +1,14 @@
+use sqlx::PgPool;
+
+use crate::error::DurableError;
+
+static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
+
+pub async fn migrate(pool: &PgPool) -> Result<(), DurableError> {
+    MIGRATOR.run(pool).await?;
+    Ok(())
+}
+
+pub fn migrator() -> &'static sqlx::migrate::Migrator {
+    &MIGRATOR
+}

crates/remora-durable/src/outbox.rs

@@ -0,0 +1,294 @@
+use serde_json::Value;
+use sqlx::PgPool;
+use time::OffsetDateTime;
+
+use crate::error::DurableError;
+use crate::tx::{self, PgTx};
+
+/// Ownership token returned by claim; required for projection and release.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct OutboxClaim {
+    pub message_id: i64,
+    pub worker_id: String,
+    pub generation: i64,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct OutboxMessage {
+    pub id: i64,
+    pub topic: String,
+    pub payload: Value,
+    pub created_at: OffsetDateTime,
+    pub available_at: OffsetDateTime,
+    pub attempt_count: i32,
+    pub claim: OutboxClaim,
+}
+
+/// Selects which outbox rows a claim may take.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum OutboxClaimFilter<'a> {
+    /// Any claimable topic.
+    Any,
+    /// Exact topic match.
+    Topic(&'a str),
+}
+
+pub async fn enqueue(tx: &mut PgTx<'_>, topic: &str, payload: Value) -> Result<i64, DurableError> {
+    let (id,): (i64,) = sqlx::query_as(
+        "INSERT INTO outbox_messages (topic, payload, available_at)
+         VALUES ($1, $2, now())
+         RETURNING id",
+    )
+    .bind(topic)
+    .bind(payload)
+    .fetch_one(tx.as_mut())
+    .await?;
+    Ok(id)
+}
+
+/// Enqueue work available after a delay measured by the database clock.
+pub async fn enqueue_after(
+    tx: &mut PgTx<'_>,
+    topic: &str,
+    payload: Value,
+    delay: time::Duration,
+) -> Result<i64, DurableError> {
+    let delay_secs = delay.as_seconds_f64().max(0.0);
+    let (id,): (i64,) = sqlx::query_as(
+        "INSERT INTO outbox_messages (topic, payload, available_at)
+         VALUES ($1, $2, now() + ($3 * interval '1 second'))
+         RETURNING id",
+    )
+    .bind(topic)
+    .bind(payload)
+    .bind(delay_secs)
+    .fetch_one(tx.as_mut())
+    .await?;
+    Ok(id)
+}
+
+/// Claim claimable outbox rows for exclusive projection using
+/// `FOR UPDATE SKIP LOCKED`. Lease eligibility and expiry use the database
+/// clock. Each successful claim advances `claim_generation` and returns an
+/// ownership token required for projection or release.
+pub async fn claim_batch(
+    pool: &PgPool,
+    worker_id: &str,
+    filter: OutboxClaimFilter<'_>,
+    limit: i64,
+    lease: time::Duration,
+) -> Result<Vec<OutboxMessage>, DurableError> {
+    let worker_id = worker_id.to_owned();
+    let topic = match filter {
+        OutboxClaimFilter::Any => None,
+        OutboxClaimFilter::Topic(topic) => Some(topic.to_owned()),
+    };
+    tx::with_transaction(pool, move |tx| {
+        Box::pin(async move {
+            let filter = match topic.as_deref() {
+                None => OutboxClaimFilter::Any,
+                Some(topic) => OutboxClaimFilter::Topic(topic),
+            };
+            claim_batch_in_tx(tx, &worker_id, filter, limit, lease).await
+        })
+    })
+    .await
+}
+
+pub async fn claim_batch_in_tx(
+    tx: &mut PgTx<'_>,
+    worker_id: &str,
+    filter: OutboxClaimFilter<'_>,
+    limit: i64,
+    lease: time::Duration,
+) -> Result<Vec<OutboxMessage>, DurableError> {
+    let lease_secs = lease.as_seconds_f64().max(0.0);
+    let topic = match filter {
+        OutboxClaimFilter::Any => None,
+        OutboxClaimFilter::Topic(topic) => Some(topic),
+    };
+    let rows = sqlx::query_as::<
+        _,
+        (
+            i64,
+            String,
+            Value,
+            OffsetDateTime,
+            OffsetDateTime,
+            i32,
+            i64,
+            String,
+        ),
+    >(
+        "WITH candidates AS (
+            SELECT id
+            FROM outbox_messages
+            WHERE projected_at IS NULL
+              AND available_at <= now()
+              AND (claimed_until IS NULL OR claimed_until < now())
+              AND ($4::text IS NULL OR topic = $4)
+            ORDER BY id
+            FOR UPDATE SKIP LOCKED
+            LIMIT $1
+         )
+         UPDATE outbox_messages AS messages
+         SET claimed_by = $2,
+             claimed_until = now() + ($3 * interval '1 second'),
+             claim_generation = messages.claim_generation + 1,
+             attempt_count = messages.attempt_count + 1
+         FROM candidates
+         WHERE messages.id = candidates.id
+         RETURNING
+            messages.id,
+            messages.topic,
+            messages.payload,
+            messages.created_at,
+            messages.available_at,
+            messages.attempt_count,
+            messages.claim_generation,
+            messages.claimed_by",
+    )
+    .bind(limit)
+    .bind(worker_id)
+    .bind(lease_secs)
+    .bind(topic)
+    .fetch_all(tx.as_mut())
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(
+            |(
+                id,
+                topic,
+                payload,
+                created_at,
+                available_at,
+                attempt_count,
+                claim_generation,
+                claimed_by,
+            )| {
+                OutboxMessage {
+                    id,
+                    topic,
+                    payload,
+                    created_at,
+                    available_at,
+                    attempt_count,
+                    claim: OutboxClaim {
+                        message_id: id,
+                        worker_id: claimed_by,
+                        generation: claim_generation,
+                    },
+                }
+            },
+        )
+        .collect())
+}
+
+/// Mark a claimed message projected. Idempotent when the row is already
+/// projected. Requires the live claim token when still pending.
+pub async fn mark_projected(tx: &mut PgTx<'_>, claim: &OutboxClaim) -> Result<(), DurableError> {
+    let result = sqlx::query(
+        "UPDATE outbox_messages
+         SET projected_at = now(),
+             claimed_by = NULL,
+             claimed_until = NULL,
+             last_error = NULL
+         WHERE id = $1
+           AND projected_at IS NULL
+           AND claimed_by = $2
+           AND claim_generation = $3",
+    )
+    .bind(claim.message_id)
+    .bind(&claim.worker_id)
+    .bind(claim.generation)
+    .execute(tx.as_mut())
+    .await?;
+    if result.rows_affected() == 1 {
+        return Ok(());
+    }
+
+    let projected: Option<(bool,)> = sqlx::query_as(
+        "SELECT projected_at IS NOT NULL
+         FROM outbox_messages
+         WHERE id = $1",
+    )
+    .bind(claim.message_id)
+    .fetch_optional(tx.as_mut())
+    .await?;
+
+    match projected {
+        Some((true,)) => Ok(()),
+        Some((false,)) => Err(DurableError::OutboxClaimMismatch {
+            message_id: claim.message_id,
+        }),
+        None => Err(DurableError::OutboxMissing {
+            message_id: claim.message_id,
+        }),
+    }
+}
+
+pub async fn mark_projected_pool(pool: &PgPool, claim: &OutboxClaim) -> Result<(), DurableError> {
+    let claim = claim.clone();
+    tx::with_transaction(pool, move |tx| {
+        Box::pin(async move { mark_projected(tx, &claim).await })
+    })
+    .await
+}
+
+/// Release a claimed message for later retry. Fenced to the claim token so a
+/// stale claimant cannot clear a reclaimed lease. Retry availability uses the
+/// database clock.
+pub async fn release_with_error(
+    tx: &mut PgTx<'_>,
+    claim: &OutboxClaim,
+    error: &str,
+    retry_delay: time::Duration,
+) -> Result<(), DurableError> {
+    let delay_secs = retry_delay.as_seconds_f64().max(0.0);
+    let result = sqlx::query(
+        "UPDATE outbox_messages
+         SET claimed_by = NULL,
+             claimed_until = NULL,
+             available_at = now() + ($4 * interval '1 second'),
+             last_error = $5
+         WHERE id = $1
+           AND projected_at IS NULL
+           AND claimed_by = $2
+           AND claim_generation = $3",
+    )
+    .bind(claim.message_id)
+    .bind(&claim.worker_id)
+    .bind(claim.generation)
+    .bind(delay_secs)
+    .bind(error)
+    .execute(tx.as_mut())
+    .await?;
+    if result.rows_affected() != 1 {
+        return Err(DurableError::OutboxClaimMismatch {
+            message_id: claim.message_id,
+        });
+    }
+    Ok(())
+}
+
+pub async fn unprojected_count(
+    pool: &PgPool,
+    filter: OutboxClaimFilter<'_>,
+) -> Result<i64, DurableError> {
+    let topic = match filter {
+        OutboxClaimFilter::Any => None,
+        OutboxClaimFilter::Topic(topic) => Some(topic),
+    };
+    let (count,): (i64,) = sqlx::query_as(
+        "SELECT COUNT(*)::bigint
+         FROM outbox_messages
+         WHERE projected_at IS NULL
+           AND ($1::text IS NULL OR topic = $1)",
+    )
+    .bind(topic)
+    .fetch_one(pool)
+    .await?;
+    Ok(count)
+}

crates/remora-durable/src/pool.rs

@@ -0,0 +1,41 @@
+use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
+use sqlx::{ConnectOptions, PgPool};
+use std::str::FromStr;
+use std::time::Duration;
+
+use crate::error::DurableError;
+
+#[derive(Clone, Debug)]
+pub struct PoolConfig {
+    pub max_connections: u32,
+    pub acquire_timeout: Duration,
+    pub idle_timeout: Option<Duration>,
+}
+
+impl Default for PoolConfig {
+    fn default() -> Self {
+        Self {
+            max_connections: 10,
+            acquire_timeout: Duration::from_secs(5),
+            idle_timeout: Some(Duration::from_secs(60)),
+        }
+    }
+}
+
+pub async fn connect(database_url: &str, config: PoolConfig) -> Result<PgPool, DurableError> {
+    let mut options = PgConnectOptions::from_str(database_url)
+        .map_err(|error| DurableError::Config(format!("invalid database URL: {error}")))?;
+    options = options
+        .application_name("remora-durable")
+        .statement_cache_capacity(100);
+    options = options.disable_statement_logging();
+
+    let mut builder = PgPoolOptions::new()
+        .max_connections(config.max_connections)
+        .acquire_timeout(config.acquire_timeout);
+    if let Some(idle_timeout) = config.idle_timeout {
+        builder = builder.idle_timeout(idle_timeout);
+    }
+
+    Ok(builder.connect_with(options).await?)
+}

crates/remora-durable/src/tx.rs

@@ -0,0 +1,42 @@
+use std::future::Future;
+use std::pin::Pin;
+
+use sqlx::{PgPool, Postgres, Transaction};
+
+use crate::error::DurableError;
+
+pub type PgTx<'a> = Transaction<'a, Postgres>;
+pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
+
+pub async fn with_transaction<T, F>(pool: &PgPool, work: F) -> Result<T, DurableError>
+where
+    F: for<'c> FnOnce(&'c mut PgTx<'_>) -> BoxFuture<'c, Result<T, DurableError>>,
+{
+    let mut tx = pool.begin().await?;
+    match work(&mut tx).await {
+        Ok(value) => {
+            tx.commit().await?;
+            Ok(value)
+        }
+        Err(error) => {
+            let _ = tx.rollback().await;
+            Err(error)
+        }
+    }
+}
+
+/// Bound how long this transaction waits for row or advisory locks.
+pub async fn set_lock_timeout(tx: &mut PgTx<'_>) -> Result<(), DurableError> {
+    sqlx::query("SET LOCAL lock_timeout = '5s'")
+        .execute(tx.as_mut())
+        .await?;
+    Ok(())
+}
+
+pub async fn advisory_xact_lock(tx: &mut PgTx<'_>, key: i64) -> Result<(), DurableError> {
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(key)
+        .execute(tx.as_mut())
+        .await?;
+    Ok(())
+}

crates/remora-durable/tests/substrate.rs

@@ -0,0 +1,976 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use remora_domain::ports::MutationScope;
+use remora_domain::{OpaqueId, fingerprint};
+use remora_durable::{
+    CursorCodec, DurableError, KeysetPosition, MutationDisposition, MutationStatus, OutboxClaim,
+    OutboxClaimFilter, PoolConfig, advisory_xact_lock, cas_set_text, claim_batch,
+    claim_batch_in_tx, connect, enqueue, enqueue_after, execute_mutation, first_page_cursor,
+    mark_projected, mark_projected_pool, migrate, next_page_cursor, release_with_error,
+    unprojected_count, with_transaction,
+};
+use serde_json::json;
+use sqlx::PgPool;
+use time::{Duration as TimeDuration, OffsetDateTime};
+use tokio::sync::Barrier;
+
+fn database_url() -> Option<String> {
+    std::env::var("DATABASE_URL")
+        .ok()
+        .filter(|value| !value.is_empty())
+}
+
+async fn test_pool() -> Option<PgPool> {
+    let Some(url) = database_url() else {
+        if std::env::var("REMORA_DURABLE_REQUIRE_DB").is_ok() {
+            panic!("DATABASE_URL is required when REMORA_DURABLE_REQUIRE_DB is set");
+        }
+        eprintln!("skipping durable substrate integration test: DATABASE_URL is unset");
+        return None;
+    };
+    let pool = connect(
+        &url,
+        PoolConfig {
+            max_connections: 8,
+            acquire_timeout: Duration::from_secs(10),
+            idle_timeout: Some(Duration::from_secs(30)),
+        },
+    )
+    .await
+    .expect("connect to postgres");
+    migrate(&pool).await.expect("run substrate migrations");
+    Some(pool)
+}
+
+fn subject(label: &str) -> OpaqueId {
+    OpaqueId::parse(format!("subject-{label}-aaaaaaaa")).expect("opaque id")
+}
+
+fn scope<'a>(subject: &'a OpaqueId, operation: &'a str, key: &'a str) -> MutationScope<'a> {
+    MutationScope {
+        subject,
+        operation,
+        key,
+    }
+}
+
+/// Example authority binding: attempt, fence, worker incarnation.
+fn eligibility(attempt: &str, fence: u64, worker: &str) -> Vec<u8> {
+    format!("attempt={attempt}|fence={fence}|worker={worker}").into_bytes()
+}
+
+#[tokio::test]
+async fn migration_round_trip_is_idempotent() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    migrate(&pool).await.expect("second migrate");
+    let (table_count,): (i64,) = sqlx::query_as(
+        "SELECT COUNT(*)::bigint
+         FROM information_schema.tables
+         WHERE table_schema = 'public'
+           AND table_name IN ('mutation_records', 'outbox_messages')",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("count substrate tables");
+    assert_eq!(table_count, 2);
+
+    let applied = remora_durable::migrator().iter().count();
+    assert_eq!(applied, 1);
+    let (version_count,): (i64,) = sqlx::query_as("SELECT COUNT(*)::bigint FROM _sqlx_migrations")
+        .fetch_one(&pool)
+        .await
+        .expect("count migrations");
+    assert!(version_count >= 1);
+}
+
+#[tokio::test]
+async fn transaction_rollback_discards_writes() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS substrate_tx_probe (
+            id TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+         )",
+    )
+    .execute(&pool)
+    .await
+    .expect("create probe table");
+    sqlx::query("DELETE FROM substrate_tx_probe WHERE id = 'rollback-row'")
+        .execute(&pool)
+        .await
+        .expect("clean probe row");
+
+    let result = with_transaction(&pool, |tx| {
+        Box::pin(async move {
+            sqlx::query("INSERT INTO substrate_tx_probe (id, value) VALUES ($1, $2)")
+                .bind("rollback-row")
+                .bind("visible-only-if-committed")
+                .execute(tx.as_mut())
+                .await?;
+            Err::<(), _>(DurableError::Config("force rollback".into()))
+        })
+    })
+    .await;
+    assert!(result.is_err());
+
+    let (count,): (i64,) =
+        sqlx::query_as("SELECT COUNT(*)::bigint FROM substrate_tx_probe WHERE id = 'rollback-row'")
+            .fetch_one(&pool)
+            .await
+            .expect("count probe row");
+    assert_eq!(count, 0);
+}
+
+#[tokio::test]
+async fn mutation_ledger_exact_replay_conflict_and_rejection() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    let owner = subject("replay");
+    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}");
+
+    let first = execute_mutation(
+        &pool,
+        scope(&owner, operation, key),
+        "submit",
+        &claim,
+        digest,
+        {
+            let probe = probe.clone();
+            move |tx| {
+                Box::pin(async move {
+                    apply_effect(tx, &probe).await?;
+                    Ok((MutationStatus::Accepted, json!({"job_id": "job-abcdefgh"})))
+                })
+            }
+        },
+    )
+    .await
+    .expect("first observation");
+    assert_eq!(first.disposition, MutationDisposition::First);
+    assert_eq!(first.status, MutationStatus::Accepted);
+    assert_eq!(first.response, json!({"job_id": "job-abcdefgh"}));
+
+    let replay = execute_mutation(
+        &pool,
+        scope(&owner, operation, key),
+        "submit",
+        &claim,
+        digest,
+        {
+            let probe = probe.clone();
+            move |tx| {
+                Box::pin(async move {
+                    apply_effect(tx, &probe).await?;
+                    Ok((
+                        MutationStatus::Accepted,
+                        json!({"job_id": "should-not-write"}),
+                    ))
+                })
+            }
+        },
+    )
+    .await
+    .expect("exact replay");
+    assert_eq!(replay.disposition, MutationDisposition::Replay);
+    assert_eq!(replay.status, MutationStatus::Accepted);
+    assert_eq!(replay.response, json!({"job_id": "job-abcdefgh"}));
+
+    let changed = fingerprint(&json!({"source": "http", "bucket": "other"})).expect("fp");
+    let conflict = execute_mutation(
+        &pool,
+        scope(&owner, operation, key),
+        "submit",
+        &claim,
+        changed,
+        {
+            let probe = probe.clone();
+            move |tx| {
+                Box::pin(async move {
+                    apply_effect(tx, &probe).await?;
+                    Ok((MutationStatus::Accepted, json!({"job_id": "conflict"})))
+                })
+            }
+        },
+    )
+    .await
+    .expect("conflict observation");
+    assert_eq!(conflict.disposition, MutationDisposition::Conflict);
+    assert_eq!(conflict.status, MutationStatus::Rejected);
+
+    let kind_conflict = execute_mutation(
+        &pool,
+        scope(&owner, operation, key),
+        "cancel",
+        &claim,
+        digest,
+        {
+            let probe = probe.clone();
+            move |tx| {
+                Box::pin(async move {
+                    apply_effect(tx, &probe).await?;
+                    Ok((MutationStatus::Accepted, json!({"ok": true})))
+                })
+            }
+        },
+    )
+    .await
+    .expect("kind conflict");
+    assert_eq!(kind_conflict.disposition, MutationDisposition::Conflict);
+
+    // Same kind and payload fingerprint with a different authority claim must
+    // not replay the stored result.
+    let stale_authority = eligibility("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");
+    let eligibility_conflict = execute_mutation(
+        &pool,
+        scope(&owner, operation, key),
+        "submit",
+        &stale_authority,
+        digest,
+        {
+            let probe = probe.clone();
+            move |tx| {
+                Box::pin(async move {
+                    apply_effect(tx, &probe).await?;
+                    Ok((MutationStatus::Accepted, json!({"job_id": "stale"})))
+                })
+            }
+        },
+    )
+    .await
+    .expect("eligibility conflict");
+    assert_eq!(
+        eligibility_conflict.disposition,
+        MutationDisposition::Conflict
+    );
+
+    let (hits,): (i64,) = sqlx::query_as("SELECT hits FROM substrate_effect_probe WHERE id = $1")
+        .bind(&probe)
+        .fetch_one(&pool)
+        .await
+        .expect("effect hits");
+    assert_eq!(hits, 1, "effect must apply at most once");
+
+    let reject_key = "op-reject-0001";
+    let reject_digest = fingerprint(&json!({"op": "reject"})).expect("fp");
+    let reject_claim = eligibility("attempt-cccccccc", 1, "worker-cccccccc");
+    let rejected = execute_mutation(
+        &pool,
+        scope(&owner, operation, reject_key),
+        "submit",
+        &reject_claim,
+        reject_digest,
+        |_tx| {
+            Box::pin(async move {
+                Ok((
+                    MutationStatus::Rejected,
+                    json!({"error": "precondition_failed"}),
+                ))
+            })
+        },
+    )
+    .await
+    .expect("rejected first observation");
+    assert_eq!(rejected.disposition, MutationDisposition::First);
+    assert_eq!(rejected.status, MutationStatus::Rejected);
+
+    let rejected_replay = execute_mutation(
+        &pool,
+        scope(&owner, operation, reject_key),
+        "submit",
+        &reject_claim,
+        reject_digest,
+        |_tx| {
+            Box::pin(async move { Ok((MutationStatus::Accepted, json!({"should": "not apply"}))) })
+        },
+    )
+    .await
+    .expect("rejected replay");
+    assert_eq!(rejected_replay.disposition, MutationDisposition::Replay);
+    assert_eq!(rejected_replay.status, MutationStatus::Rejected);
+    assert_eq!(
+        rejected_replay.response,
+        json!({"error": "precondition_failed"})
+    );
+}
+
+#[tokio::test]
+async fn concurrent_first_observation_serializes_to_one_effect() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    let owner = subject("race");
+    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}");
+    let workers = 8_usize;
+    let barrier = Arc::new(Barrier::new(workers));
+    let mut handles = Vec::with_capacity(workers);
+
+    for _ in 0..workers {
+        let pool = pool.clone();
+        let barrier = Arc::clone(&barrier);
+        let owner = owner.clone();
+        let probe = probe.clone();
+        let claim = claim.clone();
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            execute_mutation(
+                &pool,
+                scope(&owner, operation, key),
+                "submit",
+                &claim,
+                digest,
+                {
+                    let probe = probe.clone();
+                    move |tx| {
+                        Box::pin(async move {
+                            apply_effect(tx, &probe).await?;
+                            Ok((MutationStatus::Accepted, json!({"job_id": "job-race-01"})))
+                        })
+                    }
+                },
+            )
+            .await
+        }));
+    }
+
+    let mut first = 0_u32;
+    let mut replay = 0_u32;
+    for handle in handles {
+        let outcome = handle.await.expect("join").expect("mutation observation");
+        match outcome.disposition {
+            MutationDisposition::First => {
+                first += 1;
+                assert_eq!(outcome.status, MutationStatus::Accepted);
+                assert_eq!(outcome.response, json!({"job_id": "job-race-01"}));
+            }
+            MutationDisposition::Replay => {
+                replay += 1;
+                assert_eq!(outcome.status, MutationStatus::Accepted);
+                assert_eq!(outcome.response, json!({"job_id": "job-race-01"}));
+            }
+            MutationDisposition::Conflict => {
+                panic!("exact concurrent request must not conflict")
+            }
+        }
+    }
+    assert_eq!(first, 1);
+    assert_eq!(replay, u32::try_from(workers - 1).expect("workers fit u32"));
+
+    let (hits,): (i64,) = sqlx::query_as("SELECT hits FROM substrate_effect_probe WHERE id = $1")
+        .bind(&probe)
+        .fetch_one(&pool)
+        .await
+        .expect("effect hits");
+    assert_eq!(hits, 1, "effect must apply at most once under contention");
+}
+
+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",
+    )
+    .bind(probe_id)
+    .execute(tx.as_mut())
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn concurrent_cas_admits_exactly_one_writer() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS substrate_cas_rows (
+            id TEXT PRIMARY KEY,
+            value TEXT NOT NULL,
+            revision BIGINT NOT NULL
+         )",
+    )
+    .execute(&pool)
+    .await
+    .expect("create cas table");
+    sqlx::query(
+        "INSERT INTO substrate_cas_rows (id, value, revision) VALUES ($1, $2, 0)
+         ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value, revision = 0",
+    )
+    .bind("cas-row")
+    .bind("initial")
+    .execute(&pool)
+    .await
+    .expect("seed cas row");
+
+    let barrier = Arc::new(Barrier::new(8));
+    let mut handles = Vec::new();
+    for index in 0..8 {
+        let pool = pool.clone();
+        let barrier = Arc::clone(&barrier);
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            with_transaction(&pool, |tx| {
+                Box::pin(async move {
+                    cas_set_text(
+                        tx,
+                        "substrate_cas_rows",
+                        "id",
+                        "cas-row",
+                        "value",
+                        &format!("writer-{index}"),
+                        0,
+                    )
+                    .await
+                })
+            })
+            .await
+        }));
+    }
+
+    let mut successes = 0_u32;
+    let mut conflicts = 0_u32;
+    for handle in handles {
+        match handle.await.expect("join") {
+            Ok(revision) => {
+                assert_eq!(revision, 1);
+                successes += 1;
+            }
+            Err(DurableError::CasConflict) => conflicts += 1,
+            Err(error) => panic!("unexpected CAS error: {error}"),
+        }
+    }
+    assert_eq!(successes, 1);
+    assert_eq!(conflicts, 7);
+
+    let (value, revision): (String, i64) =
+        sqlx::query_as("SELECT value, revision FROM substrate_cas_rows WHERE id = 'cas-row'")
+            .fetch_one(&pool)
+            .await
+            .expect("read cas row");
+    assert_eq!(revision, 1);
+    assert!(value.starts_with("writer-"));
+}
+
+/// Namespace topics so concurrent cargo-nextest cases never claim each other.
+fn outbox_topic(suite: &str, name: &str) -> String {
+    format!("substrate.outbox.{suite}.{name}")
+}
+
+async fn clear_outbox_suite(pool: &PgPool, suite: &str) {
+    let pattern = format!("substrate.outbox.{suite}.%");
+    sqlx::query("DELETE FROM outbox_messages WHERE topic LIKE $1")
+        .bind(pattern)
+        .execute(pool)
+        .await
+        .expect("clear suite outbox rows");
+}
+
+#[tokio::test]
+async fn outbox_claim_projection_recovery_and_claim_fencing() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    const SUITE: &str = "fencing";
+    clear_outbox_suite(&pool, SUITE).await;
+    let batch_topic = outbox_topic(SUITE, "batch");
+    let recover_topic = outbox_topic(SUITE, "recover");
+    let retry_topic = outbox_topic(SUITE, "retry");
+    let batch_filter = OutboxClaimFilter::Topic(&batch_topic);
+    let recover_filter = OutboxClaimFilter::Topic(&recover_topic);
+    let retry_filter = OutboxClaimFilter::Topic(&retry_topic);
+
+    with_transaction(&pool, {
+        let batch_topic = batch_topic.clone();
+        move |tx| {
+            Box::pin(async move {
+                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-1"})).await?;
+                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-2"})).await?;
+                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-3"})).await?;
+                Ok(())
+            })
+        }
+    })
+    .await
+    .expect("enqueue");
+
+    let claimed = claim_batch(
+        &pool,
+        "worker-a",
+        batch_filter,
+        2,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("claim batch");
+    assert_eq!(claimed.len(), 2);
+    assert_eq!(claimed[0].attempt_count, 1);
+    assert_eq!(claimed[0].claim.worker_id, "worker-a");
+    assert_eq!(claimed[0].claim.generation, 1);
+
+    let overlapping = claim_batch(
+        &pool,
+        "worker-b",
+        batch_filter,
+        10,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("second claim");
+    assert_eq!(
+        overlapping.len(),
+        1,
+        "lease exclusion leaves only the unclaimed remainder"
+    );
+
+    // Unclaimed or foreign claims cannot complete projection.
+    let foreign = OutboxClaim {
+        message_id: claimed[0].id,
+        worker_id: "worker-intruder".into(),
+        generation: claimed[0].claim.generation,
+    };
+    assert!(matches!(
+        mark_projected_pool(&pool, &foreign).await,
+        Err(DurableError::OutboxClaimMismatch { .. })
+    ));
+    let unclaimed = OutboxClaim {
+        message_id: claimed[0].id,
+        worker_id: "worker-a".into(),
+        generation: 0,
+    };
+    assert!(matches!(
+        mark_projected_pool(&pool, &unclaimed).await,
+        Err(DurableError::OutboxClaimMismatch { .. })
+    ));
+
+    for message in &claimed {
+        mark_projected_pool(&pool, &message.claim)
+            .await
+            .expect("mark projected");
+        // Retry after success is idempotent for ambiguous-commit recovery.
+        mark_projected_pool(&pool, &message.claim)
+            .await
+            .expect("idempotent mark projected");
+    }
+    mark_projected_pool(&pool, &overlapping[0].claim)
+        .await
+        .expect("mark remaining");
+
+    assert_eq!(
+        unprojected_count(&pool, batch_filter).await.expect("count"),
+        0
+    );
+
+    let recovered_id = with_transaction(&pool, {
+        let recover_topic = recover_topic.clone();
+        move |tx| {
+            Box::pin(
+                async move { enqueue(tx, &recover_topic, json!({"job_id": "job-recover"})).await },
+            )
+        }
+    })
+    .await
+    .expect("enqueue recovery");
+
+    let first_claim = claim_batch(
+        &pool,
+        "worker-crash",
+        recover_filter,
+        1,
+        TimeDuration::milliseconds(1),
+    )
+    .await
+    .expect("claim before crash");
+    assert_eq!(first_claim.len(), 1);
+    assert_eq!(first_claim[0].id, recovered_id);
+    let stale = first_claim[0].claim.clone();
+    tokio::time::sleep(Duration::from_millis(20)).await;
+
+    let reclaimed = claim_batch(
+        &pool,
+        "worker-recover",
+        recover_filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("reclaim after lease expiry");
+    assert_eq!(reclaimed.len(), 1);
+    assert_eq!(reclaimed[0].id, recovered_id);
+    assert_eq!(reclaimed[0].attempt_count, 2);
+    assert_eq!(reclaimed[0].claim.generation, 2);
+    assert_eq!(reclaimed[0].claim.worker_id, "worker-recover");
+
+    // Stale claimant must not project or release after reclaim.
+    assert!(matches!(
+        mark_projected_pool(&pool, &stale).await,
+        Err(DurableError::OutboxClaimMismatch { .. })
+    ));
+    with_transaction(&pool, |tx| {
+        let stale = stale.clone();
+        Box::pin(
+            async move { release_with_error(tx, &stale, "stale", TimeDuration::seconds(1)).await },
+        )
+    })
+    .await
+    .expect_err("stale release must fail");
+
+    mark_projected_pool(&pool, &reclaimed[0].claim)
+        .await
+        .expect("project recovered");
+
+    let retry_id = with_transaction(&pool, {
+        let retry_topic = retry_topic.clone();
+        move |tx| {
+            Box::pin(async move {
+                enqueue_after(
+                    tx,
+                    &retry_topic,
+                    json!({"job_id": "job-retry-error"}),
+                    TimeDuration::ZERO,
+                )
+                .await
+            })
+        }
+    })
+    .await
+    .expect("enqueue retry");
+
+    let failed_claim = claim_batch(
+        &pool,
+        "worker-fail",
+        retry_filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("claim for release_with_error");
+    assert_eq!(failed_claim.len(), 1);
+    assert_eq!(failed_claim[0].id, retry_id);
+
+    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("release with error");
+
+    let early = claim_batch(
+        &pool,
+        "worker-early",
+        retry_filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("early claim");
+    assert!(
+        early.is_empty(),
+        "release_with_error must delay availability on the database clock"
+    );
+
+    // Advance availability using the database clock relative update.
+    sqlx::query("UPDATE outbox_messages SET available_at = now() WHERE id = $1")
+        .bind(retry_id)
+        .execute(&pool)
+        .await
+        .expect("advance available_at for recovery");
+
+    let after_error = claim_batch(
+        &pool,
+        "worker-retry",
+        retry_filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("reclaim after release_with_error");
+    assert_eq!(after_error.len(), 1);
+    assert_eq!(after_error[0].id, retry_id);
+    assert_eq!(after_error[0].attempt_count, 2);
+
+    let (last_error,): (Option<String>,) =
+        sqlx::query_as("SELECT last_error FROM outbox_messages WHERE id = $1")
+            .bind(retry_id)
+            .fetch_one(&pool)
+            .await
+            .expect("read last_error");
+    assert_eq!(last_error.as_deref(), Some("projection failed"));
+
+    let concurrent = claim_batch(
+        &pool,
+        "worker-other",
+        retry_filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("concurrent claim while leased");
+    assert!(concurrent.is_empty());
+
+    mark_projected_pool(&pool, &after_error[0].claim)
+        .await
+        .expect("project after release recovery");
+    assert_eq!(
+        unprojected_count(&pool, retry_filter).await.expect("count"),
+        0
+    );
+}
+
+#[tokio::test]
+async fn outbox_skip_locked_excludes_concurrent_open_claim() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    const SUITE: &str = "skip";
+    clear_outbox_suite(&pool, SUITE).await;
+    let topic = outbox_topic(SUITE, "dispatch");
+    let filter = OutboxClaimFilter::Topic(&topic);
+
+    with_transaction(&pool, {
+        let topic = topic.clone();
+        move |tx| {
+            Box::pin(async move {
+                enqueue(tx, &topic, json!({"job_id": "job-skip-locked"})).await?;
+                Ok(())
+            })
+        }
+    })
+    .await
+    .expect("enqueue");
+
+    // Hold the claim open in an uncommitted transaction so another connection
+    // must observe SKIP LOCKED rather than only claimed_until exclusion.
+    let mut holder = pool.begin().await.expect("holder begin");
+    let held = claim_batch_in_tx(
+        &mut holder,
+        "worker-hold",
+        filter,
+        1,
+        TimeDuration::seconds(30),
+    )
+    .await
+    .expect("hold claim");
+    assert_eq!(held.len(), 1);
+
+    let concurrent = claim_batch(&pool, "worker-other", filter, 1, TimeDuration::seconds(30))
+        .await
+        .expect("concurrent claim");
+    assert!(
+        concurrent.is_empty(),
+        "SKIP LOCKED must hide a row locked by another transaction"
+    );
+
+    mark_projected(&mut holder, &held[0].claim)
+        .await
+        .expect("project under holder");
+    holder.commit().await.expect("holder commit");
+    assert_eq!(unprojected_count(&pool, filter).await.expect("count"), 0);
+}
+
+#[tokio::test]
+async fn stable_keyset_pages_respect_high_water_and_cursor_binding() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS substrate_keyset_jobs (
+            id TEXT PRIMARY KEY,
+            subject TEXT NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL
+         )",
+    )
+    .execute(&pool)
+    .await
+    .expect("create keyset table");
+    sqlx::query("DELETE FROM substrate_keyset_jobs")
+        .execute(&pool)
+        .await
+        .expect("clean keyset table");
+
+    let owner = "subject-keyset-aaaaaaaa";
+    let base = OffsetDateTime::now_utc() - TimeDuration::seconds(100);
+    for index in 0..10 {
+        sqlx::query(
+            "INSERT INTO substrate_keyset_jobs (id, subject, created_at)
+             VALUES ($1, $2, $3)",
+        )
+        .bind(format!("job-keyset-{index:04}"))
+        .bind(owner)
+        .bind(base + TimeDuration::seconds(i64::from(index)))
+        .execute(&pool)
+        .await
+        .expect("insert job");
+    }
+
+    let codec = CursorCodec::new(b"integration-cursor-secret").expect("codec");
+    let filters = json!({"lifecycle": "visible"});
+    let filter_fp = CursorCodec::filter_fingerprint(&filters).expect("filter fp");
+
+    let first_page: Vec<(String, OffsetDateTime)> = sqlx::query_as(
+        "SELECT id, created_at
+         FROM substrate_keyset_jobs
+         WHERE subject = $1
+         ORDER BY created_at DESC, id DESC
+         LIMIT 3",
+    )
+    .bind(owner)
+    .fetch_all(&pool)
+    .await
+    .expect("first page");
+    assert_eq!(first_page.len(), 3);
+    assert_eq!(first_page[0].0, "job-keyset-0009");
+
+    let cursor = first_page_cursor(
+        owner,
+        filter_fp,
+        KeysetPosition {
+            created_at: first_page[0].1,
+            id: first_page[0].0.clone(),
+        },
+        KeysetPosition {
+            created_at: first_page[2].1,
+            id: first_page[2].0.clone(),
+        },
+    );
+    let token = codec.encode(&cursor).expect("encode");
+
+    sqlx::query(
+        "INSERT INTO substrate_keyset_jobs (id, subject, created_at)
+         VALUES ($1, $2, $3)",
+    )
+    .bind("job-keyset-9999")
+    .bind(owner)
+    .bind(OffsetDateTime::now_utc())
+    .execute(&pool)
+    .await
+    .expect("insert newer job");
+
+    let bound = codec.bind(&token, owner, filter_fp).expect("bind cursor");
+    let after = bound.after.clone().expect("after position");
+    let second_page: Vec<(String, OffsetDateTime)> = sqlx::query_as(
+        "SELECT id, created_at
+         FROM substrate_keyset_jobs
+         WHERE subject = $1
+           AND (created_at, id) <= ($2, $3)
+           AND (created_at, id) < ($4, $5)
+         ORDER BY created_at DESC, id DESC
+         LIMIT 3",
+    )
+    .bind(owner)
+    .bind(bound.high_water.created_at)
+    .bind(&bound.high_water.id)
+    .bind(after.created_at)
+    .bind(&after.id)
+    .fetch_all(&pool)
+    .await
+    .expect("second page");
+
+    assert_eq!(
+        second_page
+            .iter()
+            .map(|row| row.0.as_str())
+            .collect::<Vec<_>>(),
+        vec!["job-keyset-0006", "job-keyset-0005", "job-keyset-0004"]
+    );
+    assert!(
+        second_page.iter().all(|row| row.0 != "job-keyset-9999"),
+        "jobs newer than the high-water mark must not enter a stable traversal"
+    );
+
+    let advanced = next_page_cursor(
+        &bound,
+        KeysetPosition {
+            created_at: second_page[2].1,
+            id: second_page[2].0.clone(),
+        },
+    );
+    let advanced_token = codec.encode(&advanced).expect("encode advanced");
+    assert!(matches!(
+        codec.bind(&advanced_token, "other-subject-bbbbbbbb", filter_fp),
+        Err(DurableError::CursorBindingMismatch)
+    ));
+}
+
+#[tokio::test]
+async fn advisory_lock_serializes_critical_section() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS substrate_lock_probe (
+            id TEXT PRIMARY KEY,
+            value BIGINT NOT NULL
+         )",
+    )
+    .execute(&pool)
+    .await
+    .expect("create lock probe");
+    sqlx::query(
+        "INSERT INTO substrate_lock_probe (id, value) VALUES ('lock-row', 0)
+         ON CONFLICT (id) DO UPDATE SET value = 0",
+    )
+    .execute(&pool)
+    .await
+    .expect("seed lock probe");
+
+    let barrier = Arc::new(Barrier::new(4));
+    let mut handles = Vec::new();
+    for _ in 0..4 {
+        let pool = pool.clone();
+        let barrier = Arc::clone(&barrier);
+        handles.push(tokio::spawn(async move {
+            barrier.wait().await;
+            with_transaction(&pool, |tx| {
+                Box::pin(async move {
+                    advisory_xact_lock(tx, 42).await?;
+                    let (value,): (i64,) = sqlx::query_as(
+                        "SELECT value FROM substrate_lock_probe WHERE id = 'lock-row'",
+                    )
+                    .fetch_one(tx.as_mut())
+                    .await?;
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                    sqlx::query("UPDATE substrate_lock_probe SET value = $1 WHERE id = 'lock-row'")
+                        .bind(value + 1)
+                        .execute(tx.as_mut())
+                        .await?;
+                    Ok::<_, DurableError>(())
+                })
+            })
+            .await
+        }));
+    }
+    for handle in handles {
+        handle.await.expect("join").expect("lock section");
+    }
+    let (value,): (i64,) =
+        sqlx::query_as("SELECT value FROM substrate_lock_probe WHERE id = 'lock-row'")
+            .fetch_one(&pool)
+            .await
+            .expect("read lock probe");
+    assert_eq!(value, 4);
+}