fix(worker): re-check document after the legacy claim_ingestion_jobs row lock

supabase/applied-migration-hashes.json

@@ -1,7 +1,7 @@
 {
   "note": "sha256 of every migration this repository has shipped, with newlines normalised. An entry whose hash changes means an applied migration was edited, which cannot reach the live database. See scripts/check-migration-immutability.mjs.",
   "generator": "scripts/check-migration-immutability.mjs",
-  "sealed_at": "2026-10-18T05:38:30.657Z",
+  "sealed_at": "2026-10-18T06:29:14.181Z",
   "versions": {
     "20260527000000_bulk_ingestion.sql": "813e8e314064a6dcbaaa55e2765a8ce901d5783b304b4a7af7bb7417a07daf7c",
     "20260527001000_documents_policy_cleanup.sql": "559c26f3172bf75e545a41adef75538b2bf648ea7d53c8eb3d3ea7c244c65a48",
@@ -322,6 +322,7 @@
     "20261016210000_adversarial_review_rpc_and_ttl_hardening.sql": "8d58d892eec74bc19561065beb01ec46c98c767519295337a61d7b32e0adda41",
     "20261017120000_teaching_display_revoke_null_creator_guard.sql": "3b92e22983206cbaa8950c07a6b3e6910b3701fa9c49a9f9e5504c232b4397e4",
     "20261017175448_scope_ingestion_claim_atomically.sql": "af192755dd24aa8a6e6146013e1b7abd1b2954a53410c301dedfcb7576c88eba",
-    "20261018053558_recheck_ingestion_claim_scope_after_job_lock.sql": "a427b8aacce70579ad26bd701adb65912a345ab78d54e5b34386bbd05d261b6c"
+    "20261018053558_recheck_ingestion_claim_scope_after_job_lock.sql": "a427b8aacce70579ad26bd701adb65912a345ab78d54e5b34386bbd05d261b6c",
+    "20261018062736_recheck_legacy_ingestion_claim_document_after_job_lock.sql": "39f08a91ed4c15a8aff0dff80341242604101886de873c5cadafa65e9b92e3c8"
   }
 }

supabase/drift-manifest.json

