diff --git a/CHANGELOG.md b/CHANGELOG.md index b64db50f..e0dc02b8 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`) 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.25.3] - 2026-10-26 - A share invitation can read its link record when the links host resolves to the same addresses as another configured service. Details: TC-1737. diff --git a/tinycloud-core/src/read_audit.rs b/tinycloud-core/src/read_audit.rs index 12d3c08f..7704a312 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,292 @@ 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. 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() + .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) 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; + 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); + + 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 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. + 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); + // 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. + assert_eq!( + bounded(&pipeline, 1).await, + Err(ReadAuditError::Unavailable) + ); + assert_eq!(pipeline.health().timed_out_batches, 2); + until(|| pool.size() == 0).await; + 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); 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