From 73d4a05aadb9cbf697db8554663610534c2193d2 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 7 Oct 2026 21:52:59 +0800 Subject: [PATCH] perf(mst2): validate and seal fixed metadata installation once Mint an opaque capability from complete immutable preparation evidence and freeze its durable identity and membership. PREPARING batches use bounded exact-member checks while preserving current primary, scope, state, physical lifetime and payload validation. COMMITTED replay and finalize retain their complete oracle. Native validation and measured performance remain pending. --- src/api/router/snapshot_content_tests.rs | 3 + .../snapshot_generation_upgrade_tests.rs | 1 + .../snapshot_install_capability_fixture.rs | 33 + ...1007_000500_add_mst2_install_capability.rs | 21 + .../m20261007_000500_install_capability.sql | 138 ++++ src/jupiter/migration/mod.rs | 14 +- .../storage/native_metadata_install.rs | 8 + .../native_metadata_install_capability.rs | 464 +++++++++++ ...metadata_install_capability_fault_tests.rs | 217 +++++ ...ative_metadata_install_capability_tests.rs | 782 ++++++++++++++++++ .../storage/native_metadata_install_tests.rs | 3 + .../storage/native_snapshot_session.rs | 6 +- 12 files changed, 1685 insertions(+), 5 deletions(-) create mode 100644 src/api/router/snapshot_install_capability_fixture.rs create mode 100644 src/jupiter/migration/m20261007_000500_add_mst2_install_capability.rs create mode 100644 src/jupiter/migration/m20261007_000500_install_capability.sql create mode 100644 src/jupiter/storage/native_metadata_install_capability.rs create mode 100644 src/jupiter/storage/native_metadata_install_capability_fault_tests.rs create mode 100644 src/jupiter/storage/native_metadata_install_capability_tests.rs diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index 79bc47ef..5399ad42 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -91,6 +91,9 @@ mod generation_history_fixture; #[path = "snapshot_generation_qualified_fixture.rs"] mod generation_qualified_fixture; +#[path = "snapshot_install_capability_fixture.rs"] +mod install_capability_fixture; + #[derive(Default)] struct ReadCounts { whole: AtomicUsize, diff --git a/src/api/router/snapshot_generation_upgrade_tests.rs b/src/api/router/snapshot_generation_upgrade_tests.rs index 9f14f96d..08684866 100644 --- a/src/api/router/snapshot_generation_upgrade_tests.rs +++ b/src/api/router/snapshot_generation_upgrade_tests.rs @@ -42,6 +42,7 @@ async fn mst2_generation_additive_upgrade_preserves_legacy_v3_sid_lease_and_new_ ); let pages = scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await; let original = success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + super::install_capability_fixture::restore_pre_capability_schema(db).await; super::generation_history_fixture::restore_empty_g1_schema(db).await; // The unchanged v3 installer created these durable rows. Remove only the // empty additive schema in this isolated fixture to reproduce the previous diff --git a/src/api/router/snapshot_install_capability_fixture.rs b/src/api/router/snapshot_install_capability_fixture.rs new file mode 100644 index 00000000..1806c7ab --- /dev/null +++ b/src/api/router/snapshot_install_capability_fixture.rs @@ -0,0 +1,33 @@ +use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; + +pub(super) async fn restore_pre_capability_schema(db: &DatabaseConnection) { + let bound:i64=db.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT (SELECT count(*) FROM mst2_metadata_prepare WHERE storage_seal IS NOT NULL + OR canonical_bindings IS NOT NULL OR bindings_digest IS NOT NULL OR primary_scope IS NOT NULL + OR graph_domain IS NOT NULL)+(SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation IS NOT NULL) + +(SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NOT NULL)")) + .await.unwrap().unwrap().try_get_by_index(0).unwrap(); + assert_eq!( + bound, 0, + "old-schema fixture only contains unbound legacy installations" + ); + // Reproduce the deployment before this additive migration. Preserve every + // original plan, member, byte, context, lease and protection root. + db.execute_unprepared( + "DROP TRIGGER mst2_00_install_capability_barrier ON mst2_metadata_prepare; + DROP TRIGGER mst2_00_install_capability_barrier ON mst2_metadata_prepare_page; + DROP TRIGGER mst2_install_capability_truncate_guard ON mst2_metadata_prepare; + DROP TRIGGER mst2_install_capability_truncate_guard ON mst2_metadata_prepare_page; + DROP TRIGGER mst2_install_capability_prepare_guard ON mst2_metadata_prepare; + DROP TRIGGER mst2_install_capability_mapping_guard ON mst2_metadata_prepare_page; + DROP TABLE mst2_metadata_install_seal; + DROP FUNCTION mst2_install_capability_barrier(); + DROP FUNCTION mst2_install_capability_truncate_guard(); + DROP FUNCTION mst2_install_capability_prepare_guard(); + DROP FUNCTION mst2_install_capability_mapping_guard(); + DROP FUNCTION mst2_install_capability_register(); + DELETE FROM seaql_migrations WHERE version='m20261007_000500_add_mst2_install_capability'", + ) + .await + .unwrap(); +} diff --git a/src/jupiter/migration/m20261007_000500_add_mst2_install_capability.rs b/src/jupiter/migration/m20261007_000500_add_mst2_install_capability.rs new file mode 100644 index 00000000..95ddaca5 --- /dev/null +++ b/src/jupiter/migration/m20261007_000500_add_mst2_install_capability.rs @@ -0,0 +1,21 @@ +//! Freeze a fully validated legacy plan without granting generation authority. + +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> { + manager + .get_connection() + .execute_unprepared(include_str!("m20261007_000500_install_capability.sql")) + .await?; + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} diff --git a/src/jupiter/migration/m20261007_000500_install_capability.sql b/src/jupiter/migration/m20261007_000500_install_capability.sql new file mode 100644 index 00000000..b8114adf --- /dev/null +++ b/src/jupiter/migration/m20261007_000500_install_capability.sql @@ -0,0 +1,138 @@ +LOCK TABLE mst2_metadata_prepare,mst2_metadata_prepare_page IN ACCESS EXCLUSIVE MODE; + +CREATE TABLE mst2_metadata_install_seal ( + prepare_id text PRIMARY KEY REFERENCES mst2_metadata_prepare(prepare_id), + operation_id text NOT NULL CHECK (octet_length(operation_id) BETWEEN 1 AND 255), + manifest_digest bytea NOT NULL CHECK (octet_length(manifest_digest)=32), + members_digest bytea NOT NULL CHECK (octet_length(members_digest)=32), + primary_scope bytea NOT NULL CHECK (octet_length(primary_scope) BETWEEN 1 AND 16384), + install_seal bytea NOT NULL CHECK (octet_length(install_seal)=32), + created_at timestamptz NOT NULL DEFAULT clock_timestamp() +); + +-- The actual regulated table, rather than caller search_path, selects the lock. +CREATE FUNCTION mst2_install_capability_barrier() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF pg_catalog.current_schema() IS DISTINCT FROM TG_TABLE_SCHEMA + OR pg_catalog.pg_is_in_recovery() + OR pg_catalog.current_setting('transaction_isolation')<>'read committed' THEN + RAISE EXCEPTION 'install capability mutation requires its actual primary schema and READ COMMITTED'; + END IF; + PERFORM pg_catalog.set_config('search_path',pg_catalog.quote_ident(TG_TABLE_SCHEMA)||',pg_catalog,pg_temp',true); + PERFORM pg_catalog.set_config('lock_timeout','5000ms',true); + PERFORM pg_catalog.pg_advisory_xact_lock(1296717362,pg_catalog.hashtext(TG_TABLE_SCHEMA)); + RETURN NULL; +END $$; + +CREATE FUNCTION mst2_install_capability_register() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE p record; members bytea; actual jsonb; member_count bigint; member_bytes bigint; has_root boolean; +BEGIN + IF TG_OP<>'INSERT' THEN RAISE EXCEPTION 'install capability registration is immutable'; END IF; + IF pg_catalog.current_schema() IS DISTINCT FROM TG_TABLE_SCHEMA THEN + RAISE EXCEPTION 'install capability registration is outside its actual schema'; + END IF; + EXECUTE pg_catalog.format('SELECT * FROM %I.mst2_metadata_prepare WHERE prepare_id=$1',TG_TABLE_SCHEMA) + INTO p USING NEW.prepare_id; + IF p.prepare_id IS NULL OR p.operation_id IS DISTINCT FROM NEW.operation_id + OR p.manifest_digest IS DISTINCT FROM NEW.manifest_digest + OR pg_catalog.sha256(p.canonical_plan) IS DISTINCT FROM NEW.manifest_digest + OR p.state NOT IN ('PREPARING','COMMITTED') OR p.coverage_retired_at IS NOT NULL + OR p.canonical_bindings IS NOT NULL OR p.bindings_digest IS NOT NULL + OR p.primary_scope IS NOT NULL OR p.storage_seal IS NOT NULL OR p.graph_domain IS NOT NULL THEN + RAISE EXCEPTION 'install capability requires a fixed unbound legacy preparation'; + END IF; + EXECUTE pg_catalog.format( + 'SELECT count(*),sum(expected_size),bool_or(page_id=$2), + sha256(string_agg(page_id||int4send(expected_size)|| + CASE WHEN generation IS NULL THEN decode(''00'',''hex'') + ELSE decode(''01'',''hex'')||int8send(generation) END,''''::bytea ORDER BY page_id)) + FROM %I.mst2_metadata_prepare_page WHERE prepare_id=$1',TG_TABLE_SCHEMA) + INTO member_count,member_bytes,has_root,members USING NEW.prepare_id,p.metadata_root; + IF member_count<>p.node_count OR member_bytes<>p.total_bytes OR has_root IS DISTINCT FROM true + OR members IS DISTINCT FROM NEW.members_digest THEN + RAISE EXCEPTION 'install capability complete membership proof disagrees'; + END IF; + EXECUTE pg_catalog.format( + 'SELECT EXISTS(SELECT 1 FROM %I.mst2_metadata_prepare_page WHERE prepare_id=$1 AND generation IS NOT NULL)', + TG_TABLE_SCHEMA) INTO has_root USING NEW.prepare_id; + IF has_root THEN RAISE EXCEPTION 'legacy install capability cannot adopt generation bindings'; END IF; + EXECUTE pg_catalog.format( + 'SELECT jsonb_build_array(s.storage_uuid,current_database(),d.oid::bigint,$1,n.oid::bigint, + inet_server_addr()::text,inet_server_port()) + FROM %I.mst2_metadata_storage_scope s + JOIN pg_catalog.pg_database d ON d.datname=current_database() + JOIN pg_catalog.pg_namespace n ON n.nspname=$1 WHERE s.singleton=1',TG_TABLE_SCHEMA) + INTO actual USING TG_TABLE_SCHEMA; + IF actual IS NULL OR pg_catalog.convert_from(NEW.primary_scope,'UTF8')::jsonb IS DISTINCT FROM actual + OR NEW.install_seal IS DISTINCT FROM pg_catalog.sha256( + pg_catalog.convert_to('MST2-LEGACY-INSTALL-CAPABILITY-1','UTF8')||pg_catalog.decode('00','hex')|| + pg_catalog.uuid_send(NEW.prepare_id::uuid)|| + pg_catalog.int4send(pg_catalog.octet_length(NEW.operation_id))||pg_catalog.convert_to(NEW.operation_id,'UTF8')|| + NEW.manifest_digest||NEW.members_digest|| + pg_catalog.int4send(pg_catalog.octet_length(NEW.primary_scope))||NEW.primary_scope) THEN + RAISE EXCEPTION 'install capability seal or actual primary scope disagrees'; + END IF; + RETURN NEW; +END $$; + +CREATE FUNCTION mst2_install_capability_prepare_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE registered boolean; +BEGIN + IF pg_catalog.current_schema() IS DISTINCT FROM TG_TABLE_SCHEMA THEN + RAISE EXCEPTION 'install capability preparation is outside its actual schema'; + END IF; + EXECUTE pg_catalog.format('SELECT EXISTS(SELECT 1 FROM %I.mst2_metadata_install_seal WHERE prepare_id=$1)',TG_TABLE_SCHEMA) + INTO registered USING OLD.prepare_id; + IF registered AND (TG_OP='DELETE' OR + (pg_catalog.to_jsonb(NEW)-ARRAY['state','committed_at','aborted_at','coverage_retired_at']) IS DISTINCT FROM + (pg_catalog.to_jsonb(OLD)-ARRAY['state','committed_at','aborted_at','coverage_retired_at'])) THEN + RAISE EXCEPTION 'registered metadata preparation identity is immutable'; + END IF; + IF TG_OP='DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END $$; + +CREATE FUNCTION mst2_install_capability_mapping_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE old_id text; new_id text; registered boolean; +BEGIN + IF pg_catalog.current_schema() IS DISTINCT FROM TG_TABLE_SCHEMA THEN + RAISE EXCEPTION 'install capability membership is outside its actual schema'; + END IF; + IF TG_OP<>'INSERT' THEN old_id:=OLD.prepare_id; END IF; + IF TG_OP<>'DELETE' THEN new_id:=NEW.prepare_id; END IF; + EXECUTE pg_catalog.format( + 'SELECT EXISTS(SELECT 1 FROM %I.mst2_metadata_install_seal WHERE prepare_id=$1 OR prepare_id=$2)',TG_TABLE_SCHEMA) + INTO registered USING old_id,new_id; + IF registered THEN RAISE EXCEPTION 'registered metadata membership is immutable'; END IF; + IF TG_OP='DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END $$; + +CREATE FUNCTION mst2_install_capability_truncate_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +DECLARE registered boolean; +BEGIN + IF pg_catalog.current_schema() IS DISTINCT FROM TG_TABLE_SCHEMA + OR pg_catalog.pg_is_in_recovery() + OR pg_catalog.current_setting('transaction_isolation')<>'read committed' THEN + RAISE EXCEPTION 'install capability truncation requires its actual primary schema and READ COMMITTED'; + END IF; + PERFORM pg_catalog.set_config('lock_timeout','5000ms',true); + PERFORM pg_catalog.pg_advisory_xact_lock(1296717362,pg_catalog.hashtext(TG_TABLE_SCHEMA)); + EXECUTE pg_catalog.format('SELECT EXISTS(SELECT 1 FROM %I.mst2_metadata_install_seal)',TG_TABLE_SCHEMA) + INTO registered; + IF registered THEN RAISE EXCEPTION 'registered metadata evidence cannot be truncated'; END IF; + RETURN NULL; +END $$; + +DO $$ DECLARE t text; BEGIN + FOREACH t IN ARRAY ARRAY['mst2_metadata_prepare','mst2_metadata_prepare_page','mst2_metadata_install_seal'] LOOP + EXECUTE pg_catalog.format('CREATE TRIGGER mst2_00_install_capability_barrier BEFORE INSERT OR UPDATE OR DELETE ON %I FOR EACH STATEMENT EXECUTE FUNCTION mst2_install_capability_barrier()',t); + EXECUTE pg_catalog.format('CREATE TRIGGER mst2_install_capability_truncate_guard BEFORE TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION mst2_install_capability_truncate_guard()',t); + END LOOP; +END $$; +CREATE TRIGGER mst2_install_capability_register BEFORE INSERT OR UPDATE OR DELETE + ON mst2_metadata_install_seal FOR EACH ROW EXECUTE FUNCTION mst2_install_capability_register(); +CREATE TRIGGER mst2_install_capability_prepare_guard BEFORE UPDATE OR DELETE + ON mst2_metadata_prepare FOR EACH ROW EXECUTE FUNCTION mst2_install_capability_prepare_guard(); +CREATE TRIGGER mst2_install_capability_mapping_guard BEFORE INSERT OR UPDATE OR DELETE + ON mst2_metadata_prepare_page FOR EACH ROW EXECUTE FUNCTION mst2_install_capability_mapping_guard(); diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 384af69a..fa611fce 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -149,6 +149,7 @@ 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 m20261007_000500_add_mst2_install_capability; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -286,6 +287,7 @@ impl MigratorTrait for Migrator { 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), + Box::new(m20261007_000500_add_mst2_install_capability::Migration), ] } } @@ -1196,7 +1198,7 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 12..names.len() - 3], + &names[names.len() - 13..names.len() - 4], &[ "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), @@ -1211,17 +1213,21 @@ mod tests { "native retention, metadata installation and view tables follow media paging" ); assert_eq!( - &names[names.len() - 3], + &names[names.len() - 4], "m20261007_000200_add_mst2_metadata_generations" ); assert_eq!( - &names[names.len() - 2], + &names[names.len() - 3], "m20261007_000300_add_mst2_metadata_lifetime_history" ); assert_eq!( - names.last().unwrap(), + &names[names.len() - 2], "m20261007_000400_add_mst2_qualified_metadata_gc" ); + assert_eq!( + names.last().unwrap(), + "m20261007_000500_add_mst2_install_capability" + ); let db = alias_db().await; insert_repo(&db, 1, "/third-party//a").await; diff --git a/src/jupiter/storage/native_metadata_install.rs b/src/jupiter/storage/native_metadata_install.rs index d4be4cad..735dcc54 100644 --- a/src/jupiter/storage/native_metadata_install.rs +++ b/src/jupiter/storage/native_metadata_install.rs @@ -1050,6 +1050,14 @@ fn unavailable(message: &str) -> SnapshotError { SnapshotError::new(SnapshotErrorCode::ObjectUnavailable, message) } +#[path = "native_metadata_install_capability.rs"] +mod capability; +pub(crate) use capability::ValidatedLegacyInstallCapability; + +#[cfg(test)] +#[path = "native_metadata_install_capability_tests.rs"] +mod capability_tests; + // Additive repository only. HTTP sessions continue using the existing installer // until session anchors and lease adoption bind the generation seal. #[path = "native_metadata_generations.rs"] diff --git a/src/jupiter/storage/native_metadata_install_capability.rs b/src/jupiter/storage/native_metadata_install_capability.rs new file mode 100644 index 00000000..348405e1 --- /dev/null +++ b/src/jupiter/storage/native_metadata_install_capability.rs @@ -0,0 +1,464 @@ +//! Reuse pure plan validation only after its durable membership is frozen. + +use sha2::{Digest, Sha256}; + +use super::{ + BTreeMap, BTreeSet, ConnectionTrait, DatabaseTransaction, MetadataCommitPhase, + MetadataInstallError, MetadataInstallIdentity, MetadataPagePayload, MetadataPrepareIntent, + PostgresMetadataInstallRepository, PrimaryStorageScope, QueryResult, SnapshotError, + SnapshotErrorCode, check_payload_coverage, commit, integrity, internal, json, + load_installed_dag, require_plan, statement, unavailable, validate_payload, verify_graph, +}; + +const SEAL_DOMAIN: &[u8] = b"MST2-LEGACY-INSTALL-CAPABILITY-1\0"; + +#[derive(Debug)] +pub(crate) struct ValidatedLegacyInstallCapability { + intent: MetadataPrepareIntent, + identity: MetadataInstallIdentity, + root: [u8; 32], + edge_count: usize, + total_bytes: u64, + members: BTreeMap<[u8; 32], (i32, Option)>, + members_digest: [u8; 32], + scope: PrimaryStorageScope, + primary_scope: Vec, + install_seal: [u8; 32], +} + +impl PostgresMetadataInstallRepository { + pub(crate) async fn mint_legacy_install_capability( + &self, + intent: &MetadataPrepareIntent, + ) -> Result { + let txn = self.transaction().await?; + let result = async { + self.capability_barrier(&txn).await?; + let stored = require_plan(&txn, intent).await?; + let record = &stored.record; + if record.storage_seal.is_some() + || record.canonical_bindings.is_some() + || record.bindings_digest.is_some() + || record.primary_scope.is_some() + || record.graph_domain.is_some() + || record.coverage_retired_at.is_some() + || record.aborted_at.is_some() + { + return Err(unavailable("install capability requires an active unbound legacy preparation")); + } + let mut members = BTreeMap::new(); + for member in &stored.prepare_pages { + if member.generation.is_some() { + return Err(integrity("legacy install capability cannot adopt generation bindings")); + } + let id = member.page_id.as_slice().try_into().map_err(internal)?; + members.insert(id, (member.expected_size, member.generation)); + } + if record.state == "COMMITTED" { + load_installed_dag(&txn, &stored).await?; + check_payload_coverage(&txn, &stored).await?; + verify_graph(&txn, &stored).await?; + } + let members_digest = member_digest(&members); + let primary_scope = scope_bytes(&self.storage_scope)?; + let install_seal = seal(intent, &members_digest, &primary_scope)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_install_seal(prepare_id,operation_id,manifest_digest,members_digest,primary_scope,install_seal) + VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(prepare_id) DO NOTHING", + [intent.prepare_id.clone().into(),intent.operation_id.clone().into(), + intent.manifest_digest.to_vec().into(),members_digest.to_vec().into(), + primary_scope.clone().into(),install_seal.to_vec().into()], + )).await.map_err(internal)?; + let capability = ValidatedLegacyInstallCapability { + intent: intent.clone(), identity: stored.plan.identity, root: stored.plan.root, + edge_count: stored.plan.edges.len(), total_bytes: stored.plan.total_bytes, + members,members_digest,scope:self.storage_scope.clone(),primary_scope,install_seal, + }; + read_registered_prepare(&txn, &capability).await?; + Ok(capability) + }.await; + commit( + txn, + result, + &intent.operation_id, + intent.manifest_digest, + MetadataCommitPhase::Intent, + ) + .await + } + + pub(crate) async fn install_pages_validated( + &self, + capability: &ValidatedLegacyInstallCapability, + payloads: &[MetadataPagePayload], + ) -> Result<(), MetadataInstallError> { + if capability.scope != self.storage_scope { + return Err( + integrity("install capability belongs to another captured primary scope").into(), + ); + } + if payloads.is_empty() || payloads.len() > 64 { + return Err(SnapshotError::new( + SnapshotErrorCode::LimitExceeded, + "metadata installation batch must contain 1..=64 pages", + ) + .into()); + } + let mut ids = BTreeSet::new(); + for payload in payloads { + validate_payload(payload)?; + if !ids.insert(payload.id) { + return Err(integrity("duplicate page in metadata installation batch").into()); + } + if capability + .members + .get(&payload.id) + .map(|member| member.0 as u64) + != Some(payload.size) + { + return Err(integrity( + "metadata payload is not a member of its validated installation", + ) + .into()); + } + } + let pages: Vec<_> = payloads + .iter() + .map(|p| { + json!({ + "page_id":hex::encode(p.id),"size":p.size,"payload":hex::encode(&p.bytes) + }) + }) + .collect(); + let encoded = serde_json::to_string(&pages).map_err(internal)?; + let txn = self.transaction().await?; + let result = async { + self.capability_barrier(&txn).await?; + let state = read_registered_prepare(&txn,capability).await?; + check_requested_members(&txn,capability,&encoded,payloads.len()).await?; + if state == "COMMITTED" { + // Committed replay retains the complete receipt oracle and never repairs bytes. + let stored = require_plan(&txn,&capability.intent).await?; + load_installed_dag(&txn,&stored).await?; + check_payload_coverage(&txn,&stored).await?; + verify_graph(&txn,&stored).await?; + } else { + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_payload(page_id,metadata_codec,byte_size,payload) + SELECT decode(p.page_id,'hex'),$1,p.size,decode(p.payload,'hex') + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer,payload text) + ON CONFLICT(page_id) DO NOTHING", + [(capability.identity.metadata_codec as i16).into(),encoded.clone().into()], + )).await.map_err(internal)?; + } + let bad = txn.query_one_raw(statement( + "SELECT p.page_id FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer,payload text) + LEFT JOIN mst2_metadata_payload b ON b.page_id=decode(p.page_id,'hex') + WHERE b.page_id IS NULL OR b.metadata_codec<>$1 OR b.byte_size<>p.size + OR b.payload<>decode(p.payload,'hex') LIMIT 1", + [(capability.identity.metadata_codec as i16).into(),encoded.into()], + )).await.map_err(internal)?; + if bad.is_some() { + return Err(integrity("immutable metadata payload identity conflicts with stored bytes")); + } + Ok(()) + }.await; + commit( + txn, + result, + &capability.intent.operation_id, + capability.intent.manifest_digest, + MetadataCommitPhase::Payload, + ) + .await + } + + pub(super) async fn capability_barrier( + &self, + txn: &DatabaseTransaction, + ) -> Result<(), SnapshotError> { + let schema = txn + .query_one_raw(statement( + "SELECT pg_catalog.current_schema() AS schema", + [], + )) + .await + .map_err(internal)? + .ok_or_else(|| internal("primary schema is missing"))? + .try_get::("", "schema") + .map_err(internal)?; + if schema != self.storage_scope.schema { + return Err(integrity( + "install capability transaction is outside its captured schema", + )); + } + txn.execute_raw(statement( + "SELECT pg_catalog.set_config('search_path',pg_catalog.quote_ident($1)||',pg_catalog,pg_temp',true)", + [schema.into()], + )).await.map_err(internal)?; + self.barrier(txn).await + } +} + +async fn read_registered_prepare( + txn: &DatabaseTransaction, + capability: &ValidatedLegacyInstallCapability, +) -> Result { + let row=txn.query_one_raw(statement( + "SELECT p.prepare_id,p.operation_id,p.manifest_digest,p.source_domain,p.tagged_root_tree_oid,p.scope, + p.schema_version,p.metadata_codec,p.materialization_policy,p.fs_semantics,p.access_projection, + p.verification_revision,p.projection_revision,p.metadata_root,p.node_count,p.edge_count,p.total_bytes, + p.state,p.committed_at IS NOT NULL AS committed,p.aborted_at IS NOT NULL AS aborted, + p.coverage_retired_at IS NOT NULL AS retired, + p.storage_seal IS NOT NULL OR p.canonical_bindings IS NOT NULL OR p.bindings_digest IS NOT NULL + OR p.primary_scope IS NOT NULL OR p.graph_domain IS NOT NULL AS generation_bound, + s.operation_id AS sealed_operation,s.manifest_digest AS sealed_manifest,s.members_digest, + s.primary_scope AS sealed_scope,s.install_seal + FROM mst2_metadata_prepare p JOIN mst2_metadata_install_seal s USING(prepare_id) + WHERE p.prepare_id=$1", + [capability.intent.prepare_id.clone().into()], + )).await.map_err(internal)?.ok_or_else(||unavailable("validated install registration is missing"))?; + let identity = &capability.identity; + for (column, expected) in [ + ("prepare_id", &capability.intent.prepare_id), + ("operation_id", &capability.intent.operation_id), + ("sealed_operation", &capability.intent.operation_id), + ("source_domain", &identity.source_domain), + ("tagged_root_tree_oid", &identity.tagged_root_tree_oid), + ("scope", &identity.scope), + ] { + if row.try_get::("", column).map_err(internal)? != *expected { + return Err(integrity( + "validated install identity differs from durable registration", + )); + } + } + for (column, expected) in [ + ("schema_version", identity.schema_version), + ("metadata_codec", identity.metadata_codec), + ("materialization_policy", identity.materialization_policy), + ("fs_semantics", identity.fs_semantics), + ("access_projection", identity.access_projection), + ("projection_revision", identity.projection_revision), + ] { + if row.try_get::("", column).map_err(internal)? != expected as i16 { + return Err(integrity( + "validated install profile differs from durable registration", + )); + } + } + for (column, expected) in [ + ( + "manifest_digest", + capability.intent.manifest_digest.as_slice(), + ), + ( + "sealed_manifest", + capability.intent.manifest_digest.as_slice(), + ), + ("metadata_root", capability.root.as_slice()), + ("members_digest", capability.members_digest.as_slice()), + ("sealed_scope", capability.primary_scope.as_slice()), + ("install_seal", capability.install_seal.as_slice()), + ] { + if row + .try_get::>("", column) + .map_err(internal)? + .as_slice() + != expected + { + return Err(integrity( + "validated install proof differs from durable registration", + )); + } + } + let state = row.try_get::("", "state").map_err(internal)?; + if row + .try_get::("", "verification_revision") + .map_err(internal)? + != identity.verification_revision + || row.try_get::("", "node_count").map_err(internal)? as usize + != capability.members.len() + || row.try_get::("", "edge_count").map_err(internal)? as usize != capability.edge_count + || row.try_get::("", "total_bytes").map_err(internal)? as u64 != capability.total_bytes + || !["PREPARING", "COMMITTED"].contains(&state.as_str()) + || row.try_get::("", "committed").map_err(internal)? != (state == "COMMITTED") + || row.try_get::("", "aborted").map_err(internal)? + || row.try_get::("", "retired").map_err(internal)? + || row + .try_get::("", "generation_bound") + .map_err(internal)? + { + return Err(unavailable( + "validated install preparation is no longer active with its fixed profile", + )); + } + Ok(state) +} + +async fn check_requested_members( + txn: &DatabaseTransaction, + capability: &ValidatedLegacyInstallCapability, + encoded: &str, + expected_count: usize, +) -> Result<(), SnapshotError> { + let rows=txn.query_all_raw(statement( + "SELECT decode(p.page_id,'hex') AS page_id,m.expected_size,m.generation AS member_generation, + b.page_id IS NOT NULL AS payload_present,b.generation AS payload_generation, + c.generation AS current_generation,l.generation AS lifetime_generation,l.graph_domain, + l.state AS lifetime_state,l.metadata_codec AS lifetime_codec,l.expected_size AS lifetime_size, + n.state AS graph_state,n.kind AS graph_kind,n.bytes AS graph_bytes, + EXISTS(SELECT 1 FROM mst2_retention_gc_op g WHERE g.node_id='page:sha256:'||p.page_id + AND g.operation='REMOVE' AND g.state IN ('PENDING','APPLIED')) AS tombstone + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer,payload text) + LEFT JOIN mst2_metadata_prepare_page m ON m.prepare_id=$1 AND m.page_id=decode(p.page_id,'hex') + LEFT JOIN mst2_metadata_payload b ON b.page_id=m.page_id + LEFT JOIN mst2_metadata_current c ON c.page_id=m.page_id + LEFT JOIN mst2_metadata_lifetime l ON l.page_id=c.page_id AND l.generation=c.generation + LEFT JOIN mst2_retention_node n ON n.node_id='page:sha256:'||p.page_id", + [capability.intent.prepare_id.clone().into(),encoded.into()], + )).await.map_err(internal)?; + if rows.len() != expected_count { + return Err(integrity("requested install membership 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 size = row + .try_get::>("", "expected_size") + .map_err(internal)?; + let generation = row + .try_get::>("", "member_generation") + .map_err(internal)?; + if capability.members.get(&page).copied() != size.map(|size| (size, generation)) { + return Err(integrity( + "requested member differs from validated installation", + )); + } + check_physical_member( + &row, + capability.identity.metadata_codec as i16, + size.ok_or_else(|| integrity("requested member is missing"))?, + )?; + } + Ok(()) +} + +fn check_physical_member(row: &QueryResult, codec: i16, size: i32) -> Result<(), SnapshotError> { + let current = row + .try_get::>("", "current_generation") + .map_err(internal)?; + let lifetime = row + .try_get::>("", "lifetime_generation") + .map_err(internal)?; + let payload = row + .try_get::>("", "payload_generation") + .map_err(internal)?; + let graph = row + .try_get::>("", "graph_state") + .map_err(internal)?; + if row.try_get::("", "tombstone").map_err(internal)? + || graph.as_deref().is_some_and(|state| state != "LIVE") + || row + .try_get::>("", "graph_kind") + .map_err(internal)? + .is_some_and(|kind| kind != "page") + || row + .try_get::>("", "graph_bytes") + .map_err(internal)? + .is_some_and(|bytes| bytes != size as i64) + { + return Err(unavailable( + "requested generic metadata graph is unavailable", + )); + } + if payload.is_some() && (payload != current || payload != lifetime) { + return Err(integrity( + "generic payload is not its exact current physical lifetime", + )); + } + if current.is_some() { + let state = row + .try_get::>("", "lifetime_state") + .map_err(internal)?; + if current != lifetime + || current.is_some_and(|generation| generation <= 0) + || row + .try_get::>("", "graph_domain") + .map_err(internal)? + .as_deref() + != Some("generic-v1") + || !matches!(state.as_deref(), Some("RESERVED" | "LIVE")) + || row + .try_get::>("", "lifetime_codec") + .map_err(internal)? + != Some(codec) + || row + .try_get::>("", "lifetime_size") + .map_err(internal)? + != Some(size) + || (state.as_deref() == Some("LIVE") + && (graph.is_none() + || payload != current + || !row + .try_get::("", "payload_present") + .map_err(internal)?)) + { + return Err(unavailable( + "requested physical lifetime cannot be shared by legacy generic installation", + )); + } + } + Ok(()) +} + +fn member_digest(members: &BTreeMap<[u8; 32], (i32, Option)>) -> [u8; 32] { + let mut hash = Sha256::new(); + for (id, (size, generation)) in members { + hash.update(id); + hash.update(size.to_be_bytes()); + match generation { + None => hash.update([0]), + Some(generation) => { + hash.update([1]); + hash.update(generation.to_be_bytes()); + } + } + } + hash.finalize().into() +} + +fn scope_bytes(scope: &PrimaryStorageScope) -> Result, SnapshotError> { + let bytes = serde_json::to_vec(&( + &scope.storage_uuid, + &scope.database, + scope.database_oid, + &scope.schema, + scope.schema_oid, + &scope.server_address, + scope.server_port, + )) + .map_err(internal)?; + if bytes.is_empty() || bytes.len() > 16384 { + return Err(integrity("invalid primary install scope size")); + } + Ok(bytes) +} + +fn seal( + intent: &MetadataPrepareIntent, + members: &[u8; 32], + scope: &[u8], +) -> Result<[u8; 32], SnapshotError> { + let id = uuid::Uuid::parse_str(&intent.prepare_id).map_err(internal)?; + let mut hash = Sha256::new(); + hash.update(SEAL_DOMAIN); + hash.update(id.as_bytes()); + hash.update((intent.operation_id.len() as u32).to_be_bytes()); + hash.update(intent.operation_id.as_bytes()); + hash.update(intent.manifest_digest); + hash.update(members); + hash.update((scope.len() as u32).to_be_bytes()); + hash.update(scope); + Ok(hash.finalize().into()) +} diff --git a/src/jupiter/storage/native_metadata_install_capability_fault_tests.rs b/src/jupiter/storage/native_metadata_install_capability_fault_tests.rs new file mode 100644 index 00000000..9903085e --- /dev/null +++ b/src/jupiter/storage/native_metadata_install_capability_fault_tests.rs @@ -0,0 +1,217 @@ +use super::*; + +async fn count(db: &DatabaseConnection, sql: &str) -> i64 { + db.query_one_raw(statement(sql, [])) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +async fn mint_fault(fault: Fault) { + let (direct, recovery, _schema, url) = fixture().await; + let proxy = PgCommitFaultProxy::start(&url, fault).await; + let mut options = sea_orm::ConnectOptions::new(proxy.url.clone()); + options.max_connections(1).min_connections(1); + let repository = + PostgresMetadataInstallRepository::new(Database::connect(options).await.unwrap()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("cap-mint-fault", &pages) + .await + .unwrap(); + proxy.armed.store(true, Ordering::SeqCst); + let error = tokio::time::timeout( + Duration::from_secs(15), + repository.mint_legacy_install_capability(&intent), + ) + .await + .unwrap() + .unwrap_err(); + assert!(matches!( + error, + MetadataInstallError::CommitUncertain { + phase: MetadataCommitPhase::Intent, + .. + } + )); + proxy.wait_for_fault().await; + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_install_seal").await, + if matches!(fault, Fault::AfterCommit) { + 1 + } else { + 0 + } + ); + assert_eq!( + proxy.commit_observed.load(Ordering::SeqCst), + matches!(fault, Fault::AfterCommit) + ); + let restarted = PostgresMetadataInstallRepository::new(recovery.clone()) + .await + .unwrap(); + let cap = restarted + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 1 + ); + restarted + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .unwrap(); + restarted.finalize(&intent).await.unwrap(); +} + +async fn batch_fault(fault: Fault) { + let (direct, recovery, _schema, url) = fixture().await; + let proxy = PgCommitFaultProxy::start(&url, fault).await; + let mut options = sea_orm::ConnectOptions::new(proxy.url.clone()); + options.max_connections(1).min_connections(1); + let repository = + PostgresMetadataInstallRepository::new(Database::connect(options).await.unwrap()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("cap-batch-fault", &pages) + .await + .unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + proxy.armed.store(true, Ordering::SeqCst); + let error = tokio::time::timeout( + Duration::from_secs(15), + repository.install_pages_validated(&cap, pages.dag().payloads()), + ) + .await + .unwrap() + .unwrap_err(); + assert!(matches!( + error, + MetadataInstallError::CommitUncertain { + phase: MetadataCommitPhase::Payload, + .. + } + )); + proxy.wait_for_fault().await; + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 1 + ); + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_payload").await, + if matches!(fault, Fault::AfterCommit) { + pages.dag().payloads().len() as i64 + } else { + 0 + } + ); + let restarted = PostgresMetadataInstallRepository::new(recovery.clone()) + .await + .unwrap(); + let fresh = restarted + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + restarted + .install_pages_validated(&fresh, pages.dag().payloads()) + .await + .unwrap(); + let receipt = restarted.finalize(&intent).await.unwrap(); + assert_eq!(receipt.metadata_root(), pages.dag().root()); + assert_eq!( + count( + &direct, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + pages.dag().payloads().len() as i64 + ); + restarted + .install_pages_validated(&fresh, pages.dag().payloads()) + .await + .unwrap(); + assert_eq!(restarted.finalize(&intent).await.unwrap(), receipt); +} + +#[tokio::test] +async fn install_capability_real_registration_commit_response_loss_recovers_fresh_proof() { + mint_fault(Fault::AfterCommit).await; +} + +#[tokio::test] +async fn install_capability_real_registration_rollback_leaves_no_freeze_or_capability() { + mint_fault(Fault::BeforeCommit).await; +} + +#[tokio::test] +async fn install_capability_real_batch_commit_response_loss_preserves_exact_bytes() { + batch_fault(Fault::AfterCommit).await; +} + +#[tokio::test] +async fn install_capability_real_batch_connection_loss_rolls_back_without_false_success() { + batch_fault(Fault::BeforeCommit).await; +} + +#[tokio::test] +async fn install_capability_failed_registration_trigger_rolls_back_freeze_atomically() { + let (direct, recovery, _schema, _url) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(direct.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("cap-register-rollback", &pages) + .await + .unwrap(); + direct.execute_unprepared("CREATE FUNCTION test_cap_registration_fault() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'registration fault after production guard'; END $$; + CREATE TRIGGER test_cap_registration_fault AFTER INSERT ON mst2_metadata_install_seal FOR EACH ROW + EXECUTE FUNCTION test_cap_registration_fault()").await.unwrap(); + assert!( + repository + .mint_legacy_install_capability(&intent) + .await + .is_err() + ); + assert_eq!( + count(&direct, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 0 + ); + recovery + .execute_unprepared("UPDATE mst2_metadata_prepare_page SET expected_size=expected_size") + .await + .unwrap(); + direct + .execute_unprepared( + "DROP TRIGGER test_cap_registration_fault ON mst2_metadata_install_seal; + DROP FUNCTION test_cap_registration_fault()", + ) + .await + .unwrap(); + let restarted = PostgresMetadataInstallRepository::new(recovery) + .await + .unwrap(); + let cap = restarted + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + restarted + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .unwrap(); +} diff --git a/src/jupiter/storage/native_metadata_install_capability_tests.rs b/src/jupiter/storage/native_metadata_install_capability_tests.rs new file mode 100644 index 00000000..fb9df08f --- /dev/null +++ b/src/jupiter/storage/native_metadata_install_capability_tests.rs @@ -0,0 +1,782 @@ +use std::sync::Arc; + +use mst2_codec::metapage::{Entry, EntryKind}; +use sea_orm::{Database, PaginatorTrait}; +use sea_orm_migration::MigratorTrait; + +use super::*; +use crate::{ + ceres::snapshot::retention_dag::MetadataDagBuilder, + jupiter::{ + migration::Migrator, + tests::{TestSchemaGuard, test_db_config}, + }, +}; + +fn prepared(children: usize) -> PreparedNativeMetadataRetention { + let mut builder = MetadataDagBuilder::new(MetadataDagLimits::default()); + let mut entries = Vec::new(); + for index in 0..children { + let file = [Entry::file( + EntryKind::Regular, + b"file", + 3, + [index as u8; 32], + )]; + let child = Page::build(&file).unwrap(); + builder.add_directory(&child, &file).unwrap(); + entries.push(Entry::dir( + format!("dir-{index:03}").as_bytes(), + page_id(&child), + )); + } + let root = Page::build(&entries).unwrap(); + builder.add_directory(&root, &entries).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 install( + repository: &PostgresMetadataInstallRepository, + capability: &ValidatedLegacyInstallCapability, + prepared: &PreparedNativeMetadataRetention, +) { + for batch in prepared.dag().payloads().chunks(64) { + repository + .install_pages_validated(capability, batch) + .await + .unwrap(); + } +} + +async fn assert_guard(db: &DatabaseConnection, sql: &str) { + let error = db.execute_unprepared(sql).await.unwrap_err(); + assert!( + error.to_string().contains("install capability") + || error.to_string().contains("registered metadata"), + "{error}" + ); +} + +#[tokio::test] +async fn install_capability_many_batches_concurrent_mint_and_fresh_repository_finalize() { + let (first, second, _schema) = fixture().await; + let a = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let b = PostgresMetadataInstallRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared(130); + assert_eq!(pages.dag().payloads().len(), 131); + let intent = a.begin_intent("many", &pages).await.unwrap(); + let (left, right) = tokio::join!( + a.mint_legacy_install_capability(&intent), + b.mint_legacy_install_capability(&intent) + ); + let left = left.unwrap(); + let right = right.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 1 + ); + for (index, batch) in pages.dag().payloads().chunks(64).enumerate() { + if index == 0 { + a.install_pages_validated(&left, batch).await.unwrap(); + } else { + b.install_pages_validated(&right, batch).await.unwrap(); + } + } + drop(left); + drop(right); + drop(a); + drop(b); + let restarted = PostgresMetadataInstallRepository::new(second.clone()) + .await + .unwrap(); + let fresh = restarted + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + install(&restarted, &fresh, &pages).await; + let receipt = restarted.finalize(&intent).await.unwrap(); + assert_eq!(receipt.metadata_root(), pages.dag().root()); + assert_eq!(receipt.payload_bytes(), pages.dag().payload_bytes()); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NULL" + ) + .await, + 131 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_retention_node WHERE state='LIVE'" + ) + .await, + 131 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_edge").await, + 130 + ); + let before = scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_retention_node", + ) + .await; + install(&restarted, &fresh, &pages).await; + assert_eq!(restarted.finalize(&intent).await.unwrap(), receipt); + assert_eq!( + scalar( + &first, + "SELECT sum(incoming_refs)::bigint FROM mst2_retention_node" + ) + .await, + before + ); +} + +#[tokio::test] +async fn install_capability_freezes_every_identity_member_anchor_and_truncate_path() { + let (first, _second, _schema) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(2); + let intent = repository.begin_intent("freeze", &pages).await.unwrap(); + let other = repository + .begin_intent("unregistered", &pages) + .await + .unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + let before = mst2_metadata_prepare::Entity::find_by_id(intent.prepare_id().to_owned()) + .one(&first) + .await + .unwrap() + .unwrap(); + for mutation in [ + "operation_id='changed'", + "manifest_digest=decode(repeat('01',32),'hex')", + "canonical_plan=canonical_plan||decode('ff','hex')", + "source_domain='other'", + "tagged_root_tree_oid='sha1:bad'", + "scope='/changed'", + "schema_version=3", + "metadata_codec=2", + "materialization_policy=2", + "fs_semantics=2", + "access_projection=2", + "verification_revision=2", + "projection_revision=2", + "metadata_root=decode(repeat('01',32),'hex')", + "node_count=node_count+1", + "edge_count=edge_count+1", + "total_bytes=total_bytes+1", + "created_at=created_at+interval '1 second'", + ] { + assert_guard( + &first, + &format!( + "UPDATE mst2_metadata_prepare SET {mutation} WHERE prepare_id='{}'", + intent.prepare_id() + ), + ) + .await; + } + for sql in [ + format!( + "DELETE FROM mst2_metadata_prepare WHERE prepare_id='{}'", + intent.prepare_id() + ), + format!( + "UPDATE mst2_metadata_prepare_page SET expected_size=expected_size+1 WHERE prepare_id='{}'", + intent.prepare_id() + ), + format!( + "UPDATE mst2_metadata_prepare_page SET generation=1 WHERE prepare_id='{}'", + intent.prepare_id() + ), + format!( + "UPDATE mst2_metadata_prepare_page SET prepare_id='{}' WHERE prepare_id='{}'", + other.prepare_id(), + intent.prepare_id() + ), + format!( + "UPDATE mst2_metadata_prepare_page SET prepare_id='{}' WHERE prepare_id='{}'", + intent.prepare_id(), + other.prepare_id() + ), + format!( + "DELETE FROM mst2_metadata_prepare_page WHERE prepare_id='{}'", + intent.prepare_id() + ), + format!( + "INSERT INTO mst2_metadata_prepare_page(prepare_id,page_id,expected_size) VALUES('{}',decode(repeat('ff',32),'hex'),64)", + intent.prepare_id() + ), + "UPDATE mst2_metadata_install_seal SET members_digest=decode(repeat('ff',32),'hex')".into(), + "DELETE FROM mst2_metadata_install_seal".into(), + "TRUNCATE mst2_metadata_install_seal".into(), + "TRUNCATE mst2_metadata_prepare_page".into(), + "TRUNCATE mst2_metadata_prepare CASCADE".into(), + ] { + assert_guard(&first, &sql).await; + } + assert_eq!( + mst2_metadata_prepare::Entity::find_by_id(intent.prepare_id().to_owned()) + .one(&first) + .await + .unwrap() + .unwrap(), + before + ); + assert_eq!( + mst2_metadata_prepare_page::Entity::find() + .filter(mst2_metadata_prepare_page::Column::PrepareId.eq(intent.prepare_id())) + .count(&first) + .await + .unwrap(), + 3 + ); + install(&repository, &cap, &pages).await; + repository.finalize(&intent).await.unwrap(); +} + +#[tokio::test] +async fn install_capability_unregistered_corruption_and_seal_conflict_cannot_mint() { + let (first, _second, _schema) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(2); + let intent = repository.begin_intent("corrupt", &pages).await.unwrap(); + first + .execute_unprepared("UPDATE mst2_metadata_prepare SET projection_revision=2") + .await + .unwrap(); + assert!( + repository + .mint_legacy_install_capability(&intent) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 0 + ); + first.execute_unprepared("UPDATE mst2_metadata_prepare SET projection_revision=1,canonical_plan=canonical_plan||decode('ff','hex')").await.unwrap(); + assert!( + repository + .mint_legacy_install_capability(&intent) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 0 + ); + let mapping = repository + .begin_intent("mapping-corrupt", &pages) + .await + .unwrap(); + first.execute_raw(statement( + "UPDATE mst2_metadata_prepare_page SET expected_size=expected_size+1 WHERE prepare_id=$1", + [mapping.prepare_id().into()], + )).await.unwrap(); + assert!( + repository + .mint_legacy_install_capability(&mapping) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 0 + ); + first.execute_raw(statement( + "UPDATE mst2_metadata_prepare_page SET expected_size=expected_size-1 WHERE prepare_id=$1", + [mapping.prepare_id().into()], + )).await.unwrap(); + let intent = repository.begin_intent("conflict", &pages).await.unwrap(); + let sql=format!("INSERT INTO mst2_metadata_install_seal(prepare_id,operation_id,manifest_digest,members_digest,primary_scope,install_seal) + SELECT prepare_id,operation_id,manifest_digest,decode(repeat('00',32),'hex'),convert_to('[]','UTF8'),decode(repeat('00',32),'hex') + FROM mst2_metadata_prepare WHERE prepare_id='{}'",intent.prepare_id()); + assert_guard(&first, &sql).await; + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + install(&repository, &cap, &pages).await; +} + +#[tokio::test] +async fn install_capability_bounded_invalid_batches_are_atomic() { + let (first, _second, _schema) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(70); + let intent = repository.begin_intent("invalid", &pages).await.unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + let payload = &pages.dag().payloads()[0]; + assert!(repository.install_pages_validated(&cap, &[]).await.is_err()); + assert!( + repository + .install_pages_validated(&cap, &pages.dag().payloads()[..65]) + .await + .is_err() + ); + assert!( + repository + .install_pages_validated(&cap, &[payload.clone(), payload.clone()]) + .await + .is_err() + ); + let mut wrong = payload.clone(); + wrong.size += 1; + assert!( + repository + .install_pages_validated(&cap, &[payload.clone(), wrong]) + .await + .is_err() + ); + let mut wrong = payload.clone(); + wrong.id = [255; 32]; + assert!( + repository + .install_pages_validated(&cap, &[wrong]) + .await + .is_err() + ); + let mut wrong = payload.clone(); + wrong.bytes = vec![0; PAGE_MAX_BYTES + 1]; + wrong.size = wrong.bytes.len() as u64; + wrong.id = page_id(&wrong.bytes); + assert!( + repository + .install_pages_validated(&cap, &[wrong]) + .await + .is_err() + ); + let mut wrong = payload.clone(); + wrong.bytes = vec![0; HEADER_LEN]; + wrong.size = wrong.bytes.len() as u64; + wrong.id = page_id(&wrong.bytes); + assert!( + repository + .install_pages_validated(&cap, &[wrong]) + .await + .is_err() + ); + let stranger = prepared(71); + let stranger = stranger + .dag() + .payloads() + .iter() + .find(|p| { + !pages + .dag() + .payloads() + .iter() + .any(|member| member.id == p.id) + }) + .unwrap(); + assert!( + repository + .install_pages_validated(&cap, std::slice::from_ref(stranger)) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + let mut wrong = payload.bytes.clone(); + let last = wrong.len() - 1; + wrong[last] ^= 1; + 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(),wrong.into()])).await.unwrap(); + assert!( + repository + .install_pages_validated(&cap, &pages.dag().payloads()[..64]) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); +} + +#[tokio::test] +async fn install_capability_raw_committed_missing_bytes_does_not_repair_and_retirement_rejects() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(2); + let intent = repository.begin_intent("promoted", &pages).await.unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .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!( + repository + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + assert!( + repository + .mint_legacy_install_capability(&intent) + .await + .is_err() + ); + let intent = repository.begin_intent("retired", &pages).await.unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + install(&repository, &cap, &pages).await; + repository.finalize(&intent).await.unwrap(); + second.execute_raw(statement("UPDATE mst2_metadata_prepare SET coverage_retired_at=clock_timestamp() WHERE prepare_id=$1", + [intent.prepare_id().into()])).await.unwrap(); + assert!( + repository + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .is_err() + ); + assert!( + repository + .mint_legacy_install_capability(&intent) + .await + .is_err() + ); +} + +#[tokio::test] +async fn install_capability_cross_primary_schema_repeatable_read_and_temp_spoof_refuse() { + let (first, second, _schema) = fixture().await; + let (alien, _unused, _alien_schema) = fixture().await; + let a = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let b = PostgresMetadataInstallRepository::new(alien.clone()) + .await + .unwrap(); + let pages = prepared(2); + let intent = a.begin_intent("scope", &pages).await.unwrap(); + let cap = a.mint_legacy_install_capability(&intent).await.unwrap(); + let alien_intent = b.begin_intent("scope", &pages).await.unwrap(); + assert_ne!(alien_intent.prepare_id(), intent.prepare_id()); + let copied_prepare = first.query_one_raw(statement( + "SELECT row_to_json(p)::text AS copied FROM mst2_metadata_prepare p WHERE prepare_id=$1", + [intent.prepare_id().into()], + )).await.unwrap().unwrap().try_get::("", "copied").unwrap(); + let copied_members = first.query_one_raw(statement( + "SELECT json_agg(m)::text AS copied FROM mst2_metadata_prepare_page m WHERE prepare_id=$1", + [intent.prepare_id().into()], + )).await.unwrap().unwrap().try_get::("", "copied").unwrap(); + alien + .execute_raw(statement( + "DELETE FROM mst2_metadata_prepare_page WHERE prepare_id=$1", + [alien_intent.prepare_id().into()], + )) + .await + .unwrap(); + alien + .execute_raw(statement( + "DELETE FROM mst2_metadata_prepare WHERE prepare_id=$1", + [alien_intent.prepare_id().into()], + )) + .await + .unwrap(); + alien.execute_raw(statement("INSERT INTO mst2_metadata_prepare SELECT * FROM json_populate_record(NULL::mst2_metadata_prepare,$1::json)", + [copied_prepare.into()])).await.unwrap(); + alien.execute_raw(statement("INSERT INTO mst2_metadata_prepare_page SELECT * FROM json_populate_recordset(NULL::mst2_metadata_prepare_page,$1::json)", + [copied_members.into()])).await.unwrap(); + let own_scope_cap = b.mint_legacy_install_capability(&intent).await.unwrap(); + assert!( + b.install_pages_validated(&cap, pages.dag().payloads()) + .await + .is_err() + ); + assert_eq!( + scalar(&alien, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + install(&b, &own_scope_cap, &pages).await; + let txn = second + .begin_with_config(Some(IsolationLevel::RepeatableRead), None) + .await + .unwrap(); + assert!(a.capability_barrier(&txn).await.is_err()); + assert!( + txn.execute_unprepared("UPDATE mst2_metadata_prepare SET scope='/rr'") + .await + .is_err() + ); + txn.rollback().await.unwrap(); + let txn = second.begin().await.unwrap(); + let schema = a.storage_scope.schema.clone(); + txn.execute_unprepared("SET LOCAL search_path=pg_catalog") + .await + .unwrap(); + assert!(a.capability_barrier(&txn).await.is_err()); + let sql = format!( + "UPDATE \"{}\".mst2_metadata_prepare SET scope='/wrongpath'", + schema.replace('"', "\"\"") + ); + assert!(txn.execute_unprepared(&sql).await.is_err()); + txn.rollback().await.unwrap(); + let txn = second.begin().await.unwrap(); + txn.execute_unprepared( + "CREATE TEMP TABLE mst2_metadata_install_seal(prepare_id text); + CREATE TEMP TABLE mst2_metadata_storage_scope(singleton integer,storage_uuid text)", + ) + .await + .unwrap(); + let raw_sql = format!( + "UPDATE \"{}\".mst2_metadata_prepare SET scope='/raw-temp-spoof'", + schema.replace('"', "\"\"") + ); + assert!( + txn.execute_unprepared(&raw_sql) + .await + .unwrap_err() + .to_string() + .contains("registered metadata") + ); + txn.rollback().await.unwrap(); + let txn = second.begin().await.unwrap(); + txn.execute_unprepared( + "CREATE TEMP TABLE mst2_metadata_install_seal(prepare_id text); + CREATE TEMP TABLE mst2_metadata_storage_scope(singleton integer,storage_uuid text)", + ) + .await + .unwrap(); + a.capability_barrier(&txn).await.unwrap(); + assert_eq!( + scalar(&txn, "SELECT count(*) FROM mst2_metadata_install_seal").await, + 1 + ); + assert!( + txn.execute_unprepared("UPDATE mst2_metadata_prepare SET scope='/spoof'") + .await + .is_err() + ); + txn.rollback().await.unwrap(); + install(&a, &cap, &pages).await; +} + +#[tokio::test] +async fn install_capability_reuses_bound_generic_bytes_without_adopting_qualified_generation() { + let (first, _second, _schema) = fixture().await; + let generic = generations::PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let legacy = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(2); + let bound = generic.begin_intent("bound-generic", &pages).await.unwrap(); + generic + .install_pages(&bound, pages.dag().payloads()) + .await + .unwrap(); + generic.finalize(&bound).await.unwrap(); + let intent = legacy.begin_intent("legacy-share", &pages).await.unwrap(); + let cap = legacy + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + install(&legacy, &cap, &pages).await; + legacy.finalize(&intent).await.unwrap(); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NOT NULL" + ) + .await, + 3 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation IS NULL" + ) + .await, + 3 + ); + let bound_intent = MetadataPrepareIntent { + prepare_id: bound.prepare_id().into(), + operation_id: bound.operation_id().into(), + manifest_digest: bound.manifest_digest(), + }; + assert!( + legacy + .mint_legacy_install_capability(&bound_intent) + .await + .is_err() + ); + let (qualified_db, _other, _q_schema) = fixture().await; + let qualified = + generations::qualified::PostgresQualifiedMetadataRepository::new(qualified_db.clone()) + .await + .unwrap(); + let q_legacy = PostgresMetadataInstallRepository::new(qualified_db.clone()) + .await + .unwrap(); + let bound = qualified.begin_intent("qualified", &pages).await.unwrap(); + qualified + .install_pages(&bound, pages.dag().payloads()) + .await + .unwrap(); + let intent = q_legacy + .begin_intent("generic-refusal", &pages) + .await + .unwrap(); + let cap = q_legacy + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + assert!( + q_legacy + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .is_err() + ); + assert_eq!( + scalar(&qualified_db, "SELECT count(*) FROM mst2_metadata_payload").await, + 3 + ); + assert_eq!( + scalar(&qualified_db, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); +} + +#[tokio::test] +async fn install_capability_batch_fault_rolls_back_and_raw_writer_waits_before_row_lock() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(2); + let intent = repository + .begin_intent("batch-fault", &pages) + .await + .unwrap(); + let cap = repository + .mint_legacy_install_capability(&intent) + .await + .unwrap(); + let failed = hex::encode(pages.dag().payloads()[1].id); + first.execute_unprepared(&format!("CREATE FUNCTION test_install_batch_fault() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN IF NEW.page_id=decode('{failed}','hex') THEN RAISE EXCEPTION 'injected batch fault'; END IF; RETURN NEW; END $$; + CREATE TRIGGER test_install_batch_fault AFTER INSERT ON mst2_metadata_payload FOR EACH ROW EXECUTE FUNCTION test_install_batch_fault()")).await.unwrap(); + assert!( + repository + .install_pages_validated(&cap, pages.dag().payloads()) + .await + .is_err() + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + first.execute_unprepared("DROP TRIGGER test_install_batch_fault ON mst2_metadata_payload;DROP FUNCTION test_install_batch_fault()").await.unwrap(); + let held = first + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .unwrap(); + repository.capability_barrier(&held).await.unwrap(); + let worker = second + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .unwrap(); + let pid = scalar(&worker, "SELECT pg_backend_pid()::bigint").await; + let id = intent.prepare_id().to_owned(); + let blocked = tokio::spawn(async move { + let result = worker + .execute_raw(statement( + "UPDATE mst2_metadata_prepare SET scope='/blocked' WHERE prepare_id=$1", + [id.into()], + )) + .await; + worker.rollback().await.unwrap(); + result + }); + let deadline = tokio::time::Instant::now() + Duration::from_secs(4); + loop { + let waiting=held.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 writer must wait before frozen-row proof" + ); + tokio::task::yield_now().await; + } + let row_locked=held.query_one_raw(statement("SELECT EXISTS(SELECT 1 FROM pg_locks WHERE pid=$1 AND locktype='tuple' AND relation='mst2_metadata_prepare'::regclass) AS row_locked",[(pid as i32).into()])).await.unwrap().unwrap().try_get::("","row_locked").unwrap(); + assert!( + !row_locked, + "statement retention barrier precedes prepare row locking" + ); + held.query_one_raw(statement( + "SELECT prepare_id FROM mst2_metadata_prepare WHERE prepare_id=$1 FOR UPDATE", + [intent.prepare_id().into()], + )) + .await + .unwrap() + .unwrap(); + held.commit().await.unwrap(); + assert!( + blocked + .await + .unwrap() + .unwrap_err() + .to_string() + .contains("registered metadata") + ); + install(&repository, &cap, &pages).await; +} diff --git a/src/jupiter/storage/native_metadata_install_tests.rs b/src/jupiter/storage/native_metadata_install_tests.rs index 4b47fe38..897e3f50 100644 --- a/src/jupiter/storage/native_metadata_install_tests.rs +++ b/src/jupiter/storage/native_metadata_install_tests.rs @@ -1224,3 +1224,6 @@ async fn native_metadata_query_loss_after_recovery_barrier_preserves_typed_unkno MetadataPrepareObservation::Committed(_) )); } + +#[path = "native_metadata_install_capability_fault_tests.rs"] +mod capability_fault_tests; diff --git a/src/jupiter/storage/native_snapshot_session.rs b/src/jupiter/storage/native_snapshot_session.rs index 415228c8..d84d075a 100644 --- a/src/jupiter/storage/native_snapshot_session.rs +++ b/src/jupiter/storage/native_snapshot_session.rs @@ -69,9 +69,13 @@ impl PostgresNativeSessionRepository { .begin_intent(&format!("http:{}", built.snapshot_id), prepared) .await .map_err(install_error)?; + let capability = installer + .mint_legacy_install_capability(&intent) + .await + .map_err(install_error)?; for pages in prepared.dag().payloads().chunks(64) { installer - .install_pages(&intent, pages) + .install_pages_validated(&capability, pages) .await .map_err(install_error)?; }