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 new file mode 100644 index 000000000..b68801e3a --- /dev/null +++ b/changes.d/pglite/atomic-kv-store.md @@ -0,0 +1,21 @@ +--- +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, 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 05b4409bd..b7f3a6665 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..672166774 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -480,7 +480,31 @@ 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. 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. + [`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 +1008,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 +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` | 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 +1029,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 +1637,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..b29f3231a 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,86 @@ 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()`. +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 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 +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..889913dcf --- /dev/null +++ b/packages/pglite/src/atomic-kv.test.ts @@ -0,0 +1,650 @@ +// 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 tableName = randomName(); + const store = new PgliteKvStore(pg, { tableName }); + 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); + } + + const failure = new Error("filesystem synchronization failed"); + 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, + ); + 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..8899c3f82 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,35 @@ 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, + * 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. + * * 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 +157,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 +176,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 =