From 625406f4d3fc688559ca018d8c79b6a1ca161a89 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Mon, 12 Oct 2026 18:41:24 +0700 Subject: [PATCH 1/5] fix(platform)!: refuse audited requests on a stalled stdout audit stream A stdout audit destination whose reader stops reading held every audited request for as long as the reader stayed away: each append waited on the stream lock behind the stuck write, nothing bounded the wait, readiness stayed true, and dropping the writer joined the stuck thread. One thread now owns the stream and writes the lines handed to it in order. An append waits at most 5 seconds for its line. A destination holding one line past that deadline is stalled: the waiting request is refused with the added AuditUnavailableReason::Stalled, later requests are refused at once, and ready() is false until the destination takes the line, after which the writer accepts entries again with no restart. A failed write still stops the writer until restart. The queue behind the line being written is bounded, and a full queue refuses instead of growing. Dropping the writer waits one deadline for the thread, then leaves it. The start and the end of a stall are each logged once, with no entry content. The line the destination is holding when its caller's deadline passes is still written when the destination takes it; the crate README records that caveat with the review notes. BREAKING: AuditUnavailableReason gains the Stalled variant, and a stream append waits on the Tokio timer, so the runtime that appends needs its time driver. Without one the append is refused before its line is queued. The schedulingctl test that appended on a runtime without one enables it. Refs #1566 Signed-off-by: Jeremi Joslin --- crates/registry-platform-audit/README.md | 67 ++ crates/registry-platform-audit/src/writer.rs | 905 +++++++++++++++--- crates/registry-scheduling/src/audit.rs | 75 ++ crates/registry-schedulingctl/src/lib.rs | 2 + products/platform/CHANGELOG.md | 11 + products/platform/docs/SECURITY_PRINCIPLES.md | 2 +- 6 files changed, 911 insertions(+), 151 deletions(-) diff --git a/crates/registry-platform-audit/README.md b/crates/registry-platform-audit/README.md index d0bfa1d59b..eac055a658 100644 --- a/crates/registry-platform-audit/README.md +++ b/crates/registry-platform-audit/README.md @@ -67,6 +67,18 @@ async fn write_audit_entries() -> Result<(), Box> { and keeps its files owner-only. - A failed write stops the writer until restart, so every later audited request fails closed. +- A stalled `stdout` destination is not a failed write. A destination that + holds one line for 5 seconds is stalled: the request waiting on that line is + refused, every later audited request is refused at once, and `ready()` + reports false. When the destination takes the line the writer accepts + entries again, with no restart. The start and the end of a stall are each + logged once, with no entry content. The deadline is fixed and has no + configuration key. +- A `stdout` destination writes on a thread of its own, at most 4096 lines + behind the one being written; a line that finds the queue full is refused. + An append waits for its line on the Tokio timer, so the runtime that appends + needs its time driver; without one the append is refused before its line is + queued. - An active file whose last line was torn by an interrupted write is recovered at open: the bytes after the last complete line are copied to the owner-only side file `.torn`, the active file is truncated to its last complete @@ -102,6 +114,61 @@ async fn write_audit_entries() -> Result<(), Box> { - Redaction helpers intentionally avoid preserving email local parts, phone digits, or sensitive query values. +### Stream write deadline: review record + +A `stdout` destination whose reader stops reading used to hold every audited +request for as long as the reader stayed away. One thread now owns the +stream, and a caller waits one fixed deadline for its line. The four elements +of the change: + +- **Invariant.** A caller is told an entry was accepted only after its line + was written and flushed, and no caller waits on a stream destination longer + than the write deadline (5 seconds, `STREAM_WRITE_DEADLINE`). A stall + recovers by itself; a write error stops the writer until restart, as before. + The deadline sits below the 10-second request timeout the Base Registry + Engine defaults to and the Evidence reference projects set, so a stalled + destination is answered as an unavailable audit destination, not as a + timed-out request. +- **Enforcement point.** `LineStream` in `src/writer.rs`. `append` refuses + before queueing when the stream has failed or the write in progress is + older than the deadline, hands the line over with `try_send` so a full + queue refuses, and waits for the outcome under `tokio::time::timeout`. The + `audit-stream` thread is the only code that touches the sink; it skips a + line whose caller already stopped waiting, so a refused caller's queued + line is never written. `ready()` is false while the write in progress is + older than the deadline. Dropping the writer closes the queue and waits one + deadline for the thread, then leaves it and reports that once. +- **Executed proof.** In `src/writer.rs`, against a sink that holds its write: + `a_stalled_stream_refuses_the_append_at_the_deadline_and_later_ones_at_once`, + `readiness_follows_a_stall_and_returns_once_the_stream_drains`, + `a_write_error_still_stops_the_stream_after_its_sink_recovers`, + `a_stalled_stream_leaves_the_runtime_to_other_tasks`, + `dropping_the_writer_waits_at_most_one_deadline_for_a_blocked_stream`, + `dropping_the_writer_writes_the_entries_still_queued`, + `a_full_stream_queue_refuses_a_line_instead_of_growing`, and + `a_runtime_without_a_timer_refuses_the_append_before_queueing_its_line`. + Through a product, + `a_stalled_destination_is_not_ready_until_it_drains` in + `registry-scheduling` holds readiness to the writer over the real deadline. +- **Remaining uncertainty.** The line the destination is holding when its + caller's deadline passes cannot be withdrawn: the caller is refused and the + line is still written when the destination takes it, which is logged. That + is one line for each stall, and it records more than happened, never less. + A `request` entry written that way has no `response` entry: its request + was refused and ran no protected I/O. A `response` entry written that way + is followed by the request's unfinished response. A destination slow + enough that lines wait in the queue for most of a deadline can do the same + with no stall reported. The start of a stall is logged by the first append + or readiness check that sees it, not by a timer, so a writer nothing is + asking logs it late. A full queue refuses lines without turning `ready()` + false. A destination that never takes its line keeps one thread until the + process exits, and unfinished response entries queued behind it when the + writer is dropped are not written, which is reported once. An operator who + sets a request timeout under 5 seconds still sees a timed-out request + first. A refused entry does not undo work a product already committed: a + product that writes its outcome entry after its transaction commits reports + that operation as unresolved, as it does for a failed write. + ## Testing ```sh diff --git a/crates/registry-platform-audit/src/writer.rs b/crates/registry-platform-audit/src/writer.rs index cebba5d246..37204cbd07 100644 --- a/crates/registry-platform-audit/src/writer.rs +++ b/crates/registry-platform-audit/src/writer.rs @@ -10,6 +10,12 @@ //! after the entry's bytes are `fsync`ed, and appends arriving during one //! durable write share the next write and `fsync` (group commit). The `stdout` //! destination writes one line and flushes; it is best-effort by nature. +//! +//! A destination can refuse entries in two ways. A failed write stops the +//! writer until the process restarts. A `stdout` destination that holds one +//! line past a fixed write deadline is stalled instead: the writer refuses +//! entries while the stall lasts and accepts them again once the destination +//! takes the line it holds. use std::{ fs::{self, File, OpenOptions, TryLockError}, @@ -21,9 +27,10 @@ use std::{ path::{Component, Path, PathBuf}, sync::{ atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, - Arc, Mutex as StdMutex, + mpsc::{Receiver, SyncSender, TrySendError}, + Arc, Condvar, Mutex as StdMutex, MutexGuard, PoisonError, }, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; use serde::{Deserialize, Serialize}; @@ -87,6 +94,18 @@ const _: () = assert!(SEQUENCE_SUFFIX.len() <= 1 + SEGMENT_SEQUENCE_DIGITS); const _: () = assert!(SEQUENCE_TEMPORARY_SUFFIX.len() <= 1 + SEGMENT_SEQUENCE_DIGITS); const _: () = assert!(TORN_LINE_SUFFIX.len() <= 1 + SEGMENT_SEQUENCE_DIGITS); const DETACHED_LOCK_ATTEMPTS: usize = 1024; +/// How long a stream destination may hold one line. Past it the destination +/// is stalled: the caller waiting on a line is refused, and so is every later +/// append until the destination takes the line it holds. It is shorter than +/// the request timeouts products default to, so a stalled destination is +/// answered as an unavailable audit destination and not as a timed-out +/// request. +const STREAM_WRITE_DEADLINE: Duration = Duration::from_secs(5); +/// How many lines may wait behind the one a stream destination is writing. A +/// line waits only while an earlier one is written, so a full queue means the +/// destination is not keeping up, and refusing the next line bounds the +/// memory held for it. +const STREAM_QUEUE_LINES: usize = 4096; const SECONDS_PER_DAY: u64 = 24 * 60 * 60; const TIME_FORMAT: &[FormatItem<'static>] = format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"); @@ -231,6 +250,11 @@ pub enum AuditUnavailableReason { /// An earlier write failed, so the writer refuses every later append until /// the process restarts. Stopped, + /// A stream destination is not taking lines: it did not take this one + /// within the write deadline, it is still holding an earlier one past + /// that deadline, or its queue is full. Unlike [`Self::Stopped`], the + /// writer accepts entries again once the destination does. + Stalled, } /// The writer did not accept an entry. Products map this to their existing @@ -773,7 +797,7 @@ impl Drop for InFlightResponse { enum WriterInner { File(Arc), - Stream(Arc, DetachedLines), + Stream(LineStream), } impl std::fmt::Debug for AuditWriter { @@ -811,6 +835,10 @@ impl AuditWriter { /// A writer over an arbitrary line sink, with `stdout` semantics. For /// tests that need to observe emitted lines without a file. + /// + /// Like `stdout`, the sink is written on a thread of its own, and an + /// append waits for it on the Tokio timer: the runtime that appends needs + /// its time driver. #[must_use] pub fn from_line_sink(sink: Box) -> Self { Self { @@ -823,6 +851,10 @@ impl AuditWriter { /// entry's bytes are durable. Canceling the caller does not cancel an /// enqueued file write or the other entries in its group commit. /// + /// For a stream destination this returns once the line is written and + /// flushed, or refuses with [`AuditUnavailableReason::Stalled`] when the + /// destination has not taken it within the write deadline. + /// /// An accepted `response` entry answers the oldest [`AuditRequest`] still /// open under the same schema and correlation with no response of its own /// being written. @@ -867,12 +899,7 @@ impl AuditWriter { .await .map_err(|_| AuditUnavailable::new(AuditUnavailableReason::Stopped))? } - WriterInner::Stream(stream, _) => { - let stream = Arc::clone(stream); - tokio::task::spawn_blocking(move || stream.append(&line)) - .await - .map_err(|_| AuditUnavailable::new(AuditUnavailableReason::Stopped))? - } + WriterInner::Stream(stream) => stream.append(line).await, } } @@ -890,7 +917,11 @@ impl AuditWriter { /// /// That pairing holds while the process runs. A process killed or /// exited without unwinding, or a runtime shut down while an entry is - /// being written, can leave a request entry without its response. + /// being written, can leave a request entry without its response. So can + /// a stalled stream destination: the one line it is holding when the + /// write deadline passes is refused to its caller and still written when + /// the destination takes it, which leaves a refused request's entry + /// unpaired, or a response entry followed by the unfinished one. pub async fn begin( &self, schema: impl Into, @@ -944,14 +975,15 @@ impl AuditWriter { } /// Block until every response entry a dropped [`AuditRequest`] handed - /// to a stream destination has been written. Those entries are written - /// on a dedicated thread so a drop never blocks the runtime; this is for - /// tests and shutdown paths that read the stream right after a drop. A - /// file destination queues its entries into the group commit instead, - /// and this returns at once for it. + /// to a stream destination has been written or refused. Those entries + /// are written on the stream's own thread so a drop never blocks the + /// runtime; this is for tests and shutdown paths that read the stream + /// right after a drop, and it waits as long as the destination holds a + /// line. A file destination queues its entries into the group commit + /// instead, and this returns at once for it. pub fn wait_for_detached_entries(&self) { - if let WriterInner::Stream(_, detached) = self.inner.as_ref() { - detached.wait(); + if let WriterInner::Stream(stream) = self.inner.as_ref() { + stream.wait_for_detached(); } } @@ -966,19 +998,22 @@ impl AuditWriter { }; match self.inner.as_ref() { WriterInner::File(file) => file.enqueue_detached(line), - // A dedicated thread writes it, so a drop never blocks a runtime - // thread on a slow stream, and the thread is joined when the - // writer is dropped, so shutdown does not lose it. - WriterInner::Stream(_, detached) => detached.send(line), + // The stream's thread writes it, so a drop never blocks a runtime + // thread on a slow stream, and dropping the writer gives that + // thread one deadline to write what is queued, so shutdown does + // not lose it. + WriterInner::Stream(stream) => stream.append_detached(line), } } /// Report whether the writer can still accept entries. For the file /// destination this also confirms the writer still owns the active file. + /// A stream destination holding one line past the write deadline is not + /// ready until it takes that line. pub async fn ready(&self) -> bool { match self.inner.as_ref() { WriterInner::File(file) => file.ready().await, - WriterInner::Stream(stream, _) => stream.healthy(), + WriterInner::Stream(stream) => stream.healthy(), } } @@ -1139,156 +1174,370 @@ impl Drop for AuditRequest { } } -/// The unfinished response entries dropped requests hand to a stream -/// destination, written in order on one dedicated thread. -struct DetachedLines { - stream: Arc, - sender: StdMutex>>, - thread: StdMutex>>, - /// Lines handed over and not yet written, with a signal on each write. - pending: Arc<(StdMutex, std::sync::Condvar)>, +impl WriterInner { + fn stream(out: Box) -> Self { + Self::Stream(LineStream::new(out)) + } } -impl DetachedLines { - fn new(stream: Arc) -> Self { - Self { - stream, - sender: StdMutex::new(None), - thread: StdMutex::new(None), - pending: Arc::new((StdMutex::new(0), std::sync::Condvar::new())), +/// One line handed to a stream destination's thread. +struct StreamJob { + line: String, + /// Where the caller waiting on the line hears its outcome. `None` for the + /// unfinished response of a dropped request, which nobody waits on. + written: Option>>, +} + +/// What a stream destination reports to the operational log. No report +/// carries entry content. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum StreamEvent { + /// The destination has held one line past the write deadline. + StallStarted, + /// The destination took the line it was holding, after this long. + StallEnded(Duration), + /// A line was written although its caller had stopped waiting on it. + WrittenAfterRefusal, + /// An unfinished response entry was refused. + DetachedNotAccepted, + /// The writer was dropped while the destination still held a line. + AbandonedAtDrop, +} + +/// A stream destination. One thread owns the sink and writes the lines handed +/// to it in order, so a destination that stops taking lines holds that thread +/// and never a caller: a caller waits one deadline at most. +/// +/// [`LineStream::append`] waits on the Tokio timer, so the runtime it is +/// called on must have its time driver enabled. +struct LineStream { + /// The queue to the thread. `None` when the thread did not start. + jobs: Option>, + thread: Option>, + state: Arc, +} + +/// What a stream destination's thread shares with its callers. +struct StreamState { + /// Set by a failed write and never cleared. + failed: AtomicBool, + /// How long the destination may hold one line. + deadline: Duration, + progress: StdMutex, + /// Signaled when a detached line settles and when the thread ends. + changed: Condvar, + /// Lines handed to the thread so far. + #[cfg(test)] + handed_over: AtomicUsize, + /// Every report so far, so a test can observe what was logged. + #[cfg(test)] + events: StdMutex>, +} + +#[derive(Default)] +struct StreamProgress { + /// When the write in progress started. + writing_since: Option, + /// Whether the write in progress was already reported as stalled. + stall_reported: bool, + /// Detached lines handed over and neither written nor refused yet. + detached: usize, + /// Whether the thread has ended. + ended: bool, +} + +impl LineStream { + fn new(out: Box) -> Self { + Self::with_limits(out, STREAM_WRITE_DEADLINE, STREAM_QUEUE_LINES) + } + + fn with_limits(out: Box, deadline: Duration, queue_lines: usize) -> Self { + let state = Arc::new(StreamState { + failed: AtomicBool::new(false), + deadline, + progress: StdMutex::default(), + changed: Condvar::new(), + #[cfg(test)] + handed_over: AtomicUsize::new(0), + #[cfg(test)] + events: StdMutex::default(), + }); + let (jobs, queued) = std::sync::mpsc::sync_channel(queue_lines); + let spawned = std::thread::Builder::new() + .name("audit-stream".to_owned()) + .spawn({ + let state = Arc::clone(&state); + move || state.run(out, queued) + }); + match spawned { + Ok(thread) => Self { + jobs: Some(jobs), + thread: Some(thread), + state, + }, + Err(error) => { + state.failed.store(true, Ordering::Release); + tracing::error!(%error, "audit stream thread did not start; refusing audited requests"); + Self { + jobs: None, + thread: None, + state, + } + } } } - fn send(&self, line: String) { - let Ok(mut sender) = self.sender.lock() else { - tracing::error!("an unfinished response entry was not written"); - return; - }; - if sender.is_none() { - let (lines, received) = std::sync::mpsc::channel::(); - let stream = Arc::clone(&self.stream); - let pending = Arc::clone(&self.pending); - let spawned = std::thread::Builder::new() - .name("audit-detached".to_owned()) - .spawn(move || { - for line in received { - if stream.append(&line).is_err() { - tracing::error!("an unfinished response entry was not accepted"); - } - let (count, written) = &*pending; - if let Ok(mut count) = count.lock() { - *count = count.saturating_sub(1); - } - written.notify_all(); - } - }); - match spawned { - Ok(thread) => { - if let Ok(mut slot) = self.thread.lock() { - *slot = Some(thread); + fn healthy(&self) -> bool { + !self.state.failed() && !self.state.stalled() + } + + /// Hand `line` to the thread and wait, one deadline at most, until it is + /// written and flushed. + async fn append(&self, line: String) -> Result<(), AuditUnavailable> { + if self.state.failed() { + return Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)); + } + if self.state.stalled() { + return Err(AuditUnavailable::new(AuditUnavailableReason::Stalled)); + } + let (written, mut outcome) = tokio::sync::oneshot::channel(); + // The wait is set up before the line is handed over: a runtime + // without a time driver panics here, and its caller is refused with + // nothing queued. + let answered = tokio::time::timeout(self.state.deadline, &mut outcome); + self.hand_over(StreamJob { + line, + written: Some(written), + })?; + match answered.await { + Ok(Ok(result)) => result, + // The thread ended without answering. + Ok(Err(_)) => Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)), + Err(_) => { + // Closing settles it: either the line was answered as the + // deadline passed, or the thread now finds this caller gone. + outcome.close(); + match outcome.try_recv() { + Ok(result) => result, + Err(_) => { + // Looked at for its report: a stall that began under + // this caller is logged as this caller is refused. + self.state.stalled(); + Err(AuditUnavailable::new(AuditUnavailableReason::Stalled)) } - *sender = Some(lines); - } - Err(error) => { - tracing::error!(%error, "an unfinished response entry was not written"); - return; } } } - if let Ok(mut count) = self.pending.0.lock() { - *count += 1; + } + + /// Hand `line` to the thread without waiting for its outcome. + fn append_detached(&self, line: String) { + self.state.progress().detached += 1; + let handed_over = self.hand_over(StreamJob { + line, + written: None, + }); + if handed_over.is_err() { + self.state.report(StreamEvent::DetachedNotAccepted); + self.state.settle_detached(); } - if sender - .as_ref() - .is_some_and(|lines| lines.send(line).is_err()) - { - tracing::error!("an unfinished response entry was not written"); - if let Ok(mut count) = self.pending.0.lock() { - *count = count.saturating_sub(1); + } + + /// Queue `job` behind the line being written, without waiting for room. + fn hand_over(&self, job: StreamJob) -> Result<(), AuditUnavailable> { + let Some(jobs) = &self.jobs else { + return Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)); + }; + match jobs.try_send(job) { + Ok(()) => { + #[cfg(test)] + self.state.handed_over.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + Err(TrySendError::Full(_)) => { + Err(AuditUnavailable::new(AuditUnavailableReason::Stalled)) + } + Err(TrySendError::Disconnected(_)) => { + Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)) } } } - fn wait(&self) { - let (count, written) = &*self.pending; - let Ok(mut count) = count.lock() else { - return; - }; - while *count > 0 { - count = match written.wait(count) { - Ok(count) => count, - Err(_) => return, - }; + /// Block until every detached line handed over is written or refused, or + /// the thread has ended. + fn wait_for_detached(&self) { + let mut progress = self.state.progress(); + while progress.detached > 0 && !progress.ended { + progress = self + .state + .changed + .wait(progress) + .unwrap_or_else(PoisonError::into_inner); } } } -impl Drop for DetachedLines { - /// Close the queue and let the thread write what is left. +impl Drop for LineStream { + /// Close the queue and give the thread one deadline to write what is + /// left. A destination still holding a line after that keeps the thread, + /// not the caller dropping the writer. fn drop(&mut self) { - if let Ok(mut sender) = self.sender.lock() { - sender.take(); - } - let thread = self.thread.lock().ok().and_then(|mut thread| thread.take()); - if let Some(thread) = thread { + self.jobs.take(); + let Some(thread) = self.thread.take() else { + return; + }; + let (progress, _) = self + .state + .changed + .wait_timeout_while(self.state.progress(), self.state.deadline, |progress| { + !progress.ended + }) + .unwrap_or_else(PoisonError::into_inner); + let ended = progress.ended; + drop(progress); + if ended { + // A sink that panicked already ended the thread and stopped the + // stream; there is nothing further to report here. let _ = thread.join(); + } else { + self.state.report(StreamEvent::AbandonedAtDrop); } } } -impl WriterInner { - fn stream(out: Box) -> Self { - let stream = Arc::new(LineStream::new(out)); - Self::Stream(Arc::clone(&stream), DetachedLines::new(stream)) - } -} +/// Marks the stream stopped when its thread ends, however it ends, so no +/// caller waits on a thread that is gone. +struct StreamEnd<'a>(&'a StreamState); -struct LineStream { - out: StdMutex>, - healthy: AtomicBool, - #[cfg(test)] - before_lock_hook: Option>, +impl Drop for StreamEnd<'_> { + fn drop(&mut self) { + self.0.failed.store(true, Ordering::Release); + self.0.progress().ended = true; + self.0.changed.notify_all(); + } } -impl LineStream { - fn new(out: Box) -> Self { - Self { - out: StdMutex::new(out), - healthy: AtomicBool::new(true), - #[cfg(test)] - before_lock_hook: None, +impl StreamState { + fn failed(&self) -> bool { + self.failed.load(Ordering::Acquire) + } + + /// The progress record. Each holder leaves it consistent, so a poisoned + /// lock is still read. + fn progress(&self) -> MutexGuard<'_, StreamProgress> { + self.progress.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Whether the destination has held the line being written for a whole + /// deadline. The first caller to see a stall reports it. + fn stalled(&self) -> bool { + let mut progress = self.progress(); + let stalled = progress + .writing_since + .is_some_and(|since| since.elapsed() >= self.deadline); + let report = stalled && !progress.stall_reported; + progress.stall_reported |= stalled; + drop(progress); + if report { + self.report(StreamEvent::StallStarted); + } + stalled + } + + fn settle_detached(&self) { + let mut progress = self.progress(); + progress.detached = progress.detached.saturating_sub(1); + drop(progress); + self.changed.notify_all(); + } + + /// The thread: write each line handed over, in order, until the queue + /// closes. + fn run(&self, mut out: Box, jobs: Receiver) { + let _end = StreamEnd(self); + for StreamJob { line, written } in jobs { + match written { + Some(written) => { + // The caller stopped waiting and was refused, so its line + // is not written. + if written.is_closed() { + continue; + } + let result = self.write(out.as_mut(), &line); + if written.send(result).is_err() && result.is_ok() { + self.report(StreamEvent::WrittenAfterRefusal); + } + } + None => { + if self.write(out.as_mut(), &line).is_err() { + self.report(StreamEvent::DetachedNotAccepted); + } + self.settle_detached(); + } + } } } - fn healthy(&self) -> bool { - self.healthy.load(Ordering::Acquire) - } - - fn append(&self, line: &str) -> Result<(), AuditUnavailable> { - if !self.healthy() { - return Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)); - } - #[cfg(test)] - if let Some(hook) = &self.before_lock_hook { - hook(); - } - let mut out = self - .out - .lock() - .map_err(|_| AuditUnavailable::new(AuditUnavailableReason::Stopped))?; - // A prior holder can stop the stream while this append waits. - if !self.healthy() { + /// Write one line and flush it. A failed write stops the stream until + /// restart. + fn write(&self, out: &mut dyn Write, line: &str) -> Result<(), AuditUnavailable> { + if self.failed() { return Err(AuditUnavailable::new(AuditUnavailableReason::Stopped)); } - match out.write_all(line.as_bytes()).and_then(|()| out.flush()) { - Ok(()) => Ok(()), + self.progress().writing_since = Some(Instant::now()); + let written = out.write_all(line.as_bytes()).and_then(|()| out.flush()); + // A failure is recorded before the write stops counting as in + // progress, so no append sees a stream that is neither stalled nor + // stopped. + if written.is_err() { + self.failed.store(true, Ordering::Release); + } + let mut progress = self.progress(); + let reported = std::mem::take(&mut progress.stall_reported); + let took = progress.writing_since.take().map(|since| since.elapsed()); + drop(progress); + match written { + Ok(()) => { + if let Some(took) = took.filter(|took| reported || *took >= self.deadline) { + self.report(StreamEvent::StallEnded(took)); + } + Ok(()) + } Err(error) => { - self.healthy.store(false, Ordering::Release); tracing::error!(%error, "audit stream write failed; refusing later audited requests"); Err(AuditUnavailable::new(AuditUnavailableReason::WriteFailed)) } } } + + fn report(&self, event: StreamEvent) { + match event { + StreamEvent::StallStarted => tracing::error!( + deadline_ms = self.deadline.as_millis(), + "audit stream has held one entry past the write deadline; refusing audited \ + requests until it takes the entry" + ), + StreamEvent::StallEnded(took) => tracing::warn!( + held_ms = took.as_millis(), + "audit stream took the entry it was holding; accepting audited requests again" + ), + StreamEvent::WrittenAfterRefusal => tracing::warn!( + "an audit entry was written after its request stopped waiting for it" + ), + StreamEvent::DetachedNotAccepted => { + tracing::error!("an unfinished response entry was not accepted"); + } + StreamEvent::AbandonedAtDrop => tracing::error!( + "audit stream still held an entry when its writer was dropped; entries queued \ + behind it may not be written" + ), + } + #[cfg(test)] + self.events + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(event); + } } /// Group commit over one [`SegmentedFile`]: appends that arrive while a @@ -3773,29 +4022,38 @@ mod tests { } let (entered, first_write) = std::sync::mpsc::channel(); - let (queued, second_append) = std::sync::mpsc::channel(); let (release, failure_gate) = std::sync::mpsc::channel(); let writes = Arc::new(AtomicU64::new(0)); - let mut stream = LineStream::new(Box::new(FailOnce { + let writer = AuditWriter::from_line_sink(Box::new(FailOnce { entered, release: failure_gate, writes: writes.clone(), })); - let calls = AtomicU64::new(0); - stream.before_lock_hook = Some(Arc::new(move || { - if calls.fetch_add(1, Ordering::Relaxed) == 1 { - queued.send(()).expect("second append passed readiness"); + let state = stream_state(&writer); + let append = |correlation: &'static str| { + let writer = writer.clone(); + move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime") + .block_on(writer.append(request(correlation))) } - })); + }; std::thread::scope(|scope| { - let first = scope.spawn(|| stream.append("first\n")); + let first = scope.spawn(append("first")); first_write .recv_timeout(Duration::from_secs(5)) .expect("first write started"); - let second = scope.spawn(|| stream.append("second\n")); - second_append - .recv_timeout(Duration::from_secs(5)) - .expect("second append queued"); + let second = scope.spawn(append("second")); + let waiting = std::time::Instant::now(); + while state.handed_over.load(Ordering::SeqCst) < 2 { + assert!( + waiting.elapsed() < Duration::from_secs(5), + "second append queued" + ); + std::thread::yield_now(); + } release.send(()).expect("release failed write"); assert_eq!( first @@ -5459,16 +5717,20 @@ mod tests { } panic!("the sink never received write {writes}"); } - } - impl Write for GatedSink { - fn write(&mut self, bytes: &[u8]) -> io::Result { + /// Count one write and hold it until the test releases the gate. + fn hold(&self) { self.entered.fetch_add(1, Ordering::SeqCst); let mut open = self.open.0.lock().expect("gate"); while !*open { open = self.open.1.wait(open).expect("gate"); } - drop(open); + } + } + + impl Write for GatedSink { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.hold(); self.buffer.write(bytes) } @@ -5854,4 +6116,347 @@ mod tests { drop(request); assert_paired(&settled_lines(&writer, &buffer), "unfinished"); } + + /// A stream deadline short enough to reach a stall without a long wait. + const TEST_DEADLINE: Duration = Duration::from_millis(200); + + /// A stream writer over `sink` with its own deadline and queue size. + fn limited( + sink: impl Write + Send + 'static, + deadline: Duration, + queue_lines: usize, + ) -> AuditWriter { + AuditWriter { + inner: Arc::new(WriterInner::Stream(LineStream::with_limits( + Box::new(sink), + deadline, + queue_lines, + ))), + open: Arc::default(), + } + } + + fn stream_state(writer: &AuditWriter) -> Arc { + match writer.inner.as_ref() { + WriterInner::Stream(stream) => Arc::clone(&stream.state), + WriterInner::File(_) => panic!("a stream writer"), + } + } + + /// What the stream reported to the operational log, in order. + fn reported(state: &StreamState) -> Vec { + state.events.lock().expect("events").clone() + } + + async fn readiness_becomes(writer: &AuditWriter, ready: bool) { + for _ in 0..500 { + if writer.ready().await == ready { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("the writer never reported ready = {ready}"); + } + + fn correlations(buffer: &SharedBuffer) -> Vec { + buffered_lines(buffer) + .iter() + .map(|line| { + line["correlation"] + .as_str() + .expect("correlation") + .to_owned() + }) + .collect() + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_stalled_stream_refuses_the_append_at_the_deadline_and_later_ones_at_once() { + // The sink never takes a line. + let sink = GatedSink::new(); + let writer = limited(sink.clone(), TEST_DEADLINE, 16); + let started = std::time::Instant::now(); + let refused = writer + .append(request("stuck")) + .await + .expect_err("refused at the deadline"); + let waited = started.elapsed(); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + assert!(waited >= TEST_DEADLINE, "{waited:?}"); + assert!(waited < Duration::from_secs(5), "{waited:?}"); + + readiness_becomes(&writer, false).await; + let state = stream_state(&writer); + for correlation in ["later-1", "later-2"] { + let refused = writer + .append(request(correlation)) + .await + .expect_err("refused while stalled"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + } + let refused = writer + .begin( + SCHEMA, + "later-3", + json!({"operationId": "read"}), + unfinished(), + ) + .await + .expect_err("refused while stalled"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + // Refused at once: a line that waited out a deadline would have been + // handed to the stream first. + assert_eq!( + state.handed_over.load(Ordering::SeqCst), + 1, + "no later line was queued" + ); + assert_eq!( + sink.entered.load(Ordering::SeqCst), + 1, + "no later line reached the sink" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn readiness_follows_a_stall_and_returns_once_the_stream_drains() { + let sink = GatedSink::new(); + let writer = limited(sink.clone(), TEST_DEADLINE, 16); + let state = stream_state(&writer); + assert!(writer.ready().await); + + let stuck = tokio::spawn({ + let writer = writer.clone(); + async move { writer.append(request("stuck")).await } + }); + sink.wait_entered(1).await; + // A second line queues behind the stuck one before the stall shows. + let queued = tokio::spawn({ + let writer = writer.clone(); + async move { writer.append(request("queued")).await } + }); + for outcome in [stuck.await, queued.await] { + let refused = outcome.expect("join").expect_err("refused at the deadline"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + } + assert!(!writer.ready().await, "a stalled stream is not ready"); + let refused = writer + .append(request("refused")) + .await + .expect_err("refused while stalled"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + + // The sink takes the line it was holding: the stream recovers with no + // restart. + sink.release(); + readiness_becomes(&writer, true).await; + writer.append(request("after")).await.expect("accepted"); + + // The line being written when its caller was refused is written + // once. The line whose caller gave up in the queue is not. + assert_eq!(correlations(&sink.buffer), ["stuck", "after"]); + let reported = reported(&state); + assert!( + matches!( + reported.as_slice(), + [ + StreamEvent::StallStarted, + StreamEvent::StallEnded(_), + StreamEvent::WrittenAfterRefusal + ] + ), + "{reported:?}" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_write_error_still_stops_the_stream_after_its_sink_recovers() { + /// Holds its first write until released, then fails it. Every later + /// write would succeed. + struct FailsHeldWrite { + gate: GatedSink, + failed: bool, + } + + impl Write for FailsHeldWrite { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.gate.hold(); + if self.failed { + return self.gate.buffer.write(bytes); + } + self.failed = true; + Err(io::Error::other("stream closed")) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + let gate = GatedSink::new(); + let writer = limited( + FailsHeldWrite { + gate: gate.clone(), + failed: false, + }, + TEST_DEADLINE, + 16, + ); + let state = stream_state(&writer); + let refused = writer + .append(request("stuck")) + .await + .expect_err("refused at the deadline"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + readiness_becomes(&writer, false).await; + + // The stall ends in a failed write. Unlike a stall, that is final. + gate.release(); + let mut reason = AuditUnavailableReason::Stalled; + for _ in 0..500 { + reason = writer + .append(request("later")) + .await + .expect_err("refused") + .reason(); + if reason == AuditUnavailableReason::Stopped { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(reason, AuditUnavailableReason::Stopped); + tokio::time::sleep(TEST_DEADLINE * 2).await; + assert!(!writer.ready().await, "a failed write stays failed"); + let refused = writer.append(request("last")).await.expect_err("stopped"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stopped); + assert_eq!(gate.entered.load(Ordering::SeqCst), 1); + assert!(buffered_lines(&gate.buffer).is_empty()); + assert_eq!(reported(&state), [StreamEvent::StallStarted]); + } + + #[tokio::test] + async fn a_stalled_stream_leaves_the_runtime_to_other_tasks() { + let sink = GatedSink::new(); + let writer = limited(sink.clone(), TEST_DEADLINE, 16); + let ticks = Arc::new(AtomicUsize::new(0)); + let ticking = tokio::spawn({ + let ticks = Arc::clone(&ticks); + async move { + loop { + tokio::time::sleep(Duration::from_millis(10)).await; + ticks.fetch_add(1, Ordering::SeqCst); + } + } + }); + let refused = writer + .append(request("stuck")) + .await + .expect_err("refused at the deadline"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + // This runtime has one worker: the other task only ran if the + // stalled append left that worker free. + let ticks = ticks.load(Ordering::SeqCst); + assert!(ticks >= 2, "{ticks} ticks during the stalled append"); + ticking.abort(); + } + + #[test] + fn dropping_the_writer_waits_at_most_one_deadline_for_a_blocked_stream() { + // The sink never takes a line. + let sink = GatedSink::new(); + let writer = limited(sink.clone(), TEST_DEADLINE, 16); + let state = stream_state(&writer); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + runtime + .block_on(writer.append(request("stuck"))) + .expect_err("refused at the deadline"); + writer.append_detached(&AuditEntry::response(SCHEMA, "queued", unfinished())); + + let started = std::time::Instant::now(); + drop(writer); + let waited = started.elapsed(); + assert!(waited >= TEST_DEADLINE, "{waited:?}"); + assert!(waited < Duration::from_secs(5), "{waited:?}"); + assert!(buffered_lines(&sink.buffer).is_empty()); + assert_eq!( + reported(&state), + [StreamEvent::StallStarted, StreamEvent::AbandonedAtDrop] + ); + } + + #[test] + fn dropping_the_writer_writes_the_entries_still_queued() { + let (writer, buffer) = buffered(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + let request = runtime + .block_on(writer.begin( + SCHEMA, + "req-1", + json!({"operationId": "read"}), + unfinished(), + )) + .expect("request"); + drop(request); + drop(writer); + assert_paired(&buffered_lines(&buffer), "unfinished"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_full_stream_queue_refuses_a_line_instead_of_growing() { + let sink = GatedSink::new(); + // One line may wait behind the one being written. The deadline is + // long, so every refusal here is the queue's and not a stall's. + let writer = limited(sink.clone(), Duration::from_secs(60), 1); + let state = stream_state(&writer); + let writing = tokio::spawn({ + let writer = writer.clone(); + async move { writer.append(request("writing")).await } + }); + sink.wait_entered(1).await; + writer.append_detached(&AuditEntry::response(SCHEMA, "queued", unfinished())); + + let started = std::time::Instant::now(); + let refused = writer + .append(request("refused")) + .await + .expect_err("the queue is full"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stalled); + assert!(started.elapsed() < Duration::from_secs(5)); + assert!(reported(&state).is_empty()); + writer.append_detached(&AuditEntry::response( + SCHEMA, + "refused-detached", + unfinished(), + )); + assert_eq!(reported(&state), [StreamEvent::DetachedNotAccepted]); + + sink.release(); + writing.await.expect("join").expect("written"); + writer.wait_for_detached_entries(); + assert_eq!(correlations(&sink.buffer), ["writing", "queued"]); + assert_eq!(reported(&state), [StreamEvent::DetachedNotAccepted]); + } + + #[test] + fn a_runtime_without_a_timer_refuses_the_append_before_queueing_its_line() { + let (writer, buffer) = buffered(); + let state = stream_state(&writer); + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("runtime"); + let refused = runtime + .block_on(writer.append(request("no-timer"))) + .expect_err("refused without a timer"); + assert_eq!(refused.reason(), AuditUnavailableReason::Stopped); + // The caller was refused, so its line must never reach the sink. + assert_eq!(state.handed_over.load(Ordering::SeqCst), 0); + drop(writer); + assert!(buffered_lines(&buffer).is_empty()); + } } diff --git a/crates/registry-scheduling/src/audit.rs b/crates/registry-scheduling/src/audit.rs index a9b0215cb3..3553ebaf26 100644 --- a/crates/registry-scheduling/src/audit.rs +++ b/crates/registry-scheduling/src/audit.rs @@ -256,6 +256,12 @@ pub use capture::AuditCapture; #[cfg(test)] mod tests { + use std::{ + io::{self, Write}, + sync::{Arc, Condvar, Mutex}, + time::Duration, + }; + use serde_json::json; use super::*; @@ -321,4 +327,73 @@ mod tests { assert!(capture.entries().is_empty()); assert!(!audit.ready().await); } + + /// A line sink that holds every write until the test opens it, as a + /// standard output nobody reads does. + #[derive(Clone, Default)] + struct HeldSink(Arc<(Mutex, Condvar)>); + + impl HeldSink { + fn open(&self) { + *self.0 .0.lock().expect("held sink") = true; + self.0 .1.notify_all(); + } + } + + impl Write for HeldSink { + fn write(&mut self, bytes: &[u8]) -> io::Result { + let mut open = self.0 .0.lock().expect("held sink"); + while !*open { + open = self.0 .1.wait(open).expect("held sink"); + } + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + /// Whether the destination reports `ready` within `limit`. + async fn readiness_becomes(audit: &SchedulingAudit, ready: bool, limit: Duration) -> bool { + let deadline = tokio::time::Instant::now() + limit; + while tokio::time::Instant::now() < deadline { + if audit.ready().await == ready { + return true; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + false + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_stalled_destination_is_not_ready_until_it_drains() { + let sink = HeldSink::default(); + let audit = SchedulingAudit::new(AuditWriter::from_line_sink(Box::new(sink.clone()))); + assert!(audit.ready().await); + + // The destination stops taking lines. The request waiting on it is + // answered with a refusal, and readiness follows the stall. + let answered = tokio::time::timeout( + Duration::from_secs(15), + audit.request(Uuid::new_v4(), json!({"operation": "hold.create"})), + ) + .await; + let not_ready_while_stalled = + readiness_becomes(&audit, false, Duration::from_secs(2)).await; + sink.open(); + assert!( + matches!(answered, Ok(Err(_))), + "a request waiting on a stalled destination is refused: {answered:?}" + ); + assert!(not_ready_while_stalled); + + // The destination drains: readiness returns and requests are accepted + // again, with no restart. + assert!(readiness_becomes(&audit, true, Duration::from_secs(5)).await); + audit + .request(Uuid::new_v4(), json!({"operation": "hold.create"})) + .await + .expect("the drained destination accepts requests"); + } } diff --git a/crates/registry-schedulingctl/src/lib.rs b/crates/registry-schedulingctl/src/lib.rs index 0abd0fb2e1..c0eabe52ce 100644 --- a/crates/registry-schedulingctl/src/lib.rs +++ b/crates/registry-schedulingctl/src/lib.rs @@ -1159,7 +1159,9 @@ mod tests { let audit = registry_scheduling::audit::SchedulingAudit::new( registry_platform_audit::AuditWriter::from_line_sink(Box::new(Refusing)), ); + // A stream append waits on the Tokio timer. let unavailable = tokio::runtime::Builder::new_current_thread() + .enable_time() .build() .unwrap() .block_on(audit.activation_request(uuid::Uuid::new_v4(), json!({}))) diff --git a/products/platform/CHANGELOG.md b/products/platform/CHANGELOG.md index 19e2741a7e..fc95d96330 100644 --- a/products/platform/CHANGELOG.md +++ b/products/platform/CHANGELOG.md @@ -2,6 +2,17 @@ ## Unreleased +- BREAKING: `registry-platform-audit` bounds a `stdout` audit destination + that stops taking lines. One thread writes the stream, and an append waits + at most 5 seconds for its line. A destination that holds one line past that + deadline is stalled: audited requests are refused with the added + `AuditUnavailableReason::Stalled` and `ready()` is false until the + destination takes the line, after which the writer accepts entries again + with no restart. A failed write still stops the writer until restart. The + added variant breaks an exhaustive match on `AuditUnavailableReason`, and a + stream append now waits on the Tokio timer, so the runtime that appends + needs its time driver. The deadline is fixed and has no configuration key. + ## 0.41.0 - 2026-10-13 - `registry-platform-oidc` gains a `task_grant` module holding the exact diff --git a/products/platform/docs/SECURITY_PRINCIPLES.md b/products/platform/docs/SECURITY_PRINCIPLES.md index 09c16f92cc..b62d0e617b 100644 --- a/products/platform/docs/SECURITY_PRINCIPLES.md +++ b/products/platform/docs/SECURITY_PRINCIPLES.md @@ -50,7 +50,7 @@ Platform crates outside `audit` deliberately do not take a hard dependency on `a PII-bearing identifiers must be hashed or redacted before an audit entry is built. Use keyed HMAC hashing in operator-facing environments; `unkeyed_dev_only()` is only for local fixtures and tests. -The writer fails closed: an entry the destination does not accept is an error, and a failed write stops the writer until the process restarts. One process writes one file path and holds a lock beside it for its whole life. Entries are not hash-chained or signed, so a host that can write the audit directory can edit, reorder, or delete entries without detection. Tamper evidence and completeness are deployment properties, not properties of the local file: ship sealed files, or the standard-output stream, to append-only storage outside the host before retention deletes them. +The writer fails closed: an entry the destination does not accept is an error, and a failed write stops the writer until the process restarts. A standard-output destination that holds one line past the fixed write deadline is stalled, not failed: the writer refuses entries while the stall lasts and accepts them again once the destination takes the line. One process writes one file path and holds a lock beside it for its whole life. Entries are not hash-chained or signed, so a host that can write the audit directory can edit, reorder, or delete entries without detection. Tamper evidence and completeness are deployment properties, not properties of the local file: ship sealed files, or the standard-output stream, to append-only storage outside the host before retention deletes them. ## 10. Keep Workspace Hygiene Canonical From 81213e3cc38bf7fb43e4966d91c25fad1afbc0bb Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:17:25 +0700 Subject: [PATCH 2/5] fix(platform): report a stall's start before its end when nothing saw it A stream destination that held only the unfinished response of a dropped request past the write deadline logged the end of the stall and no start, because the start is reported by the append or readiness check that first sees the stall and nothing looked at the stream. The write that ends such a stall now reports its start first, so the operator reads both records. Signed-off-by: Jeremi Joslin --- crates/registry-platform-audit/src/writer.rs | 30 ++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/crates/registry-platform-audit/src/writer.rs b/crates/registry-platform-audit/src/writer.rs index 37204cbd07..dae979e47e 100644 --- a/crates/registry-platform-audit/src/writer.rs +++ b/crates/registry-platform-audit/src/writer.rs @@ -1499,6 +1499,11 @@ impl StreamState { match written { Ok(()) => { if let Some(took) = took.filter(|took| reported || *took >= self.deadline) { + // A stall no append or readiness check saw has its start + // reported here, so no end is reported without a start. + if !reported { + self.report(StreamEvent::StallStarted); + } self.report(StreamEvent::StallEnded(took)); } Ok(()) @@ -6269,6 +6274,31 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_stall_nothing_looked_at_reports_its_start_before_its_end() { + let sink = GatedSink::new(); + let writer = limited(sink.clone(), TEST_DEADLINE, 16); + let state = stream_state(&writer); + // The held line is an unfinished response, which nobody waits on, and + // no append or readiness check looks at the stream while it is held. + writer.append_detached(&AuditEntry::response(SCHEMA, "held", unfinished())); + sink.wait_entered(1).await; + tokio::time::sleep(TEST_DEADLINE).await; + assert!(reported(&state).is_empty()); + + sink.release(); + writer.wait_for_detached_entries(); + assert_eq!(correlations(&sink.buffer), ["held"]); + let reported = reported(&state); + assert!( + matches!( + reported.as_slice(), + [StreamEvent::StallStarted, StreamEvent::StallEnded(_)] + ), + "{reported:?}" + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_write_error_still_stops_the_stream_after_its_sink_recovers() { /// Holds its first write until released, then fails it. Every later From 85748efcbd2fe105d974fefce8d61621c9dab2a5 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:17:33 +0700 Subject: [PATCH 3/5] test(platform): hold the writer drop test to a stall a readiness check saw An append's wait starts before its line is handed to the stream thread, so when that wait runs out the write in progress can be a little younger than the deadline and the refused caller reports no stall. The drop test asserted the start report straight after the refusal and failed whenever the thread picked the line up late. It now waits for readiness to turn false first, and the comment on the refusal states what the code holds. Signed-off-by: Jeremi Joslin --- crates/registry-platform-audit/src/writer.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/registry-platform-audit/src/writer.rs b/crates/registry-platform-audit/src/writer.rs index dae979e47e..2109d6d39f 100644 --- a/crates/registry-platform-audit/src/writer.rs +++ b/crates/registry-platform-audit/src/writer.rs @@ -1320,8 +1320,11 @@ impl LineStream { match outcome.try_recv() { Ok(result) => result, Err(_) => { - // Looked at for its report: a stall that began under - // this caller is logged as this caller is refused. + // Looked at for its report. This wait began before + // the line was handed over, so the write in progress + // can still be younger than the deadline: its stall + // is then reported by the next append or readiness + // check that sees it, or as the write ends. self.state.stalled(); Err(AuditUnavailable::new(AuditUnavailableReason::Stalled)) } @@ -6403,6 +6406,10 @@ mod tests { runtime .block_on(writer.append(request("stuck"))) .expect_err("refused at the deadline"); + // The write began after the refused caller's wait did, so the stall + // is reported once a readiness check finds the line held a whole + // deadline. + runtime.block_on(readiness_becomes(&writer, false)); writer.append_detached(&AuditEntry::response(SCHEMA, "queued", unfinished())); let started = std::time::Instant::now(); From dddff08298822dc2c54e5b18ef42424677cc403d Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:17:33 +0700 Subject: [PATCH 4/5] docs(platform): state where the stream write deadline stops in a product The review record read as an end-to-end result. It now says that the writer's reports reach an operator only where the product's log filter admits the writer's target, names the products on each side, and says that a product logging to the standard output its audit stream writes can block in its own log write, which the deadline does not bound. Signed-off-by: Jeremi Joslin --- crates/registry-platform-audit/README.md | 43 +++++++++++++++++++++--- 1 file changed, 38 insertions(+), 5 deletions(-) diff --git a/crates/registry-platform-audit/README.md b/crates/registry-platform-audit/README.md index eac055a658..4457470b65 100644 --- a/crates/registry-platform-audit/README.md +++ b/crates/registry-platform-audit/README.md @@ -128,7 +128,8 @@ of the change: The deadline sits below the 10-second request timeout the Base Registry Engine defaults to and the Evidence reference projects set, so a stalled destination is answered as an unavailable audit destination, not as a - timed-out request. + timed-out request. The deadline bounds the wait on the audit destination, + not anything else the process writes to the same stream. - **Enforcement point.** `LineStream` in `src/writer.rs`. `append` refuses before queueing when the stream has failed or the write in progress is older than the deadline, hands the line over with `try_send` so a full @@ -160,15 +161,47 @@ of the change: enough that lines wait in the queue for most of a deadline can do the same with no stall reported. The start of a stall is logged by the first append or readiness check that sees it, not by a timer, so a writer nothing is - asking logs it late. A full queue refuses lines without turning `ready()` - false. A destination that never takes its line keeps one thread until the - process exits, and unfinished response entries queued behind it when the - writer is dropped are not written, which is reported once. An operator who + asking logs it late: at the latest when the destination takes the line, + just before the end. The two reports are made outside the stream's lock, + so a destination that takes its line at the instant a caller first sees + the stall can log the end ahead of the start. A full queue refuses lines + without turning `ready()` false and without a report. A destination that + never takes its line keeps one thread until the process exits, and + unfinished response entries queued behind it when the writer is dropped + are not written, which is reported once. An operator who sets a request timeout under 5 seconds still sees a timed-out request first. A refused entry does not undo work a product already committed: a product that writes its outcome entry after its transaction commits reports that operation as unresolved, as it does for a failed write. + Two more limits hold in a product. Both are read from each binary's + logging setup and neither was exercised: + + - Every report this writer makes, the stall and loss reports included, is + logged under the target `registry_platform_audit::writer`, and an + operator sees it only where the product's log filter admits that target. + Evidence (`EVIDENCE_LOG`) and Registry Scheduling (`RUST_LOG`) admit + every target at `info` by default and show the reports. Registry Render + defaults to `registry_render=info` and shows them only when `RUST_LOG` + admits the target. The Base Registry Engine, Registry Casework, Registry + Messaging, `breg-mcp`, and `breg-review` admit only their own targets and + have no setting that admits another, so their operators see the refusals + and the failed readiness, never these reports. + - The Base Registry Engine, Registry Casework, `breg-mcp`, and + `breg-review` write their operational log to standard output, the stream + a `stdout` audit destination writes. While that stream holds an audit + line, a log record written by any thread waits for it, with no deadline. + The Base Registry Engine logs one `info` record for every request and + every probe before the response leaves, at its default level, and + `breg-mcp` and `breg-review` log an `error` record for each refused + audit entry before they answer, so in those three a request the writer + refused can hang instead of being answered. Registry Casework logs no + record on that path and blocks wherever it does log, such as a + background pass that fails. Evidence, Registry Scheduling, Registry + Messaging, and Registry Render log to standard error. The executed proof + above holds its sink through `from_line_sink`; no test blocks a real + standard output. + ## Testing ```sh From 34182ef6c4655ea24785067eed88c48740b8afdc Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:30:16 +0700 Subject: [PATCH 5/5] docs(platform): say which report a full stream queue goes without A full queue refuses an awaited line with no stall report, but an unfinished response entry it refuses is still reported as not accepted. The review record said a full queue refuses lines without a report, which read as covering both. Signed-off-by: Jeremi Joslin --- crates/registry-platform-audit/README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/registry-platform-audit/README.md b/crates/registry-platform-audit/README.md index 4457470b65..29e50b782f 100644 --- a/crates/registry-platform-audit/README.md +++ b/crates/registry-platform-audit/README.md @@ -165,8 +165,8 @@ of the change: just before the end. The two reports are made outside the stream's lock, so a destination that takes its line at the instant a caller first sees the stall can log the end ahead of the start. A full queue refuses lines - without turning `ready()` false and without a report. A destination that - never takes its line keeps one thread until the process exits, and + without turning `ready()` false and without a stall report. A destination + that never takes its line keeps one thread until the process exits, and unfinished response entries queued behind it when the writer is dropped are not written, which is reported once. An operator who sets a request timeout under 5 seconds still sees a timed-out request