diff --git a/src/jupiter/storage/native_publication_storage.rs b/src/jupiter/storage/native_publication_storage.rs index 6ccb3fc9..2a5d0ff0 100644 --- a/src/jupiter/storage/native_publication_storage.rs +++ b/src/jupiter/storage/native_publication_storage.rs @@ -435,6 +435,14 @@ impl MonoStorage { "operation did not change its selected native ref", )); } + let resolved_path_tree = self + .resolve_path_tree_hash_in_txn(&root.tree, &prepared.path, txn) + .await + .map_err(|_| integrity("native path tree lookup failed"))? + .ok_or_else(|| integrity("native path tree is not materialized"))?; + if resolved_path_tree != path.tree { + return Err(integrity("native path tree does not match selected ref")); + } let next = prepared .head .token diff --git a/src/jupiter/storage/native_publication_tests.rs b/src/jupiter/storage/native_publication_tests.rs index d7eeaa78..a225eb09 100644 --- a/src/jupiter/storage/native_publication_tests.rs +++ b/src/jupiter/storage/native_publication_tests.rs @@ -1,5 +1,9 @@ use std::sync::Arc; +use git_internal::{ + hash::{ObjectHash, get_hash_kind}, + internal::object::tree::{TreeItem, TreeItemMode}, +}; use sea_orm::{DatabaseTransaction, PaginatorTrait}; use super::*; @@ -34,6 +38,24 @@ async fn fixture() -> (tempfile::TempDir, MonoStorage, PushQueueStorage) { [id.into(), path.into(), MEGA_BRANCH_NAME.into(), commit.to_string().repeat(40).into(), tree.to_string().repeat(40).into()], )).await.unwrap(); } + let child_tree = "d".repeat(40); + let root_tree = "b".repeat(40); + let project = TreeItem::new( + TreeItemMode::Tree, + ObjectHash::from_hex_for_kind(get_hash_kind(), &child_tree).unwrap(), + "project".to_owned(), + ); + for (id, tree, sub_trees) in [ + (1_i64, child_tree, Vec::new()), + (2_i64, root_tree, project.to_data()), + ] { + mono.get_connection().execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + "INSERT INTO mega_tree (id,tree_id,sub_trees,size,created_at,pack_id,pack_offset,commit_id) \ + VALUES ($1,$2,$3,0,now(),$4,0,$5)", + [id.into(), tree.into(), sub_trees.into(), "".into(), "".into()], + )).await.unwrap(); + } mono.initialize_native_publication(INSTANCE).await.unwrap(); ( temp, @@ -363,6 +385,83 @@ async fn reservation_owner_and_unchanged_selected_ref_cannot_issue_a_certificate ); } +#[tokio::test] +async fn missing_native_path_tree_keeps_head_initializing_and_rolls_back_receipt() { + let (_temp, mono, queue) = fixture().await; + let row = enqueue_claim(&queue, &"c".repeat(40), &"e".repeat(40), 1).await; + mono.get_connection() + .execute_raw(Statement::from_sql_and_values( + mono.get_connection().get_database_backend(), + "DELETE FROM mega_tree WHERE tree_id = $1", + ["d".repeat(40).into()], + )) + .await + .unwrap(); + + let txn = mono.get_connection().begin().await.unwrap(); + let request = PublicationRequest::from_trunk_queue(&row).unwrap(); + let PublicationPreparation::Prepared(origin) = + mono.begin_publication_in_txn(&txn, request).await.unwrap() + else { + panic!("fresh publication"); + }; + let native = mono + .reserve_native_publication_in_txn(&txn, &row, INSTANCE) + .await + .unwrap(); + assert!( + mono.cas_update_root_main_ref_in_txn( + &txn, + row.expected_commit_hash.as_deref(), + row.expected_tree_hash.as_deref(), + row.expected_commit_hash.as_deref().unwrap(), + row.expected_tree_hash.as_deref().unwrap(), + ) + .await + .unwrap() + ); + txn.execute_raw(Statement::from_sql_and_values( + txn.get_database_backend(), + "UPDATE mega_refs SET ref_commit_hash = $1 WHERE path='/project' AND ref_name=$2 AND is_cl=false", + [row.new_id.clone().into(), MEGA_BRANCH_NAME.into()], + )) + .await + .unwrap(); + let committed = mono + .record_publication_in_txn( + &txn, + origin, + row.expected_commit_hash.as_deref().unwrap(), + &row.new_id, + ) + .await + .unwrap(); + let error = mono + .record_native_publication_in_txn(&txn, native, &committed) + .await + .unwrap_err(); + assert!( + matches!(error, PublicationReceiptError::Integrity(message) if message.contains("path tree")) + ); + txn.rollback().await.unwrap(); + + assert!(mono.read_native_publication_head(INSTANCE).await.is_err()); + 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 + ); +} + #[cfg(unix)] pub(super) fn crash_checkpoint(phase: &str) { if std::env::var("MEGA_MST2_NATIVE_CRASH_PHASE") diff --git a/tests/integration_api_write_trunk.rs b/tests/integration_api_write_trunk.rs index 569340ce..48e710d6 100644 --- a/tests/integration_api_write_trunk.rs +++ b/tests/integration_api_write_trunk.rs @@ -232,6 +232,63 @@ struct ServiceProcess { reaped: bool, } +#[cfg(target_os = "linux")] +fn process_owns_tcp_listener(pid: u32, port: u16) -> std::io::Result { + let proc = PathBuf::from(format!("/proc/{pid}")); + let sockets: std::collections::HashSet<_> = fs::read_dir(proc.join("fd"))? + .filter_map(Result::ok) + .filter_map(|entry| fs::read_link(entry.path()).ok()) + .filter_map(|target| { + target + .to_str()? + .strip_prefix("socket:[")? + .strip_suffix(']') + .map(str::to_owned) + }) + .collect(); + for table in ["tcp", "tcp6"] { + let rows = fs::read_to_string(proc.join("net").join(table))?; + for row in rows.lines().skip(1) { + let fields: Vec<_> = row.split_whitespace().collect(); + if fields.len() < 10 || fields[3] != "0A" || !sockets.contains(fields[9]) { + continue; + } + let Some((address, encoded_port)) = fields[1].split_once(':') else { + continue; + }; + if u16::from_str_radix(encoded_port, 16).ok() == Some(port) + && matches!( + address, + "00000000" | "0100007F" | "00000000000000000000000000000000" + ) + { + return Ok(true); + } + } + } + Ok(false) +} + +#[cfg(target_os = "linux")] +#[test] +fn service_readiness_accepts_its_own_kernel_listener() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + assert!(process_owns_tcp_listener(std::process::id(), port).unwrap()); +} + +#[cfg(target_os = "linux")] +#[test] +fn service_readiness_rejects_a_connectable_foreign_listener() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut command = Command::new("sh"); + command.args(["-c", "exec sleep 60"]); + let service = ServiceProcess::spawn(command); + assert!(TcpStream::connect(("127.0.0.1", port)).is_ok()); + assert!(!process_owns_tcp_listener(service.child.id(), port).unwrap()); +} + impl ServiceProcess { fn spawn(mut command: Command) -> Self { let child = command.spawn().expect("spawn mega2 service"); @@ -251,9 +308,6 @@ impl ServiceProcess { ) { 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!( @@ -262,6 +316,14 @@ impl ServiceProcess { read_log(stderr_path), ); } + // Another test's short-lived socket cannot establish this child's readiness. + #[cfg(target_os = "linux")] + let owns_listener = process_owns_tcp_listener(self.child.id(), port).unwrap_or(false); + #[cfg(not(target_os = "linux"))] + let owns_listener = true; + if owns_listener && TcpStream::connect(("127.0.0.1", port)).is_ok() { + return; + } if Instant::now() >= deadline { panic!( "service did not bind port {port} within {timeout:?}\nstdout:\n{}\nstderr:\n{}",