perf(mst2): validate and seal fixed metadata installation once

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,

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

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();
+}

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(())
+    }
+}

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();

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;

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"]