diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..4c15c3f8f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -429,10 +429,23 @@ To be released. the index exists, writes whose expiration overflows PostgreSQL's timestamp range fail atomically, including writes from older versions. [[#1309], [#1334]] + - Added atomic multi-key snapshots, writes, deletions, and conditional updates + to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional + batches and multi-key deletions require pooled or sequentially used + `READ COMMITTED` transaction clients; reserved clients reject these + scopes. Nonempty batches reject active row-level security to prevent + policies from filtering individual mutations. Mutating single-key `cas()` + calls also serialize table writes and roll back trigger-skipped matching + mutations instead of returning `false`. `PostgresKvStore.cas()` now + rejects a `ttl` option when deleting (`newValue` is `undefined`), where it + previously ignored the TTL. Omit `ttl` when deleting. + [[#1361], [#1364], [#1429]] [index migration guide]: https://fedify.dev/manual/kv#postgresql-expiry-index [#1309]: https://github.com/fedify-dev/fedify/issues/1309 [#1334]: https://github.com/fedify-dev/fedify/pull/1334 +[#1364]: https://github.com/fedify-dev/fedify/issues/1364 +[#1429]: https://github.com/fedify-dev/fedify/pull/1429 ### @fedify/sqlite diff --git a/changes.d/postgres/atomic-kv-store.md b/changes.d/postgres/atomic-kv-store.md new file mode 100644 index 000000000..f57686cca --- /dev/null +++ b/changes.d/postgres/atomic-kv-store.md @@ -0,0 +1,17 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1364': https://github.com/fedify-dev/fedify/issues/1364 + '#1429': https://github.com/fedify-dev/fedify/pull/1429 +--- + - Added atomic multi-key snapshots, writes, deletions, and conditional updates + to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional + batches and multi-key deletions require pooled or sequentially used + `READ COMMITTED` transaction clients; reserved clients reject these + scopes. Nonempty batches reject active row-level security to prevent + policies from filtering individual mutations. Mutating single-key `cas()` + calls also serialize table writes and roll back trigger-skipped matching + mutations instead of returning `false`. `PostgresKvStore.cas()` now + rejects a `ttl` option when deleting (`newValue` is `undefined`), where it + previously ignored the TTL. Omit `ttl` when deleting. + [[#1361], [#1364], [#1429]] diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b7f3a6665..897716916 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -482,8 +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`, `SqliteKvStore`, and `PgliteKvStore` implement the full atomic -interface. +`MemoryKvStore`, `PostgresKvStore`, `SqliteKvStore`, and `PgliteKvStore` +implement the atomic interface in their documented scopes. 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 @@ -504,8 +504,8 @@ engine insertion failures roll back, while mismatch does not prove proposed writes would fit. Do not infer crash durability, browser persistence, or worker-proxy support. Read the PGlite package README's synchronization policy. -The PostgreSQL, Redis, and MySQL atomic extensions are tracked separately in -issues #1364, #1365, and #1368; Deno KV and Netlify Blobs have no atomic adapter +The Redis and MySQL atomic extensions are tracked separately in +issues #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 @@ -519,6 +519,30 @@ the store. Production codecs can support different values and comparison rules. Cache cleanup examples apply only to cache prefixes; authoritative state such as budgets and replay cutoffs needs its own retention policy. +Mutating single-key PostgreSQL `cas()` takes a table write lock and rolls back +trigger-skipped matching mutations instead of returning `false`. It supports +reserved clients using a per-call transaction or a savepoint in an existing +manual transaction. Caller-owned connections require exclusive sequential use; +CAS locks remain until outer commit even on mismatch. + +For PostgreSQL atomic operations, use one ordinary table or declarative +partition and the default JSON codec (`jsonb` equality ignores object member +order). Nonempty `casMany()` and multi-key `deleteMany()` require a pool or +READ COMMITTED transaction-scoped Postgres.js client; reserved clients reject +those scopes. Caller transactions require exclusive sequential use during +each atomic call, across other wrappers and raw SQL too. Nested savepoints +do not isolate concurrent calls on one outer transaction. Successful caller +batches are provisional and retain locks until outer commit. Table-wide write +locks protect absent keys but serialize writers and can interrupt autovacuum; +lock errors reject without automatic retry. After lock-upgrade deadlock, +roll back the outer transaction before retrying. Active row-level security, +ordinary inheritance and irreversible JSON/value transforms are unsupported for +batches. TTLs use server time and PostgreSQL calendar intervals at microsecond +precision. Backend storage constraints may fail an attempted write; a mismatch +performs no insertion. Atomicity does not override unlogged/WAL/replica +durability. Consult the [PostgreSQL atomic scope] before choosing a client or +codec. + For `PostgresKvStore`, existing populated tables and externally managed DDL (`initialized: true`) need the [expiry expression index] to accelerate cleanup. Run concurrent index creation outside a transaction, followed by `ANALYZE`. @@ -539,6 +563,7 @@ index too. Keep the original TTL columns and calendar semantics. See and . [atomic support matrix]: https://fedify.dev/manual/kv.md#current-support +[PostgreSQL atomic scope]: https://fedify.dev/manual/kv#postgresql-atomic-operations [expiry expression index]: https://fedify.dev/manual/kv.md#postgresql-expiry-index diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 672166774..0ad4d1f6d 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -318,6 +318,81 @@ const federation = createFederation({ }); ~~~~ +#### PostgreSQL atomic operations + +*This API is available since Fedify 2.5.0.* + +`PostgresKvStore` implements `AtomicKvStore` within one configured ordinary +table or declarative partition, using the existing schema and +PostgreSQL-representable Unicode string keys. Ordinary inheritance parents and +children are unsupported for nonempty batches. Tables where row-level security +is active for the current role are also rejected, because policies can filter +individual mutations. `getMany()` returns one +input-ordered snapshot; `setMany()` writes one batch; `casMany()` compares all +expectations before applying any changes. JSON `null` remains distinct from +missing or expired `undefined`. + +Nonempty `casMany()` and `deleteMany()` with more than one key require a pool +or a `READ COMMITTED` transaction-scoped Postgres.js client. Reserved clients +reject those scopes before SQL; they still support ordinary methods, the +single-statement batches, single-key deletion, and empty batches. Unsupported +caller isolation rejects. Method presence does not certify a client's scope. + +Caller-owned transactions must be used exclusively and sequentially throughout +each atomic call, including initialization, every store wrapper, and raw SQL. +Await each call before sending another command on that transaction: nested +savepoints do not isolate overlapping calls. Success remains provisional until +the outer commit, and successful batch locks remain held until then. + +Mutating forms of `cas()` acquire the table write lock and use one server +timestamp for comparison and the replacement's TTL origin. A `cas()` from +`undefined` to `undefined` stays read-only. If a trigger skips a matching +write, the call rejects and rolls back its database effects; `false` means only +a comparison mismatch. `cas()` owns a transaction on pooled clients and uses a +savepoint on transaction-scoped clients. Reserved clients use a savepoint +inside an existing manual transaction or a transaction owned by the call. +Caller transaction isolation is unchanged. Use reserved clients exclusively +and sequentially during the call. CAS locks in caller-owned transactions +remain until the outer commit, including after a mismatch. + +Conditional batches and multi-key deletions acquire a +[table write lock] before a +fresh snapshot, protecting absent keys against ordinary and direct SQL writers. +This serializes writes to the whole table and its descendants, and can interfere +with autovacuum. A mismatch releases locks acquired in the batch savepoint; +previous writes and setup may still hold locks. Prior caller writes can cause +lock-upgrade deadlocks, requiring outer rollback before retry. Errors and +configured lock timeouts reject; no automatic retry occurs. + +The default JSON codec uses PostgreSQL `jsonb` equality: object member order +is ignored, array order matters, Dates normalize to strings, and nonfinite +numbers normalize to JSON `null`. Big integers, circular/unencodable values, +NUL characters, and malformed Unicode reject. Irreversible value/JSON +transforms are unsupported; identifier transforms remain usable. Raw SQL +values must round-trip through the JavaScript JSON codec for CAS expectations. + +TTLs retain PostgreSQL interval, session-timezone, and calendar semantics at +microsecond precision, using one server expiry cutoff per operation. Omitting +a replacement TTL clears expiry; zero durations expire immediately, and +negative Temporal durations reject. A blocked write's TTL origin may precede +its commit. Invalid JSON/TTL inputs reject even on mismatch. There is no +fixed batch-count or key-length maximum; backend field, index, protocol, and +memory limits apply without splitting. A physical index-size error arises on an +attempted write; a mismatch does not attempt insertion, while a failed +attempted write rolls back the batch. + +Caller-owned batches with `initialized: false` repeat idempotent table setup +and driver probing after possible outer rollback. Setup precedes the batch +savepoint and its errors can abort the outer transaction. `initialized: true` +keeps schema responsibility with the caller. Only pooled batches and `cas()` +start best-effort cleanup after commit; logical expiry never depends on cleanup. + +Atomic operations do not strengthen unlogged-table durability or override +server WAL, `synchronous_commit`, storage, or replica settings. Lost responses +can leave commit status unknown. See the +[package's atomic scope guide] +for the full client, codec, limit, and transaction contract. + #### PostgreSQL expiry index *This optimization is available since Fedify 2.5.0.* @@ -412,6 +487,8 @@ its `AT TIME ZONE` equivalent, and has either no predicate or `WHERE ttl IS NOT NULL`. [`PostgresKvStore`]: https://jsr.io/@fedify/postgres/doc/kv/~/PostgresKvStore +[table write lock]: https://www.postgresql.org/docs/18/sql-lock.html +[package's atomic scope guide]: https://github.com/fedify-dev/fedify/tree/main/packages/postgres#atomic-keyvalue-operations [partitioned index procedure]: https://www.postgresql.org/docs/current/ddl-partitioning.html#DDL-PARTITIONING-DECLARATIVE-MAINTENANCE ### [`PgliteKvStore`] @@ -1008,15 +1085,15 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full -`AtomicKvStore` contract. -Existing single-key CAS support does not imply atomic multi-key support. -The linked issues track separate adapter implementations: +`MemoryKvStore`, `PostgresKvStore`, `SqliteKvStore`, and `PgliteKvStore` +implement `AtomicKvStore` in their documented scopes. Existing single-key CAS +support does not imply atomic multi-key support. The linked issues track +separate adapter implementations: | Implementation | `AtomicKvStore` | Status | | --------------------- | --------------- | ------------------------------------------------------------------------------------------------- | | `MemoryKvStore` | Yes | Development and testing only; one instance. | -| `PostgresKvStore` | No | Planned in [#1364]. | +| `PostgresKvStore` | Yes | One table; see [client and transaction restrictions](#postgresql-atomic-operations). | | `RedisKvStore` | No | Planned in [#1365]. | | `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | | `SqliteKvStore` | Yes | One configured table/database; writable handles, JSON values; see [SQLite scope](#sqlitekvstore). | @@ -1027,7 +1104,6 @@ The linked issues track separate adapter implementations: | `WorkersD1KvStore` | No | Base adapter [PR #1360] supports single-key CAS; atomic follow-up is [#1379]. | [PR #1360]: https://github.com/fedify-dev/fedify/pull/1360 -[#1364]: https://github.com/fedify-dev/fedify/issues/1364 [#1365]: https://github.com/fedify-dev/fedify/issues/1365 [#1368]: https://github.com/fedify-dev/fedify/issues/1368 [#1379]: https://github.com/fedify-dev/fedify/issues/1379 @@ -1642,6 +1718,14 @@ 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. +For PostgreSQL adapters, locking only existing rows does not protect absent +keys. `PostgresKvStore` uses a table write lock followed by a fresh READ +COMMITTED snapshot; its [client restrictions](#postgresql-atomic-operations) +also require exclusive sequential use of caller-owned transactions. PostgreSQL +locks and nested savepoints do not isolate overlapping calls in the same outer +transaction. Test supported independent clients rather than assuming one +transaction-scoped handle provides that concurrency. + 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/postgres/README.md b/packages/postgres/README.md index d7e5e897e..c5fcbeee3 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -60,3 +60,119 @@ yarn add @fedify/postgres # Yarn bun add @fedify/postgres # Bun deno add jsr:@fedify/postgres # Deno ~~~~ + + +Atomic key–value operations +--------------------------- + +Since Fedify 2.5.0, `PostgresKvStore` implements `AtomicKvStore`. Keys in one +configured ordinary table or declarative partition form its atomic scope. +Ordinary inheritance parents and children are unsupported for nonempty batches. +Nonempty batches also reject tables where row-level security is active for the +current role, since policies can filter individual mutations. +The table must retain the adapter's `text[]` primary key and compatible schema; +concurrent incompatible schema changes are outside this scope. + +~~~~ typescript +const store = new PostgresKvStore(sql); +const keys = [["notice"], ["budget"]] as const; +const [notice, budget] = await store.getMany(keys); +const committed = await store.casMany([ + { key: keys[0], expectedValue: notice, newValue: { version: 1 } }, + { key: keys[1], expectedValue: budget, newValue: 10 }, +]); +// On false, read again and re-evaluate the application rule. +~~~~ + +`getMany()` reads one input-ordered snapshot, including repeated keys. Missing +and logically expired keys return `undefined`; stored JSON `null` stays `null`. +`setMany()` and `deleteMany()` change the entire batch. `casMany()` changes +all entries only when every expectation matches. Mutation keys must be +distinct; batch arrays and nonempty string-array keys must not contain holes. +Key strings must be representable as PostgreSQL Unicode text. `set()` and +`setMany()` reject `undefined`; in CAS, it means absence or deletion. A +deletion cannot specify a non-null TTL, even when another comparison mismatches. + +### Clients and transactions + +Nonempty `casMany()` and `deleteMany()` with more than one key require a pool +or a transaction-scoped Postgres.js client at `READ COMMITTED` isolation +(`READ UNCOMMITTED` has the same PostgreSQL behavior). Reserved connections +reject these calls before issuing SQL, whether or not a manual transaction is +open. Reserved clients still support ordinary methods, `getMany()`, +`setMany()`, single-key `deleteMany()`, and all empty batches. Method presence +alone does not establish support for every client scope. + +Caller-owned transactions require exclusive, sequential use for the duration +of each atomic call, including initialization, other store wrappers, and raw +SQL. Await one call before issuing another command on that transaction. +Nested savepoints do not isolate overlapping operations in one outer +transaction. Successful calls remain provisional until the outer commit. The +adapter does not commit, roll back, or release that outer transaction. + +Mutating forms of `cas()` acquire the table write lock and use one server +timestamp for comparison and the replacement's TTL origin. A `cas()` from +`undefined` to `undefined` stays read-only. If a trigger skips a matching +write, the call rejects and rolls back its database effects; `false` means only +a comparison mismatch. `cas()` owns a transaction on pooled clients and uses a +savepoint on transaction-scoped clients. Reserved clients use a savepoint +inside an existing manual transaction or a transaction owned by the call. +Caller transaction isolation is unchanged. Use reserved clients exclusively +and sequentially during the call. CAS locks in caller-owned transactions +remain until the outer commit, including after a mismatch. + +Conditional batches and multi-key deletions acquire a [table write lock] before +a fresh comparison/deletion snapshot. This protects absent keys against +ordinary and direct SQL writers, but serializes all writes to the table and its +descendants. Successful caller batches retain locks until the outer transaction +ends. A mismatch rolls back locks acquired in its savepoint; earlier writes +or setup can still retain locks. Prior caller writes can cause lock-upgrade +deadlocks; roll back the outer transaction before retrying the deciding work. +Database errors and configured lock timeouts reject; the adapter does not retry. +Frequent conflicting locks can also interrupt [ordinary autovacuum]. + +With `initialized: false`, caller-owned batches repeat idempotent table setup +and driver probing because an outer rollback can undo earlier DDL. Setup runs +before the batch savepoint, so a setup error can abort the caller transaction. +With `initialized: true`, providing the schema remains the caller's +responsibility. Pooled batch mutations and mutating forms of `cas()` on pooled +clients start best-effort expiry cleanup after commit. Caller-owned batches +and caller-owned `cas()` calls do not sweep expired rows. Logical expiry is +independent of cleanup. + +[table write lock]: https://www.postgresql.org/docs/18/sql-lock.html +[ordinary autovacuum]: https://www.postgresql.org/docs/18/routine-vacuuming.html#AUTOVACUUM + +### Values, expiry, and limits + +Values use JavaScript JSON encoding and PostgreSQL [`jsonb` equality]. Object +member order is ignored; array order matters. Dates and `toJSON()` values are +normalized, and nonfinite numbers become JSON `null`. Big integers, circular +values, unencodable top-level values, NUL characters, and malformed Unicode +reject. Use passive inputs and the default value codec; irreversible JSON/value +transforms, including property-name changes in nested JSON, are outside the +atomic codec scope. Table identifier transforms remain supported. Raw SQL +values must round-trip through the JavaScript JSON codec to be usable as CAS +expectations. + +TTLs use the server clock, PostgreSQL intervals, and microsecond precision. +Calendar days/months follow the session timezone. Each operation uses one +expiry cutoff; a replacement without a TTL clears the old expiry. Zero +durations expire immediately; negative Temporal durations reject. A write +waiting on another transaction can have a TTL origin earlier than its commit. +Unrepresentable JSON or TTLs reject even on comparison mismatch. + +There is no fixed adapter batch-count or key-length maximum. PostgreSQL field, +index, protocol, and available-memory constraints still apply; calls are never +split to fit them. A physical index-size error can arise only when a write is +attempted: a mismatch performs no insertion and returns `false`, while an +attempted write that fails rolls back the complete batch. Explicit invalid +input and unsupported scopes always reject. + +Atomicity does not change durability. Unlogged tables are crash-truncated and +not replicated; logged tables still depend on server WAL, `synchronous_commit`, +storage, and replica settings. A lost commit response can leave the outcome +unknown. Use application versions for ABA detection and receipts for external +effects. + +[`jsonb` equality]: https://www.postgresql.org/docs/18/datatype-json.html diff --git a/packages/postgres/src/kv.atomic-rls.test.ts b/packages/postgres/src/kv.atomic-rls.test.ts new file mode 100644 index 000000000..0ab131bd5 --- /dev/null +++ b/packages/postgres/src/kv.atomic-rls.test.ts @@ -0,0 +1,76 @@ +import { PostgresKvStore } from "@fedify/postgres/kv"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { test } from "node:test"; +import postgres from "postgres"; + +const dbUrl = process.env.POSTGRES_URL; + +test( + "PostgresKvStore rejects active row-level security before partial atomic mutations", + { + skip: dbUrl == null, + timeout: 30_000, + }, + async () => { + if (dbUrl == null) return; + const admin = postgres(dbUrl, { max: 1 }); + const suffix = crypto.randomUUID().replaceAll("-", ""); + const tableName = `atomic_rls_${suffix}`; + const role = `atomic_rls_role_${suffix}`; + const owner = new PostgresKvStore(admin, { tableName }); + let client: postgres.Sql | undefined; + try { + await owner.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + await admin`CREATE ROLE ${admin(role)}`; + await admin`GRANT SELECT, INSERT, UPDATE, DELETE ON ${admin(tableName)} TO ${admin(role)}`; + await admin`ALTER TABLE ${admin(tableName)} ENABLE ROW LEVEL SECURITY`; + await admin`CREATE POLICY read_rows ON ${admin(tableName)} FOR SELECT USING (true)`; + await admin`CREATE POLICY insert_rows ON ${admin(tableName)} FOR INSERT WITH CHECK (true)`; + await admin`CREATE POLICY update_rows ON ${admin(tableName)} FOR UPDATE USING (true) WITH CHECK (true)`; + await admin`CREATE POLICY delete_rows ON ${admin(tableName)} FOR DELETE USING (false)`; + client = postgres(dbUrl, { max: 1, connection: { role } }); + const store = new PostgresKvStore(client, { + tableName, + initialized: true, + }); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: 1, newValue: undefined }, + { key: ["b"], expectedValue: 2, newValue: 3 }, + ]), + TypeError, + ); + assert.deepEqual(await owner.getMany([["a"], ["b"]]), [1, 2]); + await assert.rejects( + store.setMany([{ key: ["b"], value: 4 }]), + TypeError, + ); + await assert.rejects(store.deleteMany([["a"], ["b"]]), TypeError); + await assert.rejects(store.getMany([["a"]]), TypeError); + assert.deepEqual(await owner.getMany([["a"], ["b"]]), [1, 2]); + await client.begin(async (transaction) => { + const scoped = new PostgresKvStore(transaction, { + tableName, + initialized: true, + }); + await assert.rejects( + scoped.casMany([ + { key: ["a"], expectedValue: 1, newValue: undefined }, + { key: ["b"], expectedValue: 2, newValue: 5 }, + ]), + TypeError, + ); + assert.equal((await transaction`SELECT 1 AS healthy`)[0].healthy, 1); + }); + } finally { + await client?.end(); + await owner.drop(); + await admin`DROP ROLE IF EXISTS ${admin(role)}`; + await admin.end(); + } + }, +); diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts new file mode 100644 index 000000000..18ec17555 --- /dev/null +++ b/packages/postgres/src/kv.atomic.test.ts @@ -0,0 +1,1726 @@ +// Database schedules, setup and disposal must run in order. +/* oxlint-disable eslint/no-await-in-loop */ +// Proxy forwarding observes calls without changing the real client. +/* oxlint-disable anti-slop/no-reflect-apply */ +import type { AtomicKvStore, KvKey } from "@fedify/fedify"; +import { PostgresKvStore } from "@fedify/postgres/kv"; +import { testAtomicKvStore, testKvStore } from "@fedify/testing"; +import * as temporal from "@js-temporal/polyfill"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { setTimeout as delay } from "node:timers/promises"; +import { test } from "node:test"; +import postgres from "postgres"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const dbUrl = process.env.POSTGRES_URL; +const key = ["a"] as const; + +function fixture(sql = postgres(dbUrl!, { max: 1 })) { + const tableName = `fedify_atomic_${crypto.randomUUID().replaceAll("-", "")}`; + const store = new PostgresKvStore(sql, { tableName }); + return { + sql, + tableName, + store, + async dispose() { + try { + await store.drop(); + } finally { + await sql.end(); + } + }, + }; +} + +function dbTest(name: string, run: () => Promise): void { + test(name, { skip: dbUrl == null, timeout: 30_000 }, async () => { + if (dbUrl != null) await run(); // Bun ignores the skip option. + }); +} + +async function state(sql: postgres.Sql, tableName: string) { + return Array.from( + await sql` + SELECT key, value, created::text AS created, ttl::text AS ttl + FROM ${sql(tableName)} ORDER BY key + `, + ); +} + +async function waiting(sql: postgres.Sql, pid: number): Promise { + for (let attempt = 0; attempt < 300; attempt++) { + const [row] = await sql`SELECT EXISTS ( + SELECT 1 FROM pg_locks WHERE pid = ${pid} AND NOT granted + ) AS waiting`; + if (row.waiting) return; + await delay(10); + } + assert.fail("The competing operation did not wait for a database lock"); +} + +async function promptly(operation: Promise): Promise { + let timeout: ReturnType | undefined; + try { + return await Promise.race([ + operation, + new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error("Unexpected retained lock")), + 2000, + ); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +test( + "PostgresKvStore atomic conformance with independent max:1 clients", + { + skip: dbUrl == null, + timeout: 120_000, + }, + async () => { + if (dbUrl == null) return; + await testAtomicKvStore( + async () => { + const f = fixture(); + const other = postgres(dbUrl, { max: 1 }); + try { + await f.store.initialize(); + const second: AtomicKvStore = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + return { + first: f.store, + second, + async dispose() { + try { + await f.dispose(); + } finally { + await other.end(); + } + }, + }; + } catch (error) { + await f.dispose(); + await other.end(); + throw error; + } + }, + { objectKeyOrder: "ignored" }, + ); + }, +); + +dbTest("PostgresKvStore ordinary shared KV regression", async () => { + const f = fixture(); + try { + await testKvStore( + () => f.store, + () => {}, + ); + } finally { + await f.dispose(); + } +}); + +dbTest( + "PostgresKvStore snapshots input and normalizes its JSON codec", + async () => { + const f = fixture(); + try { + const value = { + nested: { a: 1, b: [null, true] }, + date: new Date("2026-01-01Z"), + }; + const keys: [string, ...string[]] = ["one", "two"]; + const pending = f.store.setMany([ + { key: keys, value }, + { key, value: NaN }, + ]); + keys[0] = "changed"; + value.nested.a = 99; + await pending; + assert.deepEqual( + await f.store.getMany([["one", "two"], ["changed", "two"], key, key]), + [ + { + nested: { a: 1, b: [null, true] }, + date: "2026-01-01T00:00:00.000Z", + }, + undefined, + null, + null, + ], + ); + assert.equal(await f.store.cas(key, Infinity, { b: 2, a: 1 }), true); + assert.equal( + await f.store.casMany([ + { key, expectedValue: { a: 1, b: 2 }, newValue: null }, + ]), + true, + ); + assert.equal( + await f.store.casMany([{ key, expectedValue: null, newValue: [1, 2] }]), + true, + ); + assert.equal(await f.store.cas(key, [2, 1], "wrong"), false); + for (const value of [undefined, 1n, () => 1, Symbol("invalid")]) { + await assert.rejects(f.store.set(key, value), TypeError); + if (value !== undefined) + await assert.rejects(f.store.cas(key, value, 1), TypeError); + } + const cyclic: { self?: unknown } = {}; + cyclic.self = cyclic; + await assert.rejects( + f.store.setMany([ + { key, value: 1 }, + { key: ["later"], value: cyclic }, + ]), + ); + assert.deepEqual(await f.store.get(key), [1, 2]); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore validates JSON and TTLs even on CAS mismatch", + async () => { + const f = fixture(); + try { + await f.store.set(key, "old", { + ttl: Temporal.Duration.from({ hours: 1 }), + }); + const before = await state(f.sql, f.tableName); + const ttl = Temporal.Duration.from({ years: 1_000_000 }); + for (const expectedValue of ["old", "wrong"]) { + await assert.rejects(f.store.cas(key, expectedValue, "new", { ttl })); + await assert.rejects( + f.store.casMany([ + { key, expectedValue, newValue: "new" }, + { + key: ["later"], + expectedValue: undefined, + newValue: 1, + options: { ttl }, + }, + ]), + ); + for (const newValue of ["\0", "\ud800"]) { + await assert.rejects(f.store.cas(key, expectedValue, newValue)); + await assert.rejects( + f.store.casMany([{ key, expectedValue, newValue }]), + ); + } + } + await assert.rejects(f.store.set(key, "bad", { ttl })); + await assert.rejects(f.store.set(["absent"], "bad", { ttl })); + await assert.rejects( + f.store.setMany([ + { key, value: "bad" }, + { key: ["later"], value: 1, options: { ttl } }, + ]), + ); + assert.deepEqual(await state(f.sql, f.tableName), before); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore rejects sparse keys and batches before initialization", + async () => { + const f = fixture(); + try { + const sparse = new Array(2); + sparse[0] = key; + await assert.rejects(f.store.getMany(sparse), TypeError); + await assert.rejects(f.store.deleteMany(sparse), TypeError); + await assert.rejects( + f.store.getMany([new Array(2) as unknown as KvKey]), + TypeError, + ); + await assert.rejects( + f.store.setMany([ + { key, value: 1 }, + { key: ["a"], value: 2 }, + ]), + TypeError, + ); + await assert.rejects( + f.store.casMany([ + { key, expectedValue: "mismatch", newValue: 1 }, + { key: ["a"], expectedValue: undefined, newValue: undefined }, + ]), + TypeError, + ); + await assert.rejects( + f.store.cas(key, "wrong", undefined, { + ttl: Temporal.Duration.from({ seconds: 1 }), + }), + TypeError, + ); + const [table] = + await f.sql`SELECT to_regclass(${f.tableName}) AS relation`; + assert.equal(table.relation, null); + } finally { + await f.dispose(); + } + }, +); + +for (const operation of ["deleteMany", "casMany"] as const) { + dbTest( + `PostgresKvStore.${operation} sees a writer committed during lock wait`, + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + const observer = postgres(dbUrl!, { max: 1 }); + const ready = Promise.withResolvers(); + const release = Promise.withResolvers(); + let holder: Promise | undefined; + let pending: Promise | undefined; + try { + await f.store.set(key, 0); + const second = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + await second.initialize(); + const [{ pid }] = await other`SELECT pg_backend_pid() AS pid`; + holder = f.sql.begin(async (sql) => { + await sql`INSERT INTO ${sql(f.tableName)} (key, value, created) + VALUES (ARRAY['a'], '1', statement_timestamp()), (ARRAY['b'], '1', statement_timestamp()) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`; + ready.resolve(); + await release.promise; + }); + void holder.catch(ready.reject); + await ready.promise; + pending = + operation === "deleteMany" + ? second.deleteMany([key, ["b"]]) + : second.casMany([ + { key, expectedValue: 1, newValue: 2 }, + { key: ["b"], expectedValue: 1, newValue: 2 }, + ]); + await waiting(observer, pid); + release.resolve(); + await holder; + const result = await pending; + if (operation === "casMany") assert.equal(result, true); + assert.deepEqual( + await f.store.getMany([key, ["b"]]), + operation === "deleteMany" ? [undefined, undefined] : [2, 2], + ); + } finally { + release.resolve(); + await Promise.allSettled([holder, pending]); + await f.dispose(); + await other.end(); + await observer.end(); + } + }, + ); +} + +dbTest( + "PostgresKvStore.casMany uses post-lock expiry and one TTL origin", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + const observer = postgres(dbUrl!, { max: 1 }); + const ready = Promise.withResolvers(); + const release = Promise.withResolvers(); + let holder: Promise | undefined; + let pending: Promise | undefined; + try { + await f.store.initialize(); + await f.sql`INSERT INTO ${f.sql(f.tableName)} (key,value,created,ttl) + VALUES (ARRAY['a'],'0',statement_timestamp(),interval '150 milliseconds')`; + const second = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + await second.initialize(); + const [{ pid }] = await other`SELECT pg_backend_pid() AS pid`; + holder = f.sql.begin(async (sql) => { + await sql`LOCK TABLE ${sql(f.tableName)} IN ROW EXCLUSIVE MODE`; + ready.resolve(); + await release.promise; + }); + void holder.catch(ready.reject); + await ready.promise; + pending = second.casMany([ + { + key, + expectedValue: undefined, + newValue: 1, + options: { ttl: Temporal.Duration.from({ seconds: 5 }) }, + }, + { + key: ["b"], + expectedValue: undefined, + newValue: 1, + options: { ttl: Temporal.Duration.from({ seconds: 5 }) }, + }, + ]); + await waiting(observer, pid); + await observer`SELECT pg_sleep(0.2)`; // Let the locked value expire, after observing the wait. + release.resolve(); + await holder; + assert.equal(await pending, true); + const [row] = await f.sql`SELECT count(DISTINCT created)::int AS origins, + bool_and(created + ttl > statement_timestamp()) AS live FROM ${f.sql(f.tableName)}`; + assert.equal(row.origins, 1); + assert.equal(row.live, true); + } finally { + release.resolve(); + await Promise.allSettled([holder, pending]); + await f.dispose(); + await other.end(); + await observer.end(); + } + }, +); + +dbTest( + "PostgresKvStore crossed setMany batches use stable key order", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + try { + await f.store.initialize(); + const second = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + for (let round = 0; round < 8; round++) { + await Promise.all([ + f.store.setMany([ + { key, value: 1 }, + { key: ["b"], value: 1 }, + ]), + second.setMany([ + { key: ["b"], value: 2 }, + { key, value: 2 }, + ]), + ]); + const [a, b] = await f.store.getMany([key, ["b"]]); + assert.equal(a, b); + assert.ok(a === 1 || a === 2); + } + } finally { + await f.dispose(); + await other.end(); + } + }, +); + +dbTest( + "PostgresKvStore later trigger failure rolls back values and TTLs", + async () => { + const f = fixture(); + const functionName = `${f.tableName}_fail`; + try { + await f.store.setMany([ + { + key, + value: 1, + options: { ttl: Temporal.Duration.from({ hours: 1 }) }, + }, + { key: ["keep"], value: 2 }, + ]); + await f.sql`CREATE FUNCTION ${f.sql(functionName)}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN IF NEW.key[1] = 'fail' THEN RAISE EXCEPTION 'injected failure'; END IF; RETURN NEW; END $$`; + await f.sql`CREATE TRIGGER failure BEFORE INSERT OR UPDATE ON ${f.sql(f.tableName)} + FOR EACH ROW EXECUTE FUNCTION ${f.sql(functionName)}()`; + const before = await state(f.sql, f.tableName); + const entries = [ + { key, expectedValue: 1, newValue: undefined }, + { key: ["fail"] as const, expectedValue: undefined, newValue: 3 }, + ]; + await assert.rejects(f.store.casMany(entries), /injected failure/); + assert.deepEqual(await state(f.sql, f.tableName), before); + await f.sql.begin(async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + await sql`INSERT INTO ${sql(f.tableName)} (key,value,created) VALUES (ARRAY['prior'],'9',statement_timestamp())`; + await assert.rejects(store.casMany(entries), /injected failure/); + assert.equal((await sql`SELECT 1 AS one`)[0].one, 1); + assert.equal(await store.get(["prior"]), 9); + assert.equal(await store.get(key), 1); + }); + assert.equal(await f.store.get(["prior"]), 9); + await assert.rejects( + f.store.setMany([ + { key, value: 7 }, + { key: ["fail"], value: 3 }, + ]), + /injected failure/, + ); + assert.equal(await f.store.get(key), 1); + } finally { + await f.sql`DROP FUNCTION IF EXISTS ${f.sql(functionName)}() CASCADE`; + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore sequential caller batches preserve outer rollback and release mismatch locks", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + const rollback = new Error("outer rollback"); + try { + await f.store.setMany([ + { key, value: 1 }, + { key: ["b"], value: 1 }, + ]); + const second = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + await second.initialize(); + await assert.rejects( + f.sql.begin(async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + assert.equal( + await store.casMany([{ key, expectedValue: 0, newValue: 2 }]), + false, + ); + await promptly(second.set(["free"], 3)); + assert.equal( + await store.casMany([{ key, expectedValue: 1, newValue: 2 }]), + true, + ); + await store.deleteMany([key, ["b"]]); + assert.deepEqual(await store.getMany([key, ["b"]]), [ + undefined, + undefined, + ]); + throw rollback; + }), + (error) => error === rollback, + ); + assert.deepEqual( + await f.store.getMany([key, ["b"], ["free"]]), + [1, 1, 3], + ); + } finally { + await f.dispose(); + await other.end(); + } + }, +); + +dbTest( + "PostgresKvStore lazy caller initialization survives mismatch rollback", + async () => { + const f = fixture(); + try { + await f.sql.begin(async (sql) => { + const store = new PostgresKvStore(sql, { tableName: f.tableName }); + assert.equal( + await store.casMany([{ key, expectedValue: "wrong", newValue: 1 }]), + false, + ); + await store.set(key, "committed"); + }); + assert.equal(await f.store.get(key), "committed"); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore batch reads reuse existing tables in read-only transactions", + async () => { + const f = fixture(); + try { + await f.store.setMany([{ key, value: "existing" }]); + await f.sql.begin("read only", async (sql) => { + const store = new PostgresKvStore(sql, { tableName: f.tableName }); + for (let attempt = 0; attempt < 2; attempt++) { + assert.deepEqual(await store.getMany([key, ["missing"]]), [ + "existing", + undefined, + ]); + } + assert.equal((await sql`SELECT 1 AS healthy`)[0].healthy, 1); + }); + const reserved = await f.sql.reserve(); + try { + await reserved`BEGIN READ ONLY`; + const store = new PostgresKvStore(reserved, { tableName: f.tableName }); + assert.deepEqual(await store.getMany([key]), ["existing"]); + assert.equal((await reserved`SELECT 1 AS healthy`)[0].healthy, 1); + } finally { + await reserved`ROLLBACK`; + reserved.release(); + } + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore reserved batch initialization refreshes after rollback and failure", + async () => { + const f = fixture(); + const reserved = await f.sql.reserve(); + try { + const store = new PostgresKvStore(reserved, { tableName: f.tableName }); + await reserved`BEGIN`; + await store.setMany([{ key, value: "rolled back" }]); + await reserved`ROLLBACK`; + await reserved`BEGIN READ ONLY`; + await assert.rejects(store.setMany([{ key, value: "must reject" }])); + await reserved`ROLLBACK`; + assert.deepEqual(await store.getMany([key]), [undefined]); + await store.setMany([{ key, value: "recovered" }]); + assert.deepEqual(await store.getMany([key]), ["recovered"]); + } finally { + await reserved`ROLLBACK`; + reserved.release(); + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore reserved restrictions reject before SQL and preserve caller transaction", + async () => { + const f = fixture(); + await f.store.initialize(); + const reserved = await f.sql.reserve(); + try { + for (const transaction of [false, true]) { + if (transaction) await reserved`BEGIN`; + let calls = 0; + const observed = new Proxy(reserved, { + apply(target, receiver, args) { + calls++; + return Reflect.apply(target, receiver, args); + }, + }); + const store = new PostgresKvStore(observed, { + tableName: f.tableName, + initialized: true, + }); + await assert.rejects( + store.casMany([{ key, expectedValue: undefined, newValue: 1 }]), + TypeError, + ); + await assert.rejects(store.deleteMany([key, ["b"]]), TypeError); + assert.deepEqual(await store.getMany([]), []); + await store.setMany([]); + await store.deleteMany([]); + assert.equal(await store.casMany([]), true); + assert.equal(calls, 0); + await store.setMany([{ key, value: 1 }]); + await store.deleteMany([key]); + assert.deepEqual(await store.getMany([key]), [undefined]); + assert.equal((await reserved`SELECT 1 AS one`)[0].one, 1); + if (transaction) await reserved`COMMIT`; + } + } finally { + await reserved`ROLLBACK`; + reserved.release(); + await f.dispose(); + } + }, +); + +for (const isolation of ["repeatable read", "serializable"]) { + dbTest( + `PostgresKvStore rejects caller ${isolation} atomic mutation scope`, + async () => { + const f = fixture(); + try { + await f.store.set(key, 1); + await f.sql.begin(`isolation level ${isolation}`, async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + await assert.rejects( + store.casMany([{ key, expectedValue: 1, newValue: 2 }]), + TypeError, + ); + await assert.rejects(store.deleteMany([key, ["b"]]), TypeError); + assert.equal((await sql`SELECT 1 AS one`)[0].one, 1); + }); + assert.equal(await f.store.get(key), 1); + } finally { + await f.dispose(); + } + }, + ); +} + +for (const identity of [false, true]) { + dbTest( + `PostgresKvStore batch codec with identity JSON serializer=${identity} and non-ISO DateStyle`, + async () => { + const sql = identity + ? postgres(dbUrl!, { + max: 1, + connection: { DateStyle: "SQL, DMY", TimeZone: "Australia/Sydney" }, + types: { + json: { + to: 3802, + from: [114, 3802], + serialize: (value: unknown) => value as string, + parse: JSON.parse, + }, + }, + }) + : postgres(dbUrl!, { + max: 1, + connection: { DateStyle: "SQL, DMY", TimeZone: "Australia/Sydney" }, + }); + const f = fixture(sql); + try { + await f.store.initialize(); + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + await store.setMany([ + { key, value: null }, + { key: ["b"], value: { nested: [null, 1] } }, + ]); + assert.equal( + await store.casMany([ + { + key, + expectedValue: null, + newValue: { a: 1 }, + options: { + ttl: Temporal.Duration.from({ seconds: 60, microseconds: 1 }), + }, + }, + { + key: ["b"], + expectedValue: { nested: [null, 1] }, + newValue: null, + }, + ]), + true, + ); + assert.deepEqual(await store.getMany([key, ["b"]]), [{ a: 1 }, null]); + const [row] = await sql`SELECT count(DISTINCT created)::int AS origins, + bool_and(ttl IS NULL OR ttl = interval '60.000001 seconds') AS precision FROM ${sql(f.tableName)}`; + assert.equal(row.origins, 1); + assert.equal(row.precision, true); + assert.equal(await store.cas(key, { a: 1 }, null), true); + assert.equal(await store.get(key), null); + } finally { + await f.dispose(); + } + }, + ); +} +for (const membership of ["parent", "child"]) { + dbTest( + `PostgresKvStore rejects ordinary inheritance ${membership} batches`, + async () => { + const f = fixture(); + const child = `${f.tableName}_child`; + try { + await f.store.initialize(); + await f.sql`CREATE TABLE ${f.sql(child)} () INHERITS (${f.sql(f.tableName)})`; + await f.sql`INSERT INTO ${f.sql(child)} (key,value,created) VALUES (ARRAY['a'],'1',statement_timestamp())`; + const store = new PostgresKvStore(f.sql, { + tableName: membership === "parent" ? f.tableName : child, + initialized: true, + }); + await assert.rejects(store.getMany([key]), TypeError); + await assert.rejects(store.setMany([{ key: ["absent"], value: 2 }])); + await assert.rejects(store.deleteMany([key]), TypeError); + await assert.rejects(store.deleteMany([key, ["absent"]]), TypeError); + await assert.rejects( + store.casMany([{ key, expectedValue: 1, newValue: 2 }]), + TypeError, + ); + await f.sql`INSERT INTO ${f.sql(f.tableName)} (key,value,created) VALUES (ARRAY['a'],'3',statement_timestamp())`; + await assert.rejects(store.getMany([key, key]), TypeError); + assert.equal( + ( + await f.sql`SELECT count(*)::int AS size FROM ${f.sql(f.tableName)}` + )[0].size, + 2, + ); + } finally { + await f.sql`DROP TABLE IF EXISTS ${f.sql(child)}`; + await f.dispose(); + } + }, + ); +} + +dbTest( + "PostgresKvStore declarative parent and leaf batches coordinate direct partition writes", + async () => { + const f = fixture(); + const child = `${f.tableName}_part`; + const other = postgres(dbUrl!, { max: 1 }); + const observer = postgres(dbUrl!, { max: 1 }); + const ready = Promise.withResolvers(); + const release = Promise.withResolvers(); + let holder: Promise | undefined; + let pending: Promise | undefined; + try { + await f.sql`CREATE TABLE ${f.sql(f.tableName)} ( + key text[] PRIMARY KEY,value jsonb NOT NULL,created timestamptz,ttl interval + ) PARTITION BY HASH (key)`; + await f.sql`CREATE TABLE ${f.sql(child)} PARTITION OF ${f.sql(f.tableName)} FOR VALUES WITH (MODULUS 1, REMAINDER 0)`; + const parent = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + const leaf = new PostgresKvStore(f.sql, { + tableName: child, + initialized: true, + }); + await leaf.setMany([ + { key, value: 0 }, + { key: ["b"], value: 0 }, + ]); + assert.deepEqual(await parent.getMany([key, ["b"]]), [0, 0]); + const [{ pid }] = await other`SELECT pg_backend_pid() AS pid`; + holder = f.sql.begin(async (sql) => { + await sql`UPDATE ${sql(child)} SET value = '1' WHERE key = ARRAY['a']`; + ready.resolve(); + await release.promise; + }); + void holder.catch(ready.reject); + await ready.promise; + pending = parent.casMany([ + { key, expectedValue: 1, newValue: 2 }, + { key: ["b"], expectedValue: 0, newValue: 2 }, + ]); + await waiting(observer, pid); + release.resolve(); + await holder; + assert.equal(await pending, true); + assert.equal( + await leaf.casMany([{ key, expectedValue: 2, newValue: 3 }]), + true, + ); + await leaf.deleteMany([key, ["b"]]); + assert.deepEqual(await parent.getMany([key, ["b"]]), [ + undefined, + undefined, + ]); + } finally { + release.resolve(); + await Promise.allSettled([holder, pending]); + await f.sql`DROP TABLE IF EXISTS ${f.sql(child)}`; + await f.dispose(); + await other.end(); + await observer.end(); + } + }, +); + +dbTest( + "PostgresKvStore lock timeout is an error and preserves state", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { + max: 1, + connection: { lock_timeout: 100 }, + }); + const ready = Promise.withResolvers(); + const release = Promise.withResolvers(); + let holder: Promise | undefined; + try { + await f.store.setMany([ + { key, value: 1 }, + { key: ["b"], value: 1 }, + ]); + const second = new PostgresKvStore(other, { + tableName: f.tableName, + initialized: true, + }); + holder = f.sql.begin(async (sql) => { + await sql`LOCK TABLE ${sql(f.tableName)} IN ROW EXCLUSIVE MODE`; + ready.resolve(); + await release.promise; + }); + void holder.catch(ready.reject); + await ready.promise; + await assert.rejects( + second.casMany([{ key, expectedValue: 1, newValue: 2 }]), + (error: unknown) => + error instanceof postgres.PostgresError && error.code === "55P03", + ); + await assert.rejects( + second.deleteMany([key, ["b"]]), + (error: unknown) => + error instanceof postgres.PostgresError && error.code === "55P03", + ); + release.resolve(); + await holder; + assert.deepEqual(await f.store.getMany([key, ["b"]]), [1, 1]); + } finally { + release.resolve(); + await Promise.allSettled([holder]); + await f.dispose(); + await other.end(); + } + }, +); + +dbTest( + "PostgresKvStore caller lock-upgrade deadlock rejects without a partial batch", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + const entered = Promise.withResolvers(); + let writers = 0; + try { + await f.store.setMany([ + { key, value: 0 }, + { key: ["b"], value: 0 }, + ]); + const attempt = (client: postgres.Sql, name: string) => + client.begin(async (sql) => { + await sql`INSERT INTO ${sql(f.tableName)} (key,value,created) VALUES (${[name]},'9',statement_timestamp())`; + if (++writers === 2) entered.resolve(); + await entered.promise; + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + try { + assert.equal( + await store.casMany([ + { key, expectedValue: 0, newValue: 1 }, + { key: ["b"], expectedValue: 0, newValue: 1 }, + ]), + true, + ); + } catch (error) { + assert.equal((await sql`SELECT 1 AS one`)[0].one, 1); + throw error; // Release earlier ROW EXCLUSIVE locks with outer rollback. + } + }); + const results = await Promise.allSettled([ + attempt(f.sql, "left"), + attempt(other, "right"), + ]); + const failures = results.filter((result) => result.status === "rejected"); + assert.equal(failures.length, 1); + assert.ok(failures[0].reason instanceof postgres.PostgresError); + assert.equal(failures[0].reason.code, "40P01"); + assert.deepEqual(await f.store.getMany([key, ["b"]]), [1, 1]); + const markers = await f.store.getMany([["left"], ["right"]]); + assert.equal(markers.filter((value) => value === 9).length, 1); + } finally { + await f.dispose(); + await other.end(); + } + }, +); + +dbTest( + "PostgresKvStore backend index limit rejects attempted mutation but not mismatch", + async () => { + const f = fixture(); + // Incompressible text exceeds the default B-tree entry size; no adapter cap. + const oversized = [ + Array.from({ length: 400 }, () => crypto.randomUUID()).join(""), + ] as const; + try { + await f.store.set(key, 1); + assert.equal( + await f.store.casMany([ + { key, expectedValue: "wrong", newValue: undefined }, + { key: oversized, expectedValue: undefined, newValue: 2 }, + ]), + false, + ); + await assert.rejects( + f.store.casMany([ + { key, expectedValue: 1, newValue: undefined }, + { key: oversized, expectedValue: undefined, newValue: 2 }, + ]), + ); + assert.deepEqual(await f.store.getMany([key, oversized]), [1, undefined]); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore qualified transformed identifiers preserve scalar atomic values", + async () => { + const sql = postgres(dbUrl!, { max: 1, transform: postgres.camel }); + const suffix = crypto.randomUUID().replaceAll("-", ""); + const schema = `AtomicSchema${suffix}`; + const tableName = `${schema}.AtomicValues`; + const store = new PostgresKvStore(sql, { tableName }); + try { + await sql`CREATE SCHEMA ${sql(schema)}`; + await store.setMany([ + { key, value: null }, + { key: ["b"], value: 1 }, + ]); + assert.equal( + await store.casMany([{ key, expectedValue: null, newValue: 2 }]), + true, + ); + assert.deepEqual(await store.getMany([key, ["b"], key]), [2, 1, 2]); + await store.deleteMany([key, ["b"]]); + } finally { + await store.drop(); + await sql`DROP SCHEMA IF EXISTS ${sql(schema)}`; + await sql.end(); + } + }, +); + +dbTest( + "PostgresKvStore single CAS liveness and TTL origin follow lock acquisition", + async () => { + const f = fixture(); + const validated = Promise.withResolvers(); + const release = Promise.withResolvers(); + let pending: Promise | undefined; + try { + await f.store.initialize(); + await f.sql`INSERT INTO ${f.sql(f.tableName)} (key,value,created,ttl) + VALUES (ARRAY['a'],'0',statement_timestamp(),interval '300 milliseconds')`; + const observed = new Proxy(f.sql, { + apply(target, receiver, args) { + const query = args[0]; + const result = Reflect.apply(target, receiver, args); + if ( + Array.isArray(query) && + query.join("").includes("count((item->>'ttl')::interval)") + ) { + return (async () => { + const rows = await result; + validated.resolve(); + await release.promise; + return rows; + })(); + } + return result; + }, + }); + const store = new PostgresKvStore(observed, { + tableName: f.tableName, + initialized: true, + }); + pending = store.cas(key, undefined, "fresh", { + ttl: Temporal.Duration.from({ seconds: 5 }), + }); + await validated.promise; + const [{ marker }] = + await f.sql`SELECT statement_timestamp()::text AS marker`; + await f.sql`SELECT pg_sleep(0.35)`; + release.resolve(); + assert.equal(await pending, true); + const [row] = + await f.sql`SELECT value, created > ${marker}::text::timestamptz AS fresh, + created + ttl > statement_timestamp() AS live FROM ${f.sql(f.tableName)}`; + assert.equal(row.value, "fresh"); + assert.equal(row.fresh, true); + assert.equal(row.live, true); + } finally { + release.resolve(); + await Promise.allSettled([pending]); + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore pooled batches reclaim expired rows after commit", + async () => { + const f = fixture(); + const swept = Promise.withResolvers(); + try { + await f.store.initialize(); + await f.sql`INSERT INTO ${f.sql(f.tableName)} (key,value,created,ttl) + VALUES (ARRAY['expired'],'0',statement_timestamp()-interval '1 hour',interval '1 second')`; + const observed = new Proxy(f.sql, { + apply(target, receiver, args) { + const query = args[0]; + const result = Reflect.apply(target, receiver, args); + if ( + Array.isArray(query) && + query.join("").includes("WITH sweep_lock") + ) { + return (async () => { + const rows = await result; + swept.resolve(); + return rows; + })(); + } + return result; + }, + }); + const store = new PostgresKvStore(observed, { + tableName: f.tableName, + initialized: true, + }); + await store.setMany([{ key, value: 1 }]); + await promptly(swept.promise); + const [row] = + await f.sql`SELECT count(*)::int AS size FROM ${f.sql(f.tableName)} WHERE key=ARRAY['expired']`; + assert.equal(row.size, 0); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore pooled cleanup failure cannot reject a committed batch and drop drains it", + async () => { + const f = fixture(); + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + let dropping: Promise | undefined; + try { + await f.store.initialize(); + const observed = new Proxy(f.sql, { + apply(target, receiver, args) { + const query = args[0]; + if ( + Array.isArray(query) && + query.join("").includes("WITH sweep_lock") + ) { + started.resolve(); + return release.promise.then(() => { + throw new Error("injected cleanup failure"); + }); + } + return Reflect.apply(target, receiver, args); + }, + }); + const store = new PostgresKvStore(observed, { + tableName: f.tableName, + initialized: true, + }); + await promptly( + store.casMany([{ key, expectedValue: undefined, newValue: 1 }]), + ); + await started.promise; + assert.equal(await f.store.get(key), 1); + let dropped = false; + dropping = store.drop().then(() => { + dropped = true; + }); + await f.sql`SELECT 1`; + assert.equal(dropped, false); + release.resolve(); + await dropping; + assert.equal(dropped, true); + } finally { + release.resolve(); + await dropping; + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore single and batch CAS recover logically absent legacy rows", + async () => { + const f = fixture(); + try { + await f.store.initialize(); + await f.sql`INSERT INTO ${f.sql(f.tableName)} (key,value,created,ttl) + VALUES (ARRAY['a'],'1',NULL,interval '1 day'),(ARRAY['b'],'1',NULL,interval '1 day')`; + assert.deepEqual(await f.store.getMany([key, ["b"]]), [ + undefined, + undefined, + ]); + assert.equal(await f.store.cas(key, undefined, "single"), true); + assert.equal( + await f.store.casMany([ + { key: ["b"], expectedValue: undefined, newValue: "batch" }, + ]), + true, + ); + assert.deepEqual(await f.store.getMany([key, ["b"]]), [ + "single", + "batch", + ]); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore zero TTL expires and negative Temporal TTL rejects atomically", + async () => { + const f = fixture(); + try { + await f.store.set(key, "old", { + ttl: Temporal.Duration.from({ hours: 1 }), + }); + const before = await state(f.sql, f.tableName); + const options = { ttl: Temporal.Duration.from({ seconds: -1 }) }; + await assert.rejects(f.store.set(key, "negative", options)); + await assert.rejects( + f.store.setMany([ + { key, value: "changed" }, + { key: ["negative"], value: 1, options }, + ]), + ); + for (const expectedValue of ["old", "mismatch"]) { + await assert.rejects( + f.store.cas(key, expectedValue, "negative", options), + ); + await assert.rejects( + f.store.casMany([ + { key, expectedValue, newValue: "changed" }, + { + key: ["negative"], + expectedValue: undefined, + newValue: 1, + options, + }, + ]), + ); + } + assert.deepEqual(await state(f.sql, f.tableName), before); + const zero = { ttl: Temporal.Duration.from({ seconds: 0 }) }; + await f.store.set(["zero-single"], 1, zero); + await f.store.setMany([{ key: ["zero-batch"], value: 1, options: zero }]); + assert.equal(await f.store.cas(["zero-cas"], undefined, 1, zero), true); + assert.equal( + await f.store.casMany([ + { + key: ["zero-cas-batch"], + expectedValue: undefined, + newValue: 1, + options: zero, + }, + ]), + true, + ); + assert.deepEqual( + await f.store.getMany([ + ["zero-single"], + ["zero-batch"], + ["zero-cas"], + ["zero-cas-batch"], + ]), + [undefined, undefined, undefined, undefined], + ); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore transformed table probes match DDL and read-only reads", + async () => { + const sql = postgres(dbUrl!, { max: 1, transform: postgres.camel }); + const admin = postgres(dbUrl!, { max: 1 }); + const tableName = `atomicValues${crypto.randomUUID().replaceAll("-", "")}`; + const actual = tableName.replace("atomicValues", "atomic_values"); + const store = new PostgresKvStore(sql, { tableName }); + try { + await sql`CREATE UNLOGGED TABLE ${sql(tableName)} ( + key text[] PRIMARY KEY, value jsonb NOT NULL, + created timestamptz DEFAULT CURRENT_TIMESTAMP, ttl interval + )`; + await store.initialize(); + const [relation] = await admin`SELECT relpersistence FROM pg_class + WHERE oid = to_regclass(${`"${actual}"`})`; + assert.equal(relation.relpersistence, "p"); + await store.setMany([{ key, value: "existing" }]); + await sql.begin("read only", async (transaction) => { + const reader = new PostgresKvStore(transaction, { tableName }); + assert.deepEqual(await reader.getMany([key]), ["existing"]); + }); + } finally { + await store.drop(); + await sql.end(); + await admin.end(); + } + }, +); + +dbTest( + "PostgresKvStore transformed initialization ignores a raw-name collision", + async () => { + const admin = postgres(dbUrl!, { max: 1 }); + const tableName = `atomic_${crypto.randomUUID().replaceAll("-", "")}`; + const sql = postgres(dbUrl!, { + max: 1, + transform: { column: { to: (name) => `x_${name}` } }, + }); + const store = new PostgresKvStore(sql, { tableName }); + try { + await admin`CREATE TABLE ${admin(tableName)} (key text[] PRIMARY KEY, value jsonb NOT NULL, + created timestamptz DEFAULT CURRENT_TIMESTAMP, ttl interval)`; + await store.setMany([{ key, value: "transformed" }]); + assert.deepEqual(await store.getMany([key]), ["transformed"]); + assert.equal( + (await admin`SELECT count(*)::int AS size FROM ${admin(tableName)}`)[0] + .size, + 0, + ); + } finally { + await store.drop(); + await admin`DROP TABLE IF EXISTS ${admin(tableName)}`; + await sql.end(); + await admin.end(); + } + }, +); + +dbTest( + "PostgresKvStore table names cannot be shadowed by atomic CTEs", + async () => { + const sql = postgres(dbUrl!, { max: 1 }); + const schema = `atomic_names_${crypto.randomUUID().replaceAll("-", "")}`; + try { + await sql`CREATE SCHEMA ${sql(schema)}`; + await sql`SET search_path TO ${sql(schema)}`; + for (const tableName of [ + "input", + "support", + "written", + "deleted", + "targets", + ]) { + const store = new PostgresKvStore(sql, { tableName }); + try { + await store.setMany([ + { key, value: 1 }, + { key: ["b"], value: 2 }, + ]); + assert.deepEqual(await store.getMany([key, ["b"]]), [1, 2]); + assert.equal( + await store.casMany([ + { key, expectedValue: 1, newValue: undefined }, + { key: ["b"], expectedValue: 2, newValue: 3 }, + ]), + true, + ); + assert.deepEqual(await store.getMany([key, ["b"]]), [undefined, 3]); + await store.deleteMany([key, ["b"]]); + assert.deepEqual(await store.getMany([key, ["b"]]), [ + undefined, + undefined, + ]); + } finally { + await store.drop(); + } + } + } finally { + await sql`RESET search_path`; + await sql`DROP SCHEMA IF EXISTS ${sql(schema)} CASCADE`; + await sql.end(); + } + }, +); + +for (const [operation, client] of [ + ["insert", "pool"], + ["update", "pool"], + ["delete", "pool"], + ["cas-delete", "pool"], + ["cas-update", "pool"], + ["insert", "reserved"], + ["cas-update", "transaction"], +] as const) { + dbTest( + `PostgresKvStore rolls back skipped ${operation} rows on a ${client} client`, + async () => { + const f = fixture(); + const audit = `${f.tableName}_audit`; + const fn = `${f.tableName}_filter`; + let reserved: postgres.ReservedSql | undefined; + try { + await f.store.initialize(); + if (operation !== "insert") { + await f.store.setMany([ + { key, value: 1 }, + { key: ["b"], value: 2 }, + ]); + } + const before = await state(f.sql, f.tableName); + await f.sql`CREATE TABLE ${f.sql(audit)} (event text)`; + await f.sql`CREATE FUNCTION ${f.sql(fn)}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO ${f.sql(audit)} VALUES (TG_OP); + IF TG_OP = 'DELETE' THEN + IF OLD.key = ARRAY['a'] THEN RETURN NULL; END IF; + RETURN OLD; + END IF; + IF NEW.key = ARRAY['a'] THEN RETURN NULL; END IF; + RETURN NEW; + END + $$`; + const event = + operation === "insert" + ? f.sql`INSERT` + : operation === "delete" || operation === "cas-delete" + ? f.sql`DELETE` + : f.sql`UPDATE`; + await f.sql`CREATE TRIGGER filter_rows BEFORE ${event} ON ${f.sql(f.tableName)} + FOR EACH ROW EXECUTE FUNCTION ${f.sql(fn)}()`; + const invoke = (store: PostgresKvStore) => { + if (operation === "insert" || operation === "update") { + return store.setMany([ + { key, value: 11 }, + { key: ["b"], value: 22 }, + ]); + } + if (operation === "delete") + return store.deleteMany([key, ["b"], ["missing"]]); + if (operation === "cas-delete") + return store.casMany([ + { key, expectedValue: 1, newValue: undefined }, + { key: ["b"], expectedValue: 2, newValue: 22 }, + ]); + return store.casMany([ + { key: ["b"], expectedValue: 2, newValue: undefined }, + { key, expectedValue: 1, newValue: 11 }, + ]); + }; + if (client === "transaction") { + await f.sql.begin(async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + await assert.rejects(invoke(store)); + assert.equal((await sql`SELECT 1 AS healthy`)[0].healthy, 1); + }); + } else { + if (client === "reserved") reserved = await f.sql.reserve(); + const store = + reserved == null + ? f.store + : new PostgresKvStore(reserved, { + tableName: f.tableName, + initialized: true, + }); + await assert.rejects(invoke(store)); + reserved?.release(); + reserved = undefined; + } + assert.deepEqual(await state(f.sql, f.tableName), before); + assert.equal( + (await f.sql`SELECT count(*)::int AS size FROM ${f.sql(audit)}`)[0] + .size, + 0, + ); + } finally { + reserved?.release(); + await f.sql`DROP TRIGGER IF EXISTS filter_rows ON ${f.sql(f.tableName)}`; + await f.sql`DROP FUNCTION IF EXISTS ${f.sql(fn)}()`; + await f.sql`DROP TABLE IF EXISTS ${f.sql(audit)}`; + await f.dispose(); + } + }, + ); +} + +dbTest( + "PostgresKvStore deletion counts preserve missing-key and concurrent-delete no-ops", + async () => { + const f = fixture(); + const other = postgres(dbUrl!, { max: 1 }); + const blocker = await other.reserve(); + let pending: Promise | undefined; + try { + await f.store.setMany([{ key, value: 1 }]); + await f.store.deleteMany([["missing-a"], ["missing-b"]]); + assert.equal( + await f.store.casMany([ + { key: ["missing"], expectedValue: undefined, newValue: undefined }, + { key, expectedValue: 1, newValue: 2 }, + ]), + true, + ); + const [{ pid }] = await f.sql`SELECT pg_backend_pid() AS pid`; + await blocker`BEGIN`; + await blocker`DELETE FROM ${blocker(f.tableName)} WHERE key = ARRAY['a']`; + pending = f.store.deleteMany([key]); + await waiting(blocker, pid); + await blocker`COMMIT`; + await promptly(pending); + pending = undefined; + assert.deepEqual(await f.store.getMany([key]), [undefined]); + } finally { + await blocker`ROLLBACK`; + await pending?.catch(() => {}); + blocker.release(); + await other.end(); + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore resolves relation names that shadow built-in types", + async () => { + const sql = postgres(dbUrl!, { max: 1 }); + const schema = `fedify_types_${crypto.randomUUID().replaceAll("-", "")}`; + try { + await sql`CREATE SCHEMA ${sql(schema)}`; + await sql`SET search_path TO ${sql(schema)}`; + const store = new PostgresKvStore(sql, { tableName: "text" }); + await store.set(key, 1); + assert.equal(await store.get(key), 1); + assert.deepEqual(await store.getMany([key]), [1]); + await store.setMany([ + { key, value: 2 }, + { key: ["b"], value: 3 }, + ]); + assert.equal( + await store.casMany([{ key, expectedValue: 2, newValue: 4 }]), + true, + ); + await store.deleteMany([["b"]]); + assert.deepEqual(await store.getMany([key, ["b"]]), [4, undefined]); + } finally { + await sql`RESET search_path`; + await sql`DROP SCHEMA IF EXISTS ${sql(schema)} CASCADE`; + await sql.end(); + } + }, +); + +for (const operation of ["setMany", "deleteMany"] as const) { + dbTest( + `PostgresKvStore ${operation} rejects unsupported tables before statement triggers`, + async () => { + const f = fixture(); + const child = `${f.tableName}_child`; + const audit = `${f.tableName}_audit`; + const fn = `${f.tableName}_audit_fn`; + try { + await f.store.set(key, 1); + const before = await state(f.sql, f.tableName); + await f.sql`CREATE TABLE ${f.sql(child)} () INHERITS (${f.sql(f.tableName)})`; + await f.sql`CREATE TABLE ${f.sql(audit)} (event text)`; + await f.sql`CREATE FUNCTION ${f.sql(fn)}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO ${f.sql(audit)} VALUES (TG_OP); + RETURN NULL; + END + $$`; + await f.sql`CREATE TRIGGER audit_statement AFTER INSERT OR UPDATE OR DELETE + ON ${f.sql(f.tableName)} FOR EACH STATEMENT EXECUTE FUNCTION ${f.sql(fn)}()`; + await assert.rejects( + operation === "setMany" + ? f.store.setMany([{ key, value: 2 }]) + : f.store.deleteMany([key]), + TypeError, + ); + assert.deepEqual(await state(f.sql, f.tableName), before); + assert.equal( + (await f.sql`SELECT count(*)::int AS size FROM ${f.sql(audit)}`)[0] + .size, + 0, + ); + } finally { + await f.sql`DROP TRIGGER IF EXISTS audit_statement ON ${f.sql(f.tableName)}`; + await f.sql`DROP FUNCTION IF EXISTS ${f.sql(fn)}()`; + await f.sql`DROP TABLE IF EXISTS ${f.sql(audit)}`; + await f.sql`DROP TABLE IF EXISTS ${f.sql(child)}`; + await f.dispose(); + } + }, + ); +} + +for (const operation of [ + "insert", + "update", + "delete", + "expired-update", +] as const) { + for (const client of [ + "pool", + "transaction", + "reserved", + "reserved-transaction", + ] as const) { + dbTest( + `PostgresKvStore single CAS rolls back skipped ${operation} on ${client}`, + async () => { + const f = fixture(); + const audit = `${f.tableName}_audit`; + const fn = `${f.tableName}_filter`; + try { + await f.store.initialize(); + if (operation !== "insert") await f.store.set(key, 1); + if (operation === "expired-update") { + await f.sql`UPDATE ${f.sql(f.tableName)} SET created = statement_timestamp() - interval '1 hour', ttl = interval '1 second'`; + } + const before = await state(f.sql, f.tableName); + await f.sql`CREATE TABLE ${f.sql(audit)} (event text)`; + await f.sql`CREATE FUNCTION ${f.sql(fn)}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO ${f.sql(audit)} VALUES (TG_OP); + RETURN NULL; + END + $$`; + const event = + operation === "insert" + ? f.sql`INSERT` + : operation === "delete" + ? f.sql`DELETE` + : f.sql`UPDATE`; + await f.sql`CREATE TRIGGER filter_rows BEFORE ${event} ON ${f.sql(f.tableName)} + FOR EACH ROW EXECUTE FUNCTION ${f.sql(fn)}()`; + const run = async (sql: postgres.Sql | postgres.TransactionSql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + const value = operation === "delete" ? undefined : 2; + // A pooled mismatch can sweep expired rows; retain the legacy row + // when specifically exercising the upsert's UPDATE trigger. + if (operation !== "expired-update") { + assert.equal(await store.cas(key, "wrong", value), false); + } + await assert.rejects( + store.cas( + key, + operation === "insert" || operation === "expired-update" + ? undefined + : 1, + value, + ), + ); + assert.equal((await sql`SELECT 1 AS healthy`)[0].healthy, 1); + }; + if (client === "pool") await run(f.sql); + else if (client === "transaction") await f.sql.begin(run); + else { + const sql = await f.sql.reserve(); + let transaction = client === "reserved-transaction"; + try { + if (transaction) await sql`BEGIN`; + await run(sql); + if (transaction) await sql`COMMIT`; + transaction = false; + } finally { + if (transaction) await sql`ROLLBACK`; + sql.release(); + } + } + assert.deepEqual(await state(f.sql, f.tableName), before); + assert.equal( + (await f.sql`SELECT count(*)::int AS size FROM ${f.sql(audit)}`)[0] + .size, + 0, + ); + } finally { + await f.sql`DROP TRIGGER IF EXISTS filter_rows ON ${f.sql(f.tableName)}`; + await f.sql`DROP FUNCTION IF EXISTS ${f.sql(fn)}()`; + await f.sql`DROP TABLE IF EXISTS ${f.sql(audit)}`; + await f.dispose(); + } + }, + ); + } +} + +for (const outer of [false, true]) { + dbTest( + `PostgresKvStore single CAS preserves reserved transaction ownership outer=${outer}`, + async () => { + const f = fixture(); + try { + await f.store.initialize(); + const sql = await f.sql.reserve(); + let transaction = outer; + try { + if (outer) await sql`BEGIN`; + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + assert.equal(await store.cas(key, undefined, 1), true); + assert.equal(await store.cas(key, "wrong", 2), false); + assert.equal(await store.cas(key, 1, 2), true); + if (outer) await sql`ROLLBACK`; + transaction = false; + } finally { + if (transaction) await sql`ROLLBACK`; + sql.release(); + } + assert.equal(await f.store.get(key), outer ? undefined : 2); + } finally { + await f.dispose(); + } + }, + ); +} + +dbTest( + "PostgresKvStore single CAS retains repeatable-read caller support", + async () => { + const f = fixture(); + try { + await f.store.set(key, 1); + await f.sql.begin("isolation level repeatable read", async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + assert.equal(await store.cas(key, 1, 2), true); + assert.equal(await store.cas(key, 1, 3), false); + }); + assert.equal(await f.store.get(key), 2); + } finally { + await f.dispose(); + } + }, +); + +dbTest( + "PostgresKvStore single CAS absence checks retain read-only caller support", + async () => { + const f = fixture(); + try { + await f.store.set(key, 1); + await f.sql.begin("read only", async (sql) => { + const store = new PostgresKvStore(sql, { + tableName: f.tableName, + initialized: true, + }); + assert.equal(await store.cas(["missing"], undefined, undefined), true); + assert.equal(await store.cas(key, undefined, undefined), false); + }); + } finally { + await f.dispose(); + } + }, +); diff --git a/packages/postgres/src/kv.expiry.test.ts b/packages/postgres/src/kv.expiry.test.ts index 36f5fa427..2908af566 100644 --- a/packages/postgres/src/kv.expiry.test.ts +++ b/packages/postgres/src/kv.expiry.test.ts @@ -864,8 +864,8 @@ test( `; assert.equal(written.fresh, true, "CAS must not backdate its TTL"); } - // These rows expire after the transaction began. The first CAS above - // has already put cleanup on cooldown, so they remain physically present. + // These rows expire after the transaction began. Caller-owned CAS + // defers cleanup, so they remain physically present. await tx` INSERT INTO ${tx(tableName)} (key, value, created, ttl) SELECT ARRAY[key], '"expired"'::jsonb, statement_timestamp(), INTERVAL '0' diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index fc985ad77..d00e1a798 100644 --- a/packages/postgres/src/kv.index.test.ts +++ b/packages/postgres/src/kv.index.test.ts @@ -138,6 +138,41 @@ test( }, ); +for (const difference of ["timezone", "offset"] as const) { + test( + `PostgresKvStore does not reuse an expiry index with a different ${difference}`, + { skip }, + async () => { + if (!dbUrl) return; + const { sql, tableName, statements } = database(); + const store = new PostgresKvStore(sql, { tableName }); + try { + await legacyTable(sql, tableName); + const expression = + difference === "timezone" + ? sql`timezone('Etc/GMT+1', timezone('UTC', created) + ttl)` + : sql`timezone('UTC', timezone('UTC', created) + ttl + interval '1 second')`; + await sql`CREATE INDEX ${sql(`near_match_${tableName}`)} ON ${sql(tableName)} + ((${expression})) WHERE ttl IS NOT NULL`; + assert.equal((await indexes(sql, tableName)).length, 1); + statements.length = 0; + await store.initialize(); + assert.equal((await indexes(sql, tableName)).length, 2); + assert(statements.some((query) => /CREATE INDEX/.test(query))); + statements.length = 0; + await new PostgresKvStore(sql, { tableName }).initialize(); + assert.equal((await indexes(sql, tableName)).length, 2); + assert( + !statements.some((query) => /LOCK TABLE|CREATE INDEX/.test(query)), + ); + } finally { + await store.drop(); + await sql.end(); + } + }, + ); +} + test( "PostgresKvStore leaves populated legacy tables unlocked until manual migration", { skip }, @@ -239,9 +274,10 @@ test( )} VALUES (ARRAY['null-created'], '1', NULL, INTERVAL '1 day')`; assert.equal(await store.get(["null-created"]), undefined); assert.equal( - await store.cas(["null-created"], undefined, "wrong"), - false, + await store.cas(["null-created"], undefined, "recovered"), + true, ); + assert.equal(await store.get(["null-created"]), "recovered"); } finally { await store.drop(); await sql.end(); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 0e1e1d553..bcd6951df 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -1,23 +1,65 @@ import type { + AtomicKvStore, KvKey, - KvStore, + KvStoreCasEntry, KvStoreListEntry, + KvStoreSetEntry, KvStoreSetOptions, } from "@fedify/fedify"; import { getLogger } from "@logtape/logtape"; -import type { JSONValue, Sql } from "postgres"; +import type { JSONValue, Sql, TransactionSql } from "postgres"; import { driverSerializesJson } from "./utils.ts"; const logger = getLogger(["fedify", "postgres", "kv"]); -function quoteIdentifier(identifier: string): string { - return `"${identifier.replaceAll('"', '""').replaceAll(".", '"."')}"`; -} - const expiryBatchSize = 1000; const expiryInterval = 60_000; const expiryCatchUpInterval = 1000; +function snapshotKey(key: KvKey): KvKey { + if (!Array.isArray(key) || key.length === 0) { + throw new TypeError("A key must be a nonempty array of strings."); + } + for (const part of key) { + if (typeof part !== "string") { + throw new TypeError("A key must be a dense array of strings."); + } + } + return [...key] as [string, ...string[]]; +} + +function snapshotValue(value: unknown): JSONValue { + const encoded = JSON.stringify(value); + if (encoded === undefined) + throw new TypeError("The value is not JSON encodable."); + return JSON.parse(encoded) as JSONValue; +} + +function snapshotTtl(options?: KvStoreSetOptions): string | null { + return options?.ttl == null + ? null + : Temporal.Duration.from(options.ttl).toString(); +} + +function snapshotBatch( + entries: readonly T[], + prepare: (entry: T) => U, +): U[] { + if (!Array.isArray(entries)) throw new TypeError("A batch must be an array."); + // Array.from visits holes, unlike map(), so sparse batches reject too. + return Array.from(entries, prepare); +} + +function uniqueKeys(entries: readonly { readonly key: KvKey }[]): void { + const seen = new Set(); + for (const { key } of entries) { + const encoded = JSON.stringify(key); + if (seen.has(encoded)) + throw new TypeError("Duplicate keys in an atomic mutation."); + seen.add(encoded); + } +} + /** * Options for the PostgreSQL key–value store. */ @@ -53,6 +95,24 @@ export interface PostgresKvStoreOptions { /** * A key–value store that uses PostgreSQL as the underlying storage. * + * Implements {@link AtomicKvStore} within one ordinary table or declarative + * partitioned table, using the default JSON codec and PostgreSQL `jsonb` + * equality. Object member order is ignored; array order is significant. + * TTLs use server time and PostgreSQL intervals at microsecond precision. + * Zero durations expire immediately; negative Temporal durations reject. + * Ordinary inheritance, active row-level security and irreversible value/JSON + * transforms are unsupported for batches. Batches are never split; backend + * field, index and memory limits apply, with index-storage failures arising + * only on attempted writes. + * + * Nonempty {@link casMany} and multi-key {@link deleteMany} require a pool or + * a READ COMMITTED transaction client; reserved clients reject those scopes. + * They serialize writes to the entire table. Caller transactions must be used + * exclusively and sequentially during each atomic call, across all wrappers + * and raw SQL. Nested savepoints do not isolate concurrent calls on one outer + * transaction. Successful calls remain provisional and retain locks until the + * caller commits. See the package README for scope and durability details. + * * @example * ```ts * import { createFederation } from "@fedify/fedify"; @@ -65,15 +125,14 @@ export interface PostgresKvStoreOptions { * }); * ``` */ -export class PostgresKvStore implements KvStore { - // postgres.js uses Sql<{}> to mean an untyped SQL client. - // oxlint-disable-next-line typescript/no-empty-object-type - readonly #sql: Sql<{}>; +export class PostgresKvStore implements AtomicKvStore { + readonly #sql: Sql | TransactionSql; readonly #tableName: string; readonly #unlogged: boolean; readonly #skipDdl: boolean; #initialized = false; #initializing?: Promise; + #batchInitializing?: Promise; #driverSerializesJson = false; #expiring?: Promise; #nextExpiry = 0; @@ -83,12 +142,7 @@ export class PostgresKvStore implements KvStore { * @param sql The PostgreSQL client to use. * @param options The options for the key–value store. */ - constructor( - // postgres.js uses Sql<{}> to mean an untyped SQL client. - // oxlint-disable-next-line typescript/no-empty-object-type - sql: Sql<{}>, - options: PostgresKvStoreOptions = {}, - ) { + constructor(sql: Sql | TransactionSql, options: PostgresKvStoreOptions = {}) { this.#sql = sql; this.#tableName = options.tableName ?? "fedify_kv_v2"; this.#unlogged = options.unlogged ?? false; @@ -179,19 +233,19 @@ export class PostgresKvStore implements KvStore { value: unknown, options?: KvStoreSetOptions | undefined, ): Promise { + key = snapshotKey(key); + value = snapshotValue(value); + const ttl = snapshotTtl(options); await this.initialize(); - const ttl = - options?.ttl == null - ? null - : Temporal.Duration.from(options.ttl).toString(); await this.#sql` INSERT INTO ${this.#sql(this.#tableName)} (key, value, created, ttl) - VALUES ( + SELECT ${key}, ${this.#json(value)}, statement_timestamp(), - ${ttl} - ) + ${ttl}::interval + WHERE ${ttl}::interval IS NULL + OR statement_timestamp() + ${ttl}::interval IS NOT NULL ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, @@ -211,7 +265,10 @@ export class PostgresKvStore implements KvStore { } /** - * {@inheritDoc KvStore.cas} + * {@inheritDoc AtomicKvStore.cas} + * Mutating forms use a table-wide write lock and transaction or savepoint. Caller-owned + * clients require exclusive sequential use; reserved clients are supported. + * A trigger-skipped matching write rejects and rolls back its effects. * @since 2.4.0 */ async cas( @@ -220,64 +277,486 @@ export class PostgresKvStore implements KvStore { newValue: unknown, options?: KvStoreSetOptions, ): Promise { - await this.initialize(); - const ttl = - options?.ttl == null - ? null - : Temporal.Duration.from(options.ttl).toString(); + key = snapshotKey(key); + expectedValue = + expectedValue === undefined ? undefined : snapshotValue(expectedValue); + newValue = newValue === undefined ? undefined : snapshotValue(newValue); + if (newValue === undefined && options?.ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } + const ttl = snapshotTtl(options); + const payload = JSON.stringify([ + { key, expected: expectedValue, value: newValue, ttl }, + ]); + await this.#initializeBatch(); + // Validate TTL ranges before locking; comparison and mutation below share + // one cutoff chosen after concurrent writers have finished. + await this.#validatePayload(this.#sql, payload, ttl === null ? 0 : 1); + // This pure absence check needs no mutation or write lock. if (expectedValue === undefined && newValue === undefined) { return (await this.get(key)) === undefined; } - let result; - if (expectedValue === undefined) { - result = await this.#sql` - INSERT INTO ${this.#sql(this.#tableName)} AS existing - (key, value, created, ttl) - VALUES ( - ${key}, - ${this.#json(newValue)}, - statement_timestamp(), - ${ttl} - ) - 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 <= statement_timestamp() - RETURNING key; - `; - } else if (newValue === undefined) { - result = await this.#sql` - DELETE FROM ${this.#sql(this.#tableName)} - WHERE key = ${key} - AND (ttl IS NULL OR created + ttl > statement_timestamp()) - AND value = ${this.#json(expectedValue)} - RETURNING key; + const applied = await this.#casScope(async (sql) => { + // A fresh comparison after this lock also protects absent keys from + // ordinary writers. Caller-owned locks survive until the outer commit. + await sql`LOCK TABLE ${sql(this.#tableName)} IN SHARE ROW EXCLUSIVE MODE`; + const table = sql.unsafe(await this.#resolvedTable(sql)); + const live = sql`key = ${key} AND + (ttl IS NULL OR created + ttl > statement_timestamp())`; + const matches = + expectedValue === undefined + ? sql`NOT EXISTS (SELECT 1 FROM ${table} WHERE ${live})` + : sql`EXISTS (SELECT 1 FROM ${table} WHERE ${live} + AND value = ${this.#json(expectedValue)})`; + const [comparison] = await sql` + SELECT ${matches} AS matches, + to_char(statement_timestamp() AT TIME ZONE 'UTC', + 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cutoff `; + if (!comparison.matches) return false; + const cutoff = sql`${comparison.cutoff}::text::timestamptz`; + let result; + if (expectedValue === undefined) { + result = await sql` + INSERT INTO ${sql(this.#tableName)} AS existing + (key, value, created, ttl) + VALUES ( + ${key}, + ${this.#json(newValue)}, + ${cutoff}, + ${ttl} + ) + ON CONFLICT (key) + DO UPDATE SET + value = EXCLUDED.value, + created = EXCLUDED.created, + ttl = EXCLUDED.ttl + WHERE NOT coalesce(existing.ttl IS NULL + OR existing.created + existing.ttl > ${cutoff}, false) + RETURNING key; + `; + } else if (newValue === undefined) { + result = await sql` + DELETE FROM ${sql(this.#tableName)} + WHERE key = ${key} + AND (ttl IS NULL OR created + ttl > ${cutoff}) + AND value = ${this.#json(expectedValue)} + RETURNING key; + `; + } else { + result = await sql` + UPDATE ${sql(this.#tableName)} + SET + value = ${this.#json(newValue)}, + created = ${cutoff}, + ttl = ${ttl} + WHERE key = ${key} + AND (ttl IS NULL OR created + ttl > ${cutoff}) + AND value = ${this.#json(expectedValue)} + RETURNING key; + `; + } + + if (result.length !== 1) { + throw new Error( + "A matching CAS mutation did not affect exactly one row.", + ); + } + return true; + }); + if (!this.#callerOwned()) await this.#expire(); + return applied; + } + + /** + * {@inheritDoc AtomicKvStore.getMany} + * @since 2.5.0 + */ + async getMany( + keys: readonly KvKey[], + ): Promise<(T | undefined)[]> { + const entries = snapshotBatch(keys, (key) => ({ key: snapshotKey(key) })); + if (entries.length === 0) return []; + const payload = JSON.stringify(entries); + await this.#initializeBatch(); + const sql = this.#sql; + const table = sql.unsafe(await this.#resolvedTable(sql)); + const rows = await sql` + WITH support AS (${this.#support(sql)}), input AS (${this.#input(sql, payload)}) + SELECT support.supported, input.ord, existing.value, + coalesce(existing.key IS NOT NULL AND + (existing.ttl IS NULL OR existing.created + existing.ttl > statement_timestamp()), false) + AS present + FROM support CROSS JOIN input + LEFT JOIN ${table} AS existing ON existing.key = input.key + ORDER BY input.ord + `; + this.#assertSupport(rows[0]?.supported); + if (rows.length !== entries.length) + throw new TypeError("Nonunique backing keys."); + return rows.map((row) => (row.present ? (row.value as T) : undefined)); + } + + /** + * {@inheritDoc AtomicKvStore.setMany} + * @since 2.5.0 + */ + async setMany(entries: readonly KvStoreSetEntry[]): Promise { + const prepared = snapshotBatch(entries, (entry) => ({ + key: snapshotKey(entry.key), + value: snapshotValue(entry.value), + ttl: snapshotTtl(entry.options), + })); + uniqueKeys(prepared); + if (prepared.length === 0) return; + const payload = JSON.stringify(prepared); + await this.#initializeBatch(); + await this.#writeBatch(this.#sql, payload); + if (!this.#callerOwned()) await this.#expire(); + } + + /** + * {@inheritDoc AtomicKvStore.deleteMany} + * More than one key requires a pool or a sequentially used READ COMMITTED + * transaction client; reserved clients reject that scope before SQL. + * @since 2.5.0 + */ + async deleteMany(keys: readonly KvKey[]): Promise { + const entries = snapshotBatch(keys, (key) => ({ key: snapshotKey(key) })); + uniqueKeys(entries); + if (entries.length === 0) return; + const payload = JSON.stringify(entries); + if (entries.length > 1) { + await this.#locked(async (sql) => { + await this.#deleteBatch(sql, payload); + return true; + }); } else { - result = await this.#sql` - UPDATE ${this.#sql(this.#tableName)} - SET - value = ${this.#json(newValue)}, - created = statement_timestamp(), - ttl = ${ttl} - WHERE key = ${key} - AND (ttl IS NULL OR created + ttl > statement_timestamp()) - AND value = ${this.#json(expectedValue)} - RETURNING key; - `; + await this.#initializeBatch(); + await this.#deleteBatch(this.#sql, payload); } + if (!this.#callerOwned()) await this.#expire(); + } - await this.#expire(); - return result.length > 0; + /** + * {@inheritDoc AtomicKvStore.casMany} + * Nonempty batches require a pool or an exclusively, sequentially used + * READ COMMITTED transaction client. Reserved clients reject before SQL. + * The table-wide write lock survives success until the outer commit. + * @since 2.5.0 + */ + async casMany(entries: readonly KvStoreCasEntry[]): Promise { + const prepared = snapshotBatch(entries, (entry) => { + const expected = entry.expectedValue; + const value = entry.newValue; + const ttl = snapshotTtl(entry.options); + if (value === undefined && ttl !== null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } + return { + key: snapshotKey(entry.key), + absent: expected === undefined, + expected: expected === undefined ? undefined : snapshotValue(expected), + remove: value === undefined, + value: value === undefined ? undefined : snapshotValue(value), + ttl, + }; + }); + uniqueKeys(prepared); + if (prepared.length === 0) return true; + const payload = JSON.stringify(prepared); + const deletions = JSON.stringify(prepared.filter((entry) => entry.remove)); + const writes = JSON.stringify(prepared.filter((entry) => !entry.remove)); + const ttls = prepared.filter((entry) => entry.ttl !== null).length; + const mismatch = new Error("Atomic comparison mismatch."); + try { + await this.#locked(async (sql) => { + const table = sql.unsafe(await this.#resolvedTable(sql)); + const [result] = await sql` + WITH input AS (${this.#input(sql, payload)}) + SELECT bool_and(CASE WHEN (input.item->>'absent')::boolean THEN + NOT coalesce(existing.key IS NOT NULL AND + (existing.ttl IS NULL OR existing.created + existing.ttl > statement_timestamp()), false) + ELSE coalesce(existing.key IS NOT NULL AND + (existing.ttl IS NULL OR existing.created + existing.ttl > statement_timestamp()), false) + AND existing.value = input.item->'expected' END) AS matches, + count(*)::int AS size, + count((input.item->>'ttl')::interval)::int AS ttls, + count(statement_timestamp() + (input.item->>'ttl')::interval)::int AS expiry, + to_char(statement_timestamp() AT TIME ZONE 'UTC', + 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cutoff + FROM input + LEFT JOIN ${table} AS existing ON existing.key = input.key + `; + if (result.size !== prepared.length) + throw new TypeError("Nonunique backing keys."); + if (result.ttls !== ttls || result.expiry !== ttls) { + throw new TypeError("Invalid TTL representation."); + } + if (!result.matches) throw mismatch; + if (deletions !== "[]") await this.#deleteBatch(sql, deletions); + if (writes !== "[]") await this.#writeBatch(sql, writes, result.cutoff); + return true; + }); + } catch (error) { + // The driver has rolled back before this private sentinel is translated. + // Never hide rollback failures or any backend error as a mismatch. + if (error === mismatch) return false; + throw error; + } + if (!this.#callerOwned()) await this.#expire(); + return true; + } + + #input(sql: Sql | TransactionSql, payload: string) { + return sql` + SELECT item, ord, + ARRAY(SELECT part FROM jsonb_array_elements_text(item->'key') + WITH ORDINALITY AS parts(part, position) ORDER BY position) AS key + FROM jsonb_array_elements(${payload}::text::jsonb) + WITH ORDINALITY AS items(item, ord) + `; + } + + async #casScope( + operation: (sql: Sql | TransactionSql) => Promise, + ): Promise { + const sql = this.#sql; + if ("savepoint" in sql && typeof sql.savepoint === "function") { + return await sql.savepoint(operation); + } + if (!("release" in sql)) { + return await (sql as Sql).begin( + "isolation level read committed", + operation, + ); + } + + // Reserved clients do not expose transaction status or savepoint(). Probe + // for an outer transaction without risking a COMMIT of caller-owned work. + const name = `fedify_cas_${crypto.randomUUID().replaceAll("-", "")}`; + let nested = true; + try { + await sql`SAVEPOINT ${sql(name)}`; + } catch (error) { + if ( + !( + typeof error === "object" && + error !== null && + "code" in error && + error.code === "25P01" + ) + ) + throw error; + nested = false; + await sql`BEGIN`; + } + try { + const result = await operation(sql); + if (nested) await sql`RELEASE SAVEPOINT ${sql(name)}`; + else await sql`COMMIT`; + return result; + } catch (error) { + // Rollback errors must reject too, never turn into a comparison mismatch. + if (nested) { + await sql`ROLLBACK TO SAVEPOINT ${sql(name)}`; + await sql`RELEASE SAVEPOINT ${sql(name)}`; + } else await sql`ROLLBACK`; + throw error; + } + } + + #escapedTableName(sql: Sql | TransactionSql): string { + // Scoped clients do not expose options. The driver's identifier helper + // contains the exact escaped SQL name, including its column.to transform. + const identifier = sql(this.#tableName); + if (!("value" in identifier) || typeof identifier.value !== "string") { + throw new TypeError( + "The PostgreSQL client must expose its escaped identifier.", + ); + } + return identifier.value; + } + + async #resolvedTable(sql: Sql | TransactionSql): Promise { + // Qualify the actual relation so internal CTE names cannot shadow it. + // PostgreSQL formats both components; do not transform them a second time. + const [relation] = await sql` + SELECT pg_catalog.format('%I.%I', n.nspname, c.relname) AS name + FROM pg_catalog.pg_class AS c + JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace + WHERE c.oid = pg_catalog.to_regclass(${this.#escapedTableName(sql)}) + `; + if (typeof relation?.name !== "string") { + throw new TypeError("The backing table could not be resolved."); + } + return relation.name; + } + + #support(sql: Sql | TransactionSql) { + // Ordinary inheritance does not extend the parent's primary-key constraint. + // Declarative partitions are different: their unique key includes the + // partition key. Resolve through the driver's identifier transforms once. + return sql` + SELECT c.relkind IN ('r', 'p') + AND NOT pg_catalog.row_security_active(c.oid) + AND (c.relispartition OR NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_inherits WHERE inhrelid = c.oid + )) + AND NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_inherits AS i + JOIN pg_catalog.pg_class AS child ON child.oid = i.inhrelid + WHERE i.inhparent = c.oid AND NOT child.relispartition + ) AS supported + FROM pg_catalog.pg_class AS c + WHERE c.oid = pg_catalog.to_regclass(${this.#escapedTableName(sql)}) + `; + } + + #assertSupport(supported: unknown): void { + if (supported !== true) { + throw new TypeError( + "Atomic batches require an ordinary table or declarative partition without ordinary inheritance or active row-level security.", + ); + } + } + + async #validatePayload( + sql: Sql | TransactionSql, + payload: string, + ttls: number, + ): Promise { + const [result] = await sql` + SELECT count((item->>'ttl')::interval)::int AS ttls, + count(statement_timestamp() + (item->>'ttl')::interval)::int AS expiry + FROM jsonb_array_elements(${payload}::text::jsonb) AS items(item) + `; + if (result.ttls !== ttls || result.expiry !== ttls) { + throw new TypeError("Invalid TTL representation."); + } + } + + async #writeBatch( + sql: Sql | TransactionSql, + payload: string, + cutoff?: string, + ): Promise { + const [support] = await this.#support(sql); + this.#assertSupport(support?.supported); + const origin = + cutoff === undefined + ? sql`statement_timestamp()` + : sql`${cutoff}::text::timestamptz`; + const [result] = await sql` + WITH support AS (${this.#support(sql)}), input AS (${this.#input(sql, payload)}), + written AS ( + INSERT INTO ${sql(this.#tableName)} (key, value, created, ttl) + SELECT input.key, input.item->'value', ${origin}, (input.item->>'ttl')::interval + FROM input CROSS JOIN support + WHERE support.supported AND ((input.item->>'ttl')::interval IS NULL + OR ${origin} + (input.item->>'ttl')::interval IS NOT NULL) + ORDER BY input.key + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, + created = EXCLUDED.created, ttl = EXCLUDED.ttl + RETURNING 1 + ) + -- Fail inside the statement so pooled/reserved clients roll back every + -- write and trigger side effect when a BEFORE trigger skips a row. + SELECT supported, 1 / CASE WHEN NOT supported OR + (SELECT count(*) FROM written) = (SELECT count(*) FROM input) + THEN 1 ELSE 0 END AS complete FROM support + `; + this.#assertSupport(result?.supported); + } + + async #deleteBatch( + sql: Sql | TransactionSql, + payload: string, + ): Promise { + const [support] = await this.#support(sql); + this.#assertSupport(support?.supported); + const table = sql.unsafe(await this.#resolvedTable(sql)); + const [result] = await sql` + WITH support AS (${this.#support(sql)}), input AS (${this.#input(sql, payload)}), + targets AS MATERIALIZED ( + SELECT existing.key FROM ${table} AS existing + WHERE (SELECT supported FROM support) + AND existing.key IN (SELECT key FROM input) + ORDER BY existing.key FOR UPDATE OF existing + ), deleted AS ( + DELETE FROM ${sql(this.#tableName)} AS existing + WHERE (SELECT supported FROM support) + AND existing.key IN (SELECT key FROM targets) + RETURNING 1 + ) + -- Count locked targets, not requested keys: absent/deleted keys are + -- no-ops. A skipped target must abort this entire statement. + SELECT supported, 1 / CASE WHEN NOT supported OR + (SELECT count(*) FROM deleted) = (SELECT count(*) FROM targets) + THEN 1 ELSE 0 END AS complete FROM support + `; + this.#assertSupport(result?.supported); + } + + async #initializeBatch(): Promise { + if (!this.#callerOwned() || this.#skipDdl) return await this.initialize(); + // Caller rollback can undo DDL while this object and a fulfilled ordinary + // initialization promise survive. Retain only concurrent in-flight setup. + const pending = (this.#batchInitializing ??= (async () => { + await this.#initializeTable(); + this.#driverSerializesJson = await driverSerializesJson(this.#sql); + })()); + try { + await pending; + } finally { + if (this.#batchInitializing === pending) + this.#batchInitializing = undefined; + } + } + + async #locked( + operation: (sql: Sql | TransactionSql) => Promise, + ): Promise { + const transaction = + "savepoint" in this.#sql && typeof this.#sql.savepoint === "function"; + if ( + !transaction && + ("release" in this.#sql || + !("begin" in this.#sql) || + typeof this.#sql.begin !== "function") + ) { + throw new TypeError( + "This atomic scope requires a pool or READ COMMITTED transaction client.", + ); + } + if (transaction) { + const [settings] = + await this.#sql`SELECT current_setting('transaction_isolation') AS isolation`; + if ( + settings.isolation !== "read committed" && + settings.isolation !== "read uncommitted" + ) { + throw new TypeError( + "This atomic scope requires READ COMMITTED isolation.", + ); + } + } + await this.#initializeBatch(); + const run = async (sql: Sql | TransactionSql): Promise => { + await sql`LOCK TABLE ${sql(this.#tableName)} IN SHARE ROW EXCLUSIVE MODE`; + const [support] = await this.#support(sql); + this.#assertSupport(support?.supported); + return await operation(sql); + }; + return transaction + ? await (this.#sql as TransactionSql).savepoint(run) + : await (this.#sql as Sql).begin("isolation level read committed", run); } /** - * {@inheritDoc KvStore.list} + * {@inheritDoc AtomicKvStore.list} * @since 1.10.0 */ async *list(prefix?: KvKey): AsyncIterable { @@ -340,8 +819,15 @@ export class PostgresKvStore implements KvStore { } async #initializeTable(): Promise { + // Caller rollback can remove a previously created table. Check each time, + // but avoid write-class DDL when reads reuse a table in READ ONLY scope. + const table = this.#escapedTableName(this.#sql); + const [existing] = await this.#sql` + SELECT to_regclass(${table}) IS NOT NULL AS present + `; if (this.#unlogged) { - await this.#sql` + if (!existing.present) + await this.#sql` CREATE UNLOGGED TABLE IF NOT EXISTS ${this.#sql(this.#tableName)} ( key text[] PRIMARY KEY, value jsonb NOT NULL, @@ -350,7 +836,8 @@ export class PostgresKvStore implements KvStore { ); `; } else { - await this.#sql` + if (!existing.present) + await this.#sql` CREATE TABLE IF NOT EXISTS ${this.#sql(this.#tableName)} ( key text[] PRIMARY KEY, value jsonb NOT NULL, @@ -361,7 +848,7 @@ export class PostgresKvStore implements KvStore { const persistence = await this.#sql` SELECT relpersistence FROM pg_class - WHERE oid = to_regclass(${quoteIdentifier(this.#tableName)}); + WHERE oid = to_regclass(${table}); `; if (persistence[0]?.relpersistence === "u") { await this.#sql` @@ -380,12 +867,14 @@ export class PostgresKvStore implements KvStore { async #initializeExpiryIndex(): Promise { try { - await this.#sql.begin("isolation level read committed", async (sql) => { - await sql`SET LOCAL lock_timeout = '250ms'`; - await sql`SET LOCAL statement_timeout = '2s'`; - // Resolve identifiers through the driver once. DDL below is quoted - // by PostgreSQL, without applying identifier transforms a second time. - const [table] = await sql` + await (this.#sql as Sql).begin( + "isolation level read committed", + async (sql) => { + await sql`SET LOCAL lock_timeout = '250ms'`; + await sql`SET LOCAL statement_timeout = '2s'`; + // Resolve identifiers through the driver once. DDL below is quoted + // by PostgreSQL, without applying identifier transforms a second time. + const [table] = await sql` SELECT c.oid, c.relkind, c.relrowsecurity, pg_has_role(current_user, c.relowner, 'USAGE') AS owner, EXISTS (SELECT 1 FROM pg_inherits WHERE inhparent = c.oid) AS children, @@ -397,8 +886,8 @@ export class PostgresKvStore implements KvStore { WHERE c.oid = (SELECT typrelid FROM pg_type WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) `; - const indexed = async () => { - const [result] = await sql` + const indexed = async () => { + const [result] = await sql` SELECT EXISTS ( SELECT 1 FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid @@ -409,7 +898,9 @@ export class PostgresKvStore implements KvStore { AND NOT i.indisunique AND i.indnatts = 1 AND a.amname = 'btree' AND i.indkey[0] = 0 AND o.opcdefault AND o.opcintype = 'timestamptz'::regtype - AND pg_get_indexdef(i.indexrelid, 1, false) IN ( + -- Index-column DDL adds version-dependent outer parentheses + -- for AT TIME ZONE. Compare the catalog expression itself. + AND pg_get_expr(i.indexprs, i.indrelid, false) IN ( ${"timezone('UTC'::text, (timezone('UTC'::text, created) + ttl))"}, ${"(((created AT TIME ZONE 'UTC'::text) + ttl) AT TIME ZONE 'UTC'::text)"} ) @@ -417,59 +908,60 @@ export class PostgresKvStore implements KvStore { pg_get_expr(i.indpred, i.indrelid) = '(ttl IS NOT NULL)') ) AS present `; - return result.present; - }; - if (await indexed()) return; - const manual = () => - logger.warn( - "Provision the expiry index for {tableName} manually; see {guide}.", - { - tableName: table.name, - guide: "https://fedify.dev/manual/kv#postgresql-expiry-index", - }, - ); - if ( - !table.owner || - table.relrowsecurity || - (table.relkind !== "r" && table.relkind !== "p") || - (table.relkind === "r" && table.children) - ) { - manual(); - return; - } - const empty = async () => { - const [result] = await sql.unsafe( - `SELECT NOT EXISTS (SELECT 1 FROM ${table.name} LIMIT 1) AS empty`, - ); - return result.empty; - }; - // Never queue a write-blocking lock against a populated legacy table. - if (!(await empty())) { - manual(); - return; - } - const [lock] = await sql` + return result.present; + }; + if (await indexed()) return; + const manual = () => + logger.warn( + "Provision the expiry index for {tableName} manually; see {guide}.", + { + tableName: table.name, + guide: "https://fedify.dev/manual/kv#postgresql-expiry-index", + }, + ); + if ( + !table.owner || + table.relrowsecurity || + (table.relkind !== "r" && table.relkind !== "p") || + (table.relkind === "r" && table.children) + ) { + manual(); + return; + } + const empty = async () => { + const [result] = await sql.unsafe( + `SELECT NOT EXISTS (SELECT 1 FROM ${table.name} LIMIT 1) AS empty`, + ); + return result.empty; + }; + // Never queue a write-blocking lock against a populated legacy table. + if (!(await empty())) { + manual(); + return; + } + const [lock] = await sql` SELECT pg_try_advisory_xact_lock( (hashtext('fedify.postgres.kv.index')::bigint << 32) | ${table.oid}::bigint ) AS acquired `; - if (!lock.acquired) return; - await sql.unsafe(`LOCK TABLE ${table.name} IN SHARE MODE`); - // READ COMMITTED gives this recheck a fresh snapshot after the lock. - if (await indexed()) return; - if (!(await empty())) { - manual(); - return; - } - const [ddl] = await sql` + if (!lock.acquired) return; + await sql.unsafe(`LOCK TABLE ${table.name} IN SHARE MODE`); + // READ COMMITTED gives this recheck a fresh snapshot after the lock. + if (await indexed()) return; + if (!(await empty())) { + manual(); + return; + } + const [ddl] = await sql` SELECT format( 'CREATE INDEX %I ON %s ((timezone(''UTC'', timezone(''UTC'', created) + ttl))) WHERE ttl IS NOT NULL', ${table.idx}::text, ${table.name}::text ) AS command `; - await sql.unsafe(ddl.command); - }); + await sql.unsafe(ddl.command); + }, + ); } catch (error) { // This private transaction has rolled back. Optional index setup must // not abort a caller's transaction or disable an otherwise usable store. diff --git a/packages/postgres/src/utils.ts b/packages/postgres/src/utils.ts index 7d93d3072..7f92cbd7e 100644 --- a/packages/postgres/src/utils.ts +++ b/packages/postgres/src/utils.ts @@ -1,8 +1,8 @@ -import type { Sql } from "postgres"; +import type { Sql, TransactionSql } from "postgres"; -// postgres.js uses Sql<{}> to mean an untyped SQL client. -// oxlint-disable-next-line typescript/no-empty-object-type -export async function driverSerializesJson(sql: Sql<{}>): Promise { +export async function driverSerializesJson( + sql: Sql | TransactionSql, +): Promise { const result = await sql`SELECT ${sql.json('{"foo":1}')}::jsonb AS test;`; return result[0].test === '{"foo":1}'; }