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

fix: retain local Session binding causes and factory sources safely

Browse Source
This commit is contained in:
Hare
2026-11-04 19:06:41 +09:00
parent 1d8c205b2e
commit 9fadd1bcaf
9 changed files with 462 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -5040,6 +5040,7 @@ dependencies = [
"agen",
"async-trait",
"client",
"error-contract",
"feature-storage",
"fs4",
"futures",
crates/standalone/Cargo.toml
+1
View File
@@ -28,6 +28,7 @@ uuid = { workspace = true, features = ["v7"] }
worker.workspace = true
[dev-dependencies]
error-contract.workspace = true
async-trait.workspace = true
futures.workspace = true
tempfile.workspace = true
crates/standalone/src/subjektiv.rs
+31 -5
View File
@@ -275,6 +275,8 @@ impl StandaloneSubjects {
writer,
self_weak: weak.clone(),
journal_dispatch: Mutex::new(false),
#[cfg(test)]
registration_before_attribution: Mutex::new(None),
language: self.memory_language.clone(),
catalog: self,
binding: binding.clone(),
@@ -373,6 +375,8 @@ pub(crate) struct LocalSubjectHost {
writer: subjektiv::journal::JournalWriterBinding,
self_weak: std::sync::Weak<Self>,
journal_dispatch: Mutex<bool>,
#[cfg(test)]
registration_before_attribution: Mutex<Option<Box<dyn FnOnce() + Send>>>,
catalog: StandaloneSubjects,
binding: StandaloneSubjectBinding,
worker_id: protocol::WorkerId,
@@ -467,7 +471,11 @@ impl LocalSubjectHost {
context: &SubjektivHostContext,
create: bool,
) -> Result<SubjectSessionAttribution, SubjektivHostError> {
self.body(context)?;
self.registration_authority(context)?;
#[cfg(test)]
if let Some(checkpoint) = self.registration_before_attribution.lock().unwrap().take() {
checkpoint();
}
let store = &self.catalog.store;
if let Some(record) = store
.session_attribution(&context.session_id)
@@ -484,7 +492,7 @@ impl LocalSubjectHost {
}
store
.bind_journal_session(&self.writer, record.clone())
.map_err(journal::journal_error)?;
.map_err(registration_journal_error)?;
return Ok(record);
}
if !create {
@@ -507,11 +515,25 @@ impl LocalSubjectHost {
.map_err(registration_domain_error)?;
store
.bind_journal_session(&self.writer, record.clone())
.map_err(journal::journal_error)?;
.map_err(registration_journal_error)?;
Ok(record)
}
}
fn registration_journal_error(error: subjektiv::journal::JournalError) -> SubjektivHostError {
let cause = match &error {
subjektiv::journal::JournalError::PermissionDenied(_) => {
SessionRegistrationCause::binding_mismatch()
}
subjektiv::journal::JournalError::Subject(
subjektiv::SubjektivError::SessionSubjectConflict { .. }
| subjektiv::SubjektivError::SessionAttributionConflict { .. },
) => SessionRegistrationCause::binding_mismatch(),
_ => SessionRegistrationCause::outcome_unknown(),
};
SubjektivHostError::session_registration(cause, error)
}
fn registration_domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError {
let cause = match &error {
subjektiv::SubjektivError::SessionSubjectConflict { .. }
@@ -868,7 +890,11 @@ impl SubjektivHost for LocalSubjectHost {
// An explicit operator connection issues this local body capability.
// Revalidate its live writer on every access/supply, not Feature enablement.
// Job decisions and Backend conversations are not local body capabilities.
self.body(context)?;
if operation == "session.record" {
self.registration_authority(context)?;
} else {
self.body(context)?;
}
match operation {
"session.record"
| "memory.resident.read"
@@ -1095,7 +1121,7 @@ impl SubjektivHost for LocalSubjectHost {
) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
self.registration_authority(context)?;
let attribution = self.attribution(context, create_if_missing)?;
self.reserve_journal_session(context)?;
self.reserve_journal_session(context, &attribution)?;
Ok(SubjektivRecordSessionResponse {
subject_id: attribution.subject_id,
session_id: attribution.session_id,
crates/standalone/src/subjektiv/journal.rs
+34 -17
View File
@@ -220,6 +220,12 @@ pub(super) fn journal_error(error: JournalError) -> SubjektivHostError {
other => SubjektivHostError::Unavailable(other.to_string()),
}
}
fn registration_reservation_error(
error: impl std::error::Error + Send + Sync + 'static,
) -> SubjektivHostError {
SubjektivHostError::session_registration(SessionRegistrationCause::outcome_unknown(), error)
}
fn source_error(error: session_store::StoreError) -> SubjektivHostError {
SubjektivHostError::Unavailable(error.to_string())
}
@@ -368,45 +374,56 @@ impl LocalSubjectHost {
pub(super) fn reserve_journal_session(
&self,
context: &SubjektivHostContext,
attribution: &SubjectSessionAttribution,
) -> Result<(), SubjektivHostError> {
self.body(context)?;
if self.journal_session_erased(&context.session_id)? {
self.registration_authority(context)?;
let journal_context =
JournalHostContext::validated_session(&self.catalog.store, attribution.clone())
.map_err(registration_journal_error)?;
if self
.catalog
.store
.journal_session_erased(&journal_context)
.map_err(registration_journal_error)?
{
return Ok(());
}
let session: session_store::SessionId = context
.session_id
.parse()
.map_err(|_| SubjektivHostError::InvalidArgument("invalid Session".into()))?;
let session: session_store::SessionId = context.session_id.parse().map_err(|error| {
SubjektivHostError::session_registration(
SessionRegistrationCause::invalid_response(),
error,
)
})?;
let root = self
.catalog
.state_root
.join(self.worker_id.to_string())
.join("sessions");
reject_symlink_components(&root)
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
reject_symlink_components(&root).map_err(registration_reservation_error)?;
let source_store = session_store::FsStore::new_with_retention(
&root,
crate::store::session_retention(&self.catalog.state_root),
)
.map_err(source_error)?;
.map_err(registration_reservation_error)?;
{
let _guard = source_store
.retention()
.lock_live(session)
.map_err(source_error)?;
.map_err(registration_reservation_error)?;
let path = root.join(session.to_string());
match fs::create_dir(&path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
let metadata = fs::symlink_metadata(&path)
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
let metadata =
fs::symlink_metadata(&path).map_err(registration_reservation_error)?;
if !metadata.is_dir() || metadata.file_type().is_symlink() {
return Err(SubjektivHostError::InvalidArgument(
"Session retention source is not an owned directory".into(),
));
return Err(registration_reservation_error(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Session retention source is not an owned directory",
)));
}
}
Err(error) => return Err(SubjektivHostError::Unavailable(error.to_string())),
Err(error) => return Err(registration_reservation_error(error)),
}
}
// Establish retention at attribution, before the first Run can publish bytes. This
@@ -416,7 +433,7 @@ impl LocalSubjectHost {
session,
session_store::retention::SessionRetentionPin(session),
)
.map_err(source_error)
.map_err(registration_reservation_error)
}
pub(super) fn admit_journal(
crates/standalone/src/subjektiv/tests.rs
+139
View File
@@ -324,3 +324,142 @@ fn session_registration_uses_real_capability_and_attribution_decisions() {
SessionRegistrationCause::denied(SessionPrincipal::Subject),
);
}
#[derive(Clone)]
struct RegistrationUnusedModel;
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for RegistrationUnusedModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
_: agen::llm_client::Request,
) -> Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError> {
panic!("Session registration must not invoke a model")
}
}
#[tokio::test]
async fn local_writer_revocation_during_registration_retains_known_cause_and_real_journal_source() {
use session_store::{CombinedStore, FsStore, FsWorkerStore, WorkerMetadataStore};
use worker::{
WorkerBootstrap, WorkerBootstrapLayout, WorkerFilesystemAuthority, WorkerWorkspaceContext,
};
// The direct non-Browser caller uses the real bootstrap, local capability and
// transactional Journal store. Only the unused model and revocation timing are substituted.
for existing in [false, true] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("state");
let catalog = StandaloneSubjects::open(&root).unwrap();
let subject = catalog.create("registration-race", None).unwrap();
let id = protocol::WorkerId::now_v7();
let mut connection = catalog.connect(&subject.id, id).unwrap();
let port = connection.host.clone();
let name = format!("standalone-{id}");
let sessions = FsStore::new_with_retention(
root.join(id.to_string()).join("sessions"),
crate::store::session_retention(&root),
)
.unwrap();
let metadata_root = temp.path().join("metadata");
let metadata = FsWorkerStore::new(&metadata_root).unwrap();
let mut manifest = manifest::WorkerManifest::from_toml(&format!("[worker]\nname='{name}'\n[model]\nscheme='anthropic'\nmodel_id='unused'\n[engine]\n[scope]\nallow=[]")).unwrap();
manifest.feature.subjektiv.profile.enabled = true;
manifest.feature.subjektiv.profile.extraction.enabled = false;
let mut prepared = WorkerBootstrap::new(
manifest,
CombinedStore::new(sessions, metadata.clone()),
worker::PromptCatalogSource::builtins_only(),
WorkerWorkspaceContext::no_workspace(),
WorkerFilesystemAuthority::None,
WorkerBootstrapLayout::Direct {
runtime_base: temp.path().join("runtime"),
},
)
.with_model_client(RegistrationUnusedModel)
.prepare()
.await
.unwrap();
let worker = prepared.worker_mut();
worker.enable_worker_metadata_write_through().unwrap();
worker.bind_subjektiv_host(Some(port.clone()));
let session = worker.session_id();
let context = SubjektivHostContext {
worker_id: name.clone(),
session_id: session.to_string(),
};
if existing {
port.record_session(&context, true).unwrap();
}
let revoke_store = port.catalog.store.clone();
let writer = port.writer.clone();
*port.registration_before_attribution.lock().unwrap() = Some(Box::new(move || {
// This actual release follows the last preflight, before transactional bind.
revoke_store.release_journal_writer(&writer).unwrap();
}));
let error = worker
.finalize_subjektiv_session_attribution_retaining_startup_failure(
true,
worker::SubjektivSessionAttributionLifecycle::NewSession,
)
.unwrap_err();
assert_eq!(
error.session_registration_cause(),
Some(SessionRegistrationCause::binding_mismatch())
);
let mut source = Some(&error as &(dyn std::error::Error + 'static));
let mut journal_source = false;
while let Some(current) = source {
if let Some(journal) = current.downcast_ref::<subjektiv::journal::JournalError>() {
assert!(matches!(
journal,
subjektiv::journal::JournalError::PermissionDenied(_)
));
journal_source = true;
}
source = current.source();
}
assert!(
journal_source,
"original store refusal must survive the Worker wrapper"
);
assert_eq!(
error.startup_metadata_save(),
Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted)
);
drop(prepared);
drop(metadata);
let reopened = FsWorkerStore::new(metadata_root)
.unwrap()
.read_by_name(&name)
.unwrap()
.unwrap();
let session_store::worker_metadata::StartupFailureRead::Present(
error_contract::DecodeFailure::Known(cause),
) = reopened.decode_startup_failure::<SessionRegistrationCause>()
else {
panic!("typed cause must really reopen")
};
assert_eq!(
*cause.facts().cause(),
SessionRegistrationCause::binding_mismatch()
);
assert!(
port.catalog
.store
.session_attribution(&session.to_string())
.unwrap()
.is_some(),
"committed attribution is independent from denied writer binding"
);
assert!(
!port
.catalog
.store
.journal_writer_active(&port.writer)
.unwrap()
);
connection.close().unwrap();
}
}
crates/worker-runtime/src/worker_backend.rs
+47 -10
View File
@@ -257,7 +257,9 @@ enum RuntimeRestoreCleanup {
/// Diagnostics are fixed owner-selected causes, not arbitrary provider messages,
/// paths, configuration, or credentials. Cleanup cannot overwrite the first cause.
#[derive(Debug, Clone, PartialEq, Eq)]
/// The original in-process source is privately shared across clones; it is neither
/// formatted into diagnostics nor included in the wire/storage projection.
#[derive(Clone)]
pub struct RuntimeRestoreFailure {
pub stage: RuntimeRestoreStage,
pub resources: RuntimeRestoreResources,
@@ -270,6 +272,7 @@ pub struct RuntimeRestoreFailure {
secondary: Vec<protocol::startup_failure::StartupSecondaryFailure>,
legacy_diagnostic: Option<String>,
metadata_save: Option<protocol::startup_failure::StartupPersistenceOutcome>,
source: Option<Arc<dyn std::error::Error + Send + Sync>>,
}
impl RuntimeRestoreFailure {
@@ -301,7 +304,13 @@ impl RuntimeRestoreFailure {
self.io_kind
}
fn from_worker(stage: RuntimeRestoreStage, error: &WorkerError) -> Self {
fn from_worker(stage: RuntimeRestoreStage, error: WorkerError) -> Self {
let mut failure = Self::from_worker_ref(stage, &error);
failure.source = Some(Arc::new(error));
failure
}
fn from_worker_ref(stage: RuntimeRestoreStage, error: &WorkerError) -> Self {
let mut failure = Self::started(stage, restore_worker_cause(error));
failure.primary = error
.startup_failure_at(stage.into())
@@ -354,6 +363,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -370,6 +380,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -386,6 +397,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -397,6 +409,25 @@ impl RuntimeRestoreFailure {
}
}
impl std::fmt::Debug for RuntimeRestoreFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RuntimeRestoreFailure")
.field("stage", &self.stage)
.field("resources", &self.resources)
.field("cause", &self.cause)
.field("cleanup", &self.cleanup)
.field("io_kind", &self.io_kind)
.field("cleanup_failure", &self.cleanup_failure)
.finish_non_exhaustive()
}
}
impl std::error::Error for RuntimeRestoreFailure {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.source.as_deref().map(|source| source as _)
}
}
impl std::fmt::Display for RuntimeRestoreFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let cause: String = self
@@ -2225,7 +2256,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
None => bootstrap,
};
let mut prepared = bootstrap.prepare().await.map_err(|error| {
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error)
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, error)
})?;
let worker = prepared.worker_mut();
if let Some(snapshot) = &request.request.plugin_execution {
@@ -2299,7 +2330,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let workspace_client = worker.workspace_client_handle();
let started = prepared.start().await.map_err(|error| {
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error)
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, error)
})?;
let (handle, shutdown_rx, controller_task) =
(started.handle, started.shutdown, started.controller_task);
@@ -2542,7 +2573,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
.await
.map_err(|error| {
self.retain_construction_failure(&request, &error);
let failure = RuntimeRestoreFailure::from_worker(stage, &error);
let failure = RuntimeRestoreFailure::from_worker(stage, error);
let cause = failure.cause;
preparation_failure = Some(failure);
cause
@@ -2818,7 +2849,7 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate {
Some(source.kind()),
),
};
let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, &error);
let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, error);
failure.cause = cause;
failure.durable_effects_possible = true;
failure.io_kind = io_kind;
@@ -3090,11 +3121,17 @@ 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)
let cause = restore_worker_cause(&error);
primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, error));
cause
}
WorkerBootstrapError::Controller { source, .. } => {
io_kind = Some(source.kind());
error @ WorkerBootstrapError::Controller { .. } => {
let failure = RuntimeRestoreFailure::from_bootstrap(
RuntimeRestoreStage::Controller,
error,
);
io_kind = failure.io_kind;
primary_failure = Some(failure);
"restored Worker preparation failed"
}
})?;
crates/worker-runtime/src/worker_backend/startup_binding_tests.rs
+85 -1
View File
@@ -1,7 +1,8 @@
mod startup_binding_tests {
//! Exercise the factory's production binding step with a constructed Worker,
//! real aggregate/Session stores and reopen. Only the Workspace response is
//! synthetic; this suite does not test domain authorization or network transport.
//! synthetic in the binding cases; the direct internal Profile factory case uses
//! a refused real loopback transport. Domain authorization is tested by Server.
use super::*;
use protocol::session_failure::{
SessionPrincipal, SessionRegistrationCause, session_registration_attachment,
@@ -154,6 +155,21 @@ mod startup_binding_tests {
let failure = startup_failure::finalize_startup_binding(&worker, true)
.err()
.expect("failure must remain a failure");
if matches!(response, Response::Lost | Response::Rejected) {
let mut source = std::error::Error::source(&failure);
let mut worker = false;
let mut host = false;
let mut client = false;
while let Some(current) = source {
worker |= current.is::<WorkerError>();
host |= current.is::<worker::subjektiv::SubjektivHostError>();
client |= current.is::<WorkspaceClientError>();
source = current.source();
}
assert!(worker && host, "common factory binding step retains in-process domain sources");
assert_eq!(client, matches!(response, Response::Lost));
assert!(!format!("{failure:?} {failure}").contains("response deliberately lost"));
}
let result = failure.execution_result(WorkerExecutionOperation::Spawn);
if matches!(response, Response::BeforeSession | Response::BeforeOwner) {
let primary = result.startup_failure::<RuntimeStartupCause>().unwrap();
@@ -260,6 +276,74 @@ mod startup_binding_tests {
}
}
}
#[cfg(feature = "http-server")]
#[tokio::test]
#[serial_test::serial(worker_allocation)]
async fn internal_profile_factory_transport_failure_retains_source_chain_and_reopened_cause() {
require_restore_test_environment();
let root = tempfile::tempdir().unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let endpoint = format!("http://{}", listener.local_addr().unwrap());
drop(listener);
let worker_ref = WorkerRef::new(WorkerId::now_v7());
let mut request = create_request("internal factory source");
request.worker_id = worker_ref.worker_id;
request.workspace_api = Some(WorkspaceApiRef { workspace_id: "adapter-test".into(), base_url: endpoint.clone() });
request.subjektiv_attached = true;
request.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { workspace_id: "adapter-test".into(), language: "English".into() });
let archive = crate::profile_archive::ProfileSourceArchive::build_evaluated_profile(
"profile-source-archive:internal-source".into(), "builtin:companion".into(),
serde_json::json!({"slug":"internal-source", "description":"internal source test", "scope":"workspace_read",
"model":{"scheme":"anthropic", "model_id":"unused", "auth":{"kind":"none"}},
"engine":{"max_tokens":100}, "feature":{"subjektiv":{"enabled":true,"extraction":{"enabled":false}}}})
).unwrap();
let mut bundle = test_bundle();
bundle.profile_source_archive = Some(archive.clone());
let bundle = bundle.with_computed_digest();
request.profile_source = crate::catalog::ProfileSourceArchiveSource::Embedded { archive };
request.config_bundle = Some(ConfigBundleRef { id: bundle.metadata.id.clone(), digest: bundle.metadata.digest.clone() });
let request = WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(), request,
workspace_scope: None,
context: test_execution_context(worker_ref.clone()),
workdir_attachments: BTreeMap::new(), config_bundle: Some(bundle),
};
// Direct internal caller, not a Browser handler or Runtime result decoder.
// The real transport is refused at an unused loopback socket; only the model is replaced.
let factory = ProfileRuntimeWorkerFactory::new(root.path())
.with_runtime_id("internal-source-runtime")
.with_runtime_store_dir(root.path().join("runtime"))
.with_spawn_model_client_factory(|| Box::new(MockClient::new(vec![])));
let failure = match factory.spawn_controller(request).await {
Err(failure) => failure,
Ok(_) => panic!("disconnected Backend must not create a Controller"),
};
assert_eq!(failure.stage, RuntimeRestoreStage::Binding);
let mut source = std::error::Error::source(&failure);
let mut worker = false; let mut host = false; let mut transport = false;
while let Some(current) = source {
worker |= current.is::<WorkerError>();
host |= current.is::<worker::subjektiv::SubjektivHostError>();
if let Some(error) = current.downcast_ref::<WorkspaceClientError>() {
assert!(matches!(error, WorkspaceClientError::Request(_)));
transport = true;
}
source = current.source();
}
assert!(worker && host && transport, "factory must retain original typed in-process source");
let result = failure.execution_result(WorkerExecutionOperation::Spawn);
let Some(error_contract::DecodeFailure::Known(cause)) = result.startup_failure::<SessionRegistrationCause>() else { panic!("Session cause must remain typed") };
assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown());
assert_eq!(result.startup_failure_persistence, Some(StartupPersistenceOutcome::Persisted));
let aggregate = root.path().join("runtime/workers").join(worker_ref.worker_id.to_string());
let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref);
let metadata = WorkerAggregateStore::new(aggregate, &name).unwrap().read_by_name(&name).unwrap().unwrap();
let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = metadata.decode_startup_failure::<SessionRegistrationCause>() else { panic!("saved factory cause must reopen") };
assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown());
assert!(!format!("{failure:?} {failure}").contains(&endpoint));
assert!(!serde_json::to_string(&result).unwrap().contains(&endpoint));
}
#[tokio::test]
async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() {
check_binding(Response::Rejected).await;
crates/worker-runtime/src/worker_backend/startup_failure.rs
+8 -6
View File
@@ -32,9 +32,9 @@ impl From<RuntimeRestoreCleanupFailure> for protocol::startup_failure::StartupCl
}
impl RuntimeRestoreFailure {
pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: &WorkerBootstrapError) -> Self {
match error {
WorkerBootstrapError::Worker(source) => Self::from_worker(stage, source),
pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: WorkerBootstrapError) -> Self {
let mut failure = match &error {
WorkerBootstrapError::Worker(source) => Self::from_worker_ref(stage, source),
WorkerBootstrapError::Controller { source, .. } => {
let mut failure = Self::started(
RuntimeRestoreStage::Controller,
@@ -43,7 +43,9 @@ impl RuntimeRestoreFailure {
failure.io_kind = Some(source.kind());
failure
}
}
};
failure.source = Some(Arc::new(error));
failure
}
}
@@ -51,7 +53,7 @@ impl From<session_store::WorkerStoreError> for RuntimeRestoreFailure {
fn from(error: session_store::WorkerStoreError) -> Self {
Self::from_worker(
RuntimeRestoreStage::Metadata,
&WorkerError::WorkerStore(error),
WorkerError::WorkerStore(error),
)
}
}
@@ -114,7 +116,7 @@ where
attached,
SubjektivSessionAttributionLifecycle::NewSession,
)
.map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error))
.map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, error))
}
#[cfg(test)]
crates/workspace-server/src/hosts.rs
+116 -3
View File
@@ -6867,7 +6867,10 @@ fn runtime_api_diagnostic(
) -> RuntimeDiagnostic {
match error {
runtime_api::client_support::ClientError::Public { status, error } => {
let startup = error.error.startup.and_then(internal_startup_transfer);
if let Some(startup) = error.error.startup {
return remote_startup_diagnostic(runtime_id, status, startup);
}
let startup = None;
let remote_code = error.error.code;
let message = sanitize_remote_runtime_message(&remote_code, &error.error.message);
let (code, severity) = match status {
@@ -6946,6 +6949,9 @@ fn remote_http_status_diagnostic(
.ok()
.filter(|_| body.len() <= MAX_RUNTIME_PING_RESPONSE_BYTES)
.and_then(|_| serde_json::from_slice::<RuntimeHttpErrorResponse>(&body).ok());
if let Some(startup) = error.as_ref().and_then(|error| error.error.startup.clone()) {
return remote_startup_diagnostic(runtime_id, status, startup);
}
let remote_code = error
.as_ref()
.map(|error| error.error.code.as_str())
@@ -6980,6 +6986,29 @@ fn remote_http_status_diagnostic(
projection
}
fn remote_startup_diagnostic(
runtime_id: &str,
status: StatusCode,
startup: runtime_api::WorkerStartupFailure,
) -> RuntimeDiagnostic {
let startup = internal_startup_transfer(startup);
let message = startup.as_ref().map_or("Worker startup failed", |startup| {
worker::startup_failure::internal_startup_failure_summary(&startup.primary)
});
let code = if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
"remote_runtime_auth_failed"
} else {
"remote_runtime_startup_failed"
};
let mut projection = diagnostic(
code,
HostDiagnosticSeverity::Error,
format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}"),
);
projection.startup = startup;
projection
}
/// Treat transferred details as untrusted observations and reproject through the
/// same domain allowlist; HTTP status and text never supply a missing cause.
fn internal_startup_transfer(
@@ -7219,7 +7248,7 @@ mod tests {
let mut wire = runtime_api::RuntimeApiError::new(
500,
"worker_execution_rejected",
"unused remote prose",
"TEST_ONLY_CREDENTIAL private-subject-id raw_http_body",
);
wire.error.startup = embedded.startup.clone();
let received = serde_json::from_str(&serde_json::to_string(&wire).unwrap()).unwrap();
@@ -7237,8 +7266,14 @@ mod tests {
public.get("startup").is_none(),
"no blind public/storage attachment forwarding"
);
for secret in ["TEST_ONLY_CREDENTIAL", "/private/path", "raw_http_body"] {
for secret in [
"TEST_ONLY_CREDENTIAL",
"private-subject-id",
"/private/path",
"raw_http_body",
] {
assert!(!serde_json::to_string(&embedded).unwrap().contains(secret));
assert!(!serde_json::to_string(&remote).unwrap().contains(secret));
}
let legacy = runtime_api_diagnostic(
@@ -7257,6 +7292,84 @@ mod tests {
assert!(internal_startup_transfer(corrupt).is_none());
}
#[test]
fn remote_startup_receivers_discard_untrusted_prose_and_code_even_without_paths() {
use protocol::session_failure::{
SessionRegistrationCause, session_registration_attachment,
};
let valid = runtime_api::WorkerStartupFailure {
primary: session_registration_attachment(SessionRegistrationCause::binding_mismatch())
.unwrap(),
secondary: vec![],
persistence: None,
resources: None,
io_kind: None,
};
for corrupt in [false, true] {
for status in [StatusCode::FORBIDDEN, StatusCode::INTERNAL_SERVER_ERROR] {
let mut startup = valid.clone();
if corrupt {
startup.primary = serde_json::from_value(json!("corrupt attachment")).unwrap();
}
let mut wire = runtime_api::RuntimeApiError::new(
status.as_u16(),
"PRIVATE_CODE_SENTINEL",
"TEST_ONLY_CREDENTIAL private-subject-id raw_http_body",
);
wire.error.startup = Some(startup);
let body = serde_json::to_string(&wire).unwrap();
let typed = runtime_api_diagnostic(
"remote",
runtime_api::client_support::ClientError::Public {
status,
error: serde_json::from_str(&body).unwrap(),
},
);
let (url, server) = serve_mock_http(vec![mock_response(
"GET",
"/",
false,
status.as_u16(),
body,
)]);
let response = reqwest::blocking::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
.unwrap()
.get(url)
.send()
.unwrap();
let http = remote_http_status_diagnostic("remote", status, response);
server.join().unwrap();
for diagnostic in [typed, http] {
assert_eq!(diagnostic.startup.is_some(), !corrupt);
assert_eq!(diagnostic.message.contains("session.record"), !corrupt);
assert_eq!(
diagnostic.code,
if status == StatusCode::FORBIDDEN {
"remote_runtime_auth_failed"
} else {
"remote_runtime_startup_failed"
}
);
let public = serde_json::to_string(&diagnostic).unwrap();
for secret in [
"PRIVATE_CODE_SENTINEL",
"TEST_ONLY_CREDENTIAL",
"private-subject-id",
"raw_http_body",
"corrupt attachment",
] {
assert!(
!public.contains(secret),
"received prose/code must never bypass startup projection: {public}"
);
}
}
}
}
}
#[tokio::test]
async fn async_workdir_authorization_uses_store_capacity_without_bearer_fallback() {
use crate::db_executor::{DbExecutor, DbExecutorLimits};