Separate orchestration by protocol action

crates/remora-api/src/public/credentials.rs

@@ -0,0 +1,404 @@
+use super::*;
+
+pub(super) async fn create_source_credential(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    request: web::Json<SourceCredentialCreateDto>,
+) -> Result<HttpResponse, PublicApiError> {
+    validate_request("SourceCredentialCreate", &*request)?;
+    let principal = state.principal(&http).await?;
+    let key = idempotency_key(&http)?;
+    let request = request.into_inner();
+    let request_fingerprint =
+        fingerprint(&request).map_err(|_| PublicApiError::invalid_request())?;
+    let create = create_credential(request)?;
+    let credentials = state.credentials.clone();
+    let owner = principal.id.clone();
+    let scope = MutationScope {
+        subject: &principal.id,
+        operation: "credential.create",
+        key,
+    };
+    if let Some(outcome) = observe_mutation(
+        &state.pool,
+        scope,
+        "create_source_credential",
+        &[],
+        request_fingerprint,
+    )
+    .await?
+    {
+        return credential_mutation_response(outcome, StatusCode::CREATED);
+    }
+    let prepared = credentials
+        .prepare_create(&owner, create)
+        .await
+        .map_err(PublicApiError::from_credential)?;
+    let outcome = execute_mutation(
+        &state.pool,
+        scope,
+        "create_source_credential",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                let credential = CredentialRegistry::commit_prepared_create_in_tx(tx, prepared)
+                    .await
+                    .map_err(|error| DurableError::Effect(error.to_string()))?;
+                let response = serde_json::to_value(
+                    credential_dto(credential)
+                        .map_err(|error| DurableError::Effect(error.to_string()))?,
+                )
+                .map_err(|error| DurableError::Effect(error.to_string()))?;
+                Ok((MutationStatus::Accepted, response))
+            })
+        },
+    )
+    .await?;
+    credential_mutation_response(outcome, StatusCode::CREATED)
+}
+
+pub(super) async fn list_source_credentials(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    query: web::Query<PageQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let caller = credential_caller(&principal);
+    let filter = CursorCodec::filter_fingerprint(&json!({
+        "resource": "credentials",
+        "authorization": authorization_snapshot(&principal)
+    }))?;
+    let cursor = query
+        .cursor
+        .as_deref()
+        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
+        .transpose()
+        .map_err(PublicApiError::from_cursor)?;
+    let registry_cursor = cursor
+        .as_ref()
+        .and_then(|value| value.after.as_ref())
+        .map(|value| value.id.as_str());
+    let page = state
+        .credentials
+        .list(&caller, page_limit(query.limit)?, registry_cursor)
+        .await
+        .map_err(PublicApiError::from_credential)?;
+    let next_cursor = page
+        .next_cursor
+        .as_deref()
+        .map(|raw| {
+            let next = KeysetPosition {
+                created_at: OffsetDateTime::UNIX_EPOCH,
+                id: raw.to_owned(),
+            };
+            let bound = match cursor.as_ref() {
+                Some(current) => next_page_cursor(current, next),
+                None => first_page_cursor(
+                    principal.id.as_str(),
+                    filter,
+                    KeysetPosition {
+                        created_at: OffsetDateTime::UNIX_EPOCH,
+                        id: String::new(),
+                    },
+                    next,
+                ),
+            };
+            state.cursors.encode(&bound)
+        })
+        .transpose()?;
+    let dto = SourceCredentialPageDto {
+        items: page
+            .items
+            .into_iter()
+            .map(credential_dto)
+            .collect::<Result<_, _>>()?,
+        next_cursor: NullableDto(next_cursor),
+    };
+    Ok(HttpResponse::Ok().json(dto))
+}
+
+pub(super) async fn get_source_credential(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let credential_id = parse_id(path.into_inner())?;
+    let credential = state
+        .credentials
+        .get(&credential_caller(&principal), &credential_id)
+        .await
+        .map_err(PublicApiError::from_credential)?
+        .ok_or_else(PublicApiError::not_found)?;
+    Ok(HttpResponse::Ok().json(credential_dto(credential)?))
+}
+
+pub(super) async fn rotate_source_credential(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+    request: web::Json<SourceCredentialRotateDto>,
+) -> Result<HttpResponse, PublicApiError> {
+    validate_request("SourceCredentialRotate", &*request)?;
+    let principal = state.principal(&http).await?;
+    let credential_id = parse_id(path.into_inner())?;
+    let key = idempotency_key(&http)?;
+    let request = request.into_inner();
+    let request_fingerprint = fingerprint(&json!({
+        "credential_id": credential_id,
+        "request": request
+    }))
+    .map_err(|_| PublicApiError::invalid_request())?;
+    let credentials = state.credentials.clone();
+    let caller = credential_caller(&principal);
+    let scope = MutationScope {
+        subject: &principal.id,
+        operation: "credential.rotate",
+        key,
+    };
+    if let Some(outcome) = observe_mutation(
+        &state.pool,
+        scope,
+        "rotate_source_credential",
+        &[],
+        request_fingerprint,
+    )
+    .await?
+    {
+        return credential_mutation_response(outcome, StatusCode::CREATED);
+    }
+    let existing = credentials
+        .get(&caller, &credential_id)
+        .await
+        .map_err(PublicApiError::from_credential)?
+        .ok_or_else(PublicApiError::not_found)?;
+    let material = credential_material_for_kind(request.material, existing.kind);
+    let prepared = credentials
+        .prepare_rotate(&caller, &credential_id, material)
+        .await
+        .map_err(PublicApiError::from_credential)?;
+    let outcome = execute_mutation(
+        &state.pool,
+        scope,
+        "rotate_source_credential",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                match CredentialRegistry::commit_prepared_rotate_in_tx(tx, prepared).await? {
+                    RotateTxOutcome::Applied(credential) => {
+                        let response = serde_json::to_value(
+                            credential_dto(credential)
+                                .map_err(|error| DurableError::Effect(error.to_string()))?,
+                        )
+                        .map_err(|error| DurableError::Effect(error.to_string()))?;
+                        Ok((MutationStatus::Accepted, response))
+                    }
+                    RotateTxOutcome::Revoked => Ok((
+                        MutationStatus::Rejected,
+                        public_error_body(
+                            ApiErrorCodeDto::PreconditionFailed,
+                            "precondition failed",
+                        ),
+                    )),
+                    RotateTxOutcome::IntentClaimed => Err(DurableError::Effect(
+                        "credential rotation intent is unavailable".into(),
+                    )),
+                }
+            })
+        },
+    )
+    .await?;
+    credential_mutation_response(outcome, StatusCode::CREATED)
+}
+
+pub(super) async fn revoke_source_credential(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let credential_id = parse_id(path.into_inner())?;
+    let key = idempotency_key(&http)?;
+    let request_fingerprint = fingerprint(&json!({ "credential_id": credential_id }))
+        .map_err(|_| PublicApiError::internal())?;
+    let caller = credential_caller(&principal);
+    let scope = MutationScope {
+        subject: &principal.id,
+        operation: "credential.revoke",
+        key,
+    };
+    if let Some(outcome) = observe_mutation(
+        &state.pool,
+        scope,
+        "revoke_source_credential",
+        &[],
+        request_fingerprint,
+    )
+    .await?
+    {
+        return credential_mutation_response(outcome, StatusCode::OK);
+    }
+    state
+        .credentials
+        .get(&caller, &credential_id)
+        .await
+        .map_err(PublicApiError::from_credential)?
+        .ok_or_else(PublicApiError::not_found)?;
+    let credential_for_effect = credential_id.clone();
+    let outcome = execute_mutation(
+        &state.pool,
+        scope,
+        "revoke_source_credential",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                let credential =
+                    CredentialRegistry::revoke_in_tx(tx, &caller, &credential_for_effect)
+                        .await
+                        .map_err(|error| DurableError::Effect(error.to_string()))?;
+                let response = serde_json::to_value(
+                    credential_dto(credential)
+                        .map_err(|error| DurableError::Effect(error.to_string()))?,
+                )
+                .map_err(|error| DurableError::Effect(error.to_string()))?;
+                Ok((MutationStatus::Accepted, response))
+            })
+        },
+    )
+    .await?;
+    credential_mutation_response(outcome, StatusCode::OK)
+}
+
+pub(super) fn create_credential(
+    request: SourceCredentialCreateDto,
+) -> Result<CreateCredential, PublicApiError> {
+    let (label, material) = match request {
+        SourceCredentialCreateDto::BasicAuth(value) => (
+            value.label,
+            SecretMaterial::BasicAuth {
+                username: value.material.username,
+                password: value.material.password,
+            },
+        ),
+        SourceCredentialCreateDto::BearerToken(value) => (
+            value.label,
+            SecretMaterial::BearerToken {
+                token: value.material.token,
+            },
+        ),
+        SourceCredentialCreateDto::HttpHeaders(value) => (
+            value.label,
+            SecretMaterial::HttpHeaders {
+                headers: value.material.headers,
+            },
+        ),
+        SourceCredentialCreateDto::ObjectStore(value) => (
+            value.label,
+            SecretMaterial::ObjectStore {
+                access_key_id: value.material.access_key_id,
+                secret_access_key: value.material.secret_access_key,
+                session_token: value.material.session_token,
+            },
+        ),
+        SourceCredentialCreateDto::HuggingFace(value) => (
+            value.label,
+            SecretMaterial::HuggingFace {
+                token: value.material.token,
+            },
+        ),
+    };
+    Ok(CreateCredential { label, material })
+}
+
+pub(super) fn credential_material(material: CredentialMaterialDto) -> SecretMaterial {
+    match material {
+        CredentialMaterialDto::BasicAuth(value) => SecretMaterial::BasicAuth {
+            username: value.username,
+            password: value.password,
+        },
+        CredentialMaterialDto::Token(value) => SecretMaterial::BearerToken { token: value.token },
+        CredentialMaterialDto::HttpHeaders(value) => SecretMaterial::HttpHeaders {
+            headers: value.headers,
+        },
+        CredentialMaterialDto::ObjectStore(value) => SecretMaterial::ObjectStore {
+            access_key_id: value.access_key_id,
+            secret_access_key: value.secret_access_key,
+            session_token: value.session_token,
+        },
+    }
+}
+
+pub(super) fn credential_material_for_kind(
+    material: CredentialMaterialDto,
+    kind: SourceCredentialKind,
+) -> SecretMaterial {
+    match (material, kind) {
+        (CredentialMaterialDto::Token(value), SourceCredentialKind::HuggingFace) => {
+            SecretMaterial::HuggingFace { token: value.token }
+        }
+        (material, _) => credential_material(material),
+    }
+}
+
+pub(super) fn credential_mutation_response(
+    outcome: remora_durable::MutationOutcome,
+    accepted_status: StatusCode,
+) -> Result<HttpResponse, PublicApiError> {
+    if outcome.disposition == MutationDisposition::Conflict {
+        return Err(PublicApiError::idempotency_conflict());
+    }
+    let response = if outcome.status == MutationStatus::Rejected {
+        HttpResponse::PreconditionFailed()
+            .content_type("application/problem+json")
+            .json(outcome.response)
+    } else {
+        HttpResponse::build(accepted_status).json(outcome.response)
+    };
+    Ok(with_idempotency_expiry(response))
+}
+
+pub(super) fn credential_dto(
+    value: SourceCredential,
+) -> Result<SourceCredentialDto, PublicApiError> {
+    let kind = match value.kind {
+        SourceCredentialKind::BasicAuth => SourceCredentialKindDto::BasicAuth,
+        SourceCredentialKind::BearerToken => SourceCredentialKindDto::BearerToken,
+        SourceCredentialKind::HttpHeaders => SourceCredentialKindDto::HttpHeaders,
+        SourceCredentialKind::ObjectStore => SourceCredentialKindDto::ObjectStore,
+        SourceCredentialKind::HuggingFace => SourceCredentialKindDto::HuggingFace,
+    };
+    let status = match value.status {
+        SourceCredentialStatus::Active => SourceCredentialStatusDto::Active,
+        SourceCredentialStatus::Revoked => SourceCredentialStatusDto::Revoked,
+    };
+    Ok(SourceCredentialDto {
+        id: value.id,
+        owner: value.owner,
+        label: value.label,
+        kind,
+        status,
+        active_version: NullableDto(
+            value
+                .active_version
+                .map(|version| {
+                    Ok::<_, PublicApiError>(SourceCredentialVersionDto {
+                        id: version.id,
+                        created_at: datetime(version.created_at)?,
+                    })
+                })
+                .transpose()?,
+        ),
+        created_at: datetime(value.created_at)?,
+        updated_at: datetime(value.updated_at)?,
+    })
+}
+
+pub(super) fn credential_caller(principal: &Principal) -> CredentialCaller {
+    CredentialCaller {
+        principal_id: principal.id.clone(),
+        administrator: principal.administrator,
+    }
+}

crates/remora-api/src/public/deletions.rs

@@ -0,0 +1,100 @@
+use super::*;
+
+pub(super) async fn soft_delete_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    lifecycle_mutation(state, http, path, LifecycleAction::SoftDelete).await
+}
+
+pub(super) async fn restore_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    lifecycle_mutation(state, http, path, LifecycleAction::Restore).await
+}
+
+pub(super) async fn lifecycle_mutation(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+    action: LifecycleAction,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    let key = idempotency_key(&http)?.to_owned();
+    let (operation, management) = match action {
+        LifecycleAction::SoftDelete => ("job.soft_delete", ManagementOperation::SoftDelete),
+        LifecycleAction::Restore => ("job.restore", ManagementOperation::Restore),
+    };
+    authorize_job(&state, &principal, &job_id, management).await?;
+    let action = match action {
+        LifecycleAction::SoftDelete => JobMutationAction::SoftDelete,
+        LifecycleAction::Restore => JobMutationAction::Restore,
+    };
+    let job = idempotent_job_mutation(&state, &principal, &job_id, &key, operation, action).await?;
+    Ok(with_idempotency_expiry(HttpResponse::Ok().json(job)))
+}
+
+pub(super) async fn hard_delete_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    authorize_job(&state, &principal, &job_id, ManagementOperation::HardDelete).await?;
+    let key = idempotency_key(&http)?.to_owned();
+    let request_fingerprint =
+        fingerprint(&json!({ "job_id": job_id })).map_err(|_| PublicApiError::internal())?;
+    let principal_for_effect = principal.clone();
+    let job_for_effect = job_id.clone();
+    let outcome = execute_mutation(
+        &state.pool,
+        MutationScope {
+            subject: &principal.id,
+            operation: "job.hard_delete",
+            key: &key,
+        },
+        "hard_delete_job",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                let outcome =
+                    DeletionService::hard_delete_in_tx(tx, &principal_for_effect, &job_for_effect)
+                        .await
+                        .map_err(DurableError::effect)?;
+                Ok((
+                    MutationStatus::Accepted,
+                    deletion_projection(&outcome.operation)
+                        .map_err(|error| DurableError::Effect(error.to_string()))?,
+                ))
+            })
+        },
+    )
+    .await?;
+    if outcome.disposition == MutationDisposition::Conflict {
+        return Err(PublicApiError::idempotency_conflict());
+    }
+    Ok(with_idempotency_expiry(
+        HttpResponse::Accepted().json(outcome.response),
+    ))
+}
+
+pub(super) async fn get_deletion_operation(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let operation_id = parse_id(path.into_inner())?;
+    let operation = state
+        .deletion
+        .operation(&principal, &operation_id)
+        .await
+        .map_err(PublicApiError::from_deletion)?;
+    Ok(HttpResponse::Ok().json(deletion_projection(&operation)?))
+}

crates/remora-api/src/public/events.rs

@@ -21,6 +21,21 @@ use super::{
     current_progress, encode_event_cursor, event_cursor, load_job_row, parse_id, request_id,
 };
 
+mod connection;
+mod frames;
+mod progress;
+mod projections;
+mod tickets;
+
+#[cfg(test)]
+use connection::decode_event_cursor;
+pub(super) use connection::{connect, prune_history};
+use frames::*;
+use progress::*;
+use projections::*;
+pub(super) use tickets::issue_ticket;
+use tickets::redeem_ticket;
+
 const EVENT_TOPICS: [&str; 2] = ["job.updated", "deletion.updated"];
 const EVENT_BATCH: i64 = 100;
 
@@ -46,366 +61,11 @@ struct TicketScope {
     prefix: String,
 }
 
