Batched job processing (opt-in)

.github/workflows/ci.yml

@@ -40,7 +40,9 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
       - run: yarn --frozen-lockfile
-      - run: yarn node --experimental-vm-modules node_modules/.bin/jest -i --ci
+      - run: yarn prepack
+      - run: yarn test:setupdb
+      - run: yarn test:only --ci
 
   lint:
     runs-on: ubuntu-latest
@@ -89,8 +91,9 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
       - run: yarn --frozen-lockfile
-      # - run: yarn lint # No need to lint altschema
-      - run: yarn node --experimental-vm-modules node_modules/.bin/jest -i --ci
+      - run: yarn prepack
+      - run: yarn test:setupdb
+      - run: yarn test:only --ci
 
   database_updated:
     runs-on: ubuntu-latest

.gitignore

@@ -1,3 +1,4 @@
+/.git
 dist
 npm-debug.log*
 yarn-debug.log*

__tests__/events.test.ts

@@ -6,7 +6,7 @@ import deferred, { Deferred } from "../src/deferred";
 import { Task, TaskList, WorkerSharedOptions } from "../src/interfaces";
 import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
-  jobCount,
+  expectJobCount,
   reset,
   sleep,
   sleepUntil,
@@ -132,7 +132,7 @@ test("emits the expected events", () =>
       expect(eventCount("worker:release")).toEqual(CONCURRENCY);
       expect(eventCount("worker:stop")).toEqual(CONCURRENCY);
       expect(eventCount("pool:release")).toEqual(1);
-      expect(await jobCount(pgPool)).toEqual(0);
+      await expectJobCount(pgPool, 0);
     } finally {
       Object.values(jobPromises).forEach((p) => p?.resolve());
     }

__tests__/forbiddenFlags.test.ts

@@ -5,20 +5,14 @@ import {
   TaskList,
   WorkerSharedOptions,
 } from "../src/index";
