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

feat: runtime commit restore decisions before execution publication

Browse Source 
Persist explicit restore commit evidence, reconcile canonical write uncertainty, and publish the same prepared controller before Accepted. Keep owned cleanup and publication failures distinct. Preserve optional transport/store feature boundaries and adapt external controller constructors.
This commit is contained in:
Hare
2026-10-26 11:50:54 +09:00
parent 848a1e9a5b
commit 1b13563e19
18 changed files with 3348 additions and 598 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/Cargo.toml
+2 -2
View File
@@ -22,14 +22,14 @@ fs-store = ["dep:fs4"]
# Offline fixture tooling only; never enabled by the product defaults.
test-support = ["fs-store", "ws-server"]
http-server = ["dep:axum", "dep:tower", "dep:reqwest"]
ws-server = ["http-server", "axum/ws", "dep:futures", "tokio/sync"]
ws-server = ["http-server", "axum/ws", "tokio/sync"]
[dependencies]
agen.workspace = true
async-trait.workspace = true
base64.workspace = true
axum = { workspace = true, optional = true }
futures = { workspace = true, optional = true }
futures.workspace = true
decodal.workspace = true
flow = { path = "../flow" }
fs4 = { workspace = true, optional = true }
crates/worker-runtime/src/execution.rs
+27
View File
@@ -650,6 +650,24 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
self.restore_worker(request)
}
/// Discard only the exact unpublished operation after Runtime has proved
/// its durable decision uncommitted. This is not Stop and must not affect a
/// connected execution. RolledBack requires confirmed cleanup and no repair.
fn discard_uncommitted_restore(
&self,
_operation_id: WorkerLifecycleOperationId,
_worker_ref: &WorkerRef,
) -> WorkerExecutionSpawnResult {
WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::unsupported(
WorkerExecutionOperation::Restore,
"backend cannot confirm unpublished restore cleanup",
),
worker_state: None,
workdir_attachments: Vec::new(),
}
}
fn activate_restored_worker(
&self,
_operation_id: WorkerLifecycleOperationId,
@@ -927,6 +945,15 @@ impl WorkerExecutionBackendRef {
self.backend.reconcile_restore(request)
}
pub(crate) fn discard_uncommitted_restore(
&self,
operation_id: WorkerLifecycleOperationId,
worker_ref: &WorkerRef,
) -> WorkerExecutionSpawnResult {
self.backend
.discard_uncommitted_restore(operation_id, worker_ref)
}
pub(crate) fn activate_restored_worker(
&self,
operation_id: WorkerLifecycleOperationId,
crates/worker-runtime/src/fs_store.rs
+554 -33
View File
@@ -43,7 +43,8 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
#[path = "test_support.rs"]
pub mod test_support;
const SCHEMA_VERSION: u32 = 12;
pub(crate) const SCHEMA_VERSION: u32 = 13;
const RESTORE_ADMISSION_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;
@@ -124,6 +125,8 @@ pub struct FsRuntimeStore {
root: PathBuf,
_owner_lock: Option<Arc<RuntimeStoreOwnerLock>>,
worker_write_fault: Arc<AtomicU8>,
#[cfg(test)]
worker_reconcile_sync_fault: Arc<std::sync::atomic::AtomicBool>,
}
impl PartialEq for FsRuntimeStore {
@@ -164,6 +167,12 @@ impl FsRuntimeStore {
.store(fault as u8, Ordering::Release);
}
#[cfg(test)]
pub(crate) fn fail_next_worker_reconcile_sync(&self) {
self.worker_reconcile_sync_fault
.store(true, Ordering::Release);
}
pub(crate) fn open_or_create(
root: PathBuf,
runtime_id: &str,
@@ -223,6 +232,8 @@ impl FsRuntimeStore {
root,
_owner_lock: Some(owner_lock),
worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)),
#[cfg(test)]
worker_reconcile_sync_fault: Default::default(),
};
let state = if snapshot_exists {
Some(store.load_runtime_state()?)
@@ -261,6 +272,124 @@ impl FsRuntimeStore {
Ok(())
}
/// Read one canonical aggregate without migration, cleanup, or tolerant
/// startup enumeration. Reconciliation must not interpret an unreadable
/// record as absent or replace it with an older in-memory projection.
#[cfg(test)]
pub(crate) fn read_worker_record(
&self,
worker_ref: &WorkerRef,
) -> Result<PersistedWorkerRecord, RuntimeError> {
self.read_worker_record_file(worker_ref)
.map(|(record, _)| record)
}
/// Reobserve and sync canonical authority after an uncertain aggregate write.
///
/// The caller must hold the Worker operation lock; this store retains the
/// Runtime owner lock. Sync the same file that was validated, then its directory
/// and the directories owning newly created Worker paths. Read equality alone
/// cannot discharge a failed post-rename sync. Any error leaves the caller's
/// write outcome unresolved; this never writes an older memory projection.
/// A returned record proves neither a matching operation nor live activation:
/// the lifecycle owner must still compare request/operation, Stop and receipt.
pub(crate) fn read_durable_worker_record(
&self,
worker_ref: &WorkerRef,
) -> Result<PersistedWorkerRecord, RuntimeError> {
self.read_durable_worker_record_with_sync(worker_ref, |file| {
#[cfg(test)]
if self
.worker_reconcile_sync_fault
.swap(false, Ordering::AcqRel)
{
return Err(std::io::Error::other(
"injected reconciliation sync failure",
));
}
file.sync_all()
})
}
/// Compare the entire typed aggregate, including request owner, receipt,
/// attachments and Stop, after durability is confirmed under owner locks.
pub(crate) fn reconcile_worker_record(
&self,
expected: &PersistedWorkerRecord,
) -> Result<(PersistedWorkerRecord, bool), RuntimeError> {
let observed = self.read_durable_worker_record(&expected.worker_ref)?;
let document = |record: &PersistedWorkerRecord| {
serde_json::to_value(WorkerAggregateRecord::from_persisted(record))
.map_err(|error| RuntimeError::InvalidRequest(error.to_string()))
};
let matches = document(&observed)? == document(expected)?;
Ok((observed, matches))
}
fn read_durable_worker_record_with_sync(
&self,
worker_ref: &WorkerRef,
mut sync: impl FnMut(&File) -> std::io::Result<()>,
) -> Result<PersistedWorkerRecord, RuntimeError> {
if self._owner_lock.is_none() {
return Err(RuntimeError::InvalidRequest(
"Worker aggregate reconciliation requires the Runtime store owner lock".to_string(),
));
}
let (record, file) = self.read_worker_record_file(worker_ref)?;
let worker_dir = self.worker_dir(&worker_ref.worker_id);
let path = worker_dir.join(WORKER_FILE);
let unknown = |source| RuntimeError::StoreCommitOutcomeUnknown {
operation: "sync reconciled Worker aggregate",
path: path.clone(),
source,
};
sync(&file).map_err(&unknown)?;
for directory in [worker_dir, self.root.join(WORKERS_DIR), self.root.clone()] {
let file = File::open(directory).map_err(&unknown)?;
sync(&file).map_err(&unknown)?;
}
Ok(record)
}
fn read_worker_record_file(
&self,
worker_ref: &WorkerRef,
) -> Result<(PersistedWorkerRecord, File), RuntimeError> {
let worker_dir = self.worker_dir(&worker_ref.worker_id);
for directory in [&self.root, &self.root.join(WORKERS_DIR), &worker_dir] {
let metadata = fs::symlink_metadata(directory).map_err(|source| {
if source.kind() == std::io::ErrorKind::NotFound {
RuntimeError::StoreMissing {
operation: "reconcile Worker aggregate",
path: directory.to_path_buf(),
}
} else {
runtime_io_error("reconcile Worker aggregate", directory, source)
}
})?;
if !metadata.file_type().is_dir() {
return Err(RuntimeError::StoreCorrupt {
operation: "reconcile Worker aggregate",
path: directory.to_path_buf(),
message: "Worker aggregate directory is not a regular directory".to_string(),
});
}
}
let path = worker_dir.join(WORKER_FILE);
let (aggregate, file): (WorkerAggregateRecord, File) =
read_bounded_json_file(&path, "reconcile Worker aggregate")?;
let record = aggregate.validate(&path)?;
if record.worker_ref != *worker_ref || record.worker_id != worker_ref.worker_id {
return Err(RuntimeError::StoreCorrupt {
operation: "reconcile Worker aggregate",
path,
message: "Worker aggregate does not match the requested identity".to_string(),
});
}
Ok((record, file))
}
pub(crate) fn delete_worker_record(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
self.delete_worker_record_with_intent(
worker_id,
@@ -523,6 +652,7 @@ pub(crate) enum PersistedWorkerRestoreMode {
pub(crate) struct PersistedWorkerRestoreOperation {
pub(crate) operation_id: WorkerLifecycleOperationId,
pub(crate) mode: PersistedWorkerRestoreMode,
pub(crate) commit: crate::runtime::RestoreCommitFact,
pub(crate) request: CreateWorkerRequest,
pub(crate) binding: Option<PersistedWorkerExecutionBinding>,
pub(crate) last_settled_status: WorkerStatus,
@@ -648,8 +778,8 @@ fn plan_runtime_store_migration(
format!("Runtime store schema version {schema_version} is out of range"),
)
})?;
let staging = migration_sibling(root, "schema-v12-staging")?;
let backup = migration_sibling(root, "pre-schema-v12-backup")?;
let staging = migration_sibling(root, "schema-v13-staging")?;
let backup = migration_sibling(root, "pre-schema-v13-backup")?;
if staging.exists() || backup.exists() {
return Err(runtime_store_corrupt(
root,
@@ -698,11 +828,12 @@ fn plan_runtime_store_migration(
| AGGREGATE_SCHEMA_VERSION
| EXECUTION_SCHEMA_VERSION
| SOURCE_SCHEMA_VERSION
| RESTORE_ADMISSION_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}, 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}, or {SCHEMA_VERSION}"
),
));
}
@@ -1034,10 +1165,10 @@ fn migrate_worker_record(
) -> Result<WorkerAggregateRecord, RuntimeError> {
if matches!(
source_schema_version,
EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION
EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION
) {
// Schemas 10 and 11 already use the current Worker aggregate layout.
// Only advance its version; never regenerate Restore admission identity.
// Admission identity survives migration. Older pending records did not
// distinguish a committed decision from preparation or failed cleanup.
if identity_document
.get("schema_version")
.and_then(serde_json::Value::as_u64)
@@ -1048,6 +1179,7 @@ fn migrate_worker_record(
"Worker schema does not match Runtime migration source".into(),
));
}
migrate_restore_commit_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()))?;
@@ -1059,6 +1191,7 @@ fn migrate_worker_record(
remove_legacy_job_input_counter(&mut identity_document);
migrate_legacy_request_evidence(&mut identity_document)
.map_err(|message| runtime_store_corrupt(identity_path, message.into()))?;
migrate_restore_commit_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| {
@@ -1125,6 +1258,42 @@ fn migrate_worker_record(
Ok(WorkerAggregateRecord::from_v8(identity, execution))
}
// Missing commit evidence is not proof of rollback. This conversion applies
// only to versioned legacy aggregates, including Restore nested under Stop.
fn migrate_restore_commit_fact(
document: &mut serde_json::Value,
path: &Path,
) -> Result<(), RuntimeError> {
let execution = &mut document["execution_state"];
if execution["state"] != "reconciliation_required" {
return Ok(());
}
let operation = &mut execution["execution"];
let restore = match operation["operation"].as_str() {
Some("restore") => Some(&mut operation["intent"]),
Some("stop") => operation["intent"]
.as_object_mut()
.and_then(|stop| stop.get_mut("pending_restore"))
.filter(|restore| !restore.is_null()),
_ => None,
};
if let Some(restore) = restore {
let restore = restore.as_object_mut().ok_or_else(|| {
runtime_store_corrupt(path, "legacy pending Restore must be an object".into())
})?;
if restore
.insert("commit".into(), serde_json::json!("legacy_unknown"))
.is_some()
{
return Err(runtime_store_corrupt(
path,
"legacy Restore contains unexpected commit evidence".into(),
));
}
}
Ok(())
}
// Frozen schema-v9 conversion. Preserve the exact token and all Restore owners
// and receipts; generating a fresh token here would lose retry/admission authority.
fn migrate_v9_restore_guard(
@@ -1437,7 +1606,10 @@ fn migrate_runtime_document(
object
.entry("subject_create_recovery_fences")
.or_insert_with(|| serde_json::json!({}));
if source_schema_version != SOURCE_SCHEMA_VERSION {
if !matches!(
source_schema_version,
SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_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.
@@ -1505,6 +1677,8 @@ fn validate_current_runtime_store(root: &Path) -> Result<(), RuntimeError> {
root: root.to_path_buf(),
_owner_lock: None,
worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)),
#[cfg(test)]
worker_reconcile_sync_fault: Default::default(),
};
store.load_runtime_state().map(|_| ())
}
@@ -1514,6 +1688,8 @@ fn validate_completed_migration_staging(root: &Path) -> Result<(), RuntimeError>
root: root.to_path_buf(),
_owner_lock: None,
worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)),
#[cfg(test)]
worker_reconcile_sync_fault: Default::default(),
};
let state = store.load_runtime_state()?;
if let Some(worker) = state.workers.values().find(|worker| {
@@ -1559,14 +1735,16 @@ fn recover_runtime_store_migration(root: &Path) -> Result<(), RuntimeError> {
"pre-schema-v10-backup",
"schema-v11-staging",
"pre-schema-v11-backup",
"schema-v12-staging",
"pre-schema-v12-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-v12 startup".into()));
return Err(runtime_store_corrupt(root, "unfinished earlier Runtime migration requires explicit recovery before schema-v13 startup".into()));
}
}
let staging = migration_sibling(root, "schema-v12-staging")?;
let backup = migration_sibling(root, "pre-schema-v12-backup")?;
let staging = migration_sibling(root, "schema-v13-staging")?;
let backup = migration_sibling(root, "pre-schema-v13-backup")?;
let root_exists = migration_directory_exists(root)?;
let staging_exists = migration_directory_exists(&staging)?;
let backup_exists = migration_directory_exists(&backup)?;
@@ -1768,8 +1946,8 @@ fn migrate_runtime_store(
if !plan.migration_required {
return Ok(plan);
}
let staging = migration_sibling(root, "schema-v12-staging")?;
let backup = migration_sibling(root, "pre-schema-v12-backup")?;
let staging = migration_sibling(root, "schema-v13-staging")?;
let backup = migration_sibling(root, "pre-schema-v13-backup")?;
if staging.exists() || backup.exists() {
return Err(runtime_store_corrupt(
root,
@@ -1966,9 +2144,12 @@ fn migrate_runtime_store_in_place(
));
}
}
// Schema 11 metadata was validated above but needs no conversion.
// Schema 11/12 metadata was validated above but needs no conversion.
if let Some(metadata) = metadata
&& plan.current_schema_version != SOURCE_SCHEMA_VERSION
&& !matches!(
plan.current_schema_version,
SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION
)
{
atomic_write_json(
&migrated_dir.join(WORKER_METADATA_FILE),
@@ -2597,6 +2778,19 @@ impl WorkerExecutionRecord {
}
fn read_bounded_json<T>(path: &Path, operation: &'static str) -> Result<T, RuntimeError>
where
T: for<'de> Deserialize<'de>,
{
read_bounded_json_file(path, operation).map(|(value, _)| value)
}
// Keep the validated file open so reconciliation syncs the observed authority,
// not a separately reopened projection. The Runtime owner and Worker locks
// exclude cooperating writers throughout the read and sync.
fn read_bounded_json_file<T>(
path: &Path,
operation: &'static str,
) -> Result<(T, File), RuntimeError>
where
T: for<'de> Deserialize<'de>,
{
@@ -2628,13 +2822,14 @@ where
),
});
}
let file = File::open(path).map_err(|source| RuntimeError::StoreIo {
let mut file = File::open(path).map_err(|source| RuntimeError::StoreIo {
operation,
path: path.to_path_buf(),
source,
})?;
let mut bytes = Vec::with_capacity(metadata.len() as usize);
file.take(MAX_WORKER_RECORD_BYTES + 1)
(&mut file)
.take(MAX_WORKER_RECORD_BYTES + 1)
.read_to_end(&mut bytes)
.map_err(|source| RuntimeError::StoreIo {
operation,
@@ -2648,11 +2843,12 @@ where
message: format!("record exceeds limit {MAX_WORKER_RECORD_BYTES} while reading"),
});
}
serde_json::from_slice(&bytes).map_err(|source| RuntimeError::StoreCorrupt {
let value = serde_json::from_slice(&bytes).map_err(|source| RuntimeError::StoreCorrupt {
operation,
path: path.to_path_buf(),
message: source.to_string(),
})
})?;
Ok((value, file))
}
fn read_json<T>(path: &Path, operation: &'static str) -> Result<T, RuntimeError>
@@ -2913,7 +3109,7 @@ mod tests {
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_eq!(plan.target_schema_version, SCHEMA_VERSION);
assert!(plan.migration_required);
assert_eq!(plan.worker_count, 1);
assert_eq!(plan.migrated_worker_aggregate_count, 0);
@@ -2928,13 +3124,13 @@ mod tests {
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);
expected_worker["schema_version"] = serde_json::json!(SCHEMA_VERSION);
assert_eq!(
serde_json::to_value(WorkerAggregateRecord::from_persisted(persisted_worker))
.unwrap(),
expected_worker
);
runtime["schema_version"] = serde_json::json!(12);
runtime["schema_version"] = serde_json::json!(SCHEMA_VERSION);
if fence_state == "absent" {
runtime["subject_create_recovery_fences"] = serde_json::json!({});
}
@@ -2970,8 +3166,8 @@ mod tests {
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_eq!(plan.current_schema_version, SCHEMA_VERSION);
assert_eq!(plan.target_schema_version, SCHEMA_VERSION);
assert!(
!FsRuntimeStore::migration_plan(&options)
.unwrap()
@@ -2982,12 +3178,12 @@ mod tests {
assert_eq!(fs::read(&metadata_path).unwrap(), metadata);
assert_eq!(fs::read(&history_path).unwrap(), history);
assert!(
!migration_sibling(&root, "schema-v12-staging")
!migration_sibling(&root, "schema-v13-staging")
.unwrap()
.exists()
);
assert!(
!migration_sibling(&root, "pre-schema-v12-backup")
!migration_sibling(&root, "pre-schema-v13-backup")
.unwrap()
.exists()
);
@@ -3456,7 +3652,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-v12-staging").unwrap();
let staging = migration_sibling(&root, "schema-v13-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();
@@ -3471,7 +3667,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-v12-backup").unwrap();
let backup = migration_sibling(&root, "pre-schema-v13-backup").unwrap();
fs::rename(&root, &backup).unwrap();
let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
@@ -3485,10 +3681,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-v12-staging").unwrap();
let staging = migration_sibling(&root, "schema-v13-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-v12-backup").unwrap();
let backup = migration_sibling(&root, "pre-schema-v13-backup").unwrap();
fs::rename(&root, &backup).unwrap();
let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
@@ -3503,10 +3699,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-v12-staging").unwrap();
let staging = migration_sibling(&root, "schema-v13-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-v12-backup").unwrap();
let backup = migration_sibling(&root, "pre-schema-v13-backup").unwrap();
fs::rename(&root, &backup).unwrap();
let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
@@ -3523,7 +3719,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-v12-backup").unwrap();
let backup = migration_sibling(&root, "pre-schema-v13-backup").unwrap();
write_empty_schema_v8_store(&backup);
let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
@@ -3538,6 +3734,8 @@ mod tests {
"pre-schema-v10-backup",
"schema-v11-staging",
"pre-schema-v11-backup",
"schema-v12-staging",
"pre-schema-v12-backup",
] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime-store");
@@ -3692,6 +3890,7 @@ mod tests {
let error = opened.store.write_worker_record(&worker).unwrap_err();
assert!(matches!(error, RuntimeError::StoreIo { .. }));
assert!(!aggregate_path.exists());
assert!(opened.store.read_worker_record(&worker.worker_ref).is_err());
worker.created_at_ms = Some(1);
opened.store.write_worker_record(&worker).unwrap();
@@ -3707,6 +3906,17 @@ mod tests {
let committed: WorkerAggregateRecord =
read_bounded_json(&aggregate_path, "read test Worker aggregate").unwrap();
assert_eq!(committed.created_at_ms, Some(2));
let before = fs::read(&aggregate_path).unwrap();
assert_eq!(
opened
.store
.read_worker_record(&worker.worker_ref)
.unwrap()
.created_at_ms,
Some(2),
"unknown write must be reconciled from the canonical record"
);
assert_eq!(fs::read(&aggregate_path).unwrap(), before);
assert!(
!aggregate_path
.with_file_name(WORKER_EXECUTION_FILE)
@@ -3714,6 +3924,317 @@ mod tests {
);
}
#[test]
fn canonical_worker_read_rejects_unreadable_or_wrong_identity_without_repair() {
let temp = tempfile::tempdir().unwrap();
let opened =
FsRuntimeStore::open_or_create(temp.path().join("runtime-store"), "test-runtime")
.unwrap();
let requested = WorkerRef::new(WorkerId::now_v7());
let path = opened
.store
.worker_dir(&requested.worker_id)
.join(WORKER_FILE);
assert!(opened.store.read_worker_record(&requested).is_err());
assert!(!path.exists());
fs::create_dir_all(path.parent().unwrap()).unwrap();
let foreign = WorkerId::now_v7();
let foreign_record = migrate_worker_record(
schema_v7_worker_document(foreign),
LEGACY_SCHEMA_VERSION,
None,
&path
.parent()
.unwrap()
.parent()
.unwrap()
.join(foreign.to_string())
.join(WORKER_FILE),
)
.unwrap();
let documents = [
b"not-json".to_vec(),
serde_json::to_vec(&foreign_record).unwrap(),
];
for document in documents {
fs::write(&path, &document).unwrap();
assert!(opened.store.read_worker_record(&requested).is_err());
assert_eq!(fs::read(&path).unwrap(), document);
}
File::create(&path)
.unwrap()
.set_len(MAX_WORKER_RECORD_BYTES + 1)
.unwrap();
assert!(opened.store.read_worker_record(&requested).is_err());
assert_eq!(
fs::metadata(&path).unwrap().len(),
MAX_WORKER_RECORD_BYTES + 1
);
}
fn canonical_worker_store() -> (
tempfile::TempDir,
FsRuntimeStore,
PersistedWorkerRecord,
PathBuf,
) {
let temp = tempfile::tempdir().unwrap();
let opened =
FsRuntimeStore::open_or_create(temp.path().join("runtime-store"), "test-runtime")
.unwrap();
let worker_id = WorkerId::now_v7();
let path = opened.store.worker_dir(&worker_id).join(WORKER_FILE);
let mut worker = migrate_worker_record(
schema_v7_worker_document(worker_id),
LEGACY_SCHEMA_VERSION,
None,
&path,
)
.unwrap()
.validate(&path)
.unwrap();
worker.created_at_ms = Some(1);
opened.store.write_worker_record(&worker).unwrap();
(temp, opened.store, worker, path)
}
#[test]
fn legacy_pending_restore_migrates_as_unknown_and_preserves_stop_ownership() {
use crate::runtime::RestoreCommitFact;
for under_stop in [false, true] {
let (_temp, store, mut worker, path) = canonical_worker_store();
let root = store.root.clone();
let PersistedWorkerExecutionState::Available(execution) =
worker.execution_state.clone()
else {
panic!("fixture execution unavailable")
};
let restore = PersistedWorkerRestoreOperation {
operation_id: WorkerLifecycleOperationId::new(),
mode: PersistedWorkerRestoreMode::Explicit,
commit: RestoreCommitFact::Committed,
request: execution.request.clone(),
binding: execution.binding.clone(),
last_settled_status: worker.status,
};
let operation = if under_stop {
PersistedWorkerLifecycleOperation::Stop(PersistedWorkerStopOperation {
operation_id: WorkerLifecycleOperationId::new(),
pending_restore: Some(restore),
request: Some(execution.request),
binding: execution.binding,
restore_intent: execution.restore_intent,
last_settled_status: worker.status,
})
} else {
PersistedWorkerLifecycleOperation::Restore(restore)
};
worker.execution_state =
PersistedWorkerExecutionState::ReconciliationRequired(operation);
let mut legacy =
serde_json::to_value(WorkerAggregateRecord::from_persisted(&worker)).unwrap();
legacy["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION);
let intent = &mut legacy["execution_state"]["execution"]["intent"];
let restore = if under_stop {
&mut intent["pending_restore"]
} else {
intent
};
restore.as_object_mut().unwrap().remove("commit");
let expected_request = restore["request"].clone();
let before_guard = legacy["restore_guard"].clone();
fs::write(&path, serde_json::to_vec(&legacy).unwrap()).unwrap();
write_empty_schema_v8_store(&root);
let runtime_path = root.join(RUNTIME_FILE);
let mut runtime: serde_json::Value = read_json(&runtime_path, "read fixture").unwrap();
runtime["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION);
fs::write(&runtime_path, serde_json::to_vec(&runtime).unwrap()).unwrap();
drop(store);
let opened = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap();
let saved = opened.store.read_worker_record(&worker.worker_ref).unwrap();
let PersistedWorkerExecutionState::ReconciliationRequired(operation) =
saved.execution_state
else {
panic!("migration discarded pending ownership")
};
let restored = match operation {
PersistedWorkerLifecycleOperation::Restore(restore) => restore,
PersistedWorkerLifecycleOperation::Stop(stop) => stop.pending_restore.unwrap(),
};
assert_eq!(restored.commit, RestoreCommitFact::LegacyUnknown);
assert_eq!(
serde_json::to_value(restored.request).unwrap(),
expected_request
);
let disk: serde_json::Value = read_json(&path, "read migrated Worker").unwrap();
assert_eq!(disk["restore_guard"], before_guard);
assert_eq!(disk["worker_ref"], legacy["worker_ref"]);
let settled = disk["execution_state"]["execution"]["intent"].clone();
let legacy_intent = &legacy["execution_state"]["execution"]["intent"];
if under_stop {
assert_eq!(settled["operation_id"], legacy_intent["operation_id"]);
assert_eq!(
settled["pending_restore"]["operation_id"],
legacy_intent["pending_restore"]["operation_id"]
);
} else {
assert_eq!(settled["operation_id"], legacy_intent["operation_id"]);
}
drop(opened);
let reopened = FsRuntimeStore::open_or_create(root, "test-runtime").unwrap();
assert!(
reopened
.store
.read_worker_record(&worker.worker_ref)
.is_ok()
);
}
}
#[test]
fn uncertain_worker_write_sync_failure_stays_unknown_until_sync_recovers() {
// Real filesystem aggregate/rename/read/validation; substitute only a
// failing file/directory sync syscall, never the reconciliation owner.
let (_temp, store, mut worker, path) = canonical_worker_store();
worker.created_at_ms = Some(2);
store.fail_next_worker_write(AtomicWriteFault::AfterRename);
assert!(matches!(
store.write_worker_record(&worker),
Err(RuntimeError::StoreCommitOutcomeUnknown { .. })
));
let bytes = fs::read(&path).unwrap();
assert_eq!(
store
.read_worker_record(&worker.worker_ref)
.unwrap()
.created_at_ms,
Some(2)
);
// The readable replacement is not a durability confirmation. Inject
// failure at the file and at each directory owning its canonical path.
for fail_at in 0..4 {
let mut sync_count = 0;
let result = store.read_durable_worker_record_with_sync(&worker.worker_ref, |file| {
let index = sync_count;
sync_count += 1;
if index == fail_at {
Err(std::io::Error::other(
"injected reconciliation sync failure",
))
} else {
file.sync_all()
}
});
assert!(matches!(
result,
Err(RuntimeError::StoreCommitOutcomeUnknown { .. })
));
assert_eq!(fs::read(&path).unwrap(), bytes);
}
let reconciled = store
.read_durable_worker_record(&worker.worker_ref)
.unwrap();
assert_eq!(reconciled.created_at_ms, Some(2));
assert_eq!(reconciled.execution_state, worker.execution_state);
assert_eq!(fs::read(&path).unwrap(), bytes);
}
#[test]
fn pre_rename_failure_durable_reread_preserves_old_canonical_record() {
let (_temp, store, mut worker, path) = canonical_worker_store();
let bytes = fs::read(&path).unwrap();
worker.created_at_ms = Some(2);
store.fail_next_worker_write(AtomicWriteFault::BeforeRename);
assert!(matches!(
store.write_worker_record(&worker),
Err(RuntimeError::StoreIo { .. })
));
let reconciled = store
.read_durable_worker_record(&worker.worker_ref)
.unwrap();
assert_eq!(reconciled.created_at_ms, Some(1));
assert_eq!(fs::read(&path).unwrap(), bytes);
}
#[test]
fn durable_worker_reread_rejects_invalid_authority_before_sync_without_repair() {
let (_temp, store, worker, path) = canonical_worker_store();
let bytes = fs::read(&path).unwrap();
let foreign = WorkerId::now_v7();
let foreign_record = migrate_worker_record(
schema_v7_worker_document(foreign),
LEGACY_SCHEMA_VERSION,
None,
&path
.parent()
.unwrap()
.parent()
.unwrap()
.join(foreign.to_string())
.join(WORKER_FILE),
)
.unwrap();
for document in [
b"not-json".to_vec(),
serde_json::to_vec(&foreign_record).unwrap(),
] {
fs::write(&path, &document).unwrap();
assert!(matches!(
store.read_durable_worker_record_with_sync(&worker.worker_ref, |_| {
panic!("invalid authority must not reach durability confirmation")
}),
Err(RuntimeError::StoreCorrupt { .. })
));
assert_eq!(fs::read(&path).unwrap(), document);
}
File::create(&path)
.unwrap()
.set_len(MAX_WORKER_RECORD_BYTES + 1)
.unwrap();
assert!(matches!(
store.read_durable_worker_record_with_sync(&worker.worker_ref, |_| {
panic!("oversized authority must not reach durability confirmation")
}),
Err(RuntimeError::StoreCorrupt { .. })
));
assert_eq!(
fs::metadata(&path).unwrap().len(),
MAX_WORKER_RECORD_BYTES + 1
);
fs::remove_file(&path).unwrap();
assert!(matches!(
store.read_durable_worker_record_with_sync(&worker.worker_ref, |_| {
panic!("missing authority must not reach durability confirmation")
}),
Err(RuntimeError::StoreMissing { .. })
));
assert!(!path.exists());
fs::write(&path, &bytes).unwrap();
assert!(store.read_durable_worker_record(&worker.worker_ref).is_ok());
assert_eq!(fs::read(&path).unwrap(), bytes);
}
#[cfg(unix)]
#[test]
fn canonical_worker_reread_rejects_symlinked_worker_directory() {
let (_temp, store, worker, path) = canonical_worker_store();
let worker_dir = path.parent().unwrap();
let retained_dir = worker_dir.with_extension("retained");
fs::rename(worker_dir, &retained_dir).unwrap();
std::os::unix::fs::symlink(&retained_dir, worker_dir).unwrap();
let bytes = fs::read(retained_dir.join(WORKER_FILE)).unwrap();
assert!(matches!(
store.read_worker_record(&worker.worker_ref),
Err(RuntimeError::StoreCorrupt { .. })
));
assert!(matches!(
store.read_durable_worker_record(&worker.worker_ref),
Err(RuntimeError::StoreCorrupt { .. })
));
assert_eq!(fs::read(retained_dir.join(WORKER_FILE)).unwrap(), bytes);
}
#[test]
fn schema_v8_pair_migrates_to_one_worker_aggregate() {
let temp = tempfile::tempdir().unwrap();
crates/worker-runtime/src/http_server.rs
+84
View File
@@ -25,6 +25,7 @@ use crate::interaction::{WorkerInput, WorkerInteractionAck};
use crate::management::{RuntimeSummary, WorkerDeleteResult};
#[cfg(feature = "ws-server")]
use crate::observation::WorkerObservationCursor;
#[cfg(feature = "fs-store")]
use crate::retention::{
WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory,
};
@@ -1567,6 +1568,7 @@ fn protocol_error_event(message: impl Into<String>) -> protocol::Event {
}
}
#[cfg(feature = "fs-store")]
#[allow(dead_code)]
async fn worker_retention_inventory(
State(state): State<RuntimeHttpState>,
@@ -1588,6 +1590,7 @@ async fn worker_retention_inventory(
.map_err(RuntimeHttpRestError::runtime)
}
#[cfg(feature = "fs-store")]
#[allow(dead_code)]
async fn execute_worker_retention(
State(state): State<RuntimeHttpState>,
@@ -2620,6 +2623,87 @@ mod tests {
.unwrap()
}
// Generated management routes and real Workspace capability middleware stay
// connected even when this process has no filesystem retention provider.
#[cfg(not(feature = "fs-store"))]
#[tokio::test]
async fn memory_retention_routes_preserve_auth_and_return_typed_unavailable() {
let identity = RuntimeIdentityMaterial::generate("workspace-key").unwrap();
let trust = workspace_trust(&identity);
let runtime = Runtime::new_memory();
let app = runtime_http_router_with_workspace_auth(
runtime,
None,
workspace_test_auth(vec![trust.clone()]),
);
let worker_id = WorkerId::now_v7();
let body = runtime_api::WorkerRetentionExecutionRequest {
operation_id: "memory-retention".into(),
input_fingerprint: "input".into(),
archive_id: None,
workspace_id: "workspace-a".into(),
source_runtime_id: "runtime-test".into(),
worker_id,
source_created_at: "2026-01-01T00:00:00Z".into(),
removed_at: "2026-01-02T00:00:00Z".into(),
effective_profile: None,
retention_class: None,
policy_id: "policy".into(),
session_disposition: runtime_api::SessionDisposition::Purge,
diagnostics_disposition: runtime_api::DiagnosticsDisposition::Purge,
};
for (method, path, body) in [
(
"GET",
format!("/v1/workers/{worker_id}/retention/inventory"),
vec![],
),
(
"POST",
format!("/v1/workers/{worker_id}/retention/execute"),
serde_json::to_vec(&body).unwrap(),
),
] {
let unsigned = app
.clone()
.oneshot(
Request::builder()
.method(method)
.uri(&path)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(body.clone()))
.unwrap(),
)
.await
.unwrap();
assert_eq!(unsigned.status(), StatusCode::UNAUTHORIZED);
let claims = signed_claims(&trust, method, &path, &body);
let response =
signed_request(app.clone(), &identity, &claims, body, "workspace-a").await;
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
let error: RuntimeHttpErrorResponse = read_json(response).await;
assert_eq!(error.error.code, "worker_retention_unavailable");
}
for (mut value, expected_status, expected_code) in [
(body.clone(), StatusCode::NOT_FOUND, "worker_not_found"),
(body.clone(), StatusCode::BAD_REQUEST, "worker_id_mismatch"),
] {
if expected_status == StatusCode::NOT_FOUND {
value.workspace_id = "workspace-b".into();
} else {
value.worker_id = WorkerId::now_v7();
}
let path = format!("/v1/workers/{worker_id}/retention/execute");
let bytes = serde_json::to_vec(&value).unwrap();
let claims = signed_claims(&trust, "POST", &path, &bytes);
let response =
signed_request(app.clone(), &identity, &claims, bytes, "workspace-a").await;
assert_eq!(response.status(), expected_status);
let error: RuntimeHttpErrorResponse = read_json(response).await;
assert_eq!(error.error.code, expected_code);
}
}
#[tokio::test]
async fn ordinary_signed_ping_succeeds_without_handshake_and_rejects_replay() {
let identity = RuntimeIdentityMaterial::generate("workspace-key").unwrap();
crates/worker-runtime/src/http_server/runtime_management_api.rs
+24 -3
View File
@@ -856,6 +856,8 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
)
})?;
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
#[cfg(feature = "fs-store")]
{
let inventory = self
.state
.runtime
@@ -864,6 +866,16 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
.map_err(api_error)?;
response(inventory)
}
#[cfg(not(feature = "fs-store"))]
{
let _ = (scope, worker_ref);
Err(runtime_api::RuntimeApiError::new(
503,
"worker_retention_unavailable",
"Worker retention requires filesystem storage",
))
}
}
async fn execute_retention(
&self,
@@ -880,21 +892,23 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
)
})?;
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let request: WorkerRetentionExecutionRequest = request(value)?;
if request.workspace_id != scope.workspace_id {
if value.workspace_id != scope.workspace_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::NOT_FOUND.as_u16(),
"worker_not_found",
"worker does not belong to the authenticated Workspace",
));
}
if request.worker_id != worker_ref.worker_id {
if value.worker_id != worker_ref.worker_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::BAD_REQUEST.as_u16(),
"worker_id_mismatch",
"Worker retention request worker_id does not match route worker_id",
));
}
#[cfg(feature = "fs-store")]
{
let request: WorkerRetentionExecutionRequest = request(value)?;
let result = self
.state
.runtime
@@ -903,4 +917,11 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
.map_err(api_error)?;
response(result)
}
#[cfg(not(feature = "fs-store"))]
Err(runtime_api::RuntimeApiError::new(
503,
"worker_retention_unavailable",
"Worker retention requires filesystem storage",
))
}
}
crates/worker-runtime/src/resource.rs
+2
View File
@@ -1,6 +1,8 @@
#[cfg(feature = "http-server")]
use crate::auth::BACKEND_RESOURCE_FETCH_PERMISSION;
use crate::identity::WorkerId;
use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex};
#[cfg(feature = "http-server")]
use crate::workspace_request::RuntimeWorkspaceRequestClient;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
crates/worker-runtime/src/retention/runtime_v5_tests.rs
+19 -19
View File
@@ -3,7 +3,7 @@
//! only the empty segment envelope is built with current APIs.
use super::tests::{request, write_json};
use super::*;
use crate::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions};
use crate::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions, SCHEMA_VERSION};
use serde_json::{Value, json};
use session_store::{LogEntry, SegmentId, SessionId, Store, WorkerSessionStore};
use std::collections::BTreeMap;
@@ -226,7 +226,7 @@ fn completed_v4_archive(
#[test]
fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
for runtime_schema in [11, 12] {
for runtime_schema in [11, 12, SCHEMA_VERSION] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, session, segment) = runtime_fixture(&root, runtime_schema);
@@ -329,7 +329,7 @@ fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
(storage.checksum_sha256, storage.bytes, storage.file_count),
(checksum, bytes, count)
);
if runtime_schema == 12 {
if runtime_schema == SCHEMA_VERSION {
for name in [
PathBuf::from("runtime.json"),
PathBuf::from(format!("workers/{worker}/worker.json")),
@@ -346,8 +346,8 @@ fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
);
drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap());
assert_eq!(tree_bytes(&root), once, "retry rewrote migrated evidence");
assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists());
assert!(!temp.path().join(".runtime.schema-v13-staging").exists());
assert!(!temp.path().join(".runtime.pre-schema-v13-backup").exists());
// session-storage, not the immutable receipt's old checksum, attests
// current bytes. Tampering with those bytes must remain detectable.
let path = archive_session
@@ -360,7 +360,7 @@ fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() {
#[test]
fn corrupt_archive_does_not_block_startup_or_healthy_session_migration_and_repair_converges() {
for runtime_schema in [11, 12] {
for runtime_schema in [11, 12, SCHEMA_VERSION] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, session, segment) = runtime_fixture(&root, runtime_schema);
@@ -440,8 +440,8 @@ fn unmigratable_archive_preserves_evidence_across_startup_and_reopen() {
] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (live_worker, live_session, live_segment) = runtime_fixture(&root, 12);
let (worker, _, segment) = stopped_worker(&root, 12);
let (live_worker, live_session, live_segment) = runtime_fixture(&root, SCHEMA_VERSION);
let (worker, _, segment) = stopped_worker(&root, SCHEMA_VERSION);
let provider = FsWorkerRetentionProvider::new(&root);
let path = provider
.worker_dir(worker)
@@ -542,8 +542,8 @@ fn unreadable_archive_metadata_does_not_block_archive_only_runtime_startup() {
] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, _, _) = runtime_fixture(&root, 12);
completed_v4_archive(&root, 12);
let (worker, _, _) = runtime_fixture(&root, SCHEMA_VERSION);
completed_v4_archive(&root, SCHEMA_VERSION);
let provider = FsWorkerRetentionProvider::new(&root);
fs::remove_dir_all(provider.worker_dir(worker)).unwrap();
let archive = provider.archive_dir("archive-a").unwrap();
@@ -575,8 +575,8 @@ fn unreadable_archive_metadata_does_not_block_archive_only_runtime_startup() {
fn archive_isolation_does_not_bypass_corrupt_erasure_authority() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
runtime_fixture(&root, 12);
let (session, _, _, _) = completed_v4_archive(&root, 12);
runtime_fixture(&root, SCHEMA_VERSION);
let (session, _, _, _) = completed_v4_archive(&root, SCHEMA_VERSION);
let ledger = root.join("retention/session-sources");
fs::create_dir_all(&ledger).unwrap();
fs::write(ledger.join(format!("{session}.json")), b"{broken\n").unwrap();
@@ -591,7 +591,7 @@ fn archive_isolation_does_not_bypass_corrupt_erasure_authority() {
fn runtime_v5_probe_and_staging_skip_erased_originals() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, live_session, live_segment) = runtime_fixture(&root, 12);
let (worker, live_session, live_segment) = runtime_fixture(&root, SCHEMA_VERSION);
let erased = root
.join("orphaned-workers")
.join(WorkerId::now_v7().to_string());
@@ -643,7 +643,7 @@ fn runtime_v5_rejects_broken_or_future_source_without_rewriting_evidence() {
for corruption in ["broken-json", "future-schema"] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, session, segment) = runtime_fixture(&root, 12);
let (worker, session, segment) = runtime_fixture(&root, SCHEMA_VERSION);
let session_root = root
.join("workers")
.join(worker.to_string())
@@ -674,8 +674,8 @@ fn runtime_v5_rejects_broken_or_future_source_without_rewriting_evidence() {
assert_eq!(tree_bytes(&root), before);
assert!(FsRuntimeStore::migrate(&options).is_err());
assert_eq!(tree_bytes(&root), before);
assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists());
assert!(!temp.path().join(".runtime.schema-v13-staging").exists());
assert!(!temp.path().join(".runtime.pre-schema-v13-backup").exists());
}
}
@@ -686,7 +686,7 @@ fn current_runtime_session_migration_still_rejects_live_worker_sockets() {
// or alias is needed, and the real live-socket migration refusal stays tested.
let temp = tempfile::tempdir_in(".").unwrap();
let root = Path::new(temp.path().file_name().unwrap()).join("runtime");
let (worker, _, _) = runtime_fixture(&root, 12);
let (worker, _, _) = runtime_fixture(&root, SCHEMA_VERSION);
let before = tree_bytes(&root);
let runs = root.join("workers").join(worker.to_string()).join("runs");
fs::create_dir_all(runs.join("1")).unwrap();
@@ -697,7 +697,7 @@ fn current_runtime_session_migration_still_rejects_live_worker_sockets() {
drop(listener);
fs::remove_dir_all(runs).unwrap();
assert_eq!(tree_bytes(&root), before);
assert!(!temp.path().join(".runtime.schema-v12-staging").exists());
assert!(!temp.path().join(".runtime.schema-v13-staging").exists());
}
#[test]
@@ -705,7 +705,7 @@ fn current_runtime_upgrades_v5_role_metadata_through_staging_without_reserializi
{
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("runtime");
let (worker, session, segment) = runtime_fixture(&root, 12);
let (worker, session, segment) = runtime_fixture(&root, SCHEMA_VERSION);
let aggregate = root.join("workers").join(worker.to_string());
let path = aggregate
.join("session/segments")
crates/worker-runtime/src/runtime.rs
+519 -147
View File
@@ -2105,6 +2105,16 @@ impl Runtime {
// Authorization is read-only until token admission succeeds.
state.ensure_workspace_owner(scope, false)?;
}
if let Err(error) = self.lock()?.refresh_uncertain_worker_write(worker_ref) {
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_canonical_write_unknown",
sanitize_worker_create_failure_message(&error.to_string()),
)),
None,
));
}
if request.request_id.trim().is_empty()
|| request.request_id.len() > 512
|| request.expected_observation_token.is_empty()
@@ -2219,6 +2229,7 @@ impl Runtime {
.unwrap_or_else(|| PendingWorkerRestore {
operation_id: WorkerLifecycleOperationId::new(),
mode: WorkerRestoreMode::Explicit,
commit: RestoreCommitFact::NotCommitted,
last_settled_status: candidate.status,
});
let operation_id = pending.operation_id;
@@ -2242,10 +2253,9 @@ impl Runtime {
receipt: None,
},
);
match state.persist_worker_record(&candidate) {
match state.persist_restore_worker_record(&candidate) {
Ok(()) => {}
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
state.workers.insert(worker_ref.worker_id, candidate);
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
@@ -2389,10 +2399,13 @@ impl Runtime {
if candidate.restore_guard.active_request_id.is_some()
|| candidate.pending_restore.is_some()
{
if result.state != WorkerRestoreState::Accepted {
candidate.ensure_restore_rollback_allowed()?;
}
candidate.restore_guard.finish(result.clone());
candidate.pending_restore = None;
candidate.restore_candidate_context = None;
state.persist_worker_record(&candidate)?;
state.persist_restore_worker_record(&candidate)?;
state.workers.insert(worker_ref.worker_id, candidate);
if result.state == WorkerRestoreState::Accepted {
state.publish_worker_upsert(worker_ref.worker_id)?;
@@ -2431,6 +2444,15 @@ impl Runtime {
));
}
let pending_restore = worker.pending_restore;
if pending_restore
.is_some_and(|pending| pending.commit == RestoreCommitFact::LegacyUnknown)
{
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_legacy_decision_unknown",
"The saved Restore lacks commit evidence; do not replay execution or infer rollback",
));
}
let reconcile = pending_restore.is_some()
&& worker
.restore_guard
@@ -2603,7 +2625,7 @@ impl Runtime {
workdir_attachments.clone(),
);
match commit {
Ok(worker) => {
Ok(_) => {
if let Err(error) =
backend.activate_restored_worker(operation_id, worker_ref)
{
@@ -2631,6 +2653,7 @@ impl Runtime {
sanitize_worker_create_failure_message(&error.to_string()),
));
}
let worker = self.lock()?.worker(worker_ref)?.detail();
let result = RuntimeWorkerRestoreResult::accepted(worker);
if let Err(error) = self.finish_restore_receipt(worker_ref, &result) {
return Ok(RuntimeWorkerRestoreResult::failed(
@@ -2642,10 +2665,38 @@ impl Runtime {
Ok(result)
}
Err(error) => {
let known_uncommitted =
!matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. })
&& self
.lock()?
.worker(worker_ref)?
.pending_restore
.is_some_and(|pending| {
pending.operation_id == operation_id
&& pending.commit == RestoreCommitFact::NotCommitted
});
if known_uncommitted {
if let WorkerExecutionSpawnResult::RolledBack(_) =
backend.discard_uncommitted_restore(operation_id, worker_ref)
{
self.discard_restore_candidate_context(
worker_ref,
&candidate_context,
)?;
let receipt = RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::RolledBack,
"worker_restore_decision_not_committed",
sanitize_worker_create_failure_message(&error.to_string()),
);
if self.settle_restore_rollback(worker_ref, &receipt).is_ok() {
return Ok(receipt);
}
}
}
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %error,
"Worker restore activated but commit did not confirm success; preserving operation and execution"
"Worker restore decision did not confirm success; preserving operation ownership"
);
if let Err(retain_error) = self.retain_restore_execution_evidence(
worker_ref,
@@ -3601,6 +3652,7 @@ impl Runtime {
self.ensure_worker_in_workspace(scope, worker_ref)?;
}
self.lock()?.refresh_uncertain_worker_write(worker_ref)?;
self.lock()?.ensure_subject_not_reserved(worker_ref, true)?;
let (backend, pending_stop, pending_restore, status) = {
let state = self.lock()?;
@@ -4355,18 +4407,18 @@ impl Runtime {
candidate.pending_restore = Some(PendingWorkerRestore {
operation_id,
mode,
commit: candidate
.pending_restore
.map_or(RestoreCommitFact::NotCommitted, |pending| pending.commit),
last_settled_status: candidate.status,
});
candidate.restore_candidate_context = Some(candidate_context);
match state.persist_worker_record(&candidate) {
match state.persist_restore_worker_record(&candidate) {
Ok(()) => {
state.workers.insert(worker_ref.worker_id, candidate);
Ok(())
}
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
state.workers.insert(worker_ref.worker_id, candidate);
Err(error)
}
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => Err(error),
Err(error) => Err(error),
}
}
@@ -4378,11 +4430,12 @@ impl Runtime {
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
candidate.ensure_restore_rollback_allowed()?;
candidate.restore_guard.rotate();
candidate.restore_guard.finish(receipt.clone());
candidate.pending_restore = None;
candidate.restore_candidate_context = None;
state.persist_worker_record(&candidate)?;
state.persist_restore_worker_record(&candidate)?;
state.workers.insert(worker_ref.worker_id, candidate);
Ok(())
}
@@ -4398,7 +4451,11 @@ impl Runtime {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let previous = state.worker(worker_ref)?;
if previous.pending_stop.is_some() {
if previous.pending_stop.is_some()
|| previous
.pending_restore
.is_none_or(|pending| pending.commit == RestoreCommitFact::LegacyUnknown)
{
return Err(RuntimeError::RestoreObservationConflict {
worker_id: worker_ref.worker_id,
});
@@ -4446,6 +4503,7 @@ impl Runtime {
}
let _ = candidate.apply_worker_state(&worker_state);
if let Some(pending) = candidate.pending_restore.as_mut() {
pending.commit = RestoreCommitFact::Committed;
pending.last_settled_status = candidate.status;
}
candidate.restore_intent = restore_intent_for_status(candidate.status);
@@ -4459,7 +4517,7 @@ impl Runtime {
// Persist the execution projection before publishing observations. Terminal
// receipt settlement follows successful observation activation separately.
state.persist_worker_record(&candidate)?;
state.persist_restore_worker_record(&candidate)?;
let detail = candidate.detail();
state.workers.insert(worker_ref.worker_id, candidate);
@@ -4499,28 +4557,15 @@ impl Runtime {
fn retain_restore_execution_evidence(
&self,
worker_ref: &WorkerRef,
_worker_state: protocol::WorkerStateSnapshot,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
_workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
candidate_context: crate::execution::WorkerExecutionContext,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
if candidate.pending_restore.is_none() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} has no pending restore operation",
worker_ref.worker_id
)));
}
candidate.restore_guard.rotate();
candidate.execution_metadata_available = true;
candidate.execution_bound = true;
candidate.restore_candidate_context = Some(candidate_context);
candidate.workdir_attachments = workdir_attachments;
state.persist_worker_record(&candidate)?;
state.workers.insert(worker_ref.worker_id, candidate);
Ok(())
// Keep the actual resource owner even if canonical authority is unreadable.
// Execution evidence cannot authorize overwriting a post-rename decision.
state.worker_mut(worker_ref)?.restore_candidate_context = Some(candidate_context);
state.refresh_uncertain_worker_write(worker_ref)
}
fn retain_create_reconciliation_pending(
@@ -5080,6 +5125,14 @@ impl Runtime {
)
})?;
let provider = FsWorkerRetentionProvider::new(store.runtime_dir());
if state
.unreconciled_worker_writes
.contains(&request.worker_id)
{
return Err(RuntimeError::InvalidRequest(
"Worker canonical write outcome must be reconciled before retention effects".into(),
));
}
if let Some(completed) = provider.completed_for(request)? {
state.workers.remove(&request.worker_id);
state.persist_runtime_snapshot()?;
@@ -5178,6 +5231,9 @@ struct RuntimeState {
#[cfg(feature = "fs-store")]
next_diagnostic_id: u64,
workers: BTreeMap<WorkerId, WorkerRecord>,
// 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_create_recovery_fences: BTreeMap<WorkerId, SubjectCreateRecoveryFence>,
subject_release_operations: BTreeMap<
(WorkerId, String),
@@ -5216,6 +5272,7 @@ impl RuntimeState {
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
unreconciled_worker_writes: Default::default(),
subject_create_recovery_fences: BTreeMap::new(),
subject_release_operations: BTreeMap::new(),
workdir_operations: BTreeMap::new(),
@@ -5249,6 +5306,7 @@ impl RuntimeState {
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
unreconciled_worker_writes: Default::default(),
subject_create_recovery_fences: BTreeMap::new(),
subject_release_operations: BTreeMap::new(),
workdir_operations: BTreeMap::new(),
@@ -5277,112 +5335,13 @@ impl RuntimeState {
let diagnostics = persisted.diagnostics;
let next_diagnostic_id = persisted.next_diagnostic_id;
for (worker_id, worker) in persisted.workers {
let (
request,
execution_metadata_available,
execution_bound,
restore_intent,
pending_restore,
pending_stop,
) = match worker.execution_state {
PersistedWorkerExecutionState::Available(execution) => (
Some(execution.request),
true,
execution.binding.is_some(),
execution.restore_intent,
None,
None,
),
PersistedWorkerExecutionState::ReconciliationRequired(operation) => match operation
{
PersistedWorkerLifecycleOperation::Restore(operation) => {
let restore_intent =
restore_intent_for_status(operation.last_settled_status);
(
Some(operation.request),
true,
operation.binding.is_some(),
restore_intent,
Some(PendingWorkerRestore {
operation_id: operation.operation_id,
mode: operation.mode.into(),
last_settled_status: operation.last_settled_status,
}),
None,
)
let mut record = WorkerRecord::from_persisted(worker);
// A prior active Controller is not live in this process. Keep
// stopped observations stable; canonical rereads never rotate them.
if record.status.is_active() && record.execution_bound {
record.restore_guard.rotate();
}
PersistedWorkerLifecycleOperation::Stop(operation) => (
operation.request.clone(),
operation.request.is_some(),
operation.binding.is_some(),
operation.restore_intent,
operation
.pending_restore
.map(|restore| PendingWorkerRestore {
operation_id: restore.operation_id,
mode: restore.mode.into(),
last_settled_status: restore.last_settled_status,
}),
Some(PendingWorkerStop {
operation_id: operation.operation_id,
last_settled_status: operation.last_settled_status,
}),
),
},
PersistedWorkerExecutionState::Unavailable => (
None,
false,
false,
restore_intent_for_status(worker.status),
None,
None,
),
};
let logical_workdir_attachments = if worker.logical_workdir_attachments.is_empty() {
let claims = request
.as_ref()
.map(|request| request.workdir_attachments.as_slice())
.unwrap_or_default();
logical_workdir_attachments_from_statuses(&worker.workdir_attachments, claims)
} else {
worker.logical_workdir_attachments
};
workers.insert(
worker_id,
WorkerRecord {
restore_guard: {
let mut guard = worker.restore_guard;
// A prior active controller is not live in this process.
// Stopped observations remain stable across restart.
if worker.status.is_active() && execution_bound {
guard.rotate();
}
guard
},
worker_ref: worker.worker_ref,
worker_id: worker.worker_id,
status: worker.status,
worker_state: None,
last_finished_submission_request_id: worker.last_finished_submission_request_id,
execution_reconstructed: true,
workspace_id: worker.workspace_id,
profile: worker.profile,
display_name: worker.display_name,
profile_source: worker.profile_source,
config_bundle: worker.config_bundle,
request,
created_at_ms: worker.created_at_ms,
execution_metadata_available,
execution_bound,
restore_intent,
pending_restore,
pending_stop,
restore_candidate_context: None,
workdir_attachments: worker.workdir_attachments,
logical_workdir_attachments,
internal_workers: InternalWorkerActivityProjection::default(),
},
);
workers.insert(worker_id, record);
}
Ok(Self {
@@ -5396,6 +5355,7 @@ impl RuntimeState {
workspace_backend_resource_clients: BTreeMap::new(),
next_diagnostic_id,
workers,
unreconciled_worker_writes: Default::default(),
subject_create_recovery_fences: persisted.subject_create_recovery_fences,
subject_release_operations: persisted
.subject_releases
@@ -5458,8 +5418,80 @@ impl RuntimeState {
Ok(())
}
/// An uncertain Restore write is resolved only from the locked canonical
/// aggregate and a successful durability barrier, never a second blind write.
fn persist_restore_worker_record(&mut self, worker: &WorkerRecord) -> Result<(), RuntimeError> {
match self.persist_worker_record(worker) {
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
self.unreconciled_worker_writes.insert(worker.worker_id);
#[cfg(feature = "fs-store")]
if let RuntimePersistence::Fs(store) = &self.persistence {
match store.reconcile_worker_record(&worker.persisted_record()) {
Ok((observed, matches)) => {
self.apply_canonical_worker_record(observed)?;
if matches {
return Ok(());
}
}
Err(reconcile_error) => {
tracing::warn!(worker_id = %worker.worker_id, error = %reconcile_error,
"Restore aggregate remains uncertain after canonical reconciliation");
}
}
}
Err(error)
}
outcome => outcome,
}
}
#[cfg(feature = "fs-store")]
fn apply_canonical_worker_record(
&mut self,
observed: PersistedWorkerRecord,
) -> Result<(), RuntimeError> {
let previous = self.worker(&observed.worker_ref)?;
let mut current = WorkerRecord::from_persisted(observed);
// Canonical authority replaces only the saved projection. A read is
// neither process reconstruction nor evidence that a Controller closed.
current.execution_reconstructed = previous.execution_reconstructed;
current.restore_candidate_context = previous.restore_candidate_context.clone();
current.internal_workers = previous.internal_workers.clone();
if current.status == previous.status {
current.worker_state = previous.worker_state.clone();
}
self.unreconciled_worker_writes.remove(&current.worker_id);
self.workers.insert(current.worker_id, current);
Ok(())
}
fn refresh_uncertain_worker_write(
&mut self,
worker_ref: &WorkerRef,
) -> Result<(), RuntimeError> {
if !self
.unreconciled_worker_writes
.contains(&worker_ref.worker_id)
{
return Ok(());
}
#[cfg(feature = "fs-store")]
if let RuntimePersistence::Fs(store) = &self.persistence {
let observed = store.read_durable_worker_record(worker_ref)?;
return self.apply_canonical_worker_record(observed);
}
Err(RuntimeError::InvalidRequest(
"Worker canonical write outcome remains unresolved".into(),
))
}
#[cfg(feature = "fs-store")]
fn persist_worker_record(&self, worker: &WorkerRecord) -> Result<(), RuntimeError> {
if self.unreconciled_worker_writes.contains(&worker.worker_id) {
return Err(RuntimeError::InvalidRequest(
"Worker canonical write outcome must be reconciled before another mutation".into(),
));
}
if let Some(store) = self.fs_store() {
store.write_worker_record(&worker.persisted_record())?;
}
@@ -5468,20 +5500,25 @@ impl RuntimeState {
#[cfg(feature = "fs-store")]
fn persist_worker(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
if self.fs_store().is_some() {
let worker =
self.workers
.get(worker_id)
.ok_or_else(|| RuntimeError::WorkerNotFound {
worker_id: *worker_id,
})?;
store.write_worker_record(&worker.persisted_record())?;
self.persist_worker_record(worker)?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn delete_worker_record(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if self.unreconciled_worker_writes.contains(worker_id) {
return Err(RuntimeError::InvalidRequest(
"Worker canonical write outcome must be reconciled before removal".into(),
));
}
if let Some(store) = self.fs_store() {
store.delete_worker_record(worker_id)?;
}
@@ -5950,6 +5987,7 @@ impl RuntimeState {
.and_then(|worker| worker.workspace_id.clone())
.unwrap_or_else(|| "<unscoped>".to_string());
let mut candidate = self.worker(worker_ref)?.clone();
candidate.ensure_restore_rollback_allowed()?;
candidate.execution_bound = false;
candidate.restore_guard.finish(receipt.clone());
@@ -6274,10 +6312,21 @@ impl InternalWorkerActivityProjection {
}
}
/// A durable decision is separate from both admission and a live Controller.
/// Legacy pending operations lack sufficient evidence to choose either branch.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RestoreCommitFact {
NotCommitted,
Committed,
LegacyUnknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PendingWorkerRestore {
operation_id: WorkerLifecycleOperationId,
mode: WorkerRestoreMode,
commit: RestoreCommitFact,
last_settled_status: WorkerStatus,
}
@@ -6410,6 +6459,17 @@ struct WorkerRecord {
}
impl WorkerRecord {
fn ensure_restore_rollback_allowed(&self) -> Result<(), RuntimeError> {
if self
.pending_restore
.is_some_and(|pending| pending.commit != RestoreCommitFact::NotCommitted)
{
return Err(RuntimeError::InvalidRequest(
"A committed or legacy-unknown Restore cannot be settled as a rollback".into(),
));
}
Ok(())
}
fn apply_logical_workdir_attachments(
&mut self,
mut attachments: Vec<LogicalWorkdirAttachment>,
@@ -6592,6 +6652,105 @@ impl WorkerRecord {
}
}
#[cfg(feature = "fs-store")]
fn from_persisted(worker: PersistedWorkerRecord) -> Self {
let (
request,
execution_metadata_available,
execution_bound,
restore_intent,
pending_restore,
pending_stop,
) = match worker.execution_state {
PersistedWorkerExecutionState::Available(execution) => (
Some(execution.request),
true,
execution.binding.is_some(),
execution.restore_intent,
None,
None,
),
PersistedWorkerExecutionState::ReconciliationRequired(operation) => match operation {
PersistedWorkerLifecycleOperation::Restore(operation) => {
let restore_intent = restore_intent_for_status(operation.last_settled_status);
(
Some(operation.request),
true,
operation.binding.is_some(),
restore_intent,
Some(PendingWorkerRestore {
operation_id: operation.operation_id,
mode: operation.mode.into(),
commit: operation.commit,
last_settled_status: operation.last_settled_status,
}),
None,
)
}
PersistedWorkerLifecycleOperation::Stop(operation) => (
operation.request.clone(),
operation.request.is_some(),
operation.binding.is_some(),
operation.restore_intent,
operation
.pending_restore
.map(|restore| PendingWorkerRestore {
operation_id: restore.operation_id,
mode: restore.mode.into(),
commit: restore.commit,
last_settled_status: restore.last_settled_status,
}),
Some(PendingWorkerStop {
operation_id: operation.operation_id,
last_settled_status: operation.last_settled_status,
}),
),
},
PersistedWorkerExecutionState::Unavailable => (
None,
false,
false,
restore_intent_for_status(worker.status),
None,
None,
),
};
let logical_workdir_attachments = if worker.logical_workdir_attachments.is_empty() {
let claims = request
.as_ref()
.map(|request| request.workdir_attachments.as_slice())
.unwrap_or_default();
logical_workdir_attachments_from_statuses(&worker.workdir_attachments, claims)
} else {
worker.logical_workdir_attachments
};
Self {
restore_guard: worker.restore_guard,
worker_ref: worker.worker_ref,
worker_id: worker.worker_id,
status: worker.status,
worker_state: None,
last_finished_submission_request_id: worker.last_finished_submission_request_id,
execution_reconstructed: true,
workspace_id: worker.workspace_id,
profile: worker.profile,
display_name: worker.display_name,
profile_source: worker.profile_source,
config_bundle: worker.config_bundle,
request,
created_at_ms: worker.created_at_ms,
execution_metadata_available,
execution_bound,
restore_intent,
pending_restore,
pending_stop,
restore_candidate_context: None,
workdir_attachments: worker.workdir_attachments,
logical_workdir_attachments,
internal_workers: InternalWorkerActivityProjection::default(),
}
}
#[cfg(feature = "fs-store")]
fn persisted_record(&self) -> PersistedWorkerRecord {
let execution_state = if let Some(pending) = self.pending_stop {
@@ -6602,6 +6761,7 @@ impl WorkerRecord {
PersistedWorkerRestoreOperation {
operation_id: restore.operation_id,
mode: restore.mode.into(),
commit: restore.commit,
request: self
.request
.clone()
@@ -6625,6 +6785,7 @@ impl WorkerRecord {
PersistedWorkerLifecycleOperation::Restore(PersistedWorkerRestoreOperation {
operation_id: pending.operation_id,
mode: pending.mode.into(),
commit: pending.commit,
request,
binding: self
.execution_bound
@@ -8994,6 +9155,7 @@ mod tests {
.pending_restore = Some(PendingWorkerRestore {
operation_id: WorkerLifecycleOperationId::new(),
mode: WorkerRestoreMode::Explicit,
commit: RestoreCommitFact::NotCommitted,
last_settled_status: WorkerStatus::Stopped,
});
let id = "workdir-irrelevant-to-recovery";
@@ -10933,6 +11095,7 @@ mod tests {
})).unwrap());
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
runtime_store(&runtime).fail_next_worker_reconcile_sync();
assert_eq!(
runtime
.restore_worker_operation(&worker.worker_ref, request.clone())
@@ -11031,6 +11194,7 @@ mod tests {
.operation_id;
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
runtime_store(&runtime).fail_next_worker_reconcile_sync();
request.preparation = None;
assert_eq!(
runtime
@@ -11221,6 +11385,7 @@ mod tests {
);
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
runtime_store(&runtime).fail_next_worker_reconcile_sync();
assert_eq!(
runtime
.restore_worker_operation(&worker.worker_ref, request.clone())
@@ -11228,9 +11393,21 @@ mod tests {
.state,
WorkerRestoreState::ReconciliationRequired
);
// Resolve admission authority before injecting a different write fault.
// Otherwise the next failed sync would only exercise admission reread.
{
let lease = runtime.worker_operation_lock(worker.worker_id).unwrap();
let _guard = lease.lock().unwrap();
runtime
.lock()
.unwrap()
.refresh_uncertain_worker_write(&worker.worker_ref)
.unwrap();
}
// This write interrupts idempotent attachment completion after admission.
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
runtime_store(&runtime).fail_next_worker_reconcile_sync();
assert_eq!(
runtime
.restore_worker_operation(&worker.worker_ref, request.clone())
@@ -11873,6 +12050,9 @@ mod tests {
.unwrap()
.operation_id;
runtime_store(&runtime).fail_next_worker_write(fault);
if matches!(fault, crate::fs_store::AtomicWriteFault::AfterRename) {
runtime_store(&runtime).fail_next_worker_reconcile_sync();
}
gate.release();
assert_eq!(
restoring.join().unwrap().unwrap().state,
@@ -13137,6 +13317,7 @@ mod tests {
assert_eq!(summary.worker_count, 0);
}
#[cfg(feature = "fs-store")]
#[test]
fn delete_worker_persistence_failure_retains_identity_and_bounded_diagnostic() {
let root = tempfile::tempdir().unwrap();
@@ -13236,6 +13417,7 @@ mod tests {
));
}
#[cfg(feature = "fs-store")]
#[test]
fn delete_worker_waits_for_execution_shutdown_before_removing_snapshot() {
let root = tempfile::tempdir().unwrap();
@@ -13406,7 +13588,7 @@ mod tests {
})
.unwrap_err();
assert!(error.to_string().contains(
"unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, or 12"
"unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, or 13"
));
let _ = std::fs::remove_dir_all(root);
@@ -13448,7 +13630,10 @@ 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!(12));
assert_eq!(
worker_aggregate["schema_version"],
serde_json::json!(crate::fs_store::SCHEMA_VERSION)
);
assert_eq!(worker_aggregate["status"], serde_json::json!("stopped"));
assert_eq!(
worker_aggregate["execution_state"]["state"],
@@ -14802,7 +14987,10 @@ 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!(12));
assert_eq!(
migrated_aggregate["schema_version"],
serde_json::json!(crate::fs_store::SCHEMA_VERSION)
);
assert!(migrated_aggregate.get("request").is_none());
assert!(migrated_aggregate.get("execution").is_none());
assert_eq!(
@@ -14826,6 +15014,184 @@ mod tests {
let _ = std::fs::remove_dir_all(root);
}
#[cfg(feature = "fs-store")]
#[test]
fn restore_decision_reread_preserves_commit_across_uncertainty_stop_and_restart() {
// Component scope: real Runtime aggregate owner/store, substituted
// execution backend. This does not attest Controller readiness or input.
for sync_failure in [false, true] {
let dir = tempfile::tempdir().unwrap();
let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
.with_runtime_id("decision-reconciliation");
let backend = Arc::new(TestExecutionBackend::default());
let runtime =
Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime
.create_worker(task_request("decision fact"))
.unwrap();
runtime.stop_worker(&worker.worker_ref, None).unwrap();
let lease = runtime.worker_operation_lock(worker.worker_id).unwrap();
let guard = lease.lock().unwrap();
let operation_id = WorkerLifecycleOperationId::new();
runtime
.begin_restore_operation(
&worker.worker_ref,
operation_id,
WorkerRestoreMode::Explicit,
runtime.restore_candidate_execution_context(worker.worker_ref.clone()),
)
.unwrap();
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
if sync_failure {
runtime_store(&runtime).fail_next_worker_reconcile_sync();
}
let result = runtime.commit_restored_worker_execution(
&worker.worker_ref,
protocol::WorkerStatus::Idle.into(),
WorkerStatus::Idle,
Vec::new(),
);
assert_eq!(result.is_ok(), !sync_failure);
let path = options
.root
.join("workers")
.join(worker.worker_id.to_string())
.join("worker.json");
let bytes = std::fs::read(&path).unwrap();
let observed = runtime_store(&runtime)
.read_worker_record(&worker.worker_ref)
.unwrap();
let PersistedWorkerExecutionState::ReconciliationRequired(
PersistedWorkerLifecycleOperation::Restore(pending),
) = observed.execution_state
else {
panic!("committed decision lost its operation")
};
assert_eq!(pending.operation_id, operation_id);
assert_eq!(pending.commit, RestoreCommitFact::Committed);
assert_eq!(
pending.request,
runtime
.lock()
.unwrap()
.worker(&worker.worker_ref)
.unwrap()
.request
.clone()
.unwrap()
);
let rollback = RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::RolledBack,
"must-not-revert",
"committed decision cannot become rollback",
);
assert!(
runtime
.settle_restore_rollback(&worker.worker_ref, &rollback)
.is_err()
);
assert_eq!(std::fs::read(&path).unwrap(), bytes);
// Cleanup confirmation from a backend is not evidence that the
// Runtime's durable decision was uncommitted.
assert!(
runtime
.lock()
.unwrap()
.record_restore_failure(
&worker.worker_ref,
&WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
"backend confirmed cleanup after reconstruction failure",
),
&rollback,
)
.is_err()
);
assert_eq!(std::fs::read(&path).unwrap(), bytes);
if sync_failure {
assert!(
runtime
.lock()
.unwrap()
.unreconciled_worker_writes
.contains(&worker.worker_id)
);
// Substitute unreadable storage only. No production recovery
// path repairs it from an old memory record.
std::fs::write(&path, b"not-json").unwrap();
assert!(
runtime
.lock()
.unwrap()
.refresh_uncertain_worker_write(&worker.worker_ref)
.is_err()
);
assert!(
runtime
.lock()
.unwrap()
.persist_worker(&worker.worker_id)
.is_err()
);
assert!(
runtime
.begin_stop_operation(&worker.worker_ref, WorkerLifecycleOperationId::new())
.is_err()
);
assert_eq!(std::fs::read(&path).unwrap(), b"not-json");
assert_eq!(*backend.stop_count.lock().unwrap(), 1);
// End the storage fault with the exact post-rename bytes, not
// an inferred projection; recovery still performs its real sync.
std::fs::write(&path, &bytes).unwrap();
runtime
.lock()
.unwrap()
.refresh_uncertain_worker_write(&worker.worker_ref)
.unwrap();
}
let current = runtime
.lock()
.unwrap()
.worker(&worker.worker_ref)
.unwrap()
.pending_restore
.unwrap();
assert_eq!(current.commit, RestoreCommitFact::Committed);
assert_eq!(current.operation_id, operation_id);
runtime
.begin_stop_operation(&worker.worker_ref, WorkerLifecycleOperationId::new())
.unwrap();
let saved = runtime_store(&runtime)
.read_worker_record(&worker.worker_ref)
.unwrap();
let PersistedWorkerExecutionState::ReconciliationRequired(
PersistedWorkerLifecycleOperation::Stop(stop),
) = saved.execution_state
else {
panic!("Stop ownership was not saved")
};
assert_eq!(
stop.pending_restore.unwrap().commit,
RestoreCommitFact::Committed
);
drop(guard);
drop(lease);
drop(runtime);
let reopened = Runtime::with_fs_store(options).unwrap();
let state = reopened.lock().unwrap();
let restored = state.worker(&worker.worker_ref).unwrap();
assert!(restored.pending_stop.is_some());
assert_eq!(
restored.pending_restore.unwrap().commit,
RestoreCommitFact::Committed
);
assert_eq!(restored.pending_restore.unwrap().operation_id, operation_id);
}
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_preserves_an_unknown_restore_operation_commit() {
@@ -14849,9 +15215,11 @@ mod tests {
runtime_store(&runtime)
.fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
runtime_store(&runtime).fail_next_worker_reconcile_sync();
let operation_id = WorkerLifecycleOperationId::new();
let result = runtime.begin_restore_operation(
&worker.worker_ref,
WorkerLifecycleOperationId::new(),
operation_id,
WorkerRestoreMode::Explicit,
runtime.restore_candidate_execution_context(worker.worker_ref.clone()),
);
@@ -14866,14 +15234,18 @@ mod tests {
runtime.worker_detail(&worker.worker_ref).unwrap().status,
WorkerStatus::Stopped
);
let operation_id = runtime
.lock()
.unwrap()
{
let state = runtime.lock().unwrap();
assert!(state.unreconciled_worker_writes.contains(&worker.worker_id));
assert!(
state
.worker(&worker.worker_ref)
.unwrap()
.pending_restore
.unwrap()
.operation_id;
.is_none(),
"failed sync must not promote an unconfirmed in-memory candidate"
);
}
let aggregate: serde_json::Value = serde_json::from_slice(
&std::fs::read(
root.join("workers")
crates/worker-runtime/src/test_support.rs
+10 -1
View File
@@ -376,6 +376,8 @@ impl StoppedWorkerTemplate {
root: stage.clone(),
_owner_lock: Some(lock),
worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)),
#[cfg(test)]
worker_reconcile_sync_fault: Default::default(),
};
let result = (|| {
let worker = PersistedWorkerRecord {
@@ -463,8 +465,15 @@ impl StoppedWorkerTemplate {
}
fn validate_shape(&self) -> Result<(), RuntimeError> {
// Version 1 templates carry only settled Session material, never a
// lifecycle operation. Schema 12 -> 13 changes pending decision evidence,
// not that payload. Accept the fixed schema-12 artifact read-only and
// install through the current aggregate owner; never rewrite its bytes.
if self.template_version != TEMPLATE_VERSION
|| self.runtime_schema_version != SCHEMA_VERSION
|| !matches!(
self.runtime_schema_version,
RESTORE_ADMISSION_SCHEMA_VERSION | SCHEMA_VERSION
)
{
return Err(reject("unsupported template schema"));
}
crates/worker-runtime/src/test_support_tests.rs
+33
View File
@@ -563,6 +563,7 @@ fn extraction_rejects_nonstopped_pending_schema_and_metadata_mismatch() {
PersistedWorkerLifecycleOperation::Restore(PersistedWorkerRestoreOperation {
operation_id: WorkerLifecycleOperationId::new(),
mode: PersistedWorkerRestoreMode::Explicit,
commit: crate::runtime::RestoreCommitFact::NotCommitted,
request: request(),
binding: None,
last_settled_status: WorkerStatus::Stopped,
@@ -632,6 +633,38 @@ fn extraction_rejects_nonstopped_pending_schema_and_metadata_mismatch() {
drop(held);
}
#[test]
fn schema_v12_settled_template_installs_current_aggregate_without_rewriting_session() {
let (_source, _, mut template) = source();
template.runtime_schema_version = RESTORE_ADMISSION_SCHEMA_VERSION;
let bytes = serde_json::to_vec(&template).unwrap();
let (_target, options) = destination(&template);
template.install(&options, request(), &resolved()).unwrap();
assert_eq!(serde_json::to_vec(&template).unwrap(), bytes);
let worker = options.root.join(WORKERS_DIR).join(worker_id().to_string());
assert_eq!(
fs::read(worker.join("session/session.json")).unwrap(),
template.session_manifest
);
assert_eq!(
fs::read(worker.join(format!("session/segments/{}.jsonl", template.segment_id))).unwrap(),
template.segment_jsonl
);
let aggregate: WorkerAggregateRecord =
read_bounded_json(&worker.join(WORKER_FILE), "read installed fixture").unwrap();
let saved = aggregate.validate(&worker.join(WORKER_FILE)).unwrap();
assert_eq!(saved.status, WorkerStatus::Stopped);
assert!(matches!(
saved.execution_state,
PersistedWorkerExecutionState::Available(_)
));
// Unsupported template schemas do not enter the versioned reader.
template.runtime_schema_version = RESTORE_ADMISSION_SCHEMA_VERSION - 1;
assert!(template.validate_shape().is_err());
template.runtime_schema_version = SCHEMA_VERSION + 1;
assert!(template.validate_shape().is_err());
}
#[test]
fn fresh_manifest_and_memory_settings_are_required() {
let (_source, _, template) = source();
crates/worker-runtime/src/worker_backend.rs
+665 -174
View File
File diff suppressed because it is too large Load Diff
crates/worker-runtime/src/worker_backend/restore_recovery.rs
+9 -1
View File
@@ -47,18 +47,26 @@ impl ProfileRuntimeWorkerFactory {
let workspace = RuntimeWorkspaceBackendRef::from_worker_request(
&request.request,
self.runtime_id.as_deref(),
);
)
.map_err(cause)?;
#[cfg(feature = "http-server")]
let request_client = request
.request
.workspace_api
.as_ref()
.and_then(|api| self.workspace_request_clients.get(&api.workspace_id));
let context = workspace.worker_context(
#[cfg(feature = "http-server")]
&request.worker_ref,
#[cfg(feature = "http-server")]
request.workspace_scope.as_ref(),
#[cfg(feature = "http-server")]
self.worker_mutation_identity.as_ref(),
#[cfg(feature = "http-server")]
request_client,
#[cfg(feature = "http-server")]
self.embedded_worker_mutation_dispatcher.as_ref(),
#[cfg(feature = "http-server")]
Some(self.prompt_projection_cache.clone()),
);
if pending && context.workspace_id().is_some() {
crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs
+634 -13
View File
@@ -165,8 +165,19 @@ mod restore_candidate {
}
fn with_transport(transport: WorkerControllerTransport) -> Self {
Self::with_seed(transport, false)
}
fn no_model() -> Self {
Self::with_seed(WorkerControllerTransport::InProcess, true)
}
fn with_seed(transport: WorkerControllerTransport, no_model: bool) -> Self {
let root = restore_failure_tempdir();
let model = MockClient::sequential(vec![
let model = if no_model {
MockClient::new(Vec::new())
} else {
MockClient::sequential(vec![
MockResponse::Complete(vec![
LlmEvent::tool_use_start(0, "candidate-task-probe", "TaskCreate"),
LlmEvent::tool_input_delta(
@@ -181,7 +192,8 @@ mod restore_candidate {
MockResponse::Complete(simple_text_events()),
MockResponse::Complete(simple_text_events()),
MockResponse::Complete(simple_text_events()),
]);
])
};
let injected = model.clone();
let mut factory = ProfileRuntimeWorkerFactory::new(root.path())
.with_runtime_store_dir(root.path())
@@ -264,15 +276,19 @@ mod restore_candidate {
let aggregate = fixture_root.join("workers").join(worker_ref.worker_id.to_string());
let session = WorkerSessionStore::new(aggregate.join("session")).unwrap();
let metadata = WorkerAggregateStore::new(&aggregate, &manifest.worker.name).unwrap();
let mut engine = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
MockClient::sequential(vec![MockResponse::Hang(simple_text_events().into_iter().take(2).collect())]),
);
let seed_model = if no_model {
MockClient::new(Vec::new())
} else {
MockClient::sequential(vec![MockResponse::Hang(simple_text_events().into_iter().take(2).collect())])
};
let mut engine = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(seed_model);
engine.set_system_prompt("Committed restore candidate fixture prompt");
let mut worker = Worker::new(
manifest.clone(),
engine,
CombinedStore::new(session.clone(), metadata.clone()),
RuntimeWorkspaceBackendRef::from_worker_request(&create, Some("candidate-runtime"))
.unwrap()
.worker_context(&worker_ref, Some(&crate::runtime::RuntimeWorkspaceScope::new(
"workspace-saved-subjektiv", "server-main",
)), None, None, None, None),
@@ -282,6 +298,12 @@ mod restore_candidate {
worker.enable_worker_metadata_write_through().unwrap();
worker.materialize_durable_session_head().await.unwrap();
let (session_id, segment_id) = (worker.session_id(), worker.segment_id());
let receipts = if no_model {
// Persist a real empty Session without starting a seed
// Controller or submitting any model input.
drop(worker);
Vec::new()
} else {
let workspace_client = worker.workspace_client_handle();
let started = PreparedWorker::new(
worker,
@@ -293,6 +315,7 @@ mod restore_candidate {
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
controller_task: started.controller_task,
workspace_client,
restore_streams: None,
}));
let handle = seed.0.as_ref().unwrap().handle.clone();
let mut started_events = handle.subscribe();
@@ -327,6 +350,8 @@ mod restore_candidate {
}
assert_eq!(pending(&handle).await.submissions.len(), 2);
shutdown_profile_controller(seed.0.take().unwrap()).await;
receipts
};
// This orphan pin would be removed by the normal activation
// reconciliation. Preparation and discard must leave it alone.
let orphan = session.write_uploaded_file(
@@ -358,6 +383,7 @@ mod restore_candidate {
host,
_root: root,
};
if !no_model {
let checkpoint = queue_checkpoint(&fixture.entries());
assert_eq!(checkpoint["pending"].as_array().unwrap().len(), 2);
assert_eq!(
@@ -367,6 +393,7 @@ mod restore_candidate {
.len(),
1
);
}
fixture.assert_quiet();
fixture
}
@@ -443,14 +470,23 @@ mod restore_candidate {
.get(&self.request.worker_ref)
.is_none()
);
assert!(
!self
.backend
.factory
.worker_restore_run_dir(&self.request.worker_ref, self.request.operation_id,)
.unwrap()
.exists()
);
let run_dir = self.backend.factory.worker_restore_run_dir(
&self.request.worker_ref, self.request.operation_id,
).unwrap();
// Private initialization owns only disposable operation-scoped run
// artifacts. It has not published a listener, liveness files or a
// delegated-child projection, and canonical files are checked by
// each caller independently.
let mut artifacts = std::fs::read_dir(&run_dir).unwrap()
.map(|entry| entry.unwrap().file_name().into_string().unwrap()).collect::<Vec<_>>();
artifacts.sort();
assert_eq!(artifacts, ["artifacts", "spawned", "worker.err.log", "worker.out.log"]);
for log in ["worker.err.log", "worker.out.log"] {
assert_eq!(std::fs::metadata(run_dir.join(log)).unwrap().len(), 0);
}
for directory in ["artifacts", "spawned"] {
assert_eq!(std::fs::read_dir(run_dir.join(directory)).unwrap().count(), 0);
}
let table = worker::runtime::worker_allocation::LockFileGuard::open(
&worker::runtime::worker_allocation::default_allocation_path().unwrap(),
)
@@ -521,6 +557,588 @@ mod restore_candidate {
}
}
#[test]
#[serial_test::serial(worker_allocation)]
fn session_availability_requires_attachable_current_controller() {
let fixture = Fixture::no_model();
let worker_ref = &fixture.request.worker_ref;
let observe = || {
fixture
.backend
.worker_session(WorkerSessionObservationRequest {
worker_ref: worker_ref.clone(),
})
};
let assert_retained = |availability| {
assert!(matches!(
availability,
runtime_api::WorkerSessionAvailability::RetainedSnapshot { identity, .. }
if identity.session_id == fixture.session_id.to_string()
&& identity.segment_id == fixture.segment_id.to_string()
));
};
fixture.prepare();
assert_retained(observe());
let pending = fixture
.backend
.pending_restore(worker_ref)
.unwrap()
.unwrap();
fixture
.backend
.activate_restored_worker(fixture.request.operation_id, worker_ref)
.unwrap();
let execution = fixture.backend.workers.lock().unwrap()[worker_ref].clone();
assert!(!execution.handle.protocol_is_closed());
assert!(!execution.shutdown_requested.load(Ordering::Acquire));
assert!(matches!(
observe(),
runtime_api::WorkerSessionAvailability::LiveProtocol
));
#[cfg(feature = "ws-server")]
let transport = fixture
.backend
.clone()
.attach_worker_protocol(worker_ref)
.unwrap();
#[cfg(feature = "ws-server")]
transport.validate().unwrap();
// Retain the same real operation owner alongside its registered actual
// Controller to exercise the pending-resource visibility guard alone.
let operation_lock = fixture.backend.worker_lock(worker_ref).unwrap();
{
let _guard = operation_lock.lock().unwrap();
fixture
.backend
.pending_restores
.lock()
.unwrap()
.insert(worker_ref.clone(), pending);
}
let pending_session = observe();
#[cfg(feature = "ws-server")]
let pending_protocol = (
transport.validate(),
fixture.backend.clone().attach_worker_protocol(worker_ref),
);
// Reset injected visibility before assertions so Fixture can also stop
// and release its allocation when an assertion unwinds.
{
let _guard = operation_lock.lock().unwrap();
fixture.backend.remove_pending_restore(worker_ref).unwrap();
}
assert_retained(pending_session);
#[cfg(feature = "ws-server")]
{
assert!(pending_protocol.0.is_err());
assert!(matches!(
pending_protocol.1,
Err(error) if error.outcome == crate::execution::WorkerExecutionOutcome::Busy
));
}
assert!(matches!(
observe(),
runtime_api::WorkerSessionAvailability::LiveProtocol
));
{
let _guard = operation_lock.lock().unwrap();
execution.shutdown_requested.store(true, Ordering::Release);
}
let stopping_session = observe();
#[cfg(feature = "ws-server")]
let stopping_protocol = (
transport.validate(),
fixture.backend.clone().attach_worker_protocol(worker_ref),
);
{
let _guard = operation_lock.lock().unwrap();
execution.shutdown_requested.store(false, Ordering::Release);
}
assert!(!execution.handle.protocol_is_closed());
assert_retained(stopping_session);
#[cfg(feature = "ws-server")]
{
assert!(stopping_protocol.0.is_err());
assert!(stopping_protocol.1.is_err());
transport.validate().unwrap();
}
assert!(matches!(
observe(),
runtime_api::WorkerSessionAvailability::LiveProtocol
));
let handle = execution.handle.clone();
let command = next_internal_command(&execution.worker_state).unwrap();
fixture
.backend
.run_on_adapter_runtime(async move {
handle
.send(Method::Shutdown { command })
.await
.map_err(|error| error.to_string())?;
tokio::time::timeout(WAIT, handle.protocol_closed())
.await
.map_err(|_| "Controller did not close".to_string())?;
Ok(())
})
.unwrap();
assert!(execution.handle.protocol_is_closed());
assert!(!execution.shutdown_requested.load(Ordering::Acquire));
assert!(
fixture
.backend
.workers
.lock()
.unwrap()
.contains_key(worker_ref)
);
assert_retained(observe());
#[cfg(feature = "ws-server")]
{
assert!(transport.validate().is_err());
assert!(
fixture
.backend
.clone()
.attach_worker_protocol(worker_ref)
.is_err()
);
}
assert!(fixture.backend.stop_worker(worker_ref).is_accepted());
#[cfg(feature = "ws-server")]
drop(transport);
drop(execution);
fixture.assert_released();
fixture.assert_quiet();
}
// Connects Profile restore activation to ordinary protocol input and the
// durable Paused boundary; Runtime admission/Accepted ordering is outside it.
#[test]
#[cfg(feature = "ws-server")]
#[serial_test::serial(worker_allocation)]
fn activated_empty_candidate_accepts_body_input_and_queues_while_paused() {
use crate::execution::WorkerProtocolTransport;
const S1: &str = "candidate-body-s1";
const S2: &str = "candidate-body-s2";
const IDLE_ONLY: &str = "candidate-body-idle-only";
const OUTPUT: &str = "committed candidate output before pause";
// Keep one real connection (paired full snapshot + subscription) for
// every method and observation. Never reconnect to hide missing events.
fn observe(
backend: &WorkerRuntimeExecutionBackend,
mut transport: WorkerProtocolTransport,
phase: &'static str,
no_new_run: bool,
mut matches: impl FnMut(&Event) -> bool + Send + 'static,
) -> (WorkerProtocolTransport, Event) {
backend
.run_on_adapter_runtime(async move {
let mut seen = Vec::new();
let matched = tokio::time::timeout(WAIT, async {
loop {
let event = transport
.events
.recv()
.await
.expect("attached protocol stream");
seen.push(format!("{event:?}"));
if no_new_run {
assert!(
!matches!(
event,
Event::InvokeStart { .. }
| Event::TurnStart { .. }
| Event::UserMessage { .. }
| Event::TextDelta { .. }
| Event::TextDone { .. }
),
"paused input started another run during {phase}: {event:?}"
);
}
if matches(&event) {
break event;
}
assert!(
!matches!(
event,
Event::Error { .. }
| Event::SubmissionRejected { .. }
| Event::NotificationRejected { .. }
| Event::Shutdown
),
"unexpected protocol failure during {phase}: {event:?}"
);
}
})
.await
.unwrap_or_else(|_| {
panic!("protocol deadline during {phase}; observed {seen:?}")
});
Ok((transport, matched))
})
.unwrap()
}
fn user_inputs(entries: &[LogEntry]) -> Vec<Vec<protocol::Segment>> {
entries
.iter()
.filter_map(|entry| match entry {
LogEntry::AnnotatedUserInput { segments, .. } => Some(segments.clone()),
_ => None,
})
.collect()
}
// Declaration order keeps the bootstrap runtime (which owns the
// existing loopback Host task) alive through Fixture::drop, even on panic.
let bootstrap_backend;
let mut fixture = Fixture::no_model();
let script = MockClient::sequential(vec![MockResponse::Hang(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, OUTPUT),
LlmEvent::text_block_stop(0, None),
])]);
// Hang explicitly chains stream::pending() after a complete text block.
// Hang([]) would exercise pre-token rollback instead of durable Paused.
let injected = script.clone();
let factory = fixture
.backend
.factory
.as_ref()
.clone()
.with_spawn_model_client_factory(move || Box::new(injected.clone()));
bootstrap_backend = std::mem::replace(
&mut fixture.backend,
Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()),
);
let worker_ref = &fixture.request.worker_ref;
assert!(user_inputs(&fixture.entries()).is_empty());
assert!(
!fixture
.entries()
.iter()
.any(|entry| matches!(entry, LogEntry::Invoke { .. }))
);
let before = fixture.canonical_files();
fixture.prepare(); // actual Profile prepare_restore; not MockFactory::spawn
fixture.assert_quiet(); // independent empty bootstrap client remains unused
assert_eq!(script.call_count.load(Ordering::SeqCst), 0);
assert!(script.captured.lock().unwrap().is_empty());
assert_eq!(fixture.canonical_files(), before);
assert!(
fixture
.backend
.clone()
.attach_worker_protocol(worker_ref)
.is_err()
);
fixture
.backend
.activate_restored_worker(fixture.request.operation_id, worker_ref)
.unwrap();
let execution = fixture.backend.workers.lock().unwrap()[worker_ref].clone();
let transport = fixture
.backend
.clone()
.attach_worker_protocol(worker_ref)
.unwrap();
transport.validate().unwrap();
assert!(
matches!(
&transport.snapshot,
Event::Snapshot { session, state, in_flight, .. }
if state.state == protocol::WorkerState::Idle
&& session.pending_submissions == protocol::PendingSubmissionsSnapshot::default()
&& session == &session_store::public_snapshot::project_current_session_snapshot(&fixture.entries())
&& in_flight.is_empty()
),
"expected full input-ready snapshot: {:?}",
transport.snapshot
);
assert!(user_inputs(&fixture.entries()).is_empty());
assert_eq!(script.call_count.load(Ordering::SeqCst), 0);
assert!(
transport
.dispatch(Method::submit_text(S1, S1))
.unwrap()
.is_empty()
);
let (transport, _) = observe(
&fixture.backend,
transport,
"S1 committed and streaming",
false,
{
let (mut accepted, mut user_message, mut running, mut text_done) =
(false, false, false, false);
move |event| {
match event {
Event::SubmissionAccepted {
submission_request_id,
disposition,
..
} if submission_request_id == S1 => {
assert_eq!(*disposition, protocol::SubmissionDisposition::Started);
accepted = true;
}
Event::UserMessage { segments, .. } => {
assert_eq!(segments, &vec![protocol::Segment::text(S1)]);
user_message = true;
}
Event::WorkerState { snapshot } => {
running |= snapshot.state
== protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Running,
));
}
Event::TextDone { text } => {
assert_eq!(text, OUTPUT);
text_done = true;
}
_ => {}
}
accepted && user_message && running && text_done
}
},
);
// Started is only an admission receipt. Require the actual canonical
// AnnotatedUserInput plus its model-visible history before sending Pause.
assert_eq!(
user_inputs(&fixture.entries()),
vec![vec![protocol::Segment::text(S1)]]
);
let history = session_store::collect_state(&fixture.entries()).annotated_history;
assert!(history.iter().any(|entry| matches!(
&entry.item, LoggedItem::Message { content, .. }
if content.iter().any(|part| matches!(part, LoggedContentPart::Text { text } if text == S1))
)));
assert_eq!(script.call_count.load(Ordering::SeqCst), 1);
let pause = next_internal_command(&execution.worker_state).unwrap();
transport
.dispatch(Method::Pause { command: pause })
.unwrap();
let (transport, _) = observe(&fixture.backend, transport, "Pause completion", false, {
let (mut acknowledged, mut paused, mut ended) = (false, false, false);
move |event| {
match event {
Event::CommandAcknowledged { acknowledgement }
if acknowledgement.command_id == pause.command_id =>
{
assert_eq!(acknowledgement.command, protocol::WorkerCommandKind::Pause);
assert_eq!(
acknowledgement.disposition,
protocol::WorkerCommandDisposition::Accepted
);
acknowledged = true;
}
Event::WorkerState { snapshot } => {
paused |= snapshot.state
== protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Paused,
));
}
Event::RunEnd { result } => {
assert_eq!(*result, protocol::RunResult::Paused);
ended = true;
}
_ => {}
}
acknowledged && paused && ended
}
});
assert_eq!(
execution.handle.shared_state.snapshot().state,
protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Paused
))
);
let paused_entries = fixture.entries();
assert_eq!(
user_inputs(&paused_entries),
vec![vec![protocol::Segment::text(S1)]]
);
let invokes = paused_entries
.iter()
.filter(|entry| matches!(entry, LogEntry::Invoke { .. }))
.count();
assert_eq!(invokes, 1);
// The hanging model response has not appended a final assistant item.
// Prove the actual durable Paused boundary, not a streaming TextDone's
// assumed equivalence to a completed assistant history message.
assert!(paused_entries.iter().any(|entry| matches!(
entry,
LogEntry::RunCompleted {
interrupted: true,
result: agen::EngineResult::Paused,
..
}
)));
transport.dispatch(Method::submit_text(S2, S2)).unwrap();
let (transport, queued) = observe(
&fixture.backend,
transport,
"S2 Queued receipt",
true,
|event| matches!(event, Event::SubmissionAccepted { submission_request_id, .. } if submission_request_id == S2),
);
let submission_id = match &queued {
Event::SubmissionAccepted {
submission_id,
disposition,
..
} => {
assert_eq!(*disposition, protocol::SubmissionDisposition::Queued);
submission_id.clone()
}
_ => unreachable!(),
};
let checkpoint = queue_checkpoint(&fixture.entries());
assert_eq!(checkpoint["pending"].as_array().unwrap().len(), 1);
assert_eq!(checkpoint["pending"][0]["submission_request_id"], S2);
assert_eq!(checkpoint["pending"][0]["submission_id"], submission_id);
assert_eq!(
checkpoint["pending"][0]["input"],
serde_json::to_value(vec![protocol::Segment::text(S2)]).unwrap()
);
let receipts = checkpoint["receipts"].as_array().unwrap();
let s2_receipts: Vec<_> = receipts
.iter()
.filter(|receipt| receipt["submission_request_id"] == S2)
.collect();
assert_eq!(s2_receipts.len(), 1);
assert_eq!(s2_receipts[0]["submission_id"], submission_id);
assert_eq!(s2_receipts[0]["disposition"], "queued");
transport.dispatch(Method::ListPendingSubmissions).unwrap();
let (transport, listed) = observe(
&fixture.backend,
transport,
"S2 pending list",
true,
|event| matches!(event, Event::PendingSubmissionsChanged { .. }),
);
let pending = match listed {
Event::PendingSubmissionsChanged { pending } => pending,
_ => unreachable!(),
};
assert_eq!(pending.submissions.len(), 1);
assert_eq!(pending.head_id.as_deref(), Some(submission_id.as_str()));
assert_eq!(pending.submissions[0].submission_id, submission_id);
let before_retry = fixture.canonical_files();
transport.dispatch(Method::submit_text(S2, S2)).unwrap();
let (transport, replay) = observe(
&fixture.backend,
transport,
"S2 exact retry",
true,
|event| matches!(event, Event::SubmissionAccepted { submission_request_id, .. } if submission_request_id == S2),
);
assert_eq!(
serde_json::to_value(replay).unwrap(),
serde_json::to_value(queued).unwrap()
);
assert_eq!(fixture.canonical_files(), before_retry);
transport.dispatch(Method::ListPendingSubmissions).unwrap();
let (transport, listed) = observe(
&fixture.backend,
transport,
"retry retains one pending",
true,
|event| matches!(event, Event::PendingSubmissionsChanged { .. }),
);
assert!(
matches!(listed, Event::PendingSubmissionsChanged { pending: actual } if actual == pending)
);
// Idle-only admission is a different contract from ordinary Submit:
// observe the real rejection on this connection, not a Queued receipt.
let mut idle_only = WorkerInput::user(IDLE_ONLY);
idle_only.kind = WorkerInputKind::UserIfIdle;
idle_only.submission_request_id = Some(IDLE_ONLY.into());
let rejected = fixture.backend.dispatch_input(worker_ref, idle_only);
assert!(!rejected.is_accepted(), "{rejected:?}");
assert!(
rejected.message_or_default().contains("WorkerNotify"),
"{rejected:?}"
);
let (transport, _) = observe(
&fixture.backend,
transport,
"Paused idle-only rejection",
true,
|event| {
matches!(event, Event::SubmissionRejected { submission_request_id, message }
if submission_request_id == IDLE_ONLY && message.contains("WorkerNotify"))
},
);
assert_eq!(fixture.canonical_files(), before_retry);
assert_eq!(queue_checkpoint(&fixture.entries()), checkpoint);
transport
.dispatch(Method::CancelPendingSubmission { submission_id })
.unwrap();
let (transport, _) = observe(&fixture.backend, transport, "cancel S2", true, |event| {
matches!(event, Event::PendingSubmissionsChanged { pending }
if pending.submissions.is_empty() && pending.head_id.is_none())
});
transport.dispatch(Method::ListPendingSubmissions).unwrap();
let (transport, _) = observe(
&fixture.backend,
transport,
"empty pending list after cancel",
true,
|event| {
matches!(event, Event::PendingSubmissionsChanged { pending }
if pending == &protocol::PendingSubmissionsSnapshot::default())
},
);
let entries = fixture.entries();
assert!(
queue_checkpoint(&entries)["pending"]
.as_array()
.unwrap()
.is_empty()
);
assert_eq!(
user_inputs(&entries),
vec![vec![protocol::Segment::text(S1)]]
);
assert_eq!(
entries
.iter()
.filter(|entry| matches!(entry, LogEntry::Invoke { .. }))
.count(),
invokes
);
assert_eq!(
execution.handle.shared_state.snapshot().state,
protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Paused
))
);
assert_eq!(script.call_count.load(Ordering::SeqCst), 1);
assert_eq!(script.captured.lock().unwrap().len(), 1);
fixture.assert_quiet();
transport.validate().unwrap();
// Normal Stop owns shutdown and the Controller/transport abort+join
// barrier; do not simulate closure or inject a terminal state.
assert!(fixture.backend.stop_worker(worker_ref).is_accepted());
assert!(execution.handle.protocol_is_closed());
assert!(execution.tasks.tasks.lock().unwrap().is_empty());
assert!(transport.validate().is_err());
drop(transport);
drop(execution);
fixture.assert_released();
assert_eq!(script.call_count.load(Ordering::SeqCst), 1);
drop(fixture);
drop(bootstrap_backend);
}
#[test]
#[serial_test::serial(worker_allocation)]
fn prepare_saved_inputs_rejects_new_input_and_discard_preserves_authority() {
@@ -896,6 +1514,7 @@ mod restore_candidate {
}
}
#[cfg(feature = "fs-store")]
#[test]
#[serial_test::serial(worker_allocation)]
fn attachment_preparation_replay_cannot_hide_logical_authority_revocation() {
@@ -1101,4 +1720,6 @@ mod restore_candidate {
server_api::WorkerRestoreState::Accepted
);
}
include!("worker_backend_restore_publication_tests.rs");
}
crates/worker-runtime/src/worker_backend_restore_failure_tests.rs
+31 -48
View File
@@ -2,26 +2,6 @@
// Worker/bootstrap, local Git Workdir provider, and allocation table. External
// model I/O is replaced for saved-input coverage; this is not product E2E.
use crate::restore_test_environment::require_restore_test_environment;
fn restore_failure_tempdir() -> tempfile::TempDir {
require_restore_test_environment();
// Explicitly stay inside this checkout, even when TMPDIR points elsewhere.
// Keep fixture components short; no symlink/alias to evade socket limits.
let target = Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.parent()
.unwrap()
.join("target");
fs::create_dir_all(&target).unwrap();
tempfile::Builder::new()
.prefix("r")
.rand_bytes(3)
.tempdir_in(target)
.unwrap()
}
struct RestoreOpenCountingMaterializer {
inner: RuntimeGitMaterializer,
opens: Arc<AtomicUsize>,
@@ -705,13 +685,8 @@ impl ProfileFailureFixture {
#[test]
#[serial_test::serial(worker_allocation)]
fn profile_restore_prior_effects_survive_read_only_reprepare_failure() {
for (activation_stage, read_stage) in [
(RuntimeRestoreStage::Feature, RuntimeRestoreStage::Metadata),
(
RuntimeRestoreStage::Controller,
RuntimeRestoreStage::Reconstruction,
),
] {
for read_stage in [RuntimeRestoreStage::Metadata, RuntimeRestoreStage::Reconstruction] {
let activation_stage = RuntimeRestoreStage::ActivationEffects;
let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
fixture.fail(activation_stage);
@@ -908,7 +883,7 @@ fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_repr
fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation_failure() {
let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
fixture.fail(RuntimeRestoreStage::Feature);
fixture.fail(RuntimeRestoreStage::ActivationEffects);
let request = fixture
.runtime
.test_restore_request(&fixture.worker.worker_ref);
@@ -926,7 +901,7 @@ fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation
.pending_restore(&fixture.worker.worker_ref)
.unwrap()
.unwrap();
fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Feature);
fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::ActivationEffects);
let effected = persisted_files(&artifacts);
assert_ne!(effected, pinned);
@@ -1215,7 +1190,7 @@ fn profile_restore_multistage_failure_cleans_then_fresh_restore_without_stop() {
#[serial_test::serial(worker_allocation)]
fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_deferred_provider() {
let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
fixture.fail(RuntimeRestoreStage::Feature);
fixture.fail(RuntimeRestoreStage::ActivationEffects);
fixture
.backend
.factory
@@ -1256,7 +1231,7 @@ fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_defer
{
let retained = pending.result.lock().unwrap();
let failure = retained.as_ref().unwrap().as_ref().err().unwrap();
assert_eq!(failure.stage, RuntimeRestoreStage::Feature);
assert_eq!(failure.stage, RuntimeRestoreStage::ActivationEffects);
assert_eq!(failure.resources, RuntimeRestoreResources::WorkerStartup);
assert_eq!(
failure.cleanup,
@@ -1281,12 +1256,17 @@ fn profile_restore_cleanup_failure_retains_owner_and_cause_without_opening_defer
assert!(!pending.request.deferred_workdir_sessions.is_empty());
assert_eq!(owned.operation_id, pending.request.operation_id);
assert!(owned.allocation.is_some());
// The private Controller has already closed its router before the later
// factory cleanup fault. Keep the remaining cleanup owner, not live aliases.
assert!(!owned.sessions.is_empty());
assert!(
owned
.sessions
.iter()
.any(|sessions| !sessions.aliases().is_empty())
.all(|sessions| sessions.aliases().is_empty())
);
// Raw deferred handles are closed by factory cleanup after this injected
// fault is removed; do not open them while cleanup is still pending.
let repeat = fixture
.runtime
.restore_worker_operation(&fixture.worker.worker_ref, request.clone())
@@ -1484,7 +1464,7 @@ async fn failed_restore_raw_cleanup_is_operation_bound_and_does_not_follow_reuse
fn profile_restore_cleanup_timeout_keeps_running_join_and_late_failure_reconciles_without_execution()
{
let fixture = ProfileFailureFixture::new(Duration::from_secs(2));
fixture.fail(RuntimeRestoreStage::Feature);
fixture.fail(RuntimeRestoreStage::ActivationEffects);
let release = Arc::new(tokio::sync::Notify::new());
let (entered_tx, entered) = mpsc::channel();
*fixture
@@ -1579,7 +1559,7 @@ fn profile_restore_cleanup_timeout_keeps_running_join_and_late_failure_reconcile
#[cfg(unix)]
#[test]
#[serial_test::serial(worker_allocation)]
fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation() {
fn profile_restore_real_socket_failure_retains_committed_candidate_until_cleanup() {
let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
fixture
.backend
@@ -1601,7 +1581,7 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
);
let diagnostic = serde_json::to_string(&result).unwrap();
assert!(
diagnostic.contains("controller startup failed"),
diagnostic.contains("publication failed"),
"{diagnostic}"
);
assert!(diagnostic.contains("InvalidInput"), "{diagnostic}");
@@ -1624,15 +1604,14 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
assert!(!diagnostic.contains(&fixture.root.path().to_string_lossy().to_string()));
assert!(!socket.exists());
assert!(fixture.backend.workers.lock().unwrap().is_empty());
assert!(
fixture
.backend
.factory
.failed_restore_resources
.lock()
.unwrap()
.is_empty()
);
assert!(pending.owns_prepared_candidate());
assert!(pending.result.lock().unwrap().as_ref().unwrap().as_ref().unwrap().can_retry_publication());
assert!(fixture.backend.factory.failed_restore_resources.lock().unwrap()[&fixture.worker.worker_ref].allocation.is_some());
let durable: serde_json::Value = serde_json::from_slice(&fs::read(
fixture.root.path().join("s/workers").join(fixture.worker.worker_ref.worker_id.to_string()).join("worker.json")
).unwrap()).unwrap();
assert_eq!(durable["execution_state"]["execution"]["intent"]["commit"], "committed");
assert!(durable["restore_guard"]["owners"][&request.request_id]["receipt"].is_null());
assert!(
fixture
.backend
@@ -1647,10 +1626,14 @@ fn profile_restore_real_socket_start_failure_removes_partial_run_and_allocation(
.restore_faults
.socket_failure
.store(false, Ordering::SeqCst);
eprintln!(
"restore-fault stage=socket-start real-bootstrap=true partial-run=removed allocation=released"
);
fixture.resume_same_operation(request, operation_id);
// A real permanent path-limit error cannot be repaired by silently
// constructing another endpoint/transport. Explicit cleanup settles the
// retained owner. This case does not assert autonomous terminal handling.
drop(pending);
fixture.runtime.stop_worker(&fixture.worker.worker_ref, None).unwrap();
fixture.assert_released();
assert!(!socket.parent().unwrap().exists());
assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
}
#[test]
crates/worker-runtime/src/worker_backend_restore_publication_tests.rs
+535
View File
@@ -0,0 +1,535 @@
// Included in restore_candidate: real Runtime catalog/store, Profile candidate,
// Controller and protocol; only the non-network model client and loopback
// Workspace transport are substituted. These are ordinary integration tests,
// not product-process E2E or live model generation.
#[cfg(all(feature = "fs-store", feature = "ws-server"))]
mod publication {
use super::*;
use crate::fs_store::{
AtomicWriteFault, FsRuntimeStoreOptions, PersistedWorkerExecution,
PersistedWorkerExecutionState, PersistedWorkerRecord,
};
fn stopped_runtime(fixture: &Fixture) -> EmbeddedRuntime {
let options = FsRuntimeStoreOptions {
root: fixture._root.path().join("catalog"),
runtime_id: "candidate-runtime".into(),
display_name: None,
};
let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
options.clone(),
fixture.backend.clone(),
)
.unwrap();
let create = fixture.request.request.clone();
let worker_ref = fixture.request.worker_ref.clone();
runtime
.test_runtime_store()
.write_worker_record(&PersistedWorkerRecord {
restore_guard: crate::runtime::RestoreGuard::default(),
worker_ref: worker_ref.clone(),
worker_id: worker_ref.worker_id,
profile: create.profile.clone(),
display_name: None,
profile_source: create.profile_source.reference(),
config_bundle: create.config_bundle.clone(),
created_at_ms: None,
status: crate::catalog::WorkerStatus::Stopped,
last_finished_submission_request_id: None,
execution_state: PersistedWorkerExecutionState::Available(
PersistedWorkerExecution {
request: create,
binding: None,
restore_intent: crate::catalog::WorkerRestoreIntent::Explicit,
},
),
workspace_id: Some(
fixture
.request
.workspace_scope
.as_ref()
.unwrap()
.workspace_id
.clone(),
),
workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
})
.unwrap();
drop(runtime);
EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone())
.unwrap()
}
fn durable_operation(
runtime: &EmbeddedRuntime,
worker_ref: &WorkerRef,
) -> Option<crate::fs_store::PersistedWorkerRestoreOperation> {
let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
match &loaded.workers[&worker_ref.worker_id].execution_state {
PersistedWorkerExecutionState::ReconciliationRequired(
crate::fs_store::PersistedWorkerLifecycleOperation::Restore(operation),
) => Some(operation.clone()),
_ => None,
}
}
fn prepared_gate(fixture: &Fixture) -> (Arc<tokio::sync::Notify>, mpsc::Receiver<Event>) {
let release = Arc::new(tokio::sync::Notify::new());
let (entered, received) = mpsc::channel();
*fixture
.backend
.factory
.restore_faults
.prepared_gate
.lock()
.unwrap() = Some(release.clone());
*fixture
.backend
.factory
.restore_faults
.prepared_entered
.lock()
.unwrap() = Some(entered);
(release, received)
}
#[test]
#[serial_test::serial(worker_allocation)]
fn runtime_accepted_follows_real_prepared_snapshot_commit_attach_and_input() {
let bootstrap_backend;
let mut fixture = Fixture::no_model();
let script = MockClient::sequential(vec![MockResponse::Complete(simple_text_events())]);
let injected = script.clone();
let factory = fixture
.backend
.factory
.as_ref()
.clone()
.with_spawn_model_client_factory(move || Box::new(injected.clone()));
bootstrap_backend = std::mem::replace(
&mut fixture.backend,
Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()),
);
let runtime = Arc::new(stopped_runtime(&fixture));
let worker_ref = fixture.request.worker_ref.clone();
let scope = fixture.request.workspace_scope.clone().unwrap();
let request = runtime.test_restore_request(&worker_ref);
let before = fixture.canonical_files();
let (release, prepared) = prepared_gate(&fixture);
let restoring = {
let runtime = runtime.clone();
let worker_ref = worker_ref.clone();
let scope = scope.clone();
let request = request.clone();
std::thread::spawn(move || {
runtime.restore_worker_operation_scoped(&scope, &worker_ref, request)
})
};
let snapshot = prepared.recv_timeout(WAIT).unwrap();
assert!(
matches!(&snapshot, Event::Snapshot { session, greeting, state, in_flight, .. }
if session == &session_store::public_snapshot::project_current_session_snapshot(&fixture.entries())
&& !greeting.tools.is_empty() && state.state == protocol::WorkerState::Idle && in_flight.is_empty())
);
let pending = durable_operation(&runtime, &worker_ref).unwrap();
assert_eq!(
pending.commit,
crate::runtime::RestoreCommitFact::NotCommitted
);
assert_eq!(fixture.canonical_files(), before);
fixture.assert_quiet();
fixture.assert_no_allocation();
assert!(fixture.backend.workers.lock().unwrap().is_empty());
assert!(
fixture
.backend
.factory
.observation_hub
.get(&worker_ref)
.is_none()
);
// External attach shares the in-flight backend operation lock; do not
// synchronously acquire that lock while this test holds its prepare gate.
// The empty execution registry and absent hub/allocation prove private
// ownership here; attach is exercised immediately after Accepted below.
assert_eq!(script.call_count.load(Ordering::SeqCst), 0);
release.notify_one();
let accepted = restoring.join().unwrap().unwrap();
assert_eq!(
accepted.state,
server_api::WorkerRestoreState::Accepted,
"{accepted:?}"
);
assert!(durable_operation(&runtime, &worker_ref).is_none());
let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
let guard =
serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
assert!(!guard["owners"][&request.request_id]["receipt"].is_null());
assert_eq!(
fixture
.backend
.factory
.restore_faults
.executions
.load(Ordering::SeqCst),
1
);
assert!(
fixture
.backend
.factory
.observation_hub
.get(&worker_ref)
.is_some()
);
let mut transport = fixture
.backend
.clone()
.attach_worker_protocol(&worker_ref)
.unwrap();
transport.validate().unwrap();
assert_eq!(
serde_json::to_value(&transport.snapshot).unwrap(),
serde_json::to_value(snapshot).unwrap()
);
assert!(matches!(
fixture
.backend
.worker_session(crate::execution::WorkerSessionObservationRequest {
worker_ref: worker_ref.clone(),
}),
runtime_api::WorkerSessionAvailability::LiveProtocol
));
transport
.dispatch(Method::submit_text(
"vertical-body-input",
"vertical-body-input",
))
.unwrap();
fixture
.backend
.run_on_adapter_runtime(async move {
let (mut admitted, mut committed, mut text, mut ended) =
(false, false, false, false);
tokio::time::timeout(WAIT, async {
while !(admitted && committed && text && ended) {
match transport.events.recv().await.unwrap() {
Event::SubmissionAccepted {
submission_request_id,
..
} if submission_request_id == "vertical-body-input" => admitted = true,
Event::UserMessage { segments, .. } => {
assert_eq!(
segments,
vec![protocol::Segment::text("vertical-body-input")]
);
committed = true;
}
Event::TextDone { text: output } => {
assert_eq!(output, "hello from worker");
text = true;
}
Event::RunEnd { .. } => ended = true,
Event::SubmissionRejected { .. } | Event::Error { .. } => {
panic!("normal input rejected after Accepted")
}
_ => {}
}
}
})
.await
.map_err(|_| "normal input did not settle".to_string())?;
// Capture and release the whole transport, not only its events field;
// its dispatch/validation closures retain the actual execution.
drop(transport);
Ok(())
})
.unwrap();
assert!(fixture.entries().iter().any(
|entry| matches!(entry, LogEntry::AnnotatedUserInput { segments, .. }
if segments == &vec![protocol::Segment::text("vertical-body-input")])
));
assert_eq!(script.call_count.load(Ordering::SeqCst), 1);
let replay = runtime
.restore_worker_operation_scoped(&scope, &worker_ref, request.clone())
.unwrap();
assert_eq!(replay.state, server_api::WorkerRestoreState::Accepted);
assert_eq!(
fixture
.backend
.factory
.restore_faults
.executions
.load(Ordering::SeqCst),
1
);
let mut stale = request;
stale.request_id = "vertical-new-stale-intent".into();
assert!(
matches!(runtime.restore_worker_operation_scoped(&scope, &worker_ref, stale), Err(crate::error::RuntimeError::RestoreObservationConflict { worker_id }) if worker_id == worker_ref.worker_id)
);
runtime
.stop_worker_scoped(&scope, &worker_ref, None)
.unwrap();
fixture.assert_released();
assert_eq!(script.call_count.load(Ordering::SeqCst), 1);
drop(runtime);
drop(bootstrap_backend);
}
#[test]
#[serial_test::serial(worker_allocation)]
fn decision_write_faults_preserve_prepared_owner_or_confirm_uncommitted_cleanup() {
for (fault, sync_unknown) in [
(AtomicWriteFault::BeforeRename, false),
(AtomicWriteFault::AfterRename, false),
(AtomicWriteFault::AfterRename, true),
] {
let fixture = Fixture::no_model();
let runtime = Arc::new(stopped_runtime(&fixture));
let worker_ref = fixture.request.worker_ref.clone();
let scope = fixture.request.workspace_scope.clone().unwrap();
let request = runtime.test_restore_request(&worker_ref);
let before = fixture.canonical_files();
let (release, prepared) = prepared_gate(&fixture);
let restoring = {
let runtime = runtime.clone();
let worker_ref = worker_ref.clone();
let scope = scope.clone();
let request = request.clone();
std::thread::spawn(move || {
runtime.restore_worker_operation_scoped(&scope, &worker_ref, request)
})
};
prepared.recv_timeout(WAIT).unwrap();
let operation = durable_operation(&runtime, &worker_ref).unwrap();
assert_eq!(
operation.commit,
crate::runtime::RestoreCommitFact::NotCommitted
);
runtime.test_runtime_store().fail_next_worker_write(fault);
if sync_unknown {
runtime
.test_runtime_store()
.fail_next_worker_reconcile_sync();
}
release.notify_one();
let result = restoring.join().unwrap().unwrap();
match fault {
AtomicWriteFault::BeforeRename => {
assert_eq!(
result.state,
server_api::WorkerRestoreState::RolledBack,
"{result:?}"
);
assert_eq!(fixture.canonical_files(), before);
fixture.assert_released();
}
AtomicWriteFault::AfterRename if !sync_unknown => {
assert_eq!(
result.state,
server_api::WorkerRestoreState::Accepted,
"{result:?}"
);
let transport = fixture
.backend
.clone()
.attach_worker_protocol(&worker_ref)
.unwrap();
transport.validate().unwrap();
drop(transport);
runtime
.stop_worker_scoped(&scope, &worker_ref, None)
.unwrap();
fixture.assert_released();
}
AtomicWriteFault::AfterRename => {
assert_eq!(
result.state,
server_api::WorkerRestoreState::ReconciliationRequired,
"{result:?}"
);
assert_eq!(
durable_operation(&runtime, &worker_ref)
.unwrap()
.operation_id,
operation.operation_id
);
assert_eq!(
durable_operation(&runtime, &worker_ref).unwrap().commit,
crate::runtime::RestoreCommitFact::Committed
);
assert!(fixture.backend.workers.lock().unwrap().is_empty());
fixture.assert_no_allocation();
assert_eq!(fixture.canonical_files(), before);
let replay = runtime
.restore_worker_operation_scoped(&scope, &worker_ref, request)
.unwrap();
assert_eq!(
replay.state,
server_api::WorkerRestoreState::Accepted,
"{replay:?}"
);
assert_eq!(
fixture
.backend
.factory
.restore_faults
.executions
.load(Ordering::SeqCst),
1
);
runtime
.stop_worker_scoped(&scope, &worker_ref, None)
.unwrap();
fixture.assert_released();
}
_ => unreachable!(),
}
fixture.assert_quiet();
}
}
#[test]
#[serial_test::serial(worker_allocation)]
fn uncommitted_cleanup_failure_fences_discarding_candidate_from_activation() {
let fixture = Fixture::no_model();
let runtime = Arc::new(stopped_runtime(&fixture));
let worker_ref = fixture.request.worker_ref.clone();
let scope = fixture.request.workspace_scope.clone().unwrap();
let request = runtime.test_restore_request(&worker_ref);
let before = fixture.canonical_files();
let (release, prepared) = prepared_gate(&fixture);
let restoring = {
let runtime = runtime.clone();
let worker_ref = worker_ref.clone();
let scope = scope.clone();
let request = request.clone();
std::thread::spawn(move || {
runtime.restore_worker_operation_scoped(&scope, &worker_ref, request)
})
};
prepared.recv_timeout(WAIT).unwrap();
runtime.test_runtime_store().fail_next_worker_write(AtomicWriteFault::BeforeRename);
fixture.backend.factory.restore_faults.cleanup_fail.store(true, Ordering::SeqCst);
release.notify_one();
let failed = restoring.join().unwrap().unwrap();
assert_eq!(failed.state, server_api::WorkerRestoreState::ReconciliationRequired, "{failed:?}");
let pending = fixture.backend.pending_restore(&worker_ref).unwrap().unwrap();
assert!(matches!(pending.discard_result.lock().unwrap().as_ref(), Some(Err(_))));
assert_eq!(durable_operation(&runtime, &worker_ref).unwrap().commit, crate::runtime::RestoreCommitFact::NotCommitted);
for cleanup_fails in [true, false] {
fixture.backend.factory.restore_faults.cleanup_fail.store(cleanup_fails, Ordering::SeqCst);
let replay = runtime.restore_worker_operation_scoped(&scope, &worker_ref, request.clone()).unwrap();
assert_eq!(replay.state, server_api::WorkerRestoreState::ReconciliationRequired, "discarding candidate must not activate: {replay:?}");
assert!(fixture.backend.workers.lock().unwrap().is_empty());
assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none());
assert!(fixture.backend.clone().attach_worker_protocol(&worker_ref).is_err());
assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
assert_eq!(fixture.canonical_files(), before);
fixture.assert_no_allocation();
fixture.assert_quiet();
}
// Explicit teardown verifies this retained owner's cleanup. Autonomous
// cleanup convergence is a separate operation-driver contract.
drop(pending);
runtime.stop_worker_scoped(&scope, &worker_ref, None).unwrap();
fixture.assert_released();
}
#[test]
#[serial_test::serial(worker_allocation)]
fn publication_failure_retains_same_committed_candidate_without_reconstruction() {
let fixture = Fixture::no_model();
let runtime = stopped_runtime(&fixture);
let worker_ref = fixture.request.worker_ref.clone();
let scope = fixture.request.workspace_scope.clone().unwrap();
let request = runtime.test_restore_request(&worker_ref);
fixture
.backend
.factory
.restore_faults
.publication_failure
.store(true, Ordering::SeqCst);
let failed = runtime
.restore_worker_operation_scoped(&scope, &worker_ref, request.clone())
.unwrap();
assert_eq!(
failed.state,
server_api::WorkerRestoreState::ReconciliationRequired,
"{failed:?}"
);
let pending = fixture
.backend
.pending_restore(&worker_ref)
.unwrap()
.unwrap();
assert!(pending.owns_prepared_candidate());
assert!(pending.activation_started.load(Ordering::SeqCst));
assert!(
pending
.result
.lock()
.unwrap()
.as_ref()
.unwrap()
.as_ref()
.unwrap()
.can_retry_publication()
);
let durable = durable_operation(&runtime, &worker_ref).unwrap();
assert_eq!(durable.operation_id, pending.request.operation_id);
assert_eq!(durable.commit, crate::runtime::RestoreCommitFact::Committed);
assert!(fixture.backend.workers.lock().unwrap().is_empty());
assert!(
fixture
.backend
.factory
.observation_hub
.get(&worker_ref)
.is_none()
);
assert!(
fixture
.backend
.clone()
.attach_worker_protocol(&worker_ref)
.is_err()
);
fixture.assert_quiet();
let result = runtime
.restore_worker_operation_scoped(&scope, &worker_ref, request)
.unwrap();
assert_eq!(
result.state,
server_api::WorkerRestoreState::Accepted,
"{result:?}"
);
let execution = fixture.backend.workers.lock().unwrap()[&worker_ref].clone();
assert_eq!(execution.restore_operation_id, Some(durable.operation_id));
assert_eq!(
fixture
.backend
.factory
.restore_faults
.executions
.load(Ordering::SeqCst),
1
);
let transport = fixture
.backend
.clone()
.attach_worker_protocol(&worker_ref)
.unwrap();
transport.validate().unwrap();
drop(transport);
drop(execution);
drop(pending);
runtime
.stop_worker_scoped(&scope, &worker_ref, None)
.unwrap();
fixture.assert_released();
fixture.assert_quiet();
}
}
crates/worker-runtime/src/worker_source.rs
+48 -7
View File
@@ -1,23 +1,32 @@
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
#[cfg(feature = "http-server")]
use std::time::Duration;
use std::time::{SystemTime, UNIX_EPOCH};
#[cfg(feature = "http-server")]
use crate::workspace_request::{
RuntimeWorkspaceRequest, RuntimeWorkspaceRequestClient, RuntimeWorkspaceRequestError,
};
use worker::WorkspaceResponse;
#[cfg(feature = "http-server")]
use worker::{
WorkspaceBinaryRequest, WorkspaceBinaryResponse, WorkspaceClient, WorkspaceClientError,
WorkspacePromptCatalogResolution, WorkspacePromptProjection, WorkspaceRequest,
WorkspaceRequestMethod, WorkspaceResponse, WorkspaceServerOperation,
WorkspaceRequestMethod, WorkspaceServerOperation,
};
use crate::auth::{
RuntimeAuthError, RuntimeIdentityMaterial, RuntimeWorkerMutationSourceSigner,
WORKER_REMOVE_PERMISSION, WORKSPACE_REQUEST_PERMISSION, WORKSPACE_WORKER_DISCOVERY_PERMISSION,
WorkerMutationActorKind, WorkerMutationOperation, WorkerMutationSourceClaims, new_token_id,
WORKER_REMOVE_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation,
WorkerMutationSourceClaims, new_token_id,
};
#[cfg(feature = "http-server")]
use crate::auth::{WORKSPACE_REQUEST_PERMISSION, WORKSPACE_WORKER_DISCOVERY_PERMISSION};
use crate::runtime::RuntimeWorkspaceScope;
#[cfg(feature = "http-server")]
use crate::worker_backend::WorkspacePromptProjectionCache;
#[cfg(feature = "http-server")]
fn ticket_assignment_role_segment(role: server_api::TicketAssignmentRole) -> &'static str {
match role {
server_api::TicketAssignmentRole::Orchestrator => "orchestrator",
@@ -144,6 +153,7 @@ pub trait EmbeddedWorkerMutationDispatcher: Send + Sync {
#[derive(Clone)]
enum RuntimeWorkerMutationTransport {
#[cfg(feature = "http-server")]
Remote {
request_client: RuntimeWorkspaceRequestClient,
},
@@ -161,6 +171,7 @@ pub struct RuntimeWorkerMutationForwarder {
}
impl RuntimeWorkerMutationForwarder {
#[cfg(feature = "http-server")]
pub fn remote(
identity: &RuntimeIdentityMaterial,
scope: RuntimeWorkspaceScope,
@@ -206,6 +217,7 @@ impl RuntimeWorkerMutationForwarder {
target_worker_id,
)?;
match (&self.transport, proof) {
#[cfg(feature = "http-server")]
(
RuntimeWorkerMutationTransport::Remote { request_client },
RuntimeOwnedWorkerMutationProof::Remote(token),
@@ -231,6 +243,7 @@ impl RuntimeWorkerMutationForwarder {
}
}
#[cfg(feature = "http-server")]
struct RemoteWorkerRemoveHttpRequest {
request_client: RuntimeWorkspaceRequestClient,
source_worker_id: String,
@@ -240,6 +253,7 @@ struct RemoteWorkerRemoveHttpRequest {
reason: String,
}
#[cfg(feature = "http-server")]
fn execute_remote_worker_remove_http(
request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
@@ -263,6 +277,7 @@ fn execute_remote_worker_remove_http(
execute_remote_worker_remove_http_blocking(request)
}
#[cfg(feature = "http-server")]
fn execute_remote_worker_remove_http_blocking(
request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
@@ -315,6 +330,7 @@ fn execute_remote_worker_remove_http_blocking(
}
}
#[cfg(feature = "http-server")]
#[derive(Clone)]
pub struct RuntimeOwnedWorkspaceClient {
workspace_id: String,
@@ -325,6 +341,7 @@ pub struct RuntimeOwnedWorkspaceClient {
prompt_projection_cache: Option<Arc<WorkspacePromptProjectionCache>>,
}
#[cfg(feature = "http-server")]
impl RuntimeOwnedWorkspaceClient {
pub fn new(
workspace_id: impl Into<String>,
@@ -495,6 +512,7 @@ impl RuntimeOwnedWorkspaceClient {
}
}
#[cfg(feature = "http-server")]
impl std::fmt::Debug for RuntimeOwnedWorkspaceClient {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
@@ -510,6 +528,7 @@ impl std::fmt::Debug for RuntimeOwnedWorkspaceClient {
}
}
#[cfg(feature = "http-server")]
fn generated_workspace_response<T: serde::Serialize>(
response: Result<T, server_api::client_support::ClientError<server_api::RepositoryApiError>>,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
@@ -530,6 +549,7 @@ fn generated_workspace_response<T: serde::Serialize>(
}
}
#[cfg(feature = "http-server")]
impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
fn workspace_id(&self) -> Option<&str> {
Some(self.request_client.workspace_id())
@@ -929,12 +949,17 @@ fn unix_now_seconds() -> u64 {
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "http-server")]
use crate::auth::{
RUNTIME_REQUEST_SOURCE_PROOF_HEADER, WorkerMutationSourceExpectation,
decode_runtime_request_source_claims, decode_worker_mutation_source_claims,
request_body_digest, verify_worker_mutation_source_proof,
RUNTIME_REQUEST_SOURCE_PROOF_HEADER, decode_runtime_request_source_claims,
request_body_digest,
};
use crate::auth::{
WorkerMutationSourceExpectation, decode_worker_mutation_source_claims,
verify_worker_mutation_source_proof,
};
#[cfg(feature = "http-server")]
fn binary_loopback(response: Vec<u8>) -> (String, std::thread::JoinHandle<(String, Vec<u8>)>) {
use std::io::{Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
@@ -972,6 +997,7 @@ mod tests {
(base_url, server)
}
#[cfg(feature = "http-server")]
#[test]
fn generated_ticket_assignment_transport_signs_exact_authoring_and_bounded_read_subjects() {
use crate::auth::{RuntimeRequestSourceExpectation, verify_runtime_request_source};
@@ -1104,6 +1130,7 @@ mod tests {
}
}
#[cfg(feature = "http-server")]
#[test]
fn binary_roundtrip_signs_exact_query_method_worker_and_raw_body() {
use crate::auth::{RuntimeRequestSourceExpectation, verify_runtime_request_source};
@@ -1194,6 +1221,7 @@ mod tests {
}
}
#[cfg(feature = "http-server")]
#[test]
fn binary_responses_enforce_declared_and_streamed_limits_with_fixed_errors() {
for response in [
@@ -1213,6 +1241,7 @@ mod tests {
}
}
#[cfg(feature = "http-server")]
#[test]
fn binary_redirect_returns_status_without_forwarding_authentication() {
let (base_url, server) = binary_loopback(b"HTTP/1.1 307 Temporary Redirect\r\nLocation: http://127.0.0.1:1/private\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec());
@@ -1231,6 +1260,7 @@ mod tests {
assert!(result.body.is_empty());
}
#[cfg(feature = "http-server")]
#[test]
fn binary_invalid_or_normalized_targets_fail_with_fixed_errors_before_http() {
let client = RuntimeOwnedWorkspaceClient::new(
@@ -1277,6 +1307,7 @@ mod tests {
);
}
#[cfg(feature = "http-server")]
#[test]
fn current_prompt_projection_uses_the_shared_runtime_cache_without_http() {
let cache = Arc::new(WorkspacePromptProjectionCache::default());
@@ -1312,6 +1343,7 @@ mod tests {
assert!(Arc::ptr_eq(&projection.catalog, &second.catalog));
}
#[cfg(feature = "http-server")]
#[test]
fn prompt_projection_cache_miss_fetches_content_identity() {
use std::io::{Read, Write};
@@ -1356,6 +1388,7 @@ mod tests {
assert_eq!(fetched.projection.source_digest, "source-3");
}
#[cfg(feature = "http-server")]
#[test]
fn concurrent_prompt_projection_miss_fetches_once_and_shares_catalog() {
use std::io::{Read, Write};
@@ -1427,6 +1460,7 @@ mod tests {
);
}
#[cfg(feature = "http-server")]
#[test]
fn current_prompt_projection_rejects_cross_workspace_response() {
use std::io::{Read, Write};
@@ -1473,6 +1507,7 @@ mod tests {
assert!(error.to_string().contains("scope mismatch"));
}
#[cfg(feature = "http-server")]
#[test]
fn ordinary_workspace_forwarding_stamps_runtime_identity_and_signs_path_and_query() {
use std::io::{Read, Write};
@@ -1531,6 +1566,7 @@ mod tests {
);
}
#[cfg(feature = "http-server")]
#[test]
fn workspace_worker_discovery_signs_dedicated_permission_and_encoded_query() {
use std::io::{Read, Write};
@@ -1652,6 +1688,7 @@ mod tests {
);
}
#[cfg(feature = "http-server")]
#[tokio::test(flavor = "multi_thread")]
async fn remote_forwarder_is_safe_in_async_runtime_and_stamps_signed_proof() {
use std::io::{Read, Write};
@@ -1815,6 +1852,7 @@ mod tests {
assert_eq!(reason, "retire obsolete Worker");
}
#[cfg(feature = "http-server")]
#[test]
fn runtime_owned_workspace_client_returns_successful_response_without_request_timeout() {
// Effective deadline clearing is exercised with a short inherited deadline in
@@ -1838,6 +1876,7 @@ mod tests {
server.join().unwrap();
}
#[cfg(feature = "http-server")]
#[test]
fn runtime_owned_workspace_client_honors_per_request_timeouts() {
let (base_url, server) = delayed_workspace_response(Duration::from_millis(75));
@@ -1857,6 +1896,7 @@ mod tests {
server.join().unwrap();
}
#[cfg(feature = "http-server")]
#[test]
fn runtime_owned_workspace_client_reports_request_timeouts() {
let (base_url, server) = delayed_workspace_response(Duration::from_millis(75));
@@ -1878,6 +1918,7 @@ mod tests {
server.join().unwrap();
}
#[cfg(feature = "http-server")]
fn delayed_workspace_response(delay: Duration) -> (String, std::thread::JoinHandle<()>) {
use std::io::{Read, Write};
use std::net::TcpListener;
crates/workspace-server/src/server_assignment_attention_tests.rs
+1
View File
@@ -477,6 +477,7 @@ impl AttentionWorkerFactory {
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
controller_task: started.controller_task,
workspace_client,
restore_streams: None,
})
}
}
crates/workspace-server/src/server_assignment_delivery_tests.rs
+1
View File
@@ -835,6 +835,7 @@ impl DeliveryWorkerFactory {
shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
controller_task: started.controller_task,
workspace_client,
restore_streams: None,
})
}
}