From fd8c7854cbea462f70ff921959700fdf6a11db48 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sun, 18 Oct 2026 15:57:55 +0000 Subject: [PATCH 1/2] test(TC-1317): review follow-ups F1-F4 - F1: the stranger's read is denied (401/403) on /signed/kv and /encryption/networks while the read audit is held, not only on /invoke. - F2: export tinycloud_read_audit_stopped_total and the tinycloud_read_audit_drain_stopped gauge; test that a dead drain is counted and reported as a stalled drain. - F3: the slow-batch time includes the SQLite writer-lock wait; say so. - F4: the deadline test checks Retry-After: 1 and stage=read_audit. --- CHANGELOG.md | 2 +- tinycloud-core/src/read_audit.rs | 45 ++++++- tinycloud-node-server/src/prometheus.rs | 14 ++ tinycloud-node-server/src/routes/mod.rs | 170 +++++++++++++++++++----- 4 files changed, 196 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 015cde80..ef155233 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ - **The invariant is unchanged and tested:** no read's data is returned unless its audit record has committed. - **Abandoned records are skipped.** When a batch forms, records whose reader already gave up or was cut are dropped and counted instead of committed. Their data was never returned. This lets a backlog drain on its own instead of needing a restart, and keeps cut SQL requests out of the database-migration attribution evidence. - **Panic isolation.** A panic while committing one batch fails only that batch; the queue keeps working. - - **Visibility.** New Prometheus metrics: `tinycloud_read_audit_failed_batches_total`, `tinycloud_read_audit_abandoned_total`, `tinycloud_read_audit_busy_total`, and the gauges `tinycloud_read_audit_queue_depth` and `tinycloud_read_audit_in_flight_batch_age_ms`. A watchdog logs `Read audit wait expired`, `Read audit batch slow` (over 5 s), `Read audit batch failed` and `Read audit drain stalled` (a batch in flight over 30 s, including a SQLite writer-lock wait; queued records with no progress for 10 s; or a dead drain, whose reads fail as stopped), each at most once a minute, with counts and durations only. The log shipper maps them to `read_audit_busy`, `read_audit_batch_slow`, `read_audit_batch_failed` and `read_audit_drain_stalled`. A request cut by the `/invoke` deadline while it waits on its audit now logs `stage="read_audit"`. + - **Visibility.** New Prometheus metrics: `tinycloud_read_audit_failed_batches_total`, `tinycloud_read_audit_abandoned_total`, `tinycloud_read_audit_busy_total`, `tinycloud_read_audit_stopped_total` (reads that failed because the drain was gone), and the gauges `tinycloud_read_audit_queue_depth`, `tinycloud_read_audit_in_flight_batch_age_ms` and `tinycloud_read_audit_drain_stopped` (1 once the drain is gone). A watchdog logs `Read audit wait expired`, `Read audit batch slow` (a batch in flight over 5 s; on SQLite that time, and its `max_commit_ms`, include the wait for the writer lock, not only the commit), `Read audit batch failed` and `Read audit drain stalled` (a batch in flight over 30 s, including a SQLite writer-lock wait; queued records with no progress for 10 s; or a dead drain, whose reads fail as stopped), each at most once a minute, with counts and durations only. The log shipper maps them to `read_audit_busy`, `read_audit_batch_slow`, `read_audit_batch_failed` and `read_audit_drain_stalled`. A request cut by the `/invoke` deadline while it waits on its audit now logs `stage="read_audit"`. - **N0 compose.** The embedded Vector configuration is regenerated for these codes, so the reviewed N0 compose template changes to `7fabba70…` (raw `1228d819…`; `docs/releasing.md`). - **Not covered.** An overall request deadline for `/signed/kv` and `/encryption/networks` is TC-1418; the audit queue's throughput is TC-321. - SQL requests no longer wait without limit, and `/invoke` has a deadline (TC-1274). The cause of the original stall is still unknown; this change bounds the waits and reports them. diff --git a/tinycloud-core/src/read_audit.rs b/tinycloud-core/src/read_audit.rs index 3af145ab..cd4e361e 100644 --- a/tinycloud-core/src/read_audit.rs +++ b/tinycloud-core/src/read_audit.rs @@ -103,7 +103,8 @@ struct PipelineStats { failed_batches: AtomicU64, abandoned: AtomicU64, busy: AtomicU64, - /// Batches whose commit took longer than [`SLOW_BATCH`], and the longest. + /// Batches in flight longer than [`SLOW_BATCH`], and the longest. On + /// SQLite this includes the wait for the writer lock, not only the commit. slow_batches: AtomicU64, max_commit_ms: AtomicU64, /// When the batch now being committed started, in ms since `epoch` plus @@ -186,7 +187,8 @@ pub struct ReadAuditHealth { pub drain_stopped: bool, } -/// A batch commit slower than this is reported (TC-1317). +/// A batch in flight longer than this (the commit, plus the SQLite writer-lock +/// wait) is reported (TC-1317). const SLOW_BATCH: Duration = Duration::from_secs(5); /// A batch in flight longer than this, or a queue that waited longer than /// [`STALLED_QUEUE_WAIT`] with no batch completing, is a stalled drain. @@ -2213,6 +2215,45 @@ mod tests { assert_eq!(stalled(&state), Some(later + Duration::from_secs(1))); } + /// A drain that is gone answers every read `Stopped`, counts it, and is + /// reported as a stalled drain until it comes back, which it never does. + #[tokio::test] + async fn tc1317_a_dead_drain_is_counted_and_reported() { + let (sender, receiver) = mpsc::channel(CHANNEL_CAPACITY); + drop(receiver); + let pipeline = ReadAuditPipeline { + sender, + stats: Arc::default(), + wait_ms: Arc::new(AtomicU64::new(PATIENCE.as_millis() as u64)), + gate: Arc::default(), + }; + let stalled = |state: &WatchdogState| state.last[3]; + let mut state = WatchdogState::default(); + let start = Instant::now(); + + // Dead before any read: reported on the first look. + let idle = pipeline.health(); + assert!(idle.drain_stopped); + assert_eq!(idle.stopped, 0); + watch(&pipeline.stats, &mut state, idle, start); + assert_eq!(stalled(&state), Some(start)); + + for index in 0..2 { + assert_eq!( + bounded(&pipeline, index).await, + Err(ReadAuditError::Stopped) + ); + } + let failing = pipeline.health(); + assert_eq!((failing.stopped, failing.busy), (2, 0)); + watch(&pipeline.stats, &mut state, failing, start + WATCHDOG_TICK); + assert_eq!(stalled(&state), Some(start), "rate-limited"); + let later = start + WARNING_INTERVAL; + watch(&pipeline.stats, &mut state, failing, later); + assert_eq!(stalled(&state), Some(later), "still dead, reported again"); + assert_eq!(state.reported.stopped, 2); + } + #[tokio::test] async fn tc1317_a_held_drain_answers_busy_within_the_bound() { let (_directory, _conn, pipeline) = file_pipeline().await; diff --git a/tinycloud-node-server/src/prometheus.rs b/tinycloud-node-server/src/prometheus.rs index b16c7c67..d3efa80e 100644 --- a/tinycloud-node-server/src/prometheus.rs +++ b/tinycloud-node-server/src/prometheus.rs @@ -123,6 +123,18 @@ lazy_static! { "Reads answered read_audit_busy because their audit record was not committed in time." ) .unwrap(); + /// Reads answered as stopped because the read-audit drain was gone. + pub static ref READ_AUDIT_STOPPED: IntCounter = register_int_counter!( + "tinycloud_read_audit_stopped_total", + "Reads that failed because the read-audit drain was gone." + ) + .unwrap(); + /// 1 once the read-audit drain is gone and nothing can commit again. + pub static ref READ_AUDIT_DRAIN_STOPPED: IntGauge = register_int_gauge!( + "tinycloud_read_audit_drain_stopped", + "1 if the read-audit drain is gone and no read can be confirmed, else 0." + ) + .unwrap(); /// Read-audit records waiting in the queue (TC-1317). pub static ref READ_AUDIT_QUEUE_DEPTH: IntGauge = register_int_gauge!( "tinycloud_read_audit_queue_depth", @@ -227,8 +239,10 @@ pub fn observe_read_audit( .inc_by(now.failed_batches.saturating_sub(previous.failed_batches)); READ_AUDIT_ABANDONED.inc_by(now.abandoned.saturating_sub(previous.abandoned)); READ_AUDIT_BUSY.inc_by(now.busy.saturating_sub(previous.busy)); + READ_AUDIT_STOPPED.inc_by(now.stopped.saturating_sub(previous.stopped)); READ_AUDIT_QUEUE_DEPTH.set(now.queue_depth as i64); READ_AUDIT_IN_FLIGHT_MS.set(now.in_flight_batch_age_ms as i64); + READ_AUDIT_DRAIN_STOPPED.set(i64::from(now.drain_stopped)); } } diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 3851a8d6..98799081 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -9708,34 +9708,82 @@ mod tests { .await??; busy(reply.status, &reply.body, "sql"); - // A real denial is still a denial: a stranger reading the space. + // A real denial is still a denial on every reclassified route: a + // stranger reading the space. let stranger = JWK::generate_ed25519()?; let stranger_vm = tc341_verification_method(&stranger)?; - let mut capabilities = Capabilities::new(); - let (uri, action) = kv_get()?; - capabilities.with_action( - uri.parse()?, - action.parse::()?, - [std::collections::BTreeMap::::new()], + let stranger_invocation = |audience: &str| -> Result { + let mut capabilities = Capabilities::new(); + let (uri, action) = kv_get()?; + capabilities.with_action( + uri.parse()?, + action.parse::()?, + [std::collections::BTreeMap::::new()], + ); + Ok(Payload { + issuer: stranger_vm.parse::()?, + audience: audience.parse::()?, + not_before: None, + expiration: test_invocation_expiration(), + nonce: Some(fixture.next_nonce()), + facts: Some(Vec::::new()), + proof: vec![], + attenuation: capabilities, + } + .sign(stranger.get_algorithm().unwrap_or_default(), &stranger)? + .encode()?) + }; + let denial = |status: Status, body: &str, route: &str| { + assert!( + [Status::Unauthorized, Status::Forbidden].contains(&status), + "{route}: a denial must not turn into busy: {status} {body}" + ); + }; + let space_did = fixture.space.did().to_string(); + let reply = within(fixture.dispatch(stranger_invocation(&space_did)?, &[])).await?; + denial(reply.status, &reply.body, "/invoke"); + + let response = within( + fixture + .client + .post("/signed/kv") + .header(Header::new( + "Authorization", + stranger_invocation(&space_did)?, + )) + .header(ContentType::JSON) + .body( + serde_json::json!({ "space": fixture.space.to_string(), "path": "notes/a" }) + .to_string(), + ) + .dispatch(), + ) + .await?; + let status = response.status(); + denial( + status, + &response.into_string().await.unwrap_or_default(), + "/signed/kv", ); - let denied = Payload { - issuer: stranger_vm.parse::()?, - audience: fixture.space.did().to_string().parse::()?, - not_before: None, - expiration: test_invocation_expiration(), - nonce: Some(fixture.next_nonce()), - facts: Some(Vec::::new()), - proof: vec![], - attenuation: capabilities, - } - .sign(stranger.get_algorithm().unwrap_or_default(), &stranger)? - .encode()?; - let reply = within(fixture.dispatch(denied, &[])).await?; - assert!( - [Status::Unauthorized, Status::Forbidden].contains(&reply.status), - "a denial must not turn into busy: {} {}", - reply.status, - reply.body + + let response = within( + fixture + .client + .post("/encryption/networks") + .header(Header::new( + "Authorization", + stranger_invocation(&fixture.node_did)?, + )) + .header(ContentType::JSON) + .body("{}") + .dispatch(), + ) + .await?; + let status = response.status(); + denial( + status, + &response.into_string().await.unwrap_or_default(), + "/encryption/networks", ); drop(held); @@ -9756,14 +9804,62 @@ mod tests { } /// TC-1317 review R2: a deadline shorter than the read-audit wait cuts a - /// read waiting on its audit with `503 request_timeout`, no data. + /// read waiting on its audit with `503 request_timeout` and + /// `Retry-After: 1`, no data, and logs that the time went to the audit. #[tokio::test] async fn a_deadline_before_the_audit_wait_answers_request_timeout() -> Result<()> { - use rocket::http::Status; + use rocket::http::{Header, Status}; + + /// Records the `stage` of each `Request deadline exceeded` warning. + /// Spans are ignored: a span opened on this thread may close on a + /// database worker thread, which a span registry would reject. + #[derive(Clone, Default)] + struct DeadlineStages(Arc>>); + #[derive(Default)] + struct Fields { + message: String, + stage: String, + } + impl tracing::field::Visit for Fields { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "stage" { + self.stage = value.to_string(); + } + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.message = format!("{value:?}"); + } + } + } + impl tracing::Subscriber for DeadlineStages { + fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { + tracing::span::Id::from_u64(1) + } + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + let mut fields = Fields::default(); + event.record(&mut fields); + if fields.message == "Request deadline exceeded" { + self.0.lock().unwrap().push(fields.stage); + } + } + fn enter(&self, _: &tracing::span::Id) {} + fn exit(&self, _: &tracing::span::Id) {} + } + let stages = DeadlineStages::default(); + let _logging = tracing::subscriber::set_default(stages.clone()); let mut config = Config::default(); config.invocation.deadline_ms = 300; - let fixture = KvSyncFixture::new_configured(config, |rocket| rocket).await?; + let fixture = KvSyncFixture::new_configured(config, |rocket| { + rocket.attach(crate::retry_after_fairing()) + }) + .await?; fixture.owner_write("notes/a", Some(b"secret")).await?; let tinycloud = fixture .client @@ -9778,14 +9874,24 @@ mod tests { .sqlite_writer_lock() .expect("the fixture runs on SQLite"); let held = writer.lock().await; - let reply = tokio::time::timeout( + let response = tokio::time::timeout( std::time::Duration::from_secs(10), - fixture.dispatch(get, &[]), + fixture + .client + .post("/invoke") + .header(Header::new("Authorization", get)) + .dispatch(), ) .await?; drop(held); - assert_eq!(reply.status, Status::ServiceUnavailable, "{}", reply.body); - assert_eq!(reply.body, REQUEST_TIMEOUT_BODY); + assert_eq!(response.status(), Status::ServiceUnavailable); + assert_eq!(response.headers().get_one("Retry-After"), Some("1")); + assert_eq!( + response.into_string().await.as_deref(), + Some(REQUEST_TIMEOUT_BODY) + ); + + assert_eq!(*stages.0.lock().unwrap(), ["read_audit"]); Ok(()) } From bd46936bc77d439301aaa34dd40ec8e8ec7dd145 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sun, 18 Oct 2026 17:00:43 +0000 Subject: [PATCH 2/2] test(TC-1317): check the read-audit stage in core, not through logs (B1) The route test's log capture depended on test order: tracing-core caches one call-site interest per process, computed from whichever thread's dispatcher reaches the call site first. Assert RequestCancel::stage() in a core test instead; the route test keeps the 503, Retry-After: 1 and the request_timeout body. --- tinycloud-core/src/read_audit.rs | 33 ++++++++++++++++ tinycloud-node-server/src/routes/mod.rs | 50 ++----------------------- 2 files changed, 36 insertions(+), 47 deletions(-) diff --git a/tinycloud-core/src/read_audit.rs b/tinycloud-core/src/read_audit.rs index cd4e361e..12d3c08f 100644 --- a/tinycloud-core/src/read_audit.rs +++ b/tinycloud-core/src/read_audit.rs @@ -2254,6 +2254,39 @@ mod tests { assert_eq!(state.reported.stopped, 2); } + /// A request's deadline context says the time is going to the read + /// audit while a read waits for its record, and stops saying so once the + /// wait ends: a deadline that cuts the wait logs `stage = read_audit`. + #[tokio::test] + async fn tc1317_a_read_waiting_on_its_audit_is_in_the_read_audit_stage() { + use crate::request_cancel::{within, RequestCancel, Stage}; + + let (_directory, _conn, pipeline) = file_pipeline().await; + pipeline.set_wait(COMMIT_TEST_WAIT); + let context = RequestCancel::new(COMMIT_TEST_WAIT); + let held = pipeline.hold_drain().await; + let read = tokio::spawn({ + let (pipeline, context) = (pipeline.clone(), context.clone()); + async move { within(Some(context), pipeline.enqueue(record(0))).await } + }); + tokio::time::timeout(PATIENCE, async { + while context.stage() != Stage::ReadAudit { + tokio::task::yield_now().await; + } + }) + .await + .expect("the waiting read must be in the read-audit stage"); + assert!(!read.is_finished(), "still waiting on the held drain"); + + drop(held); + let outcome = tokio::time::timeout(PATIENCE, read) + .await + .expect("released drain commits") + .unwrap(); + assert_eq!(outcome, Ok(())); + assert_eq!(context.stage(), Stage::Other); + } + #[tokio::test] async fn tc1317_a_held_drain_answers_busy_within_the_bound() { let (_directory, _conn, pipeline) = file_pipeline().await; diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 98799081..96ee9e35 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -9805,55 +9805,13 @@ mod tests { /// TC-1317 review R2: a deadline shorter than the read-audit wait cuts a /// read waiting on its audit with `503 request_timeout` and - /// `Retry-After: 1`, no data, and logs that the time went to the audit. + /// `Retry-After: 1`, no data. That the wait is reported as + /// `stage = read_audit` is tested in core + /// (`tc1317_a_read_waiting_on_its_audit_is_in_the_read_audit_stage`). #[tokio::test] async fn a_deadline_before_the_audit_wait_answers_request_timeout() -> Result<()> { use rocket::http::{Header, Status}; - /// Records the `stage` of each `Request deadline exceeded` warning. - /// Spans are ignored: a span opened on this thread may close on a - /// database worker thread, which a span registry would reject. - #[derive(Clone, Default)] - struct DeadlineStages(Arc>>); - #[derive(Default)] - struct Fields { - message: String, - stage: String, - } - impl tracing::field::Visit for Fields { - fn record_str(&mut self, field: &tracing::field::Field, value: &str) { - if field.name() == "stage" { - self.stage = value.to_string(); - } - } - fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { - if field.name() == "message" { - self.message = format!("{value:?}"); - } - } - } - impl tracing::Subscriber for DeadlineStages { - fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { - true - } - fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { - tracing::span::Id::from_u64(1) - } - fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} - fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} - fn event(&self, event: &tracing::Event<'_>) { - let mut fields = Fields::default(); - event.record(&mut fields); - if fields.message == "Request deadline exceeded" { - self.0.lock().unwrap().push(fields.stage); - } - } - fn enter(&self, _: &tracing::span::Id) {} - fn exit(&self, _: &tracing::span::Id) {} - } - let stages = DeadlineStages::default(); - let _logging = tracing::subscriber::set_default(stages.clone()); - let mut config = Config::default(); config.invocation.deadline_ms = 300; let fixture = KvSyncFixture::new_configured(config, |rocket| { @@ -9890,8 +9848,6 @@ mod tests { response.into_string().await.as_deref(), Some(REQUEST_TIMEOUT_BODY) ); - - assert_eq!(*stages.0.lock().unwrap(), ["read_audit"]); Ok(()) }