linera_persistent/
file.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! A [`Persist`] backend that atomically saves the value to a locked file on disk.
5
6use std::{
7    io::{self, BufRead as _, Write as _},
8    path::Path,
9};
10
11use fs4::FileExt;
12
13use super::Persist;
14
15/// A guard that keeps an exclusive lock on a file.
16struct Lock(fs_err::File);
17
18/// The kinds of error that persisting a value to a file can produce.
19#[derive(Debug, thiserror::Error)]
20#[non_exhaustive]
21pub enum Error {
22    /// An I/O operation on the file failed.
23    #[error("I/O error: {0}")]
24    IoError(#[from] std::io::Error),
25    /// The value could not be serialized to, or deserialized from, JSON.
26    #[error("JSON error: {0}")]
27    JsonError(#[from] serde_json::Error),
28    /// The file could not be locked for exclusive access.
29    #[error("failed to lock {}: {source}", path.display())]
30    Lock {
31        /// The path that could not be locked.
32        path: std::path::PathBuf,
33        /// The underlying I/O failure.
34        #[source]
35        source: std::io::Error,
36    },
37    /// An operation failed, and so did the cleanup that followed it.
38    #[error("failed to clean up after an error: {cleanup}; the original error was: {original}")]
39    Cleanup {
40        /// The failure of the cleanup step itself.
41        #[source]
42        cleanup: Box<Error>,
43        /// The failure that prompted the cleanup.
44        original: Box<Error>,
45    },
46}
47
48/// Utility: run a fallible cleanup function if an operation failed, reporting both
49/// failures if the cleanup fails too.
50trait CleanupExt {
51    /// The success type of the operation.
52    type Ok;
53
54    /// Runs `cleanup` if the operation failed.
55    fn or_cleanup<E: Into<Error>>(
56        self,
57        cleanup: impl FnOnce() -> Result<(), E>,
58    ) -> Result<Self::Ok, Error>;
59}
60
61impl<T> CleanupExt for Result<T, Error> {
62    type Ok = T;
63
64    fn or_cleanup<E: Into<Error>>(
65        self,
66        cleanup: impl FnOnce() -> Result<(), E>,
67    ) -> Result<T, Error> {
68        self.map_err(|original| match cleanup() {
69            Ok(()) => original,
70            Err(cleanup) => Error::Cleanup {
71                cleanup: Box::new(cleanup.into()),
72                original: Box::new(original),
73            },
74        })
75    }
76}
77
78impl Lock {
79    /// Acquires an exclusive lock on a provided `file`, returning a [`Lock`] which will
80    /// release the lock when dropped.
81    pub fn new(file: fs_err::File) -> std::io::Result<Self> {
82        file.file().try_lock_exclusive()?;
83        Ok(Lock(file))
84    }
85}
86
87impl Drop for Lock {
88    fn drop(&mut self) {
89        if let Err(error) = FileExt::unlock(self.0.file()) {
90            tracing::warn!("Failed to unlock wallet file: {error}");
91        }
92    }
93}
94
95/// An implementation of [`Persist`] based on an atomically-updated file at a given path.
96/// An exclusive lock is taken using `flock(2)` to ensure that concurrent updates cannot
97/// happen, and writes are saved to a staging file before being moved over the old file,
98/// an operation that is atomic on all Unixes.
99pub struct File<T> {
100    _lock: Lock,
101    path: std::path::PathBuf,
102    value: T,
103}
104
105impl<T> std::ops::Deref for File<T> {
106    type Target = T;
107    fn deref(&self) -> &T {
108        &self.value
109    }
110}
111
112impl<T> std::ops::DerefMut for File<T> {
113    fn deref_mut(&mut self) -> &mut T {
114        &mut self.value
115    }
116}
117
118/// Returns options for opening and writing to the file, creating it if it doesn't
119/// exist. On Unix, this restricts read and write permissions to the current user.
120// TODO(#1924): Implement better key management.
121// BUG(#2053): Use a separate lock file per staging file.
122fn open_options() -> fs_err::OpenOptions {
123    let mut options = fs_err::OpenOptions::new();
124    #[cfg(target_family = "unix")]
125    fs_err::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
126    options.create(true).read(true).write(true);
127    options
128}
129
130impl<T: serde::Serialize + serde::de::DeserializeOwned> File<T> {
131    /// Creates a new persistent file at `path` containing `value`.
132    pub fn new(path: &Path, value: T) -> Result<Self, Error> {
133        let this = Self {
134            _lock: Lock::new(
135                fs_err::OpenOptions::new()
136                    .read(true)
137                    .write(true)
138                    .create(true)
139                    .open(path)?,
140            )
141            .map_err(|source| Error::Lock {
142                path: path.into(),
143                source,
144            })?,
145            path: path.into(),
146            value,
147        };
148        this.save()?;
149        Ok(this)
150    }
151
152    /// Reads the value from a file at `path`, returning an error if it does not exist.
153    pub fn read(path: &Path) -> Result<Self, Error> {
154        Self::read_or_create(path, || {
155            Err(std::io::Error::new(
156                std::io::ErrorKind::NotFound,
157                format!("file is empty or does not exist: {}", path.display()),
158            )
159            .into())
160        })
161    }
162
163    /// Reads the value from a file at `path`, calling the `value` function to create it
164    /// if it does not exist. If it does exist, `value` will not be called.
165    pub fn read_or_create(
166        path: &Path,
167        value: impl FnOnce() -> Result<T, Error>,
168    ) -> Result<Self, Error> {
169        let lock = Lock::new(open_options().read(true).open(path)?)?;
170        let mut reader = io::BufReader::new(&lock.0);
171        let file_is_empty = reader.fill_buf()?.is_empty();
172
173        let me = Self {
174            value: if file_is_empty {
175                value()?
176            } else {
177                serde_json::from_reader(reader)?
178            },
179            path: path.into(),
180            _lock: lock,
181        };
182
183        me.save()?;
184
185        Ok(me)
186    }
187
188    /// Atomically writes the current value to the file, via a temporary staging file.
189    pub fn save(&self) -> Result<(), Error> {
190        let mut temp_file_path = self.path.clone();
191        temp_file_path.set_extension("json.new");
192        let temp_file = open_options().open(&temp_file_path)?;
193        let mut temp_file_writer = std::io::BufWriter::new(temp_file);
194
195        let remove_temp_file = || fs_err::remove_file(&temp_file_path);
196
197        serde_json::to_writer_pretty(&mut temp_file_writer, &self.value)
198            .map_err(Error::from)
199            .or_cleanup(remove_temp_file)?;
200        temp_file_writer
201            .flush()
202            .map_err(Error::from)
203            .or_cleanup(remove_temp_file)?;
204        drop(temp_file_writer);
205        fs_err::rename(&temp_file_path, &self.path)?;
206        Ok(())
207    }
208}
209
210impl<T: serde::Serialize + serde::de::DeserializeOwned + Send> Persist for File<T> {
211    type Error = Error;
212
213    fn as_mut(&mut self) -> &mut T {
214        &mut self.value
215    }
216
217    /// Writes the value to disk.
218    ///
219    /// The contents of the file need to be over-written completely, so
220    /// a temporary file is created as a backup in case a crash occurs while
221    /// writing to disk.
222    ///
223    /// The temporary file is then renamed to the original filename. If
224    /// serialization or writing to disk fails, the temporary file is
225    /// deleted.
226    fn persist(&mut self) -> impl std::future::Future<Output = Result<(), Error>> {
227        let result = self.save();
228        async { result }
229    }
230
231    /// Takes the value out, releasing the lock on the persistent file.
232    fn into_value(self) -> T {
233        self.value
234    }
235}
linera_persistent/file.rs

linera_persistent/
file.rs
