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

feat: replace Subject retirement with durable explicit deletion

Browse Source
This commit is contained in:
Hare
2026-10-28 11:18:43 +09:00
parent 618880d38d
commit a06f6bfb3d
71 changed files with 7448 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/execution.rs
+31
View File
@@ -727,6 +727,23 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
Err("execution backend cannot confirm Subject create execution absence".into())
}
/// Read-only exact execution identity check before Runtime persists exclusion.
fn preflight_subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
Err("execution backend cannot validate Subject Worker identity".into())
}
/// Runtime has durably fenced admission. Confirm shutdown and discard all
/// execution candidates under the backend Worker lock; uncertainty must fail.
fn subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
Err("execution backend cannot confirm Subject Worker shutdown".into())
}
/// Must reserve the actual Controller, not infer Idle from a catalog snapshot.
fn subject_release(
&self,
@@ -1069,6 +1086,20 @@ impl WorkerExecutionBackendRef {
self.backend.recover_subject_create(request)
}
pub(crate) fn preflight_subject_delete_worker(
&self,
request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
self.backend.preflight_subject_delete_worker(request)
}
pub(crate) fn subject_delete_worker(
&self,
request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
self.backend.subject_delete_worker(request)
}
pub(crate) fn subject_release(
&self,
request: &runtime_api::SubjectReleaseRequest,
crates/worker-runtime/src/fs_store.rs
+69 -7
View File
@@ -43,7 +43,8 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
#[path = "test_support.rs"]
pub mod test_support;
pub(crate) const SCHEMA_VERSION: u32 = 14;
pub(crate) const SCHEMA_VERSION: u32 = 15;
const RESTORE_REPAIR_SCHEMA_VERSION: u32 = 14;
const RESTORE_DECISION_SCHEMA_VERSION: u32 = 13;
const RESTORE_ADMISSION_SCHEMA_VERSION: u32 = 12;
// Frozen source schemas: do not move their meaning when adding a migration.
@@ -560,6 +561,22 @@ impl FsRuntimeStore {
self.root.join(RUNTIME_FILE)
}
pub(crate) fn worker_originals_absent(
&self,
worker_id: WorkerId,
) -> Result<bool, RuntimeError> {
let path = self.worker_dir(&worker_id);
match fs::symlink_metadata(&path) {
Ok(_) => Ok(false),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(true),
Err(source) => Err(RuntimeError::StoreIo {
operation: "confirm Subject Worker original absence",
path,
source,
}),
}
}
fn worker_dir(&self, worker_id: &WorkerId) -> PathBuf {
self.root.join(WORKERS_DIR).join(worker_id.to_string())
}
@@ -659,6 +676,8 @@ pub(crate) struct OpenedFsRuntimeStore {
#[derive(Clone, Debug)]
pub(crate) struct PersistedRuntimeState {
pub(crate) subject_delete_worker_fences:
BTreeMap<WorkerId, crate::runtime::SubjectDeleteWorkerFence>,
pub(crate) subject_create_recovery_fences:
BTreeMap<WorkerId, crate::runtime::SubjectCreateRecoveryFence>,
pub(crate) subject_releases: Vec<(
@@ -876,11 +895,12 @@ fn plan_runtime_store_migration(
| SOURCE_SCHEMA_VERSION
| RESTORE_ADMISSION_SCHEMA_VERSION
| RESTORE_DECISION_SCHEMA_VERSION
| RESTORE_REPAIR_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}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_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}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_SCHEMA_VERSION}, {RESTORE_REPAIR_SCHEMA_VERSION}, or {SCHEMA_VERSION}"
),
));
}
@@ -1216,6 +1236,7 @@ fn migrate_worker_record(
| SOURCE_SCHEMA_VERSION
| RESTORE_ADMISSION_SCHEMA_VERSION
| RESTORE_DECISION_SCHEMA_VERSION
| RESTORE_REPAIR_SCHEMA_VERSION
) {
// Admission identity survives migration. Older pending records did not
// distinguish a committed decision from preparation or failed cleanup.
@@ -1229,10 +1250,12 @@ fn migrate_worker_record(
"Worker schema does not match Runtime migration source".into(),
));
}
if source_schema_version != RESTORE_DECISION_SCHEMA_VERSION {
if source_schema_version < RESTORE_DECISION_SCHEMA_VERSION {
migrate_restore_commit_fact(&mut identity_document, identity_path)?;
}
migrate_restore_repair_fact(&mut identity_document, identity_path)?;
if source_schema_version < RESTORE_REPAIR_SCHEMA_VERSION {
migrate_restore_repair_fact(&mut identity_document, identity_path)?;
}
identity_document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
let aggregate: WorkerAggregateRecord = serde_json::from_value(identity_document)
.map_err(|error| runtime_store_corrupt(identity_path, error.to_string()))?;
@@ -1691,6 +1714,9 @@ fn migrate_runtime_document(
"schema_version".to_string(),
serde_json::Value::from(SCHEMA_VERSION),
);
object
.entry("subject_delete_worker_fences")
.or_insert_with(|| serde_json::json!({}));
// Intermediate schema-11 writers may already have saved irreversible fences.
// Default only a missing field; malformed or existing fences must not be erased.
object
@@ -1698,7 +1724,10 @@ fn migrate_runtime_document(
.or_insert_with(|| serde_json::json!({}));
if !matches!(
source_schema_version,
SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION
SOURCE_SCHEMA_VERSION
| RESTORE_ADMISSION_SCHEMA_VERSION
| RESTORE_DECISION_SCHEMA_VERSION
| RESTORE_REPAIR_SCHEMA_VERSION
) {
object.remove("workers");
// This legacy cache is not Runtime lifecycle or restore authority. Bundles
@@ -2249,6 +2278,7 @@ fn migrate_runtime_store_in_place(
SOURCE_SCHEMA_VERSION
| RESTORE_ADMISSION_SCHEMA_VERSION
| RESTORE_DECISION_SCHEMA_VERSION
| RESTORE_REPAIR_SCHEMA_VERSION
)
{
atomic_write_json(
@@ -2288,6 +2318,7 @@ fn migrate_runtime_store_in_place(
#[derive(Clone, Debug, Serialize, Deserialize)]
struct RuntimeSnapshot {
subject_delete_worker_fences: BTreeMap<WorkerId, crate::runtime::SubjectDeleteWorkerFence>,
#[serde(default)]
subject_create_recovery_fences: BTreeMap<WorkerId, crate::runtime::SubjectCreateRecoveryFence>,
#[serde(default)]
@@ -2326,6 +2357,7 @@ fn record_worker_load_diagnostic(
impl RuntimeSnapshot {
fn from_persisted(state: &PersistedRuntimeState) -> Self {
Self {
subject_delete_worker_fences: state.subject_delete_worker_fences.clone(),
subject_create_recovery_fences: state.subject_create_recovery_fences.clone(),
subject_releases: state.subject_releases.clone(),
schema_version: SCHEMA_VERSION,
@@ -2350,6 +2382,33 @@ impl RuntimeSnapshot {
),
});
}
for (worker_id, fence) in &self.subject_delete_worker_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 delete Worker fence",
path: path.to_path_buf(),
message: "invalid deletion identity, receipt or Workspace owner".into(),
});
}
}
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| {
@@ -2392,6 +2451,7 @@ impl RuntimeSnapshot {
workers: BTreeMap<WorkerId, PersistedWorkerRecord>,
) -> PersistedRuntimeState {
PersistedRuntimeState {
subject_delete_worker_fences: self.subject_delete_worker_fences,
subject_create_recovery_fences: self.subject_create_recovery_fences,
subject_releases: self.subject_releases,
display_name: self.display_name,
@@ -3218,8 +3278,7 @@ mod tests {
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());
assert!(serde_json::from_value::<RuntimeSnapshot>(runtime.clone()).is_err());
let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap();
let state = opened.state.as_ref().unwrap();
@@ -3231,6 +3290,7 @@ mod tests {
expected_worker
);
runtime["schema_version"] = serde_json::json!(SCHEMA_VERSION);
runtime["subject_delete_worker_fences"] = serde_json::json!({});
if fence_state == "absent" {
runtime["subject_create_recovery_fences"] = serde_json::json!({});
}
@@ -4072,6 +4132,8 @@ mod tests {
);
}
include!("subject_delete_store_tests.rs");
fn canonical_worker_store() -> (
tempfile::TempDir,
FsRuntimeStore,
crates/worker-runtime/src/http_server.rs
+17 -1
View File
@@ -2102,7 +2102,11 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
{
return Some("workers:input");
}
if path == "/v1/subject-create-recovery" && *method == Method::POST {
if matches!(
path,
"/v1/subject-create-recovery" | "/v1/subject-delete-worker"
) && *method == Method::POST
{
return Some("workers:delete");
}
if path.ends_with("/stop")
@@ -3332,6 +3336,18 @@ mod tests {
struct AcceptingBackend;
impl WorkerExecutionBackend for AcceptingBackend {
fn preflight_subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
Ok(())
}
fn subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
Ok(())
}
fn recover_subject_create(
&self,
_request: &runtime_api::SubjectCreateRecoveryRequest,
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 subject_delete_worker(
&self,
request: runtime_api::SubjectDeleteWorkerRequest,
) -> Result<runtime_api::SubjectDeleteWorkerResponse, runtime_api::RuntimeApiError> {
let scope = material_scope(&self.state, &request.workspace_id)?;
self.state
.runtime
.subject_delete_worker_scoped(&scope, request)
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)
}
async fn recover_subject_create(
&self,
request: runtime_api::SubjectCreateRecoveryRequest,
crates/worker-runtime/src/http_server/tests/session_material_tests.rs
+48
View File
@@ -1222,6 +1222,54 @@ async fn subject_create_recovery_signed_http_requires_delete_and_matching_worksp
}
}
#[tokio::test]
async fn subject_delete_worker_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-delete-worker";
let mut request = runtime_api::SubjectDeleteWorkerRequest {
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::SubjectDeleteWorkerResponse =
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/retention/runtime_v5_tests.rs
+1 -1
View File
@@ -97,7 +97,7 @@ fn runtime_fixture(root: &Path, schema: u32) -> (WorkerId, SessionId, SegmentId)
"schema_version": schema, "display_name": "retained Runtime",
"backend": "fs_store", "status": "running", "next_diagnostic_id": 1,
"config_bundles": {}, "workspace_owners": {"workspace-a": "owner-a"},
"diagnostics": [], "subject_releases": [],
"diagnostics": [], "subject_releases": [], "subject_delete_worker_fences": {},
"subject_create_recovery_fences": {fenced_worker.to_string(): {
"request": {"workspace_id": "workspace-a", "worker_id": fenced_worker,
"create_fingerprint": "retained-create", "operation_id": "retained-recovery"},
crates/worker-runtime/src/runtime.rs
+47 -2
View File
@@ -1,3 +1,5 @@
mod subject_delete;
pub(crate) use subject_delete::SubjectDeleteWorkerFence;
#[cfg(feature = "restore-checkpoints")]
mod restore_checkpoint;
mod restore_continuation;
@@ -5645,6 +5647,7 @@ struct RuntimeState {
// Ephemeral write barrier, not authority: only canonical reread + sync can
// release it. Prevents another mutation from persisting an older projection.
unreconciled_worker_writes: std::collections::BTreeSet<WorkerId>,
subject_delete_worker_fences: BTreeMap<WorkerId, SubjectDeleteWorkerFence>,
subject_create_recovery_fences: BTreeMap<WorkerId, SubjectCreateRecoveryFence>,
subject_release_operations: BTreeMap<
(WorkerId, String),
@@ -5686,6 +5689,7 @@ impl RuntimeState {
next_diagnostic_id: 1,
workers: BTreeMap::new(),
unreconciled_worker_writes: Default::default(),
subject_delete_worker_fences: BTreeMap::new(),
subject_create_recovery_fences: BTreeMap::new(),
subject_release_operations: BTreeMap::new(),
workdir_operations: BTreeMap::new(),
@@ -5722,6 +5726,7 @@ impl RuntimeState {
next_diagnostic_id: 1,
workers: BTreeMap::new(),
unreconciled_worker_writes: Default::default(),
subject_delete_worker_fences: BTreeMap::new(),
subject_create_recovery_fences: BTreeMap::new(),
subject_release_operations: BTreeMap::new(),
workdir_operations: BTreeMap::new(),
@@ -5773,6 +5778,7 @@ impl RuntimeState {
next_diagnostic_id,
workers,
unreconciled_worker_writes: Default::default(),
subject_delete_worker_fences: persisted.subject_delete_worker_fences,
subject_create_recovery_fences: persisted.subject_create_recovery_fences,
subject_release_operations: persisted
.subject_releases
@@ -5804,6 +5810,7 @@ impl RuntimeState {
#[cfg(feature = "fs-store")]
fn persisted_state(&self) -> PersistedRuntimeState {
PersistedRuntimeState {
subject_delete_worker_fences: self.subject_delete_worker_fences.clone(),
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(),
@@ -6027,7 +6034,9 @@ impl RuntimeState {
}
fn ensure_subject_create_not_fenced(&self, worker_id: WorkerId) -> Result<(), RuntimeError> {
if self.subject_create_recovery_fences.contains_key(&worker_id) {
if self.subject_create_recovery_fences.contains_key(&worker_id)
|| self.subject_delete_worker_fences.contains_key(&worker_id)
{
return Err(RuntimeError::InvalidRequest(
"Subject create identity is irreversibly fenced".into(),
));
@@ -6040,6 +6049,14 @@ impl RuntimeState {
worker: &WorkerRef,
allow_disconnected: bool,
) -> Result<(), RuntimeError> {
if allow_disconnected
&& self
.subject_delete_worker_fences
.get(&worker.worker_id)
.is_some_and(|fence| fence.response.is_some())
{
return Ok(());
}
self.ensure_subject_create_not_fenced(worker.worker_id)?;
use runtime_api::SubjectReleaseState as State;
if self
@@ -6110,6 +6127,13 @@ impl RuntimeState {
}
fn forget_workspace_owner_if_unused(&mut self, workspace_id: &str) -> bool {
if self
.subject_delete_worker_fences
.values()
.any(|fence| fence.request.workspace_id == workspace_id)
{
return false;
}
if self
.subject_create_recovery_fences
.values()
@@ -8767,6 +8791,8 @@ mod tests {
#[derive(Default)]
struct TestExecutionBackend {
deletion_confirmed: AtomicBool,
deletion_checks: std::sync::atomic::AtomicUsize,
recovery_absent: AtomicBool,
recovery_checks: std::sync::atomic::AtomicUsize,
spawn_result: Mutex<Option<WorkerExecutionSpawnResult>>,
@@ -8848,6 +8874,23 @@ mod tests {
}
impl WorkerExecutionBackend for TestExecutionBackend {
fn preflight_subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
Ok(())
}
fn subject_delete_worker(
&self,
_request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
self.deletion_checks.fetch_add(1, Ordering::SeqCst);
if self.deletion_confirmed.load(Ordering::SeqCst) {
Ok(())
} else {
Err("shutdown is unconfirmed".into())
}
}
fn recover_subject_create(
&self,
_request: &runtime_api::SubjectCreateRecoveryRequest,
@@ -14074,7 +14117,7 @@ mod tests {
})
.unwrap_err();
assert!(error.to_string().contains(
"unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, or 14"
"unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, 14, or 15"
));
let _ = std::fs::remove_dir_all(root);
@@ -14259,6 +14302,8 @@ mod tests {
let _ = std::fs::remove_dir_all(root);
}
include!("runtime/subject_delete_tests.rs");
#[cfg(feature = "fs-store")]
#[test]
fn subject_create_recovery_is_durable_exact_and_rechecks_absence() {
crates/worker-runtime/src/runtime/subject_delete.rs
+168
View File
@@ -0,0 +1,168 @@
use super::*;
/// Permanent exclusion survives WorkerRemoval. A receipt never substitutes for a
/// fresh execution reconciliation on exact retry, including after Runtime reopen.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct SubjectDeleteWorkerFence {
pub(crate) request: runtime_api::SubjectDeleteWorkerRequest,
pub(crate) response: Option<runtime_api::SubjectDeleteWorkerResponse>,
}
impl Runtime {
pub fn subject_delete_worker_scoped(
&self,
scope: &RuntimeWorkspaceScope,
request: runtime_api::SubjectDeleteWorkerRequest,
) -> Result<runtime_api::SubjectDeleteWorkerResponse, 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 delete Worker identity".into(),
));
}
let lease = self.worker_operation_lock(request.worker_id)?;
let _operation = lease.lock().map_err(|_| RuntimeError::StatePoisoned)?;
let worker_ref = WorkerRef::new(request.worker_id);
let backend = {
let mut state = self.lock()?;
state.ensure_running()?;
if state.backend == RuntimeBackendKind::Memory {
return Err(RuntimeError::InvalidRequest(
"Subject delete requires durable storage".into(),
));
}
state.ensure_workspace_owner(scope, false)?;
if state.workers.contains_key(&request.worker_id) {
state.refresh_uncertain_worker_write(&worker_ref)?;
let worker = state.worker(&worker_ref)?;
if worker.workspace_id.as_deref() != Some(request.workspace_id.as_str())
|| worker.request.as_ref().is_none_or(|create| {
create.create_fingerprint != request.create_fingerprint
})
{
return Err(RuntimeError::InvalidRequest(
"Subject delete Worker identity mismatch".into(),
));
}
}
if let Some(recovery) = state.subject_create_recovery_fences.get(&request.worker_id)
&& (recovery.request.workspace_id != request.workspace_id
|| recovery.request.create_fingerprint != request.create_fingerprint)
{
return Err(RuntimeError::InvalidRequest(
"Subject recovery identity mismatch".into(),
));
}
if state.subject_delete_worker_fences.values().any(|fence| {
fence.request.workspace_id == request.workspace_id
&& fence.request.operation_id == request.operation_id
&& fence.request.worker_id != request.worker_id
}) {
return Err(RuntimeError::InvalidRequest(
"Subject delete operation reused for another Worker".into(),
));
}
state.execution_backend.clone()
};
// Unpublished execution identity is checked without holding the catalog
// mutex. The Worker operation lock excludes Runtime create/restore.
if let Some(backend) = backend.as_ref() {
backend
.preflight_subject_delete_worker(&request)
.map_err(RuntimeError::InvalidRequest)?;
}
{
let mut state = self.lock()?;
state.ensure_workspace_owner(scope, false)?;
if state.subject_delete_worker_fences.values().any(|fence| {
fence.request.workspace_id == request.workspace_id
&& fence.request.operation_id == request.operation_id
&& fence.request.worker_id != request.worker_id
}) {
return Err(RuntimeError::InvalidRequest(
"Subject delete operation reused for another Worker".into(),
));
}
if let Some(saved) = state.subject_delete_worker_fences.get(&request.worker_id) {
if saved.request != request {
return Err(RuntimeError::InvalidRequest(
"different Subject delete Worker request retained".into(),
));
}
} else {
state.ensure_workspace_owner(scope, true)?;
state.subject_delete_worker_fences.insert(
request.worker_id,
SubjectDeleteWorkerFence {
request: request.clone(),
response: None,
},
);
}
// Keep exclusion on persistence uncertainty; never start execution
// cleanup until this durable write has been confirmed.
state.persist_runtime_snapshot()?;
}
let backend = backend.ok_or_else(|| {
RuntimeError::InvalidRequest(
"Subject Worker shutdown cannot be confirmed without execution backend".into(),
)
})?;
backend
.subject_delete_worker(&request)
.map_err(RuntimeError::InvalidRequest)?;
let mut state = self.lock()?;
if let Some(mut worker) = state.workers.get(&request.worker_id).cloned() {
let restore_receipt =
RuntimeWorkerRestoreResult::superseded_by_stop(worker.pending_restore);
worker.pending_stop = None;
worker.pending_restore = None;
if let Some(context) = worker.restore_candidate_context.take() {
context.discard_candidate();
}
worker.restore_guard.rotate();
worker.restore_guard.finish(restore_receipt);
worker.status = WorkerStatus::Stopped;
worker.worker_state = None;
worker.restore_intent = WorkerRestoreIntent::Explicit;
worker.internal_workers.clear();
worker.execution_bound = false;
state.persist_worker_record(&worker)?;
state.workers.insert(request.worker_id, worker);
state.publish_worker_upsert(request.worker_id)?;
}
// This is storage-owner evidence under the same lock that permanently
// excludes create/restore, not an inference from catalog projection.
#[cfg(feature = "fs-store")]
let originals_absent = !state.workers.contains_key(&request.worker_id)
&& state
.fs_store()
.ok_or_else(|| {
RuntimeError::InvalidRequest(
"Subject original absence requires fs storage".into(),
)
})?
.worker_originals_absent(request.worker_id)?;
#[cfg(not(feature = "fs-store"))]
let originals_absent = false;
let response = runtime_api::SubjectDeleteWorkerResponse {
workspace_id: request.workspace_id.clone(),
worker_id: request.worker_id,
create_fingerprint: request.create_fingerprint.clone(),
operation_id: request.operation_id.clone(),
originals_absent,
};
state
.subject_delete_worker_fences
.get_mut(&request.worker_id)
.expect("delete fence retained under Worker operation lock")
.response = Some(response.clone());
state.persist_runtime_snapshot()?;
Ok(response)
}
}
crates/worker-runtime/src/runtime/subject_delete_tests.rs
+274
View File
@@ -0,0 +1,274 @@
// Runtime component tests use real filesystem persistence and replace only the
// external execution boundary. Actual Controller shutdown is tested by the adapter.
#[cfg(feature = "fs-store")]
#[test]
fn subject_delete_reserved_id_survives_unknown_reopen_and_rejects_identity_changes() {
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 create = scoped_task_request("delayed create", "workspace-a");
let request = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scoped.workspace_id.clone(),
worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(),
operation_id: "subject-delete-1".into(),
};
let backend = Arc::new(TestExecutionBackend::default());
let runtime =
Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
assert!(
runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.is_err()
);
assert!(
runtime.lock().unwrap().subject_delete_worker_fences[&create.worker_id]
.response
.is_none()
);
assert!(
runtime
.create_worker_scoped(&scoped, create.clone())
.is_err()
);
assert_eq!(backend.deletion_checks.load(Ordering::SeqCst), 1);
drop(runtime);
let runtime =
Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
assert!(
runtime
.create_worker_scoped(&scoped, create.clone())
.is_err()
);
assert!(
runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.is_err()
);
backend.deletion_confirmed.store(true, Ordering::SeqCst);
let receipt = runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.unwrap();
let mut expected = serde_json::to_value(&request).unwrap();
expected["originals_absent"] = true.into();
assert_eq!(serde_json::to_value(&receipt).unwrap(), expected);
for index in 0..4 {
let mut changed = request.clone();
match index {
0 => changed.worker_id = WorkerId::now_v7(),
1 => changed.operation_id.push_str("-changed"),
2 => changed.create_fingerprint.push_str("-changed"),
_ => changed.workspace_id.push_str("-changed"),
}
assert!(
runtime
.subject_delete_worker_scoped(&scoped, changed)
.is_err()
);
}
assert!(
runtime
.subject_delete_worker_scoped(&scope("workspace-a", "server-b"), request.clone())
.is_err()
);
assert!(
runtime
.subject_delete_worker_scoped(&scope("workspace-b", "server-a"), request.clone())
.is_err()
);
assert!(
!runtime
.lock()
.unwrap()
.forget_workspace_owner_if_unused("workspace-a")
);
drop(runtime);
// Even a saved receipt must not manufacture fresh shutdown proof on reopen.
let backend = Arc::new(TestExecutionBackend::default());
let runtime = Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap();
assert_eq!(*backend.restore_count.lock().unwrap(), 0);
assert!(
runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.is_err()
);
backend.deletion_confirmed.store(true, Ordering::SeqCst);
assert_eq!(
runtime
.subject_delete_worker_scoped(&scoped, request)
.unwrap(),
receipt
);
assert!(runtime.create_worker_scoped(&scoped, create).is_err());
}
#[cfg(feature = "fs-store")]
#[test]
fn subject_delete_existing_stopped_or_active_worker_preserves_record_and_allows_canonical_removal()
{
for stopped in [false, true] {
let root = tempfile::tempdir().unwrap();
let backend = Arc::new(TestExecutionBackend::default());
backend.deletion_confirmed.store(true, Ordering::SeqCst);
let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path())
.with_runtime_id("test-runtime");
let runtime =
Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let scoped = scope("workspace-a", "server-a");
let create = scoped_task_request("subject body", "workspace-a");
let worker = runtime
.create_worker_scoped(&scoped, create.clone())
.unwrap();
if stopped {
runtime.stop_worker(&worker.worker_ref, None).unwrap();
}
let request = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scoped.workspace_id.clone(),
worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(),
operation_id: "delete-body".into(),
};
let mut mismatch = request.clone();
mismatch.create_fingerprint.push_str("-wrong");
assert!(
runtime
.subject_delete_worker_scoped(&scoped, mismatch)
.is_err()
);
let mut mismatch = request.clone();
mismatch.workspace_id = "workspace-b".into();
assert!(
runtime
.subject_delete_worker_scoped(&scope("workspace-b", "server-b"), mismatch)
.is_err()
);
assert!(
!runtime
.lock()
.unwrap()
.subject_delete_worker_fences
.contains_key(&create.worker_id)
);
// A lost Worker settlement cannot be reported as completed deletion.
runtime
.test_runtime_store()
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
assert!(
runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.is_err()
);
let receipt = runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.unwrap();
assert!(!receipt.originals_absent, "a stopped catalog record still has original material");
assert_eq!(
runtime.worker_detail(&worker.worker_ref).unwrap().status,
WorkerStatus::Stopped
);
assert!(
runtime
.restore_worker(
&worker.worker_ref,
runtime.test_restore_request(&worker.worker_ref)
)
.is_err()
);
assert!(
runtime
.create_worker_scoped(&scoped, create.clone())
.is_err()
);
let record = runtime
.test_runtime_store()
.read_worker_record(&worker.worker_ref)
.unwrap();
assert_eq!(record.worker_id, worker.worker_ref.worker_id);
drop(runtime);
let runtime =
Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap();
assert!(
runtime
.restore_worker(
&worker.worker_ref,
runtime.test_restore_request(&worker.worker_ref)
)
.is_err()
);
assert_eq!(
runtime
.subject_delete_worker_scoped(&scoped, request)
.unwrap(),
receipt
);
runtime
.delete_worker_scoped(&scoped, &worker.worker_ref)
.unwrap();
assert!(runtime.create_worker_scoped(&scoped, create).is_err());
assert!(
!runtime
.lock()
.unwrap()
.forget_workspace_owner_if_unused("workspace-a")
);
}
}
#[cfg(feature = "fs-store")]
#[test]
fn subject_delete_unavailable_backend_never_issues_receipt_but_preserves_exclusion() {
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 create = scoped_task_request("reserved without backend", "workspace-a");
let request = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scoped.workspace_id.clone(),
worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(),
operation_id: "delete-offline".into(),
};
let runtime = Runtime::with_fs_store(options.clone()).unwrap();
assert!(
runtime
.subject_delete_worker_scoped(&scoped, request.clone())
.is_err()
);
assert!(
runtime.lock().unwrap().subject_delete_worker_fences[&create.worker_id]
.response
.is_none()
);
drop(runtime);
let backend = Arc::new(TestExecutionBackend::default());
backend.deletion_confirmed.store(true, Ordering::SeqCst);
let runtime = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap();
assert!(runtime.create_worker_scoped(&scoped, create).is_err());
runtime
.subject_delete_worker_scoped(&scoped, request)
.unwrap();
}
#[cfg(feature = "fs-store")]
#[test]
fn subject_delete_missing_catalog_does_not_claim_residual_originals_absent() {
let root = tempfile::tempdir().unwrap();
let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path()).with_runtime_id("test-runtime");
let backend = Arc::new(TestExecutionBackend::default());
backend.deletion_confirmed.store(true, Ordering::SeqCst);
let runtime = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap();
let scoped = scope("workspace-a", "server-a");
let create = scoped_task_request("unpublished original", "workspace-a");
let original = root.path().join("workers").join(create.worker_id.to_string()).join("session/original.jsonl");
std::fs::create_dir_all(original.parent().unwrap()).unwrap();
std::fs::write(&original, "preserve original text").unwrap();
let receipt = runtime.subject_delete_worker_scoped(&scoped, runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scoped.workspace_id.clone(), worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(), operation_id: "delete-unpublished".into(),
}).unwrap();
assert!(!receipt.originals_absent);
assert_eq!(std::fs::read_to_string(original).unwrap(), "preserve original text");
assert!(runtime.create_worker_scoped(&scoped, create).is_err());
}
crates/worker-runtime/src/subject_delete_backend_tests.rs
+285
View File
@@ -0,0 +1,285 @@
// Integration of Runtime's durable fence with the real adapter and Controller;
// MockFactory replaces only model/provider/profile construction for a hanging run.
#[test]
#[cfg(feature = "fs-store")]
#[serial_test::serial(worker_allocation)]
fn subject_delete_running_controller_confirms_shutdown_and_preserves_history() {
let root = tempfile::tempdir().unwrap();
let client = MockClient::sequential(vec![
MockResponse::Complete(simple_text_events()),
MockResponse::Hang(vec![]),
]);
let backend = Arc::new(
WorkerRuntimeExecutionBackend::new(MockFactory {
client: client.clone(),
runtime_base: root.path().join("control"),
cwd: root.path().to_path_buf(),
store_dir: root.path().join("sessions"),
worker_metadata_dir: root.path().join("metadata"),
observed_cwds: Arc::default(),
observed_workspace_clients: Arc::default(),
})
.unwrap(),
);
let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions::new(root.path().join("catalog"))
.with_runtime_id("delete-test"),
backend.clone(),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test");
let mut create = create_request("running Subject body");
create.workspace_api = Some(WorkspaceApiRef {
workspace_id: scope.workspace_id.clone(),
base_url: "https://workspace.invalid".into(),
});
create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot {
workspace_id: scope.workspace_id.clone(),
language: "English".into(),
});
let worker = runtime
.create_worker_scoped(&scope, create.clone())
.unwrap();
runtime
.send_input(&worker.worker_ref, WorkerInput::user("settled history"))
.unwrap();
let execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone();
backend
.run_on_adapter_runtime({
let handle = execution.handle.clone();
let client = client.clone();
async move {
tokio::time::timeout(Duration::from_secs(5), async {
while client.call_count.load(Ordering::SeqCst) < 1
|| handle.shared_state.catalog_status() != WorkerStatus::Idle
{
tokio::task::yield_now().await;
}
})
.await
.map_err(|error| error.to_string())?;
Ok(())
}
})
.unwrap();
let before = persisted_files(&root.path().join("sessions"));
assert!(!before.is_empty());
runtime
.send_input(&worker.worker_ref, WorkerInput::user("run until stopped"))
.unwrap();
backend
.run_on_adapter_runtime({
let handle = execution.handle.clone();
let client = client.clone();
async move {
tokio::time::timeout(Duration::from_secs(5), async {
while client.call_count.load(Ordering::SeqCst) < 2 {
tokio::task::yield_now().await;
}
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Running);
})
.await
.map_err(|error| error.to_string())?;
Ok(())
}
})
.unwrap();
let request = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scope.workspace_id.clone(),
worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(),
operation_id: "delete-running".into(),
};
runtime
.subject_delete_worker_scoped(&scope, request.clone())
.unwrap();
assert!(execution.handle.protocol_is_closed());
assert!(execution.tasks.tasks.lock().unwrap().is_empty());
assert!(backend.workers.lock().unwrap().is_empty());
assert!(backend.pending_restores.lock().unwrap().is_empty());
assert_eq!(
runtime.worker_detail(&worker.worker_ref).unwrap().status,
crate::catalog::WorkerStatus::Stopped
);
// Preserve the settled prefix. Ordinary Controller empty-turn rollback remains unchanged.
let after = persisted_files(&root.path().join("sessions"));
for (path, bytes) in before {
let entries = |bytes: &[u8]| {
bytes
.split(|b| *b == b'\n')
.filter(|line| !line.is_empty())
.map(|line| serde_json::from_slice::<serde_json::Value>(line).unwrap())
.collect::<Vec<_>>()
};
assert!(
entries(&after[&path]).starts_with(&entries(&bytes)),
"{path:?} BEFORE={:?} AFTER={:?}",
entries(&bytes),
entries(&after[&path])
);
}
assert!(runtime.create_worker_scoped(&scope, create).is_err());
assert!(
runtime
.restore_worker(
&worker.worker_ref,
runtime.test_restore_request(&worker.worker_ref)
)
.is_err()
);
runtime
.subject_delete_worker_scoped(&scope, request)
.unwrap();
assert_eq!(client.call_count.load(Ordering::SeqCst), 2);
}
#[test]
#[serial_test::serial(worker_allocation)]
#[cfg(feature = "http-server")]
fn subject_delete_unregistered_profile_controller_retries_unknown_shutdown_without_erasing_session()
{
let root = tempfile::tempdir().unwrap();
let (backend, request) = profile_stop_fixture(root.path());
let execution = backend.workers.lock().unwrap()[&request.worker_ref].clone();
let deletion = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: request
.workspace_scope
.as_ref()
.unwrap()
.workspace_id
.clone(),
worker_id: request.worker_ref.worker_id,
create_fingerprint: request.request.create_fingerprint.clone(),
operation_id: "delete-unregistered".into(),
};
let mut bad = deletion.clone();
bad.create_fingerprint.push_str("-wrong");
assert!(backend.preflight_subject_delete_worker(&bad).is_err());
assert!(backend.subject_delete_worker(&bad).is_err());
let mut bad = deletion.clone();
bad.workspace_id.push_str("-wrong");
assert!(backend.subject_delete_worker(&bad).is_err());
assert!(
backend.admissions.lock().unwrap()[&request.worker_ref]
.deletion
.is_none()
);
let preparing = execution.clone();
backend
.run_on_adapter_runtime(async move {
prepare_rejected_shutdown(&preparing, false, None).await;
Ok(())
})
.unwrap();
let session_root = root
.path()
.join("runtime/workers")
.join(request.worker_ref.worker_id.to_string())
.join("session");
let before = persisted_files(&session_root);
assert!(!before.is_empty());
assert!(
backend
.subject_delete_worker(&deletion)
.unwrap_err()
.contains("Conflict")
);
assert!(
backend
.workers
.lock()
.unwrap()
.contains_key(&request.worker_ref)
);
assert!(matches!(
backend.restore_worker(request.clone()),
WorkerExecutionSpawnResult::Rejected(_)
));
backend.subject_delete_worker(&deletion).unwrap();
backend.subject_delete_worker(&deletion).unwrap();
assert!(execution.handle.protocol_is_closed());
assert!(execution.tasks.tasks.lock().unwrap().is_empty());
assert!(backend.workers.lock().unwrap().is_empty());
assert_eq!(persisted_files(&session_root), before);
assert!(backend.preflight_restore(&request).is_err());
assert!(
backend
.activate_restored_worker(request.operation_id, &request.worker_ref)
.is_err()
);
}
#[test]
#[serial_test::serial(worker_allocation)]
fn subject_delete_pending_restore_fences_late_controller_and_never_restarts_it() {
let root = tempfile::tempdir().unwrap();
let restores = Arc::new(AtomicUsize::new(0));
let release = Arc::new(tokio::sync::Notify::new());
let handles = Arc::new(Mutex::new(Vec::new()));
let backend = WorkerRuntimeExecutionBackend::new(LateControllerFactory {
inner: MockFactory {
client: MockClient::new(vec![]),
runtime_base: root.path().join("control"),
cwd: root.path().to_path_buf(),
store_dir: root.path().join("sessions"),
worker_metadata_dir: root.path().join("metadata"),
observed_cwds: Arc::default(),
observed_workspace_clients: Arc::default(),
},
restores: restores.clone(),
release: release.clone(),
actual_handles: handles.clone(),
})
.unwrap()
.with_spawn_restore_timeout(Duration::from_millis(1));
let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
request.workspace_scope = Some(crate::runtime::RuntimeWorkspaceScope::new(
"ws-test",
"server-test",
));
request.request.workspace_api = Some(WorkspaceApiRef {
workspace_id: "ws-test".into(),
base_url: "https://workspace.invalid".into(),
});
assert!(matches!(
backend.restore_worker(request.clone()),
WorkerExecutionSpawnResult::ReconciliationRequired { .. }
));
let deletion = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: "ws-test".into(),
worker_id: request.worker_ref.worker_id,
create_fingerprint: request.request.create_fingerprint.clone(),
operation_id: "delete-pending".into(),
};
// Unfinished factory ownership is unknown, never shutdown success.
assert!(
backend
.subject_delete_worker(&deletion)
.unwrap_err()
.contains("pending")
);
assert!(
backend
.pending_restores
.lock()
.unwrap()
.contains_key(&request.worker_ref)
);
assert!(
backend
.activate_restored_worker(request.operation_id, &request.worker_ref)
.is_err()
);
release.notify_one();
backend.subject_delete_worker(&deletion).unwrap();
assert!(backend.pending_restores.lock().unwrap().is_empty());
assert!(backend.workers.lock().unwrap().is_empty());
assert_eq!(restores.load(Ordering::SeqCst), 1);
assert_eq!(handles.lock().unwrap().len(), 1);
assert!(handles.lock().unwrap()[0].protocol_is_closed());
assert!(matches!(
backend.restore_worker(request),
WorkerExecutionSpawnResult::Rejected(_)
));
}
crates/worker-runtime/src/subject_delete_create_tests.rs
+101
View File
@@ -0,0 +1,101 @@
// Actual create factory future remains behind Runtime's Worker operation lock;
// deletion must wait for that owner, then permanently reject the delayed retry.
struct SubjectDeleteDeferredCreate {
entered: mpsc::Sender<()>,
release: Arc<tokio::sync::Notify>,
}
#[async_trait]
impl RuntimeWorkerFactory for SubjectDeleteDeferredCreate {
async fn prepare_restore(
&self,
_: WorkerExecutionRestoreRequest,
) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
panic!("fenced create must never restore")
}
async fn spawn_controller(
&self,
_: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
self.entered.send(()).unwrap();
self.release.notified().await;
Err("create failed before any Controller".into())
}
}
#[test]
fn subject_delete_serializes_inflight_create_and_rejects_all_late_admission() {
let root = tempfile::tempdir().unwrap();
let (entered, entry) = mpsc::channel();
let release = Arc::new(tokio::sync::Notify::new());
let backend = Arc::new(
WorkerRuntimeExecutionBackend::new(SubjectDeleteDeferredCreate {
entered,
release: release.clone(),
})
.unwrap(),
);
let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions::new(root.path()).with_runtime_id("delete-test"),
backend.clone(),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test");
let mut create = create_request("inflight create");
create.workspace_api = Some(WorkspaceApiRef {
workspace_id: scope.workspace_id.clone(),
base_url: "https://workspace.invalid".into(),
});
create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot {
workspace_id: scope.workspace_id.clone(),
language: "English".into(),
});
let creating = {
let runtime = runtime.clone();
let scope = scope.clone();
let create = create.clone();
std::thread::spawn(move || runtime.create_worker_scoped(&scope, create))
};
entry.recv_timeout(Duration::from_secs(5)).unwrap();
let request = runtime_api::SubjectDeleteWorkerRequest {
workspace_id: scope.workspace_id.clone(),
worker_id: create.worker_id,
create_fingerprint: create.create_fingerprint.clone(),
operation_id: "delete-creating".into(),
};
let deleting = {
let runtime = runtime.clone();
let scope = scope.clone();
let request = request.clone();
std::thread::spawn(move || runtime.subject_delete_worker_scoped(&scope, request))
};
release.notify_one();
assert!(creating.join().unwrap().is_err());
deleting.join().unwrap().unwrap();
assert!(
runtime
.create_worker_scoped(&scope, create.clone())
.is_err()
);
assert!(backend.workers.lock().unwrap().is_empty());
let worker_ref = WorkerRef::new(create.worker_id);
let spawn = WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(),
request: create.clone(),
workspace_scope: Some(scope.clone()),
context: test_execution_context(worker_ref.clone()),
workdir_attachments: BTreeMap::new(),
config_bundle: None,
};
assert!(matches!(
backend.spawn_worker(spawn),
WorkerExecutionSpawnResult::Rejected(_)
));
let mut restore = direct_restore_request(worker_ref);
restore.request = create;
restore.workspace_scope = Some(scope);
assert!(matches!(
backend.restore_worker(restore),
WorkerExecutionSpawnResult::Rejected(_)
));
}
crates/worker-runtime/src/subject_delete_store_tests.rs
+104
View File
@@ -0,0 +1,104 @@
#[test]
fn subject_delete_schema14_migration_adds_exclusion_store_without_changing_worker_or_receipts() {
for with_fence in [false, true] {
let (_temp, store, worker, path) = canonical_worker_store();
let root = store.root.clone();
let mut legacy =
serde_json::to_value(WorkerAggregateRecord::from_persisted(&worker)).unwrap();
legacy["schema_version"] = serde_json::json!(RESTORE_REPAIR_SCHEMA_VERSION);
fs::write(&path, serde_json::to_vec(&legacy).unwrap()).unwrap();
let id = WorkerId::now_v7();
let request = serde_json::json!({"workspace_id": "workspace-a", "worker_id": id,
"create_fingerprint": "fenced-create", "operation_id": "delete-1"});
let mut response = request.clone();
response["originals_absent"] = false.into();
let fence =
serde_json::json!({ id.to_string(): {"request": request, "response": response} });
let mut runtime = serde_json::json!({"schema_version": RESTORE_REPAIR_SCHEMA_VERSION,
"display_name": "retained", "backend": "fs_store", "status": "running", "next_diagnostic_id": 1,
"workspace_owners": {"workspace-a": "server-a"}, "diagnostics": [],
"config_bundles": {}, "subject_create_recovery_fences": {}, "subject_releases": []});
if with_fence {
runtime["subject_delete_worker_fences"] = fence.clone();
}
fs::write(
root.join(RUNTIME_FILE),
serde_json::to_vec(&runtime).unwrap(),
)
.unwrap();
drop(store);
let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("test-runtime");
let before = fs::read(&path).unwrap();
assert!(
FsRuntimeStore::migration_plan(&options)
.unwrap()
.migration_required
);
assert_eq!(fs::read(&path).unwrap(), before);
let opened = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
legacy["schema_version"] = serde_json::json!(SCHEMA_VERSION);
assert_eq!(
read_json::<serde_json::Value>(&path, "read migrated Worker").unwrap(),
legacy
);
let expected = if with_fence {
fence
} else {
serde_json::json!({})
};
assert_eq!(
serde_json::to_value(&opened.state.as_ref().unwrap().subject_delete_worker_fences)
.unwrap(),
expected
);
opened
.store
.write_runtime_snapshot(opened.state.as_ref().unwrap())
.unwrap();
drop(opened);
let reopened = FsRuntimeStore::open_or_create(root, "test-runtime").unwrap();
assert_eq!(
serde_json::to_value(&reopened.state.unwrap().subject_delete_worker_fences).unwrap(),
expected
);
}
}
#[test]
fn subject_delete_store_rejects_missing_exclusion_field_malformed_receipt_and_workspace_owner() {
for invalid in 0..4 {
let (_temp, store, _worker, _path) = canonical_worker_store();
let root = store.root.clone();
let id = WorkerId::now_v7();
let request = serde_json::json!({"workspace_id": "workspace-a", "worker_id": id,
"create_fingerprint": "fenced-create", "operation_id": "delete-1"});
let mut response = request.clone();
response["originals_absent"] = false.into();
let mut runtime = serde_json::json!({"schema_version": SCHEMA_VERSION,
"display_name": "retained", "backend": "fs_store", "status": "running", "next_diagnostic_id": 1,
"workspace_owners": {"workspace-a": "server-a"}, "diagnostics": [],
"subject_delete_worker_fences": { id.to_string(): {"request": request, "response": response} }});
match invalid {
0 => {
runtime
.as_object_mut()
.unwrap()
.remove("subject_delete_worker_fences");
}
1 => {
runtime["subject_delete_worker_fences"][id.to_string()]["response"]["create_fingerprint"] =
"different".into()
}
2 => runtime["workspace_owners"] = serde_json::json!({}),
_ => {
runtime["subject_delete_worker_fences"][id.to_string()]["request"]["worker_id"] =
WorkerId::now_v7().to_string().into()
}
}
let bytes = serde_json::to_vec(&runtime).unwrap();
fs::write(root.join(RUNTIME_FILE), &bytes).unwrap();
drop(store);
assert!(FsRuntimeStore::open_or_create(root.clone(), "test-runtime").is_err());
assert_eq!(fs::read(root.join(RUNTIME_FILE)).unwrap(), bytes);
}
}
crates/worker-runtime/src/test_support.rs
+3 -1
View File
@@ -215,7 +215,8 @@ impl StoppedWorkerTemplate {
snapshot
.validate(&root.join(RUNTIME_FILE))
.map_err(|_| reject("unsupported Runtime schema"))?;
if !snapshot.subject_create_recovery_fences.is_empty()
if !snapshot.subject_delete_worker_fences.is_empty()
|| !snapshot.subject_create_recovery_fences.is_empty()
|| !snapshot.subject_releases.is_empty()
{
return Err(reject("source has lifecycle operations"));
@@ -427,6 +428,7 @@ impl StoppedWorkerTemplate {
)?;
self.validate_session(&session)?;
store.write_runtime_snapshot(&PersistedRuntimeState {
subject_delete_worker_fences: BTreeMap::new(),
subject_create_recovery_fences: BTreeMap::new(),
subject_releases: Vec::new(),
// Match normal store initialization: this is catalog availability,
crates/worker-runtime/src/worker_backend.rs
+176
View File
@@ -3415,6 +3415,14 @@ struct SubjectControllerReservation {
state: runtime_api::SubjectReleaseState,
}
// Admission identity is retained even when execution publication or stop settlement
// failed. Runtime owns durable exclusion; this guards the adapter's delayed calls.
struct WorkerAdmission {
workspace_id: Option<String>,
create_fingerprint: String,
deletion: Option<runtime_api::SubjectDeleteWorkerRequest>,
}
/// Worker execution backend backed by real worker crate Workers.
pub struct WorkerRuntimeExecutionBackend<F = ProfileRuntimeWorkerFactory> {
#[cfg(test)]
@@ -3438,6 +3446,7 @@ pub struct WorkerRuntimeExecutionBackend<F = ProfileRuntimeWorkerFactory> {
>,
>,
worker_locks: Mutex<HashMap<WorkerRef, Arc<Mutex<()>>>>,
admissions: Mutex<HashMap<WorkerRef, WorkerAdmission>>,
spawn_restore_timeout: Duration,
subject_reservations: Mutex<HashMap<(WorkerRef, String), SubjectControllerReservation>>,
shutdown_admission_timeout: Duration,
@@ -3464,12 +3473,92 @@ where
pending_restores: Mutex::new(HashMap::new()),
completed_restore_failures: Mutex::new(HashMap::new()),
worker_locks: Mutex::new(HashMap::new()),
admissions: Mutex::new(HashMap::new()),
spawn_restore_timeout: SPAWN_RESTORE_TASK_TIMEOUT,
subject_reservations: Mutex::new(HashMap::new()),
shutdown_admission_timeout: Duration::from_secs(5),
})
}
// Caller holds the adapter Worker lock.
fn admit_worker(
&self,
worker: &WorkerRef,
workspace: Option<&str>,
fingerprint: &str,
) -> Result<(), String> {
let mut admissions = self
.admissions
.lock()
.map_err(|_| "worker admission lock poisoned")?;
if let Some(saved) = admissions.get(worker) {
if saved.deletion.is_some() {
return Err("Subject Worker permanently excluded".into());
}
if saved.workspace_id.as_deref() != workspace || saved.create_fingerprint != fingerprint
{
return Err("Worker admission identity mismatch".into());
}
} else {
admissions.insert(
worker.clone(),
WorkerAdmission {
workspace_id: workspace.map(str::to_string),
create_fingerprint: fingerprint.into(),
deletion: None,
},
);
}
Ok(())
}
fn validate_subject_delete_identity(
&self,
request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
let admissions = self
.admissions
.lock()
.map_err(|_| "worker admission lock poisoned")?;
if let Some(saved) = admissions.get(&WorkerRef::new(request.worker_id)) {
if saved.workspace_id.as_deref() != Some(request.workspace_id.as_str())
|| saved.create_fingerprint != request.create_fingerprint
|| saved
.deletion
.as_ref()
.is_some_and(|saved| saved != request)
{
return Err("Subject delete execution identity mismatch".into());
}
}
if let Some(pending) = self.pending_restore(&WorkerRef::new(request.worker_id))? {
if pending
.request
.workspace_scope
.as_ref()
.map(|scope| scope.workspace_id.as_str())
!= Some(request.workspace_id.as_str())
|| pending.request.request.create_fingerprint != request.create_fingerprint
{
return Err("Subject delete pending execution identity mismatch".into());
}
}
Ok(())
}
fn ensure_not_subject_deleted(&self, worker: &WorkerRef) -> Result<(), String> {
if self
.admissions
.lock()
.map_err(|_| "worker admission lock poisoned")?
.get(worker)
.is_some_and(|admission| admission.deletion.is_some())
{
return Err("Subject Worker permanently excluded".into());
}
Ok(())
}
pub fn with_backend_id(mut self, backend_id: impl Into<String>) -> Self {
self.backend_id = backend_id.into();
self
@@ -4739,6 +4828,58 @@ where
&self.backend_id
}
fn preflight_subject_delete_worker(
&self,
request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
let worker = WorkerRef::new(request.worker_id);
let lock = self.worker_lock(&worker)?;
let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?;
self.validate_subject_delete_identity(request)
}
fn subject_delete_worker(
&self,
request: &runtime_api::SubjectDeleteWorkerRequest,
) -> Result<(), String> {
let worker = WorkerRef::new(request.worker_id);
let lock = self.worker_lock(&worker)?;
let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?;
self.validate_subject_delete_identity(request)?;
self.admissions
.lock()
.map_err(|_| "worker admission lock poisoned")?
.insert(
worker.clone(),
WorkerAdmission {
workspace_id: Some(request.workspace_id.clone()),
create_fingerprint: request.create_fingerprint.clone(),
deletion: Some(request.clone()),
},
);
let result = self.stop_worker_under_lock(&worker);
if !result.is_accepted() {
return Err(result.message_or_default());
}
// A live Controller and failed unpublished candidates can coexist. The
// ordinary stop's live branch joins the Controller; also reconcile any
// factory-owned cache/artifacts before issuing deletion proof.
let factory = self.factory.clone();
let stopped = worker.clone();
self.run_cancellable_on_adapter_runtime(RUNTIME_TASK_TIMEOUT, async move {
factory.reconcile_stopped_worker(&stopped).await
})?;
self.subject_reservations
.lock()
.map_err(|_| "Subject reservations poisoned")?
.retain(|(saved, _), _| saved != &worker);
self.completed_restore_failures
.lock()
.map_err(|_| "completed restores poisoned")?
.remove(&worker);
Ok(())
}
fn recover_subject_create(
&self,
request: &runtime_api::SubjectCreateRecoveryRequest,
@@ -5178,6 +5319,19 @@ where
));
}
};
if let Err(message) = self.admit_worker(
&request.worker_ref,
request
.workspace_scope
.as_ref()
.map(|scope| scope.workspace_id.as_str()),
&request.request.create_fingerprint,
) {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message,
));
}
match self.pending_restore(&request.worker_ref) {
Ok(None) => {}
Ok(Some(_)) => {
@@ -5405,6 +5559,10 @@ where
));
}
};
self.ensure_not_subject_deleted(&request.worker_ref)
.map_err(|message| {
WorkerExecutionResult::rejected(WorkerExecutionOperation::Restore, message)
})?;
if self
.pending_restore(&request.worker_ref)
.map_err(|message| {
@@ -5499,6 +5657,19 @@ where
));
}
};
if let Err(message) = self.admit_worker(
&request.worker_ref,
request
.workspace_scope
.as_ref()
.map(|scope| scope.workspace_id.as_str()),
&request.request.create_fingerprint,
) {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
message,
));
}
if let Some(result) =
self.completed_restore_failure(request.operation_id, &request.worker_ref)
{
@@ -6182,6 +6353,7 @@ where
return Err(message);
}
};
self.ensure_not_subject_deleted(worker_ref)?;
if let Some(pending) = self.pending_restore(worker_ref)? {
if pending.request.operation_id != operation_id {
return Err("restored Worker operation identity does not match".into());
@@ -6820,6 +6992,10 @@ mod tests {
};
use worker::PreparedWorker;
include!("subject_delete_backend_tests.rs");
#[cfg(feature = "fs-store")]
include!("subject_delete_create_tests.rs");
// Both candidate and failure suites need the same guarded, checkout-owned
// temporary root, independently of Runtime's optional filesystem catalog.
fn restore_failure_tempdir() -> tempfile::TempDir {