From 4664f225b3a005bef8d524222990524add11abe9 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 6 Nov 2026 11:57:30 +0900 Subject: [PATCH] fix(startup): retain postconnection creation facts and safe observations --- Cargo.lock | 1 + crates/runtime-api/Cargo.toml | 1 + crates/runtime-api/src/startup.rs | 16 + crates/worker-runtime/src/error.rs | 24 + crates/worker-runtime/src/http_server.rs | 11 +- crates/worker-runtime/src/lib.rs | 1 + crates/worker-runtime/src/runtime.rs | 627 +++++++++++------- .../src/runtime/create_failure_tests.rs | 626 +++++++++++++++++ .../src/runtime/restore_continuation.rs | 2 +- .../worker-runtime/src/startup_observation.rs | 205 ++++++ .../src/backend_job_upgrade_tests.rs | 22 +- crates/workspace-server/src/hosts.rs | 28 +- crates/workspace-server/src/server.rs | 15 +- .../src/server/tests/startup_result_tests.rs | 306 ++++++++- .../tests/ticket_worker_launch_tests.rs | 96 ++- .../src/server/worker_startup.rs | 68 +- .../src/subject_assignment_migration_tests.rs | 71 +- .../src/ticket_post_upgrade_tests.rs | 13 +- 18 files changed, 1803 insertions(+), 330 deletions(-) create mode 100644 crates/worker-runtime/src/runtime/create_failure_tests.rs create mode 100644 crates/worker-runtime/src/startup_observation.rs diff --git a/Cargo.lock b/Cargo.lock index 22d2f63a1..958518a8f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4334,6 +4334,7 @@ dependencies = [ "serde", "serde_json", "session-store", + "sha2 0.11.0", "tokio", "workdir", ] diff --git a/crates/runtime-api/Cargo.toml b/crates/runtime-api/Cargo.toml index 23c95a2b0..3f45ae2c6 100644 --- a/crates/runtime-api/Cargo.toml +++ b/crates/runtime-api/Cargo.toml @@ -12,6 +12,7 @@ plugin-plan = { path = "../plugin-plan" } protocol = { workspace = true, features = ["json-schema"] } error-contract = { workspace = true, features = ["json-schema"] } schemars.workspace = true +sha2.workspace = true serde_json.workspace = true session-store.workspace = true serde = { workspace = true, features = ["derive"] } diff --git a/crates/runtime-api/src/startup.rs b/crates/runtime-api/src/startup.rs index 7c47937e6..a7aff76ec 100644 --- a/crates/runtime-api/src/startup.rs +++ b/crates/runtime-api/src/startup.rs @@ -658,6 +658,22 @@ pub struct StartupCorrelation( ); impl StartupCorrelation { + /// A shared path-free lookup token for an owner-bound Runtime/Worker pair. + /// This is not the create fingerprint, operation key, or replay authority. + pub fn for_worker(runtime_id: &str, worker_id: &str) -> Self { + use sha2::{Digest, Sha256}; + let mut hash = Sha256::new(); + hash.update(runtime_id.as_bytes()); + hash.update([0]); + hash.update(worker_id.as_bytes()); + let digest = hash + .finalize() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + Self(format!("worker-{digest}")) + } + pub fn new(value: impl Into) -> Result { let value = value.into(); if value.is_empty() diff --git a/crates/worker-runtime/src/error.rs b/crates/worker-runtime/src/error.rs index 5b6e4e5af..abfd3692e 100644 --- a/crates/worker-runtime/src/error.rs +++ b/crates/worker-runtime/src/error.rs @@ -63,6 +63,15 @@ pub enum RuntimeError { result: WorkerExecutionResult, }, + /// Creation failed after connection; the original error remains internal source + /// evidence. Display deliberately never renders it or the startup attachment. + #[error("worker creation failed")] + WorkerCreateFailed { + failure: Box, + #[source] + source: Box, + }, + #[error("limit {requested} exceeds maximum {max}")] LimitTooLarge { requested: usize, max: usize }, @@ -149,11 +158,26 @@ pub enum RuntimeError { } impl RuntimeError { + /// Only an owning create boundary or a Spawn result can identify a creation + /// failure. Independent Input errors retain their ordinary API contract. + pub fn create_failure(&self) -> Option { + match self { + Self::WorkerCreateFailed { failure, .. } => Some((**failure).clone()), + Self::WorkerExecutionRejected { result, .. } => { + crate::execution::startup_create_failure(result) + .ok() + .flatten() + } + _ => None, + } + } + /// The original execution result, including primary facts and independent secondary /// failures. Display is presentation only and must not be parsed to recover causes. pub fn execution_result(&self) -> Option<&WorkerExecutionResult> { match self { Self::WorkerExecutionRejected { result, .. } => Some(result), + Self::WorkerCreateFailed { source, .. } => source.execution_result(), _ => None, } } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 6f51cba8b..1c04322d8 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2347,14 +2347,7 @@ impl RuntimeHttpRestError { fn runtime(error: RuntimeError) -> Self { let status = status_for_runtime_error(&error); let code = code_for_runtime_error(&error); - let create_failure = match &error { - RuntimeError::WorkerExecutionRejected { result, .. } => { - crate::execution::startup_create_failure(result) - .ok() - .flatten() - } - _ => None, - }; + let create_failure = error.create_failure(); let startup = match &error { RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result), _ => None, @@ -2440,6 +2433,7 @@ fn startup_transfer( fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { match error { + RuntimeError::WorkerCreateFailed { .. } => StatusCode::INTERNAL_SERVER_ERROR, RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => { StatusCode::NOT_FOUND } @@ -2549,6 +2543,7 @@ fn code_for_runtime_error(error: &RuntimeError) -> String { RuntimeError::ExecutionBackendUnavailable { .. } => { "execution_backend_unavailable".to_string() } + RuntimeError::WorkerCreateFailed { .. } => "worker_create_failed".to_string(), RuntimeError::WorkerExecutionRejected { .. } => "worker_execution_rejected".to_string(), RuntimeError::WorkspaceOwnerMismatch { .. } => "workspace_owner_mismatch".to_string(), RuntimeError::LimitTooLarge { .. } => "limit_too_large".to_string(), diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index 637d56b6b..9fa6d9143 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -31,6 +31,7 @@ mod session_material; mod session_public; mod session_retention; pub mod ssh_host_key_probe; +mod startup_observation; pub mod worker_backend; pub mod worker_source; pub mod working_directory; diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index d172db44a..823d26a7b 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -56,6 +56,11 @@ use protocol::subscription::{ SubscriptionWorkerState, SubscriptionWorkerWorkdirAttachment, }; use protocol::{Event, Method}; +use runtime_api::startup::{ + CompensationOutcome, CreationAcceptance, ExecutionOutcome, FinalizeFacts, FinalizeStage, + StartupBoundaryCause, StartupBoundaryReason, StartupCause, StartupCorrelation, + StartupOperationFacts, WorkerCreateFailure, +}; use server_api::WorkerRestoreState; use std::collections::BTreeMap; #[cfg(feature = "ws-server")] @@ -1381,7 +1386,34 @@ impl Runtime { let worker_id = request.worker_id; let workspace_id = scope.map(|scope| scope.workspace_id.as_str()); let started_at = std::time::Instant::now(); - let result = self.create_worker_with_workspace_inner(request, scope); + let result = self + .create_worker_with_workspace_inner(request, scope) + .map_err(|error| { + // Use only the host-bound local identity, never received correlation. + // A poisoned/unbound Runtime cannot invent correlation authority. + let correlation = self.lock().ok().and_then(|state| { + state.runtime_identity.as_deref().map(|runtime_id| { + StartupCorrelation::for_worker(runtime_id, &worker_id.to_string()) + }) + }); + match (error.create_failure(), correlation) { + (Some(failure), Some(correlation)) => { + let mut operation = failure.operation().clone(); + operation.correlation = Some(correlation); + let failure = Box::new(failure.with_operation(operation)); + match error { + RuntimeError::WorkerCreateFailed { source, .. } => { + RuntimeError::WorkerCreateFailed { failure, source } + } + source => RuntimeError::WorkerCreateFailed { + failure, + source: Box::new(source), + }, + } + } + _ => error, + } + }); if let Err(error) = &result { write_runtime_worker_create_failure( worker_id, @@ -1528,13 +1560,13 @@ impl Runtime { WorkerExecutionSpawnResult::Rejected(mut result) | WorkerExecutionSpawnResult::RolledBack(mut result) | WorkerExecutionSpawnResult::Errored(mut result) => { - if let Err((error, phase, outcome)) = + if let Err((_error, phase, outcome)) = self.rollback_failed_create_with_phase(&worker_ref) { tracing::error!( worker_id = %worker_ref.worker_id, ?phase, - %error, + ?outcome, "failed to roll back Worker creation" ); // A later failure must not make a confirmed record deletion @@ -1601,7 +1633,8 @@ impl Runtime { }); tracing::error!( worker_id = %worker_ref.worker_id, - %error, + ?phase, + ?outcome, "failed to persist Worker create reconciliation evidence" ); } else { @@ -1620,80 +1653,119 @@ impl Runtime { } }; - if let Some(mut initial_input) = { - let state = self.lock()?; - state - .worker(&worker_ref)? - .request - .as_ref() - .and_then(|request| request.initial_input.clone()) - } { - let expected_submission_id = initial_input - .submission_request_id - .clone() - .filter(|request_id| !request_id.trim().is_empty()) - .unwrap_or_else(|| Uuid::now_v7().to_string()); - initial_input.submission_request_id = Some(expected_submission_id.clone()); - let dispatch_result = backend.dispatch_input(&worker_ref, initial_input.clone()); - if !dispatch_result.is_accepted() { - self.cleanup_connected_failed_create(&backend, &worker_ref)?; - return Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id.clone(), - operation: dispatch_result.operation, - outcome: dispatch_result.outcome, - message: dispatch_result.message_or_default(), - result: dispatch_result, - }); - } - let has_durable_acceptance = dispatch_result - .submission - .as_ref() - .is_some_and(|ack| ack.submission_request_id == expected_submission_id); - if !has_durable_acceptance { - self.cleanup_connected_failed_create(&backend, &worker_ref)?; - let result = WorkerExecutionResult::rejected( - WorkerExecutionOperation::Input, - "execution backend accepted initial input without a durable submission acknowledgement", - ); - return Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id.clone(), - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }); - } - let detail = match self.commit_created_worker( - &worker_ref, - initial_worker_state.clone(), - workdir_attachments, - dispatch_result, - ) { - Ok(detail) => detail, - Err(error) => { - self.cleanup_connected_failed_create(&backend, &worker_ref)?; - return Err(error); - } + // From Connected onward this owner, not the independent Input API, + // supplies creation context. Later cleanup must not replace the source. + let mut execution = ExecutionOutcome::Unknown; + let mut stage = FinalizeStage::InitialInput; + let mut reason = StartupBoundaryReason::FinalizeFailed; + let mut persistence_phase = None; + let mut input_result = None; + let result = (|| { + let initial_input = { + let state = self.lock()?; + state + .worker(&worker_ref)? + .request + .as_ref() + .and_then(|request| request.initial_input.clone()) }; - if let Err(error) = self.record_input_observation(&worker_ref, initial_input) { - self.cleanup_connected_failed_create(&backend, &worker_ref)?; - return Err(error); - } - Ok(detail) - } else { - match self.commit_created_worker( - &worker_ref, - initial_worker_state, - workdir_attachments, - WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn), - ) { - Ok(detail) => Ok(detail), - Err(error) => { - self.cleanup_connected_failed_create(&backend, &worker_ref)?; - Err(error) + if let Some(mut initial_input) = initial_input { + let expected_submission_id = initial_input + .submission_request_id + .clone() + .filter(|request_id| !request_id.trim().is_empty()) + .unwrap_or_else(|| Uuid::now_v7().to_string()); + initial_input.submission_request_id = Some(expected_submission_id.clone()); + let dispatch_result = backend.dispatch_input(&worker_ref, initial_input.clone()); + let has_durable_acceptance = dispatch_result + .submission + .as_ref() + .is_some_and(|ack| ack.submission_request_id == expected_submission_id); + input_result = Some(dispatch_result.clone()); + if !dispatch_result.is_accepted() { + // Confirmed rejection is failed initial execution. A lost + // acknowledgement does not establish non-execution. + execution = match dispatch_result.outcome { + crate::execution::WorkerExecutionOutcome::Errored + if !has_durable_acceptance => + { + ExecutionOutcome::Unknown + } + _ => ExecutionOutcome::Errored, + }; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id, + operation: dispatch_result.operation, + outcome: dispatch_result.outcome, + message: dispatch_result.message_or_default(), + result: dispatch_result, + }); } + if !has_durable_acceptance { + reason = StartupBoundaryReason::InvalidResponse; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id, + operation: dispatch_result.operation, + outcome: dispatch_result.outcome, + message: "initial input has no matching durable acknowledgement".into(), + result: dispatch_result, + }); + } + execution = ExecutionOutcome::Accepted; + stage = FinalizeStage::Publication; + let detail = self.commit_created_worker( + &worker_ref, + initial_worker_state, + workdir_attachments, + dispatch_result, + &mut persistence_phase, + )?; + self.record_input_observation(&worker_ref, initial_input)?; + Ok(detail) + } else { + execution = ExecutionOutcome::Accepted; + stage = FinalizeStage::Publication; + self.commit_created_worker( + &worker_ref, + initial_worker_state, + workdir_attachments, + WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn), + &mut persistence_phase, + ) } - } + })(); + result.map_err(|source| { + let mut startup = connected_create_startup(&source, input_result.as_ref(), reason); + if let Some(phase) = persistence_phase { + push_create_secondary( + &mut startup, + StartupSecondaryFailure::Persistence { + phase, + outcome: create_persistence_outcome(&source), + }, + ); + } + let compensation = + self.cleanup_connected_failed_create(&backend, &worker_ref, &mut startup); + let failure = WorkerCreateFailure::new( + startup, + StartupOperationFacts { + acceptance: CreationAcceptance::Accepted, + execution, + finalize: Some(FinalizeFacts { + stage, + compensation, + }), + backend_compensation: None, + correlation: None, + }, + ) + .expect("connected create facts have bounded secondary observations"); + RuntimeError::WorkerCreateFailed { + failure: Box::new(failure), + source: Box::new(source), + } + }) } /// List Workers known to this Runtime. @@ -2267,7 +2339,7 @@ impl Runtime { Some(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_canonical_write_unknown", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )), None, )); @@ -2322,7 +2394,7 @@ impl Runtime { Some(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_authority_changed", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )), None, )); @@ -2432,7 +2504,7 @@ impl Runtime { Some(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_admission_commit_unknown", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )), None, )); @@ -2665,7 +2737,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_saved_binding_not_ready", - sanitize_worker_create_failure_message(&message), + sanitize_worker_restore_failure_message(&message), )); } let result = RuntimeWorkerRestoreResult::accepted(detail); @@ -2813,7 +2885,7 @@ impl Runtime { WorkerRestoreState::Rejected }, "worker_restore_preparation_rejected", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )); } }; @@ -2922,7 +2994,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_observation_activation_pending", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )); } } @@ -2933,7 +3005,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::Rejected, "worker_restore_preflight_rejected", - sanitize_worker_create_failure_message(&result.message_or_default()), + sanitize_worker_restore_failure_message(&result.message_or_default()), )); } @@ -3001,7 +3073,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_activation_pending", - sanitize_worker_create_failure_message(&error), + sanitize_worker_restore_failure_message(&error), )); } #[cfg(feature = "restore-checkpoints")] @@ -3021,7 +3093,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_observation_activation_pending", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )); } let worker = self.lock()?.worker(worker_ref)?.detail(); @@ -3030,7 +3102,7 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_receipt_commit_pending", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )); } Ok(result) @@ -3057,7 +3129,7 @@ impl Runtime { let receipt = RuntimeWorkerRestoreResult::failed( WorkerRestoreState::RolledBack, "worker_restore_decision_not_committed", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), ); if self.settle_restore_rollback(worker_ref, &receipt).is_ok() { return Ok(receipt); @@ -3084,7 +3156,7 @@ impl Runtime { Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_commit_reconciliation_required", - sanitize_worker_create_failure_message(&error.to_string()), + sanitize_worker_restore_failure_message(&error.to_string()), )) } } @@ -3094,14 +3166,14 @@ impl Runtime { return Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_reconciliation_rejected", - sanitize_worker_create_failure_message(&result.message_or_default()), + sanitize_worker_restore_failure_message(&result.message_or_default()), )); } self.discard_restore_candidate_context(worker_ref, &candidate_context)?; let receipt = RuntimeWorkerRestoreResult::failed( WorkerRestoreState::Rejected, "worker_restore_rejected", - sanitize_worker_create_failure_message(&result.message_or_default()), + sanitize_worker_restore_failure_message(&result.message_or_default()), ); if let Err(error) = self.settle_restore_rollback(worker_ref, &receipt) { tracing::error!( @@ -3121,7 +3193,7 @@ impl Runtime { let receipt = RuntimeWorkerRestoreResult::failed( WorkerRestoreState::RolledBack, "worker_restore_rolled_back", - sanitize_worker_create_failure_message(&_result.message_or_default()), + sanitize_worker_restore_failure_message(&_result.message_or_default()), ); self.discard_restore_candidate_context(worker_ref, &candidate_context)?; #[cfg(feature = "fs-store")] @@ -3174,7 +3246,7 @@ impl Runtime { Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_reconciliation_required", - sanitize_worker_create_failure_message(&format!( + sanitize_worker_restore_failure_message(&format!( "Worker restore requires reconciliation ({:?}): {}", result.outcome, result.message_or_default(), @@ -3194,7 +3266,7 @@ impl Runtime { Ok(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, "worker_restore_backend_outcome_unknown", - sanitize_worker_create_failure_message(&_result.message_or_default()), + sanitize_worker_restore_failure_message(&_result.message_or_default()), )) } } @@ -4007,6 +4079,7 @@ impl Runtime { initial_worker_state: protocol::WorkerStateSnapshot, workdir_attachments: Vec, result: WorkerExecutionResult, + persistence_phase: &mut Option, ) -> Result { let mut state = self.lock()?; let detail = { @@ -4030,41 +4103,155 @@ impl Runtime { worker.workdir_attachments = workdir_attachments; worker.detail() }; + *persistence_phase = Some(StartupFailurePhase::RuntimeSnapshotSave); state.persist_runtime_snapshot()?; + *persistence_phase = Some(StartupFailurePhase::RuntimeReconciliationRecordSave); state.persist_worker(&worker_ref.worker_id)?; + *persistence_phase = None; state.publish_worker_upsert(worker_ref.worker_id)?; Ok(detail) } + /// Observe compensation separately from the original create failure. This + /// performs the same single stop and cleanup as before; no receipt lookup, + /// retry, reexecution or new durable lifecycle operation is introduced here. fn cleanup_connected_failed_create( &self, backend: &WorkerExecutionBackendRef, worker_ref: &WorkerRef, - ) -> Result<(), RuntimeError> { + startup: &mut runtime_api::WorkerStartupFailure, + ) -> CompensationOutcome { + use protocol::startup_failure::{ + StartupCleanup, StartupCleanupFailure, StartupResourceFacts, StartupResources, + }; let operation_id = WorkerLifecycleOperationId::new(); let stop_result = backend.stop_worker_operation(WorkerExecutionStopRequest { operation_id, worker_ref: worker_ref.clone(), }); - if stop_result.is_accepted() + for secondary in &stop_result.secondary_failures { + push_create_secondary(startup, secondary.clone()); + } + if startup.io_kind.is_none() { + startup.io_kind = stop_result.startup_io_kind; + } + // Preserve the original owner's resource facts, if present. In their + // absence, connection establishes startup resources but not their release. + let had_resources = startup.resources.is_some() || stop_result.startup_resources.is_some(); + startup + .resources + .get_or_insert( + stop_result + .startup_resources + .unwrap_or(StartupResourceFacts { + resources: StartupResources::WorkerStartup, + cleanup: StartupCleanup::Required, + durable_effects_possible: true, + }), + ); + if let Some(outcome) = stop_result.startup_failure_persistence { + push_create_secondary( + startup, + StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome, + }, + ); + } + let terminated = stop_result.is_accepted() && stop_result.termination.as_ref().is_some_and(|termination| { termination.confirms_scope_shutdown(worker_ref, operation_id) - }) - { - return self.rollback_failed_create(worker_ref); + }); + if terminated { + if !had_resources { + // Confirms scope shutdown only, not removal of Workdir contents + // or undoing already accepted initial-input effects. + startup.resources.as_mut().unwrap().cleanup = StartupCleanup::Confirmed; + } + match self.rollback_failed_create_with_phase(worker_ref) { + Ok(()) => CompensationOutcome::Completed, + Err((error, phase, outcome)) => { + if matches!( + phase, + StartupFailurePhase::RuntimeSnapshotSave + | StartupFailurePhase::RuntimeRemovalPublish + ) { + push_create_secondary( + startup, + StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeRecordDelete, + outcome: StartupPersistenceOutcome::Persisted, + }, + ); + } + if phase == StartupFailurePhase::RuntimeRemovalPublish { + push_create_secondary( + startup, + StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeSnapshotSave, + outcome: StartupPersistenceOutcome::Persisted, + }, + ); + } + record_create_persistence_failure(startup, &error, phase, outcome); + if outcome == StartupPersistenceOutcome::OutcomeUnknown { + CompensationOutcome::Unknown + } else { + CompensationOutcome::Residual + } + } + } + } else { + let failed = !stop_result.is_accepted() + && stop_result.outcome != crate::execution::WorkerExecutionOutcome::Errored; + let feature_failed = stop_result.termination.as_ref().is_some_and(|receipt| { + receipt.worker_ref == *worker_ref + && receipt.stop_operation_id == operation_id + && !receipt.feature_failures.is_empty() + }); + if failed || feature_failed { + let failure = if feature_failed { + StartupCleanupFailure::Feature + } else if stop_result.outcome + == crate::execution::WorkerExecutionOutcome::Unsupported + { + StartupCleanupFailure::Unsupported + } else { + StartupCleanupFailure::Ownership + }; + push_create_secondary(startup, StartupSecondaryFailure::Cleanup { failure }); + if !had_resources { + startup.resources.as_mut().unwrap().cleanup = StartupCleanup::Failed(failure); + } + } + let mut phase = StartupFailurePhase::RuntimeSnapshotSave; + let retained = (|| { + let mut state = self.lock()?; + let record = state.worker_mut(worker_ref)?; + record.worker_state = stop_result.worker_state.clone(); + state.persist_runtime_snapshot()?; + phase = StartupFailurePhase::RuntimeReconciliationRecordSave; + state.persist_worker(&worker_ref.worker_id) + })(); + match retained { + Ok(()) => push_create_secondary( + startup, + StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::Persisted, + }, + ), + Err(error) => { + let outcome = create_persistence_outcome(&error); + record_create_persistence_failure(startup, &error, phase, outcome); + } + } + if failed || feature_failed { + CompensationOutcome::Residual + } else { + CompensationOutcome::Unknown + } } - let mut state = self.lock()?; - let record = state.worker_mut(worker_ref)?; - - record.worker_state = stop_result.worker_state.clone(); - state.persist_runtime_snapshot()?; - state.persist_worker(&worker_ref.worker_id)?; - Ok(()) - } - - fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { - self.rollback_failed_create_with_phase(worker_ref) - .map_err(|(error, _, _)| error) } // The legacy caller still receives its RuntimeError. Spawn failures instead @@ -5110,7 +5297,7 @@ impl Runtime { let mut state = self.lock()?; let diagnostic_id = state.next_diagnostic_id; state.next_diagnostic_id = state.next_diagnostic_id.saturating_add(1); - let message = sanitize_worker_create_failure_message( + let message = sanitize_worker_restore_failure_message( result .message .as_deref() @@ -7148,7 +7335,7 @@ impl RuntimeState { result: &WorkerExecutionResult, receipt: &RuntimeWorkerRestoreResult, ) -> Result<(), RuntimeError> { - let message = sanitize_worker_create_failure_message(&result.message_or_default()); + let message = sanitize_worker_restore_failure_message(&result.message_or_default()); let diagnostic_id = self.next_diagnostic_id; let workspace_id = self .workers @@ -8381,25 +8568,90 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R Ok(()) } -fn worker_execution_failure_code(result: &WorkerExecutionResult) -> &'static str { - let message = result.message.as_deref().unwrap_or_default(); - if message.contains("timed out waiting for durable Worker Submit acceptance") { - "input_commit_timeout" - } else if message.contains("worker adapter task did not complete within") { - "adapter_task_timeout" - } else if message.contains("failed to send Worker method") { - "worker_method_send_failed" - } else if message.contains("worker rejected Submit") { - "worker_submit_rejected" - } else if message.contains("before durable acceptance") { - "worker_failed_before_input_commit" - } else if message.contains("event stream closed") { - "worker_event_stream_closed" - } else { - "worker_execution_failed" +/// Keep the original result's typed facts even when publication is the failing +/// boundary. No message/path parsing is used to choose the fallback cause. +fn connected_create_startup( + source: &RuntimeError, + result: Option<&WorkerExecutionResult>, + reason: StartupBoundaryReason, +) -> runtime_api::WorkerStartupFailure { + let fallback = WorkerCreateFailure::from_cause( + StartupCause::Boundary(StartupBoundaryCause { reason }), + StartupOperationFacts { + acceptance: CreationAcceptance::Accepted, + execution: ExecutionOutcome::Unknown, + finalize: None, + backend_compensation: None, + correlation: None, + }, + ) + .expect("closed startup boundary cause is valid"); + let (mut startup, _) = fallback.into_parts(); + if let Some(result) = result { + if let Some(primary) = &result.startup_failure { + startup.primary = primary.clone(); + } + for secondary in &result.secondary_failures { + push_create_secondary(&mut startup, secondary.clone()); + } + startup.resources = result.startup_resources; + startup.persistence = result.startup_failure_persistence; + startup.io_kind = result.startup_io_kind; + } + if startup.io_kind.is_none() { + startup.io_kind = create_io_kind(source); + } + startup +} + +fn create_io_kind(error: &RuntimeError) -> Option { + match error { + RuntimeError::StoreIo { source, .. } + | RuntimeError::StoreCommitOutcomeUnknown { source, .. } => Some(source.kind().into()), + _ => None, } } +fn create_persistence_outcome(error: &RuntimeError) -> StartupPersistenceOutcome { + if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) { + StartupPersistenceOutcome::OutcomeUnknown + } else { + StartupPersistenceOutcome::NotSaved + } +} + +fn push_create_secondary( + startup: &mut runtime_api::WorkerStartupFailure, + secondary: StartupSecondaryFailure, +) { + // The create DTO has the core diagnostic bound; this is an observation, + // not another durable ledger. The original error remains boxed as source. + if startup.secondary.len() < error_contract::MAX_DIAGNOSTICS { + startup.secondary.push(secondary); + } +} + +fn record_create_persistence_failure( + startup: &mut runtime_api::WorkerStartupFailure, + error: &RuntimeError, + phase: StartupFailurePhase, + outcome: StartupPersistenceOutcome, +) { + push_create_secondary( + startup, + StartupSecondaryFailure::Persistence { phase, outcome }, + ); + if startup.io_kind.is_none() { + startup.io_kind = create_io_kind(error); + } + tracing::error!( + ?phase, + ?outcome, + "Worker create compensation persistence failed" + ); +} + +#[cfg(test)] fn runtime_worker_create_failure_fields( error: &RuntimeError, ) -> (&'static str, Option, Option) { @@ -8425,6 +8677,7 @@ fn runtime_worker_create_failure_fields( RuntimeError::ExecutionBackendUnavailable { .. } => { ("execution_backend_unavailable", None, None) } + RuntimeError::WorkerCreateFailed { .. } => ("worker_create_failed", None, None), RuntimeError::WorkerExecutionRejected { operation, outcome, .. } => ( @@ -8454,9 +8707,11 @@ fn runtime_worker_create_failure_fields( } } -const WORKER_CREATE_FAILURE_MESSAGE_LIMIT: usize = 512; +// Legacy Restore diagnostic presentation only. This is not a create failure +// projector or telemetry allowlist; creation uses startup_observation instead. +const WORKER_RESTORE_FAILURE_MESSAGE_LIMIT: usize = 512; -fn sanitize_worker_create_failure_message(message: &str) -> String { +fn sanitize_worker_restore_failure_message(message: &str) -> String { let mut redact_next = false; let detail = message .split_whitespace() @@ -8496,10 +8751,10 @@ fn sanitize_worker_create_failure_message(message: &str) -> String { }) .collect::>() .join(" "); - let truncated = detail.chars().count() > WORKER_CREATE_FAILURE_MESSAGE_LIMIT; + let truncated = detail.chars().count() > WORKER_RESTORE_FAILURE_MESSAGE_LIMIT; let mut detail = detail .chars() - .take(WORKER_CREATE_FAILURE_MESSAGE_LIMIT) + .take(WORKER_RESTORE_FAILURE_MESSAGE_LIMIT) .collect::(); if truncated { detail.push('…'); @@ -8507,43 +8762,13 @@ fn sanitize_worker_create_failure_message(message: &str) -> String { detail } -fn runtime_worker_create_failure_message(error: &RuntimeError) -> Option { - match error { - RuntimeError::WorkerExecutionRejected { message, .. } => { - Some(sanitize_worker_create_failure_message(message)) - } - _ => None, - } -} - fn write_runtime_worker_create_failure( worker_id: WorkerId, workspace_id: Option<&str>, error: &RuntimeError, elapsed: std::time::Duration, ) { - let (error_kind, operation, outcome) = runtime_worker_create_failure_fields(error); - let execution_failure_code = match error { - RuntimeError::WorkerExecutionRejected { result, .. } => { - worker_execution_failure_code(result) - } - _ => "", - }; - let execution_failure_message = runtime_worker_create_failure_message(error); - tracing::error!( - target: "yoi::worker_create", - event = "worker_create_failed", - component = "runtime", - workspace_id = workspace_id.unwrap_or(""), - worker_id = %worker_id, - error_kind, - operation = operation.as_deref().unwrap_or(""), - outcome = outcome.as_deref().unwrap_or(""), - execution_failure_code, - execution_failure_message = execution_failure_message.as_deref().unwrap_or(""), - duration_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX), - "Worker creation failed" - ); + crate::startup_observation::write_create_failure(worker_id, workspace_id, error, elapsed); } fn validate_create_workspace_scope( @@ -8692,6 +8917,11 @@ mod tests { use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Barrier, Condvar, Mutex}; + mod create_failure_tests { + use super::*; + include!("runtime/create_failure_tests.rs"); + } + #[cfg(feature = "fs-store")] mod create_rollback_tests { //! Connect the real create/rollback path to FsRuntimeStore and reopen its disk state. @@ -8911,57 +9141,6 @@ mod tests { } } - #[test] - fn worker_create_failure_fields_only_include_sanitized_execution_detail() { - let invalid_request = RuntimeError::InvalidRequest("private-token".to_string()); - let (error_kind, operation, outcome) = - runtime_worker_create_failure_fields(&invalid_request); - assert_eq!(error_kind, "invalid_request"); - assert_eq!(operation, None); - assert_eq!(outcome, None); - assert_eq!( - runtime_worker_create_failure_message(&invalid_request), - None - ); - - let result = WorkerExecutionResult::errored( - WorkerExecutionOperation::Spawn, - "failed to spawn controller in /private/path: File exists; token super-secret", - ); - let error = RuntimeError::WorkerExecutionRejected { - worker_id: WorkerId::now_v7(), - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }; - assert_eq!( - runtime_worker_create_failure_message(&error).as_deref(), - Some("failed to spawn controller in [redacted] File exists; token [redacted]") - ); - - assert_eq!( - sanitize_worker_create_failure_message("backend private-token failed"), - "backend [redacted] failed" - ); - - let truncated = sanitize_worker_create_failure_message(&"x".repeat(513)); - assert_eq!(truncated.chars().count(), 513); - assert!(truncated.ends_with('…')); - } - - #[test] - fn worker_create_failure_code_classifies_durable_acceptance_timeout() { - let timeout = WorkerExecutionResult::errored( - WorkerExecutionOperation::Input, - "timed out waiting for durable Worker Submit acceptance; private-token", - ); - assert_eq!( - worker_execution_failure_code(&timeout), - "input_commit_timeout" - ); - } - fn test_command() -> protocol::WorkerCommandEnvelope { protocol::WorkerCommandEnvelope { command_id: 1 } } @@ -9950,7 +10129,6 @@ mod tests { recovery_absent: AtomicBool, recovery_checks: std::sync::atomic::AtomicUsize, spawn_result: Mutex>, - #[cfg(feature = "fs-store")] before_spawn_result: Mutex>>, job_bindings: Mutex>>, dispatch_result: Mutex>, @@ -10197,7 +10375,6 @@ mod tests { .lock() .unwrap() .push(request.request.backend_job.clone()); - #[cfg(feature = "fs-store")] if let Some(before_result) = self.before_spawn_result.lock().unwrap().take() { before_result(); } @@ -15503,13 +15680,7 @@ mod tests { let error = runtime.create_worker(request).unwrap_err(); - assert!(matches!( - error, - RuntimeError::WorkerExecutionRejected { - outcome: crate::execution::WorkerExecutionOutcome::Rejected, - .. - } - )); + create_failure_tests::assert_ack_failure(&error); assert!(runtime.list_workers().unwrap().is_empty()); } @@ -15524,13 +15695,7 @@ mod tests { let error = runtime.create_worker(request).unwrap_err(); - assert!(matches!( - error, - RuntimeError::WorkerExecutionRejected { - outcome: crate::execution::WorkerExecutionOutcome::Rejected, - .. - } - )); + create_failure_tests::assert_ack_failure(&error); assert!(runtime.list_workers().unwrap().is_empty()); } @@ -17160,7 +17325,7 @@ mod tests { ); assert!(!diagnostic.message.contains("credential-value")); assert!(!diagnostic.message.contains("/private/path")); - assert!(diagnostic.message.chars().count() <= WORKER_CREATE_FAILURE_MESSAGE_LIMIT + 1); + assert!(diagnostic.message.chars().count() <= WORKER_RESTORE_FAILURE_MESSAGE_LIMIT + 1); let before = std::fs::read( dir.path() .join("runtime/workers") @@ -17346,7 +17511,7 @@ mod tests { assert!(!message.contains("/private"), "{message}"); assert!(!message.contains("restore-token"), "{message}"); assert!(!message.contains("hidden-value"), "{message}"); - assert!(message.chars().count() <= WORKER_CREATE_FAILURE_MESSAGE_LIMIT + 1); + assert!(message.chars().count() <= WORKER_RESTORE_FAILURE_MESSAGE_LIMIT + 1); assert!( runtime .lock() diff --git a/crates/worker-runtime/src/runtime/create_failure_tests.rs b/crates/worker-runtime/src/runtime/create_failure_tests.rs new file mode 100644 index 000000000..6e62d877f --- /dev/null +++ b/crates/worker-runtime/src/runtime/create_failure_tests.rs @@ -0,0 +1,626 @@ +// Registered by runtime.rs's normal library test module. Execution responses are +// fixtures; creation, persistence, projection and compensation use the real owner. +use protocol::startup_failure::{ + RuntimeStartupCause, RuntimeStartupReason, StartupCleanup, StartupCleanupFailure, + StartupIoKind, StartupResourceFacts, StartupResources, StartupStage, +}; + +fn boundary_reason(failure: &WorkerCreateFailure) -> StartupBoundaryReason { + let projected = failure + .startup() + .primary + .project_for::(&()) + .unwrap(); + let StartupCause::Boundary(cause) = projected.details() else { + panic!("expected a closed boundary cause"); + }; + cause.reason +} + +pub(super) fn assert_ack_failure(error: &RuntimeError) { + assert!(matches!(error, RuntimeError::WorkerCreateFailed { .. })); + let failure = error.create_failure().unwrap(); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown); + assert_eq!( + failure.operation().finalize, + Some(FinalizeFacts { + stage: FinalizeStage::InitialInput, + compensation: CompensationOutcome::Completed, + }) + ); + assert_eq!( + boundary_reason(&failure), + StartupBoundaryReason::InvalidResponse + ); + // The missing/foreign receipt is not rewritten into a confirmed rejection. + assert!(error.execution_result().unwrap().is_accepted()); + assert_eq!(error.to_string(), "worker creation failed"); +} + +fn rejected_input() -> WorkerExecutionResult { + let attachment = worker::startup_failure::runtime_startup_attachment(RuntimeStartupCause { + stage: StartupStage::Controller, + reason: RuntimeStartupReason::StartupFailed, + }) + .unwrap(); + let mut result = WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "TEST_ONLY_SECRET original rejection /private/owner/path", + ) + .with_startup_failure(attachment); + result + .secondary_failures + .push(StartupSecondaryFailure::Cleanup { + failure: StartupCleanupFailure::Workdir, + }); + result.startup_resources = Some(StartupResourceFacts { + resources: StartupResources::Unproven, + cleanup: StartupCleanup::Failed(StartupCleanupFailure::Workdir), + durable_effects_possible: true, + }); + result.startup_failure_persistence = Some(StartupPersistenceOutcome::NotSaved); + result.startup_io_kind = Some(StartupIoKind::PermissionDenied); + result +} + +#[derive(Clone, Copy, Debug)] +enum DispatchFault { + DroppedAck, + MismatchedAck, + ConfirmedRejection, + LostResponse, +} + +#[derive(Clone, Copy, Debug)] +enum CleanupFault { + None, + MissingTermination, + Rejected, + LostResponse, + MismatchedTermination, +} + +#[test] +fn connected_create_dispatch_and_cleanup_fault_matrix() { + for dispatch in [ + DispatchFault::DroppedAck, + DispatchFault::MismatchedAck, + DispatchFault::ConfirmedRejection, + DispatchFault::LostResponse, + ] { + for cleanup in [ + CleanupFault::None, + CleanupFault::MissingTermination, + CleanupFault::Rejected, + CleanupFault::LostResponse, + CleanupFault::MismatchedTermination, + ] { + let (runtime, backend) = runtime_and_backend(); + let original = match dispatch { + DispatchFault::DroppedAck => { + WorkerExecutionResult::accepted(WorkerExecutionOperation::Input) + } + DispatchFault::MismatchedAck => { + backend.preserve_submission_acknowledgement_id(); + WorkerExecutionResult::accepted_submission( + WorkerExecutionOperation::Input, + "foreign-request", + "foreign-submission", + protocol::SubmissionDisposition::Started, + ) + } + DispatchFault::ConfirmedRejection => rejected_input(), + DispatchFault::LostResponse => WorkerExecutionResult::errored( + WorkerExecutionOperation::Input, + "response lost; TEST_ONLY_SECRET", + ), + }; + backend.set_dispatch_result(original.clone()); + let compensation = match cleanup { + CleanupFault::None => CompensationOutcome::Completed, + CleanupFault::MissingTermination => { + backend.set_stop_result(WorkerExecutionResult::accepted( + WorkerExecutionOperation::Stop, + )); + CompensationOutcome::Unknown + } + CleanupFault::Rejected => { + backend.set_stop_result(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Stop, + "TEST_ONLY_SECRET stop rejected", + )); + CompensationOutcome::Residual + } + CleanupFault::LostResponse => { + backend.set_stop_result(WorkerExecutionResult::errored( + WorkerExecutionOperation::Stop, + "TEST_ONLY_SECRET stop response lost", + )); + CompensationOutcome::Unknown + } + CleanupFault::MismatchedTermination => { + let mut stop = WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop); + stop.termination = Some(crate::execution::WorkerExecutionTermination { + scope: crate::execution::WorkerExecutionTerminationScope::Controller, + worker_ref: WorkerRef::new(WorkerId::now_v7()), + stop_operation_id: WorkerLifecycleOperationId::new(), + execution_id: Uuid::now_v7(), + shutdown_command_id: None, + submissions: Vec::new(), + feature_failures: Vec::new(), + }); + backend.set_stop_result(stop); + CompensationOutcome::Unknown + } + }; + let mut request = task_request("connected create fault matrix"); + request.initial_input = Some(WorkerInput::user("start once")); + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!( + failure.operation().execution, + if matches!(dispatch, DispatchFault::ConfirmedRejection) { + ExecutionOutcome::Errored + } else { + ExecutionOutcome::Unknown + }, + "{dispatch:?}/{cleanup:?}" + ); + assert_eq!( + failure.operation().finalize, + Some(FinalizeFacts { + stage: FinalizeStage::InitialInput, + compensation, + }) + ); + assert_eq!(failure.operation().backend_compensation, None); + assert_eq!(failure.operation().correlation, None); + assert_eq!(error.execution_result(), Some(&original)); + assert_eq!(error.to_string(), "worker creation failed"); + if let Some(primary) = original.startup_failure.as_ref() { + assert_eq!(&failure.startup().primary, primary); + assert!( + failure + .startup() + .secondary + .starts_with(&original.secondary_failures) + ); + assert_eq!(failure.startup().resources, original.startup_resources); + assert_eq!( + failure.startup().persistence, + original.startup_failure_persistence + ); + assert_eq!(failure.startup().io_kind, original.startup_io_kind); + assert!( + std::error::Error::source(&error) + .unwrap() + .to_string() + .contains("TEST_ONLY_SECRET") + ); + } else { + assert_eq!( + boundary_reason(&failure), + if matches!(dispatch, DispatchFault::LostResponse) { + StartupBoundaryReason::FinalizeFailed + } else { + StartupBoundaryReason::InvalidResponse + } + ); + let resources = failure.startup().resources.unwrap(); + assert!(resources.durable_effects_possible); + assert_eq!( + resources.cleanup, + match cleanup { + CleanupFault::None => StartupCleanup::Confirmed, + CleanupFault::Rejected => + StartupCleanup::Failed(StartupCleanupFailure::Ownership), + _ => StartupCleanup::Required, + } + ); + } + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1); + assert!( + backend + .submission_receipt_queries + .lock() + .unwrap() + .is_empty() + ); + assert_eq!( + runtime.list_workers().unwrap().is_empty(), + matches!(cleanup, CleanupFault::None) + ); + } + } +} + +#[test] +fn independent_input_error_is_not_a_create_failure() { + let (runtime, backend) = runtime_and_backend(); + let worker = runtime + .create_worker(task_request("independent input")) + .unwrap(); + let original = rejected_input(); + backend.set_dispatch_result(original.clone()); + let error = runtime + .send_input(&worker.worker_ref, WorkerInput::user("ordinary input")) + .unwrap_err(); + assert!(matches!( + &error, + RuntimeError::WorkerExecutionRejected { .. } + )); + assert_eq!(error.create_failure(), None); + assert_eq!(error.execution_result(), Some(&original)); + assert_eq!(*backend.stop_count.lock().unwrap(), 0); +} + +#[test] +fn connected_create_initial_input_read_failure_keeps_create_context() { + let (runtime, backend) = runtime_and_backend(); + let mut request = task_request("initial input read failure"); + request.initial_input = Some(WorkerInput::user("never dispatched")); + let worker_id = request.worker_id; + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let runtime = runtime.clone(); + move || { + runtime.lock().unwrap().workers.remove(&worker_id); + } + })); + let error = runtime.create_worker(request).unwrap_err(); + let RuntimeError::WorkerCreateFailed { failure, source } = error else { + panic!("post-Connected read failure must remain a create failure"); + }; + assert!(matches!(*source, RuntimeError::WorkerNotFound { .. })); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown); + assert_eq!( + failure.operation().finalize.unwrap().stage, + FinalizeStage::InitialInput + ); + assert_eq!( + boundary_reason(&failure), + StartupBoundaryReason::FinalizeFailed + ); + assert!(backend.dispatched_inputs.lock().unwrap().is_empty()); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); +} + +#[test] +fn connected_create_publication_failure_preserves_known_input_acceptance() { + let (runtime, backend) = runtime_and_backend(); + let mut request = task_request("publication fails after known acceptance"); + request.initial_input = Some(WorkerInput::user("accepted once")); + let worker_ref = WorkerRef::new(request.worker_id); + // Substitute only Connected data. The invalid backend Workdir identifier + // reaches the real subscription projection after the owner commit. + backend + .contexts + .lock() + .unwrap() + .insert(request.worker_id, runtime.execution_context(worker_ref)); + *backend.spawn_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::Connected { + worker_state: protocol::WorkerStatus::Idle.into(), + workdir_attachments: vec![WorkingDirectoryAttachmentStatus { + alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory: serde_json::from_value(serde_json::json!({ + "summary": { + "working_directory_id": "", + "repository_id": "repo", + "materializer_kind": "runtime_git_clone", + "status": "active" + } + })) + .unwrap(), + }], + }); + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted); + assert_eq!( + failure.operation().finalize, + Some(FinalizeFacts { + stage: FinalizeStage::Publication, + compensation: CompensationOutcome::Completed, + }) + ); + assert_eq!( + boundary_reason(&failure), + StartupBoundaryReason::FinalizeFailed + ); + let RuntimeError::WorkerCreateFailed { source, .. } = error else { + unreachable!() + }; + assert!(matches!(*source, RuntimeError::InvalidRequest(_))); + assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); +} + +#[test] +fn create_correlation_uses_bound_runtime_identity_for_connected_and_spawn_failures() { + for connected in [false, true] { + let (runtime, backend) = runtime_and_backend(); + runtime.bind_runtime_identity("runtime-identity").unwrap(); + let mut request = task_request("safe correlation"); + let worker_id = request.worker_id; + if connected { + request.initial_input = Some(WorkerInput::user("start once")); + backend.set_dispatch_result(rejected_input()); + } else { + let mut result = rejected_input(); + result.operation = WorkerExecutionOperation::Spawn; + *backend.spawn_result.lock().unwrap() = + Some(WorkerExecutionSpawnResult::Rejected(result)); + } + let error = runtime.create_worker(request).unwrap_err(); + assert_eq!( + error.create_failure().unwrap().operation().correlation, + Some(StartupCorrelation::for_worker( + "runtime-identity", + &worker_id.to_string() + )) + ); + assert_eq!(error.to_string(), "worker creation failed"); + } +} + +#[cfg(feature = "fs-store")] +#[test] +fn connected_create_commit_write_faults_preserve_acceptance_and_io_facts() { + use crate::fs_store::AtomicWriteFault; + for fault in [ + AtomicWriteFault::BeforeRename, + AtomicWriteFault::AfterRename, + ] { + for has_initial_input in [false, true] { + let root = tempfile::tempdir().unwrap(); + let options = FsRuntimeStoreOptions::new(root.path().join("runtime")) + .with_runtime_id("create-commit-fault"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let runtime = runtime.clone(); + move || { + runtime + .lock() + .unwrap() + .fs_store() + .unwrap() + .fail_next_worker_write(fault) + } + })); + let mut request = task_request("create commit fault"); + if has_initial_input { + request.initial_input = Some(WorkerInput::user("accepted once")); + } + let worker_id = request.worker_id; + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted); + assert_eq!( + failure.operation().finalize, + Some(FinalizeFacts { + stage: FinalizeStage::Publication, + compensation: CompensationOutcome::Completed, + }) + ); + assert_eq!( + boundary_reason(&failure), + StartupBoundaryReason::FinalizeFailed + ); + assert_eq!(failure.startup().io_kind, Some(StartupIoKind::Other)); + assert_eq!(failure.startup().persistence, None); + assert!( + failure + .startup() + .secondary + .contains(&StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: if fault == AtomicWriteFault::AfterRename { + StartupPersistenceOutcome::OutcomeUnknown + } else { + StartupPersistenceOutcome::NotSaved + }, + }) + ); + let RuntimeError::WorkerCreateFailed { source, .. } = error else { + unreachable!() + }; + assert!(matches!( + *source, + RuntimeError::StoreIo { .. } | RuntimeError::StoreCommitOutcomeUnknown { .. } + )); + assert_eq!( + backend.dispatched_inputs.lock().unwrap().len(), + usize::from(has_initial_input) + ); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + assert!( + !options + .root + .join("workers") + .join(worker_id.to_string()) + .exists() + ); + drop(runtime); + drop(backend); + let reopened = Runtime::with_fs_store(options).unwrap(); + assert!(reopened.list_workers().unwrap().is_empty()); + } + } +} + +#[cfg(feature = "fs-store")] +#[test] +fn connected_create_cleanup_persistence_faults_never_replace_primary() { + for delete_fault in [false, true] { + let root = tempfile::tempdir().unwrap(); + let store_root = root.path().join("runtime"); + let options = + FsRuntimeStoreOptions::new(&store_root).with_runtime_id("create-cleanup-fault"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let mut request = task_request("create cleanup persistence fault"); + request.initial_input = Some(WorkerInput::user("rejected once")); + let worker_dir = store_root + .join("workers") + .join(request.worker_id.to_string()); + let runtime_path = store_root.join("runtime.json"); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new(move || { + if delete_fault { + std::fs::create_dir(worker_dir.join("metadata.json")).unwrap(); + } else { + std::fs::rename(&runtime_path, runtime_path.with_extension("saved")).unwrap(); + std::fs::create_dir(&runtime_path).unwrap(); + } + })); + let original = rejected_input(); + backend.set_dispatch_result(original.clone()); + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(error.execution_result(), Some(&original)); + assert_eq!(failure.startup().primary, original.startup_failure.unwrap()); + assert_eq!(failure.startup().io_kind, original.startup_io_kind); + assert_eq!(failure.operation().execution, ExecutionOutcome::Errored); + assert_eq!( + failure.operation().finalize.unwrap().compensation, + CompensationOutcome::Unknown + ); + assert!( + failure + .startup() + .secondary + .contains(&StartupSecondaryFailure::Persistence { + phase: if delete_fault { + StartupFailurePhase::RuntimeRecordDelete + } else { + StartupFailurePhase::RuntimeSnapshotSave + }, + outcome: StartupPersistenceOutcome::OutcomeUnknown, + }) + ); + if !delete_fault { + assert!( + failure + .startup() + .secondary + .contains(&StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeRecordDelete, + outcome: StartupPersistenceOutcome::Persisted, + }) + ); + } + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1); + } +} + +#[test] +fn connected_create_retains_cleanup_owner_resource_and_metadata_observations() { + let (runtime, backend) = runtime_and_backend(); + backend.set_dispatch_result(WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + )); + let mut stop = + WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, "TEST_ONLY_SECRET"); + let resources = StartupResourceFacts { + resources: StartupResources::Unproven, + cleanup: StartupCleanup::Failed(StartupCleanupFailure::Feature), + durable_effects_possible: true, + }; + stop.startup_resources = Some(resources); + stop.startup_io_kind = Some(StartupIoKind::TimedOut); + stop.startup_failure_persistence = Some(StartupPersistenceOutcome::OutcomeUnknown); + backend.set_stop_result(stop); + let mut request = task_request("cleanup owner observations"); + request.initial_input = Some(WorkerInput::user("start once")); + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(failure.startup().resources, Some(resources)); + assert_eq!(failure.startup().io_kind, Some(StartupIoKind::TimedOut)); + assert_eq!(failure.startup().persistence, None); + assert!( + failure + .startup() + .secondary + .contains(&StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome: StartupPersistenceOutcome::OutcomeUnknown, + }) + ); + assert_eq!( + failure.operation().finalize.unwrap().compensation, + CompensationOutcome::Unknown + ); + assert_eq!( + boundary_reason(&failure), + StartupBoundaryReason::InvalidResponse + ); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); +} + +#[cfg(feature = "fs-store")] +#[test] +fn connected_create_unknown_cleanup_retention_write_failure_preserves_primary() { + use crate::fs_store::AtomicWriteFault; + let root = tempfile::tempdir().unwrap(); + let options = FsRuntimeStoreOptions::new(root.path().join("runtime")) + .with_runtime_id("create-retain-fault"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let runtime = runtime.clone(); + move || { + runtime + .lock() + .unwrap() + .fs_store() + .unwrap() + .fail_next_worker_write(AtomicWriteFault::BeforeRename) + } + })); + let original = rejected_input(); + backend.set_dispatch_result(original.clone()); + // Accepted stop without a termination receipt cannot authorize deletion. + backend.set_stop_result(WorkerExecutionResult::accepted( + WorkerExecutionOperation::Stop, + )); + let mut request = task_request("unknown cleanup retention persistence failure"); + request.initial_input = Some(WorkerInput::user("rejected once")); + let error = runtime.create_worker(request).unwrap_err(); + let failure = error.create_failure().unwrap(); + assert_eq!(error.execution_result(), Some(&original)); + assert_eq!( + failure.operation().finalize.unwrap().compensation, + CompensationOutcome::Unknown + ); + assert!( + failure + .startup() + .secondary + .contains(&StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::NotSaved, + }) + ); + assert_eq!(runtime.list_workers().unwrap().len(), 1); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1); + assert!( + backend + .submission_receipt_queries + .lock() + .unwrap() + .is_empty() + ); +} diff --git a/crates/worker-runtime/src/runtime/restore_continuation.rs b/crates/worker-runtime/src/runtime/restore_continuation.rs index ba8e884c7..0ad6d69ae 100644 --- a/crates/worker-runtime/src/runtime/restore_continuation.rs +++ b/crates/worker-runtime/src/runtime/restore_continuation.rs @@ -479,7 +479,7 @@ fn run( } Ok(Step::Waiting) | Ok(Step::Contended) => None, Ok(Step::Retry(reason)) => Some(reason), - Err(error) => Some(sanitize_worker_create_failure_message(&error.to_string())), + Err(error) => Some(sanitize_worker_restore_failure_message(&error.to_string())), }; if let Some(reason) = &reason { repeated = repeated.saturating_add(1); diff --git a/crates/worker-runtime/src/startup_observation.rs b/crates/worker-runtime/src/startup_observation.rs new file mode 100644 index 000000000..dbec359cd --- /dev/null +++ b/crates/worker-runtime/src/startup_observation.rs @@ -0,0 +1,205 @@ +//! The Runtime creation responsibility sink records only projected typed fields. +//! Source/provider strings are never safe merely because a blacklist did not match. +//! Wrappers/projectors do not write events; a failing sink cannot change the result. +use crate::{error::RuntimeError, identity::WorkerId}; + +pub(crate) fn write_create_failure( + worker_id: WorkerId, + workspace_id: Option<&str>, + error: &RuntimeError, + elapsed: std::time::Duration, +) { + let failure = error.create_failure(); + let projected = failure + .as_ref() + .and_then(|failure| failure.telemetry().ok()); + let fields = projected.as_ref().map(error_contract::observation_fields); + let operation = failure.as_ref().map(|failure| failure.operation()); + tracing::error!( + target: "yoi::worker_create", + event = "worker_create_failed", + component = "runtime", + workspace_id = workspace_id.unwrap_or(""), + worker_id = %worker_id, + domain = fields.as_ref().map_or("startup_boundary", |fields| fields.domain), + reason = fields.as_ref().map_or("observation_unavailable", |fields| fields.reason), + operation = fields.as_ref().and_then(|fields| fields.operation), + stage = fields.as_ref().and_then(|fields| fields.stage), + correlation = operation.and_then(|operation| operation.correlation.as_ref()).map(|value| value.as_str()), + acceptance = ?operation.map(|operation| operation.acceptance), + execution = ?operation.map(|operation| operation.execution), + finalize = ?operation.and_then(|operation| operation.finalize), + persistence = ?failure.as_ref().and_then(|failure| failure.startup().persistence), + resources = ?failure.as_ref().and_then(|failure| failure.startup().resources), + duration_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX), + "Worker creation returned a typed failure" + ); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::catalog::{CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource}; + use crate::execution::{ + WorkerExecutionBackend, WorkerExecutionOperation, WorkerExecutionResult, + WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, + }; + use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput}; + use crate::{Runtime, RuntimeOptions}; + use runtime_api::startup::*; + use std::collections::BTreeMap; + use std::sync::{Arc, Mutex}; + use tracing_subscriber::prelude::*; + + const CANARIES: &[&str] = &[ + "aQ7xV2zN9mL5", + "/private/startup/provider", + "Bearer q3fV8aJ6", + "token=K7sM2pQ4", + ]; + struct FailingBackend; + impl WorkerExecutionBackend for FailingBackend { + fn backend_id(&self) -> &str { + "log-owner-fixture" + } + fn dispatch_input( + &self, + _: &crate::identity::WorkerRef, + _: crate::interaction::WorkerInput, + ) -> WorkerExecutionResult { + panic!("rejected creation never dispatches input") + } + fn spawn_worker(&self, _: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + let result = + WorkerExecutionResult::errored(WorkerExecutionOperation::Spawn, CANARIES.join(" ")) + .with_startup_failure( + worker::startup_failure::runtime_startup_attachment( + runtime_api::RuntimeStartupCause { + stage: runtime_api::StartupStage::Controller, + reason: runtime_api::RuntimeStartupReason::StartupFailed, + }, + ) + .unwrap(), + ); + WorkerExecutionSpawnResult::Errored(result) + } + } + #[derive(Clone, Default)] + struct Events(Arc>>>); + struct Fields(BTreeMap); + impl tracing::field::Visit for Fields { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.insert(field.name().into(), value.into()); + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().into(), format!("{value:?}")); + } + } + impl tracing_subscriber::Layer for Events { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().target() == "yoi::worker_create" { + let mut fields = Fields(BTreeMap::new()); + event.record(&mut fields); + self.0.lock().unwrap().push(fields.0); + } + } + } + fn request() -> CreateWorkerRequest { + let archive = ProfileSourceArchive::build(ProfileSourceArchiveInput { + id: "log-profile".into(), + entrypoints: BTreeMap::from([("builtin:coder".into(), "profiles/coder.dcdl".into())]), + imports: BTreeMap::new(), + sources: BTreeMap::from([("profiles/coder.dcdl".into(), "{}".into())]), + }) + .unwrap(); + CreateWorkerRequest { + worker_id: WorkerId::now_v7(), + create_fingerprint: "log-create".into(), + profile: ProfileSelector::Builtin("builtin:coder".into()), + display_name: None, + profile_source: ProfileSourceArchiveSource::Embedded { archive }, + config_bundle: None, + initial_input: None, + workdir_attachment_requests: Vec::new(), + workdir_attachments: Vec::new(), + worker_observation_enabled: false, + worker_observation_grants: Vec::new(), + workspace_api: None, + memory_settings: None, + plugin_execution: None, + subjektiv_attached: false, + backend_job: None, + } + } + #[test] + fn real_runtime_creation_sink_omits_unlabelled_provider_canaries_and_matches_typed_result() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(FailingBackend)) + .unwrap(); + runtime + .bind_runtime_identity("runtime-log-fixture") + .unwrap(); + let request = request(); + let worker_id = request.worker_id; + let events = Events::default(); + let subscriber = tracing_subscriber::registry().with(events.clone()); + let error = + tracing::subscriber::with_default(subscriber, || runtime.create_worker(request)) + .unwrap_err(); + let failure = error + .create_failure() + .expect("creation owner must retain a typed cause"); + let wire = failure + .problem( + StartupBoundary::Runtime, + &SessionFailureVisibility { + can_view_cause: true, + }, + "", + ) + .unwrap(); + let encoded = failure.encode_storage().unwrap(); + let DecodeStoredWorkerCreateFailure::Known(saved) = + decode_stored_worker_create_failure(&encoded) + else { + panic!("known storage"); + }; + assert_eq!(saved.public_facts(), failure.public_facts()); + let captured = events.0.lock().unwrap(); + assert_eq!(captured.len(), 1); + let event = &captured[0]; + let projection = failure.telemetry().unwrap(); + let fields = error_contract::observation_fields(&projection); + assert_eq!(event["domain"], fields.domain); + assert_eq!(event["reason"], fields.reason); + assert_eq!( + event["correlation"], + StartupCorrelation::for_worker("runtime-log-fixture", &worker_id.to_string()).as_str() + ); + for canary in CANARIES { + assert!(!format!("{captured:?}").contains(canary)); + assert!(!String::from_utf8_lossy(wire.body()).contains(canary)); + assert!(!encoded.contains(canary)); + } + assert!(!event.contains_key("execution_failure_message")); + } + #[test] + fn unsupported_failure_logs_only_fixed_fallback_not_source_text() { + let error = RuntimeError::InvalidRequest(CANARIES.join(" ")); + let events = Events::default(); + let subscriber = tracing_subscriber::registry().with(events.clone()); + tracing::subscriber::with_default(subscriber, || { + write_create_failure(WorkerId::now_v7(), None, &error, std::time::Duration::ZERO) + }); + let captured = events.0.lock().unwrap(); + assert_eq!(captured.len(), 1); + assert_eq!(captured[0]["reason"], "observation_unavailable"); + for canary in CANARIES { + assert!(!format!("{captured:?}").contains(canary)); + } + } +} diff --git a/crates/workspace-server/src/backend_job_upgrade_tests.rs b/crates/workspace-server/src/backend_job_upgrade_tests.rs index fe445887e..6df37b38a 100644 --- a/crates/workspace-server/src/backend_job_upgrade_tests.rs +++ b/crates/workspace-server/src/backend_job_upgrade_tests.rs @@ -1476,6 +1476,21 @@ fn captured_history_is_accepted_at_each_real_retained_prefix_and_cannot_be_forge let dir = tempfile::tempdir().unwrap(); let conn = legacy_fixture_to_85(&dir.path().join("prefixes.sqlite"), 85); let original = pre_content_values(&conn, false); + let mut expected_receipts = rows( + &conn, + "SELECT version,name FROM __yoi_schema_migrations ORDER BY version", + ); + expected_receipts.extend( + store_owner::MIGRATIONS + .iter() + .filter(|migration| migration.version > 85) + .map(|migration| { + vec![ + Value::Integer(migration.version), + Value::Text(migration.name.into()), + ] + }), + ); for migration in store_owner::MIGRATIONS .iter() .filter(|migration| migration.version >= 86 && migration.version <= 122) @@ -1496,10 +1511,13 @@ fn captured_history_is_accepted_at_each_real_retained_prefix_and_cannot_be_forge &conn, "SELECT version,name FROM __yoi_schema_migrations ORDER BY version", ); + assert_eq!(receipts, expected_receipts); assert_eq!( - receipts.last().unwrap(), - &vec![Value::Integer(123), Value::Text(MIGRATION_NAME.into())] + store_owner::current_schema_version(&conn).unwrap(), + store_owner::LATEST_SCHEMA_VERSION ); + super::super::storage_formats::verify_history(&conn, store_owner::LATEST_SCHEMA_VERSION) + .unwrap(); assert_eq!(original, pre_content_values(&conn, true)); assert!(conn.execute_batch("INSERT INTO backend_job_pre_content_history SELECT * FROM backend_job_pre_content_history LIMIT 1").is_err()); } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index cdb1387cc..509ba4dad 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -3316,14 +3316,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { diagnostics, }, Err(err) => { - let failure = match &err { - EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } => { - worker_runtime::execution::startup_create_failure(result) - .ok() - .flatten() - } - _ => None, - }; + let failure = err.create_failure(); diagnostics.push(embedded_runtime_diagnostic(&err)); let state = failure.as_ref().map_or( InternalWorkerOperationState::Rejected, @@ -6954,6 +6947,11 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti message, ), ), + EmbeddedRuntimeError::WorkerCreateFailed { failure, .. } => diagnostic( + "embedded_worker_create_failed", + HostDiagnosticSeverity::Warning, + worker::startup_failure::internal_startup_failure_summary(&failure.startup().primary), + ), EmbeddedRuntimeError::WorkerExecutionRejected { operation, outcome, @@ -7036,6 +7034,9 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti }) }); } + if let EmbeddedRuntimeError::WorkerCreateFailed { failure, .. } = error { + projection.startup = internal_startup_transfer(failure.startup().clone()); + } projection } @@ -8904,10 +8905,9 @@ mod tests { assert!(spawned.acceptance_evidence.is_empty()); assert_eq!(spawned.diagnostics.len(), 1); assert!(spawned.diagnostics.iter().any(|diagnostic| { - diagnostic.code == "embedded_worker_execution_rejected" - && diagnostic - .message - .contains("provider setup failed at [redacted]") + diagnostic.code == "embedded_worker_create_failed" + && diagnostic.startup.is_some() + && !diagnostic.message.contains("provider setup failed") && !diagnostic.message.contains("/tmp/secret-provider-config") && !diagnostic.message.contains("secret-value") && !diagnostic.message.contains("session-42") @@ -10064,7 +10064,9 @@ fn startup_failure_operation_state( failure: &runtime_api::startup::WorkerCreateFailure, ) -> InternalWorkerOperationState { use runtime_api::startup::{CreationAcceptance, ExecutionOutcome}; - if failure.operation().acceptance == CreationAcceptance::Unknown { + if failure.operation().acceptance == CreationAcceptance::Unknown + || failure.operation().execution == ExecutionOutcome::Unknown + { InternalWorkerOperationState::Unknown } else if failure.operation().execution == ExecutionOutcome::Unsupported { InternalWorkerOperationState::Unsupported diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 0c60cc702..0dc4f12b5 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -35096,12 +35096,15 @@ fn browser_worker_response_for_committed_summary( let workspace_id = api.workspace_id().to_string(); let resource_key = api .store - .worker_resource_key(&workspace_id, &worker.worker)? - .ok_or_else(|| { - Error::Store(format!( - "Workspace Worker `{worker_id}` has no resource key after registration" - )) - })?; + .worker_resource_key(&workspace_id, &worker.worker) + .and_then(|key| { + key.ok_or_else(|| { + Error::Store(format!( + "Workspace Worker `{worker_id}` has no resource key after registration" + )) + }) + }) + .map_err(|error| worker_startup::publication_error(api, &worker.worker, error.into()))?; let worker = worker_launch_worker_summary(worker); let console_href = browser_worker_console_href(&workspace_id, &resource_key); Ok(BrowserCreateWorkerResponse { diff --git a/crates/workspace-server/src/server/tests/startup_result_tests.rs b/crates/workspace-server/src/server/tests/startup_result_tests.rs index 01be8db35..731b806fb 100644 --- a/crates/workspace-server/src/server/tests/startup_result_tests.rs +++ b/crates/workspace-server/src/server/tests/startup_result_tests.rs @@ -1,8 +1,9 @@ -//! Backend result fixture: real SQLite reservation -> coordinator helper -> typed -//! read -> public ApiError response -> HTTP middleware observation. Only the route's -//! startup producer is supplied by the fixture; this does not test authorization, -//! Session callbacks, Runtime startup or the Browser creation route. Those belong -//! to session_registration_connected_tests. Runs in the ordinary server lib suite; +//! Backend result fixtures use real SQLite reservations, coordinator helpers, typed +//! reads, public ApiError responses and HTTP middleware observation. Publication +//! faults enter the production committed-summary response helper with a fixture +//! Runtime Worker; only the earlier Runtime startup is substituted. These cases do +//! not claim creation-route authorization or Session callback coverage (see +//! session_registration_connected_tests). Runs in the ordinary server lib suite; //! each case owns its tempdir/DB and subscriber, with no external services. use super::*; @@ -82,6 +83,10 @@ fn assert_semantics(actual: &WorkerCreateFailure, expected: &WorkerCreateFailure } fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef { + reserve_for_runtime(api, EMBEDDED_WORKER_RUNTIME_ID) +} + +fn reserve_for_runtime(api: &WorkspaceApi, runtime_id: &str) -> RuntimeWorkerRef { let settings = api .config_store .get_workspace_memory_settings(TEST_WORKSPACE_ID) @@ -90,7 +95,7 @@ fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef { .config_store .reserve_worker_create( TEST_WORKSPACE_ID, - EMBEDDED_WORKER_RUNTIME_ID, + runtime_id, "manual:startup-result-fixture", "sha256:startup-result-fixture", None, @@ -98,7 +103,7 @@ fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef { ) .unwrap(); RuntimeWorkerRef { - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(), + runtime_id: runtime_id.into(), worker_id: reservation.worker_id.to_string(), } } @@ -123,6 +128,9 @@ fn lifecycle_snapshot(api: &WorkspaceApi) -> Vec> { format!("SELECT {columns} FROM worker_create_reservations ORDER BY workspace_id,allocation_key"), "SELECT * FROM workspace_resource_keys ORDER BY workspace_id,resource_kind,resource_id".into(), "SELECT * FROM worker_singleton_owners ORDER BY workspace_id,singleton_key".into(), + "SELECT * FROM worker_registry ORDER BY workspace_id,runtime_id,worker_id".into(), + "SELECT * FROM ticket_worker_assignments ORDER BY workspace_id,ticket_id,assignment_id".into(), + "SELECT * FROM ticket_assignment_operations ORDER BY workspace_id,operation_id".into(), ]; let mut snapshot = Vec::new(); for query in queries { @@ -455,6 +463,290 @@ async fn context_and_public_projection_are_silent_until_explicit_http_observatio assert_log(&events[0], &expected, StatusCode::CONFLICT); } +async fn assert_publication_fault_retains_committed_worker(query_failure: bool) { + let dir = tempfile::tempdir().unwrap(); + let api = test_api(dir.path()).await; + // Freeze unrelated observation writes before deliberately damaging the DB. + api.worker_projection.shutdown_async().await; + api.store.db_executor().drain().await.unwrap(); + register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; + let stops = Arc::new(AtomicUsize::new(0)); + let stop_count = stops.clone(); + let runtime = WorkdirlessFixtureRuntime { + stop_callback: Some(Arc::new(move || { + stop_count.fetch_add(1, Ordering::SeqCst); + })), + ..Default::default() + }; + api.runtime.register_or_replace(runtime.clone()); + let worker = reserve_for_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID); + let spawned = crate::hosts::WorkspaceWorkerRuntime::spawn_worker( + &runtime, + WorkerCreateBinding { + plugin_execution: None, + worker_id: WorkerId::parse(&worker.worker_id).unwrap(), + create_fingerprint: "sha256:startup-result-fixture".into(), + }, + WorkerSpawnRequest { + intent: WorkerSpawnIntent::WorkspaceCompanion, + requested_worker_name: Some("Publication fixture".into()), + singleton_key: None, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: ProfileSelector::Builtin("builtin:companion".into()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachment_requests: Vec::new(), + resolved_workdir_attachment_requests: Vec::new(), + resolved_workdir_attachments: Vec::new(), + resolved_config_bundle: None, + resolved_worker_observation_enabled: false, + resolved_worker_observation_grants: Vec::new(), + resolved_workspace_api: None, + resolved_memory_settings: None, + resolved_subjektiv_attached: false, + resolved_control_operation: None, + resolved_execution_grant: None, + }, + ); + assert_eq!(spawned.state, InternalWorkerOperationState::Accepted); + let summary = spawned.worker.unwrap(); + assert_eq!(summary.worker, worker); + let record = record_worker_summary( + &api, + &summary, + &summary.display_name, + summary.profile.clone(), + WorkerRegistryDisplayNamePolicy::UseProvided, + ) + .unwrap(); + api.config_store + .complete_worker_create_reservation( + TEST_WORKSPACE_ID, + WorkerId::parse(&worker.worker_id).unwrap(), + ) + .unwrap(); + let ticket = browser_ticket_backend(&api) + .unwrap() + .create(ticket::NewTicket::new("Publication retains assignment")) + .unwrap(); + let assignment = TicketWorkerAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + ticket_id: ticket.id.clone(), + assignment_id: "publication-assignment".into(), + worker: worker.clone(), + assigned_by: "test-user".into(), + assigned_at: TEST_CREATED_AT.into(), + }; + api.store + .set_current_ticket_worker_assignment( + &assignment, + None, + "publication-assignment-event", + "publication-assignment-operation", + false, + ) + .unwrap(); + // Establish a working publication read first; the fault changes only that + // final read, not reservation persistence or an earlier caller-identity check. + let key = api + .store + .worker_resource_key(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + api.config_store + .with_conn(|conn| { + if query_failure { + conn.execute_batch( + "ALTER TABLE workspace_resource_keys RENAME COLUMN resource_key TO \"sqlite-secret\";", + )?; + } else { + let deleted = conn.execute( + "DELETE FROM workspace_resource_keys WHERE workspace_id=?1 AND resource_kind='worker' AND resource_key=?2", + rusqlite::params![TEST_WORKSPACE_ID, key], + )?; + assert_eq!(deleted, 1); + } + Ok(()) + }) + .unwrap(); + let before = lifecycle_snapshot(&api); + let capture = Capture::default(); + let route_api = api.clone(); + let app = Router::new() + .route( + "/api/startup-result-fixture", + axum::routing::get(move || { + let api = route_api.clone(); + let summary = summary.clone(); + async move { + let error = browser_worker_response_for_committed_summary( + &api, + summary, + "Publication fixture".into(), + Vec::new(), + ) + .unwrap_err(); + let Error::WorkerCreateFailed { source, .. } = &error.error else { + panic!("final publication read must return a typed result: {error:?}"); + }; + let source = source + .as_deref() + .expect("keep the actual read error internal"); + if query_failure { + assert!(matches!(source, Error::Sqlite(_)), "{source:?}"); + } else { + assert!(matches!(source, Error::Store(_)), "{source:?}"); + assert!(source.to_string().contains("no resource key")); + } + error.into_response() + } + }), + ) + .layer(middleware::from_fn(log_failed_api_response)); + let (status, problem, public) = public_result(app) + .with_subscriber(subscriber(capture.clone())) + .await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + primary_cause(&public), + StartupCause::Boundary(runtime_api::startup::StartupBoundaryCause { + reason: runtime_api::startup::StartupBoundaryReason::FinalizeFailed, + }) + ); + assert_eq!(public.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(public.operation().execution, ExecutionOutcome::Accepted); + assert_eq!( + public.operation().finalize, + Some(FinalizeFacts { + stage: FinalizeStage::Publication, + compensation: CompensationOutcome::Residual, + }) + ); + assert_eq!(public.operation().backend_compensation, None); + assert_eq!( + public.startup().resources, + Some(protocol::startup_failure::StartupResourceFacts { + resources: protocol::startup_failure::StartupResources::WorkerStartup, + cleanup: protocol::startup_failure::StartupCleanup::NotNeeded, + durable_effects_possible: true, + }) + ); + assert_eq!( + public.backend_result_persistence(), + Some(StartupPersistenceOutcome::Persisted) + ); + assert_eq!( + public.operation().correlation.as_ref().unwrap().as_str(), + worker_startup::correlation(&worker.runtime_id, &worker.worker_id) + ); + assert!( + problem["instance"] + .as_str() + .unwrap() + .ends_with(public.operation().correlation.as_ref().unwrap().as_str()) + ); + assert_semantics(&retained(&api, &worker), &public); + let events = capture.events(); + assert_eq!(events.len(), 1, "one primary HTTP event: {events:?}"); + let event = &events[0]; + assert_eq!(event["target"], "yoi::worker_startup"); + assert_eq!(event["event"], "worker_create_result"); + assert_eq!(event["domain"], "startup_boundary"); + assert_eq!(event["reason"], "finalize_failed"); + assert_eq!(event["acceptance"], "Accepted"); + assert_eq!(event["execution"], "Accepted"); + assert_eq!( + event["finalize"], + format!("{:?}", public.operation().finalize) + ); + assert_eq!(event["backend_compensation"], "None"); + assert_eq!(event["backend_result_persistence"], "Some(Persisted)"); + assert_eq!( + event["resources"], + format!("{:?}", public.startup().resources) + ); + assert_eq!(event["status"], status.as_u16()); + assert_eq!( + event["correlation"], + problem["yoi_startup"]["operation"]["correlation"] + ); + let exposed = format!("{problem} {events:?}"); + for internal in [ + worker.worker_id.as_str(), + "no such column", + "no resource key", + "workspace_resource_keys", + dir.path().to_str().unwrap(), + ] { + assert!( + !exposed.contains(internal), + "internal source leaked: {exposed}" + ); + } + assert_eq!(lifecycle_snapshot(&api), before); + assert_eq!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &worker) + .unwrap(), + Some(record) + ); + assert_eq!( + api.store + .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id) + .unwrap(), + Some(assignment) + ); + assert_eq!(stops.load(Ordering::SeqCst), 0); + assert_eq!(runtime.spawn_requests().len(), 1, "no reexecution"); + assert_eq!( + runtime.workers.lock().unwrap().len(), + 1, + "no Runtime deletion" + ); +} + +#[tokio::test] +async fn publication_sqlite_read_failure_retains_committed_worker_and_typed_result() { + assert_publication_fault_retains_committed_worker(true).await; +} + +#[tokio::test] +async fn publication_missing_resource_key_retains_committed_worker_and_typed_result() { + assert_publication_fault_retains_committed_worker(false).await; +} + +#[tokio::test] +async fn publication_error_preserves_existing_primary_without_claiming_compensation() { + let dir = tempfile::tempdir().unwrap(); + let api = test_api(dir.path()).await; + let worker = reserve(&api); + let original = failure_fixture(CompensationOutcome::Completed); + let error = worker_startup::publication_error(&api, &worker, api_error(original.clone())); + let Error::WorkerCreateFailed { failure, source } = &error.error else { + panic!("publication must retain a typed startup result"); + }; + assert!(matches!( + source.as_deref(), + Some(Error::WorkerCreateFailed { .. }) + )); + assert_eq!(failure.startup().primary, original.startup().primary); + assert_eq!(failure.startup().secondary, original.startup().secondary); + assert_eq!( + failure.startup().persistence, + original.startup().persistence + ); + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted); + assert_eq!(failure.operation().backend_compensation, None); + assert_eq!( + failure.operation().finalize.as_ref().unwrap().stage, + FinalizeStage::Publication + ); + assert_semantics(&retained(&api, &worker), failure); +} + #[derive(Clone)] struct FailingWriter(Arc); diff --git a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs index 26333ebc5..2615c5d19 100644 --- a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs +++ b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs @@ -12,6 +12,7 @@ use worker_runtime::execution::{ // published Profile projection; leave Worker creation/control to the real Runtime. #[derive(Default)] struct LaunchExecutionBackend { + lose_initial_ack: std::sync::atomic::AtomicBool, fail_stop: std::sync::atomic::AtomicBool, execution: DeterministicExecutionBackend, provider_close_gate: Mutex>>, @@ -197,7 +198,16 @@ impl WorkerExecutionBackend for LaunchExecutionBackend { .lock() .unwrap() .push((worker.clone(), input.clone())); - self.execution.dispatch_input(worker, input) + let mut result = self.execution.dispatch_input(worker, input); + if self + .lose_initial_ack + .swap(false, std::sync::atomic::Ordering::SeqCst) + { + // Lose only the response acknowledgement, after the real fixture + // execution committed its durable receipt. Never redispatch it. + result.submission = None; + } + result } fn termination_receipt( @@ -1160,6 +1170,90 @@ async fn read_only_ticket_targets_allow_extra_external_grants_without_widening_c ); } +/// Production create owner and Backend adapters share the same executor; only +/// its response acknowledgement and cleanup boundary are fault-injected. +#[tokio::test(flavor = "multi_thread")] +async fn initial_input_ack_loss_preserves_creation_facts_across_embedded_and_remote() { + use runtime_api::startup::*; + for failed_cleanup in [false, true] { + let mut observed = Vec::new(); + for remote in [false, true] { + let fixture = LaunchFixture::new(remote).await; + let ticket = fixture.ticket(); + let payload = fixture.payload(&ticket, "lost-initial-ack", "builtin:companion", false); + fixture + .backend + .lose_initial_ack + .store(true, std::sync::atomic::Ordering::SeqCst); + fixture + .backend + .fail_stop + .store(failed_cleanup, std::sync::atomic::Ordering::SeqCst); + let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await; + assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}"); + let DecodeStartupProblem::Known(failure) = decode_startup_problem( + &serde_json::to_vec(&body).unwrap(), + status.as_u16(), + PROBLEM_CONTENT_TYPE, + StartupBoundary::Browser, + ) else { + panic!("must retain typed creation facts: {body}"); + }; + assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted); + assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown); + let finalize = failure.operation().finalize.unwrap(); + assert_eq!(finalize.stage, FinalizeStage::InitialInput); + assert_eq!( + finalize.compensation, + if failed_cleanup { + CompensationOutcome::Unknown + } else { + CompensationOutcome::Completed + } + ); + assert_eq!(body["yoi_failure"]["reason"], "invalid_response"); + assert!( + !body + .to_string() + .contains("TEST_ONLY_PRIVATE_STOP_CREDENTIAL") + ); + let creates = fixture.backend.creates.lock().unwrap().clone(); + assert_eq!(creates.len(), 1); + let worker_id = creates[0].worker_id.to_string(); + let worker = RuntimeWorkerRef::new(&fixture.runtime_id, &worker_id); + let Some(DecodeStoredWorkerCreateFailure::Known(saved)) = + fixture.api.worker_create_failure(&worker).unwrap() + else { + panic!("retain queryable create result"); + }; + assert_eq!(saved.public_facts(), failure.public_facts()); + assert_eq!( + fixture.backend.submits_for(&worker_id).len(), + 1, + "observation failure never redispatches" + ); + assert_eq!( + fixture + .backend + .execution + .contexts + .lock() + .unwrap() + .keys() + .any(|worker| worker.worker_id.to_string() == worker_id), + failed_cleanup + ); + let mut facts = failure.public_facts(); + facts.operation.correlation = None; + observed.push((status, body["yoi_failure"]["reason"].clone(), facts)); + } + assert_eq!( + observed[0], observed[1], + "same owner facts through Embedded and Remote" + ); + } +} + #[tokio::test(flavor = "multi_thread")] async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_and_preserves_open_case() { diff --git a/crates/workspace-server/src/server/worker_startup.rs b/crates/workspace-server/src/server/worker_startup.rs index 6d4fb308e..0b92ca985 100644 --- a/crates/workspace-server/src/server/worker_startup.rs +++ b/crates/workspace-server/src/server/worker_startup.rs @@ -109,6 +109,60 @@ pub(super) fn finalize_failure( .expect("locally constructed finalization facts fit the shared boundary budget") } +/// Publication failure cannot undo committed creation or authorize reexecution. +/// Keep the Worker and its assignment: `Residual` describes deliberate retention, +/// not attempted rollback; Backend compensation remains unattempted (`None`). +/// The original store error stays internal while the independent reservation sink +/// retains only the typed result and a path-free correlation. +pub(super) fn publication_error( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + source: ApiError, +) -> ApiError { + use protocol::startup_failure::{StartupCleanup, StartupResourceFacts, StartupResources}; + use runtime_api::startup::{ + CompensationOutcome, FinalizeFacts, FinalizeStage, StartupBoundaryCause, + StartupBoundaryReason, + }; + + let operation = StartupOperationFacts { + acceptance: CreationAcceptance::Accepted, + execution: ExecutionOutcome::Accepted, + finalize: Some(FinalizeFacts { + stage: FinalizeStage::Publication, + compensation: CompensationOutcome::Residual, + }), + backend_compensation: None, + correlation: None, + }; + let failure = match &source.error { + Error::WorkerCreateFailed { failure, .. } => failure.as_ref().clone(), + _ => WorkerCreateFailure::from_cause( + StartupCause::Boundary(StartupBoundaryCause { + reason: StartupBoundaryReason::FinalizeFailed, + }), + operation.clone(), + ) + .expect("enum-only publication cause fits the boundary budget"), + }; + let (mut startup, _) = failure.into_parts(); + startup.resources = Some(StartupResourceFacts { + resources: StartupResources::WorkerStartup, + cleanup: StartupCleanup::NotNeeded, + durable_effects_possible: true, + }); + let failure = WorkerCreateFailure::new(startup, operation) + .expect("locally constructed publication facts fit the shared boundary budget"); + let failure = persist(api, worker, failure); + ApiError::with_diagnostics( + Error::WorkerCreateFailed { + failure: Box::new(failure), + source: Some(Box::new(source.error)), + }, + source.diagnostics, + ) +} + pub(super) fn unknown_result() -> WorkerCreateFailure { WorkerCreateFailure::from_cause( StartupCause::Boundary(runtime_api::startup::StartupBoundaryCause { @@ -196,17 +250,9 @@ pub(super) fn persist( } pub(super) fn correlation(runtime_id: &str, worker_id: &str) -> String { - use sha2::{Digest, Sha256}; - let mut hash = Sha256::new(); - hash.update(runtime_id.as_bytes()); - hash.update([0]); - hash.update(worker_id.as_bytes()); - let digest = hash - .finalize() - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::(); - format!("worker-{digest}") + runtime_api::startup::StartupCorrelation::for_worker(runtime_id, worker_id) + .as_str() + .to_owned() } /// Only the HTTP result boundary records the primary failure. Context wrappers, diff --git a/crates/workspace-server/src/subject_assignment_migration_tests.rs b/crates/workspace-server/src/subject_assignment_migration_tests.rs index 4323c1604..d8a758519 100644 --- a/crates/workspace-server/src/subject_assignment_migration_tests.rs +++ b/crates/workspace-server/src/subject_assignment_migration_tests.rs @@ -852,8 +852,6 @@ mod subject_assignment_migration_tests { }; const PLAN_NAME: &str = "Immutable Worker Plugin plans and retained pins"; - const SUBJECT_NAME: &str = - "Subject Ticket responsibility and current-body notification routing"; type Rows = Vec>; fn raw(path: &Path) -> Connection { @@ -1038,8 +1036,23 @@ mod subject_assignment_migration_tests { } else { String::new() }; + // Keep append identities for rowid tables; current content + // history tables are WITHOUT ROWID, so compare every column + // in deterministic value order instead of inventing rowids. + let without_rowid: bool = conn + .query_row( + "SELECT wr FROM pragma_table_list WHERE schema='main' AND name=?1", + [&table], + |row| row.get(0), + ) + .unwrap(); + let (projection, order) = if without_rowid { + (columns.clone(), columns) + } else { + (format!("rowid,{columns}"), "rowid".into()) + }; let query = format!( - "SELECT rowid,{columns} FROM \"{table}\"{sequence_filter} ORDER BY rowid" + "SELECT {projection} FROM \"{table}\"{sequence_filter} ORDER BY {order}" ); let rows = fixture_query_rows(conn, &query); (query, rows) @@ -1120,49 +1133,15 @@ mod subject_assignment_migration_tests { } fn expected_tail(version: i64) -> Vec<(i64, String)> { - let mut tail = match version { - 96 => vec![ - ( - 97, - "MR notifications and authenticated implementation results".into(), - ), - (98, "Assignment-independent implementation reports".into()), - ( - 99, - "Explicit hosting administrators and self-registration policy".into(), - ), - ( - 100, - "Provisional users and scoped passkey activation".into(), - ), - (101, "Durable explicit Subject body release".into()), - (102, "Durable failed Subject creation recovery".into()), - (103, "Passkey metadata and atomic addition results".into()), - ( - 104, - "Converge Passkey results and Job input retention".into(), - ), - ( - 105, - "Durable Subject deletion snapshots and Job admission fences".into(), - ), - (106, crate::plugin_catalog::MIGRATION_NAME.into()), - (107, PLAN_NAME.into()), - (108, SUBJECT_NAME.into()), - ], - 106 => vec![(107, PLAN_NAME.into()), (108, SUBJECT_NAME.into())], - 107 => vec![(108, SUBJECT_NAME.into())], - _ => panic!("unsupported fixture origin"), - }; - tail.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into())); - tail.push((110, "Worker Plugin launch selection associations".into())); - tail.push(( - 111, - "Typed Plugin Ticket extension data and atomic mutation receipts".into(), - )); - tail.push((112, "Ticket Service outbox and fenced consumers".into())); - tail.push((113, "Ticket work-decomposition parent links".into())); - tail + assert!( + matches!(version, 96 | 106 | 107), + "unsupported fixture origin" + ); + MIGRATIONS + .iter() + .filter(|migration| migration.version > version) + .map(|migration| (migration.version, migration.name.into())) + .collect() } fn plan_steps(version: i64) -> Vec { diff --git a/crates/workspace-server/src/ticket_post_upgrade_tests.rs b/crates/workspace-server/src/ticket_post_upgrade_tests.rs index db803191c..4a9ce8a89 100644 --- a/crates/workspace-server/src/ticket_post_upgrade_tests.rs +++ b/crates/workspace-server/src/ticket_post_upgrade_tests.rs @@ -56,12 +56,17 @@ fn ticket_posts_schema93_preserves_initial_body_creator_comments_and_archived_au drop(conn); let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); assert_eq!(plan.current_schema_version, 93); + assert_eq!(plan.target_schema_version, LATEST_SCHEMA_VERSION); assert_eq!( - plan.migrations + plan.migrations, + MIGRATIONS .iter() - .map(|step| step.version) - .collect::>(), - (94..=LATEST_SCHEMA_VERSION).collect::>() + .filter(|migration| migration.version > 93) + .map(|migration| WorkspaceSchemaMigrationStep { + version: migration.version, + name: migration.name.into(), + }) + .collect::>() ); let conn = Connection::open(&path).unwrap(); assert!(