Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 62 additions & 4 deletions src/api/router/snapshot_router.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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},
},
Expand Down Expand Up @@ -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(
Expand All @@ -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,
Expand Down Expand Up @@ -498,16 +532,40 @@ 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)?;
let ctx = runtime()
.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,
Expand Down
1 change: 1 addition & 0 deletions src/ceres/snapshot/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
26 changes: 3 additions & 23 deletions src/ceres/snapshot/pages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
}
}
}
Expand Down
Loading
Loading