diff --git a/donut-sync/src/lib.rs b/donut-sync/src/lib.rs index 68a79584..e6ac3409 100644 --- a/donut-sync/src/lib.rs +++ b/donut-sync/src/lib.rs @@ -219,6 +219,9 @@ pub async fn build_state(config: &Config) -> Result, String> { if let Some(keyring) = keyring.filter(|k| !k.unsealing()) { crate::sealer::spawn(s3, keyring); } + if let Some(direct) = sync.direct_uploads() { + crate::sealer::spawn_flusher(sync.store().clone(), direct.clone()); + } Ok(Arc::new(AppState { guard, @@ -353,6 +356,14 @@ pub async fn run() -> i32 { }; let port = listener.local_addr().map_or(port, |a| a.port()); println!("Donut Sync service running on port {port}"); - serve(listener, state, stop_signal()).await; + serve(listener, state.clone(), stop_signal()).await; + if let Some(direct) = state.sync.direct_uploads() + && let Err(e) = direct.flush(state.sync.store().as_ref()).await + { + log::warn( + "Sealer", + &format!("Direct uploads not noted at exit err=\"{e}\""), + ); + } 0 } diff --git a/donut-sync/src/memory_store.rs b/donut-sync/src/memory_store.rs index 849953ec..25edb95a 100644 --- a/donut-sync/src/memory_store.rs +++ b/donut-sync/src/memory_store.rs @@ -12,7 +12,8 @@ use md5::{Digest, Md5}; use crate::store::{ ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListObject, ListOutput, - ObjectVersion, PendingUpload, Store, StoreError, Upload, collect, once, + ObjectVersion, PendingUpload, Store, StoreError, Upload, VersionMarker, VersionsPage, collect, + once, }; pub type HeadHook = Box Result + Send + Sync>; @@ -58,6 +59,10 @@ pub struct MemoryStore { pub list_hook: Option, pub copy_hook: Mutex>, pub versioned: bool, + /// Like B2: a PUT with If-Match is refused (501), never conditional. + pub ignores_if_match: bool, + /// Versions per listed page; 0 is 1000. + pub version_page: usize, pub uploads: Mutex>, } @@ -136,10 +141,10 @@ impl MemoryStore { } /// A device's delete: a delete marker where versions are kept. - pub fn remove(&self, key: &str) { - if self.raw(key).is_some() { - self.push(key, None); - } + pub fn remove(&self, key: &str) -> Option { + self.raw(key)?; + let id = self.push(key, None); + self.versioned.then_some(id) } pub fn get(&self, key: &str) -> Option { @@ -228,10 +233,9 @@ impl Store for MemoryStore { Ok(()) } - async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + async fn delete_object(&self, key: &str) -> Result, StoreError> { self.call(); - self.remove(key); - Ok(()) + Ok(self.remove(key)) } async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> { @@ -279,11 +283,20 @@ impl Store for MemoryStore { }) } - async fn list_versions(&self, prefix: &str) -> Result>, StoreError> { + async fn list_versions( + &self, + prefix: &str, + after: Option<&VersionMarker>, + ) -> Result, StoreError> { self.call(); if !self.versioned { return Ok(None); } + let page = if self.version_page == 0 { + 1000 + } else { + self.version_page + }; let objects = self.objects.lock().unwrap(); let mut out = Vec::new(); for (key, history) in objects.range(prefix.to_string()..) { @@ -301,7 +314,24 @@ impl Store for MemoryStore { }); } } - Ok(Some(out)) + let start = match after { + Some((key, id)) => out + .iter() + .position(|v| v.key == *key && v.version_id == *id) + .map_or(out.len(), |at| at + 1), + None => 0, + }; + let versions: Vec = out.into_iter().skip(start).collect(); + let next = (versions.len() > page).then(|| { + ( + versions[page - 1].key.clone(), + versions[page - 1].version_id.clone(), + ) + }); + Ok(Some(VersionsPage { + versions: versions.into_iter().take(page).collect(), + next, + })) } async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> { @@ -394,6 +424,13 @@ impl Store for MemoryStore { if body.len() as u64 != upload.length { return Err(StoreError::Transport("length mismatch".to_string())); } + if upload.if_match.is_some() && self.ignores_if_match { + return Err(StoreError::Service { + status: 501, + code: "NotImplemented".to_string(), + message: String::new(), + }); + } if let Some(expected) = upload.if_match && self.raw(key).map(|o| etag(&o.body)).as_deref() != Some(expected) { diff --git a/donut-sync/src/migrate.rs b/donut-sync/src/migrate.rs index 7234bd02..839bb4b4 100644 --- a/donut-sync/src/migrate.rs +++ b/donut-sync/src/migrate.rs @@ -34,7 +34,12 @@ use crate::config::Config; use crate::log; use crate::original_times::{ORIGINAL_MTIME, original_mtime}; use crate::sealed_store::sealed_by; -use crate::store::{ByteStream, CopySpec, HeadOutput, ListInput, Store, StoreError, Upload}; +use crate::sealer::{Failed, Failure, read_state, retry_failed, save_state}; +use crate::store::{ + ByteStream, CopySpec, HeadOutput, ListInput, Store, StoreError, Upload, versions_of, +}; +use crate::time; +use serde::{Deserialize, Serialize}; pub const COMMAND: &str = "migrate-at-rest"; const CONTEXT: &str = "Migrate"; @@ -43,6 +48,8 @@ pub const TMP_PREFIX: &str = ".donut-sync-at-rest/tmp/"; const PROGRESS_EVERY: Duration = Duration::from_secs(10); /// Tries per object: storage has transient failures. const ATTEMPTS: u32 = 3; +/// Put-backs in a row before a settle gives up on a history that keeps moving. +const SETTLE_ROUNDS: usize = 3; const USAGE: &str = "Usage: donut-sync migrate-at-rest [--unseal] [--dry-run] [--prefix P] [--concurrency N] [--max-mib-per-sec N] [--checkpoint FILE] @@ -172,6 +179,12 @@ pub(crate) enum Outcome { Rotated(u64), Unsealed(u64), Changed, + /// Rewritten, but a device's version may lie under the rewrite: settle it + /// later with these ids. + Unsettled { + settle: Settle, + reason: String, + }, Failed(String), } @@ -247,14 +260,14 @@ fn precondition_failed(error: &StoreError) -> bool { /// Deletes `key` with every version it has, where the bucket keeps versions, /// so a temporary object leaves nothing behind. pub async fn purge(store: &dyn Store, key: &str) -> Result<(), StoreError> { - match store.list_versions(key).await? { + match versions_of(store, key).await? { Some(versions) => { - for version in versions.iter().filter(|v| v.key == key) { + for version in versions { store.delete_version(key, &version.version_id).await?; } Ok(()) } - None => store.delete_object(key).await, + None => store.delete_object(key).await.map(|_| ()), } } @@ -415,66 +428,126 @@ impl Migration { let Some(ours) = placed? else { return Ok(Outcome::Changed); }; - match self - .settle(key, head.version_id.as_deref(), ours.as_deref()) - .await? - { - true => Ok(Outcome::Changed), - false => Ok(done), + // From here the copy is in place: an error must not send the object + // back to the HEAD above, which would take our copy for current and + // leave a device's version under it. + let settle = Settle { + read: head.version_id.clone(), + ours, + }; + Ok(match self.settle(key, &settle).await { + Ok(true) => Outcome::Changed, + Ok(false) => done, + Err(reason) => Outcome::Unsettled { settle, reason }, + }) + } + + /// `settle_once`, tried again on a transient failure. + pub(crate) async fn settle(&self, key: &str, settle: &Settle) -> Result { + let mut attempt = 0; + loop { + attempt += 1; + match self.settle_once(key, settle).await { + Ok(restored) => return Ok(restored), + Err(SettleError::Store(e)) if attempt < ATTEMPTS => { + log::warn(CONTEXT, &format!("Settle retry key={key} err=\"{e}\"")); + tokio::time::sleep(Duration::from_millis(500 << attempt)).await; + } + Err(e) => return Err(e.to_string()), + } } } /// After the copy: a version made between the read and the copy (a - /// device's write or delete) is put back as the current one. `true` when - /// one was. Needs a bucket that keeps versions. - async fn settle( - &self, - key: &str, - read: Option<&str>, - ours: Option<&str>, - ) -> Result { - let (Some(read), Some(ours)) = (read, ours) else { - return Ok(false); - }; - let Some(versions) = self.store.list_versions(key).await? else { - return Ok(false); - }; - let history: Vec<_> = versions.iter().filter(|v| v.key == key).collect(); - let (Some(at_ours), Some(at_read)) = ( - history.iter().position(|v| v.version_id == ours), - history.iter().position(|v| v.version_id == read), - ) else { - return Ok(false); - }; - // Newest first: anything before ours came after it and stays current. - if at_ours != 0 || at_read <= at_ours + 1 { - return Ok(false); + /// device's write or delete) is put back as the current one, and again for + /// a version made while that was put back, until the history holds still. + /// `true` when one was. A bucket without versions has nothing to settle. + async fn settle_once(&self, key: &str, settle: &Settle) -> Result { + let (mut read, mut ours) = (settle.read.clone(), settle.ours.clone()); + let mut restored = false; + for _ in 0..SETTLE_ROUNDS { + let Some(history) = versions_of(self.store.as_ref(), key).await? else { + return Ok(restored); + }; + let (Some(read_id), Some(ours_id)) = (read.as_deref(), ours.as_deref()) else { + return Err(SettleError::Unknown( + "the bucket keeps versions but gave no version id", + )); + }; + let Some(at_ours) = history.iter().position(|v| v.version_id == ours_id) else { + return Err(SettleError::Unknown("our version is not listed")); + }; + let at_read = history + .iter() + .position(|v| v.version_id == read_id) + .unwrap_or(history.len()); + // Newest first: anything before ours came after it and stays current. + if at_ours != 0 || at_read <= at_ours + 1 { + return Ok(restored); + } + let lost = &history[at_ours + 1]; + let put_back = if lost.delete_marker { + let marker = self.store.delete_object(key).await?; + log::warn( + CONTEXT, + &format!("Delete made during the rewrite put back key={key}"), + ); + marker + } else { + let copied = self + .store + .copy_object_with( + key, + key, + CopySpec { + source_version: Some(&lost.version_id), + ..CopySpec::default() + }, + ) + .await?; + log::warn( + CONTEXT, + &format!("Write made during the rewrite put back key={key}"), + ); + copied.version_id + }; + restored = true; + // Next round: a version between this put-back and ours came while it + // was being put back. + read = ours; + ours = put_back; } - let lost = history[at_ours + 1]; - if lost.delete_marker { - self.store.delete_object(key).await?; - log::warn( - CONTEXT, - &format!("Delete made during the rewrite put back key={key}"), - ); - } else { - self - .store - .copy_object_with( - key, - key, - CopySpec { - source_version: Some(&lost.version_id), - ..CopySpec::default() - }, - ) - .await?; - log::warn( - CONTEXT, - &format!("Write made during the rewrite put back key={key}"), - ); + Err(SettleError::Unknown("the versions kept changing")) + } +} + +/// The versions a rewrite read and wrote, to settle it later if it could +/// not be settled at once. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Settle { + pub read: Option, + pub ours: Option, +} + +enum SettleError { + /// Storage failed; trying again may help. + Store(StoreError), + /// Needs a person or a later pass. + Unknown(&'static str), +} + +impl From for SettleError { + fn from(e: StoreError) -> Self { + SettleError::Store(e) + } +} + +impl std::fmt::Display for SettleError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SettleError::Store(e) => write!(f, "{e}"), + SettleError::Unknown(why) => write!(f, "{why}"), } - Ok(true) } } @@ -581,6 +654,7 @@ pub async fn migrate( start_after.as_deref().unwrap_or("-"), ), ); + let state_store = crate::sealed_store::SealedStore::new(store.clone(), keyring.clone()); let migration = Migration { store: store.clone(), keyring, @@ -595,6 +669,22 @@ pub async fn migrate( }), }; let mut report = Report::default(); + // Rewrites an earlier run or the sealer left unsettled are settled first. + let (retried, mut unsettled) = if options.dry_run { + (Vec::new(), Vec::new()) + } else { + let state = read_state(&state_store) + .await + .map_err(|e| format!("Sealer state unreadable: {e}"))?; + let open: Vec = state + .failed + .into_iter() + .filter(|f| f.settle.is_some()) + .collect(); + let retried: Vec = open.iter().map(|f| f.key.clone()).collect(); + (retried, retry_failed(&migration, &open).await) + }; + report.failed += unsettled.len() as u64; let mut token: Option = None; let mut last_progress = Instant::now(); let mut checkpoint_held = false; @@ -642,6 +732,19 @@ pub async fn migrate( report.unsealed_bytes += bytes; } Outcome::Changed => report.changed += 1, + Outcome::Unsettled { settle, reason } => { + report.failed += 1; + checkpoint_held = true; + log::warn( + CONTEXT, + &format!("Rewrite not settled key={key} err=\"{reason}\""), + ); + unsettled.push(Failure { + key, + reason, + settle: Some(settle), + }); + } Outcome::Failed(err) => { report.failed += 1; checkpoint_held = true; @@ -676,6 +779,15 @@ pub async fn migrate( ); } } + if !retried.is_empty() || !unsettled.is_empty() { + let mut state = read_state(&state_store) + .await + .map_err(|e| format!("Sealer state unreadable: {e}"))?; + state.note(&retried, unsettled, time::now_ms()); + save_state(&state_store, &state) + .await + .map_err(|e| format!("Sealer state not saved: {e}"))?; + } // A whole bucket unsealed: this server's own objects (sealed records and // state) go too, so nothing sealed is left when the key is removed. if options.unseal && !options.dry_run && options.prefix.is_none() && report.failed == 0 { @@ -1004,7 +1116,9 @@ mod tests { let raw = Arc::new(MemoryStore::versioned()); prepare(&raw).await; - race(&raw, |store| store.remove("k")); + race(&raw, |store| { + store.remove("k"); + }); assert!(matches!( rewrite(&raw, keys.clone(), direction).await, Outcome::Changed @@ -1017,22 +1131,106 @@ mod tests { #[tokio::test] async fn a_write_after_the_copy_stays_current() { let raw = Arc::new(MemoryStore::versioned()); - raw.insert("k", b"old".to_vec(), Vec::new(), None); + let read = raw.insert("k", b"old".to_vec(), Vec::new(), None); let migration = Migration::sealing(raw.clone(), keyring(K1), false); assert!(matches!( migration.object("k", None).await, Outcome::Sealed(3) )); + let ours = raw.head_object("k").await.unwrap().version_id; raw.insert("k", b"later".to_vec(), Vec::new(), None); - assert!( - !migration - .settle("k", Some("v1000"), Some("v0")) - .await - .unwrap() - ); + let settle = Settle { + read: Some(read), + ours, + }; + assert!(!migration.settle("k", &settle).await.unwrap()); assert_eq!(raw.raw("k").unwrap().body, b"later"); } + /// A store that keeps versions but names none, as no S3 store should. + #[tokio::test] + async fn missing_version_ids_are_recorded_not_ignored() { + let raw = Arc::new(MemoryStore::versioned()); + raw.insert("k", b"old".to_vec(), Vec::new(), None); + let migration = Migration::sealing(raw.clone(), keyring(K1), false); + let settle = Settle { + read: None, + ours: None, + }; + let reason = migration.settle("k", &settle).await.unwrap_err(); + assert!(reason.contains("no version id"), "{reason}"); + } + + /// Settle keeps putting back until the history holds still: here a third + /// write lands while the first put-back is made. + #[tokio::test] + async fn a_write_during_the_put_back_is_put_back_too() { + let raw = Arc::new(MemoryStore::versioned()); + raw.insert("k", b"old".to_vec(), Vec::new(), None); + // The device writes inside the rewrite's copy, and again inside the + // put-back's copy. + let copies = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = copies.clone(); + *raw.copy_hook.lock().unwrap() = Some(Box::new(move |store, to| { + if to != "k" { + return; + } + match seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) { + 0 => { + store.insert("k", b"first".to_vec(), Vec::new(), None); + } + 1 => { + store.insert("k", b"second".to_vec(), Vec::new(), None); + } + _ => {} + } + })); + let migration = Migration::sealing(raw.clone(), keyring(K1), false); + assert!(matches!( + migration.object("k", None).await, + Outcome::Changed + )); + assert_eq!(raw.raw("k").unwrap().body, b"second"); + assert_eq!(copies.load(std::sync::atomic::Ordering::SeqCst), 3); + } + + /// The settle fails after the copy: the object is not rewritten again from + /// a HEAD, and the ids are kept for a later pass, which settles it. + #[tokio::test] + async fn an_unsettled_rewrite_is_recorded_and_settled_later() { + let raw = Arc::new(MemoryStore { + ignores_if_match: true, + ..MemoryStore::versioned() + }); + raw.insert("k", b"old".to_vec(), Vec::new(), None); + let fail = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let failing = fail.clone(); + *raw.copy_hook.lock().unwrap() = Some(Box::new(move |store, to| { + if to == "k" && failing.load(std::sync::atomic::Ordering::SeqCst) { + store.insert("k", b"device".to_vec(), Vec::new(), None); + } + })); + // Every put-back copy also races a device write, so the history never + // holds still within SETTLE_ROUNDS. + let keys = keyring(K1); + let report = migrate(raw.clone(), keys.clone(), &options()) + .await + .unwrap(); + assert_eq!(report.failed, 1); + let state_store = crate::sealed_store::SealedStore::new(raw.clone(), keys.clone()); + let state = read_state(&state_store).await.unwrap(); + assert_eq!(state.failed.len(), 1); + assert!(state.failed[0].settle.is_some()); + // A later run settles it without a new rewrite. + fail.store(false, std::sync::atomic::Ordering::SeqCst); + let report = migrate(raw.clone(), keys, &options()).await.unwrap(); + assert_eq!(report.failed, 0, "{report:?}"); + assert!(read_state(&state_store).await.unwrap().failed.is_empty()); + // The device's write stands, and this run sealed it. + let served = crate::served_store(raw.clone(), Some(&keyring(K1))); + assert_eq!(served.get_object_text("k").await.unwrap(), "device"); + } + #[test] fn arguments_parse() { let args: Vec = ["--dry-run", "--prefix", "users/", "--max-mib-per-sec", "8"] diff --git a/donut-sync/src/original_times.rs b/donut-sync/src/original_times.rs index 67983467..54828217 100644 --- a/donut-sync/src/original_times.rs +++ b/donut-sync/src/original_times.rs @@ -9,8 +9,8 @@ use std::sync::Arc; use async_trait::async_trait; use crate::store::{ - ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, ObjectVersion, - PendingUpload, Store, StoreError, Upload, + ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store, + StoreError, Upload, VersionMarker, VersionsPage, }; /// User metadata (`x-amz-meta-donut-orig-mtime`): the object's LastModified @@ -77,7 +77,7 @@ impl Store for OriginalTimes { .await } - async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + async fn delete_object(&self, key: &str) -> Result, StoreError> { self.inner.delete_object(key).await } @@ -137,8 +137,12 @@ impl Store for OriginalTimes { self.inner.copy_object_with(from, to, spec).await } - async fn list_versions(&self, prefix: &str) -> Result>, StoreError> { - self.inner.list_versions(prefix).await + async fn list_versions( + &self, + prefix: &str, + after: Option<&VersionMarker>, + ) -> Result, StoreError> { + self.inner.list_versions(prefix, after).await } async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> { diff --git a/donut-sync/src/s3.rs b/donut-sync/src/s3.rs index dc0541cb..5596065c 100644 --- a/donut-sync/src/s3.rs +++ b/donut-sync/src/s3.rs @@ -12,7 +12,7 @@ use crate::jsval::js_number; use crate::sigv4::{self, Credentials}; use crate::store::{ ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListObject, ListOutput, - ObjectVersion, PendingUpload, Store, StoreError, Upload, + ObjectVersion, PendingUpload, Store, StoreError, Upload, VersionMarker, VersionsPage, }; use crate::time; @@ -557,11 +557,11 @@ fn head_of(headers: &reqwest::header::HeaderMap) -> HeadOutput { } } -type VersionsPage = (Vec, Option<(String, String)>); +type ParsedVersions = (Vec, Option); /// A ListObjectVersions reply. `None` when the store answered with something /// else, as stores without versions do. -fn parse_versions(body: &[u8]) -> Result, StoreError> { +fn parse_versions(body: &[u8]) -> Result, StoreError> { let text = std::str::from_utf8(body) .map_err(|e| StoreError::Transport(format!("ListObjectVersions reply is not UTF-8: {e}")))?; let text = hide_control_references(text); @@ -735,12 +735,17 @@ impl Store for S3Store { .map(|_| ()) } - async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + async fn delete_object(&self, key: &str) -> Result, StoreError> { let path = self.location.object_path(key); - self + let reply = self .expect_success(Method::DELETE, &path, &[], &[], &[]) - .await - .map(|_| ()) + .await?; + Ok( + reply + .headers + .get("x-amz-version-id") + .map(|v| latin1(v.as_bytes())), + ) } async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> { @@ -781,39 +786,37 @@ impl Store for S3Store { self.copy(from, to, spec).await } - async fn list_versions(&self, prefix: &str) -> Result>, StoreError> { + async fn list_versions( + &self, + prefix: &str, + after: Option<&VersionMarker>, + ) -> Result, StoreError> { let path = self.location.bucket_path(); - let mut out = Vec::new(); - let mut marker: Option<(String, String)> = None; - loop { - let mut query = vec![ - ("versions".to_string(), String::new()), - ("prefix".to_string(), prefix.to_string()), - ]; - if let Some((key, id)) = &marker { - query.push(("key-marker".to_string(), key.clone())); - query.push(("version-id-marker".to_string(), id.clone())); - } - let reply = match self - .expect_success(Method::GET, &path, &query, &[], &[]) - .await - { - Ok(reply) => reply, - Err(StoreError::NotFound) => return Ok(None), - Err(StoreError::Service { status, .. }) if status == 501 || status == 400 => { - return Ok(None); - } - Err(e) => return Err(e), - }; - let Some((versions, next)) = parse_versions(&reply.body)? else { - return Ok(None); - }; - out.extend(versions); - match next { - Some(next) if Some(&next) != marker.as_ref() => marker = Some(next), - _ => return Ok(Some(out)), - } + let mut query = vec![ + ("versions".to_string(), String::new()), + ("prefix".to_string(), prefix.to_string()), + ]; + if let Some((key, id)) = after { + query.push(("key-marker".to_string(), key.clone())); + query.push(("version-id-marker".to_string(), id.clone())); } + let reply = match self + .expect_success(Method::GET, &path, &query, &[], &[]) + .await + { + Ok(reply) => reply, + Err(StoreError::NotFound) => return Ok(None), + Err(StoreError::Service { status, .. }) if status == 501 || status == 400 => { + return Ok(None); + } + Err(e) => return Err(e), + }; + Ok( + parse_versions(&reply.body)?.map(|(versions, next)| VersionsPage { + versions, + next: next.filter(|next| Some(next) != after), + }), + ) } async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> { diff --git a/donut-sync/src/sealed_store.rs b/donut-sync/src/sealed_store.rs index ca71e150..bf87aeae 100644 --- a/donut-sync/src/sealed_store.rs +++ b/donut-sync/src/sealed_store.rs @@ -12,8 +12,8 @@ use async_trait::async_trait; use crate::at_rest::{self, Keyring, MARKER}; use crate::store::{ - ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, ObjectVersion, - PendingUpload, Store, StoreError, Upload, + ByteStream, Copied, CopySpec, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store, + StoreError, Upload, VersionMarker, VersionsPage, }; pub struct SealedStore { @@ -96,7 +96,7 @@ impl Store for SealedStore { .await } - async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + async fn delete_object(&self, key: &str) -> Result, StoreError> { self.inner.delete_object(key).await } @@ -146,8 +146,12 @@ impl Store for SealedStore { .await } - async fn list_versions(&self, prefix: &str) -> Result>, StoreError> { - self.inner.list_versions(prefix).await + async fn list_versions( + &self, + prefix: &str, + after: Option<&VersionMarker>, + ) -> Result, StoreError> { + self.inner.list_versions(prefix, after).await } async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError> { diff --git a/donut-sync/src/sealer.rs b/donut-sync/src/sealer.rs index 31e13122..7fd99221 100644 --- a/donut-sync/src/sealer.rs +++ b/donut-sync/src/sealer.rs @@ -18,7 +18,7 @@ use serde_json::{Value, json}; use crate::at_rest::Keyring; use crate::log; -use crate::migrate::{Migration, Outcome, TMP_PREFIX, honors_if_match, purge}; +use crate::migrate::{Migration, Outcome, Settle, TMP_PREFIX, honors_if_match, purge}; use crate::sealed_store::SealedStore; use crate::store::{ListInput, Store, StoreError}; use crate::time; @@ -45,6 +45,61 @@ const ABANDONED_MS: i64 = 4 * 3600 * 1000; const MAX_FAILED: usize = 1000; const CONCURRENCY: usize = 4; +/// How often the keys handed direct URLs are written as one record. A crash +/// loses at most this window; the hourly backstop pass covers it. +const FLUSH_EVERY: Duration = Duration::from_secs(10); + +/// Keys handed direct storage URLs, gathered in memory until the next flush, +/// so a presign costs no storage write. +#[derive(Default)] +pub struct DirectUploads { + pending: std::sync::Mutex<(Vec, i64)>, +} + +impl DirectUploads { + pub fn note(&self, keys: &[String], expires_ms: i64) { + let mut pending = self.pending.lock().unwrap(); + pending.0.extend_from_slice(keys); + pending.1 = pending.1.max(expires_ms); + } + + /// Writes what was noted as one record; `store` is the sealed store. + pub async fn flush(&self, store: &dyn Store) -> Result<(), StoreError> { + let (keys, expires_ms) = std::mem::take(&mut *self.pending.lock().unwrap()); + if keys.is_empty() { + return Ok(()); + } + let outcome = note_direct_uploads(store, &keys, expires_ms).await; + if outcome.is_err() { + self.note(&keys, expires_ms); + } + outcome + } +} + +/// Flushes `uploads` every `FLUSH_EVERY` for as long as the server runs. +pub fn spawn_flusher(store: Arc, uploads: Arc) { + tokio::spawn(async move { + let mut failing = false; + loop { + tokio::time::sleep(FLUSH_EVERY).await; + match uploads.flush(store.as_ref()).await { + Ok(()) if failing => { + failing = false; + log::log(CONTEXT, "Direct uploads noted again"); + } + Ok(()) => {} + Err(e) if !failing => { + failing = true; + // The keys stay in memory for the next flush. + log::warn(CONTEXT, &format!("Direct uploads not noted err=\"{e}\"")); + } + Err(_) => {} + } + } + }); +} + /// Notes `keys`, which devices may write in plaintext through direct storage /// URLs until `expires_ms`. `store` is the sealed store: the record is sealed. pub async fn note_direct_uploads( @@ -82,6 +137,17 @@ pub struct Failed { pub reason: String, /// When it first failed. pub at: i64, + /// Set when the rewrite is in place but a device's version may lie under + /// it: a later pass settles it with these ids, and does not rewrite again. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub settle: Option, +} + +/// A key that failed in a pass. +pub struct Failure { + pub key: String, + pub reason: String, + pub settle: Option, } /// What the sealer keeps between passes, sealed in `STATE`. @@ -94,18 +160,27 @@ pub struct State { } impl State { - /// `failures` from a pass over `checked`: keys that sealed leave the list, - /// keys that failed join it (keeping when they first failed). - fn note(&mut self, checked: &[String], failures: Vec<(String, String)>, now_ms: i64) { + /// `failures` from a pass that retried `retried`: retried keys that + /// worked leave the list, keys that failed join it (keeping when they + /// first failed). + pub fn note(&mut self, retried: &[String], failures: Vec, now_ms: i64) { let first: Vec<(String, i64)> = self.failed.iter().map(|f| (f.key.clone(), f.at)).collect(); - self.failed.retain(|f| !checked.contains(&f.key)); - for (key, reason) in failures { + self.failed.retain(|f| !retried.contains(&f.key)); + for failure in failures { let at = first .iter() - .find(|(k, _)| *k == key) + .find(|(k, _)| *k == failure.key) .map_or(now_ms, |(_, at)| *at); - self.failed.retain(|f| f.key != key); - self.failed.insert(0, Failed { key, reason, at }); + self.failed.retain(|f| f.key != failure.key); + self.failed.insert( + 0, + Failed { + key: failure.key, + reason: failure.reason, + at, + settle: failure.settle, + }, + ); } self.failed.truncate(MAX_FAILED); } @@ -120,6 +195,61 @@ pub async fn read_state(store: &dyn Store) -> Result { } } +pub async fn save_state(store: &dyn Store, state: &State) -> Result<(), StoreError> { + store + .put_object( + STATE, + serde_json::to_vec(state).expect("the state serializes"), + "application/json", + &[], + ) + .await +} + +/// Tries the failed keys again: a settle where the rewrite is in place, a +/// whole rewrite otherwise. Returns the ones that still fail. +pub(crate) async fn retry_failed(migration: &Migration, failed: &[Failed]) -> Vec { + let mut still = Vec::new(); + for failed in failed { + match &failed.settle { + Some(settle) => { + if let Err(reason) = migration.settle(&failed.key, settle).await { + still.push(Failure { + key: failed.key.clone(), + reason, + settle: Some(settle.clone()), + }); + } + } + None => { + if let Some(failure) = failure_of( + failed.key.clone(), + migration.object(&failed.key, None).await, + ) { + still.push(failure); + } + } + } + } + still +} + +fn failure_of(key: String, outcome: Outcome) -> Option { + match outcome { + Outcome::Failed(reason) => Some(Failure { + key, + reason, + settle: None, + }), + Outcome::Unsettled { settle, reason } => Some(Failure { + key, + reason, + settle: Some(settle), + }), + _ => None, + } +} + #[derive(Default, Debug, PartialEq)] pub struct PassReport { pub records: u64, @@ -145,12 +275,12 @@ impl Sealer { }) } - /// Seals `keys`; returns the ones that failed, with why. + /// Seals `keys`; returns the ones that failed. async fn seal_all( &self, keys: Vec<(String, Option)>, report: &mut PassReport, - ) -> Vec<(String, String)> { + ) -> Vec { let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys) .map(|(key, listed)| async move { let outcome = self.migration.object(&key, listed).await; @@ -162,149 +292,155 @@ impl Sealer { let mut failures = Vec::new(); for (key, outcome) in outcomes { report.checked += 1; - match outcome { - Outcome::Sealed(_) | Outcome::Rotated(_) => report.sealed += 1, - Outcome::Failed(err) => { - log::warn( - CONTEXT, - &format!("Object not sealed key={key} err=\"{err}\""), - ); - report.failed += 1; - failures.push((key, err)); - } - Outcome::Current | Outcome::Changed | Outcome::Unsealed(_) => {} + if matches!(outcome, Outcome::Sealed(_) | Outcome::Rotated(_)) { + report.sealed += 1; + } + if let Some(failure) = failure_of(key, outcome) { + log::warn( + CONTEXT, + &format!( + "Object not sealed key={} err=\"{}\"", + failure.key, failure.reason + ), + ); + report.failed += 1; + failures.push(failure); } } failures } - async fn list(&self, prefix: Option<&str>) -> Result)>, StoreError> { - let mut out = Vec::new(); - let mut token = None; - loop { - let page = self - .raw - .list_objects_v2(&ListInput { - prefix: prefix.map(str::to_string), - max_keys: Some("1000".to_string()), - continuation_token: token.clone(), - ..ListInput::default() - }) - .await?; - out.extend( - page - .contents - .into_iter() - .map(|object| (object.key, object.last_modified_ms)), - ); - token = page.next_continuation_token.filter(|_| page.is_truncated); - if token.is_none() { - return Ok(out); - } - } - } - - async fn save_state(&self, state: &State) -> Result<(), StoreError> { - self - .sealed - .put_object( - STATE, - serde_json::to_vec(state).expect("the state serializes"), - "application/json", - &[], - ) - .await + /// One page of the listing under `prefix`. + async fn page( + &self, + prefix: Option<&str>, + token: Option, + ) -> Result<(Vec<(String, Option)>, Option), StoreError> { + let page = self + .raw + .list_objects_v2(&ListInput { + prefix: prefix.map(str::to_string), + max_keys: Some("1000".to_string()), + continuation_token: token, + ..ListInput::default() + }) + .await?; + let next = page.next_continuation_token.filter(|_| page.is_truncated); + Ok(( + page + .contents + .into_iter() + .map(|object| (object.key, object.last_modified_ms)) + .collect(), + next, + )) } /// Seals the keys of each record whose URLs are dead, then drops the /// record. Keys that fail are kept in the state for the backstop. pub async fn pending_pass(&self, now_ms: i64) -> Result { let mut report = PassReport::default(); - let mut checked = Vec::new(); let mut failures = Vec::new(); - for (record, _) in self.list(Some(PENDING)).await? { - report.records += 1; - if !record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS) { - continue; - } - let keys: Vec = match self.sealed.get_object_text(&record).await { - Ok(text) => serde_json::from_str::(&text) - .ok() - .and_then(|v| v["keys"].as_array().cloned()) - .unwrap_or_default() - .iter() - .filter_map(|k| k.as_str().map(str::to_string)) - .collect(), - Err(StoreError::NotFound) => continue, - Err(e) => { - log::warn( - CONTEXT, - &format!("Record unreadable key={record} err=\"{e}\""), - ); - report.failed += 1; + let mut token = None; + loop { + let (records, next) = self.page(Some(PENDING), token).await?; + for (record, _) in records { + report.records += 1; + if !record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS) { continue; } - }; - failures.extend( - self - .seal_all( - keys.iter().map(|k| (k.clone(), None)).collect(), - &mut report, - ) - .await, - ); - checked.extend(keys); - purge(self.raw.as_ref(), &record).await?; - report.records_done += 1; + let keys: Vec = match self.sealed.get_object_text(&record).await { + Ok(text) => serde_json::from_str::(&text) + .ok() + .and_then(|v| v["keys"].as_array().cloned()) + .unwrap_or_default() + .iter() + .filter_map(|k| k.as_str().map(str::to_string)) + .collect(), + Err(StoreError::NotFound) => continue, + Err(e) => { + log::warn( + CONTEXT, + &format!("Record unreadable key={record} err=\"{e}\""), + ); + report.failed += 1; + continue; + } + }; + failures.extend( + self + .seal_all(keys.into_iter().map(|k| (k, None)).collect(), &mut report) + .await, + ); + purge(self.raw.as_ref(), &record).await?; + report.records_done += 1; + } + token = next; + if token.is_none() { + break; + } } if !failures.is_empty() { let mut state = read_state(&self.sealed).await?; - state.note(&checked, failures, now_ms); - self.save_state(&state).await?; + state.note(&[], failures, now_ms); + save_state(&self.sealed, &state).await?; } Ok(report) } - /// Seals everything written since the last pass and retries what failed, - /// then cleans up after interrupted rewrites and uploads. + /// Retries what failed, seals everything written since the last pass, + /// page by page, then cleans up after interrupted rewrites and uploads. pub async fn backstop(&self, now_ms: i64) -> Result { let mut state = read_state(&self.sealed).await?; let since = state.backstop_since.unwrap_or(now_ms - FIRST_BACKSTOP_MS); - let listed = self.list(None).await?; - let mut candidates: Vec<(String, Option)> = listed - .iter() - .filter(|(key, modified)| { - !key.starts_with(INTERNAL_ROOT) - && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS) - }) - .cloned() - .collect(); - for failed in &state.failed { - if !candidates.iter().any(|(key, _)| *key == failed.key) { - candidates.push((failed.key.clone(), None)); + let mut report = PassReport::default(); + let retried: Vec = state.failed.iter().map(|f| f.key.clone()).collect(); + let mut failures = retry_failed(&self.migration, &state.failed).await; + report.checked += retried.len() as u64; + report.failed += failures.len() as u64; + let mut token = None; + loop { + let (objects, next) = self.page(None, token).await?; + let recent: Vec<(String, Option)> = objects + .into_iter() + .filter(|(key, modified)| { + !key.starts_with(INTERNAL_ROOT) + && !retried.contains(key) + && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS) + }) + .collect(); + failures.extend(self.seal_all(recent, &mut report).await); + token = next; + if token.is_none() { + break; } } - let checked: Vec = candidates.iter().map(|(key, _)| key.clone()).collect(); - let mut report = PassReport::default(); - let failures = self.seal_all(candidates, &mut report).await; // The pass moves on even when some keys fail: they stay in the list. - state.note(&checked, failures, now_ms); + state.note(&retried, failures, now_ms); state.backstop_since = Some(now_ms); - self.save_state(&state).await?; + save_state(&self.sealed, &state).await?; - for (key, modified) in &listed { - if key.starts_with(TMP_PREFIX) && modified.is_some_and(|ms| now_ms - ms > TMP_STALE_MS) { - match purge(self.raw.as_ref(), key).await { - Ok(()) => log::log( - CONTEXT, - &format!("Stale temporary object removed key={key}"), - ), - Err(e) => log::warn( - CONTEXT, - &format!("Stale temporary object not removed key={key} err=\"{e}\""), - ), + let mut token = None; + loop { + let (temporary, next) = self.page(Some(TMP_PREFIX), token).await?; + for (key, modified) in temporary { + if modified.is_some_and(|ms| now_ms - ms > TMP_STALE_MS) { + match purge(self.raw.as_ref(), &key).await { + Ok(()) => log::log( + CONTEXT, + &format!("Stale temporary object removed key={key}"), + ), + Err(e) => log::warn( + CONTEXT, + &format!("Stale temporary object not removed key={key} err=\"{e}\""), + ), + } } } + token = next; + if token.is_none() { + break; + } } match self.raw.list_multipart("").await { Ok(uploads) => { @@ -548,6 +684,28 @@ mod tests { assert!(raw.list_multipart("").await.unwrap().is_empty()); } + #[tokio::test] + async fn presigns_are_noted_as_one_record_per_flush() { + let (raw, sealed, _) = setup(MemoryStore::default()).await; + let direct = DirectUploads::default(); + direct.flush(&sealed).await.unwrap(); + assert!(raw.keys().is_empty(), "nothing noted, nothing written"); + direct.note(&["a".to_string()], 5_000); + direct.note(&["b".to_string(), "c".to_string()], 7_000); + direct.flush(&sealed).await.unwrap(); + let records: Vec = raw + .keys() + .into_iter() + .filter(|k| k.starts_with(PENDING)) + .collect(); + assert_eq!(records.len(), 1); + assert_eq!(record_expiry(&records[0]), Some(7_000)); + let body = sealed.get_object_text(&records[0]).await.unwrap(); + assert_eq!(body, r#"{"keys":["a","b","c"]}"#); + direct.flush(&sealed).await.unwrap(); + assert_eq!(raw.keys().len(), 1); + } + #[test] fn records_name_their_expiry() { assert_eq!( diff --git a/donut-sync/src/slots.rs b/donut-sync/src/slots.rs index 5f3dd4e1..e5555622 100644 --- a/donut-sync/src/slots.rs +++ b/donut-sync/src/slots.rs @@ -11,9 +11,9 @@ use std::time::Duration; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; -const MAX_TRANSFERS: usize = 64; -const MAX_PER_NAMESPACE: usize = 8; -pub const SLOT_WAIT: Duration = Duration::from_secs(20); +const MAX_TRANSFERS: usize = 192; +const MAX_PER_NAMESPACE: usize = 32; +pub const SLOT_WAIT: Duration = Duration::from_secs(60); const MAX_HOLD: Duration = Duration::from_secs(600); /// Accounts whose semaphores are kept while idle. const NAMESPACES_KEPT: usize = 10_000; @@ -110,7 +110,7 @@ mod tests { #[tokio::test] async fn one_account_cannot_take_every_slot() { - let slots = Slots::new(10, MAX_HOLD); + let slots = Slots::new(MAX_PER_NAMESPACE + 2, MAX_HOLD); let mut held = Vec::new(); for _ in 0..MAX_PER_NAMESPACE { held.push(slots.acquire("users/a/x", WAIT).await.unwrap()); @@ -120,7 +120,7 @@ mod tests { held.push(slots.acquire("users/c/x", WAIT).await.unwrap()); assert!( slots.acquire("users/d/x", WAIT).await.is_none(), - "all 10 in use" + "every slot in use" ); held.pop(); assert!(slots.acquire("users/d/x", WAIT).await.is_some()); diff --git a/donut-sync/src/store.rs b/donut-sync/src/store.rs index b3b25e1d..738bdde4 100644 --- a/donut-sync/src/store.rs +++ b/donut-sync/src/store.rs @@ -115,7 +115,8 @@ pub trait Store: Send + Sync { content_type: &str, metadata: &[(String, String)], ) -> Result<(), StoreError>; - async fn delete_object(&self, key: &str) -> Result<(), StoreError>; + /// Returns the delete marker's version id, where the bucket keeps versions. + async fn delete_object(&self, key: &str) -> Result, StoreError>; /// Deletes up to 1000 keys in one request; per-key failures are ignored. async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError>; /// Server-side copy, metadata included. @@ -131,9 +132,14 @@ pub trait Store: Send + Sync { to: &str, spec: CopySpec<'_>, ) -> Result; - /// Every version and delete marker under `prefix`, newest first within a - /// key. `None` when the store keeps no versions or cannot list them. - async fn list_versions(&self, prefix: &str) -> Result>, StoreError>; + /// One page of the versions and delete markers under `prefix`, newest first + /// within a key, from `after` on. `None` when the store keeps no versions + /// or cannot list them. + async fn list_versions( + &self, + prefix: &str, + after: Option<&VersionMarker>, + ) -> Result, StoreError>; async fn delete_version(&self, key: &str, version_id: &str) -> Result<(), StoreError>; /// The first `length` bytes as stored. async fn get_range(&self, key: &str, length: u64) -> Result, StoreError>; @@ -196,6 +202,35 @@ pub struct Copied { pub version_id: Option, } +/// Where the next page of versions starts: a key and a version id. +pub type VersionMarker = (String, String); + +#[derive(Debug, Clone, Default)] +pub struct VersionsPage { + pub versions: Vec, + pub next: Option, +} + +/// Every version of `key`, newest first; `None` without versions. +pub async fn versions_of( + store: &dyn Store, + key: &str, +) -> Result>, StoreError> { + let mut out = Vec::new(); + let mut after: Option = None; + loop { + let Some(page) = store.list_versions(key, after.as_ref()).await? else { + return Ok(None); + }; + out.extend(page.versions.into_iter().filter(|v| v.key == key)); + match page.next { + // Keys after `key` itself only share its prefix. + Some(next) if next.0 == key && Some(&next) != after.as_ref() => after = Some(next), + _ => return Ok(Some(out)), + } + } +} + #[derive(Debug, Clone, PartialEq)] pub struct ObjectVersion { pub key: String, diff --git a/donut-sync/src/sync.rs b/donut-sync/src/sync.rs index 516c1848..e41bc838 100644 --- a/donut-sync/src/sync.rs +++ b/donut-sync/src/sync.rs @@ -160,6 +160,7 @@ pub struct SyncService { cloud: bool, backend: Option>, group_cache: Mutex>, + direct_uploads: Option>, } pub struct SyncOptions { @@ -175,6 +176,11 @@ pub struct SyncOptions { impl SyncService { pub fn new(store: Arc, options: SyncOptions) -> Self { + let direct_uploads = options + .transfer + .as_ref() + .filter(|signer| !signer.keyring.unsealing()) + .map(|_| Arc::new(crate::sealer::DirectUploads::default())); Self { store, presigner: options.presigner, @@ -184,6 +190,7 @@ impl SyncService { cloud: options.cloud, backend: options.backend, group_cache: Mutex::new(HashMap::new()), + direct_uploads, } } @@ -215,23 +222,17 @@ impl SyncService { /// Notes the keys handed direct storage URLs, so the sealer seals what /// older apps write there. - async fn note_direct_uploads(&self, keys: &[String], now: i64, expires_in: f64) { - let sealing = self - .transfer - .as_ref() - .is_some_and(|signer| !signer.keyring.unsealing()); - if !sealing || keys.is_empty() { - return; - } - if let Err(e) = - crate::sealer::note_direct_uploads(self.store.as_ref(), keys, expires_ms(now, expires_in)) - .await - { - // The hourly backstop pass still finds these objects. - log::warn(CONTEXT, &format!("Direct uploads not noted err=\"{e}\"")); + fn note_direct_uploads(&self, keys: &[String], now: i64, expires_in: f64) { + if let Some(direct) = &self.direct_uploads { + direct.note(keys, expires_ms(now, expires_in)); } } + /// Set while sealing: the keys handed direct URLs, for the sealer. + pub fn direct_uploads(&self) -> Option<&Arc> { + self.direct_uploads.as_ref() + } + fn get_url(&self, origin: &str, key: &str, expires_in: f64, now: i64) -> String { match &self.transfer { Some(signer) => signer.url(origin, "GET", key, expires_ms(now, expires_in), &[]), @@ -534,9 +535,7 @@ impl SyncService { .presigner .put_url(scoped, expires_in, metadata_or_none, now); let transfer = self.transfer_json(origin, scoped, expires_in, metadata_or_none, now); - self - .note_direct_uploads(&[scoped.to_string()], now, expires_in) - .await; + self.note_direct_uploads(&[scoped.to_string()], now, expires_in); if ctx.mode == Mode::Cloud && field_starts_with(raw_key, "profiles/")? { self.report_profile_usage_async(ctx); @@ -751,7 +750,7 @@ impl SyncService { out_items.push(Value::Object(item)); direct.push(scoped.to_string()); } - self.note_direct_uploads(&direct, now, expires_in).await; + self.note_direct_uploads(&direct, now, expires_in); if ctx.mode == Mode::Cloud { let mut any_profile = false; @@ -1222,11 +1221,10 @@ impl SyncService { .await .map_err(store_error)?; let mut last_modified = first.contents.first().and_then(|o| o.last_modified_ms); - // A sealed store reports a migrated object's time from before the - // migration, which only a HEAD reads. - if self.transfer.is_some() - && let Some(object) = first.contents.first() - { + // A migrated object's time from before the migration is in its + // metadata, which only a HEAD reads; it stays there after the key is + // removed. + if let Some(object) = first.contents.first() { last_modified = self .store .head_object(&object.key) diff --git a/donut-sync/src/sync_tests.rs b/donut-sync/src/sync_tests.rs index 4e580f3c..0bb17cd2 100644 --- a/donut-sync/src/sync_tests.rs +++ b/donut-sync/src/sync_tests.rs @@ -719,3 +719,39 @@ async fn cleanup_orders_migrated_profiles_by_their_time_before_the_migration() { let out = service.cleanup_excess_profiles("u1", 2).await.unwrap(); assert_eq!(out["deletedProfiles"], json!(["pa", "pb"])); } + +/// After a rollback (sealed, unsealed, key removed) profile ages are still +/// the ones from before any rewrite. +#[tokio::test] +async fn cleanup_keeps_the_original_order_after_a_rollback() { + let raw = Arc::new(MemoryStore::default()); + for id in ["pd", "pc", "pb", "pa"] { + raw.set(&format!("users/u1/profiles/{id}/metadata.json"), "{}"); + } + let sealing = Arc::new( + crate::at_rest::Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap(), + ); + let options = crate::migrate::Options { + concurrency: 1, + ..Default::default() + }; + crate::migrate::migrate(raw.clone(), sealing, &options) + .await + .unwrap(); + let unsealing = Arc::new( + crate::at_rest::Keyring::parse("unseal,k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") + .unwrap(), + ); + let unseal = crate::migrate::Options { + unseal: true, + ..options + }; + let report = crate::migrate::migrate(raw.clone(), unsealing, &unseal) + .await + .unwrap(); + assert_eq!(report.unsealed, 4); + // The key is gone: the plain service, as the server runs without it. + let service = service_on(crate::served_store(raw.clone(), None), None); + let out = service.cleanup_excess_profiles("u1", 2).await.unwrap(); + assert_eq!(out["deletedProfiles"], json!(["pa", "pb"])); +} diff --git a/donut-sync/src/transfer_parts.rs b/donut-sync/src/transfer_parts.rs index cf021333..4803cc4e 100644 --- a/donut-sync/src/transfer_parts.rs +++ b/donut-sync/src/transfer_parts.rs @@ -20,8 +20,9 @@ //! Sealing a part twice under the same nonces is safe only for the same //! bytes, so a part is read whole and hashed before anything reaches storage //! (`Ledger`): a part may be sent again with the same bytes, never with -//! others. The ledger is in memory, so after a restart the parts of an upload -//! started before it are refused, and the app starts the upload again. +//! others. The ledger is in memory, so only the process that started an +//! upload takes its parts: after a restart, or while a deploy runs the old +//! and new process side by side, the app is told to start the upload again. use base64::Engine; use bytes::Bytes; @@ -75,7 +76,8 @@ struct Token { /// Whether the parts are sealed: not when the server was unsealing at the /// start (the object then has no marker either). sealed: bool, - issued_ms: i64, + /// The process that started the upload: its ledger holds the parts. + instance: String, } /// Each part's hash and its upload's expiry, by upload id and part number. @@ -83,7 +85,7 @@ type PartHashes = HashMap<(String, u64), ([u8; 32], i64)>; /// The hash of every part sent so far, while its upload lasts. pub struct Ledger { - started_ms: i64, + instance: String, parts: Mutex, buffers: Arc, } @@ -91,7 +93,7 @@ pub struct Ledger { impl Default for Ledger { fn default() -> Self { Self { - started_ms: time::now_ms(), + instance: uuid::Uuid::new_v4().simple().to_string(), parts: Mutex::new(HashMap::new()), buffers: Arc::new(Semaphore::new(PART_BUFFERS)), } @@ -162,6 +164,14 @@ impl Token { } } +fn interrupted() -> Response { + s3_error( + 409, + "UploadInterrupted", + "The server restarted during this upload. Start it again.", + ) +} + fn invalid(code: &str, message: &str) -> Response { s3_error(400, code, message) } @@ -211,7 +221,7 @@ pub async fn start( part_size: PART_SIZE, expires_ms: url_expires_ms + UPLOAD_GRACE_MS, sealed: !signer.keyring.unsealing(), - issued_ms: time::now_ms(), + instance: signer.parts.instance.clone(), }; json_response( 200, @@ -242,6 +252,12 @@ pub async fn serve( match (&parts.method, signed.op.as_deref()) { (&Method::PUT, None) => part(store, signer, &token, signed.part.as_deref(), parts, body).await, (&Method::POST, Some("complete")) => { + // Its parts went through another process's ledger; a mode change + // (sealing or not) is a new process too. + if token.instance != signer.parts.instance { + drain(body).await; + return interrupted(); + } let response = complete(store, &token, body).await; if response.status().is_success() { signer.parts.forget(&token.upload_id); @@ -296,13 +312,9 @@ async fn part( &format!("Part {number} must be {expected} bytes, not {length}."), ); } - if token.issued_ms < signer.parts.started_ms { + if token.instance != signer.parts.instance { drain(body).await; - return s3_error( - 409, - "UploadInterrupted", - "The server restarted during this upload. Start it again.", - ); + return interrupted(); } let Ok(Ok(_buffer)) = tokio::time::timeout(SLOT_WAIT, signer.parts.buffers.clone().acquire_owned()).await @@ -424,7 +436,7 @@ mod tests { part_size: PART_SIZE, expires_ms: 1000, sealed: true, - issued_ms: 0, + instance: "i".to_string(), } } diff --git a/donut-sync/src/verify.rs b/donut-sync/src/verify.rs index bb286260..a0ce20de 100644 --- a/donut-sync/src/verify.rs +++ b/donut-sync/src/verify.rs @@ -273,6 +273,34 @@ async fn open_fully( Ok(()) } +/// The bucket's older versions under `prefix`, page by page; `None` +/// without versions. +async fn noncurrent(store: &dyn Store, prefix: &str) -> Result, String> { + let mut n = Noncurrent::default(); + let mut after = None; + loop { + let Some(page) = store + .list_versions(prefix, after.as_ref()) + .await + .map_err(|e| format!("Versions not listed: {e}"))? + else { + return Ok(None); + }; + for version in page.versions { + if version.delete_marker { + n.delete_markers += 1; + } else if !version.is_latest { + n.versions += 1; + n.bytes += version.size; + } + } + after = page.next; + if after.is_none() { + return Ok(Some(n)); + } + } +} + pub async fn verify( store: Arc, keyring: Arc, @@ -438,22 +466,7 @@ pub async fn verify( } Err(e) => return Err(format!("Sealer state unreadable: {e}")), } - report.noncurrent = store - .list_versions(options.prefix.as_deref().unwrap_or("")) - .await - .map_err(|e| format!("Versions not listed: {e}"))? - .map(|versions| { - let mut n = Noncurrent::default(); - for version in versions.iter().filter(|v| !v.is_latest || v.delete_marker) { - if version.delete_marker { - n.delete_markers += 1; - } else { - n.versions += 1; - n.bytes += version.size; - } - } - n - }); + report.noncurrent = noncurrent(store.as_ref(), options.prefix.as_deref().unwrap_or("")).await?; let line = format!( "Verify done complete={} {} elapsed_ms={}", report.complete(), @@ -605,6 +618,7 @@ mod tests { key: "x".into(), reason: "r".into(), at: 1, + settle: None, }], }; sealed diff --git a/donut-sync/tests/at_rest.rs b/donut-sync/tests/at_rest.rs index db33c0b3..2192bfd1 100644 --- a/donut-sync/tests/at_rest.rs +++ b/donut-sync/tests/at_rest.rs @@ -4,7 +4,7 @@ mod common; -use common::{SYNC_TOKEN, Server, bucket, http, post, raw_store, run_command, storage_env}; +use common::{SYNC_TOKEN, Server, fresh_bucket, http, post, raw_store, run_command, storage_env}; use donut_sync::store::{Store, Upload, collect, once}; use serde_json::{Value, json}; @@ -16,7 +16,7 @@ fn sample(len: usize) -> Vec { } async fn sealed_server(name: &str, keys: &str) -> (Server, String) { - let bucket = bucket(name); + let bucket = fresh_bucket(name).await; ( Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(keys))]).await, bucket, @@ -47,7 +47,7 @@ async fn upload(server: &Server, key: &str, body: Vec, updated_at: Option<&s presign["url"] .as_str() .unwrap() - .starts_with(common::S3_ENDPOINT) + .starts_with(&common::storage().endpoint) ); assert_eq!(presign["transfer"]["partSize"], json!(8 * 1024 * 1024)); let url = presign["transfer"]["url"].as_str().unwrap().to_string(); @@ -123,7 +123,7 @@ async fn transfers_go_through_the_server_and_are_sealed_in_storage() { // Older apps still upload to storage directly, so its address is named. assert_eq!( readyz, - json!({ "status": "ready", "s3": true, "storageEndpoint": common::S3_ENDPOINT }) + json!({ "status": "ready", "s3": true, "storageEndpoint": &common::storage().endpoint }) ); assert!( server @@ -277,13 +277,13 @@ async fn transfer_urls_refuse_what_s3_would_refuse() { #[tokio::test] async fn without_the_key_nothing_changes() { - let server = Server::start(&bucket("plain"), &[]).await; + let server = Server::start(&fresh_bucket("plain").await, &[]).await; let (_, presign) = call(&server, "presign-upload", json!({ "key": "k" })).await; assert!( presign["url"] .as_str() .unwrap() - .starts_with("http://127.0.0.1:8987/") + .starts_with(&format!("{}/", common::storage().endpoint)) ); let response = http() .get(format!("{}/v1/transfer?key=k", server.url)) @@ -321,7 +321,7 @@ async fn seed_plaintext(bucket: &str, objects: &[(&str, Vec)]) { #[tokio::test] async fn plaintext_and_sealed_objects_read_side_by_side() { - let bucket = bucket("mixed"); + let bucket = fresh_bucket("mixed").await; let old = sample(100_000); seed_plaintext(&bucket, &[("proxies/old.json", old.clone())]).await; let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await; @@ -369,7 +369,7 @@ async fn plaintext_and_sealed_objects_read_side_by_side() { #[tokio::test] async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() { - let bucket = bucket("migrate"); + let bucket = fresh_bucket("migrate").await; let objects: Vec<(String, Vec)> = (0..25) .map(|i| (format!("profiles/p/files/{i:02}"), sample(i * 9_000))) .collect(); @@ -548,7 +548,7 @@ async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() { #[tokio::test] async fn a_bad_key_stops_the_server_at_start() { - let storage = storage_env(&bucket("badkey")); + let storage = storage_env(&fresh_bucket("badkey").await); let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect(); env.push(("SYNC_TOKEN", SYNC_TOKEN)); env.push(("SYNC_AT_REST_KEY", "k1:AAAA")); @@ -766,6 +766,25 @@ async fn older_apps_upload_directly_and_the_sealer_seals_it() { // Readable before it is sealed. assert_eq!(download(&server, key).await, (200, plain.clone())); + // The server notes presigned keys in memory and writes them every 10 s. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let pending = raw + .list_objects_v2(&donut_sync::store::ListInput { + prefix: Some(".donut-sync-at-rest/pending/".to_string()), + ..Default::default() + }) + .await + .unwrap(); + if !pending.contents.is_empty() { + break; + } + assert!( + std::time::Instant::now() < deadline, + "no pending record written" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } let keyring = std::sync::Arc::new(donut_sync::at_rest::Keyring::parse(K1).unwrap()); let sealer = donut_sync::sealer::Sealer::new(raw.clone(), keyring) .await @@ -795,7 +814,7 @@ async fn older_apps_upload_directly_and_the_sealer_seals_it() { #[tokio::test] async fn a_rollback_unseals_everything_and_the_key_can_go() { - let bucket = bucket("unseal"); + let bucket = fresh_bucket("unseal").await; let key_sealed = "profiles/p/files/sealed"; let key_parts = "profiles/p/files/parts"; let small = sample(70_000); @@ -887,3 +906,84 @@ async fn a_rollback_unseals_everything_and_the_key_can_go() { assert_eq!(after["metadata"]["updated-at"], "3"); assert!(after["metadata"].get("donut-at-rest").is_none()); } + +/// What settling a rewrite relies on, checked against the store the suite +/// runs on: version ids on HEAD, copy and delete, versions listed newest +/// first, copies of a given version, and a copy conditional on the source. +#[tokio::test] +async fn the_store_keeps_what_a_settle_needs() { + use donut_sync::store::{CopySpec, StoreError, versions_of}; + let bucket = fresh_bucket("capabilities").await; + let raw = raw_store(&bucket); + raw.create_bucket().await.ok(); + raw + .put_object("k", b"one".to_vec(), "text/plain", &[]) + .await + .unwrap(); + let first = raw.head_object("k").await.unwrap(); + raw + .put_object("k", b"two".to_vec(), "text/plain", &[]) + .await + .unwrap(); + let second = raw.head_object("k").await.unwrap(); + + // A copy that names a source ETag the source no longer has. + let stale = raw + .copy_object_with( + "k", + "copy", + CopySpec { + if_match: first.etag.as_deref(), + ..CopySpec::default() + }, + ) + .await; + let honored = matches!(stale, Err(StoreError::Service { status: 412, .. })); + println!("copy_source_if_match_honored={honored}"); + + let Some(history) = versions_of(&raw, "k").await.unwrap() else { + println!("versions=unsupported"); + return; + }; + println!("versions=supported"); + assert!( + honored, + "a store that keeps versions must honor a conditional copy" + ); + let (Some(v1), Some(v2)) = (first.version_id.clone(), second.version_id.clone()) else { + panic!("HEAD gave no version id: {first:?} {second:?}"); + }; + let ids: Vec<&str> = history.iter().map(|v| v.version_id.as_str()).collect(); + assert_eq!(ids, [v2.as_str(), v1.as_str()], "newest first"); + assert!(history[0].is_latest && !history[1].is_latest); + + let copied = raw + .copy_object_with( + "k", + "k", + CopySpec { + source_version: Some(&v1), + ..CopySpec::default() + }, + ) + .await + .unwrap(); + let restored = copied.version_id.expect("a copy names its new version"); + assert_eq!( + collect(raw.get_object("k").await.unwrap().body) + .await + .unwrap(), + b"one" + ); + let marker = raw + .delete_object("k") + .await + .unwrap() + .expect("a delete names its marker"); + let history = versions_of(&raw, "k").await.unwrap().unwrap(); + assert_eq!(history[0].version_id, marker); + assert!(history[0].delete_marker); + assert_eq!(history[1].version_id, restored); + donut_sync::migrate::purge(&raw, "k").await.unwrap(); + assert!(versions_of(&raw, "k").await.unwrap().unwrap().is_empty()); +} diff --git a/donut-sync/tests/common/mod.rs b/donut-sync/tests/common/mod.rs index 1ca0c8c4..e7cd6e87 100644 --- a/donut-sync/tests/common/mod.rs +++ b/donut-sync/tests/common/mod.rs @@ -1,5 +1,11 @@ //! Runs the real binary against the S3 test server of //! `donut-sync/docker-compose.yml` (rclone on port 8987). +//! +//! The at-rest suite can run against a real S3 store instead, to check what +//! it relies on there (versions, conditional copies): set +//! `DONUT_SYNC_TEST_S3_ENDPOINT`, `_REGION`, `_ACCESS_KEY_ID`, +//! `_SECRET_ACCESS_KEY` and `_BUCKET` (a scratch bucket each test empties +//! first, so run with `--test-threads=1`). #![allow(dead_code)] @@ -15,29 +21,88 @@ use donut_sync::store::{Store, StoreError}; pub const S3_ENDPOINT: &str = "http://127.0.0.1:8987"; pub const SYNC_TOKEN: &str = "test-sync-token-0123456789abcdef"; +/// Where the at-rest suite's storage is: the compose server, or what the +/// `DONUT_SYNC_TEST_S3_*` variables name. +pub struct Storage { + pub endpoint: String, + pub region: String, + pub access_key_id: String, + pub secret_access_key: String, + /// One bucket for every test, emptied before each. + pub bucket: Option, +} + +pub fn storage() -> Storage { + let var = |name: &str, default: &str| { + std::env::var(format!("DONUT_SYNC_TEST_S3_{name}")) + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| default.to_string()) + }; + Storage { + endpoint: var("ENDPOINT", S3_ENDPOINT), + region: var("REGION", "us-east-1"), + access_key_id: var("ACCESS_KEY_ID", "minioadmin"), + secret_access_key: var("SECRET_ACCESS_KEY", "minioadmin"), + bucket: std::env::var("DONUT_SYNC_TEST_S3_BUCKET") + .ok() + .filter(|v| !v.is_empty()), + } +} + /// The test storage, without the server: what is really stored. pub fn raw_store(bucket: &str) -> S3Store { + let storage = storage(); S3Store::new( - Location::parse(S3_ENDPOINT, bucket, true).unwrap(), + Location::parse(&storage.endpoint, bucket, true).unwrap(), Credentials { - access_key_id: "minioadmin".to_string(), - secret_access_key: "minioadmin".to_string(), + access_key_id: storage.access_key_id, + secret_access_key: storage.secret_access_key, }, - "us-east-1".to_string(), + storage.region, ) } /// The storage settings the server takes, for a command run to its exit. pub fn storage_env(bucket: &str) -> Vec<(&'static str, String)> { + let storage = storage(); vec![ - ("S3_ENDPOINT", S3_ENDPOINT.to_string()), - ("S3_ACCESS_KEY_ID", "minioadmin".to_string()), - ("S3_SECRET_ACCESS_KEY", "minioadmin".to_string()), + ("S3_ENDPOINT", storage.endpoint), + ("S3_REGION", storage.region), + ("S3_ACCESS_KEY_ID", storage.access_key_id), + ("S3_SECRET_ACCESS_KEY", storage.secret_access_key), ("S3_BUCKET", bucket.to_string()), ("S3_FORCE_PATH_STYLE", "true".to_string()), ] } +/// A bucket of the test's own, or the fixed one, emptied of every version +/// and unfinished upload. +pub async fn fresh_bucket(name: &str) -> String { + let Some(fixed) = storage().bucket else { + return bucket(name); + }; + let raw = raw_store(&fixed); + loop { + let Some(page) = raw.list_versions("", None).await.unwrap() else { + break; + }; + if page.versions.is_empty() { + break; + } + for version in page.versions { + raw + .delete_version(&version.key, &version.version_id) + .await + .unwrap(); + } + } + for upload in raw.list_multipart("").await.unwrap_or_default() { + let _ = raw.abort_multipart(&upload.key, &upload.upload_id).await; + } + fixed +} + pub fn http() -> reqwest::Client { let _ = rustls::crypto::ring::default_provider().install_default(); reqwest::Client::builder().no_proxy().build().unwrap() @@ -45,13 +110,10 @@ pub fn http() -> reqwest::Client { /// Waits for the S3 test server, or fails with how to start it. pub async fn wait_for_s3() { - let store = S3Store::new( - Location::parse(S3_ENDPOINT, "probe-availability", true).unwrap(), - Credentials { - access_key_id: "minioadmin".to_string(), - secret_access_key: "minioadmin".to_string(), - }, - "us-east-1".to_string(), + let store = raw_store( + &storage() + .bucket + .unwrap_or_else(|| "probe-availability".to_string()), ); let deadline = Instant::now() + Duration::from_secs(30); loop { @@ -87,9 +149,10 @@ impl Server { let port = free_port(); let mut env: Vec<(String, String)> = vec![ ("SYNC_TOKEN".into(), SYNC_TOKEN.into()), - ("S3_ENDPOINT".into(), S3_ENDPOINT.into()), - ("S3_ACCESS_KEY_ID".into(), "minioadmin".into()), - ("S3_SECRET_ACCESS_KEY".into(), "minioadmin".into()), + ("S3_ENDPOINT".into(), storage().endpoint), + ("S3_REGION".into(), storage().region), + ("S3_ACCESS_KEY_ID".into(), storage().access_key_id), + ("S3_SECRET_ACCESS_KEY".into(), storage().secret_access_key), ("S3_BUCKET".into(), bucket.into()), ("S3_FORCE_PATH_STYLE".into(), "true".into()), ("PORT".into(), port.to_string()), diff --git a/src-tauri/src/sync/client.rs b/src-tauri/src/sync/client.rs index 7010b3ea..31e194dc 100644 --- a/src-tauri/src/sync/client.rs +++ b/src-tauri/src/sync/client.rs @@ -23,6 +23,33 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(8); /// only when nothing arrives for a full minute. const READ_TIMEOUT: Duration = Duration::from_secs(60); +/// Transfers through a sync server's own URLs (a server that encrypts at +/// rest) at a time, across every profile this app syncs: such a server +/// serves a limited number per account, and more only wait there. +const SERVER_TRANSFERS: usize = 6; +static SERVER_TRANSFER_SLOTS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| std::sync::Arc::new(tokio::sync::Semaphore::new(SERVER_TRANSFERS))); +/// The path of a sync server's own transfer URLs. +const SERVER_TRANSFER_PATH: &str = "/v1/transfer?"; + +/// How long a busy answer (503, 429) is tried again before the transfer +/// fails, and the longest single wait. +const BUSY_BUDGET: Duration = Duration::from_secs(300); +const BUSY_WAIT_MAX: Duration = Duration::from_secs(30); + +/// What a busy answer asks to wait, in seconds. +fn retry_after(response: &reqwest::Response) -> Option { + response + .headers() + .get(reqwest::header::RETRY_AFTER)? + .to_str() + .ok()? + .trim() + .parse() + .ok() + .map(Duration::from_secs) +} + /// Tries per part of an upload in parts. The file's own retries come on top. const PART_ATTEMPTS: u32 = 3; const PART_RETRY_DELAY: Duration = Duration::from_secs(1); @@ -275,6 +302,16 @@ impl SyncClient { content_type: Option<&str>, metadata: Option<&std::collections::HashMap>, ) -> SyncResult<()> { + let _slot = match transfer { + Some(_) => Some( + SERVER_TRANSFER_SLOTS + .clone() + .acquire_owned() + .await + .expect("the semaphore is never closed"), + ), + None => None, + }; match transfer { Some(transfer) if data.len() as u64 > transfer.part_size => { self @@ -308,17 +345,47 @@ impl SyncClient { req } + /// Sends `build()`. For a sync server's own transfer URL, again while the + /// answer is busy (503, 429), waiting what Retry-After asks or a growing + /// backoff, for at most `BUSY_BUDGET`: that server limits transfers per + /// account. Storage's own answers go back at once, to the file's retries. + async fn send_patiently( + &self, + url: &str, + build: impl Fn() -> reqwest::RequestBuilder, + ) -> SyncResult { + let patient = url.contains(SERVER_TRANSFER_PATH); + let started = std::time::Instant::now(); + let mut backoff = Duration::from_secs(1); + loop { + // The storage host here may come from a presigned URL, so on a + // self-hosted server it is whatever the server signed against, + // frequently an address only the server can resolve. `reqwest`'s own + // Display collapses that to "error sending request"; report the + // innermost cause and the host it names. + let response = build() + .send() + .await + .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?; + let busy = matches!(response.status().as_u16(), 429 | 503); + let wait = retry_after(&response).unwrap_or(backoff).min(BUSY_WAIT_MAX); + if !patient || !busy || started.elapsed() + wait > BUSY_BUDGET { + return Ok(response); + } + log::debug!("Transfer busy, waiting wait_ms={}", wait.as_millis()); + tokio::time::sleep(wait).await; + backoff = (backoff * 2).min(BUSY_WAIT_MAX); + } + } + /// Sends a request of an upload in parts; any answer but a success is an /// error carrying the server's own words. async fn send_part_request( &self, url: &str, - req: reqwest::RequestBuilder, + build: impl Fn() -> reqwest::RequestBuilder, ) -> SyncResult { - let response = req - .send() - .await - .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?; + let response = self.send_patiently(url, build).await?; if !response.status().is_success() { let status = response.status(); let body = response.text().await.unwrap_or_default(); @@ -339,15 +406,17 @@ impl SyncClient { content_type: Option<&str>, metadata: Option<&std::collections::HashMap>, ) -> SyncResult<()> { - let start = Self::with_upload_headers( - self - .client - .post(&transfer.url) - .header(UPLOAD_LENGTH_HEADER, data.len().to_string()) - .header("Content-Length", "0"), - content_type, - metadata, - ); + let start = || { + Self::with_upload_headers( + self + .client + .post(&transfer.url) + .header(UPLOAD_LENGTH_HEADER, data.len().to_string()) + .header("Content-Length", "0"), + content_type, + metadata, + ) + }; let started: PartsStarted = self .send_part_request(&transfer.url, start) .await? @@ -392,11 +461,13 @@ impl SyncClient { let mut attempt = 0; let etag = loop { attempt += 1; - let req = self - .client - .put(&url) - .header("Content-Length", chunk.len().to_string()) - .body(chunk.to_vec()); + let req = || { + self + .client + .put(&url) + .header("Content-Length", chunk.len().to_string()) + .body(chunk.to_vec()) + }; match self.send_part_request(&url, req).await { Ok(response) => { break response @@ -421,10 +492,12 @@ impl SyncClient { &transfer.url, &[("upload", &started.upload), ("op", "complete")], )?; - let req = self - .client - .post(&complete) - .json(&serde_json::json!({ "parts": etags })); + let req = || { + self + .client + .post(&complete) + .json(&serde_json::json!({ "parts": etags })) + }; self.send_part_request(&complete, req).await.map(|_| ()) } @@ -436,24 +509,18 @@ impl SyncClient { content_type: Option<&str>, metadata: Option<&std::collections::HashMap>, ) -> SyncResult<()> { - let req = Self::with_upload_headers( - self - .client - .put(presigned_url) - .header("Content-Length", data.len().to_string()) - .body(data.to_vec()), - content_type, - metadata, - ); - - // The storage host here comes from the presigned URL, so on a self-hosted - // server it is whatever the server signed against, frequently an address - // only the server can resolve. `reqwest`'s own Display collapses that to - // "error sending request", which is why this failure used to be - // undiagnosable; report the innermost cause and the host it names. - let response = req.send().await.map_err(|e| { - SyncError::NetworkError(super::preflight::transport_reason_for(presigned_url, &e)) - })?; + let req = || { + Self::with_upload_headers( + self + .client + .put(presigned_url) + .header("Content-Length", data.len().to_string()) + .body(data.to_vec()), + content_type, + metadata, + ) + }; + let response = self.send_patiently(presigned_url, req).await?; if !response.status().is_success() { let status = response.status(); @@ -467,9 +534,19 @@ impl SyncClient { } pub async fn download_bytes(&self, presigned_url: &str) -> SyncResult> { - let response = self.client.get(presigned_url).send().await.map_err(|e| { - SyncError::NetworkError(super::preflight::transport_reason_for(presigned_url, &e)) - })?; + let _slot = match presigned_url.contains(SERVER_TRANSFER_PATH) { + true => Some( + SERVER_TRANSFER_SLOTS + .clone() + .acquire_owned() + .await + .expect("the semaphore is never closed"), + ), + false => None, + }; + let response = self + .send_patiently(presigned_url, || self.client.get(presigned_url)) + .await?; if !response.status().is_success() { return Err(SyncError::NetworkError(format!( @@ -852,6 +929,41 @@ mod tests { .is_err()); } + #[tokio::test] + async fn a_busy_server_is_waited_for_as_it_asks() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(503).insert_header("retry-after", "1")) + .up_to_n_times(2) + .with_priority(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"done".to_vec())) + .expect(1) + .mount(&server) + .await; + let started = std::time::Instant::now(); + let url = format!("{}/v1/transfer?key=k&expires=1&sig=s", server.uri()); + assert_eq!(client().download_bytes(&url).await.unwrap(), b"done"); + assert!( + started.elapsed() >= Duration::from_secs(2), + "waited Retry-After twice" + ); + } + + #[tokio::test] + async fn a_busy_storage_url_fails_at_once_for_the_file_retries() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(503).insert_header("retry-after", "30")) + .expect(1) + .mount(&server) + .await; + let url = format!("{}/bucket/k?X-Amz-Signature=s", server.uri()); + assert!(client().download_bytes(&url).await.is_err()); + } + #[test] fn answers_from_servers_that_do_not_seal_have_no_transfer() { let old: PresignUploadResponse = diff --git a/src-tauri/src/sync/engine.rs b/src-tauri/src/sync/engine.rs index 1599ea47..d05402e9 100644 --- a/src-tauri/src/sync/engine.rs +++ b/src-tauri/src/sync/engine.rs @@ -128,22 +128,6 @@ pub async fn cancel_profile_sync(profile_id: String) -> Result { /// Upload/download concurrency limit const SYNC_CONCURRENCY: usize = 32; -/// The limit when files go through the sync server itself (a server that -/// encrypts at rest): it serves a few transfers per account at a time, and -/// more only queue there. -const SERVER_TRANSFER_CONCURRENCY: usize = 6; - -/// The path of a sync server's own transfer URLs. -const SERVER_TRANSFER_PATH: &str = "/v1/transfer?"; - -fn transfer_concurrency(through_server: bool) -> usize { - if through_server { - SERVER_TRANSFER_CONCURRENCY - } else { - SYNC_CONCURRENCY - } -} - /// Max retries for individual file uploads/downloads const MAX_FILE_RETRIES: u32 = 3; @@ -1576,8 +1560,7 @@ impl SyncEngine { .store(already_bytes, Ordering::Relaxed); tracker.emit_final(); - let through_server = url_map.values().any(|(_, transfer)| transfer.is_some()); - let semaphore = Arc::new(Semaphore::new(transfer_concurrency(through_server))); + let semaphore = Arc::new(Semaphore::new(SYNC_CONCURRENCY)); let client = self.client.clone(); let profile_dir = profile_dir.to_path_buf(); let profile_id_owned = profile_id.to_string(); @@ -1848,10 +1831,7 @@ impl SyncEngine { .store(already_bytes, Ordering::Relaxed); tracker.emit_final(); - let through_server = url_map - .values() - .any(|url| url.contains(SERVER_TRANSFER_PATH)); - let semaphore = Arc::new(Semaphore::new(transfer_concurrency(through_server))); + let semaphore = Arc::new(Semaphore::new(SYNC_CONCURRENCY)); let client = self.client.clone(); let profile_dir = profile_dir.to_path_buf(); let profile_id_owned = profile_id.to_string();