Do not consume task identity on add_job conflict

__tests__/migrate.test.ts

@@ -14,7 +14,7 @@ import {
 
 const options: WorkerSharedOptions = {};
 
-const MAX_MIGRATION_NUMBER = 19;
+const MAX_MIGRATION_NUMBER = 20;
 
 test("migration installs schema; second migration does no harm", async () => {
   await withPgClient(async (pgClient) => {

__tests__/schema.sql

@@ -55,14 +55,24 @@ begin
     limit 1;
     return v_job;
   elsif job_key_mode = 'unsafe_dedupe' then
-    -- Ensure all the tasks exist
+    -- Ensure all the tasks exist (insert only missing identifiers so identity is not consumed)
     insert into "graphile_worker"._private_tasks as tasks (identifier)
-    values (add_job.identifier)
+    select add_job.identifier
+    where not exists (
+      select 1
+      from "graphile_worker"._private_tasks as existing
+      where existing.identifier = add_job.identifier
+    )
     on conflict do nothing;
     -- Ensure all the queues exist
     if add_job.queue_name is not null then
       insert into "graphile_worker"._private_job_queues as job_queues (queue_name)
-      values (add_job.queue_name)
+      select add_job.queue_name
+      where not exists (
+        select 1
+        from "graphile_worker"._private_job_queues as existing
+        where existing.queue_name = add_job.queue_name
+      )
       on conflict do nothing;
     end if;
     -- Insert job, but if one already exists then do nothing, even if the
@@ -118,16 +128,26 @@ CREATE FUNCTION graphile_worker.add_jobs(specs graphile_worker.job_spec[], job_k
     LANGUAGE plpgsql
     AS $$
 begin
-  -- Ensure all the tasks exist
+  -- Ensure all the tasks exist (insert only missing identifiers so identity is not consumed)
   insert into "graphile_worker"._private_tasks as tasks (identifier)
   select distinct spec.identifier
   from unnest(specs) spec
+  where not exists (
+    select 1
+    from "graphile_worker"._private_tasks as existing
+    where existing.identifier = spec.identifier
+  )
   on conflict do nothing;
   -- Ensure all the queues exist
   insert into "graphile_worker"._private_job_queues as job_queues (queue_name)
   select distinct spec.queue_name
   from unnest(specs) spec
   where spec.queue_name is not null
+  and not exists (
+    select 1
+    from "graphile_worker"._private_job_queues as existing
+    where existing.queue_name = spec.queue_name
+  )
   on conflict do nothing;
   -- Ensure any locked jobs have their key cleared - in the case of locked
   -- existing job create a new job instead as it must have already started
@@ -395,4 +415,5 @@ COPY graphile_worker.migrations (id, ts, breaking) FROM stdin;
 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
+20	1970-01-01 00:00:00.000000+00	f
 \.

__tests__/workerUtils.addJob.test.ts

@@ -1,9 +1,17 @@
 import { jest } from "@jest/globals";
 
+import { makeWithPgClientFromClient } from "../src/helpers.ts";
 import type { Task, WorkerSharedOptions, WorkerUtils } from "../src/index.ts";
 import { addJobAdhoc, makeWorkerUtils, runTaskListOnce } from "../src/index.ts";
 import {
+  makeEnhancedWithPgClient,
+  processSharedOptions,
+} from "../src/lib.ts";
+import { getTaskDetails } from "../src/taskIdentifiers.ts";
+import {
+  ESCAPED_GRAPHILE_WORKER_SCHEMA,
   getJobs,
+  GRAPHILE_WORKER_SCHEMA,
   HOUR,
   reset,
   setupFakeTimers,
@@ -180,3 +188,77 @@ test("adding job respects useNodeTime", () =>
     expect(+runAt).toBeGreaterThan(timeOfAddJob - 2000);
     expect(+runAt).toBeLessThan(timeOfAddJob + 2000);
   }));
+
+test("does not consume task identity when identifier already exists (GH-619)", () =>
+  withPgClient(async (pgClient, { TEST_CONNECTION_STRING }) => {
+    await reset(pgClient, options);
+
+    utils = await makeWorkerUtils({
+      connectionString: TEST_CONNECTION_STRING,
+    });
+    // Register the identifier and a named queue once (add_jobs path)
+    await utils.addJob("my_task", { a: 1 }, { queueName: "q1" });
+
+    const {
+      rows: [{ task_count: initialTaskCount, queue_count: initialQueueCount }],
+    } = await pgClient.query(
+      `select
+        (select count(*)::int from ${ESCAPED_GRAPHILE_WORKER_SCHEMA}._private_tasks) as task_count,
+        (select count(*)::int from ${ESCAPED_GRAPHILE_WORKER_SCHEMA}._private_job_queues) as queue_count`,
+    );
+    expect(initialTaskCount).toBe(1);
+    expect(initialQueueCount).toBe(1);
+
+    await pgClient.query(
+      `select setval(pg_get_serial_sequence('${GRAPHILE_WORKER_SCHEMA}._private_tasks', 'id'), 2147483647, true)`,
+    );
+    await pgClient.query(
+      `select setval(pg_get_serial_sequence('${GRAPHILE_WORKER_SCHEMA}._private_job_queues', 'id'), 2147483647, true)`,
+    );
+
+    // Same identifier + queue must not call nextval (would raise at integer ceiling)
+    await expect(
+      utils.addJob("my_task", { a: 2 }, { queueName: "q1" }),
+    ).resolves.toBeTruthy();
+
+    // unsafe_dedupe path also must not consume identity
+    await expect(
+      utils.addJob(
+        "my_task",
+        { a: 3 },
+        { queueName: "q1", jobKey: "gh619", jobKeyMode: "unsafe_dedupe" },
+      ),
+    ).resolves.toBeTruthy();
+
+    const {
+      rows: [{ task_count, queue_count }],
+    } = await pgClient.query(
+      `select
+        (select count(*)::int from ${ESCAPED_GRAPHILE_WORKER_SCHEMA}._private_tasks) as task_count,
+        (select count(*)::int from ${ESCAPED_GRAPHILE_WORKER_SCHEMA}._private_job_queues) as queue_count`,
+    );
+    expect(task_count).toBe(1);
+    expect(queue_count).toBe(1);
+
+    // Startup getTaskDetails insert must also skip existing identifiers
+    const compiledSharedOptions = processSharedOptions({
+      connectionString: TEST_CONNECTION_STRING,
+    });
+    const details = await getTaskDetails(
+      compiledSharedOptions,
+      makeEnhancedWithPgClient(makeWithPgClientFromClient(pgClient)),
+      { my_task() {} },
+    );
+    expect(details.taskIds).toHaveLength(1);
+
+    const {
+      rows: [{ task_count: taskCountAfterDetails }],
+    } = await pgClient.query(
+      `select count(*)::int as task_count from ${ESCAPED_GRAPHILE_WORKER_SCHEMA}._private_tasks`,
+    );
+    expect(taskCountAfterDetails).toBe(1);
+
+    await utils.release();
+    utils = null;
+  }));
+

sql/000020.sql

@@ -0,0 +1,192 @@
+CREATE OR REPLACE FUNCTION :GRAPHILE_WORKER_SCHEMA.add_job(identifier text, payload json DEFAULT NULL::json, queue_name text DEFAULT NULL::text, run_at timestamp with time zone DEFAULT NULL::timestamp with time zone, max_attempts integer DEFAULT NULL::integer, job_key text DEFAULT NULL::text, priority integer DEFAULT NULL::integer, flags text[] DEFAULT NULL::text[], job_key_mode text DEFAULT 'replace'::text) RETURNS :GRAPHILE_WORKER_SCHEMA._private_jobs
+    LANGUAGE plpgsql
+    AS $$
+declare
+  v_job :GRAPHILE_WORKER_SCHEMA._private_jobs;
+begin
+  if (job_key is null or job_key_mode is null or job_key_mode in ('replace', 'preserve_run_at')) then
+    select * into v_job
+    from :GRAPHILE_WORKER_SCHEMA.add_jobs(
+      ARRAY[(
+        identifier,
+        payload,
+        queue_name,
+        run_at,
+        max_attempts::smallint,
+        job_key,
+        priority::smallint,
+        flags
+      ):::GRAPHILE_WORKER_SCHEMA.job_spec],
+      (job_key_mode = 'preserve_run_at')
+    )
+    limit 1;
+    return v_job;
+  elsif job_key_mode = 'unsafe_dedupe' then
+    -- Ensure all the tasks exist (insert only missing identifiers so identity is not consumed)
+    insert into :GRAPHILE_WORKER_SCHEMA._private_tasks as tasks (identifier)
+    select add_job.identifier
+    where not exists (
+      select 1
+      from :GRAPHILE_WORKER_SCHEMA._private_tasks as existing
+      where existing.identifier = add_job.identifier
+    )
+    on conflict do nothing;
+    -- Ensure all the queues exist
+    if add_job.queue_name is not null then
+      insert into :GRAPHILE_WORKER_SCHEMA._private_job_queues as job_queues (queue_name)
+      select add_job.queue_name
+      where not exists (
+        select 1
+        from :GRAPHILE_WORKER_SCHEMA._private_job_queues as existing
+        where existing.queue_name = add_job.queue_name
+      )
+      on conflict do nothing;
+    end if;
+    -- Insert job, but if one already exists then do nothing, even if the
+    -- existing job has already started (and thus represents an out-of-date
+    -- world state). This is dangerous because it means that whatever state
+    -- change triggered this add_job may not be acted upon (since it happened
+    -- after the existing job started executing, but no further job is being
+    -- scheduled), but it is useful in very rare circumstances for
+    -- de-duplication. If in doubt, DO NOT USE THIS.
+    insert into :GRAPHILE_WORKER_SCHEMA._private_jobs as jobs (
+      job_queue_id,
+      task_id,
+      payload,
+      run_at,
+      max_attempts,
+      key,
+      priority,
+      flags
+    )
+      select
+        job_queues.id,
+        tasks.id,
+        coalesce(add_job.payload, '{}'::json),
+        coalesce(add_job.run_at, now()),
+        coalesce(add_job.max_attempts::smallint, 25::smallint),
+        add_job.job_key,
+        coalesce(add_job.priority::smallint, 0::smallint),
+        (
+          select jsonb_object_agg(flag, true)
+          from unnest(add_job.flags) as item(flag)
+        )
+      from :GRAPHILE_WORKER_SCHEMA._private_tasks as tasks
+      left join :GRAPHILE_WORKER_SCHEMA._private_job_queues as job_queues
+      on job_queues.queue_name = add_job.queue_name
+      where tasks.identifier = add_job.identifier
+    on conflict (key)
+      -- Bump the updated_at so that there's something to return
+      do update set
+        revision = jobs.revision + 1,
+        updated_at = now()
+      returning *
+      into v_job;
+    if v_job.revision = 0 then
+      perform pg_notify('jobs:insert', '{"r":' || random()::text || ',"count":1}');
+    end if;
+    return v_job;
+  else
+    raise exception 'Invalid job_key_mode value, expected ''replace'', ''preserve_run_at'' or ''unsafe_dedupe''.' using errcode = 'GWBKM';
+  end if;
+end;
+$$;
+
+CREATE OR REPLACE FUNCTION :GRAPHILE_WORKER_SCHEMA.add_jobs(specs :GRAPHILE_WORKER_SCHEMA.job_spec[], job_key_preserve_run_at boolean DEFAULT false) RETURNS SETOF :GRAPHILE_WORKER_SCHEMA._private_jobs
+    LANGUAGE plpgsql
+    AS $$
+begin
+  -- Ensure all the tasks exist (insert only missing identifiers so identity is not consumed)
+  insert into :GRAPHILE_WORKER_SCHEMA._private_tasks as tasks (identifier)
+  select distinct spec.identifier
+  from unnest(specs) spec
+  where not exists (
+    select 1
+    from :GRAPHILE_WORKER_SCHEMA._private_tasks as existing
+    where existing.identifier = spec.identifier
+  )
+  on conflict do nothing;
+  -- Ensure all the queues exist
+  insert into :GRAPHILE_WORKER_SCHEMA._private_job_queues as job_queues (queue_name)
+  select distinct spec.queue_name
+  from unnest(specs) spec
+  where spec.queue_name is not null
+  and not exists (
+    select 1
+    from :GRAPHILE_WORKER_SCHEMA._private_job_queues as existing
+    where existing.queue_name = spec.queue_name
+  )
+  on conflict do nothing;
+  -- Ensure any locked jobs have their key cleared - in the case of locked
+  -- existing job create a new job instead as it must have already started
+  -- executing (i.e. it's world state is out of date, and the fact add_job
+  -- has been called again implies there's new information that needs to be
+  -- acted upon).
+  update :GRAPHILE_WORKER_SCHEMA._private_jobs as jobs
+  set
+    key = null,
+    attempts = jobs.max_attempts,
+    updated_at = now()
+  from unnest(specs) spec
+  where spec.job_key is not null
+  and jobs.key = spec.job_key
+  and is_available is not true;
+
+  -- WARNING: this count is not 100% accurate; 'on conflict' clause will cause it to be an overestimate
+  perform pg_notify('jobs:insert', '{"r":' || random()::text || ',"count":' || array_length(specs, 1)::text || '}');
+
+  -- TODO: is there a risk that a conflict could occur depending on the
+  -- isolation level?
+  return query insert into :GRAPHILE_WORKER_SCHEMA._private_jobs as jobs (
+    job_queue_id,
+    task_id,
+    payload,
+    run_at,
+    max_attempts,
+    key,
+    priority,
+    flags
+  )
+    select
+      job_queues.id,
+      tasks.id,
+      coalesce(spec.payload, '{}'::json),
+      coalesce(spec.run_at, now()),
+      coalesce(spec.max_attempts, 25),
+      spec.job_key,
+      coalesce(spec.priority, 0),
+      (
+        select jsonb_object_agg(flag, true)
+        from unnest(spec.flags) as item(flag)
+      )
+    from unnest(specs) spec
+    inner join :GRAPHILE_WORKER_SCHEMA._private_tasks as tasks
+    on tasks.identifier = spec.identifier
+    left join :GRAPHILE_WORKER_SCHEMA._private_job_queues as job_queues
+    on job_queues.queue_name = spec.queue_name
+  on conflict (key) do update set
+    job_queue_id = excluded.job_queue_id,
+    task_id = excluded.task_id,
+    payload =
+      case
+      when json_typeof(jobs.payload) = 'array' and json_typeof(excluded.payload) = 'array' then
+        (jobs.payload::jsonb || excluded.payload::jsonb)::json
+      else
+        excluded.payload
+      end,
+    max_attempts = excluded.max_attempts,
+    run_at = (case
+      when job_key_preserve_run_at is true and jobs.attempts = 0 then jobs.run_at
+      else excluded.run_at
+    end),
+    priority = excluded.priority,
+    revision = jobs.revision + 1,
+    flags = excluded.flags,
+    -- always reset error/retry state
+    attempts = 0,
+    last_error = null,
+    updated_at = now()
+  where jobs.locked_at is null
+  returning *;
+end;
+$$;