Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: recover unpublished Subject creation and preserve auth rejections

Browse Source
This commit is contained in:
Hare
2026-10-19 18:56:24 +09:00
parent c961f244ab
commit 8e40fae6f4
20 changed files with 2243 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/execution.rs
+16
View File
@@ -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,
crates/worker-runtime/src/fs_store.rs
+291 -35
View File
@@ -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<WorkerId, crate::runtime::SubjectCreateRecoveryFence>,
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<WorkerAggregateRecord, RuntimeError> {
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<WorkerId, crate::runtime::SubjectCreateRecoveryFence>,
#[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<WorkerId, PersistedWorkerRecord>,
) -> 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::<serde_json::Value>(&runtime_path, "read migrated Runtime").unwrap(),
runtime
);
assert_eq!(
read_json::<serde_json::Value>(&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]
crates/worker-runtime/src/http_server.rs
+24
View File
@@ -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"
}
crates/worker-runtime/src/http_server/runtime_management_api.rs
+12
View File
@@ -406,6 +406,18 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
}
}
async fn recover_subject_create(
&self,
request: runtime_api::SubjectCreateRecoveryRequest,
) -> Result<runtime_api::SubjectCreateRecoveryResponse, runtime_api::RuntimeApiError> {
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,
crates/worker-runtime/src/http_server/tests/session_material_tests.rs
+48
View File
@@ -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();
crates/worker-runtime/src/runtime.rs
+333 -3
View File
@@ -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<runtime_api::SubjectCreateRecoveryResponse>,
}
impl Runtime {
pub fn recover_subject_create_scoped(
&self,
scope: &RuntimeWorkspaceScope,
request: runtime_api::SubjectCreateRecoveryRequest,
) -> Result<runtime_api::SubjectCreateRecoveryResponse, RuntimeError> {
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<WorkerId, WorkerRecord>,
subject_create_recovery_fences: BTreeMap<WorkerId, SubjectCreateRecoveryFence>,
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<Option<WorkerExecutionSpawnResult>>,
job_bindings: Mutex<Vec<Option<crate::catalog::BackendJobExecutionBinding>>>,
dispatch_result: Mutex<Option<WorkerExecutionResult>>,
@@ -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!(
crates/worker-runtime/src/worker_backend.rs
+98
View File
@@ -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());
}