From 13d853217d793665b48a047479bf81aee10f6d40 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 7 Sep 2026 04:56:54 +0900 Subject: [PATCH] fix: log failed worker creations to stdout --- crates/worker-runtime/src/runtime.rs | 96 +++++++++++++++++++++++++++ crates/workspace-server/src/server.rs | 83 +++++++++++++++++++++++ 2 files changed, 179 insertions(+) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 9722a3ea..4cfc7af3 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -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 { + 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 { 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, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 5e73ecd4..0bad4e14 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -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( )) } +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, @@ -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 {