Add fenced attempt mutation and outcome publication

crates/remora-attempt/Cargo.toml

@@ -10,6 +10,7 @@ getrandom = "0.3.4"
 remora-coordination = { path = "../remora-coordination" }
 remora-domain.workspace = true
 remora-durable.workspace = true
+remora-result = { path = "../remora-result" }
 serde.workspace = true
 serde_json.workspace = true
 sqlx.workspace = true
@@ -21,4 +22,5 @@ remora-admission = { path = "../remora-admission" }
 remora-contracts = { path = "../remora-contracts" }
 remora-credentials = { path = "../remora-credentials" }
 remora-identity = { path = "../remora-identity" }
+remora-target = { path = "../remora-target", features = ["test-utils"] }
 tokio.workspace = true

crates/remora-attempt/migrations/0002_attempt_mutations.sql

@@ -0,0 +1,35 @@
+ALTER TABLE attempts
+    DROP CONSTRAINT attempts_state_check,
+    DROP CONSTRAINT attempts_state_timestamps_check,
+    ADD COLUMN stopped_at TIMESTAMPTZ,
+    ADD COLUMN committed_at TIMESTAMPTZ,
+    ADD COLUMN failure JSONB,
+    ADD CONSTRAINT attempts_state_check CHECK (
+        state IN ('authorized', 'executing', 'reaped', 'stopped', 'committed')
+    ),
+    ADD CONSTRAINT attempts_state_timestamps_check CHECK (
+        (state = 'authorized' AND executing_at IS NULL AND reaped_at IS NULL
+            AND stopped_at IS NULL AND committed_at IS NULL)
+        OR (state = 'executing' AND executing_at IS NOT NULL AND reaped_at IS NULL
+            AND stopped_at IS NULL AND committed_at IS NULL)
+        OR (state = 'reaped' AND executing_at IS NULL AND reaped_at IS NOT NULL
+            AND stopped_at IS NULL AND committed_at IS NULL)
+        OR (state = 'stopped' AND executing_at IS NOT NULL AND reaped_at IS NULL
+            AND stopped_at IS NOT NULL AND committed_at IS NULL)
+        OR (state = 'committed' AND executing_at IS NOT NULL AND reaped_at IS NULL
+            AND stopped_at IS NULL AND committed_at IS NOT NULL)
+    );
+
+ALTER TABLE jobs
+    ADD COLUMN cancel_requested BOOLEAN NOT NULL DEFAULT FALSE,
+    ADD COLUMN resolved_source JSONB,
+    ADD COLUMN checkpoint JSONB,
+    ADD COLUMN committed_result JSONB,
+    ADD COLUMN failure JSONB,
+    ADD COLUMN terminal_at TIMESTAMPTZ,
+    ADD CONSTRAINT jobs_terminal_projection_check CHECK (
+        ((phase = 'succeeded') = (committed_result IS NOT NULL))
+        AND ((phase = 'failed') = (failure IS NOT NULL))
+        AND ((phase IN ('succeeded', 'failed', 'cancelled')) = (terminal_at IS NOT NULL))
+        AND (committed_result IS NULL OR failure IS NULL)
+    );

crates/remora-attempt/src/lib.rs

@@ -1,10 +1,17 @@
 mod error;
 mod migrate;
+mod mutation_types;
+mod mutations;
 mod service;
 mod types;
 
 pub use error::AttemptError;
 pub use migrate::migrate;
