feat(backup): backup sync --move-store, --forget-store, --forget-backup [backup/restore 10b]

crates/backup-store/src/fs.rs

@@ -54,7 +54,9 @@
 //! blocking pool, where dropping the future does not stop it; a guard in
 //! the future sets a flag when it is dropped, and the walk checks the flag
 //! before each removal, so a dropped `delete_prefix` stops at its next
-//! entry.
+//! entry. `delete_prefix_before` checks its deadline at the same point, so a
+//! walk that reaches the deadline stops before its next removal, whatever
+//! the caller's timer has done by then.
 //!
 //! Writes go to a temporary sibling file and rename into place, so a crash
 //! never leaves a truncated file under its final name. The temporary is
@@ -126,7 +128,7 @@ use futures::{FutureExt, StreamExt, TryStreamExt};
 use tokio::io::AsyncWriteExt;
 
 use crate::key::{normalize_prefix, validate_key, validate_prefix};
-use crate::{BackupStore, ByteStream, ObjectMeta, StoreError};
+use crate::{BackupStore, ByteStream, ObjectMeta, PrefixDeletion, StoreError};
 
 /// Prefix and suffix marking in-flight temporary files, hidden from listings.
 const TEMP_PREFIX: &str = ".put-";
@@ -600,6 +602,19 @@ impl FilesystemStore {
     }
 
     async fn delete_prefix_impl(&self, prefix: &str) -> Result<u64, StoreError> {
+        self.delete_walk(prefix, None)
+            .await
+            .map(|done| done.deleted)
+    }
+
+    /// The walk of `delete_prefix` and `delete_prefix_before`: with a
+    /// `deadline`, the walk starts no removal at or after it and returns
+    /// what it removed with `finished` false.
+    async fn delete_walk(
+        &self,
+        prefix: &str,
+        deadline: Option<tokio::time::Instant>,
+    ) -> Result<PrefixDeletion, StoreError> {
         validate_prefix(prefix)?;
         let normalized = normalize_prefix(prefix).to_owned();
         let root = self.root.clone();
@@ -614,10 +629,25 @@ impl FilesystemStore {
             cancel: cancel.0.clone(),
             removal_hook: self.removal_hook.clone(),
             removals: std::sync::atomic::AtomicU64::new(0),
+            deadline,
+            out_of_time: std::sync::atomic::AtomicBool::new(false),
+            removed: std::sync::atomic::AtomicU64::new(0),
         };
-        tokio::task::spawn_blocking(move || delete_objects(&root, &normalized, &walk))
-            .await
-            .map_err(|e| StoreError::Other(format!("deletion task failed: {e}")))?
+        tokio::task::spawn_blocking(move || {
+            let outcome = delete_objects(&root, &normalized, &walk);
+            if walk.out_of_time.load(std::sync::atomic::Ordering::Acquire) {
+                return Ok(PrefixDeletion {
+                    deleted: walk.removed.load(std::sync::atomic::Ordering::Acquire),
+                    finished: false,
+                });
+            }
+            outcome.map(|deleted| PrefixDeletion {
+                deleted,
+                finished: true,
+            })
+        })
+        .await
+        .map_err(|e| StoreError::Other(format!("deletion task failed: {e}")))?
     }
 
     async fn validate_impl(&self) -> Result<(), StoreError> {
@@ -742,6 +772,14 @@ impl BackupStore for FilesystemStore {
         self.delete_prefix_impl(prefix).boxed()
     }
 
+    fn delete_prefix_before<'a>(
+        &'a self,
+        prefix: &'a str,
+        deadline: tokio::time::Instant,
+    ) -> BoxFuture<'a, Result<PrefixDeletion, StoreError>> {
+        self.delete_walk(prefix, Some(deadline)).boxed()
+    }
+
     fn validate(&self) -> BoxFuture<'_, Result<(), StoreError>> {
         self.validate_impl().boxed()
     }
@@ -1254,23 +1292,40 @@ impl Drop for CancelOnDrop {
 }
 
 /// What one `delete_prefix` walk carries down the tree: the flag its
