Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
{
"slice_class": "non_ranking",
"targeted_mechanism": "none",
"kind": "harness",
"fixtures": [
"harness-goldens"
],
"slices": [
"stderr-exit-hold-test"
],
"note": "Test-only hook (debug builds) that holds the stderr mirror until the exit fence starts, so the slow-writer exit test proves the fence delivers the terminal marker to stderr. No search, ranking or routing code changes."
}
10 changes: 10 additions & 0 deletions crates/aft/src/logging.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1616,6 +1616,16 @@ pub fn flush_stderr_on_exit(timeout: std::time::Duration) -> bool {
.is_none_or(|mirror| mirror.drain_on_exit(timeout))
}

/// Test-only (debug builds): stop the stderr mirror from writing until
/// [`flush_stderr_on_exit`] starts, so the lines logged in between are still
/// queued when the exit fence runs. Without the fence they would be lost.
#[cfg(debug_assertions)]
pub fn hold_stderr_mirror_until_exit_fence() {
if let Some(mirror) = STDERR_MIRROR.get() {
mirror.hold_until_exit_fence();
}
}

/// Persist the last process-exit line without depending on the async writer.
/// Taking its active sink prevents later rotation or reconfiguration from
/// moving the marker. Queued diagnostics may be abandoned once their bounded
Expand Down
90 changes: 89 additions & 1 deletion crates/aft/src/logging/stderr_mirror.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,44 @@ enum Message {
Drain(SyncSender<bool>),
}

/// Test-only (debug builds): parks the worker so that records logged during
/// process exit are still queued when the exit fence starts. That is the
/// state a busy host can leave them in when the process exits right after
/// logging them, and the fence must still deliver them.
#[cfg(debug_assertions)]
#[derive(Default)]
struct ExitHold {
armed: std::sync::Mutex<bool>,
released: std::sync::Condvar,
}

#[cfg(debug_assertions)]
impl ExitHold {
fn arm(&self) {
*self.armed.lock().unwrap_or_else(|e| e.into_inner()) = true;
}

fn release(&self) {
*self.armed.lock().unwrap_or_else(|e| e.into_inner()) = false;
self.released.notify_all();
}

fn wait(&self) {
let armed = self.armed.lock().unwrap_or_else(|e| e.into_inner());
// Bounded so a hold that is never released cannot keep a test
// process's worker parked forever.
let _ = self
.released
.wait_timeout_while(armed, Duration::from_secs(10), |armed| *armed);
}
}

#[derive(Clone)]
pub(super) struct Mirror {
tx: Option<SyncSender<Message>>,
health: Arc<MirrorHealth>,
#[cfg(debug_assertions)]
exit_hold: Arc<ExitHold>,
}

