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

runtime: simplify worker lifecycle

Browse Source
This commit is contained in:
Hare
2026-07-24 11:07:04 +09:00
parent e0b092dbc8
commit d3d2b28adc
5 changed files with 279 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/execution.rs
+142 -79
View File
@@ -1,4 +1,4 @@
use crate::catalog::{CreateWorkerRequest, WorkingDirectoryRequest, WorkingDirectoryStatus};
use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus};
use crate::config_bundle::ConfigBundle;
use crate::error::RuntimeError;
use crate::identity::WorkerRef;
@@ -11,24 +11,29 @@ use serde::{Deserialize, Serialize};
use std::fmt;
use std::sync::Arc;
/// Coarse execution attachment visible through Worker catalog/detail responses.
/// Persisted execution lifecycle visible through Worker catalog/detail responses.
///
/// This deliberately does not expose backend handles, process paths, sockets,
/// credentials, session files, or manifest paths. It only says whether Runtime
/// has an execution backend attached for the Worker.
/// This is intentionally a worker lifecycle projection, not a transport/backend
/// handle state. Runtime restart boundaries invalidate live handles, so a
/// persisted `alive` worker is restored on startup; restore deferral keeps the
/// worker `stopped`, while structural restore failure marks it `corrupted`.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkerExecutionBackendKind {
/// Restoreable persisted state exists, but no live execution handle is attached.
#[default]
Unconnected,
/// A durable execution binding was restored, but no live handle was recovered.
Stale,
Connected,
#[serde(alias = "unconnected", alias = "stale")]
Stopped,
/// A live execution handle is currently attached. Legacy `connected` maps here.
#[serde(alias = "connected")]
Alive,
/// Persisted execution state is structurally invalid and cannot be restored.
Corrupted,
}
/// Durable, non-authority execution binding projection.
///
/// This records only enough identity to diagnose stale mappings after restore.
/// This records only enough identity to restore through the same backend kind.
/// It is not a live handle and must not contain sockets, paths, credentials, or
/// provider-private authority.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -49,12 +54,11 @@ impl WorkerExecutionBindingIdentity {
#[serde(rename_all = "snake_case")]
pub enum WorkerExecutionRunState {
#[default]
Unconnected,
Stopped,
Idle,
Busy,
Rejected,
Errored,
Stopped,
}
/// Execution operation that produced a result.
@@ -69,7 +73,18 @@ pub enum WorkerExecutionOperation {
Cancel,
}
/// Typed execution result class.
/// Typed execution result class. Results are transient operation outcomes and
/// are not persisted as Worker lifecycle authority.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerExecutionResult {
pub operation: WorkerExecutionOperation,
pub outcome: WorkerExecutionOutcome,
pub run_state: WorkerExecutionRunState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
/// Backend result class for a Worker execution operation.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkerExecutionOutcome {
@@ -80,16 +95,6 @@ pub enum WorkerExecutionOutcome {
Unsupported,
}
/// Backend result for a Worker execution operation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerExecutionResult {
pub operation: WorkerExecutionOperation,
pub outcome: WorkerExecutionOutcome,
pub run_state: WorkerExecutionRunState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
impl WorkerExecutionResult {
pub fn accepted(
operation: WorkerExecutionOperation,
@@ -116,7 +121,7 @@ impl WorkerExecutionResult {
Self {
operation,
outcome: WorkerExecutionOutcome::Rejected,
run_state: WorkerExecutionRunState::Rejected,
run_state: WorkerExecutionRunState::Stopped,
message: Some(message.into()),
}
}
@@ -134,7 +139,7 @@ impl WorkerExecutionResult {
Self {
operation,
outcome: WorkerExecutionOutcome::Unsupported,
run_state: WorkerExecutionRunState::Rejected,
run_state: WorkerExecutionRunState::Stopped,
message: Some(message.into()),
}
}
@@ -159,28 +164,31 @@ pub struct WorkerExecutionStatus {
pub binding: Option<WorkerExecutionBindingIdentity>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryStatus>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_result: Option<WorkerExecutionResult>,
}
impl WorkerExecutionStatus {
pub fn unconnected() -> Self {
pub fn stopped() -> Self {
Self::default()
}
pub fn connected(run_state: WorkerExecutionRunState) -> Self {
pub fn alive(run_state: WorkerExecutionRunState) -> Self {
Self {
backend: WorkerExecutionBackendKind::Connected,
backend: WorkerExecutionBackendKind::Alive,
run_state,
binding: None,
working_directory: None,
last_result: None,
}
}
pub fn stale(mut previous: Self) -> Self {
previous.backend = WorkerExecutionBackendKind::Stale;
previous.run_state = WorkerExecutionRunState::Unconnected;
pub fn stopped_from(mut previous: Self) -> Self {
previous.backend = WorkerExecutionBackendKind::Stopped;
previous.run_state = WorkerExecutionRunState::Stopped;
previous
}
pub fn corrupted(mut previous: Self) -> Self {
previous.backend = WorkerExecutionBackendKind::Corrupted;
previous.run_state = WorkerExecutionRunState::Errored;
previous
}
@@ -196,7 +204,6 @@ impl WorkerExecutionStatus {
pub fn with_result(mut self, result: WorkerExecutionResult) -> Self {
self.run_state = result.run_state;
self.last_result = Some(result);
self
}
}
@@ -266,13 +273,20 @@ impl WorkerExecutionContext {
&self.worker_ref
}
/// Publish a protocol event into the Runtime observation bus.
#[cfg(feature = "ws-server")]
pub fn publish_observation(
&self,
payload: protocol::Event,
) -> Result<WorkerObservationEvent, RuntimeError> {
(self.observation_publisher)(self.worker_ref.clone(), payload)
}
#[cfg(feature = "ws-server")]
pub fn publish_protocol_event(
&self,
payload: protocol::Event,
) -> Result<WorkerObservationEvent, RuntimeError> {
(self.observation_publisher)(self.worker_ref.clone(), payload)
self.publish_observation(payload)
}
}
@@ -284,32 +298,29 @@ impl fmt::Debug for WorkerExecutionContext {
}
}
/// Spawn/initialization request passed to an execution backend.
/// Request passed to a [`WorkerExecutionBackend`] when spawning a Worker.
#[derive(Clone, Debug)]
pub struct WorkerExecutionSpawnRequest {
pub worker_ref: WorkerRef,
pub request: CreateWorkerRequest,
pub request: crate::catalog::CreateWorkerRequest,
pub context: WorkerExecutionContext,
pub working_directory: Option<WorkingDirectoryBinding>,
pub config_bundle: Option<ConfigBundle>,
}
/// Restore request passed to an execution backend for a persisted Runtime Worker.
///
/// The persisted execution status is a restore hint, not a live handle. Backends
/// must create a fresh controller/handle before returning `Connected`.
/// Request passed to a [`WorkerExecutionBackend`] when restoring a persisted Worker.
#[derive(Clone, Debug)]
pub struct WorkerExecutionRestoreRequest {
pub worker_ref: WorkerRef,
pub request: CreateWorkerRequest,
pub request: crate::catalog::CreateWorkerRequest,
pub context: WorkerExecutionContext,
pub previous_execution: WorkerExecutionStatus,
pub working_directory: Option<WorkingDirectoryBinding>,
pub config_bundle: Option<ConfigBundle>,
}
/// Result of backend Worker spawn/initialization.
#[derive(Clone, Debug, PartialEq, Eq)]
/// Backend outcome for Worker spawn/restore operations.
#[derive(Clone, Debug)]
pub enum WorkerExecutionSpawnResult {
Connected {
handle: WorkerExecutionHandle,
@@ -320,11 +331,20 @@ pub enum WorkerExecutionSpawnResult {
Errored(WorkerExecutionResult),
}
/// Backend boundary for Worker execution.
///
/// Runtime owns Worker catalog, protocol observation, and lifecycle state. A
/// backend owns concrete execution. The default Runtime has no backend, so input
/// to those Workers is rejected instead of producing providerless responses.
impl WorkerExecutionSpawnResult {
pub fn connected(
handle: WorkerExecutionHandle,
run_state: WorkerExecutionRunState,
working_directory: Option<WorkingDirectoryStatus>,
) -> Self {
Self::Connected {
handle,
run_state,
working_directory,
}
}
}
pub trait WorkerExecutionBackend: Send + Sync + 'static {
fn backend_id(&self) -> &str;
@@ -393,6 +413,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
)
}
fn worker_completions(
&self,
_handle: &WorkerExecutionHandle,
_kind: protocol::CompletionKind,
_prefix: &str,
) -> Vec<protocol::CompletionEntry> {
Vec::new()
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::unsupported(
WorkerExecutionOperation::Stop,
@@ -411,32 +440,30 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
fn worker_snapshot(&self, _handle: &WorkerExecutionHandle) -> Option<protocol::Event> {
None
}
fn worker_completions(
&self,
_handle: &WorkerExecutionHandle,
_kind: protocol::CompletionKind,
_prefix: &str,
) -> Vec<protocol::CompletionEntry> {
Vec::new()
}
}
#[derive(Clone)]
pub(crate) struct WorkerExecutionBackendRef {
id: String,
backend_id: String,
backend: Arc<dyn WorkerExecutionBackend>,
}
impl WorkerExecutionBackendRef {
pub(crate) fn new(backend: Arc<dyn WorkerExecutionBackend>) -> Result<Self, RuntimeError> {
let id = backend.backend_id().trim().to_string();
if id.is_empty() {
let backend_id = backend.backend_id().to_string();
if backend_id.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"execution backend id must not be empty".to_string(),
));
}
Ok(Self { id, backend })
Ok(Self {
backend_id,
backend,
})
}
pub(crate) fn backend_id(&self) -> &str {
&self.backend_id
}
pub(crate) fn spawn_worker(
@@ -446,12 +473,6 @@ impl WorkerExecutionBackendRef {
self.backend.spawn_worker(request)
}
#[cfg(feature = "fs-store")]
pub(crate) fn backend_id(&self) -> &str {
&self.id
}
#[cfg(feature = "fs-store")]
pub(crate) fn restore_worker(
&self,
request: WorkerExecutionRestoreRequest,
@@ -500,14 +521,6 @@ impl WorkerExecutionBackendRef {
self.backend.dispatch_method(handle, method)
}
pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
self.backend.stop_worker(handle)
}
pub(crate) fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
self.backend.cancel_worker(handle)
}
#[cfg(feature = "ws-server")]
pub(crate) fn worker_snapshot(
&self,
@@ -524,12 +537,62 @@ impl WorkerExecutionBackendRef {
) -> Vec<protocol::CompletionEntry> {
self.backend.worker_completions(handle, kind, prefix)
}
pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
self.backend.stop_worker(handle)
}
pub(crate) fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
self.backend.cancel_worker(handle)
}
}
impl fmt::Debug for WorkerExecutionBackendRef {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("WorkerExecutionBackendRef")
.field("id", &self.id)
.field("backend_id", &self.backend_id)
.finish_non_exhaustive()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn execution_backend_kind_accepts_legacy_values() {
let connected: WorkerExecutionStatus = serde_json::from_value(json!({
"backend": "connected",
"run_state": "idle",
"binding": { "backend_id": "worker-crate" }
}))
.unwrap();
assert_eq!(connected.backend, WorkerExecutionBackendKind::Alive);
let stale: WorkerExecutionStatus = serde_json::from_value(json!({
"backend": "stale",
"run_state": "stopped",
"binding": { "backend_id": "worker-crate" }
}))
.unwrap();
assert_eq!(stale.backend, WorkerExecutionBackendKind::Stopped);
let unconnected: WorkerExecutionStatus = serde_json::from_value(json!({
"backend": "unconnected",
"run_state": "stopped"
}))
.unwrap();
assert_eq!(unconnected.backend, WorkerExecutionBackendKind::Stopped);
}
#[test]
fn execution_status_serializes_without_last_result() {
let status = WorkerExecutionStatus::alive(WorkerExecutionRunState::Idle).with_result(
WorkerExecutionResult::rejected(WorkerExecutionOperation::Input, "transient"),
);
let serialized = serde_json::to_value(status).unwrap();
assert_eq!(serialized["backend"], "alive");
assert!(serialized.get("last_result").is_none());
}
}
crates/worker-runtime/src/fs_store.rs
+1 -1
View File
@@ -473,7 +473,7 @@ struct WorkerSnapshot {
worker_id: WorkerId,
status: WorkerStatus,
request: CreateWorkerRequest,
#[serde(default = "WorkerExecutionStatus::unconnected")]
#[serde(default = "WorkerExecutionStatus::stopped")]
execution: WorkerExecutionStatus,
last_event_id: u64,
}
crates/worker-runtime/src/runtime.rs
+63 -46
View File
@@ -357,7 +357,7 @@ impl Runtime {
worker_id: worker_id.clone(),
status: WorkerStatus::Running,
request: request.clone(),
execution: WorkerExecutionStatus::unconnected(),
execution: WorkerExecutionStatus::stopped(),
execution_handle: None,
last_event_id: event_id,
};
@@ -471,14 +471,13 @@ impl Runtime {
match (backend, handle) {
(Some(backend), Some(handle)) => (backend, handle),
_ => {
let result = WorkerExecutionResult::rejected(
WorkerExecutionOperation::Input,
"worker has no execution backend",
);
let worker = state.worker_mut(worker_ref)?;
let mut execution = WorkerExecutionStatus::unconnected().with_result(result);
execution.binding = worker.execution.binding.clone();
worker.execution = execution;
worker.execution =
if worker.execution.backend == WorkerExecutionBackendKind::Corrupted {
worker.execution.clone()
} else {
WorkerExecutionStatus::stopped_from(worker.execution.clone())
};
state.persist_worker(&worker_ref.worker_id)?;
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id.clone(),
@@ -510,11 +509,10 @@ impl Runtime {
let worker = state.worker_mut(worker_ref)?;
worker.last_event_id = event_id;
worker.execution = WorkerExecutionStatus {
backend: WorkerExecutionBackendKind::Connected,
backend: WorkerExecutionBackendKind::Alive,
run_state: dispatch_result.run_state,
binding: worker.execution.binding.clone(),
working_directory: worker.execution.working_directory.clone(),
last_result: Some(dispatch_result),
};
let status = worker.status;
@@ -592,14 +590,13 @@ impl Runtime {
match (backend, handle) {
(Some(backend), Some(handle)) => (backend, handle),
_ => {
let result = WorkerExecutionResult::rejected(
WorkerExecutionOperation::ProtocolMethod,
"worker has no execution backend",
);
let worker = state.worker_mut(worker_ref)?;
let mut execution = WorkerExecutionStatus::unconnected().with_result(result);
execution.binding = worker.execution.binding.clone();
worker.execution = execution;
worker.execution =
if worker.execution.backend == WorkerExecutionBackendKind::Corrupted {
worker.execution.clone()
} else {
WorkerExecutionStatus::stopped_from(worker.execution.clone())
};
state.persist_worker(&worker_ref.worker_id)?;
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id.clone(),
@@ -638,7 +635,7 @@ impl Runtime {
let binding = WorkerExecutionBindingIdentity::from_handle(&handle);
let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = Some(handle);
let mut execution = WorkerExecutionStatus::connected(run_state).with_binding(binding);
let mut execution = WorkerExecutionStatus::alive(run_state).with_binding(binding);
if let Some(status) = working_directory {
execution = execution.with_working_directory(status);
}
@@ -669,11 +666,10 @@ impl Runtime {
let mut state = self.lock()?;
let worker = state.worker_mut(worker_ref)?;
worker.execution = WorkerExecutionStatus {
backend: WorkerExecutionBackendKind::Connected,
backend: WorkerExecutionBackendKind::Alive,
run_state: result.run_state,
binding: worker.execution.binding.clone(),
working_directory: worker.execution.working_directory.clone(),
last_result: Some(result),
};
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
@@ -1048,6 +1044,24 @@ impl Runtime {
let candidates = {
let mut state = self.lock()?;
let Some(backend) = state.execution_backend.clone() else {
let worker_ids: Vec<_> = state.workers.keys().cloned().collect();
for worker_id in worker_ids {
let Some(worker) = state.workers.get(&worker_id) else {
continue;
};
if worker.execution.backend != WorkerExecutionBackendKind::Alive {
continue;
}
let worker_ref = worker.worker_ref.clone();
let worker = state.worker_mut(&worker_ref)?;
worker.execution =
if worker.execution.backend == WorkerExecutionBackendKind::Corrupted {
worker.execution.clone()
} else {
WorkerExecutionStatus::stopped_from(worker.execution.clone())
};
state.persist_worker(&worker_ref.worker_id)?;
}
return Ok(());
};
let backend_id = backend.backend_id().to_string();
@@ -1059,7 +1073,10 @@ impl Runtime {
};
if !worker.status.is_active()
|| worker.execution_handle.is_some()
|| worker.execution.backend != WorkerExecutionBackendKind::Stale
|| !matches!(
worker.execution.backend,
WorkerExecutionBackendKind::Alive | WorkerExecutionBackendKind::Stopped
)
|| worker
.execution
.binding
@@ -1152,7 +1169,7 @@ impl Runtime {
{
let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = Some(handle.clone());
let mut execution = WorkerExecutionStatus::connected(run_state)
let mut execution = WorkerExecutionStatus::alive(run_state)
.with_binding(WorkerExecutionBindingIdentity::from_handle(&handle));
if let Some(status) = working_directory {
execution = execution.with_working_directory(status);
@@ -1269,22 +1286,21 @@ impl RuntimeState {
let mut diagnostics = persisted.diagnostics;
let mut next_diagnostic_id = persisted.next_diagnostic_id;
for (worker_id, worker) in persisted.workers {
let execution = if worker.execution.binding.is_some()
&& worker.execution.backend == WorkerExecutionBackendKind::Connected
let execution = if worker.execution.backend == WorkerExecutionBackendKind::Alive
&& worker.execution.binding.is_none()
{
let stale = WorkerExecutionStatus::stale(worker.execution);
diagnostics.push(RuntimeDiagnostic {
id: next_diagnostic_id,
severity: DiagnosticSeverity::Warning,
code: "worker_execution_mapping_stale".to_string(),
severity: DiagnosticSeverity::Error,
code: "worker_execution_binding_missing".to_string(),
message: format!(
"worker {} has persisted execution binding identity but no live execution handle was restored",
"worker {} was persisted as alive but has no execution binding identity",
worker.worker_id
),
worker_ref: Some(worker.worker_ref.clone()),
});
next_diagnostic_id += 1;
stale
WorkerExecutionStatus::corrupted(worker.execution)
} else {
worker.execution
};
@@ -1596,9 +1612,7 @@ impl RuntimeState {
});
let worker = self.worker_mut(worker_ref)?;
worker.execution_handle = None;
let mut execution = WorkerExecutionStatus::stale(worker.execution.clone());
execution.last_result = Some(result);
worker.execution = execution;
worker.execution = WorkerExecutionStatus::corrupted(worker.execution.clone());
self.persist_runtime_snapshot()?;
self.persist_worker(&worker_ref.worker_id)?;
Ok(())
@@ -2623,7 +2637,7 @@ mod tests {
assert_eq!(restored_worker.status, WorkerStatus::Stopped);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Stale
WorkerExecutionBackendKind::Stopped
);
assert_eq!(
restored_worker
@@ -2633,16 +2647,6 @@ mod tests {
.map(|binding| binding.backend_id.as_str()),
Some("test-execution-backend")
);
assert!(
restored
.diagnostics()
.unwrap()
.iter()
.any(
|diagnostic| diagnostic.code == "worker_execution_mapping_stale"
&& diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
)
);
#[cfg(feature = "ws-server")]
{
let observations = restored
@@ -2708,6 +2712,19 @@ mod tests {
.unwrap();
drop(runtime);
let backendless = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
})
.unwrap();
let stopped_worker = backendless.worker_detail(&worker.worker_ref).unwrap();
assert_eq!(
stopped_worker.execution.backend,
WorkerExecutionBackendKind::Stopped
);
drop(backendless);
let restoring_backend = Arc::new(TestExecutionBackend::default());
let restored = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
@@ -2724,7 +2741,7 @@ mod tests {
assert_eq!(restored_worker.status, WorkerStatus::Running);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Connected
WorkerExecutionBackendKind::Alive
);
assert!(restored_worker.execution.binding.is_some());
restored
@@ -2748,7 +2765,7 @@ mod tests {
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_keeps_worker_stale_when_execution_restore_fails() {
fn fs_store_marks_worker_corrupted_when_execution_restore_fails() {
let root = fs_store_root("execution-restore-failed");
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
@@ -2785,7 +2802,7 @@ mod tests {
assert_eq!(restored_worker.status, WorkerStatus::Running);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Stale
WorkerExecutionBackendKind::Corrupted
);
assert!(
restored