+pub use mutation_types::{
+    AttemptMutationOutcome, CheckpointProjection, CheckpointRecord, CommittedManifestRecord,
+    CommittedResultRecord, DurableDescriptor, FailureRecord, JobMutationProjection,
+    MutationRejection, ProgressRecord,
+};
 pub use service::{AttemptConfig, AttemptService};
 pub use types::{
     ActivationOutcome, AttemptAuthority, AttemptFaultInjector, AttemptFaultPoint, AttemptState,

crates/remora-attempt/src/mutation_types.rs

@@ -0,0 +1,130 @@
+use remora_domain::{FailureDisposition, JobPhase, OpaqueId, RuntimeFailureClass};
+use remora_durable::MutationDisposition;
+use remora_result::PublicationInput;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct DurableDescriptor {
+    pub id: OpaqueId,
+    pub schema_version: u32,
+    pub byte_length: u64,
+    pub item_count: Option<u64>,
+    pub checksum: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct CheckpointRecord {
+    pub descriptor: DurableDescriptor,
+    pub resolved_source_id: OpaqueId,
+    pub validated_bytes: u64,
+    pub validated_members: u64,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "state", rename_all = "snake_case")]
+pub enum CheckpointProjection {
+    Reusable {
+        checkpoint: CheckpointRecord,
+    },
+    Complete {
+        validated_bytes: u64,
+        validated_members: u64,
+    },
+}
+
+impl CheckpointProjection {
+    pub(crate) fn reusable(&self) -> Option<&CheckpointRecord> {
+        match self {
+            Self::Reusable { checkpoint } => Some(checkpoint),
+            Self::Complete { .. } => None,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ProgressRecord {
+    pub completed_bytes: u64,
+    pub total_bytes: Option<u64>,
+    pub completed_members: u64,
+    pub total_members: Option<u64>,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MutationRejection {
+    AuthorityLost,
+    RequiredState,
+    CancellationRequested,
+    ResolvedSourceConflict,
+    ResolvedSourceRequired,
+    CheckpointSourceMismatch,
+    CheckpointRegression,
+    ProgressRegression,
+    InvalidResult,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum AttemptMutationOutcome {
+    Accepted(MutationDisposition),
+    Rejected {
+        reason: MutationRejection,
+        disposition: MutationDisposition,
+    },
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct JobMutationProjection {
+    pub phase: JobPhase,
+    pub revision: u64,
+    pub authority_attempt_id: Option<OpaqueId>,
+    pub resolved_source: Option<DurableDescriptor>,
+    pub checkpoint: Option<CheckpointProjection>,
+    pub committed_result: Option<CommittedResultRecord>,
+    pub failure: Option<FailureRecord>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct CommittedResultRecord {
+    pub attempt_id: OpaqueId,
+    pub manifest_id: OpaqueId,
+    pub schema_version: u32,
+    pub manifest: CommittedManifestRecord,
+    pub member_count: u64,
+    pub total_bytes: u64,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct CommittedManifestRecord {
+    pub store: String,
+    pub bucket: String,
+    pub key: String,
+    pub version_id: Option<String>,
+    pub sha256: String,
+    pub size: u64,
+}
+
+impl From<&PublicationInput> for CommittedResultRecord {
+    fn from(value: &PublicationInput) -> Self {
+        Self {
+            attempt_id: value.attempt_id.clone(),
+            manifest_id: value.manifest_id.clone(),
+            schema_version: value.schema_version,
+            manifest: CommittedManifestRecord {
+                store: value.manifest.identity.target_store_id.clone(),
+                bucket: value.manifest.identity.bucket.clone(),
+                key: value.manifest.identity.key.clone(),
+                version_id: value.manifest.identity.version_id.clone(),
+                sha256: value.manifest.content_sha256.as_str().into(),
+                size: value.manifest.size,
+            },
+            member_count: value.member_count,
+            total_bytes: value.total_bytes,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct FailureRecord {
+    pub class: RuntimeFailureClass,
+    pub disposition: FailureDisposition,
+}

crates/remora-attempt/src/mutations.rs

@@ -0,0 +1,637 @@
+use remora_coordination::{LeaseOwner, MAX_SAFE_INTEGER, OpCode, ProgressValue};
+use remora_domain::{
+    FailureDisposition, JobPhase, OpaqueId, RuntimeFailureClass, fingerprint, ports::MutationScope,
+};
+use remora_durable::{
+    DurableError, MutationDisposition, MutationOutcome, MutationStatus, enqueue, execute_mutation,
+};
+use remora_result::PublicationInput;
+use serde::Serialize;
+use serde_json::{Value, json};
+use sqlx::types::Json;
+use time::OffsetDateTime;
+
+use crate::{
+    AttemptAuthority, AttemptError, AttemptMutationOutcome, AttemptService, CheckpointProjection,
+    CheckpointRecord, CommittedResultRecord, DurableDescriptor, FailureRecord,
+    JobMutationProjection, MutationRejection, ProgressRecord,
+};
+
+impl AttemptService {
+    pub async fn pin_resolved_source(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        descriptor: DurableDescriptor,
+    ) -> Result<AttemptMutationOutcome, AttemptError> {
+        validate_descriptor(&descriptor)?;
+        let fingerprint_input = descriptor.clone();
+        self.execute_durable_mutation(
+            authority,
+            operation_id,
+            "pin_resolved_source",
+            &fingerprint_input,
+            move |tx, authority| {
+                Box::pin(async move {
+                    let Some(row) = lock_executing(tx, &authority).await? else {
+                        return rejected(MutationRejection::AuthorityLost);
+                    };
+                    if row.cancel_requested {
+                        return rejected(MutationRejection::CancellationRequested);
+                    }
+                    if let Some(current) = row.resolved_source {
+                        return if current == descriptor {
+                            accepted()
+                        } else {
+                            rejected(MutationRejection::ResolvedSourceConflict)
+                        };
+                    }
+                    let now = OffsetDateTime::now_utc();
+                    sqlx::query(
+                        "UPDATE jobs SET resolved_source = $2, revision = revision + 1,
+                            updated_at = $3 WHERE job_id = $1",
+                    )
+                    .bind(authority.job_id.as_str())
+                    .bind(Json(&descriptor))
+                    .bind(now)
+                    .execute(tx.as_mut())
+                    .await?;
+                    enqueue(tx, "job.updated", json!({ "job_id": authority.job_id })).await?;
+                    accepted()
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn persist_checkpoint(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        checkpoint: CheckpointRecord,
+    ) -> Result<AttemptMutationOutcome, AttemptError> {
+        validate_checkpoint(&checkpoint)?;
+        let fingerprint_input = checkpoint.clone();
+        self.execute_durable_mutation(
+            authority,
+            operation_id,
+            "persist_checkpoint",
+            &fingerprint_input,
+            move |tx, authority| {
+                Box::pin(async move {
+                    let Some(row) = lock_executing(tx, &authority).await? else {
+                        return rejected(MutationRejection::AuthorityLost);
+                    };
+                    if row.cancel_requested {
+                        return rejected(MutationRejection::CancellationRequested);
+                    }
+                    let Some(resolved_source) = row.resolved_source.as_ref() else {
+                        return rejected(MutationRejection::ResolvedSourceRequired);
+                    };
+                    if checkpoint.resolved_source_id != resolved_source.id {
+                        return rejected(MutationRejection::CheckpointSourceMismatch);
+                    }
+                    let current = row
+                        .checkpoint
+                        .as_ref()
+                        .and_then(CheckpointProjection::reusable);
+                    if current.is_some_and(|current| {
+                        checkpoint.validated_bytes < current.validated_bytes
+                            || checkpoint.validated_members < current.validated_members
+                    }) {
+                        return rejected(MutationRejection::CheckpointRegression);
+                    }
+                    if current == Some(&checkpoint) {
+                        return accepted();
+                    }
+                    let now = OffsetDateTime::now_utc();
+                    sqlx::query(
+                        "UPDATE jobs SET checkpoint = $2, revision = revision + 1,
+                            updated_at = $3 WHERE job_id = $1",
+                    )
+                    .bind(authority.job_id.as_str())
+                    .bind(Json(&CheckpointProjection::Reusable {
+                        checkpoint: checkpoint.clone(),
+                    }))
+                    .bind(now)
+                    .execute(tx.as_mut())
+                    .await?;
+                    enqueue(tx, "job.updated", json!({ "job_id": authority.job_id })).await?;
+                    accepted()
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn report_progress(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        progress: ProgressRecord,
+    ) -> Result<AttemptMutationOutcome, AttemptError> {
+        validate_progress(progress)?;
+        let coordination = self.coordination.clone();
+        self.execute_durable_mutation(
+            authority,
+            operation_id,
+            "report_progress",
+            &progress,
+            move |tx, authority| {
+                Box::pin(async move {
+                    let Some(row) = lock_executing(tx, &authority).await? else {
+                        return rejected(MutationRejection::AuthorityLost);
+                    };
+                    if row.cancel_requested {
+                        return rejected(MutationRejection::CancellationRequested);
+                    }
+                    let owner = lease_owner(&authority).map_err(DurableError::effect)?;
+                    let result = coordination
+                        .report_progress_value(
+                            authority.job_id.as_str(),
+                            &owner,
+                            ProgressValue {
+                                completed_bytes: progress.completed_bytes,
+                                total_bytes: progress.total_bytes,
+                                completed_members: progress.completed_members,
+                                total_members: progress.total_members,
+                            },
+                        )
+                        .await
+                        .map_err(DurableError::effect)?;
+                    match result.code {
+                        OpCode::Ok | OpCode::OkIdempotent => accepted(),
+                        OpCode::FailStale => rejected(MutationRejection::ProgressRegression),
+                        _ => rejected(MutationRejection::AuthorityLost),
+                    }
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn publish_success(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        result: PublicationInput,
+    ) -> Result<AttemptMutationOutcome, AttemptError> {
+        let request = PublicationFingerprint::from(&result);
+        self.execute_durable_mutation(
+            authority,
+            operation_id,
+            "publish_success",
+            &request,
+            move |tx, authority| {
+                Box::pin(async move {
+                    let Some(row) = lock_executing(tx, &authority).await? else {
+                        return rejected(MutationRejection::AuthorityLost);
+                    };
+                    if row.cancel_requested {
+                        return rejected(MutationRejection::CancellationRequested);
+                    }
+                    if result.attempt_id != authority.attempt_id
+                        || result.manifest.identity.target_store_id != row.target_store
+                        || result.manifest.identity.bucket != row.target_bucket
+                        || !sealed_result_matches(tx, &result).await?
+                    {
+                        return rejected(MutationRejection::InvalidResult);
+                    }
+                    let now = OffsetDateTime::now_utc();
+                    let stored = publication_json(&result);
+                    let complete_checkpoint = CheckpointProjection::Complete {
+                        validated_bytes: result.total_bytes,
+                        validated_members: result.member_count,
+                    };
+                    let changed = sqlx::query(
+                        "UPDATE jobs SET phase = 'succeeded', authority_attempt_id = NULL,
+                            checkpoint = $2, committed_result = $3, terminal_at = $4,
+                            revision = revision + 1, updated_at = $4
+                         WHERE job_id = $1 AND phase = 'running' AND cancel_requested = FALSE",
+                    )
+                    .bind(authority.job_id.as_str())
+                    .bind(Json(&complete_checkpoint))
+                    .bind(stored)
+                    .bind(now)
+                    .execute(tx.as_mut())
+                    .await?;
+                    if changed.rows_affected() != 1 {
+                        return rejected(MutationRejection::RequiredState);
+                    }
+                    sqlx::query(
+                        "UPDATE attempts SET state = 'committed', committed_at = $2
+                         WHERE attempt_id = $1 AND state = 'executing'",
+                    )
+                    .bind(authority.attempt_id.as_str())
+                    .bind(now)
+                    .execute(tx.as_mut())
+                    .await?;
+                    enqueue(tx, "job.updated", json!({ "job_id": authority.job_id })).await?;
+                    accepted()
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn publish_failure(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        class: RuntimeFailureClass,
+    ) -> Result<AttemptMutationOutcome, AttemptError> {
+        let max_attempts = self.config.max_attempts;
+        self.execute_durable_mutation(
+            authority,
+            operation_id,
+            "publish_failure",
+            &class,
+            move |tx, authority| {
+                Box::pin(async move {
+                    let Some(row) = lock_executing(tx, &authority).await? else {
+                        return rejected(MutationRejection::AuthorityLost);
+                    };
+                    if row.cancel_requested {
+                        return rejected(MutationRejection::CancellationRequested);
+                    }
+                    let disposition = class.disposition();
+                    let retry = disposition == FailureDisposition::Retryable
+                        && authority.fence < max_attempts;
+                    let failure = FailureRecord { class, disposition };
+                    let now = OffsetDateTime::now_utc();
+                    let changed = if retry {
+                        sqlx::query(
+                            "UPDATE jobs SET phase = 'retry_wait', authority_attempt_id = NULL,
+                                dispatch_needed = FALSE, revision = revision + 1, updated_at = $2
+                             WHERE job_id = $1 AND phase = 'running' AND cancel_requested = FALSE",
+                        )
+                        .bind(authority.job_id.as_str())
+                        .bind(now)
+                        .execute(tx.as_mut())
+                        .await?
+                    } else {
+                        sqlx::query(
+                            "UPDATE jobs SET phase = 'failed', authority_attempt_id = NULL,
+                                dispatch_needed = FALSE, failure = $2, terminal_at = $3,
+                                revision = revision + 1, updated_at = $3
+                             WHERE job_id = $1 AND phase = 'running' AND cancel_requested = FALSE",
+                        )
+                        .bind(authority.job_id.as_str())
+                        .bind(Json(&failure))
+                        .bind(now)
+                        .execute(tx.as_mut())
+                        .await?
+                    };
+                    if changed.rows_affected() != 1 {
+                        return rejected(MutationRejection::RequiredState);
+                    }
+                    sqlx::query(
+                        "UPDATE attempts SET state = 'stopped', stopped_at = $2, failure = $3
+                         WHERE attempt_id = $1 AND state = 'executing'",
+                    )
+                    .bind(authority.attempt_id.as_str())
+                    .bind(now)
+                    .bind(Json(&failure))
+                    .execute(tx.as_mut())
+                    .await?;
+                    enqueue(
+                        tx,
+                        if retry {
+                            "job.retry_wait"
+                        } else {
+                            "job.updated"
+                        },
+                        json!({ "job_id": authority.job_id }),
+                    )
+                    .await?;
+                    accepted()
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn mutation_projection(
+        &self,
+        job_id: &OpaqueId,
+    ) -> Result<Option<JobMutationProjection>, AttemptError> {
+        let row = sqlx::query_as::<_, ProjectionRow>(
+            "SELECT phase, revision, authority_attempt_id, resolved_source, checkpoint,
+                    committed_result, failure FROM jobs WHERE job_id = $1",
+        )
+        .bind(job_id.as_str())
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(DurableError::from)?;
+        row.map(ProjectionRow::try_into_projection).transpose()
+    }
+
+    async fn execute_durable_mutation<T, D>(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        kind: &'static str,
+        request: &T,
+        decide: D,
+    ) -> Result<AttemptMutationOutcome, AttemptError>
+    where
+        T: Serialize,
+        D: for<'c> FnOnce(
+                &'c mut remora_durable::PgTx<'_>,
+                AttemptAuthority,
+            ) -> remora_durable::BoxFuture<
+                'c,
+                Result<(MutationStatus, Value), DurableError>,
+            > + Send
+            + 'static,
+    {
+        validate_authority(authority)?;
+        let request_fingerprint =
+            fingerprint(request).map_err(|error| AttemptError::Invalid(error.to_string()))?;
+        let eligibility = authority_binding(authority);
+        let authority = authority.clone();
+        let outcome = execute_mutation(
+            &self.pool,
+            MutationScope {
+                subject: &self.operation_subject,
+                operation: "attempt.mutation",
+                key: operation_id.as_str(),
+            },
+            kind,
+            &eligibility,
+            request_fingerprint,
+            move |tx| decide(tx, authority),
+        )
+        .await?;
+        mutation_outcome(outcome)
+    }
+}
+
+struct LockedAttempt {
+    cancel_requested: bool,
+    target_store: String,
+    target_bucket: String,
+    resolved_source: Option<DurableDescriptor>,
+    checkpoint: Option<CheckpointProjection>,
+}
+
+async fn lock_executing(
+    tx: &mut remora_durable::PgTx<'_>,
+    authority: &AttemptAuthority,
+) -> Result<Option<LockedAttempt>, DurableError> {
+    let row = sqlx::query_as::<
+        _,
+        (
+            bool,
+            String,
+            String,
+            Option<Json<DurableDescriptor>>,
+            Option<Json<CheckpointProjection>>,
+        ),
+    >(
+        "SELECT j.cancel_requested, j.target_store, j.target_bucket,
+                j.resolved_source, j.checkpoint
+         FROM jobs j JOIN attempts a ON a.attempt_id = j.authority_attempt_id
+         WHERE j.job_id = $1 AND j.phase = 'running'
+           AND j.current_fence = $2 AND j.authority_attempt_id = $3
+           AND a.worker_incarnation = $4 AND a.state = 'executing'
+         FOR UPDATE OF j, a",
+    )
+    .bind(authority.job_id.as_str())
+    .bind(to_i64(authority.fence)?)
+    .bind(authority.attempt_id.as_str())
+    .bind(authority.worker_incarnation.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?;
+    Ok(row.map(
+        |(cancel_requested, target_store, target_bucket, resolved_source, checkpoint)| {
+            LockedAttempt {
+                cancel_requested,
+                target_store,
+                target_bucket,
+                resolved_source: resolved_source.map(|value| value.0),
+                checkpoint: checkpoint.map(|value| value.0),
+            }
+        },
+    ))
+}
+
+fn accepted() -> Result<(MutationStatus, Value), DurableError> {
+    Ok((MutationStatus::Accepted, json!({ "accepted": true })))
+}
+
+fn rejected(reason: MutationRejection) -> Result<(MutationStatus, Value), DurableError> {
+    Ok((MutationStatus::Rejected, json!({ "reason": reason })))
+}
+
+fn mutation_outcome(outcome: MutationOutcome) -> Result<AttemptMutationOutcome, AttemptError> {
+    if outcome.disposition == MutationDisposition::Conflict {
+        return Err(AttemptError::OperationConflict);
+    }
+    if outcome.status == MutationStatus::Accepted {
+        return Ok(AttemptMutationOutcome::Accepted(outcome.disposition));
+    }
+    let reason = serde_json::from_value(
+        outcome
+            .response
+            .get("reason")
+            .cloned()
+            .ok_or_else(|| AttemptError::Invalid("missing mutation rejection".into()))?,
+    )
+    .map_err(|error| AttemptError::Invalid(error.to_string()))?;
+    Ok(AttemptMutationOutcome::Rejected {
+        reason,
+        disposition: outcome.disposition,
+    })
+}
+
+fn validate_descriptor(descriptor: &DurableDescriptor) -> Result<(), AttemptError> {
+    if descriptor.schema_version == 0
+        || u64::from(descriptor.schema_version) > MAX_SAFE_INTEGER
+        || descriptor.byte_length > MAX_SAFE_INTEGER
+        || descriptor
+            .item_count
+            .is_some_and(|value| value > MAX_SAFE_INTEGER)
+        || descriptor.checksum.len() != 71
+        || !descriptor.checksum.starts_with("sha256:")
+        || !descriptor.checksum.as_bytes()[7..]
+            .iter()
+            .copied()
+            .all(|value| value.is_ascii_digit() || (b'a'..=b'f').contains(&value))
+    {
+        return Err(AttemptError::Invalid(
+            "descriptor metadata is not canonical".into(),
+        ));
+    }
+    Ok(())
+}
+
+fn validate_checkpoint(checkpoint: &CheckpointRecord) -> Result<(), AttemptError> {
+    validate_descriptor(&checkpoint.descriptor)?;
+    if (checkpoint.validated_bytes == 0 && checkpoint.validated_members == 0)
+        || checkpoint.validated_bytes > MAX_SAFE_INTEGER
+        || checkpoint.validated_members > MAX_SAFE_INTEGER
+    {
+        return Err(AttemptError::Invalid(
+            "checkpoint exceeds the safe-integer range".into(),
+        ));
+    }
+    Ok(())
+}
+
+fn validate_progress(progress: ProgressRecord) -> Result<(), AttemptError> {
+    ProgressValue {
+        completed_bytes: progress.completed_bytes,
+        total_bytes: progress.total_bytes,
+        completed_members: progress.completed_members,
+        total_members: progress.total_members,
+    }
+    .validate()?;
+    Ok(())
+}
+
+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 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 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 to_i64(value: u64) -> Result<i64, DurableError> {
+    i64::try_from(value).map_err(DurableError::effect)
+}
+
+#[derive(Serialize)]
+struct PublicationFingerprint {
+    attempt_id: OpaqueId,
+    manifest_id: OpaqueId,
+    schema_version: u32,
+    store: String,
+    bucket: String,
+    key: String,
+    version_id: Option<String>,
+    sha256: String,
+    size: u64,
+    member_count: u64,
+    total_bytes: u64,
+}
+
+impl From<&PublicationInput> for PublicationFingerprint {
+    fn from(value: &PublicationInput) -> Self {
+        Self {
+            attempt_id: value.attempt_id.clone(),
+            manifest_id: value.manifest_id.clone(),
+            schema_version: value.schema_version,
+            store: value.manifest.identity.target_store_id.clone(),
+            bucket: value.manifest.identity.bucket.clone(),
+            key: value.manifest.identity.key.clone(),
+            version_id: value.manifest.identity.version_id.clone(),
+            sha256: value.manifest.content_sha256.as_str().into(),
+            size: value.manifest.size,
+            member_count: value.member_count,
+            total_bytes: value.total_bytes,
+        }
+    }
+}
+
+async fn sealed_result_matches(
+    tx: &mut remora_durable::PgTx<'_>,
+    result: &PublicationInput,
+) -> Result<bool, DurableError> {
+    let matches: bool = sqlx::query_scalar(
+        "SELECT EXISTS (
+            SELECT 1 FROM result_inventories
+            WHERE attempt_id = $1 AND state = 'sealed'
+              AND manifest_schema_version = $2 AND manifest_id = $3
+              AND manifest_store = $4 AND manifest_bucket = $5 AND manifest_key = $6
+              AND manifest_version_id IS NOT DISTINCT FROM $7
+              AND manifest_sha256 = $8 AND manifest_size = $9
+              AND member_count = $10 AND total_bytes = $11
+         )",
+    )
+    .bind(result.attempt_id.as_str())
+    .bind(i32::try_from(result.schema_version).map_err(DurableError::effect)?)
+    .bind(result.manifest_id.as_str())
+    .bind(&result.manifest.identity.target_store_id)
+    .bind(&result.manifest.identity.bucket)
+    .bind(&result.manifest.identity.key)
+    .bind(&result.manifest.identity.version_id)
+    .bind(result.manifest.content_sha256.as_str())
+    .bind(to_i64(result.manifest.size)?)
+    .bind(to_i64(result.member_count)?)
+    .bind(to_i64(result.total_bytes)?)
+    .fetch_one(tx.as_mut())
+    .await?;
+    Ok(matches)
+}
+
+fn publication_json(result: &PublicationInput) -> Value {
+    serde_json::to_value(CommittedResultRecord::from(result))
+        .expect("committed result serialization cannot fail")
+}
+
+#[derive(sqlx::FromRow)]
+struct ProjectionRow {
+    phase: String,
+    revision: i64,
+    authority_attempt_id: Option<String>,
+    resolved_source: Option<Json<DurableDescriptor>>,
+    checkpoint: Option<Json<CheckpointProjection>>,
+    committed_result: Option<Value>,
+    failure: Option<Json<FailureRecord>>,
+}
+
+impl ProjectionRow {
+    fn try_into_projection(self) -> Result<JobMutationProjection, AttemptError> {
+        Ok(JobMutationProjection {
+            phase: parse_phase(&self.phase)?,
+            revision: u64::try_from(self.revision)
+                .map_err(|_| AttemptError::Invalid("negative job revision".into()))?,
+            authority_attempt_id: self
+                .authority_attempt_id
+                .map(OpaqueId::parse)
+                .transpose()
+                .map_err(|error| AttemptError::Invalid(error.to_string()))?,
+            resolved_source: self.resolved_source.map(|value| value.0),
+            checkpoint: self.checkpoint.map(|value| value.0),
+            committed_result: self
+                .committed_result
+                .map(serde_json::from_value)
+                .transpose()
+                .map_err(|error| AttemptError::Invalid(error.to_string()))?,
+            failure: self.failure.map(|value| value.0),
+        })
+    }
+}
+
+fn parse_phase(value: &str) -> Result<JobPhase, AttemptError> {
+    match value {
+        "queued" => Ok(JobPhase::Queued),
+        "running" => Ok(JobPhase::Running),
+        "retry_wait" => Ok(JobPhase::RetryWait),
+        "cancelling" => Ok(JobPhase::Cancelling),
+        "succeeded" => Ok(JobPhase::Succeeded),
+        "failed" => Ok(JobPhase::Failed),
+        "cancelled" => Ok(JobPhase::Cancelled),
+        _ => Err(AttemptError::Invalid(format!(
+            "unknown job phase '{value}'"
+        ))),
+    }
+}

crates/remora-attempt/src/service.rs

@@ -10,12 +10,13 @@ use remora_durable::{
 };
 use serde::Serialize;
 use serde_json::json;
-use sqlx::PgPool;
+use sqlx::{PgPool, types::Json};
 use time::OffsetDateTime;
 
 use crate::{
     ActivationOutcome, AttemptAuthority, AttemptError, AttemptFaultInjector, AttemptFaultPoint,
-    AttemptState, ClaimOutcome, ClaimRejection, RecoveryOutcome, RenewalOutcome,
+    AttemptState, CheckpointProjection, ClaimOutcome, ClaimRejection, RecoveryOutcome,
+    RenewalOutcome,
 };
 
 struct NoFaults;
@@ -41,12 +42,12 @@ impl Default for AttemptConfig {
 
 #[derive(Clone)]
 pub struct AttemptService {
-    pool: PgPool,
-    coordination: CoordinationStore,
+    pub(crate) pool: PgPool,
+    pub(crate) coordination: CoordinationStore,
     local_permits: LocalPermitGate,
-    operation_subject: OpaqueId,
+    pub(crate) operation_subject: OpaqueId,
     faults: Arc<dyn AttemptFaultInjector>,
-    config: AttemptConfig,
+    pub(crate) config: AttemptConfig,
 }
 
 impl AttemptService {
@@ -477,14 +478,26 @@ 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
+    let row = sqlx::query_as::<
+        _,
+        (
+            String,
+            bool,
+            i64,
+            Option<String>,
+            Option<Json<crate::DurableDescriptor>>,
+            Option<Json<CheckpointProjection>>,
+        ),
+    >(
+        "SELECT phase, dispatch_needed, current_fence, authority_attempt_id,
+                resolved_source, checkpoint
          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 {
+    let Some((phase, dispatch_needed, current_fence, authority, resolved_source, checkpoint)) = row
+    else {
         return Ok(claim_rejected(ClaimRejection::JobUnavailable));
     };
     if phase != "queued" || !dispatch_needed || authority.is_some() {
@@ -509,17 +522,29 @@ async fn authorize_attempt(
     input.reserved.store(true, Ordering::Release);
     input.faults.hit(AttemptFaultPoint::ClaimAfterReserve);
     let fence = current_fence + 1;
+    let initial_progress = match (resolved_source, checkpoint) {
+        (Some(source), Some(checkpoint))
+            if checkpoint
+                .0
+                .reusable()
+                .is_some_and(|value| value.resolved_source_id == source.0.id) =>
+        {
+            checkpoint.0.reusable().unwrap().validated_bytes
+        }
+        _ => 0,
+    };
     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)",
+         ) VALUES ($1, $2, $3, $4, 'authorized', $5, $6, 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(i64::try_from(initial_progress).map_err(DurableError::effect)?)
     .bind(now)
     .execute(tx.as_mut())
     .await?;

crates/remora-attempt/src/types.rs

@@ -27,6 +27,8 @@ pub enum AttemptState {
     Authorized,
     Executing,
     Reaped,
+    Stopped,
+    Committed,
 }
 
 impl AttemptState {
@@ -35,6 +37,8 @@ impl AttemptState {
             "authorized" => Self::Authorized,
             "executing" => Self::Executing,
             "reaped" => Self::Reaped,
+            "stopped" => Self::Stopped,
+            "committed" => Self::Committed,
             _ => return None,
         })
     }

crates/remora-attempt/tests/mutations.rs

@@ -0,0 +1,800 @@
+use std::time::Duration;
+
+use remora_admission::{
+    AdmissionConfig, AdmissionDecision, AdmissionService, migrate as admission_migrate,
+};
+use remora_attempt::{
+    ActivationOutcome, AttemptConfig, AttemptMutationOutcome, AttemptService, AttemptState,
+    CheckpointProjection, CheckpointRecord, CommittedResultRecord, DurableDescriptor,
+    FailureRecord, MutationRejection, ProgressRecord, migrate as attempt_migrate,
+};
+use remora_contracts::{HttpSourceSubmissionDto, JobSubmissionDto, SourceSubmissionDto};
+use remora_coordination::{CoordinationStore, StoreConfig};
+use remora_credentials::migrate as credentials_migrate;
+use remora_domain::{
+    BucketName, FailureDisposition, JobPhase, OpaqueId, RuntimeFailureClass, StorageScope,
+    TargetPrefix, TargetStoreId,
+};
+use remora_durable::{MutationDisposition, PoolConfig, connect, migrate as durable_migrate};
+use remora_identity::{DefaultTarget, Principal};
+use remora_result::{PublicationInput, migrate as result_migrate};
+use remora_target::{ContentSha256, ObjectIdentity, StoredObject, attempt_manifest_key};
+use sqlx::PgPool;
+
+struct Harness {
+    pool: PgPool,
+    coordination: CoordinationStore,
+}
+
+impl Harness {
+    async fn new(name: &str) -> 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: 16,
+                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");
+        result_migrate(&pool).await.expect("result migrate");
+        let coordination = CoordinationStore::connect(
+            &coordination_url,
+            format!("{}wp16:{name}:{}:", coordination_namespace(), nonce()),
+            StoreConfig {
+                global_limit: 16,
+                lease_ttl: Duration::from_secs(10),
+                reserve_ttl: Duration::from_secs(10),
+                progress_ttl: Duration::from_secs(10),
+                cancel_ttl: None,
+            },
+        )
+        .await
+        .expect("connect coordination store");
+        Some(Self { pool, coordination })
+    }
+
+    fn service(&self, max_attempts: u64) -> AttemptService {
+        AttemptService::new(
+            self.pool.clone(),
+            self.coordination.clone(),
+            AttemptConfig {
+                max_attempts,
+                local_limit: 8,
+            },
+        )
+        .expect("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: "mutation-tests-v1".into(),
+            },
+        )
+        .unwrap();
+        let outcome = admission
+            .submit(
+                &principal,
+                unique_id("submit").as_str(),
+                JobSubmissionDto {
+                    source: SourceSubmissionDto::Http(HttpSourceSubmissionDto {
+                        url: "https://example.invalid/artifact".into(),
+                        result_path: "artifact.bin".into(),
+                        credential_id: None,
+                    }),
+                    target: None,
+                    label: Some("mutation-test".into()),
+                },
+            )
+            .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
+    }
+
+    async fn executing(
+        &self,
+        service: &AttemptService,
+        job: &OpaqueId,
+    ) -> remora_attempt::AttemptAuthority {
+        let worker = unique_id("worker");
+        let claim = service
+            .claim(job, &worker, &unique_id("claimop"))
+            .await
+            .expect("claim");
+        let remora_attempt::ClaimOutcome::Claimed { authority, .. } = claim else {
+            panic!("claim rejected")
+        };
+        assert_eq!(authority.state, AttemptState::Authorized);
+        assert!(matches!(
+            service
+                .activate(&authority, &unique_id("activateop"))
+                .await
+                .expect("activate"),
+            ActivationOutcome::Activated(_)
+        ));
+        service
+            .authority(&authority.attempt_id)
+            .await
+            .unwrap()
+            .unwrap()
+    }
+
+    async fn sealed_result(
+        &self,
+        attempt_id: &OpaqueId,
+        member_count: u64,
+        total_bytes: u64,
+    ) -> PublicationInput {
+        let manifest_id = unique_id("manifest");
+        let sha256 = format!("sha256:{}", "11".repeat(32));
+        let key = attempt_manifest_key(attempt_id, &manifest_id);
+        let input = PublicationInput {
+            attempt_id: attempt_id.clone(),
+            manifest_id: manifest_id.clone(),
+            schema_version: 1,
+            manifest: StoredObject {
+                identity: ObjectIdentity {
+                    target_store_id: "primary".into(),
+                    bucket: "remora-results".into(),
+                    key: key.clone(),
+                    version_id: Some("version-1".into()),
+                },
+                content_sha256: ContentSha256::parse(&sha256).unwrap(),
+                size: 128,
+            },
+            member_count,
+            total_bytes,
+        };
+        sqlx::query(
+            "INSERT INTO result_inventories (
+                attempt_id, state, manifest_schema_version, manifest_id, manifest_store,
+                manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
+                manifest_size, member_count, total_bytes
+             ) VALUES ($1, 'sealed', 1, $2, 'primary', 'remora-results', $3,
+                       'version-1', $4, 128, $5, $6)",
+        )
+        .bind(attempt_id.as_str())
+        .bind(manifest_id.as_str())
+        .bind(key)
+        .bind(sha256)
+        .bind(i64::try_from(member_count).unwrap())
+        .bind(i64::try_from(total_bytes).unwrap())
+        .execute(&self.pool)
+        .await
+        .unwrap();
+        input
+    }
+}
+
+#[tokio::test]
+async fn resolved_source_checkpoint_and_progress_are_fenced_and_replayable() {
+    let Some(harness) = Harness::new("mutations").await else {
+        return;
+    };
+    let service = harness.service(3);
+    let job = harness.job().await;
+    let worker = unique_id("worker");
+    let claim = service
+        .claim(&job, &worker, &unique_id("claimop"))
+        .await
+        .unwrap();
+    let remora_attempt::ClaimOutcome::Claimed { authority, .. } = claim else {
+        panic!("claim rejected")
+    };
+    let resolved = descriptor("resolved", 4096, Some(3), 'a');
+    assert_rejected(
+        service
+            .pin_resolved_source(&authority, &unique_id("resolveop"), resolved.clone())
+            .await
+            .unwrap(),
+        MutationRejection::AuthorityLost,
+    );
+    service
+        .activate(&authority, &unique_id("activateop"))
+        .await
+        .unwrap();
+    let authority = service
+        .authority(&authority.attempt_id)
+        .await
+        .unwrap()
+        .unwrap();
+    let operation = unique_id("resolveop");
+    assert_eq!(
+        service
+            .pin_resolved_source(&authority, &operation, resolved.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::First)
+    );
+    assert_eq!(
+        service
+            .pin_resolved_source(&authority, &operation, resolved.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::Replay)
+    );
+    assert_eq!(
+        service
+            .pin_resolved_source(&authority, &unique_id("resolveop"), resolved.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::First)
+    );
+    assert_rejected(
+        service
+            .pin_resolved_source(
+                &authority,
+                &unique_id("resolveop"),
+                descriptor("resolved", 4096, Some(3), 'b'),
+            )
+            .await
+            .unwrap(),
+        MutationRejection::ResolvedSourceConflict,
+    );
+
+    assert!(
+        service
+            .persist_checkpoint(
+                &authority,
+                &unique_id("checkpointop"),
+                CheckpointRecord {
+                    descriptor: descriptor("checkpoint", 1, Some(0), '0'),
+                    resolved_source_id: resolved.id.clone(),
+                    validated_bytes: 0,
+                    validated_members: 0,
+                },
+            )
+            .await
+            .is_err(),
+        "an all-zero checkpoint does not refine PersistCheckpoint"
+    );
+
+    let checkpoint = CheckpointRecord {
+        descriptor: descriptor("checkpoint", 512, Some(1), 'c'),
+        resolved_source_id: resolved.id.clone(),
+        validated_bytes: 1024,
+        validated_members: 1,
+    };
+    let mut wrong_source = checkpoint.clone();
+    wrong_source.resolved_source_id = unique_id("resolved");
+    assert_rejected(
+        service
+            .persist_checkpoint(&authority, &unique_id("checkpointop"), wrong_source)
+            .await
+            .unwrap(),
+        MutationRejection::CheckpointSourceMismatch,
+    );
+    assert!(matches!(
+        service
+            .persist_checkpoint(&authority, &unique_id("checkpointop"), checkpoint.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(_)
+    ));
+    let mut regression = checkpoint.clone();
+    regression.validated_bytes = 1000;
+    assert_rejected(
+        service
+            .persist_checkpoint(&authority, &unique_id("checkpointop"), regression)
+            .await
+            .unwrap(),
+        MutationRejection::CheckpointRegression,
+    );
+
+    let progress = ProgressRecord {
+        completed_bytes: 2048,
+        total_bytes: Some(4096),
+        completed_members: 1,
+        total_members: Some(3),
+    };
+    let progress_operation = unique_id("progressop");
+    assert_eq!(
+        service
+            .report_progress(&authority, &progress_operation, progress)
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::First)
+    );
+    assert_eq!(
+        service
+            .report_progress(&authority, &progress_operation, progress)
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::Replay)
+    );
+    let hint = harness
+        .coordination
+        .get_progress(job.as_str())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(hint.bytes, 2048);
+    assert_eq!(hint.total_bytes, Some(4096));
+    assert_eq!(hint.completed_members, 1);
+    assert_eq!(hint.total_members, Some(3));
+    assert_rejected(
+        service
+            .report_progress(
+                &authority,
+                &unique_id("progressop"),
+                ProgressRecord {
+                    completed_bytes: 1024,
+                    ..progress
+                },
+            )
+            .await
+            .unwrap(),
+        MutationRejection::ProgressRegression,
+    );
+    assert!(
+        service
+            .persist_checkpoint(&authority, &progress_operation, checkpoint)
+            .await
+            .is_err(),
+        "an operation identity cannot change mutation kind"
+    );
+}
+
+#[tokio::test]
+async fn success_is_an_exact_fenced_commit_cas_and_terminal_state_is_immutable() {
+    let Some(harness) = Harness::new("success").await else {
+        return;
+    };
+    let service = harness.service(3);
+    let job = harness.job().await;
+    let authority = harness.executing(&service, &job).await;
+    let result = harness.sealed_result(&authority.attempt_id, 2, 8192).await;
+    let operation = unique_id("successop");
+    assert_eq!(
+        service
+            .publish_success(&authority, &operation, result.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::First)
+    );
+    assert_eq!(
+        service
+            .publish_success(&authority, &operation, result.clone())
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(MutationDisposition::Replay)
+    );
+    let mut changed = result.clone();
+    changed.total_bytes += 1;
+    assert!(
+        service
+            .publish_success(&authority, &operation, changed)
+            .await
+            .is_err()
+    );
+    assert_rejected(
+        service
+            .publish_failure(
+                &authority,
+                &unique_id("failureop"),
+                RuntimeFailureClass::Internal,
+            )
+            .await
+            .unwrap(),
+        MutationRejection::AuthorityLost,
+    );
+    assert_rejected(
+        service
+            .pin_resolved_source(
+                &authority,
+                &unique_id("resolveop"),
+                descriptor("resolved", 1, None, 'd'),
+            )
+            .await
+            .unwrap(),
+        MutationRejection::AuthorityLost,
+    );
+    let projection = service.mutation_projection(&job).await.unwrap().unwrap();
+    assert_eq!(projection.phase, JobPhase::Succeeded);
+    assert_eq!(projection.authority_attempt_id, None);
+    assert_eq!(
+        projection.checkpoint,
+        Some(CheckpointProjection::Complete {
+            validated_bytes: 8192,
+            validated_members: 2,
+        })
+    );
+    assert_eq!(
+        projection.committed_result,
+        Some(CommittedResultRecord::from(&result))
+    );
+    assert_eq!(projection.failure, None);
+    let state: String = sqlx::query_scalar("SELECT state FROM attempts WHERE attempt_id = $1")
+        .bind(authority.attempt_id.as_str())
+        .fetch_one(&harness.pool)
+        .await
+        .unwrap();
+    assert_eq!(state, "committed");
+}
+
+#[tokio::test]
+async fn invalid_result_stale_authority_and_cancellation_cannot_publish() {
+    let Some(harness) = Harness::new("rejections").await else {
+        return;
+    };
+    let service = harness.service(3);
+    let job = harness.job().await;
+    let authority = harness.executing(&service, &job).await;
+    let mut invalid = harness.sealed_result(&authority.attempt_id, 1, 64).await;
+    sqlx::query("UPDATE result_inventories SET manifest_store = 'other' WHERE attempt_id = $1")
+        .bind(authority.attempt_id.as_str())
+        .execute(&harness.pool)
+        .await
+        .unwrap();
+    invalid.manifest.identity.target_store_id = "other".into();
+    assert_rejected(
+        service
+            .publish_success(&authority, &unique_id("successop"), invalid.clone())
+            .await
+            .unwrap(),
+        MutationRejection::InvalidResult,
+    );
+    sqlx::query("UPDATE result_inventories SET manifest_store = 'primary' WHERE attempt_id = $1")
+        .bind(authority.attempt_id.as_str())
+        .execute(&harness.pool)
+        .await
+        .unwrap();
+    invalid.manifest.identity.target_store_id = "primary".into();
+    invalid.manifest.size += 1;
+    assert_rejected(
+        service
+            .publish_success(&authority, &unique_id("successop"), invalid)
+            .await
+            .unwrap(),
+        MutationRejection::InvalidResult,
+    );
+    sqlx::query("UPDATE jobs SET cancel_requested = TRUE WHERE job_id = $1")
+        .bind(job.as_str())
+        .execute(&harness.pool)
+        .await
+        .unwrap();
+    assert_rejected(
+        service
+            .publish_failure(
+                &authority,
+                &unique_id("failureop"),
+                RuntimeFailureClass::TransientSource,
+            )
+            .await
+            .unwrap(),
+        MutationRejection::CancellationRequested,
+    );
+    let valid = publication_from_database(&harness.pool, &authority.attempt_id).await;
+    assert_rejected(
+        service
+            .publish_success(&authority, &unique_id("successop"), valid)
+            .await
+            .unwrap(),
+        MutationRejection::CancellationRequested,
+    );
+    assert_eq!(
+        service
+            .mutation_projection(&job)
+            .await
+            .unwrap()
+            .unwrap()
+            .phase,
+        JobPhase::Running
+    );
+}
+
+#[tokio::test]
+async fn failure_disposition_retries_with_budget_and_becomes_terminal_when_exhausted() {
+    let Some(harness) = Harness::new("retry").await else {
+        return;
+    };
+    let service = harness.service(2);
+    let job = harness.job().await;
+    let first = harness.executing(&service, &job).await;
+    assert_eq!(first.fence, 1);
+    let retry_source = descriptor("resolved", 2048, Some(1), 'f');
+    let retry_checkpoint = CheckpointRecord {
+        descriptor: descriptor("checkpoint", 256, Some(1), 'e'),
+        resolved_source_id: retry_source.id.clone(),
+        validated_bytes: 512,
+        validated_members: 1,
+    };
+    assert_rejected(
+        service
+            .persist_checkpoint(&first, &unique_id("checkpointop"), retry_checkpoint.clone())
+            .await
+            .unwrap(),
+        MutationRejection::ResolvedSourceRequired,
+    );
+    service
+        .pin_resolved_source(&first, &unique_id("resolveop"), retry_source)
+        .await
+        .unwrap();
+    service
+        .persist_checkpoint(&first, &unique_id("checkpointop"), retry_checkpoint)
+        .await
+        .unwrap();
+    assert!(matches!(
+        service
+            .publish_failure(
+                &first,
+                &unique_id("failureop"),
+                RuntimeFailureClass::TransientTarget,
+            )
+            .await
+            .unwrap(),
+        AttemptMutationOutcome::Accepted(_)
+    ));
+    let retry = service.mutation_projection(&job).await.unwrap().unwrap();
+    assert_eq!(retry.phase, JobPhase::RetryWait);
+    assert_eq!(retry.failure, None);
+    assert_attempt_failure(
+        &harness.pool,
+        &first.attempt_id,
+        FailureRecord {
+            class: RuntimeFailureClass::TransientTarget,
+            disposition: FailureDisposition::Retryable,
+        },
+    )
+    .await;
+    sqlx::query(
+        "UPDATE jobs SET phase = 'queued', dispatch_needed = TRUE,
+            revision = revision + 1 WHERE job_id = $1",
+    )
+    .bind(job.as_str())
+    .execute(&harness.pool)
+    .await
+    .unwrap();
+    harness
+        .coordination
+        .release(
+            job.as_str(),
+            &remora_coordination::LeaseOwner::new(
+                first.attempt_id.as_str(),
+                first.fence,
+                first.worker_incarnation.as_str(),
+            )
+            .unwrap(),
+        )
+        .await
+        .unwrap();
+    service.release_local(&job);
+    harness.coordination.enqueue(job.as_str()).await.unwrap();
+    let second = harness.executing(&service, &job).await;
+    assert_eq!(second.fence, 2);
+    let initial_progress: i64 =
+        sqlx::query_scalar("SELECT initial_progress FROM attempts WHERE attempt_id = $1")
+            .bind(second.attempt_id.as_str())
+            .fetch_one(&harness.pool)
+            .await
+            .unwrap();
+    assert_eq!(initial_progress, 512);
+    service
+        .publish_failure(
+            &second,
+            &unique_id("failureop"),
+            RuntimeFailureClass::Internal,
+        )
+        .await
+        .unwrap();
+    let exhausted = service.mutation_projection(&job).await.unwrap().unwrap();
+    assert_eq!(exhausted.phase, JobPhase::Failed);
+    assert_eq!(
+        exhausted.failure,
+        Some(FailureRecord {
+            class: RuntimeFailureClass::Internal,
+            disposition: FailureDisposition::Retryable,
+        })
+    );
+
+    let permanent_job = harness.job().await;
+    let permanent = harness.executing(&service, &permanent_job).await;
+    service
+        .publish_failure(
+            &permanent,
+            &unique_id("failureop"),
+            RuntimeFailureClass::Authorization,
+        )
+        .await
+        .unwrap();
+    let projection = service
+        .mutation_projection(&permanent_job)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(projection.phase, JobPhase::Failed);
+    assert_eq!(
+        projection.failure.unwrap().disposition,
+        FailureDisposition::Permanent
+    );
+}
+
+#[tokio::test]
+async fn concurrent_success_and_failure_have_one_terminal_winner() {
+    let Some(harness) = Harness::new("race").await else {
+        return;
+    };
+    let service = harness.service(3);
+    let job = harness.job().await;
+    let authority = harness.executing(&service, &job).await;
+    let result = harness.sealed_result(&authority.attempt_id, 1, 64).await;
+    let success_service = service.clone();
+    let success_authority = authority.clone();
+    let success = tokio::spawn(async move {
+        success_service
+            .publish_success(&success_authority, &unique_id("successop"), result)
+            .await
+            .unwrap()
+    });
+    let failure_service = service.clone();
+    let failure_authority = authority.clone();
+    let failure = tokio::spawn(async move {
+        failure_service
+            .publish_failure(
+                &failure_authority,
+                &unique_id("failureop"),
+                RuntimeFailureClass::Authorization,
+            )
+            .await
+            .unwrap()
+    });
+    let (success, failure) = tokio::join!(success, failure);
+    let outcomes = [success.unwrap(), failure.unwrap()];
+    assert_eq!(
+        outcomes
+            .iter()
+            .filter(|outcome| matches!(outcome, AttemptMutationOutcome::Accepted(_)))
+            .count(),
+        1
+    );
+    let projection = service.mutation_projection(&job).await.unwrap().unwrap();
+    assert!(matches!(
+        projection.phase,
+        JobPhase::Succeeded | JobPhase::Failed
+    ));
+    assert_ne!(
+        projection.committed_result.is_some(),
+        projection.failure.is_some()
+    );
+}
+
+fn descriptor(
+    prefix: &str,
+    byte_length: u64,
+    item_count: Option<u64>,
+    digest: char,
+) -> DurableDescriptor {
+    DurableDescriptor {
+        id: unique_id(prefix),
+        schema_version: 1,
+        byte_length,
+        item_count,
+        checksum: format!("sha256:{}", digest.to_string().repeat(64)),
+    }
+}
+
+fn assert_rejected(outcome: AttemptMutationOutcome, expected: MutationRejection) {
+    assert!(
+        matches!(outcome, AttemptMutationOutcome::Rejected { reason, .. } if reason == expected),
+        "unexpected outcome: {outcome:?}"
+    );
+}
+
+async fn assert_attempt_failure(pool: &PgPool, attempt_id: &OpaqueId, expected: FailureRecord) {
+    let (state, failure): (String, serde_json::Value) =
+        sqlx::query_as("SELECT state, failure FROM attempts WHERE attempt_id = $1")
+            .bind(attempt_id.as_str())
+            .fetch_one(pool)
+            .await
+            .unwrap();
+    assert_eq!(state, "stopped");
+    assert_eq!(failure, serde_json::to_value(expected).unwrap());
+}
+
+async fn publication_from_database(pool: &PgPool, attempt_id: &OpaqueId) -> PublicationInput {
+    let row: (
+        String,
+        String,
+        String,
+        Option<String>,
+        String,
+        i64,
+        i64,
+        i64,
+    ) = sqlx::query_as(
+        "SELECT manifest_id, manifest_store, manifest_bucket, manifest_version_id,
+                manifest_sha256, manifest_size, member_count, total_bytes
+         FROM result_inventories WHERE attempt_id = $1",
+    )
+    .bind(attempt_id.as_str())
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    let manifest_id = OpaqueId::parse(row.0).unwrap();
+    PublicationInput {
+        attempt_id: attempt_id.clone(),
+        manifest_id: manifest_id.clone(),
+        schema_version: 1,
+        manifest: StoredObject {
+            identity: ObjectIdentity {
+                target_store_id: row.1,
+                bucket: row.2,
+                key: attempt_manifest_key(attempt_id, &manifest_id),
+                version_id: row.3,
+            },
+            content_sha256: ContentSha256::parse(row.4).unwrap(),
+            size: u64::try_from(row.5).unwrap(),
+        },
+        member_count: u64::try_from(row.6).unwrap(),
+        total_bytes: u64::try_from(row.7).unwrap(),
+    }
+}
+
+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()
+}