impl Mirror {
Expand All @@ -63,10 +97,16 @@ impl Mirror {
) -> io::Result<(Self, JoinHandle<()>)> {
let (tx, rx) = mpsc::sync_channel(capacity);
let worker_health = Arc::clone(&health);
#[cfg(debug_assertions)]
let exit_hold = Arc::new(ExitHold::default());
#[cfg(debug_assertions)]
let worker_hold = Arc::clone(&exit_hold);
let worker = thread::Builder::new()
.name("aft-stderr-mirror".into())
.spawn(move || {
while let Ok(message) = rx.recv() {
#[cfg(debug_assertions)]
worker_hold.wait();
// No channel or file-sink lock is held during stderr I/O.
let result = match &message {
Message::Record(bytes) => sink.write_all(bytes),
Expand All @@ -92,14 +132,21 @@ impl Mirror {
Self {
tx: Some(tx),
health,
#[cfg(debug_assertions)]
exit_hold,
},
worker,
))
}

pub(super) fn disabled(health: Arc<MirrorHealth>) -> Self {
health.start_failures.fetch_add(1, Ordering::Relaxed);
Self { tx: None, health }
Self {
tx: None,
health,
#[cfg(debug_assertions)]
exit_hold: Arc::default(),
}
}

pub(super) fn record(&self, bytes: &[u8]) {
Expand All @@ -121,6 +168,8 @@ impl Mirror {
// Only process teardown waits for this fence. Ordinary logger flushes stay
// nonblocking, including when stderr is full or its sink is wedged.
pub(super) fn drain_on_exit(&self, timeout: Duration) -> bool {
#[cfg(debug_assertions)]
self.exit_hold.release();
let Some(tx) = self.tx.as_ref() else {
return false;
};
Expand Down Expand Up @@ -148,6 +197,13 @@ impl Mirror {
.unwrap_or(false)
}

/// Test-only (debug builds): stop writing records until `drain_on_exit`
/// starts. See [`ExitHold`].
#[cfg(debug_assertions)]
pub(super) fn hold_until_exit_fence(&self) {
self.exit_hold.arm();
}

fn enqueue(&self, message: Message) {
let Some(tx) = self.tx.as_ref() else {
self.health.drop_message(&message);
Expand Down Expand Up @@ -227,6 +283,36 @@ mod tests {
worker.join().unwrap();
}

#[cfg(debug_assertions)]
#[test]
fn exit_hold_keeps_records_queued_until_the_exit_fence_starts() {
let bytes = Arc::new(Mutex::new(Vec::new()));
let (mirror, worker) = Mirror::spawn(
GateWriter {
bytes: Arc::clone(&bytes),
write_gate: None,
flush_gate: None,
},
8,
Arc::new(MirrorHealth::default()),
)
.unwrap();
mirror.hold_until_exit_fence();
mirror.record(b"process exit marker\n");
thread::sleep(Duration::from_millis(50));
let written_while_held = bytes.lock().unwrap().clone();
let drained = mirror.drain_on_exit(Duration::from_secs(5));
drop(mirror);
worker.join().unwrap();
assert!(
written_while_held.is_empty(),
"held worker wrote {:?}",
String::from_utf8_lossy(&written_while_held)
);
assert!(drained, "exit fence did not release the held worker");
assert_eq!(&*bytes.lock().unwrap(), b"process exit marker\n");
}

#[test]
fn full_queue_drops_whole_records_and_flushes_without_waiting() {
let bytes = Arc::new(Mutex::new(Vec::new()));
Expand Down Expand Up @@ -301,6 +387,8 @@ mod tests {
let mirror = Mirror {
tx: Some(tx),
health: Arc::clone(&health),
#[cfg(debug_assertions)]
exit_hold: Arc::default(),
};
mirror.record(&vec![b'x'; MAX_RECORD_BYTES + 1]);
assert!(matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty)));
Expand Down
8 changes: 8 additions & 0 deletions crates/aft/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,14 @@ fn main() {
aft::subc::exit_elapsed().as_millis()
);
let flushed = aft::logging::flush_durable_log(flush_wait);
// AFT_TEST_STDERR_EXIT_HOLD keeps every stderr line from here
// on queued until the stderr exit fence below starts, so a test
// can prove the fence delivers them rather than rely on the
// mirror thread happening to run first. Release builds ignore it.
#[cfg(debug_assertions)]
if std::env::var_os("AFT_TEST_STDERR_EXIT_HOLD").is_some() {
aft::logging::hold_stderr_mirror_until_exit_fence();
}
aft::slog_info!(
"subc exit phase=log_flush_done flushed={} elapsed_ms={}",
flushed,
Expand Down
10 changes: 10 additions & 0 deletions crates/aft/tests/integration/subc_detach_test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -847,6 +847,11 @@ fn drain_with_live_lsp_servers_and_writer(
if writer_delay == "1000" {
assert!(log.contains("phase=log_flush_done flushed=false"),
"slow writer must exhaust the async flush budget: {}", log_tail(&log));
// The terminal marker is mirrored to stderr after the queued
// flush; the stderr exit fence must deliver it there as well as
// to the durable file checked below.
assert!(log.contains("subc exit phase=process_exit mode="),
"stderr exit fence must deliver the terminal marker: {}", log_tail(&log));
}
// Print phase and LSP summary lines. The bound below uses the LSP
// phase's own elapsed value, never the outer process-exit observation.
Expand Down Expand Up @@ -1185,6 +1190,11 @@ impl ModuleProcess {
// A writer this slow cannot acknowledge a flush inside the exit
// budget, wherever in its batch delay the request lands.
command.env("AFT_TEST_LOG_FLUSH_HOLD_MS", hold);
// Keep the stderr mirror from writing the exit's last lines until
// the stderr exit fence starts, so the stderr assertions prove the
// fence delivers them instead of depending on the mirror thread
// being scheduled before the process exits.
command.env("AFT_TEST_STDERR_EXIT_HOLD", "1");
}
if let Some(bin_dir) = bin_dir {
// The test supplies its language servers through `bin_dir`. An
Expand Down
Loading