diff --git a/docs/user-guide/en/runtime/blaze.md b/docs/user-guide/en/runtime/blaze.md index 8cb98f6820..913f783b77 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`. Restore uses +`POST /v1/sandboxes/{id}/rollback/{checkpoint_id}`. ## Host Integration Boundary @@ -214,7 +215,7 @@ The lifecycle invariants behind these compatibility responses are recorded in the [lifecycle state consistency and compatibility design](../../../../src/blaze/docs/design/lifecycle-state-consistency.md). -## Checkpoint Capture and History +## Checkpoint Capture, History, and Restore Blaze captures a running sandbox through `POST /v1/sandboxes/{id}/checkpoint`. @@ -283,8 +284,38 @@ 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`. + +Restore a running sandbox with: + +```http +POST /v1/sandboxes/{id}/rollback/{checkpoint_id} +``` + +Restore requires a verified full checkpoint, an exact match for the sandbox's +policy, image, backend, and backend version, plus explicit restore support from +both the backend adapter and storage provider. The built-in mock adapter and +file provider implement this contract. Other backend adapters return HTTP 501 +before stopping the current runtime until they implement restore. + +A `checkpoint_id` that is not in canonical form is rejected with HTTP 400, and a +canonical identifier that names no committed checkpoint is reported as HTTP 404. +Both answers are final: neither changes the running sandbox, so retrying the +same selection cannot succeed. + +The file provider stages the selected root filesystem while the current +backend remains running. Blaze then stops the old backend, activates the staged +root, starts and checks the replacement owner, moves checkpoint HEAD, and +commits storage. The dividing line is whether Blaze has begun stopping the old +backend: a failure before that point, while still validating and staging the +replacement root, preserves the running sandbox untouched. Once Blaze starts +stopping the old backend, any later failure — including the stop itself failing +or Blaze being unable to confirm the old backend actually stopped — retains the +resources that actually exist and marks the sandbox `RecoveryRequired` so +destruction can finish cleanup. Restore moves checkpoint HEAD but does not +rewrite `last_checkpoint` or capture history. + +Checkpoint deletion and pruning are not provided by this API. ## Storage Artifact Synchronization diff --git a/docs/user-guide/zh/runtime/blaze.md b/docs/user-guide/zh/runtime/blaze.md index 4001c8cfd1..cdf137682b 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}/rollback/{checkpoint_id}`。 ## 主机集成边界 @@ -181,7 +182,7 @@ Blaze 仍可读取旧版本写入的 `Reset`、`Warm` 和 `start_path = "warm"` [生命周期状态一致性与兼容性设计](../../../../src/blaze/docs/design/lifecycle-state-consistency_zh.md) 中。 -## 检查点捕获与历史 +## 检查点捕获、历史与恢复 Blaze 通过 `POST /v1/sandboxes/{id}/checkpoint` 捕获运行中的 sandbox。 @@ -240,8 +241,32 @@ sandbox 或修改其生命周期记录前返回 HTTP 501。 能够确认发生在发布前的失败会删除临时数据、恢复后端，并让 sandbox 保持运行。 如果 Blaze 无法确认发布、HEAD 更新、持久化或后端恢复的结果，则会保留持久记录并 报告 `RecoveryRequired`；在 sandbox 完成恢复处理或销毁前，不应重试捕获。已经提交但 -未成为 HEAD 的检查点仍可能出现在历史列表中，其 `is_head` 为 `false`。当前版本 -不提供检查点恢复、删除或清理接口。 +未成为 HEAD 的检查点仍可能出现在历史列表中，其 `is_head` 为 `false`。 + +可以使用以下接口恢复正在运行的 sandbox： + +```http +POST /v1/sandboxes/{id}/rollback/{checkpoint_id} +``` + +恢复要求目标是经过校验的完整检查点，且策略、镜像、后端和后端版本都与当前 +sandbox 完全一致；后端适配器和存储提供程序还必须明确声明支持恢复。内置 mock +适配器与文件存储提供程序实现了这项合同。其他后端适配器在实现恢复前会返回 +HTTP 501，并且不会停止当前运行环境。 + +`checkpoint_id` 不符合规范形式时返回 HTTP 400；符合规范但没有对应已提交检查点 +时返回 HTTP 404。这两种结果都是终态：都不会改动正在运行的 sandbox，用同一个 +标识符重试也不可能成功。 + +文件存储提供程序会在当前后端仍运行时准备目标根文件系统。随后 Blaze 停止旧后端、 +启用暂存根文件系统、启动并检查替代后端、移动检查点 HEAD，最后提交存储变更。 +判断边界是 Blaze 是否已经开始停止旧后端：在此之前失败（仍处于校验和准备根文件 +系统的阶段）时，旧后端照常运行，sandbox 不受影响；一旦开始停止旧后端，此后 +任何失败——包括停止操作本身失败或无法确认旧后端是否真正停止——都会让 Blaze +保留实际存在的资源并把 sandbox 标记为 `RecoveryRequired`，以便销毁操作完成 +清理。恢复会移动检查点 HEAD，但不会改写 `last_checkpoint` 或捕获历史。 + +该接口不提供检查点删除或清理能力。 ## 存储制品同步 diff --git a/src/blaze/AGENTS.md b/src/blaze/AGENTS.md index d2c9b1f92d..17b7d19ce7 100644 --- a/src/blaze/AGENTS.md +++ b/src/blaze/AGENTS.md @@ -29,8 +29,9 @@ Platform: Linux (x86_64 + aarch64) for production. macOS builds succeed but spaw - **Daemon-only API model**: No CLI client for sandbox operations. All instance and template management is done via HTTP endpoints on UDS (`/run/blaze/api.sock`) or TCP (`:14159`). The CLI subcommands (`daemon start`, `daemon reload`, `daemon doctor`) only manage daemon lifecycle. - **BackendSpawner trait**: All backend-specific process management is behind `BackendSpawner`. Adding a new backend means implementing `spawn()`, `wait()`, `kill()`, `probe()` and registering it in `daemon::build_spawner()`. - **Policy-driven backend selection**: Workload class → policy file → prioritized backend list. The daemon probes backends at startup and selects the first available. Never hardcode backend preference in application logic. -- **Lifecycle state machine**: 9 states. The main branches are Pending → - Creating → Running and Running ↔ Paused → Checkpointed. Any non-terminal +- **Lifecycle state machine**: 10 states. The main branches are Pending → + Creating → Running, Running ↔ Paused → Checkpointed, and + Running → Restoring → Running for checkpoint restore. Any non-terminal state can enter Destroyed; incomplete cleanup enters RecoveryRequired. State transitions are enforced by `blaze_core::lifecycle`. Do not bypass via direct field mutation. diff --git a/src/blaze/README.md b/src/blaze/README.md index 8dd556beb4..dfaba26964 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}/rollback/{checkpoint_id}` | Replace a running sandbox from a verified checkpoint | | GET | `/v1/pools` | Reserved; returns `501` | | GET | `/v1/pools/{backend}/{class}` | Reserved; returns `501` | | POST | `/v1/pools/{backend}/{class}/drain` | Reserved; returns `501` | @@ -258,9 +259,30 @@ 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. -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. + +`POST /v1/sandboxes/{id}/rollback/{checkpoint_id}` is available only when the +current storage provider and checkpoint backend advertise compatible restore +capabilities. The daemon verifies the selected checkpoint, its parent chain, +runtime identity, and all artifact hashes before changing runtime state. + +The file provider stages a separate rootfs copy while the current backend is +still running. After the old backend stops, the daemon selects that copy, +starts and owns the replacement backend, moves HEAD to the selected checkpoint, +and only then releases the previous rootfs. The dividing line is whether the +daemon has begun stopping the old backend: a failure before that point, while +still validating and staging the replacement rootfs, leaves the original +runtime running untouched, as if the restore never happened. Once the daemon +starts stopping the old backend, any later failure — including the stop itself +failing or the daemon being unable to confirm the old backend actually +stopped — retains the resources that actually exist and marks the sandbox +`RecoveryRequired`, so a later destroy can finish cleanup without losing +process ownership. + +`last_checkpoint` continues to mean the most recent completed capture. Restore +moves catalog HEAD but does not rewrite capture history. + +See the [checkpoint capture and restore user guide](../../docs/user-guide/en/runtime/blaze.md#checkpoint-capture-history-and-restore) +for response fields, supported capability combinations, and failure handling. ### Guest operations diff --git a/src/blaze/README_zh.md b/src/blaze/README_zh.md index fe7fa67ef6..9cc4959a57 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}/rollback/{checkpoint_id}` | 使用经过校验的检查点替换正在运行的 sandbox | | GET | `/v1/pools` | 预留接口；返回 `501` | | GET | `/v1/pools/{backend}/{class}` | 预留接口；返回 `501` | | POST | `/v1/pools/{backend}/{class}/drain` | 预留接口；返回 `501` | @@ -218,7 +219,7 @@ daemon 才会逐个处理未结束的 sandbox。后续逐项恢复期间，如 销毁捕获中断的 sandbox，而不是从其检查点恢复。恢复失败后目前没有后台循环自动 重试。重置接口仍不可用，也不会恢复检查点。 -### 检查点捕获与历史 +### 检查点捕获、历史与恢复 当运行中的 sandbox 所使用的后端和存储提供程序都声明支持完整捕获时， `POST /v1/sandboxes/{id}/checkpoint` 会创建检查点。请求成功时，Blaze 会暂停后端， @@ -228,9 +229,35 @@ daemon 才会逐个处理未结束的 sandbox。后续逐项恢复期间，如 改变 sandbox 状态前返回 HTTP 501。 `GET /v1/sandboxes/{id}/checkpoints` 返回已提交检查点的历史摘要，包括父检查点、 -逻辑大小、是否为当前 HEAD，以及能否从 HEAD 到达。当前版本不提供检查点恢复或 -删除接口。响应字段、当前后端支持情况和失败处理方式参见 -[检查点捕获用户指南](../../docs/user-guide/zh/runtime/blaze.md#检查点捕获与历史)。 +逻辑大小、是否为当前 HEAD，以及能否从 HEAD 到达。 + +`POST /v1/sandboxes/{id}/rollback/{checkpoint_id}` 用于把一个正在运行的 +sandbox 回退到它此前捕获的某个检查点：丢弃当前的运行状态，改用该检查点保存 +的那一份状态重新运行。只有当前使用的存储提供程序，以及捕获该检查点的后端， +都支持恢复能力时，这个接口才可用；否则 Blaze 不改动 sandbox 的任何状态， +直接返回 HTTP 501。 + +在真正改动运行状态之前，Blaze 会先做一整轮校验：确认所选检查点存在、它一直 +回溯到最初检查点的整条父链完整、检查点记录的运行环境标识与当前一致，并逐个 +核对所有制品文件的哈希。任意一项不通过都会中止，sandbox 保持原样。 + +恢复过程刻意遵循“先备好新状态、再切换、最后清理旧状态”的顺序，以免中途失败 +损坏 sandbox。具体来说，旧后端还在运行时，Blaze 会先在旁边准备好一份独立的 +根文件系统；只有等旧后端完全停止，才改用这份新的根文件系统启动并接管新的 +后端，把检查点历史的当前指针（HEAD）指向所选检查点，最后才释放旧的根文件 +系统。这里的分界点是 Blaze 是否已经开始停止旧后端：如果失败发生在这之前， +也就是仍处于校验和准备新根文件系统的阶段，旧后端一直照常运行，原来的运行 +实例不受影响，相当于这次恢复没有发生。一旦 Blaze 开始停止旧后端，此后的 +任何失败——包括停止操作本身失败、无法确认旧后端是否真的已经停止——都可能 +留下清理不彻底的资源；此时 Blaze 会保留磁盘上确实存在的那部分资源，并把 +sandbox 标记为 `RecoveryRequired`（需要恢复）状态，这样之后调用销毁接口时 +仍能找到并清理这些残留资源。 + +`last_checkpoint` 字段始终指向最近一次成功捕获的检查点。回退只移动检查点 +历史的当前指针，不会改写或删除已经捕获的历史记录。 + +响应字段、受支持的能力组合和失败处理方式参见 +[检查点捕获与恢复用户指南](../../docs/user-guide/zh/runtime/blaze.md#检查点捕获历史与恢复)。 ### Guest 操作 diff --git a/src/blaze/crates/blaze-core/src/backend.rs b/src/blaze/crates/blaze-core/src/backend.rs index 98f0574366..67313ef938 100644 --- a/src/blaze/crates/blaze-core/src/backend.rs +++ b/src/blaze/crates/blaze-core/src/backend.rs @@ -99,6 +99,40 @@ pub struct SpawnRequest { pub vm: Option, } +/// Backend identity and snapshot semantics accepted by a restore adapter. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RestoreCapability { + /// Concrete backend implementation that can consume the checkpoint. + pub backend: BackendKind, + /// Exact backend version required by versioned snapshot formats. + pub version: Option, + /// Snapshot flavor accepted by the adapter. + pub snapshot_kind: SnapshotKind, +} + +/// Complete input for restoring an owned backend instance. +#[derive(Debug, Clone)] +pub struct RestoreRequest { + /// Stable sandbox identifier. + pub instance_id: Uuid, + /// Backend executable selected from the current daemon configuration. + pub binary_path: PathBuf, + /// Storage resources reconstructed for this sandbox. + pub storage: StorageSlot, + /// VM-state artifact from a committed checkpoint. + pub snapshot_path: PathBuf, + /// Guest-memory artifact from the same checkpoint. + pub mem_path: PathBuf, + /// Backend identity frozen into the checkpoint metadata. + pub checkpoint_backend: BackendKind, + /// Backend version frozen into the checkpoint metadata. + pub expected_version: Option, + /// Snapshot flavor frozen into the checkpoint metadata. + pub snapshot_kind: SnapshotKind, + /// Whether the captured runtime exposed the stable run-directory guest transport. + pub expose_guest_socket: bool, +} + /// Snapshot flavor requested from a backend. /// /// The file provider currently requires self-contained artifacts, so only diff --git a/src/blaze/crates/blaze-core/src/lifecycle.rs b/src/blaze/crates/blaze-core/src/lifecycle.rs index eb21118836..472fab4d07 100644 --- a/src/blaze/crates/blaze-core/src/lifecycle.rs +++ b/src/blaze/crates/blaze-core/src/lifecycle.rs @@ -22,6 +22,8 @@ pub enum SandboxState { Running, Paused, Checkpointed, + /// The previous backend is stopped while replacement resources are owned. + Restoring, RecoveryRequired, Reset, Warm, @@ -36,6 +38,7 @@ impl SandboxState { SandboxState::Running => "running", SandboxState::Paused => "paused", SandboxState::Checkpointed => "checkpointed", + SandboxState::Restoring => "restoring", SandboxState::RecoveryRequired => "recovery-required", SandboxState::Reset => "reset", SandboxState::Warm => "warm", @@ -52,6 +55,8 @@ pub enum OperationKind { Create, /// A point-in-time checkpoint is being captured and published. Checkpoint, + /// A running sandbox is being replaced from a selected checkpoint. + Restore, /// Runtime resources are being destroyed. Destroy, } @@ -61,6 +66,7 @@ impl OperationKind { match self { OperationKind::Create => "create", OperationKind::Checkpoint => "checkpoint", + OperationKind::Restore => "restore", OperationKind::Destroy => "destroy", } } @@ -75,8 +81,8 @@ impl std::fmt::Display for OperationKind { /// Durable boundary reached by a multi-step lifecycle operation. /// /// The journal keeps this separate from [`SandboxState`]: state describes -/// externally visible runtime availability, while the phase identifies which -/// checkpoint resources may already have been published after interruption. +/// externally visible runtime availability, while the phase identifies the +/// last resource-ownership or catalog boundary committed before interruption. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum OperationPhase { @@ -88,6 +94,20 @@ pub enum OperationPhase { CheckpointPublished, /// HEAD references the checkpoint; runtime resume is not yet committed. CheckpointHeadUpdated, + /// Restore intent is durable, but the current runtime is still owned. + RestorePreparing, + /// Replacement storage is staged without changing the live rootfs. + RestoreStorageStaged, + /// The current backend has been confirmed stopped. + RestoreBackendStopped, + /// Staged storage is active while the predecessor remains recoverable. + RestoreStorageActivated, + /// A replacement backend has started and is owned by the runtime. + RestoreBackendStarted, + /// HEAD references the restored checkpoint; storage and lifecycle commits remain. + RestoreHeadUpdated, + /// The storage replacement is committed and can no longer be aborted. + RestoreStorageCommitted, } impl OperationPhase { @@ -97,6 +117,29 @@ impl OperationPhase { OperationPhase::CheckpointPaused => "checkpoint-paused", OperationPhase::CheckpointPublished => "checkpoint-published", OperationPhase::CheckpointHeadUpdated => "checkpoint-head-updated", + OperationPhase::RestorePreparing => "restore-preparing", + OperationPhase::RestoreStorageStaged => "restore-storage-staged", + OperationPhase::RestoreBackendStopped => "restore-backend-stopped", + OperationPhase::RestoreStorageActivated => "restore-storage-activated", + OperationPhase::RestoreBackendStarted => "restore-backend-started", + OperationPhase::RestoreHeadUpdated => "restore-head-updated", + OperationPhase::RestoreStorageCommitted => "restore-storage-committed", + } + } + + const fn operation_kind(self) -> OperationKind { + match self { + OperationPhase::CheckpointPreparing + | OperationPhase::CheckpointPaused + | OperationPhase::CheckpointPublished + | OperationPhase::CheckpointHeadUpdated => OperationKind::Checkpoint, + OperationPhase::RestorePreparing + | OperationPhase::RestoreStorageStaged + | OperationPhase::RestoreBackendStopped + | OperationPhase::RestoreStorageActivated + | OperationPhase::RestoreBackendStarted + | OperationPhase::RestoreHeadUpdated + | OperationPhase::RestoreStorageCommitted => OperationKind::Restore, } } @@ -106,6 +149,13 @@ impl OperationPhase { OperationPhase::CheckpointPaused => 1, OperationPhase::CheckpointPublished => 2, OperationPhase::CheckpointHeadUpdated => 3, + OperationPhase::RestorePreparing => 0, + OperationPhase::RestoreStorageStaged => 1, + OperationPhase::RestoreBackendStopped => 2, + OperationPhase::RestoreStorageActivated => 3, + OperationPhase::RestoreBackendStarted => 4, + OperationPhase::RestoreHeadUpdated => 5, + OperationPhase::RestoreStorageCommitted => 6, } } } @@ -238,21 +288,59 @@ impl SandboxInstance { /// Advance the active checkpoint journal without replacing its identity. pub fn advance_checkpoint_phase(&mut self, phase: OperationPhase) -> Result<()> { + self.advance_operation_phase(OperationKind::Checkpoint, phase) + } + + /// Record restore intent without changing the last completed checkpoint. + pub fn begin_restore_operation(&mut self, checkpoint_id: String) -> Result<()> { + if let Some(active) = &self.operation { + return Err(BlazeError::OperationInProgress { + active: active.kind.to_string(), + requested: OperationKind::Restore.to_string(), + }); + } + let now = Utc::now(); + self.operation = Some(OperationJournal { + kind: OperationKind::Restore, + started_at: now, + checkpoint_id: Some(checkpoint_id), + phase: Some(OperationPhase::RestorePreparing), + }); + self.updated_at = now; + Ok(()) + } + + /// Advance the active restore journal without replacing its identity. + pub fn advance_restore_phase(&mut self, phase: OperationPhase) -> Result<()> { + self.advance_operation_phase(OperationKind::Restore, phase) + } + + fn advance_operation_phase( + &mut self, + requested_kind: OperationKind, + phase: OperationPhase, + ) -> Result<()> { + if phase.operation_kind() != requested_kind { + return Err(BlazeError::InvalidStateTransition { + from: requested_kind.to_string(), + to: phase.as_str().to_string(), + }); + } let operation = self .operation .as_mut() .ok_or_else(|| BlazeError::OperationInProgress { active: "none".to_string(), - requested: OperationKind::Checkpoint.to_string(), + requested: requested_kind.to_string(), })?; - if operation.kind != OperationKind::Checkpoint { + if operation.kind != requested_kind { return Err(BlazeError::OperationInProgress { active: operation.kind.to_string(), - requested: OperationKind::Checkpoint.to_string(), + requested: requested_kind.to_string(), }); } if let Some(current) = operation.phase - && phase.rank() < current.rank() + && (current.operation_kind() != requested_kind || phase.rank() < current.rank()) { return Err(BlazeError::InvalidStateTransition { from: current.as_str().to_string(), @@ -270,11 +358,24 @@ impl SandboxInstance { self.updated_at = Utc::now(); } - /// Apply a state transition. Returns + /// Apply a state transition. + /// + /// Restore transitions additionally require the durable backend-stop and + /// storage-commit boundaries before changing externally visible state. + /// Returns /// [`BlazeError::InvalidStateTransition`] when the move is not part /// of the lifecycle state graph. pub fn transition(&mut self, target: SandboxState) -> Result<()> { - if !is_valid_transition(self.state, target) { + let restore_boundary_reached = match (self.state, target) { + (_, SandboxState::Restoring) => { + self.restore_phase_reached(OperationPhase::RestoreBackendStopped) + } + (SandboxState::Restoring, SandboxState::Running) => { + self.restore_phase_reached(OperationPhase::RestoreStorageCommitted) + } + _ => true, + }; + if !restore_boundary_reached || !is_valid_transition(self.state, target) { return Err(BlazeError::InvalidStateTransition { from: self.state.to_string(), to: target.to_string(), @@ -294,6 +395,17 @@ impl SandboxInstance { Ok(()) } + fn restore_phase_reached(&self, minimum: OperationPhase) -> bool { + minimum.operation_kind() == OperationKind::Restore + && self.operation.as_ref().is_some_and(|operation| { + operation.kind == OperationKind::Restore + && operation.phase.is_some_and(|phase| { + phase.operation_kind() == OperationKind::Restore + && phase.rank() >= minimum.rank() + }) + }) + } + /// Persist this instance to `{state_dir}/{id}/state.json`. Atomic /// rename via `state.json.tmp` to avoid torn reads on daemon restart. pub fn persist(&self, state_dir: &Path) -> Result<()> { @@ -324,7 +436,7 @@ impl SandboxInstance { fn is_valid_transition(from: SandboxState, to: SandboxState) -> bool { use SandboxState::{ - Checkpointed, Creating, Destroyed, Paused, Pending, RecoveryRequired, Running, + Checkpointed, Creating, Destroyed, Paused, Pending, RecoveryRequired, Restoring, Running, }; if to == Destroyed { // `* → destroyed` is always valid (terminal sink). @@ -340,6 +452,8 @@ fn is_valid_transition(from: SandboxState, to: SandboxState) -> bool { (Paused, Checkpointed) => true, (Paused, Running) => true, // resume (Checkpointed, Running) => true, + (Running, Restoring) => true, + (Restoring, Running) => true, _ => false, } } @@ -373,6 +487,49 @@ mod tests { } } + #[test] + fn restore_state_requires_owned_replacement_boundaries() { + let mut inst = fresh(); + let error = inst + .transition(SandboxState::Restoring) + .expect_err("pending sandbox cannot restore"); + assert!(matches!(error, BlazeError::InvalidStateTransition { .. })); + assert_eq!(inst.state, SandboxState::Pending); + + inst.transition(SandboxState::Creating).expect("creating"); + inst.transition(SandboxState::Running).expect("running"); + inst.begin_restore_operation("ckpt-00000000-0000-0000-0000-000000000001".to_string()) + .expect("begin restore"); + inst.advance_restore_phase(OperationPhase::RestoreStorageStaged) + .expect("stage storage"); + let error = inst + .transition(SandboxState::Restoring) + .expect_err("running remains visible until the backend is stopped"); + assert!(matches!(error, BlazeError::InvalidStateTransition { .. })); + assert_eq!(inst.state, SandboxState::Running); + + inst.advance_restore_phase(OperationPhase::RestoreBackendStopped) + .expect("stop backend"); + inst.transition(SandboxState::Restoring) + .expect("restore starts"); + inst.advance_restore_phase(OperationPhase::RestoreStorageActivated) + .expect("activate storage"); + inst.advance_restore_phase(OperationPhase::RestoreBackendStarted) + .expect("start backend"); + inst.advance_restore_phase(OperationPhase::RestoreHeadUpdated) + .expect("update head"); + let error = inst + .transition(SandboxState::Running) + .expect_err("storage commit precedes the final running state"); + assert!(matches!(error, BlazeError::InvalidStateTransition { .. })); + assert_eq!(inst.state, SandboxState::Restoring); + + inst.advance_restore_phase(OperationPhase::RestoreStorageCommitted) + .expect("commit storage"); + inst.transition(SandboxState::Running) + .expect("restore commits"); + } + #[test] fn destroy_is_always_legal_except_from_destroyed() { let mut inst = fresh(); @@ -558,4 +715,136 @@ mod tests { )); assert_eq!(instance.operation, journal); } + + #[test] + fn restore_journal_round_trips_without_overwriting_last_checkpoint() { + let tmp = tempfile::tempdir().expect("tmp"); + let mut instance = fresh(); + let completed = "ckpt-00000000-0000-0000-0000-000000000001".to_string(); + let selected = "ckpt-00000000-0000-0000-0000-000000000002".to_string(); + instance.last_checkpoint = Some(completed.clone()); + instance + .transition(SandboxState::Creating) + .expect("creating"); + instance.transition(SandboxState::Running).expect("running"); + instance + .begin_restore_operation(selected.clone()) + .expect("begin restore"); + instance + .advance_restore_phase(OperationPhase::RestoreStorageStaged) + .expect("stage storage"); + instance + .advance_restore_phase(OperationPhase::RestoreBackendStopped) + .expect("stop backend"); + instance + .transition(SandboxState::Restoring) + .expect("restoring"); + + for phase in [ + OperationPhase::RestoreStorageActivated, + OperationPhase::RestoreBackendStarted, + OperationPhase::RestoreHeadUpdated, + OperationPhase::RestoreStorageCommitted, + ] { + instance + .advance_restore_phase(phase) + .expect("advance restore"); + assert_eq!( + instance.last_checkpoint.as_deref(), + Some(completed.as_str()) + ); + } + instance.persist(tmp.path()).expect("persist"); + + let loaded = SandboxInstance::load(tmp.path(), instance.id).expect("load"); + let journal = loaded.operation.expect("restore journal"); + assert_eq!(journal.kind, OperationKind::Restore); + assert_eq!(journal.checkpoint_id.as_deref(), Some(selected.as_str())); + assert_eq!(journal.phase, Some(OperationPhase::RestoreStorageCommitted)); + assert_eq!(loaded.last_checkpoint.as_deref(), Some(completed.as_str())); + assert_eq!( + serde_json::to_value(journal.kind).expect("serialize kind"), + serde_json::json!("restore") + ); + assert_eq!( + serde_json::to_value(journal.phase).expect("serialize phase"), + serde_json::json!("restore-storage-committed") + ); + } + + #[test] + fn restore_journal_rejects_phase_regression() { + let mut instance = fresh(); + let completed = "ckpt-00000000-0000-0000-0000-000000000001".to_string(); + instance.last_checkpoint = Some(completed.clone()); + instance + .begin_restore_operation("ckpt-00000000-0000-0000-0000-000000000002".to_string()) + .expect("begin restore"); + instance + .advance_restore_phase(OperationPhase::RestoreStorageActivated) + .expect("advance restore"); + + let error = instance + .advance_restore_phase(OperationPhase::RestoreStorageStaged) + .expect_err("restore phase must remain a durable lower bound"); + + assert!(matches!(error, BlazeError::InvalidStateTransition { .. })); + assert_eq!( + instance + .operation + .as_ref() + .and_then(|journal| journal.phase), + Some(OperationPhase::RestoreStorageActivated) + ); + assert_eq!(instance.last_checkpoint, Some(completed)); + } + + #[test] + fn operation_journals_reject_phases_from_the_other_operation() { + let mut checkpoint = fresh(); + checkpoint + .begin_checkpoint_operation("ckpt-00000000-0000-0000-0000-000000000001".to_string()) + .expect("begin checkpoint"); + let checkpoint_journal = checkpoint.operation.clone(); + let checkpoint_error = checkpoint + .advance_checkpoint_phase(OperationPhase::RestoreBackendStopped) + .expect_err("checkpoint cannot record restore progress"); + assert!(matches!( + checkpoint_error, + BlazeError::InvalidStateTransition { .. } + )); + assert_eq!(checkpoint.operation, checkpoint_journal); + + let mut restore = fresh(); + restore + .begin_restore_operation("ckpt-00000000-0000-0000-0000-000000000002".to_string()) + .expect("begin restore"); + let restore_journal = restore.operation.clone(); + let restore_error = restore + .advance_restore_phase(OperationPhase::CheckpointPublished) + .expect_err("restore cannot record checkpoint progress"); + assert!(matches!( + restore_error, + BlazeError::InvalidStateTransition { .. } + )); + assert_eq!(restore.operation, restore_journal); + } + + #[test] + fn restore_journal_cannot_replace_an_active_operation() { + let mut instance = fresh(); + instance.begin_operation(OperationKind::Create); + let journal = instance.operation.clone(); + + let error = instance + .begin_restore_operation("ckpt-00000000-0000-0000-0000-000000000001".to_string()) + .expect_err("restore must not replace create"); + + assert!(matches!( + error, + BlazeError::OperationInProgress { active, requested } + if active == "create" && requested == "restore" + )); + assert_eq!(instance.operation, journal); + } } diff --git a/src/blaze/crates/blaze-core/src/storage.rs b/src/blaze/crates/blaze-core/src/storage.rs index 526b0a0b2b..3aac9b8731 100644 --- a/src/blaze/crates/blaze-core/src/storage.rs +++ b/src/blaze/crates/blaze-core/src/storage.rs @@ -32,6 +32,19 @@ pub struct StorageSlot { pub instance_dir: PathBuf, } +/// Stable handle for one provider-owned rootfs restore transaction. +/// +/// Callers must keep this handle from staging through activation and +/// finalization. Providers must validate both fields against durable state +/// before changing storage. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StorageRestoreTransaction { + /// Stable sandbox identifier whose rootfs is being replaced. + pub instance_id: String, + /// Unique transaction identifier used to reject stale handles. + pub transaction_id: uuid::Uuid, +} + /// Storage provider capacity reported by the health endpoint. #[derive(Debug, Clone, Default, serde::Serialize)] pub struct PoolStatus { @@ -152,6 +165,73 @@ pub trait StorageProvider: Send + Sync { }) } + /// Report whether this provider can restore a self-contained checkpoint. + /// + /// The default is conservative so existing providers cannot enter a + /// partially implemented replacement flow. + fn supports_checkpoint_restore(&self) -> bool { + false + } + + /// Copy a checkpoint rootfs into provider-owned staging storage. + /// + /// Staging must leave the live rootfs unchanged so callers may prepare the + /// replacement before stopping the current runtime. + async fn stage_checkpoint_restore( + &self, + slot: &StorageSlot, + source: &Path, + ) -> Result { + let _ = (slot, source); + Err(checkpoint_restore_unsupported()) + } + + /// Select the staged rootfs while retaining the previous rootfs. + /// + /// A successful activation must remain abortable until + /// [`Self::commit_checkpoint_restore`] starts. + async fn activate_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + let _ = transaction; + Err(checkpoint_restore_unsupported()) + } + + /// Finalize an activated rootfs and release its retained predecessor. + async fn commit_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + let _ = transaction; + Err(checkpoint_restore_unsupported()) + } + + /// Restore the predecessor retained by a staged or activated transaction. + async fn abort_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + let _ = transaction; + Err(checkpoint_restore_unsupported()) + } + + /// Resolve an interrupted restore transaction after process restart. + /// + /// Implementations choose the outcome from durable transaction state: + /// work not yet committed should roll back, while a durable commit intent + /// should finish committing. + async fn reconcile_checkpoint_restore(&self, instance_id: &str) -> Result<()> { + let _ = instance_id; + Err(checkpoint_restore_unsupported()) + } + /// Return the provider's current storage capacity. fn pool_status(&self) -> PoolStatus; } + +fn checkpoint_restore_unsupported() -> BlazeError { + BlazeError::StorageError { + msg: "storage provider does not support checkpoint restore".to_string(), + } +} diff --git a/src/blaze/crates/blazed/src/api.rs b/src/blaze/crates/blazed/src/api.rs index 564873d02d..26414e7370 100644 --- a/src/blaze/crates/blazed/src/api.rs +++ b/src/blaze/crates/blazed/src/api.rs @@ -26,7 +26,7 @@ use uuid::Uuid; use crate::error::{BlazeDaemonError, Result}; use crate::guest::MAX_GUEST_FILE_BYTES; -use crate::sandbox::CreateSandbox; +use crate::sandbox::{CreateSandbox, RestoreSandbox, RestoreSandboxResult}; use crate::state::ServerState; const MAX_EXEC_TIMEOUT_SECS: u32 = 20; @@ -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, "rollback", checkpoint_id]) => { + rollback(state, id, checkpoint_id).await + } ("DELETE", ["v1", "sandboxes", id]) => destroy_sandbox(state, id).await, ("GET", ["v1", "pools"]) | ("GET", ["v1", "pools", _, _]) @@ -340,6 +343,39 @@ async fn list_checkpoints(state: &Arc, id: &str) -> Result, + id: &str, + checkpoint_id: &str, +) -> Result>> { + let uuid = parse_uuid(id)?; + let instance = state.manager.get(uuid)?; + let binary_path = state + .config + .lock() + .map_err(|_| BlazeDaemonError::Internal("config lock poisoned".into()))? + .backends + .get(instance.backend.as_str()) + .cloned() + .unwrap_or_default(); + let restored: RestoreSandboxResult = state + .manager + .restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint_id.to_string(), + binary_path, + }, + ) + .await?; + json_ok(&json!({ + "instance_id": restored.instance.id, + "checkpoint_id": restored.checkpoint_id, + "restored": true, + "state": restored.instance.state, + })) +} + async fn destroy_sandbox(state: &Arc, id: &str) -> Result>> { let uuid = parse_uuid(id)?; state.manager.destroy(uuid).await?; @@ -1230,6 +1266,92 @@ mod tests { } } + struct CaptureOnlyMockSpawner; + + #[async_trait] + impl BackendSpawner for CaptureOnlyMockSpawner { + async fn spawn( + &self, + request: BackendSpawnRequest, + ) -> std::result::Result { + MockSpawner.spawn(request).await + } + + async fn probe(&self, _binary_path: &Path) -> blaze_core::Result { + Ok(true) + } + + async fn cleanup_orphan( + &self, + instance_id: Uuid, + run_dir: &OwnedRunDir, + ) -> blaze_core::Result<()> { + MockSpawner.cleanup_orphan(instance_id, run_dir).await + } + } + + /// Spawns owners that expose the guest transport but restores owners that + /// silently drop it, exercising the restore readiness contract. + struct TransportDroppingRestoreSpawner; + + #[async_trait] + impl BackendSpawner for TransportDroppingRestoreSpawner { + async fn spawn( + &self, + request: BackendSpawnRequest, + ) -> std::result::Result { + GuestMockSpawner.spawn(request).await + } + + async fn restore_capability( + &self, + _binary_path: &Path, + ) -> blaze_core::Result> { + // Match the identity the guest-mock owner freezes into the + // checkpoint so the sweep reaches the readiness contract instead of + // stopping at the version comparison. + Ok(Some(blaze_core::backend::RestoreCapability { + backend: BackendKind::Mock, + version: Some("guest-mock-v1".to_string()), + snapshot_kind: blaze_core::backend::SnapshotKind::Full, + })) + } + + async fn restore( + &self, + request: crate::spawner::BackendRestoreRequest, + ) -> crate::spawner::RestoreResult { + // Start an owner through the plain mock spawn path so the + // replacement deliberately lacks the guest transport the captured + // runtime exposed. `MockSpawner::restore` would reject the + // guest-mock version identity before reaching this point. + let spawn = BackendSpawnRequest::new( + blaze_core::backend::SpawnRequest { + instance_id: request.instance_id, + binary_path: request.binary_path.clone(), + storage: request.storage.clone(), + backend: blaze_core::policy::BackendConfigs::default(), + vm: None, + }, + request.run_dir.clone(), + ) + .map_err(SpawnFailure::clean)?; + MockSpawner.spawn(spawn).await + } + + async fn probe(&self, _binary_path: &Path) -> blaze_core::Result { + Ok(true) + } + + async fn cleanup_orphan( + &self, + instance_id: Uuid, + run_dir: &OwnedRunDir, + ) -> blaze_core::Result<()> { + GuestMockSpawner.cleanup_orphan(instance_id, run_dir).await + } + } + struct StalledGuestOwner { instance_id: Uuid, socket: PathBuf, @@ -1884,6 +2006,559 @@ mod tests { assert!(!checkpoint_namespace.exists()); } + #[tokio::test] + async fn rollback_replaces_runtime_state_without_rewriting_capture_history() { + 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 = state.storage.reconstruct(id).await.expect("storage slot"); + + tokio::fs::write(&slot.rootfs_path, b"first-rootfs") + .await + .expect("first rootfs"); + let (_, first) = dispatched_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoint"), + Vec::new(), + ) + .await; + let first_id = first["id"].as_str().expect("first checkpoint"); + + tokio::fs::write(&slot.rootfs_path, b"second-rootfs") + .await + .expect("second rootfs"); + let (_, second) = dispatched_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/checkpoint"), + Vec::new(), + ) + .await; + let second_id = second["id"].as_str().expect("second checkpoint"); + + tokio::fs::write(&slot.rootfs_path, b"third-rootfs") + .await + .expect("third rootfs"); + + let (status, restored) = dispatched_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/rollback/{first_id}"), + Vec::new(), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(restored["instance_id"], id); + assert_eq!(restored["checkpoint_id"], first_id); + assert_eq!(restored["restored"], true); + assert_eq!(restored["state"], "running"); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("restored rootfs"), + b"first-rootfs" + ); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::Running); + assert!(lifecycle.operation.is_none()); + assert_eq!(lifecycle.last_checkpoint.as_deref(), Some(second_id)); + assert_eq!( + state + .manager + .list_checkpoints(uuid) + .await + .expect("checkpoint list") + .iter() + .find(|checkpoint| checkpoint.is_head) + .map(|checkpoint| checkpoint.id.as_str()), + Some(first_id) + ); + assert!(state.manager.backend_owner(uuid).is_some()); + for name in [ + ".rootfs.restore-copying", + ".rootfs.restore-staged", + ".rootfs.restore-backup", + ".rootfs.restore-discard", + ".rootfs.restore.json", + ".rootfs.restore-journal.tmp", + ] { + assert!(!slot.instance_dir.join(name).exists(), "{name} remains"); + } + } + + #[tokio::test] + async fn rollback_rejects_an_unavailable_adapter_before_mutation() { + 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(CaptureOnlyMockSpawner)), + 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 checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let owner = state.manager.backend_owner(uuid).expect("backend owner"); + + let error = state + .manager + .restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id, + binary_path: PathBuf::new(), + }, + ) + .await + .expect_err("restore must require an adapter"); + + assert!(matches!(error, BlazeDaemonError::UnsupportedOperation(_))); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("unchanged rootfs"), + b"current-rootfs" + ); + let retained = state.manager.backend_owner(uuid).expect("retained owner"); + assert!(Arc::ptr_eq(&owner, &retained)); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::Running); + assert!(lifecycle.operation.is_none()); + } + + #[tokio::test] + async fn rollback_missing_checkpoint_returns_not_found_without_mutation() { + 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 = state.storage.reconstruct(id).await.expect("storage slot"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let owner = state.manager.backend_owner(uuid).expect("backend owner"); + + let missing = format!("ckpt-{}", Uuid::new_v4()); + let (status, body) = handled_json( + &state, + Method::POST, + &format!("/v1/sandboxes/{id}/rollback/{missing}"), + Vec::new(), + ) + .await; + + assert_eq!( + status, + StatusCode::NOT_FOUND, + "an absent checkpoint must not surface as a retriable server failure" + ); + assert_eq!(body["status"], 404); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("unchanged rootfs"), + b"current-rootfs" + ); + let retained = state.manager.backend_owner(uuid).expect("retained owner"); + assert!(Arc::ptr_eq(&owner, &retained)); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::Running); + assert!(lifecycle.operation.is_none()); + } + + #[tokio::test] + async fn rollback_rejects_a_replacement_that_drops_the_guest_transport() { + 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(TransportDroppingRestoreSpawner)), + 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"); + // The captured runtime exposes a guest socket. + assert!( + !state + .manager + .backend_owner(uuid) + .expect("backend owner") + .guest_socket_path() + .as_os_str() + .is_empty(), + "the captured runtime must expose the guest transport" + ); + write_checkpoint_fixture(&state, id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + + let error = state + .manager + .restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id, + binary_path: PathBuf::new(), + }, + ) + .await + .expect_err("a replacement without the guest transport must not publish"); + + assert!( + matches!(error, BlazeDaemonError::RecoveryRequired(_)), + "expected RecoveryRequired, got {error:?}" + ); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!( + lifecycle.state, + SandboxState::RecoveryRequired, + "the sandbox must not be published as running without its transport" + ); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn restore_stage_failure_keeps_the_current_runtime_running() { + 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"); + let slot = write_checkpoint_fixture(&state, id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let owner = state.manager.backend_owner(uuid).expect("backend owner"); + let hook = crate::failpoint::TestFailpoint::new(&["restore-storage-stage"]); + + hook.run(state.manager.restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id, + binary_path: PathBuf::new(), + }, + )) + .await + .expect_err("stage failure"); + + let retained = state.manager.backend_owner(uuid).expect("retained owner"); + assert!(Arc::ptr_eq(&owner, &retained)); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("unchanged rootfs"), + b"current-rootfs" + ); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::Running); + assert!(lifecycle.operation.is_none()); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn uncertain_backend_stop_retains_the_current_owner_and_rootfs() { + 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"); + let slot = write_checkpoint_fixture(&state, id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let owner = state.manager.backend_owner(uuid).expect("backend owner"); + let hook = crate::failpoint::TestFailpoint::new(&["restore-backend-stop"]); + + let error = hook + .run(state.manager.restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id, + binary_path: PathBuf::new(), + }, + )) + .await + .expect_err("backend stop outcome must require recovery"); + + assert!(matches!(error, BlazeDaemonError::RecoveryRequired(_))); + let retained = state.manager.backend_owner(uuid).expect("retained owner"); + assert!(Arc::ptr_eq(&owner, &retained)); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("unchanged rootfs"), + b"current-rootfs" + ); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::RecoveryRequired); + assert_eq!(lifecycle.backend_ownership, BackendOwnership::Unknown); + assert_eq!( + lifecycle + .operation + .as_ref() + .and_then(|operation| operation.phase), + Some(OperationPhase::RestoreStorageStaged) + ); + for name in [ + ".rootfs.restore-staged", + ".rootfs.restore-backup", + ".rootfs.restore.json", + ] { + assert!(!slot.instance_dir.join(name).exists(), "{name} remains"); + } + assert!(state.manager.destroy(uuid).await.expect("destroy")); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn uncertain_head_update_retains_the_replacement_owner() { + 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"); + let slot = write_checkpoint_fixture(&state, id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"later-checkpoint-rootfs") + .await + .expect("later checkpoint rootfs"); + let latest = state + .manager + .checkpoint(uuid) + .await + .expect("later checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let hook = crate::failpoint::TestFailpoint::new(&["checkpoint-store-head-after-rename"]); + + let error = hook + .run(state.manager.restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id.clone(), + binary_path: PathBuf::new(), + }, + )) + .await + .expect_err("HEAD update must be reported"); + + assert!(matches!(error, BlazeDaemonError::RecoveryRequired(_))); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("selected rootfs"), + b"checkpoint-rootfs" + ); + assert!(state.manager.backend_owner(uuid).is_some()); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::RecoveryRequired); + assert_eq!(lifecycle.backend_ownership, BackendOwnership::Running); + assert_eq!( + lifecycle + .operation + .as_ref() + .and_then(|operation| operation.phase), + Some(OperationPhase::RestoreBackendStarted) + ); + assert_eq!( + lifecycle.last_checkpoint.as_deref(), + Some(latest.id.as_str()) + ); + assert_eq!( + state + .manager + .list_checkpoints(uuid) + .await + .expect("observable checkpoint catalog") + .iter() + .find(|item| item.is_head) + .map(|item| item.id.as_str()), + Some(checkpoint.id.as_str()) + ); + + assert!(state.manager.destroy(uuid).await.expect("destroy")); + assert_eq!( + state.manager.get(uuid).expect("destroyed").state, + SandboxState::Destroyed + ); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn final_state_failure_keeps_the_committed_restore_journal() { + 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"); + let slot = write_checkpoint_fixture(&state, id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + tokio::fs::write(&slot.rootfs_path, b"current-rootfs") + .await + .expect("current rootfs"); + let hook = crate::failpoint::TestFailpoint::new(&["restore-final-state"]); + + let error = hook + .run(state.manager.restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id.clone(), + binary_path: PathBuf::new(), + }, + )) + .await + .expect_err("final state failure"); + + assert!(matches!(error, BlazeDaemonError::RecoveryRequired(_))); + assert_eq!( + tokio::fs::read(&slot.rootfs_path) + .await + .expect("committed rootfs"), + b"checkpoint-rootfs" + ); + let lifecycle = state.manager.get(uuid).expect("lifecycle"); + assert_eq!(lifecycle.state, SandboxState::RecoveryRequired); + assert_eq!(lifecycle.backend_ownership, BackendOwnership::Running); + assert_eq!( + lifecycle + .operation + .as_ref() + .map(|operation| (operation.checkpoint_id.as_deref(), operation.phase)), + Some(( + Some(checkpoint.id.as_str()), + Some(OperationPhase::RestoreStorageCommitted) + )) + ); + assert_eq!( + state + .manager + .list_checkpoints(uuid) + .await + .expect("checkpoint list") + .iter() + .find(|item| item.is_head) + .map(|item| item.id.as_str()), + Some(checkpoint.id.as_str()) + ); + assert!(state.manager.backend_owner(uuid).is_some()); + assert!(state.manager.destroy(uuid).await.expect("destroy")); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn cancelled_restore_after_head_finishes_in_detached_supervisor() { + 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").to_string(); + let uuid = Uuid::parse_str(&id).expect("uuid"); + write_checkpoint_fixture(&state, &id).await; + let checkpoint = state.manager.checkpoint(uuid).await.expect("checkpoint"); + let hook = crate::failpoint::TestFailpoint::new(&["restore-after-head"]); + let restore_state = state.clone(); + let restore_hook = hook.clone(); + let restore = tokio::spawn(async move { + restore_hook + .run(restore_state.manager.restore( + uuid, + RestoreSandbox { + checkpoint_id: checkpoint.id, + binary_path: PathBuf::new(), + }, + )) + .await + }); + hook.wait_until_paused().await; + + let persisted = SandboxInstance::load(&configured_state_dir(&state), uuid) + .expect("persisted restore journal"); + assert_eq!(persisted.state, SandboxState::Restoring); + assert_eq!( + persisted.operation.and_then(|operation| operation.phase), + Some(OperationPhase::RestoreHeadUpdated) + ); + assert_eq!(persisted.backend_ownership, BackendOwnership::Running); + assert!(state.manager.backend_owner(uuid).is_some()); + + restore.abort(); + assert!(restore.await.expect_err("cancelled restore").is_cancelled()); + let destroy_state = state.clone(); + let mut destroy = tokio::spawn(async move { destroy_state.manager.destroy(uuid).await }); + assert!( + tokio::time::timeout(Duration::from_millis(20), &mut destroy) + .await + .is_err(), + "destroy must wait for the detached restore supervisor" + ); + + hook.release(); + tokio::time::timeout(Duration::from_secs(2), &mut destroy) + .await + .expect("detached restore supervisor and queued destroy must converge") + .expect("destroy task") + .expect("destroy completed restore"); + assert_eq!( + state.manager.get(uuid).expect("destroyed").state, + SandboxState::Destroyed + ); + assert!( + !state + .config + .lock() + .expect("config") + .storage + .instances_dir + .join(id) + .exists() + ); + } + #[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..5a5e68d6f5 100644 --- a/src/blaze/crates/blazed/src/checkpoint_store.rs +++ b/src/blaze/crates/blazed/src/checkpoint_store.rs @@ -8,6 +8,8 @@ use std::collections::{HashMap, HashSet}; use std::fs::File; use std::io::{Read, Seek, SeekFrom, Write}; +#[cfg(target_os = "linux")] +use std::os::fd::AsRawFd; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -235,13 +237,37 @@ struct OwnedArtifact { } struct VerifiedCheckpoint { - #[cfg(test)] metadata: CheckpointMetadata, directory: OwnedStateDirectory, metadata_file: OwnedArtifact, artifacts: Vec, } +/// Restore target retained through the complete replacement operation. +/// +/// The catalog, sandbox, checkpoint directory, and artifact descriptors stay +/// open so path replacement cannot redirect either restore input or HEAD. +pub(crate) struct RestoreCheckpoint { + catalog: OwnedStateDirectory, + sandbox: OwnedStateDirectory, + verified: VerifiedCheckpoint, +} + +impl RestoreCheckpoint { + pub(crate) fn metadata(&self) -> &CheckpointMetadata { + &self.verified.metadata + } + + pub(crate) fn artifact_path(&self, name: &str) -> Result { + validate_artifact_name(name)?; + let index = REQUIRED_ARTIFACTS + .iter() + .position(|candidate| *candidate == name) + .ok_or_else(|| invariant(format!("checkpoint has no required artifact {name}")))?; + Ok(self.verified.artifacts[index].stable_path()) + } +} + struct LoadedCheckpointMetadata { metadata: CheckpointMetadata, directory: OwnedStateDirectory, @@ -310,6 +336,23 @@ impl VerifiedCheckpoint { } } +impl OwnedArtifact { + fn stable_path(&self) -> PathBuf { + #[cfg(target_os = "linux")] + { + PathBuf::from(format!( + "/proc/{}/fd/{}", + std::process::id(), + self.file.as_raw_fd() + )) + } + #[cfg(not(target_os = "linux"))] + { + self.path.clone() + } + } +} + #[cfg(test)] type BeforePublishRevalidation = Arc>>>; @@ -585,6 +628,44 @@ impl CheckpointStore { .metadata) } + /// Verify and retain a restore target and its complete ancestry. + pub(crate) fn verify_restore_target( + &self, + sandbox_id: Uuid, + checkpoint_id: &str, + ) -> Result { + let catalog = self.root()?; + let sandbox_name = sandbox_id.to_string(); + let sandbox = + required_child_directory(&catalog, &sandbox_name, "open checkpoint sandbox directory")?; + self.validated_chain_from(&sandbox, sandbox_id, checkpoint_id)?; + let verified = self.verified_checkpoint(&sandbox, sandbox_id, checkpoint_id)?; + require_linked_directory(&catalog, &sandbox_name, &sandbox)?; + Ok(RestoreCheckpoint { + catalog, + sandbox, + verified, + }) + } + + /// Atomically move HEAD to a restore target retained by this process. + pub(crate) fn set_head_verified(&self, target: &RestoreCheckpoint) -> SetHeadResult<()> { + let checkpoint_id = target.verified.metadata.id.clone(); + let sandbox_name = target.verified.metadata.sandbox_id.to_string(); + self.set_head_with_revalidation(&target.sandbox, &checkpoint_id, || { + let root = self.root()?; + if !same_directory(&root, &target.catalog)? { + return Err(invariant( + "restore target belongs to a different checkpoint catalog root", + )); + } + require_linked_directory(&target.catalog, &sandbox_name, &target.sandbox)?; + target + .verified + .require_linked(&target.sandbox, &checkpoint_id) + }) + } + /// List committed checkpoints and mark the lineage reachable from HEAD. pub fn list(&self, sandbox_id: Uuid) -> Result> { let catalog_root = self.root()?; @@ -745,6 +826,22 @@ impl CheckpointStore { self.read_head_from(&sandbox, sandbox_id) } + /// Return the recorded HEAD identifier without verifying its artifacts. + /// + /// Callers that only need to report which checkpoint HEAD names must use + /// this instead of [`Self::read_head`]. Hashing a complete checkpoint would + /// make the observation cost proportional to the guest image size, and an + /// unreadable artifact would replace the recorded identifier with an + /// integrity error exactly when a caller needs the identifier to describe + /// an interrupted operation. + pub fn read_head_id(&self, sandbox_id: Uuid) -> Result> { + let catalog = self.root()?; + let Some(sandbox) = optional_child_directory(&catalog, &sandbox_id.to_string())? else { + return Ok(None); + }; + self.read_head_id_from(&sandbox) + } + /// Remove every checkpoint artifact owned by one sandbox. /// /// A missing sandbox directory is already clean. Any unexpected entry or @@ -959,7 +1056,6 @@ impl CheckpointStore { } } let verified = VerifiedCheckpoint { - #[cfg(test)] metadata, directory, metadata_file, @@ -1888,6 +1984,13 @@ mod tests { store.read_head(sandbox_id).is_err(), "reading HEAD must retain full artifact verification" ); + assert_eq!( + store + .read_head_id(sandbox_id) + .expect("observing the recorded HEAD must not hash artifacts"), + Some(head), + "an unreadable artifact must not hide which checkpoint HEAD names" + ); } #[test] diff --git a/src/blaze/crates/blazed/src/file_provider.rs b/src/blaze/crates/blazed/src/file_provider.rs index 29de4cfb69..8111653b8b 100644 --- a/src/blaze/crates/blazed/src/file_provider.rs +++ b/src/blaze/crates/blazed/src/file_provider.rs @@ -16,9 +16,12 @@ use uuid::Uuid; use blaze_core::error::{BlazeError, Result}; use blaze_core::storage::{ - AcquireOpts, PoolStatus, StorageAcquireError, StorageProvider, StorageSlot, + AcquireOpts, PoolStatus, StorageAcquireError, StorageProvider, StorageRestoreTransaction, + StorageSlot, }; +mod restore; + /// A filesystem-based provider that copies base artifacts when available and /// otherwise creates sparse rootfs and memory files at configured sizes. pub struct FileStorageProvider { @@ -533,6 +536,43 @@ impl StorageProvider for FileStorageProvider { }) } + fn supports_checkpoint_restore(&self) -> bool { + true + } + + async fn stage_checkpoint_restore( + &self, + slot: &StorageSlot, + source: &Path, + ) -> Result { + restore::stage(self, slot, source).await + } + + async fn activate_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + restore::activate(self, transaction).await + } + + async fn commit_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + restore::commit(self, transaction).await + } + + async fn abort_checkpoint_restore( + &self, + transaction: &StorageRestoreTransaction, + ) -> Result<()> { + restore::abort(self, transaction).await + } + + async fn reconcile_checkpoint_restore(&self, instance_id: &str) -> Result<()> { + restore::reconcile(self, instance_id).await + } + fn pool_status(&self) -> PoolStatus { PoolStatus::default() } diff --git a/src/blaze/crates/blazed/src/file_provider/restore.rs b/src/blaze/crates/blazed/src/file_provider/restore.rs new file mode 100644 index 0000000000..f03f76b015 --- /dev/null +++ b/src/blaze/crates/blazed/src/file_provider/restore.rs @@ -0,0 +1,1649 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Recoverable rootfs replacement for the file storage provider. + +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use uuid::Uuid; + +use blaze_core::error::{BlazeError, Result}; +use blaze_core::storage::{StorageRestoreTransaction, StorageSlot}; + +use super::{FileStorageProvider, RequiredPathType}; + +const JOURNAL_VERSION: u32 = 1; +const MAX_JOURNAL_SIZE: u64 = 16 * 1024; + +#[derive(Debug)] +struct RestorePaths { + instance_id: String, + instance_dir: PathBuf, + rootfs: PathBuf, + copying: PathBuf, + staged: PathBuf, + backup: PathBuf, + discard: PathBuf, + journal: PathBuf, + journal_temporary: PathBuf, +} + +impl RestorePaths { + fn transaction_artifacts(&self) -> [&Path; 6] { + [ + &self.copying, + &self.staged, + &self.backup, + &self.discard, + &self.journal, + &self.journal_temporary, + ] + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum RestoreState { + Staged, + Activated, + Aborting, + Committing, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct RestoreJournal { + version: u32, + instance_id: String, + transaction_id: Uuid, + state: RestoreState, +} + +impl RestoreJournal { + fn transaction(&self) -> StorageRestoreTransaction { + StorageRestoreTransaction { + instance_id: self.instance_id.clone(), + transaction_id: self.transaction_id, + } + } +} + +/// Removes files that have not yet become part of a durable transaction. +struct UnpublishedFiles { + paths: Vec, +} + +impl UnpublishedFiles { + fn new() -> Self { + Self { paths: Vec::new() } + } + + fn track(&mut self, path: &Path) { + self.paths.push(path.to_path_buf()); + } + + fn untrack(&mut self, path: &Path) { + self.paths.retain(|tracked| tracked != path); + } + + fn commit(&mut self) { + self.paths.clear(); + } +} + +impl Drop for UnpublishedFiles { + fn drop(&mut self) { + for path in self.paths.drain(..) { + let _ = std::fs::remove_file(path); + } + } +} + +pub(super) async fn stage( + provider: &FileStorageProvider, + slot: &StorageSlot, + source: &Path, +) -> Result { + let paths = restore_paths(provider, &slot.id).await?; + require_plain_file(&paths.rootfs, "live rootfs").await?; + ensure_no_transaction(&paths).await?; + + let source = canonical_plain_file(source, "restore source").await?; + let rootfs = tokio::fs::canonicalize(&paths.rootfs) + .await + .map_err(|error| storage_error(format!("canonicalize live rootfs: {error}")))?; + if same_file(&source, &rootfs).await? { + return Err(storage_error( + "restore source must be independent from the live rootfs", + )); + } + + let journal = RestoreJournal { + version: JOURNAL_VERSION, + instance_id: paths.instance_id.clone(), + transaction_id: Uuid::new_v4(), + state: RestoreState::Staged, + }; + let mut unpublished = UnpublishedFiles::new(); + + copy_for_restore(&source, &paths.copying, &mut unpublished).await?; + crate::failpoint::pause("storage-restore-after-copy").await; + rename_new_plain_file(&paths.copying, &paths.staged).await?; + unpublished.untrack(&paths.copying); + unpublished.track(&paths.staged); + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-stage").await; + + publish_new_journal(&paths, &journal, &mut unpublished).await?; + unpublished.commit(); + sync_directory(&paths.instance_dir).await?; + Ok(journal.transaction()) +} + +pub(super) async fn activate( + provider: &FileStorageProvider, + transaction: &StorageRestoreTransaction, +) -> Result<()> { + let paths = restore_paths(provider, &transaction.instance_id).await?; + ensure_no_transient_files(&paths).await?; + let mut journal = require_journal(&paths).await?; + verify_transaction(&journal, transaction)?; + + match journal.state { + RestoreState::Activated => return ensure_activated_layout(&paths).await, + RestoreState::Staged => {} + RestoreState::Aborting => { + return Err(storage_error(format!( + "restore transaction {} is aborting", + transaction.transaction_id + ))); + } + RestoreState::Committing => { + return Err(storage_error(format!( + "restore transaction {} is committing", + transaction.transaction_id + ))); + } + } + + let (live, staged, backup, discard) = inspect_layout(&paths).await?; + if discard { + return Err(invalid_layout(&paths, journal.state)); + } + + if live && staged && !backup { + rename_new_plain_file(&paths.rootfs, &paths.backup).await?; + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-backup").await; + } else if !live && staged && backup { + // Resume after the predecessor was retained. + } else if live && !staged && backup { + // Resume after the staged rootfs was selected. + } else { + return Err(invalid_layout(&paths, journal.state)); + } + + let (live, staged, backup, discard) = inspect_layout(&paths).await?; + if !live && staged && backup && !discard { + if let Err(selection) = select_staged_rootfs(&paths).await { + let rollback = match crate::failpoint::storage("storage-restore-switch-rollback") { + Ok(()) => match rename_new_plain_file(&paths.backup, &paths.rootfs).await { + Ok(()) => sync_directory(&paths.instance_dir).await, + Err(error) => Err(error), + }, + Err(error) => Err(error), + }; + return match rollback { + Ok(()) => Err(storage_error(format!( + "select staged rootfs for '{}': {selection}; predecessor restored", + paths.instance_id + ))), + Err(rollback) => Err(storage_error(format!( + "select staged rootfs for '{}': {selection}; restoring predecessor failed: \ + {rollback}", + paths.instance_id + ))), + }; + } + crate::failpoint::pause("storage-restore-after-switch").await; + } + + ensure_activated_layout(&paths).await?; + journal.state = RestoreState::Activated; + replace_journal(&paths, &journal).await +} + +async fn select_staged_rootfs(paths: &RestorePaths) -> Result<()> { + crate::failpoint::storage("storage-restore-switch")?; + rename_new_plain_file(&paths.staged, &paths.rootfs).await?; + sync_directory(&paths.instance_dir).await +} + +pub(super) async fn commit( + provider: &FileStorageProvider, + transaction: &StorageRestoreTransaction, +) -> Result<()> { + let paths = restore_paths(provider, &transaction.instance_id).await?; + ensure_no_transient_files(&paths).await?; + let Some(mut journal) = read_journal(&paths).await? else { + return ensure_finalized_layout(&paths).await; + }; + verify_transaction(&journal, transaction)?; + + match journal.state { + RestoreState::Activated => { + ensure_activated_layout(&paths).await?; + journal.state = RestoreState::Committing; + replace_journal(&paths, &journal).await?; + crate::failpoint::pause("storage-restore-after-commit-intent").await; + } + RestoreState::Committing => {} + RestoreState::Staged => { + return Err(storage_error(format!( + "restore transaction {} is not activated", + transaction.transaction_id + ))); + } + RestoreState::Aborting => { + return Err(storage_error(format!( + "restore transaction {} is aborting", + transaction.transaction_id + ))); + } + } + finish_commit(&paths).await +} + +pub(super) async fn abort( + provider: &FileStorageProvider, + transaction: &StorageRestoreTransaction, +) -> Result<()> { + let paths = restore_paths(provider, &transaction.instance_id).await?; + ensure_no_transient_files(&paths).await?; + let Some(mut journal) = read_journal(&paths).await? else { + return ensure_finalized_layout(&paths).await; + }; + verify_transaction(&journal, transaction)?; + + if journal.state == RestoreState::Committing { + return Err(storage_error(format!( + "restore transaction {} has durable commit intent", + transaction.transaction_id + ))); + } + if journal.state != RestoreState::Aborting { + journal.state = RestoreState::Aborting; + replace_journal(&paths, &journal).await?; + } + finish_abort(&paths).await +} + +pub(super) async fn reconcile(provider: &FileStorageProvider, instance_id: &str) -> Result<()> { + let paths = restore_paths(provider, instance_id).await?; + remove_plain_file_if_present(&paths.copying, "restore copying file").await?; + remove_plain_file_if_present(&paths.journal_temporary, "restore journal temporary").await?; + + let Some(mut journal) = read_journal(&paths).await? else { + return reconcile_without_journal(&paths).await; + }; + if journal.instance_id != paths.instance_id { + return Err(storage_error(format!( + "restore journal instance '{}' does not match slot '{}'", + journal.instance_id, paths.instance_id + ))); + } + + match journal.state { + RestoreState::Committing => finish_commit(&paths).await, + RestoreState::Staged | RestoreState::Activated => { + journal.state = RestoreState::Aborting; + replace_journal(&paths, &journal).await?; + finish_abort(&paths).await + } + RestoreState::Aborting => finish_abort(&paths).await, + } +} + +async fn restore_paths(provider: &FileStorageProvider, instance_id: &str) -> Result { + let slot = provider.slot_for_id(instance_id)?; + let instances_dir = canonical_plain_path( + &provider.instances_dir, + RequiredPathType::Directory, + "instances directory", + ) + .await?; + let instance_dir = canonical_plain_path( + &slot.instance_dir, + RequiredPathType::Directory, + "slot directory", + ) + .await?; + if instance_dir.parent() != Some(instances_dir.as_path()) + || instance_dir.file_name() != Some(std::ffi::OsStr::new(instance_id)) + { + return Err(storage_error(format!( + "restore slot {} is not the direct '{}' child of instances directory {}", + instance_dir.display(), + instance_id, + instances_dir.display() + ))); + } + + Ok(RestorePaths { + instance_id: instance_id.to_string(), + rootfs: instance_dir.join("rootfs.ext4"), + copying: instance_dir.join(".rootfs.restore-copying"), + staged: instance_dir.join(".rootfs.restore-staged"), + backup: instance_dir.join(".rootfs.restore-backup"), + discard: instance_dir.join(".rootfs.restore-discard"), + journal: instance_dir.join(".rootfs.restore.json"), + journal_temporary: instance_dir.join(".rootfs.restore-journal.tmp"), + instance_dir, + }) +} + +async fn canonical_plain_path( + path: &Path, + required_type: RequiredPathType, + description: &str, +) -> Result { + if matches!(required_type, RequiredPathType::File) && is_retained_descriptor_path(path) { + let metadata = tokio::fs::metadata(path).await.map_err(|error| { + storage_error(format!( + "inspect retained {description} {}: {error}", + path.display() + )) + })?; + if !metadata.is_file() { + return Err(storage_error(format!( + "{description} {} is not a retained plain file", + path.display() + ))); + } + return Ok(path.to_path_buf()); + } + let metadata = tokio::fs::symlink_metadata(path).await.map_err(|error| { + storage_error(format!("inspect {description} {}: {error}", path.display())) + })?; + if !required_type.matches(&metadata) || metadata.file_type().is_symlink() { + return Err(storage_error(format!( + "{description} {} is not a plain {}", + path.display(), + required_type.description() + ))); + } + tokio::fs::canonicalize(path).await.map_err(|error| { + storage_error(format!( + "canonicalize {description} {}: {error}", + path.display() + )) + }) +} + +async fn canonical_plain_file(path: &Path, description: &str) -> Result { + canonical_plain_path(path, RequiredPathType::File, description).await +} + +async fn require_plain_file(path: &Path, description: &str) -> Result<()> { + if plain_file_exists(path, description).await? { + Ok(()) + } else { + Err(storage_error(format!( + "{description} {} does not exist", + path.display() + ))) + } +} + +async fn plain_file_exists(path: &Path, description: &str) -> Result { + match tokio::fs::symlink_metadata(path).await { + Ok(metadata) if metadata.is_file() && !metadata.file_type().is_symlink() => Ok(true), + Ok(_) => Err(storage_error(format!( + "{description} {} is not a plain file", + path.display() + ))), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(error) => Err(storage_error(format!( + "inspect {description} {}: {error}", + path.display() + ))), + } +} + +async fn entry_exists(path: &Path) -> Result { + match tokio::fs::symlink_metadata(path).await { + Ok(_) => Ok(true), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(error) => Err(storage_error(format!( + "inspect restore artifact {}: {error}", + path.display() + ))), + } +} + +async fn ensure_no_transaction(paths: &RestorePaths) -> Result<()> { + for path in paths.transaction_artifacts() { + if entry_exists(path).await? { + return Err(storage_error(format!( + "slot '{}' has unfinished restore artifact {}; reconcile it first", + paths.instance_id, + path.display() + ))); + } + } + Ok(()) +} + +async fn ensure_no_transient_files(paths: &RestorePaths) -> Result<()> { + for (path, description) in [ + (&paths.copying, "restore copying file"), + (&paths.journal_temporary, "restore journal temporary"), + ] { + if entry_exists(path).await? { + return Err(storage_error(format!( + "slot '{}' has unfinished {description}; reconcile it first", + paths.instance_id + ))); + } + } + Ok(()) +} + +async fn copy_for_restore( + source: &Path, + destination: &Path, + unpublished: &mut UnpublishedFiles, +) -> Result<()> { + let mut source_options = tokio::fs::OpenOptions::new(); + source_options.read(true); + #[cfg(unix)] + if !is_retained_descriptor_path(source) { + source_options.custom_flags(libc::O_NOFOLLOW); + } + let source_file = source_options.open(source).await.map_err(|error| { + storage_error(format!("open restore source {}: {error}", source.display())) + })?; + if !source_file + .metadata() + .await + .map_err(|error| storage_error(format!("inspect restore source: {error}")))? + .is_file() + { + return Err(storage_error(format!( + "restore source {} is not a regular file", + source.display() + ))); + } + + let mut destination_options = tokio::fs::OpenOptions::new(); + destination_options.write(true).create_new(true); + #[cfg(unix)] + destination_options.custom_flags(libc::O_NOFOLLOW); + let destination_file = destination_options + .open(destination) + .await + .map_err(|error| { + storage_error(format!( + "create restore stage {}: {error}", + destination.display() + )) + })?; + unpublished.track(destination); + + // Reuse the capture-side sparse copy so a mostly empty guest image does not + // become fully allocated here. A dense copy would materialize every hole and + // could exhaust the filesystem at the configured logical size even when the + // live rootfs and the checkpoint both fit. + let source_file = source_file.into_std().await; + let destination_file = destination_file.into_std().await; + crate::failpoint::spawn_blocking(move || { + super::copy_sparse_file(&source_file, &destination_file)?; + destination_file.sync_all() + }) + .await + .map_err(|error| storage_error(format!("restore stage copy task failed: {error}")))? + .map_err(|error| storage_error(format!("copy restore source: {error}"))) +} + +fn is_retained_descriptor_path(path: &Path) -> bool { + let parent = PathBuf::from(format!("/proc/{}/fd", std::process::id())); + path.parent() == Some(parent.as_path()) + && path + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.parse::().is_ok()) +} + +async fn same_file(left: &Path, right: &Path) -> Result { + let left = tokio::fs::metadata(left) + .await + .map_err(|error| storage_error(format!("inspect restore source: {error}")))?; + let right = tokio::fs::metadata(right) + .await + .map_err(|error| storage_error(format!("inspect live rootfs: {error}")))?; + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + + Ok(left.dev() == right.dev() && left.ino() == right.ino()) + } + #[cfg(not(unix))] + { + Ok(false) + } +} + +async fn publish_new_journal( + paths: &RestorePaths, + journal: &RestoreJournal, + unpublished: &mut UnpublishedFiles, +) -> Result<()> { + let bytes = encode_journal(journal)?; + let mut options = tokio::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.custom_flags(libc::O_NOFOLLOW); + let mut file = options + .open(&paths.journal_temporary) + .await + .map_err(|error| storage_error(format!("create restore journal: {error}")))?; + unpublished.track(&paths.journal_temporary); + file.write_all(&bytes) + .await + .map_err(|error| storage_error(format!("write restore journal: {error}")))?; + file.sync_all() + .await + .map_err(|error| storage_error(format!("sync restore journal: {error}")))?; + drop(file); + rename_new_plain_file(&paths.journal_temporary, &paths.journal).await?; + unpublished.untrack(&paths.journal_temporary); + // The journal now owns the staged rootfs, even if the directory sync fails. + unpublished.commit(); + Ok(()) +} + +async fn replace_journal(paths: &RestorePaths, journal: &RestoreJournal) -> Result<()> { + require_plain_file(&paths.journal, "restore journal").await?; + if entry_exists(&paths.journal_temporary).await? { + return Err(storage_error(format!( + "slot '{}' has an unfinished journal update; reconcile it first", + paths.instance_id + ))); + } + + let bytes = encode_journal(journal)?; + let mut cleanup = UnpublishedFiles::new(); + let mut options = tokio::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.custom_flags(libc::O_NOFOLLOW); + let mut file = options + .open(&paths.journal_temporary) + .await + .map_err(|error| storage_error(format!("create restore journal update: {error}")))?; + cleanup.track(&paths.journal_temporary); + file.write_all(&bytes) + .await + .map_err(|error| storage_error(format!("write restore journal update: {error}")))?; + file.sync_all() + .await + .map_err(|error| storage_error(format!("sync restore journal update: {error}")))?; + drop(file); + tokio::fs::rename(&paths.journal_temporary, &paths.journal) + .await + .map_err(|error| storage_error(format!("replace restore journal: {error}")))?; + cleanup.untrack(&paths.journal_temporary); + sync_directory(&paths.instance_dir).await +} + +fn encode_journal(journal: &RestoreJournal) -> Result> { + serde_json::to_vec(journal) + .map_err(|error| storage_error(format!("encode restore journal: {error}"))) +} + +async fn read_journal(paths: &RestorePaths) -> Result> { + if !plain_file_exists(&paths.journal, "restore journal").await? { + return Ok(None); + } + let mut options = tokio::fs::OpenOptions::new(); + options.read(true); + #[cfg(unix)] + options.custom_flags(libc::O_NOFOLLOW); + let file = options + .open(&paths.journal) + .await + .map_err(|error| storage_error(format!("open restore journal: {error}")))?; + let mut bytes = Vec::new(); + file.take(MAX_JOURNAL_SIZE + 1) + .read_to_end(&mut bytes) + .await + .map_err(|error| storage_error(format!("read restore journal: {error}")))?; + if bytes.len() as u64 > MAX_JOURNAL_SIZE { + return Err(storage_error("restore journal exceeds the size limit")); + } + let journal: RestoreJournal = serde_json::from_slice(&bytes) + .map_err(|error| storage_error(format!("parse restore journal: {error}")))?; + if journal.version != JOURNAL_VERSION { + return Err(storage_error(format!( + "unsupported restore journal version {}", + journal.version + ))); + } + Ok(Some(journal)) +} + +async fn require_journal(paths: &RestorePaths) -> Result { + read_journal(paths).await?.ok_or_else(|| { + storage_error(format!( + "slot '{}' has no restore transaction", + paths.instance_id + )) + }) +} + +fn verify_transaction( + journal: &RestoreJournal, + transaction: &StorageRestoreTransaction, +) -> Result<()> { + if journal.instance_id != transaction.instance_id + || journal.transaction_id != transaction.transaction_id + { + return Err(storage_error(format!( + "restore transaction {} does not own slot '{}'", + transaction.transaction_id, transaction.instance_id + ))); + } + Ok(()) +} + +async fn inspect_layout(paths: &RestorePaths) -> Result<(bool, bool, bool, bool)> { + Ok(( + plain_file_exists(&paths.rootfs, "live rootfs").await?, + plain_file_exists(&paths.staged, "staged rootfs").await?, + plain_file_exists(&paths.backup, "retained rootfs").await?, + plain_file_exists(&paths.discard, "discarded rootfs").await?, + )) +} + +async fn ensure_activated_layout(paths: &RestorePaths) -> Result<()> { + if inspect_layout(paths).await? != (true, false, true, false) { + return Err(invalid_layout(paths, RestoreState::Activated)); + } + Ok(()) +} + +async fn ensure_finalized_layout(paths: &RestorePaths) -> Result<()> { + require_plain_file(&paths.rootfs, "live rootfs").await?; + for path in paths.transaction_artifacts() { + if entry_exists(path).await? { + return Err(storage_error(format!( + "slot '{}' has restore artifact {}; reconcile it first", + paths.instance_id, + path.display() + ))); + } + } + Ok(()) +} + +async fn finish_abort(paths: &RestorePaths) -> Result<()> { + for _ in 0..6 { + let layout = inspect_layout(paths).await?; + match layout { + (false, true, true, false) => { + rename_new_plain_file(&paths.backup, &paths.rootfs).await?; + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-rollback-rootfs").await; + } + (true, true, false, false) => { + remove_plain_file(&paths.staged, "staged rootfs").await?; + sync_directory(&paths.instance_dir).await?; + } + (true, false, true, false) => { + rename_new_plain_file(&paths.rootfs, &paths.discard).await?; + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-discard").await; + } + (false, false, true, true) => { + rename_new_plain_file(&paths.backup, &paths.rootfs).await?; + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-rollback-rootfs").await; + } + (true, false, false, true) => { + remove_plain_file(&paths.discard, "discarded rootfs").await?; + sync_directory(&paths.instance_dir).await?; + } + (true, false, false, false) => { + remove_plain_file(&paths.journal, "restore journal").await?; + sync_directory(&paths.instance_dir).await?; + return Ok(()); + } + _ => return Err(invalid_layout(paths, RestoreState::Aborting)), + } + } + Err(storage_error(format!( + "restore abort for '{}' did not converge", + paths.instance_id + ))) +} + +async fn finish_commit(paths: &RestorePaths) -> Result<()> { + let (live, staged, _backup, discard) = inspect_layout(paths).await?; + if !live || staged || discard { + return Err(invalid_layout(paths, RestoreState::Committing)); + } + remove_plain_file_if_present(&paths.backup, "retained rootfs").await?; + sync_directory(&paths.instance_dir).await?; + crate::failpoint::pause("storage-restore-after-backup-release").await; + remove_plain_file(&paths.journal, "restore journal").await?; + sync_directory(&paths.instance_dir).await +} + +async fn reconcile_without_journal(paths: &RestorePaths) -> Result<()> { + let (live, staged, backup, discard) = inspect_layout(paths).await?; + if backup || discard || !live { + return Err(storage_error(format!( + "slot '{}' has ambiguous restore artifacts without a journal", + paths.instance_id + ))); + } + if staged { + remove_plain_file(&paths.staged, "unpublished staged rootfs").await?; + sync_directory(&paths.instance_dir).await?; + } + Ok(()) +} + +async fn rename_new_plain_file(source: &Path, target: &Path) -> Result<()> { + require_plain_file(source, "restore rename source").await?; + if entry_exists(target).await? { + return Err(storage_error(format!( + "restore rename target {} already exists", + target.display() + ))); + } + tokio::fs::rename(source, target).await.map_err(|error| { + storage_error(format!( + "rename restore file {} to {}: {error}", + source.display(), + target.display() + )) + }) +} + +async fn remove_plain_file(path: &Path, description: &str) -> Result<()> { + require_plain_file(path, description).await?; + tokio::fs::remove_file(path) + .await + .map_err(|error| storage_error(format!("remove {description} {}: {error}", path.display()))) +} + +async fn remove_plain_file_if_present(path: &Path, description: &str) -> Result<()> { + if plain_file_exists(path, description).await? { + remove_plain_file(path, description).await?; + } + Ok(()) +} + +async fn sync_directory(path: &Path) -> Result<()> { + tokio::fs::File::open(path) + .await + .map_err(|error| { + storage_error(format!( + "open restore directory {}: {error}", + path.display() + )) + })? + .sync_all() + .await + .map_err(|error| { + storage_error(format!( + "sync restore directory {}: {error}", + path.display() + )) + }) +} + +fn invalid_layout(paths: &RestorePaths, state: RestoreState) -> BlazeError { + storage_error(format!( + "slot '{}' has an invalid {:?} restore layout", + paths.instance_id, state + )) +} + +fn storage_error(message: impl Into) -> BlazeError { + BlazeError::StorageError { + msg: message.into(), + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use async_trait::async_trait; + use blaze_core::storage::{ + AcquireOpts, PoolStatus, StorageAcquireError, StorageProvider, StorageSlot, + }; + + use super::*; + + /// Size used only to build test sources larger than one copy step. + #[cfg(feature = "test-failpoints")] + const COPY_BUFFER_SIZE: usize = 64 * 1024; + + struct UnsupportedStorage; + + #[async_trait] + impl StorageProvider for UnsupportedStorage { + async fn probe(&self) -> Result { + Ok(true) + } + + async fn acquire( + &self, + _opts: &AcquireOpts, + ) -> std::result::Result { + Err(StorageAcquireError::clean(storage_error( + "acquire unavailable", + ))) + } + + async fn release(&self, _slot: StorageSlot) -> Result<()> { + Ok(()) + } + + async fn reconstruct(&self, _instance_id: &str) -> Result { + Err(storage_error("reconstruct unavailable")) + } + + async fn sync_artifacts(&self, _slot: &StorageSlot) -> Result<()> { + Ok(()) + } + + fn pool_status(&self) -> PoolStatus { + PoolStatus::default() + } + } + + async fn fixture( + instance_id: &str, + ) -> (tempfile::TempDir, FileStorageProvider, StorageSlot, PathBuf) { + let temp = tempfile::tempdir().expect("temporary storage"); + let instances = temp.path().join("instances"); + let checkpoints = temp.path().join("checkpoints"); + tokio::fs::create_dir(&instances) + .await + .expect("instances directory"); + tokio::fs::create_dir(&checkpoints) + .await + .expect("checkpoints directory"); + let provider = FileStorageProvider::new(instances); + let slot = provider + .acquire(&AcquireOpts { + instance_id: instance_id.to_string(), + rootfs_size: 64, + mem_size: 32, + }) + .await + .expect("storage slot"); + tokio::fs::write(&slot.rootfs_path, b"live-rootfs") + .await + .expect("live rootfs"); + let source = checkpoints.join("rootfs.snap"); + tokio::fs::write(&source, b"checkpoint-rootfs") + .await + .expect("checkpoint rootfs"); + (temp, provider, slot, source) + } + + async fn rootfs(path: &Path) -> Vec { + tokio::fs::read(path).await.expect("read rootfs") + } + + #[tokio::test] + async fn restore_contract_is_opt_in_and_fail_closed() { + let provider = UnsupportedStorage; + let slot = StorageSlot { + id: "unsupported".to_string(), + rootfs_path: PathBuf::from("rootfs"), + mem_path: PathBuf::from("memory"), + mem_diff_path: PathBuf::from("memory-diff"), + rootfs_diff_path: PathBuf::from("rootfs-diff"), + instance_dir: PathBuf::from("instance"), + }; + let transaction = StorageRestoreTransaction { + instance_id: slot.id.clone(), + transaction_id: Uuid::new_v4(), + }; + + assert!(!provider.supports_checkpoint_restore()); + assert!( + provider + .stage_checkpoint_restore(&slot, Path::new("checkpoint")) + .await + .is_err() + ); + assert!( + provider + .activate_checkpoint_restore(&transaction) + .await + .is_err() + ); + assert!( + provider + .commit_checkpoint_restore(&transaction) + .await + .is_err() + ); + assert!( + provider + .abort_checkpoint_restore(&transaction) + .await + .is_err() + ); + assert!( + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .is_err() + ); + } + + #[tokio::test] + async fn stage_keeps_the_live_rootfs_running_image_unchanged() { + let (_temp, provider, slot, source) = fixture("stage-independent").await; + + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage restore"); + + assert!(provider.supports_checkpoint_restore()); + assert_eq!(transaction.instance_id, slot.id); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect_err("a second transaction must fail closed"); + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort staged restore"); + } + + /// Sparse allocation is filesystem dependent, so this assertion is limited + /// to the platform the daemon targets, matching the capture-side test. + #[cfg(target_os = "linux")] + #[tokio::test] + async fn staging_preserves_sparse_extents() { + use std::io::{Seek, Write}; + use std::os::unix::fs::MetadataExt; + + const LOGICAL_LEN: u64 = 64 * 1024 * 1024; + const FIRST_OFFSET: u64 = 4 * 1024; + const LAST_OFFSET: u64 = 48 * 1024 * 1024 + 137; + const FIRST_DATA: &[u8] = b"first-restore-extent"; + const LAST_DATA: &[u8] = b"last-restore-extent"; + + let (_temp, provider, slot, source) = fixture("stage-sparse").await; + let mut checkpoint = std::fs::OpenOptions::new() + .write(true) + .open(&source) + .expect("open checkpoint rootfs"); + checkpoint.set_len(LOGICAL_LEN).expect("logical length"); + checkpoint + .seek(std::io::SeekFrom::Start(FIRST_OFFSET)) + .expect("seek first extent"); + checkpoint.write_all(FIRST_DATA).expect("first extent"); + checkpoint + .seek(std::io::SeekFrom::Start(LAST_OFFSET)) + .expect("seek last extent"); + checkpoint.write_all(LAST_DATA).expect("last extent"); + checkpoint.sync_all().expect("sync checkpoint"); + let source_blocks = checkpoint.metadata().expect("source metadata").blocks(); + drop(checkpoint); + + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage sparse restore"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + provider + .commit_checkpoint_restore(&transaction) + .await + .expect("commit"); + + let metadata = std::fs::metadata(&slot.rootfs_path).expect("restored metadata"); + assert_eq!( + metadata.len(), + LOGICAL_LEN, + "logical length must be restored" + ); + assert!( + metadata.blocks().saturating_mul(512) < LOGICAL_LEN / 4, + "restore allocated {} bytes for a {LOGICAL_LEN}-byte sparse checkpoint", + metadata.blocks().saturating_mul(512) + ); + assert!( + metadata.blocks() <= source_blocks.saturating_add(32), + "restore used {} blocks for a checkpoint using {source_blocks} blocks", + metadata.blocks() + ); + + let restored = std::fs::read(&slot.rootfs_path).expect("restored rootfs"); + assert_eq!( + &restored[FIRST_OFFSET as usize..FIRST_OFFSET as usize + FIRST_DATA.len()], + FIRST_DATA + ); + assert_eq!( + &restored[LAST_OFFSET as usize..LAST_OFFSET as usize + LAST_DATA.len()], + LAST_DATA + ); + } + + #[tokio::test] + async fn activated_restore_can_be_aborted_to_the_predecessor() { + let (_temp, provider, slot, source) = fixture("activate-abort").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"checkpoint-rootfs"); + + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + ensure_finalized_layout(&restore_paths(&provider, &slot.id).await.unwrap()) + .await + .unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn activation_retains_the_original_rootfs_inode_until_finalization() { + use std::os::unix::fs::MetadataExt; + + let (_temp, provider, slot, source) = fixture("retain-inode").await; + let original_inode = tokio::fs::metadata(&slot.rootfs_path) + .await + .expect("live metadata") + .ino(); + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + assert_eq!( + tokio::fs::metadata(&slot.rootfs_path) + .await + .expect("staged live metadata") + .ino(), + original_inode + ); + + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + assert_eq!( + tokio::fs::metadata(&paths.backup) + .await + .expect("backup metadata") + .ino(), + original_inode + ); + assert_ne!( + tokio::fs::metadata(&paths.rootfs) + .await + .expect("selected metadata") + .ino(), + original_inode + ); + + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort"); + assert_eq!( + tokio::fs::metadata(&slot.rootfs_path) + .await + .expect("restored metadata") + .ino(), + original_inode + ); + } + + #[tokio::test] + async fn committed_restore_releases_the_predecessor() { + let (_temp, provider, slot, source) = fixture("activate-commit").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + + provider + .commit_checkpoint_restore(&transaction) + .await + .expect("commit"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"checkpoint-rootfs"); + ensure_finalized_layout(&restore_paths(&provider, &slot.id).await.unwrap()) + .await + .unwrap(); + } + + #[tokio::test] + async fn stale_transaction_handle_cannot_select_a_rootfs() { + let (_temp, provider, slot, source) = fixture("stale-handle").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let stale = StorageRestoreTransaction { + instance_id: transaction.instance_id.clone(), + transaction_id: Uuid::new_v4(), + }; + + provider + .activate_checkpoint_restore(&stale) + .await + .expect_err("stale transaction must be rejected"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort"); + } + + #[tokio::test] + async fn staging_rederives_provider_paths_from_the_slot_id() { + let (temp, provider, slot, source) = fixture("canonical-slot").await; + let external = temp.path().join("external-rootfs"); + tokio::fs::write(&external, b"external") + .await + .expect("external rootfs"); + let mut forged = slot.clone(); + forged.rootfs_path = external.clone(); + forged.instance_dir = temp.path().to_path_buf(); + + let transaction = provider + .stage_checkpoint_restore(&forged, &source) + .await + .expect("stage through canonical slot"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"checkpoint-rootfs"); + assert_eq!(rootfs(&external).await, b"external"); + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort"); + } + + #[cfg(unix)] + #[tokio::test] + async fn staging_rejects_linked_sources_and_slot_paths() { + use std::os::unix::fs::symlink; + + let (temp, provider, slot, source) = fixture("linked-paths").await; + let linked_source = temp.path().join("linked-source"); + symlink(&source, &linked_source).expect("source link"); + provider + .stage_checkpoint_restore(&slot, &linked_source) + .await + .expect_err("linked source must be rejected"); + + tokio::fs::remove_file(&slot.rootfs_path) + .await + .expect("remove live rootfs"); + let external = temp.path().join("external-rootfs"); + tokio::fs::write(&external, b"external") + .await + .expect("external rootfs"); + symlink(&external, &slot.rootfs_path).expect("rootfs link"); + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect_err("linked live rootfs must be rejected"); + assert_eq!(rootfs(&external).await, b"external"); + } + + #[cfg(unix)] + #[tokio::test] + async fn staging_rejects_a_linked_slot_directory() { + use std::os::unix::fs::symlink; + + let (temp, provider, slot, source) = fixture("linked-slot").await; + tokio::fs::remove_dir_all(&slot.instance_dir) + .await + .expect("remove slot"); + let external = temp.path().join("external-slot"); + tokio::fs::create_dir(&external) + .await + .expect("external slot"); + tokio::fs::write(external.join("rootfs.ext4"), b"external") + .await + .expect("external rootfs"); + symlink(&external, &slot.instance_dir).expect("slot link"); + + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect_err("linked slot directory must be rejected"); + + assert_eq!(rootfs(&external.join("rootfs.ext4")).await, b"external"); + } + + #[cfg(unix)] + #[tokio::test] + async fn staging_rejects_linked_transaction_artifacts() { + use std::os::unix::fs::symlink; + + let (temp, provider, slot, source) = fixture("linked-artifact").await; + let external = temp.path().join("external"); + tokio::fs::write(&external, b"external") + .await + .expect("external"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + symlink(&external, &paths.staged).expect("stage link"); + + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect_err("linked transaction artifact must fail closed"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + assert_eq!(rootfs(&external).await, b"external"); + } + + #[tokio::test] + async fn staging_rejects_instance_id_path_components() { + let (_temp, provider, mut slot, source) = fixture("valid-id").await; + slot.id = "../escape".to_string(); + + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect_err("path component must be rejected"); + } + + #[tokio::test] + async fn restart_aborts_a_staged_restore() { + let (_temp, provider, slot, source) = fixture("restart-staged").await; + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + ensure_finalized_layout(&restore_paths(&restarted, &slot.id).await.unwrap()) + .await + .unwrap(); + } + + #[tokio::test] + async fn restart_aborts_an_activated_restore() { + let (_temp, provider, slot, source) = fixture("restart-activated").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + } + + #[tokio::test] + async fn restart_finishes_a_durable_commit_intent() { + let (_temp, provider, slot, source) = fixture("restart-commit").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + let mut journal = require_journal(&paths).await.expect("journal"); + journal.state = RestoreState::Committing; + replace_journal(&paths, &journal) + .await + .expect("commit intent"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"checkpoint-rootfs"); + ensure_finalized_layout(&restore_paths(&restarted, &slot.id).await.unwrap()) + .await + .unwrap(); + } + + #[tokio::test] + async fn restart_cleans_a_partial_copy_without_touching_the_live_rootfs() { + let (_temp, provider, slot, _source) = fixture("restart-copying").await; + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + tokio::fs::write(&paths.copying, b"partial") + .await + .expect("partial copy"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + assert!(!paths.copying.exists()); + } + + #[tokio::test] + async fn restart_recovers_after_retaining_the_predecessor() { + let (_temp, provider, slot, source) = fixture("restart-after-backup").await; + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + rename_new_plain_file(&paths.rootfs, &paths.backup) + .await + .expect("retain predecessor"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + assert!(!paths.staged.exists()); + assert!(!paths.backup.exists()); + } + + #[tokio::test] + async fn restart_recovers_after_switching_before_journal_update() { + let (_temp, provider, slot, source) = fixture("restart-after-switch").await; + provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + rename_new_plain_file(&paths.rootfs, &paths.backup) + .await + .expect("retain predecessor"); + rename_new_plain_file(&paths.staged, &paths.rootfs) + .await + .expect("switch rootfs"); + + let restarted = FileStorageProvider::new(provider.instances_dir.clone()); + restarted + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + } + + #[tokio::test] + async fn corrupt_journal_preserves_both_rootfs_versions() { + let (_temp, provider, slot, source) = fixture("corrupt-journal").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + tokio::fs::write(&paths.journal, b"not-json") + .await + .expect("corrupt journal"); + + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .expect_err("corrupt journal must fail closed"); + + assert_eq!(rootfs(&paths.rootfs).await, b"checkpoint-rootfs"); + assert_eq!(rootfs(&paths.backup).await, b"live-rootfs"); + } + + #[cfg(feature = "test-failpoints")] + async fn cancel_at( + provider: FileStorageProvider, + slot: StorageSlot, + source: PathBuf, + failpoint: &'static str, + ) { + let hook = crate::failpoint::TestFailpoint::new(&[failpoint]); + let operation_hook = hook.clone(); + let operation = tokio::spawn(async move { + operation_hook + .run(provider.stage_checkpoint_restore(&slot, &source)) + .await + }); + hook.wait_until_paused().await; + operation.abort(); + assert!( + operation + .await + .expect_err("operation must be cancelled") + .is_cancelled() + ); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn cancelled_copy_is_removed_before_publication() { + let (_temp, provider, slot, source) = fixture("cancel-copy").await; + tokio::fs::write(&source, vec![42_u8; COPY_BUFFER_SIZE * 4]) + .await + .expect("large checkpoint"); + let instances = provider.instances_dir.clone(); + let id = slot.id.clone(); + let rootfs_path = slot.rootfs_path.clone(); + + cancel_at(provider, slot, source, "storage-restore-after-copy").await; + + let restarted = FileStorageProvider::new(instances); + restarted + .reconcile_checkpoint_restore(&id) + .await + .expect("reconcile cancellation"); + assert_eq!(rootfs(&rootfs_path).await, b"live-rootfs"); + ensure_finalized_layout(&restore_paths(&restarted, &id).await.unwrap()) + .await + .unwrap(); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn failed_second_rename_immediately_restores_the_predecessor() { + let (_temp, provider, slot, source) = fixture("failed-switch").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let hook = crate::failpoint::TestFailpoint::new(&["storage-restore-switch"]); + + hook.run(provider.activate_checkpoint_restore(&transaction)) + .await + .expect_err("selecting the staged rootfs must fail"); + + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + provider + .abort_checkpoint_restore(&transaction) + .await + .expect("abort retained stage"); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn failed_switch_compensation_remains_reconcilable() { + let (_temp, provider, slot, source) = fixture("failed-switch-rollback").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let hook = crate::failpoint::TestFailpoint::new(&[ + "storage-restore-switch", + "storage-restore-switch-rollback", + ]); + + hook.run(provider.activate_checkpoint_restore(&transaction)) + .await + .expect_err("selection and immediate compensation must fail"); + + let paths = restore_paths(&provider, &slot.id).await.expect("paths"); + assert!(!paths.rootfs.exists()); + assert_eq!(rootfs(&paths.backup).await, b"live-rootfs"); + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile retained predecessor"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn cancelled_activation_after_backup_is_reconciled() { + let (_temp, provider, slot, source) = fixture("cancel-activation").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + let hook = crate::failpoint::TestFailpoint::new(&["storage-restore-after-backup"]); + let operation_hook = hook.clone(); + let operation_provider = FileStorageProvider::new(provider.instances_dir.clone()); + let operation_transaction = transaction.clone(); + let operation = tokio::spawn(async move { + operation_hook + .run(operation_provider.activate_checkpoint_restore(&operation_transaction)) + .await + }); + hook.wait_until_paused().await; + operation.abort(); + assert!( + operation + .await + .expect_err("activation must be cancelled") + .is_cancelled() + ); + + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn cancelled_commit_intent_is_completed_on_restart() { + let (_temp, provider, slot, source) = fixture("cancel-commit").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + let hook = crate::failpoint::TestFailpoint::new(&["storage-restore-after-commit-intent"]); + let operation_hook = hook.clone(); + let operation_provider = FileStorageProvider::new(provider.instances_dir.clone()); + let operation_transaction = transaction.clone(); + let operation = tokio::spawn(async move { + operation_hook + .run(operation_provider.commit_checkpoint_restore(&operation_transaction)) + .await + }); + hook.wait_until_paused().await; + operation.abort(); + assert!( + operation + .await + .expect_err("commit must be cancelled") + .is_cancelled() + ); + + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"checkpoint-rootfs"); + } + + #[cfg(feature = "test-failpoints")] + #[tokio::test] + async fn cancelled_abort_is_completed_on_restart() { + let (_temp, provider, slot, source) = fixture("cancel-abort").await; + let transaction = provider + .stage_checkpoint_restore(&slot, &source) + .await + .expect("stage"); + provider + .activate_checkpoint_restore(&transaction) + .await + .expect("activate"); + let hook = crate::failpoint::TestFailpoint::new(&["storage-restore-after-discard"]); + let operation_hook = hook.clone(); + let operation_provider = FileStorageProvider::new(provider.instances_dir.clone()); + let operation_transaction = transaction.clone(); + let operation = tokio::spawn(async move { + operation_hook + .run(operation_provider.abort_checkpoint_restore(&operation_transaction)) + .await + }); + hook.wait_until_paused().await; + operation.abort(); + assert!( + operation + .await + .expect_err("abort must be cancelled") + .is_cancelled() + ); + + provider + .reconcile_checkpoint_restore(&slot.id) + .await + .expect("reconcile"); + assert_eq!(rootfs(&slot.rootfs_path).await, b"live-rootfs"); + } +} diff --git a/src/blaze/crates/blazed/src/sandbox.rs b/src/blaze/crates/blazed/src/sandbox.rs index 511984343b..bc7e1f8d94 100644 --- a/src/blaze/crates/blazed/src/sandbox.rs +++ b/src/blaze/crates/blazed/src/sandbox.rs @@ -3,8 +3,10 @@ mod checkpoint; mod manager; +mod restore; mod storage_sync; pub(crate) mod template; pub use manager::{CreateSandbox, SandboxManager, SandboxManagerInit}; +pub use restore::{RestoreSandbox, RestoreSandboxResult}; pub(crate) use storage_sync::StorageSyncLoop; diff --git a/src/blaze/crates/blazed/src/sandbox/manager.rs b/src/blaze/crates/blazed/src/sandbox/manager.rs index c37ab76129..a8dd76a8de 100644 --- a/src/blaze/crates/blazed/src/sandbox/manager.rs +++ b/src/blaze/crates/blazed/src/sandbox/manager.rs @@ -21,9 +21,10 @@ use crate::guest::{GuestClient, GuestExecResult, MAX_GUEST_FILE_BYTES}; use crate::metrics::Metrics; use crate::sandbox::template::TemplateCatalog; use crate::spawner::{ - BackendSpawnRequest, DynBackendInstance, SpawnerRegistry, spawn_with_runtime_directory, + BackendSpawnRequest, DynBackendInstance, DynSpawner, SpawnerRegistry, + spawn_with_runtime_directory, }; -use crate::state_store::StateStore; +use crate::state_store::{OwnedRunDir, StateStore}; const GUEST_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); @@ -111,6 +112,11 @@ pub struct SandboxManagerResources { } impl SandboxManager { + /// Return the retained runtime-directory owner for one sandbox. + pub(super) fn run_directory(&self, id: Uuid) -> Result { + self.state_store.run_dir(id) + } + /// Build a manager around state loaded from the durable state directory. pub fn new(init: SandboxManagerInit) -> (Self, SandboxManagerResources) { let SandboxManagerInit { @@ -182,6 +188,17 @@ impl SandboxManager { } } + pub(super) fn spawner(&self, backend: BackendKind) -> Option { + self.spawners.get(backend) + } + + pub(super) fn remove_backend_owner(&self, id: Uuid) -> Option { + match self.backend_instances.lock() { + Ok(mut instances) => instances.remove(&id), + Err(poisoned) => poisoned.into_inner().remove(&id), + } + } + #[cfg(test)] pub(crate) fn insert_backend_owner(&self, id: Uuid, owner: DynBackendInstance) -> Result<()> { self.backend_instances @@ -950,7 +967,7 @@ impl SandboxManager { } } - fn retain_backend(&self, id: Uuid, backend: DynBackendInstance) -> Option { + pub(super) fn retain_backend(&self, id: Uuid, backend: DynBackendInstance) -> Option { match self.backend_instances.lock() { Ok(mut instances) => { instances.insert(id, backend); diff --git a/src/blaze/crates/blazed/src/sandbox/restore.rs b/src/blaze/crates/blazed/src/sandbox/restore.rs new file mode 100644 index 0000000000..0dd6f068f9 --- /dev/null +++ b/src/blaze/crates/blazed/src/sandbox/restore.rs @@ -0,0 +1,649 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Recoverable replacement of a running sandbox from a committed checkpoint. + +use std::path::PathBuf; +use std::sync::Arc; + +use blaze_core::backend::{RestoreRequest, SnapshotKind}; +use blaze_core::checkpoint::validate_checkpoint_id; +use blaze_core::lifecycle::{BackendOwnership, OperationPhase, SandboxInstance, SandboxState}; +use blaze_core::storage::StorageRestoreTransaction; +use tokio::sync::OwnedMutexGuard; +use uuid::Uuid; + +use crate::checkpoint_store::CheckpointStoreError; +use crate::error::{BlazeDaemonError, Result}; +use crate::spawner::{BackendRestoreRequest, DynBackendInstance, restore_with_runtime_directory}; + +use super::manager::SandboxManager; + +/// Inputs resolved from the current daemon configuration. +#[derive(Debug, Clone)] +pub struct RestoreSandbox { + /// Committed checkpoint selected by the caller. + pub checkpoint_id: String, + /// Current executable for the checkpoint's backend. + pub binary_path: PathBuf, +} + +/// Result of one completed checkpoint restore. +#[derive(Debug, Clone)] +pub struct RestoreSandboxResult { + /// Updated durable sandbox record. + pub instance: SandboxInstance, + /// Checkpoint now selected by the catalog HEAD. + pub checkpoint_id: String, +} + +impl SandboxManager { + /// Replace a running backend and rootfs from one verified checkpoint. + pub async fn restore( + self: &Arc, + id: Uuid, + request: RestoreSandbox, + ) -> Result { + validate_checkpoint_id(&request.checkpoint_id) + .map_err(|error| BlazeDaemonError::BadRequest(error.to_string()))?; + let operation = self.operation_lock(id).lock_owned().await; + let manager = Arc::clone(self); + crate::failpoint::spawn( + async move { manager.restore_supervised(id, request, operation).await }, + ) + .await + .map_err(|error| { + let recovery = self.mark_recovery(id).err(); + BlazeDaemonError::RecoveryRequired(format!( + "restore supervisor stopped unexpectedly: {error}{}", + recovery + .map(|error| format!("; recovery state persistence failed: {error}")) + .unwrap_or_default() + )) + })? + } + + async fn restore_supervised( + self: Arc, + id: Uuid, + request: RestoreSandbox, + operation: OwnedMutexGuard<()>, + ) -> Result { + let manager = Arc::clone(&self); + let result = + match crate::failpoint::spawn(async move { manager.restore_worker(id, request).await }) + .await + { + Ok(result) => result, + Err(error) => { + let recovery = self.mark_recovery(id).err(); + Err(BlazeDaemonError::RecoveryRequired(format!( + "restore worker stopped unexpectedly: {error}{}", + recovery + .map(|error| format!("; recovery state persistence failed: {error}")) + .unwrap_or_default() + ))) + } + }; + drop(operation); + result + } + + async fn restore_worker( + self: Arc, + id: Uuid, + request: RestoreSandbox, + ) -> Result { + let mut instance = self.get(id)?; + if let Some(journal) = &instance.operation { + return Err(BlazeDaemonError::RecoveryRequired(format!( + "instance {id} has unfinished {} operation", + journal.kind + ))); + } + if instance.state != SandboxState::Running { + return Err(BlazeDaemonError::Conflict(format!( + "instance {id} is {}, expected running", + instance.state + ))); + } + + let checkpoint_id = request.checkpoint_id.clone(); + let verify_manager = Arc::clone(&self); + let selected_id = checkpoint_id.clone(); + let target = crate::failpoint::spawn_blocking(move || { + verify_manager + .checkpoints + .verify_restore_target(id, &checkpoint_id) + .map_err(|e| checkpoint_lookup_error(e, id, &selected_id)) + }) + .await + .map_err(|error| { + BlazeDaemonError::Internal(format!( + "checkpoint restore verification blocking task: {error}" + )) + })??; + let target_metadata = target.metadata().clone(); + let snapshot_path = target + .artifact_path("vmstate.snap") + .map_err(checkpoint_store_error)?; + let memory_path = target + .artifact_path("memory.snap") + .map_err(checkpoint_store_error)?; + let rootfs_path = target + .artifact_path("rootfs.snap") + .map_err(checkpoint_store_error)?; + if target_metadata.policy_name != instance.policy_name + || target_metadata.image_digest != instance.image_digest + || target_metadata.backend != instance.backend + { + return Err(BlazeDaemonError::Conflict(format!( + "checkpoint {} runtime identity does not match instance {id}", + request.checkpoint_id + ))); + } + if target_metadata.snapshot_kind != SnapshotKind::Full { + return Err(BlazeDaemonError::UnsupportedOperation(format!( + "checkpoint {} does not contain a full snapshot", + request.checkpoint_id + ))); + } + + let current_backend = self.backend_owner(id).ok_or_else(|| { + BlazeDaemonError::Conflict(format!("instance {id} has no backend owner")) + })?; + if current_backend.instance_id() != id || current_backend.backend() != instance.backend { + self.mark_recovery(id)?; + return Err(BlazeDaemonError::RecoveryRequired(format!( + "instance {id} backend owner identity does not match durable state" + ))); + } + self.require_restore_backend_live(id, ¤t_backend) + .await?; + if !self.storage.supports_checkpoint_restore() { + return Err(BlazeDaemonError::UnsupportedOperation(format!( + "instance {id} configured storage does not support checkpoint restore" + ))); + } + let spawner = self.spawner(target_metadata.backend).ok_or_else(|| { + BlazeDaemonError::UnsupportedOperation(format!( + "instance {id} has no restore adapter for {}", + target_metadata.backend + )) + })?; + let capability = spawner + .restore_capability(&request.binary_path) + .await? + .ok_or_else(|| { + BlazeDaemonError::UnsupportedOperation(format!( + "instance {id} backend {} does not support checkpoint restore", + target_metadata.backend + )) + })?; + if capability.backend != target_metadata.backend + || capability.version != target_metadata.backend_version + || capability.snapshot_kind != target_metadata.snapshot_kind + { + return Err(BlazeDaemonError::UnsupportedOperation(format!( + "checkpoint {} requires {} version {:?} {:?}, but the current adapter provides \ + {} version {:?} {:?}", + request.checkpoint_id, + target_metadata.backend, + target_metadata.backend_version, + target_metadata.snapshot_kind, + capability.backend, + capability.version, + capability.snapshot_kind + ))); + } + let storage = self.storage.reconstruct(&id.to_string()).await?; + let expose_guest_socket = !current_backend.guest_socket_path().as_os_str().is_empty(); + + instance.begin_restore_operation(request.checkpoint_id.clone())?; + crate::failpoint::state("restore-begin-state") + .and_then(|_| self.persist_and_retain(instance.clone()))?; + crate::failpoint::pause("restore-after-begin").await; + + let transaction = match crate::failpoint::storage("restore-storage-stage") { + Ok(()) => { + self.storage + .stage_checkpoint_restore(&storage, &rootfs_path) + .await + } + Err(error) => Err(error), + }; + let transaction = match transaction { + Ok(transaction) => transaction, + Err(error) => { + return Err(self + .fail_before_restore_stop(instance, None, error.into()) + .await); + } + }; + if let Err(error) = instance + .advance_restore_phase(OperationPhase::RestoreStorageStaged) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-staged-state")?; + self.persist_and_retain(instance.clone()) + }) + { + return Err(self + .fail_before_restore_stop(instance, Some(&transaction), error) + .await); + } + crate::failpoint::pause("restore-after-stage").await; + + let stopped = match crate::failpoint::backend("restore-backend-stop") { + Ok(()) => current_backend.kill().await, + Err(error) => Err(error), + }; + if let Err(error) = stopped { + instance.backend_ownership = BackendOwnership::Unknown; + let abort = self + .storage + .abort_checkpoint_restore(&transaction) + .await + .err(); + return Err(self.fail_after_restore_stop( + instance, + format!( + "current backend termination failed: {error}{}", + abort + .map(|error| format!("; staged storage cleanup failed: {error}")) + .unwrap_or_default() + ), + )); + } + + instance.backend_ownership = BackendOwnership::Stopped; + let stopped_state = instance + .advance_restore_phase(OperationPhase::RestoreBackendStopped) + .and_then(|_| instance.transition(SandboxState::Restoring)) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-stopped-state")?; + self.persist_and_retain(instance.clone()) + }); + if let Err(error) = stopped_state { + self.remove_backend_owner(id); + return Err(self.fail_after_restore_stop( + instance, + format!("backend stopped but lifecycle commit failed: {error}"), + )); + } + self.remove_backend_owner(id); + crate::failpoint::pause("restore-after-stop").await; + + let activated = match crate::failpoint::storage("restore-storage-activate") { + Ok(()) => self.storage.activate_checkpoint_restore(&transaction).await, + Err(error) => Err(error), + }; + if let Err(error) = activated { + let abort = self + .storage + .abort_checkpoint_restore(&transaction) + .await + .err(); + return Err(self.fail_after_restore_stop( + instance, + format!( + "replacement storage activation failed: {error}{}", + abort + .map(|error| format!("; predecessor restore failed: {error}")) + .unwrap_or_default() + ), + )); + } + if let Err(error) = instance + .advance_restore_phase(OperationPhase::RestoreStorageActivated) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-activated-state")?; + self.persist_and_retain(instance.clone()) + }) + { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement storage activated but lifecycle commit failed: {error}"), + )); + } + crate::failpoint::pause("restore-after-activate").await; + + let run_dir = match self.run_directory(id) { + Ok(run_dir) => run_dir, + Err(error) => { + return Err(self.fail_after_restore_stop( + instance, + format!("runtime directory lookup after storage activation failed: {error}"), + )); + } + }; + if let Err(error) = spawner.prepare_spawn(&run_dir).await { + return Err(self.fail_after_restore_stop( + instance, + format!("prepare replacement backend ownership failed: {error}"), + )); + } + instance.backend_ownership = BackendOwnership::Starting; + if let Err(error) = crate::failpoint::state("restore-starting-state") + .and_then(|_| self.persist_and_retain(instance.clone())) + { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement backend intent commit failed: {error}"), + )); + } + + let restored = match crate::failpoint::backend("restore-backend-start") { + Ok(()) => match BackendRestoreRequest::new( + RestoreRequest { + instance_id: id, + binary_path: request.binary_path, + storage, + snapshot_path, + mem_path: memory_path, + checkpoint_backend: target_metadata.backend, + expected_version: target_metadata.backend_version.clone(), + snapshot_kind: target_metadata.snapshot_kind, + expose_guest_socket, + }, + run_dir, + ) { + Ok(request) => restore_with_runtime_directory(spawner.as_ref(), request).await, + Err(error) => Err(crate::spawner::SpawnFailure::clean(error)), + }, + Err(error) => Err(crate::spawner::SpawnFailure::clean(error)), + }; + let restored = match restored { + Ok(owner) => owner, + Err(error) => { + let (source, owner) = error.into_parts(); + if let Some(owner) = owner { + let _ = self.retain_backend(id, owner); + instance.backend_ownership = BackendOwnership::Running; + } else { + instance.backend_ownership = BackendOwnership::Stopped; + } + return Err(self.fail_after_restore_stop( + instance, + format!("replacement backend start failed: {source}"), + )); + } + }; + if let Some(error) = self.retain_backend(id, restored.clone()) { + instance.backend_ownership = BackendOwnership::Running; + return Err(self.fail_after_restore_stop(instance, error)); + } + instance.backend_ownership = BackendOwnership::Running; + + if restored.instance_id() != id + || restored.backend() != target_metadata.backend + || restored.version().map(str::to_string) != target_metadata.backend_version + { + return Err(self.fail_after_restore_stop( + instance, + format!( + "replacement backend identity ({}, {}, {:?}) does not match checkpoint \ + identity ({id}, {}, {:?})", + restored.instance_id(), + restored.backend(), + restored.version(), + target_metadata.backend, + target_metadata.backend_version + ), + )); + } + if let Err(error) = instance + .advance_restore_phase(OperationPhase::RestoreBackendStarted) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-started-state")?; + self.persist_and_retain(instance.clone()) + }) + { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement backend started but lifecycle commit failed: {error}"), + )); + } + if let Err(error) = self + .verify_restored_backend(id, &restored, expose_guest_socket) + .await + { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement backend readiness failed: {error}"), + )); + } + + let head_updated = match crate::failpoint::storage("restore-head-update") { + Ok(()) => { + let checkpoints = self.checkpoints.clone(); + crate::failpoint::spawn_blocking(move || { + checkpoints + .set_head_verified(&target) + .map_err(checkpoint_store_error) + }) + .await + .map_err(|error| { + BlazeDaemonError::RecoveryRequired(format!( + "checkpoint HEAD update blocking task stopped unexpectedly: {error}" + )) + })? + } + Err(error) => Err(error.into()), + }; + if let Err(error) = head_updated { + let observed = self.observe_checkpoint_head(id).await; + return Err(self.fail_after_restore_stop( + instance, + format!( + "checkpoint HEAD update failed: {error}; observed HEAD after failure: \ + {observed:?}" + ), + )); + } + if let Err(error) = instance + .advance_restore_phase(OperationPhase::RestoreHeadUpdated) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-head-state")?; + self.persist_and_retain(instance.clone()) + }) + { + return Err(self.fail_after_restore_stop( + instance, + format!("checkpoint HEAD changed but lifecycle commit failed: {error}"), + )); + } + crate::failpoint::pause("restore-after-head").await; + + let committed = match crate::failpoint::storage("restore-storage-commit") { + Ok(()) => self.storage.commit_checkpoint_restore(&transaction).await, + Err(error) => Err(error), + }; + if let Err(error) = committed { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement storage commit failed: {error}"), + )); + } + if let Err(error) = instance + .advance_restore_phase(OperationPhase::RestoreStorageCommitted) + .map_err(BlazeDaemonError::from) + .and_then(|_| { + crate::failpoint::state("restore-committed-state")?; + self.persist_and_retain(instance.clone()) + }) + { + return Err(self.fail_after_restore_stop( + instance, + format!("replacement storage committed but lifecycle commit failed: {error}"), + )); + } + + let recovery_instance = instance.clone(); + instance.transition(SandboxState::Running)?; + instance.finish_operation(); + if let Err(error) = crate::failpoint::state("restore-final-state") + .and_then(|_| self.persist_and_retain(instance.clone())) + { + return Err(self.fail_after_restore_stop( + recovery_instance, + format!("replacement is live but final lifecycle commit failed: {error}"), + )); + } + Ok(RestoreSandboxResult { + instance, + checkpoint_id: request.checkpoint_id, + }) + } + + /// Report which checkpoint HEAD names after a failed HEAD update. + /// + /// The observation stays on the blocking pool because it opens the catalog, + /// and it deliberately skips artifact verification so the recorded + /// identifier reaches the operator even when an artifact is unreadable. + async fn observe_checkpoint_head(&self, id: Uuid) -> Result> { + let checkpoints = self.checkpoints.clone(); + crate::failpoint::spawn_blocking(move || { + checkpoints.read_head_id(id).map_err(checkpoint_store_error) + }) + .await + .map_err(|error| { + BlazeDaemonError::Internal(format!( + "checkpoint HEAD observation blocking task: {error}" + )) + })? + } + + async fn require_restore_backend_live( + &self, + id: Uuid, + backend: &DynBackendInstance, + ) -> Result<()> { + match backend.try_wait().await { + Ok(None) => Ok(()), + Ok(Some(result)) => { + self.mark_recovery(id)?; + Err(BlazeDaemonError::RecoveryRequired(format!( + "instance {id} backend exited before restore \ + (exit={:?}, signal={:?})", + result.exit_code, result.signal + ))) + } + Err(error) => { + self.mark_recovery(id)?; + Err(BlazeDaemonError::RecoveryRequired(format!( + "instance {id} backend liveness is unknown: {error}" + ))) + } + } + } + + async fn verify_restored_backend( + &self, + id: Uuid, + backend: &DynBackendInstance, + expose_guest_socket: bool, + ) -> Result<()> { + self.require_restore_backend_live(id, backend).await?; + if expose_guest_socket { + // `wait_for_guest_ready` treats an empty socket path as immediately + // ready, so an adapter that silently drops the guest transport would + // otherwise pass readiness and commit a `Running` sandbox whose + // exec, read, and write requests all fail with a conflict. Require + // the replacement owner to expose the transport the captured runtime + // had before publishing the restore. + if backend.guest_socket_path().as_os_str().is_empty() { + return Err(BlazeDaemonError::RecoveryRequired(format!( + "instance {id} replacement backend does not expose the guest \ + transport the checkpoint captured" + ))); + } + self.wait_for_guest_ready(backend, "restore-guest-ready") + .await?; + } + self.require_restore_backend_live(id, backend).await + } + + async fn fail_before_restore_stop( + &self, + mut instance: SandboxInstance, + transaction: Option<&StorageRestoreTransaction>, + original: BlazeDaemonError, + ) -> BlazeDaemonError { + let storage_cleanup = match transaction { + Some(transaction) => self + .storage + .abort_checkpoint_restore(transaction) + .await + .map_err(BlazeDaemonError::from), + None => self + .storage + .reconcile_checkpoint_restore(&instance.id.to_string()) + .await + .map_err(BlazeDaemonError::from), + }; + if let Err(cleanup) = storage_cleanup { + return self.fail_after_restore_stop( + instance, + format!("{original}; staged storage cleanup failed: {cleanup}"), + ); + } + instance.finish_operation(); + if let Err(error) = self.persist_and_retain(instance.clone()) { + return self.fail_after_restore_stop( + instance, + format!("{original}; restore journal cleanup failed: {error}"), + ); + } + original + } + + fn fail_after_restore_stop( + &self, + instance: SandboxInstance, + cause: impl std::fmt::Display, + ) -> BlazeDaemonError { + let id = instance.id; + let recovery = self.mark_instance_recovery(instance).err(); + BlazeDaemonError::RecoveryRequired(format!( + "restore {id}: {cause}; resources retained{}", + recovery + .map(|error| format!("; recovery state persistence failed: {error}")) + .unwrap_or_default() + )) + } +} + +fn checkpoint_store_error(error: impl std::fmt::Display) -> BlazeDaemonError { + BlazeDaemonError::Internal(format!("checkpoint store: {error}")) +} + +/// Map a catalog lookup failure for a caller-supplied checkpoint identifier. +/// +/// An absent sandbox checkpoint namespace (path leaf = sandbox_id) or an absent +/// entry for the selected checkpoint itself (path leaf = selected_id) classifies +/// as not-found. Ancestor entries resolved through `validated_chain_from` carry +/// different path leaves; their absence is catalog corruption rather than a +/// permanent client-selection error, so they keep the internal classification. +fn checkpoint_lookup_error( + error: CheckpointStoreError, + sandbox_id: Uuid, + selected_id: &str, +) -> BlazeDaemonError { + if let CheckpointStoreError::Io { + ref source, + ref path, + .. + } = error + && source.kind() == std::io::ErrorKind::NotFound + { + let leaf = path.file_name().and_then(|name| name.to_str()); + let sandbox_name = sandbox_id.to_string(); + if leaf.is_some_and(|name| name == selected_id || name == sandbox_name.as_str()) { + return BlazeDaemonError::NotFound(format!("checkpoint store: {error}")); + } + } + checkpoint_store_error(error) +} diff --git a/src/blaze/crates/blazed/src/spawner.rs b/src/blaze/crates/blazed/src/spawner.rs index b18ecd95e3..975035d5df 100644 --- a/src/blaze/crates/blazed/src/spawner.rs +++ b/src/blaze/crates/blazed/src/spawner.rs @@ -16,7 +16,9 @@ use std::time::Duration; use std::time::Instant; use async_trait::async_trait; -use blaze_core::backend::{BackendKind, SnapshotRequest, SpawnRequest}; +use blaze_core::backend::{ + BackendKind, RestoreCapability, RestoreRequest, SnapshotRequest, SpawnRequest, +}; #[cfg(test)] use blaze_core::guest_protocol::DEFAULT_MAX_RESPONSE_BYTES; use blaze_core::{BlazeError, Result}; @@ -230,6 +232,61 @@ pub(crate) async fn spawn_with_runtime_directory( } } +/// Backend restore inputs paired with the opened runtime-directory owner. +#[derive(Debug, Clone)] +pub struct BackendRestoreRequest { + request: RestoreRequest, + /// Opened directory used for all replacement runtime artifacts. + pub run_dir: OwnedRunDir, +} + +impl BackendRestoreRequest { + pub(crate) fn new(request: RestoreRequest, run_dir: OwnedRunDir) -> Result { + if request.instance_id != run_dir.instance_id() { + return Err(BlazeError::BackendError { + msg: format!( + "restore request for {} does not match runtime-directory owner for {}", + request.instance_id, + run_dir.instance_id() + ), + }); + } + Ok(Self { request, run_dir }) + } +} + +impl Deref for BackendRestoreRequest { + type Target = RestoreRequest; + + fn deref(&self) -> &Self::Target { + &self.request + } +} + +/// Restore outcome that preserves ownership when cleanup cannot be confirmed. +pub type RestoreResult = std::result::Result; + +/// Restore one backend while attaching the runtime-directory owner to every +/// returned process owner, including a partial owner carried by a failure. +pub(crate) async fn restore_with_runtime_directory( + spawner: &dyn BackendSpawner, + request: BackendRestoreRequest, +) -> RestoreResult { + let run_dir = request.run_dir.clone(); + match spawner.restore(request).await { + Ok(owner) => Ok(bind_runtime_directory(owner, run_dir)), + Err(error) => { + let (source, owner) = error.into_parts(); + Err(match owner { + Some(owner) => { + SpawnFailure::with_owner(source, bind_runtime_directory(owner, run_dir)) + } + None => SpawnFailure::clean(source), + }) + } + } +} + /// Backend start failure that may retain ownership of a started process. pub struct SpawnFailure { source: BlazeError, @@ -320,6 +377,26 @@ pub trait BackendSpawner: Send + Sync { request: BackendSpawnRequest, ) -> std::result::Result; + /// Report the restore identity of the requested backend executable. + /// + /// `None` means restore is unsupported. Implementations that return a + /// version must inspect `binary_path` for every call rather than reusing + /// mutable process-wide state. + async fn restore_capability(&self, _binary_path: &Path) -> Result> { + Ok(None) + } + + /// Start an owned backend from committed checkpoint artifacts. + /// + /// Callers prepare the PID handoff through [`Self::prepare_spawn`] first. + /// Failures transfer any owner whose cleanup could not be confirmed. + async fn restore(&self, request: BackendRestoreRequest) -> RestoreResult { + let _ = request; + Err(SpawnFailure::clean(BlazeError::BackendError { + msg: "checkpoint restore is not supported by this backend".to_string(), + })) + } + /// Probe whether the configured backend executable is usable. async fn probe(&self, binary_path: &Path) -> Result; @@ -499,6 +576,66 @@ impl BackendSpawner for MockSpawner { .map_err(SpawnFailure::from) } + async fn restore_capability(&self, _binary_path: &Path) -> Result> { + Ok(Some(RestoreCapability { + backend: BackendKind::Mock, + version: Some("mock-v1".to_string()), + snapshot_kind: blaze_core::backend::SnapshotKind::Full, + })) + } + + async fn restore(&self, request: BackendRestoreRequest) -> RestoreResult { + let RestoreRequest { + instance_id, + snapshot_path, + mem_path, + checkpoint_backend, + expected_version, + snapshot_kind, + .. + } = request.request; + if checkpoint_backend != BackendKind::Mock + || expected_version.as_deref() != Some("mock-v1") + || snapshot_kind != blaze_core::backend::SnapshotKind::Full + { + return Err(SpawnFailure::clean(BlazeError::BackendError { + msg: "mock checkpoint identity is incompatible with the restore adapter" + .to_string(), + })); + } + let vmstate: serde_json::Value = match tokio::fs::read(&snapshot_path) + .await + .map_err(BlazeError::from) + .and_then(|bytes| { + serde_json::from_slice(&bytes).map_err(|error| BlazeError::BackendError { + msg: format!("decode mock VM state: {error}"), + }) + }) { + Ok(vmstate) => vmstate, + Err(error) => return Err(SpawnFailure::clean(error)), + }; + if vmstate.get("format").and_then(serde_json::Value::as_str) != Some("blaze-mock-v1") + || vmstate + .get("instance_id") + .and_then(serde_json::Value::as_str) + != Some(instance_id.to_string().as_str()) + || vmstate.get("kind").and_then(serde_json::Value::as_str) != Some("full") + { + return Err(SpawnFailure::clean(BlazeError::BackendError { + msg: "mock VM state does not match the requested sandbox".to_string(), + })); + } + match tokio::fs::read(&mem_path).await { + Ok(bytes) if bytes == b"blaze-mock-memory-v1" => spawn_mock_instance(instance_id) + .await + .map_err(SpawnFailure::from), + Ok(_) => Err(SpawnFailure::clean(BlazeError::BackendError { + msg: "mock checkpoint memory does not match the requested sandbox".to_string(), + })), + Err(error) => Err(SpawnFailure::clean(error.into())), + } + } + async fn probe(&self, _binary_path: &Path) -> Result { Ok(true) } @@ -1281,7 +1418,7 @@ mod tests { #[cfg(target_os = "linux")] use std::time::Duration; - use blaze_core::backend::{SnapshotKind, SnapshotRequest, SpawnRequest}; + use blaze_core::backend::{RestoreRequest, SnapshotKind, SnapshotRequest, SpawnRequest}; use blaze_core::policy::BackendConfigs; use blaze_core::storage::StorageSlot; @@ -1506,6 +1643,40 @@ mod tests { assert!(instance.snapshot(request).await.is_err()); } + #[tokio::test] + async fn checkpoint_restore_defaults_fail_closed_without_an_owner() { + let temp = tempfile::tempdir().expect("temp"); + let spawn = request(temp.path()); + let run_dir = spawn.run_dir.clone(); + let restore = RestoreRequest { + instance_id: spawn.instance_id, + binary_path: spawn.binary_path.clone(), + storage: spawn.storage.clone(), + snapshot_path: temp.path().join("vmstate.snap"), + mem_path: temp.path().join("memory.snap"), + checkpoint_backend: BackendKind::Bubblewrap, + expected_version: None, + snapshot_kind: SnapshotKind::Full, + expose_guest_socket: true, + }; + let restore = BackendRestoreRequest::new(restore, run_dir).expect("restore request"); + + assert!( + BubblewrapSpawner + .restore_capability(Path::new("")) + .await + .expect("capability") + .is_none() + ); + let failure = match BubblewrapSpawner.restore(restore).await { + Ok(_) => panic!("restore must remain unsupported"), + Err(failure) => failure, + }; + let (source, owner) = failure.into_parts(); + assert!(source.to_string().contains("restore is not supported")); + assert!(owner.is_none()); + } + #[tokio::test] async fn mock_instance_captures_self_contained_state() { let temp = tempfile::tempdir().expect("temp"); diff --git a/src/blaze/docs/design/lifecycle-state-consistency.md b/src/blaze/docs/design/lifecycle-state-consistency.md index b135e4f186..14f5199484 100644 --- a/src/blaze/docs/design/lifecycle-state-consistency.md +++ b/src/blaze/docs/design/lifecycle-state-consistency.md @@ -14,8 +14,10 @@ can clean non-terminal records that contain them. This document defines all three boundaries. The inventory-publication protocol does not change the HTTP API, configuration keys, or persisted JSON format. The management API section defines the sandbox namespace and the reserved -reusable-capacity boundary. The checkpoint section defines two sandbox routes -and the durable operation fields used to recover interrupted capture. +reusable-capacity boundary. The checkpoint section defines three sandbox routes — +capture, history, and restore — the durable operation fields used to recover +interrupted capture, and the restore journal and lifecycle contract that keep an +interrupted restore recoverable. ## Terms and owned objects @@ -131,16 +133,42 @@ persistence, or backend resume has an unknown or unsafe outcome, Blaze retains the durable operation and marks the sandbox `RecoveryRequired`. Startup does not restore a checkpoint or adopt an interrupted backend; normal reconciliation cleans the owned runtime and checkpoint transaction artifacts. Committed -checkpoint history is retained until sandbox destruction. Restore, deletion, -and pruning are outside this interface. +checkpoint history is retained until sandbox destruction. Deletion and pruning +are outside this interface. + +`POST /v1/sandboxes/{id}/rollback/{checkpoint_id}` replaces a running sandbox +from one verified full checkpoint. Before mutation, Blaze verifies the complete +checkpoint ancestry and artifacts, matches the policy, image, backend, version, +and snapshot kind, and requires explicit backend and storage restore +capabilities. Unsupported combinations return `501 Not Implemented` while the +current backend and lifecycle record remain unchanged. + +Restore uses this durable order: + +1. Persist restore intent and stage an independent root filesystem while the + current backend remains owned and running. +2. Stop the current backend, record `RestoreBackendStopped`, and enter + `Restoring` only after that boundary is durable. +3. Activate the staged root while retaining its predecessor, prepare backend + ownership, and start and validate the replacement owner. +4. Move checkpoint HEAD, commit replacement storage, return the lifecycle to + `Running`, and clear the restore journal. + +A failure before Blaze begins stopping the backend aborts staged storage and +preserves the running backend. Once Blaze starts stopping it, any later failure — +including the stop itself failing or shutdown not being confirmed — retains the +backend and storage ownership that can still be proven and commits +`RecoveryRequired`; destruction uses that journal to complete cleanup. Restore +changes catalog HEAD but does not rewrite the most recently completed capture +recorded by `last_checkpoint`. ## Management API and reusable-state boundary Lifecycle and guest operations are registered under `/v1/sandboxes`. Action-style reset and destroy paths are unregistered and return `404 Not Found`. Canonical destruction remains -`DELETE /v1/sandboxes/{id}`. Checkpoint capture uses the two routes defined in -the preceding section. +`DELETE /v1/sandboxes/{id}`. Checkpoint capture, listing, and restore use the +three routes defined in the preceding section. The following reserved management routes also return `501 Not Implemented` and do not manage reusable capacity: diff --git a/src/blaze/docs/design/lifecycle-state-consistency_zh.md b/src/blaze/docs/design/lifecycle-state-consistency_zh.md index 0647ef3728..cd4792e8c9 100644 --- a/src/blaze/docs/design/lifecycle-state-consistency_zh.md +++ b/src/blaze/docs/design/lifecycle-state-consistency_zh.md @@ -10,8 +10,9 @@ Blaze 有三个相互关联的生命周期边界。提供请求服务前，它 生命周期状态。 本设计定义这三个边界。清单发布流程不改变 HTTP API 或配置项。管理 API 章节定义 -沙箱命名空间以及预留的复用容量边界。检查点章节定义两个沙箱路由，以及恢复中断 -捕获所需的持久化操作字段。清单发布流程不改变持久化 JSON 格式。 +沙箱命名空间以及预留的复用容量边界。检查点章节定义三个沙箱路由——捕获、历史和 +恢复——以及恢复中断捕获所需的持久化操作字段，还有让中断的恢复保持可恢复的 +恢复日志与生命周期契约。清单发布流程不改变持久化 JSON 格式。 ## 概念与持有对象 @@ -104,14 +105,32 @@ Blaze 会保留被拒绝的 UUID 目录及其 `state.json`，供运维人员检 如果发布、HEAD 移动、生命周期持久化或后端恢复的结果未知或不安全，Blaze 会保留 持久化操作，并把 sandbox 标记为 `RecoveryRequired`。启动过程不会从检查点恢复， 也不会接管中断的后端；常规恢复会清理由记录持有的运行环境和检查点事务制品。 -已经提交的检查点历史会保留到 sandbox 销毁。该接口不提供恢复、删除或清理能力。 +已经提交的检查点历史会保留到 sandbox 销毁。该接口不提供删除或清理能力。 + +`POST /v1/sandboxes/{id}/rollback/{checkpoint_id}` 使用一个经过校验的完整检查点 +替换运行中的 sandbox。修改资源前，Blaze 会校验完整的检查点父链和制品，确认策略、 +镜像、后端、后端版本和快照类型一致，并要求后端与存储提供程序明确声明支持恢复。 +不受支持的组合会返回 `501 Not Implemented`，当前后端和生命周期记录保持不变。 + +恢复按照以下持久化顺序执行： + +1. 持久化恢复意图，并在当前后端仍被持有且正常运行时准备独立的根文件系统。 +2. 停止当前后端，持久化 `RestoreBackendStopped`，随后才进入 `Restoring`。 +3. 启用暂存根文件系统并保留原根文件系统，准备后端所有权，再启动并检查替代后端。 +4. 移动检查点 HEAD，提交替代存储，使生命周期返回 `Running`，并清除恢复日志。 + +Blaze 开始停止当前后端之前失败时，会撤销暂存存储并保留运行中的后端。一旦开始 +停止，此后任何失败——包括停止操作本身失败或无法确认后端已经停止——都会让 Blaze +保留仍可确认的后端和存储所有权，并持久化 `RecoveryRequired`；销毁操作根据 +该日志完成清理。恢复会改变目录 HEAD，但不会改写 `last_checkpoint` 记录的最近一次 +成功捕获。 ## 管理 API 与可复用状态边界 生命周期和客户机操作注册在 `/v1/sandboxes` 下。操作式重置和销毁路径不注册， 并返回 `404 Not Found`。 -规范的销毁入口仍是 `DELETE /v1/sandboxes/{id}`。检查点捕获使用上一节定义的 -两个路由。 +规范的销毁入口仍是 `DELETE /v1/sandboxes/{id}`。检查点捕获、查询和恢复使用上一节 +定义的三个路由。 以下保留的管理路由同样返回 `501 Not Implemented`，并且不会管理复用容量： diff --git a/src/blaze/docs/design/storage-artifact-synchronization.md b/src/blaze/docs/design/storage-artifact-synchronization.md index 00e356ea4b..f5dccd8913 100644 --- a/src/blaze/docs/design/storage-artifact-synchronization.md +++ b/src/blaze/docs/design/storage-artifact-synchronization.md @@ -83,8 +83,17 @@ after request cancellation and retain the sandbox operation lock until their outcome is known. A known pre-publication failure removes only the private stage. An uncertain publication never removes a path whose identity cannot be proven. Sandbox destruction removes transaction artifacts and committed -checkpoint history under the same state-root ownership boundary. Restore, -checkpoint deletion, and pruning are not part of this protocol. +checkpoint history under the same state-root ownership boundary. Checkpoint +deletion and pruning are not part of this protocol. + +Checkpoint restore is a separate opt-in provider capability. The file provider +copies the verified checkpoint root into a private stage while the live root +remains selected. Activation atomically selects the stage and retains the +predecessor. Abort restores the predecessor; commit durably records its intent, +then removes the predecessor and transaction journal. Every transition is +idempotently reconciled from the journal after restart. Unexpected links, +replacement paths, ambiguous layouts, or an unverified transaction identity +fail closed without deleting an object whose ownership cannot be proven. ## Capability boundary diff --git a/src/blaze/docs/design/storage-artifact-synchronization_zh.md b/src/blaze/docs/design/storage-artifact-synchronization_zh.md index dd5d84447d..2c137806fb 100644 --- a/src/blaze/docs/design/storage-artifact-synchronization_zh.md +++ b/src/blaze/docs/design/storage-artifact-synchronization_zh.md @@ -66,7 +66,13 @@ sweep，避免一个正在进行的 sandbox 操作阻止 worker 继续处理其 持有 sandbox 操作锁，直到结果确定。能够确认的发布前失败只删除私有暂存条目。 发布结果不确定时，不能删除身份无法确认的路径。销毁 sandbox 时，会在同一个 state-root 所有权边界内删除事务制品和已经提交的检查点历史。本协议不提供检查点 -恢复、删除或清理。 +删除或清理。 + +检查点恢复是存储提供程序的一项独立可选能力。文件存储提供程序会在当前根文件系统 +仍生效时，把经过校验的检查点根文件系统复制到私有暂存文件。启用操作会原子选择 +暂存文件并保留原文件；撤销操作恢复原文件；提交操作先持久化提交意图，再删除原文件 +和事务日志。进程重启后，每一步都能根据日志幂等完成恢复。发现意外链接、替换路径、 +布局含糊或事务身份不匹配时，操作会安全失败，不会删除无法证明归属的对象。 ## 能力边界