From 89a82431f2e152d84b8c087544df3b1ec4517ada Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 29 Aug 2026 06:30:46 +0000 Subject: [PATCH 1/4] Replace pg-boss with owned PostgreSQL store Co-authored-by: Evan Tahler --- .env.example | 2 +- .github/workflows/test.yaml | 4 +- CLAUDE.md | 19 +- README.md | 20 +- __tests__/core/connection.test.ts | 137 +++++++++++- __tests__/core/connectionError.test.ts | 8 +- __tests__/core/queue.test.ts | 39 ++-- __tests__/utils/specHelper.ts | 25 ++- bun.lock | 17 +- docs/plans/00-overview.md | 61 +++--- docs/plans/01-repo-scaffold.md | 13 +- docs/plans/02-connection-and-schema.md | 68 +++--- docs/plans/03-queue.md | 26 +-- docs/plans/04-worker.md | 16 +- docs/plans/05-scheduler.md | 6 +- docs/plans/06-plugins.md | 6 +- docs/plans/07-multiworker.md | 3 +- docs/plans/08-conformance-tests.md | 9 +- docs/plans/09-docs-site.md | 6 +- docs/plans/10-publish-and-ci.md | 9 +- docs/plans/README.md | 4 +- migrations/001_initial.sql | 60 ++++++ package.json | 10 +- scripts/assert-node-package.mjs | 6 + src/core/connection.ts | 287 +++++++++++-------------- src/core/queue.ts | 160 ++++++-------- src/index.ts | 3 +- src/types/errorPayload.ts | 2 +- 28 files changed, 573 insertions(+), 453 deletions(-) create mode 100644 migrations/001_initial.sql diff --git a/.env.example b/.env.example index 52524ab..2a35073 100644 --- a/.env.example +++ b/.env.example @@ -1 +1 @@ -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index f78838c..c5df5b5 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -33,7 +33,7 @@ jobs: image: postgres:16 env: POSTGRES_PASSWORD: postgres - POSTGRES_DB: pgboss_queue_test + POSTGRES_DB: pg_queue_test options: >- --health-cmd pg_isready --health-interval 10s @@ -47,7 +47,7 @@ jobs: - run: bun install --frozen-lockfile - run: bun run test env: - DATABASE_URL: postgres://postgres:postgres@localhost:5432/pgboss_queue_test + DATABASE_URL: postgres://postgres:postgres@localhost:5432/pg_queue_test node-package: runs-on: ubuntu-latest diff --git a/CLAUDE.md b/CLAUDE.md index e88a5a0..6f02ae9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ Guidance for Claude Code and other agents working in this repository. ## Project -`pgboss-queue` is a TypeScript background-job library. The public API is the node-resque trio — `Queue`, `Worker`, `Scheduler` — plus `MultiWorker` and `Plugins`. Storage is PostgreSQL via [pg-boss](https://github.com/timgit/pg-boss), not Redis. +`pg-queue` is a TypeScript background-job library. The public API is the node-resque trio — `Queue`, `Worker`, `Scheduler` — plus `MultiWorker` and `Plugins`. Storage is PostgreSQL through our versioned `pgrq_*` schema, not Redis. This is an Actionhero project (sibling of [node-resque](https://github.com/actionhero/node-resque) and [keryx](https://github.com/actionhero/keryx)). It exists so Keryx and other apps can keep the resque worker/scheduler pattern without a Redis dependency for jobs. @@ -60,17 +60,17 @@ bun docs:dev # VitePress (Phase 9) Local Postgres: set `DATABASE_URL` (see `.env.example`). CI starts Postgres as a workflow service; there is no `docker-compose.yml`. ```bash -# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test +# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test ``` Tests create and tear down the configured schema per file (see `specHelper`). Never point tests at a production database. ## Architecture (do not violate) -1. **node-resque is the runtime model.** Classes, events, plugin hooks, worker names (`hostname:pid[+id]`), queue priority (array order), leader-elected scheduler. Port behavior; do not "simplify" it into pg-boss's `work()` helper. -2. **pg-boss is the job store.** Jobs live in pg-boss's `job` table. Dequeue with `fetch` + `SKIP LOCKED` (or equivalent SQL), not `LPOP`. Delayed jobs use `startAfter`. Do not reimplement a job table next to pg-boss. -3. **We own metadata pg-boss does not.** Workers, heartbeats, leader lock, plugin locks, and processed/failed counters live in *our* tables in the same schema (or a documented adjacent schema). See Phase 2. -4. **The elected scheduler is the only migrator and sweeper.** `automigrate` and completed-job deletion run on the leader, never on every worker. Workers start pg-boss with `migrate: false` and `supervise: false`. +1. **node-resque is the runtime model.** Classes, events, plugin hooks, worker names (`hostname:pid[+id]`), queue priority (array order), leader-elected scheduler. Port behavior rather than replacing it with a generic handler API. +2. **We own the job store.** Jobs live in `pgrq_jobs`; queue names live in `pgrq_queues`. Dequeue atomically with `FOR UPDATE SKIP LOCKED`, not `LPOP`. Delayed jobs use `start_after`. +3. **We own metadata.** Workers, heartbeats, leader lock, plugin locks, and processed/failed counters live in `pgrq_*` tables in the same schema. See Phase 2. +4. **The elected scheduler is the only migrator and sweeper.** `automigrate` and completed-job deletion run on the leader, never on every worker. 5. **Keryx PR #519 is research, not a copy target.** Steal: connection strings, schema isolation, SQL introspection on `job`, `short` policy for singleton pending jobs, `deleteAfterSeconds` thinking. Do not steal: dropping worker heartbeats, dropping plugins, replacing `Worker` with a single `boss.work` handler, removing the scheduler. ## Public API @@ -111,8 +111,8 @@ type ConnectionOptions = { user?: string; password?: string; ssl?: boolean | object; - pool?: import("pg").Pool; // bring-your-own (maps to pg-boss `db`) - schema?: string; // default "pgboss_queue" (was Redis namespace) + pool?: import("pg").Pool; // bring-your-own pool + schema?: string; // default "pg_queue" (was Redis namespace) }; type SchedulerOptions = ConnectionOptions & { @@ -137,7 +137,7 @@ Do **not** accept `pkg: "ioredis"`, `redis: Redis`, or `database: number`. Those - **Biome** for format/lint (keryx-style), not Prettier. - **Tests use `bun:test`**, not Jest. Port node-resque tests faithfully: same `describe` / `test` names, same assertions, Postgres `specHelper` instead of Redis. Node must still be able to import the compiled package (`node scripts/assert-node-package.mjs`); do not run the Bun suite on Node. - **Every behavior change ships with tests.** A PR with no test changes is a red flag unless it is docs-only. -- **Do not add dependencies** unless a phase plan names them. Expected runtime deps: `pg-boss`, `pg`. Dev: `typescript`, `@types/pg`, `biome`, `bun` types. +- **Do not add dependencies** unless a phase plan names them. Expected runtime dependency: `pg`. Dev: `typescript`, `@types/pg`, `biome`, `bun` types. ## Testing rules @@ -183,4 +183,3 @@ Follow keryx: bump `version` in `package.json` on every user-facing PR (patch fo - node-resque README + `src/core/*` + `__tests__/*` — source of truth for behavior - node-resque examples (`example.ts`, `multiWorker.ts`, `scheduledJobs.ts`, `retry.ts`, `stuckWorker.ts`, `cluster.ts`) - [keryx#519](https://github.com/actionhero/keryx/pull/519) — `PgBossBackend.ts`, `TaskBackend.ts`, `config/tasks.ts` -- pg-boss docs — constructor options (`connectionString`, `schema`, `migrate`, `supervise`, `schedule`), `send` / `fetch` / `complete` / `fail`, `startAfter`, `deleteAfterSeconds` diff --git a/README.md b/README.md index 1770a3e..48eb793 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,11 @@ -# pgboss-queue +# pg-queue **Background jobs in Node.js, backed by Postgres.** -`pgboss-queue` is a queue-based job system with the same Worker / Scheduler / Queue API as [node-resque](https://github.com/actionhero/node-resque): priority queues, delayed jobs, plugins, locking, failed-job management, and a leader-elected scheduler. Storage is PostgreSQL via [pg-boss](https://github.com/timgit/pg-boss) (`SELECT … FOR UPDATE SKIP LOCKED`), not Redis. +`pg-queue` is a queue-based job system with the same Worker / Scheduler / Queue API as [node-resque](https://github.com/actionhero/node-resque): priority queues, delayed jobs, plugins, locking, failed-job management, and a leader-elected scheduler. Storage is PostgreSQL (`SELECT … FOR UPDATE SKIP LOCKED`), not Redis. ```ts -import { Queue, Worker, Scheduler, Plugins } from "pgboss-queue"; +import { Queue, Worker, Scheduler, Plugins } from "pg-queue"; const connection = { connectionString: process.env.DATABASE_URL, @@ -55,7 +55,7 @@ Pass a Postgres URL (not a Redis URL): ```ts const connection = { connectionString: "postgres://user:pass@host:5432/dbname", - schema: "pgboss_queue", // optional; default "pgboss_queue" + schema: "pg_queue", // optional; default "pg_queue" }; ``` @@ -77,7 +77,7 @@ const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); const connection = { pool }; ``` -`schema` isolates this library's tables (pg-boss's `job` table plus worker/lock/leader metadata) inside one Postgres database. It must be a legal SQL identifier (`letters`, `numbers`, `_`). +`schema` isolates this library's `pgrq_*` tables inside one Postgres database. It defaults to `pg_queue` and must be a legal SQL identifier (`letters`, `numbers`, `_`). Coming from node-resque: replace `{ host, port, password, database: 0 }` / `{ redis }` / `{ namespace: "resque" }` with `{ connectionString }` or `{ pool }` and `{ schema }`. @@ -142,7 +142,7 @@ Worker names must follow `hostname:pid` or `hostname:pid+unique_id` if you run m | Option | Default | Meaning | | --- | --- | --- | -| `automigrate` | `true` | Leader applies pg-boss schema migrations and metadata tables. Workers never migrate. | +| `automigrate` | `true` | Leader applies the bundled versioned SQL migrations. Workers never migrate. | | `completeJobRetentionMs` | `24 * 60 * 60 * 1000` | Leader deletes **completed** (and cancelled) jobs older than this. Failed jobs are kept until you retry or remove them. `false` disables the sweeper. `0` deletes completed jobs as soon as the leader sees them. | | `stuckWorkerTimeout` | 1 hour | If a worker has not pinged within this window, fail its in-flight job and remove it. Set `false` to disable. | | `leaderLockTimeout` | 180 seconds | Leader lock TTL; refreshed while the leader is alive. | @@ -237,7 +237,7 @@ schedule.scheduleJob("0 * * * * *", async () => { Jobs may list plugins that extend `Plugin`. Hooks: `beforeEnqueue`, `afterEnqueue`, `beforePerform`, `afterPerform`. `before*` hooks return `true` to continue or `false` to skip. ```ts -import { Plugin } from "pgboss-queue"; +import { Plugin } from "pg-queue"; class MyPlugin extends Plugin { async beforeEnqueue() { @@ -274,7 +274,7 @@ Inspect or delete plugin locks with `queue.locks()` and `queue.delLock(key)`. `MultiWorker` wraps `Worker` and scales the number of in-process workers from `minTaskProcessors` to `maxTaskProcessors` based on event-loop delay (more workers for I/O-bound jobs, fewer when the loop is blocked). ```ts -import { MultiWorker } from "pgboss-queue"; +import { MultiWorker } from "pg-queue"; const multiWorker = new MultiWorker( { @@ -306,9 +306,9 @@ Raise your Postgres pool `max` when using a large `maxTaskProcessors`. Events ma - PostgreSQL 13+ (`SKIP LOCKED`) ```bash -npm install pgboss-queue +npm install pg-queue # or -bun add pgboss-queue +bun add pg-queue ``` ## License diff --git a/__tests__/core/connection.test.ts b/__tests__/core/connection.test.ts index f72e8a7..c9bdbdc 100644 --- a/__tests__/core/connection.test.ts +++ b/__tests__/core/connection.test.ts @@ -20,7 +20,7 @@ describe("connection", () => { // Adapt: after cleanup, no job rows and no pgrq_* rows const pool = await specHelper.connect(); const jobCount = await pool.query<{ count: string }>( - `SELECT count(*)::text AS count FROM ${specHelper.schema}.job`, + `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_jobs`, ); const lockCount = await pool.query<{ count: string }>( `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_locks`, @@ -71,7 +71,7 @@ describe("connection", () => { }); test("keys built with a custom namespace are correct", async () => { - // Adapt: `schema` option sets pg-boss schema; migrate sees that schema + // Adapt: `schema` selects the isolated pg-queue schema const customSchema = "custom_namespace_test"; const custom = new Connection({ connectionString: process.env.DATABASE_URL, @@ -137,8 +137,7 @@ describe("connection", () => { // Skip: empty schema illegal; we reject }); - test("removes the redis event listeners when end", async () => { - // Adapt: pool / boss error listeners removed on end() + test("removes the postgres event listener when end", async () => { const pool = new Pool({ connectionString: process.env.DATABASE_URL }); const connection = new Connection({ pool, @@ -146,11 +145,8 @@ describe("connection", () => { }); await connection.connect(); expect(pool.listenerCount("error")).toBe(1); - expect(connection.boss.listenerCount("error")).toBe(1); - const boss = connection.boss; await connection.end(); expect(pool.listenerCount("error")).toBe(0); - expect(boss.listenerCount("error")).toBe(0); await pool.end(); }); @@ -165,6 +161,21 @@ describe("connection", () => { await connection.end(); }); + test("connect is idempotent and supports reconnect after end", async () => { + const connection = new Connection(specHelper.cleanConnectionDetails()); + await Promise.all([connection.connect(), connection.connect()]); + expect(connection.connected).toBe(true); + expect(connection.pool.listenerCount("error")).toBe(1); + + await connection.end(); + expect(connection.connected).toBe(false); + expect(() => connection.pool).toThrow("Connection is not connected"); + + await connection.connect(); + expect(connection.connected).toBe(true); + await connection.end(); + }); + test("connect with discrete host/port/user/password/database", async () => { const databaseUrl = process.env.DATABASE_URL; expect(databaseUrl).toBeDefined(); @@ -197,7 +208,7 @@ describe("connection", () => { }); test("reject illegal schema", () => { - expect(() => new Connection({ schema: "pgboss-queue" })).toThrow( + expect(() => new Connection({ schema: "pg-queue" })).toThrow( /Invalid schema/, ); expect(() => new Connection({ schema: "public; drop" })).toThrow( @@ -230,7 +241,7 @@ describe("connection", () => { ).toThrow(/database/); }); - test("migrate() creates pg-boss job table and pgrq_* tables", async () => { + test("migrate() creates all versioned pg-queue tables", async () => { const freshSchema = "pgrq_migrate_once"; const pool = await specHelper.connect(); await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); @@ -249,17 +260,35 @@ describe("connection", () => { ORDER BY table_name`, [ freshSchema, - ["job", "pgrq_leader", "pgrq_locks", "pgrq_stats", "pgrq_workers"], + [ + "pgrq_jobs", + "pgrq_leader", + "pgrq_locks", + "pgrq_migrations", + "pgrq_queues", + "pgrq_stats", + "pgrq_workers", + ], ], ); expect(tables.rows.map((row) => row.table_name)).toEqual([ - "job", + "pgrq_jobs", "pgrq_leader", "pgrq_locks", + "pgrq_migrations", + "pgrq_queues", "pgrq_stats", "pgrq_workers", ]); + const migrations = await connection.query<{ + version: number; + name: string; + }>( + `SELECT version, name FROM ${freshSchema}.pgrq_migrations ORDER BY version`, + ); + expect(migrations.rows).toEqual([{ version: 1, name: "initial" }]); + await connection.end(); await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); }); @@ -272,6 +301,92 @@ describe("connection", () => { await connection.end(); }); + test("concurrent migrate() calls serialize safely", async () => { + const freshSchema = "pgrq_migrate_concurrent"; + const pool = await specHelper.connect(); + await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); + const a = new Connection({ + connectionString: process.env.DATABASE_URL, + schema: freshSchema, + }); + const b = new Connection({ + connectionString: process.env.DATABASE_URL, + schema: freshSchema, + }); + + await Promise.all([a.migrate(), b.migrate()]); + const result = await a.query<{ count: string }>( + `SELECT count(*)::text AS count FROM ${freshSchema}.pgrq_migrations`, + ); + expect(Number(result.rows[0]?.count)).toBe(1); + + await Promise.all([a.end(), b.end()]); + await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); + }); + + test("migrate() can establish its own connection", async () => { + const freshSchema = "pgrq_migrate_connect"; + const pool = await specHelper.connect(); + await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); + const connection = new Connection({ + connectionString: process.env.DATABASE_URL, + schema: freshSchema, + }); + + await connection.migrate(); + expect(connection.connected).toBe(true); + expect( + ( + await connection.query<{ version: number }>( + `SELECT version FROM ${freshSchema}.pgrq_migrations`, + ) + ).rows, + ).toEqual([{ version: 1 }]); + + await connection.end(); + await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`); + }); + + test("fetchJob atomically claims ready jobs and skips delayed jobs", async () => { + await specHelper.cleanup(); + const connection = new Connection(specHelper.cleanConnectionDetails()); + await connection.connect(); + await connection.query( + `INSERT INTO ${specHelper.schema}.pgrq_queues (name) VALUES ('claims')`, + ); + await connection.query( + `INSERT INTO ${specHelper.schema}.pgrq_jobs + (name, data, priority, start_after) + VALUES + ('claims', '{"value":"low"}', 0, now()), + ('claims', '{"value":"high"}', 10, now()), + ('claims', '{"value":"later"}', 100, now() + interval '1 hour')`, + ); + + const [first, second] = await Promise.all([ + connection.fetchJob<{ value: string }>("claims"), + connection.fetchJob<{ value: string }>("claims"), + ]); + expect(new Set([first?.id, second?.id]).size).toBe(2); + expect([first?.data.value, second?.data.value].sort()).toEqual([ + "high", + "low", + ]); + expect(await connection.fetchJob("claims")).toBeNull(); + + const states = await connection.query<{ state: string; count: string }>( + `SELECT state, count(*)::text AS count + FROM ${specHelper.schema}.pgrq_jobs + GROUP BY state + ORDER BY state`, + ); + expect(states.rows).toEqual([ + { state: "active", count: "2" }, + { state: "created", count: "1" }, + ]); + await connection.end(); + }); + test("tryLeader: only one of two connections wins; after expiry the other wins", async () => { await specHelper.cleanup(); const a = new Connection(specHelper.cleanConnectionDetails()); diff --git a/__tests__/core/connectionError.test.ts b/__tests__/core/connectionError.test.ts index df5b7ee..1d413bf 100644 --- a/__tests__/core/connectionError.test.ts +++ b/__tests__/core/connectionError.test.ts @@ -7,10 +7,10 @@ describe("connection error", () => { const brokenConnection = new Connection({ host: "127.0.0.1", port: 1, - database: "pgboss_queue_test", + database: "pg_queue_test", user: "postgres", password: "postgres", - schema: "pgboss_queue_test", + schema: "pg_queue_test", }); let sawErrorEvent = false; @@ -32,6 +32,10 @@ describe("connection error", () => { /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|EAI_AGAIN|connect/i, ); expect(sawErrorEvent).toBe(true); + expect(brokenConnection.connected).toBe(false); + expect(() => brokenConnection.pool).toThrow( + "Connection is not connected", + ); resolve(); }); }); diff --git a/__tests__/core/queue.test.ts b/__tests__/core/queue.test.ts index 10865a4..a143071 100644 --- a/__tests__/core/queue.test.ts +++ b/__tests__/core/queue.test.ts @@ -26,10 +26,7 @@ async function seedActiveWorker( includeId = true, ): Promise { await target.enqueue(queueName, "slowJob", args); - const jobs = await target.connection.boss.fetch(queueName, { - batchSize: 1, - }); - const job = jobs[0]; + const job = await target.connection.fetchJob(queueName); if (!job) throw new Error("expected an active job"); await target.connection.query( `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on) @@ -232,7 +229,7 @@ describe("queue", () => { const ids = async () => { const result = await queue.connection.query<{ id: string }>( `SELECT id - FROM ${specHelper.schema}.job + FROM ${specHelper.schema}.pgrq_jobs WHERE name = $1 AND state = 'created' ORDER BY created_on, id`, [specHelper.queue], @@ -401,7 +398,7 @@ describe("queue", () => { for (let id = 1; id <= 3; id += 1) { await queue.enqueue("busted-queue", "busted_job", [id, 2, 3]); await queue.connection.query( - `UPDATE ${specHelper.schema}.job + `UPDATE ${specHelper.schema}.pgrq_jobs SET state = 'failed', completed_on = now() + make_interval(secs => $1), output = $2::jsonb @@ -562,7 +559,7 @@ describe("queue", () => { await client.query("BEGIN"); await client.query( `SELECT name - FROM ${specHelper.schema}.queue + FROM ${specHelper.schema}.pgrq_queues WHERE name = 'serialized' FOR UPDATE`, ); @@ -585,7 +582,7 @@ describe("queue", () => { test("does not drop jobs that remain after delQueue", async () => { await queue.enqueue("busy", "job", [1]); await queue.connection.query( - `UPDATE ${specHelper.schema}.job + `UPDATE ${specHelper.schema}.pgrq_jobs SET state = 'active' WHERE name = 'busy'`, ); @@ -597,12 +594,11 @@ describe("queue", () => { test("forceCleanWorker fails the original active job", async () => { expect(await queue.enqueue("stuck", "slowJob", [{ a: 1 }])).toBe(true); - const fetched = await queue.connection.boss.fetch<{ + const job = await queue.connection.fetchJob<{ class: string; queue: string; args: unknown[]; - }>("stuck", { batchSize: 1 }); - const job = fetched[0]; + }>("stuck"); if (!job) throw new Error("expected an active job"); await queue.connection.query( `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on) @@ -628,7 +624,7 @@ describe("queue", () => { const active = await queue.connection.query<{ count: string }>( `SELECT count(*)::text AS count - FROM ${specHelper.schema}.job + FROM ${specHelper.schema}.pgrq_jobs WHERE name = 'stuck' AND state = 'active'`, ); expect(Number(active.rows[0]?.count)).toBe(0); @@ -637,11 +633,18 @@ describe("queue", () => { test("forceCleanWorker without a job id fails only one matching active job", async () => { await queue.enqueue("twins", "slowJob", [1]); await queue.enqueue("twins", "slowJob", [1]); - const fetched = await queue.connection.boss.fetch<{ - class: string; - queue: string; - args: unknown[]; - }>("twins", { batchSize: 2 }); + const fetched = await Promise.all([ + queue.connection.fetchJob<{ + class: string; + queue: string; + args: unknown[]; + }>("twins"), + queue.connection.fetchJob<{ + class: string; + queue: string; + args: unknown[]; + }>("twins"), + ]); expect(fetched).toHaveLength(2); await queue.connection.query( `INSERT INTO ${specHelper.schema}.pgrq_workers (name, queues, working_on) @@ -660,7 +663,7 @@ describe("queue", () => { expect(await queue.failedCount()).toBe(1); const active = await queue.connection.query<{ count: string }>( `SELECT count(*)::text AS count - FROM ${specHelper.schema}.job + FROM ${specHelper.schema}.pgrq_jobs WHERE name = 'twins' AND state = 'active'`, ); expect(Number(active.rows[0]?.count)).toBe(1); diff --git a/__tests__/utils/specHelper.ts b/__tests__/utils/specHelper.ts index 25d9875..cd27e16 100644 --- a/__tests__/utils/specHelper.ts +++ b/__tests__/utils/specHelper.ts @@ -12,7 +12,7 @@ if (!connectionString) { ); } -export const schema = "pgboss_queue_test"; +export const schema = "pg_queue_test"; export const timeout = 500; export const queue = "default"; @@ -53,7 +53,7 @@ export async function disconnect(): Promise { } /** - * Install pg-boss + `pgrq_*` tables into the test schema (idempotent). + * Install all versioned pg-queue tables into the test schema (idempotent). */ export async function migrate(): Promise { const connection = new Connection(cleanConnectionDetails()); @@ -87,17 +87,25 @@ export async function cleanup(): Promise { AND table_name = ANY($2::text[])`, [ schema, - ["pgrq_leader", "pgrq_workers", "pgrq_locks", "pgrq_stats", "job"], + [ + "pgrq_jobs", + "pgrq_queues", + "pgrq_leader", + "pgrq_workers", + "pgrq_locks", + "pgrq_stats", + ], ], ); const names = new Set(tables.rows.map((row) => row.table_name)); - if (names.has("job")) { - await connection.query(`TRUNCATE TABLE ${schema}.job CASCADE`); + if (names.has("pgrq_jobs")) { + await connection.query(`TRUNCATE TABLE ${schema}.pgrq_jobs`); } const meta = [ + "pgrq_queues", "pgrq_leader", "pgrq_workers", "pgrq_locks", @@ -128,14 +136,13 @@ export async function popFromQueue(): Promise { const connection = new Connection(cleanConnectionDetails()); await connection.connect(); try { - const jobs = await connection.boss.fetch<{ + const job = await connection.fetchJob<{ class: string; queue: string; args: unknown[]; - }>(queue, { batchSize: 1 }); - const job = jobs[0]; + }>(queue); if (!job) return null; - await connection.boss.deleteJob(queue, job.id); + await connection.deleteJob(queue, job.id); return JSON.stringify(job.data); } finally { await connection.end(); diff --git a/bun.lock b/bun.lock index 311104c..52601e5 100644 --- a/bun.lock +++ b/bun.lock @@ -3,10 +3,9 @@ "configVersion": 1, "workspaces": { "": { - "name": "pgboss-queue", + "name": "pg-queue", "dependencies": { "pg": "^8.23.0", - "pg-boss": "^12.28.0", }, "devDependencies": { "@biomejs/biome": "^2.5.10", @@ -84,16 +83,8 @@ "bun-types": ["bun-types@1.4.0", "", { "dependencies": { "@types/node": "*" } }, "sha512-iIKw23BspnQQYd3prITOBxeUsxBHnwzX6YJfGMuNOZzeNcMmVqzIIVGRm1l69ogaPQmb4wB6BN8mA5bE9YuC5Q=="], - "cron-parser": ["cron-parser@5.10.0", "", { "dependencies": { "luxon": "^3.7.2" } }, "sha512-izNAxJyRWUP8ljBoDSub5WyrVOUlT4SLGShswE7eoRBpp6QUsSycYxLBMJlbshgPBMcPT/nrfgjNY2918ayv2A=="], - - "luxon": ["luxon@3.7.2", "", {}, "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew=="], - - "non-error": ["non-error@0.1.0", "", {}, "sha512-TMB1uHiGsHRGv1uYclfhivcnf0/PdFp2pNqRxXjncaAsjYMoisaQJI+SSZCqRq+VliwRTC8tsMQfmrWjDMhkPQ=="], - "pg": ["pg@8.23.0", "", { "dependencies": { "pg-connection-string": "^2.14.0", "pg-pool": "^3.14.0", "pg-protocol": "^1.16.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.4.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg=="], - "pg-boss": ["pg-boss@12.28.0", "", { "dependencies": { "cron-parser": "^5.10.0", "pg": "^8.23.0", "serialize-error": "^13.0.1" }, "bin": { "pg-boss": "dist/cli.js" } }, "sha512-7OaS/sYcQ8jcA9fSSlcdJc3Z9GD+/7GtQD3TlfweKpJQ/UnSbZYQHs1TECXMRY2sZo8ah4khQu9+6Z9v86z1dg=="], - "pg-cloudflare": ["pg-cloudflare@1.4.0", "", {}, "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A=="], "pg-connection-string": ["pg-connection-string@2.14.0", "", {}, "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg=="], @@ -116,14 +107,8 @@ "postgres-interval": ["postgres-interval@1.2.0", "", { "dependencies": { "xtend": "^4.0.0" } }, "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ=="], - "serialize-error": ["serialize-error@13.0.1", "", { "dependencies": { "non-error": "^0.1.0", "type-fest": "^5.4.1" } }, "sha512-bBZaRwLH9PN5HbLCjPId4dP5bNGEtumcErgOX952IsvOhVPrm3/AeK1y0UHA/QaPG701eg0yEnOKsCOC6X/kaA=="], - "split2": ["split2@4.2.0", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="], - "tagged-tag": ["tagged-tag@1.0.0", "", {}, "sha512-yEFYrVhod+hdNyx7g5Bnkkb0G6si8HJurOoOEgC8B/O0uXLHlaey/65KRv6cuWBNhBgHKAROVpc7QyYqE5gFng=="], - - "type-fest": ["type-fest@5.8.0", "", { "dependencies": { "tagged-tag": "^1.0.0" } }, "sha512-YGYEVz3Fm5iy/AybuA0oyNFq7H4CgQNfRp/qfe8nurE1kuCeNm3/vfm9X4Mtl+qLyaKJUh5xrFZwogr41SMjYA=="], - "typescript": ["typescript@7.0.2", "", { "optionalDependencies": { "@typescript/typescript-aix-ppc64": "7.0.2", "@typescript/typescript-darwin-arm64": "7.0.2", "@typescript/typescript-darwin-x64": "7.0.2", "@typescript/typescript-freebsd-arm64": "7.0.2", "@typescript/typescript-freebsd-x64": "7.0.2", "@typescript/typescript-linux-arm": "7.0.2", "@typescript/typescript-linux-arm64": "7.0.2", "@typescript/typescript-linux-loong64": "7.0.2", "@typescript/typescript-linux-mips64el": "7.0.2", "@typescript/typescript-linux-ppc64": "7.0.2", "@typescript/typescript-linux-riscv64": "7.0.2", "@typescript/typescript-linux-s390x": "7.0.2", "@typescript/typescript-linux-x64": "7.0.2", "@typescript/typescript-netbsd-arm64": "7.0.2", "@typescript/typescript-netbsd-x64": "7.0.2", "@typescript/typescript-openbsd-arm64": "7.0.2", "@typescript/typescript-openbsd-x64": "7.0.2", "@typescript/typescript-sunos-x64": "7.0.2", "@typescript/typescript-win32-arm64": "7.0.2", "@typescript/typescript-win32-x64": "7.0.2" }, "bin": { "tsc": "bin/tsc" } }, "sha512-8FYau96o3NKOhbjKi/qNvG/W5jhzxkbdm5sj9AbZ/5T5sWqn3hJgLfGx27sRKZWTvyzCP8dLRBTf5tBTSRVUNA=="], "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="], diff --git a/docs/plans/00-overview.md b/docs/plans/00-overview.md index 7adccb6..9a0442d 100644 --- a/docs/plans/00-overview.md +++ b/docs/plans/00-overview.md @@ -8,7 +8,7 @@ Define the product so later phases do not have to re-litigate it: a node-resque- ## Product -`pgboss-queue` is a library, not a framework. Callers construct `Queue`, `Worker`, and `Scheduler` (and optionally `MultiWorker`), pass a Postgres connection, register a `jobs` hash, and run. +`pg-queue` is a library, not a framework. Callers construct `Queue`, `Worker`, and `Scheduler` (and optionally `MultiWorker`), pass a Postgres connection, register a `jobs` hash, and run. It is the storage/runtime Keryx should eventually sit on top of, instead of either Redis node-resque or a one-off `PgBossBackend` that drops half the resque API. @@ -17,43 +17,36 @@ It is the storage/runtime Keryx should eventually sit on top of, instead of eith - **Queues** are conveyor belts: regular work, delayed work, and failed work. - **Workers** each run one job at a time. They pull from assigned queues (left-to-right = priority), succeed or write to failed, then pull again. - **Scheduler** is a specialized worker that does not run jobs. Many instances run; **one is leader**. The leader: - 1. Makes delayed jobs eligible when their time comes (node-resque: move from delayed keys onto the work list; we: rely on pg-boss `startAfter` and emit compatible events). + 1. Makes delayed jobs eligible when their `start_after` time arrives and emits compatible events. 2. Cleans stuck workers (heartbeat older than `stuckWorkerTimeout`). 3. **New:** runs schema automigrate when `automigrate: true`. 4. **New:** sweeps completed jobs older than `completeJobRetentionMs` (default 24h). Workers and schedulers are safe to run as many processes, many machines. Dequeue is exactly-once via `SELECT … FOR UPDATE SKIP LOCKED`. -## Why pg-boss, why not "just SQL" +## Why an owned PostgreSQL store -Keryx PR [#519](https://github.com/actionhero/keryx/pull/519) compared embeddable Postgres queues and chose pg-boss for: +Keryx PR [#519](https://github.com/actionhero/keryx/pull/519) established PostgreSQL and `SKIP LOCKED` as the right storage model. Initial phases used pg-boss as a store, but Phase 3 demonstrated that nearly every node-resque inspection and administration operation still required direct SQL while pg-boss's worker, scheduler, retry, and retention systems were disabled. -- Maturity and downloads -- Verified `SKIP LOCKED` dequeue -- Owned, versioned schema (`start()` / `migrate`) -- Delayed jobs (`startAfter`) -- Retry / fail states -- Optional `LISTEN/NOTIFY` later (not required for v1) +We therefore own a deliberately small, versioned schema: `pgrq_queues`, `pgrq_jobs`, and the existing metadata tables. This is not a general pg-boss replacement. It implements only the resque lifecycle we expose: enqueue, delayed eligibility, atomic claim, complete/fail/cancel, inspection, and leader-driven retention. -We wrap pg-boss rather than vendoring a job table because we do not want to maintain partition/index/migration machinery. We **do not** wrap pg-boss's `work()` as the public Worker: that helper registers one handler per queue and hides poll/plugin/event semantics. We use pg-boss as a **store**: `send`, `fetch`, `complete`, `fail`, `cancel`, plus SQL against `"schema".job` for introspection (the pattern `PgBossBackend` already used). +## Mapping: node-resque → pg-queue -## Mapping: node-resque → pgboss-queue - -| node-resque (Redis) | pgboss-queue (Postgres) | +| node-resque (Redis) | pg-queue (Postgres) | | --- | --- | | `connection.host/port/database/password` (Redis) | `connectionString` **or** `host/port/database/user/password/ssl` **or** `pool` | -| `namespace` / keyPrefix | `schema` (default `pgboss_queue`) | -| List `queue:{name}` (`RPUSH`/`LPOP`) | pg-boss `job` rows, `state IN ('created','retry')`, `start_after <= now()` | -| `delayed:{ts}` + `delayed_queue_schedule` zset | `job.start_after` in the future, same `name` (queue) | -| `failed` list | `job.state = 'failed'` (payload mapped to `ParsedFailedJobPayload`) | -| Lua `popAndStoreJob` | `boss.fetch` / `SKIP LOCKED`; worker row in **our** `workers` table | +| `namespace` / keyPrefix | `schema` (default `pg_queue`) | +| List `queue:{name}` (`RPUSH`/`LPOP`) | `pgrq_jobs`, `state IN ('created','retry')`, `start_after <= now()` | +| `delayed:{ts}` + `delayed_queue_schedule` zset | `pgrq_jobs.start_after` in the future | +| `failed` list | `pgrq_jobs.state = 'failed'` (payload mapped to `ParsedFailedJobPayload`) | +| Lua `popAndStoreJob` | Atomic update selected with `FOR UPDATE SKIP LOCKED`; worker row in `pgrq_workers` | | `SET NX EX` leader lock | `leader` row with `expires_at` (same NX/expiry semantics) | | `worker:ping:{name}` | `workers.ping_at` | | `lock:*` / `workerslock:*` | `locks` table (`key`, `expires_at`) | | `stat:processed` / `stat:failed` | `stats` table | -| `smembers queues` | pg-boss `getQueues()` plus any queue we created | +| `smembers queues` | `pgrq_queues` | | Scheduler promotes delayed → list | Job becomes fetchable when `start_after <= now()`; leader still polls and emits `workingTimestamp` / `transferredJob` for compatibility | -| pg-boss `supervise` deleting jobs | **Off** on workers. Leader sweeper deletes `completed`/`cancelled` older than retention. **Failed jobs are kept** until `removeFailed` / retry (resque behavior). | +| Redis cleanup | Leader sweeper deletes `completed`/`cancelled` older than retention. **Failed jobs are kept** until `removeFailed` / retry. | ### Classes (public, names frozen) @@ -80,18 +73,18 @@ Same as `node-resque/src/index.ts`: ## Lessons from keryx#519 (use) 1. **Connection strings, not Redis hashes.** `config.database.connectionString` / `new PgBoss({ connectionString, schema })`. -2. **Schema isolation.** Default `keryx_tasks` there; we default `pgboss_queue`. Validate identifier safety (`^[a-zA-Z_][a-zA-Z0-9_]*$`). -3. **SQL introspection.** `queued`, `del`, `delDelayed`, `scheduledAt`, `failed*` are parameterized SQL on `"schema".job`, not missing pg-boss API methods. -4. **Payload shape.** Store `{ class, queue, args }` (resque encode) inside pg-boss `data`. Keryx used `_actionName` + inputs object because actions aren't resque jobs. We store the node-resque JSON: `{ class, queue, args }`. -5. **createQueue lazily.** Track `knownQueues`; `createQueue` on first enqueue. Optional `partition` is out of scope for v1. -6. **Workers vs CLI.** pg-boss `supervise` / `schedule` should be false on processes that only enqueue. Our Scheduler leader is the one process allowed to mutate schema and delete old rows. -7. **Recurring uniqueness.** pg-boss `short` policy + `singletonKey` = one *pending* job. We do not need this for v1 core (node-resque has no built-in CRON), but expose `singletonKey` / document `short` queues as an escape hatch and as the path Keryx will use. Leader + `scheduler.leader` remains the node-resque way to run CRON (see `examples/scheduledJobs.ts`). +2. **Schema isolation.** Default `keryx_tasks` there; we default `pg_queue`. Validate identifier safety (`^[a-zA-Z_][a-zA-Z0-9_]*$`). +3. **SQL introspection.** `queued`, `del`, `delDelayed`, `scheduledAt`, and `failed*` are parameterized SQL on `pgrq_jobs`. +4. **Payload shape.** Store `{ class, queue, args }` (resque encode) in `data`. +5. **Create queues lazily.** Insert `pgrq_queues` with `ON CONFLICT DO NOTHING` before enqueue. +6. **One maintenance owner.** The Scheduler leader is the one process allowed to migrate and delete old rows. +7. **Recurring uniqueness.** It is not part of node-resque v1. Leader + `scheduler.leader` remains the supported way to gate CRON. 8. **Retention.** keryx used `deleteAfterSeconds` default 7 days. We default **24 hours** for *completed* jobs, leader-driven, not pg-boss supervise on every instance. -9. **RetryLimit 0 by default.** node-resque does not retry unless the Retry plugin is attached. Do not enable pg-boss retries globally or we will double-retry with the plugin. +9. **No store-level retry.** node-resque does not retry unless the Retry plugin is attached. ## Lessons from keryx#519 (do not copy) -- Replacing `Worker` with `boss.work(queue, { localConcurrency })` — loses one-job-at-a-time-per-Worker, plugin `beforePerform`, queue priority walk, and events. +- Replacing `Worker` with a generic handler registration — loses one-job-at-a-time-per-Worker, plugin `beforePerform`, queue priority walk, and events. - No leader — loses `queue.leader()`, CRON gating, single migrator, single sweeper. - Dropping `locks`, `delLock`, `timestamps`, `delayedAt`, `allDelayed`, `workingOn`, `cleanOldWorkers`, `delByFunction`, `delQueue` — those are in the node-resque test suite and in resque-admin. - Removing the plugin system. @@ -136,16 +129,16 @@ Also: ### 2. `automigrate: boolean` (default `true`) -Only the **elected scheduler** applies migrations (pg-boss `migrate` + our metadata DDL). Workers and extra schedulers start with `migrate: false`. If no scheduler is running, tests/scripts call `Connection.migrate()` explicitly (specHelper will). Production docs: run ≥1 scheduler with `automigrate: true`. +Only the **elected scheduler** applies bundled, versioned migrations. If no scheduler is running, tests/deploy scripts call `Connection.migrate()` explicitly. Production docs: run ≥1 scheduler with `automigrate: true`. ### 3. `completeJobRetentionMs` (default `24 * 60 * 60 * 1000`) -Leader sweeper deletes pg-boss jobs in `completed` or `cancelled` whose completion timestamp is older than this. Failed jobs are not auto-deleted. `0` means delete completed jobs as soon as the sweeper sees them. `false` / `Infinity` disables the sweeper. +Leader sweeper deletes `pgrq_jobs` in `completed` or `cancelled` whose completion timestamp is older than this. Failed jobs are not auto-deleted. `0` means delete completed jobs as soon as the sweeper sees them. `false` / `Infinity` disables the sweeper. ## Recommended repo layout (Phase 1 creates this) ``` -pgboss-queue/ +pg-queue/ src/ index.ts core/ connection, queue, worker, scheduler, multiWorker, plugin, pluginRunner @@ -160,7 +153,7 @@ pgboss-queue/ ## Non-goals (v1) - Compatible wire protocol with Ruby Resque / Sidekiq (node-resque aimed at that via Redis keys; we will not write Redis keys) -- Cockroach / PGLite backends (pg-boss supports them; we test Postgres only) +- Cockroach / PGLite backends (we test PostgreSQL only) - Built-in HTTP dashboard - Exactly matching Redis performance characteristics - Wrapping graphile-worker or PGMQ @@ -171,4 +164,4 @@ Phase 8's matrix is green: every **relevant** node-resque test exists under the ## Lessons learned -_None yet._ +- 2026-08-29: Phase 3 showed pg-boss was only supplying schema migrations and a handful of job state methods; node-resque compatibility still required direct SQL for most Queue behavior while pg-boss workers, scheduling, retries, and supervision were disabled. We replaced it before Worker landed with a focused, owned schema and atomic `SKIP LOCKED` claim. diff --git a/docs/plans/01-repo-scaffold.md b/docs/plans/01-repo-scaffold.md index 2d02d33..6b9e99e 100644 --- a/docs/plans/01-repo-scaffold.md +++ b/docs/plans/01-repo-scaffold.md @@ -18,7 +18,7 @@ CI must exist **before** Queue/Worker code. An empty `expect(true)` that never o ### Package - `package.json` - - `name`: `pgboss-queue` + - `name`: `pg-queue` - `version`: `0.0.1` - `type`: `"module"` - `main` / `types`: `dist/index.js` / `dist/index.d.ts` @@ -27,13 +27,13 @@ CI must exist **before** Queue/Worker code. An empty `expect(true)` that never o - `license`: `Apache-2.0` - `scripts`: `"test": "bun test --max-concurrency=1"`, `"test:node-package": "node scripts/assert-node-package.mjs"`, `build`, `lint`, `format` (docs scripts wait for Phase 9) - `devDependencies`: `typescript`, `@types/node`, `@types/pg`, `@biomejs/biome`, `@types/bun` - - `dependencies`: `pg` now (smoke test uses it). Add `pg-boss` in Phase 2 if you want to keep this PR smaller — either is fine as long as CI is green. + - `dependencies`: `pg` (smoke test and queue storage use it). - `tsconfig.json` — `strict`, `noImplicitAny: true`, `ES2022`, `moduleResolution: bundler` or `nodenext`, `declaration`, `outDir: dist`, `rootDir: src`. `tsconfig.test.json` typechecks `__tests__` with `noEmit` (so implicit `any` in tests fails `build` too). - `biome.json` — match keryx reasonably (indent 2, no unused imports). `suspicious/noExplicitAny` is `error` so `: any` and `as any` fail `lint`. - `.gitignore` — `node_modules`, `dist`, `.env`, `docs/.vitepress/dist`, `*.log` - `LICENSE` — Apache-2.0 - `.nvmrc` or `.node-version` — `26` -- `.env.example` — `DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test` +- `.env.example` — `DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test` ### Source stub @@ -51,7 +51,7 @@ No `docker-compose.yml`. CI starts Postgres as a GitHub Actions service. Locally `.env.example` is the only local-DB contract: ``` -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test ``` ### Test harness (not a dummy assert) @@ -59,7 +59,7 @@ DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test `__tests__/utils/specHelper.ts` — the same helper later phases grow. In this phase it must: - Read `DATABASE_URL` (fail the suite with a clear message if unset) -- Export `connectionDetails`, `timeout` (e.g. 500), `queue` (a default queue name), `schema` (default `pgboss_queue_test`) +- Export `connectionDetails`, `timeout` (e.g. 500), `queue` (a default queue name), `schema` (default `pg_queue_test`) - `connect()` / `disconnect()` against `pg.Pool` - `cleanup()` — no-op or `SELECT 1` until Phase 2 adds truncate/migrate - `popFromQueue()` — throw `"not implemented"` until Phase 3 (do not silently return null) @@ -76,7 +76,7 @@ Do **not** ship only `expect(true).toBe(true)`. Tests use `bun:test`. Node compa `.github/workflows/test.yaml` is the product gate from this PR onward. Jobs: `lint`, `build`, `test` (Postgres 16 service, **Bun `bun:test`**), `node-package` (Node 26 imports the compiled package), `complete`. -- `test`: `bun run test` with `DATABASE_URL=postgres://postgres:postgres@localhost:5432/pgboss_queue_test` +- `test`: `bun run test` with `DATABASE_URL=postgres://postgres:postgres@localhost:5432/pg_queue_test` - `node-package`: `actions/setup-node` from `.nvmrc` (26), `bun run build`, then **`node scripts/assert-node-package.mjs`** (not `bun run`; no Postgres) See the workflow file for the YAML. Do not duplicate a second test pipeline in Phase 10. @@ -120,3 +120,4 @@ A compiling package, a shared `specHelper`, and CI that will run every subsequen - 2026-08-26: `noImplicitAny` is set on both `tsconfig.json` and `tsconfig.test.json` so relaxing `strict` later still bans implicit `any` in src and tests. - 2026-08-26: Required checks on `main` should be `complete` and `Cursor Bugbot`. The cloud agent GitHub token is not a repo admin (403 on branch protection and rulesets), so a maintainer must set that in the GitHub UI. - 2026-08-26: Phase 2 filled `specHelper.cleanup()` / `migrate()` / `dropSchema()` and added `pg-boss`. Smoke test remains valid; connection suite uses `*.test.ts` filenames because Bun will not discover bare `connection.ts`. +- 2026-08-29: The project was renamed to `pg-queue` and pg-boss was removed. `pg` is now the only runtime dependency; bundled SQL migrations are included in package files. diff --git a/docs/plans/02-connection-and-schema.md b/docs/plans/02-connection-and-schema.md index acf99c7..e1cc3ed 100644 --- a/docs/plans/02-connection-and-schema.md +++ b/docs/plans/02-connection-and-schema.md @@ -5,7 +5,7 @@ ## Goal -Callers can connect to Postgres the way they connected to Redis in node-resque, and the library can install (1) pg-boss's schema and (2) our metadata tables. Migration is a function the **scheduler leader** will invoke; this phase only implements the primitive. +Callers can connect to Postgres the way they connected to Redis in node-resque, and the library can install our versioned queue and metadata schema. Migration is a function the **scheduler leader** will invoke; this phase implements the primitive. ## Connection options @@ -27,11 +27,11 @@ export interface ConnectionOptions { */ pool?: import("pg").Pool; /** - * pg-boss schema AND our metadata schema. Default `pgboss_queue`. + * Queue and metadata schema. Default `pg_queue`. * Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (reject otherwise). */ schema?: string; - application_name?: string; // default `pgboss-queue` + application_name?: string; // default `pg-queue` } export interface QueueOptions { @@ -53,7 +53,7 @@ export interface SchedulerOptions extends QueueOptions { leaderLockTimeout?: number; // seconds, default 180 stuckWorkerTimeout?: number | false; retryStuckJobs?: boolean; - /** Leader runs pg-boss migrate + metadata DDL. Default true. */ + /** Leader runs bundled versioned migrations. Default true. */ automigrate?: boolean; /** Leader deletes completed/cancelled jobs older than this. Default 24h. `false` disables. */ completeJobRetentionMs?: number | false; @@ -71,7 +71,7 @@ export interface MultiWorkerOptions extends WorkerOptions { ### Mapping help (document in JSDoc + README) -| node-resque | pgboss-queue | +| node-resque | pg-queue | | --- | --- | | `{ host, port, password, database: 0 }` | `{ connectionString }` or `{ host, port, user, password, database: "myapp" }` | | `{ redis: ioredis }` | `{ pool: pg.Pool }` | @@ -84,15 +84,14 @@ Runtime rejection: `pkg`, `redis`, and numeric `database` throw from the `Connec Port `src/core/connection.ts` *behavior*, not Redis: -- `connect()` — create pool unless `pool` was provided; always wrap the pool as pg-boss `db.executeSql` so `connection.pool` is a real `pg.Pool`. Construct `PgBoss` with `{ db, schema, migrate: false, supervise: false, schedule: false, application_name }`. Call `boss.start()` when the schema is installed; if pg-boss reports "not installed", leave the pool connected so `migrate()` can run, then start boss after migrate. -- `end()` — `boss.stop({ graceful: true, close: false })` (we own / borrow the pool); `pool.end()` only if we created the pool; remove forwarded `error` listeners. +- `connect()` — create a pool unless `pool` was provided and verify connectivity with `SELECT 1`. +- `end()` — `pool.end()` only if we created the pool; remove forwarded `error` listeners. - `connected` boolean -- Event `error` forwarded from pool and pg-boss +- Event `error` forwarded from the pool - `key(...parts)` — **keep** as a helper for lock key strings (`["lock", func, queue, args].join(":")`) stored in `locks.key`. Do not prefix Redis-style. Tests that assert `resque-test-0:thing` are Redis-only (Phase 8 skip). Expose: -- `connection.boss: PgBoss` - `connection.pool: pg.Pool` (owned or provided) - `connection.schema: string` - `connection.query(text, values)` — parameterized; schema identifiers are validated once and interpolated only after `assertSchema` @@ -101,20 +100,40 @@ Expose: async migrate(): Promise ``` -`migrate()` is idempotent: +`migrate()` is idempotent and concurrency-safe: -1. `CREATE SCHEMA IF NOT EXISTS {schema}` -2. Short-lived `PgBoss` with `{ db, schema, migrate: true, supervise: false, schedule: false }` → `start()` / `stop({ close: false })` -3. Apply our metadata DDL (`CREATE TABLE IF NOT EXISTS` + indexes) -4. `boss.start()` on the long-lived instance if it was waiting on install +1. Begin a transaction and acquire a transaction-scoped advisory lock for the schema. +2. `CREATE SCHEMA IF NOT EXISTS {schema}` and create `pgrq_migrations`. +3. Apply each pending numbered script from `migrations/` in order. +4. Record the version and commit atomically; rollback leaves no partial migration. Workers never call this. Scheduler leader will. `specHelper.migrate()` calls it in `beforeAll`. ## Metadata DDL (ours) -All in `schema` (same as pg-boss), tables prefixed `pgrq_` so they never collide with pg-boss's `job`, `queue`, `schedule`, `version`, etc. +All tables are in `schema` and prefixed `pgrq_`. `migrations/001_initial.sql` creates the job, queue, and metadata tables. ```sql +CREATE TABLE {schema}.pgrq_queues ( + name text PRIMARY KEY, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE {schema}.pgrq_jobs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + name text NOT NULL REFERENCES {schema}.pgrq_queues(name), + data jsonb NOT NULL, + state text NOT NULL CHECK (state IN ( + 'created', 'retry', 'active', 'completed', 'cancelled', 'failed' + )), + priority integer NOT NULL DEFAULT 0, + created_on timestamptz NOT NULL DEFAULT now(), + start_after timestamptz NOT NULL DEFAULT now(), + started_on timestamptz, + completed_on timestamptz, + output jsonb +); + -- Leader election (Redis SET NX EX analogue) CREATE TABLE IF NOT EXISTS {schema}.pgrq_leader ( slot text PRIMARY KEY DEFAULT 'default', @@ -177,17 +196,11 @@ currentLeader(): Promise `tryLeader` uses a single transaction. This is the Redis `SET NX EX` + refresh-if-mine pattern from `scheduler.tryForLeader`. -## pg-boss constructor flags (every instance) - -| Flag | Worker / Queue | Scheduler (non-leader) | Scheduler (leader) | -| --- | --- | --- | --- | -| `migrate` | false | false | true iff `automigrate` | -| `supervise` | false | false | false (we sweep) | -| `schedule` | false | false | false (no pg-boss cron) | +## Job-store primitives -We do not want two maintenance systems. pg-boss's built-in delete/archive would race our retention policy and might drop failed jobs. +`Connection.fetchJob(queue)` atomically claims one ready job by selecting in priority/FIFO order with `FOR UPDATE SKIP LOCKED` and changing its state to `active`. `deleteJob(queue, id)` removes a claimed job. Worker adds completion/failure transitions in Phase 4. -Dependency: `pg-boss` (installed in this phase; currently `^12`). +The only runtime dependency is `pg`. ## Tests (this phase) @@ -200,10 +213,11 @@ Port-inspired plus the Adapt rows from Phase 8: - connect with `connectionString` - connect with discrete `host/port/user/password/database` - connect with shared `pool` (ending Connection does not end the pool) -- reject illegal `schema` (`pgboss-queue`, `public; drop`, empty) +- reject illegal `schema` (`pg-queue`, `public; drop`, empty) - reject Redis options (`pkg`, `redis`, numeric `database`) -- `migrate()` creates pg-boss `job` table and `pgrq_*` tables +- `migrate()` creates `pgrq_migrations`, `pgrq_queues`, `pgrq_jobs`, and metadata tables - second `migrate()` is a no-op +- concurrent `migrate()` calls serialize and apply each version once - `tryLeader` : only one of two connections wins; after expiry the other wins - `setLockNx` / expire / `delLock` (+ stats smoke) - connectionError (bad host / port `127.0.0.1:1`) @@ -232,4 +246,6 @@ Do not defer these to Phase 8. - 2026-08-26: pg-boss is a named ESM export (`import { PgBoss } from "pg-boss"`), not a default export. Migrator instances use `migrate: true` + `supervise: false` + `schedule: false`. - 2026-08-26: Version bumped `0.0.1` → `0.1.0` (first user-facing API: `Connection`). - 2026-08-26: Node ESM (`"type": "module"`) requires relative import specifiers with `.js` extensions in emitted `dist/` (e.g. `from "./core/connection.js"`). Without them, `node scripts/assert-node-package.mjs` fails with `ERR_MODULE_NOT_FOUND` even though `tsc` and Bun tests pass. +- 2026-08-29: Removed pg-boss after Phase 3 confirmed that its lifecycle and schema added coupling without supplying the resque runtime. Migrations now ship as numbered SQL files, execute under an advisory lock in one transaction, and are included in the npm package. +- 2026-08-29: `connect()` must execute `SELECT 1`; constructing `pg.Pool` is lazy and does not prove credentials, routing, or database availability. - 2026-08-29: Phase 3 restored node-resque's optional `QueueOptions.queue` field. Queue methods still take an explicit queue name, but retaining the constructor field lets existing typed call sites migrate without an excess-property error. diff --git a/docs/plans/03-queue.md b/docs/plans/03-queue.md index 1527af3..e0b94c8 100644 --- a/docs/plans/03-queue.md +++ b/docs/plans/03-queue.md @@ -5,7 +5,7 @@ ## Goal -Implement `Queue` so programs can enqueue work, inspect it, delete it, and manage failures — the node-resque Queue methods, backed by pg-boss `job` rows plus our metadata tables. +Implement `Queue` so programs can enqueue work, inspect it, delete it, and manage failures — the node-resque Queue methods, backed by `pgrq_jobs` plus our metadata tables. Workers are not required yet. Tests enqueue and SQL-inspect (or use a test helper `popFromQueue` that `fetch`es a job). @@ -17,19 +17,14 @@ node-resque: encode(q, func, args) => JSON.stringify({ class: func, queue: q, args }) ``` -Store that object as pg-boss `data`. Queue name is pg-boss job `name` (one pg-boss queue per resque queue). +Store that object as `pgrq_jobs.data`. Queue name is `pgrq_jobs.name`. ```ts -await boss.send(q, { class: func, queue: q, args }, { - retryLimit: 0, // Retry plugin owns retries - startAfter?: Date, // enqueueAt / enqueueIn -}); +INSERT INTO pgrq_jobs (name, data, start_after) +VALUES (q, { class: func, queue: q, args }, startAfter); ``` -`createQueue(q)` before first send if needed (keryx `ensureQueue`). Default queue options: - -- `retryLimit: 0` -- `deleteAfterSeconds`: large (e.g. 30 days) so pg-boss will not delete out from under us; **our sweeper** is the real retention (Phase 5). Alternatively omit delete policy if pg-boss leaves rows until we `deleteJob`. +Insert `pgrq_queues` with `ON CONFLICT DO NOTHING` before the job insert. The leader sweeper is the sole retention mechanism (Phase 5). ## Methods (port `src/core/queue.ts`) @@ -43,7 +38,7 @@ Implement every public method. Behavior notes where Postgres differs: ### Queue admin -- `queues()` — union of pg-boss `getQueues()` names and distinct `job.name`. +- `queues()` — names from `pgrq_queues`. - `delQueue(q)` — delete all jobs in that queue (any state except maybe `active` — document: active jobs are not deleted; match "delete the list" as best-effort). Drop from known queues. - `length(q)` — count `created`/`retry` with `start_after <= now()` (ready, not delayed). - `queued(q, start, stop)` — same filter, `ORDER BY created_on`, offset/limit. Map to `ParsedJob`. @@ -64,7 +59,7 @@ Treat delayed as `state IN ('created','retry') AND start_after > now()`. - `locks()` — all `pgrq_locks` rows (strip expired). Keys should look like `lock:…` / `workerslock:…` so plugin tests pass. - `delLock(key)` -- `stats()` — `pgrq_stats` plus optionally pg-boss state counts under extra keys (keep `processed` / `failed` names). +- `stats()` — `pgrq_stats` (keep `processed` / `failed` names). - `leader()` / `leaderKey()` — `pgrq_leader.name` (leaderKey can return the slot name for tests that only check non-empty). ### Workers (tables filled in Phase 4; methods exist now) @@ -80,7 +75,7 @@ Return types must match `ParsedWorkerPayload` / `ErrorPayload`. ### Failed jobs -pg-boss `failed` rows → `ParsedFailedJobPayload`: +Failed `pgrq_jobs` rows → `ParsedFailedJobPayload`: ```ts { @@ -99,7 +94,7 @@ node-resque `failed(start,stop)` uses list indices; we `ORDER BY completed_on` o - `failedCount()` - `failed(start, stop)` - `removeFailed(failedJob)` — delete that row. Matching: prefer `failedJob` identity. If tests pass the whole payload, match on `payload` + `failed_at` or store `id` on our mapped object as an extra enumerable field (keryx added `id`). Adding `id?: string` to the payload type is allowed if tests still pass; include it. -- `retryAndRemoveFailed(failedJob)` — `boss.retry(queue, id)` or re-`enqueue` + delete. Throw `This job is not in failed queue` if nothing matched. +- `retryAndRemoveFailed(failedJob)` — re-`enqueue` + delete. Throw `This job is not in failed queue` if nothing matched. ## Plugin runner @@ -116,7 +111,7 @@ Skip only tests that poke Redis keys directly if any remain inside queue.ts (the ## Acceptance criteria - All Queue methods exist with JSDoc copied/adapted from node-resque -- `__tests__/core/queue.test.ts` is green. Worker-status methods are tested with seeded `pgrq_workers` and active pg-boss jobs; Phase 4 will additionally exercise them through a live Worker. +- `__tests__/core/queue.test.ts` is green. Worker-status methods are tested with seeded `pgrq_workers` and active jobs; Phase 4 will additionally exercise them through a live Worker. Recommended split: @@ -143,3 +138,4 @@ Recommended split: - 2026-08-29: Bugbot: delayed duplicate lock keys embedded the encoded JSON and overflowed the `pgrq_locks` btree for large payloads. Keys now use `sha256(encoded)` plus the timestamp second. - 2026-08-29: Bugbot: `delQueue` rebuilt those keys from jsonb-loaded args, whose object key order can differ from `JSON.stringify` at enqueue time. The hash now canonicalizes nested object keys so delete and re-enqueue agree. - 2026-08-29: Coverage audit found untested `del(count)`, `delByFunction(start, stop)`, expired-lock cleanup, concurrent delayed enqueue, queue-row serialization, `cleanOldWorkers`, `retryStuckJobs`, active `workingOn`, and unknown-worker errors. These now have focused PostgreSQL tests rather than being deferred wholesale to Phase 4. +- 2026-08-29: Removed pg-boss before Worker implementation. Queue registration is now an idempotent `pgrq_queues` insert, enqueue writes `pgrq_jobs`, and test dequeues use the same atomic `Connection.fetchJob()` primitive planned for Worker. diff --git a/docs/plans/04-worker.md b/docs/plans/04-worker.md index 335429d..fc68bd5 100644 --- a/docs/plans/04-worker.md +++ b/docs/plans/04-worker.md @@ -50,20 +50,20 @@ Write `pgrq_workers`: - After a full empty pass, `pause()` (emit `pause`, `setTimeout(timeout)`, poll again). - When using `"*"`, re-`checkQueues()` at the end of a pass so new queues appear (test: `will notice new job queues when started with queues=*`). -Do **not** rely on pg-boss `work()` localConcurrency to implement this. Walk queues in JS. +Walk queues in JS so array order remains queue priority. ### `getJob` (dequeue) For the current queue name: ```ts -const [job] = await boss.fetch(queue, { batchSize: 1 }); +const job = await connection.fetchJob(queue); ``` -If pg-boss fetch API differs in the pinned version, use equivalent SQL: +`Connection.fetchJob()` uses equivalent SQL: ```sql -SELECT id, name, data FROM {schema}.job +SELECT id, name, data FROM {schema}.pgrq_jobs WHERE name = $1 AND state = 'created' -- or 'retry' AND start_after <= now() @@ -72,7 +72,7 @@ FOR UPDATE SKIP LOCKED LIMIT 1 ``` -then mark `active`. Prefer the public `fetch` API. +The implementation wraps this selection in a CTE that updates the selected row to `active` and returns it atomically. On fetch: set `pgrq_workers.working_on`. Return `data` as `ParsedJob`. @@ -82,9 +82,9 @@ Exactly-once: two workers must never receive the same `id`. Add a test: enqueue Port plugin `beforePerform` / `afterPerform`, frozen args, missing job class → failure `"No job defined for class …"`. Emit `job` before perform. `completeJob` → `succeed` or `fail`. Duration in ms on success/failure events. -**succeed:** `boss.complete(id)`, incr `processed` + `processed:{workerName}`, emit `success`. +**succeed:** update `pgrq_jobs` to `completed`, incr `processed` + `processed:{workerName}`, emit `success`. -**fail:** `boss.fail(id, error)` (so row is `failed`), incr `failed` counters, emit `failure`. Store output so `queue.failed()` can rebuild the resque error payload. If pg-boss `fail` output is the error object, that is enough. +**fail:** update `pgrq_jobs` to `failed` and store the error payload in `output`, incr `failed` counters, emit `failure`. Clear `working_on`. If `looping`, poll again. @@ -135,4 +135,4 @@ Worker rows, pings, `forceCleanWorker`, fetch/complete/fail. ## Lessons learned -_None yet._ +- 2026-08-29: The job store was brought in-house before this phase. Worker must use `Connection.fetchJob()` for the atomic `SKIP LOCKED` claim and explicit state transitions on `pgrq_jobs`. diff --git a/docs/plans/05-scheduler.md b/docs/plans/05-scheduler.md index 5d82cc3..298b7f4 100644 --- a/docs/plans/05-scheduler.md +++ b/docs/plans/05-scheduler.md @@ -50,7 +50,7 @@ Non-leaders skip 1, 3–5. ### Delayed jobs -pg-boss already hides future `start_after` from `fetch`. Workers will pick them up without a transfer. For **API and event compatibility**: +`Connection.fetchJob()` hides future `start_after` values. Workers pick jobs up without a transfer. For **API and event compatibility**: On leader poll, select delayed jobs with `start_after <= now()` that have not yet been "announced" **or** simply: @@ -107,7 +107,7 @@ WHERE state IN ('completed', 'cancelled') RETURNING id ``` -Use pg-boss `deleteJob` if that is required for partition integrity; otherwise SQL delete is what keryx used for management. +Delete retained rows directly from `pgrq_jobs`; there is no second maintenance system. - Default retention 24h - Do **not** delete `failed` or `active` or `created` @@ -161,4 +161,4 @@ Stable leader + worker cleaning. Plugins can land in parallel with leftover sche ## Lessons learned -_None yet._ +- 2026-08-29: Scheduler migration and retention now operate solely on the owned migration ledger and `pgrq_jobs`; there is no pg-boss supervisor or partition lifecycle to coordinate with. diff --git a/docs/plans/06-plugins.md b/docs/plans/06-plugins.md index 7218f93..dd5206f 100644 --- a/docs/plans/06-plugins.md +++ b/docs/plans/06-plugins.md @@ -46,7 +46,7 @@ If the same name+queue+args is already in **delayed**, skip enqueue. On failure: increment attempt, if remaining, `enqueueIn` with `retryDelay` / `backoffStrategy`, emit `reEnqueue`, clear `worker.error` so the job is not placed in failed, decr processed / incr failed stats (port exactly). After limit, cleanup keys and throw. Default `retryLimit: 1`, `retryDelay: 5000`. -Do **not** also set pg-boss `retryLimit` > 0 on these jobs. +The store performs no automatic retry; the plugin is the only retry owner. ### `Noop` @@ -80,7 +80,7 @@ Also port `__tests__/core/queue.ts` `describe("locks")` if not already green. - All five plugins exported as `Plugins.JobLock` etc. (port `src/plugins/index.ts`) - Plugin tests green - `queue.locks()` / `delLock()` work -- Retry does not double-retry with pg-boss +- Retry does not double-retry at the storage layer - **CI green** on this PR ## Next @@ -89,4 +89,4 @@ Phase 7 is independent. Phase 8 includes these tests in the matrix. ## Lessons learned -_None yet._ +- 2026-08-29: The owned job store has no automatic retry behavior, preserving the Retry plugin as the single source of retry policy. diff --git a/docs/plans/07-multiworker.md b/docs/plans/07-multiworker.md index 2075e90..62f8c2a 100644 --- a/docs/plans/07-multiworker.md +++ b/docs/plans/07-multiworker.md @@ -48,10 +48,11 @@ Port `examples/multiWorker.ts` to Postgres connection details. ## Note -pg-boss `localConcurrency` is **not** a substitute. MultiWorker must spawn real `Worker` objects so plugins, names, and heartbeats stay per-worker. +MultiWorker must spawn real `Worker` objects so plugins, names, and heartbeats stay per-worker. ## Lessons learned - 2026-08-26: Phase 1 runs tests with `node:test`, not `bun:test`. Do not assume Bun-only retry APIs when this phase is implemented. - 2026-08-26: Phase 1 reverted to `bun:test`. Bun retry APIs are available again; Node is only the compiled-package import check. +- 2026-08-29: Removing pg-boss does not change MultiWorker semantics; concurrency remains a pool of real resque-compatible Worker instances. diff --git a/docs/plans/08-conformance-tests.md b/docs/plans/08-conformance-tests.md index 1e6885e..3b38476 100644 --- a/docs/plans/08-conformance-tests.md +++ b/docs/plans/08-conformance-tests.md @@ -27,7 +27,7 @@ If a row above is missing when you start this phase, that is a **bug in an earli ## Tooling (already specified in Phase 1; complete here if gaps remain) -| node-resque | pgboss-queue | +| node-resque | pg-queue | | --- | --- | | Jest + ts-jest | `bun:test` (`bun test --max-concurrency=1`) | | `ioredis` specHelper | `__tests__/utils/specHelper.ts` | @@ -65,7 +65,7 @@ Use this table as a **checklist in this phase's PR**: tick what is already green | getKeys returns appropriate keys | **Skip** | Redis SCAN | | keys built with the default namespace | **Skip** | Redis key prefix | | ioredis transparent key prefix… | **Skip** | | -| keys built with a custom namespace | **Adapt** ✅ Phase 2 | `schema` option sets pg-boss schema; `migrate` sees that schema | +| keys built with a custom namespace | **Adapt** ✅ Phase 2 | `schema` selects the owned queue schema; `migrate` sees that schema | | keys built with a array namespace | **Skip** | array namespace not supported | | will properly build namespace strings dynamically | **Skip** | | | will select redis db from options | **Adapt** ✅ Phase 2 | `database` string selects Postgres database via discrete ConnectionOptions | @@ -126,7 +126,7 @@ Adapt any `specHelper.redis.rpop(namespace+":failed")` to `queue.failed(0,-1)`. ### `__tests__/integration/ioredis-mock.ts` -**Skip**. No in-memory pg-boss mock required for v1. +**Skip**. No in-memory PostgreSQL queue mock is required for v1. ### `__tests__/utils/*` @@ -151,7 +151,7 @@ If an assertion cannot be identical, add a row (may already have rows from earli | --- | --- | --- | --- | | keys built with a custom namespace | `connection.key("thing") === "customNamespace:thing"` | `connection.schema === customSchema` and `pgrq_locks` exists in that schema | Keys are not Redis-prefixed; schema replaces namespace | | removes the redis event listeners when end | `redis.listenerCount("error"|"end")` | `pool`/`boss` `listenerCount("error")` with BYO pool | No Redis `end` event; we forward `error` only | -| queue delayed-job tests using timestamp `10000` | Redis keeps the 1970 timestamp in a delayed list until Scheduler transfers it | Use a future rounded timestamp and assert the same seconds/ms conversions | pg-boss `startAfter` is eligibility time, so a past timestamp is immediately ready by design | +| queue delayed-job tests using timestamp `10000` | Redis keeps the 1970 timestamp in a delayed list until Scheduler transfers it | Use a future rounded timestamp and assert the same seconds/ms conversions | `start_after` is eligibility time, so a past timestamp is immediately ready by design | PRs that add rows must explain. "Postgres is different" is not enough if the Queue API can still match. @@ -179,3 +179,4 @@ Docs site can describe a real API. Phase 10 can trust tests that have been runni - 2026-08-26: Phase 2 — Bun requires `.test.ts` (or `.spec` / `_test_` / `_spec_`) in the filename. Matrix paths are `__tests__/core/.test.ts` while describe/test titles stay node-resque-identical. Later phases must not copy bare `connection.ts`-style names or CI will skip them. - 2026-08-29: Phase 3 preserves upstream Queue test titles but replaces hard-coded 1970 delayed timestamps with future rounded values. This is a required semantic adaptation because pg-boss uses `startAfter` directly rather than waiting for a scheduler to move a Redis-list item. - 2026-08-29: Phase 3's Queue suite now covers active/old worker metadata, force-clean, and retry-stuck behavior through seeded pg-boss and `pgrq_workers` rows. Phase 4 still repeats these paths with a live Worker but no Queue titles remain skipped. +- 2026-08-29: The Phase 2/3 PostgreSQL adaptations now target the owned `pgrq_*` schema and `Connection.fetchJob()` rather than pg-boss internals; test titles and externally visible assertions remain unchanged. diff --git a/docs/plans/09-docs-site.md b/docs/plans/09-docs-site.md index 5af2c5e..3891aeb 100644 --- a/docs/plans/09-docs-site.md +++ b/docs/plans/09-docs-site.md @@ -45,7 +45,7 @@ From keryx `docs/.vitepress/config.mts` / `docs.yaml`: - `appearance: "dark"` - `lastUpdated: true` - local search -- `editLink` → `https://github.com/actionhero/pgboss-queue/edit/main/docs/:path` +- `editLink` → `https://github.com/actionhero/pg-queue/edit/main/docs/:path` - nav: Guide, Reference, Changelog, GitHub, version from `package.json` - sidebar grouped by task (getting started → concepts → operations), not by folder internals - `vitepress-plugin-llms` + per-page `.md` alternate links (keryx does this; do it unless it slows us down — then defer, do not block) @@ -115,7 +115,7 @@ Prefer (1) for the small public surface, with a checklist: every public method h Permissions: `pages: write`, `id-token: write`. -Custom domain optional (`CNAME`); default `https://actionhero.github.io/pgboss-queue/` until DNS exists. +Custom domain optional (`CNAME`); default `https://actionhero.github.io/pg-queue/` until DNS exists. ## Tests @@ -138,4 +138,4 @@ Phase 10 adds Pages deploy (if not already in `docs.yaml`) and npm publish. Test ## Lessons learned -_None yet._ +- 2026-08-29: Product naming and examples must use `pg-queue`; migration documentation must describe the owned `pgrq_*` schema rather than pg-boss. diff --git a/docs/plans/10-publish-and-ci.md b/docs/plans/10-publish-and-ci.md index a3e347a..4a659a2 100644 --- a/docs/plans/10-publish-and-ci.md +++ b/docs/plans/10-publish-and-ci.md @@ -41,7 +41,7 @@ Keryx: 6. `actions/setup-node` with `registry-url: https://registry.npmjs.org` 7. `npm install -g npm@latest` (OIDC / provenance) -**Intended setup:** npm **trusted publishing** (OIDC) so no long-lived `NPM_TOKEN`. Configure the npm package to trust GitHub Actions on `actionhero/pgboss-queue`. Fallback: `NPM_TOKEN`. Document both in this phase's PR. +**Intended setup:** npm **trusted publishing** (OIDC) so no long-lived `NPM_TOKEN`. Configure the npm package to trust GitHub Actions on `actionhero/pg-queue`. Fallback: `NPM_TOKEN`. Document both in this phase's PR. ```yaml name: Publish @@ -71,7 +71,7 @@ jobs: id: version run: | LOCAL=$(node -p "require('./package.json').version") - REMOTE=$(npm view pgboss-queue version 2>/dev/null || echo "0.0.0") + REMOTE=$(npm view pg-queue version 2>/dev/null || echo "0.0.0") echo "local=$LOCAL" >> "$GITHUB_OUTPUT" echo "remote=$REMOTE" >> "$GITHUB_OUTPUT" if [ "$LOCAL" != "$REMOTE" ]; then @@ -107,7 +107,7 @@ Do not publish `0.0.1` empty stubs. First intentional bump to `0.1.0` is the fir Port node-resque `examples/` in Phases 4–7. This phase can add a compose-based example command if missing: ```bash -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test bun examples/example.ts +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test bun examples/example.ts ``` Optional: `examples/docker` like node-resque — not required for v1. @@ -126,7 +126,7 @@ README (user-facing): GitHub Test workflow badge (the Phase 1 workflow), npm ver ## After 1.0 -- Keryx consumes `pgboss-queue` instead of in-tree `PgBossBackend` +- Keryx consumes `pg-queue` instead of in-tree `PgBossBackend` - Optional admin UI package (resque-admin against SQL) - `LISTEN/NOTIFY` as an opt-in latency flag (`useListenNotify`) @@ -135,3 +135,4 @@ README (user-facing): GitHub Test workflow badge (the Phase 1 workflow), npm ver - 2026-08-26 (plan): Test CI is Phase 1. This phase is only Pages + npm publish. - 2026-08-26: Phase 1 already matrices Bun and Node 26 in `test.yaml`. Example runs use `DATABASE_URL`; there is no repo `docker-compose.yml`. - 2026-08-26: Phase 1 dropped the suite matrix. `test` is Bun; `node-package` only imports `dist/` on Node 26. +- 2026-08-29: npm package name changed to `pg-queue`; package contents must include `migrations/` because runtime schema installation reads the numbered SQL files. diff --git a/docs/plans/README.md b/docs/plans/README.md index 9250822..9ec4204 100644 --- a/docs/plans/README.md +++ b/docs/plans/README.md @@ -1,6 +1,6 @@ # Implementation plans -These documents are the **living** implementation spec for `pgboss-queue`. Execute them in order. Each phase lists **goal**, **why**, **deliverables**, **acceptance criteria**, **what the next phase needs**, and **Lessons learned**. +These documents are the **living** implementation spec for `pg-queue`. Execute them in order. Each phase lists **goal**, **why**, **deliverables**, **acceptance criteria**, **what the next phase needs**, and **Lessons learned**. When you implement or change something a phase covers, update that file in the same PR and append to **Lessons learned**. Do not leave plans stale. See `CLAUDE.md` → *Keep the phase plans current*. @@ -32,4 +32,4 @@ Each phase file starts with a status line: ## Lessons learned -Every phase file ends with an empty `## Lessons learned` section. Fill it as you go: surprises, pg-boss API mismatches, tests we had to adapt, decisions that diverged from the original plan. Newest entry last. Never delete old bullets. +Every phase file ends with an empty `## Lessons learned` section. Fill it as you go: storage API mismatches, tests we had to adapt, and decisions that diverged from the original plan. Newest entry last. Never delete old bullets. diff --git a/migrations/001_initial.sql b/migrations/001_initial.sql new file mode 100644 index 0000000..b0fa3ed --- /dev/null +++ b/migrations/001_initial.sql @@ -0,0 +1,60 @@ +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_queues ( + name text PRIMARY KEY, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_jobs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + name text NOT NULL REFERENCES {{schema}}.pgrq_queues(name), + data jsonb NOT NULL, + state text NOT NULL DEFAULT 'created' + CHECK (state IN ('created', 'retry', 'active', 'completed', 'cancelled', 'failed')), + priority integer NOT NULL DEFAULT 0, + created_on timestamptz NOT NULL DEFAULT now(), + start_after timestamptz NOT NULL DEFAULT now(), + started_on timestamptz, + completed_on timestamptz, + output jsonb +); + +CREATE INDEX IF NOT EXISTS pgrq_jobs_fetch_idx + ON {{schema}}.pgrq_jobs (name, state, start_after, priority DESC, created_on, id); + +CREATE INDEX IF NOT EXISTS pgrq_jobs_delayed_idx + ON {{schema}}.pgrq_jobs (start_after, created_on) + WHERE state IN ('created', 'retry'); + +CREATE INDEX IF NOT EXISTS pgrq_jobs_failed_idx + ON {{schema}}.pgrq_jobs (completed_on, created_on, id) + WHERE state = 'failed'; + +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_leader ( + slot text PRIMARY KEY DEFAULT 'default', + name text NOT NULL, + expires_at timestamptz NOT NULL +); + +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_workers ( + name text PRIMARY KEY, + queues text NOT NULL, + started_at timestamptz NOT NULL DEFAULT now(), + ping_at timestamptz NOT NULL DEFAULT now(), + working_on jsonb +); + +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_locks ( + key text PRIMARY KEY, + value text, + expires_at timestamptz NOT NULL +); + +CREATE TABLE IF NOT EXISTS {{schema}}.pgrq_stats ( + name text PRIMARY KEY, + value bigint NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS pgrq_workers_ping_at_idx + ON {{schema}}.pgrq_workers (ping_at); + +CREATE INDEX IF NOT EXISTS pgrq_locks_expires_at_idx + ON {{schema}}.pgrq_locks (expires_at); diff --git a/package.json b/package.json index 54864a3..7be4648 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { - "name": "pgboss-queue", - "version": "0.2.0", + "name": "pg-queue", + "version": "0.3.0", "description": "A PostgreSQL-backed background job queue with the node-resque runtime model", "type": "module", "main": "./dist/index.js", @@ -12,7 +12,8 @@ } }, "files": [ - "dist" + "dist", + "migrations" ], "engines": { "node": ">=26" @@ -26,8 +27,7 @@ "format": "biome check --write ." }, "dependencies": { - "pg": "^8.23.0", - "pg-boss": "^12.28.0" + "pg": "^8.23.0" }, "devDependencies": { "@biomejs/biome": "^2.5.10", diff --git a/scripts/assert-node-package.mjs b/scripts/assert-node-package.mjs index 392b01e..5c003fe 100644 --- a/scripts/assert-node-package.mjs +++ b/scripts/assert-node-package.mjs @@ -14,7 +14,13 @@ assert.equal( assert.ok(process.versions.node, "expected a Node.js runtime"); assert.equal(typeof pkg.name, "string"); +assert.equal(pkg.name, "pg-queue"); assert.equal(typeof pkg.exports?.["."]?.import, "string"); +assert.ok(pkg.files.includes("migrations")); +assert.match( + readFileSync(join(root, "migrations", "001_initial.sql"), "utf8"), + /CREATE TABLE IF NOT EXISTS \{\{schema\}\}\.pgrq_jobs/, +); const entry = pkg.exports["."].import; const url = pathToFileURL(join(root, entry)).href; diff --git a/src/core/connection.ts b/src/core/connection.ts index 6296291..f59f228 100644 --- a/src/core/connection.ts +++ b/src/core/connection.ts @@ -1,19 +1,30 @@ import { EventEmitter } from "node:events"; +import { readFile } from "node:fs/promises"; import { Pool, type PoolConfig, type QueryResult, type QueryResultRow, } from "pg"; -import { PgBoss } from "pg-boss"; -const DEFAULT_SCHEMA = "pgboss_queue"; -const DEFAULT_APPLICATION_NAME = "pgboss-queue"; +const DEFAULT_SCHEMA = "pg_queue"; +const DEFAULT_APPLICATION_NAME = "pg-queue"; const SCHEMA_PATTERN = /^[a-zA-Z_][a-zA-Z0-9_]*$/; const LEADER_SLOT = "default"; +const MIGRATIONS = [{ version: 1, name: "initial", file: "001_initial.sql" }]; + +/** A job atomically claimed from a queue. */ +export interface FetchedJob { + /** Stable UUID. */ + id: string; + /** Queue name. */ + name: string; + /** Application payload. */ + data: T; +} /** - * Postgres connection options for pgboss-queue. + * PostgreSQL connection options for pg-queue. * * Maps from node-resque Redis options as follows: * - `{ host, port, password, database: 0 }` → `{ connectionString }` or @@ -44,12 +55,9 @@ export interface ConnectionOptions { * Analogous to passing `redis: ioredisInstance`. */ pool?: Pool; - /** - * pg-boss schema AND our metadata schema. Default `pgboss_queue`. - * Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (reject otherwise). - */ + /** Queue schema. Default `pg_queue`; must be a legal SQL identifier. */ schema?: string; - /** Reported to Postgres as `application_name`. Default `pgboss-queue`. */ + /** Reported to Postgres as `application_name`. Default `pg-queue`. */ application_name?: string; } @@ -84,7 +92,7 @@ export interface SchedulerOptions extends QueueOptions { leaderLockTimeout?: number; stuckWorkerTimeout?: number | false; retryStuckJobs?: boolean; - /** Leader runs pg-boss migrate + metadata DDL. Default `true`. */ + /** Leader runs pg-queue migrations. Default `true`. */ automigrate?: boolean; /** * Leader deletes completed/cancelled jobs older than this. Default 24h. @@ -104,12 +112,10 @@ export interface MultiWorkerOptions extends WorkerOptions { } /** - * Postgres + pg-boss connection used by Queue, Worker, and Scheduler. + * PostgreSQL connection used by Queue, Worker, and Scheduler. * * Call {@link Connection.migrate} (via the elected scheduler, or explicitly in - * tests/deploy) before workers expect a usable schema. Instances constructed for - * workers/queues always use `migrate: false`, `supervise: false`, and - * `schedule: false` on pg-boss. + * tests/deploy) before workers expect a usable schema. */ export class Connection extends EventEmitter { /** Resolved connection options (defaults applied). */ @@ -119,14 +125,11 @@ export class Connection extends EventEmitter { private eventListeners: { poolError?: (error: Error) => void; - bossError?: (error: Error) => void; } = {}; private _pool: Pool | undefined; - private _boss: PgBoss | undefined; private _schema: string; private ownsPool = false; - private bossStarted = false; /** * @param options - Postgres connection options. Redis options are rejected. @@ -155,7 +158,7 @@ export class Connection extends EventEmitter { this.connected = false; } - /** Validated pg-boss / metadata schema name. */ + /** Validated queue schema name. */ get schema(): string { return this._schema; } @@ -172,20 +175,7 @@ export class Connection extends EventEmitter { } /** - * pg-boss client started with `migrate`/`supervise`/`schedule` disabled. - * @throws If not connected. - */ - get boss(): PgBoss { - if (!this._boss) { - throw new Error("Connection is not connected"); - } - return this._boss; - } - - /** - * Open the pool (unless `pool` was provided), construct pg-boss, and start it - * when the schema is already installed. If pg-boss is not installed yet, - * the pool stays usable so {@link Connection.migrate} can run. + * Open the pool (unless `pool` was provided) and verify it with `SELECT 1`. * * @throws On connection failure (emits `error` as well). */ @@ -193,8 +183,8 @@ export class Connection extends EventEmitter { if (this.connected) return; try { - await this.ensurePoolAndBoss(); - await this.tryStartBoss(); + this.ensurePool(); + await this.pool.query("SELECT 1"); this.connected = true; } catch (error) { const err = toError(error); @@ -205,22 +195,16 @@ export class Connection extends EventEmitter { } /** - * Stop pg-boss and, if we created the pool, end it. Provided pools are left open. - * Removes forwarded `error` listeners from the pool and boss. + * End an owned pool. Provided pools are left open. + * Removes the forwarded pool `error` listener. */ async end(): Promise { this.removeForwardedListeners(); - if (this._boss && this.bossStarted) { - await this._boss.stop({ graceful: true, close: false }); - this.bossStarted = false; - } - if (this.ownsPool && this._pool) { await this._pool.end(); } - this._boss = undefined; this._pool = undefined; this.ownsPool = false; this.connected = false; @@ -257,37 +241,103 @@ export class Connection extends EventEmitter { } /** - * Idempotently install the schema: `CREATE SCHEMA`, pg-boss migrate, and - * `pgrq_*` metadata tables/indexes. Intended for the scheduler leader (or tests). + * Atomically claim the next ready job using `FOR UPDATE SKIP LOCKED`. + * + * @param queue - Queue to claim from. + * @returns The claimed job, or `null` when no job is ready. + */ + async fetchJob(queue: string): Promise | null> { + const result = await this.query< + QueryResultRow & { id: string; name: string; data: T } + >( + `WITH candidate AS ( + SELECT id + FROM ${this._schema}.pgrq_jobs + WHERE name = $1 + AND state IN ('created', 'retry') + AND start_after <= now() + ORDER BY priority DESC, created_on, id + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + UPDATE ${this._schema}.pgrq_jobs AS job + SET state = 'active', started_on = now() + FROM candidate + WHERE job.id = candidate.id + RETURNING job.id, job.name, job.data`, + [queue], + ); + return result.rows[0] ?? null; + } + + /** + * Delete one job by queue and id. + * + * @param queue - Queue containing the job. + * @param id - Job UUID. + * @returns Whether a job was deleted. + */ + async deleteJob(queue: string, id: string): Promise { + const result = await this.query( + `DELETE FROM ${this._schema}.pgrq_jobs WHERE name = $1 AND id = $2`, + [queue, id], + ); + return (result.rowCount ?? 0) > 0; + } + + /** + * Apply all pending versioned SQL migrations under a transaction-scoped + * advisory lock. Safe to call concurrently from multiple processes. * * @throws If the pool cannot be opened or migration SQL fails. */ async migrate(): Promise { - if (!this._pool || !this._boss) { - await this.ensurePoolAndBoss(); + if (!this._pool) { + this.ensurePool(); + await this.pool.query("SELECT 1"); this.connected = true; } const schema = this._schema; - await this.pool.query(`CREATE SCHEMA IF NOT EXISTS ${schema}`); - - const migrator = new PgBoss({ - db: this.dbAdapter(), - schema, - migrate: true, - supervise: false, - schedule: false, - application_name: this.options.application_name, - }); - - await migrator.start(); - await migrator.stop({ graceful: true, close: false }); - - await this.applyMetadataDdl(); - - if (!this.bossStarted) { - await this.boss.start(); - this.bossStarted = true; + const client = await this.pool.connect(); + try { + await client.query("BEGIN"); + await client.query( + "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", + ["pg-queue:migrate", schema], + ); + await client.query(`CREATE SCHEMA IF NOT EXISTS ${schema}`); + await client.query(` + CREATE TABLE IF NOT EXISTS ${schema}.pgrq_migrations ( + version integer PRIMARY KEY, + name text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now() + ) + `); + + const applied = await client.query<{ version: number }>( + `SELECT version FROM ${schema}.pgrq_migrations`, + ); + const versions = new Set(applied.rows.map((row) => row.version)); + + for (const migration of MIGRATIONS) { + if (versions.has(migration.version)) continue; + const sql = await loadMigration(migration.file, schema); + await client.query(sql); + await client.query( + `INSERT INTO ${schema}.pgrq_migrations (version, name) + VALUES ($1, $2)`, + [migration.version, migration.name], + ); + } + await client.query("COMMIT"); + } catch (error) { + await client.query("ROLLBACK").catch(() => { + // Preserve the original migration error. + }); + throw error; + } finally { + client.release(); } } @@ -472,7 +522,7 @@ export class Connection extends EventEmitter { return result.rows[0]?.name ?? null; } - private async ensurePoolAndBoss(): Promise { + private ensurePool(): void { if (!this._pool) { if (this.options.pool) { this._pool = this.options.pool; @@ -482,71 +532,9 @@ export class Connection extends EventEmitter { this.ownsPool = true; } } - - if (!this._boss) { - this._boss = new PgBoss({ - db: this.dbAdapter(), - schema: this._schema, - migrate: false, - supervise: false, - schedule: false, - application_name: this.options.application_name, - }); - } - this.attachForwardedListeners(); } - private async tryStartBoss(): Promise { - if (!this._boss || this.bossStarted) return; - - try { - await this._boss.start(); - this.bossStarted = true; - } catch (error) { - if (isPgBossNotInstalled(error)) { - return; - } - throw error; - } - } - - private async applyMetadataDdl(): Promise { - const schema = this._schema; - await this.pool.query(` - CREATE TABLE IF NOT EXISTS ${schema}.pgrq_leader ( - slot text PRIMARY KEY DEFAULT 'default', - name text NOT NULL, - expires_at timestamptz NOT NULL - ); - - CREATE TABLE IF NOT EXISTS ${schema}.pgrq_workers ( - name text PRIMARY KEY, - queues text NOT NULL, - started_at timestamptz NOT NULL DEFAULT now(), - ping_at timestamptz NOT NULL DEFAULT now(), - working_on jsonb - ); - - CREATE TABLE IF NOT EXISTS ${schema}.pgrq_locks ( - key text PRIMARY KEY, - value text, - expires_at timestamptz NOT NULL - ); - - CREATE TABLE IF NOT EXISTS ${schema}.pgrq_stats ( - name text PRIMARY KEY, - value bigint NOT NULL DEFAULT 0 - ); - - CREATE INDEX IF NOT EXISTS pgrq_workers_ping_at_idx - ON ${schema}.pgrq_workers (ping_at); - - CREATE INDEX IF NOT EXISTS pgrq_locks_expires_at_idx - ON ${schema}.pgrq_locks (expires_at); - `); - } - private buildPoolConfig(): PoolConfig { const { connectionString, @@ -574,60 +562,33 @@ export class Connection extends EventEmitter { }; } - private dbAdapter(): { - executeSql: ( - text: string, - values?: unknown[], - ) => Promise>; - } { - return { - executeSql: (text: string, values?: unknown[]) => - this.pool.query(text, values), - }; - } - private attachForwardedListeners(): void { - if (!this._pool || !this._boss) return; + if (!this._pool) return; this.removeForwardedListeners(); this.eventListeners.poolError = (error: Error) => { this.emit("error", error); }; - this.eventListeners.bossError = (error: Error) => { - this.emit("error", error); - }; - this._pool.on("error", this.eventListeners.poolError); - this._boss.on("error", this.eventListeners.bossError); } private removeForwardedListeners(): void { if (this._pool && this.eventListeners.poolError) { this._pool.off("error", this.eventListeners.poolError); } - if (this._boss && this.eventListeners.bossError) { - this._boss.off("error", this.eventListeners.bossError); - } this.eventListeners = {}; } private async teardownPartialConnect(): Promise { this.removeForwardedListeners(); - if (this._boss && this.bossStarted) { - await this._boss.stop({ graceful: false, close: false }).catch(() => { - // best-effort cleanup after a failed connect - }); - } if (this.ownsPool && this._pool) { await this._pool.end().catch(() => { // best-effort cleanup after a failed connect }); } - this._boss = undefined; this._pool = undefined; this.ownsPool = false; - this.bossStarted = false; this.connected = false; } } @@ -666,15 +627,13 @@ function rejectRedisOptions(options: ConnectionOptions): void { } } -function isPgBossNotInstalled(error: unknown): boolean { - const message = toError(error).message.toLowerCase(); - return ( - message.includes("not installed") || - (message.includes("schema") && message.includes("missing")) - ); -} - function toError(error: unknown): Error { if (error instanceof Error) return error; return new Error(String(error)); } + +async function loadMigration(file: string, schema: string): Promise { + const url = new URL(`../../migrations/${file}`, import.meta.url); + const sql = await readFile(url, "utf8"); + return sql.replaceAll("{{schema}}", schema); +} diff --git a/src/core/queue.ts b/src/core/queue.ts index fa7ef71..74a9fde 100644 --- a/src/core/queue.ts +++ b/src/core/queue.ts @@ -10,7 +10,7 @@ import { runPlugins } from "./pluginRunner.js"; const QUEUED_STATES = ["created", "retry"] as const; const DUPLICATE_ERROR = "Job already enqueued at this time with same arguments"; -/** Payload stored in pg-boss's `job.data` column. */ +/** Payload stored in the queue job table's `data` column. */ export interface ParsedJob { /** Registered job name. */ class: string; @@ -32,13 +32,13 @@ export interface ParsedWorkerPayload { worker: string; /** Encoded job payload. */ payload: ParsedJob; - /** pg-boss job id recorded by the worker, when present. */ + /** Job id recorded by the worker, when present. */ id?: string; } /** Failed-job representation returned by Queue inspection methods. */ export interface ParsedFailedJobPayload extends ErrorPayload { - /** pg-boss job id used for precise removal and retry. */ + /** Job id used for precise removal and retry. */ id?: string; } @@ -61,15 +61,15 @@ interface WorkerRow extends QueryResultRow { /** * PostgreSQL-backed node-resque Queue API. * - * Queue methods use pg-boss for insertion and its `job` table for compatible - * inspection and administration. + * Queue methods use pg-queue's versioned PostgreSQL schema for insertion, + * inspection, and administration. */ export class Queue extends EventEmitter { /** Resolved Queue options. */ readonly options: QueueOptions; /** Named jobs used by enqueue plugins and, later, Workers. */ readonly jobs: Jobs; - /** Underlying PostgreSQL / pg-boss connection. */ + /** Underlying PostgreSQL connection. */ readonly connection: Connection; /** @@ -84,7 +84,7 @@ export class Queue extends EventEmitter { this.connection.on("error", (error: Error) => this.emit("error", error)); } - /** Connect the underlying PostgreSQL and pg-boss clients. */ + /** Connect the underlying PostgreSQL client. */ async connect(): Promise { await this.connection.connect(); } @@ -212,20 +212,12 @@ export class Queue extends EventEmitter { ); } - /** @returns All known pg-boss queue names and queue names present in jobs. */ + /** @returns All known queue names. */ async queues(): Promise { - const [configured, jobs] = await Promise.all([ - this.connection.boss.getQueues(), - this.connection.query<{ name: string }>( - `SELECT DISTINCT name FROM ${this.connection.schema}.job`, - ), - ]); - return Array.from( - new Set([ - ...configured.map((queue) => queue.name), - ...jobs.rows.map((row) => row.name), - ]), - ).sort(); + const result = await this.connection.query<{ name: string }>( + `SELECT name FROM ${this.connection.schema}.pgrq_queues ORDER BY name`, + ); + return result.rows.map((row) => row.name); } /** @@ -240,26 +232,29 @@ export class Queue extends EventEmitter { try { await client.query("BEGIN"); const locked = await client.query( - `SELECT name FROM ${schema}.queue WHERE name = $1 FOR UPDATE`, + `SELECT name FROM ${schema}.pgrq_queues WHERE name = $1 FOR UPDATE`, [q], ); const result = await client.query<{ data: unknown; start_after: Date; }>( - `DELETE FROM ${schema}.job + `DELETE FROM ${schema}.pgrq_jobs WHERE name = $1 AND state <> 'active' RETURNING data, start_after`, [q], ); const remaining = await client.query<{ exists: boolean }>( `SELECT EXISTS ( - SELECT 1 FROM ${schema}.job WHERE name = $1 + SELECT 1 FROM ${schema}.pgrq_jobs WHERE name = $1 ) AS exists`, [q], ); if (!remaining.rows[0]?.exists && (locked.rowCount ?? 0) > 0) { - await client.query(`SELECT ${schema}.delete_queue($1)`, [q]); + await client.query( + `DELETE FROM ${schema}.pgrq_queues WHERE name = $1`, + [q], + ); } await client.query("COMMIT"); @@ -299,9 +294,9 @@ export class Queue extends EventEmitter { async length(q: string): Promise { const result = await this.connection.query<{ count: string }>( `SELECT count(*)::text AS count - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY($2::${this.connection.schema}.job_state[]) + AND state = ANY($2::text[]) AND start_after <= now()`, [q, QUEUED_STATES], ); @@ -320,9 +315,9 @@ export class Queue extends EventEmitter { const range = sqlRange(start, stop); const result = await this.connection.query( `SELECT id, name, data, created_on, start_after - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY($2::${this.connection.schema}.job_state[]) + AND state = ANY($2::text[]) AND start_after <= now() ORDER BY created_on, id OFFSET $3 @@ -356,15 +351,15 @@ export class Queue extends EventEmitter { const result = await this.connection.query( `WITH selected AS ( SELECT id - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + AND state = ANY(ARRAY['created','retry']::text[]) AND start_after <= now() AND data = $2::jsonb ORDER BY created_on ${direction}, id ${direction} ${limitSql} ) - DELETE FROM ${this.connection.schema}.job + DELETE FROM ${this.connection.schema}.pgrq_jobs WHERE id IN (SELECT id FROM selected)`, values, ); @@ -390,15 +385,15 @@ export class Queue extends EventEmitter { const result = await this.connection.query( `WITH sliced AS ( SELECT id, data - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + AND state = ANY(ARRAY['created','retry']::text[]) AND start_after <= now() ORDER BY created_on, id OFFSET $3 ${range.limitSql} ) - DELETE FROM ${this.connection.schema}.job + DELETE FROM ${this.connection.schema}.pgrq_jobs WHERE id IN ( SELECT id FROM sliced WHERE data->>'class' = $2 )`, @@ -422,9 +417,9 @@ export class Queue extends EventEmitter { ): Promise { const encoded = this.encode(q, func, arrayify(args)); const result = await this.connection.query<{ start_after: Date }>( - `DELETE FROM ${this.connection.schema}.job + `DELETE FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + AND state = ANY(ARRAY['created','retry']::text[]) AND start_after > now() AND data = $2::jsonb RETURNING start_after`, @@ -456,9 +451,9 @@ export class Queue extends EventEmitter { ): Promise { const result = await this.connection.query<{ start_after: Date }>( `SELECT start_after - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE name = $1 - AND state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + AND state = ANY(ARRAY['created','retry']::text[]) AND start_after > now() AND data = $2::jsonb ORDER BY start_after, created_on, id`, @@ -473,8 +468,8 @@ export class Queue extends EventEmitter { async timestamps(): Promise { const result = await this.connection.query<{ start_after: Date }>( `SELECT DISTINCT start_after - FROM ${this.connection.schema}.job - WHERE state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + FROM ${this.connection.schema}.pgrq_jobs + WHERE state = ANY(ARRAY['created','retry']::text[]) AND start_after > now() ORDER BY start_after`, ); @@ -497,8 +492,8 @@ export class Queue extends EventEmitter { ); const result = await this.connection.query( `SELECT id, name, data, created_on, start_after - FROM ${this.connection.schema}.job - WHERE state = ANY(ARRAY['created','retry']::${this.connection.schema}.job_state[]) + FROM ${this.connection.schema}.pgrq_jobs + WHERE state = ANY(ARRAY['created','retry']::text[]) AND start_after > now() AND start_after >= to_timestamp($1) AND start_after < to_timestamp($1) + interval '1 second' @@ -651,7 +646,7 @@ export class Queue extends EventEmitter { } /** - * Mark the worker's in-flight pg-boss job failed in place. + * Mark the worker's in-flight job failed in place. * * @param working - Worker assignment, including optional job id. * @param errorPayload - Resque failure payload stored in `output`. @@ -664,7 +659,7 @@ export class Queue extends EventEmitter { const updated = await this.connection.query( `WITH selected AS ( SELECT name, id - FROM ${schema}.job + FROM ${schema}.pgrq_jobs WHERE name = $1 AND state = 'active' AND ( @@ -675,7 +670,7 @@ export class Queue extends EventEmitter { LIMIT 1 FOR UPDATE ) - UPDATE ${schema}.job AS job + UPDATE ${schema}.pgrq_jobs AS job SET state = 'failed', completed_on = now(), output = $3::jsonb @@ -693,9 +688,9 @@ export class Queue extends EventEmitter { await this.ensureQueue(working.queue); await this.connection.query( - `INSERT INTO ${this.connection.schema}.job - (name, data, state, retry_limit, completed_on, output) - VALUES ($1, $2::jsonb, 'failed', 0, now(), $3::jsonb)`, + `INSERT INTO ${this.connection.schema}.pgrq_jobs + (name, data, state, completed_on, output) + VALUES ($1, $2::jsonb, 'failed', now(), $3::jsonb)`, [ working.queue, JSON.stringify(working.payload), @@ -725,11 +720,11 @@ export class Queue extends EventEmitter { return result; } - /** @returns Number of failed pg-boss jobs. */ + /** @returns Number of failed jobs. */ async failedCount(): Promise { const result = await this.connection.query<{ count: string }>( `SELECT count(*)::text AS count - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE state = 'failed'`, ); return Number(result.rows[0]?.count ?? 0); @@ -746,7 +741,7 @@ export class Queue extends EventEmitter { const range = sqlRange(start, stop, "$2"); const result = await this.connection.query( `SELECT id, name, data, output, created_on, start_after, completed_on - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE state = 'failed' ORDER BY completed_on, created_on, id OFFSET $1 @@ -766,13 +761,13 @@ export class Queue extends EventEmitter { const candidates = failedJob.id ? await this.connection.query( `SELECT id, name, data, output, created_on, start_after, completed_on - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE id = $1 AND state = 'failed'`, [failedJob.id], ) : await this.connection.query( `SELECT id, name, data, output, created_on, start_after, completed_on - FROM ${this.connection.schema}.job + FROM ${this.connection.schema}.pgrq_jobs WHERE state = 'failed' ORDER BY completed_on, created_on, id`, ); @@ -783,7 +778,7 @@ export class Queue extends EventEmitter { if (!match) return 0; const deleted = await this.connection.query( - `DELETE FROM ${this.connection.schema}.job + `DELETE FROM ${this.connection.schema}.pgrq_jobs WHERE id = $1 AND state = 'failed'`, [match.id], ); @@ -857,41 +852,26 @@ export class Queue extends EventEmitter { payload: ParsedJob, options: { startAfter?: Date } = {}, ): Promise { - let lastError: Error | undefined; - for (let attempt = 0; attempt < 2; attempt += 1) { - await this.ensureQueue(q); - try { - const id = await this.connection.boss.send(q, payload, { - retryLimit: 0, - deleteAfterSeconds: 0, - ...options, - }); - if (!id) { - throw new Error( - `pg-boss did not enqueue job "${payload.class}" on queue "${q}"`, - ); - } - return id; - } catch (error) { - lastError = toError(error); - if (attempt === 0 && isMissingQueue(lastError, q)) continue; - throw lastError; - } - } - throw lastError ?? new Error(`pg-boss did not enqueue job on queue "${q}"`); + await this.ensureQueue(q); + const result = await this.connection.query<{ id: string }>( + `INSERT INTO ${this.connection.schema}.pgrq_jobs + (name, data, start_after) + VALUES ($1, $2::jsonb, $3) + RETURNING id`, + [q, JSON.stringify(payload), options.startAfter ?? new Date()], + ); + const id = result.rows[0]?.id; + if (!id) throw new Error(`Failed to enqueue job on queue "${q}"`); + return id; } private async ensureQueue(q: string): Promise { - const existing = await this.connection.boss.getQueue(q); - if (existing) return; - try { - await this.connection.boss.createQueue(q, { - retryLimit: 0, - deleteAfterSeconds: 0, - }); - } catch (error) { - if (!(await this.connection.boss.getQueue(q))) throw error; - } + await this.connection.query( + `INSERT INTO ${this.connection.schema}.pgrq_queues (name) + VALUES ($1) + ON CONFLICT (name) DO NOTHING`, + [q], + ); } private async acquireDelayedLock( @@ -1045,11 +1025,3 @@ function sqlRange( function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } - -function isMissingQueue(error: Error, q: string): boolean { - return error.message.includes(`Queue ${q} does not exist`); -} - -function toError(error: unknown): Error { - return error instanceof Error ? error : new Error(String(error)); -} diff --git a/src/index.ts b/src/index.ts index 569561c..b521041 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,10 +1,11 @@ /** - * pgboss-queue — node-resque runtime model on PostgreSQL via pg-boss. + * pg-queue — node-resque runtime model on PostgreSQL. */ export { assertSchema, Connection, type ConnectionOptions, + type FetchedJob, type MultiWorkerOptions, type QueueOptions, type SchedulerOptions, diff --git a/src/types/errorPayload.ts b/src/types/errorPayload.ts index 72f93fd..f850eb4 100644 --- a/src/types/errorPayload.ts +++ b/src/types/errorPayload.ts @@ -2,7 +2,7 @@ import type { ParsedJob } from "../core/queue.js"; /** Resque-compatible representation of a failed job. */ export interface ErrorPayload { - /** pg-boss job id, when the failure came from the job table. */ + /** Job id, when the failure came from the job table. */ id?: string; /** Worker that failed the job, or an empty string when unknown. */ worker: string; From 254286575e4b681cdb9e40a6eda1169f475bf993 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 29 Aug 2026 06:33:44 +0000 Subject: [PATCH 2/4] Use PostgreSQL-safe schema defaults Co-authored-by: Evan Tahler --- .env.example | 2 +- .github/workflows/test.yaml | 4 ++-- CLAUDE.md | 4 ++-- README.md | 4 ++-- __tests__/core/connection.test.ts | 9 +++++++++ __tests__/core/connectionError.test.ts | 4 ++-- __tests__/utils/specHelper.ts | 15 +++++++-------- docs/plans/00-overview.md | 4 ++-- docs/plans/01-repo-scaffold.md | 8 ++++---- docs/plans/02-connection-and-schema.md | 3 ++- docs/plans/10-publish-and-ci.md | 2 +- src/core/connection.ts | 9 +++++++-- 12 files changed, 41 insertions(+), 27 deletions(-) diff --git a/.env.example b/.env.example index 2a35073..2b729a6 100644 --- a/.env.example +++ b/.env.example @@ -1 +1 @@ -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index c5df5b5..90f973a 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -33,7 +33,7 @@ jobs: image: postgres:16 env: POSTGRES_PASSWORD: postgres - POSTGRES_DB: pg_queue_test + POSTGRES_DB: pgqueue_test options: >- --health-cmd pg_isready --health-interval 10s @@ -47,7 +47,7 @@ jobs: - run: bun install --frozen-lockfile - run: bun run test env: - DATABASE_URL: postgres://postgres:postgres@localhost:5432/pg_queue_test + DATABASE_URL: postgres://postgres:postgres@localhost:5432/pgqueue_test node-package: runs-on: ubuntu-latest diff --git a/CLAUDE.md b/CLAUDE.md index 6f02ae9..8db92f8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -60,7 +60,7 @@ bun docs:dev # VitePress (Phase 9) Local Postgres: set `DATABASE_URL` (see `.env.example`). CI starts Postgres as a workflow service; there is no `docker-compose.yml`. ```bash -# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test +# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test ``` Tests create and tear down the configured schema per file (see `specHelper`). Never point tests at a production database. @@ -112,7 +112,7 @@ type ConnectionOptions = { password?: string; ssl?: boolean | object; pool?: import("pg").Pool; // bring-your-own pool - schema?: string; // default "pg_queue" (was Redis namespace) + schema?: string; // default "pgqueue" (was Redis namespace) }; type SchedulerOptions = ConnectionOptions & { diff --git a/README.md b/README.md index 48eb793..3562be2 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ Pass a Postgres URL (not a Redis URL): ```ts const connection = { connectionString: "postgres://user:pass@host:5432/dbname", - schema: "pg_queue", // optional; default "pg_queue" + schema: "pgqueue", // optional; default "pgqueue" }; ``` @@ -77,7 +77,7 @@ const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); const connection = { pool }; ``` -`schema` isolates this library's `pgrq_*` tables inside one Postgres database. It defaults to `pg_queue` and must be a legal SQL identifier (`letters`, `numbers`, `_`). +`schema` isolates this library's `pgrq_*` tables inside one Postgres database. It defaults to `pgqueue` and must be a legal SQL identifier (`letters`, `numbers`, `_`) that does not begin with PostgreSQL's reserved `pg_` prefix. Coming from node-resque: replace `{ host, port, password, database: 0 }` / `{ redis }` / `{ namespace: "resque" }` with `{ connectionString }` or `{ pool }` and `{ schema }`. diff --git a/__tests__/core/connection.test.ts b/__tests__/core/connection.test.ts index c9bdbdc..025363a 100644 --- a/__tests__/core/connection.test.ts +++ b/__tests__/core/connection.test.ts @@ -16,6 +16,12 @@ describe("connection", () => { await specHelper.disconnect(); }); + test("uses PostgreSQL-safe project defaults", () => { + const connection = new Connection(); + expect(connection.schema).toBe("pgqueue"); + expect(connection.options.application_name).toBe("pg-queue"); + }); + test("should start with no redis keys in the namespace", async () => { // Adapt: after cleanup, no job rows and no pgrq_* rows const pool = await specHelper.connect(); @@ -215,6 +221,9 @@ describe("connection", () => { /Invalid schema/, ); expect(() => new Connection({ schema: "" })).toThrow(/Invalid schema/); + expect(() => new Connection({ schema: "pg_queue" })).toThrow( + /reserves the "pg_" prefix/, + ); }); test("reject Redis options", () => { diff --git a/__tests__/core/connectionError.test.ts b/__tests__/core/connectionError.test.ts index 1d413bf..95c9d23 100644 --- a/__tests__/core/connectionError.test.ts +++ b/__tests__/core/connectionError.test.ts @@ -7,10 +7,10 @@ describe("connection error", () => { const brokenConnection = new Connection({ host: "127.0.0.1", port: 1, - database: "pg_queue_test", + database: "pgqueue_test", user: "postgres", password: "postgres", - schema: "pg_queue_test", + schema: "pgqueue_test", }); let sawErrorEvent = false; diff --git a/__tests__/utils/specHelper.ts b/__tests__/utils/specHelper.ts index cd27e16..790f5e5 100644 --- a/__tests__/utils/specHelper.ts +++ b/__tests__/utils/specHelper.ts @@ -12,7 +12,7 @@ if (!connectionString) { ); } -export const schema = "pg_queue_test"; +export const schema = "pgqueue_test"; export const timeout = 500; export const queue = "default"; @@ -100,11 +100,8 @@ export async function cleanup(): Promise { const names = new Set(tables.rows.map((row) => row.table_name)); - if (names.has("pgrq_jobs")) { - await connection.query(`TRUNCATE TABLE ${schema}.pgrq_jobs`); - } - - const meta = [ + const dataTables = [ + "pgrq_jobs", "pgrq_queues", "pgrq_leader", "pgrq_workers", @@ -112,9 +109,11 @@ export async function cleanup(): Promise { "pgrq_stats", ].filter((name) => names.has(name)); - if (meta.length > 0) { + if (dataTables.length > 0) { await connection.query( - `TRUNCATE TABLE ${meta.map((name) => `${schema}.${name}`).join(", ")}`, + `TRUNCATE TABLE ${dataTables + .map((name) => `${schema}.${name}`) + .join(", ")}`, ); } } diff --git a/docs/plans/00-overview.md b/docs/plans/00-overview.md index 9a0442d..70e8a88 100644 --- a/docs/plans/00-overview.md +++ b/docs/plans/00-overview.md @@ -35,7 +35,7 @@ We therefore own a deliberately small, versioned schema: `pgrq_queues`, `pgrq_jo | node-resque (Redis) | pg-queue (Postgres) | | --- | --- | | `connection.host/port/database/password` (Redis) | `connectionString` **or** `host/port/database/user/password/ssl` **or** `pool` | -| `namespace` / keyPrefix | `schema` (default `pg_queue`) | +| `namespace` / keyPrefix | `schema` (default `pgqueue`) | | List `queue:{name}` (`RPUSH`/`LPOP`) | `pgrq_jobs`, `state IN ('created','retry')`, `start_after <= now()` | | `delayed:{ts}` + `delayed_queue_schedule` zset | `pgrq_jobs.start_after` in the future | | `failed` list | `pgrq_jobs.state = 'failed'` (payload mapped to `ParsedFailedJobPayload`) | @@ -73,7 +73,7 @@ Same as `node-resque/src/index.ts`: ## Lessons from keryx#519 (use) 1. **Connection strings, not Redis hashes.** `config.database.connectionString` / `new PgBoss({ connectionString, schema })`. -2. **Schema isolation.** Default `keryx_tasks` there; we default `pg_queue`. Validate identifier safety (`^[a-zA-Z_][a-zA-Z0-9_]*$`). +2. **Schema isolation.** Default `keryx_tasks` there; we default `pgqueue`. Validate identifier safety (`^[a-zA-Z_][a-zA-Z0-9_]*$`) and reject PostgreSQL's reserved `pg_` prefix. 3. **SQL introspection.** `queued`, `del`, `delDelayed`, `scheduledAt`, and `failed*` are parameterized SQL on `pgrq_jobs`. 4. **Payload shape.** Store `{ class, queue, args }` (resque encode) in `data`. 5. **Create queues lazily.** Insert `pgrq_queues` with `ON CONFLICT DO NOTHING` before enqueue. diff --git a/docs/plans/01-repo-scaffold.md b/docs/plans/01-repo-scaffold.md index 6b9e99e..054c674 100644 --- a/docs/plans/01-repo-scaffold.md +++ b/docs/plans/01-repo-scaffold.md @@ -33,7 +33,7 @@ CI must exist **before** Queue/Worker code. An empty `expect(true)` that never o - `.gitignore` — `node_modules`, `dist`, `.env`, `docs/.vitepress/dist`, `*.log` - `LICENSE` — Apache-2.0 - `.nvmrc` or `.node-version` — `26` -- `.env.example` — `DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test` +- `.env.example` — `DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test` ### Source stub @@ -51,7 +51,7 @@ No `docker-compose.yml`. CI starts Postgres as a GitHub Actions service. Locally `.env.example` is the only local-DB contract: ``` -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test ``` ### Test harness (not a dummy assert) @@ -59,7 +59,7 @@ DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test `__tests__/utils/specHelper.ts` — the same helper later phases grow. In this phase it must: - Read `DATABASE_URL` (fail the suite with a clear message if unset) -- Export `connectionDetails`, `timeout` (e.g. 500), `queue` (a default queue name), `schema` (default `pg_queue_test`) +- Export `connectionDetails`, `timeout` (e.g. 500), `queue` (a default queue name), `schema` (default `pgqueue_test`) - `connect()` / `disconnect()` against `pg.Pool` - `cleanup()` — no-op or `SELECT 1` until Phase 2 adds truncate/migrate - `popFromQueue()` — throw `"not implemented"` until Phase 3 (do not silently return null) @@ -76,7 +76,7 @@ Do **not** ship only `expect(true).toBe(true)`. Tests use `bun:test`. Node compa `.github/workflows/test.yaml` is the product gate from this PR onward. Jobs: `lint`, `build`, `test` (Postgres 16 service, **Bun `bun:test`**), `node-package` (Node 26 imports the compiled package), `complete`. -- `test`: `bun run test` with `DATABASE_URL=postgres://postgres:postgres@localhost:5432/pg_queue_test` +- `test`: `bun run test` with `DATABASE_URL=postgres://postgres:postgres@localhost:5432/pgqueue_test` - `node-package`: `actions/setup-node` from `.nvmrc` (26), `bun run build`, then **`node scripts/assert-node-package.mjs`** (not `bun run`; no Postgres) See the workflow file for the YAML. Do not duplicate a second test pipeline in Phase 10. diff --git a/docs/plans/02-connection-and-schema.md b/docs/plans/02-connection-and-schema.md index e1cc3ed..b1961d9 100644 --- a/docs/plans/02-connection-and-schema.md +++ b/docs/plans/02-connection-and-schema.md @@ -27,7 +27,7 @@ export interface ConnectionOptions { */ pool?: import("pg").Pool; /** - * Queue and metadata schema. Default `pg_queue`. + * Queue and metadata schema. Default `pgqueue`. * Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (reject otherwise). */ schema?: string; @@ -248,4 +248,5 @@ Do not defer these to Phase 8. - 2026-08-26: Node ESM (`"type": "module"`) requires relative import specifiers with `.js` extensions in emitted `dist/` (e.g. `from "./core/connection.js"`). Without them, `node scripts/assert-node-package.mjs` fails with `ERR_MODULE_NOT_FOUND` even though `tsc` and Bun tests pass. - 2026-08-29: Removed pg-boss after Phase 3 confirmed that its lifecycle and schema added coupling without supplying the resque runtime. Migrations now ship as numbered SQL files, execute under an advisory lock in one transaction, and are included in the npm package. - 2026-08-29: `connect()` must execute `SELECT 1`; constructing `pg.Pool` is lazy and does not prove credentials, routing, or database availability. +- 2026-08-29: PostgreSQL reserves schema names beginning with `pg_`, so the renamed project cannot default to `pg_queue`. The default is `pgqueue`, and validation rejects the reserved prefix before migration. - 2026-08-29: Phase 3 restored node-resque's optional `QueueOptions.queue` field. Queue methods still take an explicit queue name, but retaining the constructor field lets existing typed call sites migrate without an excess-property error. diff --git a/docs/plans/10-publish-and-ci.md b/docs/plans/10-publish-and-ci.md index 4a659a2..5e05cc7 100644 --- a/docs/plans/10-publish-and-ci.md +++ b/docs/plans/10-publish-and-ci.md @@ -107,7 +107,7 @@ Do not publish `0.0.1` empty stubs. First intentional bump to `0.1.0` is the fir Port node-resque `examples/` in Phases 4–7. This phase can add a compose-based example command if missing: ```bash -DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pg_queue_test bun examples/example.ts +DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test bun examples/example.ts ``` Optional: `examples/docker` like node-resque — not required for v1. diff --git a/src/core/connection.ts b/src/core/connection.ts index f59f228..d3b1fe6 100644 --- a/src/core/connection.ts +++ b/src/core/connection.ts @@ -7,7 +7,7 @@ import { type QueryResultRow, } from "pg"; -const DEFAULT_SCHEMA = "pg_queue"; +const DEFAULT_SCHEMA = "pgqueue"; const DEFAULT_APPLICATION_NAME = "pg-queue"; const SCHEMA_PATTERN = /^[a-zA-Z_][a-zA-Z0-9_]*$/; const LEADER_SLOT = "default"; @@ -55,7 +55,7 @@ export interface ConnectionOptions { * Analogous to passing `redis: ioredisInstance`. */ pool?: Pool; - /** Queue schema. Default `pg_queue`; must be a legal SQL identifier. */ + /** Queue schema. Default `pgqueue`; must be a legal SQL identifier. */ schema?: string; /** Reported to Postgres as `application_name`. Default `pg-queue`. */ application_name?: string; @@ -601,6 +601,11 @@ export function assertSchema(schema: string): void { if (!SCHEMA_PATTERN.test(schema)) { throw new Error(`Invalid schema "${schema}": must match ${SCHEMA_PATTERN}`); } + if (schema.toLowerCase().startsWith("pg_")) { + throw new Error( + `Invalid schema "${schema}": PostgreSQL reserves the "pg_" prefix`, + ); + } } function rejectRedisOptions(options: ConnectionOptions): void { From e6b9e241399d6c6181eae53bd7c5da5a587bba3e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 29 Aug 2026 06:43:29 +0000 Subject: [PATCH 3/4] Retry enqueue when delQueue races the queue row Restore a single retry after FK 23503 so ensureQueue plus job insert survives a concurrent queue delete. Co-authored-by: Evan Tahler --- __tests__/core/queue.test.ts | 35 ++++++++++++++++ docs/plans/03-queue.md | 1 + src/core/queue.ts | 78 ++++++++++++++++++++++++++---------- 3 files changed, 92 insertions(+), 22 deletions(-) diff --git a/__tests__/core/queue.test.ts b/__tests__/core/queue.test.ts index a143071..e235474 100644 --- a/__tests__/core/queue.test.ts +++ b/__tests__/core/queue.test.ts @@ -549,6 +549,41 @@ describe("queue", () => { await other.end(); }); + test("retries enqueue when the queue row is deleted between ensure and insert", async () => { + const originalQuery = queue.connection.query.bind(queue.connection); + let sabotaged = false; + queue.connection.query = (async ( + text: string, + values: unknown[] = [], + ) => { + if ( + !sabotaged && + text.includes("INSERT INTO") && + text.includes("pgrq_jobs") && + values[0] === "racy" + ) { + sabotaged = true; + await originalQuery( + `DELETE FROM ${specHelper.schema}.pgrq_jobs WHERE name = $1`, + ["racy"], + ); + await originalQuery( + `DELETE FROM ${specHelper.schema}.pgrq_queues WHERE name = $1`, + ["racy"], + ); + } + return originalQuery(text, values); + }) as typeof queue.connection.query; + + try { + expect(await queue.enqueue("racy", "job", [1])).toBe(true); + expect(await queue.length("racy")).toBe(1); + expect(sabotaged).toBe(true); + } finally { + queue.connection.query = originalQuery; + } + }); + test("queue row locking serializes enqueue with queue deletion", async () => { const other = new Queue({ connection: specHelper.cleanConnectionDetails(), diff --git a/docs/plans/03-queue.md b/docs/plans/03-queue.md index e0b94c8..1ee5f6c 100644 --- a/docs/plans/03-queue.md +++ b/docs/plans/03-queue.md @@ -139,3 +139,4 @@ Recommended split: - 2026-08-29: Bugbot: `delQueue` rebuilt those keys from jsonb-loaded args, whose object key order can differ from `JSON.stringify` at enqueue time. The hash now canonicalizes nested object keys so delete and re-enqueue agree. - 2026-08-29: Coverage audit found untested `del(count)`, `delByFunction(start, stop)`, expired-lock cleanup, concurrent delayed enqueue, queue-row serialization, `cleanOldWorkers`, `retryStuckJobs`, active `workingOn`, and unknown-worker errors. These now have focused PostgreSQL tests rather than being deferred wholesale to Phase 4. - 2026-08-29: Removed pg-boss before Worker implementation. Queue registration is now an idempotent `pgrq_queues` insert, enqueue writes `pgrq_jobs`, and test dequeues use the same atomic `Connection.fetchJob()` primitive planned for Worker. +- 2026-08-29: Bugbot: `sendJob` / failed-job insert can lose a race with `delQueue` between `ensureQueue` and the `pgrq_jobs` write (FK 23503). Those writes now retry once after recreating the queue row. diff --git a/src/core/queue.ts b/src/core/queue.ts index 74a9fde..81dd34e 100644 --- a/src/core/queue.ts +++ b/src/core/queue.ts @@ -686,17 +686,18 @@ export class Queue extends EventEmitter { if ((updated.rowCount ?? 0) > 0) return; if (working.id) return; - await this.ensureQueue(working.queue); - await this.connection.query( - `INSERT INTO ${this.connection.schema}.pgrq_jobs - (name, data, state, completed_on, output) - VALUES ($1, $2::jsonb, 'failed', now(), $3::jsonb)`, - [ - working.queue, - JSON.stringify(working.payload), - JSON.stringify(errorPayload), - ], - ); + await this.withQueue(working.queue, async () => { + await this.connection.query( + `INSERT INTO ${this.connection.schema}.pgrq_jobs + (name, data, state, completed_on, output) + VALUES ($1, $2::jsonb, 'failed', now(), $3::jsonb)`, + [ + working.queue, + JSON.stringify(working.payload), + JSON.stringify(errorPayload), + ], + ); + }); } /** @@ -852,17 +853,41 @@ export class Queue extends EventEmitter { payload: ParsedJob, options: { startAfter?: Date } = {}, ): Promise { - await this.ensureQueue(q); - const result = await this.connection.query<{ id: string }>( - `INSERT INTO ${this.connection.schema}.pgrq_jobs - (name, data, start_after) - VALUES ($1, $2::jsonb, $3) - RETURNING id`, - [q, JSON.stringify(payload), options.startAfter ?? new Date()], - ); - const id = result.rows[0]?.id; - if (!id) throw new Error(`Failed to enqueue job on queue "${q}"`); - return id; + return this.withQueue(q, async () => { + const result = await this.connection.query<{ id: string }>( + `INSERT INTO ${this.connection.schema}.pgrq_jobs + (name, data, start_after) + VALUES ($1, $2::jsonb, $3) + RETURNING id`, + [q, JSON.stringify(payload), options.startAfter ?? new Date()], + ); + const id = result.rows[0]?.id; + if (!id) throw new Error(`Failed to enqueue job on queue "${q}"`); + return id; + }); + } + + /** + * Recreate the queue row, then run `work`. Retry once if a concurrent + * `delQueue` removed the parent between those statements (FK 23503). + * + * @param q - Queue name. + * @param work - Job-table mutation that requires `pgrq_queues.name`. + * @returns Result of `work`. + */ + private async withQueue(q: string, work: () => Promise): Promise { + let lastError: Error | undefined; + for (let attempt = 0; attempt < 2; attempt += 1) { + await this.ensureQueue(q); + try { + return await work(); + } catch (error) { + lastError = toError(error); + if (attempt === 0 && isMissingQueueFk(error)) continue; + throw lastError; + } + } + throw lastError ?? new Error(`Failed to write job on queue "${q}"`); } private async ensureQueue(q: string): Promise { @@ -1025,3 +1050,12 @@ function sqlRange( function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } + +function isMissingQueueFk(error: unknown): boolean { + if (!isRecord(error)) return false; + return error.code === "23503" && error.table === "pgrq_jobs"; +} + +function toError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} From 42122c7b06d186b20c20dbd77517190413133f05 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 29 Aug 2026 06:55:47 +0000 Subject: [PATCH 4/4] Use database time for enqueue and tear down failed migrate Immediate jobs now COALESCE start_after to now(), and migrate() reuses connect() so a failed bootstrap cannot leak an owned pool. Co-authored-by: Evan Tahler --- __tests__/core/connectionError.test.ts | 20 ++++++++++++++++++++ __tests__/core/queue.test.ts | 9 +++++++++ docs/plans/02-connection-and-schema.md | 1 + docs/plans/03-queue.md | 1 + src/core/connection.ts | 8 ++------ src/core/queue.ts | 4 ++-- 6 files changed, 35 insertions(+), 8 deletions(-) diff --git a/__tests__/core/connectionError.test.ts b/__tests__/core/connectionError.test.ts index 95c9d23..dc8b8e0 100644 --- a/__tests__/core/connectionError.test.ts +++ b/__tests__/core/connectionError.test.ts @@ -40,4 +40,24 @@ describe("connection error", () => { }); }); }, 60_000); + + test("migrate() tears down a failed bootstrap pool", async () => { + const brokenConnection = new Connection({ + host: "127.0.0.1", + port: 1, + database: "pgqueue_test", + user: "postgres", + password: "postgres", + schema: "pgqueue_test", + }); + brokenConnection.on("error", () => { + // connect() also emits; the rejection is the assertion + }); + + await expect(brokenConnection.migrate()).rejects.toThrow( + /ECONNREFUSED|ENOTFOUND|ETIMEDOUT|EAI_AGAIN|connect/i, + ); + expect(brokenConnection.connected).toBe(false); + expect(() => brokenConnection.pool).toThrow("Connection is not connected"); + }, 60_000); }); diff --git a/__tests__/core/queue.test.ts b/__tests__/core/queue.test.ts index e235474..fd8fd53 100644 --- a/__tests__/core/queue.test.ts +++ b/__tests__/core/queue.test.ts @@ -85,6 +85,15 @@ describe("queue", () => { expect(await queue.enqueue(specHelper.queue, "someJob", [1, 2, 3])).toBe( true, ); + const timing = await queue.connection.query<{ skew_ms: string }>( + `SELECT (extract(epoch from (start_after - now())) * 1000)::text AS skew_ms + FROM ${specHelper.schema}.pgrq_jobs + WHERE name = $1`, + [specHelper.queue], + ); + expect(Math.abs(Number(timing.rows[0]?.skew_ms ?? 9999))).toBeLessThan( + 1000, + ); const raw = await specHelper.popFromQueue(); expect(raw).not.toBeNull(); const job = JSON.parse(raw ?? "{}") as ParsedJob; diff --git a/docs/plans/02-connection-and-schema.md b/docs/plans/02-connection-and-schema.md index b1961d9..5cbafe2 100644 --- a/docs/plans/02-connection-and-schema.md +++ b/docs/plans/02-connection-and-schema.md @@ -249,4 +249,5 @@ Do not defer these to Phase 8. - 2026-08-29: Removed pg-boss after Phase 3 confirmed that its lifecycle and schema added coupling without supplying the resque runtime. Migrations now ship as numbered SQL files, execute under an advisory lock in one transaction, and are included in the npm package. - 2026-08-29: `connect()` must execute `SELECT 1`; constructing `pg.Pool` is lazy and does not prove credentials, routing, or database availability. - 2026-08-29: PostgreSQL reserves schema names beginning with `pg_`, so the renamed project cannot default to `pg_queue`. The default is `pgqueue`, and validation rejects the reserved prefix before migration. +- 2026-08-29: Bugbot: `migrate()` now calls `connect()`, so a failed bootstrap `SELECT 1` tears down the owned pool and error listener instead of leaking them with `connected === false`. `connect()` tears down before emitting `error` so a missing listener cannot skip cleanup. - 2026-08-29: Phase 3 restored node-resque's optional `QueueOptions.queue` field. Queue methods still take an explicit queue name, but retaining the constructor field lets existing typed call sites migrate without an excess-property error. diff --git a/docs/plans/03-queue.md b/docs/plans/03-queue.md index 1ee5f6c..7344d23 100644 --- a/docs/plans/03-queue.md +++ b/docs/plans/03-queue.md @@ -140,3 +140,4 @@ Recommended split: - 2026-08-29: Coverage audit found untested `del(count)`, `delByFunction(start, stop)`, expired-lock cleanup, concurrent delayed enqueue, queue-row serialization, `cleanOldWorkers`, `retryStuckJobs`, active `workingOn`, and unknown-worker errors. These now have focused PostgreSQL tests rather than being deferred wholesale to Phase 4. - 2026-08-29: Removed pg-boss before Worker implementation. Queue registration is now an idempotent `pgrq_queues` insert, enqueue writes `pgrq_jobs`, and test dequeues use the same atomic `Connection.fetchJob()` primitive planned for Worker. - 2026-08-29: Bugbot: `sendJob` / failed-job insert can lose a race with `delQueue` between `ensureQueue` and the `pgrq_jobs` write (FK 23503). Those writes now retry once after recreating the queue row. +- 2026-08-29: Bugbot: immediate enqueue uses `COALESCE($start_after, now())` so eligibility is compared against PostgreSQL time, matching `fetchJob` / `length` / `queued`. Delayed jobs still pass an explicit timestamp. diff --git a/src/core/connection.ts b/src/core/connection.ts index d3b1fe6..c5a5e67 100644 --- a/src/core/connection.ts +++ b/src/core/connection.ts @@ -188,8 +188,8 @@ export class Connection extends EventEmitter { this.connected = true; } catch (error) { const err = toError(error); - this.emit("error", err); await this.teardownPartialConnect(); + this.emit("error", err); throw err; } } @@ -292,11 +292,7 @@ export class Connection extends EventEmitter { * @throws If the pool cannot be opened or migration SQL fails. */ async migrate(): Promise { - if (!this._pool) { - this.ensurePool(); - await this.pool.query("SELECT 1"); - this.connected = true; - } + await this.connect(); const schema = this._schema; const client = await this.pool.connect(); diff --git a/src/core/queue.ts b/src/core/queue.ts index 81dd34e..6177e55 100644 --- a/src/core/queue.ts +++ b/src/core/queue.ts @@ -857,9 +857,9 @@ export class Queue extends EventEmitter { const result = await this.connection.query<{ id: string }>( `INSERT INTO ${this.connection.schema}.pgrq_jobs (name, data, start_after) - VALUES ($1, $2::jsonb, $3) + VALUES ($1, $2::jsonb, COALESCE($3::timestamptz, now())) RETURNING id`, - [q, JSON.stringify(payload), options.startAfter ?? new Date()], + [q, JSON.stringify(payload), options.startAfter ?? null], ); const id = result.rows[0]?.id; if (!id) throw new Error(`Failed to enqueue job on queue "${q}"`);