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 e23a4ce..3019015 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,51 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [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)). + +## Unreleased — fork integration + +### Added + +- Opt-in automatic query summaries (`QuerySummaryMode::Auto`) retain operation and + target labels without literal values, preserving the fork's SQL summaries on SQLx 0.9. + +### Changed + +- Automatic summaries now keep physical write targets when a common table expression + shadows the target name. For example, `WITH reports AS (SELECT id FROM source) + UPDATE reports SET id = 1` changes from `UPDATE source` to `UPDATE reports` relative to + the previous fork pin. Span names and `db.query.summary` metric series change accordingly. +- Preserve the fork's database error classification: `error.type` uses the driver status + code when available and otherwise the canonical `sqlx::Error::` name; + `exception.type` always uses the canonical SQLx variant. + +## [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` 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. + +## [0.3.0] – 2026-05-05 + ### Added - `db.client.response.affected_rows` histogram – a custom metric (no OTel semconv equivalent) mirroring the existing `db.response.affected_rows` span attribute. Recorded on `execute()` calls; carries the same connection / annotation / error attribute set as the duration histogram so dashboards can slice mutation throughput by the same dimensions ([#33](https://github.com/chmodas/sqlx-otel/pull/33)). @@ -13,12 +58,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Changed -- Updated `opentelemetry`, `opentelemetry-semantic-conventions`, and the test SDK to - `0.32.x`. -- `error.type` now matches `db.response.status_code` for database errors when the driver - provides a code, as required by the database client semantic conventions. Other errors - use their canonical `sqlx::Error::` name; `exception.type` remains the canonical - SQLx variant rather than a database status code. - `db.client.connection.pool.name` previously appeared only on the `db.client.connection.count` gauge; it now also propagates to spans, the `db.client.operation.duration` / `db.client.response.returned_rows` histograms, and the rest of the `db.client.connection.*` family via the shared connection-attribute set. The `count` gauge's attribute set is unchanged ([#32](https://github.com/chmodas/sqlx-otel/pull/32)). - `db.query.text` now collapses inter-token whitespace runs to a single space and trims leading/trailing whitespace for both `QueryTextMode::Full` (default) and `QueryTextMode::Obfuscated`. Whitespace inside string literals, quoted identifiers, dollar-quoted bodies, and comments is preserved verbatim. Multi-line SQL written for source-level readability now renders as a single readable line in OTel exports without the embedded `\n` and indentation runs that came from source-level formatting. `QueryTextMode::Off` is unchanged ([#34](https://github.com/chmodas/sqlx-otel/pull/34)). @@ -52,6 +91,8 @@ 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 +[0.5.0]: https://github.com/chmodas/sqlx-otel/releases/tag/v0.5.0 diff --git a/Cargo.toml b/Cargo.toml index a8995d4..5fdbea3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sqlx-otel" -version = "0.3.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", @@ -27,14 +27,20 @@ async-stream = "0.3" 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"] @@ -43,12 +49,12 @@ all-features = true rustdoc-args = ["--cfg", "docsrs"] [dev-dependencies] -ctor = "0.10" +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" +sqlx = { version = "0.9", features = ["macros", "sqlite", "postgres", "mysql", "mysql-rsa", "runtime-tokio"] } +testcontainers = "0.28" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } [[example]] diff --git a/README.md b/README.md index 6aade43..3f8aaa2 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ The wrapper talks to the [`opentelemetry`](https://docs.rs/opentelemetry) API di - **Spans on every operation.** Every `sqlx::Executor` method emits a `SpanKind::Client` span carrying the OTel database-spans recommended set: `db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name`, `db.query.text`, returned/affected row counts, and SQLSTATE / `error.type` on failure. - **Span/metric attribute parity.** The `db.client.operation.duration` histogram carries the same bounded attribute set as spans – every annotation, every error-path attribute, every connection-level attribute – so dashboards can slice query latency by operation verb, target collection, error class, or pool name. `db.query.text` is the only span attribute deliberately excluded from metrics for cardinality. -- **Caller-supplied annotations.** The library does not parse SQL. Per-query attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) come from a small annotation API. +- **Caller-supplied annotations.** Per-query attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) come from a small annotation API. Optional automatic query summaries provide a best-effort operation/target label when annotations are absent; SQL extraction is off by default. - **Three backends, one API.** Postgres, SQLite, and MySQL behind feature flags; the wrapper API is identical across all three. - **Drop-in.** `&Pool` implements `sqlx::Executor`, so existing call sites keep working unchanged. @@ -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.5.0", features = ["postgres", "runtime-tokio"] } ``` ```rust @@ -89,8 +89,8 @@ let pool = PoolBuilder::from(raw_pool) ## Automatic query summaries -Enable conservative low-cardinality `db.query.summary` generation when a short operation -name is useful for grouping: +Automatic summaries are **off by default**. Enable best-effort `db.query.summary` +generation when a short operation name is useful for grouping: ```rust use sqlx_otel::{PoolBuilder, QuerySummaryMode}; @@ -103,7 +103,16 @@ let pool = PoolBuilder::from(raw_pool) The summary contains only the outer operation and primary target, such as `SELECT users` or `INSERT orders`. Common table expressions are resolved to their primary physical target when possible. Predicates, selected columns, literal values, and bind parameters are never -included. Explicit per-query annotations take precedence. +included for supported SQL forms. The generated summary names the span and is attached to +operation metrics, independently of `QueryTextMode` (including `Off`). It does not infer +`db.operation.name` or `db.collection.name`. + +Any explicit per-query annotations take precedence and disable inference for that query, +even if they do not set a summary. This is a lightweight extractor, not a complete SQL +parser: unsupported forms may produce an operation-only summary or keep the existing +generic span name. Summaries are capped at 255 bytes, falling back to the operation rather +than truncating an identifier. Table/schema names still contribute to metric cardinality; +use explicit summaries or keep this mode off for dynamically named or sensitive objects. ## Per-query annotations @@ -190,7 +199,7 @@ Set on every `Executor` method (`execute`, `fetch`, `fetch_all`, `fetch_one`, `f | `db.query.text` | The SQL query string with inter-token whitespace collapsed | Unless `QueryTextMode::Off` | | `db.operation.name` | Database operation (e.g. `SELECT`) | When annotated | | `db.collection.name` | Target table or collection | When annotated | -| `db.query.summary` | Low-cardinality query summary | When annotated | +| `db.query.summary` | Operation/target summary or explicit query summary | When generated or annotated | | `db.stored_procedure.name` | Stored procedure name | When annotated | | `db.response.returned_rows` | Row count | On `fetch*` methods | | `db.response.affected_rows` | Rows affected (`rows_affected()`) | On `execute` | @@ -212,7 +221,7 @@ error variant name. An `exception` event is added with the canonical SQLx varian | `db.client.response.returned_rows` | Histogram | | Number of rows returned per `fetch*` call | | `db.client.response.affected_rows` | Histogram | | Rows affected per `execute` call (custom) | -These mirror the bounded portion of the span attribute set: connection-level attributes (`db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name` – wherever set), plus annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) when present, plus error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`) on the error path. `db.query.text` is deliberately excluded for cardinality; `db.query.summary` is caller-controlled and inherits its cardinality cost from the span side. +These mirror the bounded portion of the span attribute set: connection-level attributes (`db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name` – wherever set), plus annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) when present, plus error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`) on the error path. `db.query.text` is deliberately excluded for cardinality. Automatically generated `db.query.summary` values contain only an operation/target pair; dynamically named targets still contribute to metric cardinality. Explicit summaries remain caller-controlled and inherit their cardinality cost from the span side. `db.client.response.affected_rows` is not part of the OpenTelemetry semantic conventions – we ship it for the same reason as the matching span attribute: backends report a useful database-confirmed count that's worth slicing alongside duration. It is recorded only on `execute()` calls (where `QueryResult::rows_affected()` is meaningful) and is not recorded for `execute_many` (deprecated upstream). @@ -229,32 +238,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. +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. -All three latency histograms (`db.client.operation.duration`, -`db.client.connection.wait_time`, and `db.client.connection.use_time`) use explicit -boundaries in seconds: +### 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, 2.5, 5, 10, 30, 60 +0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0 ``` -This preserves resolution for subsecond operations instead of grouping them into the -SDK's default `(0, 5]` bucket. Applications can override this instrument advice with -an OpenTelemetry SDK View. Row-count histograms retain their default boundaries. +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(); +``` -`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. +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 @@ -268,9 +286,37 @@ pending-request peaks. The remaining three are static gauges recorded once at co ## Compatibility -- **MSRV:** Rust **1.85.0**. -- **SQLx:** `0.8.x`. -- **OpenTelemetry:** `0.32.x`. +| 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` | +| `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. + +When pinning SQLx to a Git revision, patch the complete SQLx package family from the +workspace root (`sqlx`, `sqlx-core`, the enabled driver crates, `sqlx-macros`, and +`sqlx-macros-core`) so this wrapper and the application share one source and one +`sqlx-core`. The wrapper enables `offline` on its selected drivers for `Executor::describe`; +if the application enables another driver outside the wrapper's backend features, enable +that driver's `offline` feature too. Check source and feature unification with +`cargo tree -i sqlx-core -e features` before adopting a pin. + +## 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 diff --git a/src/annotations.rs b/src/annotations.rs index 5bc0852..0b85229 100644 --- a/src/annotations.rs +++ b/src/annotations.rs @@ -46,7 +46,8 @@ macro_rules! impl_with_annotations_mut { } /// Per-query annotation values that enrich OpenTelemetry spans with semantic-convention -/// attributes the library cannot derive automatically (because it does not parse SQL). +/// attributes. Explicit annotations take precedence over automatic query summaries and +/// supply operation, collection, and stored-procedure attributes the library does not infer. /// /// Use the builder methods to set whichever attributes apply to a given query, then pass /// the result through one of the equivalent annotation surfaces: diff --git a/src/attributes.rs b/src/attributes.rs index d6c7a5b..037449a 100644 --- a/src/attributes.rs +++ b/src/attributes.rs @@ -58,6 +58,10 @@ pub enum QueryTextMode { /// possible. Literal values, predicates, selected columns, and bind parameters are never /// included. /// +/// This is a best-effort extractor, not a complete SQL parser. Dynamic object names still +/// contribute to metric cardinality. Keep this mode off or supply explicit annotations if +/// object names are sensitive or unbounded. +/// /// Configured via /// [`PoolBuilder::with_query_summary_mode`](crate::PoolBuilder::with_query_summary_mode). #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] diff --git a/src/executor.rs b/src/executor.rs index 251d77a..05a578b 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -98,8 +98,8 @@ fn start_span(name: &str, span_attrs: Vec) -> (OtelContext, Instant) { /// attributes plus the four annotation-derived attributes when present, plus error-path attributes /// (`error.type`, `db.response.status_code`) appended later by `record_error`. The unbounded /// `db.query.text` attribute is deliberately excluded. Automatically generated -/// `db.query.summary` values contain only a bounded operation/target pair; explicit summaries -/// remain caller-controlled, so their cardinality cost is inherited from the span side. +/// `db.query.summary` values contain an operation/target pair; dynamically named targets still contribute to +/// cardinality. Explicit summaries remain caller-controlled. fn begin_query_span( attrs: &ConnectionAttributes, sql: Option<&str>, @@ -160,10 +160,10 @@ fn exception_type(err: &sqlx::Error) -> &'static str { /// take precedence over the generic `SQLx` error variant. Non-database errors and database /// errors without a code fall back to their canonical `SQLx` variant name. fn error_type(err: &sqlx::Error) -> Cow<'static, str> { - if let sqlx::Error::Database(db_err) = err { - if let Some(code) = db_err.code() { - return Cow::Owned(code.into_owned()); - } + if let sqlx::Error::Database(db_err) = err + && let Some(code) = db_err.code() + { + return Cow::Owned(code.into_owned()); } Cow::Borrowed(exception_type(err)) } @@ -189,15 +189,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; @@ -462,10 +462,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, @@ -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(query); Box::pin(InstrumentedStream::<_, CountAll>::new( stream, @@ -518,10 +518,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, @@ -545,10 +545,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; @@ -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_one(query); Box::pin(async move { let result = fut.await; @@ -612,10 +612,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; @@ -642,18 +642,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, @@ -665,19 +666,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, @@ -688,11 +690,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>, @@ -701,7 +705,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, @@ -936,7 +941,7 @@ mod tests { let (_, _, metric_attrs) = begin_query_span( &attrs, Some( - "WITH source_rows AS (SELECT id FROM data_ba_event WHERE token = 'secret') \ + "WITH source_rows AS (SELECT id FROM events WHERE token = 'secret') \ SELECT COUNT(*) FROM source_rows", ), None, @@ -947,7 +952,7 @@ mod tests { .map(|kv| kv.value.clone()); assert_eq!( summary, - Some(opentelemetry::Value::String("SELECT data_ba_event".into())) + Some(opentelemetry::Value::String("SELECT events".into())) ); } diff --git a/src/lib.rs b/src/lib.rs index 1b2ddbd..9d33814 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -81,7 +81,7 @@ //! | `db.response.returned_rows` | Row count | On `fetch*` methods | //! | `db.response.affected_rows` | Rows affected (`QueryResult::rows_affected()`) | On `execute` ([note](#a-note-on-dbresponseaffected_rows)) | //! | `db.response.status_code` | SQLSTATE / driver error code | On database errors | -//! | `error.type` | Error variant name | On any error | +//! | `error.type` | Database status code or canonical `SQLx` error variant | On any error | //! //! [annotated]: #per-query-annotations //! @@ -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 @@ -220,7 +239,7 @@ //! //! ```toml //! [dependencies] -//! sqlx-otel = { version = "0.1", features = ["postgres", "runtime-tokio"] } +//! sqlx-otel = { version = "0.5", features = ["postgres", "runtime-tokio"] } //! ``` //! //! All operation- and pool-level metrics other than `db.client.connection.count` work @@ -228,7 +247,7 @@ //! //! # MSRV //! -//! Minimum supported Rust version: **1.85.0**. +//! Minimum supported Rust version: **1.94.0**. #![cfg_attr(docsrs, feature(doc_cfg))] @@ -245,6 +264,7 @@ mod pool; mod pool_executor; mod pool_metrics; mod query_ext; +mod rebuilt_query; mod runtime; mod summary; mod transaction; diff --git a/src/metrics.rs b/src/metrics.rs index e5121be..d6ed1a5 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -4,10 +4,20 @@ use opentelemetry::KeyValue; use opentelemetry::metrics::{Histogram, Meter}; use opentelemetry_semantic_conventions::metric; -/// Explicit boundaries in seconds for operation, acquisition, and connection-use latency. +/// Explicit bucket boundaries, in seconds, for every latency histogram the crate records. /// -/// SDK defaults start at 0 and 5, collapsing typical subsecond durations into one bucket. -/// These are instrument advice; applications can override them with a metric view. +/// 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, @@ -78,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 1bafd64..7fedf0f 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -518,8 +518,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 d843afb..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,11 +79,16 @@ where }) } - fn prepare_with<'e, 'q: 'e>( + /// 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: &'q str, + sql: sqlx::SqlStr, parameters: &'e [DB::TypeInfo], - ) -> BoxFuture<'e, Result, Error>> + ) -> BoxFuture<'e, Result> where 'c: 'e, { @@ -70,7 +103,8 @@ where }) } - fn describe<'e, 'q: 'e>(self, sql: &'q str) -> BoxFuture<'e, Result, Error>> + /// 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/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/src/summary.rs b/src/summary.rs index 8dfe5ae..920d2bd 100644 --- a/src/summary.rs +++ b/src/summary.rs @@ -62,6 +62,10 @@ fn summarize_tokens( let (local_ctes, statement_start) = parse_ctes(tokens); let (operation, operation_index) = find_operation(tokens, statement_start)?; let target = find_target(tokens, operation_index, &operation).map(|target| { + // A write target names the physical object even when a CTE shadows its name. + if !operation.eq_ignore_ascii_case("SELECT") { + return target; + } resolve_cte_target( target, &local_ctes, @@ -522,19 +526,19 @@ mod tests { let sql = r" WITH source_rows AS ( SELECT event.id - FROM data_ba_event event - JOIN data_ba_league league ON league.id = event.league_id + FROM events event + JOIN leagues league ON league.id = event.league_id ) SELECT COUNT(*) FROM source_rows "; - assert_eq!(summarize(sql).as_deref(), Some("SELECT data_ba_event")); + assert_eq!(summarize(sql).as_deref(), Some("SELECT events")); } #[test] fn resolves_chained_ctes() { let sql = r" WITH baseline AS ( - SELECT id FROM data_ba_event + SELECT id FROM events ), source_rows AS ( SELECT id FROM baseline @@ -544,20 +548,30 @@ mod tests { ) SELECT COUNT(*) FROM evaluated "; - assert_eq!(summarize(sql).as_deref(), Some("SELECT data_ba_event")); + assert_eq!(summarize(sql).as_deref(), Some("SELECT events")); } #[test] fn insert_with_ctes_keeps_outer_write_target() { let sql = r" - WITH baseline AS (SELECT id FROM data_ba_event) - INSERT INTO normalized_event_state_artifact(event_id) + WITH baseline AS (SELECT id FROM events) + INSERT INTO event_reports(event_id) SELECT id FROM baseline "; - assert_eq!( - summarize(sql).as_deref(), - Some("INSERT normalized_event_state_artifact") - ); + assert_eq!(summarize(sql).as_deref(), Some("INSERT event_reports")); + } + + #[test] + fn write_targets_are_not_resolved_through_cte_aliases() { + for statement in [ + "INSERT INTO reports SELECT * FROM reports", + "UPDATE reports SET id = 1", + "DELETE FROM reports", + ] { + let sql = format!("WITH reports AS (SELECT id FROM source) {statement}"); + let operation = statement.split_whitespace().next().unwrap(); + assert_eq!(summarize(&sql), Some(format!("{operation} reports"))); + } } #[test] diff --git a/tests/common.rs b/tests/common.rs index 692908e..220baa5 100644 --- a/tests/common.rs +++ b/tests/common.rs @@ -191,10 +191,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()); } } } @@ -566,10 +566,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(); }}; } @@ -1588,6 +1593,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 { @@ -1896,7 +1956,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); @@ -1904,13 +1967,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); @@ -1928,7 +1991,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); @@ -1936,13 +2002,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); @@ -1960,15 +2026,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(); @@ -1994,7 +2063,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(); @@ -2005,7 +2076,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(); @@ -2014,7 +2085,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(); @@ -2034,7 +2105,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(); @@ -2044,13 +2118,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); @@ -2069,7 +2149,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(); @@ -2079,13 +2162,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); @@ -2104,17 +2193,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(); @@ -2140,7 +2238,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(); @@ -2151,7 +2251,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(); @@ -2160,7 +2260,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(); @@ -2179,7 +2279,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); @@ -2187,13 +2290,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); @@ -2211,7 +2314,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); @@ -2219,13 +2325,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); @@ -2243,15 +2349,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(); @@ -2277,7 +2386,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(); @@ -2288,7 +2399,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(); @@ -2297,7 +2408,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(); @@ -3732,7 +3843,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 74d144c..cc8491a 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"); @@ -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()]) @@ -257,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/pool_metrics.rs b/tests/pool_metrics.rs index 6860bb1..4dd310d 100644 --- a/tests/pool_metrics.rs +++ b/tests/pool_metrics.rs @@ -11,89 +11,14 @@ use std::time::Duration; const POOL_NAME: &str = "test-pool"; -#[tokio::test] -#[serial] -async fn latency_histograms_use_seconds_scale_buckets_and_allow_view_overrides() { - use opentelemetry_sdk::metrics::{ - Aggregation, InMemoryMetricExporter, Instrument, PeriodicReader, SdkMeterProvider, Stream, - }; - - const LATENCY_NAMES: [&str; 3] = [ - "db.client.operation.duration", - "db.client.connection.wait_time", - "db.client.connection.use_time", - ]; - const EXPECTED: &[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, - ]; - const OVERRIDE: &[f64] = &[0.005, 0.05, 0.5, 5.0]; - const DEFAULT_ROW_BUCKETS: &[f64] = &[ - 0.0, 5.0, 10.0, 25.0, 50.0, 75.0, 100.0, 250.0, 500.0, 750.0, 1000.0, 2500.0, 5000.0, - 7500.0, 10000.0, - ]; - - // Assert the exported layout both without a view and with one. Checking only the - // override would pass even if the instruments supplied no boundaries at all. - for view_boundaries in [None, Some(OVERRIDE)] { - let exporter = InMemoryMetricExporter::default(); - let mut builder = SdkMeterProvider::builder() - .with_reader(PeriodicReader::builder(exporter.clone()).build()); - if let Some(boundaries) = view_boundaries { - builder = builder.with_view(move |instrument: &Instrument| { - LATENCY_NAMES.contains(&instrument.name()).then(|| { - Stream::builder() - .with_aggregation(Aggregation::ExplicitBucketHistogram { - boundaries: boundaries.to_vec(), - 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("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.close().await; - drop(pool); - provider.force_flush().unwrap(); - - let collected = exporter.get_finished_metrics().unwrap(); - for name in LATENCY_NAMES.into_iter().chain([ - "db.client.response.returned_rows", - "db.client.response.affected_rows", - ]) { - let metric = find_metric(&collected, name).expect("histogram not recorded"); - let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { - panic!("expected histogram for {name}"); - }; - let expected = if LATENCY_NAMES.contains(&name) { - assert_eq!(metric.unit(), "s", "{name} unit"); - view_boundaries.unwrap_or(EXPECTED) - } else { - DEFAULT_ROW_BUCKETS - }; - let points: Vec<_> = histogram.data_points().collect(); - assert!(!points.is_empty(), "{name} has no observations"); - for point in points { - assert!(point.count() > 0, "{name} has no observations"); - assert_eq!( - point.bounds().collect::>(), - expected, - "{name} bounds" - ); - } - } - provider.shutdown().unwrap(); - } -} +/// 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(); @@ -160,8 +85,16 @@ async fn every_pool_executor_path_acquires_once_without_duplicate_query_spans() .try_collect::>() .await .unwrap(); - pool.prepare_with("SELECT id FROM test", &[]).await.unwrap(); - pool.describe("SELECT id FROM test").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(); + // `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 @@ -173,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); } @@ -309,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], diff --git a/tests/postgres.rs b/tests/postgres.rs index 16c0b27..329e7fe 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"); @@ -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()]) @@ -260,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..59e0c8b 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() { @@ -397,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 // =========================================================================== @@ -487,6 +522,74 @@ async fn query_summary_drives_span_name() { test_query_summary_drives_span_name!(test_pool().await, common::SQLITE_DIALECT); } +#[tokio::test] +#[serial] +async fn automatic_summaries_are_opt_in_and_group_without_literal_values() { + use sqlx_otel::{QueryAnnotations, QuerySummaryMode, QueryTextMode}; + + let tel = common::TestTelemetry::install(); + let default_pool = PoolBuilder::from(raw_pool().await) + .with_query_text_mode(QueryTextMode::Off) + .build(); + default_pool + .fetch_all("SELECT name FROM sqlite_master") + .await + .unwrap(); + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!(common::attr(&spans[0], "db.query.summary"), None); + tel.reset(); + + let pool = PoolBuilder::from(raw_pool().await) + .with_query_text_mode(QueryTextMode::Off) + .with_query_summary_mode(QuerySummaryMode::Auto) + .build(); + for sql in [ + "SELECT name FROM sqlite_master WHERE name = 'private_a'", + "SELECT name FROM sqlite_master WHERE name = 'private_b'", + ] { + sqlx::query(sql).fetch_all(&pool).await.unwrap(); + } + let spans = tel.spans(); + assert_eq!(spans.len(), 2); + for span in &spans { + assert_eq!(span.name, "SELECT sqlite_master"); + assert_eq!( + common::attr(span, "db.query.summary"), + Some("SELECT sqlite_master".into()) + ); + assert_eq!(common::attr(span, "db.query.text"), None); + assert_eq!(common::attr(span, "db.operation.name"), None); + assert_eq!(common::attr(span, "db.collection.name"), None); + } + let metrics = tel.metrics(); + let point = common::find_histogram_data_point_with( + &metrics, + "db.client.operation.duration", + &[("db.query.summary", "SELECT sqlite_master")], + ) + .unwrap(); + assert_eq!(point.count(), 2); + tel.reset(); + + sqlx::query("SELECT name FROM sqlite_master") + .with_annotations(QueryAnnotations::new().query_summary("schema lookup")) + .fetch_all(&pool) + .await + .unwrap(); + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!(spans[0].name, "schema lookup"); + assert!( + common::find_histogram_data_point_with( + &tel.metrics(), + "db.client.operation.duration", + &[("db.query.summary", "schema lookup")], + ) + .is_some() + ); +} + // =========================================================================== // query-side annotations: sqlx::query(...).with_annotations(...).execute(&pool) // ===========================================================================