fix: retain independent Bash Drop cleanup diagnostics
This commit is contained in:
+15
-74
@@ -19,6 +19,8 @@ use workdir::{
|
||||
WorkdirSessionRouter,
|
||||
};
|
||||
|
||||
mod cleanup;
|
||||
|
||||
const DEFAULT_TIMEOUT_SECS: u64 = 120;
|
||||
const MAX_TIMEOUT_SECS: u64 = 600;
|
||||
const INLINE_BYTE_BUDGET: usize = 12 * 1024;
|
||||
@@ -72,80 +74,19 @@ impl Drop for CommandGuard {
|
||||
let Some(handle) = self.handle.take() else {
|
||||
return;
|
||||
};
|
||||
let workdir = self.session.clone();
|
||||
let state = Arc::clone(&self.state);
|
||||
let execution_id = self.execution_id.clone();
|
||||
let execution_nonce = self.execution_nonce;
|
||||
let mut retained = self.retained.take();
|
||||
let cursor = self.cursor;
|
||||
// A dropped provider future is not terminal confirmation. Keep the live
|
||||
// execution registered until cleanup has both requested cancellation and
|
||||
// observed terminal command output, so cancellation/session teardown
|
||||
// cannot race with an apparently empty registry.
|
||||
tokio::spawn(async move {
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
||||
let _ = workdir.cancel_command(handle.clone()).await;
|
||||
let mut cursor = cursor;
|
||||
loop {
|
||||
let page = command_page(
|
||||
&workdir,
|
||||
CommandOutputRequest {
|
||||
handle: handle.clone(),
|
||||
cursor,
|
||||
limit: workdir::MAX_COMMAND_READ_BYTES,
|
||||
wait: true,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
if let Some(writer) = &mut retained {
|
||||
let mut next = cursor;
|
||||
for chunk in &page.chunks {
|
||||
if chunk.sequence != next {
|
||||
return Err(workdir::WorkdirError::Unavailable(
|
||||
"command cleanup sequence gap".into(),
|
||||
));
|
||||
}
|
||||
writer
|
||||
.append_chunk(chunk)
|
||||
.map_err(workdir::WorkdirError::Unavailable)?;
|
||||
next = next.checked_add(chunk.bytes.len() as u64).ok_or_else(|| {
|
||||
workdir::WorkdirError::Unavailable(
|
||||
"command cleanup cursor overflow".into(),
|
||||
)
|
||||
})?;
|
||||
}
|
||||
if page.next_cursor != next {
|
||||
return Err(workdir::WorkdirError::Unavailable(
|
||||
"command cleanup cursor gap".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
cursor = page.next_cursor;
|
||||
if let Some(terminal) = page.terminal {
|
||||
if let Some(writer) = &mut retained {
|
||||
writer
|
||||
.finish_transport(&terminal)
|
||||
.map_err(workdir::WorkdirError::Unavailable)?;
|
||||
}
|
||||
return Ok::<(), workdir::WorkdirError>(());
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
// Bounded cleanup can end with an unknown provider outcome. The writer then
|
||||
// terminalizes partial evidence on drop; retaining a dead collector in `active`
|
||||
// would leak registrations and falsely suggest it could still collect output.
|
||||
let _ = result;
|
||||
let mut state = state.lock().unwrap();
|
||||
if state
|
||||
.active
|
||||
.get(&execution_id)
|
||||
.is_some_and(|active| active.execution_nonce == execution_nonce)
|
||||
{
|
||||
state.active.remove(&execution_id);
|
||||
state.cancellation_requested.remove(&execution_id);
|
||||
}
|
||||
});
|
||||
let cleanup = cleanup::Cleanup {
|
||||
session: self.session.clone(),
|
||||
state: Arc::clone(&self.state),
|
||||
execution_id: self.execution_id.clone(),
|
||||
execution_nonce: self.execution_nonce,
|
||||
handle,
|
||||
cursor: self.cursor,
|
||||
retained: self
|
||||
.retained
|
||||
.take()
|
||||
.expect("started command owns output writer"),
|
||||
};
|
||||
tokio::spawn(cleanup.run());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
//! Finite Drop cleanup. A cancel acknowledgement is not termination evidence; only a
|
||||
//! validated terminal fence confirms the process result. No provider text is logged.
|
||||
|
||||
use super::*;
|
||||
use crate::bash_output::{
|
||||
CleanupCancel, CleanupDiagnostics, CleanupDrain, CleanupFailure, CleanupProviderFailure,
|
||||
CleanupRegistration,
|
||||
};
|
||||
use std::time::Duration;
|
||||
|
||||
const CLEANUP_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
// Reserve time for independent terminal observation even if cancel never responds.
|
||||
const CANCEL_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
|
||||
pub(super) struct Cleanup {
|
||||
pub session: WorkdirSessionHandle,
|
||||
pub state: Arc<Mutex<BashExecutionState>>,
|
||||
pub execution_id: String,
|
||||
pub execution_nonce: u64,
|
||||
pub handle: CommandHandle,
|
||||
pub cursor: u64,
|
||||
pub retained: BashOutputWriter,
|
||||
}
|
||||
|
||||
impl Cleanup {
|
||||
pub async fn run(mut self) {
|
||||
let deadline = tokio::time::Instant::now() + CLEANUP_TIMEOUT;
|
||||
let cancel = match tokio::time::timeout(
|
||||
CANCEL_TIMEOUT,
|
||||
self.session.cancel_command(self.handle.clone()),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Ok(())) => CleanupCancel::Acknowledged,
|
||||
Ok(Err(error)) => CleanupCancel::Failed {
|
||||
reason: CleanupProviderFailure::from(&error),
|
||||
},
|
||||
Err(_) => CleanupCancel::DeadlineExceeded,
|
||||
};
|
||||
let drain = match tokio::time::timeout_at(deadline, self.drain()).await {
|
||||
Ok(Ok(())) => CleanupDrain::TerminalConfirmed,
|
||||
Ok(Err(reason)) => CleanupDrain::Failed { reason },
|
||||
Err(_) => CleanupDrain::DeadlineExceeded,
|
||||
};
|
||||
let deadline_expired = matches!(drain, CleanupDrain::DeadlineExceeded);
|
||||
// Finalize evidence before exposing an empty registry. Keep the lock through
|
||||
// publication so an older cleanup can never remove a newer generation.
|
||||
let mut state = self.state.lock().unwrap();
|
||||
let registration = if state
|
||||
.active
|
||||
.get(&self.execution_id)
|
||||
.is_some_and(|active| active.execution_nonce == self.execution_nonce)
|
||||
{
|
||||
state.active.remove(&self.execution_id);
|
||||
state.cancellation_requested.remove(&self.execution_id);
|
||||
CleanupRegistration::Released
|
||||
} else {
|
||||
CleanupRegistration::Superseded
|
||||
};
|
||||
let diagnostics = CleanupDiagnostics {
|
||||
execution_nonce: self.execution_nonce,
|
||||
cancel,
|
||||
drain,
|
||||
deadline_expired,
|
||||
registration,
|
||||
};
|
||||
tracing::info!(
|
||||
output_id = %self.retained.id(),
|
||||
cleanup = ?diagnostics,
|
||||
"Bash Drop cleanup finished; cancellation and terminal observation are independent"
|
||||
);
|
||||
self.retained.finish_cleanup(diagnostics);
|
||||
drop(self.retained);
|
||||
}
|
||||
|
||||
async fn drain(&mut self) -> Result<(), CleanupFailure> {
|
||||
loop {
|
||||
let page = command_page(
|
||||
&self.session,
|
||||
CommandOutputRequest {
|
||||
handle: self.handle.clone(),
|
||||
cursor: self.cursor,
|
||||
limit: workdir::MAX_COMMAND_READ_BYTES,
|
||||
wait: true,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.map_err(|error| CleanupFailure::Provider {
|
||||
reason: CleanupProviderFailure::from(&error),
|
||||
})?;
|
||||
let mut next = self.cursor;
|
||||
for chunk in &page.chunks {
|
||||
if chunk.sequence != next {
|
||||
return Err(CleanupFailure::SequenceGap);
|
||||
}
|
||||
next = next
|
||||
.checked_add(chunk.bytes.len() as u64)
|
||||
.filter(|next| *next <= crate::bash_output::COMMAND_BYTES as u64)
|
||||
.ok_or(CleanupFailure::DeliveryBudgetExceeded)?;
|
||||
self.retained
|
||||
.append_chunk(chunk)
|
||||
.map_err(|_| CleanupFailure::RetainedChunkInvalid)?;
|
||||
}
|
||||
if page.next_cursor != next {
|
||||
return Err(CleanupFailure::CursorGap);
|
||||
}
|
||||
self.cursor = next;
|
||||
if let Some(terminal) = page.terminal {
|
||||
self.retained
|
||||
.finish_transport(&terminal)
|
||||
.map_err(|_| CleanupFailure::TerminalMetadataInvalid)?;
|
||||
return Ok(());
|
||||
}
|
||||
// An immediately-ready provider must not starve the cleanup deadline.
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,397 @@
|
||||
//! Component tests enter through the production CommandGuard Drop, cleanup, page
|
||||
//! validation and retained store. Only the Workdir provider is replaced; paused time
|
||||
//! and a cancel barrier make faults/deadlines deterministic without OS timing.
|
||||
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use workdir::*;
|
||||
|
||||
const PRIVATE: &str = "SECRET command /private/host/path output password=poison";
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum Fault {
|
||||
Cancel,
|
||||
CancelHang,
|
||||
Drain,
|
||||
TerminalTimeout,
|
||||
Sequence,
|
||||
Cursor,
|
||||
RetainedChunk,
|
||||
TerminalMetadata,
|
||||
CollectionIncomplete,
|
||||
Complete,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Provider {
|
||||
local: LocalWorkdirSession,
|
||||
fault: Fault,
|
||||
cancelling: tokio::sync::Notify,
|
||||
allow_drain: AtomicBool,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl WorkdirSession for Provider {
|
||||
fn workdir(&self) -> &Workdir {
|
||||
self.local.workdir()
|
||||
}
|
||||
fn capabilities(&self) -> WorkdirSessionCapabilities {
|
||||
WorkdirSessionCapabilities::ALL
|
||||
}
|
||||
async fn stat(&self, _: StatRequest) -> Result<StatResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn read(&self, _: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn write(&self, _: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn edit(&self, _: EditRequest) -> Result<EditResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn list(&self, _: ListRequest) -> Result<ListResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn glob(&self, _: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn grep(&self, _: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn start_command(&self, _: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
async fn command_status(&self, _: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
||||
Err(WorkdirError::OperationFailed)
|
||||
}
|
||||
|
||||
async fn command_output(
|
||||
&self,
|
||||
request: CommandOutputRequest,
|
||||
) -> Result<CommandOutput, WorkdirError> {
|
||||
assert_eq!(request.cursor, 3);
|
||||
while !self.allow_drain.load(Ordering::Acquire) {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
match self.fault {
|
||||
Fault::Drain => {
|
||||
return Err(WorkdirError::Transport(PRIVATE.into())
|
||||
.with_failure_reason(WorkdirFailureReason::HttpTransfer));
|
||||
}
|
||||
Fault::TerminalTimeout => return std::future::pending().await,
|
||||
_ => {}
|
||||
}
|
||||
let mut chunk = CommandByteChunk {
|
||||
sequence: 3,
|
||||
offset: 3,
|
||||
stream: CommandStream::Stdout,
|
||||
bytes: b"tail".to_vec(),
|
||||
};
|
||||
if matches!(self.fault, Fault::Sequence) {
|
||||
chunk.sequence = 4;
|
||||
}
|
||||
if matches!(self.fault, Fault::RetainedChunk) {
|
||||
chunk.offset = 42;
|
||||
}
|
||||
let mut terminal = terminal();
|
||||
if matches!(self.fault, Fault::TerminalMetadata) {
|
||||
terminal.stdout_delivered_bytes = 6;
|
||||
}
|
||||
if matches!(self.fault, Fault::CollectionIncomplete) {
|
||||
terminal.output_collection = CommandOutputCollection::Incomplete {
|
||||
failure: CommandFailure {
|
||||
stage: CommandFailureStage::StdoutDrain,
|
||||
reason: CommandIoErrorKind::Other,
|
||||
os_code: None,
|
||||
},
|
||||
};
|
||||
}
|
||||
Ok(CommandOutput {
|
||||
chunks: vec![chunk],
|
||||
next_cursor: if matches!(self.fault, Fault::Cursor) {
|
||||
8
|
||||
} else {
|
||||
7
|
||||
},
|
||||
terminal: Some(terminal),
|
||||
})
|
||||
}
|
||||
|
||||
async fn cancel_command(&self, _: CommandHandle) -> Result<(), WorkdirError> {
|
||||
self.cancelling.notify_one();
|
||||
match self.fault {
|
||||
Fault::Cancel => Err(WorkdirError::Unavailable(PRIVATE.into())),
|
||||
Fault::CancelHang => std::future::pending().await,
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
async fn close(&self) -> Result<(), WorkdirError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn terminal() -> CommandTerminal {
|
||||
CommandTerminal {
|
||||
status: CommandStatus::Completed,
|
||||
exit_code: Some(0),
|
||||
process_failure: None,
|
||||
output_collection: CommandOutputCollection::Complete,
|
||||
stdout_bytes: 7,
|
||||
stderr_bytes: 0,
|
||||
stdout_delivered_bytes: 7,
|
||||
stderr_delivered_bytes: 0,
|
||||
truncation_reason: None,
|
||||
}
|
||||
}
|
||||
|
||||
struct Fixture {
|
||||
_root: tempfile::TempDir,
|
||||
provider: Arc<Provider>,
|
||||
state: Arc<Mutex<BashExecutionState>>,
|
||||
view: BashOutputView,
|
||||
id: String,
|
||||
}
|
||||
|
||||
impl Fixture {
|
||||
fn start(fault: Fault) -> (Self, CommandGuard) {
|
||||
let base =
|
||||
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/cleanup-fixtures");
|
||||
std::fs::create_dir_all(&base).unwrap();
|
||||
let root = tempfile::tempdir_in(base).unwrap();
|
||||
let provider = Arc::new(Provider {
|
||||
local: LocalWorkdirSession::new(
|
||||
manifest::Scope::writable(root.path()).unwrap(),
|
||||
root.path().into(),
|
||||
),
|
||||
fault,
|
||||
cancelling: Default::default(),
|
||||
allow_drain: AtomicBool::new(false),
|
||||
});
|
||||
let view = BashOutputView::new();
|
||||
let mut retained = view.begin().unwrap();
|
||||
retained.append(CommandStream::Stdout, b"pre");
|
||||
let id = retained.id();
|
||||
let state = Arc::new(Mutex::new(BashExecutionState::default()));
|
||||
let handle = CommandHandle(PRIVATE.into());
|
||||
state.lock().unwrap().active.insert(
|
||||
PRIVATE.into(),
|
||||
ActiveCommand {
|
||||
call_id: PRIVATE.into(),
|
||||
execution_nonce: 1,
|
||||
session: provider.clone(),
|
||||
handle: handle.clone(),
|
||||
},
|
||||
);
|
||||
state
|
||||
.lock()
|
||||
.unwrap()
|
||||
.cancellation_requested
|
||||
.insert(PRIVATE.into());
|
||||
let guard = CommandGuard {
|
||||
session: provider.clone(),
|
||||
state: state.clone(),
|
||||
execution_id: PRIVATE.into(),
|
||||
execution_nonce: 1,
|
||||
handle: Some(handle),
|
||||
cursor: 3,
|
||||
retained: Some(retained),
|
||||
};
|
||||
(
|
||||
Self {
|
||||
_root: root,
|
||||
provider,
|
||||
state,
|
||||
view,
|
||||
id,
|
||||
},
|
||||
guard,
|
||||
)
|
||||
}
|
||||
|
||||
async fn finish(&self) -> BashOutputMetadata {
|
||||
self.provider.allow_drain.store(true, Ordering::Release);
|
||||
// The provider's read retry backoff and cleanup deadlines are virtual time.
|
||||
for _ in 0..4 {
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::advance(Duration::from_secs(3)).await;
|
||||
}
|
||||
for _ in 0..1000 {
|
||||
if let Some(metadata) = self.view.metadata(&self.id) {
|
||||
return metadata;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
panic!("finite cleanup did not publish partial evidence");
|
||||
}
|
||||
|
||||
fn assert_evidence(&self, m: &BashOutputMetadata, terminal: bool, bytes: &[u8]) {
|
||||
assert_eq!(
|
||||
self.view
|
||||
.read(&self.id, CommandStream::Stdout, 0, 64)
|
||||
.unwrap(),
|
||||
bytes
|
||||
);
|
||||
assert_eq!(m.stdout_saved, bytes.len());
|
||||
assert_eq!(m.stdout_delivered, bytes.len() as u64);
|
||||
assert_eq!(m.stdout_seen, bytes.len() as u64);
|
||||
assert_eq!(
|
||||
m.terminal.status,
|
||||
terminal.then_some(CommandStatus::Completed)
|
||||
);
|
||||
assert_eq!(m.terminal.exit_code, terminal.then_some(0));
|
||||
if !terminal {
|
||||
assert!(m.terminal.output_collection.is_none());
|
||||
assert!(m.terminal.process_failure.is_none());
|
||||
assert!(!m.terminal.transport_complete);
|
||||
}
|
||||
let diagnostics = m.cleanup.as_ref().unwrap();
|
||||
assert_eq!(diagnostics.execution_nonce, 1);
|
||||
let json = serde_json::to_string(diagnostics).unwrap();
|
||||
let debug = format!("{diagnostics:?}");
|
||||
for forbidden in [
|
||||
PRIVATE, "SECRET", "/private", "password", "poison", "tail", "pre",
|
||||
] {
|
||||
assert!(!json.contains(forbidden), "diagnostics leaked data: {json}");
|
||||
assert!(
|
||||
!debug.contains(forbidden),
|
||||
"tracing payload leaked data: {debug}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn cancel_failure_does_not_preclude_confirmed_terminal() {
|
||||
let (f, guard) = Fixture::start(Fault::Cancel);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
assert!(f.state.lock().unwrap().active.contains_key(PRIVATE));
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, true, b"pretail");
|
||||
assert!(m.terminal.transport_complete);
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().cancel,
|
||||
CleanupCancel::Failed { .. }
|
||||
));
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().drain,
|
||||
CleanupDrain::TerminalConfirmed
|
||||
));
|
||||
assert!(!m.cleanup.as_ref().unwrap().deadline_expired);
|
||||
assert!(f.state.lock().unwrap().active.is_empty());
|
||||
assert!(f.state.lock().unwrap().cancellation_requested.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn acknowledged_cancel_drain_faults_keep_distinct_reasons_and_partial_bytes() {
|
||||
for (fault, reason, bytes) in [
|
||||
(Fault::Drain, "provider", b"pre".as_slice()),
|
||||
(Fault::Sequence, "sequence_gap", b"pre"),
|
||||
(Fault::Cursor, "cursor_gap", b"pretail"),
|
||||
(Fault::RetainedChunk, "retained_chunk_invalid", b"pre"),
|
||||
(
|
||||
Fault::TerminalMetadata,
|
||||
"terminal_metadata_invalid",
|
||||
b"pretail",
|
||||
),
|
||||
] {
|
||||
let (f, guard) = Fixture::start(fault);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, false, bytes);
|
||||
let diagnostic = serde_json::to_value(m.cleanup.as_ref().unwrap()).unwrap();
|
||||
assert_eq!(diagnostic["cancel"]["outcome"], "acknowledged");
|
||||
assert_eq!(diagnostic["drain"]["outcome"], "failed");
|
||||
assert_eq!(diagnostic["drain"]["reason"]["kind"], reason);
|
||||
assert_eq!(diagnostic["registration"], "released");
|
||||
assert_eq!(diagnostic["deadline_expired"], false);
|
||||
assert!(f.state.lock().unwrap().active.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn terminal_deadline_never_claims_stop_or_leaves_active_collector() {
|
||||
let (f, guard) = Fixture::start(Fault::TerminalTimeout);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, false, b"pre");
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().cancel,
|
||||
CleanupCancel::Acknowledged
|
||||
));
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().drain,
|
||||
CleanupDrain::DeadlineExceeded
|
||||
));
|
||||
assert!(m.cleanup.as_ref().unwrap().deadline_expired);
|
||||
assert!(f.state.lock().unwrap().active.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn cancel_deadline_still_attempts_independent_terminal_observation() {
|
||||
let (f, guard) = Fixture::start(Fault::CancelHang);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, true, b"pretail");
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().cancel,
|
||||
CleanupCancel::DeadlineExceeded
|
||||
));
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().drain,
|
||||
CleanupDrain::TerminalConfirmed
|
||||
));
|
||||
assert!(!m.cleanup.as_ref().unwrap().deadline_expired);
|
||||
assert!(f.state.lock().unwrap().active.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn old_cleanup_preserves_new_execution_and_cancellation_registration() {
|
||||
let (f, guard) = Fixture::start(Fault::Complete);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
f.state
|
||||
.lock()
|
||||
.unwrap()
|
||||
.active
|
||||
.get_mut(PRIVATE)
|
||||
.unwrap()
|
||||
.execution_nonce = 2;
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, true, b"pretail");
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().registration,
|
||||
CleanupRegistration::Superseded
|
||||
));
|
||||
assert_eq!(f.state.lock().unwrap().active[PRIVATE].execution_nonce, 2);
|
||||
assert!(
|
||||
f.state
|
||||
.lock()
|
||||
.unwrap()
|
||||
.cancellation_requested
|
||||
.contains(PRIVATE)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn terminal_confirmation_is_independent_of_output_collection_completeness() {
|
||||
let (f, guard) = Fixture::start(Fault::CollectionIncomplete);
|
||||
drop(guard);
|
||||
f.provider.cancelling.notified().await;
|
||||
let m = f.finish().await;
|
||||
f.assert_evidence(&m, true, b"pretail");
|
||||
assert!(!m.terminal.transport_complete);
|
||||
assert!(matches!(
|
||||
m.terminal.output_collection,
|
||||
Some(CommandOutputCollection::Incomplete { .. })
|
||||
));
|
||||
assert!(matches!(
|
||||
m.cleanup.as_ref().unwrap().drain,
|
||||
CleanupDrain::TerminalConfirmed
|
||||
));
|
||||
assert!(f.state.lock().unwrap().active.is_empty());
|
||||
}
|
||||
@@ -19,7 +19,7 @@ use std::time::{Duration, Instant};
|
||||
use serde::Serialize;
|
||||
use workdir::{CommandStatus, CommandStream};
|
||||
|
||||
const COMMAND_BYTES: usize = 4 * 1024 * 1024;
|
||||
pub(crate) const COMMAND_BYTES: usize = 4 * 1024 * 1024;
|
||||
const STORE_BYTES: usize = 32 * 1024 * 1024;
|
||||
const STORE_COUNT: usize = 64;
|
||||
const ORDER_COUNT: usize = 4096;
|
||||
@@ -55,6 +55,7 @@ struct Record {
|
||||
transport_counts: Option<(u64, u64)>,
|
||||
transport_truncation: Option<workdir::CommandOutputTruncation>,
|
||||
terminal: Option<OutputTerminal>,
|
||||
cleanup: Option<CleanupDiagnostics>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
@@ -95,10 +96,72 @@ pub struct BashOutputMetadata {
|
||||
pub transport_truncation: Option<workdir::CommandOutputTruncation>,
|
||||
pub storage_truncated: bool,
|
||||
pub terminal: OutputTerminal,
|
||||
pub cleanup: Option<CleanupDiagnostics>,
|
||||
pub text_encoding: &'static str,
|
||||
pub ordering: &'static str,
|
||||
}
|
||||
|
||||
/// Drop cleanup observations, not assertions that cancellation stopped a process.
|
||||
/// Correlate with the enclosing output UUID and this host-assigned execution generation;
|
||||
/// caller identifiers, provider handles, commands and error text are deliberately excluded.
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
pub struct CleanupDiagnostics {
|
||||
pub execution_nonce: u64,
|
||||
pub cancel: CleanupCancel,
|
||||
pub drain: CleanupDrain,
|
||||
pub deadline_expired: bool,
|
||||
pub registration: CleanupRegistration,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(tag = "outcome", rename_all = "snake_case")]
|
||||
pub enum CleanupCancel {
|
||||
Acknowledged,
|
||||
Failed { reason: CleanupProviderFailure },
|
||||
DeadlineExceeded,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(tag = "outcome", rename_all = "snake_case")]
|
||||
pub enum CleanupDrain {
|
||||
TerminalConfirmed,
|
||||
Failed { reason: CleanupFailure },
|
||||
DeadlineExceeded,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
pub enum CleanupFailure {
|
||||
Provider { reason: CleanupProviderFailure },
|
||||
SequenceGap,
|
||||
CursorGap,
|
||||
DeliveryBudgetExceeded,
|
||||
RetainedChunkInvalid,
|
||||
TerminalMetadataInvalid,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
pub struct CleanupProviderFailure {
|
||||
pub failure_reason: Option<workdir::WorkdirFailureReason>,
|
||||
pub denial_reason: Option<workdir::WorkdirDenialReason>,
|
||||
}
|
||||
|
||||
impl From<&workdir::WorkdirError> for CleanupProviderFailure {
|
||||
fn from(error: &workdir::WorkdirError) -> Self {
|
||||
Self {
|
||||
failure_reason: error.failure_reason(),
|
||||
denial_reason: error.denial_reason(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum CleanupRegistration {
|
||||
Released,
|
||||
Superseded,
|
||||
}
|
||||
|
||||
/// A bounded reservation acquired before command start. Drop terminalizes partial evidence.
|
||||
pub struct BashOutputWriter {
|
||||
record: Arc<Mutex<Record>>,
|
||||
@@ -214,6 +277,7 @@ impl BashOutputView {
|
||||
transport_counts: None,
|
||||
transport_truncation: None,
|
||||
terminal: None,
|
||||
cleanup: None,
|
||||
}));
|
||||
store.records.push_back(record.clone());
|
||||
store.reserved += COMMAND_BYTES;
|
||||
@@ -313,6 +377,7 @@ impl Record {
|
||||
transport_truncation: self.transport_truncation.clone(),
|
||||
storage_truncated: self.truncated || self.transport_truncation.is_some(),
|
||||
terminal: self.terminal.clone().expect("finalized record"),
|
||||
cleanup: self.cleanup.clone(),
|
||||
text_encoding: "utf8_lossy",
|
||||
ordering: "provider_read_observation",
|
||||
}
|
||||
@@ -342,6 +407,14 @@ async fn reap(weak: Weak<Mutex<Store>>, mut shutdown: tokio::sync::watch::Receiv
|
||||
}
|
||||
|
||||
impl BashOutputWriter {
|
||||
pub(crate) fn id(&self) -> String {
|
||||
self.record.lock().unwrap().id.clone()
|
||||
}
|
||||
|
||||
pub(crate) fn finish_cleanup(&mut self, diagnostics: CleanupDiagnostics) {
|
||||
self.record.lock().unwrap().cleanup = Some(diagnostics);
|
||||
}
|
||||
|
||||
pub fn append_chunk(&mut self, chunk: &workdir::CommandByteChunk) -> Result<(), String> {
|
||||
let expected = {
|
||||
let r = self.record.lock().unwrap();
|
||||
|
||||
Reference in New Issue
Block a user