-/// `CancelOnDrop` sets, and the store's removal hook (set only in the
-/// crate's own tests) with the count it is called with.
+/// `CancelOnDrop` sets, the store's removal hook (set only in the crate's
+/// own tests) with the count it is called with, and for
+/// `delete_prefix_before` the deadline, whether the walk reached it, and
+/// the files it removed.
 struct DeleteWalk {
     cancel: std::sync::Arc<std::sync::atomic::AtomicBool>,
     removal_hook: Option<RemovalHook>,
     removals: std::sync::atomic::AtomicU64,
+    deadline: Option<tokio::time::Instant>,
+    out_of_time: std::sync::atomic::AtomicBool,
+    removed: std::sync::atomic::AtomicU64,
 }
 
 impl DeleteWalk {
-    /// The error a cancelled walk ends with; nobody receives it, since the
-    /// caller's future is gone.
+    /// The error a cancelled walk, or one past its deadline, ends with;
+    /// nobody receives the first, since the caller's future is gone, and
+    /// `delete_walk` turns the second into an unfinished deletion. Called
+    /// before each removal.
     fn check_cancel(&self) -> Result<(), StoreError> {
         if self.cancel.load(std::sync::atomic::Ordering::Acquire) {
             return Err(StoreError::Other(
                 "delete_prefix was dropped before it finished".to_owned(),
             ));
         }
+        if self
+            .deadline
+            .is_some_and(|deadline| tokio::time::Instant::now() >= deadline)
+        {
+            self.out_of_time
+                .store(true, std::sync::atomic::Ordering::Release);
+            return Err(StoreError::Other(
+                "delete_prefix_before reached its deadline".to_owned(),
+            ));
+        }
         Ok(())
     }
 
@@ -1368,7 +1423,11 @@ fn check_tree(dir: &Path, dir_meta: &std::fs::Metadata) -> Result<(), StoreError
 fn remove_file_counted(path: &Path, walk: &DeleteWalk) -> Result<u64, StoreError> {
     walk.before_removal();
     match std::fs::remove_file(path) {
-        Ok(()) => Ok(1),
+        Ok(()) => {
+            walk.removed
+                .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+            Ok(1)
+        }
         Err(e) if vanished(&e) => Ok(0),
         Err(e) => Err(StoreError::io(format!("remove {}", path.display()), e)),
     }

crates/backup-store/src/lib.rs

@@ -69,6 +69,17 @@ pub struct ObjectMeta {
     pub etag: Option<String>,
 }
 
+/// What one [`BackupStore::delete_prefix_before`] did.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PrefixDeletion {
+    /// The objects removed, counted as [`BackupStore::delete_prefix`]
+    /// counts them.
+    pub deleted: u64,
+    /// Whether the deletion ran to its end; `false` when it stopped at the
+    /// deadline, with the rest of the prefix left.
+    pub finished: bool,
+}
+
 /// A destination for backup objects.
 ///
 /// Implementations must be safe to share across tasks; the engine holds one
@@ -189,6 +200,44 @@ pub trait BackupStore: Send + Sync {
     /// the dropped future sets).
     fn delete_prefix<'a>(&'a self, prefix: &'a str) -> BoxFuture<'a, Result<u64, StoreError>>;
 
+    /// [`BackupStore::delete_prefix`] that starts no removal at or after
+    /// `deadline`: the filesystem store's walk checks the deadline beside its
+    /// cancel flag before each removal, the S3 store before each
+    /// `ListObjectsV2`, `DeleteObjects`, `ListMultipartUploads` and
+    /// `AbortMultipartUpload`. A call that reaches the deadline returns what
+    /// it removed so far with [`PrefixDeletion::finished`] false; what
+    /// remains is the step in flight at the deadline (one removal on a
+    /// filesystem root, one request on S3, up to 1,000 keys for a
+    /// `DeleteObjects`). The check reads the clock in the process, so it
+    /// bounds a process that keeps running; one paused between a check and
+    /// the step it admits (a stopped process, a frozen VM) runs that step
+    /// when it resumes. The caller still bounds the call with a timeout for
+    /// a step that never answers.
+    ///
+    /// The default checks the deadline once and then runs `delete_prefix`
+    /// whole: for a store whose deletion is one step (an in-memory test
+    /// store), the same rule.
+    fn delete_prefix_before<'a>(
+        &'a self,
+        prefix: &'a str,
+        deadline: tokio::time::Instant,
+    ) -> BoxFuture<'a, Result<PrefixDeletion, StoreError>> {
+        Box::pin(async move {
+            if tokio::time::Instant::now() >= deadline {
+                return Ok(PrefixDeletion {
+                    deleted: 0,
+                    finished: false,
+                });
+            }
+            self.delete_prefix(prefix)
+                .await
+                .map(|deleted| PrefixDeletion {
+                    deleted,
+                    finished: true,
+                })
+        })
+    }
+
     /// Startup validation. The filesystem store applies the owner-only rule
     /// to the root and the ancestor rule to the directories above it (the
     /// `fs` module doc states both), checks that the root is still where

crates/backup-store/src/s3.rs

@@ -102,7 +102,7 @@ use futures::{FutureExt, StreamExt, TryStreamExt};
 
 use crate::config::S3StoreConfig;
 use crate::key::{key_matches_prefix, normalize_prefix, validate_key, validate_prefix};
-use crate::{BackupStore, ByteStream, ObjectMeta, StoreError};
+use crate::{BackupStore, ByteStream, ObjectMeta, PrefixDeletion, StoreError};
 
 /// The smallest part size S3 accepts for any part except the last.
 pub const MIN_PART_SIZE_BYTES: u64 = 5 * 1024 * 1024;
