diff --git a/supabase/applied-migration-hashes.json b/supabase/applied-migration-hashes.json index a59887824d..870bd6b62f 100644 --- a/supabase/applied-migration-hashes.json +++ b/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" } } diff --git a/supabase/drift-manifest.json b/supabase/drift-manifest.json index 31e0779855..679b2eff60 100644 --- a/supabase/drift-manifest.json +++ b/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)" }, { diff --git a/supabase/migrations/20261018062736_recheck_legacy_ingestion_claim_document_after_job_lock.sql b/supabase/migrations/20261018062736_recheck_legacy_ingestion_claim_document_after_job_lock.sql new file mode 100644 index 0000000000..8a353af931 --- /dev/null +++ b/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; diff --git a/supabase/schema.sql b/supabase/schema.sql index ab95b86453..19da197013 100644 --- a/supabase/schema.sql +++ b/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; diff --git a/tests/sql/ingestion-legacy-claim-recheck-after-lock.sql b/tests/sql/ingestion-legacy-claim-recheck-after-lock.sql new file mode 100644 index 0000000000..340206db8e --- /dev/null +++ b/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;