diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 07715232f..d3304617e 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -1140,8 +1140,36 @@ pub struct SubjectReleaseResponse { pub state: SubjectReleaseState, } +/// Irreversibly fence an unpublished failed Subject create. Exact retries must retain +/// all four fields; this is distinct from releasing an existing Subject Controller. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct SubjectCreateRecoveryRequest { + pub workspace_id: String, + pub worker_id: WorkerId, + pub create_fingerprint: String, + pub operation_id: String, +} + +/// Durable admission-fence proof plus freshly confirmed execution absence. +/// An error (including a lost response) is never evidence of absence. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct SubjectCreateRecoveryResponse { + pub workspace_id: String, + pub worker_id: WorkerId, + pub create_fingerprint: String, + pub operation_id: String, +} + #[api(reqwest, axum)] pub trait RuntimeApi { + #[post("/v1/subject-create-recovery", status = 200, error_status = 400)] + async fn recover_subject_create( + &self, + #[body] request: SubjectCreateRecoveryRequest, + ) -> Result; + #[post("/v1/subject-release", status = 200, error_status = 400)] async fn subject_release( &self, @@ -1675,6 +1703,18 @@ mod tests { ) -> Result { Err(test_error(501)) } + async fn recover_subject_create( + &self, + request: SubjectCreateRecoveryRequest, + ) -> Result { + Ok(SubjectCreateRecoveryResponse { + workspace_id: request.workspace_id, + worker_id: request.worker_id, + create_fingerprint: request.create_fingerprint, + operation_id: request.operation_id, + }) + } + async fn subject_release( &self, _request: SubjectReleaseRequest, @@ -1820,6 +1860,21 @@ mod tests { .build() .unwrap(); + let recovery = SubjectCreateRecoveryRequest { + workspace_id: "workspace-a".into(), + worker_id: WorkerId::now_v7(), + create_fingerprint: "failed-create-fingerprint".into(), + operation_id: "recovery-1".into(), + }; + let response = client + .recover_subject_create(recovery.clone()) + .await + .unwrap(); + assert_eq!(response.workspace_id, recovery.workspace_id); + assert_eq!(response.worker_id, recovery.worker_id); + assert_eq!(response.create_fingerprint, recovery.create_fingerprint); + assert_eq!(response.operation_id, recovery.operation_id); + let summary = client.runtime_summary().await.unwrap(); assert_eq!(summary.runtime.display_name.as_deref(), Some("round-trip")); let workers = client diff --git a/crates/subjektiv/src/journal/binding.rs b/crates/subjektiv/src/journal/binding.rs index 866b42d03..3430a72cb 100644 --- a/crates/subjektiv/src/journal/binding.rs +++ b/crates/subjektiv/src/journal/binding.rs @@ -67,6 +67,27 @@ impl SubjektivStore { }) } + /// Absence proof for failed body creation, including bindings never attached to a Session + /// and released lifetimes. The Host must hold its lifecycle fence across this observation + /// and recovery; absence alone does not fence a delayed Runtime create. + pub fn journal_body_is_unpublished( + &self, + subject: &str, + runtime: &str, + worker: &str, + ) -> JournalResult { + self.database.try_with_connection(|tx| { + Ok(tx.query_row( + "SELECT NOT EXISTS(SELECT 1 FROM journal_writer_bindings + WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3) + AND NOT EXISTS(SELECT 1 FROM subject_session_attributions + WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3)", + params![subject, runtime, worker], + |row| row.get(0), + )?) + }) + } + pub fn journal_worker_was_released( &self, subject: &str, @@ -203,3 +224,82 @@ impl SubjektivStore { }) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unpublished_body_rejects_orphan_and_released_bindings_and_attribution_only() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = crate::tests::open_store(temp.path(), "absence-tests"); + let subject = store + .create_subject(crate::SubjectRole::new("companion").unwrap()) + .unwrap(); + let other_subject = store + .create_subject(crate::SubjectRole::new("companion").unwrap()) + .unwrap(); + assert!( + store + .journal_body_is_unpublished(&subject.id, "runtime", "worker") + .unwrap() + ); + // No Session was attached: an orphan binding is still published body evidence. + let binding = store + .claim_journal_writer(&subject.id, "runtime", "worker") + .unwrap(); + assert!( + !store + .journal_body_is_unpublished(&subject.id, "runtime", "worker") + .unwrap() + ); + for (subject_id, runtime, worker) in [ + (other_subject.id.as_str(), "runtime", "worker"), + (subject.id.as_str(), "other-runtime", "worker"), + (subject.id.as_str(), "runtime", "other-worker"), + ] { + assert!( + store + .journal_body_is_unpublished(subject_id, runtime, worker) + .unwrap() + ); + } + store.release_journal_writer(&binding).unwrap(); + assert!( + !store + .journal_body_is_unpublished(&subject.id, "runtime", "worker") + .unwrap() + ); + store + .record_session_attribution( + SubjectSessionAttribution::new( + &subject.id, + "runtime", + "attribution-only", + "session", + ) + .unwrap(), + ) + .unwrap(); + assert!( + !store + .journal_body_is_unpublished(&subject.id, "runtime", "attribution-only") + .unwrap() + ); + assert!( + store + .journal_body_is_unpublished(&subject.id, "other-runtime", "attribution-only") + .unwrap() + ); + assert!( + store + .journal_body_is_unpublished(&other_subject.id, "runtime", "attribution-only") + .unwrap() + ); + assert!( + store + .journal_body_is_unpublished(&subject.id, "runtime", "never-published") + .unwrap() + ); + } +} diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 11c8a3fde..f15d5ca4e 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -567,6 +567,15 @@ impl fmt::Debug for WorkerProtocolTransport { pub trait WorkerExecutionBackend: Send + Sync + 'static { fn backend_id(&self) -> &str; + /// Confirm actual absence under the backend Worker operation lock. Runtime has + /// already durably fenced admission; unsupported or uncertain backends must fail. + fn recover_subject_create( + &self, + _request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> Result<(), String> { + Err("execution backend cannot confirm Subject create execution absence".into()) + } + /// Must reserve the actual Controller, not infer Idle from a catalog snapshot. fn subject_release( &self, @@ -839,6 +848,13 @@ impl WorkerExecutionBackendRef { self.backend.fetch_workspace_config(request) } + pub(crate) fn recover_subject_create( + &self, + request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> Result<(), String> { + self.backend.recover_subject_create(request) + } + pub(crate) fn subject_release( &self, request: &runtime_api::SubjectReleaseRequest, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index b46888e91..5b9c35b85 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -39,9 +39,10 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 11; -const EXECUTION_SCHEMA_VERSION: u32 = 10; +const SCHEMA_VERSION: u32 = 12; // Frozen source schemas: do not move their meaning when adding a migration. +const SOURCE_SCHEMA_VERSION: u32 = 11; +const EXECUTION_SCHEMA_VERSION: u32 = 10; const AGGREGATE_SCHEMA_VERSION: u32 = 9; const PREVIOUS_SCHEMA_VERSION: u32 = 8; const LEGACY_SCHEMA_VERSION: u32 = 7; @@ -480,6 +481,8 @@ pub(crate) struct OpenedFsRuntimeStore { #[derive(Clone, Debug)] pub(crate) struct PersistedRuntimeState { + pub(crate) subject_create_recovery_fences: + BTreeMap, pub(crate) subject_releases: Vec<( runtime_api::SubjectReleaseRequest, runtime_api::SubjectReleaseState, @@ -641,8 +644,8 @@ fn plan_runtime_store_migration( format!("Runtime store schema version {schema_version} is out of range"), ) })?; - let staging = migration_sibling(root, "schema-v11-staging")?; - let backup = migration_sibling(root, "pre-schema-v11-backup")?; + let staging = migration_sibling(root, "schema-v12-staging")?; + let backup = migration_sibling(root, "pre-schema-v12-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -676,11 +679,12 @@ fn plan_runtime_store_migration( | PREVIOUS_SCHEMA_VERSION | AGGREGATE_SCHEMA_VERSION | EXECUTION_SCHEMA_VERSION + | SOURCE_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, or {SCHEMA_VERSION}" ), )); } @@ -1010,11 +1014,16 @@ fn migrate_worker_record( mapping: Option<&LegacyWorkerIdentityMapping>, identity_path: &Path, ) -> Result { - if source_schema_version == EXECUTION_SCHEMA_VERSION { + if matches!( + source_schema_version, + EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION + ) { + // Schemas 10 and 11 already use the current Worker aggregate layout. + // Only advance its version; never regenerate Restore admission identity. if identity_document .get("schema_version") .and_then(serde_json::Value::as_u64) - != Some(u64::from(EXECUTION_SCHEMA_VERSION)) + != Some(u64::from(source_schema_version)) { return Err(runtime_store_corrupt( identity_path, @@ -1405,11 +1414,18 @@ fn migrate_runtime_document( "schema_version".to_string(), serde_json::Value::from(SCHEMA_VERSION), ); - object.remove("workers"); - // This legacy cache is not Runtime lifecycle or restore authority. Bundles - // are reacquired from Workspace content authority, never renumbered/rehash-pinned here. - object.remove("config_bundles"); - object.remove("next_worker_sequence"); + // Intermediate schema-11 writers may already have saved irreversible fences. + // Default only a missing field; malformed or existing fences must not be erased. + object + .entry("subject_create_recovery_fences") + .or_insert_with(|| serde_json::json!({})); + if source_schema_version != SOURCE_SCHEMA_VERSION { + object.remove("workers"); + // This legacy cache is not Runtime lifecycle or restore authority. Bundles + // are reacquired from Workspace content authority, never renumbered/rehash-pinned here. + object.remove("config_bundles"); + object.remove("next_worker_sequence"); + } let snapshot: RuntimeSnapshot = serde_json::from_value(document.clone()).map_err(|error| { runtime_store_corrupt( runtime_path, @@ -1523,14 +1539,16 @@ fn recover_runtime_store_migration(root: &Path) -> Result<(), RuntimeError> { "pre-schema-v9-backup", "schema-v10-staging", "pre-schema-v10-backup", + "schema-v11-staging", + "pre-schema-v11-backup", ] { let earlier = migration_sibling(root, suffix)?; if earlier.exists() { - return Err(runtime_store_corrupt(root, "unfinished earlier Runtime migration requires explicit recovery before schema-v11 startup".into())); + return Err(runtime_store_corrupt(root, "unfinished earlier Runtime migration requires explicit recovery before schema-v12 startup".into())); } } - let staging = migration_sibling(root, "schema-v11-staging")?; - let backup = migration_sibling(root, "pre-schema-v11-backup")?; + let staging = migration_sibling(root, "schema-v12-staging")?; + let backup = migration_sibling(root, "pre-schema-v12-backup")?; let root_exists = migration_directory_exists(root)?; let staging_exists = migration_directory_exists(&staging)?; let backup_exists = migration_directory_exists(&backup)?; @@ -1732,8 +1750,8 @@ fn migrate_runtime_store( if !plan.migration_required { return Ok(plan); } - let staging = migration_sibling(root, "schema-v11-staging")?; - let backup = migration_sibling(root, "pre-schema-v11-backup")?; + let staging = migration_sibling(root, "schema-v12-staging")?; + let backup = migration_sibling(root, "pre-schema-v12-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -1908,7 +1926,10 @@ fn migrate_runtime_store_in_place( )); } } - if let Some(metadata) = metadata { + // Schema 11 metadata was validated above but needs no conversion. + if let Some(metadata) = metadata + && plan.current_schema_version != SOURCE_SCHEMA_VERSION + { atomic_write_json( &migrated_dir.join(WORKER_METADATA_FILE), &metadata, @@ -1927,7 +1948,11 @@ fn migrate_runtime_store_in_place( ), )); } - crate::retention::migrate_staged_sessions(root)?; + // Schema 11 already migrated Sessions; the recovery-fence bump must not + // rewrite retained history or archive attestations. + if plan.current_schema_version != SOURCE_SCHEMA_VERSION { + crate::retention::migrate_staged_sessions(root)?; + } let (document, _) = migrate_runtime_document( document, plan.current_schema_version, @@ -1944,6 +1969,8 @@ fn migrate_runtime_store_in_place( #[derive(Clone, Debug, Serialize, Deserialize)] struct RuntimeSnapshot { + #[serde(default)] + subject_create_recovery_fences: BTreeMap, #[serde(default)] subject_releases: Vec<( runtime_api::SubjectReleaseRequest, @@ -1980,6 +2007,7 @@ fn record_worker_load_diagnostic( impl RuntimeSnapshot { fn from_persisted(state: &PersistedRuntimeState) -> Self { Self { + subject_create_recovery_fences: state.subject_create_recovery_fences.clone(), subject_releases: state.subject_releases.clone(), schema_version: SCHEMA_VERSION, display_name: state.display_name.clone(), @@ -2003,6 +2031,33 @@ impl RuntimeSnapshot { ), }); } + for (worker_id, fence) in &self.subject_create_recovery_fences { + let request = &fence.request; + let response_matches = fence.response.as_ref().is_none_or(|response| { + response.worker_id == request.worker_id + && response.workspace_id == request.workspace_id + && response.create_fingerprint == request.create_fingerprint + && response.operation_id == request.operation_id + }); + if *worker_id != request.worker_id + || request.workspace_id.trim().is_empty() + || request.create_fingerprint.trim().is_empty() + || request.create_fingerprint.len() > 512 + || request.operation_id.trim().is_empty() + || request.operation_id.len() > 512 + || self + .workspace_owners + .get(&request.workspace_id) + .is_none_or(|owner| owner.trim().is_empty()) + || !response_matches + { + return Err(RuntimeError::StoreCorrupt { + operation: "read Subject create recovery fence", + path: path.to_path_buf(), + message: "invalid recovery identity, receipt or Workspace owner".into(), + }); + } + } if self.backend != RuntimeBackendKind::FsStore { return Err(RuntimeError::StoreCorrupt { operation: "read runtime snapshot", @@ -2018,6 +2073,7 @@ impl RuntimeSnapshot { workers: BTreeMap, ) -> PersistedRuntimeState { PersistedRuntimeState { + subject_create_recovery_fences: self.subject_create_recovery_fences, subject_releases: self.subject_releases, display_name: self.display_name, status: self.status, @@ -2708,6 +2764,199 @@ mod tests { document["schema_version"].as_u64().unwrap() } + #[test] + fn schema_v11_migration_preserves_identity_history_and_recovery_fences_on_reopen() { + // Cover the released schema (no field) and intermediate writers with + // either a pending fence or a completed recovery receipt. + for fence_state in ["absent", "pending", "completed"] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let worker_id = WorkerId::now_v7(); + let fenced_worker_id = WorkerId::now_v7(); + let worker_dir = root.join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(worker_dir.join("session")).unwrap(); + let runtime_path = root.join(RUNTIME_FILE); + let worker_path = worker_dir.join(WORKER_FILE); + let metadata_path = worker_dir.join(WORKER_METADATA_FILE); + let history_path = worker_dir.join("session/history.json"); + let history = b"[{\"role\":\"user\",\"content\":\"retained history\"}]\n"; + fs::write(&history_path, history).unwrap(); + let metadata = serde_json::to_vec(&serde_json::json!({ + "worker_name": runtime_worker_name(worker_id), + "workspace_id": "workspace-a", + "retained_extension": {"history": "must not be rewritten"} + })) + .unwrap(); + fs::write(&metadata_path, &metadata).unwrap(); + + let mut runtime = serde_json::json!({ + "schema_version": 11, + "display_name": "schema-11 Runtime", + "backend": "fs_store", + "status": "running", + "next_diagnostic_id": 2, + "config_bundles": {}, + "workspace_owners": {"workspace-a": "owner-a"}, + "subject_releases": [], + "diagnostics": [{ + "id": 1, "worker_ref": {"worker_id": worker_id}, + "severity": "warning", "code": "retained_history", + "message": "preserve diagnostic identity" + }] + }); + let recovery_identity = serde_json::json!({ + "workspace_id": "workspace-a", "worker_id": fenced_worker_id, + "create_fingerprint": "failed-create", "operation_id": "recover-a" + }); + if fence_state != "absent" { + runtime["subject_create_recovery_fences"] = serde_json::json!({ + fenced_worker_id.to_string(): { + "request": recovery_identity, + "response": if fence_state == "completed" { + recovery_identity.clone() + } else { + serde_json::Value::Null + } + } + }); + } + atomic_write_json(&runtime_path, &runtime, "write schema-11 fixture").unwrap(); + let runtime_before = fs::read(&runtime_path).unwrap(); + + let mut request = schema_v7_worker_document(worker_id)["request"].clone(); + request["workspace_api"] = serde_json::json!({ + "workspace_id": "workspace-a", "base_url": "https://workspace.invalid" + }); + // Serialize defaults as a schema-11 writer would, without running + // a migration or generating a fresh Restore guard for the fixture. + let request: CreateWorkerRequest = serde_json::from_value(request).unwrap(); + let worker: WorkerAggregateRecord = serde_json::from_value(serde_json::json!({ + "schema_version": 11, + "restore_guard": { + "observation_token": "retained-restore-token", + "active_request_id": null, "owners": {} + }, + "worker_ref": {"worker_id": worker_id}, "worker_id": worker_id, + "profile": request.profile, + "display_name": null, + "profile_source": request.profile_source.reference(), + "config_bundle": null, + "created_at_ms": 1234, + "status": "running", + "last_finished_submission_request_id": "finished-submission", + "workspace_id": "workspace-a", + "execution_state": {"state": "available", "execution": { + "request": request, "binding": {}, "restore_intent": "automatic" + }} + })) + .unwrap(); + let mut expected_worker = serde_json::to_value(&worker).unwrap(); + atomic_write_json(&worker_path, &worker, "write schema-11 Worker").unwrap(); + let worker_before = fs::read(&worker_path).unwrap(); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-test"); + let plan = FsRuntimeStore::migration_plan(&options).unwrap(); + assert_eq!(plan.current_schema_version, 11); + assert_eq!(plan.target_schema_version, 12); + assert!(plan.migration_required); + assert_eq!(plan.worker_count, 1); + assert_eq!(plan.migrated_worker_aggregate_count, 0); + assert!(plan.mappings.is_empty()); + assert_eq!(fs::read(&runtime_path).unwrap(), runtime_before); + assert_eq!(fs::read(&worker_path).unwrap(), worker_before); + // The direct reader still rejects a previous schema: migration is + // explicit, not permissive deserialization of a newer format. + let snapshot: RuntimeSnapshot = serde_json::from_value(runtime.clone()).unwrap(); + assert!(snapshot.validate(&runtime_path).is_err()); + + let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap(); + let state = opened.state.as_ref().unwrap(); + let persisted_worker = state.workers.get(&worker_id).unwrap(); + expected_worker["schema_version"] = serde_json::json!(12); + assert_eq!( + serde_json::to_value(WorkerAggregateRecord::from_persisted(persisted_worker)) + .unwrap(), + expected_worker + ); + runtime["schema_version"] = serde_json::json!(12); + if fence_state == "absent" { + runtime["subject_create_recovery_fences"] = serde_json::json!({}); + } + assert_eq!( + serde_json::to_value(&state.subject_create_recovery_fences).unwrap(), + runtime["subject_create_recovery_fences"] + ); + assert_eq!( + read_json::(&runtime_path, "read migrated Runtime").unwrap(), + runtime + ); + assert_eq!( + read_json::(&worker_path, "read migrated Worker").unwrap(), + expected_worker + ); + // Exercise the normal writer too: it must retain intermediate fences. + opened.store.write_runtime_snapshot(state).unwrap(); + drop(opened); + let runtime_once = fs::read(&runtime_path).unwrap(); + let worker_once = fs::read(&worker_path).unwrap(); + let reopened = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap(); + assert_eq!( + serde_json::to_value( + &reopened + .state + .as_ref() + .unwrap() + .subject_create_recovery_fences + ) + .unwrap(), + runtime["subject_create_recovery_fences"] + ); + drop(reopened); + let plan = FsRuntimeStore::migrate(&options).unwrap(); + assert!(!plan.migration_required); + assert_eq!(plan.current_schema_version, 12); + assert_eq!(plan.target_schema_version, 12); + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + assert_eq!(fs::read(&runtime_path).unwrap(), runtime_once); + assert_eq!(fs::read(&worker_path).unwrap(), worker_once); + assert_eq!(fs::read(&metadata_path).unwrap(), metadata); + assert_eq!(fs::read(&history_path).unwrap(), history); + assert!( + !migration_sibling(&root, "schema-v12-staging") + .unwrap() + .exists() + ); + assert!( + !migration_sibling(&root, "pre-schema-v12-backup") + .unwrap() + .exists() + ); + } + } + + #[test] + fn schema_v11_migration_rejects_invalid_fences_without_replacing_them() { + for fences in [ + serde_json::Value::Null, + serde_json::json!({"invalid-worker": {}}), + ] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + write_empty_schema_v8_store(&root); + let path = root.join(RUNTIME_FILE); + let mut runtime: serde_json::Value = read_json(&path, "read fixture").unwrap(); + runtime["schema_version"] = serde_json::json!(11); + runtime["subject_create_recovery_fences"] = fences; + atomic_write_json(&path, &runtime, "write invalid fence fixture").unwrap(); + let before = fs::read(&path).unwrap(); + assert!(FsRuntimeStore::open_or_create(root, "runtime-test").is_err()); + assert_eq!(fs::read(&path).unwrap(), before); + } + } + #[test] fn schema_v10_startup_migrates_stopped_sessions_atomically() { for corrupt in [false, true] { @@ -2761,7 +3010,7 @@ mod tests { assert_eq!(record["schema_version"], 10); } else { drop(result.unwrap()); - assert_eq!(stored_schema_version(&root), 11); + assert_eq!(stored_schema_version(&root), u64::from(SCHEMA_VERSION)); let manifest: serde_json::Value = read_json(&session.join("session.json"), "test").unwrap(); assert_eq!(manifest["schema_version"], 4); @@ -3151,7 +3400,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v12-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join("partial"), b"partial").unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3166,7 +3415,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v12-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3180,10 +3429,10 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v12-staging").unwrap(); copy_runtime_tree(&root, &root, &staging).unwrap(); migrate_runtime_store_in_place(&staging, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v12-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3198,10 +3447,10 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v12-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join(RUNTIME_FILE), b"{\"schema_version\":8}").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v12-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3218,7 +3467,7 @@ mod tests { let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); migrate_runtime_store_in_place(&root, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v12-backup").unwrap(); write_empty_schema_v8_store(&backup); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3228,14 +3477,21 @@ mod tests { #[test] fn earlier_migration_backup_cannot_be_mistaken_for_a_fresh_runtime() { - let temp = tempfile::tempdir().unwrap(); - let root = temp.path().join("runtime-store"); - let backup = migration_sibling(&root, "pre-schema-v9-backup").unwrap(); - write_empty_schema_v8_store(&backup); - let error = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap_err(); - assert!(error.to_string().contains("explicit recovery")); - assert!(!root.exists()); - assert!(backup.join(RUNTIME_FILE).exists()); + for suffix in [ + "pre-schema-v9-backup", + "pre-schema-v10-backup", + "schema-v11-staging", + "pre-schema-v11-backup", + ] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime-store"); + let backup = migration_sibling(&root, suffix).unwrap(); + write_empty_schema_v8_store(&backup); + let error = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap_err(); + assert!(error.to_string().contains("explicit recovery")); + assert!(!root.exists()); + assert!(backup.join(RUNTIME_FILE).exists()); + } } #[test] diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 2f8743d8b..a916a5f69 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2102,6 +2102,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s { return Some("workers:input"); } + if path == "/v1/subject-create-recovery" && *method == Method::POST { + return Some("workers:delete"); + } if path.ends_with("/stop") || path.ends_with("/cancel") || (path == "/v1/subject-release" && *method == Method::POST) @@ -2973,6 +2976,19 @@ mod tests { } } + #[test] + fn subject_create_recovery_requires_worker_delete_permission() { + assert_eq!( + required_runtime_permission(&Method::POST, "/v1/subject-create-recovery"), + Some("workers:delete") + ); + // Existing ordinary release authorization is unchanged. + assert_eq!( + required_runtime_permission(&Method::POST, "/v1/subject-release"), + Some("workers:stop") + ); + } + #[test] fn retention_routes_require_worker_delete_permission() { assert_eq!( @@ -3197,6 +3213,14 @@ mod tests { struct AcceptingBackend; impl WorkerExecutionBackend for AcceptingBackend { + fn recover_subject_create( + &self, + _request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> Result<(), String> { + // This test backend owns no real executions. + Ok(()) + } + fn backend_id(&self) -> &str { "http-test" } diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index d52b7a171..e7daf4906 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -406,6 +406,18 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { } } + async fn recover_subject_create( + &self, + request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + let scope = material_scope(&self.state, &request.workspace_id)?; + self.state + .runtime + .recover_subject_create_scoped(&scope, request) + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error) + } + async fn subject_release( &self, request: runtime_api::SubjectReleaseRequest, diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index b7bb8513b..33f8555dc 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -1086,6 +1086,54 @@ async fn canonical_http_rejects_replay_tampered_body_and_workspace_mismatch() { assert_eq!(unsigned.status(), StatusCode::UNAUTHORIZED); } +#[tokio::test] +async fn subject_create_recovery_signed_http_requires_delete_and_matching_workspace() { + let f = Fixture::new(); + let identity = RuntimeIdentityMaterial::generate("recovery-key").unwrap(); + let trust = workspace_trust(&identity); + let app = runtime_http_router_with_workspace_auth( + f.runtime.clone(), + None, + workspace_test_auth(vec![trust.clone()]), + ); + let path = "/v1/subject-create-recovery"; + let mut request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: f.scope.workspace_id.clone(), + worker_id: WorkerId::now_v7(), + create_fingerprint: "unpublished-create".into(), + operation_id: "recovery-1".into(), + }; + for operation in ["workers:read", "workers:stop", "workers:create"] { + let body = serde_json::to_vec(&request).unwrap(); + let mut claims = signed_claims(&trust, "POST", path, &body); + claims.operation = operation.into(); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + request.workspace_id = "other-workspace".into(); + let body = serde_json::to_vec(&request).unwrap(); + let claims = signed_claims(&trust, "POST", path, &body); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + request.workspace_id = f.scope.workspace_id.clone(); + for _ in 0..2 { + let body = serde_json::to_vec(&request).unwrap(); + let claims = signed_claims(&trust, "POST", path, &body); + assert_eq!(claims.operation, "workers:delete"); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), 16384) + .await + .unwrap(); + let receipt: runtime_api::SubjectCreateRecoveryResponse = + serde_json::from_slice(&bytes).unwrap(); + assert_eq!(receipt.worker_id, request.worker_id); + assert_eq!(receipt.workspace_id, request.workspace_id); + assert_eq!(receipt.create_fingerprint, request.create_fingerprint); + assert_eq!(receipt.operation_id, request.operation_id); + } +} + #[tokio::test] async fn explicit_material_erasure_requires_delete_capability_and_matching_scope() { let f = Fixture::new(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index d988a491d..2b671c6e5 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -355,7 +355,92 @@ impl Drop for WorkerOperationLease { } } +/// Irreversible admission exclusion, independent of the removable Worker aggregate. +/// A saved response is historical: retries still require a fresh backend absence check. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SubjectCreateRecoveryFence { + pub(crate) request: runtime_api::SubjectCreateRecoveryRequest, + pub(crate) response: Option, +} + impl Runtime { + pub fn recover_subject_create_scoped( + &self, + scope: &RuntimeWorkspaceScope, + request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + if request.workspace_id != scope.workspace_id + || request.create_fingerprint.trim().is_empty() + || request.operation_id.trim().is_empty() + || request.create_fingerprint.len() > 512 + || request.operation_id.len() > 512 + { + return Err(RuntimeError::InvalidRequest( + "invalid Subject create recovery identity".into(), + )); + } + let lease = self.worker_operation_lock(request.worker_id)?; + let _operation = lease.lock().map_err(|_| RuntimeError::StatePoisoned)?; + let backend = { + let mut state = self.lock()?; + state.ensure_running()?; + // A memory-only Runtime cannot issue durable proof. + if state.backend == RuntimeBackendKind::Memory { + return Err(RuntimeError::InvalidRequest( + "Subject create recovery requires durable storage".into(), + )); + } + state.ensure_workspace_owner(scope, false)?; + if state.workers.contains_key(&request.worker_id) { + return Err(RuntimeError::InvalidRequest( + "Subject create catalog record is still present".into(), + )); + } + if let Some(saved) = state.subject_create_recovery_fences.get(&request.worker_id) { + if saved.request != request { + return Err(RuntimeError::InvalidRequest( + "different Subject create recovery is retained".into(), + )); + } + } else { + state.ensure_workspace_owner(scope, true)?; + state.subject_create_recovery_fences.insert( + request.worker_id, + SubjectCreateRecoveryFence { + request: request.clone(), + response: None, + }, + ); + } + // Never roll back admission exclusion, including uncertain persistence failures. + // Persist it before even checking backend availability. + state.persist_runtime_snapshot()?; + state.execution_backend.clone().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Subject create execution absence cannot be confirmed".into(), + ) + })? + }; + backend + .recover_subject_create(&request) + .map_err(RuntimeError::InvalidRequest)?; + let response = runtime_api::SubjectCreateRecoveryResponse { + workspace_id: request.workspace_id.clone(), + worker_id: request.worker_id, + create_fingerprint: request.create_fingerprint.clone(), + operation_id: request.operation_id.clone(), + }; + let mut state = self.lock()?; + state + .subject_create_recovery_fences + .get_mut(&request.worker_id) + .expect("recovery fence retained under Worker operation lock") + .response = Some(response.clone()); + state.persist_runtime_snapshot()?; + Ok(response) + } + /// Reserves the actual Controller while the Host reconciles final saved Journal material. /// A Runtime restart loses proof of a reservation, never reconstructs it from Worker status. pub fn subject_release_scoped( @@ -1262,6 +1347,8 @@ impl Runtime { let _operation_guard = operation_lock .lock() .map_err(|_| RuntimeError::StatePoisoned)?; + self.lock()? + .ensure_subject_create_not_fenced(request.worker_id)?; if let Some(existing) = self.existing_worker_for_create(&request, scope)? { return Ok(existing); } @@ -5482,6 +5569,7 @@ struct RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: u64, workers: BTreeMap, + subject_create_recovery_fences: BTreeMap, subject_release_operations: BTreeMap< (WorkerId, String), ( @@ -5519,6 +5607,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + subject_create_recovery_fences: BTreeMap::new(), subject_release_operations: BTreeMap::new(), workdir_operations: BTreeMap::new(), workspace_owners: BTreeMap::new(), @@ -5551,6 +5640,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + subject_create_recovery_fences: BTreeMap::new(), subject_release_operations: BTreeMap::new(), workdir_operations: BTreeMap::new(), workspace_owners: BTreeMap::new(), @@ -5697,6 +5787,7 @@ impl RuntimeState { workspace_backend_resource_clients: BTreeMap::new(), next_diagnostic_id, workers, + subject_create_recovery_fences: persisted.subject_create_recovery_fences, subject_release_operations: persisted .subject_releases .into_iter() @@ -5727,6 +5818,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] fn persisted_state(&self) -> PersistedRuntimeState { PersistedRuntimeState { + subject_create_recovery_fences: self.subject_create_recovery_fences.clone(), subject_releases: self.subject_release_operations.values().cloned().collect(), display_name: self.display_name.clone(), status: self.status, @@ -5871,11 +5963,21 @@ impl RuntimeState { Ok(()) } + fn ensure_subject_create_not_fenced(&self, worker_id: WorkerId) -> Result<(), RuntimeError> { + if self.subject_create_recovery_fences.contains_key(&worker_id) { + return Err(RuntimeError::InvalidRequest( + "Subject create identity is irreversibly fenced".into(), + )); + } + Ok(()) + } + fn ensure_subject_not_reserved( &self, worker: &WorkerRef, allow_disconnected: bool, ) -> Result<(), RuntimeError> { + self.ensure_subject_create_not_fenced(worker.worker_id)?; use runtime_api::SubjectReleaseState as State; if self .subject_release_operations @@ -5945,6 +6047,13 @@ impl RuntimeState { } fn forget_workspace_owner_if_unused(&mut self, workspace_id: &str) -> bool { + if self + .subject_create_recovery_fences + .values() + .any(|fence| fence.request.workspace_id == workspace_id) + { + return false; + } if self .workers .values() @@ -8293,6 +8402,8 @@ mod tests { #[derive(Default)] struct TestExecutionBackend { + recovery_absent: AtomicBool, + recovery_checks: std::sync::atomic::AtomicUsize, spawn_result: Mutex>, job_bindings: Mutex>>, dispatch_result: Mutex>, @@ -8367,6 +8478,18 @@ mod tests { } impl WorkerExecutionBackend for TestExecutionBackend { + fn recover_subject_create( + &self, + _request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> Result<(), String> { + self.recovery_checks.fetch_add(1, Ordering::SeqCst); + if self.recovery_absent.load(Ordering::SeqCst) { + Ok(()) + } else { + Err("execution absence is unconfirmed".into()) + } + } + fn subject_release( &self, request: &runtime_api::SubjectReleaseRequest, @@ -13098,7 +13221,7 @@ mod tests { }) .unwrap_err(); assert!(error.to_string().contains( - "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, or 11" + "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, or 12" )); let _ = std::fs::remove_dir_all(root); @@ -13140,7 +13263,7 @@ mod tests { let worker_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) .unwrap(); - assert_eq!(worker_aggregate["schema_version"], serde_json::json!(11)); + assert_eq!(worker_aggregate["schema_version"], serde_json::json!(12)); assert_eq!(worker_aggregate["status"], serde_json::json!("stopped")); assert_eq!( worker_aggregate["execution_state"]["state"], @@ -13280,6 +13403,213 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[cfg(feature = "fs-store")] + #[test] + fn subject_create_recovery_is_durable_exact_and_rechecks_absence() { + let root = tempfile::tempdir().unwrap(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path()) + .with_runtime_id("test-runtime"); + let scoped = scope("workspace-a", "server-a"); + let backend = Arc::new(TestExecutionBackend::default()); + backend.recovery_absent.store(true, Ordering::SeqCst); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); + let create = scoped_task_request("unpublished create", "workspace-a"); + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: scoped.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "recovery-1".into(), + }; + runtime.store_config_bundle(test_bundle()).unwrap(); + *backend.spawn_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::RolledBack( + WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + "create failed before publication", + ), + )); + assert!( + runtime + .create_worker_scoped(&scoped, create.clone()) + .is_err() + ); + assert!( + !runtime + .lock() + .unwrap() + .workers + .contains_key(&create.worker_id) + ); + let response = runtime + .recover_subject_create_scoped(&scoped, request.clone()) + .unwrap(); + assert_eq!(response.worker_id, request.worker_id); + assert!( + runtime + .create_worker_scoped(&scoped, create.clone()) + .is_err() + ); + assert!( + !runtime + .lock() + .unwrap() + .forget_workspace_owner_if_unused("workspace-a") + ); + assert_eq!( + runtime + .lock() + .unwrap() + .workspace_owners + .get("workspace-a") + .map(String::as_str), + Some("server-a") + ); + assert_eq!( + runtime + .recover_subject_create_scoped(&scoped, request.clone()) + .unwrap(), + response + ); + assert_eq!(backend.recovery_checks.load(Ordering::SeqCst), 2); + for mut mismatch in [request.clone(), request.clone(), request.clone()] + .into_iter() + .enumerate() + { + match mismatch.0 { + 0 => mismatch.1.create_fingerprint.push_str("-wrong"), + 1 => mismatch.1.operation_id.push_str("-wrong"), + _ => mismatch.1.workspace_id.push_str("-wrong"), + } + assert!( + runtime + .recover_subject_create_scoped(&scoped, mismatch.1) + .is_err() + ); + } + assert!( + runtime + .recover_subject_create_scoped( + &scope("workspace-a", "other-server"), + request.clone() + ) + .is_err() + ); + drop(runtime); + let backend = Arc::new(TestExecutionBackend::default()); + let reopened = + Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + assert_eq!(*backend.restore_count.lock().unwrap(), 0); + assert!(reopened.create_worker_scoped(&scoped, create).is_err()); + assert!( + reopened + .recover_subject_create_scoped(&scoped, request.clone()) + .is_err(), + "saved success cannot replace fresh absence proof" + ); + backend.recovery_absent.store(true, Ordering::SeqCst); + assert_eq!( + reopened + .recover_subject_create_scoped(&scoped, request) + .unwrap(), + response + ); + assert_eq!(backend.recovery_checks.load(Ordering::SeqCst), 2); + } + + #[cfg(feature = "fs-store")] + #[test] + fn subject_create_recovery_unknown_backend_retains_fence_and_owner() { + let root = tempfile::tempdir().unwrap(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path()) + .with_runtime_id("test-runtime"); + let scoped = scope("workspace-a", "server-a"); + let runtime = Runtime::with_fs_store(options.clone()).unwrap(); + let create = scoped_task_request("unpublished create", "workspace-a"); + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: scoped.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "recovery-unknown".into(), + }; + assert!( + runtime + .recover_subject_create_scoped(&scoped, request.clone()) + .is_err() + ); + assert!( + runtime + .lock() + .unwrap() + .subject_create_recovery_fences + .get(&request.worker_id) + .unwrap() + .response + .is_none() + ); + drop(runtime); + let backend = Arc::new(TestExecutionBackend::default()); + let reopened = + Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + assert!(reopened.create_worker_scoped(&scoped, create).is_err()); + assert!( + reopened + .recover_subject_create_scoped( + &scope("workspace-a", "other-server"), + request.clone() + ) + .is_err() + ); + assert!( + reopened + .recover_subject_create_scoped(&scoped, request.clone()) + .is_err() + ); + backend.recovery_absent.store(true, Ordering::SeqCst); + reopened + .recover_subject_create_scoped(&scoped, request) + .unwrap(); + } + + #[cfg(feature = "fs-store")] + #[test] + fn subject_create_recovery_rejects_existing_catalog_even_when_stopped() { + let root = tempfile::tempdir().unwrap(); + let backend = Arc::new(TestExecutionBackend::default()); + backend.recovery_absent.store(true, Ordering::SeqCst); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions::new(root.path()) + .with_runtime_id("test-runtime"), + backend.clone(), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scoped = scope("workspace-a", "server-a"); + let create = scoped_task_request("existing subject", "workspace-a"); + let worker = runtime + .create_worker_scoped(&scoped, create.clone()) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: scoped.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "not-unpublished".into(), + }; + assert!( + runtime + .recover_subject_create_scoped(&scoped, request) + .is_err() + ); + assert_eq!(backend.recovery_checks.load(Ordering::SeqCst), 0); + assert!( + runtime + .lock() + .unwrap() + .subject_create_recovery_fences + .is_empty() + ); + } + // Persistence component test; real Controller reservation has separate coverage. #[cfg(feature = "fs-store")] #[test] @@ -14164,7 +14494,7 @@ mod tests { ); let migrated_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - assert_eq!(migrated_aggregate["schema_version"], serde_json::json!(11)); + assert_eq!(migrated_aggregate["schema_version"], serde_json::json!(12)); assert!(migrated_aggregate.get("request").is_none()); assert!(migrated_aggregate.get("execution").is_none()); assert_eq!( diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 573f5c8e8..74044cde9 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -3369,6 +3369,42 @@ where &self.backend_id } + fn recover_subject_create( + &self, + request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> Result<(), String> { + let worker_ref = WorkerRef::new(request.worker_id); + let lock = self.worker_lock(&worker_ref)?; + let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; + if self + .runtime + .lock() + .map_err(|_| "adapter runtime poisoned")? + .is_none() + { + return Err("execution backend is unavailable".into()); + } + if self + .pending_restores + .lock() + .map_err(|_| "restore registry poisoned")? + .contains_key(&worker_ref) + { + return Err("Subject create has a pending execution candidate".into()); + } + if self + .workers + .lock() + .map_err(|_| "worker registry poisoned")? + .contains_key(&worker_ref) + { + // Even a closed protocol or stopped snapshot is residual execution, + // not absence. Recovery does not stop or remove anything. + return Err("Subject create execution is still present".into()); + } + Ok(()) + } + fn subject_release( &self, request: &runtime_api::SubjectReleaseRequest, @@ -6069,6 +6105,48 @@ mod tests { assert!(backend.workers.lock().unwrap().is_empty()); } + #[test] + fn subject_create_recovery_checks_actual_absence_and_pending_candidates() { + let root = tempfile::tempdir().unwrap(); + let backend = WorkerRuntimeExecutionBackend::new( + ProfileRuntimeWorkerFactory::new(root.path()) + .with_runtime_store_dir(root.path().join("runtime")), + ) + .unwrap(); + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + let recovery = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: "ws-test".into(), + worker_id: worker_ref.worker_id, + create_fingerprint: "failed-create".into(), + operation_id: "recovery-1".into(), + }; + backend.recover_subject_create(&recovery).unwrap(); + // No Controller exists, but unresolved task ownership is not absence. + backend.pending_restores.lock().unwrap().insert( + worker_ref.clone(), + Arc::new(PendingRuntimeRestore { + request: direct_restore_request(worker_ref.clone()), + task: tokio::sync::Mutex::new(None), + result: Mutex::new(None), + failure: Mutex::new(Some("task completion unknown".into())), + }), + ); + assert!( + backend + .recover_subject_create(&recovery) + .unwrap_err() + .contains("pending") + ); + assert!( + backend + .pending_restores + .lock() + .unwrap() + .contains_key(&worker_ref) + ); + assert!(backend.workers.lock().unwrap().is_empty()); + } + #[test] fn pending_restore_join_failure_never_becomes_absence_on_stop_retry() { let root = tempfile::tempdir().unwrap(); @@ -6127,6 +6205,18 @@ mod tests { fn stop_joins_healthy_self_terminated_controller_with_closed_endpoint() { let root = tempfile::tempdir().unwrap(); let (backend, request) = profile_stop_fixture(root.path()); + let recovery = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: "ws-test".into(), + worker_id: request.worker_ref.worker_id, + create_fingerprint: "failed-create".into(), + operation_id: "recovery-1".into(), + }; + assert!( + backend + .recover_subject_create(&recovery) + .unwrap_err() + .contains("still present") + ); let execution = backend .workers .lock() @@ -6167,8 +6257,16 @@ mod tests { }) .unwrap(); assert!(!execution.shutdown_requested.load(Ordering::Acquire)); + assert!( + backend + .recover_subject_create(&recovery) + .unwrap_err() + .contains("still present"), + "closed Controller is still residual execution" + ); assert!(backend.stop_worker(&request.worker_ref).is_accepted()); assert!(backend.workers.lock().unwrap().is_empty()); + backend.recover_subject_create(&recovery).unwrap(); assert!(execution.tasks.tasks.lock().unwrap().is_empty()); assert!(backend.stop_worker(&request.worker_ref).is_accepted()); } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index c731a5d66..6af8d337b 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -11298,6 +11298,7 @@ mod build_summary_prompt_tests { enum AttributionOutcome { Success, Rejected, + Unauthorized, TransportUnknown, } @@ -11344,6 +11345,16 @@ mod build_summary_prompt_tests { AttributionOutcome::TransportUnknown => Err(WorkspaceClientError::Request( "connection closed after request send".to_string(), )), + AttributionOutcome::Unauthorized => Ok(WorkspaceResponse { + status: 401, + body: serde_json::to_string(&server_api::RepositoryApiError::new( + 401, + "Unauthorized", + "invalid runtime request proof", + Vec::new(), + )) + .unwrap(), + }), AttributionOutcome::Rejected => Ok(WorkspaceResponse { status: 503, body: "temporarily unavailable".to_string(), @@ -11576,6 +11587,7 @@ mod build_summary_prompt_tests { for (first, expected_state) in [ (AttributionOutcome::TransportUnknown, "outcome_unknown"), (AttributionOutcome::Rejected, "failed"), + (AttributionOutcome::Unauthorized, "failed"), ] { let client = Arc::new(AttributionLifecycleClient::new([ first, @@ -11589,6 +11601,15 @@ mod build_summary_prompt_tests { ) .unwrap_err(); assert!(error.to_string().contains(expected_state)); + if matches!(first, AttributionOutcome::Unauthorized) { + assert!(error.to_string().contains("rejected with HTTP 401")); + assert!(!error.to_string().contains("outcome_unknown")); + assert_eq!( + client.requests.lock().unwrap().len(), + 1, + "no automatic retry after auth rejection" + ); + } let incomplete = store .read_by_name(&worker.manifest().worker.name) .unwrap() diff --git a/crates/workspace-server/src/assignment_tracking_tests.rs b/crates/workspace-server/src/assignment_tracking_tests.rs index 00bd30455..2e9085438 100644 --- a/crates/workspace-server/src/assignment_tracking_tests.rs +++ b/crates/workspace-server/src/assignment_tracking_tests.rs @@ -475,7 +475,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip (98, "Assignment-independent implementation reports".into()), (99, "Explicit hosting administrators and self-registration policy".into()), (100, "Provisional users and scoped passkey activation".into()), - (101, "Durable explicit Subject body release".into()) + (101, "Durable explicit Subject body release".into()), + (102, "Durable failed Subject creation recovery".into()) ] ); for (table, rows) in retained.iter().zip(original.iter()) { diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index b0f895adb..f6aea2db6 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1288,6 +1288,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { }) } + fn recover_subject_create( + &self, + _request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + Err("Subject create recovery fencing is unavailable on this Runtime".into()) + } + fn subject_release( &self, _request: runtime_api::SubjectReleaseRequest, @@ -2251,6 +2258,21 @@ impl RuntimeRegistry { runtime.worker_session_attachment(workspace_id, &worker_ref, session_id, attachment_id) } + pub fn recover_subject_create( + &self, + runtime_id: &str, + request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + self.runtime(runtime_id)? + .recover_subject_create(request) + .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.into(), + code: "recover_subject_create_unproven".into(), + message, + }) + } + pub fn subject_release( &self, runtime_id: &str, @@ -3319,6 +3341,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }) } + fn recover_subject_create( + &self, + request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + if request.workspace_id != self.workspace_id { + return Err("Subject create recovery Workspace mismatch".into()); + } + let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend"); + self.runtime + .recover_subject_create_scoped(&scope, request) + .map_err(|e| e.to_string()) + } + fn subject_release( &self, request: runtime_api::SubjectReleaseRequest, @@ -4164,7 +4199,10 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s if path.ends_with("/completions") { return "workers:read"; } - if path.contains("/retention/") || (path.starts_with("/v1/workers/") && method == "DELETE") { + if path.contains("/retention/") + || (path.starts_with("/v1/workers/") && method == "DELETE") + || (path == "/v1/subject-create-recovery" && method == "POST") + { return "workers:delete"; } if path.starts_with("/v1/session-public/") && method == "POST" { @@ -5561,6 +5599,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }) } + fn recover_subject_create( + &self, + request: runtime_api::SubjectCreateRecoveryRequest, + ) -> Result { + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.recover_subject_create(request).await }, + ) + .map_err(|diagnostic| diagnostic.message) + } + fn subject_release( &self, request: runtime_api::SubjectReleaseRequest, @@ -8377,6 +8427,51 @@ mod tests { ); } + #[test] + fn remote_subject_create_recovery_uses_generated_management_transport() { + assert_eq!( + workspace_runtime_operation("POST", "/v1/subject-create-recovery"), + "workers:delete" + ); + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: "workspace-test".into(), + worker_id: EmbeddedWorkerId::now_v7(), + create_fingerprint: "sha256:failed-create".into(), + operation_id: "recover-1".into(), + }; + // Remote adapter transport only; signed Runtime authorization is tested + // in worker-runtime's actual HTTP router suite. + let (base_url, server) = serve_mock_http(vec![mock_response( + "POST", + "/v1/subject-create-recovery", + true, + 200, + serde_json::to_string(&request).unwrap(), + )]); + let registry = RuntimeRegistry::new(Vec::new()); + registry.register( + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:recover", + "Recovery transport fixture", + base_url, + Some("secret-token".into()), + ), + "workspace-test".into(), + "http://127.0.0.1:8787".into(), + ) + .unwrap(), + ); + let response = registry + .recover_subject_create("remote:recover", request.clone()) + .unwrap(); + assert_eq!(response.workspace_id, request.workspace_id); + assert_eq!(response.worker_id, request.worker_id); + assert_eq!(response.create_fingerprint, request.create_fingerprint); + assert_eq!(response.operation_id, request.operation_id); + server.join().unwrap(); + } + #[test] fn remote_session_erasure_uses_the_generated_management_transport() { // The loopback fixture substitutes only the remote Runtime. This covers diff --git a/crates/workspace-server/src/implementation_result_tests.rs b/crates/workspace-server/src/implementation_result_tests.rs index 8991f7356..c5a7cd7d2 100644 --- a/crates/workspace-server/src/implementation_result_tests.rs +++ b/crates/workspace-server/src/implementation_result_tests.rs @@ -381,6 +381,8 @@ async fn implementation_result_v97_to_v98_preserves_legacy_bytes_receipts_and_no DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations; + DROP TABLE subject_create_recoveries; + UPDATE __yoi_schema_migrations SET version=101 WHERE version=102; DROP TABLE subject_worker_releases; UPDATE __yoi_schema_migrations SET version=100 WHERE version=101; DROP TABLE user_activations; @@ -515,6 +517,8 @@ async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknow // Reconstruct the prior composition before its new business capture existed. conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER IF EXISTS implementation_delegation_capture; DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations; + DROP TABLE subject_create_recoveries; + UPDATE __yoi_schema_migrations SET version=101 WHERE version=102; DROP TABLE subject_worker_releases; UPDATE __yoi_schema_migrations SET version=100 WHERE version=101; DROP TABLE user_activations; @@ -675,7 +679,9 @@ async fn mr_event_actors_preserve_self_suppression_in_capture_and_backfill() { if historical { conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER IF EXISTS implementation_delegation_capture; DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations; - DROP TABLE subject_worker_releases; + DROP TABLE subject_create_recoveries; + UPDATE __yoi_schema_migrations SET version=101 WHERE version=102; + DROP TABLE subject_worker_releases; UPDATE __yoi_schema_migrations SET version=100 WHERE version=101; DROP TABLE user_activations; UPDATE __yoi_schema_migrations SET version=99 WHERE version=100; diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index ce49d17e7..489e78728 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -699,6 +699,19 @@ CREATE TABLE worker_create_reservations ( FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, CHECK ((singleton_key IS NULL) = (singleton_generation IS NULL)) ); +CREATE TABLE subject_create_recoveries ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + singleton_key TEXT NOT NULL, + singleton_generation INTEGER NOT NULL CHECK (singleton_generation > 0), + operation_id TEXT NOT NULL CHECK (length(operation_id) > 0), + create_fingerprint TEXT NOT NULL, + completed INTEGER NOT NULL DEFAULT 0 CHECK (completed IN (0,1)), + PRIMARY KEY (workspace_id, operation_id), + UNIQUE (workspace_id, runtime_id, worker_id, singleton_key, singleton_generation), + FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_create_reservations(workspace_id, worker_id) ON DELETE CASCADE +); CREATE TABLE subject_worker_releases ( workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index b41e2dcf6..dd256fea9 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2934,10 +2934,10 @@ fn repository_api_rejection(path: &str, status: StatusCode, message: &str) -> Re if path.starts_with("/api/w/") && path.contains("/drive/") { return (status, Json(drive::failure(status.as_u16()))).into_response(); } - if path == "/api/repositories" - || path.starts_with("/api/repositories/") - || (path.starts_with("/api/w/") && path.contains("/repositories")) - { + // Generated Workspace API clients share this error envelope, not only + // repository endpoints. A bare 401 here becomes a response-decode failure + // and incorrectly turns a known auth rejection into an unknown outcome. + if path.starts_with("/api/") { return ( status, Json(server_api::RepositoryApiError::new( @@ -5484,6 +5484,12 @@ impl WorkspaceApi { let subject = singleton_key .as_deref() .and_then(|key| key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX)); + if let Some(subject) = subject + && let Err(error) = subjektiv_journal::recover_failed_subject_create(self, subject) + { + let diagnostics = cleanup_spawn_created_workdirs(self, &spawned_workdir_ids); + return Err(ApiError::with_diagnostics(error, diagnostics)); + } let subject_gate = subject .map(|s| subjektiv_journal::lifecycle_lock(self, s)) .transpose()?; @@ -22372,15 +22378,12 @@ async fn scoped_record_subjektiv_session( .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let attribution = if request.create_if_missing { - let attribution = subjektiv_session_attribution( + subjektiv_session_attribution( &api, &path.workspace_id, &context, &request.session_id, - )?; - open_subjektiv_store(&api)? - .record_session_attribution(attribution) - .map_err(|error| Error::InvalidInput(error.to_string()))? + )? } else { require_existing_subjektiv_session_attribution( &api, @@ -47939,6 +47942,9 @@ mod tests { >, materializer: worker_runtime::working_directory::RuntimeGitMaterializer, spawn_failure: std::sync::Mutex>, + recovery_failure: std::sync::Mutex>, + recovery_requests: std::sync::Mutex>, + recovery_hook: std::sync::Mutex>>, input_failure: std::sync::Mutex>, cancel_failure: std::sync::Mutex>, input_hook: std::sync::Mutex< @@ -47974,6 +47980,9 @@ mod tests { std::env::temp_dir().join(unique), ), spawn_failure: std::sync::Mutex::new(None), + recovery_failure: std::sync::Mutex::new(None), + recovery_requests: std::sync::Mutex::new(Vec::new()), + recovery_hook: std::sync::Mutex::new(None), input_failure: std::sync::Mutex::new(None), cancel_failure: std::sync::Mutex::new(None), input_hook: std::sync::Mutex::new(None), @@ -48068,6 +48077,29 @@ mod tests { "deterministic-workspace-server-test" } + fn recover_subject_create( + &self, + request: &runtime_api::SubjectCreateRecoveryRequest, + ) -> std::result::Result<(), String> { + self.recovery_requests.lock().unwrap().push(request.clone()); + if let Some(hook) = self.recovery_hook.lock().unwrap().take() { + hook(); + } + if let Some(error) = self.recovery_failure.lock().unwrap().take() { + return Err(error); + } + if self + .contexts + .lock() + .unwrap() + .keys() + .any(|worker| worker.worker_id == request.worker_id) + { + return Err("actual execution remains".into()); + } + Ok(()) + } + fn create_working_directory( &self, request: &worker_runtime::catalog::WorkingDirectoryRequest, diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index 040e9ccf3..5abfd49b2 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -346,6 +346,11 @@ pub(super) fn bind_host_session( } } } + // Publication (including attribution) shares the recovery gate. A callback + // authenticated before a recovery intent must not leave a late attribution. + let attr = store + .record_session_attribution(attr) + .map_err(|error| Error::InvalidInput(error.to_string()))?; let writer = store .claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id) .map_err(domain)?; @@ -386,6 +391,93 @@ pub(super) fn check_start_allowed(api: &WorkspaceApi, subject: &str) -> Result<( Ok(()) } +/// Explicit creation can reconcile an earlier failed, never-published body. +/// No missing catalog/status observation is treated as a Runtime fencing receipt. +pub(super) fn recover_failed_subject_create(api: &WorkspaceApi, subject: &str) -> Result<()> { + let continuation = lifecycle_lock(api, &format!("release-continuation:{subject}"))?; + let _continuation = continuation + .lock() + .map_err(|_| Error::Store("release continuation poisoned".into()))?; + let gate = lifecycle_lock(api, subject)?; + let subject_key = crate::subjektiv::subject_worker_singleton_key(subject) + .map_err(|e| Error::InvalidInput(e.to_string()))?; + let store = store(api)?; + let (lease, request) = { + let _guard = gate + .lock() + .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?; + let Some(lease) = api + .config_store + .current_worker_singleton_owner(api.workspace_id(), &subject_key)? + else { + return Ok(()); + }; + if api + .config_store + .subject_create_recovery_candidate(api.workspace_id(), &lease)? + .is_none() + { + return Ok(()); + } + if !store + .journal_body_is_unpublished(subject, &lease.worker.runtime_id, &lease.worker.worker_id) + .map_err(domain)? + { + // Published history must go through the ordinary continuity release. + return Ok(()); + } + let operation_id = format!( + "subject-create-recovery:{}:{}", + lease.worker.worker_id, lease.generation + ); + let fingerprint = api.config_store.prepare_subject_create_recovery( + api.workspace_id(), + &lease, + &operation_id, + )?; + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: api.workspace_id().into(), + worker_id: worker_runtime::identity::WorkerId::parse(&lease.worker.worker_id) + .ok_or_else(|| Error::Store("invalid retained Worker ID".into()))?, + operation_id, + create_fingerprint: fingerprint, + }; + (lease, request) + }; + // Never hold the callback lane/gate across Runtime I/O. Unknown outcome + // leaves the durable intent in place and reuses the exact identity next time. + let response = api + .runtime + .recover_subject_create(&lease.worker.runtime_id, request.clone()) + .map_err(RuntimeRegistryError::into_error)?; + if response.workspace_id != request.workspace_id + || response.worker_id != request.worker_id + || response.operation_id != request.operation_id + || response.create_fingerprint != request.create_fingerprint + { + return Err(Error::RepositoryConflict( + "Subject create recovery acknowledgment mismatch; retain occupancy".into(), + )); + } + let _guard = gate + .lock() + .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?; + if !store + .journal_body_is_unpublished(subject, &lease.worker.runtime_id, &lease.worker.worker_id) + .map_err(domain)? + { + return Err(Error::RepositoryConflict( + "Subject publication appeared during recovery; retain occupancy".into(), + )); + } + api.config_store.complete_subject_create_recovery( + api.workspace_id(), + &lease, + &request.operation_id, + &request.create_fingerprint, + ) +} + pub(super) fn lifecycle_lock(api: &WorkspaceApi, subject: &str) -> Result>> { let mut locks = api .journal_lifecycle diff --git a/crates/workspace-server/src/server/tests/auth_logging_tests.rs b/crates/workspace-server/src/server/tests/auth_logging_tests.rs index d5bf3334a..fcb0ca9e8 100644 --- a/crates/workspace-server/src/server/tests/auth_logging_tests.rs +++ b/crates/workspace-server/src/server/tests/auth_logging_tests.rs @@ -197,7 +197,20 @@ async fn rejected_runtime_proofs_log_reason_and_verification_without_secrets_on_ .unwrap(); assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "{reason}"); let response_body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); - assert!(response_body.is_empty(), "{response_body:?}"); + let error: server_api::RepositoryApiError = + serde_json::from_slice(&response_body).unwrap(); + assert_eq!(error.error, "Unauthorized"); + assert_eq!(error.message, "invalid runtime request proof"); + assert_eq!( + serde_json::from_slice::(&response_body).unwrap(), + serde_json::to_value(server_api::RepositoryApiError::new( + 401, + "Unauthorized", + "invalid runtime request proof", + Vec::new() + )) + .unwrap() + ); let (event, text) = logs.event(); assert_eq!(event["reason"], reason); assert_eq!(event["proof_verification"], verification); @@ -338,6 +351,124 @@ async fn runtime_proof_setup_failure_logs_no_internal_error_and_bounds_request_m } } +#[tokio::test] +async fn common_api_rejections_use_json_without_changing_drive_or_non_api_contracts() { + for status in [ + StatusCode::BAD_REQUEST, + StatusCode::UNAUTHORIZED, + StatusCode::FORBIDDEN, + StatusCode::NOT_FOUND, + StatusCode::CONFLICT, + StatusCode::SERVICE_UNAVAILABLE, + ] { + for path in [ + "/api/w/ws/subjektiv/sessions", + "/api/w/ws/workers/self/workdir-session/operations", + "/api/runtime/v1/workspaces/ws/resources/fetch", + "/api/repositories", + ] { + let response = repository_api_rejection(path, status, "request rejected"); + assert_eq!(response.status(), status); + assert_eq!(response.headers()[CONTENT_TYPE], "application/json"); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&bytes).unwrap(), + serde_json::to_value(server_api::RepositoryApiError::new( + status.as_u16(), + status.canonical_reason().unwrap(), + "request rejected", + Vec::new() + )) + .unwrap() + ); + } + let drive = repository_api_rejection("/api/w/ws/drive/mutate", status, "request rejected"); + assert_eq!(drive.status(), status); + assert_eq!( + serde_json::from_slice::( + &to_bytes(drive.into_body(), usize::MAX).await.unwrap() + ) + .unwrap(), + serde_json::to_value(drive::failure(status.as_u16())).unwrap() + ); + let shell = repository_api_rejection("/w/ws", status, "request rejected"); + assert_eq!(shell.status(), status); + assert!( + to_bytes(shell.into_body(), usize::MAX) + .await + .unwrap() + .is_empty() + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn subjektiv_auth_rejection_survives_runtime_generated_client_on_both_routers() { + use worker::WorkspaceClient; + struct AbortServer(tokio::task::JoinHandle<()>); + impl Drop for AbortServer { + fn drop(&mut self) { + self.0.abort(); + } + } + + let workspace = tempfile::tempdir().unwrap(); + let mut api = test_api(workspace.path()).await; + let trusted = RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + configure_runtime_request_auth(&mut api, &trusted, "runtime-test"); + let untrusted = RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + for app in [ + build_router(api.clone()), + workspace_server_router(WorkspaceServerApi::new_for_test( + api.config.clone(), + api.store.clone(), + )), + ] { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let _server = AbortServer(tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + })); + // Exercise missing authentication, signature rejection and a valid proof + // whose Worker has no catalog membership/reservation. + for identity in [None, Some(&untrusted), Some(&trusted)] { + let client = worker_runtime::worker_source::RuntimeOwnedWorkspaceClient::new( + TEST_WORKSPACE_ID, + &url, + "runtime-test", + "unregistered-worker", + ); + let client = match identity { + Some(identity) => client.with_runtime_request_source(identity, "server-test"), + None => client, + }; + let response = tokio::task::spawn_blocking(move || { + client.execute_server_operation(worker::WorkspaceServerOperation::SubjektivRecordSession( + server_api::SubjektivRecordSessionRequest { + session_id: session_store::new_session_id().to_string(), create_if_missing: true, + }, + )) + }).await.unwrap().expect("an explicit HTTP rejection must remain a WorkspaceResponse, not an unknown transport outcome"); + assert_eq!(response.status, 401); + let error: server_api::RepositoryApiError = + serde_json::from_str(&response.body).unwrap(); + assert_eq!(error.error, "Unauthorized"); + assert_eq!( + error.message, + if identity.is_some() { + "invalid runtime request proof" + } else { + "authentication required" + } + ); + assert!(!response.body.contains("unregistered-worker")); + if let Some(identity) = identity { + assert!(!response.body.contains(&identity.private_key)); + } + } + } +} + #[tokio::test] async fn repository_runtime_proof_401_keeps_generic_envelope_on_both_routers() { let workspace = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs index d3564fd9d..f15ac57cc 100644 --- a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs @@ -330,7 +330,7 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele .reserve_worker_create( TEST_WORKSPACE_ID, EMBEDDED_RUNTIME_ID, - "rejected-create", + "manual:rejected-create", &"f".repeat(64), Some("subjektiv:rejected-create"), &memory, @@ -378,5 +378,274 @@ async fn rejected_create_releases_key_but_retains_subject_occupancy_without_rele .worker, worker, ); + // Retained Subject occupancy is not a live create reservation. A new + // manual request must not send the terminal Worker's identity to Runtime. + let retry = api.config_store.reserve_worker_create( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + "manual:retry-rejected-create", + &"f".repeat(64), + Some("subjektiv:rejected-create"), + &memory, + ); + assert!( + matches!(retry, Err(Error::RepositoryConflict(ref message)) + if message.contains("worker_singleton_release_required")), + "terminal reservation must fail before Runtime dispatch: {retry:?}" + ); } } + +async fn unpublished_recovery_fixture( + root: &std::path::Path, +) -> ( + WorkspaceApi, + Arc, + String, + crate::store::WorkerCreateReservation, +) { + let (api, backend) = test_api_with_recording_backend(root).await; + let subject = open_subjektiv_store(&api) + .unwrap() + .create_subject(crate::subjektiv::SubjectRole::new("recovery-test").unwrap()) + .unwrap(); + let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); + let memory = api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let old = api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + "manual:failed-subject", + &"f".repeat(64), + Some(&key), + &memory, + ) + .unwrap(); + api.config_store + .fail_worker_create_reservation( + TEST_WORKSPACE_ID, + EMBEDDED_RUNTIME_ID, + old.worker_id, + &old.create_fingerprint, + ) + .unwrap(); + (api, backend, subject.id, old) +} + +fn recovery_spawn_request(subject: &str) -> WorkerSpawnRequest { + serde_json::from_value(json!({ + "intent": WorkerSpawnIntent::WorkspaceCompanion, + "acceptance": WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0 }, + "profile": worker_runtime::catalog::ProfileSelector::Builtin("builtin:companion".into()), + "singleton_key": crate::subjektiv::subject_worker_singleton_key(subject).unwrap(), + })) + .unwrap() +} + +#[tokio::test] +async fn unpublished_create_recovery_retries_same_fence_then_creates_new_worker() { + let root = tempfile::tempdir().unwrap(); + let (api, backend, subject, old) = unpublished_recovery_fixture(root.path()).await; + let old_worker = RuntimeWorkerRef::new(EMBEDDED_RUNTIME_ID, old.worker_id.to_string()); + let lease = old.singleton.as_ref().unwrap(); + let late = crate::subjektiv::SubjectSessionAttribution::new( + &subject, + EMBEDDED_RUNTIME_ID, + old.worker_id.to_string(), + session_store::new_session_id().to_string(), + ) + .unwrap(); + let late_session = late.session_id.clone(); + // Models an already-authorized callback arriving while Runtime fencing is + // in flight. The Host intent must reject it BEFORE attribution insertion. + let callback_api = api.clone(); + *backend.recovery_hook.lock().unwrap() = Some(Box::new(move || { + assert!(subjektiv_journal::bind_host_session(&callback_api, late).is_err()); + })); + *backend.recovery_failure.lock().unwrap() = Some("unconfirmed execution absence".into()); + assert!( + api.spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject)) + .is_err() + ); + assert!( + api.config_store + .subject_release_pending(TEST_WORKSPACE_ID, &old_worker) + .unwrap() + ); + assert_eq!( + api.config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &lease.key) + .unwrap(), + Some(lease.clone()) + ); + let store = open_subjektiv_store(&api).unwrap(); + assert!(store.session_attribution(&late_session).unwrap().is_none()); + assert!( + store + .journal_body_is_unpublished(&subject, EMBEDDED_RUNTIME_ID, &old_worker.worker_id) + .unwrap() + ); + + let replacement = api + .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject)) + .unwrap(); + assert_eq!(replacement.state, InternalWorkerOperationState::Accepted); + let replacement = replacement.worker.unwrap().worker; + assert_ne!(replacement, old_worker); + let current = api + .config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &lease.key) + .unwrap() + .unwrap(); + assert_eq!(current.worker, replacement); + assert_eq!(current.generation, lease.generation + 1); + let requests = backend.recovery_requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0], requests[1], + "unknown result must reuse exact recovery identity" + ); + // A stale completion cannot release the replacement generation. + api.config_store + .complete_subject_create_recovery( + TEST_WORKSPACE_ID, + lease, + &requests[0].operation_id, + &old.create_fingerprint, + ) + .unwrap(); + assert_eq!( + api.config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &lease.key) + .unwrap(), + Some(current) + ); +} + +#[tokio::test] +async fn unpublished_create_recovery_refuses_any_attribution_or_orphan_writer() { + for orphan_writer in [false, true] { + let root = tempfile::tempdir().unwrap(); + let (api, backend, subject, old) = unpublished_recovery_fixture(root.path()).await; + let store = open_subjektiv_store(&api).unwrap(); + if orphan_writer { + store + .claim_journal_writer(&subject, EMBEDDED_RUNTIME_ID, &old.worker_id.to_string()) + .unwrap(); + } else { + store + .record_session_attribution( + crate::subjektiv::SubjectSessionAttribution::new( + &subject, + EMBEDDED_RUNTIME_ID, + old.worker_id.to_string(), + session_store::new_session_id().to_string(), + ) + .unwrap(), + ) + .unwrap(); + } + assert!( + api.spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject)) + .is_err() + ); + assert!(backend.recovery_requests.lock().unwrap().is_empty()); + assert_eq!( + api.config_store + .current_worker_singleton_owner( + TEST_WORKSPACE_ID, + &old.singleton.as_ref().unwrap().key + ) + .unwrap(), + old.singleton + ); + } +} + +#[tokio::test] +async fn unpublished_create_recovery_resumes_after_runtime_proof_before_host_commit() { + let root = tempfile::tempdir().unwrap(); + let (api, backend, subject, old) = unpublished_recovery_fixture(root.path()).await; + let lease = old.singleton.as_ref().unwrap(); + let op = format!( + "subject-create-recovery:{}:{}", + old.worker_id, lease.generation + ); + let fingerprint = api + .config_store + .prepare_subject_create_recovery(TEST_WORKSPACE_ID, lease, &op) + .unwrap(); + let request = runtime_api::SubjectCreateRecoveryRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker_id: old.worker_id, + create_fingerprint: fingerprint, + operation_id: op, + }; + api.runtime + .recover_subject_create(EMBEDDED_RUNTIME_ID, request) + .unwrap(); + // Simulate loss of the receipt before Host completion. The Host must not + // infer completion from missing catalog state, and can retry the exact op. + assert_eq!( + api.config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &lease.key) + .unwrap(), + old.singleton + ); + subjektiv_journal::recover_failed_subject_create(&api, &subject).unwrap(); + assert!( + api.config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &lease.key) + .unwrap() + .is_none() + ); + assert_eq!(backend.recovery_requests.lock().unwrap().len(), 2); +} + +#[tokio::test] +async fn failed_subject_spawn_can_recover_on_next_explicit_create() { + let root = tempfile::tempdir().unwrap(); + let (api, backend) = test_api_with_recording_backend(root.path()).await; + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("failed-spawn").unwrap()) + .unwrap(); + let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(); + *backend.spawn_failure.lock().unwrap() = + Some("Session registration failed before publication".into()); + let rejected = api + .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) + .unwrap(); + assert_eq!(rejected.state, InternalWorkerOperationState::Rejected); + let old = api + .config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap(); + assert!( + api.config_store + .subject_create_recovery_candidate(TEST_WORKSPACE_ID, &old) + .unwrap() + .is_some() + ); + assert!( + backend.recovery_requests.lock().unwrap().is_empty(), + "failure alone must not release occupancy" + ); + let result = api + .spawn_workspace_worker(EMBEDDED_RUNTIME_ID, recovery_spawn_request(&subject.id)) + .unwrap(); + assert_eq!(result.state, InternalWorkerOperationState::Accepted); + let current = api + .config_store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap(); + assert_ne!(current.worker, old.worker); + assert_eq!(current.generation, old.generation + 1); + assert_eq!(backend.recovery_requests.lock().unwrap().len(), 1); +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 25dccf56d..b9dd24f8a 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -119,7 +119,7 @@ pub use ticket::{ }; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 101; +const LATEST_SCHEMA_VERSION: i64 = 102; const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants"; const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs"; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; @@ -432,6 +432,11 @@ const MIGRATIONS: &[Migration] = &[ name: "Durable explicit Subject body release", apply: migrate_subject_release_v100_to_v101, }, + Migration { + version: 102, + name: "Durable failed Subject creation recovery", + apply: migrate_subject_create_recovery_v101_to_v102, + }, ]; #[derive(Clone, Copy)] @@ -2997,7 +3002,7 @@ impl SqliteWorkspaceStore { .query_row( "SELECT allocation_key, request_fingerprint, create_fingerprint, \ memory_language, singleton_key, \ - singleton_generation \ + singleton_generation, state \ FROM worker_create_reservations \ WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", params![ @@ -3013,11 +3018,21 @@ impl SqliteWorkspaceStore { row.get::<_, Option>(3)?, row.get::<_, Option>(4)?, row.get::<_, Option>(5)?, + row.get::<_, String>(6)?, )) }, ) .optional()?; - if let Some((stored_allocation_key, stored_request_fingerprint, create_fingerprint, language, stored_singleton_key, stored_generation)) = stored + // Subject occupancy survives removal until an explicit release + // proof. It must not resurrect the removed creation reservation: + // Runtime callbacks require a live reservation or catalog member. + if stored.as_ref().is_some_and(|row| row.6 == "removed") { + return Err(Error::RepositoryConflict(format!( + "worker_singleton_release_required: Worker {}:{} has a terminal create reservation but still owns the Subject; complete Subject release before creating a replacement", + owner.worker.runtime_id, owner.worker.worker_id + ))); + } + if let Some((stored_allocation_key, stored_request_fingerprint, create_fingerprint, language, stored_singleton_key, stored_generation, _)) = stored && allocation_key.starts_with("manual:") && stored_allocation_key.starts_with("manual:") && owner.worker.runtime_id == runtime_id @@ -3199,6 +3214,99 @@ impl SqliteWorkspaceStore { }) } + /// Eligibility is Backend evidence only. The Host must separately prove that Subjektiv + /// never published this body and obtain the Runtime's exact create-fence receipt. + pub(crate) fn subject_create_recovery_candidate( + &self, + workspace: &str, + lease: &WorkerSingletonLease, + ) -> Result> { + self.with_conn(|conn| subject_create_recovery_candidate(conn, workspace, lease)) + } + + pub(crate) fn prepare_subject_create_recovery( + &self, + workspace: &str, + lease: &WorkerSingletonLease, + operation_id: &str, + ) -> Result { + if operation_id.trim().is_empty() || operation_id.len() > 512 { + return Err(Error::InvalidInput( + "invalid Subject create recovery operation".into(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let fingerprint = subject_create_recovery_candidate(&tx, workspace, lease)? + .ok_or_else(|| Error::RepositoryConflict("Subject create recovery requires the current removed reservation without a registry row".into()))?; + let saved: Option<(String, String, String, u64, String)> = tx.query_row( + "SELECT runtime_id,worker_id,singleton_key,singleton_generation,create_fingerprint FROM subject_create_recoveries WHERE workspace_id=?1 AND operation_id=?2", + params![workspace, operation_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)), + ).optional()?; + if let Some(saved) = saved { + if saved != (lease.worker.runtime_id.clone(), lease.worker.worker_id.clone(), lease.key.clone(), lease.generation, fingerprint.clone()) { + return Err(Error::RepositoryConflict("Subject create recovery operation was used with different input".into())); + } + return Ok(fingerprint); + } + let busy: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM subject_create_recoveries WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5)", + params![workspace, lease.worker.runtime_id, lease.worker.worker_id, lease.key, lease.generation as i64], + |row| row.get(0), + )?; + if busy { + return Err(Error::RepositoryConflict("Subject create recovery already has a durable operation".into())); + } + tx.execute( + "INSERT INTO subject_create_recoveries(workspace_id,runtime_id,worker_id,singleton_key,singleton_generation,operation_id,create_fingerprint) VALUES(?1,?2,?3,?4,?5,?6,?7)", + params![workspace, lease.worker.runtime_id, lease.worker.worker_id, lease.key, lease.generation as i64, operation_id, fingerprint], + )?; + tx.commit()?; + Ok(fingerprint) + }) + } + + /// Called only after the Host verifies the Runtime's typed receipt for this exact create. + /// The completed bit records that proof; it neither creates a writer binding nor rewrites + /// the reservation. An exact completed replay is a receipt, not authority over a successor. + pub(crate) fn complete_subject_create_recovery( + &self, + workspace: &str, + lease: &WorkerSingletonLease, + operation_id: &str, + create_fingerprint: &str, + ) -> Result<()> { + let generation = i64::try_from(lease.generation) + .map_err(|_| Error::InvalidInput("invalid singleton generation".into()))?; + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let completed: Option = tx.query_row( + "SELECT completed FROM subject_create_recoveries WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND operation_id=?6 AND create_fingerprint=?7", + params![workspace, lease.worker.runtime_id, lease.worker.worker_id, lease.key, generation, operation_id, create_fingerprint], + |row| row.get(0), + ).optional()?; + match completed { + Some(true) => return Ok(()), + Some(false) => {}, + None => return Err(Error::RepositoryConflict("Subject create recovery intent absent or mismatched".into())), + } + let current = subject_create_recovery_candidate(&tx, workspace, lease)?; + if current.as_deref() != Some(create_fingerprint) { + return Err(Error::RepositoryConflict("Subject create recovery reservation or singleton ownership changed".into())); + } + let changed = tx.execute( + "UPDATE subject_create_recoveries SET completed=1 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND operation_id=?6 AND create_fingerprint=?7 AND completed=0", + params![workspace, lease.worker.runtime_id, lease.worker.worker_id, lease.key, generation, operation_id, create_fingerprint], + )?; + if changed != 1 { + return Err(Error::RepositoryConflict("Subject create recovery completion lost its exact intent".into())); + } + tx.commit()?; + Ok(()) + }) + } + pub(crate) fn retained_subject_release_operation( &self, workspace: &str, @@ -3213,7 +3321,7 @@ impl SqliteWorkspaceStore { workspace: &str, worker: &RuntimeWorkerRef, ) -> Result { - self.with_conn(|c| Ok(c.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND cancelled=0 AND completed=0)",params![workspace,worker.runtime_id,worker.worker_id],|r|r.get(0))?)) + self.with_conn(|c| Ok(c.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND cancelled=0 AND completed=0) OR EXISTS(SELECT 1 FROM subject_create_recoveries WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND completed=0)",params![workspace,worker.runtime_id,worker.worker_id],|r|r.get(0))?)) } pub(crate) fn require_subject_release_intent( @@ -15012,6 +15120,34 @@ fn migrate_result_reports_v97_to_v98(conn: &Connection) -> Result<()> { Ok(()) } +fn migrate_subject_create_recovery_v101_to_v102(conn: &Connection) -> Result<()> { + if current_schema_version(conn)? != 101 { + return Err(Error::Store( + "Subject create recovery migration requires schema 101".into(), + )); + } + let tx = conn.unchecked_transaction()?; + tx.execute_batch("CREATE TABLE subject_create_recoveries ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + singleton_key TEXT NOT NULL, + singleton_generation INTEGER NOT NULL CHECK (singleton_generation > 0), + operation_id TEXT NOT NULL CHECK (length(operation_id) > 0), + create_fingerprint TEXT NOT NULL, + completed INTEGER NOT NULL DEFAULT 0 CHECK (completed IN (0,1)), + PRIMARY KEY (workspace_id, operation_id), + UNIQUE (workspace_id, runtime_id, worker_id, singleton_key, singleton_generation), + FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_create_reservations(workspace_id, worker_id) ON DELETE CASCADE +);")?; + tx.execute( + "INSERT INTO __yoi_schema_migrations(version,name) VALUES(102,?1)", + ["Durable failed Subject creation recovery"], + )?; + tx.commit()?; + Ok(()) +} + fn migrate_subject_release_v100_to_v101(conn: &Connection) -> Result<()> { if current_schema_version(conn)? != 100 { return Err(Error::Store( @@ -15702,6 +15838,33 @@ pub(crate) fn validate_worker_singleton_key(key: &str) -> Result<()> { Ok(()) } +fn subject_create_recovery_candidate( + conn: &Connection, + workspace: &str, + lease: &WorkerSingletonLease, +) -> Result> { + if !lease + .key + .starts_with(crate::subjektiv::SUBJECT_WORKER_SINGLETON_PREFIX) + { + return Err(Error::InvalidInput("not a Subject singleton".into())); + } + let generation = i64::try_from(lease.generation) + .map_err(|_| Error::InvalidInput("invalid singleton generation".into()))?; + Ok(conn.query_row( + "SELECT r.create_fingerprint FROM worker_create_reservations r + JOIN worker_singleton_owners o ON o.workspace_id=r.workspace_id + AND o.runtime_id=r.runtime_id AND o.worker_id=r.worker_id + AND o.singleton_key=r.singleton_key AND o.generation=r.singleton_generation + WHERE r.workspace_id=?1 AND r.runtime_id=?2 AND r.worker_id=?3 + AND r.singleton_key=?4 AND r.singleton_generation=?5 AND r.state='removed' + AND NOT EXISTS(SELECT 1 FROM worker_registry w + WHERE w.workspace_id=r.workspace_id AND w.runtime_id=r.runtime_id AND w.worker_id=r.worker_id)", + params![workspace, lease.worker.runtime_id, lease.worker.worker_id, lease.key, generation], + |row| row.get(0), + ).optional()?) +} + fn current_worker_singleton_owner( conn: &Connection, workspace_id: &str, @@ -15712,6 +15875,7 @@ fn current_worker_singleton_owner( "SELECT owner.runtime_id, owner.worker_id, owner.generation, \ CASE WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \ AND NOT EXISTS(SELECT 1 FROM subject_worker_releases r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.completed=1) \ + AND NOT EXISTS(SELECT 1 FROM subject_create_recoveries r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.create_fingerprint=reservation.create_fingerprint AND r.completed=1) \ THEN 'created' ELSE reservation.state END \ FROM worker_singleton_owners owner \ LEFT JOIN worker_create_reservations reservation \ @@ -18739,6 +18903,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx WorkspaceSchemaMigrationStep {version:99, name:"Explicit hosting administrators and self-registration policy".into()}, WorkspaceSchemaMigrationStep {version:100, name:"Provisional users and scoped passkey activation".into()}, WorkspaceSchemaMigrationStep {version:101, name:"Durable explicit Subject body release".into()}, + WorkspaceSchemaMigrationStep {version:102, name:"Durable failed Subject creation recovery".into()}, ] ); @@ -18850,6 +19015,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx (99, "Explicit hosting administrators and self-registration policy".to_string()), (100, "Provisional users and scoped passkey activation".to_string()), (101, "Durable explicit Subject body release".to_string()), +(102, "Durable failed Subject creation recovery".to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -20907,6 +21073,47 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx conflict, Error::RepositoryConflict(message) if message.contains("worker_singleton_owned") )); + + reopened + .fail_worker_create_reservation( + "workspace-a", + "runtime-a", + reserved.worker_id, + &reserved.create_fingerprint, + ) + .unwrap(); + drop(reopened); + let reopened = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + let conflict = reopened + .reserve_worker_create( + "workspace-a", + "runtime-a", + "manual:retry-after-terminal-failure", + &stable_fingerprint, + Some("subjektiv:restart-retry"), + &memory, + ) + .unwrap_err(); + assert!(matches!(conflict, Error::RepositoryConflict(message) + if message.contains("worker_singleton_release_required"))); + let worker = RuntimeWorkerRef::new("runtime-a", reserved.worker_id.to_string()); + assert!( + !reopened + .has_active_worker_create_reservation("workspace-a", &worker) + .unwrap() + ); + assert!( + reopened + .worker_resource_key("workspace-a", &worker) + .unwrap() + .is_none() + ); + assert_eq!( + reopened + .current_worker_singleton_owner("workspace-a", "subjektiv:restart-retry") + .unwrap(), + reserved.singleton + ); } #[tokio::test] @@ -21036,6 +21243,383 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx ); } + #[test] + fn subject_create_recovery_migration_is_atomic_and_matches_fresh_schema() { + let dir = tempfile::tempdir().unwrap(); + let database = dir.path().join("server.db"); + prepare_retained_schema(&database, 101); + let conn = Connection::open(&database).unwrap(); + configure_sqlite(&conn).unwrap(); + conn.execute_batch("CREATE TRIGGER interrupt_create_recovery BEFORE INSERT ON __yoi_schema_migrations WHEN NEW.version=102 BEGIN SELECT RAISE(ABORT,'injected recovery migration interruption'); END;").unwrap(); + assert!(migrate_subject_create_recovery_v101_to_v102(&conn).is_err()); + assert_eq!(current_schema_version(&conn).unwrap(), 101); + assert!(!table_exists(&conn, "subject_create_recoveries").unwrap()); + conn.execute_batch("DROP TRIGGER interrupt_create_recovery") + .unwrap(); + migrate_subject_create_recovery_v101_to_v102(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 102); + assert!(migrate_subject_create_recovery_v101_to_v102(&conn).is_err()); + let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let schema = |c: &Connection| -> Result { + Ok(c.query_row( + "SELECT sql FROM sqlite_schema WHERE name='subject_create_recoveries'", + [], + |r| r.get::<_, String>(0), + )? + .split_whitespace() + .collect::>() + .join(" ")) + }; + assert_eq!(schema(&conn).unwrap(), fresh.with_conn(schema).unwrap()); + assert_eq!( + conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r + .get::<_, i64>( + 0 + )) + .unwrap(), + 0 + ); + drop(conn); + let reopened = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + reopened + .with_conn(|c| verify_schema_history(c, 102)) + .unwrap(); + } + + #[tokio::test] + async fn subject_create_recovery_is_durable_exact_and_cannot_release_a_new_generation() { + for baseline in [None, Some(52), Some(101)] { + let dir = tempfile::tempdir().unwrap(); + let database = dir.path().join("server.db"); + if let Some(baseline) = baseline { + prepare_retained_schema(&database, baseline); + } + let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "workspace-a".into(), + owner_account_id: "owner-account".into(), + display_name: "Workspace A".into(), + state: "active".into(), + created_at: "2026-10-01T00:00:00Z".into(), + updated_at: "2026-10-01T00:00:00Z".into(), + }) + .await + .unwrap(); + let memory = store.get_workspace_memory_settings("workspace-a").unwrap(); + let key = "subjektiv:failed-create"; + let first = store + .reserve_worker_create( + "workspace-a", + "runtime-a", + "manual:first", + "sha256:first", + Some(key), + &memory, + ) + .unwrap(); + let lease = first.singleton.unwrap(); + let fingerprint = first.create_fingerprint; + assert_eq!( + store + .subject_create_recovery_candidate("workspace-a", &lease) + .unwrap(), + None + ); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "recover-first") + .is_err() + ); + store + .fail_worker_create_reservation( + "workspace-a", + "runtime-a", + first.worker_id, + &fingerprint, + ) + .unwrap(); + assert_eq!( + store + .subject_create_recovery_candidate("workspace-a", &lease) + .unwrap(), + Some(fingerprint.clone()) + ); + assert!( + store + .prepare_subject_create_recovery("workspace-b", &lease, "recover-first") + .is_err() + ); + for wrong in [ + WorkerSingletonLease { + generation: lease.generation + 1, + ..lease.clone() + }, + WorkerSingletonLease { + key: "subjektiv:other".into(), + ..lease.clone() + }, + WorkerSingletonLease { + worker: RuntimeWorkerRef::new("runtime-b", &lease.worker.worker_id), + ..lease.clone() + }, + WorkerSingletonLease { + worker: RuntimeWorkerRef::new("runtime-a", "other-worker"), + ..lease.clone() + }, + ] { + assert_eq!( + store + .subject_create_recovery_candidate("workspace-a", &wrong) + .unwrap(), + None + ); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &wrong, "recover-first") + .is_err() + ); + } + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint + ) + .is_err() + ); + assert_eq!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "recover-first") + .unwrap(), + fingerprint + ); + assert!( + store + .subject_release_pending("workspace-a", &lease.worker) + .unwrap() + ); + assert_eq!( + store + .current_worker_singleton_owner("workspace-a", key) + .unwrap(), + Some(lease.clone()) + ); + assert!( + store + .reserve_worker_create( + "workspace-a", + "runtime-a", + "manual:retry", + "sha256:first", + Some(key), + &memory + ) + .is_err() + ); + drop(store); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + assert_eq!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "recover-first") + .unwrap(), + fingerprint + ); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "different-operation") + .is_err() + ); + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + "different-fingerprint" + ) + .is_err() + ); + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "different-operation", + &fingerprint + ) + .is_err() + ); + assert!( + store + .complete_subject_create_recovery( + "workspace-b", + &lease, + "recover-first", + &fingerprint + ) + .is_err() + ); + // A registry row appearing after prepare must block both replay and completion. + store.with_conn(|c| { + c.execute("INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at) VALUES('workspace-a',?1,?2,'body','normal','now','now')", params![lease.worker.runtime_id, lease.worker.worker_id])?; + Ok(()) + }).unwrap(); + assert_eq!( + store + .subject_create_recovery_candidate("workspace-a", &lease) + .unwrap(), + None + ); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "recover-first") + .is_err() + ); + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint + ) + .is_err() + ); + store.with_conn(|c| { + c.execute("DELETE FROM worker_registry WHERE workspace_id='workspace-a' AND runtime_id=?1 AND worker_id=?2", params![lease.worker.runtime_id, lease.worker.worker_id])?; + // Exercise a current-generation change while the operation is incomplete. + c.execute("UPDATE worker_singleton_owners SET generation=generation+1 WHERE workspace_id='workspace-a' AND singleton_key=?1", [key])?; + Ok(()) + }).unwrap(); + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint + ) + .is_err() + ); + assert!( + store + .subject_release_pending("workspace-a", &lease.worker) + .unwrap() + ); + store.with_conn(|c| { + c.execute("UPDATE worker_singleton_owners SET generation=?2 WHERE workspace_id='workspace-a' AND singleton_key=?1", params![key, lease.generation as i64])?; + Ok(()) + }).unwrap(); + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint, + ) + .unwrap(); + drop(store); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint, + ) + .unwrap(); + assert!( + !store + .subject_release_pending("workspace-a", &lease.worker) + .unwrap() + ); + assert_eq!( + store + .current_worker_singleton_owner("workspace-a", key) + .unwrap(), + None + ); + let next = store + .reserve_worker_create( + "workspace-a", + "runtime-b", + "manual:next", + "sha256:next", + Some(key), + &memory, + ) + .unwrap(); + let next_lease = next.singleton.unwrap(); + assert_eq!(next_lease.generation, lease.generation + 1); + store + .complete_subject_create_recovery( + "workspace-a", + &lease, + "recover-first", + &fingerprint, + ) + .unwrap(); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &lease, "recover-first") + .is_err() + ); + assert!( + store + .complete_subject_create_recovery( + "workspace-a", + &next_lease, + "recover-first", + &fingerprint + ) + .is_err() + ); + assert_eq!( + store + .current_worker_singleton_owner("workspace-a", key) + .unwrap(), + Some(next_lease.clone()) + ); + // The old completed proof cannot release the next removed generation. + store + .fail_worker_create_reservation( + "workspace-a", + "runtime-b", + next.worker_id, + &next.create_fingerprint, + ) + .unwrap(); + assert_eq!( + store + .current_worker_singleton_owner("workspace-a", key) + .unwrap(), + Some(next_lease.clone()) + ); + assert!( + store + .prepare_subject_create_recovery("workspace-a", &next_lease, "recover-first") + .is_err() + ); + assert_eq!( + store + .prepare_subject_create_recovery("workspace-a", &next_lease, "recover-next") + .unwrap(), + next.create_fingerprint + ); + store.with_conn(|c| { + assert_eq!(c.query_row("SELECT count(*) FROM subject_worker_releases", [], |r| r.get::<_, i64>(0))?, 0); + assert_eq!(c.query_row("SELECT state FROM worker_create_reservations WHERE workspace_id='workspace-a' AND worker_id=?1", [&lease.worker.worker_id], |r| r.get::<_, String>(0))?, "removed"); + assert!(c.execute("UPDATE subject_create_recoveries SET completed=2", []).is_err()); + assert!(c.execute("UPDATE subject_create_recoveries SET singleton_generation=0", []).is_err()); + assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0))?, 0); + Ok(()) + }).unwrap(); + } + } + #[test] fn subject_release_migration_rolls_back_schema_and_marker_then_reopens() { let dir = tempfile::tempdir().unwrap(); @@ -21056,7 +21640,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx drop(conn); let reopened = SqliteWorkspaceStore::open_for_test(database).unwrap(); reopened - .with_conn(|conn| verify_schema_history(conn, 101)) + .with_conn(|conn| verify_schema_history(conn, LATEST_SCHEMA_VERSION)) .unwrap(); } @@ -21636,7 +22220,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx (98, "Assignment-independent implementation reports".into()), (99, "Explicit hosting administrators and self-registration policy".into()), (100, "Provisional users and scoped passkey activation".into()), - (101, "Durable explicit Subject body release".into())]); + (101, "Durable explicit Subject body release".into()), + (102, "Durable failed Subject creation recovery".into())]); for (table, rows) in unchanged.iter().zip(before.iter()) { assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}"); } diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index a91209f2b..802e4e52a 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -95,6 +95,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "workdir_removal_operations", "worker_control_grants", "subject_worker_releases", + "subject_create_recoveries", "worker_create_reservations", "worker_diagnostics_archives", "worker_mutation_source_proof_jtis",