Batched job processing (opt-in)

__tests__/migrate.test.ts

@@ -14,7 +14,7 @@ import {
 
 const options: WorkerSharedOptions = {};
 
-const MAX_MIGRATION_NUMBER = 18;
+const MAX_MIGRATION_NUMBER = 19;
 
 test("migration installs schema; second migration does no harm", async () => {
   await withPgClient(async (pgClient) => {
@@ -238,6 +238,7 @@ test("throws helpful error message in migration 11", async () => {
 
     // Manually run the first 10 migrations
     const event = {
+      ctx: compiledSharedOptions,
       client: pgClient,
       postgresVersion: 120000, // TODO: use the actual postgres version
       scratchpad: Object.create(null),

package.json

@@ -1,6 +1,6 @@
 {
   "name": "graphile-worker",
-  "version": "0.16.6",
+  "version": "0.17.0-canary.9817f67",
   "type": "commonjs",
   "description": "Job queue for PostgreSQL",
   "main": "dist/index.js",
@@ -18,6 +18,7 @@
     "depcheck": "depcheck --ignores='graphile-worker,faktory-worker,@google-cloud/tasks,bullmq,jest-environment-node,@docusaurus/*,@fortawesome/*,@mdx-js/*,@types/jest,clsx,eslint_d,graphile,juice,postcss-nested,prism-react-renderer,react,react-dom,svgo,ts-node,@types/debug,tslib'",
     "db:dump": "./scripts/dump_db",
     "perfTest": "cd perfTest && node ./run.js",
+    "towerDefence": "cd towerDefence && node ./run.mjs",
     "preversion": "grep '^### Pending' RELEASE_NOTES.md && echo \"⚠️ Cannot publish with 'Pending' in RELEASE_NOTES ⚠️\" && exit 1 || true",
     "version": "node scripts/postversion.mjs && git add src/version.ts",
     "website": "cd website && yarn run"

perfTest/graphile.config.js

@@ -5,16 +5,29 @@
 
 // import { WorkerProPreset } from "../graphile-pro-worker/dist/index.js";
 
+const CONCURRENT_JOBS = 24;
+
 /** @type {GraphileConfig.Preset} */
 const preset = {
   // extends: [WorkerProPreset],
   worker: {
     connectionString:
       process.env.PERF_DATABASE_URL || "postgres:///graphile_worker_perftest",
-    concurrentJobs: 3,
     fileExtensions: [".js", ".cjs", ".mjs"],
     // fileExtensions: [".js", ".cjs", ".mjs", ".ts", ".cts", ".mts"],
     gracefulShutdownAbortTimeout: 2500,
+
+    concurrentJobs: CONCURRENT_JOBS,
+    maxPoolSize: CONCURRENT_JOBS + 1,
+
+    //localQueue: { size: -1 },
+    //completeJobBatchDelay: -1,
+    //failJobBatchDelay: -1,
+
+    localQueue: { size: 500, refetchDelay: { durationMs: 10 } },
+    completeJobBatchDelay: 0,
+    failJobBatchDelay: 0,
   },
 };
+
 module.exports = preset;

perfTest/init.js

@@ -36,21 +36,19 @@ $$ language plpgsql;`,
   } else {
     const jobs = [];
     for (let i = 0; i < jobCount; i++) {
-      jobs.push(
-        `("${taskIdentifier.replace(
-          /["\\]/g,
-          "\\$&",
-        )}","{\\"id\\":${i}}",,,,,,)`,
+      jobs.push({ identifier: taskIdentifier, payload: { id: i } });
+    }
+    console.time(`Adding jobs`);
+    while (jobs.length > 0) {
+      const jobsSlice = jobs.splice(0, 1000000);
+      const jobsString = JSON.stringify(jobsSlice);
+      console.log(`Adding ${jobsSlice.length} jobs`);
+      await pgPool.query(
+        `select 1 from graphile_worker.add_jobs(array(select json_populate_recordset(null::graphile_worker.job_spec, $1::json)));`,
+        [jobsString],
       );
+      console.log(`...added`);
     }
-    const jobsString = `{"${jobs
-      .map((j) => j.replace(/["\\]/g, "\\$&"))
-      .join('","')}"}`;
-    console.time("Adding jobs");
-    await pgPool.query(
-      `select graphile_worker.add_jobs($1::graphile_worker.job_spec[]);`,
-      [jobsString],
-    );
     console.timeEnd("Adding jobs");
   }
 

perfTest/latencyTest.js

@@ -9,10 +9,7 @@ const preset = require("./graphile.config.js");
 const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
 
 /** @type {import('../dist/index.js').WorkerPoolOptions} */
-const options = {
-  concurrency: 1,
-  preset,
-};
+const options = { preset };
 
 async function main() {
   const pgPool = new Pool({ connectionString: process.env.PERF_DATABASE_URL });

perfTest/run.js

@@ -3,10 +3,9 @@ const { execSync, exec: rawExec } = require("child_process");
 const { promisify } = require("util");
 const exec = promisify(rawExec);
 
-const JOB_COUNT = 20000;
+const JOB_COUNT = 200000;
 const STUCK_JOB_COUNT = 0;
 const PARALLELISM = 4;
-const CONCURRENCY = 10;
 
 const time = async (cb) => {
   const start = process.hrtime();
@@ -52,10 +51,7 @@ async function main() {
   console.log("Timing startup/shutdown time...");
   let result;
   const startupTime = await time(async () => {
-    result = await exec(
-      `node ../dist/cli.js --once -j ${CONCURRENCY} -m ${CONCURRENCY + 1}`,
-      execOptions,
-    );
+    result = await exec(`node ../dist/cli.js --once`, execOptions);
   });
   logResult(result);
   console.log();
@@ -81,12 +77,7 @@ async function main() {
   const dur = await time(async () => {
     const promises = [];
     for (let i = 0; i < PARALLELISM; i++) {
-      promises.push(
-        exec(
-          `node ../dist/cli.js --once -j ${CONCURRENCY} -m ${CONCURRENCY + 1}`,
-          execOptions,
-        ),
-      );
+      promises.push(exec(`node ../dist/cli.js --once`, execOptions));
     }
     (await Promise.all(promises)).map(logResult);
   });

sql/000019.sql

@@ -0,0 +1,4 @@
+--! breaking-change
+-- This is just a breaking change marker for the v0.17 worker-centric to
+-- pool-centric jump. The migration itself is not breaking.
+select 1;

src/config.ts

@@ -1,6 +1,8 @@
 import { cosmiconfigSync } from "cosmiconfig";
+import EventEmitter from "events";
 
 import { MINUTE, SECOND } from "./cronConstants";
+import type { WorkerEvents } from "./interfaces";
 import { defaultLogger } from "./logger";
 
 const cosmiconfigResult = cosmiconfigSync("graphile-worker").search();
@@ -34,6 +36,7 @@ export const makeWorkerPresetWorkerOptions = () =>
     maxResetLockedInterval: 10 * MINUTE,
     gracefulShutdownAbortTimeout: 5 * SECOND,
     useNodeTime: false,
+    events: new EventEmitter() as WorkerEvents,
   } satisfies GraphileConfig.WorkerOptions);
 
 function enforceStringOrUndefined(