From 9fcfb424fbdca747aec439a928c2c835547597a5 Mon Sep 17 00:00:00 2001 From: Weisson Date: Mon, 17 Aug 2026 23:50:18 +0800 Subject: [PATCH] feat(blaze): prune unreachable checkpoint history Add POST /v1/sandboxes/{id}/checkpoints/prune. The endpoint verifies the committed catalog, protects HEAD together with its full parent chain and the ancestor chains of any explicitly listed identifiers, and removes only unreachable entries. Publish a named tombstone before renaming a candidate directory so a restart can identify the pruned candidate and finish removing it. The paired owner file records the sandbox, checkpoint, nonce, and directory identity, and restart cleanup already treats prune tombstones and owner records the same way as capture staging leftovers. Document the route, protection rules, and restart behavior in the English and Chinese operator README and runtime user guide. Signed-off-by: Weisson --- docs/user-guide/en/runtime/blaze.md | 48 +- docs/user-guide/zh/runtime/blaze.md | 40 +- src/blaze/README.md | 14 +- src/blaze/README_zh.md | 14 +- src/blaze/crates/blazed/src/api.rs | 199 ++++++ .../crates/blazed/src/checkpoint_store.rs | 614 ++++++++++++++++++ .../crates/blazed/src/sandbox/checkpoint.rs | 86 +++ 7 files changed, 1006 insertions(+), 9 deletions(-) diff --git a/docs/user-guide/en/runtime/blaze.md b/docs/user-guide/en/runtime/blaze.md index 8cb98f6820..5611370a0e 100644 --- a/docs/user-guide/en/runtime/blaze.md +++ b/docs/user-guide/en/runtime/blaze.md @@ -96,7 +96,8 @@ to execute commands, read files, and write files inside them. Sandbox destruction uses `DELETE /v1/sandboxes/{id}`. Checkpoint capture and history use `POST /v1/sandboxes/{id}/checkpoint` and -`GET /v1/sandboxes/{id}/checkpoints`. +`GET /v1/sandboxes/{id}/checkpoints`, and +`POST /v1/sandboxes/{id}/checkpoints/prune` removes unreachable history. ## Host Integration Boundary @@ -283,8 +284,49 @@ resumes the backend, and leaves the sandbox running. If Blaze cannot prove the publication, HEAD update, persistence, or backend-resume outcome, it retains the durable record and reports `RecoveryRequired`; do not retry capture until the sandbox has been reconciled or destroyed. A committed checkpoint that did -not become HEAD can still appear in history with `is_head: false`. This release -does not provide checkpoint restore, deletion, or pruning APIs. +not become HEAD can still appear in history with `is_head: false`. + +### Pruning unreachable history + +```http +POST /v1/sandboxes/{id}/checkpoints/prune +``` + +Prune removes every committed checkpoint that is neither reachable from the +current HEAD nor listed in the request's `protected` array. The request body +is optional; when present it must be a JSON object with only the `protected` +field: + +```json +{ "protected": ["ckpt-11111111-1111-4111-8111-111111111111"] } +``` + +An empty or absent body is equivalent to `{"protected": []}`. Each protected +identifier must name a currently committed checkpoint; an unknown identifier +is rejected with HTTP 400 before any deletion. The response reports the number +of removed checkpoints and their identifiers: + +```json +{ "count": 1, "removed": ["ckpt-22222222-2222-4222-8222-222222222222"] } +``` + +Prune protects the current HEAD, its full parent chain, and the ancestor chain +of every explicitly protected identifier. It only publishes each removal after +a named tombstone directory has replaced the candidate; the tombstone and its +paired owner file let subsequent restart cleanup finish a partially completed +prune or preserve a concurrently reappearing candidate rather than silently +discarding it. The endpoint acquires the same per-sandbox operation lock as +capture and list, so prune serializes with any other checkpoint operation. + +If a failure occurs after a candidate has already left the committed catalog, +Blaze reports the confirmed removals, marks the sandbox `RecoveryRequired`, and +returns an error instead of a success body. Prune requests for a +`RecoveryRequired` sandbox are rejected with HTTP 409: destroy the sandbox or +restart the daemon so the retained `.prune.*` scratch is reconciled before +pruning again. + +This release still does not provide checkpoint restore or per-checkpoint +deletion outside prune. ## Storage Artifact Synchronization diff --git a/docs/user-guide/zh/runtime/blaze.md b/docs/user-guide/zh/runtime/blaze.md index 4001c8cfd1..e509faa825 100644 --- a/docs/user-guide/zh/runtime/blaze.md +++ b/docs/user-guide/zh/runtime/blaze.md @@ -80,7 +80,8 @@ Blaze 通过 `/v1/sandboxes` 提供沙箱生命周期和客户机操作。客户 命名空间列出、创建、查看和删除沙箱，以及在沙箱内执行命令、读取文件和写入 文件。销毁沙箱使用 `DELETE /v1/sandboxes/{id}`。检查点捕获与历史查询分别使用 `POST /v1/sandboxes/{id}/checkpoint` 和 -`GET /v1/sandboxes/{id}/checkpoints`。 +`GET /v1/sandboxes/{id}/checkpoints`； +`POST /v1/sandboxes/{id}/checkpoints/prune` 用于删除不可达的历史。 ## 主机集成边界 @@ -240,8 +241,41 @@ sandbox 或修改其生命周期记录前返回 HTTP 501。 能够确认发生在发布前的失败会删除临时数据、恢复后端，并让 sandbox 保持运行。 如果 Blaze 无法确认发布、HEAD 更新、持久化或后端恢复的结果，则会保留持久记录并 报告 `RecoveryRequired`；在 sandbox 完成恢复处理或销毁前，不应重试捕获。已经提交但 -未成为 HEAD 的检查点仍可能出现在历史列表中，其 `is_head` 为 `false`。当前版本 -不提供检查点恢复、删除或清理接口。 +未成为 HEAD 的检查点仍可能出现在历史列表中，其 `is_head` 为 `false`。 + +### 裁剪不可达的检查点历史 + +```http +POST /v1/sandboxes/{id}/checkpoints/prune +``` + +Prune 会删除既不属于当前 HEAD 父链、也不在请求 `protected` 数组中的所有已提交 +检查点。请求体可选；若存在则必须是仅包含 `protected` 字段的 JSON 对象： + +```json +{ "protected": ["ckpt-11111111-1111-4111-8111-111111111111"] } +``` + +空请求体等价于 `{"protected": []}`。`protected` 中每个标识符都必须对应当前已 +提交的检查点；未知标识符会在任何删除前以 HTTP 400 拒绝。响应给出被删除检查点 +的数量与列表： + +```json +{ "count": 1, "removed": ["ckpt-22222222-2222-4222-8222-222222222222"] } +``` + +Prune 会保护当前 HEAD、其完整父链，以及每个显式保护标识符的祖先链。删除只在 +候选目录被一个命名的墓碑目录替换之后发布；墓碑及其配对的 owner 文件让重启清理 +能够完成半途终止的 prune，或在候选目录被并发重新出现时保留它，而不是被静默 +丢弃。该接口占用与捕获、查询相同的 sandbox 操作锁，因此 prune 与其他任何检查点 +操作串行化。 + +如果失败发生在某个候选已经离开已提交目录之后，Blaze 会报告已确认删除的列表、把 +sandbox 标记为 `RecoveryRequired`，并返回错误而不是成功响应。对处于 +`RecoveryRequired` 的 sandbox 发起 prune 会以 HTTP 409 拒绝：需要先销毁该 +sandbox 或重启 daemon，让残留的 `.prune.*` 得到清理，然后才能再次 prune。 + +当前版本仍不提供除 prune 之外的检查点恢复或按标识单独删除。 ## 存储制品同步 diff --git a/src/blaze/README.md b/src/blaze/README.md index 8dd556beb4..d3fd92b09b 100644 --- a/src/blaze/README.md +++ b/src/blaze/README.md @@ -153,6 +153,7 @@ Blaze exposes sandbox lifecycle and guest operations through `/v1/sandboxes`. | POST | `/v1/sandboxes/{id}/write` | Replace a guest file | | POST | `/v1/sandboxes/{id}/checkpoint` | Capture a full checkpoint | | GET | `/v1/sandboxes/{id}/checkpoints` | List committed checkpoint history | +| POST | `/v1/sandboxes/{id}/checkpoints/prune` | Remove unreachable checkpoint history | | GET | `/v1/pools` | Reserved; returns `501` | | GET | `/v1/pools/{backend}/{class}` | Reserved; returns `501` | | POST | `/v1/pools/{backend}/{class}/drain` | Reserved; returns `501` | @@ -258,7 +259,18 @@ sandbox state. `GET /v1/sandboxes/{id}/checkpoints` returns committed history summaries, including parentage, logical size, current-HEAD status, and HEAD reachability. -This release does not provide checkpoint restore or deletion. + +`POST /v1/sandboxes/{id}/checkpoints/prune` deletes checkpoints that are not +reachable from the current HEAD and are not listed in the optional +`protected` array of the request body. The body may be empty (equivalent to +`{"protected": []}`) or must be a JSON object with only `protected`. The +response reports how many checkpoints were removed and their identifiers. The +current HEAD and its full parent chain, together with the ancestor chains of +every protected identifier, are never removed. + +This release does not provide checkpoint restore or deletion of individual +checkpoints; use prune to reclaim unreachable branches. + See the [checkpoint capture user guide](../../docs/user-guide/en/runtime/blaze.md#checkpoint-capture-and-history) for response fields, current backend support, and failure handling. diff --git a/src/blaze/README_zh.md b/src/blaze/README_zh.md index fe7fa67ef6..199689b11f 100644 --- a/src/blaze/README_zh.md +++ b/src/blaze/README_zh.md @@ -143,6 +143,7 @@ Blaze 通过 `/v1/sandboxes` 提供沙箱生命周期和客户机操作。 | POST | `/v1/sandboxes/{id}/write` | 替换 guest 文件 | | POST | `/v1/sandboxes/{id}/checkpoint` | 捕获完整检查点 | | GET | `/v1/sandboxes/{id}/checkpoints` | 列出已提交的检查点历史 | +| POST | `/v1/sandboxes/{id}/checkpoints/prune` | 删除不可达的检查点历史 | | GET | `/v1/pools` | 预留接口；返回 `501` | | GET | `/v1/pools/{backend}/{class}` | 预留接口；返回 `501` | | POST | `/v1/pools/{backend}/{class}/drain` | 预留接口；返回 `501` | @@ -228,8 +229,17 @@ daemon 才会逐个处理未结束的 sandbox。后续逐项恢复期间，如 改变 sandbox 状态前返回 HTTP 501。 `GET /v1/sandboxes/{id}/checkpoints` 返回已提交检查点的历史摘要，包括父检查点、 -逻辑大小、是否为当前 HEAD，以及能否从 HEAD 到达。当前版本不提供检查点恢复或 -删除接口。响应字段、当前后端支持情况和失败处理方式参见 +逻辑大小、是否为当前 HEAD，以及能否从 HEAD 到达。 + +`POST /v1/sandboxes/{id}/checkpoints/prune` 会删除既不属于当前 HEAD 的父链、也 +不在请求体 `protected` 数组中列出的检查点。请求体可以为空（等价于 +`{"protected": []}`），或必须是仅包含 `protected` 字段的 JSON 对象。响应给出 +被删除检查点的数量和标识列表。当前 HEAD 及其完整父链，加上 `protected` 中每个 +标识符的完整父链，永远不会被删除。 + +本版本不提供检查点恢复或按标识单独删除；使用 prune 可回收不可达的分支。 + +响应字段、当前后端支持情况和失败处理方式参见 [检查点捕获用户指南](../../docs/user-guide/zh/runtime/blaze.md#检查点捕获与历史)。 ### Guest 操作 diff --git a/src/blaze/crates/blazed/src/api.rs b/src/blaze/crates/blazed/src/api.rs index 564873d02d..4a7fc61cdd 100644 --- a/src/blaze/crates/blazed/src/api.rs +++ b/src/blaze/crates/blazed/src/api.rs @@ -142,6 +142,9 @@ async fn dispatch( ("POST", ["v1", "sandboxes", id, "write"]) => write_sandbox_file(state, id, &body).await, ("POST", ["v1", "sandboxes", id, "checkpoint"]) => checkpoint(state, id).await, ("GET", ["v1", "sandboxes", id, "checkpoints"]) => list_checkpoints(state, id).await, + ("POST", ["v1", "sandboxes", id, "checkpoints", "prune"]) => { + prune_checkpoints(state, id, &body).await + } ("DELETE", ["v1", "sandboxes", id]) => destroy_sandbox(state, id).await, ("GET", ["v1", "pools"]) | ("GET", ["v1", "pools", _, _]) @@ -340,6 +343,35 @@ async fn list_checkpoints(state: &Arc, id: &str) -> Result, +} + +async fn prune_checkpoints( + state: &Arc, + id: &str, + body: &[u8], +) -> Result>> { + let uuid = parse_uuid(id)?; + let request: PruneRequest = if body.is_empty() { + PruneRequest::default() + } else { + serde_json::from_slice(body) + .map_err(|error| BlazeDaemonError::BadRequest(format!("invalid prune body: {error}")))? + }; + let removed = state + .manager + .prune_checkpoints(uuid, request.protected) + .await?; + json_ok(&json!({ + "count": removed.len(), + "removed": removed, + })) +} + async fn destroy_sandbox(state: &Arc, id: &str) -> Result>> { let uuid = parse_uuid(id)?; state.manager.destroy(uuid).await?; @@ -1838,6 +1870,128 @@ mod tests { ); } + #[tokio::test] + async fn prune_route_reports_no_removals_when_history_is_all_reachable() { + let temp = tempfile::tempdir().expect("temp"); + let config = test_config(&temp); + let storage: Arc = Arc::new(FileStorageProvider::with_images( + config.storage.images_dir.clone(), + config.storage.instances_dir.clone(), + )); + let state = build_test_state( + config, + test_policy(BackendKind::Mock), + spawners(BackendKind::Mock, Arc::new(MockSpawner)), + BackendKind::Mock, + storage, + ); + let created = created_json(&state, &test_request()).await; + let id = created["instance"]["id"].as_str().expect("id"); + let uuid = Uuid::parse_str(id).expect("uuid"); + let slot = write_checkpoint_fixture(&state, id).await; + + let root = state + .manager + .checkpoint(uuid) + .await + .expect("root checkpoint") + .id; + tokio::fs::write(&slot.rootfs_path, b"second-rootfs") + .await + .expect("second rootfs"); + let head = state + .manager + .checkpoint(uuid) + .await + .expect("head checkpoint") + .id; + + let (status, response) = dispatched_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoints/prune"), + Vec::new(), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(response["count"], 0); + assert!(response["removed"].as_array().expect("removed").is_empty()); + + let listed = state + .manager + .list_checkpoints(uuid) + .await + .expect("list after prune"); + let remaining: std::collections::HashSet = + listed.into_iter().map(|info| info.id).collect(); + assert!(remaining.contains(&root)); + assert!(remaining.contains(&head)); + } + + #[tokio::test] + async fn prune_route_rejects_unknown_body_fields() { + let temp = tempfile::tempdir().expect("temp"); + let config = test_config(&temp); + let storage: Arc = Arc::new(FileStorageProvider::with_images( + config.storage.images_dir.clone(), + config.storage.instances_dir.clone(), + )); + let state = build_test_state( + config, + test_policy(BackendKind::Mock), + spawners(BackendKind::Mock, Arc::new(MockSpawner)), + BackendKind::Mock, + storage, + ); + let created = created_json(&state, &test_request()).await; + let id = created["instance"]["id"].as_str().expect("id"); + + let (status, body) = handled_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoints/prune"), + br#"{"unknown_field":true}"#.to_vec(), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["status"], 400); + } + + #[tokio::test] + async fn prune_route_rejects_unknown_protected_checkpoint_as_bad_request() { + let temp = tempfile::tempdir().expect("temp"); + let config = test_config(&temp); + let storage: Arc = Arc::new(FileStorageProvider::with_images( + config.storage.images_dir.clone(), + config.storage.instances_dir.clone(), + )); + let state = build_test_state( + config, + test_policy(BackendKind::Mock), + spawners(BackendKind::Mock, Arc::new(MockSpawner)), + BackendKind::Mock, + storage, + ); + let created = created_json(&state, &test_request()).await; + let id = created["instance"]["id"].as_str().expect("id"); + + let missing = format!("ckpt-{}", Uuid::new_v4()); + let body_bytes = format!(r#"{{"protected":["{missing}"]}}"#).into_bytes(); + let (status, body) = handled_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoints/prune"), + body_bytes, + ) + .await; + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "canonical but uncommitted protected id must surface as 400" + ); + assert_eq!(body["status"], 400); + } + #[cfg(feature = "test-failpoints")] #[tokio::test] async fn checkpoint_cleanup_failure_keeps_destroy_recoverable() { @@ -1884,6 +2038,51 @@ mod tests { assert!(!checkpoint_namespace.exists()); } + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn prune_route_rejects_a_recovery_required_sandbox() { + let temp = tempfile::tempdir().expect("temp"); + let state = mock_state(&temp); + let created = created_json(&state, &test_request()).await; + let id = created["instance"]["id"].as_str().expect("id"); + let uuid = Uuid::parse_str(id).expect("uuid"); + write_checkpoint_fixture(&state, id).await; + state + .manager + .checkpoint(uuid) + .await + .expect("seed checkpoint"); + + // Drive the sandbox into RecoveryRequired through an existing failure + // path so the prune guard sees the durable state, not a synthetic one. + let hook = crate::failpoint::TestFailpoint::new(&[ + "checkpoint-store-sandbox-remove-before-unlink", + ]); + hook.run(state.manager.destroy(uuid)) + .await + .expect_err("destroy must fail and mark recovery-required"); + assert_eq!( + state.manager.get(uuid).expect("lifecycle").state, + SandboxState::RecoveryRequired + ); + + // Prune keeps no operation journal, so this must be rejected by the + // explicit recovery-required guard rather than silently succeeding. + let (status, body) = handled_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoints/prune"), + Vec::new(), + ) + .await; + assert_eq!( + status, + StatusCode::CONFLICT, + "a recovery-required sandbox must not accept prune retries" + ); + assert_eq!(body["status"], 409); + } + #[cfg(feature = "test-failpoints")] #[tokio::test] async fn checkpoint_snapshot_failure_resumes_and_clears_the_journal() { diff --git a/src/blaze/crates/blazed/src/checkpoint_store.rs b/src/blaze/crates/blazed/src/checkpoint_store.rs index 1e2f0460ab..c568c09b25 100644 --- a/src/blaze/crates/blazed/src/checkpoint_store.rs +++ b/src/blaze/crates/blazed/src/checkpoint_store.rs @@ -36,6 +36,9 @@ const METADATA_FILE: &str = "metadata.json"; const HEAD_FILE: &str = "HEAD"; const STAGING_SUFFIX: &str = ".tmp"; const TOMBSTONE_SUFFIX: &str = ".tombstone"; +const PRUNE_TOMBSTONE_PREFIX: &str = ".prune."; +const PRUNE_OWNER_SUFFIX: &str = ".owner"; +const PRUNE_OWNER_FORMAT_VERSION: u32 = 1; const ABORT_TOMBSTONE_PREFIX: &str = ".abort."; const CHECKPOINT_DIRECTORY_MODE: Mode = Mode::RWXU; const CHECKPOINT_FILE_MODE: Mode = Mode::RUSR.union(Mode::WUSR); @@ -76,6 +79,27 @@ pub enum CheckpointStoreError { /// Convenient result type for checkpoint catalog operations. pub type Result = std::result::Result; +/// Outcome of a prune sweep. +/// +/// A [`PruneOutcome::Complete`] result carries the removed identifier list only +/// when every candidate reached full cleanup (tombstone unlink and owner file +/// removal). A [`PruneOutcome::PartialCleanup`] result indicates that at least +/// one candidate was already removed from the committed catalog before a later +/// step failed, so callers must treat the sandbox as recovery-required until +/// the retained `.prune.*` scratch is cleaned up by a destroy or a restart. +#[derive(Debug)] +pub enum PruneOutcome { + /// All candidates were fully removed and no prune scratch remains. + Complete { removed: Vec }, + /// One or more candidates were removed from the committed catalog before a + /// later cleanup step failed. `removed` is the confirmed prefix; `source` + /// is the failure that stopped the sweep. + PartialCleanup { + removed: Vec, + source: Box, + }, +} + /// Failure while creating a checkpoint stage. #[derive(Debug, Error)] #[error("{source}")] @@ -234,6 +258,41 @@ struct OwnedArtifact { file: File, } +/// Persisted owner record for an in-flight prune tombstone. +/// +/// A prune rename is committed only after both the tombstone directory and its +/// paired owner file are visible. Restart cleanup uses the owner to prove that +/// the tombstone belongs to this catalog before removing either object. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct PruneTombstoneOwner { + format_version: u32, + sandbox_id: Uuid, + checkpoint_id: String, + nonce: Uuid, + directory_device: u64, + directory_inode: u64, +} + +/// Identity of a single prune tombstone/owner pair. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] +struct PruneScratchKey { + checkpoint_id: String, + nonce: Uuid, +} + +/// Result of sweeping one prune candidate. +enum CandidateSweep { + /// The candidate left the committed catalog and no scratch remains. + Removed, + /// Nothing changed for this candidate; the catalog and scratch are intact. + Retained(CheckpointStoreError), + /// The candidate may already have left the committed catalog, and prune + /// scratch is still on disk. Only destroy or startup reconciliation can + /// finish the cleanup. + RemovedWithRetainedScratch { source: CheckpointStoreError }, +} + struct VerifiedCheckpoint { #[cfg(test)] metadata: CheckpointMetadata, @@ -313,6 +372,9 @@ impl VerifiedCheckpoint { #[cfg(test)] type BeforePublishRevalidation = Arc>>>; +#[cfg(test)] +type BeforePruneRename = Arc>>>; + /// Filesystem-backed checkpoint catalog. #[derive(Clone)] pub struct CheckpointStore { @@ -322,6 +384,8 @@ pub struct CheckpointStore { before_publish_revalidation: BeforePublishRevalidation, #[cfg(test)] verified_checkpoint_calls: Arc, + #[cfg(test)] + before_prune_rename: BeforePruneRename, } impl std::fmt::Debug for CheckpointStore { @@ -343,6 +407,8 @@ impl CheckpointStore { before_publish_revalidation: Arc::new(Mutex::new(None)), #[cfg(test)] verified_checkpoint_calls: Arc::new(AtomicUsize::new(0)), + #[cfg(test)] + before_prune_rename: Arc::new(Mutex::new(None)), } } @@ -736,6 +802,319 @@ impl CheckpointStore { sync_directory(sandbox).map_err(CheckpointHeadError::unknown) } + /// Remove committed checkpoints that neither belong to a protected lineage + /// nor to the current HEAD's parent chain. + /// + /// Each removal publishes a named tombstone before unlinking the candidate + /// directory. The paired owner file lets restart cleanup prove the + /// tombstone belongs to this catalog. `renameat NOREPLACE` prevents + /// silently discarding a concurrent replacement, and revalidation between + /// planning and rename rejects a catalog state that has changed since the + /// plan was built. Returns [`PruneOutcome`]: `Complete` when every removed + /// candidate finished tombstone unlink and owner cleanup, or + /// `PartialCleanup` when at least one candidate was removed from the + /// committed catalog before a later step failed. + pub fn prune_preserving(&self, sandbox_id: Uuid, protected: &[String]) -> Result { + let catalog_root = self.root()?; + let Some(sandbox) = optional_child_directory(&catalog_root, &sandbox_id.to_string())? + else { + return Ok(PruneOutcome::Complete { + removed: Vec::new(), + }); + }; + let catalog = self.load_catalog(&sandbox, sandbox_id)?; + let head = self.read_head_id_from(&sandbox)?; + let mut keep = HashSet::new(); + if let Some(head_id) = head.as_deref() { + keep.extend(lineage_from(&catalog, head_id)?); + } + for checkpoint_id in protected { + validate_checkpoint_id(checkpoint_id)?; + if !catalog.contains_key(checkpoint_id) { + // Callers must validate this before invoking `prune_preserving` + // so that a canonical but unknown identifier can be surfaced as + // a bad request rather than an internal error. Keep an assertion + // here to make the invariant visible during in-crate use. + return Err(invariant(format!( + "protected checkpoint {checkpoint_id} is not committed" + ))); + } + keep.extend(lineage_from(&catalog, checkpoint_id)?); + } + + let mut candidate_ids: Vec = catalog + .keys() + .filter(|checkpoint_id| !keep.contains(*checkpoint_id)) + .cloned() + .collect(); + candidate_ids.sort(); + + let mut removed = Vec::with_capacity(candidate_ids.len()); + for checkpoint_id in candidate_ids { + // Any failure in this iteration must preserve the confirmed + // `removed` prefix: once an earlier candidate has left the committed + // catalog, a plain error would report an ordinary server failure and + // leave the sandbox usable while retained scratch still needs + // reconciliation. `sweep_candidate` therefore reports a plain error + // only when nothing has been removed yet. + let sweep = self.sweep_candidate( + &sandbox, + sandbox_id, + &catalog, + head.as_deref(), + &checkpoint_id, + ); + match sweep { + CandidateSweep::Removed => removed.push(checkpoint_id), + CandidateSweep::Retained(source) => { + if removed.is_empty() { + return Err(source); + } + return Ok(PruneOutcome::PartialCleanup { + removed, + source: Box::new(source), + }); + } + CandidateSweep::RemovedWithRetainedScratch { source } => { + removed.push(checkpoint_id); + return Ok(PruneOutcome::PartialCleanup { + removed, + source: Box::new(source), + }); + } + } + } + Ok(PruneOutcome::Complete { removed }) + } + + /// Sweep one prune candidate. + /// + /// Returns whether the candidate left the committed catalog, and whether any + /// prune scratch is still retained on disk. `Retained` means nothing changed + /// for this candidate, so the caller may surface the failure directly when no + /// earlier candidate has been removed. + fn sweep_candidate( + &self, + sandbox: &OwnedStateDirectory, + sandbox_id: Uuid, + catalog: &HashMap, + head: Option<&str>, + checkpoint_id: &str, + ) -> CandidateSweep { + macro_rules! retain { + ($expression:expr) => { + match $expression { + Ok(value) => value, + Err(source) => return CandidateSweep::Retained(source), + } + }; + } + + let expected = match catalog.get(checkpoint_id).cloned() { + Some(expected) => expected, + None => { + return CandidateSweep::Retained(invariant( + "candidate disappeared from validated plan", + )); + } + }; + let candidate = retain!(self.load_checkpoint_metadata(sandbox, sandbox_id, checkpoint_id)); + if candidate.metadata != expected { + return CandidateSweep::Retained(invariant(format!( + "checkpoint candidate {checkpoint_id} metadata changed after planning" + ))); + } + // Revalidate the current HEAD before publishing the tombstone so + // the plan cannot outlive a concurrent HEAD change. + let current_head = retain!(self.read_head_id_from(sandbox)); + if current_head.as_deref() != head { + return CandidateSweep::Retained(invariant( + "checkpoint HEAD changed after the prune plan was validated", + )); + } + if let Some(head_id) = current_head.as_deref() + && head_id == checkpoint_id + { + return CandidateSweep::Retained(invariant( + "checkpoint HEAD appeared after the prune plan was validated", + )); + } + + let nonce = Uuid::new_v4(); + let key = PruneScratchKey { + checkpoint_id: checkpoint_id.to_string(), + nonce, + }; + let tombstone_name = prune_tombstone_name(&key); + let owner_name = prune_owner_name(&key); + let directory_stat = retain!(fstat(candidate.directory.descriptor()).map_err(|source| { + io_error( + "inspect checkpoint prune candidate", + candidate.directory.configured_path(), + std::io::Error::from(source), + ) + })); + let directory_device = retain!(prune_identity_component( + directory_stat.st_dev, + "device", + candidate.directory.configured_path(), + )); + let directory_inode = retain!(prune_identity_component( + directory_stat.st_ino, + "inode", + candidate.directory.configured_path(), + )); + let owner_record = PruneTombstoneOwner { + format_version: PRUNE_OWNER_FORMAT_VERSION, + sandbox_id, + checkpoint_id: checkpoint_id.to_string(), + nonce, + directory_device, + directory_inode, + }; + // Publishing the owner file creates prune scratch. Any failure between + // creation and the rename must remove that scratch, otherwise a later + // `Complete` result would contradict its documented invariant. + let owner = retain!(write_json_new(sandbox, &owner_name, &owner_record)); + if let Err(source) = validate_checkpoint_artifact_owner(&owner) + .and_then(|()| sync_directory(sandbox)) + .and_then(|()| { + #[cfg(test)] + self.run_before_prune_rename(); + let after_head = self.read_head_id_from(sandbox)?; + if after_head.as_deref() != head { + return Err(invariant( + "checkpoint HEAD changed between planning and tombstone rename", + )); + } + require_linked_directory(sandbox, checkpoint_id, &candidate.directory)?; + candidate.require_linked(sandbox, checkpoint_id)?; + checkpoint_store_failpoint( + "checkpoint-prune-before-rename", + &sandbox.configured_path().join(checkpoint_id), + ) + }) + { + return match remove_owned_file(sandbox, &owner_name, owner) + .and_then(|()| sync_directory(sandbox)) + { + Ok(()) => CandidateSweep::Retained(source), + Err(cleanup) => CandidateSweep::RemovedWithRetainedScratch { + source: invariant(format!( + "{source}; checkpoint prune owner cleanup failed: {cleanup}" + )), + }, + }; + } + + let rename_error = renameat_with( + sandbox.descriptor(), + checkpoint_id, + sandbox.descriptor(), + tombstone_name.as_str(), + RenameFlags::NOREPLACE, + ) + .err() + .map(|source| { + io_error( + "tombstone pruned checkpoint", + sandbox.configured_path().join(&tombstone_name), + std::io::Error::from(source), + ) + }); + + // Probe the namespace to learn whether the rename actually applied. + // These probes are themselves fallible: a transient I/O error here + // leaves the outcome unknown while the candidate may already have + // moved to `.prune.*`, so treat a probe failure as an uncertain + // post-rename boundary rather than a plain store error. + let probe = (|| -> Result<(bool, bool, bool)> { + let source_link = optional_child_directory(sandbox, checkpoint_id)?; + let tombstone_link = optional_child_directory(sandbox, &tombstone_name)?; + let source_present = source_link.is_some(); + let source_matches = source_link + .as_ref() + .map(|source| same_directory(source, &candidate.directory)) + .transpose()? + .unwrap_or(false); + let tombstone_matches = tombstone_link + .as_ref() + .map(|tombstone| same_directory(tombstone, &candidate.directory)) + .transpose()? + .unwrap_or(false); + Ok((source_present, source_matches, tombstone_matches)) + })(); + let (source_present, source_matches, tombstone_matches) = match probe { + Ok(probe) => probe, + Err(source) => { + return CandidateSweep::RemovedWithRetainedScratch { + source: invariant(format!( + "checkpoint prune rename verification failed: {source}{}", + rename_error + .as_ref() + .map(|error| format!("; rename reported: {error}")) + .unwrap_or_default() + )), + }; + } + }; + + if !source_present && tombstone_matches { + // The namespace proves the rename applied even if the kernel + // still reported an error. + } else if source_matches && !tombstone_matches { + let source = rename_error.unwrap_or_else(|| { + invariant(format!( + "checkpoint prune rename reported success but {checkpoint_id} remained committed" + )) + }); + return match remove_owned_file(sandbox, &owner_name, owner) + .and_then(|()| sync_directory(sandbox)) + { + Ok(()) => CandidateSweep::Retained(source), + Err(cleanup) => CandidateSweep::RemovedWithRetainedScratch { + source: invariant(format!( + "{source}; checkpoint prune owner cleanup failed: {cleanup}" + )), + }, + }; + } else { + // Neither state is proven: the candidate may or may not still be + // committed, so require reconciliation rather than reporting a + // plain error. + return CandidateSweep::RemovedWithRetainedScratch { + source: invariant(format!( + "checkpoint prune rename has an uncertain namespace outcome{}", + rename_error + .as_ref() + .map(|error| format!(": {error}")) + .unwrap_or_default() + )), + }; + } + + // Post-rename: the candidate is no longer part of the committed + // catalog. Any later cleanup failure leaves `.prune.*` scratch on + // disk that only destroy or startup reconciliation can remove. + let confirm_removal = || -> Result<()> { + require_linked_directory(sandbox, &tombstone_name, &candidate.directory)?; + sync_directory(sandbox)?; + crate::failpoint::pause_blocking("checkpoint-prune-after-tombstone-sync"); + checkpoint_store_failpoint( + "checkpoint-prune-after-tombstone", + &sandbox.configured_path().join(&tombstone_name), + )?; + remove_owned_directory(sandbox, &tombstone_name, candidate.directory)?; + sync_directory(sandbox)?; + remove_owned_file(sandbox, &owner_name, owner)?; + sync_directory(sandbox) + }; + match confirm_removal() { + Ok(()) => CandidateSweep::Removed, + Err(source) => CandidateSweep::RemovedWithRetainedScratch { source }, + } + } + /// Return the persisted HEAD, if present. pub fn read_head(&self, sandbox_id: Uuid) -> Result> { let catalog = self.root()?; @@ -1120,6 +1499,30 @@ impl CheckpointStore { } } + #[cfg(test)] + #[allow(dead_code)] + fn set_before_prune_rename(&self, hook: F) + where + F: FnOnce() + Send + 'static, + { + *self + .before_prune_rename + .lock() + .expect("checkpoint prune test hook lock") = Some(Box::new(hook)); + } + + #[cfg(test)] + fn run_before_prune_rename(&self) { + if let Some(hook) = self + .before_prune_rename + .lock() + .expect("checkpoint prune test hook lock") + .take() + { + hook(); + } + } + #[cfg(test)] fn configured_root(&self) -> PathBuf { self.root() @@ -1604,6 +2007,20 @@ fn classify_scratch_name(name: &str) -> Result> { parse_uuid_component(nonce, "checkpoint tombstone")?; return Ok(Some(ScratchKind::Directory)); } + if let Some(body) = name + .strip_prefix(PRUNE_TOMBSTONE_PREFIX) + .and_then(|name| name.strip_suffix(TOMBSTONE_SUFFIX)) + { + parse_prune_scratch_key(body, name)?; + return Ok(Some(ScratchKind::Directory)); + } + if let Some(body) = name + .strip_prefix(PRUNE_TOMBSTONE_PREFIX) + .and_then(|name| name.strip_suffix(PRUNE_OWNER_SUFFIX)) + { + parse_prune_scratch_key(body, name)?; + return Ok(Some(ScratchKind::File)); + } Ok(None) } @@ -1618,6 +2035,45 @@ fn parse_uuid_component(value: &str, label: &str) -> Result { Ok(uuid) } +fn prune_tombstone_name(key: &PruneScratchKey) -> String { + format!( + "{PRUNE_TOMBSTONE_PREFIX}{}.{}{TOMBSTONE_SUFFIX}", + key.checkpoint_id, key.nonce + ) +} + +fn prune_owner_name(key: &PruneScratchKey) -> String { + format!( + "{PRUNE_TOMBSTONE_PREFIX}{}.{}{PRUNE_OWNER_SUFFIX}", + key.checkpoint_id, key.nonce + ) +} + +fn parse_prune_scratch_key(body: &str, name: &str) -> Result { + let (checkpoint_id, nonce) = body + .rsplit_once('.') + .ok_or_else(|| invariant(format!("invalid checkpoint prune scratch name {name:?}")))?; + validate_checkpoint_id(checkpoint_id)?; + let nonce = parse_uuid_component(nonce, "checkpoint prune tombstone")?; + Ok(PruneScratchKey { + checkpoint_id: checkpoint_id.to_string(), + nonce, + }) +} + +fn prune_identity_component(value: T, label: &str, path: &Path) -> Result +where + T: TryInto, + >::Error: std::fmt::Display, +{ + value.try_into().map_err(|error| { + invariant(format!( + "checkpoint prune candidate {} has non-representable {label}: {error}", + path.display() + )) + }) +} + fn io_error( operation: &'static str, path: impl AsRef, @@ -1923,6 +2379,164 @@ mod tests { ); } + #[test] + fn prune_preserves_head_and_explicit_lineages() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = store(&temp); + let sandbox_id = Uuid::new_v4(); + let root = publish(&store, sandbox_id, None, true); + let head = publish(&store, sandbox_id, Some(root.clone()), true); + let fork = publish(&store, sandbox_id, Some(root.clone()), false); + let unreachable = publish(&store, sandbox_id, Some(root.clone()), false); + + let removed = match store + .prune_preserving(sandbox_id, &[fork.clone()]) + .expect("prune with protected fork") + { + PruneOutcome::Complete { removed } => removed, + other => panic!("expected Complete, got {other:?}"), + }; + assert_eq!(removed, vec![unreachable.clone()]); + + let sandbox = store.configured_root().join(sandbox_id.to_string()); + assert!(sandbox.join(&root).is_dir(), "HEAD lineage must survive"); + assert!(sandbox.join(&head).is_dir(), "HEAD checkpoint must survive"); + assert!(sandbox.join(&fork).is_dir(), "protected fork must survive"); + assert!( + !sandbox.join(&unreachable).exists(), + "unreachable checkpoint must be removed" + ); + + let remaining: HashSet = store + .list(sandbox_id) + .expect("list after prune") + .into_iter() + .map(|info| info.id) + .collect(); + assert!(remaining.contains(&root)); + assert!(remaining.contains(&head)); + assert!(remaining.contains(&fork)); + assert!(!remaining.contains(&unreachable)); + } + + #[test] + fn prune_rejects_a_protected_id_that_is_not_committed() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = store(&temp); + let sandbox_id = Uuid::new_v4(); + publish(&store, sandbox_id, None, true); + + let missing = format!("ckpt-{}", Uuid::new_v4()); + let error = store + .prune_preserving(sandbox_id, &[missing.clone()]) + .expect_err("prune must reject an unknown protected id"); + assert!( + error.to_string().contains("not committed"), + "error must mention the missing id: {error}" + ); + } + + #[test] + fn prune_leaves_a_missing_sandbox_directory_untouched() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = store(&temp); + let sandbox_id = Uuid::new_v4(); + + let outcome = store + .prune_preserving(sandbox_id, &[]) + .expect("prune on a missing sandbox is a no-op"); + match outcome { + PruneOutcome::Complete { removed } => assert!(removed.is_empty()), + other => panic!("expected Complete, got {other:?}"), + } + } + + #[test] + fn prune_removes_owner_and_tombstone_after_completion() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = store(&temp); + let sandbox_id = Uuid::new_v4(); + let root = publish(&store, sandbox_id, None, true); + let unreachable = publish(&store, sandbox_id, Some(root.clone()), false); + + store + .prune_preserving(sandbox_id, &[]) + .expect("prune the unreachable branch"); + + let sandbox = store.configured_root().join(sandbox_id.to_string()); + for entry in fs::read_dir(&sandbox).expect("scan sandbox dir") { + let entry = entry.expect("entry"); + let name = entry.file_name().to_string_lossy().to_string(); + assert!( + !name.starts_with(PRUNE_TOMBSTONE_PREFIX), + "prune scratch must not remain: {name}" + ); + } + assert!( + !sandbox.join(&unreachable).exists(), + "the pruned checkpoint must be gone" + ); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn prune_reports_partial_cleanup_when_post_tombstone_work_fails() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = store(&temp); + let sandbox_id = Uuid::new_v4(); + let root = publish(&store, sandbox_id, None, true); + let unreachable = publish(&store, sandbox_id, Some(root.clone()), false); + + // Fail only after the tombstone rename has already removed the + // candidate from the committed catalog. + let hook = crate::failpoint::TestFailpoint::new(&["checkpoint-prune-after-tombstone"]); + let outcome = hook + .run(async { store.prune_preserving(sandbox_id, &[]) }) + .await + .expect("post-tombstone failure must be reported as an outcome, not an error"); + + match outcome { + PruneOutcome::PartialCleanup { removed, source } => { + assert_eq!( + removed, + vec![unreachable.clone()], + "the candidate was already removed from the committed catalog" + ); + assert!( + !source.to_string().is_empty(), + "the stopping failure must be reported" + ); + } + other => panic!("expected PartialCleanup, got {other:?}"), + } + + let sandbox = store.configured_root().join(sandbox_id.to_string()); + assert!( + !sandbox.join(&unreachable).exists(), + "the candidate must no longer be committed" + ); + let scratch_remains = fs::read_dir(&sandbox) + .expect("scan sandbox dir") + .filter_map(|entry| entry.ok()) + .any(|entry| { + entry + .file_name() + .to_string_lossy() + .starts_with(PRUNE_TOMBSTONE_PREFIX) + }); + assert!( + scratch_remains, + "an interrupted prune must retain scratch for later cleanup" + ); + assert!( + store + .read_head(sandbox_id) + .expect("HEAD survives") + .is_some(), + "the HEAD chain must remain intact" + ); + } + #[test] fn publish_metadata_only_lineage_validation_rejects_a_missing_parent() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/src/blaze/crates/blazed/src/sandbox/checkpoint.rs b/src/blaze/crates/blazed/src/sandbox/checkpoint.rs index f561e5c7f4..7f330bd868 100644 --- a/src/blaze/crates/blazed/src/sandbox/checkpoint.rs +++ b/src/blaze/crates/blazed/src/sandbox/checkpoint.rs @@ -523,6 +523,92 @@ impl SandboxManager { })? } + /// Remove checkpoints that are neither reachable from HEAD nor listed as + /// protected. Requires the sandbox to have no active operation. + pub async fn prune_checkpoints( + self: &Arc, + id: Uuid, + protected: Vec, + ) -> Result> { + for candidate in &protected { + blaze_core::checkpoint::validate_checkpoint_id(candidate).map_err(|error| { + BlazeDaemonError::BadRequest(format!( + "invalid protected checkpoint identifier: {error}" + )) + })?; + } + let operation = self.operation_lock(id).lock_owned().await; + let instance = self.get(id)?; + if let Some(journal) = &instance.operation { + return Err(BlazeDaemonError::Conflict(format!( + "instance {id} has unfinished {} operation", + journal.kind + ))); + } + // Prune keeps no operation journal, so a sandbox left recovery-required + // by an interrupted prune would otherwise pass the journal guard above. + // Reject the retry instead: `load_catalog` ignores retained `.prune.*` + // scratch, so a retry could report success while the orphaned scratch + // still needs destroy or startup reconciliation. + if instance.state == SandboxState::RecoveryRequired { + return Err(BlazeDaemonError::Conflict(format!( + "instance {id} is recovery-required; destroy or restart must \ + reconcile retained checkpoint scratch before pruning again" + ))); + } + let manager = Arc::clone(self); + crate::failpoint::spawn_blocking(move || { + let _operation = operation; + crate::failpoint::pause_blocking("checkpoint-before-store-prune"); + // Verify every protected identifier is currently committed before + // deleting anything. A canonical but unknown identifier is a + // client mistake, not a catalog fault, and must not become 500. + if !protected.is_empty() { + let committed: std::collections::HashSet = manager + .checkpoints + .list(id) + .map_err(checkpoint_store_error)? + .into_iter() + .map(|info| info.id) + .collect(); + for candidate in &protected { + if !committed.contains(candidate) { + return Err(BlazeDaemonError::BadRequest(format!( + "protected checkpoint {candidate} is not committed" + ))); + } + } + } + match manager + .checkpoints + .prune_preserving(id, &protected) + .map_err(checkpoint_store_error)? + { + crate::checkpoint_store::PruneOutcome::Complete { removed } => Ok(removed), + crate::checkpoint_store::PruneOutcome::PartialCleanup { removed, source } => { + // The candidate is already absent from the committed + // catalog, but retained `.prune.*` scratch requires destroy + // or startup reconciliation to finish cleanup. Persist the + // recovery-required state so retries do not silently skip + // the orphaned scratch. + let recovery = manager.mark_recovery(id).err(); + Err(BlazeDaemonError::RecoveryRequired(format!( + "checkpoint prune left retained scratch after removing {removed:?}: {source}{}", + recovery + .map(|error| format!( + "; recovery state persistence failed: {error}" + )) + .unwrap_or_default() + ))) + } + } + }) + .await + .map_err(|error| { + BlazeDaemonError::Internal(format!("checkpoint prune blocking task: {error}")) + })? + } + async fn finish_failed_unpublished_checkpoint( &self, id: Uuid,