feat: sealed part uploads

donut-sync/.env.example

@@ -35,8 +35,10 @@ S3_FORCE_PATH_STYLE=true
 # S3_PUBLIC_ENDPOINT=https://storage.example.com
 
 # SYNC_AT_REST_KEY: encrypts every stored object with this server's key, even
-# when E2E encryption is off in Donut Browser. Transfers then go through this
-# server instead of straight to storage. Format id:base64 (32 bytes), e.g.
+# when E2E encryption is off in Donut Browser. Downloads and uploads from
+# current apps then go through this server; older apps still upload straight to
+# storage, and the server encrypts those files minutes later. Format id:base64
+# (32 bytes), e.g.
 # k1:$(openssl rand -base64 32). To rotate, put the new key first:
 # k2:...,k1:... The first key encrypts; every key listed decrypts.
 # Objects stored before stay readable; `donut-sync migrate-at-rest` encrypts them.

donut-sync/README.md

@@ -20,8 +20,10 @@ SYNC_TOKEN=... S3_ENDPOINT=... S3_ACCESS_KEY_ID=... S3_SECRET_ACCESS_KEY=... \
 `.env.example` lists the variables. A `.env` file in the working directory is
 also read.
 
-With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores, and
-transfers go through it instead of straight to storage. Objects stored before
+With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores.
+Downloads and uploads from current apps go through it, large files in parts.
+Older apps still upload straight to storage, and the server encrypts those
+files minutes later. Objects stored before
 stay readable; `donut-sync migrate-at-rest --help` encrypts them in place, and
 `donut-sync verify-at-rest` checks that none is left in plaintext.
 

donut-sync/src/at_rest.rs

@@ -203,25 +203,53 @@ pub struct Sealer {
   pending: Vec<u8>,
 }
 
