test(TC-1317): read-audit review follow-ups (F1-F4)

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.

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,78 @@ 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);
+    }
+
+    /// 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;

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));
     }
 }
 

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::<UcanAbility>()?,
-            [std::collections::BTreeMap::<String, serde_json::Value>::new()],
+        let stranger_invocation = |audience: &str| -> Result<String> {
+            let mut capabilities = Capabilities::new();
+            let (uri, action) = kv_get()?;
+            capabilities.with_action(
+                uri.parse()?,
+                action.parse::<UcanAbility>()?,
+                [std::collections::BTreeMap::<String, serde_json::Value>::new()],
+            );
+            Ok(Payload {
+                issuer: stranger_vm.parse::<DIDURLBuf>()?,
+                audience: audience.parse::<DIDBuf>()?,
+                not_before: None,
+                expiration: test_invocation_expiration(),
+                nonce: Some(fixture.next_nonce()),
+                facts: Some(Vec::<serde_json::Value>::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::<DIDURLBuf>()?,
-            audience: fixture.space.did().to_string().parse::<DIDBuf>()?,
-            not_before: None,
-            expiration: test_invocation_expiration(),
-            nonce: Some(fixture.next_nonce()),
-            facts: Some(Vec::<serde_json::Value>::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,20 @@ 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. 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::Status;
+        use rocket::http::{Header, Status};
 
         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 +9832,22 @@ 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)
+        );
         Ok(())
     }