-pub(super) async fn issue_ticket(
-    state: web::Data<PublicApiState>,
-    request: HttpRequest,
-) -> Result<HttpResponse, PublicApiError> {
-    let principal = state.principal(&request).await?;
-    let origin = allowed_origin(&state, &request)?;
-    let ticket = random_ticket();
-    let digest = ticket_digest(ticket.as_str());
-    let mut tx = state.pool.begin().await?;
-    sqlx::query("DELETE FROM live_event_tickets WHERE expires_at <= now()")
-        .execute(tx.as_mut())
-        .await?;
-    let expires_at: OffsetDateTime = sqlx::query_scalar(
-        "INSERT INTO live_event_tickets (
-            ticket_digest, principal, origin, created_at, expires_at
-         ) VALUES (
-            $1, $2, $3, now(),
-            now() + $4 * interval '1 second'
-         )
-         RETURNING expires_at",
-    )
-    .bind(digest.as_slice())
-    .bind(Json(ticket_principal_projection(&principal)))
-    .bind(&origin)
-    .bind(state.event_ticket_ttl.as_secs_f64())
-    .fetch_one(tx.as_mut())
-    .await?;
-    tx.commit().await?;
-    Ok(HttpResponse::Created()
-        .insert_header((header::CACHE_CONTROL, "no-store"))
-        .insert_header((header::PRAGMA, "no-cache"))
-        .json(EventTicketDto {
-            ticket,
-            expires_at: event_time(expires_at)?,
-        }))
-}
-
-pub(super) async fn connect(
-    state: web::Data<PublicApiState>,
-    request: HttpRequest,
-    body: web::Payload,
-    query: web::Query<EventQuery>,
-) -> Result<HttpResponse, PublicApiError> {
-    let bearer = request.headers().contains_key(header::AUTHORIZATION);
-    let principal = match (bearer, query.ticket.as_deref()) {
-        (true, None) => state.principal(&request).await?,
-        (false, Some(ticket)) => redeem_ticket(&state, &request, ticket).await?,
-        (false, None) => return Err(PublicApiError::unauthenticated()),
-        (true, Some(_)) => return Err(PublicApiError::invalid_request()),
-    };
-    let now = OffsetDateTime::now_utc();
-    let retention =
-        time::Duration::try_from(state.event_retention).map_err(|_| PublicApiError::internal())?;
-    prune_history(&state.pool, now - retention).await?;
-    let start = classify_cursor(&state, &principal, query.cursor.as_deref(), now).await?;
-    let (response, session, messages) =
-        actix_ws::handle(&request, body).map_err(|_| PublicApiError::invalid_request())?;
-    let state = state.get_ref().clone();
-    actix_web::rt::spawn(async move {
-        match start {
-            EventStart::Replay(after_id) => {
-                run_connection(state, principal, after_id, session, messages).await;
-            }
-            EventStart::Resync(reason) => {
-                send_resync(state, principal, reason, session).await;
-            }
-        }
-    });
-    Ok(response)
-}
-
-fn allowed_origin(state: &PublicApiState, request: &HttpRequest) -> Result<String, PublicApiError> {
-    let mut origins = request.headers().get_all(header::ORIGIN);
-    let origin = origins
-        .next()
-        .and_then(|value| value.to_str().ok())
-        .ok_or_else(PublicApiError::forbidden)?;
-    if origins.next().is_some() || !state.event_allowed_origins.contains(origin) {
-        return Err(PublicApiError::forbidden());
-    }
-    Ok(origin.to_owned())
-}
-
-async fn redeem_ticket(
-    state: &PublicApiState,
-    request: &HttpRequest,
-    ticket: &str,
-) -> Result<Principal, PublicApiError> {
-    let origin = allowed_origin(state, request)?;
-    let digest = ticket_digest(ticket);
-    let projection: Option<Json<TicketPrincipal>> = sqlx::query_scalar(
-        "DELETE FROM live_event_tickets
-         WHERE ticket_digest = $1 AND origin = $2 AND expires_at > now()
-         RETURNING principal",
-    )
-    .bind(digest.as_slice())
-    .bind(origin)
-    .fetch_optional(&state.pool)
-    .await?;
-    projection
-        .ok_or_else(PublicApiError::unauthenticated)
-        .and_then(|projection| ticket_principal(projection.0))
-}
-
-fn ticket_principal_projection(principal: &Principal) -> TicketPrincipal {
-    TicketPrincipal {
-        id: principal.id.as_str().to_owned(),
-        issuer: principal.issuer.clone(),
-        subject: principal.subject.clone(),
-        administrator: principal.administrator,
-        scopes: principal
-            .scopes
-            .iter()
-            .map(|scope| TicketScope {
-                store: scope.store.as_str().to_owned(),
-                bucket: scope.bucket.as_str().to_owned(),
-                prefix: scope.prefix.canonical(),
-            })
-            .collect(),
-    }
-}
-
-fn ticket_principal(projection: TicketPrincipal) -> Result<Principal, PublicApiError> {
-    let scopes = projection
-        .scopes
-        .into_iter()
-        .map(|scope| {
-            Ok(StorageScope {
-                store: TargetStoreId::parse(scope.store).map_err(|_| PublicApiError::internal())?,
-                bucket: BucketName::parse(scope.bucket).map_err(|_| PublicApiError::internal())?,
-                prefix: TargetPrefix::parse(&scope.prefix, std::iter::empty())
-                    .map_err(|_| PublicApiError::internal())?,
-            })
-        })
-        .collect::<Result<Vec<_>, PublicApiError>>()?;
-    Ok(Principal {
-        id: OpaqueId::parse(projection.id).map_err(|_| PublicApiError::internal())?,
-        issuer: projection.issuer,
-        subject: projection.subject,
-        administrator: projection.administrator,
-        scopes,
-    })
-}
-
-fn random_ticket() -> OpaqueId {
-    let mut bytes = [0_u8; 32];
-    getrandom::fill(&mut bytes).expect("operating system randomness is required");
-    let mut value = String::from("t");
-    for byte in bytes {
-        use std::fmt::Write as _;
-        write!(value, "{byte:02x}").expect("writing to String cannot fail");
-    }
-    OpaqueId::parse(value).expect("random event ticket is valid")
-}
-
-fn ticket_digest(ticket: &str) -> [u8; 32] {
-    Sha256::digest(ticket.as_bytes()).into()
-}
-
-pub(super) async fn prune_history(
-    pool: &PgPool,
-    before: OffsetDateTime,
-) -> Result<u64, DurableError> {
-    let mut tx = pool.begin().await?;
-    let pruned: Vec<i64> = sqlx::query_scalar(
-        "DELETE FROM outbox_messages
-         WHERE topic = ANY($1) AND created_at < $2
-         RETURNING id",
-    )
-    .bind(&EVENT_TOPICS[..])
-    .bind(before)
-    .fetch_all(tx.as_mut())
-    .await?;
-    if let Some(floor) = pruned.iter().max() {
-        sqlx::query(
-            "UPDATE live_event_history_state
-             SET floor_outbox_id = GREATEST(floor_outbox_id, $1)
-             WHERE singleton",
-        )
-        .bind(floor)
-        .execute(tx.as_mut())
-        .await?;
-    }
-    tx.commit().await?;
-    Ok(pruned.len() as u64)
-}
-
 enum EventStart {
     Replay(i64),
     Resync(ResyncReasonDto),
 }
 
