backup/
boot.rs

1//! Restore on boot: a server whose data directory holds no store restores
2//! the newest restore point of a backup prefix into it before it opens the
3//! store (`MUSHROOMDB_RESTORE_FROM_S3`).
4//!
5//! In order:
6//!
7//! 1. A data directory that holds a store, and no install marker, is never
8//!    restored over: [`BootRestore::Skipped`]. Only what killed restores into
9//!    it left is swept.
10//! 2. The prefix's bucket lease is taken. A fresh lease of another instance
11//!    means that instance may still be writing: nothing is restored and
12//!    nothing is written to the data directory ([`BootError::ForeignLease`]).
13//! 3. Under the lease, a restore on boot that a crash stopped while it moved
14//!    its files in is undone ([`crate::restore::undo_interrupted_install`]),
15//!    unless the store has changed since that install (it may have served):
16//!    then nothing is removed and the boot fails.
17//! 4. The newest restore point is restored in place
18//!    ([`crate::restore::restore_in_place`]) while the lease is renewed. The
19//!    store gets a new lineage and records the restore point it came from,
20//!    so its backup can continue under the same prefix beside the parent's
21//!    chain without ever touching it (see [`crate::worker`], *Lineage*).
22//!    A prefix with no backup at all is [`BootRestore::NoBackup`]: a first
23//!    deploy starts an empty store. Any other failure is an error, and the
24//!    data directory is left without a store.
25//!
26//! The lease stays held, and renewed, until the caller hands it to the
27//! backup worker ([`BootLease::into_lease`], then
28//! [`crate::BackupWorker::adopt_lease`]) or releases it.
29
30use crate::lease::{new_instance_id, BucketLease, LeaseError, LeaseWatch};
31use crate::restore::{
32    install_interrupted, restore_in_place, staging_leftovers, sweep_in_place,
33    undo_interrupted_install, RestoreS3Error, RestoreS3Outcome,
34};
35use crate::store::BackupConfig;
36use object_store::ObjectStore;
37use std::path::Path;
38use std::sync::Arc;
39use std::time::Duration;
40use tokio::task::JoinHandle;
41
42/// The bucket lease a restore on boot took, renewed until it is handed on.
43#[derive(Debug)]
44pub struct BootLease {
45    lease: Arc<BucketLease>,
46    renewer: JoinHandle<()>,
47    op_timeout: Duration,
48}
49
50impl BootLease {
51    fn hold(lease: BucketLease, op_timeout: Duration) -> Self {
52        let lease = Arc::new(lease);
53        let renewer = tokio::spawn(renew(Arc::clone(&lease), op_timeout));
54        Self {
55            lease,
56            renewer,
57            op_timeout,
58        }
59    }
60
61    /// The request deadline the lease was taken with; [`release`](Self::release)
62    /// waits no longer than this for the bucket.
63    pub fn op_timeout(&self) -> Duration {
64        self.op_timeout
65    }
66
67    /// Stop renewing and hand the lease on at once, without waiting: for a
68    /// caller about to [`adopt`](crate::BackupWorker::adopt_lease) it, which
69    /// renews it at once, so the lease is never left unrenewed in between.
70    pub fn detach(self) -> Arc<BucketLease> {
71        self.renewer.abort();
72        self.lease
73    }
74
75    /// The instance id the lease names.
76    pub fn instance_id(&self) -> &str {
77        self.lease.instance_id()
78    }
79
80    /// Stop renewing and hand the lease on, for
81    /// [`crate::BackupWorker::adopt_lease`].
82    pub async fn into_lease(self) -> Arc<BucketLease> {
83        self.renewer.abort();
84        let _ = self.renewer.await;
85        self.lease
86    }
87
88    /// Stop renewing and delete the lease, waiting at most
89    /// [`op_timeout`](Self::op_timeout) for the bucket.
90    pub async fn release(self) {
91        let op_timeout = self.op_timeout;
92        let lease = self.into_lease().await;
93        let Ok(lease) = Arc::try_unwrap(lease) else {
94            return;
95        };
96        match tokio::time::timeout(op_timeout, lease.release()).await {
97            Ok(Ok(())) => {}
98            Ok(Err(e)) => {
99                tracing::warn!(error = %e, "could not release the backup bucket lease; it expires by itself")
100            }
101            Err(_) => tracing::warn!("releasing the backup bucket lease timed out"),
102        }
103    }
104}
105
106async fn renew(lease: Arc<BucketLease>, op_timeout: Duration) {
107    let every = lease.renew_interval().max(Duration::from_millis(1));
108    loop {
109        tokio::time::sleep(every).await;
110        match tokio::time::timeout(op_timeout, lease.renew()).await {
111            Ok(Ok(())) => {}
112            Ok(Err(e @ LeaseError::Lost { .. })) => {
113                tracing::warn!(error = %e, "lost the backup bucket lease during the restore");
114                return;
115            }
116            Ok(Err(e)) => tracing::warn!(error = %e, "bucket lease renewal failed; retrying"),
117            Err(_) => tracing::warn!("bucket lease renewal timed out; retrying"),
118        }
119    }
120}
121
122/// What a restore on boot did.
123#[derive(Debug)]
124pub enum BootRestore {
125    /// The data directory holds a store: nothing was restored, and no lease
126    /// was taken.
127    Skipped,
128    /// The prefix holds no backup: the store starts empty, and its first
129    /// backup starts the prefix.
130    NoBackup { lease: BootLease },
131    /// The newest restore point is in place.
132    Restored {
133        outcome: Box<RestoreS3Outcome>,
134        lease: BootLease,
135        /// Files an interrupted earlier restore on boot had moved in, removed
136        /// before this one.
137        undone: Vec<String>,
138        /// Leftovers of killed restores the sweep had to leave in the data
139        /// directory (in use, too young, or no advisory locks there).
140        leftovers: Vec<std::path::PathBuf>,
141    },
142}
143
144/// Why a restore on boot failed. The data directory then holds no store.
145#[derive(Debug)]
146pub enum BootError {
147    /// Another instance holds a fresh lease on the prefix.
148    ForeignLease { holder: String },
149    /// The lease could not be taken for another reason.
150    Lease(String),
151    /// The restore failed.
152    Restore(RestoreS3Error),
153    /// The data directory could not be read or cleaned up.
154    Io(std::io::Error),
155}
156
157impl std::fmt::Display for BootError {
158    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
159        match self {
160            BootError::ForeignLease { holder } => write!(
161                f,
162                "the backup bucket's lease is held by another instance ({holder}), which may \
163                 still be writing to the backup; nothing was restored. Stop that instance, or \
164                 wait for its lease to expire (MUSHROOMDB_BACKUP_LEASE_TTL), and start again"
165            ),
166            BootError::Lease(m) => write!(f, "taking the bucket lease: {m}"),
167            BootError::Restore(e) => write!(f, "{e}"),
168            BootError::Io(e) => write!(f, "{e}"),
169        }
170    }
171}
172
173impl std::error::Error for BootError {}
174
175/// Restore the newest restore point of `config.s3_url`'s prefix into
176/// `data_dir` if it holds no store. See the [module docs](self).
177///
178/// The lease is taken with `config`'s TTL, conditional mode and request
179/// deadline.
180pub async fn restore_on_boot(
181    config: &BackupConfig,
182    store: Arc<dyn ObjectStore>,
183    data_dir: &Path,
184) -> Result<BootRestore, BootError> {
185    let prefix = config
186        .location()
187        .map_err(|e| BootError::Restore(RestoreS3Error::Config(e.to_string())))?
188        .prefix;
189    let dir = data_dir.to_path_buf();
190    let (interrupted, holds) = blocking(move || {
191        Ok((
192            install_interrupted(&dir),
193            mushroomdb::restore::holds_a_store(&dir),
194        ))
195    })
196    .await?;
197    if holds && !interrupted {
198        // Never restored over; only what killed restores left is cleaned.
199        let dir = data_dir.to_path_buf();
200        for path in blocking(move || Ok(sweep_in_place(&dir))).await? {
201            tracing::info!(path = %path.display(), "removed a killed restore's leftover");
202        }
203        return Ok(BootRestore::Skipped);
204    }
205
206    let op_timeout = Duration::from_millis(config.op_timeout_ms);
207    let (instance_id, mut watch) = (new_instance_id(), LeaseWatch::default());
208    let acquire = BucketLease::acquire_as(
209        Arc::clone(&store),
210        &prefix,
211        config.lease_ttl_s,
212        &instance_id,
213        &mut watch,
214    );
215    let lease = match tokio::time::timeout(op_timeout * 4, acquire).await {
216        Ok(Ok(lease)) => BootLease::hold(lease, op_timeout),
217        Ok(Err(LeaseError::ForeignActive { instance_id })) => {
218            return Err(BootError::ForeignLease {
219                holder: instance_id,
220            })
221        }
222        Ok(Err(e @ LeaseError::Store(_))) if config.lease_conditional => {
223            return Err(BootError::Lease(format!(
224                "{e} (if this service rejects conditional PUT headers, set \
225                 MUSHROOMDB_BACKUP_LEASE_CONDITIONAL=off)"
226            )))
227        }
228        Ok(Err(e)) => return Err(BootError::Lease(e.to_string())),
229        Err(_) => {
230            return Err(BootError::Lease(
231                "the bucket did not answer within the request deadline".into(),
232            ))
233        }
234    };
235
236    // Under the lease: a crash's half-done install is undone only by the
237    // instance that will restore again.
238    let undone = if interrupted {
239        let dir = data_dir.to_path_buf();
240        match blocking(move || undo_interrupted_install(&dir).map_err(BootError::Io)).await {
241            Ok(undone) => {
242                tracing::warn!(
243                    files = undone.len(),
244                    "removed the files of a restore on boot that a crash interrupted"
245                );
246                undone
247            }
248            Err(e) => {
249                lease.release().await;
250                return Err(e);
251            }
252        }
253    } else {
254        Vec::new()
255    };
256
257    match restore_in_place(store, &prefix, data_dir).await {
258        Ok(outcome) => {
259            let dir = data_dir.to_path_buf();
260            let leftovers = blocking(move || Ok(staging_leftovers(&dir))).await?;
261            for path in &leftovers {
262                tracing::warn!(
263                    path = %path.display(),
264                    "a killed restore's leftover stays in the data directory"
265                );
266            }
267            Ok(BootRestore::Restored {
268                outcome: Box::new(outcome),
269                lease,
270                undone,
271                leftovers,
272            })
273        }
274        Err(RestoreS3Error::NoBackup { .. }) => {
275            let dir = data_dir.to_path_buf();
276            for path in blocking(move || Ok(sweep_in_place(&dir))).await? {
277                tracing::info!(path = %path.display(), "removed a killed restore's leftover");
278            }
279            Ok(BootRestore::NoBackup { lease })
280        }
281        Err(e) => {
282            // Nothing will back this server up: let a successor in at once.
283            lease.release().await;
284            Err(BootError::Restore(e))
285        }
286    }
287}
288
289async fn blocking<T: Send + 'static>(
290    f: impl FnOnce() -> Result<T, BootError> + Send + 'static,
291) -> Result<T, BootError> {
292    tokio::task::spawn_blocking(f)
293        .await
294        .map_err(|e| BootError::Io(std::io::Error::other(e.to_string())))?
295}
backup/boot.rs

backup/
boot.rs
