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

fix: log failed worker creations to stdout

Browse Source
This commit is contained in:
Hare
2026-09-07 04:56:54 +09:00
parent cc27d57e4a
commit 13d853217d
2 changed files with 179 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/runtime.rs
+96
View File
@@ -47,6 +47,7 @@ use protocol::{Event, Method};
use std::collections::BTreeMap;
#[cfg(feature = "ws-server")]
use std::collections::VecDeque;
use std::io::Write as _;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, Weak};
#[cfg(feature = "ws-server")]
@@ -783,6 +784,20 @@ impl Runtime {
&self,
request: CreateWorkerRequest,
scope: Option<&RuntimeWorkspaceScope>,
) -> Result<WorkerDetail, RuntimeError> {
let worker_id = request.worker_id;
let workspace_id = scope.map(|scope| scope.workspace_id.as_str());
let result = self.create_worker_with_workspace_inner(request, scope);
if let Err(error) = &result {
write_runtime_worker_create_failure(worker_id, workspace_id, error);
}
result
}
fn create_worker_with_workspace_inner(
&self,
request: CreateWorkerRequest,
scope: Option<&RuntimeWorkspaceScope>,
) -> Result<WorkerDetail, RuntimeError> {
let operation_lock = self.worker_operation_lock(request.worker_id)?;
let _operation_guard = operation_lock
@@ -3362,6 +3377,69 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R
Ok(())
}
fn runtime_worker_create_failure_log_line(
worker_id: WorkerId,
workspace_id: Option<&str>,
error: &RuntimeError,
) -> String {
let (error_kind, operation, outcome) = match error {
RuntimeError::RuntimeStopped => ("runtime_stopped", None, None),
RuntimeError::InvalidInitialInputKind { .. } => ("invalid_initial_input_kind", None, None),
RuntimeError::WorkerNotFound { .. } => ("worker_not_found", None, None),
RuntimeError::WorkerExecutionUnavailable { .. } => {
("worker_execution_unavailable", None, None)
}
RuntimeError::ExecutionBackendUnavailable { .. } => {
("execution_backend_unavailable", None, None)
}
RuntimeError::WorkerExecutionRejected {
operation, outcome, ..
} => (
"worker_execution_rejected",
Some(format!("{operation:?}")),
Some(format!("{outcome:?}")),
),
RuntimeError::LimitTooLarge { .. } => ("limit_too_large", None, None),
RuntimeError::InvalidRequest(_) => ("invalid_request", None, None),
RuntimeError::WorkspaceOwnerMismatch { .. } => ("workspace_owner_mismatch", None, None),
RuntimeError::WorkingDirectory(_) => ("working_directory", None, None),
RuntimeError::ConfigBundleMissing { .. } => ("config_bundle_missing", None, None),
RuntimeError::ConfigBundleDigestMismatch { .. } => {
("config_bundle_digest_mismatch", None, None)
}
RuntimeError::InvalidProfileSelector { .. } => ("invalid_profile_selector", None, None),
RuntimeError::UnsupportedConfigDeclaration { .. } => {
("unsupported_config_declaration", None, None)
}
RuntimeError::StoreIo { .. } => ("store_io", None, None),
RuntimeError::StoreMissing { .. } => ("store_missing", None, None),
RuntimeError::StoreCorrupt { .. } => ("store_corrupt", None, None),
RuntimeError::StatePoisoned => ("state_poisoned", None, None),
};
serde_json::json!({
"level": "ERROR",
"event": "worker_create_failed",
"component": "runtime",
"workspace_id": workspace_id,
"worker_id": worker_id.to_string(),
"error_kind": error_kind,
"operation": operation,
"outcome": outcome,
})
.to_string()
}
fn write_runtime_worker_create_failure(
worker_id: WorkerId,
workspace_id: Option<&str>,
error: &RuntimeError,
) {
let line = runtime_worker_create_failure_log_line(worker_id, workspace_id, error);
let mut stdout = std::io::stdout().lock();
let _ = writeln!(stdout, "{line}");
let _ = stdout.flush();
}
fn validate_create_workspace_scope(
request: &CreateWorkerRequest,
workspace_id: Option<&str>,
@@ -3485,6 +3563,24 @@ mod tests {
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
#[test]
fn worker_create_failure_log_is_structured_for_stdout() {
let worker_id = WorkerId::now_v7();
let line = runtime_worker_create_failure_log_line(
worker_id,
Some("workspace-a"),
&RuntimeError::InvalidRequest("rejected create".to_string()),
);
let event: serde_json::Value = serde_json::from_str(&line).unwrap();
assert_eq!(event["level"], "ERROR");
assert_eq!(event["event"], "worker_create_failed");
assert_eq!(event["component"], "runtime");
assert_eq!(event["workspace_id"], "workspace-a");
assert_eq!(event["worker_id"], worker_id.to_string());
assert_eq!(event["error_kind"], "invalid_request");
assert!(event.get("message").is_none());
}
fn test_command() -> protocol::WorkerCommandEnvelope {
protocol::WorkerCommandEnvelope {
command_id: 1,
crates/workspace-server/src/server.rs
+83
View File
@@ -1,4 +1,5 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::io::Write as _;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock, Weak};
@@ -2406,6 +2407,12 @@ impl WorkspaceApi {
),
));
}
write_workspace_worker_create_failure(
runtime_id,
worker_id,
"workdir_attachment_reserve",
&diagnostics,
);
return Err(ApiError::with_diagnostics(error, diagnostics));
}
let mut result = match self
@@ -2419,6 +2426,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"runtime_spawn_transport",
None,
&compensation_context,
attachment_reservation
@@ -2438,6 +2446,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"runtime_spawn_rejected",
None,
&compensation_context,
attachment_reservation
@@ -2455,6 +2464,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"runtime_worker_identity",
Some(worker),
&compensation_context,
attachment_reservation
@@ -2481,6 +2491,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"runtime_spawn_rejected",
Some(worker),
&compensation_context,
attachment_reservation
@@ -2503,6 +2514,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"workspace_api_transport",
Some(worker),
&compensation_context,
attachment_reservation
@@ -2520,6 +2532,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"workspace_api_rejected",
Some(worker),
&compensation_context,
attachment_reservation
@@ -2550,6 +2563,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"worker_registry_identity",
Some(worker),
&compensation_context,
Some((workdir_id.as_str(), reservation_id.as_str())),
@@ -2581,6 +2595,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"worker_registry_finalize",
None,
&compensation_context,
Some((workdir_id.as_str(), reservation_id.as_str())),
@@ -2612,6 +2627,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"workdir_attachment_finalize",
None,
&compensation_context,
Some((workdir_id.as_str(), reservation_id.as_str())),
@@ -2628,6 +2644,7 @@ impl WorkspaceApi {
runtime_id,
worker_id,
&reservation_fingerprint,
"create_reservation_complete",
Some(worker),
&compensation_context,
None,
@@ -14275,6 +14292,37 @@ fn finalize_worker_spawn_stage<T>(
))
}
fn workspace_worker_create_failure_log_line(
runtime_id: &str,
worker_id: WorkerId,
phase: &str,
diagnostics: &[RuntimeDiagnostic],
) -> String {
serde_json::json!({
"level": "ERROR",
"event": "worker_create_failed",
"component": "workspace_server",
"runtime_id": runtime_id,
"worker_id": worker_id.to_string(),
"phase": phase,
"cleanup_succeeded": diagnostics.is_empty(),
"cleanup_diagnostics": diagnostics,
})
.to_string()
}
fn write_workspace_worker_create_failure(
runtime_id: &str,
worker_id: WorkerId,
phase: &str,
diagnostics: &[RuntimeDiagnostic],
) {
let line = workspace_worker_create_failure_log_line(runtime_id, worker_id, phase, diagnostics);
let mut stdout = std::io::stdout().lock();
let _ = writeln!(stdout, "{line}");
let _ = stdout.flush();
}
fn api_error_with_additional_diagnostics(
mut error: ApiError,
diagnostics: Vec<RuntimeDiagnostic>,
@@ -14288,6 +14336,7 @@ fn compensate_failed_workspace_worker_create(
runtime_id: &str,
reservation_worker_id: WorkerId,
create_fingerprint: &str,
failure_phase: &str,
worker: Option<&WorkerSummary>,
context: &WorkerSpawnCompensationContext<'_>,
attachment_reservation: Option<(&str, &str)>,
@@ -14355,6 +14404,12 @@ fn compensate_failed_workspace_worker_create(
),
));
}
write_workspace_worker_create_failure(
runtime_id,
reservation_worker_id,
failure_phase,
&diagnostics,
);
diagnostics
}
@@ -19557,6 +19612,34 @@ mod tests {
assert_eq!(sanitized, "failed to open server database");
}
#[test]
fn worker_create_failure_log_is_structured_for_stdout() {
let worker_id = WorkerId::now_v7();
let diagnostics = vec![RuntimeDiagnostic {
code: "worker_cleanup_failed".to_string(),
severity: DiagnosticSeverity::Error,
message: "cleanup failed".to_string(),
}];
let line = workspace_worker_create_failure_log_line(
"runtime-a",
worker_id,
"runtime_spawn_transport",
&diagnostics,
);
let event: serde_json::Value = serde_json::from_str(&line).unwrap();
assert_eq!(event["level"], "ERROR");
assert_eq!(event["event"], "worker_create_failed");
assert_eq!(event["component"], "workspace_server");
assert_eq!(event["runtime_id"], "runtime-a");
assert_eq!(event["worker_id"], worker_id.to_string());
assert_eq!(event["phase"], "runtime_spawn_transport");
assert_eq!(event["cleanup_succeeded"], false);
assert_eq!(
event["cleanup_diagnostics"][0]["code"],
"worker_cleanup_failed"
);
}
#[test]
fn workdir_runtime_miss_uses_exact_typed_code() {
let typed_not_found = [RuntimeDiagnostic {