From 648d82f74f63b85c797199bb0adc19af58af84cc Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 07:55:17 +0800 Subject: [PATCH 1/7] 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 2/7] 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 3/7] 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 84832260ad007a022d4055e6fd6ad37429d9264c Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:19:59 +0800 Subject: [PATCH 4/7] 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 04cf79e6ddfaf60cca28689b2de8988eedf0b374 Mon Sep 17 00:00:00 2001 From: Luxian Date: Mon, 5 Oct 2026 09:57:04 +0800 Subject: [PATCH 5/7] 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 6/7] 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 7/7] 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]