From f02bc6d31211921d62f4da7ea66452a2edfad1dd Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Tue, 30 Jun 2026 16:33:27 -0400 Subject: [PATCH 1/8] feat: Add maintain command (catalog discovery, per-table extend, CDC guard) Discovers every partitioned parent carrying a valid pgslice settings comment and extends each with add_partitions in one run, with per-table failure isolation and a read-only CDC-readiness guard (every leaf must have a replica identity usable for logical replication). Exits non-zero if any table failed or is CDC-unsafe. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/cli.ts | 2 + src/commands/maintain.test.ts | 77 +++++ src/commands/maintain.ts | 97 ++++++ src/maintain.test.ts | 569 ++++++++++++++++++++++++++++++++++ src/pgslice.ts | 139 +++++++++ src/table.ts | 40 +++ src/types.ts | 58 ++++ 7 files changed, 982 insertions(+) create mode 100644 src/commands/maintain.test.ts create mode 100644 src/commands/maintain.ts create mode 100644 src/maintain.test.ts diff --git a/src/cli.ts b/src/cli.ts index d9156d6..9f0e2a5 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -7,6 +7,7 @@ import { DisableRetiredMirroringCommand } from "./commands/disable-retired-mirro import { EnableMirroringCommand } from "./commands/enable-mirroring.js"; import { EnableRetiredMirroringCommand } from "./commands/enable-retired-mirroring.js"; import { FillCommand } from "./commands/fill.js"; +import { MaintainCommand } from "./commands/maintain.js"; import { PrepCommand } from "./commands/prep.js"; import { StatusCommand } from "./commands/status.js"; import { SwapCommand } from "./commands/swap.js"; @@ -30,6 +31,7 @@ export function createCli(): Cli { cli.register(EnableMirroringCommand); cli.register(EnableRetiredMirroringCommand); cli.register(FillCommand); + cli.register(MaintainCommand); cli.register(PrepCommand); cli.register(StatusCommand); cli.register(SwapCommand); diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts new file mode 100644 index 0000000..c3c7d6e --- /dev/null +++ b/src/commands/maintain.test.ts @@ -0,0 +1,77 @@ +import { describe, expect } from "vitest"; +import { sql, type DatabaseTransactionConnection } from "slonik"; + +import { commandTest as test } from "../testing/index.js"; +import { MaintainCommand } from "./maintain.js"; + +describe("MaintainCommand", () => { + test.scoped({ commandClass: ({}, use) => use(MaintainCommand) }); + + async function createPosts(transaction: DatabaseTransactionConnection) { + await transaction.query(sql.unsafe` + CREATE TABLE posts ( + id bigint NOT NULL, + created_at timestamp without time zone NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at) + `); + await transaction.query(sql.unsafe` + COMMENT ON TABLE posts IS 'column:created_at,period:month,cast:date,version:3' + `); + await transaction.query(sql.unsafe` + CREATE TABLE posts_y2026m01 PARTITION OF posts + FOR VALUES FROM ('2026-01-01') TO ('2026-02-01') + `); + } + + test("exits 0 and reports a per-table summary on a healthy fleet", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts(transaction); + + const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + + expect(exitCode).toBe(0); + expect(commandContext.stdout.read()?.toString()).toContain( + "public.posts [native]", + ); + }); + + test("exits 1 when a leaf has no usable replica identity", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts(transaction); + await transaction.query(sql.unsafe` + ALTER TABLE posts_y2026m01 REPLICA IDENTITY NOTHING + `); + + const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + + expect(exitCode).toBe(1); + expect(commandContext.stderr.read()?.toString()).toContain( + "usable replica identity", + ); + }); + + test("reports no managed tables when none carry a settings comment", async ({ + cli, + commandContext, + transaction, + }) => { + await transaction.query(sql.unsafe` + CREATE TABLE plain (id bigint NOT NULL, created_at timestamptz NOT NULL) + PARTITION BY RANGE (created_at) + `); + + const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + + expect(exitCode).toBe(0); + expect(commandContext.stdout.read()?.toString()).toContain( + "No managed partitioned tables found.", + ); + }); +}); diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts new file mode 100644 index 0000000..1fcc5cf --- /dev/null +++ b/src/commands/maintain.ts @@ -0,0 +1,97 @@ +import { Command, Option } from "clipanion"; +import * as t from "typanion"; + +import { BaseCommand } from "./base.js"; +import { Pgslice } from "../pgslice.js"; + +export class MaintainCommand extends BaseCommand { + static override paths = [["maintain"]]; + + static override usage = Command.Usage({ + description: + "Extend every managed partitioned table discovered from the catalog", + details: ` + Discovers every partitioned table that carries a valid pgslice settings + comment and extends each one with add_partitions, so newly partitioned + tables are picked up automatically with no per-table configuration. + + Native tables (the partitioned parent owns the primary key) and classic + pgslice tables (each partition owns its own) are both handled. After + extending each table this checks that every leaf partition has a replica + identity usable for logical replication, and exits non-zero if any does + not. + `, + examples: [ + [ + "Extend all managed tables three periods ahead", + "$0 maintain --future 3", + ], + [ + "Restrict to a single schema", + "$0 maintain --future 3 --schema analytics", + ], + ], + }); + + past = Option.String("--past", "0", { + description: "Number of past partitions to add to each table", + validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), + }); + future = Option.String("--future", "0", { + description: "Number of future partitions to add to each table", + validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), + }); + schema = Option.String("--schema", { + description: "Restrict maintenance to partitioned tables in this schema", + }); + tablespace = Option.String("--tablespace", { + description: "Tablespace to use for new partitions", + }); + inheritGrants = Option.Boolean("--inherit-grants", true, { + description: + "Copy each parent table's privileges onto its new partitions (default: true; disable with --no-inherit-grants)", + }); + + override async perform(pgslice: Pgslice): Promise { + const results = await pgslice.start((connection) => + pgslice.maintain(connection, { + past: this.past, + future: this.future, + schema: this.schema, + tablespace: this.tablespace, + inheritGrants: this.inheritGrants, + }), + ); + + if (results.length === 0) { + this.context.stdout.write("No managed partitioned tables found.\n"); + return; + } + + const problems: string[] = []; + for (const result of results) { + if (result.error) { + this.context.stdout.write( + `${result.table}: FAILED — ${result.error}\n`, + ); + problems.push(`${result.table}: ${result.error}`); + continue; + } + + this.context.stdout.write( + `${result.table} [${result.model}]: +${result.partitionsCreated.length} partition(s), ${result.partitionCount} total\n`, + ); + if (!result.replicaIdentityReady) { + problems.push( + `${result.table}: partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`, + ); + } + } + + if (problems.length > 0) { + throw new Error( + `maintain encountered problems:\n ${problems.join("\n ")}`, + ); + } + } +} diff --git a/src/maintain.test.ts b/src/maintain.test.ts new file mode 100644 index 0000000..9a1460b --- /dev/null +++ b/src/maintain.test.ts @@ -0,0 +1,569 @@ +import { afterEach, beforeEach, describe, expect, vi } from "vitest"; +import { sql } from "slonik"; +import { z } from "zod"; + +import { pgsliceTest as test } from "./testing/index.js"; +import { + addChild, + addDefault, + addMinvalueChild, + childNames, + createCdcRole, + DATE, + nativeParent, + pgsliceParent, + TS, + TSTZ, +} from "./testing/shapes.js"; + +/** + * End-to-end validation of `maintain` against the real production table shapes + * on PostgreSQL 13.20: catalog discovery, the native (parent-owned PK) vs + * classic-pgslice (per-partition PK) distinction, grant inheritance, the + * replica-identity behavior verified on PG 13 (the partitioned replica-identity + * index propagates to new partitions automatically; the per-leaf identity falls + * back to the composite key), idempotency, and the CDC-readiness guard. + */ +describe("Pgslice.maintain (fleet)", () => { + beforeEach(() => { + vi.useFakeTimers(); + // 2026-01-15 is a Thursday; its ISO week is 2026-W03 (Monday 2026-01-12). + vi.setSystemTime(new Date(Date.UTC(2026, 0, 15))); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + describe("the full production fleet on one database", () => { + test.beforeEach(async ({ transaction }) => { + // 7 native tables. + for (const posts of ["posts", "comments", "reactions"]) { + await nativeParent( + transaction, + posts, + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + posts, + `${posts}_y2026m01`, + "2026-01-01", + "2026-02-01", + ); + } + + await nativeParent( + transaction, + "daily_reports", + "created_date", + DATE, + "month", + "date", + ); + await addChild( + transaction, + "daily_reports", + "daily_reports_202601", + "2026-01-01", + "2026-02-01", + ); + + await nativeParent( + transaction, + "analytics.visits", + "created_at", + TSTZ, + "week", + "timestamptz", + ); + await addChild( + transaction, + "analytics.visits", + "visits_2026w03", + "2026-01-12 00:00:00+00", + "2026-01-19 00:00:00+00", + ); + await addDefault(transaction, "analytics.visits", "visits_default"); + + await nativeParent( + transaction, + "audit_log", + "occurred_at", + TS, + "week", + "date", + ); + await addChild( + transaction, + "audit_log", + "audit_log_y2026w02", + "2026-01-05", + "2026-01-12", + ); + + await nativeParent( + transaction, + "outbox", + "created_at", + TSTZ, + "week", + "timestamptz", + ); + // MINVALUE historic partition, ignored when anchoring. + await addMinvalueChild( + transaction, + "outbox", + "outbox_historic", + "2025-01-01 00:00:00+00", + ); + await addChild( + transaction, + "outbox", + "outbox_p2026w03", + "2026-01-12 00:00:00+00", + "2026-01-19 00:00:00+00", + ); + + // 3 classic-pgslice tables (parent has no PK; each partition owns its own). + for (const sessions of ["sessions", "tokens", "messages"]) { + await pgsliceParent( + transaction, + sessions, + "created_at", + TSTZ, + "week", + "timestamptz", + ); + await addChild( + transaction, + sessions, + `${sessions}_2026w03`, + "2026-01-12 00:00:00+00", + "2026-01-19 00:00:00+00", + ["id", "created_at"], + ); + } + + // Noise that discovery must ignore: a partitioned parent with no settings + // comment, and an ordinary (non-partitioned) table. + await transaction.query(sql.unsafe` + CREATE TABLE unmanaged_part ( + id bigint NOT NULL, + created_at timestamp with time zone NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at) + `); + await addChild( + transaction, + "unmanaged_part", + "unmanaged_part_2026m01", + "2026-01-01 00:00:00+00", + "2026-02-01 00:00:00+00", + ); + await transaction.query( + sql.unsafe`CREATE TABLE regular_table (id bigint PRIMARY KEY)`, + ); + }); + + test("discovers exactly the settings-commented partitioned tables", async ({ + pgslice, + transaction, + }) => { + const results = await pgslice.maintain(transaction, { future: 3 }); + + expect(results.map((r) => r.table).sort()).toEqual( + [ + "public.outbox", + "public.daily_reports", + "public.tokens", + "public.audit_log", + "public.reactions", + "public.comments", + "public.posts", + "public.sessions", + "public.messages", + "analytics.visits", + ].sort(), + ); + }); + + test("classifies native vs pgslice tables correctly", async ({ + pgslice, + transaction, + }) => { + const results = await pgslice.maintain(transaction, { future: 3 }); + const model = new Map(results.map((r) => [r.table, r.model])); + + for (const native of [ + "public.posts", + "public.comments", + "public.reactions", + "public.daily_reports", + "analytics.visits", + "public.audit_log", + "public.outbox", + ]) { + expect(model.get(native)).toBe("native"); + } + for (const classic of [ + "public.sessions", + "public.tokens", + "public.messages", + ]) { + expect(model.get(classic)).toBe("pgslice"); + } + }); + + test("extends every table by bounds with the expected pgslice-named partitions", async ({ + pgslice, + transaction, + }) => { + const results = await pgslice.maintain(transaction, { future: 3 }); + const created = new Map( + results.map((r) => [r.table, [...r.partitionsCreated].sort()]), + ); + + expect(created.get("public.posts")).toEqual([ + "posts_202602", + "posts_202603", + "posts_202604", + ]); + expect(created.get("public.daily_reports")).toEqual([ + "daily_reports_202602", + "daily_reports_202603", + "daily_reports_202604", + ]); + expect(created.get("analytics.visits")).toEqual([ + "visits_2026w04", + "visits_2026w05", + "visits_2026w06", + ]); + expect(created.get("public.audit_log")).toEqual([ + "audit_log_2026w03", + "audit_log_2026w04", + "audit_log_2026w05", + "audit_log_2026w06", + ]); + expect(created.get("public.outbox")).toEqual([ + "outbox_2026w04", + "outbox_2026w05", + "outbox_2026w06", + ]); + expect(created.get("public.sessions")).toEqual([ + "sessions_2026w04", + "sessions_2026w05", + "sessions_2026w06", + ]); + + // Legacy-named and special partitions are left in place. + expect(await childNames(transaction, "outbox")).toContain( + "outbox_historic", + ); + expect(await childNames(transaction, "analytics.visits")).toContain( + "visits_default", + ); + expect(await childNames(transaction, "public.posts")).toContain( + "posts_y2026m01", + ); + }); + + test("every managed table ends up CDC-ready", async ({ + pgslice, + transaction, + }) => { + const results = await pgslice.maintain(transaction, { future: 3 }); + for (const result of results) { + expect(result.replicaIdentityReady).toBe(true); + expect(result.unsafePartitions).toEqual([]); + } + }); + + test("is idempotent: a second run creates nothing", async ({ + pgslice, + transaction, + }) => { + await pgslice.maintain(transaction, { future: 3 }); + const second = await pgslice.maintain(transaction, { future: 3 }); + for (const result of second) { + expect(result.partitionsCreated).toEqual([]); + } + }); + + test("can be restricted to a single schema", async ({ + pgslice, + transaction, + }) => { + const results = await pgslice.maintain(transaction, { + future: 3, + schema: "analytics", + }); + expect(results.map((r) => r.table)).toEqual(["analytics.visits"]); + }); + }); + + describe("native tables", () => { + test("inherit the parent composite PK and the parent's grants on new partitions", async ({ + pgslice, + transaction, + }) => { + await createCdcRole(transaction, "replica_reader_maint"); + await nativeParent( + transaction, + "posts", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "posts", + "posts_y2026m01", + "2026-01-01", + "2026-02-01", + ); + await transaction.query( + sql.unsafe`GRANT SELECT ON TABLE posts TO replica_reader_maint`, + ); + + await pgslice.maintain(transaction, { future: 1 }); + + const pkDefs = await transaction.any( + sql.type(z.object({ def: z.string() }))` + SELECT pg_get_constraintdef(oid) AS def FROM pg_constraint + WHERE conrelid = 'public.posts_202602'::regclass AND contype = 'p' + `, + ); + expect(pkDefs.map((c) => c.def)).toEqual([ + "PRIMARY KEY (id, created_at)", + ]); + + const grants = await transaction.any( + sql.type(z.object({ grantee: z.string(), privilege: z.string() }))` + SELECT grantee_role.rolname AS grantee, acl.privilege_type AS privilege + FROM pg_class c + CROSS JOIN LATERAL aclexplode(c.relacl) AS acl + JOIN pg_roles grantee_role ON grantee_role.oid = acl.grantee + WHERE c.oid = 'public.posts_202602'::regclass + AND grantee_role.rolname = 'replica_reader_maint' + `, + ); + expect(grants.map((g) => g.privilege)).toContain("SELECT"); + }); + }); + + describe("classic pgslice tables", () => { + test("get a per-partition composite primary key", async ({ + pgslice, + transaction, + }) => { + await pgsliceParent( + transaction, + "sessions", + "created_at", + TSTZ, + "week", + "timestamptz", + ); + await addChild( + transaction, + "sessions", + "sessions_2026w03", + "2026-01-12 00:00:00+00", + "2026-01-19 00:00:00+00", + ["id", "created_at"], + ); + + const results = await pgslice.maintain(transaction, { future: 1 }); + expect(results[0].model).toBe("pgslice"); + + const pkDefs = await transaction.any( + sql.type(z.object({ def: z.string() }))` + SELECT pg_get_constraintdef(oid) AS def FROM pg_constraint + WHERE conrelid = 'public.sessions_2026w04'::regclass AND contype = 'p' + `, + ); + expect(pkDefs.map((c) => c.def)).toEqual([ + "PRIMARY KEY (id, created_at)", + ]); + }); + }); + + describe("replica identity", () => { + test("a parent REPLICA IDENTITY USING INDEX propagates its partitioned index to new partitions automatically", async ({ + pgslice, + transaction, + }) => { + await nativeParent( + transaction, + "analytics.visits", + "created_at", + TSTZ, + "week", + "timestamptz", + ); + await addChild( + transaction, + "analytics.visits", + "visits_2026w03", + "2026-01-12 00:00:00+00", + "2026-01-19 00:00:00+00", + ); + await addDefault(transaction, "analytics.visits", "visits_default"); + // Build the partitioned replica-identity index exactly as the runner does: + // ON ONLY the parent, then attach a child index for every existing leaf, + // then point REPLICA IDENTITY at it. + await transaction.query(sql.unsafe` + CREATE UNIQUE INDEX visits_ri_uidx + ON ONLY analytics.visits (id, created_at) + `); + await transaction.query(sql.unsafe` + CREATE UNIQUE INDEX visits_2026w03_ri + ON analytics.visits_2026w03 (id, created_at) + `); + await transaction.query(sql.unsafe` + ALTER INDEX analytics.visits_ri_uidx + ATTACH PARTITION analytics.visits_2026w03_ri + `); + await transaction.query(sql.unsafe` + CREATE UNIQUE INDEX visits_default_ri + ON analytics.visits_default (id, created_at) + `); + await transaction.query(sql.unsafe` + ALTER INDEX analytics.visits_ri_uidx + ATTACH PARTITION analytics.visits_default_ri + `); + await transaction.query(sql.unsafe` + ALTER TABLE analytics.visits + REPLICA IDENTITY USING INDEX visits_ri_uidx + `); + + const results = await pgslice.maintain(transaction, { future: 1 }); + expect(results[0].replicaIdentityReady).toBe(true); + + // The new partition auto-got a unique (id, created_at) index attached to + // the parent's partitioned replica-identity index, and it is valid. (The + // new partition also has its own inherited composite PK index — that is a + // second unique index, which is why we filter to the one attached to the + // replica-identity index here.) + const idx = await transaction.any( + sql.type(z.object({ valid: z.boolean() }))` + SELECT ix.indisvalid AS valid + FROM pg_inherits pi + JOIN pg_class parent_idx ON parent_idx.oid = pi.inhparent + JOIN pg_class ic ON ic.oid = pi.inhrelid + JOIN pg_index ix ON ix.indexrelid = ic.oid + JOIN pg_class t ON t.oid = ix.indrelid + JOIN pg_namespace n ON n.oid = t.relnamespace + WHERE parent_idx.relname = 'visits_ri_uidx' + AND n.nspname = 'analytics' + AND t.relname = 'visits_2026w04' + `, + ); + expect(idx).toHaveLength(1); + expect(idx[0].valid).toBe(true); + + // The parent's replica-identity index stays valid after extension. + const parentIdx = await transaction.one( + sql.type(z.object({ valid: z.boolean() }))` + SELECT ix.indisvalid AS valid FROM pg_index ix + JOIN pg_class ic ON ic.oid = ix.indexrelid + WHERE ic.relname = 'visits_ri_uidx' + `, + ); + expect(parentIdx.valid).toBe(true); + }); + + test("flags a pre-existing partition that has no usable replica identity", async ({ + pgslice, + transaction, + }) => { + await nativeParent( + transaction, + "weird", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "weird", + "weird_2026m01", + "2026-01-01", + "2026-02-01", + ); + // Simulate a leaf created outside pgslice that is not CDC-safe. + await transaction.query(sql.unsafe` + ALTER TABLE weird_2026m01 REPLICA IDENTITY NOTHING + `); + + const results = await pgslice.maintain(transaction, { future: 1 }); + const weird = results.find((r) => r.table === "public.weird"); + + expect(weird?.replicaIdentityReady).toBe(false); + expect(weird?.unsafePartitions).toEqual(["weird_2026m01"]); + }); + }); + + describe("failure isolation", () => { + test("a non-empty DEFAULT blocking the next partition fails that table alone; the rest of the fleet still extends", async ({ + pgslice, + transaction, + }) => { + // `blocked` has a DEFAULT holding a row that belongs in the next range, + // so creating that partition will fail (the production drain footgun). + await nativeParent( + transaction, + "blocked", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "blocked", + "blocked_y2026m01", + "2026-01-01", + "2026-02-01", + ); + await addDefault(transaction, "blocked", "blocked_default"); + await transaction.query(sql.unsafe` + INSERT INTO blocked (id, created_at, payload) VALUES (1, '2026-02-15', 'absorbed') + `); + // `healthy` is an ordinary native monthly table. + await nativeParent( + transaction, + "healthy", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "healthy", + "healthy_y2026m01", + "2026-01-01", + "2026-02-01", + ); + + const results = await pgslice.maintain(transaction, { future: 1 }); + const blocked = results.find((r) => r.table === "public.blocked"); + const healthy = results.find((r) => r.table === "public.healthy"); + + // The blocked table's CREATE is rejected (slonik surfaces it as a check + // integrity constraint violation); it is recorded as failed, not thrown. + expect(blocked?.error).toBeTruthy(); + expect(blocked?.partitionsCreated).toEqual([]); + // The healthy table is extended despite the other table's failure. + expect(healthy?.error).toBeNull(); + expect(healthy?.partitionsCreated).toEqual(["healthy_202602"]); + }); + }); +}); diff --git a/src/pgslice.ts b/src/pgslice.ts index 0006731..85079bd 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -7,6 +7,8 @@ import { } from "slonik"; import { createQueryLoggingInterceptor } from "slonik-interceptor-query-logging"; +import { z } from "zod"; + import type { AddPartitionsOptions, AnalyzeOptions, @@ -15,6 +17,9 @@ import type { EnableMirroringOptions, FillBatchResult, FillOptions, + MaintainOptions, + MaintainResult, + PartitionModel, Period, PrepOptions, StatusOptions, @@ -496,6 +501,140 @@ export class Pgslice { ); } + /** + * Maintains every managed partitioned table the connection can see: each + * partitioned parent (`relkind = 'p'`) carrying a valid pgslice settings + * comment is discovered from the catalog and extended via {@link addPartitions}. + * New partitioned tables are therefore picked up automatically with no + * per-table configuration. + * + * The native-vs-pgslice distinction is handled by {@link addPartitions}: it + * skips the per-partition primary key when the parent owns one (the inherited + * composite key and any partitioned indexes propagate automatically) and adds + * a per-partition key otherwise. New partitions are created with the default + * replica identity, so each leaf's row identity is its own (or inherited) + * primary key and no replica-identity DDL is required. After extending each + * table this verifies every leaf has a usable replica identity, surfacing a + * misconfigured table rather than silently shipping a CDC-unsafe partition. + * + * Each table is maintained independently: a failure on one table (for example + * a non-empty DEFAULT partition that blocks creating the next one) is recorded + * on that table's result and does not stop the rest of the fleet. + */ + async maintain( + connection: DatabasePoolConnection, + options: MaintainOptions, + ): Promise { + const past = options.past ?? 0; + const future = options.future ?? 0; + // Capture one reference instant for the whole fleet, so a run that straddles + // a period boundary uses a consistent horizon for every table. + const now = options.now ?? new Date(); + + const managed = await this.#discoverManagedTables( + connection, + options.schema, + ); + + const results: MaintainResult[] = []; + for (const table of managed) { + try { + const partitionsCreated = await this.addPartitions(connection, { + table: table.toString(), + past, + future, + tablespace: options.tablespace, + inheritGrants: options.inheritGrants, + now, + }); + + const { model, partitionCount, unsafePartitions } = + await connection.transaction(async (tx) => { + const parentPrimaryKey = await table.primaryKeyColumns(tx); + const partitions = await table.partitions(tx); + const unsafe = await table.unsafeReplicaIdentityPartitions(tx); + const model: PartitionModel = + parentPrimaryKey.length > 0 ? "native" : "pgslice"; + return { + model, + partitionCount: partitions.length, + unsafePartitions: unsafe, + }; + }); + + results.push({ + table: table.toString(), + model, + partitionsCreated, + partitionCount, + replicaIdentityReady: unsafePartitions.length === 0, + unsafePartitions, + error: null, + }); + } catch (error) { + results.push({ + table: table.toString(), + model: null, + partitionsCreated: [], + partitionCount: 0, + replicaIdentityReady: false, + unsafePartitions: [], + error: error instanceof Error ? error.message : String(error), + }); + } + } + + return results; + } + + /** + * Discovers the managed partitioned parents visible to this connection: a + * top-level partitioned table (`relkind = 'p'`, not itself a partition) whose + * table comment parses as pgslice settings. Optionally restricted to a single + * schema. The same `pg_partitioned_table`-style catalog scan the runway + * monitor uses, so scheduler and monitor share one source of truth. + */ + async #discoverManagedTables( + connection: DatabasePoolConnection, + schema?: string, + ): Promise { + const schemaFilter = schema + ? sql.fragment`AND n.nspname = ${schema}` + : sql.fragment``; + + const rows = await connection.any( + sql.type( + z.object({ + schema: z.string(), + name: z.string(), + comment: z.string().nullable(), + }), + )` + SELECT + n.nspname AS schema, + c.relname AS name, + obj_description(c.oid, 'pg_class') AS comment + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.relkind = 'p' + AND n.nspname NOT IN ('pg_catalog', 'information_schema') + AND NOT EXISTS ( + SELECT 1 FROM pg_inherits inh WHERE inh.inhrelid = c.oid + ) + ${schemaFilter} + ORDER BY n.nspname, c.relname + `, + ); + + return rows.flatMap((row) => { + if (!row.comment) { + return []; + } + const settings = TableSettings.parseFromComment(row.comment); + return settings ? [new Table(row.schema, row.name)] : []; + }); + } + /** * Resolves the primary key columns to place on each new partition in the * classic pgslice model (where the parent has no primary key of its own). diff --git a/src/table.ts b/src/table.ts index 6999647..c0f22d6 100644 --- a/src/table.ts +++ b/src/table.ts @@ -497,6 +497,46 @@ export class Table { }); } + /** + * Returns the leaf partitions of this table that lack a replica identity + * usable for logical replication. A partition is unsafe when it has + * `REPLICA IDENTITY NOTHING`, or the default identity with no primary key — + * in either case Postgres rejects `UPDATE`/`DELETE` on a published table. + * Partitions covered by a primary key (the native parent-owned key or a + * per-partition key), or an explicit FULL / USING INDEX identity, are safe. + * Read-only: it only inspects the catalog and never emits replica-identity + * DDL. Assumes single-level partitioning (direct children only). + */ + async unsafeReplicaIdentityPartitions( + tx: CommonQueryMethods, + ): Promise { + const rows = await tx.any( + sql.type(z.object({ name: z.string() }))` + SELECT child.relname AS name + FROM pg_inherits i + JOIN pg_class parent ON parent.oid = i.inhparent + JOIN pg_class child ON child.oid = i.inhrelid + JOIN pg_namespace nmsp_parent ON nmsp_parent.oid = parent.relnamespace + WHERE + nmsp_parent.nspname = ${this.schema} AND + parent.relname = ${this.name} AND + ( + child.relreplident = 'n' + OR ( + child.relreplident = 'd' + AND NOT EXISTS ( + SELECT 1 FROM pg_index pk + WHERE pk.indrelid = child.oid AND pk.indisprimary + ) + ) + ) + ORDER BY child.relname ASC + `, + ); + + return rows.map((r) => r.name); + } + /** * Fetches the partition settings from this table's comment. */ diff --git a/src/types.ts b/src/types.ts index e50f451..40db583 100644 --- a/src/types.ts +++ b/src/types.ts @@ -225,3 +225,61 @@ export interface TableStatus { retiredMirrorTriggerExists: boolean; originalIsPartitioned: boolean; } + +/** + * Options for the `maintain` command. + */ +export interface MaintainOptions { + past?: number; + future?: number; + /** + * Restrict maintenance to partitioned tables in this schema. When omitted, + * every managed partitioned table the connection can see is maintained. + */ + schema?: string; + tablespace?: string; + inheritGrants?: boolean; + /** + * The reference "now" for choosing each table's partition horizon. Defaults + * to a single instant captured when the run starts, so a fleet run that + * straddles a period boundary uses one consistent horizon for every table. + */ + now?: Date; +} + +/** + * How a managed table's primary key is owned, which decides how + * `add_partitions` treats each new partition. + * - "native": the partitioned parent owns the (often composite) primary key, so + * Postgres propagates it — and any partitioned indexes — to each new partition. + * - "pgslice": the classic pgslice model where the parent has no primary key and + * each partition owns its own. + */ +export type PartitionModel = "native" | "pgslice"; + +/** + * Outcome of maintaining a single managed table. + */ +export interface MaintainResult { + table: string; + /** The partitioning model, or null if maintenance failed before it was determined. */ + model: PartitionModel | null; + partitionsCreated: string[]; + partitionCount: number; + /** + * Whether every leaf partition has a replica identity usable for logical + * replication. New partitions are created with the default replica identity, + * so each leaf's row identity is its own (or inherited) primary key and no + * replica-identity DDL is required; this read-only flag surfaces a leaf + * lacking a usable identity rather than shipping a CDC-unsafe partition. + */ + replicaIdentityReady: boolean; + /** Leaf partitions that would not be CDC-safe (no usable replica identity). */ + unsafePartitions: string[]; + /** + * Error message if maintaining this table failed; null on success. One + * table's failure (e.g. a non-empty DEFAULT blocking the next partition) is + * recorded here and does not stop the rest of the fleet from being extended. + */ + error: string | null; +} From 0c1b44a90ffb2bdd9e9a35575191bbbe20c8a945 Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 10:40:35 -0400 Subject: [PATCH 2/8] feat: Give maintain per-period future horizons A single --future N means very different runway for weekly vs monthly vs yearly tables (3 weeks vs 3 months vs 3 years). Replace it with per-period --future-daily / --future-weekly / --future-monthly / --future-yearly (defaults 90 / 26 / 6 / 1); discovery returns each table's period, and each table is extended by the horizon for its own period, so the fleet gets comparable forward coverage. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/maintain.test.ts | 6 +- src/commands/maintain.ts | 30 ++++++-- src/maintain.test.ts | 138 +++++++++++++++++++++++++++++++--- src/pgslice.ts | 20 +++-- src/types.ts | 12 ++- 5 files changed, 178 insertions(+), 28 deletions(-) diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index c3c7d6e..fe71d0d 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -31,7 +31,7 @@ describe("MaintainCommand", () => { }) => { await createPosts(transaction); - const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + const exitCode = await cli.run(["maintain"], commandContext); expect(exitCode).toBe(0); expect(commandContext.stdout.read()?.toString()).toContain( @@ -49,7 +49,7 @@ describe("MaintainCommand", () => { ALTER TABLE posts_y2026m01 REPLICA IDENTITY NOTHING `); - const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + const exitCode = await cli.run(["maintain"], commandContext); expect(exitCode).toBe(1); expect(commandContext.stderr.read()?.toString()).toContain( @@ -67,7 +67,7 @@ describe("MaintainCommand", () => { PARTITION BY RANGE (created_at) `); - const exitCode = await cli.run(["maintain", "--future=1"], commandContext); + const exitCode = await cli.run(["maintain"], commandContext); expect(exitCode).toBe(0); expect(commandContext.stdout.read()?.toString()).toContain( diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts index 1fcc5cf..eedbcd9 100644 --- a/src/commands/maintain.ts +++ b/src/commands/maintain.ts @@ -23,13 +23,14 @@ export class MaintainCommand extends BaseCommand { `, examples: [ [ - "Extend all managed tables three periods ahead", - "$0 maintain --future 3", + "Extend all managed tables with the default per-period horizons", + "$0 maintain", ], [ - "Restrict to a single schema", - "$0 maintain --future 3 --schema analytics", + "Override one period's horizon (12 months of monthly runway)", + "$0 maintain --future-monthly 12", ], + ["Restrict to a single schema", "$0 maintain --schema analytics"], ], }); @@ -37,8 +38,20 @@ export class MaintainCommand extends BaseCommand { description: "Number of past partitions to add to each table", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); - future = Option.String("--future", "0", { - description: "Number of future partitions to add to each table", + futureDaily = Option.String("--future-daily", "90", { + description: "Future partitions to keep ahead for daily tables", + validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), + }); + futureWeekly = Option.String("--future-weekly", "26", { + description: "Future partitions to keep ahead for weekly tables", + validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), + }); + futureMonthly = Option.String("--future-monthly", "6", { + description: "Future partitions to keep ahead for monthly tables", + validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), + }); + futureYearly = Option.String("--future-yearly", "1", { + description: "Future partitions to keep ahead for yearly tables", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); schema = Option.String("--schema", { @@ -56,7 +69,10 @@ export class MaintainCommand extends BaseCommand { const results = await pgslice.start((connection) => pgslice.maintain(connection, { past: this.past, - future: this.future, + futureDaily: this.futureDaily, + futureWeekly: this.futureWeekly, + futureMonthly: this.futureMonthly, + futureYearly: this.futureYearly, schema: this.schema, tablespace: this.tablespace, inheritGrants: this.inheritGrants, diff --git a/src/maintain.test.ts b/src/maintain.test.ts index 9a1460b..9801d84 100644 --- a/src/maintain.test.ts +++ b/src/maintain.test.ts @@ -173,7 +173,12 @@ describe("Pgslice.maintain (fleet)", () => { pgslice, transaction, }) => { - const results = await pgslice.maintain(transaction, { future: 3 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); expect(results.map((r) => r.table).sort()).toEqual( [ @@ -195,7 +200,12 @@ describe("Pgslice.maintain (fleet)", () => { pgslice, transaction, }) => { - const results = await pgslice.maintain(transaction, { future: 3 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); const model = new Map(results.map((r) => [r.table, r.model])); for (const native of [ @@ -222,7 +232,12 @@ describe("Pgslice.maintain (fleet)", () => { pgslice, transaction, }) => { - const results = await pgslice.maintain(transaction, { future: 3 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); const created = new Map( results.map((r) => [r.table, [...r.partitionsCreated].sort()]), ); @@ -275,7 +290,12 @@ describe("Pgslice.maintain (fleet)", () => { pgslice, transaction, }) => { - const results = await pgslice.maintain(transaction, { future: 3 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); for (const result of results) { expect(result.replicaIdentityReady).toBe(true); expect(result.unsafePartitions).toEqual([]); @@ -286,8 +306,18 @@ describe("Pgslice.maintain (fleet)", () => { pgslice, transaction, }) => { - await pgslice.maintain(transaction, { future: 3 }); - const second = await pgslice.maintain(transaction, { future: 3 }); + await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); + const second = await pgslice.maintain(transaction, { + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, + }); for (const result of second) { expect(result.partitionsCreated).toEqual([]); } @@ -298,7 +328,10 @@ describe("Pgslice.maintain (fleet)", () => { transaction, }) => { const results = await pgslice.maintain(transaction, { - future: 3, + futureDaily: 3, + futureWeekly: 3, + futureMonthly: 3, + futureYearly: 3, schema: "analytics", }); expect(results.map((r) => r.table)).toEqual(["analytics.visits"]); @@ -330,7 +363,12 @@ describe("Pgslice.maintain (fleet)", () => { sql.unsafe`GRANT SELECT ON TABLE posts TO replica_reader_maint`, ); - await pgslice.maintain(transaction, { future: 1 }); + await pgslice.maintain(transaction, { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 1, + futureYearly: 1, + }); const pkDefs = await transaction.any( sql.type(z.object({ def: z.string() }))` @@ -378,7 +416,12 @@ describe("Pgslice.maintain (fleet)", () => { ["id", "created_at"], ); - const results = await pgslice.maintain(transaction, { future: 1 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 1, + futureYearly: 1, + }); expect(results[0].model).toBe("pgslice"); const pkDefs = await transaction.any( @@ -442,7 +485,12 @@ describe("Pgslice.maintain (fleet)", () => { REPLICA IDENTITY USING INDEX visits_ri_uidx `); - const results = await pgslice.maintain(transaction, { future: 1 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 1, + futureYearly: 1, + }); expect(results[0].replicaIdentityReady).toBe(true); // The new partition auto-got a unique (id, created_at) index attached to @@ -502,7 +550,12 @@ describe("Pgslice.maintain (fleet)", () => { ALTER TABLE weird_2026m01 REPLICA IDENTITY NOTHING `); - const results = await pgslice.maintain(transaction, { future: 1 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 1, + futureYearly: 1, + }); const weird = results.find((r) => r.table === "public.weird"); expect(weird?.replicaIdentityReady).toBe(false); @@ -553,7 +606,12 @@ describe("Pgslice.maintain (fleet)", () => { "2026-02-01", ); - const results = await pgslice.maintain(transaction, { future: 1 }); + const results = await pgslice.maintain(transaction, { + futureDaily: 1, + futureWeekly: 1, + futureMonthly: 1, + futureYearly: 1, + }); const blocked = results.find((r) => r.table === "public.blocked"); const healthy = results.find((r) => r.table === "public.healthy"); @@ -566,4 +624,60 @@ describe("Pgslice.maintain (fleet)", () => { expect(healthy?.partitionsCreated).toEqual(["healthy_202602"]); }); }); + + describe("per-period horizons", () => { + test("extends each table by the horizon for its own period", async ({ + pgslice, + transaction, + }) => { + // One run over a monthly and a weekly table with different per-period + // futures. The weekly table (shorter period + larger future) must get + // strictly more new partitions than the monthly one — proving the horizon + // is applied per period, not uniformly. + await nativeParent( + transaction, + "m_tbl", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "m_tbl", + "m_tbl_y2026m01", + "2026-01-01", + "2026-02-01", + ); + await nativeParent( + transaction, + "w_tbl", + "created_at", + TS, + "week", + "date", + ); + await addChild( + transaction, + "w_tbl", + "w_tbl_2026w02", + "2026-01-05", + "2026-01-12", + ); + + const results = await pgslice.maintain(transaction, { + futureMonthly: 2, + futureWeekly: 6, + }); + const m = results.find((r) => r.table === "public.m_tbl"); + const w = results.find((r) => r.table === "public.w_tbl"); + + expect(m?.error).toBeNull(); + expect(w?.error).toBeNull(); + const mCount = m?.partitionsCreated.length ?? 0; + const wCount = w?.partitionsCreated.length ?? 0; + expect(mCount).toBeGreaterThan(0); + expect(wCount).toBeGreaterThan(mCount); + }); + }); }); diff --git a/src/pgslice.ts b/src/pgslice.ts index 85079bd..76a7adc 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -526,7 +526,15 @@ export class Pgslice { options: MaintainOptions, ): Promise { const past = options.past ?? 0; - const future = options.future ?? 0; + // Future runway is set per period: "N periods" is very different runway for a + // weekly vs a monthly vs a yearly table, so each table uses the horizon for + // its own period, giving the fleet comparable forward coverage. + const futureByPeriod: Record = { + day: options.futureDaily ?? 90, + week: options.futureWeekly ?? 26, + month: options.futureMonthly ?? 6, + year: options.futureYearly ?? 1, + }; // Capture one reference instant for the whole fleet, so a run that straddles // a period boundary uses a consistent horizon for every table. const now = options.now ?? new Date(); @@ -537,12 +545,12 @@ export class Pgslice { ); const results: MaintainResult[] = []; - for (const table of managed) { + for (const { table, period } of managed) { try { const partitionsCreated = await this.addPartitions(connection, { table: table.toString(), past, - future, + future: futureByPeriod[period], tablespace: options.tablespace, inheritGrants: options.inheritGrants, now, @@ -597,7 +605,7 @@ export class Pgslice { async #discoverManagedTables( connection: DatabasePoolConnection, schema?: string, - ): Promise { + ): Promise> { const schemaFilter = schema ? sql.fragment`AND n.nspname = ${schema}` : sql.fragment``; @@ -631,7 +639,9 @@ export class Pgslice { return []; } const settings = TableSettings.parseFromComment(row.comment); - return settings ? [new Table(row.schema, row.name)] : []; + return settings + ? [{ table: new Table(row.schema, row.name), period: settings.period }] + : []; }); } diff --git a/src/types.ts b/src/types.ts index 40db583..0677ac9 100644 --- a/src/types.ts +++ b/src/types.ts @@ -231,7 +231,17 @@ export interface TableStatus { */ export interface MaintainOptions { past?: number; - future?: number; + /** + * Future partitions to keep ahead, set per period. "N periods" is very + * different runway for a weekly vs a monthly vs a yearly table, so the horizon + * is per period and every table gets comparable forward coverage; each + * discovered table uses the value for its own period. Defaults when omitted: + * daily 90, weekly 26, monthly 6, yearly 1. + */ + futureDaily?: number; + futureWeekly?: number; + futureMonthly?: number; + futureYearly?: number; /** * Restrict maintenance to partitioned tables in this schema. When omitted, * every managed partitioned table the connection can see is maintained. From c930e2f4c0f819481780acaf40f1708beea1190d Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 12:15:52 -0400 Subject: [PATCH 3/8] feat: Emit maintain progress as structured JSONL logs maintain now writes one JSON object per line so a log collector can extract the keys as attributes: - a start record (target db + per-period horizons), - one record per table extended (msg, level, target, partition counts, success), - a final summary (succeeded/failed counts and table lists). Only info and error levels are used, and the partitioning model is left out of the logs. The command signals failure through its exit code (now honored by BaseCommand.execute) instead of a plain-text error line, keeping stdout pure JSONL. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/base.ts | 2 +- src/commands/maintain.test.ts | 95 +++++++++++++++++++++++++++++------ src/commands/maintain.ts | 70 +++++++++++--------------- src/pgslice.ts | 67 ++++++++++++++++++++++-- src/types.ts | 8 +++ 5 files changed, 181 insertions(+), 61 deletions(-) diff --git a/src/commands/base.ts b/src/commands/base.ts index c90c04e..768aa6e 100644 --- a/src/commands/base.ts +++ b/src/commands/base.ts @@ -25,7 +25,7 @@ export abstract class BaseCommand extends Command { new URL(this.getDatabaseUrl()), ); - await this.perform(this.context.pgslice); + return await this.perform(this.context.pgslice); } catch (error) { if (error instanceof Error) { this.context.stderr.write(`${error.message}\n`); diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index fe71d0d..0293dc7 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -4,6 +4,25 @@ import { sql, type DatabaseTransactionConnection } from "slonik"; import { commandTest as test } from "../testing/index.js"; import { MaintainCommand } from "./maintain.js"; +interface LogEntry { + msg: string; + level: string; + target: { db: string; schema?: string; table?: string }; + future?: { daily: number; weekly: number; monthly: number; yearly: number }; + partitions?: { new: number; total: number }; + success?: number; + succeeded?: { count: number; tables: string[] }; + failed?: { count: number; tables: string[] }; +} + +/** Parse the command's stdout as JSONL (one log record per line). */ +function jsonLines(output: string | undefined): LogEntry[] { + return (output ?? "") + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line): LogEntry => JSON.parse(line)); +} + describe("MaintainCommand", () => { test.scoped({ commandClass: ({}, use) => use(MaintainCommand) }); @@ -24,7 +43,7 @@ describe("MaintainCommand", () => { `); } - test("exits 0 and reports a per-table summary on a healthy fleet", async ({ + test("emits JSONL and exits 0 on a healthy fleet", async ({ cli, commandContext, transaction, @@ -32,14 +51,47 @@ describe("MaintainCommand", () => { await createPosts(transaction); const exitCode = await cli.run(["maintain"], commandContext); - expect(exitCode).toBe(0); - expect(commandContext.stdout.read()?.toString()).toContain( - "public.posts [native]", - ); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + + // Every record shares the shape, uses only info/error, and never leaks the + // partitioning model. + for (const entry of logs) { + expect(typeof entry.msg).toBe("string"); + expect(["info", "error"]).toContain(entry.level); + expect(entry.target.db).toBeTypeOf("string"); + expect(JSON.stringify(entry)).not.toContain("model"); + } + + const start = logs[0]; + expect(start.msg).toBe("Running pgslice maintain"); + expect(start.level).toBe("info"); + expect(start.future).toEqual({ + daily: 90, + weekly: 26, + monthly: 6, + yearly: 1, + }); + + const table = logs.find((entry) => entry.target.table === "posts"); + expect(table?.msg).toBe("Extended table successfully"); + expect(table?.level).toBe("info"); + expect(table?.success).toBe(1); + expect(table?.target.schema).toBe("public"); + expect(table?.partitions).toEqual({ + new: expect.any(Number), + total: expect.any(Number), + }); + + const final = logs.at(-1); + expect(final?.msg).toBe("Finished pgslice maintain successfully"); + expect(final?.level).toBe("info"); + expect(final?.succeeded).toEqual({ count: 1, tables: ["public.posts"] }); + expect(final?.failed).toEqual({ count: 0, tables: [] }); }); - test("exits 1 when a leaf has no usable replica identity", async ({ + test("exits 1 and logs an error record when a leaf has no usable replica identity", async ({ cli, commandContext, transaction, @@ -50,14 +102,22 @@ describe("MaintainCommand", () => { `); const exitCode = await cli.run(["maintain"], commandContext); - expect(exitCode).toBe(1); - expect(commandContext.stderr.read()?.toString()).toContain( - "usable replica identity", - ); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + + const table = logs.find((entry) => entry.target.table === "posts"); + expect(table?.level).toBe("error"); + expect(table?.success).toBe(0); + expect(table?.msg).toContain("replica identity"); + + const final = logs.at(-1); + expect(final?.msg).toBe("Finished pgslice maintain with errors"); + expect(final?.level).toBe("error"); + expect(final?.failed).toEqual({ count: 1, tables: ["public.posts"] }); }); - test("reports no managed tables when none carry a settings comment", async ({ + test("exits 0 with an empty summary when no managed tables exist", async ({ cli, commandContext, transaction, @@ -68,10 +128,15 @@ describe("MaintainCommand", () => { `); const exitCode = await cli.run(["maintain"], commandContext); - expect(exitCode).toBe(0); - expect(commandContext.stdout.read()?.toString()).toContain( - "No managed partitioned tables found.", - ); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + expect(logs[0].msg).toBe("Running pgslice maintain"); + expect(logs.some((entry) => entry.target.table !== undefined)).toBe(false); + + const final = logs.at(-1); + expect(final?.msg).toBe("Finished pgslice maintain successfully"); + expect(final?.succeeded).toEqual({ count: 0, tables: [] }); + expect(final?.failed).toEqual({ count: 0, tables: [] }); }); }); diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts index eedbcd9..59f9f7f 100644 --- a/src/commands/maintain.ts +++ b/src/commands/maintain.ts @@ -3,6 +3,7 @@ import * as t from "typanion"; import { BaseCommand } from "./base.js"; import { Pgslice } from "../pgslice.js"; +import type { MaintainLog } from "../types.js"; export class MaintainCommand extends BaseCommand { static override paths = [["maintain"]]; @@ -65,49 +66,36 @@ export class MaintainCommand extends BaseCommand { "Copy each parent table's privileges onto its new partitions (default: true; disable with --no-inherit-grants)", }); - override async perform(pgslice: Pgslice): Promise { + override async perform(pgslice: Pgslice): Promise { + // maintain emits structured records (start, per-table, final); write each as + // one JSON object per line so Datadog can extract the keys as attributes. + const log: MaintainLog = (entry) => { + this.context.stdout.write(`${JSON.stringify(entry)}\n`); + }; + const results = await pgslice.start((connection) => - pgslice.maintain(connection, { - past: this.past, - futureDaily: this.futureDaily, - futureWeekly: this.futureWeekly, - futureMonthly: this.futureMonthly, - futureYearly: this.futureYearly, - schema: this.schema, - tablespace: this.tablespace, - inheritGrants: this.inheritGrants, - }), + pgslice.maintain( + connection, + { + past: this.past, + futureDaily: this.futureDaily, + futureWeekly: this.futureWeekly, + futureMonthly: this.futureMonthly, + futureYearly: this.futureYearly, + schema: this.schema, + tablespace: this.tablespace, + inheritGrants: this.inheritGrants, + }, + log, + ), ); - if (results.length === 0) { - this.context.stdout.write("No managed partitioned tables found.\n"); - return; - } - - const problems: string[] = []; - for (const result of results) { - if (result.error) { - this.context.stdout.write( - `${result.table}: FAILED — ${result.error}\n`, - ); - problems.push(`${result.table}: ${result.error}`); - continue; - } - - this.context.stdout.write( - `${result.table} [${result.model}]: +${result.partitionsCreated.length} partition(s), ${result.partitionCount} total\n`, - ); - if (!result.replicaIdentityReady) { - problems.push( - `${result.table}: partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`, - ); - } - } - - if (problems.length > 0) { - throw new Error( - `maintain encountered problems:\n ${problems.join("\n ")}`, - ); - } + // Exit non-zero if any table failed to extend or shipped a CDC-unsafe leaf; + // as a CronJob that surfaces as a failed Job. The final summary log already + // carries the detail, so no separate error line is written. + const failed = results.some( + (result) => result.error !== null || !result.replicaIdentityReady, + ); + return failed ? 1 : undefined; } } diff --git a/src/pgslice.ts b/src/pgslice.ts index 76a7adc..cb99e5c 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -17,6 +17,7 @@ import type { EnableMirroringOptions, FillBatchResult, FillOptions, + MaintainLog, MaintainOptions, MaintainResult, PartitionModel, @@ -520,10 +521,16 @@ export class Pgslice { * Each table is maintained independently: a failure on one table (for example * a non-empty DEFAULT partition that blocks creating the next one) is recorded * on that table's result and does not stop the rest of the fleet. + * + * The optional `log` sink receives structured records — one on start, one per + * table as it is extended, and one final summary — for the command to emit as + * JSONL. It defaults to a no-op, so callers that only need the results (such + * as tests) can ignore it. */ async maintain( connection: DatabasePoolConnection, options: MaintainOptions, + log: MaintainLog = () => {}, ): Promise { const past = options.past ?? 0; // Future runway is set per period: "N periods" is very different runway for a @@ -539,13 +546,33 @@ export class Pgslice { // a period boundary uses a consistent horizon for every table. const now = options.now ?? new Date(); + const { db } = await connection.one( + sql.type(z.object({ db: z.string() }))`SELECT current_database() AS db`, + ); + + log({ + msg: "Running pgslice maintain", + level: "info", + target: { db }, + future: { + daily: futureByPeriod.day, + weekly: futureByPeriod.week, + monthly: futureByPeriod.month, + yearly: futureByPeriod.year, + }, + }); + const managed = await this.#discoverManagedTables( connection, options.schema, ); const results: MaintainResult[] = []; + const succeeded: string[] = []; + const failed: string[] = []; + for (const { table, period } of managed) { + let result: MaintainResult; try { const partitionsCreated = await this.addPartitions(connection, { table: table.toString(), @@ -570,7 +597,7 @@ export class Pgslice { }; }); - results.push({ + result = { table: table.toString(), model, partitionsCreated, @@ -578,9 +605,9 @@ export class Pgslice { replicaIdentityReady: unsafePartitions.length === 0, unsafePartitions, error: null, - }); + }; } catch (error) { - results.push({ + result = { table: table.toString(), model: null, partitionsCreated: [], @@ -588,10 +615,42 @@ export class Pgslice { replicaIdentityReady: false, unsafePartitions: [], error: error instanceof Error ? error.message : String(error), - }); + }; } + + results.push(result); + + // A table is a success only when it extended without error and every leaf + // is CDC-safe; a created-but-CDC-unsafe table is a failure the run reports. + const ok = result.error === null && result.replicaIdentityReady; + (ok ? succeeded : failed).push(result.table); + + log({ + msg: ok + ? "Extended table successfully" + : (result.error ?? + `Partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`), + level: ok ? "info" : "error", + target: { db, schema: table.schema, table: table.name }, + partitions: { + new: result.partitionsCreated.length, + total: result.partitionCount, + }, + success: ok ? 1 : 0, + }); } + log({ + msg: + failed.length === 0 + ? "Finished pgslice maintain successfully" + : "Finished pgslice maintain with errors", + level: failed.length === 0 ? "info" : "error", + target: { db }, + succeeded: { count: succeeded.length, tables: succeeded }, + failed: { count: failed.length, tables: failed }, + }); + return results; } diff --git a/src/types.ts b/src/types.ts index 0677ac9..bc9ffac 100644 --- a/src/types.ts +++ b/src/types.ts @@ -257,6 +257,14 @@ export interface MaintainOptions { now?: Date; } +/** + * Sink for maintain's structured JSONL logs. Each call receives one log record + * whose keys become Datadog attributes; the `maintain` command wires this to + * stdout as one JSON object per line. Only `info` and `error` records are + * emitted — there is no debug/warning level. + */ +export type MaintainLog = (entry: Record) => void; + /** * How a managed table's primary key is owned, which decides how * `add_partitions` treats each new partition. From 793319f3cdf1fe5fb5ba69b738c31fe9c8b57bca Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 12:55:48 -0400 Subject: [PATCH 4/8] feat: Log the target host and distinguish no-op maintain runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the endpoint host (from the connection URL — host only, no credentials) to every maintain log record's target alongside the database name, so a run identifies which host and DB it extended. Distinguish a table that needed no new partitions ("Table already up to date; no extension needed") from one that was extended, and drop the deployment-specific rationale from the code comments. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/maintain.test.ts | 38 +++++++++++++++++++++++++- src/commands/maintain.ts | 7 ++++- src/maintain.test.ts | 51 +++++++++++++++++++++++++++++++++++ src/pgslice.ts | 25 ++++++++++++----- src/types.ts | 15 ++++++++--- 5 files changed, 123 insertions(+), 13 deletions(-) diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index 0293dc7..e219057 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -7,7 +7,7 @@ import { MaintainCommand } from "./maintain.js"; interface LogEntry { msg: string; level: string; - target: { db: string; schema?: string; table?: string }; + target: { db: string; host: string; schema?: string; table?: string }; future?: { daily: number; weekly: number; monthly: number; yearly: number }; partitions?: { new: number; total: number }; success?: number; @@ -61,6 +61,7 @@ describe("MaintainCommand", () => { expect(typeof entry.msg).toBe("string"); expect(["info", "error"]).toContain(entry.level); expect(entry.target.db).toBeTypeOf("string"); + expect(entry.target.host).toBe("localhost"); expect(JSON.stringify(entry)).not.toContain("model"); } @@ -110,6 +111,7 @@ describe("MaintainCommand", () => { expect(table?.level).toBe("error"); expect(table?.success).toBe(0); expect(table?.msg).toContain("replica identity"); + expect(table?.target.host).toBe("localhost"); const final = logs.at(-1); expect(final?.msg).toBe("Finished pgslice maintain with errors"); @@ -139,4 +141,38 @@ describe("MaintainCommand", () => { expect(final?.succeeded).toEqual({ count: 0, tables: [] }); expect(final?.failed).toEqual({ count: 0, tables: [] }); }); + + test("logs a no-op record when a table already has runway beyond the horizon", async ({ + cli, + commandContext, + transaction, + }) => { + // A managed parent whose only partition is far in the future: the forward + // horizon is already covered, so maintain creates nothing. + await transaction.query(sql.unsafe` + CREATE TABLE posts ( + id bigint NOT NULL, + created_at timestamp without time zone NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at) + `); + await transaction.query(sql.unsafe` + COMMENT ON TABLE posts IS 'column:created_at,period:month,cast:date,version:3' + `); + await transaction.query(sql.unsafe` + CREATE TABLE posts_y2099m01 PARTITION OF posts + FOR VALUES FROM ('2099-01-01') TO ('2099-02-01') + `); + + const exitCode = await cli.run(["maintain"], commandContext); + expect(exitCode).toBe(0); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + const table = logs.find((entry) => entry.target.table === "posts"); + expect(table?.msg).toBe("Table already up to date; no extension needed"); + expect(table?.level).toBe("info"); + expect(table?.success).toBe(1); + expect(table?.partitions).toEqual({ new: 0, total: 1 }); + expect(table?.target.host).toBe("localhost"); + }); }); diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts index 59f9f7f..b2dd6d5 100644 --- a/src/commands/maintain.ts +++ b/src/commands/maintain.ts @@ -68,15 +68,20 @@ export class MaintainCommand extends BaseCommand { override async perform(pgslice: Pgslice): Promise { // maintain emits structured records (start, per-table, final); write each as - // one JSON object per line so Datadog can extract the keys as attributes. + // one JSON object per line (JSONL). const log: MaintainLog = (entry) => { this.context.stdout.write(`${JSON.stringify(entry)}\n`); }; + // Host only — never the credentials in the URL — so the logs identify the + // endpoint being maintained without leaking secrets. + const host = new URL(this.getDatabaseUrl()).hostname; + const results = await pgslice.start((connection) => pgslice.maintain( connection, { + host, past: this.past, futureDaily: this.futureDaily, futureWeekly: this.futureWeekly, diff --git a/src/maintain.test.ts b/src/maintain.test.ts index 9801d84..719f437 100644 --- a/src/maintain.test.ts +++ b/src/maintain.test.ts @@ -16,6 +16,14 @@ import { TSTZ, } from "./testing/shapes.js"; +/** A captured maintain JSONL log record, for asserting record shape from tests. */ +interface MaintainLogRecord { + msg: string; + level: string; + success?: number; + target: { host?: string; db: string; schema?: string; table?: string }; +} + /** * End-to-end validation of `maintain` against the real production table shapes * on PostgreSQL 13.20: catalog discovery, the native (parent-owned PK) vs @@ -623,6 +631,49 @@ describe("Pgslice.maintain (fleet)", () => { expect(healthy?.error).toBeNull(); expect(healthy?.partitionsCreated).toEqual(["healthy_202602"]); }); + + test("emits an error record carrying the failure message and host", async ({ + pgslice, + transaction, + }) => { + await nativeParent( + transaction, + "blocked", + "created_at", + TS, + "month", + "date", + ); + await addChild( + transaction, + "blocked", + "blocked_y2026m01", + "2026-01-01", + "2026-02-01", + ); + await addDefault(transaction, "blocked", "blocked_default"); + await transaction.query(sql.unsafe` + INSERT INTO blocked (id, created_at, payload) VALUES (1, '2026-02-15', 'absorbed') + `); + + const lines: string[] = []; + const results = await pgslice.maintain( + transaction, + { futureMonthly: 1, host: "clone.example" }, + (entry) => { + lines.push(JSON.stringify(entry)); + }, + ); + const records = lines.map((line): MaintainLogRecord => JSON.parse(line)); + + const blocked = results.find((r) => r.table === "public.blocked"); + const record = records.find((r) => r.target.table === "blocked"); + // The caught error is surfaced verbatim as the record's msg. + expect(record?.msg).toBe(blocked?.error); + expect(record?.level).toBe("error"); + expect(record?.success).toBe(0); + expect(record?.target.host).toBe("clone.example"); + }); }); describe("per-period horizons", () => { diff --git a/src/pgslice.ts b/src/pgslice.ts index cb99e5c..e3c6f3c 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -549,11 +549,14 @@ export class Pgslice { const { db } = await connection.one( sql.type(z.object({ db: z.string() }))`SELECT current_database() AS db`, ); + // Every record carries the endpoint host (from the connection URL) and the + // database name, so a run identifies which host and DB it extended. + const target = { host: options.host, db }; log({ msg: "Running pgslice maintain", level: "info", - target: { db }, + target, future: { daily: futureByPeriod.day, weekly: futureByPeriod.week, @@ -625,13 +628,21 @@ export class Pgslice { const ok = result.error === null && result.replicaIdentityReady; (ok ? succeeded : failed).push(result.table); + let msg: string; + if (result.error !== null) { + msg = result.error; + } else if (!result.replicaIdentityReady) { + msg = `Partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`; + } else if (result.partitionsCreated.length === 0) { + msg = "Table already up to date; no extension needed"; + } else { + msg = "Extended table successfully"; + } + log({ - msg: ok - ? "Extended table successfully" - : (result.error ?? - `Partitions without a usable replica identity (CDC-unsafe): ${result.unsafePartitions.join(", ")}`), + msg, level: ok ? "info" : "error", - target: { db, schema: table.schema, table: table.name }, + target: { ...target, schema: table.schema, table: table.name }, partitions: { new: result.partitionsCreated.length, total: result.partitionCount, @@ -646,7 +657,7 @@ export class Pgslice { ? "Finished pgslice maintain successfully" : "Finished pgslice maintain with errors", level: failed.length === 0 ? "info" : "error", - target: { db }, + target, succeeded: { count: succeeded.length, tables: succeeded }, failed: { count: failed.length, tables: failed }, }); diff --git a/src/types.ts b/src/types.ts index bc9ffac..d7cfce2 100644 --- a/src/types.ts +++ b/src/types.ts @@ -242,6 +242,13 @@ export interface MaintainOptions { futureWeekly?: number; futureMonthly?: number; futureYearly?: number; + /** + * Host of the endpoint being maintained, recorded on every log record's + * `target` alongside the database name so operators can see which host was + * extended. The command derives it from the connection URL (host only, no + * credentials); when omitted, `target.host` is simply absent. + */ + host?: string; /** * Restrict maintenance to partitioned tables in this schema. When omitted, * every managed partitioned table the connection can see is maintained. @@ -258,10 +265,10 @@ export interface MaintainOptions { } /** - * Sink for maintain's structured JSONL logs. Each call receives one log record - * whose keys become Datadog attributes; the `maintain` command wires this to - * stdout as one JSON object per line. Only `info` and `error` records are - * emitted — there is no debug/warning level. + * Sink for maintain's structured JSONL logs. Each call receives one log record; + * the `maintain` command serializes it to stdout as one JSON object per line so + * downstream log tooling can lift the keys into attributes. Only `info` and + * `error` records are emitted — there is no debug/warning level. */ export type MaintainLog = (entry: Record) => void; From c708b40b68cd81af630e3bd9233ab8cccce51530 Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 13:09:17 -0400 Subject: [PATCH 5/8] feat: Stamp a per-run jobId on every maintain log record Generate one UUID per maintain run and stamp it on the start, per-table, and final records so all logs from a single invocation can be correlated. Generated with crypto.randomUUID; injectable via options for deterministic tests. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/maintain.test.ts | 8 ++++++++ src/maintain.test.ts | 5 ++++- src/pgslice.ts | 7 +++++++ src/types.ts | 6 ++++++ 4 files changed, 25 insertions(+), 1 deletion(-) diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index e219057..a09ece0 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -5,6 +5,7 @@ import { commandTest as test } from "../testing/index.js"; import { MaintainCommand } from "./maintain.js"; interface LogEntry { + jobId: string; msg: string; level: string; target: { db: string; host: string; schema?: string; table?: string }; @@ -65,6 +66,13 @@ describe("MaintainCommand", () => { expect(JSON.stringify(entry)).not.toContain("model"); } + // Every record from one run shares a single generated jobId (a UUID). + const jobIds = new Set(logs.map((entry) => entry.jobId)); + expect(jobIds.size).toBe(1); + expect([...jobIds][0]).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i, + ); + const start = logs[0]; expect(start.msg).toBe("Running pgslice maintain"); expect(start.level).toBe("info"); diff --git a/src/maintain.test.ts b/src/maintain.test.ts index 719f437..5c25f92 100644 --- a/src/maintain.test.ts +++ b/src/maintain.test.ts @@ -18,6 +18,7 @@ import { /** A captured maintain JSONL log record, for asserting record shape from tests. */ interface MaintainLogRecord { + jobId: string; msg: string; level: string; success?: number; @@ -659,7 +660,7 @@ describe("Pgslice.maintain (fleet)", () => { const lines: string[] = []; const results = await pgslice.maintain( transaction, - { futureMonthly: 1, host: "clone.example" }, + { futureMonthly: 1, host: "clone.example", jobId: "job-under-test" }, (entry) => { lines.push(JSON.stringify(entry)); }, @@ -673,6 +674,8 @@ describe("Pgslice.maintain (fleet)", () => { expect(record?.level).toBe("error"); expect(record?.success).toBe(0); expect(record?.target.host).toBe("clone.example"); + // The run's jobId is stamped on every record. + expect(records.every((r) => r.jobId === "job-under-test")).toBe(true); }); }); diff --git a/src/pgslice.ts b/src/pgslice.ts index e3c6f3c..c363aab 100644 --- a/src/pgslice.ts +++ b/src/pgslice.ts @@ -8,6 +8,7 @@ import { import { createQueryLoggingInterceptor } from "slonik-interceptor-query-logging"; import { z } from "zod"; +import { randomUUID } from "node:crypto"; import type { AddPartitionsOptions, @@ -545,6 +546,9 @@ export class Pgslice { // Capture one reference instant for the whole fleet, so a run that straddles // a period boundary uses a consistent horizon for every table. const now = options.now ?? new Date(); + // One correlation id per run, stamped on every record so all logs from this + // invocation can be grouped. + const jobId = options.jobId ?? randomUUID(); const { db } = await connection.one( sql.type(z.object({ db: z.string() }))`SELECT current_database() AS db`, @@ -554,6 +558,7 @@ export class Pgslice { const target = { host: options.host, db }; log({ + jobId, msg: "Running pgslice maintain", level: "info", target, @@ -640,6 +645,7 @@ export class Pgslice { } log({ + jobId, msg, level: ok ? "info" : "error", target: { ...target, schema: table.schema, table: table.name }, @@ -652,6 +658,7 @@ export class Pgslice { } log({ + jobId, msg: failed.length === 0 ? "Finished pgslice maintain successfully" diff --git a/src/types.ts b/src/types.ts index d7cfce2..8872461 100644 --- a/src/types.ts +++ b/src/types.ts @@ -262,6 +262,12 @@ export interface MaintainOptions { * straddles a period boundary uses one consistent horizon for every table. */ now?: Date; + /** + * Correlation id stamped on every log record for this run, so all records + * from one invocation can be grouped. A fresh id is generated per run when + * omitted. + */ + jobId?: string; } /** From 7a80474661158b5ba8af5a3f0d6c4d025d2f83fb Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 13:41:37 -0400 Subject: [PATCH 6/8] fix: Don't let advisory-lock release mask the handler's error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AdvisoryLock.withLock released the session-level lock in a finally block. When the handler aborts the transaction (e.g. a failed partition CREATE), the unlock query itself errors with "current transaction is aborted", and that finally exception replaced the handler's real error — so maintain logged an unhelpful message instead of the cause. Release now only propagates its own failure when the handler succeeded; otherwise the handler's error wins. The test harness runs with advisory locks disabled, so this only surfaced against a live database. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/advisory-lock.ts | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/advisory-lock.ts b/src/advisory-lock.ts index e6d5aa2..4891c2b 100644 --- a/src/advisory-lock.ts +++ b/src/advisory-lock.ts @@ -26,10 +26,26 @@ export abstract class AdvisoryLock { handler: () => Promise, ): Promise { const release = await this.acquire(connection, table, operation); + let handlerThrew = false; try { return await handler(); + } catch (error) { + handlerThrew = true; + throw error; } finally { - await release(); + try { + await release(); + } catch (releaseError) { + // A release failure must not mask the handler's error. When the handler + // aborts the transaction (e.g. a failed CREATE), the unlock query itself + // errors with "current transaction is aborted"; surfacing that would + // hide the real cause. The session-level lock is freed when the + // transaction or session ends regardless, so only propagate a release + // failure when the handler itself succeeded. + if (!handlerThrew) { + throw releaseError; + } + } } } From 3e5899dffa416113bb310d05ce40bb41c66d9cb6 Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 15:13:04 -0400 Subject: [PATCH 7/8] feat: Let PGSLICE_FUTURE_* env vars set maintain's per-period horizons MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Give each --future-* flag a clipanion env fallback (PGSLICE_FUTURE_DAILY / _WEEKLY / _MONTHLY / _YEARLY), so a scheduled job can configure the horizons via environment — as the Terrace ScheduledProcess will — without templating the command args. Precedence is flag > env > baked default, matching how --url already falls back to PGSLICE_URL. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/maintain.test.ts | 47 +++++++++++++++++++++++++++++++++++ src/commands/maintain.ts | 8 ++++++ 2 files changed, 55 insertions(+) diff --git a/src/commands/maintain.test.ts b/src/commands/maintain.test.ts index a09ece0..5ebd94e 100644 --- a/src/commands/maintain.test.ts +++ b/src/commands/maintain.test.ts @@ -183,4 +183,51 @@ describe("MaintainCommand", () => { expect(table?.partitions).toEqual({ new: 0, total: 1 }); expect(table?.target.host).toBe("localhost"); }); + + test("reads per-period horizons from PGSLICE_FUTURE_* env vars", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts(transaction); + process.env.PGSLICE_FUTURE_MONTHLY = "3"; + process.env.PGSLICE_FUTURE_WEEKLY = "9"; + try { + const exitCode = await cli.run(["maintain"], commandContext); + expect(exitCode).toBe(0); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + // env overrides the baked default for its period; unset periods keep theirs. + expect(logs[0].future).toEqual({ + daily: 90, + weekly: 9, + monthly: 3, + yearly: 1, + }); + } finally { + delete process.env.PGSLICE_FUTURE_MONTHLY; + delete process.env.PGSLICE_FUTURE_WEEKLY; + } + }); + + test("an explicit --future-monthly flag overrides the env var", async ({ + cli, + commandContext, + transaction, + }) => { + await createPosts(transaction); + process.env.PGSLICE_FUTURE_MONTHLY = "3"; + try { + const exitCode = await cli.run( + ["maintain", "--future-monthly", "7"], + commandContext, + ); + expect(exitCode).toBe(0); + + const logs = jsonLines(commandContext.stdout.read()?.toString()); + expect(logs[0].future?.monthly).toBe(7); + } finally { + delete process.env.PGSLICE_FUTURE_MONTHLY; + } + }); }); diff --git a/src/commands/maintain.ts b/src/commands/maintain.ts index b2dd6d5..d36f7e0 100644 --- a/src/commands/maintain.ts +++ b/src/commands/maintain.ts @@ -31,6 +31,10 @@ export class MaintainCommand extends BaseCommand { "Override one period's horizon (12 months of monthly runway)", "$0 maintain --future-monthly 12", ], + [ + "Configure a horizon via environment (e.g. from a scheduled job)", + "PGSLICE_FUTURE_MONTHLY=12 $0 maintain", + ], ["Restrict to a single schema", "$0 maintain --schema analytics"], ], }); @@ -41,18 +45,22 @@ export class MaintainCommand extends BaseCommand { }); futureDaily = Option.String("--future-daily", "90", { description: "Future partitions to keep ahead for daily tables", + env: "PGSLICE_FUTURE_DAILY", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); futureWeekly = Option.String("--future-weekly", "26", { description: "Future partitions to keep ahead for weekly tables", + env: "PGSLICE_FUTURE_WEEKLY", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); futureMonthly = Option.String("--future-monthly", "6", { description: "Future partitions to keep ahead for monthly tables", + env: "PGSLICE_FUTURE_MONTHLY", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); futureYearly = Option.String("--future-yearly", "1", { description: "Future partitions to keep ahead for yearly tables", + env: "PGSLICE_FUTURE_YEARLY", validator: t.cascade(t.isNumber(), [t.isInteger(), t.isAtLeast(0)]), }); schema = Option.String("--schema", { From 90686e4ede30334db762da7fd309abefbf0dce79 Mon Sep 17 00:00:00 2001 From: Cody Swaney Date: Wed, 1 Jul 2026 15:44:59 -0400 Subject: [PATCH 8/8] fix: Address advisory-lock leak and nested-leaf CDC gap (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - withLock takes a transaction-scoped advisory lock (pg_try_advisory_xact_lock) instead of a session lock released in a finally. It's freed when the enclosing transaction commits or rolls back, so a handler that aborts the transaction can't leak the lock into the pooled connection — and there's no unlock query left to fail on an aborted transaction and mask the handler's error (this supersedes the earlier swallow-the-error workaround). The session-based acquire() is kept for fill/synchronize, which hold a lock across batches. - unsafeReplicaIdentityPartitions walks the whole partition tree via pg_partition_tree(...) filtered to leaves, so a leaf beneath a sub-partitioned child is inspected too — the previous direct-children query could report a table CDC-ready while a nested leaf had an unusable replica identity. Adds a leak-regression test (an aborted handler no longer blocks a second session) and a nested-leaf CDC test. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/advisory-lock.test.ts | 51 ++++++++++++++++++++++++++++++++++++--- src/advisory-lock.ts | 48 ++++++++++++++++++------------------ src/table.test.ts | 41 +++++++++++++++++++++++++++++++ src/table.ts | 25 +++++++++---------- 4 files changed, 125 insertions(+), 40 deletions(-) diff --git a/src/advisory-lock.test.ts b/src/advisory-lock.test.ts index 690fd99..06beccb 100644 --- a/src/advisory-lock.test.ts +++ b/src/advisory-lock.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "vitest"; -import { createPool } from "slonik"; +import { createPool, sql } from "slonik"; import { pgsliceTest as test } from "./testing/index.js"; import { AdvisoryLock, AdvisoryLockError } from "./advisory-lock.js"; @@ -20,7 +20,9 @@ describe("AdvisoryLock.withLock", () => { expect(result).toBe("success"); }); - test("releases lock even if handler throws", async ({ transaction }) => { + test("propagates the handler's error and stays re-acquirable in the transaction", async ({ + transaction, + }) => { const table = Table.parse("test_table"); await expect( @@ -29,7 +31,7 @@ describe("AdvisoryLock.withLock", () => { }), ).rejects.toThrow("handler error"); - // Should be able to acquire the lock again since it was released + // Re-acquirable within the same transaction (xact advisory locks are re-entrant) const result = await AdvisoryLock.withLock( transaction, table, @@ -75,6 +77,49 @@ describe("AdvisoryLock.withLock", () => { await pool2.end(); } }); + + test("does not leak the lock when the handler aborts the transaction", async ({ + databaseUrl, + }) => { + const table = Table.parse("test_table"); + const operation = "test_op"; + + const pool1 = await createPool(databaseUrl.toString(), { + maximumPoolSize: 1, + queryRetryLimit: 0, + }); + const pool2 = await createPool(databaseUrl.toString(), { + maximumPoolSize: 1, + queryRetryLimit: 0, + }); + + try { + // The handler aborts its transaction with a failing statement. A + // session-scoped lock survived the rollback (leaked); a transaction-scoped + // lock is released when the transaction rolls back. + await expect( + pool1.transaction((tx1) => + AdvisoryLock.withLock(tx1, table, operation, async () => { + await tx1.query(sql.unsafe`SELECT 1 FROM pgslice_missing_relation`); + }), + ), + ).rejects.toThrow(); + + // A different session can now take the lock — proof it was released. + await pool2.transaction(async (tx2) => { + const result = await AdvisoryLock.withLock( + tx2, + table, + operation, + async () => "acquired after rollback", + ); + expect(result).toBe("acquired after rollback"); + }); + } finally { + await pool1.end(); + await pool2.end(); + } + }); }); describe("AdvisoryLock.acquire", () => { diff --git a/src/advisory-lock.ts b/src/advisory-lock.ts index 4891c2b..31f4e2c 100644 --- a/src/advisory-lock.ts +++ b/src/advisory-lock.ts @@ -16,8 +16,15 @@ export class AdvisoryLockError extends Error { export abstract class AdvisoryLock { /** - * Executes a handler while holding an advisory lock. - * The lock is automatically released when the handler completes or throws. + * Runs a handler while holding a transaction-scoped advisory lock. The lock is + * acquired on the given transaction and released automatically when that + * transaction commits or rolls back — so a handler that aborts the transaction + * cannot leak the lock, and there is no unlock query that could fail on an + * aborted transaction and mask the handler's error. + * + * Must be called inside a transaction (every pgslice command wraps it in + * `connection.transaction(...)`). For a lock that must span multiple + * transactions — e.g. a batched generator — use {@link acquire} instead. */ static async withLock( connection: CommonQueryMethods, @@ -25,28 +32,11 @@ export abstract class AdvisoryLock { operation: string, handler: () => Promise, ): Promise { - const release = await this.acquire(connection, table, operation); - let handlerThrew = false; - try { - return await handler(); - } catch (error) { - handlerThrew = true; - throw error; - } finally { - try { - await release(); - } catch (releaseError) { - // A release failure must not mask the handler's error. When the handler - // aborts the transaction (e.g. a failed CREATE), the unlock query itself - // errors with "current transaction is aborted"; surfacing that would - // hide the real cause. The session-level lock is freed when the - // transaction or session ends regardless, so only propagate a release - // failure when the handler itself succeeded. - if (!handlerThrew) { - throw releaseError; - } - } + const key = await this.#getKey(connection, table, operation); + if (!(await this.#tryAcquireXact(connection, key))) { + throw new AdvisoryLockError(table, operation); } + return handler(); } /** @@ -96,6 +86,18 @@ export abstract class AdvisoryLock { return result.acquired; } + static async #tryAcquireXact( + connection: CommonQueryMethods, + key: bigint, + ): Promise { + const result = await connection.one( + sql.type(z.object({ acquired: z.boolean() }))` + SELECT pg_try_advisory_xact_lock(${key}) AS acquired + `, + ); + return result.acquired; + } + static async #release( connection: CommonQueryMethods, key: bigint, diff --git a/src/table.test.ts b/src/table.test.ts index 36cd7b6..40414a2 100644 --- a/src/table.test.ts +++ b/src/table.test.ts @@ -738,3 +738,44 @@ describe("Table.rangePartitionBounds", () => { expect(finiteRange?.upper).toEqual(new Date("2026-01-19T00:00:00Z")); }); }); + +describe("Table.unsafeReplicaIdentityPartitions", () => { + test("flags an unsafe leaf beneath a sub-partitioned child", async ({ + transaction, + }) => { + await transaction.query(sql.unsafe` + CREATE TABLE nested_parent ( + id bigint NOT NULL, + created_at timestamp NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at) + `); + // A direct child that is itself partitioned (multi-level partitioning). + await transaction.query(sql.unsafe` + CREATE TABLE nested_parent_2026 PARTITION OF nested_parent + FOR VALUES FROM ('2026-01-01') TO ('2027-01-01') + PARTITION BY RANGE (created_at) + `); + await transaction.query(sql.unsafe` + CREATE TABLE nested_parent_2026m01 PARTITION OF nested_parent_2026 + FOR VALUES FROM ('2026-01-01') TO ('2026-02-01') + `); + await transaction.query(sql.unsafe` + CREATE TABLE nested_parent_2026m02 PARTITION OF nested_parent_2026 + FOR VALUES FROM ('2026-02-01') TO ('2026-03-01') + `); + await transaction.query(sql.unsafe` + ALTER TABLE nested_parent_2026m02 REPLICA IDENTITY NOTHING + `); + + const unsafe = + await Table.parse("nested_parent").unsafeReplicaIdentityPartitions( + transaction, + ); + + // Only true leaves are inspected. The unsafe nested leaf is caught; a + // direct-children-only check would see the (safe) sub-partitioned child and + // miss it. + expect(unsafe).toEqual(["nested_parent_2026m02"]); + }); +}); diff --git a/src/table.ts b/src/table.ts index c0f22d6..5f16d3b 100644 --- a/src/table.ts +++ b/src/table.ts @@ -505,32 +505,29 @@ export class Table { * Partitions covered by a primary key (the native parent-owned key or a * per-partition key), or an explicit FULL / USING INDEX identity, are safe. * Read-only: it only inspects the catalog and never emits replica-identity - * DDL. Assumes single-level partitioning (direct children only). + * DDL. Walks the whole partition tree, so leaf partitions beneath a + * sub-partitioned child are inspected too — not just direct children. */ async unsafeReplicaIdentityPartitions( tx: CommonQueryMethods, ): Promise { const rows = await tx.any( sql.type(z.object({ name: z.string() }))` - SELECT child.relname AS name - FROM pg_inherits i - JOIN pg_class parent ON parent.oid = i.inhparent - JOIN pg_class child ON child.oid = i.inhrelid - JOIN pg_namespace nmsp_parent ON nmsp_parent.oid = parent.relnamespace - WHERE - nmsp_parent.nspname = ${this.schema} AND - parent.relname = ${this.name} AND - ( - child.relreplident = 'n' + SELECT leaf.relname AS name + FROM pg_partition_tree(${this.regclassLiteral}::regclass) tree + JOIN pg_class leaf ON leaf.oid = tree.relid + WHERE tree.isleaf + AND ( + leaf.relreplident = 'n' OR ( - child.relreplident = 'd' + leaf.relreplident = 'd' AND NOT EXISTS ( SELECT 1 FROM pg_index pk - WHERE pk.indrelid = child.oid AND pk.indisprimary + WHERE pk.indrelid = leaf.oid AND pk.indisprimary ) ) ) - ORDER BY child.relname ASC + ORDER BY leaf.relname ASC `, );