diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index 8540f65b..f06fd35b 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -9,12 +9,11 @@ use axum::{ http::HeaderMap, response::{IntoResponse, Response}, }; -use bytes::Bytes; use futures::stream::StreamExt; use serde::Deserialize; use serde_json::json; -use super::{abs_view_path, internal, mst2_error_response, treeframe_response}; +use super::{abs_view_path, internal, mst2_error_response, request::Mst2Bytes, treeframe_response}; use crate::ceres::snapshot::{ chunks::{ChunkProjection, get_or_project}, error::{SnapshotError, SnapshotErrorCode}, @@ -174,7 +173,7 @@ const OBJECT_TOTAL_MAX: usize = 8 * 1024 * 1024; pub(super) async fn objects( state: State, AxumPath(snapshot_id): AxumPath, - body: Bytes, + Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure(&state)?; let ctx = runtime() @@ -469,7 +468,7 @@ struct Planned { pub(super) async fn chunks( state: State, AxumPath(snapshot_id): AxumPath, - body: Bytes, + Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure(&state)?; let ctx = runtime() diff --git a/src/api/router/snapshot_request.rs b/src/api/router/snapshot_request.rs new file mode 100644 index 00000000..ec43af70 --- /dev/null +++ b/src/api/router/snapshot_request.rs @@ -0,0 +1,114 @@ +//! Bounded raw JSON input for the MST/2 POST surface (spec 14). + +use std::{collections::HashSet, fmt, time::Duration}; + +use axum::{ + extract::{FromRequest, Request}, + http::StatusCode, +}; +use bytes::Bytes; +use serde::{ + Deserialize, Deserializer, + de::{self, MapAccess, SeqAccess, Visitor}, +}; + +use crate::ceres::snapshot::error::{SnapshotError, SnapshotErrorCode}; + +/// One overall read deadline, including a body which keeps trickling bytes. +/// This bounds input collection only, not handler work or response streams. +pub(super) const JSON_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +/// Preserve the original bytes for TreeFrame request-body digests. The router +/// supplies DefaultBodyLimit; its rejection is converted to the MST envelope. +pub(super) struct Mst2Bytes(pub(super) Bytes); + +/// Decode keys before comparing them, including escaped spellings. This +/// separate pass also catches duplicate optional fields whose first value is +/// null, which a derived DTO can otherwise treat as an absent field. +pub(super) fn validate_json_keys(body: &[u8]) -> Result<(), serde_json::Error> { + serde_json::from_slice::(body).map(|_| ()) +} + +struct UniqueKeys; + +impl<'de> Deserialize<'de> for UniqueKeys { + fn deserialize>(deserializer: D) -> Result { + deserializer.deserialize_any(UniqueKeyVisitor) + } +} + +struct UniqueKeyVisitor; + +impl<'de> Visitor<'de> for UniqueKeyVisitor { + type Value = UniqueKeys; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("JSON without duplicate object keys") + } + + fn visit_bool(self, _: bool) -> Result { + Ok(UniqueKeys) + } + fn visit_i64(self, _: i64) -> Result { + Ok(UniqueKeys) + } + fn visit_u64(self, _: u64) -> Result { + Ok(UniqueKeys) + } + fn visit_f64(self, _: f64) -> Result { + Ok(UniqueKeys) + } + fn visit_str(self, _: &str) -> Result { + Ok(UniqueKeys) + } + fn visit_unit(self) -> Result { + Ok(UniqueKeys) + } + + fn visit_seq>(self, mut sequence: A) -> Result { + while sequence.next_element::()?.is_some() {} + Ok(UniqueKeys) + } + + fn visit_map>(self, mut object: A) -> Result { + let mut keys = HashSet::new(); + while let Some(key) = object.next_key::()? { + if !keys.insert(key) { + return Err(de::Error::custom("duplicate JSON object key")); + } + object.next_value::()?; + } + Ok(UniqueKeys) + } +} + +impl FromRequest for Mst2Bytes +where + S: Send + Sync, +{ + type Rejection = SnapshotError; + + async fn from_request(req: Request, state: &S) -> Result { + match tokio::time::timeout(JSON_REQUEST_TIMEOUT, Bytes::from_request(req, state)).await { + Ok(Ok(body)) => Ok(Self(body)), + Ok(Err(error)) => { + let (code, message) = if error.status() == StatusCode::PAYLOAD_TOO_LARGE { + ( + SnapshotErrorCode::LimitExceeded, + "request body over the spec 14 limit", + ) + } else { + ( + SnapshotErrorCode::InvalidRequest, + "could not read request body", + ) + }; + Err(SnapshotError::new(code, message)) + } + Err(_) => Err(SnapshotError::new( + SnapshotErrorCode::TemporaryUnavailable, + "request body read deadline exceeded", + )), + } + } +} diff --git a/src/api/router/snapshot_request_tests.rs b/src/api/router/snapshot_request_tests.rs new file mode 100644 index 00000000..fc577c85 --- /dev/null +++ b/src/api/router/snapshot_request_tests.rs @@ -0,0 +1,353 @@ +use std::{ + convert::Infallible, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, + }, + task::{Context, Poll}, + time::Duration, +}; + +use axum::{ + Router, + body::{Body, to_bytes}, + http::Request, + middleware, + response::Response, + routing::get, +}; +use bytes::Bytes; +use futures::{Stream, stream}; +use serde_json::Value; +use tower::ServiceExt; + +use super::{JSON_REQUEST_LIMIT, request::JSON_REQUEST_TIMEOUT, routers}; +use crate::{ + api::{MonoApiServiceState, oauth::api_store::BrowserSessionStore}, + ceres::{ + api_service::cache::GitObjectCache, + snapshot::{descriptor::build, runtime::runtime, view::SnapshotView}, + }, + config::testing::isolated_config, + jupiter::tests::{test_redis_manager, test_storage_with_config}, + server::trace_context::{TraceContext, inject_trace_context}, +}; + +struct Fixture { + app: Router, + snapshot_id: String, + lease_id: String, + expires: u64, + _temp: tempfile::TempDir, +} + +impl Fixture { + async fn new() -> Self { + let temp = tempfile::TempDir::new().unwrap(); + let mut config = isolated_config(temp.path().join("config")); + config.mst2.enabled = true; + config.mst2.instance_uuid = Some(uuid::Uuid::new_v4().to_string()); + config.mst2.auth_token = Some("mst2-input-test".to_string()); + let storage = test_storage_with_config(temp.path(), config).await; + let state = MonoApiServiceState { + entity_store: storage.entity_store.clone(), + storage, + session_store: BrowserSessionStore::Anonymous, + git_object_cache: Arc::new(GitObjectCache { + connection: test_redis_manager().await, + prefix: String::new(), + }), + listen_addr: "127.0.0.1:0".to_string(), + }; + let view = SnapshotView::from_commit(&"1".repeat(40), &"2".repeat(40)); + let built = build(&state.storage.config().mst2, &view, "/", [3; 32]).unwrap(); + let ctx = runtime() + .insert_context(built, &view.commit_oid, &view.root_tree_oid, 60) + .expect("fixture lease registration must succeed"); + // Reproduce the server's nest-after-layer order: MST must establish + // its own request context even when the earlier layer does not run. + let app = Router::new() + .route("/outside", get(|| async { "ok" })) + .layer(middleware::from_fn(inject_trace_context)) + .nest("/api/v2", routers(state.clone()).with_state(state)); + Self { + app, + snapshot_id: ctx.built.snapshot_id, + lease_id: ctx.lease_id, + expires: ctx.lease_expires_at_unix, + _temp: temp, + } + } + + fn request(&self, suffix: &str, body: Body) -> Request { + Request::builder() + .method("POST") + .uri(format!("/api/v2/snapshots/{suffix}")) + .header("authorization", "Bearer mst2-input-test") + .header("x-mega-snapshot-lease", &self.lease_id) + .header("content-type", "application/json") + .body(body) + .unwrap() + } + + fn renew_path(&self) -> String { + format!("leases/{}/renew", self.lease_id) + } + + fn assert_lease_unchanged(&self) { + let ctx = runtime().context(&self.snapshot_id).unwrap(); + assert_eq!(ctx.lease_id, self.lease_id); + assert_eq!(ctx.lease_expires_at_unix, self.expires); + } +} + +async fn assert_error(response: Response, status: u16, code: &str, retryable: bool) -> Value { + assert_eq!(response.status().as_u16(), status); + assert_eq!(response.headers()["content-type"], "application/json"); + let request_id = response.headers()["x-request-id"] + .to_str() + .unwrap() + .to_string(); + assert!(!request_id.is_empty()); + let body = to_bytes(response.into_body(), 16 * 1024).await.unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["code"], code); + assert_eq!(value["error"]["request_id"], request_id); + assert_eq!(value["error"]["retryable"], retryable); + assert!(!value["error"]["message"].as_str().unwrap().is_empty()); + value +} + +fn chunked(bytes: Vec) -> Body { + let chunks: Vec> = bytes + .chunks(4096) + .map(|chunk| Ok(Bytes::copy_from_slice(chunk))) + .collect(); + Body::from_stream(stream::iter(chunks)) +} + +#[tokio::test] +async fn mst2_json_actual_chunked_limit_has_typed_errors_on_every_post() { + let fixture = Fixture::new().await; + let paths = [ + "resolve".to_string(), + fixture.renew_path(), + format!("{}/lookup", fixture.snapshot_id), + format!("{}/metadata/pages", fixture.snapshot_id), + format!("{}/objects", fixture.snapshot_id), + format!("{}/chunks", fixture.snapshot_id), + ]; + for path in paths { + for declared in [None, Some("2")] { + let mut request = fixture.request(&path, chunked(vec![b' '; JSON_REQUEST_LIMIT + 1])); + if let Some(length) = declared { + request + .headers_mut() + .insert("content-length", length.parse().unwrap()); + } + let response = fixture.app.clone().oneshot(request).await.unwrap(); + assert_error(response, 413, "LIMIT_EXCEEDED", false).await; + fixture.assert_lease_unchanged(); + } + } + + let mut body = br#"{"lease_seconds":120}"#.to_vec(); + body.resize(JSON_REQUEST_LIMIT, b' '); + let response = fixture + .app + .clone() + .oneshot(fixture.request(&fixture.renew_path(), chunked(body))) + .await + .unwrap(); + assert_eq!(response.status(), 200); + assert!(!response.headers()["x-request-id"].is_empty()); + let renewed = runtime().context(&fixture.snapshot_id).unwrap(); + assert!(renewed.lease_expires_at_unix > fixture.expires); +} + +#[tokio::test] +async fn mst2_json_declared_oversize_is_rejected_without_reading_body() { + let fixture = Fixture::new().await; + let polls = Arc::new(AtomicUsize::new(0)); + let observed = polls.clone(); + let body = Body::from_stream(stream::poll_fn(move |_| { + observed.fetch_add(1, Ordering::SeqCst); + Poll::>>::Pending + })); + let mut request = fixture.request(&fixture.renew_path(), body); + request.headers_mut().insert( + "content-length", + (JSON_REQUEST_LIMIT + 1).to_string().parse().unwrap(), + ); + let response = fixture.app.clone().oneshot(request).await.unwrap(); + assert_error(response, 413, "LIMIT_EXCEEDED", false).await; + assert_eq!(polls.load(Ordering::SeqCst), 0); + fixture.assert_lease_unchanged(); +} + +#[tokio::test] +async fn mst2_json_closed_dtos_duplicates_and_malformed_input_do_not_renew() { + let fixture = Fixture::new().await; + let renew = fixture.renew_path(); + let lookup = format!("{}/lookup", fixture.snapshot_id); + let metadata = format!("{}/metadata/pages", fixture.snapshot_id); + let objects = format!("{}/objects", fixture.snapshot_id); + let chunks = format!("{}/chunks", fixture.snapshot_id); + let cases: Vec<(&str, &[u8])> = vec![ + ("resolve", br#"{"target":{"kind":"latest"},"scope":"/","scope":"/"}"#), + ("resolve", br#"{"target":{"kind":"latest","kind":"latest"}}"#), + ("resolve", br#"{"target":{"kind":"latest","extra":true}}"#), + ("resolve", br#"{"target":{"kind":"latest"},"extra":true}"#), + (&renew, br#"{"lease_seconds":3600,"lease_seconds":120}"#), + (&renew, br#"{"lease_seconds":3600,"\u006cease_seconds":120}"#), + (&renew, br#"{"lease_seconds":null,"lease_seconds":120}"#), + (&renew, br#"{"lease_seconds":null,"\u006cease_seconds":120}"#), + (&renew, br#"{"lease_seconds":null,"lease_seconds":null}"#), + (&renew, br#"{"lease_seconds":3600,"extra":true}"#), + (&renew, br#"{"lease_seconds":"120"}"#), + (&renew, br#"{"lease_seconds":1.5}"#), + (&renew, br#"{"lease_seconds":-1}"#), + (&renew, br#"{"lease_seconds":NaN}"#), + (&renew, br#"{"lease_seconds":Infinity}"#), + (&renew, br#"{"lease_seconds":18446744073709551616}"#), + (&renew, br#"{"lease_seconds":"#), + (&renew, b"{\"lease_seconds\":\xff}"), + (&renew, b"{\"lease_seconds\":3600} {}"), + (&renew, b"{\"lease_seconds\":3600} x"), + (&lookup, br#"{"paths":[],"paths":[]}"#), + (&lookup, br#"{"paths":[],"extra":true}"#), + (&metadata, br#"{"items":[],"items":[]}"#), + (&metadata, br#"{"items":[],"extra":true}"#), + (&metadata, br#"{"items":[{"directory_path":"/","route":[],"route":[]}]}"#), + (&metadata, br#"{"items":[{"directory_path":"/","extra":true}]}"#), + (&objects, br#"{"items":[],"items":[]}"#), + (&objects, br#"{"items":[],"extra":true}"#), + (&objects, br#"{"items":[{"path":"/a","path":"/b","expected_digest":"x"}]}"#), + (&objects, br#"{"items":[{"path":"/a","expected_digest":"x","extra":true}]}"#), + (&chunks, br#"{"items":[],"items":[]}"#), + (&chunks, br#"{"items":[],"extra":true}"#), + (&chunks, br#"{"items":[{"path":"/a","expected_digest":"x","map_id":"x","chunk_index":"0","chunk_index":"1"}]}"#), + (&chunks, br#"{"items":[{"path":"/a","expected_digest":"x","map_id":"x","chunk_index":"0","extra":true}]}"#), + ]; + for (path, body) in cases { + let response = fixture + .app + .clone() + .oneshot(fixture.request(path, Body::from(body.to_vec()))) + .await + .unwrap(); + assert_error(response, 400, "INVALID_REQUEST", false).await; + fixture.assert_lease_unchanged(); + } + + for body in [Body::empty(), Body::from("{} \r\n\t")] { + let response = fixture + .app + .clone() + .oneshot(fixture.request(&renew, body)) + .await + .unwrap(); + assert_eq!(response.status(), 200); + assert!(!response.headers()["x-request-id"].is_empty()); + } +} + +struct SlowBody { + interval: Option, + chunks: Arc, + dropped: Arc, +} + +impl Stream for SlowBody { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + match &mut this.interval { + Some(interval) => match interval.poll_tick(cx) { + Poll::Ready(_) => { + this.chunks.fetch_add(1, Ordering::SeqCst); + Poll::Ready(Some(Ok(Bytes::from_static(b" ")))) + } + Poll::Pending => Poll::Pending, + }, + None => Poll::Pending, + } + } +} + +impl Drop for SlowBody { + fn drop(&mut self) { + self.dropped.store(true, Ordering::SeqCst); + } +} + +#[tokio::test] +async fn mst2_json_real_post_pending_and_trickle_bodies_hit_overall_deadline() { + let fixture = Fixture::new().await; + let exercise = |trickle: bool| { + let app = fixture.app.clone(); + let dropped = Arc::new(AtomicBool::new(false)); + let chunks = Arc::new(AtomicUsize::new(0)); + let body = Body::from_stream(SlowBody { + interval: trickle.then(|| tokio::time::interval(Duration::from_millis(50))), + chunks: chunks.clone(), + dropped: dropped.clone(), + }); + let request = fixture.request(&fixture.renew_path(), body); + async move { + let response = tokio::time::timeout( + JSON_REQUEST_TIMEOUT + Duration::from_secs(5), + app.oneshot(request), + ) + .await + .expect("MST input deadline must terminate the request") + .unwrap(); + assert_error(response, 503, "TEMPORARY_UNAVAILABLE", true).await; + assert!(dropped.load(Ordering::SeqCst)); + if trickle { + assert!(chunks.load(Ordering::SeqCst) > 1); + } else { + assert_eq!(chunks.load(Ordering::SeqCst), 0); + } + } + }; + tokio::join!(exercise(false), exercise(true)); + fixture.assert_lease_unchanged(); + let response = fixture + .app + .clone() + .oneshot(fixture.request(&fixture.renew_path(), Body::from("{}"))) + .await + .unwrap(); + assert_eq!(response.status(), 200); +} + +#[tokio::test] +async fn mst2_json_body_read_failure_is_typed_and_preserves_request_context() { + let fixture = Fixture::new().await; + let body = Body::from_stream(stream::once(async { + Err::(std::io::Error::other("private backend key must not leak")) + })); + let mut request = fixture.request(&fixture.renew_path(), body); + request + .headers_mut() + .insert("x-request-id", "different-inbound-id".parse().unwrap()); + request.extensions_mut().insert(TraceContext { + trace_id: Arc::from("existing-trace-id"), + }); + let response = fixture.app.clone().oneshot(request).await.unwrap(); + let value = assert_error(response, 400, "INVALID_REQUEST", false).await; + assert_eq!(value["error"]["request_id"], "existing-trace-id"); + assert_eq!(value["error"]["message"], "could not read request body"); + fixture.assert_lease_unchanged(); + + let mut request = fixture.request(&fixture.renew_path(), Body::from("{")); + request + .headers_mut() + .insert("x-request-id", "accepted-inbound-id".parse().unwrap()); + let response = fixture.app.clone().oneshot(request).await.unwrap(); + let value = assert_error(response, 400, "INVALID_REQUEST", false).await; + assert_eq!(value["error"]["request_id"], "accepted-inbound-id"); +} diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 7e8f9e0e..9330837b 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -16,6 +16,7 @@ use axum::{ use base64::Engine; use bytes::Bytes; use mst2_codec::descriptor; +use request::Mst2Bytes; use serde::Deserialize; use serde_json::json; use sha2::{Digest, Sha256}; @@ -73,6 +74,13 @@ pub fn routers(api_state: MonoApiServiceState) -> Router { #[path = "snapshot_content.rs"] mod content; +#[path = "snapshot_request.rs"] +mod request; + +#[cfg(test)] +#[path = "snapshot_request_tests.rs"] +mod request_tests; + /// Spec 14 §4: JSON request bytes hard limit. pub(crate) const JSON_REQUEST_LIMIT: usize = 131_072; @@ -141,28 +149,36 @@ const LEASE_HEADER: &str = "x-mega-snapshot-lease"; /// knowing the snapshot id alone is not a capability. async fn snapshot_auth_middleware( State(state): State, - req: axum::extract::Request, + mut req: axum::extract::Request, next: axum::middleware::Next, ) -> Response { - // Same id the global trace layer echoes on responses and logs. + // Nested routes may be added after the server's trace layer. Reuse an + // existing context or establish one here, including rejection responses. let id = req .extensions() .get::() - .map(|c| c.trace_id.to_string()) - .unwrap_or_default(); - REQUEST_ID - .scope(id, async { + .map(|c| c.trace_id.clone()) + .unwrap_or_else(|| crate::server::trace_context::resolve_trace_id(req.headers())); + req.extensions_mut() + .insert(crate::server::trace_context::TraceContext { + trace_id: id.clone(), + }); + 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 }) - .await + .await; + if let Ok(value) = HeaderValue::from_str(&id) { + response.headers_mut().insert("x-request-id", value); + } + response } -/// Spec 14 §4 enforcement with the MST/2 error envelope (DefaultBodyLimit's -/// own rejection is plain-text). Content-Length is checked here; a lying -/// chunked body still trips DefaultBodyLimit inside the extractor. +/// Reject an oversized declared length before consuming any body. Actual +/// bytes and the overall read deadline are checked by `Mst2Bytes`. async fn reject_oversize_body( req: axum::extract::Request, next: axum::middleware::Next, @@ -186,12 +202,14 @@ async fn reject_oversize_body( /// (spec 14 §5 INVALID_REQUEST). Size is enforced by the router layers. #[allow(clippy::result_large_err)] pub(crate) fn parse_json_body(body: &Bytes) -> Result { - serde_json::from_slice(body).map_err(|e| { - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::InvalidRequest, - format!("malformed request body: {e}"), - )) - }) + request::validate_json_keys(body) + .and_then(|()| serde_json::from_slice(body)) + .map_err(|e| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::InvalidRequest, + format!("malformed request body: {e}"), + )) + }) } /// Bearer-token check shared by the auth middleware; the parsing rule is the @@ -260,7 +278,9 @@ fn mst2_error_response(err: SnapshotError) -> Response { "request_id": current_request_id(), "retryable": matches!( err.code, - SnapshotErrorCode::SnapshotNotReady | SnapshotErrorCode::Internal + SnapshotErrorCode::SnapshotNotReady + | SnapshotErrorCode::TemporaryUnavailable + | SnapshotErrorCode::Internal ), } })), @@ -274,6 +294,12 @@ impl From for Response { } } +impl IntoResponse for SnapshotError { + fn into_response(self) -> Response { + mst2_error_response(self) + } +} + /// MegaError → snapshot error: storage/lease failures must surface as errors, /// never as absence (spec 00 SYS-04). fn internal(e: E) -> SnapshotError { @@ -363,7 +389,10 @@ fn ensure_enabled(state: &MonoApiServiceState) -> Result<(), SnapshotError> { // `lfs_router::enforce_lfs_access` (where the error is the rare arm and is // boxed), there is nothing to gain here, so the lint is allowed outright. #[allow(clippy::result_large_err)] -async fn resolve(state: State, body: Bytes) -> Result { +async fn resolve( + state: State, + Mst2Bytes(body): Mst2Bytes, +) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; let req: ResolveRequest = parse_json_body(&body)?; // Unknown target kinds are client errors, never a silent fallback to @@ -540,18 +569,13 @@ struct RenewRequest { async fn lease_renew( state: State, AxumPath(lease_id): AxumPath, - body: Bytes, + Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; let req: RenewRequest = if body.is_empty() { RenewRequest::default() } else { - serde_json::from_slice(&body).map_err(|e| { - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::ScopeInvalid, - format!("malformed renew body: {e}"), - )) - })? + parse_json_body(&body)? }; let renewed = runtime() .renew_lease(&lease_id, req.lease_seconds.unwrap_or(600)) @@ -911,7 +935,7 @@ async fn lookup( state: State, AxumPath(snapshot_id): AxumPath, _headers: HeaderMap, - body: Bytes, + Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; let req: LookupRequest = parse_json_body(&body)?; @@ -1068,7 +1092,7 @@ async fn metadata_pages( state: State, AxumPath(snapshot_id): AxumPath, _headers: HeaderMap, - body: Bytes, + Mst2Bytes(body): Mst2Bytes, ) -> Result { ensure_enabled(&state).map_err(mst2_error_response)?; let req: MetadataPagesRequest = parse_json_body(&body)?; diff --git a/src/callisto/mod.rs b/src/callisto/mod.rs index 8d8e5440..ba9673e6 100644 --- a/src/callisto/mod.rs +++ b/src/callisto/mod.rs @@ -63,6 +63,10 @@ pub mod mst2_native_publication; pub mod mst2_publication; pub mod mst2_publication_outbox; pub mod mst2_queue_noop_receipt; +pub mod mst2_retention_edge; +pub mod mst2_retention_gc_op; +pub mod mst2_retention_node; +pub mod mst2_retention_root; pub mod mst2_verified_object; pub mod notification_event_types; pub mod oci_blob_ref; diff --git a/src/callisto/mst2_retention_edge.rs b/src/callisto/mst2_retention_edge.rs new file mode 100644 index 00000000..2c861bdf --- /dev/null +++ b/src/callisto/mst2_retention_edge.rs @@ -0,0 +1,19 @@ +//! MST/2 retention graph edges (spec 10 §6). + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mst2_retention_edge")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = true)] + pub id: i64, + pub parent_id: String, + pub child_id: String, + pub created_at: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/mst2_retention_gc_op.rs b/src/callisto/mst2_retention_gc_op.rs new file mode 100644 index 00000000..833c949d --- /dev/null +++ b/src/callisto/mst2_retention_gc_op.rs @@ -0,0 +1,26 @@ +//! Durable MST/2 GC operation log (spec 10 §6 crash replay). + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mst2_retention_gc_op")] +pub struct Model { + /// Caller supplied idempotency key. A retry of one physical GC step must + /// address the same row, so replay cannot decrement an edge twice. + #[sea_orm(primary_key, auto_increment = false)] + pub operation_id: String, + pub node_id: String, + /// `MARK_DELETING` or `REMOVE`. + pub operation: String, + /// `PENDING`, `APPLIED`, or `FAILED`; workers may replay `PENDING`. + pub state: String, + pub attempts: i32, + pub created_at: DateTimeWithTimeZone, + pub completed_at: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/mst2_retention_node.rs b/src/callisto/mst2_retention_node.rs new file mode 100644 index 00000000..40ce2999 --- /dev/null +++ b/src/callisto/mst2_retention_node.rs @@ -0,0 +1,25 @@ +//! MST/2 retention graph nodes (spec 10 §6). + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mst2_retention_node")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub node_id: String, + pub kind: String, + pub state: String, + #[sea_orm(column_type = "BigInteger")] + pub bytes: i64, + /// Number of unique incoming retention edges. Root coverage is kept in + /// `mst2_retention_root` and checked in the same transaction. + #[sea_orm(column_type = "BigInteger")] + pub incoming_refs: i64, + pub created_at: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/mst2_retention_root.rs b/src/callisto/mst2_retention_root.rs new file mode 100644 index 00000000..9b60cb5a --- /dev/null +++ b/src/callisto/mst2_retention_root.rs @@ -0,0 +1,20 @@ +//! MST/2 retention roots (spec 10 §5/§6). + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mst2_retention_root")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = true)] + pub id: i64, + pub node_id: String, + pub root_key: String, + pub root_kind: String, + pub created_at: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/src/callisto/prelude.rs b/src/callisto/prelude.rs index 7a4cd099..33e929b9 100644 --- a/src/callisto/prelude.rs +++ b/src/callisto/prelude.rs @@ -20,6 +20,10 @@ pub use super::{ mega_tag::Entity as MegaTag, mega_tree::Entity as MegaTree, mega_webhook::Entity as MegaWebhook, mega_webhook_delivery::Entity as MegaWebhookDelivery, mega_webhook_event_type::Entity as MegaWebhookEventType, + mst2_retention_edge::Entity as Mst2RetentionEdge, + mst2_retention_gc_op::Entity as Mst2RetentionGcOp, + mst2_retention_node::Entity as Mst2RetentionNode, + mst2_retention_root::Entity as Mst2RetentionRoot, notification_event_types::Entity as NotificationEventTypes, oci_blob_ref::Entity as OciBlobRef, oci_manifest::Entity as OciManifest, oci_tag::Entity as OciTag, oci_upload::Entity as OciUpload, path_check_configs::Entity as PathCheckConfigs, diff --git a/src/ceres/snapshot/error.rs b/src/ceres/snapshot/error.rs index 4fe54274..b3b77296 100644 --- a/src/ceres/snapshot/error.rs +++ b/src/ceres/snapshot/error.rs @@ -18,6 +18,8 @@ pub enum SnapshotErrorCode { ScopeForbidden, ViewNotFound, SnapshotNotReady, + /// A bounded operation could not complete in time (spec 14 §5). + TemporaryUnavailable, /// The fixed view no longer exists: spec 14 §5 SNAPSHOT_GONE (410). SnapshotGone, PathNotFound, @@ -49,6 +51,7 @@ impl SnapshotErrorCode { SnapshotErrorCode::ScopeForbidden => "SCOPE_FORBIDDEN", SnapshotErrorCode::ViewNotFound => "VIEW_NOT_FOUND", SnapshotErrorCode::SnapshotNotReady => "SNAPSHOT_NOT_READY", + SnapshotErrorCode::TemporaryUnavailable => "TEMPORARY_UNAVAILABLE", SnapshotErrorCode::SnapshotGone => "SNAPSHOT_GONE", SnapshotErrorCode::PathNotFound => "PATH_NOT_FOUND", SnapshotErrorCode::NotDirectory => "NOT_DIRECTORY", @@ -82,7 +85,7 @@ impl SnapshotErrorCode { SnapshotErrorCode::ViewNotFound | SnapshotErrorCode::PathNotFound | SnapshotErrorCode::LeaseUnknown => 404, - SnapshotErrorCode::SnapshotNotReady => 503, + SnapshotErrorCode::SnapshotNotReady | SnapshotErrorCode::TemporaryUnavailable => 503, SnapshotErrorCode::ObjectUnavailable => 503, SnapshotErrorCode::NotDirectory | SnapshotErrorCode::Conflict diff --git a/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs b/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs new file mode 100644 index 00000000..8f223af1 --- /dev/null +++ b/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs @@ -0,0 +1,119 @@ +//! T06-B: durable retention counters and replayable GC operations. +//! +//! The original T06 graph migration intentionally supplied only the graph +//! rows. This forward-only migration adds the counter used by the atomic +//! LIVE→DELETING CAS and an idempotent operation log for crash recovery. The +//! runtime is not switched to this schema by this migration alone. + +use sea_orm_migration::{prelude::*, schema::*}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .alter_table( + Table::alter() + .table(Mst2RetentionNode::Table) + .add_column( + big_integer(Mst2RetentionNode::IncomingRefs) + .not_null() + .default(0), + ) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_mst2_retention_node_gc") + .table(Mst2RetentionNode::Table) + .col(Mst2RetentionNode::State) + .col(Mst2RetentionNode::IncomingRefs) + .to_owned(), + ) + .await?; + + manager + .create_table( + Table::create() + .table(Mst2RetentionGcOp::Table) + .if_not_exists() + .col( + string(Mst2RetentionGcOp::OperationId) + .not_null() + .primary_key(), + ) + .col(string(Mst2RetentionGcOp::NodeId).not_null()) + .col(string(Mst2RetentionGcOp::Operation).not_null()) + .col( + string(Mst2RetentionGcOp::State) + .not_null() + .default("PENDING"), + ) + .col(integer(Mst2RetentionGcOp::Attempts).not_null().default(0)) + .col( + timestamp_with_time_zone(Mst2RetentionGcOp::CreatedAt) + .not_null() + .default(Expr::current_timestamp()), + ) + .col(timestamp_with_time_zone_null( + Mst2RetentionGcOp::CompletedAt, + )) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_mst2_retention_gc_op_pending") + .table(Mst2RetentionGcOp::Table) + .col(Mst2RetentionGcOp::State) + .col(Mst2RetentionGcOp::CreatedAt) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_mst2_retention_gc_op_node") + .table(Mst2RetentionGcOp::Table) + .col(Mst2RetentionGcOp::NodeId) + .to_owned(), + ) + .await + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + // Forward-only: retention state and replay evidence are never + // dropped automatically. + Ok(()) + } +} + +#[derive(DeriveIden)] +enum Mst2RetentionNode { + Table, + State, + IncomingRefs, +} + +#[derive(DeriveIden)] +enum Mst2RetentionGcOp { + Table, + OperationId, + NodeId, + Operation, + State, + Attempts, + CreatedAt, + CompletedAt, +} diff --git a/src/jupiter/migration/m20261005_000200_harden_mst2_retention_graph.rs b/src/jupiter/migration/m20261005_000200_harden_mst2_retention_graph.rs new file mode 100644 index 00000000..c42bcd28 --- /dev/null +++ b/src/jupiter/migration/m20261005_000200_harden_mst2_retention_graph.rs @@ -0,0 +1,76 @@ +//! T06-B: backfill graph counters and reject invalid durable graph state. +//! +//! Applies after the additive schema migration. Existing edges must form a +//! valid DAG and counters are derived from unique edges before any collector +//! can use them. Constraints keep failed writers from silently dangling a +//! node or underflowing a counter. Completed GC receipts intentionally do not +//! reference the removed node through a foreign key. + +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> { + let conn = manager.get_connection(); + conn.execute_unprepared( + "LOCK TABLE mst2_retention_node, mst2_retention_edge, mst2_retention_root, \ + mst2_retention_gc_op IN ACCESS EXCLUSIVE MODE", + ) + .await?; + let cycle = conn + .query_one_raw(sea_orm::Statement::from_string( + sea_orm::DbBackend::Postgres, + "WITH RECURSIVE walk(start_id, node_id) AS ( \ + SELECT parent_id, child_id FROM mst2_retention_edge \ + UNION SELECT w.start_id, e.child_id FROM walk w \ + JOIN mst2_retention_edge e ON e.parent_id = w.node_id \ + ) SELECT start_id FROM walk WHERE start_id = node_id LIMIT 1", + )) + .await?; + if cycle.is_some() { + return Err(DbErr::Migration( + "MST/2 retention graph contains a cycle".into(), + )); + } + conn.execute_unprepared( + "UPDATE mst2_retention_node n SET incoming_refs = \ + (SELECT count(*) FROM mst2_retention_edge e WHERE e.child_id = n.node_id)", + ) + .await?; + conn.execute_unprepared( + "ALTER TABLE mst2_retention_node \ + ADD CONSTRAINT mst2_retention_node_state_check CHECK (state IN ('LIVE', 'DELETING')), \ + ADD CONSTRAINT mst2_retention_node_kind_check \ + CHECK (kind IN ('page', 'chunk_map', 'frame', 'verified_object')), \ + ADD CONSTRAINT mst2_retention_node_nonnegative CHECK (bytes >= 0 AND incoming_refs >= 0); \ + ALTER TABLE mst2_retention_edge \ + ADD CONSTRAINT mst2_retention_edge_parent_fk FOREIGN KEY (parent_id) \ + REFERENCES mst2_retention_node(node_id), \ + ADD CONSTRAINT mst2_retention_edge_child_fk FOREIGN KEY (child_id) \ + REFERENCES mst2_retention_node(node_id), \ + ADD CONSTRAINT mst2_retention_edge_no_self CHECK (parent_id <> child_id); \ + ALTER TABLE mst2_retention_root \ + ADD CONSTRAINT mst2_retention_root_node_fk FOREIGN KEY (node_id) \ + REFERENCES mst2_retention_node(node_id), \ + ADD CONSTRAINT mst2_retention_root_kind_check CHECK (root_kind IN ('lease', 'pin', 'prepare')); \ + ALTER TABLE mst2_retention_gc_op \ + ADD CONSTRAINT mst2_retention_gc_op_kind_check CHECK (operation IN ('MARK_DELETING', 'REMOVE')), \ + ADD CONSTRAINT mst2_retention_gc_op_state_check CHECK (state IN ('PENDING', 'APPLIED', 'FAILED')), \ + ADD CONSTRAINT mst2_retention_gc_op_attempts_check CHECK (attempts >= 0), \ + ADD CONSTRAINT mst2_retention_gc_op_completion_check \ + CHECK ((state = 'APPLIED') = (completed_at IS NOT NULL)); \ + CREATE UNIQUE INDEX idx_mst2_retention_gc_op_pending_node \ + ON mst2_retention_gc_op(node_id) WHERE state = 'PENDING'", + ) + .await + .map(|_| ()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + // Forward-only: preserve retained nodes and GC recovery evidence. + Ok(()) + } +} diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 39078944..b385d106 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -140,7 +140,9 @@ mod m20260923_000100_import_repo_cleanups; mod m20260923_000200_canonicalize_import_repo_paths; mod m20260925_000100_media_paging; mod m20261005_000100_add_mst2_publication_request_digest; +mod m20261005_000100_add_mst2_retention_durability; mod m20261005_000200_add_mst2_native_head; +mod m20261005_000200_harden_mst2_retention_graph; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -270,6 +272,8 @@ impl MigratorTrait for Migrator { Box::new(m20260925_000100_media_paging::Migration), Box::new(m20261005_000100_add_mst2_publication_request_digest::Migration), Box::new(m20261005_000200_add_mst2_native_head::Migration), + Box::new(m20261005_000100_add_mst2_retention_durability::Migration), + Box::new(m20261005_000200_harden_mst2_retention_graph::Migration), ] } } @@ -479,15 +483,17 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 5..], + &names[names.len() - 7..], &[ "m20260923_000100_import_repo_cleanups".to_string(), "m20260923_000200_canonicalize_import_repo_paths".to_string(), "m20260925_000100_media_paging".to_string(), "m20261005_000100_add_mst2_publication_request_digest".to_string(), "m20261005_000200_add_mst2_native_head".to_string(), + "m20261005_000100_add_mst2_retention_durability".to_string(), + "m20261005_000200_harden_mst2_retention_graph".to_string(), ], - "native publication registered after media paging and request receipts" + "retention durability registered last, after media paging" ); let db = alias_db().await; diff --git a/src/jupiter/storage/mod.rs b/src/jupiter/storage/mod.rs index 10d4cd63..8ce75f10 100644 --- a/src/jupiter/storage/mod.rs +++ b/src/jupiter/storage/mod.rs @@ -17,6 +17,7 @@ pub mod lfs_db_storage; pub mod media_paging_storage; pub mod mono_storage; pub(crate) mod mst2_publication_storage; +pub mod mst2_retention; pub(crate) mod native_publication_storage; pub mod notification_storage; pub mod object_storage; diff --git a/src/jupiter/storage/mst2_retention.rs b/src/jupiter/storage/mst2_retention.rs new file mode 100644 index 00000000..fe916803 --- /dev/null +++ b/src/jupiter/storage/mst2_retention.rs @@ -0,0 +1,633 @@ +//! Asynchronous PostgreSQL retention graph repository (T06-B, spec 10 §6). +//! +//! This repository is not yet wired into the in-process snapshot runtime. +//! Graph mutations serialize on a schema-scoped transaction advisory lock; +//! callers can include them in a publication transaction. Each mutation uses +//! a savepoint, so a rejected group cannot leave partial data in that caller's +//! READ COMMITTED transaction. Bytes must be complete and verified before +//! retaining a node. +//! +//! A GC claim commits DELETING and a PENDING REMOVE intent together. A worker +//! may repeat the idempotent physical deletion after a crash, then call +//! `complete_gc`. The outgoing edges, child counters and APPLIED receipt commit +//! together, so replay cannot subtract references twice. No method here deletes +//! bytes, and shared Git raw deletion remains disabled until GitRetentionPort +//! is integrated. Physical workers and durable lease/publication wiring remain +//! separate integration work. Removed identities are tombstoned by their GC +//! receipts: rebuilding one requires future generation/fencing integration, +//! otherwise a stale physical worker could delete its newly reconstructed bytes. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; + +use sea_orm::{ + ColumnTrait, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, EntityTrait, + QueryFilter, QueryOrder, QuerySelect, Statement, TransactionTrait, +}; +use serde_json::json; + +use crate::{ + callisto::{mst2_retention_gc_op, mst2_retention_node}, + ceres::snapshot::{ + error::{SnapshotError, SnapshotErrorCode}, + retention::{NodeState, RetentionEdge, RetentionNode, RetentionRoot}, + }, +}; + +const MAX_NODES: usize = 4096; +const MAX_EDGES: usize = 16_384; +const MAX_ROOTS: usize = 16; +const MAX_PENDING_BATCH: u64 = 1000; +const RETENTION_LOCK_KEY: i32 = 1_296_717_362; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GcClaim { + /// This transaction marked the node and recorded its pending operation. + Marked, + /// The same operation is committed and awaits physical deletion/ack. + Pending, + /// The same operation already removed the node and released its edges. + Applied, + /// A root, an incoming edge, another claim or absence prevents collection. + Unavailable, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GcCompletion { + pub replayed: bool, + /// Newly zero-reference children; roots must still be checked when claimed. + pub zero_reference_children: Vec, +} + +#[derive(Clone)] +pub struct PostgresRetentionRepository { + connection: DatabaseConnection, +} + +impl PostgresRetentionRepository { + pub fn new(connection: DatabaseConnection) -> Self { + Self { connection } + } + + pub async fn node( + &self, + id: &str, + ) -> Result, SnapshotError> { + mst2_retention_node::Entity::find_by_id(id.to_owned()) + .one(&self.connection) + .await + .map_err(internal) + } + + /// Bounded replay scan. PENDING intents survive process/worker replacement. + pub async fn pending_gc( + &self, + limit: u64, + ) -> Result, SnapshotError> { + if limit == 0 || limit > MAX_PENDING_BATCH { + return Err(limit_error("pending GC batch must be 1..=1000")); + } + mst2_retention_gc_op::Entity::find() + .filter(mst2_retention_gc_op::Column::State.eq("PENDING")) + .order_by_asc(mst2_retention_gc_op::Column::CreatedAt) + .order_by_asc(mst2_retention_gc_op::Column::OperationId) + .limit(limit) + .all(&self.connection) + .await + .map_err(internal) + } + + pub async fn retain_group( + &self, + nodes: &[RetentionNode], + edges: &[RetentionEdge], + roots: &[RetentionRoot], + ) -> Result<(), SnapshotError> { + let txn = self.connection.begin().await.map_err(internal)?; + let result = Self::retain_group_in_txn(&txn, nodes, edges, roots).await; + finish(txn, result).await + } + + /// Retain an immutable group atomically. Roots cover each supplied node; + /// callers select that group rather than scanning the reachable graph. + /// An existing parent's edges may only be replayed, not extended; a new + /// identity describes a new graph. + pub async fn retain_group_in_txn( + txn: &DatabaseTransaction, + nodes: &[RetentionNode], + edges: &[RetentionEdge], + roots: &[RetentionRoot], + ) -> Result<(), SnapshotError> { + let group = PreparedGroup::new(nodes, edges, roots)?; + let savepoint = begin_graph(txn).await?; + let result = retain_locked(&savepoint, &group).await; + finish(savepoint, result).await + } + + pub async fn release_root(&self, root: &RetentionRoot) -> Result<(), SnapshotError> { + let txn = self.connection.begin().await.map_err(internal)?; + let result = Self::release_root_in_txn(&txn, root).await; + finish(txn, result).await + } + + pub async fn release_root_in_txn( + txn: &DatabaseTransaction, + root: &RetentionRoot, + ) -> Result<(), SnapshotError> { + let (key, _) = root_identity(root)?; + let savepoint = begin_graph(txn).await?; + let result = savepoint + .execute_raw(statement( + "DELETE FROM mst2_retention_root WHERE root_key = $1", + [key.into()], + )) + .await + .map(|_| ()) + .map_err(internal); + finish(savepoint, result).await + } + + pub async fn mark_deleting( + &self, + operation_id: &str, + node_id: &str, + ) -> Result { + let txn = self.connection.begin().await.map_err(internal)?; + let result = Self::mark_deleting_in_txn(&txn, operation_id, node_id).await; + finish(txn, result).await + } + + /// The LIVE check, zero-reference check, CAS and replay intent share the + /// retention mutation lock. Acquisition either precedes this CAS or fails. + pub async fn mark_deleting_in_txn( + txn: &DatabaseTransaction, + operation_id: &str, + node_id: &str, + ) -> Result { + validate_id(operation_id)?; + validate_id(node_id)?; + let savepoint = begin_graph(txn).await?; + let result = mark_locked(&savepoint, operation_id, node_id).await; + finish(savepoint, result).await + } + + /// Acknowledge an idempotent, durable physical delete. Until this succeeds + /// DELETING parents retain every child. This changes graph rows only. + pub async fn complete_gc(&self, operation_id: &str) -> Result { + let txn = self.connection.begin().await.map_err(internal)?; + let result = Self::complete_gc_in_txn(&txn, operation_id).await; + finish(txn, result).await + } + + pub async fn complete_gc_in_txn( + txn: &DatabaseTransaction, + operation_id: &str, + ) -> Result { + validate_id(operation_id)?; + let savepoint = begin_graph(txn).await?; + let result = complete_locked(&savepoint, operation_id).await; + finish(savepoint, result).await + } +} + +struct PreparedGroup { + nodes_json: String, + edges_json: String, + roots_json: String, +} + +impl PreparedGroup { + fn new( + nodes: &[RetentionNode], + edges: &[RetentionEdge], + roots: &[RetentionRoot], + ) -> Result { + if nodes.len() > MAX_NODES || edges.len() > MAX_EDGES || roots.len() > MAX_ROOTS { + return Err(limit_error("retention group exceeds bounded batch limits")); + } + let mut unique_nodes = BTreeMap::new(); + for node in nodes { + validate_id(&node.id)?; + if node.state != NodeState::Live { + return Err(unavailable("retention acquisition requires LIVE nodes")); + } + let bytes = i64::try_from(node.bytes) + .map_err(|_| limit_error("retention bytes exceed signed database range"))?; + let definition = (node.kind.as_str(), bytes); + if unique_nodes + .insert(node.id.as_str(), definition) + .is_some_and(|old| old != definition) + { + return Err(integrity("conflicting retention node definitions")); + } + } + let mut unique_edges = BTreeSet::new(); + for edge in edges { + validate_id(&edge.parent)?; + validate_id(&edge.child)?; + unique_edges.insert((edge.parent.as_str(), edge.child.as_str())); + } + reject_cycle(&unique_edges)?; + let roots: BTreeSet<_> = roots.iter().map(root_identity).collect::>()?; + Ok(Self { + nodes_json: json!( + unique_nodes + .iter() + .map(|(id, (kind, bytes))| { + json!({"node_id": id, "kind": kind, "bytes": bytes}) + }) + .collect::>() + ) + .to_string(), + edges_json: json!( + unique_edges + .iter() + .map(|(parent, child)| { json!({"parent_id": parent, "child_id": child}) }) + .collect::>() + ) + .to_string(), + roots_json: json!( + roots + .iter() + .map(|(key, kind)| { json!({"root_key": key, "root_kind": kind}) }) + .collect::>() + ) + .to_string(), + }) + } +} + +fn reject_cycle(edges: &BTreeSet<(&str, &str)>) -> Result<(), SnapshotError> { + let mut incoming: BTreeMap<&str, usize> = BTreeMap::new(); + let mut children: BTreeMap<&str, Vec<&str>> = BTreeMap::new(); + for &(parent, child) in edges { + incoming.entry(parent).or_default(); + *incoming.entry(child).or_default() += 1; + children.entry(parent).or_default().push(child); + } + let mut ready: VecDeque<_> = incoming + .iter() + .filter_map(|(&id, &count)| (count == 0).then_some(id)) + .collect(); + let mut processed = 0; + while let Some(id) = ready.pop_front() { + processed += 1; + if let Some(children) = children.get(id) { + for child in children { + if let Some(count) = incoming.get_mut(child) { + *count -= 1; + if *count == 0 { + ready.push_back(child); + } + } + } + } + } + if processed != incoming.len() { + return Err(integrity("retention graph must be acyclic")); + } + Ok(()) +} + +async fn retain_locked( + txn: &DatabaseTransaction, + group: &PreparedGroup, +) -> Result<(), SnapshotError> { + if txn + .query_one_raw(statement( + "SELECT i.node_id FROM jsonb_to_recordset($1::jsonb) AS i(node_id text) \ + WHERE NOT EXISTS (SELECT 1 FROM mst2_retention_node n WHERE n.node_id = i.node_id) \ + AND EXISTS (SELECT 1 FROM mst2_retention_gc_op op WHERE op.node_id = i.node_id) LIMIT 1", + [group.nodes_json.clone().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(unavailable( + "removed node identity requires generation-fenced reconstruction", + )); + } + // Batch comparisons happen before any externally visible commit. An + // identity never changes kind, size or DELETING state on replay. + if txn + .query_one_raw(statement( + "SELECT n.node_id, n.state FROM mst2_retention_node n \ + JOIN jsonb_to_recordset($1::jsonb) AS i(node_id text, kind text, bytes bigint) \ + ON n.node_id = i.node_id \ + WHERE n.state <> 'LIVE' OR n.kind <> i.kind OR n.bytes <> i.bytes LIMIT 1", + [group.nodes_json.clone().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(unavailable( + "existing node is DELETING or conflicts with immutable identity", + )); + } + if txn + .query_one_raw(statement( + "SELECT e.parent_id FROM \ + jsonb_to_recordset($1::jsonb) AS e(parent_id text, child_id text) \ + JOIN mst2_retention_node n ON n.node_id = e.parent_id \ + WHERE NOT EXISTS (SELECT 1 FROM mst2_retention_edge old \ + WHERE old.parent_id = e.parent_id AND old.child_id = e.child_id) LIMIT 1", + [group.edges_json.clone().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(integrity( + "an existing retention parent's edges are immutable", + )); + } + txn.execute_raw(statement( + "INSERT INTO mst2_retention_node (node_id, kind, state, bytes, incoming_refs, created_at) \ + SELECT node_id, kind, 'LIVE', bytes, 0, now() FROM \ + jsonb_to_recordset($1::jsonb) AS i(node_id text, kind text, bytes bigint) \ + ON CONFLICT (node_id) DO NOTHING", + [group.nodes_json.clone().into()], + )) + .await + .map_err(internal)?; + if txn + .query_one_raw(statement( + "SELECT e.parent_id FROM \ + jsonb_to_recordset($1::jsonb) AS e(parent_id text, child_id text) \ + LEFT JOIN mst2_retention_node p ON p.node_id = e.parent_id \ + LEFT JOIN mst2_retention_node c ON c.node_id = e.child_id \ + WHERE p.node_id IS NULL OR c.node_id IS NULL \ + OR p.state <> 'LIVE' OR c.state <> 'LIVE' LIMIT 1", + [group.edges_json.clone().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(unavailable( + "retention edge requires two known LIVE endpoints", + )); + } + txn.execute_raw(statement( + "WITH added AS ( \ + INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) \ + SELECT parent_id, child_id, now() FROM \ + jsonb_to_recordset($1::jsonb) AS e(parent_id text, child_id text) \ + ON CONFLICT (parent_id, child_id) DO NOTHING RETURNING child_id \ + ), delta AS (SELECT child_id, count(*) AS refs FROM added GROUP BY child_id) \ + UPDATE mst2_retention_node n SET incoming_refs = n.incoming_refs + delta.refs \ + FROM delta WHERE n.node_id = delta.child_id", + [group.edges_json.clone().into()], + )) + .await + .map_err(internal)?; + txn.execute_raw(statement( + "INSERT INTO mst2_retention_root (node_id, root_key, root_kind, created_at) \ + SELECT n.node_id, r.root_key, r.root_kind, now() FROM \ + jsonb_to_recordset($1::jsonb) AS n(node_id text) CROSS JOIN \ + jsonb_to_recordset($2::jsonb) AS r(root_key text, root_kind text) \ + ON CONFLICT (node_id, root_key) DO NOTHING", + [ + group.nodes_json.clone().into(), + group.roots_json.clone().into(), + ], + )) + .await + .map_err(internal)?; + Ok(()) +} + +async fn mark_locked( + txn: &DatabaseTransaction, + operation_id: &str, + node_id: &str, +) -> Result { + if let Some(op) = mst2_retention_gc_op::Entity::find_by_id(operation_id.to_owned()) + .one(txn) + .await + .map_err(internal)? + { + if op.node_id != node_id || op.operation != "REMOVE" { + return Err(integrity("GC operation id is bound to different work")); + } + return match op.state.as_str() { + "PENDING" => Ok(GcClaim::Pending), + "APPLIED" => Ok(GcClaim::Applied), + _ => Err(integrity("GC operation is not replayable")), + }; + } + let row = txn + .query_one_raw(statement( + "SELECT n.state, n.incoming_refs, \ + (SELECT count(*) FROM mst2_retention_edge e WHERE e.child_id = n.node_id) AS actual_refs \ + FROM mst2_retention_node n WHERE n.node_id = $1", + [node_id.into()], + )) + .await + .map_err(internal)?; + let Some(row) = row else { + return Ok(GcClaim::Unavailable); + }; + let refs: i64 = row.try_get("", "incoming_refs").map_err(internal)?; + let actual: i64 = row.try_get("", "actual_refs").map_err(internal)?; + if refs != actual || refs < 0 { + return Err(integrity("retention reference audit failed; GC stopped")); + } + if row.try_get::("", "state").map_err(internal)? != "LIVE" || refs != 0 { + return Ok(GcClaim::Unavailable); + } + let changed = txn + .execute_raw(statement( + "UPDATE mst2_retention_node n SET state = 'DELETING' \ + WHERE node_id = $1 AND state = 'LIVE' AND incoming_refs = 0 \ + AND NOT EXISTS (SELECT 1 FROM mst2_retention_root r WHERE r.node_id = n.node_id) \ + AND NOT EXISTS (SELECT 1 FROM mst2_retention_edge e WHERE e.child_id = n.node_id)", + [node_id.into()], + )) + .await + .map_err(internal)?; + if changed.rows_affected() == 0 { + return Ok(GcClaim::Unavailable); + } + txn.execute_raw(statement( + "INSERT INTO mst2_retention_gc_op (operation_id, node_id, operation, state, attempts, created_at) \ + VALUES ($1, $2, 'REMOVE', 'PENDING', 0, now())", + [operation_id.into(), node_id.into()], + )) + .await + .map_err(internal)?; + Ok(GcClaim::Marked) +} + +async fn complete_locked( + txn: &DatabaseTransaction, + operation_id: &str, +) -> Result { + let op = mst2_retention_gc_op::Entity::find_by_id(operation_id.to_owned()) + .one(txn) + .await + .map_err(internal)? + .ok_or_else(|| integrity("unknown GC operation"))?; + if op.operation != "REMOVE" { + return Err(integrity("GC operation is not a removal")); + } + if op.state == "APPLIED" { + return Ok(GcCompletion { + replayed: true, + zero_reference_children: Vec::new(), + }); + } + if op.state != "PENDING" { + return Err(integrity("GC operation is not replayable")); + } + let node = mst2_retention_node::Entity::find_by_id(op.node_id.clone()) + .one(txn) + .await + .map_err(internal)? + .ok_or_else(|| integrity("pending GC node is missing"))?; + if node.state != "DELETING" || node.incoming_refs != 0 { + return Err(integrity("pending GC node is not unreferenced DELETING")); + } + if txn + .query_one_raw(statement( + "SELECT node_id FROM mst2_retention_root WHERE node_id = $1 \ + UNION ALL SELECT child_id FROM mst2_retention_edge WHERE child_id = $1 LIMIT 1", + [op.node_id.clone().into()], + )) + .await + .map_err(internal)? + .is_some() + { + return Err(integrity("pending GC node gained a reference; GC stopped")); + } + // Audit before subtracting. Even a damaged stored counter must not cause + // a child's data to be treated as unreferenced. + if txn.query_one_raw(statement( + "SELECT c.node_id FROM mst2_retention_edge e \ + JOIN mst2_retention_node c ON c.node_id = e.child_id \ + WHERE e.parent_id = $1 AND (c.state <> 'LIVE' OR c.incoming_refs <= 0 OR \ + c.incoming_refs <> (SELECT count(*) FROM mst2_retention_edge i WHERE i.child_id = c.node_id)) \ + LIMIT 1", + [op.node_id.clone().into()], + )).await.map_err(internal)?.is_some() { + return Err(integrity("child reference audit failed; GC stopped")); + } + let rows = txn.query_all_raw(statement( + "WITH removed AS (DELETE FROM mst2_retention_edge WHERE parent_id = $1 RETURNING child_id), \ + delta AS (SELECT child_id, count(*) AS refs FROM removed GROUP BY child_id) \ + UPDATE mst2_retention_node n SET incoming_refs = n.incoming_refs - delta.refs \ + FROM delta WHERE n.node_id = delta.child_id RETURNING n.node_id, n.incoming_refs", + [op.node_id.clone().into()], + )).await.map_err(internal)?; + let mut zero_reference_children = Vec::new(); + for row in rows { + if row.try_get::("", "incoming_refs").map_err(internal)? == 0 { + zero_reference_children.push(row.try_get("", "node_id").map_err(internal)?); + } + } + zero_reference_children.sort(); + txn.execute_raw(statement( + "DELETE FROM mst2_retention_node WHERE node_id = $1 AND state = 'DELETING'", + [op.node_id.into()], + )) + .await + .map_err(internal)?; + txn.execute_raw(statement( + "UPDATE mst2_retention_gc_op SET state = 'APPLIED', completed_at = now(), attempts = attempts + 1 \ + WHERE operation_id = $1 AND state = 'PENDING'", + [operation_id.into()], + )).await.map_err(internal)?; + Ok(GcCompletion { + replayed: false, + zero_reference_children, + }) +} + +async fn begin_graph(txn: &DatabaseTransaction) -> Result { + if txn.get_database_backend() != DbBackend::Postgres { + return Err(internal("retention repository requires PostgreSQL")); + } + let isolation = txn + .query_one_raw(statement("SHOW transaction_isolation", [])) + .await + .map_err(internal)? + .ok_or_else(|| internal("transaction isolation query returned no row"))? + .try_get_by_index::(0) + .map_err(internal)?; + if isolation != "read committed" { + return Err(internal( + "retention mutations require READ COMMITTED transaction isolation", + )); + } + let savepoint = txn.begin().await.map_err(internal)?; + let result = savepoint + .execute_raw(statement( + "SELECT pg_advisory_xact_lock($1, hashtext(current_schema()))", + [RETENTION_LOCK_KEY.into()], + )) + .await + .map_err(internal); + if let Err(err) = result { + return finish(savepoint, Err(err)).await; + } + Ok(savepoint) +} + +async fn finish( + txn: DatabaseTransaction, + result: Result, +) -> Result { + match result { + Ok(value) => { + txn.commit().await.map_err(internal)?; + Ok(value) + } + Err(err) => { + txn.rollback().await.map_err(internal)?; + Err(err) + } + } +} + +fn statement(sql: &str, values: [sea_orm::Value; N]) -> Statement { + Statement::from_sql_and_values(DbBackend::Postgres, sql, values) +} + +fn validate_id(id: &str) -> Result<(), SnapshotError> { + if id.is_empty() || id.len() > 255 || id.contains('\0') { + return Err(limit_error( + "retention identifiers require 1..=255 bytes without NUL", + )); + } + Ok(()) +} + +fn root_identity(root: &RetentionRoot) -> Result<(String, &'static str), SnapshotError> { + let (id, kind) = match root { + RetentionRoot::Lease(id) => (id, "lease"), + RetentionRoot::Pin(id) => (id, "pin"), + RetentionRoot::Prepare(id) => (id, "prepare"), + }; + validate_id(id)?; + let key = format!("{kind}:{id}"); + validate_id(&key)?; + Ok((key, kind)) +} + +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 unavailable(message: &str) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::ObjectUnavailable, message) +} +fn limit_error(message: &str) -> SnapshotError { + SnapshotError::new(SnapshotErrorCode::LimitExceeded, message) +} + +#[cfg(test)] +#[path = "mst2_retention_tests.rs"] +mod tests; diff --git a/src/jupiter/storage/mst2_retention_tests.rs b/src/jupiter/storage/mst2_retention_tests.rs new file mode 100644 index 00000000..a8f7c866 --- /dev/null +++ b/src/jupiter/storage/mst2_retention_tests.rs @@ -0,0 +1,629 @@ +use std::time::Duration; + +use sea_orm::{ + ConnectionTrait, DatabaseConnection, EntityTrait, IsolationLevel, PaginatorTrait, + TransactionTrait, +}; +use sea_orm_migration::MigratorTrait; + +use super::*; +use crate::{ + callisto::{mst2_retention_edge, mst2_retention_root}, + ceres::snapshot::retention::RetainedKind, + jupiter::{migration::Migrator, tests::test_db_connection}, +}; + +fn node(id: &str) -> RetentionNode { + RetentionNode { + id: id.into(), + kind: RetainedKind::Page, + state: NodeState::Live, + bytes: 7, + } +} + +fn edge(parent: &str, child: &str) -> RetentionEdge { + RetentionEdge { + parent: parent.into(), + child: child.into(), + } +} + +async fn fixture() -> (DatabaseConnection, PostgresRetentionRepository) { + let temp = tempfile::TempDir::new().expect("temp directory"); + let db = test_db_connection(temp.path()).await; + Migrator::up(&db, None) + .await + .expect("isolated schema migrations"); + let repository = PostgresRetentionRepository::new(db.clone()); + (db, repository) +} + +#[test] +fn t06b_rejects_cycles_conflicting_identities_and_unbounded_groups() { + let a = node("a"); + let b = node("b"); + assert_eq!( + PreparedGroup::new(&[a.clone(), b], &[edge("a", "b"), edge("b", "a")], &[]) + .err() + .expect("cycle rejected") + .code, + SnapshotErrorCode::IntegrityError + ); + let mut conflicting = a.clone(); + conflicting.bytes += 1; + assert!(PreparedGroup::new(&[a.clone(), conflicting], &[], &[]).is_err()); + assert_eq!( + PreparedGroup::new(&vec![a; MAX_NODES + 1], &[], &[]) + .err() + .expect("batch rejected") + .code, + SnapshotErrorCode::LimitExceeded + ); +} + +#[tokio::test] +async fn t06b_retains_unique_edges_and_independent_roots_once() { + let (db, repository) = fixture().await; + let roots = [ + RetentionRoot::Lease("one".into()), + RetentionRoot::Pin("two".into()), + ]; + let nodes = [node("parent"), node("child")]; + let edges = [edge("parent", "child"), edge("parent", "child")]; + for _ in 0..2 { + repository + .retain_group(&nodes, &edges, &roots) + .await + .expect("idempotent retain"); + } + assert_eq!( + repository + .node("child") + .await + .unwrap() + .unwrap() + .incoming_refs, + 1 + ); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 1 + ); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 4 + ); + repository.release_root(&roots[0]).await.unwrap(); + repository.release_root(&roots[0]).await.unwrap(); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 2 + ); + assert_eq!( + repository + .mark_deleting("gc-parent", "parent") + .await + .unwrap(), + GcClaim::Unavailable + ); + repository.release_root(&roots[1]).await.unwrap(); + assert_eq!( + repository + .mark_deleting("gc-parent", "parent") + .await + .unwrap(), + GcClaim::Marked + ); + assert_eq!( + repository.mark_deleting("gc-child", "child").await.unwrap(), + GcClaim::Unavailable + ); +} + +#[tokio::test] +async fn t06b_group_failure_rolls_back_savepoint_even_if_caller_commits() { + let (db, repository) = fixture().await; + repository + .retain_group(&[node("old")], &[], &[]) + .await + .unwrap(); + assert_eq!( + repository.mark_deleting("old-gc", "old").await.unwrap(), + GcClaim::Marked + ); + let txn = db.begin().await.unwrap(); + let error = PostgresRetentionRepository::retain_group_in_txn( + &txn, + &[node("new")], + &[edge("new", "old")], + &[RetentionRoot::Lease("partial".into())], + ) + .await + .expect_err("DELETING child rejects whole group"); + assert_eq!(error.code, SnapshotErrorCode::ObjectUnavailable); + txn.commit().await.unwrap(); + assert!(repository.node("new").await.unwrap().is_none()); + assert_eq!( + repository.node("old").await.unwrap().unwrap().state, + "DELETING" + ); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn t06b_publication_rollback_removes_retention_acquisition() { + let (db, repository) = fixture().await; + let txn = db.begin().await.unwrap(); + PostgresRetentionRepository::retain_group_in_txn( + &txn, + &[node("publication")], + &[], + &[RetentionRoot::Prepare("writer".into())], + ) + .await + .unwrap(); + txn.rollback().await.unwrap(); + assert!(repository.node("publication").await.unwrap().is_none()); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn t06b_root_release_rollback_preserves_coverage() { + let (db, repository) = fixture().await; + let root = RetentionRoot::Lease("reader".into()); + repository + .retain_group(&[node("page")], &[], std::slice::from_ref(&root)) + .await + .unwrap(); + let txn = db.begin().await.unwrap(); + PostgresRetentionRepository::release_root_in_txn(&txn, &root) + .await + .unwrap(); + txn.rollback().await.unwrap(); + assert_eq!( + repository.mark_deleting("page-op", "page").await.unwrap(), + GcClaim::Unavailable + ); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 1 + ); +} + +#[tokio::test] +async fn t06b_rejects_snapshot_isolation_without_mutating_outer_transaction() { + let (db, repository) = fixture().await; + let txn = db + .begin_with_config(Some(IsolationLevel::RepeatableRead), None) + .await + .unwrap(); + let error = PostgresRetentionRepository::retain_group_in_txn( + &txn, + &[node("page")], + &[], + &[RetentionRoot::Lease("reader".into())], + ) + .await + .expect_err("old MVCC snapshots cannot safely follow an advisory lock"); + assert_eq!(error.code, SnapshotErrorCode::Internal); + assert!(error.message.contains("READ COMMITTED")); + txn.commit().await.unwrap(); + assert!(repository.node("page").await.unwrap().is_none()); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn t06b_existing_parent_and_node_definition_are_immutable() { + let (db, repository) = fixture().await; + repository + .retain_group(&[node("parent"), node("child")], &[], &[]) + .await + .unwrap(); + let mut changed = node("parent"); + changed.bytes += 1; + assert!(repository.retain_group(&[changed], &[], &[]).await.is_err()); + assert!( + repository + .retain_group(&[node("parent")], &[edge("parent", "child")], &[]) + .await + .is_err() + ); + assert_eq!(repository.node("parent").await.unwrap().unwrap().bytes, 7); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); +} + +#[tokio::test] +async fn t06b_gc_replay_subtracts_edges_once_and_tombstones_removed_identity() { + let (db, repository) = fixture().await; + repository + .retain_group( + &[node("parent"), node("child")], + &[edge("parent", "child")], + &[], + ) + .await + .unwrap(); + assert_eq!( + repository + .mark_deleting("parent-op", "parent") + .await + .unwrap(), + GcClaim::Marked + ); + let restarted = PostgresRetentionRepository::new(db.clone()); + let pending = restarted.pending_gc(10).await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].operation_id, "parent-op"); + assert_eq!( + restarted + .mark_deleting("parent-op", "parent") + .await + .unwrap(), + GcClaim::Pending + ); + assert_eq!( + restarted + .mark_deleting("another-op", "parent") + .await + .unwrap(), + GcClaim::Unavailable + ); + assert!( + restarted + .mark_deleting("parent-op", "different") + .await + .is_err() + ); + + // Simulated crash/rollback during bookkeeping: the PENDING intent and + // child's reference survive and can be replayed after reaper replacement. + let txn = db.begin().await.unwrap(); + let provisional = PostgresRetentionRepository::complete_gc_in_txn(&txn, "parent-op") + .await + .unwrap(); + assert_eq!(provisional.zero_reference_children, vec!["child"]); + txn.rollback().await.unwrap(); + assert_eq!( + restarted + .node("child") + .await + .unwrap() + .unwrap() + .incoming_refs, + 1 + ); + assert_eq!(restarted.pending_gc(10).await.unwrap().len(), 1); + assert_eq!( + restarted.node("parent").await.unwrap().unwrap().state, + "DELETING" + ); + + let completed = restarted.complete_gc("parent-op").await.unwrap(); + assert!(!completed.replayed); + assert_eq!(completed.zero_reference_children, vec!["child"]); + assert!(restarted.node("parent").await.unwrap().is_none()); + assert_eq!( + restarted + .node("child") + .await + .unwrap() + .unwrap() + .incoming_refs, + 0 + ); + assert!(restarted.complete_gc("parent-op").await.unwrap().replayed); + assert_eq!( + restarted + .mark_deleting("parent-op", "parent") + .await + .unwrap(), + GcClaim::Applied + ); + assert!(restarted.pending_gc(10).await.unwrap().is_empty()); + let receipt = mst2_retention_gc_op::Entity::find_by_id("parent-op".to_owned()) + .one(&db) + .await + .unwrap() + .unwrap(); + assert_eq!(receipt.attempts, 1); + assert!(receipt.completed_at.is_some()); + + // Old pending workers must not be able to delete same-id reconstructed + // bytes. Reconstruction requires a future generation/fencing protocol. + assert_eq!( + restarted + .retain_group(&[node("parent")], &[], &[]) + .await + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + assert!(restarted.complete_gc("parent-op").await.unwrap().replayed); + assert!(restarted.node("parent").await.unwrap().is_none()); +} + +#[tokio::test] +async fn t06b_counter_audit_stops_gc_and_preserves_pending_edges() { + let (db, repository) = fixture().await; + repository + .retain_group( + &[node("parent"), node("child")], + &[edge("parent", "child")], + &[], + ) + .await + .unwrap(); + db.execute_unprepared( + "UPDATE mst2_retention_node SET incoming_refs = 0 WHERE node_id = 'child'", + ) + .await + .unwrap(); + assert_eq!( + repository + .mark_deleting("child-op", "child") + .await + .unwrap_err() + .code, + SnapshotErrorCode::IntegrityError + ); + assert_eq!( + repository + .mark_deleting("parent-op", "parent") + .await + .unwrap(), + GcClaim::Marked + ); + assert_eq!( + repository.complete_gc("parent-op").await.unwrap_err().code, + SnapshotErrorCode::IntegrityError + ); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 1 + ); + assert_eq!(repository.pending_gc(10).await.unwrap().len(), 1); + assert_eq!( + repository.node("child").await.unwrap().unwrap().state, + "LIVE" + ); + db.execute_unprepared( + "UPDATE mst2_retention_node SET incoming_refs = 1 WHERE node_id = 'child'", + ) + .await + .unwrap(); + repository.complete_gc("parent-op").await.unwrap(); + assert_eq!( + repository + .node("child") + .await + .unwrap() + .unwrap() + .incoming_refs, + 0 + ); +} + +#[tokio::test] +async fn t06b_root_acquisition_wins_before_gc_cas() { + let (db, repository) = fixture().await; + repository + .retain_group(&[node("page")], &[], &[]) + .await + .unwrap(); + let root_txn = db.begin().await.unwrap(); + PostgresRetentionRepository::retain_group_in_txn( + &root_txn, + &[node("page")], + &[], + &[RetentionRoot::Lease("winner".into())], + ) + .await + .unwrap(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let gc_repository = repository.clone(); + let mut gc = tokio::spawn(async move { + started_tx.send(()).unwrap(); + gc_repository.mark_deleting("gc-op", "page").await + }); + started_rx.await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut gc) + .await + .is_err() + ); + root_txn.commit().await.unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), gc) + .await + .unwrap() + .unwrap() + .unwrap(), + GcClaim::Unavailable + ); + assert_eq!( + repository.node("page").await.unwrap().unwrap().state, + "LIVE" + ); + assert!(repository.pending_gc(10).await.unwrap().is_empty()); +} + +#[tokio::test] +async fn t06b_gc_cas_wins_and_new_root_cannot_reference_deleting_node() { + let (db, repository) = fixture().await; + repository + .retain_group(&[node("page")], &[], &[]) + .await + .unwrap(); + let gc_txn = db.begin().await.unwrap(); + assert_eq!( + PostgresRetentionRepository::mark_deleting_in_txn(&gc_txn, "gc-op", "page") + .await + .unwrap(), + GcClaim::Marked + ); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let acquire_repository = repository.clone(); + let mut acquire = tokio::spawn(async move { + started_tx.send(()).unwrap(); + acquire_repository + .retain_group(&[node("page")], &[], &[RetentionRoot::Lease("late".into())]) + .await + }); + started_rx.await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut acquire) + .await + .is_err() + ); + gc_txn.commit().await.unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), acquire) + .await + .unwrap() + .unwrap() + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + assert_eq!( + mst2_retention_root::Entity::find() + .count(&db) + .await + .unwrap(), + 0 + ); + assert_eq!(repository.pending_gc(10).await.unwrap().len(), 1); +} + +#[tokio::test] +async fn t06b_forward_migration_backfills_old_graph_and_enforces_constraints() { + let temp = tempfile::TempDir::new().unwrap(); + let db = test_db_connection(temp.path()).await; + let at = Migrator::migrations() + .iter() + .position(|migration| migration.name() == "m20261005_000200_harden_mst2_retention_graph") + .expect("hardening migration registered"); + Migrator::up(&db, Some(at.try_into().unwrap())) + .await + .unwrap(); + db.execute_unprepared( + "INSERT INTO mst2_retention_node (node_id, kind, state, bytes, created_at) \ + VALUES ('parent', 'page', 'LIVE', 7, now()), ('child', 'page', 'LIVE', 7, now()); \ + INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('parent', 'child', now())", + ) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + let repository = PostgresRetentionRepository::new(db.clone()); + assert_eq!( + repository + .node("child") + .await + .unwrap() + .unwrap() + .incoming_refs, + 1 + ); + assert!( + db.execute_unprepared( + "INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('missing', 'child', now())" + ) + .await + .is_err() + ); + assert!( + db.execute_unprepared( + "UPDATE mst2_retention_node SET incoming_refs = -1 WHERE node_id = 'child'" + ) + .await + .is_err() + ); + let completion_error = db + .execute_unprepared( + "INSERT INTO mst2_retention_gc_op (operation_id, node_id, operation, state, attempts, created_at) \ + VALUES ('bad', 'child', 'REMOVE', 'APPLIED', 0, now())", + ) + .await + .expect_err("APPLIED receipt without completed_at must violate its completion constraint"); + assert!( + completion_error + .to_string() + .contains("mst2_retention_gc_op_completion_check"), + "unexpected constraint failure: {completion_error}" + ); +} + +#[tokio::test] +async fn t06b_forward_migration_refuses_existing_cycles() { + let temp = tempfile::TempDir::new().unwrap(); + let db = test_db_connection(temp.path()).await; + let at = Migrator::migrations() + .iter() + .position(|migration| migration.name() == "m20261005_000200_harden_mst2_retention_graph") + .unwrap(); + Migrator::up(&db, Some(at.try_into().unwrap())) + .await + .unwrap(); + db.execute_unprepared( + "INSERT INTO mst2_retention_node (node_id, kind, state, bytes, created_at) \ + VALUES ('a', 'page', 'LIVE', 7, now()), ('b', 'page', 'LIVE', 7, now()); \ + INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('a', 'b', now()), ('b', 'a', now())", + ) + .await + .unwrap(); + assert!(Migrator::up(&db, None).await.is_err()); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 2 + ); +}