diff --git a/.github/workflows/repository-gates.yml b/.github/workflows/repository-gates.yml index 313fc5a5..0e26b36f 100644 --- a/.github/workflows/repository-gates.yml +++ b/.github/workflows/repository-gates.yml @@ -66,6 +66,44 @@ jobs: docker compose -p mega2-it -f docker/docker-compose.test.yml --profile git \ up -d --build --wait postgres redis rustfs rustfs-init git-cli + - name: Verify necessary v3 publication and performance regressions + shell: bash + run: | + set -euo pipefail + source .env.test + export no_proxy="$NO_PROXY" + test_inventory="$(cargo test --lib -- --list)" + tests=( + api::router::snapshot_router::content::tests::rooted_metadata::descriptor_wire::rooted_postgres_descriptor_matches_codec_bytes_for_root_ascii_long_and_utf8_scopes + jupiter::migration::tests::import_repo_alias_rows_canonicalized + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::committed_metadata_requests_retire_owners_without_retiring_source_history + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::terminal_owner_survives_its_deferred_completion_and_prunes_in_a_later_transaction + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::reader_pruning_respects_the_64_owner_budget_with_a_committed_backlog + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::stale_actual_reader_cannot_read_or_finish_a_reissued_uuid + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::rolled_back_reader_issuance_is_unpublished_and_extreme_issuance_fails_closed + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::current_reader_retention_migration_verifies_fresh_family_without_rewriting_history + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::captured_previous_q_upgrade_preserves_old_sid_source_proofs_and_active_legacy_reader + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::captured_previous_core_without_q_upgrades_before_new_provisioning + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::captured_previous_tampered_q_rejects_upgrade_without_partial_issuance_schema + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_upgrade_preserves_nonzero_reader_issuance_all_owners_oids_and_old_sid + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_interrupted_before_reader_migration_resumes_without_reader_ddl + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_qless_canonical_policy_upgrades_before_new_provisioning + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_qless_exact_legacy_deparse_policy_upgrades_before_new_provisioning + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_tampered_decoder_rejects_upgrade_without_rewriting_owners_or_policy + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::captured_cc90_qless_arbitrary_policy_shape_is_never_resigned + api::router::snapshot_router::content::tests::mst2_rooted_streamed_fact_persists_one_source_pass_and_reuses_warm_alias_facts + jupiter::storage::qualified_metadata_family::tests::canonical_tests::actual_rooted_cold_and_zero_delta_reuse_keep_one_canonical_graph + jupiter::storage::qualified_metadata_family::tests::canonical_tests::changed_ancestor_installs_only_delta_and_accepts_independently_attested_source_aliases + api::router::snapshot_router::content::tests::rooted_metadata::rooted_wide_directory_windows_and_lookup_survive_rebuild_with_valid_proofs + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::descriptor_wire::captured_87b_descriptor_upgrade_preserves_owners_oids_and_replays_missing_ledgers + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::descriptor_wire::captured_87b_qless_descriptor_upgrade_replays_missing_ledgers_before_provisioning + api::router::snapshot_router::content::tests::rooted_metadata::reader_retention::upgrade::native_runtime::descriptor_wire::captured_87b_tampered_descriptor_rejects_without_resigning_history + ) + for test_name in "${tests[@]}"; do + grep -Fqx "$test_name: test" <<< "$test_inventory" + done + cargo test --lib -- --exact --nocapture "${tests[@]}" + - name: Run repository tests with test environment shell: bash run: | diff --git a/src/api/router/snapshot_descriptor_upgrade_tests.rs b/src/api/router/snapshot_descriptor_upgrade_tests.rs new file mode 100644 index 00000000..9cb37b3b --- /dev/null +++ b/src/api/router/snapshot_descriptor_upgrade_tests.rs @@ -0,0 +1,181 @@ +use super::*; +use crate::jupiter::storage::qualified_metadata_family::reader_previous_fixture::restore_native_runtime; + +async fn remove_upgrade_ledgers(core: &sea_orm::DatabaseConnection) { + core.execute_unprepared( + "DELETE FROM seaql_migrations WHERE version IN ( + 'm20261008_000400_add_mst2_reader_retention', + 'm20261008_000500_fix_mst2_native_runtime', + 'm20261008_000600_fix_mst2_descriptor_wire')", + ) + .await + .unwrap(); +} + +async fn assert_upgrade_ledgers(core: &sea_orm::DatabaseConnection) { + let count: i64 = core + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT count(*) FROM seaql_migrations WHERE version IN ( + 'm20261008_000400_add_mst2_reader_retention', + 'm20261008_000500_fix_mst2_native_runtime', + 'm20261008_000600_fix_mst2_descriptor_wire')", + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + assert_eq!(count, 3); +} + +async fn descriptor_matches_stored(fixture: &Fixture) -> i64 { + q_count( + fixture, + "SELECT count(*) FROM {q}.mst2_qualified_session_incarnation session + WHERE session.state='READY' AND session.canonical_descriptor= + {q}.mst2_metadata_descriptor(session.prepare_id,session.instance_id, + session.commit_oid,session.root_tree_oid)", + ) + .await +} + +#[tokio::test] +async fn captured_87b_descriptor_upgrade_preserves_owners_oids_and_replays_missing_ledgers() { + let fixture = Fixture::new_with_pg_config(true).await; + seed_live_and_terminal(&fixture).await; + let schema = q_schema(&fixture).await; + let mono = fixture.state.storage.mono_storage(); + let core = mono.get_connection(); + for missing_ledgers in [false, true] { + let rows = permanent_rows(&fixture).await; + let owners = retained_owners(&fixture).await; + restore_native_runtime(core, &schema, true).await; + assert_eq!(descriptor_matches_stored(&fixture).await, 0); + assert_eq!(permanent_rows(&fixture).await, rows); + assert_eq!(retained_owners(&fixture).await, owners); + if missing_ledgers { + remove_upgrade_ledgers(core).await; + apply_migrations(core, false).await.unwrap(); + assert_upgrade_ledgers(core).await; + } else { + test_upgrade_native_runtime(core).await.unwrap(); + } + assert_eq!(descriptor_matches_stored(&fixture).await, 1); + assert_eq!(permanent_rows(&fixture).await, rows); + assert_eq!(retained_owners(&fixture).await, owners); + let stamp = policy(core).await; + if missing_ledgers { + apply_migrations(core, false).await.unwrap(); + } else { + test_upgrade_native_runtime(core).await.unwrap(); + } + assert_eq!(policy(core).await, stamp); + assert_eq!(permanent_rows(&fixture).await, rows); + assert_eq!(retained_owners(&fixture).await, owners); + assert_old_sid_works(&fixture).await; + } + let rows = permanent_rows(&fixture).await; + let owners = retained_owners(&fixture).await; + let stamp = policy(core).await; + core.execute_unprepared( + "DELETE FROM seaql_migrations WHERE version IN ( + 'm20261008_000400_add_mst2_reader_retention', + 'm20261008_000500_fix_mst2_native_runtime')", + ) + .await + .unwrap(); + apply_migrations(core, false).await.unwrap(); + assert_upgrade_ledgers(core).await; + assert_eq!(descriptor_matches_stored(&fixture).await, 1); + assert_eq!(policy(core).await, stamp); + assert_eq!(permanent_rows(&fixture).await, rows); + assert_eq!(retained_owners(&fixture).await, owners); + assert_old_sid_works(&fixture).await; +} + +#[tokio::test] +async fn captured_87b_qless_descriptor_upgrade_replays_missing_ledgers_before_provisioning() { + let temp = tempfile::tempdir().unwrap(); + let (config, _schema) = test_db_config(temp.path()).await; + let core = crate::jupiter::storage::init::database_connection(&config) + .await + .unwrap(); + let schema: String = core + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT metadata_schema FROM mst2_metadata_namespace WHERE graph_domain='qualified-v1'", + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + restore_native_runtime(&core, &schema, false).await; + remove_upgrade_ledgers(&core).await; + apply_migrations(&core, false).await.unwrap(); + assert_upgrade_ledgers(&core).await; + let stamp = policy(&core).await; + apply_migrations(&core, false).await.unwrap(); + test_upgrade_native_runtime(&core).await.unwrap(); + assert_eq!(policy(&core).await, stamp); + let namespace = provision_or_verify_rooted_qualified_family(&core) + .await + .unwrap(); + let current_schema: String = core + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT metadata_schema FROM mst2_metadata_namespace WHERE graph_domain='qualified-v1'", + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + assert_ne!(current_schema, schema); + assert_eq!( + provision_or_verify_rooted_qualified_family(&core) + .await + .unwrap(), + namespace + ); + let high_water: i64 = core + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + format!( + "SELECT high_water FROM \"{}\".mst2_metadata_reader_issuance", + current_schema.replace('"', "\"\"") + ), + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + assert_eq!(high_water, 0); +} + +#[tokio::test] +async fn captured_87b_tampered_descriptor_rejects_without_resigning_history() { + let fixture = Fixture::new_with_pg_config(true).await; + seed_live_and_terminal(&fixture).await; + let schema = q_schema(&fixture).await; + let mono = fixture.state.storage.mono_storage(); + let core = mono.get_connection(); + restore_native_runtime(core, &schema, true).await; + let rows = permanent_rows(&fixture).await; + let owners = retained_owners(&fixture).await; + let before = policy(core).await; + core.execute_unprepared(&format!( + "CREATE OR REPLACE FUNCTION \"{}\".mst2_metadata_descriptor( + pid text,instance text,commit_id text,tree_id text) RETURNS bytea + LANGUAGE sql AS 'SELECT decode(''0000'',''hex'')'", + schema.replace('"', "\"\"") + )) + .await + .unwrap(); + assert!(test_upgrade_native_runtime(core).await.is_err()); + assert_eq!(policy(core).await, before); + assert_eq!(permanent_rows(&fixture).await, rows); + assert_eq!(retained_owners(&fixture).await, owners); +} diff --git a/src/api/router/snapshot_descriptor_wire_tests.rs b/src/api/router/snapshot_descriptor_wire_tests.rs new file mode 100644 index 00000000..f5d5536e --- /dev/null +++ b/src/api/router/snapshot_descriptor_wire_tests.rs @@ -0,0 +1,118 @@ +use mst2_codec::descriptor::ServingDescriptor; + +use super::*; + +async fn assert_postgres_descriptor_matches_codec(fixture: &Fixture, scope: &str) { + let resolved = success_json( + fixture + .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":scope}).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(), + ) + .await; + let snapshot = resolved["descriptor"]["snapshot_id"].as_str().unwrap(); + let lease = resolved["lease_id"].as_str().unwrap(); + assert_eq!( + fixture + .state + .storage + .snapshot_metadata_family(lease, true) + .await + .unwrap(), + Some(SnapshotMetadataFamily::Rooted) + ); + let context = fixture + .state + .storage + .snapshot_context(snapshot, lease) + .await + .unwrap(); + let descriptor = &context.built.descriptor; + assert_eq!(descriptor.scope, scope); + let codec_bytes = descriptor.encode().unwrap(); + let schema = q_schema(fixture).await; + let quoted = format!("\"{}\"", schema.replace('"', "\"\"")); + let row = fixture + .state + .storage + .mono_storage() + .get_connection() + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + format!( + "SELECT {q}.mst2_metadata_descriptor(session.prepare_id,session.instance_id, + session.commit_oid,session.root_tree_oid) AS descriptor, + session.canonical_descriptor,session.metadata_root,session.snapshot_id + FROM {q}.mst2_qualified_session_incarnation session + JOIN {q}.mst2_qualified_lease_binding lease USING(snapshot_id,session_incarnation) + WHERE session.snapshot_id=$1 AND lease.lease_id=$2 AND session.state='READY' + AND lease.state='ACTIVE'", + q = quoted, + ), + [snapshot.into(), lease.into()], + )) + .await + .unwrap() + .unwrap(); + let database_bytes: Vec = row.try_get("", "descriptor").unwrap(); + assert_eq!(database_bytes, codec_bytes, "scope {scope:?}"); + assert_eq!( + row.try_get::>("", "canonical_descriptor").unwrap(), + codec_bytes + ); + assert_eq!( + ServingDescriptor::decode(&database_bytes).unwrap(), + *descriptor + ); + assert_eq!( + &database_bytes[56..58], + &u16::try_from(scope.len()).unwrap().to_le_bytes() + ); + assert_eq!(&database_bytes[58..58 + scope.len()], scope.as_bytes()); + assert_eq!( + row.try_get::>("", "metadata_root").unwrap(), + descriptor.metadata_root.to_vec() + ); + assert_eq!( + resolved["descriptor"]["metadata_root"], + format!("sha256:{}", hex::encode(descriptor.metadata_root)) + ); + let codec_snapshot = format!("sha256:{}", hex::encode(descriptor.snapshot_id().unwrap())); + assert_eq!(snapshot, codec_snapshot); + assert_eq!(context.built.snapshot_id, codec_snapshot); + assert_eq!( + row.try_get::("", "snapshot_id").unwrap(), + codec_snapshot + ); +} + +#[tokio::test] +async fn rooted_postgres_descriptor_matches_codec_bytes_for_root_ascii_long_and_utf8_scopes() { + let component = "a".repeat(247); + let fixture = Fixture::new_with_pg_config_directories_and_objects( + true, + 0, + &[(format!("{component}/é/file"), b"descriptor wire".to_vec())], + ) + .await; + let long_scope = format!("/project/{component}"); + let utf8_scope = format!("{long_scope}/é"); + assert_eq!(long_scope.len(), 256); + assert_eq!(utf8_scope.len(), 259); + assert_ne!(utf8_scope.len(), utf8_scope.chars().count()); + for scope in ["/", "/project", long_scope.as_str(), utf8_scope.as_str()] { + assert_postgres_descriptor_matches_codec(&fixture, scope).await; + } +} diff --git a/src/api/router/snapshot_native_runtime_upgrade_tests.rs b/src/api/router/snapshot_native_runtime_upgrade_tests.rs index 79a76175..2fdee7d6 100644 --- a/src/api/router/snapshot_native_runtime_upgrade_tests.rs +++ b/src/api/router/snapshot_native_runtime_upgrade_tests.rs @@ -4,6 +4,9 @@ use crate::jupiter::{ storage::qualified_metadata_family::reader_previous_fixture::restore_retention, }; +#[path = "snapshot_descriptor_upgrade_tests.rs"] +mod descriptor_wire; + async fn retained_owners(fixture: &Fixture) -> Value { let txn = transaction(fixture).await; let value = txn.query_one_raw(Statement::from_string(DbBackend::Postgres, @@ -135,10 +138,11 @@ async fn captured_cc90_interrupted_before_reader_migration_resumes_without_reade let owners = retained_owners(&fixture).await; core.execute_raw(Statement::from_sql_and_values( DbBackend::Postgres, - "DELETE FROM seaql_migrations WHERE version IN ($1,$2)", + "DELETE FROM seaql_migrations WHERE version IN ($1,$2,$3)", [ "m20261008_000400_add_mst2_reader_retention".into(), "m20261008_000500_fix_mst2_native_runtime".into(), + "m20261008_000600_fix_mst2_descriptor_wire".into(), ], )) .await @@ -148,9 +152,10 @@ async fn captured_cc90_interrupted_before_reader_migration_resumes_without_reade assert_eq!(retained_owners(&fixture).await, owners); let count: i64 = core.query_one_raw(Statement::from_string(DbBackend::Postgres, "SELECT count(*) FROM seaql_migrations WHERE version IN - ('m20261008_000400_add_mst2_reader_retention','m20261008_000500_fix_mst2_native_runtime')")) + ('m20261008_000400_add_mst2_reader_retention','m20261008_000500_fix_mst2_native_runtime', + 'm20261008_000600_fix_mst2_descriptor_wire')")) .await.unwrap().unwrap().try_get_by_index(0).unwrap(); - assert_eq!(count, 2); + assert_eq!(count, 3); assert_old_sid_works(&fixture).await; } diff --git a/src/api/router/snapshot_rooted_metadata_tests.rs b/src/api/router/snapshot_rooted_metadata_tests.rs index 46316703..33179e3d 100644 --- a/src/api/router/snapshot_rooted_metadata_tests.rs +++ b/src/api/router/snapshot_rooted_metadata_tests.rs @@ -10,6 +10,9 @@ use crate::jupiter::storage::qualified_metadata_family::{ #[path = "snapshot_reader_retention_tests.rs"] mod reader_retention; +#[path = "snapshot_descriptor_wire_tests.rs"] +mod descriptor_wire; + async fn q_schema(fixture: &Fixture) -> String { fixture .state diff --git a/src/jupiter/migration/m20261008_000600_fix_mst2_descriptor_wire.rs b/src/jupiter/migration/m20261008_000600_fix_mst2_descriptor_wire.rs new file mode 100644 index 00000000..7efcf97f --- /dev/null +++ b/src/jupiter/migration/m20261008_000600_fix_mst2_descriptor_wire.rs @@ -0,0 +1,21 @@ +//! Repair canonical descriptor integer encoding without rewriting ownership. + +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> { + super::qualified_native_runtime_upgrade::upgrade_descriptor(manager).await + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } + + fn use_transaction(&self) -> Option { + Some(true) + } +} diff --git a/src/jupiter/migration/m20261008_000600_previous_descriptor.sql b/src/jupiter/migration/m20261008_000600_previous_descriptor.sql new file mode 100644 index 00000000..dab92001 --- /dev/null +++ b/src/jupiter/migration/m20261008_000600_previous_descriptor.sql @@ -0,0 +1,27 @@ +CREATE FUNCTION mst2_metadata_descriptor(pid text,instance text,commit_id text,tree_id text) +RETURNS bytea LANGUAGE plpgsql VOLATILE STRICT SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE p record; scope_bytes bytea; view_digest bytea; instance_bytes bytea; +BEGIN + SELECT tagged_root_tree_oid,scope,metadata_root INTO p FROM mst2_metadata_prepare WHERE prepare_id=pid AND plan_kind='ROOTED' + AND state='COMMITTED' AND graph_domain='qualified-v1' AND mst2_metadata_scope_matches(primary_scope); + IF NOT FOUND OR instance IS DISTINCT FROM (instance::uuid)::text + OR split_part(p.tagged_root_tree_oid,':',2) IS DISTINCT FROM tree_id + OR NOT EXISTS(SELECT 1 FROM $CORE_SCHEMA$.mega_commit c WHERE c.commit_id=$3 AND c.tree=$4) + OR octet_length(commit_id)<>octet_length(tree_id) + OR commit_id !~ '^([0-9a-f]{40}|[0-9a-f]{64})$' THEN + RAISE EXCEPTION 'qualified descriptor has no exact canonical instance and fixed source'; + END IF; + PERFORM mst2_metadata_native_profile(pid); + scope_bytes:=convert_to(p.scope,'UTF8'); + IF p.scope !~ '^/' OR octet_length(scope_bytes)>4096 OR p.scope<>'/' AND + (p.scope ~ '/$|//|/(\.|\.\.)(/|$)' OR cardinality(string_to_array(substring(p.scope FROM 2),'/'))>256) + OR EXISTS(SELECT 1 FROM unnest(string_to_array(substring(p.scope FROM 2),'/')) component + WHERE octet_length(component)>255) THEN + RAISE EXCEPTION 'qualified descriptor scope is not canonical'; + END IF; + instance_bytes:=decode(replace(instance,'-',''),'hex'); + view_digest:=sha256(convert_to('mega.mst2.namespaceview','UTF8')||decode('00','hex')||convert_to(commit_id,'UTF8')); + RETURN convert_to('MSD2','UTF8')||decode('00020001','hex')||instance_bytes||view_digest + ||decode(lpad(to_hex(octet_length(scope_bytes)),4,'0'),'hex')||scope_bytes + ||decode('0001000100000000','hex')||p.metadata_root; +END $$; diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 09eaa04d..e3a737a5 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -156,6 +156,7 @@ mod m20261008_000200_add_mst2_rooted_qualified_family; mod m20261008_000300_add_mst2_chunk_map_retention; mod m20261008_000400_add_mst2_reader_retention; mod m20261008_000500_fix_mst2_native_runtime; +mod m20261008_000600_fix_mst2_descriptor_wire; pub(crate) mod qualified_native_runtime_upgrade; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; @@ -169,9 +170,15 @@ pub(crate) async fn test_upgrade_reader_retention( use sea_orm_migration::{MigrationTrait, SchemaManager}; let txn = connection.begin().await?; - let result = m20261008_000400_add_mst2_reader_retention::Migration - .up(&SchemaManager::new(&txn)) - .await; + let result = async { + m20261008_000400_add_mst2_reader_retention::Migration + .up(&SchemaManager::new(&txn)) + .await?; + m20261008_000600_fix_mst2_descriptor_wire::Migration + .up(&SchemaManager::new(&txn)) + .await + } + .await; match result { Ok(()) => txn.commit().await, Err(error) => { @@ -189,9 +196,15 @@ pub(crate) async fn test_upgrade_native_runtime( use sea_orm_migration::{MigrationTrait, SchemaManager}; let txn = connection.begin().await?; - let result = m20261008_000500_fix_mst2_native_runtime::Migration - .up(&SchemaManager::new(&txn)) - .await; + let result = async { + m20261008_000500_fix_mst2_native_runtime::Migration + .up(&SchemaManager::new(&txn)) + .await?; + m20261008_000600_fix_mst2_descriptor_wire::Migration + .up(&SchemaManager::new(&txn)) + .await + } + .await; match result { Ok(()) => txn.commit().await, Err(error) => { @@ -341,6 +354,7 @@ impl MigratorTrait for Migrator { Box::new(m20261008_000300_add_mst2_chunk_map_retention::Migration), Box::new(m20261008_000400_add_mst2_reader_retention::Migration), Box::new(m20261008_000500_fix_mst2_native_runtime::Migration), + Box::new(m20261008_000600_fix_mst2_descriptor_wire::Migration), ] } } @@ -1250,61 +1264,33 @@ mod tests { #[tokio::test] async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); + let expected = [ + "m20260923_000200_canonicalize_import_repo_paths", + "m20260925_000100_media_paging", + "m20261005_000100_add_mst2_publication_request_digest", + "m20261005_000200_add_mst2_native_head", + "m20261005_000100_add_mst2_retention_durability", + "m20261005_000200_harden_mst2_retention_graph", + "m20261005_000300_add_mst2_metadata_install", + VIEW_MIGRATION_NAME, + "m20261007_000100_add_mst2_snapshot_sessions", + "m20261007_000200_add_mst2_metadata_generations", + "m20261007_000300_add_mst2_metadata_lifetime_history", + "m20261007_000400_add_mst2_qualified_metadata_gc", + "m20261007_000500_add_mst2_install_capability", + "m20261007_000600_add_mst2_storage_routes", + "m20261008_000100_add_mst2_chunk_maps", + "m20261008_000200_add_mst2_rooted_qualified_family", + "m20261008_000300_add_mst2_chunk_map_retention", + "m20261008_000400_add_mst2_reader_retention", + "m20261008_000500_fix_mst2_native_runtime", + "m20261008_000600_fix_mst2_descriptor_wire", + ] + .map(str::to_owned); assert_eq!( - &names[names.len() - 18..names.len() - 9], - &[ - "m20260923_000200_canonicalize_import_repo_paths".to_string(), - "m20260925_000100_media_paging".to_string(), - "m20261005_000100_add_mst2_publication_request_digest".to_string(), - "m20261005_000200_add_mst2_native_head".to_string(), - "m20261005_000100_add_mst2_retention_durability".to_string(), - "m20261005_000200_harden_mst2_retention_graph".to_string(), - "m20261005_000300_add_mst2_metadata_install".to_string(), - VIEW_MIGRATION_NAME.to_string(), - "m20261007_000100_add_mst2_snapshot_sessions".to_string(), - ], - "native retention, metadata installation and view tables follow media paging" - ); - assert_eq!( - &names[names.len() - 9], - "m20261007_000200_add_mst2_metadata_generations" - ); - assert_eq!( - &names[names.len() - 8], - "m20261007_000300_add_mst2_metadata_lifetime_history" - ); - assert_eq!( - &names[names.len() - 7], - "m20261007_000400_add_mst2_qualified_metadata_gc" - ); - assert_eq!( - &names[names.len() - 6], - "m20261007_000500_add_mst2_install_capability" - ); - assert_eq!( - &names[names.len() - 5], - "m20261007_000600_add_mst2_storage_routes" - ); - assert_eq!( - &names[names.len() - 5], - "m20261008_000100_add_mst2_chunk_maps" - ); - assert_eq!( - &names[names.len() - 4], - "m20261008_000200_add_mst2_rooted_qualified_family" - ); - - assert_eq!( - &names[names.len() - 3], - "m20261008_000300_add_mst2_chunk_map_retention" - ); - assert_eq!( - &names[names.len() - 2], - "m20261008_000400_add_mst2_reader_retention" - ); - assert_eq!( - names.last().unwrap(), - "m20261008_000500_fix_mst2_native_runtime" + &names[names.len() - expected.len()..], + expected.as_slice(), + "the complete native migration suffix follows media paging in registered order" ); let db = alias_db().await; insert_repo(&db, 1, "/third-party//a").await; diff --git a/src/jupiter/migration/qualified_native_runtime_upgrade.rs b/src/jupiter/migration/qualified_native_runtime_upgrade.rs index 0a7d4b94..85cd256b 100644 --- a/src/jupiter/migration/qualified_native_runtime_upgrade.rs +++ b/src/jupiter/migration/qualified_native_runtime_upgrade.rs @@ -13,8 +13,11 @@ pub(crate) const PREVIOUS_IMPLEMENTATION: &str = pub(crate) const RETENTION_IMPLEMENTATION: &str = "900e7a340e7365c9f31995e070edbfe2076b870ec09fe89e4dc43e56f26b91ab"; +pub(crate) const NATIVE_RUNTIME_IMPLEMENTATION: &str = + "87b104e6b5593b1d6e5904a4e4be94e6ba0814dbb2c25dd8b38a0ea950d42307"; const PREVIOUS_DECODER: &str = include_str!("m20261008_000500_previous_rooted_decoder.sql"); const PREVIOUS_READERS: &str = include_str!("m20261008_000500_previous_reader_family.sql"); +const PREVIOUS_DESCRIPTOR: &str = include_str!("m20261008_000600_previous_descriptor.sql"); fn rejected(message: &str) -> DbErr { DbErr::Custom(message.into()) @@ -157,21 +160,33 @@ pub(crate) fn render_prior_family( storage: &str, implementation: &str, ) -> Result { - if ![PREVIOUS_IMPLEMENTATION, RETENTION_IMPLEMENTATION].contains(&implementation) { + if ![ + PREVIOUS_IMPLEMENTATION, + RETENTION_IMPLEMENTATION, + NATIVE_RUNTIME_IMPLEMENTATION, + ] + .contains(&implementation) + { return Err(rejected("native runtime template version is unsupported")); } let decoder = PREVIOUS_DECODER.replace("\r\n", "\n"); let readers = PREVIOUS_READERS.replace("\r\n", "\n"); + let descriptor = PREVIOUS_DESCRIPTOR.replace("\r\n", "\n"); if hex::encode(Sha256::digest(decoder.as_bytes())) != "f14d024cfb821f9d6b7348ab9b69f48c09862ff7dd3fbb2b360b253f949e35a9" || hex::encode(Sha256::digest(readers.as_bytes())) != "ec9b422440da111e5e047e1a4c85a28a33053cf0e320a56ea944ebefeaa8ea37" + || hex::encode(Sha256::digest(descriptor.as_bytes())) + != "c833666fcb1b3c9fd0bf186e5ad732f5e8b9b9dbf4360ce8d93a0d91d17d5c1a" { return Err(rejected("native runtime historical source capture changed")); } let mut source = render_family(core, core_oid, q, q_oid, namespace, storage).replace("\r\n", "\n"); - replace_function(&mut source, "mst2_metadata_decode_rooted_plan", &decoder)?; + replace_function(&mut source, "mst2_metadata_descriptor", &descriptor)?; + if implementation != NATIVE_RUNTIME_IMPLEMENTATION { + replace_function(&mut source, "mst2_metadata_decode_rooted_plan", &decoder)?; + } if implementation == PREVIOUS_IMPLEMENTATION { let start = source .find("CREATE TABLE mst2_metadata_reader_operation (") @@ -304,6 +319,21 @@ struct Family { pub(super) async fn upgrade( manager: &SchemaManager<'_>, allow_previous_readers: bool, +) -> Result<(), DbErr> { + upgrade_to(manager, allow_previous_readers, false).await +} + +pub(super) async fn upgrade_descriptor(manager: &SchemaManager<'_>) -> Result<(), DbErr> { + // Finish an interrupted reader/runtime upgrade before the independent + // descriptor repair, without replaying reader DDL on an admitted family. + upgrade(manager, true).await?; + upgrade_to(manager, false, true).await +} + +async fn upgrade_to( + manager: &SchemaManager<'_>, + allow_previous_readers: bool, + repair_descriptor: bool, ) -> Result<(), DbErr> { if manager.get_database_backend() != DbBackend::Postgres { return Err(rejected( @@ -368,15 +398,24 @@ pub(super) async fn upgrade( format!("SELECT implementation_fingerprint,expected_shape,authority_catalog FROM {c}.mst2_qualified_family_policy WHERE singleton=1"))) .await?.ok_or_else(|| rejected("reader migration trusted policy is missing"))?; let implementation: Vec = policy.try_get("", "implementation_fingerprint")?; - let current = implementation_fingerprint(); + let compiled = implementation_fingerprint(); + let runtime = + hex::decode(NATIVE_RUNTIME_IMPLEMENTATION).map_err(|error| rejected(&error.to_string()))?; + let current = if repair_descriptor { + compiled.clone() + } else { + runtime.clone() + }; let previous = hex::decode(PREVIOUS_IMPLEMENTATION).map_err(|error| rejected(&error.to_string()))?; let retention = hex::decode(RETENTION_IMPLEMENTATION).map_err(|error| rejected(&error.to_string()))?; - if implementation != current - && implementation != retention - && !(allow_previous_readers && implementation == previous) - { + let supported = implementation == compiled + || implementation == runtime + || !repair_descriptor + && (implementation == retention + || allow_previous_readers && implementation == previous); + if !supported { return Err(rejected( "reader migration refuses an unsupported prior implementation", )); @@ -468,7 +507,7 @@ pub(super) async fn upgrade( "native runtime migration prior policy shape is not trusted", )); } - if implementation == current { + if implementation == compiled || implementation == current { return Ok(()); } @@ -490,16 +529,34 @@ pub(super) async fn upgrade( if let Some(family) = &family { let q = identifier(&family.schema); connection.execute_unprepared(&format!("LOCK TABLE {q}.mst2_metadata_reader_operation,{q}.mst2_metadata_root_anchor IN ACCESS EXCLUSIVE MODE")).await?; - let rendered = render_family( - &core, - core_oid, - &family.schema, - family.oid, - &family.namespace, - &family.storage, - ); + let rendered = if repair_descriptor { + render_family( + &core, + core_oid, + &family.schema, + family.oid, + &family.namespace, + &family.storage, + ) + } else { + render_prior_family( + &core, + core_oid, + &family.schema, + family.oid, + &family.namespace, + &family.storage, + NATIVE_RUNTIME_IMPLEMENTATION, + )? + }; let mut functions = String::new(); - let replacements: &[&str] = if implementation == previous { + let replacements: &[&str] = if repair_descriptor { + &[ + "mst2_metadata_descriptor", + "mst2_metadata_dml_barrier", + "mst2_metadata_gc_enabled", + ] + } else if implementation == previous { &[ "mst2_metadata_decode_rooted_plan", "mst2_metadata_dml_barrier", diff --git a/src/jupiter/storage/qualified_metadata_reader_previous_fixture.rs b/src/jupiter/storage/qualified_metadata_reader_previous_fixture.rs index ba120126..a7975c05 100644 --- a/src/jupiter/storage/qualified_metadata_reader_previous_fixture.rs +++ b/src/jupiter/storage/qualified_metadata_reader_previous_fixture.rs @@ -259,9 +259,40 @@ pub(crate) async fn restore_retention( keep_q: bool, legacy_shape: bool, ) { - use crate::jupiter::migration::qualified_native_runtime_upgrade::{ - RETENTION_IMPLEMENTATION, render_prior_family, - }; + restore_reader_family( + core, + q_schema, + keep_q, + legacy_shape, + crate::jupiter::migration::qualified_native_runtime_upgrade::RETENTION_IMPLEMENTATION, + ) + .await; +} + +/// Restore the exact 87b runtime family without rebuilding any owned history. +pub(crate) async fn restore_native_runtime( + core: &DatabaseConnection, + q_schema: &str, + keep_q: bool, +) { + restore_reader_family( + core, + q_schema, + keep_q, + false, + crate::jupiter::migration::qualified_native_runtime_upgrade::NATIVE_RUNTIME_IMPLEMENTATION, + ) + .await; +} + +async fn restore_reader_family( + core: &DatabaseConnection, + q_schema: &str, + keep_q: bool, + legacy_shape: bool, + implementation: &str, +) { + use crate::jupiter::migration::qualified_native_runtime_upgrade::render_prior_family; assert!(!legacy_shape || !keep_q); let txn = core.begin().await.unwrap(); @@ -290,7 +321,7 @@ pub(crate) async fn restore_retention( oid, &namespace, &storage, - RETENTION_IMPLEMENTATION, + implementation, ) .unwrap(); txn.execute_unprepared(&format!("SET LOCAL search_path={q},pg_catalog,pg_temp")) @@ -298,6 +329,7 @@ pub(crate) async fn restore_retention( .unwrap(); for name in [ "mst2_metadata_decode_rooted_plan", + "mst2_metadata_descriptor", "mst2_metadata_dml_barrier", "mst2_metadata_gc_enabled", ] { @@ -319,7 +351,7 @@ pub(crate) async fn restore_retention( &capture[a..z] .replacen("CREATE FUNCTION", "CREATE OR REPLACE FUNCTION", 1) .replace("$CORE_SCHEMA$", &c) - .replace("$IMPLEMENTATION_SHA$", RETENTION_IMPLEMENTATION), + .replace("$IMPLEMENTATION_SHA$", implementation), ) .await .unwrap(); @@ -386,7 +418,7 @@ pub(crate) async fn restore_retention( } else { 0 }, - "Q-less cc90 fixture must have no history to discard: {name}" + "Q-less captured runtime fixture must have no history to discard: {name}" ); } let water: i64 = txn @@ -423,7 +455,7 @@ pub(crate) async fn restore_retention( txn.execute_unprepared(&format!("ALTER TABLE {c}.mst2_qualified_family_policy DISABLE TRIGGER mst2_route_family_policy_immutable")).await.unwrap(); txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, format!("UPDATE {c}.mst2_qualified_family_policy SET implementation_fingerprint=$1,expected_shape=$2,authority_catalog=$3 WHERE singleton=1"), - [hex::decode(RETENTION_IMPLEMENTATION).unwrap().into(), old_shape.into(), authority.clone().into()])).await.unwrap(); + [hex::decode(implementation).unwrap().into(), old_shape.into(), authority.clone().into()])).await.unwrap(); txn.execute_unprepared(&format!("ALTER TABLE {c}.mst2_qualified_family_policy ENABLE TRIGGER mst2_route_family_policy_immutable")).await.unwrap(); if let Some(full) = &full { txn.execute_unprepared(&format!( @@ -433,10 +465,10 @@ pub(crate) async fn restore_retention( ALTER TABLE {c}.mst2_metadata_namespace DISABLE TRIGGER mst2_route_immutable")).await.unwrap(); txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, format!("UPDATE {q}.mst2_metadata_family_identity SET implementation_fingerprint=$1 WHERE singleton=1"), - [hex::decode(RETENTION_IMPLEMENTATION).unwrap().into()])).await.unwrap(); + [hex::decode(implementation).unwrap().into()])).await.unwrap(); txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, format!("UPDATE {c}.mst2_metadata_namespace SET implementation_fingerprint=$1,catalog_fingerprint=$2 WHERE namespace_uuid=$3::uuid"), - [hex::decode(RETENTION_IMPLEMENTATION).unwrap().into(), full.clone().into(), namespace.into()])).await.unwrap(); + [hex::decode(implementation).unwrap().into(), full.clone().into(), namespace.into()])).await.unwrap(); txn.execute_unprepared(&format!( "ALTER TABLE {q}.mst2_metadata_family_identity ENABLE TRIGGER mst2_00_family_barrier; ALTER TABLE {q}.mst2_metadata_family_identity ENABLE TRIGGER mst2_metadata_identity_immutable; diff --git a/src/jupiter/storage/qualified_metadata_serving.sql b/src/jupiter/storage/qualified_metadata_serving.sql index 56ae3c8f..18ad4663 100644 --- a/src/jupiter/storage/qualified_metadata_serving.sql +++ b/src/jupiter/storage/qualified_metadata_serving.sql @@ -48,9 +48,9 @@ BEGIN END IF; instance_bytes:=decode(replace(instance,'-',''),'hex'); view_digest:=sha256(convert_to('mega.mst2.namespaceview','UTF8')||decode('00','hex')||convert_to(commit_id,'UTF8')); - RETURN convert_to('MSD2','UTF8')||decode('00020001','hex')||instance_bytes||view_digest - ||decode(lpad(to_hex(octet_length(scope_bytes)),4,'0'),'hex')||scope_bytes - ||decode('0001000100000000','hex')||p.metadata_root; + RETURN convert_to('MSD2','UTF8')||decode('02000100','hex')||instance_bytes||view_digest + ||mst2_metadata_write_le(octet_length(scope_bytes),2)||scope_bytes + ||decode('0100010000000000','hex')||p.metadata_root; END $$; CREATE FUNCTION mst2_metadata_root_live(p bytea,g bigint,c bytea) RETURNS boolean