Preserve startup causes through Runtime rollback and Session metadata CAS

This commit is contained in:
2026-11-04 11:31:12 +09:00
parent 9d8175f52e
commit 326616d07e
21 changed files with 2439 additions and 250 deletions
+162
View File
@@ -40,6 +40,22 @@ impl StartupFailureAttachment {
decode_failure(self.0.as_bytes(), Self::budget())
}
/// Redisclosure requires a fresh domain/audience projection and current authorization.
/// Unsupported or corrupt stored bytes are never forwarded as public details.
pub fn project_for<A: Audience, D: WireDetails + error_contract::ProjectFor<A>>(
&self,
authorization: &D::Authorization,
) -> Result<ProjectedFailure<A, D::Details>, ContractError> {
match self.decode::<D>() {
DecodeFailure::Known(received) => {
error_contract::project_failure(received.facts(), authorization)
}
DecodeFailure::Unsupported(_) | DecodeFailure::Corrupt(_) => {
Err(ContractError::ProjectionUnavailable)
}
}
}
fn budget() -> Budget {
Budget::new(
error_contract::MAX_FAILURE_BYTES,
@@ -48,6 +64,31 @@ impl StartupFailureAttachment {
}
}
/// Storage encoding cannot be constructed by relabeling an internal-transfer attachment.
/// A loaded value remains untrusted evidence; publication still requires `project_for`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct StoredStartupFailureAttachment(StartupFailureAttachment);
impl StoredStartupFailureAttachment {
pub fn from_projected<D: WireDetails>(
failure: &ProjectedFailure<error_contract::Storage, D>,
) -> Result<Self, ContractError> {
StartupFailureAttachment::from_projected(failure).map(Self)
}
pub fn decode<D: WireDetails>(&self) -> DecodeFailure<D> {
self.0.decode()
}
pub fn project_for<A: Audience, D: WireDetails + error_contract::ProjectFor<A>>(
&self,
authorization: &D::Authorization,
) -> Result<ProjectedFailure<A, D::Details>, ContractError> {
self.0.project_for::<A, D>(authorization)
}
}
/// Owner-selected startup stage; independent of resource cleanup disposition.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
@@ -177,6 +218,9 @@ pub enum StartupCleanupFailure {
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub enum StartupSecondaryFailure {
ProjectionUnavailable {
phase: StartupFailurePhase,
},
Cleanup {
failure: StartupCleanupFailure,
},
@@ -229,3 +273,121 @@ pub struct StartupResourceFacts {
pub cleanup: StartupCleanup,
pub durable_effects_possible: bool,
}
/// Path-free IO classification. Nonportable OS details remain at the resource owner.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub enum StartupIoKind {
NotFound,
PermissionDenied,
AlreadyExists,
InvalidInput,
InvalidData,
Interrupted,
TimedOut,
WouldBlock,
UnexpectedEof,
Other,
}
impl From<std::io::ErrorKind> for StartupIoKind {
fn from(kind: std::io::ErrorKind) -> Self {
use std::io::ErrorKind;
match kind {
ErrorKind::NotFound => Self::NotFound,
ErrorKind::PermissionDenied => Self::PermissionDenied,
ErrorKind::AlreadyExists => Self::AlreadyExists,
ErrorKind::InvalidInput => Self::InvalidInput,
ErrorKind::InvalidData => Self::InvalidData,
ErrorKind::Interrupted => Self::Interrupted,
ErrorKind::TimedOut => Self::TimedOut,
ErrorKind::WouldBlock => Self::WouldBlock,
ErrorKind::UnexpectedEof => Self::UnexpectedEof,
_ => Self::Other,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use error_contract::{
Browser, FailureFacts, InternalTransfer, OperationContext, Storage, project_failure,
};
#[test]
fn startup_projection_preserves_allowlisted_facts_without_secret_context() {
let mut facts = FailureFacts::new(RuntimeStartupCause {
stage: StartupStage::Binding,
reason: RuntimeStartupReason::AttributionFailed,
})
.unwrap();
let mut context = OperationContext::new("worker_startup", "runtime").unwrap();
context.operation_ref = Some("TEST_ONLY_CREDENTIAL".into());
facts.with_context(context).unwrap();
facts
.with_context(OperationContext::new("/private/internal/path", "raw_http_body").unwrap())
.unwrap();
let transfer = project_failure::<InternalTransfer, _>(&facts, &()).unwrap();
let attachment = StartupFailureAttachment::from_projected(&transfer).unwrap();
let storage_projection = attachment
.project_for::<Storage, RuntimeStartupCause>(&())
.unwrap();
let stored = StoredStartupFailureAttachment::from_projected(&storage_projection).unwrap();
let browser = stored
.project_for::<Browser, RuntimeStartupCause>(&())
.unwrap();
assert_eq!(browser.details(), facts.cause());
assert_eq!(browser.context().operations.len(), 1);
assert_eq!(browser.context().operations[0].operation_ref, None);
let public =
encode_projected_failure(&browser, StartupFailureAttachment::budget()).unwrap();
let wire = std::str::from_utf8(public.as_bytes()).unwrap();
for secret in [
"TEST_ONLY_CREDENTIAL",
"/private/internal/path",
"raw_http_body",
] {
assert!(!wire.contains(secret));
assert!(!serde_json::to_string(&stored).unwrap().contains(secret));
}
assert!(matches!(
stored.decode::<RuntimeStartupCause>(),
DecodeFailure::Known(_)
));
}
#[test]
fn unsupported_and_corrupt_startup_attachments_cannot_be_redisclosed() {
let facts = FailureFacts::new(RuntimeStartupCause {
stage: StartupStage::Controller,
reason: RuntimeStartupReason::StartupFailed,
})
.unwrap();
let projected = project_failure::<InternalTransfer, _>(&facts, &()).unwrap();
let attachment = StartupFailureAttachment::from_projected(&projected).unwrap();
let future =
StartupFailureAttachment(attachment.0.replace("\"version\":1", "\"version\":999"));
assert!(matches!(
future.decode::<RuntimeStartupCause>(),
DecodeFailure::Unsupported(_)
));
assert!(
future
.project_for::<Browser, RuntimeStartupCause>(&())
.is_err()
);
let corrupt = StartupFailureAttachment("{invalid TEST_ONLY_SECRET".into());
assert!(matches!(
corrupt.decode::<RuntimeStartupCause>(),
DecodeFailure::Corrupt(_)
));
assert!(
corrupt
.project_for::<Browser, RuntimeStartupCause>(&())
.is_err()
);
assert!(!format!("{corrupt:?}").contains("TEST_ONLY_SECRET"));
}
}
+13
View File
@@ -64,6 +64,19 @@ pub fn generated_protocol_types() -> String {
// Large integer fields are JSON numbers on the wire and are emitted as TypeScript `number`.\n\n",
);
push_decl::<crate::startup_failure::StartupFailureAttachment>(&cfg, &mut output);
push_decl::<crate::startup_failure::StoredStartupFailureAttachment>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupStage>(&cfg, &mut output);
push_decl::<crate::startup_failure::RuntimeStartupReason>(&cfg, &mut output);
push_decl::<crate::startup_failure::RuntimeStartupCause>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupFailurePhase>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupCleanupFailure>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupPersistenceOutcome>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupSecondaryFailure>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupResources>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupCleanup>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupResourceFacts>(&cfg, &mut output);
push_decl::<crate::startup_failure::StartupIoKind>(&cfg, &mut output);
push_decl::<AlertLevel>(&cfg, &mut output);
push_decl::<AlertPosition>(&cfg, &mut output);
push_decl::<AlertSource>(&cfg, &mut output);
+4 -2
View File
@@ -11,8 +11,10 @@ use serde::{Deserialize, Serialize};
use workdir::workspace::{MaterializerKind, RuntimeWorkerRef, RuntimeWorkingDirectorySummary};
pub use protocol::startup_failure::{
RuntimeStartupCause, StartupCleanupFailure, StartupFailureAttachment, StartupFailurePhase,
StartupPersistenceOutcome, StartupSecondaryFailure, StartupStage,
RuntimeStartupCause, RuntimeStartupReason, StartupCleanup, StartupCleanupFailure,
StartupFailureAttachment, StartupFailurePhase, StartupIoKind, StartupPersistenceOutcome,
StartupResourceFacts, StartupResources, StartupSecondaryFailure, StartupStage,
StoredStartupFailureAttachment,
};
pub const RUNTIME_API_VERSION: &str = "v1";
+5 -3
View File
@@ -114,9 +114,11 @@ pub use uploaded_file::{
UploadedFileUploadContext,
};
pub use worker_metadata::{
CombinedStore, FsWorkerStore, SubjektivSessionAttributionState, WorkerActiveSegmentRef,
WorkerAggregateStore, WorkerMetadata, WorkerMetadataStore, WorkerPeer, WorkerReclaimedChild,
WorkerSpawnedChild, WorkerSpawnedScopeRule, WorkerStoreError, validate_worker_name,
CombinedStore, FsWorkerStore, SessionStartupFailure, StartupFailureBindingMismatch,
StartupFailureExpectation, StartupFailureRead, StartupFailureSaveResult,
SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerAggregateStore, WorkerMetadata,
WorkerMetadataStore, WorkerPeer, WorkerReclaimedChild, WorkerSpawnedChild,
WorkerSpawnedScopeRule, WorkerStoreError, validate_worker_name,
};
pub use worker_session_store::WorkerSessionStore;
@@ -0,0 +1,727 @@
//! Component coverage of startup attachments through real FsWorkerStore files,
//! reopen and process-local CAS. Only the directory-sync IO boundary is replaced
//! for the post-rename failure case; these tests do not claim OS-crash durability.
//! Selected by the normal session-store lib test target.
use super::*;
use error_contract::{FailureFacts, OperationContext, Storage, project_failure};
use protocol::startup_failure::{RuntimeStartupCause, RuntimeStartupReason, StartupStage};
fn attachment(stage: StartupStage) -> StoredStartupFailureAttachment {
let mut facts = FailureFacts::new(RuntimeStartupCause {
stage,
reason: RuntimeStartupReason::StartupFailed,
})
.unwrap();
facts
.with_context(OperationContext {
operation: "worker_startup".into(),
stage: "runtime".into(),
operation_ref: None,
})
.unwrap();
let projected = project_failure::<Storage, _>(&facts, &()).unwrap();
StoredStartupFailureAttachment::from_projected(&projected).unwrap()
}
fn metadata() -> WorkerMetadata {
let session_id = crate::new_session_id();
let mut metadata = WorkerMetadata::new(
"agent",
Some(WorkerActiveSegmentRef::pending_segment(session_id)),
)
.with_workspace_id("workspace-one");
metadata.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed {
session_id,
diagnostic: "legacy diagnostic is not a typed cause".into(),
});
metadata.resolved_manifest_snapshot = Some(serde_json::json!({"retained": true}));
metadata.peers.push(WorkerPeer {
worker_name: "peer".into(),
});
metadata
}
fn bound_failure(
metadata: &WorkerMetadata,
attachment: &StoredStartupFailureAttachment,
) -> SessionStartupFailure {
SessionStartupFailure::from_expected(
&StartupFailureExpectation::from_metadata(metadata).unwrap(),
attachment,
)
}
fn assert_stored_cause(metadata: &WorkerMetadata, stage: StartupStage) {
match metadata.decode_startup_failure::<RuntimeStartupCause>() {
StartupFailureRead::Present(DecodeFailure::Known(received)) => {
assert_eq!(received.facts().cause().stage, stage);
assert_eq!(
received.facts().cause().reason,
RuntimeStartupReason::StartupFailed
);
assert_eq!(received.facts().context().operations.len(), 1);
assert_eq!(
received.facts().context().operations[0].operation,
"worker_startup"
);
}
_ => panic!("expected a known typed startup cause"),
}
}
#[test]
fn startup_failure_cas_reopens_typed_cause_and_preserves_unrelated_metadata() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let cause = attachment(StartupStage::Binding);
let received = cause.clone();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Persisted
));
assert_eq!(cause, received);
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
assert_stored_cause(&restored, StartupStage::Binding);
let mut wanted = original;
wanted.startup_failure = Some(SessionStartupFailure::from_expected(&expected, &received));
assert_eq!(restored, wanted);
}
#[test]
fn startup_failure_wrapper_fingerprints_attribution_without_copying_raw_diagnostics() {
const CREDENTIAL: &str = "credential=TEST_ONLY_METADATA_SENTINEL";
const PRIVATE_PATH: &str = "/private/test-only-attribution/session.json";
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let mut original = metadata();
original.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed {
session_id: original.active.as_ref().unwrap().session_id,
diagnostic: format!("{CREDENTIAL} {PRIVATE_PATH}"),
});
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Persisted
));
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
assert_eq!(
restored.subjektiv_session_attribution,
original.subjektiv_session_attribution
);
// The original field still retains the diagnostic: absence from the wrapper
// must not be explained by removing or redacting the source metadata instead.
let retained_attribution =
serde_json::to_string(&restored.subjektiv_session_attribution).unwrap();
assert!(retained_attribution.contains(CREDENTIAL));
assert!(retained_attribution.contains(PRIVATE_PATH));
let wrapper = restored.startup_failure.as_ref().unwrap();
let wrapper_json = serde_json::to_string(wrapper).unwrap();
assert!(!wrapper_json.contains(CREDENTIAL));
assert!(!wrapper_json.contains(PRIVATE_PATH));
let wrapper_value = serde_json::to_value(wrapper).unwrap();
assert!(wrapper_value.get("attribution").is_none());
let expected_fingerprint: [u8; 32] = Sha256::digest(retained_attribution.as_bytes()).into();
assert_eq!(bound(wrapper).attribution_fingerprint, expected_fingerprint);
assert_stored_cause(&restored, StartupStage::Binding);
}
#[test]
fn startup_failure_cas_rejects_changed_session_owner_attribution_and_cause_without_rewrite() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let cause = attachment(StartupStage::Binding);
let path = temp.path().join("agent/metadata.json");
// Each row changes one compared fact. Worker-name mismatch is an on-disk
// identity mismatch, not a write to a different key.
let changed = [
{
let mut value = original.clone();
value.active = Some(WorkerActiveSegmentRef::pending_segment(
crate::new_session_id(),
));
value
},
{
let mut value = original.clone();
value.workspace_id = Some("new-owner".into());
value
},
{
let mut value = original.clone();
value.worker_name = "new-worker".into();
value
},
{
let mut value = original.clone();
value.subjektiv_session_attribution =
Some(SubjektivSessionAttributionState::Confirmed {
session_id: expected.session_id,
subject_id: "new-subject".into(),
});
value
},
{
let mut value = original.clone();
value.subjektiv_session_attribution = None;
value
},
{
let mut value = original.clone();
value.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed {
session_id: expected.session_id,
diagnostic: "new attribution observation".into(),
});
value
},
{
let mut value = original.clone();
value.startup_failure = Some(bound_failure(&value, &attachment(StartupStage::Feature)));
value
},
{
let mut value = original.clone();
value.active = None;
value
},
];
for value in changed {
fs::create_dir_all(path.parent().unwrap()).unwrap();
// Retain noncanonical whitespace to detect even a no-op rewrite.
let bytes = format!(" {}\n", serde_json::to_string(&value).unwrap());
fs::write(&path, &bytes).unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Conflict
));
assert_eq!(fs::read(&path).unwrap(), bytes.as_bytes());
assert_eq!(store.read_by_name("agent").unwrap().unwrap(), value);
}
}
#[test]
fn startup_failure_replacement_requires_the_observed_existing_attachment() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let mut original = metadata();
original.startup_failure = Some(bound_failure(&original, &attachment(StartupStage::Feature)));
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let mut stale = expected.clone();
stale.startup_failure = None;
let cause = attachment(StartupStage::Binding);
assert!(matches!(
store.compare_and_swap_startup_failure(&stale, &cause),
StartupFailureSaveResult::Conflict
));
assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original);
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Persisted
));
assert_stored_cause(
&store.read_by_name("agent").unwrap().unwrap(),
StartupStage::Binding,
);
}
#[test]
fn startup_failure_cas_missing_metadata_does_not_create_a_worker() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let expected = StartupFailureExpectation::from_metadata(&metadata()).unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Conflict
));
assert!(store.read_by_name("agent").unwrap().is_none());
assert!(!temp.path().join("agent").exists());
}
#[test]
fn startup_failure_cas_allows_segment_movement_within_expected_session() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let active =
WorkerActiveSegmentRef::active_segment(expected.session_id, crate::new_segment_id());
store
.set_active("agent", Some(active.clone()), None)
.unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Persisted
));
let restored = store.read_by_name("agent").unwrap().unwrap();
assert_eq!(restored.active, Some(active));
assert_eq!(
restored.subjektiv_session_attribution,
original.subjektiv_session_attribution
);
// A later Segment switch is not a Session/ownership change either.
store
.set_active(
"agent",
Some(WorkerActiveSegmentRef::active_segment(
expected.session_id,
crate::new_segment_id(),
)),
None,
)
.unwrap();
assert_stored_cause(
&store.read_by_name("agent").unwrap().unwrap(),
StartupStage::Binding,
);
}
#[test]
fn startup_failure_saved_binding_rejects_session_owner_and_attribution_changes_after_reopen() {
for case in 0..7 {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let cause = attachment(StartupStage::Binding);
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Persisted
));
let saved = store
.read_by_name("agent")
.unwrap()
.unwrap()
.startup_failure;
store
.update_by_name("agent", |value| match case {
0 => {
value.active = Some(WorkerActiveSegmentRef::pending_segment(
crate::new_session_id(),
))
}
1 => value.workspace_id = Some("new-workspace-owner".into()),
2 => {
value.subjektiv_session_attribution =
Some(SubjektivSessionAttributionState::Confirmed {
session_id: expected.session_id,
subject_id: "new-subject-owner".into(),
})
}
3 => {
value.subjektiv_session_attribution =
Some(SubjektivSessionAttributionState::Failed {
session_id: expected.session_id,
diagnostic: "changed attribution observation".into(),
})
}
4 => value.active = None,
5 => value.workspace_id = None,
6 => value.subjektiv_session_attribution = None,
_ => unreachable!(),
})
.unwrap();
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
// State changes retain the old cause as evidence; they do not rebind it
// to the new Session/owner and do not turn a mismatch into absence.
assert_eq!(restored.startup_failure, saved);
let path = temp.path().join("agent/metadata.json");
let bytes_before_read = fs::read(&path).unwrap();
match restored.decode_startup_failure::<RuntimeStartupCause>() {
StartupFailureRead::Mismatch(mismatch) => assert_eq!(
mismatch,
StartupFailureBindingMismatch {
worker_name: false,
session_id: matches!(case, 0 | 4),
workspace_id: matches!(case, 1 | 5),
attribution: matches!(case, 2 | 3 | 6),
}
),
_ => panic!("saved cause must not decode for a changed binding (case {case})"),
}
assert_eq!(fs::read(&path).unwrap(), bytes_before_read);
assert!(matches!(
reopened.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Conflict
));
}
}
#[test]
fn startup_failure_copy_to_another_worker_keeps_original_binding() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Persisted
));
let mut copied = store.read_by_name("agent").unwrap().unwrap();
copied.worker_name = "replacement".into();
store.write(&copied).unwrap();
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("replacement").unwrap().unwrap();
assert!(matches!(
restored.decode_startup_failure::<RuntimeStartupCause>(),
StartupFailureRead::Mismatch(StartupFailureBindingMismatch {
worker_name: true,
session_id: false,
workspace_id: false,
attribution: false,
})
));
assert_stored_cause(
&reopened.read_by_name("agent").unwrap().unwrap(),
StartupStage::Binding,
);
}
#[test]
fn startup_failure_cas_compares_existing_binding_even_when_attachment_is_unchanged() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let mut original = metadata();
original.startup_failure = Some(bound_failure(&original, &attachment(StartupStage::Feature)));
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let mut stale = expected.clone();
bound_mut(stale.startup_failure.as_mut().unwrap()).session_id = crate::new_session_id();
assert!(matches!(
store.compare_and_swap_startup_failure(&stale, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Conflict
));
assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original);
let mut changed_fingerprint = expected.clone();
bound_mut(changed_fingerprint.startup_failure.as_mut().unwrap()).attribution_fingerprint[0] ^=
1;
assert!(matches!(
store.compare_and_swap_startup_failure(
&changed_fingerprint,
&attachment(StartupStage::Binding)
),
StartupFailureSaveResult::Conflict
));
assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original);
}
#[test]
fn startup_failure_bound_to_absent_optional_owner_fields_reopens_as_present() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let mut original = metadata();
original.workspace_id = None;
original.subjektiv_session_attribution = None;
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)),
StartupFailureSaveResult::Persisted
));
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
assert_stored_cause(&restored, StartupStage::Binding);
let expected_none_fingerprint: [u8; 32] = Sha256::digest(b"null").into();
assert_eq!(
bound(restored.startup_failure.as_ref().unwrap()).attribution_fingerprint,
expected_none_fingerprint
);
}
#[test]
fn startup_failure_cas_independently_opened_stores_have_one_concurrent_winner() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let barrier = Arc::new(std::sync::Barrier::new(3));
let handles = [StartupStage::Binding, StartupStage::Feature].map(|stage| {
let store = FsWorkerStore::new(temp.path()).unwrap();
let expected = expected.clone();
let barrier = barrier.clone();
std::thread::spawn(move || {
barrier.wait();
(
stage,
store.compare_and_swap_startup_failure(&expected, &attachment(stage)),
)
})
});
barrier.wait();
let mut winner = None;
let mut conflicts = 0;
for handle in handles {
let (stage, result) = handle.join().unwrap();
match result {
StartupFailureSaveResult::Persisted => {
assert!(winner.replace(stage).is_none());
}
StartupFailureSaveResult::Conflict => conflicts += 1,
other => panic!("unexpected persistence result: {other:?}"),
}
}
assert_eq!(conflicts, 1);
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
assert_stored_cause(
&reopened.read_by_name("agent").unwrap().unwrap(),
winner.unwrap(),
);
}
#[test]
fn startup_failure_legacy_missing_unknown_and_corrupt_survive_reopen_with_attribution() {
let temp = tempfile::tempdir().unwrap();
let original = metadata();
let mut future: serde_json::Value = serde_json::from_str(
serde_json::to_value(attachment(StartupStage::Binding))
.unwrap()
.as_str()
.unwrap(),
)
.unwrap();
future["version"] = serde_json::json!(65535);
let fixtures = [
None,
Some(future.to_string()),
Some("not an envelope".to_owned()),
];
for (case, encoded) in fixtures.into_iter().enumerate() {
let mut fixture = serde_json::to_value(&original).unwrap();
assert!(fixture.get("startup_failure").is_none());
if let Some(encoded) = encoded {
// The new field is always a binding wrapper. Unknown/corrupt fixtures
// change only its opaque codec payload, not the surrounding binding.
let mut bound =
serde_json::to_value(bound_failure(&original, &attachment(StartupStage::Binding)))
.unwrap();
bound["attachment"] = serde_json::Value::String(encoded);
fixture["startup_failure"] = bound;
}
let path = temp.path().join("agent/metadata.json");
fs::create_dir_all(path.parent().unwrap()).unwrap();
fs::write(path, serde_json::to_vec(&fixture).unwrap()).unwrap();
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
match (
case,
restored.decode_startup_failure::<RuntimeStartupCause>(),
) {
(0, StartupFailureRead::Missing)
| (1, StartupFailureRead::Present(DecodeFailure::Unsupported(_)))
| (2, StartupFailureRead::Present(DecodeFailure::Corrupt(_))) => {}
_ => panic!("legacy absence, unknown and corrupt must be distinguishable"),
}
assert_eq!(
restored.subjektiv_session_attribution,
original.subjektiv_session_attribution
);
assert_eq!(restored.active, original.active);
assert_eq!(restored.workspace_id, original.workspace_id);
if case != 0 {
reopened
.update_by_name("agent", |value| {
value.workspace_id = Some("new-owner".into())
})
.unwrap();
let changed = reopened.read_by_name("agent").unwrap().unwrap();
assert_eq!(changed.startup_failure, restored.startup_failure);
assert!(matches!(
changed.decode_startup_failure::<RuntimeStartupCause>(),
StartupFailureRead::Mismatch(StartupFailureBindingMismatch {
workspace_id: true,
..
})
));
}
}
}
#[test]
fn startup_failure_save_read_error_is_not_saved_and_preserves_received_cause() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let path = temp.path().join("agent/metadata.json");
fs::write(&path, b"invalid metadata").unwrap();
let cause = attachment(StartupStage::Binding);
let received = cause.clone();
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::NotSaved(WorkerStoreError::Serde(_))
));
assert_eq!(cause, received);
assert_eq!(fs::read(path).unwrap(), b"invalid metadata");
}
#[test]
fn startup_failure_rename_error_is_not_saved_but_post_rename_sync_error_is_unknown() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let cause = attachment(StartupStage::Binding);
let received = cause.clone();
// Real IO rename failure before publication. The destination is an isolated
// directory rather than metadata.json, so no retained metadata is destroyed.
let blocked_path = temp.path().join("blocked/metadata.json");
fs::create_dir_all(&blocked_path).unwrap();
let result =
save_startup_failure_locked(&store, &store.update_lock, &expected, &cause, |value| {
write_metadata_atomically(value, &blocked_path)
});
assert!(matches!(
result,
StartupFailureSaveResult::NotSaved(WorkerStoreError::Io(_))
));
assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original);
assert_eq!(
fs::read_dir(blocked_path.parent().unwrap())
.unwrap()
.count(),
1
);
// Production serialization + file sync + rename, replacing only directory
// sync with an IO failure. Reopen sees publication, but durability is unknown.
let path = store.metadata_path("agent").unwrap();
let result =
save_startup_failure_locked(&store, &store.update_lock, &expected, &cause, |value| {
write_metadata_with_directory_sync(value, &path, |_| {
Err(std::io::Error::other("directory sync unavailable"))
})
});
assert!(matches!(
result,
StartupFailureSaveResult::OutcomeUnknown(WorkerStoreError::Io(_))
));
assert_eq!(cause, received);
drop(store);
let reopened = FsWorkerStore::new(temp.path()).unwrap();
let restored = reopened.read_by_name("agent").unwrap().unwrap();
assert_stored_cause(&restored, StartupStage::Binding);
let mut wanted = original;
wanted.startup_failure = Some(SessionStartupFailure::from_expected(&expected, &received));
assert_eq!(restored, wanted);
// An uncertain attempt is not replayed against its stale expected cause.
assert!(matches!(
reopened.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Conflict
));
}
#[test]
fn startup_failure_aggregate_and_combined_stores_use_atomic_cas() {
let temp = tempfile::tempdir().unwrap();
let aggregate = WorkerAggregateStore::new(temp.path().join("aggregate"), "agent").unwrap();
let original = metadata();
aggregate.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let store = CombinedStore::new((), aggregate);
let cause = attachment(StartupStage::Binding);
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &cause),
StartupFailureSaveResult::Persisted
));
assert!(matches!(
store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Feature)),
StartupFailureSaveResult::Conflict
));
let reopened =
WorkerAggregateStore::open_read_only(temp.path().join("aggregate"), "agent").unwrap();
assert_stored_cause(
&reopened.read_read_only().unwrap().unwrap(),
StartupStage::Binding,
);
}
fn bound(failure: &SessionStartupFailure) -> &StartupFailureBinding {
match failure {
SessionStartupFailure::Bound(binding) => binding,
_ => panic!("expected current binding"),
}
}
fn bound_mut(failure: &mut SessionStartupFailure) -> &mut StartupFailureBinding {
match failure {
SessionStartupFailure::Bound(binding) => binding,
_ => panic!("expected current binding"),
}
}
#[test]
fn startup_binding_unknown_version_and_malformed_value_do_not_destroy_lifecycle_metadata() {
let temp = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(temp.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let mut wire = serde_json::to_value(&original).unwrap();
for (value, unsupported) in [
(
serde_json::json!({"version": 900, "future_binding": true}),
true,
),
(
serde_json::json!({"version": 1, "attachment": {"wrong_type": true}}),
false,
),
(serde_json::json!("invalid binding"), false),
(serde_json::json!([1, 2]), false),
] {
wire["startup_failure"] = value.clone();
std::fs::write(
temp.path().join("agent/metadata.json"),
serde_json::to_vec(&wire).unwrap(),
)
.unwrap();
let reopened = FsWorkerStore::new(temp.path())
.unwrap()
.read_by_name("agent")
.unwrap()
.unwrap();
assert_eq!(reopened.active, original.active);
assert_eq!(
reopened.subjektiv_session_attribution,
original.subjektiv_session_attribution
);
assert_eq!(
reopened.resolved_manifest_snapshot,
original.resolved_manifest_snapshot
);
match reopened.decode_startup_failure::<RuntimeStartupCause>() {
StartupFailureRead::UnsupportedBindingVersion(900) if unsupported => {}
StartupFailureRead::CorruptBinding if !unsupported => {}
_ => panic!("binding diagnosis lost"),
}
assert_eq!(
serde_json::to_value(&reopened).unwrap()["startup_failure"],
value
);
}
}
+341 -51
View File
@@ -16,7 +16,10 @@
//! execution data, not proof of liveness.
use crate::{SegmentId, SessionId};
use error_contract::{DecodeFailure, WireDetails};
use protocol::startup_failure::StoredStartupFailureAttachment;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
@@ -181,6 +184,167 @@ pub struct WorkerMetadata {
/// absence is also the legacy representation and never grants visibility.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub subjektiv_session_attribution: Option<SubjektivSessionAttributionState>,
/// Projected startup cause bound to its Session and ownership observation.
/// Pointer/ownership changes retain this evidence, but the read accessor
/// rejects a stale binding before decoding. This is not execution state.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub startup_failure: Option<SessionStartupFailure>,
}
/// Opaque cause and the exact local binding under which it was saved.
///
/// Only the CAS save path constructs this binding from its matched expectation.
/// Attribution is fingerprinted rather than copied: its existing diagnostics may
/// contain credentials or private paths and must not enter this new attachment.
/// The fingerprint is an equality fence, not encryption or ownership authority.
/// Deserialization merely restores evidence, not current ownership or authority;
/// consumers read it through `WorkerMetadata::decode_startup_failure`.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum SessionStartupFailure {
Bound(StartupFailureBinding),
/// Malformed or future binding representations remain alongside lifecycle metadata.
Undecodable(serde_json::Value),
}
impl std::fmt::Debug for SessionStartupFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SessionStartupFailure")
.finish_non_exhaustive()
}
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StartupFailureBinding {
version: u16,
worker_name: String,
session_id: SessionId,
workspace_id: Option<String>,
attribution_fingerprint: [u8; 32],
attachment: StoredStartupFailureAttachment,
}
// The compact JSON serialization of this typed Option is the fingerprint input:
// None is `null`; Some retains every enum field in its declared Serialize order.
// There are no maps whose iteration order could vary. Changing that representation
// requires a new binding version. Version 1 uses SHA-256 over precisely these bytes.
// This enum and SessionId serialize infallibly
// to JSON; no caller-controlled Serialize implementation participates here.
fn attribution_fingerprint(attribution: &Option<SubjektivSessionAttributionState>) -> [u8; 32] {
let encoded = serde_json::to_vec(attribution).expect("typed attribution serializes to JSON");
Sha256::digest(encoded).into()
}
impl SessionStartupFailure {
fn from_expected(
expected: &StartupFailureExpectation,
attachment: &StoredStartupFailureAttachment,
) -> Self {
Self::Bound(StartupFailureBinding {
version: 1,
worker_name: expected.worker_name.clone(),
session_id: expected.session_id,
workspace_id: expected.workspace_id.clone(),
attribution_fingerprint: attribution_fingerprint(&expected.attribution),
attachment: attachment.clone(),
})
}
}
impl StartupFailureBinding {
fn binding_mismatch(&self, metadata: &WorkerMetadata) -> Option<StartupFailureBindingMismatch> {
let mismatch = StartupFailureBindingMismatch {
worker_name: self.worker_name != metadata.worker_name,
session_id: Some(self.session_id)
!= metadata.active.as_ref().map(|active| active.session_id),
workspace_id: self.workspace_id != metadata.workspace_id,
attribution: self.attribution_fingerprint
!= attribution_fingerprint(&metadata.subjektiv_session_attribution),
};
(mismatch.worker_name
|| mismatch.session_id
|| mismatch.workspace_id
|| mismatch.attribution)
.then_some(mismatch)
}
}
/// Each `true` field identifies a binding that differs from current metadata.
/// Values and legacy diagnostic text are not copied into the diagnosis.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StartupFailureBindingMismatch {
pub worker_name: bool,
/// Also true when there is no longer an active Session.
pub session_id: bool,
pub workspace_id: bool,
pub attribution: bool,
}
/// A current binding must match before the consumer's typed codec runs.
/// `Present` is only cause evidence, never execution or retry authorization.
pub enum StartupFailureRead<D> {
Missing,
/// Binding encoding is versioned independently of the domain error envelope.
UnsupportedBindingVersion(u16),
CorruptBinding,
/// Retained cause belongs to a different Session/ownership observation.
/// No cause decoding is attempted, even for unknown or corrupt attachments.
Mismatch(StartupFailureBindingMismatch),
/// Known, unsupported and corrupt causes remain distinct in the codec.
Present(DecodeFailure<D>),
}
/// The observed ownership and cause values for one startup-cause write.
///
/// Segment movement within the same Session does not invalidate this expectation.
/// Attribution is compared in full, including legacy diagnostics, never inferred
/// from those diagnostics. The existing cause comparison includes its entire saved
/// binding, not just attachment bytes. A cause write changes neither attribution
/// nor ownership; it builds the new binding from this matched expectation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartupFailureExpectation {
pub worker_name: String,
pub workspace_id: Option<String>,
pub session_id: SessionId,
pub attribution: Option<SubjektivSessionAttributionState>,
pub startup_failure: Option<SessionStartupFailure>,
}
impl StartupFailureExpectation {
/// Capture an expectation only when metadata identifies an active Session.
pub fn from_metadata(metadata: &WorkerMetadata) -> Option<Self> {
Some(Self {
worker_name: metadata.worker_name.clone(),
workspace_id: metadata.workspace_id.clone(),
session_id: metadata.active.as_ref()?.session_id,
attribution: metadata.subjektiv_session_attribution.clone(),
startup_failure: metadata.startup_failure.clone(),
})
}
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)
&& metadata.subjektiv_session_attribution == self.attribution
&& metadata.startup_failure == self.startup_failure
}
}
/// Local persistence evidence, never a replacement for the received startup cause.
#[derive(Debug)]
pub enum StartupFailureSaveResult {
/// The attachment and containing directory were synced successfully.
Persisted,
/// The observed Session, ownership, attribution, or cause no longer matches.
/// No write was attempted. Missing metadata is also a conflict, not creation.
Conflict,
/// This attempt did not publish metadata. The caller still owns its cause.
NotSaved(WorkerStoreError),
/// Publication may have happened; reconcile retained metadata before retrying.
/// For the filesystem stores this means rename succeeded but directory sync failed.
OutcomeUnknown(WorkerStoreError),
}
impl WorkerMetadata {
@@ -196,9 +360,44 @@ impl WorkerMetadata {
peers: Vec::new(),
resolved_manifest_snapshot: None,
subjektiv_session_attribution: None,
startup_failure: None,
}
}
/// Absence is legacy/missing evidence, not a cause inferred from diagnostics.
/// A retained cause whose Session or ownership observation no longer matches
/// yields `Mismatch`, never `Missing` or a decoded cause for the new owner.
/// Segment movement within the same Session does not invalidate the binding.
/// Matched attachments use only the consumer's typed codec, preserving known,
/// unsupported and corrupt results. No result authorizes execution or retry.
pub fn decode_startup_failure<D: WireDetails>(&self) -> StartupFailureRead<D> {
let Some(failure) = &self.startup_failure else {
return StartupFailureRead::Missing;
};
#[derive(Deserialize)]
struct BindingVersion {
version: u16,
}
let failure = match failure {
SessionStartupFailure::Bound(binding) if binding.version == 1 => binding,
SessionStartupFailure::Bound(binding) => {
return StartupFailureRead::UnsupportedBindingVersion(binding.version);
}
SessionStartupFailure::Undecodable(raw) => {
return match serde_json::from_value::<BindingVersion>(raw.clone()) {
Ok(header) if header.version != 1 => {
StartupFailureRead::UnsupportedBindingVersion(header.version)
}
_ => StartupFailureRead::CorruptBinding,
};
}
};
if let Some(mismatch) = failure.binding_mismatch(self) {
return StartupFailureRead::Mismatch(mismatch);
}
StartupFailureRead::Present(failure.attachment.decode())
}
pub fn with_workspace_root(mut self, workspace_root: PathBuf) -> Self {
self.workspace_root = Some(workspace_root);
self
@@ -265,6 +464,25 @@ pub trait WorkerMetadataStore: Send + Sync {
Ok(matched)
}
/// Persist a received cause only against the exact observed Session and owner.
///
/// Implementations must serialize this comparison and publication with their
/// other read/modify/write operations. The filesystem stores use the same
/// process-local mutex as `update_by_name`; this is not a cross-process lock.
/// Uncoordinated direct writes are outside that guarantee. A default backend
/// refuses rather than offering a non-atomic read/write approximation.
/// The borrowed attachment is never consumed or changed on any result.
fn compare_and_swap_startup_failure(
&self,
_expected: &StartupFailureExpectation,
_failure: &StoredStartupFailureAttachment,
) -> StartupFailureSaveResult {
StartupFailureSaveResult::NotSaved(WorkerStoreError::Io(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"atomic startup failure persistence is unsupported by this backend",
)))
}
/// Set the active pointer while preserving spawned children, workspace ownership, and manifest snapshot.
fn set_active(
&self,
@@ -404,6 +622,90 @@ pub trait WorkerMetadataStore: Send + Sync {
}
}
/// Whether an atomic metadata replacement reached its publication point.
struct MetadataWriteFailure {
published: bool,
error: WorkerStoreError,
}
fn write_metadata_atomically(
metadata: &WorkerMetadata,
path: &Path,
) -> Result<(), MetadataWriteFailure> {
write_metadata_with_directory_sync(metadata, path, |parent| {
std::fs::File::open(parent)?.sync_all()
})
}
// Keep the directory durability boundary explicit: a failure here follows rename,
// unlike failures writing/syncing the temporary file. Tests replace only this IO
// boundary, retaining the real serialization, temporary file and rename path.
fn write_metadata_with_directory_sync(
metadata: &WorkerMetadata,
path: &Path,
sync_directory: impl FnOnce(&Path) -> std::io::Result<()>,
) -> Result<(), MetadataWriteFailure> {
let mut published = false;
let mut temp = None;
let result = (|| -> Result<(), WorkerStoreError> {
use std::io::Write;
let mut content = serde_json::to_vec_pretty(metadata)?;
content.push(b'\n');
let parent = path.parent().expect("metadata path has parent");
fs::create_dir_all(parent)?;
let temporary_path = parent.join(format!(
".metadata.json.tmp-{}-{}",
std::process::id(),
uuid::Uuid::now_v7()
));
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary_path)?;
temp = Some(temporary_path.clone());
file.write_all(&content)?;
file.sync_all()?;
drop(file);
fs::rename(&temporary_path, path)?;
published = true;
sync_directory(parent)?;
Ok(())
})();
if result.is_err() {
if let Some(temp) = temp {
let _ = fs::remove_file(temp);
}
}
result.map_err(|error| MetadataWriteFailure { published, error })
}
fn save_startup_failure_locked<S: WorkerMetadataStore>(
store: &S,
update_lock: &Mutex<()>,
expected: &StartupFailureExpectation,
failure: &StoredStartupFailureAttachment,
publish: impl FnOnce(&WorkerMetadata) -> Result<(), MetadataWriteFailure>,
) -> StartupFailureSaveResult {
let _guard = update_lock.lock().expect("metadata update lock poisoned");
let mut metadata = match store.read_by_name(&expected.worker_name) {
Ok(Some(metadata)) if expected.matches(&metadata) => metadata,
Ok(_) => return StartupFailureSaveResult::Conflict,
Err(error) => return StartupFailureSaveResult::NotSaved(error),
};
metadata.startup_failure = Some(SessionStartupFailure::from_expected(expected, failure));
match publish(&metadata) {
Ok(()) => StartupFailureSaveResult::Persisted,
Err(MetadataWriteFailure {
published: false,
error,
}) => StartupFailureSaveResult::NotSaved(error),
Err(MetadataWriteFailure {
published: true,
error,
}) => StartupFailureSaveResult::OutcomeUnknown(error),
}
}
/// Metadata store for one canonical Worker aggregate.
///
/// The Worker key is fixed by Runtime authority and metadata lives directly at
@@ -489,31 +791,7 @@ impl WorkerAggregateStore {
impl WorkerMetadataStore for WorkerAggregateStore {
fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> {
self.validate_name(&metadata.worker_name)?;
let mut content = serde_json::to_vec_pretty(metadata)?;
content.push(b'\n');
let path = self.metadata_path();
let temp = self.root.join(format!(
".metadata.json.tmp-{}-{}",
std::process::id(),
uuid::Uuid::now_v7()
));
let result = (|| -> Result<(), WorkerStoreError> {
use std::io::Write;
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)?;
file.write_all(&content)?;
file.sync_all()?;
drop(file);
fs::rename(&temp, &path)?;
std::fs::File::open(&self.root)?.sync_all()?;
Ok(())
})();
if result.is_err() {
let _ = fs::remove_file(temp);
}
result
write_metadata_atomically(metadata, &self.metadata_path()).map_err(|failure| failure.error)
}
fn read_by_name(&self, worker_name: &str) -> Result<Option<WorkerMetadata>, WorkerStoreError> {
@@ -553,6 +831,16 @@ impl WorkerMetadataStore for WorkerAggregateStore {
Ok(metadata)
}
fn compare_and_swap_startup_failure(
&self,
expected: &StartupFailureExpectation,
failure: &StoredStartupFailureAttachment,
) -> StartupFailureSaveResult {
save_startup_failure_locked(self, &self.update_lock, expected, failure, |metadata| {
write_metadata_atomically(metadata, &self.metadata_path())
})
}
fn compare_and_swap_active(
&self,
worker_name: &str,
@@ -627,32 +915,7 @@ impl FsWorkerStore {
impl WorkerMetadataStore for FsWorkerStore {
fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> {
let path = self.metadata_path(&metadata.worker_name)?;
let mut content = serde_json::to_vec_pretty(metadata)?;
content.push(b'\n');
let parent = path.parent().expect("metadata path has parent");
fs::create_dir_all(parent)?;
let temp = parent.join(format!(
".metadata.json.tmp-{}-{}",
std::process::id(),
uuid::Uuid::now_v7()
));
let result = (|| -> Result<(), WorkerStoreError> {
use std::io::Write;
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)?;
file.write_all(&content)?;
file.sync_all()?;
drop(file);
fs::rename(&temp, &path)?;
std::fs::File::open(parent)?.sync_all()?;
Ok(())
})();
if result.is_err() {
let _ = fs::remove_file(temp);
}
result
write_metadata_atomically(metadata, &path).map_err(|failure| failure.error)
}
fn read_by_name(&self, worker_name: &str) -> Result<Option<WorkerMetadata>, WorkerStoreError> {
@@ -685,6 +948,20 @@ impl WorkerMetadataStore for FsWorkerStore {
Ok(metadata)
}
fn compare_and_swap_startup_failure(
&self,
expected: &StartupFailureExpectation,
failure: &StoredStartupFailureAttachment,
) -> StartupFailureSaveResult {
let path = match self.metadata_path(&expected.worker_name) {
Ok(path) => path,
Err(error) => return StartupFailureSaveResult::NotSaved(error),
};
save_startup_failure_locked(self, &self.update_lock, expected, failure, |metadata| {
write_metadata_atomically(metadata, &path)
})
}
fn compare_and_swap_active(
&self,
worker_name: &str,
@@ -1007,6 +1284,15 @@ where
self.worker_metadata_store
.update_by_name(worker_name, update)
}
fn compare_and_swap_startup_failure(
&self,
expected: &StartupFailureExpectation,
failure: &StoredStartupFailureAttachment,
) -> StartupFailureSaveResult {
self.worker_metadata_store
.compare_and_swap_startup_failure(expected, failure)
}
fn compare_and_swap_active(
&self,
worker_name: &str,
@@ -1512,3 +1798,7 @@ mod tests {
assert_eq!(winners, 1);
}
}
#[cfg(test)]
#[path = "startup_failure_tests.rs"]
mod startup_failure_tests;
+11
View File
@@ -147,3 +147,14 @@ pub enum RuntimeError {
#[error("runtime state lock was poisoned")]
StatePoisoned,
}
impl RuntimeError {
/// 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),
_ => None,
}
}
}
+9
View File
@@ -103,6 +103,8 @@ pub struct WorkerExecutionResult {
pub startup_resources: Option<protocol::startup_failure::StartupResourceFacts>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub startup_failure_persistence: Option<protocol::startup_failure::StartupPersistenceOutcome>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub startup_io_kind: Option<protocol::startup_failure::StartupIoKind>,
}
/// Backend result class for a Worker execution operation.
@@ -129,6 +131,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -158,6 +161,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -178,6 +182,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -193,6 +198,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -208,6 +214,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -223,6 +230,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
@@ -238,6 +246,7 @@ impl WorkerExecutionResult {
secondary_failures: Vec::new(),
startup_resources: None,
startup_failure_persistence: None,
startup_io_kind: None,
}
}
+303 -8
View File
@@ -47,6 +47,9 @@ use crate::retention::{
FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult,
WorkerRetentionInventory, WorkerRetentionInventorySnapshot, WorkerRetentionProvider,
};
use protocol::startup_failure::{
StartupFailurePhase, StartupPersistenceOutcome, StartupSecondaryFailure,
};
use protocol::subscription::{
EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot,
SubscriptionValidationError, SubscriptionWorkdirId, SubscriptionWorker, SubscriptionWorkerId,
@@ -1522,10 +1525,41 @@ impl Runtime {
worker_state,
workdir_attachments,
} => (worker_state, workdir_attachments),
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::RolledBack(result)
| WorkerExecutionSpawnResult::Errored(result) => {
self.rollback_failed_create(&worker_ref)?;
WorkerExecutionSpawnResult::Rejected(mut result)
| WorkerExecutionSpawnResult::RolledBack(mut result)
| WorkerExecutionSpawnResult::Errored(mut result) => {
if let Err((error, phase, outcome)) =
self.rollback_failed_create_with_phase(&worker_ref)
{
tracing::error!(
worker_id = %worker_ref.worker_id,
?phase,
%error,
"failed to roll back Worker creation"
);
// A later failure must not make a confirmed record deletion
// indistinguishable from a failed deletion with unknown effects.
if matches!(
phase,
StartupFailurePhase::RuntimeSnapshotSave
| StartupFailurePhase::RuntimeRemovalPublish
) {
result.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::RuntimeRecordDelete,
outcome: StartupPersistenceOutcome::Persisted,
});
}
if phase == StartupFailurePhase::RuntimeRemovalPublish {
result.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::RuntimeSnapshotSave,
outcome: StartupPersistenceOutcome::Persisted,
});
}
result.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase,
outcome,
});
}
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
@@ -3811,9 +3845,33 @@ impl Runtime {
}
fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
self.rollback_failed_create_with_phase(worker_ref)
.map_err(|(error, _, _)| error)
}
// The legacy caller still receives its RuntimeError. Spawn failures instead
// retain their primary result and attach the completed/uncertain rollback phases.
fn rollback_failed_create_with_phase(
&self,
worker_ref: &WorkerRef,
) -> Result<(), (RuntimeError, StartupFailurePhase, StartupPersistenceOutcome)> {
let mut state = self.lock().map_err(|error| {
(
error,
StartupFailurePhase::RuntimeRecordDelete,
StartupPersistenceOutcome::NotSaved,
)
})?;
if state.workers.contains_key(&worker_ref.worker_id) {
state.delete_worker_record(&worker_ref.worker_id)?;
state
.delete_worker_record(&worker_ref.worker_id)
.map_err(|error| {
(
error,
StartupFailurePhase::RuntimeRecordDelete,
StartupPersistenceOutcome::OutcomeUnknown,
)
})?;
let record = state
.workers
.remove(&worker_ref.worker_id)
@@ -3826,8 +3884,22 @@ impl Runtime {
state
.observation_events
.retain(|event| event.worker_ref != *worker_ref);
state.persist_runtime_snapshot()?;
state.publish_worker_removed(worker_ref.worker_id, workspace_id.as_deref())?;
state.persist_runtime_snapshot().map_err(|error| {
(
error,
StartupFailurePhase::RuntimeSnapshotSave,
StartupPersistenceOutcome::OutcomeUnknown,
)
})?;
state
.publish_worker_removed(worker_ref.worker_id, workspace_id.as_deref())
.map_err(|error| {
(
error,
StartupFailurePhase::RuntimeRemovalPublish,
StartupPersistenceOutcome::OutcomeUnknown,
)
})?;
}
Ok(())
}
@@ -8257,6 +8329,223 @@ mod tests {
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Barrier, Condvar, Mutex};
#[cfg(feature = "fs-store")]
mod create_rollback_tests {
//! Connect the real create/rollback path to FsRuntimeStore and reopen its disk state.
//! Only execution startup is a fixture. Filesystem topology faults are injected after
//! the initial record is durable; existing write hooks cover Worker writes, not these
//! delete/snapshot boundaries. Run with the normal worker-runtime library tests.
use super::*;
use error_contract::{
DecodeFailure, FailureFacts, InternalTransfer, OperationContext, project_failure,
};
use protocol::startup_failure::{
RuntimeStartupCause, StartupCleanup, StartupCleanupFailure, StartupFailureAttachment,
StartupResourceFacts, StartupResources, StartupStage,
};
#[derive(Clone, Copy)]
enum Fault {
None,
RecordDelete,
SnapshotSave,
}
fn primary_failure() -> WorkerExecutionResult {
let mut facts = FailureFacts::new(RuntimeStartupCause {
reason: protocol::startup_failure::RuntimeStartupReason::StartupFailed,
stage: StartupStage::Controller,
})
.unwrap();
facts
.with_context(OperationContext::new("worker_startup", "runtime").unwrap())
.unwrap();
let projected = project_failure::<InternalTransfer, _>(&facts, &()).unwrap();
let attachment = StartupFailureAttachment::from_projected(&projected).unwrap();
let mut result = WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
"primary startup failure",
)
.with_startup_failure(attachment)
.with_worker_state(protocol::WorkerStatus::Stopped.into());
result.push_secondary_failure(StartupSecondaryFailure::Cleanup {
failure: StartupCleanupFailure::Workdir,
});
result.startup_resources = Some(StartupResourceFacts {
resources: StartupResources::Unproven,
cleanup: StartupCleanup::Failed(StartupCleanupFailure::Workdir),
durable_effects_possible: true,
});
result
}
fn check_rollback(fault: Fault) {
let primary = primary_failure();
for spawn_result in [
WorkerExecutionSpawnResult::Rejected(primary.clone()),
WorkerExecutionSpawnResult::RolledBack(primary.clone()),
WorkerExecutionSpawnResult::Errored(primary.clone()),
] {
let root = tempfile::tempdir().unwrap();
let store_root = root.path().join("runtime");
let options =
FsRuntimeStoreOptions::new(&store_root).with_runtime_id("create-rollback");
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("failed create rollback");
let worker_ref = WorkerRef::new(request.worker_id);
let worker_dir = store_root
.join("workers")
.join(request.worker_id.to_string());
let runtime_path = store_root.join("runtime.json");
let saved_snapshot = root.path().join("saved-runtime.json");
let record_fault_path = worker_dir.join("metadata.json");
*backend.spawn_result.lock().unwrap() = Some(spawn_result);
*backend.before_spawn_result.lock().unwrap() = Some(Box::new({
let worker_dir = worker_dir.clone();
let runtime_path = runtime_path.clone();
let saved_snapshot = saved_snapshot.clone();
let record_fault_path = record_fault_path.clone();
move || {
assert!(worker_dir.join("worker.json").is_file());
match fault {
Fault::None => {}
Fault::RecordDelete => {
// Reading a directory as metadata fails before aggregate removal.
std::fs::create_dir(record_fault_path).unwrap();
}
Fault::SnapshotSave => {
std::fs::rename(&runtime_path, saved_snapshot).unwrap();
std::fs::create_dir(runtime_path).unwrap();
}
}
}
}));
let error = runtime.create_worker(request).unwrap_err();
let RuntimeError::WorkerExecutionRejected {
worker_id,
operation,
outcome,
message,
result,
} = error
else {
panic!("rollback must not replace the primary execution result: {error:?}");
};
assert_eq!(worker_id, worker_ref.worker_id);
assert_eq!(operation, primary.operation);
assert_eq!(outcome, primary.outcome);
assert_eq!(message, primary.message_or_default());
let mut expected = primary.clone();
match fault {
Fault::None => {}
Fault::RecordDelete => {
expected.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::RuntimeRecordDelete,
outcome: StartupPersistenceOutcome::OutcomeUnknown,
});
}
Fault::SnapshotSave => {
expected.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::RuntimeRecordDelete,
outcome: StartupPersistenceOutcome::Persisted,
});
expected.push_secondary_failure(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::RuntimeSnapshotSave,
outcome: StartupPersistenceOutcome::OutcomeUnknown,
});
}
}
assert_eq!(
result, expected,
"primary facts and existing cleanup must survive"
);
let Some(DecodeFailure::Known(decoded)) =
result.startup_failure::<RuntimeStartupCause>()
else {
panic!("primary typed cause must remain decodable");
};
assert_eq!(decoded.facts().cause().stage, StartupStage::Controller);
assert_eq!(
decoded.facts().context().operations,
vec![OperationContext::new("worker_startup", "runtime").unwrap()]
);
let record_retained = matches!(fault, Fault::RecordDelete);
assert_eq!(worker_dir.join("worker.json").is_file(), record_retained);
assert_eq!(runtime.worker_detail(&worker_ref).is_ok(), record_retained);
match fault {
Fault::RecordDelete => {
let persisted = runtime_store(&runtime)
.read_worker_record(&worker_ref)
.unwrap();
assert_eq!(persisted.worker_ref, worker_ref);
assert!(
runtime_store(&runtime)
.load_runtime_state()
.unwrap()
.workers
.contains_key(&worker_ref.worker_id)
);
// Remove only the injected obstruction, never the retained record.
std::fs::remove_dir(record_fault_path).unwrap();
}
Fault::SnapshotSave => {
assert!(runtime_path.is_dir());
assert!(runtime_store(&runtime).load_runtime_state().is_err());
}
Fault::None => {
assert!(
!runtime_store(&runtime)
.load_runtime_state()
.unwrap()
.workers
.contains_key(&worker_ref.worker_id)
);
}
}
drop(runtime);
drop(backend);
if matches!(fault, Fault::SnapshotSave) {
assert!(Runtime::with_fs_store(options.clone()).is_err());
// Repair the injected obstruction with the last durable snapshot,
// not a new snapshot synthesized from the in-memory rollback.
std::fs::remove_dir(runtime_path.clone()).unwrap();
std::fs::rename(saved_snapshot, runtime_path).unwrap();
}
let reopened = Runtime::with_fs_store(options).unwrap();
assert_eq!(reopened.worker_detail(&worker_ref).is_ok(), record_retained);
assert_eq!(
runtime_store(&reopened)
.load_runtime_state()
.unwrap()
.workers
.contains_key(&worker_ref.worker_id),
record_retained
);
}
}
#[test]
fn create_rollback_success_preserves_primary_and_existing_cleanup() {
check_rollback(Fault::None);
}
#[test]
fn create_rollback_record_delete_failure_preserves_primary_and_retained_record() {
check_rollback(Fault::RecordDelete);
}
#[test]
fn create_rollback_snapshot_failure_preserves_primary_and_confirmed_record_deletion() {
check_rollback(Fault::SnapshotSave);
}
}
#[test]
fn worker_create_failure_fields_only_include_sanitized_execution_detail() {
let invalid_request = RuntimeError::InvalidRequest("private-token".to_string());
@@ -9239,6 +9528,8 @@ mod tests {
recovery_absent: AtomicBool,
recovery_checks: std::sync::atomic::AtomicUsize,
spawn_result: Mutex<Option<WorkerExecutionSpawnResult>>,
#[cfg(feature = "fs-store")]
before_spawn_result: Mutex<Option<Box<dyn FnOnce() + Send>>>,
job_bindings: Mutex<Vec<Option<crate::catalog::BackendJobExecutionBinding>>>,
dispatch_result: Mutex<Option<WorkerExecutionResult>>,
stop_result: Mutex<Option<WorkerExecutionResult>>,
@@ -9446,6 +9737,10 @@ 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();
}
if let Some(result) = self.spawn_result.lock().unwrap().take() {
return result;
}
@@ -15,7 +15,7 @@ impl RuntimeWorkerFactory for SubjectDeleteDeferredCreate {
async fn spawn_controller(
&self,
_: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
self.entered.send(()).unwrap();
self.release.notified().await;
Err("create failed before any Controller".into())
+337 -166
View File
@@ -71,7 +71,7 @@ fn rollback_materialized_spawn_workdirs(
materializer: &dyn WorkingDirectoryMaterializer,
bindings: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
workdir_ids: &[String],
failure_context: &str,
failure_context: &WorkerExecutionResult,
) -> Result<(), WorkerExecutionSpawnResult> {
let mut cleanup_failures = Vec::new();
let mut uncertain_attachments = Vec::new();
@@ -93,13 +93,15 @@ fn rollback_materialized_spawn_workdirs(
return Ok(());
}
Err(WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
format!(
"{failure_context}; Workdir cleanup could not be proven: {}",
cleanup_failures.join("; ")
),
),
result: {
let mut result = failure_context.clone();
result.push_secondary_failure(
protocol::startup_failure::StartupSecondaryFailure::Cleanup {
failure: protocol::startup_failure::StartupCleanupFailure::Workdir,
},
);
result
},
worker_state: None,
workdir_attachments: uncertain_attachments,
})
@@ -165,6 +167,8 @@ fn subscribe_worker_protocol_session(handle: &WorkerHandle) -> WorkerProtocolSes
worker::ipc::protocol_session::subscribe_worker_protocol_session(handle)
}
mod startup_failure;
const DEFAULT_BACKEND_ID: &str = "worker-crate";
const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10);
const SPAWN_RESTORE_TASK_TIMEOUT: Duration = Duration::from_secs(60);
@@ -262,9 +266,81 @@ pub struct RuntimeRestoreFailure {
io_kind: Option<std::io::ErrorKind>,
cleanup_failure: Option<RuntimeRestoreCleanupFailure>,
durable_effects_possible: bool,
primary: protocol::startup_failure::StartupFailureAttachment,
secondary: Vec<protocol::startup_failure::StartupSecondaryFailure>,
legacy_diagnostic: Option<String>,
metadata_save: Option<protocol::startup_failure::StartupPersistenceOutcome>,
}
impl RuntimeRestoreFailure {
fn fallback_primary(
stage: RuntimeRestoreStage,
) -> protocol::startup_failure::StartupFailureAttachment {
worker::startup_failure::runtime_startup_attachment(
protocol::startup_failure::RuntimeStartupCause {
stage: stage.into(),
reason: protocol::startup_failure::RuntimeStartupReason::StartupFailed,
},
)
.expect("fixed Runtime startup projection fits the error-contract budget")
}
pub fn with_startup_failure(
mut self,
primary: protocol::startup_failure::StartupFailureAttachment,
) -> Self {
self.primary = primary;
self
}
pub fn startup_failure(&self) -> &protocol::startup_failure::StartupFailureAttachment {
&self.primary
}
pub fn io_kind(&self) -> Option<std::io::ErrorKind> {
self.io_kind
}
fn from_worker(stage: RuntimeRestoreStage, error: &WorkerError) -> Self {
let mut failure = Self::started(stage, restore_worker_cause(error));
failure.primary = error
.startup_failure_at(stage.into())
.expect("fixed Worker startup projection fits the error-contract budget");
failure.secondary = error.startup_secondary_failures().to_vec();
failure.metadata_save = error.startup_metadata_save();
failure
}
pub fn execution_result(&self, operation: WorkerExecutionOperation) -> WorkerExecutionResult {
use protocol::startup_failure::*;
let mut result = WorkerExecutionResult::errored(operation, self.to_string())
.with_startup_failure(self.primary.clone());
result.secondary_failures = self.secondary.clone();
result.startup_failure_persistence = self.metadata_save;
result.startup_io_kind = self.io_kind.map(Into::into);
// Retain the first cleanup failure even if a later retry completed cleanup.
if let Some(failure) = self.cleanup_failure {
result.push_secondary_failure(StartupSecondaryFailure::Cleanup {
failure: failure.into(),
});
}
result.startup_resources = Some(StartupResourceFacts {
resources: match self.resources {
RuntimeRestoreResources::None => StartupResources::None,
RuntimeRestoreResources::WorkerStartup => StartupResources::WorkerStartup,
RuntimeRestoreResources::Unproven => StartupResources::Unproven,
},
cleanup: match self.cleanup {
RuntimeRestoreCleanup::NotNeeded => StartupCleanup::NotNeeded,
RuntimeRestoreCleanup::Required => StartupCleanup::Required,
RuntimeRestoreCleanup::Confirmed => StartupCleanup::Confirmed,
RuntimeRestoreCleanup::Failed(failure) => StartupCleanup::Failed(failure.into()),
},
durable_effects_possible: self.durable_effects_possible,
});
result
}
pub fn refused(stage: RuntimeRestoreStage, cause: &'static str) -> Self {
Self {
stage,
@@ -274,6 +350,10 @@ impl RuntimeRestoreFailure {
io_kind: None,
cleanup_failure: None,
durable_effects_possible: false,
primary: Self::fallback_primary(stage),
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
}
}
@@ -286,6 +366,10 @@ impl RuntimeRestoreFailure {
io_kind: None,
cleanup_failure: None,
durable_effects_possible: false,
primary: Self::fallback_primary(stage),
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
}
}
@@ -298,6 +382,10 @@ impl RuntimeRestoreFailure {
io_kind: None,
cleanup_failure: None,
durable_effects_possible: false,
primary: Self::fallback_primary(stage),
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
}
}
@@ -311,7 +399,13 @@ impl RuntimeRestoreFailure {
impl std::fmt::Display for RuntimeRestoreFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let cause: String = self.cause.chars().take(160).collect();
let cause: String = self
.legacy_diagnostic
.as_deref()
.unwrap_or(self.cause)
.chars()
.take(160)
.collect();
write!(
f,
"restore {:?}: {}; io={:?}; resources={:?}; cleanup={:?}; prior_cleanup={:?}",
@@ -323,6 +417,7 @@ impl std::fmt::Display for RuntimeRestoreFailure {
fn restore_worker_cause(error: &WorkerError) -> &'static str {
match error {
WorkerError::ConstructionCleanup { source, .. }
| WorkerError::StartupFailure { source, .. }
| WorkerError::RestoreActivationEffects { source } => restore_worker_cause(source),
WorkerError::SegmentEmpty { .. } => "saved segment empty",
WorkerError::SegmentSystemPromptMissing { .. } => "saved segment system prompt missing",
@@ -368,14 +463,13 @@ async fn prepare_restore_operation<F: RuntimeWorkerFactory + ?Sized>(
request: WorkerExecutionRestoreRequest,
) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
let prior_effects = request.source_repair_possible;
factory.preflight_restore(&request).await.map_err(|_| {
let mut failure = RuntimeRestoreFailure::refused(
RuntimeRestoreStage::Preflight,
"restore preflight refused",
);
failure.durable_effects_possible = prior_effects;
failure
})?;
factory
.preflight_restore(&request)
.await
.map_err(|mut failure| {
failure.durable_effects_possible = prior_effects;
failure
})?;
let repaired = factory
.recover_restore_source(&request)
.await
@@ -479,12 +573,12 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
async fn spawn_controller(
&self,
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String>;
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure>;
async fn preflight_restore(
&self,
_request: &WorkerExecutionRestoreRequest,
) -> Result<(), String> {
) -> Result<(), RuntimeRestoreFailure> {
Ok(())
}
@@ -907,7 +1001,9 @@ impl ProfileRuntimeWorkerFactory {
request: &WorkerExecutionRestoreRequest,
error: &WorkerError,
) {
if let WorkerError::RestoreActivationEffects { source } = error {
if let WorkerError::RestoreActivationEffects { source }
| WorkerError::StartupFailure { source, .. } = error
{
self.retain_construction_failure(request, source);
return;
}
@@ -2008,7 +2104,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
async fn spawn_controller(
&self,
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
let worker_name = Self::runtime_worker_name(&request);
let profile = Self::runtime_profile(&request);
let has_local_filesystem = request.workdir_attachments.len() == 1;
@@ -2128,13 +2224,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
Some(factory) => bootstrap.with_model_client(factory()),
None => bootstrap,
};
let mut prepared = bootstrap.prepare().await.map_err(|error| match error {
WorkerBootstrapError::Worker(source) => {
format!("failed to create Worker from profile: {source}")
}
WorkerBootstrapError::Controller { source, .. } => {
format!("failed to prepare Worker controller: {source}")
}
let mut prepared = bootstrap.prepare().await.map_err(|error| {
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error)
})?;
let worker = prepared.worker_mut();
if let Some(snapshot) = &request.request.plugin_execution {
@@ -2160,12 +2251,34 @@ 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| format!("finalize Worker Session attribution: {error}"))?;
.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)
})?;
let workdir_capabilities = request
.workdir_attachments
.keys()
@@ -2206,19 +2319,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
return Err(format!(
"install Flow transition feature failed: {:?}",
report.reports
));
)
.into());
}
}
let workspace_client = worker.workspace_client_handle();
let started = prepared.start().await.map_err(|error| match error {
WorkerBootstrapError::Worker(source) => {
format!("failed to prepare Worker before controller start: {source}")
}
WorkerBootstrapError::Controller { source, .. } => format!(
"failed to spawn Worker controller in {}: {source}",
run_dir.display()
),
let started = prepared.start().await.map_err(|error| {
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error)
})?;
let (handle, shutdown_rx, controller_task) =
(started.handle, started.shutdown, started.controller_task);
@@ -2244,79 +2352,90 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
async fn preflight_restore(
&self,
request: &WorkerExecutionRestoreRequest,
) -> Result<(), String> {
#[cfg(not(feature = "http-server"))]
RuntimeWorkspaceBackendRef::from_worker_request(
&request.request,
self.runtime_id.as_deref(),
)?;
let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref);
let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?;
if !worker_aggregate_dir.is_dir() {
return Err("Persisted Worker aggregate metadata is unavailable".to_string());
}
if !worker_aggregate_dir.join("session").is_dir() {
return Err("Persisted Worker Session metadata is unavailable".to_string());
}
let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name)
.map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?;
let metadata = metadata_store
.read_by_name(&worker_name)
.map_err(|error| format!("failed to read Worker metadata: {error}"))?
.ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?;
let (manifest, loader) = Self::manifest_for_restore(&metadata, &request.request)?;
if let Some(active) = metadata.active.as_ref()
&& let Some(segment_id) = active.segment_id
{
// Preflight must not create lock files or migrate retained sources.
// Actual restore reacquires the live guard before any publication.
self.session_retention()?
.ensure_live(active.session_id)
.map_err(|error| format!("Worker Session source is unavailable: {error}"))?;
let session_store =
WorkerSessionStore::open_read_only(worker_aggregate_dir.join("session"))
.map_err(|error| format!("failed to read Worker Session store: {error}"))?;
if !session_store
.exists(active.session_id, segment_id)
.map_err(|error| format!("failed to inspect Worker Session segment: {error}"))?
) -> Result<(), RuntimeRestoreFailure> {
let result: Result<(), String> = async {
#[cfg(not(feature = "http-server"))]
RuntimeWorkspaceBackendRef::from_worker_request(
&request.request,
self.runtime_id.as_deref(),
)?;
let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref);
let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?;
if !worker_aggregate_dir.is_dir() {
return Err("Persisted Worker aggregate metadata is unavailable".to_string());
}
if !worker_aggregate_dir.join("session").is_dir() {
return Err("Persisted Worker Session metadata is unavailable".to_string());
}
let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name)
.map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?;
let metadata = metadata_store
.read_by_name(&worker_name)
.map_err(|error| format!("failed to read Worker metadata: {error}"))?
.ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?;
let (manifest, loader) = Self::manifest_for_restore(&metadata, &request.request)?;
if let Some(active) = metadata.active.as_ref()
&& let Some(segment_id) = active.segment_id
{
return Err("Persisted Worker Session segment is unavailable".to_string());
// Preflight must not create lock files or migrate retained sources.
// Actual restore reacquires the live guard before any publication.
self.session_retention()?
.ensure_live(active.session_id)
.map_err(|error| format!("Worker Session source is unavailable: {error}"))?;
let session_store =
WorkerSessionStore::open_read_only(worker_aggregate_dir.join("session"))
.map_err(|error| format!("failed to read Worker Session store: {error}"))?;
if !session_store
.exists(active.session_id, segment_id)
.map_err(|error| format!("failed to inspect Worker Session segment: {error}"))?
{
return Err("Persisted Worker Session segment is unavailable".to_string());
}
session_store
.read_all(active.session_id, segment_id)
.map_err(|error| format!("failed to read Worker Session segment: {error}"))?;
}
session_store
.read_all(active.session_id, segment_id)
.map_err(|error| format!("failed to read Worker Session segment: {error}"))?;
}
if let Some(api) = request.request.workspace_api.as_ref()
&& metadata
.active
.as_ref()
.is_some_and(|active| active.segment_id.is_none())
{
let bundle = request.config_bundle.as_ref().ok_or_else(|| {
"Pending Workspace Worker restore requires operation-owned launch material"
.to_string()
})?;
crate::config_bundle::validate_config_bundle(bundle)
.map_err(|error| format!("invalid pending Worker launch material: {error}"))?;
if bundle.metadata.workspace_id != api.workspace_id {
return Err(format!(
"Workspace Prompt projection scope mismatch: expected {}, got {}",
api.workspace_id, bundle.metadata.workspace_id
));
if let Some(api) = request.request.workspace_api.as_ref()
&& metadata
.active
.as_ref()
.is_some_and(|active| active.segment_id.is_none())
{
let bundle = request.config_bundle.as_ref().ok_or_else(|| {
"Pending Workspace Worker restore requires operation-owned launch material"
.to_string()
})?;
crate::config_bundle::validate_config_bundle(bundle)
.map_err(|error| format!("invalid pending Worker launch material: {error}"))?;
if bundle.metadata.workspace_id != api.workspace_id {
return Err(format!(
"Workspace Prompt projection scope mismatch: expected {}, got {}",
api.workspace_id, bundle.metadata.workspace_id
));
}
let prompt_catalog = bundle.prompt_catalog.as_ref().ok_or_else(|| {
"pending Workspace Worker restore requires a saved Workspace Prompt projection"
.to_string()
})?;
// Validate the exact launch catalog without publishing it into the
// current projection cache or allocating live Worker resources.
worker::SystemPromptTemplate::parse(
&manifest.engine.instruction,
loader.with_effective_catalog(prompt_catalog.clone()),
)
.map_err(|error| format!("invalid pending Worker launch Prompt: {error}"))?;
}
let prompt_catalog = bundle.prompt_catalog.as_ref().ok_or_else(|| {
"pending Workspace Worker restore requires a saved Workspace Prompt projection"
.to_string()
})?;
// Validate the exact launch catalog without publishing it into the
// current projection cache or allocating live Worker resources.
worker::SystemPromptTemplate::parse(
&manifest.engine.instruction,
loader.with_effective_catalog(prompt_catalog.clone()),
)
.map_err(|error| format!("invalid pending Worker launch Prompt: {error}"))?;
Ok(())
}
Ok(())
.await;
result.map_err(|diagnostic| {
let mut failure = RuntimeRestoreFailure::refused(
RuntimeRestoreStage::Preflight,
"restore preflight refused",
);
failure.legacy_diagnostic = Some(diagnostic);
failure
})
}
async fn recover_restore_source(
@@ -2347,12 +2466,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
self.runtime_id.as_deref(),
)
.map_err(|cause| RuntimeRestoreFailure::refused(RuntimeRestoreStage::Preflight, cause))?;
self.preflight_restore(&request).await.map_err(|_| {
RuntimeRestoreFailure::refused(
RuntimeRestoreStage::Preflight,
"restore preflight refused",
)
})?;
self.preflight_restore(&request).await?;
#[cfg(test)]
self.restore_faults
.executions
@@ -2455,7 +2569,10 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
.await
.map_err(|error| {
self.retain_construction_failure(&request, &error);
restore_worker_cause(&error)
let failure = RuntimeRestoreFailure::from_worker(stage, &error);
let cause = failure.cause;
preparation_failure = Some(failure);
cause
})?;
self.failed_restore_resources
.lock()
@@ -2728,7 +2845,8 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate {
Some(source.kind()),
),
};
let mut failure = RuntimeRestoreFailure::started(stage, cause);
let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, &error);
failure.cause = cause;
failure.durable_effects_possible = true;
failure.io_kind = io_kind;
failure
@@ -2870,6 +2988,7 @@ impl ProfileRuntimeWorkerFactory {
) -> Result<ProfileRestoreCandidate, RuntimeRestoreFailure> {
let mut stage = RuntimeRestoreStage::Binding;
let mut io_kind = None;
let mut primary_failure = None;
let result: Result<ProfileRestoreCandidate, &str> = async {
let feature_tasks = candidate
.retain_startup_feature_cleanup()
@@ -2998,6 +3117,7 @@ impl ProfileRuntimeWorkerFactory {
WorkerBootstrapError::Worker(error) => {
stage = RuntimeRestoreStage::Reconstruction;
self.retain_construction_failure(request, &error);
primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, &error));
restore_worker_cause(&error)
}
WorkerBootstrapError::Controller { source, .. } => {
@@ -3028,7 +3148,8 @@ impl ProfileRuntimeWorkerFactory {
}
.await;
result.map_err(|cause| {
let mut failure = RuntimeRestoreFailure::started(stage, cause);
let mut failure =
primary_failure.unwrap_or_else(|| RuntimeRestoreFailure::started(stage, cause));
failure.io_kind = io_kind;
failure
})
@@ -3744,10 +3865,16 @@ where
let factory = Arc::clone(&self.factory);
let request = request.clone();
self.run_on_adapter_runtime(async move { factory.preflight_restore(&request).await })
let outcome = self
.run_on_adapter_runtime(async move { Ok(factory.preflight_restore(&request).await) })
.map_err(|message| {
WorkerExecutionResult::rejected(WorkerExecutionOperation::Restore, message)
})?;
outcome.map_err(|failure| {
let mut result = failure.execution_result(WorkerExecutionOperation::Restore);
result.outcome = crate::execution::WorkerExecutionOutcome::Rejected;
result
})?;
Ok(())
}
@@ -5418,10 +5545,15 @@ where
let worker_ref = request.worker_ref.clone();
let spawn_result = self
.run_cancellable_on_adapter_runtime(self.spawn_restore_timeout, async move {
factory.spawn_controller(request).await
Ok(factory.spawn_controller(request).await)
});
let controller = match spawn_result {
Ok(controller) => controller,
Ok(Ok(controller)) => controller,
Ok(Err(failure)) => {
return WorkerExecutionSpawnResult::Errored(
failure.execution_result(WorkerExecutionOperation::Spawn),
);
}
Err(message) => {
return WorkerExecutionSpawnResult::Errored(
WorkerExecutionResult::errored(
@@ -5457,7 +5589,10 @@ where
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
&WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message.clone(),
),
) {
return result;
}
@@ -5479,7 +5614,10 @@ where
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
&WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message.clone(),
),
) {
return result;
}
@@ -5497,7 +5635,10 @@ where
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
&WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message.clone(),
),
) {
return result;
}
@@ -5521,7 +5662,10 @@ where
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
&WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message.clone(),
),
) {
return result;
}
@@ -5539,17 +5683,32 @@ where
let worker_ref = request.worker_ref.clone();
let spawn_result = self
.run_cancellable_on_adapter_runtime(self.spawn_restore_timeout, async move {
factory.spawn_controller(request).await
Ok(factory.spawn_controller(request).await)
});
let controller = match spawn_result {
Ok(controller) => controller,
Ok(Ok(controller)) => controller,
Ok(Err(failure)) => {
let result = failure.execution_result(WorkerExecutionOperation::Spawn);
if let Err(cleanup) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&workdir_attachments,
&rollback_workdirs,
&result,
) {
return cleanup;
}
return WorkerExecutionSpawnResult::Errored(result);
}
Err(message) => {
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&workdir_attachments,
&rollback_workdirs,
&message,
&WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message.clone(),
),
) {
return result;
}
@@ -5663,8 +5822,7 @@ where
if *completed_id != operation_id {
return None;
}
let result =
WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, failure.to_string());
let result = failure.execution_result(WorkerExecutionOperation::Restore);
Some(if failure.cleanup == RuntimeRestoreCleanup::NotNeeded {
WorkerExecutionSpawnResult::Rejected(result)
} else {
@@ -5935,7 +6093,7 @@ where
};
if let Err(message) = self.wait_pending_restore(&pending, self.spawn_restore_timeout) {
return WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message),
result: pending.failure_result(message),
worker_state: None,
workdir_attachments: Vec::new(),
};
@@ -5946,10 +6104,7 @@ where
.and_then(|()| self.wait_pending_restore(&pending, self.spawn_restore_timeout))
{
return WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
message,
),
result: pending.failure_result(message),
worker_state: None,
workdir_attachments: Vec::new(),
};
@@ -5962,10 +6117,15 @@ where
.as_ref()
{
return WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
failure.to_string(),
),
result: {
let mut result = pending.failure_result(failure.to_string());
result.push_secondary_failure(
protocol::startup_failure::StartupSecondaryFailure::Cleanup {
failure: (*failure).into(),
},
);
result
},
worker_state: None,
workdir_attachments: pending.request.previous_workdir_attachments.clone(),
};
@@ -5982,13 +6142,15 @@ where
workdir_attachments: pending.request.previous_workdir_attachments.clone(),
};
}
let message = match activated.as_ref() {
Some(Err(failure)) => failure.to_string(),
_ => "restore activation is owned by the same operation; reconcile activation"
.to_string(),
let result = match activated.as_ref() {
Some(Err(failure)) => failure.execution_result(WorkerExecutionOperation::Restore),
_ => WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
"restore activation is owned by the same operation; reconcile activation",
),
};
return WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message),
result,
worker_state: None,
workdir_attachments: pending.request.previous_workdir_attachments.clone(),
};
@@ -6010,10 +6172,7 @@ where
match retained.as_ref() {
Some(Ok(candidate)) => candidate.initial_state(),
Some(Err(failure)) => {
let result = WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
failure.to_string(),
);
let result = failure.execution_result(WorkerExecutionOperation::Restore);
let cleanup = failure.cleanup;
if failure.durable_effects_possible {
return WorkerExecutionSpawnResult::ReconciliationRequired {
@@ -6224,10 +6383,7 @@ where
.insert(worker_ref.clone(), (operation_id, failure.clone()));
self.remove_pending_restore(worker_ref)?;
Ok(WorkerExecutionSpawnResult::RolledBack(
WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
failure.to_string(),
),
failure.execution_result(WorkerExecutionOperation::Restore),
))
})();
reclaimed.unwrap_or_else(
@@ -6360,10 +6516,7 @@ where
.insert(worker_ref.clone(), (operation_id, failure.clone()));
self.remove_pending_restore(worker_ref)?;
Ok(WorkerExecutionSpawnResult::RolledBack(
WorkerExecutionResult::errored(
WorkerExecutionOperation::Restore,
failure.to_string(),
),
failure.execution_result(WorkerExecutionOperation::Restore),
))
})();
discarded.unwrap_or_else(
@@ -8876,7 +9029,11 @@ mod tests {
let before = persisted_files(&runtime_store);
let error = factory.preflight_restore(&request).await.unwrap_err();
assert!(
error.contains("Pending Workspace Worker restore requires"),
error
.legacy_diagnostic
.as_deref()
.unwrap_or_default()
.contains("Pending Workspace Worker restore requires"),
"{error}"
);
assert_eq!(persisted_files(&runtime_store), before);
@@ -8936,7 +9093,14 @@ mod tests {
(missing_instruction, "invalid pending Worker launch Prompt"),
] {
let error = factory.preflight_restore(&invalid).await.unwrap_err();
assert!(error.contains(expected), "{error}");
assert!(
error
.legacy_diagnostic
.as_deref()
.unwrap_or_default()
.contains(expected),
"{error}"
);
let error = match factory.restore_controller(invalid).await {
Ok(controller) => {
shutdown_profile_controller(controller).await;
@@ -9075,7 +9239,11 @@ mod tests {
.await
.unwrap_err();
assert!(
error.contains("subjektiv settings snapshot mismatch"),
error
.legacy_diagnostic
.as_deref()
.unwrap_or_default()
.contains("subjektiv settings snapshot mismatch"),
"{error}"
);
let error = match factory.restore_controller(wrong_settings).await {
@@ -10053,7 +10221,7 @@ mod tests {
async fn spawn_controller(
&self,
_request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
let _drop = FactoryFutureDrop(self.dropped.clone());
self.started.send(()).unwrap();
std::future::pending().await
@@ -10233,7 +10401,7 @@ mod tests {
async fn spawn_controller(
&self,
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
let manifest = WorkerManifest::from_toml(
r#"
[worker]
@@ -10664,8 +10832,8 @@ mod tests {
async fn spawn_controller(
&self,
_request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
Err("spawn failed".to_string())
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
Err("spawn failed".into())
}
async fn restore_controller(
@@ -10701,7 +10869,7 @@ mod tests {
async fn spawn_controller(
&self,
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
self.inner.spawn_controller(request).await
}
@@ -11021,7 +11189,7 @@ mod tests {
async fn spawn_controller(
&self,
_request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
panic!("stop must not spawn a Controller")
}
async fn restore_controller(
@@ -11077,11 +11245,14 @@ mod tests {
async fn spawn_controller(
&self,
_: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
panic!("reclaim must not spawn")
}
async fn preflight_restore(&self, _: &WorkerExecutionRestoreRequest) -> Result<(), String> {
async fn preflight_restore(
&self,
_: &WorkerExecutionRestoreRequest,
) -> Result<(), RuntimeRestoreFailure> {
panic!("reclaim must not preflight preparation")
}
@@ -11461,7 +11632,7 @@ mod tests {
async fn spawn_controller(
&self,
request: WorkerExecutionSpawnRequest,
) -> Result<RuntimeWorkerController, String> {
) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
self.inner.spawn_controller(request).await
}
async fn restore_controller(
@@ -0,0 +1,103 @@
//! Conversion preserves owner-selected facts. The string-only legacy seams carry
//! no domain verdict; they retain their diagnostic privately and receive a coarse
//! Runtime fallback, never a cause guessed from that diagnostic.
use super::*;
impl From<RuntimeRestoreStage> for protocol::startup_failure::StartupStage {
fn from(stage: RuntimeRestoreStage) -> Self {
match stage {
RuntimeRestoreStage::Preflight => Self::Preflight,
RuntimeRestoreStage::Metadata => Self::Metadata,
RuntimeRestoreStage::Reconstruction => Self::Reconstruction,
RuntimeRestoreStage::Recovery => Self::Recovery,
RuntimeRestoreStage::Binding => Self::Binding,
RuntimeRestoreStage::Feature => Self::Feature,
RuntimeRestoreStage::ActivationEffects => Self::ActivationEffects,
RuntimeRestoreStage::Controller => Self::Controller,
}
}
}
impl From<RuntimeRestoreCleanupFailure> for protocol::startup_failure::StartupCleanupFailure {
fn from(failure: RuntimeRestoreCleanupFailure) -> Self {
match failure {
RuntimeRestoreCleanupFailure::Ownership => Self::Ownership,
RuntimeRestoreCleanupFailure::Workdir => Self::Workdir,
RuntimeRestoreCleanupFailure::Artifacts => Self::Artifacts,
RuntimeRestoreCleanupFailure::Allocation => Self::Allocation,
RuntimeRestoreCleanupFailure::Feature => Self::Feature,
RuntimeRestoreCleanupFailure::Unsupported => Self::Unsupported,
}
}
}
impl RuntimeRestoreFailure {
pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: &WorkerBootstrapError) -> Self {
match error {
WorkerBootstrapError::Worker(source) => Self::from_worker(stage, source),
WorkerBootstrapError::Controller { source, .. } => {
let mut failure = Self::started(
RuntimeRestoreStage::Controller,
"Worker controller startup failed",
);
failure.io_kind = Some(source.kind());
failure
}
}
}
}
impl From<session_store::WorkerStoreError> for RuntimeRestoreFailure {
fn from(error: session_store::WorkerStoreError) -> Self {
Self::from_worker(
RuntimeRestoreStage::Metadata,
&WorkerError::WorkerStore(error),
)
}
}
impl From<String> for RuntimeRestoreFailure {
fn from(diagnostic: String) -> Self {
let mut failure =
Self::unproven(RuntimeRestoreStage::Reconstruction, "Worker factory failed");
failure.legacy_diagnostic = Some(diagnostic);
failure
}
}
impl From<&'static str> for RuntimeRestoreFailure {
fn from(diagnostic: &'static str) -> Self {
diagnostic.to_string().into()
}
}
impl PendingRuntimeRestore {
/// Adapter waiting/cleanup errors supplement, rather than stringify and replace,
/// any primary result already received from this exact operation's owner.
pub(super) fn failure_result(&self, message: String) -> WorkerExecutionResult {
let failure = self
.result
.lock()
.ok()
.and_then(|saved| {
saved
.as_ref()
.and_then(|result| result.as_ref().err())
.cloned()
})
.or_else(|| {
self.activated.lock().ok().and_then(|saved| {
saved
.as_ref()
.and_then(|result| result.as_ref().err())
.cloned()
})
});
let mut result = failure
.map(|failure| failure.execution_result(WorkerExecutionOperation::Restore))
.unwrap_or_else(|| {
WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message.clone())
});
result.message = Some(message);
result
}
}
+1
View File
@@ -34,6 +34,7 @@ mod subjektiv_metadata;
pub use subject_release::SubjectReleaseReservation;
pub mod skill;
pub mod spawn;
pub mod startup_failure;
pub mod subjektiv;
pub mod wip;
+352
View File
@@ -0,0 +1,352 @@
//! Typed startup diagnostics across Worker ownership wrappers. Domain causes are
//! projected at their origin; fallback Runtime facts are selected from variants,
//! never inferred from Display or legacy attribution diagnostics.
use crate::WorkerError;
use error_contract::{FailureFacts, InternalTransfer, OperationContext, project_failure};
use protocol::startup_failure::{
RuntimeStartupCause, RuntimeStartupReason, StartupFailureAttachment, StartupSecondaryFailure,
StartupStage,
};
impl WorkerError {
pub fn with_startup_failure(self, failure: StartupFailureAttachment) -> Self {
Self::StartupFailure {
source: Box::new(self),
failure,
secondary: Vec::new(),
metadata_save: None,
}
}
pub fn startup_failure(&self) -> Option<&StartupFailureAttachment> {
match self {
Self::StartupFailure { failure, .. } => Some(failure),
Self::ConstructionCleanup { source, .. }
| Self::RestoreActivationEffects { source } => source.startup_failure(),
_ => None,
}
}
pub fn startup_secondary_failures(&self) -> &[StartupSecondaryFailure] {
match self {
Self::StartupFailure { secondary, .. } => secondary,
Self::ConstructionCleanup { source, .. }
| Self::RestoreActivationEffects { source } => source.startup_secondary_failures(),
_ => &[],
}
}
pub fn startup_metadata_save(
&self,
) -> Option<protocol::startup_failure::StartupPersistenceOutcome> {
match self {
Self::StartupFailure { metadata_save, .. } => *metadata_save,
Self::ConstructionCleanup { source, .. }
| Self::RestoreActivationEffects { source } => source.startup_metadata_save(),
_ => None,
}
}
/// Preserve the original attachment when an outer resource owner adds context.
pub fn startup_failure_at(
&self,
stage: StartupStage,
) -> Result<StartupFailureAttachment, error_contract::ContractError> {
if let Some(failure) = self.startup_failure() {
return Ok(failure.clone());
}
let reason = match self {
Self::ConstructionCleanup { source, .. }
| Self::RestoreActivationEffects { source } => return source.startup_failure_at(stage),
Self::SegmentEmpty { .. } => RuntimeStartupReason::SegmentEmpty,
Self::SegmentSystemPromptMissing { .. } => RuntimeStartupReason::SystemPromptMissing,
Self::WorkerMetadataPending { .. } => RuntimeStartupReason::MetadataPending,
Self::WorkerMetadataMissing { .. } => RuntimeStartupReason::MetadataMissing,
Self::WorkerMetadataInactive { .. } => RuntimeStartupReason::MetadataInactive,
Self::WorkerMetadataManifestSnapshotMissing { .. } => {
RuntimeStartupReason::ManifestMissing
}
Self::ScopeLock(_) => RuntimeStartupReason::AllocationFailed,
Self::Store(_) => RuntimeStartupReason::SessionReadFailed,
Self::WorkerStore(_) => RuntimeStartupReason::MetadataReadFailed,
Self::Provider(_) => RuntimeStartupReason::ProviderFailed,
Self::SubjektivSessionAttribution { .. } => RuntimeStartupReason::AttributionFailed,
Self::FeatureInstall(_) | Self::FeatureLifecycle(_) => {
RuntimeStartupReason::FeatureFailed
}
_ => RuntimeStartupReason::StartupFailed,
};
runtime_startup_attachment(RuntimeStartupCause { stage, reason })
}
}
/// The local fallback accepts only enums, so provider strings and internal paths
/// cannot accidentally enter storage or public cause details.
pub fn runtime_startup_attachment(
cause: RuntimeStartupCause,
) -> Result<StartupFailureAttachment, error_contract::ContractError> {
let mut facts = FailureFacts::new(cause)?;
facts.with_context(OperationContext::new("worker_startup", "runtime")?)?;
let projected = project_failure::<InternalTransfer, _>(&facts, &())?;
StartupFailureAttachment::from_projected(&projected)
}
/// Bind a caller-owned cause to the exact pre-observed Session/attribution. Saving
/// never replaces the original error, changes ownership, or issues retry authority.
pub fn persist_startup_failure<S, D>(
store: &S,
expected: &session_store::worker_metadata::StartupFailureExpectation,
source: WorkerError,
stage: StartupStage,
authorization: &D::Authorization,
) -> WorkerError
where
S: session_store::WorkerMetadataStore,
D: error_contract::WireDetails + error_contract::ProjectFor<error_contract::Storage>,
{
use protocol::startup_failure::{StartupFailurePhase, StartupPersistenceOutcome};
use session_store::worker_metadata::StartupFailureSaveResult;
let failure = source
.startup_failure_at(stage)
.expect("fixed Worker startup projection fits the error-contract budget");
let storage = failure
.project_for::<error_contract::Storage, D>(authorization)
.and_then(|projected| {
protocol::startup_failure::StoredStartupFailureAttachment::from_projected(&projected)
});
let outcome = match storage
.as_ref()
.map(|stored| store.compare_and_swap_startup_failure(expected, stored))
{
Ok(StartupFailureSaveResult::Persisted) => StartupPersistenceOutcome::Persisted,
Ok(StartupFailureSaveResult::Conflict) => StartupPersistenceOutcome::Conflict,
Ok(StartupFailureSaveResult::NotSaved(_)) => StartupPersistenceOutcome::NotSaved,
Ok(StartupFailureSaveResult::OutcomeUnknown(_)) => {
StartupPersistenceOutcome::OutcomeUnknown
}
Err(_) => StartupPersistenceOutcome::NotSaved,
};
let mut secondary = source.startup_secondary_failures().to_vec();
if storage.is_err() {
secondary.push(StartupSecondaryFailure::ProjectionUnavailable {
phase: StartupFailurePhase::MetadataSave,
});
}
if outcome != StartupPersistenceOutcome::Persisted {
secondary.push(StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::MetadataSave,
outcome,
});
}
WorkerError::StartupFailure {
source: Box::new(source),
failure,
secondary,
metadata_save: Some(outcome),
}
}
#[cfg(test)]
mod tests {
//! Connect Worker error retention, projection and metadata CAS to the real filesystem store.
//! One wrapper deliberately loses the save response after the real CAS commits; it does not
//! replace persistence or claim a real OS crash. Run in the normal worker lib test target.
use super::*;
use protocol::startup_failure::{
StartupFailurePhase, StartupPersistenceOutcome, StoredStartupFailureAttachment,
};
use session_store::worker_metadata::{
StartupFailureExpectation, StartupFailureRead, StartupFailureSaveResult,
};
use session_store::{
FsWorkerStore, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore,
WorkerStoreError,
};
struct LostSaveResponse(FsWorkerStore);
impl WorkerMetadataStore for LostSaveResponse {
fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> {
self.0.write(metadata)
}
fn read_by_name(&self, name: &str) -> Result<Option<WorkerMetadata>, WorkerStoreError> {
self.0.read_by_name(name)
}
fn list_names(&self) -> Result<Vec<String>, WorkerStoreError> {
self.0.list_names()
}
fn delete_by_name(&self, name: &str) -> Result<(), WorkerStoreError> {
self.0.delete_by_name(name)
}
fn compare_and_swap_startup_failure(
&self,
expected: &StartupFailureExpectation,
failure: &StoredStartupFailureAttachment,
) -> StartupFailureSaveResult {
match self.0.compare_and_swap_startup_failure(expected, failure) {
StartupFailureSaveResult::Persisted => StartupFailureSaveResult::OutcomeUnknown(
WorkerStoreError::Io(std::io::Error::other("TEST_ONLY lost response")),
),
other => other,
}
}
}
fn metadata() -> WorkerMetadata {
WorkerMetadata::new(
"startup",
Some(WorkerActiveSegmentRef::pending_segment(
session_store::new_session_id(),
)),
)
.with_workspace_id("workspace")
}
fn error() -> WorkerError {
WorkerError::FeatureInstall(
"TEST_ONLY_CREDENTIAL /private/internal/path raw_http_body".into(),
)
}
#[test]
fn startup_error_save_reopens_same_cause_without_replacing_primary_or_disclosing_source() {
let dir = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(dir.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let source = error();
let primary = source.startup_failure_at(StartupStage::Feature).unwrap();
let retained = persist_startup_failure::<_, RuntimeStartupCause>(
&store,
&expected,
source,
StartupStage::Feature,
&(),
);
assert_eq!(retained.startup_failure(), Some(&primary));
assert_eq!(
retained.startup_metadata_save(),
Some(StartupPersistenceOutcome::Persisted)
);
let reopened = FsWorkerStore::new(dir.path())
.unwrap()
.read_by_name("startup")
.unwrap()
.unwrap();
assert!(matches!(
reopened.decode_startup_failure::<RuntimeStartupCause>(),
StartupFailureRead::Present(error_contract::DecodeFailure::Known(_))
));
for secret in [
"TEST_ONLY_CREDENTIAL",
"/private/internal/path",
"raw_http_body",
] {
assert!(
!serde_json::to_string(&reopened.startup_failure)
.unwrap()
.contains(secret)
);
}
assert_eq!(reopened.active, original.active);
assert_eq!(
reopened.subjektiv_session_attribution,
original.subjektiv_session_attribution
);
}
#[test]
fn startup_error_unknown_save_keeps_primary_and_requires_reconciliation_of_real_commit() {
let dir = tempfile::tempdir().unwrap();
let store = LostSaveResponse(FsWorkerStore::new(dir.path()).unwrap());
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
let source = error();
let primary = source.startup_failure_at(StartupStage::Feature).unwrap();
let retained = persist_startup_failure::<_, RuntimeStartupCause>(
&store,
&expected,
source,
StartupStage::Feature,
&(),
);
assert_eq!(retained.startup_failure(), Some(&primary));
assert_eq!(
retained.startup_metadata_save(),
Some(StartupPersistenceOutcome::OutcomeUnknown)
);
assert!(retained.startup_secondary_failures().contains(
&StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::MetadataSave,
outcome: StartupPersistenceOutcome::OutcomeUnknown,
}
));
let reopened = FsWorkerStore::new(dir.path())
.unwrap()
.read_by_name("startup")
.unwrap()
.unwrap();
assert!(matches!(
reopened.decode_startup_failure::<RuntimeStartupCause>(),
StartupFailureRead::Present(error_contract::DecodeFailure::Known(_))
));
// The original observation must not be retried as a fresh write after an unknown response.
assert!(matches!(
store.compare_and_swap_startup_failure(
&expected,
&StoredStartupFailureAttachment::from_projected(
&primary
.project_for::<error_contract::Storage, RuntimeStartupCause>(&())
.unwrap()
)
.unwrap()
),
StartupFailureSaveResult::Conflict
));
}
#[test]
fn startup_error_save_failure_and_conflict_preserve_original_primary() {
for conflict in [true, false] {
let dir = tempfile::tempdir().unwrap();
let store = FsWorkerStore::new(dir.path()).unwrap();
let original = metadata();
store.write(&original).unwrap();
let expected = StartupFailureExpectation::from_metadata(&original).unwrap();
if conflict {
store
.set_active(
"startup",
Some(WorkerActiveSegmentRef::pending_segment(
session_store::new_session_id(),
)),
None,
)
.unwrap();
} else {
let path = dir.path().join("startup/metadata.json");
std::fs::remove_file(&path).unwrap();
std::fs::create_dir(&path).unwrap(); // Real read failure before a save can be attempted.
}
let source = error();
let primary = source.startup_failure_at(StartupStage::Feature).unwrap();
let retained = persist_startup_failure::<_, RuntimeStartupCause>(
&store,
&expected,
source,
StartupStage::Feature,
&(),
);
assert_eq!(retained.startup_failure(), Some(&primary));
assert_eq!(
retained.startup_metadata_save(),
Some(if conflict {
StartupPersistenceOutcome::Conflict
} else {
StartupPersistenceOutcome::NotSaved
})
);
}
}
}
+10
View File
@@ -11457,6 +11457,16 @@ fn compact_failure_category(error: &WorkerError) -> CompactFailureCategory {
#[derive(Debug, thiserror::Error)]
pub enum WorkerError {
/// The domain-projected primary error survives wrappers and metadata-save failures.
/// This attachment is not execution cleanup or lifecycle evidence.
#[error("{source}")]
StartupFailure {
#[source]
source: Box<WorkerError>,
failure: protocol::startup_failure::StartupFailureAttachment,
secondary: Vec<protocol::startup_failure::StartupSecondaryFailure>,
metadata_save: Option<protocol::startup_failure::StartupPersistenceOutcome>,
},
/// A construction or activation step failed after acquiring resources. Retain
/// these handles until explicit cleanup succeeds. Consuming constructors drop
/// the Worker before returning; in-place activation leaves its owner responsible
@@ -66,7 +66,10 @@ impl RuntimeWorkerFactory for CapturingFactory {
async fn spawn_controller(
&self,
mut request: worker_runtime::execution::WorkerExecutionSpawnRequest,
) -> std::result::Result<RuntimeWorkerController, String> {
) -> std::result::Result<
RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let archive = worker_runtime::profile_archive::ProfileSourceArchive::build(
worker_runtime::profile_archive::ProfileSourceArchiveInput {
id: "profile-source-archive:restored-parent".into(),
@@ -93,7 +96,7 @@ impl RuntimeWorkerFactory for CapturingFactory {
async fn preflight_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), String> {
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> {
self.inner.preflight_restore(request).await
}
async fn recover_restore_source(
@@ -1002,7 +1002,10 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for RestoreAuthFactory
async fn spawn_controller(
&self,
mut request: worker_runtime::execution::WorkerExecutionSpawnRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
if request.request.backend_job.is_some() {
return self.inner.spawn_controller(request).await;
}
@@ -1040,7 +1043,7 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for RestoreAuthFactory
async fn preflight_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), String> {
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> {
self.inner.preflight_restore(request).await
}
@@ -730,9 +730,13 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for AttentionWorkerFac
async fn spawn_controller(
&self,
request: worker_runtime::execution::WorkerExecutionSpawnRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
self.controller(request.worker_ref.worker_id.to_string(), false)
.await
.map_err(Into::into)
}
async fn prepare_restore(
&self,
@@ -858,14 +858,19 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for DeliveryWorkerFact
async fn spawn_controller(
&self,
request: delivery_execution::WorkerExecutionSpawnRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
let worker_id = request.worker_ref.worker_id.to_string();
let manifest = Self::manifest(
&worker_id,
&request.request.profile,
request.config_bundle.as_ref(),
)?;
self.controller(worker_id, false, manifest).await
self.controller(worker_id, false, manifest)
.await
.map_err(Into::into)
}
async fn prepare_restore(
&self,
@@ -494,7 +494,9 @@ async fn workspace_config_apply_model_union_cases_and_composite_persist_evaluate
// Subsequent calls use the Client's returned root validator, without reinspection or
// model-supplied preconditions. All four variants share the same public contract.
fixture
.apply_observed(json!([{"$case":"update","value":{"path":"notes/edit.txt","content":"updated"}}]))
.apply_observed(
json!([{"$case":"update","value":{"path":"notes/edit.txt","content":"updated"}}]),
)
.await;
fixture
.apply_observed(
@@ -541,7 +543,8 @@ async fn workspace_config_apply_model_union_cases_and_composite_persist_evaluate
);
assert_eq!(
fragment.content,
"{\n common = {\n language = \"TYPED_BATCH\";\n };\n}\n" );
"{\n common = {\n language = \"TYPED_BATCH\";\n };\n}\n"
);
assert_eq!(
state
.snapshot
@@ -804,10 +807,7 @@ async fn workspace_config_apply_cached_contract_cannot_bypass_readonly_or_revoca
let error = config_call(&fixture.wip, "/workspace-config", "apply_changes", json!({"changes":[
{"$case":"create","value":{"path":"notes/denied.txt","content":"denied","content_type":"text"}}
]})).await.unwrap_err();
assert!(
!error.to_string().contains("outcome unknown"),
"{error}"
);
assert!(!error.to_string().contains("outcome unknown"), "{error}");
assert_eq!(fixture.state(), before);
assert_eq!(fixture.commits(), commits);
if !revoke {
@@ -845,10 +845,7 @@ async fn workspace_config_apply_known_rejection_differs_from_lost_response_witho
let error = config_call(&fixture.wip, "/workspace-config", "apply_changes", json!({"changes":[
{"$case":"create","value":{"path":"notes/unknown.txt","content":"committed once","content_type":"text"}}
]})).await.unwrap_err();
assert!(
error.to_string().contains("outcome unknown"),
"{error}"
);
assert!(error.to_string().contains("outcome unknown"), "{error}");
let state = fixture.state();
assert_eq!(
state
@@ -1165,7 +1162,10 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for ConfigCaptureFacto
async fn spawn_controller(
&self,
mut request: worker_runtime::execution::WorkerExecutionSpawnRequest,
) -> std::result::Result<worker_runtime::worker_backend::RuntimeWorkerController, String> {
) -> std::result::Result<
worker_runtime::worker_backend::RuntimeWorkerController,
worker_runtime::worker_backend::RuntimeRestoreFailure,
> {
// No LLM request is issued; explicit unauthenticated test model avoids
// depending on developer credentials during production bootstrap.
let archive = worker_runtime::profile_archive::ProfileSourceArchive::build(worker_runtime::profile_archive::ProfileSourceArchiveInput {
@@ -1188,7 +1188,7 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for ConfigCaptureFacto
async fn preflight_restore(
&self,
request: &worker_runtime::execution::WorkerExecutionRestoreRequest,
) -> std::result::Result<(), String> {
) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> {
self.inner.preflight_restore(request).await
}
@@ -2,6 +2,32 @@
// Do not edit by hand. Rust DTO authority lives in `crates/protocol`.
// Large integer fields are JSON numbers on the wire and are emitted as TypeScript `number`.
export type StartupFailureAttachment = string;
export type StoredStartupFailureAttachment = StartupFailureAttachment;
export type StartupStage = "preflight" | "metadata" | "reconstruction" | "recovery" | "binding" | "feature" | "activation_effects" | "controller";
export type RuntimeStartupReason = "startup_failed" | "segment_empty" | "system_prompt_missing" | "metadata_pending" | "metadata_missing" | "metadata_inactive" | "manifest_missing" | "allocation_failed" | "session_read_failed" | "metadata_read_failed" | "provider_failed" | "attribution_failed" | "feature_failed";
export type RuntimeStartupCause = { stage: StartupStage, reason: RuntimeStartupReason, };
export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "metadata_save";
export type StartupCleanupFailure = "ownership" | "workdir" | "artifacts" | "allocation" | "feature" | "unsupported";
export type StartupPersistenceOutcome = "persisted" | "not_saved" | "conflict" | "outcome_unknown";
export type StartupSecondaryFailure = { "kind": "projection_unavailable", phase: StartupFailurePhase, } | { "kind": "cleanup", failure: StartupCleanupFailure, } | { "kind": "persistence", phase: StartupFailurePhase, outcome: StartupPersistenceOutcome, };
export type StartupResources = "none" | "worker_startup" | "unproven";
export type StartupCleanup = "not_needed" | "required" | "confirmed" | { "failed": StartupCleanupFailure };
export type StartupResourceFacts = { resources: StartupResources, cleanup: StartupCleanup, durable_effects_possible: boolean, };
export type StartupIoKind = "not_found" | "permission_denied" | "already_exists" | "invalid_input" | "invalid_data" | "interrupted" | "timed_out" | "would_block" | "unexpected_eof" | "other";
export type AlertLevel = "warn" | "error";
export type AlertPosition = { "kind": "start" } | { "kind": "after_entry", entry_id: string, } | { "kind": "after_tool_call", call_id: string, } | { "kind": "after_in_flight", block_id: number, };