-async fn classify_cursor(
-    state: &PublicApiState,
-    principal: &Principal,
-    token: Option<&str>,
-    now: OffsetDateTime,
-) -> Result<EventStart, PublicApiError> {
-    let Some(token) = token else {
-        return Ok(EventStart::Resync(ResyncReasonDto::CursorAbsent));
-    };
-    let after_id =
-        match decode_event_cursor(&state.cursors, state.event_retention, principal, token, now)? {
-            Ok(after_id) => after_id,
-            Err(reason) => return Ok(EventStart::Resync(reason)),
-        };
-    let (floor, maximum): (i64, i64) = sqlx::query_as(
-        "SELECT history.floor_outbox_id,
-                GREATEST(
-                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
-                    history.floor_outbox_id
-                )
-         FROM live_event_history_state history
-         WHERE history.singleton",
-    )
-    .fetch_one(&state.pool)
-    .await?;
-    if after_id < floor || after_id > maximum {
-        return Ok(EventStart::Resync(ResyncReasonDto::HistoryUnavailable));
-    }
-    Ok(EventStart::Replay(after_id))
-}
-
-fn decode_event_cursor(
-    cursors: &remora_durable::CursorCodec,
-    retention: std::time::Duration,
-    principal: &Principal,
-    token: &str,
-    now: OffsetDateTime,
-) -> Result<Result<i64, ResyncReasonDto>, PublicApiError> {
-    let cursor = match cursors.decode(token) {
-        Ok(cursor) => cursor,
-        Err(_) => return Ok(Err(ResyncReasonDto::CursorInvalid)),
-    };
-    let Some(after_id) = event_position(&cursor) else {
-        return Ok(Err(ResyncReasonDto::CursorInvalid));
-    };
-    let expected = remora_durable::CursorCodec::filter_fingerprint(&serde_json::json!({
-        "resource": "events",
-        "authorization": authorization_snapshot(principal)
-    }))?;
-    if cursor.subject != principal.id.as_str() || cursor.filter_fingerprint != expected {
-        return Ok(Err(ResyncReasonDto::AuthorizationChanged));
-    }
-    let retention = time::Duration::try_from(retention).map_err(|_| PublicApiError::internal())?;
-    if cursor.high_water.created_at + retention < now {
-        return Ok(Err(ResyncReasonDto::CursorExpired));
-    }
-    Ok(Ok(after_id))
-}
-
-fn event_position(cursor: &KeysetCursor) -> Option<i64> {
-    let high_water = cursor
-        .high_water
-        .id
-        .strip_prefix("event-")?
-        .parse::<i64>()
-        .ok()?;
-    let after = cursor.after.as_ref()?;
-    let after_id = after.id.strip_prefix("event-")?.parse::<i64>().ok()?;
-    (high_water == after_id && after.created_at == cursor.high_water.created_at).then_some(after_id)
-}
-
-async fn send_resync(
-    state: PublicApiState,
-    principal: Principal,
-    reason: ResyncReasonDto,
-    mut session: actix_ws::Session,
-) {
-    let Ok(cursor) = event_cursor(&state, &principal).await else {
-        let _ = session.close(None).await;
-        return;
-    };
-    let Ok(frame) = resync_frame(reason, 1, &cursor, OffsetDateTime::now_utc()) else {
-        let _ = session.close(None).await;
-        return;
-    };
-    let _ = send_frame(&mut session, &frame).await;
-    let _ = session.close(None).await;
-}
-
-async fn run_connection(
-    state: PublicApiState,
-    principal: Principal,
-    mut after_id: i64,
-    mut session: actix_ws::Session,
-    mut messages: actix_ws::MessageStream,
-) {
-    let mut sequence = 0_u64;
-    let mut poll = tokio::time::interval(state.event_poll_interval);
-    let mut heartbeat = tokio::time::interval(state.event_heartbeat_interval);
-    poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-    heartbeat.tick().await;
-    loop {
-        tokio::select! {
-            message = messages.next() => {
-                match message {
-                    Some(Ok(actix_ws::Message::Ping(bytes))) => {
-                        if session.pong(&bytes).await.is_err() {
-                            break;
-                        }
-                    }
-                    Some(Ok(actix_ws::Message::Pong(_))) => {}
-                    Some(Ok(actix_ws::Message::Close(reason))) => {
-                        let _ = session.close(reason).await;
-                        return;
-                    }
-                    Some(Err(_)) | None => break,
-                    Some(Ok(_)) => break,
-                }
-            }
-            _ = poll.tick() => {
-                if poll_progress(&state).await.is_err() {
-                    break;
-                }
-                match replay_events(
-                    &state,
-                    &principal,
-                    &mut after_id,
-                    &mut sequence,
-                    &mut session,
-                ).await {
-                    Ok(ReplayStatus::Current) => {}
-                    Ok(ReplayStatus::HistoryUnavailable) => {
-                        let now = OffsetDateTime::now_utc();
-                        let Ok(cursor) = event_cursor(&state, &principal).await else {
-                            break;
-                        };
-                        let Ok(frame) = resync_frame(
-                            ResyncReasonDto::HistoryUnavailable,
-                            next_sequence(&mut sequence),
-                            &cursor,
-                            now,
-                        ) else {
-                            break;
-                        };
-                        let _ = send_frame(&mut session, &frame).await;
-                        break;
-                    }
-                    Err(()) => break,
-                }
-            }
-            _ = heartbeat.tick() => {
-                let now = OffsetDateTime::now_utc();
-                let Ok(cursor) = encode_event_cursor(&state, &principal, (after_id, now)) else {
-                    break;
-                };
-                let Ok(frame) = heartbeat_frame(next_sequence(&mut sequence), &cursor, now) else {
-                    break;
-                };
-                if send_frame(&mut session, &frame).await.is_err() {
-                    break;
-                }
-            }
-        }
-    }
-    let _ = session.close(None).await;
-}
-
 #[derive(FromRow)]
 struct ReplayRow {
     id: i64,
@@ -428,134 +88,6 @@ enum ReplayStatus {
     HistoryUnavailable,
 }
 
-async fn replay_events(
-    state: &PublicApiState,
-    principal: &Principal,
-    after_id: &mut i64,
-    sequence: &mut u64,
-    session: &mut actix_ws::Session,
-) -> Result<ReplayStatus, ()> {
-    for _ in 0..10 {
-        let batch = sqlx::query_as::<_, ReplayBatchRow>(
-            "SELECT history.floor_outbox_id, events.id, events.topic,
-                    events.payload, events.created_at
-             FROM live_event_history_state history
-             LEFT JOIN LATERAL (
-                 SELECT id, topic, payload, created_at
-                 FROM outbox_messages
-                 WHERE id > $1 AND topic = ANY($2)
-                 ORDER BY id
-                 LIMIT $3
-             ) events ON TRUE
-             WHERE history.singleton
-             ORDER BY events.id",
-        )
-        .bind(*after_id)
-        .bind(&EVENT_TOPICS[..])
-        .bind(EVENT_BATCH)
-        .fetch_all(&state.pool)
-        .await
-        .map_err(|_| ())?;
-        let Some(first) = batch.first() else {
-            return Err(());
-        };
-        if *after_id < first.floor_outbox_id {
-            return Ok(ReplayStatus::HistoryUnavailable);
-        }
-        let rows = batch
-            .into_iter()
-            .filter_map(|row| {
-                Some(ReplayRow {
-                    id: row.id?,
-                    topic: row.topic?,
-                    payload: row.payload?,
-                    created_at: row.created_at?,
-                })
-            })
-            .collect::<Vec<_>>();
-        let count = rows.len();
-        for row in rows {
-            *after_id = row.id;
-            let now = OffsetDateTime::now_utc();
-            let cursor = encode_event_cursor(state, principal, (row.id, now)).map_err(|_| ())?;
-            let candidate_sequence = sequence.saturating_add(1);
-            let frame = match row.topic.as_str() {
-                "job.updated" => {
-                    project_job_event(state, principal, &row, candidate_sequence, &cursor).await
-                }
-                "deletion.updated" => {
-                    project_deletion_event(state, principal, &row, candidate_sequence, &cursor)
-                        .await
-                }
-                _ => None,
-            };
-            if let Some(frame) = frame {
-                *sequence = candidate_sequence;
-                send_frame(session, &frame).await?;
-            }
-        }
-        if count < EVENT_BATCH as usize {
-            break;
-        }
-    }
-    Ok(ReplayStatus::Current)
-}
-
-async fn project_job_event(
-    state: &PublicApiState,
-    principal: &Principal,
-    event: &ReplayRow,
-    sequence: u64,
-    cursor: &str,
-) -> Option<EventFrameDto> {
-    let job_id = parse_payload_id(&event.payload, "job_id")?;
-    let row = load_job_row(&state.pool, &job_id).await.ok()??;
-    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
-    let progress = current_progress(state, &row)
-        .await
-        .ok()
-        .flatten()
-        .map(progress_fingerprint);
-    job_frame(
-        &row,
-        progress.as_ref(),
-        stable_event_id(event.id),
-        sequence,
-        cursor,
-        event.created_at,
-    )
-    .ok()
-}
-
-async fn project_deletion_event(
-    state: &PublicApiState,
-    principal: &Principal,
-    event: &ReplayRow,
-    sequence: u64,
-    cursor: &str,
-) -> Option<EventFrameDto> {
-    let operation_id = parse_payload_id(&event.payload, "operation_id")?;
-    let (job_id, state_name): (String, String) =
-        sqlx::query_as("SELECT job_id, state FROM deletion_operations WHERE operation_id = $1")
-            .bind(operation_id.as_str())
-            .fetch_optional(&state.pool)
-            .await
-            .ok()??;
-    let job_id = parse_id(job_id).ok()?;
-    let row = load_job_row(&state.pool, &job_id).await.ok()??;
-    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
-    deletion_frame(
-        job_id,
-        operation_id,
-        parse_deletion_state(&state_name)?,
-        stable_event_id(event.id),
-        sequence,
-        cursor,
-        event.created_at,
-    )
-    .ok()
-}
-
 #[derive(Clone, Debug, Eq, PartialEq)]
 struct ProgressFingerprint {
     attempt_id: OpaqueId,
@@ -566,370 +98,6 @@ struct ProgressFingerprint {
     total_members: Option<u64>,
 }
 
-fn progress_fingerprint(snapshot: remora_attempt::ProgressSnapshot) -> ProgressFingerprint {
-    ProgressFingerprint {
-        attempt_id: snapshot.attempt_id,
-        fence: snapshot.fence,
-        completed_bytes: snapshot.progress.completed_bytes,
-        total_bytes: snapshot.progress.total_bytes,
-        completed_members: snapshot.progress.completed_members,
-        total_members: snapshot.progress.total_members,
-    }
-}
-
-async fn poll_progress(state: &PublicApiState) -> Result<(), ()> {
-    let delay = state.event_poll_interval.as_secs_f64();
-    let claimed = sqlx::query_scalar::<_, bool>(
-        "UPDATE live_event_history_state
-         SET progress_poll_not_before = now() + ($1 * interval '1 second')
-         WHERE singleton AND progress_poll_not_before <= now()
-         RETURNING singleton",
-    )
-    .bind(delay)
-    .fetch_optional(&state.pool)
-    .await
-    .map_err(|_| ())?
-    .is_some();
-    if !claimed {
-        return Ok(());
-    }
-    let mut after_job = None::<String>;
-    loop {
-        let job_ids: Vec<String> = sqlx::query_scalar(
-            "SELECT job_id FROM jobs
-             WHERE phase IN ('running', 'cancelling')
-               AND ($1::text IS NULL OR job_id > $1)
-             ORDER BY job_id
-             LIMIT 200",
-        )
-        .bind(after_job.as_deref())
-        .fetch_all(&state.pool)
-        .await
-        .map_err(|_| ())?;
-        if job_ids.is_empty() {
-            break;
-        }
-        let count = job_ids.len();
-        for job_id in job_ids {
-            after_job = Some(job_id.clone());
-            let Some(row) = load_job_row(&state.pool, &parse_id(job_id.clone()).map_err(|_| ())?)
-                .await
-                .map_err(|_| ())?
-            else {
-                continue;
-            };
-            let progress = current_progress(state, &row)
-                .await
-                .ok()
-                .flatten()
-                .map(progress_fingerprint);
-            record_progress_change(&state.pool, &row, progress.as_ref())
-                .await
-                .map_err(|_| ())?;
-        }
-        if count < 200 {
-            break;
-        }
-    }
-    Ok(())
-}
-
-async fn record_progress_change(
-    pool: &PgPool,
-    row: &JobRow,
-    progress: Option<&ProgressFingerprint>,
-) -> Result<(), PublicApiError> {
-    let Some(attempt_id) = row.authority_attempt_id.as_deref() else {
-        return Ok(());
-    };
-    let mut tx = pool.begin().await?;
-    let now = OffsetDateTime::now_utc();
-    let changed = if let Some(progress) = progress {
-        sqlx::query_scalar::<_, String>(
-            "INSERT INTO live_progress_event_state (
-                job_id, attempt_id, fence, progress_present, completed_bytes,
-                total_bytes, completed_members, total_members, updated_at
-             ) VALUES ($1, $2, $3, TRUE, $4, $5, $6, $7, $8)
-             ON CONFLICT (job_id) DO UPDATE
-             SET attempt_id = EXCLUDED.attempt_id,
-                 fence = EXCLUDED.fence,
-                 progress_present = TRUE,
-                 completed_bytes = EXCLUDED.completed_bytes,
-                 total_bytes = EXCLUDED.total_bytes,
-                 completed_members = EXCLUDED.completed_members,
-                 total_members = EXCLUDED.total_members,
-                 updated_at = EXCLUDED.updated_at
-             WHERE (
-                 live_progress_event_state.attempt_id,
-                 live_progress_event_state.fence,
-                 live_progress_event_state.progress_present,
-                 live_progress_event_state.completed_bytes,
-                 live_progress_event_state.total_bytes,
-                 live_progress_event_state.completed_members,
-                 live_progress_event_state.total_members
-             ) IS DISTINCT FROM (
-                 EXCLUDED.attempt_id,
-                 EXCLUDED.fence,
-                 TRUE,
-                 EXCLUDED.completed_bytes,
-                 EXCLUDED.total_bytes,
-                 EXCLUDED.completed_members,
-                 EXCLUDED.total_members
-             )
-             RETURNING job_id",
-        )
-        .bind(&row.job_id)
-        .bind(progress.attempt_id.as_str())
-        .bind(i64::try_from(progress.fence).map_err(|_| PublicApiError::internal())?)
-        .bind(i64::try_from(progress.completed_bytes).map_err(|_| PublicApiError::internal())?)
-        .bind(
-            progress
-                .total_bytes
-                .map(i64::try_from)
-                .transpose()
-                .map_err(|_| PublicApiError::internal())?,
-        )
-        .bind(i64::try_from(progress.completed_members).map_err(|_| PublicApiError::internal())?)
-        .bind(
-            progress
-                .total_members
-                .map(i64::try_from)
-                .transpose()
-                .map_err(|_| PublicApiError::internal())?,
-        )
-        .bind(now)
-        .fetch_optional(tx.as_mut())
-        .await?
-        .is_some()
-    } else {
-        sqlx::query_scalar::<_, String>(
-            "UPDATE live_progress_event_state
-             SET attempt_id = $2, fence = $3, progress_present = FALSE,
-                 completed_bytes = NULL, total_bytes = NULL,
-                 completed_members = NULL, total_members = NULL, updated_at = $4
-             WHERE job_id = $1
-               AND (
-                   attempt_id IS DISTINCT FROM $2
-                   OR fence IS DISTINCT FROM $3
-                   OR progress_present
-                   OR completed_bytes IS NOT NULL
-                   OR total_bytes IS NOT NULL
-                   OR completed_members IS NOT NULL
-                   OR total_members IS NOT NULL
-               )
-             RETURNING job_id",
-        )
-        .bind(&row.job_id)
-        .bind(attempt_id)
-        .bind(row.current_fence)
-        .bind(now)
-        .fetch_optional(tx.as_mut())
-        .await?
-        .is_some()
-    };
-    if changed {
-        remora_durable::enqueue(
-            &mut tx,
-            "job.updated",
-            serde_json::json!({ "job_id": row.job_id }),
-        )
-        .await?;
-    }
-    tx.commit().await?;
-    Ok(())
-}
-
-fn job_frame(
-    row: &JobRow,
-    progress: Option<&ProgressFingerprint>,
-    event_id: OpaqueId,
-    sequence: u64,
-    cursor: &str,
-    occurred_at: OffsetDateTime,
-) -> Result<EventFrameDto, PublicApiError> {
-    let authoritative_attempt = row
-        .authority_attempt_id
-        .as_deref()
-        .map(|attempt_id| {
-            Ok::<_, PublicApiError>(EventAttemptDto {
-                attempt_id: parse_id(attempt_id)?,
-                fence: PositiveIntegerDto::new(
-                    u64::try_from(row.current_fence).map_err(|_| PublicApiError::internal())?,
-                )
-                .map_err(|_| PublicApiError::internal())?,
-            })
-        })
-        .transpose()?;
-    let progress = progress
-        .filter(|progress| {
-            row.authority_attempt_id.as_deref() == Some(progress.attempt_id.as_str())
-                && u64::try_from(row.current_fence).ok() == Some(progress.fence)
-        })
-        .map(event_progress)
-        .transpose()?;
-    Ok(EventFrameDto::JobChanged(JobChangedDto {
-        schema_version: SchemaVersionDto,
-        event_id,
-        sequence: positive(sequence)?,
-        cursor: event_cursor_dto(cursor)?,
-        occurred_at: event_time(occurred_at)?,
-        job_id: parse_id(row.job_id.clone())?,
-        job_revision: positive(
-            u64::try_from(row.revision).map_err(|_| PublicApiError::internal())?,
-        )?,
-        phase: parse_phase(&row.phase)?,
-        lifecycle: parse_lifecycle(&row.lifecycle)?,
-        authoritative_attempt: NullableDto(authoritative_attempt),
-        progress: NullableDto(progress),
-    }))
-}
-
-fn deletion_frame(
-    job_id: OpaqueId,
-    operation_id: OpaqueId,
-    state: DeletionStateDto,
-    event_id: OpaqueId,
-    sequence: u64,
-    cursor: &str,
-    occurred_at: OffsetDateTime,
-) -> Result<EventFrameDto, PublicApiError> {
-    Ok(EventFrameDto::DeletionChanged(DeletionChangedDto {
-        schema_version: SchemaVersionDto,
-        event_id,
-        sequence: positive(sequence)?,
-        cursor: event_cursor_dto(cursor)?,
-        occurred_at: event_time(occurred_at)?,
-        job_id,
-        operation_id,
-        state,
-    }))
-}
-
-fn resync_frame(
-    reason: ResyncReasonDto,
-    sequence: u64,
-    cursor: &str,
-    occurred_at: OffsetDateTime,
-) -> Result<EventFrameDto, PublicApiError> {
-    Ok(EventFrameDto::ResyncRequired(ResyncRequiredDto {
-        schema_version: SchemaVersionDto,
-        event_id: request_id(),
-        sequence: positive(sequence)?,
-        cursor: event_cursor_dto(cursor)?,
-        occurred_at: event_time(occurred_at)?,
-        reason,
-    }))
-}
-
-fn heartbeat_frame(
-    sequence: u64,
-    cursor: &str,
-    occurred_at: OffsetDateTime,
-) -> Result<EventFrameDto, PublicApiError> {
-    Ok(EventFrameDto::Heartbeat(HeartbeatDto {
-        schema_version: SchemaVersionDto,
-        event_id: request_id(),
-        sequence: positive(sequence)?,
-        cursor: event_cursor_dto(cursor)?,
-        occurred_at: event_time(occurred_at)?,
-    }))
-}
-
-fn event_progress(progress: &ProgressFingerprint) -> Result<EventProgressDto, PublicApiError> {
-    Ok(EventProgressDto {
-        attempt_id: progress.attempt_id.clone(),
-        fence: positive(progress.fence)?,
-        completed_bytes: safe(progress.completed_bytes)?,
-        total_bytes: progress
-            .total_bytes
-            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
-            .transpose()?,
-        completed_members: safe(progress.completed_members)?,
-        total_members: progress
-            .total_members
-            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
-            .transpose()?,
-    })
-}
-
-fn parse_payload_id(payload: &Value, field: &str) -> Option<OpaqueId> {
-    payload
-        .get(field)?
-        .as_str()
-        .and_then(|value| OpaqueId::parse(value).ok())
-}
-
-fn stable_event_id(id: i64) -> OpaqueId {
-    OpaqueId::parse(format!("event_{id:016}")).expect("positive outbox ids form opaque event ids")
-}
-
-fn parse_phase(value: &str) -> Result<JobPhase, PublicApiError> {
-    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(PublicApiError::internal()),
-    }
-}
-
-fn parse_lifecycle(value: &str) -> Result<JobLifecycle, PublicApiError> {
-    match value {
-        "visible" => Ok(JobLifecycle::Visible),
-        "soft_deleted" => Ok(JobLifecycle::SoftDeleted),
-        "hard_deleting" => Ok(JobLifecycle::HardDeleting),
-        "hard_deleted" => Ok(JobLifecycle::HardDeleted),
-        _ => Err(PublicApiError::internal()),
-    }
-}
-
-fn parse_deletion_state(value: &str) -> Option<DeletionStateDto> {
-    match value {
-        "queued" => Some(DeletionStateDto::Queued),
-        "running" => Some(DeletionStateDto::Running),
-        "retry_wait" => Some(DeletionStateDto::RetryWait),
-        "blocked" => Some(DeletionStateDto::Blocked),
-        "completed" => Some(DeletionStateDto::Completed),
-        _ => None,
-    }
-}
-
-fn positive(value: u64) -> Result<PositiveIntegerDto, PublicApiError> {
-    PositiveIntegerDto::new(value).map_err(|_| PublicApiError::internal())
-}
-
-fn safe(value: u64) -> Result<SafeIntegerDto, PublicApiError> {
-    SafeIntegerDto::new(value).map_err(|_| PublicApiError::internal())
-}
-
-fn event_cursor_dto(value: &str) -> Result<CursorDto, PublicApiError> {
-    CursorDto::parse(value).map_err(|_| PublicApiError::internal())
-}
-
-fn event_time(value: OffsetDateTime) -> Result<DateTimeDto, PublicApiError> {
-    DateTimeDto::parse(
-        value
-            .format(&Rfc3339)
-            .map_err(|_| PublicApiError::internal())?,
-    )
-    .map_err(|_| PublicApiError::internal())
-}
-
-fn next_sequence(sequence: &mut u64) -> u64 {
-    *sequence = sequence.saturating_add(1);
-    *sequence
-}
-
-async fn send_frame(session: &mut actix_ws::Session, frame: &EventFrameDto) -> Result<(), ()> {
-    session
-        .text(serde_json::to_string(frame).map_err(|_| ())?)
-        .await
-        .map_err(|_| ())
-}
-
 #[cfg(test)]
 mod tests {
     use remora_domain::{BucketName, StorageScope, TargetPrefix, TargetStoreId};

crates/remora-api/src/public/events/connection.rs

@@ -0,0 +1,304 @@
+use super::*;
+
+pub(in crate::public) async fn connect(
+    state: web::Data<PublicApiState>,
+    request: HttpRequest,
+    body: web::Payload,
+    query: web::Query<EventQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let bearer = request.headers().contains_key(header::AUTHORIZATION);
+    let principal = match (bearer, query.ticket.as_deref()) {
+        (true, None) => state.principal(&request).await?,
+        (false, Some(ticket)) => redeem_ticket(&state, &request, ticket).await?,
+        (false, None) => return Err(PublicApiError::unauthenticated()),
+        (true, Some(_)) => return Err(PublicApiError::invalid_request()),
+    };
+    let now = OffsetDateTime::now_utc();
+    let retention =
+        time::Duration::try_from(state.event_retention).map_err(|_| PublicApiError::internal())?;
+    prune_history(&state.pool, now - retention).await?;
+    let start = classify_cursor(&state, &principal, query.cursor.as_deref(), now).await?;
+    let (response, session, messages) =
+        actix_ws::handle(&request, body).map_err(|_| PublicApiError::invalid_request())?;
+    let state = state.get_ref().clone();
+    actix_web::rt::spawn(async move {
+        match start {
+            EventStart::Replay(after_id) => {
+                run_connection(state, principal, after_id, session, messages).await;
+            }
+            EventStart::Resync(reason) => {
+                send_resync(state, principal, reason, session).await;
+            }
+        }
+    });
+    Ok(response)
+}
+
+pub(in crate::public) async fn prune_history(
+    pool: &PgPool,
+    before: OffsetDateTime,
+) -> Result<u64, DurableError> {
+    let mut tx = pool.begin().await?;
+    let pruned: Vec<i64> = sqlx::query_scalar(
+        "DELETE FROM outbox_messages
+         WHERE topic = ANY($1) AND created_at < $2
+         RETURNING id",
+    )
+    .bind(&EVENT_TOPICS[..])
+    .bind(before)
+    .fetch_all(tx.as_mut())
+    .await?;
+    if let Some(floor) = pruned.iter().max() {
+        sqlx::query(
+            "UPDATE live_event_history_state
+             SET floor_outbox_id = GREATEST(floor_outbox_id, $1)
+             WHERE singleton",
+        )
+        .bind(floor)
+        .execute(tx.as_mut())
+        .await?;
+    }
+    tx.commit().await?;
+    Ok(pruned.len() as u64)
+}
+
+pub(super) async fn classify_cursor(
+    state: &PublicApiState,
+    principal: &Principal,
+    token: Option<&str>,
+    now: OffsetDateTime,
+) -> Result<EventStart, PublicApiError> {
+    let Some(token) = token else {
+        return Ok(EventStart::Resync(ResyncReasonDto::CursorAbsent));
+    };
+    let after_id =
+        match decode_event_cursor(&state.cursors, state.event_retention, principal, token, now)? {
+            Ok(after_id) => after_id,
+            Err(reason) => return Ok(EventStart::Resync(reason)),
+        };
+    let (floor, maximum): (i64, i64) = sqlx::query_as(
+        "SELECT history.floor_outbox_id,
+                GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                )
+         FROM live_event_history_state history
+         WHERE history.singleton",
+    )
+    .fetch_one(&state.pool)
+    .await?;
+    if after_id < floor || after_id > maximum {
+        return Ok(EventStart::Resync(ResyncReasonDto::HistoryUnavailable));
+    }
+    Ok(EventStart::Replay(after_id))
+}
+
+pub(super) fn decode_event_cursor(
+    cursors: &remora_durable::CursorCodec,
+    retention: std::time::Duration,
+    principal: &Principal,
+    token: &str,
+    now: OffsetDateTime,
+) -> Result<Result<i64, ResyncReasonDto>, PublicApiError> {
+    let cursor = match cursors.decode(token) {
+        Ok(cursor) => cursor,
+        Err(_) => return Ok(Err(ResyncReasonDto::CursorInvalid)),
+    };
+    let Some(after_id) = event_position(&cursor) else {
+        return Ok(Err(ResyncReasonDto::CursorInvalid));
+    };
+    let expected = remora_durable::CursorCodec::filter_fingerprint(&serde_json::json!({
+        "resource": "events",
+        "authorization": authorization_snapshot(principal)
+    }))?;
+    if cursor.subject != principal.id.as_str() || cursor.filter_fingerprint != expected {
+        return Ok(Err(ResyncReasonDto::AuthorizationChanged));
+    }
+    let retention = time::Duration::try_from(retention).map_err(|_| PublicApiError::internal())?;
+    if cursor.high_water.created_at + retention < now {
+        return Ok(Err(ResyncReasonDto::CursorExpired));
+    }
+    Ok(Ok(after_id))
+}
+
+pub(super) fn event_position(cursor: &KeysetCursor) -> Option<i64> {
+    let high_water = cursor
+        .high_water
+        .id
+        .strip_prefix("event-")?
+        .parse::<i64>()
+        .ok()?;
+    let after = cursor.after.as_ref()?;
+    let after_id = after.id.strip_prefix("event-")?.parse::<i64>().ok()?;
+    (high_water == after_id && after.created_at == cursor.high_water.created_at).then_some(after_id)
+}
+
+pub(super) async fn send_resync(
+    state: PublicApiState,
+    principal: Principal,
+    reason: ResyncReasonDto,
+    mut session: actix_ws::Session,
+) {
+    let Ok(cursor) = event_cursor(&state, &principal).await else {
+        let _ = session.close(None).await;
+        return;
+    };
+    let Ok(frame) = resync_frame(reason, 1, &cursor, OffsetDateTime::now_utc()) else {
+        let _ = session.close(None).await;
+        return;
+    };
+    let _ = send_frame(&mut session, &frame).await;
+    let _ = session.close(None).await;
+}
+
+pub(super) async fn run_connection(
+    state: PublicApiState,
+    principal: Principal,
+    mut after_id: i64,
+    mut session: actix_ws::Session,
+    mut messages: actix_ws::MessageStream,
+) {
+    let mut sequence = 0_u64;
+    let mut poll = tokio::time::interval(state.event_poll_interval);
+    let mut heartbeat = tokio::time::interval(state.event_heartbeat_interval);
+    poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    heartbeat.tick().await;
+    loop {
+        tokio::select! {
+            message = messages.next() => {
+                match message {
+                    Some(Ok(actix_ws::Message::Ping(bytes))) => {
+                        if session.pong(&bytes).await.is_err() {
+                            break;
+                        }
+                    }
+                    Some(Ok(actix_ws::Message::Pong(_))) => {}
+                    Some(Ok(actix_ws::Message::Close(reason))) => {
+                        let _ = session.close(reason).await;
+                        return;
+                    }
+                    Some(Err(_)) | None => break,
+                    Some(Ok(_)) => break,
+                }
+            }
+            _ = poll.tick() => {
+                if poll_progress(&state).await.is_err() {
+                    break;
+                }
+                match replay_events(
+                    &state,
+                    &principal,
+                    &mut after_id,
+                    &mut sequence,
+                    &mut session,
+                ).await {
+                    Ok(ReplayStatus::Current) => {}
+                    Ok(ReplayStatus::HistoryUnavailable) => {
+                        let now = OffsetDateTime::now_utc();
+                        let Ok(cursor) = event_cursor(&state, &principal).await else {
+                            break;
+                        };
+                        let Ok(frame) = resync_frame(
+                            ResyncReasonDto::HistoryUnavailable,
+                            next_sequence(&mut sequence),
+                            &cursor,
+                            now,
+                        ) else {
+                            break;
+                        };
+                        let _ = send_frame(&mut session, &frame).await;
+                        break;
+                    }
+                    Err(()) => break,
+                }
+            }
+            _ = heartbeat.tick() => {
+                let now = OffsetDateTime::now_utc();
+                let Ok(cursor) = encode_event_cursor(&state, &principal, (after_id, now)) else {
+                    break;
+                };
+                let Ok(frame) = heartbeat_frame(next_sequence(&mut sequence), &cursor, now) else {
+                    break;
+                };
+                if send_frame(&mut session, &frame).await.is_err() {
+                    break;
+                }
+            }
+        }
+    }
+    let _ = session.close(None).await;
+}
+
+pub(super) async fn replay_events(
+    state: &PublicApiState,
+    principal: &Principal,
+    after_id: &mut i64,
+    sequence: &mut u64,
+    session: &mut actix_ws::Session,
+) -> Result<ReplayStatus, ()> {
+    for _ in 0..10 {
+        let batch = sqlx::query_as::<_, ReplayBatchRow>(
+            "SELECT history.floor_outbox_id, events.id, events.topic,
+                    events.payload, events.created_at
+             FROM live_event_history_state history
+             LEFT JOIN LATERAL (
+                 SELECT id, topic, payload, created_at
+                 FROM outbox_messages
+                 WHERE id > $1 AND topic = ANY($2)
+                 ORDER BY id
+                 LIMIT $3
+             ) events ON TRUE
+             WHERE history.singleton
+             ORDER BY events.id",
+        )
+        .bind(*after_id)
+        .bind(&EVENT_TOPICS[..])
+        .bind(EVENT_BATCH)
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| ())?;
+        let Some(first) = batch.first() else {
+            return Err(());
+        };
+        if *after_id < first.floor_outbox_id {
+            return Ok(ReplayStatus::HistoryUnavailable);
+        }
+        let rows = batch
+            .into_iter()
+            .filter_map(|row| {
+                Some(ReplayRow {
+                    id: row.id?,
+                    topic: row.topic?,
+                    payload: row.payload?,
+                    created_at: row.created_at?,
+                })
+            })
+            .collect::<Vec<_>>();
+        let count = rows.len();
+        for row in rows {
+            *after_id = row.id;
+            let now = OffsetDateTime::now_utc();
+            let cursor = encode_event_cursor(state, principal, (row.id, now)).map_err(|_| ())?;
+            let candidate_sequence = sequence.saturating_add(1);
+            let frame = match row.topic.as_str() {
+                "job.updated" => {
+                    project_job_event(state, principal, &row, candidate_sequence, &cursor).await
+                }
+                "deletion.updated" => {
+                    project_deletion_event(state, principal, &row, candidate_sequence, &cursor)
+                        .await
+                }
+                _ => None,
+            };
+            if let Some(frame) = frame {
+                *sequence = candidate_sequence;
+                send_frame(session, &frame).await?;
+            }
+        }
+        if count < EVENT_BATCH as usize {
+            break;
+        }
+    }
+    Ok(ReplayStatus::Current)
+}

