From 49b1539bbee7e37cd4d93a2ec9427296ee04f790 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:27:07 +0900 Subject: [PATCH 1/7] Add atomic batches to the MySQL key-value store Implement AtomicKvStore with snapshot reads and all-or-nothing mutations on one InnoDB connection. Lock physical rows and absent ranges before comparing, use one server expiry cutoff, and retry bounded transaction conflicts with fresh comparisons. Preserve ambiguous commit outcomes and discard connections whose transaction state is uncertain. Validate the table schema even when initialization DDL is skipped, and use one JSON codec for MySQL and MariaDB. Reject undefined writes and invalid batches before I/O. Document the pool, schema, timezone and TTL requirements and cover concurrency, rollback, codecs and connection failures on both database engines. Closes https://github.com/fedify-dev/fedify/issues/1368 Related to https://github.com/fedify-dev/fedify/issues/1361 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-fable-5-2 Assisted-by: Antigravity:gemini-3.8-flash Assisted-by: OpenCode:MiniMax-M3.1-Flash Assisted-by: Claude Code:claude-opus-5-5 Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-sonnet-5-5 --- .../mysql/atomic-key-value-operations.md | 19 + claude-plugin/skills/fedify/SKILL.md | 19 +- docs/manual/kv.md | 33 +- packages/mysql/README.md | 98 ++ packages/mysql/src/atomic-kv.test.ts | 1122 +++++++++++++++++ packages/mysql/src/kv.test.ts | 59 +- packages/mysql/src/kv.ts | 629 ++++++--- 7 files changed, 1761 insertions(+), 218 deletions(-) create mode 100644 changes.d/mysql/atomic-key-value-operations.md create mode 100644 packages/mysql/src/atomic-kv.test.ts 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..db3f9694c 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,7 +495,7 @@ 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 +`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 @@ -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..7371c434c 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -563,6 +563,31 @@ 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 +[package README] +for schema, codec, clock, cleanup and resource restrictions. + [`MysqlKvStore`]: https://jsr.io/@fedify/mysql/doc/kv/~/MysqlKvStore ### `DenoKvStore` (Deno only) @@ -949,9 +974,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,7 +1030,7 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full +`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: @@ -1020,7 +1042,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 +1051,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..d75836400 100644 --- a/packages/mysql/README.md +++ b/packages/mysql/README.md @@ -44,3 +44,101 @@ 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. + +### 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..f776e55d6 --- /dev/null +++ b/packages/mysql/src/atomic-kv.test.ts @@ -0,0 +1,1122 @@ +// 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", + }); + } +}); + +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 locked.promise; + 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 changed.promise; + 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" + ); } From 916774245bb88b481e9873d6d81a83c045132d2a Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:50:04 +0900 Subject: [PATCH 2/7] Release MySQL test barriers when operations fail Race each synchronization gate against its background CAS or writer. If the operation rejects before signaling the gate, propagate its error into the existing finally block so pools close and the table is dropped. https://github.com/fedify-dev/fedify/pull/1433#discussion_r4555500928 Assisted-by: Codex:gpt-6.1-sol --- packages/mysql/src/atomic-kv.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/mysql/src/atomic-kv.test.ts b/packages/mysql/src/atomic-kv.test.ts index f776e55d6..6b1a9dab6 100644 --- a/packages/mysql/src/atomic-kv.test.ts +++ b/packages/mysql/src/atomic-kv.test.ts @@ -289,7 +289,7 @@ test( { key: key("a"), expectedValue: undefined, newValue: undefined }, { key: key("b"), expectedValue: 1, newValue: 2 }, ]); - await locked.promise; + await Promise.race([locked.promise, operation]); let settled = false; writer = p.second.set(key("a"), "writer").finally(() => { settled = true; @@ -1093,7 +1093,7 @@ test( { key: key("a"), value: 1 }, { key: key("b"), value: 1 }, ]); - await changed.promise; + await Promise.race([changed.promise, writer]); assert.deepEqual(await reader.getMany([key("a"), key("b")]), [0, 0]); const listed = async (prefix?: KvKey) => { const entries = []; From 55626358682bd5aff509eb5d9fe5194d34d2fe68 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:57:54 +0900 Subject: [PATCH 3/7] Allow database latency in MySQL expiry tests The shared conformance helper defaults to a 500 ms TTL. Under the parallel Bun CI workload, a value can expire while later keys are seeded and updated, before the test attempts to clear its TTL with CAS. Use the helper's existing MySQL-specific timing options with a five second TTL and six second wait. Keep every expiry, replacement and failed-CAS preservation assertion without changing production behavior or the shared helper. https://github.com/fedify-dev/fedify/actions/runs/43291681876/job/131439243660?pr=1433 Assisted-by: Codex:gpt-6.1-sol --- packages/mysql/src/atomic-kv.test.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/mysql/src/atomic-kv.test.ts b/packages/mysql/src/atomic-kv.test.ts index 6b1a9dab6..47686e90b 100644 --- a/packages/mysql/src/atomic-kv.test.ts +++ b/packages/mysql/src/atomic-kv.test.ts @@ -64,6 +64,11 @@ 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 }), + }, }); } }); From 72953747a832e3763c1779080e3f09c19a77fe74 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 09:32:00 +0900 Subject: [PATCH 4/7] Recognize expiry indexes on PostgreSQL 18.7 Compare the catalog expression rather than index-column DDL. PostgreSQL 18.7 adds outer parentheses around AT TIME ZONE index definitions, so the old exact DDL comparison missed compatible indexes and could provision a duplicate index. Keep the expression, predicate and index-shape checks. Retain the original one-index and no-DDL assertions and cover rejection of near-match timezone and one-second-offset expressions. https://github.com/fedify-dev/fedify/pull/1429 https://github.com/fedify-dev/fedify/actions/runs/43333332595 https://www.postgresql.org/docs/release/18.7/ Changelog: none Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-opus-5-5 (cherry picked from commit 508ab54422871543f7c28e2a8082ecc65fdcdd4e) --- packages/postgres/src/kv.index.test.ts | 35 ++++++++++++++++++++++++++ packages/postgres/src/kv.ts | 4 ++- 2 files changed, 38 insertions(+), 1 deletion(-) 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)"} ) From 113ae2731e4d531ba3b8c6132df1b2e6efc6c99c Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:29:57 +0900 Subject: [PATCH 5/7] Preserve adapter README links after rebasing Keep the merged SQLite guidance and MySQL atomic-store documentation linked to their own package READMEs. The two branches used the same reference label for different destinations, so merging the text alone would send MySQL readers to SQLite documentation. https://github.com/fedify-dev/fedify/pull/1430 https://github.com/fedify-dev/fedify/pull/1433 Assisted-by: Codex:gpt-6.1-sol --- docs/manual/kv.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 7371c434c..e55a46c7d 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`] @@ -585,10 +585,11 @@ 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 -[package README] +[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) From f255a5b2b773fdf17db1604a176d806c18ea116c Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:31:54 +0900 Subject: [PATCH 6/7] Test MariaDB key-value stores in CI Run the ordinary and atomic KV suites against MariaDB 11.8 on Node.js, Bun, and Deno. Use a healthy InnoDB service and the existing source test commands rather than treating MySQL coverage as MariaDB verification. Keep this matrix scoped to KV and document the existing message-queue compatibility failures that prevent using the whole MySQL package suite as a MariaDB gate. https://github.com/fedify-dev/fedify/pull/1433#discussion_r4558929211 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-sonnet-5-5 --- .github/workflows/main.yaml | 39 +++++++++++++++++++++++++++++++++++++ packages/mysql/README.md | 7 +++++++ 2 files changed, 46 insertions(+) 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/packages/mysql/README.md b/packages/mysql/README.md index d75836400..a2a0587da 100644 --- a/packages/mysql/README.md +++ b/packages/mysql/README.md @@ -127,6 +127,13 @@ 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 From de653b8b8cb5ef32f41383da98b4a12d1fbcbae6 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:36:54 +0900 Subject: [PATCH 7/7] Reconcile documentation after rebasing Preserve the generated MySQL changelog alongside upstream fragments and combine the PGlite and MySQL atomic support guidance after rebasing. Changelog: none Assisted-by: Codex:gpt-6.1-sol --- CHANGES.md | 19 +++++++++++++++++++ claude-plugin/skills/fedify/SKILL.md | 22 +++++++++++----------- docs/manual/kv.md | 8 ++++---- 3 files changed, 34 insertions(+), 15 deletions(-) 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/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index db3f9694c..a7c7f6a04 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -495,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`, `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. +`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 diff --git a/docs/manual/kv.md b/docs/manual/kv.md index e55a46c7d..af77c542d 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -1031,10 +1031,10 @@ check or an operation receipt before repeating effects. ### Current support -`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: +`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 | | --------------------- | --------------- | ------------------------------------------------------------------------------------------------- |