From 7268a9be4d9d2b39d46dd478ed4d1990d0f5733e Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 27 Oct 2026 14:56:04 +0900 Subject: [PATCH 1/4] Index PostgreSQL KV expiry cleanup Store UTC-calendar expiry keys in a partial expression index while retaining the original session-calendar TTL checks. Narrow bounded sweep candidates without changing reads, CAS, or housekeeping ownership. Provision indexes only on empty tables through a bounded private transaction. Document concurrent migration for populated and externally managed tables, rolling upgrades, calendar skew, and timestamp limits. Cover old writers, TTL transitions, expired-key CAS, identifier transforms, concurrency, RLS, non-owner access, and partitioned or inherited layouts. Verify actual custom and generic sweep plans over 80,000 live rows. Fixes https://github.com/fedify-dev/fedify/issues/1309 https://github.com/fedify-dev/fedify/issues/1308 https://github.com/fedify-dev/fedify/pull/1314 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-opus-5-5 --- CHANGES.md | 16 + changes.d/postgres/indexed-expiry.md | 16 + claude-plugin/skills/fedify/SKILL.md | 8 + docs/manual/kv.md | 91 ++++ packages/postgres/README.md | 7 + packages/postgres/src/kv.index.test.ts | 648 +++++++++++++++++++++++++ packages/postgres/src/kv.ts | 128 ++++- 7 files changed, 910 insertions(+), 4 deletions(-) create mode 100644 changes.d/postgres/indexed-expiry.md create mode 100644 packages/postgres/src/kv.index.test.ts diff --git a/CHANGES.md b/CHANGES.md index 2b001cae7..19989bc08 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -205,6 +205,22 @@ To be released. [#1206]: https://github.com/fedify-dev/fedify/issues/1206 [#1245]: https://github.com/fedify-dev/fedify/pull/1245 +### @fedify/postgres + + - Added an expression index for `PostgresKvStore` expiry cleanup, avoiding a + full-table scan when few or no entries have expired. Empty tables receive + the index automatically with pooled clients. For existing populated tables + or `initialized: true`, follow the [index migration guide]. Old and new + versions can share the indexed table. Logical TTL behavior is unchanged; + calendar intervals can delay physical deletion in non-UTC sessions. Once + the index exists, writes whose expiration overflows PostgreSQL's timestamp + range fail atomically, including writes from older versions. + [[#1309], [#1334]] + +[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 + ### @fedify/testing - Mock inbox contexts made by `createInboxContext()` now have the diff --git a/changes.d/postgres/indexed-expiry.md b/changes.d/postgres/indexed-expiry.md new file mode 100644 index 000000000..97dafc1a0 --- /dev/null +++ b/changes.d/postgres/indexed-expiry.md @@ -0,0 +1,16 @@ +--- +links: + '#1309': https://github.com/fedify-dev/fedify/issues/1309 + '#1334': https://github.com/fedify-dev/fedify/pull/1334 +--- + - Added an expression index for `PostgresKvStore` expiry cleanup, avoiding a + full-table scan when few or no entries have expired. Empty tables receive + the index automatically with pooled clients. For existing populated tables + or `initialized: true`, follow the [index migration guide]. Old and new + versions can share the indexed table. Logical TTL behavior is unchanged; + calendar intervals can delay physical deletion in non-UTC sessions. Once + the index exists, writes whose expiration overflows PostgreSQL's timestamp + range fail atomically, including writes from older versions. + [[#1309], [#1334]] + +[index migration guide]: https://fedify.dev/manual/kv#postgresql-expiry-index diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index a6f9d121e..b9eb20aa7 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -311,6 +311,12 @@ in development or tests. | *@fedify/denokv* | yes | yes | | *@fedify/cfworkers* | yes | yes | +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`. +The migration can precede an application upgrade; old writers maintain the +index too. Keep the original TTL columns and calendar semantics. + > [!WARNING] > `PostgresMessageQueue` and similar implementations require connection > pooling sized for parallel consumers; a single shared connection will @@ -324,6 +330,8 @@ in development or tests. See and . +[expiry expression index]: https://fedify.dev/manual/kv#postgresql-expiry-index + Observability ------------- diff --git a/docs/manual/kv.md b/docs/manual/kv.md index fb3b58cb9..9e522160a 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -265,7 +265,98 @@ const federation = createFederation({ }); ~~~~ +#### PostgreSQL expiry index + +*This optimization is available since Fedify 2.5.0.* + +Expiry cleanup can use a B-tree expression index that stores a UTC-calendar +expiration timestamp for each entry with a TTL. The index build populates +keys for existing rows; no new table column, trigger, or separate backfill is +needed. PostgreSQL maintains the index on every insert and update, including +those from older Fedify versions. + +With a pooled client, automatic initialization attempts to add the index to an +empty table. It skips populated tables, tables with row-level security, and +ordinary inheritance parents. It also skips index provisioning with +`initialized: true`, a transaction client, or a reserved connection. Missing +ownership, lock contention, or an index-build failure leaves the store usable +without the optimization. Ordinary table initialization still follows the +existing `initialized` and `unlogged` options. + +For an existing table, or when you manage DDL yourself, run this migration as +the table owner, outside a transaction block. Replace `public.fedify_kv_v2` +with your actual schema and table, and choose an unused index name in that +schema. Quote mixed-case or special identifiers, for example +`"Cache"."Federation keys"`. Use the literal `'UTC'` in both function calls: + +~~~~ sql +CREATE INDEX CONCURRENTLY fedify_kv_expiry +ON public.fedify_kv_v2 + ((timezone('UTC', timezone('UTC', created) + ttl))) +WHERE ttl IS NOT NULL; + +ANALYZE public.fedify_kv_v2; +~~~~ + +The concurrent build permits ordinary writes while it runs, but needs time, +disk space, and database resources. `ANALYZE` collects expression statistics +so the planner can choose the index when few or no entries have expired. +If the role lacks privileges, PostgreSQL skips `ANALYZE` with only a warning, +so run it as a role allowed to analyze the table. Keep normal +autovacuum and analyze maintenance enabled as the table grows. + +The index can be created before or after an application upgrade. Old and new +versions can share the table, and downgrading does not require removing the +index. Without the index, the new cleanup query remains correct but may scan +the full table. An index also adds write and storage costs, including fewer +opportunities for HOT updates. Once the index exists, a write whose expiration +timestamp falls outside PostgreSQL's supported timestamp range fails +atomically, including writes from older application versions. An existing +overflowing row can also prevent index creation; correct such data before +retrying the migration. + +Logical TTL checks in `get()`, `list()`, and `cas()` still evaluate +`created + ttl` in the connection's `TimeZone`. Cleanup uses the UTC index to +find candidates and rechecks that original predicate before deleting them. +This preserves PostgreSQL's calendar-month, calendar-day, and DST behavior. +For calendar intervals in a non-UTC session, a different UTC-calendar result +can delay physical deletion or make cleanup visit a candidate before it +logically expires; the recheck prevents early deletion. Reads and CAS still +observe the original logical expiry. Cleanup's 1,000-row limit bounds the rows +selected and deleted, not the index entries examined. + +For traditional inheritance, build the index and run `ANALYZE` on the parent +and on each child, including children added later. An index on the parent +does not cover inherited children. + +For a populated partitioned table, PostgreSQL does not support +`CREATE INDEX CONCURRENTLY` on the parent. Create the parent index with +`CREATE INDEX ... ON ONLY`, build matching indexes concurrently on each leaf, +and attach them with `ALTER INDEX ... ATTACH PARTITION`. Repeat the parent +and attachment steps at each intermediate partitioning level. After all +attachments, verify that the root index is valid, then run `ANALYZE` on the +root and each leaf partition. Autovacuum does not analyze partitioned +parents. New partitions then receive matching indexes automatically. See +PostgreSQL's [partitioned index procedure] for the full workflow. + +A failed concurrent build can leave an invalid index. Inspect its state: + +~~~~ sql +SELECT indexrelid::regclass AS index_name, indisvalid +FROM pg_index +WHERE indrelid = 'public.fedify_kv_v2'::regclass + AND NOT indisprimary; +~~~~ + +After confirming that an invalid index is the one created by your migration, +drop it with `DROP INDEX CONCURRENTLY public.fedify_kv_expiry` outside a +transaction and retry the build and `ANALYZE`. Automatic initialization never +replaces conflicting or invalid indexes. It accepts an existing valid index if +it uses the expression above or 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 +[partitioned index procedure]: https://www.postgresql.org/docs/current/ddl-partitioning.html#DDL-PARTITIONING-DECLARATIVE-MAINTENANCE ### [`PgliteKvStore`] diff --git a/packages/postgres/README.md b/packages/postgres/README.md index 5c3e0370e..afd729479 100644 --- a/packages/postgres/README.md +++ b/packages/postgres/README.md @@ -34,6 +34,12 @@ next initialization converts that table to logged storage. PostgreSQL rewrites the table and takes an `ACCESS EXCLUSIVE` lock during this one-time migration, so schedule the upgrade accordingly for a large or busy table. +Expiry cleanup can use an expression index without changing the table schema. +New empty tables receive it automatically with a pooled client. Existing +populated tables and deployments using `initialized: true` need an explicit +index migration. See the [expiry-index guide] for concurrent index creation, +rolling upgrades, and calendar-time semantics. + [JSR badge]: https://jsr.io/badges/@fedify/postgres [JSR]: https://jsr.io/@fedify/postgres [npm badge]: https://img.shields.io/npm/v/@fedify/postgres?logo=npm @@ -43,6 +49,7 @@ so schedule the upgrade accordingly for a large or busy table. [`MessageQueue`]: https://jsr.io/@fedify/fedify/doc/federation/~/MessageQueue [`PostgresKvStore`]: https://jsr.io/@fedify/postgres/doc/~/PostgresKvStore [`PostgresMessageQueue`]: https://jsr.io/@fedify/postgres/doc/~/PostgresMessageQueue +[expiry-index guide]: https://fedify.dev/manual/kv#postgresql-expiry-index Installation diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts new file mode 100644 index 000000000..34656c0e1 --- /dev/null +++ b/packages/postgres/src/kv.index.test.ts @@ -0,0 +1,648 @@ +import { PostgresKvStore } from "@fedify/postgres/kv"; +import * as temporal from "@js-temporal/polyfill"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { test } from "node:test"; +import postgres from "postgres"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const dbUrl = process.env.POSTGRES_URL; +const skip = dbUrl == null; + +function database(options: postgres.Options> = {}) { + const statements: string[] = []; + let sweep: { query: string; parameters: postgres.ParameterOrJSON[] }; + let pending: Promise | undefined; + const sql = postgres(dbUrl!, { + ...options, + debug(connection, query, parameters, types) { + if (typeof options.debug === "function") { + options.debug(connection, query, parameters, types); + } + statements.push(query); + if ( + query.includes("hashtext('fedify.postgres.kv.expire')") && + !query.startsWith("EXPLAIN") + ) { + sweep = { query, parameters }; + } + }, + }); + const observed = new Proxy(sql, { + apply(target, self, args) { + const result = Reflect.apply(target, self, args); + if ( + Array.isArray(args[0]) && + args[0].join("").includes("hashtext('fedify.postgres.kv.expire')") + ) { + pending = Promise.resolve(result); + return pending; + } + return result; + }, + }); + const tableName = `fedify_kv_index_${ + crypto.randomUUID().replaceAll("-", "") + }`; + return { + sql, + observed, + tableName, + statements, + async complete() { + assert(pending, "a sweep must have started"); + await pending; + }, + async prepared() { + return await sql`SELECT generic_plans, custom_plans FROM pg_prepared_statements + WHERE statement = ${sweep.query}`; + }, + async explain() { + assert(sweep!, "the actual sweep query must have been captured"); + return await sql.unsafe( + `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${sweep.query}`, + sweep.parameters, + { prepare: true }, + ); + }, + }; +} + +async function legacyTable(sql: postgres.Sql, name: string) { + await sql`CREATE TABLE ${sql(name)} ( + key text[] PRIMARY KEY, value jsonb NOT NULL, + created timestamptz DEFAULT CURRENT_TIMESTAMP, ttl interval + )`; +} + +async function indexes(sql: postgres.Sql, name: string) { + return await sql` + SELECT i.indisvalid, pg_get_indexdef(i.indexrelid) AS definition + FROM pg_index i + WHERE i.indrelid = (SELECT typrelid FROM pg_type + WHERE oid = pg_typeof(NULL::${sql(name)})::oid) + AND NOT i.indisprimary + `; +} + +async function manualIndex( + sql: postgres.Sql, + name: string, + atTimeZone = false, +) { + const index = `manual_expiry_${crypto.randomUUID().replaceAll("-", "")}`; + // The exact concurrent migration from the manual; no transaction wrapper. + if (atTimeZone) { + await sql`CREATE INDEX CONCURRENTLY ${sql(index)} ON ${sql(name)} + ((((created AT TIME ZONE 'UTC') + ttl) AT TIME ZONE 'UTC')) + WHERE ttl IS NOT NULL`; + } else { + await sql`CREATE INDEX CONCURRENTLY ${sql(index)} ON ${sql(name)} + ((timezone('UTC', timezone('UTC', created) + ttl))) + WHERE ttl IS NOT NULL`; + } + await sql`ANALYZE ${sql(name)}`; +} + +test( + "PostgresKvStore provisions a fresh expiry index and accepts both spellings", + { skip }, + async () => { + if (!dbUrl) return; + const { sql, tableName, statements } = database(); + const store = new PostgresKvStore(sql, { tableName }); + try { + await store.initialize(); + const initial = await indexes(sql, tableName); + assert.equal(initial.length, 1); + assert.equal(initial[0].indisvalid, true); + assert.match(initial[0].definition, /timezone\('UTC'/); + const [{ name }] = await sql` + SELECT c.relname AS name FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE i.indrelid = ${tableName}::regclass AND NOT i.indisprimary + `; + await sql`DROP INDEX ${sql(name)}`; + await manualIndex(sql, tableName, true); + statements.length = 0; + await new PostgresKvStore(sql, { tableName: `public.${tableName}` }) + .initialize(); + assert.equal((await indexes(sql, tableName)).length, 1); + assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + } finally { + await store.drop(); + await sql.end(); + } + }, +); + +test( + "PostgresKvStore leaves populated legacy tables unlocked until manual migration", + { skip }, + async () => { + if (!dbUrl) return; + const db = database(); + const { sql, tableName, statements } = db; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await legacyTable(sql, tableName); + await sql`INSERT INTO ${sql(tableName)} VALUES + (ARRAY['expired'], '1', '2000-01-01', INTERVAL '1 day'), + (ARRAY['permanent'], '2', CURRENT_TIMESTAMP, NULL)`; + statements.length = 0; + await store.initialize(); + assert.equal((await indexes(sql, tableName)).length, 0); + assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + // Index creation populates existing keys without a column backfill. + await manualIndex(sql, tableName); + const migrated = new PostgresKvStore(db.observed, { + tableName, + initialized: true, + }); + assert.equal(await migrated.get(["expired"]), undefined); + assert.equal(await migrated.get(["permanent"]), 2); + await migrated.delete(["missing"]); + await db.complete(); + const [{ count }] = await sql`SELECT count(*)::int AS count FROM ${ + sql(tableName) + }`; + assert.equal(count, 1); + assert.equal((await indexes(sql, tableName)).length, 1); + } finally { + await store.drop(); + await sql.end(); + } + }, +); + +test("PostgresKvStore indexed keys track old writers and all TTL transitions", { + skip, +}, async () => { + if (!dbUrl) return; + const db = database(); + const { sql, tableName } = db; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await store.initialize(); + // Start and drain housekeeping before inserting expired rows. Subsequent + // operations use #1314's cooldown, so tests exercise logical expiry too. + await store.delete(["missing"]); + await db.complete(); + await sql`INSERT INTO ${sql(tableName)} (key, value, created, ttl) + VALUES (ARRAY['key'], '"old"', '2000-01-01', INTERVAL '1 day')`; + assert.equal(await store.cas(["key"], "old", "wrong"), false); + assert.equal( + await store.cas(["key"], undefined, "new", { + ttl: Temporal.Duration.from({ hours: 1 }), + }), + true, + ); + assert.equal(await store.get(["key"]), "new"); + await store.set(["key"], "permanent"); + assert.equal( + await store.cas(["key"], "permanent", "zero", { + ttl: Temporal.Duration.from({ seconds: 0 }), + }), + true, + ); + assert.equal(await store.get(["key"]), undefined); + assert.equal(await store.cas(["key"], undefined, "again"), true); + // Old applications still write only these four columns. + await sql`UPDATE ${sql(tableName)} SET created = '2000-01-01', + ttl = INTERVAL '1 day' WHERE key = ARRAY['key']`; + assert.equal(await store.get(["key"]), undefined); + assert.deepEqual(await Array.fromAsync(store.list()), []); + const cleaner = new PostgresKvStore(db.observed, { + tableName, + initialized: true, + }); + await cleaner.delete(["missing"]); + await db.complete(); + const [{ count }] = await sql`SELECT count(*)::int AS count FROM ${ + sql(tableName) + }`; + assert.equal(count, 0); + await sql`INSERT INTO ${sql(tableName)} VALUES + (ARRAY['key'], '"negative"', statement_timestamp(), INTERVAL '-1 second')`; + assert.equal(await store.get(["key"]), undefined); + await store.set(["key"], "live", { + ttl: Temporal.Duration.from({ days: 30 }), + }); + assert.equal(await store.get(["key"]), "live"); + await sql`INSERT INTO ${ + sql(tableName) + } 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); + } finally { + await store.drop(); + await sql.end(); + } +}); + +for (const planMode of ["force_custom_plan", "force_generic_plan"]) { + test( + `PostgresKvStore actual sparse expiry plan uses the index (${planMode})`, + { skip }, + async () => { + if (!dbUrl) return; + const db = database({ + max: 1, + connection: { plan_cache_mode: planMode }, + }); + const { sql, tableName } = db; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await store.initialize(); + await sql`INSERT INTO ${sql(tableName)} (key, value, created, ttl) + SELECT ARRAY[n::text], '1', statement_timestamp(), + CASE WHEN n % 2 = 0 THEN INTERVAL '1 day' END + FROM generate_series(1, 80000) n`; + await sql`ANALYZE ${sql(tableName)}`; + await store.delete(["missing"]); + await db.complete(); + const [prepared] = await db.prepared(); + assert(prepared, "the production sweep must be prepared"); + assert( + Number( + prepared[ + planMode === "force_generic_plan" + ? "generic_plans" + : "custom_plans" + ], + ) > 0, + ); + for (const expired of [0, 5]) { + if (expired) { + await sql`INSERT INTO ${sql(tableName)} (key, value, created, ttl) + SELECT ARRAY['expired', n::text], '1', '2000-01-01', INTERVAL '1 day' + FROM generate_series(1, ${expired}) n`; + } + const result = await db.explain(); + const [{ Plan: plan }] = result[0]["QUERY PLAN"]; + const nodes: Record[] = []; + const visit = (node: Record) => { + nodes.push(node); + for ( + const child of (node.Plans ?? []) as Record[] + ) { + visit(child); + } + }; + visit(plan); + assert( + nodes.some((n) => + /Index|Bitmap/.test(String(n["Node Type"])) && + String(n["Index Cond"]).includes("timezone('UTC'") + ), + JSON.stringify(plan), + ); + assert( + !nodes.some((n) => n["Node Type"] === "Seq Scan"), + JSON.stringify(plan), + ); + assert( + Number(plan["Shared Hit Blocks"]) < 1000, + JSON.stringify(plan), + ); + const deleted = nodes.find((n) => n["CTE Name"] === "deleted"); + assert.equal(deleted?.["Actual Rows"], expired); + const [{ count }] = await sql`SELECT count(*)::int AS count FROM ${ + sql(tableName) + }`; + assert.equal(count, 80000); + } + } finally { + await store.drop(); + await sql.end(); + } + }, + ); +} + +for ( + const base of [ + { created: "2026-03-07 12:00-05", ttl: "1 day", live: false }, + { created: "2026-10-31 12:00-04", ttl: "1 day", live: true }, + { created: "2026-02-28 23:30-05", ttl: "1 month", live: false }, + ] +) { + test( + `PostgresKvStore retains session-calendar TTL semantics (${base.created})`, + { skip }, + async () => { + if (!dbUrl) return; + const db = database({ connection: { TimeZone: "America/New_York" } }); + const { sql, tableName } = db; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await store.initialize(); + // Shift only elapsed seconds, retaining the calendar day/month component. + // The logical expiry is +/-30 minutes from now, while the UTC index key + // lies on the other side of now (1h DST skew, or >3d month-end skew). + await sql`INSERT INTO ${sql(tableName)} (key, value, created, ttl) + SELECT ARRAY['calendar'], '"calendar"', ${base.created}::timestamptz, + ${base.ttl}::interval + make_interval(secs => extract(epoch FROM ( + statement_timestamp() - (${base.created}::timestamptz + ${base.ttl}::interval) + ))::double precision) + ${ + base.live ? "30 minutes" : "-30 minutes" + }::interval`; + const [{ live, candidate }] = await sql` + SELECT created + ttl > statement_timestamp() AS live, + timezone('UTC', timezone('UTC', created) + ttl) < statement_timestamp() AS candidate + FROM ${sql(tableName)} + `; + assert.equal(live, base.live); + assert.equal(candidate, base.live); + await store.delete(["missing"]); + await db.complete(); + const [{ count }] = await sql`SELECT count(*)::int AS count FROM ${ + sql(tableName) + }`; + assert.equal( + count, + 1, + "no premature deletion; later physical deletion is permitted", + ); + assert.equal( + await store.get(["calendar"]), + base.live ? "calendar" : undefined, + ); + assert.equal( + (await Array.fromAsync(store.list())).length, + base.live ? 1 : 0, + ); + assert.equal( + await store.cas(["calendar"], "calendar", "new"), + base.live, + ); + if (!base.live) { + assert.equal(await store.cas(["calendar"], undefined, "new"), true); + } + } finally { + await store.drop(); + await sql.end(); + } + }, + ); +} + +test( + "PostgresKvStore index provisioning resolves non-idempotent transformed identifiers once", + { skip }, + async () => { + if (!dbUrl) return; + const db = database({ + transform: { column: { to: (name) => `x_${name}` } }, + }); + const { sql, statements } = db; + const schema = `schema_${crypto.randomUUID().replaceAll("-", "")}`; + const name = `Mixed"${"a".repeat(48)}`; + const tableName = `${schema}.${name}`; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await sql`CREATE SCHEMA ${sql(schema)}`; + await store.initialize(); + assert.equal((await indexes(sql, tableName)).length, 1); + statements.length = 0; + await new PostgresKvStore(sql, { tableName }).initialize(); + assert.equal((await indexes(sql, tableName)).length, 1); + assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + await store.set(["key"], "value"); + await db.complete(); + assert.equal(await store.get(["key"]), "value"); + } finally { + await store.drop(); + await sql`DROP SCHEMA ${sql(schema)}`; + await sql.end(); + } + }, +); + +test("PostgresKvStore optional index failures preserve usable legacy storage", { + skip, +}, async () => { + if (!dbUrl) return; + const db = database(); + const { sql, tableName } = db; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await legacyTable(sql, tableName); + const [{ name }] = await sql` + SELECT 'idx_fedify_kv_expiry_' || substr(md5(format('%I.%I', + n.nspname, c.relname)), 1, 16) AS name + FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.oid = ${tableName}::regclass + `; + // A conflicting unrelated index must neither be accepted nor replaced. + await sql`CREATE INDEX ${sql(name)} ON ${sql(tableName)} (created)`; + await store.set(["key"], "works"); + await db.complete(); + assert.equal(await store.get(["key"]), "works"); + assert.equal((await indexes(sql, tableName)).length, 1); + assert.match((await indexes(sql, tableName))[0].definition, /\(created\)/); + } finally { + await store.drop(); + await sql.end(); + } +}); + +test("PostgresKvStore provisioning skips RLS", { skip }, async () => { + if (!dbUrl) return; + const { sql, tableName, statements } = database({ + connection: { default_transaction_isolation: "repeatable read" }, + }); + const store = new PostgresKvStore(sql, { tableName }); + try { + await legacyTable(sql, tableName); + await sql`ALTER TABLE ${sql(tableName)} ENABLE ROW LEVEL SECURITY`; + statements.length = 0; + await store.initialize(); + assert.equal((await indexes(sql, tableName)).length, 0); + assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + } finally { + await store.drop(); + await sql.end(); + } +}); + +test( + "PostgresKvStore serializes empty-table index provisioning", + { skip }, + async () => { + if (!dbUrl) return; + const { sql, tableName } = database(); + try { + await legacyTable(sql, tableName); + await Promise.all( + Array.from( + { length: 6 }, + () => new PostgresKvStore(sql, { tableName }).initialize(), + ), + ); + assert.equal((await indexes(sql, tableName)).length, 1); + } finally { + await sql`DROP TABLE IF EXISTS ${sql(tableName)}`; + await sql.end(); + } + }, +); + +test("PostgresKvStore indexed timestamp overflow rejects writes atomically", { + skip, +}, async () => { + if (!dbUrl) return; + const { sql, tableName } = database(); + const store = new PostgresKvStore(sql, { tableName }); + try { + await store.set(["key"], "original"); + const ttl = Temporal.Duration.from({ years: 300000 }); + await assert.rejects( + store.set(["key"], "wrong", { ttl }), + /timestamp out of range/, + ); + await assert.rejects( + store.cas(["key"], "original", "wrong", { ttl }), + /timestamp out of range/, + ); + assert.equal(await store.get(["key"]), "original"); + } finally { + await store.drop(); + await sql.end(); + } +}); + +test("PostgresKvStore rechecks emptiness after a concurrent writer commits", { + skip, +}, async () => { + if (!dbUrl) return; + const locking = Promise.withResolvers(); + const { sql, tableName } = database({ + connection: { default_transaction_isolation: "repeatable read" }, + debug(_connection, query) { + if (query.startsWith("LOCK TABLE")) locking.resolve(); + }, + }); + const store = new PostgresKvStore(sql, { tableName }); + let initializing: Promise | undefined; + let timer: ReturnType | undefined; + try { + await legacyTable(sql, tableName); + await sql.begin(async (tx) => { + // The first emptiness probe cannot see this uncommitted row. SHARE + // must wait for this writer, then the next probe must see its commit. + await tx`INSERT INTO ${tx(tableName)} VALUES + (ARRAY['concurrent'], '1', statement_timestamp(), NULL)`; + initializing = store.initialize(); + await Promise.race([ + locking.promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error("provisioning did not reach its lock")), + 3000, + ); + }), + ]); + clearTimeout(timer); + }); + await initializing; + assert.equal( + (await indexes(sql, tableName)).length, + 0, + "a stale snapshot must not allow index creation on the populated table", + ); + assert.equal(await store.get(["concurrent"]), 1); + } finally { + clearTimeout(timer); + await initializing; + await store.drop(); + await sql.end(); + } +}); + +for (const layout of ["partitioned", "inherited"] as const) { + test( + `PostgresKvStore expiry indexing supports ${layout} tables`, + { skip }, + async () => { + if (!dbUrl) return; + const db = database(); + const { sql, tableName } = db; + const child = `${tableName}_child`; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + if (layout === "partitioned") { + await sql`CREATE TABLE ${sql(tableName)} ( + key text[], value jsonb NOT NULL, created timestamptz, ttl interval + ) PARTITION BY RANGE (created)`; + await sql`CREATE TABLE ${sql(child)} PARTITION OF ${ + sql(tableName) + } DEFAULT`; + } else { + await legacyTable(sql, tableName); + await sql`CREATE TABLE ${sql(child)} () INHERITS (${sql(tableName)})`; + } + await store.initialize(); + assert.equal( + (await indexes(sql, tableName)).length, + layout === "partitioned" ? 1 : 0, + ); + if (layout === "inherited") { + await manualIndex(sql, tableName); + await manualIndex(sql, child); + } + assert.equal((await indexes(sql, child)).length, 1); + // A store naming the child directly must recognize its physical index. + await new PostgresKvStore(sql, { tableName: child }).initialize(); + assert.equal((await indexes(sql, child)).length, 1); + await sql`INSERT INTO ${sql(child)} VALUES + (ARRAY['expired'], '1', '2000-01-01', INTERVAL '1 day')`; + await store.delete(["missing"]); + await db.complete(); + const [{ count }] = await sql`SELECT count(*)::int AS count FROM ${ + sql(tableName) + }`; + assert.equal(count, 0); + } finally { + await sql`DROP TABLE IF EXISTS ${sql(tableName)} CASCADE`; + await sql.end(); + } + }, + ); +} + +test( + "PostgresKvStore keeps a non-owner's existing table usable", + { skip }, + async () => { + if (!dbUrl) return; + const db = database(); + const { sql } = db; + const schema = `kv_schema_${crypto.randomUUID().replaceAll("-", "")}`; + const tableName = `${schema}.${db.tableName}`; + const role = `kv_reader_${crypto.randomUUID().replaceAll("-", "")}`; + let client: postgres.Sql | undefined; + try { + await sql`CREATE SCHEMA ${sql(schema)}`; + await legacyTable(sql, tableName); + await sql`CREATE ROLE ${sql(role)}`; + await sql`GRANT USAGE, CREATE ON SCHEMA ${sql(schema)} TO ${sql(role)}`; + await sql`GRANT SELECT, INSERT, UPDATE, DELETE ON ${sql(tableName)} TO ${ + sql(role) + }`; + client = postgres(dbUrl, { connection: { role } }); + const store = new PostgresKvStore(client, { tableName }); + await store.set(["key"], "value"); + assert.equal(await store.get(["key"]), "value"); + assert.equal((await indexes(sql, tableName)).length, 0); + // Drains background housekeeping without requiring the non-owner to DROP. + await client.end(); + } finally { + await client?.end(); + await sql`DROP TABLE IF EXISTS ${sql(tableName)}`; + await sql`DROP SCHEMA IF EXISTS ${sql(schema)}`; + await sql`DROP ROLE IF EXISTS ${sql(role)}`; + await sql.end(); + } + }, +); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 68bd1067d..56896ac43 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -35,6 +35,8 @@ export interface PostgresKvStoreOptions { * This skips only the table's schema DDL. Driver-specific runtime setup, * such as detecting whether the driver serializes JSON parameters on its * own, still runs before the first key is read or written. + * It also skips automatic expiry-index provisioning. See the PostgreSQL + * key–value store migration guide for the index to add to your own DDL. * @default `false` */ readonly initialized?: boolean; @@ -96,9 +98,7 @@ export class PostgresKvStore implements KvStore { // detached SQL error would still abort it and could hide a rollback. // Reserved clients may run manual transactions without savepoint(). // Cleanup locks remain held until the caller ends the transaction. - const callerOwned = ("savepoint" in this.#sql && - typeof this.#sql.savepoint === "function") || - ("release" in this.#sql && typeof this.#sql.release === "function"); + const callerOwned = this.#callerOwned(); if (this.#expiring) return callerOwned ? this.#expiring : undefined; if (performance.now() < this.#nextExpiry) return; this.#expiring = (async () => { @@ -127,6 +127,10 @@ export class PostgresKvStore implements KvStore { // Resolve the logical table through its row type so partitions and // inheritance children share one lock, respecting identifier transforms. // The bigint lock namespace is separate from the queue's two-int locks. + // The immutable UTC expression is only an index search key. Recheck the + // original session-calendar predicate before LIMIT and again on deletion; + // calendar/DST differences may delay reclamation, never logical expiry. + // Keep UTC literal so generic plans can match the expression index. // CTIDs overlap across child tables, so check tableoid too. Keep the CTID // array to allow TID lookups instead of a second full-table join scan. const [result] = await this.#sql<{ count: number }[]>` @@ -140,7 +144,10 @@ export class PostgresKvStore implements KvStore { SELECT tableoid AS relation, ctid AS tid FROM ${this.#sql(this.#tableName)} WHERE (SELECT acquired FROM sweep_lock) - AND ttl IS NOT NULL AND created + ttl < statement_timestamp() + AND ttl IS NOT NULL + AND timezone('UTC', timezone('UTC', created) + ttl) + < statement_timestamp() + AND created + ttl < statement_timestamp() LIMIT ${expiryBatchSize} FOR UPDATE SKIP LOCKED ), deleted AS ( DELETE FROM ${this.#sql(this.#tableName)} @@ -311,6 +318,9 @@ export class PostgresKvStore implements KvStore { tableName: this.#tableName, }); if (!this.#skipDdl) await this.#initializeTable(); + if (!this.#skipDdl && !this.#callerOwned()) { + await this.#initializeExpiryIndex(); + } this.#driverSerializesJson = await driverSerializesJson(this.#sql); this.#initialized = true; logger.debug("Initialized the key–value store table {tableName}.", { @@ -357,6 +367,116 @@ export class PostgresKvStore implements KvStore { } } + #callerOwned(): boolean { + return ("savepoint" in this.#sql && + typeof this.#sql.savepoint === "function") || + ("release" in this.#sql && typeof this.#sql.release === "function"); + } + + 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` + 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, + format('%I.%I', n.nspname, c.relname) AS name, + 'idx_fedify_kv_expiry_' || substr(md5( + format('%I.%I', n.nspname, c.relname) + ), 1, 16) AS idx + FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.oid = (SELECT typrelid FROM pg_type + WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) + `; + const indexed = async () => { + const [result] = await sql` + SELECT EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_am a ON a.oid = c.relam + JOIN pg_opclass o ON o.oid = i.indclass[0] + WHERE i.indrelid = ${table.oid}::oid + AND i.indisvalid AND i.indisready AND i.indislive + 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 ( + ${"timezone('UTC'::text, (timezone('UTC'::text, created) + ttl))"}, + ${"(((created AT TIME ZONE 'UTC'::text) + ttl) AT TIME ZONE 'UTC'::text)"} + ) + AND (i.indpred IS NULL OR + 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` + 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` + 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); + }); + } catch (error) { + // This private transaction has rolled back. Optional index setup must + // not abort a caller's transaction or disable an otherwise usable store. + logger.warn( + "Failed to provision the expiry index for {tableName}: {error}. See {guide}.", + { + tableName: this.#tableName, + error, + guide: "https://fedify.dev/manual/kv#postgresql-expiry-index", + }, + ); + } + } + /** * Drops the table used by the key–value store. Does nothing if the table * does not exist. From 60b26d0f8a37b1e628fb687967cd8e9583c63251 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 27 Oct 2026 18:21:58 +0900 Subject: [PATCH 2/4] Use raw Markdown for the expiry skill link Keep the expiry-index reference consistent with the Fedify skill's raw Markdown convention so agents can read the source documentation. https://github.com/fedify-dev/fedify/pull/1334#discussion_r4386258891 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- claude-plugin/skills/fedify/SKILL.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b9eb20aa7..955d873f6 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -330,7 +330,7 @@ index too. Keep the original TTL columns and calendar semantics. See and . -[expiry expression index]: https://fedify.dev/manual/kv#postgresql-expiry-index +[expiry expression index]: https://fedify.dev/manual/kv.md#postgresql-expiry-index Observability From 48d9a323effa351b99abbb61d9b09ddab8d41490 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 27 Oct 2026 18:25:22 +0900 Subject: [PATCH 3/4] Clarify custom expiry index recovery Use the schema and index name chosen for the migration when removing an invalid index, and explain when each identifier needs quoting. Make the recovery order explicit: rebuild the index, then ANALYZE. https://github.com/fedify-dev/fedify/pull/1334#discussion_r4386258897 Changelog: none Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-opus-5-5 --- docs/manual/kv.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 9e522160a..759b1f3e6 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -349,8 +349,10 @@ WHERE indrelid = 'public.fedify_kv_v2'::regclass ~~~~ After confirming that an invalid index is the one created by your migration, -drop it with `DROP INDEX CONCURRENTLY public.fedify_kv_expiry` outside a -transaction and retry the build and `ANALYZE`. Automatic initialization never +run `DROP INDEX CONCURRENTLY .` outside a transaction. +Replace `` and `` with the schema and index name you chose +for the migration, quoting each identifier as needed. Then rerun the index +build, followed by `ANALYZE`. Automatic initialization never replaces conflicting or invalid indexes. It accepts an existing valid index if it uses the expression above or its `AT TIME ZONE` equivalent, and has either no predicate or `WHERE ttl IS NOT NULL`. From c2948a7c03784dfc3a2dcc10a7cf2216da58b4d8 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Wed, 28 Oct 2026 00:51:15 +0900 Subject: [PATCH 4/4] Require a dedicated PostgreSQL expiry index Accept only non-unique single-attribute indexes during expiry index provisioning. A matching first key on a wider index does not establish that cleanup has the intended dedicated index. Preserve operator-owned indexes and constraints. Cover UNIQUE and multi-key indexes, including custom and generic sweep plans over 80,000 rows, and document the stricter matching criteria. https://github.com/fedify-dev/fedify/pull/1334#pullrequestreview-5664921282 Changelog: none Assisted-by: Codex:gpt-6.1-sol --- docs/manual/kv.md | 7 +- packages/postgres/src/kv.index.test.ts | 92 ++++++++++++++++++++++++++ packages/postgres/src/kv.ts | 1 + 3 files changed, 97 insertions(+), 3 deletions(-) diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 759b1f3e6..5b2c165a9 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -353,9 +353,10 @@ run `DROP INDEX CONCURRENTLY .` outside a transaction. Replace `` and `` with the schema and index name you chose for the migration, quoting each identifier as needed. Then rerun the index build, followed by `ANALYZE`. Automatic initialization never -replaces conflicting or invalid indexes. It accepts an existing valid index if -it uses the expression above or its `AT TIME ZONE` equivalent, and has either -no predicate or `WHERE ttl IS NOT NULL`. +replaces conflicting or invalid indexes. It accepts an existing valid, +non-unique, single-attribute B-tree index if it uses the expression above or +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 [partitioned index procedure]: https://www.postgresql.org/docs/current/ddl-partitioning.html#DDL-PARTITIONING-DECLARATIVE-MAINTENANCE diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index 34656c0e1..40b6d1914 100644 --- a/packages/postgres/src/kv.index.test.ts +++ b/packages/postgres/src/kv.index.test.ts @@ -646,3 +646,95 @@ test( } }, ); + +for (const shape of ["unique", "multi-key"] as const) { + test( + `PostgresKvStore provisions a dedicated expiry index beside a ${shape} index`, + { skip }, + async () => { + if (!dbUrl) return; + const db = database({ max: 1 }); + const { sql, tableName } = db; + const existing = `existing_expiry_${ + crypto.randomUUID().replaceAll("-", "") + }`; + const store = new PostgresKvStore(db.observed, { tableName }); + try { + await legacyTable(sql, tableName); + if (shape === "unique") { + await sql`CREATE UNIQUE INDEX ${sql(existing)} ON ${sql(tableName)} + ((timezone('UTC', timezone('UTC', created) + ttl))) + WHERE ttl IS NOT NULL`; + } else { + await sql`CREATE INDEX ${sql(existing)} ON ${sql(tableName)} + ((timezone('UTC', timezone('UTC', created) + ttl)), key) + WHERE ttl IS NOT NULL`; + } + await store.initialize(); + const all = await indexes(sql, tableName); + assert.equal( + all.length, + 2, + "provision the intended index without replacing the existing one", + ); + const [dedicated] = await sql` + SELECT c.relname AS name FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE i.indrelid = ${tableName}::regclass + AND NOT i.indisprimary AND NOT i.indisunique AND i.indnatts = 1 + `; + assert( + dedicated, + "cleanup needs its own non-unique single-attribute index", + ); + await sql`INSERT INTO ${sql(tableName)} (key, value, created, ttl) + SELECT ARRAY[n::text], '1', statement_timestamp() + n * INTERVAL '1 second', + INTERVAL '1 day' FROM generate_series(1, 80000) n`; + await sql`ANALYZE ${sql(tableName)}`; + await store.delete(["missing"]); + await db.complete(); + for (const mode of ["force_custom_plan", "force_generic_plan"]) { + await sql.unsafe(`SET plan_cache_mode = ${mode}`); + const result = await db.explain(); + const [{ Plan: plan }] = result[0]["QUERY PLAN"]; + const nodes: Record[] = []; + const visit = (node: Record) => { + nodes.push(node); + for ( + const child of (node.Plans ?? []) as Record[] + ) visit(child); + }; + visit(plan); + assert( + nodes.some((node) => + String(node["Index Cond"]).includes("timezone('UTC'") + ), + JSON.stringify(plan), + ); + assert( + !nodes.some((node) => node["Node Type"] === "Seq Scan"), + JSON.stringify(plan), + ); + assert( + Number(plan["Shared Hit Blocks"]) < 1000, + JSON.stringify(plan), + ); + } + // Provisioning never drops or weakens an operator-owned constraint. + if (shape === "unique") { + await sql`INSERT INTO ${sql(tableName)} VALUES + (ARRAY['first'], '1', '2000-01-01', INTERVAL '1 day')`; + await assert.rejects( + sql`INSERT INTO ${sql(tableName)} VALUES + (ARRAY['duplicate'], '1', '2000-01-01', INTERVAL '1 day')`, + (error: unknown) => + error instanceof postgres.PostgresError && error.code === "23505", + ); + } + } finally { + await store.drop(); + await sql.end(); + } + }, + ); +} diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 56896ac43..28c7ce295 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -401,6 +401,7 @@ export class PostgresKvStore implements KvStore { JOIN pg_opclass o ON o.oid = i.indclass[0] WHERE i.indrelid = ${table.oid}::oid AND i.indisvalid AND i.indisready AND i.indislive + 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 (