Batched job processing (opt-in) (#474)

Author: benjie

__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),

__tests__/schema.sql

@@ -374,5 +374,25 @@ ALTER TABLE graphile_worker._private_job_queues ENABLE ROW LEVEL SECURITY;
 ALTER TABLE graphile_worker._private_jobs ENABLE ROW LEVEL SECURITY;
 ALTER TABLE graphile_worker._private_known_crontabs ENABLE ROW LEVEL SECURITY;
 ALTER TABLE graphile_worker._private_tasks ENABLE ROW LEVEL SECURITY;
-REVOKE USAGE ON SCHEMA public FROM PUBLIC;
-GRANT ALL ON SCHEMA public TO PUBLIC;
+SELECT pg_catalog.set_config('search_path', '', false);
+COPY graphile_worker.migrations (id, ts, breaking) FROM stdin;
+1	1970-01-01 00:00:00.000000+00	t
+2	1970-01-01 00:00:00.000000+00	f
+3	1970-01-01 00:00:00.000000+00	t
+4	1970-01-01 00:00:00.000000+00	f
+5	1970-01-01 00:00:00.000000+00	f
+6	1970-01-01 00:00:00.000000+00	f
+7	1970-01-01 00:00:00.000000+00	f
+8	1970-01-01 00:00:00.000000+00	f
+9	1970-01-01 00:00:00.000000+00	f
+10	1970-01-01 00:00:00.000000+00	f
+11	1970-01-01 00:00:00.000000+00	t
+12	1970-01-01 00:00:00.000000+00	f
+13	1970-01-01 00:00:00.000000+00	t
+14	1970-01-01 00:00:00.000000+00	t
+15	1970-01-01 00:00:00.000000+00	f
+16	1970-01-01 00:00:00.000000+00	t
+17	1970-01-01 00:00:00.000000+00	f
+18	1970-01-01 00:00:00.000000+00	f
+19	1970-01-01 00:00:00.000000+00	t
+\.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "graphile-worker",
-  "version": "0.16.6",
+  "version": "0.17.0-canary.1fcb2a0",
   "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",
@@ -51,11 +52,10 @@
   "homepage": "https://github.com/graphile/worker#readme",
   "dependencies": {
     "@graphile/logger": "^0.2.0",
-    "@tsconfig/node18": "^18.2.4",
     "@types/debug": "^4.1.10",
     "@types/pg": "^8.10.5",
     "cosmiconfig": "^8.3.6",
-    "graphile-config": "^0.0.1-beta.11",
+    "graphile-config": "^0.0.1-beta.16",
     "json5": "^2.2.3",
     "pg": "^8.11.3",
     "tslib": "^2.6.2",
@@ -72,6 +72,7 @@
     "@fortawesome/free-solid-svg-icons": "^6.5.1",
     "@fortawesome/react-fontawesome": "^0.2.0",
     "@mdx-js/react": "^1.6.22",
+    "@tsconfig/node18": "^18.2.4",
     "@types/jest": "^26.0.0",
     "@types/json5": "^2.2.0",
     "@types/node": "^20.8.7",
@@ -85,7 +86,7 @@
     "eslint-plugin-jest": "^26.0.0",
     "eslint-plugin-simple-import-sort": "^10.0.0",
     "eslint_d": "^13.0.0",
-    "graphile": "^5.0.0-beta.16",
+    "graphile": "^5.0.0-beta.41",
     "jest": "^26.0.0",
     "jest-time-helpers": "0.1.1",
     "juice": "5.2.0",

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);
   });

scripts/dump_db

@@ -6,6 +6,7 @@ dropuser graphile_worker_role || true
 psql template1 -c "CREATE USER graphile_worker_role WITH SUPERUSER PASSWORD 'password';"
 createdb graphile_worker_dump -O graphile_worker_role
 PGUSER=graphile_worker_role PGPASSWORD=password PGHOST=127.0.0.1 ts-node src/cli.ts -c postgres:///graphile_worker_dump --schema-only
-pg_dump --schema-only --no-owner graphile_worker_dump | sed -e '/^--/d' -e '/^\s*$/d' -e '/^SET /d' -e 's/EXECUTE FUNCTION/EXECUTE PROCEDURE/g' > __tests__/schema.sql
+pg_dump --schema-only --no-owner graphile_worker_dump | sed -E -e '/^--/d' -e '/^\s*$/d' -e '/^SET /d' -e 's/EXECUTE FUNCTION/EXECUTE PROCEDURE/g' -e '/^(REVOKE|GRANT) .* ON SCHEMA public (FROM|TO) PUBLIC;$/d' > __tests__/schema.sql
+pg_dump --data-only --no-owner graphile_worker_dump --table=graphile_worker.migrations --table=graphile_worker._private_pro_migrations | sed -E -e '/^--/d' -e '/^\s*$/d' -e 's/\b2[0-9]{3}-[0-9]{2}-[0-9]{2}\s[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{1,6}\+[0-9]+/1970-01-01 00:00:00.000000+00/g'  -e '/^SET /d' >> __tests__/schema.sql
 dropdb graphile_worker_dump
 dropuser graphile_worker_role

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/cron.ts

@@ -179,6 +179,7 @@ async function scheduleCronJobs(
  * performs backfilling on any crontab tasks that need it.
  */
 async function registerAndBackfillItems(
+  ctx: CompiledSharedOptions,
   { pgPool, events, cron }: { pgPool: Pool; events: WorkerEvents; cron: Cron },
   escapedWorkerSchema: string,
   parsedCronItems: ParsedCronItem[],
@@ -261,6 +262,7 @@ async function registerAndBackfillItems(
         // At this time it's not expected that backfilling will be sufficiently
         // expensive to justify optimising this further.
         events.emit("cron:backfill", {
+          ctx,
           cron,
           itemsToBackfill,
           timestamp: ts,
@@ -338,19 +340,21 @@ export const runCron = (
     }
 
     const start = new Date();
-    events.emit("cron:starting", { cron, start });
+    const ctx = compiledSharedOptions;
+    events.emit("cron:starting", { ctx, cron, start });
 
     // We must backfill BEFORE scheduling any new jobs otherwise backfill won't
     // work due to known_crontabs.last_execution having been updated.
     await registerAndBackfillItems(
+      ctx,
       { pgPool, events, cron },
       escapedWorkerSchema,
       parsedCronItems,
       new Date(+start),
       useNodeTime,
     );
 
-    events.emit("cron:started", { cron, start });
+    events.emit("cron:started", { ctx, cron, start });
 
     if (!cron._active) {
       return stop();
@@ -411,6 +415,7 @@ export const runCron = (
             },
           );
           events.emit("cron:prematureTimer", {
+            ctx,
             cron,
             currentTimestamp,
             expectedTimestamp,
@@ -427,6 +432,7 @@ export const runCron = (
             )}s behind)`,
           );
           events.emit("cron:overdueTimer", {
+            ctx,
             cron,
             currentTimestamp,
             expectedTimestamp,
@@ -449,6 +455,7 @@ export const runCron = (
         // Finally actually run the jobs.
         if (jobsAndIdentifiers.length) {
           events.emit("cron:schedule", {
+            ctx,
             cron,
             timestamp: expectedTimestamp,
             jobsAndIdentifiers,
@@ -461,6 +468,7 @@ export const runCron = (
             useNodeTime,
           );
           events.emit("cron:scheduled", {
+            ctx,
             cron,
             timestamp: expectedTimestamp,
             jobsAndIdentifiers,