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

Implement transactional Worker restore outcomes

Browse Source
This commit is contained in:
Hare
2026-09-18 00:08:00 +09:00
parent 5ae1bab4ee
commit 1d7cd9339c
11 changed files with 739 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/execution.rs
+30
View File
@@ -286,7 +286,21 @@ pub enum WorkerExecutionSpawnResult {
worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<WorkingDirectoryStatus>,
},
/// A read-only preflight rejected the operation before live work started.
Rejected(WorkerExecutionResult),
/// Live work started, failed, and all operation-owned resources were joined/released.
RolledBack(WorkerExecutionResult),
/// The operation crossed the live side-effect boundary and could not prove
/// either commit or complete cleanup. Optional handle/state fields preserve
/// concrete execution evidence for retry reconciliation.
ReconciliationRequired {
result: WorkerExecutionResult,
handle: Option<WorkerExecutionHandle>,
worker_state: Option<protocol::WorkerStateSnapshot>,
working_directory: Option<WorkingDirectoryStatus>,
},
/// Legacy spawn failure. Restore implementations must use `RolledBack` or
/// `ReconciliationRequired` after crossing their live side-effect boundary.
Errored(WorkerExecutionResult),
}
@@ -316,6 +330,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult;
/// Read-only restore validation performed before any live controller,
/// bridge, task, or external registration is created.
fn preflight_restore(
&self,
_request: &WorkerExecutionRestoreRequest,
) -> Result<(), WorkerExecutionResult> {
Ok(())
}
fn restore_worker(
&self,
_request: WorkerExecutionRestoreRequest,
@@ -509,6 +532,13 @@ impl WorkerExecutionBackendRef {
self.backend.spawn_worker(request)
}
pub(crate) fn preflight_restore(
&self,
request: &WorkerExecutionRestoreRequest,
) -> Result<(), WorkerExecutionResult> {
self.backend.preflight_restore(request)
}
pub(crate) fn restore_worker(
&self,
request: WorkerExecutionRestoreRequest,
crates/worker-runtime/src/http_server.rs
+30 -7
View File
@@ -60,6 +60,7 @@ use protocol::subscription::{
SubscriptionTerminationCode,
};
use serde::{Deserialize, Serialize};
use workspace_api::WorkerRestoreState;
use std::collections::HashMap;
use std::fmt;
use std::net::SocketAddr;
@@ -420,6 +421,20 @@ pub struct RuntimeHttpWorkerResponse {
pub worker: WorkerDetail,
}
/// Explicit restore outcome. This envelope is returned for every reachable
/// Runtime restore attempt, including preflight rejection and uncertain commit.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RuntimeHttpWorkerRestoreResponse {
pub state: WorkerRestoreState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerDetail>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reason_code: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
/// Replace the Workspace API binding for an existing Worker.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerWorkspaceApiRequest {
@@ -1199,14 +1214,21 @@ async fn restore_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
Path(worker_id): Path<String>,
) -> RestResult<RuntimeHttpWorkerResponse> {
) -> RestResult<RuntimeHttpWorkerRestoreResponse> {
let worker_ref = worker_ref_for(&state.runtime, worker_id)?;
let worker = match auth_workspace_scope(&state, auth.as_ref())? {
Some(scope) => state.runtime.restore_worker_scoped(&scope, &worker_ref),
None => state.runtime.restore_worker(&worker_ref),
let result = match auth_workspace_scope(&state, auth.as_ref())? {
Some(scope) => state
.runtime
.restore_worker_operation_scoped(&scope, &worker_ref),
None => state.runtime.restore_worker_operation(&worker_ref),
}
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkerResponse { worker }))
Ok(Json(RuntimeHttpWorkerRestoreResponse {
state: result.state,
worker: result.worker,
reason_code: result.reason_code,
message: result.message,
}))
}
#[cfg(feature = "ws-server")]
@@ -3322,8 +3344,9 @@ mod tests {
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let restored: RuntimeHttpWorkerResponse = read_json(response).await;
assert_eq!(restored.worker.status, WorkerStatus::Idle);
let restored: RuntimeHttpWorkerRestoreResponse = read_json(response).await;
assert_eq!(restored.state, WorkerRestoreState::Accepted);
assert_eq!(restored.worker.unwrap().status, WorkerStatus::Idle);
let response = authed_empty_request(
app.clone(),
crates/worker-runtime/src/runtime.rs
+408 -114
View File
@@ -54,6 +54,7 @@ use std::sync::{Arc, Mutex, MutexGuard, Weak};
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use uuid::Uuid;
use workspace_api::WorkerRestoreState;
/// Workspace-scoped Runtime authorization context supplied by a trusted backend.
#[derive(Clone, Debug, Eq, PartialEq)]
@@ -83,6 +84,47 @@ enum WorkerRestoreMode {
Automatic,
}
/// Runtime-internal restore result consumed by both embedded and HTTP
/// providers. Diagnostics are stable and path-free; detailed backend errors
/// remain in Runtime logs rather than crossing the API boundary.
#[derive(Clone, Debug)]
pub struct RuntimeWorkerRestoreResult {
pub state: WorkerRestoreState,
pub worker: Option<WorkerDetail>,
pub reason_code: Option<String>,
pub message: Option<String>,
}
impl RuntimeWorkerRestoreResult {
fn accepted(worker: WorkerDetail) -> Self {
Self {
state: WorkerRestoreState::Accepted,
worker: Some(worker),
reason_code: None,
message: None,
}
}
fn failed(state: WorkerRestoreState, reason_code: &'static str, message: &'static str) -> Self {
Self {
state,
worker: None,
reason_code: Some(reason_code.to_string()),
message: Some(message.to_string()),
}
}
fn into_legacy_result(self) -> Result<WorkerDetail, RuntimeError> {
match (self.state, self.worker) {
(WorkerRestoreState::Accepted, Some(worker)) => Ok(worker),
_ => Err(RuntimeError::InvalidRequest(
self.message
.unwrap_or_else(|| "Worker restore did not commit".to_string()),
)),
}
}
}
#[derive(Clone, Debug)]
pub struct RuntimeSubscriptionUpdate {
pub subject_revision: u64,
@@ -1047,6 +1089,7 @@ impl Runtime {
working_directory,
} => (handle, worker_state, working_directory),
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::RolledBack(result)
| WorkerExecutionSpawnResult::Errored(result) => {
self.rollback_failed_create(&worker_ref)?;
return Err(RuntimeError::WorkerExecutionRejected {
@@ -1057,6 +1100,16 @@ impl Runtime {
result,
});
}
WorkerExecutionSpawnResult::ReconciliationRequired { result, .. } => {
self.rollback_failed_create(&worker_ref)?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
outcome: result.outcome,
message: result.message_or_default(),
result,
});
}
};
if let Some(mut initial_input) = {
@@ -1377,7 +1430,16 @@ impl Runtime {
scope: &RuntimeWorkspaceScope,
worker_ref: &WorkerRef,
) -> Result<WorkerDetail, RuntimeError> {
self.restore_worker_with_scope(worker_ref, Some(scope))
self.restore_worker_operation_scoped(scope, worker_ref)?
.into_legacy_result()
}
pub fn restore_worker_operation_scoped(
&self,
scope: &RuntimeWorkspaceScope,
worker_ref: &WorkerRef,
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
self.restore_worker_operation_with_scope(worker_ref, Some(scope))
}
/// Attach a live execution to a persisted Worker definition.
@@ -1387,14 +1449,21 @@ impl Runtime {
/// converges on its already-installed execution rather than spawning another
/// controller.
pub fn restore_worker(&self, worker_ref: &WorkerRef) -> Result<WorkerDetail, RuntimeError> {
self.restore_worker_with_scope(worker_ref, None)
self.restore_worker_operation(worker_ref)?.into_legacy_result()
}
fn restore_worker_with_scope(
pub fn restore_worker_operation(
&self,
worker_ref: &WorkerRef,
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
self.restore_worker_operation_with_scope(worker_ref, None)
}
fn restore_worker_operation_with_scope(
&self,
worker_ref: &WorkerRef,
scope: Option<&RuntimeWorkspaceScope>,
) -> Result<WorkerDetail, RuntimeError> {
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?;
let _operation_guard = operation_lock
.lock()
@@ -1409,58 +1478,61 @@ impl Runtime {
&self,
worker_ref: &WorkerRef,
mode: WorkerRestoreMode,
) -> Result<WorkerDetail, RuntimeError> {
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
let (backend, request) = {
let mut state = self.lock()?;
let state = self.lock()?;
state.ensure_running()?;
let (worker_request, previous_working_directory) = {
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_some() {
if worker.status.is_active() {
return Ok(worker.detail());
}
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id,
message: "previous execution cleanup has not completed".to_string(),
});
}
match mode {
WorkerRestoreMode::Explicit
if worker.execution_metadata_available
&& worker.status != WorkerStatus::Stopped =>
{
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is not stopped",
worker_ref.worker_id
)));
}
WorkerRestoreMode::Automatic
if !worker.status.is_active()
|| worker.restore_intent != WorkerRestoreIntent::Automatic =>
{
return Ok(worker.detail());
}
_ => {}
}
let request = worker.request.clone().ok_or_else(|| {
RuntimeError::WorkerExecutionUnavailable {
worker_id: worker.worker_id,
message: "persisted Worker restore request is unavailable".to_string(),
}
})?;
(request, worker.working_directory.clone())
};
let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id,
message: "runtime has no execution backend".to_string(),
}
})?;
{
let worker = state.worker_mut(worker_ref)?;
worker.execution_bound = true;
let worker = state.worker(worker_ref)?;
if worker.execution_bound && !worker.execution_metadata_available {
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_reconciliation_pending",
"Worker restore cleanup or reconciliation is still pending",
));
}
state.persist_worker(&worker_ref.worker_id)?;
if worker.execution_handle.is_some() {
if worker.status.is_active() {
return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
}
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_cleanup_pending",
"Worker restore cleanup or reconciliation is still pending",
));
}
match mode {
WorkerRestoreMode::Explicit
if worker.execution_metadata_available
&& worker.status != WorkerStatus::Stopped =>
{
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::Rejected,
"worker_restore_not_stopped",
"Worker restore requires a stopped Worker",
));
}
WorkerRestoreMode::Automatic
if !worker.status.is_active()
|| worker.restore_intent != WorkerRestoreIntent::Automatic =>
{
return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
}
_ => {}
}
let Some(worker_request) = worker.request.clone() else {
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::Rejected,
"worker_restore_metadata_unavailable",
"Persisted Worker restore metadata is unavailable",
));
};
let Some(backend) = state.execution_backend.clone() else {
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::Rejected,
"worker_restore_backend_unavailable",
"Runtime has no Worker execution backend",
));
};
let workspace_scope = worker_request.workspace_api.as_ref().and_then(|api| {
state
.workspace_owners
@@ -1472,13 +1544,21 @@ impl Runtime {
request: worker_request,
workspace_scope,
context: self.execution_context(worker_ref.clone()),
previous_working_directory,
previous_working_directory: worker.working_directory.clone(),
working_directory: None,
config_bundle: None,
};
(backend, request)
};
if let Err(_result) = backend.preflight_restore(&request) {
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::Rejected,
"worker_restore_preflight_rejected",
"Worker restore preflight rejected the current request",
));
}
match backend.restore_worker(request) {
WorkerExecutionSpawnResult::Connected {
handle,
@@ -1488,30 +1568,117 @@ impl Runtime {
let commit = self.commit_restored_worker_execution(
worker_ref,
handle.clone(),
worker_state,
worker_state.clone(),
WorkerStatus::Idle,
working_directory,
working_directory.clone(),
);
if let Err(error) = commit {
self.cleanup_failed_restore(&backend, worker_ref, &handle)?;
return Err(error);
match commit {
Ok(worker) => Ok(RuntimeWorkerRestoreResult::accepted(worker)),
Err(error) => match self.cleanup_failed_restore(&backend, worker_ref, &handle) {
Ok(()) => {
tracing::warn!(
worker_id = %worker_ref.worker_id,
error = %error,
"Worker restore commit failed and was rolled back"
);
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::RolledBack,
"worker_restore_commit_rolled_back",
"Worker restore commit failed and cleanup completed",
))
}
Err(cleanup_error) => {
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %error,
cleanup_error = %cleanup_error,
"Worker restore requires reconciliation after commit and cleanup failures"
);
if let Err(retain_error) = self.retain_restore_execution_evidence(
worker_ref,
handle,
worker_state,
working_directory,
) {
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %retain_error,
"failed to persist restore execution evidence; in-process reconciliation remains required"
);
}
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_reconciliation_required",
"Worker restore result is uncertain; reread or retry the Worker restore",
))
}
},
}
self.worker_detail(worker_ref)
}
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::Errored(result) => {
WorkerExecutionSpawnResult::Rejected(_result) => {
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::Rejected,
"worker_restore_rejected",
"Worker restore was rejected before live work started",
))
}
WorkerExecutionSpawnResult::RolledBack(result) => {
#[cfg(feature = "fs-store")]
self.lock()?
.record_restore_failure(worker_ref, result)?;
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::RolledBack,
"worker_restore_rolled_back",
"Worker restore failed after live work started; cleanup completed",
))
}
WorkerExecutionSpawnResult::ReconciliationRequired {
result: _result,
handle,
worker_state,
working_directory,
} => {
if let (Some(handle), Some(worker_state)) = (handle, worker_state) {
if let Err(retain_error) = self.retain_restore_execution_evidence(
worker_ref,
handle,
worker_state,
working_directory,
) {
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %retain_error,
"failed to persist restore execution evidence; in-process reconciliation remains required"
);
}
} else if let Err(retain_error) =
self.retain_restore_reconciliation_pending(worker_ref)
{
self.lock()?
.record_restore_failure(worker_ref, result.clone())?;
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %retain_error,
"failed to persist pending restore reconciliation marker"
);
}
Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id,
operation: result.operation,
outcome: result.outcome,
message: result.message_or_default(),
result,
})
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_reconciliation_required",
"Worker restore result is uncertain; reread or retry the Worker restore",
))
}
WorkerExecutionSpawnResult::Errored(_result) => {
if let Err(retain_error) = self.retain_restore_reconciliation_pending(worker_ref) {
tracing::error!(
worker_id = %worker_ref.worker_id,
error = %retain_error,
"failed to persist pending restore reconciliation marker"
);
}
Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_backend_outcome_unknown",
"Worker restore backend outcome is uncertain; reread or retry the Worker restore",
))
}
}
}
@@ -2469,27 +2636,83 @@ impl Runtime {
worker_state: protocol::WorkerStateSnapshot,
status: WorkerStatus,
working_directory: Option<CatalogWorkingDirectoryStatus>,
) -> Result<(), RuntimeError> {
) -> Result<WorkerDetail, RuntimeError> {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
{
let worker = state.worker_mut(worker_ref)?;
if worker.execution_handle.is_some() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} already has a current execution",
worker_ref.worker_id
)));
}
worker.execution_handle = Some(handle);
worker.execution_metadata_available = true;
worker.execution_bound = true;
worker.status = status;
let _ = worker.apply_worker_state(&worker_state);
worker.restore_intent = restore_intent_for_status(worker.status);
worker.working_directory = working_directory;
let mut candidate = state.worker(worker_ref)?.clone();
if candidate.execution_handle.is_some() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} already has a current execution",
worker_ref.worker_id
)));
}
candidate.execution_handle = Some(handle);
candidate.execution_metadata_available = true;
candidate.execution_bound = true;
candidate.status = status;
let _ = candidate.apply_worker_state(&worker_state);
candidate.restore_intent = restore_intent_for_status(candidate.status);
candidate.working_directory = working_directory;
// Validate the exact subscription projection before persistence. The
// subsequent publish uses the same pure projection while the Worker
// operation lock keeps this candidate stable.
let _subscription_projection = state.subscription_worker(&candidate)?;
// Persist the complete candidate before making it observable. A failed
// write therefore leaves both the in-memory catalog and subscriptions
// on the pre-restore record, so cleanup can produce a true rollback.
state.persist_worker_record(&candidate)?;
let detail = candidate.detail();
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
if let Err(error) = state.persist_runtime_snapshot() {
tracing::warn!(
worker_id = %worker_ref.worker_id,
error = %error,
"Worker restore committed to its canonical record but Runtime snapshot refresh failed"
);
}
Ok(detail)
}
fn retain_restore_execution_evidence(
&self,
worker_ref: &WorkerRef,
handle: WorkerExecutionHandle,
worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<CatalogWorkingDirectoryStatus>,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
candidate.execution_handle = Some(handle);
candidate.execution_metadata_available = true;
candidate.execution_bound = true;
candidate.status = WorkerStatus::Idle;
let _ = candidate.apply_worker_state(&worker_state);
candidate.restore_intent = WorkerRestoreIntent::Automatic;
candidate.working_directory = working_directory;
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
fn retain_restore_reconciliation_pending(
&self,
worker_ref: &WorkerRef,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
candidate.execution_handle = None;
candidate.execution_metadata_available = false;
candidate.execution_bound = true;
// Active-without-metadata is the durable pending-reconciliation marker:
// it prevents a false Stopped projection and blocks duplicate restore.
candidate.status = WorkerStatus::Idle;
candidate.restore_intent = WorkerRestoreIntent::Explicit;
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
@@ -2513,18 +2736,6 @@ impl Runtime {
result,
});
}
let mut state = self.lock()?;
let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = None;
worker.execution_bound = false;
worker.status = WorkerStatus::Stopped;
worker.worker_state = None;
worker.restore_intent = WorkerRestoreIntent::Explicit;
worker.internal_workers.clear();
state.publish_worker_upsert(worker_ref.worker_id)?;
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
@@ -2908,6 +3119,14 @@ impl RuntimeState {
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_worker_record(&self, worker: &WorkerRecord) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
store.write_worker_record(&worker.persisted_record())?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_worker(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
@@ -2945,6 +3164,11 @@ impl RuntimeState {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_worker_record(&self, _worker: &WorkerRecord) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_worker(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> {
Ok(())
@@ -3689,7 +3913,7 @@ struct RemovedInternalWorkerActivity {
parent_session_id: Option<String>,
}
#[derive(Debug, Default)]
#[derive(Debug, Default, Clone)]
struct InternalWorkerActivityProjection {
workers: BTreeMap<String, InternalWorkerActivity>,
removed_workers: BTreeMap<String, RemovedInternalWorkerActivity>,
@@ -3708,7 +3932,7 @@ impl InternalWorkerActivityProjection {
}
}
#[derive(Debug)]
#[derive(Debug, Clone)]
struct WorkerRecord {
worker_ref: WorkerRef,
worker_id: WorkerId,
@@ -4998,6 +5222,7 @@ mod tests {
stop_gate: Mutex<Option<Arc<RestoreGate>>>,
stop_count: Mutex<u64>,
restore_result: Mutex<Option<WorkerExecutionSpawnResult>>,
preflight_restore_result: Mutex<Option<WorkerExecutionResult>>,
restore_gate: Mutex<Option<Arc<RestoreGate>>>,
restore_count: Mutex<u64>,
config_bundles: Mutex<Vec<Option<ConfigBundle>>>,
@@ -5117,6 +5342,16 @@ mod tests {
}
}
fn preflight_restore(
&self,
_request: &WorkerExecutionRestoreRequest,
) -> Result<(), WorkerExecutionResult> {
match self.preflight_restore_result.lock().unwrap().clone() {
Some(result) => Err(result),
None => Ok(()),
}
}
fn restore_worker(
&self,
request: WorkerExecutionRestoreRequest,
@@ -7229,6 +7464,69 @@ mod tests {
let _ = std::fs::remove_dir_all(root);
}
#[test]
fn restore_preflight_rejection_is_typed_and_side_effect_free() {
let (runtime, backend) = runtime_and_backend();
runtime.store_config_bundle(test_bundle()).unwrap();
let created = runtime
.create_worker(task_request("restore preflight"))
.unwrap();
runtime.stop_worker(&created.worker_ref, None).unwrap();
let before = runtime.worker_detail(&created.worker_ref).unwrap();
backend.preflight_restore_result.lock().unwrap().replace(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"profile is no longer permitted",
),
);
let result = runtime
.restore_worker_operation(&created.worker_ref)
.unwrap();
assert_eq!(result.state, WorkerRestoreState::Rejected);
assert!(result.worker.is_none());
assert_eq!(*backend.restore_count.lock().unwrap(), 0);
assert_eq!(runtime.worker_detail(&created.worker_ref).unwrap(), before);
}
#[test]
fn uncertain_restore_backend_outcome_requires_reconciliation() {
let (runtime, backend) = runtime_and_backend();
runtime.store_config_bundle(test_bundle()).unwrap();
let created = runtime
.create_worker(task_request("restore reconciliation"))
.unwrap();
runtime.stop_worker(&created.worker_ref, None).unwrap();
backend.restore_result.lock().unwrap().replace(
WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
"controller handshake outcome was not observable",
),
handle: None,
worker_state: None,
working_directory: None,
},
);
let result = runtime
.restore_worker_operation(&created.worker_ref)
.unwrap();
assert_eq!(result.state, WorkerRestoreState::ReconciliationRequired);
assert!(result.worker.is_none());
assert_eq!(
runtime.worker_detail(&created.worker_ref).unwrap().status,
WorkerStatus::Idle
);
let retry = runtime
.restore_worker_operation(&created.worker_ref)
.unwrap();
assert_eq!(retry.state, WorkerRestoreState::ReconciliationRequired);
assert_eq!(*backend.restore_count.lock().unwrap(), 1);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_automatically_restores_every_active_lifecycle_state() {
@@ -7564,7 +7862,7 @@ mod tests {
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_stops_worker_and_reports_when_execution_restore_fails() {
fn fs_store_marks_worker_for_reconciliation_when_execution_restore_is_uncertain() {
let root = fs_store_root("execution-restore-failed");
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
@@ -7598,17 +7896,13 @@ mod tests {
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1);
let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap();
assert_eq!(restored_worker.status, WorkerStatus::Stopped);
assert!(
restored
.diagnostics()
.unwrap()
.iter()
.any(
|diagnostic| diagnostic.code == "worker_execution_restore_failed"
&& diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
)
);
assert_eq!(restored_worker.status, WorkerStatus::Idle);
assert!(!restored_worker.execution_metadata_available);
let retry = restored
.restore_worker_operation(&worker.worker_ref)
.unwrap();
assert_eq!(retry.state, WorkerRestoreState::ReconciliationRequired);
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1);
let err = restored
.send_input(
&worker.worker_ref,
crates/worker-runtime/src/worker_backend.rs
+109 -28
View File
@@ -1666,15 +1666,34 @@ where
}) {
Ok(task) => task,
Err(message) => {
let cleanup = self
.cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state)
.err()
.map(|cleanup| format!("; controller cleanup failed: {cleanup}"))
.unwrap_or_default();
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
let cleanup = self.cleanup_unconnected_controller(
&handle,
&shutdown,
&tasks,
&worker_state,
);
let result = WorkerExecutionResult::errored(
operation,
format!("{message}{cleanup}"),
));
match &cleanup {
Ok(()) => message,
Err(cleanup) => {
format!("{message}; controller cleanup failed: {cleanup}")
}
},
);
return if operation == WorkerExecutionOperation::Restore {
match cleanup {
Ok(()) => WorkerExecutionSpawnResult::RolledBack(result),
Err(_) => WorkerExecutionSpawnResult::ReconciliationRequired {
result,
handle: None,
worker_state: None,
working_directory: None,
},
}
} else {
WorkerExecutionSpawnResult::Errored(result)
};
}
};
tasks.push("protocol bridge", bridge_task, true);
@@ -1687,28 +1706,66 @@ where
let mut workers = match self.workers.lock() {
Ok(workers) => workers,
Err(_) => {
let cleanup = self
.cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state)
.err()
.map(|cleanup| format!("; controller cleanup failed: {cleanup}"))
.unwrap_or_default();
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
let cleanup = self.cleanup_unconnected_controller(
&handle,
&shutdown,
&tasks,
&worker_state,
);
let result = WorkerExecutionResult::errored(
operation,
format!("worker adapter registry lock is poisoned{cleanup}"),
));
match &cleanup {
Ok(()) => "worker adapter registry lock is poisoned".to_string(),
Err(cleanup) => format!(
"worker adapter registry lock is poisoned; controller cleanup failed: {cleanup}"
),
},
);
return if operation == WorkerExecutionOperation::Restore {
match cleanup {
Ok(()) => WorkerExecutionSpawnResult::RolledBack(result),
Err(_) => WorkerExecutionSpawnResult::ReconciliationRequired {
result,
handle: None,
worker_state: None,
working_directory: None,
},
}
} else {
WorkerExecutionSpawnResult::Errored(result)
};
}
};
if workers.contains_key(&worker_ref) {
drop(workers);
let cleanup = self
.cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state)
.err()
.map(|cleanup| format!("; controller cleanup failed: {cleanup}"))
.unwrap_or_default();
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy(
let cleanup = self.cleanup_unconnected_controller(
&handle,
&shutdown,
&tasks,
&worker_state,
);
let result = WorkerExecutionResult::busy(
operation,
format!("Worker is already connected to execution backend{cleanup}"),
));
match &cleanup {
Ok(()) => "Worker is already connected to execution backend".to_string(),
Err(cleanup) => format!(
"Worker is already connected to execution backend; controller cleanup failed: {cleanup}"
),
},
);
return if operation == WorkerExecutionOperation::Restore {
match cleanup {
Ok(()) => WorkerExecutionSpawnResult::RolledBack(result),
Err(_) => WorkerExecutionSpawnResult::ReconciliationRequired {
result,
handle: None,
worker_state: None,
working_directory: None,
},
}
} else {
WorkerExecutionSpawnResult::Rejected(result)
};
}
let connected_worker_state = worker_state
.read()
@@ -1995,6 +2052,25 @@ where
)
}
fn preflight_restore(
&self,
request: &WorkerExecutionRestoreRequest,
) -> Result<(), WorkerExecutionResult> {
let workers = self.workers.lock().map_err(|_| {
WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
"worker adapter registry lock is poisoned",
)
})?;
if workers.contains_key(&request.worker_ref) {
return Err(WorkerExecutionResult::busy(
WorkerExecutionOperation::Restore,
"Worker is already connected to execution backend",
));
}
Ok(())
}
fn restore_worker(
&self,
mut request: WorkerExecutionRestoreRequest,
@@ -2076,10 +2152,15 @@ where
let controller = match restore_result {
Ok(controller) => controller,
Err(message) => {
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
message,
));
return WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
message,
),
handle: None,
worker_state: None,
working_directory: None,
};
}
};