+pub type Salt = [u8; SALT];
+
+pub fn new_salt() -> Salt {
+  let mut salt = [0u8; SALT];
+  rand::fill(&mut salt);
+  salt
+}
+
+fn header_of(key_id: &str, salt: &Salt) -> Vec<u8> {
+  let mut header = Vec::with_capacity(header_len(key_id) as usize);
+  header.extend_from_slice(MAGIC);
+  header.push(VERSION);
+  header.push(key_id.len() as u8);
+  header.extend_from_slice(key_id.as_bytes());
+  header.extend_from_slice(salt);
+  header
+}
+
 impl Sealer {
+  fn with(key: &Key, salt: &Salt, first_frame: u64) -> Self {
+    Self {
+      key: object_key(&key.secret, salt),
+      header: header_of(&key.id, salt),
+      index: first_frame,
+      pending: Vec::new(),
+    }
+  }
+
   /// A sealer and the header that starts its output.
   pub fn new(keyring: &Keyring) -> (Self, Bytes) {
-    let current = keyring.current();
-    let mut salt = [0u8; SALT];
-    rand::fill(&mut salt);
-    let mut header = Vec::with_capacity(header_len(&current.id) as usize);
-    header.extend_from_slice(MAGIC);
-    header.push(VERSION);
-    header.push(current.id.len() as u8);
-    header.extend_from_slice(current.id.as_bytes());
-    header.extend_from_slice(&salt);
-    let sealer = Self {
-      key: object_key(&current.secret, &salt),
-      header: header.clone(),
-      index: 0,
-      pending: Vec::new(),
-    };
-    (sealer, Bytes::from(header))
+    let sealer = Self::with(keyring.current(), &new_salt(), 0);
+    let header = Bytes::from(sealer.header.clone());
+    (sealer, header)
+  }
+
+  /// One part of an object sealed in parts under `key_id` and `salt`: its
+  /// frames are numbered from `first_frame`, and only part 1 carries the
+  /// header. `None` when the keyring does not hold `key_id`.
+  pub fn for_part(
+    keyring: &Keyring,
+    key_id: &str,
+    salt: &Salt,
+    first_frame: u64,
+  ) -> Option<(Self, Option<Bytes>)> {
+    let sealer = Self::with(keyring.find(key_id)?, salt, first_frame);
+    let header = (first_frame == 0).then(|| Bytes::from(sealer.header.clone()));
+    Some((sealer, header))
   }
 
   fn seal(&mut self, mut frame: Vec<u8>, last: bool) -> Bytes {
@@ -247,6 +275,28 @@ impl Sealer {
     let frame = std::mem::take(&mut self.pending);
     self.seal(frame, true)
   }
+
+  /// The end of a part: the object's final frame on the last part, else the
+  /// part's own last full frame.
+  fn finish_part(mut self, last: bool) -> Option<Bytes> {
+    if last {
+      return Some(self.finish());
+    }
+    let frame = std::mem::take(&mut self.pending);
+    (!frame.is_empty()).then(|| self.seal(frame, false))
+  }
+}
+
+/// The stored size of one part of `plain` bytes. A part before the last holds
+/// whole frames only.
+pub fn sealed_part_len(key_id: &str, plain: u64, first: bool, last: bool) -> u64 {
+  let header = if first { header_len(key_id) } else { 0 };
+  let frames = if last {
+    frames(plain)
+  } else {
+    plain / FRAME as u64
+  };
+  header + plain + frames * TAG as u64
 }
 
 /// Decrypts a sealed byte stream, frame by frame.
@@ -332,14 +382,15 @@ impl Opener {
 }
 
 enum Codec {
-  Seal(Option<Sealer>, Option<Bytes>),
+  /// The sealer, the header still to send, and whether this ends the object.
+  Seal(Option<Sealer>, Option<Bytes>, bool),
   Open(Option<Opener>),
 }
 
 impl Codec {
   fn push(&mut self, data: &[u8], out: &mut VecDeque<Bytes>) -> Result<(), StoreError> {
     match self {
-      Codec::Seal(sealer, header) => {
+      Codec::Seal(sealer, header, _) => {
         if let Some(header) = header.take() {
           out.push_back(header);
         }
@@ -352,11 +403,11 @@ impl Codec {
 
   fn finish(&mut self, out: &mut VecDeque<Bytes>) -> Result<(), StoreError> {
     match self {
-      Codec::Seal(sealer, header) => {
+      Codec::Seal(sealer, header, last) => {
         if let Some(header) = header.take() {
           out.push_back(header);
         }
-        out.push_back(sealer.take().expect("finished once").finish());
+        out.extend(sealer.take().expect("finished once").finish_part(*last));
         Ok(())
       }
       Codec::Open(opener) => {
@@ -414,7 +465,16 @@ fn transform(input: ByteStream, codec: Codec) -> ByteStream {
 /// `input` sealed under the current key.
 pub fn seal_stream(keyring: &Keyring, input: ByteStream) -> ByteStream {
   let (sealer, header) = Sealer::new(keyring);
-  transform(input, Codec::Seal(Some(sealer), Some(header)))
+  transform(input, Codec::Seal(Some(sealer), Some(header), true))
+}
+
+/// One part of an object sealed in parts (see `Sealer::for_part`).
+pub fn seal_part_stream(
+  (sealer, header): (Sealer, Option<Bytes>),
+  last: bool,
+  input: ByteStream,
+) -> ByteStream {
+  transform(input, Codec::Seal(Some(sealer), header, last))
 }
 
 /// `input`, a sealed object, opened. A forged, cut or wrongly keyed object

donut-sync/src/at_rest_tests.rs

@@ -385,3 +385,64 @@ mod sealed_store {
     assert_eq!(store.get_object_text("b").await.unwrap(), "x");
   }
 }
+
+#[tokio::test]
+async fn an_object_sealed_in_parts_reads_like_one_sealed_whole() {
+  let keys = keyring(K1);
+  let part = 2 * FRAME;
+  for size in [
+    0,
+    1,
+    part - 1,
+    part,
+    part + 1,
+    2 * part,
+    3 * part + FRAME + 5,
+  ] {
+    let plain = sample(size);
+    let salt = new_salt();
+    let parts = size.div_ceil(part).max(1);
+    let mut sealed = Vec::new();
+    for i in 0..parts {
+      let start = i * part;
+      let chunk = plain[start..(start + part).min(size)].to_vec();
+      let last = i + 1 == parts;
+      let sealer = Sealer::for_part(&keys, "k1", &salt, (start / FRAME) as u64).unwrap();
+      let len = chunk.len() as u64;
+      let out = collect(seal_part_stream(sealer, last, chunked(chunk, &[777])))
+        .await
+        .unwrap();
+      assert_eq!(
+        out.len() as u64,
+        sealed_part_len("k1", len, i == 0, last),
+        "size {size} part {i}"
+      );
+      sealed.extend(out);
+    }
+    assert_eq!(
+      sealed.len() as u64,
+      sealed_len("k1", size as u64),
+      "size {size}"
+    );
+    assert_eq!(open(keys.clone(), &sealed).unwrap(), plain, "size {size}");
+  }
+  assert!(Sealer::for_part(&keys, "k9", &new_salt(), 0).is_none());
+}
+
+#[tokio::test]
+async fn parts_out_of_place_do_not_open() {
+  let keys = keyring(K1);
+  let salt = new_salt();
+  let part = |first: u64, last: bool, data: Vec<u8>| {
+    let sealer = Sealer::for_part(&keys, "k1", &salt, first).unwrap();
+    collect(seal_part_stream(sealer, last, chunked(data, &[FRAME])))
+  };
+  let one = part(0, false, sample(FRAME)).await.unwrap();
+  let two = part(1, true, sample(10)).await.unwrap();
+  assert!(open(keys.clone(), &[one.clone(), two.clone()].concat()).is_ok());
+  // A part numbered as if it came later, or a non-final part last.
+  let skipped = part(2, true, sample(10)).await.unwrap();
+  assert!(open(keys.clone(), &[one.clone(), skipped].concat()).is_err());
+  let unfinished = part(1, false, sample(FRAME)).await.unwrap();
+  assert!(open(keys.clone(), &[one, unfinished].concat()).is_err());
+}

donut-sync/src/http.rs

@@ -39,7 +39,7 @@ pub fn full(bytes: impl Into<Bytes>) -> ResBody {
     .boxed_unsync()
 }
 
-fn json_response(status: u16, body: &Value) -> Response<ResBody> {
+pub(crate) fn json_response(status: u16, body: &Value) -> Response<ResBody> {
   let text = body.to_string();
   Response::builder()
     .status(status)

donut-sync/src/lib.rs

@@ -18,12 +18,14 @@ pub mod migrate;
 pub mod net;
 pub mod s3;
 pub mod sealed_store;
+pub mod sealer;
 pub mod sigv4;
 pub mod store;
 pub mod subscribe;
 pub mod sync;
 pub mod time;
 pub mod transfer;
+pub mod transfer_parts;
 pub mod verify;
 
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
@@ -167,11 +169,11 @@ pub async fn build_state(config: &Config) -> Result<Arc<AppState>, String> {
         ),
       );
       (
-        Arc::new(SealedStore::new(s3, keyring.clone())),
+        Arc::new(SealedStore::new(s3.clone(), keyring.clone())),
         Some(TransferSigner::new(keyring)),
       )
     }
-    None => (s3, None),
+    None => (s3.clone(), None),
   };
   let presigner = Presigner::new(
     Location::parse(&public_endpoint, &storage.bucket, storage.force_path_style)?,
@@ -202,6 +204,9 @@ pub async fn build_state(config: &Config) -> Result<Arc<AppState>, String> {
 
   sync.ensure_bucket_exists().await?;
   sync.warn_if_presign_endpoint_is_server_only();
+  if let Some(keyring) = keyring {
+    crate::sealer::spawn(s3, keyring);
+  }
 
   Ok(Arc::new(AppState {
     guard,

donut-sync/src/memory_store.rs

@@ -9,8 +9,8 @@ use async_trait::async_trait;
 use md5::{Digest, Md5};
 
 use crate::store::{
-  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload,
-  collect, once,
+  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, PendingUpload, Store,
+  StoreError, Upload, collect, once,
 };
 
 pub type HeadHook = Box<dyn Fn(&str) -> Result<HeadOutput, StoreError> + Send + Sync>;
@@ -24,6 +24,17 @@ pub struct Object {
   pub modified_ms: i64,
 }
 
+/// S3's smallest part, except the last.
+pub const MIN_PART: usize = 5 * 1024 * 1024;
+
+pub struct Multipart {
+  key: String,
+  content_type: Option<String>,
+  metadata: Vec<(String, String)>,
+  parts: BTreeMap<u32, Vec<u8>>,
+  initiated_ms: i64,
+}
+
 #[derive(Default)]
 pub struct MemoryStore {
   pub objects: Mutex<BTreeMap<String, Object>>,
@@ -32,6 +43,7 @@ pub struct MemoryStore {
   pub clock: AtomicI64,
   pub head_hook: Option<HeadHook>,
   pub list_hook: Option<ListHook>,
+  pub uploads: Mutex<BTreeMap<String, Multipart>>,
 }
 
 fn etag(body: &[u8]) -> String {
@@ -268,4 +280,110 @@ impl Store for MemoryStore {
     );
     Ok(())
   }
+
+  async fn create_multipart(
+    &self,
+    key: &str,
+    content_type: Option<&str>,
+    metadata: &[(String, String)],
+  ) -> Result<String, StoreError> {
+    self.call();
+    let id = uuid::Uuid::new_v4().simple().to_string();
+    self.uploads.lock().unwrap().insert(
+      id.clone(),
+      Multipart {
+        key: key.to_string(),
+        content_type: content_type.map(str::to_string),
+        metadata: metadata.to_vec(),
+        parts: BTreeMap::new(),
+        initiated_ms: self.tick(),
+      },
+    );
+    Ok(id)
+  }
+
+  async fn upload_part(
+    &self,
+    key: &str,
+    upload_id: &str,
+    part: u32,
+    body: ByteStream,
+    length: u64,
+  ) -> Result<String, StoreError> {
+    self.call();
+    let body = collect(body).await?;
+    if body.len() as u64 != length {
+      return Err(StoreError::Transport("length mismatch".to_string()));
+    }
+    let mut uploads = self.uploads.lock().unwrap();
+    let upload = uploads
+      .get_mut(upload_id)
+      .filter(|u| u.key == key)
+      .ok_or(StoreError::NotFound)?;
+    let tag = etag(&body);
+    upload.parts.insert(part, body);
+    Ok(tag)
+  }
+
+  async fn complete_multipart(
+    &self,
+    key: &str,
+    upload_id: &str,
+    etags: &[String],
+  ) -> Result<(), StoreError> {
+    self.call();
+    let upload = self
+      .uploads
+      .lock()
+      .unwrap()
+      .remove(upload_id)
+      .filter(|u| u.key == key)
+      .ok_or(StoreError::NotFound)?;
+    let mut body = Vec::new();
+    for (i, expected) in etags.iter().enumerate() {
+      let part = upload
+        .parts
+        .get(&(i as u32 + 1))
+        .filter(|part| etag(part) == *expected)
+        .ok_or_else(|| StoreError::Service {
+          status: 400,
+          code: "InvalidPart".to_string(),
+          message: String::new(),
+        })?;
+      if i + 1 < etags.len() && part.len() < MIN_PART {
+        return Err(StoreError::Service {
+          status: 400,
+          code: "EntityTooSmall".to_string(),
+          message: String::new(),
+        });
+      }
+      body.extend_from_slice(part);
+    }
+    self.insert(key, body, upload.metadata, upload.content_type);
+    Ok(())
+  }
+
+  async fn abort_multipart(&self, _key: &str, upload_id: &str) -> Result<(), StoreError> {
+    self.call();
+    self.uploads.lock().unwrap().remove(upload_id);
+    Ok(())
+  }
+
+  async fn list_multipart(&self, prefix: &str) -> Result<Vec<PendingUpload>, StoreError> {
+    self.call();
+    Ok(
+      self
+        .uploads
+        .lock()
+        .unwrap()
+        .iter()
+        .filter(|(_, u)| u.key.starts_with(prefix))
+        .map(|(id, u)| PendingUpload {
+          key: u.key.clone(),
+          upload_id: id.clone(),
+          initiated_ms: Some(u.initiated_ms),
+        })
+        .collect(),
+    )
+  }
 }

donut-sync/src/migrate.rs

@@ -29,7 +29,7 @@ use crate::store::{ByteStream, ListInput, Store, StoreError, Upload};
 pub const COMMAND: &str = "migrate-at-rest";
 const CONTEXT: &str = "Migrate";
 /// Temporary sealed copies. No client prefix matches it.
-pub const TMP_PREFIX: &str = ".donut-sync-at-rest-tmp/";
+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;
@@ -148,7 +148,7 @@ pub struct Report {
   pub failed: u64,
 }
 
-enum Outcome {
+pub(crate) enum Outcome {
   Current,
   Sealed(u64),
   Rotated(u64),
@@ -191,7 +191,7 @@ fn throttled(body: ByteStream, limit: Option<Arc<RateLimit>>) -> ByteStream {
   }
 }
 
-struct Migration {
+pub(crate) struct Migration {
   store: Arc<dyn Store>,
   keyring: Arc<Keyring>,
   dry_run: bool,
@@ -199,6 +199,19 @@ struct Migration {
   limit: Option<Arc<RateLimit>>,
 }
 
+impl Migration {
+  /// Seals in place, as the migration does, without a rate limit.
+  pub(crate) fn sealing(store: Arc<dyn Store>, keyring: Arc<Keyring>, conditional: bool) -> Self {
+    Self {
+      store,
+      keyring,
+      dry_run: false,
+      conditional,
+      limit: None,
+    }
+  }
+}
+
 fn precondition_failed(error: &StoreError) -> bool {
   matches!(error, StoreError::Service { status: 412, .. })
 }
@@ -234,7 +247,7 @@ pub async fn honors_if_match(store: &dyn Store) -> Result<bool, StoreError> {
 impl Migration {
   /// `listed_ms` is the LastModified the listing gave: it has milliseconds,
   /// where a HEAD has seconds.
-  async fn object(&self, key: &str, listed_ms: Option<i64>) -> Outcome {
+  pub(crate) async fn object(&self, key: &str, listed_ms: Option<i64>) -> Outcome {
     let mut attempt = 0;
     loop {
       attempt += 1;
@@ -432,7 +445,7 @@ pub async fn migrate(
       .contents
       .into_iter()
       .map(|object| (object.key, object.last_modified_ms))
-      .filter(|(key, _)| !key.starts_with(TMP_PREFIX))
+      .filter(|(key, _)| !key.starts_with(crate::sealer::INTERNAL_ROOT))
       .collect();
     let keys: Vec<&String> = objects.iter().map(|(key, _)| key).collect();
     let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(objects.iter())

donut-sync/src/s3.rs

@@ -11,7 +11,8 @@ use reqwest::Method;
 use crate::jsval::js_number;
 use crate::sigv4::{self, Credentials};
 use crate::store::{
-  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload,
+  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, PendingUpload, Store,
+  StoreError, Upload,
 };
 use crate::time;
 
@@ -329,6 +330,38 @@ fn restore_controls(text: &str) -> String {
     .collect()
 }
 
+type UploadsPage = (Vec<PendingUpload>, Option<(String, String)>);
+
+/// A ListMultipartUploads reply: the uploads, and where the next page starts.
+fn parse_uploads(body: &[u8]) -> Result<UploadsPage, StoreError> {
+  let text = std::str::from_utf8(body)
+    .map_err(|e| StoreError::Transport(format!("ListMultipartUploads reply is not UTF-8: {e}")))?;
+  let doc = roxmltree::Document::parse(text)
+    .map_err(|e| StoreError::Transport(format!("ListMultipartUploads reply is not XML: {e}")))?;
+  let root = doc.root_element();
+  let mut uploads = Vec::new();
+  let mut truncated = false;
+  for node in root.children().filter(|n| n.is_element()) {
+    match node.tag_name().name() {
+      "Upload" => uploads.push(PendingUpload {
+        key: child_text(node, "Key").unwrap_or("").to_string(),
+        upload_id: child_text(node, "UploadId").unwrap_or("").to_string(),
+        initiated_ms: child_text(node, "Initiated").and_then(time::parse_rfc3339_ms),
+      }),
+      "IsTruncated" => truncated = node.text().unwrap_or("").trim() == "true",
+      _ => {}
+    }
+  }
+  let next = if truncated {
+    child_text(root, "NextKeyMarker")
+      .zip(child_text(root, "NextUploadIdMarker"))
+      .map(|(key, id)| (key.to_string(), id.to_string()))
+  } else {
+    None
+  };
+  Ok((uploads, next))
+}
+
 fn parse_list(body: &[u8]) -> Result<ListOutput, StoreError> {
   let text = std::str::from_utf8(body)
     .map_err(|e| StoreError::Transport(format!("ListObjectsV2 reply is not UTF-8: {e}")))?;
@@ -748,6 +781,150 @@ impl Store for S3Store {
     })
   }
 
+  async fn create_multipart(
+    &self,
+    key: &str,
+    content_type: Option<&str>,
+    metadata: &[(String, String)],
+  ) -> Result<String, StoreError> {
+    let path = self.location.object_path(key);
+    let mut headers = Vec::new();
+    if let Some(content_type) = content_type {
+      headers.push(("content-type".to_string(), content_type.to_string()));
+    }
+    for (name, value) in metadata {
+      headers.push((format!("x-amz-meta-{name}"), value.clone()));
+    }
+    let reply = self
+      .expect_success(
+        Method::POST,
+        &path,
+        &[("uploads".to_string(), String::new())],
+        &headers,
+        &[],
+      )
+      .await?;
+    let text = String::from_utf8_lossy(&reply.body);
+    let doc = roxmltree::Document::parse(&text)
+      .map_err(|e| StoreError::Transport(format!("CreateMultipartUpload reply is not XML: {e}")))?;
+    child_text(doc.root_element(), "UploadId")
+      .filter(|id| !id.is_empty())
+      .map(str::to_string)
+      .ok_or_else(|| {
+        StoreError::Transport("CreateMultipartUpload reply has no UploadId".to_string())
+      })
+  }
+
+  /// One attempt: a stream cannot be sent twice.
+  async fn upload_part(
+    &self,
+    key: &str,
+    upload_id: &str,
+    part: u32,
+    body: ByteStream,
+    length: u64,
+  ) -> Result<String, StoreError> {
+    let path = self.location.object_path(key);
+    let query = [
+      ("partNumber".to_string(), part.to_string()),
+      ("uploadId".to_string(), upload_id.to_string()),
+    ];
+    let headers = [("content-length".to_string(), length.to_string())];
+    let response = self
+      .request(
+        &Method::PUT,
+        &path,
+        &query,
+        &headers,
+        sigv4::UNSIGNED_PAYLOAD.to_string(),
+      )
+      .body(reqwest::Body::wrap_stream(body))
+      .send()
+      .await
+      .map_err(|e| StoreError::Transport(transport_message(&e)))?;
+    if !response.status().is_success() {
+      return Err(refusal(response).await);
+    }
+    head_of(response.headers())
+      .etag
+      .ok_or_else(|| StoreError::Transport("UploadPart reply has no ETag".to_string()))
+  }
+
+  async fn complete_multipart(
+    &self,
+    key: &str,
+    upload_id: &str,
+    etags: &[String],
+  ) -> Result<(), StoreError> {
+    let mut xml = String::from("<CompleteMultipartUpload>");
+    for (i, etag) in etags.iter().enumerate() {
+      xml.push_str(&format!(
+        "<Part><PartNumber>{}</PartNumber><ETag>{}</ETag></Part>",
+        i + 1,
+        xml_escape(etag)
+      ));
+    }
+    xml.push_str("</CompleteMultipartUpload>");
+    let path = self.location.object_path(key);
+    let reply = self
+      .expect_success(
+        Method::POST,
+        &path,
+        &[("uploadId".to_string(), upload_id.to_string())],
+        &[("content-type".to_string(), "application/xml".to_string())],
+        xml.as_bytes(),
+      )
+      .await?;
+    // Like a copy, a completion can fail with 200 and an error document.
+    if let Some((code, message)) = error_document(&reply.body) {
+      return Err(StoreError::Service {
+        status: reply.status,
+        code,
+        message,
+      });
+    }
+    Ok(())
+  }
+
+  async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError> {
+    let path = self.location.object_path(key);
+    self
+      .expect_success(
+        Method::DELETE,
+        &path,
+        &[("uploadId".to_string(), upload_id.to_string())],
+        &[],
+        &[],
+      )
+      .await
+      .map(|_| ())
+  }
+
+  async fn list_multipart(&self, prefix: &str) -> Result<Vec<PendingUpload>, StoreError> {
+    let path = self.location.bucket_path();
+    let mut out = Vec::new();
+    let mut marker: Option<(String, String)> = None;
+    loop {
+      let mut query = vec![
+        ("uploads".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(("upload-id-marker".to_string(), id.clone()));
+      }
+      let reply = self
+        .expect_success(Method::GET, &path, &query, &[], &[])
+        .await?;
+      let (uploads, next) = parse_uploads(&reply.body)?;
+      out.extend(uploads);
+      match next {
+        Some(next) if Some(&next) != marker.as_ref() => marker = Some(next),
+        _ => return Ok(out),
+      }
+    }
+  }
+
   /// One attempt: a stream cannot be sent twice.
   async fn put_object_stream(
     &self,

donut-sync/src/sealed_store.rs

@@ -12,7 +12,7 @@ use async_trait::async_trait;
 
 use crate::at_rest::{self, Keyring, MARKER};
 use crate::store::{
-  ByteStream, Download, HeadOutput, ListInput, ListOutput, Store, StoreError, Upload,
+  ByteStream, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store, StoreError, Upload,
 };
 
 /// User metadata (`x-amz-meta-donut-orig-mtime`): the object's LastModified
@@ -185,4 +185,50 @@ impl Store for SealedStore {
       )
       .await
   }
+
+  // Multipart parts pass through as they come: the caller (transfer.rs)
+  // seals each part as whole frames of one object.
+
+  async fn create_multipart(
+    &self,
+    key: &str,
+    content_type: Option<&str>,
+    metadata: &[(String, String)],
+  ) -> Result<String, StoreError> {
+    self
+      .inner
+      .create_multipart(key, content_type, &self.with_marker(metadata))
+      .await
+  }
+
+  async fn upload_part(
+    &self,
+    key: &str,
+    upload_id: &str,
+    part: u32,
+    body: ByteStream,
+    length: u64,
+  ) -> Result<String, StoreError> {
+    self
+      .inner
+      .upload_part(key, upload_id, part, body, length)
+      .await
+  }
+
+  async fn complete_multipart(
+    &self,
+    key: &str,
+    upload_id: &str,
+    etags: &[String],
+  ) -> Result<(), StoreError> {
+    self.inner.complete_multipart(key, upload_id, etags).await
+  }
+
+  async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError> {
+    self.inner.abort_multipart(key, upload_id).await
+  }
+
+  async fn list_multipart(&self, prefix: &str) -> Result<Vec<PendingUpload>, StoreError> {
+    self.inner.list_multipart(prefix).await
+  }
 }

donut-sync/src/sealer.rs

@@ -0,0 +1,444 @@
+//! Seals what older apps upload in plaintext. Older apps PUT to the direct
+//! storage URL that presign-upload has always returned, because one request
+//! through a proxy such as Cloudflare cannot carry a large file. The server
+//! notes each key it hands such a URL for (`note_direct_uploads`), in the
+//! bucket so a restart keeps the list, and seals those objects in place with
+//! the migration's ETag-checked rewrite: on the first pass after the URL is
+//! handed out (`PASS_EVERY`), and once more after the URL is dead. An hourly pass over everything written
+//! since the last one catches what a lost record missed, and storage uploads
+//! in parts that were never finished are aborted.
+
+use std::collections::HashSet;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use futures_util::StreamExt;
+use serde_json::{Value, json};
+
+use crate::at_rest::Keyring;
+use crate::log;
+use crate::migrate::{Migration, Outcome, honors_if_match};
+use crate::sealed_store::SealedStore;
+use crate::store::{ListInput, Store, StoreError};
+use crate::time;
+
+/// This server's own objects. No client prefix matches it.
+pub const INTERNAL_ROOT: &str = ".donut-sync-at-rest/";
+const PENDING: &str = ".donut-sync-at-rest/pending/";
+const STATE: &str = ".donut-sync-at-rest/state.json";
+const CONTEXT: &str = "Sealer";
+const PASS_EVERY: Duration = Duration::from_secs(120);
+const BACKSTOP_EVERY: Duration = Duration::from_secs(3600);
+/// Overlap between backstop passes, for clock skew between server and storage.
+const BACKSTOP_OVERLAP_MS: i64 = 10 * 60 * 1000;
+/// A first backstop pass looks back this far; the migration covers the rest.
+const FIRST_BACKSTOP_MS: i64 = 2 * 3600 * 1000;
+/// A record is kept this long after its URLs expire, for uploads in flight.
+const RECORD_GRACE_MS: i64 = 5 * 60 * 1000;
+/// Unfinished part uploads older than this are aborted: past any upload token.
+const ABANDONED_MS: i64 = 26 * 3600 * 1000;
+const CONCURRENCY: usize = 4;
+
+/// 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(
+  store: &dyn Store,
+  keys: &[String],
+  expires_ms: i64,
+) -> Result<(), StoreError> {
+  let record = format!(
+    "{PENDING}{expires_ms:013}-{}",
+    uuid::Uuid::new_v4().simple()
+  );
+  store
+    .put_object(
+      &record,
+      json!({ "keys": keys }).to_string().into_bytes(),
+      "application/json",
+      &[],
+    )
+    .await
+}
+
+fn record_expiry(record: &str) -> Option<i64> {
+  record
+    .strip_prefix(PENDING)?
+    .split('-')
+    .next()?
+    .parse()
+    .ok()
+}
+
+#[derive(Default, Debug, PartialEq)]
+pub struct PassReport {
+  pub records: u64,
+  pub records_done: u64,
+  pub checked: u64,
+  pub sealed: u64,
+  pub failed: u64,
+}
+
+impl PassReport {
+  fn add(&mut self, outcome: &Outcome) {
+    self.checked += 1;
+    match outcome {
+      Outcome::Sealed(_) | Outcome::Rotated(_) => self.sealed += 1,
+      Outcome::Failed(_) => self.failed += 1,
+      Outcome::Current | Outcome::Changed => {}
+    }
+  }
+}
+
+pub struct Sealer {
+  raw: Arc<dyn Store>,
+  sealed: SealedStore,
+  migration: Migration,
+  /// Records sealed once while their URLs live. A restart forgets them, and
+  /// they are sealed once more: harmless.
+  first_round: Mutex<HashSet<String>>,
+}
+
+impl Sealer {
+  pub async fn new(raw: Arc<dyn Store>, keyring: Arc<Keyring>) -> Result<Self, StoreError> {
+    let conditional = honors_if_match(raw.as_ref()).await?;
+    Ok(Self {
+      sealed: SealedStore::new(raw.clone(), keyring.clone()),
+      migration: Migration::sealing(raw.clone(), keyring, conditional),
+      raw,
+      first_round: Mutex::new(HashSet::new()),
+    })
+  }
+
+  async fn seal_all(&self, keys: Vec<(String, Option<i64>)>, report: &mut PassReport) {
+    let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys)
+      .map(|(key, listed)| async move {
+        let outcome = self.migration.object(&key, listed).await;
+        (key, outcome)
+      })
+      .buffer_unordered(CONCURRENCY)
+      .collect()
+      .await;
+    for (key, outcome) in outcomes {
+      if let Outcome::Failed(err) = &outcome {
+        log::warn(
+          CONTEXT,
+          &format!("Object not sealed key={key} err=\"{err}\""),
+        );
+      }
+      report.add(&outcome);
+    }
+  }
+
+  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);
+      }
+    }
+  }
+
+  /// Seals the keys of each new record, and of each record whose URLs are
+  /// dead, which it then drops.
+  pub async fn pending_pass(&self, now_ms: i64) -> Result<PassReport, StoreError> {
+    let mut report = PassReport::default();
+    for (record, _) in self.list(Some(PENDING)).await? {
+      report.records += 1;
+      let dead = record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS);
+      if !dead && !self.first_round.lock().unwrap().insert(record.clone()) {
+        continue;
+      }
+      let keys: Vec<(String, Option<i64>)> = 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(|k| (k.to_string(), None)))
+          .collect(),
+        Err(StoreError::NotFound) => continue,
+        Err(e) => {
+          log::warn(
+            CONTEXT,
+            &format!("Record unreadable key={record} err=\"{e}\""),
+          );
+          report.failed += 1;
+          continue;
+        }
+      };
+      let failed_before = report.failed;
+      self.seal_all(keys, &mut report).await;
+      if dead && report.failed == failed_before {
+        self.raw.delete_object(&record).await?;
+        self.first_round.lock().unwrap().remove(&record);
+        report.records_done += 1;
+      }
+    }
+    Ok(report)
+  }
+
+  /// Seals everything written since the last pass, and aborts abandoned
+  /// uploads in parts.
+  pub async fn backstop(&self, now_ms: i64) -> Result<PassReport, StoreError> {
+    let since = match self.sealed.get_object_text(STATE).await {
+      Ok(text) => serde_json::from_str::<Value>(&text)
+        .ok()
+        .and_then(|v| v["backstopSince"].as_i64()),
+      Err(StoreError::NotFound) => None,
+      Err(e) => return Err(e),
+    }
+    .unwrap_or(now_ms - FIRST_BACKSTOP_MS);
+    let recent: Vec<(String, Option<i64>)> = self
+      .list(None)
+      .await?
+      .into_iter()
+      .filter(|(key, modified)| {
+        !key.starts_with(INTERNAL_ROOT)
+          && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS)
+      })
+      .collect();
+    let mut report = PassReport::default();
+    self.seal_all(recent, &mut report).await;
+    if report.failed == 0 {
+      self
+        .sealed
+        .put_object(
+          STATE,
+          json!({ "backstopSince": now_ms }).to_string().into_bytes(),
+          "application/json",
+          &[],
+        )
+        .await?;
+    }
+    match self.raw.list_multipart("").await {
+      Ok(uploads) => {
+        for upload in uploads {
+          if upload
+            .initiated_ms
+            .is_some_and(|ms| now_ms - ms > ABANDONED_MS)
+          {
+            match self
+              .raw
+              .abort_multipart(&upload.key, &upload.upload_id)
+              .await
+            {
+              Ok(()) => log::log(
+                CONTEXT,
+                &format!("Abandoned upload aborted key={}", upload.key),
+              ),
+              Err(e) => log::warn(
+                CONTEXT,
+                &format!(
+                  "Abandoned upload not aborted key={} err=\"{e}\"",
+                  upload.key
+                ),
+              ),
+            }
+          }
+        }
+      }
+      // Some stores (rclone) cannot list them, and keep none.
+      Err(StoreError::NotFound) => {}
+      Err(e) => log::warn(
+        CONTEXT,
+        &format!("Unfinished uploads not listed err=\"{e}\""),
+      ),
+    }
+    Ok(report)
+  }
+}
+
+fn log_pass(name: &str, outcome: Result<PassReport, StoreError>, started: Instant) {
+  match outcome {
+    Ok(report) if report.sealed > 0 || report.failed > 0 || report.records_done > 0 => {
+      let line = format!(
+        "{name} records={} records_done={} checked={} sealed={} failed={} elapsed_ms={}",
+        report.records,
+        report.records_done,
+        report.checked,
+        report.sealed,
+        report.failed,
+        started.elapsed().as_millis()
+      );
+      if report.failed > 0 {
+        log::warn(CONTEXT, &line);
+      } else {
+        log::log(CONTEXT, &line);
+      }
+    }
+    Ok(_) => {}
+    Err(e) => log::warn(CONTEXT, &format!("{name} stopped err=\"{e}\"")),
+  }
+}
+
+/// Runs the passes for as long as the server does.
+pub fn spawn(raw: Arc<dyn Store>, keyring: Arc<Keyring>) {
+  tokio::spawn(async move {
+    // The first pass waits too: starting writes a probe object, which
+    // should not race the server's first requests.
+    tokio::time::sleep(PASS_EVERY).await;
+    let sealer = loop {
+      match Sealer::new(raw.clone(), keyring.clone()).await {
+        Ok(sealer) => break sealer,
+        Err(e) => {
+          log::warn(CONTEXT, &format!("Sealer not started err=\"{e}\""));
+          tokio::time::sleep(PASS_EVERY).await;
+        }
+      }
+    };
+    let mut last_backstop: Option<Instant> = None;
+    loop {
+      let started = Instant::now();
+      log_pass(
+        "Pending pass",
+        sealer.pending_pass(time::now_ms()).await,
+        started,
+      );
+      if last_backstop.is_none_or(|at| at.elapsed() >= BACKSTOP_EVERY) {
+        let started = Instant::now();
+        log_pass(
+          "Backstop pass",
+          sealer.backstop(time::now_ms()).await,
+          started,
+        );
+        last_backstop = Some(Instant::now());
+      }
+      tokio::time::sleep(PASS_EVERY).await;
+    }
+  });
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use crate::memory_store::MemoryStore;
+  use crate::sealed_store::sealed_by;
+
+  const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+
+  async fn setup() -> (Arc<MemoryStore>, SealedStore, Sealer) {
+    let raw = Arc::new(MemoryStore::default());
+    let keyring = Arc::new(Keyring::parse(K1).unwrap());
+    let sealed = SealedStore::new(raw.clone(), keyring.clone());
+    let sealer = Sealer::new(raw.clone(), keyring).await.unwrap();
+    (raw, sealed, sealer)
+  }
+
+  async fn is_sealed(raw: &MemoryStore, key: &str) -> bool {
+    sealed_by(&raw.head_object(key).await.unwrap()).is_some()
+  }
+
+  #[tokio::test]
+  async fn records_seal_direct_uploads_until_their_urls_are_dead() {
+    let (raw, sealed, sealer) = setup().await;
+    note_direct_uploads(&sealed, &["a".to_string(), "b".to_string()], 10_000)
+      .await
+      .unwrap();
+    let record = raw
+      .keys()
+      .into_iter()
+      .find(|k| k.starts_with(PENDING))
+      .unwrap();
+    assert!(
+      raw.get(&record).unwrap().starts_with("DSAR"),
+      "the record is sealed"
+    );
+
+    // "a" is uploaded, "b" not yet.
+    raw.set("a", "plain a");
+    let report = sealer.pending_pass(5_000).await.unwrap();
+    assert_eq!(
+      (report.records, report.sealed, report.records_done),
+      (1, 1, 0)
+    );
+    assert!(is_sealed(&raw, "a").await);
+    assert_eq!(sealed.get_object_text("a").await.unwrap(), "plain a");
+
+    // A live record is not sealed again.
+    raw.set("b", "plain b");
+    let report = sealer.pending_pass(6_000).await.unwrap();
+    assert_eq!((report.records, report.checked), (1, 0));
+
+    // The URL writes again before it expires; the pass after it is dead
+    // seals that too.
+    raw.set("a", "plain a2");
+    let report = sealer
+      .pending_pass(10_000 + RECORD_GRACE_MS + 1)
+      .await
+      .unwrap();
+    assert_eq!((report.sealed, report.records_done), (2, 1));
+    assert_eq!(sealed.get_object_text("a").await.unwrap(), "plain a2");
+    assert!(raw.keys().iter().all(|k| !k.starts_with(PENDING)));
+  }
+
+  #[tokio::test]
+  async fn the_backstop_seals_recent_plaintext_and_remembers_where_it_stopped() {
+    use std::sync::atomic::Ordering::SeqCst;
+    let (raw, sealed, sealer) = setup().await;
+    let start = 10 * 3600 * 1000;
+    raw.clock.store(start, SeqCst);
+    raw.set("old", "x");
+    raw.clock.store(start + 3 * 3600 * 1000, SeqCst);
+    raw.set("new", "y");
+    let first = raw.clock.load(SeqCst) + 1;
+    let report = sealer.backstop(first).await.unwrap();
+    assert_eq!(
+      report.sealed, 1,
+      "a first pass looks back FIRST_BACKSTOP_MS"
+    );
+    assert!(is_sealed(&raw, "new").await);
+    assert!(
+      !is_sealed(&raw, "old").await,
+      "older objects are the migration's"
+    );
+    assert!(
+      sealed
+        .get_object_text(STATE)
+        .await
+        .unwrap()
+        .contains(&first.to_string())
+    );
+
+    raw.set("later", "z");
+    let report = sealer.backstop(raw.clock.load(SeqCst) + 1).await.unwrap();
+    assert_eq!(report.sealed, 1);
+    assert!(is_sealed(&raw, "later").await);
+    assert!(!is_sealed(&raw, "old").await);
+  }
+
+  #[tokio::test]
+  async fn abandoned_uploads_in_parts_are_aborted() {
+    let (raw, _, sealer) = setup().await;
+    raw.create_multipart("k", None, &[]).await.unwrap();
+    sealer.backstop(1).await.unwrap();
+    assert_eq!(raw.list_multipart("").await.unwrap().len(), 1);
+    sealer.backstop(ABANDONED_MS + 1_000_000).await.unwrap();
+    assert!(raw.list_multipart("").await.unwrap().is_empty());
+  }
+
+  #[test]
+  fn records_name_their_expiry() {
+    assert_eq!(
+      record_expiry(&format!("{PENDING}0000000012345-abc")),
+      Some(12345)
+    );
+    assert_eq!(record_expiry("other"), None);
+  }
+}

donut-sync/src/store.rs

@@ -138,6 +138,40 @@ pub trait Store: Send + Sync {
     body: ByteStream,
     upload: Upload<'_>,
   ) -> Result<(), StoreError>;
+  /// Starts a multipart upload; the object gets `content_type` and `metadata`.
+  async fn create_multipart(
+    &self,
+    key: &str,
+    content_type: Option<&str>,
+    metadata: &[(String, String)],
+  ) -> Result<String, StoreError>;
+  /// One part, `length` bytes from `body`. Returns its ETag.
+  async fn upload_part(
+    &self,
+    key: &str,
+    upload_id: &str,
+    part: u32,
+    body: ByteStream,
+    length: u64,
+  ) -> Result<String, StoreError>;
+  /// Joins the parts, in order, by their ETags.
+  async fn complete_multipart(
+    &self,
+    key: &str,
+    upload_id: &str,
+    etags: &[String],
+  ) -> Result<(), StoreError>;
+  async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError>;
+  /// Unfinished uploads under `prefix`.
+  async fn list_multipart(&self, prefix: &str) -> Result<Vec<PendingUpload>, StoreError>;
+}
+
+/// An unfinished multipart upload.
+#[derive(Debug, Clone, PartialEq)]
+pub struct PendingUpload {
+  pub key: String,
+  pub upload_id: String,
+  pub initiated_ms: Option<i64>,
 }
 
 pub struct Download {

donut-sync/src/sync.rs

@@ -195,17 +195,36 @@ impl SyncService {
     self.transfer.as_ref()
   }
 
-  fn put_url(
+  /// `transfer` of a presign-upload answer when the store is sealed: the
+  /// signed URL newer apps upload through, in parts. Older apps keep the
+  /// direct storage URL, whose writes the sealer seals afterwards.
+  fn transfer_json(
     &self,
     origin: &str,
     key: &str,
     expires_in: f64,
     metadata: &[(String, String)],
     now: i64,
-  ) -> String {
-    match &self.transfer {
-      Some(signer) => signer.url(origin, "PUT", key, expires_ms(now, expires_in), metadata),
-      None => self.presigner.put_url(key, expires_in, metadata, now),
+  ) -> Option<Value> {
+    let signer = self.transfer.as_ref()?;
+    Some(json!({
+      "url": signer.url(origin, "PUT", key, expires_ms(now, expires_in), metadata),
+      "partSize": crate::transfer_parts::PART_SIZE,
+    }))
+  }
+
+  /// 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) {
+    if self.transfer.is_none() || 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}\""));
     }
   }
 
@@ -240,16 +259,13 @@ impl SyncService {
   /// publish its storage host.
   /// Withheld too when transfers come here: devices never reach storage.
   pub fn diagnostic_storage_endpoint(&self) -> Option<&str> {
-    (!self.cloud && self.transfer.is_none()).then_some(self.public_endpoint.as_str())
+    (!self.cloud).then_some(self.public_endpoint.as_str())
   }
 
   /// A single-label host (`minio`, `s3`) resolves only inside a container
   /// network, so every presigned URL built from it fails for the desktop
   /// client while the server's own S3 calls succeed. Say it once at boot.
   pub fn warn_if_presign_endpoint_is_server_only(&self) {
-    if self.transfer.is_some() {
-      return;
-    }
     let Ok(url) = url::Url::parse(&self.public_endpoint) else {
       return;
     };
@@ -508,13 +524,15 @@ impl SyncService {
     let expires_in = clamp_expires_in(body.field("expiresIn")?);
     let now = time::now_ms();
     let expires_at = iso_at(now as f64 + expires_in * 1000.0);
-    let url = self.put_url(
-      origin,
-      require_key(&key)?,
-      expires_in,
-      metadata.as_deref().unwrap_or(&[]),
-      now,
-    );
+    let scoped = require_key(&key)?;
+    let metadata_or_none = metadata.as_deref().unwrap_or(&[]);
+    let url = self
+      .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;
 
     if ctx.mode == Mode::Cloud && field_starts_with(raw_key, "profiles/")? {
       self.report_profile_usage_async(ctx);
@@ -527,6 +545,9 @@ impl SyncService {
     if let Some(metadata) = &metadata {
       out.insert("metadata".to_string(), metadata_json(metadata));
     }
+    if let Some(transfer) = transfer {
+      out.insert("transfer".to_string(), transfer);
+    }
     Ok(Value::Object(out))
   }
 
@@ -632,7 +653,9 @@ impl SyncService {
     let objects: Vec<Value> = response
       .contents
       .iter()
-      .filter(|object| !object.key.ends_with(MANIFEST_KEY))
+      .filter(|object| {
+        !object.key.ends_with(MANIFEST_KEY) && !object.key.starts_with(crate::sealer::INTERNAL_ROOT)
+      })
       .map(|object| {
         let key = if !user_prefix.is_empty() {
           object.key.strip_prefix(user_prefix).unwrap_or(&object.key)
@@ -708,16 +731,23 @@ impl SyncService {
     )
     .await?;
     let mut out_items = Vec::with_capacity(keys.len());
+    let mut direct = Vec::with_capacity(keys.len());
     for (raw, key) in raw_keys.iter().zip(&keys) {
-      let url = self.put_url(origin, require_key(key)?, expires_in, &[], now);
+      let scoped = require_key(key)?;
+      let url = self.presigner.put_url(scoped, expires_in, &[], now);
       let mut item = Map::new();
       if let Field::Value(value) = raw {
         item.insert("key".to_string(), (*value).clone());
       }
       item.insert("url".to_string(), Value::String(url));
       item.insert("expiresAt".to_string(), Value::String(expires_at.clone()));
+      if let Some(transfer) = self.transfer_json(origin, scoped, expires_in, &[], now) {
+        item.insert("transfer".to_string(), transfer);
+      }
       out_items.push(Value::Object(item));
+      direct.push(scoped.to_string());
     }
+    self.note_direct_uploads(&direct, now, expires_in).await;
 
     if ctx.mode == Mode::Cloud {
       let mut any_profile = false;

donut-sync/src/transfer.rs

@@ -3,6 +3,9 @@
 //! signed and expiring like S3's, and the body streams through the sealed
 //! store. A device sends and reads exactly what it did with S3: the same
 //! method, `Content-Length`, `Content-Type` and the signed `x-amz-meta-*`.
+//!
+//! A request body through a proxy such as Cloudflare has a size and time
+//! limit, so newer apps upload large files in parts (`transfer_parts`).
 
 use std::sync::Arc;
 
@@ -19,16 +22,23 @@ use crate::at_rest::Keyring;
 use crate::http::{BoxError, ResBody, drain, full};
 use crate::store::{Store, StoreError, Upload};
 use crate::time;
+use crate::transfer_parts;
 
 pub const PATH: &str = "/v1/transfer";
 /// Transfers streamed at a time; more wait their turn. Each holds a few
 /// frames in memory.
 const MAX_TRANSFERS: usize = 64;
 const META_PARAM: &str = "meta-";
+/// Query parameters a part request adds to a signed URL. The upload token
+/// authenticates them, so they are not part of the URL's signature.
+const UPLOAD_PARAM: &str = "upload";
+const PART_PARAM: &str = "part";
+const OP_PARAM: &str = "op";
 
 pub struct TransferSigner {
   /// The current key's first; any of them verifies.
   keys: Vec<[u8; 32]>,
+  pub(crate) keyring: Arc<Keyring>,
   slots: Arc<tokio::sync::Semaphore>,
 }
 
@@ -39,6 +49,16 @@ pub struct Grant {
   pub metadata: Vec<(String, String)>,
 }
 
+/// A URL whose signature holds, with what a part request added to it.
+#[derive(Debug, PartialEq)]
+pub struct Signed {
+  pub grant: Grant,
+  pub expires_ms: i64,
+  pub upload: Option<String>,
+  pub part: Option<String>,
+  pub op: Option<String>,
+}
+
 #[derive(Debug, PartialEq)]
 pub enum Refusal {
   Expired,
@@ -51,9 +71,10 @@ fn field(mac: &mut Hmac<Sha256>, value: &str) {
 }
 
 impl TransferSigner {
-  pub fn new(keyring: &Keyring) -> Self {
+  pub fn new(keyring: &Arc<Keyring>) -> Self {
     Self {
       keys: keyring.url_keys(),
+      keyring: keyring.clone(),
       slots: Arc::new(tokio::sync::Semaphore::new(MAX_TRANSFERS)),
     }
   }
@@ -102,15 +123,28 @@ impl TransferSigner {
   }
 
   pub fn verify(&self, method: &str, query: &str, now_ms: i64) -> Result<Grant, Refusal> {
+    let signed = self.check(method, query)?;
+    if now_ms > signed.expires_ms {
+      return Err(Refusal::Expired);
+    }
+    Ok(signed.grant)
+  }
+
+  /// The signature only; the caller decides whether the URL's time counts.
+  pub fn check(&self, method: &str, query: &str) -> Result<Signed, Refusal> {
     let mut key = None;
     let mut expires = None;
     let mut signature = None;
     let mut metadata = Vec::new();
+    let (mut upload, mut part, mut op) = (None, None, None);
     for (name, value) in url::form_urlencoded::parse(query.as_bytes()) {
       match name.as_ref() {
         "key" if key.is_none() => key = Some(value.into_owned()),
         "expires" if expires.is_none() => expires = Some(value.into_owned()),
         "sig" if signature.is_none() => signature = Some(value.into_owned()),
+        UPLOAD_PARAM if upload.is_none() => upload = Some(value.into_owned()),
+        PART_PARAM if part.is_none() => part = Some(value.into_owned()),
+        OP_PARAM if op.is_none() => op = Some(value.into_owned()),
         other => match other.strip_prefix(META_PARAM) {
           Some(meta) => metadata.push((meta.to_string(), value.into_owned())),
           None => return Err(Refusal::BadSignature),
@@ -130,15 +164,34 @@ impl TransferSigner {
       return Err(Refusal::BadSignature);
     }
     let expires_ms: i64 = expires.parse().map_err(|_| Refusal::BadSignature)?;
-    if now_ms > expires_ms {
-      return Err(Refusal::Expired);
-    }
-    Ok(Grant { key, metadata })
+    Ok(Signed {
+      grant: Grant { key, metadata },
+      expires_ms,
+      upload,
+      part,
+      op,
+    })
+  }
+
+  /// HMAC of `data` under the current key, for upload tokens.
+  pub(crate) fn tag(&self, data: &[u8]) -> Vec<u8> {
+    let mut mac = Hmac::<Sha256>::new_from_slice(&self.keys[0]).expect("HMAC takes any key length");
+    mac.update(data);
+    mac.finalize().into_bytes().to_vec()
+  }
+
+  /// Whether `tag` is `data`'s under any key.
+  pub(crate) fn tag_holds(&self, data: &[u8], tag: &[u8]) -> bool {
+    self.keys.iter().any(|key| {
+      let mut mac = Hmac::<Sha256>::new_from_slice(key).expect("HMAC takes any key length");
+      mac.update(data);
+      mac.verify_slice(tag).is_ok()
+    })
   }
 }
 
 /// An S3-style error document, so a device logs what S3 would have said.
-fn s3_error(status: u16, code: &str, message: &str) -> Response<ResBody> {
+pub(crate) fn s3_error(status: u16, code: &str, message: &str) -> Response<ResBody> {
   let body = format!(
     "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<Error><Code>{code}</Code><Message>{message}</Message></Error>"
   );
@@ -150,7 +203,7 @@ fn s3_error(status: u16, code: &str, message: &str) -> Response<ResBody> {
     .expect("static response parts are valid")
 }
 
-fn store_refusal(error: StoreError) -> Response<ResBody> {
+pub(crate) fn store_refusal(error: StoreError) -> Response<ResBody> {
   match error {
     StoreError::NotFound => s3_error(404, "NoSuchKey", "The specified key does not exist."),
     StoreError::Service { status, code, .. } if (400..500).contains(&status) => {
@@ -167,7 +220,7 @@ fn store_refusal(error: StoreError) -> Response<ResBody> {
   }
 }
 
-fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option<String> {
+pub(crate) fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option<String> {
   parts
     .headers
     .get(name)
@@ -180,37 +233,110 @@ pub async fn serve(
   parts: &Parts,
   body: Incoming,
 ) -> Response<ResBody> {
-  let method = parts.method.as_str();
-  let grant = match signer.verify(method, parts.uri.query().unwrap_or(""), time::now_ms()) {
-    Ok(grant) if parts.method == Method::GET || parts.method == Method::PUT => grant,
-    Ok(_) | Err(Refusal::BadSignature) => {
+  // A GET URL reads; a PUT URL writes, in one request or in parts.
+  let signed_for = if parts.method == Method::GET {
+    "GET"
+  } else {
+    "PUT"
+  };
+  let signed = match signer.check(signed_for, parts.uri.query().unwrap_or("")) {
+    Ok(signed) => signed,
+    Err(_) => {
       drain(body).await;
-      return s3_error(
-        403,
-        "SignatureDoesNotMatch",
-        "The request signature we calculated does not match the signature you provided.",
-      );
-    }
-    Err(Refusal::Expired) => {
-      drain(body).await;
-      return s3_error(403, "AccessDenied", "Request has expired");
+      return bad_signature();
     }
   };
+  // Parts carry an upload token, which outlives the URL that started them.
+  if signed.upload.is_none() && time::now_ms() > signed.expires_ms {
+    drain(body).await;
+    return s3_error(403, "AccessDenied", "Request has expired");
+  }
   let slot = signer
     .slots
     .clone()
     .acquire_owned()
     .await
     .expect("the semaphore is never closed");
-  if parts.method == Method::GET {
-    drain(body).await;
-    return download(store, &grant.key, slot).await;
-  }
-  let response = upload(store, grant, parts, body).await;
+  let response = match (&parts.method, signed.upload.is_some()) {
+    (&Method::GET, false) => {
+      drain(body).await;
+      return download(store, &signed.grant.key, slot).await;
+    }
+    (&Method::PUT, false) => upload(store, signed.grant, parts, body).await,
+    (&Method::POST, false) => transfer_parts::start(store, signer, signed.grant, parts, body).await,
+    (&Method::PUT | &Method::POST | &Method::DELETE, true) => {
+      transfer_parts::serve(store, signer, signed, parts, body).await
+    }
+    _ => {
+      drain(body).await;
+      bad_signature()
+    }
+  };
   drop(slot);
   response
 }
 
+pub(crate) fn bad_signature() -> Response<ResBody> {
+  s3_error(
+    403,
+    "SignatureDoesNotMatch",
+    "The request signature we calculated does not match the signature you provided.",
+  )
+}
+
+/// The `x-amz-meta-*` a request sent, when they are exactly the signed ones.
+pub(crate) fn signed_metadata(
+  parts: &Parts,
+  grant: Grant,
+) -> Result<Vec<(String, String)>, Box<Response<ResBody>>> {
+  let mut sent: Vec<(String, String)> = parts
+    .headers
+    .iter()
+    .filter_map(|(name, value)| {
+      name.as_str().strip_prefix("x-amz-meta-").map(|n| {
+        (
+          n.to_string(),
+          value.as_bytes().iter().map(|&b| b as char).collect(),
+        )
+      })
+    })
+    .collect();
+  sent.sort();
+  let mut signed = grant.metadata;
+  signed.sort();
+  if sent != signed {
+    return Err(Box::new(s3_error(
+      403,
+      "AccessDenied",
+      "There were headers present in the request which were not signed",
+    )));
+  }
+  Ok(signed)
+}
+
+/// A request body as a store stream.
+pub(crate) fn body_stream(body: Incoming) -> crate::store::ByteStream {
+  Box::pin(
+    body
+      .into_data_stream()
+      .map(|chunk: Result<Bytes, hyper::Error>| {
+        chunk.map_err(|e| StoreError::Transport(e.to_string()))
+      }),
+  )
+}
+
+pub(crate) fn content_length(parts: &Parts) -> Option<u64> {
+  header_text(parts, header::CONTENT_LENGTH).and_then(|v| v.trim().parse().ok())
+}
+
+pub(crate) fn missing_length() -> Response<ResBody> {
+  s3_error(
+    411,
+    "MissingContentLength",
+    "You must provide the Content-Length HTTP header.",
+  )
+}
+
 async fn download(
   store: &Arc<dyn Store>,
   key: &str,
@@ -259,48 +385,23 @@ async fn upload(
   parts: &Parts,
   body: Incoming,
 ) -> Response<ResBody> {
-  let mut sent: Vec<(String, String)> = parts
-    .headers
-    .iter()
-    .filter_map(|(name, value)| {
-      name.as_str().strip_prefix("x-amz-meta-").map(|n| {
-        (
-          n.to_string(),
-          value.as_bytes().iter().map(|&b| b as char).collect(),
-        )
-      })
-    })
-    .collect();
-  sent.sort();
-  let mut signed = grant.metadata;
-  signed.sort();
-  if sent != signed {
+  let key = grant.key.clone();
+  let signed = match signed_metadata(parts, grant) {
+    Ok(signed) => signed,
+    Err(refusal) => {
+      drain(body).await;
+      return *refusal;
+    }
+  };
+  let Some(length) = content_length(parts) else {
     drain(body).await;
-    return s3_error(
-      403,
-      "AccessDenied",
-      "There were headers present in the request which were not signed",
-    );
-  }
-  let Some(length) = header_text(parts, header::CONTENT_LENGTH).and_then(|v| v.trim().parse().ok())
-  else {
-    drain(body).await;
-    return s3_error(
-      411,
-      "MissingContentLength",
-      "You must provide the Content-Length HTTP header.",
-    );
+    return missing_length();
   };
   let content_type = header_text(parts, header::CONTENT_TYPE);
-  let stream = body
-    .into_data_stream()
-    .map(|chunk: Result<Bytes, hyper::Error>| {
-      chunk.map_err(|e| StoreError::Transport(e.to_string()))
-    });
   match store
     .put_object_stream(
-      &grant.key,
-      Box::pin(stream),
+      &key,
+      body_stream(body),
       Upload {
         length,
         content_type: content_type.as_deref(),
@@ -324,7 +425,9 @@ mod tests {
   use super::*;
 
   fn signer() -> TransferSigner {
-    TransferSigner::new(&Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap())
+    TransferSigner::new(&Arc::new(
+      Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap(),
+    ))
   }
 
   fn query(url: &str) -> &str {
@@ -377,16 +480,16 @@ mod tests {
   fn an_old_key_still_verifies_after_rotation() {
     let old = signer();
     let url = old.url("http://h", "GET", "k", 50, &[]);
-    let rotated = TransferSigner::new(
-      &Keyring::parse(
+    let rotated = TransferSigner::new(&Arc::new(
+      Keyring::parse(
         "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=,k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
       )
       .unwrap(),
-    );
+    ));
     assert!(rotated.verify("GET", query(&url), 0).is_ok());
-    let removed = TransferSigner::new(
-      &Keyring::parse("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").unwrap(),
-    );
+    let removed = TransferSigner::new(&Arc::new(
+      Keyring::parse("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").unwrap(),
+    ));
     assert_eq!(
       removed.verify("GET", query(&url), 0),
       Err(Refusal::BadSignature)

donut-sync/src/transfer_parts.rs

@@ -0,0 +1,357 @@
+//! Uploads in parts through this server, for a store sealed at rest. A proxy
+//! in front of the server (Cloudflare) limits each request's size and time,
+//! so a large file goes up in `PART_SIZE` pieces on the signed PUT URL that
+//! presign-upload hands out as `transfer.url`:
+//!
+//! ```text
+//! POST   url                     x-donut-upload-length: N, the signed x-amz-meta-*, Content-Type
+//!        -> 200 {"upload": T, "partSize": P, "parts": n}
+//! PUT    url&upload=T&part=i     i in 1..=n, exactly P bytes (the last part: the rest)
+//!        -> 200, ETag
+//! POST   url&upload=T&op=complete  {"parts": [ETag of part 1, ...]}  -> 200
+//! DELETE url&upload=T            -> 204
+//! ```
+//!
+//! The token T is signed and names the object, the storage upload and the
+//! sealing salt, so the server keeps no state between requests. A part may be
+//! sent again. Each part is sealed as whole frames of one object, numbered
+//! where the part starts, so the joined object reads like any sealed object.
+
+use base64::Engine;
+use bytes::Bytes;
+use http_body_util::BodyExt;
+use hyper::body::Incoming;
+use hyper::http::request::Parts;
+use hyper::{Method, Response, StatusCode, header};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use crate::at_rest::{self, FRAME, Salt, Sealer};
+use crate::http::{ResBody, drain, full, json_response};
+use crate::store::Store;
+use crate::time;
+use crate::transfer::{
+  Grant, Signed, TransferSigner, bad_signature, body_stream, content_length, header_text,
+  missing_length, s3_error, signed_metadata, store_refusal,
+};
+
+/// Plaintext bytes per part: far under a proxy's request limit, and sent
+/// within its timeout on a slow uplink. A whole number of frames, and above
+/// S3's 5 MB minimum part.
+pub const PART_SIZE: u64 = 8 * 1024 * 1024;
+/// How long an upload may take once started.
+const UPLOAD_TTL_MS: i64 = 24 * 3600 * 1000;
+pub const LENGTH_HEADER: &str = "x-donut-upload-length";
+/// S3's most parts in one upload.
+const MAX_PARTS: u64 = 10_000;
+const MAX_COMPLETE_BODY: usize = 1024 * 1024;
+const TOKEN_CONTEXT: &[u8] = b"donut-sync upload token v1\n";
+
+const _: () = assert!(PART_SIZE.is_multiple_of(FRAME as u64) && PART_SIZE >= 5 * 1024 * 1024);
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+struct Token {
+  key: String,
+  upload_id: String,
+  key_id: String,
+  salt: String,
+  length: u64,
+  part_size: u64,
+  expires_ms: i64,
+}
+
+impl Token {
+  fn parts(&self) -> u64 {
+    self.length.div_ceil(self.part_size).max(1)
+  }
+
+  fn part_len(&self, part: u64) -> u64 {
+    if part < self.parts() {
+      self.part_size
+    } else {
+      self.length - (self.parts() - 1) * self.part_size
+    }
+  }
+
+  fn salt(&self) -> Option<Salt> {
+    hex::decode(&self.salt).ok()?.try_into().ok()
+  }
+
+  fn seal(&self, signer: &TransferSigner) -> String {
+    let json = serde_json::to_vec(self).expect("a token serializes");
+    let tag = signer.tag(&[TOKEN_CONTEXT, &json].concat());
+    format!(
+      "{}.{}",
+      base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&json),
+      hex::encode(tag)
+    )
+  }
+
+  /// The token, when it is ours, for `key`, and not expired.
+  fn open(signer: &TransferSigner, text: &str, key: &str, now_ms: i64) -> Option<Self> {
+    let (body, tag) = text.split_once('.')?;
+    let json = base64::engine::general_purpose::URL_SAFE_NO_PAD
+      .decode(body)
+      .ok()?;
+    if !signer.tag_holds(&[TOKEN_CONTEXT, &json].concat(), &hex::decode(tag).ok()?) {
+      return None;
+    }
+    let token: Self = serde_json::from_slice(&json).ok()?;
+    (token.key == key && now_ms <= token.expires_ms).then_some(token)
+  }
+}
+
+fn invalid(code: &str, message: &str) -> Response<ResBody> {
+  s3_error(400, code, message)
+}
+
+/// Starts an upload of `x-donut-upload-length` bytes.
+pub async fn start(
+  store: &Arc<dyn Store>,
+  signer: &TransferSigner,
+  grant: Grant,
+  parts: &Parts,
+  body: Incoming,
+) -> Response<ResBody> {
+  drain(body).await;
+  let key = grant.key.clone();
+  let metadata = match signed_metadata(parts, grant) {
+    Ok(metadata) => metadata,
+    Err(refusal) => return *refusal,
+  };
+  let Some(length) = header_text(parts, LENGTH_HEADER).and_then(|v| v.trim().parse::<u64>().ok())
+  else {
+    return invalid(
+      "MissingUploadLength",
+      "Send the file size as x-donut-upload-length.",
+    );
+  };
+  if length.div_ceil(PART_SIZE) > MAX_PARTS {
+    return invalid(
+      "EntityTooLarge",
+      "The file is too large to upload in parts.",
+    );
+  }
+  let content_type = header_text(parts, header::CONTENT_TYPE);
+  let upload_id = match store
+    .create_multipart(&key, content_type.as_deref(), &metadata)
+    .await
+  {
+    Ok(id) => id,
+    Err(e) => return store_refusal(e),
+  };
+  let token = Token {
+    key,
+    upload_id,
+    key_id: signer.keyring.current().id().to_string(),
+    salt: hex::encode(at_rest::new_salt()),
+    length,
+    part_size: PART_SIZE,
+    expires_ms: time::now_ms() + UPLOAD_TTL_MS,
+  };
+  json_response(
+    200,
+    &serde_json::json!({
+      "upload": token.seal(signer),
+      "partSize": token.part_size,
+      "parts": token.parts(),
+    }),
+  )
+}
+
+/// A part, the completion or the abort of a started upload.
+pub async fn serve(
+  store: &Arc<dyn Store>,
+  signer: &TransferSigner,
+  signed: Signed,
+  parts: &Parts,
+  body: Incoming,
+) -> Response<ResBody> {
+  let token = signed
+    .upload
+    .as_deref()
+    .and_then(|text| Token::open(signer, text, &signed.grant.key, time::now_ms()));
+  let Some(token) = token else {
+    drain(body).await;
+    return bad_signature();
+  };
+  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")) => complete(store, &token, body).await,
+    (&Method::DELETE, None) => {
+      drain(body).await;
+      match store.abort_multipart(&token.key, &token.upload_id).await {
+        Ok(()) => Response::builder()
+          .status(StatusCode::NO_CONTENT)
+          .body(full(Bytes::new()))
+          .expect("static response parts are valid"),
+        Err(e) => store_refusal(e),
+      }
+    }
+    _ => {
+      drain(body).await;
+      bad_signature()
+    }
+  }
+}
+
+async fn part(
+  store: &Arc<dyn Store>,
+  signer: &TransferSigner,
+  token: &Token,
+  number: Option<&str>,
+  parts: &Parts,
+  body: Incoming,
+) -> Response<ResBody> {
+  let Some(number) = number
+    .and_then(|n| n.parse::<u64>().ok())
+    .filter(|n| (1..=token.parts()).contains(n))
+  else {
+    drain(body).await;
+    return invalid(
+      "InvalidPart",
+      "The part number is not one of this upload's.",
+    );
+  };
+  let Some(length) = content_length(parts) else {
+    drain(body).await;
+    return missing_length();
+  };
+  let expected = token.part_len(number);
+  if length != expected {
+    drain(body).await;
+    return invalid(
+      "InvalidPart",
+      &format!("Part {number} must be {expected} bytes, not {length}."),
+    );
+  }
+  let first_frame = (number - 1) * token.part_size / FRAME as u64;
+  let sealer = token
+    .salt()
+    .and_then(|salt| Sealer::for_part(&signer.keyring, &token.key_id, &salt, first_frame));
+  let Some(sealer) = sealer else {
+    drain(body).await;
+    return invalid("InvalidUpload", "This upload's key is no longer held.");
+  };
+  let last = number == token.parts();
+  let sealed_len = at_rest::sealed_part_len(&token.key_id, length, number == 1, last);
+  let sealed = at_rest::seal_part_stream(sealer, last, body_stream(body));
+  match store
+    .upload_part(
+      &token.key,
+      &token.upload_id,
+      number as u32,
+      sealed,
+      sealed_len,
+    )
+    .await
+  {
+    Ok(etag) => Response::builder()
+      .status(200)
+      .header(header::ETAG, etag)
+      .header(header::CONTENT_LENGTH, 0)
+      .body(full(Bytes::new()))
+      .unwrap_or_else(|_| s3_error(500, "InternalError", "Storage gave an invalid ETag.")),
+    Err(e) => store_refusal(e),
+  }
+}
+
+#[derive(Deserialize)]
+struct Completion {
+  parts: Vec<String>,
+}
+
+async fn complete(store: &Arc<dyn Store>, token: &Token, body: Incoming) -> Response<ResBody> {
+  let bytes = match http_body_util::Limited::new(body, MAX_COMPLETE_BODY)
+    .collect()
+    .await
+  {
+    Ok(collected) => collected.to_bytes(),
+    Err(_) => return invalid("MalformedXML", "The completion body is too large."),
+  };
+  let Ok(completion) = serde_json::from_slice::<Completion>(&bytes) else {
+    return invalid("MalformedXML", "Send {\"parts\": [ETag, ...]}.");
+  };
+  if completion.parts.len() as u64 != token.parts() {
+    return invalid(
+      "InvalidPart",
+      &format!("This upload has {} parts.", token.parts()),
+    );
+  }
+  match store
+    .complete_multipart(&token.key, &token.upload_id, &completion.parts)
+    .await
+  {
+    Ok(()) => Response::builder()
+      .status(200)
+      .header(header::CONTENT_LENGTH, 0)
+      .body(full(Bytes::new()))
+      .expect("static response parts are valid"),
+    Err(e) => store_refusal(e),
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use crate::at_rest::Keyring;
+
+  fn signer(keys: &str) -> TransferSigner {
+    TransferSigner::new(&Arc::new(Keyring::parse(keys).unwrap()))
+  }
+
+  const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+
+  fn token(length: u64) -> Token {
+    Token {
+      key: "users/u/a".to_string(),
+      upload_id: "id".to_string(),
+      key_id: "k1".to_string(),
+      salt: hex::encode([7u8; 32]),
+      length,
+      part_size: PART_SIZE,
+      expires_ms: 1000,
+    }
+  }
+
+  #[test]
+  fn tokens_are_bound_to_their_key_signer_and_time() {
+    let signer = signer(K1);
+    let text = token(5).seal(&signer);
+    assert_eq!(
+      Token::open(&signer, &text, "users/u/a", 1000),
+      Some(token(5))
+    );
+    assert_eq!(Token::open(&signer, &text, "users/u/b", 0), None);
+    assert_eq!(Token::open(&signer, &text, "users/u/a", 1001), None);
+    let other = signer_other();
+    assert_eq!(Token::open(&other, &text, "users/u/a", 0), None);
+    let (body, tag) = text.split_once('.').unwrap();
+    let mut json = base64::engine::general_purpose::URL_SAFE_NO_PAD
+      .decode(body)
+      .unwrap();
+    let at = json.iter().position(|&b| b == b'5').unwrap();
+    json[at] = b'6';
+    let forged = format!(
+      "{}.{tag}",
+      base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json)
+    );
+    assert_eq!(Token::open(&signer, &forged, "users/u/a", 0), None);
+  }
+
+  fn signer_other() -> TransferSigner {
+    signer("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=")
+  }
+
+  #[test]
+  fn parts_split_the_length() {
+    assert_eq!((token(0).parts(), token(0).part_len(1)), (1, 0));
+    assert_eq!(token(PART_SIZE).parts(), 1);
+    let t = token(2 * PART_SIZE + 3);
+    assert_eq!(t.parts(), 3);
+    assert_eq!(
+      (t.part_len(1), t.part_len(2), t.part_len(3)),
+      (PART_SIZE, PART_SIZE, 3)
+    );
+  }
+}

donut-sync/src/verify.rs

@@ -12,7 +12,6 @@ use futures_util::StreamExt;
 use crate::at_rest::{self, Keyring, MARKER};
 use crate::config::Config;
 use crate::log;
-use crate::migrate::TMP_PREFIX;
 use crate::sealed_store::sealed_by;
 use crate::store::{ListInput, Store, StoreError};
 
@@ -135,8 +134,9 @@ pub struct Report {
   pub unknown_key: u64,
   /// A marker of another format version, or a size no sealed object has.
   pub invalid: u64,
-  /// Leftover temporary copies of an interrupted migration.
-  pub temporary: u64,
+  /// This server's own objects: records of direct uploads, and leftovers of
+  /// an interrupted migration.
+  pub internal: u64,
   /// Gone between the listing and the HEAD.
   pub vanished: u64,
   pub sampled: u64,
@@ -160,10 +160,10 @@ impl Report {
       ));
     }
     line.push_str(&format!(
-      " unknown_key={} invalid={} temporary={} vanished={} sampled={} sample_failed={}",
+      " unknown_key={} invalid={} internal={} vanished={} sampled={} sample_failed={}",
       self.unknown_key,
       self.invalid,
-      self.temporary,
+      self.internal,
       self.vanished,
       self.sampled,
       self.sample_failed
@@ -252,12 +252,12 @@ pub async fn verify(
       })
       .await
       .map_err(|e| format!("Listing failed: {e}"))?;
-    let (temporary, keys): (Vec<String>, Vec<String>) = page
+    let (internal, keys): (Vec<String>, Vec<String>) = page
       .contents
       .into_iter()
       .map(|object| object.key)
-      .partition(|key| key.starts_with(TMP_PREFIX));
-    report.temporary += temporary.len() as u64;
+      .partition(|key| key.starts_with(crate::sealer::INTERNAL_ROOT));
+    report.internal += internal.len() as u64;
     let classified: Vec<(String, Classified)> = futures_util::stream::iter(keys)
       .map(|key| {
         let store = store.clone();
@@ -394,7 +394,7 @@ mod tests {
       .put_object("c", b"ccc".to_vec(), "text/plain", &[])
       .await
       .unwrap();
-    raw.set(&format!("{TMP_PREFIX}left"), "x");
+    raw.set(&format!("{}left", crate::migrate::TMP_PREFIX), "x");
     let keys = keyring(&format!("{K1},{K2}"));
 
     let before = verify(raw.clone(), keys.clone(), &Options::default())
@@ -408,7 +408,7 @@ mod tests {
     assert_eq!(before.sealed["k2"].objects, 1);
     assert_eq!(before.sealed["k2"].plain_bytes, 3);
     assert_eq!(
-      (before.temporary, before.sampled, before.sample_failed),
+      (before.internal, before.sampled, before.sample_failed),
       (1, 1, 0)
     );
 

donut-sync/tests/at_rest.rs

@@ -33,7 +33,8 @@ async fn call(server: &Server, route: &str, body: Value) -> (u16, Value) {
   .await
 }
 
-/// What the app does: presign, then PUT with the echoed metadata.
+/// What a newer app does with a small file: presign, then PUT to the
+/// transfer URL with the echoed metadata.
 async fn upload(server: &Server, key: &str, body: Vec<u8>, updated_at: Option<&str>) -> String {
   let mut request = json!({ "key": key, "contentType": "application/octet-stream" });
   if let Some(updated_at) = updated_at {
@@ -41,7 +42,15 @@ async fn upload(server: &Server, key: &str, body: Vec<u8>, updated_at: Option<&s
   }
   let (status, presign) = call(server, "presign-upload", request).await;
   assert_eq!(status, 200, "{presign}");
-  let url = presign["url"].as_str().unwrap().to_string();
+  // Older apps keep the direct storage URL.
+  assert!(
+    presign["url"]
+      .as_str()
+      .unwrap()
+      .starts_with(common::S3_ENDPOINT)
+  );
+  assert_eq!(presign["transfer"]["partSize"], json!(8 * 1024 * 1024));
+  let url = presign["transfer"]["url"].as_str().unwrap().to_string();
   let mut put = http()
     .put(&url)
     .header("Content-Length", body.len().to_string())
@@ -111,7 +120,11 @@ async fn transfers_go_through_the_server_and_are_sealed_in_storage() {
       .await
       .unwrap(),
   );
-  assert_eq!(readyz, json!({ "status": "ready", "s3": true }));
+  // Older apps still upload to storage directly, so its address is named.
+  assert_eq!(
+    readyz,
+    json!({ "status": "ready", "s3": true, "storageEndpoint": common::S3_ENDPOINT })
+  );
   assert!(
     server
       .logs()
@@ -132,8 +145,13 @@ async fn batches_and_downloads_keep_the_s3_shapes() {
   for item in batch["items"].as_array().unwrap() {
     let key = item["key"].as_str().unwrap();
     assert!(item["expiresAt"].is_string());
+    // "a" as a newer app sends it, "b" as an older one.
+    let url = match key {
+      "a" => item["transfer"]["url"].as_str().unwrap(),
+      _ => item["url"].as_str().unwrap(),
+    };
     let response = http()
-      .put(item["url"].as_str().unwrap())
+      .put(url)
       .header("Content-Length", key.len().to_string())
       .body(key.to_string())
       .send()
@@ -178,7 +196,7 @@ async fn transfer_urls_refuse_what_s3_would_refuse() {
     json!({ "key": "k", "metadata": { "updated-at": "5" } }),
   )
   .await;
-  let url = presign["url"].as_str().unwrap();
+  let url = presign["transfer"]["url"].as_str().unwrap();
   let put = |url: String| client.put(url).header("Content-Length", "1").body("x");
 
   // The signed metadata must be sent, exactly.
@@ -538,3 +556,220 @@ async fn a_bad_key_stops_the_server_at_start() {
   assert_eq!(code, 1);
   assert!(out.contains("must be 32 bytes"), "{out}");
 }
+
+const PART: usize = 8 * 1024 * 1024;
+
+/// Presign as a newer app does, and return (transfer URL, signed metadata).
+async fn presign_transfer(server: &Server, key: &str) -> (String, Value) {
+  let (status, presign) = call(
+    server,
+    "presign-upload",
+    json!({ "key": key, "contentType": "application/octet-stream", "metadata": { "updated-at": "77" } }),
+  )
+  .await;
+  assert_eq!(status, 200, "{presign}");
+  (
+    presign["transfer"]["url"].as_str().unwrap().to_string(),
+    presign["metadata"].clone(),
+  )
+}
+
+async fn start_parts(url: &str, length: usize) -> reqwest::Response {
+  http()
+    .post(url)
+    .header("x-donut-upload-length", length.to_string())
+    .header("Content-Type", "application/octet-stream")
+    .header("x-amz-meta-updated-at", "77")
+    .send()
+    .await
+    .unwrap()
+}
+
+fn part_url(url: &str, upload: &str, part: usize) -> String {
+  format!("{url}&upload={upload}&part={part}")
+}
+
+#[tokio::test]
+async fn large_files_upload_in_parts_into_one_sealed_object() {
+  let (server, bucket) = sealed_server("parts", K1).await;
+  let key = "profiles/p/files/large";
+  let plain = sample(2 * PART + 12_345);
+  let (url, _) = presign_transfer(&server, key).await;
+
+  let started = start_parts(&url, plain.len()).await;
+  assert_eq!(started.status(), 200);
+  let started: Value = started.json().await.unwrap();
+  assert_eq!(
+    (started["partSize"].as_u64(), started["parts"].as_u64()),
+    (Some(PART as u64), Some(3))
+  );
+  let upload = started["upload"].as_str().unwrap();
+
+  let mut etags = Vec::new();
+  for (i, chunk) in plain.chunks(PART).enumerate() {
+    let response = http()
+      .put(part_url(&url, upload, i + 1))
+      .header("Content-Length", chunk.len().to_string())
+      .body(chunk.to_vec())
+      .send()
+      .await
+      .unwrap();
+    assert_eq!(response.status(), 200, "{}", response.text().await.unwrap());
+    etags.push(response.headers()["etag"].to_str().unwrap().to_string());
+  }
+  // A part sent again replaces itself.
+  let again = http()
+    .put(part_url(&url, upload, 2))
+    .header("Content-Length", PART.to_string())
+    .body(plain[PART..2 * PART].to_vec())
+    .send()
+    .await
+    .unwrap();
+  etags[1] = again.headers()["etag"].to_str().unwrap().to_string();
+  let done = http()
+    .post(format!("{url}&upload={upload}&op=complete"))
+    .json(&json!({ "parts": etags }))
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(done.status(), 200, "{}", done.text().await.unwrap());
+
+  let raw = raw_store(&bucket);
+  let head = raw.head_object(key).await.unwrap();
+  assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1"));
+  let (_, stat) = call(&server, "stat", json!({ "key": key })).await;
+  assert_eq!(stat["size"], json!(plain.len()));
+  assert_eq!(stat["metadata"]["updated-at"], "77");
+  assert_eq!(download(&server, key).await, (200, plain));
+}
+
+#[tokio::test]
+async fn part_requests_refuse_what_does_not_fit_the_upload() {
+  let (server, _) = sealed_server("partrefuse", K1).await;
+  let (url, _) = presign_transfer(&server, "k").await;
+  // Unsigned metadata, and no length.
+  let response = http()
+    .post(&url)
+    .header("x-donut-upload-length", "10")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  let response = http()
+    .post(&url)
+    .header("x-amz-meta-updated-at", "77")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 400);
+
+  let started: Value = start_parts(&url, PART + 1).await.json().await.unwrap();
+  let upload = started["upload"].as_str().unwrap().to_string();
+  let put = |part: usize, len: usize| {
+    http()
+      .put(part_url(&url, &upload, part))
+      .header("Content-Length", len.to_string())
+      .body(vec![0u8; len])
+      .send()
+  };
+  assert_eq!(
+    put(1, PART - 1).await.unwrap().status(),
+    400,
+    "a short part"
+  );
+  assert_eq!(put(3, 1).await.unwrap().status(), 400, "no part 3");
+  assert_eq!(put(0, 1).await.unwrap().status(), 400, "no part 0");
+  // A forged token, or this token on another key's URL.
+  let forged = upload.replacen('.', ".0", 1);
+  let response = http()
+    .put(part_url(&url, &forged, 2))
+    .header("Content-Length", "1")
+    .body(vec![0u8])
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  let (other, _) = presign_transfer(&server, "other").await;
+  let response = http()
+    .put(part_url(&other, &upload, 2))
+    .header("Content-Length", "1")
+    .body(vec![0u8])
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  // The wrong number of parts on completion.
+  let one = put(2, 1).await.unwrap();
+  let etag = one.headers()["etag"].to_str().unwrap().to_string();
+  let done = http()
+    .post(format!("{url}&upload={upload}&op=complete"))
+    .json(&json!({ "parts": [etag] }))
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(done.status(), 400);
+  // Aborted, the upload takes no more parts.
+  let aborted = http()
+    .delete(format!("{url}&upload={upload}"))
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(aborted.status(), 204);
+  assert_eq!(download(&server, "k").await.0, 404);
+}
+
+#[tokio::test]
+async fn older_apps_upload_directly_and_the_sealer_seals_it() {
+  let (server, bucket) = sealed_server("direct", K1).await;
+  let key = "profiles/p/files/old-app";
+  let (status, presign) = call(
+    &server,
+    "presign-upload",
+    json!({ "key": key, "metadata": { "updated-at": "5" } }),
+  )
+  .await;
+  assert_eq!(status, 200);
+  let plain = sample(300_000);
+  let response = http()
+    .put(presign["url"].as_str().unwrap())
+    .header("Content-Length", plain.len().to_string())
+    .header("x-amz-meta-updated-at", "5")
+    .body(plain.clone())
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 200);
+  let raw: std::sync::Arc<dyn Store> = std::sync::Arc::new(raw_store(&bucket));
+  assert!(
+    raw
+      .head_object(key)
+      .await
+      .unwrap()
+      .metadata_value("donut-at-rest")
+      .is_none()
+  );
+  // Readable before it is sealed.
+  assert_eq!(download(&server, key).await, (200, plain.clone()));
+
+  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
+    .unwrap();
+  let report = sealer
+    .pending_pass(donut_sync::time::now_ms())
+    .await
+    .unwrap();
+  assert_eq!((report.records, report.sealed, report.failed), (1, 1, 0));
+  let head = raw.head_object(key).await.unwrap();
+  assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1"));
+  assert_eq!(head.metadata_value("updated-at"), Some("5"));
+  assert_eq!(download(&server, key).await, (200, plain));
+  // The record stays until its URL is dead.
+  let (_, list) = call(&server, "list", json!({ "prefix": "" })).await;
+  assert_eq!(list["objects"].as_array().unwrap().len(), 1, "{list}");
+  let report = sealer
+    .pending_pass(donut_sync::time::now_ms() + 2 * 3600 * 1000)
+    .await
+    .unwrap();
+  assert_eq!(report.records_done, 1);
+}

src-tauri/src/sync/client.rs

@@ -23,6 +23,29 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);
 /// only when nothing arrives for a full minute.
 const READ_TIMEOUT: Duration = Duration::from_secs(60);
 
+/// 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);
+/// The size of a file uploaded in parts, which the server checks every part against.
+const UPLOAD_LENGTH_HEADER: &str = "x-donut-upload-length";
+
+/// The server's answer to the start of an upload in parts.
+#[derive(serde::Deserialize)]
+struct PartsStarted {
+  upload: String,
+  #[serde(rename = "partSize")]
+  part_size: u64,
+  parts: u64,
+}
+
+/// `url` with `pairs` added to its query.
+fn with_query(url: &str, pairs: &[(&str, &str)]) -> SyncResult<String> {
+  let mut url = reqwest::Url::parse(url)
+    .map_err(|e| SyncError::NetworkError(format!("Invalid transfer URL: {e}")))?;
+  url.query_pairs_mut().extend_pairs(pairs);
+  Ok(url.into())
+}
+
 #[derive(Clone)]
 pub struct SyncClient {
   client: Client,
@@ -227,42 +250,201 @@ impl SyncClient {
     Ok(all_objects)
   }
 
+  /// Uploads to a presign answer: through `transfer` when the server offers
+  /// it, else to `presigned_url`.
   pub async fn upload_bytes(
     &self,
     presigned_url: &str,
+    transfer: Option<&SealedTransfer>,
     data: &[u8],
     content_type: Option<&str>,
   ) -> SyncResult<()> {
     self
-      .upload_bytes_with_metadata(presigned_url, data, content_type, None)
+      .upload_bytes_with_metadata(presigned_url, transfer, data, content_type, None)
       .await
   }
 
-  /// PUT to a presigned URL, sending `metadata` as `x-amz-meta-*` headers. These
-  /// MUST be exactly the metadata the presign signed (from
-  /// `PresignUploadResponse::metadata`) or S3 rejects the request.
+  /// `upload_bytes`, sending `metadata` as `x-amz-meta-*` headers. These MUST
+  /// be exactly the metadata the presign signed (from
+  /// `PresignUploadResponse::metadata`) or the upload is refused.
   pub async fn upload_bytes_with_metadata(
+    &self,
+    presigned_url: &str,
+    transfer: Option<&SealedTransfer>,
+    data: &[u8],
+    content_type: Option<&str>,
+    metadata: Option<&std::collections::HashMap<String, String>>,
+  ) -> SyncResult<()> {
+    match transfer {
+      Some(transfer) if data.len() as u64 > transfer.part_size => {
+        self
+          .upload_in_parts(transfer, data, content_type, metadata)
+          .await
+      }
+      Some(transfer) => {
+        self
+          .put_once(&transfer.url, data, content_type, metadata)
+          .await
+      }
+      None => {
+        self
+          .put_once(presigned_url, data, content_type, metadata)
+          .await
+      }
+    }
+  }
+
+  fn with_upload_headers(
+    mut req: reqwest::RequestBuilder,
+    content_type: Option<&str>,
+    metadata: Option<&std::collections::HashMap<String, String>>,
+  ) -> reqwest::RequestBuilder {
+    if let Some(ct) = content_type {
+      req = req.header("Content-Type", ct);
+    }
+    for (k, v) in metadata.into_iter().flatten() {
+      req = req.header(format!("x-amz-meta-{k}"), v);
+    }
+    req
+  }
+
+  /// 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,
+  ) -> SyncResult<reqwest::Response> {
+    let response = req
+      .send()
+      .await
+      .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?;
+    if !response.status().is_success() {
+      let status = response.status();
+      let body = response.text().await.unwrap_or_default();
+      return Err(SyncError::NetworkError(format!(
+        "Upload failed with status {status}: {body}"
+      )));
+    }
+    Ok(response)
+  }
+
+  /// The server's protocol for files larger than one request may carry:
+  /// start with the size, send each part (retried on its own), then join
+  /// them. A failure aborts the upload, so storage keeps nothing half done.
+  async fn upload_in_parts(
+    &self,
+    transfer: &SealedTransfer,
+    data: &[u8],
+    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 started: PartsStarted = self
+      .send_part_request(&transfer.url, start)
+      .await?
+      .json()
+      .await
+      .map_err(|e| SyncError::SerializationError(e.to_string()))?;
+    let outcome = self.send_parts(transfer, &started, data).await;
+    if outcome.is_err() {
+      let abort = with_query(&transfer.url, &[("upload", &started.upload)])?;
+      if let Err(e) = self.client.delete(&abort).send().await {
+        log::debug!("Upload abort failed err=\"{e}\"");
+      }
+    }
+    outcome
+  }
+
+  async fn send_parts(
+    &self,
+    transfer: &SealedTransfer,
+    started: &PartsStarted,
+    data: &[u8],
+  ) -> SyncResult<()> {
+    let part_size = usize::try_from(started.part_size)
+      .ok()
+      .filter(|size| *size > 0)
+      .ok_or_else(|| SyncError::InvalidData("The server gave no part size".to_string()))?;
+    let chunks: Vec<&[u8]> = data.chunks(part_size).collect();
+    if chunks.len() as u64 != started.parts {
+      return Err(SyncError::InvalidData(format!(
+        "The server expects {} parts, the file has {}",
+        started.parts,
+        chunks.len()
+      )));
+    }
+    let mut etags = Vec::with_capacity(chunks.len());
+    for (index, chunk) in chunks.iter().enumerate() {
+      let number = (index + 1).to_string();
+      let url = with_query(
+        &transfer.url,
+        &[("upload", &started.upload), ("part", &number)],
+      )?;
+      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());
+        match self.send_part_request(&url, req).await {
+          Ok(response) => {
+            break response
+              .headers()
+              .get("etag")
+              .and_then(|v| v.to_str().ok())
+              .map(str::to_string)
+              .ok_or_else(|| {
+                SyncError::InvalidData(format!("Part {number} came back without an ETag"))
+              })?;
+          }
+          Err(e) if attempt >= PART_ATTEMPTS => return Err(e),
+          Err(e) => {
+            log::debug!("Part retry part={number} attempt={attempt} err=\"{e}\"");
+            tokio::time::sleep(PART_RETRY_DELAY * attempt).await;
+          }
+        }
+      };
+      etags.push(etag);
+    }
+    let complete = with_query(
+      &transfer.url,
+      &[("upload", &started.upload), ("op", "complete")],
+    )?;
+    let req = self
+      .client
+      .post(&complete)
+      .json(&serde_json::json!({ "parts": etags }));
+    self.send_part_request(&complete, req).await.map(|_| ())
+  }
+
+  /// One PUT, as S3 takes it.
+  async fn put_once(
     &self,
     presigned_url: &str,
     data: &[u8],
     content_type: Option<&str>,
     metadata: Option<&std::collections::HashMap<String, String>>,
   ) -> SyncResult<()> {
-    let mut req = self
-      .client
-      .put(presigned_url)
-      .header("Content-Length", data.len().to_string())
-      .body(data.to_vec());
-
-    if let Some(ct) = content_type {
-      req = req.header("Content-Type", ct);
-    }
-
-    if let Some(meta) = metadata {
-      for (k, v) in meta {
-        req = req.header(format!("x-amz-meta-{k}"), v);
-      }
-    }
+    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
@@ -491,3 +673,200 @@ impl SyncClient {
     Ok(())
   }
 }
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use std::collections::HashMap;
+  use wiremock::matchers::{body_bytes, body_json, header, method, path, query_param};
+  use wiremock::{Mock, MockServer, ResponseTemplate};
+
+  fn client() -> SyncClient {
+    SyncClient::new("http://127.0.0.1:1".to_string(), "unused".to_string())
+  }
+
+  fn transfer(server: &MockServer, part_size: u64) -> SealedTransfer {
+    SealedTransfer {
+      url: format!("{}/v1/transfer?key=k&expires=1&sig=s", server.uri()),
+      part_size,
+    }
+  }
+
+  fn meta() -> HashMap<String, String> {
+    HashMap::from([("updated-at".to_string(), "7".to_string())])
+  }
+
+  #[tokio::test]
+  async fn without_a_transfer_the_presigned_url_takes_the_file() {
+    let server = MockServer::start().await;
+    Mock::given(method("PUT"))
+      .and(path("/bucket/k"))
+      .and(body_bytes(b"data".to_vec()))
+      .respond_with(ResponseTemplate::new(200))
+      .expect(1)
+      .mount(&server)
+      .await;
+    client()
+      .upload_bytes(&format!("{}/bucket/k", server.uri()), None, b"data", None)
+      .await
+      .unwrap();
+  }
+
+  #[tokio::test]
+  async fn a_file_within_one_part_goes_through_the_server_in_one_request() {
+    let server = MockServer::start().await;
+    Mock::given(method("PUT"))
+      .and(path("/v1/transfer"))
+      .and(header("x-amz-meta-updated-at", "7"))
+      .and(header("content-type", "application/json"))
+      .and(body_bytes(b"data".to_vec()))
+      .respond_with(ResponseTemplate::new(200))
+      .expect(1)
+      .mount(&server)
+      .await;
+    client()
+      .upload_bytes_with_metadata(
+        "http://storage.invalid/never",
+        Some(&transfer(&server, 4)),
+        b"data",
+        Some("application/json"),
+        Some(&meta()),
+      )
+      .await
+      .unwrap();
+  }
+
+  #[tokio::test]
+  async fn a_larger_file_goes_up_in_parts_and_a_failed_part_is_sent_again() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+      .and(path("/v1/transfer"))
+      .and(header("x-donut-upload-length", "10"))
+      .and(header("x-amz-meta-updated-at", "7"))
+      .respond_with(
+        ResponseTemplate::new(200)
+          .set_body_json(serde_json::json!({ "upload": "T.1", "partSize": 4, "parts": 3 })),
+      )
+      .expect(1)
+      .mount(&server)
+      .await;
+    for (part, body) in [("1", &b"0123"[..]), ("2", b"4567"), ("3", b"89")] {
+      if part == "2" {
+        Mock::given(method("PUT"))
+          .and(query_param("part", part))
+          .respond_with(ResponseTemplate::new(503))
+          .up_to_n_times(1)
+          .with_priority(1)
+          .mount(&server)
+          .await;
+      }
+      Mock::given(method("PUT"))
+        .and(query_param("upload", "T.1"))
+        .and(query_param("part", part))
+        .and(query_param("key", "k"))
+        .and(body_bytes(body.to_vec()))
+        .respond_with(ResponseTemplate::new(200).insert_header("etag", format!("\"e{part}\"")))
+        .expect(1)
+        .mount(&server)
+        .await;
+    }
+    Mock::given(method("POST"))
+      .and(query_param("op", "complete"))
+      .and(query_param("upload", "T.1"))
+      .and(body_json(
+        serde_json::json!({ "parts": ["\"e1\"", "\"e2\"", "\"e3\""] }),
+      ))
+      .respond_with(ResponseTemplate::new(200))
+      .expect(1)
+      .mount(&server)
+      .await;
+    client()
+      .upload_bytes_with_metadata(
+        "http://storage.invalid/never",
+        Some(&transfer(&server, 4)),
+        b"0123456789",
+        None,
+        Some(&meta()),
+      )
+      .await
+      .unwrap();
+  }
+
+  #[tokio::test]
+  async fn a_part_that_keeps_failing_aborts_the_upload() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+      .and(path("/v1/transfer"))
+      .respond_with(
+        ResponseTemplate::new(200)
+          .set_body_json(serde_json::json!({ "upload": "T.2", "partSize": 4, "parts": 2 })),
+      )
+      .mount(&server)
+      .await;
+    Mock::given(method("PUT"))
+      .respond_with(ResponseTemplate::new(500).set_body_string("no"))
+      .expect(u64::from(PART_ATTEMPTS))
+      .mount(&server)
+      .await;
+    Mock::given(method("DELETE"))
+      .and(query_param("upload", "T.2"))
+      .respond_with(ResponseTemplate::new(204))
+      .expect(1)
+      .mount(&server)
+      .await;
+    let error = client()
+      .upload_bytes(
+        "http://storage.invalid/never",
+        Some(&transfer(&server, 4)),
+        b"01234567",
+        None,
+      )
+      .await
+      .unwrap_err();
+    assert!(error.to_string().contains("500"), "{error}");
+  }
+
+  #[tokio::test]
+  async fn a_part_count_that_disagrees_with_the_file_is_refused_before_sending() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+      .respond_with(
+        ResponseTemplate::new(200)
+          .set_body_json(serde_json::json!({ "upload": "T.3", "partSize": 4, "parts": 5 })),
+      )
+      .mount(&server)
+      .await;
+    Mock::given(method("DELETE"))
+      .respond_with(ResponseTemplate::new(204))
+      .expect(1)
+      .mount(&server)
+      .await;
+    assert!(client()
+      .upload_bytes(
+        "http://x.invalid",
+        Some(&transfer(&server, 4)),
+        b"0123456",
+        None
+      )
+      .await
+      .is_err());
+  }
+
+  #[test]
+  fn answers_from_servers_that_do_not_seal_have_no_transfer() {
+    let old: PresignUploadResponse =
+      serde_json::from_str(r#"{"url":"u","expiresAt":"t"}"#).unwrap();
+    assert_eq!(old.transfer, None);
+    let new: PresignUploadBatchItemResponse = serde_json::from_str(
+      r#"{"key":"k","url":"u","expiresAt":"t","transfer":{"url":"v","partSize":8388608}}"#,
+    )
+    .unwrap();
+    assert_eq!(
+      new.transfer,
+      Some(SealedTransfer {
+        url: "v".to_string(),
+        part_size: 8_388_608
+      })
+    );
+  }
+}

src-tauri/src/sync/engine.rs

@@ -753,6 +753,7 @@ impl SyncEngine {
       .client
       .upload_bytes_with_metadata(
         &presign.url,
+        presign.transfer.as_ref(),
         &payload,
         Some(content_type),
         presign.metadata.as_ref(),
@@ -1324,7 +1325,12 @@ impl SyncEngine {
 
     self
       .client
-      .upload_bytes(&presign.url, &upload_data, Some(content_type))
+      .upload_bytes(
+        &presign.url,
+        presign.transfer.as_ref(),
+        &upload_data,
+        Some(content_type),
+      )
       .await?;
 
     Ok(())
@@ -1525,10 +1531,10 @@ impl SyncEngine {
     let batch_response = self.client.presign_upload_batch(items).await?;
 
     // Build URL map
-    let url_map: HashMap<String, String> = batch_response
+    let url_map: HashMap<String, (String, Option<SealedTransfer>)> = batch_response
       .items
       .into_iter()
-      .map(|item| (item.key, item.url))
+      .map(|item| (item.key, (item.url, item.transfer)))
       .collect();
 
     let total_bytes: u64 = files.iter().map(|f| f.size).sum();
@@ -1602,7 +1608,7 @@ impl SyncEngine {
         continue;
       }
 
-      let url = url.unwrap();
+      let (url, transfer) = url.unwrap();
       let client = client.clone();
       let tracker = tracker.clone();
       let resume_state = resume_state.clone();
@@ -1653,7 +1659,7 @@ impl SyncEngine {
         let mut last_err = String::new();
         for attempt in 0..MAX_FILE_RETRIES {
           match client
-            .upload_bytes(&url, &upload_data, content_type.as_deref())
+            .upload_bytes(&url, transfer.as_ref(), &upload_data, content_type.as_deref())
             .await
           {
             Ok(()) => {
@@ -2674,7 +2680,12 @@ impl SyncEngine {
         .await?;
       self
         .client
-        .upload_bytes(&file_presign.url, &file_payload, Some(file_content_type))
+        .upload_bytes(
+          &file_presign.url,
+          file_presign.transfer.as_ref(),
+          &file_payload,
+          Some(file_content_type),
+        )
         .await?;
     }
 
@@ -4961,7 +4972,7 @@ mod tests {
                      ?X-Amz-Signature=deadbeef";
 
     let error = client
-      .upload_bytes(presigned, b"payload", None)
+      .upload_bytes(presigned, None, b"payload", None)
       .await
       .expect_err("a host that cannot resolve must not report a successful upload");
 

src-tauri/src/sync/types.rs

@@ -41,6 +41,20 @@ pub struct PresignUploadResponse {
   /// from older servers → client sends no metadata headers (body-GET fallback).
   #[serde(default)]
   pub metadata: Option<HashMap<String, String>>,
+  /// Set by a server that encrypts at rest: upload through it instead of
+  /// `url`, in parts when the file is larger than one.
+  #[serde(default)]
+  pub transfer: Option<SealedTransfer>,
+}
+
+/// A server's own upload URL, for a store it encrypts at rest. A request
+/// through a proxy in front of it has a size and time limit, so a file larger
+/// than `part_size` goes up in parts (`SyncClient::upload_bytes`).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct SealedTransfer {
+  pub url: String,
+  #[serde(rename = "partSize")]
+  pub part_size: u64,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -126,6 +140,8 @@ pub struct PresignUploadBatchItemResponse {
   pub url: String,
   #[serde(rename = "expiresAt")]
   pub expires_at: String,
+  #[serde(default)]
+  pub transfer: Option<SealedTransfer>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/components/agent-tasks/batch-card.tsx

@@ -1,244 +0,0 @@
-"use client";
-
-import { motion } from "motion/react";
-import { useState } from "react";
-import { useTranslation } from "react-i18next";
-import { LuRotateCcw, LuSquare, LuX } from "react-icons/lu";
-import { ProgressBar } from "@/components/agent-console-parts";
-import { useNumber } from "@/components/agent-tasks/parts";
-import { StatusDot } from "@/components/cookie-bot-shared";
-import { Button } from "@/components/ui/button";
-import { useMotionAllowed } from "@/hooks/use-motion-allowed";
-import {
-  type AgentBatch,
-  type AgentTask,
-  batchProgress,
-  dismissAgentBatch,
-  promptTitle,
-  retryAgentBatch,
-  statusTone,
-  stopAgentBatch,
-} from "@/lib/agent-tasks";
-import { translateBackendError } from "@/lib/backend-errors";
-import { MOTION_SPRING } from "@/lib/motion";
-import { showErrorToast } from "@/lib/toast-utils";
-import { cn } from "@/lib/utils";
-
-/** The one line a tile shows: the newest step, or what the status means. */
-function statusLine(
-  t: ReturnType<typeof useTranslation>["t"],
-  task: AgentTask,
-): string {
-  if (task.status === "needs_approval") return t("agent.tasks.needsYou");
-  if (task.status === "done" && task.result) return task.result;
-  if (task.lastStep && task.status !== "queued") return task.lastStep.text;
-  return t(`agent.tasks.status.${task.status}`);
-}
-
-function ProfileTile({
-  task,
-  name,
-  selected,
-  onSelect,
-}: {
-  task: AgentTask;
-  name: string;
-  selected: boolean;
-  onSelect: () => void;
-}) {
-  const { t } = useTranslation();
-  const format = useNumber();
-  const live = task.status === "running" || task.status === "launching";
-  return (
-    <button
-      type="button"
-      data-slot="agent-batch-tile"
-      data-profile-id={task.profileId}
-      data-status={task.status}
-      aria-pressed={selected}
-      onClick={onSelect}
-      className={cn(
-        "flex min-w-0 cursor-pointer flex-col gap-1 rounded-xl px-3 py-2.5 text-left transition-colors duration-150 focus-visible:ring-2 focus-visible:ring-ring focus-visible:outline-none",
-        selected ? "bg-primary/10" : "bg-foreground/3 hover:bg-foreground/5",
-      )}
-    >
-      <span className="flex min-w-0 items-center gap-2">
-        <StatusDot tone={statusTone(task.status)} pulse={live} />
-        <span className="min-w-0 flex-1 truncate text-sm font-medium">
-          {name}
-        </span>
-        {task.creditsUsed > 0 && (
-          <span className="shrink-0 text-[11px] tabular-nums text-muted-foreground">
-            {format(task.creditsUsed)}
-          </span>
-        )}
-      </span>
-      <span
-        className={cn(
-          "truncate text-xs",
-          task.status === "needs_approval"
-            ? "text-warning-text"
-            : task.status === "failed"
-              ? "text-destructive-text"
-              : "text-muted-foreground",
-        )}
-      >
-        {statusLine(t, task)}
-      </span>
-    </button>
-  );
-}
-
-/** One run on one or many profiles: totals, the batch controls and a tile per profile. */
-export function BatchCard({
-  batch,
-  profileNames,
-  selectedProfileId,
-  onSelect,
-  onDismissed,
-}: {
-  batch: AgentBatch;
-  profileNames: Map<string, string>;
-  selectedProfileId: string | null;
-  onSelect: (profileId: string) => void;
-  onDismissed: (batchId: string) => void;
-}) {
-  const { t } = useTranslation();
-  const format = useNumber();
-  const animate = useMotionAllowed();
-  const [busy, setBusy] = useState(false);
-  const { tally } = batch;
-  const total = batch.tasks.length;
-  const ended = tally.done + tally.failed + tally.stopped;
-  const nameOf = (profileId: string) =>
-    profileNames.get(profileId) ?? t("agent.tasks.unknownProfile");
-
-  const act = async (work: () => Promise<unknown>) => {
-    setBusy(true);
-    try {
-      await work();
-    } catch (error) {
-      showErrorToast(translateBackendError(t, error));
-    } finally {
-      setBusy(false);
-    }
-  };
-
-  return (
-    <motion.article
-      layout={animate ? "position" : false}
-      initial={animate ? { opacity: 0.6, y: 6 } : false}
-      animate={{ opacity: 1, y: 0 }}
-      transition={animate ? MOTION_SPRING.gentle : { duration: 0 }}
-      data-slot="agent-batch"
-      data-batch-id={batch.id}
-      data-finished={batch.finished ? "true" : "false"}
-      className="flex flex-col gap-2.5"
-    >
-      <header className="flex items-start gap-3">
-        <div className="flex min-w-0 flex-1 flex-col gap-0.5">
-          <h4 className="truncate text-sm font-medium" title={batch.prompt}>
-            {promptTitle(batch.prompt)}
-          </h4>
-          <p
-            data-slot="agent-batch-summary"
-            className="flex flex-wrap gap-x-2 text-xs text-muted-foreground tabular-nums"
-          >
-            <span>{t("agent.tasks.progress", { done: ended, total })}</span>
-            {tally.failed > 0 && (
-              <span className="text-destructive-text">
-                {t("agent.tasks.failedCount", { n: tally.failed })}
-              </span>
-            )}
-            {tally.creditsUsed > 0 && (
-              <span>
-                {t("agent.tasks.credits", { n: format(tally.creditsUsed) })}
-              </span>
-            )}
-            {batch.routineId && <span>{t("agent.tasks.fromRoutine")}</span>}
-          </p>
-        </div>
-        <div className="flex shrink-0 items-center gap-1">
-          {!batch.finished && (
-            <Button
-              type="button"
-              variant="subtle"
-              size="sm"
-              disabled={busy || batch.stopping}
-              data-slot="agent-batch-stop"
-              className="h-7 gap-1.5 rounded-lg text-xs hover:bg-destructive/10 hover:text-destructive-text"
-              onClick={() => void act(() => stopAgentBatch(batch.id))}
-            >
-              <LuSquare className="size-3 fill-current" aria-hidden="true" />
-              {total > 1 ? t("agent.tasks.stopAll") : t("agent.tasks.stop")}
-            </Button>
-          )}
-          {tally.failed > 0 && batch.finished && (
-            <Button
-              type="button"
-              variant="subtle"
-              size="sm"
-              disabled={busy}
-              data-slot="agent-batch-retry"
-              className="h-7 gap-1.5 rounded-lg text-xs"
-              onClick={() => void act(() => retryAgentBatch(batch.id))}
-            >
-              <LuRotateCcw className="size-3.5" aria-hidden="true" />
-              {t("agent.tasks.retryFailed")}
-            </Button>
-          )}
-          {batch.finished && (
-            <Button
-              type="button"
-              variant="subtle"
-              size="sm"
-              disabled={busy}
-              aria-label={t("agent.tasks.dismiss")}
-              data-slot="agent-batch-dismiss"
-              className="h-7 rounded-lg px-2"
-              onClick={() =>
-                void act(async () => {
-                  await dismissAgentBatch(batch.id);
-                  onDismissed(batch.id);
-                })
-              }
-            >
-              <LuX className="size-3.5" />
-            </Button>
-          )}
-        </div>
-      </header>
-      {total > 1 && (
-        <>
-          <ProgressBar
-            percent={batchProgress(tally, total)}
-            label={t("agent.tasks.progress", { done: ended, total })}
-          />
-          <div className="grid gap-1.5 [grid-template-columns:repeat(auto-fill,minmax(12.5rem,1fr))]">
-            {batch.tasks.map((task) => (
-              <ProfileTile
-                key={task.profileId}
-                task={task}
-                name={nameOf(task.profileId)}
-                selected={task.profileId === selectedProfileId}
-                onSelect={() => {
-                  onSelect(task.profileId);
-                }}
-              />
-            ))}
-          </div>
-        </>
-      )}
-      {total === 1 && (
-        <ProfileTile
-          task={batch.tasks[0]}
-          name={nameOf(batch.tasks[0].profileId)}
-          selected={batch.tasks[0].profileId === selectedProfileId}
-          onSelect={() => {
-            onSelect(batch.tasks[0].profileId);
-          }}
-        />
-      )}
-    </motion.article>
-  );
-}