From c68b6d5c696c603393f3467e4859888301a09fb9 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 7 Oct 2026 22:19:18 +0800 Subject: [PATCH] fix(mst2): keep lookup metadata-only and restore native regression coverage Use one fixed-path metadata walk and current verified size/digest facts for lookup without whole-content reads. Preserve the original zero-body-read assertions and HEAD behavior. Correct three observed native fixture failures without changing production publication or payload guards. Native execution of this combined source remains pending. --- src/api/router/snapshot_content.rs | 17 +- .../router/snapshot_lookup_metadata_tests.rs | 271 ++++++++++++++++++ src/api/router/snapshot_router.rs | 26 +- src/api/router/snapshot_session_tests.rs | 3 + src/contract/git_protocol/mod.rs | 1 + .../service/native_publication_push_tests.rs | 39 ++- .../storage/native_metadata_install_tests.rs | 71 ++++- 7 files changed, 392 insertions(+), 36 deletions(-) create mode 100644 src/api/router/snapshot_lookup_metadata_tests.rs diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index d1dd2a15..6e419a61 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -34,11 +34,11 @@ struct ResolvedFile { raw: Vec, } -struct ResolvedFileMetadata { +pub(super) struct ResolvedFileMetadata { fs_kind: FsKind, oid: String, - digest: [u8; 32], - size: u64, + pub(super) digest: [u8; 32], + pub(super) size: u64, } #[allow(clippy::result_large_err)] @@ -93,6 +93,17 @@ async fn resolve_file_metadata( + handler: &T, + fs_kind: FsKind, + oid: String, + path: &str, + expected_digest: Option<&str>, +) -> Result { let storage = handler.get_context(); let mut verified = storage .mono_storage() diff --git a/src/api/router/snapshot_lookup_metadata_tests.rs b/src/api/router/snapshot_lookup_metadata_tests.rs new file mode 100644 index 00000000..81c11cdb --- /dev/null +++ b/src/api/router/snapshot_lookup_metadata_tests.rs @@ -0,0 +1,271 @@ +use super::*; + +fn lookup_body(paths: &[&str]) -> Body { + Body::from(json!({"paths": paths}).to_string()) +} + +#[tokio::test] +async fn mst2_durable_http_lookup_uses_verified_metadata_after_service_rebuild_without_body_reads() +{ + let fixture = Fixture::new_with_pg_config(true).await; + let state = rebuilt(&fixture).await; + assert!(state.storage.native_snapshot_sessions.get().is_none()); + assert!(!Arc::ptr_eq( + &state.storage.native_projection_cache, + &fixture.state.storage.native_projection_cache + )); + let paths = [ + "/file", + "/alias", + "/executable", + "/empty", + "/link", + "/nested/file", + "/directory", + "/nested", + "/missing", + "/file/child", + "/link/child", + ]; + let value = success_json( + app(&state) + .oneshot(fixture.request("POST", "lookup", lookup_body(&paths))) + .await + .unwrap(), + ) + .await; + assert_eq!(value["snapshot_id"], fixture.snapshot); + let results = value["results"].as_array().unwrap(); + assert_eq!(results.len(), paths.len()); + for (result, path) in results.iter().zip(paths) { + assert_eq!(result["path"], path); + } + for (index, kind, size, content_digest) in [ + (0, "regular", fixture.raw.len(), fixture.digest_string()), + (1, "regular", fixture.raw.len(), fixture.digest_string()), + (2, "executable", fixture.raw.len(), fixture.digest_string()), + (3, "regular", 0, format!("sha256:{}", hex_of(&digest(&[])))), + ( + 4, + "symlink", + 4, + format!("sha256:{}", hex_of(&digest(b"file"))), + ), + (5, "regular", fixture.raw.len(), fixture.digest_string()), + ] { + assert_eq!(results[index]["status"], "found"); + assert_eq!( + results[index]["node"], + json!({ + "fs_kind": kind, + "name": paths[index].rsplit('/').next().unwrap(), + "size": size.to_string(), + "content_digest": content_digest, + }) + ); + } + let proofs = value["proof_pages"].as_array().unwrap(); + assert!(!proofs.is_empty()); + let mut proof_digests = std::collections::HashSet::new(); + for proof in proofs { + let bytes = STANDARD + .decode(proof["data_base64"].as_str().unwrap()) + .unwrap(); + mst2_codec::metapage::Page::decode(&bytes).unwrap(); + let digest = format!("sha256:{}", hex_of(&mst2_codec::metapage::page_id(&bytes))); + assert_eq!(proof["digest"], digest); + assert!(proof_digests.insert(digest)); + } + for index in [6, 7] { + let result = &results[index]; + assert_eq!(result["status"], "found"); + assert_eq!(result["node"]["fs_kind"], "directory"); + assert_eq!( + result["node"]["name"], + paths[index].rsplit('/').next().unwrap() + ); + assert_eq!(result["node"]["node_class"], "native_tree"); + assert_eq!(result["node"]["lifecycle"], "mutable"); + assert!(proof_digests.contains(result["node"]["directory_root"].as_str().unwrap())); + } + for (index, status) in [ + (8, "absent"), + (9, "not_directory"), + (10, "symlink_traversal"), + ] { + assert_eq!( + results[index], + json!({"path": paths[index], "status": status}) + ); + } + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_lookup_missing_and_noncurrent_facts_never_fall_back_to_body_or_cache() { + let fixture = Fixture::new_with_pg_config(true).await; + fixture.map("/file").await; + let original = fixture.fact().await; + fixture.counts.reset(); + fixture.delete_fact().await; + error( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + 503, + "METADATA_NOT_READY", + true, + ) + .await; + fixture.counts.assert(0, 0); + for (domain, kind, generation) in [ + ("git", "blob", 1), + ("other", "blob", MST2_VERIFICATION_VERSION), + ("git", "tree", MST2_VERIFICATION_VERSION), + ] { + let mut fact = original.clone(); + fact.storage_domain = domain.to_string(); + fact.object_kind = kind.to_string(); + fact.verification_version = generation; + fixture.replace_fact(fact).await; + error( + fixture + .send("POST", "lookup", lookup_body(&["/alias"])) + .await, + 503, + "METADATA_NOT_READY", + true, + ) + .await; + fixture.counts.assert(0, 0); + } + fixture.replace_fact(original).await; + success_json( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + ) + .await; + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_lookup_invalid_verified_facts_fail_before_body_reads() { + let fixture = Fixture::new_with_pg_config(true).await; + let original = fixture.fact().await; + for case in 0..5 { + let mut fact = original.clone(); + match case { + 0 => fact.state = "PENDING".to_string(), + 1 => fact.verification_version = MST2_VERIFICATION_VERSION + 1, + 2 => fact.size = -1, + 3 => fact.size = 8_796_093_022_209, + 4 => { + fact.raw_sha256.pop().unwrap(); + } + _ => unreachable!(), + } + fixture.replace_fact(fact).await; + error( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(0, 0); + } + fixture.replace_fact(original).await; + success_json( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + ) + .await; + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_lookup_symlink_facts_outside_profile_fail_without_body_reads() { + let fixture = Fixture::new_with_pg_config(true).await; + let mono = fixture.state.storage.mono_storage(); + let main = mono.get_main_ref("/").await.unwrap().unwrap(); + let handler = MonoApiService::from(&fixture.state); + let root = handler.get_tree_by_hash(&main.ref_tree_hash).await.unwrap(); + let MetadataWalkOutcome::FoundFile { oid, .. } = + resolve_abs_metadata(&handler, &root, "/project/link") + .await + .unwrap() + else { + panic!("fixture link must be a fixed file"); + }; + let original = mono + .get_verified_blobs(vec![oid.clone()]) + .await + .unwrap() + .remove(&oid) + .unwrap(); + for size in [0, 4096] { + let mut fact = original.clone().into_active_model(); + fact.size = Set(size); + fact.update(mono.get_connection()).await.unwrap(); + error( + fixture + .send("POST", "lookup", lookup_body(&["/link"])) + .await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + fixture.counts.assert(0, 0); + } + let mut restored = original.into_active_model(); + restored.size = Set(4); + restored.update(mono.get_connection()).await.unwrap(); + let result = success_json( + fixture + .send("POST", "lookup", lookup_body(&["/link"])) + .await, + ) + .await; + assert_eq!(result["results"][0]["node"]["size"], "4"); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_lookup_verified_fact_database_failure_is_typed_without_body_reads() { + let fixture = Fixture::new_with_pg_config(true).await; + let mono = fixture.state.storage.mono_storage(); + mono.get_connection() + .execute_unprepared( + "ALTER TABLE mst2_verified_object RENAME TO mst2_verified_object_unavailable", + ) + .await + .unwrap(); + error( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + 500, + "INTERNAL", + true, + ) + .await; + fixture.counts.assert(0, 0); + mono.get_connection() + .execute_unprepared( + "ALTER TABLE mst2_verified_object_unavailable RENAME TO mst2_verified_object", + ) + .await + .unwrap(); + success_json( + fixture + .send("POST", "lookup", lookup_body(&["/file"])) + .await, + ) + .await; + fixture.counts.assert(0, 0); +} diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index f2c3515f..40876942 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -28,8 +28,8 @@ use crate::{ descriptor::build as build_descriptor, error::{SnapshotError, SnapshotErrorCode}, pages::{ - WalkOutcome, base64_of, build_directory_page, build_directory_page_with_work, hex_of, - proof_pages, resolve_abs, + MetadataWalkOutcome, WalkOutcome, base64_of, build_directory_page, + build_directory_page_with_work, hex_of, proof_pages, resolve_abs, resolve_abs_metadata, }, projection_observation::{NativeResolveSource, ResolvedProjection}, runtime::{now_unix, runtime}, @@ -1299,11 +1299,11 @@ async fn lookup( validate_scope_relative_path(path).map_err(mst2_error_response)?; let abs_path = abs_view_path(&ctx.built.descriptor.scope, path); let mut entry = json!({"path": path}); - match resolve_abs(handler.as_ref(), &root_tree, &abs_path) + match resolve_abs_metadata(handler.as_ref(), &root_tree, &abs_path) .await .map_err(mst2_error_response)? { - WalkOutcome::FoundDir => { + MetadataWalkOutcome::FoundDir => { let built = build_directory_page(handler.as_ref(), &root_tree, &abs_path) .await .map_err(mst2_error_response)?; @@ -1320,25 +1320,23 @@ async fn lookup( entry["node"] = node; deepest_dirs.push(abs_path); } - WalkOutcome::FoundFile { - fs_kind, - size, - digest, - .. - } => { + MetadataWalkOutcome::FoundFile { fs_kind, oid } => { + let fact = + content::verified_file_metadata(handler.as_ref(), fs_kind, oid, path, None) + .await?; entry["status"] = json!("found"); let mut node = json!({"fs_kind": fs_kind.as_str()}); if let Some(name) = path.rsplit('/').next() { node["name"] = json!(name); } - node["size"] = json!(size.to_string()); - node["content_digest"] = json!(format!("sha256:{}", hex_of(&digest))); + node["size"] = json!(fact.size.to_string()); + node["content_digest"] = json!(format!("sha256:{}", hex_of(&fact.digest))); entry["node"] = node; } - WalkOutcome::Absent => { + MetadataWalkOutcome::Absent => { entry["status"] = json!("absent"); } - WalkOutcome::NotDirectory { symlink } => { + MetadataWalkOutcome::NotDirectory { symlink } => { entry["status"] = if symlink { json!("symlink_traversal") } else { diff --git a/src/api/router/snapshot_session_tests.rs b/src/api/router/snapshot_session_tests.rs index 187499a4..7cf495b4 100644 --- a/src/api/router/snapshot_session_tests.rs +++ b/src/api/router/snapshot_session_tests.rs @@ -1125,3 +1125,6 @@ async fn mst2_durable_http_warm_reads_renew_and_frame_delivery_reject_primary_sc success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; fixture.counts.assert(0, 0); } + +#[path = "snapshot_lookup_metadata_tests.rs"] +mod metadata_lookup; diff --git a/src/contract/git_protocol/mod.rs b/src/contract/git_protocol/mod.rs index e3b4dc7d..6a87531e 100644 --- a/src/contract/git_protocol/mod.rs +++ b/src/contract/git_protocol/mod.rs @@ -954,6 +954,7 @@ mod tests { .unwrap(); let bytes = Arc::new(Mutex::new(Vec::new())); let subscriber = tracing_subscriber::fmt() + .with_ansi(false) .with_writer(Writer(bytes.clone())) .finish(); let guard = tracing::subscriber::set_default(subscriber); diff --git a/src/jupiter/service/native_publication_push_tests.rs b/src/jupiter/service/native_publication_push_tests.rs index 8f2f9f76..4fb89aa1 100644 --- a/src/jupiter/service/native_publication_push_tests.rs +++ b/src/jupiter/service/native_publication_push_tests.rs @@ -314,13 +314,22 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d assert!(matches!(wh03_exec(&storage,id).await,ExecuteOutcome::Done{..})); tokio::time::timeout(Duration::from_secs(5),release.wait()).await.unwrap(); let ((status,delayed),delayed_observations)=tokio::time::timeout(Duration::from_secs(10),&mut reader).await.unwrap().unwrap(); - assert_eq!(status,200,"{delayed}");assert_eq!(delayed["descriptor"],v1["descriptor"]); - assert_eq!(delayed["publication_sequence"],v1["publication_sequence"]); - assert_eq!(delayed_observations.len(),1); - let delayed_identity=delayed_observations[0].test_identity(); - for field in ["root_commit_oid","root_tree_oid","native_certificate_receipt_id","native_writer_epoch","native_publication_sequence","snapshot_id","metadata_root"] { - assert_eq!(delayed_identity[field],first_identity[field],"{field} mixed with a later publication"); - } + assert_eq!(status,503,"{delayed}"); + assert_eq!(delayed["error"]["code"],"SNAPSHOT_NOT_READY"); + assert_eq!(delayed["error"]["retryable"],true); + assert!(delayed.get("descriptor").is_none()); + assert!(delayed.get("publication_sequence").is_none()); + assert!(delayed_observations.is_empty(),"rejected stale handoff emitted a success observation"); + use tower::ServiceExt; + let old_app=crate::api::router::snapshot_router::routers(state.clone()).with_state(state.clone()); + let old=old_app.oneshot(axum::http::Request::builder().method("GET") + .uri(format!("/snapshots/{}/descriptor",v1["descriptor"]["snapshot_id"].as_str().unwrap())) + .header("x-mega-snapshot-lease",v1["lease_id"].as_str().unwrap()) + .body(axum::body::Body::empty()).unwrap()).await.unwrap(); + assert_eq!(old.status(),200); + let old=axum::body::to_bytes(old.into_body(),1_048_576).await.unwrap(); + let old:serde_json::Value=serde_json::from_slice(&old).unwrap(); + assert_eq!(old["descriptor"],v1["descriptor"],"already protected old SID changed with the writer"); let ((status,v2),next_observations)=crate::ceres::snapshot::projection_observation::with_observations(http_resolve(state.clone(),"/")).await; assert_eq!(status,200,"{v2}");assert_eq!(v2["publication_sequence"],"2"); assert_ne!(v2["descriptor"]["snapshot_id"],v1["descriptor"]["snapshot_id"]); @@ -328,6 +337,14 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d let next_identity=next_observations[0].test_identity(); assert_eq!(next_identity["native_publication_sequence"],v2["publication_sequence"]); assert_eq!(next_identity["snapshot_id"],v2["descriptor"]["snapshot_id"]); + assert_eq!(next_identity["metadata_root"],v2["descriptor"]["metadata_root"]); + assert_eq!(next_identity["namespace_view_id"],v2["descriptor"]["namespace_view_id"]); + assert_eq!(next_identity["instance_id"],v2["descriptor"]["instance_id"]); + assert_eq!(next_identity["native_writer_epoch"],v2["writer_epoch"]); + let native_v2=storage.mono_storage().read_native_publication_head(storage.config().mst2.instance_uuid.as_deref().unwrap()).await.unwrap(); + assert_eq!(next_identity["root_commit_oid"],git_internal::hash::ObjectHash::from_hex_for_kind(git_internal::hash::get_hash_kind(),&native_v2.root.commit).unwrap().to_tagged_string()); + assert_eq!(next_identity["root_tree_oid"],git_internal::hash::ObjectHash::from_hex_for_kind(git_internal::hash::get_hash_kind(),&native_v2.root.tree).unwrap().to_tagged_string()); + assert_eq!(next_identity["native_certificate_receipt_id"].as_u64(),Some(native_v2.token.certificate.unwrap() as u64)); assert_ne!(next_identity["native_certificate_receipt_id"],first_identity["native_certificate_receipt_id"]); let ((status,_),failed_observations)=crate::ceres::snapshot::projection_observation::with_observations(http_resolve(state,"/absent-observation-scope")).await; assert_eq!(status,404); @@ -338,8 +355,8 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d let directory=writer.test_directory(_temp.path()); let records=std::fs::read_to_string(directory.join("records.jsonl")).unwrap(); let records:Vec=records.lines().map(|line|serde_json::from_str(line).unwrap()).collect(); - assert_eq!(records.len(),4,"failed resolve must not be a successful observation"); - for (record,observation) in [(&records[0],&first_observations[0]),(&records[2],&delayed_observations[0]),(&records[3],&next_observations[0])] { + assert_eq!(records.len(),3,"failed resolve must not be a successful observation"); + for (record,observation) in [(&records[0],&first_observations[0]),(&records[2],&next_observations[0])] { let typed=serde_json::to_value(observation.wire_record()).unwrap(); assert_eq!(record["payload"],typed,"writer must preserve the full validated operation tuple"); assert_eq!(record["payload"].as_object().unwrap().len(),38); @@ -347,8 +364,8 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d assert_ne!(records[0]["payload"]["request_id"],records[2]["payload"]["request_id"]); let status:serde_json::Value=serde_json::from_slice(&std::fs::read(directory.join("status.json")).unwrap()).unwrap(); assert_eq!(status["closed"],true); - assert_eq!(status["accepted_records"],4); - assert_eq!(status["written_records"],4); + assert_eq!(status["accepted_records"],3); + assert_eq!(status["written_records"],3); assert_eq!(status["first_error_code"],0); } } diff --git a/src/jupiter/storage/native_metadata_install_tests.rs b/src/jupiter/storage/native_metadata_install_tests.rs index 4b47fe38..f7662117 100644 --- a/src/jupiter/storage/native_metadata_install_tests.rs +++ b/src/jupiter/storage/native_metadata_install_tests.rs @@ -458,24 +458,79 @@ async fn overwrite_installed_payload_for_test( bytes: Vec, ) { assert_eq!(bytes.len(), page.bytes.len()); + let guard_modes = payload_trigger_modes_for_test(connection).await; + assert!( + guard_modes + .iter() + .any(|(name, _)| name == "mst2_metadata_payload_fenced") + ); + assert!(guard_modes.iter().all(|(_, mode)| mode == "O")); + assert!( + connection + .execute_raw(statement( + "UPDATE mst2_metadata_payload SET payload=payload WHERE page_id=$1", + [page.id.to_vec().into()], + )) + .await + .is_err(), + "production payload UPDATE fence must still reject writes" + ); let txn = connection.begin().await.unwrap(); + txn.execute_unprepared("SELECT pg_advisory_xact_lock(1296717362,hashtext(current_schema()))") + .await + .unwrap(); txn.execute_unprepared( - "ALTER TABLE mst2_metadata_payload DISABLE TRIGGER mst2_metadata_payload_immutable", + "ALTER TABLE mst2_metadata_payload DISABLE TRIGGER mst2_metadata_payload_fenced", ) .await .unwrap(); - txn.execute_raw(statement( - "UPDATE mst2_metadata_payload SET payload=$2 WHERE page_id=$1", - [page.id.to_vec().into(), bytes.into()], - )) - .await - .unwrap(); + let result = txn + .execute_raw(statement( + "UPDATE mst2_metadata_payload SET payload=$2 WHERE page_id=$1", + [page.id.to_vec().into(), bytes.into()], + )) + .await + .unwrap(); + assert_eq!(result.rows_affected(), 1); txn.execute_unprepared( - "ALTER TABLE mst2_metadata_payload ENABLE TRIGGER mst2_metadata_payload_immutable", + "ALTER TABLE mst2_metadata_payload ENABLE TRIGGER mst2_metadata_payload_fenced", ) .await .unwrap(); txn.commit().await.unwrap(); + assert_eq!( + payload_trigger_modes_for_test(connection).await, + guard_modes + ); + assert!( + connection + .execute_raw(statement( + "UPDATE mst2_metadata_payload SET payload=payload WHERE page_id=$1", + [page.id.to_vec().into()], + )) + .await + .is_err(), + "corruption injection must restore the production UPDATE fence" + ); +} + +async fn payload_trigger_modes_for_test(connection: &DatabaseConnection) -> Vec<(String, String)> { + connection + .query_all_raw(statement( + "SELECT tgname,tgenabled::text AS mode FROM pg_trigger \ + WHERE tgrelid='mst2_metadata_payload'::regclass AND NOT tgisinternal ORDER BY tgname", + [], + )) + .await + .unwrap() + .into_iter() + .map(|row| { + ( + row.try_get("", "tgname").unwrap(), + row.try_get("", "mode").unwrap(), + ) + }) + .collect() } #[tokio::test]