Validate, rename, and advance Raft state after the last chunk. Returns the `InstallSnapshotResponse` produced by `MultiRaft::handle_install_snapshot` so callers can propagate the Raft term back to the leader.
(
state: PartialSnapshotState,
multi_raft: &Arc<Mutex<MultiRaft>>,
)
| 37 | /// `MultiRaft::handle_install_snapshot` so callers can propagate the |
| 38 | /// Raft term back to the leader. |
| 39 | pub async fn commit( |
| 40 | state: PartialSnapshotState, |
| 41 | multi_raft: &Arc<Mutex<MultiRaft>>, |
| 42 | ) -> Result<InstallSnapshotResponse, ClusterError> { |
| 43 | let group_id = state.group_id; |
| 44 | let partial_path = state.partial_path.clone(); |
| 45 | let expected_crc = state.running_crc; |
| 46 | |
| 47 | // Flush and close the partial file before reading it back. |
| 48 | // `state.partial_file` may be `None` if the snapshot had zero bytes |
| 49 | // (bootstrap stub). In that case skip the I/O validation. |
| 50 | if let Some(file) = state.partial_file { |
| 51 | tokio::task::spawn_blocking(move || -> std::io::Result<()> { file.sync_all() }) |
| 52 | .await |
| 53 | .map_err(|e| ClusterError::PartialSnapshotCorrupt { |
| 54 | group_id, |
| 55 | detail: format!("spawn_blocking join error on sync: {e}"), |
| 56 | })? |
| 57 | .map_err(|e| ClusterError::Storage { |
| 58 | detail: format!("sync partial file for group {group_id}: {e}"), |
| 59 | })?; |
| 60 | } |
| 61 | |
| 62 | // CRC validation: re-read the file and compare against running CRC. |
| 63 | // If the file is empty (bootstrap stub), skip. |
| 64 | let file_bytes = tokio::task::spawn_blocking({ |
| 65 | let path = partial_path.clone(); |
| 66 | move || std::fs::read(&path) |
| 67 | }) |
| 68 | .await |
| 69 | .map_err(|e| ClusterError::PartialSnapshotCorrupt { |
| 70 | group_id, |
| 71 | detail: format!("spawn_blocking join error on read: {e}"), |
| 72 | })? |
| 73 | .map_err(|e| ClusterError::Storage { |
| 74 | detail: format!("read partial file for group {group_id}: {e}"), |
| 75 | })?; |
| 76 | |
| 77 | if !file_bytes.is_empty() { |
| 78 | let computed = crc32c::crc32c(&file_bytes); |
| 79 | if computed != expected_crc { |
| 80 | return Err(ClusterError::SnapshotCrcMismatch { |
| 81 | group_id, |
| 82 | stored: expected_crc, |
| 83 | computed, |
| 84 | }); |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | // Atomic rename: .partial → .snap |
| 89 | let snap_path = snap_path_for(&partial_path); |
| 90 | tokio::task::spawn_blocking({ |
| 91 | let from = partial_path.clone(); |
| 92 | let to = snap_path.clone(); |
| 93 | move || std::fs::rename(&from, &to) |
| 94 | }) |
| 95 | .await |
| 96 | .map_err(|e| ClusterError::PartialSnapshotCorrupt { |
no test coverage detected