From 195b9f888f6104661e2dbef7f68f8d903f851811 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 7 Oct 2026 18:07:30 +0800 Subject: [PATCH] feat(mst2): persist native HTTP snapshot sessions and guard delivery Use primary PostgreSQL authority for fixed native snapshot sources and per-lease certificates. Atomically hand off installed metadata protection to durable sessions and leases, retain only roots on warm resolve, and retire session pins with the last lease. Recheck current lease, source and captured primary scope before each actual TreeFrame or shared raw block. Add real HTTP/PostgreSQL regressions; native validation and performance remain pending. --- src/api/router/snapshot_content.rs | 44 +- src/api/router/snapshot_content_tests.rs | 69 +- src/api/router/snapshot_router.rs | 320 ++++- src/api/router/snapshot_session_tests.rs | 1121 +++++++++++++++++ src/callisto/mod.rs | 3 + src/callisto/mst2_snapshot_context.rs | 24 + src/callisto/mst2_snapshot_lease.rs | 20 + ...61007_000100_add_mst2_snapshot_sessions.rs | 72 ++ src/jupiter/migration/mod.rs | 5 +- src/jupiter/storage/mod.rs | 5 + src/jupiter/storage/mst2_retention.rs | 47 + .../storage/native_metadata_install.rs | 303 ++++- .../storage/native_publication_storage.rs | 11 +- .../storage/native_snapshot_session.rs | 678 ++++++++++ src/jupiter/tests.rs | 1 + 15 files changed, 2606 insertions(+), 117 deletions(-) create mode 100644 src/api/router/snapshot_session_tests.rs create mode 100644 src/callisto/mst2_snapshot_context.rs create mode 100644 src/callisto/mst2_snapshot_lease.rs create mode 100644 src/jupiter/migration/m20261007_000100_add_mst2_snapshot_sessions.rs create mode 100644 src/jupiter/storage/native_snapshot_session.rs diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index 08158723..d1dd2a15 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -13,7 +13,9 @@ use futures::stream::StreamExt; use serde::Deserialize; use serde_json::json; -use super::{abs_view_path, internal, mst2_error_response, request::Mst2Bytes, treeframe_response}; +use super::{ + abs_view_path, guarded_treeframe_response, internal, mst2_error_response, request::Mst2Bytes, +}; use crate::ceres::snapshot::{ chunks::{ChunkProjection, get_or_project}, error::{SnapshotError, SnapshotErrorCode}, @@ -22,7 +24,6 @@ use crate::ceres::snapshot::{ resolve_abs_metadata, }, resolver::FsKind, - runtime::runtime, view::validate_scope_relative_path, }; @@ -226,9 +227,7 @@ pub(super) async fn blob_head( headers: HeaderMap, ) -> Result { ensure(&state)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; if headers.contains_key("range") { return Err(mst2_error_response(SnapshotError::new( @@ -286,9 +285,7 @@ pub(super) async fn objects( Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure(&state)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; let req: ObjectsRequest = super::parse_json_body(&body)?; if req.items.is_empty() || req.items.len() > OBJECT_MAX_ITEMS { return Err(mst2_error_response(SnapshotError::new( @@ -365,7 +362,7 @@ pub(super) async fn objects( // then exactly one END. Encoding is negotiated per request. use crate::ceres::snapshot::frame_stream::FrameStream; let mut stream = FrameStream::new(1, encoding); - let mut out: Vec = Vec::new(); + let mut out: Vec> = Vec::new(); let mut frame: Vec<([u8; 32], Vec)> = Vec::new(); let mut frame_raw = 0usize; for (cid, data) in unique { @@ -377,7 +374,7 @@ pub(super) async fn objects( let bytes = stream .object(std::mem::take(&mut frame)) .map_err(mst2_error_response)?; - out.extend_from_slice(&bytes); + out.push(bytes); frame_raw = 0; } frame_raw += data.len(); @@ -385,18 +382,17 @@ pub(super) async fn objects( } if !frame.is_empty() { let bytes = stream.object(frame).map_err(mst2_error_response)?; - out.extend_from_slice(&bytes); + out.push(bytes); } - let end = stream.end( req.items.len() as u32, seen.len() as u32, logical_bytes, sha256_of(&body), ); - out.extend_from_slice(&end); + out.push(end); - treeframe_response(&snapshot_id, &body, out).map_err(mst2_error_response) + guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response) } #[derive(Deserialize, Debug)] @@ -467,9 +463,7 @@ pub(super) async fn chunk_map( Query(q): Query, ) -> Result { ensure(&state)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; let handler = state .api_handler(std::path::Path::new("/")) @@ -513,9 +507,7 @@ pub(super) async fn chunk_map_pages( Query(q): Query, ) -> Result { ensure(&state)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; // Canonical v3 uses `map_id` + `page_index`; retain parsing of the old // names only while the legacy client is still present in this checkout. @@ -622,9 +614,7 @@ pub(super) async fn chunks( Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure(&state)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?; let req: ChunksRequest = super::parse_json_body(&body)?; if req.items.is_empty() || req.items.len() > CHUNKS_MAX_ITEMS { return Err(mst2_error_response(SnapshotError::new( @@ -707,7 +697,7 @@ pub(super) async fn chunks( use crate::ceres::snapshot::frame_stream::FrameStream; let mut stream = FrameStream::new(1, encoding); - let mut out: Vec = Vec::new(); + let mut out: Vec> = Vec::new(); for p in planned.iter() { // Re-verified slice (digest + length) from the staged projection. let bytes = p @@ -722,7 +712,7 @@ pub(super) async fn chunks( bytes.to_vec(), ) .map_err(mst2_error_response)?; - out.extend_from_slice(&frame); + out.push(frame); } let end = stream.end( req.items.len() as u32, @@ -730,9 +720,9 @@ pub(super) async fn chunks( logical_bytes, sha256_of(&body), ); - out.extend_from_slice(&end); + out.push(end); - treeframe_response(&snapshot_id, &body, out).map_err(mst2_error_response) + guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response) } /// Strict decimal-string parse for unsigned counts (spec 04 §1: no leading diff --git a/src/api/router/snapshot_content_tests.rs b/src/api/router/snapshot_content_tests.rs index f9643d92..aa697589 100644 --- a/src/api/router/snapshot_content_tests.rs +++ b/src/api/router/snapshot_content_tests.rs @@ -63,7 +63,10 @@ use crate::{ object_storage::{MegaObjectStorageWrapper, build_object_storage}, push_queue_storage::{ClaimOutcome, EnqueueOutcome}, }, - tests::{test_redis_manager, test_storage_with_config, with_test_vault}, + tests::{ + TestSchemaGuard, test_db_config, test_redis_manager, test_storage_with_config, + with_test_vault, + }, }, orbit_api::{ error::OrbitResult, @@ -76,6 +79,9 @@ use crate::{ const TOKEN: &str = "mst2-fixed-content-test"; +#[path = "snapshot_session_tests.rs"] +mod durable_sessions; + #[derive(Default)] struct ReadCounts { whole: AtomicUsize, @@ -214,6 +220,7 @@ struct Fixture { digest: [u8; 32], counts: Arc, _temp: tempfile::TempDir, + _schema: Option, } fn tree(items: Vec) -> Tree { @@ -301,6 +308,14 @@ async fn publish_native_push( impl Fixture { async fn new() -> Self { + Self::new_with_pg_config(false).await + } + + async fn new_with_pg_config(rebuildable: bool) -> Self { + Self::new_with_pg_config_and_directories(rebuildable, 0).await + } + + async fn new_with_pg_config_and_directories(rebuildable: bool, directory_count: usize) -> Self { let temp = tempfile::tempdir().unwrap(); let mut config = isolated_config(temp.path().join("config")); config.monorepo.push_policy = PushPolicy::Trunk; @@ -310,7 +325,25 @@ impl Fixture { config.mst2.auth_token = Some(TOKEN.to_string()); let backend = build_object_storage(&config.object_storage).await.unwrap(); let counts = Arc::new(ReadCounts::default()); - let mut storage = test_storage_with_config(temp.path(), config).await; + let (mut storage, schema) = if rebuildable { + let (database, schema) = test_db_config(temp.path()).await; + config.database = database; + let connection = crate::jupiter::storage::init::database_connection(&config.database) + .await + .unwrap(); + ( + crate::jupiter::storage::Storage::new_with_connection( + Arc::new(config), + Arc::new(connection), + backend.clone(), + ) + .await + .unwrap(), + Some(schema), + ) + } else { + (test_storage_with_config(temp.path(), config).await, None) + }; storage.git_service = GitService { obj_storage: MegaObjectStorageWrapper::new(Arc::new(CountingStorage { inner: backend, @@ -344,7 +377,7 @@ impl Fixture { let empty_oid = ObjectHash::from_hex_for_kind(HashKind::Sha1, &empty_oid).unwrap(); let empty_dir = tree(vec![]); let nested = tree(vec![item(TreeItemMode::Blob, blob_oid, "file")]); - let project = tree(vec![ + let mut project_items = vec![ item(TreeItemMode::Blob, blob_oid, "alias"), item(TreeItemMode::Tree, empty_dir.id, "directory"), item(TreeItemMode::Blob, empty_oid, "empty"), @@ -352,7 +385,24 @@ impl Fixture { item(TreeItemMode::Blob, blob_oid, "file"), item(TreeItemMode::Link, link_oid, "link"), item(TreeItemMode::Tree, nested.id, "nested"), - ]); + ]; + let mut extra_trees = Vec::new(); + for index in 0..directory_count { + // Distinct names make distinct canonical pages even though all + // directories share the same already-verified immutable blob. + let child = tree(vec![item( + TreeItemMode::Blob, + blob_oid, + &format!("file-{index:03}"), + )]); + project_items.push(item( + TreeItemMode::Tree, + child.id, + &format!("wide-{index:03}"), + )); + extra_trees.push(child); + } + let project = tree(project_items); let old_tip = Commit::from_tree_id_with_kind( HashKind::Sha1, project.id, @@ -379,13 +429,9 @@ impl Fixture { ) .unwrap(); let mono = storage.mono_storage(); - mono.save_mega_trees( - vec![empty_dir, nested, project, root.clone()], - commit.id, - None, - ) - .await - .unwrap(); + let mut trees = vec![empty_dir, nested, project, root.clone()]; + trees.extend(extra_trees); + mono.save_mega_trees(trees, commit.id, None).await.unwrap(); mono.save_mega_commits(vec![commit.clone(), old_tip.clone(), new_tip.clone()], None) .await .unwrap(); @@ -474,6 +520,7 @@ impl Fixture { raw, counts, _temp: temp, + _schema: schema, } } diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 3cfa4963..4c2604d2 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -101,6 +101,14 @@ pub(crate) fn treeframe_response( snapshot_id: &str, request_body: &[u8], body: Vec, +) -> Result { + treeframe_response_body(snapshot_id, request_body, axum::body::Body::from(body)) +} + +fn treeframe_response_body( + snapshot_id: &str, + request_body: &[u8], + body: axum::body::Body, ) -> Result { let request_digest: [u8; 32] = Sha256::digest(request_body).into(); Response::builder() @@ -112,20 +120,58 @@ pub(crate) fn treeframe_response( ) .header("cache-control", "private, no-cache, no-transform") .header("vary", "Authorization, Accept") - .body(axum::body::Body::from(Bytes::from(body))) + .body(body) .map_err(|e| internal(format!("TreeFrame response build failed: {e}"))) } +fn guarded_treeframe_response( + state: &MonoApiServiceState, + context: &crate::ceres::snapshot::runtime::SnapshotContext, + snapshot_id: &str, + request_body: &[u8], + frames: Vec>, +) -> Result { + let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "request authentication context missing", + ) + })?; + let units = frames + .into_iter() + .map(Bytes::from) + .collect::>(); + let stream = futures::stream::unfold( + (units, state.clone(), context.clone(), headers), + |(mut units, state, context, headers)| async move { + let unit = units.pop_front()?; + if let Err(error) = revalidate_access(&state, &context, &headers).await { + units.clear(); + return Some((Err(error), (units, state, context, headers))); + } + Some((Ok(unit), (units, state, context, headers))) + }, + ); + treeframe_response_body( + snapshot_id, + request_body, + axum::body::Body::from_stream(stream), + ) +} + tokio::task_local! { /// Per-request id for the error envelope (spec 14 §5). Sourced from the /// global `TraceContext` so the envelope, the `X-Request-Id` response /// header and log spans all carry the same id. static REQUEST_ID: String; + static REQUEST_CONTEXT: Option; + static REQUEST_HEADERS: HeaderMap; } #[cfg(test)] tokio::task_local! { static NATIVE_RESOLVE_BARRIERS: (std::sync::Arc, std::sync::Arc); + static NATIVE_HANDOFF_BARRIERS: (std::sync::Arc, std::sync::Arc); static REJECT_NATIVE_OBSERVATION_SOURCE: bool; } @@ -147,6 +193,17 @@ pub(crate) async fn with_native_resolve_barriers( .await } +#[cfg(test)] +pub(crate) async fn with_native_handoff_barriers( + prepared: std::sync::Arc, + release: std::sync::Arc, + future: F, +) -> F::Output { + NATIVE_HANDOFF_BARRIERS + .scope((prepared, release), future) + .await +} + /// The request id of the in-flight request, for error envelopes. pub(crate) fn current_request_id() -> String { REQUEST_ID.try_with(|id| id.clone()).unwrap_or_default() @@ -158,7 +215,7 @@ const LEASE_HEADER: &str = "x-mega-snapshot-lease"; /// every endpoint except `capabilities` requires `Authorization: Bearer /// ` when the deployment configured one, and snapshot-bound endpoints /// must additionally present the lease they resolved -/// (`X-Mega-Snapshot-Lease`), validated against the in-memory lease table — +/// (`X-Mega-Snapshot-Lease`), validated against the session authority — /// knowing the snapshot id alone is not a capability. async fn snapshot_auth_middleware( State(state): State, @@ -178,10 +235,41 @@ async fn snapshot_auth_middleware( }); let mut response = REQUEST_ID .scope(id.to_string(), async { - if let Some(res) = auth_error(&state, req.headers(), req.uri().path()) { - return res; - } - next.run(req).await + let headers = req.headers().clone(); + let path = req.uri().path().to_owned(); + let context = match authenticate_request(&state, &headers, &path).await { + Ok(context) => context, + Err(response) => return response, + }; + REQUEST_HEADERS + .scope( + headers.clone(), + REQUEST_CONTEXT.scope(context.clone(), async { + let response = next.run(req).await; + if response.status().is_success() + || response.status() == StatusCode::NOT_MODIFIED + { + match context { + Some(context) => { + if let Err(error) = + revalidate_access(&state, &context, &headers).await + { + return mst2_error_response(error); + } + } + None => { + if let Err(response) = + authenticate_request(&state, &headers, &path).await + { + return response; + } + } + } + } + response + }), + ) + .await }) .await; if let Ok(value) = HeaderValue::from_str(&id) { @@ -245,7 +333,11 @@ fn unauthenticated(message: &'static str) -> Response { /// Auth decision for one request path (see [`snapshot_auth_middleware`]). /// `capabilities` stays open; lease routes need only the bearer; a /// snapshot-bound route (`/snapshots/sha256:…/…`) also needs its lease. -fn auth_error(state: &MonoApiServiceState, headers: &HeaderMap, path: &str) -> Option { +async fn authenticate_request( + state: &MonoApiServiceState, + headers: &HeaderMap, + path: &str, +) -> Result, Response> { let config = state.storage.config(); let token = config.mst2.auth_token.as_deref(); let path = path.split('?').next().unwrap_or(path); @@ -253,32 +345,105 @@ fn auth_error(state: &MonoApiServiceState, headers: &HeaderMap, path: &str) -> O // middleware runs — accept both the stripped and full forms. let rest = path .strip_prefix("/api/v2/snapshots/") - .or_else(|| path.strip_prefix("/snapshots/"))?; + .or_else(|| path.strip_prefix("/snapshots/")); + let Some(rest) = rest else { + return Ok(None); + }; if rest == "capabilities" { - return None; + return Ok(None); } if let Some(token) = token && !bearer_ok(headers, token) { - return Some(unauthenticated("missing or invalid bearer credentials")); + return Err(unauthenticated("missing or invalid bearer credentials")); } let mut segments = rest.split('/'); match segments.next() { // Lease management names the lease in the path, not a snapshot. - Some("resolve") | Some("leases") | Some("capabilities") | None => None, + Some("resolve") | Some("leases") | Some("capabilities") | None => Ok(None), Some(snapshot_id) => { let lease = headers.get(LEASE_HEADER).and_then(|v| v.to_str().ok()); match lease { - None => Some(unauthenticated("missing X-Mega-Snapshot-Lease header")), - Some(lease) => runtime() - .validate_lease(snapshot_id, lease) - .err() - .map(mst2_error_response), + None => Err(unauthenticated("missing X-Mega-Snapshot-Lease header")), + Some(lease) => state + .storage + .snapshot_context(snapshot_id, lease) + .await + .map(Some) + .map_err(mst2_error_response), } } } } +fn request_context( + state: &MonoApiServiceState, + snapshot_id: &str, +) -> Result { + if let Ok(Some(context)) = REQUEST_CONTEXT.try_with(Clone::clone) + && context.built.snapshot_id == snapshot_id + { + return Ok(context); + } + if !state.storage.config().mst2.publication_enabled { + return runtime().context(snapshot_id); + } + Err(SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "validated snapshot session missing", + )) +} + +async fn revalidate_access( + state: &MonoApiServiceState, + context: &crate::ceres::snapshot::runtime::SnapshotContext, + headers: &HeaderMap, +) -> Result<(), SnapshotError> { + let config = state.storage.config(); + if !config.mst2.enabled { + return Err(SnapshotError::new( + SnapshotErrorCode::SnapshotNotReady, + "snapshot surface disabled", + )); + } + if let Some(token) = config.mst2.auth_token.as_deref() + && !bearer_ok(headers, token) + { + return Err(SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "bearer credentials changed", + )); + } + let current = state + .storage + .snapshot_context(&context.built.snapshot_id, &context.lease_id) + .await?; + if current.built.descriptor != context.built.descriptor + || current.commit_oid != context.commit_oid + || current.root_tree_oid != context.root_tree_oid + || current.authorization_epoch != context.authorization_epoch + { + return Err(SnapshotError::new( + SnapshotErrorCode::IntegrityError, + "fixed session changed during request", + )); + } + Ok(()) +} + +async fn revalidate_request( + state: &MonoApiServiceState, + context: &crate::ceres::snapshot::runtime::SnapshotContext, +) -> Result<(), SnapshotError> { + let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { + SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "request authentication context missing", + ) + })?; + revalidate_access(state, context, &headers).await +} + fn mst2_error_response(err: SnapshotError) -> Response { let status = StatusCode::from_u16(err.code.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); @@ -451,6 +616,7 @@ async fn resolve( let config = state.storage.config(); let mut native_source = None; + let mut selected_native_head = 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( @@ -511,6 +677,7 @@ async fn resolve( None } }; + selected_native_head = Some(head.clone()); ( head.root.commit, head.root.tree, @@ -576,9 +743,46 @@ async fn resolve( 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)?; + let ctx = if let Some(head) = selected_native_head.as_ref() { + let sessions = state.storage.snapshot_sessions().await; + if let Some(context) = sessions + .open(head, &built, None, req.lease_seconds) + .await + .map_err(mst2_error_response)? + { + context + } else { + let prepared = crate::ceres::snapshot::pages::prepare_native_metadata_retention( + handler.as_ref(), + &root_tree, + &req.scope, + crate::ceres::snapshot::retention_dag::MetadataDagLimits::default(), + ) + .await + .map_err(mst2_error_response)?; + let receipt = sessions + .install(&built, &prepared) + .await + .map_err(mst2_error_response)?; + #[cfg(test)] + if let Ok((prepared, release)) = NATIVE_HANDOFF_BARRIERS.try_with(|value| value.clone()) + { + prepared.wait().await; + release.wait().await; + } + sessions + .open(head, &built, Some(&receipt), req.lease_seconds) + .await + .map_err(mst2_error_response)? + .ok_or_else(|| { + mst2_error_response(internal("durable session handoff returned no context")) + })? + } + } else { + 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(); @@ -611,6 +815,9 @@ async fn resolve( } } + revalidate_request(&state, &ctx) + .await + .map_err(mst2_error_response)?; let body = json!({ "descriptor": descriptor_json(&built), "publication_sequence": sequence, @@ -648,9 +855,7 @@ async fn descriptor_get( AxumPath(snapshot_id): AxumPath, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; // Reading the descriptor back never touches latest (spec 04 §2). let body = json!({ "snapshot_id": snapshot_id, @@ -680,8 +885,10 @@ async fn lease_renew( } else { parse_json_body(&body)? }; - let renewed = runtime() - .renew_lease(&lease_id, req.lease_seconds.unwrap_or(600)) + let renewed = state + .storage + .snapshot_renew(&lease_id, req.lease_seconds.unwrap_or(600)) + .await .map_err(mst2_error_response)?; // Renewal never changes the version or authorization (spec 04 §4). let body = json!({ @@ -701,7 +908,11 @@ async fn lease_release( ensure_enabled(&state).map_err(mst2_error_response)?; // Idempotent: releasing an unknown/already-released lease still succeeds // (spec 04 §2). This never deletes Git content. - let released = runtime().release_lease(&lease_id); + let released = state + .storage + .snapshot_release(&lease_id) + .await + .map_err(mst2_error_response)?; Ok(Json(json!({ "lease_id": lease_id, "released": released })).into_response()) } @@ -732,9 +943,7 @@ async fn directory( Query(q): Query, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; if !(1..=256).contains(&q.limit) { return Err(mst2_error_response(SnapshotError::new( @@ -939,9 +1148,7 @@ async fn blob( headers: HeaderMap, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; validate_scope_relative_path(&q.path).map_err(mst2_error_response)?; if headers.contains_key("range") { // Spec 04 section 9: raw blob has no Range semantics this profile. @@ -985,11 +1192,42 @@ async fn blob( crate::ceres::snapshot::resolver::FsKind::Symlink => "symlink", crate::ceres::snapshot::resolver::FsKind::Directory => "directory", }; + revalidate_request(&state, &ctx) + .await + .map_err(mst2_error_response)?; + let headers = REQUEST_HEADERS.try_with(Clone::clone).map_err(|_| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::Unauthenticated, + "request authentication context missing", + )) + })?; + let stream_state = state.0.clone(); + let stream_context = ctx.clone(); + let blocks = futures::stream::unfold( + ( + Bytes::from(raw), + 0usize, + stream_state, + stream_context, + headers, + ), + |(raw, offset, state, context, headers)| async move { + if offset >= raw.len() { + return None; + } + if let Err(error) = revalidate_access(&state, &context, &headers).await { + return Some((Err(error), (raw, usize::MAX, state, context, headers))); + } + let end = (offset + 1_048_576).min(raw.len()); + let bytes = raw.slice(offset..end); + Some((Ok(bytes), (raw, end, state, context, headers))) + }, + ); Response::builder() .header("etag", format!("\"sha256:{}\"", hex_of(&digest))) .header("cache-control", "private, no-cache, no-transform") .header("x-mega-fs-kind", fs_kind_str) - .body(axum::body::Body::from(Bytes::from(raw))) + .body(axum::body::Body::from_stream(blocks)) .map_err(|e| { mst2_error_response(SnapshotError::new( SnapshotErrorCode::Internal, @@ -1053,9 +1291,7 @@ async fn lookup( .api_handler(std::path::Path::new("/")) .await .map_err(internal)?; - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; let root_tree = handler .get_tree_by_hash(&ctx.root_tree_oid) .await @@ -1208,9 +1444,7 @@ async fn metadata_pages( for item in &req.items { validate_scope_relative_path(&item.directory_path).map_err(mst2_error_response)?; } - let ctx = runtime() - .context(&snapshot_id) - .map_err(mst2_error_response)?; + let ctx = request_context(&state, &snapshot_id).map_err(mst2_error_response)?; let handler = state .api_handler(std::path::Path::new("/")) @@ -1281,18 +1515,18 @@ async fn metadata_pages( .unwrap_or(crate::ceres::snapshot::frame_stream::Encoding::Identity); use crate::ceres::snapshot::frame_stream::FrameStream; let mut stream = FrameStream::new(1, encoding); - let mut out: Vec = Vec::new(); + let mut out: Vec> = Vec::new(); let mut frame: Vec<([u8; 32], Vec)> = Vec::new(); let mut frame_raw: usize = 0; let mut flush = |frame: &mut Vec<([u8; 32], Vec)>, raw: &mut usize, - out: &mut Vec| + out: &mut Vec>| -> Result<(), SnapshotError> { if frame.is_empty() { return Ok(()); } let bytes = stream.meta(std::mem::take(frame))?; - out.extend_from_slice(&bytes); + out.push(bytes); *raw = 0; Ok(()) }; @@ -1318,9 +1552,9 @@ async fn metadata_pages( logical_bytes, request_body_sha256, ); - out.extend_from_slice(&end); + out.push(end); - treeframe_response(&snapshot_id, &body, out).map_err(mst2_error_response) + guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response) } #[cfg(test)] diff --git a/src/api/router/snapshot_session_tests.rs b/src/api/router/snapshot_session_tests.rs new file mode 100644 index 00000000..43a9c440 --- /dev/null +++ b/src/api/router/snapshot_session_tests.rs @@ -0,0 +1,1121 @@ +use sea_orm::{DatabaseConnection, DbBackend, Statement, TransactionTrait}; +use tokio::sync::Barrier; + +use super::*; +use crate::{ + api::router::snapshot_router::{with_native_handoff_barriers, with_native_resolve_barriers}, + jupiter::storage::{ + mst2_retention::{GcClaim, PostgresRetentionRepository}, + push_queue_storage::PushQueueStorage, + }, +}; + +fn resolve_request(scope: &str) -> Request { + Request::builder() + .method("POST") + .uri("/api/v2/snapshots/resolve") + .header("authorization", format!("Bearer {TOKEN}")) + .header("content-type", "application/json") + .body(Body::from( + json!({"target":{"kind":"latest"},"scope":scope}).to_string(), + )) + .unwrap() +} + +async fn scalar(db: &DatabaseConnection, sql: &str) -> i64 { + db.query_one_raw(Statement::from_string(DbBackend::Postgres, sql.to_owned())) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap() +} + +async fn rebuilt(fixture: &Fixture) -> MonoApiServiceState { + let config = fixture.state.storage.config(); + let connection = crate::jupiter::storage::init::postgres_connection(&config.database) + .await + .unwrap(); + let storage = crate::jupiter::storage::Storage::new_with_connection( + config, + Arc::new(connection), + fixture.state.storage.git_service.obj_storage.clone(), + ) + .await + .unwrap(); + MonoApiServiceState { + storage, + git_object_cache: Arc::new(GitObjectCache { + connection: fixture.state.git_object_cache.connection.clone(), + prefix: uuid::Uuid::new_v4().to_string(), + }), + ..fixture.state.clone() + } +} + +fn app(state: &MonoApiServiceState) -> Router { + Router::new().nest("/api/v2", routers(state.clone()).with_state(state.clone())) +} + +async fn lease_control(fixture: &Fixture, lease: &str, method: &str, renew: bool) -> Value { + let suffix = if renew { "/renew" } else { "" }; + success_json( + fixture + .app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(format!("/api/v2/snapshots/leases/{lease}{suffix}")) + .header("authorization", format!("Bearer {TOKEN}")) + .header("content-type", "application/json") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(), + ) + .await +} + +async fn advance(fixture: &Fixture) { + let mono = fixture.state.storage.mono_storage(); + let old = mono.get_main_ref("/project").await.unwrap().unwrap(); + let old_oid = ObjectHash::from_hex_for_kind(HashKind::Sha1, &old.ref_commit_hash).unwrap(); + let oid = ObjectHash::from_hex_for_kind(HashKind::Sha1, &fixture.oid).unwrap(); + let next_tree = tree(vec![item(TreeItemMode::Blob, oid, "new-only")]); + let next = Commit::from_tree_id_with_kind( + HashKind::Sha1, + next_tree.id, + vec![old_oid], + "durable HTTP next view", + ) + .unwrap(); + mono.save_mega_trees(vec![next_tree], next.id, None) + .await + .unwrap(); + mono.save_mega_commits(vec![next.clone()], None) + .await + .unwrap(); + publish_native_push(&fixture.state.storage, "/project", old_oid, &next).await; +} + +#[tokio::test] +async fn mst2_durable_http_resolve_installs_complete_dag_and_warm_leases_share_it() { + let fixture = Fixture::new_with_pg_config(true).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + let pages = scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await; + assert!( + pages >= 3, + "scope includes nested and empty directory pages" + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_metadata_prepare_page").await, + pages + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_context").await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_snapshot_lease WHERE state='ACTIVE'" + ) + .await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + 0 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='lease'" + ) + .await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='pin'" + ) + .await, + 1 + ); + let edges = scalar(db, "SELECT count(*) FROM mst2_retention_edge").await; + let refs = scalar( + db, + "SELECT sum(incoming_refs)::bigint FROM mst2_retention_node", + ) + .await; + assert_eq!(edges, refs); + fixture.counts.reset(); + let again = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project")) + .await + .unwrap(), + ) + .await; + assert_eq!(again["descriptor"]["snapshot_id"], fixture.snapshot); + assert_ne!(again["lease_id"], fixture.lease); + fixture.counts.assert(0, 0); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await, + pages + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_metadata_prepare").await, + 1 + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_context").await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='lease'" + ) + .await, + 2 + ); + assert_eq!( + scalar( + db, + "SELECT sum(incoming_refs)::bigint FROM mst2_retention_node" + ) + .await, + refs + ); + + 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 replacement = app(&state); + let descriptor = success_json( + replacement + .clone() + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + ) + .await; + assert_eq!(descriptor["snapshot_id"], fixture.snapshot); + assert_eq!(descriptor["lease_id"], fixture.lease); + assert_eq!(descriptor["descriptor"], again["descriptor"]); + for (method, path, body) in [ + ("GET", "directory?path=/nested", Body::empty()), + ( + "POST", + "lookup", + Body::from(json!({"paths":["/nested/file"]}).to_string()), + ), + ( + "POST", + "metadata/pages", + Body::from(json!({"items":[{"directory_path":"/nested"}]}).to_string()), + ), + ("HEAD", "blob?path=/file", Body::empty()), + ] { + let response = replacement + .clone() + .oneshot(fixture.request(method, path, body)) + .await + .unwrap(); + assert_eq!(response.status(), 200); + to_bytes(response.into_body(), usize::MAX).await.unwrap(); + } + fixture.counts.assert(0, 0); + let response = replacement + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + assert_eq!( + to_bytes(response.into_body(), usize::MAX) + .await + .unwrap() + .as_ref(), + fixture.raw.as_slice() + ); + fixture.counts.assert(1, fixture.raw.len()); +} + +#[tokio::test] +async fn mst2_durable_http_old_sid_and_original_lease_survive_real_publication_and_fresh_service() { + let fixture = Fixture::new_with_pg_config(true).await; + let old = success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + advance(&fixture).await; + let next = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project")) + .await + .unwrap(), + ) + .await; + assert_eq!(next["publication_sequence"], "2"); + assert_ne!(next["descriptor"]["snapshot_id"], fixture.snapshot); + let state = rebuilt(&fixture).await; + let replacement = app(&state); + let restored = success_json( + replacement + .clone() + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + ) + .await; + assert_eq!(restored, old); + let response = replacement + .clone() + .oneshot(fixture.request("GET", "blob?path=/file", Body::empty())) + .await + .unwrap(); + assert_eq!(response.status(), 200); + assert_eq!( + to_bytes(response.into_body(), usize::MAX) + .await + .unwrap() + .as_ref(), + fixture.raw.as_slice() + ); + error( + replacement + .oneshot(fixture.request("GET", "chunk-map?path=/new-only", Body::empty())) + .await + .unwrap(), + 404, + "PATH_NOT_FOUND", + false, + ) + .await; + let new_request = Request::builder() + .uri(format!( + "/api/v2/snapshots/{}/blob?path=/new-only", + next["descriptor"]["snapshot_id"].as_str().unwrap() + )) + .header("authorization", format!("Bearer {TOKEN}")) + .header("x-mega-snapshot-lease", next["lease_id"].as_str().unwrap()) + .body(Body::empty()) + .unwrap(); + let response = app(&state).oneshot(new_request).await.unwrap(); + assert_eq!(response.status(), 200); + assert_eq!( + to_bytes(response.into_body(), usize::MAX) + .await + .unwrap() + .as_ref(), + fixture.raw.as_slice() + ); +} + +#[tokio::test] +async fn mst2_durable_http_renew_release_expiry_and_last_lease_retire_protection() { + let fixture = Fixture::new_with_pg_config(true).await; + let another = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project")) + .await + .unwrap(), + ) + .await; + let other = another["lease_id"].as_str().unwrap(); + let renewed = lease_control(&fixture, &fixture.lease, "POST", true).await; + assert_eq!(renewed["snapshot_id"], fixture.snapshot); + let state = rebuilt(&fixture).await; + let restored = success_json( + app(&state) + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + ) + .await; + assert_eq!(restored["lease_expires_at"], renewed["lease_expires_at"]); + let first = lease_control(&fixture, &fixture.lease, "DELETE", false).await; + let second = lease_control(&fixture, &fixture.lease, "DELETE", false).await; + assert_eq!(first["released"], true); + assert_eq!(second["released"], false); + error( + app(&state) + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + 410, + "LEASE_EXPIRED", + false, + ) + .await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='lease'" + ) + .await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='pin'" + ) + .await, + 1 + ); + let mut request = fixture.request("HEAD", "blob?path=/file", Body::empty()); + request + .headers_mut() + .insert("x-mega-snapshot-lease", other.parse().unwrap()); + assert_eq!(app(&state).oneshot(request).await.unwrap().status(), 200); + db.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE mst2_snapshot_lease SET expires_at_unix=0 WHERE lease_id=$1", + [other.into()], + )) + .await + .unwrap(); + let renew = Request::builder() + .method("POST") + .uri(format!("/api/v2/snapshots/leases/{other}/renew")) + .header("authorization", format!("Bearer {TOKEN}")) + .body(Body::empty()) + .unwrap(); + error( + app(&state).oneshot(renew).await.unwrap(), + 410, + "LEASE_EXPIRED", + false, + ) + .await; + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_snapshot_lease WHERE state='EXPIRED'" + ) + .await, + 1 + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); + let stored = + crate::callisto::mst2_snapshot_context::Entity::find_by_id(fixture.snapshot.clone()) + .one(db) + .await + .unwrap() + .unwrap(); + let node = format!("page:sha256:{}", hex::encode(stored.metadata_root)); + assert_eq!( + PostgresRetentionRepository::new(db.clone()) + .mark_deleting("last-lease-gc", &node) + .await + .unwrap(), + GcClaim::Marked + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await, + scalar(db, "SELECT count(*) FROM mst2_metadata_prepare_page").await, + "release does not delete payload or Git bytes" + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_live_to_deleting_winner_rejects_paused_resolve_without_leak() { + let fixture = Fixture::new_with_pg_config(true).await; + let captured = Arc::new(Barrier::new(2)); + let release = Arc::new(Barrier::new(2)); + let resolving = { + let app = fixture.app.clone(); + let captured = captured.clone(); + let release = release.clone(); + tokio::spawn(async move { + with_native_resolve_barriers( + captured, + release, + app.oneshot(resolve_request("/project")), + ) + .await + .unwrap() + }) + }; + captured.wait().await; + lease_control(&fixture, &fixture.lease, "DELETE", false).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + let context = + crate::callisto::mst2_snapshot_context::Entity::find_by_id(fixture.snapshot.clone()) + .one(db) + .await + .unwrap() + .unwrap(); + let node = format!("page:sha256:{}", hex::encode(context.metadata_root)); + let txn = db.begin().await.unwrap(); + assert_eq!( + PostgresRetentionRepository::mark_deleting_in_txn(&txn, "http-gc-winner", &node) + .await + .unwrap(), + GcClaim::Marked + ); + let txn_id = scalar(db, "SELECT count(*) FROM mst2_snapshot_lease").await; + release.wait().await; + txn.commit().await.unwrap(); + error(resolving.await.unwrap(), 503, "OBJECT_UNAVAILABLE", false).await; + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_lease").await, + txn_id + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); +} + +#[tokio::test] +async fn mst2_durable_http_lease_winner_blocks_gc_until_the_last_release() { + let fixture = Fixture::new_with_pg_config(true).await; + let another = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project")) + .await + .unwrap(), + ) + .await; + let other = another["lease_id"].as_str().unwrap(); + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + let stored = + crate::callisto::mst2_snapshot_context::Entity::find_by_id(fixture.snapshot.clone()) + .one(db) + .await + .unwrap() + .unwrap(); + let node = format!("page:sha256:{}", hex::encode(stored.metadata_root)); + let retention = PostgresRetentionRepository::new(db.clone()); + assert_eq!( + retention + .mark_deleting("lease-winner-first", &node) + .await + .unwrap(), + GcClaim::Unavailable + ); + lease_control(&fixture, &fixture.lease, "DELETE", false).await; + assert_eq!( + retention + .mark_deleting("lease-winner-second", &node) + .await + .unwrap(), + GcClaim::Unavailable + ); + lease_control(&fixture, other, "DELETE", false).await; + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); + assert_eq!( + retention + .mark_deleting("lease-winner-final", &node) + .await + .unwrap(), + GcClaim::Marked + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_renew_waits_for_lock_before_checking_database_deadline() { + let fixture = Fixture::new_with_pg_config(true).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + let held = db.begin().await.unwrap(); + held.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT pg_advisory_xact_lock($1,hashtext(current_schema()))", + [crate::jupiter::storage::mst2_retention::RETENTION_LOCK_KEY.into()], + )) + .await + .unwrap(); + let renewing = { + let app = fixture.app.clone(); + let lease = fixture.lease.clone(); + tokio::spawn(async move { + app.oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/v2/snapshots/leases/{lease}/renew")) + .header("authorization", format!("Bearer {TOKEN}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap() + }) + }; + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let blocked = scalar( + db, + "SELECT count(*) FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid + WHERE l.locktype='advisory' AND NOT l.granted AND a.datname=current_database() + AND a.application_name=current_schema()", + ) + .await; + if blocked > 0 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("HTTP renewal did not wait on the retention lock"); + held.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE mst2_snapshot_lease SET expires_at_unix=0 WHERE lease_id=$1", + [fixture.lease.clone().into()], + )) + .await + .unwrap(); + held.commit().await.unwrap(); + error(renewing.await.unwrap(), 410, "LEASE_EXPIRED", false).await; + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_snapshot_lease WHERE state='EXPIRED'" + ) + .await, + 1 + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_retention_root").await, + 0 + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_publication_advance_rejects_mixed_resolve_then_retries_current() { + let fixture = Fixture::new_with_pg_config(true).await; + let captured = Arc::new(Barrier::new(2)); + let release = Arc::new(Barrier::new(2)); + let resolving = { + let app = fixture.app.clone(); + let captured = captured.clone(); + let release = release.clone(); + tokio::spawn(async move { + with_native_resolve_barriers( + captured, + release, + app.oneshot(resolve_request("/project")), + ) + .await + .unwrap() + }) + }; + captured.wait().await; + advance(&fixture).await; + release.wait().await; + error(resolving.await.unwrap(), 503, "SNAPSHOT_NOT_READY", true).await; + let mono = fixture.state.storage.mono_storage(); + assert_eq!( + scalar( + mono.get_connection(), + "SELECT count(*) FROM mst2_snapshot_lease" + ) + .await, + 1 + ); + let next = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project")) + .await + .unwrap(), + ) + .await; + assert_eq!(next["publication_sequence"], "2"); + assert_ne!(next["descriptor"]["snapshot_id"], fixture.snapshot); + let state = rebuilt(&fixture).await; + assert_eq!( + app(&state) + .oneshot(fixture.request("HEAD", "blob?path=/file", Body::empty())) + .await + .unwrap() + .status(), + 200 + ); +} + +#[tokio::test] +async fn mst2_durable_http_install_fault_never_hands_off_or_returns_success() { + let fixture = Fixture::new_with_pg_config(true).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + db.execute_unprepared( + "CREATE FUNCTION reject_http_page() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'forced HTTP install interruption'; END $$; + CREATE TRIGGER reject_http_page BEFORE INSERT ON mst2_metadata_payload + FOR EACH ROW EXECUTE FUNCTION reject_http_page() ", + ) + .await + .unwrap(); + error( + fixture + .app + .clone() + .oneshot(resolve_request("/project/nested")) + .await + .unwrap(), + 503, + "TEMPORARY_UNAVAILABLE", + true, + ) + .await; + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_context").await, + 1 + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_lease").await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await, + 0 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_metadata_prepare WHERE state='PREPARING'" + ) + .await, + 1 + ); + assert_eq!( + fixture + .send("HEAD", "blob?path=/file", Body::empty()) + .await + .status(), + 200 + ); + db.execute_unprepared( + "DROP TRIGGER reject_http_page ON mst2_metadata_payload; DROP FUNCTION reject_http_page() ", + ) + .await + .unwrap(); + let success = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/project/nested")) + .await + .unwrap(), + ) + .await; + assert_eq!(success["descriptor"]["scope"], "/project/nested"); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_metadata_prepare WHERE state='PREPARING'" + ) + .await, + 0 + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_frame_and_raw_delivery_recheck_after_release() { + for raw in [false, true] { + let fixture = Fixture::new_with_pg_config(true).await; + let response = if raw { + fixture.send("GET", "blob?path=/file", Body::empty()).await + } else { + fixture + .send( + "POST", + "metadata/pages", + Body::from(json!({"items":[{"directory_path":"/"}]}).to_string()), + ) + .await + }; + assert_eq!(response.status(), 200); + let mut stream = response.into_body().into_data_stream(); + let first = stream.next().await.unwrap().unwrap(); + assert!(!first.is_empty()); + if raw { + assert_eq!(first.len(), 1_048_576); + } + lease_control(&fixture, &fixture.lease, "DELETE", false).await; + assert!( + stream.next().await.unwrap().is_err(), + "revocation must suppress the next raw block or END frame" + ); + assert!(stream.next().await.is_none()); + fixture.counts.assert( + if raw { 1 } else { 0 }, + if raw { fixture.raw.len() } else { 0 }, + ); + } +} + +#[tokio::test] +async fn mst2_durable_http_current_state_wrong_lease_and_corrupt_source_reject_warm_reads() { + let fixture = Fixture::new_with_pg_config(true).await; + let warm = success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + for (sid, lease) in [ + (fixture.snapshot.clone(), uuid::Uuid::new_v4().to_string()), + (format!("sha256:{}", "1".repeat(64)), fixture.lease.clone()), + ] { + let request = Request::builder() + .method("GET") + .uri(format!("/api/v2/snapshots/{sid}/descriptor")) + .header("authorization", format!("Bearer {TOKEN}")) + .header("x-mega-snapshot-lease", lease) + .header("if-none-match", format!("\"{}\"", fixture.digest_string())) + .body(Body::empty()) + .unwrap(); + error( + fixture.app.clone().oneshot(request).await.unwrap(), + 410, + "LEASE_EXPIRED", + false, + ) + .await; + } + db.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE mst2_snapshot_context SET authorization_epoch=2 WHERE snapshot_id=$1", + [fixture.snapshot.clone().into()], + )) + .await + .unwrap(); + let request = fixture.request("GET", "descriptor", Body::empty()); + error( + fixture.app.clone().oneshot(request).await.unwrap(), + 403, + "SCOPE_FORBIDDEN", + false, + ) + .await; + db.execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE mst2_snapshot_context SET authorization_epoch=1 WHERE snapshot_id=$1", + [fixture.snapshot.clone().into()], + )) + .await + .unwrap(); + db.execute_unprepared("UPDATE mst2_native_publication SET writer_epoch=2") + .await + .unwrap(); + error( + fixture.send("GET", "descriptor", Body::empty()).await, + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + let state = rebuilt(&fixture).await; + error( + app(&state) + .oneshot(fixture.request("GET", "descriptor", Body::empty())) + .await + .unwrap(), + 502, + "INTEGRITY_ERROR", + false, + ) + .await; + assert_eq!(warm["snapshot_id"], fixture.snapshot); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_large_dag_batches_handoff_without_scanning_pages_or_edges() { + let fixture = Fixture::new_with_pg_config_and_directories(true, 80).await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + assert!(scalar(db, "SELECT count(*) FROM mst2_metadata_payload").await > 64); + db.execute_unprepared( + "CREATE TABLE http_install_batch_count(singleton integer PRIMARY KEY,batches bigint NOT NULL); + INSERT INTO http_install_batch_count VALUES(1,0); + CREATE FUNCTION count_http_install_batch() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN UPDATE http_install_batch_count SET batches=batches+1 WHERE singleton=1; RETURN NULL; END $$; + CREATE TRIGGER count_http_install_batch AFTER INSERT ON mst2_metadata_payload + FOR EACH STATEMENT EXECUTE FUNCTION count_http_install_batch()", + ).await.unwrap(); + let prepared = Arc::new(Barrier::new(2)); + let release = Arc::new(Barrier::new(2)); + let resolving = { + let app = fixture.app.clone(); + let prepared = prepared.clone(); + let release = release.clone(); + tokio::spawn(async move { + with_native_handoff_barriers(prepared, release, app.oneshot(resolve_request("/"))) + .await + .unwrap() + }) + }; + prepared.wait().await; + let members = scalar( + db, + "SELECT count(*) FROM mst2_metadata_prepare_page pp JOIN mst2_metadata_prepare p + ON p.prepare_id=pp.prepare_id WHERE p.scope='/'", + ) + .await; + assert!(members > 64); + assert_eq!( + scalar(db, "SELECT batches FROM http_install_batch_count").await, + (members + 63) / 64 + ); + let writer = db.begin().await.unwrap(); + assert!( + PushQueueStorage::try_mono_write_lock(&writer) + .await + .unwrap(), + "full projection and installation completed without retaining the mono writer lock" + ); + writer.rollback().await.unwrap(); + let blocked_scans = db.begin().await.unwrap(); + blocked_scans.execute_unprepared( + "LOCK TABLE mst2_metadata_prepare_page,mst2_metadata_payload,mst2_retention_edge IN ACCESS EXCLUSIVE MODE", + ).await.unwrap(); + release.wait().await; + let response = tokio::time::timeout(Duration::from_secs(5), resolving) + .await + .expect("handoff must not read page membership, payload, or edge tables") + .unwrap(); + let resolved = success_json(response).await; + let writer = db.begin().await.unwrap(); + assert!( + PushQueueStorage::try_mono_write_lock(&writer) + .await + .unwrap() + ); + writer.rollback().await.unwrap(); + blocked_scans.rollback().await.unwrap(); + assert_eq!(resolved["descriptor"]["scope"], "/"); + assert_eq!( + scalar(db, "SELECT batches FROM http_install_batch_count").await, + (members + 63) / 64 + ); + fixture.counts.assert(0, 0); + let warm = success_json( + fixture + .app + .clone() + .oneshot(resolve_request("/")) + .await + .unwrap(), + ) + .await; + assert_eq!( + warm["descriptor"]["snapshot_id"], + resolved["descriptor"]["snapshot_id"] + ); + assert_eq!( + scalar(db, "SELECT batches FROM http_install_batch_count").await, + (members + 63) / 64 + ); + advance(&fixture).await; + let state = rebuilt(&fixture).await; + let directory = success_json( + app(&state) + .oneshot(fixture.request("GET", "directory?path=/wide-079", Body::empty())) + .await + .unwrap(), + ) + .await; + assert_eq!(directory["entries"][0]["name"], "file-079"); + assert_eq!( + app(&state) + .oneshot(fixture.request("HEAD", "blob?path=/wide-079/file-079", Body::empty())) + .await + .unwrap() + .status(), + 200 + ); + fixture.counts.assert(0, 0); +} + +#[tokio::test] +async fn mst2_durable_http_handoff_rejects_changed_plan_or_receipt_summary() { + for mutation in [ + "canonical_plan=canonical_plan||decode('00','hex')", + "projection_revision=projection_revision+1", + "total_bytes=total_bytes+1", + ] { + let fixture = Fixture::new_with_pg_config_and_directories(true, 80).await; + let prepared = Arc::new(Barrier::new(2)); + let release = Arc::new(Barrier::new(2)); + let resolving = { + let app = fixture.app.clone(); + let prepared = prepared.clone(); + let release = release.clone(); + tokio::spawn(async move { + with_native_handoff_barriers(prepared, release, app.oneshot(resolve_request("/"))) + .await + .unwrap() + }) + }; + prepared.wait().await; + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + db.execute_unprepared(&format!( + "UPDATE mst2_metadata_prepare SET {mutation} WHERE scope='/'" + )) + .await + .unwrap(); + release.wait().await; + error(resolving.await.unwrap(), 502, "INTEGRITY_ERROR", false).await; + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_context").await, + 1 + ); + assert_eq!( + scalar(db, "SELECT count(*) FROM mst2_snapshot_lease").await, + 1 + ); + assert_eq!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='lease'" + ) + .await, + 1 + ); + assert!( + scalar( + db, + "SELECT count(*) FROM mst2_retention_root WHERE root_kind='prepare'" + ) + .await + > 64 + ); + assert_eq!( + fixture + .send("HEAD", "blob?path=/file", Body::empty()) + .await + .status(), + 200 + ); + fixture.counts.assert(0, 0); + } +} + +async fn overwrite_primary_scope_for_test(db: &DatabaseConnection, storage_uuid: String) { + // Fault injection owns this isolated schema. Restore the production + // immutable trigger before commit; failed injection rolls back its DDL. + let txn = db.begin().await.unwrap(); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_storage_scope DISABLE TRIGGER mst2_metadata_scope_immutable", + ) + .await + .unwrap(); + let changed = txn + .execute_raw(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE mst2_metadata_storage_scope SET storage_uuid=$1 WHERE singleton=1", + [storage_uuid.into()], + )) + .await + .unwrap(); + assert_eq!(changed.rows_affected(), 1); + txn.execute_unprepared( + "ALTER TABLE mst2_metadata_storage_scope ENABLE TRIGGER mst2_metadata_scope_immutable", + ) + .await + .unwrap(); + txn.commit().await.unwrap(); +} + +#[tokio::test] +async fn mst2_durable_http_warm_reads_renew_and_frame_delivery_reject_primary_scope_drift() { + let fixture = Fixture::new_with_pg_config(true).await; + success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + let response = fixture + .send( + "POST", + "metadata/pages", + Body::from(json!({"items":[{"directory_path":"/nested"}]}).to_string()), + ) + .await; + assert_eq!(response.status(), 200); + let mut body = response.into_body().into_data_stream(); + assert!(body.next().await.unwrap().is_ok()); + let mono = fixture.state.storage.mono_storage(); + let db = mono.get_connection(); + let original: String = db + .query_one_raw(Statement::from_string( + DbBackend::Postgres, + "SELECT storage_uuid FROM mst2_metadata_storage_scope WHERE singleton=1".to_owned(), + )) + .await + .unwrap() + .unwrap() + .try_get_by_index(0) + .unwrap(); + overwrite_primary_scope_for_test(db, uuid::Uuid::new_v4().to_string()).await; + error( + fixture.send("GET", "descriptor", Body::empty()).await, + 500, + "INTERNAL", + true, + ) + .await; + error( + fixture + .app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/v2/snapshots/leases/{}/renew", fixture.lease)) + .header("authorization", format!("Bearer {TOKEN}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(), + 500, + "INTERNAL", + true, + ) + .await; + assert!( + body.next().await.unwrap().is_err(), + "cached validation cannot emit END after scope drift" + ); + assert!(body.next().await.is_none()); + overwrite_primary_scope_for_test(db, original).await; + success_json(fixture.send("GET", "descriptor", Body::empty()).await).await; + fixture.counts.assert(0, 0); +} diff --git a/src/callisto/mod.rs b/src/callisto/mod.rs index ca18ec38..c8d90192 100644 --- a/src/callisto/mod.rs +++ b/src/callisto/mod.rs @@ -92,3 +92,6 @@ pub mod ssh_keys; pub mod user_notification_preferences; pub mod user_notification_settings; pub mod vault; + +pub mod mst2_snapshot_context; +pub mod mst2_snapshot_lease; diff --git a/src/callisto/mst2_snapshot_context.rs b/src/callisto/mst2_snapshot_context.rs new file mode 100644 index 00000000..5e7acefc --- /dev/null +++ b/src/callisto/mst2_snapshot_context.rs @@ -0,0 +1,24 @@ +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "mst2_snapshot_context")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub snapshot_id: String, + pub canonical_descriptor: Vec, + pub instance_id: String, + pub commit_oid: String, + pub root_tree_oid: String, + pub metadata_root: Vec, + pub prepare_id: String, + pub publication_sequence: i64, + pub writer_epoch: i64, + pub certificate_receipt_id: Option, + pub authorization_epoch: i64, + pub state: String, + pub created_at: DateTimeWithTimeZone, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/mst2_snapshot_lease.rs b/src/callisto/mst2_snapshot_lease.rs new file mode 100644 index 00000000..2d5cc887 --- /dev/null +++ b/src/callisto/mst2_snapshot_lease.rs @@ -0,0 +1,20 @@ +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "mst2_snapshot_lease")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub lease_id: String, + pub snapshot_id: String, + pub authorization_epoch: i64, + pub publication_sequence: i64, + pub writer_epoch: i64, + pub certificate_receipt_id: i64, + pub expires_at_unix: i64, + pub state: String, + pub created_at: DateTimeWithTimeZone, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/jupiter/migration/m20261007_000100_add_mst2_snapshot_sessions.rs b/src/jupiter/migration/m20261007_000100_add_mst2_snapshot_sessions.rs new file mode 100644 index 00000000..be67efa5 --- /dev/null +++ b/src/jupiter/migration/m20261007_000100_add_mst2_snapshot_sessions.rs @@ -0,0 +1,72 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager.get_connection().execute_unprepared( + "CREATE TABLE mst2_snapshot_context ( + snapshot_id text PRIMARY KEY, + canonical_descriptor bytea NOT NULL, + instance_id text NOT NULL, + commit_oid text NOT NULL, + root_tree_oid text NOT NULL, + metadata_root bytea NOT NULL CHECK (octet_length(metadata_root)=32), + prepare_id text NOT NULL UNIQUE REFERENCES mst2_metadata_prepare(prepare_id), + publication_sequence bigint NOT NULL CHECK (publication_sequence>=0), + writer_epoch bigint NOT NULL CHECK (writer_epoch>0), + certificate_receipt_id bigint REFERENCES mst2_native_publication(receipt_id), + authorization_epoch bigint NOT NULL DEFAULT 1 CHECK (authorization_epoch>0), + state text NOT NULL CHECK (state IN ('READY','DISABLED')), + created_at timestamptz NOT NULL DEFAULT clock_timestamp() + ); + CREATE TABLE mst2_snapshot_lease ( + lease_id text PRIMARY KEY, + snapshot_id text NOT NULL REFERENCES mst2_snapshot_context(snapshot_id), + authorization_epoch bigint NOT NULL CHECK (authorization_epoch>0), + publication_sequence bigint NOT NULL CHECK (publication_sequence>0), + writer_epoch bigint NOT NULL CHECK (writer_epoch>0), + certificate_receipt_id bigint NOT NULL REFERENCES mst2_native_publication(receipt_id), + expires_at_unix bigint NOT NULL CHECK (expires_at_unix>=0), + state text NOT NULL CHECK (state IN ('ACTIVE','RELEASED','EXPIRED')), + created_at timestamptz NOT NULL DEFAULT clock_timestamp() + ); + CREATE FUNCTION mst2_snapshot_lease_identity_immutable() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.lease_id<>OLD.lease_id OR NEW.snapshot_id<>OLD.snapshot_id + OR NEW.authorization_epoch<>OLD.authorization_epoch OR NEW.publication_sequence<>OLD.publication_sequence + OR NEW.writer_epoch<>OLD.writer_epoch OR NEW.certificate_receipt_id<>OLD.certificate_receipt_id + OR (OLD.state<>'ACTIVE' AND NEW.state<>OLD.state) THEN + RAISE EXCEPTION 'MST2 lease source cannot change or be revived'; + END IF; + RETURN NEW; + END $$; + CREATE TRIGGER mst2_snapshot_lease_identity_immutable BEFORE UPDATE ON mst2_snapshot_lease + FOR EACH ROW EXECUTE FUNCTION mst2_snapshot_lease_identity_immutable(); + CREATE INDEX mst2_snapshot_lease_active ON mst2_snapshot_lease(snapshot_id,expires_at_unix) + WHERE state='ACTIVE'; + CREATE INDEX mst2_snapshot_lease_expiry ON mst2_snapshot_lease(expires_at_unix) + WHERE state='ACTIVE'; + CREATE FUNCTION mst2_snapshot_identity_immutable() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.snapshot_id<>OLD.snapshot_id OR NEW.canonical_descriptor<>OLD.canonical_descriptor + OR NEW.instance_id<>OLD.instance_id OR NEW.commit_oid<>OLD.commit_oid + OR NEW.root_tree_oid<>OLD.root_tree_oid OR NEW.metadata_root<>OLD.metadata_root + OR NEW.prepare_id<>OLD.prepare_id OR NEW.publication_sequence<>OLD.publication_sequence + OR NEW.writer_epoch<>OLD.writer_epoch + OR NEW.certificate_receipt_id IS DISTINCT FROM OLD.certificate_receipt_id THEN + RAISE EXCEPTION 'MST2 fixed snapshot identity cannot be changed'; + END IF; + RETURN NEW; + END $$; + CREATE TRIGGER mst2_snapshot_identity_immutable BEFORE UPDATE ON mst2_snapshot_context + FOR EACH ROW EXECUTE FUNCTION mst2_snapshot_identity_immutable();" + ).await.map(|_| ()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 7f964e4a..c102df51 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -145,6 +145,7 @@ mod m20261005_000200_add_mst2_native_head; mod m20261005_000200_harden_mst2_retention_graph; mod m20261005_000300_add_mst2_metadata_install; mod m20261006_000100_add_view_tables; +mod m20261007_000100_add_mst2_snapshot_sessions; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -278,6 +279,7 @@ impl MigratorTrait for Migrator { Box::new(m20261005_000200_harden_mst2_retention_graph::Migration), Box::new(m20261005_000300_add_mst2_metadata_install::Migration), Box::new(m20261006_000100_add_view_tables::Migration), + Box::new(m20261007_000100_add_mst2_snapshot_sessions::Migration), ] } } @@ -1188,7 +1190,7 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 8..], + &names[names.len() - 9..], &[ "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), @@ -1198,6 +1200,7 @@ mod tests { "m20261005_000200_harden_mst2_retention_graph".to_string(), "m20261005_000300_add_mst2_metadata_install".to_string(), VIEW_MIGRATION_NAME.to_string(), + "m20261007_000100_add_mst2_snapshot_sessions".to_string(), ], "native retention, metadata installation and view tables follow media paging" ); diff --git a/src/jupiter/storage/mod.rs b/src/jupiter/storage/mod.rs index 655a7d02..b130aa5f 100644 --- a/src/jupiter/storage/mod.rs +++ b/src/jupiter/storage/mod.rs @@ -20,6 +20,7 @@ pub(crate) mod mst2_publication_storage; pub mod mst2_retention; pub mod native_metadata_install; pub(crate) mod native_publication_storage; +pub(crate) mod native_snapshot_session; pub mod notification_storage; pub mod object_storage; pub mod oci_db_storage; @@ -158,6 +159,8 @@ pub struct Storage { /// Derived native projection memoization, scoped to this storage assembly. /// Clones share it; independent databases/backends never share entries. pub(crate) native_projection_cache: Arc, + pub(crate) native_snapshot_sessions: + Arc>, pub(crate) projection_observation_sink: Option>, pub cl_service: CLService, @@ -322,6 +325,7 @@ impl Storage { Ok(Storage { app_service: app_service.into(), native_projection_cache: Arc::default(), + native_snapshot_sessions: Arc::default(), projection_observation_sink: None, config_handle, config, @@ -699,6 +703,7 @@ impl Storage { Storage { app_service, native_projection_cache: Arc::default(), + native_snapshot_sessions: Arc::default(), projection_observation_sink: None, // app_service: AppService::mock(), cl_service: CLService::mock(), diff --git a/src/jupiter/storage/mst2_retention.rs b/src/jupiter/storage/mst2_retention.rs index 11c9c456..b336b7a1 100644 --- a/src/jupiter/storage/mst2_retention.rs +++ b/src/jupiter/storage/mst2_retention.rs @@ -129,6 +129,53 @@ impl PostgresRetentionRepository { finish(txn, result).await } + /// Attach a root to an already verified immutable DAG root. Incoming edges + /// retain descendants; this does not copy every page for each lease. + pub(crate) async fn acquire_existing_roots_in_txn( + txn: &DatabaseTransaction, + node_id: &str, + roots: &[RetentionRoot], + ) -> Result<(), SnapshotError> { + validate_id(node_id)?; + if roots.is_empty() || roots.len() > MAX_ROOTS { + return Err(limit_error( + "existing-root acquisition requires 1..=16 roots", + )); + } + let identities = roots + .iter() + .map(root_identity) + .collect::, _>>()?; + let roots: Vec<_> = identities + .iter() + .map(|(key, kind)| json!({"key":key,"kind":kind})) + .collect(); + let encoded = serde_json::to_string(&roots).map_err(internal)?; + let savepoint = begin_graph(txn).await?; + let result=async { + let row=savepoint.query_one_raw(statement( + "SELECT state FROM mst2_retention_node WHERE node_id=$1 FOR UPDATE",[node_id.into()] + )).await.map_err(internal)?.ok_or_else(|| unavailable("metadata root is missing"))?; + if row.try_get::("","state").map_err(internal)?!="LIVE" { + return Err(unavailable("metadata root is not LIVE")); + } + if savepoint.query_one_raw(statement( + "SELECT r.node_id FROM mst2_retention_root r JOIN jsonb_to_recordset($2::jsonb) + AS p(key text,kind text) ON r.root_key=p.key WHERE r.node_id<>$1 OR r.root_kind<>p.kind LIMIT 1", + [node_id.into(),encoded.clone().into()], + )).await.map_err(internal)?.is_some() { + return Err(integrity("retention root is bound to another DAG or kind")); + } + savepoint.execute_raw(statement( + "INSERT INTO mst2_retention_root(node_id,root_key,root_kind) + SELECT $1,p.key,p.kind FROM jsonb_to_recordset($2::jsonb) AS p(key text,kind text) + ON CONFLICT(node_id,root_key) DO NOTHING",[node_id.into(),encoded.into()], + )).await.map_err(internal)?; + Ok(()) + }.await; + finish(savepoint, result).await + } + pub async fn release_root_in_txn( txn: &DatabaseTransaction, root: &RetentionRoot, diff --git a/src/jupiter/storage/native_metadata_install.rs b/src/jupiter/storage/native_metadata_install.rs index 7f7d80ed..659989c4 100644 --- a/src/jupiter/storage/native_metadata_install.rs +++ b/src/jupiter/storage/native_metadata_install.rs @@ -1,5 +1,5 @@ //! Immutable metadata payload CAS and durable native preparation receipts. -//! No runtime, publication, lease or physical collector is enabled here. +//! The native HTTP session authority installs batches and transfers prepare pins. use std::{ collections::{BTreeMap, BTreeSet}, @@ -9,7 +9,7 @@ use std::{ use mst2_codec::metapage::{HEADER_LEN, PAGE_MAX_BYTES, Page, page_id}; use sea_orm::{ ColumnTrait, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, EntityTrait, - IsolationLevel, QueryFilter, QuerySelect, Statement, TransactionTrait, + IsolationLevel, QueryFilter, QueryResult, QuerySelect, Statement, TransactionTrait, }; use serde_json::json; @@ -21,7 +21,7 @@ use crate::{ }, ceres::snapshot::{ error::{SnapshotError, SnapshotErrorCode}, - metadata_install::MetadataInstallPlan, + metadata_install::{MAX_PLAN_BYTES, MetadataInstallIdentity, MetadataInstallPlan}, pages::PreparedNativeMetadataRetention, retention::RetentionRoot, retention_dag::{ @@ -71,8 +71,12 @@ impl MetadataPrepareIntent { #[derive(Debug, Clone, PartialEq, Eq)] pub struct PreparedMetadataReceipt { intent: MetadataPrepareIntent, + identity: MetadataInstallIdentity, metadata_root: [u8; 32], payload_bytes: u64, + root_payload_bytes: u64, + node_count: usize, + edge_count: usize, } impl PreparedMetadataReceipt { @@ -115,8 +119,16 @@ impl StoredPlan { fn receipt(&self) -> Result { Ok(PreparedMetadataReceipt { intent: self.intent()?, + identity: self.plan.identity.clone(), metadata_root: self.plan.root, payload_bytes: self.plan.total_bytes, + root_payload_bytes: *self + .plan + .pages + .get(&self.plan.root) + .ok_or_else(|| integrity("metadata preparation root page is missing"))?, + node_count: self.plan.pages.len(), + edge_count: self.plan.edges.len(), }) } } @@ -150,6 +162,49 @@ impl PostgresMetadataInstallRepository { }) } + /// These columns are returned by the same query that authorizes a lease, + /// so a warm cache cannot authorize a stale replica or another schema. + pub(crate) fn verify_primary_scope_row(&self, row: &QueryResult) -> Result<(), SnapshotError> { + let actual = PrimaryStorageScope { + storage_uuid: row + .try_get::>("", "authority_storage_uuid") + .map_err(internal)? + .ok_or_else(|| internal("session authority storage scope is missing"))?, + database: row.try_get("", "authority_database").map_err(internal)?, + database_oid: row + .try_get("", "authority_database_oid") + .map_err(internal)?, + schema: row.try_get("", "authority_schema").map_err(internal)?, + schema_oid: row.try_get("", "authority_schema_oid").map_err(internal)?, + server_address: row + .try_get("", "authority_server_address") + .map_err(internal)?, + server_port: row.try_get("", "authority_server_port").map_err(internal)?, + }; + if row + .try_get::("", "authority_replica") + .map_err(internal)? + || actual != self.storage_scope + { + return Err(internal( + "session authority no longer targets its captured primary storage scope", + )); + } + Ok(()) + } + + pub(crate) async fn verify_primary_connection( + &self, + connection: &C, + ) -> Result<(), SnapshotError> { + if read_storage_scope(connection).await? != self.storage_scope { + return Err(internal( + "session mutation no longer targets its captured primary storage scope", + )); + } + Ok(()) + } + pub async fn begin_intent( &self, operation_id: &str, @@ -203,44 +258,62 @@ impl PostgresMetadataInstallRepository { intent: &MetadataPrepareIntent, payload: &MetadataPagePayload, ) -> Result<(), MetadataInstallError> { - validate_payload(payload)?; + self.install_pages(intent, std::slice::from_ref(payload)) + .await + } + + /// At most 64 pages (1 MiB of canonical payload) per transaction. + pub async fn install_pages( + &self, + intent: &MetadataPrepareIntent, + payloads: &[MetadataPagePayload], + ) -> Result<(), MetadataInstallError> { + if payloads.is_empty() || payloads.len() > 64 { + return Err(SnapshotError::new( + SnapshotErrorCode::LimitExceeded, + "metadata installation batch must contain 1..=64 pages", + ) + .into()); + } + let mut ids = BTreeSet::new(); + for payload in payloads { + validate_payload(payload)?; + if !ids.insert(payload.id) { + return Err(integrity("duplicate page in metadata installation batch").into()); + } + } let txn = self.transaction().await?; let result = async { self.barrier(&txn).await?; let stored = require_plan(&txn, intent).await?; - if stored.plan.pages.get(&payload.id) != Some(&payload.size) { - return Err(integrity( - "metadata payload is not a member of this fixed installation", - )); + for payload in payloads { + if stored.plan.pages.get(&payload.id) != Some(&payload.size) { + return Err(integrity("metadata payload is not a member of this fixed installation")); + } } + let pages: Vec<_> = payloads.iter().map(|p| json!({ + "page_id":hex::encode(p.id), "size":p.size, "payload":hex::encode(&p.bytes) + })).collect(); + let encoded = serde_json::to_string(&pages).map_err(internal)?; txn.execute_raw(statement( "INSERT INTO mst2_metadata_payload(page_id,metadata_codec,byte_size,payload) - VALUES ($1,$2,$3,$4) ON CONFLICT(page_id) DO NOTHING", - [ - payload.id.to_vec().into(), - (stored.plan.identity.metadata_codec as i16).into(), - (payload.size as i32).into(), - payload.bytes.clone().into(), - ], - )) - .await - .map_err(internal)?; - let existing = mst2_metadata_payload::Entity::find_by_id(payload.id.to_vec()) - .one(&txn) - .await - .map_err(internal)? - .ok_or_else(|| integrity("installed metadata payload disappeared"))?; - if existing.payload != payload.bytes - || existing.byte_size as u64 != payload.size - || existing.metadata_codec != stored.plan.identity.metadata_codec as i16 - { - return Err(integrity( - "immutable metadata payload identity conflicts with stored bytes", - )); + SELECT decode(p.page_id,'hex'),$1,p.size,decode(p.payload,'hex') + FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer,payload text) + ON CONFLICT(page_id) DO NOTHING", + [(stored.plan.identity.metadata_codec as i16).into(),encoded.clone().into()], + )).await.map_err(internal)?; + let bad = txn.query_one_raw(statement( + "SELECT p.page_id FROM jsonb_to_recordset($2::jsonb) AS p(page_id text,size integer,payload text) + LEFT JOIN mst2_metadata_payload b ON b.page_id=decode(p.page_id,'hex') + WHERE b.page_id IS NULL OR b.metadata_codec<>$1 OR b.byte_size<>p.size + OR b.payload<>decode(p.payload,'hex') LIMIT 1", + [(stored.plan.identity.metadata_codec as i16).into(),encoded.into()], + )).await.map_err(internal)?; + if bad.is_some() { + return Err(integrity("immutable metadata payload identity conflicts with stored bytes")); } Ok(()) - } - .await; + }.await; commit( txn, result, @@ -251,6 +324,156 @@ impl PostgresMetadataInstallRepository { .await } + pub(crate) async fn verify_receipt_in_txn( + &self, + txn: &DatabaseTransaction, + receipt: &PreparedMetadataReceipt, + tagged_root_tree_oid: &str, + scope: &str, + ) -> Result<(), SnapshotError> { + self.barrier(txn).await?; + let identity = &receipt.identity; + if identity.tagged_root_tree_oid != tagged_root_tree_oid || identity.scope != scope { + return Err(integrity( + "metadata receipt belongs to another fixed source", + )); + } + // The private receipt can only be issued after full validation and a + // definitive commit. Hash the bounded canonical plan in PostgreSQL, + // bind all redundant identity/summary columns to that receipt, and + // check its still-protected LIVE root under the retention barrier. + // This hashes <=2 MiB; it does not transfer or scan pages/edges here. + let row = txn.query_one_raw(statement( + "SELECT p.operation_id,p.manifest_digest, + CASE WHEN octet_length(p.canonical_plan)<=$2 THEN sha256(p.canonical_plan) END AS plan_digest, + p.source_domain,p.tagged_root_tree_oid,p.scope,p.schema_version,p.metadata_codec, + p.materialization_policy,p.fs_semantics,p.access_projection,p.verification_revision, + p.projection_revision,p.metadata_root,p.node_count,p.edge_count,p.total_bytes, + p.state,p.committed_at IS NOT NULL AS committed, + n.kind AS root_kind,n.state AS root_state,n.bytes AS root_bytes, + EXISTS(SELECT 1 FROM mst2_retention_root r WHERE r.node_id=n.node_id + AND r.root_key='prepare:'||p.prepare_id AND r.root_kind='prepare') AS prepare_covered + FROM mst2_metadata_prepare p LEFT JOIN mst2_retention_node n + ON n.node_id='page:sha256:'||encode(p.metadata_root,'hex') WHERE p.prepare_id=$1", + [receipt.intent.prepare_id.clone().into(), (MAX_PLAN_BYTES as i32).into()], + )).await.map_err(internal)?.ok_or_else(|| unavailable("metadata preparation is missing"))?; + if row + .try_get::("", "operation_id") + .map_err(internal)? + != receipt.intent.operation_id + || row + .try_get::>("", "manifest_digest") + .map_err(internal)? + != receipt.intent.manifest_digest + || row + .try_get::>>("", "plan_digest") + .map_err(internal)? + .as_deref() + != Some(receipt.intent.manifest_digest.as_slice()) + || row + .try_get::("", "source_domain") + .map_err(internal)? + != identity.source_domain + || row + .try_get::("", "tagged_root_tree_oid") + .map_err(internal)? + != identity.tagged_root_tree_oid + || row.try_get::("", "scope").map_err(internal)? != identity.scope + || row.try_get::("", "schema_version").map_err(internal)? + != identity.schema_version as i16 + || row.try_get::("", "metadata_codec").map_err(internal)? + != identity.metadata_codec as i16 + || row + .try_get::("", "materialization_policy") + .map_err(internal)? + != identity.materialization_policy as i16 + || row.try_get::("", "fs_semantics").map_err(internal)? + != identity.fs_semantics as i16 + || row + .try_get::("", "access_projection") + .map_err(internal)? + != identity.access_projection as i16 + || row + .try_get::("", "verification_revision") + .map_err(internal)? + != identity.verification_revision + || row + .try_get::("", "projection_revision") + .map_err(internal)? + != identity.projection_revision as i16 + || row + .try_get::>("", "metadata_root") + .map_err(internal)? + != receipt.metadata_root + || row.try_get::("", "node_count").map_err(internal)? != receipt.node_count as i32 + || row.try_get::("", "edge_count").map_err(internal)? != receipt.edge_count as i32 + || row.try_get::("", "total_bytes").map_err(internal)? + != receipt.payload_bytes as i64 + || row.try_get::("", "state").map_err(internal)? != "COMMITTED" + || !row.try_get::("", "committed").map_err(internal)? + { + return Err(integrity( + "metadata preparation differs from its definitive receipt", + )); + } + if row + .try_get::>("", "root_kind") + .map_err(internal)? + .as_deref() + != Some("page") + || row + .try_get::>("", "root_state") + .map_err(internal)? + .as_deref() + != Some("LIVE") + || row + .try_get::>("", "root_bytes") + .map_err(internal)? + != Some(receipt.root_payload_bytes as i64) + || !row + .try_get::("", "prepare_covered") + .map_err(internal)? + { + return Err(unavailable( + "metadata receipt root is not LIVE and protected by its prepare", + )); + } + Ok(()) + } + + pub(crate) async fn restore_session_dag(&self, prepare_id: &str) -> Result<(), SnapshotError> { + let record = mst2_metadata_prepare::Entity::find_by_id(prepare_id.to_owned()) + .one(&self.connection) + .await + .map_err(internal)? + .ok_or_else(|| unavailable("snapshot preparation is missing"))?; + let digest = record + .manifest_digest + .as_slice() + .try_into() + .map_err(|_| integrity("invalid metadata preparation digest"))?; + let stored = load_plan(&self.connection, &record.operation_id, &digest) + .await? + .ok_or_else(|| unavailable("snapshot preparation disappeared"))?; + if stored.record.state != "COMMITTED" { + return Err(unavailable("snapshot preparation is not committed")); + } + load_installed_dag(&self.connection, &stored).await?; + let txn = self.transaction().await?; + let result = async { + self.barrier(&txn).await?; + verify_graph(&txn, &stored).await + } + .await; + match result { + Ok(()) => txn.commit().await.map_err(internal), + Err(error) => { + txn.rollback().await.map_err(internal)?; + Err(error) + } + } + } + /// Verify all stored bytes outside the graph transaction. Immutable payloads /// cannot change; LIVE state and coverage are checked again under the lock. pub async fn load_installed_dag( @@ -702,7 +925,21 @@ async fn verify_graph( .all(connection) .await .map_err(internal)?; - if roots.iter().any(|root| root.root_kind != "prepare") + if roots.is_empty() { + let transferred = connection + .query_one_raw(statement( + "SELECT s.snapshot_id FROM mst2_snapshot_context s + JOIN mst2_retention_root r ON r.root_key='pin:session:'||s.snapshot_id + AND r.node_id='page:sha256:'||encode(s.metadata_root,'hex') AND r.root_kind='pin' + WHERE s.prepare_id=$1 LIMIT 1", + [stored.record.prepare_id.clone().into()], + )) + .await + .map_err(internal)?; + if transferred.is_none() { + return Err(unavailable("metadata protection handoff is incomplete")); + } + } else if roots.iter().any(|root| root.root_kind != "prepare") || roots .into_iter() .map(|root| root.node_id) diff --git a/src/jupiter/storage/native_publication_storage.rs b/src/jupiter/storage/native_publication_storage.rs index 8039a3ea..2dbff900 100644 --- a/src/jupiter/storage/native_publication_storage.rs +++ b/src/jupiter/storage/native_publication_storage.rs @@ -223,7 +223,7 @@ pub(crate) fn decode_native_observation( }) } -async fn observe( +pub(crate) async fn observe( connection: &C, ) -> Result { let rows = connection @@ -268,7 +268,14 @@ impl MonoStorage { &self, instance: &str, ) -> Result { - let observation = observe(self.get_connection()).await?; + Self::read_native_publication_head_from(self.get_connection(), instance).await + } + + pub(crate) async fn read_native_publication_head_from( + connection: &C, + instance: &str, + ) -> Result { + let observation = observe(connection).await?; let head = observation .head .ok_or_else(|| integrity("native publication is not initialized"))?; diff --git a/src/jupiter/storage/native_snapshot_session.rs b/src/jupiter/storage/native_snapshot_session.rs new file mode 100644 index 00000000..415228c8 --- /dev/null +++ b/src/jupiter/storage/native_snapshot_session.rs @@ -0,0 +1,678 @@ +//! Durable native HTTP sessions. Deployment bearer authentication is enforced +//! by the router; leases retain data and do not grant per-principal permission. + +use std::{collections::HashMap, sync::Arc}; + +use mst2_codec::descriptor::ServingDescriptor; +use sea_orm::{ + ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, EntityTrait, + IsolationLevel, QueryResult, Statement, TransactionTrait, +}; +use tokio::sync::{Mutex, OnceCell}; + +use super::{ + mono_storage::MonoStorage, + mst2_retention::{PostgresRetentionRepository, RETENTION_LOCK_KEY}, + native_metadata_install::{ + MetadataInstallError, PostgresMetadataInstallRepository, PreparedMetadataReceipt, + }, + native_publication_storage::{NativePublicationHead, decode_native_observation}, + push_queue_storage::PushQueueStorage, +}; +use crate::{ + callisto::mst2_snapshot_context, + ceres::snapshot::{ + descriptor::BuiltDescriptor, + error::{SnapshotError, SnapshotErrorCode}, + pages::PreparedNativeMetadataRetention, + retention::RetentionRoot, + runtime::{LeaseRenewed, SnapshotContext}, + view::{SnapshotView, hex}, + }, +}; + +pub(crate) struct PostgresNativeSessionRepository { + connection: DatabaseConnection, + installer: OnceCell, + restored: Mutex>>>, +} + +impl PostgresNativeSessionRepository { + pub(crate) fn new(connection: DatabaseConnection) -> Self { + Self { + connection, + installer: OnceCell::new(), + restored: Mutex::new(HashMap::new()), + } + } + + async fn installer(&self) -> Result<&PostgresMetadataInstallRepository, SnapshotError> { + self.installer + .get_or_try_init(|| PostgresMetadataInstallRepository::new(self.connection.clone())) + .await + } + + pub(crate) async fn install( + &self, + built: &BuiltDescriptor, + prepared: &PreparedNativeMetadataRetention, + ) -> Result { + if prepared.dag().root() != built.descriptor.metadata_root + || prepared.scope() != built.descriptor.scope + { + return Err(integrity( + "prepared DAG differs from the serving descriptor", + )); + } + let installer = self.installer().await?; + let intent = installer + .begin_intent(&format!("http:{}", built.snapshot_id), prepared) + .await + .map_err(install_error)?; + for pages in prepared.dag().payloads().chunks(64) { + installer + .install_pages(&intent, pages) + .await + .map_err(install_error)?; + } + installer.finalize(&intent).await.map_err(install_error) + } + + /// Projection and payload installation happen before this boundary. + /// Cold handoff hashes a <=2 MiB plan, without a page/edge database scan. + /// The formal native head and LIVE root are selected with lease creation. + pub(crate) async fn open( + &self, + expected: &NativePublicationHead, + built: &BuiltDescriptor, + receipt: Option<&PreparedMetadataReceipt>, + lease_seconds: u64, + ) -> Result, SnapshotError> { + let installer = self.installer().await?; + let txn = self.transaction().await?; + let result = async { + PushQueueStorage::acquire_mono_write_lock(&txn).await.map_err(internal)?; + installer.verify_primary_connection(&txn).await?; + let current = MonoStorage::read_native_publication_head_from(&txn, &expected.instance_id) + .await.map_err(|_| not_ready("native publication is not ready"))?; + if current.root != expected.root || current.token != expected.token { + return Err(not_ready("native publication advanced during preparation; retry resolve")); + } + retention_lock(&txn).await?; + let existing = mst2_snapshot_context::Entity::find_by_id(built.snapshot_id.clone()) + .one(&txn).await.map_err(internal)?; + let prepare_id = if let Some(existing) = existing { + if existing.canonical_descriptor != built.descriptor.encode().map_err(internal)? + || existing.commit_oid != expected.root.commit || existing.root_tree_oid != expected.root.tree + || existing.instance_id != expected.instance_id + { + return Err(integrity("immutable session source conflicts with selected publication")); + } + if existing.state != "READY" || existing.authorization_epoch != 1 { + return Err(forbidden()); + } + existing.prepare_id + } else { + let Some(receipt) = receipt else { return Ok(None); }; + if receipt.metadata_root() != built.descriptor.metadata_root { + return Err(integrity("metadata receipt root differs from descriptor")); + } + let prepare_id = receipt.intent().prepare_id().to_owned(); + let tagged_tree=git_internal::hash::ObjectHash::from_hex_for_kind( + git_internal::hash::get_hash_kind(),&expected.root.tree).map_err(internal)?.to_tagged_string(); + self.installer().await?.verify_receipt_in_txn(&txn, receipt,&tagged_tree,&built.descriptor.scope).await?; + txn.execute_raw(statement( + "INSERT INTO mst2_snapshot_context(snapshot_id,canonical_descriptor,instance_id,commit_oid, + root_tree_oid,metadata_root,prepare_id,publication_sequence,writer_epoch,certificate_receipt_id, + authorization_epoch,state) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,1,'READY')", + [built.snapshot_id.clone().into(),built.descriptor.encode().map_err(internal)?.into(), + expected.instance_id.clone().into(),expected.root.commit.clone().into(),expected.root.tree.clone().into(), + built.descriptor.metadata_root.to_vec().into(),prepare_id.clone().into(),expected.token.sequence.into(), + expected.token.epoch.into(),expected.token.certificate.into()], + )).await.map_err(internal)?; + prepare_id + }; + let node = root_node(built); + let lease_id = uuid::Uuid::new_v4().to_string(); + PostgresRetentionRepository::acquire_existing_roots_in_txn(&txn,&node,&[ + RetentionRoot::Pin(format!("session:{}",built.snapshot_id)),RetentionRoot::Lease(lease_id.clone()) + ]).await?; + PostgresRetentionRepository::release_root_in_txn(&txn,&RetentionRoot::Prepare(prepare_id)).await?; + let row = txn.query_one_raw(statement( + "INSERT INTO mst2_snapshot_lease(lease_id,snapshot_id,authorization_epoch,expires_at_unix,state, + publication_sequence,writer_epoch,certificate_receipt_id) + VALUES($1,$2,1,floor(extract(epoch FROM clock_timestamp()))::bigint+$3,'ACTIVE',$4,$5,$6) + RETURNING expires_at_unix", + [lease_id.clone().into(),built.snapshot_id.clone().into(),(lease_seconds.clamp(1,3600) as i64).into(), + expected.token.sequence.into(),expected.token.epoch.into(),expected.token.certificate.into()], + )).await.map_err(internal)?.ok_or_else(|| internal("lease insert returned no deadline"))?; + expire_locked(&txn).await?; + Ok(Some(SnapshotContext { built:built.clone(),commit_oid:expected.root.commit.clone(), + root_tree_oid:expected.root.tree.clone(),lease_id, + lease_expires_at_unix:row.try_get::("","expires_at_unix").map_err(internal)? as u64, + authorization_epoch:1 })) + }.await; + let context = finish(txn, result).await?; + if let Some(context) = &context + && receipt.is_some() + { + let _ = self + .verification_cell(&context.built.snapshot_id) + .await + .set(()); + } + Ok(context) + } + + pub(crate) async fn context( + &self, + snapshot_id: &str, + lease_id: &str, + instance_id: &str, + ) -> Result { + let installer = self.installer().await?; + let row = self + .connection + .query_one_raw(statement( + SESSION_SQL, + [snapshot_id.into(), lease_id.into()], + )) + .await + .map_err(internal)? + .ok_or_else(expired)?; + installer.verify_primary_scope_row(&row)?; + let context = match decode_context(&row, snapshot_id, lease_id, instance_id) { + Ok(context) => context, + Err(error) => { + if error.code == SnapshotErrorCode::LeaseExpired { + let txn = self.transaction().await?; + let result = async { + retention_lock(&txn).await?; + expire_specific_locked(&txn, lease_id).await + } + .await; + finish(txn, result).await?; + } + return Err(error); + } + }; + let prepare_id: String = row.try_get("", "prepare_id").map_err(internal)?; + let cell = self.verification_cell(snapshot_id).await; + cell.get_or_try_init(|| async { + self.installer() + .await? + .restore_session_dag(&prepare_id) + .await + }) + .await?; + Ok(context) + } + + pub(crate) async fn renew( + &self, + lease_id: &str, + seconds: u64, + instance: &str, + ) -> Result { + let txn = self.transaction().await?; + let result = async { + retention_lock(&txn).await?; + let lease = txn.query_one_raw(statement( + "SELECT snapshot_id FROM mst2_snapshot_lease WHERE lease_id=$1 FOR UPDATE", + [lease_id.into()], + )).await.map_err(internal)?.ok_or_else(|| SnapshotError::new(SnapshotErrorCode::LeaseUnknown,"unknown lease_id"))?; + let sid: String = lease.try_get("","snapshot_id").map_err(internal)?; + let row = txn.query_one_raw(statement(SESSION_SQL,[sid.clone().into(),lease_id.into()])) + .await.map_err(internal)?.ok_or_else(expired)?; + self.installer().await?.verify_primary_scope_row(&row)?; + if let Err(error)=decode_context(&row,&sid,lease_id,instance) { + if error.code==SnapshotErrorCode::LeaseExpired { + expire_specific_locked(&txn,lease_id).await?; + return Ok(Err(error)); + } + return Err(error); + } + let updated = txn.query_one_raw(statement( + "UPDATE mst2_snapshot_lease SET expires_at_unix=greatest(expires_at_unix, + floor(extract(epoch FROM clock_timestamp()))::bigint)+$2 WHERE lease_id=$1 + AND state='ACTIVE' AND expires_at_unix>=floor(extract(epoch FROM clock_timestamp()))::bigint + RETURNING expires_at_unix", + [lease_id.into(),(seconds.clamp(1,3600) as i64).into()], + )).await.map_err(internal)?; + let Some(updated)=updated else { + expire_specific_locked(&txn,lease_id).await?; + return Ok(Err(expired())); + }; + Ok(Ok(LeaseRenewed { lease_id:lease_id.into(),snapshot_id:sid, + expires_at_unix:updated.try_get::("","expires_at_unix").map_err(internal)? as u64 })) + }.await; + finish(txn, result).await? + } + + pub(crate) async fn release(&self, lease_id: &str) -> Result { + let installer = self.installer().await?; + let txn = self.transaction().await?; + let result = async { + retention_lock(&txn).await?; + installer.verify_primary_connection(&txn).await?; + let changed=txn.query_one_raw(statement( + "UPDATE mst2_snapshot_lease SET state='RELEASED' WHERE lease_id=$1 AND state='ACTIVE' RETURNING snapshot_id", + [lease_id.into()], + )).await.map_err(internal)?; + PostgresRetentionRepository::release_root_in_txn(&txn,&RetentionRoot::Lease(lease_id.into())).await?; + expire_locked(&txn).await?; + if let Some(row)=&changed { + let sid:String=row.try_get("","snapshot_id").map_err(internal)?; + retire_session_pin_locked(&txn,&sid).await?; + } + Ok(changed.is_some()) + }.await; + finish(txn, result).await + } + + async fn verification_cell(&self, sid: &str) -> Arc> { + let mut restored = self.restored.lock().await; + if restored.len() >= 128 + && !restored.contains_key(sid) + && let Some(old) = restored.keys().next().cloned() + { + restored.remove(&old); + } + restored.entry(sid.to_owned()).or_default().clone() + } + + async fn transaction(&self) -> Result { + self.connection + .begin_with_config(Some(IsolationLevel::ReadCommitted), None) + .await + .map_err(internal) + } +} + +const SESSION_SQL: &str = "SELECT s.snapshot_id,s.canonical_descriptor,s.commit_oid,s.root_tree_oid, + (SELECT storage_uuid FROM mst2_metadata_storage_scope WHERE singleton=1) AS authority_storage_uuid, + current_database() AS authority_database, + (SELECT oid::bigint FROM pg_catalog.pg_database WHERE datname=current_database()) AS authority_database_oid, + current_schema() AS authority_schema, + (SELECT oid::bigint FROM pg_catalog.pg_namespace WHERE nspname=current_schema()) AS authority_schema_oid, + inet_server_addr()::text AS authority_server_address,inet_server_port() AS authority_server_port, + pg_is_in_recovery() AS authority_replica, + s.metadata_root,s.prepare_id,s.authorization_epoch,s.state AS session_state,l.lease_id, + l.authorization_epoch AS lease_epoch,l.expires_at_unix,l.state AS lease_state, + floor(extract(epoch FROM clock_timestamp()))::bigint AS db_now,n.state AS root_state, + EXISTS(SELECT 1 FROM mst2_retention_root rr WHERE rr.node_id=n.node_id AND rr.root_key='lease:'||l.lease_id + AND rr.root_kind='lease') AS lease_covered, + EXISTS(SELECT 1 FROM mst2_retention_root rr WHERE rr.node_id=n.node_id AND rr.root_key='pin:session:'||s.snapshot_id + AND rr.root_kind='pin') AS session_covered, + p.state AS prepare_state,p.metadata_root AS prepared_root,p.tagged_root_tree_oid,p.scope AS prepared_scope, + p.source_domain,p.schema_version,p.metadata_codec,p.materialization_policy,p.fs_semantics,p.access_projection, + p.verification_revision,p.projection_revision, + 1::bigint AS root_count,s.commit_oid AS root_commit,s.root_tree_oid AS root_tree,s.instance_id, + l.publication_sequence AS sequence,l.writer_epoch,'READY'::text AS state,s.commit_oid AS head_commit, + s.root_tree_oid AS head_tree,l.certificate_receipt_id,c.receipt_id AS certificate_id, + c.namespace AS certificate_namespace,c.instance_id AS certificate_instance,c.sequence AS certificate_sequence, + c.writer_epoch AS certificate_epoch,c.old_root_commit AS certificate_old_commit, + c.root_commit AS certificate_commit,c.root_tree AS certificate_tree,c.origin_path,c.origin_ref,c.old_path_commit,c.path_commit, + r.id AS receipt_id,r.namespace AS receipt_namespace,r.sequence AS receipt_sequence,r.operation_id,r.old_oid,r.new_oid, + r.writer_epoch AS receipt_epoch,r.writer_kind,r.request_digest,r.request_digest_version,r.native_certificate_version, + o.id AS outbox_id,o.namespace AS outbox_namespace,o.sequence AS outbox_sequence + FROM mst2_snapshot_context s JOIN mst2_snapshot_lease l ON l.snapshot_id=s.snapshot_id AND l.lease_id=$2 + LEFT JOIN mst2_retention_node n ON n.node_id='page:sha256:'||encode(s.metadata_root,'hex') + LEFT JOIN mst2_metadata_prepare p ON p.prepare_id=s.prepare_id + LEFT JOIN mst2_native_publication c ON c.receipt_id=l.certificate_receipt_id + LEFT JOIN mst2_publication r ON r.id=c.receipt_id + LEFT JOIN mst2_publication_outbox o ON o.operation_id=r.operation_id WHERE s.snapshot_id=$1"; + +fn decode_context( + row: &QueryResult, + sid: &str, + lease: &str, + instance: &str, +) -> Result { + let deadline: i64 = row.try_get("", "expires_at_unix").map_err(internal)?; + if row.try_get::("", "lease_state").map_err(internal)? != "ACTIVE" + || deadline < row.try_get::("", "db_now").map_err(internal)? + { + return Err(expired()); + } + let epoch: i64 = row.try_get("", "authorization_epoch").map_err(internal)?; + if row + .try_get::("", "session_state") + .map_err(internal)? + != "READY" + || epoch != 1 + || row.try_get::("", "lease_epoch").map_err(internal)? != epoch + { + return Err(forbidden()); + } + if row + .try_get::>("", "root_state") + .map_err(internal)? + .as_deref() + != Some("LIVE") + || !row.try_get::("", "lease_covered").map_err(internal)? + || !row + .try_get::("", "session_covered") + .map_err(internal)? + { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + "snapshot root protection is unavailable", + )); + } + let head = decode_native_observation(row) + .map_err(|_| integrity("durable native source certificate is invalid"))? + .head + .ok_or_else(|| integrity("durable native source is missing"))?; + if head.instance_id != instance { + return Err(forbidden()); + } + let canonical: Vec = row.try_get("", "canonical_descriptor").map_err(internal)?; + let descriptor = ServingDescriptor::decode(&canonical) + .map_err(|_| integrity("invalid durable serving descriptor"))?; + let view = SnapshotView::from_commit(&head.root.commit, &head.root.tree); + let root: Vec = row.try_get("", "metadata_root").map_err(internal)?; + let prepared_root: Option> = row.try_get("", "prepared_root").map_err(internal)?; + let tagged_tree = git_internal::hash::ObjectHash::from_hex_for_kind( + git_internal::hash::get_hash_kind(), + &head.root.tree, + ) + .map_err(|_| integrity("invalid durable tree identity"))? + .to_tagged_string(); + if format!( + "sha256:{}", + hex(&descriptor.snapshot_id().map_err(internal)?) + ) != sid + || uuid::Uuid::from_bytes(descriptor.instance_uuid).to_string() != instance + || format!("sha256:{}", hex(&descriptor.namespace_view_id)) != view.view_id + || descriptor.metadata_root.as_slice() != root.as_slice() + || prepared_root.as_deref() != Some(root.as_slice()) + || row + .try_get::>("", "prepare_state") + .map_err(internal)? + .as_deref() + != Some("COMMITTED") + || row + .try_get::>("", "tagged_root_tree_oid") + .map_err(internal)? + .as_deref() + != Some(tagged_tree.as_str()) + || row + .try_get::>("", "prepared_scope") + .map_err(internal)? + .as_deref() + != Some(descriptor.scope.as_str()) + { + return Err(integrity( + "durable snapshot identity or source proof is inconsistent", + )); + } + let profile = [ + ("schema_version", mst2_codec::descriptor::SCHEMA_VERSION), + ("metadata_codec", mst2_codec::descriptor::METADATA_CODEC), + ( + "materialization_policy", + mst2_codec::descriptor::MATERIALIZATION_POLICY_GIT_RAW_V1, + ), + ( + "fs_semantics", + mst2_codec::descriptor::FS_SEMANTICS_LINUX_CODE_V1, + ), + ( + "access_projection", + mst2_codec::descriptor::ACCESS_PROJECTION_EXACT_FULL, + ), + ( + "projection_revision", + crate::ceres::snapshot::projection_observation::NATIVE_PROJECTION_REVISION, + ), + ]; + if profile.iter().any(|(column, value)| { + row.try_get::>("", column).ok().flatten() != Some(*value as i16) + }) || row + .try_get::>("", "source_domain") + .map_err(internal)? + .as_deref() + != Some("native-git") + || row + .try_get::>("", "verification_revision") + .map_err(internal)? + != Some(super::mono_storage::MST2_VERIFICATION_VERSION) + { + return Err(integrity( + "durable snapshot materialization profile changed", + )); + } + Ok(SnapshotContext { + built: BuiltDescriptor { + instance_id: instance.into(), + snapshot_id: sid.into(), + metadata_root: format!("sha256:{}", hex(&descriptor.metadata_root)), + descriptor, + }, + commit_oid: head.root.commit, + root_tree_oid: head.root.tree, + lease_id: lease.into(), + lease_expires_at_unix: deadline as u64, + authorization_epoch: epoch as u64, + }) +} + +async fn retention_lock(txn: &DatabaseTransaction) -> Result<(), SnapshotError> { + txn.execute_raw(statement( + "SELECT pg_advisory_xact_lock($1,hashtext(current_schema()))", + [RETENTION_LOCK_KEY.into()], + )) + .await + .map_err(internal) + .map(|_| ()) +} + +async fn retire_session_pin_locked( + txn: &DatabaseTransaction, + sid: &str, +) -> Result<(), SnapshotError> { + txn.execute_raw(statement( + "DELETE FROM mst2_retention_root r WHERE r.root_kind='pin' AND r.root_key='pin:session:'||$1 + AND NOT EXISTS(SELECT 1 FROM mst2_snapshot_lease l WHERE l.snapshot_id=$1 AND l.state='ACTIVE' + AND l.expires_at_unix>=floor(extract(epoch FROM clock_timestamp()))::bigint)",[sid.into()] + )).await.map_err(internal).map(|_| ()) +} + +async fn expire_specific_locked( + txn: &DatabaseTransaction, + lease: &str, +) -> Result<(), SnapshotError> { + let row=txn.query_one_raw(statement( + "UPDATE mst2_snapshot_lease SET state='EXPIRED' WHERE lease_id=$1 AND state='ACTIVE' + AND expires_at_unix("", "snapshot_id").map_err(internal)?, + ) + .await?; + } + Ok(()) +} + +async fn expire_locked(txn: &DatabaseTransaction) -> Result<(), SnapshotError> { + let rows = txn + .query_all_raw(statement( + "UPDATE mst2_snapshot_lease SET state='EXPIRED' WHERE lease_id IN + (SELECT lease_id FROM mst2_snapshot_lease WHERE state='ACTIVE' + AND expires_at_unix("","lease_id").map_err(internal)?, + "sid":row.try_get::("","snapshot_id").map_err(internal)? + })) + }) + .collect::, SnapshotError>>()?; + let encoded = serde_json::to_string(&rows).map_err(internal)?; + txn.execute_raw(statement( + "DELETE FROM mst2_retention_root r USING jsonb_to_recordset($1::jsonb) AS p(lease text,sid text) + WHERE r.root_kind='lease' AND r.root_key='lease:'||p.lease",[encoded.clone().into()] + )).await.map_err(internal)?; + txn.execute_raw(statement( + "DELETE FROM mst2_retention_root r USING jsonb_to_recordset($1::jsonb) AS p(lease text,sid text) + WHERE r.root_kind='pin' AND r.root_key='pin:session:'||p.sid + AND NOT EXISTS(SELECT 1 FROM mst2_snapshot_lease l WHERE l.snapshot_id=p.sid AND l.state='ACTIVE' + AND l.expires_at_unix>=floor(extract(epoch FROM clock_timestamp()))::bigint)",[encoded.into()] + )).await.map_err(internal).map(|_| ()) +} + +async fn finish( + txn: DatabaseTransaction, + result: Result, +) -> Result { + match result { + Ok(value) => { + txn.commit().await.map_err(|error| { + tracing::error!(error=%error,"native session commit outcome unknown"); + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "session commit outcome unknown; retry resolve", + ) + })?; + Ok(value) + } + Err(error) => { + txn.rollback().await.map_err(internal)?; + Err(error) + } + } +} +fn statement(sql: &str, values: [sea_orm::Value; N]) -> Statement { + Statement::from_sql_and_values(DbBackend::Postgres, sql, values) +} +fn root_node(built: &BuiltDescriptor) -> String { + format!("page:{}", built.metadata_root) +} +fn internal(error: impl std::fmt::Display) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::Internal, error.to_string()) +} +fn integrity(message: &str) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::IntegrityError, message) +} +fn not_ready(message: &str) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::SnapshotNotReady, message) +} +fn expired() -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::LeaseExpired, + "lease is not active for this snapshot; re-resolve", + ) +} +fn forbidden() -> SnapshotError { + SnapshotError::new( + SnapshotErrorCode::ScopeForbidden, + "snapshot serving state or authorization epoch changed", + ) +} +fn install_error(error: MetadataInstallError) -> SnapshotError { + match error { + MetadataInstallError::Rejected(error) if error.code == SnapshotErrorCode::Internal => { + tracing::error!(%error,"native metadata installation unavailable"); + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "native metadata installation unavailable", + ) + } + MetadataInstallError::Rejected(error) => error, + MetadataInstallError::CommitUncertain { + operation_id, + phase, + .. + } => { + tracing::error!(%operation_id,?phase,"native metadata commit outcome unknown"); + SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "metadata commit outcome unknown; retry resolve", + ) + } + } +} + +impl super::Storage { + pub(crate) async fn snapshot_sessions(&self) -> &PostgresNativeSessionRepository { + use super::base_storage::StorageConnector; + self.native_snapshot_sessions + .get_or_init(|| async { + PostgresNativeSessionRepository::new(self.mono_storage().get_connection().clone()) + }) + .await + } + + pub(crate) async fn snapshot_context( + &self, + sid: &str, + lease: &str, + ) -> Result { + let config = self.config(); + if config.mst2.publication_enabled { + let instance = config + .mst2 + .instance_uuid + .as_deref() + .ok_or_else(|| not_ready("native instance missing"))?; + let instance = uuid::Uuid::parse_str(instance) + .map_err(|_| not_ready("invalid native instance"))? + .to_string(); + self.snapshot_sessions() + .await + .context(sid, lease, &instance) + .await + } else { + let runtime = crate::ceres::snapshot::runtime::runtime(); + runtime.validate_lease(sid, lease)?; + runtime.context(sid) + } + } + + pub(crate) async fn snapshot_renew( + &self, + lease: &str, + seconds: u64, + ) -> Result { + let config = self.config(); + if config.mst2.publication_enabled { + let instance = config + .mst2 + .instance_uuid + .as_deref() + .ok_or_else(|| not_ready("native instance missing"))?; + let instance = uuid::Uuid::parse_str(instance) + .map_err(|_| not_ready("invalid native instance"))? + .to_string(); + self.snapshot_sessions() + .await + .renew(lease, seconds, &instance) + .await + } else { + crate::ceres::snapshot::runtime::runtime().renew_lease(lease, seconds) + } + } + + pub(crate) async fn snapshot_release(&self, lease: &str) -> Result { + if self.config().mst2.publication_enabled { + self.snapshot_sessions().await.release(lease).await + } else { + Ok(crate::ceres::snapshot::runtime::runtime().release_lease(lease)) + } + } +} diff --git a/src/jupiter/tests.rs b/src/jupiter/tests.rs index f79aeb41..2b03da07 100644 --- a/src/jupiter/tests.rs +++ b/src/jupiter/tests.rs @@ -378,6 +378,7 @@ pub async fn test_storage_with_config(temp_dir: impl AsRef, config: Config Storage { app_service: Arc::new(svc), native_projection_cache: Arc::default(), + native_snapshot_sessions: Arc::default(), projection_observation_sink: None, cl_service: CLService::mock(), push_queue_service: PushQueueService::new(