fix: retry rejected automatic worker restores
This commit is contained in:
@@ -1646,10 +1646,15 @@ impl Runtime {
|
|||||||
"Worker restore cleanup or reconciliation is still pending",
|
"Worker restore cleanup or reconciliation is still pending",
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
match mode {
|
// An active persisted Worker can still have no live handle when its
|
||||||
WorkerRestoreMode::Explicit
|
// startup restore was rejected before side effects (for example,
|
||||||
if worker.execution_metadata_available
|
// before Workspace Backend renewed operation-scoped Repository
|
||||||
&& worker.status != WorkerStatus::Stopped =>
|
// access). Preserve its automatic intent so an explicit restore can
|
||||||
|
// serve as the authorized retry for that state.
|
||||||
|
if mode == WorkerRestoreMode::Explicit
|
||||||
|
&& worker.execution_metadata_available
|
||||||
|
&& worker.status != WorkerStatus::Stopped
|
||||||
|
&& worker.restore_intent != WorkerRestoreIntent::Automatic
|
||||||
{
|
{
|
||||||
return Ok(RuntimeWorkerRestoreResult::failed(
|
return Ok(RuntimeWorkerRestoreResult::failed(
|
||||||
WorkerRestoreState::Rejected,
|
WorkerRestoreState::Rejected,
|
||||||
@@ -1657,14 +1662,12 @@ impl Runtime {
|
|||||||
"Worker restore requires a stopped Worker",
|
"Worker restore requires a stopped Worker",
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
WorkerRestoreMode::Automatic
|
if mode == WorkerRestoreMode::Automatic
|
||||||
if !worker.status.is_active()
|
&& (!worker.status.is_active()
|
||||||
|| worker.restore_intent != WorkerRestoreIntent::Automatic =>
|
|| worker.restore_intent != WorkerRestoreIntent::Automatic)
|
||||||
{
|
{
|
||||||
return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
|
return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
|
||||||
}
|
}
|
||||||
_ => {}
|
|
||||||
}
|
|
||||||
let Some(worker_request) = worker.request.clone() else {
|
let Some(worker_request) = worker.request.clone() else {
|
||||||
return Ok(RuntimeWorkerRestoreResult::failed(
|
return Ok(RuntimeWorkerRestoreResult::failed(
|
||||||
WorkerRestoreState::Rejected,
|
WorkerRestoreState::Rejected,
|
||||||
@@ -8035,6 +8038,59 @@ mod tests {
|
|||||||
let _ = std::fs::remove_dir_all(root);
|
let _ = std::fs::remove_dir_all(root);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(feature = "fs-store")]
|
||||||
|
#[test]
|
||||||
|
fn fs_store_explicitly_retries_rejected_automatic_restore() {
|
||||||
|
let root = fs_store_root("retry-rejected-automatic-restore");
|
||||||
|
let options = crate::fs_store::FsRuntimeStoreOptions {
|
||||||
|
root: root.clone(),
|
||||||
|
runtime_id: "test-runtime".to_string(),
|
||||||
|
display_name: None,
|
||||||
|
};
|
||||||
|
let runtime = Runtime::with_fs_store_and_execution_backend(
|
||||||
|
options.clone(),
|
||||||
|
Arc::new(TestExecutionBackend::default()),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
runtime.store_config_bundle(test_bundle()).unwrap();
|
||||||
|
let worker = runtime
|
||||||
|
.create_worker(task_request("retry rejected automatic restore"))
|
||||||
|
.unwrap();
|
||||||
|
drop(runtime);
|
||||||
|
|
||||||
|
let restoring_backend = Arc::new(TestExecutionBackend::default());
|
||||||
|
*restoring_backend.restore_result.lock().unwrap() = Some(
|
||||||
|
WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
|
||||||
|
WorkerExecutionOperation::Restore,
|
||||||
|
"repository access must be reacquired",
|
||||||
|
)),
|
||||||
|
);
|
||||||
|
let restored =
|
||||||
|
Runtime::with_fs_store_and_execution_backend(options, restoring_backend.clone())
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1);
|
||||||
|
let pending_worker = restored.worker_detail(&worker.worker_ref).unwrap();
|
||||||
|
assert_eq!(pending_worker.status, WorkerStatus::Idle);
|
||||||
|
assert!(pending_worker.execution_metadata_available);
|
||||||
|
|
||||||
|
restoring_backend.restore_result.lock().unwrap().take();
|
||||||
|
let retry = restored
|
||||||
|
.restore_worker_operation(&worker.worker_ref)
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(retry.state, WorkerRestoreState::Accepted);
|
||||||
|
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 2);
|
||||||
|
restored
|
||||||
|
.send_input(
|
||||||
|
&worker.worker_ref,
|
||||||
|
WorkerInput::user("after repository access renewal"),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(root);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn restore_preflight_rejection_is_typed_and_side_effect_free() {
|
fn restore_preflight_rejection_is_typed_and_side_effect_free() {
|
||||||
let (runtime, backend) = runtime_and_backend();
|
let (runtime, backend) = runtime_and_backend();
|
||||||
|
|||||||
Reference in New Issue
Block a user