@@ -1,9 +1,9 @@
 {
-  "generated_at": "2026-10-18T05:44:24.096Z",
+  "generated_at": "2026-10-18T06:37:18.983Z",
   "generator": "scripts/generate-drift-manifest.ts",
   "postgres_image": "supabase/postgres:17.6.1.127@sha256:be60aee15997daca475b710b734bc6bfe52cd544dcd7e9fd2ff58210b6747d83",
-  "schema_sha256": "012a8b157207c836e1ec0dee4caa7be9c35412eefb87fd454be9fcd5eacfb8d9",
-  "replay_seconds": 16,
+  "schema_sha256": "58e7d9c92b1264426f6cd239fdc136f978afe90770e31f950a0b4c505628e4fb",
+  "replay_seconds": 14,
   "snapshot": {
     "views": [
       {
@@ -18649,7 +18649,7 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "64a776fa765a7d8cdc164e55d99bfa9a",
+        "def_hash": "80066ae9391d629393d44dc61b0fc181",
         "signature": "public.claim_ingestion_jobs(text,integer,integer)"
       },
       {

supabase/migrations/20261018062736_recheck_legacy_ingestion_claim_document_after_job_lock.sql

@@ -0,0 +1,55 @@
+-- Follow-up to 20261018053558 (#3633), for the legacy three-argument claim. The job
+-- row can be re-pointed at another document between the candidate pick and its row
+-- lock, so the authority and sibling-lease checks proven for the picked document do
+-- not hold for the row actually leased. After the job row lock, require the locked job
+-- to still belong to the picked (and authority-locked) document; otherwise leave it
+-- pending with no lease and no consumed attempt. Signature, defaults, grants and every
+-- other behaviour are unchanged.
+
+set local lock_timeout = '5s';
+set local statement_timeout = '30s';
+
+create or replace function public.claim_ingestion_jobs(p_worker_id text, p_claim_limit integer default 1, p_stale_after_minutes integer default 45)
+returns setof jsonb language plpgsql security definer set search_path = '' as $$
+declare v_candidate record; v_locked jsonb; v_job public.ingestion_jobs%rowtype; v_count integer := 0; v_now timestamptz; v_generation uuid;
+begin
+  if p_worker_id is null or btrim(p_worker_id) = '' or p_claim_limit is null or p_claim_limit not between 1 and 100
+    or p_stale_after_minutes not between 1 and 1440 or p_stale_after_minutes is null then raise exception 'invalid_ingestion_claim_configuration'; end if;
+  for v_candidate in select j.id, j.document_id from public.ingestion_jobs j
+    where j.attempt_count < j.max_attempts
+      and (j.status = 'pending' and coalesce(j.next_run_at, pg_catalog.clock_timestamp()) <= pg_catalog.clock_timestamp()
+        or j.status = 'processing' and j.lease_expires_at <= pg_catalog.clock_timestamp())
+    order by j.created_at, j.id
+  loop
+    exit when v_count >= p_claim_limit;
+    v_locked := public.lock_ingestion_document_authority(v_candidate.document_id, true);
+    if v_locked is null then continue; end if;
+    perform 1 from public.ingestion_jobs where document_id = v_candidate.document_id order by id for update skip locked;
+    perform 1 from public.indexing_v3_agent_jobs where document_id = v_candidate.document_id order by id for update skip locked;
+    v_now := pg_catalog.clock_timestamp();
+    if exists (select 1 from public.ingestion_jobs where document_id = v_candidate.document_id and status = 'processing' and lease_expires_at > v_now)
+      or exists (select 1 from public.indexing_v3_agent_jobs where document_id = v_candidate.document_id and status = 'processing' and lease_expires_at > v_now) then continue; end if;
+    select * into v_job from public.ingestion_jobs where id = v_candidate.id for update skip locked;
+    if not found or v_job.attempt_count >= v_job.max_attempts or not (
+      v_job.status = 'pending' and coalesce(v_job.next_run_at, v_now) <= v_now
+      or v_job.status = 'processing' and v_job.lease_expires_at <= v_now) then continue; end if;
+    -- The pick-time document is a snapshot. Authority and sibling leases were proven for
+    -- the picked document, so a job that now belongs to any other document is not claimed.
+    if v_job.document_id is distinct from v_candidate.document_id then continue; end if;
+    v_generation := (v_locked->'document'->>'index_generation_id')::uuid;
+    if v_generation is null and (v_locked->'document'->'metadata' ? 'index_generation_id') then continue; end if;
+
+    update public.ingestion_jobs set status = 'processing', locked_by = p_worker_id, locked_at = v_now,
+      lease_token = gen_random_uuid(), lease_expected_generation_id = v_generation,
+      lease_write_generation_id = gen_random_uuid(),
+      lease_authority_digest = v_locked->>'authority_digest', lease_expires_at = v_now + make_interval(secs => p_stale_after_minutes * 60),
+      lease_duration_seconds = p_stale_after_minutes * 60, attempt_count = attempt_count + 1, next_run_at = null
+    where id = v_job.id returning * into v_job;
+    return next to_jsonb(v_job) || jsonb_build_object('job_kind','node', 'documents',v_locked->'document');
+    v_count := v_count + 1;
+  end loop;
+end $$;
+
+alter function public.claim_ingestion_jobs(text, integer, integer) owner to postgres;
+revoke all on function public.claim_ingestion_jobs(text, integer, integer) from public, anon, authenticated;
+grant execute on function public.claim_ingestion_jobs(text, integer, integer) to service_role;

supabase/schema.sql

@@ -34865,3 +34865,59 @@ end $$;
 alter function public.claim_ingestion_jobs(text, integer, integer, uuid[], text[]) owner to postgres;
 revoke all on function public.claim_ingestion_jobs(text, integer, integer, uuid[], text[]) from public, anon, authenticated;
 grant execute on function public.claim_ingestion_jobs(text, integer, integer, uuid[], text[]) to service_role;
+
+-- Follow-up to 20261018053558 (#3633), for the legacy three-argument claim. The job
+-- row can be re-pointed at another document between the candidate pick and its row
+-- lock, so the authority and sibling-lease checks proven for the picked document do
+-- not hold for the row actually leased. After the job row lock, require the locked job
+-- to still belong to the picked (and authority-locked) document; otherwise leave it
+-- pending with no lease and no consumed attempt. Signature, defaults, grants and every
+-- other behaviour are unchanged.
+
+set local lock_timeout = '5s';
+set local statement_timeout = '30s';
+
+create or replace function public.claim_ingestion_jobs(p_worker_id text, p_claim_limit integer default 1, p_stale_after_minutes integer default 45)
+returns setof jsonb language plpgsql security definer set search_path = '' as $$
+declare v_candidate record; v_locked jsonb; v_job public.ingestion_jobs%rowtype; v_count integer := 0; v_now timestamptz; v_generation uuid;
+begin
+  if p_worker_id is null or btrim(p_worker_id) = '' or p_claim_limit is null or p_claim_limit not between 1 and 100
+    or p_stale_after_minutes not between 1 and 1440 or p_stale_after_minutes is null then raise exception 'invalid_ingestion_claim_configuration'; end if;
+  for v_candidate in select j.id, j.document_id from public.ingestion_jobs j
+    where j.attempt_count < j.max_attempts
+      and (j.status = 'pending' and coalesce(j.next_run_at, pg_catalog.clock_timestamp()) <= pg_catalog.clock_timestamp()
+        or j.status = 'processing' and j.lease_expires_at <= pg_catalog.clock_timestamp())
+    order by j.created_at, j.id
+  loop
+    exit when v_count >= p_claim_limit;
+    v_locked := public.lock_ingestion_document_authority(v_candidate.document_id, true);
+    if v_locked is null then continue; end if;
+    perform 1 from public.ingestion_jobs where document_id = v_candidate.document_id order by id for update skip locked;
+    perform 1 from public.indexing_v3_agent_jobs where document_id = v_candidate.document_id order by id for update skip locked;
+    v_now := pg_catalog.clock_timestamp();
+    if exists (select 1 from public.ingestion_jobs where document_id = v_candidate.document_id and status = 'processing' and lease_expires_at > v_now)
+      or exists (select 1 from public.indexing_v3_agent_jobs where document_id = v_candidate.document_id and status = 'processing' and lease_expires_at > v_now) then continue; end if;
+    select * into v_job from public.ingestion_jobs where id = v_candidate.id for update skip locked;
+    if not found or v_job.attempt_count >= v_job.max_attempts or not (
+      v_job.status = 'pending' and coalesce(v_job.next_run_at, v_now) <= v_now
+      or v_job.status = 'processing' and v_job.lease_expires_at <= v_now) then continue; end if;
+    -- The pick-time document is a snapshot. Authority and sibling leases were proven for
+    -- the picked document, so a job that now belongs to any other document is not claimed.
+    if v_job.document_id is distinct from v_candidate.document_id then continue; end if;
+    v_generation := (v_locked->'document'->>'index_generation_id')::uuid;
+    if v_generation is null and (v_locked->'document'->'metadata' ? 'index_generation_id') then continue; end if;
+
+    update public.ingestion_jobs set status = 'processing', locked_by = p_worker_id, locked_at = v_now,
+      lease_token = gen_random_uuid(), lease_expected_generation_id = v_generation,
+      lease_write_generation_id = gen_random_uuid(),
+      lease_authority_digest = v_locked->>'authority_digest', lease_expires_at = v_now + make_interval(secs => p_stale_after_minutes * 60),
+      lease_duration_seconds = p_stale_after_minutes * 60, attempt_count = attempt_count + 1, next_run_at = null
+    where id = v_job.id returning * into v_job;
+    return next to_jsonb(v_job) || jsonb_build_object('job_kind','node', 'documents',v_locked->'document');
+    v_count := v_count + 1;
+  end loop;
+end $$;
+
+alter function public.claim_ingestion_jobs(text, integer, integer) owner to postgres;
+revoke all on function public.claim_ingestion_jobs(text, integer, integer) from public, anon, authenticated;
+grant execute on function public.claim_ingestion_jobs(text, integer, integer) to service_role;

tests/sql/ingestion-legacy-claim-recheck-after-lock.sql

@@ -0,0 +1,91 @@
+-- Disposable database only. The legacy three-argument claim_ingestion_jobs must not lease
+-- a job re-pointed at another document between its candidate pick and its job row lock
+-- (follow-up to #3633, which fixed the scoped five-argument claim).
+--
+-- The interleaving is made deterministic by wrapping the authority lock, which the claim
+-- takes after the pick and before the job row lock. The wrapper performs the concurrent
+-- writer's re-point, then calls the real lock. Under READ COMMITTED the claim's later
+-- statements see that write exactly as they would see another session's committed one.
+-- Everything, including the wrapper, rolls back.
+\set ON_ERROR_STOP on
+begin;
+set local statement_timeout = '60s';
+create function pg_temp.assert_true(ok boolean, message text) returns void
+language plpgsql as $$ begin
+  if ok is distinct from true then raise exception '%', message; end if;
+end $$;
+select pg_temp.assert_true(not exists (select 1 from public.documents), 'requires an empty disposable database');
+insert into auth.users(id) values ('11111111-1111-4111-8111-111111111111');
+insert into public.documents(id, owner_id, title, file_name, file_type, storage_path, status, metadata)
+values
+  ('00000000-0000-4000-8000-00000000000a', '11111111-1111-4111-8111-111111111111',
+   'Synthetic legacy picked PDF', 'picked.pdf', 'application/pdf', 'synthetic/legacy-recheck-picked', 'queued', '{}'),
+  ('00000000-0000-4000-8000-00000000000b', '11111111-1111-4111-8111-111111111111',
+   'Synthetic legacy other TXT', 'other.txt', 'text/plain', 'synthetic/legacy-recheck-txt', 'queued', '{}'),
+  ('00000000-0000-4000-8000-00000000000c', '11111111-1111-4111-8111-111111111111',
+   'Synthetic legacy other PDF', 'other.pdf', 'application/pdf', 'synthetic/legacy-recheck-pdf', 'queued', '{}');
+
+alter function public.lock_ingestion_document_authority(uuid, boolean)
+  rename to lock_ingestion_document_authority_under_test;
+create function public.lock_ingestion_document_authority(p_document_id uuid, p_try boolean)
+returns jsonb language plpgsql set search_path = '' as $$
+declare
+  v_job uuid := nullif(pg_catalog.current_setting('claim_recheck.job_id', true), '')::uuid;
+  v_target uuid := nullif(pg_catalog.current_setting('claim_recheck.target_document_id', true), '')::uuid;
+begin
+  if v_job is not null then
+    update public.ingestion_jobs set document_id = v_target where id = v_job and document_id = p_document_id;
+    perform pg_catalog.set_config('claim_recheck.job_id', '', true);
+  end if;
+  return public.lock_ingestion_document_authority_under_test(p_document_id, p_try);
+end $$;
+
+-- One due job, on the picked document. The other documents have no open job, so the
+-- re-point cannot collide with the one-open-job-per-document index.
+create function pg_temp.reset_queue() returns void language plpgsql as $$ begin
+  delete from public.ingestion_jobs;
+  insert into public.ingestion_jobs(id, document_id, status, created_at)
+  values ('00000000-0000-4000-9000-00000000000a', '00000000-0000-4000-8000-00000000000a', 'pending', now() - interval '1 day');
+end $$;
+create function pg_temp.repoint_mid_claim(p_target uuid) returns void language sql as $$
+  select pg_catalog.set_config('claim_recheck.job_id', '00000000-0000-4000-9000-00000000000a', true),
+         pg_catalog.set_config('claim_recheck.target_document_id', p_target::text, true);
+$$;
+create function pg_temp.job_untouched(p_document uuid) returns boolean language sql as $$
+  select count(*) = 1 and bool_and(document_id = p_document and status = 'pending' and attempt_count = 0
+    and locked_by is null and lease_token is null)
+  from public.ingestion_jobs where id = '00000000-0000-4000-9000-00000000000a';
+$$;
+
+-- Control: the wrapper alone changes nothing, so the legacy claim still leases the job.
+select pg_temp.reset_queue();
+create temp table claims as
+select job from public.claim_ingestion_jobs('synthetic-worker', 1, 45) job;
+select pg_temp.assert_true((select count(*) = 1 and bool_and(job->>'document_id' = '00000000-0000-4000-8000-00000000000a'
+  and job->'documents'->>'id' = '00000000-0000-4000-8000-00000000000a') from claims),
+  'control: an unmoved job must still be claimed by the legacy claim');
+
+-- The job moves to another document mid-claim. That document's authority was never
+-- locked, so the legacy claim must skip the job rather than lease it under the old one.
+select pg_temp.reset_queue();
+select pg_temp.repoint_mid_claim('00000000-0000-4000-8000-00000000000c');
+select pg_temp.assert_true(not exists (select 1 from public.claim_ingestion_jobs('synthetic-worker', 1, 45)),
+  'legacy claim: a job whose document changed after the pick must not be leased under the old document''s authority');
+select pg_temp.assert_true(pg_temp.job_untouched('00000000-0000-4000-8000-00000000000c'),
+  'legacy claim: the re-pointed job must stay pending without a lease or a consumed attempt');
+
+-- Same for a re-point onto a document of another file type.
+select pg_temp.reset_queue();
+select pg_temp.repoint_mid_claim('00000000-0000-4000-8000-00000000000b');
+select pg_temp.assert_true(not exists (select 1 from public.claim_ingestion_jobs('synthetic-worker', 1, 45)),
+  'legacy claim: a job re-pointed to another file type after the pick must not be leased');
+select pg_temp.assert_true(pg_temp.job_untouched('00000000-0000-4000-8000-00000000000b'),
+  'legacy claim: the file-type re-pointed job must stay pending without a lease or a consumed attempt');
+
+-- The skipped job is not lost: the next legacy claim picks it under its new document.
+truncate claims;
+insert into claims select job from public.claim_ingestion_jobs('synthetic-worker', 1, 45) job;
+select pg_temp.assert_true((select count(*) = 1 and bool_and(job->>'document_id' = '00000000-0000-4000-8000-00000000000b'
+  and job->'documents'->>'id' = '00000000-0000-4000-8000-00000000000b') from claims),
+  'legacy claim: the next claim must lease the job with its current document''s authority');
+rollback;