From 23bae17db78d5c8b80c335c6760ef7dfcbb28fa7 Mon Sep 17 00:00:00 2001 From: Yi Zhan Date: Thu, 20 Aug 2026 17:21:17 +0000 Subject: [PATCH] Do not consume task identity on add_job conflict Insert missing task/queue identifiers only so nextval is not called when the name already exists. --- __tests__/migrate.test.ts | 2 +- __tests__/schema.sql | 29 +++- __tests__/workerUtils.addJob.test.ts | 82 ++++++++++++ sql/000020.sql | 192 ++++++++++++++++++++++++++ src/generated/sql.ts | 193 +++++++++++++++++++++++++++ src/taskIdentifiers.ts | 2 +- 6 files changed, 494 insertions(+), 6 deletions(-) create mode 100644 sql/000020.sql diff --git a/__tests__/migrate.test.ts b/__tests__/migrate.test.ts index 8a211dbb..bcf5a6a3 100644 --- a/__tests__/migrate.test.ts +++ b/__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) => { diff --git a/__tests__/schema.sql b/__tests__/schema.sql index 1e545a57..aa05d45a 100644 --- a/__tests__/schema.sql +++ b/__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 \. diff --git a/__tests__/workerUtils.addJob.test.ts b/__tests__/workerUtils.addJob.test.ts index ae244e0e..59493d5a 100644 --- a/__tests__/workerUtils.addJob.test.ts +++ b/__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; + })); + diff --git a/sql/000020.sql b/sql/000020.sql new file mode 100644 index 00000000..7e4005e4 --- /dev/null +++ b/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; +$$; diff --git a/src/generated/sql.ts b/src/generated/sql.ts index df57995a..058f66ad 100644 --- a/src/generated/sql.ts +++ b/src/generated/sql.ts @@ -2364,5 +2364,198 @@ $$; -- 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; +`, + "000020.sql": String.raw`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; +$$; `, }; diff --git a/src/taskIdentifiers.ts b/src/taskIdentifiers.ts index b2df694c..eb5d630a 100644 --- a/src/taskIdentifiers.ts +++ b/src/taskIdentifiers.ts @@ -43,7 +43,7 @@ export function getTaskDetails( cache.lastDigest = (async () => { const { rows } = await withPgClient.withRetries(async (client) => { await client.query({ - text: `insert into ${escapedWorkerSchema}._private_tasks as tasks (identifier) select unnest($1::text[]) on conflict do nothing`, + text: `insert into ${escapedWorkerSchema}._private_tasks as tasks (identifier) select i from unnest($1::text[]) as u(i) where not exists (select 1 from ${escapedWorkerSchema}._private_tasks as existing where existing.identifier = u.i) on conflict do nothing`, values: [supportedTaskNames], }); return client.query<{ id: number; identifier: string }>({