crates/remora-api/src/public/events/frames.rs

@@ -0,0 +1,197 @@
+use super::*;
+
+pub(super) fn job_frame(
+    row: &JobRow,
+    progress: Option<&ProgressFingerprint>,
+    event_id: OpaqueId,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    let authoritative_attempt = row
+        .authority_attempt_id
+        .as_deref()
+        .map(|attempt_id| {
+            Ok::<_, PublicApiError>(EventAttemptDto {
+                attempt_id: parse_id(attempt_id)?,
+                fence: PositiveIntegerDto::new(
+                    u64::try_from(row.current_fence).map_err(|_| PublicApiError::internal())?,
+                )
+                .map_err(|_| PublicApiError::internal())?,
+            })
+        })
+        .transpose()?;
+    let progress = progress
+        .filter(|progress| {
+            row.authority_attempt_id.as_deref() == Some(progress.attempt_id.as_str())
+                && u64::try_from(row.current_fence).ok() == Some(progress.fence)
+        })
+        .map(event_progress)
+        .transpose()?;
+    Ok(EventFrameDto::JobChanged(JobChangedDto {
+        schema_version: SchemaVersionDto,
+        event_id,
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        job_id: parse_id(row.job_id.clone())?,
+        job_revision: positive(
+            u64::try_from(row.revision).map_err(|_| PublicApiError::internal())?,
+        )?,
+        phase: parse_phase(&row.phase)?,
+        lifecycle: parse_lifecycle(&row.lifecycle)?,
+        authoritative_attempt: NullableDto(authoritative_attempt),
+        progress: NullableDto(progress),
+    }))
+}
+
+pub(super) fn deletion_frame(
+    job_id: OpaqueId,
+    operation_id: OpaqueId,
+    state: DeletionStateDto,
+    event_id: OpaqueId,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::DeletionChanged(DeletionChangedDto {
+        schema_version: SchemaVersionDto,
+        event_id,
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        job_id,
+        operation_id,
+        state,
+    }))
+}
+
+pub(super) fn resync_frame(
+    reason: ResyncReasonDto,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::ResyncRequired(ResyncRequiredDto {
+        schema_version: SchemaVersionDto,
+        event_id: request_id(),
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        reason,
+    }))
+}
+
+pub(super) fn heartbeat_frame(
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::Heartbeat(HeartbeatDto {
+        schema_version: SchemaVersionDto,
+        event_id: request_id(),
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+    }))
+}
+
+pub(super) fn event_progress(
+    progress: &ProgressFingerprint,
+) -> Result<EventProgressDto, PublicApiError> {
+    Ok(EventProgressDto {
+        attempt_id: progress.attempt_id.clone(),
+        fence: positive(progress.fence)?,
+        completed_bytes: safe(progress.completed_bytes)?,
+        total_bytes: progress
+            .total_bytes
+            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
+            .transpose()?,
+        completed_members: safe(progress.completed_members)?,
+        total_members: progress
+            .total_members
+            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
+            .transpose()?,
+    })
+}
+
+pub(super) fn parse_payload_id(payload: &Value, field: &str) -> Option<OpaqueId> {
+    payload
+        .get(field)?
+        .as_str()
+        .and_then(|value| OpaqueId::parse(value).ok())
+}
+
+pub(super) fn stable_event_id(id: i64) -> OpaqueId {
+    OpaqueId::parse(format!("event_{id:016}")).expect("positive outbox ids form opaque event ids")
+}
+
+pub(super) fn parse_phase(value: &str) -> Result<JobPhase, PublicApiError> {
+    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(PublicApiError::internal()),
+    }
+}
+
+pub(super) fn parse_lifecycle(value: &str) -> Result<JobLifecycle, PublicApiError> {
+    match value {
+        "visible" => Ok(JobLifecycle::Visible),
+        "soft_deleted" => Ok(JobLifecycle::SoftDeleted),
+        "hard_deleting" => Ok(JobLifecycle::HardDeleting),
+        "hard_deleted" => Ok(JobLifecycle::HardDeleted),
+        _ => Err(PublicApiError::internal()),
+    }
+}
+
+pub(super) fn parse_deletion_state(value: &str) -> Option<DeletionStateDto> {
+    match value {
+        "queued" => Some(DeletionStateDto::Queued),
+        "running" => Some(DeletionStateDto::Running),
+        "retry_wait" => Some(DeletionStateDto::RetryWait),
+        "blocked" => Some(DeletionStateDto::Blocked),
+        "completed" => Some(DeletionStateDto::Completed),
+        _ => None,
+    }
+}
+
+pub(super) fn positive(value: u64) -> Result<PositiveIntegerDto, PublicApiError> {
+    PositiveIntegerDto::new(value).map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn safe(value: u64) -> Result<SafeIntegerDto, PublicApiError> {
+    SafeIntegerDto::new(value).map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn event_cursor_dto(value: &str) -> Result<CursorDto, PublicApiError> {
+    CursorDto::parse(value).map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn event_time(value: OffsetDateTime) -> Result<DateTimeDto, PublicApiError> {
+    DateTimeDto::parse(
+        value
+            .format(&Rfc3339)
+            .map_err(|_| PublicApiError::internal())?,
+    )
+    .map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn next_sequence(sequence: &mut u64) -> u64 {
+    *sequence = sequence.saturating_add(1);
+    *sequence
+}
+
+pub(super) async fn send_frame(
+    session: &mut actix_ws::Session,
+    frame: &EventFrameDto,
+) -> Result<(), ()> {
+    session
+        .text(serde_json::to_string(frame).map_err(|_| ())?)
+        .await
+        .map_err(|_| ())
+}

crates/remora-api/src/public/events/progress.rs

@@ -0,0 +1,176 @@
+use super::*;
+
+pub(super) fn progress_fingerprint(
+    snapshot: remora_attempt::ProgressSnapshot,
+) -> ProgressFingerprint {
+    ProgressFingerprint {
+        attempt_id: snapshot.attempt_id,
+        fence: snapshot.fence,
+        completed_bytes: snapshot.progress.completed_bytes,
+        total_bytes: snapshot.progress.total_bytes,
+        completed_members: snapshot.progress.completed_members,
+        total_members: snapshot.progress.total_members,
+    }
+}
+
+pub(super) async fn poll_progress(state: &PublicApiState) -> Result<(), ()> {
+    let delay = state.event_poll_interval.as_secs_f64();
+    let claimed = sqlx::query_scalar::<_, bool>(
+        "UPDATE live_event_history_state
+         SET progress_poll_not_before = now() + ($1 * interval '1 second')
+         WHERE singleton AND progress_poll_not_before <= now()
+         RETURNING singleton",
+    )
+    .bind(delay)
+    .fetch_optional(&state.pool)
+    .await
+    .map_err(|_| ())?
+    .is_some();
+    if !claimed {
+        return Ok(());
+    }
+    let mut after_job = None::<String>;
+    loop {
+        let job_ids: Vec<String> = sqlx::query_scalar(
+            "SELECT job_id FROM jobs
+             WHERE phase IN ('running', 'cancelling')
+               AND ($1::text IS NULL OR job_id > $1)
+             ORDER BY job_id
+             LIMIT 200",
+        )
+        .bind(after_job.as_deref())
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| ())?;
+        if job_ids.is_empty() {
+            break;
+        }
+        let count = job_ids.len();
+        for job_id in job_ids {
+            after_job = Some(job_id.clone());
+            let Some(row) = load_job_row(&state.pool, &parse_id(job_id.clone()).map_err(|_| ())?)
+                .await
+                .map_err(|_| ())?
+            else {
+                continue;
+            };
+            let progress = current_progress(state, &row)
+                .await
+                .ok()
+                .flatten()
+                .map(progress_fingerprint);
+            record_progress_change(&state.pool, &row, progress.as_ref())
+                .await
+                .map_err(|_| ())?;
+        }
+        if count < 200 {
+            break;
+        }
+    }
+    Ok(())
+}
+
+pub(super) async fn record_progress_change(
+    pool: &PgPool,
+    row: &JobRow,
+    progress: Option<&ProgressFingerprint>,
+) -> Result<(), PublicApiError> {
+    let Some(attempt_id) = row.authority_attempt_id.as_deref() else {
+        return Ok(());
+    };
+    let mut tx = pool.begin().await?;
+    let now = OffsetDateTime::now_utc();
+    let changed = if let Some(progress) = progress {
+        sqlx::query_scalar::<_, String>(
+            "INSERT INTO live_progress_event_state (
+                job_id, attempt_id, fence, progress_present, completed_bytes,
+                total_bytes, completed_members, total_members, updated_at
+             ) VALUES ($1, $2, $3, TRUE, $4, $5, $6, $7, $8)
+             ON CONFLICT (job_id) DO UPDATE
+             SET attempt_id = EXCLUDED.attempt_id,
+                 fence = EXCLUDED.fence,
+                 progress_present = TRUE,
+                 completed_bytes = EXCLUDED.completed_bytes,
+                 total_bytes = EXCLUDED.total_bytes,
+                 completed_members = EXCLUDED.completed_members,
+                 total_members = EXCLUDED.total_members,
+                 updated_at = EXCLUDED.updated_at
+             WHERE (
+                 live_progress_event_state.attempt_id,
+                 live_progress_event_state.fence,
+                 live_progress_event_state.progress_present,
+                 live_progress_event_state.completed_bytes,
+                 live_progress_event_state.total_bytes,
+                 live_progress_event_state.completed_members,
+                 live_progress_event_state.total_members
+             ) IS DISTINCT FROM (
+                 EXCLUDED.attempt_id,
+                 EXCLUDED.fence,
+                 TRUE,
+                 EXCLUDED.completed_bytes,
+                 EXCLUDED.total_bytes,
+                 EXCLUDED.completed_members,
+                 EXCLUDED.total_members
+             )
+             RETURNING job_id",
+        )
+        .bind(&row.job_id)
+        .bind(progress.attempt_id.as_str())
+        .bind(i64::try_from(progress.fence).map_err(|_| PublicApiError::internal())?)
+        .bind(i64::try_from(progress.completed_bytes).map_err(|_| PublicApiError::internal())?)
+        .bind(
+            progress
+                .total_bytes
+                .map(i64::try_from)
+                .transpose()
+                .map_err(|_| PublicApiError::internal())?,
+        )
+        .bind(i64::try_from(progress.completed_members).map_err(|_| PublicApiError::internal())?)
+        .bind(
+            progress
+                .total_members
+                .map(i64::try_from)
+                .transpose()
+                .map_err(|_| PublicApiError::internal())?,
+        )
+        .bind(now)
+        .fetch_optional(tx.as_mut())
+        .await?
+        .is_some()
+    } else {
+        sqlx::query_scalar::<_, String>(
+            "UPDATE live_progress_event_state
+             SET attempt_id = $2, fence = $3, progress_present = FALSE,
+                 completed_bytes = NULL, total_bytes = NULL,
+                 completed_members = NULL, total_members = NULL, updated_at = $4
+             WHERE job_id = $1
+               AND (
+                   attempt_id IS DISTINCT FROM $2
+                   OR fence IS DISTINCT FROM $3
+                   OR progress_present
+                   OR completed_bytes IS NOT NULL
+                   OR total_bytes IS NOT NULL
+                   OR completed_members IS NOT NULL
+                   OR total_members IS NOT NULL
+               )
+             RETURNING job_id",
+        )
+        .bind(&row.job_id)
+        .bind(attempt_id)
+        .bind(row.current_fence)
+        .bind(now)
+        .fetch_optional(tx.as_mut())
+        .await?
+        .is_some()
+    };
+    if changed {
+        remora_durable::enqueue(
+            &mut tx,
+            "job.updated",
+            serde_json::json!({ "job_id": row.job_id }),
+        )
+        .await?;
+    }
+    tx.commit().await?;
+    Ok(())
+}

crates/remora-api/src/public/events/projections.rs

@@ -0,0 +1,56 @@
+use super::*;
+
+pub(super) async fn project_job_event(
+    state: &PublicApiState,
+    principal: &Principal,
+    event: &ReplayRow,
+    sequence: u64,
+    cursor: &str,
+) -> Option<EventFrameDto> {
+    let job_id = parse_payload_id(&event.payload, "job_id")?;
+    let row = load_job_row(&state.pool, &job_id).await.ok()??;
+    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
+    let progress = current_progress(state, &row)
+        .await
+        .ok()
+        .flatten()
+        .map(progress_fingerprint);
+    job_frame(
+        &row,
+        progress.as_ref(),
+        stable_event_id(event.id),
+        sequence,
+        cursor,
+        event.created_at,
+    )
+    .ok()
+}
+
+pub(super) async fn project_deletion_event(
+    state: &PublicApiState,
+    principal: &Principal,
+    event: &ReplayRow,
+    sequence: u64,
+    cursor: &str,
+) -> Option<EventFrameDto> {
+    let operation_id = parse_payload_id(&event.payload, "operation_id")?;
+    let (job_id, state_name): (String, String) =
+        sqlx::query_as("SELECT job_id, state FROM deletion_operations WHERE operation_id = $1")
+            .bind(operation_id.as_str())
+            .fetch_optional(&state.pool)
+            .await
+            .ok()??;
+    let job_id = parse_id(job_id).ok()?;
+    let row = load_job_row(&state.pool, &job_id).await.ok()??;
+    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
+    deletion_frame(
+        job_id,
+        operation_id,
+        parse_deletion_state(&state_name)?,
+        stable_event_id(event.id),
+        sequence,
+        cursor,
+        event.created_at,
+    )
+    .ok()
+}

crates/remora-api/src/public/events/tickets.rs

@@ -0,0 +1,129 @@
+use super::*;
+
+pub(in crate::public) async fn issue_ticket(
+    state: web::Data<PublicApiState>,
+    request: HttpRequest,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&request).await?;
+    let origin = allowed_origin(&state, &request)?;
+    let ticket = random_ticket();
+    let digest = ticket_digest(ticket.as_str());
+    let mut tx = state.pool.begin().await?;
+    sqlx::query("DELETE FROM live_event_tickets WHERE expires_at <= now()")
+        .execute(tx.as_mut())
+        .await?;
+    let expires_at: OffsetDateTime = sqlx::query_scalar(
+        "INSERT INTO live_event_tickets (
+            ticket_digest, principal, origin, created_at, expires_at
+         ) VALUES (
+            $1, $2, $3, now(),
+            now() + $4 * interval '1 second'
+         )
+         RETURNING expires_at",
+    )
+    .bind(digest.as_slice())
+    .bind(Json(ticket_principal_projection(&principal)))
+    .bind(&origin)
+    .bind(state.event_ticket_ttl.as_secs_f64())
+    .fetch_one(tx.as_mut())
+    .await?;
+    tx.commit().await?;
+    Ok(HttpResponse::Created()
+        .insert_header((header::CACHE_CONTROL, "no-store"))
+        .insert_header((header::PRAGMA, "no-cache"))
+        .json(EventTicketDto {
+            ticket,
+            expires_at: event_time(expires_at)?,
+        }))
+}
+
+pub(super) fn allowed_origin(
+    state: &PublicApiState,
+    request: &HttpRequest,
+) -> Result<String, PublicApiError> {
+    let mut origins = request.headers().get_all(header::ORIGIN);
+    let origin = origins
+        .next()
+        .and_then(|value| value.to_str().ok())
+        .ok_or_else(PublicApiError::forbidden)?;
+    if origins.next().is_some() || !state.event_allowed_origins.contains(origin) {
+        return Err(PublicApiError::forbidden());
+    }
+    Ok(origin.to_owned())
+}
+
+pub(super) async fn redeem_ticket(
+    state: &PublicApiState,
+    request: &HttpRequest,
+    ticket: &str,
+) -> Result<Principal, PublicApiError> {
+    let origin = allowed_origin(state, request)?;
+    let digest = ticket_digest(ticket);
+    let projection: Option<Json<TicketPrincipal>> = sqlx::query_scalar(
+        "DELETE FROM live_event_tickets
+         WHERE ticket_digest = $1 AND origin = $2 AND expires_at > now()
+         RETURNING principal",
+    )
+    .bind(digest.as_slice())
+    .bind(origin)
+    .fetch_optional(&state.pool)
+    .await?;
+    projection
+        .ok_or_else(PublicApiError::unauthenticated)
+        .and_then(|projection| ticket_principal(projection.0))
+}
+
+pub(super) fn ticket_principal_projection(principal: &Principal) -> TicketPrincipal {
+    TicketPrincipal {
+        id: principal.id.as_str().to_owned(),
+        issuer: principal.issuer.clone(),
+        subject: principal.subject.clone(),
+        administrator: principal.administrator,
+        scopes: principal
+            .scopes
+            .iter()
+            .map(|scope| TicketScope {
+                store: scope.store.as_str().to_owned(),
+                bucket: scope.bucket.as_str().to_owned(),
+                prefix: scope.prefix.canonical(),
+            })
+            .collect(),
+    }
+}
+
+pub(super) fn ticket_principal(projection: TicketPrincipal) -> Result<Principal, PublicApiError> {
+    let scopes = projection
+        .scopes
+        .into_iter()
+        .map(|scope| {
+            Ok(StorageScope {
+                store: TargetStoreId::parse(scope.store).map_err(|_| PublicApiError::internal())?,
+                bucket: BucketName::parse(scope.bucket).map_err(|_| PublicApiError::internal())?,
+                prefix: TargetPrefix::parse(&scope.prefix, std::iter::empty())
+                    .map_err(|_| PublicApiError::internal())?,
+            })
+        })
+        .collect::<Result<Vec<_>, PublicApiError>>()?;
+    Ok(Principal {
+        id: OpaqueId::parse(projection.id).map_err(|_| PublicApiError::internal())?,
+        issuer: projection.issuer,
+        subject: projection.subject,
+        administrator: projection.administrator,
+        scopes,
+    })
+}
+
+pub(super) fn random_ticket() -> OpaqueId {
+    let mut bytes = [0_u8; 32];
+    getrandom::fill(&mut bytes).expect("operating system randomness is required");
+    let mut value = String::from("t");
+    for byte in bytes {
+        use std::fmt::Write as _;
+        write!(value, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    OpaqueId::parse(value).expect("random event ticket is valid")
+}
+
+pub(super) fn ticket_digest(ticket: &str) -> [u8; 32] {
+    Sha256::digest(ticket.as_bytes()).into()
+}

crates/remora-api/src/public/job_store.rs

@@ -0,0 +1,354 @@
+use super::*;
+
+pub(super) async fn idempotent_job_mutation(
+    state: &PublicApiState,
+    principal: &Principal,
+    job_id: &OpaqueId,
+    key: &str,
+    operation: &'static str,
+    action: JobMutationAction,
+) -> Result<Value, PublicApiError> {
+    let request_fingerprint =
+        fingerprint(&json!({ "job_id": job_id })).map_err(|_| PublicApiError::internal())?;
+    let state_for_projection = state.clone();
+    let principal_for_projection = principal.clone();
+    let job_for_projection = job_id.clone();
+    let outcome = execute_mutation(
+        &state.pool,
+        MutationScope {
+            subject: &principal.id,
+            operation,
+            key,
+        },
+        operation,
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                match action {
+                    JobMutationAction::Cancel => {
+                        AttemptService::request_cancel_in_tx(tx, &job_for_projection).await?;
+                    }
+                    JobMutationAction::SoftDelete => {
+                        DeletionService::soft_delete_in_tx(
+                            tx,
+                            &principal_for_projection,
+                            &job_for_projection,
+                        )
+                        .await
+                        .map_err(DurableError::effect)?;
+                    }
+                    JobMutationAction::Restore => {
+                        DeletionService::restore_in_tx(
+                            tx,
+                            &principal_for_projection,
+                            &job_for_projection,
+                        )
+                        .await
+                        .map_err(DurableError::effect)?;
+                    }
+                    JobMutationAction::Revoke { grant_id } => {
+                        remora_admission::AdmissionService::revoke_execution_grant_in_tx(
+                            tx,
+                            &principal_for_projection,
+                            &grant_id,
+                        )
+                        .await?;
+                        AttemptService::request_cancel_in_tx(tx, &job_for_projection).await?;
+                    }
+                }
+                let projection = load_authorized_job_in_tx(
+                    tx,
+                    &state_for_projection,
+                    &principal_for_projection,
+                    &job_for_projection,
+                    true,
+                )
+                .await
+                .map_err(|error| DurableError::Effect(error.to_string()))?;
+                Ok((MutationStatus::Accepted, projection))
+            })
+        },
+    )
+    .await?;
+    if outcome.disposition == MutationDisposition::Conflict {
+        Err(PublicApiError::idempotency_conflict())
+    } else {
+        Ok(outcome.response)
+    }
+}
+
+pub(super) async fn load_job_row(
+    pool: &PgPool,
+    job_id: &OpaqueId,
+) -> Result<Option<JobRow>, PublicApiError> {
+    Ok(sqlx::query_as::<_, JobRow>(
+        "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
+                target_store, target_bucket, target_prefix, execution_grant_id,
+                authority_attempt_id,
+                (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
+                    AS authority_worker_incarnation,
+                current_fence, checkpoint, committed_result, failure,
+                deletion_operation_id, hard_delete_started_at, created_at, updated_at
+         FROM jobs WHERE job_id = $1",
+    )
+    .bind(job_id.as_str())
+    .fetch_optional(pool)
+    .await?)
+}
+
+pub(super) async fn load_job_row_in_tx(
+    tx: &mut PgTx<'_>,
+    job_id: &OpaqueId,
+) -> Result<Option<JobRow>, PublicApiError> {
+    Ok(sqlx::query_as::<_, JobRow>(
+        "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
+                target_store, target_bucket, target_prefix, execution_grant_id,
+                authority_attempt_id,
+                (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
+                    AS authority_worker_incarnation,
+                current_fence, checkpoint, committed_result, failure,
+                deletion_operation_id, hard_delete_started_at, created_at, updated_at
+         FROM jobs WHERE job_id = $1",
+    )
+    .bind(job_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?)
+}
+
+pub(super) async fn load_authorized_job_in_tx(
+    tx: &mut PgTx<'_>,
+    state: &PublicApiState,
+    principal: &Principal,
+    job_id: &OpaqueId,
+    include_deleted: bool,
+) -> Result<Value, PublicApiError> {
+    let row = load_job_row_in_tx(tx, job_id)
+        .await?
+        .ok_or_else(PublicApiError::not_found)?;
+    if !include_deleted && row.lifecycle != "visible" {
+        return Err(PublicApiError::not_found());
+    }
+    authorize_row(principal, &row, ManagementOperation::Read)?;
+    let cursor = event_cursor_in_tx(tx, state, principal).await?;
+    job_row_projection(&row, None, &cursor)
+}
+
+pub(super) async fn load_authorized_job(
+    state: &PublicApiState,
+    principal: &Principal,
+    job_id: &OpaqueId,
+    include_deleted: bool,
+) -> Result<Value, PublicApiError> {
+    let cursor = event_cursor(state, principal).await?;
+    let row = load_job_row(&state.pool, job_id)
+        .await?
+        .ok_or_else(PublicApiError::not_found)?;
+    if !include_deleted && row.lifecycle != "visible" {
+        return Err(PublicApiError::not_found());
+    }
+    authorize_row(principal, &row, ManagementOperation::Read)?;
+    let progress = current_progress(state, &row).await?;
+    job_row_projection(&row, progress, &cursor)
+}
+
+pub(super) async fn load_visible_job_row(
+    state: &PublicApiState,
+    principal: &Principal,
+    job_id: &OpaqueId,
+) -> Result<JobRow, PublicApiError> {
+    let row = load_job_row(&state.pool, job_id)
+        .await?
+        .ok_or_else(PublicApiError::not_found)?;
+    if row.lifecycle != "visible" {
+        return Err(PublicApiError::not_found());
+    }
+    authorize_row(principal, &row, ManagementOperation::Read)?;
+    Ok(row)
+}
+
+pub(super) async fn authorize_job(
+    state: &PublicApiState,
+    principal: &Principal,
+    job_id: &OpaqueId,
+    operation: ManagementOperation,
+) -> Result<JobRow, PublicApiError> {
+    let row = load_job_row(&state.pool, job_id)
+        .await?
+        .ok_or_else(PublicApiError::not_found)?;
+    authorize_row(principal, &row, operation)?;
+    Ok(row)
+}
+
+pub(super) fn authorize_row(
+    principal: &Principal,
+    row: &JobRow,
+    operation: ManagementOperation,
+) -> Result<(), PublicApiError> {
+    let resource = AuthorizedResource {
+        owner: parse_id(row.owner_id.clone())?,
+        target: row_target(row)?,
+    };
+    if may_manage(principal, &resource, operation) {
+        Ok(())
+    } else {
+        Err(PublicApiError::not_found())
+    }
+}
+
+pub(super) async fn load_job_page_rows(
+    pool: &PgPool,
+    principal: &Principal,
+    include_deleted: bool,
+    cursor: Option<&KeysetCursor>,
+    wanted: u32,
+) -> Result<Vec<JobRow>, PublicApiError> {
+    let mut after = cursor.and_then(|value| value.after.clone());
+    let high_water = cursor.map(|value| value.high_water.clone());
+    let mut visible = Vec::new();
+    loop {
+        let rows = sqlx::query_as::<_, JobRow>(
+            "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
+                    target_store, target_bucket, target_prefix, execution_grant_id,
+                    authority_attempt_id,
+                    (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
+                        AS authority_worker_incarnation,
+                    current_fence, checkpoint, committed_result, failure,
+                    deletion_operation_id, hard_delete_started_at, created_at, updated_at
+             FROM jobs
+             WHERE ($1::text IS NULL OR owner_id = $1)
+               AND ($2 OR lifecycle = 'visible')
+               AND ($3::timestamptz IS NULL OR (created_at, job_id) <= ($3, $4))
+               AND ($5::timestamptz IS NULL OR (created_at, job_id) < ($5, $6))
+             ORDER BY created_at DESC, job_id DESC
+             LIMIT 200",
+        )
+        .bind((!principal.administrator).then(|| principal.id.as_str()))
+        .bind(include_deleted)
+        .bind(high_water.as_ref().map(|value| value.created_at))
+        .bind(high_water.as_ref().map(|value| value.id.as_str()))
+        .bind(after.as_ref().map(|value| value.created_at))
+        .bind(after.as_ref().map(|value| value.id.as_str()))
+        .fetch_all(pool)
+        .await?;
+        if rows.is_empty() {
+            break;
+        }
+        let count = rows.len();
+        for row in rows {
+            after = Some(position(row.created_at, &row.job_id));
+            if authorize_row(principal, &row, ManagementOperation::List).is_ok() {
+                visible.push(row);
+                if visible.len() >= wanted as usize {
+                    return Ok(visible);
+                }
+            }
+        }
+        if count < 200 {
+            break;
+        }
+    }
+    Ok(visible)
+}
+
+pub(super) async fn current_progress(
+    state: &PublicApiState,
+    row: &JobRow,
+) -> Result<Option<ProgressSnapshot>, PublicApiError> {
+    let Some(authority_attempt_id) = row.authority_attempt_id.as_deref() else {
+        return Ok(None);
+    };
+    let Ok(progress) = state
+        .attempts
+        .progress_snapshot(&parse_id(row.job_id.clone())?)
+        .await
+    else {
+        return Ok(None);
+    };
+    Ok(progress.filter(|progress| {
+        progress.attempt_id.as_str() == authority_attempt_id
+            && i64::try_from(progress.fence).ok() == Some(row.current_fence)
+    }))
+}
+
+pub(super) async fn event_cursor(
+    state: &PublicApiState,
+    principal: &Principal,
+) -> Result<String, PublicApiError> {
+    let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
+        "SELECT GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                ),
+                now()
+         FROM live_event_history_state history
+         WHERE history.singleton",
+    )
+    .fetch_one(&state.pool)
+    .await?;
+    encode_event_cursor(state, principal, high_water)
+}
+
+pub(super) async fn event_cursor_in_tx(
+    tx: &mut PgTx<'_>,
+    state: &PublicApiState,
+    principal: &Principal,
+) -> Result<String, PublicApiError> {
+    let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
+        "SELECT GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                ),
+                now()
+         FROM live_event_history_state history
+         WHERE history.singleton",
+    )
+    .fetch_one(tx.as_mut())
+    .await?;
+    encode_event_cursor(state, principal, high_water)
+}
+
+pub(super) fn encode_event_cursor(
+    state: &PublicApiState,
+    principal: &Principal,
+    high_water: (i64, OffsetDateTime),
+) -> Result<String, PublicApiError> {
+    let created_at = high_water.1;
+    let id = format!("event-{}", high_water.0);
+    let filter = CursorCodec::filter_fingerprint(&json!({
+        "resource": "events",
+        "authorization": authorization_snapshot(principal)
+    }))?;
+    state
+        .cursors
+        .encode(&first_page_cursor(
+            principal.id.as_str(),
+            filter,
+            KeysetPosition {
+                created_at,
+                id: id.clone(),
+            },
+            KeysetPosition { created_at, id },
+        ))
+        .map_err(Into::into)
+}
+
+pub(super) fn authorization_snapshot(principal: &Principal) -> Value {
+    let mut scopes = principal
+        .scopes
+        .iter()
+        .map(|scope| {
+            (
+                scope.store.as_str(),
+                scope.bucket.as_str(),
+                scope.prefix.canonical(),
+            )
+        })
+        .collect::<Vec<_>>();
+    scopes.sort_unstable();
+    scopes.dedup();
+    json!({
+        "administrator": principal.administrator,
+        "scopes": scopes
+    })
+}

crates/remora-api/src/public/jobs.rs

@@ -0,0 +1,281 @@
+use super::*;
+
+pub(super) async fn submit_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    request: web::Json<JobSubmissionDto>,
+) -> Result<HttpResponse, PublicApiError> {
+    validate_request("JobSubmission", &*request)?;
+    let principal = state.principal(&http).await?;
+    let key = idempotency_key(&http)?;
+    let request = request.into_inner();
+    let request_fingerprint =
+        fingerprint(&request).map_err(|_| PublicApiError::invalid_request())?;
+    let admission = state.admission.clone();
+    let state_for_projection = state.get_ref().clone();
+    let principal_for_effect = principal.clone();
+    let outcome = execute_mutation(
+        &state.pool,
+        MutationScope {
+            subject: &principal.id,
+            operation: "job.submit",
+            key,
+        },
+        "submit_job",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                match admission
+                    .submit_in_tx(tx, &principal_for_effect, request)
+                    .await?
+                {
+                    AdmissionDecision::Accepted(accepted) => {
+                        let projection = load_authorized_job_in_tx(
+                            tx,
+                            &state_for_projection,
+                            &principal_for_effect,
+                            &accepted.job_id,
+                            false,
+                        )
+                        .await
+                        .map_err(|error| DurableError::Effect(error.to_string()))?;
+                        Ok((MutationStatus::Accepted, projection))
+                    }
+                    AdmissionDecision::Rejected { .. } => Ok((
+                        MutationStatus::Rejected,
+                        public_error_body(ApiErrorCodeDto::InvalidRequest, "invalid request"),
+                    )),
+                }
+            })
+        },
+    )
+    .await?;
+    if outcome.disposition == MutationDisposition::Conflict {
+        return Err(PublicApiError::idempotency_conflict());
+    }
+    if outcome.status == MutationStatus::Accepted {
+        let job_id = outcome
+            .response
+            .get("id")
+            .and_then(Value::as_str)
+            .ok_or_else(PublicApiError::internal)?;
+        let job_id = parse_id(job_id.to_owned())?;
+        authorize_job(&state, &principal, &job_id, ManagementOperation::Read).await?;
+    }
+    let response = if outcome.status == MutationStatus::Rejected {
+        HttpResponse::BadRequest()
+            .content_type("application/problem+json")
+            .json(outcome.response)
+    } else {
+        HttpResponse::Accepted().json(outcome.response)
+    };
+    Ok(with_idempotency_expiry(response))
+}
+
+pub(super) async fn get_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    let job = load_authorized_job(&state, &principal, &job_id, true).await?;
+    Ok(HttpResponse::Ok().json(job))
+}
+
+pub(super) async fn list_jobs(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    query: web::Query<JobPageQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let limit = page_limit(query.limit)?;
+    let filters = json!({
+        "resource": "jobs",
+        "include_deleted": query.include_deleted,
+        "authorization": authorization_snapshot(&principal)
+    });
+    let filter = CursorCodec::filter_fingerprint(&filters)?;
+    let cursor = query
+        .cursor
+        .as_deref()
+        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
+        .transpose()
+        .map_err(PublicApiError::from_cursor)?;
+    let event_cursor = event_cursor(&state, &principal).await?;
+    let mut rows = load_job_page_rows(
+        &state.pool,
+        &principal,
+        query.include_deleted,
+        cursor.as_ref(),
+        limit + 1,
+    )
+    .await?;
+    let has_more = rows.len() > limit as usize;
+    rows.truncate(limit as usize);
+    let mut items = Vec::with_capacity(rows.len());
+    for row in &rows {
+        items.push(job_row_projection(
+            row,
+            current_progress(&state, row).await?,
+            &event_cursor,
+        )?);
+    }
+    let next_cursor = if has_more {
+        let last = rows.last().ok_or_else(PublicApiError::internal)?;
+        let last = position(last.created_at, &last.job_id);
+        let cursor = match cursor.as_ref() {
+            Some(current) => next_page_cursor(current, last),
+            None => first_page_cursor(
+                principal.id.as_str(),
+                filter,
+                position(rows[0].created_at, &rows[0].job_id),
+                last,
+            ),
+        };
+        Some(state.cursors.encode(&cursor)?)
+    } else {
+        None
+    };
+    Ok(HttpResponse::Ok().json(json!({
+        "items": items,
+        "next_cursor": next_cursor,
+        "event_cursor": event_cursor,
+        "management_capabilities": {
+            "revoke_execution_grants": principal.administrator
+        }
+    })))
+}
+
+pub(super) async fn cancel_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    authorize_job(&state, &principal, &job_id, ManagementOperation::Cancel).await?;
+    let key = idempotency_key(&http)?.to_owned();
+    let job = idempotent_job_mutation(
+        &state,
+        &principal,
+        &job_id,
+        &key,
+        "job.cancel",
+        JobMutationAction::Cancel,
+    )
+    .await?;
+    Ok(with_idempotency_expiry(HttpResponse::Accepted().json(job)))
+}
+
+pub(super) async fn revoke_job(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    if !principal.administrator {
+        return Err(PublicApiError::forbidden());
+    }
+    let job_id = parse_id(path.into_inner())?;
+    let row = load_job_row(&state.pool, &job_id)
+        .await?
+        .ok_or_else(PublicApiError::not_found)?;
+    let key = idempotency_key(&http)?.to_owned();
+    let grant_id = parse_id(row.execution_grant_id.clone())?;
+    let job = idempotent_job_mutation(
+        &state,
+        &principal,
+        &job_id,
+        &key,
+        "job.revoke",
+        JobMutationAction::Revoke { grant_id },
+    )
+    .await?;
+    Ok(with_idempotency_expiry(HttpResponse::Accepted().json(job)))
+}
+
+pub(super) async fn list_job_attempts(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+    query: web::Query<PageQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    authorize_job(&state, &principal, &job_id, ManagementOperation::Read).await?;
+    let limit = page_limit(query.limit)?;
+    let filter = CursorCodec::filter_fingerprint(&json!({
+        "resource": "attempts",
+        "job_id": job_id,
+        "authorization": authorization_snapshot(&principal)
+    }))?;
+    let cursor = query
+        .cursor
+        .as_deref()
+        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
+        .transpose()
+        .map_err(PublicApiError::from_cursor)?;
+    let after = cursor
+        .as_ref()
+        .and_then(|value| value.after.as_ref())
+        .map(|value| value.id.parse::<i64>())
+        .transpose()
+        .map_err(|_| PublicApiError::cursor_invalid())?;
+    let rows = sqlx::query_as::<_, AttemptRow>(
+        "SELECT a.attempt_id, a.fence, a.worker_incarnation, a.state, a.failure,
+                CASE WHEN j.committed_result->>'attempt_id' = a.attempt_id
+                     THEN j.committed_result END AS result,
+                a.authorized_at,
+                COALESCE(a.committed_at, a.stopped_at, a.reaped_at, a.executing_at, a.authorized_at)
+                    AS updated_at
+         FROM attempts a
+         JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.job_id = $1 AND ($2::bigint IS NULL OR a.fence < $2)
+         ORDER BY a.fence DESC
+         LIMIT $3",
+    )
+    .bind(job_id.as_str())
+    .bind(after)
+    .bind(i64::from(limit) + 1)
+    .fetch_all(&state.pool)
+    .await?;
+    let has_more = rows.len() > limit as usize;
+    let visible = rows.iter().take(limit as usize).collect::<Vec<_>>();
+    let progress = state
+        .attempts
+        .progress_snapshot(&job_id)
+        .await
+        .unwrap_or(None);
+    let items = visible
+        .iter()
+        .map(|row| attempt_projection(row, progress.as_ref()))
+        .collect::<Result<Vec<_>, _>>()?;
+    let next_cursor = if has_more {
+        let last = visible.last().ok_or_else(PublicApiError::internal)?;
+        let next = KeysetPosition {
+            created_at: last.authorized_at,
+            id: last.fence.to_string(),
+        };
+        let cursor = match cursor.as_ref() {
+            Some(current) => next_page_cursor(current, next),
+            None => first_page_cursor(
+                principal.id.as_str(),
+                filter,
+                KeysetPosition {
+                    created_at: rows[0].authorized_at,
+                    id: rows[0].fence.to_string(),
+                },
+                next,
+            ),
+        };
+        Some(state.cursors.encode(&cursor)?)
+    } else {
+        None
+    };
+    Ok(HttpResponse::Ok().json(json!({
+        "items": items,
+        "next_cursor": next_cursor
+    })))
+}

