From 6a6a9d9e006d40e3e2ade96f9c1ef68eb1294d63 Mon Sep 17 00:00:00 2001 From: dryrun Date: Sun, 25 Oct 2026 21:19:13 +0000 Subject: [PATCH 1/2] Read audit: bound the batch commit and discard a stalled connection A batch commit now has a 10 s bound (below the 20 s read wait). When it does not finish, the batch is failed as the existing retryable unavailable_retryable (503) class, the drain continues with the next batch, and the connection the commit held is closed and its pool slot freed instead of being released to the pool (sqlx pings a released connection, and the ping waits for the abandoned statement's reply). RecoveringConnection::execute_idempotent, whose only caller is the read-audit batch, acquires the PostgreSQL pool connection itself and detaches it if its future is dropped before the result arrives. Retry on a lost connection is unchanged. The timeout is logged as a warning with error_class read_audit_batch_timeout (batch size, elapsed and timeout in ms) and counted in tinycloud_read_audit_batch_timeout_total. Details: TC-2035 --- CHANGELOG.md | 3 + Cargo.lock | 1 + tinycloud-core/Cargo.toml | 6 +- tinycloud-core/src/db_pool.rs | 75 ++++- tinycloud-core/src/read_audit.rs | 364 +++++++++++++++++++++++- tinycloud-node-server/src/prometheus.rs | 10 + 6 files changed, 451 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 484063f4..2985105c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## [Unreleased] +### Changes +- Read audit: a batch commit that does not complete within 10 seconds is failed as a retryable 503 (`unavailable_retryable`) and its database connection is closed and its pool slot freed, instead of being returned to the pool, so the audit queue keeps draining; records whose readers have gone are skipped on the next batch as before. No read's data is returned without a committed record. The timeout is logged as `Read audit batch timed out` (`error_class = "read_audit_batch_timeout"`: batch size, elapsed and timeout in ms) and counted in `tinycloud_read_audit_batch_timeout_total`; log-shipper classification of the new warning follows. Details: TC-2035. + ## [1.23.0] - 2026-10-25 ### Release notes diff --git a/Cargo.lock b/Cargo.lock index 7120cc31..ef2ad9b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9909,6 +9909,7 @@ dependencies = [ "ryu-js", "sea-orm", "sea-orm-migration", + "sea-query-binder", "serde", "serde_ipld_dagcbor 0.3.0", "serde_json", diff --git a/tinycloud-core/Cargo.toml b/tinycloud-core/Cargo.toml index 0cda3535..4f72183a 100644 --- a/tinycloud-core/Cargo.toml +++ b/tinycloud-core/Cargo.toml @@ -8,7 +8,7 @@ edition = "2021" default = ["sqlite", "postgres", "mysql", "tokio"] sqlite = ["sea-orm/sqlx-sqlite", "sea-orm-migration/sqlx-sqlite"] -postgres = ["sea-orm/sqlx-postgres", "sea-orm-migration/sqlx-postgres"] +postgres = ["sea-orm/sqlx-postgres", "sea-orm-migration/sqlx-postgres", "dep:sea-query-binder"] mysql = ["sea-orm/sqlx-mysql", "sea-orm-migration/sqlx-mysql"] duckdb = ["dep:arrow", "dep:duckdb"] # TC-409: opt-in, deterministic verification-count test hook @@ -35,6 +35,10 @@ dashmap = "5.5" async-trait.workspace = true sea-orm = { version = "1.1", default-features = false, features = ["macros", "with-time", "with-json", "postgres-array"] } sea-orm-migration = { version = "1.1", default-features = false } +# TC-2035: binds a `Statement`'s values to a sqlx query on a connection the +# node holds itself (`RecoveringConnection::execute_idempotent`), the same +# binder sea-orm uses. +sea-query-binder = { version = "0.7", default-features = false, features = ["sqlx-postgres"], optional = true } futures.workspace = true pin-project.workspace = true time.workspace = true diff --git a/tinycloud-core/src/db_pool.rs b/tinycloud-core/src/db_pool.rs index cf1dab56..f1f6d3ce 100644 --- a/tinycloud-core/src/db_pool.rs +++ b/tinycloud-core/src/db_pool.rs @@ -238,10 +238,16 @@ impl RecoveringConnection { /// statement whose effect or result depends on whether it already ran: /// the replay guard's `INSERT ... ON CONFLICT DO NOTHING` reports a /// conflict as a replay, so it must not be retried. + /// + /// If the returned future is dropped before the statement's result + /// arrives (a caller's timeout), its connection is closed and its pool + /// slot freed instead of being returned to the pool: a released + /// connection is pinged, and the ping waits for the abandoned + /// statement's reply, which never comes on a stalled connection (TC-2035). pub async fn execute_idempotent(&self, stmt: Statement) -> Result { let started = Instant::now(); let in_flight = InFlightWrite::arm(true); - let result = match self.conn.execute(stmt.clone()).await { + let result = match self.execute_discarding_on_cancel(stmt.clone()).await { Err(error) if is_connection_error(&error) && !self.within_retry_window(started) => { self.lost("execute_idempotent", &error, false); Err(error) @@ -251,7 +257,10 @@ impl RecoveringConnection { if let Some(recovery) = &self.recovery { recovery.idempotent_retries.fetch_add(1, Ordering::Relaxed); } - self.retried("execute_idempotent", self.conn.execute(stmt).await) + self.retried( + "execute_idempotent", + self.execute_discarding_on_cancel(stmt).await, + ) } result => result, }; @@ -260,6 +269,24 @@ impl RecoveringConnection { result } + /// [`ConnectionTrait::execute`], except that on PostgreSQL a future + /// dropped before its result arrives closes the connection it was using. + /// + /// sqlx returns a dropped statement's connection to the pool, which pings + /// it first; the ping waits for the statement's reply, so a connection + /// that never answers is held (with its pool slot) with no bound. + async fn execute_discarding_on_cancel(&self, stmt: Statement) -> Result { + #[cfg(feature = "postgres")] + if matches!(self.conn, DatabaseConnection::SqlxPostgresPoolConnection(_)) { + return execute_postgres_discarding_on_cancel( + self.conn.get_postgres_connection_pool(), + stmt, + ) + .await; + } + self.conn.execute(stmt).await + } + /// The outcome of the one retry. It is never retried again, but a /// connection lost again is still recorded. fn retried(&self, operation: &'static str, result: Result) -> Result { @@ -808,6 +835,50 @@ impl Future for KeepConnection { } } +/// Run `stmt` the way sea-orm's PostgreSQL `execute` does (same acquire, +/// bind and error mapping), but on a connection that is closed, not +/// released, if this future is dropped before the statement's result +/// arrives (TC-2035). +/// +/// A released connection is pinged by sqlx before it re-enters the pool, and +/// the ping first reads the abandoned statement's reply. On a connection +/// that never answers, that task, the connection and its pool slot are held +/// for good. `detach` frees the slot at once and the dropped socket closes. +#[cfg(feature = "postgres")] +async fn execute_postgres_discarding_on_cancel( + pool: &sea_orm::sqlx::PgPool, + stmt: Statement, +) -> Result { + use sea_orm::{error::ConnAcquireErr, sqlx::pool::PoolConnection, sqlx::Postgres}; + + struct DiscardOnDrop(Option>); + impl Drop for DiscardOnDrop { + fn drop(&mut self) { + if let Some(connection) = self.0.take() { + drop(connection.detach()); + } + } + } + + let values = stmt + .values + .unwrap_or_else(|| sea_orm::sea_query::Values(Vec::new())); + let query = sea_orm::sqlx::query_with(&stmt.sql, sea_query_binder::SqlxValues(values)); + let connection = pool.acquire().await.map_err(|error| match error { + SqlxError::PoolTimedOut => DbErr::ConnectionAcquire(ConnAcquireErr::Timeout), + SqlxError::PoolClosed => DbErr::ConnectionAcquire(ConnAcquireErr::ConnectionClosed), + error => DbErr::Conn(RuntimeErr::SqlxError(error)), + })?; + let mut held = DiscardOnDrop(Some(connection)); + let connection = held.0.as_mut().expect("connection held until released"); + let result = query.execute(&mut **connection).await; + // The result arrived: release the connection to the pool as usual. + drop(held.0.take()); + result + .map(ExecResult::from) + .map_err(|error| DbErr::Exec(RuntimeErr::SqlxError(error))) +} + /// Whether `error` means the connection it ran on is gone, so the statement /// may never have reached the server or its result was lost. /// diff --git a/tinycloud-core/src/read_audit.rs b/tinycloud-core/src/read_audit.rs index 12d3c08f..98ec6504 100644 --- a/tinycloud-core/src/read_audit.rs +++ b/tinycloud-core/src/read_audit.rs @@ -116,10 +116,22 @@ struct PipelineStats { last_failure_class: std::sync::Mutex<&'static str>, /// Reads whose pipeline was gone when they queued or waited. stopped: AtomicU64, + /// Batches whose commit did not finish within `batch_timeout_ms` + /// (TC-2035); also counted in `failed_batches`. + timed_out_batches: AtomicU64, + /// The bound on one batch's commit, in ms. + batch_timeout_ms: AtomicU64, epoch: Instant, /// Test-only fault: this pipeline's next batch commit panics. #[cfg(test)] panic_next_batch: std::sync::atomic::AtomicBool, + /// Test-only fault: this pipeline's next batch commit never starts. + #[cfg(test)] + hang_next_batch: std::sync::atomic::AtomicBool, + /// Test-only fault: this pipeline's next batch commits, then its result + /// is never delivered (a lost response). + #[cfg(test)] + hang_after_persist: std::sync::atomic::AtomicBool, } impl Default for PipelineStats { @@ -136,9 +148,15 @@ impl Default for PipelineStats { last_queue_wait_ms: AtomicU64::new(0), last_failure_class: std::sync::Mutex::new("other"), stopped: AtomicU64::new(0), + timed_out_batches: AtomicU64::new(0), + batch_timeout_ms: AtomicU64::new(BATCH_COMMIT_TIMEOUT.as_millis() as u64), epoch: Instant::now(), #[cfg(test)] panic_next_batch: std::sync::atomic::AtomicBool::new(false), + #[cfg(test)] + hang_next_batch: std::sync::atomic::AtomicBool::new(false), + #[cfg(test)] + hang_after_persist: std::sync::atomic::AtomicBool::new(false), } } } @@ -183,6 +201,9 @@ pub struct ReadAuditHealth { pub last_queue_wait_ms: u64, /// Reads answered `Stopped` because the pipeline was gone. pub stopped: u64, + /// Batches abandoned because their commit did not finish in time; each is + /// also in `failed_batches`. + pub timed_out_batches: u64, /// The drain is gone: nothing will ever commit a record again. pub drain_stopped: bool, } @@ -190,6 +211,17 @@ pub struct ReadAuditHealth { /// 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); +/// The bound on one batch's commit, from its connection acquire to its result +/// (TC-2035). A commit that does not finish is failed as a retryable error and +/// its connection is closed, so the drain moves on to the next batch. +/// +/// It is below [`DEFAULT_READ_AUDIT_WAIT`], so a read that queued behind the +/// stuck batch is still waiting when the next batch forms, and below +/// [`STALLED_BATCH`], so the stalled-drain warning fires only if this bound +/// itself fails to end a batch (for example a SQLite writer-lock wait, which +/// it does not cover). It is above [`SLOW_BATCH`] and well above a healthy +/// commit, which is one statement. +const BATCH_COMMIT_TIMEOUT: Duration = Duration::from_secs(10); /// 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. const STALLED_BATCH: Duration = Duration::from_secs(30); @@ -244,6 +276,14 @@ impl ReadAuditPipeline { .store(wait.as_millis() as u64, Ordering::Relaxed); } + /// Bound one batch's commit. Tests only: they cannot wait ten seconds. + #[cfg(test)] + pub(crate) fn set_batch_timeout(&self, timeout: Duration) { + self.stats + .batch_timeout_ms + .store(timeout.as_millis() as u64, Ordering::Relaxed); + } + /// Hold every batch before it is committed until the guard drops. #[cfg(test)] pub(crate) async fn hold_drain(&self) -> tokio::sync::OwnedMutexGuard<()> { @@ -319,6 +359,7 @@ fn health(stats: &PipelineStats, sender: &mpsc::Sender) -> ReadAuditHea in_flight_batch_age_ms: stats.in_flight_age().as_millis() as u64, last_queue_wait_ms: stats.last_queue_wait_ms.load(Ordering::Relaxed), stopped: stats.stopped.load(Ordering::Relaxed), + timed_out_batches: stats.timed_out_batches.load(Ordering::Relaxed), drain_stopped: sender.is_closed(), } } @@ -508,13 +549,29 @@ async fn commit_batch( if stats.panic_next_batch.swap(false, Ordering::SeqCst) { panic!("injected read-audit batch panic"); } - persist_batch(conn, &batch).await + #[cfg(test)] + if stats.hang_next_batch.swap(false, Ordering::SeqCst) { + std::future::pending::<()>().await; + } + let persisted = persist_batch(conn, &batch).await; + #[cfg(test)] + if stats.hang_after_persist.swap(false, Ordering::SeqCst) { + std::future::pending::<()>().await; + } + persisted }, )); + // A commit that does not finish in time fails this batch's waiters as + // retryable, like a transient database error, and drops the commit + // future; dropping it closes the connection it holds (TC-2035). + let limit = Duration::from_millis(stats.batch_timeout_ms.load(Ordering::Relaxed)); + let began = Instant::now(); // A panic fails this batch's waiters; the drain goes on (TC-1317). - let result = match futures::FutureExt::catch_unwind(persisted).await { - Ok(Ok(())) => Ok(()), - Ok(Err(error)) => { + let result = match tokio::time::timeout(limit, futures::FutureExt::catch_unwind(persisted)) + .await + { + Ok(Ok(Ok(()))) => Ok(()), + Ok(Ok(Err(error))) => { let class = transient_error_class(&error).unwrap_or("other"); *stats .last_failure_class @@ -522,13 +579,29 @@ async fn commit_batch( .unwrap_or_else(std::sync::PoisonError::into_inner) = class; Err(batch_error(error)) } - Err(_) => { + Ok(Err(_)) => { *stats .last_failure_class .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) = "panic"; Err(ReadAuditError::Failed("read audit batch panicked".into())) } + Err(_) => { + *stats + .last_failure_class + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = "read_audit_batch_timeout"; + stats.timed_out_batches.fetch_add(1, Ordering::Relaxed); + // Counts and durations only: never a key or an identifier. + tracing::warn!( + error_class = "read_audit_batch_timeout", + batch_size = batch.len(), + elapsed_ms = began.elapsed().as_millis() as u64, + timeout_ms = limit.as_millis() as u64, + "Read audit batch timed out" + ); + Err(ReadAuditError::Unavailable) + } }; let snapshot = metrics.snapshot(); // At most 64 commands: scanning prior entries avoids a recipient allocation @@ -2423,4 +2496,285 @@ mod tests { .unwrap(); assert!(committed(&conn, 1).await); } + + // TC-2035: a commit that never returns must not stop the drain. + + /// A batch whose commit never returns is failed as retryable once the + /// bound passes, never confirmed, and the next read commits. + #[tokio::test] + async fn tc2035_a_commit_that_never_returns_fails_retryably_and_the_drain_moves_on() { + let (_directory, conn, pipeline) = file_pipeline().await; + pipeline.set_batch_timeout(Duration::from_millis(200)); + pipeline.stats.hang_next_batch.store(true, Ordering::SeqCst); + + let started = Instant::now(); + let outcome = bounded(&pipeline, 0).await; + assert_eq!(outcome, Err(ReadAuditError::Unavailable)); + assert!( + started.elapsed() < Duration::from_secs(5), + "answered at the commit bound, not the read wait" + ); + assert!(!committed(&conn, 0).await); + let health = pipeline.health(); + assert_eq!((health.timed_out_batches, health.failed_batches), (1, 1)); + assert_eq!(health.in_flight_batch_age_ms, 0); + + bounded(&pipeline, 1) + .await + .expect("a later read in the same pipeline commits"); + assert!(committed(&conn, 1).await); + assert_eq!(pipeline.health().timed_out_batches, 1); + } + + /// Records queued behind a batch that hangs are skipped once their + /// waiters are gone, when the next batch forms after the timeout. + #[tokio::test] + async fn tc2035_abandoned_records_behind_a_hung_batch_are_skipped_after_the_timeout() { + let (_directory, conn, pipeline) = file_pipeline().await; + pipeline.set_wait(Duration::from_millis(100)); + pipeline.set_batch_timeout(Duration::from_secs(2)); + pipeline.stats.hang_next_batch.store(true, Ordering::SeqCst); + + // Record 0 forms the batch that hangs; its waiter gives up. + assert_eq!(bounded(&pipeline, 0).await, Err(ReadAuditError::Busy)); + until(|| pipeline.health().queue_depth == 0).await; + // Records 1-5 queue behind it and their waiters give up too. + for index in 1..=5 { + assert_eq!(bounded(&pipeline, index).await, Err(ReadAuditError::Busy)); + } + assert_eq!(pipeline.health().queue_depth, 5); + assert_eq!(pipeline.health().abandoned, 0, "the drain is stuck"); + + until(|| pipeline.health().abandoned == 5).await; + assert_eq!(pipeline.health().timed_out_batches, 1); + for index in 0..=5 { + assert!(!committed(&conn, index).await, "record {index}"); + } + + pipeline.set_wait(DEFAULT_READ_AUDIT_WAIT); + bounded(&pipeline, 6).await.expect("the drain recovered"); + assert!(committed(&conn, 6).await); + } + + /// A PostgreSQL server that completes every connection's startup and then + /// never answers, and what it saw of its sockets. + struct BlackHole { + port: u16, + accepted: Arc, + closed: Arc, + } + + async fn black_hole() -> BlackHole { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let accepted = Arc::new(AtomicU64::new(0)); + let closed = Arc::new(AtomicU64::new(0)); + let (seen_accepted, seen_closed) = (accepted.clone(), closed.clone()); + tokio::spawn(async move { + while let Ok((mut stream, _)) = listener.accept().await { + seen_accepted.fetch_add(1, Ordering::SeqCst); + let closed = seen_closed.clone(); + tokio::spawn(async move { + let mut length = [0u8; 4]; + if stream.read_exact(&mut length).await.is_ok() { + // The startup message, then AuthenticationOk and + // ReadyForQuery. Nothing after that is ever answered. + let mut startup = vec![0u8; (u32::from_be_bytes(length) as usize) - 4]; + let _ = stream.read_exact(&mut startup).await; + let _ = stream + .write_all(&[b'R', 0, 0, 0, 8, 0, 0, 0, 0, b'Z', 0, 0, 0, 5, b'I']) + .await; + let mut sink = [0u8; 4096]; + while matches!(stream.read(&mut sink).await, Ok(read) if read > 0) {} + } + closed.fetch_add(1, Ordering::SeqCst); + }); + } + }); + BlackHole { + port, + accepted, + closed, + } + } + + /// A one-connection pool (so a leaked slot is visible) to `server`, with + /// that connection already open. + async fn black_hole_pool(server: &BlackHole) -> sea_orm::sqlx::PgPool { + use sea_orm::sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgSslMode}; + + PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(2)) + .test_before_acquire(false) + .connect_with( + PgConnectOptions::new() + .host("127.0.0.1") + .port(server.port) + .username("audit") + .database("audit") + .ssl_mode(PgSslMode::Disable), + ) + .await + .expect("the black hole completes the startup") + } + + /// A batch whose connection stalls (the server never answers) times out, + /// and the stalled connection is closed and its pool slot freed rather + /// than returned: the pool, limited to one connection, serves the next + /// acquire on a new socket, and the drain keeps going. + #[tokio::test] + async fn tc2035_a_stalled_connection_is_closed_and_its_pool_slot_freed() { + let server = black_hole().await; + let pool = black_hole_pool(&server).await; + let conn = sea_orm::SqlxPostgresConnector::from_sqlx_postgres_pool(pool.clone()); + let pipeline = ReadAuditPipeline::start(RecoveringConnection::new(conn), None); + pipeline.set_batch_timeout(Duration::from_millis(300)); + assert_eq!((pool.size(), pool.num_idle()), (1, 1)); + assert_eq!(server.accepted.load(Ordering::SeqCst), 1); + + assert_eq!( + bounded(&pipeline, 0).await, + Err(ReadAuditError::Unavailable) + ); + let health = pipeline.health(); + assert_eq!((health.timed_out_batches, health.failed_batches), (1, 1)); + // The slot is free again, not held by a connection waiting for a + // reply that never comes. + assert_eq!((pool.size(), pool.num_idle()), (0, 0)); + until(|| server.closed.load(Ordering::SeqCst) == 1).await; + + // A following acquire against the (otherwise healthy) server opens a + // new connection within the one-connection limit. + let fresh = tokio::time::timeout(PATIENCE, pool.acquire()) + .await + .expect("acquire must not hang") + .expect("the freed slot serves a new connection"); + assert_eq!(pool.size(), 1); + assert_eq!(server.accepted.load(Ordering::SeqCst), 2); + // Not released: the pool's release ping would wait on this server + // for ever, which is why the batch's connection is discarded. + drop(fresh.detach()); + assert_eq!(pool.size(), 0); + until(|| server.closed.load(Ordering::SeqCst) == 2).await; + + // The drain went on: the next batch opens a connection, stalls, + // times out, and closes it as well. + assert_eq!( + bounded(&pipeline, 1).await, + Err(ReadAuditError::Unavailable) + ); + assert_eq!(pipeline.health().timed_out_batches, 2); + assert_eq!(pool.size(), 0); + until(|| server.closed.load(Ordering::SeqCst) == 3).await; + assert_eq!(server.accepted.load(Ordering::SeqCst), 3); + } + + /// A batch that did commit but whose result never arrived is failed like + /// any timed-out batch (its read got no data), and the client's retry of + /// the same record leaves exactly the rows a single commit would: no + /// duplicates, no gaps. The same batch persisted twice is also a no-op. + async fn assert_timed_out_commit_is_idempotent(conn: DatabaseConnection) { + let parent = crate::hash::hash(b"timeout-parent"); + insert_parent_delegation(&conn, parent).await; + let get = ability("tinycloud.kv/get"); + let first = rich_record( + "timeout-1", + "did:key:timeout", + vec![( + resource("tinycloud:did:key:ne-delegator:files/kv/a"), + get.clone(), + )], + vec![parent], + ); + let second = rich_record( + "timeout-2", + "did:key:timeout-other", + vec![(resource("tinycloud:did:key:ne-delegator:files/kv/b"), get)], + vec![parent], + ); + // `enqueue` consumes a record; rebuild the same one by content. + let again = |record: &ReadAuditRecord| ReadAuditRecord { + id: record.id, + invoker: record.invoker.clone(), + issued_at: OffsetDateTime::now_utc(), + serialization: record.serialization.clone(), + abilities: record.abilities.clone(), + parents: record.parents.clone(), + }; + + let pipeline = ReadAuditPipeline::start(conn.clone(), None); + pipeline.set_batch_timeout(Duration::from_secs(3)); + pipeline + .stats + .hang_after_persist + .store(true, Ordering::SeqCst); + let outcome = tokio::time::timeout(PATIENCE, pipeline.enqueue(again(&first))) + .await + .expect("the timed-out batch's waiter must be answered"); + assert_eq!(outcome, Err(ReadAuditError::Unavailable)); + // The statement had committed: over-audit, never a lost record. + assert_eq!( + audit_row_sets(&conn).await, + expected_rows(std::slice::from_ref(&first)) + ); + + tokio::time::timeout(PATIENCE, pipeline.enqueue(again(&second))) + .await + .expect("the drain moved on") + .unwrap(); + // The client's retry of the timed-out read, and a repeat of the + // other one. + for record in [&first, &second] { + tokio::time::timeout(PATIENCE, pipeline.enqueue(again(record))) + .await + .expect("the drain answers") + .unwrap(); + } + let expected = expected_rows(&[again(&first), again(&second)]); + assert_eq!(audit_row_sets(&conn).await, expected); + // Raw counts, in case the sets hide duplicates the schema allowed. + assert_eq!(invocation::Entity::find().count(&conn).await.unwrap(), 2); + assert_eq!( + invoked_abilities::Entity::find() + .count(&conn) + .await + .unwrap(), + 2 + ); + assert_eq!( + parent_delegations::Entity::find() + .count(&conn) + .await + .unwrap(), + 2 + ); + + // The same batch persisted twice directly: still a no-op. + let batch = vec![command(again(&first)), command(again(&second))]; + let recovering = RecoveringConnection::new(conn.clone()); + persist_batch(&recovering, &batch).await.unwrap(); + persist_batch(&recovering, &batch).await.unwrap(); + assert_eq!(audit_row_sets(&conn).await, expected); + assert_eq!(invocation::Entity::find().count(&conn).await.unwrap(), 2); + } + + #[tokio::test] + async fn tc2035_a_timed_out_commit_is_idempotent_on_retry() { + let (_directory, conn, _pipeline) = file_pipeline().await; + assert_timed_out_commit_is_idempotent(conn).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn postgres_read_audit_timed_out_commit_is_idempotent_on_retry() { + use crate::test_support::PostgresTestSchema; + + const TEST: &str = "postgres_read_audit_timed_out_commit_is_idempotent_on_retry"; + let Some(schema) = PostgresTestSchema::new(TEST).await else { + return; + }; + assert_timed_out_commit_is_idempotent(schema.conn.clone()).await; + } } diff --git a/tinycloud-node-server/src/prometheus.rs b/tinycloud-node-server/src/prometheus.rs index d6719684..a42eb01c 100644 --- a/tinycloud-node-server/src/prometheus.rs +++ b/tinycloud-node-server/src/prometheus.rs @@ -149,6 +149,12 @@ lazy_static! { "Reads answered read_audit_busy because their audit record was not committed in time." ) .unwrap(); + /// Read-audit batches abandoned at the commit bound (TC-2035). + pub static ref READ_AUDIT_BATCH_TIMEOUTS: IntCounter = register_int_counter!( + "tinycloud_read_audit_batch_timeout_total", + "Read-audit commit batches abandoned because their commit did not finish within its bound." + ) + .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", @@ -277,6 +283,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_BATCH_TIMEOUTS.inc_by( + now.timed_out_batches + .saturating_sub(previous.timed_out_batches), + ); 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); From 468a6c06558958307df387059ac2da34c0905e54 Mon Sep 17 00:00:00 2001 From: dryrun Date: Mon, 26 Oct 2026 09:45:35 +0000 Subject: [PATCH 2/2] Read audit: bound the pool's release ping instead of acquiring explicitly Revert RecoveringConnection::execute_idempotent to its original form (the explicit acquire bypassed sea-orm's metric callback). The held pool slot is fixed at its root in the vendored sqlx-core: the release ping is bounded by a named timeout (5 s, per-pool option) and the connection is closed hard, releasing its permit, when it does not answer. Details: TC-2035 --- CHANGELOG.md | 2 +- Cargo.lock | 4 +- tinycloud-core/Cargo.toml | 8 +-- tinycloud-core/src/db_pool.rs | 87 +++++-------------------- tinycloud-core/src/read_audit.rs | 37 ++++++----- vendor/sqlx-core/PATCH.md | 41 ++++++++++++ vendor/sqlx-core/src/pool/connection.rs | 27 +++++++- vendor/sqlx-core/src/pool/options.rs | 12 ++++ 8 files changed, 121 insertions(+), 97 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2985105c..f899ab8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ ## [Unreleased] ### Changes -- Read audit: a batch commit that does not complete within 10 seconds is failed as a retryable 503 (`unavailable_retryable`) and its database connection is closed and its pool slot freed, instead of being returned to the pool, so the audit queue keeps draining; records whose readers have gone are skipped on the next batch as before. No read's data is returned without a committed record. The timeout is logged as `Read audit batch timed out` (`error_class = "read_audit_batch_timeout"`: batch size, elapsed and timeout in ms) and counted in `tinycloud_read_audit_batch_timeout_total`; log-shipper classification of the new warning follows. Details: TC-2035. +- Read audit: a batch commit that does not complete within 10 seconds is failed as a retryable 503 (`unavailable_retryable`) so the audit queue keeps draining; records whose readers have gone are skipped on the next batch as before. No read's data is returned without a committed record. The timeout is logged as `Read audit batch timed out` (`error_class = "read_audit_batch_timeout"`: batch size, elapsed and timeout in ms) and counted in `tinycloud_read_audit_batch_timeout_total`; log-shipper classification of the new warning follows. Database pool: a released connection whose health ping does not answer within 5 seconds is now closed and its slot freed, instead of being held with no bound (vendored `sqlx-core`, `vendor/sqlx-core/PATCH.md`). Details: TC-2035. ## [1.23.0] - 2026-10-25 diff --git a/Cargo.lock b/Cargo.lock index ef2ad9b0..af898e06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9896,6 +9896,7 @@ dependencies = [ "hex", "hkdf", "hmac 0.12.1", + "idna", "iri-string", "k256", "libc", @@ -9909,7 +9910,6 @@ dependencies = [ "ryu-js", "sea-orm", "sea-orm-migration", - "sea-query-binder", "serde", "serde_ipld_dagcbor 0.3.0", "serde_json", @@ -9933,7 +9933,7 @@ dependencies = [ [[package]] name = "tinycloud-node" -version = "1.23.0" +version = "1.25.2" dependencies = [ "aes-gcm", "anyhow", diff --git a/tinycloud-core/Cargo.toml b/tinycloud-core/Cargo.toml index 4f72183a..6425231d 100644 --- a/tinycloud-core/Cargo.toml +++ b/tinycloud-core/Cargo.toml @@ -8,7 +8,7 @@ edition = "2021" default = ["sqlite", "postgres", "mysql", "tokio"] sqlite = ["sea-orm/sqlx-sqlite", "sea-orm-migration/sqlx-sqlite"] -postgres = ["sea-orm/sqlx-postgres", "sea-orm-migration/sqlx-postgres", "dep:sea-query-binder"] +postgres = ["sea-orm/sqlx-postgres", "sea-orm-migration/sqlx-postgres"] mysql = ["sea-orm/sqlx-mysql", "sea-orm-migration/sqlx-mysql"] duckdb = ["dep:arrow", "dep:duckdb"] # TC-409: opt-in, deterministic verification-count test hook @@ -35,10 +35,6 @@ dashmap = "5.5" async-trait.workspace = true sea-orm = { version = "1.1", default-features = false, features = ["macros", "with-time", "with-json", "postgres-array"] } sea-orm-migration = { version = "1.1", default-features = false } -# TC-2035: binds a `Statement`'s values to a sqlx query on a connection the -# node holds itself (`RecoveringConnection::execute_idempotent`), the same -# binder sea-orm uses. -sea-query-binder = { version = "0.7", default-features = false, features = ["sqlx-postgres"], optional = true } futures.workspace = true pin-project.workspace = true time.workspace = true @@ -85,6 +81,8 @@ ryu-js = "0.2.2" # canonicalization. Replaces the hand-coded subset that only composed a # narrow Latin precomposed-pair table. unicode-normalization = "0.1" +# App ids are bound to the ASCII (punycode) form of a host name. +idna = "1" [dev-dependencies] sea-orm = { version = "1.1", features = ["runtime-tokio-rustls", "sqlx-sqlite"] } diff --git a/tinycloud-core/src/db_pool.rs b/tinycloud-core/src/db_pool.rs index f1f6d3ce..0201f074 100644 --- a/tinycloud-core/src/db_pool.rs +++ b/tinycloud-core/src/db_pool.rs @@ -238,16 +238,10 @@ impl RecoveringConnection { /// statement whose effect or result depends on whether it already ran: /// the replay guard's `INSERT ... ON CONFLICT DO NOTHING` reports a /// conflict as a replay, so it must not be retried. - /// - /// If the returned future is dropped before the statement's result - /// arrives (a caller's timeout), its connection is closed and its pool - /// slot freed instead of being returned to the pool: a released - /// connection is pinged, and the ping waits for the abandoned - /// statement's reply, which never comes on a stalled connection (TC-2035). pub async fn execute_idempotent(&self, stmt: Statement) -> Result { let started = Instant::now(); let in_flight = InFlightWrite::arm(true); - let result = match self.execute_discarding_on_cancel(stmt.clone()).await { + let result = match self.conn.execute(stmt.clone()).await { Err(error) if is_connection_error(&error) && !self.within_retry_window(started) => { self.lost("execute_idempotent", &error, false); Err(error) @@ -257,10 +251,7 @@ impl RecoveringConnection { if let Some(recovery) = &self.recovery { recovery.idempotent_retries.fetch_add(1, Ordering::Relaxed); } - self.retried( - "execute_idempotent", - self.execute_discarding_on_cancel(stmt).await, - ) + self.retried("execute_idempotent", self.conn.execute(stmt).await) } result => result, }; @@ -269,22 +260,20 @@ impl RecoveringConnection { result } - /// [`ConnectionTrait::execute`], except that on PostgreSQL a future - /// dropped before its result arrives closes the connection it was using. - /// - /// sqlx returns a dropped statement's connection to the pool, which pings - /// it first; the ping waits for the statement's reply, so a connection - /// that never answers is held (with its pool slot) with no bound. - async fn execute_discarding_on_cancel(&self, stmt: Statement) -> Result { - #[cfg(feature = "postgres")] - if matches!(self.conn, DatabaseConnection::SqlxPostgresPoolConnection(_)) { - return execute_postgres_discarding_on_cancel( - self.conn.get_postgres_connection_pool(), - stmt, - ) - .await; - } - self.conn.execute(stmt).await + /// Pre-upload authorization and orphan registration share one connection. + /// Reserve SQLite's database writer before authorization reads, just like + /// the other write paths: a separate collector connection must not commit + /// after our snapshot and turn registration into SQLITE_BUSY_SNAPSHOT. + /// This transaction changes only retryable candidate metadata and commits + /// before blob persistence, so its uncertain outcome cannot mean a + /// user-visible KV write committed. + pub(crate) async fn begin_preparatory(&self) -> Result { + let transaction = self + .run("begin_preparatory", true, || { + crate::db::begin_write_transaction(&self.conn, None) + }) + .await?; + self.scope_local_timeout(transaction).await } /// The outcome of the one retry. It is never retried again, but a @@ -835,50 +824,6 @@ impl Future for KeepConnection { } } -/// Run `stmt` the way sea-orm's PostgreSQL `execute` does (same acquire, -/// bind and error mapping), but on a connection that is closed, not -/// released, if this future is dropped before the statement's result -/// arrives (TC-2035). -/// -/// A released connection is pinged by sqlx before it re-enters the pool, and -/// the ping first reads the abandoned statement's reply. On a connection -/// that never answers, that task, the connection and its pool slot are held -/// for good. `detach` frees the slot at once and the dropped socket closes. -#[cfg(feature = "postgres")] -async fn execute_postgres_discarding_on_cancel( - pool: &sea_orm::sqlx::PgPool, - stmt: Statement, -) -> Result { - use sea_orm::{error::ConnAcquireErr, sqlx::pool::PoolConnection, sqlx::Postgres}; - - struct DiscardOnDrop(Option>); - impl Drop for DiscardOnDrop { - fn drop(&mut self) { - if let Some(connection) = self.0.take() { - drop(connection.detach()); - } - } - } - - let values = stmt - .values - .unwrap_or_else(|| sea_orm::sea_query::Values(Vec::new())); - let query = sea_orm::sqlx::query_with(&stmt.sql, sea_query_binder::SqlxValues(values)); - let connection = pool.acquire().await.map_err(|error| match error { - SqlxError::PoolTimedOut => DbErr::ConnectionAcquire(ConnAcquireErr::Timeout), - SqlxError::PoolClosed => DbErr::ConnectionAcquire(ConnAcquireErr::ConnectionClosed), - error => DbErr::Conn(RuntimeErr::SqlxError(error)), - })?; - let mut held = DiscardOnDrop(Some(connection)); - let connection = held.0.as_mut().expect("connection held until released"); - let result = query.execute(&mut **connection).await; - // The result arrived: release the connection to the pool as usual. - drop(held.0.take()); - result - .map(ExecResult::from) - .map_err(|error| DbErr::Exec(RuntimeErr::SqlxError(error))) -} - /// Whether `error` means the connection it ran on is gone, so the statement /// may never have reached the server or its result was lost. /// diff --git a/tinycloud-core/src/read_audit.rs b/tinycloud-core/src/read_audit.rs index 98ec6504..7704a312 100644 --- a/tinycloud-core/src/read_audit.rs +++ b/tinycloud-core/src/read_audit.rs @@ -2601,13 +2601,15 @@ mod tests { } /// A one-connection pool (so a leaked slot is visible) to `server`, with - /// that connection already open. + /// that connection already open. The release-ping bound is shortened from + /// its 5 s default so the test is fast; it is a per-pool option. async fn black_hole_pool(server: &BlackHole) -> sea_orm::sqlx::PgPool { use sea_orm::sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgSslMode}; PgPoolOptions::new() .max_connections(1) .acquire_timeout(Duration::from_secs(2)) + .release_ping_timeout(Duration::from_millis(300)) .test_before_acquire(false) .connect_with( PgConnectOptions::new() @@ -2621,10 +2623,12 @@ mod tests { .expect("the black hole completes the startup") } - /// A batch whose connection stalls (the server never answers) times out, - /// and the stalled connection is closed and its pool slot freed rather - /// than returned: the pool, limited to one connection, serves the next - /// acquire on a new socket, and the drain keeps going. + /// A batch whose connection stalls (the server never answers) fails + /// within the commit bound as retryable. The stalled connection's release + /// ping is then bounded too: its socket is closed and its pool permit + /// freed within the commit bound plus the ping bound, instead of being + /// held for ever, so the one-connection pool serves the next acquire on a + /// new socket, and the drain keeps going. #[tokio::test] async fn tc2035_a_stalled_connection_is_closed_and_its_pool_slot_freed() { let server = black_hole().await; @@ -2635,16 +2639,20 @@ mod tests { assert_eq!((pool.size(), pool.num_idle()), (1, 1)); assert_eq!(server.accepted.load(Ordering::SeqCst), 1); + let started = Instant::now(); assert_eq!( bounded(&pipeline, 0).await, Err(ReadAuditError::Unavailable) ); + assert!( + started.elapsed() < Duration::from_secs(2), + "answered at the commit bound" + ); let health = pipeline.health(); assert_eq!((health.timed_out_batches, health.failed_batches), (1, 1)); - // The slot is free again, not held by a connection waiting for a - // reply that never comes. - assert_eq!((pool.size(), pool.num_idle()), (0, 0)); - until(|| server.closed.load(Ordering::SeqCst) == 1).await; + // The permit and socket recover once the release ping's bound passes. + until(|| pool.size() == 0 && server.closed.load(Ordering::SeqCst) == 1).await; + assert_eq!(pool.num_idle(), 0); // A following acquire against the (otherwise healthy) server opens a // new connection within the one-connection limit. @@ -2654,11 +2662,10 @@ mod tests { .expect("the freed slot serves a new connection"); assert_eq!(pool.size(), 1); assert_eq!(server.accepted.load(Ordering::SeqCst), 2); - // Not released: the pool's release ping would wait on this server - // for ever, which is why the batch's connection is discarded. - drop(fresh.detach()); - assert_eq!(pool.size(), 0); - until(|| server.closed.load(Ordering::SeqCst) == 2).await; + // A connection released with nothing in flight is pinged too, and this + // server never answers: the same bound closes it. + drop(fresh); + until(|| pool.size() == 0 && server.closed.load(Ordering::SeqCst) == 2).await; // The drain went on: the next batch opens a connection, stalls, // times out, and closes it as well. @@ -2667,7 +2674,7 @@ mod tests { Err(ReadAuditError::Unavailable) ); assert_eq!(pipeline.health().timed_out_batches, 2); - assert_eq!(pool.size(), 0); + until(|| pool.size() == 0).await; until(|| server.closed.load(Ordering::SeqCst) == 3).await; assert_eq!(server.accepted.load(Ordering::SeqCst), 3); } diff --git a/vendor/sqlx-core/PATCH.md b/vendor/sqlx-core/PATCH.md index e9eefd04..ab3d18be 100644 --- a/vendor/sqlx-core/PATCH.md +++ b/vendor/sqlx-core/PATCH.md @@ -79,3 +79,44 @@ test runtime, require it in CI, and otherwise print an explicit local skip. They use SELECT-only statements and test raw BEGIN/SAVEPOINT through SQLx and SeaORM, validation, success/failure, awaited commit/rollback, and acquisition timeout/closed-pool failure. MySQL has no integration-test server in this suite. + +## Bounded release ping (TC-2035) + +### Why + +sqlx returns a dropped connection to the pool through +`Floating::return_to_pool`, which pings it first. A connection whose statement +future was dropped mid-flight (a caller's timeout, a cancelled request) still +owes that statement's reply, and the ping waits for it. If the peer never +answers (a stalled proxy or socket), nothing bounds that wait: the spawned +release task, the socket and the pool permit are held for good. The read-audit +drain hit this: a batch commit that never returned stopped the drain, and +dropping its future from a timeout would have held the pool slot instead. + +### The delta + +- `src/pool/connection.rs`: `RELEASE_PING_TIMEOUT` (5 s). In + `Floating::return_to_pool`, the release ping runs under + `crate::rt::timeout(self.guard.pool.options.release_ping_timeout, ..)`. A + timeout is handled like a failed ping: `warn!` log, then `close_hard()`, itself + bounded by `CLOSE_ON_DROP_TIMEOUT` (on that timeout the future is dropped, which + closes the socket and releases the permit). The pool opens a replacement on + the next acquire; the permit is released on every path. +- `src/pool/options.rs`: `PoolOptions::release_ping_timeout(Duration)` and a + `release_ping_timeout` field (default `RELEASE_PING_TIMEOUT`, copied in `Clone` + and `new`). The node leaves the default; a test shortens it for one pool. + +### Effect + +Pool-wide. Any released connection whose ping does not answer within the bound is +closed instead of held. A live server finishing a long dropped statement is no +longer waited for; closing the socket makes the server abort it. + +### Re-applying on a vendor bump + +Re-apply the two hunks above to the new `pool/connection.rs` (the ping in +`return_to_pool`, and the new const) and `pool/options.rs` (the field, `Clone`, +`new`, the setter). The regression is +`read_audit::tests::tc2035_a_stalled_connection_is_closed_and_its_pool_slot_freed` +in `tinycloud-core`, which uses a listener that never answers after startup. Drop +this patch when upstream bounds the release ping. diff --git a/vendor/sqlx-core/src/pool/connection.rs b/vendor/sqlx-core/src/pool/connection.rs index c029fec6..b89b705c 100644 --- a/vendor/sqlx-core/src/pool/connection.rs +++ b/vendor/sqlx-core/src/pool/connection.rs @@ -15,6 +15,16 @@ use std::future::Future; const CLOSE_ON_DROP_TIMEOUT: Duration = Duration::from_secs(5); +/// How long the ping that tests a released connection may take (TinyCloud patch, TC-2035). +/// +/// A connection whose statement future was dropped mid-flight (a caller's timeout, a cancelled +/// request) is still owed that statement's reply, and the release ping waits for it. If the peer +/// never answers (a stalled proxy or socket), that wait has no bound of its own, so the ping, the +/// socket and the pool permit are held for good. Past this bound the connection is closed hard +/// and its permit released, so the pool opens a replacement. Overridable per pool with +/// `PoolOptions::release_ping_timeout`. +pub(super) const RELEASE_PING_TIMEOUT: Duration = Duration::from_secs(5); + /// A connection managed by a [`Pool`][crate::pool::Pool]. /// /// Will be returned to the pool on-drop. @@ -311,14 +321,25 @@ impl Floating> { // returned to the pool; also of course, if it was dropped due to an error // this is simply a band-aid as SQLx-next connections should be able // to recover from cancellations - if let Err(error) = self.raw.ping().await { + let ping_timeout = self.guard.pool.options.release_ping_timeout; + let pinged = match crate::rt::timeout(ping_timeout, self.raw.ping()).await { + Ok(pinged) => pinged, + Err(_) => Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "the release ping did not answer in time", + ))), + }; + if let Err(error) = pinged { tracing::warn!( %error, "error occurred while testing the connection on-release", ); - // Connection is broken, don't try to gracefully close. - self.close_hard().await; + // Connection is broken, don't try to gracefully close. This also covers a ping that + // did not answer within `release_ping_timeout`. The close is bounded too: on timeout + // the future, and with it the connection and the pool permit, is dropped, which + // closes the socket and releases the permit. + let _ = crate::rt::timeout(CLOSE_ON_DROP_TIMEOUT, self.close_hard()).await; false } else { // if the connection is still viable, release it to the pool diff --git a/vendor/sqlx-core/src/pool/options.rs b/vendor/sqlx-core/src/pool/options.rs index 3d048f17..7010335c 100644 --- a/vendor/sqlx-core/src/pool/options.rs +++ b/vendor/sqlx-core/src/pool/options.rs @@ -79,6 +79,7 @@ pub struct PoolOptions { pub(crate) acquire_slow_level: LevelFilter, pub(crate) acquire_slow_threshold: Duration, pub(crate) acquire_timeout: Duration, + pub(crate) release_ping_timeout: Duration, pub(crate) min_connections: u32, pub(crate) max_lifetime: Option, pub(crate) idle_timeout: Option, @@ -102,6 +103,7 @@ impl Clone for PoolOptions { acquire_slow_threshold: self.acquire_slow_threshold, acquire_slow_level: self.acquire_slow_level, acquire_timeout: self.acquire_timeout, + release_ping_timeout: self.release_ping_timeout, min_connections: self.min_connections, max_lifetime: self.max_lifetime, idle_timeout: self.idle_timeout, @@ -158,6 +160,7 @@ impl PoolOptions { // to not flag typical time to add a new connection to a pool. acquire_slow_threshold: Duration::from_secs(2), acquire_timeout: Duration::from_secs(30), + release_ping_timeout: super::connection::RELEASE_PING_TIMEOUT, idle_timeout: Some(Duration::from_secs(10 * 60)), max_lifetime: Some(Duration::from_secs(30 * 60)), fair: true, @@ -263,6 +266,15 @@ impl PoolOptions { self } + /// Set the longest a connection's release ping may take before the connection is closed + /// instead of returned to the pool (TinyCloud patch, TC-2035). + /// + /// The default is 5 seconds. See `vendor/sqlx-core/PATCH.md`. + pub fn release_ping_timeout(mut self, timeout: Duration) -> Self { + self.release_ping_timeout = timeout; + self + } + /// Get the maximum amount of time to spend waiting for a connection in [`Pool::acquire()`]. pub fn get_acquire_timeout(&self) -> Duration { self.acquire_timeout