From 0574bee142ee6a77828b0b953acaf404843c8afb Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 12:24:33 +0800 Subject: [PATCH 01/10] fix(git): respect shallow depth and v0 negotiation boundaries --- src/ceres/pack/monorepo.rs | 164 +++++++++++++++++++++++++++++- src/ceres/protocol/smart.rs | 29 ++++-- src/contract/git_protocol/http.rs | 3 + 3 files changed, 182 insertions(+), 14 deletions(-) diff --git a/src/ceres/pack/monorepo.rs b/src/ceres/pack/monorepo.rs index 99db2992..8e5f79f7 100644 --- a/src/ceres/pack/monorepo.rs +++ b/src/ceres/pack/monorepo.rs @@ -282,17 +282,21 @@ impl RepoHandler for Monorepo { for level in 0..depth { let mut next_level: Vec = Vec::new(); for commit in ¤t_level { + // Depth counts the wanted commit. Cut its parent edges at + // the last included level, without packing the next level. + if level + 1 == depth { + if !commit.parent_commit_ids.is_empty() { + shallow_commits.push(commit.id.to_string()); + } + continue; + } for p_commit_id in &commit.parent_commit_ids { let p_id = p_commit_id.to_string(); if visited.insert(p_id.clone()) && let Some(model) = storage.get_commit_by_hash(&p_id).await.unwrap() { let parent = Commit::from_mega_model(model); - if level + 1 == depth { - shallow_commits.push(p_id); - } else { - next_level.push(parent.clone()); - } + next_level.push(parent.clone()); all_commits.push(parent); } } @@ -3716,6 +3720,156 @@ mod tests { ); } + #[tokio::test] + async fn shallow_pack_respects_depth_and_excludes_parent_only_objects() { + let temp = TempDir::new().expect("temp dir"); + let storage = test_storage(temp.path()).await; + let mono_storage = storage.mono_storage(); + let conn = mono_storage.get_connection(); + let sig = Signature::new( + SignatureType::Author, + "Shallow Test".into(), + "shallow@example.invalid".into(), + ); + let mut commits: Vec = Vec::new(); + let mut objects = Vec::new(); + for index in 0..3 { + let blob = Blob::from_content_bytes(format!("unique depth {index}").into_bytes()); + storage + .git_service + .put_objects(vec![blob.clone()]) + .await + .unwrap(); + let tree = Tree::from_tree_items(vec![TreeItem::new( + TreeItemMode::Blob, + blob.id, + format!("file-{index}.txt"), + )]) + .unwrap(); + mega_tree::Entity::insert( + tree.clone() + .into_mega_model(EntryMeta::default()) + .into_active_model(), + ) + .exec(conn) + .await + .unwrap(); + let parents = commits.last().map(|c| vec![c.id]).unwrap_or_default(); + let commit = Commit::new(sig.clone(), sig.clone(), tree.id, parents, "depth fixture"); + let model: mega_commit::Model = commit.clone().into_mega_model(EntryMeta::default()); + mega_commit::Entity::insert(model.into_active_model()) + .exec(conn) + .await + .unwrap(); + objects.push([ + blob.id.to_string(), + tree.id.to_string(), + commit.id.to_string(), + ]); + commits.push(commit); + } + let repo = + trunk_monorepo(&storage, "/", vec![], HashSet::new(), HashSet::new(), None).await; + for depth in [1, 2, 3, 5] { + let (stream, boundaries) = repo + .shallow_pack(vec![commits[2].id.to_string()], depth, false) + .await + .unwrap(); + let included = (depth as usize).min(commits.len()); + let expected_boundaries = if included < commits.len() { + vec![commits[commits.len() - included].id.to_string()] + } else { + vec![] + }; + assert_eq!( + boundaries, expected_boundaries, + "depth {depth}: cut at the included boundary commit" + ); + let pack: Vec = stream.concat().await; + let declared = u32::from_be_bytes(pack[8..12].try_into().unwrap()) as usize; + let mut decoder = Pack::new_with_hash_kind( + repo.object_hash_kind().unwrap(), + Some(1), + Some(64 * 1024 * 1024), + Some(temp.path().to_path_buf()), + true, + ); + let ids = Arc::new(Mutex::new(Vec::new())); + let sink = ids.clone(); + decoder + .decode( + &mut std::io::Cursor::new(pack), + move |entry| sink.lock().unwrap().push(entry.inner.hash.to_string()), + None::, + ) + .unwrap(); + let expected: HashSet<_> = objects[objects.len() - included..] + .iter() + .flatten() + .cloned() + .collect(); + let ids = ids.lock().unwrap(); + assert_eq!( + declared, + expected.len(), + "depth {depth}: exact pack header count" + ); + assert_eq!(ids.len(), expected.len(), "depth {depth}: no duplicates"); + assert_eq!( + ids.iter().cloned().collect::>(), + expected, + "depth {depth}: no parent-only objects" + ); + } + + let state = crate::ceres::api_service::state::ProtocolApiState { + storage: storage.clone(), + git_object_cache: repo.git_object_cache.clone(), + entity_store: storage.entity_store(), + }; + for depth in [1, 3] { + for done in [false, true] { + use crate::ceres::protocol::{ServiceType, SmartSession, TransportProtocol, smart}; + let mut session = SmartSession::from_state( + PathBuf::from("/"), + ServiceType::UploadPack, + TransportProtocol::Http, + &state, + ) + .unwrap(); + let mut request = bytes::BytesMut::new(); + smart::add_pkt_line_string(&mut request, format!("want {}\n", commits[2].id)); + smart::add_pkt_line_string(&mut request, format!("deepen {depth}\n")); + request.extend_from_slice(smart::PKT_LINE_END_MARKER); + if done { + smart::add_pkt_line_string(&mut request, "done\n".to_string()); + } + let (stream, response) = session + .git_upload_pack(&state, &mut request.freeze()) + .await + .unwrap(); + let mut expected = bytes::BytesMut::new(); + if depth == 1 { + smart::add_pkt_line_string( + &mut expected, + format!("shallow {}\n", commits[2].id), + ); + } + expected.extend_from_slice(smart::PKT_LINE_END_MARKER); + if done { + smart::add_pkt_line_string(&mut expected, "NAK\n".to_string()); + } + assert_eq!( + response, expected, + "depth {depth}, done {done}: v0 negotiation" + ); + let pack: Vec = stream.concat().await; + assert_eq!(pack.starts_with(b"PACK"), done); + assert_eq!(pack.is_empty(), !done); + } + } + } + /// Regression (fetch duplicate/count defect): two want commits sharing /// one root tree (net-zero roll-ups) must produce a pack where every /// object appears exactly once and the entry count matches the header. diff --git a/src/ceres/protocol/smart.rs b/src/ceres/protocol/smart.rs index afb4dad3..2f460795 100644 --- a/src/ceres/protocol/smart.rs +++ b/src/ceres/protocol/smart.rs @@ -6,7 +6,7 @@ use std::{ use anyhow::Result; use bytes::{Buf, BufMut, Bytes, BytesMut}; -use futures::{Stream, stream}; +use futures::{Stream, StreamExt, stream}; use git_internal::hash::HashKind; use tokio_stream::wrappers::ReceiverStream; @@ -126,7 +126,7 @@ impl SmartSession { let mut last_common_commit = String::new(); let mut deepen_depth: Option = None; let mut deepen_relative = false; - let mut shallow_commits: Vec = Vec::new(); + let mut done = false; let mut read_first_line = false; loop { @@ -162,7 +162,10 @@ impl SmartSession { let oid = self.parse_pkt_object_id(&dst, 5, "have")?; have.insert(oid); } - b"done" => break, + b"done" => { + done = true; + break; + } b"deep" => { let payload = &dst[4..]; if payload.starts_with(b"en ") { @@ -255,14 +258,26 @@ impl SmartSession { if have.is_empty() { if let Some(depth) = deepen_depth { - let (stream, shallows) = repo_handler + let (stream, shallow_commits) = repo_handler .shallow_pack(want, depth, deepen_relative) .await .map_err(|e| { ProtocolError::InvalidInput(format!("shallow pack generation failed: {e}")) })?; pack_data = stream; - shallow_commits = shallows; + for shallow in &shallow_commits { + add_pkt_line_string(&mut protocol_buf, format!("shallow {shallow}\n")); + } + // The shallow-info section ends before the ACK/NAK section, + // even when the requested depth reaches the root commit. + protocol_buf.extend_from_slice(PKT_LINE_END_MARKER); + // Stateless clients first request shallow boundaries, then + // repeat the wants with `done` to request the pack. + if !done { + pack_data.for_each(|_| async {}).await; + let (_tx, rx) = tokio::sync::mpsc::channel(1); + return Ok((ReceiverStream::new(rx), protocol_buf)); + } } else { pack_data = repo_handler.full_pack(want).await.map_err(|e| { ProtocolError::InvalidInput(format!("pack generation failed: {e}")) @@ -304,10 +319,6 @@ impl SmartSession { add_pkt_line_string(&mut protocol_buf, format!("ACK {last_common_commit} \n")); } - for shallow in &shallow_commits { - add_pkt_line_string(&mut protocol_buf, format!("shallow {shallow}\n")); - } - Ok((pack_data, protocol_buf)) } diff --git a/src/contract/git_protocol/http.rs b/src/contract/git_protocol/http.rs index 54fd20a9..28e041cc 100644 --- a/src/contract/git_protocol/http.rs +++ b/src/contract/git_protocol/http.rs @@ -250,6 +250,9 @@ pub async fn git_upload_pack( let body_stream = async_stream::stream! { tracing::info!("send ack/nak message buf: --------> {:?}", &protocol_buf); yield Ok::<_, Infallible>(Bytes::copy_from_slice(&protocol_buf)); + if protocol_buf.ends_with(smart::PKT_LINE_END_MARKER) { + return; + } while let Some(chunk) = send_pack_data.next().await { let mut reader = chunk.as_slice(); loop { From d7d630f3304914352ac633dff5b18b2b3b05ea30 Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 12:24:33 +0800 Subject: [PATCH 02/10] test(redis): accept OS-assigned standalone ports --- src/jupiter/redis/lock.rs | 2 +- src/jupiter/redis/snowflake_worker.rs | 2 +- src/jupiter/tests.rs | 12 ++++++++++++ 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/jupiter/redis/lock.rs b/src/jupiter/redis/lock.rs index e3792c91..5b8d4063 100644 --- a/src/jupiter/redis/lock.rs +++ b/src/jupiter/redis/lock.rs @@ -406,7 +406,7 @@ mod tests { eprintln!("redis-server not found; skipping redis lock tests"); return None; } - let server = RedisServer::new(); + let server = crate::jupiter::tests::test_redis_server(); let url = server.client_addr().to_owned(); println!("starting redis mock server at: {}", url); let client = redis::Client::open(url).unwrap(); diff --git a/src/jupiter/redis/snowflake_worker.rs b/src/jupiter/redis/snowflake_worker.rs index f45215d2..991e8415 100644 --- a/src/jupiter/redis/snowflake_worker.rs +++ b/src/jupiter/redis/snowflake_worker.rs @@ -126,7 +126,7 @@ mod tests { eprintln!("redis-server not found; skipping snowflake worker tests"); return None; } - let server = RedisServer::new(); + let server = crate::jupiter::tests::test_redis_server(); let url = server.client_addr().to_owned(); let client = redis::Client::open(url).unwrap(); let conn = ConnectionManager::new(client).await.unwrap(); diff --git a/src/jupiter/tests.rs b/src/jupiter/tests.rs index 78a33808..aab57e64 100644 --- a/src/jupiter/tests.rs +++ b/src/jupiter/tests.rs @@ -412,6 +412,18 @@ pub async fn with_test_vault(storage: Storage, dir: impl AsRef) -> Storage storage.with_vault(vault) } +/// Start a standalone Redis instance on an OS-assigned test port. +pub fn test_redis_server() -> redis_test::server::RedisServer { + use redis_test::server::RedisServer; + + // RedisServer::new reserves room for a cluster bus port and rejects + // ports >= 55535. These standalone servers need no cluster bus port. + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("free Redis test port"); + let port = listener.local_addr().expect("Redis test address").port(); + drop(listener); + RedisServer::new_with_addr_and_modules(RedisServer::get_addr(port), &[], false) +} + /// Redis manager for RedLock-backed server-signing key init. pub async fn test_redis_manager() -> crate::jupiter::redis::ConnectionManager { let url = From 8e1b2042785e7639e9fa6d80d4744d4148d0e103 Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 12:24:33 +0800 Subject: [PATCH 03/10] test(s3): allow isolated RustFS integration endpoints --- tests/common/mod.rs | 24 ++++++++++++++++++++++++ tests/integration_fastcdc_libra.rs | 12 +++++------- tests/integration_git_lfs.rs | 5 +++-- tests/integration_vault.rs | 4 ++-- 4 files changed, 34 insertions(+), 11 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 29c69717..cc9b194d 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -11,6 +11,30 @@ use std::{fs, path::Path}; +#[allow( + dead_code, + reason = "RustFS endpoint is used by S3 integration targets only" +)] +pub fn rustfs_endpoint() -> String { + std::env::var("MEGA2_IT_RUSTFS_URL") + .unwrap_or_else(|_| "http://127.0.0.1:19000".to_string()) + .trim_end_matches('/') + .to_string() +} + +#[allow( + dead_code, + reason = "RustFS probe is used by S3 integration targets only" +)] +pub fn rustfs_available(endpoint: &str) -> bool { + let url = url::Url::parse(endpoint).expect("valid RustFS test endpoint"); + std::net::TcpStream::connect(( + url.host_str().expect("RustFS test host"), + url.port_or_known_default().expect("RustFS test port"), + )) + .is_ok() +} + pub fn write_full_config(path: &Path) { // Product sample defaults to `admin = ["mega2"]` (local convenience), but // the git/SSH/LFS IT harness and seeded cedar fixtures still use the diff --git a/tests/integration_fastcdc_libra.rs b/tests/integration_fastcdc_libra.rs index 1b6c5855..e112fb12 100644 --- a/tests/integration_fastcdc_libra.rs +++ b/tests/integration_fastcdc_libra.rs @@ -173,20 +173,18 @@ impl FastcdcEnv { .env("MEGA_OBJECT_STORAGE__LOCAL__ROOT_DIR", &self.object_root); } StorageBackend::S3 => { + let endpoint = common::rustfs_endpoint(); // Fail closed if RustFS is missing — no skip (MF-07). assert!( - TcpStream::connect(("127.0.0.1", 19000)).is_ok(), - "media_publication_s3 requires RustFS at http://127.0.0.1:19000; \ + common::rustfs_available(&endpoint), + "media_publication_s3 requires RustFS at {endpoint}; \ run `docker compose -p mega2-it -f docker/docker-compose.test.yml up -d --wait postgres redis rustfs rustfs-init`" ); command .env("MEGA_OBJECT_STORAGE__STORAGE_TYPE", "s3compatible") .env("MEGA_OBJECT_STORAGE__S3__REGION", "us-east-1") .env("MEGA_OBJECT_STORAGE__S3__BUCKET", "mega2") - .env( - "MEGA_OBJECT_STORAGE__S3__ENDPOINT_URL", - "http://127.0.0.1:19000", - ) + .env("MEGA_OBJECT_STORAGE__S3__ENDPOINT_URL", endpoint) .env("MEGA_OBJECT_STORAGE__S3__ACCESS_KEY_ID", "rustfs") .env( "MEGA_OBJECT_STORAGE__S3__SECRET_ACCESS_KEY", @@ -1240,7 +1238,7 @@ async fn build_publication_store( s3: S3Config { region: "us-east-1".into(), bucket: "mega2".into(), - endpoint_url: "http://127.0.0.1:19000".into(), + endpoint_url: common::rustfs_endpoint(), public_endpoint_url: String::new(), access_key_id: "rustfs".into(), secret_access_key: "rustfs_secret".into(), diff --git a/tests/integration_git_lfs.rs b/tests/integration_git_lfs.rs index bae543e2..60a5034e 100644 --- a/tests/integration_git_lfs.rs +++ b/tests/integration_git_lfs.rs @@ -1561,13 +1561,14 @@ fn integration_git_lfs_storage_events_presigned_gap() { let payload = wh05_unique_payload(); let oid = hex::encode(sha2::Sha256::digest(&payload)); + let rustfs_endpoint = common::rustfs_endpoint(); let s3_env: [(&str, &str); 6] = [ ("MEGA_OBJECT_STORAGE__STORAGE_TYPE", "s3compatible"), ("MEGA_OBJECT_STORAGE__S3__REGION", "us-east-1"), ("MEGA_OBJECT_STORAGE__S3__BUCKET", "mega2"), ( "MEGA_OBJECT_STORAGE__S3__ENDPOINT_URL", - "http://127.0.0.1:19000", + rustfs_endpoint.as_str(), ), ("MEGA_OBJECT_STORAGE__S3__ACCESS_KEY_ID", "rustfs"), ( @@ -1633,7 +1634,7 @@ fn integration_git_lfs_storage_events_presigned_gap() { .expect("batch must carry an upload action for the fresh OID") .to_owned(); assert!( - upload_href.contains("127.0.0.1:19000"), + upload_href.starts_with(&format!("{rustfs_endpoint}/")), "upload action must be a presigned RustFS URL, got {upload_href}" ); let probe_status = probe.shutdown_via_sigint(Duration::from_secs(60)); diff --git a/tests/integration_vault.rs b/tests/integration_vault.rs index d8a77a6d..5bc5a742 100644 --- a/tests/integration_vault.rs +++ b/tests/integration_vault.rs @@ -927,8 +927,8 @@ fn integration_object_storage_s3_compatible_smoke() { // 启动真实 RustFS 服务后，通过 `debug storage-smoke` 对 S3-compatible 后端 // 执行 put/get/delete round-trip，验证对象存储后端在 post-vault 启动路径 // 正确解析 endpoint、bucket、credential 并完成真实 I/O。 - let rustfs_endpoint = "http://127.0.0.1:19000"; - let rustfs_available = std::net::TcpStream::connect("127.0.0.1:19000").is_ok(); + let rustfs_endpoint = common::rustfs_endpoint(); + let rustfs_available = common::rustfs_available(&rustfs_endpoint); if !rustfs_available { eprintln!( "integration_object_storage_s3_compatible_smoke requires RustFS at {}; \ From 24ce6f17b7bfcf236a067ea19dc2309027e773e6 Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 14:50:26 +0800 Subject: [PATCH 04/10] fix(git): preserve SSH shallow negotiation and isolate test ports --- src/ceres/protocol/smart.rs | 25 ++++- src/contract/git_protocol/ssh.rs | 55 ++++++++-- tests/common/git_cli.rs | 76 +++++++++++++- tests/integration_git_cli.rs | 10 +- tests/integration_git_lfs.rs | 8 +- tests/integration_git_ssh.rs | 175 ++++++++++++++++++++++++++++++- 6 files changed, 314 insertions(+), 35 deletions(-) diff --git a/src/ceres/protocol/smart.rs b/src/ceres/protocol/smart.rs index 2f460795..4587e2f6 100644 --- a/src/ceres/protocol/smart.rs +++ b/src/ceres/protocol/smart.rs @@ -118,6 +118,19 @@ impl SmartSession { &mut self, state: &ProtocolApiState, upload_request: &mut Bytes, + ) -> Result<(ReceiverStream>, BytesMut), ProtocolError> { + self.git_upload_pack_with_shallow_info(state, upload_request, false) + .await + } + + /// Stateful SSH clients send their wants once, read shallow-info, then + /// send `done`. The channel replays the wants for pack generation, but + /// must not send the already consumed shallow-info section again. + pub(crate) async fn git_upload_pack_with_shallow_info( + &mut self, + state: &ProtocolApiState, + upload_request: &mut Bytes, + shallow_info_sent: bool, ) -> Result<(ReceiverStream>, BytesMut), ProtocolError> { let repo_handler = self.repo_handler_with_commands(state, Vec::new()).await?; @@ -265,12 +278,14 @@ impl SmartSession { ProtocolError::InvalidInput(format!("shallow pack generation failed: {e}")) })?; pack_data = stream; - for shallow in &shallow_commits { - add_pkt_line_string(&mut protocol_buf, format!("shallow {shallow}\n")); + if !shallow_info_sent { + for shallow in &shallow_commits { + add_pkt_line_string(&mut protocol_buf, format!("shallow {shallow}\n")); + } + // The shallow-info section ends before the ACK/NAK section, + // even when the requested depth reaches the root commit. + protocol_buf.extend_from_slice(PKT_LINE_END_MARKER); } - // The shallow-info section ends before the ACK/NAK section, - // even when the requested depth reaches the root commit. - protocol_buf.extend_from_slice(PKT_LINE_END_MARKER); // Stateless clients first request shallow boundaries, then // repeat the wants with `done` to request the pack. if !done { diff --git a/src/contract/git_protocol/ssh.rs b/src/contract/git_protocol/ssh.rs index 2ff31809..f793c992 100644 --- a/src/contract/git_protocol/ssh.rs +++ b/src/contract/git_protocol/ssh.rs @@ -58,6 +58,8 @@ pub struct SshServer { pub struct GitSshChannelState { pub smart_protocol: SmartSession, pub data_combined: BytesMut, + /// Wants/depth from a v0 shallow-info round; SSH sends only `done` next. + pub pending_shallow_request: Option, } impl server::Server for SshServer { @@ -185,6 +187,7 @@ impl server::Handler for SshServer { GitSshChannelState { smart_protocol, data_combined: BytesMut::new(), + pending_shallow_request: None, }, ); session.data(channel, v2_adv.to_vec())?; @@ -196,6 +199,7 @@ impl server::Handler for SshServer { GitSshChannelState { smart_protocol, data_combined: BytesMut::new(), + pending_shallow_request: None, }, ); session.data(channel, res.to_vec())?; @@ -363,13 +367,16 @@ impl server::Handler for SshServer { match service_type { ServiceType::UploadPack => { // Git may deliver the upload-pack request across multiple SSH - // data packets. Process only once the buffer ends in a flush - // pkt-line (`0000`), matching HTTP's full-body collection. + // data packets. Process only a complete round, terminated + // by a flush pkt-line (`0000`) or a complete `done` packet. // Handling a partial/flush-only chunk as a complete request // previously ran pack generation with want=[] and returned // `error: …` (`bad line length character: erro` on the client). state.data_combined.extend_from_slice(data); - while upload_pack_buffer_complete(&state.data_combined) { + while upload_pack_buffer_complete( + &state.data_combined, + state.pending_shallow_request.is_some(), + ) { let request = take_complete_upload_pack_request(&mut state.data_combined); handle_upload_pack(state, &self.state, channel, &request, session).await; } @@ -575,8 +582,19 @@ async fn handle_upload_pack( return; } + let previous = state.pending_shallow_request.take(); + let shallow_info_sent = previous.is_some(); + let mut request = BytesMut::new(); + if let Some(previous) = previous { + request.extend_from_slice(&previous); + } + request.extend_from_slice(data); + let request = request.freeze(); + body = request.clone(); let smart_protocol = &mut state.smart_protocol; - let (mut send_pack_data, buf) = match smart_protocol.git_upload_pack(api_state, &mut body).await + let (mut send_pack_data, buf) = match smart_protocol + .git_upload_pack_with_shallow_info(api_state, &mut body, shallow_info_sent) + .await { Ok(result) => result, Err(e) => { @@ -588,6 +606,10 @@ async fn handle_upload_pack( tracing::info!("buf is {:?}", buf); let _ = session.data(channel, buf.to_vec()); + if buf.ends_with(smart::PKT_LINE_END_MARKER) { + state.pending_shallow_request = Some(request); + return; + } while let Some(chunk) = send_pack_data.next().await { let mut reader = chunk.as_slice(); @@ -613,15 +635,19 @@ async fn handle_upload_pack( /// True when `buf` holds at least one terminated upload-pack/v2 command. /// -/// A flush-only (`0000`) or empty buffer is *not* complete — processing those -/// as a full request yields `want=[]` and a protocol error to the client. -fn upload_pack_buffer_complete(buf: &[u8]) -> bool { +/// A flush-only (`0000`) or empty buffer is *not* complete. A done-only +/// continuation is complete when the channel already holds the wants/depth. +fn upload_pack_buffer_complete(buf: &[u8], shallow_negotiation_pending: bool) -> bool { if buf.len() < 8 { return false; } let text = String::from_utf8_lossy(buf); - let has_payload = text.contains("want ") || text.contains("command=") || text.contains("have "); - let terminated = buf.ends_with(smart::PKT_LINE_END_MARKER) || text.contains("0009done"); + let has_done = text.contains("0009done\n"); + let has_payload = text.contains("want ") + || text.contains("command=") + || text.contains("have ") + || (shallow_negotiation_pending && has_done); + let terminated = buf.ends_with(smart::PKT_LINE_END_MARKER) || has_done; has_payload && terminated } @@ -1078,6 +1104,7 @@ mod tests { TransportProtocol::Ssh, ), data_combined: BytesMut::new(), + pending_shallow_request: None, }; let mut channel_b = GitSshChannelState { smart_protocol: SmartSession::new( @@ -1086,6 +1113,7 @@ mod tests { TransportProtocol::Ssh, ), data_combined: BytesMut::new(), + pending_shallow_request: None, }; channel_a.data_combined.extend_from_slice(b"payload-a"); @@ -1094,4 +1122,13 @@ mod tests { assert_eq!(&channel_a.data_combined[..], b"payload-a"); assert_eq!(&channel_b.data_combined[..], b"payload-b"); } + + #[test] + fn upload_pack_accepts_done_only_after_shallow_info() { + assert!(!upload_pack_buffer_complete(b"0009done\n", false)); + assert!(upload_pack_buffer_complete(b"0009done\n", true)); + assert!(!upload_pack_buffer_complete(b"0009don", true)); + assert!(!upload_pack_buffer_complete(b"0009done", true)); + assert!(!upload_pack_buffer_complete(b"0000", true)); + } } diff --git a/tests/common/git_cli.rs b/tests/common/git_cli.rs index c37be6ca..9f5078a6 100644 --- a/tests/common/git_cli.rs +++ b/tests/common/git_cli.rs @@ -5,12 +5,13 @@ // so vault and other bin test targets do not compile these symbols. use std::{ + collections::HashSet, env, fs, io::Read, net::{TcpListener, TcpStream}, path::{Path, PathBuf}, process::{Command, Output, Stdio}, - sync::OnceLock, + sync::{Mutex, OnceLock}, thread::{self, sleep}, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; @@ -113,19 +114,84 @@ pub fn record_service_pid(pid: u32) { append_evidence_line("MEGA2_IT_PIDS_FILE", &pid.to_string()); } -/// Bind `127.0.0.1:0`, capture the OS-assigned port, then drop the listener. +/// Allocate a port that no other service in this test process has claimed. #[allow( dead_code, - reason = "SSH lifecycle helper; path-included into HTTP targets that do not call it yet" + reason = "HTTP/SSH lifecycle helper; some path-included targets do not call it" )] pub fn reserve_ephemeral_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind ephemeral port") + bind_ephemeral_listener() .local_addr() .expect("local addr") .port() } +pub fn bind_ephemeral_listener() -> TcpListener { + static CLAIMED_PORTS: OnceLock>> = OnceLock::new(); + bind_unique_listener(CLAIMED_PORTS.get_or_init(Default::default), || { + TcpListener::bind("0.0.0.0:0") + }) +} + +fn bind_unique_listener( + claimed: &Mutex>, + mut bind: impl FnMut() -> std::io::Result, +) -> TcpListener { + for _ in 0..1024 { + let listener = bind().expect("bind ephemeral port"); + let port = listener.local_addr().expect("local addr").port(); + // Keep the listener until the claim is recorded, and never recycle + // a port during this target's lifetime: services bind after DB setup. + if claimed.lock().expect("port allocator lock").insert(port) { + return listener; + } + } + panic!("could not allocate a new service port after 1024 attempts") +} + +#[test] +fn service_port_allocator_rejects_an_os_reused_port() { + let listener = TcpListener::bind("127.0.0.1:0").expect("test listener"); + let first = listener.local_addr().unwrap().port(); + let claimed = Mutex::new(HashSet::from([first])); + let mut reused = Some(listener); + let mut attempts = 0; + let second = bind_unique_listener(&claimed, || { + attempts += 1; + if attempts == 1 { + Ok(reused.take().unwrap()) + } else { + TcpListener::bind("127.0.0.1:0") + } + }) + .local_addr() + .unwrap() + .port(); + assert_ne!(first, second); + assert!(attempts >= 2); +} + +#[test] +fn concurrent_service_port_claims_are_unique() { + let ports = thread::scope(|scope| { + let workers: Vec<_> = (0..8) + .map(|_| { + scope.spawn(|| { + (0..16) + .map(|_| reserve_ephemeral_port()) + .collect::>() + }) + }) + .collect(); + workers + .into_iter() + .flat_map(|worker| worker.join().expect("port worker")) + .collect::>() + }); + assert_eq!(ports.len(), 128); + assert_eq!(ports.iter().copied().collect::>().len(), 128); +} + #[allow( dead_code, reason = "SSH lifecycle helper; path-included into HTTP targets that do not call it yet" diff --git a/tests/integration_git_cli.rs b/tests/integration_git_cli.rs index 262ee5c3..4faa926a 100644 --- a/tests/integration_git_cli.rs +++ b/tests/integration_git_cli.rs @@ -20,7 +20,7 @@ use std::{ collections::BTreeMap, fs, io::{Read, Write}, - net::{TcpListener, TcpStream}, + net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -367,7 +367,7 @@ fn spawn_website_session_stub(username: &str) -> u16 { /// subjects. UN-19 needs that: the same CL has to be offered to an admin and to /// a non-admin, and only one of them may merge it. fn spawn_website_session_stub_for(sessions: &[(&str, &str)]) -> u16 { - let listener = TcpListener::bind("127.0.0.1:0").expect("bind session stub"); + let listener = git_cli::bind_ephemeral_listener(); let port = listener.local_addr().expect("stub addr").port(); let sessions: Vec<(String, String)> = sessions .iter() @@ -5092,11 +5092,7 @@ fn with_runtime(future: impl std::future::Future) -> T { } fn reserve_free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind ephemeral port") - .local_addr() - .expect("local addr") - .port() + git_cli::reserve_ephemeral_port() } fn create_log_file(path: &Path) -> fs::File { diff --git a/tests/integration_git_lfs.rs b/tests/integration_git_lfs.rs index 60a5034e..5fda1ddc 100644 --- a/tests/integration_git_lfs.rs +++ b/tests/integration_git_lfs.rs @@ -15,7 +15,7 @@ mod git_cli; use std::{ fs, io::{Read, Write}, - net::{TcpListener, TcpStream}, + net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -1219,11 +1219,7 @@ fn with_runtime(future: impl std::future::Future) -> T { } fn reserve_free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind ephemeral port") - .local_addr() - .expect("local addr") - .port() + git_cli::reserve_ephemeral_port() } fn create_log_file(path: &Path) -> fs::File { diff --git a/tests/integration_git_ssh.rs b/tests/integration_git_ssh.rs index 9c5204c1..27c66192 100644 --- a/tests/integration_git_ssh.rs +++ b/tests/integration_git_ssh.rs @@ -14,10 +14,10 @@ mod common; mod git_cli; use std::{ - collections::BTreeMap, + collections::{BTreeMap, HashSet}, fs, io::Read, - net::{TcpListener, TcpStream}, + net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -452,6 +452,175 @@ fn integration_git_ssh_authenticated_clone() { drop(env); } +#[test] +fn integration_git_ssh_v0_shallow_fetch_resumes_after_shallow_info() { + assert!( + !git_cli::git_cli_skip_requested(), + "MEGA2_IT_SKIP_GIT_CLI must be unset/0 for integration_git_ssh; skipping is not a green path" + ); + git_cli::require_git_cli_runner(); + let env = GitSshEnv::new(); + let (mut service, port, _stdout_path, stderr_path, git_ssh, remote) = + prepare_authenticated_ssh(&env); + let git_stdout = |args: &[&str]| { + let output = git_cli::git_cli_ssh(&env.case_dir, &git_ssh, args); + git_cli::assert_git_success(&output, "SSH shallow fetch fixture/verification"); + String::from_utf8(output.stdout).expect("Git output UTF-8") + }; + + // Publish one child of the initial monorepo commit as a CL. Removing a + // parent-only file makes an over-deep pack observable in the object set. + let seed_name = "ssh-shallow-seed"; + git_stdout(&["clone", &remote, seed_name]); + git_stdout(&["-C", seed_name, "config", "user.name", "SSH Shallow Test"]); + git_stdout(&[ + "-C", + seed_name, + "config", + "user.email", + "ssh-shallow@example.invalid", + ]); + fs::remove_file(env.case_dir.join(seed_name).join("project/.gitkeep")) + .expect("remove parent-only fixture file"); + fs::write( + env.case_dir.join(seed_name).join("ssh-shallow-tip.txt"), + b"only the shallow tip contains this file\n", + ) + .expect("write shallow tip fixture"); + git_stdout(&["-C", seed_name, "add", "-A"]); + git_stdout(&["-C", seed_name, "commit", "-m", "SSH shallow tip"]); + let before = ls_remote_cl_refs_ssh(&env.case_dir, &git_ssh, &remote); + git_stdout(&[ + "-C", + seed_name, + "-c", + "pack.window=0", + "-c", + "pack.depth=0", + "push", + "origin", + "HEAD:refs/heads/ssh-shallow-test", + ]); + let cl_ref = ls_remote_cl_refs_ssh(&env.case_dir, &git_ssh, &remote) + .into_iter() + .find(|reference| !before.contains(reference)) + .expect("new SSH shallow CL ref"); + + // Freeze the actual server history as the oracle; server CL publication + // may assign commit identities independently of the client's seed. + let reference = "ssh-shallow-reference"; + git_stdout(&["init", reference]); + git_stdout(&["-C", reference, "fetch", &remote, &cl_ref]); + git_stdout(&["-C", reference, "checkout", "--detach", "FETCH_HEAD"]); + let history: Vec = git_stdout(&["-C", reference, "rev-list", "HEAD"]) + .lines() + .map(str::to_owned) + .collect(); + assert_eq!(history.len(), 2, "fixture must have a tip and its parent"); + let expected_worktree = snapshot_workdir(&env.case_dir.join(reference)); + assert_eq!( + expected_worktree, + snapshot_workdir(&env.case_dir.join(seed_name)), + "server must preserve the fixture tree" + ); + + for depth in [1, 2, 5] { + let repo = format!("ssh-shallow-depth-{depth}"); + let depth_arg = depth.to_string(); + git_stdout(&["init", &repo]); + // v0 SSH sends want/deepen/flush, consumes shallow-info, and sends + // only `done` next. Fresh repositories exercise the clone handshake. + git_stdout(&[ + "-C", + &repo, + "-c", + "protocol.version=0", + "fetch", + "--depth", + &depth_arg, + &remote, + &cl_ref, + ]); + git_stdout(&["-C", &repo, "checkout", "--detach", "FETCH_HEAD"]); + let included = (depth as usize).min(history.len()); + let actual_history: Vec = git_stdout(&["-C", &repo, "rev-list", "HEAD"]) + .lines() + .map(str::to_owned) + .collect(); + assert_eq!( + actual_history, + history[..included], + "depth {depth}: commits" + ); + let shallow_path = env.case_dir.join(&repo).join(".git/shallow"); + let boundaries: Vec = fs::read_to_string(shallow_path) + .unwrap_or_default() + .lines() + .map(str::to_owned) + .collect(); + if included < history.len() { + assert_eq!( + boundaries, + vec![history[included - 1].clone()], + "depth {depth}: boundary" + ); + } else { + // Git may retain a shallow marker at the root even when every + // commit fits within the requested depth. It cuts no real edge. + assert!( + boundaries.is_empty() || boundaries == vec![history.last().unwrap().clone()], + "depth {depth}: only a root marker is valid for complete history: {boundaries:?}" + ); + } + + let mut expected_objects = HashSet::new(); + for commit in &history[..included] { + expected_objects.insert(commit.clone()); + let tree = git_stdout(&["-C", reference, "rev-parse", &format!("{commit}^{{tree}}")]) + .trim() + .to_owned(); + expected_objects.insert(tree.clone()); + let entries = git_stdout(&["-C", reference, "ls-tree", "-r", "-t", &tree]); + for entry in entries.lines() { + expected_objects.insert( + entry + .split_whitespace() + .nth(2) + .expect("tree object id") + .to_owned(), + ); + } + } + let actual_objects: HashSet = git_stdout(&[ + "-C", + &repo, + "cat-file", + "--batch-all-objects", + "--batch-check=%(objectname)", + ]) + .lines() + .map(str::to_owned) + .collect(); + assert_eq!( + actual_objects, expected_objects, + "depth {depth}: exact objects" + ); + assert_eq!( + snapshot_workdir(&env.case_dir.join(&repo)), + expected_worktree, + "depth {depth}: worktree" + ); + git_stdout(&["-C", &repo, "fsck", "--full", "--strict"]); + } + let status = service.shutdown_via_sigint(Duration::from_secs(10)); + assert!( + status.success(), + "SSH service shutdown failed: {status}\n{}", + read_log(&stderr_path) + ); + git_cli::wait_until_port_closed(port, Duration::from_secs(5)); +} + fn git_ssh_command_loopback_password(case_dir: &Path, port: u16) -> String { let known = case_dir.join("ssh").join("known_hosts"); // BatchMode must stay off: OpenSSH disables password/ASKPASS when BatchMode=yes. @@ -2434,7 +2603,7 @@ const SESSION_COOKIE: &str = "better-auth.session_token=it-un16-ssh-session"; /// this stub answers with the admin whose ACL change is being merged. Returns /// the port it listens on. fn spawn_website_session_stub(username: &str) -> u16 { - let listener = TcpListener::bind("127.0.0.1:0").expect("bind session stub"); + let listener = git_cli::bind_ephemeral_listener(); let port = listener.local_addr().expect("stub addr").port(); let body = format!( r#"{{"session":{{"id":"it-session","userId":"{username}"}},"user":{{"id":"{username}","name":"{username}","email":"{username}@example.invalid"}}}}"# From 689e7eb1a83f74d6ff35b3b57cebce8a2738ea48 Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 18:05:41 +0800 Subject: [PATCH 05/10] Validate MST2 verified facts and preserve raw blob content --- src/callisto/mst2_verified_object.rs | 6 +- src/ceres/snapshot/error.rs | 6 + src/ceres/snapshot/pages.rs | 610 ++++++++++++++++++++++++--- src/ceres/snapshot/view.rs | 19 +- src/jupiter/storage/mono_storage.rs | 252 ++++++++++- tests/integration_api_write_trunk.rs | 11 +- 6 files changed, 828 insertions(+), 76 deletions(-) diff --git a/src/callisto/mst2_verified_object.rs b/src/callisto/mst2_verified_object.rs index 851f5cd5..9933e936 100644 --- a/src/callisto/mst2_verified_object.rs +++ b/src/callisto/mst2_verified_object.rs @@ -2,9 +2,9 @@ //! //! One row per verified object: Git identity + raw content digest + 64-bit //! size + verification state. The natural key is -//! `(storage_domain, git_oid, object_kind)`; rows are insert-only with -//! ON CONFLICT DO NOTHING — a row only ever appears after the content was -//! fetched and hashed by this service (write-through verification). +//! `(storage_domain, git_oid, object_kind)`. Current-generation facts retain +//! first-write-wins semantics; valid legacy facts can be atomically upgraded +//! only after this service fetches and hashes the raw body again. use sea_orm::entity::prelude::*; use serde::{Deserialize, Serialize}; diff --git a/src/ceres/snapshot/error.rs b/src/ceres/snapshot/error.rs index 8d9cb316..4fe54274 100644 --- a/src/ceres/snapshot/error.rs +++ b/src/ceres/snapshot/error.rs @@ -34,6 +34,8 @@ pub enum SnapshotErrorCode { DigestMismatch, RangeNotSupported, SymlinkTraversal, + ObjectUnavailable, + IntegrityError, Internal, } @@ -60,6 +62,8 @@ impl SnapshotErrorCode { SnapshotErrorCode::DigestMismatch => "EXPECTED_DIGEST_MISMATCH", SnapshotErrorCode::RangeNotSupported => "RANGE_NOT_SUPPORTED", SnapshotErrorCode::SymlinkTraversal => "SYMLINK_TRAVERSAL", + SnapshotErrorCode::ObjectUnavailable => "OBJECT_UNAVAILABLE", + SnapshotErrorCode::IntegrityError => "INTEGRITY_ERROR", SnapshotErrorCode::Internal => "INTERNAL", } } @@ -79,6 +83,7 @@ impl SnapshotErrorCode { | SnapshotErrorCode::PathNotFound | SnapshotErrorCode::LeaseUnknown => 404, SnapshotErrorCode::SnapshotNotReady => 503, + SnapshotErrorCode::ObjectUnavailable => 503, SnapshotErrorCode::NotDirectory | SnapshotErrorCode::Conflict | SnapshotErrorCode::CursorStale @@ -87,6 +92,7 @@ impl SnapshotErrorCode { SnapshotErrorCode::UnsupportedEntry => 422, SnapshotErrorCode::LimitExceeded | SnapshotErrorCode::ProofBudgetExceeded => 413, SnapshotErrorCode::LeaseExpired | SnapshotErrorCode::SnapshotGone => 410, + SnapshotErrorCode::IntegrityError => 502, SnapshotErrorCode::Internal => 500, } } diff --git a/src/ceres/snapshot/pages.rs b/src/ceres/snapshot/pages.rs index 74824a38..555ed913 100644 --- a/src/ceres/snapshot/pages.rs +++ b/src/ceres/snapshot/pages.rs @@ -15,13 +15,16 @@ use mst2_codec::metapage::{Entry, EntryKind, Page, page_id}; use sea_orm::ActiveValue::Set; use sha2::{Digest, Sha256}; -use crate::ceres::{ - api_service::ApiHandler, - snapshot::{ - error::{SnapshotError, SnapshotErrorCode}, - resolver::FsKind, - view::hex, +use crate::{ + ceres::{ + api_service::ApiHandler, + snapshot::{ + error::{SnapshotError, SnapshotErrorCode}, + resolver::FsKind, + view::hex, + }, }, + jupiter::storage::mono_storage::MST2_VERIFICATION_VERSION, }; /// A directory page plus everything the JSON layer needs. @@ -99,37 +102,43 @@ async fn build_directory_page_uncached( let dirents = crate::ceres::snapshot::resolver::direct_entries(&tree)?; // T03 write-through verification: consult verified records first, then - // fetch + hash the misses and persist them. A verification-record read - // failure degrades to recomputation (safe: strictly more verification). + // fetch + hash the misses and persist them. Invalid records and lookup + // failures are errors, never a missing verification fact. let blob_oids: Vec = dirents .iter() .filter(|(_, k, _)| *k != FsKind::Directory) .map(|(_, _, oid)| oid.clone()) .collect(); - let verified = match handler + let verified = handler .get_context() .mono_storage() .get_verified_blobs(blob_oids) .await - { - Ok(map) => map, - Err(e) => { - tracing::warn!(error = %e, "verified blob lookup failed; recomputing digests"); - HashMap::new() - } - }; + .map_err(|e| { + let code = if matches!( + e, + crate::common::errors::MegaError::ObjStorageInconsistent(_) + ) { + SnapshotErrorCode::IntegrityError + } else { + SnapshotErrorCode::Internal + }; + tracing::warn!(error = %e, "verified blob lookup failed"); + SnapshotError::new(code, "verified blob lookup failed") + })?; - let mut new_verified: Vec = Vec::new(); + let mut new_verified = HashMap::new(); let mut entries = Vec::with_capacity(dirents.len()); let mut codec_entries = Vec::with_capacity(dirents.len()); for (name, fs_kind, oid) in dirents { + let child_rel = if rel_path == "/" { + format!("/{name}") + } else { + format!("{rel_path}/{name}") + }; + crate::ceres::snapshot::view::validate_scope_relative_path(&child_rel)?; match fs_kind { FsKind::Directory => { - let child_rel = if rel_path == "/" { - format!("/{name}") - } else { - format!("{rel_path}/{name}") - }; let child_page = Box::pin(build_directory_page(handler, root_tree, &child_rel)).await?; codec_entries.push(Entry::dir(name.as_bytes(), child_page.page_id)); @@ -145,24 +154,37 @@ async fn build_directory_page_uncached( FsKind::Regular | FsKind::Executable | FsKind::Symlink => { let (size, digest) = if let Some(v) = verified.get(&oid) { // Verified record: 64-bit size + raw digest, no content read. - let d: [u8; 32] = v.raw_sha256.clone().try_into().unwrap_or([0u8; 32]); - (v.size as u64, d) + let d: [u8; 32] = v.raw_sha256.as_slice().try_into().map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "invalid verified content digest", + ) + })?; + let size = u64::try_from(v.size).map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "invalid verified content size", + ) + })?; + (size, d) } else { let raw = fetch_raw_blob(handler, &oid).await?; let mut h = Sha256::new(); h.update(&raw); let digest: [u8; 32] = h.finalize().into(); - new_verified.push(crate::callisto::mst2_verified_object::ActiveModel { - id: sea_orm::ActiveValue::NotSet, - storage_domain: Set("git".to_string()), - git_oid: Set(oid.clone()), - object_kind: Set("blob".to_string()), - raw_sha256: Set(digest.to_vec()), - size: Set(raw.len() as i64), - verification_version: Set(1), - state: Set("VERIFIED".to_string()), - created_at: Set(chrono_now()), - }); + new_verified.entry(oid.clone()).or_insert( + crate::callisto::mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(digest.to_vec()), + size: Set(raw.len() as i64), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono_now()), + }, + ); (raw.len() as u64, digest) }; let kind = match fs_kind { @@ -189,7 +211,7 @@ async fn build_directory_page_uncached( if let Err(e) = handler .get_context() .mono_storage() - .insert_verified_blobs(new_verified) + .insert_verified_blobs(new_verified.into_values().collect()) .await { tracing::warn!(error = %e, "verified blob persistence failed"); @@ -242,6 +264,7 @@ async fn fetch_tree( root_tree: &git_internal::internal::object::tree::Tree, rel_path: &str, ) -> Result { + crate::ceres::snapshot::view::validate_scope_relative_path(rel_path)?; if rel_path == "/" { return Ok(root_tree.clone()); } @@ -329,6 +352,7 @@ pub async fn resolve_abs( root_tree: &git_internal::internal::object::tree::Tree, abs_path: &str, ) -> Result { + crate::ceres::snapshot::view::validate_scope_relative_path(abs_path)?; if abs_path == "/" { return Ok(WalkOutcome::FoundDir); } @@ -379,7 +403,7 @@ pub async fn resolve_abs( unreachable!("loop returns on the last component") } -/// Raw blob bytes with the `blob \0` git header stripped. +/// GitService stores and returns raw file content, without an object header. pub async fn fetch_raw_blob( handler: &T, oid: &str, @@ -387,24 +411,19 @@ pub async fn fetch_raw_blob( let data = handler.get_raw_blob_by_hash(oid).await.map_err(|e| { // Missing/corrupt content must be an error, never an empty file // (spec 00 SYS-04). - SnapshotError::new( - SnapshotErrorCode::Internal, - format!("blob fetch failed for {oid}: {e}"), - ) + let code = match e { + crate::common::errors::MegaError::ObjStorageNotFound(_) => { + SnapshotErrorCode::ObjectUnavailable + } + crate::common::errors::MegaError::ObjStorageInconsistent(_) => { + SnapshotErrorCode::IntegrityError + } + _ => SnapshotErrorCode::Internal, + }; + tracing::warn!(error = %e, "fixed-view blob fetch failed"); + SnapshotError::new(code, "fixed-view content could not be read") })?; - Ok(strip_git_blob_header(&data)) -} - -fn strip_git_blob_header(data: &[u8]) -> Vec { - if let Some(pos) = data.iter().position(|&b| b == 0) - && data.starts_with(b"blob ") - && let Ok(len) = std::str::from_utf8(&data[5..pos]).map(|s| s.parse::()) - && let Ok(len) = len - && data.len() == pos + 1 + len - { - return data[pos + 1..].to_vec(); - } - data.to_vec() + Ok(data) } pub fn sha256_hex(data: &[u8]) -> String { @@ -426,3 +445,486 @@ fn chrono_now() -> chrono::DateTime { pub fn hex_of(id: &[u8; 32]) -> String { hex(id) } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use axum::{body::Body, http::Request}; + use bytes::Bytes; + use git_internal::internal::{ + metadata::EntryMeta, + object::{ + blob::Blob, + tree::{Tree, TreeItem, TreeItemMode}, + }, + }; + use sea_orm::{ActiveValue::Set, EntityTrait, IntoActiveModel}; + use tower::ServiceExt; + + use super::*; + use crate::{ + api::{ + MonoApiServiceState, oauth::api_store::BrowserSessionStore, router::snapshot_router, + }, + callisto::{mega_refs, mst2_verified_object}, + ceres::api_service::cache::GitObjectCache, + config::testing::isolated_config, + jupiter::{ + storage::base_storage::StorageConnector, + tests::{test_redis_manager, test_storage_with_config}, + utils::converter::IntoMegaModel, + }, + }; + + async fn state(temp: &std::path::Path) -> MonoApiServiceState { + let mut config = isolated_config(temp.join("config")); + config.mst2.enabled = true; + config.mst2.instance_uuid = Some(uuid::Uuid::new_v4().to_string()); + config.mst2.auth_token = Some("mst2-projection-test".to_string()); + let storage = test_storage_with_config(temp, config).await; + MonoApiServiceState { + entity_store: storage.entity_store.clone(), + storage, + session_store: BrowserSessionStore::Anonymous, + git_object_cache: Arc::new(GitObjectCache { + connection: test_redis_manager().await, + prefix: String::new(), + }), + listen_addr: "127.0.0.1:0".to_string(), + } + } + + async fn seed(state: &MonoApiServiceState, filename: &str, raw: &[u8]) -> String { + seed_path(state, &format!("/{filename}"), raw).await + } + + async fn seed_path(state: &MonoApiServiceState, path: &str, raw: &[u8]) -> String { + let blob = Blob::from_content_bytes(raw.to_vec()); + state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(raw)) + .await + .unwrap(); + let mut child_id = blob.id; + let mut child_mode = TreeItemMode::Blob; + let mut trees = Vec::new(); + for name in path[1..].split('/').rev() { + let tree = + Tree::from_tree_items(vec![TreeItem::new(child_mode, child_id, name.to_string())]) + .unwrap(); + child_id = tree.id; + child_mode = TreeItemMode::Tree; + trees.push(tree.into_mega_model(EntryMeta::new()).into_active_model()); + } + let mono = state.storage.mono_storage(); + mono.batch_save_model(trees).await.unwrap(); + mono.save_refs( + mega_refs::Model::new( + "/", + "refs/heads/main".to_string(), + blob.id.to_string(), + child_id.to_string(), + false, + ), + None, + ) + .await + .unwrap(); + blob.id.to_string() + } + + async fn seed_legacy(state: &MonoApiServiceState, oid: &str, raw: &[u8]) { + mst2_verified_object::Entity::insert(mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.to_string()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(Sha256::digest(raw).to_vec()), + size: Set(raw.len() as i64), + verification_version: Set(1), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }) + .exec(state.storage.mono_storage().get_connection()) + .await + .unwrap(); + } + + async fn snapshot_id(state: &MonoApiServiceState, scope: &str) -> String { + let main = state + .storage + .mono_storage() + .get_main_ref("/") + .await + .unwrap() + .unwrap(); + let handler = crate::ceres::api_service::mono_api_service::MonoApiService::from(state); + let tree = handler.get_tree_by_hash(&main.ref_tree_hash).await.unwrap(); + let built = build_directory_page(&handler, &tree, scope).await.unwrap(); + crate::ceres::snapshot::descriptor::build( + &state.storage.config().mst2, + &crate::ceres::snapshot::view::SnapshotView::from_commit( + &main.ref_commit_hash, + &main.ref_tree_hash, + ), + scope, + built.page_id, + ) + .unwrap() + .snapshot_id + } + + fn request(method: &str, uri: &str, lease: Option<&str>, body: Body) -> Request { + let mut request = Request::builder() + .method(method) + .uri(uri) + .header("authorization", "Bearer mst2-projection-test") + .header("content-type", "application/json"); + if let Some(lease) = lease { + request = request.header("x-mega-snapshot-lease", lease); + } + request.body(body).unwrap() + } + + #[tokio::test] + async fn mst2_raw_header_like_bytes_survive_projection_and_http_reads() { + assert_header_like_blob_http(false).await; + } + + #[tokio::test] + async fn mst2_legacy_header_like_fact_is_reverified_before_http_projection() { + assert_header_like_blob_http(true).await; + } + + async fn assert_header_like_blob_http(legacy: bool) { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let raw = b"blob 3\0abc"; + let filename = format!("binary-{}", uuid::Uuid::new_v4()); + let oid = seed(&state, &filename, raw).await; + if legacy { + seed_legacy(&state, &oid, b"abc").await; + } + let app = snapshot_router::routers(state.clone()).with_state(state.clone()); + let resolve = app + .clone() + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(resolve.status(), 200); + let resolved: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(resolve.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + let snapshot = snapshot_id(&state, "/").await; + let lease = resolved["lease_id"].as_str().unwrap(); + let digest = sha256_hex(raw); + let rows = state + .storage + .mono_storage() + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap(); + assert_eq!(rows[&oid].size, raw.len() as i64); + assert_eq!(rows[&oid].raw_sha256, Sha256::digest(raw).to_vec()); + assert_eq!(rows[&oid].verification_version, MST2_VERIFICATION_VERSION); + + let directory = app + .clone() + .oneshot(request( + "GET", + &format!("/snapshots/{snapshot}/directory"), + Some(lease), + Body::empty(), + )) + .await + .unwrap(); + assert_eq!(directory.status(), 200); + let directory: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(directory.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(directory["entries"][0]["size"], raw.len().to_string()); + assert_eq!( + directory["entries"][0]["content_digest"], + format!("sha256:{digest}") + ); + let uri = + format!("/snapshots/{snapshot}/blob?path=/{filename}&expected_digest=sha256:{digest}"); + let get = app + .clone() + .oneshot(request("GET", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(get.status(), 200); + assert_eq!( + axum::body::to_bytes(get.into_body(), 128) + .await + .unwrap() + .as_ref(), + raw + ); + let head = app + .clone() + .oneshot(request("HEAD", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(head.status(), 200); + assert_eq!(head.headers()["content-length"], raw.len().to_string()); + assert!( + axum::body::to_bytes(head.into_body(), 128) + .await + .unwrap() + .is_empty() + ); + let again = app + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(again.status(), 200); + let again: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(again.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(again["descriptor"], resolved["descriptor"]); + crate::ceres::snapshot::runtime::runtime() + .release_lease(again["lease_id"].as_str().unwrap()); + crate::ceres::snapshot::runtime::runtime().release_lease(lease); + } + + #[tokio::test] + async fn mst2_invalid_verified_record_is_not_recomputed_into_a_successful_snapshot() { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let filename = format!("corrupt-{}", uuid::Uuid::new_v4()); + let raw = b"valid stored content"; + let oid = seed(&state, &filename, raw).await; + mst2_verified_object::Entity::insert(mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![7; 31]), + size: Set(raw.len() as i64), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }) + .exec(state.storage.mono_storage().get_connection()) + .await + .unwrap(); + let response = snapshot_router::routers(state.clone()) + .with_state(state) + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(response.status(), 502); + let body: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["error"]["code"], "INTEGRITY_ERROR"); + } + + #[tokio::test] + async fn mst2_legacy_fact_cannot_upgrade_when_raw_storage_is_unavailable() { + use sea_orm::{ColumnTrait, QueryFilter}; + + use crate::orbit_api::object_storage::{ObjectKey, ObjectNamespace}; + + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let filename = format!("missing-{}", uuid::Uuid::new_v4()); + let oid = seed(&state, &filename, b"blob 3\0abc").await; + seed_legacy(&state, &oid, b"abc").await; + state + .storage + .git_service + .obj_storage + .inner + .delete(&ObjectKey { + namespace: ObjectNamespace::Git, + key: oid.clone(), + }) + .await + .unwrap(); + let response = snapshot_router::routers(state.clone()) + .with_state(state.clone()) + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from(r#"{"target":{"kind":"latest"}}"#), + )) + .await + .unwrap(); + assert_eq!(response.status(), 503); + let body: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["error"]["code"], "OBJECT_UNAVAILABLE"); + let saved = mst2_verified_object::Entity::find() + .filter(mst2_verified_object::Column::GitOid.eq(oid)) + .one(state.storage.mono_storage().get_connection()) + .await + .unwrap() + .unwrap(); + assert_eq!(saved.verification_version, 1); + assert_eq!(saved.size, 3); + assert_eq!(saved.raw_sha256, Sha256::digest(b"abc").to_vec()); + } + + #[tokio::test] + async fn mst2_shared_blob_paths_upgrade_one_verification_record() { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let raw = b"blob 3\0abc"; + let oid = state + .storage + .git_service + .save_object_from_raw(Bytes::copy_from_slice(raw)) + .await + .unwrap(); + seed_legacy(&state, &oid, b"abc").await; + let blob = Blob::from_content_bytes(raw.to_vec()); + let tree = Tree::from_tree_items(vec![ + TreeItem::new( + TreeItemMode::Blob, + blob.id, + format!("a-{}", uuid::Uuid::new_v4()), + ), + TreeItem::new( + TreeItemMode::Blob, + blob.id, + format!("b-{}", uuid::Uuid::new_v4()), + ), + ]) + .unwrap(); + let handler = crate::ceres::api_service::mono_api_service::MonoApiService::from(&state); + let built = build_directory_page_uncached(&handler, &tree, "/") + .await + .unwrap(); + assert_eq!(built.entries.len(), 2); + assert!(built.entries.iter().all(|entry| entry.size == Some(10) + && entry.content_digest == Some(Sha256::digest(raw).into()))); + let rows = state + .storage + .mono_storage() + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[&oid].size, 10); + assert_eq!(rows[&oid].verification_version, MST2_VERIFICATION_VERSION); + } + + #[tokio::test] + async fn mst2_projection_checks_full_leaf_path_before_resolve_and_http_reads() { + for (component_boundary, valid) in + [(true, true), (true, false), (false, true), (false, false)] + { + let temp = tempfile::TempDir::new().unwrap(); + let state = state(temp.path()).await; + let marker = uuid::Uuid::new_v4().to_string(); + let mut components = if component_boundary { + let mut parts = vec!["d".to_string(); if valid { 255 } else { 256 }]; + parts[0] = marker; + parts.push("file".to_string()); + parts + } else { + let mut parts = vec!["d".repeat(255); 15]; + parts[0].replace_range(..marker.len(), &marker); + parts.push("d".repeat(127)); + parts.push("f".repeat(if valid { 127 } else { 128 })); + parts + }; + let filename = components.pop().unwrap(); + let scope = format!("/{}", components.join("/")); + let full_path = format!("{scope}/{filename}"); + if component_boundary { + assert_eq!( + full_path[1..].split('/').count(), + if valid { 256 } else { 257 } + ); + } else { + assert_eq!(full_path.len(), if valid { 4096 } else { 4097 }); + } + let raw = b"boundary file bytes"; + seed_path(&state, &full_path, raw).await; + let app = snapshot_router::routers(state.clone()).with_state(state.clone()); + let resolve = app + .clone() + .oneshot(request( + "POST", + "/snapshots/resolve", + None, + Body::from( + serde_json::json!({"target":{"kind":"latest"}, "scope":scope}).to_string(), + ), + )) + .await + .unwrap(); + let status = resolve.status(); + let resolved: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(resolve.into_body(), 1_048_576) + .await + .unwrap(), + ) + .unwrap(); + if !valid { + assert_eq!(status, 400); + assert_eq!(resolved["error"]["code"], "SCOPE_INVALID"); + assert!(resolved.get("lease_id").is_none()); + continue; + } + assert_eq!(status, 200); + let snapshot = snapshot_id(&state, &scope).await; + let lease = resolved["lease_id"].as_str().unwrap(); + let uri = format!( + "/snapshots/{snapshot}/blob?path=/{filename}&expected_digest=sha256:{}", + sha256_hex(raw) + ); + let get = app + .oneshot(request("GET", &uri, Some(lease), Body::empty())) + .await + .unwrap(); + assert_eq!(get.status(), 200); + assert_eq!( + axum::body::to_bytes(get.into_body(), 128) + .await + .unwrap() + .as_ref(), + raw + ); + crate::ceres::snapshot::runtime::runtime().release_lease(lease); + } + } +} diff --git a/src/ceres/snapshot/view.rs b/src/ceres/snapshot/view.rs index 5f81cba5..88d1352c 100644 --- a/src/ceres/snapshot/view.rs +++ b/src/ceres/snapshot/view.rs @@ -62,15 +62,20 @@ pub fn validate_scope_relative_path(path: &str) -> Result<(), SnapshotError> { if path == "/" { return Ok(()); } + let mut component_count = 0; for comp in path[1..].split('/') { + component_count += 1; + if component_count > 256 { + return Err(invalid("path has more than 256 components")); + } if comp.is_empty() { return Err(invalid("empty path component")); } if comp == "." || comp == ".." { return Err(invalid("dot component")); } - if comp.len() > 256 { - return Err(invalid("path component over 256 bytes")); + if comp.len() > 255 { + return Err(invalid("path component over 255 bytes")); } } Ok(()) @@ -99,4 +104,14 @@ mod tests { assert!(validate_scope_relative_path("/a/").is_err()); assert!(validate_scope_relative_path("/a\0").is_err()); } + + #[test] + fn scope_relative_path_enforces_byte_and_component_boundaries() { + assert!(validate_scope_relative_path(&format!("/{}", "a".repeat(255))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", "a".repeat(256))).is_err()); + assert!(validate_scope_relative_path(&format!("/{}a", "é".repeat(127))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", "é".repeat(128))).is_err()); + assert!(validate_scope_relative_path(&format!("/{}", vec!["a"; 256].join("/"))).is_ok()); + assert!(validate_scope_relative_path(&format!("/{}", vec!["a"; 257].join("/"))).is_err()); + } } diff --git a/src/jupiter/storage/mono_storage.rs b/src/jupiter/storage/mono_storage.rs index 550d8ced..dc140b2e 100644 --- a/src/jupiter/storage/mono_storage.rs +++ b/src/jupiter/storage/mono_storage.rs @@ -47,6 +47,10 @@ pub struct MonoStorage { pub base: BaseStorage, } +/// Generation 2 hashes the raw file body without guessing a Git envelope. +pub const MST2_VERIFICATION_VERSION: i32 = 2; +const MST2_MAX_FILE_SIZE: i64 = 8_796_093_022_208; + impl Deref for MonoStorage { type Target = BaseStorage; fn deref(&self) -> &Self::Target { @@ -1730,12 +1734,29 @@ impl MonoStorage { .filter(mst2_verified_object::Column::GitOid.is_in(oids)) .all(self.get_connection()) .await?; - Ok(rows.into_iter().map(|m| (m.git_oid.clone(), m)).collect()) + for row in &rows { + if row.state != "VERIFIED" + || ![1, MST2_VERIFICATION_VERSION].contains(&row.verification_version) + || !(0..=MST2_MAX_FILE_SIZE).contains(&row.size) + || row.raw_sha256.len() != 32 + { + return Err(MegaError::ObjStorageInconsistent( + "invalid MST/2 verified blob record".to_string(), + )); + } + } + // Generation 1 may have stripped header-like file bytes. Reverify + // valid legacy rows from object storage before trusting their facts. + Ok(rows + .into_iter() + .filter(|m| m.verification_version == MST2_VERIFICATION_VERSION) + .map(|m| (m.git_oid.clone(), m)) + .collect()) } /// Write-through verification: rows are only written after the content /// was fetched and hashed. Conflicts (concurrent verification of the same - /// oid) resolve to the existing row. + /// oid) keep current facts. Only valid generation-1 facts may be upgraded. pub async fn insert_verified_blobs( &self, rows: Vec, @@ -1743,6 +1764,23 @@ impl MonoStorage { if rows.is_empty() { return Ok(()); } + for row in &rows { + if row.state.try_as_ref().map(String::as_str) != Some("VERIFIED") + || row.verification_version.try_as_ref() != Some(&MST2_VERIFICATION_VERSION) + || !row + .size + .try_as_ref() + .is_some_and(|size| (0..=MST2_MAX_FILE_SIZE).contains(size)) + || !row + .raw_sha256 + .try_as_ref() + .is_some_and(|digest| digest.len() == 32) + { + return Err(MegaError::ObjStorageInconsistent( + "invalid MST/2 verified blob write".to_string(), + )); + } + } match mst2_verified_object::Entity::insert_many(rows) .on_conflict( OnConflict::columns([ @@ -1750,15 +1788,43 @@ impl MonoStorage { mst2_verified_object::Column::GitOid, mst2_verified_object::Column::ObjectKind, ]) - .do_nothing() + .update_columns([ + mst2_verified_object::Column::RawSha256, + mst2_verified_object::Column::Size, + mst2_verified_object::Column::VerificationVersion, + mst2_verified_object::Column::CreatedAt, + ]) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::VerificationVersion, + )) + .eq(1), + ) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::State, + )) + .eq("VERIFIED"), + ) + .action_and_where( + Expr::col(( + mst2_verified_object::Entity, + mst2_verified_object::Column::Size, + )) + .between(0, MST2_MAX_FILE_SIZE), + ) + .action_and_where(Expr::cust( + "length(\"mst2_verified_object\".\"raw_sha256\") = 32", + )) .to_owned(), ) .exec(self.get_connection()) .await { - // ON CONFLICT DO NOTHING returns RecordNotInserted when every - // row was a duplicate; that is the intended first-write-wins - // outcome (same convention as next_insert_retry). + // A skipped conditional conflict returns RecordNotInserted; + // current-generation facts retain first-write-wins semantics. Ok(_) | Err(DbErr::RecordNotInserted) => Ok(()), Err(e) => Err(e.into()), } @@ -3339,7 +3405,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(digest.clone()), size: Set(4_294_967_296), // > i32::MAX: the size64 gap this table closes - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3364,7 +3430,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(vec![9u8; 32]), size: Set(1), - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3381,8 +3447,7 @@ mod tests { #[tokio::test] async fn mst2_verified_blob_insert_real_error_is_not_swallowed() { - // The ON CONFLICT DO NOTHING sentinel (RecordNotInserted on an - // all-duplicate batch) is treated as idempotent success; a genuine + // A skipped conditional conflict is idempotent success; a genuine // database error must still surface, never be masked as Ok. let temp = tempfile::TempDir::new().unwrap(); let storage = test_storage(temp.path()).await; @@ -3402,7 +3467,7 @@ mod tests { object_kind: Set("blob".to_string()), raw_sha256: Set(vec![2u8; 32]), size: Set(2), - verification_version: Set(1), + verification_version: Set(MST2_VERIFICATION_VERSION), state: Set("VERIFIED".to_string()), created_at: Set(chrono::Utc::now().fixed_offset()), }]) @@ -3412,4 +3477,169 @@ mod tests { "a real DB error must propagate, not be swallowed by the ON CONFLICT sentinel match" ); } + + #[tokio::test] + async fn mst2_verified_blob_invalid_records_reject_the_entire_lookup() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + let valid_oid = "f".repeat(40); + let row = |oid: String, state: &str, version, size, digest: Vec| { + mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(digest), + size: Set(size), + verification_version: Set(version), + state: Set(state.to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + } + }; + mono.insert_verified_blobs(vec![row( + valid_oid.clone(), + "VERIFIED", + MST2_VERIFICATION_VERSION, + 0, + vec![7; 32], + )]) + .await + .unwrap(); + for (index, (state, version, size, digest_len)) in [ + ("PENDING", 1, 1, 32), + ("VERIFIED", 0, 1, 32), + ("VERIFIED", 3, 1, 32), + ("VERIFIED", 1, -1, 32), + ("VERIFIED", 1, 8_796_093_022_209, 32), + ("VERIFIED", 1, 1, 31), + ("VERIFIED", 1, 1, 33), + ("VERIFIED", MST2_VERIFICATION_VERSION, -1, 32), + ("VERIFIED", MST2_VERIFICATION_VERSION, 8_796_093_022_209, 32), + ("VERIFIED", MST2_VERIFICATION_VERSION, 1, 31), + ("VERIFIED", MST2_VERIFICATION_VERSION, 1, 33), + ] + .into_iter() + .enumerate() + { + let bad_oid = format!("{index:040x}"); + mst2_verified_object::Entity::insert(row( + bad_oid.clone(), + state, + version, + size, + vec![8; digest_len], + )) + .exec(mono.get_connection()) + .await + .unwrap(); + assert!(matches!( + mono.get_verified_blobs(vec![valid_oid.clone(), bad_oid]) + .await, + Err(MegaError::ObjStorageInconsistent(_)) + )); + assert_eq!( + mono.get_verified_blobs(vec![valid_oid.clone()]) + .await + .unwrap() + .len(), + 1 + ); + } + } + + #[tokio::test] + async fn mst2_verified_blob_accepts_the_eight_tib_boundary() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + let oid = "e".repeat(40); + mono.insert_verified_blobs(vec![mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![9; 32]), + size: Set(8_796_093_022_208), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }]) + .await + .unwrap(); + let rows = mono.get_verified_blobs(vec![oid.clone()]).await.unwrap(); + assert_eq!(rows[&oid].size, 8_796_093_022_208); + } + + #[tokio::test] + async fn mst2_verified_blob_upgrade_only_replaces_valid_legacy_facts() { + let temp = tempfile::TempDir::new().unwrap(); + let storage = test_storage(temp.path()).await; + let mono = storage.mono_storage(); + for (index, (version, state, size, digest_len)) in [ + (1, "VERIFIED", 3, 32), + (1, "PENDING", 3, 32), + (1, "VERIFIED", -1, 32), + (1, "VERIFIED", MST2_MAX_FILE_SIZE + 1, 32), + (1, "VERIFIED", 3, 31), + (MST2_VERIFICATION_VERSION, "VERIFIED", 3, 32), + (3, "VERIFIED", 3, 32), + ] + .into_iter() + .enumerate() + { + let oid = format!("{index:040x}"); + let legacy = mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid.clone()), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![7; digest_len]), + size: Set(size), + verification_version: Set(version), + state: Set(state.to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }; + let old = mst2_verified_object::Entity::insert(legacy) + .exec_with_returning(mono.get_connection()) + .await + .unwrap(); + if index == 0 { + assert!( + mono.get_verified_blobs(vec![oid.clone()]) + .await + .unwrap() + .is_empty() + ); + } + mono.insert_verified_blobs(vec![mst2_verified_object::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + storage_domain: Set("git".to_string()), + git_oid: Set(oid), + object_kind: Set("blob".to_string()), + raw_sha256: Set(vec![9; 32]), + size: Set(10), + verification_version: Set(MST2_VERIFICATION_VERSION), + state: Set("VERIFIED".to_string()), + created_at: Set(chrono::Utc::now().fixed_offset()), + }]) + .await + .unwrap(); + let saved = mst2_verified_object::Entity::find_by_id(old.id) + .one(mono.get_connection()) + .await + .unwrap() + .unwrap(); + if index == 0 { + assert_eq!(saved.verification_version, MST2_VERIFICATION_VERSION); + assert_eq!(saved.size, 10); + assert_eq!(saved.raw_sha256, vec![9; 32]); + } else { + assert_eq!( + saved, old, + "invalid, unknown and current rows are never replaced" + ); + } + } + } } diff --git a/tests/integration_api_write_trunk.rs b/tests/integration_api_write_trunk.rs index 938f5a6e..569340ce 100644 --- a/tests/integration_api_write_trunk.rs +++ b/tests/integration_api_write_trunk.rs @@ -18,7 +18,7 @@ mod git_cli; use std::{ fs, io::Read, - net::{TcpListener, TcpStream}, + net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, ExitStatus, Stdio}, sync::atomic::{AtomicUsize, Ordering}, @@ -235,6 +235,7 @@ struct ServiceProcess { impl ServiceProcess { fn spawn(mut command: Command) -> Self { let child = command.spawn().expect("spawn mega2 service"); + git_cli::record_service_pid(child.id()); Self { child, reaped: false, @@ -711,11 +712,9 @@ fn with_runtime(future: impl std::future::Future) -> T { } fn reserve_free_port() -> u16 { - TcpListener::bind("127.0.0.1:0") - .expect("bind ephemeral port") - .local_addr() - .expect("local addr") - .port() + let port = git_cli::reserve_ephemeral_port(); + git_cli::record_allocated_port(port); + port } fn create_log_file(path: &Path) -> fs::File { From c7a5e3aa33f16245eff1ae13228463d703834571 Mon Sep 17 00:00:00 2001 From: Luxian Date: Sun, 4 Oct 2026 19:30:51 +0800 Subject: [PATCH 06/10] Isolate retention lifecycle test runtime --- src/ceres/snapshot/runtime.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/ceres/snapshot/runtime.rs b/src/ceres/snapshot/runtime.rs index 03cafef6..770c4725 100644 --- a/src/ceres/snapshot/runtime.rs +++ b/src/ceres/snapshot/runtime.rs @@ -349,7 +349,15 @@ mod tests { use crate::ceres::snapshot::retention::{ NodeState, RetainedKind, RetentionRoot, RetentionStore as _, }; - let r = runtime(); + let r = Mst2Runtime { + hmac_key: blake3_key(), + contexts: Mutex::new(HashMap::new()), + leases: Mutex::new(HashMap::new()), + retention: crate::ceres::snapshot::retention::RetentionCoordinator::new( + crate::ceres::snapshot::retention::mem::InMemoryRetentionStore::default(), + ), + tip_state: Mutex::new((String::new(), 0u64)), + }; // Retain the snapshot's derived nodes directly, covered by the lease // root that insert_context created. let lease_root = RetentionRoot::Lease("gc-lease-1".to_string()); From 91724621730ef53de35aab9f26e09e801dae05b8 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:07:03 +0800 Subject: [PATCH 07/10] ci: run isolated Linux repository gates on pull requests --- .github/workflows/repository-gates.yml | 77 ++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 .github/workflows/repository-gates.yml diff --git a/.github/workflows/repository-gates.yml b/.github/workflows/repository-gates.yml new file mode 100644 index 00000000..80a0717e --- /dev/null +++ b/.github/workflows/repository-gates.yml @@ -0,0 +1,77 @@ +name: Repository gates + +on: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: repository-gates-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + linux: + name: Linux repository gates + runs-on: ubuntu-latest + timeout-minutes: 120 + env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: "0" + CARGO_BUILD_JOBS: "2" + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_PROFILE_TEST_DEBUG: "0" + RUST_BACKTRACE: "1" + NO_PROXY: localhost,127.0.0.1,::1,host.docker.internal + no_proxy: localhost,127.0.0.1,::1,host.docker.internal + steps: + - uses: actions/checkout@v4 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y pkg-config libssl-dev protobuf-compiler + + - uses: dtolnay/rust-toolchain@nightly + with: + components: rustfmt + + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy + + - uses: Swatinem/rust-cache@v2 + + - name: Check formatting + run: cargo +nightly fmt --all --check + + - name: Check all targets and features + run: cargo clippy --all-targets --all-features -- -D warnings + + - name: Build + run: cargo build + + - name: Build tests + run: cargo build --tests + + - name: Start isolated test dependencies + shell: bash + run: | + set -euo pipefail + source scripts/lib/mega2-it.sh + mega2_it_ensure_env_test + mega2_it_prepare_git_workdir + docker compose -p mega2-it -f docker/docker-compose.test.yml --profile git \ + up -d --build --wait postgres redis rustfs rustfs-init git-cli + + - name: Run repository tests with test environment + shell: bash + run: | + set -euo pipefail + source .env.test + cargo test --all + + - name: Stop isolated test dependencies + if: always() + run: docker compose -p mega2-it -f docker/docker-compose.test.yml --profile git down -v From c460711cf3183d04a87207e860138c55736e8c1b Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:09:26 +0800 Subject: [PATCH 08/10] ci: avoid duplicate case-insensitive workflow environment keys --- .github/workflows/repository-gates.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/repository-gates.yml b/.github/workflows/repository-gates.yml index 80a0717e..48e7382b 100644 --- a/.github/workflows/repository-gates.yml +++ b/.github/workflows/repository-gates.yml @@ -24,7 +24,6 @@ jobs: CARGO_PROFILE_TEST_DEBUG: "0" RUST_BACKTRACE: "1" NO_PROXY: localhost,127.0.0.1,::1,host.docker.internal - no_proxy: localhost,127.0.0.1,::1,host.docker.internal steps: - uses: actions/checkout@v4 @@ -70,6 +69,7 @@ jobs: run: | set -euo pipefail source .env.test + export no_proxy="$NO_PROXY" cargo test --all - name: Stop isolated test dependencies From e611be0473b28d83a54d46e04336bad15a9f0fdd Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:17:39 +0800 Subject: [PATCH 09/10] ci: handle generated future lint and retain failed gate cache --- .github/workflows/repository-gates.yml | 2 ++ src/ceres/pack/import_repo.rs | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/.github/workflows/repository-gates.yml b/.github/workflows/repository-gates.yml index 48e7382b..b656c665 100644 --- a/.github/workflows/repository-gates.yml +++ b/.github/workflows/repository-gates.yml @@ -41,6 +41,8 @@ jobs: components: clippy - uses: Swatinem/rust-cache@v2 + with: + save-if: "true" - name: Check formatting run: cargo +nightly fmt --all --check diff --git a/src/ceres/pack/import_repo.rs b/src/ceres/pack/import_repo.rs index 6b6bd6ce..b98cbb39 100644 --- a/src/ceres/pack/import_repo.rs +++ b/src/ceres/pack/import_repo.rs @@ -452,6 +452,10 @@ impl RepoHandler for ImportRepo { } } +#[allow( + clippy::double_must_use, + reason = "async_recursion adds must_use to the generated boxed Future" +)] #[async_recursion] pub(crate) async fn collect_git_blob_filepaths( storage: GitDbStorage, From 814820a82458e1cbcf49868c680512067125b5b6 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Mon, 5 Oct 2026 18:14:05 +0800 Subject: [PATCH 10/10] feat(mst2): index fixed namespace bindings by path components (#40) * feat(mst2): add fixed namespace component index * test(mst2): keep namespace oracle inside private domain * test(mst2): register internal namespace regression suite --- src/ceres/snapshot/mod.rs | 7 +- src/ceres/snapshot/namespace.rs | 405 +++++++++++++ src/ceres/snapshot/namespace/tests.rs | 838 ++++++++++++++++++++++++++ 3 files changed, 1247 insertions(+), 3 deletions(-) create mode 100644 src/ceres/snapshot/namespace.rs create mode 100644 src/ceres/snapshot/namespace/tests.rs diff --git a/src/ceres/snapshot/mod.rs b/src/ceres/snapshot/mod.rs index 7ac7fc25..a7c44a61 100644 --- a/src/ceres/snapshot/mod.rs +++ b/src/ceres/snapshot/mod.rs @@ -1,12 +1,13 @@ //! MST/2 snapshot domain (spec 02/03): fixed views over pinned Git trees. //! -//! Slice scope (T02-a): native monorepo views only — no bindings, imports or -//! release policy yet (T04/T05). Everything here resolves against fixed OIDs; -//! no code path in this module may read current refs. +//! Serving remains native-only. The independent namespace index is an unwired +//! composition seam; identity, attestation and publication integration remain +//! separate gates. Fixed-source readers must never look up current refs. pub mod chunks; pub mod descriptor; pub mod error; pub mod frame_stream; +pub mod namespace; pub mod pages; pub mod publication; pub mod resolver; diff --git a/src/ceres/snapshot/namespace.rs b/src/ceres/snapshot/namespace.rs new file mode 100644 index 00000000..720a6171 --- /dev/null +++ b/src/ceres/snapshot/namespace.rs @@ -0,0 +1,405 @@ +//! Immutable component-prefix routing for an already attested namespace. +//! +//! `S` is a caller-owned fixed source handle, not a registry path or ref selector. +//! This module neither encodes NamespaceView identity nor grants authorization, +//! retention or release-write permission. It is not wired into serving/publication. + +use std::collections::{BTreeMap, BTreeSet}; + +use thiserror::Error; + +use super::{ + error::{SnapshotError, SnapshotErrorCode}, + resolver::FsKind, + view::validate_scope_relative_path, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BindingPolicy { + Mutable, + ImmutableRelease, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SourceKind { + Native, + Import, +} + +#[derive(Debug, Clone)] +pub struct FixedBinding { + pub mount_path: String, + pub source: S, + pub source_kind: SourceKind, + /// Canonical source-root-relative path; the source root is the empty string. + pub source_subpath: String, + pub policy: BindingPolicy, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NodeClass { + NativeTree, + NativeCheckoutRoot, + ImportRoot, + ImportTree, + Aggregate, +} + +#[derive(Debug, Clone)] +pub struct SourceNode { + pub fs_kind: FsKind, + pub oid: String, +} + +#[derive(Debug, Clone)] +pub struct SourceEntry { + pub name: String, + pub node: SourceNode, +} + +#[derive(Debug, Clone)] +pub enum SourceOutcome { + Found(SourceNode), + /// Absence established by reading a fixed parent tree, never an I/O failure. + AbsentProven, +} + +/// An adapter must read only the attested fixed root held by `source`, verify +/// membership, and preserve storage errors. Paths are relative to that root; +/// `scope_path` must not be prepended again. Attestation/auth/retention are external. +/// The adapter also enforces the attested scope plus source-path byte budget. +pub trait FixedSourceReader { + fn resolve(&mut self, source: &S, source_path: &str) -> Result; + fn list(&mut self, source: &S, source_path: &str) -> Result, SnapshotError>; +} + +#[derive(Debug)] +pub struct SourceRoute<'a, S> { + pub source: &'a S, + pub source_path: String, + pub source_kind: SourceKind, + pub mount_path: Option<&'a str>, + /// Selected binding provenance, not an inherited release protection or a + /// write permission. A persistent release registry must fence every writer. + pub policy: BindingPolicy, +} + +#[derive(Debug)] +pub struct NamespaceEntry<'a, S> { + pub name: String, + pub fs_kind: FsKind, + /// Synthetic ancestors have neither an object ID nor a source context. + pub oid: Option, + pub source_context: Option>, +} + +#[derive(Debug)] +pub struct NamespaceDirectory<'a, S> { + pub node_class: NodeClass, + pub source_context: Option>, + pub entries: Vec>, +} + +#[derive(Debug, Error)] +pub enum NamespaceError { + #[error(transparent)] + Snapshot(#[from] SnapshotError), + #[error("NAMESPACE_CONFLICT at {0}")] + Conflict(String), + #[error("duplicate mount path {0}")] + DuplicateMount(String), + #[error("binding source subpath is not a directory at {0}")] + SourceSubpathNotDirectory(String), + #[error("invalid source directory entry {0}")] + InvalidSourceEntry(String), +} + +#[derive(Debug, Default)] +struct PrefixNode { + binding: Option, + children: BTreeMap, +} + +/// Construct a new index for each fixed view; no binding/source mutators exist. +/// The trie restricts each request to its path and immediate indexed children. +#[derive(Debug)] +pub struct FixedNamespaceIndex { + native: S, + bindings: Vec>, + root: PrefixNode, + native_checkouts: BTreeSet, +} + +impl FixedNamespaceIndex { + pub fn new>( + native: S, + bindings: Vec>, + native_checkouts: BTreeSet, + reader: &mut R, + ) -> Result { + let mut root = PrefixNode::default(); + for (id, binding) in bindings.iter().enumerate() { + validate_scope_relative_path(&binding.mount_path)?; + validate_relative_path(&binding.source_subpath)?; + let mut node = &mut root; + for component in components(&binding.mount_path) { + node = node.children.entry(component.to_string()).or_default(); + } + if node.binding.replace(id).is_some() { + return Err(NamespaceError::DuplicateMount(binding.mount_path.clone())); + } + } + for checkout in &native_checkouts { + validate_scope_relative_path(checkout)?; + } + let index = Self { + native, + bindings, + root, + native_checkouts, + }; + index.validate_sources(reader)?; + Ok(index) + } + + pub fn route(&self, path: &str) -> Result, NamespaceError> { + self.route_inner(path, false) + } + + fn route_inner( + &self, + path: &str, + exclude_exact: bool, + ) -> Result, NamespaceError> { + validate_scope_relative_path(path)?; + let comps = components(path); + let mut prefix = &self.root; + let mut selected = prefix + .binding + .filter(|_| !(exclude_exact && comps.is_empty())); + let mut selected_depth = 0; + for (depth, component) in comps.iter().enumerate() { + let Some(child) = prefix.children.get(*component) else { + break; + }; + prefix = child; + if let Some(id) = prefix.binding + && !(exclude_exact && depth + 1 == comps.len()) + { + selected = Some(id); + selected_depth = depth + 1; + } + } + let route = if let Some(id) = selected { + let binding = &self.bindings[id]; + let suffix = comps[selected_depth..].join("/"); + let source_path = join_relative(&binding.source_subpath, &suffix); + SourceRoute { + source: &binding.source, + source_path, + source_kind: binding.source_kind, + mount_path: Some(&binding.mount_path), + policy: binding.policy, + } + } else { + SourceRoute { + source: &self.native, + source_path: comps.join("/"), + source_kind: SourceKind::Native, + mount_path: None, + policy: BindingPolicy::Mutable, + } + }; + validate_relative_path(&route.source_path)?; + Ok(route) + } + + fn prefix(&self, path: &str) -> Option<&PrefixNode> { + let mut node = &self.root; + for component in components(path) { + node = node.children.get(component)?; + } + Some(node) + } + + /// Validate all mount crossings once when constructing the index. The + /// binding at the endpoint is excluded so it cannot hide a source file. + fn validate_sources>( + &self, + reader: &mut R, + ) -> Result<(), NamespaceError> { + match reader.resolve(&self.native, "")? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => {} + _ => return Err(NamespaceError::Conflict("/".to_string())), + } + for binding in &self.bindings { + match reader.resolve(&binding.source, &binding.source_subpath)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => {} + _ => { + return Err(NamespaceError::SourceSubpathNotDirectory( + binding.mount_path.clone(), + )); + } + } + let comps = components(&binding.mount_path); + for length in 0..=comps.len() { + let path = format!("/{}", comps[..length].join("/")); + let route = self.route_inner(&path, length == comps.len())?; + if let SourceOutcome::Found(node) = + reader.resolve(route.source, &route.source_path)? + && node.fs_kind != FsKind::Directory + { + return Err(NamespaceError::Conflict(path)); + } + } + } + Ok(()) + } + + /// Return proven absence as None; failures remain errors. Exact mounts + /// replace the underlying source subtree, with only explicit descendants + /// added. No registry enumeration or implicit native/import union occurs. + pub fn list>( + &self, + path: &str, + reader: &mut R, + ) -> Result>, NamespaceError> { + let route = self.route(path)?; + let indexed = self.prefix(path); + let has_children = indexed.is_some_and(|node| !node.children.is_empty()); + let mut entries = BTreeMap::new(); + let mut aggregate = false; + let source_context = match reader.resolve(route.source, &route.source_path)? { + SourceOutcome::AbsentProven if !has_children => return Ok(None), + SourceOutcome::AbsentProven => { + aggregate = true; + None + } + SourceOutcome::Found(node) if node.fs_kind != FsKind::Directory => { + return Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "namespace path is not a directory", + ) + .into()); + } + SourceOutcome::Found(_) => { + for entry in reader.list(route.source, &route.source_path)? { + validate_entry_name(&entry.name)?; + let child_path = append_absolute(path, &entry.name); + let context = self.route(&child_path)?; + let name = entry.name; + if entries + .insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: entry.node.fs_kind, + oid: Some(entry.node.oid), + source_context: Some(context), + }, + ) + .is_some() + { + return Err(NamespaceError::InvalidSourceEntry(name)); + } + } + Some(route) + } + }; + if let Some(prefix) = indexed { + for (name, child) in &prefix.children { + let child_path = append_absolute(path, name); + if child.binding.is_some() { + let route = self.route(&child_path)?; + let node = match reader.resolve(route.source, &route.source_path)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => node, + _ => return Err(NamespaceError::SourceSubpathNotDirectory(child_path)), + }; + aggregate = true; + entries.insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: node.fs_kind, + oid: Some(node.oid), + source_context: Some(route), + }, + ); + } else if let Some(entry) = entries.get(name) { + if entry.fs_kind != FsKind::Directory { + return Err(NamespaceError::Conflict(child_path)); + } + } else { + aggregate = true; + entries.insert( + name.clone(), + NamespaceEntry { + name: name.clone(), + fs_kind: FsKind::Directory, + oid: None, + source_context: None, + }, + ); + } + } + } + let node_class = if aggregate { + NodeClass::Aggregate + } else if let Some(route) = &source_context { + match route.source_kind { + SourceKind::Import if route.mount_path == Some(path) => NodeClass::ImportRoot, + SourceKind::Import => NodeClass::ImportTree, + SourceKind::Native if self.native_checkouts.contains(path) => { + NodeClass::NativeCheckoutRoot + } + SourceKind::Native => NodeClass::NativeTree, + } + } else { + NodeClass::Aggregate + }; + Ok(Some(NamespaceDirectory { + node_class, + source_context, + entries: entries.into_values().collect(), + })) + } +} + +fn components(path: &str) -> Vec<&str> { + if path == "/" { + Vec::new() + } else { + path[1..].split('/').collect() + } +} + +fn join_relative(parent: &str, suffix: &str) -> String { + match (parent.is_empty(), suffix.is_empty()) { + (true, _) => suffix.to_string(), + (_, true) => parent.to_string(), + _ => format!("{parent}/{suffix}"), + } +} + +fn append_absolute(parent: &str, name: &str) -> String { + if parent == "/" { + format!("/{name}") + } else { + format!("{parent}/{name}") + } +} + +fn validate_relative_path(path: &str) -> Result<(), SnapshotError> { + validate_scope_relative_path(&format!("/{path}")) +} + +fn validate_entry_name(name: &str) -> Result<(), NamespaceError> { + if name.contains('/') || validate_relative_path(name).is_err() || name.is_empty() { + return Err(NamespaceError::InvalidSourceEntry(name.to_string())); + } + Ok(()) +} + +#[cfg(test)] +#[path = "namespace/tests.rs"] +mod tests; diff --git a/src/ceres/snapshot/namespace/tests.rs b/src/ceres/snapshot/namespace/tests.rs new file mode 100644 index 00000000..6b2fa27e --- /dev/null +++ b/src/ceres/snapshot/namespace/tests.rs @@ -0,0 +1,838 @@ +use std::{ + collections::{BTreeMap, BTreeSet}, + io::Write, + path::Path, + process::{Command, Stdio}, +}; + +use tempfile::TempDir; + +use crate::ceres::snapshot::{ + error::{SnapshotError, SnapshotErrorCode}, + namespace::{ + BindingPolicy, FixedBinding, FixedNamespaceIndex, FixedSourceReader, NamespaceError, + NodeClass, SourceEntry, SourceKind, SourceNode, SourceOutcome, + }, + resolver::FsKind, +}; + +#[derive(Default)] +struct MemoryReader { + nodes: BTreeMap<(String, String), SourceNode>, + fail_path: Option, + fail_listing: bool, + resolves: usize, + lists: usize, +} + +impl MemoryReader { + fn add(&mut self, source: &str, path: &str, fs_kind: FsKind) { + self.nodes.insert( + (source.to_string(), path.to_string()), + SourceNode { + fs_kind, + oid: format!("{source}:{path}"), + }, + ); + } + + fn roots() -> Self { + let mut reader = Self::default(); + for source in ["native", "a20", "a21", "b30"] { + reader.add(source, "", FsKind::Directory); + } + reader + } +} + +impl FixedSourceReader for MemoryReader { + fn resolve(&mut self, source: &String, path: &str) -> Result { + self.resolves += 1; + if self.fail_path.as_deref() == Some(path) { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "injected missing backend object", + )); + } + Ok(self + .nodes + .get(&(source.clone(), path.to_string())) + .cloned() + .map_or(SourceOutcome::AbsentProven, SourceOutcome::Found)) + } + + fn list(&mut self, source: &String, path: &str) -> Result, SnapshotError> { + self.lists += 1; + if self.fail_listing { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "injected listing failure", + )); + } + let prefix = if path.is_empty() { + String::new() + } else { + format!("{path}/") + }; + Ok(self + .nodes + .iter() + .filter_map(|((owner, node_path), node)| { + if owner != source { + return None; + } + let suffix = node_path.strip_prefix(&prefix)?; + if suffix.is_empty() || suffix.contains('/') { + return None; + } + Some(SourceEntry { + name: suffix.to_string(), + node: node.clone(), + }) + }) + .collect()) + } +} + +fn binding(path: &str, source: &str, subpath: &str) -> FixedBinding { + FixedBinding { + mount_path: path.to_string(), + source: source.to_string(), + source_kind: SourceKind::Import, + source_subpath: subpath.to_string(), + policy: BindingPolicy::Mutable, + } +} + +#[test] +fn component_prefix_source_subpath_and_nested_routing_match_manual_table() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "src", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/x/lib", "a20", "src"), + binding("/x/lib/vendor/b", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + for (path, expected_source, expected_path) in [ + ("/", "native", ""), + ("/x/lib", "a20", "src"), + ("/x/lib/a", "a20", "src/a"), + ("/x/library/a", "native", "x/library/a"), + ("/x/lib/vendor", "a20", "src/vendor"), + ("/x/lib/vendor/b/z", "b30", "z"), + ("/x/lib/vendor/beta/z", "a20", "src/vendor/beta/z"), + ("/x/lib/é+\\name", "a20", "src/é+\\name"), + ] { + let route = index.route(path).unwrap(); + assert_eq!(route.source, expected_source, "{path}"); + assert_eq!(route.source_path, expected_path, "{path}"); + } +} + +#[test] +fn synthetic_ancestors_have_no_source_and_exact_mount_has_no_native_union() { + let mut reader = MemoryReader::roots(); + reader.add("native", "lib", FsKind::Directory); + reader.add("native", "lib/native-only", FsKind::Regular); + reader.add("a20", "src", FsKind::Directory); + reader.add("a20", "src/import-only", FsKind::Executable); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/lib", "a20", "src"), + binding("/missing/deep/mount", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let synthetic = index.list("/missing", &mut reader).unwrap().unwrap(); + assert_eq!(synthetic.node_class, NodeClass::Aggregate); + assert!(synthetic.source_context.is_none()); + assert_eq!(synthetic.entries[0].name, "deep"); + assert!(synthetic.entries[0].oid.is_none()); + assert!(synthetic.entries[0].source_context.is_none()); + let imported = index.list("/lib", &mut reader).unwrap().unwrap(); + assert_eq!(imported.node_class, NodeClass::ImportRoot); + assert_eq!(imported.entries.len(), 1); + assert_eq!(imported.entries[0].name, "import-only"); + assert_eq!(imported.entries[0].fs_kind, FsKind::Executable); + assert!(index.list("/absent", &mut reader).unwrap().is_none()); +} + +#[test] +fn node_class_depends_on_direct_composition_and_explicit_checkout() { + let mut reader = MemoryReader::roots(); + reader.add("native", "app", FsKind::Directory); + reader.add("native", "plain", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/app/vendor/lib", "a20", "")], + BTreeSet::from(["/app".to_string(), "/plain".to_string()]), + &mut reader, + ) + .unwrap(); + assert_eq!( + index.list("/", &mut reader).unwrap().unwrap().node_class, + NodeClass::NativeTree + ); + let app = index.list("/app", &mut reader).unwrap().unwrap(); + assert_eq!(app.node_class, NodeClass::Aggregate); + assert!(app.source_context.is_some()); + assert_eq!( + index + .list("/plain", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::NativeCheckoutRoot + ); +} + +#[test] +fn file_and_symlink_crossings_and_exact_mount_targets_are_conflicts() { + for kind in [FsKind::Regular, FsKind::Executable, FsKind::Symlink] { + for mount in ["/blocked", "/blocked/child"] { + let mut reader = MemoryReader::roots(); + reader.add("native", "blocked", kind); + let error = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding(mount, "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap_err(); + assert!(matches!(error, NamespaceError::Conflict(path) if path == "/blocked")); + } + } + let mut reader = MemoryReader::roots(); + reader.add("a20", "file", FsKind::Regular); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/import", "a20", "file")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::SourceSubpathNotDirectory(_)) + )); +} + +#[test] +fn nested_binding_cannot_hide_an_outer_source_file() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "blocked", FsKind::Symlink); + let result = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/import", "a20", ""), + binding("/import/blocked/nested", "b30", ""), + ], + BTreeSet::new(), + &mut reader, + ); + assert!(matches!(result, Err(NamespaceError::Conflict(path)) if path == "/import/blocked")); +} + +#[test] +fn removing_binding_restores_native_in_new_index_and_preserves_old_index() { + let mut reader = MemoryReader::roots(); + reader.add("native", "lib", FsKind::Directory); + reader.add("native", "lib/native-only", FsKind::Regular); + reader.add("a20", "old-only", FsKind::Regular); + reader.add("a21", "new-only", FsKind::Regular); + let old = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let new = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert_eq!( + old.list("/lib", &mut reader).unwrap().unwrap().entries[0].name, + "old-only" + ); + assert_eq!( + new.list("/lib", &mut reader).unwrap().unwrap().entries[0].name, + "native-only" + ); + assert_eq!(old.route("/lib/unvisited/deep").unwrap().source, "a20"); +} + +#[test] +fn duplicate_mounts_and_noncanonical_paths_are_rejected() { + let mut reader = MemoryReader::roots(); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/same", "a20", ""), binding("/same", "b30", "")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::DuplicateMount(_)) + )); + for path in ["", "relative", "/a/", "/a//b", "/a/..", "/a\0"] { + assert!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding(path, "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .is_err(), + "{path:?}" + ); + } + for subpath in ["/src", "src/", "src//a", "src/..", "src\0"] { + assert!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", subpath)], + BTreeSet::new(), + &mut reader, + ) + .is_err(), + "{subpath:?}" + ); + } +} + +#[test] +fn composed_source_path_limits_are_checked_after_prefix_replacement() { + let mut reader = MemoryReader::roots(); + let subpath = vec!["a"; 256].join("/"); + reader.add("a20", &subpath, FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", &subpath)], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(index.route("/lib").is_ok()); + assert!(index.route("/lib/extra").is_err()); + assert!(index.route(&format!("/{}", "a".repeat(256))).is_err()); + assert!( + index + .route(&format!("/{}", vec!["a"; 257].join("/"))) + .is_err() + ); + let max_bytes = vec!["a".repeat(255); 16].join("/"); + assert_eq!(max_bytes.len() + 1, 4096); + reader.add("a20", &max_bytes, FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", &max_bytes)], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(index.route("/lib").is_ok()); + assert!(index.route("/lib/x").is_err()); +} + +#[test] +fn import_root_class_yields_to_direct_binding_and_import_tree_class_is_retained() { + let mut reader = MemoryReader::roots(); + reader.add("a20", "src", FsKind::Directory); + reader.add("a20", "src/plain", FsKind::Directory); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![ + binding("/lib", "a20", "src"), + binding("/lib/nested", "b30", ""), + ], + BTreeSet::from(["/lib".to_string()]), + &mut reader, + ) + .unwrap(); + let root = index.list("/lib", &mut reader).unwrap().unwrap(); + assert_eq!(root.node_class, NodeClass::Aggregate); + assert_eq!(root.source_context.as_ref().unwrap().source, "a20"); + assert_eq!( + index + .list("/lib/plain", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::ImportTree + ); + assert_eq!( + index + .list("/lib/nested", &mut reader) + .unwrap() + .unwrap() + .node_class, + NodeClass::ImportRoot + ); +} + +#[test] +fn storage_failure_is_preserved_in_construction_and_listing() { + let mut reader = MemoryReader::roots(); + let index = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + reader.fail_path = Some("missing".to_string()); + assert!(matches!( + index.list("/missing", &mut reader), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); + assert!(matches!( + FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/lib", "a20", "missing")], + BTreeSet::new(), + &mut reader, + ), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); + reader.fail_path = None; + reader.fail_listing = true; + assert!(matches!( + index.list("/", &mut reader), + Err(NamespaceError::Snapshot(error)) if error.code == SnapshotErrorCode::ObjectUnavailable + )); +} + +#[test] +fn root_binding_replaces_entire_native_source() { + let mut reader = MemoryReader::roots(); + reader.add("native", "native-only", FsKind::Regular); + reader.add("a20", "import-only", FsKind::Regular); + let index = FixedNamespaceIndex::new( + "native".to_string(), + vec![binding("/", "a20", "")], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert_eq!(index.route("/").unwrap().source, "a20"); + assert_eq!(index.route("/anything").unwrap().source_path, "anything"); + let root = index.list("/", &mut reader).unwrap().unwrap(); + assert_eq!(root.node_class, NodeClass::ImportRoot); + assert_eq!(root.entries.len(), 1); + assert_eq!(root.entries[0].name, "import-only"); +} + +#[test] +fn invalid_and_duplicate_source_entries_are_rejected() { + struct MalformedReader(Vec); + impl FixedSourceReader for MalformedReader { + fn resolve(&mut self, _: &String, _: &str) -> Result { + Ok(SourceOutcome::Found(SourceNode { + fs_kind: FsKind::Directory, + oid: "root".to_string(), + })) + } + + fn list(&mut self, _: &String, _: &str) -> Result, SnapshotError> { + Ok(self + .0 + .iter() + .map(|name| SourceEntry { + name: name.clone(), + node: SourceNode { + fs_kind: FsKind::Regular, + oid: "blob".to_string(), + }, + }) + .collect()) + } + } + for names in [vec![""], vec![".."], vec!["a/b"], vec!["same", "same"]] { + let mut reader = MalformedReader(names.iter().map(|s| s.to_string()).collect()); + let index = FixedNamespaceIndex::new( + "native".to_string(), + Vec::new(), + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + assert!(matches!( + index.list("/", &mut reader), + Err(NamespaceError::InvalidSourceEntry(_)) + )); + } +} + +#[test] +fn unrelated_binding_count_does_not_expand_source_reads_for_a_listing() { + let mut reader = MemoryReader::roots(); + reader.add("native", "app", FsKind::Directory); + let bindings = (0..512) + .map(|i| binding(&format!("/elsewhere-{i}/lib"), "a20", "")) + .collect(); + let index = + FixedNamespaceIndex::new("native".to_string(), bindings, BTreeSet::new(), &mut reader) + .unwrap(); + reader.resolves = 0; + reader.lists = 0; + index.list("/app", &mut reader).unwrap().unwrap(); + assert_eq!((reader.resolves, reader.lists), (1, 1)); +} + +fn git(repo: &Path, args: &[&str], input: Option<&[u8]>) -> String { + let mut child = Command::new("git") + .current_dir(repo) + .env("GIT_AUTHOR_NAME", "MST2 Oracle") + .env("GIT_AUTHOR_EMAIL", "mst2-oracle@example.invalid") + .env("GIT_COMMITTER_NAME", "MST2 Oracle") + .env("GIT_COMMITTER_EMAIL", "mst2-oracle@example.invalid") + .env("GIT_AUTHOR_DATE", "2026-09-15T00:00:00Z") + .env("GIT_COMMITTER_DATE", "2026-09-15T00:00:00Z") + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + if let Some(input) = input { + child.stdin.take().unwrap().write_all(input).unwrap(); + } else { + drop(child.stdin.take()); + } + let output = child.wait_with_output().unwrap(); + assert!( + output.status.success(), + "{args:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +} + +fn make_tree(repo: &Path, files: &[(&str, &str)]) -> String { + let mut items = Vec::new(); + let mut directories: BTreeMap<&str, Vec<(&str, &str)>> = BTreeMap::new(); + for (path, bytes) in files { + if let Some((head, tail)) = path.split_once('/') { + directories.entry(head).or_default().push((tail, bytes)); + } else { + let oid = git( + repo, + &["hash-object", "-w", "--stdin"], + Some(bytes.as_bytes()), + ); + items.push(format!("100644 blob {}\t{path}\n", oid.trim())); + } + } + for (name, children) in directories { + let oid = make_tree(repo, &children); + items.push(format!("040000 tree {oid}\t{name}\n")); + } + git(repo, &["mktree"], Some(items.concat().as_bytes())) + .trim() + .to_string() +} + +#[derive(Debug, Clone)] +struct GitSource { + root: String, + commit: String, + scope_path: String, +} + +struct GitReader<'a> { + repo: &'a Path, +} + +impl FixedSourceReader for GitReader<'_> { + fn resolve(&mut self, source: &GitSource, path: &str) -> Result { + if path.is_empty() { + return Ok(SourceOutcome::Found(SourceNode { + fs_kind: FsKind::Directory, + oid: source.root.clone(), + })); + } + let mut tree = source.root.clone(); + let components: Vec<_> = path.split('/').collect(); + for (position, component) in components.iter().enumerate() { + let Some(entry) = self + .read_tree(&tree) + .into_iter() + .find(|entry| entry.name == *component) + else { + return Ok(SourceOutcome::AbsentProven); + }; + if position + 1 == components.len() { + return Ok(SourceOutcome::Found(entry.node)); + } + if entry.node.fs_kind != FsKind::Directory { + return Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "fixed Git path crossing", + )); + } + tree = entry.node.oid; + } + unreachable!() + } + + fn list(&mut self, source: &GitSource, path: &str) -> Result, SnapshotError> { + match self.resolve(source, path)? { + SourceOutcome::Found(node) if node.fs_kind == FsKind::Directory => { + Ok(self.read_tree(&node.oid)) + } + _ => Err(SnapshotError::new( + SnapshotErrorCode::NotDirectory, + "fixed Git listing", + )), + } + } +} + +impl GitReader<'_> { + fn read_tree(&self, oid: &str) -> Vec { + git(self.repo, &["ls-tree", "-z", oid], None) + .split('\0') + .filter(|row| !row.is_empty()) + .map(|row| { + let (metadata, name) = row.split_once('\t').unwrap(); + let fields: Vec<_> = metadata.split_whitespace().collect(); + SourceEntry { + name: name.to_string(), + node: SourceNode { + fs_kind: if fields[0] == "040000" { + FsKind::Directory + } else { + FsKind::Regular + }, + oid: fields[2].to_string(), + }, + } + }) + .collect() + } +} + +fn fixed_source(repo: &Path, files: &[(&str, &str)]) -> GitSource { + let root = make_tree(repo, files); + let commit = git( + repo, + &["commit-tree", &root], + Some(b"fixed oracle fixture\n"), + ) + .trim() + .to_string(); + GitSource { + root, + commit, + scope_path: "/".to_string(), + } +} + +#[test] +fn attested_scoped_root_is_not_prefixed_again_when_subpath_and_nested_mount_are_used() { + let repo = TempDir::new().unwrap(); + git( + repo.path(), + &["init", "--bare", "--object-format=sha1", "--quiet"], + None, + ); + let native = fixed_source(repo.path(), &[("native.txt", "native")]); + let full = fixed_source( + repo.path(), + &[ + ("scope/src/deep/leaf.txt", "scoped"), + ("other/outside.txt", "outside"), + ], + ); + let scoped = GitSource { + root: git( + repo.path(), + &["rev-parse", &format!("{}:scope", full.commit)], + None, + ) + .trim() + .to_string(), + commit: full.commit.clone(), + scope_path: "/scope".to_string(), + }; + let nested = fixed_source(repo.path(), &[("inside/nested.txt", "nested")]); + let mut reader = GitReader { repo: repo.path() }; + let index = FixedNamespaceIndex::new( + native, + vec![ + git_binding("/lib", &scoped, "src"), + git_binding("/lib/deep/nested", &nested, "inside"), + ], + BTreeSet::new(), + &mut reader, + ) + .unwrap(); + let route = index.route("/lib/deep/leaf.txt").unwrap(); + assert_eq!(route.source.scope_path, "/scope"); + assert_eq!(route.source_path, "src/deep/leaf.txt"); + let route = index.route("/lib/deep/nested/nested.txt").unwrap(); + assert_eq!(route.source_path, "inside/nested.txt"); + let mut files = BTreeMap::new(); + traverse(&index, &mut reader, "/", &mut files); + assert_eq!( + files, + BTreeMap::from([ + ("/native.txt".to_string(), b"native".to_vec()), + ("/lib/deep/leaf.txt".to_string(), b"scoped".to_vec()), + ( + "/lib/deep/nested/nested.txt".to_string(), + b"nested".to_vec() + ), + ]) + ); +} + +fn git_binding(path: &str, source: &GitSource, subpath: &str) -> FixedBinding { + FixedBinding { + mount_path: path.to_string(), + source: source.clone(), + source_kind: SourceKind::Import, + source_subpath: subpath.to_string(), + policy: BindingPolicy::Mutable, + } +} + +/// Independently flatten commits using Git, then compose a handwritten fixture. +/// This oracle does not call the production trie or source reader to select paths. +fn flatten_commit(repo: &Path, source: &GitSource) -> BTreeMap> { + git( + repo, + &["ls-tree", "-rz", "--full-tree", &source.commit], + None, + ) + .split('\0') + .filter(|row| !row.is_empty()) + .map(|row| { + let (metadata, path) = row.split_once('\t').unwrap(); + let oid = metadata.split_whitespace().nth(2).unwrap(); + ( + path.to_string(), + git(repo, &["cat-file", "blob", oid], None).into_bytes(), + ) + }) + .collect() +} + +fn traverse( + index: &FixedNamespaceIndex, + reader: &mut GitReader<'_>, + path: &str, + files: &mut BTreeMap>, +) { + for entry in index.list(path, reader).unwrap().unwrap().entries { + let child = if path == "/" { + format!("/{}", entry.name) + } else { + format!("{path}/{}", entry.name) + }; + if entry.fs_kind == FsKind::Directory { + traverse(index, reader, &child, files); + } else { + files.insert( + child, + git( + reader.repo, + &["cat-file", "blob", entry.oid.as_deref().unwrap()], + None, + ) + .into_bytes(), + ); + } + } +} + +#[test] +fn composed_listing_matches_independent_fixed_git_oracle_after_ref_and_path_reuse() { + let repo = TempDir::new().unwrap(); + git( + repo.path(), + &["init", "--bare", "--object-format=sha1", "--quiet"], + None, + ); + let native = fixed_source( + repo.path(), + &[ + ("project/app/native.txt", "native"), + ("lib/native-hidden.txt", "hidden"), + ("library/keep.txt", "neighbor"), + ], + ); + let a20 = fixed_source( + repo.path(), + &[ + ("src/deep/old.txt", "A20"), + ("src/vendor/b/hidden.txt", "outer-hidden"), + ("elsewhere/not-visible.txt", "outside-subpath"), + ], + ); + let a21 = fixed_source(repo.path(), &[("src/deep/new.txt", "A21")]); + let b30 = fixed_source(repo.path(), &[("nested.txt", "B30")]); + let bindings = vec![ + git_binding("/lib", &a20, "src"), + git_binding("/lib/vendor/b", &b30, ""), + git_binding("/missing/deep/import", &b30, ""), + ]; + let mut reader = GitReader { repo: repo.path() }; + let old = + FixedNamespaceIndex::new(native.clone(), bindings, BTreeSet::new(), &mut reader).unwrap(); + // Simulate advancing current refs and reassigning a registry path. The + // reader has only the captured fixed handle; no registry/ref read is possible. + git( + repo.path(), + &["update-ref", "refs/heads/import-current", &a21.commit], + None, + ); + let current_registry = BTreeMap::from([("/lib", a21.clone())]); + assert_ne!( + current_registry["/lib"].commit, + old.route("/lib/deep/old.txt").unwrap().source.commit + ); + let mut expected = BTreeMap::new(); + for (path, bytes) in flatten_commit(repo.path(), &native) { + if !path.starts_with("lib/") { + expected.insert(format!("/{path}"), bytes); + } + } + for (path, bytes) in flatten_commit(repo.path(), &a20) { + if let Some(suffix) = path.strip_prefix("src/") + && !suffix.starts_with("vendor/b/") + { + expected.insert(format!("/lib/{suffix}"), bytes); + } + } + for (path, bytes) in flatten_commit(repo.path(), &b30) { + expected.insert(format!("/lib/vendor/b/{path}"), bytes.clone()); + expected.insert(format!("/missing/deep/import/{path}"), bytes); + } + let mut actual = BTreeMap::new(); + traverse(&old, &mut reader, "/", &mut actual); + assert_eq!(actual, expected); + assert!(actual.contains_key("/lib/deep/old.txt")); + assert!(!actual.contains_key("/lib/deep/new.txt")); + let unbound = + FixedNamespaceIndex::new(native.clone(), Vec::new(), BTreeSet::new(), &mut reader).unwrap(); + let mut restored = BTreeMap::new(); + traverse(&unbound, &mut reader, "/", &mut restored); + let expected_native = flatten_commit(repo.path(), &native) + .into_iter() + .map(|(path, bytes)| (format!("/{path}"), bytes)) + .collect::>(); + assert_eq!(restored, expected_native); + let mut old_again = BTreeMap::new(); + traverse(&old, &mut reader, "/", &mut old_again); + assert_eq!(old_again, expected); +}