From c77edd8abb781e4ef70f0c225e2f6cdeafc13e7d Mon Sep 17 00:00:00 2001 From: Mario Savarese Date: Mon, 20 Jul 2026 13:53:12 +0100 Subject: [PATCH 1/8] fix: only stop task when every clone of it has been dropped This fixes an issue where the shutdown flag gets permanently flipped to true (which signals that the polling task should be stopped) any time a cloned pool is dropped. Because all the pool clones share the same `Arc`, the first clone to be dropped incorrectly flips the flag, which is not what the documentation implies. This also adds a regression test to make sure the metric task surives a cloned pool being dropped --- src/pool_metrics.rs | 21 +++++++++++++-------- tests/pool_metrics.rs | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 8 deletions(-) diff --git a/src/pool_metrics.rs b/src/pool_metrics.rs index 82a9dc1..6b15a45 100644 --- a/src/pool_metrics.rs +++ b/src/pool_metrics.rs @@ -3,12 +3,13 @@ //! //! Spawned by [`PoolBuilder::build()`](crate::PoolBuilder::build) when a pool name is //! configured and a runtime feature (e.g. `runtime-tokio`) is enabled. The task stops -//! when the [`Pool`](crate::Pool) is dropped. +//! only once the [`Pool`](crate::Pool) and every clone of it have been dropped. use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -/// Handle that signals the background polling task to stop when dropped. +/// Handle that signals the background polling task to stop when the last clone is +/// dropped. #[derive(Clone)] pub(crate) struct ShutdownHandle { flag: Arc, @@ -16,7 +17,9 @@ pub(crate) struct ShutdownHandle { impl Drop for ShutdownHandle { fn drop(&mut self) { - self.flag.store(true, Ordering::Relaxed); + if Arc::strong_count(&self.flag) == 1 { + self.flag.store(true, Ordering::Relaxed); + } } } @@ -61,8 +64,8 @@ pub(crate) fn spawn( count.record(used, &used_attrs); } - let shutdown = Arc::new(AtomicBool::new(false)); - let flag = shutdown.clone(); + let flag = Arc::new(AtomicBool::new(false)); + let shutdown = Arc::downgrade(&flag); R::spawn(async move { let meter: Meter = opentelemetry::global::meter("sqlx-otel"); @@ -77,10 +80,12 @@ pub(crate) fn spawn( loop { R::sleep(interval).await; - if shutdown.load(Ordering::Relaxed) { - break; + match shutdown.upgrade() { + Some(flag) if !flag.load(Ordering::Relaxed) => { + record(&count, &pool, &base_attrs); + } + _ => break, } - record(&count, &pool, &base_attrs); } }); diff --git a/tests/pool_metrics.rs b/tests/pool_metrics.rs index 1f48866..e91003b 100644 --- a/tests/pool_metrics.rs +++ b/tests/pool_metrics.rs @@ -376,6 +376,42 @@ mod tokio_runtime { drop(conn); } + #[tokio::test] + #[serial] + async fn background_task_survives_clone_drop() { + let tel = common::TestTelemetry::install(); + let raw = sqlx::SqlitePool::connect(":memory:").await.unwrap(); + let pool = PoolBuilder::from(raw) + .with_pool_name(POOL_NAME) + .with_pool_metrics_interval(Duration::from_millis(50)) + .build(); + + let clone = pool.clone(); + // Dropping a clone must NOT stop the polling task + drop(clone); + + let conn = pool.acquire().await.unwrap(); + + let used = poll_for(Duration::from_secs(2), || { + let snapshot = tel.metrics(); + let used = gauge_value( + &snapshot, + "db.client.connection.count", + "db.client.connection.state", + "used", + )?; + if used >= 1 { Some(used) } else { None } + }) + .await; + + assert!( + used.is_some(), + "connection.count should keep updating after a clone is dropped" + ); + + drop(conn); + } + #[tokio::test] #[serial] async fn no_pool_metrics_without_pool_name() { From 0116f91ce411e88c7f55a2d1e846756a13da1b20 Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Sat, 8 Aug 2026 15:26:30 +0100 Subject: [PATCH 2/8] chore: Update dependencies --- Cargo.toml | 10 +++++----- tests/mysql.rs | 10 ++++++---- tests/postgres.rs | 10 ++++++---- 3 files changed, 17 insertions(+), 13 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f33f1d8..b0f7da5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,8 +24,8 @@ pedantic = "warn" [dependencies] futures = "0.3" -opentelemetry = "0.31" -opentelemetry-semantic-conventions = { version = "0.31", features = ["semconv_experimental"] } +opentelemetry = "0.32" +opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_experimental"] } sqlx = { version = "0.8", default-features = false } async-std = { version = "1", optional = true } tokio = { version = "1", features = ["rt", "time"], optional = true } @@ -42,12 +42,12 @@ all-features = true rustdoc-args = ["--cfg", "docsrs"] [dev-dependencies] -ctor = "0.10" -opentelemetry_sdk = { version = "0.31", features = ["testing", "rt-tokio"] } +dtor = "1" +opentelemetry_sdk = { version = "0.32", features = ["testing", "rt-tokio"] } proptest = "1" serial_test = "3" sqlx = { version = "0.8", features = ["macros", "sqlite", "postgres", "mysql", "runtime-tokio"] } -testcontainers = "0.27" +testcontainers = "0.28" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } [[example]] diff --git a/tests/mysql.rs b/tests/mysql.rs index 92269ec..ab8e133 100644 --- a/tests/mysql.rs +++ b/tests/mysql.rs @@ -63,11 +63,13 @@ async fn shared_container() -> &'static SharedContainer { /// Stop and remove the shared container at process exit. Required because the /// `ContainerAsync` value lives in a `'static` (`CONTAINER`), so the language never -/// runs its `Drop`. Using `ctor::dtor` schedules a synchronous shell-out to +/// runs its `Drop`. Using `dtor::dtor` schedules a synchronous shell-out to /// `docker rm -f` that fires after `main` returns – equivalent to the per-test -/// RAII cleanup that existed before the shared-container refactor (commit c29f995). -#[ctor::dtor] -fn drop_container() { +/// RAII cleanup that existed before the shared-container refactor (commit c29f995). The fn is +/// `unsafe` because `dtor` runs it outside the Rust runtime, where most std facilities carry no +/// guarantees; spawning a subprocess and reading a `OnceLock` is safe in practice. +#[dtor::dtor] +unsafe fn drop_container() { if let Some(id) = CONTAINER_ID.get() { let _ = std::process::Command::new("docker") .args(["rm", "-f", id.as_str()]) diff --git a/tests/postgres.rs b/tests/postgres.rs index d3f4e0d..0380ba3 100644 --- a/tests/postgres.rs +++ b/tests/postgres.rs @@ -63,11 +63,13 @@ async fn shared_container() -> &'static SharedContainer { /// Stop and remove the shared container at process exit. Required because the /// `ContainerAsync` value lives in a `'static` (`CONTAINER`), so the language never -/// runs its `Drop`. Using `ctor::dtor` schedules a synchronous shell-out to +/// runs its `Drop`. Using `dtor::dtor` schedules a synchronous shell-out to /// `docker rm -f` that fires after `main` returns – equivalent to the per-test -/// RAII cleanup that existed before the shared-container refactor (commit c29f995). -#[ctor::dtor] -fn drop_container() { +/// RAII cleanup that existed before the shared-container refactor (commit c29f995). The fn is +/// `unsafe` because `dtor` runs it outside the Rust runtime, where most std facilities carry no +/// guarantees; spawning a subprocess and reading a `OnceLock` is safe in practice. +#[dtor::dtor] +unsafe fn drop_container() { if let Some(id) = CONTAINER_ID.get() { let _ = std::process::Command::new("docker") .args(["rm", "-f", id.as_str()]) From 947ec61fd4aabd47316c895b36c737eabaed915d Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Sat, 8 Aug 2026 17:59:32 +0100 Subject: [PATCH 3/8] chore: Bump to 0.4.0 Cuts the release covering the opentelemetry 0.32 upgrade and the pool-metrics shutdown fix. Adds a per-release compatibility matrix and writes down the versioning policy the crate has followed since 0.1.0. Both sqlx and opentelemetry have to be upgraded in lockstep with this crate, and they fail differently when they are not: sqlx types are in the public signatures, so a mismatch is a compile error, whereas opentelemetry is coupled only through its `global` provider registry, so a mismatch compiles fine and silently emits nothing. Neither constraint was documented anywhere. Corrects a CHANGELOG slip: v0.3.0 was tagged and published on 2026-05-05 but its notes were still sitting under `## [Unreleased]`. --- CHANGELOG.md | 19 +++++++++++++++++-- Cargo.toml | 2 +- README.md | 27 +++++++++++++++++++++++---- 3 files changed, 41 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4e4d83..6126779 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,21 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.4.0] – 2026-08-08 + +### Changed + +- **Action required:** Upgraded to `opentelemetry` / `opentelemetry-semantic-conventions` `0.32`. Upgrade your own `opentelemetry` and `opentelemetry_sdk` to match, or spans and metrics stop being emitted. + +### Fixed + +- The pool-metrics polling task no longer stops when a single cloned `Pool` is dropped. All clones share one `Arc` shutdown flag, so the first clone to drop was flipping it and silently ending `db.client.connection.*` collection for every surviving clone. The task now stops only once the last clone is dropped, matching what the documentation already implied. + +### Internal + +- Test-only: replaced `ctor` with the `dtor` crate, whose `#[dtor]` macro was split out of `ctor` and removed in `ctor` 1.0, and marked the container-cleanup destructors `unsafe` per `dtor` 1.0's deprecation of the safe form. Bumped `testcontainers` to `0.28`. No effect on the published crate. + +## [0.3.0] – 2026-05-05 ### Added @@ -46,6 +60,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - SQLite backend ([#3](https://github.com/chmodas/sqlx-otel/pull/3)). - Backend-agnostic instrumentation core: `Pool`, `PoolBuilder`, `Transaction`, `PoolConnection`, and the `Executor` trait wiring that emits OpenTelemetry-native spans and metrics following the [database calls and systems](https://opentelemetry.io/docs/specs/semconv/db/) semantic conventions ([#2](https://github.com/chmodas/sqlx-otel/pull/2)). -[Unreleased]: https://github.com/chmodas/sqlx-otel/compare/v0.1.0...HEAD [0.1.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.1.0 [0.2.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.2.0 +[0.3.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.3.0 +[0.4.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.4.0 diff --git a/Cargo.toml b/Cargo.toml index b0f7da5..9f438b8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sqlx-otel" -version = "0.3.0" +version = "0.4.0" edition = "2024" authors = ["Borislav Borisov "] license = "MIT OR Apache-2.0" diff --git a/README.md b/README.md index 19d8642..434155b 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ The wrapper talks to the [`opentelemetry`](https://docs.rs/opentelemetry) API di ```toml [dependencies] -sqlx-otel = { version = "0.2.0", features = ["postgres", "runtime-tokio"] } +sqlx-otel = { version = "0.4.0", features = ["postgres", "runtime-tokio"] } ``` ```rust @@ -221,9 +221,28 @@ The first four are recorded inline on every `acquire()` / connection drop – no ## Compatibility -- **MSRV:** Rust **1.85.0**. -- **SQLx:** `0.8.x`. -- **OpenTelemetry:** `0.31.x`. +| sqlx-otel | sqlx | opentelemetry | MSRV | +| --------- | ------- | ------------- | -------- | +| `0.4.x` | `0.8.x` | `0.32.x` | `1.85.0` | +| `0.3.x` | `0.8.x` | `0.31.x` | `1.85.0` | +| `0.2.x` | `0.8.x` | `0.31.x` | `1.85.0` | +| `0.1.x` | `0.8.x` | `0.31.x` | `1.85.0` | + +Pick the row matching the `sqlx` and `opentelemetry` versions your application already uses. Older releases stay on crates.io indefinitely and are never yanked, so pinning to an earlier row is a supported way to stay put. + +## Versioning policy + +`sqlx` and `opentelemetry` are both pre-1.0, so Cargo compares their *minor* version for compatibility: `0.31` and `0.32` are separate ranges that never unify, whether or not the release actually changed anything. If your application's versions drift from the ones sqlx-otel was built against, Cargo resolves both copies into the build graph, and the two libraries fail differently. + +**`sqlx` fails at compile time.** Its types are part of this crate's public API, and types from semver-incompatible releases are distinct types. A `Pool` from `sqlx` `0.9` will not go into a wrapper built against `sqlx` `0.8`. + +**`opentelemetry` fails silently at runtime.** It coordinates through process-global state, so two copies means two registries. Your application installs its provider in one; the wrapper emits into the other. Spans and metrics disappear, with no error and no warning. + +Hence: + +- Either library bumping its minor triggers a minor release here. +- Bugs are not backported. Fixes ship in the next release. +- MSRV tracks [sqlx's MSRV policy](https://github.com/transact-rs/sqlx/blob/main/FAQ.md#MSRV). ## License From e6bb1f8b12827c08590f8834ebc888b5e05a9487 Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Sun, 9 Aug 2026 10:41:05 +0100 Subject: [PATCH 4/8] chore: Bump to 0.5.0 Upgrades to sqlx 0.9, which raises the minimum Rust version to 1.94. sqlx 0.9 dropped the `'q` lifetime from statements and arguments. That changed the signatures of `prepare`, `prepare_with` and `describe`, and made `Execute::sql` consume the query rather than borrow it. The wrapper needs the query text for `db.query.text` and still has to pass the query on to the inner executor, which a consuming `sql` rules out, so queries are now taken apart and reassembled first. See `src/rebuilt_query.rs`. `sqlx-core` and the per-backend driver crates become direct dependencies so their `offline` feature can be enabled. Without it `Executor::describe` is absent from the trait; with it, implementing it is mandatory. --- ...bf617409027a30287fa9c840b4dbee9edea39.json | 6 +- ...b665004336f5d6208dcff09652c90c8f1137c.json | 3 +- ...8fd16bff357c676870d1445cbb7bae98b142c.json | 6 +- ...7f5156e02304c09b2781f347f52ed6ba15aad.json | 3 +- CHANGELOG.md | 13 +- Cargo.toml | 20 +- README.md | 3 +- src/executor.rs | 75 ++-- src/lib.rs | 1 + src/query_ext.rs | 16 +- src/rebuilt_query.rs | 329 ++++++++++++++++++ tests/common.rs | 200 ++++++++--- tests/mysql.rs | 14 +- tests/postgres.rs | 14 +- tests/sqlite.rs | 12 + 15 files changed, 608 insertions(+), 107 deletions(-) create mode 100644 src/rebuilt_query.rs diff --git a/.sqlx/query-85f87901b3da4789fefdedfa1b0bf617409027a30287fa9c840b4dbee9edea39.json b/.sqlx/query-85f87901b3da4789fefdedfa1b0bf617409027a30287fa9c840b4dbee9edea39.json index bf47f43..4fd4c1f 100644 --- a/.sqlx/query-85f87901b3da4789fefdedfa1b0bf617409027a30287fa9c840b4dbee9edea39.json +++ b/.sqlx/query-85f87901b3da4789fefdedfa1b0bf617409027a30287fa9c840b4dbee9edea39.json @@ -9,7 +9,8 @@ "type_info": { "type": "Long", "flags": "NOT_NULL | PRIMARY_KEY | NO_DEFAULT_VALUE", - "max_size": 11 + "max_size": 11, + "collation": 63 } }, { @@ -18,7 +19,8 @@ "type_info": { "type": "VarString", "flags": "NOT_NULL | NO_DEFAULT_VALUE", - "max_size": 1020 + "max_size": 1020, + "collation": 45 } } ], diff --git a/.sqlx/query-b5729cb4e4c5fea522e4e1bc241b665004336f5d6208dcff09652c90c8f1137c.json b/.sqlx/query-b5729cb4e4c5fea522e4e1bc241b665004336f5d6208dcff09652c90c8f1137c.json index a9bdbe7..5c18415 100644 --- a/.sqlx/query-b5729cb4e4c5fea522e4e1bc241b665004336f5d6208dcff09652c90c8f1137c.json +++ b/.sqlx/query-b5729cb4e4c5fea522e4e1bc241b665004336f5d6208dcff09652c90c8f1137c.json @@ -9,7 +9,8 @@ "type_info": { "type": "VarString", "flags": "NOT_NULL | NO_DEFAULT_VALUE", - "max_size": 1020 + "max_size": 1020, + "collation": 45 } } ], diff --git a/.sqlx/query-c4c1bfc60ab673a3dd91f31203e8fd16bff357c676870d1445cbb7bae98b142c.json b/.sqlx/query-c4c1bfc60ab673a3dd91f31203e8fd16bff357c676870d1445cbb7bae98b142c.json index a1d6850..4591f65 100644 --- a/.sqlx/query-c4c1bfc60ab673a3dd91f31203e8fd16bff357c676870d1445cbb7bae98b142c.json +++ b/.sqlx/query-c4c1bfc60ab673a3dd91f31203e8fd16bff357c676870d1445cbb7bae98b142c.json @@ -9,7 +9,8 @@ "type_info": { "type": "Long", "flags": "NOT_NULL | PRIMARY_KEY | NO_DEFAULT_VALUE", - "max_size": 11 + "max_size": 11, + "collation": 63 } }, { @@ -18,7 +19,8 @@ "type_info": { "type": "VarString", "flags": "NOT_NULL | NO_DEFAULT_VALUE", - "max_size": 1020 + "max_size": 1020, + "collation": 45 } } ], diff --git a/.sqlx/query-c74c0d0a1f4a4b60988a49740837f5156e02304c09b2781f347f52ed6ba15aad.json b/.sqlx/query-c74c0d0a1f4a4b60988a49740837f5156e02304c09b2781f347f52ed6ba15aad.json index a999bd0..9a17680 100644 --- a/.sqlx/query-c74c0d0a1f4a4b60988a49740837f5156e02304c09b2781f347f52ed6ba15aad.json +++ b/.sqlx/query-c74c0d0a1f4a4b60988a49740837f5156e02304c09b2781f347f52ed6ba15aad.json @@ -9,7 +9,8 @@ "type_info": { "type": "Long", "flags": "NOT_NULL | PRIMARY_KEY | NO_DEFAULT_VALUE", - "max_size": 11 + "max_size": 11, + "collation": 63 } } ], diff --git a/CHANGELOG.md b/CHANGELOG.md index 6126779..1448e02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,20 +4,22 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.5.0] – 2026-08-09 + +### Changed + +- **BREAKING:** Upgraded to `sqlx` `0.9`, which raised the minimum supported Rust version from `1.85.0` to `1.94.0`. + ## [0.4.0] – 2026-08-08 ### Changed -- **Action required:** Upgraded to `opentelemetry` / `opentelemetry-semantic-conventions` `0.32`. Upgrade your own `opentelemetry` and `opentelemetry_sdk` to match, or spans and metrics stop being emitted. +- **Action required:** Upgraded to `opentelemetry` and `opentelemetry-semantic-conventions` `0.32`. Applications must move their own `opentelemetry` and `opentelemetry_sdk` to `0.32` as well, or spans and metrics stop being emitted. ### Fixed - The pool-metrics polling task no longer stops when a single cloned `Pool` is dropped. All clones share one `Arc` shutdown flag, so the first clone to drop was flipping it and silently ending `db.client.connection.*` collection for every surviving clone. The task now stops only once the last clone is dropped, matching what the documentation already implied. -### Internal - -- Test-only: replaced `ctor` with the `dtor` crate, whose `#[dtor]` macro was split out of `ctor` and removed in `ctor` 1.0, and marked the container-cleanup destructors `unsafe` per `dtor` 1.0's deprecation of the safe form. Bumped `testcontainers` to `0.28`. No effect on the published crate. - ## [0.3.0] – 2026-05-05 ### Added @@ -64,3 +66,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), [0.2.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.2.0 [0.3.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.3.0 [0.4.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.4.0 +[0.5.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.5.0 diff --git a/Cargo.toml b/Cargo.toml index 9f438b8..990c021 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sqlx-otel" -version = "0.4.0" +version = "0.5.0" edition = "2024" authors = ["Borislav Borisov "] license = "MIT OR Apache-2.0" @@ -11,7 +11,7 @@ documentation = "https://docs.rs/sqlx-otel/" description = "Thin wrapper around SQLx that emits OpenTelemetry spans and metrics following the database client semantic conventions." categories = ["database", "asynchronous", "development-tools::profiling"] keywords = ["sqlx", "opentelemetry", "metrics", "database", "instrumentation"] -rust-version = "1.85.0" +rust-version = "1.94.0" exclude = [ ".gitignore", @@ -26,14 +26,20 @@ pedantic = "warn" futures = "0.3" opentelemetry = "0.32" opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_experimental"] } -sqlx = { version = "0.8", default-features = false } +# `offline` is what makes `Executor::describe` a present, required # trait method, and it has to +# be set on `sqlx-core` and on every driver. +sqlx = { version = "0.9", default-features = false } +sqlx-core = { version = "0.9", default-features = false, features = ["offline"] } +sqlx-sqlite = { version = "0.9", default-features = false, features = ["offline"], optional = true } +sqlx-postgres = { version = "0.9", default-features = false, features = ["offline"], optional = true } +sqlx-mysql = { version = "0.9", default-features = false, features = ["offline"], optional = true } async-std = { version = "1", optional = true } tokio = { version = "1", features = ["rt", "time"], optional = true } [features] -sqlite = ["sqlx/sqlite"] -postgres = ["sqlx/postgres"] -mysql = ["sqlx/mysql"] +sqlite = ["sqlx/sqlite", "dep:sqlx-sqlite"] +postgres = ["sqlx/postgres", "dep:sqlx-postgres"] +mysql = ["sqlx/mysql", "dep:sqlx-mysql"] runtime-tokio = ["dep:tokio"] runtime-async-std = ["dep:async-std"] @@ -46,7 +52,7 @@ dtor = "1" opentelemetry_sdk = { version = "0.32", features = ["testing", "rt-tokio"] } proptest = "1" serial_test = "3" -sqlx = { version = "0.8", features = ["macros", "sqlite", "postgres", "mysql", "runtime-tokio"] } +sqlx = { version = "0.9", features = ["macros", "sqlite", "postgres", "mysql", "mysql-rsa", "runtime-tokio"] } testcontainers = "0.28" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } diff --git a/README.md b/README.md index 434155b..6fa69b3 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ The wrapper talks to the [`opentelemetry`](https://docs.rs/opentelemetry) API di ```toml [dependencies] -sqlx-otel = { version = "0.4.0", features = ["postgres", "runtime-tokio"] } +sqlx-otel = { version = "0.5.0", features = ["postgres", "runtime-tokio"] } ``` ```rust @@ -223,6 +223,7 @@ The first four are recorded inline on every `acquire()` / connection drop – no | sqlx-otel | sqlx | opentelemetry | MSRV | | --------- | ------- | ------------- | -------- | +| `0.5.x` | `0.9.x` | `0.32.x` | `1.94.0` | | `0.4.x` | `0.8.x` | `0.32.x` | `1.85.0` | | `0.3.x` | `0.8.x` | `0.31.x` | `1.85.0` | | `0.2.x` | `0.8.x` | `0.31.x` | `1.85.0` | diff --git a/src/executor.rs b/src/executor.rs index be6128c..ad20356 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -156,15 +156,15 @@ fn record_error(cx: &OtelContext, err: &sqlx::Error, metric_attrs: &mut Vec, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, mut metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let fut = ($inner).execute(query); Box::pin(async move { let result = fut.await; @@ -429,10 +429,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let stream = ($inner).execute_many(query); Box::pin(InstrumentedStream::<_, CountNone>::new( stream, @@ -452,10 +452,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let stream = ($inner).fetch(query); Box::pin(InstrumentedStream::<_, CountAll>::new( stream, @@ -485,10 +485,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let stream = ($inner).fetch_many(query); Box::pin(InstrumentedStream::<_, CountRight>::new( stream, @@ -512,10 +512,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, mut metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let fut = ($inner).fetch_all(query); Box::pin(async move { let result = fut.await; @@ -546,10 +546,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, mut metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let fut = ($inner).fetch_one(query); Box::pin(async move { let result = fut.await; @@ -579,10 +579,10 @@ macro_rules! impl_executor { E: 'q + sqlx::Execute<'q, DB>, 'c: 'e, { - let sql = query.sql().to_owned(); + let query = crate::rebuilt_query::RebuiltQuery::::split(query); let state = $self_.state.clone(); let (cx, start, mut metric_attrs) = - begin_query_span(&state.attrs, Some(&sql), $ann); + begin_query_span(&state.attrs, Some(query.sql_str()), $ann); let fut = ($inner).fetch_optional(query); Box::pin(async move { let result = fut.await; @@ -609,18 +609,19 @@ macro_rules! impl_executor { /// /// This explicit API is provided to allow access to the statement metadata /// available after it prepared but before the first row is returned. - fn prepare<'e, 'q: 'e>( + fn prepare<'e>( $self_, - query: &'q str, + query: sqlx::SqlStr, ) -> futures::future::BoxFuture< 'e, - Result<::Statement<'q>, sqlx::Error>, + Result<::Statement, sqlx::Error>, > where 'c: 'e, { let state = $self_.state.clone(); - let (cx, start, metric_attrs) = begin_query_span(&state.attrs, Some(query), $ann); + let (cx, start, metric_attrs) = + begin_query_span(&state.attrs, Some(query.as_str()), $ann); let fut = ($inner).prepare(query); Box::pin(execute_instrumented( fut, cx, start, state.metrics, metric_attrs, @@ -632,19 +633,20 @@ macro_rules! impl_executor { /// /// Only some database drivers (Postgres, MSSQL) can take advantage of /// this extra information to influence parameter type inference. - fn prepare_with<'e, 'q: 'e>( + fn prepare_with<'e>( $self_, - sql: &'q str, + sql: sqlx::SqlStr, parameters: &'e [::TypeInfo], ) -> futures::future::BoxFuture< 'e, - Result<::Statement<'q>, sqlx::Error>, + Result<::Statement, sqlx::Error>, > where 'c: 'e, { let state = $self_.state.clone(); - let (cx, start, metric_attrs) = begin_query_span(&state.attrs, Some(sql), $ann); + let (cx, start, metric_attrs) = + begin_query_span(&state.attrs, Some(sql.as_str()), $ann); let fut = ($inner).prepare_with(sql, parameters); Box::pin(execute_instrumented( fut, cx, start, state.metrics, metric_attrs, @@ -655,11 +657,13 @@ macro_rules! impl_executor { /// and results. /// /// This is used by compile-time verification in the query macros to - /// power their type inference. + /// power their type inference. The macros call it on a connection they open + /// themselves from `DATABASE_URL`, never through this wrapper, so a span emitted + /// here is always a caller invoking `describe` at runtime against a live pool. #[doc(hidden)] - fn describe<'e, 'q: 'e>( + fn describe<'e>( $self_, - sql: &'q str, + sql: sqlx::SqlStr, ) -> futures::future::BoxFuture< 'e, Result, sqlx::Error>, @@ -668,7 +672,8 @@ macro_rules! impl_executor { 'c: 'e, { let state = $self_.state.clone(); - let (cx, start, metric_attrs) = begin_query_span(&state.attrs, Some(sql), $ann); + let (cx, start, metric_attrs) = + begin_query_span(&state.attrs, Some(sql.as_str()), $ann); let fut = ($inner).describe(sql); Box::pin(execute_instrumented( fut, cx, start, state.metrics, metric_attrs, diff --git a/src/lib.rs b/src/lib.rs index abe425a..0141313 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -236,6 +236,7 @@ mod obfuscate; mod pool; mod pool_metrics; mod query_ext; +mod rebuilt_query; mod runtime; mod transaction; diff --git a/src/query_ext.rs b/src/query_ext.rs index 9eda556..8577a56 100644 --- a/src/query_ext.rs +++ b/src/query_ext.rs @@ -452,7 +452,7 @@ macro_rules! impl_annotated_query_fetch_forwarders { /// Emit the `bind` method on a specialised impl block. /// -/// `SQLx` restricts `bind` to the default `::Arguments<'q>` parameter, so +/// `SQLx` restricts `bind` to the default `::Arguments` parameter, so /// each builder family has its own dedicated impl block keyed on the default arguments. macro_rules! impl_annotated_query_bind { () => { @@ -473,7 +473,7 @@ macro_rules! impl_annotated_query_bind { impl<'q, DB, A> AnnotatedQuery> where DB: Database, - A: 'q + Send + sqlx::IntoArguments<'q, DB>, + A: 'q + Send + sqlx::IntoArguments, { impl_annotated_query_fetch_forwarders!(row = DB::Row, extra_bounds = ()); @@ -547,7 +547,7 @@ where } } -impl<'q, DB> AnnotatedQuery::Arguments<'q>>> +impl<'q, DB> AnnotatedQuery::Arguments>> where DB: sqlx::Database, { @@ -559,13 +559,13 @@ where impl<'q, DB, O, A> AnnotatedQuery> where DB: Database, - A: 'q + Send + sqlx::IntoArguments<'q, DB>, + A: 'q + Send + sqlx::IntoArguments, O: Send + Unpin + for<'r> sqlx::FromRow<'r, DB::Row>, { impl_annotated_query_fetch_forwarders!(row = O, extra_bounds = (DB: 'e, O: 'e,)); } -impl<'q, DB, O> AnnotatedQuery::Arguments<'q>>> +impl<'q, DB, O> AnnotatedQuery::Arguments>> where DB: sqlx::Database, { @@ -577,14 +577,14 @@ where impl<'q, DB, O, A> AnnotatedQuery> where DB: Database, - A: 'q + Send + sqlx::IntoArguments<'q, DB>, + A: 'q + Send + sqlx::IntoArguments, O: Send + Unpin, (O,): Send + Unpin + for<'r> sqlx::FromRow<'r, DB::Row>, { impl_annotated_query_fetch_forwarders!(row = O, extra_bounds = (DB: 'e, O: 'e,)); } -impl<'q, DB, O> AnnotatedQuery::Arguments<'q>>> +impl<'q, DB, O> AnnotatedQuery::Arguments>> where DB: sqlx::Database, { @@ -598,7 +598,7 @@ where DB: Database, F: FnMut(DB::Row) -> Result + Send, O: Send + Unpin, - A: 'q + Send + sqlx::IntoArguments<'q, DB>, + A: 'q + Send + sqlx::IntoArguments, { impl_annotated_query_fetch_forwarders!(row = O, extra_bounds = (DB: 'e, F: 'e, O: 'e,)); diff --git a/src/rebuilt_query.rs b/src/rebuilt_query.rs new file mode 100644 index 0000000..b4e9a63 --- /dev/null +++ b/src/rebuilt_query.rs @@ -0,0 +1,329 @@ +//! Taking a [`sqlx::Execute`] value apart, so the SQL text can be read for `db.query.text` while +//! the query is still forwarded to the inner executor. [`sqlx::Execute::sql`] consumes the query, +//! which would otherwise make those two mutually exclusive. +//! +//! A driver reads four things off an `Execute` value: `sql`, `statement`, `take_arguments` and +//! `persistent`. [`RebuiltQuery`] holds all four and reports each one exactly as the original +//! query would have, so the driver cannot tell it apart from what the caller passed in. Rebuilding +//! through `sqlx::query_with` / `sqlx::raw_sql` instead would not manage that: `raw_sql` forces +//! `persistent` to `false`, and neither constructor can carry a cached statement handle. + +use sqlx::error::BoxDynError; + +use crate::database::Database; + +/// A [`sqlx::Execute`] value taken apart into the four values a driver reads from it, retaining +/// the SQL text for span and metric attributes. +/// +/// Constructed by [`RebuiltQuery::split`] and immediately handed to the inner executor. It is never +/// held across an await point and never escapes the `Executor` method that built it. +pub(crate) struct RebuiltQuery { + /// The query text, moved out of the original query. Also the source for `db.query.text`. + sql: sqlx::SqlStr, + /// A cached statement handle, when the caller built the query from a prepared statement. + /// Postgres reads it to skip re-resolving statement metadata. + statement: Option, + /// The arguments, already taken from the original query. `None` once taken by the driver; + /// `Some(Ok(None))` means "use the simple query protocol and do not prepare"; `Some(Err(_))` + /// carries an encoding failure that must surface at execution time, not here. + arguments: Option, BoxDynError>>, + /// Whether the driver should cache the prepared statement. + persistent: bool, +} + +impl RebuiltQuery { + /// Take a query apart, retaining everything the driver will ask for. + /// + /// Do not reorder the reads below. `Map::persistent()` is `self.inner.arguments.is_some()`, so + /// it reports `false` once the arguments have been taken. Drivers take the arguments first, so + /// `false` is what they see; reading `persistent` before `take_arguments` would capture `true` + /// instead and switch on statement caching that sqlx does not do for `Map` – the shape every + /// `query!()` / `query_as!()` expands to. + pub(crate) fn split<'q, E>(mut query: E) -> Self + where + E: sqlx::Execute<'q, DB>, + { + // 1. Borrowed, so it must be cloned before the query is consumed in step 4. `Statement` is + // `Clone` by trait bound and the concrete impls are cheap (an `SqlStr` plus an `Arc`). + let statement = query.statement().cloned(); + // 2. Must precede the `persistent` read – see the note above. + let arguments = query.take_arguments(); + // 3. Reflects the post-take state, matching what the driver would have observed. + let persistent = query.persistent(); + // 4. Consumes `query`; nothing may touch it afterwards. + let sql = query.sql(); + + Self { + sql, + statement, + arguments: Some(arguments), + persistent, + } + } + + /// Borrow the query text for span and metric attributes. + pub(crate) fn sql_str(&self) -> &str { + self.sql.as_str() + } +} + +impl sqlx::Execute<'_, DB> for RebuiltQuery { + fn sql(self) -> sqlx::SqlStr { + self.sql + } + + fn statement(&self) -> Option<&DB::Statement> { + self.statement.as_ref() + } + + /// Yield the arguments taken in [`RebuiltQuery::split`], exactly once. A second call returns + /// `Ok(None)`, as `sqlx::query::Query` does. + fn take_arguments(&mut self) -> Result, BoxDynError> { + self.arguments.take().unwrap_or(Ok(None)) + } + + fn persistent(&self) -> bool { + self.persistent + } +} + +#[cfg(all(test, feature = "sqlite"))] +mod tests { + use proptest::prelude::*; + use sqlx::{Arguments as _, Execute as _, Sqlite, Statement as _}; + + use super::*; + + /// The four values a driver reads off a [`sqlx::Execute`], flattened so an original and its + /// rebuilt counterpart can be compared directly. + /// + /// Arguments are reduced to a classification plus a length rather than compared structurally, + /// because `Arguments` exposes no equality and the encoded buffer is driver-private. The + /// distinction that matters – `Some` vs `None` vs `Err` – is preserved exactly, and `None` is + /// the one that decides between the prepared and simple query protocols. + #[derive(Debug, PartialEq, Eq)] + struct Observables { + sql: String, + statement_sql: Option, + arguments: ArgClass, + persistent: bool, + } + + #[derive(Debug, PartialEq, Eq)] + enum ArgClass { + /// Prepared, carrying this many encoded arguments. + Some(usize), + /// Simple query protocol – the driver must not prepare this statement. + None, + /// Argument encoding failed before execution. + Err, + } + + /// Read all four in the order every sqlx driver uses. Consumes the query, because + /// `sql()` does. + fn observe<'q, E: sqlx::Execute<'q, Sqlite>>(mut query: E) -> Observables { + let statement_sql = query + .statement() + .map(|s| sqlx::Statement::sql(s).as_str().to_owned()); + let arguments = match query.take_arguments() { + Ok(Some(a)) => ArgClass::Some(a.len()), + Ok(None) => ArgClass::None, + Err(_) => ArgClass::Err, + }; + let persistent = query.persistent(); + Observables { + sql: query.sql().as_str().to_owned(), + statement_sql, + arguments, + persistent, + } + } + + /// The shapes of `Execute` value this crate can be handed. Each maps to a different + /// combination of those four values, so covering all six is what makes the parity property + /// meaningful. + #[derive(Debug, Clone)] + enum Kind { + /// `sqlx::query(..)` with no binds. + Plain, + /// `sqlx::query(..)` with `n` binds. + Prepared(usize), + /// `sqlx::raw_sql(..)` – the `Ok(None)` + `persistent == false` case. + Raw, + /// `.map(..)`, which `query!()` / `query_as!()` expand to. Its `persistent()` is derived + /// from whether arguments are still present, so it is the only shape that detects a + /// reordering of the reads in `split`. Do not drop it from this generator. + Mapped, + /// An `AssertSqlSafe(String)`, routed through the blanket `Execute for T: SqlSafeStr` impl + /// – the same path a bare `&'static str` takes. + Bare, + /// A query carrying an argument that fails to encode, so `take_arguments` yields `Err`. + FailedEncode, + } + + /// A bind value whose `Encode` impl always fails, so `Query::bind` stores an `Err` in place of + /// the arguments. Nothing in sqlx's own `SQLite` types can be made to fail on demand, and the + /// `Err` branch is the one that must not be silently swallowed into `Ok(None)` – which would + /// turn a failed encode into a successful unprepared execution. + struct AlwaysFailsToEncode; + + impl sqlx::Type for AlwaysFailsToEncode { + fn type_info() -> ::TypeInfo { + >::type_info() + } + } + + impl sqlx::Encode<'_, Sqlite> for AlwaysFailsToEncode { + fn encode_by_ref( + &self, + _buf: &mut ::ArgumentBuffer, + ) -> Result { + Err("deliberate encode failure".into()) + } + } + + fn kind_strategy() -> impl Strategy { + prop_oneof![ + Just(Kind::Plain), + (0usize..4).prop_map(Kind::Prepared), + Just(Kind::Raw), + Just(Kind::Mapped), + Just(Kind::Bare), + Just(Kind::FailedEncode), + ] + } + + /// Observe a freshly built query twice – once directly, once through [`RebuiltQuery::split`] – + /// and require the two to agree. + /// + /// The builder expression is evaluated twice rather than the value being cloned, because + /// `Execute` is consumed by inspection and is not `Clone`. Kept as a macro rather than a + /// function taking `impl Fn`: the concrete query types (`Query<'q, ..>`, `Map<'q, ..>`) borrow + /// from their SQL, so a trait-object or HRTB erasure would not admit them. + macro_rules! assert_split_parity { + ($build:expr) => {{ + let direct = observe($build); + let split = observe(RebuiltQuery::::split($build)); + prop_assert_eq!(direct, split); + }}; + } + + /// Non-empty, non-control SQL text. The content is never parsed – only carried – so the + /// generator targets byte-level fidelity rather than syntactic validity. + fn sql_strategy() -> impl Strategy { + prop_oneof![ + Just("SELECT 1".to_owned()), + Just(String::new()), + Just("SELECT '\u{1F600} πŸ¦€ \u{202E}'".to_owned()), + Just("SELECT ?1, ?2, ?3".to_owned()), + "[ -~]{0,64}".prop_map(|s| s), + any::(), + ] + } + + proptest! { + #![proptest_config(ProptestConfig::with_cases(256))] + + /// Algorithmic parity: splitting a query and observing the rebuilt value must yield + /// exactly what observing the original would have. This is the property that proves all + /// three argument branches – `Some`, `None`, and `Err` – round-trip, and that the cached + /// statement handle and `persistent` flag survive with them. `Kind::FailedEncode` is what + /// reaches the `Err` arm; without it the branch would be unreachable from this generator. + #[test] + fn split_preserves_all_observables(kind in kind_strategy(), sql in sql_strategy()) { + let owned = || sqlx::AssertSqlSafe(sql.clone()); + match kind { + Kind::Plain => assert_split_parity!(sqlx::query::(owned())), + Kind::Prepared(n) => assert_split_parity!({ + let mut q = sqlx::query::(owned()); + for i in 0..n { + q = q.bind(i32::try_from(i).unwrap_or(i32::MAX)); + } + q + }), + Kind::Raw => assert_split_parity!(sqlx::raw_sql(owned())), + Kind::Mapped => assert_split_parity!(sqlx::query::(owned()).map(|_| ())), + Kind::Bare => assert_split_parity!(owned()), + Kind::FailedEncode => { + assert_split_parity!(sqlx::query::(owned()).bind(AlwaysFailsToEncode)); + } + } + } + + /// The SQL text survives byte-for-byte, whatever the input contains. `db.query.text` is + /// derived from this, so any lossiness here would corrupt exported spans. + #[test] + fn sql_text_round_trips_byte_for_byte(sql in any::()) { + let q = sqlx::query::(sqlx::AssertSqlSafe(sql.clone())); + let rebuilt = RebuiltQuery::::split(q); + prop_assert_eq!(rebuilt.sql_str(), sql.as_str()); + } + + /// Splitting never panics, whatever the SQL and bind count. + #[test] + fn split_never_panics(sql in any::(), binds in 0usize..8) { + let mut q = sqlx::query::(sqlx::AssertSqlSafe(sql)); + for i in 0..binds { + q = q.bind(i32::try_from(i).unwrap_or(i32::MAX)); + } + let _ = RebuiltQuery::::split(q); + } + } + + /// A statement handle from `query_statement` is preserved rather than downgraded to plain + /// text. Postgres reads it to skip re-resolving statement metadata. + #[tokio::test] + async fn statement_handle_survives_split() { + use sqlx::{Connection as _, Executor as _}; + + let mut conn = sqlx::SqliteConnection::connect(":memory:").await.unwrap(); + let stmt = conn + .prepare(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); + let rebuilt = RebuiltQuery::::split(stmt.query()); + + let carried = sqlx::Execute::::statement(&rebuilt); + assert!(carried.is_some(), "cached statement handle must survive"); + assert_eq!(carried.map(|s| s.sql().as_str()), Some("SELECT 1")); + } + + /// Arguments are yielded exactly once. A second take degrades to `Ok(None)` rather than + /// panicking, matching `sqlx::query::Query`, whose `take_arguments` is + /// `self.arguments.take().transpose()`. + #[test] + fn take_arguments_is_idempotent_after_the_first_call() { + let mut rebuilt = RebuiltQuery::::split(sqlx::query::("SELECT ?1").bind(1)); + + assert!(matches!(rebuilt.take_arguments(), Ok(Some(_)))); + assert!(matches!(rebuilt.take_arguments(), Ok(None))); + assert!(matches!(rebuilt.take_arguments(), Ok(None))); + } + + /// An encoding failure surfaces once, at execution time, and does not resurrect on a second + /// take. Constructed directly because provoking a real encode failure through the public + /// builder is driver-specific; `RebuiltQuery` is crate-private, so this is legitimate. + #[test] + fn argument_encoding_error_surfaces_once_then_degrades_to_none() { + let mut rebuilt = RebuiltQuery:: { + sql: sqlx::SqlStr::from_static("SELECT 1"), + statement: None, + arguments: Some(Err("encode failure".into())), + persistent: true, + }; + + assert!(rebuilt.take_arguments().is_err()); + assert!(matches!(rebuilt.take_arguments(), Ok(None))); + } + + /// `raw_sql` is the one shape carrying `Ok(None)` *and* `persistent == false`. Collapsing it + /// into a `query_with` rebuild would silently promote an unprepared statement to a prepared + /// one, so the combination is pinned explicitly. + #[test] + fn raw_sql_keeps_simple_protocol_and_non_persistent() { + let rebuilt = RebuiltQuery::::split(sqlx::raw_sql("SELECT 1; SELECT 2;")); + + assert!(!sqlx::Execute::::persistent(&rebuilt)); + assert_eq!(rebuilt.sql_str(), "SELECT 1; SELECT 2;"); + } +} diff --git a/tests/common.rs b/tests/common.rs index a3749cb..ba1a069 100644 --- a/tests/common.rs +++ b/tests/common.rs @@ -177,10 +177,10 @@ pub fn find_duration_data_point( if metric.name() != "db.client.operation.duration" { continue; } - if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() { - if let Some(dp) = hist.data_points().next() { - return Some(dp.clone()); - } + if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() + && let Some(dp) = hist.data_points().next() + { + return Some(dp.clone()); } } } @@ -552,10 +552,15 @@ pub const MYSQL_DIALECT: Dialect = Dialect { macro_rules! fresh_table { ($pool:expr, $table:expr, $columns:expr) => {{ use sqlx::Executor as _; + // `AssertSqlSafe` because a bare `&str` must be `'static`; these are built at runtime from + // test-controlled table and column names. let drop_sql = format!("DROP TABLE IF EXISTS {}", $table); - $pool.execute(drop_sql.as_str()).await.unwrap(); + $pool.execute(sqlx::AssertSqlSafe(drop_sql)).await.unwrap(); let create_sql = format!("CREATE TABLE {} ({})", $table, $columns); - $pool.execute(create_sql.as_str()).await.unwrap(); + $pool + .execute(sqlx::AssertSqlSafe(create_sql)) + .await + .unwrap(); }}; } @@ -1574,6 +1579,61 @@ macro_rules! test_fetch_all_records_error { }}; } +/// `raw_sql` through the wrapper: the batch executes, and all of it reaches `db.query.text`. +/// +/// This does not show that the simple query protocol was selected. `sqlx::query` passes here too, +/// because `SQLite` tolerates multi-statement text on the prepared path. That `Ok(None)` and +/// `persistent == false` round-trip is proved by `split_preserves_all_observables` in +/// `src/rebuilt_query.rs`; this covers the wiring to it. +#[macro_export] +macro_rules! test_raw_sql_preserves_simple_protocol { + ($pool_factory:expr, $dialect:expr) => {{ + use sqlx::Executor as _; + let tel = $crate::common::TestTelemetry::install(); + let pool = $pool_factory; + + (&pool) + .execute(sqlx::raw_sql("SELECT 1; SELECT 2;")) + .await + .expect("raw_sql batch must stay on the simple query protocol through the wrapper"); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + $crate::common::assert_common_span_attributes(&spans[0], $dialect.system); + assert_eq!( + $crate::common::attr(&spans[0], "db.query.text"), + Some(opentelemetry::Value::String("SELECT 1; SELECT 2;".into())), + "the whole batch must reach db.query.text, not just the first statement" + ); + }}; +} + +/// A query built from an owned `String` reports its SQL verbatim. +/// +/// Runtime-built SQL arrives as `AssertSqlSafe`, a different `SqlStr` representation than a string +/// literal. `db.query.text` must come out the same either way. +#[macro_export] +macro_rules! test_owned_sql_string_query_text { + ($pool_factory:expr, $dialect:expr) => {{ + use sqlx::Executor as _; + let tel = $crate::common::TestTelemetry::install(); + let pool = $pool_factory; + + let sql = format!("SELECT {}", 1); + (&pool) + .fetch_one(sqlx::AssertSqlSafe(sql.clone())) + .await + .unwrap(); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!( + $crate::common::attr(&spans[0], "db.query.text"), + Some(opentelemetry::Value::String(sql.into())), + ); + }}; +} + /// `fetch_one` against the wrapped pool. Returns 1 row. #[macro_export] macro_rules! test_fetch_one_via_pool { @@ -1882,7 +1942,10 @@ macro_rules! test_prepare_via_pool { let tel = $crate::common::TestTelemetry::install(); let pool = $pool_factory; - let _stmt = (&pool).prepare("SELECT 1").await.unwrap(); + let _stmt = (&pool) + .prepare(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); let spans = tel.spans(); assert_eq!(spans.len(), 1); @@ -1890,13 +1953,13 @@ macro_rules! test_prepare_via_pool { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); pool.with_annotations($crate::common::test_annotations()) - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); pool.with_operation("SELECT", "users") - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -1914,7 +1977,10 @@ macro_rules! test_prepare_via_connection { let pool = $pool_factory; let mut conn = pool.acquire().await.unwrap(); - let _stmt = (&mut conn).prepare("SELECT 1").await.unwrap(); + let _stmt = (&mut conn) + .prepare(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); let spans = tel.spans(); assert_eq!(spans.len(), 1); @@ -1922,13 +1988,13 @@ macro_rules! test_prepare_via_connection { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); conn.with_annotations($crate::common::test_annotations()) - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); conn.with_operation("SELECT", "users") - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -1946,15 +2012,18 @@ macro_rules! test_prepare_via_transaction { let pool = $pool_factory; let mut tx = pool.begin().await.unwrap(); - let _stmt = (&mut tx).prepare("SELECT 1").await.unwrap(); + let _stmt = (&mut tx) + .prepare(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); tx.with_annotations($crate::common::test_annotations()) - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); tx.with_operation("SELECT", "users") - .prepare("SELECT 1") + .prepare(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); @@ -1980,7 +2049,9 @@ macro_rules! test_prepare_records_error { let pool = $pool_factory; let mut conn = pool.acquire().await.unwrap(); - let result = (&mut conn).prepare("INVALID SQL GIBBERISH").await; + let result = (&mut conn) + .prepare(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) + .await; assert!(result.is_err()); let spans = tel.spans(); @@ -1991,7 +2062,7 @@ macro_rules! test_prepare_records_error { let result = conn .with_annotations($crate::common::test_annotations()) - .prepare("INVALID SQL GIBBERISH") + .prepare(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -2000,7 +2071,7 @@ macro_rules! test_prepare_records_error { let result = conn .with_operation("SELECT", "users") - .prepare("INVALID SQL GIBBERISH") + .prepare(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -2020,7 +2091,10 @@ macro_rules! test_prepare_with_via_pool { let pool = $pool_factory; let _stmt = (&pool) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); @@ -2030,13 +2104,19 @@ macro_rules! test_prepare_with_via_pool { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); pool.with_annotations($crate::common::test_annotations()) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); pool.with_operation("SELECT", "users") - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -2055,7 +2135,10 @@ macro_rules! test_prepare_with_via_connection { let mut conn = pool.acquire().await.unwrap(); let _stmt = (&mut conn) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); @@ -2065,13 +2148,19 @@ macro_rules! test_prepare_with_via_connection { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); conn.with_annotations($crate::common::test_annotations()) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); conn.with_operation("SELECT", "users") - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -2090,17 +2179,26 @@ macro_rules! test_prepare_with_via_transaction { let mut tx = pool.begin().await.unwrap(); let _stmt = (&mut tx) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); tx.with_annotations($crate::common::test_annotations()) - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); tx.with_operation("SELECT", "users") - .prepare_with($dialect.prepare_with_select_sql, &[]) + .prepare_with( + sqlx::SqlStr::from_static($dialect.prepare_with_select_sql), + &[], + ) .await .unwrap(); @@ -2126,7 +2224,9 @@ macro_rules! test_prepare_with_records_error { let pool = $pool_factory; let mut conn = pool.acquire().await.unwrap(); - let result = (&mut conn).prepare_with("INVALID SQL GIBBERISH", &[]).await; + let result = (&mut conn) + .prepare_with(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH"), &[]) + .await; assert!(result.is_err()); let spans = tel.spans(); @@ -2137,7 +2237,7 @@ macro_rules! test_prepare_with_records_error { let result = conn .with_annotations($crate::common::test_annotations()) - .prepare_with("INVALID SQL GIBBERISH", &[]) + .prepare_with(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH"), &[]) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -2146,7 +2246,7 @@ macro_rules! test_prepare_with_records_error { let result = conn .with_operation("SELECT", "users") - .prepare_with("INVALID SQL GIBBERISH", &[]) + .prepare_with(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH"), &[]) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -2165,7 +2265,10 @@ macro_rules! test_describe_via_pool { let tel = $crate::common::TestTelemetry::install(); let pool = $pool_factory; - let _desc = (&pool).describe("SELECT 1").await.unwrap(); + let _desc = (&pool) + .describe(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); let spans = tel.spans(); assert_eq!(spans.len(), 1); @@ -2173,13 +2276,13 @@ macro_rules! test_describe_via_pool { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); pool.with_annotations($crate::common::test_annotations()) - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); pool.with_operation("SELECT", "users") - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -2197,7 +2300,10 @@ macro_rules! test_describe_via_connection { let pool = $pool_factory; let mut conn = pool.acquire().await.unwrap(); - let _desc = (&mut conn).describe("SELECT 1").await.unwrap(); + let _desc = (&mut conn) + .describe(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); let spans = tel.spans(); assert_eq!(spans.len(), 1); @@ -2205,13 +2311,13 @@ macro_rules! test_describe_via_connection { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); conn.with_annotations($crate::common::test_annotations()) - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); conn.with_operation("SELECT", "users") - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); @@ -2229,15 +2335,18 @@ macro_rules! test_describe_via_transaction { let pool = $pool_factory; let mut tx = pool.begin().await.unwrap(); - let _desc = (&mut tx).describe("SELECT 1").await.unwrap(); + let _desc = (&mut tx) + .describe(sqlx::SqlStr::from_static("SELECT 1")) + .await + .unwrap(); tx.with_annotations($crate::common::test_annotations()) - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); tx.with_operation("SELECT", "users") - .describe("SELECT 1") + .describe(sqlx::SqlStr::from_static("SELECT 1")) .await .unwrap(); @@ -2263,7 +2372,9 @@ macro_rules! test_describe_records_error { let pool = $pool_factory; let mut conn = pool.acquire().await.unwrap(); - let result = (&mut conn).describe("INVALID SQL GIBBERISH").await; + let result = (&mut conn) + .describe(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) + .await; assert!(result.is_err()); let spans = tel.spans(); @@ -2274,7 +2385,7 @@ macro_rules! test_describe_records_error { let result = conn .with_annotations($crate::common::test_annotations()) - .describe("INVALID SQL GIBBERISH") + .describe(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -2283,7 +2394,7 @@ macro_rules! test_describe_records_error { let result = conn .with_operation("SELECT", "users") - .describe("INVALID SQL GIBBERISH") + .describe(sqlx::SqlStr::from_static("INVALID SQL GIBBERISH")) .await; assert!(result.is_err()); let last = tel.spans().last().unwrap().clone(); @@ -3713,7 +3824,10 @@ macro_rules! test_transaction_rollback { let mut tx = pool.begin().await.unwrap(); let create_sql = format!("CREATE TABLE rollback_test (id {})", $dialect.id_pk_column); - (&mut tx).execute(create_sql.as_str()).await.unwrap(); + (&mut tx) + .execute(sqlx::AssertSqlSafe(create_sql)) + .await + .unwrap(); tx.rollback().await.unwrap(); let spans = tel.spans(); diff --git a/tests/mysql.rs b/tests/mysql.rs index ab8e133..3472abd 100644 --- a/tests/mysql.rs +++ b/tests/mysql.rs @@ -44,7 +44,7 @@ async fn shared_container() -> &'static SharedContainer { .with_env_var("MYSQL_ROOT_PASSWORD", "test") .with_env_var("MYSQL_DATABASE", "testdb") .with_env_var("MYSQL_ALLOW_EMPTY_PASSWORD", "no") - .with_startup_timeout(Duration::from_secs(120)) + .with_startup_timeout(Duration::from_mins(2)) .start() .await .expect("starting mysql container"); @@ -259,6 +259,18 @@ async fn fetch_one_via_pool() { test_fetch_one_via_pool!(test_pool().await, common::MYSQL_DIALECT); } +#[tokio::test] +#[serial] +async fn raw_sql_preserves_simple_protocol() { + test_raw_sql_preserves_simple_protocol!(test_pool().await, common::MYSQL_DIALECT); +} + +#[tokio::test] +#[serial] +async fn owned_sql_string_query_text() { + test_owned_sql_string_query_text!(test_pool().await, common::MYSQL_DIALECT); +} + #[tokio::test] #[serial] async fn fetch_one_via_connection() { diff --git a/tests/postgres.rs b/tests/postgres.rs index 0380ba3..d848008 100644 --- a/tests/postgres.rs +++ b/tests/postgres.rs @@ -44,7 +44,7 @@ async fn shared_container() -> &'static SharedContainer { .with_env_var("POSTGRES_USER", "postgres") .with_env_var("POSTGRES_DB", "testdb") .with_env_var("POSTGRES_HOST_AUTH_METHOD", "trust") - .with_startup_timeout(Duration::from_secs(60)) + .with_startup_timeout(Duration::from_mins(1)) .start() .await .expect("starting postgres container"); @@ -262,6 +262,18 @@ async fn fetch_one_via_pool() { test_fetch_one_via_pool!(test_pool().await, common::POSTGRES_DIALECT); } +#[tokio::test] +#[serial] +async fn raw_sql_preserves_simple_protocol() { + test_raw_sql_preserves_simple_protocol!(test_pool().await, common::POSTGRES_DIALECT); +} + +#[tokio::test] +#[serial] +async fn owned_sql_string_query_text() { + test_owned_sql_string_query_text!(test_pool().await, common::POSTGRES_DIALECT); +} + #[tokio::test] #[serial] async fn fetch_one_via_connection() { diff --git a/tests/sqlite.rs b/tests/sqlite.rs index 18c6d21..830a163 100644 --- a/tests/sqlite.rs +++ b/tests/sqlite.rs @@ -195,6 +195,18 @@ async fn fetch_one_via_pool() { test_fetch_one_via_pool!(test_pool().await, common::SQLITE_DIALECT); } +#[tokio::test] +#[serial] +async fn raw_sql_preserves_simple_protocol() { + test_raw_sql_preserves_simple_protocol!(test_pool().await, common::SQLITE_DIALECT); +} + +#[tokio::test] +#[serial] +async fn owned_sql_string_query_text() { + test_owned_sql_string_query_text!(test_pool().await, common::SQLITE_DIALECT); +} + #[tokio::test] #[serial] async fn fetch_one_via_connection() { From 5b6d19e84931ee54f2c6903adf2f55158cc81f2c Mon Sep 17 00:00:00 2001 From: Alex Torchenko Date: Wed, 9 Sep 2026 16:53:08 +0700 Subject: [PATCH 5/8] fix: instrument all pool acquisition paths --- Cargo.toml | 1 + README.md | 14 ++- src/connection.rs | 8 +- src/executor.rs | 4 +- src/lib.rs | 10 +- src/pool.rs | 49 ++++++++-- src/pool_executor.rs | 79 +++++++++++++++ src/transaction.rs | 9 +- tests/pool_metrics.rs | 218 ++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 376 insertions(+), 16 deletions(-) create mode 100644 src/pool_executor.rs diff --git a/Cargo.toml b/Cargo.toml index 990c021..5fdbea3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,7 @@ exclude = [ pedantic = "warn" [dependencies] +async-stream = "0.3" futures = "0.3" opentelemetry = "0.32" opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_experimental"] } diff --git a/README.md b/README.md index 6fa69b3..8ef1622 100644 --- a/README.md +++ b/README.md @@ -207,7 +207,19 @@ These mirror the bounded portion of the span attribute set: connection-level att | `db.client.connection.idle.max` | Gauge | | Maximum idle connections (equals `max` in SQLx) | | `db.client.connection.idle.min` | Gauge | | Configured minimum connections | -The first four are recorded inline on every `acquire()` / connection drop – no sampling gaps. `db.client.connection.count` is polled by a background task and requires both a runtime feature (`runtime-tokio` or `runtime-async-std`) and a pool name set via `PoolBuilder::with_pool_name`. The remaining three are static gauges recorded once at pool construction. +Explicit `acquire()`, implicit pool queries (including annotated and streaming queries), +and `begin()` share one instrumented acquisition. Queries within a transaction do not +acquire again. Wait time covers completed acquisition attempts, including semaphore wait, +connection creation and validation, but excludes query execution and `BEGIN`. Dropping a +waiting future decrements pending requests without recording a completed wait or a pool +timeout. Only `sqlx::Error::PoolTimedOut` increments the timeout counter. Pending and +timeout series are initialized at zero; wait buckets resolve submillisecond acquisitions. +Connection-use time follows the lease through transaction commit, rollback, or drop. + +`db.client.connection.count` is polled by a background task and requires both a runtime +feature (`runtime-tokio` or `runtime-async-std`) and a pool name set via +`PoolBuilder::with_pool_name`. Polling and export intervals can miss brief utilization or +pending-request peaks. The remaining three are static gauges recorded once at construction. ## Backend support diff --git a/src/connection.rs b/src/connection.rs index 161d056..7b53e1c 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -34,12 +34,18 @@ use crate::pool::SharedState; pub struct PoolConnection { pub(crate) inner: sqlx::pool::PoolConnection, pub(crate) state: SharedState, + pub(crate) usage: ConnectionUsage, +} + +/// Moves with an owned connection into a transaction, preserving the complete lease duration. +#[derive(Debug)] +pub(crate) struct ConnectionUsage { pub(crate) use_time: Arc>, pub(crate) acquired_at: Instant, pub(crate) base_attrs: Vec, } -impl Drop for PoolConnection { +impl Drop for ConnectionUsage { fn drop(&mut self) { self.use_time .record(self.acquired_at.elapsed().as_secs_f64(), &self.base_attrs); diff --git a/src/executor.rs b/src/executor.rs index ad20356..9fbc235 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -687,14 +687,14 @@ macro_rules! impl_executor { // Executor impls for each wrapper type // --------------------------------------------------------------------------- -impl_executor!(&'_ crate::Pool, self => &self.inner); +impl_executor!(&'_ crate::Pool, self => crate::pool_executor::PoolExecutor(self.clone())); impl_executor!(&'c mut crate::PoolConnection, self => self.inner.as_mut()); impl_executor!(&'c mut crate::Transaction<'_, DB>, self => &mut *self.inner); // Annotated wrappers – same instrumentation with per-query annotations threaded through. impl_executor!( crate::annotations::Annotated<'c, crate::Pool>, - self => &self.inner.inner, + self => crate::pool_executor::PoolExecutor(self.inner.clone()), annotations: Some(&self.annotations) ); impl_executor!( diff --git a/src/lib.rs b/src/lib.rs index 0141313..d4c950c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -111,8 +111,13 @@ //! | `db.client.connection.idle.max` | Gauge | | Maximum idle connections (equals `max` in `SQLx`) | //! | `db.client.connection.idle.min` | Gauge | | Configured minimum connections | //! -//! The first four are recorded inline on every `acquire()` and connection drop – no sampling -//! gaps. `connection.count` is polled by a background task and requires both +//! Explicit acquisition, implicit pool queries (including annotated and streaming queries), +//! and `begin()` share one instrumented acquisition. Queries within a transaction do not +//! acquire again. Wait time covers completed acquisition attempts and excludes `BEGIN` and +//! query execution. Caller cancellation clears pending without recording a completed wait +//! or a pool timeout. Connection use time follows the lease through transaction completion. +//! Pending and timeout series start at zero. Polling/export can miss brief gauge peaks. +//! `connection.count` is polled by a background task and requires both //! [`PoolBuilder::with_pool_name`] and a runtime feature (`runtime-tokio` or //! `runtime-async-std`); without either, the gauge is silent. The remaining three are static //! gauges recorded once at [`PoolBuilder::build`]. @@ -234,6 +239,7 @@ mod executor; mod metrics; mod obfuscate; mod pool; +mod pool_executor; mod pool_metrics; mod query_ext; mod rebuilt_query; diff --git a/src/pool.rs b/src/pool.rs index 1c6fa17..e297cbb 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -224,7 +224,7 @@ impl PoolBuilder { .build() .record(min_conns, &base_attrs); - Pool { + let pool = Pool { inner: self.pool, state: SharedState { attrs, metrics }, metrics_shutdown, @@ -235,6 +235,10 @@ impl PoolBuilder { .with_description( "The time it took to obtain an open connection from the pool.", ) + .with_boundaries(vec![ + 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, + 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, + ]) .build(), ), use_time: Arc::new( @@ -260,7 +264,10 @@ impl PoolBuilder { .with_description("The number of pending requests for an open connection.") .build(), ), - } + }; + pool.pending_requests.add(0, &base_attrs); + pool.timeouts.add(0, &base_attrs); + pool } /// Spawn the pool metrics background task if a pool name is set and a runtime is @@ -382,9 +389,13 @@ impl Pool { pub async fn acquire(&self) -> Result, sqlx::Error> { let attrs = self.state.attrs.base_key_values(); self.pending_requests.add(1, &attrs); + let pending = PendingAcquisition { + counter: &self.pending_requests, + attrs: &attrs, + }; let start = std::time::Instant::now(); let result = self.inner.acquire().await; - self.pending_requests.add(-1, &attrs); + drop(pending); self.wait_time.record(start.elapsed().as_secs_f64(), &attrs); if let Err(sqlx::Error::PoolTimedOut) = &result { @@ -394,9 +405,11 @@ impl Pool { result.map(|inner| PoolConnection { inner, state: self.state.clone(), - use_time: self.use_time.clone(), - acquired_at: std::time::Instant::now(), - base_attrs: attrs, + usage: crate::connection::ConnectionUsage { + use_time: self.use_time.clone(), + acquired_at: std::time::Instant::now(), + base_attrs: attrs, + }, }) } @@ -413,9 +426,16 @@ impl Pool { /// Returns `sqlx::Error` if `BEGIN` fails – typically due to a connection problem or /// because the underlying connection cannot start a new transaction. pub async fn begin(&self) -> Result, sqlx::Error> { - self.inner.begin().await.map(|inner| Transaction { + let PoolConnection { inner, - state: self.state.clone(), + state, + usage, + } = self.acquire().await?; + let inner = sqlx::Transaction::begin(inner, None).await?; + Ok(Transaction { + inner, + state, + usage, }) } @@ -484,3 +504,16 @@ impl Pool { ) } } + +// A caller timeout or task abort may drop acquire() while it is still awaiting SQLx. +// Keep pending truthful on that path without classifying caller cancellation as PoolTimedOut. +struct PendingAcquisition<'a> { + counter: &'a opentelemetry::metrics::UpDownCounter, + attrs: &'a [opentelemetry::KeyValue], +} + +impl Drop for PendingAcquisition<'_> { + fn drop(&mut self) { + self.counter.add(-1, self.attrs); + } +} diff --git a/src/pool_executor.rs b/src/pool_executor.rs new file mode 100644 index 0000000..5ea8d16 --- /dev/null +++ b/src/pool_executor.rs @@ -0,0 +1,79 @@ +//! Acquire through the instrumented pool, then execute on the raw connection. +//! Query spans/metrics remain owned by executor.rs, so each query is recorded once. + +use crate::{Database, Pool}; +use futures::{TryStreamExt, future::BoxFuture, stream::BoxStream}; +use sqlx::{Describe, Either, Error, Execute, Executor}; + +#[derive(Debug)] +pub(crate) struct PoolExecutor(pub Pool); + +impl<'c, DB> Executor<'c> for PoolExecutor +where + DB: Database, + for<'a> &'a mut DB::Connection: Executor<'a, Database = DB>, +{ + type Database = DB; + + fn fetch_many<'e, 'q: 'e, E>( + self, + query: E, + ) -> BoxStream<'e, Result, Error>> + where + E: 'q + Execute<'q, DB>, + 'c: 'e, + { + Box::pin(async_stream::try_stream! { + let mut connection = self.0.acquire().await?; + let mut rows = connection.inner.as_mut().fetch_many(query); + while let Some(row) = rows.try_next().await? { + yield row; + } + }) + } + + fn fetch_optional<'e, 'q: 'e, E>( + self, + query: E, + ) -> BoxFuture<'e, Result, Error>> + where + E: 'q + Execute<'q, DB>, + 'c: 'e, + { + Box::pin(async move { + self.0 + .acquire() + .await? + .inner + .as_mut() + .fetch_optional(query) + .await + }) + } + + fn prepare_with<'e>( + self, + sql: sqlx::SqlStr, + parameters: &'e [DB::TypeInfo], + ) -> BoxFuture<'e, Result> + where + 'c: 'e, + { + Box::pin(async move { + self.0 + .acquire() + .await? + .inner + .as_mut() + .prepare_with(sql, parameters) + .await + }) + } + + fn describe<'e>(self, sql: sqlx::SqlStr) -> BoxFuture<'e, Result, Error>> + where + 'c: 'e, + { + Box::pin(async move { self.0.acquire().await?.inner.as_mut().describe(sql).await }) + } +} diff --git a/src/transaction.rs b/src/transaction.rs index 104fd1a..9fefbd9 100644 --- a/src/transaction.rs +++ b/src/transaction.rs @@ -33,6 +33,7 @@ use crate::pool::SharedState; pub struct Transaction<'c, DB: sqlx::Database> { pub(crate) inner: sqlx::Transaction<'c, DB>, pub(crate) state: SharedState, + pub(crate) usage: crate::connection::ConnectionUsage, } impl Transaction<'_, DB> @@ -45,7 +46,9 @@ where /// /// Returns `sqlx::Error` if the commit fails. pub async fn commit(self) -> Result<(), sqlx::Error> { - self.inner.commit().await + let result = self.inner.commit().await; + drop(self.usage); + result } /// Roll back the transaction. @@ -54,7 +57,9 @@ where /// /// Returns `sqlx::Error` if the rollback fails. pub async fn rollback(self) -> Result<(), sqlx::Error> { - self.inner.rollback().await + let result = self.inner.rollback().await; + drop(self.usage); + result } impl_with_annotations_mut!(); diff --git a/tests/pool_metrics.rs b/tests/pool_metrics.rs index e91003b..cfc149f 100644 --- a/tests/pool_metrics.rs +++ b/tests/pool_metrics.rs @@ -11,6 +11,224 @@ use std::time::Duration; const POOL_NAME: &str = "test-pool"; +fn histogram_count(tel: &common::TestTelemetry, name: &str) -> u64 { + tel.reset(); + let metrics = tel.metrics(); + let Some(metric) = find_metric(&metrics, name) else { + return 0; + }; + let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() else { + panic!("expected histogram") + }; + hist.data_points() + .map(opentelemetry_sdk::metrics::data::HistogramDataPoint::count) + .sum() +} + +fn pending_count(tel: &common::TestTelemetry) -> i64 { + tel.reset(); + let metrics = tel.metrics(); + let metric = find_metric(&metrics, "db.client.connection.pending_requests").unwrap(); + let AggregatedMetrics::I64(MetricData::Sum(sum)) = metric.data() else { + panic!("expected pending sum") + }; + sum.data_points() + .map(opentelemetry_sdk::metrics::data::SumDataPoint::value) + .sum() +} + +fn timeout_count(tel: &common::TestTelemetry) -> u64 { + tel.reset(); + let metrics = tel.metrics(); + let metric = find_metric(&metrics, "db.client.connection.timeouts").unwrap(); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("expected timeout sum") + }; + sum.data_points() + .map(opentelemetry_sdk::metrics::data::SumDataPoint::value) + .sum() +} + +#[tokio::test] +#[serial] +async fn every_pool_executor_path_acquires_once_without_duplicate_query_spans() { + use futures::TryStreamExt; + use sqlx_otel::QueryAnnotateExt; + let tel = common::TestTelemetry::install(); + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + pool.execute("CREATE TABLE test (id integer)") + .await + .unwrap(); + pool.execute("INSERT INTO test VALUES (1)").await.unwrap(); + pool.fetch_all("SELECT id FROM test").await.unwrap(); + pool.fetch_one("SELECT id FROM test").await.unwrap(); + pool.fetch_optional("SELECT id FROM test").await.unwrap(); + pool.fetch("SELECT id FROM test") + .try_collect::>() + .await + .unwrap(); + #[allow(deprecated)] + pool.fetch_many("SELECT id FROM test") + .try_collect::>() + .await + .unwrap(); + pool.execute_many("UPDATE test SET id = 2") + .try_collect::>() + .await + .unwrap(); + pool.prepare_with(sqlx::SqlStr::from_static("SELECT id FROM test"), &[]) + .await + .unwrap(); + pool.describe(sqlx::SqlStr::from_static("SELECT id FROM test")) + .await + .unwrap(); + pool.with_operation("SELECT", "test") + .fetch_one("SELECT id FROM test") + .await + .unwrap(); + sqlx::query("SELECT id FROM test") + .with_operation("SELECT", "test") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + tel.spans().len(), + 12, + "query instrumentation must not be nested" + ); + assert_eq!(histogram_count(&tel, "db.client.connection.wait_time"), 12); + assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 12); + assert_eq!(pending_count(&tel), 0); + assert_eq!(timeout_count(&tel), 0); +} + +#[tokio::test] +#[serial] +async fn transactions_keep_one_acquisition_and_measure_the_complete_lease() { + let tel = common::TestTelemetry::install(); + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + for finish in 0..3 { + let mut transaction = pool.begin().await.unwrap(); + sqlx::query("SELECT 1") + .execute(&mut transaction) + .await + .unwrap(); + sqlx::query("SELECT 2") + .execute(&mut transaction) + .await + .unwrap(); + assert_eq!( + histogram_count(&tel, "db.client.connection.wait_time"), + finish + 1 + ); + assert_eq!( + histogram_count(&tel, "db.client.connection.use_time"), + finish + ); + match finish { + 0 => transaction.commit().await.unwrap(), + 1 => transaction.rollback().await.unwrap(), + _ => drop(transaction), + } + assert_eq!( + histogram_count(&tel, "db.client.connection.use_time"), + finish + 1 + ); + } + assert_eq!(pending_count(&tel), 0); +} + +async fn waiting_operation( + pool: &sqlx_otel::Pool, + path: u8, +) -> Result<(), sqlx::Error> { + use futures::TryStreamExt; + match path { + 0 => { + drop(pool.acquire().await?); + } + 1 => { + pool.fetch_optional("SELECT 1").await?; + } + 2 => { + drop(pool.begin().await?); + } + 3 => { + pool.fetch("SELECT 1").try_next().await?; + } + _ => { + pool.with_operation("SELECT", "test") + .fetch_optional("SELECT 1") + .await?; + } + } + Ok(()) +} + +#[tokio::test] +#[serial] +async fn all_acquisition_paths_clear_pending_on_cancel_and_report_real_timeouts() { + let tel = common::TestTelemetry::install(); + let raw = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_millis(100)) + .connect(":memory:") + .await + .unwrap(); + let pool = PoolBuilder::from(raw).build(); + let held = pool.acquire().await.unwrap(); + for path in 0..5 { + let mut cancelled = Box::pin(waiting_operation(&pool, path)); + assert!(futures::poll!(&mut cancelled).is_pending()); + assert_eq!(pending_count(&tel), 1); + drop(cancelled); + assert_eq!( + pending_count(&tel), + 0, + "cancelled acquisition leaked pending" + ); + assert_eq!( + timeout_count(&tel), + u64::from(path), + "caller cancellation is not a pool timeout" + ); + assert!(matches!( + waiting_operation(&pool, path).await, + Err(sqlx::Error::PoolTimedOut) + )); + assert_eq!(pending_count(&tel), 0); + assert_eq!(timeout_count(&tel), u64::from(path) + 1); + assert_eq!( + histogram_count(&tel, "db.client.connection.wait_time"), + u64::from(path) + 2 + ); + } + drop(held); +} + +#[tokio::test] +#[serial] +async fn dropping_a_partially_consumed_stream_releases_its_only_connection() { + use futures::TryStreamExt; + let tel = common::TestTelemetry::install(); + let raw = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect(":memory:") + .await + .unwrap(); + let pool = PoolBuilder::from(raw).build(); + let mut rows = pool.fetch("SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3"); + rows.try_next().await.unwrap().unwrap(); + assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 0); + drop(rows); + assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 1); + tokio::time::timeout(Duration::from_secs(1), pool.acquire()) + .await + .unwrap() + .unwrap(); + assert_eq!(pending_count(&tel), 0); +} + /// Helper to find a named metric in the collected resource metrics. fn find_metric<'a>( resource_metrics: &'a [opentelemetry_sdk::metrics::data::ResourceMetrics], From 6f1bbe31490a27df96e29566d723f000e6ad0699 Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Mon, 21 Sep 2026 12:17:47 +0100 Subject: [PATCH 6/8] fix: Apply seconds-scale buckets to every latency histogram MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `db.client.operation.duration` and `db.client.connection.use_time` inherited the OpenTelemetry SDK's default bucket boundaries (`0, 5, 10, 25, ... 10000`), which are shaped for milliseconds. Semconv requires both be recorded in seconds, so every observation below five seconds collapsed into a single bucket and any quantile derived from it was an interpolation artefact rather than a measurement. `wait_time` already carried explicit boundaries; hoist that set into a shared `LATENCY_BUCKETS_SECONDS` constant and apply it to all three. The row-count histograms keep the SDK defaults, which already suit counts. The boundaries are instrument advice, so an application that registers a `View` with a compatible explicit aggregation still overrides them. Document that in the README with a worked example, and pin it with a test that observes the same workload with and without a view — asserting the crate's advice applies when no view matches, so the override is demonstrably an override rather than a value filling a vacuum. Also close gaps left by the acquisition-path work: record why `PoolExecutor` owns its pool rather than borrowing, and why it implements only four `Executor` methods — sqlx's defaults funnel into those, which is what holds acquisition to once per query, and an added override would double every acquisition metric with no compile error. Cover the error-after-acquisition, closed-pool, multi-statement-stream, and `prepare` paths. --- README.md | 48 ++++-- src/lib.rs | 33 +++- src/metrics.rs | 71 ++++++++ src/pool.rs | 16 +- src/pool_executor.rs | 34 ++++ tests/pool_metrics.rs | 369 +++++++++++++++++++++++++++++++++++++++++- 6 files changed, 542 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index 8ef1622..4d2182f 100644 --- a/README.md +++ b/README.md @@ -207,19 +207,41 @@ These mirror the bounded portion of the span attribute set: connection-level att | `db.client.connection.idle.max` | Gauge | | Maximum idle connections (equals `max` in SQLx) | | `db.client.connection.idle.min` | Gauge | | Configured minimum connections | -Explicit `acquire()`, implicit pool queries (including annotated and streaming queries), -and `begin()` share one instrumented acquisition. Queries within a transaction do not -acquire again. Wait time covers completed acquisition attempts, including semaphore wait, -connection creation and validation, but excludes query execution and `BEGIN`. Dropping a -waiting future decrements pending requests without recording a completed wait or a pool -timeout. Only `sqlx::Error::PoolTimedOut` increments the timeout counter. Pending and -timeout series are initialized at zero; wait buckets resolve submillisecond acquisitions. -Connection-use time follows the lease through transaction commit, rollback, or drop. - -`db.client.connection.count` is polled by a background task and requires both a runtime -feature (`runtime-tokio` or `runtime-async-std`) and a pool name set via -`PoolBuilder::with_pool_name`. Polling and export intervals can miss brief utilization or -pending-request peaks. The remaining three are static gauges recorded once at construction. +Explicit `acquire()`, implicit pool queries (including annotated and streaming queries), and `begin()` share one instrumented acquisition; queries issued inside a transaction do not acquire again. Wait time covers completed acquisition attempts – semaphore wait, connection creation, and validation – but excludes `BEGIN` and query execution. Dropping a waiting future decrements pending requests without recording a completed wait or a timeout, since the pool never failed to deliver: the caller left. Only `sqlx::Error::PoolTimedOut` increments the timeout counter, so `PoolClosed` on shutdown does not inflate your timeout rate. Pending-request and timeout series are initialised at zero so dashboards start from a known baseline rather than waiting for the first event. Connection use time follows the lease through transaction commit, rollback, or drop. + +`db.client.connection.count` is polled by a background task and requires both a runtime feature (`runtime-tokio` or `runtime-async-std`) and a pool name set via `PoolBuilder::with_pool_name`. Polling and export intervals can miss brief utilisation or pending-request peaks. The remaining three are static gauges recorded once at construction. + +### Histogram buckets + +The three latency histograms – `db.client.operation.duration`, `db.client.connection.wait_time`, and `db.client.connection.use_time` – are recorded in seconds, as the semantic conventions require. The OpenTelemetry SDK's default bucket boundaries (`0, 5, 10, 25, … 10000`) are shaped for milliseconds, so on a seconds-valued histogram every observation below five seconds falls into a single bucket and any quantile derived from it is an interpolation artefact rather than a measurement. The crate therefore ships explicit boundaries spanning the range a database client actually occupies, from sub-millisecond local round trips up to the 30-second acquire timeout SQLx applies by default: + +```text +0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, +0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0 +``` + +These are instrument *advice*, not a fixed configuration. The SDK applies advice only when no matching view has already set an aggregation, so an application that registers a `View` with its own explicit bucket boundaries overrides what the crate ships – no API is needed from this crate to retune them. (A view whose aggregation the SDK rejects as incompatible with the instrument is discarded, and the crate's advice applies after all, so match histogram names only.) + +```rust,ignore +use opentelemetry_sdk::metrics::{Aggregation, Instrument, SdkMeterProvider, Stream}; + +let provider = SdkMeterProvider::builder() + .with_reader(reader) + .with_view(|i: &Instrument| { + (i.name() == "db.client.operation.duration").then(|| { + Stream::builder() + .with_aggregation(Aggregation::ExplicitBucketHistogram { + boundaries: vec![0.005, 0.05, 0.5, 5.0], + record_min_max: true, + }) + .build() + .unwrap() + }) + }) + .build(); +``` + +The two row-count histograms (`db.client.response.returned_rows` and `db.client.response.affected_rows`) are left on the SDK defaults, which are already reasonable for counts. ## Backend support diff --git a/src/lib.rs b/src/lib.rs index d4c950c..d9a0f53 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -111,17 +111,36 @@ //! | `db.client.connection.idle.max` | Gauge | | Maximum idle connections (equals `max` in `SQLx`) | //! | `db.client.connection.idle.min` | Gauge | | Configured minimum connections | //! -//! Explicit acquisition, implicit pool queries (including annotated and streaming queries), -//! and `begin()` share one instrumented acquisition. Queries within a transaction do not -//! acquire again. Wait time covers completed acquisition attempts and excludes `BEGIN` and -//! query execution. Caller cancellation clears pending without recording a completed wait -//! or a pool timeout. Connection use time follows the lease through transaction completion. -//! Pending and timeout series start at zero. Polling/export can miss brief gauge peaks. -//! `connection.count` is polled by a background task and requires both +//! Explicit acquisition, implicit pool queries (including annotated and streaming queries), and +//! `begin()` share one instrumented acquisition; queries inside a transaction do not acquire +//! again. Wait time covers completed acquisition attempts and excludes `BEGIN` and query +//! execution. Caller cancellation clears pending without recording a completed wait or a pool +//! timeout, since the pool never failed to deliver – the caller left. Only +//! `sqlx::Error::PoolTimedOut` increments the timeout counter, so `PoolClosed` on shutdown does +//! not inflate the timeout rate. Connection use time follows the lease through transaction +//! completion. Pending and timeout series start at zero. Polling/export can miss brief gauge +//! peaks. `connection.count` is polled by a background task and requires both //! [`PoolBuilder::with_pool_name`] and a runtime feature (`runtime-tokio` or //! `runtime-async-std`); without either, the gauge is silent. The remaining three are static //! gauges recorded once at [`PoolBuilder::build`]. //! +//! ## Histogram buckets +//! +//! `db.client.operation.duration`, `db.client.connection.wait_time`, and +//! `db.client.connection.use_time` are recorded in seconds, as semconv requires. The `OTel` SDK's +//! default bucket boundaries (`0, 5, 10, 25, ... 10000`) are shaped for milliseconds, so on a +//! seconds-valued histogram every observation below five seconds collapses into one bucket and +//! quantiles derived from it are interpolation artefacts rather than measurements. All three +//! therefore ship explicit boundaries spanning sub-millisecond round trips up to the 30-second +//! acquire timeout `SQLx` applies by default. The two row-count histograms are left on the SDK +//! defaults, which suit counts already. +//! +//! These boundaries are instrument *advice*. The SDK applies advice only where no matching view +//! has already set an aggregation, so an application that registers a `View` with a compatible +//! explicit aggregation overrides them – this crate deliberately exposes no API of its own for +//! retuning them. A view whose aggregation the SDK rejects as incompatible falls back to the +//! implicit default view, which applies this advice after all. See the README for an example. +//! //! ## A note on `db.response.affected_rows` //! //! `db.response.affected_rows` is **not part of the OpenTelemetry semantic conventions** at diff --git a/src/metrics.rs b/src/metrics.rs index ba038ba..d6ed1a5 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -4,6 +4,25 @@ use opentelemetry::KeyValue; use opentelemetry::metrics::{Histogram, Meter}; use opentelemetry_semantic_conventions::metric; +/// Explicit bucket boundaries, in seconds, for every latency histogram the crate records. +/// +/// The `OTel` SDK's default boundaries (`0, 5, 10, 25, ... 10000`) are shaped for milliseconds, +/// but semconv requires these histograms be recorded in seconds. Left on the defaults, every +/// observation below five seconds falls into a single bucket, and quantiles computed from it are +/// interpolation artefacts rather than measurements. These boundaries span the range a database +/// client actually occupies: sub-millisecond local round trips up to the 30-second acquire +/// timeout `SQLx` applies by default. +/// +/// This is instrument *advice*, not a hard configuration. An application that registers a +/// `View` with its own explicit aggregation overrides these boundaries, because the SDK only +/// applies advice when no matching view has already set an aggregation. The exception is a view +/// whose aggregation is incompatible with the instrument: the SDK rejects it, falls back to the +/// implicit default view, and applies this advice after all. +pub(crate) const LATENCY_BUCKETS_SECONDS: &[f64] = &[ + 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, + 30.0, 60.0, +]; + /// Holds the OpenTelemetry metric instruments for per-operation recording. /// /// Created once per [`Pool`](crate::Pool) and shared (via `Arc`) across all wrapper types @@ -28,6 +47,7 @@ impl Metrics { .f64_histogram(metric::DB_CLIENT_OPERATION_DURATION) .with_unit("s") .with_description("Duration of database client operations.") + .with_boundaries(LATENCY_BUCKETS_SECONDS.to_vec()) .build(); let returned_rows = meter .f64_histogram(metric::DB_CLIENT_RESPONSE_RETURNED_ROWS) @@ -68,3 +88,54 @@ impl Metrics { } } } + +#[cfg(test)] +mod tests { + use super::LATENCY_BUCKETS_SECONDS; + + /// The boundary set must be well-formed for an explicit-bucket histogram. + /// + /// The `OTel` SDK requires strictly ascending, finite boundaries; a duplicate or out-of-order + /// entry produces an empty or misattributed bucket rather than an error at build time. This + /// reports a mistyped edit here, rather than as a vector diff in an integration test. + #[test] + fn latency_buckets_are_strictly_ascending_and_positive() { + assert!( + !LATENCY_BUCKETS_SECONDS.is_empty(), + "boundary set must not be empty" + ); + // Evaluate the comparison for every pair up front rather than inside an `assert!` + // message, so the check itself runs on the passing path instead of only on failure. + let ascending = LATENCY_BUCKETS_SECONDS + .windows(2) + .all(|pair| pair[0] < pair[1]); + assert!( + ascending, + "boundaries must strictly ascend: {LATENCY_BUCKETS_SECONDS:?}" + ); + for &bound in LATENCY_BUCKETS_SECONDS { + assert!(bound.is_finite(), "boundary {bound} is not finite"); + assert!(bound > 0.0, "boundary {bound} must be positive"); + } + } + + /// The set must cover the range a database client operates in. + /// + /// The lower bound has to resolve sub-millisecond local round trips, which is why the crate + /// ships boundaries at all. The upper bound has to reach past `SQLx`'s 30-second default + /// acquire timeout, or every timed-out acquisition lands in the overflow bucket and the p99 + /// of `wait_time` stops being readable when it matters most. + #[test] + fn latency_buckets_span_submillisecond_to_beyond_the_default_acquire_timeout() { + let first = LATENCY_BUCKETS_SECONDS[0]; + let last = LATENCY_BUCKETS_SECONDS[LATENCY_BUCKETS_SECONDS.len() - 1]; + assert!( + first <= 0.001, + "lowest boundary {first} cannot resolve submillisecond acquisition" + ); + assert!( + last >= 30.0, + "highest boundary {last} does not reach the default acquire timeout" + ); + } +} diff --git a/src/pool.rs b/src/pool.rs index e297cbb..e5b2f9d 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -235,10 +235,7 @@ impl PoolBuilder { .with_description( "The time it took to obtain an open connection from the pool.", ) - .with_boundaries(vec![ - 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, - 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, - ]) + .with_boundaries(crate::metrics::LATENCY_BUCKETS_SECONDS.to_vec()) .build(), ), use_time: Arc::new( @@ -248,6 +245,7 @@ impl PoolBuilder { .with_description( "The time between borrowing a connection and returning it to the pool.", ) + .with_boundaries(crate::metrics::LATENCY_BUCKETS_SECONDS.to_vec()) .build(), ), timeouts: Arc::new( @@ -505,8 +503,14 @@ impl Pool { } } -// A caller timeout or task abort may drop acquire() while it is still awaiting SQLx. -// Keep pending truthful on that path without classifying caller cancellation as PoolTimedOut. +/// Decrements `db.client.connection.pending_requests` on every exit path from +/// [`Pool::acquire`]. +/// +/// A caller timeout or task abort can drop the `acquire()` future while it is still awaiting +/// `SQLx`, which would otherwise leave the counter permanently incremented. Tying the decrement +/// to a `Drop` guard covers success, error, and cancellation alike. Cancellation deliberately +/// stops there: it records no completed wait and is not classified as `PoolTimedOut`, because +/// the pool never failed to deliver – the caller left. struct PendingAcquisition<'a> { counter: &'a opentelemetry::metrics::UpDownCounter, attrs: &'a [opentelemetry::KeyValue], diff --git a/src/pool_executor.rs b/src/pool_executor.rs index 5ea8d16..bf06adc 100644 --- a/src/pool_executor.rs +++ b/src/pool_executor.rs @@ -5,6 +5,19 @@ use crate::{Database, Pool}; use futures::{TryStreamExt, future::BoxFuture, stream::BoxStream}; use sqlx::{Describe, Either, Error, Execute, Executor}; +/// Routes an implicit pool query through [`Pool::acquire`] instead of the raw `SQLx` pool. +/// +/// Queries written against a pool (`query(..).fetch_one(&pool)`) never call `acquire()` +/// themselves, so without this indirection they would borrow a connection through +/// `sqlx::Pool` directly and contribute nothing to `db.client.connection.wait_time`, +/// `use_time`, `pending_requests`, or `timeouts`. This wrapper restores that accounting +/// while leaving query-level spans and metrics to `executor.rs`. +/// +/// Holds an owned [`Pool`] rather than a borrow. The `Executor` impls for `&Pool` and +/// `Annotated<'_, Pool>` are generic over a lifetime unrelated to the pool reference, so +/// borrowing here would tie returned streams and futures to the pool's lifetime and break +/// callers that keep them alive independently. `sqlx::Pool`'s own executor clones for the same +/// reason; it costs a few `Arc` bumps per query. #[derive(Debug)] pub(crate) struct PoolExecutor(pub Pool); @@ -15,6 +28,20 @@ where { type Database = DB; + // Only `fetch_many`, `fetch_optional`, `prepare_with` and `describe` are implemented here, + // on purpose. `SQLx`'s default `execute`, `execute_many`, `fetch`, `fetch_all`, `fetch_one` + // and `prepare` are all defined in terms of these four, so every entry point ends up at + // exactly one `acquire()`. Adding an override that acquires again would double every + // acquisition metric for that path, with no compile error and no obvious test failure. + // `every_pool_executor_path_acquires_once_without_duplicate_query_spans` in + // `tests/pool_metrics.rs` guards this by asserting one wait-time observation per query + // across every one of those entry points. + + /// Acquire, then stream rows from the borrowed connection. + /// + /// The connection is acquired on first poll, not when the stream is created, and is held + /// until the stream completes or is dropped – so `use_time` covers the full lease, and a + /// partially consumed stream still returns its connection to the pool. fn fetch_many<'e, 'q: 'e, E>( self, query: E, @@ -32,6 +59,7 @@ where }) } + /// Acquire, run the query, and release the connection before returning. fn fetch_optional<'e, 'q: 'e, E>( self, query: E, @@ -51,6 +79,11 @@ where }) } + /// Acquire, prepare the statement, and release the connection before returning. + /// + /// The returned statement is owned, so it outlives the connection that produced it. Any + /// driver-side statement cache entry stays with that connection, exactly as it does when + /// preparing through `sqlx::Pool`. fn prepare_with<'e>( self, sql: sqlx::SqlStr, @@ -70,6 +103,7 @@ where }) } + /// Acquire, describe the query, and release the connection before returning. fn describe<'e>(self, sql: sqlx::SqlStr) -> BoxFuture<'e, Result, Error>> where 'c: 'e, diff --git a/tests/pool_metrics.rs b/tests/pool_metrics.rs index cfc149f..663ef19 100644 --- a/tests/pool_metrics.rs +++ b/tests/pool_metrics.rs @@ -11,6 +11,15 @@ use std::time::Duration; const POOL_NAME: &str = "test-pool"; +/// The boundaries the crate ships on every latency histogram. +/// +/// Hardcoded on purpose: importing `LATENCY_BUCKETS_SECONDS` would make the assertion +/// tautological, and it would pass through any edit to the constant. +const EXPECTED_LATENCY_BUCKETS: &[f64] = &[ + 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, + 30.0, 60.0, +]; + fn histogram_count(tel: &common::TestTelemetry, name: &str) -> u64 { tel.reset(); let metrics = tel.metrics(); @@ -82,6 +91,10 @@ async fn every_pool_executor_path_acquires_once_without_duplicate_query_spans() pool.describe(sqlx::SqlStr::from_static("SELECT id FROM test")) .await .unwrap(); + // `prepare` defaults into `prepare_with`, so it is a funnel path of its own. + pool.prepare(sqlx::SqlStr::from_static("SELECT id FROM test")) + .await + .unwrap(); pool.with_operation("SELECT", "test") .fetch_one("SELECT id FROM test") .await @@ -93,11 +106,11 @@ async fn every_pool_executor_path_acquires_once_without_duplicate_query_spans() .unwrap(); assert_eq!( tel.spans().len(), - 12, + 13, "query instrumentation must not be nested" ); - assert_eq!(histogram_count(&tel, "db.client.connection.wait_time"), 12); - assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 12); + assert_eq!(histogram_count(&tel, "db.client.connection.wait_time"), 13); + assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 13); assert_eq!(pending_count(&tel), 0); assert_eq!(timeout_count(&tel), 0); } @@ -229,6 +242,356 @@ async fn dropping_a_partially_consumed_stream_releases_its_only_connection() { assert_eq!(pending_count(&tel), 0); } +/// Resolve a latency histogram's bucket boundaries as the SDK actually configured them. +fn histogram_bounds(tel: &common::TestTelemetry, name: &str) -> Vec { + tel.reset(); + let metrics = tel.metrics(); + let metric = find_metric(&metrics, name).expect("histogram not recorded"); + let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() else { + panic!("expected histogram") + }; + hist.data_points() + .next() + .expect("no data point") + .bounds() + .collect() +} + +/// Every latency histogram must carry the shared seconds-scale boundaries. +/// +/// The `OTel` SDK default boundaries (`0, 5, 10, 25, ... 10000`) are millisecond-shaped, so on a +/// seconds-valued histogram every observation below five seconds collapses into one bucket and +/// quantiles become interpolation artefacts. This pins all three latency histograms to the +/// shared set; row-count histograms are deliberately left on the defaults. +#[tokio::test] +#[serial] +async fn latency_histograms_use_seconds_scale_buckets() { + let tel = common::TestTelemetry::install(); + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + drop(pool.acquire().await.unwrap()); + pool.execute("SELECT 1").await.unwrap(); + pool.fetch_all("SELECT 1").await.unwrap(); + + for name in LATENCY_HISTOGRAMS { + assert_eq!( + histogram_bounds(&tel, name), + EXPECTED_LATENCY_BUCKETS, + "{name} boundaries" + ); + } + + // The negative half. Row counts are not durations, so the seconds-scale set must not reach + // them – and the shared constant now sits two lines from their builders, which is exactly + // where an accidental copy-paste would land. + for name in [ + "db.client.response.returned_rows", + "db.client.response.affected_rows", + ] { + assert_ne!( + histogram_bounds(&tel, name), + EXPECTED_LATENCY_BUCKETS, + "{name} must keep the SDK default boundaries" + ); + } +} + +/// Sub-millisecond acquisitions must land in the low buckets, not be swallowed by a coarse one. +/// +/// This is the behaviour the boundaries exist for: an in-memory `SQLite` pool acquires in tens of +/// microseconds, which the SDK defaults would record as indistinguishable from a four-second +/// wait. The assertion is deliberately made against the 5ms boundary rather than the tightest +/// one measurement allows – still three orders of magnitude finer than the SDK default's first +/// edge, without making the test hostage to a scheduler hiccup on a loaded CI runner. +#[tokio::test] +#[serial] +async fn submillisecond_acquisition_lands_below_the_first_millisecond_boundary() { + let tel = common::TestTelemetry::install(); + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + for _ in 0..8 { + drop(pool.acquire().await.unwrap()); + } + + tel.reset(); + let metrics = tel.metrics(); + let metric = find_metric(&metrics, "db.client.connection.wait_time").unwrap(); + let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() else { + panic!("expected histogram") + }; + let dp = hist.data_points().next().unwrap(); + let bounds: Vec = dp.bounds().collect(); + let counts: Vec = dp.bucket_counts().collect(); + let five_ms = bounds + .iter() + .position(|b| (b - 0.005).abs() < f64::EPSILON) + .expect("0.005 boundary missing from the shipped set"); + let below_five_ms: u64 = counts[..=five_ms].iter().sum(); + assert_eq!( + below_five_ms, + dp.count(), + "warm in-memory acquisitions should all record under 5ms; got {counts:?}" + ); +} + +/// A query that fails after acquisition must still release its connection and clear pending. +/// +/// The failure arrives from the driver once the connection is already leased, so the guard and +/// the usage lease both have to unwind on the error path, not just on success. +#[tokio::test] +#[serial] +async fn queries_failing_after_acquisition_release_the_connection() { + let tel = common::TestTelemetry::install(); + let raw = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_millis(200)) + .connect(":memory:") + .await + .unwrap(); + let pool = PoolBuilder::from(raw).build(); + + assert!(pool.fetch_optional("SELECT * FROM missing").await.is_err()); + assert!(pool.execute("NOT VALID SQL").await.is_err()); + assert!(pool.fetch_all("SELECT * FROM missing").await.is_err()); + + // Collect spans before touching any metric helper: they all call `TestTelemetry::reset`, + // which drains the span exporter as well as the metric one. + // + // Query-level instrumentation must survive the error path, not just acquisition accounting: + // each failure should still produce a span marked with the error. + let spans = tel.spans(); + assert_eq!(spans.len(), 3, "one span per failed query"); + assert!( + spans.iter().all(|span| span + .attributes + .iter() + .any(|kv| kv.key.as_str() == "error.type")), + "every failed query span must carry error.type" + ); + + assert_eq!(pending_count(&tel), 0); + assert_eq!( + timeout_count(&tel), + 0, + "a query error is not a pool timeout" + ); + assert_eq!(histogram_count(&tel, "db.client.connection.wait_time"), 3); + assert_eq!(histogram_count(&tel, "db.client.connection.use_time"), 3); + assert_eq!( + histogram_count(&tel, "db.client.operation.duration"), + 3, + "a failed query should still record its duration" + ); + // The inner `expect` carries the diagnostic: these pools set a 200ms acquire timeout, so a + // leak surfaces as `Err(PoolTimedOut)` long before the outer 1s timeout could fire. + tokio::time::timeout(Duration::from_secs(1), pool.acquire()) + .await + .expect("acquire hung") + .expect("failed queries exhausted the pool"); +} + +/// A closed pool fails acquisition without leasing a connection or counting a timeout. +/// +/// `PoolClosed` and `PoolTimedOut` are both acquisition failures but only the latter belongs in +/// `db.client.connection.timeouts`; conflating them would inflate the timeout rate on shutdown. +#[tokio::test] +#[serial] +async fn closed_pool_records_no_lease_and_no_timeout() { + let tel = common::TestTelemetry::install(); + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + pool.close().await; + + assert!(pool.fetch_optional("SELECT 1").await.is_err()); + assert!(pool.begin().await.is_err()); + + assert_eq!(pending_count(&tel), 0); + assert_eq!(timeout_count(&tel), 0, "PoolClosed is not a timeout"); + assert_eq!( + histogram_count(&tel, "db.client.connection.use_time"), + 0, + "no connection was ever leased" + ); +} + +/// A multi-statement stream holds one connection across every statement, and releases it when +/// abandoned between them. +/// +/// Row-only streams are already covered by `dropping_a_partially_consumed_stream_releases_its_ +/// only_connection`. The distinct axis here is `fetch_many` yielding `Either::Left(QueryResult)` +/// at each statement boundary: the generator resumes across those boundaries on the same lease, +/// so a single acquisition must cover all three statements no matter where the caller stops. +#[tokio::test] +#[serial] +async fn multi_statement_stream_holds_one_lease_across_statement_boundaries() { + use futures::TryStreamExt; + let tel = common::TestTelemetry::install(); + let raw = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_millis(200)) + .connect(":memory:") + .await + .unwrap(); + let pool = PoolBuilder::from(raw).build(); + pool.execute("CREATE TABLE t (id integer)").await.unwrap(); + + // The SDK aggregates histograms cumulatively and `TestTelemetry::reset` only drops exported + // batches, so the setup query's observations persist. Compare against a baseline rather than + // against zero. + let waits_before = histogram_count(&tel, "db.client.connection.wait_time"); + let uses_before = histogram_count(&tel, "db.client.connection.use_time"); + + // Three statements: each completion yields a `Left(QueryResult)` before the next begins. + #[allow(deprecated)] + let mut items = Box::pin(pool.fetch_many( + "INSERT INTO t VALUES (1); INSERT INTO t VALUES (2); INSERT INTO t VALUES (3);", + )); + + let first = items.try_next().await.unwrap().expect("no first item"); + assert!( + first.is_left(), + "expected a QueryResult at the first statement boundary" + ); + assert_eq!( + histogram_count(&tel, "db.client.connection.wait_time"), + waits_before + 1, + "one acquisition covers the whole multi-statement stream" + ); + assert_eq!( + histogram_count(&tel, "db.client.connection.use_time"), + uses_before, + "the lease is still open while the stream is mid-flight" + ); + + // Abandon between statements, not between rows. + drop(items); + + assert_eq!( + histogram_count(&tel, "db.client.connection.use_time"), + uses_before + 1, + "abandoning the stream must close out exactly one lease" + ); + assert_eq!( + histogram_count(&tel, "db.client.connection.wait_time"), + waits_before + 1, + "no statement boundary may trigger a second acquisition" + ); + assert_eq!(pending_count(&tel), 0); + tokio::time::timeout(Duration::from_secs(1), pool.acquire()) + .await + .expect("acquire hung") + .expect("stream abandoned between statements leaked its connection"); +} + +/// The three latency histogram names the crate ships explicit boundaries for. +const LATENCY_HISTOGRAMS: [&str; 3] = [ + "db.client.connection.wait_time", + "db.client.connection.use_time", + "db.client.operation.duration", +]; + +/// Build a pool under a freshly installed meter provider and report each latency histogram's +/// bucket boundaries. +/// +/// Takes an optional view so the same pool workload can be observed with and without one. Each +/// call installs its own provider because `PoolBuilder::build` resolves its instruments from +/// whatever provider is global at that moment, so the view has to be registered first. +async fn latency_bounds_under_view(view_boundaries: Option>) -> Vec> { + use opentelemetry_sdk::metrics::{ + Aggregation, InMemoryMetricExporter, Instrument, PeriodicReader, SdkMeterProvider, Stream, + }; + + let exporter = InMemoryMetricExporter::default(); + let mut builder = + SdkMeterProvider::builder().with_reader(PeriodicReader::builder(exporter.clone()).build()); + if let Some(boundaries) = view_boundaries { + // Match only the three histograms. A broader predicate would also hit the counters and + // gauges, where the SDK rejects a histogram aggregation and silently falls back to the + // implicit default view – passing through an error-recovery path instead of this one. + builder = builder.with_view(move |i: &Instrument| { + LATENCY_HISTOGRAMS.contains(&i.name()).then(|| { + Stream::builder() + .with_aggregation(Aggregation::ExplicitBucketHistogram { + boundaries: boundaries.clone(), + record_min_max: true, + }) + .build() + .unwrap() + }) + }); + } + let provider = builder.build(); + opentelemetry::global::set_meter_provider(provider.clone()); + + let pool = PoolBuilder::from(sqlx::SqlitePool::connect(":memory:").await.unwrap()).build(); + drop(pool.acquire().await.unwrap()); + pool.execute("SELECT 1").await.unwrap(); + provider.force_flush().unwrap(); + + let collected = exporter.get_finished_metrics().unwrap(); + LATENCY_HISTOGRAMS + .iter() + .map(|name| { + let metric = find_metric(&collected, name).expect("histogram not recorded"); + let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() else { + panic!("expected histogram for {name}") + }; + hist.data_points() + .next() + .expect("no data point") + .bounds() + .collect() + }) + .collect() +} + +/// An application `View` must be able to override the bucket boundaries the crate ships. +/// +/// The crate sets boundaries as instrument *advice*, and the SDK applies advice only where no +/// matching view has already set an aggregation. That is what lets callers retune the histograms +/// without this crate exposing an API for it, and the README documents it. +/// +/// Both halves are asserted in one test on purpose. Checking only that a view produces the +/// boundaries it asked for would pass even if the crate shipped no advice at all – it would be +/// testing the SDK, not this crate. +/// +/// The baseline does most of the work: it shows the crate's boundaries are applied when no view +/// is registered, so the second half is genuinely replacing them rather than supplying a value +/// where there was none. The `assert_ne!` covers the one case the baseline cannot – an override +/// that happens to match the shipped set, which would leave the two halves indistinguishable. +/// +/// This test installs its own meter provider rather than using `TestTelemetry`, which has no +/// hook for registering views. That makes `#[serial]` matter more here than elsewhere: nothing +/// restores the previous provider afterwards, and the next test's `TestTelemetry` installation +/// is what reclaims it. +#[tokio::test] +#[serial] +async fn an_application_view_overrides_the_shipped_bucket_boundaries() { + const OVERRIDE: &[f64] = &[0.005, 0.05, 0.5, 5.0]; + + // Without a view, the pool must come up on the boundaries the crate advises. + let baseline = latency_bounds_under_view(None).await; + for (name, bounds) in LATENCY_HISTOGRAMS.iter().zip(&baseline) { + assert_eq!( + bounds, EXPECTED_LATENCY_BUCKETS, + "{name}: crate advice must apply when no view matches" + ); + } + + // With one, the view's aggregation must win outright. + let overridden = latency_bounds_under_view(Some(OVERRIDE.to_vec())).await; + for (name, bounds) in LATENCY_HISTOGRAMS.iter().zip(&overridden) { + assert_eq!( + bounds, OVERRIDE, + "{name}: a view's aggregation must beat the crate's advice" + ); + } + + // The guard that keeps this test honest: if the crate ever stopped shipping advice, the two + // observations would coincide and the assertions above would still pass individually. + assert_ne!( + baseline, overridden, + "override must actually change the bucket layout" + ); +} + /// Helper to find a named metric in the collected resource metrics. fn find_metric<'a>( resource_metrics: &'a [opentelemetry_sdk::metrics::data::ResourceMetrics], From 0bca4370f0ae053abcd0b7f72133b41c804b1dbc Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Mon, 21 Sep 2026 13:01:24 +0100 Subject: [PATCH 7/8] docs: Stamp CHANGELOG for the pool acquisition work Covers both unreleased pull requests. #40 routed every acquisition path through the instrumented `Pool::acquire()`, fixed the pending-requests leak on cancellation, and gave `wait_time` explicit bucket boundaries; #42 extended those boundaries to `operation.duration` and `use_time`, which had the same millisecond-shaped defaults. Entries are attributed per pull request rather than per behaviour, so the bucket change names both: bisecting a shifted dashboard should land on the commit that moved the histogram in question. --- CHANGELOG.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1448e02..e7ca794 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- Implicit pool queries and `Pool::begin()` now route through the instrumented `Pool::acquire()`, so `db.client.connection.wait_time`, `use_time`, `pending_requests`, and `timeouts` account for every acquisition rather than only explicit `acquire()` calls. A workload written as `query(..).fetch_one(&pool)` previously borrowed its connection straight from `sqlx::Pool` and could saturate the pool without any of its waits or timeouts reaching telemetry. Queries issued inside a transaction still acquire once, and connection use time now follows the lease through transaction commit, rollback, or drop ([#40](https://github.com/chmodas/sqlx-otel/pull/40)). +- Dropping an in-flight `acquire()` – a caller timeout or an aborted task – no longer leaves `db.client.connection.pending_requests` permanently incremented. The decrement moved to a `Drop` guard covering success, error, and cancellation alike. Cancellation records neither a completed wait nor a pool timeout, because the pool did not fail to deliver; only `sqlx::Error::PoolTimedOut` increments the timeout counter, so `PoolClosed` during shutdown no longer inflates the timeout rate ([#40](https://github.com/chmodas/sqlx-otel/pull/40)). + +### Changed + +- `db.client.connection.wait_time` declares explicit bucket boundaries instead of inheriting the OpenTelemetry SDK defaults ([#40](https://github.com/chmodas/sqlx-otel/pull/40)), and `db.client.operation.duration` and `db.client.connection.use_time` now use the same set ([#42](https://github.com/chmodas/sqlx-otel/pull/42)). The SDK defaults (`0, 5, 10, 25, … 10000`) are shaped for milliseconds, but semconv requires these histograms be recorded in seconds, so every observation below five seconds landed in a single bucket and any quantile computed from it was an interpolation artefact rather than a measurement. The boundaries run from `0.0001` to `60.0` seconds, covering sub-millisecond local round trips through the 30-second acquire timeout SQLx applies by default. **Dashboards and recording rules reading these histograms will see their bucket layout change.** The row-count histograms `db.client.response.returned_rows` and `db.client.response.affected_rows` keep the SDK defaults, which already suit counts. The boundaries are instrument advice, so an application that registers a `View` with a compatible explicit aggregation overrides them and the crate exposes no API of its own for retuning; the README carries a worked example ([#42](https://github.com/chmodas/sqlx-otel/pull/42)). +- `db.client.connection.pending_requests` and `db.client.connection.timeouts` are initialised at zero when the pool is built, so both series exist from startup instead of appearing only after the first waiting caller or first timeout ([#40](https://github.com/chmodas/sqlx-otel/pull/40)). + ## [0.5.0] – 2026-08-09 ### Changed From 83ee8779a5faa699a13013e1fd0eb2d36dd2c810 Mon Sep 17 00:00:00 2001 From: Alex Torchenko Date: Wed, 9 Sep 2026 16:49:35 +0700 Subject: [PATCH 8/8] fix: preserve owned lifetime of pool transactions --- src/pool.rs | 2 +- tests/sqlite.rs | 23 +++++++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/pool.rs b/src/pool.rs index e5b2f9d..a57d4c3 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -423,7 +423,7 @@ impl Pool { /// /// Returns `sqlx::Error` if `BEGIN` fails – typically due to a connection problem or /// because the underlying connection cannot start a new transaction. - pub async fn begin(&self) -> Result, sqlx::Error> { + pub async fn begin(&self) -> Result, sqlx::Error> { let PoolConnection { inner, state, diff --git a/tests/sqlite.rs b/tests/sqlite.rs index 830a163..17fdac6 100644 --- a/tests/sqlite.rs +++ b/tests/sqlite.rs @@ -409,6 +409,29 @@ async fn transaction_rollback() { test_transaction_rollback!(test_pool().await, common::SQLITE_DIALECT); } +/// Returning an owned transaction must not borrow the pool handle. This helper +/// deliberately requires the public API to preserve `SQLx`'s static lifetime. +async fn owned_transaction( + pool: &Pool, +) -> Result, sqlx::Error> { + pool.begin().await +} + +#[tokio::test] +#[serial] +async fn pool_transaction_can_outlive_pool_handle() { + let mut transaction = { + let pool = test_pool().await; + owned_transaction(&pool).await.unwrap() + }; + + sqlx::query("SELECT 1") + .execute(&mut transaction) + .await + .unwrap(); + transaction.rollback().await.unwrap(); +} + // =========================================================================== // PoolBuilder with_* methods // ===========================================================================