From 6810f51f460ac0027628156bab5945423c148c3e Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 05:53:58 +0900 Subject: [PATCH] fix: retain safe Drive and config adapter failure diagnostics --- crates/worker/src/feature/builtin.rs | 1 + .../src/feature/builtin/backend_diagnostic.rs | 195 ++++++ .../src/feature/builtin/drive/backend.rs | 237 +++++-- .../feature/builtin/drive/backend_tests.rs | 644 ++++++++++++++++++ .../builtin/workspace_config/backend.rs | 117 +++- .../builtin/workspace_config/backend_tests.rs | 580 ++++++++++++++++ .../builtin/workspace_config/content.rs | 74 +- .../workspace_config/production_tests.rs | 161 ++++- .../feature/builtin/workspace_config/wip.rs | 9 +- 9 files changed, 1947 insertions(+), 71 deletions(-) create mode 100644 crates/worker/src/feature/builtin/backend_diagnostic.rs create mode 100644 crates/worker/src/feature/builtin/drive/backend_tests.rs create mode 100644 crates/worker/src/feature/builtin/workspace_config/backend_tests.rs diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index dac21d1d5..26b4d9a4e 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -4,6 +4,7 @@ //! same descriptor-approved registry path used by feature modules. They are not //! an external plugin-loading surface. +mod backend_diagnostic; pub mod backend_job_result; pub(crate) mod bash; diff --git a/crates/worker/src/feature/builtin/backend_diagnostic.rs b/crates/worker/src/feature/builtin/backend_diagnostic.rs new file mode 100644 index 000000000..ac7d7db44 --- /dev/null +++ b/crates/worker/src/feature/builtin/backend_diagnostic.rs @@ -0,0 +1,195 @@ +//! Internal adapter diagnostics are independent of public failure classifications. +//! +//! Only closed vocabularies, numeric status and opaque correlation tokens cross this boundary. +//! In particular, a transport error's Display/Debug and server-controlled text are never recorded. +use sha2::{Digest, Sha256}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Adapter { + Drive, + WorkspaceConfig, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Stage { + Request, + Response, + Completion, + Reconcile, + Attach, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Reason { + Deadline, + TaskJoin, + TransportUnavailable, + InvalidPath, + Communication, + Decode, + RequestSize, + ResponseSize, + ResponseLength, + Identity, + InvalidMetadata, + BackendRejected, + AccessCeiling, + AttachConversion, + BackendUnknown, + UnrecognizedResponse, + NotCommittedSnapshot, + IncompleteReceipt, + Reconciled, +} + +/// One logical call, including its original-request reconciliation. Request identities are hashed +/// rather than logged verbatim: callers and responses need not contain safe log text. +#[derive(Clone)] +pub(super) struct Diagnostic { + adapter: Adapter, + operation: &'static str, + mutation: bool, + correlation: String, + dispatch: tracing::Dispatch, + stage_override: Option, +} + +impl Diagnostic { + pub(super) fn new( + adapter: Adapter, + operation: &'static str, + mutation: bool, + request_id: Option<&str>, + ) -> Self { + Self { + adapter, + operation, + mutation, + correlation: request_id + .map(|id| { + let hash: String = Sha256::digest(id.as_bytes()) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + format!("request:{hash}") + }) + .unwrap_or_else(|| uuid::Uuid::now_v7().to_string()), + dispatch: tracing::dispatcher::get_default(Clone::clone), + stage_override: None, + } + } + + pub(super) fn at_stage(&self, stage: Stage) -> Self { + let mut diagnostic = self.clone(); + diagnostic.stage_override = Some(stage); + diagnostic + } + + pub(super) fn record(&self, stage: Stage, reason: Reason, status: Option) { + let stage = self.stage_override.unwrap_or(stage); + tracing::dispatcher::with_default(&self.dispatch, || { + tracing::warn!( + target: "worker::backend_diagnostic", + adapter = ?self.adapter, + operation = self.operation, + mutation = self.mutation, + correlation = %self.correlation, + stage = ?stage, + reason = ?reason, + status, + "Workspace adapter diagnostic" + ); + }); + } + + pub(super) fn transport(&self, error: &crate::worker::WorkspaceClientError) { + use crate::worker::WorkspaceClientError; + self.record( + Stage::Request, + match error { + WorkspaceClientError::Unavailable(_) => Reason::TransportUnavailable, + WorkspaceClientError::InvalidPath(_) => Reason::InvalidPath, + WorkspaceClientError::Request(_) => Reason::Communication, + }, + None, + ); + } +} + +#[cfg(test)] +pub(super) mod tests { + use super::{Adapter, Diagnostic, Reason, Stage}; + use std::sync::{Arc, Mutex}; + use tracing::{Event, Metadata, Subscriber, field::Visit, span}; + + #[test] + fn unsafe_request_identity_is_correlated_without_logging_its_value() { + let capture = Capture::default(); + tracing::dispatcher::with_default(&capture.dispatch(), || { + for id in [ + "secret /host/path\ncredential", + "secret /host/path\ncredential", + "different", + ] { + Diagnostic::new(Adapter::Drive, "mutate", true, Some(id)).record( + Stage::Request, + Reason::Communication, + None, + ); + } + }); + let text = capture.text(); + for secret in ["secret", "/host/path", "credential", "different"] { + assert!(!text.contains(secret)); + } + let correlations: Vec<_> = text + .lines() + .map(|line| { + line.split("correlation=") + .nth(1) + .unwrap() + .split(';') + .next() + .unwrap() + }) + .collect(); + assert_eq!(correlations.len(), 3); + assert_eq!(correlations[0], correlations[1]); + assert_ne!(correlations[0], correlations[2]); + } + + #[derive(Clone, Default)] + pub(crate) struct Capture(pub(crate) Arc>>); + impl Capture { + pub(crate) fn dispatch(&self) -> tracing::Dispatch { + tracing::Dispatch::new(self.clone()) + } + pub(crate) fn text(&self) -> String { + self.0.lock().unwrap().join("\n") + } + } + struct Fields(String); + impl Visit for Fields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + use std::fmt::Write; + write!(&mut self.0, "{}={value:?};", field.name()).unwrap(); + } + } + impl Subscriber for Capture { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + metadata.target() == "worker::backend_diagnostic" + } + fn new_span(&self, _: &span::Attributes<'_>) -> span::Id { + span::Id::from_u64(1) + } + fn record(&self, _: &span::Id, _: &span::Record<'_>) {} + fn record_follows_from(&self, _: &span::Id, _: &span::Id) {} + fn event(&self, event: &Event<'_>) { + let mut fields = Fields(String::new()); + event.record(&mut fields); + self.0.lock().unwrap().push(fields.0); + } + fn enter(&self, _: &span::Id) {} + fn exit(&self, _: &span::Id) {} + } +} diff --git a/crates/worker/src/feature/builtin/drive/backend.rs b/crates/worker/src/feature/builtin/drive/backend.rs index 4c799d1ef..5730dcf0a 100644 --- a/crates/worker/src/feature/builtin/drive/backend.rs +++ b/crates/worker/src/feature/builtin/drive/backend.rs @@ -11,6 +11,8 @@ //! cases retain an unknown outcome with the original request ID. Never blindly resend with a new ID. use std::{sync::Arc, time::Duration}; +use super::super::backend_diagnostic::{Adapter, Diagnostic, Reason, Stage}; + use crate::worker::{ WorkspaceBinaryRequest, WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, }; @@ -23,6 +25,7 @@ const RESPONSE_LIMIT: usize = 2 * 1024 * 1024; #[derive(Clone)] pub struct DriveBackend { client: Arc, + deadline: Duration, } #[derive(Debug, Clone, thiserror::Error)] @@ -49,7 +52,10 @@ pub enum DriveError { impl DriveBackend { pub fn new(client: Arc) -> Self { - Self { client } + Self { + client, + deadline: DEADLINE, + } } pub fn workspace(&self) -> Result<&str, DriveError> { self.client @@ -61,15 +67,28 @@ impl DriveBackend { Ok(format!("/api/w/{}/drive", encode(self.workspace()?))) } pub(super) fn validate_entry(&self, entry: &DriveEntry) -> Result<(), DriveError> { - self.validate_ref(&entry.entry)?; + self.validate_entry_at( + entry, + &Diagnostic::new(Adapter::Drive, "validate_entry", false, None), + ) + } + fn validate_entry_at( + &self, + entry: &DriveEntry, + diagnostic: &Diagnostic, + ) -> Result<(), DriveError> { + self.validate_ref(&entry.entry) + .inspect_err(|_| diagnostic.record(Stage::Completion, Reason::Identity, None))?; if let Some(parent) = &entry.parent { - self.validate_ref(parent)?; + self.validate_ref(parent) + .inspect_err(|_| diagnostic.record(Stage::Completion, Reason::Identity, None))?; } if (entry.last_mutation_id.len() > 128 || entry.last_mutation_id.chars().any(char::is_control) || (entry.parent.is_some() && entry.last_mutation_id.is_empty())) || entry.size.is_some_and(|n| n > DRIVE_FILE_MAX_BYTES) { + diagnostic.record(Stage::Completion, Reason::InvalidMetadata, None); return Err(DriveError::Unavailable); } Ok(()) @@ -90,11 +109,13 @@ impl DriveBackend { request: WorkspaceRequest, mutation: bool, request_id: &str, + diagnostic: &Diagnostic, ) -> Result { let client = self.client.clone(); + let deadline = self.deadline; let response = tokio::time::timeout( - DEADLINE, - tokio::task::spawn_blocking(move || client.execute_with_timeout(request, DEADLINE)), + deadline, + tokio::task::spawn_blocking(move || client.execute_with_timeout(request, deadline)), ) .await; let unknown = || { @@ -105,10 +126,20 @@ impl DriveBackend { } }; let response = response - .map_err(|_| unknown())? - .map_err(|_| unknown())? - .map_err(|_| unknown())?; + .map_err(|_| { + diagnostic.record(Stage::Request, Reason::Deadline, None); + unknown() + })? + .map_err(|_| { + diagnostic.record(Stage::Request, Reason::TaskJoin, None); + unknown() + })? + .map_err(|error| { + diagnostic.transport(&error); + unknown() + })?; if response.body.len() > RESPONSE_LIMIT { + diagnostic.record(Stage::Response, Reason::ResponseSize, None); return Err(if mutation { unknown() } else { @@ -121,42 +152,57 @@ impl DriveBackend { response.body.as_bytes(), mutation, request_id, + diagnostic, )); } - serde_json::from_str(&response.body).map_err(|_| unknown()) + serde_json::from_str(&response.body).map_err(|_| { + diagnostic.record(Stage::Response, Reason::Decode, None); + unknown() + }) } async fn get( &self, - suffix: &str, + suffix: &'static str, query: &Q, + diagnostic: &Diagnostic, ) -> Result { let query = query_string(query)?; self.json( WorkspaceRequest::get(format!("{}{suffix}{query}", self.base()?)), false, "", + diagnostic, ) .await } pub async fn capabilities(&self) -> Result { - self.get("/capabilities", &serde_json::json!({})).await + self.get( + "/capabilities", + &serde_json::json!({}), + &Diagnostic::new(Adapter::Drive, "capabilities", false, None), + ) + .await } pub async fn root(&self) -> Result { + let diagnostic = Diagnostic::new(Adapter::Drive, "root", false, None); let entry: DriveEntry = self .json( WorkspaceRequest::get(format!("{}/root", self.base()?)), false, "", + &diagnostic, ) .await?; - self.validate_entry(&entry)?; + self.validate_entry_at(&entry, &diagnostic)?; if entry.kind != DriveEntryKind::Folder || entry.parent.is_some() { + diagnostic.record(Stage::Completion, Reason::InvalidMetadata, None); return Err(DriveError::Unavailable); } Ok(entry) } pub async fn metadata(&self, entry: &DriveEntryRef) -> Result { self.validate_ref(entry)?; + let diagnostic = Diagnostic::new(Adapter::Drive, "metadata", false, None); let result: DriveEntry = self .get( "/metadata", @@ -164,10 +210,12 @@ impl DriveBackend { entry_workspace_id: entry.workspace_id.clone(), id: entry.node_id.clone(), }, + &diagnostic, ) .await?; - self.validate_entry(&result)?; + self.validate_entry_at(&result, &diagnostic)?; if result.entry != *entry { + diagnostic.record(Stage::Completion, Reason::Identity, None); return Err(DriveError::Unavailable); } Ok(result) @@ -179,6 +227,7 @@ impl DriveBackend { after: Option, ) -> Result { self.validate_ref(parent)?; + let diagnostic = Diagnostic::new(Adapter::Drive, "list", false, None); let result: DriveListResponse = self .get( "/list", @@ -188,6 +237,7 @@ impl DriveBackend { limit, after, }, + &diagnostic, ) .await?; self.validate_page( @@ -195,33 +245,43 @@ impl DriveBackend { limit .unwrap_or(DRIVE_PAGE_MAX_LIMIT) .min(DRIVE_PAGE_MAX_LIMIT), + &diagnostic, )?; if result .entries .iter() .any(|entry| entry.parent.as_ref() != Some(parent)) { + diagnostic.record(Stage::Completion, Reason::Identity, None); return Err(DriveError::Unavailable); } Ok(result) } pub async fn search(&self, query: DriveSearchQuery) -> Result { - let result: DriveListResponse = self.get("/search", &query).await?; + let diagnostic = Diagnostic::new(Adapter::Drive, "search", false, None); + let result: DriveListResponse = self.get("/search", &query, &diagnostic).await?; self.validate_page( &result, query .limit .unwrap_or(DRIVE_SEARCH_MAX_LIMIT) .min(DRIVE_SEARCH_MAX_LIMIT), + &diagnostic, )?; Ok(result) } - fn validate_page(&self, page: &DriveListResponse, limit: u32) -> Result<(), DriveError> { + fn validate_page( + &self, + page: &DriveListResponse, + limit: u32, + diagnostic: &Diagnostic, + ) -> Result<(), DriveError> { if page.entries.len() > limit as usize { + diagnostic.record(Stage::Completion, Reason::ResponseSize, None); return Err(DriveError::Limit); } for entry in &page.entries { - self.validate_entry(entry)?; + self.validate_entry_at(entry, diagnostic)?; } Ok(()) } @@ -231,6 +291,7 @@ impl DriveBackend { max_bytes: u32, ) -> Result { self.validate_ref(entry)?; + let diagnostic = Diagnostic::new(Adapter::Drive, "read_text", false, None); let result: DriveReadTextResponse = self .get( "/read-text", @@ -239,10 +300,16 @@ impl DriveBackend { id: entry.node_id.clone(), max_bytes, }, + &diagnostic, ) .await?; - self.validate_entry(&result.entry)?; - if result.entry.entry != *entry || result.text.len() > max_bytes as usize { + self.validate_entry_at(&result.entry, &diagnostic)?; + if result.entry.entry != *entry { + diagnostic.record(Stage::Completion, Reason::Identity, None); + return Err(DriveError::Unavailable); + } + if result.text.len() > max_bytes as usize { + diagnostic.record(Stage::Completion, Reason::ResponseSize, None); return Err(DriveError::Unavailable); } Ok(result) @@ -255,6 +322,7 @@ impl DriveBackend { if total as usize > limit { return Err(DriveError::Limit); } + let diagnostic = Diagnostic::new(Adapter::Drive, "read_chunk", false, None); let mut bytes = Vec::with_capacity(total as usize); // Even an empty file gets a read bound to the observed committed mutation and current authorization. loop { @@ -274,21 +342,42 @@ impl DriveBackend { max_response_bytes: DRIVE_CHUNK_MAX_BYTES as usize, }; let client = self.client.clone(); + let deadline = self.deadline; let response = tokio::time::timeout( - DEADLINE, + deadline, tokio::task::spawn_blocking(move || { - client.execute_binary_with_timeout(request, DEADLINE) + client.execute_binary_with_timeout(request, deadline) }), ) .await - .map_err(|_| DriveError::Unavailable)? - .map_err(|_| DriveError::Unavailable)? - .map_err(|_| DriveError::Unavailable)?; + .map_err(|_| { + diagnostic.record(Stage::Request, Reason::Deadline, None); + DriveError::Unavailable + })? + .map_err(|_| { + diagnostic.record(Stage::Request, Reason::TaskJoin, None); + DriveError::Unavailable + })? + .map_err(|error| { + diagnostic.transport(&error); + DriveError::Unavailable + })?; + if response.body.len() > DRIVE_CHUNK_MAX_BYTES as usize { + diagnostic.record(Stage::Response, Reason::ResponseSize, None); + return Err(DriveError::Unavailable); + } if !(200..300).contains(&response.status) { - return Err(classify(response.status, &response.body, false, "")); + return Err(classify( + response.status, + &response.body, + false, + "", + &diagnostic, + )); } let expected = total.saturating_sub(offset).min(length) as usize; if response.body.len() != expected { + diagnostic.record(Stage::Response, Reason::ResponseLength, None); return Err(DriveError::Unavailable); } bytes.extend_from_slice(&response.body); @@ -301,22 +390,40 @@ impl DriveBackend { &self, id: &str, result: Result, + diagnostic: &Diagnostic, ) -> Result { // A parseable success can still be an invalid completion. Classify it // before reconciliation so it also gets one original-request inquiry. let result = result.and_then(|response| { - self.validate_completion(id, &response) + self.validate_completion(id, &response, diagnostic) .map_err(|_| DriveError::OutcomeUnknown(id.into()))?; Ok(response) }); match result { - Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await { + Err(DriveError::OutcomeUnknown(_)) => match self + .status_with_diagnostic(id, &diagnostic.at_stage(Stage::Reconcile)) + .await + { Ok(DriveRequestStatusResponse { state: DriveRequestState::Committed, response: Some(response), .. - }) => Ok(response), // status validates both IDs and entry metadata - _ => Err(DriveError::OutcomeUnknown(id.into())), + }) => { + diagnostic.record(Stage::Reconcile, Reason::Reconciled, None); + Ok(response) // status validates both IDs and entry metadata + } + Ok(status) => { + diagnostic.record( + Stage::Reconcile, + match status.state { + DriveRequestState::Uncommitted => Reason::NotCommittedSnapshot, + DriveRequestState::Committed => Reason::IncompleteReceipt, + }, + None, + ); + Err(DriveError::OutcomeUnknown(id.into())) + } + Err(_) => Err(DriveError::OutcomeUnknown(id.into())), }, other => other, } @@ -325,12 +432,14 @@ impl DriveBackend { &self, id: &str, response: &DriveMutationResponse, + diagnostic: &Diagnostic, ) -> Result<(), DriveError> { if response.request_id != id { + diagnostic.record(Stage::Completion, Reason::Identity, None); return Err(DriveError::Unavailable); } if let Some(entry) = &response.entry { - self.validate_entry(entry)?; + self.validate_entry_at(entry, diagnostic)?; } Ok(()) } @@ -338,9 +447,11 @@ impl DriveBackend { &self, request: DriveMutationRequest, ) -> Result { + let diagnostic = Diagnostic::new(Adapter::Drive, "mutate", true, Some(&request.request_id)); let body = serde_json::to_string(&request) .map_err(|_| DriveError::Invalid("invalid mutation".into()))?; if body.len() > 512 * 1024 { + diagnostic.record(Stage::Request, Reason::RequestSize, None); return Err(DriveError::Limit); } let result = self @@ -352,16 +463,20 @@ impl DriveBackend { ), true, &request.request_id, + &diagnostic, ) .await; - self.reconcile(&request.request_id, result).await + self.reconcile(&request.request_id, result, &diagnostic) + .await } pub async fn upload( &self, query: DriveUploadQuery, bytes: Vec, ) -> Result { + let diagnostic = Diagnostic::new(Adapter::Drive, "upload", true, Some(&query.request_id)); if bytes.len() > DRIVE_FILE_MAX_BYTES as usize { + diagnostic.record(Stage::Request, Reason::RequestSize, None); return Err(DriveError::Limit); } let path = format!("{}/upload{}", self.base()?, query_string(&query)?); @@ -372,29 +487,56 @@ impl DriveBackend { max_response_bytes: RESPONSE_LIMIT, }; let client = self.client.clone(); + let deadline = self.deadline; let result = tokio::time::timeout( - DEADLINE, + deadline, tokio::task::spawn_blocking(move || { - client.execute_binary_with_timeout(request, DEADLINE) + client.execute_binary_with_timeout(request, deadline) }), ) .await; let result = match result { + Ok(Ok(Ok(response))) if response.body.len() > RESPONSE_LIMIT => { + diagnostic.record(Stage::Response, Reason::ResponseSize, None); + Err(DriveError::OutcomeUnknown(query.request_id.clone())) + } Ok(Ok(Ok(response))) if (200..300).contains(&response.status) => { - serde_json::from_slice(&response.body) - .map_err(|_| DriveError::OutcomeUnknown(query.request_id.clone())) + serde_json::from_slice(&response.body).map_err(|_| { + diagnostic.record(Stage::Response, Reason::Decode, None); + DriveError::OutcomeUnknown(query.request_id.clone()) + }) } Ok(Ok(Ok(response))) => Err(classify( response.status, &response.body, true, &query.request_id, + &diagnostic, )), - _ => Err(DriveError::OutcomeUnknown(query.request_id.clone())), + Err(_) => { + diagnostic.record(Stage::Request, Reason::Deadline, None); + Err(DriveError::OutcomeUnknown(query.request_id.clone())) + } + Ok(Err(_)) => { + diagnostic.record(Stage::Request, Reason::TaskJoin, None); + Err(DriveError::OutcomeUnknown(query.request_id.clone())) + } + Ok(Ok(Err(error))) => { + diagnostic.transport(&error); + Err(DriveError::OutcomeUnknown(query.request_id.clone())) + } }; - self.reconcile(&query.request_id, result).await + self.reconcile(&query.request_id, result, &diagnostic).await } pub async fn status(&self, request_id: &str) -> Result { + let diagnostic = Diagnostic::new(Adapter::Drive, "request_status", false, Some(request_id)); + self.status_with_diagnostic(request_id, &diagnostic).await + } + async fn status_with_diagnostic( + &self, + request_id: &str, + diagnostic: &Diagnostic, + ) -> Result { if request_id.is_empty() || request_id.len() > 256 { return Err(DriveError::Invalid("invalid request ID".into())); } @@ -403,13 +545,15 @@ impl DriveBackend { WorkspaceRequest::get(format!("{}/requests/{}", self.base()?, encode(request_id))), false, "", + diagnostic, ) .await?; if response.request_id != request_id { + diagnostic.record(Stage::Reconcile, Reason::Identity, None); return Err(DriveError::Unavailable); } if let Some(result) = &response.response { - self.validate_completion(request_id, result)?; + self.validate_completion(request_id, result, diagnostic)?; } Ok(response) } @@ -463,11 +607,21 @@ fn query_string(value: &T) -> Result { format!("?{}", pairs.join("&")) }) } -fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError { +fn classify( + status: u16, + body: &[u8], + mutation: bool, + id: &str, + diagnostic: &Diagnostic, +) -> DriveError { if let Ok(error) = serde_json::from_slice::(body) { - if error.classification == DriveFailureClassification::Unknown { + if error.classification == DriveFailureClassification::Unknown + || error.code == DriveApiErrorCode::OutcomeUnknown + { + diagnostic.record(Stage::Response, Reason::BackendUnknown, Some(status)); return DriveError::OutcomeUnknown(id.into()); } + diagnostic.record(Stage::Response, Reason::BackendRejected, Some(status)); return match error.code { DriveApiErrorCode::Denied => DriveError::Denied, DriveApiErrorCode::NotFound => DriveError::NotFound, @@ -478,6 +632,7 @@ fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError { DriveApiErrorCode::OutcomeUnknown => DriveError::OutcomeUnknown(id.into()), }; } + diagnostic.record(Stage::Response, Reason::UnrecognizedResponse, Some(status)); if matches!(status, 401 | 403) { DriveError::Denied } else if mutation { @@ -486,3 +641,7 @@ fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError { DriveError::Unavailable } } + +#[cfg(test)] +#[path = "backend_tests.rs"] +mod tests; diff --git a/crates/worker/src/feature/builtin/drive/backend_tests.rs b/crates/worker/src/feature/builtin/drive/backend_tests.rs new file mode 100644 index 000000000..46109b3cc --- /dev/null +++ b/crates/worker/src/feature/builtin/drive/backend_tests.rs @@ -0,0 +1,644 @@ +//! Production Drive adapter fault injection. Only the Workspace transport boundary is substituted; +//! retained fixture effects and original-request inquiries are inspected independently of replies. +//! The HTTP case additionally connects the adapter to reqwest and an isolated loopback router. +use super::*; +use crate::feature::builtin::backend_diagnostic::{Reason, tests::Capture}; +use crate::worker::{WorkspaceBinaryResponse, WorkspaceClientError, WorkspaceResponse}; +use std::sync::{Condvar, Mutex}; +use tracing::instrument::WithSubscriber; + +const SECRET: &str = "secret-token /private/drive credential=unsafe"; +const ID: &str = "host-request-1"; + +fn entry() -> DriveEntry { + DriveEntry { + entry: DriveEntryRef { + workspace_id: "workspace".into(), + node_id: "2".into(), + }, + parent: Some(DriveEntryRef { + workspace_id: "workspace".into(), + node_id: "1".into(), + }), + name: SECRET.into(), + kind: DriveEntryKind::File, + last_mutation_id: "old-request".into(), + size: Some(5), + content_type: Some("text/plain".into()), + updated_by: SECRET.into(), + updated_at: SECRET.into(), + latest_url: SECRET.into(), + } +} +fn mutation() -> DriveMutationRequest { + DriveMutationRequest { + request_id: ID.into(), + mutation: DriveMutation::UpdateText { + id: entry().entry, + expected_mutation_id: "old-request".into(), + text: SECRET.into(), + content_type: "text/plain".into(), + }, + } +} +fn json(value: impl Serialize) -> WorkspaceResponse { + WorkspaceResponse { + status: 200, + body: serde_json::to_string(&value).unwrap(), + } +} +#[derive(Clone, Copy, Debug)] +enum Fault { + Communication, + Unavailable, + InvalidPath, + Decode, + Size, + Length, + Identity, + Metadata, + Rejected, + Unknown, + Unrecognized, + Incomplete, + Panic, + Deadline, + None, +} +#[derive(Debug, Default)] +struct State { + mutations: usize, + saved: Option, + requests: Vec, + binary: Vec, + released: bool, + started: bool, + finished: bool, +} +#[derive(Debug)] +struct Client { + fault: Fault, + receipt: Fault, + state: Mutex, + changed: Condvar, + started: tokio::sync::Notify, +} +impl Client { + fn new(fault: Fault, receipt: Fault) -> Arc { + Arc::new(Self { + fault, + receipt, + state: Mutex::new(State::default()), + changed: Condvar::new(), + started: tokio::sync::Notify::new(), + }) + } + fn fail( + &self, + fault: Fault, + success: WorkspaceResponse, + ) -> Result { + match fault { + Fault::Communication => Err(WorkspaceClientError::Request(SECRET.into())), + Fault::Unavailable => Err(WorkspaceClientError::Unavailable(SECRET.into())), + Fault::InvalidPath => Err(WorkspaceClientError::InvalidPath(SECRET.into())), + Fault::Panic => panic!("injected adapter task panic"), + Fault::Decode => Ok(WorkspaceResponse { + status: 200, + body: SECRET.into(), + }), + Fault::Size => Ok(WorkspaceResponse { + status: 200, + body: SECRET.repeat(RESPONSE_LIMIT / SECRET.len() + 1), + }), + Fault::Rejected | Fault::Unknown => Ok(WorkspaceResponse { + status: 409, + body: serde_json::to_string(&DriveApiError { + code: DriveApiErrorCode::Conflict, + message: SECRET.into(), + classification: if matches!(fault, Fault::Unknown) { + DriveFailureClassification::Unknown + } else { + DriveFailureClassification::NotCommitted + }, + }) + .unwrap(), + }), + Fault::Unrecognized => Ok(WorkspaceResponse { + status: 503, + body: SECRET.into(), + }), + _ => Ok(success), + } + } + fn release(&self) { + self.state.lock().unwrap().released = true; + self.changed.notify_all(); + } + fn wait(&self, finished: bool) { + let (state, timeout) = self + .changed + .wait_timeout_while(self.state.lock().unwrap(), Duration::from_secs(5), |s| { + if finished { !s.finished } else { !s.started } + }) + .unwrap(); + assert!(!timeout.timed_out()); + assert!(if finished { + state.finished + } else { + state.started + }); + } +} +struct Release(Arc); +impl Drop for Release { + fn drop(&mut self) { + self.0.release(); + } +} +impl WorkspaceClient for Client { + fn workspace_id(&self) -> Option<&str> { + Some("workspace") + } + fn kind(&self) -> &str { + "drive-diagnostic-fixture" + } + fn is_available(&self) -> bool { + true + } + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + let is_status = request.path.contains("/requests/"); + let fault = if is_status { self.receipt } else { self.fault }; + let mutation = request.method == WorkspaceRequestMethod::Post; + { + let mut state = self.state.lock().unwrap(); + if mutation + && !matches!( + fault, + Fault::Rejected | Fault::InvalidPath | Fault::Unavailable + ) + { + let input: DriveMutationRequest = + serde_json::from_str(request.body.as_ref().unwrap()).unwrap(); + assert_eq!(input, super::tests::mutation()); + state.mutations += 1; + if let DriveMutation::UpdateText { text, .. } = input.mutation { + state.saved = Some(text); + } + } + state.requests.push(request); + state.started = true; + self.started.notify_one(); + self.changed.notify_all(); + if matches!(fault, Fault::Deadline) { + let (_state, timeout) = self + .changed + .wait_timeout_while(state, Duration::from_secs(5), |s| !s.released) + .unwrap(); + assert!( + !timeout.timed_out(), + "transport must be explicitly released" + ); + } + } + let mut metadata = entry(); + if matches!(fault, Fault::Identity) { + metadata.entry.node_id = "3".into(); + } + if matches!(fault, Fault::Metadata) { + metadata.size = Some(DRIVE_FILE_MAX_BYTES + 1); + } + let completion = DriveMutationResponse { + request_id: if matches!(fault, Fault::Identity) { + "other-request" + } else { + ID + } + .into(), + entry: Some(metadata.clone()), + }; + let value = if is_status { + json(DriveRequestStatusResponse { + request_id: if matches!(fault, Fault::Identity) { + "other-request" + } else { + ID + } + .into(), + state: if matches!( + fault, + Fault::None | Fault::Identity | Fault::Metadata | Fault::Incomplete + ) { + DriveRequestState::Committed + } else { + DriveRequestState::Uncommitted + }, + response: if matches!(fault, Fault::None | Fault::Identity | Fault::Metadata) { + Some(completion) + } else { + None + }, + }) + } else if mutation { + json(completion) + } else { + json(metadata) + }; + if !is_status { + self.state.lock().unwrap().finished = true; + } + self.changed.notify_all(); + self.fail(fault, value) + } + fn execute_binary( + &self, + request: WorkspaceBinaryRequest, + ) -> Result { + let upload = request.method == WorkspaceRequestMethod::Put; + { + let mut state = self.state.lock().unwrap(); + if upload { + state.mutations += 1; + state.saved = Some(SECRET.into()); + } + state.binary.push(request); + } + let success = if upload { + json(DriveMutationResponse { + request_id: ID.into(), + entry: Some(entry()), + }) + } else { + WorkspaceResponse { + status: 200, + body: if matches!(self.fault, Fault::Length) { + "x" + } else { + "12345" + } + .into(), + } + }; + self.fail(self.fault, success) + .map(|r| WorkspaceBinaryResponse { + status: r.status, + body: r.body.into_bytes(), + }) + } +} +fn safe(capture: &Capture, reason: Reason, public: &str) { + let text = capture.text(); + assert!(text.contains(&format!("reason={reason:?}")), "{text}"); + assert!(text.contains("adapter=Drive"), "{text}"); + assert!(text.contains("correlation="), "{text}"); + for secret in [ + SECRET, + "secret-token", + "/private/drive", + "credential=unsafe", + ] { + assert!(!text.contains(secret), "internal diagnostic leaked content"); + assert!(!public.contains(secret), "public error leaked content"); + } +} +fn reason(fault: Fault) -> Reason { + match fault { + Fault::Communication => Reason::Communication, + Fault::Unavailable => Reason::TransportUnavailable, + Fault::InvalidPath => Reason::InvalidPath, + Fault::Decode => Reason::Decode, + Fault::Size => Reason::ResponseSize, + Fault::Length => Reason::ResponseLength, + Fault::Identity => Reason::Identity, + Fault::Metadata => Reason::InvalidMetadata, + Fault::Rejected => Reason::BackendRejected, + Fault::Unknown => Reason::BackendUnknown, + Fault::Unrecognized => Reason::UnrecognizedResponse, + Fault::Panic => Reason::TaskJoin, + Fault::Deadline => Reason::Deadline, + Fault::Incomplete => Reason::IncompleteReceipt, + Fault::None => panic!("no fault"), + } +} +#[tokio::test] +async fn production_read_failures_retain_public_classification_and_safe_cause() { + for fault in [ + Fault::Communication, + Fault::Unavailable, + Fault::InvalidPath, + Fault::Decode, + Fault::Size, + Fault::Identity, + Fault::Metadata, + Fault::Rejected, + Fault::Unknown, + Fault::Unrecognized, + Fault::Panic, + ] { + let client = Client::new(fault, Fault::None); + let backend = DriveBackend::new(client.clone()); + let capture = Capture::default(); + let error = backend + .metadata(&entry().entry) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err(); + match fault { + Fault::Size => assert!(matches!(error, DriveError::Limit)), + Fault::Rejected => assert!(matches!(error, DriveError::Conflict)), + Fault::Unknown => assert!(matches!(error, DriveError::OutcomeUnknown(_))), + _ => assert!(matches!(error, DriveError::Unavailable)), + } + safe(&capture, reason(fault), &error.to_string()); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests.len(), 1); + assert_eq!(state.mutations, 0); + } +} +#[tokio::test] +async fn production_mutation_loss_reconciles_only_original_identity_and_never_resends() { + for fault in [ + Fault::Communication, + Fault::Decode, + Fault::Size, + Fault::Identity, + Fault::Metadata, + Fault::Unknown, + Fault::Unrecognized, + Fault::Panic, + ] { + for receipt in [ + Fault::None, + Fault::Identity, + Fault::Metadata, + Fault::Decode, + Fault::Communication, + Fault::Unrecognized, + Fault::Length, + Fault::Incomplete, + ] { + let client = Client::new(fault, receipt); + let backend = DriveBackend::new(client.clone()); + let capture = Capture::default(); + let result = backend + .mutate(mutation()) + .with_subscriber(capture.dispatch()) + .await; + if matches!(receipt, Fault::None) { + assert_eq!(result.unwrap().request_id, ID); + } else { + assert!(matches!(result, Err(DriveError::OutcomeUnknown(ref id)) if id == ID)); + } + safe(&capture, reason(fault), ""); + let text = capture.text(); + if matches!(receipt, Fault::Length | Fault::Incomplete) { + safe( + &capture, + if matches!(receipt, Fault::Length) { + Reason::NotCommittedSnapshot + } else { + Reason::IncompleteReceipt + }, + "", + ); + } + assert!(text.contains("stage=Reconcile"), "{text}"); + let correlations: std::collections::BTreeSet<_> = text + .lines() + .filter_map(|l| l.split("correlation=").nth(1)) + .filter_map(|l| l.split(';').next()) + .collect(); + assert_eq!( + correlations.len(), + 1, + "original loss and inquiry must correlate" + ); + assert!( + !text.contains(ID), + "request identity must be opaque in logs" + ); + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, 1); + assert_eq!(state.saved.as_deref(), Some(SECRET)); + assert_eq!(state.requests.len(), 2); + assert_eq!(state.requests[1].method, WorkspaceRequestMethod::Get); + assert!(state.requests[1].path.ends_with(&format!("/requests/{ID}"))); + } + } +} +#[tokio::test] +async fn production_rejection_has_no_effect_or_status_inquiry() { + let client = Client::new(Fault::Rejected, Fault::None); + let backend = DriveBackend::new(client.clone()); + let capture = Capture::default(); + let error = backend + .mutate(mutation()) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err(); + assert!(matches!(error, DriveError::Conflict)); + safe(&capture, Reason::BackendRejected, &error.to_string()); + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, 0); + assert_eq!(state.requests.len(), 1); +} +#[tokio::test] +async fn production_binary_length_transport_and_upload_loss_are_diagnosed() { + for fault in [ + Fault::Length, + Fault::Communication, + Fault::Unavailable, + Fault::Panic, + ] { + let client = Client::new(fault, Fault::None); + let backend = DriveBackend::new(client.clone()); + let capture = Capture::default(); + let error = backend + .bytes(&entry(), 5) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err(); + assert!(matches!(error, DriveError::Unavailable)); + safe(&capture, reason(fault), &error.to_string()); + assert_eq!(client.state.lock().unwrap().binary.len(), 1); + } + for fault in [Fault::Decode, Fault::Communication, Fault::Size] { + let client = Client::new(fault, Fault::Unrecognized); + let backend = DriveBackend::new(client.clone()); + let capture = Capture::default(); + let query = DriveUploadQuery { + operation: DriveUploadOperation::Update, + request_id: ID.into(), + entry_workspace_id: "workspace".into(), + parent_id: None, + name: None, + id: Some("2".into()), + expected_mutation_id: Some("old-request".into()), + content_type: "text/plain".into(), + size: 5, + sha256: "hash".into(), + }; + let error = backend + .upload(query, b"12345".to_vec()) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err(); + assert!(matches!(error, DriveError::OutcomeUnknown(ref id) if id == ID)); + safe(&capture, reason(fault), &error.to_string()); + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, 1); + assert_eq!(state.binary.len(), 1); + assert_eq!(state.requests.len(), 1); + assert!(state.requests[0].path.ends_with(&format!("/requests/{ID}"))); + } +} +#[tokio::test] +async fn production_deadline_does_not_cancel_effect_or_infer_uncommitted_from_snapshot() { + for mutation_call in [false, true] { + let client = Client::new(Fault::Deadline, Fault::Unrecognized); + let release = Release(client.clone()); + let mut backend = DriveBackend::new(client.clone()); + backend.deadline = Duration::from_secs(1); + let capture = Capture::default(); + let request = async { + if mutation_call { + backend.mutate(mutation()).await.map(|_| ()) + } else { + backend.metadata(&entry().entry).await.map(|_| ()) + } + } + .with_subscriber(capture.dispatch()); + tokio::pin!(request); + tokio::select! { + biased; + _ = client.started.notified() => {}, + _ = &mut request => panic!("request must start before timeout"), + } + { + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, usize::from(mutation_call)); + assert!(!state.released && !state.finished); + } + let result = request.await; + let error = result.unwrap_err(); + assert!(if mutation_call { + matches!(error, DriveError::OutcomeUnknown(ref id) if id == ID) + } else { + matches!(error, DriveError::Unavailable) + }); + safe(&capture, Reason::Deadline, &error.to_string()); + drop(release); + let waiting = client.clone(); + tokio::task::spawn_blocking(move || waiting.wait(true)) + .await + .unwrap(); + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, usize::from(mutation_call)); + assert_eq!(state.saved.is_some(), mutation_call); + assert_eq!(state.requests.len(), if mutation_call { 2 } else { 1 }); + } +} + +#[tokio::test] +async fn http_reply_loss_keeps_original_receipt_and_retained_effect_without_reexecution() { + use axum::{ + Router, + extract::State as HttpState, + routing::{get, post}, + }; + #[derive(Default)] + struct Stored { + updates: usize, + content: String, + ids: Vec, + } + struct Server(tokio::task::JoinHandle<()>); + impl Drop for Server { + fn drop(&mut self) { + self.0.abort(); + } + } + let stored = Arc::new(Mutex::new(Stored::default())); + let router = Router::new() + .route("/api/w/workspace/drive/metadata", get(|| async { SECRET })) + .route( + "/api/w/workspace/drive/mutate", + post( + |HttpState(state): HttpState>>, + body: axum::body::Bytes| async move { + let request: DriveMutationRequest = serde_json::from_slice(&body).unwrap(); + let mut state = state.lock().unwrap(); + state.ids.push(request.request_id); + if let DriveMutation::UpdateText { + text, + expected_mutation_id, + .. + } = request.mutation + { + assert_eq!(expected_mutation_id, "old-request"); + state.content = text; + state.updates += 1; + } else { + panic!("unexpected mutation"); + } + // The effect is retained; only the success response is malformed. + SECRET + }, + ), + ) + .route( + "/api/w/workspace/drive/requests/{id}", + get( + |HttpState(state): HttpState>>, + axum::extract::Path(id): axum::extract::Path| async move { + let state = state.lock().unwrap(); + assert_eq!(state.ids, [id.clone()]); + axum::Json(DriveRequestStatusResponse { + request_id: id.clone(), + state: DriveRequestState::Committed, + response: Some(DriveMutationResponse { + request_id: id, + entry: Some(entry()), + }), + }) + }, + ), + ) + .with_state(stored.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = Server(tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + })); + let client = Arc::new(crate::worker::TestWorkspaceHttpClient::new( + "workspace", + format!("http://{address}"), + )); + let backend = DriveBackend::new(client); + let capture = Capture::default(); + let error = backend + .metadata(&entry().entry) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err(); + assert!(matches!(error, DriveError::Unavailable)); + safe(&capture, Reason::Decode, &error.to_string()); + let capture = Capture::default(); + let response = backend + .mutate(mutation()) + .with_subscriber(capture.dispatch()) + .await + .unwrap(); + assert_eq!(response.request_id, ID); + safe(&capture, Reason::Decode, ""); + assert!(capture.text().contains("reason=Reconciled")); + let state = stored.lock().unwrap(); + assert_eq!(state.updates, 1); + assert_eq!(state.content, SECRET); + assert_eq!(state.ids, [ID]); + drop(state); + drop(server); +} diff --git a/crates/worker/src/feature/builtin/workspace_config/backend.rs b/crates/worker/src/feature/builtin/workspace_config/backend.rs index f10c60eda..d5d58e376 100644 --- a/crates/worker/src/feature/builtin/workspace_config/backend.rs +++ b/crates/worker/src/feature/builtin/workspace_config/backend.rs @@ -7,6 +7,7 @@ //! undecodable mutation responses do not prove non-commit; dispatched requests are never resent. //! This adapter does not reimplement Backend source-persistence transactions or activation. use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use serde::{Serialize, de::DeserializeOwned}; @@ -20,6 +21,7 @@ use super::{ ATTACHMENT_ALIAS, CONTENT_ROOT, ConfigAccess, ConfigAttachError, ConfigAttachment, DEADLINE, WorkspaceConfigAttachmentBackend, }; +use crate::feature::builtin::backend_diagnostic::{Adapter, Diagnostic, Reason, Stage}; use crate::wip::WipOperationError; use crate::worker::{ WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, @@ -34,11 +36,21 @@ const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024; #[derive(Clone)] pub struct WorkspaceConfigBackend { client: Arc, + deadline: Duration, } impl WorkspaceConfigBackend { pub fn new(client: Arc) -> Self { - Self { client } + Self { + client, + deadline: DEADLINE, + } + } + + #[cfg(test)] + pub(super) fn with_deadline(mut self, deadline: Duration) -> Self { + self.deadline = deadline; + self } fn base(&self) -> Result { @@ -47,6 +59,7 @@ impl WorkspaceConfigBackend { .workspace_id() .filter(|id| !id.is_empty()) .ok_or_else(|| { + diagnostic("authority", false).record(Stage::Request, Reason::Identity, None); failure( ProtocolErrorCode::PermissionDenied, "Workspace config authority unavailable", @@ -69,11 +82,13 @@ impl WorkspaceConfigBackend { &self, request: WorkspaceRequest, mutation: bool, + diagnostic: &Diagnostic, ) -> Result { let response = self .client - .execute_with_timeout(request, DEADLINE) + .execute_with_timeout(request, self.deadline) .map_err(|error| { + diagnostic.transport(&error); if mutation && !matches!(error, WorkspaceClientError::InvalidPath(_)) { unknown() } else { @@ -84,6 +99,7 @@ impl WorkspaceConfigBackend { } })?; if response.body.len() > MAX_RESPONSE_BYTES { + diagnostic.record(Stage::Response, Reason::ResponseSize, Some(response.status)); return Err(if mutation { unknown() } else { @@ -96,8 +112,18 @@ impl WorkspaceConfigBackend { if !response.is_success() { if let Ok(error) = serde_json::from_str::(&response.body) { if error.classification == WorkspaceConfigFailureClassification::Unknown { + diagnostic.record( + Stage::Response, + Reason::BackendUnknown, + Some(response.status), + ); return Err(unknown()); } + diagnostic.record( + Stage::Response, + Reason::BackendRejected, + Some(response.status), + ); // Do not reflect arbitrary transport/provider text into model output. let code = match error.status { 401 | 403 | 404 => ProtocolErrorCode::PermissionDenied, @@ -111,6 +137,11 @@ impl WorkspaceConfigBackend { "Workspace config request rejected without commit", )); } + diagnostic.record( + Stage::Response, + Reason::UnrecognizedResponse, + Some(response.status), + ); return Err(if mutation { unknown() } else { @@ -125,6 +156,7 @@ impl WorkspaceConfigBackend { }); } serde_json::from_str(&response.body).map_err(|_| { + diagnostic.record(Stage::Response, Reason::Decode, Some(response.status)); if mutation { unknown() } else { @@ -138,10 +170,17 @@ impl WorkspaceConfigBackend { /// GET checks the current grant even when the Worker has not attached yet. pub fn current(&self) -> Result, WipOperationError> { + self.current_diagnosed(&diagnostic("current", false)) + } + + fn current_diagnosed( + &self, + diagnostic: &Diagnostic, + ) -> Result, WipOperationError> { let result: Option = - self.request(WorkspaceRequest::get(self.base()?), false)?; + self.request(WorkspaceRequest::get(self.base()?), false, diagnostic)?; if let Some(attachment) = &result { - self.validate_attachment(attachment)?; + self.validate_attachment(attachment, diagnostic)?; } Ok(result) } @@ -149,6 +188,7 @@ impl WorkspaceConfigBackend { fn validate_attachment( &self, attachment: &WorkspaceConfigAttachment, + diagnostic: &Diagnostic, ) -> Result<(), WipOperationError> { if Some(attachment.workspace_id.as_str()) != self.client.workspace_id() || attachment.alias != ATTACHMENT_ALIAS @@ -160,6 +200,7 @@ impl WorkspaceConfigBackend { .bytes() .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_')) { + diagnostic.record(Stage::Response, Reason::Identity, None); return Err(failure( ProtocolErrorCode::PermissionDenied, "Invalid Workspace config attachment identity", @@ -173,14 +214,34 @@ impl WorkspaceConfigBackend { suffix: &str, request: &R, mutation: bool, + ) -> Result { + let operation = match suffix { + "/observe" => "observe", + "/read" => "read", + "/commit" => "commit", + "" => "attach", + _ => "post", + }; + self.post_diagnosed(suffix, request, mutation, diagnostic(operation, mutation)) + .await + } + + pub(super) async fn post_diagnosed( + &self, + suffix: &str, + request: &R, + mutation: bool, + diagnostic: Diagnostic, ) -> Result { let body = serde_json::to_string(request).map_err(|_| { + diagnostic.record(Stage::Request, Reason::Decode, None); failure( ProtocolErrorCode::InvalidArguments, "Invalid Workspace config request", ) })?; if body.len() > MAX_RESPONSE_BYTES { + diagnostic.record(Stage::Request, Reason::RequestSize, None); return Err(failure( ProtocolErrorCode::ResourceLimitExceeded, "Workspace config request exceeds limit", @@ -192,12 +253,16 @@ impl WorkspaceConfigBackend { body, ); let this = self.clone(); + let request_diagnostic = diagnostic.clone(); tokio::time::timeout( - DEADLINE, - tokio::task::spawn_blocking(move || this.request(request, mutation)), + self.deadline, + tokio::task::spawn_blocking(move || { + this.request(request, mutation, &request_diagnostic) + }), ) .await .map_err(|_| { + diagnostic.record(Stage::Completion, Reason::Deadline, None); if mutation { unknown() } else { @@ -208,6 +273,7 @@ impl WorkspaceConfigBackend { } })? .map_err(|_| { + diagnostic.record(Stage::Completion, Reason::TaskJoin, None); if mutation { unknown() } else { @@ -223,18 +289,22 @@ impl WorkspaceConfigBackend { &self, ) -> Result, WipOperationError> { let this = self.clone(); + let diagnostic = diagnostic("current", false); + let request_diagnostic = diagnostic.clone(); tokio::time::timeout( - DEADLINE, - tokio::task::spawn_blocking(move || this.current()), + self.deadline, + tokio::task::spawn_blocking(move || this.current_diagnosed(&request_diagnostic)), ) .await .map_err(|_| { + diagnostic.record(Stage::Completion, Reason::Deadline, None); failure( ProtocolErrorCode::ResourceLimitExceeded, "Workspace config request deadline exceeded", ) })? .map_err(|_| { + diagnostic.record(Stage::Completion, Reason::TaskJoin, None); failure( ProtocolErrorCode::Internal, "Workspace config transport unavailable", @@ -249,10 +319,14 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend { self.client.is_available() } fn validate_access(&self, access: Option) -> Result<(), ConfigAttachError> { - let current = self.current().map_err(attach_error)?; + let diagnostic = diagnostic("validate_access", false); + let current = self + .current_diagnosed(&diagnostic) + .map_err(|error| attach_error(error, &diagnostic))?; if access == Some(ConfigAccess::ReadWrite) && current.is_some_and(|a| a.access == WorkspaceConfigAccess::ReadOnly) { + diagnostic.record(Stage::Attach, Reason::AccessCeiling, None); return Err(ConfigAttachError::Denied); } Ok(()) @@ -271,10 +345,16 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend { ConfigAccess::ReadWrite => WorkspaceConfigAccess::ReadWrite, }), }; - let attachment: WorkspaceConfigAttachment = - self.post("", &request, true).await.map_err(attach_error)?; - self.validate_attachment(&attachment) - .map_err(|_| ConfigAttachError::OutcomeUnknown)?; + let diagnostic = diagnostic("attach", true); + let attachment: WorkspaceConfigAttachment = self + .post_diagnosed("", &request, true, diagnostic.clone()) + .await + .map_err(|error| attach_error(error, &diagnostic))?; + self.validate_attachment(&attachment, &diagnostic) + .map_err(|_| { + diagnostic.record(Stage::Attach, Reason::Identity, None); + ConfigAttachError::OutcomeUnknown + })?; Ok(ConfigAttachment { connection_id: attachment.connection_id, access: match attachment.access { @@ -286,7 +366,8 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend { } } -fn attach_error(error: WipOperationError) -> ConfigAttachError { +fn attach_error(error: WipOperationError, diagnostic: &Diagnostic) -> ConfigAttachError { + diagnostic.record(Stage::Attach, Reason::AttachConversion, None); match error { WipOperationError::OutcomeUnknown(_) => ConfigAttachError::OutcomeUnknown, WipOperationError::Protocol(error) => match error.code { @@ -298,6 +379,10 @@ fn attach_error(error: WipOperationError) -> ConfigAttachError { } } +pub(super) fn diagnostic(operation: &'static str, mutation: bool) -> Diagnostic { + Diagnostic::new(Adapter::WorkspaceConfig, operation, mutation, None) +} + pub(super) fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError { WipOperationError::Protocol(ProtocolError { code, @@ -307,3 +392,7 @@ pub(super) fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationErr pub(super) fn unknown() -> WipOperationError { WipOperationError::OutcomeUnknown("Workspace config outcome unknown; refresh before any subsequent operation; no automatic retry".into()) } + +#[cfg(test)] +#[path = "backend_tests.rs"] +mod tests; diff --git a/crates/worker/src/feature/builtin/workspace_config/backend_tests.rs b/crates/worker/src/feature/builtin/workspace_config/backend_tests.rs new file mode 100644 index 000000000..b617107c7 --- /dev/null +++ b/crates/worker/src/feature/builtin/workspace_config/backend_tests.rs @@ -0,0 +1,580 @@ +//! Production WorkspaceConfigBackend with only WorkspaceClient substituted. The transport fixture +//! applies effects before injected reply loss, retains its state for rereading, and gates blocking +//! completion explicitly. These ordinary worker lib tests do not claim Backend transaction coverage. +use super::*; +use crate::feature::builtin::backend_diagnostic::tests::Capture; +use crate::worker::WorkspaceResponse; +use std::sync::{Condvar, Mutex}; +use tracing::instrument::WithSubscriber; + +const UNSAFE: &str = "secret-token /private/config credential=unsafe"; + +#[derive(Clone, Copy, Debug)] +enum Fault { + Unavailable, + InvalidPath, + Communication, + Decode, + Size, + Rejected(u16), + Unknown, + Unrecognized, + Identity, + Panic, + Deadline, +} + +#[derive(Debug, Default)] +struct Effects { + requests: usize, + mutations: usize, + saved: Option, + completed: bool, + released: bool, +} + +#[derive(Debug)] +struct Client { + fault: Fault, + state: Mutex, + changed: Condvar, + started: tokio::sync::Notify, +} +impl Client { + fn new(fault: Fault) -> Arc { + Arc::new(Self { + fault, + state: Mutex::new(Effects::default()), + changed: Condvar::new(), + started: tokio::sync::Notify::new(), + }) + } + fn finish(&self) { + self.state.lock().unwrap().completed = true; + self.changed.notify_all(); + } + fn release(&self) { + self.state.lock().unwrap().released = true; + self.changed.notify_all(); + } + fn wait_completed(&self) { + let (state, timeout) = self + .changed + .wait_timeout_while(self.state.lock().unwrap(), Duration::from_secs(5), |s| { + !s.completed + }) + .unwrap(); + assert!( + !timeout.timed_out() && state.completed, + "blocking request did not finish" + ); + } +} +struct Release(Arc); +impl Drop for Release { + fn drop(&mut self) { + self.0.release(); + } +} + +fn attachment() -> WorkspaceConfigAttachment { + WorkspaceConfigAttachment { + workspace_id: "workspace".into(), + connection_id: "connection-1".into(), + alias: ATTACHMENT_ALIAS.into(), + content_path: CONTENT_ROOT.into(), + working_directory_id: "grant".into(), + access: WorkspaceConfigAccess::ReadWrite, + name: UNSAFE.into(), + purpose: UNSAFE.into(), + already_attached: false, + } +} + +impl WorkspaceClient for Client { + fn workspace_id(&self) -> Option<&str> { + Some("workspace") + } + fn kind(&self) -> &str { + "diagnostic-test-boundary" + } + fn is_available(&self) -> bool { + true + } + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + { + let mut state = self.state.lock().unwrap(); + state.requests += 1; + if request.method == WorkspaceRequestMethod::Post + && !request.path.ends_with("/read") + && !matches!( + self.fault, + Fault::Unavailable | Fault::InvalidPath | Fault::Rejected(_) + ) + { + state.mutations += 1; + state.saved = request.body.clone(); + } + self.started.notify_one(); + if matches!(self.fault, Fault::Deadline) { + let (_state, timeout) = self + .changed + .wait_timeout_while(state, Duration::from_secs(5), |s| !s.released) + .unwrap(); + assert!(!timeout.timed_out(), "test must release blocked transport"); + } + } + self.finish(); + match self.fault { + Fault::Unavailable => Err(WorkspaceClientError::Unavailable(UNSAFE.into())), + Fault::InvalidPath => Err(WorkspaceClientError::InvalidPath(UNSAFE.into())), + Fault::Communication => Err(WorkspaceClientError::Request(UNSAFE.into())), + Fault::Panic => panic!("injected blocking task failure"), + Fault::Decode => Ok(WorkspaceResponse { + status: 200, + body: UNSAFE.into(), + }), + Fault::Size => Ok(WorkspaceResponse { + status: 200, + body: UNSAFE.repeat(MAX_RESPONSE_BYTES / UNSAFE.len() + 1), + }), + Fault::Rejected(_) | Fault::Unknown => { + let status = match self.fault { + Fault::Rejected(status) => status, + _ => 503, + }; + Ok(WorkspaceResponse { + status, + body: serde_json::to_string(&WorkspaceConfigApiError { + status, + code: UNSAFE.into(), + message: UNSAFE.into(), + classification: if matches!(self.fault, Fault::Unknown) { + WorkspaceConfigFailureClassification::Unknown + } else { + WorkspaceConfigFailureClassification::NotCommitted + }, + }) + .unwrap(), + }) + } + Fault::Unrecognized => Ok(WorkspaceResponse { + status: 503, + body: UNSAFE.into(), + }), + Fault::Identity | Fault::Deadline => { + let mut value = attachment(); + if matches!(self.fault, Fault::Identity) { + value.workspace_id = UNSAFE.into(); + } + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string(&value).unwrap(), + }) + } + } + } +} + +fn assert_safe(capture: &Capture, reason: Reason, stage: Stage, public: &str) { + let text = capture.text(); + assert!( + text.contains(&format!("reason={reason:?}")), + "missing reason: {text}" + ); + assert!( + text.contains(&format!("stage={stage:?}")), + "missing stage: {text}" + ); + assert!(text.contains("adapter=WorkspaceConfig"), "{text}"); + assert!(text.contains("correlation="), "{text}"); + for secret in [ + UNSAFE, + "secret-token", + "/private/config", + "credential=unsafe", + ] { + assert!( + !text.contains(secret), + "diagnostics leaked unsafe boundary data" + ); + assert!( + !public.contains(secret), + "public result leaked unsafe boundary data" + ); + } +} +fn public_error( + result: Result, + code: Option, +) -> String { + match (result, code) { + (Err(WipOperationError::OutcomeUnknown(message)), None) => { + assert!(message.contains("no automatic retry")); + message + } + (Err(WipOperationError::Protocol(error)), Some(expected)) => { + assert_eq!(error.code, expected); + error.message + } + _ => panic!("unexpected public error classification"), + } +} + +#[tokio::test] +async fn production_reply_failures_retain_reasons_without_retrying_mutations() { + let cases = [ + ( + Fault::Unavailable, + Reason::TransportUnavailable, + Stage::Request, + ProtocolErrorCode::Internal, + false, + ), + ( + Fault::InvalidPath, + Reason::InvalidPath, + Stage::Request, + ProtocolErrorCode::Internal, + false, + ), + ( + Fault::Communication, + Reason::Communication, + Stage::Request, + ProtocolErrorCode::Internal, + true, + ), + ( + Fault::Decode, + Reason::Decode, + Stage::Response, + ProtocolErrorCode::Internal, + true, + ), + ( + Fault::Size, + Reason::ResponseSize, + Stage::Response, + ProtocolErrorCode::ResourceLimitExceeded, + true, + ), + ( + Fault::Unknown, + Reason::BackendUnknown, + Stage::Response, + ProtocolErrorCode::Internal, + true, + ), + ( + Fault::Unrecognized, + Reason::UnrecognizedResponse, + Stage::Response, + ProtocolErrorCode::Internal, + true, + ), + ( + Fault::Panic, + Reason::TaskJoin, + Stage::Completion, + ProtocolErrorCode::Internal, + true, + ), + ]; + for (fault, reason, stage, read_code, applies) in cases { + for mutation in [false, true] { + let client = Client::new(fault); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + let result = backend + .post::( + if mutation { "/commit" } else { "/read" }, + &serde_json::json!({"content": UNSAFE}), + mutation, + ) + .with_subscriber(capture.dispatch()) + .await; + let code = if (mutation && !matches!(fault, Fault::InvalidPath)) + || matches!(fault, Fault::Unknown) + { + None + } else { + Some(read_code) + }; + let public = public_error(result, code); + assert_safe(&capture, reason, stage, &public); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 1); + assert_eq!(state.mutations, usize::from(mutation && applies)); + assert_eq!(state.saved.is_some(), mutation && applies); + } + } +} + +#[tokio::test] +async fn production_typed_rejections_preserve_public_codes_and_never_apply() { + for (status, code) in [ + (403, ProtocolErrorCode::PermissionDenied), + (409, ProtocolErrorCode::ValidatorMismatch), + (413, ProtocolErrorCode::ResourceLimitExceeded), + (422, ProtocolErrorCode::InvalidArguments), + (500, ProtocolErrorCode::Internal), + ] { + for mutation in [false, true] { + let client = Client::new(Fault::Rejected(status)); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + let public = public_error( + backend + .post::( + if mutation { "/commit" } else { "/read" }, + &(), + mutation, + ) + .with_subscriber(capture.dispatch()) + .await, + Some(code), + ); + assert_safe(&capture, Reason::BackendRejected, Stage::Response, &public); + assert!(capture.text().contains(&format!("status={status}"))); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 1); + assert_eq!(state.mutations, 0); + assert!(state.saved.is_none()); + } + } +} + +#[tokio::test] +async fn production_deadline_is_unknown_after_effect_and_blocking_completion_is_not_retried() { + for mutation in [false, true] { + let client = Client::new(Fault::Deadline); + let release = Release(client.clone()); + let backend = + WorkspaceConfigBackend::new(client.clone()).with_deadline(Duration::from_secs(1)); + let capture = Capture::default(); + let body = serde_json::json!({"content": UNSAFE}); + let request = backend + .post::( + if mutation { "/commit" } else { "/read" }, + &body, + mutation, + ) + .with_subscriber(capture.dispatch()); + tokio::pin!(request); + tokio::select! { + biased; + _ = client.started.notified() => {}, + _ = &mut request => panic!("request must start and apply its effect before timeout is observed"), + } + { + let state = client.state.lock().unwrap(); + assert_eq!(state.mutations, usize::from(mutation)); + assert_eq!(state.saved.is_some(), mutation); + assert!(!state.completed && !state.released); + } + let result = request.await; + let public = public_error( + result, + if mutation { + None + } else { + Some(ProtocolErrorCode::ResourceLimitExceeded) + }, + ); + assert_safe(&capture, Reason::Deadline, Stage::Completion, &public); + drop(release); + let completed = client.clone(); + tokio::task::spawn_blocking(move || completed.wait_completed()) + .await + .unwrap(); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 1); + assert_eq!(state.mutations, usize::from(mutation)); + assert_eq!(state.saved.is_some(), mutation); + } +} + +#[tokio::test] +async fn production_current_identity_and_task_loss_remain_safe() { + for (fault, reason, stage, code) in [ + ( + Fault::Identity, + Reason::Identity, + Stage::Response, + ProtocolErrorCode::PermissionDenied, + ), + ( + Fault::Panic, + Reason::TaskJoin, + Stage::Completion, + ProtocolErrorCode::Internal, + ), + ( + Fault::Decode, + Reason::Decode, + Stage::Response, + ProtocolErrorCode::Internal, + ), + ] { + let client = Client::new(fault); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + let result = backend + .current_async() + .with_subscriber(capture.dispatch()) + .await; + let message = match result { + Err(WipOperationError::Protocol(error)) => { + assert_eq!(error.code, code); + error.message + } + _ => panic!("unexpected current classification"), + }; + assert_safe(&capture, reason, stage, &message); + assert_eq!(client.state.lock().unwrap().mutations, 0); + } +} + +#[tokio::test] +async fn production_attach_conversion_retains_original_reason_and_effect_count() { + for (fault, reason, expected, mutations) in [ + ( + Fault::Communication, + Reason::Communication, + ConfigAttachError::OutcomeUnknown, + 1, + ), + ( + Fault::Identity, + Reason::Identity, + ConfigAttachError::OutcomeUnknown, + 1, + ), + ( + Fault::Rejected(403), + Reason::BackendRejected, + ConfigAttachError::Denied, + 0, + ), + ( + Fault::Rejected(422), + Reason::BackendRejected, + ConfigAttachError::InvalidRequest, + 0, + ), + ( + Fault::Rejected(500), + Reason::BackendRejected, + ConfigAttachError::Unavailable, + 0, + ), + ] { + let client = Client::new(fault); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + assert_eq!( + backend + .attach("host-key-not-logged", None) + .with_subscriber(capture.dispatch()) + .await, + Err(expected) + ); + assert_safe(&capture, reason, Stage::Attach, expected.message()); + let text = capture.text(); + if !matches!(fault, Fault::Identity) { + assert!(text.contains("reason=AttachConversion"), "{text}"); + } + assert!(!text.contains("host-key-not-logged")); + let correlations: std::collections::BTreeSet<_> = text + .lines() + .filter_map(|line| line.split("correlation=").nth(1)) + .filter_map(|tail| tail.split(';').next()) + .collect(); + assert_eq!( + correlations.len(), + 1, + "request and attach conversion must correlate" + ); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 1); + assert_eq!(state.mutations, mutations); + assert_eq!(state.saved.is_some(), mutations == 1); + } +} + +#[tokio::test] +async fn production_oversized_request_is_rejected_before_dispatch() { + let client = Client::new(Fault::Communication); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + let public = public_error( + backend + .post::("/commit", &"x".repeat(MAX_RESPONSE_BYTES), true) + .with_subscriber(capture.dispatch()) + .await, + Some(ProtocolErrorCode::ResourceLimitExceeded), + ); + assert_safe(&capture, Reason::RequestSize, Stage::Request, &public); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 0); + assert_eq!(state.mutations, 0); +} + +#[tokio::test] +async fn production_current_deadline_remains_bounded_and_has_no_mutation() { + let client = Client::new(Fault::Deadline); + let release = Release(client.clone()); + let backend = WorkspaceConfigBackend::new(client.clone()).with_deadline(Duration::from_secs(1)); + let capture = Capture::default(); + let request = backend.current_async().with_subscriber(capture.dispatch()); + tokio::pin!(request); + tokio::select! { + biased; + _ = client.started.notified() => {}, + _ = &mut request => panic!("current request must enter the gate before timeout is observed"), + } + assert_eq!(client.state.lock().unwrap().mutations, 0); + let error = request.await; + let message = match error { + Err(WipOperationError::Protocol(error)) => { + assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded); + error.message + } + _ => panic!("current deadline must remain a public resource-limit error"), + }; + assert_safe(&capture, Reason::Deadline, Stage::Completion, &message); + drop(release); + let completed = client.clone(); + tokio::task::spawn_blocking(move || completed.wait_completed()) + .await + .unwrap(); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 1); + assert_eq!(state.mutations, 0); +} + +#[tokio::test] +async fn production_request_encoding_failure_is_safe_and_never_dispatched() { + struct Unencodable; + impl Serialize for Unencodable { + fn serialize(&self, _: S) -> Result { + Err(serde::ser::Error::custom(UNSAFE)) + } + } + let client = Client::new(Fault::Communication); + let backend = WorkspaceConfigBackend::new(client.clone()); + let capture = Capture::default(); + let result = backend + .post::("/commit", &Unencodable, true) + .with_subscriber(capture.dispatch()) + .await; + let message = public_error(result, Some(ProtocolErrorCode::InvalidArguments)); + assert_safe(&capture, Reason::Decode, Stage::Request, &message); + let state = client.state.lock().unwrap(); + assert_eq!(state.requests, 0); + assert_eq!(state.mutations, 0); +} diff --git a/crates/worker/src/feature/builtin/workspace_config/content.rs b/crates/worker/src/feature/builtin/workspace_config/content.rs index b2f67d342..ae6c64b41 100644 --- a/crates/worker/src/feature/builtin/workspace_config/content.rs +++ b/crates/worker/src/feature/builtin/workspace_config/content.rs @@ -36,9 +36,11 @@ use wip_protocol::{ }; use super::backend::{ - MAX_DEPTH, MAX_NODES, MAX_PATH_BYTES, MAX_TEXT_BYTES, WorkspaceConfigBackend, failure, unknown, + MAX_DEPTH, MAX_NODES, MAX_PATH_BYTES, MAX_TEXT_BYTES, WorkspaceConfigBackend, diagnostic, + failure, unknown, }; use super::{CONTENT_ROOT, WorkspaceConfigFeature}; +use crate::feature::builtin::backend_diagnostic::{Diagnostic, Reason, Stage}; use crate::wip::{ WipCallContext, WipOperationError, WipOperationHandler, WipOperationOutput, WipProjection, WipProjectionKind, WipSubtreeProvider, contextual_reference, json_to_wip, @@ -138,6 +140,23 @@ async fn observe( connection: &str, paths: Vec, depth: u32, +) -> Result { + observe_diagnosed( + backend, + connection, + paths, + depth, + diagnostic("observe", false), + ) + .await +} + +async fn observe_diagnosed( + backend: &WorkspaceConfigBackend, + connection: &str, + paths: Vec, + depth: u32, + diagnostic: Diagnostic, ) -> Result { if paths.len() > MAX_NODES || depth > MAX_DEPTH { return Err(invalid()); @@ -148,7 +167,7 @@ async fn observe( } } let result: WorkspaceConfigObserveResponse = backend - .post( + .post_diagnosed( "/observe", &WorkspaceConfigObserveRequest { connection_id: connection.into(), @@ -156,6 +175,7 @@ async fn observe( depth, }, false, + diagnostic.clone(), ) .await?; if result.connection_id != connection @@ -164,24 +184,37 @@ async fn observe( || result.validator.len() > 4096 || result.digest.is_empty() { + diagnostic.record( + Stage::Response, + if result.connection_id != connection { + Reason::Identity + } else { + Reason::InvalidMetadata + }, + None, + ); return Err(failure( ProtocolErrorCode::Internal, "Invalid Workspace config metadata", )); } if result.entrypoints.is_empty() || result.entrypoints.len() > MAX_NODES { + diagnostic.record(Stage::Response, Reason::InvalidMetadata, None); return Err(failure( ProtocolErrorCode::Internal, "Invalid Workspace config entrypoints", )); } for path in &result.entrypoints { - validate_source(path)?; + validate_source(path) + .inspect_err(|_| diagnostic.record(Stage::Response, Reason::InvalidMetadata, None))?; } let mut seen = BTreeSet::new(); for node in &result.nodes { if !node.path.is_empty() { - validate_source(&node.path)?; + validate_source(&node.path).inspect_err(|_| { + diagnostic.record(Stage::Response, Reason::InvalidMetadata, None) + })?; } if !seen.insert(&node.path) || node.validator.is_empty() @@ -198,6 +231,7 @@ async fn observe( suffix.is_some_and(|s| s.split('/').count() <= depth as usize) }) { + diagnostic.record(Stage::Response, Reason::InvalidMetadata, None); return Err(failure( ProtocolErrorCode::Internal, "Invalid Workspace config metadata scope", @@ -490,9 +524,10 @@ fn content_type(value: Option<&Value>) -> Result Result { + let diagnostic = diagnostic("read", false); let response: WorkspaceConfigReadResponse = self .backend - .post( + .post_diagnosed( "/read", &WorkspaceConfigReadRequest { connection_id: self.attachment.connection_id.clone(), @@ -500,6 +535,7 @@ impl ConfigHandler { validator: self.node.validator.clone(), }, false, + diagnostic.clone(), ) .await?; if response.path != self.node.path @@ -507,12 +543,14 @@ impl ConfigHandler { || Some(&response.digest) != self.node.digest.as_ref() || Some(&response.content_type) != self.node.content_type.as_ref() { + diagnostic.record(Stage::Response, Reason::Identity, None); return Err(failure( ProtocolErrorCode::Internal, "Invalid Workspace config read response", )); } if response.content.len() > MAX_TEXT_BYTES { + diagnostic.record(Stage::Response, Reason::ResponseSize, None); return Err(failure( ProtocolErrorCode::ResourceLimitExceeded, "Workspace config source exceeds text limit", @@ -630,12 +668,17 @@ impl WipOperationHandler for ConfigHandler { context: WipCallContext, ) -> Result { let dispatched = std::sync::atomic::AtomicBool::new(false); + let diagnostic = diagnostic( + "operation", + !matches!(operation, "read" | "read_lines" | "search_text"), + ); tokio::time::timeout( super::DEADLINE, - self.execute(operation, arguments, context, &dispatched), + self.execute(operation, arguments, context, &dispatched, &diagnostic), ) .await .map_err(|_| { + diagnostic.record(Stage::Completion, Reason::Deadline, None); if dispatched.load(std::sync::atomic::Ordering::Acquire) { unknown() } else { @@ -655,6 +698,7 @@ impl ConfigHandler { arguments: &BTreeMap, context: WipCallContext, dispatched: &std::sync::atomic::AtomicBool, + operation_diagnostic: &Diagnostic, ) -> Result { if !self.operations.iter().any(|name| name == operation) { return Err(failure( @@ -798,9 +842,10 @@ impl ConfigHandler { }; let changes = self.canonical_changes(changes).await?; dispatched.store(true, std::sync::atomic::Ordering::Release); + let diagnostic = operation_diagnostic.clone(); let result: WorkspaceConfigCommitResponse = self .backend - .post( + .post_diagnosed( "/commit", &WorkspaceConfigCommitRequest { connection_id: self.attachment.connection_id.clone(), @@ -812,28 +857,37 @@ impl ConfigHandler { }, }, true, + diagnostic.clone(), ) .await?; // Commit response supplies ROOT state. Obtain the exact post-commit PATH // validator, never substitute the root validator for a file validator. - let updated = observe( + let updated = observe_diagnosed( &self.backend, &self.attachment.connection_id, vec![self.node.path.clone()], 0, + diagnostic.at_stage(Stage::Reconcile), ) .await .map_err(|_| unknown())?; if updated.digest != result.digest || updated.validator != result.validator { + diagnostic.record(Stage::Reconcile, Reason::Identity, None); return Err(unknown()); } let node = updated .nodes .iter() .find(|n| n.path == self.node.path) - .ok_or_else(unknown)?; + .ok_or_else(|| { + diagnostic.record(Stage::Reconcile, Reason::InvalidMetadata, None); + unknown() + })?; let value = json_to_wip(&json!({"digest": result.digest, - "activation": "Saved to canonical Workspace configuration; existing Worker manifests are not rewritten"})).map_err(|_| unknown())?; + "activation": "Saved to canonical Workspace configuration; existing Worker manifests are not rewritten"})).map_err(|_| { + diagnostic.record(Stage::Completion, Reason::Decode, None); + unknown() + })?; Ok(WipOperationOutput::native_with_validator( value, object_validator( diff --git a/crates/worker/src/feature/builtin/workspace_config/production_tests.rs b/crates/worker/src/feature/builtin/workspace_config/production_tests.rs index 71649555c..27ee6f161 100644 --- a/crates/worker/src/feature/builtin/workspace_config/production_tests.rs +++ b/crates/worker/src/feature/builtin/workspace_config/production_tests.rs @@ -1,8 +1,10 @@ -//! Worker-side E2E across real Feature invocation, WIP Client/Host and typed -//! Workspace transport. Canonical persistence/projection tests live in Backend; -//! this deterministic router models its published auth/CAS/response contract. +//! Worker-side integration across real Feature invocation, WIP Client/Host and the production +//! WorkspaceConfigBackend. Only WorkspaceClient transport is substituted by a deterministic router +//! retaining applied effects and modeling the published auth/CAS/response contract. These ordinary +//! worker lib tests do not replace canonical persistence/projection tests owned by Backend. use super::*; use crate::feature::FeatureRegistryBuilder; +use crate::feature::builtin::backend_diagnostic::tests::Capture; use crate::hook::HookRegistryBuilder; use crate::wip::{WipMountRegistry, WipRuntime}; use crate::worker::{ @@ -16,6 +18,7 @@ use serde_json::{Value as Json, json}; use server_api::*; use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, Mutex}; +use tracing::instrument::WithSubscriber; #[derive(Debug)] struct State { @@ -26,6 +29,10 @@ struct State { entries: BTreeMap, commits: Vec, requests: Vec, + applied_commits: usize, + malformed_read_reply: Option<&'static str>, + malformed_observe_reply: Option<&'static str>, + lose_postcommit_observe_reply: bool, fail_commit: bool, fail_current: bool, lose_commit_reply: bool, @@ -58,6 +65,10 @@ impl Router { ]), commits: Vec::new(), requests: Vec::new(), + applied_commits: 0, + malformed_read_reply: None, + malformed_observe_reply: None, + lose_postcommit_observe_reply: false, fail_commit: false, fail_current: false, lose_commit_reply: false, @@ -222,6 +233,11 @@ impl WorkspaceClient for Router { } match suffix { "/observe" => { + if state.lose_postcommit_observe_reply && state.applied_commits > 0 { + return Err(WorkspaceClientError::Request( + "secret-token /private/host/config".into(), + )); + } let request: WorkspaceConfigObserveRequest = serde_json::from_value(input).unwrap(); assert!(request.depth <= 8 && request.paths.len() <= 256); let mut kinds = @@ -268,13 +284,23 @@ impl WorkspaceClient for Router { ) }) .collect(); - Ok(response(&WorkspaceConfigObserveResponse { + let mut result = WorkspaceConfigObserveResponse { connection_id: state.active.as_ref().unwrap().connection_id.clone(), validator: state.validator(""), digest: state.tree_digest(), entrypoints: vec!["main.dcdl".into()], nodes, - })) + }; + match state.malformed_observe_reply { + Some("connection") => { + result.connection_id = "secret-token /private/host/config".into() + } + Some("entrypoints") => result.entrypoints.clear(), + Some("scope") => result.nodes[0].path = "outside/path.dcdl".into(), + None => {} + Some(_) => unreachable!(), + } + Ok(response(&result)) } "/read" => { let request: WorkspaceConfigReadRequest = serde_json::from_value(input).unwrap(); @@ -290,13 +316,20 @@ impl WorkspaceClient for Router { WorkspaceConfigFailureClassification::NotCommitted, )); }; - Ok(response(&WorkspaceConfigReadResponse { + let mut result = WorkspaceConfigReadResponse { path: request.path.clone(), content: content.clone(), content_type: *content_type, digest: digest(content), validator: state.validator(&request.path), - })) + }; + match state.malformed_read_reply { + Some("identity") => result.path = "secret-token /private/host/config".into(), + Some("size") => result.content = "x".repeat(256 * 1024 + 1), + None => {} + Some(_) => unreachable!(), + } + Ok(response(&result)) } "/commit" => { let request: WorkspaceConfigCommitRequest = serde_json::from_value(input).unwrap(); @@ -403,6 +436,7 @@ impl WorkspaceClient for Router { )); } state.entries = entries; + state.applied_commits += 1; if state.lose_commit_reply { state.lose_commit_reply = false; return Err(WorkspaceClientError::Request( @@ -797,14 +831,20 @@ async fn post_commit_observation_mismatch_reports_unknown_without_replaying_writ let path = "/workspace-config/main.dcdl"; observe_interface(&runtime, path, true).await; client.state.lock().unwrap().mismatched_commit_reply = Some(field); + let capture = Capture::default(); let error = call(&runtime, path, "write", json!({"content":"saved"})) + .with_subscriber(capture.dispatch()) .await .unwrap_err() .to_string(); assert!(error.contains("unknown"), "{field}: {error}"); + assert!(capture.text().contains("stage=Reconcile")); + assert!(capture.text().contains("reason=Identity")); + assert!(!capture.text().contains("saved")); let state = client.state.lock().unwrap(); assert_eq!(state.commits.len(), 1); assert_eq!(state.entries["main.dcdl"].1, "saved"); + assert_eq!(state.applied_commits, 1); } } @@ -1155,3 +1195,110 @@ async fn workspace_config_backend_validators_reject_path_and_connection_transpla assert!(result.content.unwrap().contains("model = updated")); assert_eq!(client.state.lock().unwrap().commits.len(), 1); } + +#[tokio::test] +async fn production_read_validation_retains_reason_without_committing() { + for (fault, reason) in [("identity", "Identity"), ("size", "ResponseSize")] { + let client = Router::new(); + let (feature, runtime) = runtime(client.clone()); + feature.attach("initial", None).await.unwrap(); + let path = "/workspace-config/main.dcdl"; + observe_interface(&runtime, path, true).await; + client.state.lock().unwrap().malformed_read_reply = Some(fault); + let capture = Capture::default(); + let error = call(&runtime, path, "read", json!({})) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err() + .to_string(); + let text = capture.text(); + assert!(text.contains(&format!("reason={reason}")), "{text}"); + assert!(text.contains("stage=Response")); + assert!(!error.contains("unknown")); + assert!(!text.contains("secret-token") && !text.contains("/private/host")); + assert!(!error.contains("secret-token") && !error.contains("/private/host")); + let state = client.state.lock().unwrap(); + assert_eq!(state.applied_commits, 0); + assert_eq!(state.entries["main.dcdl"].1, "model = old"); + } +} + +#[tokio::test] +async fn production_observation_validation_retains_reason_without_committing() { + for (fault, reason) in [ + ("connection", "Identity"), + ("entrypoints", "InvalidMetadata"), + ("scope", "InvalidMetadata"), + ] { + let client = Router::new(); + let (feature, runtime) = runtime(client.clone()); + feature.attach("initial", None).await.unwrap(); + client.state.lock().unwrap().malformed_observe_reply = Some(fault); + let capture = Capture::default(); + let error = runtime + .tree("/workspace-config/main.dcdl".into(), 0, true) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err() + .to_string(); + let text = capture.text(); + assert!(text.contains(&format!("reason={reason}")), "{text}"); + assert!(text.contains("stage=Response")); + assert!(!text.contains("secret-token") && !text.contains("/private/host")); + assert!(!error.contains("secret-token") && !error.contains("/private/host")); + assert_eq!(client.state.lock().unwrap().applied_commits, 0); + } +} + +#[tokio::test] +async fn production_postcommit_transport_loss_retains_cause_and_unknown_without_retry() { + let client = Router::new(); + let (feature, runtime) = runtime(client.clone()); + feature.attach("initial", None).await.unwrap(); + let path = "/workspace-config/main.dcdl"; + observe_interface(&runtime, path, true).await; + client.state.lock().unwrap().lose_postcommit_observe_reply = true; + let capture = Capture::default(); + let error = call( + &runtime, + path, + "write", + json!({"content":"saved-before-observation-loss"}), + ) + .with_subscriber(capture.dispatch()) + .await + .unwrap_err() + .to_string(); + assert!(error.contains("unknown")); + let text = capture.text(); + assert!( + text.lines().any( + |event| event.contains("reason=Communication") && event.contains("stage=Reconcile") + ), + "{text}" + ); + assert!( + !text.contains("reason=UnrecognizedResponse"), + "must not invent a response-decode cause for transport loss" + ); + assert!(!text.contains("secret-token") && !text.contains("/private/host")); + assert!(!text.contains("saved-before-observation-loss")); + assert!(!error.contains("secret-token") && !error.contains("/private/host")); + let correlations: BTreeSet<_> = text + .lines() + .filter_map(|line| line.split("correlation=").nth(1)) + .filter_map(|tail| tail.split(';').next()) + .collect(); + assert_eq!( + correlations.len(), + 1, + "lost postcommit observation must correlate to original commit" + ); + let state = client.state.lock().unwrap(); + assert_eq!(state.commits.len(), 1); + assert_eq!(state.applied_commits, 1); + assert_eq!( + state.entries["main.dcdl"].1, + "saved-before-observation-loss" + ); +} diff --git a/crates/worker/src/feature/builtin/workspace_config/wip.rs b/crates/worker/src/feature/builtin/workspace_config/wip.rs index e03c90db3..0f77f854c 100644 --- a/crates/worker/src/feature/builtin/workspace_config/wip.rs +++ b/crates/worker/src/feature/builtin/workspace_config/wip.rs @@ -161,7 +161,14 @@ impl WipOperationHandler for AttachHandler { "attachment_path": format!("/workdir-attachments/{}", encode_identity(&result.connection_id)), "content_path": CONTENT_ROOT, "already_attached": result.already_attached, - })).map_err(|_| WipOperationError::OutcomeUnknown("Completed config attachment response unavailable".into()))?; + })).map_err(|_| { + super::backend::diagnostic("attach_output", true).record( + crate::feature::builtin::backend_diagnostic::Stage::Attach, + crate::feature::builtin::backend_diagnostic::Reason::Decode, + None, + ); + WipOperationError::OutcomeUnknown("Completed config attachment response unavailable".into()) + })?; Ok(WipOperationOutput::native(value)) } }