diff --git a/.cargo/audit.toml b/.cargo/audit.toml
index 7b8b282..f52934c 100644
--- a/.cargo/audit.toml
+++ b/.cargo/audit.toml
@@ -9,7 +9,9 @@ ignore = [
     # plaintext from RSA *private-key decryption* (PKCS#1 v1.5). No fixed `rsa`
     # release exists.
     #
-    # Pulled in only by `sqlx-mysql` (via the `mysql` feature / sqlx dev-dependency).
+    # Pulled in only by `sqlx-mysql`'s optional `rsa` feature, which since sqlx 0.9
+    # only the `cargo-hammerwork` and `hammerwork-web` binaries enable (sqlx feature
+    # `mysql-rsa`); the library itself does not.
     # sqlx uses it solely in `connection/auth.rs` to *encrypt* the password with the
     # server's RSA public key (`RsaPublicKey` + OAEP) for the `caching_sha2_password`
     # / `sha256_password` auth plugins over non-TLS connections. Hammerwork never
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1ad040e..c3183d6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -43,7 +43,7 @@ jobs:
   # Builds every crate and target with the declared `rust-version` against the committed
   # Cargo.lock, which the MSRV-aware resolver (resolver = "3") keeps compatible with it.
   msrv:
-    name: MSRV (1.91.1)
+    name: MSRV (1.94.0)
     runs-on: ubuntu-latest
     timeout-minutes: 45
     steps:
@@ -52,7 +52,7 @@ jobs:
         run: sudo apt-get update && sudo apt-get install -y --no-install-recommends ${{ env.APT_PACKAGES }}
       - uses: dtolnay/rust-toolchain@master
         with:
-          toolchain: 1.91.1
+          toolchain: 1.94.0
       - uses: Swatinem/rust-cache@v2
         with:
           key: msrv
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5c952f1..3a68b1c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Breaking changes
 
+See [Upgrading from 2.x to 3.0](docs/upgrading-to-3.0.md) for what to change.
+
+- **sqlx 0.9** (#85). Hammerwork now depends on sqlx 0.9 instead of 0.8. sqlx types are
+  part of the public API (the pool passed to `JobQueue::new`, `JobQueue<DB: sqlx::Database>`,
+  `DatabaseQueue`, the migration runners, sqlx `Encode`/`Decode` for `JobStatus` and
+  `BatchStatus`), so applications must move their own sqlx dependency to 0.9 as well.
+  sqlx 0.9 also changes behaviour that applications may notice through their own sqlx
+  code: the combined runtime/TLS features are gone (`runtime-tokio-rustls` is now
+  `runtime-tokio` + `tls-rustls-ring-webpki`), dynamic SQL strings must be wrapped in
+  `sqlx::AssertSqlSafe`, MySQL RSA password encryption for non-TLS connections needs the
+  `mysql-rsa` feature (the CLI and dashboard enable it; the library does not), and MySQL
+  connections no longer force the `utf8mb4_unicode_ci` collation. No schema migration is
+  needed.
+- **MSRV is now Rust 1.94.0** (was 1.91.1), as required by sqlx 0.9.
 - **No features are enabled by default** (#92). `default` used to be `["metrics", "alerting", "webhooks"]`, which pulled `warp`, `reqwest`, `lettre` and `prometheus` into every build. It is now empty. Enable what you use, or the new `full` feature (`metrics`, `alerting`, `webhooks`, `streaming`, `tracing`, `encryption`; no database backend, KMS key source, event stream sink or `test-queue`). Code that uses `hammerwork::metrics`, `hammerwork::alerting` or `hammerwork::webhooks` must now enable those features. `cargo-hammerwork` and `hammerwork-web` enable the features they need themselves.
 - **`azure-kv` is renamed `azure-kms`**, to match `aws-kms`, `gcp-kms` and `vault-kms` (#92). There is no alias. The same rename applies to the `cargo-hammerwork` feature.
 - **`test` is renamed `test-queue`**, after the `TestQueue` it enables (#92). There is no alias.
@@ -20,6 +34,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - CI checks the feature combinations `full`, `postgres,full`, `mysql,full` and `test-queue`.
+- The library no longer depends on the `rsa` crate (RUSTSEC-2023-0071), which sqlx 0.9
+  made optional. It stays in the lockfile only through the CLI and dashboard binaries.
 
 ## [2.5.0] - 2026-10-11
 
diff --git a/CLAUDE.md b/CLAUDE.md
index 02ce495..0cc1c7b 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -16,6 +16,9 @@ Hammerwork is a high-performance, database-driven job queue library for Rust wit
 - Tracing is integrated for observability
 - Examples demonstrate database configurations, cron scheduling, and prioritization
 - Use edition 2024 as the preference
+- sqlx 0.9; MSRV is Rust 1.94.0 (workspace `rust-version`, checked by the CI MSRV job)
+- sqlx 0.9 only accepts `&'static str` SQL: wrap runtime-built SQL in `sqlx::AssertSqlSafe(...)`
+  (owned `String` preferred) after checking that only bind parameters carry user input
 
 ## Common Development Commands
 
diff --git a/Cargo.lock b/Cargo.lock
index 38aa7b3..5240c9f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -285,7 +285,7 @@ dependencies = [
  "fastrand",
  "hex",
  "http 1.5.0",
- "sha1",
+ "sha1 0.10.7",
  "time",
  "tokio",
  "tracing",
@@ -785,6 +785,12 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "base16ct"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fd307490d624467aa6f74b0eabb77633d1f758a7b25f12bceb0b22e08d9726f6"
+
 [[package]]
 name = "base64"
 version = "0.22.1"
@@ -1152,12 +1158,6 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
-[[package]]
-name = "const-oid"
-version = "0.9.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
-
 [[package]]
 name = "const-oid"
 version = "0.10.2"
@@ -1196,6 +1196,12 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48"
 
+[[package]]
+name = "cpubits"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae"
+
 [[package]]
 name = "cpufeatures"
 version = "0.2.17"
@@ -1287,6 +1293,20 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "crypto-bigint"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a52aa3fcda4e6302a9f48734f234d35d4721b96f8fe07d073f07ce9df4f0271"
+dependencies = [
+ "cpubits",
+ "ctutils",
+ "num-traits",
+ "rand_core 0.10.1",
+ "serdect",
+ "zeroize",
+]
+
 [[package]]
 name = "crypto-common"
 version = "0.1.7"
@@ -1307,6 +1327,16 @@ dependencies = [
  "hybrid-array",
 ]
 
+[[package]]
+name = "crypto-primes"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3633a51a39c69ebbaa4feaa694bd83d241e4093901c84a0963b19d9bb3f0cf8f"
+dependencies = [
+ "crypto-bigint",
+ "rand_core 0.10.1",
+]
+
 [[package]]
 name = "ctr"
 version = "0.9.2"
@@ -1374,11 +1404,11 @@ checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
 
 [[package]]
 name = "der"
-version = "0.7.10"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
+checksum = "a878c850e9e421b20262e9b41f9c860e4785fa07541c266b62ff9d1ef998a80a"
 dependencies = [
- "const-oid 0.9.6",
+ "const-oid",
  "pem-rfc7468",
  "zeroize",
 ]
@@ -1430,7 +1460,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
  "block-buffer 0.10.4",
- "const-oid 0.9.6",
  "crypto-common 0.1.7",
  "subtle",
 ]
@@ -1442,7 +1471,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
 dependencies = [
  "block-buffer 0.12.1",
- "const-oid 0.10.2",
+ "const-oid",
  "crypto-common 0.2.2",
  "ctutils",
 ]
@@ -1569,13 +1598,12 @@ dependencies = [
 
 [[package]]
 name = "etcetera"
-version = "0.8.0"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "136d1b5283a1ab77bd9257427ffd09d8667ced0570b6f938942bc7568ed5b943"
+checksum = "de48cc4d1c1d97a20fd819def54b890cadde72ed3ad0c614822a0a433361be96"
 dependencies = [
  "cfg-if",
- "home",
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -1623,9 +1651,9 @@ dependencies = [
 
 [[package]]
 name = "flume"
-version = "0.11.1"
+version = "0.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
+checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -1638,12 +1666,6 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
-[[package]]
-name = "foldhash"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
-
 [[package]]
 name = "foldhash"
 version = "0.2.0"
@@ -2042,17 +2064,6 @@ dependencies = [
  "warp",
 ]
 
-[[package]]
-name = "hashbrown"
-version = "0.15.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
-dependencies = [
- "allocator-api2",
- "equivalent",
- "foldhash 0.1.5",
-]
-
 [[package]]
 name = "hashbrown"
 version = "0.16.1"
@@ -2061,7 +2072,7 @@ checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
 dependencies = [
  "allocator-api2",
  "equivalent",
- "foldhash 0.2.0",
+ "foldhash",
 ]
 
 [[package]]
@@ -2072,11 +2083,11 @@ checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
 [[package]]
 name = "hashlink"
-version = "0.10.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1"
+checksum = "824e001ac4f3012dd16a264bec811403a67ca9deb6c102fc5049b32c4574b35f"
 dependencies = [
- "hashbrown 0.15.5",
+ "hashbrown 0.16.1",
 ]
 
 [[package]]
@@ -2091,7 +2102,7 @@ dependencies = [
  "http 1.5.0",
  "httpdate",
  "mime",
- "sha1",
+ "sha1 0.10.7",
 ]
 
 [[package]]
@@ -2123,11 +2134,11 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
 
 [[package]]
 name = "hkdf"
-version = "0.12.4"
+version = "0.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7"
+checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018"
 dependencies = [
- "hmac 0.12.1",
+ "hmac 0.13.0",
 ]
 
 [[package]]
@@ -2280,7 +2291,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tower-service",
- "webpki-roots 1.0.9",
+ "webpki-roots",
 ]
 
 [[package]]
@@ -2623,7 +2634,7 @@ dependencies = [
  "pem",
  "serde",
  "serde_json",
- "signature",
+ "signature 2.2.0",
  "simple_asn1",
  "zeroize",
 ]
@@ -2633,9 +2644,6 @@ name = "lazy_static"
 version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb"
-dependencies = [
- "spin",
-]
 
 [[package]]
 name = "lettre"
@@ -2662,7 +2670,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "url",
- "webpki-roots 1.0.9",
+ "webpki-roots",
 ]
 
 [[package]]
@@ -2695,22 +2703,13 @@ dependencies = [
  "rle-decode-fast",
 ]
 
-[[package]]
-name = "libm"
-version = "0.2.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
-
 [[package]]
 name = "libredox"
 version = "0.1.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d"
 dependencies = [
- "bitflags",
  "libc",
- "plain",
- "redox_syscall 0.9.4",
 ]
 
 [[package]]
@@ -2785,12 +2784,12 @@ dependencies = [
 
 [[package]]
 name = "md-5"
-version = "0.10.6"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
 dependencies = [
  "cfg-if",
- "digest 0.10.7",
+ "digest 0.11.3",
 ]
 
 [[package]]
@@ -2926,22 +2925,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "num-bigint-dig"
-version = "0.8.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7"
-dependencies = [
- "lazy_static",
- "libm",
- "num-integer",
- "num-iter",
- "num-traits",
- "rand 0.8.8",
- "smallvec",
- "zeroize",
-]
-
 [[package]]
 name = "num-conv"
 version = "0.2.2"
@@ -2957,16 +2940,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "num-iter"
-version = "0.1.46"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c92800bd69a1eac91786bcfe9da64a897eb72911b8dc3095decbd07429e8048b"
-dependencies = [
- "num-integer",
- "num-traits",
-]
-
 [[package]]
 name = "num-traits"
 version = "0.2.19"
@@ -2974,7 +2947,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
 dependencies = [
  "autocfg",
- "libm",
 ]
 
 [[package]]
@@ -3156,7 +3128,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
 dependencies = [
  "cfg-if",
  "libc",
- "redox_syscall 0.5.18",
+ "redox_syscall",
  "smallvec",
  "windows-link",
 ]
@@ -3193,9 +3165,9 @@ dependencies = [
 
 [[package]]
 name = "pem-rfc7468"
-version = "0.7.0"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412"
+checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9"
 dependencies = [
  "base64ct",
 ]
@@ -3278,20 +3250,19 @@ checksum = "13bee6c73da26345c729282832b60b0363cf3dd9f4bfd81d8551b7a1c889a113"
 
 [[package]]
 name = "pkcs1"
-version = "0.7.5"
+version = "0.8.0-rc.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f"
+checksum = "f485dbafe293ea6ccaa7d82bc591651a796eb8a9aec3df040b057b706073dbc3"
 dependencies = [
  "der",
- "pkcs8",
  "spki",
 ]
 
 [[package]]
 name = "pkcs8"
-version = "0.10.2"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
+checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7"
 dependencies = [
  "der",
  "spki",
@@ -3303,12 +3274,6 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548"
 
-[[package]]
-name = "plain"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
-
 [[package]]
 name = "poly1305"
 version = "0.8.0"
@@ -3698,15 +3663,6 @@ dependencies = [
  "bitflags",
 ]
 
-[[package]]
-name = "redox_syscall"
-version = "0.9.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "737970939a87c6fa31e7acad13307bccbb017a073b695b6089a2c484f929e20e"
-dependencies = [
- "bitflags",
-]
-
 [[package]]
 name = "redox_users"
 version = "0.4.6"
@@ -3794,7 +3750,7 @@ dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "web-sys",
- "webpki-roots 1.0.9",
+ "webpki-roots",
 ]
 
 [[package]]
@@ -3881,21 +3837,19 @@ dependencies = [
 
 [[package]]
 name = "rsa"
-version = "0.9.10"
+version = "0.10.0-rc.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d"
+checksum = "33405e321e16ef941ca66e254f6904c3aafd3c7ff001fb88bf11c73eb75c2b85"
 dependencies = [
- "const-oid 0.9.6",
- "digest 0.10.7",
- "num-bigint-dig",
- "num-integer",
- "num-traits",
+ "const-oid",
+ "crypto-bigint",
+ "crypto-primes",
+ "digest 0.11.3",
  "pkcs1",
  "pkcs8",
- "rand_core 0.6.4",
- "signature",
+ "rand_core 0.10.1",
+ "signature 3.0.0",
  "spki",
- "subtle",
  "zeroize",
 ]
 
@@ -4173,6 +4127,16 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serdect"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66cf8fedced2fcf12406bcb34223dffb92eaf34908ede12fed414c82b7f00b3e"
+dependencies = [
+ "base16ct",
+ "serde",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.7"
@@ -4184,6 +4148,17 @@ dependencies = [
  "digest 0.10.7",
 ]
 
+[[package]]
+name = "sha1"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.3.1",
+ "digest 0.11.3",
+]
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -4237,10 +4212,19 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de"
 dependencies = [
- "digest 0.10.7",
  "rand_core 0.6.4",
 ]
 
+[[package]]
+name = "signature"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "28d567dcbaf0049cb8ac2608a76cd95ff9e4412e1899d389ee400918ca7537f5"
+dependencies = [
+ "digest 0.11.3",
+ "rand_core 0.10.1",
+]
+
 [[package]]
 name = "simd-adler32"
 version = "0.3.10"
@@ -4317,9 +4301,9 @@ dependencies = [
 
 [[package]]
 name = "spki"
-version = "0.7.3"
+version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
+checksum = "7ef958a98b9d5da290cfc78946e9f3e61e1e62a18db0d92cac0b83cc161491a9"
 dependencies = [
  "base64ct",
  "der",
@@ -4327,9 +4311,9 @@ dependencies = [
 
 [[package]]
 name = "sqlx"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc"
+checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71"
 dependencies = [
  "sqlx-core",
  "sqlx-macros",
@@ -4340,12 +4324,13 @@ dependencies = [
 
 [[package]]
 name = "sqlx-core"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6"
+checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb"
 dependencies = [
  "base64 0.22.1",
  "bytes",
+ "cfg-if",
  "chrono",
  "crc",
  "crossbeam-queue",
@@ -4355,12 +4340,11 @@ dependencies = [
  "futures-intrusive",
  "futures-io",
  "futures-util",
- "hashbrown 0.15.5",
+ "hashbrown 0.16.1",
  "hashlink",
  "indexmap",
  "log",
  "memchr",
- "once_cell",
  "percent-encoding",
  "rustls",
  "serde",
@@ -4373,14 +4357,14 @@ dependencies = [
  "tracing",
  "url",
  "uuid",
- "webpki-roots 0.26.11",
+ "webpki-roots",
 ]
 
 [[package]]
 name = "sqlx-macros"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d"
+checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4391,15 +4375,15 @@ dependencies = [
 
 [[package]]
 name = "sqlx-macros-core"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b"
+checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3"
 dependencies = [
+ "cfg-if",
  "dotenvy",
  "either",
  "heck 0.5.0",
  "hex",
- "once_cell",
  "proc-macro2",
  "quote",
  "serde",
@@ -4410,59 +4394,46 @@ dependencies = [
  "sqlx-postgres",
  "sqlx-sqlite",
  "syn 2.0.119",
+ "thiserror 2.0.21",
  "tokio",
  "url",
 ]
 
 [[package]]
 name = "sqlx-mysql"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526"
+checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27"
 dependencies = [
- "atoi",
- "base64 0.22.1",
  "bitflags",
  "byteorder",
  "bytes",
  "chrono",
  "crc",
- "digest 0.10.7",
+ "digest 0.11.3",
  "dotenvy",
  "either",
- "futures-channel",
  "futures-core",
- "futures-io",
  "futures-util",
  "generic-array",
- "hex",
- "hkdf",
- "hmac 0.12.1",
- "itoa",
  "log",
- "md-5",
- "memchr",
- "once_cell",
  "percent-encoding",
- "rand 0.8.8",
+ "rand 0.10.3",
  "rsa",
  "serde",
- "sha1",
- "sha2 0.10.9",
- "smallvec",
+ "sha1 0.11.0",
+ "sha2 0.11.0",
  "sqlx-core",
- "stringprep",
  "thiserror 2.0.21",
  "tracing",
  "uuid",
- "whoami",
 ]
 
 [[package]]
 name = "sqlx-postgres"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46"
+checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e"
 dependencies = [
  "atoi",
  "base64 0.22.1",
@@ -4477,17 +4448,15 @@ dependencies = [
  "futures-util",
  "hex",
  "hkdf",
- "hmac 0.12.1",
- "home",
+ "hmac 0.13.0",
  "itoa",
  "log",
  "md-5",
  "memchr",
- "once_cell",
- "rand 0.8.8",
+ "rand 0.10.3",
  "serde",
  "serde_json",
- "sha2 0.10.9",
+ "sha2 0.11.0",
  "smallvec",
  "sqlx-core",
  "stringprep",
@@ -4499,13 +4468,14 @@ dependencies = [
 
 [[package]]
 name = "sqlx-sqlite"
-version = "0.8.6"
+version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea"
+checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c"
 dependencies = [
  "atoi",
  "chrono",
  "flume",
+ "form_urlencoded",
  "futures-channel",
  "futures-core",
  "futures-executor",
@@ -4515,7 +4485,6 @@ dependencies = [
  "log",
  "percent-encoding",
  "serde",
- "serde_urlencoded",
  "sqlx-core",
  "thiserror 2.0.21",
  "tracing",
@@ -4995,7 +4964,7 @@ dependencies = [
  "tower-layer",
  "tower-service",
  "tracing",
- "webpki-roots 1.0.9",
+ "webpki-roots",
 ]
 
 [[package]]
@@ -5184,7 +5153,7 @@ dependencies = [
  "httparse",
  "log",
  "rand 0.8.8",
- "sha1",
+ "sha1 0.10.7",
  "thiserror 1.0.69",
  "url",
  "utf-8",
@@ -5202,7 +5171,7 @@ dependencies = [
  "httparse",
  "log",
  "rand 0.9.5",
- "sha1",
+ "sha1 0.10.7",
  "thiserror 2.0.21",
 ]
 
@@ -5519,12 +5488,6 @@ dependencies = [
  "wit-bindgen",
 ]
 
-[[package]]
-name = "wasite"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
-
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.129"
@@ -5623,15 +5586,6 @@ dependencies = [
  "rustls-pki-types",
 ]
 
-[[package]]
-name = "webpki-roots"
-version = "0.26.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
-dependencies = [
- "webpki-roots 1.0.9",
-]
-
 [[package]]
 name = "webpki-roots"
 version = "1.0.9"
@@ -5643,13 +5597,9 @@ dependencies = [
 
 [[package]]
 name = "whoami"
-version = "1.6.1"
+version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d"
-dependencies = [
- "libredox",
- "wasite",
-]
+checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c"
 
 [[package]]
 name = "winapi"
diff --git a/Cargo.toml b/Cargo.toml
index 5f13c45..30b5f50 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,12 +16,12 @@ repository = "https://github.com/CodingAnarchy/hammerwork"
 authors = ["CodingAnarchy <noreply@codinganarchy.com>"]
 homepage = "https://github.com/CodingAnarchy/hammerwork"
 documentation = "https://docs.rs/hammerwork"
-rust-version = "1.91.1"
+rust-version = "1.94.0"
 
 [workspace.dependencies]
 hammerwork = { version = "2.5.0", path = "." }
 tokio = { version = "1.0", features = ["full"] }
-sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "chrono", "uuid", "json"] }
+sqlx = { version = "0.9", features = ["runtime-tokio", "tls-rustls-ring-webpki", "chrono", "uuid", "json"] }
 chrono = { version = "0.4", features = ["serde"] }
 chrono-tz = "0.8"
 cron = "0.12"
diff --git a/README.md b/README.md
index 7a2e8de..fa6c41c 100644
--- a/README.md
+++ b/README.md
@@ -97,6 +97,7 @@ See the [Quick Start Guide](docs/quick-start.md) for complete examples with Post
 
 ## Documentation
 
+- **[Upgrading to 3.0](docs/upgrading-to-3.0.md)** - Breaking changes from 2.x and how to migrate (see also the [CHANGELOG](CHANGELOG.md))
 - **[Quick Start Guide](docs/quick-start.md)** - Get started with PostgreSQL and MySQL
 - **[TestQueue Framework](docs/testing.md)** - In-memory testing with MockClock for unit tests and time control
 - **[Web Dashboard](hammerwork-web/README.md)** - Real-time web interface for queue monitoring and job management
diff --git a/cargo-hammerwork/Cargo.toml b/cargo-hammerwork/Cargo.toml
index d794111..74a5942 100644
--- a/cargo-hammerwork/Cargo.toml
+++ b/cargo-hammerwork/Cargo.toml
@@ -23,7 +23,7 @@ tokio = { workspace = true }
 anyhow = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
-sqlx = { workspace = true, features = ["postgres", "mysql"] }
+sqlx = { workspace = true, features = ["postgres", "mysql", "mysql-rsa"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 chrono = { workspace = true }
diff --git a/cargo-hammerwork/src/commands/backup.rs b/cargo-hammerwork/src/commands/backup.rs
index d994b0e..d49175b 100644
--- a/cargo-hammerwork/src/commands/backup.rs
+++ b/cargo-hammerwork/src/commands/backup.rs
@@ -208,18 +208,20 @@ async fn fetch_backup_rows(
         include_failed,
     );
     match pool {
-        DatabasePool::Postgres(pg_pool) => bind_pg(sqlx::query(&query), &binds)
+        DatabasePool::Postgres(pg_pool) => bind_pg(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
             .fetch_all(pg_pool)
             .await?
             .iter()
             .map(|row| job_rows::pg_row_to_json(row, columns))
             .collect(),
-        DatabasePool::MySQL(mysql_pool) => bind_mysql(sqlx::query(&query), &binds)
-            .fetch_all(mysql_pool)
-            .await?
-            .iter()
-            .map(|row| job_rows::mysql_row_to_json(row, columns))
-            .collect(),
+        DatabasePool::MySQL(mysql_pool) => {
+            bind_mysql(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
+                .fetch_all(mysql_pool)
+                .await?
+                .iter()
+                .map(|row| job_rows::mysql_row_to_json(row, columns))
+                .collect()
+        }
     }
 }
 
@@ -435,14 +437,18 @@ async fn check_job_exists(pool: &DatabasePool, id: &str) -> Result<bool> {
     let id = params.uuid(id);
     let sql = format!("SELECT 1 AS present FROM hammerwork_jobs WHERE id = {id}");
     Ok(match pool {
-        DatabasePool::Postgres(pg_pool) => bind_pg(sqlx::query(&sql), params.binds())
-            .fetch_optional(pg_pool)
-            .await?
-            .is_some(),
-        DatabasePool::MySQL(mysql_pool) => bind_mysql(sqlx::query(&sql), params.binds())
-            .fetch_optional(mysql_pool)
-            .await?
-            .is_some(),
+        DatabasePool::Postgres(pg_pool) => {
+            bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
+                .fetch_optional(pg_pool)
+                .await?
+                .is_some()
+        }
+        DatabasePool::MySQL(mysql_pool) => {
+            bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
+                .fetch_optional(mysql_pool)
+                .await?
+                .is_some()
+        }
     })
 }
 
@@ -1363,9 +1369,9 @@ mod tests {
                     })
                     .collect::<Vec<_>>()
                     .join(", ");
-                sqlx::query_scalar(&format!(
+                sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
                     "SELECT CONCAT_WS('|', {parts}) FROM hammerwork_jobs WHERE id = ?"
-                ))
+                )))
                 .bind(id)
                 .fetch_one(p)
                 .await
diff --git a/cargo-hammerwork/src/commands/cron.rs b/cargo-hammerwork/src/commands/cron.rs
index a6359d5..eebd70c 100644
--- a/cargo-hammerwork/src/commands/cron.rs
+++ b/cargo-hammerwork/src/commands/cron.rs
@@ -279,12 +279,18 @@ pub async fn fetch_cron_jobs(
     let mut out = Vec::new();
     match pool {
         DatabasePool::Postgres(pg) => {
-            for row in bind_pg(sqlx::query(&sql), &binds).fetch_all(pg).await? {
+            for row in bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), &binds)
+                .fetch_all(pg)
+                .await?
+            {
                 out.push(cron_row!(&row));
             }
         }
         DatabasePool::MySQL(my) => {
-            for row in bind_mysql(sqlx::query(&sql), &binds).fetch_all(my).await? {
+            for row in bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), &binds)
+                .fetch_all(my)
+                .await?
+            {
                 out.push(cron_row!(&row));
             }
         }
diff --git a/cargo-hammerwork/src/commands/maintenance.rs b/cargo-hammerwork/src/commands/maintenance.rs
index a6dc435..02afbb6 100644
--- a/cargo-hammerwork/src/commands/maintenance.rs
+++ b/cargo-hammerwork/src/commands/maintenance.rs
@@ -175,8 +175,11 @@ async fn count_expired_encrypted_jobs(pool: &DatabasePool) -> Result<(i64, i64)>
         DatabasePool::Postgres(pg) => {
             let (jobs, archived) = expired_encrypted_count_queries("$1");
             (
-                sqlx::query_scalar(&jobs).bind(now).fetch_one(pg).await?,
-                sqlx::query_scalar(&archived)
+                sqlx::query_scalar(sqlx::AssertSqlSafe(jobs))
+                    .bind(now)
+                    .fetch_one(pg)
+                    .await?,
+                sqlx::query_scalar(sqlx::AssertSqlSafe(archived))
                     .bind(now)
                     .fetch_one(pg)
                     .await?,
@@ -185,8 +188,11 @@ async fn count_expired_encrypted_jobs(pool: &DatabasePool) -> Result<(i64, i64)>
         DatabasePool::MySQL(my) => {
             let (jobs, archived) = expired_encrypted_count_queries("?");
             (
-                sqlx::query_scalar(&jobs).bind(now).fetch_one(my).await?,
-                sqlx::query_scalar(&archived)
+                sqlx::query_scalar(sqlx::AssertSqlSafe(jobs))
+                    .bind(now)
+                    .fetch_one(my)
+                    .await?,
+                sqlx::query_scalar(sqlx::AssertSqlSafe(archived))
                     .bind(now)
                     .fetch_one(my)
                     .await?,
@@ -419,9 +425,12 @@ async fn reindex_database(pool: DatabasePool, confirm: bool) -> Result<()> {
             .await?;
             for table in &tables {
                 // Names come from the catalog; quote them anyway.
-                sqlx::query(&format!("REINDEX TABLE \"{}\"", table.replace('"', "\"\"")))
-                    .execute(pg_pool)
-                    .await?;
+                sqlx::query(sqlx::AssertSqlSafe(format!(
+                    "REINDEX TABLE \"{}\"",
+                    table.replace('"', "\"\"")
+                )))
+                .execute(pg_pool)
+                .await?;
             }
 
             println!("✅ PostgreSQL indexes rebuilt ({} tables)", tables.len());
diff --git a/cargo-hammerwork/src/commands/monitor.rs b/cargo-hammerwork/src/commands/monitor.rs
index ff402df..cd9aadc 100644
--- a/cargo-hammerwork/src/commands/monitor.rs
+++ b/cargo-hammerwork/src/commands/monitor.rs
@@ -508,10 +508,13 @@ pub async fn fetch_metrics(
 
     Ok(match pool {
         DatabasePool::Postgres(pg_pool) => {
-            let t = bind_pg(sqlx::query(&throughput_query), &throughput_binds)
-                .fetch_one(pg_pool)
-                .await?;
-            let a = bind_pg(sqlx::query(&avg_query), &avg_binds)
+            let t = bind_pg(
+                sqlx::query(sqlx::AssertSqlSafe(throughput_query)),
+                &throughput_binds,
+            )
+            .fetch_one(pg_pool)
+            .await?;
+            let a = bind_pg(sqlx::query(sqlx::AssertSqlSafe(avg_query)), &avg_binds)
                 .fetch_one(pg_pool)
                 .await?;
             Metrics {
@@ -522,10 +525,13 @@ pub async fn fetch_metrics(
             }
         }
         DatabasePool::MySQL(mysql_pool) => {
-            let t = bind_mysql(sqlx::query(&throughput_query), &throughput_binds)
-                .fetch_one(mysql_pool)
-                .await?;
-            let a = bind_mysql(sqlx::query(&avg_query), &avg_binds)
+            let t = bind_mysql(
+                sqlx::query(sqlx::AssertSqlSafe(throughput_query)),
+                &throughput_binds,
+            )
+            .fetch_one(mysql_pool)
+            .await?;
+            let a = bind_mysql(sqlx::query(sqlx::AssertSqlSafe(avg_query)), &avg_binds)
                 .fetch_one(mysql_pool)
                 .await?;
             Metrics {
diff --git a/cargo-hammerwork/src/commands/spawn.rs b/cargo-hammerwork/src/commands/spawn.rs
index 130b0d8..add5f30 100644
--- a/cargo-hammerwork/src/commands/spawn.rs
+++ b/cargo-hammerwork/src/commands/spawn.rs
@@ -379,7 +379,7 @@ pub async fn fetch_spawn_list(
     let mut out = Vec::new();
     match pool {
         DatabasePool::Postgres(pg_pool) => {
-            for row in bind_pg(sqlx::query(&query), &binds)
+            for row in bind_pg(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
                 .fetch_all(pg_pool)
                 .await?
             {
@@ -397,7 +397,7 @@ pub async fn fetch_spawn_list(
             }
         }
         DatabasePool::MySQL(mysql_pool) => {
-            for row in bind_mysql(sqlx::query(&query), &binds)
+            for row in bind_mysql(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
                 .fetch_all(mysql_pool)
                 .await?
             {
@@ -486,7 +486,7 @@ pub async fn fetch_spawn_stats(
 
     macro_rules! stats {
         ($pool:expr, $bind:ident) => {{
-            let row = $bind(sqlx::query(&total_query), &total_binds)
+            let row = $bind(sqlx::query(sqlx::AssertSqlSafe(total_query)), &total_binds)
                 .fetch_one($pool)
                 .await?;
             let mut stats = SpawnStats {
@@ -497,9 +497,12 @@ pub async fn fetch_spawn_stats(
             };
             if detailed {
                 let mut lines = Vec::new();
-                for row in $bind(sqlx::query(&breakdown_query), &breakdown_binds)
-                    .fetch_all($pool)
-                    .await?
+                for row in $bind(
+                    sqlx::query(sqlx::AssertSqlSafe(breakdown_query)),
+                    &breakdown_binds,
+                )
+                .fetch_all($pool)
+                .await?
                 {
                     lines.push(QueueSpawnStats {
                         queue_name: row.try_get("queue_name")?,
@@ -570,7 +573,7 @@ pub async fn fetch_pending_spawns(
     let mut out = Vec::new();
     match pool {
         DatabasePool::Postgres(pg_pool) => {
-            for row in bind_pg(sqlx::query(&query), &binds)
+            for row in bind_pg(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
                 .fetch_all(pg_pool)
                 .await?
             {
@@ -584,7 +587,7 @@ pub async fn fetch_pending_spawns(
             }
         }
         DatabasePool::MySQL(mysql_pool) => {
-            for row in bind_mysql(sqlx::query(&query), &binds)
+            for row in bind_mysql(sqlx::query(sqlx::AssertSqlSafe(query)), &binds)
                 .fetch_all(mysql_pool)
                 .await?
             {
diff --git a/cargo-hammerwork/src/commands/worker.rs b/cargo-hammerwork/src/commands/worker.rs
index ead5477..98e6281 100644
--- a/cargo-hammerwork/src/commands/worker.rs
+++ b/cargo-hammerwork/src/commands/worker.rs
@@ -241,7 +241,7 @@ pub async fn fetch_running_jobs(
                 sql.push_str(" AND queue_name = $1");
             }
             sql.push_str(" ORDER BY queue_name, started_at");
-            let mut q = sqlx::query(&sql);
+            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
             if let Some(name) = queue {
                 q = q.bind(name.to_string());
             }
@@ -272,7 +272,7 @@ pub async fn fetch_running_jobs(
                 sql.push_str(" AND queue_name = ?");
             }
             sql.push_str(" ORDER BY queue_name, started_at");
-            let mut q = sqlx::query(&sql);
+            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
             if let Some(name) = queue {
                 q = q.bind(name.to_string());
             }
@@ -407,7 +407,7 @@ mod tests {
             };
             match &pool {
                 DatabasePool::Postgres(p) => {
-                    sqlx::query(&sql)
+                    sqlx::query(sqlx::AssertSqlSafe(sql))
                         .bind(id)
                         .bind(&queue)
                         .execute(p)
@@ -415,7 +415,7 @@ mod tests {
                         .unwrap();
                 }
                 DatabasePool::MySQL(p) => {
-                    sqlx::query(&sql)
+                    sqlx::query(sqlx::AssertSqlSafe(sql))
                         .bind(id)
                         .bind(&queue)
                         .execute(p)
diff --git a/cargo-hammerwork/src/commands/workflow.rs b/cargo-hammerwork/src/commands/workflow.rs
index 67cc7af..e2365b8 100644
--- a/cargo-hammerwork/src/commands/workflow.rs
+++ b/cargo-hammerwork/src/commands/workflow.rs
@@ -174,14 +174,14 @@ pub async fn fetch_workflows(
     Ok(match pool {
         DatabasePool::Postgres(p) => {
             summaries!(
-                bind_pg(sqlx::query(&sql), params.binds())
+                bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
                     .fetch_all(p)
                     .await?
             )
         }
         DatabasePool::MySQL(p) => {
             summaries!(
-                bind_mysql(sqlx::query(&sql), params.binds())
+                bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
                     .fetch_all(p)
                     .await?
             )
diff --git a/cargo-hammerwork/src/utils/job_ops.rs b/cargo-hammerwork/src/utils/job_ops.rs
index 29ee1b6..a48adc4 100644
--- a/cargo-hammerwork/src/utils/job_ops.rs
+++ b/cargo-hammerwork/src/utils/job_ops.rs
@@ -134,13 +134,17 @@ pub async fn select_job_ids(pool: &DatabasePool, selector: &JobSelector) -> Resu
     let (sql, binds) = selector.sql(pool.backend());
     match pool {
         DatabasePool::Postgres(pg) => {
-            let rows = bind_pg(sqlx::query(&sql), &binds).fetch_all(pg).await?;
+            let rows = bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), &binds)
+                .fetch_all(pg)
+                .await?;
             rows.iter()
                 .map(|row| row.try_get::<JobId, _>("id").map_err(Into::into))
                 .collect()
         }
         DatabasePool::MySQL(my) => {
-            let rows = bind_mysql(sqlx::query(&sql), &binds).fetch_all(my).await?;
+            let rows = bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), &binds)
+                .fetch_all(my)
+                .await?;
             rows.iter()
                 .map(|row| {
                     let id: String = row.try_get("id")?;
diff --git a/cargo-hammerwork/src/utils/job_rows.rs b/cargo-hammerwork/src/utils/job_rows.rs
index 305cc46..d758f0d 100644
--- a/cargo-hammerwork/src/utils/job_rows.rs
+++ b/cargo-hammerwork/src/utils/job_rows.rs
@@ -444,7 +444,7 @@ pub async fn insert_rows(pool: &DatabasePool, rows: &[RowInsert]) -> Result<usiz
             let mut tx = p.begin().await?;
             for row in rows {
                 let sql = row.sql(Backend::Postgres);
-                let affected = bind_pg(sqlx::query(&sql), &row.params)
+                let affected = bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), &row.params)
                     .execute(&mut *tx)
                     .await
                     .map_err(|e| anyhow!("Cannot restore job {}: {}", row.id, e))?
@@ -457,7 +457,7 @@ pub async fn insert_rows(pool: &DatabasePool, rows: &[RowInsert]) -> Result<usiz
             let mut tx = p.begin().await?;
             for row in rows {
                 let sql = row.sql(Backend::MySql);
-                match bind_mysql(sqlx::query(&sql), &row.params)
+                match bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), &row.params)
                     .execute(&mut *tx)
                     .await
                 {
diff --git a/cargo-hammerwork/src/utils/sql.rs b/cargo-hammerwork/src/utils/sql.rs
index f72abe4..792867a 100644
--- a/cargo-hammerwork/src/utils/sql.rs
+++ b/cargo-hammerwork/src/utils/sql.rs
@@ -189,11 +189,15 @@ pub async fn fetch_i64(
 ) -> Result<i64> {
     match pool {
         DatabasePool::Postgres(p) => {
-            let row = bind_pg(sqlx::query(sql), binds).fetch_one(p).await?;
+            let row = bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
+                .fetch_one(p)
+                .await?;
             Ok(row.try_get(column)?)
         }
         DatabasePool::MySQL(p) => {
-            let row = bind_mysql(sqlx::query(sql), binds).fetch_one(p).await?;
+            let row = bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
+                .fetch_one(p)
+                .await?;
             Ok(row.try_get(column)?)
         }
     }
@@ -202,11 +206,11 @@ pub async fn fetch_i64(
 /// Run a write statement and return the number of affected rows.
 pub async fn execute_binds(pool: &DatabasePool, sql: &str, binds: &[Bind]) -> Result<u64> {
     match pool {
-        DatabasePool::Postgres(p) => Ok(bind_pg(sqlx::query(sql), binds)
+        DatabasePool::Postgres(p) => Ok(bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
             .execute(p)
             .await?
             .rows_affected()),
-        DatabasePool::MySQL(p) => Ok(bind_mysql(sqlx::query(sql), binds)
+        DatabasePool::MySQL(p) => Ok(bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
             .execute(p)
             .await?
             .rows_affected()),
@@ -223,9 +227,14 @@ macro_rules! fetch_rows {
     ($pool:expr, $sql:expr, $binds:expr, |$row:ident| $map:expr) => {
         match $pool {
             $crate::utils::database::DatabasePool::Postgres(pool) => {
-                let rows = $crate::utils::sql::bind_pg(sqlx::query($sql), $binds)
-                    .fetch_all(pool)
-                    .await?;
+                let rows = $crate::utils::sql::bind_pg(
+                    sqlx::query(sqlx::AssertSqlSafe(::std::convert::AsRef::<str>::as_ref(
+                        $sql,
+                    ))),
+                    $binds,
+                )
+                .fetch_all(pool)
+                .await?;
                 let mut mapped = Vec::with_capacity(rows.len());
                 for $row in rows {
                     mapped.push($map);
@@ -233,9 +242,14 @@ macro_rules! fetch_rows {
                 mapped
             }
             $crate::utils::database::DatabasePool::MySQL(pool) => {
-                let rows = $crate::utils::sql::bind_mysql(sqlx::query($sql), $binds)
-                    .fetch_all(pool)
-                    .await?;
+                let rows = $crate::utils::sql::bind_mysql(
+                    sqlx::query(sqlx::AssertSqlSafe(::std::convert::AsRef::<str>::as_ref(
+                        $sql,
+                    ))),
+                    $binds,
+                )
+                .fetch_all(pool)
+                .await?;
                 let mut mapped = Vec::with_capacity(rows.len());
                 for $row in rows {
                     mapped.push($map);
diff --git a/cargo-hammerwork/src/utils/test_support.rs b/cargo-hammerwork/src/utils/test_support.rs
index 0a7562b..0c52fd8 100644
--- a/cargo-hammerwork/src/utils/test_support.rs
+++ b/cargo-hammerwork/src/utils/test_support.rs
@@ -88,13 +88,13 @@ impl ScratchDb {
             .unwrap();
         match &admin {
             DatabasePool::Postgres(p) => {
-                sqlx::query(&format!("CREATE DATABASE {name}"))
+                sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}")))
                     .execute(p)
                     .await
                     .unwrap();
             }
             DatabasePool::MySQL(p) => {
-                sqlx::query(&format!("CREATE DATABASE {name}"))
+                sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}")))
                     .execute(p)
                     .await
                     .unwrap();
@@ -138,7 +138,7 @@ impl ScratchDb {
                     .await
                     .ok();
                 for t in tables {
-                    sqlx::query(&format!("DROP TABLE `{t}`"))
+                    sqlx::query(sqlx::AssertSqlSafe(format!("DROP TABLE `{t}`")))
                         .execute(p)
                         .await
                         .unwrap();
@@ -156,20 +156,23 @@ impl ScratchDb {
         match (&self.pool, &self.admin) {
             (DatabasePool::Postgres(p), DatabasePool::Postgres(a)) => {
                 p.close().await;
-                sqlx::query(&format!(
+                sqlx::query(sqlx::AssertSqlSafe(format!(
                     "DROP DATABASE IF EXISTS {} WITH (FORCE)",
                     self.name
-                ))
+                )))
                 .execute(a)
                 .await
                 .unwrap();
             }
             (DatabasePool::MySQL(p), DatabasePool::MySQL(a)) => {
                 p.close().await;
-                sqlx::query(&format!("DROP DATABASE IF EXISTS {}", self.name))
-                    .execute(a)
-                    .await
-                    .unwrap();
+                sqlx::query(sqlx::AssertSqlSafe(format!(
+                    "DROP DATABASE IF EXISTS {}",
+                    self.name
+                )))
+                .execute(a)
+                .await
+                .unwrap();
             }
             _ => unreachable!(),
         }
@@ -206,12 +209,12 @@ pub async fn job_column(pool: &DatabasePool, id: &str, column: &str) -> Option<S
     };
     let sql = format!("SELECT {cast} AS v FROM hammerwork_jobs WHERE id = {id}");
     match pool {
-        DatabasePool::Postgres(p) => bind_pg(sqlx::query(&sql), params.binds())
+        DatabasePool::Postgres(p) => bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
             .fetch_optional(p)
             .await
             .unwrap()
             .and_then(|r| r.try_get::<Option<String>, _>("v").unwrap()),
-        DatabasePool::MySQL(p) => bind_mysql(sqlx::query(&sql), params.binds())
+        DatabasePool::MySQL(p) => bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), params.binds())
             .fetch_optional(p)
             .await
             .unwrap()
@@ -298,7 +301,7 @@ pub async fn seed(pool: &DatabasePool, job: &SeedJob<'_>) -> String {
                 CASE WHEN $8 THEN '0 0 * * * *' ELSE NULL END, $8, \
                 CASE WHEN $8 THEN NOW() + INTERVAL '1 hour' ELSE NULL END, \
                 CASE WHEN $9::text IS NULL THEN '{}'::uuid[] ELSE ARRAY[$9::uuid] END)";
-            sqlx::query(sql)
+            sqlx::query(sqlx::AssertSqlSafe(sql))
                 .bind(&id)
                 .bind(job.queue)
                 .bind(job.payload)
@@ -322,7 +325,7 @@ pub async fn seed(pool: &DatabasePool, job: &SeedJob<'_>) -> String {
                 IF(?, '0 0 * * * *', NULL), ?, \
                 IF(?, DATE_ADD(UTC_TIMESTAMP(6), INTERVAL 1 HOUR), NULL), \
                 IF(? IS NULL, JSON_ARRAY(), JSON_ARRAY(?)))";
-            sqlx::query(sql)
+            sqlx::query(sqlx::AssertSqlSafe(sql))
                 .bind(&id)
                 .bind(job.queue)
                 .bind(job.payload)
@@ -373,14 +376,14 @@ pub async fn column_strings(
     column: &str,
 ) -> Vec<String> {
     match pool {
-        DatabasePool::Postgres(p) => bind_pg(sqlx::query(sql), binds)
+        DatabasePool::Postgres(p) => bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
             .fetch_all(p)
             .await
             .unwrap()
             .iter()
             .map(|r| r.try_get::<String, _>(column).unwrap())
             .collect(),
-        DatabasePool::MySQL(p) => bind_mysql(sqlx::query(sql), binds)
+        DatabasePool::MySQL(p) => bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql)), binds)
             .fetch_all(p)
             .await
             .unwrap()
@@ -407,8 +410,16 @@ pub async fn table_exists(pool: &DatabasePool) -> bool {
 /// Run a raw statement (test fixtures only; never build these from user input).
 pub async fn exec_sql(pool: &DatabasePool, sql: &str) -> u64 {
     match pool {
-        DatabasePool::Postgres(p) => sqlx::query(sql).execute(p).await.unwrap().rows_affected(),
-        DatabasePool::MySQL(p) => sqlx::query(sql).execute(p).await.unwrap().rows_affected(),
+        DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
+            .execute(p)
+            .await
+            .unwrap()
+            .rows_affected(),
+        DatabasePool::MySQL(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
+            .execute(p)
+            .await
+            .unwrap()
+            .rows_affected(),
     }
 }
 
diff --git a/cargo-hammerwork/tests/sql_query_tests.rs b/cargo-hammerwork/tests/sql_query_tests.rs
index 1f9aa8c..c978d4d 100644
--- a/cargo-hammerwork/tests/sql_query_tests.rs
+++ b/cargo-hammerwork/tests/sql_query_tests.rs
@@ -540,13 +540,13 @@ mod builder_tests {
             if name.starts_with("job_purge") || name == "vacuum" && sql.starts_with("DELETE") {
                 // Writes: run them inside a rolled-back transaction
                 let mut tx = pool.begin().await?;
-                bind_pg(sqlx::query(&sql), &binds)
+                bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql.as_str())), &binds)
                     .execute(&mut *tx)
                     .await
                     .unwrap_or_else(|e| panic!("{name}: {e}\n{sql}"));
                 tx.rollback().await?;
             } else {
-                bind_pg(sqlx::query(&sql), &binds)
+                bind_pg(sqlx::query(sqlx::AssertSqlSafe(sql.as_str())), &binds)
                     .fetch_all(&pool)
                     .await
                     .unwrap_or_else(|e| panic!("{name}: {e}\n{sql}"));
@@ -573,13 +573,13 @@ mod builder_tests {
         for (name, sql, binds) in all {
             if name.starts_with("job_purge") || name == "vacuum" && sql.starts_with("DELETE") {
                 let mut tx = pool.begin().await?;
-                bind_mysql(sqlx::query(&sql), &binds)
+                bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql.as_str())), &binds)
                     .execute(&mut *tx)
                     .await
                     .unwrap_or_else(|e| panic!("{name}: {e}\n{sql}"));
                 tx.rollback().await?;
             } else {
-                bind_mysql(sqlx::query(&sql), &binds)
+                bind_mysql(sqlx::query(sqlx::AssertSqlSafe(sql.as_str())), &binds)
                     .fetch_all(&pool)
                     .await
                     .unwrap_or_else(|e| panic!("{name}: {e}\n{sql}"));
diff --git a/docs/quick-start.md b/docs/quick-start.md
index a607772..64ac586 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -32,7 +32,9 @@ hammerwork = { version = "2.5", features = ["postgres", "tracing"] }
 # Metrics, alerting, webhooks, streaming, tracing and encryption together
 hammerwork = { version = "2.5", features = ["postgres", "full"] }
 
-# Additional dependencies for examples
+# Additional dependencies for examples. Your sqlx must be the same major
+# version as Hammerwork's (0.9), since you pass Hammerwork a sqlx pool.
+sqlx = { version = "0.9", features = ["runtime-tokio", "postgres"] }  # or "mysql"
 serde_json = "1.0"
 tokio = { version = "1.0", features = ["full"] }
 tracing = "0.1"
diff --git a/docs/upgrading-to-3.0.md b/docs/upgrading-to-3.0.md
index 14ae72c..f8f0fa0 100644
--- a/docs/upgrading-to-3.0.md
+++ b/docs/upgrading-to-3.0.md
@@ -1,6 +1,103 @@
 # Upgrading from 2.x to 3.0
 
-This page lists what to change in your project when you move from Hammerwork 2.x to 3.0.
+Hammerwork 3.0 is a major release. This guide lists each breaking change, what it
+affects and what you need to change in your code, configuration or deployment. Read
+it together with the `3.0.0` entry in the [CHANGELOG](../CHANGELOG.md).
+
+## sqlx 0.9 and Rust 1.94
+
+Hammerwork 3.0 depends on [sqlx](https://crates.io/crates/sqlx) 0.9 instead of 0.8,
+and its minimum supported Rust version (MSRV) is now **1.94.0** (it was 1.91.1).
+sqlx 0.9 requires Rust 1.94.
+
+sqlx is part of Hammerwork's public API. You create the `sqlx::Pool` that
+`JobQueue::new` takes, `JobQueue<DB>` and `DatabaseQueue` are generic over
+`sqlx::Database`, the `MigrationRunner`s take sqlx pools, and `JobStatus` and
+`BatchStatus` implement sqlx's `Type`, `Encode` and `Decode`. Your application must
+therefore use the same sqlx major version as Hammerwork. If it still depends on sqlx
+0.8, the build fails with errors such as "expected `Pool<Postgres>`, found
+`Pool<Postgres>`", because two different versions of the `sqlx` crate are in the
+dependency graph.
+
+### What to change
+
+1. **Use Rust 1.94.0 or newer.**
+
+2. **Upgrade your own sqlx dependency to 0.9.** sqlx 0.9 removed the combined
+   runtime and TLS features, so name them separately:
+
+   ```toml
+   # Before (2.x)
+   sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres"] }
+
+   # After (3.0)
+   sqlx = { version = "0.9", features = ["runtime-tokio", "tls-rustls-ring-webpki", "postgres"] }
+   ```
+
+   `runtime-tokio-native-tls` becomes `runtime-tokio` plus `tls-native-tls`.
+   Hammerwork itself enables `runtime-tokio` and `tls-rustls-ring-webpki`.
+
+3. **Wrap dynamic SQL strings in your own queries.** sqlx 0.9's `query`,
+   `query_as`, `query_scalar`, `raw_sql` and `Executor::execute` accept only string
+   literals (`&'static str`) unless you mark the string as reviewed:
+
+   ```rust,no_run
+   # async fn example(pool: sqlx::PgPool, table: &str) -> Result<(), sqlx::Error> {
+   // Before (no longer compiles):
+   // sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table} WHERE status = $1"))
+   let pending: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
+       "SELECT COUNT(*) FROM {table} WHERE status = $1"
+   )))
+   .bind("Pending")
+   .fetch_one(&pool)
+   .await?;
+   # Ok(())
+   # }
+   ```
+
+   This affects only SQL your application writes itself, for example reporting
+   queries against `hammerwork_jobs`. No Hammerwork API takes an SQL string.
+
+4. **MySQL without TLS: enable `mysql-rsa` if you need it.** sqlx 0.9 moved the RSA
+   password encryption used by MySQL's `caching_sha2_password` and `sha256_password`
+   authentication over unencrypted connections behind a new `mysql-rsa` feature.
+   Without it, connecting with `ssl-mode=disabled`, or to a server without TLS,
+   fails with "RSA auth backend disabled" whenever the server needs a full
+   authentication (for example the first login after a server restart). The
+   library does not enable `mysql-rsa`. If you connect without TLS, either enable
+   TLS or add the feature to your own sqlx dependency; Cargo's feature unification
+   turns it on for Hammerwork too:
+
+   ```toml
+   sqlx = { version = "0.9", features = ["runtime-tokio", "tls-rustls-ring-webpki", "mysql", "mysql-rsa"] }
+   ```
+
+   The `cargo-hammerwork` CLI and the `hammerwork-web` dashboard enable `mysql-rsa`,
+   so they connect as before.
+
+5. **MySQL connection collation.** sqlx 0.8 ran `SET NAMES utf8mb4 COLLATE
+   utf8mb4_unicode_ci` on every new MySQL connection. sqlx 0.9 runs `SET NAMES
+   utf8mb4` and lets the server pick its default collation (`utf8mb4_0900_ai_ci` on
+   MySQL 8.0). Hammerwork's queries compare columns with parameters, where the column
+   collation applies, so Hammerwork behaves the same. Queries of your own that
+   compare string literals or parameters with each other may sort or match
+   differently. To keep the old behaviour, set it on the connect options:
+   `MySqlConnectOptions::from_str(url)?.collation("utf8mb4_unicode_ci")`.
+
+For other sqlx 0.9 changes that may affect your own sqlx code (`RawSql`,
+`IntoArguments` and `Arguments` lost their lifetime parameter, the `Migrate` trait,
+`sqlx.toml`), see the
+[sqlx 0.9.0 changelog](https://github.com/launchbadge/sqlx/blob/main/CHANGELOG.md#090---2026-05-06).
+
+### What does not change
+
+- No database migration is needed. Hammerwork's migrations, and the way they are run
+  (one statement at a time, over the simple text protocol on MySQL), are the same.
+- Timestamps decode as before: PostgreSQL `TIMESTAMPTZ` and MySQL `TIMESTAMP(6)`
+  columns map to `chrono::DateTime<Utc>`, and sqlx still sets the MySQL session
+  `time_zone` to `+00:00`.
+- Hammerwork's own types and function signatures are unchanged by this upgrade;
+  only the sqlx version behind them moves.
 
 ## Feature flags
 
diff --git a/hammerwork-web/Cargo.toml b/hammerwork-web/Cargo.toml
index ede5247..24ebeeb 100644
--- a/hammerwork-web/Cargo.toml
+++ b/hammerwork-web/Cargo.toml
@@ -37,7 +37,7 @@ clap = { workspace = true, features = ["derive"] }
 toml = { workspace = true }
 
 # Database
-sqlx = { workspace = true, features = ["postgres", "mysql"] }
+sqlx = { workspace = true, features = ["postgres", "mysql", "mysql-rsa"] }
 
 # Utility
 anyhow = { workspace = true }
diff --git a/hammerwork-web/src/api/history.rs b/hammerwork-web/src/api/history.rs
index 2935dd7..cf5f4f6 100644
--- a/hammerwork-web/src/api/history.rs
+++ b/hammerwork-web/src/api/history.rs
@@ -742,7 +742,9 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
              FROM hammerwork_jobs WHERE status = 'Completed' \
              AND completed_at >= $1 AND completed_at < $2{queue_clause} GROUP BY 1"
         );
-        let mut q = sqlx::query(&completed_sql).bind(first).bind(upper);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(completed_sql))
+            .bind(first)
+            .bind(upper);
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -761,7 +763,9 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
              AND COALESCE(failed_at, timed_out_at) >= $1 AND COALESCE(failed_at, timed_out_at) < $2{queue_clause} \
              GROUP BY 1"
         );
-        let mut q = sqlx::query(&failed_sql).bind(first).bind(upper);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(failed_sql))
+            .bind(first)
+            .bind(upper);
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -812,7 +816,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
             "SELECT COUNT(*)::bigint FROM hammerwork_jobs {}",
             finished_filter(kind, queue.is_some(), older_than.is_some(), true)
         );
-        let mut q = sqlx::query_scalar::<_, i64>(&sql);
+        let mut q = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -832,7 +836,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
         else {
             return Ok(0);
         };
-        let mut q = sqlx::query_scalar::<_, i64>(&sql);
+        let mut q = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -852,7 +856,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
             "DELETE FROM hammerwork_jobs {}",
             finished_filter(kind, queue.is_some(), older_than.is_some(), true)
         );
-        let mut q = sqlx::query(&sql);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -870,9 +874,12 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
     ) -> Result<(Vec<ArchivedJob>, u64)> {
         let (conditions, binds) = archive_conditions(filter, true);
         let count_sql = format!("SELECT COUNT(*) FROM hammerwork_jobs_archive {conditions}");
-        let total = bind_archive_filter!(sqlx::query_scalar::<_, i64>(&count_sql), &binds)
-            .fetch_one(&self.pool)
-            .await?;
+        let total = bind_archive_filter!(
+            sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql)),
+            &binds
+        )
+        .fetch_one(&self.pool)
+        .await?;
         let (limit_ph, offset_ph) = (
             format!("${}", binds.len() + 1),
             format!("${}", binds.len() + 2),
@@ -881,7 +888,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
             "SELECT {ARCHIVED_JOB_COLUMNS} FROM hammerwork_jobs_archive {conditions} \
              ORDER BY archived_at DESC, id DESC LIMIT {limit_ph} OFFSET {offset_ph}"
         );
-        let rows = bind_archive_filter!(sqlx::query(&page_sql), &binds)
+        let rows = bind_archive_filter!(sqlx::query(sqlx::AssertSqlSafe(page_sql)), &binds)
             .bind(i64::from(limit))
             .bind(i64::from(offset))
             .fetch_all(&self.pool)
@@ -911,7 +918,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
             CHANGE_COLUMNS.join(", "),
             changed_since_filter(true)
         );
-        let rows = sqlx::query(&sql)
+        let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
             .bind(since)
             .bind(i64::from(limit))
             .fetch_all(&self.pool)
@@ -938,7 +945,7 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
         limit: u32,
     ) -> Result<Vec<hammerwork::Job>> {
         let sql = jobs_with_status_sql(queue_name.is_some(), true);
-        let mut query = sqlx::query(&sql).bind(status.to_string());
+        let mut query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(status.to_string());
         if let Some(name) = queue_name {
             query = query.bind(name.to_string());
         }
@@ -961,7 +968,10 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
              FROM hammerwork_jobs WHERE id = ANY($1)",
             CHANGE_COLUMNS.join(", ")
         );
-        let rows = sqlx::query(&sql).bind(&ids).fetch_all(&self.pool).await?;
+        let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
+            .bind(&ids)
+            .fetch_all(&self.pool)
+            .await?;
         Ok(pg_job_changes(&rows))
     }
 
@@ -984,7 +994,9 @@ impl JobHistory for JobQueue<sqlx::Postgres> {
             let statement = operation.postgres_statement(table);
             if !dry_run {
                 // Simple-query protocol: VACUUM may not run inside a transaction block.
-                sqlx::raw_sql(&statement).execute(&self.pool).await?;
+                sqlx::raw_sql(sqlx::AssertSqlSafe(statement.as_str()))
+                    .execute(&self.pool)
+                    .await?;
             }
             ran.push(MaintenanceStatement {
                 table,
@@ -1021,7 +1033,9 @@ impl JobHistory for JobQueue<sqlx::MySql> {
              FROM hammerwork_jobs WHERE status = 'Completed' \
              AND completed_at >= ? AND completed_at < ?{queue_clause} GROUP BY bucket"
         );
-        let mut q = sqlx::query(&completed_sql).bind(first).bind(upper);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(completed_sql))
+            .bind(first)
+            .bind(upper);
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -1042,7 +1056,9 @@ impl JobHistory for JobQueue<sqlx::MySql> {
              AND COALESCE(failed_at, timed_out_at) >= ? AND COALESCE(failed_at, timed_out_at) < ?{queue_clause} \
              GROUP BY bucket"
         );
-        let mut q = sqlx::query(&failed_sql).bind(first).bind(upper);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(failed_sql))
+            .bind(first)
+            .bind(upper);
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -1095,7 +1111,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             "SELECT CAST(COUNT(*) AS SIGNED) FROM hammerwork_jobs {}",
             finished_filter(kind, queue.is_some(), older_than.is_some(), false)
         );
-        let mut q = sqlx::query_scalar::<_, i64>(&sql);
+        let mut q = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -1115,7 +1131,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
         else {
             return Ok(0);
         };
-        let mut q = sqlx::query_scalar::<_, i64>(&sql);
+        let mut q = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -1135,7 +1151,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             "DELETE FROM hammerwork_jobs {}",
             finished_filter(kind, queue.is_some(), older_than.is_some(), false)
         );
-        let mut q = sqlx::query(&sql);
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
         if let Some(name) = queue {
             q = q.bind(name.to_string());
         }
@@ -1153,14 +1169,17 @@ impl JobHistory for JobQueue<sqlx::MySql> {
     ) -> Result<(Vec<ArchivedJob>, u64)> {
         let (conditions, binds) = archive_conditions(filter, false);
         let count_sql = format!("SELECT COUNT(*) FROM hammerwork_jobs_archive {conditions}");
-        let total = bind_archive_filter!(sqlx::query_scalar::<_, i64>(&count_sql), &binds)
-            .fetch_one(&self.pool)
-            .await?;
+        let total = bind_archive_filter!(
+            sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql)),
+            &binds
+        )
+        .fetch_one(&self.pool)
+        .await?;
         let page_sql = format!(
             "SELECT {ARCHIVED_JOB_COLUMNS} FROM hammerwork_jobs_archive {conditions} \
              ORDER BY archived_at DESC, id DESC LIMIT ? OFFSET ?"
         );
-        let rows = bind_archive_filter!(sqlx::query(&page_sql), &binds)
+        let rows = bind_archive_filter!(sqlx::query(sqlx::AssertSqlSafe(page_sql)), &binds)
             .bind(i64::from(limit))
             .bind(i64::from(offset))
             .fetch_all(&self.pool)
@@ -1188,7 +1207,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             mysql_latest_change(),
             changed_since_filter(false)
         );
-        let mut query = sqlx::query(&sql);
+        let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
         for _ in CHANGE_COLUMNS {
             query = query.bind(since);
         }
@@ -1215,7 +1234,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
         limit: u32,
     ) -> Result<Vec<hammerwork::Job>> {
         let sql = jobs_with_status_sql(queue_name.is_some(), false);
-        let mut query = sqlx::query(&sql).bind(status.to_string());
+        let mut query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(status.to_string());
         if let Some(name) = queue_name {
             query = query.bind(name.to_string());
         }
@@ -1242,7 +1261,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             mysql_latest_change(),
             vec!["?"; ids.len()].join(", ")
         );
-        let mut query = sqlx::query(&sql);
+        let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
         for id in &ids {
             query = query.bind(id);
         }
@@ -1263,7 +1282,7 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             "SELECT CAST(TABLE_NAME AS CHAR) FROM information_schema.TABLES \
              WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN ({placeholders})"
         );
-        let mut q = sqlx::query_scalar::<_, String>(&sql);
+        let mut q = sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(sql));
         for table in tables {
             q = q.bind(*table);
         }
@@ -1274,7 +1293,10 @@ impl JobHistory for JobQueue<sqlx::MySql> {
             let mut messages = Vec::new();
             if !dry_run {
                 // OPTIMIZE TABLE reports failures as result rows, not as SQL errors.
-                for row in sqlx::raw_sql(&statement).fetch_all(&self.pool).await? {
+                for row in sqlx::raw_sql(sqlx::AssertSqlSafe(statement.as_str()))
+                    .fetch_all(&self.pool)
+                    .await?
+                {
                     let kind = text_column(&row, "Msg_type");
                     let text = text_column(&row, "Msg_text");
                     if kind.eq_ignore_ascii_case("error") {
diff --git a/hammerwork-web/src/api/mod.rs b/hammerwork-web/src/api/mod.rs
index c0f383a..55082b9 100644
--- a/hammerwork-web/src/api/mod.rs
+++ b/hammerwork-web/src/api/mod.rs
@@ -794,7 +794,10 @@ mod undecodable_row_tests {
         let sql = |statement: String| {
             let pool = pool.clone();
             async move {
-                sqlx::query(&statement).execute(&pool).await.unwrap();
+                sqlx::query(sqlx::AssertSqlSafe(statement))
+                    .execute(&pool)
+                    .await
+                    .unwrap();
             }
         };
         listings_skip_undecodable_rows(queue, sql, false).await;
@@ -809,7 +812,10 @@ mod undecodable_row_tests {
         let sql = |statement: String| {
             let pool = pool.clone();
             async move {
-                sqlx::query(&statement).execute(&pool).await.unwrap();
+                sqlx::query(sqlx::AssertSqlSafe(statement))
+                    .execute(&pool)
+                    .await
+                    .unwrap();
             }
         };
         listings_skip_undecodable_rows(queue, sql, true).await;
diff --git a/src/encryption/key_manager.rs b/src/encryption/key_manager.rs
index c0282c9..163df9d 100644
--- a/src/encryption/key_manager.rs
+++ b/src/encryption/key_manager.rs
@@ -3994,7 +3994,7 @@ mod tests {
             .await
             .expect("failed to run PostgreSQL migrations");
         for table in ["hammerwork_key_audit_log", "hammerwork_encryption_keys"] {
-            sqlx::query(&format!("DELETE FROM {table}"))
+            sqlx::query(sqlx::AssertSqlSafe(format!("DELETE FROM {table}")))
                 .execute(&pool)
                 .await
                 .expect("failed to clear key tables");
@@ -4019,7 +4019,7 @@ mod tests {
             .await
             .expect("failed to run MySQL migrations");
         for table in ["hammerwork_key_audit_log", "hammerwork_encryption_keys"] {
-            sqlx::query(&format!("DELETE FROM {table}"))
+            sqlx::query(sqlx::AssertSqlSafe(format!("DELETE FROM {table}")))
                 .execute(&pool)
                 .await
                 .expect("failed to clear key tables");
@@ -4681,7 +4681,7 @@ mod tests {
     async fn persistence_roundtrip<DB: KeyManagerBackend>(pool: Pool<DB>)
     where
         for<'c> &'c mut DB::Connection: sqlx::Executor<'c, Database = DB>,
-        for<'q> <DB as Database>::Arguments<'q>: sqlx::IntoArguments<'q, DB>,
+        <DB as Database>::Arguments: sqlx::IntoArguments<DB>,
         for<'r> Vec<u8>: sqlx::Decode<'r, DB> + sqlx::Type<DB>,
         for<'r> String: sqlx::Decode<'r, DB> + sqlx::Type<DB>,
         for<'n> &'n str: sqlx::ColumnIndex<DB::Row>,
diff --git a/src/lib.rs b/src/lib.rs
index 33be1a9..2de8082 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -460,4 +460,7 @@ mod doc_examples {
 
     #[doc = include_str!("../docs/encryption.md")]
     struct EncryptionDoctests;
+
+    #[doc = include_str!("../docs/upgrading-to-3.0.md")]
+    struct UpgradingTo30Doctests;
 }
diff --git a/src/migrations/mysql.rs b/src/migrations/mysql.rs
index 63adbea..4646082 100644
--- a/src/migrations/mysql.rs
+++ b/src/migrations/mysql.rs
@@ -4,7 +4,7 @@ use super::split::{Dialect, split_statements};
 use super::{Migration, MigrationRecord, MigrationRunner};
 use crate::Result;
 use chrono::Utc;
-use sqlx::{Executor, MySql, MySqlConnection, MySqlPool, Row, pool::PoolConnection};
+use sqlx::{AssertSqlSafe, Executor, MySql, MySqlConnection, MySqlPool, Row, pool::PoolConnection};
 use tokio::sync::{Mutex, MutexGuard};
 use tracing::{debug, info};
 
@@ -130,7 +130,7 @@ impl MigrationRunner<sqlx::MySql> for MySqlMigrationRunner {
 
             // Use the text protocol: dynamic DDL (PREPARE/EXECUTE/DEALLOCATE) is not
             // supported through MySQL's prepared statement protocol.
-            tx.execute(full_statement.as_str()).await?;
+            tx.execute(AssertSqlSafe(full_statement)).await?;
         }
 
         tx.commit().await?;
diff --git a/src/migrations/postgres.rs b/src/migrations/postgres.rs
index 3823469..8e41a3a 100644
--- a/src/migrations/postgres.rs
+++ b/src/migrations/postgres.rs
@@ -4,7 +4,7 @@ use super::split::{Dialect, split_statements};
 use super::{Migration, MigrationRecord, MigrationRunner};
 use crate::Result;
 use chrono::Utc;
-use sqlx::{PgConnection, PgPool, Postgres, Row, pool::PoolConnection};
+use sqlx::{AssertSqlSafe, PgConnection, PgPool, Postgres, Row, pool::PoolConnection};
 use tokio::sync::{Mutex, MutexGuard};
 use tracing::{debug, error, info};
 
@@ -124,7 +124,10 @@ impl MigrationRunner<sqlx::Postgres> for PostgresMigrationRunner {
                 full_statement
             );
 
-            if let Err(e) = sqlx::query(&full_statement).execute(&mut *tx).await {
+            if let Err(e) = sqlx::query(AssertSqlSafe(full_statement.as_str()))
+                .execute(&mut *tx)
+                .await
+            {
                 error!(
                     "Failed to execute statement {}: {} - Error: {}",
                     i + 1,
diff --git a/src/queue/mysql.rs b/src/queue/mysql.rs
index aed4556..b0afad0 100644
--- a/src/queue/mysql.rs
+++ b/src/queue/mysql.rs
@@ -23,7 +23,7 @@ use crate::{
 };
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use sqlx::{FromRow, MySql, Row};
+use sqlx::{AssertSqlSafe, FromRow, MySql, Row};
 use std::{collections::HashMap, time::Duration};
 
 #[derive(FromRow, Clone)]
@@ -103,9 +103,9 @@ async fn copy_encryption_columns(
         .map(|column| format!("dst.{column} = src.{column}"))
         .collect::<Vec<_>>()
         .join(", ");
-    sqlx::query(&format!(
+    sqlx::query(AssertSqlSafe(format!(
         "UPDATE {to} dst JOIN {from} src ON dst.id = src.id SET {assignments} WHERE src.id = ?"
-    ))
+    )))
     .bind(id.to_string())
     .execute(&mut *conn)
     .await?;
@@ -344,7 +344,7 @@ impl crate::queue::JobQueue<MySql> {
                     ""
                 }
             );
-            let mut select = sqlx::query(&query).bind(queue_name).bind(queue_name);
+            let mut select = sqlx::query(AssertSqlSafe(query)).bind(queue_name).bind(queue_name);
             if let Some(priority) = priority {
                 select = select.bind(priority.as_i32());
             }
@@ -361,11 +361,11 @@ impl crate::queue::JobQueue<MySql> {
                 .map_err(crate::HammerworkError::from)
                 .and_then(JobRow::into_job);
 
-            sqlx::query(&format!(
+            sqlx::query(AssertSqlSafe(format!(
                 "UPDATE hammerwork_jobs SET status = ?, started_at = ?, attempts = attempts + 1, \
                  last_heartbeat_at = ?, lease_expires_at = LEAST(DATE_ADD(?, INTERVAL ? MICROSECOND), \
                  CAST('{TIMESTAMP_MAX_SQL}' AS DATETIME(6))) WHERE id = ?"
-            ))
+            )))
             .bind(JobStatus::Running)
             .bind(started_at)
             .bind(started_at)
@@ -412,7 +412,7 @@ impl crate::queue::JobQueue<MySql> {
                     ""
                 }
             );
-            let mut select = sqlx::query(&query).bind(queue_name).bind(queue_name);
+            let mut select = sqlx::query(AssertSqlSafe(query)).bind(queue_name).bind(queue_name);
             if let Some(priority) = priority {
                 select = select.bind(priority.as_i32());
             }
@@ -447,7 +447,7 @@ impl crate::queue::JobQueue<MySql> {
                  last_heartbeat_at = ?, lease_expires_at = LEAST(DATE_ADD(?, INTERVAL ? MICROSECOND), \
                  CAST('{TIMESTAMP_MAX_SQL}' AS DATETIME(6))) WHERE id IN ({placeholders})"
             );
-            let mut update = sqlx::query(&sql)
+            let mut update = sqlx::query(AssertSqlSafe(sql))
                 .bind(JobStatus::Running)
                 .bind(started_at)
                 .bind(started_at)
@@ -477,12 +477,13 @@ impl crate::queue::JobQueue<MySql> {
     ) -> Result<Option<Job>> {
         // Which priority levels have a runnable job: one index probe per level, so
         // every level is considered however many jobs of other levels are queued.
-        let levels: Vec<i64> =
-            sqlx::query_scalar(&super::runnable_priorities_sql(RUNNABLE_IN_QUEUE_SQL))
-                .bind(queue_name)
-                .bind(queue_name)
-                .fetch_all(&self.pool)
-                .await?;
+        let levels: Vec<i64> = sqlx::query_scalar(AssertSqlSafe(super::runnable_priorities_sql(
+            RUNNABLE_IN_QUEUE_SQL,
+        )))
+        .bind(queue_name)
+        .bind(queue_name)
+        .fetch_all(&self.pool)
+        .await?;
         let mut candidates: Vec<JobPriority> = levels
             .into_iter()
             .filter_map(|level| i32::try_from(level).ok())
@@ -613,12 +614,13 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         };
 
         // Which priority levels have a runnable job (see `dequeue_leased`).
-        let levels: Vec<i64> =
-            sqlx::query_scalar(&super::runnable_priorities_sql(RUNNABLE_IN_QUEUE_SQL))
-                .bind(queue_name)
-                .bind(queue_name)
-                .fetch_all(&self.pool)
-                .await?;
+        let levels: Vec<i64> = sqlx::query_scalar(AssertSqlSafe(super::runnable_priorities_sql(
+            RUNNABLE_IN_QUEUE_SQL,
+        )))
+        .bind(queue_name)
+        .bind(queue_name)
+        .fetch_all(&self.pool)
+        .await?;
         let candidates: Vec<JobPriority> = levels
             .into_iter()
             .filter_map(|level| i32::try_from(level).ok())
@@ -738,10 +740,10 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
     }
 
     async fn get_job(&self, job_id: JobId) -> Result<Option<Job>> {
-        let row = sqlx::query_as::<_, JobRow>(&format!(
+        let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {} FROM hammerwork_jobs WHERE id = ?",
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(job_id.to_string())
         .fetch_optional(&self.pool)
         .await?;
@@ -751,9 +753,9 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         }
 
         // Archiving moves a job out of hammerwork_jobs; report it with status Archived.
-        let archived = sqlx::query(&format!(
+        let archived = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive WHERE id = ?"
-        ))
+        )))
         .bind(job_id.to_string())
         .fetch_optional(&self.pool)
         .await?;
@@ -893,10 +895,10 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
     }
 
     async fn get_batch_jobs(&self, batch_id: crate::batch::BatchId) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {} FROM hammerwork_jobs WHERE batch_id = ? ORDER BY created_at ASC",
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(batch_id.to_string())
         .fetch_all(&self.pool)
         .await?;
@@ -949,9 +951,9 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         let limit = limit.unwrap_or(100) as i64;
         let offset = offset.unwrap_or(0) as i64;
 
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE status = ? ORDER BY failed_at DESC LIMIT ? OFFSET ?"
-        ))
+        )))
         .bind(JobStatus::Dead)
         .bind(limit)
         .bind(offset)
@@ -976,9 +978,9 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         let limit = limit.unwrap_or(100) as i64;
         let offset = offset.unwrap_or(0) as i64;
 
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE status = ? AND queue_name = ? ORDER BY failed_at DESC LIMIT ? OFFSET ?"
-        ))
+        )))
         .bind(JobStatus::Dead)
         .bind(queue_name)
         .bind(limit)
@@ -1298,14 +1300,14 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         };
 
         let rows: Vec<JobRow> = if let Some(queue) = queue_name {
-            sqlx::query_as(&query)
+            sqlx::query_as(AssertSqlSafe(query))
                 .bind(queue)
                 .bind(start_time)
                 .bind(end_time)
                 .fetch_all(&self.pool)
                 .await?
         } else {
-            sqlx::query_as(&query)
+            sqlx::query_as(AssertSqlSafe(query))
                 .bind(start_time)
                 .bind(end_time)
                 .fetch_all(&self.pool)
@@ -1350,13 +1352,13 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         };
 
         let rows = if let Some(queue) = queue_name {
-            sqlx::query_as::<_, JobRow>(&query)
+            sqlx::query_as::<_, JobRow>(AssertSqlSafe(query))
                 .bind(queue)
                 .bind(JobStatus::Pending)
                 .fetch_all(&self.pool)
                 .await?
         } else {
-            sqlx::query_as::<_, JobRow>(&query)
+            sqlx::query_as::<_, JobRow>(AssertSqlSafe(query))
                 .bind(JobStatus::Pending)
                 .fetch_all(&self.pool)
                 .await?
@@ -1394,7 +1396,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
 
     async fn get_recurring_jobs(&self, queue_name: &str) -> Result<Vec<Job>> {
         let rows = sqlx::query_as::<_, JobRow>(
-            &format!("SELECT {} FROM hammerwork_jobs WHERE queue_name = ? AND recurring = TRUE ORDER BY next_run_at ASC", JOB_SELECT_FIELDS)
+            AssertSqlSafe(format!("SELECT {} FROM hammerwork_jobs WHERE queue_name = ? AND recurring = TRUE ORDER BY next_run_at ASC", JOB_SELECT_FIELDS))
         )
         .bind(queue_name)
         .fetch_all(&self.pool)
@@ -1428,9 +1430,9 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
             let mut conn = self.pool.acquire().await?;
             let mut tx = Self::begin_claim_transaction(&mut conn).await?;
             let result = async {
-                let row = sqlx::query_as::<_, JobRow>(&format!(
+                let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
                     "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE id = ? FOR UPDATE"
-                ))
+                )))
                 .bind(job_id.to_string())
                 .fetch_optional(&mut *tx)
                 .await?;
@@ -1686,7 +1688,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
     }
 
     async fn get_ready_jobs(&self, queue_name: &str, limit: u32) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             r#"
                 SELECT {}
                 FROM hammerwork_jobs
@@ -1698,7 +1700,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
                 LIMIT ?
                 "#,
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(queue_name)
         .bind(limit)
         .fetch_all(&self.pool)
@@ -1725,7 +1727,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         &self,
         workflow_id: crate::workflow::WorkflowId,
     ) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             r#"
                 SELECT {}
                 FROM hammerwork_jobs
@@ -1733,7 +1735,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
                 ORDER BY created_at ASC
                 "#,
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(workflow_id.to_string())
         .fetch_all(&self.pool)
         .await?;
@@ -1743,11 +1745,11 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
             .collect::<Result<Vec<_>>>()?;
 
         // Archived members, with status `Archived` (as `get_job` reports them).
-        let archived = sqlx::query(&format!(
+        let archived = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive a \
              WHERE a.workflow_id = ? \
              AND NOT EXISTS (SELECT 1 FROM hammerwork_jobs j WHERE j.id = a.id)"
-        ))
+        )))
         .bind(workflow_id.to_string())
         .fetch_all(&self.pool)
         .await?;
@@ -1828,7 +1830,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
             query.push_str(&format!(" OFFSET {}", offset));
         }
 
-        let mut sql_query = sqlx::query(&query);
+        let mut sql_query = sqlx::query(AssertSqlSafe(query));
 
         if let Some(queue) = queue_name {
             sql_query = sql_query.bind(queue);
@@ -1916,7 +1918,7 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
             base_query.push_str(" WHERE queue_name = ?");
         }
 
-        let mut query = sqlx::query(&base_query);
+        let mut query = sqlx::query(AssertSqlSafe(base_query));
         if let Some(queue) = queue_name {
             query = query.bind(queue);
         }
@@ -2013,12 +2015,12 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
         // dequeued is extended: a stale worker whose job was reclaimed and claimed again
         // must not keep the new run's lease alive.
         let started = super::run_started_range(run);
-        let result = sqlx::query(&format!(
+        let result = sqlx::query(AssertSqlSafe(format!(
             "UPDATE hammerwork_jobs SET last_heartbeat_at = UTC_TIMESTAMP(6), \
              lease_expires_at = LEAST(DATE_ADD(UTC_TIMESTAMP(6), INTERVAL ? MICROSECOND), \
              CAST('{TIMESTAMP_MAX_SQL}' AS DATETIME(6))) WHERE id = ? AND status = ? \
              AND attempts = ? AND (? IS NULL OR (started_at > ? AND started_at < ?))"
-        ))
+        )))
         .bind(lease_micros(lease))
         .bind(run.id.to_string())
         .bind(JobStatus::Running)
@@ -2041,12 +2043,12 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
             .fetch_one(&self.pool)
             .await?;
         let started = super::run_started_range(run);
-        let result = sqlx::query(&format!(
+        let result = sqlx::query(AssertSqlSafe(format!(
             "UPDATE hammerwork_jobs SET started_at = ?, last_heartbeat_at = ?, \
              lease_expires_at = LEAST(DATE_ADD(?, INTERVAL ? MICROSECOND), \
              CAST('{TIMESTAMP_MAX_SQL}' AS DATETIME(6))) WHERE id = ? AND status = ? \
              AND attempts = ? AND (? IS NULL OR (started_at > ? AND started_at < ?))"
-        ))
+        )))
         .bind(now)
         .bind(now)
         .bind(now)
@@ -2088,12 +2090,12 @@ impl DatabaseQueue for crate::queue::JobQueue<MySql> {
     async fn purge_expired_encrypted_jobs(&self) -> Result<super::EncryptedJobPurge> {
         let mut tx = self.pool.begin().await?;
         let now = db_now(&mut tx).await?;
-        let jobs = sqlx::query(&format!(
+        let jobs = sqlx::query(AssertSqlSafe(format!(
             "DELETE FROM hammerwork_jobs WHERE is_encrypted = true \
              AND retention_delete_at IS NOT NULL AND retention_delete_at <= ? \
              AND status IN ({})",
             super::FINISHED_STATUS_SQL
-        ))
+        )))
         .bind(now)
         .execute(&mut *tx)
         .await?
@@ -2148,7 +2150,7 @@ impl crate::queue::JobQueue<MySql> {
         // A run's lease is written when it is claimed (`last_heartbeat_at = started_at`),
         // so only its expiry counts. The `started_at` cutoff applies only to runs without
         // a lease of their own: claimed by an older version that wrote none.
-        let candidates = sqlx::query_as::<_, JobRow>(&format!(
+        let candidates = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             r#"
             SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs
             WHERE status = ?
@@ -2162,7 +2164,7 @@ impl crate::queue::JobQueue<MySql> {
               )
             FOR UPDATE SKIP LOCKED
             "#
-        ))
+        )))
         .bind(JobStatus::Running)
         .bind(now)
         .bind(cutoff)
@@ -2518,7 +2520,8 @@ async fn insert_dependent_jobs(conn: &mut sqlx::MySqlConnection, jobs: &mut [Job
              LEFT JOIN hammerwork_workflows w ON w.id = j.workflow_id \
              WHERE j.id IN ({placeholders}) ORDER BY j.id FOR SHARE OF j"
         );
-        let mut select = sqlx::query_as::<_, (String, String, Option<String>)>(&query);
+        let mut select =
+            sqlx::query_as::<_, (String, String, Option<String>)>(AssertSqlSafe(query));
         for id in chunk {
             select = select.bind(id.to_string());
         }
@@ -2532,7 +2535,7 @@ async fn insert_dependent_jobs(conn: &mut sqlx::MySqlConnection, jobs: &mut [Job
         // Finished dependencies may already have been archived.
         let query =
             format!("SELECT id, status FROM hammerwork_jobs_archive WHERE id IN ({placeholders})");
-        let mut select = sqlx::query_as::<_, (String, String)>(&query);
+        let mut select = sqlx::query_as::<_, (String, String)>(AssertSqlSafe(query));
         for id in chunk {
             select = select.bind(id.to_string());
         }
@@ -2580,9 +2583,9 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
     type Conn = sqlx::MySqlConnection;
 
     async fn lock_job(conn: &mut sqlx::MySqlConnection, job_id: JobId) -> Result<Option<Job>> {
-        let row = sqlx::query_as::<_, JobRow>(&format!(
+        let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE id = ? FOR UPDATE"
-        ))
+        )))
         .bind(job_id.to_string())
         .fetch_optional(&mut *conn)
         .await?;
@@ -2715,7 +2718,7 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
                 "SELECT id, status FROM hammerwork_jobs WHERE id IN ({placeholders}) \
                  ORDER BY id FOR SHARE"
             );
-            let mut select = sqlx::query_as::<_, (String, String)>(&query);
+            let mut select = sqlx::query_as::<_, (String, String)>(AssertSqlSafe(query));
             for id in chunk {
                 select = select.bind(id.to_string());
             }
@@ -2728,7 +2731,7 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
             let query = format!(
                 "SELECT id, status FROM hammerwork_jobs_archive WHERE id IN ({placeholders})"
             );
-            let mut select = sqlx::query_as::<_, (String, String)>(&query);
+            let mut select = sqlx::query_as::<_, (String, String)>(AssertSqlSafe(query));
             for id in chunk {
                 select = select.bind(id.to_string());
             }
@@ -2830,7 +2833,7 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
                  UNION SELECT id FROM hammerwork_jobs_archive WHERE id IN ({placeholders}) \
                  AND status = 'Completed'"
             );
-            let mut select = sqlx::query_scalar::<_, String>(&query);
+            let mut select = sqlx::query_scalar::<_, String>(AssertSqlSafe(query));
             for _ in 0..2 {
                 for id in chunk {
                     select = select.bind(*id);
@@ -2850,7 +2853,7 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
                 "UPDATE hammerwork_jobs SET dependency_status = 'satisfied' \
                  WHERE id IN ({placeholders})"
             );
-            let mut update = sqlx::query(&query);
+            let mut update = sqlx::query(AssertSqlSafe(query));
             for id in chunk {
                 update = update.bind(id);
             }
@@ -2870,9 +2873,9 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
         let mut failed = Vec::new();
         let mut to_check = vec![failed_job_id];
         while let Some(current) = to_check.pop() {
-            let candidates: Vec<String> = sqlx::query_scalar(&format!(
+            let candidates: Vec<String> = sqlx::query_scalar(AssertSqlSafe(format!(
                 "SELECT id FROM hammerwork_jobs WHERE JSON_CONTAINS(depends_on, ?) AND {CONDITION}"
-            ))
+            )))
             .bind(serde_json::json!([current.to_string()]))
             .fetch_all(&mut *conn)
             .await?;
@@ -3019,11 +3022,11 @@ impl TransitionStore for crate::queue::JobQueue<MySql> {
             .await?;
         }
 
-        let unfinished: i64 = sqlx::query_scalar(&format!(
+        let unfinished: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
             "SELECT EXISTS (SELECT 1 FROM hammerwork_jobs WHERE batch_id = ? \
              AND status IN ({}))",
             lifecycle::UNFINISHED_STATUS_SQL
-        ))
+        )))
         .bind(&batch)
         .fetch_one(&mut *conn)
         .await?;
@@ -3143,7 +3146,7 @@ impl crate::queue::JobQueue<MySql> {
                 "SELECT {columns} FROM hammerwork_jobs WHERE id IN ({placeholders}) \
                  AND {condition} ORDER BY id FOR UPDATE"
             );
-            let mut select = sqlx::query(&query);
+            let mut select = sqlx::query(AssertSqlSafe(query));
             for id in chunk {
                 select = select.bind(id);
             }
@@ -3173,7 +3176,7 @@ impl crate::queue::JobQueue<MySql> {
                  dependency_status = CASE WHEN dependency_status IN {failed_dependencies} \
                  THEN 'failed' ELSE dependency_status END WHERE id IN ({placeholders})"
             );
-            let mut update = sqlx::query(&query).bind(now).bind(error);
+            let mut update = sqlx::query(AssertSqlSafe(query)).bind(now).bind(error);
             for id in chunk {
                 update = update.bind(id);
             }
@@ -3196,10 +3199,10 @@ impl crate::queue::JobQueue<MySql> {
         Self::lock_workflow(conn, workflow_id).await?;
         let now = db_now(conn).await?;
         let id = workflow_id.to_string();
-        let candidates: Vec<String> = sqlx::query_scalar(&format!(
+        let candidates: Vec<String> = sqlx::query_scalar(AssertSqlSafe(format!(
             "SELECT id FROM hammerwork_jobs WHERE workflow_id = ? AND status IN ({})",
             lifecycle::UNFINISHED_STATUS_SQL
-        ))
+        )))
         .bind(&id)
         .fetch_all(&mut *conn)
         .await?;
@@ -3221,7 +3224,7 @@ impl crate::queue::JobQueue<MySql> {
                  THEN 'failed' ELSE dependency_status END, \
                  last_heartbeat_at = NULL, lease_expires_at = NULL WHERE id IN ({placeholders})"
             );
-            let mut update = sqlx::query(&query)
+            let mut update = sqlx::query(AssertSqlSafe(query))
                 .bind(now)
                 .bind(lifecycle::WORKFLOW_CANCELLED_MESSAGE);
             for id in chunk {
@@ -3380,7 +3383,7 @@ impl crate::queue::JobQueue<MySql> {
         while jobs_to_archive.len() < policy.batch_size {
             let wanted = policy.batch_size - jobs_to_archive.len();
             let select = select(wanted, examined.len());
-            let mut query = sqlx::query_as::<_, JobRow>(&select);
+            let mut query = sqlx::query_as::<_, JobRow>(AssertSqlSafe(select));
             if let Some(queue) = queue_name {
                 query = query.bind(queue);
             }
@@ -3538,9 +3541,9 @@ impl crate::queue::JobQueue<MySql> {
         let mut tx = self.pool.begin().await?;
 
         // Lock the archived row so two concurrent restores cannot both re-insert it.
-        let archived_row = sqlx::query(&format!(
+        let archived_row = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive WHERE id = ? FOR UPDATE"
-        ))
+        )))
         .bind(job_id.to_string())
         .fetch_optional(&mut *tx)
         .await?
diff --git a/src/queue/postgres.rs b/src/queue/postgres.rs
index cf2b9cd..387955f 100644
--- a/src/queue/postgres.rs
+++ b/src/queue/postgres.rs
@@ -23,7 +23,7 @@ use crate::{
 };
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use sqlx::{FromRow, Postgres, Row};
+use sqlx::{AssertSqlSafe, FromRow, Postgres, Row};
 use std::collections::HashMap;
 
 #[derive(FromRow, Clone)]
@@ -106,10 +106,10 @@ async fn copy_encryption_columns(
         .map(|column| format!("{column} = src.{column}"))
         .collect::<Vec<_>>()
         .join(", ");
-    sqlx::query(&format!(
+    sqlx::query(AssertSqlSafe(format!(
         "UPDATE {to} dst SET {assignments} FROM {from} src \
          WHERE dst.id = src.id AND src.id = ANY($1)"
-    ))
+    )))
     .bind(ids)
     .execute(&mut *conn)
     .await?;
@@ -352,7 +352,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             Some(weights) if !weights.is_strict() => weights,
             // Strict priority: the highest priority, then the oldest job.
             _ => {
-                let row = sqlx::query_as::<_, JobRow>(&claim_sql(false))
+                let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(claim_sql(false)))
                     .bind(queue_name)
                     .bind(lease)
                     .fetch_optional(&self.pool)
@@ -363,11 +363,12 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
 
         // Which priority levels have a runnable job: one index probe per level, so
         // every level is considered however many jobs of other levels are queued.
-        let levels: Vec<i32> =
-            sqlx::query_scalar(&super::runnable_priorities_sql(RUNNABLE_IN_QUEUE_SQL))
-                .bind(queue_name)
-                .fetch_all(&self.pool)
-                .await?;
+        let levels: Vec<i32> = sqlx::query_scalar(AssertSqlSafe(super::runnable_priorities_sql(
+            RUNNABLE_IN_QUEUE_SQL,
+        )))
+        .bind(queue_name)
+        .fetch_all(&self.pool)
+        .await?;
         let mut candidates: Vec<JobPriority> = levels
             .into_iter()
             .filter_map(|level| JobPriority::from_i32(level).ok())
@@ -381,7 +382,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             weights,
             super::weighted_seed(queue_name, attempt),
         ) {
-            let claimed = sqlx::query_as::<_, JobRow>(&claim_sql(true))
+            let claimed = sqlx::query_as::<_, JobRow>(AssertSqlSafe(claim_sql(true)))
                 .bind(queue_name)
                 .bind(lease)
                 .bind(priority.as_i32())
@@ -412,7 +413,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             Some(weights) if !weights.is_strict() => weights,
             // Strict priority: highest priority first, then the oldest jobs.
             _ => {
-                let rows = sqlx::query_as::<_, JobRow>(&claim_batch_sql(false))
+                let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(claim_batch_sql(false)))
                     .bind(queue_name)
                     .bind(lease)
                     .bind(max_jobs as i64)
@@ -423,11 +424,12 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         };
 
         // Which priority levels have a runnable job (see `dequeue_leased`).
-        let levels: Vec<i32> =
-            sqlx::query_scalar(&super::runnable_priorities_sql(RUNNABLE_IN_QUEUE_SQL))
-                .bind(queue_name)
-                .fetch_all(&self.pool)
-                .await?;
+        let levels: Vec<i32> = sqlx::query_scalar(AssertSqlSafe(super::runnable_priorities_sql(
+            RUNNABLE_IN_QUEUE_SQL,
+        )))
+        .bind(queue_name)
+        .fetch_all(&self.pool)
+        .await?;
         let candidates: Vec<JobPriority> = levels
             .into_iter()
             .filter_map(|level| JobPriority::from_i32(level).ok())
@@ -441,7 +443,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             weights,
             max_jobs,
             |priority, room| async move {
-                let rows = sqlx::query_as::<_, JobRow>(&claim_batch_sql(true))
+                let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(claim_batch_sql(true)))
                     .bind(queue_name)
                     .bind(lease)
                     .bind(room as i64)
@@ -546,10 +548,10 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         .await
     }
     async fn get_job(&self, job_id: JobId) -> Result<Option<Job>> {
-        let row = sqlx::query_as::<_, JobRow>(&format!(
+        let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {} FROM hammerwork_jobs WHERE id = $1",
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(job_id)
         .fetch_optional(&self.pool)
         .await?;
@@ -559,9 +561,9 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         }
 
         // Archiving moves a job out of hammerwork_jobs; report it with status Archived.
-        let archived = sqlx::query(&format!(
+        let archived = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive WHERE id = $1"
-        ))
+        )))
         .bind(job_id)
         .fetch_optional(&self.pool)
         .await?;
@@ -697,10 +699,10 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
     }
 
     async fn get_batch_jobs(&self, batch_id: crate::batch::BatchId) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {} FROM hammerwork_jobs WHERE batch_id = $1 ORDER BY created_at ASC",
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(batch_id)
         .fetch_all(&self.pool)
         .await?;
@@ -751,9 +753,9 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         let limit = limit.unwrap_or(100) as i64;
         let offset = offset.unwrap_or(0) as i64;
 
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE status = $1 ORDER BY failed_at DESC LIMIT $2 OFFSET $3"
-        ))
+        )))
         .bind(JobStatus::Dead)
         .bind(limit)
         .bind(offset)
@@ -776,9 +778,9 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         let limit = limit.unwrap_or(100) as i64;
         let offset = offset.unwrap_or(0) as i64;
 
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE status = $1 AND queue_name = $2 ORDER BY failed_at DESC LIMIT $3 OFFSET $4"
-        ))
+        )))
         .bind(JobStatus::Dead)
         .bind(queue_name)
         .bind(limit)
@@ -795,13 +797,13 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
 
     async fn retry_dead_job(&self, job_id: JobId) -> Result<()> {
         let transition = JobTransition::RetryDead;
-        let result = sqlx::query(&format!(
+        let result = sqlx::query(AssertSqlSafe(format!(
             "UPDATE hammerwork_jobs SET status = $1, attempts = 0, scheduled_at = now(), \
              started_at = NULL, failed_at = NULL, timed_out_at = NULL, \
              last_heartbeat_at = NULL, lease_expires_at = NULL \
              WHERE id = $2 AND status IN ({})",
             transition.sql_status_list()
-        ))
+        )))
         .bind(JobStatus::Pending)
         .bind(job_id)
         .execute(&self.pool)
@@ -1104,14 +1106,14 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         };
 
         let rows: Vec<JobRow> = if let Some(queue) = queue_name {
-            sqlx::query_as(&query)
+            sqlx::query_as(AssertSqlSafe(query))
                 .bind(queue)
                 .bind(start_time)
                 .bind(end_time)
                 .fetch_all(&self.pool)
                 .await?
         } else {
-            sqlx::query_as(&query)
+            sqlx::query_as(AssertSqlSafe(query))
                 .bind(start_time)
                 .bind(end_time)
                 .fetch_all(&self.pool)
@@ -1154,13 +1156,13 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         };
 
         let rows = if let Some(queue) = queue_name {
-            sqlx::query_as::<_, JobRow>(&query)
+            sqlx::query_as::<_, JobRow>(AssertSqlSafe(query))
                 .bind(queue)
                 .bind(JobStatus::Pending)
                 .fetch_all(&self.pool)
                 .await?
         } else {
-            sqlx::query_as::<_, JobRow>(&query)
+            sqlx::query_as::<_, JobRow>(AssertSqlSafe(query))
                 .bind(JobStatus::Pending)
                 .fetch_all(&self.pool)
                 .await?
@@ -1171,7 +1173,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
 
     async fn reschedule_cron_job(&self, job_id: JobId, next_run_at: DateTime<Utc>) -> Result<()> {
         let transition = JobTransition::RescheduleCron;
-        let result = sqlx::query(&format!(
+        let result = sqlx::query(AssertSqlSafe(format!(
             r#"
             UPDATE hammerwork_jobs
             SET status = $1,
@@ -1188,7 +1190,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             WHERE id = $3 AND recurring = TRUE AND status IN ({})
             "#,
             transition.sql_status_list()
-        ))
+        )))
         .bind(JobStatus::Pending)
         .bind(next_run_at)
         .bind(job_id)
@@ -1203,7 +1205,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
 
     async fn get_recurring_jobs(&self, queue_name: &str) -> Result<Vec<Job>> {
         let rows = sqlx::query_as::<_, JobRow>(
-            &format!("SELECT {} FROM hammerwork_jobs WHERE queue_name = $1 AND recurring = TRUE ORDER BY next_run_at ASC", JOB_SELECT_FIELDS)
+            AssertSqlSafe(format!("SELECT {} FROM hammerwork_jobs WHERE queue_name = $1 AND recurring = TRUE ORDER BY next_run_at ASC", JOB_SELECT_FIELDS))
         )
         .bind(queue_name)
         .fetch_all(&self.pool)
@@ -1235,9 +1237,9 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         super::retry_on_conflict(|| async {
             let mut tx = self.pool.begin().await?;
             let result = async {
-                let row = sqlx::query_as::<_, JobRow>(&format!(
+                let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
                     "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE id = $1 FOR UPDATE"
-                ))
+                )))
                 .bind(job_id)
                 .fetch_optional(&mut *tx)
                 .await?;
@@ -1489,7 +1491,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
     }
 
     async fn get_ready_jobs(&self, queue_name: &str, limit: u32) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             r#"
                 SELECT {}
                 FROM hammerwork_jobs
@@ -1501,7 +1503,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
                 LIMIT $2
                 "#,
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(queue_name)
         .bind(i64::from(limit))
         .fetch_all(&self.pool)
@@ -1527,7 +1529,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
         &self,
         workflow_id: crate::workflow::WorkflowId,
     ) -> Result<Vec<Job>> {
-        let rows = sqlx::query_as::<_, JobRow>(&format!(
+        let rows = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             r#"
                 SELECT {}
                 FROM hammerwork_jobs
@@ -1535,7 +1537,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
                 ORDER BY created_at ASC
                 "#,
             JOB_SELECT_FIELDS
-        ))
+        )))
         .bind(workflow_id)
         .fetch_all(&self.pool)
         .await?;
@@ -1545,11 +1547,11 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             .collect::<Result<Vec<_>>>()?;
 
         // Archived members, with status `Archived` (as `get_job` reports them).
-        let archived = sqlx::query(&format!(
+        let archived = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive a \
              WHERE a.workflow_id = $1 \
              AND NOT EXISTS (SELECT 1 FROM hammerwork_jobs j WHERE j.id = a.id)"
-        ))
+        )))
         .bind(workflow_id)
         .fetch_all(&self.pool)
         .await?;
@@ -1637,7 +1639,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             query.push_str(&format!(" OFFSET ${}", param_num));
         }
 
-        let mut sql_query = sqlx::query(&query);
+        let mut sql_query = sqlx::query(AssertSqlSafe(query));
 
         if let Some(queue) = queue_name {
             sql_query = sql_query.bind(queue);
@@ -1734,7 +1736,7 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
             base_query.push_str(" WHERE queue_name = $1");
         }
 
-        let mut query = sqlx::query(&base_query);
+        let mut query = sqlx::query(AssertSqlSafe(base_query));
         if let Some(queue) = queue_name {
             query = query.bind(queue);
         }
@@ -1895,12 +1897,12 @@ impl DatabaseQueue for crate::queue::JobQueue<Postgres> {
     async fn purge_expired_encrypted_jobs(&self) -> Result<super::EncryptedJobPurge> {
         let mut tx = self.pool.begin().await?;
         let now = db_now(&mut tx).await?;
-        let jobs = sqlx::query(&format!(
+        let jobs = sqlx::query(AssertSqlSafe(format!(
             "DELETE FROM hammerwork_jobs WHERE is_encrypted = true \
              AND retention_delete_at IS NOT NULL AND retention_delete_at <= $1 \
              AND status IN ({})",
             super::FINISHED_STATUS_SQL
-        ))
+        )))
         .bind(now)
         .execute(&mut *tx)
         .await?
@@ -2008,9 +2010,9 @@ impl crate::queue::JobQueue<Postgres> {
         // A reclaimed job that died is a terminal failure like any other: reschedule it
         // if recurring, otherwise apply it to its dependents, workflow and batch.
         for id in &recovery.dead {
-            let row = sqlx::query_as::<_, JobRow>(&format!(
+            let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
                 "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE id = $1"
-            ))
+            )))
             .bind(id)
             .fetch_one(&mut *conn)
             .await?;
@@ -2317,9 +2319,9 @@ impl TransitionStore for crate::queue::JobQueue<Postgres> {
     type Conn = sqlx::PgConnection;
 
     async fn lock_job(conn: &mut sqlx::PgConnection, job_id: JobId) -> Result<Option<Job>> {
-        let row = sqlx::query_as::<_, JobRow>(&format!(
+        let row = sqlx::query_as::<_, JobRow>(AssertSqlSafe(format!(
             "SELECT {JOB_SELECT_FIELDS} FROM hammerwork_jobs WHERE id = $1 FOR UPDATE"
-        ))
+        )))
         .bind(job_id)
         .fetch_optional(&mut *conn)
         .await?;
@@ -2701,11 +2703,11 @@ impl TransitionStore for crate::queue::JobQueue<Postgres> {
             Vec::new()
         };
 
-        let unfinished: bool = sqlx::query_scalar(&format!(
+        let unfinished: bool = sqlx::query_scalar(AssertSqlSafe(format!(
             "SELECT EXISTS (SELECT 1 FROM hammerwork_jobs WHERE batch_id = $1 \
              AND status IN ({}))",
             lifecycle::UNFINISHED_STATUS_SQL
-        ))
+        )))
         .bind(batch_id)
         .fetch_one(&mut *conn)
         .await?;
@@ -2812,7 +2814,7 @@ impl crate::queue::JobQueue<Postgres> {
         // Serializes with the terminal transitions of the workflow's jobs.
         Self::lock_workflow(conn, workflow_id).await?;
         let now = db_now(conn).await?;
-        sqlx::query(&format!(
+        sqlx::query(AssertSqlSafe(format!(
             r#"
             UPDATE hammerwork_jobs
             SET status = 'Failed',
@@ -2826,7 +2828,7 @@ impl crate::queue::JobQueue<Postgres> {
               AND status IN ({})
             "#,
             lifecycle::UNFINISHED_STATUS_SQL
-        ))
+        )))
         .bind(workflow_id)
         .bind(now)
         .bind(lifecycle::WORKFLOW_CANCELLED_MESSAGE)
@@ -2940,7 +2942,7 @@ impl crate::queue::JobQueue<Postgres> {
         while jobs_to_archive.len() < policy.batch_size {
             let wanted = policy.batch_size - jobs_to_archive.len();
             let select = select(wanted);
-            let mut query = sqlx::query_as::<_, JobRow>(&select);
+            let mut query = sqlx::query_as::<_, JobRow>(AssertSqlSafe(select));
             if let Some(queue) = queue_name {
                 query = query.bind(queue);
             }
@@ -3098,9 +3100,9 @@ impl crate::queue::JobQueue<Postgres> {
         let mut tx = self.pool.begin().await?;
 
         // Lock the archived row so two concurrent restores cannot both re-insert it.
-        let archived_row = sqlx::query(&format!(
+        let archived_row = sqlx::query(AssertSqlSafe(format!(
             "SELECT {ARCHIVED_JOB_FIELDS} FROM hammerwork_jobs_archive WHERE id = $1 FOR UPDATE"
-        ))
+        )))
         .bind(job_id)
         .fetch_optional(&mut *tx)
         .await?
diff --git a/tests/migration_idempotency_tests.rs b/tests/migration_idempotency_tests.rs
index 6f4fa53..dd5653b 100644
--- a/tests/migration_idempotency_tests.rs
+++ b/tests/migration_idempotency_tests.rs
@@ -57,7 +57,7 @@ mod mysql_tests {
 
         let mut admin = MySqlConnection::connect_with(&base).await.unwrap();
         admin
-            .execute(format!("CREATE DATABASE `{scratch}`").as_str())
+            .execute(sqlx::AssertSqlSafe(format!("CREATE DATABASE `{scratch}`")))
             .await
             .unwrap();
 
@@ -93,7 +93,9 @@ mod mysql_tests {
         .await;
 
         admin
-            .execute(format!("DROP DATABASE IF EXISTS `{scratch}`").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "DROP DATABASE IF EXISTS `{scratch}`"
+            )))
             .await
             .unwrap();
         outcome.unwrap();
@@ -143,7 +145,9 @@ mod postgres_tests {
 
         let mut admin = PgConnection::connect_with(&base).await.unwrap();
         admin
-            .execute(format!("CREATE DATABASE \"{scratch}\"").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "CREATE DATABASE \"{scratch}\""
+            )))
             .await
             .unwrap();
 
@@ -178,7 +182,9 @@ mod postgres_tests {
         .await;
 
         admin
-            .execute(format!("DROP DATABASE IF EXISTS \"{scratch}\" WITH (FORCE)").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "DROP DATABASE IF EXISTS \"{scratch}\" WITH (FORCE)"
+            )))
             .await
             .unwrap();
         outcome.unwrap();
diff --git a/tests/migration_lock_tests.rs b/tests/migration_lock_tests.rs
index 8247be7..0acac49 100644
--- a/tests/migration_lock_tests.rs
+++ b/tests/migration_lock_tests.rs
@@ -90,7 +90,7 @@ mod postgres_tests {
         .await
         .unwrap();
         assert_eq!(recorded, 1);
-        sqlx::query(&format!("DROP TABLE {table}"))
+        sqlx::query(sqlx::AssertSqlSafe(format!("DROP TABLE {table}")))
             .execute(pool)
             .await
             .expect("the probe migration created its table");
@@ -117,7 +117,9 @@ mod postgres_tests {
         let scratch = format!("hw_lock_{}", uuid::Uuid::new_v4().simple());
         let mut admin = PgConnection::connect_with(&base).await.unwrap();
         admin
-            .execute(format!("CREATE DATABASE \"{scratch}\"").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "CREATE DATABASE \"{scratch}\""
+            )))
             .await
             .unwrap();
 
@@ -146,7 +148,9 @@ mod postgres_tests {
         .await;
 
         admin
-            .execute(format!("DROP DATABASE IF EXISTS \"{scratch}\" WITH (FORCE)").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "DROP DATABASE IF EXISTS \"{scratch}\" WITH (FORCE)"
+            )))
             .await
             .unwrap();
         outcome.unwrap();
@@ -203,7 +207,7 @@ mod mysql_tests {
                 .await
                 .unwrap();
         assert_eq!(recorded, 1);
-        sqlx::query(&format!("DROP TABLE {table}"))
+        sqlx::query(sqlx::AssertSqlSafe(format!("DROP TABLE {table}")))
             .execute(pool)
             .await
             .expect("the probe migration created its table");
@@ -222,7 +226,7 @@ mod mysql_tests {
         let scratch = format!("hw_lock_{}", uuid::Uuid::new_v4().simple());
         let mut admin = MySqlConnection::connect_with(&base).await.unwrap();
         admin
-            .execute(format!("CREATE DATABASE `{scratch}`").as_str())
+            .execute(sqlx::AssertSqlSafe(format!("CREATE DATABASE `{scratch}`")))
             .await
             .unwrap();
 
@@ -251,7 +255,9 @@ mod mysql_tests {
         .await;
 
         admin
-            .execute(format!("DROP DATABASE IF EXISTS `{scratch}`").as_str())
+            .execute(sqlx::AssertSqlSafe(format!(
+                "DROP DATABASE IF EXISTS `{scratch}`"
+            )))
             .await
             .unwrap();
         outcome.unwrap();
diff --git a/tests/payload_encryption_tests.rs b/tests/payload_encryption_tests.rs
index 534b815..7e48c65 100644
--- a/tests/payload_encryption_tests.rs
+++ b/tests/payload_encryption_tests.rs
@@ -85,11 +85,13 @@ impl Inspect for sqlx::Postgres {
             })
             .collect::<Vec<_>>()
             .join(" || '|' || ");
-        let hex: String = sqlx::query_scalar(&format!("SELECT {parts} FROM {table} WHERE id = $1"))
-            .bind(id)
-            .fetch_one(pool)
-            .await
-            .unwrap();
+        let hex: String = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
+            "SELECT {parts} FROM {table} WHERE id = $1"
+        )))
+        .bind(id)
+        .fetch_one(pool)
+        .await
+        .unwrap();
         hex.to_lowercase()
     }
 
@@ -160,9 +162,9 @@ impl Inspect for sqlx::MySql {
             .map(|column| format!("COALESCE(HEX(`{column}`), '')"))
             .collect::<Vec<_>>()
             .join(", ");
-        let hex: String = sqlx::query_scalar(&format!(
+        let hex: String = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
             "SELECT CONCAT_WS('|', {parts}) FROM {table} WHERE id = ?"
-        ))
+        )))
         .bind(id.to_string())
         .fetch_one(pool)
         .await
diff --git a/tests/queue_followup_tests.rs b/tests/queue_followup_tests.rs
index 3953c01..877d332 100644
--- a/tests/queue_followup_tests.rs
+++ b/tests/queue_followup_tests.rs
@@ -581,9 +581,9 @@ mod postgres_tests {
             ("retry_strategy", "'{\"NoSuchStrategy\": {}}'::jsonb"),
         ] {
             let id = queue.enqueue(job(&queue_name, column)).await.unwrap();
-            sqlx::query(&format!(
+            sqlx::query(sqlx::AssertSqlSafe(format!(
                 "UPDATE hammerwork_jobs SET {column} = {value} WHERE id = $1"
-            ))
+            )))
             .bind(id)
             .execute(queue.get_pool())
             .await
@@ -669,9 +669,9 @@ mod mysql_tests {
             ),
         ] {
             let id = queue.enqueue(job(&queue_name, column)).await.unwrap();
-            sqlx::query(&format!(
+            sqlx::query(sqlx::AssertSqlSafe(format!(
                 "UPDATE hammerwork_jobs SET {column} = {value} WHERE id = ?"
-            ))
+            )))
             .bind(id.to_string())
             .execute(queue.get_pool())
             .await
diff --git a/tests/undecodable_row_tests.rs b/tests/undecodable_row_tests.rs
index dbea735..4b02ddb 100644
--- a/tests/undecodable_row_tests.rs
+++ b/tests/undecodable_row_tests.rs
@@ -133,7 +133,10 @@ async fn postgres_listings_skip_undecodable_rows() {
     listings_skip_undecodable_rows(queue.as_ref(), |statement| {
         let pool = pool.clone();
         async move {
-            sqlx::query(&statement).execute(&pool).await.unwrap();
+            sqlx::query(sqlx::AssertSqlSafe(statement))
+                .execute(&pool)
+                .await
+                .unwrap();
         }
     })
     .await;
@@ -150,7 +153,10 @@ async fn mysql_listings_skip_undecodable_rows() {
     listings_skip_undecodable_rows(queue.as_ref(), |statement| {
         let pool = pool.clone();
         async move {
-            sqlx::query(&statement).execute(&pool).await.unwrap();
+            sqlx::query(sqlx::AssertSqlSafe(statement))
+                .execute(&pool)
+                .await
+                .unwrap();
         }
     })
     .await;
@@ -272,7 +278,10 @@ fn postgres_sql(
     move |statement| {
         let pool = pool.clone();
         Box::pin(async move {
-            sqlx::query(&statement).execute(&pool).await.unwrap();
+            sqlx::query(sqlx::AssertSqlSafe(statement))
+                .execute(&pool)
+                .await
+                .unwrap();
         })
     }
 }
@@ -284,7 +293,10 @@ fn mysql_sql(
     move |statement| {
         let pool = pool.clone();
         Box::pin(async move {
-            sqlx::query(&statement).execute(&pool).await.unwrap();
+            sqlx::query(sqlx::AssertSqlSafe(statement))
+                .execute(&pool)
+                .await
+                .unwrap();
         })
     }
 }