feat(mst2): persist fixed metadata generations for native preparation

src/api/router/snapshot_content_tests.rs

@@ -82,6 +82,9 @@ const TOKEN: &str = "mst2-fixed-content-test";
 #[path = "snapshot_session_tests.rs"]
 mod durable_sessions;
 
+#[path = "snapshot_generation_upgrade_tests.rs"]
+mod generation_upgrade;
+
 #[derive(Default)]
 struct ReadCounts {
     whole: AtomicUsize,

src/api/router/snapshot_generation_upgrade_tests.rs

@@ -0,0 +1,178 @@
+use sea_orm_migration::MigratorTrait;
+
+use super::*;
+use crate::jupiter::migration::Migrator;
+
+async fn scalar(db: &sea_orm::DatabaseConnection, sql: &str) -> i64 {
+    db.query_one_raw(sea_orm::Statement::from_string(
+        sea_orm::DbBackend::Postgres,
+        sql,
+    ))
+    .await
+    .unwrap()
+    .unwrap()
+    .try_get_by_index(0)
+    .unwrap()
+}
+
+#[tokio::test]
+async fn mst2_generation_additive_upgrade_preserves_legacy_v3_sid_lease_and_new_resolve() {
+    let fixture = Fixture::new_with_pg_config(true).await;
+    let mono = fixture.state.storage.mono_storage();
+    let db = mono.get_connection();
+    assert_eq!(
+        scalar(db, "SELECT count(*) FROM mst2_metadata_lifetime").await,
+        0
+    );
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_metadata_prepare WHERE storage_seal IS NOT NULL"
+        )
+        .await,
+        0
+    );
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NOT NULL"
+        )
+        .await,
+        0
+    );
+    let pages = scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await;
+    let original = success_json(fixture.send("GET", "descriptor", Body::empty()).await).await;
+    // The unchanged v3 installer created these durable rows. Remove only the
+    // empty additive schema in this isolated fixture to reproduce the previous
+    // deployed schema; the SID, lease, CAS bytes and all original guards survive.
+    db.execute_unprepared(
+        "DROP TRIGGER mst2_metadata_generation_mapping_guard ON mst2_metadata_prepare_page;
+         DROP TRIGGER mst2_metadata_generation_seal_guard ON mst2_metadata_prepare;
+         DROP FUNCTION mst2_metadata_generation_mapping_guard();
+         DROP FUNCTION mst2_metadata_generation_seal_guard();
+         ALTER TABLE mst2_metadata_prepare DROP COLUMN canonical_bindings,DROP COLUMN bindings_digest,
+           DROP COLUMN primary_scope,DROP COLUMN storage_seal;
+         ALTER TABLE mst2_metadata_prepare_page DROP COLUMN generation;
+         ALTER TABLE mst2_metadata_payload DROP COLUMN generation;
+         DROP TABLE mst2_metadata_lifetime;
+         DROP FUNCTION mst2_metadata_lifetime_guard();
+         DELETE FROM seaql_migrations WHERE version='m20261007_000200_add_mst2_metadata_generations'"
+    ).await.unwrap();
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_snapshot_lease WHERE state='ACTIVE'"
+        )
+        .await,
+        1
+    );
+    assert_eq!(
+        scalar(db, "SELECT count(*) FROM mst2_snapshot_context").await,
+        1
+    );
+    assert_eq!(
+        success_json(fixture.send("GET", "descriptor", Body::empty()).await).await,
+        original
+    );
+    Migrator::up(db, None).await.unwrap();
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NULL"
+        )
+        .await,
+        pages
+    );
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_metadata_prepare WHERE storage_seal IS NULL"
+        )
+        .await,
+        1
+    );
+    assert_eq!(
+        success_json(fixture.send("GET", "descriptor", Body::empty()).await).await,
+        original
+    );
+
+    let config = fixture.state.storage.config();
+    let connection = crate::jupiter::storage::init::postgres_connection(&config.database)
+        .await
+        .unwrap();
+    let storage = crate::jupiter::storage::Storage::new_with_connection(
+        config,
+        Arc::new(connection),
+        fixture.state.storage.git_service.obj_storage.clone(),
+    )
+    .await
+    .unwrap();
+    let state = MonoApiServiceState {
+        storage,
+        ..fixture.state.clone()
+    };
+    let app = Router::new().nest("/api/v2", routers(state.clone()).with_state(state));
+    let restored = success_json(
+        app.clone()
+            .oneshot(fixture.request("GET", "descriptor", Body::empty()))
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(restored, original);
+    let next = success_json(
+        app.clone()
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/api/v2/snapshots/resolve")
+                    .header("authorization", format!("Bearer {TOKEN}"))
+                    .header("content-type", "application/json")
+                    .body(Body::from(
+                        json!({"target":{"kind":"latest"},"scope":"/project"}).to_string(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(next["descriptor"]["snapshot_id"], fixture.snapshot);
+    assert_ne!(next["lease_id"], fixture.lease);
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_snapshot_lease WHERE state='ACTIVE'"
+        )
+        .await,
+        2
+    );
+    let content = app
+        .oneshot(fixture.request("GET", "blob?path=/file", Body::empty()))
+        .await
+        .unwrap();
+    assert_eq!(content.status(), 200);
+    assert_eq!(
+        to_bytes(content.into_body(), usize::MAX)
+            .await
+            .unwrap()
+            .as_ref(),
+        fixture.raw.as_slice()
+    );
+    assert_eq!(
+        scalar(db, "SELECT count(*) FROM mst2_metadata_lifetime").await,
+        0
+    );
+    assert_eq!(
+        scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await,
+        pages
+    );
+    assert_eq!(
+        scalar(
+            db,
+            "SELECT count(*) FROM mst2_metadata_prepare WHERE storage_seal IS NULL"
+        )
+        .await,
+        1
+    );
+}

src/callisto/mod.rs

@@ -66,6 +66,7 @@ pub mod mega_view_root_chain_scan;
 pub mod mega_webhook;
 pub mod mega_webhook_delivery;
 pub mod mega_webhook_event_type;
+pub mod mst2_metadata_lifetime;
 pub mod mst2_metadata_payload;
 pub mod mst2_metadata_prepare;
 pub mod mst2_metadata_prepare_page;

src/callisto/mst2_metadata_lifetime.rs

@@ -0,0 +1,17 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "mst2_metadata_lifetime")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub page_id: Vec<u8>,
+    pub node_id: String,
+    pub generation: i64,
+    pub state: String,
+    pub metadata_codec: i16,
+    pub expected_size: i32,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+impl ActiveModelBehavior for ActiveModel {}

src/callisto/mst2_metadata_payload.rs

@@ -5,6 +5,7 @@ use sea_orm::entity::prelude::*;
 pub struct Model {
     #[sea_orm(primary_key, auto_increment = false)]
     pub page_id: Vec<u8>,
+    pub generation: Option<i64>,
     pub metadata_codec: i16,
     pub byte_size: i32,
     pub payload: Vec<u8>,

src/callisto/mst2_metadata_prepare.rs

@@ -8,6 +8,10 @@ pub struct Model {
     pub operation_id: String,
     pub manifest_digest: Vec<u8>,
     pub canonical_plan: Vec<u8>,
+    pub canonical_bindings: Option<Vec<u8>>,
+    pub bindings_digest: Option<Vec<u8>>,
+    pub primary_scope: Option<Vec<u8>>,
+    pub storage_seal: Option<Vec<u8>>,
     pub source_domain: String,
     pub tagged_root_tree_oid: String,
     pub scope: String,

src/callisto/mst2_metadata_prepare_page.rs

@@ -8,6 +8,7 @@ pub struct Model {
     #[sea_orm(primary_key, auto_increment = false)]
     pub page_id: Vec<u8>,
     pub expected_size: i32,
+    pub generation: Option<i64>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

src/jupiter/migration/m20261007_000200_add_mst2_metadata_generations.rs

@@ -0,0 +1,91 @@
+//! Fixed metadata lifetimes. Legacy rows remain explicitly unbound.
+
+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> {
+        manager.get_connection().execute_unprepared(&format!(
+            "CREATE TABLE mst2_metadata_lifetime (
+               page_id bytea PRIMARY KEY CHECK (octet_length(page_id)=32),
+               node_id text NOT NULL UNIQUE CHECK (node_id='page:sha256:'||encode(page_id,'hex')),
+               generation bigint NOT NULL CHECK (generation>0),
+               state text NOT NULL CHECK (state IN ('RESERVED','LIVE','DELETING','REMOVED')),
+               metadata_codec smallint NOT NULL CHECK (metadata_codec=1),
+               expected_size integer NOT NULL CHECK (expected_size BETWEEN {HEADER_LEN} AND {PAGE_MAX_BYTES}),
+               UNIQUE(page_id,generation)
+             );
+             CREATE INDEX idx_mst2_metadata_lifetime_state_page ON mst2_metadata_lifetime(state,page_id);
+             CREATE 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_lifetime_guard BEFORE UPDATE OR DELETE
+               ON mst2_metadata_lifetime FOR EACH ROW EXECUTE FUNCTION mst2_metadata_lifetime_guard();
+             ALTER TABLE mst2_metadata_payload ADD COLUMN generation bigint CHECK (generation>0),
+               ADD CONSTRAINT mst2_metadata_payload_lifetime_fk FOREIGN KEY(page_id,generation)
+                 REFERENCES mst2_metadata_lifetime(page_id,generation);
+             ALTER TABLE mst2_metadata_prepare_page ADD COLUMN generation bigint CHECK (generation>0),
+               ADD CONSTRAINT mst2_metadata_prepare_page_lifetime_fk FOREIGN KEY(page_id,generation)
+                 REFERENCES mst2_metadata_lifetime(page_id,generation);
+             CREATE INDEX idx_mst2_metadata_prepare_page_lifetime
+               ON mst2_metadata_prepare_page(page_id,generation,prepare_id);
+             ALTER TABLE mst2_metadata_prepare
+               ADD COLUMN canonical_bindings bytea,
+               ADD COLUMN bindings_digest bytea CHECK (octet_length(bindings_digest)=32),
+               ADD COLUMN primary_scope bytea CHECK (octet_length(primary_scope) BETWEEN 1 AND 16384),
+               ADD COLUMN storage_seal bytea CHECK (octet_length(storage_seal)=32),
+               ADD CONSTRAINT mst2_metadata_prepare_generation_seal_check CHECK (
+                 (canonical_bindings IS NULL AND bindings_digest IS NULL AND primary_scope IS NULL AND storage_seal IS NULL)
+                 OR (canonical_bindings IS NOT NULL AND octet_length(canonical_bindings) BETWEEN 60 AND 196620
+                   AND bindings_digest IS NOT NULL AND primary_scope IS NOT NULL AND storage_seal IS NOT NULL)
+               );
+             CREATE FUNCTION mst2_metadata_generation_seal_guard() RETURNS trigger LANGUAGE plpgsql AS $$
+             BEGIN
+               IF NEW.canonical_bindings IS DISTINCT FROM OLD.canonical_bindings
+                 OR NEW.bindings_digest IS DISTINCT FROM OLD.bindings_digest
+                 OR NEW.primary_scope IS DISTINCT FROM OLD.primary_scope
+                 OR NEW.storage_seal IS DISTINCT FROM OLD.storage_seal THEN
+                 RAISE EXCEPTION 'metadata generation seal cannot be rebound';
+               END IF;
+               RETURN NEW;
+             END $$;
+             CREATE TRIGGER mst2_metadata_generation_seal_guard BEFORE UPDATE ON mst2_metadata_prepare
+               FOR EACH ROW EXECUTE FUNCTION mst2_metadata_generation_seal_guard();
+             CREATE FUNCTION mst2_metadata_generation_mapping_guard() RETURNS trigger LANGUAGE plpgsql AS $$
+             DECLARE fixed boolean; preparing boolean;
+             BEGIN
+               SELECT storage_seal IS NOT NULL,state='PREPARING' INTO fixed,preparing
+                 FROM mst2_metadata_prepare WHERE prepare_id=CASE WHEN TG_OP='DELETE' THEN OLD.prepare_id ELSE NEW.prepare_id END FOR UPDATE;
+               IF fixed AND (TG_OP<>'INSERT' OR NOT preparing) THEN
+                 RAISE EXCEPTION 'metadata generation mappings are immutable';
+               END IF;
+               IF TG_OP='UPDATE' AND OLD.prepare_id IS DISTINCT FROM NEW.prepare_id AND EXISTS (
+                 SELECT 1 FROM mst2_metadata_prepare WHERE prepare_id=OLD.prepare_id AND storage_seal IS NOT NULL
+               ) THEN RAISE EXCEPTION 'metadata generation mapping cannot leave its prepare'; END IF;
+               IF TG_OP='DELETE' THEN RETURN OLD; END IF;
+               RETURN NEW;
+             END $$;
+             CREATE TRIGGER mst2_metadata_generation_mapping_guard BEFORE INSERT OR UPDATE OR DELETE
+               ON mst2_metadata_prepare_page FOR EACH ROW EXECUTE FUNCTION mst2_metadata_generation_mapping_guard()"
+        )).await.map(|_| ())
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        Ok(())
+    }
+}

src/jupiter/migration/mod.rs

@@ -146,6 +146,7 @@ mod m20261005_000200_harden_mst2_retention_graph;
 mod m20261005_000300_add_mst2_metadata_install;
 mod m20261006_000100_add_view_tables;
 mod m20261007_000100_add_mst2_snapshot_sessions;
+mod m20261007_000200_add_mst2_metadata_generations;
 mod runner;
 pub use m20260905_000100_add_push_queue::ensure_queue_control_seed;
 pub use runner::apply_migrations;
@@ -280,6 +281,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20261005_000300_add_mst2_metadata_install::Migration),
             Box::new(m20261006_000100_add_view_tables::Migration),
             Box::new(m20261007_000100_add_mst2_snapshot_sessions::Migration),
+            Box::new(m20261007_000200_add_mst2_metadata_generations::Migration),
         ]
     }
 }
@@ -1190,7 +1192,7 @@ mod tests {
     async fn import_repo_alias_rows_canonicalized() {
         let names = migration_names();
         assert_eq!(
-            &names[names.len() - 9..],
+            &names[names.len() - 10..names.len() - 1],
             &[
                 "m20260923_000200_canonicalize_import_repo_paths".to_string(),
                 "m20260925_000100_media_paging".to_string(),
@@ -1204,6 +1206,10 @@ mod tests {
             ],
             "native retention, metadata installation and view tables follow media paging"
         );
+        assert_eq!(
+            names.last().unwrap(),
+            "m20261007_000200_add_mst2_metadata_generations"
+        );
 
         let db = alias_db().await;
         insert_repo(&db, 1, "/third-party//a").await;