From 6bd9b9c991d89f9040cdafbc6729e2cb583d4ac1 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 16:20:59 +0800 Subject: [PATCH 1/2] Bind native resolve projection traces to fixed publication source --- src/api/router/snapshot_router.rs | 66 ++- src/ceres/snapshot/mod.rs | 1 + src/ceres/snapshot/pages.rs | 26 +- src/ceres/snapshot/projection_observation.rs | 554 ++++++++++++++++++ .../service/native_publication_push_tests.rs | 36 +- 5 files changed, 651 insertions(+), 32 deletions(-) create mode 100644 src/ceres/snapshot/projection_observation.rs diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 9330837b..b6fe5813 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -15,6 +15,7 @@ use axum::{ }; use base64::Engine; use bytes::Bytes; +use git_internal::hash::{ObjectHash, get_hash_kind}; use mst2_codec::descriptor; use request::Mst2Bytes; use serde::Deserialize; @@ -26,7 +27,11 @@ use crate::{ ceres::snapshot::{ descriptor::build as build_descriptor, error::{SnapshotError, SnapshotErrorCode}, - pages::{WalkOutcome, base64_of, build_directory_page, hex_of, proof_pages, resolve_abs}, + pages::{ + WalkOutcome, base64_of, build_directory_page, build_directory_page_with_work, hex_of, + proof_pages, resolve_abs, + }, + projection_observation::{NativeResolveSource, ResolvedProjection}, runtime::{now_unix, runtime}, view::{SnapshotView, validate_scope_relative_path}, }, @@ -423,6 +428,7 @@ async fn resolve( })?; let config = state.storage.config(); + let mut native_source = None; let (commit_oid, tree_oid, sequence, writer_epoch) = if config.mst2.publication_enabled { let Some(instance) = config.mst2.instance_uuid.as_deref() else { return Err(mst2_error_response(SnapshotError::new( @@ -442,6 +448,34 @@ async fn resolve( "native publication is not ready", )) })?; + let observation_source = ObjectHash::from_hex_for_kind(get_hash_kind(), &head.root.commit) + .and_then(|commit| { + ObjectHash::from_hex_for_kind(get_hash_kind(), &head.root.tree) + .map(|tree| (commit, tree)) + }) + .map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "invalid fixed native source identity", + ) + }) + .and_then(|(commit, tree)| { + NativeResolveSource::capture( + &head.instance_id, + commit, + tree, + head.token.certificate, + head.token.epoch, + head.token.sequence, + ) + }); + native_source = match observation_source { + Ok(source) => Some(source), + Err(_) => { + tracing::warn!("native resolve observation source rejected"); + None + } + }; ( head.root.commit, head.root.tree, @@ -498,9 +532,12 @@ async fn resolve( .map_err(internal)?; // The scope root doubles as metadata_root; building it also validates // that the scope exists and is a directory in this view. - let scope_page = build_directory_page(handler.as_ref(), &root_tree, &req.scope) - .await - .map_err(mst2_error_response)?; + let projection_started = std::time::Instant::now(); + let (scope_page, projection_work) = + build_directory_page_with_work(handler.as_ref(), &root_tree, &req.scope) + .await + .map_err(mst2_error_response)?; + let projection_elapsed = projection_started.elapsed(); let built = build_descriptor(&config.mst2, &view, &req.scope, scope_page.page_id) .map_err(mst2_error_response)?; @@ -508,6 +545,27 @@ async fn resolve( .insert_context(built.clone(), &commit_oid, &tree_oid, req.lease_seconds) .map_err(mst2_error_response)?; + if let Some(source) = native_source { + let request_id = current_request_id(); + match source.observe( + ResolvedProjection { + descriptor: &ctx.built.descriptor, + snapshot_id: &ctx.built.snapshot_id, + metadata_root: &ctx.built.metadata_root, + context_commit: &ctx.commit_oid, + context_root_tree: &ctx.root_tree_oid, + fixed_root_tree: root_tree.id, + requested_scope: &req.scope, + request_id: &request_id, + }, + projection_work, + projection_elapsed, + ) { + Ok(observation) => observation.emit(), + Err(_) => tracing::warn!("native resolve observation context rejected"), + } + } + let body = json!({ "descriptor": descriptor_json(&built), "publication_sequence": sequence, diff --git a/src/ceres/snapshot/mod.rs b/src/ceres/snapshot/mod.rs index b5a55d9e..3c408ace 100644 --- a/src/ceres/snapshot/mod.rs +++ b/src/ceres/snapshot/mod.rs @@ -8,6 +8,7 @@ pub mod descriptor; pub mod error; pub mod frame_stream; pub mod pages; +pub(crate) mod projection_observation; pub mod publication; pub mod resolver; pub mod retention; diff --git a/src/ceres/snapshot/pages.rs b/src/ceres/snapshot/pages.rs index d3666eb2..f7e3c7b8 100644 --- a/src/ceres/snapshot/pages.rs +++ b/src/ceres/snapshot/pages.rs @@ -23,11 +23,13 @@ use mst2_codec::{ use sea_orm::ActiveValue::Set; use sha2::{Digest, Sha256}; +pub use crate::ceres::snapshot::projection_observation::ProjectionWork; use crate::{ ceres::{ api_service::ApiHandler, snapshot::{ error::{SnapshotError, SnapshotErrorCode}, + projection_observation::NATIVE_PROJECTION_REVISION, resolver::FsKind, retention_dag::{MetadataDagBuilder, MetadataDagLimits, ValidatedMetadataDag}, view::hex, @@ -82,28 +84,6 @@ impl DescendantPathBudget { } } -/// Actual work in one on-demand native projection. These are memoization -/// counters, not durable publication or retention accounting. Page counters -/// cover returned directory-root pages only: the codec does not expose the -/// radix pages it internally encodes while building a root. -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct ProjectionWork { - pub directories_rebuilt: u64, - pub reused_subtree_roots: u64, - pub directory_root_pages_built: u64, - pub directory_root_page_bytes_built: u64, - pub directory_root_pages_reused: u64, - pub directory_root_page_bytes_reused: u64, - pub directory_entries_scanned: u64, - pub scope_path_entries_examined: u64, - /// Backend tree requests, excluding the caller-supplied fixed root tree. - pub tree_fetches: u64, - pub verified_blob_hits: u64, - pub verified_blob_misses: u64, - pub raw_bytes_fetched: u64, - pub raw_bytes_hashed: u64, -} - #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct NativeProjectionKey { source_domain: &'static str, @@ -129,7 +109,7 @@ impl NativeProjectionKey { fs_semantics: FS_SEMANTICS_LINUX_CODE_V1, access_projection: ACCESS_PROJECTION_EXACT_FULL, verification_revision: MST2_VERIFICATION_VERSION, - projection_revision: 1, + projection_revision: NATIVE_PROJECTION_REVISION, } } } diff --git a/src/ceres/snapshot/projection_observation.rs b/src/ceres/snapshot/projection_observation.rs new file mode 100644 index 00000000..3e0dc083 --- /dev/null +++ b/src/ceres/snapshot/projection_observation.rs @@ -0,0 +1,554 @@ +//! Operation-local resolve diagnostics, not publication or retention receipts. + +use std::time::Duration; + +use git_internal::hash::ObjectHash; +use mst2_codec::descriptor::{ + ACCESS_PROJECTION_EXACT_FULL, FS_SEMANTICS_LINUX_CODE_V1, MATERIALIZATION_POLICY_GIT_RAW_V1, + METADATA_CODEC, SCHEMA_VERSION, ServingDescriptor, +}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::{ + ceres::snapshot::error::{SnapshotError, SnapshotErrorCode}, + jupiter::storage::mono_storage::MST2_VERIFICATION_VERSION, +}; + +pub(crate) const NATIVE_PROJECTION_REVISION: u16 = 1; + +/// Work in one directory projection. Page counters cover returned directory +/// roots, excluding codec-internal radix encoding and later route traversal. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ProjectionWork { + pub directories_rebuilt: u64, + /// Cache-hit boundary visits, not all descendants or unique source OIDs. + pub reused_subtree_roots: u64, + pub directory_root_pages_built: u64, + pub directory_root_page_bytes_built: u64, + pub directory_root_pages_reused: u64, + pub directory_root_page_bytes_reused: u64, + pub directory_entries_scanned: u64, + pub scope_path_entries_examined: u64, + /// Backend tree requests, excluding the caller-supplied fixed root tree. + pub tree_fetches: u64, + /// Verification-record outcomes per file entry, not unique body OIDs. + pub verified_blob_hits: u64, + pub verified_blob_misses: u64, + /// Returned raw-body lengths, not physical disk/network traffic. + pub raw_bytes_fetched: u64, + /// Bytes passed to this builder's explicit raw-file SHA-256. + pub raw_bytes_hashed: u64, +} + +/// Captured only from the validated native head, before resolve projects it. +pub(crate) struct NativeResolveSource { + instance: Uuid, + commit: ObjectHash, + tree: ObjectHash, + certificate_receipt_id: u64, + writer_epoch: u64, + publication_sequence: u64, +} + +pub(crate) struct ResolvedProjection<'a> { + pub descriptor: &'a ServingDescriptor, + pub snapshot_id: &'a str, + pub metadata_root: &'a str, + pub context_commit: &'a str, + pub context_root_tree: &'a str, + pub fixed_root_tree: ObjectHash, + pub requested_scope: &'a str, + pub request_id: &'a str, +} + +/// Immutable successful resolve observation. This grants no authorization, +/// retention, durable publication or proof of codec-internal construction work. +pub(crate) struct NativeProjectionObservation { + source: NativeResolveSource, + scope: String, + namespace_view_id: String, + snapshot_id: String, + metadata_root: String, + request_id: String, + projection_elapsed_micros: u64, + work: ProjectionWork, +} + +fn invalid() -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "native resolve observation does not match its fixed source and context", + ) +} + +impl NativeResolveSource { + pub(crate) fn capture( + instance_id: &str, + commit: ObjectHash, + tree: ObjectHash, + certificate_receipt_id: Option, + writer_epoch: i64, + publication_sequence: i64, + ) -> Result { + let instance = Uuid::parse_str(instance_id).map_err(|_| invalid())?; + let positive = |value: i64| { + u64::try_from(value) + .ok() + .filter(|value| *value > 0) + .ok_or_else(invalid) + }; + if instance.to_string() != instance_id || commit.kind() != tree.kind() { + return Err(invalid()); + } + Ok(Self { + instance, + commit, + tree, + certificate_receipt_id: positive(certificate_receipt_id.ok_or_else(invalid)?)?, + writer_epoch: positive(writer_epoch)?, + publication_sequence: positive(publication_sequence)?, + }) + } + + pub(crate) fn observe( + self, + resolved: ResolvedProjection<'_>, + work: ProjectionWork, + projection_elapsed: Duration, + ) -> Result { + let mut view_hash = Sha256::new(); + view_hash.update(b"mega.mst2.namespaceview\0"); + view_hash.update(self.commit.to_string().as_bytes()); + let expected_view: [u8; 32] = view_hash.finalize().into(); + let descriptor = resolved.descriptor; + let digest_text = |bytes: &[u8]| format!("sha256:{}", hex::encode(bytes)); + let snapshot = descriptor.snapshot_id().map_err(|_| invalid())?; + let bounded_request_id = !resolved.request_id.is_empty() + && resolved.request_id.len() <= 128 + && resolved.request_id.bytes().all(|byte| { + matches!(byte, b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b':' | b'/') + }); + if resolved.context_commit != self.commit.to_string() + || resolved.context_root_tree != self.tree.to_string() + || resolved.fixed_root_tree != self.tree + || descriptor.instance_uuid != *self.instance.as_bytes() + || descriptor.namespace_view_id != expected_view + || descriptor.scope != resolved.requested_scope + || resolved.snapshot_id != digest_text(&snapshot) + || resolved.metadata_root != digest_text(&descriptor.metadata_root) + || !bounded_request_id + { + return Err(invalid()); + } + Ok(NativeProjectionObservation { + source: self, + scope: descriptor.scope.clone(), + namespace_view_id: digest_text(&descriptor.namespace_view_id), + snapshot_id: resolved.snapshot_id.to_owned(), + metadata_root: resolved.metadata_root.to_owned(), + request_id: resolved.request_id.to_owned(), + projection_elapsed_micros: u64::try_from(projection_elapsed.as_micros()) + .map_err(|_| invalid())?, + work, + }) + } +} + +impl NativeProjectionObservation { + pub(crate) fn emit(self) { + tracing::debug!( + target: "mst2::native_projection_observation", + observation_revision = 1u16, + phase = "resolve_directory_projection", + source_domain = "native-git", + request_id = %self.request_id, + instance_id = %self.source.instance, + root_commit_oid = %self.source.commit.to_tagged_string(), + root_tree_oid = %self.source.tree.to_tagged_string(), + native_certificate_receipt_id = self.source.certificate_receipt_id, + native_writer_epoch = self.source.writer_epoch, + native_publication_sequence = self.source.publication_sequence, + scope = ?self.scope, + schema_version = SCHEMA_VERSION, + metadata_codec = METADATA_CODEC, + materialization_policy = MATERIALIZATION_POLICY_GIT_RAW_V1, + fs_semantics = FS_SEMANTICS_LINUX_CODE_V1, + access_projection = ACCESS_PROJECTION_EXACT_FULL, + verification_revision = MST2_VERIFICATION_VERSION, + projection_revision = NATIVE_PROJECTION_REVISION, + namespace_view_id = %self.namespace_view_id, + snapshot_id = %self.snapshot_id, + metadata_root = %self.metadata_root, + projection_elapsed_micros = self.projection_elapsed_micros, + page_counter_scope = "returned-directory-root-pages", + codec_radix_work = "NOT_EXPOSED", + directories_rebuilt = self.work.directories_rebuilt, + reused_subtree_roots = self.work.reused_subtree_roots, + directory_root_pages_built = self.work.directory_root_pages_built, + directory_root_page_bytes_built = self.work.directory_root_page_bytes_built, + directory_root_pages_reused = self.work.directory_root_pages_reused, + directory_root_page_bytes_reused = self.work.directory_root_page_bytes_reused, + directory_entries_scanned = self.work.directory_entries_scanned, + scope_path_entries_examined = self.work.scope_path_entries_examined, + tree_fetches = self.work.tree_fetches, + verified_blob_hits = self.work.verified_blob_hits, + verified_blob_misses = self.work.verified_blob_misses, + raw_bytes_fetched = self.work.raw_bytes_fetched, + raw_bytes_hashed = self.work.raw_bytes_hashed, + "native resolve directory projection succeeded" + ); + #[cfg(test)] + let _ = OBSERVATIONS.try_with(|observations| observations.lock().unwrap().push(self)); + } +} + +#[cfg(test)] +tokio::task_local! { + static OBSERVATIONS: std::sync::Arc>>; +} + +#[cfg(test)] +pub(crate) async fn with_observations( + future: F, +) -> (F::Output, Vec) { + let observations = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let result = OBSERVATIONS.scope(observations.clone(), future).await; + let records = std::mem::take(&mut *observations.lock().unwrap()); + (result, records) +} + +#[cfg(test)] +impl NativeProjectionObservation { + pub(crate) fn test_identity(&self) -> serde_json::Value { + serde_json::json!({ + "instance_id": self.source.instance.to_string(), + "root_commit_oid": self.source.commit.to_tagged_string(), + "root_tree_oid": self.source.tree.to_tagged_string(), + "native_certificate_receipt_id": self.source.certificate_receipt_id, + "native_writer_epoch": self.source.writer_epoch.to_string(), + "native_publication_sequence": self.source.publication_sequence.to_string(), + "scope": self.scope, + "namespace_view_id": self.namespace_view_id, + "snapshot_id": self.snapshot_id, + "metadata_root": self.metadata_root, + "request_id": self.request_id, + }) + } +} + +#[cfg(test)] +mod tests { + use std::{ + collections::BTreeMap, + sync::{Arc, Mutex}, + }; + + use git_internal::hash::HashKind; + use tracing::{ + Event, Metadata, Subscriber, + field::{Field, Visit}, + span::{Attributes, Id, Record}, + }; + + use super::*; + + const INSTANCE: &str = "11111111-2222-4333-8444-555555555559"; + + struct Fixture { + commit: ObjectHash, + tree: ObjectHash, + descriptor: ServingDescriptor, + snapshot_id: String, + metadata_root: String, + } + + impl Fixture { + fn new() -> Self { + let commit = ObjectHash::from_hex_for_kind(HashKind::Sha256, &"a".repeat(64)).unwrap(); + let tree = ObjectHash::from_hex_for_kind(HashKind::Sha256, &"b".repeat(64)).unwrap(); + let mut view_hash = Sha256::new(); + view_hash.update(b"mega.mst2.namespaceview\0"); + view_hash.update(commit.to_string().as_bytes()); + let descriptor = ServingDescriptor { + instance_uuid: *Uuid::parse_str(INSTANCE).unwrap().as_bytes(), + namespace_view_id: view_hash.finalize().into(), + scope: "/project".to_owned(), + metadata_root: [0xc; 32], + }; + Self { + commit, + tree, + snapshot_id: format!("sha256:{}", hex::encode(descriptor.snapshot_id().unwrap())), + metadata_root: format!("sha256:{}", hex::encode(descriptor.metadata_root)), + descriptor, + } + } + + fn source(&self) -> NativeResolveSource { + NativeResolveSource::capture(INSTANCE, self.commit, self.tree, Some(97), 3, 7).unwrap() + } + + fn resolved<'a>(&'a self, commit: &'a str, tree: &'a str) -> ResolvedProjection<'a> { + ResolvedProjection { + descriptor: &self.descriptor, + snapshot_id: &self.snapshot_id, + metadata_root: &self.metadata_root, + context_commit: commit, + context_root_tree: tree, + fixed_root_tree: self.tree, + requested_scope: "/project", + request_id: "resolve:request-1", + } + } + } + + #[test] + fn observation_binds_fixed_source_context_descriptor_and_operation_work() { + let fixture = Fixture::new(); + let commit = fixture.commit.to_string(); + let tree = fixture.tree.to_string(); + let work = ProjectionWork { + directories_rebuilt: 3, + reused_subtree_roots: 70, + directory_entries_scanned: 104, + verified_blob_misses: 1, + ..ProjectionWork::default() + }; + let observation = fixture + .source() + .observe( + fixture.resolved(&commit, &tree), + work.clone(), + Duration::from_micros(123), + ) + .unwrap(); + assert_eq!(observation.source.certificate_receipt_id, 97); + assert_eq!(observation.source.publication_sequence, 7); + assert_eq!(observation.source.writer_epoch, 3); + assert_eq!( + observation.source.tree.to_tagged_string(), + format!("sha256:{tree}") + ); + assert_eq!(observation.projection_elapsed_micros, 123); + assert_eq!(observation.work, work); + assert_eq!(observation.scope, "/project"); + assert_eq!(observation.snapshot_id, fixture.snapshot_id); + } + + #[test] + fn observation_rejects_each_changed_binding_or_unbounded_trace_token() { + let fixture = Fixture::new(); + let commit = fixture.commit.to_string(); + let tree = fixture.tree.to_string(); + for mode in 0..8 { + let mut descriptor = fixture.descriptor.clone(); + let mut resolved = fixture.resolved(&commit, &tree); + match mode { + 0 => resolved.context_commit = &tree, + 1 => resolved.context_root_tree = &commit, + 2 => resolved.fixed_root_tree = fixture.commit, + 3 => descriptor.instance_uuid[0] ^= 1, + 4 => descriptor.namespace_view_id[0] ^= 1, + 5 => resolved.requested_scope = "/different", + 6 => resolved.snapshot_id = &fixture.metadata_root, + 7 => resolved.metadata_root = &fixture.snapshot_id, + _ => unreachable!(), + } + resolved.descriptor = &descriptor; + assert!( + fixture + .source() + .observe(resolved, ProjectionWork::default(), Duration::ZERO) + .is_err(), + "mode {mode}" + ); + } + for request_id in [ + "".to_owned(), + "x".repeat(129), + "bad\ntrace".to_owned(), + "actor secret".to_owned(), + ] { + let mut resolved = fixture.resolved(&commit, &tree); + resolved.request_id = &request_id; + assert!( + fixture + .source() + .observe(resolved, ProjectionWork::default(), Duration::ZERO) + .is_err() + ); + } + } + + #[test] + fn capture_requires_positive_native_certificate_and_preserves_hash_kind() { + let fixture = Fixture::new(); + for (certificate, epoch, sequence) in [ + (None, 1, 1), + (Some(0), 1, 1), + (Some(-1), 1, 1), + (Some(1), 0, 1), + (Some(1), 1, 0), + ] { + assert!( + NativeResolveSource::capture( + INSTANCE, + fixture.commit, + fixture.tree, + certificate, + epoch, + sequence + ) + .is_err() + ); + } + let blake_tree = + ObjectHash::from_hex_for_kind(HashKind::Blake3, &fixture.tree.to_string()).unwrap(); + assert!( + NativeResolveSource::capture(INSTANCE, fixture.commit, blake_tree, Some(1), 1, 1) + .is_err() + ); + } + + struct TraceCapture(Arc>>>); + + impl Subscriber for TraceCapture { + fn enabled(&self, _: &Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + fn record(&self, _: &Id, _: &Record<'_>) {} + fn record_follows_from(&self, _: &Id, _: &Id) {} + fn enter(&self, _: &Id) {} + fn exit(&self, _: &Id) {} + fn event(&self, event: &Event<'_>) { + if event.metadata().target() != "mst2::native_projection_observation" { + return; + } + struct Fields(BTreeMap); + impl Visit for Fields { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().to_owned(), format!("{value:?}")); + } + fn record_u64(&mut self, field: &Field, value: u64) { + self.0.insert(field.name().to_owned(), value.to_string()); + } + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().to_owned(), value.to_owned()); + } + } + let mut fields = Fields(BTreeMap::new()); + event.record(&mut fields); + self.0.lock().unwrap().push(fields.0); + } + } + + #[test] + fn actual_trace_has_closed_identity_profile_counter_fields_and_no_body_or_credentials() { + let fixture = Fixture::new(); + let commit = fixture.commit.to_string(); + let tree = fixture.tree.to_string(); + let observation = fixture + .source() + .observe( + fixture.resolved(&commit, &tree), + ProjectionWork::default(), + Duration::from_micros(4), + ) + .unwrap(); + let events = Arc::new(Mutex::new(Vec::new())); + tracing::subscriber::with_default(TraceCapture(events.clone()), || observation.emit()); + let events = events.lock().unwrap(); + assert_eq!(events.len(), 1); + let fields = &events[0]; + let expected = [ + "observation_revision", + "phase", + "source_domain", + "request_id", + "instance_id", + "root_commit_oid", + "root_tree_oid", + "native_certificate_receipt_id", + "native_writer_epoch", + "native_publication_sequence", + "scope", + "schema_version", + "metadata_codec", + "materialization_policy", + "fs_semantics", + "access_projection", + "verification_revision", + "projection_revision", + "namespace_view_id", + "snapshot_id", + "metadata_root", + "projection_elapsed_micros", + "page_counter_scope", + "codec_radix_work", + "directories_rebuilt", + "reused_subtree_roots", + "directory_root_pages_built", + "directory_root_page_bytes_built", + "directory_root_pages_reused", + "directory_root_page_bytes_reused", + "directory_entries_scanned", + "scope_path_entries_examined", + "tree_fetches", + "verified_blob_hits", + "verified_blob_misses", + "raw_bytes_fetched", + "raw_bytes_hashed", + "message", + ]; + assert_eq!(fields.len(), expected.len()); + assert!(expected.iter().all(|field| fields.contains_key(*field))); + assert_eq!(fields["phase"], "resolve_directory_projection"); + assert_eq!( + fields["page_counter_scope"], + "returned-directory-root-pages" + ); + assert_eq!(fields["codec_radix_work"], "NOT_EXPOSED"); + assert_eq!(fields["projection_elapsed_micros"], "4"); + assert_eq!(fields["native_certificate_receipt_id"], "97"); + assert_eq!(fields["scope"], "\"/project\""); + assert_eq!(fields["root_tree_oid"], fixture.tree.to_tagged_string()); + } + + #[tokio::test] + async fn emitted_observations_are_operation_local_with_no_publication_aggregation() { + let collect = |request: &'static str, sequence: i64| async move { + let fixture = Fixture::new(); + let commit = fixture.commit.to_string(); + let tree = fixture.tree.to_string(); + let mut resolved = fixture.resolved(&commit, &tree); + resolved.request_id = request; + let source = NativeResolveSource::capture( + INSTANCE, + fixture.commit, + fixture.tree, + Some(97), + 3, + sequence, + ) + .unwrap(); + source + .observe(resolved, ProjectionWork::default(), Duration::ZERO) + .unwrap() + .emit(); + }; + let (left, right) = tokio::join!( + with_observations(collect("request-left", 7)), + with_observations(collect("request-right", 8)) + ); + assert_eq!(left.1.len(), 1); + assert_eq!(right.1.len(), 1); + assert_eq!(left.1[0].request_id, "request-left"); + assert_eq!(right.1[0].request_id, "request-right"); + assert_eq!(left.1[0].source.publication_sequence, 7); + assert_eq!(right.1[0].source.publication_sequence, 8); + } +} diff --git a/src/jupiter/service/native_publication_push_tests.rs b/src/jupiter/service/native_publication_push_tests.rs index f24f591c..2bd0f211 100644 --- a/src/jupiter/service/native_publication_push_tests.rs +++ b/src/jupiter/service/native_publication_push_tests.rs @@ -229,19 +229,32 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d let id=wh03_enqueue_push(&storage,&path,&tip.id.to_string(),&first,&payload).await; assert!(matches!(wh03_exec(&storage,id).await,ExecuteOutcome::Done{..})); let state=api_state(storage.clone()).await; - let (status,v1)=http_resolve(state.clone(),"/").await; + let ((status,v1),first_observations)=crate::ceres::snapshot::projection_observation::with_observations(http_resolve(state.clone(),"/")).await; assert_eq!(status,200,"{v1}"); + assert_eq!(first_observations.len(),1); + let first_identity=first_observations[0].test_identity(); + assert_eq!(first_identity["snapshot_id"],v1["descriptor"]["snapshot_id"]); + assert_eq!(first_identity["metadata_root"],v1["descriptor"]["metadata_root"]); + assert_eq!(first_identity["namespace_view_id"],v1["descriptor"]["namespace_view_id"]); + assert_eq!(first_identity["instance_id"],v1["descriptor"]["instance_id"]); + assert_eq!(first_identity["native_publication_sequence"],v1["publication_sequence"]); + assert_eq!(first_identity["native_writer_epoch"],v1["writer_epoch"]); + assert!(first_identity["native_certificate_receipt_id"].as_u64().unwrap()>0); + assert!(!first_identity["request_id"].as_str().unwrap().is_empty()); let (status,subscope)=http_resolve(state.clone(),&path).await; assert_eq!(status,200,"{subscope}"); assert_eq!(v1["publication_sequence"],subscope["publication_sequence"]); assert_eq!(v1["writer_epoch"],subscope["writer_epoch"]); let root_v1=storage.mono_storage().get_main_ref("/").await.unwrap().unwrap(); - let _ = root_v1; + assert_eq!(first_identity["root_commit_oid"],git_internal::hash::ObjectHash::from_hex_for_kind(git_internal::hash::get_hash_kind(),&root_v1.ref_commit_hash).unwrap().to_tagged_string()); + assert_eq!(first_identity["root_tree_oid"],git_internal::hash::ObjectHash::from_hex_for_kind(git_internal::hash::get_hash_kind(),&root_v1.ref_tree_hash).unwrap().to_tagged_string()); + let native_v1=storage.mono_storage().read_native_publication_head(storage.config().mst2.instance_uuid.as_deref().unwrap()).await.unwrap(); + assert_eq!(first_identity["native_certificate_receipt_id"].as_u64(),Some(native_v1.token.certificate.unwrap() as u64)); let captured=Arc::new(tokio::sync::Barrier::new(2)); let release=Arc::new(tokio::sync::Barrier::new(2)); let task_state=state.clone();let task_captured=captured.clone();let task_release=release.clone(); let mut reader=tokio::spawn(async move { - crate::api::router::snapshot_router::with_native_resolve_barriers(task_captured,task_release,http_resolve(task_state,"/")).await + crate::ceres::snapshot::projection_observation::with_observations(crate::api::router::snapshot_router::with_native_resolve_barriers(task_captured,task_release,http_resolve(task_state,"/"))).await }); tokio::time::timeout(Duration::from_secs(5),captured.wait()).await.unwrap(); let first_commit=storage.mono_storage().get_commit_by_hash(&first).await.unwrap().unwrap(); @@ -251,10 +264,23 @@ async fn http_resolve_captured_before_commit_never_mixes_new_sequence_with_old_d let id=wh03_enqueue_push(&storage,&path,&first,&next,&next_payload).await; assert!(matches!(wh03_exec(&storage,id).await,ExecuteOutcome::Done{..})); tokio::time::timeout(Duration::from_secs(5),release.wait()).await.unwrap(); - let (status,delayed)=tokio::time::timeout(Duration::from_secs(10),&mut reader).await.unwrap().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"]); - let (status,v2)=http_resolve(state,"/").await; + 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"); + } + 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"]); + assert_eq!(next_observations.len(),1); + 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_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); + assert!(failed_observations.is_empty(),"failed projection emitted success observation"); } From b1e1d2104e044ce44d1c676eaf6916f6b9687b2b Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 17:23:40 +0800 Subject: [PATCH 2/2] Recheck exact git-cli identity after concurrent passwd creation --- tests/common/git_cli.rs | 82 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 76 insertions(+), 6 deletions(-) diff --git a/tests/common/git_cli.rs b/tests/common/git_cli.rs index 9f5078a6..b2fa7c2a 100644 --- a/tests/common/git_cli.rs +++ b/tests/common/git_cli.rs @@ -1283,12 +1283,7 @@ pub fn ensure_git_cli_passwd_for_runtime_uid() { "unexpected git-cli id output: {id_pair}" ); - let script = format!( - "if getent passwd {uid} >/dev/null 2>&1; then exit 0; fi; \ - if ! getent group {gid} >/dev/null 2>&1; then addgroup -g {gid} gitcliruntime || true; fi; \ - group_name=\"$(getent group {gid} | cut -d: -f1)\"; \ - adduser -D -u {uid} -G \"$group_name\" -h /home/gitcli -s /bin/sh gitcliruntime" - ); + let script = git_cli_runtime_passwd_script(&uid, &gid); let mut command = docker_exec_base(); command .arg("-u") @@ -1303,6 +1298,81 @@ pub fn ensure_git_cli_passwd_for_runtime_uid() { ); } +fn git_cli_runtime_passwd_script(uid: &str, gid: &str) -> String { + assert!( + !uid.is_empty() + && !gid.is_empty() + && uid.bytes().all(|byte| byte.is_ascii_digit()) + && gid.bytes().all(|byte| byte.is_ascii_digit()), + "git-cli runtime uid/gid must be numeric" + ); + format!( + r#"passwd_matches() {{ + getent passwd {uid} | awk -F: -v uid={uid} -v gid={gid} \ + '$3 == uid && $4 == gid {{ matched++ }} END {{ exit matched != 1 }}' + }} + if getent passwd {uid} >/dev/null 2>&1; then passwd_matches; exit $?; fi + if ! getent group {gid} >/dev/null 2>&1; then addgroup -g {gid} gitcliruntime || true; fi + group_name="$(getent group {gid} | cut -d: -f1)" + test -n "$group_name" || exit 1 + if adduser -D -u {uid} -G "$group_name" -h /home/gitcli -s /bin/sh gitcliruntime; then + passwd_matches + else + status=$? + # Another SSH test can win the check/create race. Accept only the + # exact runtime identity, never an unrelated name or primary GID. + if passwd_matches; then exit 0; fi + exit "$status" + fi"# + ) +} + +#[cfg(target_os = "linux")] +#[test] +fn git_cli_passwd_creation_race_requires_exact_runtime_identity() { + let fixture = r#" + getent() { + if [ "$1" = group ]; then printf 'gitcliruntime:x:1001:\n'; return 0; fi + if [ "$SCENARIO" = existing ]; then printf 'gitcliruntime:x:1001:1001::/home/gitcli:/bin/sh\n'; return 0; fi + if [ "$SCENARIO" = existing_wrong_gid ]; then printf 'gitcliruntime:x:1001:1000::/home/gitcli:/bin/sh\n'; return 0; fi + if [ ! -f "$MOCK_STATE" ]; then return 2; fi + cat "$MOCK_STATE" + } + adduser() { + case "$SCENARIO" in + raced|created) printf 'gitcliruntime:x:1001:1001::/home/gitcli:/bin/sh\n' > "$MOCK_STATE" ;; + wrong_uid) printf 'gitcliruntime:x:1000:1001::/home/gitcli:/bin/sh\n' > "$MOCK_STATE" ;; + wrong_gid|created_wrong_gid) printf 'gitcliruntime:x:1001:1000::/home/gitcli:/bin/sh\n' > "$MOCK_STATE" ;; + esac + if [ "$SCENARIO" = created ] || [ "$SCENARIO" = created_wrong_gid ]; then return 0; fi + return 17 + } + "#; + let temp = tempfile::tempdir().expect("passwd race fixture"); + let script = format!( + "{fixture}\n{}", + git_cli_runtime_passwd_script("1001", "1001") + ); + for (scenario, expected) in [ + ("existing", 0), + ("existing_wrong_gid", 1), + ("raced", 0), + ("created", 0), + ("missing", 17), + ("wrong_uid", 17), + ("wrong_gid", 17), + ("created_wrong_gid", 1), + ] { + let mut command = Command::new("sh"); + command + .args(["-c", &script]) + .env("SCENARIO", scenario) + .env("MOCK_STATE", temp.path().join(scenario)); + let output = output_with_timeout(command, Duration::from_secs(5), "passwd race fixture"); + assert_eq!(output.status.code(), Some(expected), "{scenario}"); + } +} + /// Build ADR-GM-05 `GIT_SSH_COMMAND` for the selected runner (container paths under `/work`). #[allow( dead_code,