diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..dadcd2745 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -167,8 +167,23 @@ To be released. its table with a D1 migration and pass `initialized: true` in production to avoid an extra query per request. + - Added atomic multi-key snapshots, writes, deletions and conditional + updates to `WorkersD1KvStore`. Applications can now reserve a record + and update a shared budget together, with every comparison checked + before any change commits. Single-key and multi-key CAS share JSON + comparison and millisecond TTL rules; oversized calls reject instead + of splitting. [[#1361], [#1379], [#1432]] + + Production D1 migrations must create both the entry table and its + `_atomic` guard table, plus the expiry index, before using + `initialized: true`. Pass the D1 binding for fresh primary snapshots; + caller-owned sessions can read stale replicas. `WorkersKvStore` + remains without CAS or atomic multi-key support. + [#1135]: https://github.com/fedify-dev/fedify/issues/1135 [#1360]: https://github.com/fedify-dev/fedify/pull/1360 +[#1379]: https://github.com/fedify-dev/fedify/issues/1379 +[#1432]: https://github.com/fedify-dev/fedify/pull/1432 ### @fedify/cli @@ -429,6 +444,11 @@ To be released. the index exists, writes whose expiration overflows PostgreSQL's timestamp range fail atomically, including writes from older versions. [[#1309], [#1334]] + - Fixed `PostgresKvStore` creating a redundant expiry index on PostgreSQL + 18.7 when the existing index uses the documented `AT TIME ZONE` + spelling. Index detection now compares the expression itself rather + than the version-dependent index-column wrapper. + [[#1361], [#1379], [#1432]] [index migration guide]: https://fedify.dev/manual/kv#postgresql-expiry-index [#1309]: https://github.com/fedify-dev/fedify/issues/1309 diff --git a/changes.d/cfworkers/atomic-kv-store.md b/changes.d/cfworkers/atomic-kv-store.md new file mode 100644 index 000000000..e54ec7569 --- /dev/null +++ b/changes.d/cfworkers/atomic-kv-store.md @@ -0,0 +1,18 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1379': https://github.com/fedify-dev/fedify/issues/1379 + '#1432': https://github.com/fedify-dev/fedify/pull/1432 +--- + - Added atomic multi-key snapshots, writes, deletions and conditional + updates to `WorkersD1KvStore`. Applications can now reserve a record + and update a shared budget together, with every comparison checked + before any change commits. Single-key and multi-key CAS share JSON + comparison and millisecond TTL rules; oversized calls reject instead + of splitting. [[#1361], [#1379], [#1432]] + + Production D1 migrations must create both the entry table and its + `_atomic` guard table, plus the expiry index, before using + `initialized: true`. Pass the D1 binding for fresh primary snapshots; + caller-owned sessions can read stale replicas. `WorkersKvStore` + remains without CAS or atomic multi-key support. diff --git a/changes.d/postgres/expiry-index-deparsing.md b/changes.d/postgres/expiry-index-deparsing.md new file mode 100644 index 000000000..d4749ffb0 --- /dev/null +++ b/changes.d/postgres/expiry-index-deparsing.md @@ -0,0 +1,11 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1379': https://github.com/fedify-dev/fedify/issues/1379 + '#1432': https://github.com/fedify-dev/fedify/pull/1432 +--- + - Fixed `PostgresKvStore` creating a redundant expiry index on PostgreSQL + 18.7 when the existing index uses the documented `AT TIME ZONE` + spelling. Index detection now compares the expression itself rather + than the version-dependent index-column wrapper. + [[#1361], [#1379], [#1432]] diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b7f3a6665..f256cc908 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -458,9 +458,17 @@ existing Deno applications throughout the 2.x series. On Cloudflare Workers, use `WorkersD1KvStore` (backed by D1) rather than `WorkersKvStore` (backed by Workers KV). Fedify's key–value workload is balanced between reads and writes and needs strong consistency, and only -`WorkersD1KvStore` supports `cas()`. Pass the `D1Database` binding itself, -not a D1 session, and create its table with a D1 migration and -`initialized: true`, since the store is constructed per request. +`WorkersD1KvStore` implements `AtomicKvStore`. Pass the `D1Database` binding +for fresh primary reads; caller-owned sessions can serve stale `getMany()` +snapshots even after their first `first-primary` query. CAS always uses a +write transaction with a guard row, including absence-only CAS. Create both +the entry table and `_atomic` guard table plus the expiry index +with a D1 migration before using `initialized: true`. The entry table needs +a single-column `key` primary index with `BINARY` collation. Follow the +complete schema and codec/TTL/batch limits in the +[KV manual][atomic support matrix]; a table name ending in `_atomic` is +reserved. Single and multi-key writes share the same protocol; oversized calls +reject instead of splitting. For application transitions across several keys, require `AtomicKvStore` from `@fedify/fedify` and narrow an existing `KvStore` with @@ -482,17 +490,17 @@ for a shared backend. Separate calls do not form a transaction; uncertain commit results need receipts or version checks, and ABA detection needs a non-reused version. -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full atomic -interface. -SQLite covers one configured table/database on a writable, normally locking -connection. Nonempty batches and ordinary mutations reject caller-owned -transactions and use one transaction; snapshots and mismatching CAS can reject -busy, including under rollback-journal reader contention. Configure busy -timeout on the connection; the KV store has no retry policy. Use decoded -JSON expectations, structural comparison, and the documented millisecond TTL -precision and resource limits. `SqliteMessageQueue` defaults to WAL on shared -files; concurrent WAL use needs a patched SQLite engine (Bun 1.2.22 -SQLite 3.50.4 is affected by the WAL-reset bug). See the adapter documentation. +`MemoryKvStore`, `SqliteKvStore`, `PgliteKvStore`, and `WorkersD1KvStore` +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 @@ -506,8 +514,9 @@ 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 -child issue. `WorkersKvStore` lacks CAS. `WorkersD1KvStore` supports -single-key CAS; its atomic extension is tracked in #1379. Check the +child issue. `WorkersKvStore` lacks CAS. D1 operations cover one +configured table in one database, with JSON serialization and millisecond TTLs. +Check the [atomic support matrix] and installed declarations before selecting a production adapter. diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 672166774..3f106eb3b 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -657,14 +657,14 @@ database. Fedify writes to its key–value store about as often as it reads from it, and some of its records, such as idempotency records and signature nonces, need strong consistency. D1 fits this workload better than [Cloudflare Workers KV], which is tuned for read-heavy workloads and is -eventually consistent. [`WorkersD1KvStore`] also supports -`~KvStore.cas()`, which some Fedify features need to work atomically. +eventually consistent. [`WorkersD1KvStore`] implements `AtomicKvStore`, +including single-key CAS, snapshot reads and conditional multi-key writes. Best for : Production use in Cloudflare Workers environments. Pros -: Persistent storage, strong consistency, supports `~KvStore.cas()`. +: Persistent storage, primary reads and atomic multi-key operations. Cons : Only available in Cloudflare Workers runtime. @@ -690,10 +690,16 @@ export default { The store keeps its entries in a table named `fedify_kv` by default, which you can change with the `tableName` option. Without `initialized: true`, -each store instance creates the table on its first use. Since you construct -the store for every request, that costs an extra D1 query per request, so -create the table with a [D1 migration][D1 migrations] in production and pass -`initialized: true`: +each store instance creates and checks its schema on first use. The names +`_atomic` and `idx__expires` are reserved for the guard +table and expiry index; table names ending in `_atomic` are rejected. +The entry table must have a single-column `key` primary index using `BINARY` +collation, so differently cased keys remain distinct. Automatic initialization +rejects incompatible primary indexes. Key lookups, mutations, and listing +explicitly use binary comparisons even if the column declares another collation. + +Create both tables and the index with a [D1 migration][D1 migrations] in +production, then pass `initialized: true` to avoid repeated schema queries: ~~~~ sql CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -702,12 +708,51 @@ CREATE TABLE IF NOT EXISTS fedify_kv ( expires INTEGER ); CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); +CREATE TABLE IF NOT EXISTS fedify_kv_atomic ( + singleton INTEGER PRIMARY KEY, + cutoff INTEGER NOT NULL, + matched INTEGER NOT NULL +); ~~~~ -Expired entries are never returned. Every `~KvStore.set()` and every -`~KvStore.cas()` that writes also purges up to 100 expired entries from the -table, so expired entries do not pile up while the store is in use. Entries -set without a TTL never expire and stay until they are deleted. +Atomic operations cover all keys in one configured table in one D1 database. +`getMany()` reads one snapshot and preserves input order, missing values and +duplicate keys. `setMany()`, `deleteMany()` and `casMany()` commit each batch +as a unit; `casMany()` compares every expectation before changing any entry. +Single-key `set()` and `cas()` use the same guard protocol. Each nonempty +set or CAS writes and removes a private guard row, including absence-only +CAS. A mismatch leaves all user values and TTLs unchanged. Backend errors +reject, and a lost response can leave commit status unknown. Re-read and +re-evaluate the application rule after a mismatch; include a non-reused +version in values if ABA changes must be detected. + +Values use `JSON.stringify()`/`JSON.parse()`, including their coercions: +nonfinite numbers become `null`, and undefined object members are omitted. +Top-level `undefined` is only an absence/deletion sentinel in CAS; it cannot +be stored. Functions, symbols, cycles and `BigInt` values without a JSON +serialization reject. CAS compares the persisted JSON structurally, ignoring +object member order while retaining array order and types. Values deeper +than SQLite's 1,000-level JSON limit use exact serialized comparison instead. +External guard writes, triggers and custom constraints that alter the +protocol, and raw JSON rows with duplicate object members, are unsupported. + +TTLs use one database-clock cutoff per operation and are truncated to integer +milliseconds. Nonpositive durations, including positive durations shorter +than one millisecond after truncation, expire immediately. Durations whose +absolute total exceeds `8.64e15` milliseconds, nonfinite totals, and calendar +durations requiring a reference date reject. Replacing without a TTL clears +the previous expiry. Successful sets and CAS also purge up to 100 entries +expired at the same cutoff; a mismatch does not run that cleanup. + +Snapshots accept up to 256 input keys, counting duplicates, and at most +8 MiB of encoded keys and values. For maximum-size values, the byte budget +usually binds first. Mutations accept up to 128 entries and 16 MiB of encoded +keys, expectations and replacements combined. Each encoded key/value row +and each bound string is limited to 1,999,000 UTF-8 bytes. `getMany()` and +`deleteMany()` keys also have a 1,999,000-byte aggregate encoded-key limit, and +the final escaped JSON key payload must fit the same limit. Oversized +operations reject without splitting; backend or runtime limits can also +reject smaller calls. > [!WARNING] > Pass the `D1Database` binding itself (e.g., `env.DB`) to get strongly @@ -1008,10 +1053,10 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full -`AtomicKvStore` contract. -Existing single-key CAS support does not imply atomic multi-key support. -The linked issues track separate adapter implementations: +`MemoryKvStore`, `SqliteKvStore`, `PgliteKvStore`, and `WorkersD1KvStore` +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 | | --------------------- | --------------- | ------------------------------------------------------------------------------------------------- | @@ -1024,13 +1069,11 @@ The linked issues track separate adapter implementations: | `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. | -| `WorkersD1KvStore` | No | Base adapter [PR #1360] supports single-key CAS; atomic follow-up is [#1379]. | +| `WorkersD1KvStore` | Yes | One configured table in one D1 database; see its codec, TTL, size and session restrictions above. | -[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 Bounding how long cache entries live diff --git a/packages/cfworkers/README.md b/packages/cfworkers/README.md index e3886cf99..ed4b132b3 100644 --- a/packages/cfworkers/README.md +++ b/packages/cfworkers/README.md @@ -72,14 +72,20 @@ export default { `WorkersD1KvStore` is a key–value store implementation for [Cloudflare Workers] that uses [Cloudflare D1]. D1 is strongly consistent, which suits Fedify's balanced read and write workload, and -`WorkersD1KvStore` supports the atomic `cas()` operation. It is the +`WorkersD1KvStore` implements `AtomicKvStore`, including atomic snapshots and +conditional multi-key writes. It is the recommended key–value store for Cloudflare Workers applications. -Without `initialized: true`, each store instance creates its table -(`fedify_kv` by default, configurable with the `tableName` option) on its -first use, which costs an extra D1 query per request. In production, create -the table with [D1 migrations] and pass `initialized: true`, as in the example -above: +Without `initialized: true`, each store instance creates and checks its schema +on first use. The entry table is `fedify_kv` by default, configurable with +`tableName`; `_atomic` and `idx__expires` are reserved +auxiliary names. The entry table must have a single-column `key` primary +index using `BINARY` collation, so differently cased keys remain distinct. +Automatic initialization rejects incompatible primary indexes. Key lookups, +mutations, and listing explicitly use binary comparisons even if the column +declares another collation. Table names ending in `_atomic` are rejected. In +production, create both tables and the index with [D1 migrations], then pass +`initialized: true` to avoid repeating schema queries per request: ~~~~ sql CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -88,10 +94,58 @@ CREATE TABLE IF NOT EXISTS fedify_kv ( expires INTEGER ); CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); +CREATE TABLE IF NOT EXISTS fedify_kv_atomic ( + singleton INTEGER PRIMARY KEY, + cutoff INTEGER NOT NULL, + matched INTEGER NOT NULL +); ~~~~ -Pass the `D1Database` binding itself rather than a D1 session: sessions only -provide sequential consistency and can read stale data from a replica. +Pass the `D1Database` binding itself for fresh primary reads. A caller-owned +D1 session provides sequential consistency, but `get()` and `getMany()` may +read stale replicas. `first-primary` affects only the session's first query; +it does not pin later reads to the primary. CAS uses a write transaction +with either input. The adapter keeps the caller's session and does not call +`withSession()` on its behalf. + +Atomic operations cover all keys in one configured table in one D1 database. +`getMany()` reads one snapshot and preserves input order, missing values and +duplicate keys. `setMany()`, `deleteMany()` and `casMany()` commit each batch +as a unit; `casMany()` compares every expectation before changing any entry. +Single-key `set()` and `cas()` use the same guard protocol. Each nonempty +set or CAS writes and removes a private guard row, including absence-only +CAS. A mismatch leaves all user values and TTLs unchanged. Backend errors +reject, and a lost response can leave commit status unknown. Re-read and +re-evaluate the application rule after a mismatch; include a non-reused +version in values if ABA changes must be detected. + +Values use `JSON.stringify()`/`JSON.parse()`, including their coercions: +nonfinite numbers become `null`, and undefined object members are omitted. +Top-level `undefined` is only an absence/deletion sentinel in CAS; it cannot +be stored. Functions, symbols, cycles and `BigInt` values without a JSON +serialization reject. CAS compares the persisted JSON structurally, ignoring +object member order while retaining array order and types. Values deeper +than SQLite's 1,000-level JSON limit use exact serialized comparison instead. +External guard writes, triggers and custom constraints that alter the +protocol, and raw JSON rows with duplicate object members, are unsupported. + +TTLs use one database-clock cutoff per operation and are truncated to integer +milliseconds. Nonpositive durations, including positive durations shorter +than one millisecond after truncation, expire immediately. Durations whose +absolute total exceeds `8.64e15` milliseconds, nonfinite totals, and calendar +durations requiring a reference date reject. Replacing without a TTL clears +the previous expiry. Successful sets and CAS also purge up to 100 entries +expired at the same cutoff; a mismatch does not run that cleanup. + +Snapshots accept up to 256 input keys, counting duplicates, and at most +8 MiB of encoded keys and values. For maximum-size values, the byte budget +usually binds first. Mutations accept up to 128 entries and 16 MiB of encoded +keys, expectations and replacements combined. Each encoded key/value row +and each bound string is limited to 1,999,000 UTF-8 bytes. `getMany()` and +`deleteMany()` keys also have a 1,999,000-byte aggregate encoded-key limit, and +the final escaped JSON key payload must fit the same limit. Oversized +operations reject without splitting; backend or runtime limits can also +reject smaller calls. [Cloudflare D1]: https://developers.cloudflare.com/d1/ [D1 migrations]: https://developers.cloudflare.com/d1/reference/migrations/ diff --git a/packages/cfworkers/package.json b/packages/cfworkers/package.json index cd3c3a4d2..a13f67720 100644 --- a/packages/cfworkers/package.json +++ b/packages/cfworkers/package.json @@ -67,7 +67,9 @@ "devDependencies": { "@cloudflare/vitest-pool-workers": "^0.8.31", "@cloudflare/workers-types": "catalog:", + "@fedify/testing": "workspace:^", "semver": "^7.7.2", + "temporal-polyfill": "catalog:", "tsdown": "catalog:", "typescript": "catalog:", "vitest": "~3.2.0", @@ -79,6 +81,6 @@ "prepack": "pnpm build", "prepublish": "pnpm build", "pretest": "pnpm build", - "test": "tsc -p test/typecheck/tsconfig.json --noEmit && node --test test/peer-dependencies.check.mjs && vitest run" + "test": "tsc -p test/typecheck/tsconfig.json --noEmit && node --test test/peer-dependencies.check.mjs test/d1.atomic.check.mjs && vitest run" } } diff --git a/packages/cfworkers/src/mod.ts b/packages/cfworkers/src/mod.ts index 814a4f9b7..21fbcc550 100644 --- a/packages/cfworkers/src/mod.ts +++ b/packages/cfworkers/src/mod.ts @@ -11,6 +11,9 @@ */ import type { MessageSendRequest, Queue } from "@cloudflare/workers-types"; import type { + AtomicKvStore, + KvStoreCasEntry, + KvStoreSetEntry, KvKey, KvStore, KvStoreListEntry, @@ -285,19 +288,20 @@ export interface WorkersD1KvStoreOptions { /** * The table name to use for the key–value store. It must be 1 to 64 * characters long, consist of ASCII letters, digits, and underscores, and - * not start with a digit. Names starting with `sqlite_` or `_cf_` + * not start with a digit or end with `_atomic`. Names starting with `sqlite_` or `_cf_` * (case-insensitively) are reserved by SQLite and D1, so they are rejected * too. The store also creates an index named `idx__expires`, - * so no other table or index may have that name. + * and an auxiliary table named `_atomic`, so no unrelated object + * may use either name. * `"fedify_kv"` by default. * @default `"fedify_kv"` */ readonly tableName?: string; /** - * Whether the table has already been created, for example, by a D1 + * Whether both tables and the index have already been created, for example, by a D1 * migration. If `false`, each store instance creates the table on its - * first use, which costs an extra D1 query per instance. Since a store is + * first use, then validates the schema with metadata queries. Since a store is * usually constructed per request on Cloudflare Workers, create the table * with a D1 migration and set this to `true` in production. * `false` by default. @@ -306,6 +310,13 @@ export interface WorkersD1KvStoreOptions { readonly initialized?: boolean; } +interface D1PreparedMutation { + readonly key: string; + readonly expected: string | undefined; + readonly value: string | undefined; + readonly ttl: number | null; +} + /** * The current time in milliseconds since the Unix epoch, as seen by D1. * Comparing expiration against the database clock keeps each statement @@ -335,22 +346,46 @@ const D1_LIST_PAGE_BYTES = 8 * 1024 * 1024; const D1_EXPIRE_BATCH_SIZE = 100; /** - * Implementation of the {@link KvStore} interface for Cloudflare D1 binding. + * An {@link AtomicKvStore} backed by Cloudflare D1. All keys in one configured + * table share an atomic scope. Each nonempty set or CAS uses one transactional + * batch and a private guard row to freeze its decision and database-clock + * expiry cutoff. This also routes absence-only CAS through a write. + * + * Pass the `D1Database` binding for fresh primary reads. Caller-owned + * `D1DatabaseSession` objects are also accepted: their reads, including the + * single-statement snapshot from {@link getMany}, may use stale replicas. + * Sessions retain their own bookmark and sequential-consistency behavior; + * the adapter never replaces them or calls `withSession()`. * - * Unlike {@link WorkersKvStore}, this store is strongly consistent when given - * a `D1Database` binding, which suits Fedify's balanced read/write usage of - * the key–value store. It also supports the {@link KvStore.cas} operation: - * since D1 has no interactive transactions, each compare-and-swap is - * performed by a single conditional SQL statement. + * Values round-trip through `JSON.stringify()` and `JSON.parse()`, including + * JSON coercions such as nonfinite numbers becoming `null` and omitted + * undefined object members. Values without a JSON serialization reject. + * CAS compares persisted JSON structurally, ignoring object member order but + * preserving array order and types. Beyond SQLite's 1,000-level JSON depth + * limit, comparison falls back to exact serialized equality. Include a + * non-reused version in values when ABA changes must be detected. * - * A `D1DatabaseSession` is accepted too, but sessions only provide D1's - * sequential consistency, and reads may be served by a stale replica. - * Fedify relies on fresh reads for idempotency checks, so pass the - * `D1Database` binding itself (e.g., `env.DB`) unless you know you can - * tolerate stale reads. + * TTLs are truncated to integer milliseconds, with an absolute maximum of + * 8.64e15 milliseconds. Calendar durations requiring a reference date reject. + * Nonpositive TTLs expire immediately; omitting a TTL clears previous expiry. + * Successful sets and CAS purge at most 100 rows expired at the same cutoff. + * Mismatch leaves all user rows and their TTLs unchanged. * - * The store uses a table with the following schema, which each instance - * creates on its first use unless the `initialized` option is set: + * Snapshots accept at most 256 input keys, including duplicates, and return at + * most 8 MiB of encoded keys and values. Mutations accept at most 128 entries + * and 16 MiB of encoded keys, expectations and replacements combined. Each + * encoded key/value row, bound string, aggregate snapshot/deletion key input, + * and final JSON key payload is limited to 1,999,000 UTF-8 bytes. These are + * conservative adapter limits; backend/runtime limits may reject smaller + * calls. Operations are never split. Errors propagate without write retry; + * a lost response may leave commit status unknown. `false` means only a + * proven CAS mismatch, not a backend failure or a durable operation receipt. + * + * Automatic initialization creates and checks the following schema. Use a + * D1 migration with both tables and the index, and `initialized: true`, to + * avoid repeating schema queries in each request. Derived object names are + * reserved; external guard writes, triggers and custom constraints that alter + * this protocol are unsupported. The guard table must normally be empty. * * ```sql * CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -359,6 +394,11 @@ const D1_EXPIRE_BATCH_SIZE = 100; * expires INTEGER * ); * CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); + * CREATE TABLE IF NOT EXISTS fedify_kv_atomic ( + * singleton INTEGER PRIMARY KEY, + * cutoff INTEGER NOT NULL, + * matched INTEGER NOT NULL + * ); * ``` * * @example @@ -379,7 +419,7 @@ const D1_EXPIRE_BATCH_SIZE = 100; * ``` * @since 2.5.0 */ -export class WorkersD1KvStore implements KvStore { +export class WorkersD1KvStore implements AtomicKvStore { static readonly #defaultTableName = "fedify_kv"; static readonly #tableNameRegex = /^[A-Za-z_][A-Za-z0-9_]{0,63}$/; static readonly #reservedTableNameRegex = /^(sqlite_|_cf_)/i; @@ -403,7 +443,8 @@ export class WorkersD1KvStore implements KvStore { this.#tableName = options.tableName ?? WorkersD1KvStore.#defaultTableName; if ( !WorkersD1KvStore.#tableNameRegex.test(this.#tableName) || - WorkersD1KvStore.#reservedTableNameRegex.test(this.#tableName) + WorkersD1KvStore.#reservedTableNameRegex.test(this.#tableName) || + /_atomic$/i.test(this.#tableName) ) { throw new TypeError( `Invalid table name for the key–value store: ${this.#tableName}`, @@ -413,192 +454,465 @@ export class WorkersD1KvStore implements KvStore { } /** - * Creates the table used by the key–value store if it does not already - * exist. Does nothing if the table already exists. It is called + * Creates both tables and the expiry index if absent, and verifies the + * binary key primary index, auxiliary table, and expiry index metadata. + * It is called * automatically on the first use of the store instance unless * the `initialized` option is set. */ initialize(): Promise { this.#initialized ??= this.#db .batch([ - this.#db.prepare(` - CREATE TABLE IF NOT EXISTS "${this.#tableName}" ( - key TEXT PRIMARY KEY NOT NULL, - value TEXT NOT NULL, - expires INTEGER - ) - `), - this.#db.prepare(` - CREATE INDEX IF NOT EXISTS "idx_${this.#tableName}_expires" - ON "${this.#tableName}" (expires) - `), + this.#db.prepare(`CREATE TABLE IF NOT EXISTS "${this.#tableName}" ( + key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL, expires INTEGER + )`), + this.#db + .prepare(`CREATE INDEX IF NOT EXISTS "idx_${this.#tableName}_expires" + ON "${this.#tableName}" (expires)`), + this.#db + .prepare(`CREATE TABLE IF NOT EXISTS "${this.#tableName}_atomic" ( + singleton INTEGER PRIMARY KEY, cutoff INTEGER NOT NULL, matched INTEGER NOT NULL + )`), ]) - .then( - () => undefined, - (error) => { - // Let a later call retry the initialization: - this.#initialized = null; - throw error; - }, - ); + .then(async () => { + const [ + { results: columns }, + owner, + { results: index }, + { results: indexes }, + object, + ] = await Promise.all([ + this.#db + .prepare(`PRAGMA table_xinfo("${this.#tableName}_atomic")`) + .all<{ + name: string; + type: string; + notnull: number; + pk: number; + hidden: number; + }>(), + this.#db + .prepare( + "SELECT tbl_name FROM sqlite_master WHERE type = 'index' AND name = ?1 COLLATE NOCASE", + ) + .bind(`idx_${this.#tableName}_expires`) + .first<{ tbl_name: string }>(), + this.#db + .prepare(`PRAGMA index_info("idx_${this.#tableName}_expires")`) + .all<{ name: string | null }>(), + this.#db.prepare(`PRAGMA index_list("${this.#tableName}")`).all<{ + name: string; + unique: number; + partial: number; + origin: string; + }>(), + this.#db + .prepare( + "SELECT type FROM sqlite_master WHERE name = ?1 COLLATE NOCASE", + ) + .bind(`${this.#tableName}_atomic`) + .first<{ type: string }>(), + ]); + const primaryIndex = indexes.find((entry) => entry.origin === "pk"); + const primaryColumns = + primaryIndex == null + ? [] + : ( + await this.#db + .prepare( + `PRAGMA index_xinfo("${primaryIndex.name.replaceAll('"', '""')}")`, + ) + .all<{ name: string | null; coll: string; key: number }>() + ).results; + const keyColumns = primaryColumns.filter((column) => column.key === 1); + const shape = new Map( + columns.map((column) => [column.name.toLowerCase(), column]), + ); + const expiryIndex = indexes.find( + (entry) => + entry.name.toLowerCase() === + `idx_${this.#tableName}_expires`.toLowerCase(), + ); + if ( + primaryIndex?.unique !== 1 || + primaryIndex?.partial !== 0 || + keyColumns.length !== 1 || + keyColumns[0].name?.toLowerCase() !== "key" || + keyColumns[0].coll.toUpperCase() !== "BINARY" || + object?.type !== "table" || + columns.length !== 3 || + columns.some((column) => column.hidden !== 0) || + shape.get("singleton")?.type.toUpperCase() !== "INTEGER" || + shape.get("singleton")?.pk !== 1 || + shape.get("cutoff")?.type.toUpperCase() !== "INTEGER" || + shape.get("cutoff")?.notnull !== 1 || + shape.get("cutoff")?.pk !== 0 || + shape.get("matched")?.type.toUpperCase() !== "INTEGER" || + shape.get("matched")?.notnull !== 1 || + shape.get("matched")?.pk !== 0 || + owner?.tbl_name.toLowerCase() !== this.#tableName.toLowerCase() || + index.length !== 1 || + index[0].name?.toLowerCase() !== "expires" || + expiryIndex?.unique !== 0 || + expiryIndex?.partial !== 0 + ) { + throw new TypeError( + `Incompatible D1 KV schema for "${this.#tableName}" or its atomic table/expiry index.`, + ); + } + }) + .catch((error) => { + this.#initialized = null; + throw error; + }); return this.#initialized; } /** - * Drops the table used by the key–value store. Does nothing if the table - * does not exist. Mostly useful for tests. + * Drops both tables reserved by this store. Mostly useful for tests. + * No unrelated object may use either table name. */ async drop(): Promise { - await this.#db.prepare(`DROP TABLE IF EXISTS "${this.#tableName}"`).run(); + await this.#db.batch([ + this.#db.prepare(`DROP TABLE IF EXISTS "${this.#tableName}_atomic"`), + this.#db.prepare(`DROP TABLE IF EXISTS "${this.#tableName}"`), + ]); this.#initialized = null; } #encodeKey(key: KvKey): string { - return JSON.stringify(key); + if (!Array.isArray(key)) { + throw new TypeError("A KV key must be a non-empty array of strings."); + } + const length = key.length; + if (!Number.isSafeInteger(length) || length < 1) { + throw new TypeError("A KV key must be a non-empty array of strings."); + } + const parts: string[] = []; + for (let index = 0; index < length; index++) { + if (!Object.hasOwn(key, index)) + throw new TypeError("A KV key must not be sparse."); + const part = key[index]; + if (typeof part !== "string") + throw new TypeError("A KV key must contain only strings."); + parts.push(part); + } + const encoded = JSON.stringify(parts); + this.#checkBytes(encoded); + return encoded; } - #getTtl(options?: KvStoreSetOptions): number | null { - return options?.ttl == null ? null : options.ttl.total("milliseconds"); + #checkBytes(value: string): number { + const bytes = new TextEncoder().encode(value).byteLength; + if (bytes > 1_999_000) + throw new RangeError("D1 KV string/row size limit exceeded."); + return bytes; } - /** - * Deletes a bounded number of expired entries, so that a write never has to - * sweep an unbounded backlog of them at once. - */ - #expire(): WorkersD1PreparedStatementLike { - return this.#db.prepare(` - DELETE FROM "${this.#tableName}" - WHERE key IN ( - SELECT key FROM "${this.#tableName}" - WHERE expires <= ${D1_NOW} - LIMIT ${D1_EXPIRE_BATCH_SIZE} - ) - `); + #array(items: readonly T[], limit: number): T[] { + if (!Array.isArray(items)) + throw new TypeError("A KV batch must be an array."); + const length = items.length; + if (!Number.isSafeInteger(length) || length < 0) + throw new TypeError("Invalid KV batch length."); + if (length > limit) + throw new RangeError(`D1 KV batch exceeds ${limit} entries.`); + const captured: T[] = []; + for (let index = 0; index < length; index++) { + if (!Object.hasOwn(items, index)) + throw new TypeError("A KV batch must not be sparse."); + captured.push(items[index]); + } + return captured; } - /** - * {@inheritDoc KvStore.get} - */ + #encodeValue(value: unknown): string { + const encoded = JSON.stringify(value); + if (encoded === undefined) + throw new TypeError("Value has no JSON serialization."); + this.#checkBytes(encoded); + return encoded; + } + + #prepare( + entries: readonly KvStoreCasEntry[], + compare: boolean, + ): D1PreparedMutation[] { + const captured = this.#array(entries, 128).map((entry) => ({ + key: entry.key, + expected: entry.expectedValue, + value: entry.newValue, + ttl: entry.options?.ttl, + })); + if (captured.length > 128) + throw new RangeError("D1 KV mutation exceeds 128 entries."); + const seen = new Set(); + let bytes = 0; + return captured.map((entry) => { + const key = this.#encodeKey(entry.key); + if (seen.has(key)) + throw new TypeError("Duplicate keys in an atomic mutation."); + seen.add(key); + if (!compare && entry.value === undefined) { + throw new TypeError("undefined is not a storable KV value."); + } + if (entry.value === undefined && entry.ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } + const expected = + entry.expected === undefined + ? undefined + : this.#encodeValue(entry.expected); + const value = + entry.value === undefined ? undefined : this.#encodeValue(entry.value); + let ttl: number | null = null; + if (entry.ttl != null) { + const total = entry.ttl.total("milliseconds"); + if (!Number.isFinite(total) || Math.abs(total) > 8.64e15) { + throw new RangeError( + "D1 KV TTL exceeds the supported millisecond range.", + ); + } + ttl = Math.trunc(total); + } + this.#checkBytes(key + (value ?? "")); + bytes += + this.#checkBytes(key) + + (expected == null ? 0 : this.#checkBytes(expected)) + + (value == null ? 0 : this.#checkBytes(value)); + if (bytes > 16 * 1024 * 1024) + throw new RangeError("D1 KV mutation exceeds 16 MiB."); + return { key, expected, value, ttl }; + }); + } + + // Resolve JSON virtual tables outside main so application tables cannot shadow them. + #matches(current: string, expected: string): string { + return `CASE WHEN json_valid(${current}) AND json_valid(${expected}) THEN + NOT EXISTS ( + SELECT fullkey, type, atom FROM temp.json_tree(${current}) + EXCEPT SELECT fullkey, type, atom FROM temp.json_tree(${expected}) + ) AND NOT EXISTS ( + SELECT fullkey, type, atom FROM temp.json_tree(${expected}) + EXCEPT SELECT fullkey, type, atom FROM temp.json_tree(${current}) + ) ELSE ${current} = ${expected} END`; + } + + async #mutate( + entries: readonly D1PreparedMutation[], + compare: boolean, + ): Promise { + if (entries.length === 0) return true; + await this.initialize(); + const guard = `"${this.#tableName}_atomic"`; + const cutoff = `(SELECT cutoff FROM ${guard} WHERE singleton = 1)`; + const permitted = `(SELECT matched FROM ${guard} WHERE singleton = 1) = 1`; + // The first write captures one cutoff and forces a primary transaction. + const statements: WorkersD1PreparedStatementLike[] = [ + this.#db.prepare(` + INSERT INTO ${guard} (singleton, cutoff, matched) VALUES (1, ${D1_NOW}, 1) + `), + ]; + // Freeze every comparison before changing any user row. + if (compare) { + for (const entry of entries) { + const live = `SELECT 1 FROM "${this.#tableName}" AS current + WHERE current.key COLLATE BINARY = ?1 AND (current.expires IS NULL OR current.expires > ${cutoff})`; + const predicate = + entry.expected === undefined + ? `NOT EXISTS (${live})` + : `EXISTS (${live} AND ${this.#matches("current.value", "?2")})`; + const statement = this.#db.prepare(`UPDATE ${guard} SET matched = 0 + WHERE singleton = 1 AND matched = 1 AND NOT (${predicate})`); + statements.push( + entry.expected === undefined + ? statement.bind(entry.key) + : statement.bind(entry.key, entry.expected), + ); + } + } + for (const entry of entries) { + statements.push( + entry.value === undefined + ? this.#db + .prepare(` + DELETE FROM "${this.#tableName}" WHERE key COLLATE BINARY = ?1 AND ${permitted} + `) + .bind(entry.key) + : this.#db + .prepare(` + INSERT INTO "${this.#tableName}" (key, value, expires) + SELECT ?1, ?2, ${cutoff} + ?3 WHERE ${permitted} + ON CONFLICT (key COLLATE BINARY) DO UPDATE SET value = excluded.value, expires = excluded.expires + `) + .bind(entry.key, entry.value, entry.ttl), + ); + } + statements.push( + this.#db.prepare(`DELETE FROM "${this.#tableName}" + WHERE ${permitted} AND key COLLATE BINARY IN (SELECT key FROM "${this.#tableName}" + WHERE expires <= ${cutoff} LIMIT ${D1_EXPIRE_BATCH_SIZE})`), + ); + // These two result counts distinguish mismatch from a missing guard. + const resultIndex = statements.length; + statements.push( + this.#db.prepare( + `DELETE FROM ${guard} WHERE singleton = 1 AND matched = 1`, + ), + ); + statements.push( + this.#db.prepare(`DELETE FROM ${guard} WHERE singleton = 1`), + ); + const results = await this.#db.batch(statements); + const matched = results[resultIndex].meta.changes; + const cleaned = results[resultIndex + 1].meta.changes; + if (matched === 1 && cleaned === 0) return true; + if (matched === 0 && cleaned === 1) return false; + throw new Error( + `D1 KV atomic guard integrity failure for "${this.#tableName}_atomic".`, + ); + } + + /** {@inheritDoc KvStore.get} */ async get(key: KvKey): Promise { + const encoded = this.#encodeKey(key); await this.initialize(); const row = await this.#db .prepare(` SELECT value FROM "${this.#tableName}" - WHERE key = ? AND (expires IS NULL OR expires > ${D1_NOW}) + WHERE key COLLATE BINARY = ?1 AND (expires IS NULL OR expires > ${D1_NOW}) `) - .bind(this.#encodeKey(key)) + .bind(encoded) .first<{ value: string }>(); return row == null ? undefined : (JSON.parse(row.value) as T); } - /** - * {@inheritDoc KvStore.set} - */ + /** {@inheritDoc AtomicKvStore.getMany} */ + async getMany( + keys: readonly KvKey[], + ): Promise<(T | undefined)[]> { + const captured = this.#array(keys, 256); + if (captured.length > 256) + throw new RangeError("D1 KV snapshot exceeds 256 input keys."); + let keyBytes = 0; + const encoded = captured.map((key) => { + const value = this.#encodeKey(key); + keyBytes += this.#checkBytes(value); + if (keyBytes > 1_999_000) + throw new RangeError("D1 KV snapshot key size limit exceeded."); + return value; + }); + const unique = [...new Set(encoded)]; + if (unique.length > 256) + throw new RangeError("D1 KV snapshot exceeds 256 unique keys."); + const payload = JSON.stringify(unique); + this.#checkBytes(payload); + if (unique.length === 0) return []; + await this.initialize(); + const { results } = await this.#db + .prepare(` + WITH cutoff(now) AS MATERIALIZED (SELECT ${D1_NOW}), + live AS MATERIALIZED ( + SELECT current.key, octet_length(current.key) + octet_length(current.value) AS size FROM temp.json_each(?1) AS input + JOIN main."${this.#tableName}" AS current ON current.key COLLATE BINARY = input.value + CROSS JOIN cutoff + WHERE current.expires IS NULL OR current.expires > cutoff.now + ), budget AS (SELECT COALESCE(SUM(size), 0) AS bytes FROM live) + SELECT live.key, CASE WHEN bytes <= 8388608 THEN current.value ELSE NULL END AS value, bytes + FROM live CROSS JOIN budget JOIN main."${this.#tableName}" AS current ON current.key COLLATE BINARY = live.key + `) + .bind(payload) + .all<{ key: string; value: string | null; bytes: number }>(); + if (results.some((row) => row.bytes > 8 * 1024 * 1024)) { + throw new RangeError("D1 KV snapshot exceeds 8 MiB."); + } + const values = new Map( + results.map((row) => [row.key, JSON.parse(row.value!) as T] as const), + ); + return encoded.map((key) => values.get(key)); + } + + /** {@inheritDoc AtomicKvStore.set} */ async set( key: KvKey, value: unknown, options?: KvStoreSetOptions, ): Promise { - await this.initialize(); - await this.#db.batch([ - this.#db - .prepare(` - INSERT INTO "${this.#tableName}" (key, value, expires) - VALUES (?1, ?2, ${D1_NOW} + ?3) - ON CONFLICT (key) DO UPDATE SET - value = excluded.value, - expires = excluded.expires - `) - .bind( - this.#encodeKey(key), - JSON.stringify(value), - this.#getTtl(options), - ), - this.#expire(), - ]); + await this.setMany([{ key, value, options }]); } - /** - * {@inheritDoc KvStore.delete} - */ + /** {@inheritDoc AtomicKvStore.setMany} */ + async setMany(entries: readonly KvStoreSetEntry[]): Promise { + const captured = this.#array(entries, 128).map((entry) => ({ + key: entry.key, + newValue: entry.value, + expectedValue: undefined, + options: entry.options, + })); + if (!(await this.#mutate(this.#prepare(captured, false), false))) { + throw new Error("D1 KV mutation lost its atomic guard."); + } + } + + /** {@inheritDoc KvStore.delete} */ async delete(key: KvKey): Promise { + const encoded = this.#encodeKey(key); await this.initialize(); await this.#db - .prepare(`DELETE FROM "${this.#tableName}" WHERE key = ?`) - .bind(this.#encodeKey(key)) + .prepare(`DELETE FROM "${this.#tableName}" WHERE key COLLATE BINARY = ?1`) + .bind(encoded) .run(); } - /** - * {@inheritDoc KvStore.cas} - */ + /** {@inheritDoc AtomicKvStore.deleteMany} */ + async deleteMany(keys: readonly KvKey[]): Promise { + if (!Array.isArray(keys)) + throw new TypeError("A KV batch must be an array."); + const length = keys.length; + if (!Number.isSafeInteger(length) || length < 0) + throw new TypeError("Invalid KV batch length."); + if (length > 128) + throw new RangeError("D1 KV mutation exceeds 128 entries."); + let bytes = 0; + const seen = new Set(); + const encoded: string[] = []; + for (let index = 0; index < length; index++) { + if (!Object.hasOwn(keys, index)) + throw new TypeError("A KV batch must not be sparse."); + const value = this.#encodeKey(keys[index]); + if (seen.has(value)) + throw new TypeError("Duplicate keys in an atomic mutation."); + bytes += this.#checkBytes(value); + if (bytes > 1_999_000) + throw new RangeError("D1 KV deletion key size limit exceeded."); + seen.add(value); + encoded.push(value); + } + const payload = JSON.stringify(encoded); + this.#checkBytes(payload); + if (encoded.length === 0) return; + await this.initialize(); + await this.#db.batch([ + this.#db + .prepare(`DELETE FROM "${this.#tableName}" + WHERE key COLLATE BINARY IN (SELECT value FROM temp.json_each(?1))`) + .bind(payload), + ]); + } + + /** {@inheritDoc AtomicKvStore.cas} */ async cas( key: KvKey, expectedValue: unknown, newValue: unknown, options?: KvStoreSetOptions, ): Promise { - await this.initialize(); - const encodedKey = this.#encodeKey(key); - let statement: WorkersD1PreparedStatementLike; - - if (expectedValue === undefined) { - // No mutation is needed; this read is the operation's atomic point: - if (newValue === undefined) return (await this.get(key)) === undefined; - // Insert the entry unless an unexpired entry already exists: - statement = this.#db - .prepare(` - INSERT INTO "${this.#tableName}" (key, value, expires) - VALUES (?1, ?2, ${D1_NOW} + ?3) - ON CONFLICT (key) DO UPDATE SET - value = excluded.value, - expires = excluded.expires - WHERE "${this.#tableName}".expires <= ${D1_NOW} - `) - .bind(encodedKey, JSON.stringify(newValue), this.#getTtl(options)); - } else { - // Compares the JSON trees in both directions so that the order of - // object members is ignored, while the order of array elements, - // container types, and scalar types matter. The predicate runs inside - // the mutation itself, so neither a read–write race nor an interactive - // transaction is involved. json_tree() rejects JSON nested deeper than - // 1,000 levels, so such values are compared by their serialization. - // The column must be qualified, since json_tree() has a value column: - const current = `"${this.#tableName}".value`; - const matches = ` - key = ?1 AND (expires IS NULL OR expires > ${D1_NOW}) - AND CASE WHEN json_valid(${current}) AND json_valid(?2) THEN - NOT EXISTS ( - SELECT fullkey, type, atom FROM json_tree(${current}) - EXCEPT SELECT fullkey, type, atom FROM json_tree(?2) - ) - AND NOT EXISTS ( - SELECT fullkey, type, atom FROM json_tree(?2) - EXCEPT SELECT fullkey, type, atom FROM json_tree(${current}) - ) - ELSE ${current} = ?2 END - `; - const expectedJson = JSON.stringify(expectedValue); - statement = - newValue === undefined - ? this.#db - .prepare(`DELETE FROM "${this.#tableName}" WHERE ${matches}`) - .bind(encodedKey, expectedJson) - : this.#db - .prepare(` - UPDATE "${this.#tableName}" - SET value = ?3, expires = ${D1_NOW} + ?4 - WHERE ${matches} - `) - .bind( - encodedKey, - expectedJson, - JSON.stringify(newValue), - this.#getTtl(options), - ); - } - const [{ meta }] = await this.#db.batch([statement, this.#expire()]); - return meta.changes > 0; + return await this.casMany([{ key, expectedValue, newValue, options }]); + } + + /** {@inheritDoc AtomicKvStore.casMany} */ + async casMany(entries: readonly KvStoreCasEntry[]): Promise { + return await this.#mutate(this.#prepare(entries, true), true); } /** @@ -627,23 +941,23 @@ export class WorkersD1KvStore implements KvStore { const statement = this.#db.prepare(` WITH page AS ( SELECT key, octet_length(key) + octet_length(value) AS size - FROM "${this.#tableName}" - WHERE key > ?1 AND key < ?2 + FROM main."${this.#tableName}" + WHERE key COLLATE BINARY > ?1 AND key COLLATE BINARY < ?2 AND (expires IS NULL OR expires > ${D1_NOW}) - ORDER BY key + ORDER BY key COLLATE BINARY LIMIT ${D1_LIST_PAGE_ROWS} ), sized AS ( SELECT key, size, - SUM(size) OVER (ORDER BY key ROWS UNBOUNDED PRECEDING) AS total, + SUM(size) OVER (ORDER BY key COLLATE BINARY ROWS UNBOUNDED PRECEDING) AS total, COUNT(*) OVER () AS page_rows FROM page ) SELECT t.key, t.value, sized.page_rows - FROM sized JOIN "${this.#tableName}" AS t ON t.key = sized.key + FROM sized JOIN main."${this.#tableName}" AS t ON t.key COLLATE BINARY = sized.key WHERE sized.total - sized.size < ${D1_LIST_PAGE_BYTES} - ORDER BY t.key + ORDER BY t.key COLLATE BINARY `); while (true) { const { results } = await statement diff --git a/packages/cfworkers/test/d1.atomic.check.mjs b/packages/cfworkers/test/d1.atomic.check.mjs new file mode 100644 index 000000000..ff5d2d78d --- /dev/null +++ b/packages/cfworkers/test/d1.atomic.check.mjs @@ -0,0 +1,62 @@ +// Run the shared timing-sensitive suite in Node: the Workers Vitest runner's +// performance/timer clock can lag SQLite's wall clock after many binding calls. +// Wrangler still executes every SQL operation in local workerd/Miniflare D1. +import { testAtomicKvStore } from "@fedify/testing"; +import { randomUUID } from "node:crypto"; +import { test } from "node:test"; +import { fileURLToPath } from "node:url"; +import { Temporal } from "temporal-polyfill"; +import { getPlatformProxy } from "wrangler"; +import { WorkersD1KvStore } from "../dist/mod.js"; + +test( + "D1 shared atomic conformance through a local binding", + { timeout: 600_000 }, + async () => { + const proxy = await getPlatformProxy({ + configPath: fileURLToPath(new URL("../wrangler.jsonc", import.meta.url)), + persist: false, + remoteBindings: false, + }); + try { + await testAtomicKvStore( + () => { + const tableName = `atomic_${randomUUID().replaceAll("-", "")}`; + const first = new WorkersD1KvStore( + proxy.env.D1.withSession("first-primary"), + { tableName }, + ); + const second = new WorkersD1KvStore( + proxy.env.D1.withSession("first-primary"), + { tableName }, + ); + return { first, second, dispose: () => first.drop() }; + }, + { + objectKeyOrder: "ignored", + ttl: { + short: Temporal.Duration.from({ seconds: 5 }), + wait: Temporal.Duration.from({ seconds: 6 }), + }, + rejections: (key) => ({ + getMany: Array.from({ length: 257 }, (_, i) => key("read", `${i}`)), + setMany: Array.from({ length: 129 }, (_, i) => ({ + key: key("set", `${i}`), + value: i, + })), + deleteMany: Array.from({ length: 129 }, (_, i) => + key("delete", `${i}`), + ), + casMany: Array.from({ length: 129 }, (_, i) => ({ + key: key("cas", `${i}`), + expectedValue: 0, + newValue: 1, + })), + }), + }, + ); + } finally { + await proxy.dispose(); + } + }, +); diff --git a/packages/cfworkers/test/d1.test.ts b/packages/cfworkers/test/d1.test.ts index d73cdbd73..1414ed488 100644 --- a/packages/cfworkers/test/d1.test.ts +++ b/packages/cfworkers/test/d1.test.ts @@ -1,3 +1,4 @@ +import { Temporal } from "temporal-polyfill"; import { env } from "cloudflare:test"; import { describe, expect, it } from "vitest"; import { @@ -7,13 +8,15 @@ import { } from "../src/mod.ts"; // Mock Temporal.Duration for testing in Cloudflare Workers environment -const mockDuration = (seconds: number) => ({ - total: (unit: string) => { - if (unit === "milliseconds" || unit === "millisecond") return seconds * 1000; - if (unit === "seconds" || unit === "second") return seconds; - return seconds; - }, -}) as unknown as Temporal.Duration; +const mockDuration = (seconds: number) => + ({ + total: (unit: string) => { + if (unit === "milliseconds" || unit === "millisecond") + return seconds * 1000; + if (unit === "seconds" || unit === "second") return seconds; + return seconds; + }, + }) as unknown as Temporal.Duration; async function insertRaw( key: readonly string[], @@ -24,7 +27,9 @@ async function insertRaw( await env.D1.prepare( `INSERT OR REPLACE INTO "${tableName}" (key, value, expires) VALUES (?, ?, ?)`, - ).bind(JSON.stringify(key), JSON.stringify(value), expires).run(); + ) + .bind(JSON.stringify(key), JSON.stringify(value), expires) + .run(); } const delay = (ms: number) => @@ -35,6 +40,9 @@ interface D1SpyOptions { readonly gate?: (queries: readonly string[]) => Promise | void; /** Whether to fail a batch instead of sending it, by its statements. */ readonly failBatch?: (queries: readonly string[]) => boolean; + readonly rewrite?: (query: string) => string; + readonly appendFailure?: boolean; + readonly guardCounts?: readonly [number, number]; } /** A prepared statement that remembers its query and the underlying one. */ @@ -54,9 +62,7 @@ class SpyStatement implements WorkersD1PreparedStatementLike { return await this.inner.first(); } - async all>(): Promise< - { readonly results: T[] } - > { + async all>(): Promise<{ readonly results: T[] }> { await this.spy.send([this.query]); return await this.inner.all(); } @@ -81,7 +87,8 @@ class D1Spy implements WorkersD1DatabaseLike { ) {} prepare(query: string): SpyStatement { - return new SpyStatement(this, query, this.db.prepare(query)); + const rewritten = this.options.rewrite?.(query) ?? query; + return new SpyStatement(this, rewritten, this.db.prepare(rewritten)); } async send(queries: string[]): Promise { @@ -97,7 +104,24 @@ class D1Spy implements WorkersD1DatabaseLike { this.batches.push(queries); if (this.options.failBatch?.(queries)) throw new Error("Injected failure"); await this.send(queries); - return await this.db.batch(spyStatements.map((s) => s.inner)); + const inner = spyStatements.map((s) => s.inner); + const atomic = queries.some((q) => + /INSERT INTO "fedify_kv_atomic"/.test(q), + ); + if (atomic && this.options.appendFailure) { + inner.push( + this.db.prepare( + "INSERT INTO fedify_kv (key, value) VALUES ('injected', NULL)", + ), + ); + } + const results = await this.db.batch(inner); + if (atomic && this.options.guardCounts != null) { + const [matched, cleaned] = this.options.guardCounts; + results[results.length - 2].meta.changes = matched; + results[results.length - 1].meta.changes = cleaned; + } + return results; } get ddlBatches(): number { @@ -117,8 +141,9 @@ async function collect( describe("WorkersD1KvStore", () => { it("rejects an invalid table name", () => { for (const tableName of ["foo; DROP", "sqlite_kv", "SQLITE_kv", "_cf_KV"]) { - expect(() => new WorkersD1KvStore(env.D1, { tableName })) - .toThrow(TypeError); + expect(() => new WorkersD1KvStore(env.D1, { tableName })).toThrow( + TypeError, + ); } }); @@ -128,7 +153,9 @@ describe("WorkersD1KvStore", () => { await store.initialize(); const row = await env.D1.prepare( "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", - ).bind("init_test").first<{ name: string }>(); + ) + .bind("init_test") + .first<{ name: string }>(); expect(row?.name).toBe("init_test"); }); @@ -155,7 +182,9 @@ describe("WorkersD1KvStore", () => { expect(await store.get(["ttl"])).toBe("value"); const row = await env.D1.prepare( "SELECT expires FROM fedify_kv WHERE key = ?", - ).bind(JSON.stringify(["ttl"])).first<{ expires: number }>(); + ) + .bind(JSON.stringify(["ttl"])) + .first<{ expires: number }>(); expect(row?.expires).toBeGreaterThan(Date.now() + 3_500_000); }); @@ -175,7 +204,9 @@ describe("WorkersD1KvStore", () => { await store.set(["other"], "value"); const row = await env.D1.prepare( "SELECT count(*) AS count FROM fedify_kv WHERE key = ?", - ).bind(JSON.stringify(["expired"])).first<{ count: number }>(); + ) + .bind(JSON.stringify(["expired"])) + .first<{ count: number }>(); expect(row?.count).toBe(0); }); @@ -192,7 +223,9 @@ describe("WorkersD1KvStore", () => { await store.set(["custom"], "value"); const row = await env.D1.prepare( "SELECT value FROM custom_kv WHERE key = ?", - ).bind(JSON.stringify(["custom"])).first<{ value: string }>(); + ) + .bind(JSON.stringify(["custom"])) + .first<{ value: string }>(); expect(row?.value).toBe('"value"'); }); @@ -225,8 +258,9 @@ describe("WorkersD1KvStore", () => { const store = new WorkersD1KvStore(env.D1); await store.set(["obj"], { a: 1, b: [1, 2, { c: null }] }); // Order of object members does not matter: - expect(await store.cas(["obj"], { b: [1, 2, { c: null }], a: 1 }, 1)) - .toBe(true); + expect(await store.cas(["obj"], { b: [1, 2, { c: null }], a: 1 }, 1)).toBe( + true, + ); expect(await store.get(["obj"])).toBe(1); await store.set(["arr"], [1, 2]); @@ -266,7 +300,9 @@ describe("WorkersD1KvStore", () => { expect(await store.cas(["ttl"], "a", "b")).toBe(true); const row = await env.D1.prepare( "SELECT expires FROM fedify_kv WHERE key = ?", - ).bind(JSON.stringify(["ttl"])).first<{ expires: number | null }>(); + ) + .bind(JSON.stringify(["ttl"])) + .first<{ expires: number | null }>(); expect(row?.expires).toBeNull(); }); @@ -340,7 +376,9 @@ describe("WorkersD1KvStore", () => { expect(spy.ddlBatches).toBe(1); const index = await env.D1.prepare( "SELECT name FROM sqlite_master WHERE type = 'index' AND name = ?", - ).bind("idx_fedify_kv_expires").first<{ name: string }>(); + ) + .bind("idx_fedify_kv_expires") + .first<{ name: string }>(); expect(index?.name).toBe("idx_fedify_kv_expires"); }); @@ -376,8 +414,15 @@ describe("WorkersD1KvStore", () => { await store.drop(); const row = await env.D1.prepare( "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", - ).bind("drop_test").first(); + ) + .bind("drop_test") + .first(); expect(row).toBeNull(); + expect( + await env.D1.prepare( + "SELECT name FROM sqlite_master WHERE name = 'drop_test_atomic'", + ).first(), + ).toBeNull(); // The store re-creates the table on its next use: expect(await store.get(["a"])).toBeUndefined(); }); @@ -397,18 +442,18 @@ describe("WorkersD1KvStore", () => { await store.initialize(); const expired = Date.now() - 1000; await env.D1.batch( - Array.from( - { length: 150 }, - (_, i) => - env.D1.prepare( - "INSERT INTO fedify_kv (key, value, expires) VALUES (?, ?, ?)", - ).bind(JSON.stringify(["expired", `${i}`]), "null", expired), + Array.from({ length: 150 }, (_, i) => + env.D1.prepare( + "INSERT INTO fedify_kv (key, value, expires) VALUES (?, ?, ?)", + ).bind(JSON.stringify(["expired", `${i}`]), "null", expired), ), ); const countExpired = async () => - (await env.D1.prepare( - "SELECT count(*) AS count FROM fedify_kv WHERE expires IS NOT NULL", - ).first<{ count: number }>())?.count; + ( + await env.D1.prepare( + "SELECT count(*) AS count FROM fedify_kv WHERE expires IS NOT NULL", + ).first<{ count: number }>() + )?.count; await store.set(["a"], 1); expect(await countExpired()).toBe(50); // cas() purges them as well: @@ -426,7 +471,8 @@ describe("WorkersD1KvStore", () => { const delayed = new WorkersD1KvStore( new D1Spy(env.D1, { async gate(queries) { - if (!queries.some((q) => q.trimStart().startsWith("UPDATE"))) return; + if (!queries.some((q) => /INSERT INTO "fedify_kv_atomic"/.test(q))) + return; // The entry must still be live when the CAS is sent: liveAtGate = (await store.get(key)) !== undefined; // Hold the CAS until D1's clock passes the expiration, and let @@ -452,7 +498,9 @@ describe("WorkersD1KvStore", () => { ).toBe(true); const row = await env.D1.prepare( "SELECT expires FROM fedify_kv WHERE key = ?", - ).bind(JSON.stringify(["ttl-update"])).first<{ expires: number }>(); + ) + .bind(JSON.stringify(["ttl-update"])) + .first<{ expires: number }>(); expect(row?.expires).toBeGreaterThan(Date.now() + 3_500_000); expect(row?.expires).toBeLessThanOrEqual(Date.now() + 3_600_000); }); @@ -507,11 +555,13 @@ describe("WorkersD1KvStore", () => { ); for (let i = 0; i < keys.length; i += 500) { await env.D1.batch( - keys.slice(i, i + 500).map((key) => - env.D1.prepare( - "INSERT INTO fedify_kv (key, value, expires) VALUES (?, ?, NULL)", - ).bind(JSON.stringify(key), JSON.stringify(key[1])), - ), + keys + .slice(i, i + 500) + .map((key) => + env.D1.prepare( + "INSERT INTO fedify_kv (key, value, expires) VALUES (?, ?, NULL)", + ).bind(JSON.stringify(key), JSON.stringify(key[1])), + ), ); } await store.set(["paged-other"], "other"); @@ -549,7 +599,10 @@ describe("WorkersD1KvStore", () => { const store = new WorkersD1KvStore(spy); // About 1.9 MB each, so 8 MiB fits five of them per query: const value = "x".repeat(1_900_000); - const keys = Array.from({ length: 6 }, (_, i) => ["large", `${i}`] as const); + const keys = Array.from( + { length: 6 }, + (_, i) => ["large", `${i}`] as const, + ); for (const key of keys) await store.set(key, value); const before = spy.queries.length; @@ -589,3 +642,520 @@ describe("WorkersD1KvStore", () => { expect(await listKeys(["1", "0"])).toEqual([["1", "0"]]); }); }); + +describe("WorkersD1KvStore atomic operations", () => { + it("compares the pre-mutation state and preserves physical rows on mismatch", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + expect( + await store.casMany([ + { key: ["a"], expectedValue: 1, newValue: 2 }, + { key: ["b"], expectedValue: 2, newValue: 1 }, + ]), + ).toBe(true); + await insertRaw(["expired"], "stale", 1); + const before = await env.D1.prepare( + "SELECT * FROM fedify_kv ORDER BY key", + ).all(); + expect( + await store.casMany([ + { key: ["a"], expectedValue: 2, newValue: 3 }, + { key: ["b"], expectedValue: 999, newValue: undefined }, + ]), + ).toBe(false); + expect( + (await env.D1.prepare("SELECT * FROM fedify_kv ORDER BY key").all()) + .results, + ).toEqual(before.results); + expect( + (await env.D1.prepare("SELECT * FROM fedify_kv_atomic").all()).results, + ).toEqual([]); + }); + + it("allows one competing reservation across separate instances", async () => { + const stores = Array.from( + { length: 10 }, + () => new WorkersD1KvStore(env.D1), + ); + await stores[0].set(["budget"], 1); + const results = await Promise.all( + stores.map((store, i) => + store.casMany([ + { + key: ["reservation", `${i}`], + expectedValue: undefined, + newValue: true, + }, + { key: ["budget"], expectedValue: 1, newValue: 0 }, + ]), + ), + ); + expect(results.filter(Boolean)).toHaveLength(1); + expect(await stores[0].get(["budget"])).toBe(0); + expect(await collect(stores[0].list(["reservation"]))).toHaveLength(1); + }); + + it("rolls back an actual later SQL error, including TTLs and guard cleanup", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["a"], 1, { ttl: Temporal.Duration.from({ hours: 1 }) }); + const before = await env.D1.prepare("SELECT * FROM fedify_kv").all(); + const failing = new WorkersD1KvStore( + new D1Spy(env.D1, { appendFailure: true }), + { initialized: true }, + ); + await expect( + failing.casMany([ + { key: ["a"], expectedValue: 1, newValue: 2 }, + { key: ["new"], expectedValue: undefined, newValue: 3 }, + ]), + ).rejects.toThrow(); + expect( + (await env.D1.prepare("SELECT * FROM fedify_kv").all()).results, + ).toEqual(before.results); + expect( + (await env.D1.prepare("SELECT * FROM fedify_kv_atomic").all()).results, + ).toEqual([]); + await expect( + failing.setMany([ + { key: ["a"], value: 4 }, + { key: ["new"], value: 5 }, + ]), + ).rejects.toThrow(); + expect( + (await env.D1.prepare("SELECT * FROM fedify_kv").all()).results, + ).toEqual(before.results); + expect(await store.cas(["a"], 1, 2)).toBe(true); + }); + + it("uses one database cutoff for comparison, replacement and cleanup", async () => { + const ordinary = new WorkersD1KvStore(env.D1); + await ordinary.initialize(); + const cutoff = Date.now() + 3_600_000; + await insertRaw(["expired-at-cutoff"], "old", cutoff - 1); + const spy = new D1Spy(env.D1, { + rewrite: (query) => + /INSERT INTO "fedify_kv_atomic"/.test(query) + ? query.replace( + "CAST(unixepoch('subsec') * 1000 AS INTEGER)", + `${cutoff}`, + ) + : query, + }); + const store = new WorkersD1KvStore(spy, { initialized: true }); + expect( + await store.casMany([ + { + key: ["expired-at-cutoff"], + expectedValue: undefined, + newValue: "new", + options: { ttl: mockDuration(2) }, + }, + { + key: ["second"], + expectedValue: undefined, + newValue: 1, + options: { ttl: mockDuration(2) }, + }, + ]), + ).toBe(true); + const { results } = await env.D1.prepare( + "SELECT expires FROM fedify_kv ORDER BY key", + ).all<{ expires: number }>(); + expect(results.map((row) => row.expires)).toEqual([ + cutoff + 2000, + cutoff + 2000, + ]); + await store.set(["immediate"], 1, { ttl: mockDuration(0.0005) }); + expect( + await env.D1.prepare("SELECT key FROM fedify_kv WHERE key = ?") + .bind(JSON.stringify(["immediate"])) + .first(), + ).toBeNull(); + }); + + it("rejects impossible guard result pairs as errors rather than mismatch", async () => { + await new WorkersD1KvStore(env.D1).initialize(); + for (const guardCounts of [ + [0, 0], + [1, 1], + [2, 0], + ] as const) { + const store = new WorkersD1KvStore(new D1Spy(env.D1, { guardCounts }), { + initialized: true, + }); + await expect(store.cas(["a"], undefined, undefined)).rejects.toThrow( + "integrity", + ); + await expect(store.set(["a"], 1)).rejects.toThrow("integrity"); + } + // Altered response metadata is an error after execution, not a rollback. + expect(await new WorkersD1KvStore(env.D1).get(["a"])).toBe(1); + }); + + it("keeps empty batches free of backend calls", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + expect(await store.getMany([])).toEqual([]); + await store.setMany([]); + await store.deleteMany([]); + expect(await store.casMany([])).toBe(true); + expect(spy.queries).toEqual([]); + }); + + it("rejects invalid batches before initialization or mutation", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + await expect(store.set(["a"], undefined)).rejects.toThrow(TypeError); + await expect(store.set(["a"], () => 1)).rejects.toThrow(TypeError); + await expect(store.cas(["a"], Symbol(), 1)).rejects.toThrow(TypeError); + await expect( + store.cas(["a"], 99, undefined, { ttl: mockDuration(1) }), + ).rejects.toThrow(TypeError); + await expect( + store.setMany([ + { key: ["a"], value: 1 }, + { key: ["a"], value: 2 }, + ]), + ).rejects.toThrow(TypeError); + await expect(store.setMany(new Array(2))).rejects.toThrow(TypeError); + await expect(store.getMany(new Array(2))).rejects.toThrow(TypeError); + await expect(store.deleteMany(new Array(2))).rejects.toThrow(TypeError); + await expect(store.casMany(new Array(2))).rejects.toThrow(TypeError); + await expect( + store.set(["a"], 1, { ttl: mockDuration(Infinity) }), + ).rejects.toThrow(RangeError); + await expect( + store.set(["a"], 1, { ttl: Temporal.Duration.from({ months: 1 }) }), + ).rejects.toThrow(RangeError); + const circular: unknown[] = []; + circular.push(circular); + await expect( + store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: circular }, + ]), + ).rejects.toThrow(); + expect(spy.queries).toEqual([]); + }); + + it("validates indexed contents even when array iterators disagree", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + const sparse = new Array(1); + Object.defineProperty(sparse, Symbol.iterator, { value: function* () {} }); + await expect(store.getMany(sparse)).rejects.toThrow(TypeError); + await expect(store.setMany(sparse)).rejects.toThrow(TypeError); + await expect(store.deleteMany(sparse)).rejects.toThrow(TypeError); + await expect(store.casMany(sparse)).rejects.toThrow(TypeError); + const empty: readonly [string][] = []; + Object.defineProperty(empty, Symbol.iterator, { + value: function* () { + for (let i = 0; i < 129; i++) yield [`${i}`]; + }, + }); + await store.deleteMany(empty); + expect(await store.getMany(empty)).toEqual([]); + const invalid = [42] as unknown as readonly [string]; + Object.defineProperty(invalid, Symbol.iterator, { + value: function* () { + yield "masked"; + }, + }); + await expect(store.set(invalid, 1)).rejects.toThrow(TypeError); + expect(spy.queries).toEqual([]); + }); + + it("bounds deletion work before visiting excessive or later keys", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + const excessive = Array.from({ length: 129 }, () => ["x"] as const); + Object.defineProperty(excessive, 0, { + get() { + throw new Error("visited"); + }, + }); + await expect(store.deleteMany(excessive)).rejects.toThrow(RangeError); + const keys = [ + ["x".repeat(1_000_000)], + ["y".repeat(1_000_000)], + ["late"], + ] as const; + Object.defineProperty(keys, 2, { + get() { + throw new Error("visited later"); + }, + }); + await expect(store.deleteMany(keys)).rejects.toThrow(RangeError); + expect(spy.queries).toEqual([]); + }); + + it("keeps ordinary reads and deletions usable for escaped keys", async () => { + const store = new WorkersD1KvStore(env.D1); + const key = ['"'.repeat(600_000)] as const; + await store.set(key, 1); + expect(await store.get(key)).toBe(1); + await expect(store.getMany([key])).rejects.toThrow(RangeError); + await expect(store.deleteMany([key])).rejects.toThrow(RangeError); + expect(await store.get(key)).toBe(1); + await store.delete(key); + expect(await store.get(key)).toBeUndefined(); + }); + + it("rejects encoded row, expected-value and escaped key-payload limits", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + const huge = "한".repeat(700_000); + await expect(store.set(["a"], huge)).rejects.toThrow(RangeError); + await expect(store.cas(["a"], huge, undefined)).rejects.toThrow(RangeError); + await expect( + store.set(["x".repeat(1_100_000)], "y".repeat(1_100_000)), + ).rejects.toThrow(RangeError); + const escaped = ['"'.repeat(600_000)] as const; + await expect(store.getMany([escaped])).rejects.toThrow(RangeError); + await expect(store.deleteMany([escaped])).rejects.toThrow(RangeError); + await expect( + store.setMany( + Array.from({ length: 10 }, (_, i) => ({ + key: [`${i}`], + value: "x".repeat(1_800_000), + })), + ), + ).rejects.toThrow(RangeError); + expect(spy.queries).toEqual([]); + }); + + it("rejects a snapshot over budget and expands duplicates after one read", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.setMany( + Array.from({ length: 5 }, (_, i) => ({ + key: [`${i}`], + value: "x".repeat(1_800_000), + })), + ); + await expect( + store.getMany(Array.from({ length: 5 }, (_, i) => [`${i}`] as const)), + ).rejects.toThrow(RangeError); + const spy = new D1Spy(env.D1); + const reader = new WorkersD1KvStore(spy, { initialized: true }); + const repeated = await reader.getMany( + Array.from({ length: 20 }, () => ["0"] as const), + ); + expect(repeated).toHaveLength(20); + expect(repeated.every((value) => value === repeated[0])).toBe(true); + expect(spy.queries).toHaveLength(1); + }); + + it("compares persisted JSON coercions and unusual member names", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["nan"], NaN); + expect(await store.cas(["nan"], null, 1)).toBe(true); + await store.set(["omitted"], { a: undefined }); + expect( + await store.casMany([ + { key: ["omitted"], expectedValue: {}, newValue: null }, + ]), + ).toBe(true); + const value = { "a.b": 1, "a[0]": 2, '\"\\\u0000': { "한😀": [1, null] } }; + await store.set(["object"], value); + expect( + await store.casMany([ + { key: ["object"], expectedValue: { ...value, "a.b": 9 }, newValue: 0 }, + ]), + ).toBe(false); + expect( + await store.cas( + ["object"], + { '\"\\\u0000': { "한😀": [1, null] }, "a[0]": 2, "a.b": 1 }, + 1, + ), + ).toBe(true); + await store.set(["number"], 1.2345678901234568e29); + expect(await store.cas(["number"], 1.234567890123457e29, 0)).toBe(false); + }); + + it("rejects incompatible auxiliary schemas and retries after repair", async () => { + for (const ddl of [ + "CREATE TABLE bad_atomic (singleton INTEGER PRIMARY KEY, cutoff INTEGER NOT NULL, matched INTEGER NOT NULL, extra INTEGER GENERATED ALWAYS AS (1) VIRTUAL)", + "CREATE TABLE bad_atomic (singleton INTEGER PRIMARY KEY, cutoff TEXT NOT NULL, matched INTEGER NOT NULL)", + ]) { + await env.D1.prepare(ddl).run(); + const store = new WorkersD1KvStore(env.D1, { tableName: "bad" }); + await expect(store.initialize()).rejects.toThrow("Incompatible"); + await env.D1.prepare("DROP TABLE bad_atomic").run(); + await store.set(["ok"], 1); + await store.drop(); + } + expect( + () => new WorkersD1KvStore(env.D1, { tableName: "bad_ATOMIC" }), + ).toThrow(TypeError); + }); + + it("supports backing table names that match snapshot CTEs", async () => { + for (const tableName of ["live", "cutoff", "budget"]) { + const store = new WorkersD1KvStore(env.D1, { tableName }); + await store.setMany([ + { key: ["a"], value: "first" }, + { key: ["b"], value: "second" }, + ]); + expect(await store.getMany([["b"], ["missing"], ["a"], ["b"]])).toEqual([ + "second", + undefined, + "first", + "second", + ]); + await store.drop(); + } + }); + + it.each(["json_each", "json_tree", "shadowed_functions"])( + "keeps JSON functions available with application table %s", + async (tableName) => { + const unrelated = ["json_each", "json_tree"].filter((name) => name !== tableName); + for (const name of unrelated) { + await env.D1.prepare(`CREATE TABLE "${name}" (unrelated TEXT)`).run(); + } + const store = new WorkersD1KvStore(env.D1, { tableName }); + await store.setMany([ + { key: ["a"], value: { first: 1, second: [2] } }, + { key: ["b"], value: 2 }, + ]); + expect(await store.getMany([["b"], ["missing"], ["a"], ["b"]])).toEqual([ + 2, undefined, { first: 1, second: [2] }, 2, + ]); + expect(await store.cas(["a"], { second: [2], first: 1 }, 3)).toBe(true); + expect(await store.casMany([ + { key: ["a"], expectedValue: 0, newValue: 4 }, + { key: ["b"], expectedValue: 2, newValue: 5 }, + ])).toBe(false); + await store.deleteMany([["a"]]); + expect(await store.getMany([["a"], ["b"]])).toEqual([undefined, 2]); + await store.drop(); + for (const name of unrelated) { + await env.D1.prepare(`DROP TABLE "${name}"`).run(); + } + }, + ); + + it("supports backing table names that match list CTEs", async () => { + for (const tableName of ["page", "sized"]) { + const store = new WorkersD1KvStore(env.D1, { tableName }); + await store.setMany([ + { key: ["a"], value: "first" }, + { key: ["a", "child"], value: "child" }, + { key: ["b"], value: "second" }, + ]); + const expected = [ + { key: ["a", "child"], value: "child" }, + { key: ["a"], value: "first" }, + { key: ["b"], value: "second" }, + ]; + expect(await Array.fromAsync(store.list())).toEqual(expected); + expect(await Array.fromAsync(store.list(["a"]))).toEqual(expected.slice(0, 2)); + await store.drop(); + } + }); + + it("uses binary key comparisons with a NOCASE column and binary primary index", async () => { + await env.D1.prepare(`CREATE TABLE mixed_collation ( + key TEXT COLLATE NOCASE NOT NULL, value TEXT NOT NULL, expires INTEGER, + PRIMARY KEY (key COLLATE BINARY) + )`).run(); + const store = new WorkersD1KvStore(env.D1, { tableName: "mixed_collation" }); + await store.setMany([ + { key: ["A"], value: "upper" }, + { key: ["a"], value: "lower" }, + ]); + expect(await store.get(["a"])).toBe("lower"); + expect(await store.getMany([["a"], ["A"], ["a"]])).toEqual(["lower", "upper", "lower"]); + expect(await store.cas(["a"], "upper", "wrong")).toBe(false); + expect(await store.casMany([ + { key: ["A"], expectedValue: "upper", newValue: "UPPER" }, + { key: ["a"], expectedValue: "lower", newValue: "LOWER" }, + ])).toBe(true); + expect(await store.getMany([["A"], ["a"]])).toEqual(["UPPER", "LOWER"]); + expect(await Array.fromAsync(store.list(["a"]))).toEqual([{ key: ["a"], value: "LOWER" }]); + await store.delete(["A"]); + expect(await store.getMany([["A"], ["a"]])).toEqual([undefined, "LOWER"]); + await store.set(["A"], "upper"); + await store.deleteMany([["a"]]); + expect(await store.getMany([["A"], ["a"]])).toEqual(["upper", undefined]); + await store.set(["a"], "lower"); + await insertRaw(["A"], "expired", 0, "mixed_collation"); + await store.set(["trigger"], true); + expect(await store.get(["a"])).toBe("lower"); + // Exercise ordering and the cursor beyond the 1,000-row page boundary. + for (let start = 0; start < 1002; start += 128) { + await store.setMany(Array.from({ length: Math.min(128, 1002 - start) }, (_, i) => ({ + key: ["pages", i % 2 ? "a" : "A", (start + i).toString().padStart(4, "0")], + value: start + i, + }))); + } + const rows = await Array.fromAsync(store.list(["pages"])); + expect(rows.map((row) => row.value)).toEqual([ + ...Array.from({ length: 501 }, (_, i) => i * 2), + ...Array.from({ length: 501 }, (_, i) => i * 2 + 1), + ]); + await store.drop(); + }); + + it("rejects nonbinary key primary indexes before batch writes", async () => { + for (const collation of ["NOCASE", "RTRIM"]) { + await env.D1.prepare(`CREATE TABLE key_collation ( + key TEXT PRIMARY KEY COLLATE ${collation} NOT NULL, + value TEXT NOT NULL, expires INTEGER + )`).run(); + const store = new WorkersD1KvStore(env.D1, { + tableName: "key_collation", + }); + await expect( + store.setMany([ + { key: ["A"], value: "upper" }, + { key: ["a"], value: "lower" }, + ]), + ).rejects.toThrow("Incompatible"); + const row = await env.D1.prepare( + "SELECT COUNT(*) AS count FROM key_collation", + ).first<{ count: number }>(); + expect(row?.count).toBe(0); + await env.D1.prepare("DROP TABLE key_collation").run(); + // Failed validation must not cache initialization success. + await store.setMany([ + { key: ["A"], value: "upper" }, + { key: ["a"], value: "lower" }, + ]); + expect(await store.getMany([["A"], ["a"]])).toEqual(["upper", "lower"]); + await store.drop(); + } + }); + + it("rejects colliding expiry indexes", async () => { + await env.D1.prepare("CREATE TABLE other (expires INTEGER)").run(); + await env.D1.prepare( + "CREATE INDEX idx_collision_expires ON other (expires)", + ).run(); + await expect( + new WorkersD1KvStore(env.D1, { tableName: "collision" }).initialize(), + ).rejects.toThrow("Incompatible"); + }); + + it("coexists atomically with direct base-adapter SQL writes", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.setMany([ + { key: ["a"], value: 0 }, + { key: ["b"], value: 0 }, + ]); + const [committed] = await Promise.all([ + store.casMany([ + { key: ["a"], expectedValue: 0, newValue: 1 }, + { key: ["b"], expectedValue: 0, newValue: 1 }, + ]), + env.D1.prepare("UPDATE fedify_kv SET value = '2' WHERE key = ?") + .bind(JSON.stringify(["a"])) + .run(), + ]); + expect(await store.getMany([["a"], ["b"]])).toEqual([2, committed ? 1 : 0]); + }); +}); diff --git a/packages/cfworkers/test/typecheck/workers-types.ts b/packages/cfworkers/test/typecheck/workers-types.ts index 87f3a4fd2..471773c93 100644 --- a/packages/cfworkers/test/typecheck/workers-types.ts +++ b/packages/cfworkers/test/typecheck/workers-types.ts @@ -4,7 +4,11 @@ import type { KVNamespace, Queue, } from "@cloudflare/workers-types"; -import type { KvStore, MessageQueue } from "@fedify/fedify/federation"; +import type { + AtomicKvStore, + KvStore, + MessageQueue, +} from "@fedify/fedify/federation"; import { WorkersD1KvStore, WorkersKvStore, @@ -21,8 +25,8 @@ declare const kv: KVNamespace; declare const queue: Queue; export const store: KvStore = new WorkersKvStore(kv); -export const d1Store: KvStore = new WorkersD1KvStore(d1); -export const d1SessionStore: KvStore = new WorkersD1KvStore(d1Session, { +export const d1Store: AtomicKvStore = new WorkersD1KvStore(d1); +export const d1SessionStore: AtomicKvStore = new WorkersD1KvStore(d1Session, { tableName: "fedify_kv", initialized: true, }); diff --git a/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts b/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts index 45927c269..c56c59d9d 100644 --- a/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts +++ b/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts @@ -1,4 +1,4 @@ -import type { KvStore } from "@fedify/fedify/federation"; +import type { AtomicKvStore } from "@fedify/fedify/federation"; import { WorkersD1KvStore } from "../../dist/mod.js"; interface GeneratedD1Meta { @@ -66,8 +66,8 @@ interface GeneratedD1Database { declare const generatedD1: GeneratedD1Database; -export const store: KvStore = new WorkersD1KvStore(generatedD1); -export const sessionStore: KvStore = new WorkersD1KvStore( +export const store: AtomicKvStore = new WorkersD1KvStore(generatedD1); +export const sessionStore: AtomicKvStore = new WorkersD1KvStore( generatedD1.withSession("first-primary"), { initialized: true }, ); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 0e1e1d553..51bfd1b9e 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -397,6 +397,8 @@ export class PostgresKvStore implements KvStore { WHERE c.oid = (SELECT typrelid FROM pg_type WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) `; + // Index-column DDL can add version-dependent wrappers. Compare the + // expression itself after restricting the index to one expression. const indexed = async () => { const [result] = await sql` SELECT EXISTS ( @@ -409,7 +411,7 @@ 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 ( + AND pg_get_expr(i.indexprs, i.indrelid) IN ( ${"timezone('UTC'::text, (timezone('UTC'::text, created) + ttl))"}, ${"(((created AT TIME ZONE 'UTC'::text) + ttl) AT TIME ZONE 'UTC'::text)"} ) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index bf04f92ed..820f1c81f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1182,9 +1182,15 @@ importers: '@cloudflare/workers-types': specifier: 'catalog:' version: 4.20260511.1 + '@fedify/testing': + specifier: workspace:^ + version: link:../testing semver: specifier: ^7.7.2 version: 7.8.0 + temporal-polyfill: + specifier: 'catalog:' + version: 1.0.1 tsdown: specifier: 'catalog:' version: 0.22.0(tsx@4.21.0)(typescript@6.0.3)(unrun@0.2.34(@emnapi/core@1.11.3)(@emnapi/runtime@1.11.3))