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

feat: harden Worker restore reconciliation

Browse Source
This commit is contained in:
Hare
2026-09-18 01:16:05 +09:00
parent fd5199e8bd bd47d064b5
commit 7a2834c95c
12 changed files with 951 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/backend_runtime.rs
+2 -1
View File
@@ -18,7 +18,8 @@ pub use workspace_api::{
WorkerLaunchRuntimeOption as BackendWorkerLaunchRuntimeOption,
WorkerOperationState as BackendWorkerOperationState,
WorkerRestoreResponse as BackendWorkerRestoreResponse,
WorkerRestoreResult as BackendWorkerRestoreResult, WorkerSummary as BackendWorkerSummary,
WorkerRestoreResult as BackendWorkerRestoreResult,
WorkerRestoreState as BackendWorkerRestoreState, WorkerSummary as BackendWorkerSummary,
WorkerWorkspaceSummary as BackendWorkerWorkspaceSummary,
WorkingDirectoryCreateRequest as BackendWorkingDirectoryCreateRequest,
WorkingDirectoryCreateResponse as BackendWorkingDirectoryCreateResponse,
crates/client/src/lib.rs
+4 -4
View File
@@ -27,10 +27,10 @@ pub use backend_runtime::{
BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerLaunchOptions,
BackendWorkerLaunchProfileCandidate, BackendWorkerLaunchRuntimeOption,
BackendWorkerLaunchTarget, BackendWorkerOperationState, BackendWorkerRestoreResponse,
BackendWorkerRestoreResult, BackendWorkerSummary, BackendWorkerWorkspaceSummary,
BackendWorkingDirectorySummary, connect_backend_runtime, create_backend_worker,
get_backend_worker_launch_options, list_backend_stopped_workers, list_backend_workers,
restore_backend_worker,
BackendWorkerRestoreResult, BackendWorkerRestoreState, BackendWorkerSummary,
BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, connect_backend_runtime,
create_backend_worker, get_backend_worker_launch_options, list_backend_stopped_workers,
list_backend_workers, restore_backend_worker,
};
pub use backend_workspace::{
BackendWorkspace, BackendWorkspaceCatalogTarget, BackendWorkspaceClientError,
crates/tui/src/backend_worker_picker.rs
+10 -9
View File
@@ -3,7 +3,7 @@ use std::io;
use std::time::Duration;
use client::{
BackendRuntimeListTarget, BackendWorkerOperationState, BackendWorkerRestoreResponse,
BackendRuntimeListTarget, BackendWorkerRestoreResponse, BackendWorkerRestoreState,
BackendWorkerSummary, list_backend_stopped_workers, list_backend_workers,
restore_backend_worker,
};
@@ -110,7 +110,7 @@ pub(crate) async fn run(
}
fn restored_worker(response: BackendWorkerRestoreResponse) -> Result<BackendWorkerSummary, String> {
if response.result.state != BackendWorkerOperationState::Accepted {
if response.result.state != BackendWorkerRestoreState::Accepted {
let diagnostics = response
.result
.diagnostics
@@ -119,9 +119,10 @@ fn restored_worker(response: BackendWorkerRestoreResponse) -> Result<BackendWork
.collect::<Vec<_>>()
.join("; ");
let state = match response.result.state {
BackendWorkerOperationState::Accepted => unreachable!(),
BackendWorkerOperationState::Rejected => "rejected",
BackendWorkerOperationState::Unsupported => "unsupported",
BackendWorkerRestoreState::Accepted => unreachable!(),
BackendWorkerRestoreState::Rejected => "rejected",
BackendWorkerRestoreState::RolledBack => "rolled back",
BackendWorkerRestoreState::ReconciliationRequired => "reconciliation required",
};
return Err(if diagnostics.is_empty() {
format!("restore was {state} without a diagnostic")
@@ -494,7 +495,7 @@ mod tests {
}
fn restore_response(
state: BackendWorkerOperationState,
state: BackendWorkerRestoreState,
worker: Option<BackendWorkerSummary>,
diagnostics: Vec<BackendDiagnostic>,
) -> BackendWorkerRestoreResponse {
@@ -513,7 +514,7 @@ mod tests {
#[test]
fn rejected_restore_surfaces_diagnostic_instead_of_attaching_selected_worker() {
let error = restored_worker(restore_response(
BackendWorkerOperationState::Rejected,
BackendWorkerRestoreState::Rejected,
None,
vec![BackendDiagnostic {
code: "working_directory_not_found".to_string(),
@@ -532,7 +533,7 @@ mod tests {
#[test]
fn accepted_restore_requires_returned_worker_snapshot() {
let error = restored_worker(restore_response(
BackendWorkerOperationState::Accepted,
BackendWorkerRestoreState::Accepted,
None,
Vec::new(),
))
@@ -545,7 +546,7 @@ mod tests {
fn accepted_restore_returns_authoritative_worker_snapshot() {
let worker = worker("runtime-a", "worker-a", Some("builtin:companion"));
let restored = restored_worker(restore_response(
BackendWorkerOperationState::Accepted,
BackendWorkerRestoreState::Accepted,
Some(worker.clone()),
Vec::new(),
))
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
@@ -74,6 +74,7 @@ use workdir::{
WorkdirTransportError, WorkdirTransportErrorCode,
},
};
use workspace_api::WorkerRestoreState;
const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800;
pub const RUNTIME_HTTP_PROTOCOL_MIN_VERSION: u32 = 1;
@@ -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
+409 -120
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,22 @@ 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 +1479,54 @@ 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(),
});
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_some() {
if worker.status.is_active() {
return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
}
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;
return Ok(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_cleanup_pending",
"Worker restore cleanup or reconciliation is still pending",
));
}
state.persist_worker(&worker_ref.worker_id)?;
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 +1538,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 +1562,118 @@ 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 +2631,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 +2731,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 +3114,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 +3159,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 +3908,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 +3927,7 @@ impl InternalWorkerActivityProjection {
}
}
#[derive(Debug)]
#[derive(Debug, Clone)]
struct WorkerRecord {
worker_ref: WorkerRef,
worker_id: WorkerId,
@@ -4998,6 +5217,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 +5337,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 +7459,72 @@ 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
);
backend.restore_result.lock().unwrap().take();
let retry = runtime
.restore_worker_operation(&created.worker_ref)
.unwrap();
assert_eq!(retry.state, WorkerRestoreState::Accepted);
assert_eq!(*backend.restore_count.lock().unwrap(), 2);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_automatically_restores_every_active_lifecycle_state() {
@@ -7564,7 +7860,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,27 +7894,20 @@ 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)
)
);
let err = restored
assert_eq!(restored_worker.status, WorkerStatus::Idle);
assert!(!restored_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 failed restore"),
WorkerInput::user("after reconciled restore"),
)
.unwrap_err();
assert!(matches!(
err,
RuntimeError::WorkerExecutionUnavailable { .. }
));
.unwrap();
let _ = std::fs::remove_dir_all(root);
}
crates/worker-runtime/src/worker_backend.rs
+296 -40
View File
@@ -18,7 +18,7 @@ use crate::auth::{BACKEND_RESOURCE_FETCH_PERMISSION, RuntimeIdentityMaterial};
use crate::catalog::{
CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation,
RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectoryStatusKind,
};
use crate::config_bundle::{ConfigBundle, workspace_config_etag};
use crate::execution::{
@@ -59,7 +59,9 @@ fn next_internal_command(
.max(floor);
Ok(WorkerCommandEnvelope::new(command_id))
}
use session_store::{CombinedStore, WorkerAggregateStore, WorkerSessionStore};
use session_store::{
CombinedStore, Store, WorkerAggregateStore, WorkerMetadataStore, WorkerSessionStore,
};
#[cfg(test)]
use session_store::{FsStore, FsWorkerStore};
use tokio::runtime::Runtime;
@@ -122,6 +124,13 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String>;
async fn preflight_restore(
&self,
_request: &WorkerExecutionRestoreRequest,
) -> Result<(), String> {
Ok(())
}
async fn restore_controller(
&self,
request: WorkerExecutionRestoreRequest,
@@ -1002,6 +1011,56 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
})
}
async fn preflight_restore(
&self,
request: &WorkerExecutionRestoreRequest,
) -> Result<(), String> {
let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref);
let (mut manifest, _) = Self::restore_fallback_manifest(&worker_name)?;
bind_workspace_memory_settings(&mut manifest, &request.request)?;
let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?;
if !worker_aggregate_dir.is_dir() {
return Err("Persisted Worker aggregate metadata is unavailable".to_string());
}
if !worker_aggregate_dir.join("session").is_dir() {
return Err("Persisted Worker Session metadata is unavailable".to_string());
}
let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name)
.map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?;
let metadata = metadata_store
.read_by_name(&worker_name)
.map_err(|error| format!("failed to read Worker metadata: {error}"))?
.ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?;
if let Some(active) = metadata.active.as_ref()
&& let Some(segment_id) = active.segment_id
{
let session_store = WorkerSessionStore::new(worker_aggregate_dir.join("session"))
.map_err(|error| format!("failed to read Worker Session store: {error}"))?;
if !session_store
.exists(active.session_id, segment_id)
.map_err(|error| format!("failed to inspect Worker Session segment: {error}"))?
{
return Err("Persisted Worker Session segment is unavailable".to_string());
}
session_store
.read_all(active.session_id, segment_id)
.map_err(|error| format!("failed to read Worker Session segment: {error}"))?;
}
if request.request.workspace_api.is_some()
&& metadata
.active
.as_ref()
.is_some_and(|active| active.segment_id.is_none())
&& request.config_bundle.is_none()
{
return Err(
"Pending Workspace Worker restore requires current profile launch authority"
.to_string(),
);
}
Ok(())
}
async fn restore_controller(
&self,
request: WorkerExecutionRestoreRequest,
@@ -1310,6 +1369,7 @@ pub struct WorkerRuntimeExecutionBackend<F = ProfileRuntimeWorkerFactory> {
working_directory_materializer: Option<Arc<dyn WorkingDirectoryMaterializer>>,
runtime: Mutex<Option<Runtime>>,
workers: Mutex<HashMap<crate::identity::WorkerRef, RuntimeWorkerExecution>>,
restore_lock: Mutex<()>,
spawn_restore_timeout: Duration,
}
@@ -1338,6 +1398,7 @@ where
working_directory_materializer: None,
runtime: Mutex::new(Some(runtime)),
workers: Mutex::new(HashMap::new()),
restore_lock: Mutex::new(()),
spawn_restore_timeout: SPAWN_RESTORE_TASK_TIMEOUT,
})
}
@@ -1569,6 +1630,39 @@ where
.unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message))
}
fn retain_uncertain_unconnected_controller(
&self,
worker_ref: &crate::identity::WorkerRef,
handle: WorkerHandle,
shutdown: Arc<tokio::sync::Mutex<Option<worker::ShutdownReceiver>>>,
tasks: RuntimeExecutionTaskScope,
worker_state: Arc<RwLock<protocol::WorkerStateSnapshot>>,
workspace_client: Option<Arc<dyn WorkspaceClient>>,
) -> Result<WorkerExecutionHandle, String> {
let mut workers = self
.workers
.lock()
.map_err(|_| "worker adapter registry lock is poisoned".to_string())?;
if workers.contains_key(worker_ref) {
return Err("Worker already has a retained execution handle".to_string());
}
workers.insert(
worker_ref.clone(),
RuntimeWorkerExecution {
handle,
shutdown,
shutdown_requested: Arc::new(AtomicBool::new(false)),
tasks,
worker_state,
workspace_client,
},
);
Ok(WorkerExecutionHandle::new(
worker_ref.clone(),
self.backend_id(),
))
}
fn cleanup_unconnected_controller(
&self,
handle: &WorkerHandle,
@@ -1618,11 +1712,19 @@ where
working_directory: Option<WorkingDirectoryBinding>,
workspace_client: Option<Arc<dyn WorkspaceClient>>,
) -> WorkerExecutionSpawnResult {
let worker_state = Arc::new(RwLock::new(handle.shared_state.snapshot()));
#[cfg(feature = "ws-server")]
let streams = subscribe_worker_protocol_session(&handle);
#[cfg(feature = "ws-server")]
let worker_state_snapshot = match &streams.snapshot_event {
Event::Snapshot { state, .. } => state.clone(),
_ => unreachable!("Worker protocol subscription snapshot must be a snapshot event"),
};
#[cfg(not(feature = "ws-server"))]
let worker_state_snapshot = handle.shared_state.snapshot();
let worker_state = Arc::new(RwLock::new(worker_state_snapshot));
let tasks = RuntimeExecutionTaskScope::new(controller_task);
#[cfg(feature = "ws-server")]
{
let streams = subscribe_worker_protocol_session(&handle);
let mut events = streams.events;
let mut entry_events = streams.log_entries;
let bridge_worker_state = worker_state.clone();
@@ -1666,15 +1768,50 @@ 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(_) => {
let retained_state =
worker_state.read().ok().map(|state| state.clone());
let retained_working_directory =
working_directory.as_ref().map(|binding| binding.status());
let retained_handle = self
.retain_uncertain_unconnected_controller(
&worker_ref,
handle,
shutdown,
tasks,
worker_state,
workspace_client,
)
.ok();
WorkerExecutionSpawnResult::ReconciliationRequired {
result,
handle: retained_handle,
worker_state: retained_state,
working_directory: retained_working_directory,
}
}
}
} else {
WorkerExecutionSpawnResult::Errored(result)
};
}
};
tasks.push("protocol bridge", bridge_task, true);
@@ -1684,31 +1821,40 @@ where
let _ = bridge_context;
}
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(
operation,
format!("worker adapter registry lock is poisoned{cleanup}"),
));
}
};
let mut workers = self
.workers
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if workers.contains_key(&worker_ref) {
let existing_worker_state = workers
.get(&worker_ref)
.and_then(|worker| worker.worker_state.read().ok().map(|state| state.clone()));
let existing_handle = WorkerExecutionHandle::new(worker_ref.clone(), self.backend_id());
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: Some(existing_handle),
worker_state: existing_worker_state,
working_directory: None,
},
}
} else {
WorkerExecutionSpawnResult::Rejected(result)
};
}
let connected_worker_state = worker_state
.read()
@@ -1884,11 +2030,15 @@ where
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
let _start_guard = self
.restore_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if self
.workers
.lock()
.map(|workers| workers.contains_key(&request.worker_ref))
.unwrap_or(false)
.unwrap_or_else(|poisoned| poisoned.into_inner())
.contains_key(&request.worker_ref)
{
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy(
WorkerExecutionOperation::Spawn,
@@ -1995,10 +2145,111 @@ where
)
}
fn preflight_restore(
&self,
request: &WorkerExecutionRestoreRequest,
) -> Result<(), WorkerExecutionResult> {
let workers = self
.workers
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if workers.contains_key(&request.worker_ref) {
return Err(WorkerExecutionResult::busy(
WorkerExecutionOperation::Restore,
"Worker is already connected to execution backend",
));
}
drop(workers);
if let Some(previous) = request.previous_working_directory.as_ref() {
let materializer = self
.working_directory_materializer
.as_ref()
.ok_or_else(|| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir binding cannot be restored by this Runtime",
)
})?;
let current = materializer
.working_directory_status(&previous.summary.working_directory_id)
.map_err(|message| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!("Persisted Worker Workdir is unavailable: {message}"),
)
})?;
if current.summary.status != WorkingDirectoryStatusKind::Active {
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir is not active",
));
}
}
if request.previous_working_directory.is_none()
&& request.request.working_directory_request.is_some()
{
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir allocation is unavailable",
));
}
if request.previous_working_directory.is_none()
&& request.request.working_directory.is_some()
&& self.working_directory_materializer.is_none()
{
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir claim cannot be restored by this Runtime",
));
}
if let Some(workspace_api) = request.request.workspace_api.as_ref() {
let Some(scope) = request.workspace_scope.as_ref() else {
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Workspace restore authorization is unavailable",
));
};
if scope.workspace_id != workspace_api.workspace_id {
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Workspace restore authorization does not match Worker authority",
));
}
}
let factory = Arc::clone(&self.factory);
let request = request.clone();
self.run_on_adapter_runtime(async move { factory.preflight_restore(&request).await })
.map_err(|message| {
WorkerExecutionResult::rejected(WorkerExecutionOperation::Restore, message)
})?;
Ok(())
}
fn restore_worker(
&self,
mut request: WorkerExecutionRestoreRequest,
) -> WorkerExecutionSpawnResult {
// Serialize backend restore starts so duplicate detection is completed
// before any controller or bridge side effect. Runtime also serializes
// per Worker; this guard protects direct/concurrent backend callers.
let _restore_guard = self
.restore_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let workers = self
.workers
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if workers.contains_key(&request.worker_ref) {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy(
WorkerExecutionOperation::Restore,
"Worker is already connected to execution backend",
));
}
drop(workers);
let working_directory = match request.previous_working_directory.clone() {
Some(status) => {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
@@ -2076,10 +2327,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,
};
}
};
crates/workspace-api/src/lib.rs
+58 -1
View File
@@ -2327,15 +2327,36 @@ pub enum WorkerOperationState {
Rejected,
}
/// Cross-layer outcome of an explicit Worker restore operation.
///
/// `Rejected` is reserved for read-only preflight failures. Once live restore
/// work starts, failure is either a confirmed `RolledBack` operation or a
/// `ReconciliationRequired` result whose commit state must be reread/retried.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum WorkerRestoreState {
Accepted,
Rejected,
RolledBack,
ReconciliationRequired,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerRestoreResult {
pub state: WorkerOperationState,
pub state: WorkerRestoreState,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(optional = nullable))]
pub worker: Option<WorkerSummary>,
#[serde(default)]
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerRestoreResponse {
pub workspace_id: String,
pub runtime_id: String,
@@ -3482,6 +3503,42 @@ mod workdir_typescript_tests {
mod tests {
use super::*;
#[test]
fn worker_restore_state_round_trips_and_rejects_unknown_variants() {
for (state, wire) in [
(WorkerRestoreState::Accepted, "accepted"),
(WorkerRestoreState::Rejected, "rejected"),
(WorkerRestoreState::RolledBack, "rolled_back"),
(
WorkerRestoreState::ReconciliationRequired,
"reconciliation_required",
),
] {
assert_eq!(
serde_json::to_value(state).unwrap(),
serde_json::json!(wire)
);
assert_eq!(
serde_json::from_value::<WorkerRestoreState>(serde_json::json!(wire)).unwrap(),
state
);
}
assert!(
serde_json::from_value::<WorkerRestoreState>(serde_json::json!("unknown")).is_err()
);
}
#[test]
fn worker_restore_result_rejects_unknown_fields() {
let result = serde_json::from_value::<WorkerRestoreResult>(serde_json::json!({
"state": "rejected",
"worker": null,
"diagnostics": [],
"unexpected": true
}));
assert!(result.is_err());
}
#[test]
fn historical_http_repository_source_kind_decodes_as_invalid_evidence() {
let source: RepositorySource = serde_json::from_value(serde_json::json!({
crates/workspace-server/src/hosts.rs
+72 -20
View File
@@ -52,9 +52,10 @@ use worker_runtime::http_server::{
RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
RuntimeHttpWorkerRestoreResponse, RuntimeHttpWorkerWorkspaceApiRequest,
RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse,
RuntimeHttpWorkingDirectoryResponse, RuntimeHttpWorkspacePromptProjectionRequest,
RuntimeHttpWorkspacePromptProjectionResponse,
};
use worker_runtime::identity::{
RuntimeWorkerRef, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef,
@@ -378,8 +379,9 @@ pub(crate) fn workspace_worker_summary(
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerRestoreResult {
pub state: WorkerOperationState,
pub state: workspace_api::WorkerRestoreState,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
@@ -882,7 +884,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult {
WorkerRestoreResult {
state: WorkerOperationState::Unsupported,
state: workspace_api::WorkerRestoreState::Rejected,
worker: None,
diagnostics: vec![diagnostic(
"worker_restore_unsupported",
@@ -2336,7 +2338,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult {
let Some(worker_ref) = self.worker_ref(worker_id) else {
return WorkerRestoreResult {
state: WorkerOperationState::Rejected,
state: workspace_api::WorkerRestoreState::Rejected,
worker: None,
diagnostics: vec![diagnostic(
"embedded_worker_id_invalid",
@@ -2345,14 +2347,22 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
)],
};
};
match self.runtime.restore_worker(&worker_ref) {
Ok(detail) => WorkerRestoreResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(detail)),
diagnostics: Vec::new(),
},
match self.runtime.restore_worker_operation(&worker_ref) {
Ok(result) => {
let diagnostics = match (result.reason_code, result.message) {
(Some(code), Some(message)) => {
vec![diagnostic(code, DiagnosticSeverity::Warning, message)]
}
_ => Vec::new(),
};
WorkerRestoreResult {
state: result.state,
worker: result.worker.map(|detail| self.map_worker_detail(detail)),
diagnostics,
}
}
Err(err) => WorkerRestoreResult {
state: WorkerOperationState::Rejected,
state: workspace_api::WorkerRestoreState::ReconciliationRequired,
worker: None,
diagnostics: vec![embedded_runtime_diagnostic(&err)],
},
@@ -4164,17 +4174,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult {
match self.post_json::<_, RuntimeHttpWorkerResponse>(
match self.post_json::<_, RuntimeHttpWorkerRestoreResponse>(
&format!("/v1/workers/{worker_id}/restore"),
&serde_json::json!({}),
) {
Ok(response) => WorkerRestoreResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response.worker)),
diagnostics: Vec::new(),
},
Ok(response) => {
let diagnostics = match (response.reason_code, response.message) {
(Some(code), Some(message)) => {
vec![diagnostic(code, DiagnosticSeverity::Warning, message)]
}
_ => Vec::new(),
};
WorkerRestoreResult {
state: response.state,
worker: response.worker.map(|worker| self.map_worker_detail(worker)),
diagnostics,
}
}
Err(diagnostic) => WorkerRestoreResult {
state: WorkerOperationState::Rejected,
// A transport/protocol failure cannot prove that the Runtime
// rejected before side effects. Preserve uncertainty.
state: workspace_api::WorkerRestoreState::ReconciliationRequired,
worker: None,
diagnostics: vec![diagnostic],
},
@@ -5428,6 +5448,38 @@ mod tests {
use std::sync::{Arc, Mutex};
use std::thread;
#[test]
fn remote_restore_transport_failure_requires_reconciliation() {
let provider = RemoteWorkerRuntime::new(
RemoteRuntimeConfig {
runtime_id: "runtime-remote".to_string(),
workspace_id: Some("workspace-a".to_string()),
display_name: "Remote Runtime".to_string(),
base_url: "http://127.0.0.1:9".to_string(),
bearer_token: None,
workspace_authorization: None,
strict_public_egress: false,
cached_worker_creation_available: true,
cached_os: "linux".to_string(),
cached_arch: "x86_64".to_string(),
cached_status: "ready".to_string(),
timeout: Duration::from_millis(100),
},
"workspace-a".to_string(),
"http://127.0.0.1:3000".to_string(),
)
.unwrap();
let result = provider.restore_worker("worker-transport-uncertain");
assert_eq!(
result.state,
workspace_api::WorkerRestoreState::ReconciliationRequired
);
assert!(result.worker.is_none());
assert_eq!(result.diagnostics.len(), 1);
}
#[test]
fn embedded_delete_persistence_failure_diagnostic_is_bounded_and_path_free() {
let diagnostic =
crates/workspace-server/src/server.rs
+10 -10
View File
@@ -2814,7 +2814,7 @@ impl WorkspaceApi {
.map_err(|error| error.into_error())?;
if binding.state != WorkerOperationState::Accepted {
return Ok(WorkerRestoreResult {
state: binding.state,
state: workspace_api::WorkerRestoreState::Rejected,
worker: binding.worker,
diagnostics: binding.diagnostics,
});
@@ -10422,7 +10422,7 @@ async fn scoped_start_workspace_orchestrator(
.runtime
.restore_worker(&existing.worker)
.map_err(|error| error.into_error())?;
if restored.state != WorkerOperationState::Accepted {
if restored.state != workspace_api::WorkerRestoreState::Accepted {
return Err(ApiError::with_diagnostics(
Error::RuntimeOperationFailed {
runtime_id: existing.worker.runtime_id.clone(),
@@ -12501,7 +12501,7 @@ async fn scoped_restore_runtime_worker(
runtime_id: runtime_id.clone(),
worker_id: worker_id.clone(),
result: workspace_api::WorkerRestoreResult {
state: workspace_api::WorkerOperationState::Accepted,
state: workspace_api::WorkerRestoreState::Accepted,
worker: Some(worker),
diagnostics: Vec::new(),
},
@@ -15213,7 +15213,7 @@ async fn restore_runtime_worker(
runtime_id: runtime_id.clone(),
worker_id: worker_id.clone(),
result: workspace_api::WorkerRestoreResult {
state: result.state.into(),
state: result.state,
worker: projected_worker,
diagnostics: result.diagnostics.into_iter().map(Into::into).collect(),
},
@@ -20722,7 +20722,7 @@ mod tests {
}
#[tokio::test]
async fn production_profile_backend_rejects_unrecoverable_pending_orchestrator_restore() {
async fn production_profile_backend_preflight_rejects_pending_orchestrator_restore() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let config = test_server_config(workspace.path());
@@ -20772,11 +20772,11 @@ mod tests {
}),
)
.await
.expect_err("pending Workspace Orchestrator restore without durable Prompt must fail");
.expect_err(
"pending Workspace Orchestrator restore without durable Prompt must be rejected",
);
assert!(
format!("{error:?}").contains(
"pending Workspace Worker restore requires operation-owned launch material"
),
format!("{error:?}").contains("worker_restore_preflight_rejected"),
"unexpected restore error: {error:?}"
);
}
@@ -25277,7 +25277,7 @@ mod tests {
assert_eq!(retried_restore.worker_id, first_worker.worker.worker_id);
assert_eq!(
retried_restore.result.state,
workspace_api::WorkerOperationState::Accepted
workspace_api::WorkerRestoreState::Accepted
);
let restored_assignment = api
.store
docs/report/2026-08-19-coder-toolchain-missing-rustfmt.md
+17
View File
@@ -0,0 +1,17 @@
# Ticket validation image lacks `rustfmt`
While implementing T-613, `cargo fmt --all` failed before reading project sources because the active Rust toolchain does not provide the `rustfmt` component. The standalone `rustfmt` binary is also absent.
Observed commands:
```text
$ cargo fmt --all
error: 'cargo-fmt' is not installed for the toolchain '1.90.0-x86_64-unknown-linux-gnu'.
$ rustfmt --version
bash: rustfmt: command not found
```
This prevents Workers from executing the repository-required formatting check or automatically normalizing Rust changes. The Coder can still run compile/tests and `git diff --check`, but formatting evidence remains weaker and review may surface avoidable formatting differences.
Suggested improvement: include the `rustfmt` component in the standard coding Workdir/toolchain image and verify `cargo fmt --all -- --check` during image qualification.
web/workspace/src/lib/workspace/sidebar/types.ts
+13
View File
@@ -112,6 +112,19 @@ export type Worker = {
export type WorkerOperationState = "accepted" | "unsupported" | "rejected";
/** Typed result of restore across Runtime, Workspace, Web, and TUI clients. */
export type WorkerRestoreState =
| "accepted"
| "rejected"
| "rolled_back"
| "reconciliation_required";
export type WorkerRestoreResult = {
state: WorkerRestoreState;
worker?: WorkerSummary | null;
diagnostics: Diagnostic[];
};
export type WorkerLaunchRuntimeOption = SharedWorkerLaunchRuntimeOption;
export type WorkerLaunchProfileCandidate = SharedWorkerLaunchProfileCandidate;
export type WorkingDirectoryRepositoryOption =