1/*
2 * SPDX-FileCopyrightText: 2026 Copyright (c) Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 */
13
14//! Startup recovery for interrupted transactions.
15//!
16//! When [`LocalStorage`](crate::LocalStorage) is created, it calls [`recover`] to check for
17//! leftover WAL, staging, and `.bak` files from a previous run.
18//!
19//! Recovery decisions are driven by the WAL file header status:
20//!
21//! - **No WAL file (or corrupt/unreadable header):** Either no transaction was in progress, or
22//! a commit completed successfully but cleanup was interrupted. Any orphaned `.bak` files are
23//! deleted and staging is cleaned.
24//! - **WAL with `RECORDING` status:** The transaction was still recording operations. Nothing
25//! was applied to collections. Safe to discard the WAL and staging files.
26//! - **WAL with `COMMITTING` status:** The commit phase started but did not complete (the WAL
27//! was not yet deleted). Operations may have been partially applied. Recovery reads the WAL
28//! entries, removes newly created artifacts, and restores `.bak` files.
29
30use std::path::Path;
31
32use cda_interfaces::storage_api::{Operation, StorageError};
33
34use crate::wal::{self, WalStatus};
35
36/// File extension used for backup files created during the commit phase.
37pub(crate) const BACKUP_EXTENSION: &str = "bak";
38
39/// File extension used for staging (temporary) files created during write operations.
40pub(crate) const STAGING_EXTENSION: &str = "tmp";
41
42/// Perform startup recovery on the given journal and collections directories.
43///
44/// This function is idempotent. Calling it multiple times on a clean state is a no-op.
45///
46/// # Errors
47///
48/// Returns a [`StorageError`] if recovery cannot complete (e.g., filesystem is read-only or
49/// corrupted beyond repair).
[docs] 50/// [[ dimpl~storage-crash-recovery, Startup recovery of interrupted transactions from the WAL, dimpl ]]
51pub(crate) fn recover(journal_dir: &Path, collections_dir: &Path) -> Result<(), StorageError> {
52 let wal_path = journal_dir.join(wal::WAL_FILE_NAME);
53 let staging_dir = journal_dir.join(wal::STAGING_DIR_NAME);
54
55 let status = wal::open_wal(&wal_path)?;
56
57 match status {
58 Some((WalStatus::Committing, wal_data)) => {
59 // Commit was in progress. Operations may have been partially applied.
60 tracing::info!("WAL indicates commit was in progress -- rolling back");
61
62 // Read the operations from the WAL so we know what to undo.
63 let wal_result = wal::read_wal(&wal_data)?;
64
65 if wal_result.truncated {
66 // Check if any operations were actually applied to the filesystem
67 // to determine if it was a crash during recording or a corruption of the fs.
68 let has_backups = has_backup_files(collections_dir);
69 let has_applied_artifacts =
70 has_new_artifacts(collections_dir, &wal_result.operations);
71
72 if has_backups || has_applied_artifacts {
73 // Operations were partially applied but WAL is corrupted.
74 // Unable to guarantee a consistent rollback.
75 return Err(StorageError::Corruption(
76 "WAL is truncated and operations were partially applied, indicating \
77 possible filesystem corruption."
78 .to_string(),
79 ));
80 }
81 // No evidence of the transaction being actually applied.
82 // -> Safe to discard the WAL.
83 tracing::info!(
84 "WAL is truncated but no operations were applied. Discarding safely"
85 );
86 } else {
87 // Full WAL available -- perform complete rollback.
88 // Remove newly created artifacts first. This must happen before
89 // rollback_partial_commit as restoring .bak files could make
90 // previously-overwritten files look like new artifacts.
91 remove_new_artifacts(collections_dir, &wal_result.operations)?;
92
93 // Restore .bak files (undoes overwrites and deletes).
94 rollback_partial_commit(collections_dir)?;
95 }
96 }
97 Some((WalStatus::Recording, _wal_data)) => {
98 // Transaction was still recording. Nothing was applied.
99 tracing::info!("WAL indicates transaction was still recording -- discarding");
100 }
101 None => {
102 // No valid WAL. Either clean state or commit completed with interrupted cleanup.
103 // Delete any orphaned .bak files (leftover from a completed commit whose cleanup
104 // was interrupted between WAL deletion and .bak removal).
105 if has_backup_files(collections_dir) {
106 tracing::info!("No WAL but found orphaned .bak files -- cleaning up");
107 delete_all_backup_files(collections_dir)?;
108 }
109 }
110 }
111
112 // Always clean up WAL and staging files.
113 if wal_path.exists() {
114 std::fs::remove_file(&wal_path)?;
115 }
116 cleanup_staging_dir(&staging_dir)?;
117
118 tracing::info!("Recovery complete");
119 Ok(())
120}
121
122/// Roll back a partially applied commit by restoring `.bak` files.
123///
124/// This is also called by the commit path in `local_storage.rs` when `apply_operations` fails
125/// mid-commit.
126///
127/// For each `.bak` file found under `collections_dir`:
128/// 1. If the corresponding non-`.bak` file exists (i.e., the new file was already renamed into
129/// place), remove it.
130/// 2. Rename the `.bak` file back to its original name.
131pub(crate) fn rollback_partial_commit(collections_dir: &Path) -> Result<(), StorageError> {
132 if !collections_dir.exists() {
133 return Ok(());
134 }
135
136 for collection_entry in std::fs::read_dir(collections_dir)? {
137 let collection_entry = collection_entry?;
138 let collection_path = collection_entry.path();
139 if !collection_path.is_dir() {
140 continue;
141 }
142
143 // Check for .bak files that are actually directories (collection-level backups from
144 // delete_collection).
145 let bak_ext = std::ffi::OsStr::new(BACKUP_EXTENSION);
146 if collection_path.extension() == Some(bak_ext) {
147 let original = collection_path.with_extension("");
148 if original.exists() {
149 std::fs::remove_dir_all(&original)?;
150 }
151 std::fs::rename(&collection_path, &original)?;
152 tracing::info!(
153 path = %original.display(),
154 "Restored collection directory from backup"
155 );
156 continue;
157 }
158
159 restore_bak_files_in_dir(&collection_path)?;
160 }
161
162 Ok(())
163}
164
165/// Remove files and directories that were newly created by a partially applied commit.
166///
167/// This handles the case where operations created new artifacts (write to new key,
168/// `CreateCollection`, `CopyCollection`) that have no `.bak` counterpart and would otherwise
169/// remain after `.bak` restoration.
170pub(crate) fn remove_new_artifacts(
171 collections_dir: &Path,
172 operations: &[Operation],
173) -> Result<(), StorageError> {
174 for op in operations.iter().rev() {
175 match op {
176 Operation::Write {
177 collection, key, ..
178 } => {
179 let target = collections_dir.join(collection.as_str()).join(key);
180 let backup = append_bak_extension(&target);
181 // If no .bak exists, this was a new file (not an overwrite). Remove it.
182 if target.exists() && !backup.exists() {
183 std::fs::remove_file(&target)?;
184 tracing::debug!(path = %target.display(), "Removed newly created file");
185 }
186 }
187 Operation::CreateCollection { name } => {
188 let dir = collections_dir.join(name.as_str());
189 // Only remove if it is empty (its contents, if any, are handled by other ops).
190 if dir.exists() && is_dir_empty(&dir)? {
191 std::fs::remove_dir(&dir)?;
192 tracing::debug!(path = %dir.display(), "Removed newly created collection dir");
193 }
194 }
195 Operation::CopyCollection {
196 dest, dest_existed, ..
197 } => {
198 let dest_dir = collections_dir.join(dest.as_str());
199 let backup = append_bak_extension(&dest_dir);
200 // Existing destinations are restored from their .bak file. Only remove a
201 // destination that was created by this transaction and has no backup.
202 if !dest_existed && dest_dir.exists() && !backup.exists() {
203 std::fs::remove_dir_all(&dest_dir)?;
204 tracing::debug!(
205 path = %dest_dir.display(),
206 "Removed newly created copy-collection dir"
207 );
208 }
209 }
210 // Delete and DeleteAll only create .bak files -- handled by rollback_partial_commit.
211 Operation::Delete { .. }
212 | Operation::DeleteAll { .. }
213 | Operation::DeleteCollection { .. } => {}
214 }
215 }
216 Ok(())
217}
218
219/// Check whether any `.bak` files or directories exist under `collections_dir`.
220fn has_backup_files(collections_dir: &Path) -> bool {
221 if !collections_dir.exists() {
222 return false;
223 }
224
225 let bak_ext = std::ffi::OsStr::new(BACKUP_EXTENSION);
226
227 let Ok(entries) = std::fs::read_dir(collections_dir) else {
228 return false;
229 };
230
231 for entry in entries {
232 let Ok(entry) = entry else {
233 continue;
234 };
235 let path = entry.path();
236
237 // Collection-level .bak directory.
238 if path.extension() == Some(bak_ext) {
239 return true;
240 }
241
242 // Check inside each collection directory for .bak files.
243 if path.is_dir() {
244 let Ok(inner_entries) = std::fs::read_dir(&path) else {
245 continue;
246 };
247 for inner in inner_entries {
248 let Ok(inner) = inner else {
249 continue;
250 };
251 if inner.path().extension() == Some(bak_ext) {
252 return true;
253 }
254 }
255 }
256 }
257
258 false
259}
260
261/// Check whether any operations from the WAL appear to have been applied to the filesystem.
262///
263/// This looks for artifacts that would have been created by the given operations (new files,
264/// new collection directories) that exist on disk without a corresponding `.bak` backup.
265fn has_new_artifacts(collections_dir: &Path, operations: &[Operation]) -> bool {
266 for op in operations {
267 match op {
268 Operation::Write {
269 collection, key, ..
270 } => {
271 let target = collections_dir.join(collection.as_str()).join(key);
272 let backup = append_bak_extension(&target);
273 // A file exists without a .bak -> it was newly created by this operation.
274 if target.exists() && !backup.exists() {
275 return true;
276 }
277 }
278 Operation::CreateCollection { name } => {
279 let dir = collections_dir.join(name.as_str());
280 if dir.exists() {
281 return true;
282 }
283 }
284 Operation::CopyCollection {
285 dest, dest_existed, ..
286 } => {
287 let dest_dir = collections_dir.join(dest.as_str());
288 let backup = append_bak_extension(&dest_dir);
289 if !dest_existed && dest_dir.exists() && !backup.exists() {
290 return true;
291 }
292 }
293 Operation::Delete { .. }
294 | Operation::DeleteAll { .. }
295 | Operation::DeleteCollection { .. } => {}
296 }
297 }
298 false
299}
300
301/// Delete all `.bak` files and directories under `collections_dir`.
302///
303/// Used when no WAL is present but orphaned `.bak` files remain from a successfully committed
304/// transaction whose post-commit cleanup was interrupted.
305fn delete_all_backup_files(collections_dir: &Path) -> Result<(), StorageError> {
306 let bak_ext = std::ffi::OsStr::new(BACKUP_EXTENSION);
307
308 for collection_entry in std::fs::read_dir(collections_dir)? {
309 let collection_entry = collection_entry?;
310 let path = collection_entry.path();
311
312 // Collection-level .bak directory.
313 if path.is_dir() && path.extension() == Some(bak_ext) {
314 std::fs::remove_dir_all(&path)?;
315 tracing::debug!(path = %path.display(), "Removed orphaned .bak directory");
316 continue;
317 }
318
319 // File-level .bak files inside collection directories.
320 if path.is_dir() {
321 for entry in std::fs::read_dir(&path)? {
322 let entry = entry?;
323 let file_path = entry.path();
324 if file_path.extension() == Some(bak_ext) {
325 std::fs::remove_file(&file_path)?;
326 tracing::debug!(path = %file_path.display(), "Removed orphaned .bak file");
327 }
328 }
329 }
330 }
331
332 Ok(())
333}
334
335/// Restore all `.bak` files in a single directory to their original names.
336fn restore_bak_files_in_dir(dir: &Path) -> Result<(), StorageError> {
337 let bak_ext = std::ffi::OsStr::new(BACKUP_EXTENSION);
338
339 for entry in std::fs::read_dir(dir)? {
340 let entry = entry?;
341 let path = entry.path();
342
343 if path.extension() == Some(bak_ext) {
344 let original = path.with_extension("");
345 // Remove the partially-applied new file if it exists.
346 if original.exists() {
347 std::fs::remove_file(&original)?;
348 }
349 std::fs::rename(&path, &original)?;
350 tracing::info!(path = %original.display(), "Restored file from backup");
351 }
352
353 // Also clean up any orphaned .tmp files (staging files that were renamed into the
354 // collection directory during commit but shouldn't be there).
355 let tmp_ext = std::ffi::OsStr::new(STAGING_EXTENSION);
356 if path.extension() == Some(tmp_ext) {
357 std::fs::remove_file(&path)?;
358 tracing::warn!(path = %path.display(), "Removed orphaned staging file");
359 }
360 }
361
362 Ok(())
363}
364
365/// Remove all files in the staging directory.
366fn cleanup_staging_dir(staging_dir: &Path) -> Result<(), StorageError> {
367 if !staging_dir.exists() {
368 return Ok(());
369 }
370
371 for entry in std::fs::read_dir(staging_dir)? {
372 let entry = entry?;
373 let path = entry.path();
374 if path.is_file() {
375 std::fs::remove_file(&path)?;
376 }
377 }
378
379 Ok(())
380}
381
382/// Append `.bak` to a path.
383fn append_bak_extension(path: &Path) -> std::path::PathBuf {
384 let mut s = path.as_os_str().to_owned();
385 s.push(".");
386 s.push(BACKUP_EXTENSION);
387 std::path::PathBuf::from(s)
388}
389
390/// Check if a directory is empty.
391fn is_dir_empty(dir: &Path) -> Result<bool, StorageError> {
392 let mut entries = std::fs::read_dir(dir)?;
393 Ok(entries.next().is_none())
394}
Back to top