From 8aead3bbaa6e10d0285e1f8460aa353959759c7d Mon Sep 17 00:00:00 2001 From: BeiyanYunyi <14120445+BeiyanYunyi@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:10:02 -0500 Subject: [PATCH 1/2] Add WorkersD1KvStore backed by Cloudflare D1 Fedify writes to its key-value store about as often as it reads from it, and idempotency records and signature nonces need strong consistency. Workers KV is tuned for read-heavy workloads, is eventually consistent, and cannot support cas(), so WorkersKvStore is a poor fit for Fedify on Cloudflare Workers. This commit includes the following changes: - Added WorkersD1KvStore to @fedify/cfworkers, with cas() implemented as a single conditional SQL statement per call, since D1 has no interactive transactions. Values are compared structurally with json_tree(), ignoring the order of object members. - Added WorkersD1DatabaseLike and WorkersD1PreparedStatementLike so that both D1Database and D1DatabaseSession bindings are accepted. - Added binding-based tests and type checks for the new store. - Updated the documentation and the Fedify skill to recommend WorkersD1KvStore over WorkersKvStore on Cloudflare Workers. https://github.com/fedify-dev/fedify/issues/1135 Assisted-by: Claude Code:claude-opus-5-5 --- CHANGES.md | 15 + changes.d/cfworkers/d1-kv-store.md | 14 + claude-plugin/skills/fedify/SKILL.md | 5 + docs/manual/deploy.md | 40 ++- docs/manual/kv.md | 127 ++++++- docs/manual/opentelemetry.md | 2 + docs/manual/tasks.md | 7 +- packages/cfworkers/README.md | 46 ++- packages/cfworkers/src/mod.ts | 309 +++++++++++++++++- packages/cfworkers/test/d1.test.ts | 248 ++++++++++++++ packages/cfworkers/test/env.d.ts | 6 +- .../cfworkers/test/typecheck/workers-types.ts | 24 +- packages/cfworkers/wrangler.jsonc | 7 + 13 files changed, 805 insertions(+), 45 deletions(-) create mode 100644 changes.d/cfworkers/d1-kv-store.md create mode 100644 packages/cfworkers/test/d1.test.ts diff --git a/CHANGES.md b/CHANGES.md index 0a038110a..b09125853 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -107,6 +107,21 @@ To be released. [#1353]: https://github.com/fedify-dev/fedify/pull/1353 [#1359]: https://github.com/fedify-dev/fedify/pull/1359 +### @fedify/cfworkers + + - Added `WorkersD1KvStore`, a `KvStore` implementation backed by + Cloudflare D1. 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, which Workers KV does not + provide. `WorkersD1KvStore` is strongly consistent and supports `cas()`, + so features that need atomic updates no longer fall back to non-atomic + code paths. It is now the recommended `KvStore` for Cloudflare Workers; + consider migrating from `WorkersKvStore`. + [[#1135], [#1360] by Beiyan Yunyi\] + +[#1135]: https://github.com/fedify-dev/fedify/issues/1135 +[#1360]: https://github.com/fedify-dev/fedify/pull/1360 + ### @fedify/cli - Added a `fedify-cli` Agent Skill for command-line ActivityPub diagnostics diff --git a/changes.d/cfworkers/d1-kv-store.md b/changes.d/cfworkers/d1-kv-store.md new file mode 100644 index 000000000..2cc0729c7 --- /dev/null +++ b/changes.d/cfworkers/d1-kv-store.md @@ -0,0 +1,14 @@ +--- +links: + '#1135': https://github.com/fedify-dev/fedify/issues/1135 + '#1360': https://github.com/fedify-dev/fedify/pull/1360 +--- + - Added `WorkersD1KvStore`, a `KvStore` implementation backed by + Cloudflare D1. 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, which Workers KV does not + provide. `WorkersD1KvStore` is strongly consistent and supports `cas()`, + so features that need atomic updates no longer fall back to non-atomic + code paths. It is now the recommended `KvStore` for Cloudflare Workers; + consider migrating from `WorkersKvStore`. + [[#1135], [#1360] by Beiyan Yunyi] diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index 26bd6c7df..8e89dd177 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -385,6 +385,11 @@ Do not choose *@fedify/denokv* for new projects: it is deprecated since Fedify 2.5.0 and will be removed in Fedify 3.0.0. It remains supported for 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 +write-heavy and needs strong consistency, and only `WorkersD1KvStore` +supports `cas()`. + For `PostgresKvStore`, existing populated tables and externally managed DDL (`initialized: true`) need the [expiry expression index] to accelerate cleanup. Run concurrent index creation outside a transaction, followed by `ANALYZE`. diff --git a/docs/manual/deploy.md b/docs/manual/deploy.md index 4c2c92223..dbc28f00d 100644 --- a/docs/manual/deploy.md +++ b/docs/manual/deploy.md @@ -171,16 +171,16 @@ The in-memory defaults are for development only. in-flight activity; neither survives horizontal scaling. Pick a persistent backend before you take traffic: -| Backend | KV store | Message queue | When to choose | -| ---------------------------------- | ----------------- | ---------------------- | ------------------------------------- | -| PostgreSQL | `PostgresKvStore` | `PostgresMessageQueue` | You already run Postgres for app data | -| Redis | `RedisKvStore` | `RedisMessageQueue` | Dedicated cache/queue infrastructure | -| MySQL/MariaDB | `MysqlKvStore` | `MysqlMessageQueue` | You already run MySQL or MariaDB | -| SQLite | `SqliteKvStore` | `SqliteMessageQueue` | Single-node/embedded deployments | -| RabbitMQ | — | `AmqpMessageQueue` | Existing AMQP infrastructure | -| Deno KV (deprecated) | `DenoKvStore` | `DenoKvMessageQueue` | Existing Deno Deploy applications | -| Cloudflare KV + Queues | `WorkersKvStore` | `WorkersMessageQueue` | Cloudflare Workers | -| Netlify Database + Async Workloads | `PostgresKvStore` | `NetlifyMessageQueue` | Netlify Functions | +| Backend | KV store | Message queue | When to choose | +| ---------------------------------- | ------------------ | ---------------------- | ------------------------------------- | +| PostgreSQL | `PostgresKvStore` | `PostgresMessageQueue` | You already run Postgres for app data | +| Redis | `RedisKvStore` | `RedisMessageQueue` | Dedicated cache/queue infrastructure | +| MySQL/MariaDB | `MysqlKvStore` | `MysqlMessageQueue` | You already run MySQL or MariaDB | +| SQLite | `SqliteKvStore` | `SqliteMessageQueue` | Single-node/embedded deployments | +| RabbitMQ | — | `AmqpMessageQueue` | Existing AMQP infrastructure | +| Deno KV (deprecated) | `DenoKvStore` | `DenoKvMessageQueue` | Existing Deno Deploy applications | +| Cloudflare D1 + Queues | `WorkersD1KvStore` | `WorkersMessageQueue` | Cloudflare Workers | +| Netlify Database + Async Workloads | `PostgresKvStore` | `NetlifyMessageQueue` | Netlify Functions | See the [*Key–value store*](./kv.md) and [*Message queue*](./mq.md) chapters for setup details and trade-offs between backends (connection pooling, @@ -919,8 +919,12 @@ and no mutable global state between invocations. Platform services—KV, Queues, R2, D1—are exposed through the `env` parameter of the request handler rather than as ambient imports. Fedify accommodates this through the [builder pattern](./federation.md#builder-pattern-for-structuring) and -the [`@fedify/cfworkers`] package, which provides `WorkersKvStore` and -`WorkersMessageQueue`. +the [`@fedify/cfworkers`] package, which provides `WorkersD1KvStore`, +`WorkersKvStore`, and `WorkersMessageQueue`. Prefer `WorkersD1KvStore` +over `WorkersKvStore`: Fedify's key–value workload is write-heavy and needs +strong consistency, which D1 provides and Workers KV does not. See the +[*`WorkersD1KvStore`*](./kv.md#workersd1kvstore-cloudflare-workers-only) +section for details. #### Node.js compatibility @@ -944,13 +948,13 @@ dispatchers and build the `Federation` object per request: ~~~~ typescript twoslash // @noErrors: 2345 type Env = { - KV_NAMESPACE: KVNamespace; + DB: D1Database; QUEUE: Queue; }; import { Person } from "@fedify/vocab"; // ---cut-before--- import { createFederationBuilder } from "@fedify/fedify"; -import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; +import { WorkersD1KvStore, WorkersMessageQueue } from "@fedify/cfworkers"; const builder = createFederationBuilder(); @@ -964,7 +968,7 @@ builder.setActorDispatcher("/users/{identifier}", async (ctx, identifier) => { export default { async fetch(request: Request, env: Env): Promise { const federation = await builder.build({ - kv: new WorkersKvStore(env.KV_NAMESPACE), + kv: new WorkersD1KvStore(env.DB), queue: new WorkersMessageQueue(env.QUEUE), }); return federation.fetch(request, { contextData: env }); @@ -982,10 +986,10 @@ manually: ~~~~ typescript twoslash // @noErrors: 2345 import { createFederationBuilder, type Message } from "@fedify/fedify"; -import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; +import { WorkersD1KvStore, WorkersMessageQueue } from "@fedify/cfworkers"; type Env = { - KV_NAMESPACE: KVNamespace; + DB: D1Database; QUEUE: Queue; }; @@ -996,7 +1000,7 @@ export default { async queue(batch: MessageBatch, env: Env): Promise { const federation = await builder.build({ - kv: new WorkersKvStore(env.KV_NAMESPACE), + kv: new WorkersD1KvStore(env.DB), queue: new WorkersMessageQueue(env.QUEUE), }); diff --git a/docs/manual/kv.md b/docs/manual/kv.md index af804ab6a..cebffc662 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -497,7 +497,7 @@ const federation = createFederation({ > `@fedify/denokv` throughout the 2.x series, but the package will be removed > in Fedify 3.0.0. Use [`SqliteKvStore`] for local or single-node deployments, > or [`PostgresKvStore`], [`RedisKvStore`], [`MysqlKvStore`], or -> [`WorkersKvStore`] for production. See [#1336] for the removal plan. +> [`WorkersD1KvStore`] for production. See [#1336] for the removal plan. To use the [`DenoKvStore`], you need to install the *@fedify/denokv* package first: @@ -536,12 +536,116 @@ const federation = createFederation({ ~~~~ [deno-announcement]: https://deno.com/blog/cloudflare -[`WorkersKvStore`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersKvStore +[`WorkersD1KvStore`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersD1KvStore [#1336]: https://github.com/fedify-dev/fedify/issues/1336 [`DenoKvStore`]: https://jsr.io/@fedify/denokv/doc/~/DenoKvStore [Deno]: https://deno.com/ [`Deno.openKv()`]: https://docs.deno.com/api/deno/~/Deno.openKv +### `WorkersD1KvStore` (Cloudflare Workers only) + +*This API is available since Fedify 2.5.0.* + +To use the [`WorkersD1KvStore`], you need to install the *@fedify/cfworkers* +package first: + +::: code-group + +~~~~ bash [Deno] +deno add jsr:@fedify/cfworkers +~~~~ + +~~~~ bash [npm] +npm add @fedify/cfworkers +~~~~ + +~~~~ bash [pnpm] +pnpm add @fedify/cfworkers +~~~~ + +~~~~ bash [Yarn] +yarn add @fedify/cfworkers +~~~~ + +~~~~ bash [Bun] +bun add @fedify/cfworkers +~~~~ + +::: + +[`WorkersD1KvStore`] is a key–value store implementation for +[Cloudflare Workers] that uses [Cloudflare D1], Cloudflare's serverless SQL +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. + +Best for +: Production use in Cloudflare Workers environments. + +Pros +: Persistent storage, strong consistency, supports `~KvStore.cas()`. + +Cons +: Only available in Cloudflare Workers runtime. + +~~~~ typescript twoslash +// @noErrors: 2345 +import type { FederationBuilder } from "@fedify/fedify"; +const builder = undefined as unknown as FederationBuilder; +// ---cut-before--- +import type { Federation } from "@fedify/fedify"; +import { WorkersD1KvStore } from "@fedify/cfworkers"; + +export default { + async fetch(request, env, ctx) { + const federation: Federation = await builder.build({ + kv: new WorkersD1KvStore(env.DB), + }); + return await federation.fetch(request, { contextData: undefined }); + } +} satisfies ExportedHandler<{ DB: D1Database }>; +~~~~ + +The store keeps its entries in a table named `fedify_kv` by default, which +you can change with the `tableName` option. It creates the table on its +first use in each isolate. If you would rather manage the schema with +[D1 migrations], create the table in a migration and pass +`initialized: true` to skip the check: + +~~~~ sql +CREATE TABLE IF NOT EXISTS fedify_kv ( + key TEXT PRIMARY KEY NOT NULL, + value TEXT NOT NULL, + expires INTEGER +); +CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); +~~~~ + +~~~~ typescript twoslash +import { WorkersD1KvStore } from "@fedify/cfworkers"; +const env = undefined as unknown as { DB: D1Database }; +// ---cut-before--- +const kv = new WorkersD1KvStore(env.DB, { initialized: true }); +~~~~ + +Expired entries are never returned, and they are purged from the table +whenever a value is set. + +> [!NOTE] +> Since your `D1Database` is not bound to a global variable, but rather +> passed as an argument to the `fetch()` method, you need to instantiate +> your `Federation` object inside the `fetch()` method, rather than the top +> level. See the [*Builder pattern for structuring* +> section](./federation.md#builder-pattern-for-structuring) for details. + +[Cloudflare Workers]: https://workers.cloudflare.com/ +[Cloudflare D1]: https://developers.cloudflare.com/d1/ +[Cloudflare Workers KV]: https://developers.cloudflare.com/kv/ +[D1 migrations]: https://developers.cloudflare.com/d1/reference/migrations/ + ### `WorkersKvStore` (Cloudflare Workers only) *This API is available since Fedify 1.6.0.* @@ -575,17 +679,23 @@ bun add @fedify/cfworkers [`WorkersKvStore`] is a key–value store implementation for [Cloudflare Workers] that uses Cloudflare's built-in [Cloudflare Workers KV] API. It provides -persistent storage and good performance for Cloudflare Workers environments. -It's suitable for production use in Cloudflare Workers applications. +persistent storage with fast reads for Cloudflare Workers environments. + +However, Workers KV is tuned for read-heavy workloads and is eventually +consistent, while Fedify writes to its key–value store about as often as it +reads from it. It also does not support `~KvStore.cas()`, so some features +fall back to non-atomic code paths. For new Cloudflare Workers deployments, +prefer [`WorkersD1KvStore`]. Best for -: Production use in Cloudflare Workers environments. +: Existing Cloudflare Workers applications that already use Workers KV. Pros -: Persistent storage, good performance, easy to set up. +: Persistent storage, fast reads, easy to set up. Cons -: Only available in Cloudflare Workers runtime. +: Only available in Cloudflare Workers runtime, eventually consistent, + priced for read-heavy workloads, and does not support `~KvStore.cas()`. ~~~~ typescript twoslash // @noErrors: 2345 @@ -616,8 +726,7 @@ export default { > object. See the [*Builder pattern for structuring* > section](./federation.md#builder-pattern-for-structuring) for details. -[Cloudflare Workers]: https://workers.cloudflare.com/ -[Cloudflare Workers KV]: https://developers.cloudflare.com/kv/ +[`WorkersKvStore`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersKvStore ### `NetlifyBlobsKvStore` diff --git a/docs/manual/opentelemetry.md b/docs/manual/opentelemetry.md index ddc9aa522..4dfea6098 100644 --- a/docs/manual/opentelemetry.md +++ b/docs/manual/opentelemetry.md @@ -1385,4 +1385,6 @@ The following `KvStore` implementations support the required operations: - `PostgresKvStore` from *@fedify/postgres* (supports `list()`) - `SqliteKvStore` from *@fedify/sqlite* (supports `list()`) - `DenoKvStore` from *@fedify/denokv* (supports both `list()` and `cas()`) + - `WorkersD1KvStore` from *@fedify/cfworkers* (supports both `list()` and + `cas()`) - `WorkersKvStore` from *@fedify/cfworkers* (supports `list()`) diff --git a/docs/manual/tasks.md b/docs/manual/tasks.md index 5f54ce39a..c1eb6e3a3 100644 --- a/docs/manual/tasks.md +++ b/docs/manual/tasks.md @@ -321,9 +321,10 @@ How the key is resolved depends on the queue and the key–value store: ~~~~ Among the first-party adapters, the in-memory, SQLite, MySQL, -PostgreSQL, Redis, Netlify Blobs, and Deno KV key–value stores implement -`~KvStore.cas`; Cloudflare Workers KV does not, so those deployments take the -`taskDeduplicationFallback` branch until per-adapter follow-ups add it. +PostgreSQL, Redis, Netlify Blobs, Deno KV, and Cloudflare D1 key–value stores +implement `~KvStore.cas`; Cloudflare Workers KV does not, so those deployments +take the `taskDeduplicationFallback` branch. Use `WorkersD1KvStore` on +Cloudflare Workers to get atomic deduplication. For `~Context.enqueueTaskMany()`, a single `deduplicationKey` applies to the whole batch: the batch enqueues as a unit or is skipped as a unit, never diff --git a/packages/cfworkers/README.md b/packages/cfworkers/README.md index fcf566d32..ccfcdadc9 100644 --- a/packages/cfworkers/README.md +++ b/packages/cfworkers/README.md @@ -12,17 +12,18 @@ This package provides [Fedify]'s [`KvStore`] and [`MessageQueue`] implementations for [Cloudflare Workers]: + - [`WorkersD1KvStore`] - [`WorkersKvStore`] - [`WorkersMessageQueue`] ~~~~ typescript import { createFederation, type Message } from "@fedify/fedify"; -import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; +import { WorkersD1KvStore, WorkersMessageQueue } from "@fedify/cfworkers"; export default { async fetch(request, env, ctx) { const federation = createFederation({ - kv: new WorkersKvStore(env.KV_BINDING), + kv: new WorkersD1KvStore(env.DB_BINDING), queue: new WorkersMessageQueue(env.QUEUE_BINDING), // ... other options }); @@ -32,7 +33,7 @@ export default { async queue(batch, env, ctx) { const federation = createFederation({ - kv: new WorkersKvStore(env.KV_BINDING), + kv: new WorkersD1KvStore(env.DB_BINDING), queue: new WorkersMessageQueue(env.QUEUE_BINDING), // ... other options }); @@ -45,7 +46,7 @@ export default { } } } satisfies ExportedHandler<{ - KV_BINDING: KVNamespace; + DB_BINDING: D1Database; QUEUE_BINDING: Queue; }>; ~~~~ @@ -60,17 +61,48 @@ export default { [`KvStore`]: https://jsr.io/@fedify/fedify/doc/federation/~/KvStore [`MessageQueue`]: https://jsr.io/@fedify/fedify/doc/federation/~/MessageQueue [Cloudflare Workers]: https://workers.cloudflare.com/ +[`WorkersD1KvStore`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersD1KvStore [`WorkersKvStore`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersKvStore [`WorkersMessageQueue`]: https://jsr.io/@fedify/cfworkers/doc/~/WorkersMessageQueue +`WorkersD1KvStore` +------------------ + +*This API is available since Fedify 2.5.0.* + +`WorkersD1KvStore` is a key–value store implementation for +[Cloudflare Workers] that uses [Cloudflare D1]. D1 is strongly consistent +and suits Fedify's balanced read and write workload, and `WorkersD1KvStore` +supports the atomic `cas()` operation. It is the recommended key–value store +for Cloudflare Workers applications. + +The store creates its table (`fedify_kv` by default, configurable with +the `tableName` option) on its first use. If you manage the schema with +[D1 migrations] instead, create the table in a migration and pass +`initialized: true`: + +~~~~ sql +CREATE TABLE IF NOT EXISTS fedify_kv ( + key TEXT PRIMARY KEY NOT NULL, + value TEXT NOT NULL, + expires INTEGER +); +CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); +~~~~ + +[Cloudflare D1]: https://developers.cloudflare.com/d1/ +[D1 migrations]: https://developers.cloudflare.com/d1/reference/migrations/ + + `WorkersKvStore` ---------------- `WorkersKvStore` is a key–value store implementation for [Cloudflare Workers] that uses Cloudflare's built-in [Cloudflare Workers KV] API. It provides -persistent storage and good performance for Cloudflare Workers environments. -It's suitable for production use in Cloudflare Workers applications. +persistent storage with fast reads. However, Workers KV is tuned for +read-heavy workloads and is eventually consistent, and `WorkersKvStore` does +not support `cas()`. Prefer `WorkersD1KvStore` for new applications. [Cloudflare Workers KV]: https://developers.cloudflare.com/kv/ @@ -85,7 +117,7 @@ Cloudflare Workers environments. It requires a Cloudflare Queues setup and management. > [!NOTE] -> Since your `KVNamespace` and `Queue` are not bound to global variables, but +> Since your `D1Database` and `Queue` are not bound to global variables, but > rather passed as arguments to the `fetch()` and `queue()` methods, you need > to instantiate your `Federation` object inside these methods, rather than at > the top level. diff --git a/packages/cfworkers/src/mod.ts b/packages/cfworkers/src/mod.ts index 4e5a5416b..533bc0f7b 100644 --- a/packages/cfworkers/src/mod.ts +++ b/packages/cfworkers/src/mod.ts @@ -2,8 +2,9 @@ * `KvStore` & `MessageQueue` adapters for Cloudflare Workers * ========================================================== * - * This package provides `KvStore` and `MessageQueue` implementations that use - * Cloudflare Workers' KV and Queues bindings, respectively. + * This package provides `KvStore` implementations that use Cloudflare Workers' + * KV and D1 bindings, and a `MessageQueue` implementation that uses + * Cloudflare Queues binding. * * @module * @since 1.9.0 @@ -248,6 +249,310 @@ export class WorkersKvStore implements KvStore { } } +/** + * Minimal Cloudflare D1 prepared statement shape used by this package. + * Compatible with both `@cloudflare/workers-types` and `wrangler types` + * generated declarations. + * @since 2.5.0 + */ +export interface WorkersD1PreparedStatementLike { + bind(...values: unknown[]): WorkersD1PreparedStatementLike; + first>(): Promise; + all>(): Promise<{ readonly results: T[] }>; + run(): Promise<{ readonly meta: { readonly changes: number } }>; +} + +/** + * Minimal Cloudflare D1 binding shape used by this package. Compatible with + * both `D1Database` and `D1DatabaseSession`, as declared by + * `@cloudflare/workers-types` and `wrangler types`. + * @since 2.5.0 + */ +export interface WorkersD1DatabaseLike { + prepare(query: string): WorkersD1PreparedStatementLike; + batch(statements: WorkersD1PreparedStatementLike[]): Promise; +} + +/** + * Options for {@link WorkersD1KvStore}. + * @since 2.5.0 + */ +export interface WorkersD1KvStoreOptions { + /** + * The table name to use for the key–value store. + * Only letters, digits, and underscores are allowed. + * `"fedify_kv"` by default. + * @default `"fedify_kv"` + */ + readonly tableName?: string; + + /** + * Whether the table has already been created, for example, by a D1 + * migration. If `false`, the store creates the table on its first use. + * `false` by default. + * @default `false` + */ + readonly initialized?: boolean; +} + +/** + * Implementation of the {@link KvStore} interface for Cloudflare D1 binding. + * + * Unlike {@link WorkersKvStore}, this store is strongly consistent and is + * priced for balanced read and write workloads, which suits Fedify's 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. + * + * The store uses a table with the following schema, which it creates on + * its first use unless the `initialized` option is set: + * + * ```sql + * CREATE TABLE IF NOT EXISTS fedify_kv ( + * key TEXT PRIMARY KEY NOT NULL, + * value TEXT NOT NULL, + * expires INTEGER + * ); + * CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); + * ``` + * + * @example + * ```typescript ignore + * import { createFederation } from "@fedify/fedify"; + * import { WorkersD1KvStore } from "@fedify/cfworkers"; + * + * export default { + * async fetch(request, env, ctx) { + * const federation = createFederation({ + * kv: new WorkersD1KvStore(env.DB), + * // ... other options + * }); + * return federation.fetch(request, { contextData: env }); + * }, + * } satisfies ExportedHandler<{ DB: D1Database }>; + * ``` + * @since 2.5.0 + */ +export class WorkersD1KvStore implements KvStore { + static readonly #defaultTableName = "fedify_kv"; + static readonly #tableNameRegex = /^[A-Za-z_][A-Za-z0-9_]{0,63}$/; + readonly #db: WorkersD1DatabaseLike; + readonly #tableName: string; + #initialized: Promise | null; + + /** + * Constructs a new {@link WorkersD1KvStore} with the given D1 binding. + * @param db The Cloudflare D1 binding (or a D1 session). + * @param options Options for the key–value store. + * @throws {TypeError} If the table name is invalid. + */ + constructor( + db: WorkersD1DatabaseLike, + options: WorkersD1KvStoreOptions = {}, + ) { + this.#db = db; + this.#tableName = options.tableName ?? + WorkersD1KvStore.#defaultTableName; + if (!WorkersD1KvStore.#tableNameRegex.test(this.#tableName)) { + throw new TypeError( + `Invalid table name for the key–value store: ${this.#tableName}`, + ); + } + this.#initialized = options.initialized ? Promise.resolve() : null; + } + + /** + * 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 + * automatically on the first use of the store 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 "${this.#tableName}_expires" + ON "${this.#tableName}" (expires) + `), + ]).then(() => undefined, (error) => { + // Let a later call retry the initialization: + this.#initialized = null; + throw error; + }); + return this.#initialized; + } + + #encodeKey(key: KvKey): string { + return JSON.stringify(key); + } + + #getExpires(now: number, options?: KvStoreSetOptions): number | null { + return options?.ttl == null + ? null + : now + options.ttl.total("milliseconds"); + } + + /** + * {@inheritDoc KvStore.get} + */ + async get(key: KvKey): Promise { + await this.initialize(); + const row = await this.#db.prepare(` + SELECT value FROM "${this.#tableName}" + WHERE key = ? AND (expires IS NULL OR expires > ?) + `).bind(this.#encodeKey(key), Date.now()).first<{ value: string }>(); + return row == null ? undefined : JSON.parse(row.value) as T; + } + + /** + * {@inheritDoc KvStore.set} + */ + async set( + key: KvKey, + value: unknown, + options?: KvStoreSetOptions, + ): Promise { + await this.initialize(); + const now = Date.now(); + // Purge expired entries in the same round trip: + await this.#db.batch([ + this.#db.prepare(` + DELETE FROM "${this.#tableName}" + WHERE expires IS NOT NULL AND expires <= ? + `).bind(now), + this.#db.prepare(` + INSERT INTO "${this.#tableName}" (key, value, expires) + VALUES (?, ?, ?) + ON CONFLICT (key) DO UPDATE SET + value = excluded.value, + expires = excluded.expires + `).bind( + this.#encodeKey(key), + JSON.stringify(value), + this.#getExpires(now, options), + ), + ]); + } + + /** + * {@inheritDoc KvStore.delete} + */ + async delete(key: KvKey): Promise { + await this.initialize(); + await this.#db.prepare(`DELETE FROM "${this.#tableName}" WHERE key = ?`) + .bind(this.#encodeKey(key)) + .run(); + } + + /** + * {@inheritDoc KvStore.cas} + */ + async cas( + key: KvKey, + expectedValue: unknown, + newValue: unknown, + options?: KvStoreSetOptions, + ): Promise { + await this.initialize(); + const encodedKey = this.#encodeKey(key); + const now = Date.now(); + + 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: + const { meta } = await this.#db.prepare(` + INSERT INTO "${this.#tableName}" (key, value, expires) + VALUES (?, ?, ?) + ON CONFLICT (key) DO UPDATE SET + value = excluded.value, + expires = excluded.expires + WHERE "${this.#tableName}".expires IS NOT NULL + AND "${this.#tableName}".expires <= ? + `).bind( + encodedKey, + JSON.stringify(newValue), + this.#getExpires(now, options), + now, + ).run(); + return meta.changes > 0; + } + + // 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: + const expectedJson = JSON.stringify(expectedValue); + // The column must be qualified, since json_tree() has a value column too: + const current = `"${this.#tableName}".value`; + const matches = ` + key = ?1 AND (expires IS NULL OR expires > ?2) + AND NOT EXISTS ( + SELECT fullkey, type, atom FROM json_tree(${current}) + EXCEPT SELECT fullkey, type, atom FROM json_tree(?3) + ) + AND NOT EXISTS ( + SELECT fullkey, type, atom FROM json_tree(?3) + EXCEPT SELECT fullkey, type, atom FROM json_tree(${current}) + ) + `; + const statement = newValue === undefined + ? this.#db.prepare(`DELETE FROM "${this.#tableName}" WHERE ${matches}`) + .bind(encodedKey, now, expectedJson) + : this.#db.prepare(` + UPDATE "${this.#tableName}" SET value = ?4, expires = ?5 + WHERE ${matches} + `).bind( + encodedKey, + now, + expectedJson, + JSON.stringify(newValue), + this.#getExpires(now, options), + ); + const { meta } = await statement.run(); + return meta.changes > 0; + } + + /** + * {@inheritDoc KvStore.list} + */ + async *list(prefix?: KvKey): AsyncIterable { + await this.initialize(); + const now = Date.now(); + let statement: WorkersD1PreparedStatementLike; + if (prefix == null || prefix.length === 0) { + statement = this.#db.prepare(` + SELECT key, value FROM "${this.#tableName}" + WHERE expires IS NULL OR expires > ? + ORDER BY key + `).bind(now); + } else { + // Keys are JSON encoded, so '["prefix","a"]' and '["prefix"]' both + // start with '["prefix"', while '["prefixed"]' does not: + const encodedPrefix = JSON.stringify(prefix).slice(0, -1); + statement = this.#db.prepare(` + SELECT key, value FROM "${this.#tableName}" + WHERE substr(key, 1, length(?1)) = ?1 + AND (expires IS NULL OR expires > ?2) + ORDER BY key + `).bind(encodedPrefix, now); + } + const { results } = await statement.all<{ key: string; value: string }>(); + for (const row of results) { + yield { + key: JSON.parse(row.key) as KvKey, + value: JSON.parse(row.value), + }; + } + } +} + /** * Options for {@link WorkersMessageQueue}. * @since 2.0.0 diff --git a/packages/cfworkers/test/d1.test.ts b/packages/cfworkers/test/d1.test.ts new file mode 100644 index 000000000..0916e8652 --- /dev/null +++ b/packages/cfworkers/test/d1.test.ts @@ -0,0 +1,248 @@ +import { env } from "cloudflare:test"; +import { describe, expect, it } from "vitest"; +import { WorkersD1KvStore } 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; + +async function insertRaw( + key: readonly string[], + value: unknown, + expires: number | null, + tableName = "fedify_kv", +): Promise { + await env.D1.prepare( + `INSERT OR REPLACE INTO "${tableName}" (key, value, expires) + VALUES (?, ?, ?)`, + ).bind(JSON.stringify(key), JSON.stringify(value), expires).run(); +} + +async function collect( + iterable: AsyncIterable<{ key: readonly string[]; value: unknown }>, +): Promise<{ key: readonly string[]; value: unknown }[]> { + const entries = []; + for await (const entry of iterable) entries.push(entry); + return entries; +} + +describe("WorkersD1KvStore", () => { + it("rejects an invalid table name", () => { + expect(() => new WorkersD1KvStore(env.D1, { tableName: "foo; DROP" })) + .toThrow(TypeError); + }); + + it("initialize() creates the table", async () => { + const store = new WorkersD1KvStore(env.D1, { tableName: "init_test" }); + await store.initialize(); + 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 }>(); + expect(row?.name).toBe("init_test"); + }); + + it("set() & get()", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["foo", "bar"], { foo: 1, bar: 2 }); + expect(await store.get(["foo", "bar"])).toEqual({ foo: 1, bar: 2 }); + expect(await store.get(["foo"])).toBeUndefined(); + + await store.set(["foo", "bar"], [1, "two"]); + expect(await store.get(["foo", "bar"])).toEqual([1, "two"]); + }); + + it("keeps stored null distinct from a missing key", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["null-value"], null); + expect(await store.get(["null-value"])).toBeNull(); + expect(await store.get(["missing"])).toBeUndefined(); + }); + + it("set() with TTL", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["ttl"], "value", { ttl: mockDuration(3600) }); + 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 }>(); + expect(row?.expires).toBeGreaterThan(Date.now() + 3_500_000); + }); + + it("get() excludes expired entries", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["expired"], "stale", Date.now() - 1000); + await insertRaw(["fresh"], "live", Date.now() + 60_000); + expect(await store.get(["expired"])).toBeUndefined(); + expect(await store.get(["fresh"])).toBe("live"); + }); + + it("set() purges expired entries", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["expired"], "stale", Date.now() - 1000); + 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 }>(); + expect(row?.count).toBe(0); + }); + + it("delete()", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["delete"], "value"); + await store.delete(["delete"]); + expect(await store.get(["delete"])).toBeUndefined(); + await store.delete(["nonexistent"]); + }); + + it("uses a custom table name", async () => { + const store = new WorkersD1KvStore(env.D1, { tableName: "custom_kv" }); + 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 }>(); + expect(row?.value).toBe('"value"'); + }); + + it("cas()", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["foo", "bar"], "foobar"); + expect(await store.cas(["foo", "bar"], "bar", "baz")).toBe(false); + expect(await store.get(["foo", "bar"])).toBe("foobar"); + expect(await store.cas(["foo", "bar"], "foobar", "baz")).toBe(true); + expect(await store.get(["foo", "bar"])).toBe("baz"); + + // Creates a missing entry only when undefined is expected: + expect(await store.cas(["new"], "x", "y")).toBe(false); + expect(await store.get(["new"])).toBeUndefined(); + expect(await store.cas(["new"], undefined, "created")).toBe(true); + expect(await store.get(["new"])).toBe("created"); + expect(await store.cas(["new"], undefined, "again")).toBe(false); + expect(await store.get(["new"])).toBe("created"); + + // Deletes the entry when the new value is undefined: + expect(await store.cas(["new"], "wrong", undefined)).toBe(false); + expect(await store.cas(["new"], "created", undefined)).toBe(true); + expect(await store.get(["new"])).toBeUndefined(); + expect(await store.cas(["new"], undefined, undefined)).toBe(true); + await store.set(["new"], null); + expect(await store.cas(["new"], undefined, undefined)).toBe(false); + }); + + it("cas() compares values structurally", async () => { + 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.get(["obj"])).toBe(1); + + await store.set(["arr"], [1, 2]); + // Order of array elements matters: + expect(await store.cas(["arr"], [2, 1], "x")).toBe(false); + // Missing or extra elements and members do not match: + expect(await store.cas(["arr"], [1], "x")).toBe(false); + expect(await store.cas(["arr"], [1, 2, 3], "x")).toBe(false); + // Scalar types matter: + expect(await store.cas(["arr"], ["1", "2"], "x")).toBe(false); + expect(await store.get(["arr"])).toEqual([1, 2]); + + await store.set(["empty"], {}); + expect(await store.cas(["empty"], [], "x")).toBe(false); + expect(await store.cas(["empty"], { a: 1 }, "x")).toBe(false); + expect(await store.cas(["empty"], {}, "x")).toBe(true); + + await store.set(["nullish"], null); + expect(await store.cas(["nullish"], false, "x")).toBe(false); + expect(await store.cas(["nullish"], null, "x")).toBe(true); + }); + + it("cas() treats expired entries as missing", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["expired"], "stale", Date.now() - 1000); + expect(await store.cas(["expired"], "stale", "x")).toBe(false); + expect(await store.cas(["expired"], undefined, "fresh")).toBe(true); + expect(await store.get(["expired"])).toBe("fresh"); + }); + + it("cas() with TTL", async () => { + const store = new WorkersD1KvStore(env.D1); + expect( + await store.cas(["ttl"], undefined, "a", { ttl: mockDuration(3600) }), + ).toBe(true); + 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 }>(); + expect(row?.expires).toBeNull(); + }); + + it("cas() is atomic under concurrency", async () => { + const store = new WorkersD1KvStore(env.D1); + const results = await Promise.all( + Array.from({ length: 10 }, (_, i) => store.cas(["lock"], undefined, i)), + ); + expect(results.filter((r) => r).length).toBe(1); + + await store.set(["counter"], 0); + const swaps = await Promise.all( + Array.from({ length: 10 }, () => store.cas(["counter"], 0, 1)), + ); + expect(swaps.filter((r) => r).length).toBe(1); + }); + + it("list()", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["prefix"], "root"); + await store.set(["prefix", "a"], "value-a"); + await store.set(["prefix", "b"], "value-b"); + await store.set(["prefix", "nested", "c"], "value-c"); + await store.set(["prefixed", "x"], "value-x"); + await store.set(["other", "x"], "value-x"); + + // Entries are ordered by their JSON-encoded keys: + expect(await collect(store.list(["prefix"]))).toEqual([ + { key: ["prefix", "a"], value: "value-a" }, + { key: ["prefix", "b"], value: "value-b" }, + { key: ["prefix", "nested", "c"], value: "value-c" }, + { key: ["prefix"], value: "root" }, + ]); + expect(await collect(store.list(["prefix", "nested"]))).toEqual([ + { key: ["prefix", "nested", "c"], value: "value-c" }, + ]); + expect((await collect(store.list())).length).toBe(6); + expect(await collect(store.list(["missing"]))).toEqual([]); + }); + + it("list() handles LIKE wildcards literally", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["a%", "1"], 1); + await store.set(["a_", "2"], 2); + await store.set(["ab", "3"], 3); + expect(await collect(store.list(["a%"]))).toEqual([ + { key: ["a%", "1"], value: 1 }, + ]); + expect(await collect(store.list(["a_"]))).toEqual([ + { key: ["a_", "2"], value: 2 }, + ]); + }); + + it("list() excludes expired entries", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["expired-list"], "stale-root", Date.now() - 1000); + await insertRaw(["expired-list", "stale"], "stale", Date.now() - 1000); + await insertRaw(["expired-list", "fresh"], "fresh", null); + expect(await collect(store.list(["expired-list"]))).toEqual([ + { key: ["expired-list", "fresh"], value: "fresh" }, + ]); + }); +}); diff --git a/packages/cfworkers/test/env.d.ts b/packages/cfworkers/test/env.d.ts index 63748bf08..062670713 100644 --- a/packages/cfworkers/test/env.d.ts +++ b/packages/cfworkers/test/env.d.ts @@ -1,7 +1,9 @@ -// KVNamespace and Queue are globals provided by @cloudflare/workers-types, -// which is a global declaration file without named exports to import from. +// D1Database, KVNamespace, and Queue are globals provided by +// @cloudflare/workers-types, which is a global declaration file without named +// exports to import from. declare module "cloudflare:test" { interface ProvidedEnv { + D1: D1Database; KV1: KVNamespace; Q1: Queue; } diff --git a/packages/cfworkers/test/typecheck/workers-types.ts b/packages/cfworkers/test/typecheck/workers-types.ts index 32c0b6297..87f3a4fd2 100644 --- a/packages/cfworkers/test/typecheck/workers-types.ts +++ b/packages/cfworkers/test/typecheck/workers-types.ts @@ -1,15 +1,31 @@ -import type { KVNamespace, Queue } from "@cloudflare/workers-types"; +import type { + D1Database, + D1DatabaseSession, + KVNamespace, + Queue, +} from "@cloudflare/workers-types"; import type { KvStore, MessageQueue } from "@fedify/fedify/federation"; -import { WorkersKvStore, WorkersMessageQueue } from "../../dist/mod.js"; +import { + WorkersD1KvStore, + WorkersKvStore, + WorkersMessageQueue, +} from "../../dist/mod.js"; -// Checks that the public API accepts the KV and Queue bindings declared by -// `@cloudflare/workers-types` itself, in addition to the `wrangler types` +// Checks that the public API accepts the D1, KV, and Queue bindings declared +// by `@cloudflare/workers-types` itself, in addition to the `wrangler types` // declarations covered by wrangler-generated-kv.ts. +declare const d1: D1Database; +declare const d1Session: D1DatabaseSession; 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, { + tableName: "fedify_kv", + initialized: true, +}); export const mq: MessageQueue = new WorkersMessageQueue(queue); export const orderedMq: MessageQueue = new WorkersMessageQueue(queue, { orderingKv: kv, diff --git a/packages/cfworkers/wrangler.jsonc b/packages/cfworkers/wrangler.jsonc index 216fcad18..5aa4fea6d 100644 --- a/packages/cfworkers/wrangler.jsonc +++ b/packages/cfworkers/wrangler.jsonc @@ -9,6 +9,13 @@ "id": "00000000000000000000000000000000" } ], + "d1_databases": [ + { + "binding": "D1", + "database_name": "fedify-cfworkers-test", + "database_id": "00000000-0000-0000-0000-000000000000" + } + ], "queues": { "producers": [ { From ef3249512d99faa5b2dce9f4462e48a868fd885f Mon Sep 17 00:00:00 2001 From: BeiyanYunyi <14120445+BeiyanYunyi@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:43:00 -0500 Subject: [PATCH 2/2] Address review feedback on WorkersD1KvStore The review of https://github.com/fedify-dev/fedify/pull/1360 found that cas() compared expiration against a timestamp taken before the D1 request, so a delayed CAS could overwrite an entry that a concurrent reader had already observed as expired. It also pointed out unbounded work in list() and set(), per-request initialization, and documentation that overstated the store's guarantees. This commit includes the following changes: - Compared and computed expiration with D1's own clock inside each statement, and added a regression test with a delayed binding. - Fetched list() results in bounded, keyset-paginated batches over an indexed key range instead of buffering one unbounded scan. - Bounded the expired-entry sweep, and ran it on cas() writes too. - Fell back to comparing serializations in cas() for values nested deeper than json_tree() accepts. - Rejected table names reserved by SQLite and D1, renamed the index to idx__expires to match SqliteKvStore, and added drop(). - Documented that D1 sessions only provide sequential consistency, and recommended D1 migrations with initialized: true in production. - Updated the Deno KV deprecation guidance and remaining examples to recommend WorkersD1KvStore on Cloudflare Workers. - Added tests for initialization, sessions, competing instances, TTLs on update, pagination, and escaped or Unicode prefixes, and a type check against wrangler-generated D1 declarations. Assisted-by: Claude Code:claude-opus-5-5 --- CHANGES.md | 5 + changes.d/cfworkers/d1-kv-store.md | 5 + claude-plugin/skills/fedify/SKILL.md | 6 +- docs/manual/deploy.md | 10 +- docs/manual/kv.md | 35 ++- docs/manual/lint.md | 5 +- docs/manual/mq.md | 6 +- packages/cfworkers/README.md | 26 +- packages/cfworkers/src/mod.ts | 256 +++++++++------ packages/cfworkers/test/d1.test.ts | 297 +++++++++++++++++- .../test/typecheck/wrangler-generated-d1.ts | 73 +++++ packages/denokv/src/mod.ts | 12 +- packages/lint/src/lib/messages.ts | 5 +- 13 files changed, 594 insertions(+), 147 deletions(-) create mode 100644 packages/cfworkers/test/typecheck/wrangler-generated-d1.ts diff --git a/CHANGES.md b/CHANGES.md index b09125853..02fca8d09 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -119,6 +119,11 @@ To be released. consider migrating from `WorkersKvStore`. [[#1135], [#1360] by Beiyan Yunyi\] + Pass the `D1Database` binding itself, since D1 sessions can read stale + data from replicas. Because the store is constructed per request, create + its table with a D1 migration and pass `initialized: true` in production + to avoid an extra query per request. + [#1135]: https://github.com/fedify-dev/fedify/issues/1135 [#1360]: https://github.com/fedify-dev/fedify/pull/1360 diff --git a/changes.d/cfworkers/d1-kv-store.md b/changes.d/cfworkers/d1-kv-store.md index 2cc0729c7..337cab36e 100644 --- a/changes.d/cfworkers/d1-kv-store.md +++ b/changes.d/cfworkers/d1-kv-store.md @@ -12,3 +12,8 @@ links: code paths. It is now the recommended `KvStore` for Cloudflare Workers; consider migrating from `WorkersKvStore`. [[#1135], [#1360] by Beiyan Yunyi] + + Pass the `D1Database` binding itself, since D1 sessions can read stale + data from replicas. Because the store is constructed per request, create + its table with a D1 migration and pass `initialized: true` in production + to avoid an extra query per request. diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index 8e89dd177..b2654ca21 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -387,8 +387,10 @@ 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 -write-heavy and needs strong consistency, and only `WorkersD1KvStore` -supports `cas()`. +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. For `PostgresKvStore`, existing populated tables and externally managed DDL (`initialized: true`) need the [expiry expression index] to accelerate cleanup. diff --git a/docs/manual/deploy.md b/docs/manual/deploy.md index dbc28f00d..a46afc7a8 100644 --- a/docs/manual/deploy.md +++ b/docs/manual/deploy.md @@ -921,8 +921,9 @@ handler rather than as ambient imports. Fedify accommodates this through the [builder pattern](./federation.md#builder-pattern-for-structuring) and the [`@fedify/cfworkers`] package, which provides `WorkersD1KvStore`, `WorkersKvStore`, and `WorkersMessageQueue`. Prefer `WorkersD1KvStore` -over `WorkersKvStore`: Fedify's key–value workload is write-heavy and needs -strong consistency, which D1 provides and Workers KV does not. See the +over `WorkersKvStore`: Fedify writes to its key–value store about as often as +it reads from it and needs strong consistency, which D1 provides and Workers KV +does not. See the [*`WorkersD1KvStore`*](./kv.md#workersd1kvstore-cloudflare-workers-only) section for details. @@ -968,7 +969,8 @@ builder.setActorDispatcher("/users/{identifier}", async (ctx, identifier) => { export default { async fetch(request: Request, env: Env): Promise { const federation = await builder.build({ - kv: new WorkersD1KvStore(env.DB), + // The table is created by a D1 migration: + kv: new WorkersD1KvStore(env.DB, { initialized: true }), queue: new WorkersMessageQueue(env.QUEUE), }); return federation.fetch(request, { contextData: env }); @@ -1000,7 +1002,7 @@ export default { async queue(batch: MessageBatch, env: Env): Promise { const federation = await builder.build({ - kv: new WorkersD1KvStore(env.DB), + kv: new WorkersD1KvStore(env.DB, { initialized: true }), queue: new WorkersMessageQueue(env.QUEUE), }); diff --git a/docs/manual/kv.md b/docs/manual/kv.md index cebffc662..a88273102 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -496,8 +496,9 @@ const federation = createFederation({ > primarily for Deno Deploy, which is shutting down. Fedify keeps supporting > `@fedify/denokv` throughout the 2.x series, but the package will be removed > in Fedify 3.0.0. Use [`SqliteKvStore`] for local or single-node deployments, -> or [`PostgresKvStore`], [`RedisKvStore`], [`MysqlKvStore`], or -> [`WorkersD1KvStore`] for production. See [#1336] for the removal plan. +> [`PostgresKvStore`], [`RedisKvStore`], or [`MysqlKvStore`] for production, +> or [`WorkersD1KvStore`] if you move to Cloudflare Workers. See [#1336] for +> the removal plan. To use the [`DenoKvStore`], you need to install the *@fedify/denokv* package first: @@ -602,7 +603,8 @@ import { WorkersD1KvStore } from "@fedify/cfworkers"; export default { async fetch(request, env, ctx) { const federation: Federation = await builder.build({ - kv: new WorkersD1KvStore(env.DB), + // The table is created by a D1 migration; see below: + kv: new WorkersD1KvStore(env.DB, { initialized: true }), }); return await federation.fetch(request, { contextData: undefined }); } @@ -610,10 +612,11 @@ export default { ~~~~ The store keeps its entries in a table named `fedify_kv` by default, which -you can change with the `tableName` option. It creates the table on its -first use in each isolate. If you would rather manage the schema with -[D1 migrations], create the table in a migration and pass -`initialized: true` to skip the check: +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`: ~~~~ sql CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -621,18 +624,18 @@ CREATE TABLE IF NOT EXISTS fedify_kv ( value TEXT NOT NULL, expires INTEGER ); -CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); +CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); ~~~~ -~~~~ typescript twoslash -import { WorkersD1KvStore } from "@fedify/cfworkers"; -const env = undefined as unknown as { DB: D1Database }; -// ---cut-before--- -const kv = new WorkersD1KvStore(env.DB, { initialized: true }); -~~~~ +Expired entries are never returned. Each write also purges a bounded number +of expired entries from the table, so the table does not grow without limit. -Expired entries are never returned, and they are purged from the table -whenever a value is set. +> [!WARNING] +> Pass the `D1Database` binding itself (e.g., `env.DB`) to get strongly +> consistent reads. `WorkersD1KvStore` also accepts a D1 session created +> with `D1Database.withSession()`, but sessions only provide sequential +> consistency and can read stale data from a replica, which can make Fedify +> process the same activity twice. > [!NOTE] > Since your `D1Database` is not bound to a global variable, but rather diff --git a/docs/manual/lint.md b/docs/manual/lint.md index 149c1573e..5b69f15d1 100644 --- a/docs/manual/lint.md +++ b/docs/manual/lint.md @@ -1799,8 +1799,9 @@ covered. Deno KV was built primarily for Deno Deploy, which is shutting down. `@fedify/denokv` remains supported throughout Fedify 2.x, but it will be removed in Fedify 3.0.0. Use `SqliteKvStore` from *@fedify/sqlite* for local -or single-node deployments, or `PostgresKvStore`, `RedisKvStore`, -`MysqlKvStore`, or `WorkersKvStore` for production. +or single-node deployments, `PostgresKvStore`, `RedisKvStore`, or +`MysqlKvStore` for production, or `WorkersD1KvStore` from *@fedify/cfworkers* +on Cloudflare Workers. ~~~~ typescript // ❌ Bad: importing the deprecated adapter diff --git a/docs/manual/mq.md b/docs/manual/mq.md index bc103d0d4..807130155 100644 --- a/docs/manual/mq.md +++ b/docs/manual/mq.md @@ -641,10 +641,10 @@ export default { > > ~~~~ typescript > import { createFederationBuilder, type Message } from "@fedify/fedify"; -> import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; +> import { WorkersD1KvStore, WorkersMessageQueue } from "@fedify/cfworkers"; > > type Env = { -> KV_NAMESPACE: KVNamespace; +> DB: D1Database; > QUEUE_BINDING: Queue; > ORDERING_KV: KVNamespace; > }; @@ -657,7 +657,7 @@ export default { > orderingKv: env.ORDERING_KV, > }); > const federation = await builder.build({ -> kv: new WorkersKvStore(env.KV_NAMESPACE), +> kv: new WorkersD1KvStore(env.DB, { initialized: true }), > queue, > }); > diff --git a/packages/cfworkers/README.md b/packages/cfworkers/README.md index ccfcdadc9..6f34b3fe5 100644 --- a/packages/cfworkers/README.md +++ b/packages/cfworkers/README.md @@ -23,7 +23,7 @@ import { WorkersD1KvStore, WorkersMessageQueue } from "@fedify/cfworkers"; export default { async fetch(request, env, ctx) { const federation = createFederation({ - kv: new WorkersD1KvStore(env.DB_BINDING), + kv: new WorkersD1KvStore(env.DB_BINDING, { initialized: true }), queue: new WorkersMessageQueue(env.QUEUE_BINDING), // ... other options }); @@ -33,7 +33,7 @@ export default { async queue(batch, env, ctx) { const federation = createFederation({ - kv: new WorkersD1KvStore(env.DB_BINDING), + kv: new WorkersD1KvStore(env.DB_BINDING, { initialized: true }), queue: new WorkersMessageQueue(env.QUEUE_BINDING), // ... other options }); @@ -72,15 +72,16 @@ export default { *This API is available since Fedify 2.5.0.* `WorkersD1KvStore` is a key–value store implementation for -[Cloudflare Workers] that uses [Cloudflare D1]. D1 is strongly consistent -and suits Fedify's balanced read and write workload, and `WorkersD1KvStore` -supports the atomic `cas()` operation. It is the recommended key–value store -for Cloudflare Workers applications. +[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 +recommended key–value store for Cloudflare Workers applications. -The store creates its table (`fedify_kv` by default, configurable with -the `tableName` option) on its first use. If you manage the schema with -[D1 migrations] instead, create the table in a migration and pass -`initialized: true`: +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: ~~~~ sql CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -88,9 +89,12 @@ CREATE TABLE IF NOT EXISTS fedify_kv ( value TEXT NOT NULL, expires INTEGER ); -CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); +CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); ~~~~ +Pass the `D1Database` binding itself rather than a D1 session: sessions only +provide sequential consistency and can read stale data from a replica. + [Cloudflare D1]: https://developers.cloudflare.com/d1/ [D1 migrations]: https://developers.cloudflare.com/d1/reference/migrations/ diff --git a/packages/cfworkers/src/mod.ts b/packages/cfworkers/src/mod.ts index 533bc0f7b..061689dce 100644 --- a/packages/cfworkers/src/mod.ts +++ b/packages/cfworkers/src/mod.ts @@ -270,7 +270,9 @@ export interface WorkersD1PreparedStatementLike { */ export interface WorkersD1DatabaseLike { prepare(query: string): WorkersD1PreparedStatementLike; - batch(statements: WorkersD1PreparedStatementLike[]): Promise; + batch( + statements: WorkersD1PreparedStatementLike[], + ): Promise<{ readonly meta: { readonly changes: number } }[]>; } /** @@ -280,7 +282,8 @@ export interface WorkersD1DatabaseLike { export interface WorkersD1KvStoreOptions { /** * The table name to use for the key–value store. - * Only letters, digits, and underscores are allowed. + * Only letters, digits, and underscores are allowed, and names starting + * with `sqlite_` or `_cf_` are reserved by SQLite and D1. * `"fedify_kv"` by default. * @default `"fedify_kv"` */ @@ -288,24 +291,53 @@ export interface WorkersD1KvStoreOptions { /** * Whether the table has already been created, for example, by a D1 - * migration. If `false`, the store creates the table on its first use. + * 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 + * usually constructed per request on Cloudflare Workers, create the table + * with a D1 migration and set this to `true` in production. * `false` by default. * @default `false` */ readonly initialized?: boolean; } +/** + * The current time in milliseconds since the Unix epoch, as seen by D1. + * Comparing expiration against the database clock keeps each statement + * consistent with the moment it actually runs, rather than the moment the + * Worker sent it. + */ +const D1_NOW = "CAST(unixepoch('subsec') * 1000 AS INTEGER)"; + +/** + * The maximum number of rows to fetch per query in + * {@link WorkersD1KvStore.list}. D1 rows can be up to 2 MB each, so this + * keeps a page well under D1's 50 MB limit on query results. + */ +const D1_LIST_BATCH_SIZE = 20; + +/** + * The maximum number of expired rows to delete per write. + */ +const D1_EXPIRE_BATCH_SIZE = 100; + /** * Implementation of the {@link KvStore} interface for Cloudflare D1 binding. * - * Unlike {@link WorkersKvStore}, this store is strongly consistent and is - * priced for balanced read and write workloads, which suits Fedify's 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. + * 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. * - * The store uses a table with the following schema, which it creates on - * its first use unless the `initialized` option is set: + * 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. + * + * The store uses a table with the following schema, which each instance + * creates on its first use unless the `initialized` option is set: * * ```sql * CREATE TABLE IF NOT EXISTS fedify_kv ( @@ -313,7 +345,7 @@ export interface WorkersD1KvStoreOptions { * value TEXT NOT NULL, * expires INTEGER * ); - * CREATE INDEX IF NOT EXISTS fedify_kv_expires ON fedify_kv (expires); + * CREATE INDEX IF NOT EXISTS idx_fedify_kv_expires ON fedify_kv (expires); * ``` * * @example @@ -324,7 +356,8 @@ export interface WorkersD1KvStoreOptions { * export default { * async fetch(request, env, ctx) { * const federation = createFederation({ - * kv: new WorkersD1KvStore(env.DB), + * // The table is created by a D1 migration: + * kv: new WorkersD1KvStore(env.DB, { initialized: true }), * // ... other options * }); * return federation.fetch(request, { contextData: env }); @@ -336,15 +369,17 @@ export interface WorkersD1KvStoreOptions { export class WorkersD1KvStore implements KvStore { static readonly #defaultTableName = "fedify_kv"; static readonly #tableNameRegex = /^[A-Za-z_][A-Za-z0-9_]{0,63}$/; + static readonly #reservedTableNameRegex = /^(sqlite_|_cf_)/i; readonly #db: WorkersD1DatabaseLike; readonly #tableName: string; #initialized: Promise | null; /** * Constructs a new {@link WorkersD1KvStore} with the given D1 binding. - * @param db The Cloudflare D1 binding (or a D1 session). + * @param db The Cloudflare D1 binding. A D1 session is also accepted, but + * it only provides sequential consistency. * @param options Options for the key–value store. - * @throws {TypeError} If the table name is invalid. + * @throws {TypeError} If the table name is invalid or reserved. */ constructor( db: WorkersD1DatabaseLike, @@ -353,7 +388,10 @@ export class WorkersD1KvStore implements KvStore { this.#db = db; this.#tableName = options.tableName ?? WorkersD1KvStore.#defaultTableName; - if (!WorkersD1KvStore.#tableNameRegex.test(this.#tableName)) { + if ( + !WorkersD1KvStore.#tableNameRegex.test(this.#tableName) || + WorkersD1KvStore.#reservedTableNameRegex.test(this.#tableName) + ) { throw new TypeError( `Invalid table name for the key–value store: ${this.#tableName}`, ); @@ -364,8 +402,8 @@ 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 - * automatically on the first use of the store unless the `initialized` - * option is set. + * automatically on the first use of the store instance unless + * the `initialized` option is set. */ initialize(): Promise { this.#initialized ??= this.#db.batch([ @@ -377,7 +415,7 @@ export class WorkersD1KvStore implements KvStore { ) `), this.#db.prepare(` - CREATE INDEX IF NOT EXISTS "${this.#tableName}_expires" + CREATE INDEX IF NOT EXISTS "idx_${this.#tableName}_expires" ON "${this.#tableName}" (expires) `), ]).then(() => undefined, (error) => { @@ -388,14 +426,36 @@ export class WorkersD1KvStore implements KvStore { return this.#initialized; } + /** + * Drops the table used by the key–value store. Does nothing if the table + * does not exist. Mostly useful for tests. + */ + async drop(): Promise { + await this.#db.prepare(`DROP TABLE IF EXISTS "${this.#tableName}"`).run(); + this.#initialized = null; + } + #encodeKey(key: KvKey): string { return JSON.stringify(key); } - #getExpires(now: number, options?: KvStoreSetOptions): number | null { - return options?.ttl == null - ? null - : now + options.ttl.total("milliseconds"); + #getTtl(options?: KvStoreSetOptions): number | null { + return options?.ttl == null ? null : options.ttl.total("milliseconds"); + } + + /** + * 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} + ) + `); } /** @@ -405,8 +465,8 @@ export class WorkersD1KvStore implements KvStore { await this.initialize(); const row = await this.#db.prepare(` SELECT value FROM "${this.#tableName}" - WHERE key = ? AND (expires IS NULL OR expires > ?) - `).bind(this.#encodeKey(key), Date.now()).first<{ value: string }>(); + WHERE key = ? AND (expires IS NULL OR expires > ${D1_NOW}) + `).bind(this.#encodeKey(key)).first<{ value: string }>(); return row == null ? undefined : JSON.parse(row.value) as T; } @@ -419,24 +479,19 @@ export class WorkersD1KvStore implements KvStore { options?: KvStoreSetOptions, ): Promise { await this.initialize(); - const now = Date.now(); - // Purge expired entries in the same round trip: await this.#db.batch([ - this.#db.prepare(` - DELETE FROM "${this.#tableName}" - WHERE expires IS NOT NULL AND expires <= ? - `).bind(now), this.#db.prepare(` INSERT INTO "${this.#tableName}" (key, value, expires) - VALUES (?, ?, ?) + 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.#getExpires(now, options), + this.#getTtl(options), ), + this.#expire(), ]); } @@ -461,61 +516,58 @@ export class WorkersD1KvStore implements KvStore { ): Promise { await this.initialize(); const encodedKey = this.#encodeKey(key); - const now = Date.now(); + 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: - const { meta } = await this.#db.prepare(` + statement = this.#db.prepare(` INSERT INTO "${this.#tableName}" (key, value, expires) - VALUES (?, ?, ?) + VALUES (?1, ?2, ${D1_NOW} + ?3) ON CONFLICT (key) DO UPDATE SET value = excluded.value, expires = excluded.expires - WHERE "${this.#tableName}".expires IS NOT NULL - AND "${this.#tableName}".expires <= ? - `).bind( - encodedKey, - JSON.stringify(newValue), - this.#getExpires(now, options), - now, - ).run(); - return meta.changes > 0; + 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), + ); } - - // 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: - const expectedJson = JSON.stringify(expectedValue); - // The column must be qualified, since json_tree() has a value column too: - const current = `"${this.#tableName}".value`; - const matches = ` - key = ?1 AND (expires IS NULL OR expires > ?2) - AND NOT EXISTS ( - SELECT fullkey, type, atom FROM json_tree(${current}) - EXCEPT SELECT fullkey, type, atom FROM json_tree(?3) - ) - AND NOT EXISTS ( - SELECT fullkey, type, atom FROM json_tree(?3) - EXCEPT SELECT fullkey, type, atom FROM json_tree(${current}) - ) - `; - const statement = newValue === undefined - ? this.#db.prepare(`DELETE FROM "${this.#tableName}" WHERE ${matches}`) - .bind(encodedKey, now, expectedJson) - : this.#db.prepare(` - UPDATE "${this.#tableName}" SET value = ?4, expires = ?5 - WHERE ${matches} - `).bind( - encodedKey, - now, - expectedJson, - JSON.stringify(newValue), - this.#getExpires(now, options), - ); - const { meta } = await statement.run(); + const [{ meta }] = await this.#db.batch([statement, this.#expire()]); return meta.changes > 0; } @@ -524,31 +576,37 @@ export class WorkersD1KvStore implements KvStore { */ async *list(prefix?: KvKey): AsyncIterable { await this.initialize(); - const now = Date.now(); - let statement: WorkersD1PreparedStatementLike; + // Keys are JSON-encoded arrays of strings, so they all fall in the range + // ('[', '\\'). Keys with a prefix such as ["a"] start with '["a"', and + // since a closing quote is followed by ',' or ']', they all sort below + // '["a#'. Neither bound is a valid key itself: + let cursor: string; + let upperBound: string; if (prefix == null || prefix.length === 0) { - statement = this.#db.prepare(` - SELECT key, value FROM "${this.#tableName}" - WHERE expires IS NULL OR expires > ? - ORDER BY key - `).bind(now); + cursor = "["; + upperBound = "\\"; } else { - // Keys are JSON encoded, so '["prefix","a"]' and '["prefix"]' both - // start with '["prefix"', while '["prefixed"]' does not: - const encodedPrefix = JSON.stringify(prefix).slice(0, -1); - statement = this.#db.prepare(` - SELECT key, value FROM "${this.#tableName}" - WHERE substr(key, 1, length(?1)) = ?1 - AND (expires IS NULL OR expires > ?2) - ORDER BY key - `).bind(encodedPrefix, now); + cursor = JSON.stringify(prefix).slice(0, -1); + upperBound = cursor.slice(0, -1) + "#"; } - const { results } = await statement.all<{ key: string; value: string }>(); - for (const row of results) { - yield { - key: JSON.parse(row.key) as KvKey, - value: JSON.parse(row.value), - }; + const statement = this.#db.prepare(` + SELECT key, value FROM "${this.#tableName}" + WHERE key > ?1 AND key < ?2 + AND (expires IS NULL OR expires > ${D1_NOW}) + ORDER BY key + LIMIT ${D1_LIST_BATCH_SIZE} + `); + while (true) { + const { results } = await statement.bind(cursor, upperBound) + .all<{ key: string; value: string }>(); + for (const row of results) { + yield { + key: JSON.parse(row.key) as KvKey, + value: JSON.parse(row.value), + }; + } + if (results.length < D1_LIST_BATCH_SIZE) break; + cursor = results[results.length - 1].key; } } } diff --git a/packages/cfworkers/test/d1.test.ts b/packages/cfworkers/test/d1.test.ts index 0916e8652..bf0aac04c 100644 --- a/packages/cfworkers/test/d1.test.ts +++ b/packages/cfworkers/test/d1.test.ts @@ -1,6 +1,10 @@ import { env } from "cloudflare:test"; import { describe, expect, it } from "vitest"; -import { WorkersD1KvStore } from "../src/mod.ts"; +import { + WorkersD1KvStore, + type WorkersD1DatabaseLike, + type WorkersD1PreparedStatementLike, +} from "../src/mod.ts"; // Mock Temporal.Duration for testing in Cloudflare Workers environment const mockDuration = (seconds: number) => ({ @@ -23,6 +27,85 @@ async function insertRaw( ).bind(JSON.stringify(key), JSON.stringify(value), expires).run(); } +const delay = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +interface D1SpyOptions { + /** Milliseconds to delay a query or batch before sending it. */ + readonly delay?: (queries: readonly string[]) => number; + /** Whether to fail a batch instead of sending it, by its statements. */ + readonly failBatch?: (queries: readonly string[]) => boolean; +} + +/** A prepared statement that remembers its query and the underlying one. */ +class SpyStatement implements WorkersD1PreparedStatementLike { + constructor( + readonly spy: D1Spy, + readonly query: string, + readonly inner: D1PreparedStatement, + ) {} + + bind(...values: unknown[]): SpyStatement { + return new SpyStatement(this.spy, this.query, this.inner.bind(...values)); + } + + async first>(): Promise { + await this.spy.send([this.query]); + return await this.inner.first(); + } + + async all>(): Promise< + { readonly results: T[] } + > { + await this.spy.send([this.query]); + return await this.inner.all(); + } + + async run(): Promise<{ readonly meta: { readonly changes: number } }> { + await this.spy.send([this.query]); + return await this.inner.run(); + } +} + +/** + * Wraps a D1 binding to record the queries it runs, and optionally to delay + * or fail batches. + */ +class D1Spy implements WorkersD1DatabaseLike { + readonly queries: string[] = []; + readonly batches: string[][] = []; + + constructor( + readonly db: D1Database, + readonly options: D1SpyOptions = {}, + ) {} + + prepare(query: string): SpyStatement { + return new SpyStatement(this, query, this.db.prepare(query)); + } + + async send(queries: string[]): Promise { + this.queries.push(...queries); + await delay(this.options.delay?.(queries) ?? 0); + } + + async batch( + statements: WorkersD1PreparedStatementLike[], + ): Promise<{ readonly meta: { readonly changes: number } }[]> { + const spyStatements = statements as SpyStatement[]; + const queries = spyStatements.map((s) => s.query); + 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)); + } + + get ddlBatches(): number { + return this.batches.filter((b) => b.some((q) => q.includes("CREATE"))) + .length; + } +} + async function collect( iterable: AsyncIterable<{ key: readonly string[]; value: unknown }>, ): Promise<{ key: readonly string[]; value: unknown }[]> { @@ -33,8 +116,10 @@ async function collect( describe("WorkersD1KvStore", () => { it("rejects an invalid table name", () => { - expect(() => new WorkersD1KvStore(env.D1, { tableName: "foo; DROP" })) - .toThrow(TypeError); + for (const tableName of ["foo; DROP", "sqlite_kv", "SQLITE_kv", "_cf_KV"]) { + expect(() => new WorkersD1KvStore(env.D1, { tableName })) + .toThrow(TypeError); + } }); it("initialize() creates the table", async () => { @@ -245,4 +330,210 @@ describe("WorkersD1KvStore", () => { { key: ["expired-list", "fresh"], value: "fresh" }, ]); }); + + it("initializes the table once per store instance", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + await Promise.all([store.get(["a"]), store.get(["b"])]); + await store.set(["a"], 1); + await store.cas(["a"], 1, 2); + 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 }>(); + expect(index?.name).toBe("idx_fedify_kv_expires"); + }); + + it("skips initialization when initialized is set", async () => { + await new WorkersD1KvStore(env.D1).initialize(); + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy, { initialized: true }); + await store.set(["a"], 1); + expect(await store.get(["a"])).toBe(1); + expect(await store.cas(["a"], 1, 2)).toBe(true); + await collect(store.list()); + await store.delete(["a"]); + expect(spy.ddlBatches).toBe(0); + expect(spy.queries.some((q) => q.includes("CREATE"))).toBe(false); + }); + + it("retries initialization after a failure", async () => { + let fail = true; + const spy = new D1Spy(env.D1, { + failBatch: (queries) => fail && queries.some((q) => q.includes("CREATE")), + }); + const store = new WorkersD1KvStore(spy); + await expect(store.get(["a"])).rejects.toThrow("Injected failure"); + fail = false; + await store.set(["a"], 1); + expect(await store.get(["a"])).toBe(1); + expect(spy.ddlBatches).toBe(2); + }); + + it("drop()", async () => { + const store = new WorkersD1KvStore(env.D1, { tableName: "drop_test" }); + await store.set(["a"], 1); + await store.drop(); + const row = await env.D1.prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", + ).bind("drop_test").first(); + expect(row).toBeNull(); + // The store re-creates the table on its next use: + expect(await store.get(["a"])).toBeUndefined(); + }); + + it("works with a D1 session", async () => { + const store = new WorkersD1KvStore(env.D1.withSession("first-primary")); + await store.set(["session"], { a: 1 }); + expect(await store.get(["session"])).toEqual({ a: 1 }); + expect(await store.cas(["session"], { a: 1 }, 2)).toBe(true); + expect(await collect(store.list(["session"]))).toEqual([ + { key: ["session"], value: 2 }, + ]); + }); + + it("set() purges a bounded number of expired entries", async () => { + const store = new WorkersD1KvStore(env.D1); + 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), + ), + ); + const countExpired = async () => + (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: + await store.cas(["a"], 1, 2); + expect(await countExpired()).toBe(0); + }); + + it("cas() does not match an entry that expires while in flight", async () => { + // Regression test: the expiration must be checked against the time the + // statement runs, not the time the Worker sent it. + const delayed = new WorkersD1KvStore( + new D1Spy(env.D1, { + delay: (queries) => + queries.some((q) => q.trimStart().startsWith("UPDATE")) ? 600 : 0, + }), + ); + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["in-flight"], "old", Date.now() + 300); + const swap = delayed.cas(["in-flight"], "old", "new"); + await delay(400); + expect(await store.get(["in-flight"])).toBeUndefined(); + expect(await store.cas(["in-flight"], undefined, undefined)).toBe(true); + expect(await swap).toBe(false); + expect(await store.get(["in-flight"])).toBeUndefined(); + }); + + it("cas() with an explicit TTL on update", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.set(["ttl-update"], "a"); + expect( + await store.cas(["ttl-update"], "a", "b", { ttl: mockDuration(3600) }), + ).toBe(true); + const row = await env.D1.prepare( + "SELECT expires FROM fedify_kv WHERE key = ?", + ).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); + }); + + it("cas() does not delete an expired entry", async () => { + const store = new WorkersD1KvStore(env.D1); + await store.initialize(); + await insertRaw(["expired-delete"], "stale", Date.now() - 1000); + expect(await store.cas(["expired-delete"], "stale", undefined)).toBe(false); + }); + + it("cas() compares deeply nested values", async () => { + const store = new WorkersD1KvStore(env.D1); + // Deeper than json_tree() can parse: + let deep: unknown = "leaf"; + for (let i = 0; i < 1001; i++) deep = [deep]; + await store.set(["deep"], deep); + expect(await store.get(["deep"])).toEqual(deep); + expect(await store.cas(["deep"], ["other"], 1)).toBe(false); + expect(await store.cas(["deep"], deep, 1)).toBe(true); + expect(await store.cas(["deep"], 1, deep)).toBe(true); + expect(await store.cas(["deep"], deep, undefined)).toBe(true); + expect(await store.get(["deep"])).toBeUndefined(); + }); + + it("cas() lets only one of separate store instances win", async () => { + const stores = Array.from( + { length: 5 }, + () => new WorkersD1KvStore(env.D1), + ); + await stores[0].initialize(); + const absent = await Promise.all( + stores.map((s, i) => s.cas(["absent"], undefined, i)), + ); + expect(absent.filter((r) => r).length).toBe(1); + + await insertRaw(["expired-race"], "stale", Date.now() - 1000); + const expired = await Promise.all( + stores.map((s, i) => s.cas(["expired-race"], undefined, i)), + ); + expect(expired.filter((r) => r).length).toBe(1); + }); + + it("list() fetches entries in bounded batches", async () => { + const spy = new D1Spy(env.D1); + const store = new WorkersD1KvStore(spy); + const keys = Array.from( + { length: 45 }, + (_, i) => ["paged", `${i}`.padStart(2, "0")] as const, + ); + for (const key of keys) await store.set(key, key[1]); + await store.set(["paged-other"], "other"); + await store.set(["other"], "other"); + + const before = spy.queries.length; + const entries = await collect(store.list(["paged"])); + expect(entries).toEqual(keys.map((key) => ({ key, value: key[1] }))); + // 45 entries in batches of 20: + expect(spy.queries.length - before).toBe(3); + expect((await collect(store.list())).length).toBe(47); + }); + + it("list() handles escaped, Unicode, and numeric-looking prefixes", async () => { + const store = new WorkersD1KvStore(env.D1); + const keys = [ + ['a"b'], + ['a"b', "x"], + ['a"bc'], + ["a\\"], + ["a\\", "x"], + ["한글"], + ["한글", "1"], + ["한글x"], + ["😀"], + ["1"], + ["1", "0"], + ["10"], + ["-1"], + ["-10"], + ] as const; + for (const key of keys) await store.set(key, key.join("/")); + const listKeys = async (prefix: readonly [string, ...string[]]) => + (await collect(store.list(prefix))).map((e) => e.key); + expect(await listKeys(['a"b'])).toEqual([['a"b', "x"], ['a"b']]); + expect(await listKeys(["a\\"])).toEqual([["a\\", "x"], ["a\\"]]); + expect(await listKeys(["한글"])).toEqual([["한글", "1"], ["한글"]]); + expect(await listKeys(["😀"])).toEqual([["😀"]]); + expect(await listKeys(["1"])).toEqual([["1", "0"], ["1"]]); + expect(await listKeys(["-1"])).toEqual([["-1"]]); + expect(await listKeys(["1", "0"])).toEqual([["1", "0"]]); + }); }); diff --git a/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts b/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts new file mode 100644 index 000000000..45927c269 --- /dev/null +++ b/packages/cfworkers/test/typecheck/wrangler-generated-d1.ts @@ -0,0 +1,73 @@ +import type { KvStore } from "@fedify/fedify/federation"; +import { WorkersD1KvStore } from "../../dist/mod.js"; + +interface GeneratedD1Meta { + duration: number; + size_after: number; + rows_read: number; + rows_written: number; + last_row_id: number; + changed_db: boolean; + changes: number; + served_by_region?: string; + served_by_primary?: boolean; +} + +interface GeneratedD1Response { + success: true; + meta: GeneratedD1Meta & Record; + error?: never; +} + +type GeneratedD1Result = GeneratedD1Response & { + results: T[]; +}; + +interface GeneratedD1ExecResult { + count: number; + duration: number; +} + +/** + * Mirrors the D1 prepared statement declaration shape emitted by + * `wrangler types`, but comes from a distinct local declaration source. + */ +interface GeneratedD1PreparedStatement { + bind(...values: unknown[]): GeneratedD1PreparedStatement; + first(colName: string): Promise; + first>(): Promise; + run>(): Promise>; + all>(): Promise>; + raw(options: { + columnNames: true; + }): Promise<[string[], ...T[]]>; + raw(options?: { columnNames?: false }): Promise; +} + +interface GeneratedD1DatabaseSession { + prepare(query: string): GeneratedD1PreparedStatement; + batch( + statements: GeneratedD1PreparedStatement[], + ): Promise[]>; + getBookmark(): string | null; +} + +/** + * Mirrors the D1 binding declaration shape emitted by `wrangler types`. + */ +interface GeneratedD1Database { + prepare(query: string): GeneratedD1PreparedStatement; + batch( + statements: GeneratedD1PreparedStatement[], + ): Promise[]>; + exec(query: string): Promise; + withSession(constraintOrBookmark?: string): GeneratedD1DatabaseSession; +} + +declare const generatedD1: GeneratedD1Database; + +export const store: KvStore = new WorkersD1KvStore(generatedD1); +export const sessionStore: KvStore = new WorkersD1KvStore( + generatedD1.withSession("first-primary"), + { initialized: true }, +); diff --git a/packages/denokv/src/mod.ts b/packages/denokv/src/mod.ts index ffbe743c3..395b70d03 100644 --- a/packages/denokv/src/mod.ts +++ b/packages/denokv/src/mod.ts @@ -39,8 +39,9 @@ const DEPRECATION_MESSAGE = const STORE_DEPRECATION_MESSAGE = "The `DenoKvStore` class from `@fedify/denokv` is deprecated and will be " + "removed in Fedify 3.0.0. Use `SqliteKvStore` from `@fedify/sqlite` for " + - "local or single-node deployments, or `PostgresKvStore`, `RedisKvStore`, " + - "`MysqlKvStore`, or `WorkersKvStore` for production. " + DEPRECATION_MESSAGE; + "local or single-node deployments, `PostgresKvStore`, `RedisKvStore`, " + + "or `MysqlKvStore` for production, or `WorkersD1KvStore` from " + + "`@fedify/cfworkers` on Cloudflare Workers. " + DEPRECATION_MESSAGE; const QUEUE_DEPRECATION_MESSAGE = "The `DenoKvMessageQueue` class from `@fedify/denokv` is deprecated and " + @@ -58,9 +59,10 @@ const QUEUE_DEPRECATION_MESSAGE = * down. Fedify keeps supporting `@fedify/denokv` throughout the * 2.x series, but this class will be removed in Fedify 3.0.0. * Use `SqliteKvStore` from `@fedify/sqlite` for local or - * single-node deployments, or `PostgresKvStore`, - * `RedisKvStore`, `MysqlKvStore`, or `WorkersKvStore` for - * production deployments. See + * single-node deployments, `PostgresKvStore`, `RedisKvStore`, + * or `MysqlKvStore` for production deployments, or + * `WorkersD1KvStore` from `@fedify/cfworkers` on Cloudflare + * Workers. See * for details. */ export class DenoKvStore implements KvStore { diff --git a/packages/lint/src/lib/messages.ts b/packages/lint/src/lib/messages.ts index 8d806ab95..2e2a87198 100644 --- a/packages/lint/src/lib/messages.ts +++ b/packages/lint/src/lib/messages.ts @@ -74,8 +74,9 @@ export const DENO_KV_STORE_DEPRECATED = "`DenoKvStore` from `@fedify/denokv` is deprecated since Fedify 2.5.0 and " + "will be removed in Fedify 3.0.0. Use `SqliteKvStore` from " + "`@fedify/sqlite` for local or single-node deployments, or " + - "`PostgresKvStore`, `RedisKvStore`, `MysqlKvStore`, or `WorkersKvStore` " + - "for production. See https://github.com/fedify-dev/fedify/issues/1336 " + + "`PostgresKvStore`, `RedisKvStore`, or `MysqlKvStore` for production, " + + "or `WorkersD1KvStore` from `@fedify/cfworkers` on Cloudflare Workers. " + + "See https://github.com/fedify-dev/fedify/issues/1336 " + "for the removal plan."; /**