diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index eaa318e4c..47b064be7 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -270,6 +270,45 @@ jobs: env: AMQP_ORDERING_TEST: "true" + test-mariadb-kv: + name: MariaDB KV (${{ matrix.runtime }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + runtime: [node, bun, deno] + services: + mariadb: + image: mariadb:11.8 + env: + MARIADB_ROOT_PASSWORD: mariadb + MARIADB_DATABASE: fedify + ports: + - 3306:3306 + options: >- + --health-cmd "healthcheck.sh --connect --innodb_initialized" + --health-interval 10s + --health-timeout 5s + --health-retries 10 + env: + MISE_DISABLE_TOOLS: ${{ matrix.runtime != 'deno' && 'deno' || '' }} + MYSQL_URL: mysql://root:mariadb@127.0.0.1:3306/fedify + steps: + - uses: actions/checkout@v6 + - uses: ./.github/actions/setup-mise + - name: Test MariaDB KV on Node.js + if: matrix.runtime == 'node' + run: mise exec -- node --experimental-transform-types --test packages/mysql/src/kv.test.ts packages/mysql/src/atomic-kv.test.ts + - name: Test MariaDB KV on Bun + if: matrix.runtime == 'bun' + run: mise exec -- bun test --timeout=60000 packages/mysql/src/kv.test.ts packages/mysql/src/atomic-kv.test.ts + - name: Prepare Deno workspace imports + if: matrix.runtime == 'deno' + run: mise run deno:cache + - name: Test MariaDB KV on Deno + if: matrix.runtime == 'deno' + run: mise run --skip-deps test:deno packages/mysql/src/kv.test.ts packages/mysql/src/atomic-kv.test.ts + test-cfworkers: runs-on: ubuntu-latest env: diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..5f7bf7f2d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -381,6 +381,25 @@ To be released. [#1356]: https://github.com/fedify-dev/fedify/pull/1356 [#1420]: https://github.com/fedify-dev/fedify/pull/1420 +### @fedify/mysql + + - Added `AtomicKvStore` support to `MysqlKvStore`: `getMany()`, `setMany()`, + `deleteMany()`, and `casMany()` read or change related keys together on + MySQL and MariaDB InnoDB tables. [[#1361], [#1368], [#1433]] + + Single-key operations use the same transactions and JSON codec, so reads + include transaction setup and commit round trips. `set(key, undefined)` + now rejects; use deletion instead. Initialization validates the schema + even with `initialized: true`, and custom pool connections need + `destroy()`. Use `autocommit`, default mysql2 decoding and one fixed-offset + or UTC session time zone; `NO_BACKSLASH_ESCAPES` is unsupported. Batches + accept up to 1,000 entries and serialized keys up to 768 Unicode code + points. TTLs are truncated to safe integer microseconds, approximately + ±285 years. See the package README for schema and server restrictions. + +[#1368]: https://github.com/fedify-dev/fedify/issues/1368 +[#1433]: https://github.com/fedify-dev/fedify/pull/1433 + ### @fedify/nestjs - Fixed the `ContextDataFactory` type to declare the Express `Request` and diff --git a/changes.d/mysql/atomic-key-value-operations.md b/changes.d/mysql/atomic-key-value-operations.md new file mode 100644 index 000000000..516e48da1 --- /dev/null +++ b/changes.d/mysql/atomic-key-value-operations.md @@ -0,0 +1,19 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1368': https://github.com/fedify-dev/fedify/issues/1368 + '#1433': https://github.com/fedify-dev/fedify/pull/1433 +--- + - Added `AtomicKvStore` support to `MysqlKvStore`: `getMany()`, `setMany()`, + `deleteMany()`, and `casMany()` read or change related keys together on + MySQL and MariaDB InnoDB tables. [[#1361], [#1368], [#1433]] + + Single-key operations use the same transactions and JSON codec, so reads + include transaction setup and commit round trips. `set(key, undefined)` + now rejects; use deletion instead. Initialization validates the schema + even with `initialized: true`, and custom pool connections need + `destroy()`. Use `autocommit`, default mysql2 decoding and one fixed-offset + or UTC session time zone; `NO_BACKSLASH_ESCAPES` is unsupported. Batches + accept up to 1,000 entries and serialized keys up to 768 Unicode code + points. TTLs are truncated to safe integer microseconds, approximately + ±285 years. See the package README for schema and server restrictions. diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b7f3a6665..a7c7f6a04 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -471,6 +471,19 @@ atomicity or durability. `getMany()` returns an input-ordered snapshot; `casMany()` checks every expectation before committing all changes. On `false`, reread and re-evaluate application eligibility before retrying. +`MysqlKvStore` implements `AtomicKvStore` on an InnoDB table in one primary +database. It supports up to 1,000 entries per call and serialized keys up to +768 Unicode code points. Use a normal mysql2 pool with `autocommit` enabled, +default object/type decoding, and one fixed-offset or UTC session time zone +across clients. Initialization validates the schema even with +`initialized: true`; do not use caller-owned pool transactions, replica reads, +or change the schema during operations. Values round-trip through JSON and +CAS compares decoded structures; `set(undefined)` rejects and omitted TTLs +clear expiry. TTLs are truncated to safe integer microseconds. Deadlocks and +MariaDB snapshot conflicts retry with fresh comparisons up to five attempts; +backend errors reject, and lost commit responses can have unknown outcomes. +See *packages/mysql/README.md* for the full schema, codec and server limits. + `expectedValue: undefined` means absence; `newValue: undefined` means deletion in CAS. Atomic stores reject `set(key, undefined)` and `setMany()` entries whose value is `undefined`. They also reject TTLs on deletions and duplicate @@ -482,17 +495,17 @@ for a shared backend. Separate calls do not form a transaction; uncertain commit results need receipts or version checks, and ABA detection needs a non-reused version. -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full atomic -interface. -SQLite covers one configured table/database on a writable, normally locking -connection. Nonempty batches and ordinary mutations reject caller-owned -transactions and use one transaction; snapshots and mismatching CAS can reject -busy, including under rollback-journal reader contention. Configure busy -timeout on the connection; the KV store has no retry policy. Use decoded -JSON expectations, structural comparison, and the documented millisecond TTL -precision and resource limits. `SqliteMessageQueue` defaults to WAL on shared -files; concurrent WAL use needs a patched SQLite engine (Bun 1.2.22 -SQLite 3.50.4 is affected by the WAL-reset bug). See the adapter documentation. +`MemoryKvStore`, `SqliteKvStore`, `PgliteKvStore`, and `MysqlKvStore` implement +the full atomic interface. SQLite covers one configured table/database on a +writable, normally locking connection. Nonempty batches and ordinary mutations +reject caller-owned transactions and use one transaction; snapshots and +mismatching CAS can reject busy, including under rollback-journal reader +contention. Configure busy timeout on the connection; the KV store has no +retry policy. Use decoded JSON expectations, structural comparison, and the +documented millisecond TTL precision and resource limits. `SqliteMessageQueue` +defaults to WAL on shared files; concurrent WAL use needs a patched SQLite +engine (Bun 1.2.22 SQLite 3.50.4 is affected by the WAL-reset bug). See the +adapter documentation. For PGlite, share one normal owner and table across wrappers, require standard JSONB/text-array codecs, and keep raw/manual transaction control and @@ -504,8 +517,8 @@ engine insertion failures roll back, while mismatch does not prove proposed writes would fit. Do not infer crash durability, browser persistence, or worker-proxy support. Read the PGlite package README's synchronization policy. -The PostgreSQL, Redis, and MySQL atomic extensions are tracked separately in -issues #1364, #1365, and #1368; Deno KV and Netlify Blobs have no atomic adapter +The PostgreSQL and Redis atomic extensions are tracked separately in +issues #1364 and #1365; Deno KV and Netlify Blobs have no atomic adapter child issue. `WorkersKvStore` lacks CAS. `WorkersD1KvStore` supports single-key CAS; its atomic extension is tracked in #1379. Check the [atomic support matrix] and installed declarations before selecting a diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 672166774..af77c542d 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -124,8 +124,8 @@ Concurrent WAL writes/checkpoints require the [SQLite WAL-reset fix] (3.51.3 or the 3.50.7/3.44.6 backports, or a verified equivalent fix). Bun 1.2.22's SQLite 3.50.4 is affected. `SqliteMessageQueue` enables WAL by default on shared database files, so the prerequisite also applies to combined -KV and queue deployments. See the [package README] for the complete value, -expiry, connection and durability restrictions. +KV and queue deployments. See the [SQLite package README] for the complete +value, expiry, connection and durability restrictions. Best for : Development and testing. @@ -199,7 +199,7 @@ const federation = createFederation({ [`bun:sqlite`]: https://bun.com/docs/api/sqlite [busy timeout]: https://sqlite.org/c3ref/busy_timeout.html [SQLite WAL-reset fix]: https://sqlite.org/wal.html#walresetbug -[package README]: https://github.com/fedify-dev/fedify/tree/main/packages/sqlite#atomic-operations +[SQLite package README]: https://github.com/fedify-dev/fedify/tree/main/packages/sqlite#atomic-operations [conditional exports]: https://nodejs.org/api/packages.html#conditional-exports ### [`RedisKvStore`] @@ -563,7 +563,33 @@ const federation = createFederation({ }); ~~~~ +Since Fedify 2.5.0, `MysqlKvStore` implements `AtomicKvStore`, with atomic +snapshots and conditional batches across keys in one primary database/table. +It uses InnoDB transactions with explicit `REPEATABLE READ`, including for +single-key reads and CAS. Use `getMany()` for a snapshot and `casMany()` to +compare every expectation before committing all replacements or deletions. +A mismatch returns `false`; reread and re-evaluate before retrying. + +The adapter validates its InnoDB schema even with `initialized: true`, which +skips DDL but requires metadata visibility. Use the default schema, a normal +mysql2 pool with `autocommit` enabled and default object/type decoding, and the +same fixed-offset or UTC session `time_zone` across clients. Caller-owned +transactions, replica routing, `NO_BACKSLASH_ESCAPES`, custom decoding, and +schema changes during use are unsupported. Values use JSON conversion and +decoded structural equality; `set(key, undefined)` rejects. Batches have at +most 1,000 input entries, and serialized keys at most 768 Unicode code points. +TTLs are truncated to safe integer microseconds, approximately ±285 years; +omitting a replacement TTL clears its previous expiry. + +Deadlocks and MariaDB snapshot conflicts retry with fresh comparisons, up to +five total attempts. Other backend failures reject, and a lost commit +response can leave the outcome unknown. Cleanup runs after commit; its +failures are logged, while logical expiry always applies. See the +[MySQL package README] +for schema, codec, clock, cleanup and resource restrictions. + [`MysqlKvStore`]: https://jsr.io/@fedify/mysql/doc/kv/~/MysqlKvStore +[MySQL package README]: https://github.com/fedify-dev/fedify/blob/main/packages/mysql/README.md#atomic-key-value-operations ### `DenoKvStore` (Deno only) @@ -949,9 +975,6 @@ from `list()` too. `set()` and `setMany()` reject `undefined` on an `AtomicKvStore`; skip the write to preserve the former no-op, or use `delete()` or `deleteMany()` for unconditional removal. For conditional removal, use `cas()` or `casMany()` with `newValue: undefined` and omit the TTL. -These stronger rules do not change legacy ordinary stores: for example, MySQL -currently treats -`set(key, undefined)` as a no-op. Each set or conditional entry can supply its own `options.ttl`: @@ -1008,10 +1031,10 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full -`AtomicKvStore` contract. -Existing single-key CAS support does not imply atomic multi-key support. -The linked issues track separate adapter implementations: +`MemoryKvStore`, `SqliteKvStore`, `PgliteKvStore`, and `MysqlKvStore` implement +the full `AtomicKvStore` contract. Existing single-key CAS support does not +imply atomic multi-key support. The linked issues track separate adapter +implementations: | Implementation | `AtomicKvStore` | Status | | --------------------- | --------------- | ------------------------------------------------------------------------------------------------- | @@ -1020,7 +1043,7 @@ The linked issues track separate adapter implementations: | `RedisKvStore` | No | Planned in [#1365]. | | `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | | `SqliteKvStore` | Yes | One configured table/database; writable handles, JSON values; see [SQLite scope](#sqlitekvstore). | -| `MysqlKvStore` | No | Planned in [#1368]. | +| `MysqlKvStore` | Yes | InnoDB; one primary database/table, up to 1,000 entries per operation. | | `DenoKvStore` | No | No atomic adapter child issue; deprecated since 2.5.0. | | `NetlifyBlobsKvStore` | No | Single-key CAS only; no atomic adapter child issue. | | `WorkersKvStore` | No | No CAS; cannot implement this contract with Workers KV. | @@ -1029,7 +1052,6 @@ The linked issues track separate adapter implementations: [PR #1360]: https://github.com/fedify-dev/fedify/pull/1360 [#1364]: https://github.com/fedify-dev/fedify/issues/1364 [#1365]: https://github.com/fedify-dev/fedify/issues/1365 -[#1368]: https://github.com/fedify-dev/fedify/issues/1368 [#1379]: https://github.com/fedify-dev/fedify/issues/1379 diff --git a/packages/mysql/README.md b/packages/mysql/README.md index 2acf7daf5..a2a0587da 100644 --- a/packages/mysql/README.md +++ b/packages/mysql/README.md @@ -44,3 +44,108 @@ yarn add @fedify/mysql mysql2 # Yarn bun add @fedify/mysql mysql2 # Bun deno add jsr:@fedify/mysql # Deno ~~~~ + + +Atomic key-value operations +--------------------------- + +Since Fedify 2.5.0, `MysqlKvStore` implements `AtomicKvStore`. Use `getMany()` +to read related keys from one snapshot, `setMany()` or `deleteMany()` to change +them together, and `casMany()` to commit only when every expectation matches. +Single-key `cas()` follows the same rules as `casMany()`. After a `false` +result, read again and re-evaluate the application rule before retrying. + +~~~~ typescript +const store = new MysqlKvStore(pool); +await store.setMany([ + { key: ["cursor"], value: 0 }, + { key: ["budget"], value: 10 }, +]); +const [cursor, budget] = await store.getMany([["cursor"], ["budget"]]); +const committed = await store.casMany([ + { key: ["cursor"], expectedValue: cursor, newValue: (cursor ?? 0) + 1 }, + { key: ["budget"], expectedValue: budget, newValue: (budget ?? 0) - 1 }, +]); +~~~~ + +### Scope and configuration + +All keys in the same primary database and table share an atomic scope across +independent pools and processes. Each operation runs in its own transaction on +one pooled connection, with explicit `REPEATABLE READ` isolation. `get()` uses +the same snapshot transaction as `getMany()`, so reads now require transaction +setup and commit round trips. Successive calls do not form one transaction. A +failed commit response can leave its outcome unknown; use a non-reused version +or an application receipt before repeating external effects. Crash durability +depends on the server and storage configuration. + +Use a normal mysql2 pool with `autocommit` enabled, object rows and default type +casting. The `jsonStrings` option is supported. Do not supply connections +already in caller-owned transactions, share borrowed connections with raw SQL, +or route reads to replicas. All cooperating clients must use the same +fixed-offset or UTC session `time_zone`; `DATETIME` expiry values do not carry +a time zone. `NO_BACKSLASH_ESCAPES`, custom `typeCast`, and `rowsAsArray` are +unsupported. Keep server settings and the table schema stable while stores +are in use. `initialize()` and `drop()` must not overlap other operations. + +Initialization requires an InnoDB base table with the adapter's schema: a +`VARCHAR(768)` key in `utf8mb4_bin`, a full-column primary `BTREE` index with +no other unique indexes, a non-null native JSON or `utf8mb4` `LONGTEXT` value, +and a nullable `DATETIME(6)` expiry. It rejects incompatible existing tables +instead of migrating them. `initialized: true` skips DDL, but still requires +metadata visibility for schema validation. Custom pool-like objects must +provide mysql2's connection `destroy()` method. The legacy +`innodb_locks_unsafe_for_binlog` setting must be disabled if present. + +### Values, TTLs and limits + +Values use `JSON.stringify()` on write and `JSON.parse()` on every read path, +including `list()`. JSON primitives, arrays and objects round-trip; dates and +`toJSON()` values follow normal JSON conversion. Unsupported object members +are omitted, unsupported array members and non-finite numbers become `null`, +and cycles, `BigInt` and values without a root JSON representation reject. +`set(key, undefined)` now rejects; use `delete()` instead. CAS compares decoded +values with `isEqual()`, ignoring object member order and preserving array +order and type distinctions. Expectations are not JSON-normalized: use the ISO +string returned by `get()` as the expected value for a stored date, rather than +a `Date`. Do not mutate expected values while an operation is pending. + +Each batch accepts at most 1,000 input entries, including repeated keys in a +read. Mutation keys must be distinct. A JSON-serialized key, including its +punctuation and escapes, has at most 768 Unicode code points. Server packet, +JSON nesting, transaction and runtime resource limits also apply. Operations +reject rather than split a batch to fit a limit. + +A stored `null` is distinct from absence. Expired entries count as absent; +`expectedValue: undefined` checks absence and `newValue: undefined` deletes. +Deletion rejects a non-null TTL, even when an expectation would mismatch. +Each replacement has its own TTL; omitting it clears previous expiry. TTLs +are truncated to safe integer microseconds, approximately ±285 years; larger +or server-unrepresentable durations reject. Calendar units use the current +local date for conversion. Non-positive TTLs expire immediately. One +server cutoff applies to each attempt: CAS captures it after acquiring all +locks, and unconditional writes capture it before any write-side lock waits, so +a short TTL can already have elapsed at commit. + +### Backend verification + +CI runs the ordinary and atomic key-value suites against MariaDB 11.8 on +Node.js, Bun, and Deno, alongside the full-package MySQL 8 jobs. The MariaDB +jobs run only the key-value suites because the message-queue suite has +pre-existing MariaDB compatibility failures. + +### Conflicts and cleanup + +CAS locks physical rows and absent key ranges before comparing. Expired rows +remain locked, and all expectations match before the first mutation. Deadlocks +(`1213`) and MariaDB snapshot conflicts (`1020`) retry the whole transaction, +with fresh comparisons and cutoff, for at most five total attempts. Other +backend errors reject. Lock timeout `1205` rejects after rollback; disabling +`innodb_deadlock_detect` can turn contention into such timeouts. A comparison +mismatch alone returns `false` and preserves values and TTLs. + +Expiry cleanup runs after commit and after releasing the transaction's +connection. It is awaited and may wait for the server's lock timeout, but its +failures are logged without changing the committed result. The existing +`expireCleanupRate` option controls this housekeeping cost; `0` disables it. +Logical expiry works regardless of physical cleanup. diff --git a/packages/mysql/src/atomic-kv.test.ts b/packages/mysql/src/atomic-kv.test.ts new file mode 100644 index 000000000..47686e90b --- /dev/null +++ b/packages/mysql/src/atomic-kv.test.ts @@ -0,0 +1,1127 @@ +// Transaction schedules and lifecycle cleanup must run sequentially. +/* oxlint-disable eslint/no-await-in-loop */ +import { + isAtomicKvStore, + type KvKey, + type KvStoreCasEntry, + type KvStoreSetEntry, +} from "@fedify/fedify"; +import { MysqlKvStore } from "@fedify/mysql/kv"; +import { testAtomicKvStore, testKvStore } from "@fedify/testing"; +import * as temporal from "@js-temporal/polyfill"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { test } from "node:test"; +import mysql from "mysql2/promise"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const dbUrl = process.env.MYSQL_URL; +const live = { skip: dbUrl == null, timeout: 180_000 }; +const stamp = (): string => `atomic_${crypto.randomUUID().replace(/-/g, "")}`; +const key = (value: string): KvKey => [value]; + +async function pair(isolation?: string) { + const firstPool = mysql.createPool(dbUrl!); + const secondPool = mysql.createPool(dbUrl!); + const tableName = stamp(); + const wrap = (pool: mysql.Pool) => ({ + query: pool.query.bind(pool), + async getConnection() { + const conn = await pool.getConnection(); + if (isolation != null) + await conn.query( + `SET SESSION TRANSACTION ISOLATION LEVEL ${isolation}`, + ); + return conn; + }, + }); + const first = new MysqlKvStore(wrap(firstPool), { + tableName, + expireCleanupRate: 0, + }); + const second = new MysqlKvStore(wrap(secondPool), { + tableName, + expireCleanupRate: 0, + }); + const dispose = async () => { + try { + await first.drop(); + } finally { + await Promise.all([firstPool.end(), secondPool.end()]); + } + }; + try { + await first.initialize(); + await second.initialize(); + return { first, second, firstPool, secondPool, tableName, dispose }; + } catch (error) { + await dispose(); + throw error; + } +} + +test("MysqlKvStore AtomicKvStore conformance", live, async () => { + for (const isolation of [undefined, "READ COMMITTED", "READ UNCOMMITTED"]) { + await testAtomicKvStore(() => pair(isolation), { + objectKeyOrder: "ignored", + // Seeding and updating TTLs takes several transactions on a shared runner. + ttl: { + short: Temporal.Duration.from({ seconds: 5 }), + wait: Temporal.Duration.from({ seconds: 6 }), + }, + }); + } +}); + +test("MysqlKvStore ordinary KV conformance", live, async () => { + const p = await pair(); + await testKvStore( + () => p.first, + () => p.dispose(), + ); +}); + +test("MysqlKvStore validates complete batches before I/O", async () => { + const store = new MysqlKvStore({ + query() { + throw new Error("Unexpected I/O"); + }, + getConnection() { + throw new Error("Unexpected I/O"); + }, + } as unknown as mysql.Pool); + assert.ok(isAtomicKvStore(store)); + const k = key("a"); + await assert.rejects(store.set(k, undefined), TypeError); + await assert.rejects( + store.setMany([ + { key: k, value: 1 }, + { key: k, value: 2 }, + ]), + TypeError, + ); + await assert.rejects(store.deleteMany([k, k]), TypeError); + await assert.rejects( + store.casMany([ + { key: k, expectedValue: "mismatch", newValue: 2 }, + { key: k, expectedValue: undefined, newValue: 1 }, + ]), + TypeError, + ); + await assert.rejects( + store.cas(k, "mismatch", undefined, { + ttl: Temporal.Duration.from({ seconds: 1 }), + }), + TypeError, + ); + await assert.rejects( + store.setMany([ + { key: k, value: 1 }, + { key: key("b"), value: undefined }, + ]), + TypeError, + ); + const cycle: { self?: unknown } = {}; + cycle.self = cycle; + for (const value of [undefined, () => 1, Symbol("bad"), 1n, cycle]) { + await assert.rejects(store.set(k, value), TypeError); + } + for (const entries of [new Array(1), [k, , k]]) { + await assert.rejects(store.getMany(entries as KvKey[]), TypeError); + await assert.rejects(store.deleteMany(entries as KvKey[]), TypeError); + } + await assert.rejects(store.setMany(new Array(2)), TypeError); + await assert.rejects(store.casMany(new Array(2)), TypeError); + await assert.rejects( + store.getMany(Array.from({ length: 1001 }, () => k)), + RangeError, + ); + const oversized = Array.from({ length: 1001 }, (_, i) => key(String(i))); + await assert.rejects( + store.setMany(oversized.map((key) => ({ key, value: 1 }))), + RangeError, + ); + await assert.rejects(store.deleteMany(oversized), RangeError); + await assert.rejects( + store.casMany( + oversized.map((key) => ({ key, expectedValue: undefined, newValue: 1 })), + ), + RangeError, + ); + await assert.rejects(store.set(["x".repeat(765)], 1), RangeError); + await assert.rejects( + store.cas(k, "mismatch", 1, { + ttl: Temporal.Duration.from({ years: 1000 }), + }), + RangeError, + ); + assert.deepEqual(await store.getMany([]), []); + await store.setMany([]); + await store.deleteMany([]); + assert.equal(await store.casMany([]), true); +}); + +test( + "MysqlKvStore rolls back real writes and retains expiry on a later failure", + live, + async () => { + const p = await pair(); + try { + await p.first.setMany([ + { + key: key("a"), + value: 1, + options: { ttl: Temporal.Duration.from({ hours: 1 }) }, + }, + { key: key("b"), value: 2 }, + ]); + const [before] = await p.firstPool.query( + `SELECT DATE_FORMAT(expires, '%Y-%m-%d %H:%i:%s.%f') AS expires FROM \`${p.tableName}\` WHERE \`key\`=?`, + [JSON.stringify(key("a"))], + ); + let writes = 0; + let mutation = "INSERT"; + const failing = new MysqlKvStore( + { + query: p.firstPool.query.bind(p.firstPool), + async getConnection() { + const conn = await p.firstPool.getConnection(); + return { + async query( + sql: string, + values?: unknown, + ): Promise<[T, unknown]> { + if (sql.startsWith(mutation) && ++writes === 2) + throw new Error("injected second write"); + return (await conn.query(sql, parameters(values))) as [ + T, + unknown, + ]; + }, + beginTransaction: conn.beginTransaction.bind(conn), + commit: conn.commit.bind(conn), + rollback: conn.rollback.bind(conn), + release: conn.release.bind(conn), + destroy: conn.destroy.bind(conn), + }; + }, + }, + { tableName: p.tableName, initialized: true, expireCleanupRate: 0 }, + ); + await assert.rejects( + failing.setMany([ + { key: key("a"), value: 10 }, + { key: key("b"), value: 20 }, + ]), + /injected second write/, + ); + assert.deepEqual(await p.second.getMany([key("a"), key("b")]), [1, 2]); + writes = 0; + await assert.rejects( + failing.casMany([ + { key: key("a"), expectedValue: 1, newValue: 10 }, + { key: key("b"), expectedValue: 2, newValue: 20 }, + ]), + /injected second write/, + ); + writes = 0; + mutation = "DELETE"; + await assert.rejects( + failing.deleteMany([key("a"), key("b")]), + /injected second write/, + ); + assert.deepEqual(await p.second.getMany([key("a"), key("b")]), [1, 2]); + const [after] = await p.firstPool.query( + `SELECT DATE_FORMAT(expires, '%Y-%m-%d %H:%i:%s.%f') AS expires FROM \`${p.tableName}\` WHERE \`key\`=?`, + [JSON.stringify(key("a"))], + ); + assert.deepEqual(after, before); + } finally { + await p.dispose(); + } + }, +); + +function gate() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +test( + "MysqlKvStore holds an absent range even for a CAS that never inserts that key", + live, + async () => { + const p = await pair(); + const locked = gate(); + const resume = gate(); + let operation: Promise | undefined; + let writer: Promise | undefined; + try { + await p.first.set(key("b"), 1); + const paused = new MysqlKvStore( + { + query: p.firstPool.query.bind(p.firstPool), + async getConnection() { + const conn = await p.firstPool.getConnection(); + return { + async query( + sql: string, + values?: unknown, + ): Promise<[T, unknown]> { + if (sql.startsWith("SELECT DATE_FORMAT(NOW")) { + locked.resolve(); + await resume.promise; + } + return (await conn.query(sql, parameters(values))) as [ + T, + unknown, + ]; + }, + beginTransaction: conn.beginTransaction.bind(conn), + commit: conn.commit.bind(conn), + rollback: conn.rollback.bind(conn), + release: conn.release.bind(conn), + destroy: conn.destroy.bind(conn), + }; + }, + }, + { tableName: p.tableName, initialized: true, expireCleanupRate: 0 }, + ); + operation = paused.casMany([ + { key: key("a"), expectedValue: undefined, newValue: undefined }, + { key: key("b"), expectedValue: 1, newValue: 2 }, + ]); + await Promise.race([locked.promise, operation]); + let settled = false; + writer = p.second.set(key("a"), "writer").finally(() => { + settled = true; + }); + await new Promise((done) => setTimeout(done, 300)); + assert.equal( + settled, + false, + "independent writer must wait for the absent-range lock", + ); + const [processes] = await p.firstPool.query( + "SHOW FULL PROCESSLIST", + ); + assert.ok( + processes.some((row) => + String(row.Info).includes(`INSERT INTO \`${p.tableName}\``), + ), + "writer query must actually have reached the server", + ); + resume.resolve(); + assert.equal(await operation, true); + await writer; + assert.deepEqual(await p.first.getMany([key("a"), key("b")]), [ + "writer", + 2, + ]); + } finally { + resume.resolve(); + await Promise.allSettled([operation, writer]); + await p.dispose(); + } + }, +); + +test( + "MysqlKvStore one-slot pools release before cleanup and keep committed success on cleanup failure", + live, + async () => { + const pool = mysql.createPool({ uri: dbUrl!, connectionLimit: 1 }); + const tableName = stamp(); + const store = new MysqlKvStore(pool, { tableName }); + try { + await store.set(key("a"), 1, { + ttl: Temporal.Duration.from({ hours: 1 }), + }); + assert.equal( + await store.casMany([{ key: key("a"), expectedValue: 1, newValue: 2 }]), + true, + ); + const brokenCleanup = new MysqlKvStore( + { + async query(sql: string, values?: unknown): Promise<[T, unknown]> { + if (sql.startsWith("DELETE") && sql.includes("NOW(6)")) + throw new Error("cleanup failure"); + return (await pool.query(sql, parameters(values))) as [T, unknown]; + }, + getConnection: pool.getConnection.bind(pool), + }, + { tableName, initialized: true }, + ); + await brokenCleanup.set(key("a"), 3); + assert.equal(await store.get(key("a")), 3); + } finally { + await store.drop(); + await pool.end(); + } + }, +); + +test( + "MysqlKvStore rejects incompatible precreated tables without altering them", + live, + async () => { + const p = await pair(); + const names: string[] = []; + try { + for (const definition of [ + { + engine: "MyISAM", + value: "JSON NOT NULL", + primary: "PRIMARY KEY (`key`(100))", + expiry: "DATETIME(6) NULL", + }, + { + engine: "InnoDB", + value: "LONGTEXT CHARACTER SET latin1 NOT NULL", + primary: "PRIMARY KEY (`key`)", + expiry: "DATETIME(6) NULL", + }, + { + engine: "InnoDB", + value: "LONGTEXT CHARACTER SET utf8mb3 NOT NULL", + primary: "PRIMARY KEY (`key`)", + expiry: "DATETIME(6) NULL", + }, + { + engine: "InnoDB", + value: "JSON NOT NULL", + primary: "PRIMARY KEY (`key`(100))", + expiry: "DATETIME(6) NULL", + }, + { + engine: "InnoDB", + value: "JSON NOT NULL", + primary: "PRIMARY KEY (`key`), UNIQUE (expires)", + expiry: "DATETIME(6) NULL", + }, + { + engine: "InnoDB", + value: "JSON NOT NULL", + primary: "PRIMARY KEY (`key`)", + expiry: "DATETIME(6) NOT NULL", + }, + ]) { + const name = stamp(); + names.push(name); + await p.firstPool.query( + `CREATE TABLE \`${name}\` (\`key\` VARCHAR(768) NOT NULL, value ${definition.value}, expires ${definition.expiry}, ${definition.primary}) ENGINE=${definition.engine} DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`, + ); + const [before] = await p.firstPool.query( + `SHOW CREATE TABLE \`${name}\``, + ); + for (const initialized of [true, false]) { + const store = new MysqlKvStore(p.firstPool, { + tableName: name, + initialized, + }); + await assert.rejects(store.set(key("a"), 1), TypeError); + } + const [after] = await p.firstPool.query( + `SHOW CREATE TABLE \`${name}\``, + ); + assert.deepEqual( + after, + before, + "rejection must not alter an existing schema", + ); + const [rows] = await p.firstPool.query( + `SELECT COUNT(*) AS count FROM \`${name}\``, + ); + assert.equal(Number(rows[0].count), 0); + } + } finally { + for (const name of names) + await p.firstPool.query(`DROP TABLE IF EXISTS \`${name}\``); + await p.dispose(); + } + }, +); + +test( + "MysqlKvStore JSON, Unicode key boundary, TTL precision and raw expected semantics", + live, + async () => { + const p = await pair(); + try { + const boundary: KvKey = ["😀".repeat(764)]; // Four JSON punctuation characters. + assert.equal([...JSON.stringify(boundary)].length, 768); + await p.first.set(boundary, { + text: "🦄한글", + nested: [null, '"json"', 0.1], + }); + assert.deepEqual(await p.second.get(boundary), { + text: "🦄한글", + nested: [null, '"json"', 0.1], + }); + await assert.rejects(p.first.set(["😀".repeat(765)], 1), RangeError); + const listed = []; + for await (const entry of p.second.list()) listed.push(entry.value); + assert.deepEqual(listed, [ + { text: "🦄한글", nested: [null, '"json"', 0.1] }, + ]); + const date = new Date("2020-01-01T00:00:00Z"); + await p.first.set(key("date"), date); + assert.equal(await p.first.cas(key("date"), date, 1), false); + assert.equal(await p.first.cas(key("date"), date.toJSON(), 1), true); + await p.first.set(key("micro"), 1, { + ttl: Temporal.Duration.from({ microseconds: 500001, nanoseconds: 999 }), + }); + await p.first.set(key("micro"), 2); + const [rows] = await p.firstPool.query( + `SELECT expires FROM \`${p.tableName}\` WHERE \`key\`=?`, + [JSON.stringify(key("micro"))], + ); + assert.equal(rows[0].expires, null); + await p.first.set(key("zero"), 1, { + ttl: Temporal.Duration.from({ nanoseconds: 999 }), + }); + assert.equal(await p.second.get(key("zero")), undefined); + } finally { + await p.dispose(); + } + }, +); + +interface FakeOptions { + query?: (sql: string, values: unknown, attempt: number) => unknown[] | void; + begin?: () => void; + commit?: () => void; + rollback?: () => void; + metadata?: (sql: string) => unknown[] | undefined; +} + +function fakePool(options: FakeOptions = {}) { + const events: string[] = []; + let attempts = 0; + const pool = { + query(sql: string): Promise<[T, unknown]> { + events.push(`pool:${sql}`); + let rows = options.metadata?.(sql); + if (rows == null) { + if (sql.includes("information_schema.TABLES")) + rows = [{ engine: "InnoDB", table_type: "BASE TABLE" }]; + else if (sql.includes("information_schema.COLUMNS")) + rows = [ + { + name: "key", + type: "varchar", + nullable: "NO", + length: 768, + charset: "utf8mb4", + collation: "utf8mb4_bin", + }, + { name: "value", type: "json", nullable: "NO", charset: null }, + { + name: "expires", + type: "datetime", + nullable: "YES", + precision: 6, + }, + ]; + else if (sql.includes("information_schema.STATISTICS")) + rows = [ + { + name: "PRIMARY", + column_name: "key", + position: 1, + prefix: null, + type: "BTREE", + }, + ]; + else rows = []; + } + return Promise.resolve<[T, unknown]>([rows as T, undefined]); + }, + getConnection() { + const attempt = ++attempts; + events.push("acquire"); + return Promise.resolve({ + query(sql: string, values?: unknown): Promise<[T, unknown]> { + events.push(sql); + let rows = options.query?.(sql, values, attempt); + if (rows == null) + rows = sql.startsWith("SELECT DATE_FORMAT(NOW") + ? [{ cutoff: "2026-11-12 12:00:00.123456" }] + : []; + return Promise.resolve<[T, unknown]>([rows as T, undefined]); + }, + beginTransaction() { + events.push("begin"); + options.begin?.(); + return Promise.resolve(); + }, + commit() { + events.push("commit"); + options.commit?.(); + return Promise.resolve(); + }, + rollback() { + events.push("rollback"); + options.rollback?.(); + return Promise.resolve(); + }, + release() { + events.push("release"); + }, + destroy() { + events.push("destroy"); + }, + }); + }, + }; + return { + pool, + events, + get attempts() { + return attempts; + }, + }; +} + +for (const conflict of [1213, 1020]) { + test(`MysqlKvStore retries ${conflict} from a fresh transaction and rechecks expectations`, async () => { + const f = fakePool({ + query(sql, _values, attempt) { + if (sql.includes("FOR UPDATE")) { + if (attempt === 1) + throw Object.assign(new Error("retry conflict"), { + errno: conflict, + }); + return [{ key: '["a"]', value: '"winner"', expires: null }]; + } + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + assert.equal(await store.cas(key("a"), undefined, "loser"), false); + assert.equal(f.attempts, 2); + assert.equal( + f.events.filter( + (e) => e === "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ", + ).length, + 2, + ); + assert.equal(f.events.filter((e) => e === "rollback").length, 2); + assert.equal(f.events.filter((e) => e === "release").length, 2); + assert.ok(f.events.indexOf("release") < f.events.lastIndexOf("acquire")); + assert.equal( + f.events.some((e) => e.startsWith("INSERT")), + false, + ); + }); +} + +test("MysqlKvStore bounds conflict retries and rejects lock timeouts without retry", async () => { + for (const errno of [1213, 1020, 1205]) { + const problem = Object.assign(new Error(`server ${errno}`), { errno }); + const f = fakePool({ + query(sql) { + if (sql.startsWith("INSERT")) throw problem; + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + await assert.rejects(store.set(key("a"), 1), (error) => error === problem); + assert.equal(f.attempts, errno === 1205 ? 1 : 5); + assert.equal(f.events.filter((e) => e === "rollback").length, f.attempts); + assert.equal(f.events.filter((e) => e === "release").length, f.attempts); + assert.equal(f.events.includes("destroy"), false); + } +}); + +test("MysqlKvStore preserves rollback errors and discards poisoned connections", async () => { + const primary = new Error("write failure"), + rollback = new Error("rollback failure"); + const f = fakePool({ + query(sql) { + if (sql.startsWith("INSERT")) throw primary; + }, + rollback() { + throw rollback; + }, + }); + await assert.rejects( + new MysqlKvStore(f.pool, { initialized: true }).set(key("a"), 1), + (error) => { + assert.ok(error instanceof AggregateError); + assert.deepEqual(error.errors, [primary, rollback]); + return true; + }, + ); + assert.equal(f.attempts, 1); + assert.ok(f.events.includes("destroy")); + assert.equal(f.events.includes("release"), false); +}); + +test("MysqlKvStore does not retry an ambiguous commit, failed setup or mismatch rollback", async () => { + const fault = new Error("connection failure"); + for (const mode of [ + "commit", + "begin", + "isolation", + "mismatch-rollback", + ] as const) { + const f = fakePool({ + query(sql) { + if (mode === "isolation" && sql.startsWith("SET TRANSACTION")) + throw fault; + if (mode === "mismatch-rollback" && sql.includes("FOR UPDATE")) + return [{ key: '["a"]', value: "1", expires: null }]; + }, + begin() { + if (mode === "begin") throw fault; + }, + commit() { + if (mode === "commit") throw fault; + }, + rollback() { + if (mode === "mismatch-rollback") throw fault; + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + const call = + mode === "mismatch-rollback" + ? store.cas(key("a"), undefined, 2) + : store.set(key("a"), 1); + await assert.rejects(call, (error) => error === fault); + assert.equal(f.attempts, 1); + assert.ok(f.events.includes("destroy")); + assert.equal(f.events.includes("release"), false); + if (mode === "commit") assert.equal(f.events.includes("rollback"), false); + } +}); + +test("MysqlKvStore locks sorted physical keys before one cutoff and compares everything before writes", async () => { + const locks: unknown[] = []; + const writes: unknown[] = []; + const f = fakePool({ + query(sql, values) { + if (sql.includes("FOR UPDATE")) { + assert.equal(sql.includes("expires` >"), false); + locks.push((values as string[])[0]); + return [ + { + key: (values as string[])[0], + value: "1", + expires: "2026-11-12 12:00:00.123456", + }, + ]; + } + if (sql.startsWith("INSERT")) writes.push(values); + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + assert.equal( + await store.casMany([ + { key: key("b"), expectedValue: undefined, newValue: 2 }, + { key: key("a"), expectedValue: undefined, newValue: 3 }, + ]), + true, + ); + assert.deepEqual(locks, ['["a"]', '["b"]']); + assert.equal( + f.events.filter((e) => e.startsWith("SELECT DATE_FORMAT(NOW")).length, + 1, + ); + assert.ok( + f.events.findIndex((e) => e.startsWith("SELECT DATE_FORMAT(NOW")) > + f.events.findLastIndex((e) => e.includes("FOR UPDATE")), + ); + assert.equal(writes.length, 2); +}); + +test("MysqlKvStore validates derived TTLs even on mismatch and truncates microseconds", async () => { + let captured: unknown; + const f = fakePool({ + query(sql, values) { + if (sql.includes("FOR UPDATE")) + return [{ key: '["a"]', value: "1", expires: null }]; + if (sql.startsWith("SELECT DATE_FORMAT(DATE_ADD")) { + captured = values; + return [{ e0: null }]; + } + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + await assert.rejects( + store.cas(key("a"), "mismatch", 2, { + ttl: Temporal.Duration.from({ microseconds: 10, nanoseconds: 999 }), + }), + RangeError, + ); + assert.deepEqual(captured, ["2026-11-12 12:00:00.123456", 10]); + assert.equal( + f.events.some((e) => e.startsWith("INSERT")), + false, + ); + assert.ok(f.events.includes("rollback")); +}); + +test("MysqlKvStore retries initialization after failure and invalidates after failed drop", async () => { + let failInitialization = true, + failDrop = false; + const problem = new Error("metadata unavailable"); + const f = fakePool({ + metadata(sql) { + if (sql.includes("information_schema.TABLES") && failInitialization) { + failInitialization = false; + throw problem; + } + if (sql.startsWith("DROP") && failDrop) throw problem; + }, + }); + const store = new MysqlKvStore(f.pool, { + initialized: true, + expireCleanupRate: 0, + }); + await assert.rejects(store.initialize(), (error) => error === problem); + await Promise.all([store.initialize(), store.initialize()]); + assert.equal( + f.events.filter((e) => e.includes("information_schema.TABLES")).length, + 2, + ); + assert.equal( + f.events.some((e) => e.startsWith("pool:CREATE")), + false, + ); + failDrop = true; + await assert.rejects(store.drop(), (error) => error === problem); + await store.initialize(); + assert.equal( + f.events.filter((e) => e.includes("information_schema.TABLES")).length, + 3, + ); + assert.ok(f.events.some((e) => e.startsWith("pool:CREATE TABLE"))); +}); + +test("MysqlKvStore rejects unsafe gap configuration and accepts compatible JSON alias metadata", async () => { + const unsafe = fakePool({ + metadata(sql) { + if (sql.startsWith("SHOW VARIABLES")) return [{ Value: "ON" }]; + }, + }); + await assert.rejects( + new MysqlKvStore(unsafe.pool, { initialized: true }).initialize(), + /gap locking/, + ); + const valid = fakePool({ + metadata(sql) { + if (sql.includes("information_schema.COLUMNS")) + return [ + { + name: "key", + type: "varchar", + nullable: "NO", + length: 768, + charset: "utf8mb4", + collation: "utf8mb4_bin", + }, + { + name: "value", + type: "longtext", + nullable: "NO", + charset: "utf8mb4", + }, + { name: "expires", type: "datetime", nullable: "YES", precision: 6 }, + ]; + }, + }); + await new MysqlKvStore(valid.pool, { initialized: true }).initialize(); +}); + +test( + "MysqlKvStore distinct absent-key contention has bounded, all-or-nothing outcomes", + live, + async () => { + const p = await pair(); + try { + for (const count of [4, 16]) { + const pools = Array.from({ length: count }, () => + mysql.createPool(dbUrl!), + ); + const stores = pools.map( + (pool) => + new MysqlKvStore(pool, { + tableName: p.tableName, + initialized: true, + expireCleanupRate: 0, + }), + ); + const keys = stores.map((_, i) => key(`contender-${i}`)); + try { + await Promise.all(stores.map((store) => store.initialize())); + const results = await Promise.allSettled( + stores.map((store, i) => store.cas(keys[i], undefined, i)), + ); + const expected = results.map((result, i) => { + if (result.status === "fulfilled") { + assert.equal(result.value, true); + return i; + } + assert.ok( + [1213, 1020].includes(result.reason?.errno), + "only documented conflict exhaustion may reject", + ); + return undefined; + }); + if (count === 4) + assert.equal( + results.every((r) => r.status === "fulfilled"), + true, + ); + assert.ok(results.some((r) => r.status === "fulfilled")); + assert.deepEqual(await p.first.getMany(keys), expected); + await p.first.deleteMany(keys); + } finally { + await Promise.all(pools.map((pool) => pool.end())); + } + } + } finally { + await p.dispose(); + } + }, +); + +function parameters(value: unknown) { + if (value !== undefined && !Array.isArray(value)) + throw new TypeError("Expected adapter query parameters."); + return value; +} + +test( + "MysqlKvStore accepts the batch boundary and rejects oversized operations without partial writes", + live, + async () => { + const p = await pair(); + try { + const keys = Array.from({ length: 1000 }, (_, i) => key(`limit-${i}`)); + await p.first.setMany(keys.map((key, i) => ({ key, value: i }))); + assert.deepEqual( + await p.second.getMany(keys), + keys.map((_, i) => i), + ); + const over = [...keys, key("overflow")]; + await assert.rejects(p.first.getMany(over), RangeError); + await assert.rejects( + p.first.setMany(over.map((key) => ({ key, value: "changed" }))), + RangeError, + ); + await assert.rejects(p.first.deleteMany(over), RangeError); + await assert.rejects( + p.first.casMany( + over.map((key, i) => ({ + key, + expectedValue: i < 1000 ? i : undefined, + newValue: "changed", + })), + ), + RangeError, + ); + assert.deepEqual( + await p.second.getMany(keys), + keys.map((_, i) => i), + ); + assert.equal(await p.second.get(key("overflow")), undefined); + await p.first.deleteMany(keys); + assert.deepEqual( + await p.second.getMany(keys), + keys.map(() => undefined), + ); + } finally { + await p.dispose(); + } + }, +); + +test( + "MysqlKvStore metadata respects mixed-case names and case-sensitive twins", + live, + async () => { + const pool = mysql.createPool(dbUrl!); + const name = `Atomic_${crypto.randomUUID().replace(/-/g, "")}`; + const mixed = new MysqlKvStore(pool, { + tableName: name, + expireCleanupRate: 0, + }); + let twin: MysqlKvStore | undefined; + try { + await mixed.set(key("a"), 1); + const [config] = await pool.query( + "SELECT @@lower_case_table_names AS mode", + ); + if (Number(config[0].mode) === 0) { + twin = new MysqlKvStore(pool, { + tableName: name.toLowerCase(), + expireCleanupRate: 0, + }); + await twin.set(key("a"), 2); + assert.equal(await mixed.get(key("a")), 1); + assert.equal(await twin.get(key("a")), 2); + } else { + const alias = new MysqlKvStore(pool, { + tableName: name.toLowerCase(), + initialized: true, + expireCleanupRate: 0, + }); + assert.equal(await alias.get(key("a")), 1); + } + } finally { + try { + await mixed.drop(); + if (twin != null) await twin.drop(); + } finally { + await pool.end(); + } + } + }, +); + +test("MysqlKvStore captures keys, JSON and TTL before asynchronous initialization", async () => { + const ready = gate(); + let written: unknown; + let ttl: unknown; + const f = fakePool({ + query(sql, values) { + if (sql.startsWith("INSERT")) written = values; + if (sql.startsWith("SELECT DATE_FORMAT(DATE_ADD")) { + ttl = values; + return [{ e0: "2026-11-12 12:00:00.123457" }]; + } + }, + }); + const store = new MysqlKvStore( + { + async query(sql: string): Promise<[T, unknown]> { + await ready.promise; + return await f.pool.query(sql); + }, + getConnection: f.pool.getConnection, + }, + { initialized: true, expireCleanupRate: 0 }, + ); + const parts: [string] = ["original"]; + const value = { count: 1 }; + const options = { ttl: Temporal.Duration.from({ microseconds: 1 }) }; + const pending = store.set(parts, value, options); + parts[0] = "changed"; + value.count = 2; + options.ttl = Temporal.Duration.from({ hours: 1 }); + ready.resolve(); + await pending; + assert.deepEqual(ttl, ["2026-11-12 12:00:00.123456", 1]); + assert.deepEqual(written, [ + '["original"]', + '{"count":1}', + "2026-11-12 12:00:00.123457", + '{"count":1}', + "2026-11-12 12:00:00.123457", + ]); +}); + +test( + "MysqlKvStore snapshot reads exclude a paused partial batch under READ UNCOMMITTED", + live, + async () => { + const p = await pair(); + const changed = gate(), + resume = gate(); + let writer: Promise | undefined; + try { + await p.first.setMany([ + { key: key("a"), value: 0 }, + { key: key("b"), value: 0 }, + ]); + const reader = new MysqlKvStore( + { + query: p.secondPool.query.bind(p.secondPool), + async getConnection() { + const conn = await p.secondPool.getConnection(); + await conn.query( + "SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED", + ); + return conn; + }, + }, + { tableName: p.tableName, initialized: true, expireCleanupRate: 0 }, + ); + await reader.initialize(); + const paused = new MysqlKvStore( + { + query: p.firstPool.query.bind(p.firstPool), + async getConnection() { + const conn = await p.firstPool.getConnection(); + let writes = 0; + return { + async query( + sql: string, + values?: unknown, + ): Promise<[T, unknown]> { + const result = await conn.query(sql, parameters(values)); + if (sql.startsWith("INSERT") && ++writes === 1) { + changed.resolve(); + await resume.promise; + } + return result as [T, unknown]; + }, + beginTransaction: conn.beginTransaction.bind(conn), + commit: conn.commit.bind(conn), + rollback: conn.rollback.bind(conn), + release: conn.release.bind(conn), + destroy: conn.destroy.bind(conn), + }; + }, + }, + { tableName: p.tableName, initialized: true, expireCleanupRate: 0 }, + ); + writer = paused.setMany([ + { key: key("a"), value: 1 }, + { key: key("b"), value: 1 }, + ]); + await Promise.race([changed.promise, writer]); + assert.deepEqual(await reader.getMany([key("a"), key("b")]), [0, 0]); + const listed = async (prefix?: KvKey) => { + const entries = []; + for await (const entry of reader.list(prefix)) entries.push(entry); + return entries; + }; + assert.deepEqual(await listed(), [ + { key: key("a"), value: 0 }, + { key: key("b"), value: 0 }, + ]); + assert.deepEqual(await listed(key("a")), [{ key: key("a"), value: 0 }]); + resume.resolve(); + await writer; + assert.deepEqual(await reader.getMany([key("a"), key("b")]), [1, 1]); + assert.deepEqual(await listed(), [ + { key: key("a"), value: 1 }, + { key: key("b"), value: 1 }, + ]); + assert.deepEqual(await listed(key("a")), [{ key: key("a"), value: 1 }]); + } finally { + resume.resolve(); + await Promise.allSettled([writer]); + await p.dispose(); + } + }, +); diff --git a/packages/mysql/src/kv.test.ts b/packages/mysql/src/kv.test.ts index a253c4686..f122e2005 100644 --- a/packages/mysql/src/kv.test.ts +++ b/packages/mysql/src/kv.test.ts @@ -71,7 +71,7 @@ test( // Insert an already-expired row directly await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, [JSON.stringify(["expired-key"]), JSON.stringify("expired-value")], ); @@ -169,7 +169,7 @@ test("MysqlKvStore.get()", { skip: dbUrl == null }, async () => { await store.initialize(); await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`) - VALUES (?, CAST(? AS JSON))`, + VALUES (?, ?)`, [JSON.stringify(["foo", "bar"]), JSON.stringify(["foobar"])], ); assert.deepStrictEqual(await store.get(["foo", "bar"]), ["foobar"]); @@ -177,7 +177,7 @@ test("MysqlKvStore.get()", { skip: dbUrl == null }, async () => { // Insert with immediately expired TTL await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, [JSON.stringify(["foo", "bar", "ttl"]), JSON.stringify(["foobar"])], ); assert.strictEqual(await store.get(["foo", "bar", "ttl"]), undefined); @@ -194,31 +194,31 @@ test("MysqlKvStore.set()", { skip: dbUrl == null }, async () => { try { await store.set(["foo", "baz"], "baz"); const [rows] = await pool.query( - `SELECT * FROM \`${tableName}\` WHERE \`key\` = ?`, + `SELECT CAST(value AS CHAR CHARACTER SET utf8mb4) AS value, expires FROM \`${tableName}\` WHERE \`key\` = ?`, [JSON.stringify(["foo", "baz"])], ); assert.strictEqual(rows.length, 1); - assert.strictEqual(rows[0].value, "baz"); + assert.strictEqual(JSON.parse(rows[0].value), "baz"); assert.strictEqual(rows[0].expires, null); await store.set(["foo", "qux"], "qux", { ttl: Temporal.Duration.from({ days: 1 }), }); const [rows2] = await pool.query( - `SELECT * FROM \`${tableName}\` WHERE \`key\` = ?`, + `SELECT CAST(value AS CHAR CHARACTER SET utf8mb4) AS value, expires FROM \`${tableName}\` WHERE \`key\` = ?`, [JSON.stringify(["foo", "qux"])], ); assert.strictEqual(rows2.length, 1); - assert.strictEqual(rows2[0].value, "qux"); + assert.strictEqual(JSON.parse(rows2[0].value), "qux"); assert.notStrictEqual(rows2[0].expires, null); await store.set(["foo", "quux"], true); const [rows3] = await pool.query( - `SELECT * FROM \`${tableName}\` WHERE \`key\` = ?`, + `SELECT CAST(value AS CHAR CHARACTER SET utf8mb4) AS value, expires FROM \`${tableName}\` WHERE \`key\` = ?`, [JSON.stringify(["foo", "quux"])], ); assert.strictEqual(rows3.length, 1); - assert.strictEqual(rows3[0].value, true); + assert.strictEqual(JSON.parse(rows3[0].value), true); assert.strictEqual(rows3[0].expires, null); } finally { await store.drop(); @@ -226,28 +226,17 @@ test("MysqlKvStore.set()", { skip: dbUrl == null }, async () => { } }); -test( - "MysqlKvStore.set() - undefined value is a no-op", - { skip: dbUrl == null }, - async () => { - if (dbUrl == null) return; - - const { pool, store } = getStore(); - try { - // Setting undefined on a nonexistent key should leave the key absent - await store.set(["foo"], undefined); - assert.strictEqual(await store.get(["foo"]), undefined); - - // Setting a real value then overwriting with undefined should be a no-op - await store.set(["bar"], "value"); - await store.set(["bar"], undefined); - assert.deepStrictEqual(await store.get(["bar"]), "value"); - } finally { - await store.drop(); - await pool.end(); - } - }, -); +test("MysqlKvStore.set() rejects undefined before I/O", async () => { + const store = new MysqlKvStore({ + query() { + throw new Error("Unexpected I/O"); + }, + getConnection() { + throw new Error("Unexpected I/O"); + }, + } as unknown as mysql.Pool); + await assert.rejects(store.set(["foo"], undefined), TypeError); +}); test("MysqlKvStore.delete()", { skip: dbUrl == null }, async () => { if (dbUrl == null) return; @@ -348,7 +337,7 @@ test( // Insert expired entry directly await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 30 MINUTE))`, [ JSON.stringify(["list-test", "expired"]), @@ -533,7 +522,7 @@ test( // set()'s #expire() does not clean it up await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, [JSON.stringify(["expired-key"]), JSON.stringify("expired-value")], ); @@ -577,7 +566,7 @@ test( // table but should be treated as absent by cas(). await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, [JSON.stringify(["key"]), JSON.stringify("old-value")], ); @@ -627,7 +616,7 @@ test( // write—violating CAS atomicity. await pool.query( `INSERT INTO \`${tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, + VALUES (?, ?, DATE_SUB(NOW(6), INTERVAL 1 SECOND))`, [JSON.stringify(["key"]), JSON.stringify("stale")], ); diff --git a/packages/mysql/src/kv.ts b/packages/mysql/src/kv.ts index 31aabb16b..e66c049fa 100644 --- a/packages/mysql/src/kv.ts +++ b/packages/mysql/src/kv.ts @@ -1,6 +1,10 @@ +// Transactions acquire locks and execute statements sequentially. +/* oxlint-disable eslint/no-await-in-loop */ import type { KvKey, - KvStore, + AtomicKvStore, + KvStoreCasEntry, + KvStoreSetEntry, KvStoreListEntry, KvStoreSetOptions, } from "@fedify/fedify"; @@ -22,6 +26,7 @@ interface MysqlPoolConnection extends MysqlQueryable { commit(): Promise; rollback(): Promise; release(): void; + destroy(): void; } interface MysqlPool extends MysqlQueryable { @@ -43,7 +48,8 @@ export interface MysqlKvStoreOptions { readonly tableName?: string; /** - * Whether the table has been initialized. `false` by default. + * Whether the table has been created. `false` by default. With `true`, + * initialization skips DDL but still validates the InnoDB schema. * @default `false` * @since 2.1.0 */ @@ -60,7 +66,27 @@ export interface MysqlKvStoreOptions { } /** - * A key-value store that uses MySQL (or MariaDB) as the underlying storage. + * An atomic key-value store backed by an InnoDB table on MySQL or MariaDB. + * + * All keys in one primary database/table share an atomic scope. Batches have + * at most 1,000 entries and serialized keys at most 768 Unicode code points; + * server packet and resource limits also apply. Values use JSON serialization + * and decoded structural equality (`isEqual`); object member order is ignored. + * Do not mutate expected values while a CAS is pending. + * + * TTLs are truncated to safe integer microseconds (approximately ±285 years), + * evaluated using one server cutoff per attempt. Calendar units use the local + * current date for conversion. Omitted TTL clears the previous expiry. + * + * Use a normal mysql2 pool with autocommit enabled, object rows, default type + * casting, and the same fixed-offset or UTC session time zone on every client. + * Caller-owned transactions, replica routing, concurrent schema changes, + * `NO_BACKSLASH_ESCAPES`, and disabled InnoDB gap locking are unsupported. + * Initialization checks the schema even with `initialized: true`. + * + * Deadlocks and MariaDB snapshot conflicts retry up to five total attempts. + * Other failures reject; a failed commit response can leave its outcome + * unknown. Atomicity does not promise crash durability or external effects. * * @example * ```ts @@ -78,11 +104,12 @@ export interface MysqlKvStoreOptions { * * @since 2.1.0 */ -export class MysqlKvStore implements KvStore { +export class MysqlKvStore implements AtomicKvStore { readonly #pool: MysqlPool; readonly #tableName: string; readonly #expireCleanupRate: number; #initialized: boolean; + #initialization: Promise | undefined; /** * Creates a new MySQL key-value store. @@ -131,82 +158,49 @@ export class MysqlKvStore implements KvStore { ) { return; } - await this.#pool.query( - `DELETE FROM \`${this.#tableName}\` - WHERE \`expires\` IS NOT NULL AND \`expires\` < NOW(6)`, - ); + try { + await this.#pool.query( + `DELETE FROM \`${this.#tableName}\` + WHERE \`expires\` IS NOT NULL AND \`expires\` < NOW(6)`, + ); + } catch (error) { + // Housekeeping must not turn an already committed mutation into failure. + logger.warn("Failed to clean up expired key-value entries: {error}", { + error, + }); + } } /** - * {@inheritDoc KvStore.get} + * {@inheritDoc AtomicKvStore.get} * @since 2.1.0 */ async get(key: KvKey): Promise { - await this.initialize(); - const serializedKey = JSON.stringify([...key]); - const [rows] = await this.#pool.query( - `SELECT \`value\` FROM \`${this.#tableName}\` - WHERE \`key\` = ? - AND (\`expires\` IS NULL OR \`expires\` > NOW(6))`, - [serializedKey], - ); - if (rows.length < 1) return undefined; - return rows[0].value as T; + return (await this.getMany([key]))[0]; } /** - * {@inheritDoc KvStore.set} + * {@inheritDoc AtomicKvStore.set} * @since 2.1.0 */ async set( key: KvKey, value: unknown, - options?: KvStoreSetOptions | undefined, + options?: KvStoreSetOptions, ): Promise { - if (value === undefined) return; - await this.initialize(); - const serializedKey = JSON.stringify([...key]); - const jsonValue = JSON.stringify(value); - if (options?.ttl != null) { - const ttlSeconds = durationToSeconds(options.ttl); - await this.#pool.query( - `INSERT INTO \`${this.#tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), - DATE_ADD(NOW(6), INTERVAL ? SECOND)) - ON DUPLICATE KEY UPDATE - \`value\` = VALUES(\`value\`), - \`expires\` = VALUES(\`expires\`)`, - [serializedKey, jsonValue, ttlSeconds], - ); - } else { - await this.#pool.query( - `INSERT INTO \`${this.#tableName}\` (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), NULL) - ON DUPLICATE KEY UPDATE - \`value\` = VALUES(\`value\`), - \`expires\` = NULL`, - [serializedKey, jsonValue], - ); - } - await this.#expire(); + await this.setMany([{ key, value, options }]); } /** - * {@inheritDoc KvStore.delete} + * {@inheritDoc AtomicKvStore.delete} * @since 2.1.0 */ async delete(key: KvKey): Promise { - await this.initialize(); - const serializedKey = JSON.stringify([...key]); - await this.#pool.query( - `DELETE FROM \`${this.#tableName}\` WHERE \`key\` = ?`, - [serializedKey], - ); - await this.#expire(); + await this.deleteMany([key]); } /** - * {@inheritDoc KvStore.cas} + * {@inheritDoc AtomicKvStore.cas} * @since 2.1.0 */ async cas( @@ -215,158 +209,371 @@ export class MysqlKvStore implements KvStore { newValue: unknown, options?: KvStoreSetOptions, ): Promise { + return await this.casMany([{ key, expectedValue, newValue, options }]); + } + + /** + * {@inheritDoc AtomicKvStore.getMany} + * @since 2.5.0 + */ + async getMany( + keys: readonly KvKey[], + ): Promise<(T | undefined)[]> { + validateBatch(keys); + const serialized = Array.from(keys, serializeKey); + if (serialized.length === 0) return []; await this.initialize(); - const serializedKey = JSON.stringify([...key]); - let conn: MysqlPoolConnection | undefined; - try { - conn = await this.#pool.getConnection(); - await conn.beginTransaction(); - - // NOTE: The expiry predicate is intentionally omitted from the WHERE - // clause here so that `FOR UPDATE` always locks the physical row (if - // it exists), even when that row has an expired `expires` value. - // Including `AND (expires IS NULL OR expires > NOW(6))` in the WHERE - // would cause MySQL to skip (and therefore not lock) physically-present - // but expired rows, allowing two concurrent `cas(key, undefined, ...)` - // calls to both pass the expected-value check and both insert—a CAS - // atomicity violation. - // - // NOTE: Under InnoDB REPEATABLE READ (the default isolation level), a - // `FOR UPDATE` on a missing row acquires a gap lock preventing concurrent - // inserts. Under READ COMMITTED, that gap lock is not applied, so - // concurrent CAS operations on a logically-absent key may race. - const [rows] = await conn.query< - (RowDataPacket & { is_expired: 0 | 1 })[] - >( - `SELECT - \`value\`, - (\`expires\` IS NOT NULL AND \`expires\` <= NOW(6)) AS \`is_expired\` - FROM \`${this.#tableName}\` - WHERE \`key\` = ? - FOR UPDATE`, - [serializedKey], + return await this.#transaction(async (conn) => { + const [rows] = await conn.query( + `SELECT \`key\`, CAST(\`value\` AS CHAR CHARACTER SET utf8mb4) AS \`value\` + FROM \`${this.#tableName}\` WHERE \`key\` IN (?) + AND (\`expires\` IS NULL OR \`expires\` > NOW(6))`, + [serialized], + ); + const values = new Map( + rows.map((row) => [row.key, JSON.parse(row.value) as T]), ); + return serialized.map((key) => values.get(key)); + }); + } + + /** + * {@inheritDoc AtomicKvStore.setMany} + * @since 2.5.0 + */ + async setMany(entries: readonly KvStoreSetEntry[]): Promise { + validateBatch(entries); + const captured = Array.from(entries, (entry): EncodedEntry => ({ + key: serializeKey(entry.key), + value: serializeValue(entry.value), + ttl: encodeTtl(entry.options), + })); + validateDistinct(captured); + if (captured.length === 0) return; + await this.#mutate(captured, false); + } - const row = rows[0]; - const currentValue = - !row || row.is_expired ? undefined : (row as RowDataPacket).value; + /** + * {@inheritDoc AtomicKvStore.deleteMany} + * @since 2.5.0 + */ + async deleteMany(keys: readonly KvKey[]): Promise { + validateBatch(keys); + const captured = Array.from(keys, (key): EncodedEntry => ({ + key: serializeKey(key), + value: undefined, + ttl: null, + })); + validateDistinct(captured); + if (captured.length === 0) return; + await this.#mutate(captured, false); + } - if (!isEqual(currentValue, expectedValue)) { - await conn.rollback(); - return false; + /** + * {@inheritDoc AtomicKvStore.casMany} + * @since 2.5.0 + */ + async casMany(entries: readonly KvStoreCasEntry[]): Promise { + validateBatch(entries); + const captured = Array.from(entries, (entry): EncodedEntry => { + const value = entry.newValue; + const options = entry.options; + if (value === undefined && options?.ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); } + return { + key: serializeKey(entry.key), + expectedValue: entry.expectedValue, + value: value === undefined ? undefined : serializeValue(value), + ttl: encodeTtl(options), + }; + }); + validateDistinct(captured); + if (captured.length === 0) return true; + return await this.#mutate(captured, true); + } - if (newValue === undefined) { - await conn.query( - `DELETE FROM \`${this.#tableName}\` WHERE \`key\` = ?`, - [serializedKey], + async #mutate( + entries: EncodedEntry[], + conditional: boolean, + ): Promise { + entries.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)); + await this.initialize(); + const matched = await this.#transaction(async (conn) => { + // Do not add consistent table reads before these locks: MariaDB snapshot + // isolation can reject newer records with ER_CHECKREAD (1020). + const rows = new Map(); + if (conditional) { + for (const entry of entries) { + // Lock physical rows AND absent ranges, including expired rows. + const [found] = await conn.query( + `SELECT \`key\`, CAST(\`value\` AS CHAR CHARACTER SET utf8mb4) AS \`value\`, + DATE_FORMAT(\`expires\`, '%Y-%m-%d %H:%i:%s.%f') AS \`expires\` + FROM \`${this.#tableName}\` WHERE \`key\` = ? FOR UPDATE`, + [entry.key], + ); + if (found.length > 0) rows.set(entry.key, found[0]); + } + } + const [clock] = await conn.query<(RowDataPacket & { cutoff: string })[]>( + "SELECT DATE_FORMAT(NOW(6), '%Y-%m-%d %H:%i:%s.%f') AS cutoff", + ); + const cutoff = clock[0].cutoff; + const durations = [ + ...new Set(entries.flatMap((e) => (e.ttl == null ? [] : [e.ttl]))), + ]; + const expiries = new Map(); + if (durations.length > 0) { + const expressions = durations.map( + (_, i) => + `DATE_FORMAT(DATE_ADD(?, INTERVAL ? MICROSECOND), '%Y-%m-%d %H:%i:%s.%f') AS e${i}`, ); - } else { - const jsonValue = JSON.stringify(newValue); - if (options?.ttl != null) { - const ttlSeconds = durationToSeconds(options.ttl); + const [computed] = await conn.query( + `SELECT ${expressions.join(", ")}`, + durations.flatMap((ttl) => [cutoff, ttl]), + ); + for (let i = 0; i < durations.length; i++) { + const expiry = computed[0][`e${i}`] as string | null; + if (expiry == null) + throw new RangeError("TTL exceeds the server DATETIME range."); + expiries.set(durations[i], expiry); + } + } + if (conditional) { + for (const entry of entries) { + const row = rows.get(entry.key); + const value = + row == null || (row.expires != null && row.expires <= cutoff) + ? undefined + : JSON.parse(row.value); + if (!isEqual(value, entry.expectedValue)) return false; + } + } + for (const entry of entries) { + if (entry.value === undefined) { await conn.query( - `INSERT INTO \`${this.#tableName}\` - (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), - DATE_ADD(NOW(6), INTERVAL ? SECOND)) - ON DUPLICATE KEY UPDATE - \`value\` = VALUES(\`value\`), - \`expires\` = VALUES(\`expires\`)`, - [serializedKey, jsonValue, ttlSeconds], + `DELETE FROM \`${this.#tableName}\` WHERE \`key\` = ?`, + [entry.key], ); } else { + const expiry = entry.ttl == null ? null : expiries.get(entry.ttl)!; await conn.query( - `INSERT INTO \`${this.#tableName}\` - (\`key\`, \`value\`, \`expires\`) - VALUES (?, CAST(? AS JSON), NULL) - ON DUPLICATE KEY UPDATE - \`value\` = VALUES(\`value\`), - \`expires\` = NULL`, - [serializedKey, jsonValue], + `INSERT INTO \`${this.#tableName}\` (\`key\`, \`value\`, \`expires\`) + VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE \`value\` = ?, \`expires\` = ?`, + [entry.key, entry.value, expiry, entry.value, expiry], ); } } - - await conn.commit(); - await this.#expire(); return true; - } catch (e) { - if (conn) await conn.rollback(); - throw e; - } finally { - if (conn) conn.release(); + }); + // #transaction has released its connection, including with a one-slot pool. + if (matched) await this.#expire(); + return matched; + } + + async #transaction( + run: (conn: MysqlPoolConnection) => Promise, + ): Promise { + for (let attempt = 0; ; attempt++) { + const conn = await this.#pool.getConnection(); + let phase: "setup" | "active" | "rollback" | "commit" = "setup"; + let reusable = false; + try { + await conn.query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"); + await conn.beginTransaction(); + phase = "active"; + const result = await run(conn); + if (result === false) { + phase = "rollback"; + await conn.rollback(); + } else { + phase = "commit"; + await conn.commit(); + } + reusable = true; + return result; + } catch (error) { + if (phase !== "active") throw error; + try { + await conn.rollback(); + reusable = true; + } catch (rollbackError) { + throw new AggregateError( + [error, rollbackError], + "MySQL transaction and rollback failed.", + ); + } + if (attempt >= 4 || !retryable(error)) throw error; + } finally { + if (reusable) conn.release(); + else conn.destroy(); + } + // Release before backoff/reacquisition, even when the pool has one slot. + await new Promise((resolve) => + setTimeout(resolve, 5 + Math.random() * Math.min(45, 5 * 2 ** attempt)), + ); } } /** - * {@inheritDoc KvStore.list} + * {@inheritDoc AtomicKvStore.list} * @since 2.1.0 */ async *list(prefix?: KvKey): AsyncIterable { await this.initialize(); - let rows: RowDataPacket[]; - if (prefix == null || prefix.length === 0) { - [rows] = await this.#pool.query( - `SELECT \`key\`, \`value\` FROM \`${this.#tableName}\` - WHERE \`expires\` IS NULL OR \`expires\` > NOW(6) - ORDER BY \`key\``, - ); - } else { - const serializedPrefix = JSON.stringify([...prefix]); - // Escape LIKE special characters in the prefix - const likePrefix = - serializedPrefix.slice(0, -1).replace(/[%_\\]/g, "\\$&") + ",%"; - [rows] = await this.#pool.query( - `SELECT \`key\`, \`value\` FROM \`${this.#tableName}\` - WHERE (\`key\` = ? OR \`key\` LIKE ? ESCAPE '\\\\') - AND (\`expires\` IS NULL OR \`expires\` > NOW(6)) - ORDER BY \`key\``, - [serializedPrefix, likePrefix], - ); - } + const rows = await this.#transaction(async (conn) => { + let rows: RowDataPacket[]; + if (prefix == null || prefix.length === 0) { + [rows] = await conn.query( + `SELECT \`key\`, CAST(\`value\` AS CHAR CHARACTER SET utf8mb4) AS \`value\` FROM \`${this.#tableName}\` + WHERE \`expires\` IS NULL OR \`expires\` > NOW(6) + ORDER BY \`key\``, + ); + } else { + const serializedPrefix = JSON.stringify([...prefix]); + // Escape LIKE special characters in the prefix + const likePrefix = + serializedPrefix.slice(0, -1).replace(/[%_\\]/g, "\\$&") + ",%"; + [rows] = await conn.query( + `SELECT \`key\`, CAST(\`value\` AS CHAR CHARACTER SET utf8mb4) AS \`value\` FROM \`${this.#tableName}\` + WHERE (\`key\` = ? OR \`key\` LIKE ? ESCAPE '\\\\') + AND (\`expires\` IS NULL OR \`expires\` > NOW(6)) + ORDER BY \`key\``, + [serializedPrefix, likePrefix], + ); + } + return rows; + }); for (const row of rows) { yield { key: JSON.parse(row.key) as KvKey, - value: row.value, + value: JSON.parse(row.value), }; } } /** * Creates the table used by the key-value store if it does not already exist. - * Does nothing if the table already exists. + * Validates the existing schema, including when `initialized: true` was + * supplied. Successful initialization is cached until {@link MysqlKvStore.drop}. + * Initialization and dropping must not overlap store operations. * * @since 2.1.0 */ async initialize(): Promise { - if (this.#initialized) return; - logger.debug("Initializing the key-value store table {tableName}...", { - tableName: this.#tableName, - }); - await this.#pool.query( - `CREATE TABLE IF NOT EXISTS \`${this.#tableName}\` ( - \`key\` VARCHAR(768) NOT NULL, - \`value\` JSON NOT NULL, - \`expires\` DATETIME(6) NULL DEFAULT NULL, - PRIMARY KEY (\`key\`) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`, - ); - try { + if (this.#initialization == null) { + const pending = this.#initialize(); + this.#initialization = pending; + // Clear only this attempt; a failed older attempt must not erase a newer one. + pending.catch(() => { + if (this.#initialization === pending) this.#initialization = undefined; + }); + } + await this.#initialization; + } + + async #initialize(): Promise { + if (!this.#initialized) { await this.#pool.query( - `CREATE INDEX \`idx_${this.#tableName}_expires\` - ON \`${this.#tableName}\` (\`expires\`)`, + `CREATE TABLE IF NOT EXISTS \`${this.#tableName}\` ( + \`key\` VARCHAR(768) NOT NULL, + \`value\` JSON NOT NULL, + \`expires\` DATETIME(6) NULL DEFAULT NULL, + PRIMARY KEY (\`key\`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`, ); - } catch (e) { - // Ignore if the index already exists (ER_DUP_KEYNAME) - if ((e as { code?: string }).code !== "ER_DUP_KEYNAME") throw e; } - this.#initialized = true; - logger.debug("Initialized the key-value store table {tableName}.", { - tableName: this.#tableName, - }); + // Metadata collations can ignore case even on case-sensitive file systems. + const predicate = `TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? + AND (@@lower_case_table_names <> 0 OR + (CAST(TABLE_SCHEMA AS BINARY) = CAST(DATABASE() AS BINARY) + AND CAST(TABLE_NAME AS BINARY) = CAST(? AS BINARY)))`; + const params = [this.#tableName, this.#tableName]; + const [tables] = await this.#pool.query( + `SELECT ENGINE AS \`engine\`, TABLE_TYPE AS \`table_type\` + FROM information_schema.TABLES WHERE ${predicate}`, + params, + ); + if ( + tables.length !== 1 || + tables[0].engine?.toLowerCase() !== "innodb" || + tables[0].table_type !== "BASE TABLE" + ) { + throw new TypeError( + "MysqlKvStore requires an InnoDB base table in the pool's default database.", + ); + } + const [columns] = await this.#pool.query( + `SELECT COLUMN_NAME AS \`name\`, DATA_TYPE AS \`type\`, IS_NULLABLE AS \`nullable\`, + CHARACTER_MAXIMUM_LENGTH AS \`length\`, CHARACTER_SET_NAME AS \`charset\`, + COLLATION_NAME AS \`collation\`, DATETIME_PRECISION AS \`precision\` + FROM information_schema.COLUMNS WHERE ${predicate}`, + params, + ); + const key = columns.find((c) => c.name === "key"); + const value = columns.find((c) => c.name === "value"); + const expires = columns.find((c) => c.name === "expires"); + if ( + key?.type !== "varchar" || + Number(key.length) !== 768 || + key.charset !== "utf8mb4" || + key.collation !== "utf8mb4_bin" || + key.nullable !== "NO" || + value?.nullable !== "NO" || + !( + value.type === "json" || + (value.type === "longtext" && value.charset === "utf8mb4") + ) || + expires?.type !== "datetime" || + Number(expires.precision) !== 6 || + expires.nullable !== "YES" + ) { + throw new TypeError( + "Incompatible MysqlKvStore key, JSON value, or nullable DATETIME(6) expiry columns.", + ); + } + const [indexes] = await this.#pool.query( + `SELECT INDEX_NAME AS \`name\`, COLUMN_NAME AS \`column_name\`, + SEQ_IN_INDEX AS \`position\`, SUB_PART AS \`prefix\`, INDEX_TYPE AS \`type\` + FROM information_schema.STATISTICS WHERE ${predicate} AND NON_UNIQUE = 0`, + params, + ); + if ( + indexes.length !== 1 || + indexes[0].name !== "PRIMARY" || + indexes[0].column_name !== "key" || + Number(indexes[0].position) !== 1 || + indexes[0].prefix != null || + indexes[0].type !== "BTREE" + ) { + throw new TypeError( + "MysqlKvStore requires a full key BTREE primary index and no other unique indexes.", + ); + } + const [unsafe] = await this.#pool.query( + "SHOW VARIABLES WHERE Variable_name = 'innodb_locks_unsafe_for_binlog'", + ); + if ( + unsafe.some( + (row) => !["off", "0"].includes(String(row.Value).toLowerCase()), + ) + ) { + throw new TypeError( + "MysqlKvStore requires InnoDB gap locking to be enabled.", + ); + } + if (!this.#initialized) { + try { + await this.#pool.query( + `CREATE INDEX \`idx_${this.#tableName}_expires\` + ON \`${this.#tableName}\` (\`expires\`)`, + ); + } catch (error) { + if ((error as { code?: string }).code !== "ER_DUP_KEYNAME") throw error; + } + } } /** @@ -377,14 +584,88 @@ export class MysqlKvStore implements KvStore { * @since 2.1.0 */ async drop(): Promise { - await this.#pool.query(`DROP TABLE IF EXISTS \`${this.#tableName}\``); - this.#initialized = false; + try { + await this.#pool.query(`DROP TABLE IF EXISTS \`${this.#tableName}\``); + } finally { + this.#initialization = undefined; + this.#initialized = false; + } + } +} + +interface KvRow extends RowDataPacket { + key: string; + value: string; + expires?: string | null; +} + +interface EncodedEntry { + readonly key: string; + readonly value: string | undefined; + readonly ttl: number | null; + readonly expectedValue?: unknown; +} + +function validateBatch(entries: readonly unknown[]): void { + if (!Array.isArray(entries)) + throw new TypeError("An atomic batch must be an array."); + if (entries.length > 1000) + throw new RangeError("An atomic batch has at most 1,000 entries."); + for (let i = 0; i < entries.length; i++) { + if (!Object.hasOwn(entries, i)) + throw new TypeError("Atomic batches must not be sparse."); + } +} + +function serializeKey(key: KvKey): string { + if (!Array.isArray(key) || key.length === 0) + throw new TypeError("A key must be a nonempty array of strings."); + for (let i = 0; i < key.length; i++) { + if (!Object.hasOwn(key, i) || typeof key[i] !== "string") + throw new TypeError("Key parts must be strings."); + } + const encoded = JSON.stringify([...key]); + if ([...encoded].length > 768) + throw new RangeError( + "A serialized key has at most 768 Unicode code points.", + ); + return encoded; +} + +function serializeValue(value: unknown): string { + const encoded = JSON.stringify(value); + if (encoded === undefined) + throw new TypeError( + "The value must have a JSON representation; undefined cannot be stored.", + ); + return encoded; +} + +function encodeTtl(options?: KvStoreSetOptions): number | null { + if (options?.ttl == null) return null; + const micros = Math.trunc( + options.ttl.total({ + unit: "microsecond", + relativeTo: Temporal.Now.plainDateTimeISO(), + }), + ); + if (!Number.isSafeInteger(micros)) + throw new RangeError("TTL must fit in safe integer microseconds."); + return micros; +} + +function validateDistinct(entries: readonly EncodedEntry[]): void { + if (new Set(entries.map((entry) => entry.key)).size !== entries.length) { + throw new TypeError("Duplicate keys in an atomic mutation."); } } -function durationToSeconds(duration: Temporal.Duration): number { - return duration.total({ - unit: "second", - relativeTo: Temporal.Now.plainDateTimeISO(), - }); +function retryable(error: unknown): boolean { + const e = error as { errno?: number; code?: string } | null; + return ( + e?.errno === 1213 || + e?.errno === 1020 || + e?.code === "ER_LOCK_DEADLOCK" || + e?.code === "ER_CHECKREAD" + ); } diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index fc985ad77..5d0f96cbe 100644 --- a/packages/postgres/src/kv.index.test.ts +++ b/packages/postgres/src/kv.index.test.ts @@ -138,6 +138,41 @@ test( }, ); +for (const difference of ["timezone", "offset"] as const) { + test( + `PostgresKvStore does not reuse an expiry index with a different ${difference}`, + { skip }, + async () => { + if (!dbUrl) return; + const { sql, tableName, statements } = database(); + const store = new PostgresKvStore(sql, { tableName }); + try { + await legacyTable(sql, tableName); + const expression = + difference === "timezone" + ? sql`timezone('Etc/GMT+1', timezone('UTC', created) + ttl)` + : sql`timezone('UTC', timezone('UTC', created) + ttl + interval '1 second')`; + await sql`CREATE INDEX ${sql(`near_match_${tableName}`)} ON ${sql(tableName)} + ((${expression})) WHERE ttl IS NOT NULL`; + assert.equal((await indexes(sql, tableName)).length, 1); + statements.length = 0; + await store.initialize(); + assert.equal((await indexes(sql, tableName)).length, 2); + assert(statements.some((query) => /CREATE INDEX/.test(query))); + statements.length = 0; + await new PostgresKvStore(sql, { tableName }).initialize(); + assert.equal((await indexes(sql, tableName)).length, 2); + assert( + !statements.some((query) => /LOCK TABLE|CREATE INDEX/.test(query)), + ); + } finally { + await store.drop(); + await sql.end(); + } + }, + ); +} + test( "PostgresKvStore leaves populated legacy tables unlocked until manual migration", { skip }, diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 0e1e1d553..f170877f5 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -409,7 +409,9 @@ export class PostgresKvStore implements KvStore { AND NOT i.indisunique AND i.indnatts = 1 AND a.amname = 'btree' AND i.indkey[0] = 0 AND o.opcdefault AND o.opcintype = 'timestamptz'::regtype - AND pg_get_indexdef(i.indexrelid, 1, false) IN ( + -- Index-column DDL adds version-dependent outer parentheses + -- for AT TIME ZONE. Compare the catalog expression itself. + AND pg_get_expr(i.indexprs, i.indrelid, false) IN ( ${"timezone('UTC'::text, (timezone('UTC'::text, created) + ttl))"}, ${"(((created AT TIME ZONE 'UTC'::text) + ttl) AT TIME ZONE 'UTC'::text)"} )