crates/remora-api/src/public/projections.rs

@@ -0,0 +1,297 @@
+use super::*;
+
+pub(super) fn progress_projection(progress: &ProgressSnapshot) -> Result<Value, PublicApiError> {
+    Ok(json!({
+        "attempt_id": progress.attempt_id,
+        "fence": progress.fence,
+        "completed_bytes": progress.progress.completed_bytes,
+        "total_bytes": progress.progress.total_bytes,
+        "completed_members": progress.progress.completed_members,
+        "total_members": progress.progress.total_members,
+        "observed_at": format_time(progress.observed_at)?
+    }))
+}
+
+pub(super) fn job_row_projection(
+    row: &JobRow,
+    progress: Option<ProgressSnapshot>,
+    event_cursor: &str,
+) -> Result<Value, PublicApiError> {
+    let target = json!({
+        "store": row.target_store,
+        "bucket": row.target_bucket,
+        "prefix": row.target_prefix.0.join("/")
+    });
+    if row.lifecycle == "hard_deleted" || row.lifecycle == "hard_deleting" {
+        return Ok(json!({
+            "resource_kind": "tombstone",
+            "id": row.job_id,
+            "owner": row.owner_id,
+            "target": target,
+            "terminal_phase": phase_projection(&row.phase)?,
+            "deletion_operation_id": row.deletion_operation_id,
+            "created_at": format_time(row.created_at)?,
+            "hard_delete_started_at": row.hard_delete_started_at.map(format_time).transpose()?,
+            "updated_at": format_time(row.updated_at)?,
+            "event_cursor": event_cursor
+        }));
+    }
+    let authority = row
+        .authority_attempt_id
+        .as_ref()
+        .zip(row.authority_worker_incarnation.as_ref())
+        .map(|(attempt_id, worker_incarnation)| {
+            json!({
+                "attempt_id": attempt_id,
+                "fence": row.current_fence,
+                "worker_incarnation": worker_incarnation
+            })
+        });
+    let checkpoint = row
+        .checkpoint
+        .as_ref()
+        .map(|value| checkpoint_projection(&value.0))
+        .transpose()?;
+    let result = committed_result(row)?.as_ref().map(result_projection);
+    let failure = row
+        .failure
+        .as_ref()
+        .map(|value| failure_projection(&value.0))
+        .transpose()?;
+    Ok(json!({
+        "resource_kind": "job",
+        "id": row.job_id,
+        "owner": row.owner_id,
+        "label": row.label,
+        "revision": row.revision,
+        "phase": phase_projection(&row.phase)?,
+        "lifecycle": row.lifecycle,
+        "source": row.source_projection.0,
+        "target": target,
+        "authoritative_attempt": authority,
+        "progress": progress.as_ref().map(progress_projection).transpose()?,
+        "checkpoint": checkpoint,
+        "result": result,
+        "failure": failure,
+        "created_at": format_time(row.created_at)?,
+        "updated_at": format_time(row.updated_at)?,
+        "event_cursor": event_cursor
+    }))
+}
+
+pub(super) fn phase_projection(value: &str) -> Result<&'static str, PublicApiError> {
+    match value {
+        "queued" => Ok("Queued"),
+        "running" => Ok("Running"),
+        "retry_wait" => Ok("RetryWait"),
+        "cancelling" => Ok("Cancelling"),
+        "succeeded" => Ok("Succeeded"),
+        "failed" => Ok("Failed"),
+        "cancelled" => Ok("Cancelled"),
+        _ => Err(PublicApiError::internal()),
+    }
+}
+
+pub(super) fn checkpoint_projection(value: &Value) -> Result<Value, PublicApiError> {
+    let state = value
+        .get("state")
+        .and_then(Value::as_str)
+        .ok_or_else(PublicApiError::internal)?;
+    match state {
+        "reusable" => {
+            let checkpoint = value
+                .get("checkpoint")
+                .ok_or_else(PublicApiError::internal)?;
+            Ok(json!({
+                "reusable": true,
+                "validated_bytes": checkpoint.get("validated_bytes"),
+                "validated_members": checkpoint.get("validated_members")
+            }))
+        }
+        "complete" => Ok(json!({
+            "reusable": false,
+            "validated_bytes": value.get("validated_bytes"),
+            "validated_members": value.get("validated_members")
+        })),
+        _ => Err(PublicApiError::internal()),
+    }
+}
+
+pub(super) fn failure_projection(value: &Value) -> Result<Value, PublicApiError> {
+    let class = value
+        .get("class")
+        .and_then(Value::as_str)
+        .ok_or_else(PublicApiError::internal)?;
+    let disposition = value
+        .get("disposition")
+        .and_then(Value::as_str)
+        .ok_or_else(PublicApiError::internal)?;
+    Ok(json!({
+        "class": class,
+        "message": "job failed",
+        "retryable": disposition == "retryable"
+    }))
+}
+
+pub(super) fn row_target(row: &JobRow) -> Result<Target, PublicApiError> {
+    Ok(Target {
+        store: TargetStoreId::parse(row.target_store.clone())
+            .map_err(|_| PublicApiError::internal())?,
+        bucket: BucketName::parse(row.target_bucket.clone())
+            .map_err(|_| PublicApiError::internal())?,
+        prefix: TargetPrefix::parse(&row.target_prefix.0.join("/"), std::iter::empty())
+            .map_err(|_| PublicApiError::internal())?,
+    })
+}
+
+pub(super) fn committed_result(row: &JobRow) -> Result<Option<PublicationInput>, PublicApiError> {
+    let Some(value) = row.committed_result.as_ref() else {
+        return Ok(None);
+    };
+    let record: CommittedResultRecord =
+        serde_json::from_value(value.0.clone()).map_err(|_| PublicApiError::internal())?;
+    Ok(Some(PublicationInput {
+        attempt_id: record.attempt_id,
+        manifest_id: record.manifest_id,
+        schema_version: record.schema_version,
+        manifest: StoredObject {
+            identity: ObjectIdentity {
+                target_store_id: record.manifest.store,
+                bucket: record.manifest.bucket,
+                key: record.manifest.key,
+                version_id: record.manifest.version_id,
+            },
+            content_sha256: ContentSha256::parse(record.manifest.sha256)
+                .map_err(|_| PublicApiError::internal())?,
+            size: record.manifest.size,
+        },
+        member_count: record.member_count,
+        total_bytes: record.total_bytes,
+    }))
+}
+
+pub(super) fn result_projection(result: &CommittedResult) -> Value {
+    json!({
+        "kind": "manifest",
+        "manifest_id": result.manifest_id,
+        "schema_version": result.schema_version,
+        "manifest": object_identity_projection(
+            &result.manifest.identity.target_store_id,
+            &result.manifest.identity.bucket,
+            &result.manifest.identity.key,
+            result.manifest.identity.version_id.as_deref(),
+            result.manifest.content_sha256.as_str(),
+            result.manifest.size
+        ),
+        "member_count": result.member_count,
+        "total_bytes": result.total_bytes
+    })
+}
+
+pub(super) fn object_identity_projection(
+    store: &str,
+    bucket: &str,
+    key: &str,
+    version_id: Option<&str>,
+    checksum: &str,
+    size: u64,
+) -> Value {
+    let mut object = json!({
+        "store": store,
+        "bucket": bucket,
+        "key": key,
+        "checksum": checksum,
+        "size": size
+    });
+    if let Some(version_id) = version_id {
+        object["version_id"] = json!(version_id);
+    }
+    object
+}
+
+pub(super) fn attempt_projection(
+    row: &AttemptRow,
+    progress: Option<&ProgressSnapshot>,
+) -> Result<Value, PublicApiError> {
+    let state = match row.state.as_str() {
+        "authorized" => "AttemptAuthorized",
+        "executing" => "AttemptExecuting",
+        "reaped" | "stopped" => "AttemptStopped",
+        "committed" => "AttemptCommitted",
+        _ => return Err(PublicApiError::internal()),
+    };
+    let result = row
+        .result
+        .as_ref()
+        .map(|value| result_record_projection(&value.0))
+        .transpose()?;
+    let progress = progress
+        .filter(|progress| {
+            progress.attempt_id.as_str() == row.attempt_id
+                && i64::try_from(progress.fence).ok() == Some(row.fence)
+        })
+        .map(progress_projection)
+        .transpose()?;
+    Ok(json!({
+        "id": row.attempt_id,
+        "fence": row.fence,
+        "worker_incarnation": row.worker_incarnation,
+        "state": state,
+        "progress": progress,
+        "result": result,
+        "failure": row
+            .failure
+            .as_ref()
+            .map(|value| failure_projection(&value.0))
+            .transpose()?,
+        "created_at": format_time(row.authorized_at)?,
+        "updated_at": format_time(row.updated_at)?
+    }))
+}
+
+pub(super) fn result_record_projection(value: &Value) -> Result<Value, PublicApiError> {
+    let record: CommittedResultRecord =
+        serde_json::from_value(value.clone()).map_err(|_| PublicApiError::internal())?;
+    Ok(json!({
+        "kind": "manifest",
+        "manifest_id": record.manifest_id,
+        "schema_version": record.schema_version,
+        "manifest": object_identity_projection(
+            &record.manifest.store,
+            &record.manifest.bucket,
+            &record.manifest.key,
+            record.manifest.version_id.as_deref(),
+            &record.manifest.sha256,
+            record.manifest.size
+        ),
+        "member_count": record.member_count,
+        "total_bytes": record.total_bytes
+    }))
+}
+
+pub(super) fn deletion_projection(operation: &DeletionOperation) -> Result<Value, PublicApiError> {
+    let state = match operation.state {
+        DeletionState::Queued => "queued",
+        DeletionState::Running => "running",
+        DeletionState::RetryWait => "retry_wait",
+        DeletionState::Blocked => "blocked",
+        DeletionState::Completed => "completed",
+    };
+    let problem = operation.problem.as_ref().map(|problem| {
+        let class = match problem.class {
+            DeletionProblemClass::Authorization => "authorization",
+            DeletionProblemClass::TargetUnavailable => "target_unavailable",
+            DeletionProblemClass::Configuration => "configuration",
+            DeletionProblemClass::Internal => "internal",
+        };
+        json!({ "class": class, "message": problem.message })
+    });
+    Ok(json!({
+        "id": operation.id,
+        "job_id": operation.job_id,
+        "state": state,
+        "problem": problem,
+        "created_at": format_time(operation.created_at)?,
+        "updated_at": format_time(operation.updated_at)?
+    }))
+}

