334 lines
32 KiB
PL/PgSQL
334 lines
32 KiB
PL/PgSQL
-- 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_at<a.authority_expires_at) THEN RETURN false; END IF;
|
|
SELECT count(*) INTO n FROM public.server_poll_intents WHERE receipt_id=id;
|
|
SELECT count(*) INTO pubs FROM public.server_poll_intents WHERE receipt_id=id AND stage LIKE 'publish.%';
|
|
IF n<>pubs+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'<clock_timestamp();
|
|
END $$;
|
|
CREATE FUNCTION server_poll_storage_source_payload(receipt_uuid uuid) RETURNS jsonb LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$
|
|
DECLARE payload jsonb;
|
|
BEGIN
|
|
IF NOT public.server_poll_storage_compact_allowed(receipt_uuid) THEN RETURN NULL; END IF;
|
|
SELECT jsonb_build_object('admission',to_jsonb(a),'journal',jsonb_build_object(
|
|
'execution',(SELECT to_jsonb(e) FROM public.server_poll_executions e WHERE e.receipt_id=receipt_uuid),
|
|
'intents',(SELECT COALESCE(jsonb_agg(to_jsonb(i) ORDER BY i.stage),'[]'::jsonb) FROM public.server_poll_intents i WHERE i.receipt_id=receipt_uuid),
|
|
'acks',(SELECT COALESCE(jsonb_agg(to_jsonb(k) ORDER BY k.stage),'[]'::jsonb) FROM public.server_poll_acks k WHERE k.receipt_id=receipt_uuid),
|
|
'receipt',(SELECT to_jsonb(r) FROM public.server_poll_receipts r WHERE r.receipt_id=receipt_uuid),
|
|
'quarantines',(SELECT COALESCE(jsonb_agg(to_jsonb(q) ORDER BY q.created_at,q.receipt_id),'[]'::jsonb) FROM public.server_poll_quarantines q WHERE q.receipt_id=receipt_uuid)),
|
|
'terminal',jsonb_build_object('operation',to_jsonb(o),'plan',to_jsonb(p),'step',to_jsonb(s))) INTO payload
|
|
FROM public.server_poll_storage_admissions a JOIN public.mcp_operations o ON o.id=a.execution_id AND o.connection_id=a.connection_id
|
|
JOIN public.mcp_plans p ON p.id=o.plan_id AND p.connection_id=o.connection_id JOIN public.mcp_operation_steps s ON s.operation_id=o.id AND s.step_index=0
|
|
WHERE a.receipt_id=receipt_uuid;
|
|
IF octet_length(payload::text)>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 admissions<CASE WHEN scope='global' THEN 128 ELSE 8 END
|
|
AND reserved_rows<=CASE WHEN scope='global' THEN 514432 ELSE 32152 END-4019
|
|
AND reserved_bytes<=CASE WHEN scope='global' THEN 1077936128 ELSE 67371008 END-8421376)<>2 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'<clock_timestamp();
|
|
END $$;
|
|
CREATE FUNCTION server_poll_archive_immutable() RETURNS trigger LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$ BEGIN
|
|
IF TG_OP='DELETE' AND public.server_poll_storage_gc_allowed(OLD.receipt_id) THEN RETURN OLD; END IF;
|
|
RAISE EXCEPTION 'poll archive is immutable until original full retention GC';
|
|
END $$;
|
|
CREATE TRIGGER server_poll_archive_immutable BEFORE UPDATE OR DELETE ON server_poll_archives FOR EACH ROW EXECUTE FUNCTION server_poll_archive_immutable();
|
|
CREATE FUNCTION server_poll_archive_final_guard() RETURNS trigger LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$
|
|
DECLARE id uuid; a public.server_poll_storage_admissions%ROWTYPE;
|
|
BEGIN
|
|
id:=CASE WHEN TG_OP='DELETE' THEN OLD.receipt_id ELSE NEW.receipt_id END;
|
|
SELECT * INTO a FROM public.server_poll_storage_admissions WHERE receipt_id=id;
|
|
IF FOUND AND a.storage_state='compacted' AND(EXISTS(SELECT 1 FROM public.server_poll_executions WHERE receipt_id=id)
|
|
OR EXISTS(SELECT 1 FROM public.server_poll_intents WHERE receipt_id=id) OR EXISTS(SELECT 1 FROM public.server_poll_acks WHERE receipt_id=id)
|
|
OR EXISTS(SELECT 1 FROM public.server_poll_receipts WHERE receipt_id=id) OR EXISTS(SELECT 1 FROM public.server_poll_quarantines WHERE receipt_id=id))
|
|
THEN RAISE EXCEPTION 'compacted poll retained incomplete original children'; END IF;
|
|
IF NOT public.server_poll_storage_accounting_valid() THEN RAISE EXCEPTION 'poll final archive or exact ledger mismatch'; END IF;
|
|
RETURN NULL;
|
|
END $$;
|
|
CREATE CONSTRAINT TRIGGER server_poll_archive_final_guard AFTER INSERT OR DELETE ON server_poll_archives DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION server_poll_archive_final_guard();
|
|
CREATE CONSTRAINT TRIGGER server_poll_admission_final_guard AFTER INSERT OR DELETE ON server_poll_storage_admissions DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION server_poll_archive_final_guard();
|
|
CREATE CONSTRAINT TRIGGER server_poll_compact_final_guard AFTER UPDATE ON server_poll_storage_admissions DEFERRABLE INITIALLY DEFERRED FOR EACH ROW WHEN(OLD.storage_state IS DISTINCT FROM NEW.storage_state) EXECUTE FUNCTION server_poll_archive_final_guard();
|
|
-- Existing row guard additionally refuses compacted tombstones, including late
|
|
-- quarantine attempts; the original meter and immutable identity stay intact.
|
|
CREATE FUNCTION server_poll_compacted_row_guard() RETURNS trigger LANGUAGE plpgsql SET search_path=pg_catalog,public AS $$ BEGIN
|
|
PERFORM pg_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0));
|
|
IF EXISTS(SELECT 1 FROM public.server_poll_storage_admissions WHERE receipt_id=NEW.receipt_id AND storage_state='compacted') THEN RAISE EXCEPTION 'compacted poll journal cannot reopen'; END IF;
|
|
RETURN NEW;
|
|
END $$;
|
|
CREATE TRIGGER server_poll_compacted_anchor_guard BEFORE INSERT ON server_poll_executions FOR EACH ROW EXECUTE FUNCTION server_poll_compacted_row_guard();
|
|
CREATE TRIGGER server_poll_compacted_intent_guard BEFORE INSERT ON server_poll_intents FOR EACH ROW EXECUTE FUNCTION server_poll_compacted_row_guard();
|
|
CREATE TRIGGER server_poll_compacted_ack_guard BEFORE INSERT ON server_poll_acks FOR EACH ROW EXECUTE FUNCTION server_poll_compacted_row_guard();
|
|
CREATE TRIGGER server_poll_compacted_receipt_guard BEFORE INSERT ON server_poll_receipts FOR EACH ROW EXECUTE FUNCTION server_poll_compacted_row_guard();
|
|
CREATE TRIGGER server_poll_compacted_quarantine_guard BEFORE INSERT ON server_poll_quarantines FOR EACH ROW EXECUTE FUNCTION server_poll_compacted_row_guard();
|
|
INSERT INTO schema_migrations(id) VALUES('0041_server_poll_terminal_archives') ON CONFLICT(id) DO NOTHING;
|
|
COMMIT;
|