fix: audit validated external workdir operation outcomes
This commit is contained in:
@@ -1046,10 +1046,17 @@ impl ExternalProviderWorkdirSession {
|
||||
})
|
||||
}
|
||||
|
||||
async fn operate(
|
||||
async fn operate<T, F, Fut>(
|
||||
&self,
|
||||
operation: WorkdirSessionOperation,
|
||||
) -> std::result::Result<WorkdirSessionOperationResult, workdir::WorkdirError> {
|
||||
validate: F,
|
||||
) -> std::result::Result<T, workdir::WorkdirError>
|
||||
where
|
||||
T: Send,
|
||||
F: FnOnce(std::result::Result<WorkdirSessionOperationResult, workdir::WorkdirError>) -> Fut
|
||||
+ Send,
|
||||
Fut: std::future::Future<Output = std::result::Result<T, workdir::WorkdirError>> + 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<bool, workdir::WorkdirError> {
|
||||
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<workdir::CheckoutObservation, workdir::WorkdirError> {
|
||||
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<workdir::StatResult, workdir::WorkdirError> {
|
||||
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<workdir::ReadResult, workdir::WorkdirError> {
|
||||
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<workdir::ReadBytesResult, workdir::WorkdirError> {
|
||||
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<workdir::WriteResult, workdir::WorkdirError> {
|
||||
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<workdir::EditResult, workdir::WorkdirError> {
|
||||
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<workdir::ListResult, workdir::WorkdirError> {
|
||||
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<workdir::GlobResult, workdir::WorkdirError> {
|
||||
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<workdir::GrepResult, workdir::WorkdirError> {
|
||||
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::<Vec<_>>()
|
||||
};
|
||||
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::<Vec<_>>()
|
||||
};
|
||||
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<workdir::CommandStatus, workdir::WorkdirError> {
|
||||
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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user