@@ -925,12 +925,34 @@ impl S3Store {
     }
 
     async fn delete_prefix_impl(&self, prefix: &str) -> Result<u64, StoreError> {
+        self.delete_requests(prefix, None)
+            .await
+            .map(|done| done.deleted)
+    }
+
+    /// The requests of `delete_prefix` and `delete_prefix_before`: with a
+    /// `deadline`, no request is sent at or after it, and the call returns
+    /// what it deleted with `finished` false.
+    async fn delete_requests(
+        &self,
+        prefix: &str,
+        deadline: Option<tokio::time::Instant>,
+    ) -> Result<PrefixDeletion, StoreError> {
         validate_prefix(prefix)?;
         let match_prefix = normalize_prefix(prefix);
         let request_prefix = format!("{}{match_prefix}", self.prefix);
         let mut deleted: u64 = 0;
         let mut token: Option<String> = None;
+        let out_of_time =
+            || deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline);
+        let unfinished = |deleted| PrefixDeletion {
+            deleted,
+            finished: false,
+        };
         loop {
+            if out_of_time() {
+                return Ok(unfinished(deleted));
+            }
             // Raw bucket-side keys, not `list_page`'s filtered view: an
             // object whose key fails the store rule is still under the
             // prefix and still has to go.
@@ -964,6 +986,9 @@ impl S3Store {
                     })
                     .collect::<Result<Vec<_>, _>>()?;
                 let count = objects.len() as u64;
+                if out_of_time() {
+                    return Ok(unfinished(deleted));
+                }
                 let delete = Delete::builder()
                     .set_objects(Some(objects))
                     .quiet(true)
@@ -1006,9 +1031,16 @@ impl S3Store {
         // does not see them and S3 bills their parts until they are aborted.
         // A process that died mid-put left one; the reconciler's prefix
         // deletion is the one place that sweeps it.
-        self.abort_uploads_under(&request_prefix, match_prefix)
-            .await?;
-        Ok(deleted)
+        if !self
+            .abort_uploads_under(&request_prefix, match_prefix, deadline)
+            .await?
+        {
+            return Ok(unfinished(deleted));
+        }
+        Ok(PrefixDeletion {
+            deleted,
+            finished: true,
+        })
     }
 
     /// Abort every in-progress multipart upload under the normalized store
@@ -1027,14 +1059,22 @@ impl S3Store {
     /// backstop for such uploads, and `validate` reports either missing
     /// grant at startup (it probes both). A token code (`ExpiredToken` and
     /// the others in `is_token_code`) is not a denial and fails the call.
+    /// With a `deadline`, no request is sent at or after it, and the sweep
+    /// answers `false` (not finished); otherwise `true`.
     async fn abort_uploads_under(
         &self,
         request_prefix: &str,
         match_prefix: &str,
-    ) -> Result<(), StoreError> {
+        deadline: Option<tokio::time::Instant>,
+    ) -> Result<bool, StoreError> {
+        let out_of_time =
+            || deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline);
         let mut key_marker: Option<String> = None;
         let mut id_marker: Option<String> = None;
         loop {
+            if out_of_time() {
+                return Ok(false);
+            }
             let output = match self
                 .client
                 .list_multipart_uploads()
@@ -1056,7 +1096,7 @@ impl S3Store {
                          there stay open until the bucket's AbortIncompleteMultipartUpload \
                          lifecycle rule removes them"
                     );
-                    return Ok(());
+                    return Ok(true);
                 }
                 Err(e) => return Err(map_sdk_error("ListMultipartUploads", e)),
             };
@@ -1070,6 +1110,9 @@ impl S3Store {
                 if !matches {
                     continue;
                 }
+                if out_of_time() {
+                    return Ok(false);
+                }
                 match self
                     .client
                     .abort_multipart_upload()
@@ -1110,7 +1153,7 @@ impl S3Store {
                 break;
             }
         }
-        Ok(())
+        Ok(true)
     }
 
     async fn validate_impl(&self) -> Result<(), StoreError> {
@@ -1325,6 +1368,14 @@ impl BackupStore for S3Store {
         self.delete_prefix_impl(prefix).boxed()
     }
 
+    fn delete_prefix_before<'a>(
+        &'a self,
+        prefix: &'a str,
+        deadline: tokio::time::Instant,
+    ) -> BoxFuture<'a, Result<PrefixDeletion, StoreError>> {
+        self.delete_requests(prefix, Some(deadline)).boxed()
+    }
+
     /// Lists one key under the prefix (`ListObjectsV2`, `s3:ListBucket`)
     /// rather than calling `HeadBucket`, since only a GET-shaped error has a
     /// body: a 403 keeps S3's code, a 404 names the missing bucket, a 301 the