diff --git a/src/api/router/snapshot_reader_retention_tests.rs b/src/api/router/snapshot_reader_retention_tests.rs new file mode 100644 index 00000000..546ece7a --- /dev/null +++ b/src/api/router/snapshot_reader_retention_tests.rs @@ -0,0 +1,431 @@ +use sea_orm::DatabaseTransaction; + +use super::*; + +#[path = "snapshot_reader_retention_upgrade_tests.rs"] +mod upgrade; + +async fn transaction(fixture: &Fixture) -> DatabaseTransaction { + let schema = q_schema(fixture).await; + let mono = fixture.state.storage.mono_storage(); + let txn = mono.get_connection().begin().await.unwrap(); + txn.execute_unprepared(&format!( + "SET LOCAL search_path=\"{}\",pg_catalog,pg_temp", + schema.replace('"', "\"\"") + )) + .await + .unwrap(); + txn +} + +async fn admission(txn: &DatabaseTransaction, fixture: &Fixture) -> (String, i64) { + let reader = txn.query_one_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT operation_id::text,reader_issuance FROM mst2_metadata_begin_reader($1,$2, + (SELECT instance_id FROM mst2_qualified_session_incarnation WHERE snapshot_id=$1 AND state='READY' LIMIT 1))", + [fixture.snapshot.clone().into(),fixture.lease.clone().into()])).await.unwrap().unwrap(); + ( + reader.try_get("", "operation_id").unwrap(), + reader.try_get("", "reader_issuance").unwrap(), + ) +} + +async fn finish(txn: &DatabaseTransaction, ticket: &(String, i64)) { + txn.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT mst2_metadata_finish_reader($1::uuid,$2::bigint)", + [ticket.0.clone().into(), ticket.1.into()], + )) + .await + .unwrap(); +} + +async fn prune(txn: &DatabaseTransaction, maximum: i32) -> i64 { + txn.query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT mst2_metadata_prune_readers($1)", + [maximum.into()], + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +#[tokio::test] +async fn committed_metadata_requests_retire_owners_without_retiring_source_history() { + let fixture = Fixture::new_with_pg_config(true).await; + let initial = q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance", + ) + .await; + let certificate_count = q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_page_certificate", + ) + .await; + let source_count = q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_source_root_attestation", + ) + .await; + for _ in 0..96 { + success_json( + fixture + .app + .clone() + .oneshot(fixture.request( + "POST", + "lookup", + Body::from(json!({"paths":["/file"]}).to_string()), + )) + .await + .unwrap(), + ) + .await; + } + assert_eq!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await, + initial + 96 + ); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 1 + ); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE state='ACTIVE'" + ) + .await, + 0 + ); + assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_root_anchor WHERE anchor_kind IN ('REQUEST','READER')").await,0); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_page_certificate" + ) + .await, + certificate_count + ); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_source_root_attestation" + ) + .await, + source_count + ); + success_json( + fixture + .app + .clone() + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + ) + .await; + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn terminal_owner_survives_its_deferred_completion_and_prunes_in_a_later_transaction() { + let fixture = Fixture::new_with_pg_config(true).await; + let txn = transaction(&fixture).await; + let ticket = admission(&txn, &fixture).await; + txn.commit().await.unwrap(); + for forbidden in [ + "DELETE FROM mst2_metadata_reader_operation", + "UPDATE mst2_metadata_reader_operation SET state='EXPIRED'", + "UPDATE mst2_metadata_reader_issuance SET high_water=0", + "DELETE FROM mst2_metadata_reader_issuance", + "TRUNCATE mst2_metadata_reader_issuance", + "SELECT mst2_metadata_prune_readers(65)", + ] { + let txn = transaction(&fixture).await; + assert!(txn.execute_unprepared(forbidden).await.is_err()); + txn.rollback().await.unwrap(); + } + let txn = transaction(&fixture).await; + finish(&txn, &ticket).await; + assert_eq!(prune(&txn, 64).await, 0); + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + // A terminal no-op must not enqueue another deferred row lookup. + txn.execute_unprepared("UPDATE mst2_metadata_reader_operation SET state=state") + .await + .unwrap(); + assert_eq!(prune(&txn, 0).await, 0); + assert_eq!(prune(&txn, 64).await, 1); + txn.commit().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 0 + ); + assert_eq!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await, + ticket.1 + ); +} + +#[tokio::test] +async fn reader_pruning_respects_the_64_owner_budget_with_a_committed_backlog() { + let fixture = Fixture::new_with_pg_config(true).await; + let txn = transaction(&fixture).await; + let mut tickets = Vec::new(); + for _ in 0..65 { + tickets.push(admission(&txn, &fixture).await); + } + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + for ticket in &tickets { + finish(&txn, ticket).await; + } + assert_eq!(prune(&txn, 64).await, 0); + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + assert_eq!(prune(&txn, 64).await, 64); + txn.commit().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 1 + ); + let txn = transaction(&fixture).await; + assert_eq!(prune(&txn, 64).await, 1); + txn.commit().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 0 + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn stale_actual_reader_cannot_read_or_finish_a_reissued_uuid() { + let fixture = Fixture::new_with_pg_config(true).await; + let admitted = Arc::new(Barrier::new(2)); + let resume = Arc::new(Barrier::new(2)); + let request = fixture.request( + "POST", + "metadata/pages", + Body::from( + json!({"encoding":"identity","items":[{"directory_path":"/nested"}]}).to_string(), + ), + ); + let app = fixture.app.clone(); + let pending = tokio::spawn(with_rooted_reader_barriers( + admitted.clone(), + resume.clone(), + async move { app.oneshot(request).await.unwrap() }, + )); + tokio::time::timeout(Duration::from_secs(4), admitted.wait()) + .await + .unwrap(); + let txn = transaction(&fixture).await; + let owner=txn.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT operation_id::text,reader_issuance,to_jsonb(r) AS owner FROM mst2_metadata_reader_operation r WHERE state='ACTIVE'")) + .await.unwrap().unwrap(); + let old = ( + owner.try_get::("", "operation_id").unwrap(), + owner.try_get::("", "reader_issuance").unwrap(), + ); + let old_row: Value = owner.try_get("", "owner").unwrap(); + let anchors:Value=txn.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT jsonb_agg(to_jsonb(a)) AS anchors FROM mst2_metadata_root_anchor a WHERE anchor_kind IN ('REQUEST','READER')")) + .await.unwrap().unwrap().try_get("","anchors").unwrap(); + finish(&txn, &old).await; + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + assert_eq!(prune(&txn, 64).await, 1); + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + assert!(txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "INSERT INTO mst2_metadata_reader_operation SELECT * FROM jsonb_populate_record(NULL::mst2_metadata_reader_operation,$1::jsonb)", + [old_row.clone().into()])).await.is_err()); + txn.rollback().await.unwrap(); + let next = old.1 + 1; + let txn = transaction(&fixture).await; + txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "INSERT INTO mst2_metadata_reader_operation SELECT * FROM jsonb_populate_record(NULL::mst2_metadata_reader_operation, + $1::jsonb||jsonb_build_object('reader_issuance',$2::bigint))",[old_row.into(),next.into()])).await.unwrap(); + txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "INSERT INTO mst2_metadata_root_anchor SELECT (jsonb_populate_record(NULL::mst2_metadata_root_anchor, + value||jsonb_build_object('reader_issuance',$2::bigint))).* FROM jsonb_array_elements($1::jsonb)", + [anchors.into(),next.into()])).await.unwrap(); + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + let source = txn + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT a.attestation_id::text,a.root_page,a.root_generation,a.root_certificate_digest + FROM mst2_metadata_source_root_attestation a JOIN mst2_metadata_reader_operation r + ON r.root_page=a.root_page AND r.root_generation=a.root_generation + WHERE r.operation_id=$1::uuid AND r.reader_issuance=$2 LIMIT 1", + [old.0.clone().into(), next.into()], + )) + .await + .unwrap() + .unwrap(); + let source_id: String = source.try_get("", "attestation_id").unwrap(); + let source_root: Vec = source.try_get("", "root_page").unwrap(); + let source_generation: i64 = source.try_get("", "root_generation").unwrap(); + let source_certificate: Vec = source.try_get("", "root_certificate_digest").unwrap(); + assert!(txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT * FROM mst2_metadata_read_source_entries($1::uuid,$2::bigint,$3::uuid,$4,$5,$6,'[]'::jsonb)", + [old.0.clone().into(),old.1.into(),source_id.clone().into(),source_root.clone().into(),source_generation.into(),source_certificate.clone().into()])).await.is_err()); + txn.rollback().await.unwrap(); + let txn = transaction(&fixture).await; + assert!(txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT * FROM mst2_metadata_read_source_entries($1::uuid,$2::bigint,$3::uuid,$4,$5,$6,'[]'::jsonb)", + [old.0.clone().into(),next.into(),source_id.into(),source_root.into(),source_generation.into(),source_certificate.into()])).await.unwrap().is_empty()); + txn.commit().await.unwrap(); + resume.wait().await; + let response = tokio::time::timeout(Duration::from_secs(4), pending) + .await + .unwrap() + .unwrap(); + assert!(response.status().is_server_error()); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE state='ACTIVE'" + ) + .await, + 1 + ); + assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_root_anchor WHERE anchor_kind IN ('REQUEST','READER')").await,2); + let txn = transaction(&fixture).await; + finish(&txn, &(old.0, next)).await; + txn.commit().await.unwrap(); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn rolled_back_reader_issuance_is_unpublished_and_extreme_issuance_fails_closed() { + let fixture = Fixture::new_with_pg_config(true).await; + let initial = q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance", + ) + .await; + let txn = transaction(&fixture).await; + let unpublished = admission(&txn, &fixture).await; + txn.rollback().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await, + initial + ); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 0 + ); + let txn = transaction(&fixture).await; + let published = admission(&txn, &fixture).await; + assert_eq!(published.1, unpublished.1); + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + assert_eq!( + txn.query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT mst2_metadata_next_reader_issuance(9223372036854775806)" + )) + .await + .unwrap() + .unwrap() + .try_get_by_index::(0) + .unwrap(), + i64::MAX + ); + assert!( + txn.query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT mst2_metadata_next_reader_issuance(9223372036854775807)" + )) + .await + .is_err() + ); + txn.rollback().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await, + published.1 + ); + let txn = transaction(&fixture).await; + finish(&txn, &published).await; + txn.commit().await.unwrap(); +} + +#[tokio::test] +#[ignore = "capacity soak: run explicitly against dedicated PostgreSQL"] +async fn reader_history_remains_bounded_after_more_than_65536_committed_http_requests() { + let fixture = Fixture::new_with_pg_config(true).await; + for _ in 0..65_537 { + success_json( + fixture + .app + .clone() + .oneshot(fixture.request( + "POST", + "lookup", + Body::from(json!({"paths":["/file"]}).to_string()), + )) + .await + .unwrap(), + ) + .await; + } + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation" + ) + .await, + 1 + ); + assert!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await + > 65_536 + ); + fixture.counts.assert(0, 0); +} diff --git a/src/api/router/snapshot_reader_retention_upgrade_tests.rs b/src/api/router/snapshot_reader_retention_upgrade_tests.rs new file mode 100644 index 00000000..1f798a8e --- /dev/null +++ b/src/api/router/snapshot_reader_retention_upgrade_tests.rs @@ -0,0 +1,199 @@ +use super::*; +use crate::jupiter::{ + migration::test_upgrade_reader_retention, + storage::qualified_metadata_family::{ + RootedLookupStatus, RootedQualifiedMetadataRepository, + provision_or_verify_rooted_qualified_family, reader_previous_fixture::restore_previous, + }, +}; + +async fn permanent_rows(fixture: &Fixture) -> Value { + let txn = transaction(fixture).await; + let rows=txn.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT jsonb_build_object( + 'sessions',(SELECT jsonb_agg(to_jsonb(x) ORDER BY snapshot_id,session_incarnation) FROM mst2_qualified_session_incarnation x), + 'leases',(SELECT jsonb_agg(to_jsonb(x) ORDER BY lease_id) FROM mst2_qualified_lease_binding x), + 'sources',(SELECT jsonb_agg(to_jsonb(x) ORDER BY attestation_id) FROM mst2_metadata_source_root_attestation x), + 'certificates',(SELECT jsonb_agg(to_jsonb(x) ORDER BY page_id,generation) FROM mst2_metadata_page_certificate x)) AS rows")) + .await.unwrap().unwrap().try_get("","rows").unwrap(); + txn.commit().await.unwrap(); + rows +} + +#[tokio::test] +async fn current_reader_retention_migration_verifies_fresh_family_without_rewriting_history() { + let fixture = Fixture::new_with_pg_config(true).await; + let before = permanent_rows(&fixture).await; + let high_water = q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance", + ) + .await; + let mono = fixture.state.storage.mono_storage(); + let core = mono.get_connection(); + test_upgrade_reader_retention(core).await.unwrap(); + assert_eq!(permanent_rows(&fixture).await, before); + assert_eq!( + q_count( + &fixture, + "SELECT high_water FROM {q}.mst2_metadata_reader_issuance" + ) + .await, + high_water + ); + provision_or_verify_rooted_qualified_family(core) + .await + .unwrap(); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn captured_previous_q_upgrade_preserves_old_sid_source_proofs_and_active_legacy_reader() { + let fixture = Fixture::new_with_pg_config(true).await; + let pinned = fixture + .state + .storage + .snapshot_context(&fixture.snapshot, &fixture.lease) + .await + .unwrap(); + let before = permanent_rows(&fixture).await; + let txn = transaction(&fixture).await; + let active = admission(&txn, &fixture).await; + txn.commit().await.unwrap(); + let schema = q_schema(&fixture).await; + let mono = fixture.state.storage.mono_storage(); + let core = mono.get_connection(); + restore_previous(core, &schema, true).await; + assert_eq!(permanent_rows(&fixture).await, before); + test_upgrade_reader_retention(core).await.unwrap(); + assert_eq!(permanent_rows(&fixture).await, before); + assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE reader_issuance=0 AND state='ACTIVE'").await,1); + assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_root_anchor WHERE reader_issuance=0 AND anchor_kind IN ('REQUEST','READER')").await,2); + let restarted = + RootedQualifiedMetadataRepository::open(core, &fixture.state.storage.config().database) + .await + .unwrap(); + let fixed = restarted + .context(&fixture.snapshot, &fixture.lease, &pinned.built.instance_id) + .await + .unwrap(); + assert_eq!(fixed.built.descriptor, pinned.built.descriptor); + assert_eq!(fixed.commit_oid, pinned.commit_oid); + let lookup = restarted + .lookup_metadata(&fixed, &["/file".to_owned()]) + .await + .unwrap(); + assert!(matches!( + lookup.results.as_slice(), + [RootedLookupStatus::File { .. }] + )); + assert_eq!(permanent_rows(&fixture).await, before); + assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE reader_issuance=0 AND state='ACTIVE'").await,1); + let txn = transaction(&fixture).await; + finish(&txn, &(active.0.clone(), 0)).await; + txn.commit().await.unwrap(); + let txn = transaction(&fixture).await; + assert!(prune(&txn, 64).await >= 1); + txn.commit().await.unwrap(); + assert_eq!( + q_count( + &fixture, + "SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE reader_issuance=0" + ) + .await, + 0 + ); + provision_or_verify_rooted_qualified_family(core) + .await + .unwrap(); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn captured_previous_core_without_q_upgrades_before_new_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 old_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_previous(&core, &old_schema, false).await; + test_upgrade_reader_retention(&core).await.unwrap(); + provision_or_verify_rooted_qualified_family(&core) + .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(); + assert_ne!(schema, old_schema); + let water: i64 = core + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + format!( + "SELECT high_water FROM \"{}\".mst2_metadata_reader_issuance", + schema.replace('"', "\"\"") + ), + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + assert_eq!(water, 0); + provision_or_verify_rooted_qualified_family(&core) + .await + .unwrap(); +} + +#[tokio::test] +async fn captured_previous_tampered_q_rejects_upgrade_without_partial_issuance_schema() { + let fixture = Fixture::new_with_pg_config(true).await; + let before = permanent_rows(&fixture).await; + let schema = q_schema(&fixture).await; + let mono = fixture.state.storage.mono_storage(); + let core = mono.get_connection(); + restore_previous(core, &schema, true).await; + core.execute_unprepared(&format!( + "ALTER TABLE \"{}\".mst2_metadata_reader_operation ADD COLUMN unexpected integer", + schema.replace('"', "\"\"") + )) + .await + .unwrap(); + assert!(test_upgrade_reader_retention(core).await.is_err()); + assert_eq!(permanent_rows(&fixture).await, before); + let exists: bool = core + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT to_regclass($1) IS NULL AS absent", + [format!("{schema}.mst2_metadata_reader_issuance").into()], + )) + .await + .unwrap() + .unwrap() + .try_get("", "absent") + .unwrap(); + assert!(exists); + let implementation:String=core.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT encode(implementation_fingerprint,'hex') FROM mst2_qualified_family_policy WHERE singleton=1")) + .await.unwrap().unwrap().try_get_by_index(0).unwrap(); + assert_eq!( + implementation, + "6d7095dc052e60bf6de21a987b72e23fdfbfe819355f0e847d2a7f3c1cbd3f27" + ); +} diff --git a/src/api/router/snapshot_rooted_metadata_tests.rs b/src/api/router/snapshot_rooted_metadata_tests.rs index 7c944913..46316703 100644 --- a/src/api/router/snapshot_rooted_metadata_tests.rs +++ b/src/api/router/snapshot_rooted_metadata_tests.rs @@ -7,6 +7,9 @@ use crate::jupiter::storage::qualified_metadata_family::{ with_rooted_source_temporary_shadow, }; +#[path = "snapshot_reader_retention_tests.rs"] +mod reader_retention; + async fn q_schema(fixture: &Fixture) -> String { fixture .state diff --git a/src/jupiter/migration/m20261008_000400_add_mst2_reader_retention.rs b/src/jupiter/migration/m20261008_000400_add_mst2_reader_retention.rs new file mode 100644 index 00000000..5be3fd26 --- /dev/null +++ b/src/jupiter/migration/m20261008_000400_add_mst2_reader_retention.rs @@ -0,0 +1,397 @@ +//! Upgrade only the trusted Q reader lifecycle, preserving fixed source history. + +use sea_orm::{ConnectionTrait, DbBackend, Statement, Value}; +use sea_orm_migration::prelude::*; + +use crate::jupiter::storage::qualified_metadata_family::{ + implementation_fingerprint, render_family, +}; + +const PREVIOUS_IMPLEMENTATION: &str = + "6d7095dc052e60bf6de21a987b72e23fdfbfe819355f0e847d2a7f3c1cbd3f27"; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +fn rejected(message: &str) -> DbErr { + DbErr::Custom(message.into()) +} + +fn identifier(value: &str) -> String { + format!("\"{}\"", value.replace('"', "\"\"")) +} + +fn literal(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +fn trusted_function(source: &str, name: &str) -> Result { + let prefix = format!("CREATE FUNCTION {name}("); + let start = source + .find(&prefix) + .ok_or_else(|| rejected("reader migration trusted function is missing"))?; + let function = &source[start..]; + let body = function + .find("AS $") + .ok_or_else(|| rejected("reader migration function body is missing"))? + + 3; + let tag_end = function[body + 1..] + .find('$') + .ok_or_else(|| rejected("reader migration function delimiter is missing"))? + + body + + 2; + let tag = &function[body..tag_end]; + let end = function[tag_end..] + .find(tag) + .ok_or_else(|| rejected("reader migration function terminator is missing"))? + + tag_end + + tag.len(); + if function.as_bytes().get(end) != Some(&b';') { + return Err(rejected("reader migration function terminator is invalid")); + } + Ok(function[..=end].replacen("CREATE FUNCTION", "CREATE OR REPLACE FUNCTION", 1)) +} + +async fn digest( + connection: &C, + sql: String, + values: Vec, +) -> Result, DbErr> { + connection + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + sql, + values, + )) + .await? + .ok_or_else(|| rejected("reader migration physical fingerprint is missing"))? + .try_get("", "fingerprint") +} + +async fn catalog( + connection: &C, + core_oid: i64, + q_oid: i64, + exempt_oid: i64, +) -> Result, DbErr> { + let source = include_str!("../storage/qualified_family_catalog.sql") + .replace("$CORE_OID$", "$1::bigint::oid") + .replace("$Q_OID$", "$2::bigint::oid") + .replace("$EXEMPT_Q_OID$", "$3::bigint::oid"); + digest( + connection, + source, + vec![core_oid.into(), q_oid.into(), exempt_oid.into()], + ) + .await +} + +async fn shape( + connection: &C, + q_oid: i64, + namespace: &str, + storage: &str, +) -> Result, DbErr> { + let source = include_str!("../storage/qualified_family_shape.sql") + .replace("q_oid", "$1::bigint::oid") + .replace("n_uuid", "$2::uuid") + .replace("s_uuid", "$3::text"); + digest( + connection, + source, + vec![q_oid.into(), namespace.into(), storage.into()], + ) + .await +} + +struct Family { + schema: String, + oid: i64, + namespace: String, + storage: String, + catalog: Vec, +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + if manager.get_database_backend() != DbBackend::Postgres { + return Err(rejected( + "qualified reader retention requires primary PostgreSQL", + )); + } + let connection = manager.get_connection(); + let captured = connection.query_one_raw(Statement::from_string(DbBackend::Postgres, + "SELECT n.nspname,n.oid::bigint,registry.mono_lock_key2 FROM pg_catalog.pg_namespace n + JOIN mst2_metadata_namespace registry ON registry.singleton=1 AND registry.core_schema=n.nspname + AND registry.core_schema_oid=n.oid WHERE n.nspname=current_schema()")) + .await?.ok_or_else(|| rejected("reader migration requires its captured core schema"))?; + let core: String = captured.try_get("", "nspname")?; + let core_oid: i64 = captured.try_get("", "oid")?; + let mono: i32 = captured.try_get("", "mono_lock_key2")?; + let c = identifier(&core); + connection + .execute_unprepared("SELECT pg_catalog.set_config('lock_timeout','5000ms',true)") + .await?; + connection + .execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT pg_catalog.pg_advisory_xact_lock(1297043024,$1)", + [mono.into()], + )) + .await?; + connection + .execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT pg_catalog.pg_advisory_xact_lock(1296718001,pg_catalog.hashtext($1))", + [core.clone().into()], + )) + .await?; + let namespaces = connection.query_all_raw(Statement::from_string(DbBackend::Postgres, + format!("SELECT metadata_schema FROM {c}.mst2_metadata_namespace ORDER BY namespace_uuid"))).await?; + if !(1..=2).contains(&namespaces.len()) { + return Err(rejected( + "reader migration namespace registry is not bounded", + )); + } + for namespace in namespaces { + let schema: String = namespace.try_get("", "metadata_schema")?; + connection + .execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT pg_catalog.pg_advisory_xact_lock(1296717362,pg_catalog.hashtext($1))", + [schema.into()], + )) + .await?; + } + connection.execute_unprepared(&format!( + "LOCK TABLE {c}.mst2_metadata_namespace,{c}.mst2_qualified_family_policy IN ACCESS EXCLUSIVE MODE" + )).await?; + let policy=connection.query_one_raw(Statement::from_string(DbBackend::Postgres, + 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 previous = + hex::decode(PREVIOUS_IMPLEMENTATION).map_err(|error| rejected(&error.to_string()))?; + if implementation != previous && implementation != current { + return Err(rejected( + "reader migration refuses an unsupported prior implementation", + )); + } + let rows=connection.query_all_raw(Statement::from_string(DbBackend::Postgres,format!( + "SELECT namespace_uuid::text,metadata_schema,metadata_schema_oid::bigint,metadata_storage_uuid, + implementation_fingerprint,catalog_fingerprint FROM {c}.mst2_metadata_namespace WHERE graph_domain='qualified-v1'"))).await?; + let family = match rows.as_slice() { + [] => None, + [row] => Some(Family { + schema: row.try_get("", "metadata_schema")?, + oid: row.try_get("", "metadata_schema_oid")?, + namespace: row.try_get("", "namespace_uuid")?, + storage: row.try_get("", "metadata_storage_uuid")?, + catalog: row.try_get("", "catalog_fingerprint")?, + }), + _ => return Err(rejected("reader migration requires at most one Q family")), + }; + let q_oid = family.as_ref().map_or(0, |family| family.oid); + if policy.try_get::>("", "authority_catalog")? + != catalog(connection, core_oid, 0, q_oid).await? + { + return Err(rejected( + "reader migration prior core authority catalog changed", + )); + } + let scope = connection + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + format!("SELECT {c}.mst2_route_scope_valid($1) AS valid"), + [core.clone().into()], + )) + .await? + .ok_or_else(|| rejected("reader migration core scope is missing"))?; + if !scope.try_get::("", "valid")? { + return Err(rejected("reader migration prior core scope changed")); + } + if let Some(family) = &family { + let q = identifier(&family.schema); + let identity=connection.query_one_raw(Statement::from_sql_and_values(DbBackend::Postgres,format!( + "SELECT EXISTS(SELECT 1 FROM {c}.mst2_metadata_namespace n + JOIN {c}.mst2_metadata_namespace g ON g.singleton=1 + JOIN {q}.mst2_metadata_family_identity i ON i.singleton=1 + JOIN {q}.mst2_metadata_storage_scope s ON s.singleton=1 + JOIN pg_catalog.pg_namespace physical ON physical.oid=n.metadata_schema_oid AND physical.nspname=n.metadata_schema + WHERE n.namespace_uuid=$1::uuid AND n.metadata_schema='mst2q_'||replace(n.namespace_uuid::text,'-','') + AND n.singleton IS NULL AND n.family_identity='v3-rooted-qualified-1' AND n.graph_domain='qualified-v1' + AND n.admission_state='ROOTED_Q_ADMITTED' AND n.collector_state='ENABLED' + AND ROW(n.core_schema,n.core_schema_oid,n.database_name,n.database_oid,n.storage_uuid,n.server_address,n.server_port,n.mono_lock_key2) + IS NOT DISTINCT FROM ROW(g.core_schema,g.core_schema_oid,g.database_name,g.database_oid,g.storage_uuid,g.server_address,g.server_port,g.mono_lock_key2) + AND n.metadata_storage_uuid<>g.storage_uuid AND n.implementation_fingerprint=$2 + AND ROW(i.namespace_uuid,i.storage_uuid,i.core_schema_oid,i.metadata_schema_oid,i.family_identity,i.implementation_fingerprint) + IS NOT DISTINCT FROM ROW(n.namespace_uuid,n.metadata_storage_uuid,n.core_schema_oid,n.metadata_schema_oid,n.family_identity,n.implementation_fingerprint) + AND s.storage_uuid=i.storage_uuid) AS valid"),[family.namespace.clone().into(),implementation.clone().into()])).await? + .ok_or_else(|| rejected("reader migration prior Q identity is missing"))?; + if !identity.try_get::("", "valid")? + || catalog(connection, core_oid, family.oid, 0).await? != family.catalog + || shape(connection, family.oid, &family.namespace, &family.storage).await? + != policy.try_get::>("", "expected_shape")? + { + return Err(rejected( + "reader migration prior Q physical stamp or shape changed", + )); + } + } + if implementation == current { + return Ok(()); + } + + let registration = include_str!("m20261008_000200_rooted_qualified_family.sql") + .replace("$CORE_SCHEMA$", &c) + .replace("$CORE_LITERAL$", &literal(&core)) + .replace("$IMPLEMENTATION_SHA$", &hex::encode(¤t)); + connection + .execute_unprepared(&trusted_function( + ®istration, + "mst2_route_family_registration_guard", + )?) + .await?; + + let template_namespace = uuid::Uuid::new_v4().to_string(); + let template_schema = format!("mst2q_{}", template_namespace.replace('-', "")); + let template_storage = uuid::Uuid::new_v4().to_string(); + let t = identifier(&template_schema); + connection + .execute_unprepared(&format!("CREATE SCHEMA {t}")) + .await?; + let template_oid: i64 = connection + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT oid::bigint FROM pg_catalog.pg_namespace WHERE nspname=$1", + [template_schema.clone().into()], + )) + .await? + .ok_or_else(|| rejected("reader migration trusted template schema is missing"))? + .try_get("", "oid")?; + connection + .execute_unprepared(&render_family( + &core, + core_oid, + &template_schema, + template_oid, + &template_namespace, + &template_storage, + )) + .await?; + let expected_shape = shape( + connection, + template_oid, + &template_namespace, + &template_storage, + ) + .await?; + connection + .execute_unprepared(&format!( + "SET LOCAL search_path={c},pg_catalog,pg_temp; DROP SCHEMA {t} CASCADE" + )) + .await?; + + 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 mut functions = String::new(); + for name in [ + "mst2_metadata_dml_barrier", + "mst2_metadata_gc_enabled", + "mst2_metadata_root_anchor_guard", + "mst2_metadata_serving_complete", + "mst2_metadata_next_reader_issuance", + "mst2_metadata_reader_issuance_guard", + "mst2_metadata_reader_guard", + "mst2_metadata_prune_readers", + "mst2_metadata_begin_reader", + "mst2_metadata_finish_reader", + "mst2_metadata_read_source_entries", + "mst2_metadata_cleanup_expired", + "mst2_metadata_gc_owner_cleanup", + ] { + functions.push_str(&trusted_function(&rendered, name)?); + functions.push('\n'); + } + let upgrade = include_str!("m20261008_000400_reader_retention_upgrade.sql") + .replace("$Q_SCHEMA$", &q) + .replace("$READER_FUNCTIONS_SQL$", &functions); + connection.execute_unprepared(&upgrade).await?; + if shape(connection, family.oid, &family.namespace, &family.storage).await? + != expected_shape + { + return Err(rejected( + "reader migration upgraded Q differs from the trusted complete shape", + )); + } + } + + // Fingerprints include trigger identities/enabled state, but not table + // values. Preserve trigger OIDs while writing only these controlled stamps. + let authority = catalog(connection, core_oid, 0, q_oid).await?; + let full = if q_oid == 0 { + None + } else { + Some(catalog(connection, core_oid, q_oid, 0).await?) + }; + connection.execute_unprepared(&format!("ALTER TABLE {c}.mst2_qualified_family_policy DISABLE TRIGGER mst2_route_family_policy_immutable")).await?; + connection.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"), + [current.clone().into(),expected_shape.clone().into(),authority.clone().into()])).await?; + connection.execute_unprepared(&format!("ALTER TABLE {c}.mst2_qualified_family_policy ENABLE TRIGGER mst2_route_family_policy_immutable")).await?; + if let (Some(family), Some(full)) = (&family, &full) { + let q = identifier(&family.schema); + connection.execute_unprepared(&format!( + "ALTER TABLE {q}.mst2_metadata_family_identity DISABLE TRIGGER mst2_00_family_barrier; + ALTER TABLE {q}.mst2_metadata_family_identity DISABLE TRIGGER mst2_metadata_identity_immutable; + ALTER TABLE {c}.mst2_metadata_namespace DISABLE TRIGGER mst2_00_route_statement_barrier; + ALTER TABLE {c}.mst2_metadata_namespace DISABLE TRIGGER mst2_route_immutable")).await?; + connection.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres, + format!("UPDATE {q}.mst2_metadata_family_identity SET implementation_fingerprint=$1 WHERE singleton=1"),[current.clone().into()])).await?; + connection.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"), + [current.clone().into(),full.clone().into(),family.namespace.clone().into()])).await?; + connection.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; + ALTER TABLE {c}.mst2_metadata_namespace ENABLE TRIGGER mst2_00_route_statement_barrier; + ALTER TABLE {c}.mst2_metadata_namespace ENABLE TRIGGER mst2_route_immutable")).await?; + } + connection + .execute_unprepared(&format!("SET LOCAL search_path={c},pg_catalog,pg_temp")) + .await?; + if catalog(connection, core_oid, 0, q_oid).await? != authority { + return Err(rejected( + "reader migration final core authority catalog changed", + )); + } + if let (Some(family), Some(full)) = (&family, &full) + && (catalog(connection, core_oid, q_oid, 0).await? != *full + || shape(connection, family.oid, &family.namespace, &family.storage).await? + != expected_shape) + { + return Err(rejected("reader migration final Q fingerprint changed")); + } + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + // Preserve committed issuance fences and the complete reader identity. + Ok(()) + } + + fn use_transaction(&self) -> Option { + Some(true) + } +} diff --git a/src/jupiter/migration/m20261008_000400_reader_retention_upgrade.sql b/src/jupiter/migration/m20261008_000400_reader_retention_upgrade.sql new file mode 100644 index 00000000..cb7b6876 --- /dev/null +++ b/src/jupiter/migration/m20261008_000400_reader_retention_upgrade.sql @@ -0,0 +1,46 @@ +SET LOCAL search_path=$Q_SCHEMA$,pg_catalog,pg_temp; +ALTER TABLE mst2_metadata_reader_operation DISABLE TRIGGER mst2_00_family_barrier; +ALTER TABLE mst2_metadata_reader_operation DISABLE TRIGGER mst2_metadata_reader_guard; +ALTER TABLE mst2_metadata_reader_operation DISABLE TRIGGER mst2_metadata_reader_complete; +ALTER TABLE mst2_metadata_root_anchor DISABLE TRIGGER mst2_00_family_barrier; +ALTER TABLE mst2_metadata_root_anchor DISABLE TRIGGER mst2_metadata_root_anchor_guard; + +ALTER TABLE mst2_metadata_reader_operation + ADD COLUMN reader_issuance bigint NOT NULL DEFAULT 0 CHECK(reader_issuance>=0), + ADD COLUMN terminal_xid bigint; +UPDATE mst2_metadata_reader_operation SET terminal_xid=0 WHERE state IN ('FINISHED','EXPIRED'); +ALTER TABLE mst2_metadata_reader_operation ALTER COLUMN reader_issuance DROP DEFAULT; +ALTER TABLE mst2_metadata_reader_operation + ADD CONSTRAINT mst2_metadata_reader_identity UNIQUE(operation_id,reader_issuance), + ADD CONSTRAINT mst2_metadata_reader_terminal_state CHECK((state='ACTIVE')=(terminal_xid IS NULL)); +CREATE TABLE mst2_metadata_reader_issuance ( + singleton smallint PRIMARY KEY CHECK(singleton=1),high_water bigint NOT NULL CHECK(high_water>=0) +); +INSERT INTO mst2_metadata_reader_issuance VALUES(1,0); +CREATE INDEX mst2_metadata_reader_terminal ON mst2_metadata_reader_operation(terminal_xid,operation_id) WHERE state IN ('FINISHED','EXPIRED'); +ALTER TABLE mst2_metadata_root_anchor ADD COLUMN reader_issuance bigint; +UPDATE mst2_metadata_root_anchor SET reader_issuance=0 WHERE reader_operation_id IS NOT NULL; +ALTER TABLE mst2_metadata_root_anchor + DROP CONSTRAINT mst2_metadata_root_anchor_reader_operation_id_fkey, + ADD CONSTRAINT mst2_metadata_root_anchor_reader_identity FOREIGN KEY(reader_operation_id,reader_issuance) + REFERENCES mst2_metadata_reader_operation(operation_id,reader_issuance) MATCH FULL, + ADD CONSTRAINT mst2_metadata_root_anchor_reader_kind CHECK((anchor_kind IN ('REQUEST','READER'))=(reader_operation_id IS NOT NULL)), + ADD CONSTRAINT mst2_metadata_root_anchor_reader_fields CHECK((reader_operation_id IS NULL)=(reader_issuance IS NULL)); +CREATE INDEX mst2_metadata_root_anchor_reader ON mst2_metadata_root_anchor(reader_operation_id,reader_issuance) WHERE reader_operation_id IS NOT NULL; +DROP FUNCTION mst2_metadata_begin_reader(text,text,text); +DROP FUNCTION mst2_metadata_finish_reader(uuid); +DROP FUNCTION mst2_metadata_read_source_entries(uuid,uuid,bytea,bigint,bytea,jsonb); + +$READER_FUNCTIONS_SQL$ + +CREATE TRIGGER mst2_00_family_barrier BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_issuance + FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_dml_barrier(); +CREATE TRIGGER mst2_metadata_truncate_guard BEFORE TRUNCATE ON mst2_metadata_reader_issuance + FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_immutable(); +CREATE TRIGGER mst2_metadata_reader_issuance_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_issuance + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_reader_issuance_guard(); +ALTER TABLE mst2_metadata_reader_operation ENABLE TRIGGER mst2_00_family_barrier; +ALTER TABLE mst2_metadata_reader_operation ENABLE TRIGGER mst2_metadata_reader_guard; +ALTER TABLE mst2_metadata_reader_operation ENABLE TRIGGER mst2_metadata_reader_complete; +ALTER TABLE mst2_metadata_root_anchor ENABLE TRIGGER mst2_00_family_barrier; +ALTER TABLE mst2_metadata_root_anchor ENABLE TRIGGER mst2_metadata_root_anchor_guard; diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 018716bc..d44427fc 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -154,10 +154,31 @@ mod m20261007_000600_add_mst2_storage_routes; mod m20261008_000100_add_mst2_chunk_maps; mod m20261008_000200_add_mst2_rooted_qualified_family; mod m20261008_000300_add_mst2_chunk_map_retention; +mod m20261008_000400_add_mst2_reader_retention; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; +#[cfg(test)] +pub(crate) async fn test_upgrade_reader_retention( + connection: &sea_orm::DatabaseConnection, +) -> Result<(), sea_orm::DbErr> { + use sea_orm::TransactionTrait; + 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; + match result { + Ok(()) => txn.commit().await, + Err(error) => { + txn.rollback().await?; + Err(error) + } + } +} + /// Primary key `BIGINT` (not DB auto-increment); the application assigns `id` (e.g. `idgenerator::IdInstance::next_id`). fn pk_bigint(name: T) -> ColumnDef { big_integer(name).primary_key().take() @@ -296,6 +317,7 @@ impl MigratorTrait for Migrator { Box::new(m20261008_000100_add_mst2_chunk_maps::Migration), Box::new(m20261008_000200_add_mst2_rooted_qualified_family::Migration), Box::new(m20261008_000300_add_mst2_chunk_map_retention::Migration), + Box::new(m20261008_000400_add_mst2_reader_retention::Migration), ] } } @@ -1206,7 +1228,7 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 17..names.len() - 8], + &names[names.len() - 18..names.len() - 9], &[ "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), @@ -1221,38 +1243,42 @@ mod tests { "native retention, metadata installation and view tables follow media paging" ); assert_eq!( - &names[names.len() - 8], + &names[names.len() - 9], "m20261007_000200_add_mst2_metadata_generations" ); assert_eq!( - &names[names.len() - 7], + &names[names.len() - 8], "m20261007_000300_add_mst2_metadata_lifetime_history" ); assert_eq!( - &names[names.len() - 6], + &names[names.len() - 7], "m20261007_000400_add_mst2_qualified_metadata_gc" ); assert_eq!( - &names[names.len() - 5], + &names[names.len() - 6], "m20261007_000500_add_mst2_install_capability" ); assert_eq!( - &names[names.len() - 4], + &names[names.len() - 5], "m20261007_000600_add_mst2_storage_routes" ); assert_eq!( - &names[names.len() - 3], + &names[names.len() - 4], "m20261008_000100_add_mst2_chunk_maps" ); assert_eq!( - &names[names.len() - 2], + &names[names.len() - 3], "m20261008_000200_add_mst2_rooted_qualified_family" ); assert_eq!( - names.last().unwrap(), + &names[names.len() - 2], "m20261008_000300_add_mst2_chunk_map_retention" ); + assert_eq!( + names.last().unwrap(), + "m20261008_000400_add_mst2_reader_retention" + ); let db = alias_db().await; insert_repo(&db, 1, "/third-party//a").await; insert_repo(&db, 2, "/third-party/b/").await; diff --git a/src/jupiter/storage/qualified_metadata_anchors.sql b/src/jupiter/storage/qualified_metadata_anchors.sql index 08b3d7ea..e7db38e1 100644 --- a/src/jupiter/storage/qualified_metadata_anchors.sql +++ b/src/jupiter/storage/qualified_metadata_anchors.sql @@ -46,8 +46,16 @@ CREATE TABLE mst2_metadata_reader_operation ( snapshot_id text NOT NULL,session_incarnation uuid NOT NULL,root_page bytea NOT NULL,root_generation bigint NOT NULL, lease_epoch bigint NOT NULL CHECK(lease_epoch>0),hard_deadline_unix bigint NOT NULL, state text NOT NULL CHECK(state IN ('ACTIVE','FINISHED','EXPIRED')), + reader_issuance bigint NOT NULL CHECK(reader_issuance>=0),terminal_xid bigint, + CONSTRAINT mst2_metadata_reader_identity UNIQUE(operation_id,reader_issuance), + CONSTRAINT mst2_metadata_reader_terminal_state CHECK((state='ACTIVE')=(terminal_xid IS NULL)), FOREIGN KEY(snapshot_id,session_incarnation) REFERENCES mst2_qualified_session_incarnation(snapshot_id,session_incarnation) ); +CREATE TABLE mst2_metadata_reader_issuance ( + singleton smallint PRIMARY KEY CHECK(singleton=1),high_water bigint NOT NULL CHECK(high_water>=0) +); +INSERT INTO mst2_metadata_reader_issuance VALUES(1,0); +CREATE INDEX mst2_metadata_reader_terminal ON mst2_metadata_reader_operation(terminal_xid,operation_id) WHERE state IN ('FINISHED','EXPIRED'); CREATE INDEX mst2_metadata_reader_active_lease ON mst2_metadata_reader_operation(lease_id,state,operation_id); CREATE INDEX mst2_metadata_reader_active_deadline ON mst2_metadata_reader_operation(hard_deadline_unix,operation_id) WHERE state='ACTIVE'; CREATE TABLE mst2_metadata_root_anchor ( @@ -56,13 +64,17 @@ CREATE TABLE mst2_metadata_root_anchor ( root_page bytea NOT NULL,root_generation bigint NOT NULL,root_certificate_digest bytea NOT NULL, prepare_id text REFERENCES mst2_metadata_prepare(prepare_id), snapshot_id text,session_incarnation uuid,lease_id text REFERENCES mst2_qualified_lease_binding(lease_id), - reader_operation_id uuid REFERENCES mst2_metadata_reader_operation(operation_id), + reader_operation_id uuid,reader_issuance bigint, + CONSTRAINT mst2_metadata_root_anchor_reader_identity FOREIGN KEY(reader_operation_id,reader_issuance) REFERENCES mst2_metadata_reader_operation(operation_id,reader_issuance) MATCH FULL, + CONSTRAINT mst2_metadata_root_anchor_reader_kind CHECK((anchor_kind IN ('REQUEST','READER'))=(reader_operation_id IS NOT NULL)), + CONSTRAINT mst2_metadata_root_anchor_reader_fields CHECK((reader_operation_id IS NULL)=(reader_issuance IS NULL)), UNIQUE(anchor_kind,owner_key,root_page,root_generation), FOREIGN KEY(root_page,root_generation) REFERENCES mst2_metadata_graph_node(page_id,generation), FOREIGN KEY(root_page,root_generation,root_certificate_digest) REFERENCES mst2_metadata_page_certificate(page_id,generation,certificate_digest), FOREIGN KEY(snapshot_id,session_incarnation) REFERENCES mst2_qualified_session_incarnation(snapshot_id,session_incarnation) ); +CREATE INDEX mst2_metadata_root_anchor_reader ON mst2_metadata_root_anchor(reader_operation_id,reader_issuance) WHERE reader_operation_id IS NOT NULL; CREATE INDEX mst2_metadata_root_anchor_page ON mst2_metadata_root_anchor(root_page,root_generation,anchor_kind,owner_key); CREATE INDEX mst2_metadata_root_anchor_prepare ON mst2_metadata_root_anchor(prepare_id,anchor_kind,anchor_id); CREATE INDEX mst2_metadata_root_anchor_lease ON mst2_metadata_root_anchor(lease_id,anchor_kind,anchor_id); @@ -327,7 +339,7 @@ BEGIN RAISE EXCEPTION 'lease root still has its active lease or readers'; END IF; ELSE - IF NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation r WHERE r.operation_id=OLD.reader_operation_id + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation r WHERE r.operation_id=OLD.reader_operation_id AND r.reader_issuance=OLD.reader_issuance AND (r.state='FINISHED' OR r.state='EXPIRED' AND r.hard_deadline_unix<=floor(extract(epoch FROM clock_timestamp()))::bigint)) THEN RAISE EXCEPTION 'reader root still has an active operation'; END IF; @@ -376,7 +388,7 @@ BEGIN ELSE IF NEW.owner_key IS DISTINCT FROM NEW.reader_operation_id::text OR NEW.prepare_id IS NOT NULL OR NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation r JOIN mst2_qualified_lease_binding l USING(lease_id) - WHERE r.operation_id=NEW.reader_operation_id AND r.lease_id=NEW.lease_id AND r.snapshot_id=NEW.snapshot_id + WHERE r.operation_id=NEW.reader_operation_id AND r.reader_issuance=NEW.reader_issuance AND r.lease_id=NEW.lease_id AND r.snapshot_id=NEW.snapshot_id AND r.session_incarnation=NEW.session_incarnation AND r.root_page=NEW.root_page AND r.root_generation=NEW.root_generation AND r.state='ACTIVE' AND l.state='ACTIVE' AND l.lease_epoch=r.lease_epoch AND r.hard_deadline_unix<=l.expires_at_unix AND r.hard_deadline_unix>floor(extract(epoch FROM clock_timestamp()))::bigint) THEN diff --git a/src/jupiter/storage/qualified_metadata_family.rs b/src/jupiter/storage/qualified_metadata_family.rs index 024d325b..c44c8a94 100644 --- a/src/jupiter/storage/qualified_metadata_family.rs +++ b/src/jupiter/storage/qualified_metadata_family.rs @@ -32,11 +32,15 @@ const SOURCE_REVISION_SQL: &str = include_str!("qualified_source_revision.sql"); const ROOTED_SQL: &str = include_str!("qualified_metadata_rooted.sql"); const SERVING_SQL: &str = include_str!("qualified_metadata_serving.sql"); const GC_SQL: &str = include_str!("qualified_metadata_gc.sql"); +const READER_LIFECYCLE_SQL: &str = include_str!("qualified_metadata_reader_lifecycle.sql"); #[path = "qualified_metadata_rooted.rs"] mod rooted; pub(crate) use rooted::{RootedLookupStatus, RootedQualifiedMetadataRepository}; #[cfg(test)] +#[path = "qualified_metadata_reader_previous_fixture.rs"] +pub(crate) mod reader_previous_fixture; +#[cfg(test)] pub(crate) use rooted::{ RootedPrepareIntent, with_rooted_reader_barriers, with_rooted_source_fact_barriers, with_rooted_source_temporary_shadow, @@ -139,6 +143,10 @@ pub(crate) fn implementation_fingerprint() -> Vec { ("indexed-source-read-revision-1", SOURCE_READ_SQL.as_bytes()), ("captured-source-revision-1", SOURCE_REVISION_SQL.as_bytes()), ("rooted-collector-revision-1", GC_SQL.as_bytes()), + ( + "bounded-reader-lifecycle-revision-1", + READER_LIFECYCLE_SQL.as_bytes(), + ), ("typed-certificate-revision-1", CERTIFICATES_SQL.as_bytes()), ( "source-attestation-and-anchor-revision-1", @@ -189,6 +197,7 @@ pub(crate) fn render_family( .replace("$ROOTED_SQL$", ROOTED_SQL) .replace("$SERVING_SQL$", SERVING_SQL) .replace("$GC_SQL$", GC_SQL) + .replace("$READER_LIFECYCLE_SQL$", READER_LIFECYCLE_SQL) .replace("$CORE_SCHEMA$", &identifier(core_schema)) .replace("$CORE_LITERAL$", &literal(core_schema)) .replace("$Q_SCHEMA$", &identifier(q_schema)) diff --git a/src/jupiter/storage/qualified_metadata_family.sql b/src/jupiter/storage/qualified_metadata_family.sql index 853834e5..fae04446 100644 --- a/src/jupiter/storage/qualified_metadata_family.sql +++ b/src/jupiter/storage/qualified_metadata_family.sql @@ -473,7 +473,7 @@ DO $$ DECLARE t text; BEGIN 'mst2_metadata_prepare_page','mst2_metadata_graph_node','mst2_metadata_graph_edge','mst2_metadata_graph_root', 'mst2_metadata_gc_op','mst2_qualified_session_incarnation','mst2_qualified_lease_binding', 'mst2_metadata_page_certificate','mst2_metadata_verified_ref','mst2_metadata_source_root_attestation', - 'mst2_metadata_prepare_reuse_root','mst2_metadata_reuse_index','mst2_metadata_reader_operation','mst2_metadata_root_anchor', + 'mst2_metadata_prepare_reuse_root','mst2_metadata_reuse_index','mst2_metadata_reader_operation','mst2_metadata_reader_issuance','mst2_metadata_root_anchor', 'mst2_metadata_source_entry_reference','mst2_metadata_scope_source_reference'] LOOP EXECUTE format('CREATE TRIGGER mst2_00_family_barrier BEFORE INSERT OR UPDATE OR DELETE ON %I FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_dml_barrier()',t); EXECUTE format('CREATE TRIGGER mst2_metadata_truncate_guard BEFORE TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION mst2_metadata_immutable()',t); diff --git a/src/jupiter/storage/qualified_metadata_gc.sql b/src/jupiter/storage/qualified_metadata_gc.sql index 13b41998..61276556 100644 --- a/src/jupiter/storage/qualified_metadata_gc.sql +++ b/src/jupiter/storage/qualified_metadata_gc.sql @@ -299,8 +299,8 @@ DO $$ DECLARE name text; BEGIN END LOOP; END $$; --- Append-only identity and replay evidence remains bounded independently of --- resident payloads. The admission check reads actual stored rows, never hints. +-- Permanent identity evidence and bounded reader owners have independent +-- resident quotas. Admission reads actual stored rows, never hints. CREATE FUNCTION mst2_metadata_history_capacity_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ DECLARE actual bigint; bytes bigint; row_limit integer; size_expression text; @@ -529,23 +529,23 @@ BEGIN examined:=0; readers_expired:=0; leases_expired:=0; prepares_aborted:=0; handovers_retired:=0; orphans_retired:=0; now_unix:=floor(extract(epoch FROM clock_timestamp()))::bigint; FOR item IN SELECT * FROM ( - (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline + (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline,reader_issuance FROM mst2_metadata_reader_operation WHERE state='ACTIVE' AND hard_deadline_unix<=now_unix ORDER BY hard_deadline_unix,operation_id LIMIT bound) UNION ALL - (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix FROM mst2_qualified_lease_binding + (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix,NULL::bigint FROM mst2_qualified_lease_binding WHERE state='ACTIVE' AND expires_at_unix<=now_unix ORDER BY expires_at_unix,lease_id LIMIT bound) UNION ALL - (SELECT 'PREPARE'::text,prepare_id,NULL::text,floor(extract(epoch FROM orphan_expires_at))::bigint + (SELECT 'PREPARE'::text,prepare_id,NULL::text,floor(extract(epoch FROM orphan_expires_at))::bigint,NULL::bigint FROM mst2_metadata_prepare WHERE (state='PREPARING' OR state='COMMITTED' AND coverage_retired_at IS NULL) AND orphan_expires_at<=clock_timestamp() ORDER BY orphan_expires_at,prepare_id LIMIT bound) ) expired ORDER BY deadline,kind,owner LIMIT bound LOOP examined:=examined+1; IF item.kind='READER' THEN - UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND state='ACTIVE'; + UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND reader_issuance=item.reader_issuance AND state='ACTIVE'; IF NOT FOUND THEN RAISE EXCEPTION 'qualified expired reader changed behind its mutation barrier'; END IF; readers_expired:=readers_expired+1; - DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND anchor_kind IN ('REQUEST','READER'); + DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND reader_issuance=item.reader_issuance AND anchor_kind IN ('REQUEST','READER'); PERFORM mst2_metadata_cleanup_lease(item.lease_id); ELSIF item.kind='LEASE' THEN UPDATE mst2_qualified_lease_binding SET state='EXPIRED',lease_epoch=lease_epoch+1 WHERE lease_id=item.owner AND state='ACTIVE'; diff --git a/src/jupiter/storage/qualified_metadata_reader.rs b/src/jupiter/storage/qualified_metadata_reader.rs index 6e28360a..85b22ff4 100644 --- a/src/jupiter/storage/qualified_metadata_reader.rs +++ b/src/jupiter/storage/qualified_metadata_reader.rs @@ -119,9 +119,15 @@ pub(crate) struct RootedLookupBatch { pub results: Vec, pub proof_pages: Vec<([u8; 32], Vec)>, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ReaderIdentity { + operation_id: uuid::Uuid, + issuance: i64, +} + struct Reader<'a> { txn: &'a DatabaseTransaction, - operation: uuid::Uuid, + operation: ReaderIdentity, bindings: HashMap<[u8; 32], Binding>, cache: HashMap<[u8; 32], StoredPage>, work: PersistedMetadataReadWork, @@ -284,7 +290,7 @@ impl RootedQualifiedMetadataRepository { async fn admit_reader( &self, pinned: &SnapshotContext, - ) -> Result<(uuid::Uuid, Binding, uuid::Uuid), SnapshotError> { + ) -> Result<(ReaderIdentity, Binding, uuid::Uuid), SnapshotError> { let txn = self.transaction().await?; let admitted = async { let row = self @@ -312,7 +318,7 @@ impl RootedQualifiedMetadataRepository { } let reader = txn .query_one_raw(sql( - "SELECT operation_id::text,root_generation,certificate_digest + "SELECT operation_id::text,reader_issuance,root_generation,certificate_digest FROM mst2_metadata_begin_reader($1,$2,$3)", [ pinned.built.snapshot_id.clone().into(), @@ -324,12 +330,15 @@ impl RootedQualifiedMetadataRepository { .map_err(database_error)? .ok_or_else(|| unavailable("qualified reader admission returned no owned root"))?; Ok(( - uuid::Uuid::parse_str( - &reader - .try_get::("", "operation_id") - .map_err(internal)?, - ) - .map_err(internal)?, + ReaderIdentity { + operation_id: uuid::Uuid::parse_str( + &reader + .try_get::("", "operation_id") + .map_err(internal)?, + ) + .map_err(internal)?, + issuance: reader.try_get("", "reader_issuance").map_err(internal)?, + }, Binding { generation: reader.try_get("", "root_generation").map_err(internal)?, certificate: digest_column(&reader, "certificate_digest")?, @@ -349,7 +358,7 @@ impl RootedQualifiedMetadataRepository { async fn finish_reader( &self, - operation: uuid::Uuid, + operation: ReaderIdentity, result: Result, ) -> Result { let cleanup = self.transaction().await; @@ -357,8 +366,11 @@ impl RootedQualifiedMetadataRepository { Ok(txn) => { let finished = txn .execute_raw(sql( - "SELECT mst2_metadata_finish_reader($1::uuid)", - [operation.to_string().into()], + "SELECT mst2_metadata_finish_reader($1::uuid,$2::bigint)", + [ + operation.operation_id.to_string().into(), + operation.issuance.into(), + ], )) .await .map_err(internal) @@ -381,7 +393,7 @@ impl RootedQualifiedMetadataRepository { &self, pinned: &SnapshotContext, requests: &[MetadataRouteRequest<'_>], - operation: uuid::Uuid, + operation: ReaderIdentity, binding: Binding, source: uuid::Uuid, ) -> Result { @@ -424,7 +436,7 @@ impl RootedQualifiedMetadataRepository { impl Reader<'_> { fn new( txn: &DatabaseTransaction, - operation: uuid::Uuid, + operation: ReaderIdentity, root: [u8; 32], binding: Binding, source: uuid::Uuid, @@ -456,8 +468,8 @@ impl Reader<'_> { .iter() .map(|entry| hex::encode(&entry.name)) .collect(); - let rows = self.txn.query_all_raw(sql("SELECT * FROM mst2_metadata_read_source_entries($1::uuid,$2::uuid,$3,$4,$5,$6::jsonb)", - [self.operation.to_string().into(), source.to_string().into(), directory.to_vec().into(), + let rows = self.txn.query_all_raw(sql("SELECT * FROM mst2_metadata_read_source_entries($1::uuid,$2::bigint,$3::uuid,$4,$5,$6,$7::jsonb)", + [self.operation.operation_id.to_string().into(), self.operation.issuance.into(), source.to_string().into(), directory.to_vec().into(), binding.generation.into(), binding.certificate.to_vec().into(),json!(names).into()])).await.map_err(database_error)?; #[cfg(test)] if entries.iter().any(|entry| !entry.is_dir()) @@ -712,7 +724,8 @@ impl Reader<'_> { id.to_vec().into(), binding.generation.into(), binding.certificate.to_vec().into(), - self.operation.to_string().into(), + self.operation.operation_id.to_string().into(), + self.operation.issuance.into(), ], )) .await @@ -947,9 +960,9 @@ const PAGE_SQL:&str="SELECT body.payload,body.byte_size,proof.canonical_proof->' AND octet_length(body.payload)=proof.byte_size AND NOT EXISTS(SELECT 1 FROM mst2_metadata_gc_op gc WHERE gc.page_id=proof.page_id AND gc.generation=proof.generation) AND EXISTS(SELECT 1 FROM mst2_metadata_reader_operation reader JOIN mst2_metadata_root_anchor anchor - ON anchor.reader_operation_id=reader.operation_id AND anchor.anchor_kind='READER' + ON anchor.reader_operation_id=reader.operation_id AND anchor.reader_issuance=reader.reader_issuance AND anchor.anchor_kind='READER' AND anchor.root_page=reader.root_page AND anchor.root_generation=reader.root_generation - WHERE reader.operation_id=$4::uuid AND reader.state='ACTIVE' + WHERE reader.operation_id=$4::uuid AND reader.reader_issuance=$5::bigint AND reader.state='ACTIVE' AND reader.hard_deadline_unix>floor(extract(epoch FROM clock_timestamp()))::bigint) AND (SELECT count(*) FROM mst2_metadata_verified_ref ref WHERE ref.parent_page=proof.page_id AND ref.parent_generation=proof.generation)=jsonb_array_length(proof.canonical_proof->'references') diff --git a/src/jupiter/storage/qualified_metadata_reader_lifecycle.sql b/src/jupiter/storage/qualified_metadata_reader_lifecycle.sql new file mode 100644 index 00000000..058c9a84 --- /dev/null +++ b/src/jupiter/storage/qualified_metadata_reader_lifecycle.sql @@ -0,0 +1,110 @@ +CREATE FUNCTION mst2_metadata_next_reader_issuance(issued bigint) RETURNS bigint LANGUAGE plpgsql IMMUTABLE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +BEGIN + IF issued IS NULL OR issued<0 OR issued=9223372036854775807 THEN + RAISE EXCEPTION 'qualified reader issuance is exhausted or invalid'; END IF; + RETURN issued+1; +END $$; + +CREATE FUNCTION mst2_metadata_reader_issuance_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +BEGIN + IF TG_OP<>'UPDATE' THEN RAISE EXCEPTION 'qualified reader issuance cannot be reset or removed'; END IF; + IF NEW.singleton IS DISTINCT FROM OLD.singleton OR NEW.high_water IS DISTINCT FROM mst2_metadata_next_reader_issuance(OLD.high_water) THEN + RAISE EXCEPTION 'qualified reader issuance must advance exactly once'; END IF; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_reader_issuance_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_issuance + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_reader_issuance_guard(); + +CREATE FUNCTION mst2_metadata_reader_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE l mst2_qualified_lease_binding%ROWTYPE; issued bigint; + now_unix bigint:=floor(extract(epoch FROM clock_timestamp()))::bigint; +BEGIN + IF TG_OP='DELETE' THEN + IF OLD.state NOT IN ('FINISHED','EXPIRED') OR OLD.terminal_xid IS NULL OR OLD.terminal_xid=txid_current() + OR OLD.state='EXPIRED' AND OLD.hard_deadline_unix>now_unix + OR EXISTS(SELECT 1 FROM mst2_metadata_root_anchor WHERE reader_operation_id=OLD.operation_id) THEN + RAISE EXCEPTION 'qualified reader still has its active owner, deferred completion, or owned roots'; END IF; + RETURN OLD; + END IF; + IF TG_OP='UPDATE' THEN + IF (to_jsonb(NEW)-'state') IS DISTINCT FROM (to_jsonb(OLD)-'state') + OR OLD.state<>'ACTIVE' AND NEW.state<>OLD.state OR NEW.state NOT IN ('ACTIVE','FINISHED','EXPIRED') + OR NEW.state='EXPIRED' AND OLD.hard_deadline_unix>now_unix THEN + RAISE EXCEPTION 'qualified reader identity cannot change or be prematurely expired'; END IF; + IF OLD.state<>'ACTIVE' THEN RETURN NULL; END IF; + IF NEW.state<>'ACTIVE' THEN NEW.terminal_xid:=txid_current(); END IF; + RETURN NEW; + END IF; + SELECT * INTO l FROM mst2_qualified_lease_binding WHERE lease_id=NEW.lease_id AND state='ACTIVE' AND expires_at_unix>now_unix; + IF NOT FOUND OR NEW.state<>'ACTIVE' OR NEW.terminal_xid IS NOT NULL OR NEW.reader_issuance<=0 + OR substr(NEW.operation_id::text,15,1)<>'4' OR substr(NEW.operation_id::text,20,1) NOT IN ('8','9','a','b') + OR ROW(NEW.snapshot_id,NEW.session_incarnation,NEW.root_page,NEW.root_generation,NEW.lease_epoch) IS DISTINCT FROM + ROW(l.snapshot_id,l.session_incarnation,l.metadata_root,l.root_generation,l.lease_epoch) + OR NEW.hard_deadline_unix<=now_unix OR NEW.hard_deadline_unix>least(l.expires_at_unix,now_unix+60) + OR NOT mst2_metadata_incarnation_proof(l.snapshot_id,l.session_incarnation,true) THEN + RAISE EXCEPTION 'qualified reader lacks its exact active lease and bounded deadline'; END IF; + SELECT high_water INTO STRICT issued FROM mst2_metadata_reader_issuance WHERE singleton=1; + IF NEW.reader_issuance<>mst2_metadata_next_reader_issuance(issued) THEN RAISE EXCEPTION 'qualified reader cannot replay an issued identity'; END IF; + UPDATE mst2_metadata_reader_issuance SET high_water=NEW.reader_issuance WHERE singleton=1; + RETURN NEW; +END $$; +CREATE TRIGGER mst2_metadata_reader_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_operation + FOR EACH ROW EXECUTE FUNCTION mst2_metadata_reader_guard(); + +CREATE FUNCTION mst2_metadata_prune_readers(maximum integer) RETURNS bigint LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE removed bigint; now_unix bigint:=floor(extract(epoch FROM clock_timestamp()))::bigint; +BEGIN + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + IF maximum IS NULL OR maximum NOT BETWEEN 0 AND 64 THEN RAISE EXCEPTION 'qualified reader prune budget must be 0..=64'; END IF; + WITH candidates AS ( + SELECT r.operation_id,r.reader_issuance FROM mst2_metadata_reader_operation r + WHERE r.state IN ('FINISHED','EXPIRED') AND r.terminal_xid (usize, usize) { + let start = source.find(&format!("CREATE FUNCTION {name}(")).unwrap(); + let f = &source[start..]; + let body = f.find("AS $").unwrap() + 3; + let tag_end = f[body + 1..].find('$').unwrap() + body + 2; + let tag = &f[body..tag_end]; + let end = f[tag_end..].find(tag).unwrap() + tag_end + tag.len() + 1; + assert_eq!(f.as_bytes()[end - 1], b';'); + (start, start + end) +} + +fn render_previous( + core: &str, + core_oid: i64, + q: &str, + q_oid: i64, + namespace: &str, + storage: &str, +) -> String { + let capture = CAPTURE.replace("\r\n", "\n"); + assert_eq!( + hex::encode(Sha256::digest(capture.as_bytes())), + "599a5a6749cb41afd8f7b05287a482ab6ba921d4ffe6142159b0bbf0f00e33f4" + ); + let mut sql = render_family(core, core_oid, q, q_oid, namespace, storage).replace("\r\n", "\n"); + let a = sql + .find("CREATE TABLE mst2_metadata_reader_operation (") + .unwrap(); + let z = sql[a..] + .find("CREATE FUNCTION mst2_metadata_session_covers_prepare(") + .unwrap() + + a; + let old_a = capture.find("-- READER DDL BEGIN\n").unwrap() + "-- READER DDL BEGIN\n".len(); + let old_z = capture.find("-- READER DDL END").unwrap(); + sql.replace_range(a..z, &capture[old_a..old_z]); + for name in [ + "mst2_metadata_dml_barrier", + "mst2_metadata_gc_enabled", + "mst2_metadata_root_anchor_guard", + "mst2_metadata_serving_complete", + "mst2_metadata_reader_guard", + "mst2_metadata_begin_reader", + "mst2_metadata_finish_reader", + "mst2_metadata_read_source_entries", + "mst2_metadata_cleanup_expired", + "mst2_metadata_gc_owner_cleanup", + ] { + let (a, z) = function(&sql, name); + let (old_a, old_z) = function(&capture, name); + sql.replace_range(a..z, &capture[old_a..old_z]); + } + for name in [ + "mst2_metadata_next_reader_issuance", + "mst2_metadata_reader_issuance_guard", + "mst2_metadata_prune_readers", + ] { + let (a, z) = function(&sql, name); + sql.replace_range(a..z, ""); + } + sql=sql.replace("CREATE TRIGGER mst2_metadata_reader_issuance_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_issuance\n FOR EACH ROW EXECUTE FUNCTION mst2_metadata_reader_issuance_guard();","") + .replace("'mst2_metadata_reader_issuance',","") + .replace(&hex::encode(implementation_fingerprint()),PREVIOUS) + .replace("$CORE_SCHEMA$",&identifier(core)).replace("$CORE_LITERAL$",&literal(core)) + .replace("$Q_SCHEMA$",&identifier(q)).replace("$Q_LITERAL$",&literal(q)) + .replace("$CORE_OID$",&core_oid.to_string()).replace("$Q_OID$",&q_oid.to_string()) + .replace("$NAMESPACE_UUID$",namespace).replace("$STORAGE_UUID$",storage).replace("$IMPLEMENTATION_SHA$",PREVIOUS); + assert!(!sql.contains("reader_issuance")); + assert!(!sql.contains("terminal_xid")); + sql +} + +struct Table { + name: String, + rows: Value, + dependencies: BTreeSet, +} + +async fn fingerprint(txn: &DatabaseTransaction, core_oid: i64, q_oid: i64, exempt: i64) -> Vec { + catalog_with_exemption(txn, core_oid, q_oid, exempt) + .await + .unwrap() +} + +/// Recreate Q relations from captured old DDL within an owned test schema. +/// Preserve its namespace OID, scope and real source/certificate/session rows. +pub(crate) async fn restore_previous(core: &DatabaseConnection, q_schema: &str, keep_q: bool) { + let txn = core.begin().await.unwrap(); + let (core_schema, core_oid) = captured_core(&txn).await.unwrap(); + assert!(core_schema.starts_with("mega2_test_")); + assert!(q_schema.starts_with("mst2q_")); + let c = identifier(&core_schema); + let q = identifier(q_schema); + txn.execute_unprepared(&format!( + "SELECT {c}.mst2_route_enter({})", + literal(&core_schema) + )) + .await + .unwrap(); + let registry=txn.query_one_raw(Statement::from_sql_and_values(DbBackend::Postgres, + format!("SELECT metadata_schema_oid::bigint,namespace_uuid::text,metadata_storage_uuid FROM {c}.mst2_metadata_namespace WHERE metadata_schema=$1 AND graph_domain='qualified-v1'"), + [q_schema.into()])).await.unwrap().unwrap(); + let q_oid: i64 = registry.try_get("", "metadata_schema_oid").unwrap(); + let namespace: String = registry.try_get("", "namespace_uuid").unwrap(); + let storage: String = registry.try_get("", "metadata_storage_uuid").unwrap(); + let table_rows=txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT relname FROM pg_catalog.pg_class WHERE relnamespace=$1::bigint::oid AND relkind='r' ORDER BY relname",[q_oid.into()])).await.unwrap(); + let mut tables = Vec::new(); + let mut all_names = Vec::new(); + for row in table_rows { + let name: String = row.try_get("", "relname").unwrap(); + all_names.push(name.clone()); + if [ + "mst2_metadata_storage_scope", + "mst2_metadata_family_identity", + "mst2_metadata_reader_issuance", + ] + .contains(&name.as_str()) + { + continue; + } + let rows:Value=txn.query_one_raw(Statement::from_string(DbBackend::Postgres,format!( + "SELECT coalesce(jsonb_agg(to_jsonb(data)),'[]'::jsonb) AS rows FROM {q}.{} data",identifier(&name)))) + .await.unwrap().unwrap().try_get("","rows").unwrap(); + if !keep_q { + assert_eq!( + rows, + json!([]), + "Q-less fixture must have no history to discard" + ); + } + let dependencies=txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT target.relname FROM pg_catalog.pg_constraint fk JOIN pg_catalog.pg_class source ON source.oid=fk.conrelid + JOIN pg_catalog.pg_class target ON target.oid=fk.confrelid WHERE source.relnamespace=$1::bigint::oid + AND target.relnamespace=source.relnamespace AND source.relname=$2 AND target.relname<>source.relname AND fk.contype='f'", + [q_oid.into(),name.clone().into()])).await.unwrap().into_iter() + .map(|row|row.try_get("","relname").unwrap()).collect(); + tables.push(Table { + name, + rows, + dependencies, + }); + } + txn.execute_unprepared(&format!( + "DROP TABLE {} CASCADE", + all_names + .iter() + .map(|name| format!("{q}.{}", identifier(name))) + .collect::>() + .join(",") + )) + .await + .unwrap(); + let funcs=txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT proname,pg_catalog.pg_get_function_identity_arguments(oid) AS args FROM pg_catalog.pg_proc WHERE pronamespace=$1::bigint::oid",[q_oid.into()])).await.unwrap(); + for row in funcs { + let name: String = row.try_get("", "proname").unwrap(); + let args: String = row.try_get("", "args").unwrap(); + txn.execute_unprepared(&format!( + "DROP FUNCTION IF EXISTS {q}.{}({args}) CASCADE", + identifier(&name) + )) + .await + .unwrap(); + } + txn.execute_unprepared(&render_previous( + &core_schema, + core_oid, + q_schema, + q_oid, + &namespace, + &storage, + )) + .await + .unwrap(); + let old_tables=txn.query_all_raw(Statement::from_sql_and_values(DbBackend::Postgres, + "SELECT relname FROM pg_catalog.pg_class WHERE relnamespace=$1::bigint::oid AND relkind='r'",[q_oid.into()])).await.unwrap(); + let old_names: Vec = old_tables + .into_iter() + .map(|row| row.try_get("", "relname").unwrap()) + .collect(); + for name in &old_names { + txn.execute_unprepared(&format!( + "ALTER TABLE {q}.{} DISABLE TRIGGER USER", + identifier(name) + )) + .await + .unwrap(); + } + let mut restored = BTreeSet::from([ + "mst2_metadata_storage_scope".to_owned(), + "mst2_metadata_family_identity".to_owned(), + ]); + while !tables.is_empty() { + let index = tables + .iter() + .position(|table| table.dependencies.is_subset(&restored)) + .expect("captured Q foreign keys are acyclic"); + let table = tables.remove(index); + txn.execute_raw(Statement::from_sql_and_values(DbBackend::Postgres,format!( + "INSERT INTO {q}.{t} SELECT * FROM jsonb_populate_recordset(NULL::{q}.{t},$1::jsonb)",t=identifier(&table.name)),[table.rows.into()])).await.unwrap(); + restored.insert(table.name); + } + for name in &old_names { + txn.execute_unprepared(&format!( + "ALTER TABLE {q}.{} ENABLE TRIGGER USER", + identifier(name) + )) + .await + .unwrap(); + } + let capture = CAPTURE.replace("\r\n", "\n"); + let (a, z) = function(&capture, "mst2_route_family_registration_guard"); + let registration = capture[a..z] + .replacen("CREATE FUNCTION", "CREATE OR REPLACE FUNCTION", 1) + .replace("$CORE_SCHEMA$", &c) + .replace("$IMPLEMENTATION_SHA$", PREVIOUS); + txn.execute_unprepared(®istration).await.unwrap(); + let old_shape: Vec = txn + .query_one_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + format!( + "SELECT {c}.mst2_route_family_shape($1::bigint::oid,$2::uuid,$3) AS fingerprint" + ), + [q_oid.into(), namespace.clone().into(), storage.into()], + )) + .await + .unwrap() + .unwrap() + .try_get("", "fingerprint") + .unwrap(); + txn.execute_unprepared(&format!("SET LOCAL search_path={c},pg_catalog,pg_temp")) + .await + .unwrap(); + if !keep_q { + txn.execute_unprepared(&format!( + "ALTER TABLE {c}.mst2_metadata_namespace DISABLE TRIGGER mst2_route_immutable" + )) + .await + .unwrap(); + txn.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + format!("DELETE FROM {c}.mst2_metadata_namespace WHERE namespace_uuid=$1::uuid"), + [namespace.clone().into()], + )) + .await + .unwrap(); + txn.execute_unprepared(&format!("ALTER TABLE {c}.mst2_metadata_namespace ENABLE TRIGGER mst2_route_immutable; DROP SCHEMA {q} CASCADE")).await.unwrap(); + } + let authority = fingerprint(&txn, core_oid, 0, if keep_q { q_oid } else { 0 }).await; + let full = if keep_q { + Some(fingerprint(&txn, core_oid, q_oid, 0).await) + } else { + None + }; + 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(PREVIOUS).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!( + "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 {c}.mst2_metadata_namespace SET implementation_fingerprint=$1,catalog_fingerprint=$2 WHERE namespace_uuid=$3::uuid"), + [hex::decode(PREVIOUS).unwrap().into(),full.clone().into(),namespace.into()])).await.unwrap(); + txn.execute_unprepared(&format!( + "ALTER TABLE {c}.mst2_metadata_namespace ENABLE TRIGGER mst2_route_immutable" + )) + .await + .unwrap(); + } + assert_eq!( + fingerprint(&txn, core_oid, 0, if keep_q { q_oid } else { 0 }).await, + authority + ); + if let Some(full) = full { + assert_eq!(fingerprint(&txn, core_oid, q_oid, 0).await, full); + } + txn.commit().await.unwrap(); +} diff --git a/src/jupiter/storage/qualified_metadata_reader_previous_fixture.sql b/src/jupiter/storage/qualified_metadata_reader_previous_fixture.sql new file mode 100644 index 00000000..3b087462 --- /dev/null +++ b/src/jupiter/storage/qualified_metadata_reader_previous_fixture.sql @@ -0,0 +1,425 @@ +-- Captured exactly from c1ff280281440891e732ecfaed4732ee59c41a90; test-only. +-- READER DDL BEGIN +CREATE TABLE mst2_metadata_reader_operation ( + operation_id uuid PRIMARY KEY,lease_id text NOT NULL REFERENCES mst2_qualified_lease_binding(lease_id), + snapshot_id text NOT NULL,session_incarnation uuid NOT NULL,root_page bytea NOT NULL,root_generation bigint NOT NULL, + lease_epoch bigint NOT NULL CHECK(lease_epoch>0),hard_deadline_unix bigint NOT NULL, + state text NOT NULL CHECK(state IN ('ACTIVE','FINISHED','EXPIRED')), + FOREIGN KEY(snapshot_id,session_incarnation) REFERENCES mst2_qualified_session_incarnation(snapshot_id,session_incarnation) +); +CREATE INDEX mst2_metadata_reader_active_lease ON mst2_metadata_reader_operation(lease_id,state,operation_id); +CREATE INDEX mst2_metadata_reader_active_deadline ON mst2_metadata_reader_operation(hard_deadline_unix,operation_id) WHERE state='ACTIVE'; +CREATE TABLE mst2_metadata_root_anchor ( + anchor_id uuid PRIMARY KEY,anchor_kind text NOT NULL CHECK(anchor_kind IN ('PREPARE','REUSE','SESSION','LEASE','REQUEST','READER')), + owner_key text NOT NULL CHECK(octet_length(owner_key) BETWEEN 1 AND 512), + root_page bytea NOT NULL,root_generation bigint NOT NULL,root_certificate_digest bytea NOT NULL, + prepare_id text REFERENCES mst2_metadata_prepare(prepare_id), + snapshot_id text,session_incarnation uuid,lease_id text REFERENCES mst2_qualified_lease_binding(lease_id), + reader_operation_id uuid REFERENCES mst2_metadata_reader_operation(operation_id), + UNIQUE(anchor_kind,owner_key,root_page,root_generation), + FOREIGN KEY(root_page,root_generation) REFERENCES mst2_metadata_graph_node(page_id,generation), + FOREIGN KEY(root_page,root_generation,root_certificate_digest) + REFERENCES mst2_metadata_page_certificate(page_id,generation,certificate_digest), + FOREIGN KEY(snapshot_id,session_incarnation) REFERENCES mst2_qualified_session_incarnation(snapshot_id,session_incarnation) +); +CREATE INDEX mst2_metadata_root_anchor_page ON mst2_metadata_root_anchor(root_page,root_generation,anchor_kind,owner_key); +CREATE INDEX mst2_metadata_root_anchor_prepare ON mst2_metadata_root_anchor(prepare_id,anchor_kind,anchor_id); +CREATE INDEX mst2_metadata_root_anchor_lease ON mst2_metadata_root_anchor(lease_id,anchor_kind,anchor_id); + +-- READER DDL END + +CREATE FUNCTION mst2_metadata_dml_barrier() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ +BEGIN + IF pg_catalog.current_schema() IS DISTINCT FROM $Q_LITERAL$ OR TG_TABLE_SCHEMA IS DISTINCT FROM $Q_LITERAL$ + OR NOT EXISTS(SELECT 1 FROM pg_catalog.pg_class WHERE oid=TG_RELID AND relnamespace='$Q_OID$'::oid) + OR NOT EXISTS(SELECT 1 FROM pg_catalog.pg_namespace WHERE oid='$Q_OID$'::oid AND nspname=$Q_LITERAL$) + OR pg_catalog.pg_is_in_recovery() OR pg_catalog.current_setting('transaction_isolation')<>'read committed' THEN + RAISE EXCEPTION 'qualified mutation requires its captured primary family and READ COMMITTED'; + END IF; + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + IF NOT EXISTS(SELECT 1 FROM $CORE_SCHEMA$.mst2_metadata_namespace n + WHERE n.namespace_uuid='$NAMESPACE_UUID$'::uuid AND n.metadata_schema=$Q_LITERAL$ + AND n.metadata_schema_oid='$Q_OID$'::oid AND n.core_schema_oid=$CORE_OID$ + AND n.metadata_storage_uuid='$STORAGE_UUID$' AND n.admission_state='ROOTED_Q_ADMITTED' + AND n.collector_state='ENABLED' AND n.implementation_fingerprint=pg_catalog.decode('$IMPLEMENTATION_SHA$','hex') + AND n.catalog_fingerprint=$CORE_SCHEMA$.mst2_route_family_catalog($CORE_OID$,'$Q_OID$'::oid)) THEN + RAISE EXCEPTION 'qualified physical family catalog fingerprint is unavailable'; + END IF; + RETURN NULL; +END $$; + +CREATE FUNCTION mst2_metadata_gc_enabled() RETURNS boolean LANGUAGE sql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ + SELECT NOT pg_is_in_recovery() AND current_setting('transaction_isolation')='read committed' + AND EXISTS(SELECT 1 FROM $CORE_SCHEMA$.mst2_metadata_namespace n WHERE n.namespace_uuid='$NAMESPACE_UUID$'::uuid + AND n.graph_domain='qualified-v1' AND n.metadata_schema=$Q_LITERAL$ AND n.metadata_schema_oid='$Q_OID$'::oid + AND n.core_schema_oid=$CORE_OID$ AND n.metadata_storage_uuid='$STORAGE_UUID$' + AND n.admission_state='ROOTED_Q_ADMITTED' AND n.collector_state='ENABLED' + AND n.implementation_fingerprint=decode('$IMPLEMENTATION_SHA$','hex') + AND n.catalog_fingerprint=$CORE_SCHEMA$.mst2_route_family_catalog($CORE_OID$,'$Q_OID$'::oid)) +$$; + +CREATE FUNCTION mst2_metadata_root_anchor_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +BEGIN + IF TG_OP='UPDATE' THEN RAISE EXCEPTION 'rooted anchors cannot retarget immutable owner or root identities'; END IF; + IF TG_OP='DELETE' THEN + IF OLD.anchor_kind IN ('PREPARE','REUSE') THEN + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare WHERE prepare_id=OLD.prepare_id AND state IN ('PREPARING','COMMITTED','ABORTED')) THEN + RAISE EXCEPTION 'temporary anchor owner history is missing'; + END IF; + ELSIF OLD.anchor_kind='SESSION' THEN + IF NOT EXISTS(SELECT 1 FROM mst2_qualified_session_incarnation s + WHERE s.snapshot_id=OLD.snapshot_id AND s.session_incarnation=OLD.session_incarnation AND s.state='RETIRED') + OR EXISTS(SELECT 1 FROM mst2_qualified_lease_binding l WHERE l.snapshot_id=OLD.snapshot_id + AND l.session_incarnation=OLD.session_incarnation AND l.state='ACTIVE') THEN + RAISE EXCEPTION 'session root still has its active incarnation or leases'; + END IF; + ELSIF OLD.anchor_kind='LEASE' THEN + IF NOT EXISTS(SELECT 1 FROM mst2_qualified_lease_binding WHERE lease_id=OLD.lease_id AND state IN ('RELEASED','EXPIRED')) + OR EXISTS(SELECT 1 FROM mst2_metadata_reader_operation WHERE lease_id=OLD.lease_id AND state='ACTIVE') THEN + RAISE EXCEPTION 'lease root still has its active lease or readers'; + END IF; + ELSE + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation r WHERE r.operation_id=OLD.reader_operation_id + AND (r.state='FINISHED' OR r.state='EXPIRED' AND r.hard_deadline_unix<=floor(extract(epoch FROM clock_timestamp()))::bigint)) THEN + RAISE EXCEPTION 'reader root still has an active operation'; + END IF; + END IF; + RETURN OLD; + END IF; + IF NOT EXISTS(SELECT 1 FROM mst2_metadata_graph_node node + JOIN mst2_metadata_current cur USING(page_id,generation) + JOIN mst2_metadata_lifetime life USING(page_id,generation) + JOIN mst2_metadata_page_certificate proof USING(page_id,generation) + WHERE node.page_id=NEW.root_page AND node.generation=NEW.root_generation AND node.state='LIVE' + AND node.certificate_digest=NEW.root_certificate_digest AND proof.certificate_digest=NEW.root_certificate_digest + AND life.state IN ('RESERVED','LIVE') AND life.graph_domain='qualified-v1' + AND NOT EXISTS(SELECT 1 FROM mst2_metadata_gc_op gc WHERE gc.page_id=node.page_id AND gc.generation=node.generation)) THEN + RAISE EXCEPTION 'rooted anchor does not protect its exact canonical current graph'; + END IF; + IF NEW.anchor_kind IN ('PREPARE','REUSE') THEN + IF NEW.owner_key IS DISTINCT FROM NEW.prepare_id OR NEW.snapshot_id IS NOT NULL OR NEW.session_incarnation IS NOT NULL + OR NEW.lease_id IS NOT NULL OR NEW.reader_operation_id IS NOT NULL + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare WHERE prepare_id=NEW.prepare_id + AND state IN ('PREPARING','COMMITTED') AND coverage_retired_at IS NULL) + OR NEW.anchor_kind='PREPARE' AND NOT (EXISTS(SELECT 1 FROM mst2_metadata_prepare_page + WHERE prepare_id=NEW.prepare_id AND page_id=NEW.root_page AND generation=NEW.root_generation) + OR EXISTS(SELECT 1 FROM mst2_metadata_prepare q JOIN mst2_metadata_prepare_reuse_root r USING(prepare_id) + WHERE q.prepare_id=NEW.prepare_id AND q.plan_kind='ROOTED' AND q.metadata_root=NEW.root_page + AND r.root_page=NEW.root_page AND r.root_generation=NEW.root_generation)) + OR NEW.anchor_kind='REUSE' AND NOT EXISTS(SELECT 1 FROM mst2_metadata_prepare_reuse_root + WHERE prepare_id=NEW.prepare_id AND root_page=NEW.root_page AND root_generation=NEW.root_generation) THEN + RAISE EXCEPTION 'temporary anchor differs from its immutable delta or reused-root owner'; + END IF; + ELSIF NEW.anchor_kind='SESSION' THEN + IF NEW.owner_key IS DISTINCT FROM NEW.snapshot_id||':'||NEW.session_incarnation::text OR NEW.prepare_id IS NOT NULL + OR NEW.lease_id IS NOT NULL OR NEW.reader_operation_id IS NOT NULL + OR NOT EXISTS(SELECT 1 FROM mst2_qualified_session_incarnation s WHERE s.snapshot_id=NEW.snapshot_id + AND s.session_incarnation=NEW.session_incarnation AND s.metadata_root=NEW.root_page AND s.root_generation=NEW.root_generation AND s.state='READY') THEN + RAISE EXCEPTION 'session anchor differs from its exact ready incarnation'; + END IF; + ELSIF NEW.anchor_kind='LEASE' THEN + IF NEW.owner_key IS DISTINCT FROM NEW.lease_id OR NEW.prepare_id IS NOT NULL OR NEW.reader_operation_id IS NOT NULL + OR NOT EXISTS(SELECT 1 FROM mst2_qualified_lease_binding l WHERE l.lease_id=NEW.lease_id + AND l.snapshot_id=NEW.snapshot_id AND l.session_incarnation=NEW.session_incarnation + AND l.metadata_root=NEW.root_page AND l.root_generation=NEW.root_generation AND l.state='ACTIVE' + AND l.expires_at_unix>floor(extract(epoch FROM clock_timestamp()))::bigint) THEN + RAISE EXCEPTION 'lease anchor differs from its exact active lease'; + END IF; + ELSE + IF NEW.owner_key IS DISTINCT FROM NEW.reader_operation_id::text OR NEW.prepare_id IS NOT NULL + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation r JOIN mst2_qualified_lease_binding l USING(lease_id) + WHERE r.operation_id=NEW.reader_operation_id AND r.lease_id=NEW.lease_id AND r.snapshot_id=NEW.snapshot_id + AND r.session_incarnation=NEW.session_incarnation AND r.root_page=NEW.root_page AND r.root_generation=NEW.root_generation + AND r.state='ACTIVE' AND l.state='ACTIVE' AND l.lease_epoch=r.lease_epoch + AND r.hard_deadline_unix<=l.expires_at_unix AND r.hard_deadline_unix>floor(extract(epoch FROM clock_timestamp()))::bigint) THEN + RAISE EXCEPTION 'reader anchor differs from its exact active lease operation'; + END IF; + END IF; + RETURN NEW; +END $$; + +CREATE FUNCTION mst2_metadata_serving_complete() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE s mst2_qualified_session_incarnation%ROWTYPE; l mst2_qualified_lease_binding%ROWTYPE; + r mst2_metadata_reader_operation%ROWTYPE; +BEGIN + IF TG_TABLE_NAME='mst2_qualified_session_incarnation' THEN + SELECT * INTO STRICT s FROM mst2_qualified_session_incarnation + WHERE snapshot_id=NEW.snapshot_id AND session_incarnation=NEW.session_incarnation; + IF NOT mst2_metadata_incarnation_proof(s.snapshot_id,s.session_incarnation,s.state='READY') + OR s.state='READY' AND NOT EXISTS(SELECT 1 FROM mst2_qualified_lease_binding lease + WHERE lease.snapshot_id=s.snapshot_id AND lease.session_incarnation=s.session_incarnation AND lease.state='ACTIVE') + OR s.state='RETIRED' AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor a + WHERE a.anchor_kind='SESSION' AND a.snapshot_id=s.snapshot_id AND a.session_incarnation=s.session_incarnation) THEN + RAISE EXCEPTION 'qualified session cannot commit without its exact final serving roots'; END IF; + ELSIF TG_TABLE_NAME='mst2_qualified_lease_binding' THEN + SELECT * INTO STRICT l FROM mst2_qualified_lease_binding WHERE lease_id=NEW.lease_id; + IF NOT EXISTS(SELECT 1 FROM $CORE_SCHEMA$.mst2_lease_storage_route route + WHERE route.lease_id=l.lease_id AND route.namespace_uuid=l.namespace_uuid AND route.snapshot_id=l.snapshot_id + AND route.session_incarnation=l.session_incarnation AND route.prepare_id=l.prepare_id AND route.metadata_root=l.metadata_root + AND route.authorization_epoch=l.authorization_epoch AND route.publication_sequence=l.publication_sequence + AND route.writer_epoch=l.writer_epoch AND route.certificate_receipt_id=l.certificate_receipt_id) + OR l.state='ACTIVE' AND (l.expires_at_unix<=floor(extract(epoch FROM clock_timestamp()))::bigint + OR NOT mst2_metadata_incarnation_proof(l.snapshot_id,l.session_incarnation,true) + OR NOT EXISTS(SELECT 1 FROM mst2_metadata_root_anchor a WHERE a.anchor_kind='LEASE' AND a.owner_key=l.lease_id + AND a.lease_id=l.lease_id AND a.root_page=l.metadata_root AND a.root_generation=l.root_generation)) + OR l.state<>'ACTIVE' AND NOT EXISTS(SELECT 1 FROM mst2_metadata_reader_operation operation + WHERE operation.lease_id=l.lease_id AND operation.state='ACTIVE') + AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor a WHERE a.anchor_kind='LEASE' AND a.lease_id=l.lease_id) THEN + RAISE EXCEPTION 'qualified lease cannot commit without its exact final route and owned protection'; END IF; + ELSE + SELECT * INTO STRICT r FROM mst2_metadata_reader_operation WHERE operation_id=NEW.operation_id; + IF r.state='ACTIVE' AND (r.hard_deadline_unix<=floor(extract(epoch FROM clock_timestamp()))::bigint + OR (SELECT count(*) FROM mst2_metadata_root_anchor a WHERE a.reader_operation_id=r.operation_id + AND a.anchor_kind IN ('REQUEST','READER') AND a.owner_key=r.operation_id::text + AND a.lease_id=r.lease_id AND a.snapshot_id=r.snapshot_id AND a.session_incarnation=r.session_incarnation + AND a.root_page=r.root_page AND a.root_generation=r.root_generation)<>2) + OR r.state<>'ACTIVE' AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor a WHERE a.reader_operation_id=r.operation_id) THEN + RAISE EXCEPTION 'qualified reader cannot commit without both exact owned roots or definitive cleanup'; END IF; + END IF; + RETURN NULL; +END $$; + +CREATE FUNCTION mst2_metadata_reader_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE l mst2_qualified_lease_binding%ROWTYPE; now_unix bigint:=floor(extract(epoch FROM clock_timestamp()))::bigint; +BEGIN + IF TG_OP='DELETE' THEN RAISE EXCEPTION 'qualified reader operation history is immutable'; END IF; + IF TG_OP='UPDATE' THEN + IF (to_jsonb(NEW)-'state') IS DISTINCT FROM (to_jsonb(OLD)-'state') + OR OLD.state<>'ACTIVE' AND NEW.state<>OLD.state OR NEW.state NOT IN ('ACTIVE','FINISHED','EXPIRED') + OR NEW.state='EXPIRED' AND OLD.hard_deadline_unix>now_unix THEN + RAISE EXCEPTION 'qualified reader identity cannot change or be prematurely expired'; END IF; + RETURN NEW; + END IF; + SELECT * INTO l FROM mst2_qualified_lease_binding WHERE lease_id=NEW.lease_id AND state='ACTIVE' AND expires_at_unix>now_unix; + IF NOT FOUND OR NEW.state<>'ACTIVE' OR substr(NEW.operation_id::text,15,1)<>'4' + OR substr(NEW.operation_id::text,20,1) NOT IN ('8','9','a','b') + OR ROW(NEW.snapshot_id,NEW.session_incarnation,NEW.root_page,NEW.root_generation,NEW.lease_epoch) IS DISTINCT FROM + ROW(l.snapshot_id,l.session_incarnation,l.metadata_root,l.root_generation,l.lease_epoch) + OR NEW.hard_deadline_unix<=now_unix OR NEW.hard_deadline_unix>least(l.expires_at_unix,now_unix+60) + OR NOT mst2_metadata_incarnation_proof(l.snapshot_id,l.session_incarnation,true) THEN + RAISE EXCEPTION 'qualified reader lacks its exact active lease and bounded deadline'; END IF; + RETURN NEW; +END $$; + +CREATE FUNCTION mst2_metadata_begin_reader(sid text,lid text,instance text) +RETURNS TABLE(operation_id uuid,root_generation bigint,certificate_digest bytea) LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE l mst2_qualified_lease_binding%ROWTYPE; s record; op uuid:=gen_random_uuid(); deadline bigint; kind text; +BEGIN + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + PERFORM mst2_metadata_cleanup_expired(64); + SELECT * INTO s FROM mst2_metadata_session_row(sid,lid,instance); + IF NOT FOUND THEN RETURN; END IF; + SELECT * INTO STRICT l FROM mst2_qualified_lease_binding WHERE lease_id=lid; + deadline:=least(l.expires_at_unix,floor(extract(epoch FROM clock_timestamp()))::bigint+60); + INSERT INTO mst2_metadata_reader_operation(operation_id,lease_id,snapshot_id,session_incarnation,root_page,root_generation, + lease_epoch,hard_deadline_unix,state) VALUES(op,lid,sid,l.session_incarnation,l.metadata_root,l.root_generation,l.lease_epoch,deadline,'ACTIVE'); + FOREACH kind IN ARRAY ARRAY['REQUEST','READER'] LOOP + INSERT INTO mst2_metadata_root_anchor(anchor_id,anchor_kind,owner_key,root_page,root_generation,root_certificate_digest, + snapshot_id,session_incarnation,lease_id,reader_operation_id) VALUES(gen_random_uuid(),kind,op::text, + l.metadata_root,l.root_generation,s.certificate_digest,sid,l.session_incarnation,lid,op); + END LOOP; + RETURN QUERY SELECT op,l.root_generation,s.certificate_digest::bytea; +END $$; + +CREATE FUNCTION mst2_metadata_finish_reader(op uuid) RETURNS void LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE r mst2_metadata_reader_operation%ROWTYPE; +BEGIN + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + SELECT * INTO r FROM mst2_metadata_reader_operation WHERE operation_id=op; + IF NOT FOUND THEN RETURN; END IF; + IF r.state='ACTIVE' THEN UPDATE mst2_metadata_reader_operation SET state='FINISHED' WHERE operation_id=op; END IF; + DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=op AND anchor_kind IN ('REQUEST','READER'); + PERFORM mst2_metadata_cleanup_lease(r.lease_id); +END $$; + +CREATE FUNCTION mst2_metadata_read_source_entries(op uuid,source_id uuid,p bytea,g bigint,c bytea,names jsonb) +RETURNS TABLE(name bytea,git_oid text,kind smallint,byte_size bigint,content_digest bytea,child_root bytea, + child_generation bigint,child_certificate_digest bytea,child_attestation_id uuid,fact_state text) +LANGUAGE plpgsql VOLATILE SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE a record; profile jsonb; +BEGIN + IF jsonb_typeof(names)<>'array' OR jsonb_array_length(names)>256 + OR EXISTS(SELECT 1 FROM jsonb_array_elements_text(names) wanted + WHERE wanted !~ '^([0-9a-f]{2}){1,255}$') THEN + RAISE EXCEPTION 'qualified source-name read exceeds its bounded exact request'; END IF; + SELECT source.namespace_uuid,source.source_profile,source.tagged_tree_oid,source.source_body_digest,source.source_revision, + source.root_page,source.root_generation,source.root_certificate_digest + INTO a FROM mst2_metadata_source_root_attestation source + JOIN mst2_metadata_prepare origin ON origin.prepare_id=source.origin_prepare_id + WHERE source.attestation_id=source_id AND origin.state='COMMITTED' + AND source.root_page=p AND source.root_generation=g AND source.root_certificate_digest=c + AND source.namespace_uuid='$NAMESPACE_UUID$'::uuid; + IF NOT FOUND THEN RAISE EXCEPTION 'qualified selected directory has no definitive source attestation'; END IF; + SELECT mst2_metadata_native_profile(session.prepare_id) INTO profile + FROM mst2_metadata_reader_operation reader JOIN mst2_qualified_session_incarnation session + ON session.snapshot_id=reader.snapshot_id AND session.session_incarnation=reader.session_incarnation + WHERE reader.operation_id=op AND reader.state='ACTIVE' + AND reader.hard_deadline_unix>floor(extract(epoch FROM clock_timestamp()))::bigint + AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor anchor WHERE anchor.reader_operation_id=reader.operation_id + AND anchor.anchor_kind='READER' AND anchor.root_page=reader.root_page AND anchor.root_generation=reader.root_generation); + IF NOT FOUND OR profile IS DISTINCT FROM a.source_profile + OR NOT mst2_metadata_root_live(a.root_page,a.root_generation,a.root_certificate_digest) THEN + RAISE EXCEPTION 'qualified source read lost its exact reader profile and current directory'; END IF; + IF NOT $CORE_SCHEMA$.mst2_route_source_tree_matches(split_part(a.tagged_tree_oid,':',2),a.source_revision,a.source_body_digest) THEN + RAISE EXCEPTION 'qualified selected directory source body changed'; END IF; + RETURN QUERY SELECT reference.name,reference.git_oid,reference.kind,reference.byte_size,reference.content_digest, + reference.child_root,reference.child_generation,reference.child_certificate_digest,child.attestation_id, + CASE WHEN reference.kind=4 THEN CASE WHEN child.attestation_id IS NULL THEN 'SOURCE_UNAVAILABLE' ELSE 'READY' END + WHEN fact.git_oid IS NULL THEN 'MISSING' + WHEN fact.state<>'VERIFIED' OR fact.verification_version NOT IN (1,2) + OR fact.size NOT BETWEEN 0 AND 8796093022208 OR octet_length(fact.raw_sha256)<>32 THEN 'INVALID' + WHEN fact.verification_version=1 THEN 'MISSING' + WHEN fact.size IS DISTINCT FROM reference.byte_size OR reference.kind=3 AND fact.size NOT BETWEEN 1 AND 4095 + OR CASE WHEN octet_length(fact.raw_sha256)=32 THEN fact.raw_sha256 ELSE NULL END + IS DISTINCT FROM reference.content_digest THEN 'INVALID' + ELSE 'READY' END + FROM (SELECT DISTINCT decode(value,'hex') AS name FROM jsonb_array_elements_text(names)) wanted + JOIN mst2_metadata_source_entry_reference reference ON reference.attestation_id=source_id AND reference.name=wanted.name + LEFT JOIN LATERAL ( + SELECT verified.git_oid,verified.state,verified.verification_version,verified.size,verified.raw_sha256 + FROM $CORE_SCHEMA$.mst2_verified_object verified WHERE reference.kind<>4 AND verified.storage_domain='git' + AND verified.object_kind='blob' AND verified.git_oid=split_part(reference.git_oid,':',2) + FOR SHARE OF verified NOWAIT + ) fact ON true + LEFT JOIN LATERAL ( + SELECT candidate.attestation_id FROM mst2_metadata_source_root_attestation candidate + JOIN mst2_metadata_prepare origin ON origin.prepare_id=candidate.origin_prepare_id + JOIN $CORE_SCHEMA$.mega_tree source_tree ON source_tree.tree_id=split_part(candidate.tagged_tree_oid,':',2) + WHERE reference.kind=4 AND candidate.namespace_uuid=a.namespace_uuid AND origin.state='COMMITTED' + AND candidate.tagged_tree_oid=reference.git_oid AND candidate.source_profile=a.source_profile + AND candidate.root_page=reference.child_root AND candidate.root_generation=reference.child_generation + AND candidate.root_certificate_digest=reference.child_certificate_digest + AND $CORE_SCHEMA$.mst2_route_source_tree_matches(split_part(candidate.tagged_tree_oid,':',2),candidate.source_revision,candidate.source_body_digest) + AND mst2_metadata_root_live(candidate.root_page,candidate.root_generation,candidate.root_certificate_digest) + ORDER BY candidate.attestation_id LIMIT 1 + ) child ON true ORDER BY reference.name; +END $$; + +CREATE FUNCTION mst2_metadata_cleanup_expired(maximum integer DEFAULT 64) RETURNS void LANGUAGE plpgsql VOLATILE +SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE item record; now_unix bigint; bound integer:=least(64,greatest(0,maximum)); +BEGIN + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + now_unix:=floor(extract(epoch FROM clock_timestamp()))::bigint; + FOR item IN SELECT * FROM ( + (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline + FROM mst2_metadata_reader_operation WHERE state='ACTIVE' AND hard_deadline_unix<=now_unix + ORDER BY hard_deadline_unix,operation_id LIMIT bound) + UNION ALL + (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix + FROM mst2_qualified_lease_binding WHERE state='ACTIVE' AND expires_at_unix<=now_unix + ORDER BY expires_at_unix,lease_id LIMIT bound) + ) expired ORDER BY deadline,kind,owner LIMIT bound LOOP + IF item.kind='READER' THEN + UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND state='ACTIVE'; + DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND anchor_kind IN ('REQUEST','READER'); + ELSE + UPDATE mst2_qualified_lease_binding SET state='EXPIRED',lease_epoch=lease_epoch+1 WHERE lease_id=item.owner AND state='ACTIVE'; + END IF; + PERFORM mst2_metadata_cleanup_lease(item.lease_id); + END LOOP; +END $$; + +CREATE FUNCTION mst2_metadata_gc_owner_cleanup(maximum integer) +RETURNS TABLE(examined bigint,readers_expired bigint,leases_expired bigint,prepares_aborted bigint, + handovers_retired bigint,orphans_retired bigint) +LANGUAGE plpgsql VOLATILE SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE bound integer:=maximum; item record; q mst2_metadata_prepare%ROWTYPE; now_unix bigint; +BEGIN + PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); + IF maximum IS NULL OR maximum NOT BETWEEN 0 AND 64 THEN RAISE EXCEPTION 'qualified owner cleanup budget must be 0..=64'; END IF; + examined:=0; readers_expired:=0; leases_expired:=0; prepares_aborted:=0; handovers_retired:=0; orphans_retired:=0; + now_unix:=floor(extract(epoch FROM clock_timestamp()))::bigint; + FOR item IN SELECT * FROM ( + (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline + FROM mst2_metadata_reader_operation WHERE state='ACTIVE' AND hard_deadline_unix<=now_unix + ORDER BY hard_deadline_unix,operation_id LIMIT bound) + UNION ALL + (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix FROM mst2_qualified_lease_binding + WHERE state='ACTIVE' AND expires_at_unix<=now_unix ORDER BY expires_at_unix,lease_id LIMIT bound) + UNION ALL + (SELECT 'PREPARE'::text,prepare_id,NULL::text,floor(extract(epoch FROM orphan_expires_at))::bigint + FROM mst2_metadata_prepare WHERE (state='PREPARING' OR state='COMMITTED' AND coverage_retired_at IS NULL) + AND orphan_expires_at<=clock_timestamp() ORDER BY orphan_expires_at,prepare_id LIMIT bound) + ) expired ORDER BY deadline,kind,owner LIMIT bound LOOP + examined:=examined+1; + IF item.kind='READER' THEN + UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND state='ACTIVE'; + IF NOT FOUND THEN RAISE EXCEPTION 'qualified expired reader changed behind its mutation barrier'; END IF; + readers_expired:=readers_expired+1; + DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND anchor_kind IN ('REQUEST','READER'); + PERFORM mst2_metadata_cleanup_lease(item.lease_id); + ELSIF item.kind='LEASE' THEN + UPDATE mst2_qualified_lease_binding SET state='EXPIRED',lease_epoch=lease_epoch+1 WHERE lease_id=item.owner AND state='ACTIVE'; + IF NOT FOUND THEN RAISE EXCEPTION 'qualified expired lease changed behind its mutation barrier'; END IF; + leases_expired:=leases_expired+1; PERFORM mst2_metadata_cleanup_lease(item.lease_id); + ELSE + SELECT * INTO STRICT q FROM mst2_metadata_prepare WHERE prepare_id=item.owner; + IF q.state='PREPARING' THEN + DELETE FROM mst2_metadata_graph_root WHERE prepare_id=q.prepare_id; + UPDATE mst2_metadata_prepare SET state='ABORTED',aborted_at=clock_timestamp() WHERE prepare_id=q.prepare_id; + DELETE FROM mst2_metadata_root_anchor WHERE prepare_id=q.prepare_id AND anchor_kind IN ('PREPARE','REUSE'); + prepares_aborted:=prepares_aborted+1; + ELSIF mst2_metadata_session_covers_prepare(q.prepare_id) THEN + UPDATE mst2_metadata_prepare SET coverage_retired_at=clock_timestamp() WHERE prepare_id=q.prepare_id; + DELETE FROM mst2_metadata_root_anchor WHERE prepare_id=q.prepare_id AND anchor_kind IN ('PREPARE','REUSE'); + handovers_retired:=handovers_retired+1; + ELSIF mst2_metadata_orphan_prepare_eligible(q.prepare_id) THEN + UPDATE mst2_metadata_prepare SET coverage_retired_at=clock_timestamp() WHERE prepare_id=q.prepare_id; + DELETE FROM mst2_metadata_root_anchor WHERE prepare_id=q.prepare_id AND anchor_kind IN ('PREPARE','REUSE'); + orphans_retired:=orphans_retired+1; + END IF; + END IF; + END LOOP; + RETURN NEXT; +END $$; + +CREATE FUNCTION mst2_route_family_registration_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE +SET search_path=$CORE_SCHEMA$,pg_catalog,pg_temp AS $$ +DECLARE g mst2_metadata_namespace%ROWTYPE; stamp record; +BEGIN + SELECT * INTO STRICT g FROM mst2_metadata_namespace WHERE singleton=1; + IF (SELECT count(*) FROM mst2_metadata_namespace)<>1 OR NEW.singleton IS NOT NULL + OR substr(NEW.namespace_uuid::text,15,1)<>'4' OR substr(NEW.namespace_uuid::text,20,1) NOT IN ('8','9','a','b') + OR NEW.graph_domain<>'qualified-v1' OR NEW.family_identity<>'v3-rooted-qualified-1' + OR NEW.admission_state<>'ROOTED_Q_ADMITTED' OR NEW.collector_state<>'ENABLED' + OR ROW(NEW.core_schema,NEW.core_schema_oid,NEW.database_name,NEW.database_oid,NEW.storage_uuid, + NEW.server_address,NEW.server_port,NEW.mono_lock_key2) IS DISTINCT FROM + ROW(g.core_schema,g.core_schema_oid,g.database_name,g.database_oid,g.storage_uuid, + g.server_address,g.server_port,g.mono_lock_key2) + OR NEW.metadata_schema IS DISTINCT FROM 'mst2q_'||replace(NEW.namespace_uuid::text,'-','') + OR NEW.metadata_storage_uuid IS NOT DISTINCT FROM g.storage_uuid + OR NEW.metadata_storage_uuid IS DISTINCT FROM (NEW.metadata_storage_uuid::uuid)::text + OR substr(NEW.metadata_storage_uuid,15,1)<>'4' OR substr(NEW.metadata_storage_uuid,20,1) NOT IN ('8','9','a','b') + OR NOT EXISTS(SELECT 1 FROM pg_catalog.pg_namespace n + WHERE n.oid=NEW.metadata_schema_oid AND n.nspname=NEW.metadata_schema) + OR NEW.implementation_fingerprint IS DISTINCT FROM decode('$IMPLEMENTATION_SHA$','hex') + OR NOT mst2_route_lock_held(1297043024,g.mono_lock_key2) + OR NOT mst2_route_lock_held(1296718001,pg_catalog.hashtext(g.core_schema)) + OR NOT mst2_route_lock_held(1296717362,pg_catalog.hashtext(NEW.metadata_schema)) + OR NOT mst2_route_lock_held(1296717362,pg_catalog.hashtext(g.metadata_schema)) THEN + RAISE EXCEPTION 'qualified namespace registration has no exact admitted rooted family scope and lock set'; + END IF; + EXECUTE pg_catalog.format('SELECT * FROM %I.mst2_metadata_family_identity WHERE singleton=1',NEW.metadata_schema) + INTO STRICT stamp; + IF ROW(stamp.namespace_uuid,stamp.storage_uuid,stamp.core_schema_oid,stamp.metadata_schema_oid, + stamp.family_identity,stamp.implementation_fingerprint) IS DISTINCT FROM + ROW(NEW.namespace_uuid,NEW.metadata_storage_uuid,NEW.core_schema_oid,NEW.metadata_schema_oid, + NEW.family_identity,NEW.implementation_fingerprint) + OR NEW.catalog_fingerprint IS DISTINCT FROM mst2_route_family_catalog(NEW.core_schema_oid,NEW.metadata_schema_oid) THEN + RAISE EXCEPTION 'qualified namespace registration fingerprint disagrees with its physical family'; + END IF; + IF NOT EXISTS(SELECT 1 FROM mst2_qualified_family_policy p WHERE p.singleton=1 + AND p.implementation_fingerprint=NEW.implementation_fingerprint + AND p.authority_catalog=mst2_route_family_catalog(NEW.core_schema_oid,0::oid,NEW.metadata_schema_oid) + AND p.expected_shape=mst2_route_family_shape(NEW.metadata_schema_oid,NEW.namespace_uuid,NEW.metadata_storage_uuid)) THEN + RAISE EXCEPTION 'qualified namespace does not have the trusted complete physical family shape'; + END IF; + RETURN NEW; +END $$; diff --git a/src/jupiter/storage/qualified_metadata_serving.sql b/src/jupiter/storage/qualified_metadata_serving.sql index fa3d729e..56ae3c8f 100644 --- a/src/jupiter/storage/qualified_metadata_serving.sql +++ b/src/jupiter/storage/qualified_metadata_serving.sql @@ -263,30 +263,7 @@ END $$; CREATE TRIGGER mst2_metadata_lease_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_qualified_lease_binding FOR EACH ROW EXECUTE FUNCTION mst2_metadata_lease_guard(); -CREATE FUNCTION mst2_metadata_reader_guard() RETURNS trigger LANGUAGE plpgsql VOLATILE -SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ -DECLARE l mst2_qualified_lease_binding%ROWTYPE; now_unix bigint:=floor(extract(epoch FROM clock_timestamp()))::bigint; -BEGIN - IF TG_OP='DELETE' THEN RAISE EXCEPTION 'qualified reader operation history is immutable'; END IF; - IF TG_OP='UPDATE' THEN - IF (to_jsonb(NEW)-'state') IS DISTINCT FROM (to_jsonb(OLD)-'state') - OR OLD.state<>'ACTIVE' AND NEW.state<>OLD.state OR NEW.state NOT IN ('ACTIVE','FINISHED','EXPIRED') - OR NEW.state='EXPIRED' AND OLD.hard_deadline_unix>now_unix THEN - RAISE EXCEPTION 'qualified reader identity cannot change or be prematurely expired'; END IF; - RETURN NEW; - END IF; - SELECT * INTO l FROM mst2_qualified_lease_binding WHERE lease_id=NEW.lease_id AND state='ACTIVE' AND expires_at_unix>now_unix; - IF NOT FOUND OR NEW.state<>'ACTIVE' OR substr(NEW.operation_id::text,15,1)<>'4' - OR substr(NEW.operation_id::text,20,1) NOT IN ('8','9','a','b') - OR ROW(NEW.snapshot_id,NEW.session_incarnation,NEW.root_page,NEW.root_generation,NEW.lease_epoch) IS DISTINCT FROM - ROW(l.snapshot_id,l.session_incarnation,l.metadata_root,l.root_generation,l.lease_epoch) - OR NEW.hard_deadline_unix<=now_unix OR NEW.hard_deadline_unix>least(l.expires_at_unix,now_unix+60) - OR NOT mst2_metadata_incarnation_proof(l.snapshot_id,l.session_incarnation,true) THEN - RAISE EXCEPTION 'qualified reader lacks its exact active lease and bounded deadline'; END IF; - RETURN NEW; -END $$; -CREATE TRIGGER mst2_metadata_reader_guard BEFORE INSERT OR UPDATE OR DELETE ON mst2_metadata_reader_operation - FOR EACH ROW EXECUTE FUNCTION mst2_metadata_reader_guard(); +$READER_LIFECYCLE_SQL$ CREATE FUNCTION mst2_metadata_serving_complete() RETURNS trigger LANGUAGE plpgsql VOLATILE SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ @@ -318,9 +295,9 @@ BEGIN AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor a WHERE a.anchor_kind='LEASE' AND a.lease_id=l.lease_id) THEN RAISE EXCEPTION 'qualified lease cannot commit without its exact final route and owned protection'; END IF; ELSE - SELECT * INTO STRICT r FROM mst2_metadata_reader_operation WHERE operation_id=NEW.operation_id; + SELECT * INTO STRICT r FROM mst2_metadata_reader_operation WHERE operation_id=NEW.operation_id AND reader_issuance=NEW.reader_issuance; IF r.state='ACTIVE' AND (r.hard_deadline_unix<=floor(extract(epoch FROM clock_timestamp()))::bigint - OR (SELECT count(*) FROM mst2_metadata_root_anchor a WHERE a.reader_operation_id=r.operation_id + OR (SELECT count(*) FROM mst2_metadata_root_anchor a WHERE a.reader_operation_id=r.operation_id AND a.reader_issuance=r.reader_issuance AND a.anchor_kind IN ('REQUEST','READER') AND a.owner_key=r.operation_id::text AND a.lease_id=r.lease_id AND a.snapshot_id=r.snapshot_id AND a.session_incarnation=r.session_incarnation AND a.root_page=r.root_page AND a.root_generation=r.root_generation)<>2) @@ -488,17 +465,17 @@ BEGIN PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); now_unix:=floor(extract(epoch FROM clock_timestamp()))::bigint; FOR item IN SELECT * FROM ( - (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline + (SELECT 'READER'::text AS kind,operation_id::text AS owner,lease_id,hard_deadline_unix AS deadline,reader_issuance FROM mst2_metadata_reader_operation WHERE state='ACTIVE' AND hard_deadline_unix<=now_unix ORDER BY hard_deadline_unix,operation_id LIMIT bound) UNION ALL - (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix + (SELECT 'LEASE'::text,lease_id,lease_id,expires_at_unix,NULL::bigint FROM mst2_qualified_lease_binding WHERE state='ACTIVE' AND expires_at_unix<=now_unix ORDER BY expires_at_unix,lease_id LIMIT bound) ) expired ORDER BY deadline,kind,owner LIMIT bound LOOP IF item.kind='READER' THEN - UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND state='ACTIVE'; - DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND anchor_kind IN ('REQUEST','READER'); + UPDATE mst2_metadata_reader_operation SET state='EXPIRED' WHERE operation_id=item.owner::uuid AND reader_issuance=item.reader_issuance AND state='ACTIVE'; + DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=item.owner::uuid AND reader_issuance=item.reader_issuance AND anchor_kind IN ('REQUEST','READER'); ELSE UPDATE mst2_qualified_lease_binding SET state='EXPIRED',lease_epoch=lease_epoch+1 WHERE lease_id=item.owner AND state='ACTIVE'; END IF; @@ -543,39 +520,6 @@ BEGIN RETURN changed; END $$; -CREATE FUNCTION mst2_metadata_begin_reader(sid text,lid text,instance text) -RETURNS TABLE(operation_id uuid,root_generation bigint,certificate_digest bytea) LANGUAGE plpgsql VOLATILE -SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ -DECLARE l mst2_qualified_lease_binding%ROWTYPE; s record; op uuid:=gen_random_uuid(); deadline bigint; kind text; -BEGIN - PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); - PERFORM mst2_metadata_cleanup_expired(64); - SELECT * INTO s FROM mst2_metadata_session_row(sid,lid,instance); - IF NOT FOUND THEN RETURN; END IF; - SELECT * INTO STRICT l FROM mst2_qualified_lease_binding WHERE lease_id=lid; - deadline:=least(l.expires_at_unix,floor(extract(epoch FROM clock_timestamp()))::bigint+60); - INSERT INTO mst2_metadata_reader_operation(operation_id,lease_id,snapshot_id,session_incarnation,root_page,root_generation, - lease_epoch,hard_deadline_unix,state) VALUES(op,lid,sid,l.session_incarnation,l.metadata_root,l.root_generation,l.lease_epoch,deadline,'ACTIVE'); - FOREACH kind IN ARRAY ARRAY['REQUEST','READER'] LOOP - INSERT INTO mst2_metadata_root_anchor(anchor_id,anchor_kind,owner_key,root_page,root_generation,root_certificate_digest, - snapshot_id,session_incarnation,lease_id,reader_operation_id) VALUES(gen_random_uuid(),kind,op::text, - l.metadata_root,l.root_generation,s.certificate_digest,sid,l.session_incarnation,lid,op); - END LOOP; - RETURN QUERY SELECT op,l.root_generation,s.certificate_digest::bytea; -END $$; - -CREATE FUNCTION mst2_metadata_finish_reader(op uuid) RETURNS void LANGUAGE plpgsql VOLATILE -SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ -DECLARE r mst2_metadata_reader_operation%ROWTYPE; -BEGIN - PERFORM $CORE_SCHEMA$.mst2_route_enter($CORE_LITERAL$); - SELECT * INTO r FROM mst2_metadata_reader_operation WHERE operation_id=op; - IF NOT FOUND THEN RETURN; END IF; - IF r.state='ACTIVE' THEN UPDATE mst2_metadata_reader_operation SET state='FINISHED' WHERE operation_id=op; END IF; - DELETE FROM mst2_metadata_root_anchor WHERE reader_operation_id=op AND anchor_kind IN ('REQUEST','READER'); - PERFORM mst2_metadata_cleanup_lease(r.lease_id); -END $$; - DROP TRIGGER mst2_01_family_closed ON mst2_qualified_session_incarnation; DROP TRIGGER mst2_01_family_closed ON mst2_qualified_lease_binding; DROP TRIGGER mst2_01_family_closed ON mst2_metadata_reader_operation; diff --git a/src/jupiter/storage/qualified_metadata_source_read.sql b/src/jupiter/storage/qualified_metadata_source_read.sql index 232a0e6d..f2442e1b 100644 --- a/src/jupiter/storage/qualified_metadata_source_read.sql +++ b/src/jupiter/storage/qualified_metadata_source_read.sql @@ -68,7 +68,7 @@ END $$; CREATE CONSTRAINT TRIGGER mst2_metadata_source_entries_complete AFTER INSERT ON mst2_metadata_source_root_attestation DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION mst2_metadata_source_entries_complete(); -CREATE FUNCTION mst2_metadata_read_source_entries(op uuid,source_id uuid,p bytea,g bigint,c bytea,names jsonb) +CREATE FUNCTION mst2_metadata_read_source_entries(op uuid,issuance bigint,source_id uuid,p bytea,g bigint,c bytea,names jsonb) RETURNS TABLE(name bytea,git_oid text,kind smallint,byte_size bigint,content_digest bytea,child_root bytea, child_generation bigint,child_certificate_digest bytea,child_attestation_id uuid,fact_state text) LANGUAGE plpgsql VOLATILE SET search_path=$Q_SCHEMA$,pg_catalog,pg_temp AS $$ @@ -89,9 +89,9 @@ BEGIN SELECT mst2_metadata_native_profile(session.prepare_id) INTO profile FROM mst2_metadata_reader_operation reader JOIN mst2_qualified_session_incarnation session ON session.snapshot_id=reader.snapshot_id AND session.session_incarnation=reader.session_incarnation - WHERE reader.operation_id=op AND reader.state='ACTIVE' + WHERE reader.operation_id=op AND reader.reader_issuance=issuance AND reader.state='ACTIVE' AND reader.hard_deadline_unix>floor(extract(epoch FROM clock_timestamp()))::bigint - AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor anchor WHERE anchor.reader_operation_id=reader.operation_id + AND EXISTS(SELECT 1 FROM mst2_metadata_root_anchor anchor WHERE anchor.reader_operation_id=reader.operation_id AND anchor.reader_issuance=reader.reader_issuance AND anchor.anchor_kind='READER' AND anchor.root_page=reader.root_page AND anchor.root_generation=reader.root_generation); IF NOT FOUND OR profile IS DISTINCT FROM a.source_profile OR NOT mst2_metadata_root_live(a.root_page,a.root_generation,a.root_certificate_digest) THEN