fix(mst2): gate projection writer on ready native head

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

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")

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<bool> {
+    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{}",