From 21fe1a7a3167cf9dce70763004fc1f1c2ef02a9f Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 12:18:37 +0900 Subject: [PATCH] fix: bind startup cause persistence to confirmed Session transitions --- crates/protocol/src/startup_failure.rs | 4 + crates/session-store/src/worker_metadata.rs | 3 +- crates/worker-runtime/src/runtime.rs | 57 +++- .../runtime/create_reconciliation_tests.rs | 113 ++++++++ crates/worker-runtime/src/worker_backend.rs | 30 +-- .../worker_backend/startup_binding_tests.rs | 253 ++++++++++++++++++ .../src/worker_backend/startup_failure.rs | 15 ++ crates/worker/src/subjektiv_metadata.rs | 18 +- crates/worker/src/worker.rs | 95 +++++-- web/workspace/src/lib/generated/protocol.ts | 2 +- 10 files changed, 516 insertions(+), 74 deletions(-) create mode 100644 crates/worker-runtime/src/runtime/create_reconciliation_tests.rs create mode 100644 crates/worker-runtime/src/worker_backend/startup_binding_tests.rs diff --git a/crates/protocol/src/startup_failure.rs b/crates/protocol/src/startup_failure.rs index ce94c7765..8a40e48ee 100644 --- a/crates/protocol/src/startup_failure.rs +++ b/crates/protocol/src/startup_failure.rs @@ -196,6 +196,10 @@ pub enum StartupFailurePhase { RuntimeRecordDelete, RuntimeSnapshotSave, RuntimeRemovalPublish, + /// Retaining create reconciliation observations in the canonical Worker record. + RuntimeReconciliationRecordSave, + /// Publishing a retained record to Runtime subscribers; not a durable write. + RuntimeReconciliationPublish, MetadataSave, } diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 2617aa557..bfa5b00d4 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -323,7 +323,8 @@ impl StartupFailureExpectation { }) } - fn matches(&self, metadata: &WorkerMetadata) -> bool { + /// Compare the complete observation without adopting newer ownership or causes. + pub fn matches(&self, metadata: &WorkerMetadata) -> bool { metadata.worker_name == self.worker_name && metadata.workspace_id == self.workspace_id && metadata.active.as_ref().map(|active| active.session_id) == Some(self.session_id) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 3ed474699..3c57b93c0 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1569,7 +1569,7 @@ impl Runtime { }); } WorkerExecutionSpawnResult::ReconciliationRequired { - result, + mut result, worker_state, workdir_attachments, } => { @@ -1582,12 +1582,33 @@ impl Runtime { } else { self.retain_create_reconciliation_pending(&worker_ref, workdir_attachments) }; - if let Err(error) = retain_result { + if let Err((phase, error)) = retain_result { + let outcome = if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) + { + StartupPersistenceOutcome::OutcomeUnknown + } else { + StartupPersistenceOutcome::NotSaved + }; + if phase == StartupFailurePhase::RuntimeReconciliationPublish { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::Persisted, + }); + } + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase, + outcome, + }); tracing::error!( worker_id = %worker_ref.worker_id, %error, "failed to persist Worker create reconciliation evidence" ); + } else { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::Persisted, + }); } return Err(RuntimeError::WorkerExecutionRejected { worker_id: worker_ref.worker_id.clone(), @@ -5193,9 +5214,10 @@ impl Runtime { worker_state: protocol::WorkerStateSnapshot, workdir_attachments: Vec, - ) -> Result<(), RuntimeError> { - let mut state = self.lock()?; - let mut candidate = state.worker(worker_ref)?.clone(); + ) -> Result<(), (StartupFailurePhase, RuntimeError)> { + let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error); + let mut state = self.lock().map_err(record_error)?; + let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone(); candidate.restore_guard.rotate(); candidate.execution_metadata_available = true; @@ -5204,9 +5226,13 @@ impl Runtime { candidate.apply_worker_state(&worker_state); candidate.restore_intent = WorkerRestoreIntent::Automatic; candidate.workdir_attachments = workdir_attachments; - state.persist_worker_record(&candidate)?; + state + .persist_worker_record(&candidate) + .map_err(record_error)?; state.workers.insert(worker_ref.worker_id, candidate); - state.publish_worker_upsert(worker_ref.worker_id)?; + state + .publish_worker_upsert(worker_ref.worker_id) + .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?; Ok(()) } @@ -5228,9 +5254,10 @@ impl Runtime { &self, worker_ref: &WorkerRef, workdir_attachments: Vec, - ) -> Result<(), RuntimeError> { - let mut state = self.lock()?; - let mut candidate = state.worker(worker_ref)?.clone(); + ) -> Result<(), (StartupFailurePhase, RuntimeError)> { + let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error); + let mut state = self.lock().map_err(record_error)?; + let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone(); candidate.restore_guard.rotate(); candidate.execution_metadata_available = false; @@ -5246,8 +5273,12 @@ impl Runtime { logical_workdir_attachments_from_statuses(&workdir_attachments, claims); candidate.workdir_attachments = workdir_attachments; state.workers.insert(worker_ref.worker_id, candidate); - state.persist_worker(&worker_ref.worker_id)?; - state.publish_worker_upsert(worker_ref.worker_id)?; + state + .persist_worker(&worker_ref.worker_id) + .map_err(record_error)?; + state + .publish_worker_upsert(worker_ref.worker_id) + .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?; Ok(()) } @@ -8379,6 +8410,8 @@ mod tests { result } + include!("runtime/create_reconciliation_tests.rs"); + fn check_rollback(fault: Fault) { let primary = primary_failure(); for spawn_result in [ diff --git a/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs b/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs new file mode 100644 index 000000000..d58f6eecc --- /dev/null +++ b/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs @@ -0,0 +1,113 @@ +mod create_reconciliation_tests { + //! The real create retention and FsRuntimeStore write/reopen path; only spawn + //! disposition and the existing deterministic atomic-write hook are substituted. + use super::*; + use crate::fs_store::AtomicWriteFault; + + fn check_reconciliation(fault: AtomicWriteFault) { + for has_state in [false, true] { + let root = tempfile::tempdir().unwrap(); + let options = FsRuntimeStoreOptions::new(root.path().join("runtime")) + .with_runtime_id("create-reconciliation"); + 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(); + let request = task_request("reconciliation failure"); + let worker_ref = WorkerRef::new(request.worker_id); + let primary = primary_failure(); + *backend.spawn_result.lock().unwrap() = + Some(WorkerExecutionSpawnResult::ReconciliationRequired { + result: primary.clone(), + worker_state: has_state.then(|| protocol::WorkerStatus::Stopped.into()), + workdir_attachments: Vec::new(), + }); + let initial = Arc::new(Mutex::new(None)); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let store = runtime_store(&runtime); + let initial = initial.clone(); + let worker_ref = worker_ref.clone(); + move || { + *initial.lock().unwrap() = Some(store.read_worker_record(&worker_ref).unwrap()); + store.fail_next_worker_write(fault); + } + })); + let RuntimeError::WorkerExecutionRejected { result, .. } = + runtime.create_worker(request).unwrap_err() + else { + panic!("retention failure must not replace primary result") + }; + assert_eq!(result.startup_failure, primary.startup_failure); + assert_eq!(result.startup_resources, primary.startup_resources); + assert_eq!(result.outcome, primary.outcome); + assert!( + result + .secondary_failures + .contains(&primary.secondary_failures[0]) + ); + let outcome = match fault { + AtomicWriteFault::BeforeRename => StartupPersistenceOutcome::NotSaved, + AtomicWriteFault::AfterRename => StartupPersistenceOutcome::OutcomeUnknown, + AtomicWriteFault::None => StartupPersistenceOutcome::Persisted, + }; + assert!(result.secondary_failures.iter().any(|failure| matches!(failure, StartupSecondaryFailure::Persistence { phase: StartupFailurePhase::RuntimeReconciliationRecordSave, outcome: observed } if *observed == outcome)), "retention persistence must be explicit: {result:?}"); + let disk = runtime_store(&runtime) + .read_worker_record(&worker_ref) + .unwrap(); + let initial = initial.lock().unwrap().take().unwrap(); + if fault == AtomicWriteFault::BeforeRename { + assert_eq!(disk.execution_state, initial.execution_state); + assert_eq!( + disk.restore_guard.observation_token, + initial.restore_guard.observation_token + ); + } else { + assert_ne!( + disk.restore_guard.observation_token, + initial.restore_guard.observation_token + ); + if has_state { + assert!( + matches!(disk.execution_state, crate::fs_store::PersistedWorkerExecutionState::Available(ref execution) if execution.binding.is_some()) + ); + } else { + assert_eq!( + disk.execution_state, + crate::fs_store::PersistedWorkerExecutionState::Unavailable + ); + } + } + // A pre-rename failure left the initial record; an uncertain response + // here followed a real rename and is reconciled from disk, not guessed. + let on_disk = std::fs::read( + root.path() + .join("runtime/workers") + .join(worker_ref.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(); + drop(runtime); + drop(backend); + let reopened = Runtime::with_fs_store(options).unwrap(); + let reread = runtime_store(&reopened) + .read_worker_record(&worker_ref) + .unwrap(); + assert_eq!(reread.execution_state, disk.execution_state); + assert!(reopened.worker_detail(&worker_ref).is_ok()); + assert!(!on_disk.is_empty()); + } + } + #[test] + fn create_reconciliation_write_failure_retains_primary_and_unsaved_evidence() { + check_reconciliation(AtomicWriteFault::BeforeRename); + } + #[test] + fn create_reconciliation_unknown_write_retains_primary_and_reopens_actual_record() { + check_reconciliation(AtomicWriteFault::AfterRename); + } + #[test] + fn create_reconciliation_success_retains_primary_and_durable_evidence() { + check_reconciliation(AtomicWriteFault::None); + } +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 4a6f58347..09b93f923 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -2251,34 +2251,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .map_err(str::to_string)?; } validate_worker_memory_settings(worker.manifest(), &request.request)?; - let expected = worker - .store() - .read_by_name(&worker_name)? - .and_then(|metadata| { - session_store::worker_metadata::StartupFailureExpectation::from_metadata(&metadata) - }); - worker - .finalize_subjektiv_session_attribution( - request.request.subjektiv_attached, - SubjektivSessionAttributionLifecycle::NewSession, - ) - .map_err(|error| { - let error = if let Some(expected) = &expected { - worker::startup_failure::persist_startup_failure::< - _, - protocol::startup_failure::RuntimeStartupCause, - >( - worker.store(), - expected, - error, - protocol::startup_failure::StartupStage::Binding, - &(), - ) - } else { - error - }; - RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error) - })?; + startup_failure::finalize_startup_binding(worker, request.request.subjektiv_attached)?; let workdir_capabilities = request .workdir_attachments .keys() @@ -7187,6 +7160,7 @@ mod tests { use worker::PreparedWorker; include!("subject_delete_backend_tests.rs"); + include!("worker_backend/startup_binding_tests.rs"); #[cfg(feature = "fs-store")] include!("subject_delete_create_tests.rs"); diff --git a/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs new file mode 100644 index 000000000..1a532fc26 --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs @@ -0,0 +1,253 @@ +mod startup_binding_tests { + //! Exercise the factory's production binding step with a constructed Worker, + //! real aggregate/Session stores and reopen. Only the Workspace response is + //! synthetic; this suite does not test domain authorization or network transport. + use super::*; + use protocol::startup_failure::{RuntimeStartupCause, StartupPersistenceOutcome}; + use session_store::worker_metadata::StartupFailureRead; + use worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; + + #[derive(Debug, Clone, Copy)] + enum Response { + Rejected, + Lost, + BeforeSession, + AfterSession, + BeforeOwner, + AfterOwner, + AfterAttribution, + } + struct AttributionClient { + response: Response, + store: WorkerAggregateStore, + name: String, + } + impl std::fmt::Debug for AttributionClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("AttributionClient") + } + } + impl WorkspaceClient for AttributionClient { + fn workspace_id(&self) -> Option<&str> { + Some("startup-binding-workspace") + } + fn kind(&self) -> &str { + "startup-binding-fixture" + } + fn is_available(&self) -> bool { + true + } + fn execute(&self, _: WorkspaceRequest) -> Result { + match self.response { + Response::AfterSession => self + .store + .update_by_name(&self.name, |metadata| { + metadata.active = Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )); + metadata.subjektiv_session_attribution = None; + }) + .unwrap(), + Response::AfterOwner => self + .store + .update_by_name(&self.name, |metadata| { + metadata.workspace_id = Some("replacement-owner".into()); + }) + .unwrap(), + Response::AfterAttribution => self + .store + .update_by_name(&self.name, |metadata| { + metadata.subjektiv_session_attribution = + Some(session_store::SubjektivSessionAttributionState::Confirmed { + session_id: metadata.active.as_ref().unwrap().session_id, + subject_id: "replacement-subject".into(), + }); + }) + .unwrap(), + _ => self.store.read_by_name(&self.name).unwrap().unwrap(), + }; + if matches!(self.response, Response::Lost) { + return Err(WorkspaceClientError::Unavailable( + "response deliberately lost".into(), + )); + } + Ok(WorkspaceResponse { + status: 403, + body: "synthetic rejection with private token /internal/path".into(), + }) + } + } + async fn check_binding(response: Response) { + require_restore_test_environment(); + let root = tempfile::tempdir().unwrap(); + let name = "startup-binding"; + let store = WorkerAggregateStore::new(root.path(), name).unwrap(); + let session_store = WorkerSessionStore::new(root.path().join("session")).unwrap(); + let client = Arc::new(AttributionClient { + response, + store: store.clone(), + name: name.into(), + }); + let manifest = WorkerManifest::from_toml( + r#" + [worker] + name = "startup-binding" + [model] + scheme = "anthropic" + model_id = "unused" + auth = { kind = "none" } + [engine] + max_tokens = 100 + [scope] + allow = [] + "#, + ) + .unwrap(); + let llm: Box = Box::new(MockClient::new(Vec::new())); + let mut worker = Worker::new( + manifest, + Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(llm), + CombinedStore::new(session_store, store.clone()), + WorkerWorkspaceContext::with_client( + Some(WorkspaceId::new("startup-binding-workspace").unwrap()), + client, + ), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.enable_worker_metadata_write_through().unwrap(); + let origin = worker.session_id(); + match response { + Response::BeforeSession => store + .update_by_name(name, |metadata| { + metadata.active = Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )); + }) + .unwrap(), + Response::BeforeOwner => store + .update_by_name(name, |metadata| { + metadata.workspace_id = Some("replacement-owner".into()); + }) + .unwrap(), + _ => store.read_by_name(name).unwrap().unwrap(), + }; + let failure = startup_failure::finalize_startup_binding(&worker, true) + .err() + .expect("failure must remain a failure"); + let result = failure.execution_result(WorkerExecutionOperation::Spawn); + let primary = result.startup_failure::().unwrap(); + assert!(matches!(primary, error_contract::DecodeFailure::Known(_))); + drop(worker); + drop(store); + let reopened = WorkerAggregateStore::new(root.path(), name).unwrap(); + let metadata = reopened.read_by_name(name).unwrap().unwrap(); + match response { + Response::Rejected | Response::Lost => { + assert_eq!( + result.startup_failure_persistence, + Some(StartupPersistenceOutcome::Persisted) + ); + let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = + metadata.decode_startup_failure::() + else { + panic!("same-Session failure must actually save a typed cause") + }; + assert_eq!( + cause.facts().cause().reason, + protocol::startup_failure::RuntimeStartupReason::AttributionFailed + ); + assert_eq!(metadata.active.unwrap().session_id, origin); + let saved = serde_json::to_string(&metadata.startup_failure).unwrap(); + assert!(!saved.contains("private token")); + assert!(!saved.contains("/internal/path")); + assert_eq!( + metadata + .subjektiv_session_attribution + .as_ref() + .unwrap() + .session_id(), + origin + ); + assert_eq!( + matches!( + metadata.subjektiv_session_attribution, + Some( + session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. } + ) + ), + matches!(response, Response::Lost) + ); + } + _ => { + assert!( + matches!( + metadata.decode_startup_failure::(), + StartupFailureRead::Missing + ), + "replacement Session/owner must not acquire stale cause" + ); + assert_ne!( + result.startup_failure_persistence, + Some(StartupPersistenceOutcome::Persisted) + ); + if matches!(response, Response::BeforeSession | Response::AfterSession) { + assert_ne!(metadata.active.unwrap().session_id, origin); + } + if matches!( + response, + Response::BeforeOwner | Response::BeforeSession | Response::AfterSession + ) { + assert!(metadata.subjektiv_session_attribution.is_none()); + } + if matches!(response, Response::AfterOwner) { + assert!(matches!( + metadata.subjektiv_session_attribution, + Some( + session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. } + ) + )); + } + if matches!(response, Response::BeforeOwner | Response::AfterOwner) { + assert_eq!(metadata.workspace_id.as_deref(), Some("replacement-owner")); + } + if matches!(response, Response::AfterAttribution) { + assert!(matches!( + metadata.subjektiv_session_attribution, + Some(session_store::SubjektivSessionAttributionState::Confirmed { .. }) + )); + } + } + } + } + #[tokio::test] + async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() { + check_binding(Response::Rejected).await; + } + #[tokio::test] + async fn factory_unknown_response_saves_cause_without_claiming_attribution_success() { + check_binding(Response::Lost).await; + } + #[tokio::test] + async fn factory_session_replaced_before_capture_rejects_stale_cause() { + check_binding(Response::BeforeSession).await; + } + #[tokio::test] + async fn factory_session_replaced_after_capture_rejects_stale_cause() { + check_binding(Response::AfterSession).await; + } + #[tokio::test] + async fn factory_owner_replaced_before_capture_rejects_stale_cause() { + check_binding(Response::BeforeOwner).await; + } + #[tokio::test] + async fn factory_owner_replaced_after_capture_rejects_stale_cause() { + check_binding(Response::AfterOwner).await; + } + #[tokio::test] + async fn factory_attribution_replaced_after_capture_preserves_new_attribution() { + check_binding(Response::AfterAttribution).await; + } +} diff --git a/crates/worker-runtime/src/worker_backend/startup_failure.rs b/crates/worker-runtime/src/worker_backend/startup_failure.rs index f99ec985c..8f70f3bde 100644 --- a/crates/worker-runtime/src/worker_backend/startup_failure.rs +++ b/crates/worker-runtime/src/worker_backend/startup_failure.rs @@ -102,6 +102,21 @@ impl PendingRuntimeRestore { } } +pub(super) fn finalize_startup_binding( + worker: &Worker, St>, + attached: bool, +) -> Result<(), RuntimeRestoreFailure> +where + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, +{ + worker + .finalize_subjektiv_session_attribution_retaining_startup_failure( + attached, + SubjektivSessionAttributionLifecycle::NewSession, + ) + .map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error)) +} + #[cfg(test)] mod tests { //! Connect the typed factory seam to the real Tokio adapter and protocol codec. diff --git a/crates/worker/src/subjektiv_metadata.rs b/crates/worker/src/subjektiv_metadata.rs index e24b4c926..198bba479 100644 --- a/crates/worker/src/subjektiv_metadata.rs +++ b/crates/worker/src/subjektiv_metadata.rs @@ -3,8 +3,7 @@ use crate::metadata_lifecycle::SessionMetadataLifecycle; use session_store::{ - SessionId, SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, - WorkerMetadataStore, + SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, }; pub(crate) struct SessionAttributionMetadata; @@ -25,18 +24,13 @@ impl SessionMetadataLifecycle for SessionAttributionMetadata { pub(crate) fn compare_and_swap_attribution( store: &St, - worker_name: &str, - session_id: SessionId, - expected: Option<&SubjektivSessionAttributionState>, + expected: &mut session_store::worker_metadata::StartupFailureExpectation, replacement: SubjektivSessionAttributionState, ) -> Result<(), crate::WorkerError> { let mut matched = false; - store.update_by_name(worker_name, |metadata| { - if metadata.active.as_ref().map(|active| active.session_id) == Some(session_id) - && metadata.subjektiv_session_attribution.as_ref() == expected - && replacement.session_id() == session_id - { - metadata.subjektiv_session_attribution = Some(replacement); + store.update_by_name(&expected.worker_name, |metadata| { + if expected.matches(metadata) && replacement.session_id() == expected.session_id { + metadata.subjektiv_session_attribution = Some(replacement.clone()); matched = true; } })?; @@ -47,6 +41,8 @@ pub(crate) fn compare_and_swap_attribution( .into(), }); } + // Advance only after the exact owner/Session CAS returned confirmed success. + expected.attribution = Some(replacement); Ok(()) } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 4460290af..707752550 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -9698,6 +9698,48 @@ where &self, subjektiv_attached: bool, lifecycle: SubjektivSessionAttributionLifecycle, + ) -> Result<(), WorkerError> { + self.finalize_subjektiv_session_attribution_observed( + subjektiv_attached, + lifecycle, + &mut None, + ) + } + + /// Retain a failed startup against its constructed Session and Workspace. Only + /// this operation's successful CAS transitions advance the attribution expectation; + /// a reread never adopts a replacement owner or another operation's attribution. + pub fn finalize_subjektiv_session_attribution_retaining_startup_failure( + &self, + subjektiv_attached: bool, + lifecycle: SubjektivSessionAttributionLifecycle, + ) -> Result<(), WorkerError> { + let mut expected = None; + self.finalize_subjektiv_session_attribution_observed( + subjektiv_attached, + lifecycle, + &mut expected, + ) + .map_err(|error| match expected { + Some(expected) => crate::startup_failure::persist_startup_failure::< + _, + protocol::startup_failure::RuntimeStartupCause, + >( + &self.store, + &expected, + error, + protocol::startup_failure::StartupStage::Binding, + &(), + ), + None => error, + }) + } + + fn finalize_subjektiv_session_attribution_observed( + &self, + subjektiv_attached: bool, + lifecycle: SubjektivSessionAttributionLifecycle, + observation: &mut Option, ) -> Result<(), WorkerError> { let _admission = self.segment_state.mutation_authority.enter()?; let worker_name = self.manifest.worker.name.as_str(); @@ -9726,6 +9768,31 @@ where }); } + if metadata.worker_name != worker_name + || metadata.workspace_id.as_deref() != self.workspace_id().map(|id| id.as_str()) + { + return Err(WorkerError::SubjektivSessionAttribution { + state: "owner_mismatch", + message: "Worker metadata does not match constructed Workspace ownership".into(), + }); + } + *observation = + session_store::worker_metadata::StartupFailureExpectation::from_metadata(&metadata); + let mut persist_attribution = + |previous: Option<&SubjektivSessionAttributionState>, + next: SubjektivSessionAttributionState| { + let expected = observation + .as_mut() + .expect("validated active Session has an expectation"); + if expected.attribution.as_ref() != previous { + return Err(WorkerError::SubjektivSessionAttribution { + state: "attribution_conflict", + message: "startup attribution expectation changed".into(), + }); + } + crate::subjektiv_metadata::compare_and_swap_attribution(&self.store, expected, next) + }; + if self .backend_job() .is_some_and(crate::BackendJobExecutionBinding::has_subjektiv_job_grant) @@ -9783,7 +9850,7 @@ where create_if_missing, initial_diagnostic.to_string(), ); - self.persist_subjektiv_session_attribution( + persist_attribution( metadata.subjektiv_session_attribution.as_ref(), pending_attribution.clone(), )?; @@ -9819,7 +9886,7 @@ where message.clone(), ) }; - self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; + persist_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing && rejected { "failed" @@ -9844,7 +9911,7 @@ where let message = bounded_attribution_diagnostic(format!( "Session attribution transport outcome is unknown: {error}" )); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9874,7 +9941,7 @@ where diagnostic: message.clone(), } }; - self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; + persist_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing { "failed" @@ -9891,7 +9958,7 @@ where let message = bounded_attribution_diagnostic(format!( "Session attribution response outcome is unknown: {error}" )); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9914,7 +9981,7 @@ where let message = "Session attribution response did not confirm the requested immutable identity" .to_string(); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9927,7 +9994,7 @@ where message, }); } - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), SubjektivSessionAttributionState::Confirmed { session_id, @@ -9972,20 +10039,6 @@ where Ok(()) } - fn persist_subjektiv_session_attribution( - &self, - expected: Option<&SubjektivSessionAttributionState>, - state: SubjektivSessionAttributionState, - ) -> Result<(), WorkerError> { - crate::subjektiv_metadata::compare_and_swap_attribution( - &self.store, - &self.manifest.worker.name, - self.session_id(), - expected, - state, - ) - } - /// Build an in-process Internal Worker without machine-wide allocation or durable Worker metadata. pub(crate) async fn from_internal_manifest_with_context( manifest: WorkerManifest, diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 7ec8ec247..6fb931d91 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -12,7 +12,7 @@ export type RuntimeStartupReason = "startup_failed" | "segment_empty" | "system_ export type RuntimeStartupCause = { stage: StartupStage, reason: RuntimeStartupReason, }; -export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "metadata_save"; +export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "runtime_reconciliation_record_save" | "runtime_reconciliation_publish" | "metadata_save"; export type StartupCleanupFailure = "ownership" | "workdir" | "artifacts" | "allocation" | "feature" | "unsupported";