diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index 3a7c8390..5c85ffbd 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -85,6 +85,9 @@ mod durable_sessions; #[path = "snapshot_generation_upgrade_tests.rs"] mod generation_upgrade; +#[path = "snapshot_generation_history_fixture.rs"] +mod generation_history_fixture; + #[derive(Default)] struct ReadCounts { whole: AtomicUsize, diff --git a/src/api/router/snapshot_generation_history_fixture.rs b/src/api/router/snapshot_generation_history_fixture.rs new file mode 100644 index 00000000..9a694424 --- /dev/null +++ b/src/api/router/snapshot_generation_history_fixture.rs @@ -0,0 +1,31 @@ +use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement}; + +pub(super) async fn restore_empty_g1_schema(db: &DatabaseConnection) { + let count = db + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT count(*)::bigint AS n FROM mst2_metadata_current", + )) + .await + .unwrap() + .unwrap() + .try_get::("", "n") + .unwrap(); + assert_eq!( + count, 0, + "upgrade fixture has no generation-bound incarnation to erase" + ); + db.execute_unprepared( + "DROP TABLE mst2_metadata_current; + DROP FUNCTION mst2_metadata_current_guard(); + DROP INDEX idx_mst2_metadata_lifetime_node_generation; + ALTER TABLE mst2_metadata_lifetime DROP CONSTRAINT mst2_metadata_lifetime_pkey, + ADD PRIMARY KEY(page_id),ADD UNIQUE(node_id); + ALTER TABLE mst2_metadata_prepare DROP CONSTRAINT mst2_metadata_prepare_state_check, + DROP CONSTRAINT mst2_metadata_prepare_terminal_check, + ADD CONSTRAINT mst2_metadata_prepare_state_check CHECK (state IN ('PREPARING','COMMITTED')), + ADD CONSTRAINT mst2_metadata_prepare_check CHECK ((state='COMMITTED')=(committed_at IS NOT NULL)), + DROP COLUMN graph_domain,DROP COLUMN aborted_at,DROP COLUMN coverage_retired_at; + DELETE FROM seaql_migrations WHERE version='m20261007_000300_add_mst2_metadata_lifetime_history'" + ).await.unwrap(); +} diff --git a/src/api/router/snapshot_generation_upgrade_tests.rs b/src/api/router/snapshot_generation_upgrade_tests.rs index c2afda08..9f14f96d 100644 --- a/src/api/router/snapshot_generation_upgrade_tests.rs +++ b/src/api/router/snapshot_generation_upgrade_tests.rs @@ -42,6 +42,7 @@ async fn mst2_generation_additive_upgrade_preserves_legacy_v3_sid_lease_and_new_ ); let pages = scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await; let original = success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + super::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 // deployed schema; the SID, lease, CAS bytes and all original guards survive. diff --git a/src/callisto/mod.rs b/src/callisto/mod.rs index 8eb43be5..8bdddc74 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_current; pub mod mst2_metadata_lifetime; pub mod mst2_metadata_payload; pub mod mst2_metadata_prepare; diff --git a/src/callisto/mst2_metadata_current.rs b/src/callisto/mst2_metadata_current.rs new file mode 100644 index 00000000..0c759a1c --- /dev/null +++ b/src/callisto/mst2_metadata_current.rs @@ -0,0 +1,13 @@ +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "mst2_metadata_current")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub page_id: Vec, + pub generation: i64, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/mst2_metadata_lifetime.rs b/src/callisto/mst2_metadata_lifetime.rs index 9807c27a..d9bc99da 100644 --- a/src/callisto/mst2_metadata_lifetime.rs +++ b/src/callisto/mst2_metadata_lifetime.rs @@ -6,6 +6,7 @@ pub struct Model { #[sea_orm(primary_key, auto_increment = false)] pub page_id: Vec, pub node_id: String, + #[sea_orm(primary_key, auto_increment = false)] pub generation: i64, pub state: String, pub metadata_codec: i16, diff --git a/src/callisto/mst2_metadata_prepare.rs b/src/callisto/mst2_metadata_prepare.rs index 98b017a8..c586cf0d 100644 --- a/src/callisto/mst2_metadata_prepare.rs +++ b/src/callisto/mst2_metadata_prepare.rs @@ -12,6 +12,7 @@ pub struct Model { pub bindings_digest: Option>, pub primary_scope: Option>, pub storage_seal: Option>, + pub graph_domain: Option, pub source_domain: String, pub tagged_root_tree_oid: String, pub scope: String, @@ -29,6 +30,8 @@ pub struct Model { pub state: String, pub created_at: DateTimeWithTimeZone, pub committed_at: Option, + pub aborted_at: Option, + pub coverage_retired_at: Option, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/src/jupiter/migration/m20261007_000300_add_mst2_metadata_lifetime_history.rs b/src/jupiter/migration/m20261007_000300_add_mst2_metadata_lifetime_history.rs new file mode 100644 index 00000000..51dcee8f --- /dev/null +++ b/src/jupiter/migration/m20261007_000300_add_mst2_metadata_lifetime_history.rs @@ -0,0 +1,67 @@ +//! Preserve historical incarnations and explicit preparation terminal states. + +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( + "ALTER TABLE mst2_metadata_lifetime + DROP CONSTRAINT mst2_metadata_lifetime_pkey, + DROP CONSTRAINT mst2_metadata_lifetime_node_id_key, + ADD PRIMARY KEY(page_id,generation); + CREATE TABLE mst2_metadata_current ( + page_id bytea PRIMARY KEY CHECK (octet_length(page_id)=32), + generation bigint NOT NULL CHECK (generation>0), + FOREIGN KEY(page_id,generation) REFERENCES mst2_metadata_lifetime(page_id,generation) + ); + INSERT INTO mst2_metadata_current(page_id,generation) + SELECT page_id,generation FROM mst2_metadata_lifetime; + CREATE FUNCTION mst2_metadata_current_guard() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'metadata current watermark requires generation-fenced collection'; END $$; + CREATE TRIGGER mst2_metadata_current_guard BEFORE UPDATE OR DELETE + ON mst2_metadata_current FOR EACH ROW EXECUTE FUNCTION mst2_metadata_current_guard(); + ALTER TABLE mst2_metadata_prepare + ADD COLUMN graph_domain text CHECK (graph_domain IN ('generic-v1','qualified-v1')), + ADD COLUMN aborted_at timestamptz, + ADD COLUMN coverage_retired_at timestamptz, + DROP CONSTRAINT mst2_metadata_prepare_state_check, + DROP CONSTRAINT mst2_metadata_prepare_check, + ADD CONSTRAINT mst2_metadata_prepare_state_check CHECK (state IN ('PREPARING','COMMITTED','ABORTED')), + ADD CONSTRAINT mst2_metadata_prepare_terminal_check CHECK ( + (state='COMMITTED')=(committed_at IS NOT NULL) + AND (state='ABORTED')=(aborted_at IS NOT NULL) + AND (coverage_retired_at IS NULL OR state='COMMITTED') + AND (state<>'ABORTED' OR storage_seal IS NOT NULL) + AND (graph_domain IS NULL OR storage_seal IS NOT NULL) + ); + CREATE OR REPLACE 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 + OR NEW.graph_domain IS DISTINCT FROM OLD.graph_domain THEN + RAISE EXCEPTION 'metadata generation seal cannot be rebound'; + END IF; + IF OLD.state IN ('COMMITTED','ABORTED') AND NEW.state IS DISTINCT FROM OLD.state THEN + RAISE EXCEPTION 'metadata preparation terminal state cannot be revived'; + END IF; + IF OLD.committed_at IS NOT NULL AND NEW.committed_at IS DISTINCT FROM OLD.committed_at + OR OLD.aborted_at IS NOT NULL AND NEW.aborted_at IS DISTINCT FROM OLD.aborted_at + OR OLD.coverage_retired_at IS NOT NULL AND NEW.coverage_retired_at IS DISTINCT FROM OLD.coverage_retired_at THEN + RAISE EXCEPTION 'metadata preparation terminal receipt cannot be rewritten'; + END IF; + RETURN NEW; + END $$; + CREATE INDEX idx_mst2_metadata_lifetime_node_generation ON mst2_metadata_lifetime(node_id,generation)" + ).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 6ceda285..4a302ee3 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -147,6 +147,7 @@ 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 m20261007_000300_add_mst2_metadata_lifetime_history; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -282,6 +283,7 @@ impl MigratorTrait for Migrator { 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), + Box::new(m20261007_000300_add_mst2_metadata_lifetime_history::Migration), ] } } @@ -1192,7 +1194,7 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 10..names.len() - 1], + &names[names.len() - 11..names.len() - 2], &[ "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), @@ -1207,9 +1209,13 @@ mod tests { "native retention, metadata installation and view tables follow media paging" ); assert_eq!( - names.last().unwrap(), + &names[names.len() - 2], "m20261007_000200_add_mst2_metadata_generations" ); + assert_eq!( + names.last().unwrap(), + "m20261007_000300_add_mst2_metadata_lifetime_history" + ); let db = alias_db().await; insert_repo(&db, 1, "/third-party//a").await; diff --git a/src/jupiter/storage/native_metadata_generations.rs b/src/jupiter/storage/native_metadata_generations.rs index f7957de9..29189469 100644 --- a/src/jupiter/storage/native_metadata_generations.rs +++ b/src/jupiter/storage/native_metadata_generations.rs @@ -6,6 +6,33 @@ use super::*; const MAX_BINDINGS_BYTES: usize = 12 + 48 * 4096; const MAX_PRIMARY_SCOPE_BYTES: usize = 16384; +const GENERIC_GRAPH_DOMAIN: &str = "generic-v1"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum GraphDomain { + LegacyGeneric, + Generic, + Qualified, +} + +impl GraphDomain { + fn from_stored(domain: Option<&str>) -> Result { + match domain { + None => Ok(Self::LegacyGeneric), + Some(GENERIC_GRAPH_DOMAIN) => Ok(Self::Generic), + Some("qualified-v1") => Ok(Self::Qualified), + _ => Err(integrity("unknown stored metadata graph domain")), + } + } + + fn stored(self) -> Option<&'static str> { + match self { + Self::LegacyGeneric => None, + Self::Generic => Some(GENERIC_GRAPH_DOMAIN), + Self::Qualified => Some("qualified-v1"), + } + } +} #[derive(Debug, Clone, PartialEq, Eq)] pub struct GenerationPrepareIntent { @@ -13,8 +40,9 @@ pub struct GenerationPrepareIntent { metadata_root: [u8; 32], root_generation: i64, bindings_digest: [u8; 32], - primary_scope: Vec, + primary_scope: Box<[u8]>, storage_seal: [u8; 32], + graph_domain: GraphDomain, } impl GenerationPrepareIntent { @@ -125,12 +153,14 @@ struct FixedPlan { #[derive(Clone)] pub struct PostgresMetadataGenerationRepository { inner: PostgresMetadataInstallRepository, + graph_domain: &'static str, } impl PostgresMetadataGenerationRepository { pub async fn new(connection: DatabaseConnection) -> Result { Ok(Self { inner: PostgresMetadataInstallRepository::new(connection).await?, + graph_domain: GENERIC_GRAPH_DOMAIN, }) } @@ -149,7 +179,7 @@ impl PostgresMetadataGenerationRepository { 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 bindings = allocate_lifetimes(&txn, &plan, self.graph_domain).await?; let canonical_bindings = bindings.encode()?; let bindings_digest = Sha256::digest(&canonical_bindings).into(); let prepare_id = uuid::Uuid::new_v4().to_string(); @@ -158,16 +188,17 @@ impl PostgresMetadataGenerationRepository { 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, + storage_seal: seal(&prepare_id, &digest, &plan.root, &bindings_digest, &primary_scope,Some(self.graph_domain))?, + primary_scope:primary_scope.into_boxed_slice(), + graph_domain:GraphDomain::from_stored(Some(self.graph_domain))?, }; 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)", + node_count,edge_count,total_bytes,state,canonical_bindings,bindings_digest,primary_scope,storage_seal,graph_domain) + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,'PREPARING',$19,$20,$21,$22,$23)", [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(), @@ -175,7 +206,7 @@ impl PostgresMetadataGenerationRepository { (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()], + intent.bindings_digest.to_vec().into(),intent.primary_scope.to_vec().into(),intent.storage_seal.to_vec().into(),self.graph_domain.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(); @@ -380,24 +411,38 @@ impl PostgresMetadataGenerationRepository { .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, + "SELECT p.bindings_digest,p.primary_scope,p.storage_seal,p.graph_domain,p.coverage_retired_at IS NOT NULL AS retired, 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_current c ON c.page_id=l.page_id AND c.generation=l.generation LEFT JOIN mst2_metadata_payload b ON b.page_id=m.page_id - WHERE p.prepare_id=$1", + WHERE p.prepare_id=$1 AND c.page_id IS NOT NULL", [receipt.intent.prepare_id().into(),(MAX_BINDINGS_BYTES as i32).into()], )).await.map_err(internal)?.ok_or_else(|| unavailable("metadata generation receipt is missing"))?; + let domain = row + .try_get::>("", "graph_domain") + .map_err(internal)?; + if GraphDomain::from_stored(domain.as_deref())? != receipt.intent.graph_domain { + return Err(integrity( + "metadata generation receipt graph domain differs from storage", + )); + } + if row.try_get::("", "retired").map_err(internal)? { + return Err(unavailable( + "metadata generation receipt coverage was retired", + )); + } 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()), + ("primary_scope", receipt.intent.primary_scope.as_ref()), ("storage_seal", receipt.intent.storage_seal.as_slice()), ] { if row @@ -525,6 +570,9 @@ impl PostgresMetadataGenerationRepository { let Some(stored) = load_plan(connection, operation_id, digest).await? else { return Ok(None); }; + if stored.record.coverage_retired_at.is_some() { + return Err(unavailable("metadata preparation coverage was retired")); + } let canonical = stored .record .canonical_bindings @@ -543,6 +591,7 @@ impl PostgresMetadataGenerationRepository { &stored.plan.root, &bindings_digest, &primary_scope, + stored.record.graph_domain.as_deref(), )?; if stored.record.bindings_digest.as_deref() != Some(bindings_digest.as_slice()) || stored.record.storage_seal.as_deref() != Some(storage_seal.as_slice()) @@ -558,8 +607,9 @@ impl PostgresMetadataGenerationRepository { .ok_or_else(|| integrity("root lifetime is missing"))? .0, bindings_digest, - primary_scope, + primary_scope: primary_scope.into_boxed_slice(), storage_seal, + graph_domain: GraphDomain::from_stored(stored.record.graph_domain.as_deref())?, }; self.require_scope(&intent)?; let mut actual = BTreeMap::new(); @@ -602,11 +652,16 @@ impl PostgresMetadataGenerationRepository { } fn require_scope(&self, intent: &GenerationPrepareIntent) -> Result<(), SnapshotError> { - if intent.primary_scope != self.primary_scope()? { + if intent.primary_scope.as_ref() != self.primary_scope()?.as_slice() { return Err(integrity( "metadata seal belongs to another captured primary storage scope", )); } + if intent.graph_domain.stored().unwrap_or(GENERIC_GRAPH_DOMAIN) != self.graph_domain { + return Err(integrity( + "metadata seal belongs to another immutable graph domain", + )); + } Ok(()) } } @@ -617,13 +672,24 @@ fn seal( root: &[u8; 32], bindings: &[u8; 32], scope: &[u8], + graph_domain: Option<&str>, ) -> 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"); + match graph_domain { + None => hash.update(b"MST2-METADATA-STORAGE-SEAL-1\0"), + Some(domain) => { + if ![GENERIC_GRAPH_DOMAIN, "qualified-v1"].contains(&domain) { + return Err(integrity("unknown metadata graph domain")); + } + hash.update(b"MST2-METADATA-STORAGE-SEAL-2\0"); + hash.update((domain.len() as u32).to_be_bytes()); + hash.update(domain.as_bytes()); + } + } hash.update(id.as_bytes()); hash.update(manifest); hash.update(root); @@ -636,6 +702,7 @@ fn seal( async fn allocate_lifetimes( txn: &DatabaseTransaction, plan: &MetadataInstallPlan, + graph_domain: &str, ) -> Result { let pages: Vec<_> = plan .pages @@ -647,9 +714,26 @@ async fn allocate_lifetimes( "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", + ON CONFLICT(page_id,generation) DO NOTHING", [(plan.identity.metadata_codec as i16).into(),encoded.clone().into()], )).await.map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_metadata_current(page_id,generation) + SELECT decode(p.page_id,'hex'),1 FROM jsonb_to_recordset($1::jsonb) AS p(page_id text,size integer) + ON CONFLICT(page_id) DO NOTHING", + [encoded.clone().into()], + )).await.map_err(internal)?; + if txn.query_one_raw(statement( + "SELECT p.page_id FROM jsonb_to_recordset($1::jsonb) AS p(page_id text,size integer) + JOIN mst2_metadata_prepare_page m ON m.page_id=decode(p.page_id,'hex') + JOIN mst2_metadata_current c ON c.page_id=m.page_id AND c.generation=m.generation + JOIN mst2_metadata_prepare q ON q.prepare_id=m.prepare_id + WHERE coalesce(q.graph_domain,'generic-v1')<>$2 + AND (q.state='PREPARING' OR (q.state='COMMITTED' AND q.coverage_retired_at IS NULL)) LIMIT 1", + [encoded.clone().into(),graph_domain.into()], + )).await.map_err(internal)?.is_some() { + return Err(unavailable("metadata lifetime is covered by another graph domain")); + } 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, @@ -658,7 +742,8 @@ async fn allocate_lifetimes( 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') + JOIN mst2_metadata_current c ON c.page_id=decode(p.page_id,'hex') + JOIN mst2_metadata_lifetime l ON l.page_id=c.page_id AND l.generation=c.generation LEFT JOIN mst2_retention_node n ON n.node_id=l.node_id LEFT JOIN mst2_metadata_payload b ON b.page_id=l.page_id ORDER BY l.page_id", [encoded.into()], @@ -748,9 +833,10 @@ async fn check_lifetimes( 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 + FROM mst2_metadata_prepare_page p LEFT JOIN mst2_metadata_lifetime l ON l.page_id=p.page_id AND l.generation=p.generation + LEFT JOIN mst2_metadata_current c ON c.page_id=p.page_id AND c.generation=p.generation 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 + AND (l.page_id IS NULL OR c.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) @@ -864,3 +950,6 @@ async fn lock_prepare(txn: &DatabaseTransaction, prepare_id: &str) -> Result<(), #[cfg(test)] #[path = "native_metadata_generation_tests.rs"] mod tests; + +#[path = "native_metadata_history.rs"] +pub mod history; diff --git a/src/jupiter/storage/native_metadata_history.rs b/src/jupiter/storage/native_metadata_history.rs new file mode 100644 index 00000000..e093a4cc --- /dev/null +++ b/src/jupiter/storage/native_metadata_history.rs @@ -0,0 +1,357 @@ +//! Explicit preparation termination. Payload deletion remains forbidden. + +use super::*; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetadataTerminalAction { + Abort, + RetireCoverage, +} + +#[derive(Debug, thiserror::Error)] +pub enum MetadataTerminalError { + #[error(transparent)] + Rejected(#[from] SnapshotError), + #[error("metadata {action:?} commit outcome is unknown for operation {operation_id}")] + CommitUncertain { + operation_id: String, + manifest_digest: [u8; 32], + action: MetadataTerminalAction, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MetadataTerminalReceipt { + intent: GenerationPrepareIntent, + action: MetadataTerminalAction, + terminal_at: sea_orm::prelude::DateTimeWithTimeZone, +} + +impl MetadataTerminalReceipt { + pub fn intent(&self) -> &GenerationPrepareIntent { + &self.intent + } + + pub fn action(&self) -> MetadataTerminalAction { + self.action + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MetadataTerminalObservation { + Active, + Terminated(Box), +} + +impl PostgresMetadataGenerationRepository { + pub async fn abort( + &self, + intent: &GenerationPrepareIntent, + ) -> Result { + self.terminate(intent, MetadataTerminalAction::Abort, None) + .await + } + + pub async fn retire_prepare_coverage( + &self, + receipt: &GenerationMetadataReceipt, + ) -> Result { + self.terminate( + &receipt.intent, + MetadataTerminalAction::RetireCoverage, + Some(receipt), + ) + .await + } + + async fn terminate( + &self, + intent: &GenerationPrepareIntent, + action: MetadataTerminalAction, + receipt: Option<&GenerationMetadataReceipt>, + ) -> Result { + self.require_scope(intent)?; + let txn = self.inner.transaction().await?; + let result = self.terminate_in_txn(&txn, intent, action, receipt).await; + match result { + Ok(receipt) => { + txn.commit().await.map_err(|_| uncertain(intent, action))?; + Ok(receipt) + } + Err(error) => { + txn.rollback().await.map_err(internal)?; + Err(error.into()) + } + } + } + + async fn terminate_in_txn( + &self, + txn: &DatabaseTransaction, + intent: &GenerationPrepareIntent, + action: MetadataTerminalAction, + receipt: Option<&GenerationMetadataReceipt>, + ) -> Result { + self.inner.barrier(txn).await?; + lock_prepare(txn, intent.prepare_id()).await?; + let record = self.require_terminal_record(txn, intent).await?; + if let Some(terminal) = terminal_receipt(&record, intent, action)? { + verify_no_prepare_coverage(txn, intent).await?; + return Ok(terminal); + } + let expected = match action { + MetadataTerminalAction::Abort => "PREPARING", + MetadataTerminalAction::RetireCoverage => "COMMITTED", + }; + if record.state != expected { + return Err(SnapshotError::new( + SnapshotErrorCode::Conflict, + "metadata prepare is in another terminal state", + )); + } + let fixed = self.require_fixed_plan(txn, intent).await?; + if action == MetadataTerminalAction::RetireCoverage { + let receipt = receipt + .ok_or_else(|| integrity("coverage retirement requires a definitive receipt"))?; + if receipt.legacy != fixed.stored.receipt()? || receipt.intent != fixed.intent { + return Err(integrity( + "coverage retirement receipt differs from its fixed preparation", + )); + } + check_generation_payloads(txn, &fixed).await?; + verify_graph(txn, &fixed.stored).await?; + } + let roots = mst2_retention_root::Entity::find() + .filter( + mst2_retention_root::Column::RootKey.eq(format!("prepare:{}", intent.prepare_id())), + ) + .limit((MetadataDagLimits::default().nodes + 1) as u64) + .all(txn) + .await + .map_err(internal)?; + let expected_nodes: BTreeSet<_> = fixed.bindings.0.keys().map(node_id).collect(); + if roots + .iter() + .any(|root| root.root_kind != "prepare" || !expected_nodes.contains(&root.node_id)) + { + return Err(integrity( + "terminal prepare coverage differs from its fixed lifetime mappings", + )); + } + if action == MetadataTerminalAction::RetireCoverage + && roots + .iter() + .map(|root| root.node_id.clone()) + .collect::>() + != expected_nodes + { + return Err(unavailable( + "coverage retirement cannot recover missing or transferred prepare roots", + )); + } + txn.execute_raw(statement( + "DELETE FROM mst2_retention_root r USING mst2_metadata_prepare_page p + WHERE p.prepare_id=$1 AND r.root_key='prepare:'||p.prepare_id AND r.root_kind='prepare' + AND r.node_id='page:sha256:'||encode(p.page_id,'hex')", + [intent.prepare_id().into()], + )) + .await + .map_err(internal)?; + let sql=match action { + MetadataTerminalAction::Abort=> + "UPDATE mst2_metadata_prepare SET state='ABORTED',aborted_at=clock_timestamp() + WHERE prepare_id=$1 AND state='PREPARING' AND storage_seal=$2 RETURNING *", + MetadataTerminalAction::RetireCoverage=> + "UPDATE mst2_metadata_prepare SET coverage_retired_at=clock_timestamp() + WHERE prepare_id=$1 AND state='COMMITTED' AND coverage_retired_at IS NULL AND storage_seal=$2 RETURNING *", + }; + let row = txn + .query_one_raw(statement( + sql, + [ + intent.prepare_id().into(), + intent.storage_seal.to_vec().into(), + ], + )) + .await + .map_err(internal)? + .ok_or_else(|| integrity("metadata terminal CAS changed during its barrier"))?; + let column = match action { + MetadataTerminalAction::Abort => "aborted_at", + MetadataTerminalAction::RetireCoverage => "coverage_retired_at", + }; + let terminal_at = row.try_get("", column).map_err(internal)?; + Ok(MetadataTerminalReceipt { + intent: intent.clone(), + action, + terminal_at, + }) + } + + /// A fresh same-primary barrier resolves a lost terminal commit response. + /// An old receipt is replayed without consulting or changing a newer lifetime. + pub async fn inspect_terminal( + &self, + fresh_primary: &DatabaseConnection, + intent: &GenerationPrepareIntent, + action: MetadataTerminalAction, + ) -> Result { + self.require_scope(intent)?; + let txn = fresh_primary + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .map_err(|_| uncertain(intent, action))?; + if self.inner.barrier(&txn).await.is_err() { + let _ = txn.rollback().await; + return Err(uncertain(intent, action)); + } + let result = async { + let record = self.require_terminal_record(&txn, intent).await?; + match terminal_receipt(&record, intent, action)? { + Some(receipt) => { + verify_no_prepare_coverage(&txn, intent).await?; + Ok(MetadataTerminalObservation::Terminated(Box::new(receipt))) + } + None => Ok(MetadataTerminalObservation::Active), + } + } + .await; + txn.rollback() + .await + .map_err(|_| uncertain(intent, action))?; + result.map_err(|error: SnapshotError| { + if error.code == SnapshotErrorCode::Internal { + uncertain(intent, action) + } else { + MetadataTerminalError::Rejected(error) + } + }) + } + + async fn require_terminal_record( + &self, + connection: &C, + intent: &GenerationPrepareIntent, + ) -> Result { + self.require_scope(intent)?; + let record = mst2_metadata_prepare::Entity::find() + .filter(mst2_metadata_prepare::Column::OperationId.eq(intent.operation_id())) + .one(connection) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("sealed metadata preparation is missing"))?; + let canonical = record + .canonical_bindings + .as_deref() + .ok_or_else(|| unavailable("legacy preparation has no generation seal"))?; + let plan = MetadataInstallPlan::decode(&record.canonical_plan, &intent.manifest_digest())?; + let bindings = GenerationBindings::decode(canonical, &plan)?; + let actual_digest: [u8; 32] = Sha256::digest(canonical).into(); + let domain = GraphDomain::from_stored(record.graph_domain.as_deref())?; + if record.prepare_id != intent.prepare_id() + || record.manifest_digest.as_slice() != intent.manifest_digest() + || record.metadata_root.as_slice() != intent.metadata_root + || record.bindings_digest.as_deref() != Some(intent.bindings_digest.as_slice()) + || actual_digest != intent.bindings_digest + || record.primary_scope.as_deref() != Some(intent.primary_scope.as_ref()) + || record.storage_seal.as_deref() != Some(intent.storage_seal.as_slice()) + || domain != intent.graph_domain + || bindings.0.get(&intent.metadata_root).map(|pair| pair.0) + != Some(intent.root_generation) + || seal( + intent.prepare_id(), + &intent.manifest_digest(), + &intent.metadata_root, + &intent.bindings_digest, + &intent.primary_scope, + domain.stored(), + )? != intent.storage_seal + { + return Err(integrity( + "terminal metadata action differs from its immutable generation seal", + )); + } + let pages = mst2_metadata_prepare_page::Entity::find() + .filter(mst2_metadata_prepare_page::Column::PrepareId.eq(intent.prepare_id())) + .limit((MetadataDagLimits::default().nodes + 1) as u64) + .all(connection) + .await + .map_err(internal)?; + let mut actual = BTreeMap::new(); + for page in pages { + let id: [u8; 32] = page.page_id.as_slice().try_into().map_err(internal)?; + actual.insert( + id, + ( + page.generation + .ok_or_else(|| integrity("terminal preparation has an unbound mapping"))?, + page.expected_size as u64, + ), + ); + } + if actual != bindings.0 { + return Err(integrity( + "terminal metadata mappings differ from the fixed seal", + )); + } + Ok(record) + } +} + +fn terminal_receipt( + record: &mst2_metadata_prepare::Model, + intent: &GenerationPrepareIntent, + action: MetadataTerminalAction, +) -> Result, SnapshotError> { + let terminal_at = match action { + MetadataTerminalAction::Abort if record.state == "ABORTED" => Some( + record + .aborted_at + .ok_or_else(|| integrity("aborted metadata preparation has no terminal receipt"))?, + ), + MetadataTerminalAction::RetireCoverage if record.state == "COMMITTED" => { + record.coverage_retired_at + } + _ => None, + }; + Ok(terminal_at.map(|terminal_at| MetadataTerminalReceipt { + intent: intent.clone(), + action, + terminal_at, + })) +} + +async fn verify_no_prepare_coverage( + connection: &C, + intent: &GenerationPrepareIntent, +) -> Result<(), SnapshotError> { + if connection + .query_one_raw(statement( + "SELECT node_id FROM mst2_retention_root WHERE root_key='prepare:'||$1 LIMIT 1", + [intent.prepare_id().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(integrity( + "terminal metadata preparation still has prepare coverage", + )); + } + Ok(()) +} + +fn uncertain( + intent: &GenerationPrepareIntent, + action: MetadataTerminalAction, +) -> MetadataTerminalError { + MetadataTerminalError::CommitUncertain { + operation_id: intent.operation_id().into(), + manifest_digest: intent.manifest_digest(), + action, + } +} + +#[cfg(test)] +#[path = "native_metadata_history_tests.rs"] +mod tests; diff --git a/src/jupiter/storage/native_metadata_history_tests.rs b/src/jupiter/storage/native_metadata_history_tests.rs new file mode 100644 index 00000000..250d99a6 --- /dev/null +++ b/src/jupiter/storage/native_metadata_history_tests.rs @@ -0,0 +1,634 @@ +use sea_orm::{Database, PaginatorTrait}; +use sea_orm_migration::MigratorTrait; + +use super::*; +use crate::{ + callisto::{mst2_metadata_current, mst2_metadata_lifetime}, + jupiter::{ + migration::Migrator, + tests::{TestSchemaGuard, test_db_config}, + }, +}; + +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 prepared() -> PreparedNativeMetadataRetention { + let entries = [mst2_codec::metapage::Entry::file( + mst2_codec::metapage::EntryKind::Regular, + b"file", + 3, + [42; 32], + )]; + let child = Page::build(&entries).unwrap(); + let root_entries = [mst2_codec::metapage::Entry::dir(b"one", page_id(&child))]; + let root = Page::build(&root_entries).unwrap(); + let mut builder = crate::ceres::snapshot::retention_dag::MetadataDagBuilder::new( + MetadataDagLimits::default(), + ); + builder.add_directory(&child, &entries).unwrap(); + builder.add_directory(&root, &root_entries).unwrap(); + PreparedNativeMetadataRetention::test_installation( + std::sync::Arc::new(builder.finish(page_id(&root)).unwrap()), + "/", + ) +} + +async fn scalar(db: &DatabaseConnection, sql: &str) -> i64 { + db.query_one_raw(statement(sql, [])) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +fn rejected(error: MetadataTerminalError) -> SnapshotErrorCode { + match error { + MetadataTerminalError::Rejected(error) => error.code, + _ => panic!("expected definite terminal rejection"), + } +} + +#[tokio::test] +async fn history_keeps_old_composite_fk_bindings_while_current_is_an_independent_watermark() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository.begin_intent("history", &pages).await.unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let page = &pages.dag().payloads()[0]; + second.execute_raw(statement( + "INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size) + VALUES($1,$2,2,'RESERVED',1,$3)", + [page.id.to_vec().into(),node_id(&page.id).into(),(page.size as i32).into()], + )).await.unwrap(); + assert!( + mst2_metadata_lifetime::Entity::find_by_id((page.id.to_vec(), 1)) + .one(&second) + .await + .unwrap() + .is_some() + ); + assert!( + mst2_metadata_lifetime::Entity::find_by_id((page.id.to_vec(), 2)) + .one(&second) + .await + .unwrap() + .is_some() + ); + assert_eq!( + mst2_metadata_current::Entity::find_by_id(page.id.to_vec()) + .one(&second) + .await + .unwrap() + .unwrap() + .generation, + 1 + ); + assert!( + second + .execute_raw(statement( + "UPDATE mst2_metadata_current SET generation=2 WHERE page_id=$1", + [page.id.to_vec().into()] + )) + .await + .is_err() + ); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_prepare_page p JOIN mst2_metadata_lifetime l ON l.page_id=p.page_id AND l.generation=p.generation WHERE p.generation=1").await,2); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_payload b JOIN mst2_metadata_lifetime l ON l.page_id=b.page_id AND l.generation=b.generation WHERE b.generation=1").await,2); + repository.finalize(&intent).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_lifetime").await, + 3 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_current").await, + 2 + ); +} + +#[tokio::test] +async fn history_abort_fences_a_late_installer_waiting_on_the_real_retention_barrier() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let other = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository + .begin_intent("abort-late-install", &pages) + .await + .unwrap(); + repository + .install_page(&intent, &pages.dag().payloads()[0]) + .await + .unwrap(); + let held = repository.inner.transaction().await.unwrap(); + repository.inner.barrier(&held).await.unwrap(); + let late_intent = intent.clone(); + let late_pages = pages.dag().payloads().to_vec(); + let late = tokio::spawn(async move { other.install_pages(&late_intent, &late_pages).await }); + let deadline = tokio::time::Instant::now() + Duration::from_secs(4); + loop { + let waiting=held.query_one_raw(statement( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND NOT granted + AND classid=$1::integer::oid AND objid=hashtext(current_schema())::oid + AND database=(SELECT oid FROM pg_database WHERE datname=current_database())) AS waiting", + [RETENTION_LOCK_KEY.into()], + )).await.unwrap().unwrap().try_get::("","waiting").unwrap(); + if waiting { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "late installer must wait on the captured schema barrier" + ); + tokio::task::yield_now().await; + } + let terminal = repository + .terminate_in_txn(&held, &intent, MetadataTerminalAction::Abort, None) + .await + .unwrap(); + held.commit().await.unwrap(); + assert!(matches!( + late.await.unwrap(), + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::IntegrityError, + .. + })) + )); + assert_eq!(repository.abort(&intent).await.unwrap(), terminal); + assert_eq!( + repository + .inspect_terminal(&second, &intent, MetadataTerminalAction::Abort) + .await + .unwrap(), + MetadataTerminalObservation::Terminated(Box::new(terminal)) + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_prepare_page").await, + 2 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_node").await, + 0 + ); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_prepare WHERE state='ABORTED' AND aborted_at IS NOT NULL").await,1); +} + +#[tokio::test] +async fn history_abort_after_observation_rejects_old_finalize_and_preserves_other_prepares() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let other = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let pages = prepared(); + let owner = repository + .begin_intent("committed-owner", &pages) + .await + .unwrap(); + repository + .install_pages(&owner, pages.dag().payloads()) + .await + .unwrap(); + let owner_receipt = repository.finalize(&owner).await.unwrap(); + let abandoned = other + .begin_intent("abandoned-shared", &pages) + .await + .unwrap(); + let observation = other.observe_installed_dag(&abandoned).await.unwrap(); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_root").await, + 4 + ); + other.abort(&abandoned).await.unwrap(); + assert!(matches!( + other.finalize_observation(&abandoned, &observation).await, + Err(MetadataInstallError::Rejected(_)) + )); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_root").await, + 2 + ); + assert_eq!(repository.finalize(&owner).await.unwrap(), owner_receipt); + assert_eq!( + rejected(repository.abort(&owner).await.unwrap_err()), + SnapshotErrorCode::Conflict + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_lifetime WHERE state='LIVE' AND generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar(&second, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); +} + +#[tokio::test] +async fn history_terminal_outer_rollback_keeps_preparing_protection_and_durable_recovery_active() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository + .begin_intent("abort-rollback", &pages) + .await + .unwrap(); + let held = repository.inner.transaction().await.unwrap(); + repository + .terminate_in_txn(&held, &intent, MetadataTerminalAction::Abort, None) + .await + .unwrap(); + held.rollback().await.unwrap(); + assert_eq!( + repository + .inspect_terminal(&second, &intent, MetadataTerminalAction::Abort) + .await + .unwrap(), + MetadataTerminalObservation::Active + ); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_prepare WHERE state='PREPARING' AND aborted_at IS NULL").await,1); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + repository.finalize(&intent).await.unwrap(); +} + +#[tokio::test] +async fn history_retirement_replays_original_receipt_and_never_recreates_lost_prepare_coverage() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository.begin_intent("retire", &pages).await.unwrap(); + repository + .install_pages(&intent, pages.dag().payloads()) + .await + .unwrap(); + let receipt = repository.finalize(&intent).await.unwrap(); + let terminal = repository.retire_prepare_coverage(&receipt).await.unwrap(); + assert_eq!( + repository.retire_prepare_coverage(&receipt).await.unwrap(), + terminal + ); + assert_eq!( + repository + .inspect_terminal(&second, &intent, MetadataTerminalAction::RetireCoverage) + .await + .unwrap(), + MetadataTerminalObservation::Terminated(Box::new(terminal)) + ); + assert!(matches!( + repository + .install_pages(&intent, pages.dag().payloads()) + .await, + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::ObjectUnavailable, + .. + })) + )); + assert!(repository.finalize(&intent).await.is_err()); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); + 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' AND coverage_retired_at IS NOT NULL").await,1); + assert!( + second + .execute_unprepared("UPDATE mst2_metadata_prepare SET coverage_retired_at=NULL") + .await + .is_err() + ); +} + +#[tokio::test] +async fn history_committed_missing_payload_rejects_a_late_installer_without_repairing_corruption() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository + .begin_intent("committed-missing", &pages) + .await + .unwrap(); + repository + .install_page(&intent, &pages.dag().payloads()[0]) + .await + .unwrap(); + // Persist a corrupt COMMITTED fixture without deleting bytes or disabling + // any guard. A late installer must refuse to fill its missing page. + let txn = second.begin().await.unwrap(); + repository.inner.barrier(&txn).await.unwrap(); + PostgresRetentionRepository::retain_group_in_txn( + &txn, + pages.dag().nodes(), + pages.dag().edges(), + &[RetentionRoot::Prepare(intent.prepare_id().into())], + ) + .await + .unwrap(); + txn.execute_unprepared("UPDATE mst2_metadata_lifetime SET state='LIVE'") + .await + .unwrap(); + txn.execute_raw(statement("UPDATE mst2_metadata_prepare SET state='COMMITTED',committed_at=clock_timestamp() WHERE prepare_id=$1",[intent.prepare_id().into()])).await.unwrap(); + txn.commit().await.unwrap(); + assert!(matches!( + repository + .install_pages(&intent, pages.dag().payloads()) + .await, + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::ObjectUnavailable, + .. + })) + )); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 1 + ); + assert_eq!( + scalar(&second, "SELECT count(*) FROM mst2_retention_root").await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE state='COMMITTED'" + ) + .await, + 1 + ); +} + +#[tokio::test] +async fn history_graph_domain_is_durable_sealed_and_cannot_be_changed_by_a_new_repository() { + let (first, second, _schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let mut qualified = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + qualified.graph_domain = "qualified-v1"; + let pages = prepared(); + let intent = repository + .begin_intent("generic-domain", &pages) + .await + .unwrap(); + assert_eq!(intent.graph_domain, GraphDomain::Generic); + assert!(matches!( + qualified.begin_intent("generic-domain", &pages).await, + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::IntegrityError, + .. + })) + )); + assert!(matches!( + qualified.begin_intent("cross-domain-page", &pages).await, + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::ObjectUnavailable, + .. + })) + )); + assert!(matches!( + qualified + .install_pages(&intent, pages.dag().payloads()) + .await, + Err(MetadataInstallError::Rejected(SnapshotError { + code: SnapshotErrorCode::IntegrityError, + .. + })) + )); + assert!( + second + .execute_unprepared("UPDATE mst2_metadata_prepare SET graph_domain='qualified-v1'") + .await + .is_err() + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE graph_domain='generic-v1'" + ) + .await, + 1 + ); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 0 + ); + assert_eq!( + scalar(&second, "SELECT count(*) FROM mst2_metadata_current").await, + 2 + ); +} + +#[tokio::test] +async fn history_terminal_wrong_primary_is_unknown_and_current_delete_payload_update_stay_forbidden() + { + let (first, second, _schema) = fixture().await; + let (other, _other_connection, _other_schema) = fixture().await; + let repository = PostgresMetadataGenerationRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let intent = repository + .begin_intent("terminal-primary", &pages) + .await + .unwrap(); + repository + .install_page(&intent, &pages.dag().payloads()[0]) + .await + .unwrap(); + assert!(matches!( + repository + .inspect_terminal(&other, &intent, MetadataTerminalAction::Abort) + .await, + Err(MetadataTerminalError::CommitUncertain { .. }) + )); + assert_eq!( + repository + .inspect_terminal(&second, &intent, MetadataTerminalAction::Abort) + .await + .unwrap(), + MetadataTerminalObservation::Active + ); + for sql in [ + "DELETE FROM mst2_metadata_current", + "DELETE FROM mst2_metadata_lifetime", + "DELETE FROM mst2_metadata_payload", + "UPDATE mst2_metadata_payload SET payload=payload", + "DELETE FROM mst2_metadata_storage_scope", + ] { + assert!(second.execute_unprepared(sql).await.is_err(), "{sql}"); + } + repository.abort(&intent).await.unwrap(); + assert!( + second + .execute_unprepared( + "UPDATE mst2_metadata_prepare SET state='PREPARING',aborted_at=NULL" + ) + .await + .is_err() + ); + assert_eq!( + mst2_metadata_current::Entity::find() + .count(&first) + .await + .unwrap(), + 2 + ); +} + +#[tokio::test] +async fn history_migration_preserves_preexisting_g1_null_domain_seal_and_composite_mappings() { + let temp = tempfile::tempdir().unwrap(); + let (config, _schema) = test_db_config(temp.path()).await; + let first = Database::connect(config.db_url.clone()).await.unwrap(); + let second = Database::connect(config.db_url).await.unwrap(); + let at = Migrator::migrations() + .iter() + .position(|migration| { + migration.name() == "m20261007_000300_add_mst2_metadata_lifetime_history" + }) + .unwrap(); + Migrator::up(&first, Some(at.try_into().unwrap())) + .await + .unwrap(); + let inner = PostgresMetadataInstallRepository::new(first.clone()) + .await + .unwrap(); + let pages = prepared(); + let plan = pages.install_plan().unwrap(); + let manifest = plan.digest().unwrap(); + let bindings = GenerationBindings( + plan.pages + .iter() + .map(|(page, size)| (*page, (1, *size))) + .collect(), + ); + let canonical = bindings.encode().unwrap(); + let digest: [u8; 32] = Sha256::digest(&canonical).into(); + let scope = serde_json::to_vec(&( + &inner.storage_scope.storage_uuid, + &inner.storage_scope.database, + inner.storage_scope.database_oid, + &inner.storage_scope.schema, + inner.storage_scope.schema_oid, + &inner.storage_scope.server_address, + inner.storage_scope.server_port, + )) + .unwrap(); + let prepare_id = uuid::Uuid::new_v4().to_string(); + let old_seal = seal(&prepare_id, &manifest, &plan.root, &digest, &scope, None).unwrap(); + let txn = first.begin().await.unwrap(); + inner.barrier(&txn).await.unwrap(); + for page in pages.dag().payloads() { + txn.execute_raw(statement("INSERT INTO mst2_metadata_lifetime(page_id,node_id,generation,state,metadata_codec,expected_size) VALUES($1,$2,1,'RESERVED',1,$3)", + [page.id.to_vec().into(),node_id(&page.id).into(),(page.size as i32).into()])).await.unwrap(); + txn.execute_raw(statement("INSERT INTO mst2_metadata_payload(page_id,generation,metadata_codec,byte_size,payload) VALUES($1,1,1,$2,$3)", + [page.id.to_vec().into(),(page.size as i32).into(),page.bytes.clone().into()])).await.unwrap(); + } + 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,'g1-before-history',$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,'PREPARING',$18,$19,$20,$21)", + [prepare_id.clone().into(),manifest.to_vec().into(),plan.encode().unwrap().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.into(),digest.to_vec().into(),scope.into(),old_seal.to_vec().into()], + )).await.unwrap(); + for (page, size) in &plan.pages { + txn.execute_raw(statement("INSERT INTO mst2_metadata_prepare_page(prepare_id,page_id,generation,expected_size) VALUES($1,$2,1,$3)", + [prepare_id.clone().into(),page.to_vec().into(),(*size as i32).into()])).await.unwrap(); + } + PostgresRetentionRepository::retain_group_in_txn( + &txn, + pages.dag().nodes(), + pages.dag().edges(), + &[RetentionRoot::Prepare(prepare_id.clone())], + ) + .await + .unwrap(); + txn.execute_unprepared("UPDATE mst2_metadata_lifetime SET state='LIVE'") + .await + .unwrap(); + txn.execute_raw(statement("UPDATE mst2_metadata_prepare SET state='COMMITTED',committed_at=clock_timestamp() WHERE prepare_id=$1",[prepare_id.clone().into()])).await.unwrap(); + txn.commit().await.unwrap(); + Migrator::up(&first, Some(1)).await.unwrap(); + let repository = PostgresMetadataGenerationRepository::new(second.clone()) + .await + .unwrap(); + let restored = repository + .begin_intent("g1-before-history", &pages) + .await + .unwrap(); + assert_eq!(restored.graph_domain, GraphDomain::LegacyGeneric); + assert_eq!(restored.storage_seal, old_seal); + assert_eq!(restored.prepare_id(), prepare_id); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_current WHERE generation=1" + ) + .await, + 2 + ); + assert_eq!( + scalar( + &first, + "SELECT count(*) FROM mst2_metadata_prepare WHERE graph_domain IS NULL" + ) + .await, + 1 + ); + let receipt = repository.finalize(&restored).await.unwrap(); + repository.retire_prepare_coverage(&receipt).await.unwrap(); + assert_eq!(scalar(&first,"SELECT count(*) FROM mst2_metadata_prepare_page p JOIN mst2_metadata_lifetime l ON l.page_id=p.page_id AND l.generation=p.generation").await,2); + assert_eq!( + scalar(&first, "SELECT count(*) FROM mst2_metadata_payload").await, + 2 + ); +}