fix(worker): re-check document and allowlist scope after 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-17T18:01:56.227Z",
+  "sealed_at": "2026-10-18T05:38:30.657Z",
   "versions": {
     "20260527000000_bulk_ingestion.sql": "813e8e314064a6dcbaaa55e2765a8ce901d5783b304b4a7af7bb7417a07daf7c",
     "20260527001000_documents_policy_cleanup.sql": "559c26f3172bf75e545a41adef75538b2bf648ea7d53c8eb3d3ea7c244c65a48",
@@ -321,6 +321,7 @@
     "20261016134500_teaching_admin_attendance_export.sql": "a9fc28336706aee579e18ec782213627e26eb6ca1cd60ab46f4625cf5efee94e",
     "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"
+    "20261017175448_scope_ingestion_claim_atomically.sql": "af192755dd24aa8a6e6146013e1b7abd1b2954a53410c301dedfcb7576c88eba",
+    "20261018053558_recheck_ingestion_claim_scope_after_job_lock.sql": "a427b8aacce70579ad26bd701adb65912a345ab78d54e5b34386bbd05d261b6c"
   }
 }

supabase/drift-manifest.json

@@ -1,9 +1,9 @@
 {
-  "generated_at": "2026-10-17T18:37:12.668Z",
+  "generated_at": "2026-10-18T05:44:24.096Z",
   "generator": "scripts/generate-drift-manifest.ts",
   "postgres_image": "supabase/postgres:17.6.1.127@sha256:be60aee15997daca475b710b734bc6bfe52cd544dcd7e9fd2ff58210b6747d83",
-  "schema_sha256": "6e5913909f23c2e627413a4fcce0508e2002f129b958eafd27868ed5ab0f8b6a",
-  "replay_seconds": 18,
+  "schema_sha256": "012a8b157207c836e1ec0dee4caa7be9c35412eefb87fd454be9fcd5eacfb8d9",
+  "replay_seconds": 16,
   "snapshot": {
     "views": [
       {
@@ -18641,7 +18641,7 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "8d0a15ae0b00bcfd8b0d9701287d48d2",
+        "def_hash": "c1ef5a420ef6a43ea937c6aa80482575",
         "signature": "public.claim_ingestion_jobs(text,integer,integer,uuid[],text[])"
       },
       {

supabase/migrations/20261018053558_recheck_ingestion_claim_scope_after_job_lock.sql

@@ -0,0 +1,65 @@
+-- Follow-up to 20261017175448 (#3628). The job row can be re-pointed at another
+-- document between the candidate pick and its row lock, so the scope proven at
+-- pick time does 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 and to still sit inside the worker's document allowlist; otherwise
+-- leave it unclaimed. Signature, 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, p_stale_after_minutes integer, p_allowed_document_ids uuid[], p_allowed_file_types text[])
+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 (p_allowed_document_ids is null or j.document_id = any(p_allowed_document_ids))
+      and (p_allowed_file_types is null or exists (select 1 from public.documents d where d.id = j.document_id
+        and lower(btrim(split_part(d.file_type, ';', 1))) = any(p_allowed_file_types)))
+      and 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;
+    -- File type can change while authority is acquired. Recheck the locked document
+    -- before minting a lease or incrementing attempts; the client snapshot is not authority.
+    if p_allowed_file_types is not null and not coalesce(
+      lower(btrim(split_part(v_locked->'document'->>'file_type', ';', 1))) = any(p_allowed_file_types), false)
+      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 and allowlist checks are snapshots. Recheck them on the
+    -- locked row: authority, sibling leases and file type were all 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
+      or (p_allowed_document_ids is not null and not coalesce(v_job.document_id = any(p_allowed_document_ids), false))
+      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, 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;

supabase/schema.sql

@@ -34799,3 +34799,69 @@ 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 20261017175448 (#3628). The job row can be re-pointed at another
+-- document between the candidate pick and its row lock, so the scope proven at
+-- pick time does 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 and to still sit inside the worker's document allowlist; otherwise
+-- leave it unclaimed. Signature, 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, p_stale_after_minutes integer, p_allowed_document_ids uuid[], p_allowed_file_types text[])
+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 (p_allowed_document_ids is null or j.document_id = any(p_allowed_document_ids))
+      and (p_allowed_file_types is null or exists (select 1 from public.documents d where d.id = j.document_id
+        and lower(btrim(split_part(d.file_type, ';', 1))) = any(p_allowed_file_types)))
+      and 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;
+    -- File type can change while authority is acquired. Recheck the locked document
+    -- before minting a lease or incrementing attempts; the client snapshot is not authority.
+    if p_allowed_file_types is not null and not coalesce(
+      lower(btrim(split_part(v_locked->'document'->>'file_type', ';', 1))) = any(p_allowed_file_types), false)
+      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 and allowlist checks are snapshots. Recheck them on the
+    -- locked row: authority, sibling leases and file type were all 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
+      or (p_allowed_document_ids is not null and not coalesce(v_job.document_id = any(p_allowed_document_ids), false))
+      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, 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;

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

@@ -0,0 +1,101 @@
+-- Disposable database only. A job re-pointed at another document between the claim's
+-- candidate pick and its job row lock must stay unclaimed (follow-up to #3628).
+--
+-- 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 picked PDF', 'picked.pdf', 'application/pdf', 'synthetic/recheck-picked', 'queued', '{}'),
+  ('00000000-0000-4000-8000-00000000000b', '11111111-1111-4111-8111-111111111111',
+   'Synthetic out-of-scope TXT', 'other.txt', 'text/plain', 'synthetic/recheck-txt', 'queued', '{}'),
+  ('00000000-0000-4000-8000-00000000000c', '11111111-1111-4111-8111-111111111111',
+   'Synthetic in-scope PDF', 'other.pdf', 'application/pdf', 'synthetic/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 scoped 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,
+  array['00000000-0000-4000-8000-00000000000a']::uuid[], null::text[]) 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 in-scope job must still be claimed');
+
+-- 1. Document allowlist: the job moves to a document outside the allowlist mid-claim.
+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,
+  array['00000000-0000-4000-8000-00000000000a']::uuid[], null::text[])),
+  'a job re-pointed outside the document allowlist after the pick must not be claimed');
+select pg_temp.assert_true(pg_temp.job_untouched('00000000-0000-4000-8000-00000000000b'),
+  'the re-pointed job must stay pending without a lease or a consumed attempt');
+
+-- 2. File-type allowlist: the job moves to a TXT document while a PDF-only worker claims.
+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,
+  null::uuid[], array['application/pdf'])),
+  'a job re-pointed to a disallowed file type after the pick must not be claimed');
+select pg_temp.assert_true(pg_temp.job_untouched('00000000-0000-4000-8000-00000000000b'),
+  'the file-type re-pointed job must stay pending without a lease or a consumed attempt');
+
+-- 3. Unscoped: the new document is in scope, but its authority was never locked.
+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,
+  null::uuid[], null::text[])),
+  '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'),
+  'the in-scope re-pointed job must stay pending without a lease or a consumed attempt');
+
+-- The skipped job is not lost: the next claim picks it under its new document.
+truncate claims;
+insert into claims select job from public.claim_ingestion_jobs('synthetic-worker', 1, 45, null::uuid[], null::text[]) job;
+select pg_temp.assert_true((select count(*) = 1 and bool_and(job->>'document_id' = '00000000-0000-4000-8000-00000000000c'
+  and job->'documents'->>'id' = '00000000-0000-4000-8000-00000000000c') from claims),
+  'the next claim must lease the job with its current document''s authority');
+rollback;