-- Additive terminal poll archives. Existing 0040 full journals are retained. -- No compaction, retention run, source backfill or worker activation occurs here. BEGIN; DO $$ BEGIN PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0)); IF EXISTS(SELECT 1 FROM pg_class WHERE relname='server_poll_archives') OR EXISTS(SELECT 1 FROM pg_attribute WHERE attrelid='public.server_poll_storage_admissions'::regclass AND NOT attisdropped AND attname IN('storage_state','compact_at','archive_bytes','original_activity_at','compaction_claim','compaction_claimed_at','compaction_claim_xid')) OR EXISTS(SELECT 1 FROM pg_attribute WHERE attrelid='public.server_poll_storage_usage'::regclass AND NOT attisdropped AND attname IN('compacted_admissions','archive_bytes')) OR EXISTS(SELECT 1 FROM pg_proc WHERE proname IN('server_poll_storage_source_payload','server_poll_storage_compact_allowed','server_poll_storage_accounting_valid','server_poll_archive_insert_guard','server_poll_archive_insert_finish','server_poll_archive_immutable','server_poll_archive_final_guard','server_poll_compacted_row_guard')) OR EXISTS(SELECT 1 FROM pg_trigger WHERE tgname IN('server_poll_archive_insert_guard','server_poll_archive_insert_finish','server_poll_archive_immutable','server_poll_archive_final_guard','server_poll_admission_final_guard','server_poll_compact_final_guard','server_poll_compacted_anchor_guard','server_poll_compacted_intent_guard','server_poll_compacted_ack_guard','server_poll_compacted_receipt_guard','server_poll_compacted_quarantine_guard')) THEN RAISE EXCEPTION 'unmarked poll archive objects require explicit reconciliation'; END IF; END $$; LOCK TABLE server_poll_storage_usage,server_poll_storage_admissions,server_poll_executions,server_poll_intents,server_poll_acks,server_poll_receipts,server_poll_quarantines IN ACCESS EXCLUSIVE MODE; DO $$ BEGIN IF (SELECT count(*) FROM server_poll_storage_usage WHERE scope='global')<>1 OR EXISTS(SELECT 1 FROM server_poll_storage_usage u WHERE u.admissions<>(SELECT count(*) FROM server_poll_storage_admissions a WHERE u.scope='global' OR u.scope='owner:'||a.owner_id) OR u.reserved_rows<>u.admissions*4019 OR u.reserved_bytes<>u.admissions*8421376) OR EXISTS(SELECT 1 FROM server_poll_storage_admissions a WHERE NOT EXISTS(SELECT 1 FROM server_poll_storage_usage u WHERE u.scope='owner:'||a.owner_id)) THEN RAISE EXCEPTION 'existing poll accounting requires explicit reconciliation'; END IF; IF NOT EXISTS(SELECT 1 FROM pg_constraint WHERE conrelid='public.server_poll_storage_usage'::regclass AND conname='server_poll_storage_usage_check' AND pg_get_constraintdef(oid)='CHECK ((reserved_rows = (admissions * 4019)))') OR NOT EXISTS(SELECT 1 FROM pg_constraint WHERE conrelid='public.server_poll_storage_usage'::regclass AND conname='server_poll_storage_usage_check1' AND pg_get_constraintdef(oid)='CHECK ((reserved_bytes = (admissions * 8421376)))') THEN RAISE EXCEPTION 'original poll accounting constraints require explicit reconciliation'; END IF; END $$; ALTER TABLE server_poll_storage_admissions ADD COLUMN storage_state text NOT NULL DEFAULT 'full' CHECK(storage_state IN('full','compacted')), ADD COLUMN compact_at timestamptz, ADD COLUMN archive_bytes integer, ADD COLUMN original_activity_at timestamptz, ADD COLUMN compaction_claim uuid, ADD COLUMN compaction_claimed_at timestamptz, ADD COLUMN compaction_claim_xid xid8, ADD CONSTRAINT server_poll_compaction_claim_shape CHECK( (compaction_claim IS NULL AND compaction_claimed_at IS NULL AND compaction_claim_xid IS NULL) OR(compaction_claim IS NOT NULL AND compaction_claimed_at IS NOT NULL AND compaction_claim_xid IS NOT NULL AND isfinite(compaction_claimed_at))), ADD CONSTRAINT server_poll_storage_compact_shape CHECK( (storage_state='full' AND compact_at IS NULL AND archive_bytes IS NULL AND original_activity_at IS NULL) OR(storage_state='compacted' AND identity_kind='mcp' AND claim_token IS NULL AND claim_attempted_at IS NOT NULL AND compaction_claim IS NOT NULL AND compaction_claimed_at IS NOT NULL AND compaction_claim_xid IS NOT NULL AND compact_at IS NOT NULL AND archive_bytes IS NOT NULL AND original_activity_at IS NOT NULL AND isfinite(compact_at) AND archive_bytes BETWEEN 1 AND 131072 AND isfinite(original_activity_at) AND original_activity_at>=created_at AND compact_at>authority_expires_at+safety_ms*interval '1 millisecond')); ALTER TABLE server_poll_storage_usage DROP CONSTRAINT server_poll_storage_usage_check, DROP CONSTRAINT server_poll_storage_usage_check1, ADD COLUMN compacted_admissions bigint NOT NULL DEFAULT 0 CHECK(compacted_admissions>=0), ADD COLUMN archive_bytes bigint NOT NULL DEFAULT 0 CHECK(archive_bytes>=0), ADD CONSTRAINT server_poll_storage_rows_charge CHECK(reserved_rows=admissions*4019+compacted_admissions*2 AND reserved_rows<=CASE WHEN scope='global' THEN 514432 ELSE 32152 END), ADD CONSTRAINT server_poll_storage_bytes_charge CHECK(reserved_bytes=admissions*8421376+compacted_admissions*32768+archive_bytes AND reserved_bytes<=CASE WHEN scope='global' THEN 1077936128 ELSE 67371008 END), ADD CONSTRAINT server_poll_storage_archive_charge CHECK(archive_bytes>=compacted_admissions AND archive_bytes<=compacted_admissions*131072); CREATE TABLE server_poll_archives ( receipt_id uuid PRIMARY KEY REFERENCES server_poll_storage_admissions(receipt_id),version smallint NOT NULL CHECK(version=1), execution_id uuid NOT NULL UNIQUE,connection_id uuid NOT NULL,owner_id text NOT NULL, binding_proof text NOT NULL CHECK(binding_proof ~ '^[a-f0-9]{64}$'),context_proof text NOT NULL CHECK(context_proof ~ '^[a-f0-9]{64}$'), source_digest text NOT NULL CHECK(source_digest ~ '^[a-f0-9]{64}$'),source_sha256 text NOT NULL CHECK(source_sha256 ~ '^[a-f0-9]{64}$'), terminal_digest text NOT NULL CHECK(terminal_digest ~ '^[a-f0-9]{64}$'), original_activity_at timestamptz NOT NULL CHECK(isfinite(original_activity_at)),original_expires_at timestamptz NOT NULL CHECK(isfinite(original_expires_at)), decoded_bytes integer NOT NULL CHECK(decoded_bytes BETWEEN 1 AND 16777216),compressed_bytes integer NOT NULL CHECK(compressed_bytes BETWEEN 1 AND 98048), ciphertext text NOT NULL,ciphertext_bytes integer NOT NULL CHECK(ciphertext_bytes BETWEEN 1 AND 131072 AND ciphertext_bytes=octet_length(ciphertext)), archive_proof text NOT NULL CHECK(archive_proof ~ '^[a-f0-9]{64}$'), created_at timestamptz NOT NULL,compaction_xid xid8 NOT NULL ); -- SQL checks structure/ownership/current rows. The fixed private producer and -- historical reader additionally authenticate every existing keyed proof and -- the archive MAC; SQL never receives an encryption/HMAC key or boolean grant. CREATE FUNCTION server_poll_storage_compact_allowed(id uuid) RETURNS boolean LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$ DECLARE a public.server_poll_storage_admissions%ROWTYPE; o public.mcp_operations%ROWTYPE; p public.mcp_plans%ROWTYPE; s public.mcp_operation_steps%ROWTYPE; n integer; pubs integer; activity timestamptz; BEGIN PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0)); SELECT * INTO a FROM public.server_poll_storage_admissions WHERE receipt_id=id FOR UPDATE; IF NOT FOUND OR a.storage_state<>'full' OR a.identity_kind<>'mcp' OR a.claim_token IS NOT NULL OR a.claim_attempted_at IS NULL OR a.authority_expires_at+a.safety_ms*interval '1 millisecond'>=clock_timestamp() OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=id OR(identity_kind=a.identity_kind AND execution_id=a.execution_id)) THEN RETURN false; END IF; IF NOT pg_try_advisory_xact_lock(hashtextextended('mcp-operation:'||a.execution_id::text,0)) THEN RETURN false; END IF; IF NOT pg_try_advisory_xact_lock(hashtextextended('server-poll-execution-v1:mcp:'||a.execution_id::text,0)) THEN RETURN false; END IF; IF NOT pg_try_advisory_xact_lock(hashtextextended('server-poll-receipt-v1:'||a.receipt_id::text,0)) THEN RETURN false; END IF; BEGIN SELECT * INTO o FROM public.mcp_operations WHERE mcp_operations.id=a.execution_id AND connection_id=a.connection_id FOR UPDATE NOWAIT; IF NOT FOUND OR o.state<>'applied' OR o.verification<>'passed' OR o.error_code IS NOT NULL OR o.finished_at IS NULL THEN RETURN false; END IF; SELECT * INTO p FROM public.mcp_plans WHERE mcp_plans.id=o.plan_id AND connection_id=a.connection_id FOR UPDATE NOWAIT; IF NOT FOUND OR p.operation_id<>'servers.poll' OR p.operation_version<>1 OR p.state<>'completed' OR p.approved_by IS NULL OR p.approved_at IS NULL OR p.created_at>p.approved_at OR p.approved_at>o.created_at OR o.created_at>=LEAST(p.expires_at,a.authority_expires_at) THEN RETURN false; END IF; SELECT * INTO s FROM public.mcp_operation_steps WHERE operation_id=o.id AND step_index=0 FOR UPDATE NOWAIT; IF NOT FOUND OR s.state<>'verified' OR s.server_id IS NOT NULL OR s.result IS NULL OR s.finished_at IS NULL OR(SELECT count(*) FROM public.mcp_operation_steps WHERE operation_id=o.id)<>1 OR NOT EXISTS(SELECT 1 FROM public.mcp_connections c WHERE c.id=a.connection_id AND c.owner_id=a.owner_id) OR s.intended_state->>'kind' IS DISTINCT FROM 'service' OR s.intended_state->>'service' IS DISTINCT FROM 'servers.poll' OR s.intended_state#>>'{input,receiptId}' IS DISTINCT FROM id::text OR s.result->>'operationId' IS DISTINCT FROM o.id::text OR s.result->>'receiptId' IS DISTINCT FROM id::text OR s.result->'nativeReads' IS DISTINCT FROM '4'::jsonb OR s.result->'complete' IS DISTINCT FROM 'true'::jsonb OR s.result->'maintenanceComplete' IS DISTINCT FROM 'true'::jsonb OR s.result->'configurationVerified' IS DISTINCT FROM 'true'::jsonb OR s.result->'automaticReplay' IS DISTINCT FROM 'false'::jsonb THEN RETURN false; END IF; EXCEPTION WHEN lock_not_available THEN RETURN false; END; IF NOT EXISTS(SELECT 1 FROM public.server_poll_executions e WHERE e.receipt_id=id AND e.identity_kind=a.identity_kind AND e.execution_id=a.execution_id AND e.binding_proof=a.binding_proof AND e.context_proof=a.context_proof) OR NOT EXISTS(SELECT 1 FROM public.server_poll_receipts r WHERE r.receipt_id=id AND r.created_atpubs+4 OR pubs NOT IN(a.max_stages-6,a.max_stages-4) OR a.used_rows<>2*n+2 OR(SELECT count(*) FROM public.server_poll_acks WHERE receipt_id=id)<>n OR s.result->'confirmedPublications' IS DISTINCT FROM to_jsonb(pubs) OR s.result->'completedPublications' IS DISTINCT FROM to_jsonb(pubs) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=id AND outcome<>'confirmed') OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=id AND created_at>=a.authority_expires_at) OR EXISTS(SELECT 1 FROM unnest(ARRAY['native.identity','native.resource','native.interfaces','native.addresses']) wanted(stage) WHERE NOT EXISTS(SELECT 1 FROM public.server_poll_intents i JOIN public.server_poll_acks k USING(receipt_id,stage) WHERE i.receipt_id=id AND i.stage=wanted.stage AND k.outcome='confirmed')) OR EXISTS(SELECT 1 FROM generate_series(0,pubs-1) wanted(stage) WHERE NOT EXISTS(SELECT 1 FROM public.server_poll_intents i JOIN public.server_poll_acks k USING(receipt_id,stage) WHERE i.receipt_id=id AND i.stage='publish.'||wanted.stage::text AND k.outcome='confirmed')) THEN RETURN false; END IF; SELECT GREATEST(a.authority_expires_at,a.created_at,a.claim_attempted_at,o.finished_at,p.expires_at,s.finished_at, (SELECT created_at FROM public.server_poll_executions WHERE receipt_id=id),(SELECT max(created_at) FROM public.server_poll_intents WHERE receipt_id=id), (SELECT max(created_at) FROM public.server_poll_acks WHERE receipt_id=id),(SELECT created_at FROM public.server_poll_receipts WHERE receipt_id=id)) INTO activity; RETURN isfinite(activity) AND activity+a.safety_ms*interval '1 millisecond'16777216 THEN RETURN NULL; END IF; RETURN payload; END $$; CREATE FUNCTION server_poll_storage_accounting_valid() RETURNS boolean LANGUAGE sql SET search_path=pg_catalog,public AS $$ WITH charges AS ( SELECT owner_id,CASE WHEN storage_state='full' THEN 1 ELSE 0 END::bigint AS heavy, CASE WHEN storage_state='compacted' THEN 1 ELSE 0 END::bigint AS compact, COALESCE(archive_bytes,0)::bigint AS archives, CASE WHEN storage_state='full' THEN 4019 ELSE 2 END::bigint AS rows, CASE WHEN storage_state='full' THEN 8421376 ELSE 32768+archive_bytes END::bigint AS bytes FROM public.server_poll_storage_admissions ), expected AS ( SELECT 'global' AS scope,COALESCE(sum(heavy),0) AS heavy,COALESCE(sum(compact),0) AS compact,COALESCE(sum(archives),0) AS archives,COALESCE(sum(rows),0) AS rows,COALESCE(sum(bytes),0) AS bytes FROM charges UNION ALL SELECT 'owner:'||owner_id,sum(heavy),sum(compact),sum(archives),sum(rows),sum(bytes) FROM charges GROUP BY owner_id ) SELECT NOT EXISTS(SELECT 1 FROM public.server_poll_storage_usage u FULL JOIN expected e USING(scope) WHERE u.scope IS NULL OR u.admissions<>COALESCE(e.heavy,0) OR u.compacted_admissions<>COALESCE(e.compact,0) OR u.archive_bytes<>COALESCE(e.archives,0) OR u.reserved_rows<>COALESCE(e.rows,0) OR u.reserved_bytes<>COALESCE(e.bytes,0)) AND NOT EXISTS(SELECT 1 FROM public.server_poll_storage_admissions a LEFT JOIN public.server_poll_archives r USING(receipt_id) WHERE (a.storage_state='full' AND r.receipt_id IS NOT NULL) OR(a.storage_state='compacted' AND(r.receipt_id IS NULL OR a.archive_bytes<>r.ciphertext_bytes))) $$; CREATE FUNCTION server_poll_archive_insert_guard() RETURNS trigger LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$ DECLARE a public.server_poll_storage_admissions%ROWTYPE; payload jsonb; activity timestamptz; BEGIN payload:=public.server_poll_storage_source_payload(NEW.receipt_id); IF payload IS NULL OR NOT public.server_poll_storage_accounting_valid() THEN RAISE EXCEPTION 'poll source is not eligible for terminal archive'; END IF; SELECT * INTO a FROM public.server_poll_storage_admissions WHERE receipt_id=NEW.receipt_id FOR UPDATE; SELECT GREATEST(a.authority_expires_at,a.created_at,a.claim_attempted_at,(payload#>>'{terminal,operation,finished_at}')::timestamptz, (payload#>>'{terminal,plan,expires_at}')::timestamptz,(payload#>>'{terminal,step,finished_at}')::timestamptz, (payload#>>'{journal,execution,created_at}')::timestamptz,(payload#>>'{journal,receipt,created_at}')::timestamptz, (SELECT max((v->>'created_at')::timestamptz) FROM jsonb_array_elements(payload#>'{journal,intents}') v), (SELECT max((v->>'created_at')::timestamptz) FROM jsonb_array_elements(payload#>'{journal,acks}') v)) INTO activity; IF a.compaction_claim IS NULL OR a.compaction_claimed_at IS NULL OR NOT isfinite(a.compaction_claimed_at) OR a.compaction_claim_xid IS NULL OR a.compaction_claim_xid=pg_current_xact_id() OR NEW.version<>1 OR NEW.execution_id<>a.execution_id OR NEW.connection_id<>a.connection_id OR NEW.owner_id<>a.owner_id OR NEW.binding_proof<>a.binding_proof OR NEW.context_proof<>a.context_proof OR NEW.original_expires_at<>a.authority_expires_at OR NEW.original_activity_at<>activity OR NEW.decoded_bytes<>octet_length(payload::text) OR NEW.source_sha256<>encode(sha256(convert_to(payload::text,'UTF8')),'hex') OR NEW.ciphertext_bytes<>octet_length(NEW.ciphertext) OR NEW.ciphertext_bytes>131072 OR NEW.ciphertext !~ '^[A-Za-z0-9+/]+={0,2}$' OR replace(encode(decode(NEW.ciphertext,'base64'),'base64'),E'\n','')<>NEW.ciphertext OR octet_length(decode(NEW.ciphertext,'base64'))<>NEW.compressed_bytes+28 THEN RAISE EXCEPTION 'poll archive does not match the entire immutable source'; END IF; NEW.created_at:=clock_timestamp(); NEW.compaction_xid:=pg_current_xact_id(); RETURN NEW; END $$; CREATE TRIGGER server_poll_archive_insert_guard BEFORE INSERT ON server_poll_archives FOR EACH ROW EXECUTE FUNCTION server_poll_archive_insert_guard(); CREATE OR REPLACE FUNCTION server_poll_journal_immutable() RETURNS trigger LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$ BEGIN IF TG_OP='DELETE' AND EXISTS(SELECT 1 FROM public.server_poll_archives r JOIN public.server_poll_storage_admissions a USING(receipt_id) WHERE r.receipt_id=OLD.receipt_id AND r.compaction_xid=pg_current_xact_id() AND a.storage_state='full') THEN RETURN OLD; END IF; IF TG_OP='DELETE' AND public.server_poll_storage_gc_allowed(OLD.receipt_id) THEN RETURN OLD; END IF; RAISE EXCEPTION 'server poll journal is immutable outside complete archive or original lifetime GC'; END $$; CREATE OR REPLACE FUNCTION server_poll_storage_admission_guard() RETURNS trigger LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$ DECLARE retention integer; charged integer; r public.server_poll_archives%ROWTYPE; BEGIN PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0)); IF TG_OP='DELETE' THEN IF NOT public.server_poll_storage_gc_allowed(OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_executions WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_intents WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_receipts WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_archives WHERE receipt_id=OLD.receipt_id) THEN RAISE EXCEPTION 'poll lifetime or children prevent admission deletion'; END IF; UPDATE public.server_poll_storage_usage SET admissions=admissions-CASE WHEN OLD.storage_state='full' THEN 1 ELSE 0 END, compacted_admissions=compacted_admissions-CASE WHEN OLD.storage_state='compacted' THEN 1 ELSE 0 END, archive_bytes=archive_bytes-COALESCE(OLD.archive_bytes,0), reserved_rows=reserved_rows-CASE WHEN OLD.storage_state='full' THEN 4019 ELSE 2 END, reserved_bytes=reserved_bytes-CASE WHEN OLD.storage_state='full' THEN 8421376 ELSE 32768+OLD.archive_bytes END WHERE scope IN('global','owner:'||OLD.owner_id); GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll quota refund mismatch'; END IF; RETURN OLD; END IF; IF TG_OP='UPDATE' THEN -- This durable one-shot marker is written and ACKed in its OWN transaction -- before archive construction. A committed/unknown claim never resets and -- the worker never retries destructive work against an already claimed full -- source. Claim time is maintenance metadata, not renewed execution activity. IF OLD.storage_state='full' AND NEW.storage_state='full' AND OLD.compaction_claim IS NULL AND OLD.compaction_claimed_at IS NULL AND OLD.compaction_claim_xid IS NULL AND NEW.compaction_claim IS NOT NULL AND NEW.compaction_claim_xid IS NULL AND(to_jsonb(OLD)-ARRAY['compaction_claim','compaction_claimed_at','compaction_claim_xid'])=(to_jsonb(NEW)-ARRAY['compaction_claim','compaction_claimed_at','compaction_claim_xid']) AND public.server_poll_storage_compact_allowed(OLD.receipt_id) THEN NEW.compaction_claimed_at:=clock_timestamp(); NEW.compaction_claim_xid:=pg_current_xact_id(); RETURN NEW; END IF; IF(to_jsonb(OLD)-ARRAY['claim_token','claim_attempted_at','used_rows','used_bytes','storage_state','compact_at','archive_bytes','original_activity_at']) IS DISTINCT FROM (to_jsonb(NEW)-ARRAY['claim_token','claim_attempted_at','used_rows','used_bytes','storage_state','compact_at','archive_bytes','original_activity_at']) THEN RAISE EXCEPTION 'poll admission identity is immutable'; END IF; IF OLD.storage_state<>'full' THEN RAISE EXCEPTION 'compacted poll identity cannot reopen or change'; END IF; IF NEW.storage_state='compacted' THEN SELECT * INTO r FROM public.server_poll_archives WHERE receipt_id=OLD.receipt_id AND compaction_xid=pg_current_xact_id(); IF NOT FOUND OR NEW.used_rows<>OLD.used_rows OR NEW.used_bytes<>OLD.used_bytes OR NEW.claim_token IS DISTINCT FROM OLD.claim_token OR NEW.claim_attempted_at IS DISTINCT FROM OLD.claim_attempted_at OR NEW.archive_bytes<>r.ciphertext_bytes OR NEW.original_activity_at<>r.original_activity_at OR EXISTS(SELECT 1 FROM public.server_poll_executions WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_intents WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_receipts WHERE receipt_id=OLD.receipt_id) OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=OLD.receipt_id) THEN RAISE EXCEPTION 'poll compaction requires the complete current-transaction archive'; END IF; NEW.compact_at:=r.created_at; UPDATE public.server_poll_storage_usage SET admissions=admissions-1,compacted_admissions=compacted_admissions+1, archive_bytes=archive_bytes+r.ciphertext_bytes,reserved_rows=reserved_rows-4019+2,reserved_bytes=reserved_bytes-8421376+32768+r.ciphertext_bytes WHERE scope IN('global','owner:'||OLD.owner_id) AND admissions>=1; GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll compaction charge mismatch'; END IF; RETURN NEW; END IF; IF NEW.compact_at IS NOT NULL OR NEW.archive_bytes IS NOT NULL OR NEW.original_activity_at IS NOT NULL THEN RAISE EXCEPTION 'full poll cannot carry archive metadata'; END IF; IF OLD.claim_token IS NOT NULL AND NEW.claim_token IS NULL AND OLD.claim_attempted_at IS NULL AND NEW.claim_attempted_at IS NOT NULL AND NEW.used_rows=OLD.used_rows AND NEW.used_bytes=OLD.used_bytes AND OLD.authority_expires_at>clock_timestamp() THEN NEW.claim_attempted_at:=clock_timestamp(); RETURN NEW; END IF; IF OLD.claim_token IS NOT DISTINCT FROM NEW.claim_token AND OLD.claim_attempted_at IS NOT DISTINCT FROM NEW.claim_attempted_at AND NEW.used_rows=OLD.used_rows+1 AND NEW.used_bytes>OLD.used_bytes THEN RETURN NEW; END IF; RAISE EXCEPTION 'poll admission cannot refresh, reopen or refund itself'; END IF; IF NEW.storage_state<>'full' OR NEW.compact_at IS NOT NULL OR NEW.archive_bytes IS NOT NULL OR NEW.original_activity_at IS NOT NULL OR NEW.compaction_claim IS NOT NULL OR NEW.compaction_claimed_at IS NOT NULL OR NEW.compaction_claim_xid IS NOT NULL OR NOT public.server_poll_storage_accounting_valid() THEN RAISE EXCEPTION 'new poll requires full reservation and exact ledger'; END IF; SELECT requests_retention_days INTO retention FROM public.mcp_settings WHERE id=1; IF retention IS NULL OR retention NOT BETWEEN 1 AND 365 OR NEW.authority_issued_at>clock_timestamp() OR NEW.authority_expires_at<=clock_timestamp() OR NEW.claim_token IS NULL OR NEW.claim_attempted_at IS NOT NULL OR NEW.used_rows<>0 OR NEW.used_bytes<>0 THEN RAISE EXCEPTION 'new poll admission needs original live authority'; END IF; IF EXISTS(SELECT 1 FROM public.server_poll_storage_admissions WHERE receipt_id=NEW.receipt_id OR(identity_kind=NEW.identity_kind AND execution_id=NEW.execution_id)) THEN RAISE EXCEPTION 'poll identity already spent'; END IF; NEW.created_at:=clock_timestamp(); NEW.retention_days:=retention; NEW.safety_ms:=60000; NEW.gc_not_before:=NEW.authority_expires_at+make_interval(days=>retention)+interval '60 seconds'; INSERT INTO public.server_poll_storage_usage(scope) VALUES('owner:'||NEW.owner_id) ON CONFLICT DO NOTHING; IF(SELECT count(*) FROM public.server_poll_storage_usage WHERE scope IN('global','owner:'||NEW.owner_id) AND admissions2 THEN RAISE EXCEPTION 'poll admission quota exhausted'; END IF; UPDATE public.server_poll_storage_usage SET admissions=admissions+1,reserved_rows=reserved_rows+4019,reserved_bytes=reserved_bytes+8421376 WHERE scope IN('global','owner:'||NEW.owner_id); GET DIAGNOSTICS charged=ROW_COUNT; IF charged<>2 THEN RAISE EXCEPTION 'poll admission quota unavailable'; END IF; RETURN NEW; END $$; CREATE FUNCTION server_poll_archive_insert_finish() RETURNS trigger LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$ BEGIN DELETE FROM public.server_poll_acks WHERE receipt_id=NEW.receipt_id; DELETE FROM public.server_poll_intents WHERE receipt_id=NEW.receipt_id; DELETE FROM public.server_poll_receipts WHERE receipt_id=NEW.receipt_id; DELETE FROM public.server_poll_executions WHERE receipt_id=NEW.receipt_id; UPDATE public.server_poll_storage_admissions SET storage_state='compacted',compact_at=NEW.created_at,archive_bytes=NEW.ciphertext_bytes,original_activity_at=NEW.original_activity_at WHERE receipt_id=NEW.receipt_id AND storage_state='full'; IF NOT FOUND THEN RAISE EXCEPTION 'poll admission did not transition with archive'; END IF; RETURN NEW; END $$; CREATE TRIGGER server_poll_archive_insert_finish AFTER INSERT ON server_poll_archives FOR EACH ROW EXECUTE FUNCTION server_poll_archive_insert_finish(); CREATE OR REPLACE FUNCTION server_poll_storage_gc_allowed(id uuid) RETURNS boolean LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$ DECLARE a public.server_poll_storage_admissions%ROWTYPE; retention integer; activity timestamptz; terminal_at timestamptz; BEGIN IF current_setting('mm.poll_storage_gc',true) IS DISTINCT FROM 'on' THEN RETURN false; END IF; PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0)); SELECT * INTO a FROM public.server_poll_storage_admissions WHERE receipt_id=id FOR UPDATE; IF NOT FOUND OR a.authority_expires_at>=clock_timestamp() OR a.gc_not_before>=clock_timestamp() THEN RETURN false; END IF; SELECT requests_retention_days INTO retention FROM public.mcp_settings WHERE mcp_settings.id=1 FOR SHARE; IF retention IS NULL OR retention NOT BETWEEN 1 AND 365 THEN RETURN false; END IF; terminal_at:=a.created_at; IF a.identity_kind='mcp' THEN IF NOT pg_try_advisory_xact_lock(hashtextextended('mcp-operation:'||a.execution_id::text,0)) THEN RETURN false; END IF; BEGIN SELECT GREATEST(o.finished_at,p.expires_at) INTO terminal_at FROM public.mcp_operations o JOIN public.mcp_plans p ON p.id=o.plan_id AND p.connection_id=o.connection_id JOIN public.mcp_connections c ON c.id=o.connection_id WHERE o.id=a.execution_id AND o.connection_id=a.connection_id AND c.owner_id=a.owner_id AND p.operation_id='servers.poll' AND o.state IN('applied','failed','unknown','rejected','partial','cancelled') AND o.finished_at IS NOT NULL FOR UPDATE OF o,p NOWAIT; IF NOT FOUND OR terminal_at IS NULL THEN RETURN false; END IF; EXCEPTION WHEN lock_not_available THEN RETURN false; END; END IF; IF a.storage_state='compacted' THEN activity:=GREATEST(a.original_activity_at,a.authority_expires_at,terminal_at); ELSE SELECT GREATEST(a.authority_expires_at,a.created_at,COALESCE(a.claim_attempted_at,a.created_at),terminal_at, COALESCE((SELECT created_at FROM public.server_poll_executions WHERE receipt_id=id),a.created_at), COALESCE((SELECT max(created_at) FROM public.server_poll_intents WHERE receipt_id=id),a.created_at), COALESCE((SELECT max(created_at) FROM public.server_poll_acks WHERE receipt_id=id),a.created_at), COALESCE((SELECT created_at FROM public.server_poll_receipts WHERE receipt_id=id),a.created_at), COALESCE((SELECT created_at FROM public.server_poll_quarantines WHERE receipt_id=id),a.created_at)) INTO activity; END IF; RETURN isfinite(activity) AND activity+make_interval(days=>GREATEST(a.retention_days,retention))+a.safety_ms*interval '1 millisecond'