diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 11a33bbc6..f9377801c 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1046,10 +1046,17 @@ impl ExternalProviderWorkdirSession { }) } - async fn operate( + async fn operate( &self, operation: WorkdirSessionOperation, - ) -> std::result::Result { + validate: F, + ) -> std::result::Result + where + T: Send, + F: FnOnce(std::result::Result) -> Fut + + Send, + Fut: std::future::Future> + Send, + { let operation_kind = external_workdir_operation_kind(&operation); let operation_timeout = external_workdir_operation_timeout(&operation); if self.closed.load(Ordering::Acquire) { @@ -1169,6 +1176,12 @@ impl ExternalProviderWorkdirSession { ) } }; + // Provider dispatch is finished; validation may issue bounded command cleanup + // and must not retain a data-operation slot while awaiting another operation. + drop(_admission); + // A received response is not evidence of operation success until its requested + // type and semantics (including mutation uncertainty) have been checked. + let result = validate(result).await; let outcome = match &result { Ok(_) => Outcome::Succeeded, Err(error) @@ -1275,9 +1288,15 @@ impl ExternalProviderWorkdirSession { // Do not acquire the output lock until cancellation has unblocked any reader. // The reader records its last known cursor before cleanup acknowledges bytes. let _ = self - .operate(WorkdirSessionOperation::CommandCancel( - command.handle.clone(), - )) + .operate( + WorkdirSessionOperation::CommandCancel(command.handle.clone()), + |result| async move { + match result? { + WorkdirSessionOperationResult::CommandCancel => Ok(()), + _ => Err(Self::mismatch("command_cancel")), + } + }, + ) .await; let mut cursor = tokio::time::timeout(std::time::Duration::from_secs(30), command.cursor.lock()) @@ -1292,31 +1311,31 @@ impl ExternalProviderWorkdirSession { } let drain = async { loop { - match self - .operate(WorkdirSessionOperation::CommandOutput( - workdir::CommandOutputRequest { + let output = self + .operate( + WorkdirSessionOperation::CommandOutput(workdir::CommandOutputRequest { handle: command.handle.clone(), cursor: *cursor, limit: workdir::MAX_COMMAND_READ_BYTES, wait: true, + }), + |result| async move { + match result? { + WorkdirSessionOperationResult::CommandOutput(output) => Ok(output), + _ => Err(Self::mismatch("command_output")), + } }, - )) - .await? - { - WorkdirSessionOperationResult::CommandOutput(output) => { - *cursor = output.next_cursor; - if output.terminal.is_some() { - *command.terminal_at.lock().unwrap() = Some(std::time::Instant::now()); - return Ok(()); - } - if output.chunks.is_empty() { - return Err(workdir::WorkdirError::Unavailable( - "External Workdir provider did not confirm command termination" - .to_string(), - )); - } - } - _ => return Err(Self::mismatch("command_output")), + ) + .await?; + *cursor = output.next_cursor; + if output.terminal.is_some() { + *command.terminal_at.lock().unwrap() = Some(std::time::Instant::now()); + return Ok(()); + } + if output.chunks.is_empty() { + return Err(workdir::WorkdirError::Unavailable( + "External Workdir provider did not confirm command termination".to_string(), + )); } } }; @@ -1384,46 +1403,55 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { request: workdir::WorkdirScopeAuthorizationRequest, ) -> std::result::Result<(), workdir::WorkdirError> { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self - .operate(WorkdirSessionOperation::AuthorizeScope(request)) - .await? - { - WorkdirSessionOperationResult::AuthorizeScope => Ok(()), - _ => Err(Self::mismatch("authorize_scope")), - } + self.operate( + WorkdirSessionOperation::AuthorizeScope(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::AuthorizeScope => Ok(()), + _ => Err(Self::mismatch("authorize_scope")), + } + }, + ) + .await } async fn scope_rules_overlap( &self, request: workdir::WorkdirScopeOverlapRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self - .operate(WorkdirSessionOperation::ScopeRulesOverlap(request)) - .await? - { - WorkdirSessionOperationResult::ScopeRulesOverlap { overlaps } => Ok(overlaps), - _ => Err(Self::mismatch("scope_rules_overlap")), - } + self.operate( + WorkdirSessionOperation::ScopeRulesOverlap(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::ScopeRulesOverlap { overlaps } => Ok(overlaps), + _ => Err(Self::mismatch("scope_rules_overlap")), + } + }, + ) + .await } async fn checkout_observe( &self, path: workdir::WorkdirPath, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self - .operate(WorkdirSessionOperation::CheckoutObserve(path.clone())) - .await? - { - WorkdirSessionOperationResult::CheckoutObserve(mut value) - if value.path == path - && !value.validator.is_empty() - && value.validator.len() <= 256 => - { - value.capabilities = value.capabilities.intersection(self.capabilities()); - Ok(value) - } - _ => Err(Self::mismatch("checkout_observe")), - } + self.operate( + WorkdirSessionOperation::CheckoutObserve(path.clone()), + |result| async move { + match result? { + WorkdirSessionOperationResult::CheckoutObserve(mut value) + if value.path == path + && !value.validator.is_empty() + && value.validator.len() <= 256 => + { + value.capabilities = value.capabilities.intersection(self.capabilities()); + Ok(value) + } + _ => Err(Self::mismatch("checkout_observe")), + } + }, + ) + .await } async fn checkout_execute( &self, @@ -1433,63 +1461,67 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { self.ensure_capability(cap)?; let target = request.target.clone(); let mutation = cap != workdir::WorkdirSessionCapability::Read; - match self - .operate(WorkdirSessionOperation::CheckoutExecute(request)) - .await - { - Ok(WorkdirSessionOperationResult::CheckoutExecute(mut value)) - if value.observation.path == target - && !value.observation.validator.is_empty() - && value.observation.validator.len() <= 256 - && matches!( - (&value.output, cap), - ( - workdir::CheckoutOutput::Read(_), - workdir::WorkdirSessionCapability::Read - ) | ( - workdir::CheckoutOutput::Write(_), - workdir::WorkdirSessionCapability::Write - ) | ( - workdir::CheckoutOutput::Edit(_), - workdir::WorkdirSessionCapability::Edit - ) - ) => - { - value.observation.capabilities = value - .observation - .capabilities - .intersection(self.capabilities()); - Ok(value) - } - Ok(_) if mutation => Err(workdir::WorkdirError::OutcomeUnknown( - "mismatched checkout mutation response".into(), - ) - .with_failure_reason(workdir::WorkdirFailureReason::OperationResultMismatch)), - Ok(_) => Err(Self::mismatch("checkout_execute")), - Err(error) - if mutation - && matches!( - error.classification_source(), - workdir::WorkdirError::Unavailable(_) | workdir::WorkdirError::Transport(_) - ) => - { - let outcome = workdir::WorkdirError::OutcomeUnknown( - "checkout provider response lost; inspect effects before retry".into(), - ); - let outcome = match error.failure_reason() { - Some(reason) => outcome.with_failure_reason(reason), - None => outcome, - }; - Err(match error.denial_reason() { - Some(reason) => workdir::WorkdirError::DenialContext { - reason, - source: Box::new(outcome), - }, - None => outcome, - }) - } - Err(error) => Err(error), - } + self.operate( + WorkdirSessionOperation::CheckoutExecute(request), + |result| async move { + match result { + Ok(WorkdirSessionOperationResult::CheckoutExecute(mut value)) + if value.observation.path == target + && !value.observation.validator.is_empty() + && value.observation.validator.len() <= 256 + && matches!( + (&value.output, cap), + ( + workdir::CheckoutOutput::Read(_), + workdir::WorkdirSessionCapability::Read + ) | ( + workdir::CheckoutOutput::Write(_), + workdir::WorkdirSessionCapability::Write + ) | ( + workdir::CheckoutOutput::Edit(_), + workdir::WorkdirSessionCapability::Edit + ) + ) => + { + value.observation.capabilities = value + .observation + .capabilities + .intersection(self.capabilities()); + Ok(value) + } + Ok(_) if mutation => Err(workdir::WorkdirError::OutcomeUnknown( + "mismatched checkout mutation response".into(), + ) + .with_failure_reason(workdir::WorkdirFailureReason::OperationResultMismatch)), + Ok(_) => Err(Self::mismatch("checkout_execute")), + Err(error) + if mutation + && matches!( + error.classification_source(), + workdir::WorkdirError::Unavailable(_) + | workdir::WorkdirError::Transport(_) + ) => + { + let outcome = workdir::WorkdirError::OutcomeUnknown( + "checkout provider response lost; inspect effects before retry".into(), + ); + let outcome = match error.failure_reason() { + Some(reason) => outcome.with_failure_reason(reason), + None => outcome, + }; + Err(match error.denial_reason() { + Some(reason) => workdir::WorkdirError::DenialContext { + reason, + source: Box::new(outcome), + }, + None => outcome, + }) + } + Err(error) => Err(error), + } + }, + ) + .await } async fn checkout_search( &self, @@ -1505,108 +1537,153 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { Op::Grep(_) => Cap::Grep, }; self.ensure_capability(cap)?; - match self - .operate(WorkdirSessionOperation::CheckoutSearch(request)) - .await? - { - WorkdirSessionOperationResult::CheckoutSearch(value) - if matches!( - (&value, cap), - (Result::List(_), Cap::Read) - | (Result::Glob(_), Cap::Glob) - | (Result::Grep(_), Cap::Grep) - ) => - { - Ok(value) - } - _ => Err(Self::mismatch("checkout_search")), - } + self.operate( + WorkdirSessionOperation::CheckoutSearch(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::CheckoutSearch(value) + if matches!( + (&value, cap), + (Result::List(_), Cap::Read) + | (Result::Glob(_), Cap::Glob) + | (Result::Grep(_), Cap::Grep) + ) => + { + Ok(value) + } + _ => Err(Self::mismatch("checkout_search")), + } + }, + ) + .await } async fn stat( &self, request: workdir::StatRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self.operate(WorkdirSessionOperation::Stat(request)).await? { - WorkdirSessionOperationResult::Stat(value) => Ok(value), - _ => Err(Self::mismatch("stat")), - } + self.operate( + WorkdirSessionOperation::Stat(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Stat(value) => Ok(value), + _ => Err(Self::mismatch("stat")), + } + }, + ) + .await } async fn read( &self, request: workdir::ReadRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self.operate(WorkdirSessionOperation::Read(request)).await? { - WorkdirSessionOperationResult::Read(value) => Ok(value), - _ => Err(Self::mismatch("read")), - } + self.operate( + WorkdirSessionOperation::Read(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Read(value) => Ok(value), + _ => Err(Self::mismatch("read")), + } + }, + ) + .await } async fn read_bytes( &self, request: workdir::ReadBytesRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self - .operate(WorkdirSessionOperation::ReadBytes(request)) - .await? - { - WorkdirSessionOperationResult::ReadBytes(value) => Ok(value), - _ => Err(Self::mismatch("read_bytes")), - } + self.operate( + WorkdirSessionOperation::ReadBytes(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::ReadBytes(value) => Ok(value), + _ => Err(Self::mismatch("read_bytes")), + } + }, + ) + .await } async fn write( &self, request: workdir::WriteRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Write)?; - match self - .operate(WorkdirSessionOperation::Write(request)) - .await? - { - WorkdirSessionOperationResult::Write(value) => Ok(value), - _ => Err(Self::mismatch("write")), - } + self.operate( + WorkdirSessionOperation::Write(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Write(value) => Ok(value), + _ => Err(Self::mismatch("write")), + } + }, + ) + .await } async fn edit( &self, request: workdir::EditRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Edit)?; - match self.operate(WorkdirSessionOperation::Edit(request)).await? { - WorkdirSessionOperationResult::Edit(value) => Ok(value), - _ => Err(Self::mismatch("edit")), - } + self.operate( + WorkdirSessionOperation::Edit(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Edit(value) => Ok(value), + _ => Err(Self::mismatch("edit")), + } + }, + ) + .await } async fn list( &self, request: workdir::ListRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Read)?; - match self.operate(WorkdirSessionOperation::List(request)).await? { - WorkdirSessionOperationResult::List(value) => Ok(value), - _ => Err(Self::mismatch("list")), - } + self.operate( + WorkdirSessionOperation::List(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::List(value) => Ok(value), + _ => Err(Self::mismatch("list")), + } + }, + ) + .await } async fn glob( &self, request: workdir::GlobRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Glob)?; - match self.operate(WorkdirSessionOperation::Glob(request)).await? { - WorkdirSessionOperationResult::Glob(value) => Ok(value), - _ => Err(Self::mismatch("glob")), - } + self.operate( + WorkdirSessionOperation::Glob(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Glob(value) => Ok(value), + _ => Err(Self::mismatch("glob")), + } + }, + ) + .await } async fn grep( &self, request: workdir::GrepRequest, ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Grep)?; - match self.operate(WorkdirSessionOperation::Grep(request)).await? { - WorkdirSessionOperationResult::Grep(value) => Ok(value), - _ => Err(Self::mismatch("grep")), - } + self.operate( + WorkdirSessionOperation::Grep(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::Grep(value) => Ok(value), + _ => Err(Self::mismatch("grep")), + } + }, + ) + .await } async fn start_command( &self, @@ -1631,62 +1708,68 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { "External Workdir active command limit reached".to_string(), )); } - match self - .operate(WorkdirSessionOperation::CommandStart(request)) - .await? - { - WorkdirSessionOperationResult::CommandStart(provider_handle) => { - let duplicate_handles = { - let commands = self.active_commands.lock().await; - commands - .iter() - .filter(|(_, active)| active.handle == provider_handle) - .map(|(external, _)| external.clone()) - .collect::>() - }; - if !duplicate_handles.is_empty() { - // A provider handle must identify exactly one command in one - // generation. Remove every ambiguous route before attempting - // cleanup so no caller can address the wrong process. - let mut commands = self.active_commands.lock().await; - let mut quarantined = None; - for external_handle in duplicate_handles { - if let Some(command) = commands.remove(&external_handle) { - quarantined.get_or_insert(command); + self.operate( + WorkdirSessionOperation::CommandStart(request), + |result| async move { + match result? { + WorkdirSessionOperationResult::CommandStart(provider_handle) => { + let duplicate_handles = { + let commands = self.active_commands.lock().await; + commands + .iter() + .filter(|(_, active)| active.handle == provider_handle) + .map(|(external, _)| external.clone()) + .collect::>() + }; + if !duplicate_handles.is_empty() { + // A provider handle must identify exactly one command in one + // generation. Remove every ambiguous route before attempting + // cleanup so no caller can address the wrong process. + let mut commands = self.active_commands.lock().await; + let mut quarantined = None; + for external_handle in duplicate_handles { + if let Some(command) = commands.remove(&external_handle) { + quarantined.get_or_insert(command); + } + } + let quarantine_handle = workdir::CommandHandle(format!( + "quarantined-command-{}", + Uuid::now_v7() + )); + let command = quarantined.expect("duplicate command route disappeared"); + *command.terminal_at.lock().unwrap() = None; + commands.insert(quarantine_handle.0.clone(), command.clone()); + drop(commands); + if self + .terminate_provider_command(command.clone(), true) + .await + .is_ok() + { + self.remove_active_command_if_matches(&quarantine_handle, &command) + .await; + } else { + // An earlier read of the reused handle cannot establish that + // the newly started, ambiguous process has been cleaned up. + *command.terminal_at.lock().unwrap() = None; + } + return Err(workdir::WorkdirError::Unavailable( + "External Workdir provider reused an active command handle" + .to_string(), + )); } + let external_handle = + workdir::CommandHandle(format!("external-command-{}", Uuid::now_v7())); + self.active_commands.lock().await.insert( + external_handle.0.clone(), + Arc::new(ExternalProviderCommandState::new(provider_handle)), + ); + Ok(external_handle) } - let quarantine_handle = - workdir::CommandHandle(format!("quarantined-command-{}", Uuid::now_v7())); - let command = quarantined.expect("duplicate command route disappeared"); - *command.terminal_at.lock().unwrap() = None; - commands.insert(quarantine_handle.0.clone(), command.clone()); - drop(commands); - if self - .terminate_provider_command(command.clone(), true) - .await - .is_ok() - { - self.remove_active_command_if_matches(&quarantine_handle, &command) - .await; - } else { - // An earlier read of the reused handle cannot establish that - // the newly started, ambiguous process has been cleaned up. - *command.terminal_at.lock().unwrap() = None; - } - return Err(workdir::WorkdirError::Unavailable( - "External Workdir provider reused an active command handle".to_string(), - )); + _ => Err(Self::mismatch("command_start")), } - let external_handle = - workdir::CommandHandle(format!("external-command-{}", Uuid::now_v7())); - self.active_commands.lock().await.insert( - external_handle.0.clone(), - Arc::new(ExternalProviderCommandState::new(provider_handle)), - ); - Ok(external_handle) - } - _ => Err(Self::mismatch("command_start")), - } + }, + ) + .await } async fn command_status( &self, @@ -1694,13 +1777,16 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { ) -> std::result::Result { self.ensure_capability(workdir::WorkdirSessionCapability::Command)?; let provider_handle = self.provider_command_handle(&external_handle).await?; - match self - .operate(WorkdirSessionOperation::CommandStatus(provider_handle)) - .await? - { - WorkdirSessionOperationResult::CommandStatus(status) => Ok(status), - _ => Err(Self::mismatch("command_status")), - } + self.operate( + WorkdirSessionOperation::CommandStatus(provider_handle), + |result| async move { + match result? { + WorkdirSessionOperationResult::CommandStatus(status) => Ok(status), + _ => Err(Self::mismatch("command_status")), + } + }, + ) + .await } async fn command_output( &self, @@ -1709,29 +1795,33 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { self.ensure_capability(workdir::WorkdirSessionCapability::Command)?; let external_handle = request.handle.clone(); let command = self.provider_command(&external_handle).await?; + let command = &command; let mut cursor = command.cursor.lock().await; request.handle = command.handle.clone(); // If the response is lost, this request cursor still replays the provider page. - match self - .operate(WorkdirSessionOperation::CommandOutput(request)) - .await - { - Ok(WorkdirSessionOperationResult::CommandOutput(output)) => { - *cursor = output.next_cursor; - if output.terminal.is_some() { - command - .terminal_at - .lock() - .unwrap() - .get_or_insert_with(std::time::Instant::now); + self.operate( + WorkdirSessionOperation::CommandOutput(request), + |result| async move { + match result { + Ok(WorkdirSessionOperationResult::CommandOutput(output)) => { + *cursor = output.next_cursor; + if output.terminal.is_some() { + command + .terminal_at + .lock() + .unwrap() + .get_or_insert_with(std::time::Instant::now); + } + drop(cursor); + self.prune_terminal_commands().await; + Ok(output) + } + Ok(_) => Err(Self::mismatch("command_output")), + Err(error) => Err(error), } - drop(cursor); - self.prune_terminal_commands().await; - Ok(output) - } - Ok(_) => Err(Self::mismatch("command_output")), - Err(error) => Err(error), - } + }, + ) + .await } async fn cancel_command( &self, @@ -1739,13 +1829,16 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { ) -> std::result::Result<(), workdir::WorkdirError> { self.ensure_capability(workdir::WorkdirSessionCapability::Command)?; let provider_handle = self.provider_command_handle(&external_handle).await?; - match self - .operate(WorkdirSessionOperation::CommandCancel(provider_handle)) - .await? - { - WorkdirSessionOperationResult::CommandCancel => Ok(()), - _ => Err(Self::mismatch("command_cancel")), - } + self.operate( + WorkdirSessionOperation::CommandCancel(provider_handle), + |result| async move { + match result? { + WorkdirSessionOperationResult::CommandCancel => Ok(()), + _ => Err(Self::mismatch("command_cancel")), + } + }, + ) + .await } async fn close(&self) -> std::result::Result<(), workdir::WorkdirError> { let _close_guard = self.close_lock.lock().await; diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs index a6f76f548..bdf75c837 100644 --- a/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs @@ -46,7 +46,10 @@ struct ProviderProbe { } impl ProviderProbe { - async fn receive_write(&mut self, pending: &mut PendingHttpWrite) -> (String, ProviderReply) { + async fn receive_operation( + &mut self, + pending: &mut PendingHttpWrite, + ) -> (String, WorkdirSessionOperation, ProviderReply) { let command = tokio::time::timeout(BOUND, async { tokio::select! { command = self.receiver.recv() => command.expect("provider channel closed"), @@ -66,6 +69,11 @@ impl ProviderProbe { panic!("expected write dispatch, received cancellation") }; self.dispatches += 1; + (operation_id.as_str().to_owned(), operation, response) + } + + async fn receive_write(&mut self, pending: &mut PendingHttpWrite) -> (String, ProviderReply) { + let (operation_id, operation, response) = self.receive_operation(pending).await; let WorkdirSessionOperation::Write(request) = operation else { panic!("expected mutation, got {operation:?}") }; @@ -75,7 +83,7 @@ impl ProviderProbe { ); assert_eq!(request.content, WRITE_SECRET); assert_eq!(request.expected_hash, None); - (operation_id.as_str().to_owned(), response) + (operation_id, response) } fn assert_dispatches(&mut self, expected: usize) { @@ -200,14 +208,18 @@ impl AuditHttpFixture { } fn start_write(&self) -> PendingHttpWrite { + self.start_operation(WorkdirSessionOperation::Write(workdir::WriteRequest { + path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(), + content: WRITE_SECRET.to_vec(), + expected_hash: None, + })) + } + + fn start_operation(&self, operation: WorkdirSessionOperation) -> PendingHttpWrite { let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations"); let body = serde_json::to_vec(&WorkspaceWorkdirSessionOperationRequest { target_workdir: "attachment".into(), - operation: WorkdirSessionOperation::Write(workdir::WriteRequest { - path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(), - content: WRITE_SECRET.to_vec(), - expected_hash: None, - }), + operation, }) .unwrap(); let request = self.fixture.request( @@ -259,6 +271,7 @@ impl AuditHttpFixture { encoded_content.as_str(), PATH_SECRET, PROVIDER_SECRET, + PROVIDER_SECRET.trim_start_matches('/'), pending.proof.as_str(), worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, self.fixture.identity.private_key.as_str(), @@ -276,6 +289,10 @@ impl AuditHttpFixture { } fn assert_audits(&self, expected: &[(&str, &str)]) { + self.assert_operation_audits(expected, "write"); + } + + fn assert_operation_audits(&self, expected: &[(&str, &str)], operation: &str) { let connection = rusqlite::Connection::open(&self.fixture.api.config.database_path).unwrap(); let mut statement = connection @@ -298,7 +315,10 @@ impl AuditHttpFixture { .collect(); assert_eq!(actual, expected, "persisted audits: {rows:?}"); for (_, _, summary) in rows { - assert!(summary.contains("operation_kind=write"), "{summary}"); + assert!( + summary.contains(&format!("operation_kind={operation}")), + "{summary}" + ); for secret in [ AUDIT_SECRET, PATH_SECRET, @@ -697,3 +717,169 @@ async fn provider_reported_unknown_preserves_received_response_and_unknown_effec provider.assert_dispatches(1); h.assert_audits(&[(&dispatched_id, "requested")]); } + +#[derive(Clone, Copy, Debug)] +enum CheckoutReplyFault { + WrongOperation, + WrongPath, + EmptyValidator, + WrongOutput, + ProviderUnavailable, + ProviderTransport, +} + +async fn assert_checkout_unknown_audit(fault: CheckoutReplyFault, fail_completion: bool) { + let (h, mut provider) = AuditHttpFixture::new().await; + if fail_completion { + h.fail_audit(false); + } + let target = workdir::WorkdirPath::new(PATH_SECRET).unwrap(); + let mut pending = h.start_operation(WorkdirSessionOperation::CheckoutExecute( + workdir::CheckoutRequest { + target: target.clone(), + validator: vec![1], + operation: workdir::CheckoutOperation::Write { + content: WRITE_SECRET.to_vec(), + expected_hash: [0; 32], + }, + }, + )); + let (operation_id, operation, reply) = provider.receive_operation(&mut pending).await; + let WorkdirSessionOperation::CheckoutExecute(request) = operation else { + panic!("expected checkout mutation, got {operation:?}"); + }; + assert_eq!(request.target, target); + assert_eq!(request.validator, vec![1]); + assert!( + matches!(request.operation, workdir::CheckoutOperation::Write { ref content, .. } if content == WRITE_SECRET) + ); + h.apply_provider_write(); + let result = match fault { + CheckoutReplyFault::WrongOperation => Ok(write_result()), + CheckoutReplyFault::ProviderUnavailable => Err(WorkdirTransportError::from_workdir_error( + &workdir::WorkdirError::Unavailable(PROVIDER_SECRET.into()), + )), + CheckoutReplyFault::ProviderTransport => Err(WorkdirTransportError::from_workdir_error( + &workdir::WorkdirError::Transport(PROVIDER_SECRET.into()), + )), + _ => { + let mut result = workdir::CheckoutResult { + observation: workdir::CheckoutObservation { + path: target, + validator: vec![2], + kind: workdir::EntryKind::File, + size: WRITE_SECRET.len() as u64, + capabilities: workdir::WorkdirSessionCapabilities::READ_WRITE, + }, + output: workdir::CheckoutOutput::Write(workdir::WriteResult { + bytes_written: WRITE_SECRET.len(), + created: false, + }), + }; + match fault { + CheckoutReplyFault::WrongPath => { + result.observation.path = + workdir::WorkdirPath::new(PROVIDER_SECRET.trim_start_matches('/')).unwrap() + } + CheckoutReplyFault::EmptyValidator => result.observation.validator.clear(), + CheckoutReplyFault::WrongOutput => { + result.output = workdir::CheckoutOutput::Edit(workdir::EditResult { + replacements: 1, + content_hash: [0; 32], + bytes_written: 1, + }) + } + _ => unreachable!(), + } + Ok(WorkdirSessionOperationResult::CheckoutExecute(result)) + } + }; + reply.send(result).unwrap(); + let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!(body["code"], "outcome_unknown", "{fault:?}: {body}"); + assert_eq!( + body["message"], + "Workdir operation may have effects; inspect before retrying" + ); + if !matches!( + fault, + CheckoutReplyFault::ProviderUnavailable | CheckoutReplyFault::ProviderTransport + ) { + assert_failure_reason(&events, "operation_result_mismatch", "response_validation"); + } + if fail_completion { + assert_eq!( + body["audit_failures"], + json!([{ + "operation_id": operation_id, + "stage": "response_received", + "reason": "audit_store_unavailable", + "operation_outcome": {"kind":"unknown"}, + }]), + "{fault:?}: {body}" + ); + let audit_logs: Vec<_> = events + .iter() + .filter(|e| e["event"] == "external_workdir_audit_failed") + .collect(); + assert_eq!(audit_logs.len(), 1, "{events:?}"); + let event = audit_logs[0]; + assert_eq!(event["operation_id"], operation_id); + assert_eq!(event["operation"], "checkout_execute"); + assert_eq!(event["audit_stage"], "ResponseReceived"); + assert_eq!(event["operation_outcome"], "Unknown"); + h.assert_operation_audits(&[(&operation_id, "requested")], "checkout_execute"); + } else { + assert!(body.get("audit_failures").is_none(), "{body}"); + assert!( + events + .iter() + .all(|e| e["event"] != "external_workdir_audit_failed") + ); + h.assert_operation_audits( + &[ + (&operation_id, "requested"), + (&operation_id, "outcome_unknown"), + ], + "checkout_execute", + ); + } + assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET); + provider.assert_dispatches(1); +} + +#[tokio::test] +async fn checkout_invalid_reply_records_unknown_in_healthy_completion_audit() { + for fault in [ + CheckoutReplyFault::WrongOperation, + CheckoutReplyFault::WrongPath, + CheckoutReplyFault::EmptyValidator, + CheckoutReplyFault::WrongOutput, + ] { + assert_checkout_unknown_audit(fault, false).await; + } +} + +#[tokio::test] +async fn checkout_invalid_reply_records_unknown_in_failed_completion_receipt_and_log() { + for fault in [ + CheckoutReplyFault::WrongOperation, + CheckoutReplyFault::WrongPath, + CheckoutReplyFault::EmptyValidator, + CheckoutReplyFault::WrongOutput, + ] { + assert_checkout_unknown_audit(fault, true).await; + } +} + +#[tokio::test] +async fn checkout_uncertain_reply_normalization_matches_audit_outcome() { + for fault in [ + CheckoutReplyFault::ProviderUnavailable, + CheckoutReplyFault::ProviderTransport, + ] { + for fail_completion in [false, true] { + assert_checkout_unknown_audit(fault, fail_completion).await; + } + } +}