From 37ae4111a43f0248e929b80113b004a9266fc9ca Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 02:51:57 +0900 Subject: [PATCH 1/2] Add atomic operations to PGlite KV store Use PGlite's exclusive owner transaction callback for consistent snapshots and all-or-nothing multi-key changes, including absent keys. Share single and batch CAS validation, preserve JSONB comparison and calendar expiry, and roll back mutation housekeeping with the writes. Preserve awaited owner filesystem synchronization after mutation commits. Document that synchronization errors can reject committed changes without replay. Cover concurrency, late failures, expiry, input capture, sync ordering and failures, and clean filesystem reopen within the supported owner, codec, and storage scope. Closes https://github.com/fedify-dev/fedify/issues/1366 Related to https://github.com/fedify-dev/fedify/issues/1361 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-fable-5-2 Assisted-by: Claude Code:claude-opus-5-5 Assisted-by: Codex:gpt-6-astra --- changes.d/pglite/atomic-kv-store.md | 20 + claude-plugin/skills/fedify/SKILL.md | 23 +- docs/manual/kv.md | 33 +- packages/pglite/README.md | 84 +++- packages/pglite/src/atomic-kv.test.ts | 619 ++++++++++++++++++++++++++ packages/pglite/src/kv.ts | 364 ++++++++++----- 6 files changed, 1024 insertions(+), 119 deletions(-) create mode 100644 changes.d/pglite/atomic-kv-store.md create mode 100644 packages/pglite/src/atomic-kv.test.ts diff --git a/changes.d/pglite/atomic-kv-store.md b/changes.d/pglite/atomic-kv-store.md new file mode 100644 index 000000000..2e6c18ffd --- /dev/null +++ b/changes.d/pglite/atomic-kv-store.md @@ -0,0 +1,20 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1366': https://github.com/fedify-dev/fedify/issues/1366 + '#1431': https://github.com/fedify-dev/fedify/pull/1431 +--- + - Added atomic snapshots and multi-key writes, deletions, and conditional + updates to `PgliteKvStore` through `AtomicKvStore`. Wrappers sharing one + PGlite owner can commit related keys together with JSONB comparison and + per-entry TTLs. [[#1361], [#1366], [#1431]] + + Malformed keys, negative or overflowing TTLs, and TTLs on CAS deletions + now reject before mutation. Mutation cleanup shares the transaction, + so cleanup failures roll back writes; legacy invalid TTL rows require + removal together or direct SQL remediation. + + Existing single-key mutations and the new batch mutations await the + PGlite owner's filesystem synchronization policy after a successful + transaction. If synchronization fails, the operation rejects after its + changes have committed; retrying blindly can repeat the change. diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index 05b4409bd..7d5fdac4b 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -444,6 +444,7 @@ in development or tests. | Package | `KvStore` | `MessageQueue` | | ------------------- | --------- | -------------- | | *@fedify/sqlite* | yes | yes | +| *@fedify/pglite* | yes | no | | *@fedify/postgres* | yes | yes | | *@fedify/mysql* | yes | yes | | *@fedify/redis* | yes | yes | @@ -481,7 +482,8 @@ 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` and `SqliteKvStore` implement the full atomic interface. +`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 @@ -491,10 +493,21 @@ 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. -The persistent PostgreSQL, Redis, PGlite, and MySQL extensions remain tracked -separately in issues #1364–#1366 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 + +For PGlite, share one normal owner and table across wrappers, require standard +JSONB/text-array codecs, and keep raw/manual transaction control and +caller-owned transaction callbacks outside store calls. CAS ignores object +member order and preserves array order. Keys reject malformed Unicode and +NUL; negative/overflowing TTLs and deletion TTLs reject. Calendar expiry uses +the session time zone and one cutoff per call. Large batches block the owner; +engine insertion failures roll back, while mismatch does not prove proposed +writes would fit. Do not infer crash durability, browser persistence, or +worker-proxy support. Read the PGlite package README's synchronization policy. + +The PostgreSQL, Redis, and MySQL atomic extensions are tracked separately in +issues #1364, #1365, and #1368; Deno KV and Netlify Blobs have no atomic adapter +child issue. `WorkersKvStore` lacks CAS. `WorkersD1KvStore` supports single-key +CAS; its atomic extension is tracked in #1379. Check the [atomic support matrix] and installed declarations before selecting a production adapter. diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 03539a197..ddceb8483 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -480,7 +480,29 @@ const federation = createFederation({ > directory from multiple worker processes. Use one PGlite instance per data > directory and runtime isolate. +Since Fedify 2.5.0, this store also implements `AtomicKvStore`. +Each snapshot or batch uses one PGlite transaction and one expiry cutoff. +Wrappers must share one normal PGlite owner and table. Do not call the store +inside the owner's transaction callback or bypass its query queue with raw +protocol or manual transaction control. Worker proxies and browser +persistence are outside the documented atomic scope. + +CAS compares JSONB values: object member order is ignored, array order +matters, and stored null differs from absence. Keys must be nonempty arrays +of well-formed Unicode strings without U+0000. Negative TTLs, timestamp +overflow, and TTLs on CAS deletion reject; zero TTL expires immediately. +Intervals round to microseconds and calendar units follow the session time +zone. There is no fixed batch-entry cap. Engine key insertion limits depend +on compression; a mismatch does not prove that unattempted writes would fit. +Statement failures roll back the batch; commit or post-commit synchronization +failures can reject after changes have committed. Long batches occupy the +shared instance. Mutations await the owner's `syncToFs()` after commit; atomic +execution does not establish crash durability. See the +[package's atomic-operation guidance] for codecs, legacy-row recovery, and the +owner's synchronization policy. + [`PgliteKvStore`]: https://jsr.io/@fedify/pglite/doc/kv/~/PgliteKvStore +[package's atomic-operation guidance]: https://github.com/fedify-dev/fedify/tree/main/packages/pglite#atomic-operations ### [`MysqlKvStore`] @@ -984,7 +1006,8 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore` and `SqliteKvStore` implement the full `AtomicKvStore` contract. +`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: @@ -993,7 +1016,7 @@ The linked issues track separate adapter implementations: | `MemoryKvStore` | Yes | Development and testing only; one instance. | | `PostgresKvStore` | No | Planned in [#1364]. | | `RedisKvStore` | No | Planned in [#1365]. | -| `PgliteKvStore` | No | Planned in [#1366]. | +| `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | | `SqliteKvStore` | Yes | One configured table/database; writable handles, JSON values; see [SQLite scope](#sqlitekvstore). | | `MysqlKvStore` | No | Planned in [#1368]. | | `DenoKvStore` | No | No atomic adapter child issue; deprecated since 2.5.0. | @@ -1004,7 +1027,6 @@ The linked issues track separate adapter implementations: [PR #1360]: https://github.com/fedify-dev/fedify/pull/1360 [#1364]: https://github.com/fedify-dev/fedify/issues/1364 [#1365]: https://github.com/fedify-dev/fedify/issues/1365 -[#1366]: https://github.com/fedify-dev/fedify/issues/1366 [#1368]: https://github.com/fedify-dev/fedify/issues/1368 [#1379]: https://github.com/fedify-dev/fedify/issues/1379 @@ -1613,6 +1635,11 @@ scope. A process-local mutex cannot coordinate other handles or processes. On a deadlock retry, roll back and re-evaluate the complete expected set. Return `false` only for a comparison mismatch; propagate backend errors. +For example, `PgliteKvStore` uses the exclusive transaction callback of +one supported PGlite owner. Independent store wrappers share that owner, whose +client queue excludes other normal queries for the callback's entire lifetime, +including checks for absent keys. + Run the [shared atomic conformance tests](./test.md), using independent clients sharing an isolated backend for persistent adapters. Add backend-specific competing-client, rollback, expiry, and injected-failure diff --git a/packages/pglite/README.md b/packages/pglite/README.md index 126c91785..57d6385ed 100644 --- a/packages/pglite/README.md +++ b/packages/pglite/README.md @@ -5,7 +5,7 @@ [![JSR][JSR badge]][JSR] This package provides [`PgliteKvStore`], a PGlite-backed implementation of -Fedify's [`KvStore`]: +Fedify's [`AtomicKvStore`]: ~~~~ typescript import { PGlite } from "@electric-sql/pglite"; @@ -33,7 +33,7 @@ rolling restart. See [PGlite issue #489] and [PGlite pull request #892]. [JSR badge]: https://jsr.io/badges/@fedify/pglite [JSR]: https://jsr.io/@fedify/pglite [`PgliteKvStore`]: https://jsr.io/@fedify/pglite/doc/~/PgliteKvStore -[`KvStore`]: https://jsr.io/@fedify/fedify/doc/federation/~/KvStore +[`AtomicKvStore`]: https://jsr.io/@fedify/fedify/doc/federation/~/AtomicKvStore [PGlite issue #489]: https://github.com/electric-sql/pglite/issues/489#issuecomment-2587892190 [PGlite pull request #892]: https://github.com/electric-sql/pglite/pull/892 @@ -48,3 +48,83 @@ yarn add @fedify/pglite @electric-sql/pglite # Yarn bun add @fedify/pglite @electric-sql/pglite # Bun deno add jsr:@fedify/pglite npm:@electric-sql/pglite # Deno ~~~~ + + +Atomic operations +----------------- + +Since Fedify 2.5.0, `PgliteKvStore` implements `AtomicKvStore`. Use +`getMany()` for an ordered snapshot and `casMany()` to compare and change +related keys together. A comparison mismatch returns `false` without changing +values or TTLs; reread and re-evaluate the application rule before retrying. +`setMany()` and `deleteMany()` also commit as complete batches. + +For example, compare a generation and its associated state before replacing +both. The application computes the replacement from its own rule: + +~~~~ typescript +const store = new PgliteKvStore(pg); +const keys = [["example", "generation"], ["example", "state"]] as const; +const [generation, state] = await store.getMany(keys); +const committed = await store.casMany([ + { key: keys[0], expectedValue: generation, newValue: crypto.randomUUID() }, + { key: keys[1], expectedValue: state, newValue: { status: "ready" } }, +]); +~~~~ + +All wrappers for a table must share one caller-owned PGlite instance. Each +batch uses PGlite's transaction callback, which excludes other normal queries +and transactions on that instance, including when a compared key is absent. +Do not call store methods from inside the owner's transaction callback, issue +manual transaction commands, or use concurrent raw protocol operations. +Independent PGlite owners and processes must not open the same data directory. +The standard text-array and JSONB codecs are required; custom parser or +serializer overrides are outside this guarantee. + +Values use `JSON.stringify()` followed by PostgreSQL JSONB input, and CAS uses +JSONB equality. Object member order is ignored and array order matters. +JSON `null` remains a value; `undefined` checks absence or deletes in CAS and +cannot be stored. JSON serialization can omit unsupported object properties +or convert nonfinite numbers to `null`; cyclic values and `bigint` reject. +JSONB also rejects U+0000 and invalid surrogate sequences. Keys must contain +one or more well-formed Unicode strings without U+0000. Mutation batches +reject duplicate keys and sparse inputs before writing. + +Each entry has its own TTL, and every operation uses one transaction-start +expiry cutoff. Omitting a replacement TTL clears the previous expiry. Zero +TTL expires immediately; negative durations reject, as do durations whose +addition exceeds the timestamp range. TTL uses PostgreSQL calendar intervals +with microsecond rounding and the session time zone; the host clock may have +only millisecond resolution. CAS deletion must not specify a non-null TTL. + +There is no fixed adapter batch-entry limit. Statements run sequentially in +one transaction, so a large batch occupies the shared instance and consumes +memory proportional to its input and results. New keys are subject to the +engine's B-tree entry limit, about 2700 bytes after compression with its default +8 KiB pages; this is a storage constraint rather than an adapter input cap. +A mismatching CAS batch can return `false` without attempting an insertion or +proving that insertion would fit. An insertion error that actually occurs +rejects and rolls back the complete batch. Deterministically invalid keys, +JSONB values and TTLs still reject even on a comparison mismatch. + +Atomic execution does not promise crash durability or exactly-once effects. +The supported scope is a normal PGlite instance on Node.js, Bun, or Deno, +using memory or the host filesystem. Worker proxies and browser persistence +are outside this scope. Host-filesystem storage supports clean close/reopen; +memory-only instances remain ephemeral. After a mutation transaction completes, +the adapter awaits the owner's `syncToFs()` outside the transaction, preserving +the synchronization policy previously used by `set()`, `delete()`, and `cas()`. +PGlite 0.5.8 does not perform that post-commit synchronization itself. +The owner's filesystem and `relaxedDurability` setting determine what this +call guarantees. A synchronization failure rejects after the changes have +committed; it cannot roll them back. A lost response or commit failure can +also leave commit status unknown. Do not +automatically replay an uncertain change; use a non-reused version or an +operation receipt when the application needs it. + +Mutation housekeeping is part of the same transaction. A legacy row with an +overflowing TTL can make expiry cleanup fail; other mutations then roll back +until all offending keys are removed together with `deleteMany()` or direct +SQL. Inspect the raw key, creation time, and TTL columns through direct SQL +without adding the TTL. Malformed legacy keys, such as empty arrays or SQL +null segments, require direct SQL remediation. diff --git a/packages/pglite/src/atomic-kv.test.ts b/packages/pglite/src/atomic-kv.test.ts new file mode 100644 index 000000000..f1873c5b7 --- /dev/null +++ b/packages/pglite/src/atomic-kv.test.ts @@ -0,0 +1,619 @@ +// Cases and transaction statements must run sequentially on each owner. +/* oxlint-disable eslint/no-await-in-loop */ +import { + PGlite, + MemoryFS, + type PGliteInterface, + type Transaction, + type QueryOptions, +} from "@electric-sql/pglite"; +import type { KvKey, KvStoreCasEntry } from "@fedify/fedify"; +import { PgliteKvStore } from "@fedify/pglite/kv"; +import { testAtomicKvStore } from "@fedify/testing"; +import * as temporal from "@js-temporal/polyfill"; +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const randomName = (): string => + "atomic_" + crypto.randomUUID().replaceAll("-", ""); + +async function fixture( + run: (pg: PGlite, store: PgliteKvStore, table: string) => Promise, +): Promise { + const pg = new PGlite(); + const tableName = randomName(); + const store = new PgliteKvStore(pg, { tableName }); + try { + await store.initialize(); + await run(pg, store, '"' + tableName + '"'); + } finally { + await pg.close(); + } +} + +async function physical(pg: PGlite, table: string): Promise { + return ( + await pg.query( + "SELECT key, value, created::text AS created, ttl::text AS ttl FROM " + + table + + " ORDER BY key", + ) + ).rows; +} + +function code(expected: string): (error: unknown) => boolean { + return (error) => + error instanceof Error && "code" in error && error.code === expected; +} + +async function listed( + store: PgliteKvStore, + prefix?: KvKey, +): Promise { + const values: unknown[] = []; + for await (const value of store.list(prefix)) values.push(value); + return values; +} + +test("PgliteKvStore conforms to AtomicKvStore on one shared owner", async () => { + const pg = new PGlite(); + try { + await testAtomicKvStore( + async () => { + const tableName = randomName(); + const first = new PgliteKvStore(pg, { tableName }); + await first.initialize(); + return { + first, + second: new PgliteKvStore(pg, { tableName, initialized: true }), + dispose: () => first.drop(), + }; + }, + { objectKeyOrder: "ignored" }, + ); + } finally { + await pg.close(); + } +}); + +test("PgliteKvStore rolls back late writes, CAS, deletes and codec failures", () => + fixture(async (pg, store, table) => { + const ttl = Temporal.Duration.from({ hours: 1 }); + await store.setMany([ + { key: ["a"], value: "before-a", options: { ttl } }, + { key: ["b"], value: "before-b", options: { ttl } }, + ]); + const before = await physical(pg, table); + await pg.exec( + "ALTER TABLE " + table + " ADD CHECK (value <> '\"blocked\"'::jsonb)", + ); + await assert.rejects( + store.setMany([ + { key: ["a"], value: "after-a" }, + { key: ["b"], value: "blocked" }, + ]), + code("23514"), + ); + assert.deepStrictEqual(await physical(pg, table), before); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: "before-a", newValue: "after-a" }, + { key: ["b"], expectedValue: "before-b", newValue: "blocked" }, + ]), + code("23514"), + ); + assert.deepStrictEqual(await physical(pg, table), before); + await assert.rejects( + store.setMany([ + { key: ["a"], value: "after-a" }, + { key: ["b"], value: "\0" }, + ]), + ); + assert.deepStrictEqual(await physical(pg, table), before); + + const functionName = randomName(); + try { + await pg.exec( + "CREATE FUNCTION " + + functionName + + "() RETURNS trigger LANGUAGE plpgsql AS $$ " + + "BEGIN IF OLD.key = ARRAY['b'] THEN RAISE EXCEPTION 'injected delete failure'; END IF; RETURN OLD; END $$; " + + "CREATE TRIGGER fail_delete BEFORE DELETE ON " + + table + + " FOR EACH ROW EXECUTE FUNCTION " + + functionName + + "()", + ); + await assert.rejects(store.deleteMany([["a"], ["b"]]), code("P0001")); + assert.deepStrictEqual(await physical(pg, table), before); + await pg.exec("DROP TRIGGER fail_delete ON " + table); + } finally { + await pg.exec("DROP FUNCTION IF EXISTS " + functionName + "() CASCADE"); + } + await store.set(["usable"], true); + assert.strictEqual(await store.get(["usable"]), true); + })); + +test("PgliteKvStore preflights every CAS codec and TTL even after mismatch", () => + fixture(async (pg, store, table) => { + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + const before = await physical(pg, table); + const overflow = Temporal.Duration.from({ years: 300000 }); + const invalid: KvStoreCasEntry[] = [ + { key: ["b"], expectedValue: 2, newValue: "\0" }, + { key: ["b"], expectedValue: "\ud800", newValue: 3 }, + { key: ["b"], expectedValue: 2, newValue: 3, options: { ttl: overflow } }, + { + key: ["b"], + expectedValue: 2, + newValue: undefined, + options: { ttl: Temporal.Duration.from({ seconds: 1 }) }, + }, + { + key: ["b"], + expectedValue: 2, + newValue: 3, + options: { ttl: Temporal.Duration.from({ seconds: -1 }) }, + }, + ]; + for (const entry of invalid) { + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: "wrong", newValue: 100 }, + entry, + ]), + ); + assert.deepStrictEqual(await physical(pg, table), before); + } + for (const write of [ + () => store.set(["overflow"], 1, { ttl: overflow }), + () => + store.setMany([ + { key: ["overflow"], value: 1, options: { ttl: overflow } }, + ]), + () => store.cas(["overflow"], undefined, 1, { ttl: overflow }), + () => + store.casMany([ + { + key: ["overflow"], + expectedValue: undefined, + newValue: 1, + options: { ttl: overflow }, + }, + ]), + ]) { + await assert.rejects(write, code("22008")); + assert.deepStrictEqual(await physical(pg, table), before); + } + })); + +test("PgliteKvStore captures input and rejects malformed keys without aliases", () => + fixture(async (_pg, store) => { + const key: [string] = ["captured"]; + const value = { nested: { value: 1 } }; + const write = store.setMany([{ key, value }]); + key[0] = "mutated"; + value.nested.value = 2; + await write; + assert.deepStrictEqual(await store.get(["captured"]), { + nested: { value: 1 }, + }); + assert.strictEqual(await store.get(["mutated"]), undefined); + const readKey: [string] = ["captured"]; + const read = store.getMany([readKey]); + readKey[0] = "mutated"; + assert.deepStrictEqual(await read, [{ nested: { value: 1 } }]); + const replacement = { value: 3 }; + const change = store.cas( + ["captured"], + { nested: { value: 1 } }, + replacement, + ); + replacement.value = 4; + assert.strictEqual(await change, true); + assert.deepStrictEqual(await store.get(["captured"]), { value: 3 }); + + const invalidKeys = [ + [], + ["\ud800"], + ["\udc00"], + ["a\0b"], + ["a", null], + new Array(2), + ] as unknown as KvKey[]; + for (const bad of invalidKeys) { + for (const operation of [ + () => store.get(bad), + () => store.getMany([bad]), + () => store.set(bad, 1), + () => store.setMany([{ key: bad, value: 1 }]), + () => store.delete(bad), + () => store.deleteMany([bad]), + () => store.cas(bad, undefined, 1), + () => + store.casMany([ + { key: ["captured"], expectedValue: "wrong", newValue: 9 }, + { key: bad, expectedValue: undefined, newValue: 1 }, + ]), + ]) + await assert.rejects(operation, TypeError); + if (bad.length > 0) await assert.rejects(listed(store, bad), TypeError); + } + const keys: KvKey[] = [ + ["a", "b:c"], + ["a:b", "c"], + [""], + ['"\\{}'], + ["한😀"], + ["\ufffd"], + ]; + await store.setMany(keys.map((key, value) => ({ key, value }))); + assert.deepStrictEqual( + await store.getMany(keys), + keys.map((_, index) => index), + ); + assert.deepStrictEqual( + await listed(store, [] as unknown as KvKey), + await listed(store), + ); + })); + +test("PgliteKvStore preserves large compressible keys and rolls back index failures", () => + fixture(async (pg, store, table) => { + const large: KvKey = ["a".repeat(100000)]; + await store.set(large, 1); + assert.strictEqual(await store.get(large), 1); + assert.strictEqual(await store.cas(large, 1, 2), true); + await store.setMany([{ key: large, value: 3 }]); + assert.deepStrictEqual(await store.getMany([large]), [3]); + assert.strictEqual( + await store.casMany([{ key: large, expectedValue: 3, newValue: 4 }]), + true, + ); + await store.deleteMany([large]); + await store.set(large, 5); + await store.delete(large); + assert.strictEqual(await store.get(large), undefined); + + const oversized: KvKey = [ + Array.from(crypto.getRandomValues(new Uint8Array(4096)), (x) => + x.toString(16).padStart(2, "0"), + ).join(""), + ]; + await store.set(["a"], "before", { + ttl: Temporal.Duration.from({ hours: 1 }), + }); + const before = await physical(pg, table); + assert.strictEqual( + await store.casMany([ + { key: ["a"], expectedValue: "wrong", newValue: "after" }, + { key: oversized, expectedValue: undefined, newValue: 1 }, + ]), + false, + ); + assert.deepStrictEqual(await physical(pg, table), before); + await assert.rejects( + store.setMany([ + { key: ["a"], value: "after" }, + { key: oversized, value: 1 }, + ]), + code("54000"), + ); + assert.deepStrictEqual(await physical(pg, table), before); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: "before", newValue: "after" }, + { key: oversized, expectedValue: undefined, newValue: 1 }, + ]), + code("54000"), + ); + assert.deepStrictEqual(await physical(pg, table), before); + })); + +test("PgliteKvStore handles interval rounding, exact expiry and legacy poison rows", () => + fixture(async (pg, store, table) => { + await store.setMany([ + { + key: ["half"], + value: 1, + options: { ttl: Temporal.Duration.from({ nanoseconds: 500 }) }, + }, + { + key: ["up"], + value: 1, + options: { ttl: Temporal.Duration.from({ nanoseconds: 600 }) }, + }, + { + key: ["month"], + value: 1, + options: { ttl: Temporal.Duration.from({ months: 1 }) }, + }, + ]); + const intervals = await pg.query<{ key: KvKey; ttl: string }>( + "SELECT key, ttl::text AS ttl FROM " + table + " ORDER BY key", + ); + assert.deepStrictEqual(intervals.rows, [ + { key: ["half"], ttl: "00:00:00" }, + { key: ["month"], ttl: "1 mon" }, + { key: ["up"], ttl: "00:00:00.000001" }, + ]); + assert.strictEqual(await store.get(["half"]), undefined); + await pg.query( + "INSERT INTO " + + table + + " (key,value,created,ttl) VALUES ($1::text[], '1', NULL, '1 day')", + [["legacy-null"]], + ); + assert.strictEqual(await store.get(["legacy-null"]), undefined); + assert.strictEqual(await store.cas(["legacy-null"], undefined, 2), true); + assert.strictEqual(await store.get(["legacy-null"]), 2); + + await pg.exec( + "INSERT INTO " + + table + + " (key,value,ttl) VALUES (ARRAY['poison-a'], '1', '300000 years'), (ARRAY['poison-b'], '2', '300000 years')", + ); + const before = await physical(pg, table); + await assert.rejects(store.get(["poison-a"]), code("22008")); + await assert.rejects(store.getMany([["poison-b"]]), code("22008")); + await assert.rejects(store.set(["cleanup-failure"], 1), code("22008")); + assert.deepStrictEqual(await physical(pg, table), before); + await assert.rejects(store.delete(["poison-a"]), code("22008")); + assert.deepStrictEqual(await physical(pg, table), before); + await store.deleteMany([["poison-a"], ["poison-b"]]); + await store.set(["recovered"], true); + assert.strictEqual(await store.get(["recovered"]), true); + })); + +test("PgliteKvStore queues absent-key CAS, snapshots and ordinary writes behind the owner", () => + fixture(async (pg, store, table) => { + const second = new PgliteKvStore(pg, { + tableName: table.slice(1, -1), + initialized: true, + }); + let enter!: () => void; + let release!: () => void; + const entered = new Promise((resolve) => { + enter = resolve; + }); + const gate = new Promise((resolve) => { + release = resolve; + }); + const transaction = pg.transaction(async (tx) => { + await tx.query( + "INSERT INTO " + table + " (key,value) VALUES (ARRAY['a'],'1')", + ); + enter(); + await gate; + await tx.query( + "INSERT INTO " + + table + + " (key,value) VALUES (ARRAY['b'],'1'), (ARRAY['absent'],'1')", + ); + }); + await entered; + let settled = false; + const reads = store.getMany([["a"], ["b"]]); + const condition = second.cas(["absent"], undefined, 2); + const ordinary = second.set(["ordinary"], true); + const pending = [reads, condition, ordinary]; + for (const operation of pending) + void operation.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + try { + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.strictEqual(settled, false); + } finally { + release(); + await Promise.allSettled([transaction, ...pending]); + } + await transaction; + assert.deepStrictEqual(await reads, [1, 1]); + assert.strictEqual(await condition, false); + await ordinary; + assert.strictEqual(await store.get(["ordinary"]), true); + })); + +function instrumentTransactions( + pg: PGlite, + before: (tx: Transaction) => Promise, + afterQuery: (tx: Transaction, sql: string) => Promise, +): PGliteInterface { + return { + waitReady: pg.waitReady, + query: pg.query.bind(pg), + transaction: (callback: (tx: Transaction) => Promise): Promise => + pg.transaction(async (tx) => { + await before(tx); + return await callback({ + ...tx, + query: async ( + sql: string, + params?: Parameters[1], + options?: QueryOptions, + ) => { + const result = await tx.query(sql, params, options); + await afterQuery(tx, sql); + return result; + }, + }); + }), + } as PGliteInterface; +} + +test("PgliteKvStore uses one expiry cutoff across delayed snapshot statements", () => + fixture(async (pg, store, table) => { + let delayed = false; + const observed = instrumentTransactions( + pg, + async (tx) => { + await tx.query( + "INSERT INTO " + + table + + " (key,value,created,ttl) VALUES " + + "(ARRAY['a'],'1',CURRENT_TIMESTAMP,'50 milliseconds'), " + + "(ARRAY['b'],'1',CURRENT_TIMESTAMP,'50 milliseconds'), " + + "(ARRAY['boundary'],'1',CURRENT_TIMESTAMP,'0 seconds')", + ); + }, + async (tx, sql) => { + if (!delayed && sql.includes("SELECT value FROM")) { + delayed = true; + await tx.query("SELECT pg_sleep(0.1)"); + } + }, + ); + const snapshot = new PgliteKvStore(observed, { + tableName: table.slice(1, -1), + initialized: true, + }); + assert.deepStrictEqual( + await snapshot.getMany([["a"], ["b"], ["a"], ["boundary"]]), + [1, 1, 1, undefined], + ); + assert.strictEqual(delayed, true); + assert.deepStrictEqual(await store.getMany([["a"], ["b"]]), [ + undefined, + undefined, + ]); + })); + +test("PgliteKvStore batch changes and TTLs survive clean filesystem reopen", async () => { + const dataDir = await mkdtemp(join(tmpdir(), "fedify-pglite-atomic-")); + const tableName = randomName(); + let pg: PGlite | undefined; + try { + pg = new PGlite(dataDir); + let store = new PgliteKvStore(pg, { tableName }); + await store.setMany([ + { key: ["a"], value: { version: 1 } }, + { key: ["b"], value: null }, + { + key: ["ttl"], + value: true, + options: { ttl: Temporal.Duration.from({ days: 1 }) }, + }, + ]); + assert.strictEqual( + await store.casMany([ + { key: ["a"], expectedValue: { version: 1 }, newValue: { version: 2 } }, + { key: ["b"], expectedValue: null, newValue: undefined }, + { key: ["c"], expectedValue: undefined, newValue: "created" }, + ]), + true, + ); + await pg.close(); + pg = new PGlite(dataDir); + store = new PgliteKvStore(pg, { tableName, initialized: true }); + assert.deepStrictEqual( + await store.getMany([["a"], ["b"], ["c"], ["ttl"]]), + [{ version: 2 }, undefined, "created", true], + ); + const rows = await pg.query<{ ttl: string }>( + 'SELECT ttl::text AS ttl FROM "' + tableName + '" WHERE key = $1::text[]', + [["ttl"]], + ); + assert.strictEqual(rows.rows[0].ttl, "1 day"); + await store.drop(); + } finally { + if (pg != null && !pg.closed) await pg.close(); + await rm(dataDir, { recursive: true, force: true }); + } +}); + +test("PgliteKvStore awaits post-commit filesystem sync and does not replay failures", async () => { + class ObservedMemoryFS extends MemoryFS { + onSync: () => Promise = async () => {}; + override async syncToFs(): Promise { + await this.onSync(); + } + } + const fs = new ObservedMemoryFS(); + const pg = new PGlite({ fs, relaxedDurability: false }); + const store = new PgliteKvStore(pg, { tableName: randomName() }); + const events: string[] = []; + const transaction = pg.transaction.bind(pg); + pg.transaction = async ( + callback: (tx: Transaction) => Promise, + ): Promise => { + const result = await transaction(callback); + events.push("committed"); + return result; + }; + try { + await store.initialize(); + const operations = [ + () => store.set(["single"], 1), + () => store.cas(["single"], 1, 2), + () => store.delete(["single"]), + () => store.setMany([{ key: ["batch"], value: 1 }]), + () => store.casMany([{ key: ["batch"], expectedValue: 1, newValue: 2 }]), + () => store.deleteMany([["batch"]]), + ]; + for (const operation of operations) { + events.length = 0; + let release!: () => void; + let enter!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const entered = new Promise((resolve) => { + enter = resolve; + }); + fs.onSync = async () => { + // BEGIN also syncs in 0.5.8; only pause the sync after COMMIT. + if (!events.includes("committed")) return; + events.push("sync"); + enter(); + await gate; + }; + let settled = false; + const pending = operation().then(() => { + settled = true; + }); + try { + // Also finish if the regression returns without a post-commit sync. + await Promise.race([entered, pending]); + assert.deepStrictEqual(events, ["committed", "sync"]); + assert.strictEqual(settled, false); + } finally { + release(); + await pending; + fs.onSync = async () => {}; + } + assert.strictEqual(settled, true); + } + + events.length = 0; + const failure = new Error("filesystem synchronization failed"); + fs.onSync = () => + events.includes("committed") + ? Promise.reject(failure) + : Promise.resolve(); + await assert.rejects( + store.set(["committed-despite-rejection"], 42), + (error) => error === failure, + ); + fs.onSync = async () => {}; + assert.deepStrictEqual(events, ["committed"]); + assert.strictEqual(await store.get(["committed-despite-rejection"]), 42); + await store.set(["connection-usable"], true); + assert.strictEqual(await store.get(["connection-usable"]), true); + } finally { + fs.onSync = async () => {}; + await pg.close(); + } +}); diff --git a/packages/pglite/src/kv.ts b/packages/pglite/src/kv.ts index 87b00f92c..80fbf6a20 100644 --- a/packages/pglite/src/kv.ts +++ b/packages/pglite/src/kv.ts @@ -1,9 +1,13 @@ /// -import type { PGliteInterface } from "@electric-sql/pglite"; +// Statements must stay sequential within the owner transaction. +/* oxlint-disable eslint/no-await-in-loop */ +import type { PGliteInterface, Transaction } from "@electric-sql/pglite"; import type { KvKey, - KvStore, + AtomicKvStore, + KvStoreCasEntry, + KvStoreSetEntry, KvStoreListEntry, KvStoreSetOptions, } from "@fedify/fedify"; @@ -31,6 +35,55 @@ function serializeJson(value: unknown): string { return serialized; } +function captureKey(key: KvKey): KvKey { + const parts = captureArray(key); + if ( + parts.length === 0 || + parts.some( + (part) => + typeof part !== "string" || !part.isWellFormed() || part.includes("\0"), + ) + ) { + throw new TypeError( + "Keys must be nonempty arrays of well-formed strings without NUL.", + ); + } + return parts as [string, ...string[]]; +} + +function captureArray(items: readonly T[]): T[] { + if (!Array.isArray(items)) throw new TypeError("A batch must be an array."); + for (let i = 0; i < items.length; i++) { + if (!Object.hasOwn(items, i)) + throw new TypeError("Sparse arrays are not supported."); + } + return Array.from(items); +} + +function captureKeys(keys: readonly KvKey[], unique = false): KvKey[] { + const captured = captureArray(keys).map(captureKey); + if ( + unique && + new Set(captured.map((key) => JSON.stringify(key))).size !== captured.length + ) { + throw new TypeError("Duplicate keys in an atomic mutation."); + } + return captured; +} + +function serializeTtl(options?: KvStoreSetOptions): string | null { + if (options?.ttl == null) return null; + const ttl = Temporal.Duration.from(options.ttl); + if (ttl.sign < 0) throw new RangeError("The TTL must not be negative."); + return ttl.toString(); +} + +interface PreparedWrite { + readonly key: KvKey; + readonly value: string; + readonly ttl: string | null; +} + /** * Options for the PGlite key–value store. */ @@ -54,6 +107,33 @@ export interface PgliteKvStoreOptions { /** * A key–value store backed by an embedded PGlite database. * + * Since Fedify 2.5.0, this implements {@link AtomicKvStore}. All keys in one + * table share an atomic scope when wrappers share one normal PGlite owner, + * using its standard text-array and JSONB codecs. Do not invoke store methods + * inside the owner's transaction callback or concurrently bypass its normal + * query/transaction queue with raw protocol or manual transaction control. + * Worker proxies and independent owners sharing a directory are outside this + * scope. + * + * Values use JSON.stringify() and JSONB equality: object member order is + * ignored, array order matters, and JSON null differs from absence. Keys must + * be nonempty arrays of well-formed strings without NUL. Unsupported JSONB + * values, duplicate mutation keys, sparse arrays, negative TTLs, timestamp + * overflow, and TTLs on CAS deletion reject. Omitting a replacement TTL clears + * expiry. Zero TTL expires immediately; intervals round to microseconds on a + * host clock that may resolve only milliseconds. Calendar units use the + * session time zone. Each operation uses one transaction-start expiry cutoff. + * + * No fixed batch-entry limit is imposed; large batches occupy the shared + * owner and consume its finite memory. Engine insertion limits depend on + * compression and are checked when writes are attempted. A mismatch does not + * prove hypothetical insertability. Statement failures reject and roll back; + * commit or post-commit synchronization failures may leave committed changes + * despite rejection and must not trigger replay. Successful mutation + * transactions await the owner's syncToFs() policy outside the transaction. + * Atomicity does not promise crash durability; see the package README for + * supported storage and the owner's durability policy. + * * The caller owns the PGlite instance and is responsible for closing it after * every operation using the store has completed. Use only one PGlite instance * for a data directory within a runtime isolate. @@ -75,7 +155,7 @@ export interface PgliteKvStoreOptions { * ``` * @since 2.4.0 */ -export class PgliteKvStore implements KvStore { +export class PgliteKvStore implements AtomicKvStore { readonly #pg: PGliteInterface; readonly #tableName: string; readonly #quotedTableName: string; @@ -94,143 +174,209 @@ export class PgliteKvStore implements KvStore { this.#initialized = options.initialized ?? false; } - async #expire(): Promise { - await this.#pg.query(` + async #expire(tx: Transaction): Promise { + await tx.query(` DELETE FROM ${this.#quotedTableName} WHERE ttl IS NOT NULL AND created + ttl < CURRENT_TIMESTAMP; `); } + async #write(tx: Transaction, entry: PreparedWrite): Promise { + await tx.query( + ` + INSERT INTO ${this.#quotedTableName} (key, value, created, ttl) + VALUES ($1::text[], $2::text::jsonb, CURRENT_TIMESTAMP, $3::text::interval) + ON CONFLICT (key) + DO UPDATE SET value = EXCLUDED.value, + created = EXCLUDED.created, ttl = EXCLUDED.ttl; + `, + [entry.key, entry.value, entry.ttl], + ); + } + + async #remove(tx: Transaction, key: KvKey): Promise { + await tx.query( + `DELETE FROM ${this.#quotedTableName} WHERE key = $1::text[];`, + [key], + ); + } + + /** {@inheritDoc AtomicKvStore.get} */ async get(key: KvKey): Promise { + const captured = captureKey(key); await this.initialize(); const result = await this.#pg.query<{ value: T }>( ` - SELECT value - FROM ${this.#quotedTableName} + SELECT value FROM ${this.#quotedTableName} WHERE key = $1::text[] AND (ttl IS NULL OR created + ttl > CURRENT_TIMESTAMP); - `, - [key], + `, + [captured], ); return result.rows[0]?.value; } - async set( - key: KvKey, - value: unknown, - options?: KvStoreSetOptions, - ): Promise { - await this.initialize(); - const ttl = - options?.ttl == null - ? null - : Temporal.Duration.from(options.ttl).toString(); - await this.#pg.query( - ` - INSERT INTO ${this.#quotedTableName} (key, value, ttl) - VALUES ($1::text[], $2::text::jsonb, $3::text::interval) - ON CONFLICT (key) - DO UPDATE SET - value = EXCLUDED.value, - created = CURRENT_TIMESTAMP, - ttl = EXCLUDED.ttl; - `, - [key, serializeJson(value), ttl], - ); - await this.#expire(); + /** {@inheritDoc AtomicKvStore.set} */ + set(key: KvKey, value: unknown, options?: KvStoreSetOptions): Promise { + return this.setMany([{ key, value, options }]); } - async delete(key: KvKey): Promise { - await this.initialize(); - await this.#pg.query( - ` - DELETE FROM ${this.#quotedTableName} - WHERE key = $1::text[]; - `, - [key], - ); - await this.#expire(); + /** {@inheritDoc AtomicKvStore.delete} */ + delete(key: KvKey): Promise { + return this.deleteMany([key]); } - /** {@inheritDoc KvStore.cas} */ - async cas( + /** {@inheritDoc AtomicKvStore.cas} */ + cas( key: KvKey, expectedValue: unknown, newValue: unknown, options?: KvStoreSetOptions, ): Promise { + return this.casMany([{ key, expectedValue, newValue, options }]); + } + + /** + * {@inheritDoc AtomicKvStore.getMany} + * @since 2.5.0 + */ + async getMany( + keys: readonly KvKey[], + ): Promise<(T | undefined)[]> { + const captured = captureKeys(keys); + if (captured.length === 0) return []; await this.initialize(); - const ttl = - options?.ttl == null - ? null - : Temporal.Duration.from(options.ttl).toString(); + return await this.#pg.transaction(async (tx) => { + const values: (T | undefined)[] = []; + for (const key of captured) { + const result = await tx.query<{ value: T }>( + `SELECT value FROM ${this.#quotedTableName} + WHERE key = $1::text[] + AND (ttl IS NULL OR created + ttl > CURRENT_TIMESTAMP);`, + [key], + ); + values.push(result.rows[0]?.value); + } + return values; + }); + } - if (expectedValue === undefined && newValue === undefined) { - const matched = (await this.get(key)) === undefined; - await this.#expire(); - return matched; - } + /** + * {@inheritDoc AtomicKvStore.setMany} + * @since 2.5.0 + */ + async setMany(entries: readonly KvStoreSetEntry[]): Promise { + const captured = captureArray(entries); + const keys = captureKeys( + captured.map((entry) => entry.key), + true, + ); + const writes = captured.map((entry, index): PreparedWrite => ({ + key: keys[index], + value: serializeJson(entry.value), + ttl: serializeTtl(entry.options), + })); + if (writes.length === 0) return; + await this.initialize(); + await this.#pg.transaction(async (tx) => { + for (const entry of writes) { + await tx.query( + "SELECT (CURRENT_TIMESTAMP + $1::text::interval) IS NULL;", + [entry.ttl], + ); + } + for (const entry of writes) await this.#write(tx, entry); + await this.#expire(tx); + }); + // PGlite 0.5.8 does not sync after COMMIT; keep the owner's policy. + await this.#pg.syncToFs(); + } - let rowCount: number; - if (expectedValue === undefined) { - const result = await this.#pg.query( - ` - INSERT INTO ${this.#quotedTableName} AS existing - (key, value, created, ttl) - VALUES ( - $1::text[], - $2::text::jsonb, - CURRENT_TIMESTAMP, - $3::text::interval - ) - ON CONFLICT (key) - DO UPDATE SET - value = EXCLUDED.value, - created = EXCLUDED.created, - ttl = EXCLUDED.ttl - WHERE existing.ttl IS NOT NULL - AND existing.created + existing.ttl <= CURRENT_TIMESTAMP - RETURNING key; - `, - [key, serializeJson(newValue), ttl], - ); - rowCount = result.rows.length; - } else if (newValue === undefined) { - const result = await this.#pg.query( - ` - DELETE FROM ${this.#quotedTableName} - WHERE key = $1::text[] - AND (ttl IS NULL OR created + ttl > CURRENT_TIMESTAMP) - AND value = $2::text::jsonb - RETURNING key; - `, - [key, serializeJson(expectedValue)], - ); - rowCount = result.rows.length; - } else { - const result = await this.#pg.query( - ` - UPDATE ${this.#quotedTableName} - SET - value = $3::text::jsonb, - created = CURRENT_TIMESTAMP, - ttl = $4::text::interval - WHERE key = $1::text[] - AND (ttl IS NULL OR created + ttl > CURRENT_TIMESTAMP) - AND value = $2::text::jsonb - RETURNING key; - `, - [key, serializeJson(expectedValue), serializeJson(newValue), ttl], - ); - rowCount = result.rows.length; - } + /** + * {@inheritDoc AtomicKvStore.deleteMany} + * @since 2.5.0 + */ + async deleteMany(keys: readonly KvKey[]): Promise { + const captured = captureKeys(keys, true); + if (captured.length === 0) return; + await this.initialize(); + await this.#pg.transaction(async (tx) => { + for (const key of captured) await this.#remove(tx, key); + await this.#expire(tx); + }); + // PGlite 0.5.8 does not sync after COMMIT; keep the owner's policy. + await this.#pg.syncToFs(); + } - await this.#expire(); - return rowCount > 0; + /** + * {@inheritDoc AtomicKvStore.casMany} + * @since 2.5.0 + */ + async casMany(entries: readonly KvStoreCasEntry[]): Promise { + const captured = captureArray(entries); + const keys = captureKeys( + captured.map((entry) => entry.key), + true, + ); + const changes = captured.map((entry, index) => { + if (entry.newValue === undefined && entry.options?.ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } + return { + key: keys[index], + expected: + entry.expectedValue === undefined + ? undefined + : serializeJson(entry.expectedValue), + value: + entry.newValue === undefined + ? undefined + : serializeJson(entry.newValue), + ttl: serializeTtl(entry.options), + }; + }); + if (changes.length === 0) return true; + await this.initialize(); + const committed = await this.#pg.transaction(async (tx) => { + // Validate every backend codec and expiry before a mismatch can return. + // Let SQL errors escape so PGlite rolls the transaction back. + for (const entry of changes) { + await tx.query( + `SELECT ($1::text::jsonb) IS NULL, ($2::text::jsonb) IS NULL, + (CURRENT_TIMESTAMP + $3::text::interval) IS NULL;`, + [entry.expected ?? null, entry.value ?? null, entry.ttl], + ); + } + for (const entry of changes) { + const result = await tx.query<{ matched: boolean }>( + `SELECT ${entry.expected === undefined ? "NOT " : ""}EXISTS ( + SELECT 1 FROM ${this.#quotedTableName} + WHERE key = $1::text[] + AND (ttl IS NULL OR created + ttl > CURRENT_TIMESTAMP) + ${entry.expected === undefined ? "" : "AND value = $2::text::jsonb"} + ) AS matched;`, + entry.expected === undefined + ? [entry.key] + : [entry.key, entry.expected], + ); + if (!result.rows[0].matched) return false; + } + for (const entry of changes) { + if (entry.value === undefined) await this.#remove(tx, entry.key); + else await this.#write(tx, { ...entry, value: entry.value }); + } + await this.#expire(tx); + return true; + }); + await this.#pg.syncToFs(); + return committed; } - /** {@inheritDoc KvStore.list} */ + /** {@inheritDoc AtomicKvStore.list} */ async *list(prefix?: KvKey): AsyncIterable { + if (prefix != null && (!Array.isArray(prefix) || prefix.length > 0)) { + prefix = captureKey(prefix); + } await this.initialize(); const result = From ba14fc04c70943323c66be4a8a2aed0d83a190cb Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:06:30 +0900 Subject: [PATCH 2/2] Clarify CAS mismatch synchronization failures Keep the established PGlite owner synchronization policy on read-only CAS mismatches. Backend synchronization errors reject rather than return false, without changing values or TTLs on that path. Document read-only completion separately from committed mutations and cover failure and healthy-backend results for single and batch CAS. https://github.com/fedify-dev/fedify/pull/1431#discussion_r4554686956 Assisted-by: Codex:gpt-6.1-sol --- CHANGES.md | 21 +++++++++++++++ changes.d/pglite/atomic-kv-store.md | 5 ++-- claude-plugin/skills/fedify/SKILL.md | 4 +-- docs/manual/kv.md | 8 +++--- packages/pglite/README.md | 13 ++++++---- packages/pglite/src/atomic-kv.test.ts | 37 ++++++++++++++++++++++++--- packages/pglite/src/kv.ts | 4 ++- 7 files changed, 76 insertions(+), 16 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index de985a020..012ee8b83 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -397,6 +397,27 @@ To be released. [#1287]: https://github.com/fedify-dev/fedify/issues/1287 [#1357]: https://github.com/fedify-dev/fedify/pull/1357 +### @fedify/pglite + + - Added atomic snapshots and multi-key writes, deletions, and conditional + updates to `PgliteKvStore` through `AtomicKvStore`. Wrappers sharing one + PGlite owner can commit related keys together with JSONB comparison and + per-entry TTLs. [[#1361], [#1366], [#1431]] + + Malformed keys, negative or overflowing TTLs, and TTLs on CAS deletions + now reject before mutation. Mutation cleanup shares the transaction, + so cleanup failures roll back writes; legacy invalid TTL rows require + removal together or direct SQL remediation. + + Existing single-key mutations and the new batch mutations await the + PGlite owner's filesystem synchronization policy after a successful + transaction, including read-only CAS mismatches. If synchronization fails, + the operation rejects instead of returning `false`; any committed mutations + remain applied, so retrying blindly can repeat a change. + +[#1366]: https://github.com/fedify-dev/fedify/issues/1366 +[#1431]: https://github.com/fedify-dev/fedify/pull/1431 + ### @fedify/postgres - Added an expression index for `PostgresKvStore` expiry cleanup, avoiding a diff --git a/changes.d/pglite/atomic-kv-store.md b/changes.d/pglite/atomic-kv-store.md index 2e6c18ffd..b68801e3a 100644 --- a/changes.d/pglite/atomic-kv-store.md +++ b/changes.d/pglite/atomic-kv-store.md @@ -16,5 +16,6 @@ links: Existing single-key mutations and the new batch mutations await the PGlite owner's filesystem synchronization policy after a successful - transaction. If synchronization fails, the operation rejects after its - changes have committed; retrying blindly can repeat the change. + transaction, including read-only CAS mismatches. If synchronization fails, + the operation rejects instead of returning `false`; any committed mutations + remain applied, so retrying blindly can repeat a change. diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index 7d5fdac4b..b7f3a6665 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -506,8 +506,8 @@ 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. `WorkersD1KvStore` supports +single-key CAS; its atomic extension is tracked in #1379. 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 ddceb8483..672166774 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -496,8 +496,10 @@ zone. There is no fixed batch-entry cap. Engine key insertion limits depend on compression; a mismatch does not prove that unattempted writes would fit. Statement failures roll back the batch; commit or post-commit synchronization failures can reject after changes have committed. Long batches occupy the -shared instance. Mutations await the owner's `syncToFs()` after commit; atomic -execution does not establish crash durability. See the +shared instance. Nonempty mutation calls await the owner's `syncToFs()` after +transaction completion, including read-only CAS mismatches. Backend failures +reject instead of returning `false`; any committed mutations remain applied. +Atomic execution does not establish crash durability. See the [package's atomic-operation guidance] for codecs, legacy-row recovery, and the owner's synchronization policy. @@ -1016,7 +1018,7 @@ The linked issues track separate adapter implementations: | `MemoryKvStore` | Yes | Development and testing only; one instance. | | `PostgresKvStore` | No | Planned in [#1364]. | | `RedisKvStore` | No | Planned in [#1365]. | -| `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | +| `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | | `SqliteKvStore` | Yes | One configured table/database; writable handles, JSON values; see [SQLite scope](#sqlitekvstore). | | `MysqlKvStore` | No | Planned in [#1368]. | | `DenoKvStore` | No | No atomic adapter child issue; deprecated since 2.5.0. | diff --git a/packages/pglite/README.md b/packages/pglite/README.md index 57d6385ed..b29f3231a 100644 --- a/packages/pglite/README.md +++ b/packages/pglite/README.md @@ -114,13 +114,16 @@ are outside this scope. Host-filesystem storage supports clean close/reopen; memory-only instances remain ephemeral. After a mutation transaction completes, the adapter awaits the owner's `syncToFs()` outside the transaction, preserving the synchronization policy previously used by `set()`, `delete()`, and `cas()`. +This includes a nonempty CAS call whose comparisons mismatch: its read-only +transaction still completes, and a backend synchronization failure rejects +instead of returning `false`. PGlite 0.5.8 does not perform that post-commit synchronization itself. The owner's filesystem and `relaxedDurability` setting determine what this -call guarantees. A synchronization failure rejects after the changes have -committed; it cannot roll them back. A lost response or commit failure can -also leave commit status unknown. Do not -automatically replay an uncertain change; use a non-reused version or an -operation receipt when the application needs it. +call guarantees. A synchronization failure rejects after transaction +completion; any committed mutations remain applied. A lost response or commit +failure can also leave commit status unknown. Do not automatically replay an +uncertain change; use a non-reused version or an operation receipt when the +application needs it. Mutation housekeeping is part of the same transaction. A legacy row with an overflowing TTL can make expiry cleanup fail; other mutations then roll back diff --git a/packages/pglite/src/atomic-kv.test.ts b/packages/pglite/src/atomic-kv.test.ts index f1873c5b7..889913dcf 100644 --- a/packages/pglite/src/atomic-kv.test.ts +++ b/packages/pglite/src/atomic-kv.test.ts @@ -543,7 +543,8 @@ test("PgliteKvStore awaits post-commit filesystem sync and does not replay failu } const fs = new ObservedMemoryFS(); const pg = new PGlite({ fs, relaxedDurability: false }); - const store = new PgliteKvStore(pg, { tableName: randomName() }); + const tableName = randomName(); + const store = new PgliteKvStore(pg, { tableName }); const events: string[] = []; const transaction = pg.transaction.bind(pg); pg.transaction = async ( @@ -597,12 +598,42 @@ test("PgliteKvStore awaits post-commit filesystem sync and does not replay failu assert.strictEqual(settled, true); } - events.length = 0; const failure = new Error("filesystem synchronization failed"); - fs.onSync = () => + const failAfterCommit = () => events.includes("committed") ? Promise.reject(failure) : Promise.resolve(); + await store.set( + ["retained"], + { version: 1 }, + { + ttl: Temporal.Duration.from({ hours: 1 }), + }, + ); + const beforeMismatch = await physical(pg, '"' + tableName + '"'); + const mismatches = [ + () => store.cas(["retained"], { version: 0 }, { version: 2 }), + () => + store.casMany([ + { key: ["retained"], expectedValue: { version: 0 }, newValue: 2 }, + { key: ["untouched"], expectedValue: undefined, newValue: 3 }, + ]), + ]; + for (const operation of mismatches) { + events.length = 0; + fs.onSync = failAfterCommit; + await assert.rejects(operation(), (error) => error === failure); + fs.onSync = async () => {}; + assert.deepStrictEqual(events, ["committed"]); + assert.deepStrictEqual( + await physical(pg, '"' + tableName + '"'), + beforeMismatch, + ); + assert.strictEqual(await operation(), false); + } + + events.length = 0; + fs.onSync = failAfterCommit; await assert.rejects( store.set(["committed-despite-rejection"], 42), (error) => error === failure, diff --git a/packages/pglite/src/kv.ts b/packages/pglite/src/kv.ts index 80fbf6a20..8899c3f82 100644 --- a/packages/pglite/src/kv.ts +++ b/packages/pglite/src/kv.ts @@ -130,7 +130,9 @@ export interface PgliteKvStoreOptions { * prove hypothetical insertability. Statement failures reject and roll back; * commit or post-commit synchronization failures may leave committed changes * despite rejection and must not trigger replay. Successful mutation - * transactions await the owner's syncToFs() policy outside the transaction. + * transactions await the owner's syncToFs() policy outside the transaction, + * including read-only CAS mismatches. Backend synchronization failures reject + * rather than returning false. * Atomicity does not promise crash durability; see the package README for * supported storage and the owner's durability policy. *