HealthLog/tests/integration/queue-singleton-policy.integration.test.ts

/**
 * Queue de-duplication semantics, asserted against a real pg-boss on a real
 * Postgres.
 *
 * This suite exists because the bug it pins is invisible to a mocked test. A
 * bare `singletonKey` on a `send()` LOOKS like de-duplication at every call
 * site, and a unit test with a stubbed boss will happily agree. But pg-boss
 * only constrains `singleton_key` through partial unique indexes that are
 * scoped BY QUEUE POLICY, and under the default `standard` policy no such index
 * exists — so every bare `singletonKey` in the tree de-duplicated nothing.
 * Only a real insert against a real index can tell the two apart.
 *
 * The `standard` case below is deliberately kept as a permanent, executable
 * statement of the original defect: if someone drops a policy entry from a
 * registrar table, the matching assertion here goes red instead of the
 * behaviour silently reverting to "the key does nothing".
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { PgBoss } from "pg-boss";

let boss: PgBoss;

/** Unique per run so a re-run never collides with a queue left behind. */
const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`;
const q = (name: string) => `test-${name}-${suffix}`;

beforeAll(async () => {
  boss = new PgBoss({
    connectionString: process.env.DATABASE_URL,
    // Keep the maintenance/monitor loops out of the way; this suite only
    // exercises createQueue/send/fetch.
    schedule: false,
    supervise: false,
  });
  await boss.start();
}, 120_000);

afterAll(async () => {
  await boss?.stop({ graceful: false });
});

describe("pg-boss singleton semantics by queue policy", () => {
  it("standard: a bare singletonKey de-duplicates NOTHING (the original defect)", async () => {
    const name = q("standard");
    await boss.createQueue(name);

    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });

    // Both inserted. This is the finding: under the default policy the key is
    // inert, so every "coalesced by singletonKey" comment in the tree was
    // describing a guarantee that did not exist.
    expect(first).toBeTruthy();
    expect(second).toBeTruthy();
    expect(second).not.toBe(first);
  });

  it("exclusive: suppresses a duplicate while the first is QUEUED", async () => {
    const name = q("excl-queued");
    await boss.createQueue(name, { policy: "exclusive" });

    const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" });

    expect(first).toBeTruthy();
    // pg-boss inserts with ON CONFLICT DO NOTHING ... RETURNING, so a
    // suppressed send resolves to null rather than throwing. Every boot
    // discovery helper in the tree counts exactly this as `skipped`.
    expect(second).toBeNull();
  });

  it("exclusive: still suppresses once the first job is ACTIVE", async () => {
    const name = q("excl-active");
    await boss.createQueue(name, { policy: "exclusive" });

    const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
    expect(first).toBeTruthy();

    // Move the job out of `created` and into `active`.
    const fetched = await boss.fetch(name);
    expect(fetched).toHaveLength(1);

    // This is the case that matters for the backfills: a worker restarting
    // during a heavy account's multi-hour pass used to append another
    // identical full-history job, because `short` semantics would have let it
    // through. `exclusive` covers queued OR active.
    const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" });
    expect(second).toBeNull();
  });

  it("exclusive: a genuinely distinct key still enqueues", async () => {
    const name = q("excl-distinct");
    await boss.createQueue(name, { policy: "exclusive" });

    const a = await boss.send(name, { n: 1 }, { singletonKey: "user-a" });
    const b = await boss.send(name, { n: 2 }, { singletonKey: "user-b" });

    expect(a).toBeTruthy();
    expect(b).toBeTruthy();
    expect(b).not.toBe(a);
  });

  it("short: suppresses a duplicate while the first is QUEUED", async () => {
    const name = q("short-queued");
    await boss.createQueue(name, { policy: "short" });

    const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" });

    expect(first).toBeTruthy();
    expect(second).toBeNull();
  });

  it("short: ADMITS a send once the first job is ACTIVE", async () => {
    const name = q("short-active");
    await boss.createQueue(name, { policy: "short" });

    const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" });
    expect(first).toBeTruthy();

    const fetched = await boss.fetch(name);
    expect(fetched).toHaveLength(1);

    // This is the property that makes `short` the correct choice for the
    // rollup recompute and the per-document jobs: the running job already read
    // its input, so a write that lands afterwards MUST be able to enqueue its
    // own recompute or it would be stranded in a stale bucket. `exclusive`
    // would return null here and silently drop that work.
    const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" });
    expect(second).toBeTruthy();
    expect(second).not.toBe(first);
  });

  it("short: a genuinely distinct key still enqueues", async () => {
    const name = q("short-distinct");
    await boss.createQueue(name, { policy: "short" });

    const a = await boss.send(name, { n: 1 }, { singletonKey: "bucket-1" });
    const b = await boss.send(name, { n: 2 }, { singletonKey: "bucket-2" });

    expect(a).toBeTruthy();
    expect(b).toBeTruthy();
    expect(b).not.toBe(a);
  });
});

describe("policy reconcile for queues that already exist", () => {
  it("createQueue does NOT change the policy of an existing queue", async () => {
    const name = q("recreate");
    await boss.createQueue(name);

    // pg-boss's create_queue() ends in ON CONFLICT DO NOTHING, so this is a
    // no-op rather than an upgrade. Without the reconcile in
    // `registrar-shared.ts`, shipping the policy tables would therefore change
    // nothing at all on an instance whose queues already exist -- which is
    // every running instance.
    await boss.createQueue(name, { policy: "exclusive" });

    const queue = await boss.getQueue(name);
    expect(queue?.policy).toBe("standard");

    // And the behaviour confirms it: the key is still inert.
    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
    expect(first).toBeTruthy();
    expect(second).toBeTruthy();
  });

  it("updateQueue refuses a policy change outright", async () => {
    const name = q("update");
    await boss.createQueue(name);

    await expect(
      // @ts-expect-error -- policy is excluded from UpdateQueueOptions by
      // design; this asserts the runtime guard that forces the direct-column
      // reconcile the registrar performs.
      boss.updateQueue(name, { policy: "exclusive" }),
    ).rejects.toThrow(/policy cannot be changed/i);
  });

  it("writing the policy column takes effect on the next send", async () => {
    const name = q("reconciled");
    await boss.createQueue(name);

    // Same statement `reconcileQueuePolicies` issues. The partial unique
    // indexes for every policy already exist on the shared `job_common` table
    // (this deployment does not partition per queue), so claiming a policy
    // later is enough -- no schema change, no worker restart.
    await boss.getDb().executeSql(
      `UPDATE pgboss.queue SET policy = $2, updated_on = now()
         WHERE name = $1 AND policy IS DISTINCT FROM $2`,
      [name, "exclusive"],
    );

    const queue = await boss.getQueue(name);
    expect(queue?.policy).toBe("exclusive");

    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
    expect(first).toBeTruthy();
    expect(second).toBeNull();
  });
});

describe("createAndSchedule applies the registrar's policy table", () => {
  it("provisions a NEW queue with the decided policy", async () => {
    const { createAndSchedule } =
      await import("@/lib/jobs/reminder/registrar-shared");
    const name = q("cas-new");

    await createAndSchedule(boss, [name], [], {
      [name]: { policy: "exclusive", reason: "test" },
    });

    const queue = await boss.getQueue(name);
    expect(queue?.policy).toBe("exclusive");

    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
    expect(first).toBeTruthy();
    expect(second).toBeNull();
  });

  it("reconciles an EXISTING standard queue onto the decided policy", async () => {
    const { createAndSchedule } =
      await import("@/lib/jobs/reminder/registrar-shared");
    const name = q("cas-existing");

    // Simulate a running instance: the queue already exists under `standard`,
    // which is how every queue on every deployed instance was provisioned.
    await boss.createQueue(name);
    expect((await boss.getQueue(name))?.policy).toBe("standard");

    await createAndSchedule(boss, [name], [], {
      [name]: { policy: "short", reason: "test" },
    });

    expect((await boss.getQueue(name))?.policy).toBe("short");

    const first = await boss.send(name, { n: 1 }, { singletonKey: "same" });
    const second = await boss.send(name, { n: 2 }, { singletonKey: "same" });
    expect(first).toBeTruthy();
    expect(second).toBeNull();
  });

  it("leaves a queue absent from the table on standard", async () => {
    const { createAndSchedule } =
      await import("@/lib/jobs/reminder/registrar-shared");
    const name = q("cas-absent");

    await createAndSchedule(boss, [name], [], {});

    expect((await boss.getQueue(name))?.policy).toBe("standard");
  });
});

describe("boot-discovery skipped counter", () => {
  it("can actually be non-zero once the queue carries a policy", async () => {
    const name = q("skipped");
    await boss.createQueue(name, { policy: "exclusive" });

    // Mirror the shape every boot-discovery helper uses: iterate a candidate
    // set, count a null job id as `skipped`. Two of these three users are
    // already represented, which under `standard` could never happen -- which
    // is why `skipped` was structurally always zero in the boot log lines.
    const candidates = ["u1", "u1", "u2", "u2", "u3"];
    let enqueued = 0;
    let skipped = 0;
    for (const userId of candidates) {
      const jobId = await boss.send(name, { userId }, { singletonKey: userId });
      if (jobId) {
        enqueued += 1;
      } else {
        skipped += 1;
      }
    }

    expect(enqueued).toBe(3);
    expect(skipped).toBe(2);
  });
});