diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index 5c85ffbd..79bc47ef 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -88,6 +88,9 @@ mod generation_upgrade; #[path = "snapshot_generation_history_fixture.rs"] mod generation_history_fixture; +#[path = "snapshot_generation_qualified_fixture.rs"] +mod generation_qualified_fixture; + #[derive(Default)] struct ReadCounts { whole: AtomicUsize, diff --git a/src/api/router/snapshot_generation_history_fixture.rs b/src/api/router/snapshot_generation_history_fixture.rs index 9a694424..0f6e7005 100644 --- a/src/api/router/snapshot_generation_history_fixture.rs +++ b/src/api/router/snapshot_generation_history_fixture.rs @@ -1,6 +1,7 @@ use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; pub(super) async fn restore_empty_g1_schema(db: &DatabaseConnection) { + super::generation_qualified_fixture::restore_empty_history_schema(db).await; let count = db .query_one_raw(Statement::from_string( DbBackend::Postgres, diff --git a/src/api/router/snapshot_generation_qualified_fixture.rs b/src/api/router/snapshot_generation_qualified_fixture.rs new file mode 100644 index 00000000..e7020864 --- /dev/null +++ b/src/api/router/snapshot_generation_qualified_fixture.rs @@ -0,0 +1,56 @@ +use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; + +pub(super) async fn restore_empty_history_schema(db: &DatabaseConnection) { + let n:i64=db.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT (SELECT count(*) FROM mst2_metadata_current)+(SELECT count(*) FROM mst2_metadata_graph_node) + +(SELECT count(*) FROM mst2_metadata_graph_root)+(SELECT count(*) FROM mst2_metadata_graph_edge) + +(SELECT count(*) FROM mst2_metadata_gc_op) AS n")) + .await.unwrap().unwrap().try_get("","n").unwrap(); + assert_eq!( + n, 0, + "upgrade fixture has no qualified or current incarnation to erase" + ); + db.execute_unprepared( + "DO $$ DECLARE t text; tr text; BEGIN + FOREACH t IN ARRAY ARRAY['mst2_metadata_lifetime','mst2_metadata_current','mst2_metadata_payload', + 'mst2_metadata_prepare','mst2_metadata_prepare_page','mst2_retention_node','mst2_retention_edge', + 'mst2_retention_root','mst2_retention_gc_op','mst2_snapshot_context','mst2_snapshot_lease'] LOOP + FOREACH tr IN ARRAY ARRAY['mst2_metadata_statement_barrier','mst2_metadata_generic_domain_guard', + 'mst2_metadata_session_domain_guard','mst2_metadata_qualified_mapping_guard', + 'mst2_metadata_lifetime_insert_guard','mst2_metadata_lifetime_removed', + 'mst2_metadata_current_insert_guard','mst2_metadata_current_protected', + 'mst2_metadata_payload_fenced','mst2_metadata_payload_removed'] LOOP + EXECUTE format('DROP TRIGGER IF EXISTS %I ON %I',tr,t); + END LOOP; + END LOOP; + END $$; + DROP TABLE mst2_metadata_graph_root,mst2_metadata_graph_edge,mst2_metadata_graph_node,mst2_metadata_gc_op; + ALTER TABLE mst2_metadata_lifetime DROP COLUMN graph_domain; + ALTER TABLE mst2_metadata_prepare_page DROP CONSTRAINT mst2_metadata_prepare_page_prepare_id_page_id_generation_key; + ALTER TABLE mst2_metadata_prepare DROP CONSTRAINT mst2_metadata_prepare_prepare_id_storage_seal_key; + DROP INDEX idx_mst2_snapshot_context_metadata_root; + DO $$ DECLARE function_row record; BEGIN + FOR function_row IN SELECT proc.oid::regprocedure::text AS signature FROM pg_proc proc JOIN pg_namespace n ON n.oid=proc.pronamespace + WHERE n.nspname=current_schema() AND proc.proname LIKE 'mst2_metadata_%' + AND proc.proname NOT IN ('mst2_metadata_lifetime_guard','mst2_metadata_current_guard', + 'mst2_metadata_generation_seal_guard','mst2_metadata_generation_mapping_guard','mst2_metadata_payload_immutable') LOOP + EXECUTE 'DROP FUNCTION '||function_row.signature; + END LOOP; + END $$; + CREATE OR REPLACE FUNCTION mst2_metadata_current_guard() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'metadata current watermark requires generation-fenced collection'; END $$; + CREATE OR REPLACE FUNCTION mst2_metadata_lifetime_guard() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF TG_OP='DELETE' THEN RAISE EXCEPTION 'metadata lifetime watermark cannot be deleted'; END IF; + IF NEW.page_id IS DISTINCT FROM OLD.page_id OR NEW.node_id IS DISTINCT FROM OLD.node_id + OR NEW.generation IS DISTINCT FROM OLD.generation OR NEW.metadata_codec IS DISTINCT FROM OLD.metadata_codec + OR NEW.expected_size IS DISTINCT FROM OLD.expected_size THEN RAISE EXCEPTION 'metadata lifetime identity is immutable'; END IF; + IF NEW.state IS DISTINCT FROM OLD.state AND NOT (OLD.state='RESERVED' AND NEW.state='LIVE') THEN + RAISE EXCEPTION 'metadata lifetime transition requires generation collector'; END IF; + RETURN NEW; + END $$; + CREATE TRIGGER mst2_metadata_payload_immutable BEFORE UPDATE OR DELETE ON mst2_metadata_payload + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_payload_immutable(); + DELETE FROM seaql_migrations WHERE version='m20261007_000400_add_mst2_qualified_metadata_gc'" + ).await.unwrap(); +} diff --git a/src/callisto/mst2_metadata_lifetime.rs b/src/callisto/mst2_metadata_lifetime.rs index d9bc99da..ba0fa4bc 100644 --- a/src/callisto/mst2_metadata_lifetime.rs +++ b/src/callisto/mst2_metadata_lifetime.rs @@ -11,6 +11,7 @@ pub struct Model { pub state: String, pub metadata_codec: i16, pub expected_size: i32, + pub graph_domain: String, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/src/jupiter/migration/m20261007_000400_add_mst2_qualified_metadata_gc.rs b/src/jupiter/migration/m20261007_000400_add_mst2_qualified_metadata_gc.rs new file mode 100644 index 00000000..0faacb16 --- /dev/null +++ b/src/jupiter/migration/m20261007_000400_add_mst2_qualified_metadata_gc.rs @@ -0,0 +1,22 @@ +//! Exact-incarnation metadata graph and atomic, replayable payload collection. + +use mst2_codec::metapage::{HEADER_LEN, PAGE_MAX_BYTES}; +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let sql = include_str!("m20261007_000400_qualified_metadata_gc.sql") + .replace("$HEADER_LEN$", &HEADER_LEN.to_string()) + .replace("$PAGE_MAX_BYTES$", &PAGE_MAX_BYTES.to_string()); + manager.get_connection().execute_unprepared(&sql).await?; + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} diff --git a/src/jupiter/migration/m20261007_000400_qualified_metadata_gc.sql b/src/jupiter/migration/m20261007_000400_qualified_metadata_gc.sql new file mode 100644 index 00000000..a2a84cf0 --- /dev/null +++ b/src/jupiter/migration/m20261007_000400_qualified_metadata_gc.sql @@ -0,0 +1,639 @@ +LOCK TABLE mst2_metadata_lifetime,mst2_metadata_current,mst2_metadata_payload, + mst2_metadata_prepare,mst2_metadata_prepare_page,mst2_retention_node, + mst2_retention_edge,mst2_retention_root,mst2_retention_gc_op, + mst2_snapshot_context,mst2_snapshot_lease IN ACCESS EXCLUSIVE MODE; + +ALTER TABLE mst2_metadata_lifetime ADD COLUMN graph_domain text NOT NULL DEFAULT 'generic-v1' + CHECK (graph_domain IN ('generic-v1','qualified-v1')); +ALTER TABLE mst2_metadata_prepare_page ADD UNIQUE(prepare_id,page_id,generation); +ALTER TABLE mst2_metadata_prepare ADD UNIQUE(prepare_id,storage_seal); +CREATE INDEX idx_mst2_snapshot_context_metadata_root ON mst2_snapshot_context(metadata_root,snapshot_id); +CREATE INDEX idx_mst2_metadata_lifetime_qualified_scan ON mst2_metadata_lifetime(page_id,generation) + WHERE graph_domain='qualified-v1' AND state IN ('RESERVED','LIVE'); + +CREATE TABLE mst2_metadata_graph_node ( + page_id bytea NOT NULL CHECK (octet_length(page_id)=32), + generation bigint NOT NULL CHECK (generation>0), + state text NOT NULL CHECK (state IN ('LIVE','DELETING')), + metadata_codec smallint NOT NULL CHECK (metadata_codec=1), + bytes bigint NOT NULL CHECK (bytes BETWEEN $HEADER_LEN$ AND $PAGE_MAX_BYTES$), + incoming_refs bigint NOT NULL DEFAULT 0 CHECK (incoming_refs>=0), + PRIMARY KEY(page_id,generation), + FOREIGN KEY(page_id,generation) REFERENCES mst2_metadata_lifetime(page_id,generation) +); +CREATE INDEX idx_mst2_metadata_graph_node_gc ON mst2_metadata_graph_node(state,incoming_refs,page_id,generation); +CREATE TABLE mst2_metadata_graph_edge ( + parent_page bytea NOT NULL, + parent_generation bigint NOT NULL, + child_page bytea NOT NULL, + child_generation bigint NOT NULL, + PRIMARY KEY(parent_page,parent_generation,child_page,child_generation), + FOREIGN KEY(parent_page,parent_generation) REFERENCES mst2_metadata_graph_node(page_id,generation), + FOREIGN KEY(child_page,child_generation) REFERENCES mst2_metadata_graph_node(page_id,generation), + CHECK (parent_page<>child_page OR parent_generation<>child_generation) +); +CREATE INDEX idx_mst2_metadata_graph_edge_child ON mst2_metadata_graph_edge(child_page,child_generation,parent_page,parent_generation); +CREATE TABLE mst2_metadata_graph_root ( + prepare_id text NOT NULL, + storage_seal bytea NOT NULL CHECK (octet_length(storage_seal)=32), + page_id bytea NOT NULL, + generation bigint NOT NULL, + PRIMARY KEY(prepare_id,page_id,generation), + FOREIGN KEY(prepare_id,storage_seal) REFERENCES mst2_metadata_prepare(prepare_id,storage_seal), + FOREIGN KEY(prepare_id,page_id,generation) REFERENCES mst2_metadata_prepare_page(prepare_id,page_id,generation), + FOREIGN KEY(page_id,generation) REFERENCES mst2_metadata_graph_node(page_id,generation) +); +CREATE INDEX idx_mst2_metadata_graph_root_page ON mst2_metadata_graph_root(page_id,generation,prepare_id); +CREATE TABLE mst2_metadata_gc_op ( + operation_id uuid PRIMARY KEY, + page_id bytea NOT NULL CHECK (octet_length(page_id)=32), + generation bigint NOT NULL CHECK (generation>0), + primary_scope bytea NOT NULL CHECK (octet_length(primary_scope) BETWEEN 1 AND 16384), + graph_domain text NOT NULL CHECK (graph_domain='qualified-v1'), + metadata_codec smallint NOT NULL CHECK (metadata_codec=1), + expected_size integer NOT NULL CHECK (expected_size BETWEEN $HEADER_LEN$ AND $PAGE_MAX_BYTES$), + graph_present boolean NOT NULL, + had_payload boolean NOT NULL, + payload_delete_xid bigint, + state text NOT NULL CHECK (state IN ('PENDING','APPLIED')), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + completed_at timestamptz, + UNIQUE(page_id,generation), + FOREIGN KEY(page_id,generation) REFERENCES mst2_metadata_lifetime(page_id,generation), + CHECK ((state='APPLIED')=(completed_at IS NOT NULL)) +); +CREATE INDEX idx_mst2_metadata_gc_op_pending ON mst2_metadata_gc_op(created_at,operation_id) WHERE state='PENDING'; + +CREATE FUNCTION mst2_metadata_dml_barrier() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF pg_is_in_recovery() OR current_setting('transaction_isolation')<>'read committed' THEN + RAISE EXCEPTION 'metadata mutation requires primary READ COMMITTED'; + END IF; + PERFORM set_config('lock_timeout','5000ms',true); + PERFORM pg_advisory_xact_lock(1296717362,hashtext(current_schema())); + RETURN NULL; +END $$; +DO $$ DECLARE t text; BEGIN + FOREACH t IN ARRAY ARRAY['mst2_metadata_lifetime','mst2_metadata_current','mst2_metadata_payload', + 'mst2_metadata_prepare','mst2_metadata_prepare_page','mst2_metadata_graph_node', + 'mst2_metadata_graph_edge','mst2_metadata_graph_root','mst2_metadata_gc_op', + 'mst2_retention_node','mst2_retention_edge','mst2_retention_root','mst2_retention_gc_op', + 'mst2_snapshot_context','mst2_snapshot_lease'] LOOP + EXECUTE format('CREATE TRIGGER mst2_metadata_statement_barrier BEFORE INSERT OR UPDATE OR DELETE ON %I FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_dml_barrier()',t); + END LOOP; +END $$; + +CREATE FUNCTION mst2_metadata_scope_matches(s bytea) RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$ +DECLARE actual jsonb; +BEGIN + IF pg_is_in_recovery() OR current_setting('transaction_isolation')<>'read committed' THEN RETURN false; END IF; + SELECT jsonb_build_array(x.storage_uuid,current_database(),d.oid::bigint,current_schema(),n.oid::bigint, + inet_server_addr()::text,inet_server_port()) INTO actual + FROM mst2_metadata_storage_scope x JOIN pg_database d ON d.datname=current_database() + JOIN pg_namespace n ON n.nspname=current_schema() WHERE x.singleton=1; + RETURN actual IS NOT NULL AND convert_from(s,'UTF8')::jsonb=actual; +EXCEPTION WHEN OTHERS THEN RETURN false; +END $$; + +CREATE FUNCTION mst2_metadata_assert_uncovered(p bytea,g bigint) RETURNS void LANGUAGE plpgsql VOLATILE AS $$ +DECLARE nid text:='page:sha256:'||encode(p,'hex'); +BEGIN + IF EXISTS(SELECT 1 FROM mst2_metadata_graph_root WHERE page_id=p AND generation=g) + OR EXISTS(SELECT 1 FROM mst2_metadata_graph_edge WHERE child_page=p AND child_generation=g) + OR EXISTS(SELECT 1 FROM mst2_retention_node WHERE node_id=nid) + OR EXISTS(SELECT 1 FROM mst2_retention_edge WHERE parent_id=nid OR child_id=nid) + OR EXISTS(SELECT 1 FROM mst2_retention_root WHERE node_id=nid) + OR EXISTS(SELECT 1 FROM mst2_retention_gc_op WHERE node_id=nid) + OR EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + WHERE m.page_id=p AND (q.state='PREPARING' OR (q.state='COMMITTED' AND q.coverage_retired_at IS NULL))) + OR EXISTS(SELECT 1 FROM mst2_snapshot_context WHERE metadata_root=p) + OR EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_snapshot_context s USING(prepare_id) + WHERE m.page_id=p) THEN + RAISE EXCEPTION 'metadata lifetime has durable or ambiguous coverage'; + END IF; +END $$; + +CREATE FUNCTION mst2_metadata_gc_proof(p bytea,g bigint,stage text,graph_present boolean,payload_present boolean) +RETURNS void LANGUAGE plpgsql VOLATILE AS $$ +DECLARE l mst2_metadata_lifetime%ROWTYPE; n mst2_metadata_graph_node%ROWTYPE; b mst2_metadata_payload%ROWTYPE; +BEGIN + SELECT h.* INTO l FROM mst2_metadata_current c JOIN mst2_metadata_lifetime h USING(page_id,generation) + WHERE c.page_id=p AND c.generation=g FOR UPDATE OF c,h; + IF NOT FOUND OR l.graph_domain<>'qualified-v1' + OR (stage='CLAIM' AND l.state NOT IN ('RESERVED','LIVE')) + OR (stage='PENDING' AND l.state<>'DELETING') + OR (stage='REMOVED' AND l.state<>'REMOVED') + OR stage NOT IN ('CLAIM','PENDING','REMOVED') THEN + RAISE EXCEPTION 'metadata GC is not the exact current qualified lifetime'; + END IF; + PERFORM mst2_metadata_assert_uncovered(p,g); + SELECT * INTO n FROM mst2_metadata_graph_node WHERE page_id=p AND generation=g FOR UPDATE; + IF FOUND IS DISTINCT FROM graph_present THEN RAISE EXCEPTION 'metadata GC graph presence changed'; END IF; + IF graph_present AND (n.metadata_codec<>l.metadata_codec OR n.bytes<>l.expected_size OR n.incoming_refs<>0 + OR (stage='CLAIM' AND n.state<>'LIVE') OR (stage='PENDING' AND n.state<>'DELETING')) THEN + RAISE EXCEPTION 'metadata GC graph profile or counter changed'; + END IF; + IF graph_present AND (SELECT count(*) FROM (SELECT 1 FROM mst2_metadata_graph_edge + WHERE parent_page=p AND parent_generation=g LIMIT 16385) bounded)>16384 THEN + RAISE EXCEPTION 'metadata GC outgoing edge limit exceeded'; + END IF; + SELECT * INTO b FROM mst2_metadata_payload WHERE page_id=p FOR UPDATE; + IF FOUND IS DISTINCT FROM payload_present THEN RAISE EXCEPTION 'metadata GC payload presence changed'; END IF; + IF payload_present AND (b.generation IS DISTINCT FROM g OR b.metadata_codec<>l.metadata_codec OR b.byte_size<>l.expected_size) THEN + RAISE EXCEPTION 'metadata GC payload profile or generation changed'; + END IF; + IF stage='CLAIM' AND l.state='LIVE' AND NOT (graph_present AND payload_present) THEN + RAISE EXCEPTION 'LIVE metadata corruption is not collectable'; + END IF; + IF stage='CLAIM' AND l.state='RESERVED' AND (graph_present OR NOT EXISTS( + SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + WHERE m.page_id=p AND m.generation=g AND q.graph_domain='qualified-v1' + AND q.state='ABORTED' AND q.storage_seal IS NOT NULL AND m.expected_size=l.expected_size) + OR EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + WHERE m.page_id=p AND m.generation=g AND q.state<>'ABORTED')) THEN + RAISE EXCEPTION 'RESERVED metadata needs an explicitly aborted qualified origin'; + END IF; +END $$; + +CREATE FUNCTION mst2_metadata_gc_op_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE l mst2_metadata_lifetime%ROWTYPE; +BEGIN + IF TG_OP='DELETE' THEN RAISE EXCEPTION 'metadata GC evidence cannot be deleted'; END IF; + IF NOT mst2_metadata_scope_matches(NEW.primary_scope) THEN RAISE EXCEPTION 'metadata GC wrong primary scope'; END IF; + IF TG_OP='INSERT' THEN + IF NEW.state<>'PENDING' OR NEW.completed_at IS NOT NULL OR NEW.payload_delete_xid IS NOT NULL THEN RAISE EXCEPTION 'metadata GC must start PENDING'; END IF; + PERFORM mst2_metadata_gc_proof(NEW.page_id,NEW.generation,'CLAIM',NEW.graph_present,NEW.had_payload); + SELECT * INTO l FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND generation=NEW.generation; + IF NEW.graph_domain<>l.graph_domain OR NEW.metadata_codec<>l.metadata_codec OR NEW.expected_size<>l.expected_size THEN + RAISE EXCEPTION 'metadata GC immutable profile conflict'; + END IF; + NEW.created_at:=clock_timestamp(); + ELSE + IF ROW(NEW.operation_id,NEW.page_id,NEW.generation,NEW.primary_scope,NEW.graph_domain, + NEW.metadata_codec,NEW.expected_size,NEW.graph_present,NEW.had_payload,NEW.created_at) + IS DISTINCT FROM ROW(OLD.operation_id,OLD.page_id,OLD.generation,OLD.primary_scope,OLD.graph_domain, + OLD.metadata_codec,OLD.expected_size,OLD.graph_present,OLD.had_payload,OLD.created_at) THEN + RAISE EXCEPTION 'metadata GC identity cannot change'; + END IF; + IF OLD.state='APPLIED' AND (NEW.state<>OLD.state OR NEW.completed_at IS DISTINCT FROM OLD.completed_at + OR NEW.payload_delete_xid IS DISTINCT FROM OLD.payload_delete_xid) THEN + RAISE EXCEPTION 'metadata GC receipt cannot change'; + END IF; + IF OLD.state='PENDING' AND NEW.state='APPLIED' THEN + IF OLD.had_payload AND OLD.payload_delete_xid IS DISTINCT FROM txid_current() THEN + RAISE EXCEPTION 'metadata GC has no same-transaction payload delete proof'; + END IF; + PERFORM mst2_metadata_gc_proof(NEW.page_id,NEW.generation,'REMOVED',false,false); + NEW.completed_at:=clock_timestamp(); + ELSIF NEW.state IS DISTINCT FROM OLD.state OR NEW.completed_at IS DISTINCT FROM OLD.completed_at THEN + RAISE EXCEPTION 'metadata GC invalid receipt transition'; + END IF; + IF NEW.payload_delete_xid IS DISTINCT FROM OLD.payload_delete_xid THEN + IF OLD.state<>'PENDING' OR NEW.state<>'PENDING' OR NOT OLD.had_payload THEN + RAISE EXCEPTION 'invalid metadata delete transaction proof'; + END IF; + PERFORM mst2_metadata_gc_proof(OLD.page_id,OLD.generation,'PENDING',OLD.graph_present,true); + NEW.payload_delete_xid:=txid_current(); + END IF; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_gc_op_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_gc_op + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_gc_op_guard(); + +CREATE OR REPLACE FUNCTION mst2_metadata_lifetime_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op mst2_metadata_gc_op%ROWTYPE; +BEGIN + IF TG_OP='DELETE' THEN RAISE EXCEPTION 'metadata lifetime watermark cannot be deleted'; END IF; + IF ROW(NEW.page_id,NEW.node_id,NEW.generation,NEW.metadata_codec,NEW.expected_size,NEW.graph_domain) + IS DISTINCT FROM ROW(OLD.page_id,OLD.node_id,OLD.generation,OLD.metadata_codec,OLD.expected_size,OLD.graph_domain) THEN + RAISE EXCEPTION 'metadata lifetime identity is immutable'; + END IF; + IF NEW.state=OLD.state THEN RETURN NEW; END IF; + IF OLD.graph_domain='generic-v1' THEN + IF OLD.state='RESERVED' AND NEW.state='LIVE' THEN RETURN NEW; END IF; + RAISE EXCEPTION 'generic lifetime transition requires original collector'; + END IF; + IF OLD.state='RESERVED' AND NEW.state='LIVE' THEN + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_graph_node n JOIN mst2_metadata_payload b USING(page_id,generation) + WHERE n.page_id=OLD.page_id AND n.generation=OLD.generation AND n.state='LIVE' + AND n.metadata_codec=OLD.metadata_codec AND n.bytes=OLD.expected_size + AND b.metadata_codec=OLD.metadata_codec AND b.byte_size=OLD.expected_size) + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_graph_root r WHERE r.page_id=OLD.page_id AND r.generation=OLD.generation) THEN + RAISE EXCEPTION 'qualified LIVE transition needs its graph payload and prepare root'; + END IF; + RETURN NEW; + END IF; + SELECT * INTO op FROM mst2_metadata_gc_op WHERE page_id=OLD.page_id AND generation=OLD.generation AND state='PENDING'; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN RAISE EXCEPTION 'metadata transition has no exact GC op'; END IF; + IF OLD.state IN ('RESERVED','LIVE') AND NEW.state='DELETING' THEN + PERFORM mst2_metadata_assert_uncovered(OLD.page_id,OLD.generation); + RETURN NEW; + END IF; + IF OLD.state='DELETING' AND NEW.state='REMOVED' THEN + IF op.had_payload AND op.payload_delete_xid IS DISTINCT FROM txid_current() THEN + RAISE EXCEPTION 'metadata removal has no same-transaction payload delete proof'; + END IF; + PERFORM mst2_metadata_gc_proof(OLD.page_id,OLD.generation,'PENDING',false,false); + RETURN NEW; + END IF; + RAISE EXCEPTION 'metadata lifetime invalid state transition'; +END $$; + +CREATE FUNCTION mst2_metadata_lifetime_removed() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF OLD.state='DELETING' AND NEW.state='REMOVED' THEN + UPDATE mst2_metadata_gc_op SET state='APPLIED' + WHERE page_id=NEW.page_id AND generation=NEW.generation AND state='PENDING'; + IF NOT FOUND THEN RAISE EXCEPTION 'metadata removal lost its GC operation'; END IF; + END IF; + RETURN NULL; +END $$; +CREATE TRIGGER mst2_metadata_lifetime_removed AFTER UPDATE ON mst2_metadata_lifetime + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_lifetime_removed(); + +CREATE OR REPLACE FUNCTION mst2_metadata_current_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op mst2_metadata_gc_op%ROWTYPE; l mst2_metadata_lifetime%ROWTYPE; +BEGIN + IF TG_OP='DELETE' THEN RAISE EXCEPTION 'metadata current watermark cannot be deleted'; END IF; + IF TG_OP='INSERT' THEN + IF NEW.generation<>1 OR EXISTS(SELECT 1 FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND generation<>1) THEN + RAISE EXCEPTION 'initial metadata current cannot reset history'; + END IF; + SELECT * INTO l FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND generation=NEW.generation; + IF l.graph_domain='qualified-v1' THEN + PERFORM mst2_metadata_assert_uncovered(NEW.page_id,NEW.generation); + IF EXISTS(SELECT 1 FROM mst2_metadata_payload WHERE page_id=NEW.page_id) + OR EXISTS(SELECT 1 FROM mst2_metadata_graph_node WHERE page_id=NEW.page_id) THEN + RAISE EXCEPTION 'initial qualified current cannot adopt old bytes or graph'; + END IF; + END IF; + RETURN NEW; + END IF; + IF NEW.page_id<>OLD.page_id OR OLD.generation=9223372036854775807 OR NEW.generation<>OLD.generation+1 THEN + RAISE EXCEPTION 'metadata current requires exact next-generation CAS'; + END IF; + SELECT * INTO op FROM mst2_metadata_gc_op WHERE page_id=OLD.page_id AND generation=OLD.generation AND state='APPLIED'; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN RAISE EXCEPTION 'fresh metadata requires exact APPLIED proof'; END IF; + PERFORM mst2_metadata_gc_proof(OLD.page_id,OLD.generation,'REMOVED',false,false); + SELECT * INTO l FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND generation=NEW.generation; + IF NOT FOUND OR l.state<>'RESERVED' OR l.graph_domain<>'qualified-v1' + OR l.metadata_codec<>op.metadata_codec OR l.expected_size<>op.expected_size THEN + RAISE EXCEPTION 'fresh metadata incarnation profile mismatch'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_current_insert_guard BEFORE INSERT ON mst2_metadata_current + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_current_guard(); + +CREATE FUNCTION mst2_metadata_current_protected() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF EXISTS(SELECT 1 FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND generation=NEW.generation AND graph_domain='qualified-v1') + AND NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + JOIN mst2_metadata_current c ON c.page_id=m.page_id AND c.generation=m.generation + WHERE m.page_id=NEW.page_id AND m.generation=NEW.generation AND q.graph_domain='qualified-v1' + AND q.storage_seal IS NOT NULL AND (q.state='PREPARING' OR (q.state='COMMITTED' AND q.coverage_retired_at IS NULL))) THEN + RAISE EXCEPTION 'qualified current change must commit with fresh prepare protection'; + END IF; + RETURN NULL; +END $$; +CREATE CONSTRAINT TRIGGER mst2_metadata_current_protected AFTER INSERT OR UPDATE ON mst2_metadata_current + DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION mst2_metadata_current_protected(); + +CREATE FUNCTION mst2_metadata_lifetime_insert_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE c mst2_metadata_current%ROWTYPE; old_owner text; op mst2_metadata_gc_op%ROWTYPE; +BEGIN + SELECT * INTO c FROM mst2_metadata_current WHERE page_id=NEW.page_id; + IF FOUND THEN + SELECT graph_domain INTO old_owner FROM mst2_metadata_lifetime WHERE page_id=c.page_id AND generation=c.generation; + IF old_owner IS DISTINCT FROM NEW.graph_domain THEN RAISE EXCEPTION 'metadata incarnation domain cannot be adopted'; END IF; + IF NEW.graph_domain='qualified-v1' AND NEW.generation<>c.generation THEN + SELECT * INTO op FROM mst2_metadata_gc_op WHERE page_id=c.page_id AND generation=c.generation AND state='APPLIED'; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) OR c.generation=9223372036854775807 + OR NEW.generation<>c.generation+1 OR NEW.state<>'RESERVED' + OR NEW.metadata_codec<>op.metadata_codec OR NEW.expected_size<>op.expected_size THEN + RAISE EXCEPTION 'fresh historical incarnation needs exact removal proof'; + END IF; + PERFORM mst2_metadata_gc_proof(c.page_id,c.generation,'REMOVED',false,false); + END IF; + ELSIF NEW.graph_domain='qualified-v1' AND (NEW.generation<>1 OR NEW.state<>'RESERVED' + OR EXISTS(SELECT 1 FROM mst2_metadata_lifetime WHERE page_id=NEW.page_id AND graph_domain<>NEW.graph_domain)) THEN + RAISE EXCEPTION 'initial qualified incarnation cannot adopt old history'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_lifetime_insert_guard BEFORE INSERT ON mst2_metadata_lifetime + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_lifetime_insert_guard(); + +CREATE FUNCTION mst2_metadata_qualified_mapping_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE owner text; domain text; st text; +BEGIN + IF TG_OP<>'INSERT' THEN RETURN NEW; END IF; + SELECT l.graph_domain INTO owner FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE c.page_id=NEW.page_id; + SELECT graph_domain INTO domain FROM mst2_metadata_prepare WHERE prepare_id=NEW.prepare_id; + IF owner='qualified-v1' OR domain='qualified-v1' THEN + SELECT l.state INTO st FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE c.page_id=NEW.page_id AND c.generation=NEW.generation AND l.graph_domain='qualified-v1'; + IF owner IS DISTINCT FROM domain OR st IS NULL OR st NOT IN ('RESERVED','LIVE') + OR EXISTS(SELECT 1 FROM mst2_metadata_gc_op WHERE page_id=NEW.page_id AND generation=NEW.generation) + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare q WHERE q.prepare_id=NEW.prepare_id + AND q.state='PREPARING' AND q.storage_seal IS NOT NULL AND mst2_metadata_scope_matches(q.primary_scope)) THEN + RAISE EXCEPTION 'qualified mapping cannot cross domain/current/state fence'; + END IF; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_qualified_mapping_guard BEFORE INSERT ON mst2_metadata_prepare_page + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_qualified_mapping_guard(); + +CREATE FUNCTION mst2_metadata_generic_domain_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE record jsonb; ident text; key text; +BEGIN + FOREACH key IN ARRAY ARRAY['node_id','parent_id','child_id'] LOOP + IF TG_OP<>'INSERT' THEN + record:=to_jsonb(OLD); ident:=record->>key; + IF EXISTS(SELECT 1 FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE l.graph_domain='qualified-v1' AND l.node_id=ident) THEN RAISE EXCEPTION 'generic graph cannot touch qualified incarnation'; END IF; + END IF; + IF TG_OP<>'DELETE' THEN + record:=to_jsonb(NEW); ident:=record->>key; + IF EXISTS(SELECT 1 FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE l.graph_domain='qualified-v1' AND l.node_id=ident) THEN RAISE EXCEPTION 'generic graph cannot adopt qualified incarnation'; END IF; + END IF; + END LOOP; + IF TG_OP='DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END $$; +DO $$ DECLARE t text; BEGIN + FOREACH t IN ARRAY ARRAY['mst2_retention_node','mst2_retention_edge','mst2_retention_root','mst2_retention_gc_op'] LOOP + EXECUTE format('CREATE TRIGGER mst2_metadata_generic_domain_guard BEFORE INSERT OR UPDATE OR DELETE ON %I FOR EACH ROW EXECUTE FUNCTION mst2_metadata_generic_domain_guard()',t); + END LOOP; +END $$; + +CREATE FUNCTION mst2_metadata_session_domain_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE sid text; pid text; +BEGIN + IF TG_OP='DELETE' THEN RETURN OLD; END IF; + IF TG_TABLE_NAME='mst2_snapshot_context' THEN pid:=NEW.prepare_id; + ELSE sid:=NEW.snapshot_id; SELECT prepare_id INTO pid FROM mst2_snapshot_context WHERE snapshot_id=sid; END IF; + IF EXISTS(SELECT 1 FROM mst2_metadata_prepare WHERE prepare_id=pid AND graph_domain='qualified-v1') THEN + RAISE EXCEPTION 'qualified production session adoption is closed'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_session_domain_guard BEFORE INSERT OR UPDATE ON mst2_snapshot_context + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_session_domain_guard(); +CREATE TRIGGER mst2_metadata_session_domain_guard BEFORE INSERT OR UPDATE ON mst2_snapshot_lease + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_session_domain_guard(); + +CREATE FUNCTION mst2_metadata_graph_node_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE l mst2_metadata_lifetime%ROWTYPE; op mst2_metadata_gc_op%ROWTYPE; +BEGIN + IF TG_OP='DELETE' THEN + SELECT * INTO op FROM mst2_metadata_gc_op WHERE page_id=OLD.page_id AND generation=OLD.generation AND state='PENDING'; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) OR OLD.state<>'DELETING' THEN + RAISE EXCEPTION 'qualified node deletion needs exact pending operation'; + END IF; + IF op.had_payload AND op.payload_delete_xid IS DISTINCT FROM txid_current() THEN RAISE EXCEPTION 'node deletion has no payload delete proof'; END IF; + PERFORM mst2_metadata_gc_proof(OLD.page_id,OLD.generation,'PENDING',true,false); + IF EXISTS(SELECT 1 FROM mst2_metadata_graph_edge WHERE parent_page=OLD.page_id AND parent_generation=OLD.generation) THEN + RAISE EXCEPTION 'qualified node still has outgoing edges'; + END IF; + RETURN OLD; + END IF; + SELECT l0.* INTO l FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l0 USING(page_id,generation) + WHERE c.page_id=NEW.page_id AND c.generation=NEW.generation; + IF NOT FOUND OR l.graph_domain<>'qualified-v1' OR l.metadata_codec<>NEW.metadata_codec OR l.expected_size<>NEW.bytes THEN + RAISE EXCEPTION 'qualified node does not match exact lifetime'; + END IF; + IF NEW.incoming_refs<>(SELECT count(*) FROM mst2_metadata_graph_edge WHERE child_page=NEW.page_id AND child_generation=NEW.generation) THEN + RAISE EXCEPTION 'qualified node counter differs from actual unique edges'; + END IF; + IF TG_OP='INSERT' THEN + IF NEW.state<>'LIVE' OR l.state NOT IN ('RESERVED','LIVE') OR NOT EXISTS( + SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + WHERE m.page_id=NEW.page_id AND m.generation=NEW.generation AND q.graph_domain='qualified-v1' AND q.state='PREPARING') THEN + RAISE EXCEPTION 'qualified node creation needs active fixed preparation'; + END IF; + ELSE + IF ROW(NEW.page_id,NEW.generation,NEW.metadata_codec,NEW.bytes) IS DISTINCT FROM ROW(OLD.page_id,OLD.generation,OLD.metadata_codec,OLD.bytes) THEN + RAISE EXCEPTION 'qualified node identity is immutable'; + END IF; + IF NEW.state<>OLD.state AND NOT (OLD.state='LIVE' AND NEW.state='DELETING' AND EXISTS( + SELECT 1 FROM mst2_metadata_gc_op WHERE page_id=OLD.page_id AND generation=OLD.generation AND state='PENDING')) THEN + RAISE EXCEPTION 'qualified node invalid state transition'; + END IF; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_graph_node_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_graph_node + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_graph_node_guard(); + +CREATE FUNCTION mst2_metadata_graph_root_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF TG_OP='UPDATE' THEN RAISE EXCEPTION 'qualified roots cannot be retargeted'; END IF; + IF TG_OP='DELETE' THEN RETURN OLD; END IF; + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare q JOIN mst2_metadata_current c ON c.page_id=NEW.page_id AND c.generation=NEW.generation + JOIN mst2_metadata_lifetime l USING(page_id,generation) JOIN mst2_metadata_graph_node n USING(page_id,generation) + WHERE q.prepare_id=NEW.prepare_id AND q.storage_seal=NEW.storage_seal AND q.state='PREPARING' + AND q.graph_domain='qualified-v1' AND l.graph_domain='qualified-v1' AND l.state IN ('RESERVED','LIVE') + AND n.state='LIVE' AND mst2_metadata_scope_matches(q.primary_scope)) THEN + RAISE EXCEPTION 'qualified root requires its active exact sealed preparation'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_graph_root_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_graph_root + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_graph_root_guard(); + +CREATE FUNCTION mst2_metadata_graph_edge_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op mst2_metadata_gc_op%ROWTYPE; +BEGIN + IF TG_OP='UPDATE' THEN RAISE EXCEPTION 'qualified edge identity is immutable'; END IF; + IF TG_OP='DELETE' THEN + SELECT * INTO op FROM mst2_metadata_gc_op WHERE page_id=OLD.parent_page AND generation=OLD.parent_generation AND state='PENDING'; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN RAISE EXCEPTION 'edge deletion has no exact operation'; END IF; + IF op.had_payload AND op.payload_delete_xid IS DISTINCT FROM txid_current() THEN RAISE EXCEPTION 'edge deletion has no payload delete proof'; END IF; + IF EXISTS(SELECT 1 FROM mst2_metadata_payload WHERE page_id=OLD.parent_page) + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_graph_node WHERE page_id=OLD.parent_page + AND generation=OLD.parent_generation AND state='DELETING' AND incoming_refs=0) THEN + RAISE EXCEPTION 'edge deletion is outside its atomic byte-removal phase'; + END IF; + RETURN OLD; + END IF; + IF (SELECT count(*) FROM mst2_metadata_graph_node n JOIN mst2_metadata_current c USING(page_id,generation) + JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE ((n.page_id=NEW.parent_page AND n.generation=NEW.parent_generation) + OR (n.page_id=NEW.child_page AND n.generation=NEW.child_generation)) + AND n.state='LIVE' AND l.state IN ('RESERVED','LIVE') AND l.graph_domain='qualified-v1')<>2 + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare_page a JOIN mst2_metadata_prepare_page b USING(prepare_id) + JOIN mst2_metadata_prepare q USING(prepare_id) WHERE a.page_id=NEW.parent_page AND a.generation=NEW.parent_generation + AND b.page_id=NEW.child_page AND b.generation=NEW.child_generation AND q.state='PREPARING' AND q.graph_domain='qualified-v1') THEN + RAISE EXCEPTION 'edge creation requires active exact qualified endpoints'; + END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_graph_edge_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_graph_edge + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_graph_edge_guard(); + +CREATE FUNCTION mst2_metadata_edges_added() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE reachable text[]; links jsonb; adjacency jsonb; indegrees bigint[]; + queue integer[]:=ARRAY[]::integer[]; head integer:=1; parent_index integer; child_index integer; i integer; +BEGIN + IF (SELECT count(*) FROM added_edges)>16384 THEN RAISE EXCEPTION 'qualified edge batch exceeds limit'; END IF; + IF NOT EXISTS(SELECT 1 FROM added_edges) THEN RETURN NULL; END IF; + IF EXISTS(SELECT 1 FROM mst2_metadata_graph_node n JOIN ( + SELECT child_page,child_generation,count(*) AS delta FROM added_edges GROUP BY child_page,child_generation + ) d ON d.child_page=n.page_id AND d.child_generation=n.generation + WHERE n.incoming_refs+d.delta<>(SELECT count(*) FROM mst2_metadata_graph_edge x WHERE x.child_page=n.page_id AND x.child_generation=n.generation)) THEN + RAISE EXCEPTION 'qualified insertion found preexisting counter drift'; + END IF; + UPDATE mst2_metadata_graph_node n SET incoming_refs=n.incoming_refs+d.delta FROM ( + SELECT child_page,child_generation,count(*) AS delta FROM added_edges GROUP BY child_page,child_generation + ) d WHERE n.page_id=d.child_page AND n.generation=d.child_generation; + -- One bounded closure per statement; topological accounting below is local. + WITH RECURSIVE walk(page_id,generation) AS ( + SELECT page_id,generation FROM ( + SELECT parent_page AS page_id,parent_generation AS generation FROM added_edges + UNION SELECT child_page,child_generation FROM added_edges + ) starts UNION + SELECT x.child_page,x.child_generation FROM walk w JOIN mst2_metadata_graph_edge x + ON x.parent_page=w.page_id AND x.parent_generation=w.generation + ) SELECT array_agg(encode(page_id,'hex')||':'||generation ORDER BY page_id,generation) + INTO reachable FROM (SELECT * FROM walk LIMIT 4097) bounded; + IF cardinality(reachable)>4096 THEN RAISE EXCEPTION 'qualified graph node audit overflow'; END IF; + WITH nodes AS ( + SELECT ordinality::integer AS i,decode(split_part(key,':',1),'hex') AS page_id, + split_part(key,':',2)::bigint AS generation FROM unnest(reachable) WITH ORDINALITY n(key,ordinality) + ) SELECT coalesce(jsonb_agg(jsonb_build_array(p.i,c.i)),'[]'::jsonb) INTO links FROM ( + SELECT e.* FROM nodes p JOIN mst2_metadata_graph_edge e + ON e.parent_page=p.page_id AND e.parent_generation=p.generation LIMIT 16385 + ) e JOIN nodes p ON p.page_id=e.parent_page AND p.generation=e.parent_generation + JOIN nodes c ON c.page_id=e.child_page AND c.generation=e.child_generation; + IF jsonb_array_length(links)>16384 THEN RAISE EXCEPTION 'qualified graph edge audit overflow'; END IF; + SELECT coalesce(jsonb_object_agg(parent,children),'{}'::jsonb) INTO adjacency FROM ( + SELECT item->>0 AS parent,jsonb_agg((item->>1)::integer) AS children + FROM jsonb_array_elements(links) x(item) GROUP BY item->>0 + ) grouped; + SELECT array_agg(coalesce(counts.refs,0) ORDER BY n.i) INTO indegrees + FROM generate_series(1,cardinality(reachable)) n(i) LEFT JOIN ( + SELECT (item->>1)::integer AS child,count(*) AS refs FROM jsonb_array_elements(links) x(item) GROUP BY item->>1 + ) counts ON counts.child=n.i; + FOR i IN 1..cardinality(reachable) LOOP + IF indegrees[i]=0 THEN queue:=array_append(queue,i); END IF; + END LOOP; + WHILE head<=cardinality(queue) LOOP + parent_index:=queue[head]; head:=head+1; + FOR child_index IN SELECT value::text::integer FROM jsonb_array_elements(coalesce(adjacency->parent_index::text,'[]'::jsonb)) LOOP + indegrees[child_index]:=indegrees[child_index]-1; + IF indegrees[child_index]=0 THEN queue:=array_append(queue,child_index); END IF; + END LOOP; + END LOOP; + IF cardinality(queue)<>cardinality(reachable) THEN RAISE EXCEPTION 'qualified graph contains a cycle'; END IF; + RETURN NULL; +END $$; +CREATE TRIGGER mst2_metadata_edges_added AFTER INSERT ON mst2_metadata_graph_edge + REFERENCING NEW TABLE AS added_edges FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_edges_added(); +CREATE FUNCTION mst2_metadata_edges_removed() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF EXISTS(SELECT 1 FROM mst2_metadata_graph_node n JOIN ( + SELECT child_page,child_generation,count(*) AS delta FROM removed_edges GROUP BY child_page,child_generation + ) d ON d.child_page=n.page_id AND d.child_generation=n.generation + WHERE n.incoming_refs(SELECT count(*) FROM mst2_metadata_graph_edge x WHERE x.child_page=n.page_id AND x.child_generation=n.generation)) THEN + RAISE EXCEPTION 'qualified subtraction found counter drift'; + END IF; + UPDATE mst2_metadata_graph_node n SET incoming_refs=n.incoming_refs-d.delta FROM ( + SELECT child_page,child_generation,count(*) AS delta FROM removed_edges GROUP BY child_page,child_generation + ) d WHERE n.page_id=d.child_page AND n.generation=d.child_generation; + RETURN NULL; +END $$; +CREATE TRIGGER mst2_metadata_edges_removed AFTER DELETE ON mst2_metadata_graph_edge + REFERENCING OLD TABLE AS removed_edges FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_edges_removed(); + +CREATE FUNCTION mst2_metadata_gc_claimed() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + UPDATE mst2_metadata_graph_node SET state='DELETING' WHERE page_id=NEW.page_id AND generation=NEW.generation; + UPDATE mst2_metadata_lifetime SET state='DELETING' WHERE page_id=NEW.page_id AND generation=NEW.generation; + RETURN NULL; +END $$; +CREATE TRIGGER mst2_metadata_gc_claimed AFTER INSERT ON mst2_metadata_gc_op + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_gc_claimed(); + +CREATE FUNCTION mst2_metadata_gc_finish(id uuid) RETURNS void LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op mst2_metadata_gc_op%ROWTYPE; +BEGIN + IF pg_is_in_recovery() OR current_setting('transaction_isolation')<>'read committed' THEN RAISE EXCEPTION 'GC finish needs primary READ COMMITTED'; END IF; + PERFORM set_config('lock_timeout','5000ms',true); + PERFORM pg_advisory_xact_lock(1296717362,hashtext(current_schema())); + SELECT * INTO op FROM mst2_metadata_gc_op WHERE operation_id=id FOR UPDATE; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN RAISE EXCEPTION 'GC finish wrong exact operation scope'; END IF; + IF op.state='APPLIED' THEN RETURN; END IF; + IF op.had_payload AND op.payload_delete_xid IS DISTINCT FROM txid_current() THEN + RAISE EXCEPTION 'GC finish has no same-transaction payload delete proof'; + END IF; + PERFORM mst2_metadata_gc_proof(op.page_id,op.generation,'PENDING',op.graph_present,false); + DELETE FROM mst2_metadata_graph_edge WHERE parent_page=op.page_id AND parent_generation=op.generation; + DELETE FROM mst2_metadata_graph_node WHERE page_id=op.page_id AND generation=op.generation; + UPDATE mst2_metadata_lifetime SET state='REMOVED' WHERE page_id=op.page_id AND generation=op.generation AND state='DELETING'; + IF NOT FOUND THEN RAISE EXCEPTION 'GC finish lost exact lifetime'; END IF; +END $$; + +CREATE FUNCTION mst2_metadata_payload_fenced() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE l mst2_metadata_lifetime%ROWTYPE; op mst2_metadata_gc_op%ROWTYPE; selector text; +BEGIN + IF TG_OP='UPDATE' THEN RAISE EXCEPTION 'MST2 metadata payload UPDATE is forbidden'; END IF; + IF TG_OP='INSERT' THEN + SELECT h.* INTO l FROM mst2_metadata_current c JOIN mst2_metadata_lifetime h USING(page_id,generation) WHERE c.page_id=NEW.page_id; + IF FOUND AND l.graph_domain='qualified-v1' THEN + IF NEW.generation IS DISTINCT FROM l.generation OR l.state NOT IN ('RESERVED','LIVE') + OR NEW.metadata_codec<>l.metadata_codec OR NEW.byte_size<>l.expected_size + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m JOIN mst2_metadata_prepare q USING(prepare_id) + WHERE m.page_id=NEW.page_id AND m.generation=NEW.generation AND q.state='PREPARING' + AND q.graph_domain='qualified-v1' AND q.storage_seal IS NOT NULL AND mst2_metadata_scope_matches(q.primary_scope)) THEN + RAISE EXCEPTION 'qualified payload INSERT crossed its exact active generation'; + END IF; + END IF; + RETURN NEW; + END IF; + selector:=current_setting('mega2.metadata_gc_operation',true); + IF OLD.generation IS NULL OR selector IS NULL OR selector='' THEN RAISE EXCEPTION 'payload DELETE has no qualified operation selector'; END IF; + SELECT * INTO op FROM mst2_metadata_gc_op WHERE operation_id=selector::uuid AND page_id=OLD.page_id AND generation=OLD.generation FOR UPDATE; + IF NOT FOUND OR op.state<>'PENDING' OR NOT op.had_payload OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN + RAISE EXCEPTION 'payload DELETE requires exact persistent pending proof'; + END IF; + PERFORM mst2_metadata_gc_proof(OLD.page_id,OLD.generation,'PENDING',op.graph_present,true); + UPDATE mst2_metadata_gc_op SET payload_delete_xid=txid_current() WHERE operation_id=op.operation_id; + RETURN OLD; +END $$; +DROP TRIGGER mst2_metadata_payload_immutable ON mst2_metadata_payload; +CREATE TRIGGER mst2_metadata_payload_fenced BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_payload + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_payload_fenced(); +CREATE FUNCTION mst2_metadata_payload_removed() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op uuid; +BEGIN + SELECT operation_id INTO op FROM mst2_metadata_gc_op WHERE page_id=OLD.page_id AND generation=OLD.generation AND state='PENDING'; + IF NOT FOUND THEN RAISE EXCEPTION 'deleted payload lost exact pending operation'; END IF; + PERFORM mst2_metadata_gc_finish(op); + RETURN NULL; +END $$; +CREATE TRIGGER mst2_metadata_payload_removed AFTER DELETE ON mst2_metadata_payload + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_payload_removed(); + +CREATE FUNCTION mst2_metadata_gc_apply(id uuid) RETURNS void LANGUAGE plpgsql VOLATILE AS $$ +DECLARE op mst2_metadata_gc_op%ROWTYPE; +BEGIN + IF pg_is_in_recovery() OR current_setting('transaction_isolation')<>'read committed' THEN RAISE EXCEPTION 'GC apply needs primary READ COMMITTED'; END IF; + PERFORM set_config('lock_timeout','5000ms',true); + PERFORM pg_advisory_xact_lock(1296717362,hashtext(current_schema())); + SELECT * INTO op FROM mst2_metadata_gc_op WHERE operation_id=id FOR UPDATE; + IF NOT FOUND OR NOT mst2_metadata_scope_matches(op.primary_scope) THEN RAISE EXCEPTION 'GC apply missing exact scope'; END IF; + IF op.state='APPLIED' THEN RETURN; END IF; + PERFORM mst2_metadata_gc_proof(op.page_id,op.generation,'PENDING',op.graph_present,op.had_payload); + IF op.had_payload THEN + PERFORM set_config('mega2.metadata_gc_operation',op.operation_id::text,true); + DELETE FROM mst2_metadata_payload WHERE page_id=op.page_id AND generation=op.generation; + IF NOT FOUND THEN RAISE EXCEPTION 'GC apply lost captured payload'; END IF; + ELSE + PERFORM mst2_metadata_gc_finish(id); + END IF; +END $$; diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 4a302ee3..384af69a 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -148,6 +148,7 @@ mod m20261006_000100_add_view_tables; mod m20261007_000100_add_mst2_snapshot_sessions; mod m20261007_000200_add_mst2_metadata_generations; mod m20261007_000300_add_mst2_metadata_lifetime_history; +mod m20261007_000400_add_mst2_qualified_metadata_gc; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -284,6 +285,7 @@ impl MigratorTrait for Migrator { Box::new(m20261007_000100_add_mst2_snapshot_sessions::Migration), Box::new(m20261007_000200_add_mst2_metadata_generations::Migration), Box::new(m20261007_000300_add_mst2_metadata_lifetime_history::Migration), + Box::new(m20261007_000400_add_mst2_qualified_metadata_gc::Migration), ] } } @@ -1194,7 +1196,7 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 11..names.len() - 2], + &names[names.len() - 12..names.len() - 3], &[ "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), @@ -1209,13 +1211,17 @@ mod tests { "native retention, metadata installation and view tables follow media paging" ); assert_eq!( - &names[names.len() - 2], + &names[names.len() - 3], "m20261007_000200_add_mst2_metadata_generations" ); assert_eq!( - names.last().unwrap(), + &names[names.len() - 2], "m20261007_000300_add_mst2_metadata_lifetime_history" ); + assert_eq!( + names.last().unwrap(), + "m20261007_000400_add_mst2_qualified_metadata_gc" + ); let db = alias_db().await; insert_repo(&db, 1, "/third-party//a").await; diff --git a/src/jupiter/storage/native_metadata_generations.rs b/src/jupiter/storage/native_metadata_generations.rs index 9ea9f766..a2cad697 100644 --- a/src/jupiter/storage/native_metadata_generations.rs +++ b/src/jupiter/storage/native_metadata_generations.rs @@ -168,6 +168,15 @@ impl PostgresMetadataGenerationRepository { &self, operation_id: &str, prepared: &PreparedNativeMetadataRetention, + ) -> Result { + self.begin_with_reopens(operation_id, prepared, &[]).await + } + + async fn begin_with_reopens( + &self, + operation_id: &str, + prepared: &PreparedNativeMetadataRetention, + reopens: &[qualified::MetadataGcReceipt], ) -> Result { validate_operation_id(operation_id)?; let plan = prepared.install_plan()?; @@ -177,8 +186,12 @@ impl PostgresMetadataGenerationRepository { let result = async { self.inner.barrier(&txn).await?; if let Some(fixed) = self.load_fixed_plan(&txn, operation_id, &digest).await? { + qualified::verify_reopen_replay(&txn, &fixed, reopens).await?; return Ok(fixed.intent); } + if !reopens.is_empty() { + qualified::reopen_in_txn(&txn, &plan, reopens).await?; + } let bindings = allocate_lifetimes(&txn, &plan, self.graph_domain).await?; let canonical_bindings = bindings.encode()?; let bindings_digest = Sha256::digest(&canonical_bindings).into(); @@ -218,6 +231,9 @@ impl PostgresMetadataGenerationRepository { )).await.map_err(internal)?; // Reuse only already LIVE graph rows. RESERVED pages without a graph // are protected by the durable mappings, without inventing a DAG. + if self.graph_domain == "qualified-v1" { + qualified::retain_existing_roots(&txn, &intent).await?; + } else { txn.execute_raw(statement( "INSERT INTO mst2_retention_root(node_id,root_key,root_kind,created_at) SELECT l.node_id,'prepare:'||p.prepare_id,'prepare',now() @@ -227,6 +243,7 @@ impl PostgresMetadataGenerationRepository { WHERE p.prepare_id=$1 ON CONFLICT(node_id,root_key) DO NOTHING", [prepare_id.into()], )).await.map_err(internal)?; + } Ok(intent) }.await; commit( @@ -288,6 +305,7 @@ impl PostgresMetadataGenerationRepository { "INSERT INTO mst2_metadata_payload(page_id,generation,metadata_codec,byte_size,payload) SELECT decode(p.page_id,'hex'),p.generation,$1,p.size,decode(p.payload,'hex') FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,generation bigint,size integer,payload text) + WHERE NOT EXISTS(SELECT 1 FROM mst2_metadata_payload b WHERE b.page_id=decode(p.page_id,'hex')) ON CONFLICT(page_id) DO NOTHING", [fixed.stored.record.metadata_codec.into(),encoded.clone().into()], )).await.map_err(internal)?; @@ -379,10 +397,13 @@ impl PostgresMetadataGenerationRepository { lock_prepare(txn, intent.prepare_id()).await?; let fixed = self.require_fixed_plan(txn, intent).await?; check_generation_payloads(txn, &fixed).await?; - let legacy = self - .inner - .finalize_stored_plan_in_txn(txn, &intent.legacy, &observation.dag, fixed.stored) - .await?; + let legacy = if intent.graph_domain == GraphDomain::Qualified { + qualified::finalize_graph(txn, &fixed, &observation.dag).await? + } else { + self.inner + .finalize_stored_plan_in_txn(txn, &intent.legacy, &observation.dag, fixed.stored) + .await? + }; txn.execute_raw(statement( "UPDATE mst2_metadata_lifetime l SET state='LIVE' FROM mst2_metadata_prepare_page p WHERE p.prepare_id=$1 AND p.page_id=l.page_id @@ -407,6 +428,9 @@ impl PostgresMetadataGenerationRepository { scope: &str, ) -> Result<(), SnapshotError> { self.require_scope(&receipt.intent)?; + if receipt.intent.graph_domain == GraphDomain::Qualified { + return Err(unavailable("qualified production handoff is not admitted")); + } self.inner .verify_receipt_in_txn(txn, &receipt.legacy, tagged_root_tree_oid, scope) .await?; @@ -515,7 +539,7 @@ impl PostgresMetadataGenerationRepository { Some(fixed) if fixed.stored.record.state == "COMMITTED" => { check_generation_payloads(&txn, &fixed).await?; load_fixed_dag(&txn, &fixed).await?; - verify_graph(&txn, &fixed.stored).await?; + verify_fixed_graph(&txn, &fixed).await?; Ok(GenerationPrepareObservation::Committed(Box::new( GenerationMetadataReceipt { legacy: fixed.stored.receipt()?, @@ -704,6 +728,9 @@ async fn allocate_lifetimes( plan: &MetadataInstallPlan, graph_domain: &str, ) -> Result { + if graph_domain == "qualified-v1" { + return qualified::allocate_lifetimes(txn, plan).await; + } let pages: Vec<_> = plan .pages .iter() @@ -828,6 +855,9 @@ async fn check_lifetimes( connection: &C, stored: &StoredPlan, ) -> Result<(), SnapshotError> { + if stored.record.graph_domain.as_deref() == Some("qualified-v1") { + return qualified::check_lifetimes(connection, stored).await; + } let row=connection.query_one_raw(statement( "SELECT p.page_id,l.generation,l.state,l.metadata_codec,l.expected_size,p.generation AS expected_generation, n.state AS graph_state,n.kind AS graph_kind,n.bytes AS graph_bytes, @@ -953,3 +983,17 @@ mod tests; #[path = "native_metadata_history.rs"] pub mod history; + +#[path = "native_metadata_qualified.rs"] +pub mod qualified; + +async fn verify_fixed_graph( + connection: &C, + fixed: &FixedPlan, +) -> Result<(), SnapshotError> { + if fixed.intent.graph_domain == GraphDomain::Qualified { + qualified::verify_graph(connection, fixed).await + } else { + verify_graph(connection, &fixed.stored).await + } +} diff --git a/src/jupiter/storage/native_metadata_history.rs b/src/jupiter/storage/native_metadata_history.rs index e093a4cc..6ea7f821 100644 --- a/src/jupiter/storage/native_metadata_history.rs +++ b/src/jupiter/storage/native_metadata_history.rs @@ -39,11 +39,77 @@ impl MetadataTerminalReceipt { #[derive(Debug, Clone, PartialEq, Eq)] pub enum MetadataTerminalObservation { + /// The stored state matches this action; mutation still requires its full CAS proof. Active, Terminated(Box), } impl PostgresMetadataGenerationRepository { + /// Recover an immutable sealed intent after restart, including retired/aborted history. + /// Current lifetimes and payloads are deliberately not consulted or revived. + pub async fn capture_terminal_intent( + &self, + operation_id: &str, + manifest_digest: [u8; 32], + ) -> Result { + validate_operation_id(operation_id)?; + let txn = self.inner.transaction().await?; + self.inner.barrier(&txn).await?; + let result = async { + let record = mst2_metadata_prepare::Entity::find() + .filter(mst2_metadata_prepare::Column::OperationId.eq(operation_id)) + .one(&txn) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("terminal preparation history is missing"))?; + if record.manifest_digest.as_slice() != manifest_digest { + return Err(integrity( + "terminal recovery manifest differs from durable history", + )); + } + let plan = MetadataInstallPlan::decode(&record.canonical_plan, &manifest_digest)?; + let canonical = record.canonical_bindings.as_deref().ok_or_else(|| { + unavailable("legacy unsealed preparation has no fixed terminal capability") + })?; + let bindings = GenerationBindings::decode(canonical, &plan)?; + let digest: [u8; 32] = Sha256::digest(canonical).into(); + let primary = record + .primary_scope + .as_deref() + .ok_or_else(|| integrity("terminal history has no primary scope"))?; + let domain = GraphDomain::from_stored(record.graph_domain.as_deref())?; + let intent = GenerationPrepareIntent { + legacy: MetadataPrepareIntent { + prepare_id: record.prepare_id.clone(), + operation_id: operation_id.into(), + manifest_digest, + }, + metadata_root: plan.root, + root_generation: bindings + .0 + .get(&plan.root) + .ok_or_else(|| integrity("terminal root binding is missing"))? + .0, + bindings_digest: digest, + primary_scope: primary.into(), + graph_domain: domain, + storage_seal: seal( + &record.prepare_id, + &manifest_digest, + &plan.root, + &digest, + primary, + domain.stored(), + )?, + }; + self.require_terminal_record(&txn, &intent).await?; + Ok(intent) + } + .await; + txn.rollback().await.map_err(internal)?; + result + } + pub async fn abort( &self, intent: &GenerationPrepareIntent, @@ -85,7 +151,7 @@ impl PostgresMetadataGenerationRepository { } } - async fn terminate_in_txn( + pub(super) async fn terminate_in_txn( &self, txn: &DatabaseTransaction, intent: &GenerationPrepareIntent, @@ -119,44 +185,63 @@ impl PostgresMetadataGenerationRepository { )); } check_generation_payloads(txn, &fixed).await?; - verify_graph(txn, &fixed.stored).await?; + verify_fixed_graph(txn, &fixed).await?; } - let roots = mst2_retention_root::Entity::find() - .filter( - mst2_retention_root::Column::RootKey.eq(format!("prepare:{}", intent.prepare_id())), + if intent.graph_domain == GraphDomain::Qualified { + qualified::verify_roots( + txn, + &fixed, + action == MetadataTerminalAction::RetireCoverage, ) - .limit((MetadataDagLimits::default().nodes + 1) as u64) - .all(txn) + .await?; + txn.execute_raw(statement( + "DELETE FROM mst2_metadata_graph_root WHERE prepare_id=$1 AND storage_seal=$2", + [ + intent.prepare_id().into(), + intent.storage_seal.to_vec().into(), + ], + )) .await .map_err(internal)?; - let expected_nodes: BTreeSet<_> = fixed.bindings.0.keys().map(node_id).collect(); - if roots - .iter() - .any(|root| root.root_kind != "prepare" || !expected_nodes.contains(&root.node_id)) - { - return Err(integrity( - "terminal prepare coverage differs from its fixed lifetime mappings", - )); - } - if action == MetadataTerminalAction::RetireCoverage - && roots + } else { + let roots = mst2_retention_root::Entity::find() + .filter( + mst2_retention_root::Column::RootKey + .eq(format!("prepare:{}", intent.prepare_id())), + ) + .limit((MetadataDagLimits::default().nodes + 1) as u64) + .all(txn) + .await + .map_err(internal)?; + let expected_nodes: BTreeSet<_> = fixed.bindings.0.keys().map(node_id).collect(); + if roots .iter() - .map(|root| root.node_id.clone()) - .collect::>() - != expected_nodes - { - return Err(unavailable( - "coverage retirement cannot recover missing or transferred prepare roots", - )); - } - txn.execute_raw(statement( - "DELETE FROM mst2_retention_root r USING mst2_metadata_prepare_page p + .any(|root| root.root_kind != "prepare" || !expected_nodes.contains(&root.node_id)) + { + return Err(integrity( + "terminal prepare coverage differs from its fixed lifetime mappings", + )); + } + if action == MetadataTerminalAction::RetireCoverage + && roots + .iter() + .map(|root| root.node_id.clone()) + .collect::>() + != expected_nodes + { + return Err(unavailable( + "coverage retirement cannot recover missing or transferred prepare roots", + )); + } + txn.execute_raw(statement( + "DELETE FROM mst2_retention_root r USING mst2_metadata_prepare_page p WHERE p.prepare_id=$1 AND r.root_key='prepare:'||p.prepare_id AND r.root_kind='prepare' AND r.node_id='page:sha256:'||encode(p.page_id,'hex')", - [intent.prepare_id().into()], - )) - .await - .map_err(internal)?; + [intent.prepare_id().into()], + )) + .await + .map_err(internal)?; + } let sql=match action { MetadataTerminalAction::Abort=> "UPDATE mst2_metadata_prepare SET state='ABORTED',aborted_at=clock_timestamp() @@ -212,7 +297,19 @@ impl PostgresMetadataGenerationRepository { verify_no_prepare_coverage(&txn, intent).await?; Ok(MetadataTerminalObservation::Terminated(Box::new(receipt))) } - None => Ok(MetadataTerminalObservation::Active), + None => { + let expected = match action { + MetadataTerminalAction::Abort => "PREPARING", + MetadataTerminalAction::RetireCoverage => "COMMITTED", + }; + if record.state != expected { + return Err(SnapshotError::new( + SnapshotErrorCode::Conflict, + "metadata prepare is in another terminal state", + )); + } + Ok(MetadataTerminalObservation::Active) + } } } .await; @@ -325,6 +422,22 @@ async fn verify_no_prepare_coverage( connection: &C, intent: &GenerationPrepareIntent, ) -> Result<(), SnapshotError> { + if intent.graph_domain == GraphDomain::Qualified { + if connection + .query_one_raw(statement( + "SELECT prepare_id FROM mst2_metadata_graph_root WHERE prepare_id=$1 LIMIT 1", + [intent.prepare_id().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(integrity( + "terminal qualified preparation still has coverage", + )); + } + return Ok(()); + } if connection .query_one_raw(statement( "SELECT node_id FROM mst2_retention_root WHERE root_key='prepare:'||$1 LIMIT 1", diff --git a/src/jupiter/storage/native_metadata_qualified.rs b/src/jupiter/storage/native_metadata_qualified.rs new file mode 100644 index 00000000..5a29c9c6 --- /dev/null +++ b/src/jupiter/storage/native_metadata_qualified.rs @@ -0,0 +1,936 @@ +//! Qualified metadata incarnations. Runtime serving adoption is intentionally closed. + +use std::ops::Deref; + +use super::*; + +pub struct PostgresQualifiedMetadataRepository { + inner: PostgresMetadataGenerationRepository, +} + +impl Deref for PostgresQualifiedMetadataRepository { + type Target = PostgresMetadataGenerationRepository; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MetadataLifetime { + page_id: [u8; 32], + generation: i64, + metadata_codec: i16, + expected_size: i32, + primary_scope: Box<[u8]>, +} + +impl MetadataLifetime { + pub fn page_id(&self) -> [u8; 32] { + self.page_id + } + + pub fn generation(&self) -> i64 { + self.generation + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetadataGcPhase { + Claim, + Apply, +} + +#[derive(Debug, thiserror::Error)] +pub enum MetadataGcError { + #[error(transparent)] + Rejected(#[from] SnapshotError), + #[error("metadata GC {phase:?} commit outcome is unknown for {operation_id}")] + CommitUncertain { + operation_id: String, + page_id: [u8; 32], + generation: i64, + phase: MetadataGcPhase, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MetadataGcClaim { + operation_id: String, + lifetime: MetadataLifetime, + graph_present: bool, + had_payload: bool, +} + +impl MetadataGcClaim { + pub fn operation_id(&self) -> &str { + &self.operation_id + } + + pub fn lifetime(&self) -> &MetadataLifetime { + &self.lifetime + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MetadataGcReceipt { + claim: MetadataGcClaim, + created_at: sea_orm::prelude::DateTimeWithTimeZone, + completed_at: sea_orm::prelude::DateTimeWithTimeZone, +} + +impl MetadataGcReceipt { + pub fn claim(&self) -> &MetadataGcClaim { + &self.claim + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MetadataGcObservation { + Absent, + Pending(Box), + Applied(Box), +} + +struct GcRecord { + claim: MetadataGcClaim, + state: String, + created_at: sea_orm::prelude::DateTimeWithTimeZone, + completed_at: Option, +} + +impl GcRecord { + fn receipt(self) -> Result { + if self.state != "APPLIED" { + return Err(unavailable("metadata GC operation is not APPLIED")); + } + Ok(MetadataGcReceipt { + claim: self.claim, + created_at: self.created_at, + completed_at: self + .completed_at + .ok_or_else(|| integrity("APPLIED metadata GC receipt has no timestamp"))?, + }) + } +} + +impl PostgresQualifiedMetadataRepository { + pub async fn new(connection: DatabaseConnection) -> Result { + Ok(Self { + inner: PostgresMetadataGenerationRepository { + inner: PostgresMetadataInstallRepository::new(connection).await?, + graph_domain: "qualified-v1", + }, + }) + } + + pub async fn lifetime(&self, page_id: [u8; 32]) -> Result { + self.inner + .inner + .verify_primary_connection(&self.inner.inner.connection) + .await?; + let row = self + .inner + .inner + .connection + .query_one_raw(statement( + "SELECT l.generation,l.metadata_codec,l.expected_size FROM mst2_metadata_current c + JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE c.page_id=$1 AND l.graph_domain='qualified-v1'", + [page_id.to_vec().into()], + )) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("qualified lifetime is missing"))?; + Ok(MetadataLifetime { + page_id, + generation: row.try_get("", "generation").map_err(internal)?, + metadata_codec: row.try_get("", "metadata_codec").map_err(internal)?, + expected_size: row.try_get("", "expected_size").map_err(internal)?, + primary_scope: self.inner.primary_scope()?.into_boxed_slice(), + }) + } + + /// Read-only recovery selector; claim still requires this tuple to be current. + pub async fn historical_lifetime( + &self, + page_id: [u8; 32], + generation: i64, + ) -> Result { + if generation <= 0 { + return Err(integrity("historical generation must be positive")); + } + self.inner + .inner + .verify_primary_connection(&self.inner.inner.connection) + .await?; + let row = self + .inner + .inner + .connection + .query_one_raw(statement( + "SELECT metadata_codec,expected_size FROM mst2_metadata_lifetime + WHERE page_id=$1 AND generation=$2 AND graph_domain='qualified-v1'", + [page_id.to_vec().into(), generation.into()], + )) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("qualified historical incarnation is missing"))?; + Ok(MetadataLifetime { + page_id, + generation, + metadata_codec: row.try_get("", "metadata_codec").map_err(internal)?, + expected_size: row.try_get("", "expected_size").map_err(internal)?, + primary_scope: self.inner.primary_scope()?.into_boxed_slice(), + }) + } + + fn require_lifetime_scope(&self, lifetime: &MetadataLifetime) -> Result<(), SnapshotError> { + if lifetime.primary_scope.as_ref() != self.inner.primary_scope()?.as_slice() { + return Err(integrity( + "metadata GC belongs to another captured primary scope", + )); + } + Ok(()) + } + + /// Bounded indexed discovery only. Every returned tuple still needs a full claim proof. + pub async fn scan_current_lifetimes( + &self, + after: Option<([u8; 32], i64)>, + limit: u16, + ) -> Result, SnapshotError> { + if !(1..=256).contains(&limit) { + return Err(integrity("metadata scan limit must be 1..=256")); + } + self.inner + .inner + .verify_primary_connection(&self.inner.inner.connection) + .await?; + let (page, generation) = after.map(|(p, g)| (p.to_vec(), g)).unwrap_or_default(); + let rows=self.inner.inner.connection.query_all_raw(statement( + "SELECT l.page_id,l.generation,l.metadata_codec,l.expected_size FROM mst2_metadata_lifetime l + JOIN mst2_metadata_current c USING(page_id,generation) + WHERE l.graph_domain='qualified-v1' AND l.state IN ('RESERVED','LIVE') AND (l.page_id,l.generation)>($1,$2) + ORDER BY l.page_id,l.generation LIMIT $3", + [page.into(),generation.into(),i64::from(limit).into()], + )).await.map_err(internal)?; + let scope = self.inner.primary_scope()?.into_boxed_slice(); + rows.into_iter() + .map(|row| { + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + Ok(MetadataLifetime { + page_id: page.as_slice().try_into().map_err(internal)?, + generation: row.try_get("", "generation").map_err(internal)?, + metadata_codec: row.try_get("", "metadata_codec").map_err(internal)?, + expected_size: row.try_get("", "expected_size").map_err(internal)?, + primary_scope: scope.clone(), + }) + }) + .collect() + } + + pub async fn claim( + &self, + operation_id: &str, + lifetime: &MetadataLifetime, + ) -> Result { + validate_gc_operation(operation_id)?; + self.require_lifetime_scope(lifetime)?; + let txn = self.inner.inner.transaction().await?; + let result = self.claim_in_txn(&txn, operation_id, lifetime).await; + finish_gc_transaction(txn, result, operation_id, lifetime, MetadataGcPhase::Claim).await + } + + async fn claim_in_txn( + &self, + txn: &DatabaseTransaction, + operation_id: &str, + lifetime: &MetadataLifetime, + ) -> Result { + self.require_lifetime_scope(lifetime)?; + self.inner.inner.barrier(txn).await?; + if let Some(record) = load_gc(txn, operation_id).await? { + if &record.claim.lifetime != lifetime { + return Err(integrity( + "GC operation cannot be retargeted to another lifetime", + )); + } + return Ok(record.claim); + } + let row=txn.query_one_raw(statement( + "SELECT EXISTS(SELECT 1 FROM mst2_metadata_graph_node n WHERE n.page_id=l.page_id AND n.generation=l.generation) AS graph_present, + EXISTS(SELECT 1 FROM mst2_metadata_payload b WHERE b.page_id=l.page_id) AS had_payload + FROM mst2_metadata_current c JOIN mst2_metadata_lifetime l USING(page_id,generation) + WHERE l.page_id=$1 AND l.generation=$2 AND l.graph_domain='qualified-v1' + AND l.metadata_codec=$3 AND l.expected_size=$4 FOR UPDATE OF c,l", + [lifetime.page_id.to_vec().into(),lifetime.generation.into(),lifetime.metadata_codec.into(),lifetime.expected_size.into()], + )).await.map_err(internal)?.ok_or_else(|| unavailable("GC candidate no longer names its captured current incarnation"))?; + let graph_present: bool = row.try_get("", "graph_present").map_err(internal)?; + let had_payload: bool = row.try_get("", "had_payload").map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_gc_op(operation_id,page_id,generation,primary_scope,graph_domain, + metadata_codec,expected_size,graph_present,had_payload,state) + VALUES($1::uuid,$2,$3,$4,'qualified-v1',$5,$6,$7,$8,'PENDING')", + [operation_id.into(),lifetime.page_id.to_vec().into(),lifetime.generation.into(), + lifetime.primary_scope.to_vec().into(),lifetime.metadata_codec.into(),lifetime.expected_size.into(), + graph_present.into(),had_payload.into()], + )).await.map_err(internal)?; + Ok(load_gc(txn, operation_id) + .await? + .ok_or_else(|| integrity("claimed GC operation disappeared"))? + .claim) + } + + pub async fn apply( + &self, + claim: &MetadataGcClaim, + ) -> Result { + self.require_lifetime_scope(&claim.lifetime)?; + let txn = self.inner.inner.transaction().await?; + let result = self.apply_in_txn(&txn, claim).await; + finish_gc_transaction( + txn, + result, + &claim.operation_id, + &claim.lifetime, + MetadataGcPhase::Apply, + ) + .await + } + + async fn apply_in_txn( + &self, + txn: &DatabaseTransaction, + claim: &MetadataGcClaim, + ) -> Result { + self.require_lifetime_scope(&claim.lifetime)?; + self.inner.inner.barrier(txn).await?; + let record = load_gc(txn, &claim.operation_id) + .await? + .ok_or_else(|| unavailable("metadata GC operation is missing"))?; + if &record.claim != claim { + return Err(integrity( + "metadata GC claim differs from immutable operation", + )); + } + if record.state == "APPLIED" { + return record.receipt(); + } + txn.query_one_raw(statement( + "SELECT mst2_metadata_gc_apply($1::uuid)", + [claim.operation_id.clone().into()], + )) + .await + .map_err(internal)?; + load_gc(txn, &claim.operation_id) + .await? + .ok_or_else(|| integrity("applied metadata GC operation disappeared"))? + .receipt() + } + + /// Absence is definitive only after a fresh same-primary completion barrier. + pub async fn inspect_gc( + &self, + fresh_primary: &DatabaseConnection, + operation_id: &str, + lifetime: &MetadataLifetime, + phase: MetadataGcPhase, + ) -> Result { + validate_gc_operation(operation_id)?; + self.require_lifetime_scope(lifetime)?; + let txn = fresh_primary + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .map_err(|_| gc_uncertain(operation_id, lifetime, phase))?; + if self.inner.inner.barrier(&txn).await.is_err() { + let _ = txn.rollback().await; + return Err(gc_uncertain(operation_id, lifetime, phase)); + } + let result = async { + let Some(record) = load_gc(&txn, operation_id).await? else { + return Ok(MetadataGcObservation::Absent); + }; + if &record.claim.lifetime != lifetime { + return Err(integrity( + "GC inspection differs from its captured lifetime", + )); + } + match record.state.as_str() { + "PENDING" => Ok(MetadataGcObservation::Pending(Box::new(record.claim))), + "APPLIED" => Ok(MetadataGcObservation::Applied(Box::new(record.receipt()?))), + _ => Err(integrity("unknown metadata GC operation state")), + } + } + .await; + txn.rollback() + .await + .map_err(|_| gc_uncertain(operation_id, lifetime, phase))?; + result.map_err(|error: SnapshotError| { + if error.code == SnapshotErrorCode::Internal { + gc_uncertain(operation_id, lifetime, phase) + } else { + MetadataGcError::Rejected(error) + } + }) + } + + pub async fn pending_gc(&self, limit: u16) -> Result, SnapshotError> { + if !(1..=256).contains(&limit) { + return Err(integrity("metadata pending GC limit must be 1..=256")); + } + self.inner + .inner + .verify_primary_connection(&self.inner.inner.connection) + .await?; + let rows=self.inner.inner.connection.query_all_raw(statement( + "SELECT operation_id::text,page_id,generation,primary_scope,metadata_codec,expected_size, + graph_present,had_payload,state,created_at,completed_at FROM mst2_metadata_gc_op WHERE state='PENDING' + ORDER BY created_at,operation_id LIMIT $1", + [i64::from(limit).into()], + )).await.map_err(internal)?; + let mut claims = Vec::with_capacity(rows.len()); + for row in rows { + let record = gc_record(row)?; + self.require_lifetime_scope(&record.claim.lifetime)?; + claims.push(record.claim); + } + Ok(claims) + } + + pub async fn begin_fresh_intent( + &self, + operation_id: &str, + prepared: &PreparedNativeMetadataRetention, + applied: &[MetadataGcReceipt], + ) -> Result { + if applied.is_empty() || applied.len() > MetadataDagLimits::default().nodes { + return Err(integrity("fresh begin requires bounded exact APPLIED receipts").into()); + } + for receipt in applied { + self.require_lifetime_scope(&receipt.claim.lifetime)?; + } + self.inner + .begin_with_reopens(operation_id, prepared, applied) + .await + } +} + +fn validate_gc_operation(operation: &str) -> Result<(), SnapshotError> { + let uuid = uuid::Uuid::parse_str(operation).map_err(internal)?; + if uuid.get_version_num() != 4 || uuid.to_string() != operation { + return Err(integrity("metadata GC operation must be canonical UUID v4")); + } + Ok(()) +} + +fn gc_uncertain( + operation: &str, + lifetime: &MetadataLifetime, + phase: MetadataGcPhase, +) -> MetadataGcError { + MetadataGcError::CommitUncertain { + operation_id: operation.into(), + page_id: lifetime.page_id, + generation: lifetime.generation, + phase, + } +} + +async fn finish_gc_transaction( + txn: DatabaseTransaction, + result: Result, + operation: &str, + lifetime: &MetadataLifetime, + phase: MetadataGcPhase, +) -> Result { + match result { + Ok(value) => { + txn.commit() + .await + .map_err(|_| gc_uncertain(operation, lifetime, phase))?; + Ok(value) + } + Err(error) => { + txn.rollback().await.map_err(internal)?; + Err(error.into()) + } + } +} + +async fn load_gc( + connection: &C, + operation: &str, +) -> Result, SnapshotError> { + let Some(row)=connection.query_one_raw(statement( + "SELECT operation_id::text,page_id,generation,primary_scope,metadata_codec,expected_size, + graph_present,had_payload,state,created_at,completed_at FROM mst2_metadata_gc_op WHERE operation_id=$1::uuid", + [operation.into()], + )).await.map_err(internal)? else { return Ok(None); }; + Ok(Some(gc_record(row)?)) +} + +fn gc_record(row: QueryResult) -> Result { + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + Ok(GcRecord { + claim: MetadataGcClaim { + operation_id: row.try_get("", "operation_id").map_err(internal)?, + lifetime: MetadataLifetime { + page_id: page.as_slice().try_into().map_err(internal)?, + generation: row.try_get("", "generation").map_err(internal)?, + metadata_codec: row.try_get("", "metadata_codec").map_err(internal)?, + expected_size: row.try_get("", "expected_size").map_err(internal)?, + primary_scope: row + .try_get::>("", "primary_scope") + .map_err(internal)? + .into_boxed_slice(), + }, + graph_present: row.try_get("", "graph_present").map_err(internal)?, + had_payload: row.try_get("", "had_payload").map_err(internal)?, + }, + state: row.try_get("", "state").map_err(internal)?, + created_at: row.try_get("", "created_at").map_err(internal)?, + completed_at: row.try_get("", "completed_at").map_err(internal)?, + }) +} + +async fn validate_applied_receipts( + connection: &C, + receipts: &[MetadataGcReceipt], +) -> Result<(), SnapshotError> { + if receipts.is_empty() { + return Ok(()); + } + let operations: Vec<_> = receipts + .iter() + .map(|r| r.claim.operation_id.as_str()) + .collect(); + let rows=connection.query_all_raw(statement( + "SELECT operation_id::text,page_id,generation,primary_scope,metadata_codec,expected_size, + graph_present,had_payload,state,created_at,completed_at FROM mst2_metadata_gc_op + WHERE operation_id IN (SELECT jsonb_array_elements_text($1::jsonb)::uuid) LIMIT 4097", + [serde_json::to_string(&operations).map_err(internal)?.into()], + )).await.map_err(internal)?; + let mut actual = BTreeMap::new(); + for row in rows { + let record = gc_record(row)?; + actual.insert(record.claim.operation_id.clone(), record.receipt()?); + } + if actual.len() != receipts.len() + || receipts + .iter() + .any(|r| actual.get(&r.claim.operation_id) != Some(r)) + { + return Err(integrity( + "fresh proof differs from exact immutable APPLIED operation and timestamps", + )); + } + Ok(()) +} + +pub(super) async fn reopen_in_txn( + txn: &DatabaseTransaction, + plan: &MetadataInstallPlan, + receipts: &[MetadataGcReceipt], +) -> Result<(), SnapshotError> { + let mut pages = BTreeSet::new(); + validate_applied_receipts(txn, receipts).await?; + for receipt in receipts { + let lifetime = &receipt.claim.lifetime; + if !pages.insert(lifetime.page_id) + || plan.pages.get(&lifetime.page_id).copied() != Some(lifetime.expected_size as u64) + || plan.identity.metadata_codec as i16 != lifetime.metadata_codec + { + return Err(integrity( + "fresh receipts are not unique exact members of the new plan", + )); + } + lifetime + .generation + .checked_add(1) + .ok_or_else(|| unavailable("metadata generation watermark exhausted"))?; + } + let operations: Vec<_> = receipts + .iter() + .map(|r| r.claim.operation_id.as_str()) + .collect(); + let encoded = serde_json::to_string(&operations).map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size,graph_domain) + SELECT o.page_id,'page:sha256:'||encode(o.page_id,'hex'),o.generation+1,'RESERVED',o.metadata_codec,o.expected_size,'qualified-v1' + FROM mst2_metadata_gc_op o WHERE operation_id IN (SELECT jsonb_array_elements_text($1::jsonb)::uuid) + ORDER BY o.page_id ON CONFLICT(page_id,generation) DO NOTHING", + [encoded.clone().into()], + )).await.map_err(internal)?; + let changed=txn.execute_raw(statement( + "UPDATE mst2_metadata_current c SET generation=o.generation+1 FROM mst2_metadata_gc_op o + WHERE o.operation_id IN (SELECT jsonb_array_elements_text($1::jsonb)::uuid) + AND c.page_id=o.page_id AND c.generation=o.generation", + [encoded.into()], + )).await.map_err(internal)?; + if changed.rows_affected() != receipts.len() as u64 { + return Err(unavailable( + "fresh receipts do not name all current removed incarnations", + )); + } + Ok(()) +} + +pub(super) async fn verify_reopen_replay( + connection: &C, + fixed: &FixedPlan, + receipts: &[MetadataGcReceipt], +) -> Result<(), SnapshotError> { + validate_applied_receipts(connection, receipts).await?; + let mut pages = BTreeSet::new(); + for receipt in receipts { + let old = &receipt.claim.lifetime; + if !pages.insert(old.page_id) + || old.primary_scope != fixed.intent.primary_scope + || old.generation.checked_add(1) != fixed.bindings.0.get(&old.page_id).map(|b| b.0) + || Some(old.expected_size as u64) != fixed.bindings.0.get(&old.page_id).map(|b| b.1) + { + return Err(integrity( + "fresh replay receipts differ from the fixed new incarnation bindings", + )); + } + } + Ok(()) +} + +pub(super) async fn allocate_lifetimes( + txn: &DatabaseTransaction, + plan: &MetadataInstallPlan, +) -> Result { + let pages: Vec<_> = plan + .pages + .iter() + .map(|(id, size)| json!({"id":hex::encode(id),"size":size})) + .collect(); + let encoded = serde_json::to_string(&pages).map_err(internal)?; + if txn.query_one_raw(statement( + "SELECT l.page_id FROM jsonb_to_recordset($1::jsonb) p(id text,size integer) + JOIN mst2_metadata_current c ON c.page_id=decode(p.id,'hex') + JOIN mst2_metadata_lifetime l USING(page_id,generation) WHERE l.graph_domain<>'qualified-v1' LIMIT 1", + [encoded.clone().into()], + )).await.map_err(internal)?.is_some() { + return Err(unavailable("metadata incarnation permanently belongs to another graph domain")); + } + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size,graph_domain) + SELECT decode(p.id,'hex'),'page:sha256:'||p.id,1,'RESERVED',$1,p.size,'qualified-v1' + FROM jsonb_to_recordset($2::jsonb) p(id text,size integer) + WHERE NOT EXISTS(SELECT 1 FROM mst2_metadata_current c WHERE c.page_id=decode(p.id,'hex')) + ON CONFLICT(page_id,generation) DO NOTHING", + [(plan.identity.metadata_codec as i16).into(),encoded.clone().into()], + )).await.map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_current(page_id,generation) SELECT decode(p.id,'hex'),1 + FROM jsonb_to_recordset($1::jsonb) p(id text,size integer) + WHERE NOT EXISTS(SELECT 1 FROM mst2_metadata_current c WHERE c.page_id=decode(p.id,'hex'))", + [encoded.clone().into()], + )).await.map_err(internal)?; + let rows=txn.query_all_raw(statement( + "SELECT l.page_id,l.generation,l.state,l.graph_domain,l.metadata_codec,l.expected_size,p.size, + n.state AS graph_state,n.metadata_codec AS graph_codec,n.bytes AS graph_bytes, + b.page_id IS NOT NULL AS payload_present,b.generation AS payload_generation,b.metadata_codec AS payload_codec,b.byte_size AS payload_size, + EXISTS(SELECT 1 FROM mst2_metadata_gc_op o WHERE o.page_id=l.page_id AND o.generation=l.generation) AS tombstone, + EXISTS(SELECT 1 FROM mst2_retention_node x WHERE x.node_id=l.node_id) + OR EXISTS(SELECT 1 FROM mst2_retention_gc_op x WHERE x.node_id=l.node_id) AS generic_graph + FROM jsonb_to_recordset($1::jsonb) p(id text,size integer) + JOIN mst2_metadata_current c ON c.page_id=decode(p.id,'hex') + JOIN mst2_metadata_lifetime l USING(page_id,generation) + LEFT JOIN mst2_metadata_graph_node n USING(page_id,generation) + LEFT JOIN mst2_metadata_payload b ON b.page_id=l.page_id ORDER BY l.page_id", + [encoded.into()], + )).await.map_err(internal)?; + let mut bindings = BTreeMap::new(); + for row in rows { + let state: String = row.try_get("", "state").map_err(internal)?; + let generation: i64 = row.try_get("", "generation").map_err(internal)?; + let size: i32 = row.try_get("", "size").map_err(internal)?; + let graph: Option = row.try_get("", "graph_state").map_err(internal)?; + let payload: bool = row.try_get("", "payload_present").map_err(internal)?; + if !["RESERVED", "LIVE"].contains(&state.as_str()) + || row.try_get::("", "tombstone").map_err(internal)? + || row.try_get::("", "generic_graph").map_err(internal)? + || graph.as_deref().is_some_and(|s| s != "LIVE") + { + return Err(unavailable( + "qualified incarnation is removed, deleting, or has generic evidence", + )); + } + if state == "LIVE" && (!payload || graph.is_none()) { + return Err(unavailable( + "LIVE qualified incarnation lost bytes or graph", + )); + } + if row + .try_get::("", "graph_domain") + .map_err(internal)? + != "qualified-v1" + || row.try_get::("", "metadata_codec").map_err(internal)? + != plan.identity.metadata_codec as i16 + || row.try_get::("", "expected_size").map_err(internal)? != size + || (graph.is_some() + && (row + .try_get::>("", "graph_codec") + .map_err(internal)? + != Some(plan.identity.metadata_codec as i16) + || row + .try_get::>("", "graph_bytes") + .map_err(internal)? + != Some(i64::from(size)))) + || (payload + && (row + .try_get::>("", "payload_generation") + .map_err(internal)? + != Some(generation) + || row + .try_get::>("", "payload_codec") + .map_err(internal)? + != Some(plan.identity.metadata_codec as i16) + || row + .try_get::>("", "payload_size") + .map_err(internal)? + != Some(size))) + { + return Err(integrity( + "qualified incarnation immutable profile mismatch", + )); + } + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + bindings.insert( + page.as_slice().try_into().map_err(internal)?, + (generation, size as u64), + ); + } + if bindings.len() != plan.pages.len() { + return Err(integrity("qualified lifetime allocation is incomplete")); + } + Ok(GenerationBindings(bindings)) +} + +pub(super) async fn retain_existing_roots( + txn: &DatabaseTransaction, + intent: &GenerationPrepareIntent, +) -> Result<(), SnapshotError> { + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_graph_root(prepare_id,storage_seal,page_id,generation) + SELECT q.prepare_id,q.storage_seal,m.page_id,m.generation FROM mst2_metadata_prepare q + JOIN mst2_metadata_prepare_page m USING(prepare_id) + JOIN mst2_metadata_graph_node n USING(page_id,generation) WHERE q.prepare_id=$1 AND n.state='LIVE' + ON CONFLICT(prepare_id,page_id,generation) DO NOTHING", + [intent.prepare_id().into()], + )).await.map_err(internal)?; + Ok(()) +} + +pub(super) async fn check_lifetimes( + connection: &C, + stored: &StoredPlan, +) -> Result<(), SnapshotError> { + if connection.query_one_raw(statement( + "SELECT m.page_id FROM mst2_metadata_prepare_page m + LEFT JOIN mst2_metadata_current c ON c.page_id=m.page_id AND c.generation=m.generation + LEFT JOIN mst2_metadata_lifetime l ON l.page_id=m.page_id AND l.generation=m.generation + LEFT JOIN mst2_metadata_graph_node n ON n.page_id=m.page_id AND n.generation=m.generation + WHERE m.prepare_id=$1 AND (c.page_id IS NULL OR l.page_id IS NULL OR l.graph_domain<>'qualified-v1' + OR l.state NOT IN ('RESERVED','LIVE') OR l.metadata_codec<>$2 OR l.expected_size<>m.expected_size + OR ($3='COMMITTED' AND l.state<>'LIVE') OR (l.state='LIVE' AND n.page_id IS NULL) + OR n.state<>'LIVE' OR n.metadata_codec<>$2 OR n.bytes<>m.expected_size + OR EXISTS(SELECT 1 FROM mst2_metadata_gc_op o WHERE o.page_id=m.page_id AND o.generation=m.generation) + OR EXISTS(SELECT 1 FROM mst2_retention_node x WHERE x.node_id=l.node_id) + OR EXISTS(SELECT 1 FROM mst2_retention_gc_op x WHERE x.node_id=l.node_id)) LIMIT 1", + [stored.record.prepare_id.clone().into(),stored.record.metadata_codec.into(),stored.record.state.clone().into()], + )).await.map_err(internal)?.is_some() { return Err(unavailable("fixed qualified incarnation is no longer installable")); } + Ok(()) +} + +pub(super) async fn finalize_graph( + txn: &DatabaseTransaction, + fixed: &FixedPlan, + dag: &ValidatedMetadataDag, +) -> Result { + let actual_pages: BTreeSet<_> = dag.payloads().iter().map(|p| (p.id, p.size)).collect(); + let actual_edges: BTreeSet<_> = dag + .edges() + .iter() + .map(|e| (e.parent.clone(), e.child.clone())) + .collect(); + if dag.root() != fixed.stored.plan.root + || actual_pages + != fixed + .stored + .plan + .pages + .iter() + .map(|(p, s)| (*p, *s)) + .collect() + || actual_edges + != fixed + .stored + .plan + .edges + .iter() + .map(|(p, c)| (node_id(p), node_id(c))) + .collect() + { + return Err(integrity( + "qualified DAG observation differs from immutable plan", + )); + } + if fixed.stored.record.state == "COMMITTED" { + verify_graph(txn, fixed).await?; + return fixed.stored.receipt(); + } + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_graph_node(page_id,generation,state,metadata_codec,bytes) + SELECT m.page_id,m.generation,'LIVE',$2,m.expected_size FROM mst2_metadata_prepare_page m + WHERE m.prepare_id=$1 AND NOT EXISTS(SELECT 1 FROM mst2_metadata_graph_node n WHERE n.page_id=m.page_id AND n.generation=m.generation)", + [fixed.intent.prepare_id().into(),fixed.stored.record.metadata_codec.into()], + )).await.map_err(internal)?; + let edges:Vec<_>=fixed.stored.plan.edges.iter().map(|(p,c)|json!({"parent":hex::encode(p),"pg":fixed.bindings.0[p].0,"child":hex::encode(c),"cg":fixed.bindings.0[c].0})).collect(); + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_graph_edge(parent_page,parent_generation,child_page,child_generation) + SELECT decode(e.parent,'hex'),e.pg,decode(e.child,'hex'),e.cg + FROM jsonb_to_recordset($1::jsonb) e(parent text,pg bigint,child text,cg bigint) + ON CONFLICT(parent_page,parent_generation,child_page,child_generation) DO NOTHING", + [serde_json::to_string(&edges).map_err(internal)?.into()], + )).await.map_err(internal)?; + retain_existing_roots(txn, &fixed.intent).await?; + verify_graph(txn, fixed).await?; + let changed = txn + .execute_raw(statement( + "UPDATE mst2_metadata_prepare SET state='COMMITTED',committed_at=clock_timestamp() + WHERE prepare_id=$1 AND state='PREPARING' AND storage_seal=$2", + [ + fixed.intent.prepare_id().into(), + fixed.intent.storage_seal.to_vec().into(), + ], + )) + .await + .map_err(internal)?; + if changed.rows_affected() != 1 { + return Err(integrity("qualified finalize lost its prepare CAS")); + } + fixed.stored.receipt() +} + +pub(super) async fn verify_graph( + connection: &C, + fixed: &FixedPlan, +) -> Result<(), SnapshotError> { + let rows=connection.query_all_raw(statement( + "SELECT n.page_id,n.generation,n.state,n.metadata_codec,n.bytes,n.incoming_refs, + (SELECT count(*) FROM mst2_metadata_graph_edge e WHERE e.child_page=n.page_id AND e.child_generation=n.generation) AS actual_refs + FROM mst2_metadata_prepare_page m JOIN mst2_metadata_graph_node n USING(page_id,generation) + WHERE m.prepare_id=$1 ORDER BY n.page_id LIMIT 4097", + [fixed.intent.prepare_id().into()], + )).await.map_err(internal)?; + if rows.len() != fixed.bindings.0.len() { + return Err(unavailable("qualified graph coverage is incomplete")); + } + for row in rows { + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + let page: [u8; 32] = page.as_slice().try_into().map_err(internal)?; + let &(generation, size) = fixed + .bindings + .0 + .get(&page) + .ok_or_else(|| integrity("qualified graph has an extra node"))?; + if row.try_get::("", "generation").map_err(internal)? != generation + || row.try_get::("", "state").map_err(internal)? != "LIVE" + || row.try_get::("", "metadata_codec").map_err(internal)? + != fixed.stored.record.metadata_codec + || row.try_get::("", "bytes").map_err(internal)? != size as i64 + || row.try_get::("", "incoming_refs").map_err(internal)? + != row.try_get::("", "actual_refs").map_err(internal)? + { + return Err(integrity("qualified graph profile or counter drift")); + } + } + let rows=connection.query_all_raw(statement( + "SELECT e.parent_page,e.parent_generation,e.child_page,e.child_generation FROM mst2_metadata_prepare_page m + JOIN mst2_metadata_graph_edge e ON e.parent_page=m.page_id AND e.parent_generation=m.generation + WHERE m.prepare_id=$1 LIMIT 16385", + [fixed.intent.prepare_id().into()], + )).await.map_err(internal)?; + let mut actual = BTreeSet::new(); + for row in rows { + let parent: Vec = row.try_get("", "parent_page").map_err(internal)?; + let child: Vec = row.try_get("", "child_page").map_err(internal)?; + actual.insert(( + parent, + row.try_get::("", "parent_generation") + .map_err(internal)?, + child, + row.try_get::("", "child_generation") + .map_err(internal)?, + )); + } + let expected: BTreeSet<_> = fixed + .stored + .plan + .edges + .iter() + .map(|(p, c)| { + ( + p.to_vec(), + fixed.bindings.0[p].0, + c.to_vec(), + fixed.bindings.0[c].0, + ) + }) + .collect(); + if actual != expected { + return Err(integrity( + "qualified graph edges differ from immutable plan", + )); + } + verify_roots(connection, fixed, true).await +} + +pub(super) async fn verify_roots( + connection: &C, + fixed: &FixedPlan, + complete: bool, +) -> Result<(), SnapshotError> { + let rows=connection.query_all_raw(statement( + "SELECT page_id,generation,storage_seal FROM mst2_metadata_graph_root WHERE prepare_id=$1 LIMIT 4097", + [fixed.intent.prepare_id().into()], + )).await.map_err(internal)?; + let mut actual = BTreeSet::new(); + for row in rows { + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + let page: [u8; 32] = page.as_slice().try_into().map_err(internal)?; + let generation: i64 = row.try_get("", "generation").map_err(internal)?; + let seal: Vec = row.try_get("", "storage_seal").map_err(internal)?; + if seal.as_slice() != fixed.intent.storage_seal + || fixed.bindings.0.get(&page).map(|b| b.0) != Some(generation) + { + return Err(integrity( + "qualified prepare root differs from its immutable seal", + )); + } + actual.insert((page, generation)); + } + if complete + && actual + != fixed + .bindings + .0 + .iter() + .map(|(p, (g, _))| (*p, *g)) + .collect() + { + return Err(unavailable( + "qualified prepare lost or transferred original roots", + )); + } + Ok(()) +} + +#[cfg(test)] +#[path = "native_metadata_qualified_tests.rs"] +mod tests; diff --git a/src/jupiter/storage/native_metadata_qualified_tests.rs b/src/jupiter/storage/native_metadata_qualified_tests.rs new file mode 100644 index 00000000..a7b73493 --- /dev/null +++ b/src/jupiter/storage/native_metadata_qualified_tests.rs @@ -0,0 +1,1261 @@ +use std::sync::Arc; + +use mst2_codec::metapage::{Entry, EntryKind}; +use sea_orm::Database; +use sea_orm_migration::MigratorTrait; + +use super::{ + super::history::{MetadataTerminalAction, MetadataTerminalError, MetadataTerminalObservation}, + *, +}; +use crate::{ + ceres::snapshot::retention_dag::MetadataDagBuilder, + jupiter::{ + migration::Migrator, + tests::{TestSchemaGuard, test_db_config}, + }, +}; + +fn prepared() -> PreparedNativeMetadataRetention { + let entries = [Entry::file(EntryKind::Regular, b"file", 3, [42; 32])]; + let child = Page::build(&entries).unwrap(); + let roots = [ + Entry::dir(b"one", page_id(&child)), + Entry::dir(b"two", page_id(&child)), + ]; + let root = Page::build(&roots).unwrap(); + let mut builder = MetadataDagBuilder::new(MetadataDagLimits::default()); + builder.add_directory(&child, &entries).unwrap(); + builder.add_directory(&root, &roots).unwrap(); + PreparedNativeMetadataRetention::test_installation( + Arc::new(builder.finish(page_id(&root)).unwrap()), + "/", + ) +} + +async fn fixture() -> (DatabaseConnection, DatabaseConnection, TestSchemaGuard) { + let temp = tempfile::tempdir().unwrap(); + let (config, schema) = test_db_config(temp.path()).await; + let first = Database::connect(config.db_url.clone()).await.unwrap(); + Migrator::up(&first, None).await.unwrap(); + let second = Database::connect(config.db_url).await.unwrap(); + (first, second, schema) +} + +async fn scalar(db: &C, sql: &str) -> i64 { + db.query_one_raw(statement(sql, [])) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +async fn installed( + db: &DatabaseConnection, + operation: &str, +) -> ( + PostgresQualifiedMetadataRepository, + PreparedNativeMetadataRetention, + GenerationMetadataReceipt, +) { + let repo = PostgresQualifiedMetadataRepository::new(db.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repo.begin_intent(operation, &pages).await.unwrap(); + repo.install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = repo.finalize(&intent).await.unwrap(); + (repo, pages, receipt) +} + +async fn claim_root( + repo: &PostgresQualifiedMetadataRepository, + receipt: &GenerationMetadataReceipt, +) -> MetadataGcClaim { + let lifetime = repo.lifetime(receipt.metadata_root()).await.unwrap(); + repo.claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .unwrap() +} + +async fn wait_for_waiter(txn: &DatabaseTransaction) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(4); + loop { + let waiting:bool=txn.query_one_raw(statement( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND NOT granted + AND classid=$1::integer::oid AND objid=hashtext(current_schema())::oid + AND database=(SELECT oid FROM pg_database WHERE datname=current_database())) AS waiting", + [RETENTION_LOCK_KEY.into()], + )).await.unwrap().unwrap().try_get("","waiting").unwrap(); + if waiting { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "real independent connection must wait on retention barrier" + ); + tokio::task::yield_now().await; + } +} + +#[tokio::test] +async fn qualified_graph_exact_fks_unique_edges_and_permanent_domains() { + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "qualified-graph").await; + repo.install_pages(receipt.intent(), pages.dag().payloads()) + .await + .unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_node").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_edge").await, + 1 + ); + assert_eq!( + scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_metadata_graph_node" + ) + .await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE graph_domain='qualified-v1'" + ) + .await, + 2 + ); + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let generic = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + assert!( + generic + .begin_intent("cannot-adopt-retired-qualified", &pages) + .await + .is_err() + ); + for sql in [ + "UPDATE mst2_metadata_lifetime SET graph_domain='generic-v1'", + "UPDATE mst2_metadata_graph_node SET incoming_refs=incoming_refs+1", + "UPDATE mst2_metadata_graph_edge SET child_generation=child_generation+1", + "DELETE FROM mst2_metadata_current", + "UPDATE mst2_metadata_payload SET payload=payload", + "DELETE FROM mst2_metadata_storage_scope", + ] { + assert!(second.execute_unprepared(sql).await.is_err(), "{sql}"); + } + let claim = claim_root(&repo, &receipt).await; + let applied = repo.apply(&claim).await.unwrap(); + assert_eq!(applied.claim(), &claim); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_gc_op WHERE state='APPLIED'" + ) + .await, + 1 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 2 + ); +} + +#[tokio::test] +async fn qualified_shared_child_and_other_prepare_are_real_collection_vetoes() { + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "shared-owner-one").await; + let other = PostgresQualifiedMetadataRepository::new(second) + .await + .unwrap(); + let intent = other + .begin_intent("shared-owner-two", &pages) + .await + .unwrap(); + let shared = other.finalize(&intent).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_root").await, + 4 + ); + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let root = repo.lifetime(receipt.metadata_root()).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &root) + .await + .is_err() + ); + other.retire_prepare_coverage(&shared).await.unwrap(); + let child = pages + .dag() + .payloads() + .iter() + .find(|p| p.id != receipt.metadata_root()) + .unwrap(); + let child = repo.lifetime(child.id).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &child) + .await + .is_err() + ); + let root_claim = repo + .claim(&uuid::Uuid::new_v4().to_string(), &root) + .await + .unwrap(); + repo.apply(&root_claim).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT incoming_refs FROM mst2_metadata_graph_node").await, + 0 + ); + let child_claim = repo + .claim(&uuid::Uuid::new_v4().to_string(), &child) + .await + .unwrap(); + repo.apply(&child_claim).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); +} + +#[tokio::test] +async fn qualified_pending_preparation_and_lost_roots_fail_closed() { + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "lost-roots").await; + second + .execute_raw(statement( + "DELETE FROM mst2_metadata_graph_root WHERE prepare_id=$1", + [receipt.intent().prepare_id().into()], + )) + .await + .unwrap(); + assert!(repo.retire_prepare_coverage(&receipt).await.is_err()); + let lifetime = repo.lifetime(receipt.metadata_root()).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .is_err() + ); + let preparing = repo.begin_intent("pending-owner", &pages).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .is_err() + ); + repo.abort(&preparing).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_gc_op").await, + 0 + ); +} + +#[tokio::test] +async fn qualified_explicit_aborted_orphans_with_and_without_payload_collect() { + for partial in [false, true] { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repo.begin_intent("orphan", &pages).await.unwrap(); + if partial { + repo.install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + } + let lifetime = repo.lifetime(pages.dag().root()).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .is_err() + ); + repo.abort(&intent).await.unwrap(); + let claim = repo + .claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .unwrap(); + assert!(!claim.graph_present); + assert_eq!(claim.had_payload, partial); + repo.apply(&claim).await.unwrap(); + assert_eq!( + repo.inspect_gc( + &second, + claim.operation_id(), + &lifetime, + MetadataGcPhase::Apply + ) + .await + .unwrap(), + MetadataGcObservation::Applied(Box::new(repo.apply(&claim).await.unwrap())) + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE state='REMOVED'" + ) + .await, + 1 + ); + } +} + +#[tokio::test] +async fn qualified_abort_and_collection_fence_a_late_installer_at_actual_pg_lock() { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repo.begin_intent("late-installer", &pages).await.unwrap(); + repo.install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let lifetime = repo.lifetime(pages.dag().root()).await.unwrap(); + let held = first.begin().await.unwrap(); + repo.inner.inner.barrier(&held).await.unwrap(); + let late_intent = intent.clone(); + let late_pages = pages.dag().payloads().to_vec(); + let late = tokio::spawn(async move { + let late_repo = PostgresQualifiedMetadataRepository::new(second) + .await + .unwrap(); + late_repo.install_pages(&late_intent, &late_pages).await + }); + wait_for_waiter(&held).await; + repo.terminate_in_txn(&held, &intent, MetadataTerminalAction::Abort, None) + .await + .unwrap(); + let claim = repo + .claim_in_txn(&held, &uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .unwrap(); + repo.apply_in_txn(&held, &claim).await.unwrap(); + held.commit().await.unwrap(); + assert!(late.await.unwrap().is_err()); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_gc_op WHERE state='APPLIED'" + ) + .await, + 1 + ); +} + +#[tokio::test] +async fn qualified_statement_barrier_raw_writer_waits_before_row_lock_and_rechecks_domain() { + for generic in [true, false] { + let (first, second, _schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "raw-writer").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let root = repo.lifetime(receipt.metadata_root()).await.unwrap(); + let held = first.begin().await.unwrap(); + repo.inner.inner.barrier(&held).await.unwrap(); + let sql = if generic { + format!( + "INSERT INTO mst2_retention_node(node_id,kind,bytes,state) VALUES('page:sha256:{}','page',{},'LIVE')", + hex::encode(root.page_id), + root.expected_size + ) + } else { + format!( + "UPDATE mst2_metadata_graph_node SET state='LIVE' WHERE page_id=decode('{}','hex') AND generation={}", + hex::encode(root.page_id), + root.generation + ) + }; + let writer = tokio::spawn(async move { second.execute_unprepared(&sql).await }); + wait_for_waiter(&held).await; + held.query_one_raw(statement("SELECT page_id FROM mst2_metadata_graph_node WHERE page_id=$1 AND generation=$2 FOR UPDATE NOWAIT", + [root.page_id.to_vec().into(),root.generation.into()])).await.unwrap().unwrap(); + let claim = repo + .claim_in_txn(&held, &uuid::Uuid::new_v4().to_string(), &root) + .await + .unwrap(); + held.commit().await.unwrap(); + assert!(writer.await.unwrap().is_err()); + repo.apply(&claim).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + } +} + +#[tokio::test] +async fn qualified_claim_outer_rollback_and_lost_commit_response_use_fresh_barrier() { + let (first, second, _schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "claim-recovery").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let lifetime = repo.lifetime(receipt.metadata_root()).await.unwrap(); + let id = uuid::Uuid::new_v4().to_string(); + let held = first.begin().await.unwrap(); + repo.claim_in_txn(&held, &id, &lifetime).await.unwrap(); + held.rollback().await.unwrap(); + assert_eq!( + repo.inspect_gc(&second, &id, &lifetime, MetadataGcPhase::Claim) + .await + .unwrap(), + MetadataGcObservation::Absent + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_graph_node WHERE state='LIVE'" + ) + .await, + 2 + ); + let held = first.begin().await.unwrap(); + let claim = repo.claim_in_txn(&held, &id, &lifetime).await.unwrap(); + held.commit().await.unwrap(); // The caller deliberately discards the commit response. + assert_eq!( + repo.inspect_gc(&second, &id, &lifetime, MetadataGcPhase::Claim) + .await + .unwrap(), + MetadataGcObservation::Pending(Box::new(claim.clone())) + ); + assert_eq!(repo.pending_gc(1).await.unwrap(), vec![claim.clone()]); + repo.apply(&claim).await.unwrap(); +} + +#[tokio::test] +async fn qualified_apply_faults_rollback_bytes_graph_counters_and_receipt_with_guards_enabled() { + for (table, event, predicate) in [ + ("mst2_metadata_payload", "AFTER DELETE", "true"), + ("mst2_metadata_graph_edge", "AFTER DELETE", "true"), + ( + "mst2_metadata_gc_op", + "BEFORE UPDATE", + "NEW.state='APPLIED'", + ), + ] { + let (first, second, _schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "apply-fault").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let claim = claim_root(&repo, &receipt).await; + let granularity = if table == "mst2_metadata_graph_edge" { + "STATEMENT" + } else { + "ROW" + }; + let sql = format!( + "CREATE FUNCTION test_gc_fault() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF {predicate} THEN RAISE EXCEPTION 'injected atomic GC failure'; END IF; RETURN NEW; END $$; CREATE TRIGGER zzz_test_gc_fault {event} ON {table} FOR EACH {granularity} EXECUTE FUNCTION test_gc_fault()" + ); + second.execute_unprepared(&sql).await.unwrap(); + assert!(repo.apply(&claim).await.is_err()); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_edge").await, + 1 + ); + assert_eq!( + scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_metadata_graph_node" + ) + .await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_node").await, + 2 + ); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_gc_op WHERE state='PENDING' AND payload_delete_xid IS NULL").await,1); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE state='DELETING'" + ) + .await, + 1 + ); + assert_eq!(scalar(&first,"SELECT count(*) FROM pg_trigger WHERE tgname IN ('mst2_metadata_payload_fenced','mst2_metadata_payload_removed','mst2_metadata_graph_node_guard','mst2_metadata_gc_op_guard') AND tgenabled='O' AND tgrelid IN (SELECT oid FROM pg_class WHERE relnamespace=(SELECT oid FROM pg_namespace WHERE nspname=current_schema()))").await,4); + second + .execute_unprepared(&format!("DROP TRIGGER zzz_test_gc_fault ON {table}")) + .await + .unwrap(); + repo.apply(&claim).await.unwrap(); + } +} + +#[tokio::test] +async fn qualified_direct_selected_delete_completes_all_bookkeeping_and_guc_is_not_authority() { + let (first, second, _schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "direct-delete").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let claim = claim_root(&repo, &receipt).await; + for sql in [ + "UPDATE mst2_metadata_gc_op SET had_payload=false", + "UPDATE mst2_metadata_gc_op SET graph_present=false", + "UPDATE mst2_metadata_gc_op SET generation=generation+1", + ] { + assert!(second.execute_unprepared(sql).await.is_err(), "{sql}"); + } + assert!( + second + .query_one_raw(statement( + "SELECT mst2_metadata_gc_finish($1::uuid)", + [claim.operation_id.clone().into()] + )) + .await + .is_err() + ); + second + .execute_unprepared("UPDATE mst2_metadata_gc_op SET payload_delete_xid=1") + .await + .unwrap(); + let marked = second.begin().await.unwrap(); + marked + .execute_raw(statement( + "UPDATE mst2_metadata_gc_op SET payload_delete_xid=42 WHERE operation_id=$1::uuid", + [claim.operation_id.clone().into()], + )) + .await + .unwrap(); + let same_xid:bool=marked.query_one_raw(statement( + "SELECT payload_delete_xid=txid_current() AS same_xid FROM mst2_metadata_gc_op WHERE operation_id=$1::uuid", + [claim.operation_id.clone().into()], + )).await.unwrap().unwrap().try_get("","same_xid").unwrap(); + assert!( + same_xid, + "database must replace a caller marker with the actual transaction identity" + ); + let error = marked + .query_one_raw(statement( + "SELECT mst2_metadata_gc_finish($1::uuid)", + [claim.operation_id.clone().into()], + )) + .await + .unwrap_err(); + assert!( + error.to_string().contains("payload presence changed"), + "{error}" + ); + marked.rollback().await.unwrap(); + assert!( + second + .query_one_raw(statement( + "SELECT mst2_metadata_gc_finish($1::uuid)", + [claim.operation_id.clone().into()] + )) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + assert!( + second + .execute_unprepared("DELETE FROM mst2_metadata_payload") + .await + .is_err() + ); + let txn = second.begin().await.unwrap(); + txn.execute_raw(statement( + "SELECT set_config('mega2.metadata_gc_operation',$1,true)", + [uuid::Uuid::new_v4().to_string().into()], + )) + .await + .unwrap(); + assert!( + txn.execute_raw(statement( + "DELETE FROM mst2_metadata_payload WHERE page_id=$1 AND generation=$2", + [ + claim.lifetime.page_id.to_vec().into(), + claim.lifetime.generation.into() + ] + )) + .await + .is_err() + ); + txn.rollback().await.unwrap(); + let txn = second.begin().await.unwrap(); + txn.execute_raw(statement( + "SELECT set_config('mega2.metadata_gc_operation',$1,true)", + [claim.operation_id.clone().into()], + )) + .await + .unwrap(); + txn.execute_raw(statement( + "DELETE FROM mst2_metadata_payload WHERE page_id=$1 AND generation=$2", + [ + claim.lifetime.page_id.to_vec().into(), + claim.lifetime.generation.into(), + ], + )) + .await + .unwrap(); + assert_eq!( + scalar( + &txn, + "SELECT count(*) FROM mst2_metadata_gc_op WHERE state='APPLIED'" + ) + .await, + 1 + ); + assert_eq!( + scalar(&txn, "SELECT count(*) FROM mst2_metadata_graph_edge").await, + 0 + ); + assert_eq!( + scalar(&txn, "SELECT incoming_refs FROM mst2_metadata_graph_node").await, + 0 + ); + txn.commit().await.unwrap(); + repo.apply(&claim).await.unwrap(); + assert!( + second + .execute_unprepared("DELETE FROM mst2_metadata_gc_op") + .await + .is_err() + ); +} + +#[tokio::test] +async fn qualified_fresh_aba_replays_old_receipts_without_deleting_g2_or_reviving_old_intents() { + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "aba-old").await; + let observation = repo.observe_installed_dag(receipt.intent()).await.unwrap(); + let terminal = repo.retire_prepare_coverage(&receipt).await.unwrap(); + let claim = claim_root(&repo, &receipt).await; + second.execute_unprepared("CREATE TABLE test_payload_deletes(n integer NOT NULL); CREATE FUNCTION test_record_delete() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN INSERT INTO test_payload_deletes VALUES(1); RETURN NULL; END $$; CREATE TRIGGER zzz_test_record_delete AFTER DELETE ON mst2_metadata_payload FOR EACH ROW EXECUTE FUNCTION test_record_delete()").await.unwrap(); + let held = first.begin().await.unwrap(); + let applied = repo.apply_in_txn(&held, &claim).await.unwrap(); + held.commit().await.unwrap(); // Deliberately lose delivery; inspect recovers the exact durable receipt. + assert_eq!( + repo.inspect_gc( + &second, + claim.operation_id(), + claim.lifetime(), + MetadataGcPhase::Apply + ) + .await + .unwrap(), + MetadataGcObservation::Applied(Box::new(applied.clone())) + ); + assert!( + repo.begin_intent("ordinary-cannot-reopen", &pages) + .await + .is_err() + ); + let fresh = repo + .begin_fresh_intent("aba-fresh", &pages, std::slice::from_ref(&applied)) + .await + .unwrap(); + assert_eq!( + repo.lifetime(receipt.metadata_root()) + .await + .unwrap() + .generation(), + 2 + ); + repo.install_pages(&fresh, pages.dag().payloads()) + .await + .unwrap(); + repo.finalize(&fresh).await.unwrap(); + assert!( + repo.install_pages(receipt.intent(), pages.dag().payloads()) + .await + .is_err() + ); + assert!( + repo.finalize_observation(receipt.intent(), &observation) + .await + .is_err() + ); + let restarted = PostgresQualifiedMetadataRepository::new(second.clone()) + .await + .unwrap(); + let old = restarted + .historical_lifetime(claim.lifetime.page_id, 1) + .await + .unwrap(); + let recovered = restarted + .inspect_gc(&second, claim.operation_id(), &old, MetadataGcPhase::Apply) + .await + .unwrap(); + let MetadataGcObservation::Applied(recovered) = recovered else { + panic!("restart must recover historical APPLIED receipt"); + }; + assert_eq!(restarted.apply(recovered.claim()).await.unwrap(), applied); + assert!( + restarted + .claim(&uuid::Uuid::new_v4().to_string(), &old) + .await + .is_err() + ); + let terminal_intent = restarted + .capture_terminal_intent( + receipt.intent().operation_id(), + receipt.intent().manifest_digest(), + ) + .await + .unwrap(); + assert_eq!( + restarted + .inspect_terminal( + &second, + &terminal_intent, + MetadataTerminalAction::RetireCoverage + ) + .await + .unwrap(), + MetadataTerminalObservation::Terminated(Box::new(terminal.clone())) + ); + assert_eq!( + repo.retire_prepare_coverage(&receipt).await.unwrap(), + terminal + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM test_payload_deletes").await, + 1 + ); + let stale = second.begin().await.unwrap(); + stale + .execute_raw(statement( + "SELECT set_config('mega2.metadata_gc_operation',$1,true)", + [claim.operation_id.clone().into()], + )) + .await + .unwrap(); + assert!( + stale + .execute_raw(statement( + "DELETE FROM mst2_metadata_payload WHERE page_id=$1 AND generation=2", + [claim.lifetime.page_id.to_vec().into()] + )) + .await + .is_err() + ); + stale.rollback().await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 3 + ); + assert_eq!( + repo.begin_fresh_intent("aba-fresh", &pages, std::slice::from_ref(&applied)) + .await + .unwrap(), + fresh + ); + let mut forged = applied.clone(); + forged.created_at += chrono::Duration::seconds(1); + assert!( + repo.begin_fresh_intent("aba-fresh", &pages, &[forged]) + .await + .is_err() + ); + assert!( + repo.begin_fresh_intent("stale-fresh-proof", &pages, &[applied]) + .await + .is_err() + ); +} + +#[tokio::test] +async fn qualified_multi_page_fresh_failure_rolls_back_all_watermarks_and_new_history() { + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "multi-old").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let root = claim_root(&repo, &receipt).await; + let root = repo.apply(&root).await.unwrap(); + let child_id = pages + .dag() + .payloads() + .iter() + .find(|p| p.id != receipt.metadata_root()) + .unwrap() + .id; + let child = repo.lifetime(child_id).await.unwrap(); + let child = repo + .claim(&uuid::Uuid::new_v4().to_string(), &child) + .await + .unwrap(); + let child = repo.apply(&child).await.unwrap(); + second.execute_unprepared("CREATE FUNCTION test_fresh_fault() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.generation=2 AND NEW.page_id=(SELECT page_id FROM mst2_metadata_current ORDER BY page_id DESC LIMIT 1) THEN RAISE EXCEPTION 'injected multi-page fresh CAS failure'; END IF; RETURN NEW; END $$; CREATE TRIGGER zzz_test_fresh_fault BEFORE UPDATE ON mst2_metadata_current FOR EACH ROW EXECUTE FUNCTION test_fresh_fault()").await.unwrap(); + assert!( + repo.begin_fresh_intent("multi-fresh", &pages, &[root.clone(), child.clone()]) + .await + .is_err() + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_current WHERE generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE operation_id='multi-fresh'" + ) + .await, + 0 + ); + second + .execute_unprepared("DROP TRIGGER zzz_test_fresh_fault ON mst2_metadata_current") + .await + .unwrap(); + let intent = repo + .begin_fresh_intent("multi-fresh", &pages, &[root, child]) + .await + .unwrap(); + repo.install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + repo.finalize(&intent).await.unwrap(); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_current WHERE generation=2" + ) + .await, + 2 + ); +} + +#[tokio::test] +async fn qualified_wrong_primary_and_repeatable_read_never_prove_absence_or_authorize_delete() { + let (first, second, _schema) = fixture().await; + let (other, _other_second, _other_schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "scope").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let claim = claim_root(&repo, &receipt).await; + assert!(matches!( + repo.inspect_gc( + &other, + claim.operation_id(), + claim.lifetime(), + MetadataGcPhase::Apply + ) + .await, + Err(MetadataGcError::CommitUncertain { .. }) + )); + let txn = second + .begin_with_config(Some(IsolationLevel::RepeatableRead), None) + .await + .unwrap(); + assert!( + txn.execute_unprepared("DELETE FROM mst2_metadata_payload") + .await + .is_err() + ); + txn.rollback().await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + repo.apply(&claim).await.unwrap(); +} + +#[tokio::test] +async fn qualified_null_bytes_and_retired_generic_incarnations_cannot_be_adopted() { + let (first, second, _schema) = fixture().await; + let pages = prepared(); + let payload = &pages.dag().payloads()[0]; + first.execute_raw(statement("INSERT INTO mst2_metadata_payload(page_id,metadata_codec,byte_size,payload) VALUES($1,1,$2,$3)", + [payload.id.to_vec().into(),(payload.size as i32).into(),payload.bytes.clone().into()])).await.unwrap(); + let qualified = PostgresQualifiedMetadataRepository::new(second.clone()) + .await + .unwrap(); + assert!( + qualified + .begin_intent("cannot-adopt-null", &pages) + .await + .is_err() + ); + assert!( + second + .execute_unprepared("DELETE FROM mst2_metadata_payload") + .await + .is_err() + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NULL" + ) + .await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_current").await, + 0 + ); + let (first, second, _second_schema) = fixture().await; + let generic = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let intent = generic + .begin_intent("generic-history", &pages) + .await + .unwrap(); + generic + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = generic.finalize(&intent).await.unwrap(); + generic.retire_prepare_coverage(&receipt).await.unwrap(); + let qualified = PostgresQualifiedMetadataRepository::new(second) + .await + .unwrap(); + assert!( + qualified + .begin_intent("cannot-adopt-retired-generic", &pages) + .await + .is_err() + ); + assert!(qualified.lifetime(payload.id).await.is_err()); +} + +#[tokio::test] +async fn qualified_committed_partial_payload_is_rejected_without_repair() { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repo + .begin_intent("corrupt-partial-committed", &pages) + .await + .unwrap(); + repo.install_page(&intent, &pages.dag().payloads()[0]) + .await + .unwrap(); + second.execute_raw(statement("UPDATE mst2_metadata_prepare SET state='COMMITTED',committed_at=clock_timestamp() WHERE prepare_id=$1", + [intent.prepare_id().into()])).await.unwrap(); + assert!( + repo.install_pages(&intent, pages.dag().payloads()) + .await + .is_err() + ); + let lifetime = repo.lifetime(pages.dag().root()).await.unwrap(); + assert!( + repo.claim(&uuid::Uuid::new_v4().to_string(), &lifetime) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); +} + +#[tokio::test] +async fn qualified_cross_action_terminal_inspection_conflicts_and_production_adoption_stays_closed() +{ + let (first, second, _schema) = fixture().await; + let (repo, pages, receipt) = installed(&first, "cross-action-committed").await; + assert!(matches!( + repo.inspect_terminal(&second, receipt.intent(), MetadataTerminalAction::Abort) + .await, + Err(MetadataTerminalError::Rejected(SnapshotError { + code: SnapshotErrorCode::Conflict, + .. + })) + )); + let pending = repo + .begin_intent("cross-action-aborted", &pages) + .await + .unwrap(); + repo.abort(&pending).await.unwrap(); + assert!(matches!( + repo.inspect_terminal(&second, &pending, MetadataTerminalAction::RetireCoverage) + .await, + Err(MetadataTerminalError::Rejected(SnapshotError { + code: SnapshotErrorCode::Conflict, + .. + })) + )); + assert!(repo.abort(receipt.intent()).await.is_err()); + let sql="INSERT INTO mst2_snapshot_context(snapshot_id,canonical_descriptor,instance_id,commit_oid,root_tree_oid, + metadata_root,prepare_id,publication_sequence,writer_epoch,state) VALUES('blocked',$1,'i','c','r',$2,$3,0,1,'READY')"; + assert!( + second + .execute_raw(statement( + sql, + [ + vec![1_u8].into(), + receipt.metadata_root().to_vec().into(), + receipt.intent().prepare_id().into() + ] + )) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_snapshot_context").await, + 0 + ); + assert_eq!( + repo.inspect_terminal( + &second, + receipt.intent(), + MetadataTerminalAction::RetireCoverage + ) + .await + .unwrap(), + MetadataTerminalObservation::Active + ); +} + +#[tokio::test] +async fn qualified_deferred_current_guard_denies_unprotected_raw_fresh_commit() { + let (first, second, _schema) = fixture().await; + let (repo, _pages, receipt) = installed(&first, "raw-fresh").await; + repo.retire_prepare_coverage(&receipt).await.unwrap(); + let claim = claim_root(&repo, &receipt).await; + repo.apply(&claim).await.unwrap(); + let txn = second.begin().await.unwrap(); + txn.execute_raw(statement("INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size,graph_domain) VALUES($1,$2,2,'RESERVED',1,$3,'qualified-v1')", + [claim.lifetime.page_id.to_vec().into(),node_id(&claim.lifetime.page_id).into(),claim.lifetime.expected_size.into()])).await.unwrap(); + txn.execute_raw(statement( + "UPDATE mst2_metadata_current SET generation=2 WHERE page_id=$1", + [claim.lifetime.page_id.to_vec().into()], + )) + .await + .unwrap(); + assert!(txn.commit().await.is_err()); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_current WHERE generation=2" + ) + .await, + 0 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE generation=2" + ) + .await, + 0 + ); +} + +#[tokio::test] +async fn qualified_statement_dag_audit_accepts_wide_and_chain_graphs_and_rejects_a_cycle() { + for wide in [true, false] { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first.clone()) + .await + .unwrap(); + let mut builder = MetadataDagBuilder::new(MetadataDagLimits::default()); + let mut root_entries = Vec::new(); + let mut last = [0_u8; 32]; + for i in 0..64_u8 { + let entries = if wide || i == 0 { + vec![Entry::file(EntryKind::Regular, b"file", 3, [i; 32])] + } else { + vec![Entry::dir(b"child", last)] + }; + let page = Page::build(&entries).unwrap(); + builder.add_directory(&page, &entries).unwrap(); + last = page_id(&page); + root_entries.push(Entry::dir(format!("d{i:03}").as_bytes(), last)); + } + let root = if wide { + let root = Page::build(&root_entries).unwrap(); + builder.add_directory(&root, &root_entries).unwrap(); + page_id(&root) + } else { + last + }; + let pages = PreparedNativeMetadataRetention::test_installation( + Arc::new(builder.finish(root).unwrap()), + "/", + ); + let intent = repo.begin_intent("statement-dag", &pages).await.unwrap(); + repo.install_pages(&intent, &pages.dag().payloads()[..64]) + .await + .unwrap(); + if wide { + repo.install_page(&intent, &pages.dag().payloads()[64]) + .await + .unwrap(); + } + let receipt = repo.finalize(&intent).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_node").await, + if wide { 65 } else { 64 } + ); + assert_eq!( + scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_metadata_graph_node" + ) + .await, + if wide { 64 } else { 63 } + ); + let pending = repo.begin_intent("cycle-attempt", &pages).await.unwrap(); + let leaf = pages + .dag() + .payloads() + .iter() + .find(|p| { + !pages + .dag() + .edges() + .iter() + .any(|e| e.parent == node_id(&p.id)) + }) + .unwrap() + .id; + let error=second.execute_raw(statement( + "INSERT INTO mst2_metadata_graph_edge(parent_page,parent_generation,child_page,child_generation) VALUES($1,1,$2,1)", + [leaf.to_vec().into(),root.to_vec().into()], + )).await.unwrap_err(); + assert!(error.to_string().contains("cycle"), "{error}"); + assert_eq!( + scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_metadata_graph_node" + ) + .await, + if wide { 64 } else { 63 } + ); + repo.abort(&pending).await.unwrap(); + repo.retire_prepare_coverage(&receipt).await.unwrap(); + } +} + +#[tokio::test] +async fn qualified_statement_dag_overflow_rolls_back_without_disabling_production_guards() { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repo.begin_intent("overflow-proof", &pages).await.unwrap(); + assert_eq!(MetadataDagLimits::default().nodes, 4096); + assert_eq!(MetadataDagLimits::default().edges, 16384); + let txn = second.begin().await.unwrap(); + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size,graph_domain) + SELECT sha256(convert_to('bound-node-'||i,'UTF8')),'page:sha256:'||encode(sha256(convert_to('bound-node-'||i,'UTF8')),'hex'), + 1,'RESERVED',1,$1,'qualified-v1' FROM generate_series(1,4097) x(i)", + [(pages.dag().payloads()[0].size as i32).into()], + )).await.unwrap(); + txn.execute_unprepared("INSERT INTO mst2_metadata_current(page_id,generation) SELECT l.page_id,l.generation FROM mst2_metadata_lifetime l WHERE NOT EXISTS(SELECT 1 FROM mst2_metadata_current c WHERE c.page_id=l.page_id)").await.unwrap(); + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_prepare_page(prepare_id,page_id,generation,expected_size) + SELECT $1,l.page_id,l.generation,l.expected_size FROM mst2_metadata_lifetime l + WHERE NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare_page m WHERE m.prepare_id=$1 AND m.page_id=l.page_id)", + [intent.prepare_id().into()], + )).await.unwrap(); + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_graph_node(page_id,generation,state,metadata_codec,bytes) + SELECT page_id,generation,'LIVE',1,expected_size FROM mst2_metadata_prepare_page WHERE prepare_id=$1", + [intent.prepare_id().into()], + )).await.unwrap(); + let error=txn.execute_unprepared( + "INSERT INTO mst2_metadata_graph_edge(parent_page,parent_generation,child_page,child_generation) + SELECT p.page_id,1,c.page_id,1 FROM (SELECT page_id FROM mst2_metadata_graph_node ORDER BY page_id LIMIT 1) p + CROSS JOIN mst2_metadata_graph_node c WHERE c.page_id<>p.page_id" + ).await.unwrap_err(); + assert!(error.to_string().contains("overflow"), "{error}"); + txn.rollback().await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_graph_node").await, + 0 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_current").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_prepare_page").await, + 2 + ); + let captured = repo.scan_current_lifetimes(None, 1).await.unwrap(); + assert_eq!(captured.len(), 1); + let rest = repo + .scan_current_lifetimes(Some((captured[0].page_id(), captured[0].generation())), 1) + .await + .unwrap(); + assert_eq!(rest.len(), 1); + assert_ne!(rest[0].page_id(), captured[0].page_id()); +} + +#[tokio::test] +async fn qualified_aborted_intent_is_recovered_from_history_after_repository_restart() { + let (first, second, _schema) = fixture().await; + let repo = PostgresQualifiedMetadataRepository::new(first) + .await + .unwrap(); + let pages = prepared(); + let intent = repo.begin_intent("restart-aborted", &pages).await.unwrap(); + let terminal = repo.abort(&intent).await.unwrap(); + drop(repo); + let restarted = PostgresQualifiedMetadataRepository::new(second.clone()) + .await + .unwrap(); + let recovered = restarted + .capture_terminal_intent("restart-aborted", intent.manifest_digest()) + .await + .unwrap(); + assert_eq!( + restarted + .inspect_terminal(&second, &recovered, MetadataTerminalAction::Abort) + .await + .unwrap(), + MetadataTerminalObservation::Terminated(Box::new(terminal)) + ); + assert!( + restarted + .install_pages(&recovered, pages.dag().payloads()) + .await + .is_err() + ); + assert_eq!( + scalar(&second, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); +}