From 73e6f5dba028a3a84af3acc275a51d9cf198039c Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 05:09:48 +0530 Subject: [PATCH 1/3] fix(observability): bound repeated durable-append failure reports The drain loop reported every failed durable append with a bare `eprintln!`, so a persistent sink failure (read-only volume, full disk) emitted one identical line per arriving observation and drowned the host's stderr. Failed appends were also uncounted, unlike queue-full drops, leaving the flood itself as the only signal that the durable log was losing data. Replace it with a per-worker failure-run state machine local to the drain thread: the first failure of a run reports at ERROR, subsequent failures are counted silently with a WARN reminder at most once per APPEND_REPORT_COOLDOWN (5 minutes) carrying the latest error, the first success emits one WARN recovery summary with the number of observations lost, and a shutdown that is still degraded emits a final WARN so a never-recovering run is not silently quiet. Reporting moves to `tracing` on the `tinyagents::observability` target with a `sink` field, matching the crate's existing emission idiom and removing its only `eprintln!`. Add an `append_failures` counter mirroring `dropped` so durable-log loss has a subscriber-independent signal, and keep attempting every item while degraded: the attempt is what detects recovery, it runs off the run's critical path, and skipping it would turn a transient blip into guaranteed loss of everything still queued. BEHAVIOR CHANGE: the stderr line is gone. An embedder with no tracing subscriber installed now sees nothing on append failure and must read `append_failures` instead. The message keeps the literal substring "durable append failed" so existing log searches still match. --- src/harness/observability/test.rs | 107 ++++++++++++++++++++ src/harness/observability/worker.rs | 145 ++++++++++++++++++++++++++-- 2 files changed, 245 insertions(+), 7 deletions(-) diff --git a/src/harness/observability/test.rs b/src/harness/observability/test.rs index 3d2ed68..c222021 100644 --- a/src/harness/observability/test.rs +++ b/src/harness/observability/test.rs @@ -4,6 +4,7 @@ use std::sync::Arc; +use crate::error::TinyAgentsError; use crate::harness::events::{AgentEvent, EventListener, EventRecord, HarnessRunStatus, LimitKind}; use crate::harness::ids::{CallId, ComponentId, EventId, ExecutionStatus, RunId, ThreadId}; use crate::harness::observability::AppendWorker; @@ -362,6 +363,112 @@ async fn append_worker_drops_and_counts_when_queue_is_full() { ); } +#[tokio::test] +async fn append_worker_counts_failed_appends() { + // A sink that rejects every append: each failure is counted, and that count + // is kept separate from the queue-full drop count (nothing was dropped — + // every item reached the sink and was rejected by it). + let worker = AppendWorker::spawn("test-failing", 64, move |_n: u64| async move { + Err(TinyAgentsError::Storage("sink offline".into())) + }); + + for n in 0..12 { + worker.submit(n); + } + worker.flush(); + + // Both loss counters are surfaced on the debug view, so an operator dumping + // a sink can tell "never reached the backend" from "backend rejected it". + let debug = format!("{worker:?}"); + assert!( + debug.contains("append_failures: 12") && debug.contains("dropped: 0"), + "debug view must distinguish append failures from queue-full drops, got {debug}" + ); + + assert_eq!( + worker.append_failures(), + 12, + "every failed durable append must be counted" + ); + assert_eq!( + worker.dropped(), + 0, + "append failures must not be conflated with queue-full drops" + ); +} + +#[tokio::test] +async fn append_worker_failure_then_recovery_keeps_attempting() { + use std::sync::Mutex; + + // The sink fails the first 3 appends, then starts accepting. A worker that + // stopped attempting while degraded would never persist anything after the + // failure run; keeping attempts is what detects the recovery. + let attempts = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let counter = Arc::clone(&attempts); + let sink = Arc::clone(&seen); + // A zero cooldown makes the "still failing" reminder fire on every failure + // after the first, exercising the suppression path without a wall clock. + let worker = AppendWorker::spawn_with_cooldown( + "test-flaky", + 64, + std::time::Duration::ZERO, + move |n: u64| { + let counter = Arc::clone(&counter); + let sink = Arc::clone(&sink); + async move { + if counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed) < 3 { + return Err(TinyAgentsError::Storage("sink offline".into())); + } + sink.lock().unwrap().push(n); + Ok(()) + } + }, + ); + + for n in 0..8 { + worker.submit(n); + } + worker.flush(); + + assert_eq!( + worker.append_failures(), + 3, + "only the appends the sink rejected are counted as failures" + ); + assert_eq!( + *seen.lock().unwrap(), + (3..8).collect::>(), + "the worker must keep attempting while degraded, so items persist once the sink recovers" + ); +} + +#[test] +fn should_report_bounds_repeats_to_one_per_cooldown() { + use crate::harness::observability::worker::should_report; + use std::time::{Duration, Instant}; + + let cooldown = Duration::from_secs(300); + let start = Instant::now(); + + // The first failure of a run has never reported, so it always reports. + assert!(should_report(None, start, cooldown)); + // A failure inside the cooldown window is suppressed (counted, not logged). + assert!(!should_report( + Some(start), + start + Duration::from_secs(299), + cooldown + )); + // Once the cooldown has elapsed, the run is due for a reminder again. + assert!(should_report(Some(start), start + cooldown, cooldown)); + assert!(should_report( + Some(start), + start + Duration::from_secs(600), + cooldown + )); +} + /// Collects forwarded records for assertions. struct Collector { records: std::sync::Mutex>, diff --git a/src/harness/observability/worker.rs b/src/harness/observability/worker.rs index 2a27094..3d8e46b 100644 --- a/src/harness/observability/worker.rs +++ b/src/harness/observability/worker.rs @@ -19,9 +19,33 @@ //! the run). Callers that need a lossless log can inspect the dropped count. //! //! # Error policy -//! Append errors are **not** silently discarded: the drain loop reports each -//! failure to stderr with the sink name. Persistence remains best-effort — an -//! error never propagates back into the run — but it is observable. +//! Append errors are **not** silently discarded, but neither are they allowed to +//! flood the host's log. Every failed append is counted in +//! [`AppendWorker::append_failures`] (a lifetime total, distinct from the +//! queue-full [`AppendWorker::dropped`] count), and the drain loop reports +//! through `tracing` on the `tinyagents::observability` target with a `sink` +//! field: +//! +//! - the **first** failure of a failure run is reported at `ERROR` — durable +//! observations are being lost; +//! - subsequent failures are counted silently, with a reminder at `WARN` at most +//! once per [`APPEND_REPORT_COOLDOWN`]; each reminder carries the *latest* +//! error, so a changed cause (read-only volume → full disk) still surfaces +//! within one cooldown; +//! - the first success after a failure run emits one `WARN` recovery summary +//! carrying how many observations were lost while the sink was down; +//! - if the worker shuts down while still failing, it emits one final `WARN` +//! summary so a never-recovering run is not silently quiet. +//! +//! The worker keeps attempting every item while degraded: the attempt *is* the +//! recovery detector, it runs off the run's critical path, and skipping it would +//! turn a transient blip into guaranteed loss of everything still queued. Items +//! that fail are lost, but counted. +//! +//! Persistence remains best-effort — an error never propagates back into the run +//! — but it is observable. Note that reporting goes through `tracing`, so an +//! embedder with no subscriber installed sees nothing; inspect +//! [`AppendWorker::append_failures`] for a subscriber-independent signal. //! //! # Ordering & durability boundary //! A single drain thread preserves submit order. [`AppendWorker::flush`] blocks @@ -35,6 +59,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc::{Sender, SyncSender, TrySendError, sync_channel}; use std::thread::JoinHandle; +use std::time::{Duration, Instant}; use crate::error::Result; @@ -44,6 +69,27 @@ use crate::error::Result; /// the drop policy engages, without letting the queue grow without bound. pub(crate) const DEFAULT_DRAIN_CAPACITY: usize = 1024; +/// Minimum interval between repeated reports of an *ongoing* append-failure run. +/// +/// A persistent sink failure (a volume flipped read-only, a full disk) fails +/// every queued observation identically; without a cooldown that is one log line +/// per arriving event. The first failure is always reported; after that the +/// drain loop stays quiet for this long between reminders. +pub(crate) const APPEND_REPORT_COOLDOWN: Duration = Duration::from_secs(300); + +/// Whether an ongoing failure run is due for a reminder report. +/// +/// `last` is when this run last reported (`None` before its first report). +/// Pure and clock-free so the cooldown policy is unit-testable without a +/// subscriber or a fake clock. +pub(super) fn should_report(last: Option, now: Instant, cooldown: Duration) -> bool { + match last { + None => true, + // Saturating: a non-monotonic `now` yields zero, not a panic. + Some(previous) => now.saturating_duration_since(previous) >= cooldown, + } +} + /// Messages carried over the drain channel. enum Msg { /// A payload to persist. @@ -63,6 +109,8 @@ pub(crate) struct AppendWorker { tx: Option>>, /// Count of payloads dropped because the queue was full (or disconnected). dropped: Arc, + /// Lifetime count of payloads whose durable append returned an error. + append_failures: Arc, /// Handle to the drain thread, joined on drop. handle: Option>, /// Human-readable sink name used in error reports. @@ -76,11 +124,31 @@ impl AppendWorker { /// runtime; it returns the async append future. `name` labels the drain /// thread and error reports. pub(crate) fn spawn(name: &'static str, capacity: usize, append: F) -> Self + where + F: Fn(T) -> Fut + Send + 'static, + Fut: Future>, + { + Self::spawn_with_cooldown(name, capacity, APPEND_REPORT_COOLDOWN, append) + } + + /// Spawns a drain worker with an explicit failure-report cooldown. + /// + /// Same as [`Self::spawn`], which supplies [`APPEND_REPORT_COOLDOWN`]. The + /// cooldown is a parameter so the suppression and reminder paths can be + /// exercised deterministically, without waiting on wall-clock time. + pub(crate) fn spawn_with_cooldown( + name: &'static str, + capacity: usize, + cooldown: Duration, + append: F, + ) -> Self where F: Fn(T) -> Fut + Send + 'static, Fut: Future>, { let (tx, rx) = sync_channel::>(capacity.max(1)); + let append_failures = Arc::new(AtomicU64::new(0)); + let failures = Arc::clone(&append_failures); let handle = std::thread::Builder::new() .name(format!("tinyagents-{name}-drain")) .spawn(move || { @@ -96,24 +164,72 @@ impl AppendWorker { } }; rt.block_on(async move { + // Failure-run state: how many appends have failed since the + // last success, and when this run last reported. Local to + // the drain thread — no shared state, no timer tasks. + let mut failure_run: u64 = 0; + let mut last_report: Option = None; while let Ok(msg) = rx.recv() { match msg { - Msg::Item(item) => { - if let Err(e) = append(item).await { - eprintln!("tinyagents: {name} durable append failed: {e}"); + Msg::Item(item) => match append(item).await { + Ok(()) => { + if failure_run > 0 { + tracing::warn!( + target: "tinyagents::observability", + sink = name, + lost = failure_run, + "[observability] durable append recovered; {failure_run} observation(s) lost while the sink was failing" + ); + failure_run = 0; + last_report = None; + } } - } + Err(error) => { + failures.fetch_add(1, Ordering::Relaxed); + failure_run += 1; + let now = Instant::now(); + if failure_run == 1 { + last_report = Some(now); + tracing::error!( + target: "tinyagents::observability", + sink = name, + error = %error, + "[observability] durable append failed; the observation is lost and repeats are suppressed until the sink recovers" + ); + } else if should_report(last_report, now, cooldown) { + last_report = Some(now); + tracing::warn!( + target: "tinyagents::observability", + sink = name, + error = %error, + failures = failure_run, + "[observability] durable append failed; still failing after {failure_run} consecutive observations" + ); + } + } + }, Msg::Flush(ack) => { let _ = ack.send(()); } } } + // The channel closed mid-failure: report once on the way out + // so a run that never recovered is not silently quiet. + if failure_run > 0 { + tracing::warn!( + target: "tinyagents::observability", + sink = name, + lost = failure_run, + "[observability] durable append failed; sink never recovered before shutdown, {failure_run} observation(s) lost" + ); + } }); }) .expect("spawn durable-drain thread"); Self { tx: Some(tx), dropped: Arc::new(AtomicU64::new(0)), + append_failures, handle: Some(handle), name, } @@ -141,6 +257,17 @@ impl AppendWorker { self.dropped.load(Ordering::Relaxed) } + /// Returns the number of payloads whose durable append returned an error. + /// + /// Distinct from [`Self::dropped`]: those never reached the sink, these were + /// attempted and rejected. Reporting is `tracing`-based and suppressed while + /// a failure run continues, so this counter is the only subscriber-free + /// signal of durable-log loss. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn append_failures(&self) -> u64 { + self.append_failures.load(Ordering::Relaxed) + } + /// Blocks until every payload submitted before this call has been persisted. pub(crate) fn flush(&self) { let Some(tx) = self.tx.as_ref() else { @@ -160,6 +287,10 @@ impl fmt::Debug for AppendWorker { f.debug_struct("AppendWorker") .field("name", &self.name) .field("dropped", &self.dropped.load(Ordering::Relaxed)) + .field( + "append_failures", + &self.append_failures.load(Ordering::Relaxed), + ) .finish_non_exhaustive() } } From 34c28cb5b7c64dd950440f2c9327c3477106495b Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 05:09:55 +0530 Subject: [PATCH 2/3] docs(observability): align error policy with tracing-based reporting Both module READMEs still described backend errors as "reported to stderr", which the rate-limited tracing reporter makes false. Describe the actual policy: errors are counted in `append_failures` and reported on the `tinyagents::observability` target, rate-limited to one ERROR per failure run plus a WARN reminder per cooldown and a WARN recovery summary, with the worker still attempting every item while degraded. Note that reporting requires a tracing subscriber, so `append_failures` is the subscriber-independent signal. --- src/graph/observability/README.md | 7 +++++-- src/harness/observability/README.md | 22 +++++++++++++++++----- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/src/graph/observability/README.md b/src/graph/observability/README.md index 46cd611..08654de 100644 --- a/src/graph/observability/README.md +++ b/src/graph/observability/README.md @@ -56,8 +56,11 @@ so existing runs are unchanged. async journal API through a background `AppendWorker`: `emit` hands the observation to a bounded channel drained on a dedicated thread, so it never blocks the executor on I/O. Persistence is **best-effort**: a full queue drops -(and counts) the overflow, backend errors are reported to stderr rather than -propagated, and neither aborts the run. The executor calls `GraphEventSink::flush` +(and counts) the overflow, backend errors are counted and reported through +`tracing` (rate-limited to one report per failure run plus a reminder every five +minutes — see the harness `observability` README for the full error policy) +rather than propagated, and neither aborts the run. The executor calls +`GraphEventSink::flush` after the terminal run event (and callers can call it directly) to block until the durable log has caught up. Do not rely on the sink for delivery guarantees stronger than "usually persisted, never run-blocking." diff --git a/src/harness/observability/README.md b/src/harness/observability/README.md index 30e55a1..55a30c1 100644 --- a/src/harness/observability/README.md +++ b/src/harness/observability/README.md @@ -56,11 +56,23 @@ Persisting sinks (`JournalSink`, `JsonlSink`) bridge the synchronous background `AppendWorker`: `on_event` hands the observation to a **bounded** channel drained on a dedicated thread, so it never blocks the run on I/O. Persistence is **best-effort**: if the queue is full the observation is dropped -(and counted), backend errors are reported to stderr rather than propagated, -and neither ever aborts the run. Call `JournalSink::flush` / `JsonlSink::flush` -to block until the durable log has caught up (for example before reading it back -or shutting down). Do not rely on a sink for delivery guarantees stronger than -"usually persisted, never run-blocking." +(and counted in `AppendWorker::dropped`), backend errors are counted (in +`AppendWorker::append_failures`) and reported through `tracing` rather than +propagated, and neither ever aborts the run. Call `JournalSink::flush` / +`JsonlSink::flush` to block until the durable log has caught up (for example +before reading it back or shutting down). Do not rely on a sink for delivery +guarantees stronger than "usually persisted, never run-blocking." + +A persistent backend failure (read-only volume, full disk) fails every queued +observation identically, so reporting is **rate-limited** rather than +per-observation: the first failure of a run logs at `ERROR` on the +`tinyagents::observability` target, further failures are counted silently with a +`WARN` reminder at most once per `APPEND_REPORT_COOLDOWN` (5 minutes), and the +first success afterwards logs one `WARN` recovery summary carrying how many +observations were lost. The worker keeps attempting every item while degraded — +the attempt is what detects recovery. Because reporting goes through `tracing`, +an embedder with no subscriber installed sees nothing; read `append_failures` +for a subscriber-independent signal. ## Latency metrics semantics From 98cfdce84e32d078d610b1b4d3a0185f36727b42 Mon Sep 17 00:00:00 2001 From: oxoxDev Date: Wed, 12 Aug 2026 11:06:56 +0530 Subject: [PATCH 3/3] docs(observability): stop promising a counter a host cannot read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both the module docs and the README told embedders to "read append_failures for a subscriber-independent signal". They cannot: like the queue-full `dropped` count it mirrors, the counter is `pub(crate)` and neither `JournalSink` nor `JsonlSink` exposes it. That mattered because it was the stated consolation for removing the stderr line — so the one claim softening the behaviour change was the one that was not true. Say plainly that a host with no subscriber sees nothing, and that installing one is how durable-log loss is observed. --- src/harness/observability/README.md | 6 ++++-- src/harness/observability/worker.rs | 8 ++++++-- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/src/harness/observability/README.md b/src/harness/observability/README.md index 55a30c1..6abc7f2 100644 --- a/src/harness/observability/README.md +++ b/src/harness/observability/README.md @@ -71,8 +71,10 @@ per-observation: the first failure of a run logs at `ERROR` on the first success afterwards logs one `WARN` recovery summary carrying how many observations were lost. The worker keeps attempting every item while degraded — the attempt is what detects recovery. Because reporting goes through `tracing`, -an embedder with no subscriber installed sees nothing; read `append_failures` -for a subscriber-independent signal. +an embedder with no subscriber installed sees nothing — install a subscriber to +observe durable-log loss. `AppendWorker::append_failures` counts it, but like +the queue-full `dropped` count it is crate-internal, so it is not a signal a +host application can read today. ## Latency metrics semantics diff --git a/src/harness/observability/worker.rs b/src/harness/observability/worker.rs index 3d8e46b..7484da2 100644 --- a/src/harness/observability/worker.rs +++ b/src/harness/observability/worker.rs @@ -44,8 +44,12 @@ //! //! Persistence remains best-effort — an error never propagates back into the run //! — but it is observable. Note that reporting goes through `tracing`, so an -//! embedder with no subscriber installed sees nothing; inspect -//! [`AppendWorker::append_failures`] for a subscriber-independent signal. +//! embedder with no subscriber installed sees nothing at all: installing one is +//! how a host observes durable-log loss. [`AppendWorker::append_failures`] +//! counts it, but — like the queue-full [`AppendWorker::dropped`] count it +//! mirrors — it is crate-internal and is not reachable from a host application +//! through [`JournalSink`](super::JournalSink) or +//! [`JsonlSink`](super::JsonlSink). //! //! # Ordering & durability boundary //! A single drain thread preserves submit order. [`AppendWorker::flush`] blocks