-import {
-  getJobs,
-  reset,
-  TEST_CONNECTION_STRING,
-  withPgClient,
-  withPgPool,
-} from "./helpers";
+import { getJobs, reset, withPgClient, withPgPool } from "./helpers";
 
 const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
 const options: WorkerSharedOptions = {};
 
 test("supports the flags API", () =>
-  withPgClient(async (pgClient) => {
+  withPgClient(async (pgClient, { TEST_CONNECTION_STRING }) => {
     await reset(pgClient, options);
 
     // Schedule a job

__tests__/helpers.ts

@@ -1,6 +1,7 @@
+import { randomBytes } from "crypto";
 import { EventEmitter } from "events";
+import { sleep, sleepUntil as baseSleepUntil } from "jest-time-helpers";
 import * as pg from "pg";
-import { parse } from "pg-connection-string";
 
 import defer from "../src/deferred";
 import {
@@ -32,23 +33,62 @@ export {
   SECOND,
   setupFakeTimers,
   sleep,
-  sleepUntil,
   WEEK,
 } from "jest-time-helpers";
 
+export function sleepUntil(condition: () => boolean, ms?: number) {
+  // Bump the default timeout from 2000ms for CI
+  return baseSleepUntil(condition, ms ?? 5000);
+}
+
 // Sometimes CI's clock can get interrupted (it is shared infra!) so this
 // extends the default timeout just in case.
-jest.setTimeout(15000);
+jest.setTimeout(20000);
 
 // process.env.GRAPHILE_LOGGER_DEBUG = "1";
 
-export const TEST_CONNECTION_STRING =
-  process.env.TEST_CONNECTION_STRING || "postgres:///graphile_worker_test";
+async function createTestDatabase() {
+  const id = randomBytes(8).toString("hex");
+  const PGDATABASE = `graphile_worker_test_${id}`;
+  {
+    const client = new pg.Client({ connectionString: `postgres:///template1` });
+    await client.connect();
+    await client.query(
+      `create database ${pg.Client.prototype.escapeIdentifier(
+        PGDATABASE,
+      )} with template = graphile_worker_testtemplate;`,
+    );
+    await client.end();
+  }
+  const TEST_CONNECTION_STRING = `postgres:///${PGDATABASE}`;
+  const PGHOST = process.env.PGHOST;
+  async function release() {
+    const client = new pg.Client({ connectionString: `postgres:///template1` });
+    await client.connect();
+    await client.query(
+      `drop database ${pg.Client.prototype.escapeIdentifier(PGDATABASE)};`,
+    );
+    await client.end();
+  }
+
+  return {
+    TEST_CONNECTION_STRING,
+    PGHOST,
+    PGDATABASE,
+    release,
+  };
+}
 
-const parsed = parse(TEST_CONNECTION_STRING);
+export let databaseDetails: Awaited<
+  ReturnType<typeof createTestDatabase>
+> | null = null;
 
-export const PGHOST = parsed.host || process.env.PGHOST;
-export const PGDATABASE = parsed.database || undefined;
+beforeAll(async () => {
+  databaseDetails = await createTestDatabase();
+});
+afterAll(async () => {
+  databaseDetails?.release();
+});
 
 export const GRAPHILE_WORKER_SCHEMA =
   process.env.GRAPHILE_WORKER_SCHEMA || "graphile_worker";
@@ -58,6 +98,7 @@ export const ESCAPED_GRAPHILE_WORKER_SCHEMA =
 export async function withPgPool<T>(
   cb: (pool: pg.Pool) => Promise<T>,
 ): Promise<T> {
+  const { TEST_CONNECTION_STRING } = databaseDetails!;
   const pool = new pg.Pool({
     connectionString: TEST_CONNECTION_STRING,
   });
@@ -75,12 +116,17 @@ afterEach(() => {
 });
 
 export async function withPgClient<T>(
-  cb: (client: pg.PoolClient) => Promise<T>,
+  cb: (
+    client: pg.PoolClient,
+    extra: {
+      TEST_CONNECTION_STRING: string;
+    },
+  ) => Promise<T>,
 ): Promise<T> {
   return withPgPool(async (pool) => {
     const client = await pool.connect();
     try {
-      return await cb(client);
+      return await cb(client, databaseDetails!);
     } finally {
       client.release();
     }
@@ -125,7 +171,18 @@ export async function reset(
   }
 }
 
-export async function jobCount(
+/**
+ * Counts the number of jobs currently in DB.
+ *
+ * If you have a pool, you may hit race conditions with this method, instead
+ * use `expectJobCount()` which will try multiple times to give time for
+ * multiple clients to synchronize.
+ */
+export async function jobCount(pgClient: pg.PoolClient): Promise<number> {
+  return _jobCount(pgClient);
+}
+
+async function _jobCount(
   pgPoolOrClient: pg.Pool | pg.PoolClient,
 ): Promise<number> {
   const {
@@ -296,3 +353,28 @@ export function withOptions<T>(
     }),
   );
 }
+
+/**
+ * Wait for the job count to match the expected count, handles
+ * issues with different connections to the database not
+ * reflecting the same data by retrying.
+ */
+export async function expectJobCount(
+  // NOTE: if you have a pgClient then you shouldn't need to
+  // use this - just call `jobCount()` directly since you're
+  // in the same client
+  pool: pg.Pool,
+  expectedCount: number,
+) {
+  let count: number = Infinity;
+  for (let i = 0; i < 8; i++) {
+    if (i > 0) {
+      await sleep(i * 50);
+    }
+    count = await _jobCount(pool);
+    if (count === expectedCount) {
+      break;
+    }
+  }
+  expect(count).toEqual(expectedCount);
+}

__tests__/jobsView.test.ts

@@ -3,14 +3,13 @@ import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
   makeSelectionOfJobs,
   reset,
-  TEST_CONNECTION_STRING,
   withPgClient,
 } from "./helpers";
 
 const options: WorkerSharedOptions = {};
 
 test("jobs view renders jobs", () =>
-  withPgClient(async (pgClient) => {
+  withPgClient(async (pgClient, { TEST_CONNECTION_STRING }) => {
     await reset(pgClient, options);
 
     const utils = await makeWorkerUtils({

__tests__/main.runTaskList.test.ts

@@ -6,7 +6,7 @@ import { Task, TaskList, WorkerSharedOptions } from "../src/interfaces";
 import { runTaskList } from "../src/main";
 import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
-  jobCount,
+  expectJobCount,
   reset,
   sleep,
   sleepUntil,
@@ -70,7 +70,7 @@ test("main will execute jobs as they come up, and exits cleanly", () =>
       await sleep(1);
       expect(finished).toBeTruthy();
       await workerPool.promise;
-      expect(await jobCount(pgPool)).toEqual(0);
+      await expectJobCount(pgPool, 0);
       expect(process.listeners("SIGTERM")).toHaveLength(0);
     } finally {
       Object.values(jobPromises).forEach((p) => p?.resolve());

__tests__/runner.helpers.getTaskName.test.ts

@@ -3,9 +3,9 @@ import { Pool, PoolClient } from "pg";
 import { DbJobSpec, Runner, RunnerOptions } from "../src/interfaces";
 import { run } from "../src/runner";
 import {
+  databaseDetails,
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
   sleepUntil,
-  TEST_CONNECTION_STRING,
   withPgClient,
 } from "./helpers";
 
@@ -15,7 +15,7 @@ let runner: Runner | null = null;
 const JOB_COUNT = 10;
 beforeAll(() => {
   pgPool = new Pool({
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
     max: JOB_COUNT * 2 + 5,
   });
 });

__tests__/runner.runOnce.test.ts

@@ -4,12 +4,7 @@ import { makeWorkerPresetWorkerOptions } from "../src/config";
 import { RunnerOptions } from "../src/interfaces";
 import { WorkerPreset } from "../src/preset";
 import { runOnce } from "../src/runner";
-import {
-  PGDATABASE,
-  PGHOST,
-  TEST_CONNECTION_STRING,
-  withPgPool,
-} from "./helpers";
+import { databaseDetails, withPgPool } from "./helpers";
 
 delete process.env.DATABASE_URL;
 delete process.env.PGDATABASE;
@@ -57,7 +52,7 @@ async function runOnceErrorAssertion(
 
 test("either a list of tasks or a existent task directory must be provided", async () => {
   const options: RunnerOptions = {
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
   };
   await runOnceErrorAssertion(
     options,
@@ -67,7 +62,7 @@ test("either a list of tasks or a existent task directory must be provided", asy
 
 test("taskList and taskDirectory cannot be provided a the same time", async () => {
   const options: RunnerOptions = {
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
     taskDirectory: "foo",
     taskList: { task: () => {} },
   };
@@ -90,7 +85,7 @@ test("at least a connectionString, a pgPool, the DATABASE_URL or PGDATABASE envv
 test("connectionString and a pgPool cannot provided a the same time", async () => {
   const options: RunnerOptions = {
     taskList: { task: () => {} },
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
     pgPool: new Pool(),
   };
   await runOnceErrorAssertion(
@@ -100,34 +95,42 @@ test("connectionString and a pgPool cannot provided a the same time", async () =
 });
 
 test("providing just a DATABASE_URL is possible", async () => {
-  return withEnv({ DATABASE_URL: TEST_CONNECTION_STRING }, async () => {
-    const options: RunnerOptions = {
-      taskList: { task: () => {} },
-    };
-    expect.assertions(0);
-    await runOnce(options);
-  });
+  return withEnv(
+    { DATABASE_URL: databaseDetails!.TEST_CONNECTION_STRING },
+    async () => {
+      const options: RunnerOptions = {
+        taskList: { task: () => {} },
+      };
+      expect.assertions(0);
+      await runOnce(options);
+    },
+  );
 });
 
 test("providing just PGHOST and PGDATABASE is possible", async () => {
-  return withEnv({ PGHOST, PGDATABASE }, async () => {
-    const options: RunnerOptions = {
-      taskList: { task: () => {} },
-    };
-    expect.assertions(0);
-    await runOnce(options);
-  });
+  return withEnv(
+    {
+      PGHOST: databaseDetails!.PGHOST,
+      PGDATABASE: databaseDetails!.PGDATABASE,
+    },
+    async () => {
+      const options: RunnerOptions = {
+        taskList: { task: () => {} },
+      };
+      expect.assertions(0);
+      await runOnce(options);
+    },
+  );
 });
 
 test("providing just a connectionString is possible", async () => {
   const options: RunnerOptions = {
     taskList: { task: () => {} },
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
   };
   expect.assertions(0);
   await runOnce(options);
 });
-
 test("providing just a pgPool is possible", async () =>
   withPgPool(async (pgPool) => {
     const options: RunnerOptions = {