Add crash-safe attempt authority lifecycle

Cargo.lock

@@ -2881,6 +2881,26 @@ dependencies = [
 name = "remora-api"
 version = "0.0.0"
 
+[[package]]
+name = "remora-attempt"
+version = "0.0.0"
+dependencies = [
+ "getrandom 0.3.4",
+ "remora-admission",
+ "remora-contracts",
+ "remora-coordination",
+ "remora-credentials",
+ "remora-domain",
+ "remora-durable",
+ "remora-identity",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "thiserror",
+ "time",
+ "tokio",
+]
+
 [[package]]
 name = "remora-contracts"
 version = "0.0.0"

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
     "crates/remora-admission",
     "crates/remora-api",
+    "crates/remora-attempt",
     "crates/remora-contracts",
     "crates/remora-coordination",
     "crates/remora-credentials",

Dockerfile.test

@@ -214,6 +214,35 @@ COPY crates crates
 COPY --from=kind-admission-build /tmp/remora-admission-tests.tar.zst /tests/remora-admission-tests.tar.zst
 CMD ["bash", "-ec", "test -s /run/remora-fixture/profile.env && test -n \"$DATABASE_URL\" && export REMORA_ADMISSION_REQUIRE=1 && cargo-nextest nextest run --archive-file /tests/remora-admission-tests.tar.zst --workspace-remap /workspace --test-threads=1"]
 
+FROM rust-tools AS kind-attempt-build
+WORKDIR /workspace
+COPY Cargo.toml Cargo.lock rust-toolchain.toml ./
+COPY .config .config
+COPY contracts contracts
+COPY crates crates
+COPY docs/api docs/api
+RUN cargo nextest archive \
+    --package remora-attempt \
+    --build-jobs 4 \
+    --locked \
+    --archive-file /tmp/remora-attempt-tests.tar.zst
+
+FROM kind-attempt-build AS kind-attempt-verified
+RUN cargo clippy \
+    --package remora-attempt \
+    --all-targets \
+    --all-features \
+    --locked \
+    -- -D warnings \
+    && cargo test --package remora-attempt --doc --locked
+
+FROM rust-tools AS kind-attempt-runner
+WORKDIR /workspace
+COPY Cargo.toml ./
+COPY crates crates
+COPY --from=kind-attempt-verified /tmp/remora-attempt-tests.tar.zst /tests/remora-attempt-tests.tar.zst
+CMD ["bash", "-ec", "test -s /run/remora-fixture/profile.env && test -n \"$DATABASE_URL\" && test -n \"$REMORA_COORDINATION_URL\" && export REMORA_ATTEMPT_REQUIRE=1 && cargo-nextest nextest run --archive-file /tests/remora-attempt-tests.tar.zst --workspace-remap /workspace --test-threads=1"]
+
 FROM rust-tools AS kind-http-source-build
 WORKDIR /workspace
 COPY Cargo.toml Cargo.lock rust-toolchain.toml ./

Makefile

@@ -1,4 +1,4 @@
-.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-admission check-stream check-target check-result check-outbound check-http-source check-object-source check-hf-source check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
+.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-admission check-attempt check-stream check-target check-result check-outbound check-http-source check-object-source check-hf-source check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
 
 CONTAINER_ENGINE ?= nerdctl
 CONTAINER_RUN_FLAGS ?= --memory=16g --memory-swap=24g --cpus=4
@@ -71,6 +71,9 @@ check-credentials:
 check-admission:
 	./tests/kind/run-admission-job.sh
 
+check-attempt:
+	./tests/kind/run-attempt-job.sh
+
 check-stream:
 	cargo fmt --all --check
 	cargo clippy --package remora-stream --all-targets --all-features -- -D warnings

crates/remora-attempt/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "remora-attempt"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+getrandom = "0.3.4"
+remora-coordination = { path = "../remora-coordination" }
+remora-domain.workspace = true
+remora-durable.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+sqlx.workspace = true
+thiserror.workspace = true
+time.workspace = true
+
+[dev-dependencies]
+remora-admission = { path = "../remora-admission" }
+remora-contracts = { path = "../remora-contracts" }
+remora-credentials = { path = "../remora-credentials" }
+remora-identity = { path = "../remora-identity" }
+tokio.workspace = true

crates/remora-attempt/migrations/0001_attempt_authority.sql

@@ -0,0 +1,38 @@
+ALTER TABLE jobs
+    ADD COLUMN current_fence BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN authority_attempt_id TEXT,
+    ADD CONSTRAINT jobs_current_fence_check CHECK (
+        current_fence >= 0 AND current_fence <= 9007199254740991
+    );
+
+CREATE TABLE attempts (
+    attempt_id TEXT PRIMARY KEY,
+    job_id TEXT NOT NULL REFERENCES jobs (job_id),
+    fence BIGINT NOT NULL CHECK (fence BETWEEN 1 AND 9007199254740991),
+    worker_incarnation TEXT NOT NULL,
+    state TEXT NOT NULL CHECK (state IN ('authorized', 'executing', 'reaped')),
+    initial_progress BIGINT NOT NULL DEFAULT 0 CHECK (initial_progress >= 0),
+    authorized_at TIMESTAMPTZ NOT NULL,
+    executing_at TIMESTAMPTZ,
+    reaped_at TIMESTAMPTZ,
+    CONSTRAINT attempts_state_timestamps_check CHECK (
+        (state = 'authorized' AND executing_at IS NULL AND reaped_at IS NULL)
+        OR (state = 'executing' AND executing_at IS NOT NULL AND reaped_at IS NULL)
+        OR (state = 'reaped' AND executing_at IS NULL AND reaped_at IS NOT NULL)
+    ),
+    CONSTRAINT attempts_job_fence_unique UNIQUE (job_id, fence),
+    CONSTRAINT attempts_job_authority_unique UNIQUE (attempt_id, job_id, fence),
+    CONSTRAINT attempts_authority_tuple_unique UNIQUE (
+        attempt_id, job_id, fence, worker_incarnation
+    )
+);
+
+ALTER TABLE jobs
+    ADD CONSTRAINT jobs_authority_attempt_fk
+    FOREIGN KEY (authority_attempt_id, job_id, current_fence)
+    REFERENCES attempts (attempt_id, job_id, fence)
+    DEFERRABLE INITIALLY DEFERRED;
+
+CREATE INDEX attempts_job_authorized_idx
+    ON attempts (job_id, authorized_at)
+    WHERE state = 'authorized';

crates/remora-attempt/src/error.rs

@@ -0,0 +1,14 @@
+use remora_coordination::CoordinationError;
+use remora_durable::DurableError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AttemptError {
+    #[error("invalid attempt request: {0}")]
+    Invalid(String),
+    #[error("attempt operation identity was reused with different inputs")]
+    OperationConflict,
+    #[error("coordination operation failed: {0}")]
+    Coordination(#[from] CoordinationError),
+    #[error("durable operation failed: {0}")]
+    Durable(#[from] DurableError),
+}

crates/remora-attempt/src/lib.rs

@@ -0,0 +1,12 @@
+mod error;
+mod migrate;
+mod service;
+mod types;
+
+pub use error::AttemptError;
+pub use migrate::migrate;
+pub use service::{AttemptConfig, AttemptService};
+pub use types::{
+    ActivationOutcome, AttemptAuthority, AttemptFaultInjector, AttemptFaultPoint, AttemptState,
+    ClaimOutcome, ClaimRejection, RecoveryOutcome, RenewalOutcome,
+};

crates/remora-attempt/src/migrate.rs

@@ -0,0 +1,12 @@
+use sqlx::PgPool;
+
+use crate::AttemptError;
+
+pub async fn migrate(pool: &PgPool) -> Result<(), AttemptError> {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name("_remora_attempt_migrations");
+    migrator
+        .run(pool)
+        .await
+        .map_err(|error| AttemptError::Durable(remora_durable::DurableError::Migrate(error)))
+}

crates/remora-attempt/src/service.rs

@@ -0,0 +1,722 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use remora_coordination::{
+    ActivateOptions, CoordinationStore, LeaseOwner, LocalPermitGate, MAX_SAFE_INTEGER, OpCode,
+};
+use remora_domain::{OpaqueId, fingerprint, ports::MutationScope};
+use remora_durable::{
+    DurableError, MutationDisposition, MutationStatus, enqueue, execute_mutation, with_transaction,
+};
+use serde::Serialize;
+use serde_json::json;
+use sqlx::PgPool;
+use time::OffsetDateTime;
+
+use crate::{
+    ActivationOutcome, AttemptAuthority, AttemptError, AttemptFaultInjector, AttemptFaultPoint,
+    AttemptState, ClaimOutcome, ClaimRejection, RecoveryOutcome, RenewalOutcome,
+};
+
+struct NoFaults;
+
+impl AttemptFaultInjector for NoFaults {
+    fn hit(&self, _point: AttemptFaultPoint) {}
+}
+
+#[derive(Clone, Debug)]
+pub struct AttemptConfig {
+    pub max_attempts: u64,
+    pub local_limit: u64,
+}
+
+impl Default for AttemptConfig {
+    fn default() -> Self {
+        Self {
+            max_attempts: 3,
+            local_limit: 4,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct AttemptService {
+    pool: PgPool,
+    coordination: CoordinationStore,
+    local_permits: LocalPermitGate,
+    operation_subject: OpaqueId,
+    faults: Arc<dyn AttemptFaultInjector>,
+    config: AttemptConfig,
+}
+
+impl AttemptService {
+    pub fn new(
+        pool: PgPool,
+        coordination: CoordinationStore,
+        config: AttemptConfig,
+    ) -> Result<Self, AttemptError> {
+        Self::new_with_fault_injector(pool, coordination, config, Arc::new(NoFaults))
+    }
+
+    pub fn new_with_fault_injector(
+        pool: PgPool,
+        coordination: CoordinationStore,
+        config: AttemptConfig,
+        faults: Arc<dyn AttemptFaultInjector>,
+    ) -> Result<Self, AttemptError> {
+        if config.max_attempts == 0 || config.max_attempts > MAX_SAFE_INTEGER {
+            return Err(AttemptError::Invalid(
+                "max_attempts must fit the coordination safe-integer range".into(),
+            ));
+        }
+        if config.local_limit == 0 {
+            return Err(AttemptError::Invalid(
+                "local_limit must be greater than zero".into(),
+            ));
+        }
+        Ok(Self {
+            pool,
+            coordination,
+            local_permits: LocalPermitGate::new(config.local_limit),
+            operation_subject: OpaqueId::parse("remora_attempt_service")
+                .map_err(|error| AttemptError::Invalid(error.to_string()))?,
+            faults,
+            config,
+        })
+    }
+
+    pub async fn claim(
+        &self,
+        job_id: &OpaqueId,
+        worker_incarnation: &OpaqueId,
+        operation_id: &OpaqueId,
+    ) -> Result<ClaimOutcome, AttemptError> {
+        if self.local_permits.acquire(job_id.as_str()).is_err() {
+            return Ok(ClaimOutcome::Rejected {
+                reason: ClaimRejection::LocalCapacity,
+                disposition: None,
+            });
+        }
+        let request = ClaimRequest {
+            job_id,
+            worker_incarnation,
+        };
+        let request_fingerprint =
+            fingerprint(&request).map_err(|error| AttemptError::Invalid(error.to_string()))?;
+        let attempt_id = new_id('a')?;
+        let job = job_id.clone();
+        let worker = worker_incarnation.clone();
+        let max_attempts = self.config.max_attempts;
+        let coordination = self.coordination.clone();
+        let faults = self.faults.clone();
+        let reserved = Arc::new(AtomicBool::new(false));
+        let effect_reserved = reserved.clone();
+        let outcome = execute_mutation(
+            &self.pool,
+            MutationScope {
+                subject: &self.operation_subject,
+                operation: "attempt.mutation",
+                key: operation_id.as_str(),
+            },
+            "claim_attempt",
+            &[],
+            request_fingerprint,
+            move |tx| {
+                Box::pin(async move {
+                    let decision = authorize_attempt(
+                        tx,
+                        AuthorizeInput {
+                            coordination: &coordination,
+                            job_id: &job,
+                            attempt_id: &attempt_id,
+                            worker: &worker,
+                            max_attempts,
+                            faults: faults.as_ref(),
+                            reserved: effect_reserved.as_ref(),
+                        },
+                    )
+                    .await?;
+                    let status = if matches!(decision, ClaimOutcome::Claimed { .. }) {
+                        MutationStatus::Accepted
+                    } else {
+                        MutationStatus::Rejected
+                    };
+                    let response = serde_json::to_value(decision)
+                        .map_err(|error| DurableError::Effect(error.to_string()))?;
+                    Ok((status, response))
+                })
+            },
+        )
+        .await;
+        let outcome = match outcome {
+            Ok(outcome) => outcome,
+            Err(error) => {
+                if reserved.load(Ordering::Acquire) {
+                    self.compensate_reservation(job_id, worker_incarnation)
+                        .await;
+                } else {
+                    self.local_permits.release(job_id.as_str());
+                }
+                return Err(error.into());
+            }
+        };
+        self.faults.hit(AttemptFaultPoint::ClaimAfterDurable);
+        if outcome.disposition == MutationDisposition::Conflict {
+            self.local_permits.release(job_id.as_str());
+            return Err(AttemptError::OperationConflict);
+        }
+        let decision: ClaimOutcome = serde_json::from_value(outcome.response)
+            .map_err(|error| AttemptError::Invalid(error.to_string()))?;
+        if matches!(decision, ClaimOutcome::Rejected { .. }) {
+            self.local_permits.release(job_id.as_str());
+        }
+        Ok(decision.with_disposition(outcome.disposition))
+    }
+
+    pub async fn activate(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+    ) -> Result<ActivationOutcome, AttemptError> {
+        validate_authority(authority)?;
+        let request_fingerprint = fingerprint(&ActivationRequest {
+            job_id: &authority.job_id,
+            attempt_id: &authority.attempt_id,
+            fence: authority.fence,
+            worker_incarnation: &authority.worker_incarnation,
+        })
+        .map_err(|error| AttemptError::Invalid(error.to_string()))?;
+        let eligibility = authority_binding(authority);
+        let authority_for_effect = authority.clone();
+        let coordination = self.coordination.clone();
+        let faults = self.faults.clone();
+        let outcome = execute_mutation(
+            &self.pool,
+            MutationScope {
+                subject: &self.operation_subject,
+                operation: "attempt.mutation",
+                key: operation_id.as_str(),
+            },
+            "activate_attempt",
+            &eligibility,
+            request_fingerprint,
+            move |tx| {
+                Box::pin(async move {
+                    let accepted =
+                        activate_attempt(tx, &coordination, &authority_for_effect, faults.as_ref())
+                            .await?;
+                    Ok((
+                        if accepted {
+                            MutationStatus::Accepted
+                        } else {
+                            MutationStatus::Rejected
+                        },
+                        json!({ "activated": accepted }),
+                    ))
+                })
+            },
+        )
+        .await;
+        let outcome = match outcome {
+            Ok(outcome) => outcome,
+            Err(error) => {
+                if self
+                    .exact_authority_is_executing_and_live(authority)
+                    .await?
+                {
+                    return Ok(ActivationOutcome::Activated(MutationDisposition::First));
+                }
+                self.release_exact(authority).await;
+                return Err(error.into());
+            }
+        };
+        self.faults.hit(AttemptFaultPoint::ActivationAfterDurable);
+        if outcome.disposition == MutationDisposition::Conflict {
+            return Err(AttemptError::OperationConflict);
+        }
+        if outcome.status == MutationStatus::Rejected {
+            self.release_exact(authority).await;
+            return Ok(ActivationOutcome::PreconditionFailed);
+        }
+        if !self
+            .exact_authority_is_executing_and_live(authority)
+            .await?
+        {
+            self.release_exact(authority).await;
+            return Ok(ActivationOutcome::PreconditionFailed);
+        }
+        Ok(ActivationOutcome::Activated(outcome.disposition))
+    }
+
+    pub async fn renew(
+        &self,
+        authority: &AttemptAuthority,
+    ) -> Result<RenewalOutcome, AttemptError> {
+        validate_authority(authority)?;
+        if !self.exact_authority_is_executing(authority).await? {
+            return Ok(RenewalOutcome::AuthorityLost);
+        }
+        self.faults.hit(AttemptFaultPoint::RenewalBeforeLease);
+        let result = self
+            .coordination
+            .renew(authority.job_id.as_str(), &lease_owner(authority)?)
+            .await?;
+        self.faults.hit(AttemptFaultPoint::RenewalAfterLease);
+        if !result.ok() || !self.exact_authority_is_executing(authority).await? {
+            return Ok(RenewalOutcome::AuthorityLost);
+        }
+        Ok(RenewalOutcome::Renewed)
+    }
+
+    pub async fn recover_inactive(
+        &self,
+        authority: &AttemptAuthority,
+    ) -> Result<RecoveryOutcome, AttemptError> {
+        validate_authority(authority)?;
+        let job_id = authority.job_id.clone();
+        let attempt_id = authority.attempt_id.clone();
+        let worker = authority.worker_incarnation.clone();
+        let fence = authority.fence;
+        let coordination = self.coordination.clone();
+        let faults = self.faults.clone();
+        let outcome = with_transaction(&self.pool, move |tx| {
+            Box::pin(async move {
+                let row = lock_authority(tx, &attempt_id).await?;
+                let Some(row) = row else {
+                    return Ok(RecoveryOutcome::AlreadyInactive);
+                };
+                if !row.matches(&job_id, fence, &worker) {
+                    return Ok(RecoveryOutcome::AlreadyInactive);
+                }
+                if row.state == AttemptState::Executing {
+                    return Ok(RecoveryOutcome::StillOwned);
+                }
+                if row.state != AttemptState::Authorized {
+                    return Ok(RecoveryOutcome::AlreadyInactive);
+                }
+                faults.hit(AttemptFaultPoint::RecoveryBeforeRevalidation);
+                if coordination
+                    .get_reservation(job_id.as_str())
+                    .await
+                    .map_err(DurableError::effect)?
+                    .is_some()
+                {
+                    return Ok(RecoveryOutcome::StillOwned);
+                }
+                if let Some(lease) = coordination
+                    .get_live_lease(job_id.as_str())
+                    .await
+                    .map_err(DurableError::effect)?
+                {
+                    if lease.record.owner()
+                        == LeaseOwner::new(attempt_id.as_str(), fence, worker.as_str())
+                            .map_err(DurableError::effect)?
+                    {
+                        return Ok(RecoveryOutcome::StillOwned);
+                    }
+                    return Err(DurableError::Effect(
+                        "foreign live lease conflicts with durable authority".into(),
+                    ));
+                }
+                let now = OffsetDateTime::now_utc();
+                faults.hit(AttemptFaultPoint::RecoveryBeforeDurable);
+                sqlx::query(
+                    "UPDATE attempts
+                     SET state = 'reaped', reaped_at = $2
+                     WHERE attempt_id = $1 AND state = 'authorized'",
+                )
+                .bind(attempt_id.as_str())
+                .bind(now)
+                .execute(tx.as_mut())
+                .await?;
+                let changed = sqlx::query(
+                    "UPDATE jobs
+                     SET phase = 'queued', authority_attempt_id = NULL,
+                         dispatch_needed = TRUE, revision = revision + 1, updated_at = $2
+                     WHERE job_id = $1 AND authority_attempt_id = $3
+                       AND current_fence = $4 AND phase = 'running'",
+                )
+                .bind(job_id.as_str())
+                .bind(now)
+                .bind(attempt_id.as_str())
+                .bind(i64::try_from(fence).map_err(DurableError::effect)?)
+                .execute(tx.as_mut())
+                .await?;
+                if changed.rows_affected() != 1 {
+                    return Err(DurableError::Effect(
+                        "attempt authority changed while recovery held its lock".into(),
+                    ));
+                }
+                enqueue(tx, "job.dispatch", json!({ "job_id": job_id })).await?;
+                Ok(RecoveryOutcome::Reaped)
+            })
+        })
+        .await?;
+        self.faults.hit(AttemptFaultPoint::RecoveryAfterDurable);
+        if matches!(
+            outcome,
+            RecoveryOutcome::Reaped | RecoveryOutcome::AlreadyInactive
+        ) {
+            self.release_exact(authority).await;
+        }
+        Ok(outcome)
+    }
+
+    pub async fn authority(
+        &self,
+        attempt_id: &OpaqueId,
+    ) -> Result<Option<AttemptAuthority>, AttemptError> {
+        load_authority(&self.pool, attempt_id)
+            .await
+            .map_err(Into::into)
+    }
+
+    pub fn release_local(&self, job_id: &OpaqueId) -> bool {
+        self.local_permits.release_all(job_id.as_str())
+    }
+
+    #[must_use]
+    pub fn local_used(&self) -> u64 {
+        self.local_permits.used()
+    }
+
+    async fn compensate_reservation(&self, job_id: &OpaqueId, worker: &OpaqueId) {
+        let _ = self
+            .coordination
+            .free_reserve(job_id.as_str(), Some(worker.as_str()))
+            .await;
+        self.local_permits.release(job_id.as_str());
+    }
+
+    async fn release_exact(&self, authority: &AttemptAuthority) {
+        if let Ok(owner) = lease_owner(authority) {
+            let _ = self
+                .coordination
+                .release(authority.job_id.as_str(), &owner)
+                .await;
+        }
+    }
+
+    async fn exact_authority_is_executing(
+        &self,
+        authority: &AttemptAuthority,
+    ) -> Result<bool, AttemptError> {
+        Ok(load_authority(&self.pool, &authority.attempt_id)
+            .await?
+            .is_some_and(|current| current == authority.clone_with_state(AttemptState::Executing)))
+    }
+
+    async fn exact_authority_is_executing_and_live(
+        &self,
+        authority: &AttemptAuthority,
+    ) -> Result<bool, AttemptError> {
+        if !self.exact_authority_is_executing(authority).await? {
+            return Ok(false);
+        }
+        let Some(lease) = self
+            .coordination
+            .get_live_lease(authority.job_id.as_str())
+            .await?
+        else {
+            return Ok(false);
+        };
+        Ok(lease.record.owner() == lease_owner(authority)?)
+    }
+}
+
+impl AttemptAuthority {
+    fn clone_with_state(&self, state: AttemptState) -> Self {
+        Self {
+            job_id: self.job_id.clone(),
+            attempt_id: self.attempt_id.clone(),
+            fence: self.fence,
+            worker_incarnation: self.worker_incarnation.clone(),
+            state,
+        }
+    }
+}
+
+#[derive(Serialize)]
+struct ClaimRequest<'a> {
+    job_id: &'a OpaqueId,
+    worker_incarnation: &'a OpaqueId,
+}
+
+#[derive(Serialize)]
+struct ActivationRequest<'a> {
+    job_id: &'a OpaqueId,
+    attempt_id: &'a OpaqueId,
+    fence: u64,
+    worker_incarnation: &'a OpaqueId,
+}
+
+struct AuthorityRow {
+    job_id: OpaqueId,
+    fence: u64,
+    worker: OpaqueId,
+    state: AttemptState,
+}
+
+impl AuthorityRow {
+    fn matches(&self, job_id: &OpaqueId, fence: u64, worker: &OpaqueId) -> bool {
+        self.job_id == *job_id && self.fence == fence && self.worker == *worker
+    }
+}
+
+struct AuthorizeInput<'a> {
+    coordination: &'a CoordinationStore,
+    job_id: &'a OpaqueId,
+    attempt_id: &'a OpaqueId,
+    worker: &'a OpaqueId,
+    max_attempts: u64,
+    faults: &'a dyn AttemptFaultInjector,
+    reserved: &'a AtomicBool,
+}
+
+async fn authorize_attempt(
+    tx: &mut remora_durable::PgTx<'_>,
+    input: AuthorizeInput<'_>,
+) -> Result<ClaimOutcome, DurableError> {
+    let row = sqlx::query_as::<_, (String, bool, i64, Option<String>)>(
+        "SELECT phase, dispatch_needed, current_fence, authority_attempt_id
+         FROM jobs WHERE job_id = $1 FOR UPDATE",
+    )
+    .bind(input.job_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?;
+    let Some((phase, dispatch_needed, current_fence, authority)) = row else {
+        return Ok(claim_rejected(ClaimRejection::JobUnavailable));
+    };
+    if phase != "queued" || !dispatch_needed || authority.is_some() {
+        return Ok(claim_rejected(ClaimRejection::JobUnavailable));
+    }
+    let current_fence = u64::try_from(current_fence).map_err(DurableError::effect)?;
+    if current_fence >= input.max_attempts {
+        return Ok(claim_rejected(ClaimRejection::AttemptsExhausted));
+    }
+    input.faults.hit(AttemptFaultPoint::ClaimBeforeReserve);
+    let reservation = input
+        .coordination
+        .reserve(input.job_id.as_str(), input.worker.as_str())
+        .await
+        .map_err(DurableError::effect)?;
+    if !reservation.ok() {
+        return Ok(claim_rejected(match reservation.code {
+            OpCode::FailFull => ClaimRejection::GlobalCapacity,
+            _ => ClaimRejection::ReservationHeld,
+        }));
+    }
+    input.reserved.store(true, Ordering::Release);
+    input.faults.hit(AttemptFaultPoint::ClaimAfterReserve);
+    let fence = current_fence + 1;
+    let now = OffsetDateTime::now_utc();
+    sqlx::query(
+        "INSERT INTO attempts (
+            attempt_id, job_id, fence, worker_incarnation, state,
+            initial_progress, authorized_at, executing_at, reaped_at
+         ) VALUES ($1, $2, $3, $4, 'authorized', 0, $5, NULL, NULL)",
+    )
+    .bind(input.attempt_id.as_str())
+    .bind(input.job_id.as_str())
+    .bind(i64::try_from(fence).map_err(DurableError::effect)?)
+    .bind(input.worker.as_str())
+    .bind(now)
+    .execute(tx.as_mut())
+    .await?;
+    sqlx::query(
+        "UPDATE jobs
+         SET phase = 'running', current_fence = $2, authority_attempt_id = $3,
+             dispatch_needed = FALSE, revision = revision + 1, updated_at = $4
+         WHERE job_id = $1",
+    )
+    .bind(input.job_id.as_str())
+    .bind(i64::try_from(fence).map_err(DurableError::effect)?)
+    .bind(input.attempt_id.as_str())
+    .bind(now)
+    .execute(tx.as_mut())
+    .await?;
+    Ok(ClaimOutcome::Claimed {
+        authority: AttemptAuthority {
+            job_id: input.job_id.clone(),
+            attempt_id: input.attempt_id.clone(),
+            fence,
+            worker_incarnation: input.worker.clone(),
+            state: AttemptState::Authorized,
+        },
+        disposition: None,
+    })
+}
+
+async fn activate_attempt(
+    tx: &mut remora_durable::PgTx<'_>,
+    coordination: &CoordinationStore,
+    authority: &AttemptAuthority,
+    faults: &dyn AttemptFaultInjector,
+) -> Result<bool, DurableError> {
+    let Some(row) = lock_authority(tx, &authority.attempt_id).await? else {
+        return Ok(false);
+    };
+    if !row.matches(
+        &authority.job_id,
+        authority.fence,
+        &authority.worker_incarnation,
+    ) {
+        return Ok(false);
+    }
+    if row.state == AttemptState::Executing {
+        let live = coordination
+            .get_live_lease(authority.job_id.as_str())
+            .await
+            .map_err(DurableError::effect)?;
+        let expected = lease_owner(authority).map_err(DurableError::effect)?;
+        if !live.is_some_and(|lease| lease.record.owner() == expected) {
+            return Ok(false);
+        }
+        return Ok(coordination
+            .renew(authority.job_id.as_str(), &expected)
+            .await
+            .map_err(DurableError::effect)?
+            .ok());
+    }
+    if row.state != AttemptState::Authorized {
+        return Ok(false);
+    }
+    faults.hit(AttemptFaultPoint::ActivationBeforeLease);
+    let owner = lease_owner(authority).map_err(DurableError::effect)?;
+    let coordinated = coordination
+        .activate(
+            authority.job_id.as_str(),
+            &owner,
+            ActivateOptions {
+                seed_progress: true,
+                require_reserve: true,
+            },
+        )
+        .await
+        .map_err(DurableError::effect)?;
+    if !coordinated.ok() {
+        return Ok(false);
+    }
+    faults.hit(AttemptFaultPoint::ActivationAfterLease);
+    let live = coordination
+        .get_live_lease(authority.job_id.as_str())
+        .await
+        .map_err(DurableError::effect)?;
+    let expected = lease_owner(authority).map_err(DurableError::effect)?;
+    if !live.is_some_and(|lease| lease.record.owner() == expected) {
+        return Ok(false);
+    }
+    let now = OffsetDateTime::now_utc();
+    sqlx::query(
+        "UPDATE attempts SET state = 'executing', executing_at = $2
+         WHERE attempt_id = $1 AND state = 'authorized'",
+    )
+    .bind(authority.attempt_id.as_str())
+    .bind(now)
+    .execute(tx.as_mut())
+    .await?;
+    Ok(true)
+}
+
+async fn lock_authority(
+    tx: &mut remora_durable::PgTx<'_>,
+    attempt_id: &OpaqueId,
+) -> Result<Option<AuthorityRow>, DurableError> {
+    let row = sqlx::query_as::<_, (String, i64, String, String)>(
+        "SELECT a.job_id, a.fence, a.worker_incarnation, a.state
+         FROM attempts a
+         JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.attempt_id = $1
+           AND j.authority_attempt_id = a.attempt_id
+           AND j.current_fence = a.fence
+           AND j.phase = 'running'
+         FOR UPDATE OF a, j",
+    )
+    .bind(attempt_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?;
+    row.map(authority_row).transpose()
+}
+
+async fn load_authority(
+    pool: &PgPool,
+    attempt_id: &OpaqueId,
+) -> Result<Option<AttemptAuthority>, DurableError> {
+    let row = sqlx::query_as::<_, (String, String, i64, String, String)>(
+        "SELECT a.job_id, a.attempt_id, a.fence, a.worker_incarnation, a.state
+         FROM attempts a
+         JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.attempt_id = $1
+           AND j.authority_attempt_id = a.attempt_id
+           AND j.current_fence = a.fence
+           AND j.phase = 'running'",
+    )
+    .bind(attempt_id.as_str())
+    .fetch_optional(pool)
+    .await?;
+    row.map(|(job, attempt, fence, worker, state)| {
+        Ok(AttemptAuthority {
+            job_id: OpaqueId::parse(job).map_err(DurableError::effect)?,
+            attempt_id: OpaqueId::parse(attempt).map_err(DurableError::effect)?,
+            fence: u64::try_from(fence).map_err(DurableError::effect)?,
+            worker_incarnation: OpaqueId::parse(worker).map_err(DurableError::effect)?,
+            state: AttemptState::parse(&state)
+                .ok_or_else(|| DurableError::Config(format!("unknown attempt state '{state}'")))?,
+        })
+    })
+    .transpose()
+}
+
+fn authority_row(row: (String, i64, String, String)) -> Result<AuthorityRow, DurableError> {
+    Ok(AuthorityRow {
+        job_id: OpaqueId::parse(row.0).map_err(DurableError::effect)?,
+        fence: u64::try_from(row.1).map_err(DurableError::effect)?,
+        worker: OpaqueId::parse(row.2).map_err(DurableError::effect)?,
+        state: AttemptState::parse(&row.3)
+            .ok_or_else(|| DurableError::Config(format!("unknown attempt state '{}'", row.3)))?,
+    })
+}
+
+fn validate_authority(authority: &AttemptAuthority) -> Result<(), AttemptError> {
+    if authority.fence == 0 || authority.fence > MAX_SAFE_INTEGER {
+        return Err(AttemptError::Invalid("invalid fencing token".into()));
+    }
+    Ok(())
+}
+
+fn lease_owner(authority: &AttemptAuthority) -> Result<LeaseOwner, AttemptError> {
+    LeaseOwner::new(
+        authority.attempt_id.as_str(),
+        authority.fence,
+        authority.worker_incarnation.as_str(),
+    )
+    .map_err(Into::into)
+}
+
+fn authority_binding(authority: &AttemptAuthority) -> Vec<u8> {
+    format!(
+        "job={}|attempt={}|fence={}|worker={}",
+        authority.job_id, authority.attempt_id, authority.fence, authority.worker_incarnation
+    )
+    .into_bytes()
+}
+
+fn claim_rejected(reason: ClaimRejection) -> ClaimOutcome {
+    ClaimOutcome::Rejected {
+        reason,
+        disposition: None,
+    }
+}
+
+fn new_id(prefix: char) -> Result<OpaqueId, AttemptError> {
+    let mut bytes = [0u8; 16];
+    getrandom::fill(&mut bytes)
+        .map_err(|error| AttemptError::Invalid(format!("identifier entropy failed: {error}")))?;
+    let mut value = String::with_capacity(33);
+    value.push(prefix);
+    for byte in bytes {
+        use std::fmt::Write as _;
+        write!(value, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    OpaqueId::parse(value).map_err(|error| AttemptError::Invalid(error.to_string()))
+}

crates/remora-attempt/src/types.rs

@@ -0,0 +1,105 @@
+use remora_domain::OpaqueId;
+use remora_durable::MutationDisposition;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum AttemptFaultPoint {
+    ClaimBeforeReserve,
+    ClaimAfterReserve,
+    ClaimAfterDurable,
+    ActivationBeforeLease,
+    ActivationAfterLease,
+    ActivationAfterDurable,
+    RenewalBeforeLease,
+    RenewalAfterLease,
+    RecoveryBeforeRevalidation,
+    RecoveryBeforeDurable,
+    RecoveryAfterDurable,
+}
+
+pub trait AttemptFaultInjector: Send + Sync {
+    fn hit(&self, point: AttemptFaultPoint);
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AttemptState {
+    Authorized,
+    Executing,
+    Reaped,
+}
+
+impl AttemptState {
+    pub(crate) fn parse(value: &str) -> Option<Self> {
+        Some(match value {
+            "authorized" => Self::Authorized,
+            "executing" => Self::Executing,
+            "reaped" => Self::Reaped,
+            _ => return None,
+        })
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct AttemptAuthority {
+    pub job_id: OpaqueId,
+    pub attempt_id: OpaqueId,
+    pub fence: u64,
+    pub worker_incarnation: OpaqueId,
+    pub state: AttemptState,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ClaimRejection {
+    JobUnavailable,
+    AttemptsExhausted,
+    LocalCapacity,
+    GlobalCapacity,
+    ReservationHeld,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "status", rename_all = "snake_case")]
+pub enum ClaimOutcome {
+    Claimed {
+        authority: AttemptAuthority,
+        #[serde(skip)]
+        disposition: Option<MutationDisposition>,
+    },
+    Rejected {
+        reason: ClaimRejection,
+        #[serde(skip)]
+        disposition: Option<MutationDisposition>,
+    },
+}
+
+impl ClaimOutcome {
+    pub(crate) fn with_disposition(mut self, value: MutationDisposition) -> Self {
+        match &mut self {
+            Self::Claimed { disposition, .. } | Self::Rejected { disposition, .. } => {
+                *disposition = Some(value);
+            }
+        }
+        self
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ActivationOutcome {
+    Activated(MutationDisposition),
+    PreconditionFailed,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum RenewalOutcome {
+    Renewed,
+    AuthorityLost,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum RecoveryOutcome {
+    StillOwned,
+    Reaped,
+    AlreadyInactive,
+}

crates/remora-attempt/tests/authority.rs

@@ -0,0 +1,796 @@
+use std::collections::HashSet;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use remora_admission::{
+    AdmissionConfig, AdmissionDecision, AdmissionService, migrate as admission_migrate,
+};
+use remora_attempt::{
+    ActivationOutcome, AttemptConfig, AttemptFaultInjector, AttemptFaultPoint, AttemptService,
+    AttemptState, ClaimOutcome, RecoveryOutcome, RenewalOutcome, migrate as attempt_migrate,
+};
+use remora_contracts::{HttpSourceSubmissionDto, JobSubmissionDto, SourceSubmissionDto};
+use remora_coordination::{ActivateOptions, CoordinationStore, LeaseOwner, StoreConfig};
+use remora_credentials::migrate as credentials_migrate;
+use remora_domain::{BucketName, OpaqueId, StorageScope, TargetPrefix, TargetStoreId};
+use remora_durable::{MutationDisposition, PoolConfig, connect, migrate as durable_migrate};
+use remora_identity::{DefaultTarget, Principal};
+use sqlx::PgPool;
+use sqlx::postgres::PgPoolOptions;
+use tokio::time::sleep;
+
+struct Harness {
+    pool: PgPool,
+    coordination: CoordinationStore,
+}
+
+impl Harness {
+    async fn new(name: &str, ttl: Duration, global_limit: u64) -> Option<Self> {
+        let database_url = required_env("DATABASE_URL")?;
+        let coordination_url = required_env("REMORA_COORDINATION_URL")?;
+        let pool = connect(
+            &database_url,
+            PoolConfig {
+                max_connections: 12,
+                acquire_timeout: Duration::from_secs(10),
+                idle_timeout: Some(Duration::from_secs(30)),
+            },
+        )
+        .await
+        .expect("connect postgres");
+        durable_migrate(&pool).await.expect("durable migrate");
+        credentials_migrate(&pool)
+            .await
+            .expect("credentials migrate");
+        admission_migrate(&pool).await.expect("admission migrate");
+        attempt_migrate(&pool).await.expect("attempt migrate");
+        let namespace = format!("{}wp15:{name}:{}:", coordination_namespace(), nonce());
+        let coordination = CoordinationStore::connect(
+            &coordination_url,
+            namespace,
+            StoreConfig {
+                global_limit,
+                lease_ttl: ttl,
+                reserve_ttl: ttl,
+                progress_ttl: ttl,
+                cancel_ttl: None,
+            },
+        )
+        .await
+        .expect("connect coordination store");
+        Some(Self { pool, coordination })
+    }
+
+    fn service(&self, max_attempts: u64, local_limit: u64) -> AttemptService {
+        AttemptService::new(
+            self.pool.clone(),
+            self.coordination.clone(),
+            AttemptConfig {
+                max_attempts,
+                local_limit,
+            },
+        )
+        .expect("attempt service")
+    }
+
+    fn service_with_fault(
+        &self,
+        point: AttemptFaultPoint,
+        max_attempts: u64,
+        local_limit: u64,
+    ) -> AttemptService {
+        AttemptService::new_with_fault_injector(
+            self.pool.clone(),
+            self.coordination.clone(),
+            AttemptConfig {
+                max_attempts,
+                local_limit,
+            },
+            Arc::new(PanicOnce::new(point)),
+        )
+        .expect("fault-injected attempt service")
+    }
+
+    async fn job(&self) -> OpaqueId {
+        let owner = unique_id("owner");
+        let store = TargetStoreId::parse("primary").unwrap();
+        let bucket = BucketName::parse("remora-results").unwrap();
+        let prefix = TargetPrefix::parse("users/alice", ["_remora"]).unwrap();
+        let principal = Principal {
+            id: owner,
+            issuer: "https://issuer.example".into(),
+            subject: "subject".into(),
+            administrator: false,
+            scopes: [StorageScope {
+                store: store.clone(),
+                bucket: bucket.clone(),
+                prefix: prefix.clone(),
+            }]
+            .into_iter()
+            .collect(),
+        };
+        let admission = AdmissionService::new(
+            self.pool.clone(),
+            AdmissionConfig {
+                default_target: DefaultTarget {
+                    store,
+                    bucket,
+                    prefix,
+                },
+                reserved_target_segments: vec!["_remora".into()],
+                policy_version: "attempt-tests-v1".into(),
+            },
+        )
+        .unwrap();
+        let request = JobSubmissionDto {
+            source: SourceSubmissionDto::Http(HttpSourceSubmissionDto {
+                url: "https://example.invalid/artifact".into(),
+                result_path: "artifact.bin".into(),
+                credential_id: None,
+            }),
+            target: None,
+            label: Some("attempt-test".into()),
+        };
+        let outcome = admission
+            .submit(&principal, unique_id("submit").as_str(), request)
+            .await
+            .expect("submit job");
+        let AdmissionDecision::Accepted(job) = outcome.decision else {
+            panic!("job rejected");
+        };
+        self.coordination
+            .enqueue(job.job_id.as_str())
+            .await
+            .expect("project queue");
+        job.job_id
+    }
+}
+
+#[tokio::test]
+async fn claim_replay_conflict_and_fence_allocation_are_durable() {
+    let Some(harness) = Harness::new("claim", Duration::from_secs(2), 8).await else {
+        return;
+    };
+    let job = harness.job().await;
+    let worker = unique_id("worker");
+    let operation = unique_id("claimop");
+    let service = harness.service(2, 2);
+
+    let first = claimed(
+        service
+            .claim(&job, &worker, &operation)
+            .await
+            .expect("claim"),
+    );
+    assert_eq!(first.0, 1);
+    assert_eq!(first.1, MutationDisposition::First);
+    let replay = claimed(
+        service
+            .claim(&job, &worker, &operation)
+            .await
+            .expect("claim replay"),
+    );
+    assert_eq!(replay.0, 1);
+    assert_eq!(replay.1, MutationDisposition::Replay);
+    assert_eq!(first.2, replay.2);
+
+    let other_job = harness.job().await;
+    assert!(
+        service
+            .claim(&other_job, &worker, &operation)
+            .await
+            .is_err(),
+        "one operation identity cannot be rebound to another job"
+    );
+
+    sleep(Duration::from_millis(2200)).await;
+    let authority = service.authority(&first.2).await.unwrap().unwrap();
+    assert_eq!(
+        service.recover_inactive(&authority).await.unwrap(),
+        RecoveryOutcome::Reaped
+    );
+    service.release_local(&job);
+    harness.coordination.enqueue(job.as_str()).await.unwrap();
+    let second = claimed(
+        service
+            .claim(&job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    assert_eq!(second.0, 2);
+    sleep(Duration::from_millis(2200)).await;
+    let second_authority = service.authority(&second.2).await.unwrap().unwrap();
+    service.recover_inactive(&second_authority).await.unwrap();
+    service.release_local(&job);
+    harness.coordination.enqueue(job.as_str()).await.unwrap();
+    let exhausted = service
+        .claim(&job, &worker, &unique_id("claimop"))
+        .await
+        .unwrap();
+    assert!(matches!(exhausted, ClaimOutcome::Rejected { .. }));
+}
+
+#[tokio::test]
+async fn activation_replay_requires_the_complete_live_authority_tuple() {
+    let Some(harness) = Harness::new("activation", Duration::from_secs(3), 8).await else {
+        return;
+    };
+    let job = harness.job().await;
+    let worker = unique_id("worker");
+    let service = harness.service(3, 2);
+    let claim_operation = unique_id("claimop");
+    let (_, _, attempt) = claimed(
+        service
+            .claim(&job, &worker, &claim_operation)
+            .await
+            .unwrap(),
+    );
+    let authority = service.authority(&attempt).await.unwrap().unwrap();
+    assert!(matches!(
+        service
+            .claim(&job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+        ClaimOutcome::Rejected { .. }
+    ));
+    let unavailable_pool = PgPoolOptions::new()
+        .acquire_timeout(Duration::from_millis(100))
+        .connect_lazy("postgres://remora:[email protected]:1/remora")
+        .unwrap();
+    let unavailable = AttemptService::new(
+        unavailable_pool,
+        harness.coordination.clone(),
+        AttemptConfig {
+            max_attempts: 3,
+            local_limit: 2,
+        },
+    )
+    .unwrap();
+    assert!(
+        unavailable
+            .claim(&job, &worker, &claim_operation)
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        harness
+            .coordination
+            .get_reservation(job.as_str())
+            .await
+            .unwrap()
+            .unwrap()
+            .worker_id,
+        worker.as_str(),
+        "durable outage before mutation observation must preserve the claim reservation"
+    );
+    assert!(
+        service
+            .activate(&authority, &claim_operation)
+            .await
+            .is_err(),
+        "an operation identity cannot change mutation kind"
+    );
+    let operation = unique_id("activateop");
+    assert_eq!(
+        service.activate(&authority, &operation).await.unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::First)
+    );
+    let executing = service.authority(&attempt).await.unwrap().unwrap();
+    assert_eq!(executing.state, AttemptState::Executing);
+    assert_eq!(
+        service.activate(&executing, &operation).await.unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::Replay)
+    );
+    assert_eq!(
+        service
+            .activate(&executing, &unique_id("activateop"))
+            .await
+            .unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::First)
+    );
+    assert_eq!(
+        service.renew(&authority).await.unwrap(),
+        RenewalOutcome::Renewed
+    );
+
+    let mut forged = authority.clone();
+    forged.worker_incarnation = unique_id("worker");
+    assert!(service.activate(&forged, &operation).await.is_err());
+    assert_eq!(
+        service.renew(&authority).await.unwrap(),
+        RenewalOutcome::Renewed,
+        "conflicting operation must preserve the established lease"
+    );
+    assert_eq!(
+        service.recover_inactive(&authority).await.unwrap(),
+        RecoveryOutcome::StillOwned
+    );
+    assert_eq!(
+        service.renew(&authority).await.unwrap(),
+        RenewalOutcome::Renewed,
+        "inactive recovery must preserve an exact executing authority"
+    );
+
+    harness
+        .coordination
+        .release(job.as_str(), &owner(&authority))
+        .await
+        .unwrap();
+    assert_eq!(
+        service.activate(&authority, &operation).await.unwrap(),
+        ActivationOutcome::PreconditionFailed
+    );
+    assert_eq!(
+        service.renew(&authority).await.unwrap(),
+        RenewalOutcome::AuthorityLost
+    );
+    assert_eq!(
+        service.recover_inactive(&authority).await.unwrap(),
+        RecoveryOutcome::StillOwned,
+        "missing coordination never reverses executing durable authority"
+    );
+}
+
+#[tokio::test]
+async fn activation_and_inactive_recovery_serialize_on_durable_authority() {
+    let Some(harness) = Harness::new("serialize", Duration::from_millis(700), 8).await else {
+        return;
+    };
+    let job = harness.job().await;
+    let worker = unique_id("worker");
+    let service = harness.service(3, 2);
+    let (_, _, attempt) = claimed(
+        service
+            .claim(&job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    let authority = service.authority(&attempt).await.unwrap().unwrap();
+    sleep(Duration::from_millis(800)).await;
+
+    let activation_service = service.clone();
+    let activation_authority = authority.clone();
+    let activation = tokio::spawn(async move {
+        activation_service
+            .activate(&activation_authority, &unique_id("activateop"))
+            .await
+            .unwrap()
+    });
+    let recovery_service = service.clone();
+    let recovery_authority = authority.clone();
+    let recovery = tokio::spawn(async move {
+        recovery_service
+            .recover_inactive(&recovery_authority)
+            .await
+            .unwrap()
+    });
+    let activation = activation.await.unwrap();
+    let recovery = recovery.await.unwrap();
+    assert!(
+        matches!(
+            (activation, recovery),
+            (
+                ActivationOutcome::PreconditionFailed,
+                RecoveryOutcome::Reaped
+            )
+        ),
+        "expired reservation cannot activate while recovery reaps: {activation:?} {recovery:?}"
+    );
+    assert!(service.authority(&attempt).await.unwrap().is_none());
+}
+
+#[tokio::test]
+async fn local_and_global_permits_bound_claims_and_survive_lease_expiry() {
+    let Some(harness) = Harness::new("permits", Duration::from_millis(500), 1).await else {
+        return;
+    };
+    let service = harness.service(3, 1);
+    let worker = unique_id("worker");
+    let first_job = harness.job().await;
+    let (_, _, attempt) = claimed(
+        service
+            .claim(&first_job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    let authority = service.authority(&attempt).await.unwrap().unwrap();
+    service
+        .activate(&authority, &unique_id("activateop"))
+        .await
+        .unwrap();
+    sleep(Duration::from_millis(600)).await;
+    assert_eq!(service.local_used(), 1);
+
+    let second_job = harness.job().await;
+    assert!(matches!(
+        service
+            .claim(&second_job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+        ClaimOutcome::Rejected { .. }
+    ));
+    service.release_local(&first_job);
+    harness
+        .coordination
+        .expire_lease(first_job.as_str())
+        .await
+        .unwrap();
+    assert_eq!(service.local_used(), 0);
+}
+
+#[tokio::test]
+async fn precommit_lease_is_compensated_when_durable_authority_was_reaped() {
+    let Some(harness) = Harness::new("compensate", Duration::from_millis(700), 8).await else {
+        return;
+    };
+    let job = harness.job().await;
+    let worker = unique_id("worker");
+    let service = harness.service(3, 2);
+    let (_, _, attempt) = claimed(
+        service
+            .claim(&job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    let authority = service.authority(&attempt).await.unwrap().unwrap();
+    let lease = harness
+        .coordination
+        .activate(
+            job.as_str(),
+            &owner(&authority),
+            ActivateOptions {
+                seed_progress: true,
+                require_reserve: true,
+            },
+        )
+        .await
+        .unwrap();
+    assert!(lease.ok());
+    harness
+        .coordination
+        .release(job.as_str(), &owner(&authority))
+        .await
+        .unwrap();
+    sleep(Duration::from_millis(800)).await;
+    assert_eq!(
+        service.recover_inactive(&authority).await.unwrap(),
+        RecoveryOutcome::Reaped
+    );
+    assert_eq!(
+        service
+            .activate(&authority, &unique_id("activateop"))
+            .await
+            .unwrap(),
+        ActivationOutcome::PreconditionFailed
+    );
+    assert!(
+        harness
+            .coordination
+            .get_live_lease(job.as_str())
+            .await
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
+    let Some(harness) = Harness::new("crash-matrix", Duration::from_millis(500), 32).await else {
+        return;
+    };
+    let worker = unique_id("worker");
+    let normal = harness.service(32, 32);
+
+    let before_reserve_job = harness.job().await;
+    crash_claim(
+        harness.service_with_fault(AttemptFaultPoint::ClaimBeforeReserve, 32, 32),
+        before_reserve_job.clone(),
+        worker.clone(),
+        unique_id("claimop"),
+    )
+    .await;
+    assert!(
+        harness
+            .coordination
+            .get_reservation(before_reserve_job.as_str())
+            .await
+            .unwrap()
+            .is_none()
+    );
+
+    let after_reserve_job = harness.job().await;
+    crash_claim(
+        harness.service_with_fault(AttemptFaultPoint::ClaimAfterReserve, 32, 32),
+        after_reserve_job.clone(),
+        worker.clone(),
+        unique_id("claimop"),
+    )
+    .await;
+    assert!(
+        harness
+            .coordination
+            .get_reservation(after_reserve_job.as_str())
+            .await
+            .unwrap()
+            .is_some()
+    );
+    harness
+        .coordination
+        .free_reserve(after_reserve_job.as_str(), Some(worker.as_str()))
+        .await
+        .unwrap();
+
+    let lost_claim_job = harness.job().await;
+    let lost_claim_operation = unique_id("claimop");
+    crash_claim(
+        harness.service_with_fault(AttemptFaultPoint::ClaimAfterDurable, 32, 32),
+        lost_claim_job.clone(),
+        worker.clone(),
+        lost_claim_operation.clone(),
+    )
+    .await;
+    let (_, disposition, lost_claim_attempt) = claimed(
+        normal
+            .claim(&lost_claim_job, &worker, &lost_claim_operation)
+            .await
+            .unwrap(),
+    );
+    assert_eq!(disposition, MutationDisposition::Replay);
+    let lost_claim_authority = normal
+        .authority(&lost_claim_attempt)
+        .await
+        .unwrap()
+        .unwrap();
+
+    let before_lease_operation = unique_id("activateop");
+    crash_activate(
+        harness.service_with_fault(AttemptFaultPoint::ActivationBeforeLease, 32, 32),
+        lost_claim_authority.clone(),
+        before_lease_operation.clone(),
+    )
+    .await;
+    assert!(
+        harness
+            .coordination
+            .get_live_lease(lost_claim_job.as_str())
+            .await
+            .unwrap()
+            .is_none()
+    );
+    assert_eq!(
+        normal
+            .activate(&lost_claim_authority, &before_lease_operation)
+            .await
+            .unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::First)
+    );
+
+    let after_lease_job = harness.job().await;
+    let (_, _, after_lease_attempt) = claimed(
+        normal
+            .claim(&after_lease_job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    let after_lease_authority = normal
+        .authority(&after_lease_attempt)
+        .await
+        .unwrap()
+        .unwrap();
+    let after_lease_operation = unique_id("activateop");
+    crash_activate(
+        harness.service_with_fault(AttemptFaultPoint::ActivationAfterLease, 32, 32),
+        after_lease_authority.clone(),
+        after_lease_operation.clone(),
+    )
+    .await;
+    assert!(
+        harness
+            .coordination
+            .get_live_lease(after_lease_job.as_str())
+            .await
+            .unwrap()
+            .is_some()
+    );
+    assert_eq!(
+        normal
+            .activate(&after_lease_authority, &after_lease_operation)
+            .await
+            .unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::First)
+    );
+
+    let lost_activation_job = harness.job().await;
+    let (_, _, lost_activation_attempt) = claimed(
+        normal
+            .claim(&lost_activation_job, &worker, &unique_id("claimop"))
+            .await
+            .unwrap(),
+    );
+    let lost_activation_authority = normal
+        .authority(&lost_activation_attempt)
+        .await
+        .unwrap()
+        .unwrap();
+    let lost_activation_operation = unique_id("activateop");
+    crash_activate(
+        harness.service_with_fault(AttemptFaultPoint::ActivationAfterDurable, 32, 32),
+        lost_activation_authority.clone(),
+        lost_activation_operation.clone(),
+    )
+    .await;
+    let executing = normal
+        .authority(&lost_activation_attempt)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        normal
+            .activate(&executing, &lost_activation_operation)
+            .await
+            .unwrap(),
+        ActivationOutcome::Activated(MutationDisposition::Replay)
+    );
+
+    crash_renew(
+        harness.service_with_fault(AttemptFaultPoint::RenewalBeforeLease, 32, 32),
+        executing.clone(),
+    )
+    .await;
+    crash_renew(
+        harness.service_with_fault(AttemptFaultPoint::RenewalAfterLease, 32, 32),
+        executing.clone(),
+    )
+    .await;
+    assert_eq!(
+        normal.renew(&executing).await.unwrap(),
+        RenewalOutcome::Renewed
+    );
+
+    let mut recovery_cases = Vec::new();
+    for point in [
+        AttemptFaultPoint::RecoveryBeforeRevalidation,
+        AttemptFaultPoint::RecoveryBeforeDurable,
+        AttemptFaultPoint::RecoveryAfterDurable,
+    ] {
+        let job = harness.job().await;
+        let (_, _, attempt) = claimed(
+            normal
+                .claim(&job, &worker, &unique_id("claimop"))
+                .await
+                .unwrap(),
+        );
+        let authority = normal.authority(&attempt).await.unwrap().unwrap();
+        recovery_cases.push((point, authority));
+    }
+    sleep(Duration::from_millis(600)).await;
+    for (point, authority) in recovery_cases {
+        let used_before = harness.coordination.global_used().await.unwrap();
+        crash_recovery(harness.service_with_fault(point, 32, 32), authority.clone()).await;
+        let expected = if point == AttemptFaultPoint::RecoveryAfterDurable {
+            RecoveryOutcome::AlreadyInactive
+        } else {
+            RecoveryOutcome::Reaped
+        };
+        assert_eq!(normal.recover_inactive(&authority).await.unwrap(), expected);
+        assert_eq!(
+            harness.coordination.global_used().await.unwrap(),
+            used_before - 1,
+            "recovery retry must release the expired attempt's global slot"
+        );
+    }
+}
+
+fn required_env(name: &str) -> Option<String> {
+    match std::env::var(name).ok().filter(|value| !value.is_empty()) {
+        Some(value) => Some(value),
+        None if std::env::var("REMORA_ATTEMPT_REQUIRE").is_ok() => {
+            panic!("{name} is required when REMORA_ATTEMPT_REQUIRE is set")
+        }
+        None => None,
+    }
+}
+
+fn coordination_namespace() -> String {
+    std::env::var("REMORA_COORDINATION_NAMESPACE")
+        .ok()
+        .filter(|value| !value.is_empty())
+        .unwrap_or_else(|| "remora:test:".into())
+}
+
+fn unique_id(prefix: &str) -> OpaqueId {
+    OpaqueId::parse(format!("{prefix}{:032x}", nonce())).unwrap()
+}
+
+fn nonce() -> u128 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_nanos()
+}
+
+fn claimed(outcome: ClaimOutcome) -> (u64, MutationDisposition, OpaqueId) {
+    let ClaimOutcome::Claimed {
+        authority,
+        disposition: Some(disposition),
+    } = outcome
+    else {
+        panic!("claim rejected")
+    };
+    assert_eq!(authority.state, AttemptState::Authorized);
+    (authority.fence, disposition, authority.attempt_id)
+}
+
+fn owner(authority: &remora_attempt::AttemptAuthority) -> LeaseOwner {
+    LeaseOwner::new(
+        authority.attempt_id.as_str(),
+        authority.fence,
+        authority.worker_incarnation.as_str(),
+    )
+    .unwrap()
+}
+
+struct PanicOnce {
+    armed: Mutex<HashSet<AttemptFaultPoint>>,
+}
+
+impl PanicOnce {
+    fn new(point: AttemptFaultPoint) -> Self {
+        Self {
+            armed: Mutex::new(HashSet::from([point])),
+        }
+    }
+}
+
+impl AttemptFaultInjector for PanicOnce {
+    fn hit(&self, point: AttemptFaultPoint) {
+        if self.armed.lock().unwrap().remove(&point) {
+            panic!("injected crash at {point:?}");
+        }
+    }
+}
+
+async fn crash_claim(
+    service: AttemptService,
+    job: OpaqueId,
+    worker: OpaqueId,
+    operation: OpaqueId,
+) {
+    assert!(
+        tokio::spawn(async move { service.claim(&job, &worker, &operation).await })
+            .await
+            .expect_err("claim must crash")
+            .is_panic()
+    );
+}
+
+async fn crash_activate(
+    service: AttemptService,
+    authority: remora_attempt::AttemptAuthority,
+    operation: OpaqueId,
+) {
+    assert!(
+        tokio::spawn(async move { service.activate(&authority, &operation).await })
+            .await
+            .expect_err("activation must crash")
+            .is_panic()
+    );
+}
+
+async fn crash_renew(service: AttemptService, authority: remora_attempt::AttemptAuthority) {
+    assert!(
+        tokio::spawn(async move { service.renew(&authority).await })
+            .await
+            .expect_err("renewal must crash")
+            .is_panic()
+    );
+}
+
+async fn crash_recovery(service: AttemptService, authority: remora_attempt::AttemptAuthority) {
+    assert!(
+        tokio::spawn(async move { service.recover_inactive(&authority).await })
+            .await
+            .expect_err("recovery must crash")
+            .is_panic()
+    );
+}

tests/kind/attempt-job.yaml.in

@@ -0,0 +1,34 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: @@JOB_NAME@@
+  namespace: remora-testbed
+spec:
+  activeDeadlineSeconds: 600
+  backoffLimit: 0
+  ttlSecondsAfterFinished: 300
+  template:
+    spec:
+      automountServiceAccountToken: false
+      enableServiceLinks: false
+      restartPolicy: Never
+      containers:
+        - name: tests
+          image: @@TEST_IMAGE@@
+          imagePullPolicy: Never
+          envFrom:
+            - configMapRef: {name: @@PROFILE_NAME@@}
+          env:
+            - name: REMORA_ATTEMPT_REQUIRE
+              value: "1"
+          volumeMounts:
+            - {name: fixture-profile, mountPath: /run/remora-fixture, readOnly: true}
+          resources:
+            requests: {cpu: 250m, memory: 256Mi}
+            limits: {cpu: "2", memory: 1Gi}
+      volumes:
+        - name: fixture-profile
+          configMap:
+            name: @@PROFILE_NAME@@-file
+            items:
+              - {key: profile.env, path: profile.env}

tests/kind/run-attempt-job.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+source "$ROOT/tests/kind/lib.sh"
+
+USER_KUBECONFIG_BEFORE="$(user_kubeconfig_digest)"
+SUCCEEDED=false
+OWNS_FIXTURE=false
+
+cleanup() {
+  local status=$?
+  trap - EXIT INT TERM
+  if [[ "${REMORA_TESTBED_RETAIN:-0}" = 1 ]] && [[ "$SUCCEEDED" != true ]]; then
+    printf 'retained kubeconfig: %s\n' "$KUBECONFIG_PATH"
+    printf 'retained context: %s\n' "$CONTEXT"
+    exit "$status"
+  fi
+  if [[ "$OWNS_FIXTURE" = true ]]; then
+    destroy_fixture || status=1
+  fi
+  if [[ "$(user_kubeconfig_digest)" != "$USER_KUBECONFIG_BEFORE" ]]; then
+    printf 'user kubeconfig changed unexpectedly\n' >&2
+    status=1
+  fi
+  exit "$status"
+}
+
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+acquire_fixture_lock
+REMORA_REAL_NERDCTL="$(command -v nerdctl)"
+export REMORA_REAL_NERDCTL
+"$REMORA_REAL_NERDCTL" --namespace default build \
+  --file "$ROOT/Dockerfile.test" \
+  --target kind-attempt-runner \
+  --tag remora-kind-attempt:local \
+  "$ROOT"
+OWNS_FIXTURE=true
+setup_fixture durable coordination
+load_local_image remora-kind-attempt:local
+run_job_matrix \
+  attempt-authority \
+  "$TESTBED/attempt-job.yaml.in" \
+  remora-kind-attempt:local \
+  durable coordination
+SUCCEEDED=true
+printf 'attempt authority package test Jobs passed\n'