From 4f82a6a5053398471b74d14fdfbd0ac70b761e58 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 02:46:31 +0900 Subject: [PATCH 01/12] Add atomic PostgreSQL KV operations Add ordered snapshots and atomic multi-key writes, deletions, and CAS through the existing PostgresKvStore. Use database table locks for conditional transitions and multi-key deletion so absent keys remain protected against ordinary and legacy writers without changing schema. Keep work on the scoped SQL client and document caller transactions, reserved-client limits, JSONB comparison, TTLs, and durability boundaries. Cover independent-client races, rollback, and runtime compatibility. Closes https://github.com/fedify-dev/fedify/issues/1364 Related to https://github.com/fedify-dev/fedify/issues/1361 Assisted-by: Codex:gpt-6.1-sol Assisted-by: OpenCode:kimi-k3.1-pro Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-fable-5-2 --- CHANGES.md | 7 + changes.d/postgres/atomic-kv-store.md | 11 + claude-plugin/skills/fedify/SKILL.md | 26 +- docs/manual/kv.md | 83 +- packages/postgres/README.md | 102 ++ packages/postgres/src/kv.atomic.test.ts | 1204 +++++++++++++++++++++++ packages/postgres/src/kv.index.test.ts | 5 +- packages/postgres/src/kv.ts | 516 ++++++++-- packages/postgres/src/utils.ts | 8 +- 9 files changed, 1868 insertions(+), 94 deletions(-) create mode 100644 changes.d/postgres/atomic-kv-store.md create mode 100644 packages/postgres/src/kv.atomic.test.ts diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..0dc2cce94 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -429,10 +429,17 @@ 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 connections reject these + scopes. [[#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..c5deff9b3 --- /dev/null +++ b/changes.d/postgres/atomic-kv-store.md @@ -0,0 +1,11 @@ +--- +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 connections reject these + scopes. [[#1361], [#1364], [#1429]] diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b7f3a6665..c02d0e78b 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,23 @@ 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. +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. 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 +556,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..f298bf8dd 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -318,6 +318,68 @@ 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. `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. + +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 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 +474,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 +1072,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 +1091,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 +1705,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..456e42b08 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -60,3 +60,105 @@ 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. +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. + +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 start best-effort expiry cleanup after +commit; caller-owned batches 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.test.ts b/packages/postgres/src/kv.atomic.test.ts new file mode 100644 index 000000000..28e4f0715 --- /dev/null +++ b/packages/postgres/src/kv.atomic.test.ts @@ -0,0 +1,1204 @@ +// 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; + }); + 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; + }); + 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 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({ microseconds: 100_001 }), + }, + }, + { + 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 '0.100001 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; + }); + 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; + }); + 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 its final statement", + 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(); + } + }, +); diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index fc985ad77..4926c1de4 100644 --- a/packages/postgres/src/kv.index.test.ts +++ b/packages/postgres/src/kv.index.test.ts @@ -239,9 +239,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..9556fc557 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -1,11 +1,13 @@ 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"]); @@ -18,6 +20,50 @@ 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 +99,23 @@ 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 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 +128,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 +145,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 +236,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 +268,7 @@ export class PostgresKvStore implements KvStore { } /** - * {@inheritDoc KvStore.cas} + * {@inheritDoc AtomicKvStore.cas} * @since 2.4.0 */ async cas( @@ -220,11 +277,21 @@ export class PostgresKvStore implements KvStore { newValue: unknown, options?: KvStoreSetOptions, ): Promise { + 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.initialize(); - const ttl = - options?.ttl == null - ? null - : Temporal.Duration.from(options.ttl).toString(); + // Range validation is not an expiry decision. The final CAS statement + // alone supplies its liveness cutoff and replacement's TTL origin. + await this.#validatePayload(this.#sql, payload, ttl === null ? 0 : 1); if (expectedValue === undefined && newValue === undefined) { return (await this.get(key)) === undefined; @@ -246,8 +313,8 @@ export class PostgresKvStore implements KvStore { value = EXCLUDED.value, created = EXCLUDED.created, ttl = EXCLUDED.ttl - WHERE existing.ttl IS NOT NULL - AND existing.created + existing.ttl <= statement_timestamp() + WHERE NOT coalesce(existing.ttl IS NULL + OR existing.created + existing.ttl > statement_timestamp(), false) RETURNING key; `; } else if (newValue === undefined) { @@ -277,7 +344,297 @@ export class PostgresKvStore implements KvStore { } /** - * {@inheritDoc KvStore.list} + * {@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 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 ${sql(this.#tableName)} 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 { + await this.#initializeBatch(); + await this.#deleteBatch(this.#sql, payload); + } + if (!this.#callerOwned()) await this.#expire(); + } + + /** + * {@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 [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 ${sql(this.#tableName)} 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) + `; + } + + #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 (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 = (SELECT typrelid FROM pg_catalog.pg_type + WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) + `; + } + + #assertSupport(supported: unknown): void { + if (supported !== true) { + throw new TypeError( + "Atomic batches require an ordinary table or declarative partition without ordinary inheritance.", + ); + } + } + + 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 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 + ) + SELECT supported, (SELECT count(*) FROM written) AS size FROM support + `; + this.#assertSupport(result?.supported); + } + + async #deleteBatch( + sql: Sql | TransactionSql, + payload: string, + ): Promise { + const [result] = await sql` + WITH support AS (${this.#support(sql)}), input AS (${this.#input(sql, payload)}), + deleted AS ( + DELETE FROM ${sql(this.#tableName)} AS existing + WHERE (SELECT supported FROM support) + AND existing.key IN (SELECT key FROM input) + RETURNING 1 + ) + SELECT supported, (SELECT count(*) FROM deleted) AS size 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 AtomicKvStore.list} * @since 1.10.0 */ async *list(prefix?: KvKey): AsyncIterable { @@ -380,12 +737,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 +756,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 @@ -417,59 +776,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}'; } From 548b3fa03243305fed493c471c4166e28f52c750 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 03:56:40 +0900 Subject: [PATCH 02/12] Allow PostgreSQL batch reads in read-only mode Check table existence before issuing CREATE TABLE during initialization. Continue checking after caller rollback and retain the logged-table repair, while allowing batch reads to reuse an existing table inside read-only transactions. Cover repeated transaction-client reads, reserved-client reads and transaction health with a regression that fails before this fix. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4554407393 Assisted-by: Codex:gpt-6.1-sol Assisted-by: CodeRabbit:unknown --- packages/postgres/src/kv.atomic.test.ts | 32 +++++++++++++++++++++++++ packages/postgres/src/kv.ts | 11 +++++++-- 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index 28e4f0715..f73b76b29 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -545,6 +545,38 @@ dbTest( }, ); +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 () => { diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 9556fc557..8b9251077 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -697,8 +697,14 @@ export class PostgresKvStore implements AtomicKvStore { } 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 [existing] = await this.#sql` + SELECT to_regclass(${quoteIdentifier(this.#tableName)}) 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, @@ -707,7 +713,8 @@ export class PostgresKvStore implements AtomicKvStore { ); `; } 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, From b97cb513958af34674f98fac4c1ce24c1f7811dd Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:00:27 +0900 Subject: [PATCH 03/12] Document PostgreSQL CAS deletion TTL rejection Tell existing callers to omit the ttl option when deleting with cas(). The option now rejects rather than being silently ignored. Keep the materialized changelog synchronized with its source fragment. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4554407360 Assisted-by: Codex:gpt-6.1-sol Assisted-by: CodeRabbit:unknown --- changes.d/postgres/atomic-kv-store.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/changes.d/postgres/atomic-kv-store.md b/changes.d/postgres/atomic-kv-store.md index c5deff9b3..2c5e76ca8 100644 --- a/changes.d/postgres/atomic-kv-store.md +++ b/changes.d/postgres/atomic-kv-store.md @@ -8,4 +8,6 @@ links: to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional batches and multi-key deletions require pooled or sequentially used READ COMMITTED transaction clients; reserved connections reject these - scopes. [[#1361], [#1364], [#1429]] + scopes. `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]] From 4145ad7003c840016a6281ba1c74ea27b2571073 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:46:46 +0900 Subject: [PATCH 04/12] Resolve PostgreSQL backing table identifiers Use the client's escaped identifier for existence and persistence probes so column transforms apply to the same relation as the DDL. Qualify batch reads through the resolved schema and table to avoid CTE name shadowing. Cover transformed read-only reads, logged-table repair, raw-name collisions and backing tables named after the internal CTEs. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4555056502 https://github.com/fedify-dev/fedify/pull/1429#discussion_r4555149330 https://github.com/fedify-dev/fedify/pull/1429#discussion_r4555149338 Assisted-by: Codex:gpt-6.1-sol Assisted-by: CodeRabbit:unknown Assisted-by: Codex:unknown --- packages/postgres/src/kv.atomic.test.ts | 100 ++++++++++++++++++++++++ packages/postgres/src/kv.ts | 39 +++++++-- 2 files changed, 131 insertions(+), 8 deletions(-) diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index f73b76b29..eb555f14b 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -1234,3 +1234,103 @@ dbTest( } }, ); + +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"]) { + 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(); + } + }, +); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 8b9251077..27e949639 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -12,10 +12,6 @@ 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; @@ -355,6 +351,7 @@ export class PostgresKvStore implements AtomicKvStore { 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, @@ -362,7 +359,7 @@ export class PostgresKvStore implements AtomicKvStore { (existing.ttl IS NULL OR existing.created + existing.ttl > statement_timestamp()), false) AS present FROM support CROSS JOIN input - LEFT JOIN ${sql(this.#tableName)} AS existing ON existing.key = input.key + LEFT JOIN ${table} AS existing ON existing.key = input.key ORDER BY input.ord `; this.#assertSupport(rows[0]?.supported); @@ -445,6 +442,7 @@ export class PostgresKvStore implements AtomicKvStore { 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 @@ -459,7 +457,7 @@ export class PostgresKvStore implements AtomicKvStore { to_char(statement_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cutoff FROM input - LEFT JOIN ${sql(this.#tableName)} AS existing ON existing.key = input.key + LEFT JOIN ${table} AS existing ON existing.key = input.key `; if (result.size !== prepared.length) throw new TypeError("Nonunique backing keys."); @@ -491,6 +489,22 @@ export class PostgresKvStore implements AtomicKvStore { `; } + 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 = (SELECT typrelid FROM pg_catalog.pg_type + WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) + `; + 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 @@ -699,8 +713,17 @@ export class PostgresKvStore implements AtomicKvStore { 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. + // Scoped clients do not expose options. The driver's identifier helper + // contains the exact escaped SQL name, including its column.to transform. + const identifier = this.#sql(this.#tableName); + if (!("value" in identifier) || typeof identifier.value !== "string") { + throw new TypeError( + "The PostgreSQL client must expose its escaped identifier.", + ); + } + const table = identifier.value; const [existing] = await this.#sql` - SELECT to_regclass(${quoteIdentifier(this.#tableName)}) IS NOT NULL AS present + SELECT to_regclass(${table}) IS NOT NULL AS present `; if (this.#unlogged) { if (!existing.present) @@ -725,7 +748,7 @@ export class PostgresKvStore implements AtomicKvStore { 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` From d2e7cae30b707bb96d91de0ea5c4ed6613b1a35c Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:47:35 +0900 Subject: [PATCH 05/12] Reject PostgreSQL batches under active RLS Reject nonempty batches when row-level security applies to the current role. Policies can otherwise filter a deletion while a replacement in the same conditional batch commits, violating its all-or-nothing scope. Cover selective DELETE filtering and caller transaction health with a non-owner role, and document the unsupported scope in the adapter API, README, manual, user skill and changelog. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4555149311 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Codex:unknown --- CHANGES.md | 4 +- changes.d/postgres/atomic-kv-store.md | 8 ++- claude-plugin/skills/fedify/SKILL.md | 13 ++-- docs/manual/kv.md | 4 +- packages/postgres/README.md | 2 + packages/postgres/src/kv.atomic-rls.test.ts | 76 +++++++++++++++++++++ packages/postgres/src/kv.ts | 10 +-- 7 files changed, 102 insertions(+), 15 deletions(-) create mode 100644 packages/postgres/src/kv.atomic-rls.test.ts diff --git a/CHANGES.md b/CHANGES.md index 0dc2cce94..8b66bb48e 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -433,7 +433,9 @@ To be released. to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional batches and multi-key deletions require pooled or sequentially used READ COMMITTED transaction clients; reserved connections reject these - scopes. [[#1361], [#1364], [#1429]] + scopes. `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 diff --git a/changes.d/postgres/atomic-kv-store.md b/changes.d/postgres/atomic-kv-store.md index 2c5e76ca8..b469f3f59 100644 --- a/changes.d/postgres/atomic-kv-store.md +++ b/changes.d/postgres/atomic-kv-store.md @@ -8,6 +8,8 @@ links: to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional batches and multi-key deletions require pooled or sequentially used READ COMMITTED transaction clients; reserved connections reject these - scopes. `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]] + scopes. Nonempty batches reject active row-level security to prevent + policies from filtering individual mutations. `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 c02d0e78b..e3dd826ba 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -529,12 +529,13 @@ 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. 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. +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. diff --git a/docs/manual/kv.md b/docs/manual/kv.md index f298bf8dd..a8c850ee2 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -325,7 +325,9 @@ const federation = createFederation({ `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. `getMany()` returns one +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`. diff --git a/packages/postgres/README.md b/packages/postgres/README.md index 456e42b08..2bd92a1fe 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -68,6 +68,8 @@ 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. 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.ts b/packages/postgres/src/kv.ts index 27e949639..ef40d812e 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -100,9 +100,10 @@ export interface PostgresKvStoreOptions { * 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 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. + * 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. @@ -511,6 +512,7 @@ export class PostgresKvStore implements AtomicKvStore { // 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 )) @@ -528,7 +530,7 @@ export class PostgresKvStore implements AtomicKvStore { #assertSupport(supported: unknown): void { if (supported !== true) { throw new TypeError( - "Atomic batches require an ordinary table or declarative partition without ordinary inheritance.", + "Atomic batches require an ordinary table or declarative partition without ordinary inheritance or active row-level security.", ); } } From 1f0f8fef699bedf1c259aadafefb0a756c59f45c Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 09:32:00 +0900 Subject: [PATCH 06/12] Recognize expiry indexes on PostgreSQL 18.7 Compare the catalog expression rather than index-column DDL. PostgreSQL 18.7 adds outer parentheses around AT TIME ZONE index definitions, so the old exact DDL comparison missed compatible indexes and could provision a duplicate index. Keep the expression, predicate and index-shape checks. Retain the original one-index and no-DDL assertions and cover rejection of near-match timezone and one-second-offset expressions. https://github.com/fedify-dev/fedify/pull/1429 https://github.com/fedify-dev/fedify/actions/runs/43333332595 https://www.postgresql.org/docs/release/18.7/ Changelog: none Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-opus-5-5 --- packages/postgres/src/kv.index.test.ts | 35 ++++++++++++++++++++++++++ packages/postgres/src/kv.ts | 4 ++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index 4926c1de4..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 }, diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index ef40d812e..9ffcaad71 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -800,7 +800,9 @@ export class PostgresKvStore implements AtomicKvStore { 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)"} ) From 685a6dd2e850b44b23649125a07c0b33e7225a36 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:24:54 +0900 Subject: [PATCH 07/12] Roll back trigger-skipped PostgreSQL batches Assert affected row counts inside each SQL mutation statement so a BEFORE trigger returning NULL cannot commit a partial batch. Count locked deletion targets to preserve absent and concurrently deleted keys as no-ops, and cover trigger side effects and caller scopes. Regenerate the materialized changelog after rebasing onto main. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4558928924 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- CHANGES.md | 8 +- packages/postgres/src/kv.atomic.test.ts | 147 +++++++++++++++++++++++- packages/postgres/src/kv.ts | 22 +++- 3 files changed, 169 insertions(+), 8 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 8b66bb48e..da60f95e6 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -433,9 +433,11 @@ To be released. to `PostgresKvStore` through the `AtomicKvStore` interface. Conditional batches and multi-key deletions require pooled or sequentially used READ COMMITTED transaction clients; reserved connections reject these - scopes. `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]] + scopes. Nonempty batches reject active row-level security to prevent + policies from filtering individual mutations. `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 diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index eb555f14b..64865d744 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -1302,7 +1302,13 @@ dbTest( try { await sql`CREATE SCHEMA ${sql(schema)}`; await sql`SET search_path TO ${sql(schema)}`; - for (const tableName of ["input", "support", "written", "deleted"]) { + for (const tableName of [ + "input", + "support", + "written", + "deleted", + "targets", + ]) { const store = new PostgresKvStore(sql, { tableName }); try { await store.setMany([ @@ -1334,3 +1340,142 @@ dbTest( } }, ); + +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(); + } + }, +); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 9ffcaad71..d2e589fef 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -572,7 +572,11 @@ export class PostgresKvStore implements AtomicKvStore { created = EXCLUDED.created, ttl = EXCLUDED.ttl RETURNING 1 ) - SELECT supported, (SELECT count(*) FROM written) AS size FROM support + -- 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); } @@ -581,15 +585,25 @@ export class PostgresKvStore implements AtomicKvStore { sql: Sql | TransactionSql, payload: string, ): Promise { + const table = sql.unsafe(await this.#resolvedTable(sql)); const [result] = await sql` WITH support AS (${this.#support(sql)}), input AS (${this.#input(sql, payload)}), - deleted AS ( - DELETE FROM ${sql(this.#tableName)} AS existing + 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 ) - SELECT supported, (SELECT count(*) FROM deleted) AS size FROM support + -- 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); } From 940da1e0faff240c1b5f756609dbb07782204ed7 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:59:33 +0900 Subject: [PATCH 08/12] Validate PostgreSQL backing relations before writes Resolve escaped table identifiers as relations instead of composite types so built-in type names do not hide valid backing tables. Check table support before executing mutation statements so rejecting an unsupported scope cannot commit statement-trigger side effects. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4559740662 https://github.com/fedify-dev/fedify/pull/1429#discussion_r4559756514 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- packages/postgres/src/kv.atomic.test.ts | 74 +++++++++++++++++++++++++ packages/postgres/src/kv.ts | 32 ++++++----- 2 files changed, 93 insertions(+), 13 deletions(-) diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index 64865d744..c5ff2624c 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -1479,3 +1479,77 @@ dbTest( } }, ); + +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(); + } + }, + ); +} diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index d2e589fef..9c4115ee5 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -490,6 +490,18 @@ export class PostgresKvStore implements AtomicKvStore { `; } + #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. @@ -497,8 +509,7 @@ export class PostgresKvStore implements AtomicKvStore { 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 = (SELECT typrelid FROM pg_catalog.pg_type - WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) + 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."); @@ -522,8 +533,7 @@ export class PostgresKvStore implements AtomicKvStore { WHERE i.inhparent = c.oid AND NOT child.relispartition ) AS supported FROM pg_catalog.pg_class AS c - WHERE c.oid = (SELECT typrelid FROM pg_catalog.pg_type - WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) + WHERE c.oid = pg_catalog.to_regclass(${this.#escapedTableName(sql)}) `; } @@ -555,6 +565,8 @@ export class PostgresKvStore implements AtomicKvStore { payload: string, cutoff?: string, ): Promise { + const [support] = await this.#support(sql); + this.#assertSupport(support?.supported); const origin = cutoff === undefined ? sql`statement_timestamp()` @@ -585,6 +597,8 @@ export class PostgresKvStore implements AtomicKvStore { 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)}), @@ -729,15 +743,7 @@ export class PostgresKvStore implements AtomicKvStore { 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. - // Scoped clients do not expose options. The driver's identifier helper - // contains the exact escaped SQL name, including its column.to transform. - const identifier = this.#sql(this.#tableName); - if (!("value" in identifier) || typeof identifier.value !== "string") { - throw new TypeError( - "The PostgreSQL client must expose its escaped identifier.", - ); - } - const table = identifier.value; + const table = this.#escapedTableName(this.#sql); const [existing] = await this.#sql` SELECT to_regclass(${table}) IS NOT NULL AS present `; From bfa1ac2a7f1b45b25f8b9d8cd0356141030a484e Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 14:05:27 +0900 Subject: [PATCH 09/12] Keep PostgreSQL codec fixtures live during tests Use a 60-second TTL plus one microsecond, exceeding the test timeout, and assert the matching stored interval. Preserve the precision check without making codec assertions race a 100-millisecond expiry. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4559756510 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- packages/postgres/src/kv.atomic.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index c5ff2624c..6ca5a1bd1 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -711,7 +711,7 @@ for (const identity of [false, true]) { expectedValue: null, newValue: { a: 1 }, options: { - ttl: Temporal.Duration.from({ microseconds: 100_001 }), + ttl: Temporal.Duration.from({ seconds: 60, microseconds: 1 }), }, }, { @@ -724,7 +724,7 @@ for (const identity of [false, 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 '0.100001 seconds') AS precision FROM ${sql(f.tableName)}`; + 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); From f1696b5366cbb9fa7a7fbe86550d07edba71eebe Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 20:11:27 +0900 Subject: [PATCH 10/12] Propagate PostgreSQL test holder setup failures Reject each readiness barrier when its holder fails before acquiring its lock, allowing the existing finally block to release clients. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4560585102 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- packages/postgres/src/kv.atomic.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index 6ca5a1bd1..603142cb4 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -302,6 +302,7 @@ for (const operation of ["deleteMany", "casMany"] as const) { ready.resolve(); await release.promise; }); + void holder.catch(ready.reject); await ready.promise; pending = operation === "deleteMany" @@ -355,6 +356,7 @@ dbTest( ready.resolve(); await release.promise; }); + void holder.catch(ready.reject); await ready.promise; pending = second.casMany([ { @@ -808,6 +810,7 @@ dbTest( ready.resolve(); await release.promise; }); + void holder.catch(ready.reject); await ready.promise; pending = parent.casMany([ { key, expectedValue: 1, newValue: 2 }, @@ -862,6 +865,7 @@ dbTest( ready.resolve(); await release.promise; }); + void holder.catch(ready.reject); await ready.promise; await assert.rejects( second.casMany([{ key, expectedValue: 1, newValue: 2 }]), From a1190b73050c5f3a5ab92daa950eca0a3686ef60 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 20:20:01 +0900 Subject: [PATCH 11/12] Roll back skipped single-key PostgreSQL CAS writes Compare and mutate under one table lock and transaction, sharing a server-time cutoff. Reject a matching mutation that affects no row and roll back trigger side effects instead of reporting a mismatch. Preserve reserved clients with an owned transaction or nested manual savepoint, retain caller isolation, and keep pure absence checks read-only. Document the table-wide locking and transaction lifetime. https://github.com/fedify-dev/fedify/pull/1429#pullrequestreview-5887355870 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-sonnet-5-5 --- CHANGES.md | 6 +- changes.d/postgres/atomic-kv-store.md | 6 +- claude-plugin/skills/fedify/SKILL.md | 6 + docs/manual/kv.md | 15 ++- packages/postgres/README.md | 15 ++- packages/postgres/src/kv.atomic.test.ts | 169 +++++++++++++++++++++++- packages/postgres/src/kv.expiry.test.ts | 4 +- packages/postgres/src/kv.ts | 168 ++++++++++++++++------- 8 files changed, 333 insertions(+), 56 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index da60f95e6..4c15c3f8f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -432,9 +432,11 @@ To be released. - 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 connections reject these + `READ COMMITTED` transaction clients; reserved clients reject these scopes. Nonempty batches reject active row-level security to prevent - policies from filtering individual mutations. `PostgresKvStore.cas()` now + 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/changes.d/postgres/atomic-kv-store.md b/changes.d/postgres/atomic-kv-store.md index b469f3f59..f57686cca 100644 --- a/changes.d/postgres/atomic-kv-store.md +++ b/changes.d/postgres/atomic-kv-store.md @@ -7,9 +7,11 @@ links: - 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 connections reject these + `READ COMMITTED` transaction clients; reserved clients reject these scopes. Nonempty batches reject active row-level security to prevent - policies from filtering individual mutations. `PostgresKvStore.cas()` now + 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 e3dd826ba..897716916 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -519,6 +519,12 @@ 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 diff --git a/docs/manual/kv.md b/docs/manual/kv.md index a8c850ee2..0ad4d1f6d 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -344,6 +344,17 @@ 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. @@ -373,8 +384,8 @@ 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 start -best-effort cleanup after commit; logical expiry never depends on cleanup. +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 diff --git a/packages/postgres/README.md b/packages/postgres/README.md index 2bd92a1fe..1cfdf7fb0 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -110,6 +110,17 @@ 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 @@ -125,8 +136,8 @@ 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 start best-effort expiry cleanup after -commit; caller-owned batches do not sweep expired rows. Logical expiry is -independent of cleanup. +commit; caller-owned batches and `cas()` 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 diff --git a/packages/postgres/src/kv.atomic.test.ts b/packages/postgres/src/kv.atomic.test.ts index 603142cb4..18ec17555 100644 --- a/packages/postgres/src/kv.atomic.test.ts +++ b/packages/postgres/src/kv.atomic.test.ts @@ -1000,7 +1000,7 @@ dbTest( ); dbTest( - "PostgresKvStore single CAS liveness and TTL origin follow its final statement", + "PostgresKvStore single CAS liveness and TTL origin follow lock acquisition", async () => { const f = fixture(); const validated = Promise.withResolvers(); @@ -1557,3 +1557,170 @@ for (const operation of ["setMany", "deleteMany"] as const) { }, ); } + +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.ts b/packages/postgres/src/kv.ts index 9c4115ee5..bcd6951df 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -266,6 +266,9 @@ export class PostgresKvStore implements AtomicKvStore { /** * {@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( @@ -285,59 +288,86 @@ export class PostgresKvStore implements AtomicKvStore { const payload = JSON.stringify([ { key, expected: expectedValue, value: newValue, ttl }, ]); - await this.initialize(); - // Range validation is not an expiry decision. The final CAS statement - // alone supplies its liveness cutoff and replacement's TTL origin. + 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 NOT coalesce(existing.ttl IS NULL - OR existing.created + existing.ttl > statement_timestamp(), false) - 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; - `; - } 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; + 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; + `; + } - await this.#expire(); - return result.length > 0; + 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; } /** @@ -490,6 +520,54 @@ export class PostgresKvStore implements AtomicKvStore { `; } + 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. From 18fcb97747acb4bfc33ae2e7679793d3ef3730e1 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 20:53:51 +0900 Subject: [PATCH 12/12] Clarify PostgreSQL CAS cleanup client scopes Distinguish pooled batch mutations and mutating CAS forms from caller-owned calls. Preserve the read-only absence-check exception and the independence of logical expiry from physical cleanup. https://github.com/fedify-dev/fedify/pull/1429#discussion_r4562769962 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- packages/postgres/README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/postgres/README.md b/packages/postgres/README.md index 1cfdf7fb0..c5fcbeee3 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -135,9 +135,10 @@ 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 start best-effort expiry cleanup after -commit; caller-owned batches and `cas()` do not sweep expired rows. Logical -expiry is independent of cleanup. +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