Postgres World permits concurrent ownership of the same hook token

{
  "name": "workflow-postgres-hook-token-race-repro",
  "private": true,
  "type": "module",
  "scripts": {
    "repro": "node race.mjs"
  },
  "dependencies": {
    "@workflow/world-postgres": "5.0.0-beta.40",
    "pg": "8.20.0"
  }
}

import assert from "node:assert/strict";
import { fork } from "node:child_process";
import { once } from "node:events";
import { readFileSync } from "node:fs";

import { Pool } from "pg";

const database = new URL(process.env.WORKFLOW_POSTGRES_URL ?? "http://invalid");
if (
  !["127.0.0.1", "localhost", "[::1]"].includes(database.hostname) ||
  !database.pathname.startsWith("/b2_repro_")
) {
  throw new Error("Requires a dedicated local b2_repro_ database");
}
const processCount = 12;
const token = `hook-token-race-${Date.now()}`;
const children = [];
const results = [];
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 30_000);
const pool = new Pool({ connectionString: database.href, query_timeout: 5000 });
try {
  await Promise.all(
    Array.from({ length: processCount }, async () => {
      const worker = fork(new URL("race-worker.mjs", import.meta.url), [], {
        execArgv: [],
        stdio: ["ignore", "inherit", "inherit", "ipc"],
      });
      const exited = once(worker, "exit", { signal: controller.signal });
      // Observe exit immediately, including setup failure before the ready message.
      const ready = Promise.race([
        once(worker, "message", { signal: controller.signal }),
        exited.then(() => {
          throw new Error("Worker exited before becoming ready");
        }),
      ]);
      children.push({ exited, worker });
      worker.on("message", (message) => {
        if (!message.ready) {
          results.push(message);
        }
      });
      const [message] = await ready;
      if (!message.ready) {
        throw new Error("Worker setup failed");
      }
    })
  );
  const completed = Promise.all(children.map((child) => child.exited));
  if (process.argv.includes("--sequential")) {
    for (const child of children) {
      child.worker.send({ token });
      // eslint-disable-next-line no-await-in-loop -- This is the serial positive control.
      await child.exited;
    }
  } else {
    for (const child of children) {
      child.worker.send({ token });
    }
  }
  await completed;
  if (
    results.length !== processCount ||
    results.some((result) => result.error)
  ) {
    throw new Error("Incomplete or failed worker results; inconclusive");
  }
  const counts = {};
  for (const result of results) {
    counts[result.eventType] = (counts[result.eventType] ?? 0) + 1;
  }
  const rows = await pool.query(
    "select run_id from workflow.workflow_hooks where token = $1",
    [token]
  );
  const { version } = JSON.parse(
    readFileSync("node_modules/@workflow/world-postgres/package.json", "utf-8")
  );
  const evidence = {
    counts,
    distinctOwners: new Set(rows.rows.map((row) => row.run_id)).size,
    hookRows: rows.rowCount,
    package: `@workflow/world-postgres@${version}`,
    processes: processCount,
    returnedHooks: results.filter((result) => result.hasHook).length,
  };
  console.log(JSON.stringify(evidence, null, 2));
  assert.equal(
    counts.hook_created,
    1,
    "exactly one caller must own a hook token"
  );
  assert.equal(
    counts.hook_conflict,
    processCount - 1,
    "losers must receive conflicts"
  );
  assert.equal(evidence.returnedHooks, 1);
  assert.equal(evidence.hookRows, 1);
  assert.equal(evidence.distinctOwners, 1);
} catch (error) {
  console.error(error);
  process.exitCode = error instanceof assert.AssertionError ? 1 : 2;
} finally {
  clearTimeout(timeout);
  for (const child of children) {
    if (child.worker.exitCode === null) {
      child.worker.kill("SIGTERM");
    }
  }
  await Promise.allSettled(children.map((child) => child.exited));
  await pool.end();
}

import { once } from "node:events";

import { createWorld } from "@workflow/world-postgres";

const world = createWorld({
  connectionString: process.env.WORKFLOW_POSTGRES_URL,
});

try {
  const created = await world.events.create(null, {
    eventData: {
      deploymentId: "hook-token-race-repro",
      input: [],
      workflowName: "hook-token-race-repro",
    },
    eventType: "run_created",
  });

  process.send({ ready: true });
  const [{ token }] = await once(process, "message");

  const result = await world.events.create(created.run.runId, {
    correlationId: `hook-${created.run.runId}`,
    eventData: { token },
    eventType: "hook_created",
  });

  process.send({
    eventType: result.event.eventType,
    hasHook: result.hook !== undefined,
    runId: created.run.runId,
  });
} catch (error) {
  process.send({
    error: error instanceof Error ? error.message : String(error),
  });
  process.exitCode = 1;
} finally {
  await world.close();
  process.disconnect();
}