diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index aa697589..3a7c8390 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/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, diff --git a/src/api/router/snapshot_generation_upgrade_tests.rs b/src/api/router/snapshot_generation_upgrade_tests.rs new file mode 100644 index 00000000..c2afda08 --- /dev/null +++ b/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 + ); +} diff --git a/src/callisto/mod.rs b/src/callisto/mod.rs index c8d90192..8eb43be5 100644 --- a/src/callisto/mod.rs +++ b/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; diff --git a/src/callisto/mst2_metadata_lifetime.rs b/src/callisto/mst2_metadata_lifetime.rs new file mode 100644 index 00000000..9807c27a --- /dev/null +++ b/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, + 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 {} diff --git a/src/callisto/mst2_metadata_payload.rs b/src/callisto/mst2_metadata_payload.rs index 798aab27..9ec1a65b 100644 --- a/src/callisto/mst2_metadata_payload.rs +++ b/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, + pub generation: Option, pub metadata_codec: i16, pub byte_size: i32, pub payload: Vec, diff --git a/src/callisto/mst2_metadata_prepare.rs b/src/callisto/mst2_metadata_prepare.rs index 9401282b..98b017a8 100644 --- a/src/callisto/mst2_metadata_prepare.rs +++ b/src/callisto/mst2_metadata_prepare.rs @@ -8,6 +8,10 @@ pub struct Model { pub operation_id: String, pub manifest_digest: Vec, pub canonical_plan: Vec, + pub canonical_bindings: Option>, + pub bindings_digest: Option>, + pub primary_scope: Option>, + pub storage_seal: Option>, pub source_domain: String, pub tagged_root_tree_oid: String, pub scope: String, diff --git a/src/callisto/mst2_metadata_prepare_page.rs b/src/callisto/mst2_metadata_prepare_page.rs index 61299f7e..7946022e 100644 --- a/src/callisto/mst2_metadata_prepare_page.rs +++ b/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, pub expected_size: i32, + pub generation: Option, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/src/jupiter/migration/m20261007_000200_add_mst2_metadata_generations.rs b/src/jupiter/migration/m20261007_000200_add_mst2_metadata_generations.rs new file mode 100644 index 00000000..2c098c19 --- /dev/null +++ b/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(()) + } +} diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index c102df51..6ceda285 100644 --- a/src/jupiter/migration/mod.rs +++ b/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; diff --git a/src/jupiter/storage/native_metadata_generation_tests.rs b/src/jupiter/storage/native_metadata_generation_tests.rs new file mode 100644 index 00000000..b565c091 --- /dev/null +++ b/src/jupiter/storage/native_metadata_generation_tests.rs @@ -0,0 +1,906 @@ +use std::sync::Arc; + +use mst2_codec::metapage::{Entry, EntryKind}; +use sea_orm::{Database, PaginatorTrait}; +use sea_orm_migration::MigratorTrait; + +use super::*; +use crate::{ + callisto::mst2_metadata_lifetime, + ceres::snapshot::retention_dag::MetadataDagBuilder, + jupiter::{ + migration::Migrator, + storage::mst2_retention::GcClaim, + tests::{TestSchemaGuard, test_db_config}, + }, +}; + +fn prepared(scope: &str) -> PreparedNativeMetadataRetention { + let entries = [Entry::file(EntryKind::Regular, b"file", 3, [42; 32])]; + let child = Page::build(&entries).unwrap(); + let root_entries = [ + Entry::dir(b"one", page_id(&child)), + Entry::dir(b"two", page_id(&child)), + ]; + let root = Page::build(&root_entries).unwrap(); + let mut builder = MetadataDagBuilder::new(MetadataDagLimits::default()); + builder.add_directory(&child, &entries).unwrap(); + builder.add_directory(&root, &root_entries).unwrap(); + PreparedNativeMetadataRetention::test_installation( + Arc::new(builder.finish(page_id(&root)).unwrap()), + scope, + ) +} + +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) +} + +fn rejected(error: MetadataInstallError) -> SnapshotErrorCode { + match error { + MetadataInstallError::Rejected(error) => error.code, + _ => panic!("expected definitive rejection"), + } +} + +async fn scalar(db: &DatabaseConnection, sql: &str) -> i64 { + db.query_one_raw(statement(sql, [])) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +async fn mappings(db: &DatabaseConnection, intent: &GenerationPrepareIntent) -> GenerationBindings { + let repository = PostgresMetadataGenerationRepository::new(db.clone()) + .await + .unwrap(); + repository + .require_fixed_plan(db, intent) + .await + .unwrap() + .bindings +} + +async fn wait_for_retention_waiter(txn: &DatabaseTransaction) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(4); + loop { + let waiting = txn.query_one_raw(statement( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND NOT granted + AND classid=$1::integer::oid AND objid=hashtext(current_schema())::oid + AND database=(SELECT oid FROM pg_database WHERE datname=current_database())) AS waiting", + [RETENTION_LOCK_KEY.into()], + )).await.unwrap().unwrap().try_get::("", "waiting").unwrap(); + if waiting { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "independent connection must wait on the schema retention barrier" + ); + tokio::task::yield_now().await; + } +} + +#[tokio::test] +async fn generation_begin_serializes_full_reserved_bindings_before_any_payload() { + let (first, second, _schema) = fixture().await; + let a = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let b = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let left = a.begin_intent("reserved-a", &pages).await.unwrap(); + let held = a.inner.transaction().await.unwrap(); + a.inner.barrier(&held).await.unwrap(); + let right_pages = prepared("/"); + let blocked = tokio::spawn(async move { b.begin_intent("reserved-b", &right_pages).await }); + wait_for_retention_waiter(&held).await; + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); + held.commit().await.unwrap(); + let right = blocked.await.unwrap().unwrap(); + assert_ne!(left.prepare_id(), right.prepare_id()); + assert_ne!(left.storage_seal, right.storage_seal); + assert_eq!(left.bindings_digest, right.bindings_digest); + assert_eq!( + mappings(&first, &left).await, + mappings(&second, &right).await + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE generation=1 AND state='RESERVED'" + ) + .await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation=1" + ) + .await, + 4 + ); + let replay = a.begin_intent("reserved-a", &pages).await.unwrap(); + assert_eq!(left, replay); +} + +#[tokio::test] +async fn generation_partial_prepare_recovers_exact_complete_mapping_on_second_connection() { + let (first, second, _schema) = fixture().await; + let a = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = a.begin_intent("partial", &pages).await.unwrap(); + a.install_pages(&intent, &pages.dag().payloads()[..1]) + .await + .unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + drop(a); + let restarted = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + assert_eq!( + restarted + .inspect_prepare( + &second, + "partial", + intent.manifest_digest(), + MetadataCommitPhase::Payload + ) + .await + .unwrap(), + GenerationPrepareObservation::Preparing(intent.clone()) + ); + assert_eq!( + restarted + .observe_installed_dag(&intent) + .await + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + restarted + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = restarted.finalize(&intent).await.unwrap(); + assert_eq!(receipt.intent(), &intent); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE state='LIVE' AND generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + 2 + ); +} + +#[tokio::test] +async fn generation_shared_live_prepare_pins_batch_before_generic_gc_can_claim() { + let (first, second, _schema) = fixture().await; + let a = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let b = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let old = a.begin_intent("live-old", &pages).await.unwrap(); + a.install_pages(&old, pages.dag().payloads()).await.unwrap(); + a.finalize(&old).await.unwrap(); + let shared = b.begin_intent("live-shared", &pages).await.unwrap(); + assert_eq!( + mappings(&first, &old).await, + mappings(&second, &shared).await + ); + let graph = PostgresRetentionRepository::new(first.clone()); + graph + .release_root(&RetentionRoot::Prepare(old.prepare_id().into())) + .await + .unwrap(); + assert_eq!( + graph + .mark_deleting("generic-cannot-claim-shared", &node_id(&pages.dag().root())) + .await + .unwrap(), + GcClaim::Unavailable + ); + assert_eq!( + scalar( + &second, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + 2 + ); + b.install_pages(&shared, pages.dag().payloads()) + .await + .unwrap(); + b.finalize(&shared).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_edge").await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT max(generation) FROM mst2_metadata_lifetime").await, + 1 + ); +} + +#[tokio::test] +async fn generation_observation_is_bound_to_prepare_not_only_the_same_dag() { + let (first, second, _schema) = fixture().await; + let a = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let b = PostgresMetadataGenerationRepository::new(second) + .await + .unwrap(); + let pages = prepared("/"); + let left = a.begin_intent("observation-left", &pages).await.unwrap(); + let right = b.begin_intent("observation-right", &pages).await.unwrap(); + a.install_pages(&left, pages.dag().payloads()) + .await + .unwrap(); + let observation = a.observe_installed_dag(&left).await.unwrap(); + assert_eq!( + rejected( + b.finalize_observation(&right, &observation) + .await + .unwrap_err() + ), + SnapshotErrorCode::IntegrityError + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + let receipt = a.finalize_observation(&left, &observation).await.unwrap(); + assert_eq!(receipt, a.finalize(&left).await.unwrap()); +} + +#[tokio::test] +async fn generation_finalization_rechecks_changed_lifetime_after_lock_free_observation() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("state-change", &pages) + .await + .unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let observed = repository.observe_installed_dag(&intent).await.unwrap(); + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + // Fault injection only: G1 deliberately has no collector transition API. + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_lifetime DISABLE TRIGGER mst2_metadata_lifetime_guard", + ) + .await + .unwrap(); + txn.execute_raw(statement( + "UPDATE mst2_metadata_lifetime SET state='DELETING' WHERE page_id=$1", + [pages.dag().root().to_vec().into()], + )) + .await + .unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_lifetime ENABLE TRIGGER mst2_metadata_lifetime_guard", + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + assert_eq!( + rejected( + repository + .finalize_observation(&intent, &observed) + .await + .unwrap_err() + ), + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE state='PREPARING'" + ) + .await, + 1 + ); + assert_eq!( + mst2_metadata_payload::Entity::find() + .count(&first) + .await + .unwrap(), + 2 + ); +} + +#[tokio::test] +async fn generation_deleting_removed_and_generic_tombstone_are_never_reopened() { + for state in ["DELETING", "REMOVED"] { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let old = repository.begin_intent("old", &pages).await.unwrap(); + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_lifetime DISABLE TRIGGER mst2_metadata_lifetime_guard", + ) + .await + .unwrap(); + txn.execute_raw(statement( + "UPDATE mst2_metadata_lifetime SET state=$1 WHERE page_id=$2", + [state.into(), pages.dag().root().to_vec().into()], + )) + .await + .unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_lifetime ENABLE TRIGGER mst2_metadata_lifetime_guard", + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + assert_eq!( + rejected( + repository + .begin_intent("fresh-disallowed", &pages) + .await + .unwrap_err() + ), + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + rejected( + repository + .install_pages(&old, pages.dag().payloads()) + .await + .unwrap_err() + ), + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_prepare").await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT max(generation) FROM mst2_metadata_lifetime").await, + 1 + ); + } + let (first, second, _schema) = fixture().await; + let pages = prepared("/"); + let graph = PostgresRetentionRepository::new(second.clone()); + graph + .retain_group(pages.dag().nodes(), pages.dag().edges(), &[]) + .await + .unwrap(); + assert_eq!( + graph + .mark_deleting("legacy-remove", &node_id(&pages.dag().root())) + .await + .unwrap(), + GcClaim::Marked + ); + graph.complete_gc("legacy-remove").await.unwrap(); + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + assert_eq!( + rejected( + repository + .begin_intent("no-resurrection", &pages) + .await + .unwrap_err() + ), + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_prepare").await, + 0 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 0 + ); + assert_eq!( + graph + .retain_group(pages.dag().nodes(), pages.dag().edges(), &[]) + .await + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); +} + +#[tokio::test] +async fn generation_legacy_null_capability_remains_legacy_and_does_not_block_existing_installer() { + let (first, second, _schema) = fixture().await; + let legacy = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let fresh = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let old = legacy + .begin_intent("legacy-operation", &pages) + .await + .unwrap(); + legacy + .install_pages(&old, pages.dag().payloads()) + .await + .unwrap(); + let original_receipt = legacy.finalize(&old).await.unwrap(); + assert_eq!( + rejected( + fresh + .begin_intent("legacy-operation", &pages) + .await + .unwrap_err() + ), + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + rejected( + fresh + .begin_intent("new-over-legacy-cas", &pages) + .await + .unwrap_err() + ), + SnapshotErrorCode::IntegrityError + ); + assert_eq!(legacy.finalize(&old).await.unwrap(), original_receipt); + let another = legacy + .begin_intent("legacy-new-resolve", &pages) + .await + .unwrap(); + legacy + .install_pages(&another, pages.dag().payloads()) + .await + .unwrap(); + legacy.finalize(&another).await.unwrap(); + assert_eq!( + scalar( + &second, + "SELECT count(*) FROM mst2_metadata_payload WHERE generation IS NULL" + ) + .await, + 2 + ); + assert_eq!( + scalar( + &second, + "SELECT count(*) FROM mst2_metadata_prepare WHERE storage_seal IS NULL" + ) + .await, + 2 + ); + assert_eq!( + mst2_metadata_lifetime::Entity::find() + .count(&first) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn generation_storage_seal_scope_and_complete_child_binding_cannot_be_substituted() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("seal-tamper", &pages) + .await + .unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let mut wrong_bindings = intent.clone(); + wrong_bindings.bindings_digest[0] ^= 1; + let mut wrong_seal = intent.clone(); + wrong_seal.storage_seal[0] ^= 1; + let mut wrong_root = intent.clone(); + wrong_root.metadata_root[0] ^= 1; + for wrong in [wrong_bindings, wrong_seal, wrong_root] { + assert_eq!( + rejected( + repository + .install_pages(&wrong, pages.dag().payloads()) + .await + .unwrap_err() + ), + SnapshotErrorCode::IntegrityError + ); + } + let mut wrong_scope = intent.clone(); + wrong_scope.primary_scope[0] ^= 1; + assert_eq!( + repository + .observe_installed_dag(&wrong_scope) + .await + .unwrap_err() + .code, + SnapshotErrorCode::IntegrityError + ); + let child = pages + .dag() + .payloads() + .iter() + .find(|page| page.id != pages.dag().root()) + .unwrap(); + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + txn.execute_unprepared("ALTER TABLE mst2_metadata_prepare_page DISABLE TRIGGER mst2_metadata_generation_mapping_guard").await.unwrap(); + txn.execute_raw(statement("UPDATE mst2_metadata_prepare_page SET expected_size=expected_size+1 WHERE prepare_id=$1 AND page_id=$2", [intent.prepare_id().into(), child.id.to_vec().into()])).await.unwrap(); + txn.execute_unprepared("ALTER TABLE mst2_metadata_prepare_page ENABLE TRIGGER mst2_metadata_generation_mapping_guard").await.unwrap(); + txn.commit().await.unwrap(); + assert_eq!( + repository + .observe_installed_dag(&intent) + .await + .unwrap_err() + .code, + SnapshotErrorCode::IntegrityError + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); +} + +#[tokio::test] +async fn generation_compact_receipt_checks_actual_seal_without_loading_the_dag() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository.begin_intent("compact", &pages).await.unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = repository.finalize(&intent).await.unwrap(); + let source = receipt.legacy.identity.tagged_root_tree_oid.clone(); + let txn = second.begin().await.unwrap(); + repository + .verify_receipt_in_txn(&txn, &receipt, &source, "/") + .await + .unwrap(); + txn.rollback().await.unwrap(); + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_prepare DISABLE TRIGGER mst2_metadata_generation_seal_guard", + ) + .await + .unwrap(); + txn.execute_raw(statement( + "UPDATE mst2_metadata_prepare SET storage_seal=$1 WHERE prepare_id=$2", + [vec![0u8; 32].into(), intent.prepare_id().into()], + )) + .await + .unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_prepare ENABLE TRIGGER mst2_metadata_generation_seal_guard", + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + let txn = second.begin().await.unwrap(); + assert_eq!( + repository + .verify_receipt_in_txn(&txn, &receipt, &source, "/") + .await + .unwrap_err() + .code, + SnapshotErrorCode::IntegrityError + ); + txn.rollback().await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_prepare DISABLE TRIGGER mst2_metadata_generation_seal_guard", + ) + .await + .unwrap(); + txn.execute_raw(statement( + "UPDATE mst2_metadata_prepare SET storage_seal=$1, + canonical_bindings=set_byte(canonical_bindings,60,get_byte(canonical_bindings,60)#1) WHERE prepare_id=$2", + [intent.storage_seal.to_vec().into(),intent.prepare_id().into()], + )).await.unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_prepare ENABLE TRIGGER mst2_metadata_generation_seal_guard", + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + let txn = second.begin().await.unwrap(); + assert_eq!( + repository + .verify_receipt_in_txn(&txn, &receipt, &source, "/") + .await + .unwrap_err() + .code, + SnapshotErrorCode::IntegrityError + ); + txn.rollback().await.unwrap(); +} + +#[tokio::test] +async fn generation_db_guards_preserve_watermark_payload_scope_and_committed_mappings() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository.begin_intent("guards", &pages).await.unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + repository.finalize(&intent).await.unwrap(); + for sql in [ + "DELETE FROM mst2_metadata_lifetime", + "UPDATE mst2_metadata_lifetime SET generation=2", + "UPDATE mst2_metadata_lifetime SET state='REMOVED'", + "DELETE FROM mst2_metadata_payload", + "UPDATE mst2_metadata_payload SET generation=1", + "DELETE FROM mst2_metadata_storage_scope", + "UPDATE mst2_metadata_storage_scope SET storage_uuid=storage_uuid", + "DELETE FROM mst2_metadata_prepare_page", + "UPDATE mst2_metadata_prepare_page SET generation=1", + "UPDATE mst2_metadata_prepare SET storage_seal=NULL,canonical_bindings=NULL,bindings_digest=NULL,primary_scope=NULL", + ] { + assert!( + second.execute_unprepared(sql).await.is_err(), + "guard must reject {sql}" + ); + } + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE generation=1 AND state='LIVE'" + ) + .await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare_page WHERE generation=1" + ) + .await, + 2 + ); + assert_eq!( + repository.finalize(&intent).await.unwrap().intent(), + &intent + ); + assert_eq!(scalar(&first, "SELECT count(*) FROM pg_indexes WHERE schemaname=current_schema() AND indexname='idx_mst2_metadata_prepare_page_lifetime'").await, 1); +} + +#[tokio::test] +async fn generation_finalize_row_lock_fences_a_late_mapping_insert_on_another_connection() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("late-mapping", &pages) + .await + .unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let observation = repository.observe_installed_dag(&intent).await.unwrap(); + let held = repository.inner.transaction().await.unwrap(); + repository.inner.barrier(&held).await.unwrap(); + lock_prepare(&held, intent.prepare_id()).await.unwrap(); + let contender = second.begin().await.unwrap(); + let pid = contender + .query_one_raw(statement("SELECT pg_backend_pid() AS pid", [])) + .await + .unwrap() + .unwrap() + .try_get::("", "pid") + .unwrap(); + let prepare_id = intent.prepare_id().to_owned(); + let root = pages.dag().root(); + let size = pages.install_plan().unwrap().pages[&root] as i32; + let blocked = tokio::spawn(async move { + let result = contender.execute_raw(statement( + "INSERT INTO mst2_metadata_prepare_page(prepare_id,page_id,generation,expected_size) VALUES($1,$2,1,$3)", + [prepare_id.into(), root.to_vec().into(), size.into()], + )).await; + contender.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 pid=$1 AND NOT granted) AS waiting", + [pid.into()], + )) + .await + .unwrap() + .unwrap() + .try_get::("", "waiting") + .unwrap(); + if waiting { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "late mapping writer must wait for the prepare row lock" + ); + tokio::task::yield_now().await; + } + repository + .finalize_in_txn(&held, &intent, &observation) + .await + .unwrap(); + held.commit().await.unwrap(); + let error = blocked.await.unwrap().unwrap_err(); + assert!( + error + .to_string() + .contains("metadata generation mappings are immutable"), + "{error}" + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_prepare_page").await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE state='COMMITTED'" + ) + .await, + 1 + ); + repository.finalize(&intent).await.unwrap(); +} + +#[tokio::test] +async fn generation_wrong_primary_never_upgrades_intent_or_reports_false_absence() { + let (first, same_primary, _schema) = fixture().await; + let (other, _other_connection, _other_schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let wrong = PostgresMetadataGenerationRepository::new(other.clone()) + .await + .unwrap(); + let pages = prepared("/"); + let intent = repository + .begin_intent("captured-primary", &pages) + .await + .unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = repository.finalize(&intent).await.unwrap(); + assert_eq!( + rejected( + wrong + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap_err() + ), + SnapshotErrorCode::IntegrityError + ); + assert_eq!( + wrong.observe_installed_dag(&intent).await.unwrap_err().code, + SnapshotErrorCode::IntegrityError + ); + assert!(matches!( + repository + .inspect_prepare( + &other, + "captured-primary", + intent.manifest_digest(), + MetadataCommitPhase::Finalize + ) + .await, + Err(MetadataInstallError::CommitUncertain { + phase: MetadataCommitPhase::Finalize, + .. + }) + )); + assert_eq!( + repository + .inspect_prepare( + &same_primary, + "captured-primary", + intent.manifest_digest(), + MetadataCommitPhase::Finalize + ) + .await + .unwrap(), + GenerationPrepareObservation::Committed(Box::new(receipt)) + ); + assert_eq!( + scalar(&other, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + 2 + ); +} diff --git a/src/jupiter/storage/native_metadata_generations.rs b/src/jupiter/storage/native_metadata_generations.rs new file mode 100644 index 00000000..fd1323cd --- /dev/null +++ b/src/jupiter/storage/native_metadata_generations.rs @@ -0,0 +1,866 @@ +//! Generation-bound preparation repository, isolated from HTTP lease authority. + +use sha2::{Digest, Sha256}; + +use super::*; + +const MAX_BINDINGS_BYTES: usize = 12 + 48 * 4096; +const MAX_PRIMARY_SCOPE_BYTES: usize = 16384; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GenerationPrepareIntent { + legacy: MetadataPrepareIntent, + metadata_root: [u8; 32], + root_generation: i64, + bindings_digest: [u8; 32], + primary_scope: Vec, + storage_seal: [u8; 32], +} + +impl GenerationPrepareIntent { + pub fn prepare_id(&self) -> &str { + self.legacy.prepare_id() + } + + pub fn manifest_digest(&self) -> [u8; 32] { + self.legacy.manifest_digest() + } + + pub fn operation_id(&self) -> &str { + self.legacy.operation_id() + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GenerationMetadataReceipt { + legacy: PreparedMetadataReceipt, + intent: GenerationPrepareIntent, +} + +impl GenerationMetadataReceipt { + pub fn intent(&self) -> &GenerationPrepareIntent { + &self.intent + } + + pub fn metadata_root(&self) -> [u8; 32] { + self.intent.metadata_root + } + + pub fn payload_bytes(&self) -> u64 { + self.legacy.payload_bytes() + } +} + +#[derive(Debug)] +pub struct GenerationDagObservation { + intent: GenerationPrepareIntent, + dag: ValidatedMetadataDag, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GenerationPrepareObservation { + Absent, + Preparing(GenerationPrepareIntent), + Committed(Box), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct GenerationBindings(BTreeMap<[u8; 32], (i64, u64)>); + +impl GenerationBindings { + fn encode(&self) -> Result, SnapshotError> { + if self.0.is_empty() || self.0.len() > MetadataDagLimits::default().nodes { + return Err(integrity("invalid metadata generation binding count")); + } + let mut bytes = Vec::with_capacity(12 + 48 * self.0.len()); + bytes.extend_from_slice(b"MST2GEN1"); + bytes.extend_from_slice(&(self.0.len() as u32).to_be_bytes()); + for (page, (generation, size)) in &self.0 { + if *generation <= 0 || !(HEADER_LEN as u64..=PAGE_MAX_BYTES as u64).contains(size) { + return Err(integrity("invalid metadata lifetime generation or size")); + } + bytes.extend_from_slice(page); + bytes.extend_from_slice(&generation.to_be_bytes()); + bytes.extend_from_slice(&size.to_be_bytes()); + } + Ok(bytes) + } + + fn decode(bytes: &[u8], plan: &MetadataInstallPlan) -> Result { + if bytes.len() < 60 || bytes.len() > MAX_BINDINGS_BYTES || &bytes[..8] != b"MST2GEN1" { + return Err(integrity("invalid metadata generation binding encoding")); + } + let count = u32::from_be_bytes(bytes[8..12].try_into().map_err(internal)?) as usize; + if count != plan.pages.len() || bytes.len() != 12 + count * 48 { + return Err(integrity( + "metadata generation binding coverage differs from plan", + )); + } + let mut bindings = BTreeMap::new(); + for entry in bytes[12..].chunks_exact(48) { + let page: [u8; 32] = entry[..32].try_into().map_err(internal)?; + let generation = i64::from_be_bytes(entry[32..40].try_into().map_err(internal)?); + let size = u64::from_be_bytes(entry[40..48].try_into().map_err(internal)?); + if generation <= 0 + || plan.pages.get(&page) != Some(&size) + || bindings.insert(page, (generation, size)).is_some() + { + return Err(integrity("invalid fixed metadata lifetime binding")); + } + } + let result = Self(bindings); + if result.encode()? != bytes { + return Err(integrity("noncanonical metadata generation bindings")); + } + Ok(result) + } +} + +struct FixedPlan { + stored: StoredPlan, + bindings: GenerationBindings, + intent: GenerationPrepareIntent, +} + +#[derive(Clone)] +pub struct PostgresMetadataGenerationRepository { + inner: PostgresMetadataInstallRepository, +} + +impl PostgresMetadataGenerationRepository { + pub async fn new(connection: DatabaseConnection) -> Result { + Ok(Self { + inner: PostgresMetadataInstallRepository::new(connection).await?, + }) + } + + pub async fn begin_intent( + &self, + operation_id: &str, + prepared: &PreparedNativeMetadataRetention, + ) -> Result { + validate_operation_id(operation_id)?; + let plan = prepared.install_plan()?; + let digest = plan.digest()?; + let primary_scope = self.primary_scope()?; + let txn = self.inner.transaction().await?; + let result = async { + self.inner.barrier(&txn).await?; + if let Some(fixed) = self.load_fixed_plan(&txn, operation_id, &digest).await? { + return Ok(fixed.intent); + } + let bindings = allocate_lifetimes(&txn, &plan).await?; + let canonical_bindings = bindings.encode()?; + let bindings_digest = Sha256::digest(&canonical_bindings).into(); + let prepare_id = uuid::Uuid::new_v4().to_string(); + let intent = GenerationPrepareIntent { + legacy: MetadataPrepareIntent { prepare_id: prepare_id.clone(), operation_id: operation_id.into(), manifest_digest: digest }, + metadata_root: plan.root, + root_generation: bindings.0.get(&plan.root).ok_or_else(|| integrity("root lifetime is missing"))?.0, + bindings_digest, + storage_seal: seal(&prepare_id, &digest, &plan.root, &bindings_digest, &primary_scope)?, + primary_scope, + }; + let identity = &plan.identity; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_prepare(prepare_id,operation_id,manifest_digest,canonical_plan, + source_domain,tagged_root_tree_oid,scope,schema_version,metadata_codec,materialization_policy, + fs_semantics,access_projection,verification_revision,projection_revision,metadata_root, + node_count,edge_count,total_bytes,state,canonical_bindings,bindings_digest,primary_scope,storage_seal) + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,'PREPARING',$19,$20,$21,$22)", + [prepare_id.clone().into(),operation_id.into(),digest.to_vec().into(),plan.encode()?.into(), + identity.source_domain.clone().into(),identity.tagged_root_tree_oid.clone().into(),identity.scope.clone().into(), + (identity.schema_version as i16).into(),(identity.metadata_codec as i16).into(), + (identity.materialization_policy as i16).into(),(identity.fs_semantics as i16).into(), + (identity.access_projection as i16).into(),identity.verification_revision.into(), + (identity.projection_revision as i16).into(),plan.root.to_vec().into(),(plan.pages.len() as i32).into(), + (plan.edges.len() as i32).into(),(plan.total_bytes as i64).into(),canonical_bindings.into(), + intent.bindings_digest.to_vec().into(),intent.primary_scope.clone().into(),intent.storage_seal.to_vec().into()], + )).await.map_err(internal)?; + let pages: Vec<_> = bindings.0.iter().map(|(page,(generation,size))| + json!({"page_id":hex::encode(page),"generation":generation,"size":size})).collect(); + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_prepare_page(prepare_id,page_id,generation,expected_size) + SELECT $1,decode(p.page_id,'hex'),p.generation,p.size + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,generation bigint,size integer)", + [prepare_id.clone().into(),serde_json::to_string(&pages).map_err(internal)?.into()], + )).await.map_err(internal)?; + // Reuse only already LIVE graph rows. RESERVED pages without a graph + // are protected by the durable mappings, without inventing a DAG. + txn.execute_raw(statement( + "INSERT INTO mst2_retention_root(node_id,root_key,root_kind) + SELECT l.node_id,'prepare:'||p.prepare_id,'prepare' + FROM mst2_metadata_prepare_page p JOIN mst2_metadata_lifetime l + ON l.page_id=p.page_id AND l.generation=p.generation + JOIN mst2_retention_node n ON n.node_id=l.node_id AND n.state='LIVE' + WHERE p.prepare_id=$1 ON CONFLICT(node_id,root_key) DO NOTHING", + [prepare_id.into()], + )).await.map_err(internal)?; + Ok(intent) + }.await; + commit( + txn, + result, + operation_id, + digest, + MetadataCommitPhase::Intent, + ) + .await + } + + pub async fn install_page( + &self, + intent: &GenerationPrepareIntent, + payload: &MetadataPagePayload, + ) -> Result<(), MetadataInstallError> { + self.install_pages(intent, std::slice::from_ref(payload)) + .await + } + + pub async fn install_pages( + &self, + intent: &GenerationPrepareIntent, + payloads: &[MetadataPagePayload], + ) -> Result<(), MetadataInstallError> { + 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()); + } + } + let txn = self.inner.transaction().await?; + let result = async { + self.inner.barrier(&txn).await?; + let fixed = self.require_fixed_plan(&txn, intent).await?; + if fixed.stored.record.state == "COMMITTED" { + check_generation_payloads(&txn, &fixed).await?; + } + let mut pages = Vec::with_capacity(payloads.len()); + for page in payloads { + let &(generation,size) = fixed.bindings.0.get(&page.id) + .ok_or_else(|| integrity("metadata payload is not a member of its fixed generation installation"))?; + if size != page.size { + return Err(integrity("metadata payload size differs from its fixed lifetime")); + } + pages.push(json!({"page_id":hex::encode(page.id),"generation":generation,"size":size,"payload":hex::encode(&page.bytes)})); + } + let encoded = serde_json::to_string(&pages).map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_payload(page_id,generation,metadata_codec,byte_size,payload) + SELECT decode(p.page_id,'hex'),p.generation,$1,p.size,decode(p.payload,'hex') + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,generation bigint,size integer,payload text) + ON CONFLICT(page_id) DO NOTHING", + [fixed.stored.record.metadata_codec.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,generation bigint,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.generation IS DISTINCT FROM p.generation OR b.metadata_codec<>$1 + OR b.byte_size<>p.size OR b.payload<>decode(p.payload,'hex') LIMIT 1", + [fixed.stored.record.metadata_codec.into(),encoded.into()], + )).await.map_err(internal)?; + if bad.is_some() { + return Err(integrity("immutable payload conflicts with its exact metadata lifetime")); + } + Ok(()) + }.await; + commit( + txn, + result, + &intent.legacy.operation_id, + intent.manifest_digest(), + MetadataCommitPhase::Payload, + ) + .await + } + + /// Read, hash and parse the actual fixed-generation bytes outside the lock. + pub async fn observe_installed_dag( + &self, + intent: &GenerationPrepareIntent, + ) -> Result { + self.inner + .verify_primary_connection(&self.inner.connection) + .await?; + let fixed = self + .require_fixed_plan(&self.inner.connection, intent) + .await?; + let dag = load_fixed_dag(&self.inner.connection, &fixed).await?; + Ok(GenerationDagObservation { + intent: intent.clone(), + dag, + }) + } + + pub async fn finalize( + &self, + intent: &GenerationPrepareIntent, + ) -> Result { + let observation = self.observe_installed_dag(intent).await?; + self.finalize_observation(intent, &observation).await + } + + pub async fn finalize_observation( + &self, + intent: &GenerationPrepareIntent, + observation: &GenerationDagObservation, + ) -> Result { + if &observation.intent != intent { + return Err( + integrity("metadata observation belongs to another fixed installation").into(), + ); + } + let txn = self.inner.transaction().await?; + let result = self.finalize_in_txn(&txn, intent, observation).await; + commit( + txn, + result, + &intent.legacy.operation_id, + intent.manifest_digest(), + MetadataCommitPhase::Finalize, + ) + .await + } + + // Only a successful outer commit or the recovery barrier exposes a receipt. + async fn finalize_in_txn( + &self, + txn: &DatabaseTransaction, + intent: &GenerationPrepareIntent, + observation: &GenerationDagObservation, + ) -> Result { + if &observation.intent != intent { + return Err(integrity( + "metadata observation belongs to another fixed installation", + )); + } + self.inner.barrier(txn).await?; + // Mapping INSERT guards take this same row lock. No late insertion + // may cross coverage validation and the COMMITTED transition. + lock_prepare(txn, intent.prepare_id()).await?; + let fixed = self.require_fixed_plan(txn, intent).await?; + check_generation_payloads(txn, &fixed).await?; + let legacy = self + .inner + .finalize_stored_plan_in_txn(txn, &intent.legacy, &observation.dag, fixed.stored) + .await?; + txn.execute_raw(statement( + "UPDATE mst2_metadata_lifetime l SET state='LIVE' + FROM mst2_metadata_prepare_page p WHERE p.prepare_id=$1 AND p.page_id=l.page_id + AND p.generation=l.generation AND l.state='RESERVED'", + [intent.prepare_id().into()], + )) + .await + .map_err(internal)?; + Ok(GenerationMetadataReceipt { + legacy, + intent: intent.clone(), + }) + } + + /// Compact proof for a future atomic handoff. No DAG/page scan under the + /// mono writer lock: hash <=196620 binding bytes and check the root lifetime. + pub(crate) async fn verify_receipt_in_txn( + &self, + txn: &DatabaseTransaction, + receipt: &GenerationMetadataReceipt, + tagged_root_tree_oid: &str, + scope: &str, + ) -> Result<(), SnapshotError> { + self.require_scope(&receipt.intent)?; + self.inner + .verify_receipt_in_txn(txn, &receipt.legacy, tagged_root_tree_oid, scope) + .await?; + let row=txn.query_one_raw(statement( + "SELECT p.bindings_digest,p.primary_scope,p.storage_seal, + CASE WHEN octet_length(p.canonical_bindings)<=$2 THEN sha256(p.canonical_bindings) END AS actual_bindings_digest, + l.generation,l.state,l.metadata_codec,l.expected_size,m.generation AS root_generation, + b.generation AS payload_generation,b.byte_size + FROM mst2_metadata_prepare p LEFT JOIN mst2_metadata_prepare_page m + ON m.prepare_id=p.prepare_id AND m.page_id=p.metadata_root + LEFT JOIN mst2_metadata_lifetime l ON l.page_id=m.page_id AND l.generation=m.generation + LEFT JOIN mst2_metadata_payload b ON b.page_id=m.page_id + WHERE p.prepare_id=$1", + [receipt.intent.prepare_id().into(),(MAX_BINDINGS_BYTES as i32).into()], + )).await.map_err(internal)?.ok_or_else(|| unavailable("metadata generation receipt is missing"))?; + for (column, expected) in [ + ("bindings_digest", receipt.intent.bindings_digest.as_slice()), + ( + "actual_bindings_digest", + receipt.intent.bindings_digest.as_slice(), + ), + ("primary_scope", receipt.intent.primary_scope.as_slice()), + ("storage_seal", receipt.intent.storage_seal.as_slice()), + ] { + if row + .try_get::>>("", column) + .map_err(internal)? + .as_deref() + != Some(expected) + { + return Err(integrity( + "metadata generation receipt seal differs from durable storage", + )); + } + } + let generation = row + .try_get::>("", "generation") + .map_err(internal)?; + if generation != Some(receipt.intent.root_generation) + || row + .try_get::>("", "root_generation") + .map_err(internal)? + != generation + || row + .try_get::>("", "payload_generation") + .map_err(internal)? + != generation + || row + .try_get::>("", "state") + .map_err(internal)? + .as_deref() + != Some("LIVE") + || row + .try_get::>("", "metadata_codec") + .map_err(internal)? + != Some(receipt.legacy.identity.metadata_codec as i16) + || row + .try_get::>("", "expected_size") + .map_err(internal)? + != Some(receipt.legacy.root_payload_bytes as i32) + || row + .try_get::>("", "byte_size") + .map_err(internal)? + != Some(receipt.legacy.root_payload_bytes as i32) + { + return Err(unavailable( + "metadata generation receipt root lifetime is unavailable", + )); + } + Ok(()) + } + + pub async fn inspect_prepare( + &self, + fresh_primary: &DatabaseConnection, + operation_id: &str, + digest: [u8; 32], + phase: MetadataCommitPhase, + ) -> Result { + validate_operation_id(operation_id)?; + let txn = fresh_primary + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .map_err(|_| uncertain(operation_id, digest, phase))?; + if self.inner.barrier(&txn).await.is_err() { + let _ = txn.rollback().await; + return Err(uncertain(operation_id, digest, phase)); + } + let result = async { + match self.load_fixed_plan(&txn, operation_id, &digest).await? { + None => Ok(GenerationPrepareObservation::Absent), + Some(fixed) if fixed.stored.record.state == "COMMITTED" => { + check_generation_payloads(&txn, &fixed).await?; + load_fixed_dag(&txn, &fixed).await?; + verify_graph(&txn, &fixed.stored).await?; + Ok(GenerationPrepareObservation::Committed(Box::new( + GenerationMetadataReceipt { + legacy: fixed.stored.receipt()?, + intent: fixed.intent, + }, + ))) + } + Some(fixed) => Ok(GenerationPrepareObservation::Preparing(fixed.intent)), + } + } + .await; + txn.rollback() + .await + .map_err(|_| uncertain(operation_id, digest, phase))?; + result.map_err(|error: SnapshotError| { + if error.code == SnapshotErrorCode::Internal { + uncertain(operation_id, digest, phase) + } else { + MetadataInstallError::Rejected(error) + } + }) + } + + async fn require_fixed_plan( + &self, + connection: &C, + intent: &GenerationPrepareIntent, + ) -> Result { + self.require_scope(intent)?; + let fixed = self + .load_fixed_plan( + connection, + &intent.legacy.operation_id, + &intent.manifest_digest(), + ) + .await? + .ok_or_else(|| unavailable("fixed metadata preparation is missing"))?; + if &fixed.intent != intent { + return Err(integrity( + "metadata intent differs from its fixed generation seal", + )); + } + Ok(fixed) + } + + async fn load_fixed_plan( + &self, + connection: &C, + operation_id: &str, + digest: &[u8; 32], + ) -> Result, SnapshotError> { + let Some(stored) = load_plan(connection, operation_id, digest).await? else { + return Ok(None); + }; + let canonical = stored + .record + .canonical_bindings + .as_deref() + .ok_or_else(|| unavailable("legacy preparation has no generation seal"))?; + let bindings = GenerationBindings::decode(canonical, &stored.plan)?; + let bindings_digest: [u8; 32] = Sha256::digest(canonical).into(); + let primary_scope = stored + .record + .primary_scope + .clone() + .ok_or_else(|| integrity("fixed preparation primary scope is missing"))?; + let storage_seal = seal( + &stored.record.prepare_id, + digest, + &stored.plan.root, + &bindings_digest, + &primary_scope, + )?; + if stored.record.bindings_digest.as_deref() != Some(bindings_digest.as_slice()) + || stored.record.storage_seal.as_deref() != Some(storage_seal.as_slice()) + { + return Err(integrity("metadata generation seal is corrupt")); + } + let intent = GenerationPrepareIntent { + legacy: stored.intent()?, + metadata_root: stored.plan.root, + root_generation: bindings + .0 + .get(&stored.plan.root) + .ok_or_else(|| integrity("root lifetime is missing"))? + .0, + bindings_digest, + primary_scope, + storage_seal, + }; + self.require_scope(&intent)?; + let mut actual = BTreeMap::new(); + for row in &stored.prepare_pages { + let page: [u8; 32] = row.page_id.as_slice().try_into().map_err(internal)?; + let generation = row + .generation + .ok_or_else(|| integrity("fixed metadata page has no generation"))?; + actual.insert(page, (generation, row.expected_size as u64)); + } + if actual != bindings.0 { + return Err(integrity( + "metadata mappings differ from complete fixed generation bindings", + )); + } + check_lifetimes(connection, &stored).await?; + Ok(Some(FixedPlan { + stored, + bindings, + intent, + })) + } + + fn primary_scope(&self) -> Result, SnapshotError> { + let scope = &self.inner.storage_scope; + 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() > MAX_PRIMARY_SCOPE_BYTES { + return Err(integrity("invalid captured metadata primary scope size")); + } + Ok(bytes) + } + + fn require_scope(&self, intent: &GenerationPrepareIntent) -> Result<(), SnapshotError> { + if intent.primary_scope != self.primary_scope()? { + return Err(integrity( + "metadata seal belongs to another captured primary storage scope", + )); + } + Ok(()) + } +} + +fn seal( + prepare_id: &str, + manifest: &[u8; 32], + root: &[u8; 32], + bindings: &[u8; 32], + scope: &[u8], +) -> Result<[u8; 32], SnapshotError> { + if scope.is_empty() || scope.len() > MAX_PRIMARY_SCOPE_BYTES { + return Err(integrity("invalid metadata generation seal primary scope")); + } + let id = uuid::Uuid::parse_str(prepare_id).map_err(internal)?; + let mut hash = Sha256::new(); + hash.update(b"MST2-METADATA-STORAGE-SEAL-1\0"); + hash.update(id.as_bytes()); + hash.update(manifest); + hash.update(root); + hash.update(bindings); + hash.update((scope.len() as u32).to_be_bytes()); + hash.update(scope); + Ok(hash.finalize().into()) +} + +async fn allocate_lifetimes( + txn: &DatabaseTransaction, + plan: &MetadataInstallPlan, +) -> Result { + let pages: Vec<_> = plan + .pages + .iter() + .map(|(page, size)| json!({"page_id":hex::encode(page),"size":size})) + .collect(); + let encoded = serde_json::to_string(&pages).map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size) + SELECT decode(p.page_id,'hex'),'page:sha256:'||p.page_id,1,'RESERVED',$1,p.size + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer) + ON CONFLICT(page_id) DO NOTHING", + [(plan.identity.metadata_codec as i16).into(),encoded.clone().into()], + )).await.map_err(internal)?; + let rows=txn.query_all_raw(statement( + "SELECT decode(p.page_id,'hex') AS page_id,p.size,l.generation,l.state,l.metadata_codec,l.expected_size, + n.state AS graph_state,n.kind AS graph_kind,n.bytes AS graph_bytes, + b.page_id IS NOT NULL AS payload_present,b.generation AS payload_generation, + b.metadata_codec AS payload_codec,b.byte_size AS payload_size, + EXISTS(SELECT 1 FROM mst2_retention_gc_op g WHERE g.node_id=l.node_id + AND g.operation='REMOVE' AND g.state IN ('PENDING','APPLIED')) AS tombstone + FROM jsonb_to_recordset($1::jsonb) AS p(page_id text,size integer) + JOIN mst2_metadata_lifetime l ON l.page_id=decode(p.page_id,'hex') + LEFT JOIN mst2_retention_node n ON n.node_id=l.node_id + LEFT JOIN mst2_metadata_payload b ON b.page_id=l.page_id ORDER BY l.page_id", + [encoded.into()], + )).await.map_err(internal)?; + let mut bindings = BTreeMap::new(); + for row in rows { + let state: String = row.try_get("", "state").map_err(internal)?; + let generation: i64 = row.try_get("", "generation").map_err(internal)?; + let size: i32 = row.try_get("", "size").map_err(internal)?; + if state == "LIVE" + && (!row + .try_get::("", "payload_present") + .map_err(internal)? + || row + .try_get::>("", "graph_state") + .map_err(internal)? + .is_none()) + { + return Err(unavailable( + "LIVE metadata lifetime has lost its payload or graph", + )); + } + if !["RESERVED", "LIVE"].contains(&state.as_str()) + || row.try_get::("", "tombstone").map_err(internal)? + || row + .try_get::>("", "graph_state") + .map_err(internal)? + .is_some_and(|state| state != "LIVE") + { + return Err(unavailable("metadata lifetime is deleting or removed")); + } + if generation <= 0 + || row.try_get::("", "metadata_codec").map_err(internal)? + != plan.identity.metadata_codec as i16 + || row.try_get::("", "expected_size").map_err(internal)? != size + || 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(integrity( + "metadata lifetime profile conflicts with fixed plan", + )); + } + if row + .try_get::("", "payload_present") + .map_err(internal)? + && (row + .try_get::>("", "payload_generation") + .map_err(internal)? + != Some(generation) + || row + .try_get::>("", "payload_codec") + .map_err(internal)? + != Some(plan.identity.metadata_codec as i16) + || row + .try_get::>("", "payload_size") + .map_err(internal)? + != Some(size)) + { + return Err(integrity( + "existing metadata payload has another or unbound lifetime", + )); + } + let page: Vec = row.try_get("", "page_id").map_err(internal)?; + bindings.insert( + page.as_slice().try_into().map_err(internal)?, + (generation, size as u64), + ); + } + if bindings.len() != plan.pages.len() { + return Err(integrity("incomplete metadata lifetime allocation")); + } + Ok(GenerationBindings(bindings)) +} + +async fn check_lifetimes( + connection: &C, + stored: &StoredPlan, +) -> Result<(), SnapshotError> { + let row=connection.query_one_raw(statement( + "SELECT p.page_id,l.generation,l.state,l.metadata_codec,l.expected_size,p.generation AS expected_generation, + n.state AS graph_state,n.kind AS graph_kind,n.bytes AS graph_bytes, + EXISTS(SELECT 1 FROM mst2_retention_gc_op g WHERE g.node_id=l.node_id + AND g.operation='REMOVE' AND g.state IN ('PENDING','APPLIED')) AS tombstone + FROM mst2_metadata_prepare_page p LEFT JOIN mst2_metadata_lifetime l ON l.page_id=p.page_id + LEFT JOIN mst2_retention_node n ON n.node_id=l.node_id WHERE p.prepare_id=$1 + AND (l.page_id IS NULL OR l.generation IS DISTINCT FROM p.generation OR l.metadata_codec<>$2 + OR l.expected_size<>p.expected_size OR l.state NOT IN ('RESERVED','LIVE') + OR ($3='COMMITTED' AND l.state<>'LIVE') + OR (l.state='LIVE' AND n.node_id IS NULL) + OR n.state<>'LIVE' OR n.kind<>'page' OR n.bytes<>p.expected_size + OR EXISTS(SELECT 1 FROM mst2_retention_gc_op g WHERE g.node_id=l.node_id + AND g.operation='REMOVE' AND g.state IN ('PENDING','APPLIED'))) LIMIT 1", + [stored.record.prepare_id.clone().into(),stored.record.metadata_codec.into(),stored.record.state.clone().into()], + )).await.map_err(internal)?; + if let Some(row) = row { + if row + .try_get::>("", "state") + .map_err(internal)? + .is_some_and(|state| !["RESERVED", "LIVE"].contains(&state.as_str())) + || row + .try_get::>("", "graph_state") + .map_err(internal)? + .is_some_and(|state| state != "LIVE") + || row.try_get::("", "tombstone").map_err(internal)? + { + return Err(unavailable( + "fixed metadata lifetime is no longer installable", + )); + } + return Err(integrity( + "fixed metadata lifetime registry conflicts with its bindings", + )); + } + Ok(()) +} + +async fn check_generation_payloads( + connection: &C, + fixed: &FixedPlan, +) -> Result<(), SnapshotError> { + let row=connection.query_one_raw(statement( + "SELECT p.page_id FROM mst2_metadata_prepare_page p LEFT JOIN mst2_metadata_payload b ON b.page_id=p.page_id + WHERE p.prepare_id=$1 AND (b.page_id IS NULL OR b.generation IS DISTINCT FROM p.generation + OR b.metadata_codec<>$2 OR b.byte_size<>p.expected_size OR octet_length(b.payload)<>p.expected_size) LIMIT 1", + [fixed.intent.prepare_id().into(),fixed.stored.record.metadata_codec.into()], + )).await.map_err(internal)?; + if row.is_some() { + return Err(unavailable( + "exact metadata generation payload coverage is incomplete", + )); + } + Ok(()) +} + +async fn load_fixed_dag( + connection: &C, + fixed: &FixedPlan, +) -> Result { + let rows = connection + .query_all_raw(statement( + "SELECT b.page_id,b.generation,b.metadata_codec,b.byte_size,b.payload + FROM mst2_metadata_prepare_page p JOIN mst2_metadata_payload b + ON b.page_id=p.page_id AND b.generation=p.generation WHERE p.prepare_id=$1", + [fixed.intent.prepare_id().into()], + )) + .await + .map_err(internal)?; + let mut pages = Vec::with_capacity(rows.len()); + for row in rows { + let id: Vec = row.try_get("", "page_id").map_err(internal)?; + let id: [u8; 32] = id.as_slice().try_into().map_err(internal)?; + let generation: i64 = row.try_get("", "generation").map_err(internal)?; + let size: i32 = row.try_get("", "byte_size").map_err(internal)?; + if fixed.bindings.0.get(&id) != Some(&(generation, size as u64)) + || row.try_get::("", "metadata_codec").map_err(internal)? + != fixed.stored.record.metadata_codec + { + return Err(integrity( + "installed bytes differ from their fixed lifetime binding", + )); + } + let payload = MetadataPagePayload { + id, + size: size as u64, + bytes: row.try_get("", "payload").map_err(internal)?, + }; + validate_payload(&payload)?; + pages.push(payload); + } + if pages.len() != fixed.bindings.0.len() { + return Err(unavailable( + "exact metadata generation payload coverage is incomplete", + )); + } + ValidatedMetadataDag::validate( + MetadataDagCandidate { + metadata_codec: fixed.stored.plan.identity.metadata_codec, + root: fixed.stored.plan.root, + pages, + edges: fixed.stored.plan.edges.iter().copied().collect(), + }, + MetadataDagLimits::default(), + ) +} + +async fn lock_prepare(txn: &DatabaseTransaction, prepare_id: &str) -> Result<(), SnapshotError> { + txn.query_one_raw(statement( + "SELECT prepare_id FROM mst2_metadata_prepare WHERE prepare_id=$1 FOR UPDATE", + [prepare_id.into()], + )) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("fixed metadata prepare is missing"))?; + Ok(()) +} + +#[cfg(test)] +#[path = "native_metadata_generation_tests.rs"] +mod tests; diff --git a/src/jupiter/storage/native_metadata_install.rs b/src/jupiter/storage/native_metadata_install.rs index 659989c4..d4be4cad 100644 --- a/src/jupiter/storage/native_metadata_install.rs +++ b/src/jupiter/storage/native_metadata_install.rs @@ -101,6 +101,7 @@ pub enum MetadataPrepareObservation { struct StoredPlan { record: mst2_metadata_prepare::Model, plan: MetadataInstallPlan, + prepare_pages: Vec, } impl StoredPlan { @@ -511,6 +512,17 @@ impl PostgresMetadataInstallRepository { ) -> Result { self.barrier(txn).await?; let stored = require_plan(txn, intent).await?; + self.finalize_stored_plan_in_txn(txn, intent, dag, stored) + .await + } + + async fn finalize_stored_plan_in_txn( + &self, + txn: &DatabaseTransaction, + intent: &MetadataPrepareIntent, + dag: &ValidatedMetadataDag, + stored: StoredPlan, + ) -> Result { let expected: BTreeSet<_> = dag .payloads() .iter() @@ -768,7 +780,7 @@ async fn load_plan( .await .map_err(internal)?; let mut coverage = BTreeSet::new(); - for row in rows { + for row in &rows { let page: [u8; 32] = row .page_id .as_slice() @@ -781,7 +793,11 @@ async fn load_plan( "stored metadata preparation coverage differs from its canonical plan", )); } - Ok(Some(StoredPlan { record, plan })) + Ok(Some(StoredPlan { + record, + plan, + prepare_pages: rows, + })) } async fn require_plan( @@ -1034,6 +1050,11 @@ fn unavailable(message: &str) -> SnapshotError { SnapshotError::new(SnapshotErrorCode::ObjectUnavailable, message) } +// 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"] +pub mod generations; + #[cfg(test)] #[path = "native_metadata_install_tests.rs"] mod tests;