From 1fd630eed4850388eb3dea1e388261f6cf04c1d7 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 13:16:29 +0800 Subject: [PATCH 1/5] Add controlled native publication initialization command --- src/commands/mod.rs | 2 +- src/commands/service/mod.rs | 17 +- .../service/native_publication_init.rs | 199 ++++++++++ .../native_publication_init_preflight.rs | 162 ++++++++ .../service/native_publication_push_tests.rs | 47 ++- .../storage/native_publication_storage.rs | 119 +++++- .../storage/native_publication_tests.rs | 360 ++++++++++++++++++ 7 files changed, 899 insertions(+), 7 deletions(-) create mode 100644 src/commands/service/native_publication_init.rs create mode 100644 src/commands/service/native_publication_init_preflight.rs diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 92361218..4c7fc6d3 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -119,7 +119,7 @@ pub(crate) fn builtin_exec(cmd: &str) -> Option { pub(crate) fn load_mode(cmd: &str, args: &ArgMatches) -> Option { match cmd { "service" => match args.subcommand_name() { - Some("init") => Some(LoadMode::ParsedExistingConfig), + Some("init" | "native-publication-init") => Some(LoadMode::ParsedExistingConfig), _ => Some(LoadMode::FullAppContext), }, "debug" => Some(LoadMode::FullAppContext), diff --git a/src/commands/service/mod.rs b/src/commands/service/mod.rs index 666dfb7c..4e11019d 100644 --- a/src/commands/service/mod.rs +++ b/src/commands/service/mod.rs @@ -17,12 +17,19 @@ use crate::{ pub mod http; pub mod init; pub mod multi; +pub mod native_publication_init; pub mod ssh; const CONFIG_RELOAD_POLL_INTERVAL: Duration = Duration::from_secs(5); pub fn cli() -> Command { - let subcommands = vec![init::cli(), http::cli(), ssh::cli(), multi::cli()]; + let subcommands = vec![ + init::cli(), + native_publication_init::cli(), + http::cli(), + ssh::cli(), + multi::cli(), + ]; Command::new("service") .about("Start different kinds of server: for example https or ssh") .subcommands(subcommands) @@ -30,6 +37,9 @@ pub fn cli() -> Command { #[tokio::main] pub(crate) async fn exec(ctx: CommandContext, args: &ArgMatches) -> MegaResult { + if let Some(("native-publication-init", subcommand_args)) = args.subcommand() { + return native_publication_init::exec(ctx, subcommand_args).await; + } let config_path = ctx.config_path.clone(); let config_profile_path = ctx.config_profile_path.clone(); let config = require_config(ctx, "service")?; @@ -256,7 +266,10 @@ mod tests { .map(|cmd| cmd.get_name().to_owned()) .collect::>(); - assert_eq!(names, vec!["init", "http", "ssh", "multi"]); + assert_eq!( + names, + vec!["init", "native-publication-init", "http", "ssh", "multi"] + ); } #[test] diff --git a/src/commands/service/native_publication_init.rs b/src/commands/service/native_publication_init.rs new file mode 100644 index 00000000..6ced8b9a --- /dev/null +++ b/src/commands/service/native_publication_init.rs @@ -0,0 +1,199 @@ +use std::sync::Arc; + +use clap::{Arg, ArgAction, ArgMatches, Command}; +use sea_orm_migration::MigratorTrait; + +use crate::{ + commands::{CommandContext, require_config}, + common::errors::{MegaError, MegaResult}, + config::{DbConfig, loader::ConfigSource}, + jupiter::{ + migration::Migrator, + storage::{ + base_storage::BaseStorage, + init::{postgres_connection, read_only_database_connection}, + mono_storage::MonoStorage, + native_publication_storage::NativeRoot, + }, + }, +}; + +#[path = "native_publication_init_preflight.rs"] +mod preflight; +use preflight::InitializationTarget; + +pub fn cli() -> Command { + let mut command = Command::new("native-publication-init") + .about("Prepare native publication in a stopped, paused and drained SHA-1 deployment"); + for (name, help) in [ + ("instance", "Deployment UUID; must match mst2.instance_uuid"), + ( + "expected-root-commit", + "Exact current native root commit object ID", + ), + ( + "expected-root-tree", + "Exact current native root tree object ID", + ), + ] { + command = command.arg(Arg::new(name).long(name).required(true).help(help)); + } + for (name, help) in [ + ("yes", "Confirm preparation of the native publication head"), + ( + "writers-stopped", + "Confirm all old writer processes have been stopped", + ), + ] { + command = command.arg( + Arg::new(name) + .long(name) + .action(ArgAction::SetTrue) + .required(true) + .help(help), + ); + } + command +} + +pub(crate) async fn exec(ctx: CommandContext, args: &ArgMatches) -> MegaResult { + if !ctx + .config_summary + .as_ref() + .is_some_and(|summary| matches!(summary.source, ConfigSource::Cli | ConfigSource::Env)) + { + return Err(MegaError::Other( + "MST2_NATIVE_INIT_CONFIG_REQUIRED: name the deployment with --config or MEGA_CONFIG" + .into(), + )); + } + if !args.get_flag("yes") || !args.get_flag("writers-stopped") { + return Err(MegaError::Other( + "MST2_NATIVE_INIT_CONFIRMATION_REQUIRED: --yes and --writers-stopped are required" + .into(), + )); + } + let config = require_config(ctx, "service native-publication-init")?; + config.validate()?; + if !config.mst2.enabled || !config.mst2.publication_enabled { + return Err(MegaError::Other( + "MST2_NATIVE_INIT_DISABLED: mst2.enabled and mst2.publication_enabled must be true" + .into(), + )); + } + let argument = |name| { + args.get_one::(name) + .map(String::as_str) + .ok_or_else(|| MegaError::Other(format!("--{name} is required"))) + }; + let target = InitializationTarget::parse( + config.monorepo.object_format.as_str(), + config.mst2.instance_uuid.as_deref(), + argument("instance")?, + argument("expected-root-commit")?, + argument("expected-root-tree")?, + ) + .map_err(|error| MegaError::Other(error.to_string()))?; + ensure_schema_current(&config.database).await?; + let db = Arc::new(postgres_connection(&config.database).await?); + let mono = MonoStorage { + base: BaseStorage::new(db.clone()), + }; + let result = mono + .initialize_native_publication_for_maintenance( + &target.instance, + &NativeRoot { + commit: target.commit, + tree: target.tree, + }, + ) + .await + .map_err(|error| MegaError::Other(error.to_string())); + let close = db.close_by_ref().await; + result?; + close?; + tracing::info!( + instance = %target.instance, + state = "INITIALIZING", + "Native publication initialization prepared; queue remains paused" + ); + Ok(()) +} + +async fn ensure_schema_current(config: &DbConfig) -> MegaResult { + let db = read_only_database_connection(config).await?; + let pending = Migrator::get_pending_migrations_read_only(&db).await; + let _ = db.close().await; + let pending = pending.map_err(|error| { + MegaError::Other(format!( + "MST2_NATIVE_INIT_SCHEMA_MISMATCH: cannot confirm the current schema: {error}" + )) + })?; + if !pending.is_empty() { + return Err(MegaError::Other(format!( + "MST2_NATIVE_INIT_SCHEMA_MISMATCH: {} migrations are pending; this command does not migrate", + pending.len() + ))); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::commands::{LoadMode, load_mode}; + + #[test] + fn maintenance_cli_requires_each_confirmation_and_fixed_target() { + let arguments = [ + "native-publication-init", + "--yes", + "--writers-stopped", + "--instance", + "instance", + "--expected-root-commit", + "commit", + "--expected-root-tree", + "tree", + ]; + assert!(cli().try_get_matches_from(arguments).is_ok()); + for remove in [1, 2, 3, 5, 7] { + let count = if remove <= 2 { 1 } else { 2 }; + let mut incomplete = arguments.to_vec(); + incomplete.drain(remove..remove + count); + assert!(cli().try_get_matches_from(incomplete).is_err()); + } + let mut service_arguments = vec!["service"]; + service_arguments.extend(arguments); + let args = super::super::cli() + .try_get_matches_from(service_arguments) + .unwrap(); + assert_eq!( + load_mode("service", &args), + Some(LoadMode::ParsedExistingConfig) + ); + } + + #[tokio::test] + async fn maintenance_command_refuses_unnamed_config_before_any_database_io() { + let args = cli() + .try_get_matches_from([ + "native-publication-init", + "--yes", + "--writers-stopped", + "--instance", + "instance", + "--expected-root-commit", + "commit", + "--expected-root-tree", + "tree", + ]) + .unwrap(); + let error = exec(CommandContext::default(), &args).await.unwrap_err(); + assert!( + error + .to_string() + .starts_with("MST2_NATIVE_INIT_CONFIG_REQUIRED") + ); + } +} diff --git a/src/commands/service/native_publication_init_preflight.rs b/src/commands/service/native_publication_init_preflight.rs new file mode 100644 index 00000000..360751b6 --- /dev/null +++ b/src/commands/service/native_publication_init_preflight.rs @@ -0,0 +1,162 @@ +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub(super) enum InitializationTargetError { + #[error( + "MST2_NATIVE_INIT_UNSUPPORTED_OBJECT_FORMAT: {0}; this maintenance entry supports sha1 only" + )] + UnsupportedObjectFormat(String), + #[error("MST2_NATIVE_INIT_INVALID_INSTANCE: both deployment instances must be non-nil UUIDs")] + InvalidInstance, + #[error("MST2_NATIVE_INIT_INSTANCE_MISMATCH: --instance differs from mst2.instance_uuid")] + InstanceMismatch, + #[error( + "MST2_NATIVE_INIT_INVALID_ROOT: expected commit and tree must be canonical SHA-1 object IDs" + )] + InvalidRoot, +} + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct InitializationTarget { + pub(super) instance: String, + pub(super) commit: String, + pub(super) tree: String, +} + +impl InitializationTarget { + pub(super) fn parse( + object_format: &str, + configured_instance: Option<&str>, + requested_instance: &str, + commit: &str, + tree: &str, + ) -> Result { + if object_format != "sha1" { + return Err(InitializationTargetError::UnsupportedObjectFormat( + object_format.into(), + )); + } + let parse_instance = |value: &str| { + uuid::Uuid::parse_str(value) + .ok() + .filter(|id| !id.is_nil()) + .ok_or(InitializationTargetError::InvalidInstance) + }; + let configured = + parse_instance(configured_instance.ok_or(InitializationTargetError::InvalidInstance)?)?; + let requested = parse_instance(requested_instance)?; + if configured != requested { + return Err(InitializationTargetError::InstanceMismatch); + } + if [commit, tree].iter().any(|value| { + value.len() != 40 + || !value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + }) { + return Err(InitializationTargetError::InvalidRoot); + } + Ok(Self { + instance: configured.to_string(), + commit: commit.into(), + tree: tree.into(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const INSTANCE: &str = "6ab219b0-4275-45ba-9d7b-7b0b633018cd"; + + #[test] + fn target_is_canonical_and_bound_to_the_configured_instance() { + let target = InitializationTarget::parse( + "sha1", + Some(INSTANCE), + &INSTANCE.to_uppercase(), + &"a".repeat(40), + &"b".repeat(40), + ) + .unwrap(); + assert_eq!(target.instance, INSTANCE); + assert_eq!(target.commit, "a".repeat(40)); + assert_eq!(target.tree, "b".repeat(40)); + } + + #[test] + fn target_rejects_other_formats_without_inferring_from_equal_width() { + for format in ["sha256", "blake3", "unknown"] { + assert_eq!( + InitializationTarget::parse( + format, + Some(INSTANCE), + INSTANCE, + &"a".repeat(64), + &"b".repeat(64), + ), + Err(InitializationTargetError::UnsupportedObjectFormat( + format.into() + )) + ); + } + } + + #[test] + fn target_rejects_nil_missing_invalid_or_different_instances() { + for instance in [ + None, + Some("invalid"), + Some("00000000-0000-0000-0000-000000000000"), + ] { + assert_eq!( + InitializationTarget::parse( + "sha1", + instance, + INSTANCE, + &"a".repeat(40), + &"b".repeat(40) + ), + Err(InitializationTargetError::InvalidInstance) + ); + } + for instance in ["invalid", "00000000-0000-0000-0000-000000000000"] { + assert_eq!( + InitializationTarget::parse( + "sha1", + Some(INSTANCE), + instance, + &"a".repeat(40), + &"b".repeat(40) + ), + Err(InitializationTargetError::InvalidInstance) + ); + } + assert_eq!( + InitializationTarget::parse( + "sha1", + Some(INSTANCE), + "11111111-2222-4333-8444-555555555555", + &"a".repeat(40), + &"b".repeat(40) + ), + Err(InitializationTargetError::InstanceMismatch) + ); + } + + #[test] + fn target_rejects_noncanonical_or_wrong_width_object_ids() { + for value in [ + "a".repeat(39), + "a".repeat(64), + "A".repeat(40), + "z".repeat(40), + ] { + for (commit, tree) in [(&value, &"b".repeat(40)), (&"a".repeat(40), &value)] { + assert_eq!( + InitializationTarget::parse("sha1", Some(INSTANCE), INSTANCE, commit, tree), + Err(InitializationTargetError::InvalidRoot) + ); + } + } + } +} diff --git a/src/jupiter/service/native_publication_push_tests.rs b/src/jupiter/service/native_publication_push_tests.rs index f24f591c..3bb41a9b 100644 --- a/src/jupiter/service/native_publication_push_tests.rs +++ b/src/jupiter/service/native_publication_push_tests.rs @@ -41,7 +41,15 @@ async fn native_fixture() -> (tempfile::TempDir, crate::jupiter::storage::Storag mono.save_refs(mega_refs::Model::new( path.clone(), MEGA_BRANCH_NAME.to_owned(), tip.id.to_string(), child.id.to_string(), false, ), None).await.unwrap(); - storage.mono_storage().initialize_native_publication(NATIVE_INSTANCE).await.unwrap(); + storage.push_queue_service.push_queue_storage.set_control_flags(Some(true), None, None).await.unwrap(); + mono.initialize_native_publication_for_maintenance( + NATIVE_INSTANCE, + &crate::jupiter::storage::native_publication_storage::NativeRoot { + commit: root_commit.id.to_string(), tree: root_tree.id.to_string(), + }, + ).await.unwrap(); + assert!(mono.read_native_publication_head(NATIVE_INSTANCE).await.is_err()); + storage.push_queue_service.push_queue_storage.set_control_flags(Some(false), None, None).await.unwrap(); (temp, storage, tip, path) } @@ -82,6 +90,43 @@ async fn real_same_tree_push_advances_one_global_certificate_and_n0_replay_stays assert_eq!(mst2_native_publication::Entity::find().count(mono.get_connection()).await.unwrap(),1); } +#[tokio::test] +async fn maintenance_cannot_reinitialize_published_history_after_head_loss() { + let (_temp, storage, tip, path) = native_fixture().await; + let mono = storage.mono_storage(); + let (new, payload) = save_same_tree_commit(&storage, &tip).await; + let id = wh03_enqueue_push(&storage, &path, &tip.id.to_string(), &new, &payload).await; + assert!(matches!(wh03_exec(&storage, id).await, ExecuteOutcome::Done { .. })); + let head = mono.read_native_publication_head(NATIVE_INSTANCE).await.unwrap(); + let certificates = mst2_native_publication::Entity::find().all(mono.get_connection()).await.unwrap(); + let receipts = mst2_publication::Entity::find().all(mono.get_connection()).await.unwrap(); + let outbox = mst2_publication_outbox::Entity::find().all(mono.get_connection()).await.unwrap(); + assert_eq!(certificates.len(), 1); + storage.push_queue_service.push_queue_storage.set_control_flags(Some(true), None, None).await.unwrap(); + mono.get_connection().execute_unprepared("DELETE FROM mst2_native_head").await.unwrap(); + for instance in [NATIVE_INSTANCE, "11111111-2222-4333-8444-555555555555"] { + let error = mono.initialize_native_publication_for_maintenance(instance, &head.root).await.unwrap_err(); + assert!(error.to_string().contains("native publication history exists")); + } + assert_eq!(mst2_native_head::Entity::find().count(mono.get_connection()).await.unwrap(), 0); + assert_eq!(mst2_native_publication::Entity::find().all(mono.get_connection()).await.unwrap(), certificates); + assert_eq!(mst2_publication::Entity::find().all(mono.get_connection()).await.unwrap(), receipts); + assert_eq!(mst2_publication_outbox::Entity::find().all(mono.get_connection()).await.unwrap(), outbox); + assert!(storage.push_queue_service.push_queue_storage.get_control().await.unwrap().paused); + mono.get_connection().execute_unprepared("DELETE FROM mst2_native_publication").await.unwrap(); + for instance in [NATIVE_INSTANCE, "11111111-2222-4333-8444-555555555555"] { + let error = mono.initialize_native_publication_for_maintenance(instance, &head.root).await.unwrap_err(); + assert!(error.to_string().contains("native publication history exists")); + } + assert_eq!(mst2_native_head::Entity::find().count(mono.get_connection()).await.unwrap(), 0); + assert_eq!(mst2_native_publication::Entity::find().count(mono.get_connection()).await.unwrap(), 0); + assert_eq!(mst2_publication::Entity::find().all(mono.get_connection()).await.unwrap(), receipts); + assert_eq!(mst2_publication_outbox::Entity::find().all(mono.get_connection()).await.unwrap(), outbox); + assert!(storage.push_queue_service.push_queue_storage.get_control().await.unwrap().paused); + let root = mono.get_main_ref("/").await.unwrap().unwrap(); + assert_eq!((root.ref_commit_hash, root.ref_tree_hash), (head.root.commit, head.root.tree)); +} + #[tokio::test] async fn real_merge_fails_closed_when_native_publication_is_enabled() { let (_temp, storage, tip, path) = native_fixture().await; diff --git a/src/jupiter/storage/native_publication_storage.rs b/src/jupiter/storage/native_publication_storage.rs index 6ccb3fc9..c106c101 100644 --- a/src/jupiter/storage/native_publication_storage.rs +++ b/src/jupiter/storage/native_publication_storage.rs @@ -281,7 +281,104 @@ impl MonoStorage { Ok(head) } - /// Maintenance-only, with stopped old writers and a drained queue; never invoked by resolve. + /// The command confirms stopped writers; this transaction checks the + /// paused admission gate, drained queue and exact native root. + pub(crate) async fn initialize_native_publication_for_maintenance( + &self, + instance: &str, + expected_root: &NativeRoot, + ) -> Result<(), PublicationReceiptError> { + let instance = canonical_instance(instance)?; + if uuid::Uuid::parse_str(&instance).is_ok_and(|id| id.is_nil()) { + return Err(integrity("native deployment instance must not be nil")); + } + for oid in [&expected_root.commit, &expected_root.tree] { + if oid.len() != 40 + || !oid + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(integrity( + "maintenance requires canonical SHA-1 root identities", + )); + } + } + let txn = self.get_connection().begin().await?; + if !PushQueueStorage::try_mono_write_lock(&txn) + .await + .map_err(|error| integrity(&error.to_string()))? + { + return Err(PublicationReceiptError::Conflict( + "native initialization refused while a writer holds the mono write lock".into(), + )); + } + let control = txn + .query_one_raw(Statement::from_string( + txn.get_database_backend(), + "SELECT paused, hard_stopped FROM queue_control WHERE id=1 FOR UPDATE NOWAIT" + .to_owned(), + )) + .await? + .ok_or_else(|| { + integrity( + "queue_control row missing; maintenance cannot establish admission exclusion", + ) + })?; + if !control.try_get::("", "paused")? { + return Err(integrity( + "queue admission must be paused before native initialization", + )); + } + if control.try_get::("", "hard_stopped")? { + return Err(integrity( + "hard-stopped queue must be investigated before native initialization", + )); + } + let history = txn + .query_one_raw(Statement::from_string( + txn.get_database_backend(), + "SELECT EXISTS(SELECT 1 FROM mst2_native_publication) OR \ + EXISTS(SELECT 1 FROM mst2_publication WHERE native_certificate_version IS NOT NULL) AS present" + .to_owned(), + )) + .await? + .ok_or_else(|| integrity("native history observation missing"))?; + if history.try_get::("", "present")? { + return Err(integrity( + "native publication history exists; initialization cannot repair or replace it", + )); + } + let objects = txn + .query_one_raw(Statement::from_sql_and_values( + txn.get_database_backend(), + "SELECT (SELECT count(*) FROM mega_commit WHERE commit_id=$1) AS commits, \ + (SELECT max(tree) FROM mega_commit WHERE commit_id=$1) AS commit_tree, \ + (SELECT count(*) FROM mega_tree WHERE tree_id=$2) AS trees", + [ + expected_root.commit.clone().into(), + expected_root.tree.clone().into(), + ], + )) + .await? + .ok_or_else(|| integrity("native object observation missing"))?; + if objects.try_get::("", "commits")? != 1 + || objects + .try_get::>("", "commit_tree")? + .as_deref() + != Some(expected_root.tree.as_str()) + || objects.try_get::("", "trees")? != 1 + { + return Err(integrity( + "native root requires a unique stored commit with the expected tree and a unique stored tree", + )); + } + self.initialize_native_publication_in_txn(&txn, &instance, Some(expected_root)) + .await?; + txn.commit().await?; + Ok(()) + } + + #[cfg(test)] pub(crate) async fn initialize_native_publication( &self, instance: &str, @@ -293,13 +390,30 @@ impl MonoStorage { .map_err(|error| integrity(&error.to_string()))?; txn.execute_unprepared("SELECT id FROM queue_control WHERE id=1 FOR UPDATE") .await?; - let observation = observe(&txn).await?; + self.initialize_native_publication_in_txn(&txn, &instance, None) + .await?; + txn.commit().await?; + Ok(()) + } + + async fn initialize_native_publication_in_txn( + &self, + txn: &DatabaseTransaction, + instance: &str, + expected_root: Option<&NativeRoot>, + ) -> Result<(), PublicationReceiptError> { + let observation = observe(txn).await?; if observation.head.is_some() { return Err(integrity("native head already initialized")); } let root = observation .root .ok_or_else(|| integrity("native root missing"))?; + if expected_root.is_some_and(|expected| *expected != root) { + return Err(PublicationReceiptError::Conflict( + "native root differs from the maintenance command's expected commit/tree".into(), + )); + } let row = txn.query_one_raw(Statement::from_string(txn.get_database_backend(), "SELECT count(*)::bigint AS active FROM push_queue WHERE status IN ('Queued', 'Running')".to_owned())).await? .ok_or_else(|| integrity("queue observation missing"))?; @@ -331,7 +445,6 @@ impl MonoStorage { VALUES ('/', $1, $2, $3, $4, $5, 'INITIALIZING')", [instance.into(), floor.into(), NATIVE_EPOCH.into(), root.commit.into(), root.tree.into()], )).await?; - txn.commit().await?; Ok(()) } diff --git a/src/jupiter/storage/native_publication_tests.rs b/src/jupiter/storage/native_publication_tests.rs index d7eeaa78..1e2f2da3 100644 --- a/src/jupiter/storage/native_publication_tests.rs +++ b/src/jupiter/storage/native_publication_tests.rs @@ -376,3 +376,363 @@ pub(super) fn crash_checkpoint(phase: &str) { panic!("SIGKILL did not terminate the native worker"); } } + +async fn maintenance_fixture() -> (tempfile::TempDir, MonoStorage, PushQueueStorage, NativeRoot) { + let (temp, mono, queue) = fixture().await; + mono.get_connection() + .execute_unprepared("DELETE FROM mst2_native_head") + .await + .unwrap(); + queue + .set_control_flags(Some(true), None, None) + .await + .unwrap(); + let tree = git_internal::internal::object::tree::Tree::from_tree_items(vec![]).unwrap(); + let commit = git_internal::internal::object::commit::Commit::from_tree_id( + tree.id, + vec![], + "maintenance root", + ); + mono.save_mega_trees(vec![tree.clone()], commit.id, None) + .await + .unwrap(); + mono.save_mega_commits(vec![commit.clone()], None) + .await + .unwrap(); + let root = NativeRoot { + commit: commit.id.to_string(), + tree: tree.id.to_string(), + }; + mono.get_connection().execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + "UPDATE mega_refs SET ref_commit_hash=$1,ref_tree_hash=$2 WHERE path='/' AND ref_name=$3 AND is_cl=false", + [root.commit.clone().into(), root.tree.clone().into(), MEGA_BRANCH_NAME.into()], + )).await.unwrap(); + (temp, mono, queue, root) +} + +async fn assert_no_maintenance_head(mono: &MonoStorage) { + assert_eq!( + mst2_native_head::Entity::find() + .count(mono.get_connection()) + .await + .unwrap(), + 0 + ); + assert_eq!( + mst2_native_publication::Entity::find() + .count(mono.get_connection()) + .await + .unwrap(), + 0 + ); + assert_eq!( + mst2_publication_outbox::Entity::find() + .count(mono.get_connection()) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn maintenance_initializes_only_a_floor_and_preserves_the_paused_gate() { + let (_temp, mono, queue, root) = maintenance_fixture().await; + mono.get_connection() + .execute_unprepared( + "INSERT INTO mst2_namespace_seq(namespace,sequence,epoch) VALUES('/older',23,1)", + ) + .await + .unwrap(); + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .unwrap(); + let before = mst2_native_head::Entity::find() + .one(mono.get_connection()) + .await + .unwrap() + .unwrap(); + assert_eq!(before.instance_id, INSTANCE); + assert_eq!(before.sequence, 23); + assert_eq!(before.writer_epoch, 1); + assert_eq!( + (&before.root_commit, &before.root_tree), + (&root.commit, &root.tree) + ); + assert_eq!(before.state, "INITIALIZING"); + assert_eq!(before.certificate_receipt_id, None); + assert!(queue.get_control().await.unwrap().paused); + assert!(mono.read_native_publication_head(INSTANCE).await.is_err()); + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .is_err() + ); + assert!( + mono.initialize_native_publication_for_maintenance( + "11111111-2222-4333-8444-555555555555", + &root + ) + .await + .is_err() + ); + assert_eq!( + mst2_native_head::Entity::find() + .one(mono.get_connection()) + .await + .unwrap() + .unwrap(), + before + ); + assert_eq!( + selected_ref(mono.get_connection(), "/").await.unwrap(), + Some(root) + ); + assert_eq!( + mst2_native_publication::Entity::find() + .count(mono.get_connection()) + .await + .unwrap(), + 0 + ); + assert_eq!( + mst2_publication_outbox::Entity::find() + .count(mono.get_connection()) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn maintenance_rejects_unpaused_missing_or_hard_stopped_control_without_writes() { + for state in ["unpaused", "missing", "hard-stopped"] { + let (_temp, mono, queue, root) = maintenance_fixture().await; + match state { + "unpaused" => queue + .set_control_flags(Some(false), None, None) + .await + .unwrap(), + "hard-stopped" => queue + .set_control_flags(None, Some(true), None) + .await + .unwrap(), + "missing" => { + mono.get_connection() + .execute_unprepared("DELETE FROM queue_control") + .await + .unwrap(); + } + _ => unreachable!(), + } + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .is_err(), + "{state}" + ); + assert_no_maintenance_head(&mono).await; + let control = crate::callisto::queue_control::Entity::find() + .one(mono.get_connection()) + .await + .unwrap(); + match state { + "missing" => assert!(control.is_none()), + "hard-stopped" => assert!(control.unwrap().hard_stopped), + _ => assert!(!control.unwrap().paused), + } + } +} + +#[tokio::test] +async fn maintenance_refuses_queued_and_running_work_until_drained() { + for status in ["Queued", "Running"] { + let (_temp, mono, queue, root) = maintenance_fixture().await; + queue + .set_control_flags(Some(false), None, None) + .await + .unwrap(); + let EnqueueOutcome::Inserted { id } = queue + .enqueue_atomic(EnqueueParams { + kind: PushQueueKindEnum::Push, + operation_id: "maintenance-pending", + path: "/project", + old_id: &"c".repeat(40), + new_id: &"e".repeat(40), + requester: None, + payload: serde_json::json!({"n":1,"commits":["e".repeat(40)]}), + }) + .await + .unwrap() + else { + panic!("fresh pending operation"); + }; + mono.get_connection() + .execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + "UPDATE push_queue SET status=$1::push_queue_status_enum WHERE id=$2", + [status.into(), id.into()], + )) + .await + .unwrap(); + queue + .set_control_flags(Some(true), None, None) + .await + .unwrap(); + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .is_err(), + "{status}" + ); + assert_no_maintenance_head(&mono).await; + assert!(queue.get_control().await.unwrap().paused); + } +} + +#[tokio::test] +async fn maintenance_refuses_busy_writer_or_admission_locks_and_can_retry() { + for lock in ["writer", "admission"] { + let (_temp, mono, _queue, root) = maintenance_fixture().await; + let other = mono.get_connection().begin().await.unwrap(); + if lock == "writer" { + PushQueueStorage::acquire_mono_write_lock(&other) + .await + .unwrap(); + } else { + other + .execute_unprepared("SELECT id FROM queue_control WHERE id=1 FOR UPDATE") + .await + .unwrap(); + } + let result = tokio::time::timeout( + std::time::Duration::from_secs(5), + mono.initialize_native_publication_for_maintenance(INSTANCE, &root), + ) + .await; + assert!( + result + .expect("maintenance lock refusal must be bounded") + .is_err(), + "{lock}" + ); + assert_no_maintenance_head(&mono).await; + other.rollback().await.unwrap(); + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .unwrap(); + } +} + +#[tokio::test] +async fn maintenance_rejects_wrong_or_ambiguous_root_and_invalid_counters() { + for damage in [ + "commit", + "tree", + "missing", + "ambiguous", + "negative", + "exhausted", + ] { + let (_temp, mono, _queue, mut root) = maintenance_fixture().await; + match damage { + "commit" => root.commit = "e".repeat(40), + "tree" => root.tree = "e".repeat(40), + "missing" => { + mono.get_connection() + .execute_unprepared("DELETE FROM mega_refs WHERE path='/'") + .await + .unwrap(); + } + "ambiguous" => { + // Simulate a corrupted/old schema, independently of the normal + // uniqueness protection on selected refs. + mono.get_connection() + .execute_unprepared("DROP INDEX uniq_mref_path") + .await + .unwrap(); + mono.get_connection().execute_unprepared("INSERT INTO mega_refs(id,path,ref_name,ref_commit_hash,ref_tree_hash,created_at,updated_at,is_cl) SELECT 3,path,ref_name,ref_commit_hash,ref_tree_hash,created_at,updated_at,is_cl FROM mega_refs WHERE path='/'").await.unwrap(); + } + "negative" | "exhausted" => { + let floor = if damage == "negative" { + -1_i64 + } else { + i64::MAX + }; + mono.get_connection().execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + "INSERT INTO mst2_namespace_seq(namespace,sequence,epoch) VALUES('/older',$1,1)", [floor.into()], + )).await.unwrap(); + } + _ => unreachable!(), + } + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .is_err(), + "{damage}" + ); + assert_no_maintenance_head(&mono).await; + } +} + +#[tokio::test] +async fn maintenance_rejects_nil_instance_and_non_sha1_expected_roots() { + let (_temp, mono, _queue, root) = maintenance_fixture().await; + for instance in ["invalid", "00000000-0000-0000-0000-000000000000"] { + assert!( + mono.initialize_native_publication_for_maintenance(instance, &root) + .await + .is_err() + ); + } + for commit in ["a".repeat(64), "A".repeat(40)] { + let wrong = NativeRoot { + commit, + tree: root.tree.clone(), + }; + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &wrong) + .await + .is_err() + ); + } + assert_no_maintenance_head(&mono).await; +} + +#[tokio::test] +async fn maintenance_refuses_missing_commit_tree_or_a_mismatched_commit_tree() { + for damage in ["commit", "tree", "wrong-tree"] { + let (_temp, mono, _queue, root) = maintenance_fixture().await; + let sql = match damage { + "commit" => "DELETE FROM mega_commit WHERE commit_id=$1", + "tree" => "DELETE FROM mega_tree WHERE tree_id=$1", + "wrong-tree" => "UPDATE mega_commit SET tree=repeat('e',40) WHERE commit_id=$1", + _ => unreachable!(), + }; + let id = if damage == "tree" { + &root.tree + } else { + &root.commit + }; + mono.get_connection() + .execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + sql, + [id.clone().into()], + )) + .await + .unwrap(); + assert!( + mono.initialize_native_publication_for_maintenance(INSTANCE, &root) + .await + .is_err(), + "{damage}" + ); + assert_no_maintenance_head(&mono).await; + assert_eq!( + selected_ref(mono.get_connection(), "/").await.unwrap(), + Some(root) + ); + } +} From cfb79b4a14672ee6132e0f00d18199ef67bf820d Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 13:26:48 +0800 Subject: [PATCH 2/5] fix(mst2): import maintenance storage constructor trait --- src/commands/service/native_publication_init.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/commands/service/native_publication_init.rs b/src/commands/service/native_publication_init.rs index 6ced8b9a..87002d22 100644 --- a/src/commands/service/native_publication_init.rs +++ b/src/commands/service/native_publication_init.rs @@ -10,7 +10,7 @@ use crate::{ jupiter::{ migration::Migrator, storage::{ - base_storage::BaseStorage, + base_storage::{BaseStorage, StorageConnector}, init::{postgres_connection, read_only_database_connection}, mono_storage::MonoStorage, native_publication_storage::NativeRoot, From 1935868b358bc4f83718b021d919b87c7dcd49ad Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 13:51:52 +0800 Subject: [PATCH 3/5] test(runtime): read bind failure diagnostic after child exits --- tests/integration_storage_events_runtime.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/integration_storage_events_runtime.rs b/tests/integration_storage_events_runtime.rs index 9443a443..ab8c4908 100644 --- a/tests/integration_storage_events_runtime.rs +++ b/tests/integration_storage_events_runtime.rs @@ -363,12 +363,6 @@ fn integration_storage_events_runtime_occupied_port_exits_through_cleanup() { // evidence that the cleanup tail ran. let receipt = wait_for_shutdown_receipt(&stdout_path, &stderr_path, Duration::from_secs(60)); assert_shutdown_receipt_sanitized(&receipt); - // The failure must be the port bind, not some unrelated startup error. - let combined = format!("{}\n{}", read_log(&stdout_path), read_log(&stderr_path)); - assert!( - combined.contains("failed to bind HTTP listener"), - "occupied-port case must fail at the HTTP bind:\n{combined}" - ); let status = service .wait_for_exit(Duration::from_secs(60)) .expect("occupied-port startup failure must exit on its own"); @@ -378,6 +372,13 @@ fn integration_storage_events_runtime_occupied_port_exits_through_cleanup() { read_log(&stdout_path), read_log(&stderr_path), ); + // main prints the startup error after cleanup returns; wait for exit + // before reading the final diagnostic that must identify the bind failure. + let combined = format!("{}\n{}", read_log(&stdout_path), read_log(&stderr_path)); + assert!( + combined.contains("failed to bind HTTP listener"), + "occupied-port case must fail at the HTTP bind:\n{combined}" + ); } #[test] From 116b012fea8e24a5da8cc60e7ddb966be98d9853 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 14:31:17 +0800 Subject: [PATCH 4/5] test(mst2): build valid initialization fixture and match coded error --- .../service/native_publication_init.rs | 5 ++-- .../storage/native_publication_tests.rs | 29 +++++++++++++++---- 2 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/commands/service/native_publication_init.rs b/src/commands/service/native_publication_init.rs index 87002d22..483f647c 100644 --- a/src/commands/service/native_publication_init.rs +++ b/src/commands/service/native_publication_init.rs @@ -191,9 +191,8 @@ mod tests { .unwrap(); let error = exec(CommandContext::default(), &args).await.unwrap_err(); assert!( - error - .to_string() - .starts_with("MST2_NATIVE_INIT_CONFIG_REQUIRED") + matches!(&error, MegaError::Other(message) if message.starts_with("MST2_NATIVE_INIT_CONFIG_REQUIRED")), + "unexpected error: {error}" ); } } diff --git a/src/jupiter/storage/native_publication_tests.rs b/src/jupiter/storage/native_publication_tests.rs index 1e2f2da3..89b6c555 100644 --- a/src/jupiter/storage/native_publication_tests.rs +++ b/src/jupiter/storage/native_publication_tests.rs @@ -378,6 +378,15 @@ pub(super) fn crash_checkpoint(phase: &str) { } async fn maintenance_fixture() -> (tempfile::TempDir, MonoStorage, PushQueueStorage, NativeRoot) { + use git_internal::{ + hash::HashKind, + internal::object::{ + blob::Blob, + commit::Commit, + tree::{Tree, TreeItem, TreeItemMode}, + }, + }; + let (temp, mono, queue) = fixture().await; mono.get_connection() .execute_unprepared("DELETE FROM mst2_native_head") @@ -387,12 +396,20 @@ async fn maintenance_fixture() -> (tempfile::TempDir, MonoStorage, PushQueueStor .set_control_flags(Some(true), None, None) .await .unwrap(); - let tree = git_internal::internal::object::tree::Tree::from_tree_items(vec![]).unwrap(); - let commit = git_internal::internal::object::commit::Commit::from_tree_id( - tree.id, - vec![], - "maintenance root", - ); + let blob = + Blob::from_content_bytes_with_kind(HashKind::Sha1, b"maintenance root".to_vec()).unwrap(); + let tree = Tree::from_tree_items_with_kind( + HashKind::Sha1, + vec![TreeItem::new( + TreeItemMode::Blob, + blob.id, + ".gitkeep".to_owned(), + )], + ) + .unwrap(); + let commit = + Commit::from_tree_id_with_kind(HashKind::Sha1, tree.id, vec![], "maintenance root") + .unwrap(); mono.save_mega_trees(vec![tree.clone()], commit.id, None) .await .unwrap(); From f7f6f7e109254a56a59e077a4b9f353185824fae Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 14:42:58 +0800 Subject: [PATCH 5/5] test(git): require bounded HTTP readiness before integration pushes --- tests/integration_git_cli.rs | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/tests/integration_git_cli.rs b/tests/integration_git_cli.rs index 4faa926a..82392c78 100644 --- a/tests/integration_git_cli.rs +++ b/tests/integration_git_cli.rs @@ -20,7 +20,6 @@ use std::{ collections::BTreeMap, fs, io::{Read, Write}, - net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -203,34 +202,51 @@ impl ServiceProcess { service } - fn wait_until_listening( + fn wait_until_openapi_ready( &mut self, port: u16, timeout: Duration, stdout_path: &Path, stderr_path: &Path, ) { + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(5)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .expect("build HTTP readiness client"); + let url = format!("http://127.0.0.1:{port}/api/openapi.json"); let deadline = Instant::now() + timeout; loop { - if TcpStream::connect(("127.0.0.1", port)).is_ok() { - return; - } if let Some(status) = self.child.try_wait().expect("poll service") { self.reaped = true; panic!( - "service exited before binding port {port} (status {status})\nstdout:\n{}\nstderr:\n{}", + "service exited before OpenAPI was ready on port {port} (status {status})\nstdout:\n{}\nstderr:\n{}", read_log(stdout_path), read_log(stderr_path), ); } - if Instant::now() >= deadline { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { panic!( - "service did not bind port {port} within {timeout:?}\nstdout:\n{}\nstderr:\n{}", + "service OpenAPI not ready on port {port} within {timeout:?}\nstdout:\n{}\nstderr:\n{}", read_log(stdout_path), read_log(stderr_path), ); } - sleep(Duration::from_millis(200)); + if let Ok(response) = client + .get(&url) + .timeout(remaining.min(Duration::from_secs(5))) + .send() + && response.status() == reqwest::StatusCode::OK + && Instant::now() <= deadline + { + return; + } + sleep( + deadline + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(200)), + ); } } @@ -344,7 +360,7 @@ fn boot_service_http_with_env( .stderr(Stdio::from(create_log_file(&stderr_path))); let mut service = ServiceProcess::spawn(command); - service.wait_until_listening(port, Duration::from_secs(90), &stdout_path, &stderr_path); + service.wait_until_openapi_ready(port, Duration::from_secs(90), &stdout_path, &stderr_path); (service, port, stdout_path, stderr_path) }