fix: second at-rest review

donut-sync/src/lib.rs

@@ -219,6 +219,9 @@ pub async fn build_state(config: &Config) -> Result<Arc<AppState>, String> {
   if let Some(keyring) = keyring.filter(|k| !k.unsealing()) {
     crate::sealer::spawn(s3, keyring);
   }
+  if let Some(direct) = sync.direct_uploads() {
+    crate::sealer::spawn_flusher(sync.store().clone(), direct.clone());
+  }
 
   Ok(Arc::new(AppState {
     guard,
@@ -353,6 +356,14 @@ pub async fn run() -> i32 {
   };
   let port = listener.local_addr().map_or(port, |a| a.port());
   println!("Donut Sync service running on port {port}");
-  serve(listener, state, stop_signal()).await;
+  serve(listener, state.clone(), stop_signal()).await;
+  if let Some(direct) = state.sync.direct_uploads()
+    && let Err(e) = direct.flush(state.sync.store().as_ref()).await
+  {
+    log::warn(
+      "Sealer",
+      &format!("Direct uploads not noted at exit err=\"{e}\""),
+    );
+  }
   0
 }

donut-sync/src/memory_store.rs

@@ -12,7 +12,8 @@ use md5::{Digest, Md5};
 
 use crate::store::{
   ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListObject, ListOutput,
-  ObjectVersion, PendingUpload, Store, StoreError, Upload, collect, once,
+  ObjectVersion, PendingUpload, Store, StoreError, Upload, VersionMarker, VersionsPage, collect,
+  once,
 };
 
 pub type HeadHook = Box<dyn Fn(&str) -> Result<HeadOutput, StoreError> + Send + Sync>;
@@ -58,6 +59,10 @@ pub struct MemoryStore {
   pub list_hook: Option<ListHook>,
   pub copy_hook: Mutex<Option<CopyHook>>,
   pub versioned: bool,
+  /// Like B2: a PUT with If-Match is refused (501), never conditional.
+  pub ignores_if_match: bool,
+  /// Versions per listed page; 0 is 1000.
+  pub version_page: usize,
   pub uploads: Mutex<BTreeMap<String, Multipart>>,
 }
 
@@ -136,10 +141,10 @@ impl MemoryStore {
   }
 
   /// A device's delete: a delete marker where versions are kept.
-  pub fn remove(&self, key: &str) {
-    if self.raw(key).is_some() {
-      self.push(key, None);
-    }
+  pub fn remove(&self, key: &str) -> Option<String> {
+    self.raw(key)?;
+    let id = self.push(key, None);
+    self.versioned.then_some(id)
   }
 
   pub fn get(&self, key: &str) -> Option<String> {
@@ -228,10 +233,9 @@ impl Store for MemoryStore {
     Ok(())
   }
 
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+  async fn delete_object(&self, key: &str) -> Result<Option<String>, StoreError> {
     self.call();
-    self.remove(key);
-    Ok(())
+    Ok(self.remove(key))
   }
 
   async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> {
@@ -279,11 +283,20 @@ impl Store for MemoryStore {
     })
   }
 
-  async fn list_versions(&self, prefix: &str) -> Result<Option<Vec<ObjectVersion>>, StoreError> {
+  async fn list_versions(
+    &self,
+    prefix: &str,
+    after: Option<&VersionMarker>,
+  ) -> Result<Option<VersionsPage>, StoreError> {
     self.call();
     if !self.versioned {
       return Ok(None);
     }
+    let page = if self.version_page == 0 {
+      1000
+    } else {
+      self.version_page
+    };
     let objects = self.objects.lock().unwrap();
     let mut out = Vec::new();
     for (key, history) in objects.range(prefix.to_string()..) {
@@ -301,7 +314,24 @@ impl Store for MemoryStore {
         });
       }
     }
-    Ok(Some(out))
+    let start = match after {
+      Some((key, id)) => out
+        .iter()
+        .position(|v| v.key == *key && v.version_id == *id)
+        .map_or(out.len(), |at| at + 1),
+      None => 0,
+    };
+    let versions: Vec<ObjectVersion> = out.into_iter().skip(start).collect();
+    let next = (versions.len() > page).then(|| {
+      (
+        versions[page - 1].key.clone(),
+        versions[page - 1].version_id.clone(),
+      )
+    });
+    Ok(Some(VersionsPage {
+      versions: versions.into_iter().take(page).collect(),
+      next,
+    }))
   }
 
   async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> {
@@ -394,6 +424,13 @@ impl Store for MemoryStore {
     if body.len() as u64 != upload.length {
       return Err(StoreError::Transport("length mismatch".to_string()));
     }
+    if upload.if_match.is_some() && self.ignores_if_match {
+      return Err(StoreError::Service {
+        status: 501,
+        code: "NotImplemented".to_string(),
+        message: String::new(),
+      });
+    }
     if let Some(expected) = upload.if_match
       && self.raw(key).map(|o| etag(&o.body)).as_deref() != Some(expected)
     {

donut-sync/src/migrate.rs

@@ -34,7 +34,12 @@ use crate::config::Config;
 use crate::log;
 use crate::original_times::{ORIGINAL_MTIME, original_mtime};
 use crate::sealed_store::sealed_by;
-use crate::store::{ByteStream, CopySpec, HeadOutput, ListInput, Store, StoreError, Upload};
+use crate::sealer::{Failed, Failure, read_state, retry_failed, save_state};
+use crate::store::{
+  ByteStream, CopySpec, HeadOutput, ListInput, Store, StoreError, Upload, versions_of,
+};
+use crate::time;
+use serde::{Deserialize, Serialize};
 
 pub const COMMAND: &str = "migrate-at-rest";
 const CONTEXT: &str = "Migrate";
@@ -43,6 +48,8 @@ pub const TMP_PREFIX: &str = ".donut-sync-at-rest/tmp/";
 const PROGRESS_EVERY: Duration = Duration::from_secs(10);
 /// Tries per object: storage has transient failures.
 const ATTEMPTS: u32 = 3;
+/// Put-backs in a row before a settle gives up on a history that keeps moving.
+const SETTLE_ROUNDS: usize = 3;
 
 const USAGE: &str = "Usage: donut-sync migrate-at-rest [--unseal] [--dry-run] [--prefix P] [--concurrency N] [--max-mib-per-sec N] [--checkpoint FILE]
 
@@ -172,6 +179,12 @@ pub(crate) enum Outcome {
   Rotated(u64),
   Unsealed(u64),
   Changed,
+  /// Rewritten, but a device's version may lie under the rewrite: settle it
+  /// later with these ids.
+  Unsettled {
+    settle: Settle,
+    reason: String,
+  },
   Failed(String),
 }
 
@@ -247,14 +260,14 @@ fn precondition_failed(error: &StoreError) -> bool {
 /// Deletes `key` with every version it has, where the bucket keeps versions,
 /// so a temporary object leaves nothing behind.
 pub async fn purge(store: &dyn Store, key: &str) -> Result<(), StoreError> {
-  match store.list_versions(key).await? {
+  match versions_of(store, key).await? {
     Some(versions) => {
-      for version in versions.iter().filter(|v| v.key == key) {
+      for version in versions {
         store.delete_version(key, &version.version_id).await?;
       }
       Ok(())
     }
-    None => store.delete_object(key).await,
+    None => store.delete_object(key).await.map(|_| ()),
   }
 }
 
@@ -415,66 +428,126 @@ impl Migration {
     let Some(ours) = placed? else {
       return Ok(Outcome::Changed);
     };
-    match self
-      .settle(key, head.version_id.as_deref(), ours.as_deref())
-      .await?
-    {
-      true => Ok(Outcome::Changed),
-      false => Ok(done),
+    // From here the copy is in place: an error must not send the object
+    // back to the HEAD above, which would take our copy for current and
+    // leave a device's version under it.
+    let settle = Settle {
+      read: head.version_id.clone(),
+      ours,
+    };
+    Ok(match self.settle(key, &settle).await {
+      Ok(true) => Outcome::Changed,
+      Ok(false) => done,
+      Err(reason) => Outcome::Unsettled { settle, reason },
+    })
+  }
+
+  /// `settle_once`, tried again on a transient failure.
+  pub(crate) async fn settle(&self, key: &str, settle: &Settle) -> Result<bool, String> {
+    let mut attempt = 0;
+    loop {
+      attempt += 1;
+      match self.settle_once(key, settle).await {
+        Ok(restored) => return Ok(restored),
+        Err(SettleError::Store(e)) if attempt < ATTEMPTS => {
+          log::warn(CONTEXT, &format!("Settle retry key={key} err=\"{e}\""));
+          tokio::time::sleep(Duration::from_millis(500 << attempt)).await;
+        }
+        Err(e) => return Err(e.to_string()),
+      }
     }
   }
 
   /// After the copy: a version made between the read and the copy (a
-  /// device's write or delete) is put back as the current one. `true` when
-  /// one was. Needs a bucket that keeps versions.
-  async fn settle(
-    &self,
-    key: &str,
-    read: Option<&str>,
-    ours: Option<&str>,
-  ) -> Result<bool, StoreError> {
-    let (Some(read), Some(ours)) = (read, ours) else {
-      return Ok(false);
-    };
-    let Some(versions) = self.store.list_versions(key).await? else {
-      return Ok(false);
-    };
-    let history: Vec<_> = versions.iter().filter(|v| v.key == key).collect();
-    let (Some(at_ours), Some(at_read)) = (
-      history.iter().position(|v| v.version_id == ours),
-      history.iter().position(|v| v.version_id == read),
-    ) else {
-      return Ok(false);
-    };
-    // Newest first: anything before ours came after it and stays current.
-    if at_ours != 0 || at_read <= at_ours + 1 {
-      return Ok(false);
+  /// device's write or delete) is put back as the current one, and again for
+  /// a version made while that was put back, until the history holds still.
+  /// `true` when one was. A bucket without versions has nothing to settle.
+  async fn settle_once(&self, key: &str, settle: &Settle) -> Result<bool, SettleError> {
+    let (mut read, mut ours) = (settle.read.clone(), settle.ours.clone());
+    let mut restored = false;
+    for _ in 0..SETTLE_ROUNDS {
+      let Some(history) = versions_of(self.store.as_ref(), key).await? else {
+        return Ok(restored);
+      };
+      let (Some(read_id), Some(ours_id)) = (read.as_deref(), ours.as_deref()) else {
+        return Err(SettleError::Unknown(
+          "the bucket keeps versions but gave no version id",
+        ));
+      };
+      let Some(at_ours) = history.iter().position(|v| v.version_id == ours_id) else {
+        return Err(SettleError::Unknown("our version is not listed"));
+      };
+      let at_read = history
+        .iter()
+        .position(|v| v.version_id == read_id)
+        .unwrap_or(history.len());
+      // Newest first: anything before ours came after it and stays current.
+      if at_ours != 0 || at_read <= at_ours + 1 {
+        return Ok(restored);
+      }
+      let lost = &history[at_ours + 1];
+      let put_back = if lost.delete_marker {
+        let marker = self.store.delete_object(key).await?;
+        log::warn(
+          CONTEXT,
+          &format!("Delete made during the rewrite put back key={key}"),
+        );
+        marker
+      } else {
+        let copied = self
+          .store
+          .copy_object_with(
+            key,
+            key,
+            CopySpec {
+              source_version: Some(&lost.version_id),
+              ..CopySpec::default()
+            },
+          )
+          .await?;
+        log::warn(
+          CONTEXT,
+          &format!("Write made during the rewrite put back key={key}"),
+        );
+        copied.version_id
+      };
+      restored = true;
+      // Next round: a version between this put-back and ours came while it
+      // was being put back.
+      read = ours;
+      ours = put_back;
     }
-    let lost = history[at_ours + 1];
-    if lost.delete_marker {
-      self.store.delete_object(key).await?;
-      log::warn(
-        CONTEXT,
-        &format!("Delete made during the rewrite put back key={key}"),
-      );
-    } else {
-      self
-        .store
-        .copy_object_with(
-          key,
-          key,
-          CopySpec {
-            source_version: Some(&lost.version_id),
-            ..CopySpec::default()
-          },
-        )
-        .await?;
-      log::warn(
-        CONTEXT,
-        &format!("Write made during the rewrite put back key={key}"),
-      );
+    Err(SettleError::Unknown("the versions kept changing"))
+  }
+}
+
+/// The versions a rewrite read and wrote, to settle it later if it could
+/// not be settled at once.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Settle {
+  pub read: Option<String>,
+  pub ours: Option<String>,
+}
+
+enum SettleError {
+  /// Storage failed; trying again may help.
+  Store(StoreError),
+  /// Needs a person or a later pass.
+  Unknown(&'static str),
+}
+
+impl From<StoreError> for SettleError {
+  fn from(e: StoreError) -> Self {
+    SettleError::Store(e)
+  }
+}
+
+impl std::fmt::Display for SettleError {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    match self {
+      SettleError::Store(e) => write!(f, "{e}"),
+      SettleError::Unknown(why) => write!(f, "{why}"),
     }
-    Ok(true)
   }
 }
 
@@ -581,6 +654,7 @@ pub async fn migrate(
       start_after.as_deref().unwrap_or("-"),
     ),
   );
+  let state_store = crate::sealed_store::SealedStore::new(store.clone(), keyring.clone());
   let migration = Migration {
     store: store.clone(),
     keyring,
@@ -595,6 +669,22 @@ pub async fn migrate(
     }),
   };
   let mut report = Report::default();
+  // Rewrites an earlier run or the sealer left unsettled are settled first.
+  let (retried, mut unsettled) = if options.dry_run {
+    (Vec::new(), Vec::new())
+  } else {
+    let state = read_state(&state_store)
+      .await
+      .map_err(|e| format!("Sealer state unreadable: {e}"))?;
+    let open: Vec<Failed> = state
+      .failed
+      .into_iter()
+      .filter(|f| f.settle.is_some())
+      .collect();
+    let retried: Vec<String> = open.iter().map(|f| f.key.clone()).collect();
+    (retried, retry_failed(&migration, &open).await)
+  };
+  report.failed += unsettled.len() as u64;
   let mut token: Option<String> = None;
   let mut last_progress = Instant::now();
   let mut checkpoint_held = false;
@@ -642,6 +732,19 @@ pub async fn migrate(
           report.unsealed_bytes += bytes;
         }
         Outcome::Changed => report.changed += 1,
+        Outcome::Unsettled { settle, reason } => {
+          report.failed += 1;
+          checkpoint_held = true;
+          log::warn(
+            CONTEXT,
+            &format!("Rewrite not settled key={key} err=\"{reason}\""),
+          );
+          unsettled.push(Failure {
+            key,
+            reason,
+            settle: Some(settle),
+          });
+        }
         Outcome::Failed(err) => {
           report.failed += 1;
           checkpoint_held = true;
@@ -676,6 +779,15 @@ pub async fn migrate(
       );
     }
   }
+  if !retried.is_empty() || !unsettled.is_empty() {
+    let mut state = read_state(&state_store)
+      .await
+      .map_err(|e| format!("Sealer state unreadable: {e}"))?;
+    state.note(&retried, unsettled, time::now_ms());
+    save_state(&state_store, &state)
+      .await
+      .map_err(|e| format!("Sealer state not saved: {e}"))?;
+  }
   // A whole bucket unsealed: this server's own objects (sealed records and
   // state) go too, so nothing sealed is left when the key is removed.
   if options.unseal && !options.dry_run && options.prefix.is_none() && report.failed == 0 {
@@ -1004,7 +1116,9 @@ mod tests {
 
       let raw = Arc::new(MemoryStore::versioned());
       prepare(&raw).await;
-      race(&raw, |store| store.remove("k"));
+      race(&raw, |store| {
+        store.remove("k");
+      });
       assert!(matches!(
         rewrite(&raw, keys.clone(), direction).await,
         Outcome::Changed
@@ -1017,22 +1131,106 @@ mod tests {
   #[tokio::test]
   async fn a_write_after_the_copy_stays_current() {
     let raw = Arc::new(MemoryStore::versioned());
-    raw.insert("k", b"old".to_vec(), Vec::new(), None);
+    let read = raw.insert("k", b"old".to_vec(), Vec::new(), None);
     let migration = Migration::sealing(raw.clone(), keyring(K1), false);
     assert!(matches!(
       migration.object("k", None).await,
       Outcome::Sealed(3)
     ));
+    let ours = raw.head_object("k").await.unwrap().version_id;
     raw.insert("k", b"later".to_vec(), Vec::new(), None);
-    assert!(
-      !migration
-        .settle("k", Some("v1000"), Some("v0"))
-        .await
-        .unwrap()
-    );
+    let settle = Settle {
+      read: Some(read),
+      ours,
+    };
+    assert!(!migration.settle("k", &settle).await.unwrap());
     assert_eq!(raw.raw("k").unwrap().body, b"later");
   }
 
+  /// A store that keeps versions but names none, as no S3 store should.
+  #[tokio::test]
+  async fn missing_version_ids_are_recorded_not_ignored() {
+    let raw = Arc::new(MemoryStore::versioned());
+    raw.insert("k", b"old".to_vec(), Vec::new(), None);
+    let migration = Migration::sealing(raw.clone(), keyring(K1), false);
+    let settle = Settle {
+      read: None,
+      ours: None,
+    };
+    let reason = migration.settle("k", &settle).await.unwrap_err();
+    assert!(reason.contains("no version id"), "{reason}");
+  }
+
+  /// Settle keeps putting back until the history holds still: here a third
+  /// write lands while the first put-back is made.
+  #[tokio::test]
+  async fn a_write_during_the_put_back_is_put_back_too() {
+    let raw = Arc::new(MemoryStore::versioned());
+    raw.insert("k", b"old".to_vec(), Vec::new(), None);
+    // The device writes inside the rewrite's copy, and again inside the
+    // put-back's copy.
+    let copies = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let seen = copies.clone();
+    *raw.copy_hook.lock().unwrap() = Some(Box::new(move |store, to| {
+      if to != "k" {
+        return;
+      }
+      match seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) {
+        0 => {
+          store.insert("k", b"first".to_vec(), Vec::new(), None);
+        }
+        1 => {
+          store.insert("k", b"second".to_vec(), Vec::new(), None);
+        }
+        _ => {}
+      }
+    }));
+    let migration = Migration::sealing(raw.clone(), keyring(K1), false);
+    assert!(matches!(
+      migration.object("k", None).await,
+      Outcome::Changed
+    ));
+    assert_eq!(raw.raw("k").unwrap().body, b"second");
+    assert_eq!(copies.load(std::sync::atomic::Ordering::SeqCst), 3);
+  }
+
+  /// The settle fails after the copy: the object is not rewritten again from
+  /// a HEAD, and the ids are kept for a later pass, which settles it.
+  #[tokio::test]
+  async fn an_unsettled_rewrite_is_recorded_and_settled_later() {
+    let raw = Arc::new(MemoryStore {
+      ignores_if_match: true,
+      ..MemoryStore::versioned()
+    });
+    raw.insert("k", b"old".to_vec(), Vec::new(), None);
+    let fail = Arc::new(std::sync::atomic::AtomicBool::new(true));
+    let failing = fail.clone();
+    *raw.copy_hook.lock().unwrap() = Some(Box::new(move |store, to| {
+      if to == "k" && failing.load(std::sync::atomic::Ordering::SeqCst) {
+        store.insert("k", b"device".to_vec(), Vec::new(), None);
+      }
+    }));
+    // Every put-back copy also races a device write, so the history never
+    // holds still within SETTLE_ROUNDS.
+    let keys = keyring(K1);
+    let report = migrate(raw.clone(), keys.clone(), &options())
+      .await
+      .unwrap();
+    assert_eq!(report.failed, 1);
+    let state_store = crate::sealed_store::SealedStore::new(raw.clone(), keys.clone());
+    let state = read_state(&state_store).await.unwrap();
+    assert_eq!(state.failed.len(), 1);
+    assert!(state.failed[0].settle.is_some());
+    // A later run settles it without a new rewrite.
+    fail.store(false, std::sync::atomic::Ordering::SeqCst);
+    let report = migrate(raw.clone(), keys, &options()).await.unwrap();
+    assert_eq!(report.failed, 0, "{report:?}");
+    assert!(read_state(&state_store).await.unwrap().failed.is_empty());
+    // The device's write stands, and this run sealed it.
+    let served = crate::served_store(raw.clone(), Some(&keyring(K1)));
+    assert_eq!(served.get_object_text("k").await.unwrap(), "device");
+  }
+
   #[test]
   fn arguments_parse() {
     let args: Vec<String> = ["--dry-run", "--prefix", "users/", "--max-mib-per-sec", "8"]

donut-sync/src/original_times.rs

@@ -9,8 +9,8 @@ use std::sync::Arc;
 use async_trait::async_trait;
 
 use crate::store::{
-  ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, ObjectVersion,
-  PendingUpload, Store, StoreError, Upload,
+  ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store,
+  StoreError, Upload, VersionMarker, VersionsPage,
 };
 
 /// User metadata (`x-amz-meta-donut-orig-mtime`): the object's LastModified
@@ -77,7 +77,7 @@ impl Store for OriginalTimes {
       .await
   }
 
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+  async fn delete_object(&self, key: &str) -> Result<Option<String>, StoreError> {
     self.inner.delete_object(key).await
   }
 
@@ -137,8 +137,12 @@ impl Store for OriginalTimes {
     self.inner.copy_object_with(from, to, spec).await
   }
 
-  async fn list_versions(&self, prefix: &str) -> Result<Option<Vec<ObjectVersion>>, StoreError> {
-    self.inner.list_versions(prefix).await
+  async fn list_versions(
+    &self,
+    prefix: &str,
+    after: Option<&VersionMarker>,
+  ) -> Result<Option<VersionsPage>, StoreError> {
+    self.inner.list_versions(prefix, after).await
   }
 
   async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> {

donut-sync/src/s3.rs

@@ -12,7 +12,7 @@ use crate::jsval::js_number;
 use crate::sigv4::{self, Credentials};
 use crate::store::{
   ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListObject, ListOutput,
-  ObjectVersion, PendingUpload, Store, StoreError, Upload,
+  ObjectVersion, PendingUpload, Store, StoreError, Upload, VersionMarker, VersionsPage,
 };
 use crate::time;
 
@@ -557,11 +557,11 @@ fn head_of(headers: &reqwest::header::HeaderMap) -> HeadOutput {
   }
 }
 
-type VersionsPage = (Vec<ObjectVersion>, Option<(String, String)>);
+type ParsedVersions = (Vec<ObjectVersion>, Option<VersionMarker>);
 
 /// A ListObjectVersions reply. `None` when the store answered with something
 /// else, as stores without versions do.
-fn parse_versions(body: &[u8]) -> Result<Option<VersionsPage>, StoreError> {
+fn parse_versions(body: &[u8]) -> Result<Option<ParsedVersions>, StoreError> {
   let text = std::str::from_utf8(body)
     .map_err(|e| StoreError::Transport(format!("ListObjectVersions reply is not UTF-8: {e}")))?;
   let text = hide_control_references(text);
@@ -735,12 +735,17 @@ impl Store for S3Store {
       .map(|_| ())
   }
 
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+  async fn delete_object(&self, key: &str) -> Result<Option<String>, StoreError> {
     let path = self.location.object_path(key);
-    self
+    let reply = self
       .expect_success(Method::DELETE, &path, &[], &[], &[])
-      .await
-      .map(|_| ())
+      .await?;
+    Ok(
+      reply
+        .headers
+        .get("x-amz-version-id")
+        .map(|v| latin1(v.as_bytes())),
+    )
   }
 
   async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> {
@@ -781,39 +786,37 @@ impl Store for S3Store {
     self.copy(from, to, spec).await
   }
 
-  async fn list_versions(&self, prefix: &str) -> Result<Option<Vec<ObjectVersion>>, StoreError> {
+  async fn list_versions(
+    &self,
+    prefix: &str,
+    after: Option<&VersionMarker>,
+  ) -> Result<Option<VersionsPage>, StoreError> {
     let path = self.location.bucket_path();
-    let mut out = Vec::new();
-    let mut marker: Option<(String, String)> = None;
-    loop {
-      let mut query = vec![
-        ("versions".to_string(), String::new()),
-        ("prefix".to_string(), prefix.to_string()),
-      ];
-      if let Some((key, id)) = &marker {
-        query.push(("key-marker".to_string(), key.clone()));
-        query.push(("version-id-marker".to_string(), id.clone()));
-      }
-      let reply = match self
-        .expect_success(Method::GET, &path, &query, &[], &[])
-        .await
-      {
-        Ok(reply) => reply,
-        Err(StoreError::NotFound) => return Ok(None),
-        Err(StoreError::Service { status, .. }) if status == 501 || status == 400 => {
-          return Ok(None);
-        }
-        Err(e) => return Err(e),
-      };
-      let Some((versions, next)) = parse_versions(&reply.body)? else {
-        return Ok(None);
-      };
-      out.extend(versions);
-      match next {
-        Some(next) if Some(&next) != marker.as_ref() => marker = Some(next),
-        _ => return Ok(Some(out)),
-      }
+    let mut query = vec![
+      ("versions".to_string(), String::new()),
+      ("prefix".to_string(), prefix.to_string()),
+    ];
+    if let Some((key, id)) = after {
+      query.push(("key-marker".to_string(), key.clone()));
+      query.push(("version-id-marker".to_string(), id.clone()));
     }
+    let reply = match self
+      .expect_success(Method::GET, &path, &query, &[], &[])
+      .await
+    {
+      Ok(reply) => reply,
+      Err(StoreError::NotFound) => return Ok(None),
+      Err(StoreError::Service { status, .. }) if status == 501 || status == 400 => {
+        return Ok(None);
+      }
+      Err(e) => return Err(e),
+    };
+    Ok(
+      parse_versions(&reply.body)?.map(|(versions, next)| VersionsPage {
+        versions,
+        next: next.filter(|next| Some(next) != after),
+      }),
+    )
   }
 
   async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> {

donut-sync/src/sealed_store.rs

@@ -12,8 +12,8 @@ use async_trait::async_trait;
 
 use crate::at_rest::{self, Keyring, MARKER};
 use crate::store::{
-  ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, ObjectVersion,
-  PendingUpload, Store, StoreError, Upload,
+  ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store,
+  StoreError, Upload, VersionMarker, VersionsPage,
 };
 
 pub struct SealedStore {
@@ -96,7 +96,7 @@ impl Store for SealedStore {
       .await
   }
 
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+  async fn delete_object(&self, key: &str) -> Result<Option<String>, StoreError> {
     self.inner.delete_object(key).await
   }
 
@@ -146,8 +146,12 @@ impl Store for SealedStore {
       .await
   }
 
-  async fn list_versions(&self, prefix: &str) -> Result<Option<Vec<ObjectVersion>>, StoreError> {
-    self.inner.list_versions(prefix).await
+  async fn list_versions(
+    &self,
+    prefix: &str,
+    after: Option<&VersionMarker>,
+  ) -> Result<Option<VersionsPage>, StoreError> {
+    self.inner.list_versions(prefix, after).await
   }
 
   async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> {

donut-sync/src/sealer.rs

@@ -18,7 +18,7 @@ use serde_json::{Value, json};
 
 use crate::at_rest::Keyring;
 use crate::log;
-use crate::migrate::{Migration, Outcome, TMP_PREFIX, honors_if_match, purge};
+use crate::migrate::{Migration, Outcome, Settle, TMP_PREFIX, honors_if_match, purge};
 use crate::sealed_store::SealedStore;
 use crate::store::{ListInput, Store, StoreError};
 use crate::time;
@@ -45,6 +45,61 @@ const ABANDONED_MS: i64 = 4 * 3600 * 1000;
 const MAX_FAILED: usize = 1000;
 const CONCURRENCY: usize = 4;
 
+/// How often the keys handed direct URLs are written as one record. A crash
+/// loses at most this window; the hourly backstop pass covers it.
+const FLUSH_EVERY: Duration = Duration::from_secs(10);
+
+/// Keys handed direct storage URLs, gathered in memory until the next flush,
+/// so a presign costs no storage write.
+#[derive(Default)]
+pub struct DirectUploads {
+  pending: std::sync::Mutex<(Vec<String>, i64)>,
+}
+
+impl DirectUploads {
+  pub fn note(&self, keys: &[String], expires_ms: i64) {
+    let mut pending = self.pending.lock().unwrap();
+    pending.0.extend_from_slice(keys);
+    pending.1 = pending.1.max(expires_ms);
+  }
+
+  /// Writes what was noted as one record; `store` is the sealed store.
+  pub async fn flush(&self, store: &dyn Store) -> Result<(), StoreError> {
+    let (keys, expires_ms) = std::mem::take(&mut *self.pending.lock().unwrap());
+    if keys.is_empty() {
+      return Ok(());
+    }
+    let outcome = note_direct_uploads(store, &keys, expires_ms).await;
+    if outcome.is_err() {
+      self.note(&keys, expires_ms);
+    }
+    outcome
+  }
+}
+
+/// Flushes `uploads` every `FLUSH_EVERY` for as long as the server runs.
+pub fn spawn_flusher(store: Arc<dyn Store>, uploads: Arc<DirectUploads>) {
+  tokio::spawn(async move {
+    let mut failing = false;
+    loop {
+      tokio::time::sleep(FLUSH_EVERY).await;
+      match uploads.flush(store.as_ref()).await {
+        Ok(()) if failing => {
+          failing = false;
+          log::log(CONTEXT, "Direct uploads noted again");
+        }
+        Ok(()) => {}
+        Err(e) if !failing => {
+          failing = true;
+          // The keys stay in memory for the next flush.
+          log::warn(CONTEXT, &format!("Direct uploads not noted err=\"{e}\""));
+        }
+        Err(_) => {}
+      }
+    }
+  });
+}
+
 /// Notes `keys`, which devices may write in plaintext through direct storage
 /// URLs until `expires_ms`. `store` is the sealed store: the record is sealed.
 pub async fn note_direct_uploads(
@@ -82,6 +137,17 @@ pub struct Failed {
   pub reason: String,
   /// When it first failed.
   pub at: i64,
+  /// Set when the rewrite is in place but a device's version may lie under
+  /// it: a later pass settles it with these ids, and does not rewrite again.
+  #[serde(default, skip_serializing_if = "Option::is_none")]
+  pub settle: Option<Settle>,
+}
+
+/// A key that failed in a pass.
+pub struct Failure {
+  pub key: String,
+  pub reason: String,
+  pub settle: Option<Settle>,
 }
 
 /// What the sealer keeps between passes, sealed in `STATE`.
@@ -94,18 +160,27 @@ pub struct State {
 }
 
 impl State {
-  /// `failures` from a pass over `checked`: keys that sealed leave the list,
-  /// keys that failed join it (keeping when they first failed).
-  fn note(&mut self, checked: &[String], failures: Vec<(String, String)>, now_ms: i64) {
+  /// `failures` from a pass that retried `retried`: retried keys that
+  /// worked leave the list, keys that failed join it (keeping when they
+  /// first failed).
+  pub fn note(&mut self, retried: &[String], failures: Vec<Failure>, now_ms: i64) {
     let first: Vec<(String, i64)> = self.failed.iter().map(|f| (f.key.clone(), f.at)).collect();
-    self.failed.retain(|f| !checked.contains(&f.key));
-    for (key, reason) in failures {
+    self.failed.retain(|f| !retried.contains(&f.key));
+    for failure in failures {
       let at = first
         .iter()
-        .find(|(k, _)| *k == key)
+        .find(|(k, _)| *k == failure.key)
         .map_or(now_ms, |(_, at)| *at);
-      self.failed.retain(|f| f.key != key);
-      self.failed.insert(0, Failed { key, reason, at });
+      self.failed.retain(|f| f.key != failure.key);
+      self.failed.insert(
+        0,
+        Failed {
+          key: failure.key,
+          reason: failure.reason,
+          at,
+          settle: failure.settle,
+        },
+      );
     }
     self.failed.truncate(MAX_FAILED);
   }
@@ -120,6 +195,61 @@ pub async fn read_state(store: &dyn Store) -> Result<State, StoreError> {
   }
 }
 
+pub async fn save_state(store: &dyn Store, state: &State) -> Result<(), StoreError> {
+  store
+    .put_object(
+      STATE,
+      serde_json::to_vec(state).expect("the state serializes"),
+      "application/json",
+      &[],
+    )
+    .await
+}
+
+/// Tries the failed keys again: a settle where the rewrite is in place, a
+/// whole rewrite otherwise. Returns the ones that still fail.
+pub(crate) async fn retry_failed(migration: &Migration, failed: &[Failed]) -> Vec<Failure> {
+  let mut still = Vec::new();
+  for failed in failed {
+    match &failed.settle {
+      Some(settle) => {
+        if let Err(reason) = migration.settle(&failed.key, settle).await {
+          still.push(Failure {
+            key: failed.key.clone(),
+            reason,
+            settle: Some(settle.clone()),
+          });
+        }
+      }
+      None => {
+        if let Some(failure) = failure_of(
+          failed.key.clone(),
+          migration.object(&failed.key, None).await,
+        ) {
+          still.push(failure);
+        }
+      }
+    }
+  }
+  still
+}
+
+fn failure_of(key: String, outcome: Outcome) -> Option<Failure> {
+  match outcome {
+    Outcome::Failed(reason) => Some(Failure {
+      key,
+      reason,
+      settle: None,
+    }),
+    Outcome::Unsettled { settle, reason } => Some(Failure {
+      key,
+      reason,
+      settle: Some(settle),
+    }),
+    _ => None,
+  }
+}
+
 #[derive(Default, Debug, PartialEq)]
 pub struct PassReport {
   pub records: u64,
@@ -145,12 +275,12 @@ impl Sealer {
     })
   }
 
-  /// Seals `keys`; returns the ones that failed, with why.
+  /// Seals `keys`; returns the ones that failed.
   async fn seal_all(
     &self,
     keys: Vec<(String, Option<i64>)>,
     report: &mut PassReport,
-  ) -> Vec<(String, String)> {
+  ) -> Vec<Failure> {
     let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys)
       .map(|(key, listed)| async move {
         let outcome = self.migration.object(&key, listed).await;
@@ -162,149 +292,155 @@ impl Sealer {
     let mut failures = Vec::new();
     for (key, outcome) in outcomes {
       report.checked += 1;
-      match outcome {
-        Outcome::Sealed(_) | Outcome::Rotated(_) => report.sealed += 1,
-        Outcome::Failed(err) => {
-          log::warn(
-            CONTEXT,
-            &format!("Object not sealed key={key} err=\"{err}\""),
-          );
-          report.failed += 1;
-          failures.push((key, err));
-        }
-        Outcome::Current | Outcome::Changed | Outcome::Unsealed(_) => {}
+      if matches!(outcome, Outcome::Sealed(_) | Outcome::Rotated(_)) {
+        report.sealed += 1;
+      }
+      if let Some(failure) = failure_of(key, outcome) {
+        log::warn(
+          CONTEXT,
+          &format!(
+            "Object not sealed key={} err=\"{}\"",
+            failure.key, failure.reason
+          ),
+        );
+        report.failed += 1;
+        failures.push(failure);
       }
     }
     failures
   }
 
-  async fn list(&self, prefix: Option<&str>) -> Result<Vec<(String, Option<i64>)>, StoreError> {
-    let mut out = Vec::new();
-    let mut token = None;
-    loop {
-      let page = self
-        .raw
-        .list_objects_v2(&ListInput {
-          prefix: prefix.map(str::to_string),
-          max_keys: Some("1000".to_string()),
-          continuation_token: token.clone(),
-          ..ListInput::default()
-        })
-        .await?;
-      out.extend(
-        page
-          .contents
-          .into_iter()
-          .map(|object| (object.key, object.last_modified_ms)),
-      );
-      token = page.next_continuation_token.filter(|_| page.is_truncated);
-      if token.is_none() {
-        return Ok(out);
-      }
-    }
-  }
-
-  async fn save_state(&self, state: &State) -> Result<(), StoreError> {
-    self
-      .sealed
-      .put_object(
-        STATE,
-        serde_json::to_vec(state).expect("the state serializes"),
-        "application/json",
-        &[],
-      )
-      .await
+  /// One page of the listing under `prefix`.
+  async fn page(
+    &self,
+    prefix: Option<&str>,
+    token: Option<String>,
+  ) -> Result<(Vec<(String, Option<i64>)>, Option<String>), StoreError> {
+    let page = self
+      .raw
+      .list_objects_v2(&ListInput {
+        prefix: prefix.map(str::to_string),
+        max_keys: Some("1000".to_string()),
+        continuation_token: token,
+        ..ListInput::default()
+      })
+      .await?;
+    let next = page.next_continuation_token.filter(|_| page.is_truncated);
+    Ok((
+      page
+        .contents
+        .into_iter()
+        .map(|object| (object.key, object.last_modified_ms))
+        .collect(),
+      next,
+    ))
   }
 
   /// Seals the keys of each record whose URLs are dead, then drops the
   /// record. Keys that fail are kept in the state for the backstop.
   pub async fn pending_pass(&self, now_ms: i64) -> Result<PassReport, StoreError> {
     let mut report = PassReport::default();
-    let mut checked = Vec::new();
     let mut failures = Vec::new();
-    for (record, _) in self.list(Some(PENDING)).await? {
-      report.records += 1;
-      if !record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS) {
-        continue;
-      }
-      let keys: Vec<String> = match self.sealed.get_object_text(&record).await {
-        Ok(text) => serde_json::from_str::<Value>(&text)
-          .ok()
-          .and_then(|v| v["keys"].as_array().cloned())
-          .unwrap_or_default()
-          .iter()
-          .filter_map(|k| k.as_str().map(str::to_string))
-          .collect(),
-        Err(StoreError::NotFound) => continue,
-        Err(e) => {
-          log::warn(
-            CONTEXT,
-            &format!("Record unreadable key={record} err=\"{e}\""),
-          );
-          report.failed += 1;
+    let mut token = None;
+    loop {
+      let (records, next) = self.page(Some(PENDING), token).await?;
+      for (record, _) in records {
+        report.records += 1;
+        if !record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS) {
           continue;
         }
-      };
-      failures.extend(
-        self
-          .seal_all(
-            keys.iter().map(|k| (k.clone(), None)).collect(),
-            &mut report,
-          )
-          .await,
-      );
-      checked.extend(keys);
-      purge(self.raw.as_ref(), &record).await?;
-      report.records_done += 1;
+        let keys: Vec<String> = match self.sealed.get_object_text(&record).await {
+          Ok(text) => serde_json::from_str::<Value>(&text)
+            .ok()
+            .and_then(|v| v["keys"].as_array().cloned())
+            .unwrap_or_default()
+            .iter()
+            .filter_map(|k| k.as_str().map(str::to_string))
+            .collect(),
+          Err(StoreError::NotFound) => continue,
+          Err(e) => {
+            log::warn(
+              CONTEXT,
+              &format!("Record unreadable key={record} err=\"{e}\""),
+            );
+            report.failed += 1;
+            continue;
+          }
+        };
+        failures.extend(
+          self
+            .seal_all(keys.into_iter().map(|k| (k, None)).collect(), &mut report)
+            .await,
+        );
+        purge(self.raw.as_ref(), &record).await?;
+        report.records_done += 1;
+      }
+      token = next;
+      if token.is_none() {
+        break;
+      }
     }
     if !failures.is_empty() {
       let mut state = read_state(&self.sealed).await?;
-      state.note(&checked, failures, now_ms);
-      self.save_state(&state).await?;
+      state.note(&[], failures, now_ms);
+      save_state(&self.sealed, &state).await?;
     }
     Ok(report)
   }
 
-  /// Seals everything written since the last pass and retries what failed,
-  /// then cleans up after interrupted rewrites and uploads.
+  /// Retries what failed, seals everything written since the last pass,
+  /// page by page, then cleans up after interrupted rewrites and uploads.
   pub async fn backstop(&self, now_ms: i64) -> Result<PassReport, StoreError> {
     let mut state = read_state(&self.sealed).await?;
     let since = state.backstop_since.unwrap_or(now_ms - FIRST_BACKSTOP_MS);
-    let listed = self.list(None).await?;
-    let mut candidates: Vec<(String, Option<i64>)> = listed
-      .iter()
-      .filter(|(key, modified)| {
-        !key.starts_with(INTERNAL_ROOT)
-          && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS)
-      })
-      .cloned()
-      .collect();
-    for failed in &state.failed {
-      if !candidates.iter().any(|(key, _)| *key == failed.key) {
-        candidates.push((failed.key.clone(), None));
+    let mut report = PassReport::default();
+    let retried: Vec<String> = state.failed.iter().map(|f| f.key.clone()).collect();
+    let mut failures = retry_failed(&self.migration, &state.failed).await;
+    report.checked += retried.len() as u64;
+    report.failed += failures.len() as u64;
+    let mut token = None;
+    loop {
+      let (objects, next) = self.page(None, token).await?;
+      let recent: Vec<(String, Option<i64>)> = objects
+        .into_iter()
+        .filter(|(key, modified)| {
+          !key.starts_with(INTERNAL_ROOT)
+            && !retried.contains(key)
+            && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS)
+        })
+        .collect();
+      failures.extend(self.seal_all(recent, &mut report).await);
+      token = next;
+      if token.is_none() {
+        break;
       }
     }
-    let checked: Vec<String> = candidates.iter().map(|(key, _)| key.clone()).collect();
-    let mut report = PassReport::default();
-    let failures = self.seal_all(candidates, &mut report).await;
     // The pass moves on even when some keys fail: they stay in the list.
-    state.note(&checked, failures, now_ms);
+    state.note(&retried, failures, now_ms);
     state.backstop_since = Some(now_ms);
-    self.save_state(&state).await?;
+    save_state(&self.sealed, &state).await?;
 
-    for (key, modified) in &listed {
-      if key.starts_with(TMP_PREFIX) && modified.is_some_and(|ms| now_ms - ms > TMP_STALE_MS) {
-        match purge(self.raw.as_ref(), key).await {
-          Ok(()) => log::log(
-            CONTEXT,
-            &format!("Stale temporary object removed key={key}"),
-          ),
-          Err(e) => log::warn(
-            CONTEXT,
-            &format!("Stale temporary object not removed key={key} err=\"{e}\""),
-          ),
+    let mut token = None;
+    loop {
+      let (temporary, next) = self.page(Some(TMP_PREFIX), token).await?;
+      for (key, modified) in temporary {
+        if modified.is_some_and(|ms| now_ms - ms > TMP_STALE_MS) {
+          match purge(self.raw.as_ref(), &key).await {
+            Ok(()) => log::log(
+              CONTEXT,
+              &format!("Stale temporary object removed key={key}"),
+            ),
+            Err(e) => log::warn(
+              CONTEXT,
+              &format!("Stale temporary object not removed key={key} err=\"{e}\""),
+            ),
+          }
         }
       }
+      token = next;
+      if token.is_none() {
+        break;
+      }
     }
     match self.raw.list_multipart("").await {
       Ok(uploads) => {
@@ -548,6 +684,28 @@ mod tests {
     assert!(raw.list_multipart("").await.unwrap().is_empty());
   }
 
+  #[tokio::test]
+  async fn presigns_are_noted_as_one_record_per_flush() {
+    let (raw, sealed, _) = setup(MemoryStore::default()).await;
+    let direct = DirectUploads::default();
+    direct.flush(&sealed).await.unwrap();
+    assert!(raw.keys().is_empty(), "nothing noted, nothing written");
+    direct.note(&["a".to_string()], 5_000);
+    direct.note(&["b".to_string(), "c".to_string()], 7_000);
+    direct.flush(&sealed).await.unwrap();
+    let records: Vec<String> = raw
+      .keys()
+      .into_iter()
+      .filter(|k| k.starts_with(PENDING))
+      .collect();
+    assert_eq!(records.len(), 1);
+    assert_eq!(record_expiry(&records[0]), Some(7_000));
+    let body = sealed.get_object_text(&records[0]).await.unwrap();
+    assert_eq!(body, r#"{"keys":["a","b","c"]}"#);
+    direct.flush(&sealed).await.unwrap();
+    assert_eq!(raw.keys().len(), 1);
+  }
+
   #[test]
   fn records_name_their_expiry() {
     assert_eq!(

donut-sync/src/slots.rs

@@ -11,9 +11,9 @@ use std::time::Duration;
 
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
-const MAX_TRANSFERS: usize = 64;
-const MAX_PER_NAMESPACE: usize = 8;
-pub const SLOT_WAIT: Duration = Duration::from_secs(20);
+const MAX_TRANSFERS: usize = 192;
+const MAX_PER_NAMESPACE: usize = 32;
+pub const SLOT_WAIT: Duration = Duration::from_secs(60);
 const MAX_HOLD: Duration = Duration::from_secs(600);
 /// Accounts whose semaphores are kept while idle.
 const NAMESPACES_KEPT: usize = 10_000;
@@ -110,7 +110,7 @@ mod tests {
 
   #[tokio::test]
   async fn one_account_cannot_take_every_slot() {
-    let slots = Slots::new(10, MAX_HOLD);
+    let slots = Slots::new(MAX_PER_NAMESPACE + 2, MAX_HOLD);
     let mut held = Vec::new();
     for _ in 0..MAX_PER_NAMESPACE {
       held.push(slots.acquire("users/a/x", WAIT).await.unwrap());
@@ -120,7 +120,7 @@ mod tests {
     held.push(slots.acquire("users/c/x", WAIT).await.unwrap());
     assert!(
       slots.acquire("users/d/x", WAIT).await.is_none(),
-      "all 10 in use"
+      "every slot in use"
     );
     held.pop();
     assert!(slots.acquire("users/d/x", WAIT).await.is_some());

donut-sync/src/store.rs

@@ -115,7 +115,8 @@ pub trait Store: Send + Sync {
     content_type: &str,
     metadata: &[(String, String)],
   ) -> Result<(), StoreError>;
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError>;
+  /// Returns the delete marker's version id, where the bucket keeps versions.
+  async fn delete_object(&self, key: &str) -> Result<Option<String>, StoreError>;
   /// Deletes up to 1000 keys in one request; per-key failures are ignored.
   async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError>;
   /// Server-side copy, metadata included.
@@ -131,9 +132,14 @@ pub trait Store: Send + Sync {
     to: &str,
     spec: CopySpec<'_>,
   ) -> Result<Copied, StoreError>;
-  /// Every version and delete marker under `prefix`, newest first within a
-  /// key. `None` when the store keeps no versions or cannot list them.
-  async fn list_versions(&self, prefix: &str) -> Result<Option<Vec<ObjectVersion>>, StoreError>;
+  /// One page of the versions and delete markers under `prefix`, newest first
+  /// within a key, from `after` on. `None` when the store keeps no versions
+  /// or cannot list them.
+  async fn list_versions(
+    &self,
+    prefix: &str,
+    after: Option<&VersionMarker>,
+  ) -> Result<Option<VersionsPage>, StoreError>;
   async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError>;
   /// The first `length` bytes as stored.
   async fn get_range(&self, key: &str, length: u64) -> Result<Vec<u8>, StoreError>;
@@ -196,6 +202,35 @@ pub struct Copied {
   pub version_id: Option<String>,
 }
 
+/// Where the next page of versions starts: a key and a version id.
+pub type VersionMarker = (String, String);
+
+#[derive(Debug, Clone, Default)]
+pub struct VersionsPage {
+  pub versions: Vec<ObjectVersion>,
+  pub next: Option<VersionMarker>,
+}
+
+/// Every version of `key`, newest first; `None` without versions.
+pub async fn versions_of(
+  store: &dyn Store,
+  key: &str,
+) -> Result<Option<Vec<ObjectVersion>>, StoreError> {
+  let mut out = Vec::new();
+  let mut after: Option<VersionMarker> = None;
+  loop {
+    let Some(page) = store.list_versions(key, after.as_ref()).await? else {
+      return Ok(None);
+    };
+    out.extend(page.versions.into_iter().filter(|v| v.key == key));
+    match page.next {
+      // Keys after `key` itself only share its prefix.
+      Some(next) if next.0 == key && Some(&next) != after.as_ref() => after = Some(next),
+      _ => return Ok(Some(out)),
+    }
+  }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct ObjectVersion {
   pub key: String,

donut-sync/src/sync.rs

@@ -160,6 +160,7 @@ pub struct SyncService {
   cloud: bool,
   backend: Option<Arc<Backend>>,
   group_cache: Mutex<HashMap<String, (Group, Instant)>>,
+  direct_uploads: Option<Arc<crate::sealer::DirectUploads>>,
 }
 
 pub struct SyncOptions {
@@ -175,6 +176,11 @@ pub struct SyncOptions {
 
 impl SyncService {
   pub fn new(store: Arc<dyn Store>, options: SyncOptions) -> Self {
+    let direct_uploads = options
+      .transfer
+      .as_ref()
+      .filter(|signer| !signer.keyring.unsealing())
+      .map(|_| Arc::new(crate::sealer::DirectUploads::default()));
     Self {
       store,
       presigner: options.presigner,
@@ -184,6 +190,7 @@ impl SyncService {
       cloud: options.cloud,
       backend: options.backend,
       group_cache: Mutex::new(HashMap::new()),
+      direct_uploads,
     }
   }
 
@@ -215,23 +222,17 @@ impl SyncService {
 
   /// Notes the keys handed direct storage URLs, so the sealer seals what
   /// older apps write there.
-  async fn note_direct_uploads(&self, keys: &[String], now: i64, expires_in: f64) {
-    let sealing = self
-      .transfer
-      .as_ref()
-      .is_some_and(|signer| !signer.keyring.unsealing());
-    if !sealing || keys.is_empty() {
-      return;
-    }
-    if let Err(e) =
-      crate::sealer::note_direct_uploads(self.store.as_ref(), keys, expires_ms(now, expires_in))
-        .await
-    {
-      // The hourly backstop pass still finds these objects.
-      log::warn(CONTEXT, &format!("Direct uploads not noted err=\"{e}\""));
+  fn note_direct_uploads(&self, keys: &[String], now: i64, expires_in: f64) {
+    if let Some(direct) = &self.direct_uploads {
+      direct.note(keys, expires_ms(now, expires_in));
     }
   }
 
+  /// Set while sealing: the keys handed direct URLs, for the sealer.
+  pub fn direct_uploads(&self) -> Option<&Arc<crate::sealer::DirectUploads>> {
+    self.direct_uploads.as_ref()
+  }
+
   fn get_url(&self, origin: &str, key: &str, expires_in: f64, now: i64) -> String {
     match &self.transfer {
       Some(signer) => signer.url(origin, "GET", key, expires_ms(now, expires_in), &[]),
@@ -534,9 +535,7 @@ impl SyncService {
       .presigner
       .put_url(scoped, expires_in, metadata_or_none, now);
     let transfer = self.transfer_json(origin, scoped, expires_in, metadata_or_none, now);
-    self
-      .note_direct_uploads(&[scoped.to_string()], now, expires_in)
-      .await;
+    self.note_direct_uploads(&[scoped.to_string()], now, expires_in);
 
     if ctx.mode == Mode::Cloud && field_starts_with(raw_key, "profiles/")? {
       self.report_profile_usage_async(ctx);
@@ -751,7 +750,7 @@ impl SyncService {
       out_items.push(Value::Object(item));
       direct.push(scoped.to_string());
     }
-    self.note_direct_uploads(&direct, now, expires_in).await;
+    self.note_direct_uploads(&direct, now, expires_in);
 
     if ctx.mode == Mode::Cloud {
       let mut any_profile = false;
@@ -1222,11 +1221,10 @@ impl SyncService {
           .await
           .map_err(store_error)?;
         let mut last_modified = first.contents.first().and_then(|o| o.last_modified_ms);
-        // A sealed store reports a migrated object's time from before the
-        // migration, which only a HEAD reads.
-        if self.transfer.is_some()
-          && let Some(object) = first.contents.first()
-        {
+        // A migrated object's time from before the migration is in its
+        // metadata, which only a HEAD reads; it stays there after the key is
+        // removed.
+        if let Some(object) = first.contents.first() {
           last_modified = self
             .store
             .head_object(&object.key)

donut-sync/src/sync_tests.rs

@@ -719,3 +719,39 @@ async fn cleanup_orders_migrated_profiles_by_their_time_before_the_migration() {
   let out = service.cleanup_excess_profiles("u1", 2).await.unwrap();
   assert_eq!(out["deletedProfiles"], json!(["pa", "pb"]));
 }
+
+/// After a rollback (sealed, unsealed, key removed) profile ages are still
+/// the ones from before any rewrite.
+#[tokio::test]
+async fn cleanup_keeps_the_original_order_after_a_rollback() {
+  let raw = Arc::new(MemoryStore::default());
+  for id in ["pd", "pc", "pb", "pa"] {
+    raw.set(&format!("users/u1/profiles/{id}/metadata.json"), "{}");
+  }
+  let sealing = Arc::new(
+    crate::at_rest::Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap(),
+  );
+  let options = crate::migrate::Options {
+    concurrency: 1,
+    ..Default::default()
+  };
+  crate::migrate::migrate(raw.clone(), sealing, &options)
+    .await
+    .unwrap();
+  let unsealing = Arc::new(
+    crate::at_rest::Keyring::parse("unseal,k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")
+      .unwrap(),
+  );
+  let unseal = crate::migrate::Options {
+    unseal: true,
+    ..options
+  };
+  let report = crate::migrate::migrate(raw.clone(), unsealing, &unseal)
+    .await
+    .unwrap();
+  assert_eq!(report.unsealed, 4);
+  // The key is gone: the plain service, as the server runs without it.
+  let service = service_on(crate::served_store(raw.clone(), None), None);
+  let out = service.cleanup_excess_profiles("u1", 2).await.unwrap();
+  assert_eq!(out["deletedProfiles"], json!(["pa", "pb"]));
+}

donut-sync/src/transfer_parts.rs

@@ -20,8 +20,9 @@
 //! Sealing a part twice under the same nonces is safe only for the same
 //! bytes, so a part is read whole and hashed before anything reaches storage
 //! (`Ledger`): a part may be sent again with the same bytes, never with
-//! others. The ledger is in memory, so after a restart the parts of an upload
-//! started before it are refused, and the app starts the upload again.
+//! others. The ledger is in memory, so only the process that started an
+//! upload takes its parts: after a restart, or while a deploy runs the old
+//! and new process side by side, the app is told to start the upload again.
 
 use base64::Engine;
 use bytes::Bytes;
@@ -75,7 +76,8 @@ struct Token {
   /// Whether the parts are sealed: not when the server was unsealing at the
   /// start (the object then has no marker either).
   sealed: bool,
-  issued_ms: i64,
+  /// The process that started the upload: its ledger holds the parts.
+  instance: String,
 }
 
 /// Each part's hash and its upload's expiry, by upload id and part number.
@@ -83,7 +85,7 @@ type PartHashes = HashMap<(String, u64), ([u8; 32], i64)>;
 
 /// The hash of every part sent so far, while its upload lasts.
 pub struct Ledger {
-  started_ms: i64,
+  instance: String,
   parts: Mutex<PartHashes>,
   buffers: Arc<Semaphore>,
 }
@@ -91,7 +93,7 @@ pub struct Ledger {
 impl Default for Ledger {
   fn default() -> Self {
     Self {
-      started_ms: time::now_ms(),
+      instance: uuid::Uuid::new_v4().simple().to_string(),
       parts: Mutex::new(HashMap::new()),
       buffers: Arc::new(Semaphore::new(PART_BUFFERS)),
     }
@@ -162,6 +164,14 @@ impl Token {
   }
 }
 
+fn interrupted() -> Response<ResBody> {
+  s3_error(
+    409,
+    "UploadInterrupted",
+    "The server restarted during this upload. Start it again.",
+  )
+}
+
 fn invalid(code: &str, message: &str) -> Response<ResBody> {
   s3_error(400, code, message)
 }
@@ -211,7 +221,7 @@ pub async fn start(
     part_size: PART_SIZE,
     expires_ms: url_expires_ms + UPLOAD_GRACE_MS,
     sealed: !signer.keyring.unsealing(),
-    issued_ms: time::now_ms(),
+    instance: signer.parts.instance.clone(),
   };
   json_response(
     200,
@@ -242,6 +252,12 @@ pub async fn serve(
   match (&parts.method, signed.op.as_deref()) {
     (&Method::PUT, None) => part(store, signer, &token, signed.part.as_deref(), parts, body).await,
     (&Method::POST, Some("complete")) => {
+      // Its parts went through another process's ledger; a mode change
+      // (sealing or not) is a new process too.
+      if token.instance != signer.parts.instance {
+        drain(body).await;
+        return interrupted();
+      }
       let response = complete(store, &token, body).await;
       if response.status().is_success() {
         signer.parts.forget(&token.upload_id);
@@ -296,13 +312,9 @@ async fn part(
       &format!("Part {number} must be {expected} bytes, not {length}."),
     );
   }
-  if token.issued_ms < signer.parts.started_ms {
+  if token.instance != signer.parts.instance {
     drain(body).await;
-    return s3_error(
-      409,
-      "UploadInterrupted",
-      "The server restarted during this upload. Start it again.",
-    );
+    return interrupted();
   }
   let Ok(Ok(_buffer)) =
     tokio::time::timeout(SLOT_WAIT, signer.parts.buffers.clone().acquire_owned()).await
@@ -424,7 +436,7 @@ mod tests {
       part_size: PART_SIZE,
       expires_ms: 1000,
       sealed: true,
-      issued_ms: 0,
+      instance: "i".to_string(),
     }
   }
 

donut-sync/src/verify.rs

@@ -273,6 +273,34 @@ async fn open_fully(
   Ok(())
 }
 
+/// The bucket's older versions under `prefix`, page by page; `None`
+/// without versions.
+async fn noncurrent(store: &dyn Store, prefix: &str) -> Result<Option<Noncurrent>, String> {
+  let mut n = Noncurrent::default();
+  let mut after = None;
+  loop {
+    let Some(page) = store
+      .list_versions(prefix, after.as_ref())
+      .await
+      .map_err(|e| format!("Versions not listed: {e}"))?
+    else {
+      return Ok(None);
+    };
+    for version in page.versions {
+      if version.delete_marker {
+        n.delete_markers += 1;
+      } else if !version.is_latest {
+        n.versions += 1;
+        n.bytes += version.size;
+      }
+    }
+    after = page.next;
+    if after.is_none() {
+      return Ok(Some(n));
+    }
+  }
+}
+
 pub async fn verify(
   store: Arc<dyn Store>,
   keyring: Arc<Keyring>,
@@ -438,22 +466,7 @@ pub async fn verify(
     }
     Err(e) => return Err(format!("Sealer state unreadable: {e}")),
   }
-  report.noncurrent = store
-    .list_versions(options.prefix.as_deref().unwrap_or(""))
-    .await
-    .map_err(|e| format!("Versions not listed: {e}"))?
-    .map(|versions| {
-      let mut n = Noncurrent::default();
-      for version in versions.iter().filter(|v| !v.is_latest || v.delete_marker) {
-        if version.delete_marker {
-          n.delete_markers += 1;
-        } else {
-          n.versions += 1;
-          n.bytes += version.size;
-        }
-      }
-      n
-    });
+  report.noncurrent = noncurrent(store.as_ref(), options.prefix.as_deref().unwrap_or("")).await?;
   let line = format!(
     "Verify done complete={} {} elapsed_ms={}",
     report.complete(),
@@ -605,6 +618,7 @@ mod tests {
         key: "x".into(),
         reason: "r".into(),
         at: 1,
+        settle: None,
       }],
     };
     sealed

donut-sync/tests/at_rest.rs

@@ -4,7 +4,7 @@
 
 mod common;
 
-use common::{SYNC_TOKEN, Server, bucket, http, post, raw_store, run_command, storage_env};
+use common::{SYNC_TOKEN, Server, fresh_bucket, http, post, raw_store, run_command, storage_env};
 use donut_sync::store::{Store, Upload, collect, once};
 use serde_json::{Value, json};
 
@@ -16,7 +16,7 @@ fn sample(len: usize) -> Vec<u8> {
 }
 
 async fn sealed_server(name: &str, keys: &str) -> (Server, String) {
-  let bucket = bucket(name);
+  let bucket = fresh_bucket(name).await;
   (
     Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(keys))]).await,
     bucket,
@@ -47,7 +47,7 @@ async fn upload(server: &Server, key: &str, body: Vec<u8>, updated_at: Option<&s
     presign["url"]
       .as_str()
       .unwrap()
-      .starts_with(common::S3_ENDPOINT)
+      .starts_with(&common::storage().endpoint)
   );
   assert_eq!(presign["transfer"]["partSize"], json!(8 * 1024 * 1024));
   let url = presign["transfer"]["url"].as_str().unwrap().to_string();
@@ -123,7 +123,7 @@ async fn transfers_go_through_the_server_and_are_sealed_in_storage() {
   // Older apps still upload to storage directly, so its address is named.
   assert_eq!(
     readyz,
-    json!({ "status": "ready", "s3": true, "storageEndpoint": common::S3_ENDPOINT })
+    json!({ "status": "ready", "s3": true, "storageEndpoint": &common::storage().endpoint })
   );
   assert!(
     server
@@ -277,13 +277,13 @@ async fn transfer_urls_refuse_what_s3_would_refuse() {
 
 #[tokio::test]
 async fn without_the_key_nothing_changes() {
-  let server = Server::start(&bucket("plain"), &[]).await;
+  let server = Server::start(&fresh_bucket("plain").await, &[]).await;
   let (_, presign) = call(&server, "presign-upload", json!({ "key": "k" })).await;
   assert!(
     presign["url"]
       .as_str()
       .unwrap()
-      .starts_with("http://127.0.0.1:8987/")
+      .starts_with(&format!("{}/", common::storage().endpoint))
   );
   let response = http()
     .get(format!("{}/v1/transfer?key=k", server.url))
@@ -321,7 +321,7 @@ async fn seed_plaintext(bucket: &str, objects: &[(&str, Vec<u8>)]) {
 
 #[tokio::test]
 async fn plaintext_and_sealed_objects_read_side_by_side() {
-  let bucket = bucket("mixed");
+  let bucket = fresh_bucket("mixed").await;
   let old = sample(100_000);
   seed_plaintext(&bucket, &[("proxies/old.json", old.clone())]).await;
   let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await;
@@ -369,7 +369,7 @@ async fn plaintext_and_sealed_objects_read_side_by_side() {
 
 #[tokio::test]
 async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() {
-  let bucket = bucket("migrate");
+  let bucket = fresh_bucket("migrate").await;
   let objects: Vec<(String, Vec<u8>)> = (0..25)
     .map(|i| (format!("profiles/p/files/{i:02}"), sample(i * 9_000)))
     .collect();
@@ -548,7 +548,7 @@ async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() {
 
 #[tokio::test]
 async fn a_bad_key_stops_the_server_at_start() {
-  let storage = storage_env(&bucket("badkey"));
+  let storage = storage_env(&fresh_bucket("badkey").await);
   let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect();
   env.push(("SYNC_TOKEN", SYNC_TOKEN));
   env.push(("SYNC_AT_REST_KEY", "k1:AAAA"));
@@ -766,6 +766,25 @@ async fn older_apps_upload_directly_and_the_sealer_seals_it() {
   // Readable before it is sealed.
   assert_eq!(download(&server, key).await, (200, plain.clone()));
 
+  // The server notes presigned keys in memory and writes them every 10 s.
+  let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+  loop {
+    let pending = raw
+      .list_objects_v2(&donut_sync::store::ListInput {
+        prefix: Some(".donut-sync-at-rest/pending/".to_string()),
+        ..Default::default()
+      })
+      .await
+      .unwrap();
+    if !pending.contents.is_empty() {
+      break;
+    }
+    assert!(
+      std::time::Instant::now() < deadline,
+      "no pending record written"
+    );
+    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+  }
   let keyring = std::sync::Arc::new(donut_sync::at_rest::Keyring::parse(K1).unwrap());
   let sealer = donut_sync::sealer::Sealer::new(raw.clone(), keyring)
     .await
@@ -795,7 +814,7 @@ async fn older_apps_upload_directly_and_the_sealer_seals_it() {
 
 #[tokio::test]
 async fn a_rollback_unseals_everything_and_the_key_can_go() {
-  let bucket = bucket("unseal");
+  let bucket = fresh_bucket("unseal").await;
   let key_sealed = "profiles/p/files/sealed";
   let key_parts = "profiles/p/files/parts";
   let small = sample(70_000);
@@ -887,3 +906,84 @@ async fn a_rollback_unseals_everything_and_the_key_can_go() {
   assert_eq!(after["metadata"]["updated-at"], "3");
   assert!(after["metadata"].get("donut-at-rest").is_none());
 }
+
+/// What settling a rewrite relies on, checked against the store the suite
+/// runs on: version ids on HEAD, copy and delete, versions listed newest
+/// first, copies of a given version, and a copy conditional on the source.
+#[tokio::test]
+async fn the_store_keeps_what_a_settle_needs() {
+  use donut_sync::store::{CopySpec, StoreError, versions_of};
+  let bucket = fresh_bucket("capabilities").await;
+  let raw = raw_store(&bucket);
+  raw.create_bucket().await.ok();
+  raw
+    .put_object("k", b"one".to_vec(), "text/plain", &[])
+    .await
+    .unwrap();
+  let first = raw.head_object("k").await.unwrap();
+  raw
+    .put_object("k", b"two".to_vec(), "text/plain", &[])
+    .await
+    .unwrap();
+  let second = raw.head_object("k").await.unwrap();
+
+  // A copy that names a source ETag the source no longer has.
+  let stale = raw
+    .copy_object_with(
+      "k",
+      "copy",
+      CopySpec {
+        if_match: first.etag.as_deref(),
+        ..CopySpec::default()
+      },
+    )
+    .await;
+  let honored = matches!(stale, Err(StoreError::Service { status: 412, .. }));
+  println!("copy_source_if_match_honored={honored}");
+
+  let Some(history) = versions_of(&raw, "k").await.unwrap() else {
+    println!("versions=unsupported");
+    return;
+  };
+  println!("versions=supported");
+  assert!(
+    honored,
+    "a store that keeps versions must honor a conditional copy"
+  );
+  let (Some(v1), Some(v2)) = (first.version_id.clone(), second.version_id.clone()) else {
+    panic!("HEAD gave no version id: {first:?} {second:?}");
+  };
+  let ids: Vec<&str> = history.iter().map(|v| v.version_id.as_str()).collect();
+  assert_eq!(ids, [v2.as_str(), v1.as_str()], "newest first");
+  assert!(history[0].is_latest && !history[1].is_latest);
+
+  let copied = raw
+    .copy_object_with(
+      "k",
+      "k",
+      CopySpec {
+        source_version: Some(&v1),
+        ..CopySpec::default()
+      },
+    )
+    .await
+    .unwrap();
+  let restored = copied.version_id.expect("a copy names its new version");
+  assert_eq!(
+    collect(raw.get_object("k").await.unwrap().body)
+      .await
+      .unwrap(),
+    b"one"
+  );
+  let marker = raw
+    .delete_object("k")
+    .await
+    .unwrap()
+    .expect("a delete names its marker");
+  let history = versions_of(&raw, "k").await.unwrap().unwrap();
+  assert_eq!(history[0].version_id, marker);
+  assert!(history[0].delete_marker);
+  assert_eq!(history[1].version_id, restored);
+  donut_sync::migrate::purge(&raw, "k").await.unwrap();
+  assert!(versions_of(&raw, "k").await.unwrap().unwrap().is_empty());
+}

donut-sync/tests/common/mod.rs

@@ -1,5 +1,11 @@
 //! Runs the real binary against the S3 test server of
 //! `donut-sync/docker-compose.yml` (rclone on port 8987).
+//!
+//! The at-rest suite can run against a real S3 store instead, to check what
+//! it relies on there (versions, conditional copies): set
+//! `DONUT_SYNC_TEST_S3_ENDPOINT`, `_REGION`, `_ACCESS_KEY_ID`,
+//! `_SECRET_ACCESS_KEY` and `_BUCKET` (a scratch bucket each test empties
+//! first, so run with `--test-threads=1`).
 
 #![allow(dead_code)]
 
@@ -15,29 +21,88 @@ use donut_sync::store::{Store, StoreError};
 pub const S3_ENDPOINT: &str = "http://127.0.0.1:8987";
 pub const SYNC_TOKEN: &str = "test-sync-token-0123456789abcdef";
 
+/// Where the at-rest suite's storage is: the compose server, or what the
+/// `DONUT_SYNC_TEST_S3_*` variables name.
+pub struct Storage {
+  pub endpoint: String,
+  pub region: String,
+  pub access_key_id: String,
+  pub secret_access_key: String,
+  /// One bucket for every test, emptied before each.
+  pub bucket: Option<String>,
+}
+
+pub fn storage() -> Storage {
+  let var = |name: &str, default: &str| {
+    std::env::var(format!("DONUT_SYNC_TEST_S3_{name}"))
+      .ok()
+      .filter(|v| !v.is_empty())
+      .unwrap_or_else(|| default.to_string())
+  };
+  Storage {
+    endpoint: var("ENDPOINT", S3_ENDPOINT),
+    region: var("REGION", "us-east-1"),
+    access_key_id: var("ACCESS_KEY_ID", "minioadmin"),
+    secret_access_key: var("SECRET_ACCESS_KEY", "minioadmin"),
+    bucket: std::env::var("DONUT_SYNC_TEST_S3_BUCKET")
+      .ok()
+      .filter(|v| !v.is_empty()),
+  }
+}
+
 /// The test storage, without the server: what is really stored.
 pub fn raw_store(bucket: &str) -> S3Store {
+  let storage = storage();
   S3Store::new(
-    Location::parse(S3_ENDPOINT, bucket, true).unwrap(),
+    Location::parse(&storage.endpoint, bucket, true).unwrap(),
     Credentials {
-      access_key_id: "minioadmin".to_string(),
-      secret_access_key: "minioadmin".to_string(),
+      access_key_id: storage.access_key_id,
+      secret_access_key: storage.secret_access_key,
     },
-    "us-east-1".to_string(),
+    storage.region,
   )
 }
 
 /// The storage settings the server takes, for a command run to its exit.
 pub fn storage_env(bucket: &str) -> Vec<(&'static str, String)> {
+  let storage = storage();
   vec![
-    ("S3_ENDPOINT", S3_ENDPOINT.to_string()),
-    ("S3_ACCESS_KEY_ID", "minioadmin".to_string()),
-    ("S3_SECRET_ACCESS_KEY", "minioadmin".to_string()),
+    ("S3_ENDPOINT", storage.endpoint),
+    ("S3_REGION", storage.region),
+    ("S3_ACCESS_KEY_ID", storage.access_key_id),
+    ("S3_SECRET_ACCESS_KEY", storage.secret_access_key),
     ("S3_BUCKET", bucket.to_string()),
     ("S3_FORCE_PATH_STYLE", "true".to_string()),
   ]
 }
 
+/// A bucket of the test's own, or the fixed one, emptied of every version
+/// and unfinished upload.
+pub async fn fresh_bucket(name: &str) -> String {
+  let Some(fixed) = storage().bucket else {
+    return bucket(name);
+  };
+  let raw = raw_store(&fixed);
+  loop {
+    let Some(page) = raw.list_versions("", None).await.unwrap() else {
+      break;
+    };
+    if page.versions.is_empty() {
+      break;
+    }
+    for version in page.versions {
+      raw
+        .delete_version(&version.key, &version.version_id)
+        .await
+        .unwrap();
+    }
+  }
+  for upload in raw.list_multipart("").await.unwrap_or_default() {
+    let _ = raw.abort_multipart(&upload.key, &upload.upload_id).await;
+  }
+  fixed
+}
+
 pub fn http() -> reqwest::Client {
   let _ = rustls::crypto::ring::default_provider().install_default();
   reqwest::Client::builder().no_proxy().build().unwrap()
@@ -45,13 +110,10 @@ pub fn http() -> reqwest::Client {
 
 /// Waits for the S3 test server, or fails with how to start it.
 pub async fn wait_for_s3() {
-  let store = S3Store::new(
-    Location::parse(S3_ENDPOINT, "probe-availability", true).unwrap(),
-    Credentials {
-      access_key_id: "minioadmin".to_string(),
-      secret_access_key: "minioadmin".to_string(),
-    },
-    "us-east-1".to_string(),
+  let store = raw_store(
+    &storage()
+      .bucket
+      .unwrap_or_else(|| "probe-availability".to_string()),
   );
   let deadline = Instant::now() + Duration::from_secs(30);
   loop {
@@ -87,9 +149,10 @@ impl Server {
     let port = free_port();
     let mut env: Vec<(String, String)> = vec![
       ("SYNC_TOKEN".into(), SYNC_TOKEN.into()),
-      ("S3_ENDPOINT".into(), S3_ENDPOINT.into()),
-      ("S3_ACCESS_KEY_ID".into(), "minioadmin".into()),
-      ("S3_SECRET_ACCESS_KEY".into(), "minioadmin".into()),
+      ("S3_ENDPOINT".into(), storage().endpoint),
+      ("S3_REGION".into(), storage().region),
+      ("S3_ACCESS_KEY_ID".into(), storage().access_key_id),
+      ("S3_SECRET_ACCESS_KEY".into(), storage().secret_access_key),
       ("S3_BUCKET".into(), bucket.into()),
       ("S3_FORCE_PATH_STYLE".into(), "true".into()),
       ("PORT".into(), port.to_string()),

src-tauri/src/sync/client.rs

@@ -23,6 +23,33 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);
 /// only when nothing arrives for a full minute.
 const READ_TIMEOUT: Duration = Duration::from_secs(60);
 
+/// Transfers through a sync server's own URLs (a server that encrypts at
+/// rest) at a time, across every profile this app syncs: such a server
+/// serves a limited number per account, and more only wait there.
+const SERVER_TRANSFERS: usize = 6;
+static SERVER_TRANSFER_SLOTS: std::sync::LazyLock<std::sync::Arc<tokio::sync::Semaphore>> =
+  std::sync::LazyLock::new(|| std::sync::Arc::new(tokio::sync::Semaphore::new(SERVER_TRANSFERS)));
+/// The path of a sync server's own transfer URLs.
+const SERVER_TRANSFER_PATH: &str = "/v1/transfer?";
+
+/// How long a busy answer (503, 429) is tried again before the transfer
+/// fails, and the longest single wait.
+const BUSY_BUDGET: Duration = Duration::from_secs(300);
+const BUSY_WAIT_MAX: Duration = Duration::from_secs(30);
+
+/// What a busy answer asks to wait, in seconds.
+fn retry_after(response: &reqwest::Response) -> Option<Duration> {
+  response
+    .headers()
+    .get(reqwest::header::RETRY_AFTER)?
+    .to_str()
+    .ok()?
+    .trim()
+    .parse()
+    .ok()
+    .map(Duration::from_secs)
+}
+
 /// Tries per part of an upload in parts. The file's own retries come on top.
 const PART_ATTEMPTS: u32 = 3;
 const PART_RETRY_DELAY: Duration = Duration::from_secs(1);
@@ -275,6 +302,16 @@ impl SyncClient {
     content_type: Option<&str>,
     metadata: Option<&std::collections::HashMap<String, String>>,
   ) -> SyncResult<()> {
+    let _slot = match transfer {
+      Some(_) => Some(
+        SERVER_TRANSFER_SLOTS
+          .clone()
+          .acquire_owned()
+          .await
+          .expect("the semaphore is never closed"),
+      ),
+      None => None,
+    };
     match transfer {
       Some(transfer) if data.len() as u64 > transfer.part_size => {
         self
@@ -308,17 +345,47 @@ impl SyncClient {
     req
   }
 
+  /// Sends `build()`. For a sync server's own transfer URL, again while the
+  /// answer is busy (503, 429), waiting what Retry-After asks or a growing
+  /// backoff, for at most `BUSY_BUDGET`: that server limits transfers per
+  /// account. Storage's own answers go back at once, to the file's retries.
+  async fn send_patiently(
+    &self,
+    url: &str,
+    build: impl Fn() -> reqwest::RequestBuilder,
+  ) -> SyncResult<reqwest::Response> {
+    let patient = url.contains(SERVER_TRANSFER_PATH);
+    let started = std::time::Instant::now();
+    let mut backoff = Duration::from_secs(1);
+    loop {
+      // The storage host here may come from a presigned URL, so on a
+      // self-hosted server it is whatever the server signed against,
+      // frequently an address only the server can resolve. `reqwest`'s own
+      // Display collapses that to "error sending request"; report the
+      // innermost cause and the host it names.
+      let response = build()
+        .send()
+        .await
+        .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?;
+      let busy = matches!(response.status().as_u16(), 429 | 503);
+      let wait = retry_after(&response).unwrap_or(backoff).min(BUSY_WAIT_MAX);
+      if !patient || !busy || started.elapsed() + wait > BUSY_BUDGET {
+        return Ok(response);
+      }
+      log::debug!("Transfer busy, waiting wait_ms={}", wait.as_millis());
+      tokio::time::sleep(wait).await;
+      backoff = (backoff * 2).min(BUSY_WAIT_MAX);
+    }
+  }
+
   /// Sends a request of an upload in parts; any answer but a success is an
   /// error carrying the server's own words.
   async fn send_part_request(
     &self,
     url: &str,
-    req: reqwest::RequestBuilder,
+    build: impl Fn() -> reqwest::RequestBuilder,
   ) -> SyncResult<reqwest::Response> {
-    let response = req
-      .send()
-      .await
-      .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?;
+    let response = self.send_patiently(url, build).await?;
     if !response.status().is_success() {
       let status = response.status();
       let body = response.text().await.unwrap_or_default();
@@ -339,15 +406,17 @@ impl SyncClient {
     content_type: Option<&str>,
     metadata: Option<&std::collections::HashMap<String, String>>,
   ) -> SyncResult<()> {
-    let start = Self::with_upload_headers(
-      self
-        .client
-        .post(&transfer.url)
-        .header(UPLOAD_LENGTH_HEADER, data.len().to_string())
-        .header("Content-Length", "0"),
-      content_type,
-      metadata,
-    );
+    let start = || {
+      Self::with_upload_headers(
+        self
+          .client
+          .post(&transfer.url)
+          .header(UPLOAD_LENGTH_HEADER, data.len().to_string())
+          .header("Content-Length", "0"),
+        content_type,
+        metadata,
+      )
+    };
     let started: PartsStarted = self
       .send_part_request(&transfer.url, start)
       .await?
@@ -392,11 +461,13 @@ impl SyncClient {
       let mut attempt = 0;
       let etag = loop {
         attempt += 1;
-        let req = self
-          .client
-          .put(&url)
-          .header("Content-Length", chunk.len().to_string())
-          .body(chunk.to_vec());
+        let req = || {
+          self
+            .client
+            .put(&url)
+            .header("Content-Length", chunk.len().to_string())
+            .body(chunk.to_vec())
+        };
         match self.send_part_request(&url, req).await {
           Ok(response) => {
             break response
@@ -421,10 +492,12 @@ impl SyncClient {
       &transfer.url,
       &[("upload", &started.upload), ("op", "complete")],
     )?;
-    let req = self
-      .client
-      .post(&complete)
-      .json(&serde_json::json!({ "parts": etags }));
+    let req = || {
+      self
+        .client
+        .post(&complete)
+        .json(&serde_json::json!({ "parts": etags }))
+    };
     self.send_part_request(&complete, req).await.map(|_| ())
   }
 
@@ -436,24 +509,18 @@ impl SyncClient {
     content_type: Option<&str>,
     metadata: Option<&std::collections::HashMap<String, String>>,
   ) -> SyncResult<()> {
-    let req = Self::with_upload_headers(
-      self
-        .client
-        .put(presigned_url)
-        .header("Content-Length", data.len().to_string())
-        .body(data.to_vec()),
-      content_type,
-      metadata,
-    );
-
-    // The storage host here comes from the presigned URL, so on a self-hosted
-    // server it is whatever the server signed against, frequently an address
-    // only the server can resolve. `reqwest`'s own Display collapses that to
-    // "error sending request", which is why this failure used to be
-    // undiagnosable; report the innermost cause and the host it names.
-    let response = req.send().await.map_err(|e| {
-      SyncError::NetworkError(super::preflight::transport_reason_for(presigned_url, &e))
-    })?;
+    let req = || {
+      Self::with_upload_headers(
+        self
+          .client
+          .put(presigned_url)
+          .header("Content-Length", data.len().to_string())
+          .body(data.to_vec()),
+        content_type,
+        metadata,
+      )
+    };
+    let response = self.send_patiently(presigned_url, req).await?;
 
     if !response.status().is_success() {
       let status = response.status();
@@ -467,9 +534,19 @@ impl SyncClient {
   }
 
   pub async fn download_bytes(&self, presigned_url: &str) -> SyncResult<Vec<u8>> {
-    let response = self.client.get(presigned_url).send().await.map_err(|e| {
-      SyncError::NetworkError(super::preflight::transport_reason_for(presigned_url, &e))
-    })?;
+    let _slot = match presigned_url.contains(SERVER_TRANSFER_PATH) {
+      true => Some(
+        SERVER_TRANSFER_SLOTS
+          .clone()
+          .acquire_owned()
+          .await
+          .expect("the semaphore is never closed"),
+      ),
+      false => None,
+    };
+    let response = self
+      .send_patiently(presigned_url, || self.client.get(presigned_url))
+      .await?;
 
     if !response.status().is_success() {
       return Err(SyncError::NetworkError(format!(
@@ -852,6 +929,41 @@ mod tests {
       .is_err());
   }
 
+  #[tokio::test]
+  async fn a_busy_server_is_waited_for_as_it_asks() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+      .respond_with(ResponseTemplate::new(503).insert_header("retry-after", "1"))
+      .up_to_n_times(2)
+      .with_priority(1)
+      .mount(&server)
+      .await;
+    Mock::given(method("GET"))
+      .respond_with(ResponseTemplate::new(200).set_body_bytes(b"done".to_vec()))
+      .expect(1)
+      .mount(&server)
+      .await;
+    let started = std::time::Instant::now();
+    let url = format!("{}/v1/transfer?key=k&expires=1&sig=s", server.uri());
+    assert_eq!(client().download_bytes(&url).await.unwrap(), b"done");
+    assert!(
+      started.elapsed() >= Duration::from_secs(2),
+      "waited Retry-After twice"
+    );
+  }
+
+  #[tokio::test]
+  async fn a_busy_storage_url_fails_at_once_for_the_file_retries() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+      .respond_with(ResponseTemplate::new(503).insert_header("retry-after", "30"))
+      .expect(1)
+      .mount(&server)
+      .await;
+    let url = format!("{}/bucket/k?X-Amz-Signature=s", server.uri());
+    assert!(client().download_bytes(&url).await.is_err());
+  }
+
   #[test]
   fn answers_from_servers_that_do_not_seal_have_no_transfer() {
     let old: PresignUploadResponse =

src-tauri/src/sync/engine.rs

@@ -128,22 +128,6 @@ pub async fn cancel_profile_sync(profile_id: String) -> Result<bool, String> {
 /// Upload/download concurrency limit
 const SYNC_CONCURRENCY: usize = 32;
 
-/// The limit when files go through the sync server itself (a server that
-/// encrypts at rest): it serves a few transfers per account at a time, and
-/// more only queue there.
-const SERVER_TRANSFER_CONCURRENCY: usize = 6;
-
-/// The path of a sync server's own transfer URLs.
-const SERVER_TRANSFER_PATH: &str = "/v1/transfer?";
-
-fn transfer_concurrency(through_server: bool) -> usize {
-  if through_server {
-    SERVER_TRANSFER_CONCURRENCY
-  } else {
-    SYNC_CONCURRENCY
-  }
-}
-
 /// Max retries for individual file uploads/downloads
 const MAX_FILE_RETRIES: u32 = 3;
 
@@ -1576,8 +1560,7 @@ impl SyncEngine {
       .store(already_bytes, Ordering::Relaxed);
     tracker.emit_final();
 
-    let through_server = url_map.values().any(|(_, transfer)| transfer.is_some());
-    let semaphore = Arc::new(Semaphore::new(transfer_concurrency(through_server)));
+    let semaphore = Arc::new(Semaphore::new(SYNC_CONCURRENCY));
     let client = self.client.clone();
     let profile_dir = profile_dir.to_path_buf();
     let profile_id_owned = profile_id.to_string();
@@ -1848,10 +1831,7 @@ impl SyncEngine {
       .store(already_bytes, Ordering::Relaxed);
     tracker.emit_final();
 
-    let through_server = url_map
-      .values()
-      .any(|url| url.contains(SERVER_TRANSFER_PATH));
-    let semaphore = Arc::new(Semaphore::new(transfer_concurrency(through_server)));
+    let semaphore = Arc::new(Semaphore::new(SYNC_CONCURRENCY));
     let client = self.client.clone();
     let profile_dir = profile_dir.to_path_buf();
     let profile_id_owned = profile_id.to_string();