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(