Prepare complete validated native metadata retention DAGs

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,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,

src/callisto/mod.rs

@@ -58,6 +58,9 @@ pub mod mega_tree;
 pub mod mega_webhook;
 pub mod mega_webhook_delivery;
 pub mod mega_webhook_event_type;
+pub mod mst2_metadata_payload;
+pub mod mst2_metadata_prepare;
+pub mod mst2_metadata_prepare_page;
 pub mod mst2_native_head;
 pub mod mst2_native_publication;
 pub mod mst2_publication;

src/callisto/mst2_metadata_payload.rs

@@ -0,0 +1,16 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "mst2_metadata_payload")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub page_id: Vec<u8>,
+    pub metadata_codec: i16,
+    pub byte_size: i32,
+    pub payload: Vec<u8>,
+    pub created_at: DateTimeWithTimeZone,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+impl ActiveModelBehavior for ActiveModel {}

src/callisto/mst2_metadata_prepare.rs

@@ -0,0 +1,32 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "mst2_metadata_prepare")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub prepare_id: String,
+    pub operation_id: String,
+    pub manifest_digest: Vec<u8>,
+    pub canonical_plan: Vec<u8>,
+    pub source_domain: String,
+    pub tagged_root_tree_oid: String,
+    pub scope: String,
+    pub schema_version: i16,
+    pub metadata_codec: i16,
+    pub materialization_policy: i16,
+    pub fs_semantics: i16,
+    pub access_projection: i16,
+    pub verification_revision: i32,
+    pub projection_revision: i16,
+    pub metadata_root: Vec<u8>,
+    pub node_count: i32,
+    pub edge_count: i32,
+    pub total_bytes: i64,
+    pub state: String,
+    pub created_at: DateTimeWithTimeZone,
+    pub committed_at: Option<DateTimeWithTimeZone>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+impl ActiveModelBehavior for ActiveModel {}

src/callisto/mst2_metadata_prepare_page.rs

@@ -0,0 +1,15 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "mst2_metadata_prepare_page")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub prepare_id: String,
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub page_id: Vec<u8>,
+    pub expected_size: i32,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+impl ActiveModelBehavior for ActiveModel {}