From 648d82f74f63b85c797199bb0adc19af58af84cc Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 07:55:17 +0800 Subject: [PATCH 01/13] fix(mst2): make retention acquisition atomic Protect snapshot retention roots and edges with one validated commit, reject DELETING references, and propagate lease lifecycle failures. --- src/api/router/snapshot_router.rs | 4 +- src/ceres/snapshot/retention.rs | 447 ++++++++++++++++++++++++----- src/ceres/snapshot/runtime.rs | 452 ++++++++++++++++++++++++++---- 3 files changed, 773 insertions(+), 130 deletions(-) diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 556897e3..e66633cc 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -396,7 +396,9 @@ async fn resolve(state: State, body: Bytes) -> Result, /// Bytes that would be / were reclaimed. pub reclaimed_bytes: u64, @@ -140,16 +140,30 @@ impl Reaper for NoopReaper { pub trait RetentionStore { fn node(&self, id: &str) -> Option; fn root_covers(&self, node_id: &str) -> bool; + /// Count incoming edges from parents not yet removed, including + /// DELETING parents whose physical reclaim has not succeeded. fn live_incoming(&self, node_id: &str) -> usize; - /// Idempotent: create the node if absent, upsert edges and roots in - /// one atomic step (spec §6 "no empty window"). + /// Idempotent: retain one LIVE node with its edges and root coverage. fn retain( &self, node: RetentionNode, edges: &[RetentionEdge], roots: &[RetentionRoot], + ) -> Result<(), SnapshotError> { + self.retain_group(std::slice::from_ref(&node), edges, roots) + } + /// Atomically retain the entire group, covering every supplied node + /// with `roots`. All nodes and edge endpoints must be LIVE; acquiring + /// a DELETING node is forbidden. Any error leaves the graph unchanged. + fn retain_group( + &self, + nodes: &[RetentionNode], + edges: &[RetentionEdge], + roots: &[RetentionRoot], ) -> Result<(), SnapshotError>; - /// Atomically CAS a node LIVE→DELETING; false if it is no longer LIVE. + /// Atomically check zero roots/incoming references and CAS LIVE→DELETING. + /// A parent still protects its children until the parent is removed, + /// including while the parent is DELETING and physical reclaim is pending. fn mark_deleting(&self, id: &str) -> bool; /// Remove a DELETING node after the reaper succeeds. fn remove(&self, id: &str); @@ -182,14 +196,6 @@ impl RetentionCoordinator { parents: &[String], roots: &[RetentionRoot], ) -> Result<(), SnapshotError> { - for p in parents { - if self.store.node(p).is_none() { - return Err(SnapshotError::new( - SnapshotErrorCode::Internal, - format!("retention edge to unknown parent {p}"), - )); - } - } let edges: Vec = parents .iter() .map(|p| RetentionEdge { @@ -200,38 +206,24 @@ impl RetentionCoordinator { self.store.retain(node, &edges, roots) } - /// Add coverage from a root to a node, atomically re-lifting it out of - /// a DELETING state (spec §10). The root is also materialized as a live - /// anchor node so reachability can start from it. The store's `retain` - /// is the upsert, so both the fresh and the already-known cases take - /// the same path. + /// Acquire root coverage for the complete group in one store operation. + /// The root is also materialized as a LIVE anchor. If any covered node + /// is DELETING, neither the anchor nor any partial coverage is retained. pub fn pin_root( &self, root: &RetentionRoot, covered: &[RetentionNode], ) -> Result<(), SnapshotError> { - let roots = std::slice::from_ref(root); - self.store.retain( - RetentionNode { - id: root.key(), - kind: RetainedKind::Frame, - state: NodeState::Live, - bytes: 0, - }, - &[], - roots, - )?; - for n in covered { - self.store.retain( - RetentionNode { - state: NodeState::Live, - ..n.clone() - }, - &[], - roots, - )?; - } - Ok(()) + let mut nodes = Vec::with_capacity(covered.len() + 1); + nodes.push(RetentionNode { + id: root.key(), + kind: RetainedKind::Frame, + state: NodeState::Live, + bytes: 0, + }); + nodes.extend_from_slice(covered); + self.store + .retain_group(&nodes, &[], std::slice::from_ref(root)) } pub fn release(&self, root: &RetentionRoot) { @@ -239,11 +231,10 @@ impl RetentionCoordinator { } /// One collection pass (spec 10 §6 steps 1–7): - /// mark unreachable LIVE nodes DELETING, then reap. Nodes marked - /// DELETING are skipped entirely on later passes if a new root/edge - /// raced in only after the CAS — here reachability is recomputed under - /// the store lock, so a retain concurrent with a pass either observes - /// the node (keeping it LIVE) or lands before the next pass. + /// mark unreachable LIVE nodes DELETING, then reap. The reachability + /// scan selects candidates; the store rechecks roots and incoming + /// references atomically with the CAS. Acquisition either wins before + /// that CAS or fails because the node is already DELETING. pub fn collect(&self, reaper: &R) -> Result { let live = self.store.all_live(); let reachable = self.reachable_set(); @@ -254,18 +245,11 @@ impl RetentionCoordinator { continue; } if !self.store.mark_deleting(&node.id) { - // Lost the CAS: a new reference made it LIVE again. + // Another collector or a newly acquired reference won. continue; } report.unreachable.push(node.id.clone()); report.reclaimed_bytes += node.bytes; - // Re-verify after the CAS: the store serializes retain against - // mark, so a zero incoming count here is stable for this pass. - if self.store.root_covers(&node.id) || self.store.live_incoming(&node.id) > 0 { - // A concurrent retain resurrected it; the CAS semantics of - // the store keep it LIVE, so do not reap. - continue; - } if reaper.physical() { reaper.reap(&node)?; self.store.remove(&node.id); @@ -311,6 +295,8 @@ pub mod mem { edges: HashSet, /// node id -> set of root keys covering it. roots: HashMap>, + #[cfg(test)] + fail_next_retain: bool, } /// Single-Mutex store; the lock is the serialization point that makes @@ -327,6 +313,14 @@ pub mod mem { } } + #[cfg(test)] + impl InMemoryRetentionStore { + /// Inject a backend failure after validation and before committing. + pub(crate) fn fail_next_retain_for_test(&self) { + self.inner.lock().unwrap().fail_next_retain = true; + } + } + impl RetentionStore for InMemoryRetentionStore { fn node(&self, id: &str) -> Option { self.inner.lock().unwrap().nodes.get(id).cloned() @@ -350,41 +344,82 @@ pub mod mem { .count() } - fn retain( + fn retain_group( &self, - node: RetentionNode, + nodes: &[RetentionNode], edges: &[RetentionEdge], roots: &[RetentionRoot], ) -> Result<(), SnapshotError> { let mut g = self.inner.lock().unwrap(); - // Atomic re-lift: never downgrade a LIVE node to DELETING. - g.nodes - .entry(node.id.clone()) - .and_modify(|existing| { - existing.bytes = node.bytes; - existing.kind = node.kind; - existing.state = NodeState::Live; - }) - .or_insert_with(|| node.clone()); - // Defensive: do not insert an edge to a missing child node. - for e in edges { - if !g.nodes.contains_key(&e.child) || !g.nodes.contains_key(&e.parent) { + // Validate the entire transaction before mutating any graph data. + let mut staged = HashMap::new(); + for node in nodes { + if node.state != NodeState::Live + || g.nodes + .get(&node.id) + .is_some_and(|existing| existing.state != NodeState::Live) + { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + format!("retention acquire requires LIVE node {}", node.id), + )); + } + if staged + .insert(node.id.as_str(), node) + .is_some_and(|previous| previous != node) + { return Err(SnapshotError::new( SnapshotErrorCode::Internal, - "retention edge references missing node", + "retention group has conflicting node definitions", )); } - g.edges.insert(e.clone()); } - let entry = g.roots.entry(node.id.clone()).or_default(); - for r in roots { - entry.insert(r.key()); + for e in edges { + for id in [&e.parent, &e.child] { + match staged.get(id.as_str()).copied().or_else(|| g.nodes.get(id)) { + Some(node) if node.state == NodeState::Live => {} + Some(_) => { + return Err(SnapshotError::new( + SnapshotErrorCode::ObjectUnavailable, + format!("retention edge requires LIVE endpoint {id}"), + )); + } + None => { + return Err(SnapshotError::new( + SnapshotErrorCode::Internal, + format!("retention edge references missing node {id}"), + )); + } + } + } + } + #[cfg(test)] + if std::mem::take(&mut g.fail_next_retain) { + return Err(SnapshotError::new( + SnapshotErrorCode::Internal, + "injected retention commit failure", + )); + } + for node in nodes { + g.nodes.insert(node.id.clone(), node.clone()); + let entry = g.roots.entry(node.id.clone()).or_default(); + for root in roots { + entry.insert(root.key()); + } } + g.edges.extend(edges.iter().cloned()); Ok(()) } fn mark_deleting(&self, id: &str) -> bool { let mut g = self.inner.lock().unwrap(); + if g.roots.get(id).is_some_and(|roots| !roots.is_empty()) + || g.edges + .iter() + .any(|edge| edge.child == id && g.nodes.contains_key(&edge.parent)) + { + return false; + } match g.nodes.get_mut(id) { Some(n) if n.state == NodeState::Live => { n.state = NodeState::Deleting; @@ -396,6 +431,13 @@ pub mod mem { fn remove(&self, id: &str) { let mut g = self.inner.lock().unwrap(); + if !g + .nodes + .get(id) + .is_some_and(|node| node.state == NodeState::Deleting) + { + return; + } g.nodes.remove(id); g.edges.retain(|e| e.parent != id && e.child != id); g.roots.remove(id); @@ -435,6 +477,15 @@ pub mod mem { #[cfg(test)] mod tests { + use std::{ + sync::{ + Arc, + mpsc::{Receiver, SyncSender, sync_channel}, + }, + thread, + time::Duration, + }; + use super::{mem::InMemoryRetentionStore, *}; fn node(id: &str, bytes: u64) -> RetentionNode { @@ -450,6 +501,243 @@ mod tests { RetentionCoordinator::new(InMemoryRetentionStore::default()) } + /// Pause a collection after its scan, immediately before the store CAS. + /// Timeouts make both sides of the forced interleaving bounded. + struct PausedMarkStore { + inner: InMemoryRetentionStore, + ready: SyncSender<()>, + resume: Mutex>, + } + + impl RetentionStore for PausedMarkStore { + fn node(&self, id: &str) -> Option { + self.inner.node(id) + } + + fn root_covers(&self, node_id: &str) -> bool { + self.inner.root_covers(node_id) + } + + fn live_incoming(&self, node_id: &str) -> usize { + self.inner.live_incoming(node_id) + } + + fn retain_group( + &self, + nodes: &[RetentionNode], + edges: &[RetentionEdge], + roots: &[RetentionRoot], + ) -> Result<(), SnapshotError> { + self.inner.retain_group(nodes, edges, roots) + } + + fn mark_deleting(&self, id: &str) -> bool { + self.ready.send(()).unwrap(); + self.resume + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(5)) + .expect("test did not resume deletion"); + self.inner.mark_deleting(id) + } + + fn remove(&self, id: &str) { + self.inner.remove(id); + } + + fn release_root(&self, root: &RetentionRoot) { + self.inner.release_root(root); + } + + fn all_live(&self) -> Vec { + self.inner.all_live() + } + + fn children(&self, parent: &str) -> Vec { + self.inner.children(parent) + } + } + + #[test] + fn failed_pin_group_leaves_no_anchor_or_partial_coverage() { + let c = coord(); + let original = node("existing", 10); + c.store().retain(original.clone(), &[], &[]).unwrap(); + c.store().retain(node("deleting", 7), &[], &[]).unwrap(); + assert!(c.store().mark_deleting("deleting")); + let root = RetentionRoot::Lease("new".into()); + + // The last node fails after earlier entries would have been written + // by the old per-node pin loop, including an update to existing. + let err = c + .pin_root( + &root, + &[node("fresh", 1), node("existing", 99), node("deleting", 7)], + ) + .unwrap_err(); + assert_eq!(err.code, SnapshotErrorCode::ObjectUnavailable); + assert!(c.store().node(&root.key()).is_none()); + assert!(c.store().node("fresh").is_none()); + assert_eq!(c.store().node("existing"), Some(original)); + assert!(!c.store().root_covers("existing")); + assert!(!c.store().root_covers("deleting")); + assert_eq!( + c.store().node("deleting").unwrap().state, + NodeState::Deleting + ); + } + + #[test] + fn supplied_deleting_node_cannot_be_created_or_pinned() { + let c = coord(); + let root = RetentionRoot::Pin("invalid".into()); + let mut deleting = node("absent", 1); + deleting.state = NodeState::Deleting; + assert_eq!( + c.pin_root(&root, &[deleting]).unwrap_err().code, + SnapshotErrorCode::ObjectUnavailable + ); + assert!(c.store().all_live().is_empty()); + assert!(c.store().node("absent").is_none()); + assert!(c.store().node(&root.key()).is_none()); + } + + #[test] + fn late_invalid_edge_rolls_back_nodes_edges_and_roots() { + let c = coord(); + c.store().retain(node("parent", 0), &[], &[]).unwrap(); + let root = RetentionRoot::Lease("new".into()); + let edges = [ + RetentionEdge { + parent: "parent".into(), + child: "fresh".into(), + }, + RetentionEdge { + parent: "missing".into(), + child: "fresh".into(), + }, + ]; + let err = c + .store() + .retain(node("fresh", 2), &edges, &[root]) + .unwrap_err(); + assert_eq!(err.code, SnapshotErrorCode::Internal); + assert!(c.store().node("fresh").is_none()); + assert!(c.store().children("parent").is_empty()); + assert!(!c.store().root_covers("fresh")); + assert_eq!(c.store().live_incoming("fresh"), 0); + } + + #[test] + fn injected_commit_failure_preserves_existing_root_and_can_retry() { + let c = coord(); + let old = RetentionRoot::Pin("old".into()); + let new = RetentionRoot::Lease("new".into()); + c.store() + .retain(node("existing", 1), &[], std::slice::from_ref(&old)) + .unwrap(); + c.store().fail_next_retain_for_test(); + assert_eq!( + c.pin_root(&new, &[node("existing", 1), node("fresh", 2)]) + .unwrap_err() + .code, + SnapshotErrorCode::Internal + ); + assert!(c.store().node(&new.key()).is_none()); + assert!(c.store().node("fresh").is_none()); + assert!(c.store().root_covers("existing")); + c.release(&old); + assert!( + !c.store().root_covers("existing"), + "failed pin added no new root" + ); + + c.pin_root(&new, &[node("existing", 1), node("fresh", 2)]) + .unwrap(); + assert!(c.is_retained("existing")); + assert!(c.is_retained("fresh")); + } + + #[test] + fn collection_rechecks_roots_and_edges_acquired_after_its_scan() { + for via_edge in [false, true] { + let (ready_tx, ready_rx) = sync_channel(1); + let (resume_tx, resume_rx) = sync_channel(1); + let c = Arc::new(RetentionCoordinator::new(PausedMarkStore { + inner: InMemoryRetentionStore::default(), + ready: ready_tx, + resume: Mutex::new(resume_rx), + })); + c.store().retain(node("candidate", 1), &[], &[]).unwrap(); + let root = RetentionRoot::Lease("late".into()); + if via_edge { + c.store() + .retain(node("parent", 0), &[], std::slice::from_ref(&root)) + .unwrap(); + } + let collector = { + let c = Arc::clone(&c); + thread::spawn(move || c.collect(&NoopReaper)) + }; + ready_rx + .recv_timeout(Duration::from_secs(5)) + .expect("collector did not reach its deletion CAS"); + // Both references land after candidate selection, before CAS. + if via_edge { + c.retain(node("candidate", 1), &["parent".into()], &[]) + .unwrap(); + } else { + c.pin_root(&root, &[node("candidate", 1)]).unwrap(); + } + resume_tx.send(()).unwrap(); + let report = collector.join().unwrap().unwrap(); + assert!(report.unreachable.is_empty()); + assert_eq!(c.store().node("candidate").unwrap().state, NodeState::Live); + assert!(c.is_retained("candidate")); + } + } + + #[test] + fn deleting_parent_rejects_new_edges_and_protects_existing_children_until_removed() { + let c = coord(); + c.store().retain(node("parent", 0), &[], &[]).unwrap(); + c.retain(node("child", 1), &["parent".into()], &[]).unwrap(); + assert!(c.store().mark_deleting("parent")); + assert!(!c.store().mark_deleting("child")); + assert_eq!(c.store().live_incoming("child"), 1); + assert_eq!( + c.retain(node("new-child", 2), &["parent".into()], &[]) + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + assert!(c.store().node("new-child").is_none()); + c.store().remove("parent"); + assert_eq!(c.store().live_incoming("child"), 0); + assert!(c.store().mark_deleting("child")); + } + + #[test] + fn retain_cannot_add_an_edge_to_a_deleting_child() { + let c = coord(); + c.store().retain(node("child", 1), &[], &[]).unwrap(); + assert!(c.store().mark_deleting("child")); + let edge = RetentionEdge { + parent: "new-parent".into(), + child: "child".into(), + }; + assert_eq!( + c.store() + .retain(node("new-parent", 0), &[edge], &[]) + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + assert!(c.store().node("new-parent").is_none()); + assert_eq!(c.store().node("child").unwrap().state, NodeState::Deleting); + assert_eq!(c.store().live_incoming("child"), 0); + } + #[test] fn leaf_dies_when_root_released_but_other_root_keeps_it() { // root A and B both cover a shared leaf; releasing A must not @@ -486,8 +774,8 @@ mod tests { let r = c.collect(&NoopReaper).unwrap(); assert_eq!(r.unreachable, vec!["orphan".to_string()]); assert_eq!(r.reclaimed_bytes, 9); - // Releasing the root cascades unreachability to a and b only after - // the anchor itself is collected (the anchor node is root-covered). + // Releasing the root makes the chain unreachable, but children stay + // LIVE while their parents await physical removal under NoopReaper. c.release(&root); let _ = c.collect(&NoopReaper).unwrap(); assert!(!c.is_retained("b")); @@ -521,7 +809,7 @@ mod tests { ) .unwrap(); assert_eq!(c.store().live_incoming("shared"), 2); - // Remove p1's coverage and p1 itself; shared still has p2. + // Drop p1's coverage; its pending deletion must not affect p2's view. c.release(&RetentionRoot::Pin("r1".into())); c.collect(&NoopReaper).unwrap(); assert!(c.store().node("shared").is_some()); @@ -562,12 +850,12 @@ mod tests { let c = coord(); c.store().retain(node("x", 3), &[], &[]).unwrap(); assert!(c.collect(&Fail).is_err()); - // Node remains (DELETING) so a retry can replay; it is not lost. - assert!(c.store().node("x").is_some()); + // Node remains DELETING; physical retry/recovery is a later slice. + assert_eq!(c.store().node("x").unwrap().state, NodeState::Deleting); } #[test] - fn new_root_revives_a_node_before_it_is_collected() { + fn new_root_protects_a_live_node_before_it_is_collected() { let c = coord(); c.store().retain(node("x", 4), &[], &[]).unwrap(); // Before any collect, a lease covers it. @@ -644,7 +932,12 @@ mod tests { c.retain(node("ch", 2), &["p".to_string()], &[]).unwrap(); c.release(&RetentionRoot::Lease("l".into())); c.collect(&Delete).unwrap(); + // A child selected before its parent was removed waits for the + // next pass; iteration order must not affect the final result. + c.collect(&Delete).unwrap(); // Both nodes and their edge are gone; no stale incoming edge. + assert!(c.store().node("p").is_none()); + assert!(c.store().node("ch").is_none()); assert_eq!(c.store().live_incoming("ch"), 0); } } diff --git a/src/ceres/snapshot/runtime.rs b/src/ceres/snapshot/runtime.rs index 770c4725..b8b4bce3 100644 --- a/src/ceres/snapshot/runtime.rs +++ b/src/ceres/snapshot/runtime.rs @@ -11,7 +11,7 @@ use std::{ collections::HashMap, - sync::{Mutex, OnceLock}, + sync::{Mutex, MutexGuard, OnceLock}, time::{Duration, SystemTime, UNIX_EPOCH}, }; @@ -101,16 +101,16 @@ impl Mst2Runtime { st.1 } - /// Insert a context and create its first lease. The lease also becomes - /// a retention root covering the snapshot's derived metadata root - /// (spec 10 §5: active leases pin what a fixed view needs). + /// Acquire the complete in-memory retention group before exposing a + /// context or lease. These placeholder anchors do not yet represent + /// the complete durable page/chunk graph required by T06. pub fn insert_context( &self, built: BuiltDescriptor, commit_oid: &str, root_tree_oid: &str, lease_seconds: u64, - ) -> SnapshotContext { + ) -> Result { let lease_id = Uuid::new_v4().to_string(); let expires = now_unix() + lease_seconds.clamp(1, 3600); let snapshot_id = built.snapshot_id.clone(); @@ -122,23 +122,9 @@ impl Mst2Runtime { lease_expires_at_unix: expires, authorization_epoch: 1, }; - self.contexts.lock().unwrap().insert( - snapshot_id.clone(), - ContextData { - built: built.clone(), - commit_oid: commit_oid.to_string(), - root_tree_oid: root_tree_oid.to_string(), - }, - ); - self.leases.lock().unwrap().insert( - lease_id.clone(), - LeaseRec { - snapshot_id: snapshot_id.clone(), - expires_at_unix: expires, - }, - ); - // Retention root for the lease: the metadata-root page node and the - // snapshot's chunk projections stay reachable while it is active. + // Serialize registration with release/expiry/renewal. No path holds + // contexts while acquiring leases, so this lock order cannot cycle. + let mut leases = self.leases.lock().unwrap(); let root = crate::ceres::snapshot::retention::RetentionRoot::Lease(lease_id.clone()); let page_node = crate::ceres::snapshot::retention::RetentionNode { id: format!("page:{}", built.metadata_root), @@ -152,15 +138,24 @@ impl Mst2Runtime { state: crate::ceres::snapshot::retention::NodeState::Live, bytes: 0, }; - if let Err(e) = self - .retention - .pin_root(&root, &[page_node, projection_node]) - { - // Coverage failure is surfaced, not swallowed: a lease that - // cannot pin its view must not pretend to. - tracing::warn!(error = %e, "retention pin for new lease failed"); - } - ctx + self.retention + .pin_root(&root, &[page_node, projection_node])?; + self.contexts.lock().unwrap().insert( + snapshot_id.clone(), + ContextData { + built, + commit_oid: commit_oid.to_string(), + root_tree_oid: root_tree_oid.to_string(), + }, + ); + leases.insert( + lease_id, + LeaseRec { + snapshot_id, + expires_at_unix: expires, + }, + ); + Ok(ctx) } /// Look up a snapshot context by snapshot_id. It is servable while at @@ -194,7 +189,7 @@ impl Mst2Runtime { /// as it goes. fn active_lease(&self, snapshot_id: &str) -> Result<(String, u64), SnapshotError> { let mut leases = self.leases.lock().unwrap(); - leases.retain(|_, r| r.expires_at_unix >= now_unix()); + self.prune_expired_leases(&mut leases, now_unix()); leases .iter() .find(|(_, r)| r.snapshot_id == snapshot_id) @@ -207,6 +202,22 @@ impl Mst2Runtime { }) } + /// The leases lock serializes expiry with renewal and registration; + /// releasing one expired lease never removes another lease's coverage. + /// The current in-memory store's release operation is infallible. + fn prune_expired_leases(&self, leases: &mut HashMap, now: u64) { + leases.retain(|lease_id, rec| { + if rec.expires_at_unix >= now { + return true; + } + self.retention + .release(&crate::ceres::snapshot::retention::RetentionRoot::Lease( + lease_id.clone(), + )); + false + }); + } + /// Validate the lease a request presented (spec 04 §1: identity lives in /// the `X-Mega-Snapshot-Lease` header, and a read path must check it — /// knowing the snapshot id alone is not a capability). Any lease that is @@ -216,8 +227,8 @@ impl Mst2Runtime { /// existed is answered identically: probing with lease ids must not /// distinguish released from forged. /// - /// Expired-row pruning happens in `active_lease`, not here: this runs on - /// every authenticated read and must stay a single key lookup. + /// Expired-row pruning happens in `active_lease` and collection: this + /// runs on every authenticated read and stays a single key lookup. pub fn validate_lease(&self, snapshot_id: &str, lease_id: &str) -> Result<(), SnapshotError> { let leases = self.leases.lock().unwrap(); match leases.get(lease_id) { @@ -237,8 +248,24 @@ impl Mst2Runtime { lease_id: &str, lease_seconds: u64, ) -> Result { - let now = now_unix(); - let mut leases = self.leases.lock().unwrap(); + Self::renew_lease_with_clock( + lease_id, + lease_seconds, + || self.leases.lock().unwrap(), + now_unix, + ) + } + + /// Sample the clock only after acquiring the lease lock. Separate + /// sources let tests force a deadline crossing during lock contention. + fn renew_lease_with_clock<'a>( + lease_id: &str, + lease_seconds: u64, + acquire_leases: impl FnOnce() -> MutexGuard<'a, HashMap>, + clock: impl FnOnce() -> u64, + ) -> Result { + let mut leases = acquire_leases(); + let now = clock(); let Some(rec) = leases.get_mut(lease_id) else { return Err(SnapshotError::new( SnapshotErrorCode::LeaseUnknown, @@ -268,7 +295,8 @@ impl Mst2Runtime { /// only drops the retention root so a later collection pass may /// reclaim derived pages no other lease/pin covers. pub fn release_lease(&self, lease_id: &str) -> bool { - let removed = self.leases.lock().unwrap().remove(lease_id).is_some(); + let mut leases = self.leases.lock().unwrap(); + let removed = leases.remove(lease_id).is_some(); self.retention .release(&crate::ceres::snapshot::retention::RetentionRoot::Lease( lease_id.to_string(), @@ -282,6 +310,7 @@ impl Mst2Runtime { pub fn collect_retention( &self, ) -> Result { + self.prune_expired_leases(&mut self.leases.lock().unwrap(), now_unix()); self.retention .collect(&crate::ceres::snapshot::retention::NoopReaper) } @@ -336,20 +365,24 @@ pub fn rfc3339(unix: u64) -> String { #[cfg(test)] mod tests { - use super::*; + use std::{ + sync::{ + Arc, TryLockError, + atomic::{AtomicU64, Ordering}, + mpsc::sync_channel, + }, + thread, + }; - #[test] - fn rfc3339_known_values() { - assert_eq!(rfc3339(0), "1970-01-01T00:00:00Z"); - assert_eq!(rfc3339(1_789_525_722), "2026-09-16T02:28:42Z"); - } + use mst2_codec::descriptor::ServingDescriptor; - #[tokio::test] - async fn lease_lifecycle_drives_retention() { - use crate::ceres::snapshot::retention::{ - NodeState, RetainedKind, RetentionRoot, RetentionStore as _, - }; - let r = Mst2Runtime { + use super::*; + use crate::ceres::snapshot::retention::{ + NodeState, RetainedKind, RetentionRoot, RetentionStore as _, + }; + + fn isolated_runtime() -> Mst2Runtime { + Mst2Runtime { hmac_key: blake3_key(), contexts: Mutex::new(HashMap::new()), leases: Mutex::new(HashMap::new()), @@ -357,9 +390,324 @@ mod tests { crate::ceres::snapshot::retention::mem::InMemoryRetentionStore::default(), ), tip_state: Mutex::new((String::new(), 0u64)), + } + } + + fn descriptor(root_byte: u8) -> BuiltDescriptor { + let descriptor = ServingDescriptor { + instance_uuid: [0x11; 16], + namespace_view_id: [0x22; 32], + scope: "/".into(), + metadata_root: [root_byte; 32], + }; + BuiltDescriptor { + instance_id: Uuid::from_bytes(descriptor.instance_uuid).to_string(), + snapshot_id: format!( + "sha256:{}", + crate::ceres::snapshot::view::hex(&descriptor.snapshot_id().unwrap()) + ), + metadata_root: format!( + "sha256:{}", + crate::ceres::snapshot::view::hex(&descriptor.metadata_root) + ), + descriptor, + } + } + + #[test] + fn pin_failure_exposes_no_context_lease_or_partial_graph_and_can_retry() { + let r = isolated_runtime(); + let built = descriptor(3); + r.retention.store().fail_next_retain_for_test(); + assert_eq!( + r.insert_context(built.clone(), "commit", "tree", 60) + .unwrap_err() + .code, + SnapshotErrorCode::Internal + ); + assert!(r.contexts.lock().unwrap().is_empty()); + assert!(r.leases.lock().unwrap().is_empty()); + assert!(r.retention.store().all_live().is_empty()); + assert_eq!( + r.context(&built.snapshot_id).unwrap_err().code, + SnapshotErrorCode::SnapshotGone + ); + + let ctx = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + r.validate_lease(&built.snapshot_id, &ctx.lease_id).unwrap(); + assert!( + r.retention + .is_retained(&format!("page:{}", built.metadata_root)) + ); + assert!(r.retention.is_retained("projection:commit")); + } + + #[test] + fn failed_resolve_preserves_an_existing_active_snapshot() { + let r = isolated_runtime(); + let old = descriptor(3); + let active = r + .insert_context(old.clone(), "old-commit", "old-tree", 60) + .unwrap(); + let new = descriptor(4); + r.retention.store().fail_next_retain_for_test(); + assert!( + r.insert_context(new.clone(), "new-commit", "new-tree", 60) + .is_err() + ); + assert_eq!(r.contexts.lock().unwrap().len(), 1); + assert_eq!(r.leases.lock().unwrap().len(), 1); + assert_eq!( + r.context(&old.snapshot_id).unwrap().lease_id, + active.lease_id + ); + assert_eq!( + r.context(&new.snapshot_id).unwrap_err().code, + SnapshotErrorCode::SnapshotGone + ); + assert!( + r.retention + .store() + .node(&format!("page:{}", new.metadata_root)) + .is_none() + ); + assert!(r.retention.store().node("projection:new-commit").is_none()); + } + + #[test] + fn resolve_cannot_create_a_lease_for_a_deleting_group() { + let r = isolated_runtime(); + let built = descriptor(3); + let ctx = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + assert!(r.release_lease(&ctx.lease_id)); + assert!(r.collect_retention().unwrap().reaped.is_empty()); + assert_eq!( + r.insert_context(built.clone(), "commit", "tree", 60) + .unwrap_err() + .code, + SnapshotErrorCode::ObjectUnavailable + ); + assert!(r.leases.lock().unwrap().is_empty()); + assert!( + r.retention.store().all_live().is_empty(), + "no new lease anchor leaked" + ); + assert_eq!( + r.context(&built.snapshot_id).unwrap_err().code, + SnapshotErrorCode::LeaseExpired + ); + assert_eq!( + r.retention + .store() + .node(&format!("page:{}", built.metadata_root)) + .unwrap() + .state, + NodeState::Deleting + ); + } + + #[test] + fn context_expiry_releases_only_the_expired_lease_on_a_shared_snapshot() { + let r = isolated_runtime(); + let built = descriptor(3); + let expired = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + let active = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + r.leases + .lock() + .unwrap() + .get_mut(&expired.lease_id) + .unwrap() + .expires_at_unix = 0; + + assert_eq!( + r.context(&built.snapshot_id).unwrap().lease_id, + active.lease_id + ); + assert!(!r.leases.lock().unwrap().contains_key(&expired.lease_id)); + assert!( + !r.retention + .store() + .root_covers(&format!("lease:{}", expired.lease_id)) + ); + assert!( + r.retention + .store() + .root_covers(&format!("lease:{}", active.lease_id)) + ); + assert!( + r.retention + .is_retained(&format!("page:{}", built.metadata_root)) + ); + r.validate_lease(&built.snapshot_id, &active.lease_id) + .unwrap(); + assert_eq!( + r.validate_lease(&built.snapshot_id, &expired.lease_id) + .unwrap_err() + .code, + SnapshotErrorCode::LeaseExpired + ); + assert_eq!( + r.collect_retention().unwrap().unreachable, + vec![format!("lease:{}", expired.lease_id)] + ); + assert_eq!( + r.context(&built.snapshot_id).unwrap().lease_id, + active.lease_id + ); + + assert!(r.release_lease(&active.lease_id)); + assert!( + !r.retention + .store() + .root_covers(&format!("page:{}", built.metadata_root)) + ); + assert!(!r.retention.store().root_covers("projection:commit")); + } + + #[test] + fn collection_prunes_expired_roots_without_a_context_read() { + let r = isolated_runtime(); + let built = descriptor(3); + let ctx = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + r.leases + .lock() + .unwrap() + .get_mut(&ctx.lease_id) + .unwrap() + .expires_at_unix = 0; + let report = r.collect_retention().unwrap(); + assert_eq!(report.unreachable.len(), 3, "anchor, page and projection"); + assert!(report.reaped.is_empty(), "NoopReaper stays fail-closed"); + assert!(r.leases.lock().unwrap().is_empty()); + for id in [ + format!("lease:{}", ctx.lease_id), + format!("page:{}", built.metadata_root), + "projection:commit".into(), + ] { + assert!(report.unreachable.contains(&id)); + assert!(!r.retention.store().root_covers(&id)); + assert_eq!( + r.retention.store().node(&id).unwrap().state, + NodeState::Deleting + ); + } + } + + #[test] + fn renewal_winning_before_expiry_keeps_its_root_and_expiry_winning_prevents_renewal() { + let r = isolated_runtime(); + let built = descriptor(3); + let ctx = r + .insert_context(built.clone(), "commit", "tree", 60) + .unwrap(); + let renewed = r.renew_lease(&ctx.lease_id, 60).unwrap(); + { + let mut leases = r.leases.lock().unwrap(); + // Advance past the old deadline, before the renewed deadline. + r.prune_expired_leases(&mut leases, ctx.lease_expires_at_unix + 1); + } + r.validate_lease(&built.snapshot_id, &ctx.lease_id).unwrap(); + assert!( + r.retention + .store() + .root_covers(&format!("lease:{}", ctx.lease_id)) + ); + { + let mut leases = r.leases.lock().unwrap(); + r.prune_expired_leases(&mut leases, renewed.expires_at_unix + 1); + } + assert!( + !r.retention + .store() + .root_covers(&format!("lease:{}", ctx.lease_id)) + ); + assert_eq!( + r.renew_lease(&ctx.lease_id, 60).unwrap_err().code, + SnapshotErrorCode::LeaseUnknown + ); + assert_eq!( + r.context(&built.snapshot_id).unwrap_err().code, + SnapshotErrorCode::LeaseExpired + ); + } + + #[test] + fn renewal_waiting_for_the_lease_lock_cannot_use_a_pre_expiry_clock_sample() { + let r = Arc::new(isolated_runtime()); + let ctx = r + .insert_context(descriptor(3), "commit", "tree", 60) + .unwrap(); + let clock = Arc::new(AtomicU64::new(9)); + let (blocked_tx, blocked_rx) = sync_channel(1); + let mut leases = r.leases.lock().unwrap(); + leases.get_mut(&ctx.lease_id).unwrap().expires_at_unix = 10; + let renewal = { + let r = Arc::clone(&r); + let clock = Arc::clone(&clock); + let lease_id = ctx.lease_id.clone(); + thread::spawn(move || { + Mst2Runtime::renew_lease_with_clock( + &lease_id, + 60, + || { + // Prove contention before allowing the clock to cross + // the deadline. A pre-lock sample would observe 9. + assert!(matches!(r.leases.try_lock(), Err(TryLockError::WouldBlock))); + blocked_tx.send(()).unwrap(); + r.leases.lock().unwrap() + }, + || clock.load(Ordering::SeqCst), + ) + }) }; - // Retain the snapshot's derived nodes directly, covered by the lease - // root that insert_context created. + blocked_rx + .recv_timeout(Duration::from_secs(5)) + .expect("renewal did not contend for the lease lock"); + clock.store(11, Ordering::SeqCst); + drop(leases); + + assert_eq!( + renewal.join().unwrap().unwrap_err().code, + SnapshotErrorCode::LeaseExpired + ); + assert_eq!( + r.leases + .lock() + .unwrap() + .get(&ctx.lease_id) + .unwrap() + .expires_at_unix, + 10 + ); + assert_eq!(r.leases.lock().unwrap().len(), 1, "no replacement lease"); + assert!(r.collect_retention().unwrap().reaped.is_empty()); + assert!( + !r.retention + .store() + .root_covers(&format!("lease:{}", ctx.lease_id)) + ); + } + + #[test] + fn rfc3339_known_values() { + assert_eq!(rfc3339(0), "1970-01-01T00:00:00Z"); + assert_eq!(rfc3339(1_789_525_722), "2026-09-16T02:28:42Z"); + } + + #[test] + fn lease_lifecycle_drives_retention() { + let r = isolated_runtime(); + // Cover a derived node directly with a manual test lease root. let lease_root = RetentionRoot::Lease("gc-lease-1".to_string()); let page = crate::ceres::snapshot::retention::RetentionNode { id: "page:sha256:test".to_string(), @@ -382,7 +730,7 @@ mod tests { assert_eq!(report.unreachable, vec!["page:sha256:test".to_string()]); assert_eq!(report.reclaimed_bytes, 100); assert!(report.reaped.is_empty(), "fail-closed reaper"); - // The node is DELETING, not gone: a re-resolve re-lifts it. + // The node stays DELETING; a new acquire must not resurrect it. assert_eq!( r.retention.store().node("page:sha256:test").unwrap().state, NodeState::Deleting From 0ef0bd07d7132226e26bf02adf70de9868d1a1df Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 08:14:53 +0800 Subject: [PATCH 02/13] feat(mst2): add durable retention GC schema Add SeaORM retention entities and a forward migration for incoming reference counts and idempotent GC operation replay. --- src/callisto/mod.rs | 4 + src/callisto/mst2_retention_edge.rs | 19 +++ src/callisto/mst2_retention_gc_op.rs | 26 ++++ src/callisto/mst2_retention_node.rs | 25 ++++ src/callisto/mst2_retention_root.rs | 20 +++ src/callisto/prelude.rs | 4 + ...05_000100_add_mst2_retention_durability.rs | 119 ++++++++++++++++++ src/jupiter/migration/mod.rs | 2 + 8 files changed, 219 insertions(+) create mode 100644 src/callisto/mst2_retention_edge.rs create mode 100644 src/callisto/mst2_retention_gc_op.rs create mode 100644 src/callisto/mst2_retention_node.rs create mode 100644 src/callisto/mst2_retention_root.rs create mode 100644 src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs diff --git a/src/callisto/mod.rs b/src/callisto/mod.rs index 3285765e..a841858c 100644 --- a/src/callisto/mod.rs +++ b/src/callisto/mod.rs @@ -60,6 +60,10 @@ pub mod mega_webhook_delivery; pub mod mega_webhook_event_type; pub mod mst2_publication; pub mod mst2_publication_outbox; +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/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..778ec2df --- /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::*; + +#[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/mod.rs b/src/jupiter/migration/mod.rs index 822a1f72..b3e6c010 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -139,6 +139,7 @@ mod m20260921_000100_fix_git_tag_unique; mod m20260923_000100_import_repo_cleanups; mod m20260923_000200_canonicalize_import_repo_paths; mod m20260925_000100_media_paging; +mod m20261005_000100_add_mst2_retention_durability; mod runner; pub use m20260905_000100_add_push_queue::ensure_queue_control_seed; pub use runner::apply_migrations; @@ -266,6 +267,7 @@ impl MigratorTrait for Migrator { Box::new(m20260923_000100_import_repo_cleanups::Migration), Box::new(m20260923_000200_canonicalize_import_repo_paths::Migration), Box::new(m20260925_000100_media_paging::Migration), + Box::new(m20261005_000100_add_mst2_retention_durability::Migration), ] } } From c8482bc55c31b248a5a42b08572eb386cc9b2668 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 08:17:19 +0800 Subject: [PATCH 03/13] test(mst2): include retention migration in ordering check --- src/jupiter/migration/mod.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index b3e6c010..aaee69e7 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -477,13 +477,14 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 3..], + &names[names.len() - 4..], &[ "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_retention_durability".to_string(), ], - "media paging registered last, after the FU-14/16 migrations" + "retention durability registered last, after media paging" ); let db = alias_db().await; From a39be1100e92710add368377a9c19198185c35e2 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:12:24 +0800 Subject: [PATCH 04/13] fix(mst2): bound JSON body reads and preserve typed rejections --- src/api/router/snapshot_content.rs | 6 +- src/api/router/snapshot_request.rs | 50 ++++ src/api/router/snapshot_request_tests.rs | 349 +++++++++++++++++++++++ src/api/router/snapshot_router.rs | 65 +++-- src/ceres/snapshot/error.rs | 5 +- 5 files changed, 450 insertions(+), 25 deletions(-) create mode 100644 src/api/router/snapshot_request.rs create mode 100644 src/api/router/snapshot_request_tests.rs diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index b52dbba6..8f27b5ad 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -14,7 +14,7 @@ use futures::stream::StreamExt; use serde::Deserialize; use serde_json::json; -use super::{abs_view_path, internal, mst2_error_response}; +use super::{abs_view_path, internal, mst2_error_response, request::Mst2Bytes}; use crate::ceres::snapshot::{ chunks::{ChunkProjection, get_or_project}, error::{SnapshotError, SnapshotErrorCode}, @@ -174,7 +174,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() @@ -473,7 +473,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..728162a5 --- /dev/null +++ b/src/api/router/snapshot_request.rs @@ -0,0 +1,50 @@ +//! Bounded raw JSON input for the MST/2 POST surface (spec 14). + +use std::time::Duration; + +use axum::{ + extract::{FromRequest, Request}, + http::StatusCode, +}; +use bytes::Bytes; + +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); + +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..0c898da8 --- /dev/null +++ b/src/api/router/snapshot_request_tests.rs @@ -0,0 +1,349 @@ +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); + // 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("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/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":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 556897e3..d18c5c79 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -19,6 +19,8 @@ use mst2_codec::descriptor; use serde::Deserialize; use serde_json::json; +use request::Mst2Bytes; + use crate::{ api::MonoApiServiceState, ceres::snapshot::{ @@ -72,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; @@ -97,28 +106,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, @@ -216,7 +233,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 ), } })), @@ -230,6 +249,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 { @@ -319,7 +344,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 @@ -483,18 +511,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)) @@ -854,7 +877,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)?; @@ -1011,7 +1034,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/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 From ba5438deee717a07684cd2fbaebc72d8bfd7ddac Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:14:00 +0800 Subject: [PATCH 05/13] fix(mst2): reject duplicate JSON keys including null and escapes --- src/api/router/snapshot_request.rs | 66 +++++++++++++++++++++++- src/api/router/snapshot_request_tests.rs | 2 + src/api/router/snapshot_router.rs | 14 ++--- 3 files changed, 75 insertions(+), 7 deletions(-) diff --git a/src/api/router/snapshot_request.rs b/src/api/router/snapshot_request.rs index 728162a5..ec43af70 100644 --- a/src/api/router/snapshot_request.rs +++ b/src/api/router/snapshot_request.rs @@ -1,12 +1,16 @@ //! Bounded raw JSON input for the MST/2 POST surface (spec 14). -use std::time::Duration; +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}; @@ -18,6 +22,66 @@ pub(super) const JSON_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); /// 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, diff --git a/src/api/router/snapshot_request_tests.rs b/src/api/router/snapshot_request_tests.rs index 0c898da8..0de0c411 100644 --- a/src/api/router/snapshot_request_tests.rs +++ b/src/api/router/snapshot_request_tests.rs @@ -200,6 +200,8 @@ async fn mst2_json_closed_dtos_duplicates_and_malformed_input_do_not_renew() { (&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}"#), diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 36a1fee2..9fa25e6e 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -186,12 +186,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 From 84832260ad007a022d4055e6fd6ad37429d9264c Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:19:59 +0800 Subject: [PATCH 06/13] feat(mst2): add atomic PostgreSQL retention repository --- ...05_000100_add_mst2_retention_durability.rs | 2 +- ...1005_000200_harden_mst2_retention_graph.rs | 76 +++ src/jupiter/migration/mod.rs | 5 +- src/jupiter/storage/mod.rs | 1 + src/jupiter/storage/mst2_retention.rs | 633 ++++++++++++++++++ src/jupiter/storage/mst2_retention_tests.rs | 617 +++++++++++++++++ 6 files changed, 1332 insertions(+), 2 deletions(-) create mode 100644 src/jupiter/migration/m20261005_000200_harden_mst2_retention_graph.rs create mode 100644 src/jupiter/storage/mst2_retention.rs create mode 100644 src/jupiter/storage/mst2_retention_tests.rs diff --git a/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs b/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs index 778ec2df..8f223af1 100644 --- a/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs +++ b/src/jupiter/migration/m20261005_000100_add_mst2_retention_durability.rs @@ -5,7 +5,7 @@ //! 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::*; +use sea_orm_migration::{prelude::*, schema::*}; #[derive(DeriveMigrationName)] pub struct Migration; 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 aaee69e7..47df8efe 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -140,6 +140,7 @@ mod m20260923_000100_import_repo_cleanups; mod m20260923_000200_canonicalize_import_repo_paths; mod m20260925_000100_media_paging; mod m20261005_000100_add_mst2_retention_durability; +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; @@ -268,6 +269,7 @@ impl MigratorTrait for Migrator { Box::new(m20260923_000200_canonicalize_import_repo_paths::Migration), Box::new(m20260925_000100_media_paging::Migration), Box::new(m20261005_000100_add_mst2_retention_durability::Migration), + Box::new(m20261005_000200_harden_mst2_retention_graph::Migration), ] } } @@ -477,12 +479,13 @@ mod tests { async fn import_repo_alias_rows_canonicalized() { let names = migration_names(); assert_eq!( - &names[names.len() - 4..], + &names[names.len() - 5..], &[ "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_retention_durability".to_string(), + "m20261005_000200_harden_mst2_retention_graph".to_string(), ], "retention durability registered last, after media paging" ); diff --git a/src/jupiter/storage/mod.rs b/src/jupiter/storage/mod.rs index deb4fe34..46cb4a0a 100644 --- a/src/jupiter/storage/mod.rs +++ b/src/jupiter/storage/mod.rs @@ -16,6 +16,7 @@ pub mod issue_storage; pub mod lfs_db_storage; pub mod media_paging_storage; pub mod mono_storage; +pub mod mst2_retention; pub mod notification_storage; pub mod object_storage; pub mod oci_db_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..6bd60c15 --- /dev/null +++ b/src/jupiter/storage/mst2_retention_tests.rs @@ -0,0 +1,617 @@ +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) \ + VALUES ('parent', 'page', 'LIVE', 7), ('child', 'page', 'LIVE', 7); \ + INSERT INTO mst2_retention_edge (parent_id, child_id) VALUES ('parent', 'child')", + ) + .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) VALUES ('missing', 'child')" + ) + .await + .is_err() + ); + assert!( + db.execute_unprepared( + "UPDATE mst2_retention_node SET incoming_refs = -1 WHERE node_id = 'child'" + ) + .await + .is_err() + ); + assert!(db.execute_unprepared("INSERT INTO mst2_retention_gc_op (operation_id, node_id, operation, state) VALUES ('bad', 'child', 'REMOVE', 'APPLIED')").await.is_err()); +} + +#[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) \ + VALUES ('a', 'page', 'LIVE', 7), ('b', 'page', 'LIVE', 7); \ + INSERT INTO mst2_retention_edge (parent_id, child_id) VALUES ('a', 'b'), ('b', 'a')", + ) + .await + .unwrap(); + assert!(Migrator::up(&db, None).await.is_err()); + assert_eq!( + mst2_retention_edge::Entity::find() + .count(&db) + .await + .unwrap(), + 2 + ); +} From 01ac68f46579f0ed9661b8a25b50ae7a7162373e Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:27:24 +0800 Subject: [PATCH 07/13] fix(mst2): remove obsolete body extractor import --- src/api/router/snapshot_content.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/api/router/snapshot_content.rs b/src/api/router/snapshot_content.rs index 852f4e4f..f06fd35b 100644 --- a/src/api/router/snapshot_content.rs +++ b/src/api/router/snapshot_content.rs @@ -9,7 +9,6 @@ use axum::{ http::HeaderMap, response::{IntoResponse, Response}, }; -use bytes::Bytes; use futures::stream::StreamExt; use serde::Deserialize; use serde_json::json; From c6c1d5c839f73a717192fec6e403c1ed320148f1 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:28:26 +0800 Subject: [PATCH 08/13] test(mst2): account for integrated native and retention migrations --- src/jupiter/migration/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/jupiter/migration/mod.rs b/src/jupiter/migration/mod.rs index 52a3779b..b385d106 100644 --- a/src/jupiter/migration/mod.rs +++ b/src/jupiter/migration/mod.rs @@ -483,11 +483,13 @@ 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(), ], From 19965e42efb2c471125939e550015116823026cc Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:40:40 +0800 Subject: [PATCH 09/13] fix(mst2): repair native publication compile gates --- src/api/router/snapshot_router.rs | 105 +++++++++--------- .../service/native_publication_push_tests.rs | 3 +- src/jupiter/service/push_queue_service.rs | 3 +- src/jupiter/storage/push_queue_storage.rs | 11 +- 4 files changed, 60 insertions(+), 62 deletions(-) diff --git a/src/api/router/snapshot_router.rs b/src/api/router/snapshot_router.rs index 6f5aef1c..9330837b 100644 --- a/src/api/router/snapshot_router.rs +++ b/src/api/router/snapshot_router.rs @@ -422,55 +422,55 @@ async fn resolve( )) })?; - let (commit_oid, tree_oid, sequence, writer_epoch) = - if state.storage.config().mst2.publication_enabled { - let Some(instance) = state.storage.config().mst2.instance_uuid.as_deref() else { - return Err(mst2_error_response(SnapshotError::new( - SnapshotErrorCode::SnapshotNotReady, - "native publication instance missing", - ))); - }; - let head = state - .storage - .mono_storage() - .read_native_publication_head(instance) - .await - .map_err(|error| { - tracing::error!(error = %error, "native publication observation failed"); - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::SnapshotNotReady, - "native publication is not ready", - )) - })?; - ( - head.root.commit, - head.root.tree, - head.token.sequence.to_string(), - head.token.epoch.to_string(), - ) - } else { - let main = state - .storage - .mono_storage() - .get_main_ref("/") - .await - .map_err(internal)? - .ok_or_else(|| { - mst2_error_response(SnapshotError::new( - SnapshotErrorCode::SnapshotNotReady, - "monorepo main ref missing", - )) - })?; - let sequence = runtime() - .publication_sequence(&main.ref_commit_hash) - .to_string(); - ( - main.ref_commit_hash, - main.ref_tree_hash, - sequence, - "1".to_owned(), - ) + let config = state.storage.config(); + 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( + SnapshotErrorCode::SnapshotNotReady, + "native publication instance missing", + ))); }; + let head = state + .storage + .mono_storage() + .read_native_publication_head(instance) + .await + .map_err(|error| { + tracing::error!(error = %error, "native publication observation failed"); + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::SnapshotNotReady, + "native publication is not ready", + )) + })?; + ( + head.root.commit, + head.root.tree, + head.token.sequence.to_string(), + head.token.epoch.to_string(), + ) + } else { + let main = state + .storage + .mono_storage() + .get_main_ref("/") + .await + .map_err(internal)? + .ok_or_else(|| { + mst2_error_response(SnapshotError::new( + SnapshotErrorCode::SnapshotNotReady, + "monorepo main ref missing", + )) + })?; + let sequence = runtime() + .publication_sequence(&main.ref_commit_hash) + .to_string(); + ( + main.ref_commit_hash, + main.ref_tree_hash, + sequence, + "1".to_owned(), + ) + }; #[cfg(test)] if let Ok((captured, release)) = NATIVE_RESOLVE_BARRIERS.try_with(|value| value.clone()) { @@ -502,13 +502,8 @@ async fn resolve( .await .map_err(mst2_error_response)?; - let built = build_descriptor( - &state.storage.config().mst2, - &view, - &req.scope, - scope_page.page_id, - ) - .map_err(mst2_error_response)?; + 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)?; diff --git a/src/jupiter/service/native_publication_push_tests.rs b/src/jupiter/service/native_publication_push_tests.rs index 9f77b988..6ebb703c 100644 --- a/src/jupiter/service/native_publication_push_tests.rs +++ b/src/jupiter/service/native_publication_push_tests.rs @@ -1,11 +1,10 @@ -use super::*; - use bytes::Bytes; use crate::callisto::{ mst2_native_head, mst2_native_publication, mst2_publication, mst2_publication_outbox, }; use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, Statement}; +use crate::jupiter::utils::converter::FromMegaModel; const NATIVE_INSTANCE: &str = "6ab219b0-4275-45ba-9d7b-7b0b633018cd"; diff --git a/src/jupiter/service/push_queue_service.rs b/src/jupiter/service/push_queue_service.rs index be66fcfd..8d7b33a8 100644 --- a/src/jupiter/service/push_queue_service.rs +++ b/src/jupiter/service/push_queue_service.rs @@ -1381,7 +1381,8 @@ impl PushQueueService { let ctx = push_ctx.ok_or_else(|| { MegaError::Other("native publication requires the real push writer".into()) })?; - let Some(instance) = ctx.storage.config().mst2.instance_uuid.as_deref() else { + let config = ctx.storage.config(); + let Some(instance) = config.mst2.instance_uuid.as_deref() else { return self .b3_fail_merge( txn, diff --git a/src/jupiter/storage/push_queue_storage.rs b/src/jupiter/storage/push_queue_storage.rs index 8f7616b3..5be79779 100644 --- a/src/jupiter/storage/push_queue_storage.rs +++ b/src/jupiter/storage/push_queue_storage.rs @@ -8,7 +8,7 @@ use serde_json::Value as JsonValue; use crate::{ callisto::{ - authz_notify_outbox, mega_refs, push_queue, queue_control, + authz_notify_outbox, push_queue, queue_control, sea_orm_active_enums::{PushQueueKindEnum, PushQueueStatusEnum}, }, common::{errors::MegaError, utils::MEGA_BRANCH_NAME}, @@ -1414,9 +1414,12 @@ mod tests { use serde_json::json; use super::*; - use crate::jupiter::{ - migration::apply_migrations, storage::base_storage::StorageConnector, - tests::test_db_connection, + use crate::{ + callisto::mega_refs, + jupiter::{ + migration::apply_migrations, storage::base_storage::StorageConnector, + tests::test_db_connection, + }, }; async fn storage() -> (tempfile::TempDir, PushQueueStorage) { From 74f4373e691f388d58d7fee6fde9dbc7498d9420 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:40:45 +0800 Subject: [PATCH 10/13] test(mst2): require successful bounded-input lease fixture registration --- src/api/router/snapshot_request_tests.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/api/router/snapshot_request_tests.rs b/src/api/router/snapshot_request_tests.rs index 0de0c411..fc577c85 100644 --- a/src/api/router/snapshot_request_tests.rs +++ b/src/api/router/snapshot_request_tests.rs @@ -62,7 +62,9 @@ impl Fixture { }; 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); + 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() From 04cf79e6ddfaf60cca28689b2de8988eedf0b374 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:57:04 +0800 Subject: [PATCH 11/13] test(mst2): seed retention migration timestamps --- src/jupiter/storage/mst2_retention_tests.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/jupiter/storage/mst2_retention_tests.rs b/src/jupiter/storage/mst2_retention_tests.rs index 6bd60c15..9db9545c 100644 --- a/src/jupiter/storage/mst2_retention_tests.rs +++ b/src/jupiter/storage/mst2_retention_tests.rs @@ -554,8 +554,8 @@ async fn t06b_forward_migration_backfills_old_graph_and_enforces_constraints() { .await .unwrap(); db.execute_unprepared( - "INSERT INTO mst2_retention_node (node_id, kind, state, bytes) \ - VALUES ('parent', 'page', 'LIVE', 7), ('child', 'page', 'LIVE', 7); \ + "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) VALUES ('parent', 'child')", ) .await @@ -600,8 +600,8 @@ async fn t06b_forward_migration_refuses_existing_cycles() { .await .unwrap(); db.execute_unprepared( - "INSERT INTO mst2_retention_node (node_id, kind, state, bytes) \ - VALUES ('a', 'page', 'LIVE', 7), ('b', 'page', 'LIVE', 7); \ + "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) VALUES ('a', 'b'), ('b', 'a')", ) .await From 0873479b9eb04be8c94259ee8320b9ff83d98993 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 10:17:07 +0800 Subject: [PATCH 12/13] test(mst2): seed legacy retention edge timestamps --- src/jupiter/storage/mst2_retention_tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/jupiter/storage/mst2_retention_tests.rs b/src/jupiter/storage/mst2_retention_tests.rs index 9db9545c..b24a58c4 100644 --- a/src/jupiter/storage/mst2_retention_tests.rs +++ b/src/jupiter/storage/mst2_retention_tests.rs @@ -556,7 +556,7 @@ async fn t06b_forward_migration_backfills_old_graph_and_enforces_constraints() { 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) VALUES ('parent', 'child')", + INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('parent', 'child', now())", ) .await .unwrap(); @@ -573,7 +573,7 @@ async fn t06b_forward_migration_backfills_old_graph_and_enforces_constraints() { ); assert!( db.execute_unprepared( - "INSERT INTO mst2_retention_edge (parent_id, child_id) VALUES ('missing', 'child')" + "INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('missing', 'child', now())" ) .await .is_err() @@ -602,7 +602,7 @@ async fn t06b_forward_migration_refuses_existing_cycles() { 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) VALUES ('a', 'b'), ('b', 'a')", + INSERT INTO mst2_retention_edge (parent_id, child_id, created_at) VALUES ('a', 'b', now()), ('b', 'a', now())", ) .await .unwrap(); From 7087a441a3d03ead48b9e702fe240dfffe8e39b5 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 10:18:01 +0800 Subject: [PATCH 13/13] test(mst2): assert the GC receipt completion constraint --- src/jupiter/storage/mst2_retention_tests.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/jupiter/storage/mst2_retention_tests.rs b/src/jupiter/storage/mst2_retention_tests.rs index b24a58c4..a8f7c866 100644 --- a/src/jupiter/storage/mst2_retention_tests.rs +++ b/src/jupiter/storage/mst2_retention_tests.rs @@ -585,7 +585,19 @@ async fn t06b_forward_migration_backfills_old_graph_and_enforces_constraints() { .await .is_err() ); - assert!(db.execute_unprepared("INSERT INTO mst2_retention_gc_op (operation_id, node_id, operation, state) VALUES ('bad', 'child', 'REMOVE', 'APPLIED')").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]