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//! Local filesystem implementation of the [`Storage`] trait.
15
16use std::{
17 path::{Path, PathBuf},
18 sync::{
19 Arc,
20 atomic::{AtomicBool, Ordering},
21 },
22};
23
24use async_trait::async_trait;
25use cda_interfaces::storage_api::{
26 CollectionName, Operation, Storage, StorageError, Transaction, TransactionCommitter,
27};
28
29use crate::{
30 local_collection::LocalCollection,
31 paths,
32 recovery::{self, BACKUP_EXTENSION},
33 wal,
34};
35
36/// Local filesystem implementation of the [`Storage`] trait.
37///
38/// Collections are stored as subdirectories under `{root}/collections/`. The WAL and staging
39/// files live under `{root}/journal/`.
[docs] 40/// [[ dimpl~storage-local-filesystem-implementation, Local filesystem implementation of the Storage Access API, dimpl ]]
41pub struct LocalStorage {
42 /// Base directory for collection data.
43 collections_dir: PathBuf,
44 /// Directory containing the WAL file and staging subdirectory.
45 journal_dir: PathBuf,
46 /// Read-write lock: reads take shared, commit takes exclusive.
47 data_lock: Arc<tokio::sync::RwLock<()>>,
48 /// Flag to enforce single-transaction-at-a-time.
49 tx_active: Arc<AtomicBool>,
50}
51
52impl LocalStorage {
53 /// Create a new `LocalStorage` rooted at the given directory.
54 ///
55 /// On construction, performs startup recovery to handle any incomplete transactions from a
56 /// previous run.
57 ///
58 /// # Errors
59 ///
60 /// Returns a [`StorageError`] if directory creation fails or recovery fails.
61 pub fn new(root: impl Into<PathBuf>) -> Result<Self, StorageError> {
62 let root = root.into();
63 let collections_dir = root.join("collections");
64 let journal_dir = root.join("journal");
65 let staging_dir = journal_dir.join(wal::STAGING_DIR_NAME);
66
67 // Ensure directories exist.
68 std::fs::create_dir_all(&collections_dir)?;
69 std::fs::create_dir_all(&staging_dir)?;
70
71 // Run startup recovery before accepting any operations.
72 recovery::recover(&journal_dir, &collections_dir)?;
73
74 tracing::info!(root = %root.display(), "Local storage initialized");
75
76 Ok(Self {
77 collections_dir,
78 journal_dir,
79 data_lock: Arc::new(tokio::sync::RwLock::new(())),
80 tx_active: Arc::new(AtomicBool::new(false)),
81 })
82 }
83
84 /// Return the path to a collection's directory.
85 ///
86 /// # Errors
87 ///
88 /// Returns a [`StorageError`] if the collection name fails path sanitization (when the
89 /// `sanitize-paths` feature is enabled).
90 fn collection_dir(&self, name: &CollectionName) -> Result<PathBuf, StorageError> {
91 paths::sanitize_path_segment(name.as_str())?;
92 Ok(self.collections_dir.join(name.as_str()))
93 }
94}
95
96impl Storage for LocalStorage {
97 type CollectionHandle = LocalCollection;
98
99 async fn get_collection(
100 &self,
101 name: &CollectionName,
102 ) -> Result<Arc<LocalCollection>, StorageError> {
103 let _guard = self.data_lock.read().await;
104 let dir = self.collection_dir(name)?;
105 if !dir.exists() {
106 return Err(StorageError::CollectionNotFound(name.to_string()));
107 }
108 Ok(Arc::new(LocalCollection::new(
109 name.clone(),
110 dir,
111 Arc::clone(&self.data_lock),
112 )))
113 }
114
115 async fn get_or_create_collection(
116 &self,
117 name: &CollectionName,
118 ) -> Result<Arc<LocalCollection>, StorageError> {
119 let dir = self.collection_dir(name)?;
120
121 let read_guard = self.data_lock.read().await;
122 if dir.exists() {
123 return Ok(Arc::new(LocalCollection::new(
124 name.clone(),
125 dir,
126 Arc::clone(&self.data_lock),
127 )));
128 }
129 drop(read_guard);
130
131 // Collection does not exist -- acquire a write lock to create it.
132 // Reject if a transaction is active.
133 if self.tx_active.load(Ordering::Acquire) {
134 return Err(StorageError::TransactionBusy);
135 }
136 let write_guard = self.data_lock.write().await;
137 // Re-check after acquiring write lock
138 if !dir.exists() {
139 std::fs::create_dir_all(&dir)?;
140 tracing::debug!(collection = %name, "Created collection directory");
141 }
142 drop(write_guard);
143
144 Ok(Arc::new(LocalCollection::new(
145 name.clone(),
146 dir,
147 Arc::clone(&self.data_lock),
148 )))
149 }
150
151 fn begin_transaction(&self) -> Result<Transaction, StorageError> {
152 // Enforce single-transaction-at-a-time.
153 let was_active = self.tx_active.swap(true, Ordering::AcqRel);
154 if was_active {
155 return Err(StorageError::TransactionBusy);
156 }
157
158 let wal_path = self.journal_dir.join(wal::WAL_FILE_NAME);
159 let staging_dir = self.journal_dir.join(wal::STAGING_DIR_NAME);
160
161 // Create a fresh WAL file.
162 wal::create_wal(&wal_path).inspect_err(|_| {
163 // Reset flag if WAL creation fails.
164 self.tx_active.store(false, Ordering::Release);
165 })?;
166
167 // Ensure staging directory exists.
168 std::fs::create_dir_all(&staging_dir).map_err(|e| {
169 self.tx_active.store(false, Ordering::Release);
170 StorageError::Io(e)
171 })?;
172
173 tracing::debug!("Transaction started");
174
175 Ok(Transaction::new(
176 wal_path,
177 staging_dir,
178 Arc::new(LocalStorageCommitter {
179 collections_dir: self.collections_dir.clone(),
180 data_lock: Arc::clone(&self.data_lock),
181 tx_active: Arc::clone(&self.tx_active),
182 }),
183 ))
184 }
185
186 fn create_collection(
187 &self,
188 tx: &mut Transaction,
189 name: &CollectionName,
190 ) -> impl std::future::Future<Output = Result<Arc<LocalCollection>, StorageError>> + Send {
191 let result = (|| {
192 let dir = self.collection_dir(name)?;
193 if dir.exists() {
194 return Err(StorageError::TransactionConflict(format!(
195 "Collection already exists: {name}"
196 )));
197 }
198
199 let op = Operation::CreateCollection { name: name.clone() };
200 wal::append_operation(tx.journal_path(), &op)?;
201 tx.record(op);
202
203 // Return a collection handle that points to where the directory *will* be after commit.
204 Ok(Arc::new(LocalCollection::new(
205 name.clone(),
206 dir,
207 Arc::clone(&self.data_lock),
208 )))
209 })();
210 std::future::ready(result)
211 }
212
213 fn delete_collection(
214 &self,
215 tx: &mut Transaction,
216 name: &CollectionName,
217 ) -> impl std::future::Future<Output = Result<(), StorageError>> + Send {
218 let result = (|| {
219 let dir = self.collection_dir(name)?;
220 if !dir.exists() {
221 return Err(StorageError::CollectionNotFound(name.to_string()));
222 }
223
224 let op = Operation::DeleteCollection { name: name.clone() };
225 wal::append_operation(tx.journal_path(), &op)?;
226 tx.record(op);
227
228 Ok(())
229 })();
230 std::future::ready(result)
231 }
232
233 fn copy_collection(
234 &self,
235 tx: &mut Transaction,
236 source: &CollectionName,
237 dest: &CollectionName,
238 ) -> impl std::future::Future<Output = Result<(), StorageError>> + Send {
239 let result = (|| {
240 let source_dir = self.collection_dir(source)?;
241 if !source_dir.exists() {
242 return Err(StorageError::CollectionNotFound(source.to_string()));
243 }
244
245 let op = Operation::CopyCollection {
246 source: source.clone(),
247 dest: dest.clone(),
248 dest_existed: self.collection_dir(dest)?.exists(),
249 };
250 wal::append_operation(tx.journal_path(), &op)?;
251 tx.record(op);
252
253 Ok(())
254 })();
255 std::future::ready(result)
256 }
257}
258
259/// Handles commit and rollback callbacks from [`Transaction`].
260///
261/// This struct is held by the `Transaction` via `Arc<dyn TransactionCommitter>` and provides the
262/// bridge back into the storage backend for applying or discarding operations.
263struct LocalStorageCommitter {
264 collections_dir: PathBuf,
265 data_lock: Arc<tokio::sync::RwLock<()>>,
266 tx_active: Arc<AtomicBool>,
267}
268
269#[async_trait]
270impl TransactionCommitter for LocalStorageCommitter {
[docs]271 /// [[ dimpl~storage-atomic-commit, Atomic commit of a staged transaction via WAL and backup-rename, dimpl ]]
272 async fn apply(
273 &self,
274 operations: Vec<Operation>,
275 journal_path: PathBuf,
276 ) -> Result<(), StorageError> {
277 async fn apply_inner(
278 committer: &LocalStorageCommitter,
279 operations: Vec<Operation>,
280 journal_path: PathBuf,
281 ) -> Result<(), StorageError> {
282 let staging_dir = journal_path.parent().map(|p| p.join(wal::STAGING_DIR_NAME));
283 let journal_dir = journal_path
284 .parent()
285 .ok_or_else(|| StorageError::Other("WAL path has no parent directory".to_string()))?
286 .to_path_buf();
287
288 // Step 1: Mark the WAL as COMMITTING (in-place pwrite, no fsync yet).
289 wal::mark_committing(&journal_path)?;
290
291 // Step 2: Fsync all staging files and the WAL in one batch.
292 // This single durability barrier makes both the COMMITTING status and all operation
293 // entries durable. If the process crashes after this point, recovery will see
294 // COMMITTING and know that operations may have been partially applied.
295 if let Some(ref staging) = staging_dir {
296 wal::sync_staging_files(staging)?;
297 }
298 wal::sync_wal(&journal_path)?;
299
300 // Step 3: Acquire exclusive access. Blocks all reads.
301 let _write_guard = committer.data_lock.write().await;
302
303 // Step 4: Apply each operation, creating backups for rollback.
304 let result = committer.apply_operations(&operations);
305
306 if let Err(e) = &result {
307 tracing::warn!(
308 error = %e,
309 "Commit failed, rolling back partially applied operations"
310 );
311 if let Err(rb_err) =
312 rollback_applied_operations(&committer.collections_dir, &operations)
313 {
314 tracing::warn!(error = %rb_err, "Rollback also encountered an error");
315 }
316 } else {
317 // Step 5: Fsync collection directories to make renames durable.
318 sync_collection_dirs(&committer.collections_dir, &operations)?;
319 }
320
321 // Step 6: Delete the WAL file and fsync the journal directory.
322 // This is the point of no return. Once the WAL deletion is durable, recovery will
323 // treat any remaining .bak files as orphans from a successfully committed transaction.
324 if journal_path.exists() {
325 std::fs::remove_file(&journal_path)?;
326 }
327 wal::sync_journal_dir(&journal_dir)?;
328
329 // Step 7: Clean up .bak files and staging (non-critical, best-effort after point of
330 // no return).
331 if result.is_ok()
332 && let Err(e) = cleanup_backup_files(&committer.collections_dir)
333 {
334 tracing::warn!(error = %e, "Failed to clean up .bak files after commit");
335 }
336 if let Some(ref staging) = staging_dir
337 && let Err(e) = cleanup_directory_contents(staging)
338 {
339 tracing::warn!(error = %e, "Failed to clean up staging directory after commit");
340 }
341
342 result
343 }
344 let result = apply_inner(self, operations, journal_path).await;
345
346 // Release the transaction flag.
347 self.tx_active.store(false, Ordering::Release);
348
349 if result.is_ok() {
350 tracing::info!("Transaction committed successfully");
351 }
352 result
353 }
354
355 fn discard(&self, staging_dir: &Path, journal_path: &Path) {
356 // Best-effort cleanup on rollback.
357 if let Err(e) = cleanup_directory_contents(staging_dir) {
358 tracing::warn!(error = %e, "Failed to clean up staging directory during rollback");
359 }
360 if journal_path.exists()
361 && let Err(e) = std::fs::remove_file(journal_path)
362 {
363 tracing::warn!(error = %e, "Failed to remove WAL file during rollback");
364 }
365 self.tx_active.store(false, Ordering::Release);
366 tracing::debug!("Transaction rolled back");
367 }
368}
369
370impl LocalStorageCommitter {
371 /// Apply all operations from the journal.
372 ///
373 /// The Delete* Operations ignore the case that the file or directory was already deleted
374 /// on the fs due to external modifications.
375 /// In these cases, it's better to continue with the commit instead of failing it,
376 /// as the end result is the same (the file/directory is gone).
377 fn apply_operations(&self, operations: &[Operation]) -> Result<(), StorageError> {
378 operations
379 .iter()
380 .try_for_each(|op| -> Result<(), StorageError> {
381 match op {
382 Operation::Write {
383 collection,
384 key,
385 staged_path,
386 } => {
387 let target_dir = self.collections_dir.join(collection.as_str());
388 std::fs::create_dir_all(&target_dir)?;
389 let target = target_dir.join(key);
390
391 // Backup existing file if present.
392 if target.exists() {
393 let backup = append_extension(&target, BACKUP_EXTENSION);
394 std::fs::rename(&target, &backup)?;
395 }
396
397 // Move staged file into place.
398 std::fs::rename(staged_path, &target)?;
399 }
400 Operation::Delete { collection, key } => {
401 let target = self.collections_dir.join(collection.as_str()).join(key);
402 if target.exists() {
403 let backup = append_extension(&target, BACKUP_EXTENSION);
404 std::fs::rename(&target, &backup)?;
405 }
406 }
407 Operation::DeleteAll { collection } => {
408 let dir = self.collections_dir.join(collection.as_str());
409 if dir.exists() {
410 backup_all_files_in_dir(&dir)?;
411 }
412 }
413 Operation::CreateCollection { name } => {
414 let dir = self.collections_dir.join(name.as_str());
415 std::fs::create_dir_all(&dir)?;
416 }
417 Operation::DeleteCollection { name } => {
418 let dir = self.collections_dir.join(name.as_str());
419 if dir.exists() {
420 let backup = append_extension(&dir, BACKUP_EXTENSION);
421 std::fs::rename(&dir, &backup)?;
422 }
423 }
424 Operation::CopyCollection { source, dest, .. } => {
425 let source_dir = self.collections_dir.join(source.as_str());
426 let dest_dir = self.collections_dir.join(dest.as_str());
427
428 // Back up an existing destination before creating the replacement.
429 if dest_dir.exists() {
430 let backup = append_extension(&dest_dir, BACKUP_EXTENSION);
431 std::fs::rename(&dest_dir, &backup)?;
432 }
433
434 // Create a fresh directory to provide replace rather than merge semantics.
435 std::fs::create_dir_all(&dest_dir)?;
436 copy_dir_contents(&source_dir, &dest_dir)?;
437 }
438 }
439 Ok(())
440 })
441 }
442}
443
444/// Append an extension to a path (e.g., `foo.txt` -> `foo.txt.bak`).
445fn append_extension(path: &Path, ext: &str) -> PathBuf {
446 let mut s = path.as_os_str().to_owned();
447 s.push(".");
448 s.push(ext);
449 PathBuf::from(s)
450}
451
452/// Rename every file in a directory to have a `.bak` extension.
453fn backup_all_files_in_dir(dir: &Path) -> Result<(), StorageError> {
454 for entry in std::fs::read_dir(dir)? {
455 let entry = entry?;
456 let path = entry.path();
457 if path.is_file() {
458 let backup = append_extension(&path, BACKUP_EXTENSION);
459 std::fs::rename(&path, &backup)?;
460 }
461 }
462 Ok(())
463}
464
465/// Copy all files from `source` to `dest`.
466fn copy_dir_contents(source: &Path, dest: &Path) -> Result<(), StorageError> {
467 for entry in std::fs::read_dir(source)? {
468 let entry = entry?;
469 let path = entry.path();
470 if path.is_file()
471 && let Some(name) = path.file_name()
472 {
473 let target = dest.join(name);
474 std::fs::copy(&path, &target)?;
475 }
476 }
477 Ok(())
478}
479
480/// Remove all `.bak` files and directories under `collections_dir`.
481fn cleanup_backup_files(collections_dir: &Path) -> Result<(), StorageError> {
482 let bak_ext = std::ffi::OsStr::new(BACKUP_EXTENSION);
483
484 for collection_entry in std::fs::read_dir(collections_dir)? {
485 let collection_entry = collection_entry?;
486 let path = collection_entry.path();
487
488 // Collection-level backups (directories with .bak extension).
489 if path.is_dir() && path.extension() == Some(bak_ext) {
490 std::fs::remove_dir_all(&path)?;
491 continue;
492 }
493
494 if path.is_dir() {
495 for entry in std::fs::read_dir(&path)? {
496 let entry = entry?;
497 let file_path = entry.path();
498 if file_path.extension() == Some(bak_ext) {
499 std::fs::remove_file(&file_path)?;
500 }
501 }
502 }
503 }
504
505 Ok(())
506}
507
508/// Roll back partially applied operations during a failed commit.
509///
510/// This removes newly created files and directories (for writes to new keys,
511/// `CreateCollection`, `CopyCollection`) and then restores `.bak` files (for
512/// overwrites/deletes). The order matters: new artifacts must be identified while `.bak` files
513/// are still in place (before restoration changes the filesystem state).
514fn rollback_applied_operations(
515 collections_dir: &Path,
516 operations: &[Operation],
517) -> Result<(), StorageError> {
518 recovery::remove_new_artifacts(collections_dir, operations)?;
519
520 recovery::rollback_partial_commit(collections_dir)?;
521
522 Ok(())
523}
524
525/// Fsync the directories that were modified during commit to ensure renames are durable.
526///
527/// On POSIX systems, `rename()` is atomic but not durable until the *directory* containing the
528/// entry is fsynced. This function fsyncs each unique collection directory that was touched by
529/// the operations.
530///
531/// On Windows this is a no-op because NTFS journals metadata changes internally.
532// disable the lint on windows to keep signature consistent
533#[cfg_attr(windows, allow(clippy::unnecessary_wraps))]
534fn sync_collection_dirs(
535 collections_dir: &Path,
536 operations: &[Operation],
537) -> Result<(), StorageError> {
538 // On Windows, NTFS journals metadata changes internally so explicit directory fsync is
539 // not needed.
540 // Additionally, `File::open` on a directory fails due to `FILE_FLAG_BACKUP_SEMANTICS`
541 // not being set by the standard library.
542 // The early return avoids both the unnecessary work and the platform-specific fsync
543 // implementation.
544 #[cfg(windows)]
545 {
546 // Suppress unused variable warnings.
547 let _ = &collections_dir;
548 let _ = &operations;
549 }
550 #[cfg(not(windows))]
551 {
552 let mut synced = std::collections::HashSet::new();
553
554 for op in operations {
555 let dir_name = match op {
556 Operation::Write { collection, .. }
557 | Operation::Delete { collection, .. }
558 | Operation::DeleteAll { collection } => Some(collection.as_str()),
559 Operation::CreateCollection { name } | Operation::DeleteCollection { name } => {
560 Some(name.as_str())
561 }
562 Operation::CopyCollection { dest, .. } => Some(dest.as_str()),
563 };
564
565 if let Some(name) = dir_name
566 && synced.insert(name.to_string())
567 {
568 let dir = collections_dir.join(name);
569 if dir.exists() {
570 let f = std::fs::File::open(&dir)?;
571 f.sync_all()?;
572 }
573 }
574 }
575
576 // Also fsync the collections directory itself (for create/delete collection operations).
577 let f = std::fs::File::open(collections_dir)?;
578 f.sync_all()?;
579 }
580
581 Ok(())
582}
583
584/// Remove all files (not directories) in the given directory.
585fn cleanup_directory_contents(dir: &Path) -> Result<(), StorageError> {
586 if !dir.exists() {
587 return Ok(());
588 }
589 for entry in std::fs::read_dir(dir)? {
590 let entry = entry?;
591 let path = entry.path();
592 if path.is_file() {
593 std::fs::remove_file(&path)?;
594 }
595 }
596 Ok(())
597}
Back to top