crates/remora-api/src/public/request.rs

@@ -0,0 +1,80 @@
+use super::*;
+
+pub(super) fn public_error_body(code: ApiErrorCodeDto, message: &'static str) -> Value {
+    json!({
+        "code": code,
+        "message": message,
+        "request_id": request_id()
+    })
+}
+
+pub(super) fn page_limit(limit: Option<u32>) -> Result<u32, PublicApiError> {
+    let limit = limit.unwrap_or(DEFAULT_LIMIT);
+    if !(1..=MAX_LIMIT).contains(&limit) {
+        return Err(PublicApiError::invalid_request());
+    }
+    Ok(limit)
+}
+
+pub(super) fn parse_id(value: impl Into<String>) -> Result<OpaqueId, PublicApiError> {
+    OpaqueId::parse(value).map_err(|_| PublicApiError::invalid_request())
+}
+
+pub(super) fn idempotency_key(request: &HttpRequest) -> Result<&str, PublicApiError> {
+    let value = request
+        .headers()
+        .get("Idempotency-Key")
+        .and_then(|value| value.to_str().ok())
+        .ok_or_else(PublicApiError::invalid_request)?;
+    if !(16..=128).contains(&value.len())
+        || !value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || b"._~-".contains(&byte))
+    {
+        return Err(PublicApiError::invalid_request());
+    }
+    Ok(value)
+}
+
+pub(super) fn with_idempotency_expiry(mut response: HttpResponse) -> HttpResponse {
+    let expiry = OffsetDateTime::now_utc() + time::Duration::hours(24);
+    if let Ok(value) = expiry.format(&Rfc3339)
+        && let Ok(value) = header::HeaderValue::from_str(&value)
+    {
+        response.headers_mut().insert(
+            header::HeaderName::from_static("idempotency-expires-at"),
+            value,
+        );
+    }
+    response
+}
+
+pub(super) fn format_time(value: OffsetDateTime) -> Result<String, PublicApiError> {
+    value
+        .format(&Rfc3339)
+        .map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn datetime(value: OffsetDateTime) -> Result<DateTimeDto, PublicApiError> {
+    DateTimeDto::parse(format_time(value)?).map_err(|_| PublicApiError::internal())
+}
+
+pub(super) fn request_id() -> OpaqueId {
+    let mut bytes = [0_u8; 16];
+    getrandom::fill(&mut bytes).expect("operating system randomness is required");
+    let mut value = String::from("r");
+    for byte in bytes {
+        use std::fmt::Write as _;
+        write!(value, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    OpaqueId::parse(value).expect("random request id is valid")
+}
+
+pub(super) fn validate_request(
+    schema: &'static str,
+    request: &impl Serialize,
+) -> Result<(), PublicApiError> {
+    remora_contracts::request_is_valid(schema, request)
+        .then_some(())
+        .ok_or_else(PublicApiError::invalid_request)
+}

crates/remora-api/src/public/results.rs

@@ -0,0 +1,122 @@
+use super::*;
+
+pub(super) async fn get_job_result(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    let row = load_visible_job_row(&state, &principal, &job_id).await?;
+    let result = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
+    state.results.resolve_committed(&result).await?;
+    Ok(HttpResponse::Ok().json(result_projection(&result)))
+}
+
+pub(super) async fn list_result_members(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<String>,
+    query: web::Query<PageQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let job_id = parse_id(path.into_inner())?;
+    let row = load_visible_job_row(&state, &principal, &job_id).await?;
+    let committed = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
+    let filter = CursorCodec::filter_fingerprint(&json!({
+        "resource": "result_members",
+        "job_id": job_id,
+        "authorization": authorization_snapshot(&principal)
+    }))?;
+    let cursor = query
+        .cursor
+        .as_deref()
+        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
+        .transpose()
+        .map_err(PublicApiError::from_cursor)?;
+    let after = cursor
+        .as_ref()
+        .and_then(|value| value.after.as_ref())
+        .map(|value| value.id.as_str());
+    let page_size = PageSize::new(page_limit(query.limit)?)?;
+    let page = state
+        .results
+        .page_members(&committed, after, page_size)
+        .await?;
+    let next_cursor = if let Some(after) = page.next_after.as_deref() {
+        let next = KeysetPosition {
+            created_at: row.created_at,
+            id: after.to_owned(),
+        };
+        let cursor = match cursor.as_ref() {
+            Some(current) => next_page_cursor(current, next),
+            None => first_page_cursor(
+                principal.id.as_str(),
+                filter,
+                KeysetPosition {
+                    created_at: row.created_at,
+                    id: String::new(),
+                },
+                next,
+            ),
+        };
+        Some(state.cursors.encode(&cursor)?)
+    } else {
+        None
+    };
+    let items = page
+        .items
+        .iter()
+        .map(|member| {
+            json!({
+                "id": member.id(),
+                "path": member.path(),
+                "size": member.size(),
+                "object": object_identity_projection(
+                    member.object().store(),
+                    member.object().bucket(),
+                    member.object().key(),
+                    member.object().version_id(),
+                    member.content_sha256().as_str(),
+                    member.size()
+                )
+            })
+        })
+        .collect::<Vec<_>>();
+    Ok(HttpResponse::Ok().json(json!({
+        "items": items,
+        "next_cursor": next_cursor
+    })))
+}
+
+pub(super) async fn access_result_member(
+    state: web::Data<PublicApiState>,
+    http: HttpRequest,
+    path: web::Path<(String, String)>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&http).await?;
+    let (job_id, member_id) = path.into_inner();
+    let job_id = parse_id(job_id)?;
+    let member_id = parse_id(member_id)?;
+    let row = load_visible_job_row(&state, &principal, &job_id).await?;
+    let committed = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
+    let member = state.results.lookup_member(&committed, &member_id).await?;
+    let target = state
+        .targets
+        .resolve(member.object().store(), member.object().bucket())
+        .ok_or_else(PublicApiError::unavailable)?;
+    let issued_at = OffsetDateTime::now_utc();
+    let access = target
+        .presign_read(&member.exact_identity(), state.member_access_ttl)
+        .await
+        .map_err(|_| PublicApiError::unavailable())?;
+    let duration =
+        time::Duration::try_from(access.expires_in).map_err(|_| PublicApiError::internal())?;
+    Ok(HttpResponse::Ok()
+        .insert_header((header::CACHE_CONTROL, "no-store"))
+        .insert_header((header::PRAGMA, "no-cache"))
+        .json(ResultMemberAccessDto {
+            url: access.url,
+            expires_at: datetime(issued_at + duration)?,
+        }))
+}

crates/remora-api/src/public/routes.rs

@@ -1,41 +1,45 @@
 use actix_web::web;
 
-use super::{
+use super::{credentials, deletions, events as event_handlers, jobs, results};
-    access_result_member, cancel_job, create_source_credential, events as event_handlers,
-    get_deletion_operation, get_job, get_job_result, get_source_credential, hard_delete_job,
-    list_job_attempts, list_jobs, list_result_members, list_source_credentials, restore_job,
-    revoke_job, revoke_source_credential, rotate_source_credential, soft_delete_job, submit_job,
-};
 
 pub(super) fn jobs(config: &mut web::ServiceConfig) {
     config
-        .route("/jobs", web::post().to(submit_job))
+        .route("/jobs", web::post().to(jobs::submit_job))
-        .route("/jobs", web::get().to(list_jobs))
+        .route("/jobs", web::get().to(jobs::list_jobs))
-        .route("/jobs/{job_id}", web::get().to(get_job))
+        .route("/jobs/{job_id}", web::get().to(jobs::get_job))
-        .route("/jobs/{job_id}/cancel", web::post().to(cancel_job))
+        .route("/jobs/{job_id}/cancel", web::post().to(jobs::cancel_job))
         .route(
             "/jobs/{job_id}/soft-delete",
-            web::post().to(soft_delete_job),
+            web::post().to(deletions::soft_delete_job),
+        )
+        .route(
+            "/jobs/{job_id}/restore",
+            web::post().to(deletions::restore_job),
         )
-        .route("/jobs/{job_id}/restore", web::post().to(restore_job))
         .route(
             "/jobs/{job_id}/hard-delete",
-            web::post().to(hard_delete_job),
+            web::post().to(deletions::hard_delete_job),
         )
-        .route("/jobs/{job_id}/revoke", web::post().to(revoke_job))
+        .route("/jobs/{job_id}/revoke", web::post().to(jobs::revoke_job))
-        .route("/jobs/{job_id}/attempts", web::get().to(list_job_attempts));
+        .route(
+            "/jobs/{job_id}/attempts",
+            web::get().to(jobs::list_job_attempts),
+        );
 }
 
 pub(super) fn results(config: &mut web::ServiceConfig) {
     config
-        .route("/jobs/{job_id}/result", web::get().to(get_job_result))
+        .route(
+            "/jobs/{job_id}/result",
+            web::get().to(results::get_job_result),
+        )
         .route(
             "/jobs/{job_id}/result/members",
-            web::get().to(list_result_members),
+            web::get().to(results::list_result_members),
         )
         .route(
             "/jobs/{job_id}/result/members/{member_id}/content",
-            web::get().to(access_result_member),
+            web::get().to(results::access_result_member),
         );
 }
 
@@ -43,30 +47,30 @@ pub(super) fn credentials(config: &mut web::ServiceConfig) {
     config
         .route(
             "/source-credentials",
-            web::post().to(create_source_credential),
+            web::post().to(credentials::create_source_credential),
         )
         .route(
             "/source-credentials",
-            web::get().to(list_source_credentials),
+            web::get().to(credentials::list_source_credentials),
         )
         .route(
             "/source-credentials/{credential_id}",
-            web::get().to(get_source_credential),
+            web::get().to(credentials::get_source_credential),
         )
         .route(
             "/source-credentials/{credential_id}/versions",
-            web::post().to(rotate_source_credential),
+            web::post().to(credentials::rotate_source_credential),
         )
         .route(
             "/source-credentials/{credential_id}/revoke",
-            web::post().to(revoke_source_credential),
+            web::post().to(credentials::revoke_source_credential),
         );
 }
 
 pub(super) fn deletions(config: &mut web::ServiceConfig) {
     config.route(
         "/deletions/{operation_id}",
-        web::get().to(get_deletion_operation),
+        web::get().to(deletions::get_deletion_operation),
     );
 }
 

crates/remora-api/src/worker.rs

@@ -30,7 +30,10 @@ use remora_target::{ContentSha256, ObjectIdentity, StoredObject, attempt_staging
 use sha2::{Digest, Sha256};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 
+mod authority;
+mod execution;
 mod routes;
+mod staging;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct AuthenticatedWorker {
@@ -104,478 +107,6 @@ pub fn configure_worker_api(config: &mut web::ServiceConfig) {
     );
 }
 
-async fn worker_ready(_: AuthenticatedWorker) -> HttpResponse {
-    HttpResponse::Ok().finish()
-}
-
-async fn register_staging_object(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptStagingObjectRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptStagingObjectRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    let accepted = state
-        .attempts
-        .register_staging_object_operation(
-            &authority,
-            &request.operation_id,
-            StagingObject {
-                target_store: request.object.store.clone(),
-                bucket: request.object.bucket.clone(),
-                key: request.object.key.clone(),
-                version_id: request.object.version_id.clone(),
-            },
-            request.completed_upload_id.clone(),
-        )
-        .await
-        .map_err(WorkerApiError::from_attempt)?;
-    if accepted {
-        Ok(HttpResponse::NoContent().finish())
-    } else {
-        Err(WorkerApiError::precondition_failed())
-    }
-}
-
-async fn register_staging_write_intent(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptStagingWriteIntentRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptStagingWriteIntentRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    let accepted = state
-        .attempts
-        .register_staging_write_intent_operation(
-            &authority,
-            &request.operation_id,
-            StagingWriteIntent {
-                target_store: request.intent.store.clone(),
-                bucket: request.intent.bucket.clone(),
-                key: request.intent.key.clone(),
-            },
-        )
-        .await
-        .map_err(WorkerApiError::from_attempt)?;
-    if accepted {
-        Ok(HttpResponse::NoContent().finish())
-    } else {
-        Err(WorkerApiError::precondition_failed())
-    }
-}
-
-async fn register_staging_multipart(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptStagingMultipartRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptStagingMultipartRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    let accepted = state
-        .attempts
-        .register_staging_multipart_operation(
-            &authority,
-            &request.operation_id,
-            StagingMultipart {
-                target_store: request.multipart.store.clone(),
-                bucket: request.multipart.bucket.clone(),
-                key: request.multipart.key.clone(),
-                upload_id: request.multipart.upload_id.clone(),
-            },
-        )
-        .await
-        .map_err(WorkerApiError::from_attempt)?;
-    if accepted {
-        Ok(HttpResponse::NoContent().finish())
-    } else {
-        Err(WorkerApiError::precondition_failed())
-    }
-}
-
-async fn stop(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptStopRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptStopRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    match state
-        .attempts
-        .cooperative_stop_operation(&authority, &request.operation_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?
-    {
-        remora_attempt::CooperativeStopOutcome::Stopped
-        | remora_attempt::CooperativeStopOutcome::StopNotRequired => {
-            Ok(HttpResponse::NoContent().finish())
-        }
-        remora_attempt::CooperativeStopOutcome::AuthorityMismatch => {
-            Err(WorkerApiError::precondition_failed())
-        }
-    }
-}
-
-async fn claim(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<ClaimRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("ClaimRequest", &*request)?;
-    let job_id = parse_id(path.into_inner())?;
-    let outcome = state
-        .attempts
-        .claim(&job_id, &request.worker_incarnation, &request.operation_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?;
-    let ClaimOutcome::Claimed { authority, .. } = outcome else {
-        return Err(WorkerApiError::precondition_failed());
-    };
-    let execution = load_execution_envelope(&state.attempts, &authority).await?;
-    Ok(HttpResponse::Ok().json(ClaimResponseDto {
-        authority: authority_dto(&authority)?,
-        execution,
-    }))
-}
-
-async fn renew(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptRenewRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptRenewRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    match state
-        .attempts
-        .renew_operation(&authority, &request.operation_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?
-    {
-        RenewalMutationOutcome::Renewed { lease, .. } => {
-            Ok(HttpResponse::Ok().json(lease_dto(lease)?))
-        }
-        RenewalMutationOutcome::AuthorityLost { .. } => Err(WorkerApiError::precondition_failed()),
-    }
-}
-
-async fn activate(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptActivateRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptActivateRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    let lease = match state
-        .attempts
-        .activate_operation(&authority, &request.operation_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?
-    {
-        ActivationMutationOutcome::Activated { lease, .. } => lease,
-        ActivationMutationOutcome::PreconditionFailed => {
-            return Err(WorkerApiError::precondition_failed());
-        }
-    };
-    let execution = load_execution_envelope(&state.attempts, &authority).await?;
-    Ok(HttpResponse::Ok().json(ActivationResponseDto {
-        authority: authority_dto(&authority)?,
-        execution,
-        lease: lease_dto(lease)?,
-    }))
-}
-
-async fn pin_resolved_source(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptResolvedSourceRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptResolvedSourceRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    accepted(
-        state
-            .attempts
-            .pin_resolved_source(
-                &authority,
-                &request.operation_id,
-                descriptor_from_resolved(&request.resolved_source_ref)?,
-            )
-            .await
-            .map_err(WorkerApiError::from_attempt)?,
-    )
-}
-
-async fn persist_checkpoint(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptCheckpointRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptCheckpointRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    accepted(
-        state
-            .attempts
-            .persist_checkpoint(
-                &authority,
-                &request.operation_id,
-                CheckpointRecord {
-                    descriptor: descriptor_from_checkpoint(&request.checkpoint_ref)?,
-                    resolved_source_id: request.resolved_source_id.clone(),
-                    validated_bytes: request.validated_bytes.get(),
-                    validated_members: request.validated_members.get(),
-                    staging_multipart: request.staging_multipart.as_ref().map(|identity| {
-                        CheckpointMultipart {
-                            target_store: identity.store.clone(),
-                            bucket: identity.bucket.clone(),
-                            key: identity.key.clone(),
-                            upload_id: identity.upload_id.clone(),
-                        }
-                    }),
-                },
-            )
-            .await
-            .map_err(WorkerApiError::from_attempt)?,
-    )
-}
-
-async fn report_progress(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptProgressRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptProgressRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    accepted(
-        state
-            .attempts
-            .report_progress(
-                &authority,
-                &request.operation_id,
-                ProgressRecord {
-                    completed_bytes: request.progress.completed_bytes.get(),
-                    total_bytes: request.progress.total_bytes.map(SafeIntegerDto::get),
-                    completed_members: request.progress.completed_members.get(),
-                    total_members: request.progress.total_members.map(SafeIntegerDto::get),
-                },
-            )
-            .await
-            .map_err(WorkerApiError::from_attempt)?,
-    )
-}
-
-async fn report_outcome(
-    _: AuthenticatedWorker,
-    state: web::Data<WorkerApiState>,
-    path: web::Path<String>,
-    request: web::Json<AttemptOutcomeRequestDto>,
-) -> Result<HttpResponse, WorkerApiError> {
-    validate_request("AttemptOutcomeRequest", &*request)?;
-    let authority = request_authority(
-        &state,
-        path.into_inner(),
-        request.fence.get(),
-        &request.worker_incarnation,
-    )
-    .await?;
-    let outcome = match &request.outcome {
-        AttemptOutcomeDto::Success(success) => {
-            let result = &success.result;
-            state
-                .attempts
-                .publish_success_operation(
-                    &authority,
-                    &request.operation_id,
-                    PublicationInput {
-                        attempt_id: authority.attempt_id.clone(),
-                        manifest_id: result.manifest_id.clone(),
-                        schema_version: u32::try_from(result.schema_version.get())
-                            .map_err(|_| WorkerApiError::invalid_request())?,
-                        manifest: StoredObject {
-                            identity: ObjectIdentity {
-                                target_store_id: result.manifest.store.clone(),
-                                bucket: result.manifest.bucket.clone(),
-                                key: result.manifest.key.clone(),
-                                version_id: result.manifest.version_id.clone(),
-                            },
-                            content_sha256: ContentSha256::parse(&result.manifest.checksum)
-                                .map_err(|_| WorkerApiError::invalid_request())?,
-                            size: result.manifest.size.get(),
-                        },
-                        member_count: result.member_count.get(),
-                        total_bytes: result.total_bytes.get(),
-                    },
-                )
-                .await
-        }
-        AttemptOutcomeDto::Failure(failure) => {
-            state
-                .attempts
-                .publish_failure_operation(&authority, &request.operation_id, failure.failure.class)
-                .await
-        }
-    }
-    .map_err(WorkerApiError::from_attempt)?;
-    let phase = match outcome {
-        OutcomePublication::Accepted { phase, .. } => phase,
-        OutcomePublication::Rejected { reason, .. } => {
-            return Err(WorkerApiError::from_rejection(reason));
-        }
-    };
-    Ok(HttpResponse::Ok().json(JobPhaseProjectionDto { phase }))
-}
-
-fn accepted(outcome: AttemptMutationOutcome) -> Result<HttpResponse, WorkerApiError> {
-    match outcome {
-        AttemptMutationOutcome::Accepted(_) => Ok(HttpResponse::NoContent().finish()),
-        AttemptMutationOutcome::Rejected { reason, .. } => {
-            Err(WorkerApiError::from_rejection(reason))
-        }
-    }
-}
-
-async fn request_authority(
-    state: &WorkerApiState,
-    attempt_id: String,
-    fence: u64,
-    worker: &OpaqueId,
-) -> Result<AttemptAuthority, WorkerApiError> {
-    let attempt_id = parse_id(attempt_id)?;
-    state
-        .attempts
-        .historical_authority(&attempt_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?
-        .filter(|authority| authority.fence == fence && authority.worker_incarnation == *worker)
-        .ok_or_else(WorkerApiError::precondition_failed)
-}
-
-async fn load_execution_envelope(
-    attempts: &AttemptService,
-    authority: &AttemptAuthority,
-) -> Result<ExecutionEnvelopeDto, WorkerApiError> {
-    let row = attempts
-        .execution_envelope(&authority.attempt_id)
-        .await
-        .map_err(WorkerApiError::from_attempt)?
-        .ok_or_else(WorkerApiError::precondition_failed)?;
-    if row.job_id != authority.job_id
-        || row.attempt_id != authority.attempt_id
-        || row.fence != authority.fence
-        || row.worker_incarnation != authority.worker_incarnation
-    {
-        return Err(WorkerApiError::precondition_failed());
-    }
-    let source: SourceProjectionDto =
-        serde_json::from_value(row.source).map_err(|_| WorkerApiError::unavailable())?;
-    let target_prefix = row.target_prefix.join("/");
-    let staging_prefix = attempt_staging_prefix(&target_prefix, &row.staging_origin_attempt_id);
-    let resolved_source_ref = row.resolved_source.map(|descriptor| {
-        NullableDto(Some(descriptor_ref(
-            descriptor,
-            DurableDescriptorKindDto::ResolvedSource,
-        )))
-    });
-    let checkpoint_ref = row.checkpoint.and_then(|checkpoint| match checkpoint {
-        CheckpointProjection::Reusable { checkpoint } => Some(NullableDto(Some(descriptor_ref(
-            checkpoint.descriptor,
-            DurableDescriptorKindDto::Checkpoint,
-        )))),
-        CheckpointProjection::Complete { .. } => None,
-    });
-    Ok(ExecutionEnvelopeDto {
-        id: row.id,
-        job_id: authority.job_id.clone(),
-        attempt_id: authority.attempt_id.clone(),
-        fence: positive(authority.fence)?,
-        worker_incarnation: authority.worker_incarnation.clone(),
-        source,
-        sensitive_source_ref: row.sensitive_source_ref,
-        credential_version: NullableDto(match (row.credential_id, row.credential_version_id) {
-            (Some(credential_id), Some(version_id)) => Some(CredentialGrantVersionDto {
-                credential_id,
-                version_id,
-            }),
-            (None, None) => None,
-            _ => return Err(WorkerApiError::unavailable()),
-        }),
-        resolved_source_ref: resolved_source_ref.unwrap_or(NullableDto(None)),
-        target: TargetDto {
-            store: row.target_store.clone(),
-            bucket: row.target_bucket.clone(),
-            prefix: target_prefix,
-        },
-        staging_target: StagingTargetDto {
-            store: row.target_store,
-            bucket: row.target_bucket,
-            prefix: staging_prefix,
-        },
-        execution_grant_ref: row.execution_grant_id,
-        checkpoint_ref: checkpoint_ref.unwrap_or(NullableDto(None)),
-    })
-}
-
 fn descriptor_ref(
     descriptor: DurableDescriptor,
     kind: DurableDescriptorKindDto,

crates/remora-api/src/worker/authority.rs

@@ -0,0 +1,199 @@
+use super::*;
+
+pub(super) async fn worker_ready(_: AuthenticatedWorker) -> HttpResponse {
+    HttpResponse::Ok().finish()
+}
+
+pub(super) async fn stop(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStopRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptStopRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    match state
+        .attempts
+        .cooperative_stop_operation(&authority, &request.operation_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+    {
+        remora_attempt::CooperativeStopOutcome::Stopped
+        | remora_attempt::CooperativeStopOutcome::StopNotRequired => {
+            Ok(HttpResponse::NoContent().finish())
+        }
+        remora_attempt::CooperativeStopOutcome::AuthorityMismatch => {
+            Err(WorkerApiError::precondition_failed())
+        }
+    }
+}
+
+pub(super) async fn claim(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<ClaimRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("ClaimRequest", &*request)?;
+    let job_id = parse_id(path.into_inner())?;
+    let outcome = state
+        .attempts
+        .claim(&job_id, &request.worker_incarnation, &request.operation_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    let ClaimOutcome::Claimed { authority, .. } = outcome else {
+        return Err(WorkerApiError::precondition_failed());
+    };
+    let execution = load_execution_envelope(&state.attempts, &authority).await?;
+    Ok(HttpResponse::Ok().json(ClaimResponseDto {
+        authority: authority_dto(&authority)?,
+        execution,
+    }))
+}
+
+pub(super) async fn renew(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptRenewRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptRenewRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    match state
+        .attempts
+        .renew_operation(&authority, &request.operation_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+    {
+        RenewalMutationOutcome::Renewed { lease, .. } => {
+            Ok(HttpResponse::Ok().json(lease_dto(lease)?))
+        }
+        RenewalMutationOutcome::AuthorityLost { .. } => Err(WorkerApiError::precondition_failed()),
+    }
+}
+
+pub(super) async fn activate(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptActivateRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptActivateRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let lease = match state
+        .attempts
+        .activate_operation(&authority, &request.operation_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+    {
+        ActivationMutationOutcome::Activated { lease, .. } => lease,
+        ActivationMutationOutcome::PreconditionFailed => {
+            return Err(WorkerApiError::precondition_failed());
+        }
+    };
+    let execution = load_execution_envelope(&state.attempts, &authority).await?;
+    Ok(HttpResponse::Ok().json(ActivationResponseDto {
+        authority: authority_dto(&authority)?,
+        execution,
+        lease: lease_dto(lease)?,
+    }))
+}
+
+pub(super) async fn request_authority(
+    state: &WorkerApiState,
+    attempt_id: String,
+    fence: u64,
+    worker: &OpaqueId,
+) -> Result<AttemptAuthority, WorkerApiError> {
+    let attempt_id = parse_id(attempt_id)?;
+    state
+        .attempts
+        .historical_authority(&attempt_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+        .filter(|authority| authority.fence == fence && authority.worker_incarnation == *worker)
+        .ok_or_else(WorkerApiError::precondition_failed)
+}
+
+pub(super) async fn load_execution_envelope(
+    attempts: &AttemptService,
+    authority: &AttemptAuthority,
+) -> Result<ExecutionEnvelopeDto, WorkerApiError> {
+    let row = attempts
+        .execution_envelope(&authority.attempt_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+        .ok_or_else(WorkerApiError::precondition_failed)?;
+    if row.job_id != authority.job_id
+        || row.attempt_id != authority.attempt_id
+        || row.fence != authority.fence
+        || row.worker_incarnation != authority.worker_incarnation
+    {
+        return Err(WorkerApiError::precondition_failed());
+    }
+    let source: SourceProjectionDto =
+        serde_json::from_value(row.source).map_err(|_| WorkerApiError::unavailable())?;
+    let target_prefix = row.target_prefix.join("/");
+    let staging_prefix = attempt_staging_prefix(&target_prefix, &row.staging_origin_attempt_id);
+    let resolved_source_ref = row.resolved_source.map(|descriptor| {
+        NullableDto(Some(descriptor_ref(
+            descriptor,
+            DurableDescriptorKindDto::ResolvedSource,
+        )))
+    });
+    let checkpoint_ref = row.checkpoint.and_then(|checkpoint| match checkpoint {
+        CheckpointProjection::Reusable { checkpoint } => Some(NullableDto(Some(descriptor_ref(
+            checkpoint.descriptor,
+            DurableDescriptorKindDto::Checkpoint,
+        )))),
+        CheckpointProjection::Complete { .. } => None,
+    });
+    Ok(ExecutionEnvelopeDto {
+        id: row.id,
+        job_id: authority.job_id.clone(),
+        attempt_id: authority.attempt_id.clone(),
+        fence: positive(authority.fence)?,
+        worker_incarnation: authority.worker_incarnation.clone(),
+        source,
+        sensitive_source_ref: row.sensitive_source_ref,
+        credential_version: NullableDto(match (row.credential_id, row.credential_version_id) {
+            (Some(credential_id), Some(version_id)) => Some(CredentialGrantVersionDto {
+                credential_id,
+                version_id,
+            }),
+            (None, None) => None,
+            _ => return Err(WorkerApiError::unavailable()),
+        }),
+        resolved_source_ref: resolved_source_ref.unwrap_or(NullableDto(None)),
+        target: TargetDto {
+            store: row.target_store.clone(),
+            bucket: row.target_bucket.clone(),
+            prefix: target_prefix,
+        },
+        staging_target: StagingTargetDto {
+            store: row.target_store,
+            bucket: row.target_bucket,
+            prefix: staging_prefix,
+        },
+        execution_grant_ref: row.execution_grant_id,
+        checkpoint_ref: checkpoint_ref.unwrap_or(NullableDto(None)),
+    })
+}

crates/remora-api/src/worker/execution.rs

@@ -0,0 +1,171 @@
+use super::authority::request_authority;
+use super::*;
+
+pub(super) async fn pin_resolved_source(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptResolvedSourceRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptResolvedSourceRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    accepted(
+        state
+            .attempts
+            .pin_resolved_source(
+                &authority,
+                &request.operation_id,
+                descriptor_from_resolved(&request.resolved_source_ref)?,
+            )
+            .await
+            .map_err(WorkerApiError::from_attempt)?,
+    )
+}
+
+pub(super) async fn persist_checkpoint(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptCheckpointRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptCheckpointRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    accepted(
+        state
+            .attempts
+            .persist_checkpoint(
+                &authority,
+                &request.operation_id,
+                CheckpointRecord {
+                    descriptor: descriptor_from_checkpoint(&request.checkpoint_ref)?,
+                    resolved_source_id: request.resolved_source_id.clone(),
+                    validated_bytes: request.validated_bytes.get(),
+                    validated_members: request.validated_members.get(),
+                    staging_multipart: request.staging_multipart.as_ref().map(|identity| {
+                        CheckpointMultipart {
+                            target_store: identity.store.clone(),
+                            bucket: identity.bucket.clone(),
+                            key: identity.key.clone(),
+                            upload_id: identity.upload_id.clone(),
+                        }
+                    }),
+                },
+            )
+            .await
+            .map_err(WorkerApiError::from_attempt)?,
+    )
+}
+
+pub(super) async fn report_progress(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptProgressRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptProgressRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    accepted(
+        state
+            .attempts
+            .report_progress(
+                &authority,
+                &request.operation_id,
+                ProgressRecord {
+                    completed_bytes: request.progress.completed_bytes.get(),
+                    total_bytes: request.progress.total_bytes.map(SafeIntegerDto::get),
+                    completed_members: request.progress.completed_members.get(),
+                    total_members: request.progress.total_members.map(SafeIntegerDto::get),
+                },
+            )
+            .await
+            .map_err(WorkerApiError::from_attempt)?,
+    )
+}
+
+pub(super) async fn report_outcome(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptOutcomeRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptOutcomeRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let outcome = match &request.outcome {
+        AttemptOutcomeDto::Success(success) => {
+            let result = &success.result;
+            state
+                .attempts
+                .publish_success_operation(
+                    &authority,
+                    &request.operation_id,
+                    PublicationInput {
+                        attempt_id: authority.attempt_id.clone(),
+                        manifest_id: result.manifest_id.clone(),
+                        schema_version: u32::try_from(result.schema_version.get())
+                            .map_err(|_| WorkerApiError::invalid_request())?,
+                        manifest: StoredObject {
+                            identity: ObjectIdentity {
+                                target_store_id: result.manifest.store.clone(),
+                                bucket: result.manifest.bucket.clone(),
+                                key: result.manifest.key.clone(),
+                                version_id: result.manifest.version_id.clone(),
+                            },
+                            content_sha256: ContentSha256::parse(&result.manifest.checksum)
+                                .map_err(|_| WorkerApiError::invalid_request())?,
+                            size: result.manifest.size.get(),
+                        },
+                        member_count: result.member_count.get(),
+                        total_bytes: result.total_bytes.get(),
+                    },
+                )
+                .await
+        }
+        AttemptOutcomeDto::Failure(failure) => {
+            state
+                .attempts
+                .publish_failure_operation(&authority, &request.operation_id, failure.failure.class)
+                .await
+        }
+    }
+    .map_err(WorkerApiError::from_attempt)?;
+    let phase = match outcome {
+        OutcomePublication::Accepted { phase, .. } => phase,
+        OutcomePublication::Rejected { reason, .. } => {
+            return Err(WorkerApiError::from_rejection(reason));
+        }
+    };
+    Ok(HttpResponse::Ok().json(JobPhaseProjectionDto { phase }))
+}
+
+pub(super) fn accepted(outcome: AttemptMutationOutcome) -> Result<HttpResponse, WorkerApiError> {
+    match outcome {
+        AttemptMutationOutcome::Accepted(_) => Ok(HttpResponse::NoContent().finish()),
+        AttemptMutationOutcome::Rejected { reason, .. } => {
+            Err(WorkerApiError::from_rejection(reason))
+        }
+    }
+}

crates/remora-api/src/worker/routes.rs

@@ -1,33 +1,44 @@
 use actix_web::web;
 
-use super::{
+use super::{authority as authority_handlers, execution, staging};
-    activate, claim, persist_checkpoint, pin_resolved_source, register_staging_multipart,
-    register_staging_object, register_staging_write_intent, renew, report_outcome, report_progress,
-    stop, worker_ready,
-};
 
 pub(super) fn authority(config: &mut web::ServiceConfig) {
     config
-        .route("/health/ready", web::get().to(worker_ready))
+        .route(
-        .route("/jobs/{job_id}/claim", web::post().to(claim))
+            "/health/ready",
-        .route("/attempts/{attempt_id}/renew", web::post().to(renew))
+            web::get().to(authority_handlers::worker_ready),
-        .route("/attempts/{attempt_id}/activate", web::post().to(activate))
+        )
-        .route("/attempts/{attempt_id}/stop", web::post().to(stop));
+        .route(
+            "/jobs/{job_id}/claim",
+            web::post().to(authority_handlers::claim),
+        )
+        .route(
+            "/attempts/{attempt_id}/renew",
+            web::post().to(authority_handlers::renew),
+        )
+        .route(
+            "/attempts/{attempt_id}/activate",
+            web::post().to(authority_handlers::activate),
+        )
+        .route(
+            "/attempts/{attempt_id}/stop",
+            web::post().to(authority_handlers::stop),
+        );
 }
 
 pub(super) fn staging(config: &mut web::ServiceConfig) {
     config
         .route(
             "/attempts/{attempt_id}/staging/write-intent",
-            web::post().to(register_staging_write_intent),
+            web::post().to(staging::register_staging_write_intent),
         )
         .route(
             "/attempts/{attempt_id}/staging/object",
-            web::post().to(register_staging_object),
+            web::post().to(staging::register_staging_object),
         )
         .route(
             "/attempts/{attempt_id}/staging/multipart",
-            web::post().to(register_staging_multipart),
+            web::post().to(staging::register_staging_multipart),
         );
 }
 
@@ -35,18 +46,18 @@ pub(super) fn execution(config: &mut web::ServiceConfig) {
     config
         .route(
             "/attempts/{attempt_id}/resolved-source",
-            web::post().to(pin_resolved_source),
+            web::post().to(execution::pin_resolved_source),
         )
         .route(
             "/attempts/{attempt_id}/checkpoint",
-            web::post().to(persist_checkpoint),
+            web::post().to(execution::persist_checkpoint),
         )
         .route(
             "/attempts/{attempt_id}/progress",
-            web::post().to(report_progress),
+            web::post().to(execution::report_progress),
         )
         .route(
             "/attempts/{attempt_id}/outcome",
-            web::post().to(report_outcome),
+            web::post().to(execution::report_outcome),
         );
 }

crates/remora-api/src/worker/staging.rs

@@ -0,0 +1,107 @@
+use super::authority::request_authority;
+use super::*;
+
+pub(super) async fn register_staging_object(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingObjectRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptStagingObjectRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_object_operation(
+            &authority,
+            &request.operation_id,
+            StagingObject {
+                target_store: request.object.store.clone(),
+                bucket: request.object.bucket.clone(),
+                key: request.object.key.clone(),
+                version_id: request.object.version_id.clone(),
+            },
+            request.completed_upload_id.clone(),
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}
+
+pub(super) async fn register_staging_write_intent(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingWriteIntentRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptStagingWriteIntentRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_write_intent_operation(
+            &authority,
+            &request.operation_id,
+            StagingWriteIntent {
+                target_store: request.intent.store.clone(),
+                bucket: request.intent.bucket.clone(),
+                key: request.intent.key.clone(),
+            },
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}
+
+pub(super) async fn register_staging_multipart(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingMultipartRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    validate_request("AttemptStagingMultipartRequest", &*request)?;
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_multipart_operation(
+            &authority,
+            &request.operation_id,
+            StagingMultipart {
+                target_store: request.multipart.store.clone(),
+                bucket: request.multipart.bucket.clone(),
+                key: request.multipart.key.clone(),
+                upload_id: request.multipart.upload_id.clone(),
+            },
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}