1use crate::SyncError;
45use crate::dav::{self, DAV};
46use serde::{Deserialize, Serialize};
47use std::collections::{BTreeMap, BTreeSet};
48use std::io::{Read, Seek, SeekFrom, Write};
49use std::path::{Path, PathBuf};
50use std::sync::Mutex;
51use std::time::{Duration, Instant, SystemTime};
52
53#[derive(Debug, Clone, Copy)]
57pub(crate) struct Limits {
58 pub(crate) wait: Duration,
59 pub(crate) small: Duration,
60 pub(crate) large: Duration,
61 pub(crate) pull: Duration,
62 pub(crate) blob: Duration,
64 pub(crate) floor: u64,
67 pub(crate) stall: Duration,
68}
69
70pub(crate) const LIMITS: Limits = Limits { wait: Duration::from_secs(10), small: Duration::from_secs(10), large: Duration::from_secs(60), pull: Duration::from_secs(30), blob: Duration::from_secs(15 * 60), floor: FLOOR, stall: Duration::from_secs(30) };
71
72pub const FLOOR: u64 = 16 << 10;
75
76pub(crate) const SEND_LIMITS: Limits = Limits { wait: Duration::from_secs(20), small: Duration::from_secs(30), large: Duration::from_secs(60), pull: Duration::from_secs(20 * 60), blob: Duration::from_secs(15 * 60), floor: FLOOR, stall: Duration::from_secs(30) };
80
81pub const LOOK_AGAIN: i64 = 6 * 3600;
83const SEAL_AGAIN: i64 = 6 * 3600;
85const LIST_AGAIN: i64 = 10 * 60;
87const LARGEST: u64 = 64 << 20;
89const TEXTS_LARGEST: u64 = 4 << 20;
91const LISTING: u64 = 4 << 20;
93const OVERLAP: u64 = 1 << 20;
95
96const XML: &str = "application/xml; charset=utf-8";
97const LIST: &str = r#"<?xml version="1.0" encoding="utf-8"?><d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns"><d:prop><d:resourcetype/><d:getetag/><d:getcontentlength/><oc:checksums/></d:prop></d:propfind>"#;
98const OC: &str = "http://owncloud.org/ns";
100const PRINCIPAL: &str = r#"<?xml version="1.0" encoding="utf-8"?><d:propfind xmlns:d="DAV:"><d:prop><d:current-user-principal/></d:prop></d:propfind>"#;
101
102pub fn cache_of(memory: &Path) -> PathBuf {
106 memory.with_file_name("remote")
107}
108
109fn state_path(memory: &Path) -> PathBuf {
111 memory.with_file_name("remote.toml")
112}
113
114fn shown(folder: &Path) -> String {
115 folder.display().to_string()
116}
117
118pub fn host_of(url: &str) -> String {
120 let rest = url.split_once("://").map_or(url, |(_, rest)| rest);
121 let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
122 authority.rsplit('@').next().unwrap_or("").to_string()
123}
124
125#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
131pub struct State {
132 #[serde(default, skip_serializing_if = "Option::is_none")]
134 pub on: Option<bool>,
135 #[serde(default, skip_serializing_if = "String::is_empty")]
138 pub given: String,
139 #[serde(default)]
141 pub asked: i64,
142 #[serde(default, skip_serializing_if = "Option::is_none")]
145 pub send: Option<bool>,
146 #[serde(default, skip_serializing_if = "String::is_empty")]
148 pub folder: String,
149 #[serde(default, skip_serializing_if = "String::is_empty")]
151 pub account: String,
152 #[serde(default, skip_serializing_if = "String::is_empty")]
153 pub url: String,
154 #[serde(default)]
156 pub confirmed: i64,
157 #[serde(default)]
159 pub sealed: i64,
160 #[serde(default)]
162 pub looked: i64,
163 #[serde(default)]
165 pub last: i64,
166 #[serde(default)]
167 pub tried: i64,
168 #[serde(default, skip_serializing_if = "String::is_empty")]
171 pub said: String,
172 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
177 pub left: BTreeMap<String, u64>,
178 #[serde(default, skip_serializing_if = "Option::is_none")]
183 pub own_entry: Option<(u64, i64)>,
184 #[serde(default, skip_serializing_if = "String::is_empty")]
188 pub mode: String,
189 #[serde(default, skip_serializing_if = "String::is_empty")]
191 pub place: String,
192 #[serde(default)]
194 pub pushed: i64,
195 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
197 pub files: BTreeMap<String, Fetched>,
198 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
200 pub folders: BTreeMap<String, Looked>,
201 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
203 pub sent: BTreeMap<String, Sent>,
204}
205
206#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
208pub struct Fetched {
209 #[serde(default)]
210 pub etag: String,
211 #[serde(default)]
212 pub size: u64,
213}
214
215#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
217pub struct Looked {
218 #[serde(default)]
219 pub etag: String,
220 #[serde(default)]
221 pub at: i64,
222}
223
224#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
227pub struct Sent {
228 #[serde(default)]
229 pub size: u64,
230 #[serde(default)]
231 pub modified: u64,
232 #[serde(default)]
233 pub etag: String,
234}
235
236impl State {
237 pub fn load(memory: &Path) -> State {
238 std::fs::read_to_string(state_path(memory)).ok().and_then(|text| toml::from_str(&text).ok()).unwrap_or_default()
239 }
240
241 fn write(&self, memory: &Path) -> Result<(), String> {
242 let text = toml::to_string(self).map_err(|e| e.to_string())?;
243 crate::share::write_atomically(&state_path(memory), format!("# Sioul: the sharing folder fetched from its server too (docs/database.md). Never shared.\n{text}").as_bytes())
244 }
245
246 fn save(&self, memory: &Path) -> Result<(), String> {
249 let path = state_path(memory);
250 sioul_core::filelock::with_lock(&path, || {
251 let there = State::load(memory);
252 State { on: there.on, given: there.given, asked: there.asked, send: there.send, ..self.clone() }.write(memory)
253 })
254 }
255
256 fn save_pulled(&self, memory: &Path, ended: i64, loaded_left: &BTreeMap<String, u64>) -> Result<(), String> {
273 let path = state_path(memory);
274 sioul_core::filelock::with_lock(&path, || {
275 let mut merged = State::load(memory);
276 if (merged.url.as_str(), merged.folder.as_str(), merged.account.as_str()) != (self.url.as_str(), self.folder.as_str(), self.account.as_str()) && !merged.url.is_empty() {
277 return Ok(());
278 }
279 let ours_later = self.tried >= merged.tried || merged.tried > ended;
280 let (later, earlier) = if ours_later { (self.clone(), merged.clone()) } else { (merged.clone(), self.clone()) };
281 merged.url = later.url.clone();
282 merged.folder = later.folder.clone();
283 merged.account = later.account.clone();
284 merged.tried = later.tried;
285 merged.last = [later.last, earlier.last].into_iter().filter(|last| *last <= ended).max().unwrap_or(self.last);
287 merged.said = later.said.clone();
288 merged.confirmed = later.confirmed;
289 merged.sealed = later.sealed.max(earlier.sealed);
290 merged.files = earlier.files.into_iter().chain(later.files).collect();
291 merged.folders = earlier.folders.into_iter().chain(later.folders).collect();
292 let brought: Vec<&String> = loaded_left.keys().filter(|name| !self.left.contains_key(*name)).collect();
293 merged.left.retain(|name, _| !brought.contains(&name));
294 merged.left.extend(self.left.iter().filter(|(name, size)| loaded_left.get(*name) != Some(*size)).map(|(name, size)| (name.clone(), *size)));
296 merged.own_entry = [earlier.own_entry, later.own_entry].into_iter().flatten().max_by_key(|(_, at)| *at);
297 merged.write(memory)
298 })
299 }
300
301 pub fn choose(memory: &Path, change: impl FnOnce(&mut State)) -> Result<State, String> {
303 let path = state_path(memory);
304 sioul_core::filelock::with_lock(&path, || {
305 let mut state = State::load(memory);
306 change(&mut state);
307 state.write(memory)?;
308 Ok(state)
309 })
310 }
311
312 pub fn confirmed_for(&self, folder: &Path) -> bool {
314 self.confirmed > 0 && !self.url.is_empty() && self.folder == shown(folder)
315 }
316
317 pub fn fetching(&self) -> bool {
319 self.on != Some(false)
320 }
321
322 pub fn pulls_well(&self, folder: &Path) -> bool {
329 self.mode != MIRROR && self.fetching() && self.confirmed_for(folder) && self.last > 0 && self.said.is_empty() && self.left.is_empty()
330 }
331
332 pub fn went_through_since(&self, folder: &Path, since: i64, now: i64) -> bool {
338 self.pulls_well(folder) && (since..=now).contains(&self.last)
339 }
340
341 pub fn sending(&self) -> bool {
344 self.fetching() && self.send != Some(false)
345 }
346
347 pub fn host(&self) -> String {
349 host_of(&self.url)
350 }
351
352 pub fn look_due(&self, folder: &Path, now: i64) -> bool {
356 self.folder != shown(folder) || self.asked > self.looked || (!self.confirmed_for(folder) && (self.looked == 0 || now - self.looked >= LOOK_AGAIN))
357 }
358}
359
360type Stamp = [Option<(u64, SystemTime)>; 3];
366static ATTACHED: Mutex<BTreeMap<PathBuf, (Stamp, Option<PathBuf>)>> = Mutex::new(BTreeMap::new());
367
368fn stamp_of(path: &Path) -> Option<(u64, SystemTime)> {
369 std::fs::metadata(path).ok().map(|m| (m.len(), m.modified().unwrap_or(SystemTime::UNIX_EPOCH)))
370}
371
372pub fn attach(folder: &Path, memory: &Path) -> Option<PathBuf> {
378 let cache = cache_of(memory);
379 let stamp = [stamp_of(&state_path(memory)), stamp_of(&folder.join("seal.toml")), stamp_of(&cache.join("seal.toml"))];
380 let mut attached = ATTACHED.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
381 if let Some((was, chosen)) = attached.get(folder)
382 && *was == stamp
383 {
384 return chosen.clone();
385 }
386 let chosen = usable(folder, memory).then_some(cache);
387 attached.insert(folder.to_path_buf(), (stamp, chosen.clone()));
388 chosen
389}
390
391pub fn overlay(folder: &Path) -> Option<PathBuf> {
394 ATTACHED.lock().ok()?.get(folder).and_then(|(_, chosen)| chosen.clone())
395}
396
397fn seal_of(folder: &Path) -> Option<Vec<u8>> {
399 let path = folder.join("seal.toml");
400 let size = std::fs::metadata(&path).ok()?.len();
401 (size <= crate::share::SMALL_FILE).then(|| std::fs::read(&path).ok()).flatten()
402}
403
404fn usable(folder: &Path, memory: &Path) -> bool {
406 let state = State::load(memory);
407 state.mode != MIRROR && state.folder == shown(folder) && seal_of(folder).is_some_and(|here| seal_of(&cache_of(memory)).is_some_and(|fetched| fetched == here))
408}
409
410pub fn forget(memory: &Path, folder: &Path) {
412 let _ = std::fs::remove_dir_all(cache_of(memory));
413 let _ = std::fs::remove_file(state_path(memory));
414 let _ = std::fs::remove_file(crate::blobs::Wanted::path(memory));
415 let _ = std::fs::remove_file(Lane::Main.memory(memory));
416 let _ = std::fs::remove_file(Lane::Urgent.memory(memory));
417 if let Ok(mut attached) = ATTACHED.lock() {
418 attached.remove(folder);
419 }
420}
421
422pub struct Login {
428 pub account: String,
429 pub url: String,
430 pub user: String,
431 pub password: Option<String>,
432}
433
434pub(crate) struct Server {
440 agent: ureq::Agent,
441 authorization: String,
442 started: Instant,
443 limits: Limits,
444}
445
446#[derive(Debug, Default)]
448pub(crate) struct Listing {
449 pub(crate) etag: String,
450 pub(crate) items: BTreeMap<String, Item>,
451}
452
453#[derive(Debug, Default, Clone)]
454pub(crate) struct Item {
455 pub(crate) dir: bool,
456 pub(crate) etag: String,
457 pub(crate) size: u64,
458 pub(crate) checksum: String,
460}
461
462pub(crate) struct Got {
465 start: u64,
466 bytes: Vec<u8>,
467 error: Option<SyncError>,
468}
469
470#[derive(Debug)]
474pub(crate) struct PutFailed {
475 pub(crate) error: SyncError,
476 pub(crate) kind: &'static str,
477 pub(crate) words: String,
478 pub(crate) seconds: u64,
479 pub(crate) sent: u64,
480 pub(crate) total: u64,
481}
482
483struct Counted {
485 inner: std::io::Cursor<Vec<u8>>,
486 moved: std::sync::Arc<std::sync::atomic::AtomicU64>,
487}
488
489impl Read for Counted {
490 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
491 let n = self.inner.read(buf)?;
492 self.moved.fetch_add(n as u64, std::sync::atomic::Ordering::Relaxed);
493 Ok(n)
494 }
495}
496
497fn failed(e: &ureq::Error) -> SyncError {
498 match e {
499 ureq::Error::Tls(_) => SyncError::Tls(e.to_string()),
500 _ if format!("{e:?}").starts_with("Rustls") => SyncError::Tls(e.to_string()),
501 _ => SyncError::Network(e.to_string()),
502 }
503}
504
505fn header(response: &ureq::http::Response<ureq::Body>, name: &str) -> Option<String> {
506 response.headers().get(name).and_then(|v| v.to_str().ok()).map(str::to_string)
507}
508
509impl Server {
510 pub(crate) fn new(login: &Login, limits: Limits) -> Result<Server, SyncError> {
511 let password = login.password.as_deref().ok_or(SyncError::NoPassword)?;
512 Ok(Server {
513 agent: dav::shared_agent(),
514 authorization: format!("Basic {}", sioul_core::lines::base64_encode(format!("{}:{password}", login.user).as_bytes())),
515 started: Instant::now(),
516 limits,
517 })
518 }
519
520 pub(crate) fn request(&self, large: bool, method: &str, url: &str, headers: &[(&str, &str)], body: Vec<u8>) -> Result<ureq::http::Response<ureq::Body>, SyncError> {
521 self.request_with(if large { self.limits.large } else { self.limits.small }, method, url, headers, body)
522 }
523
524 fn request_with(&self, whole: Duration, method: &str, url: &str, headers: &[(&str, &str)], body: Vec<u8>) -> Result<ureq::http::Response<ureq::Body>, SyncError> {
528 dav::allowed(url)?;
529 if self.started.elapsed() > self.limits.pull {
530 return Err(SyncError::Network("out of time".into()));
531 }
532 let mut request = ureq::http::Request::builder().method(method).uri(url).header("Authorization", &self.authorization).header("User-Agent", "Sioul");
533 for (name, value) in headers {
534 request = request.header(*name, *value);
535 }
536 let request = request.body(body).map_err(|e| SyncError::Server(e.to_string()))?;
537 let request = self.agent.configure_request(request).timeout_connect(Some(self.limits.wait)).timeout_global(Some(whole)).build();
538 crate::stalls::ask(self.limits.wait, false);
539 let response = self.agent.run(request).map_err(|e| failed(&e))?;
540 if response.status().as_u16() == 401 {
541 return Err(SyncError::Login(format!("{method}: 401")));
542 }
543 Ok(response)
544 }
545
546 pub(crate) fn list(&self, url: &str) -> Result<Listing, SyncError> {
548 let mut response = self.request(false, "PROPFIND", url, &[("Depth", "1"), ("Content-Type", XML)], LIST.as_bytes().to_vec())?;
549 match response.status().as_u16() {
550 207 => {}
551 404 => return Err(SyncError::NotFound(String::new())),
552 status => return Err(SyncError::Server(format!("PROPFIND: {status}"))),
553 }
554 let body = response.body_mut().with_config().limit(LISTING).read_to_string().map_err(|e| failed(&e))?;
555 let (responses, _) = dav::multistatus(&body)?;
556 let sums = checksums(&body, url);
557 let me = dav::path_key(url);
558 let mut listing = Listing::default();
559 for response in responses {
560 let path = dav::path_key(&dav::absolute(url, &response.href));
561 let etag = response.text(DAV, "getetag").unwrap_or_default();
562 if path == me {
563 listing.etag = etag;
564 continue;
565 }
566 let Some(name) = path.strip_prefix(&me).and_then(|rest| rest.strip_prefix('/')).filter(|n| !n.is_empty() && !n.contains('/')) else { continue };
567 let size = response.text(DAV, "getcontentlength").and_then(|t| t.parse().ok()).unwrap_or(0);
568 listing.items.insert(name.to_string(), Item { dir: response.is(DAV, "collection"), etag, size, checksum: sums.get(&path).cloned().unwrap_or_default() });
569 }
570 Ok(listing)
571 }
572
573 pub(crate) fn stat(&self, url: &str) -> Result<Option<Item>, SyncError> {
576 let mut response = self.request(false, "PROPFIND", url, &[("Depth", "0"), ("Content-Type", XML)], LIST.as_bytes().to_vec())?;
577 match response.status().as_u16() {
578 207 => {}
579 404 => return Ok(None),
580 status => return Err(SyncError::Server(format!("PROPFIND: {status}"))),
581 }
582 let body = response.body_mut().with_config().limit(LISTING).read_to_string().map_err(|e| failed(&e))?;
583 let (responses, _) = dav::multistatus(&body)?;
584 let sums = checksums(&body, url);
585 Ok(responses.first().map(|r| Item {
586 dir: r.is(DAV, "collection"),
587 etag: r.text(DAV, "getetag").unwrap_or_default(),
588 size: r.text(DAV, "getcontentlength").and_then(|t| t.parse().ok()).unwrap_or(0),
589 checksum: sums.get(&dav::path_key(&dav::absolute(url, &r.href))).cloned().unwrap_or_default(),
590 }))
591 }
592
593 pub(crate) fn put_file(&self, url: &str, body: Vec<u8>, seen: Option<&str>, modified_s: u64, sha1: &str) -> Result<(u16, Option<String>), PutFailed> {
605 let total = body.len() as u64;
606 let started = Instant::now();
607 let gave_up = |error: SyncError, kind: &'static str, sent: u64| PutFailed { words: error.to_string(), error, kind, seconds: started.elapsed().as_millis().div_ceil(1000) as u64, sent, total };
608 dav::allowed(url).map_err(|e| gave_up(e, "refused", 0))?;
609 if self.started.elapsed() > self.limits.pull {
610 return Err(gave_up(SyncError::Network("out of time".into()), "out-of-time", 0));
611 }
612 let precondition = match seen.filter(|etag| !etag.is_empty()) {
613 Some(etag) => ("If-Match", etag),
614 None => ("If-None-Match", "*"),
615 };
616 let whole = self.limits.large + Duration::from_secs(total / self.limits.floor.max(1));
617 let moved = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
618 let reader = Counted { inner: std::io::Cursor::new(body), moved: std::sync::Arc::clone(&moved) };
619 let request = ureq::http::Request::builder()
620 .method("PUT")
621 .uri(url)
622 .header("Authorization", &self.authorization)
623 .header("User-Agent", "Sioul")
624 .header(precondition.0, precondition.1)
625 .header("Content-Type", "application/octet-stream")
626 .header("Content-Length", total.to_string())
627 .header("X-OC-Mtime", modified_s.to_string())
628 .header("OC-Checksum", format!("SHA1:{sha1}"))
629 .body(ureq::SendBody::from_owned_reader(reader))
630 .map_err(|e| gave_up(SyncError::Server(e.to_string()), "server", 0))?;
631 let request = self.agent.configure_request(request).timeout_connect(Some(self.limits.wait)).timeout_global(Some(whole)).build();
635 crate::stalls::ask(self.limits.stall, true);
636 match self.agent.run(request) {
637 Ok(response) if response.status().as_u16() == 401 => Err(gave_up(SyncError::Login("PUT: 401".into()), "login", moved.load(std::sync::atomic::Ordering::Relaxed))),
638 Ok(response) => Ok((response.status().as_u16(), header(&response, "ETag").or_else(|| header(&response, "OC-ETag")))),
639 Err(e) => {
640 let sent = moved.load(std::sync::atomic::Ordering::Relaxed);
641 let kind = match &e {
642 ureq::Error::Timeout(ureq::Timeout::Connect | ureq::Timeout::Resolve) | ureq::Error::ConnectionFailed | ureq::Error::HostNotFound => "unreachable",
643 ureq::Error::Timeout(_) | ureq::Error::Io(_) if started.elapsed() + Duration::from_secs(2) < whole && sent > 0 => "stalled",
645 ureq::Error::Timeout(_) => "timeout",
646 ureq::Error::Tls(_) => "tls",
647 _ if format!("{e:?}").starts_with("Rustls") => "tls",
648 _ => "network",
649 };
650 Err(gave_up(failed(&e), kind, sent))
651 }
652 }
653 }
654
655 pub(crate) fn small(&self, url: &str) -> Result<Option<(Vec<u8>, Option<String>)>, SyncError> {
659 self.small_up_to(url, crate::share::SMALL_FILE)
660 }
661
662 pub(crate) fn small_up_to(&self, url: &str, cap: u64) -> Result<Option<(Vec<u8>, Option<String>)>, SyncError> {
664 let mut response = self.request(false, "GET", url, &[("Accept-Encoding", "identity")], Vec::new())?;
665 match response.status().as_u16() {
666 200 => {}
667 404 | 410 => return Ok(None),
668 status => return Err(SyncError::Server(format!("GET: {status}"))),
669 }
670 let etag = header(&response, "ETag");
671 let length = response.body().content_length();
672 let chunked = header(&response, "Transfer-Encoding").is_some_and(|t| t.to_ascii_lowercase().contains("chunked"));
673 if length.is_some_and(|l| l > cap) {
674 return Ok(None);
675 }
676 let bytes = match response.body_mut().with_config().limit(cap).read_to_vec() {
677 Ok(bytes) => bytes,
678 Err(ureq::Error::BodyExceedsLimit(_)) => return Ok(None),
679 Err(e) => return Err(failed(&e)),
680 };
681 if !(length == Some(bytes.len() as u64) || (length.is_none() && chunked)) {
682 return Err(SyncError::Network("GET: cut".into()));
683 }
684 Ok(Some((bytes, etag)))
685 }
686
687 fn tail(&self, url: &str, from: u64) -> Result<Option<Got>, SyncError> {
691 let range = format!("bytes={from}-");
692 let mut headers = vec![("Accept-Encoding", "identity")];
693 if from > 0 {
694 headers.push(("Range", range.as_str()));
695 }
696 let mut response = self.request(true, "GET", url, &headers, Vec::new())?;
697 let start = match response.status().as_u16() {
698 200 => 0,
699 206 => {
700 let said = header(&response, "Content-Range").unwrap_or_default();
702 said.trim().strip_prefix("bytes").and_then(|r| r.trim().split('-').next()).and_then(|s| s.trim().parse().ok()).ok_or_else(|| SyncError::Server(format!("GET: Content-Range {said}")))?
703 }
704 404 | 410 | 416 => return Ok(None),
705 status => return Err(SyncError::Server(format!("GET: {status}"))),
706 };
707 let length = response.body().content_length();
708 let mut reader = response.body_mut().with_config().limit(LARGEST).reader();
709 let mut bytes = Vec::new();
710 let mut piece = vec![0u8; 64 << 10];
711 let mut error = loop {
712 match reader.read(&mut piece) {
713 Ok(0) => break None,
714 Ok(n) => bytes.extend_from_slice(&piece[..n]),
715 Err(e) => break Some(SyncError::Network(e.to_string())),
716 }
717 };
718 if error.is_none() && length.is_some_and(|l| l != bytes.len() as u64) {
719 error = Some(SyncError::Network("GET: cut".into()));
720 }
721 Ok(Some(Got { start, bytes, error }))
722 }
723
724 pub(crate) fn put(&self, url: &str, body: Vec<u8>, seen: Option<&str>) -> Result<(u16, Option<String>), SyncError> {
728 let precondition = match seen {
729 Some(etag) => ("If-Match", etag),
730 None => ("If-None-Match", "*"),
731 };
732 let response = self.request(true, "PUT", url, &[precondition, ("Content-Type", "application/octet-stream")], body)?;
733 Ok((response.status().as_u16(), header(&response, "ETag").or_else(|| header(&response, "OC-ETag"))))
734 }
735
736 pub(crate) fn mkcol(&self, url: &str) -> Result<u16, SyncError> {
738 Ok(self.request(false, "MKCOL", url, &[], Vec::new())?.status().as_u16())
739 }
740
741 pub(crate) fn delete(&self, url: &str, seen: Option<&str>) -> Result<u16, SyncError> {
743 let headers: Vec<(&str, &str)> = seen.map(|etag| vec![("If-Match", etag)]).unwrap_or_default();
744 Ok(self.request(false, "DELETE", url, &headers, Vec::new())?.status().as_u16())
745 }
746
747 pub(crate) fn etag(&self, url: &str) -> Result<Option<String>, SyncError> {
749 let mut response = self.request(false, "PROPFIND", url, &[("Depth", "0"), ("Content-Type", XML)], LIST.as_bytes().to_vec())?;
750 match response.status().as_u16() {
751 207 => {}
752 404 => return Ok(None),
753 status => return Err(SyncError::Server(format!("PROPFIND: {status}"))),
754 }
755 let body = response.body_mut().with_config().limit(LISTING).read_to_string().map_err(|e| failed(&e))?;
756 let (responses, _) = dav::multistatus(&body)?;
757 Ok(responses.first().map(|r| r.text(DAV, "getetag").unwrap_or_default()))
758 }
759
760 pub(crate) fn download(&self, url: &str, path: &Path, limit: u64) -> Result<Option<u64>, SyncError> {
765 let mut response = self.request_with(self.limits.blob, "GET", url, &[("Accept-Encoding", "identity")], Vec::new())?;
766 match response.status().as_u16() {
767 200 => {}
768 404 | 410 => return Ok(None),
769 status => return Err(SyncError::Server(format!("GET: {status}"))),
770 }
771 let length = response.body().content_length();
772 let chunked = header(&response, "Transfer-Encoding").is_some_and(|t| t.to_ascii_lowercase().contains("chunked"));
773 if length.is_some_and(|l| l > limit) {
774 return Ok(None);
775 }
776 if length.is_none() && !chunked {
777 return Err(SyncError::Network("GET: its end cannot be told".into()));
778 }
779 let temporary = crate::share::temporary(path);
780 let disk = |e: std::io::Error| SyncError::Disk(format!("{}: {e}", path.display()));
781 let written = (|| -> Result<u64, SyncError> {
782 if let Some(parent) = path.parent() {
783 crate::share::private_dirs(parent).map_err(disk)?;
784 }
785 let mut out = crate::share::new_private(&temporary).map_err(disk)?;
786 let mut reader = response.body_mut().with_config().limit(limit).reader();
787 let copied = std::io::copy(&mut reader, &mut out).map_err(|e| SyncError::Network(e.to_string()))?;
788 if length.is_some_and(|l| l != copied) {
789 return Err(SyncError::Network("GET: cut".into()));
790 }
791 out.sync_all().map_err(disk)?;
792 std::fs::rename(&temporary, path).map_err(disk)?;
793 Ok(copied)
794 })();
795 if written.is_err() {
796 let _ = std::fs::remove_file(&temporary);
797 }
798 written.map(Some)
799 }
800}
801
802pub(crate) fn encode_path(path: &str) -> String {
804 path.split('/')
805 .map(|name| name.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-._~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") }).collect::<String>())
806 .collect::<Vec<_>>()
807 .join("/")
808}
809
810fn checksums(body: &str, url: &str) -> BTreeMap<String, String> {
813 let Ok(doc) = roxmltree::Document::parse(body) else { return BTreeMap::new() };
814 doc.descendants()
815 .filter(|n| n.has_tag_name((DAV, "response")))
816 .filter_map(|response| {
817 let href = response.children().find(|c| c.has_tag_name((DAV, "href")))?.text()?.trim().to_string();
818 let sums = response.descendants().filter(|n| n.has_tag_name((OC, "checksum"))).filter_map(|n| n.text()).map(str::trim).filter(|t| !t.is_empty()).collect::<Vec<_>>().join(" ");
819 (!sums.is_empty()).then(|| (dav::path_key(&dav::absolute(url, &href)), sums))
820 })
821 .collect()
822}
823
824fn sha1_hex(bytes: &[u8]) -> String {
826 use sha1::Digest;
827 sha1::Sha1::digest(bytes).iter().map(|b| format!("{b:02x}")).collect()
828}
829
830fn sha1_of(checksums: &str) -> Option<String> {
832 checksums.split_whitespace().find_map(|sum| sum.split_once(':').filter(|(kind, _)| kind.eq_ignore_ascii_case("sha1")).map(|(_, hex)| hex.to_ascii_lowercase()))
833}
834
835fn code(error: &SyncError, host: &str) -> String {
837 let kind = match error {
838 SyncError::Login(_) | SyncError::AppPassword(_) => "login",
839 SyncError::Tls(_) => "tls",
840 SyncError::Network(_) => "network",
841 SyncError::NoPassword => "no-password",
842 SyncError::Disk(_) => "disk",
843 SyncError::Server(d) if d == "quota" => "quota",
844 _ => "server",
845 };
846 format!("{kind}:{host}")
847}
848
849pub(crate) fn files_root(server: &Server, login: &Login) -> Result<Option<String>, SyncError> {
856 let Some(at) = login.url.find("/remote.php/") else { return Ok(None) };
857 let base = format!("{}/remote.php/dav/", &login.url[..at]);
858 let mut response = server.request(false, "PROPFIND", &base, &[("Depth", "0"), ("Content-Type", XML)], PRINCIPAL.as_bytes().to_vec())?;
859 if response.status().as_u16() != 207 {
860 return Ok(None);
861 }
862 let body = response.body_mut().with_config().limit(LISTING).read_to_string().map_err(|e| failed(&e))?;
863 let (responses, _) = dav::multistatus(&body)?;
864 let principal = responses.iter().find_map(|r| r.prop(DAV, "current-user-principal").and_then(|p| p.hrefs.first().cloned()));
865 let root = principal.and_then(|href| {
866 let at = href.find("/principals/users/")?;
867 let user = href[at + "/principals/users/".len()..].trim_end_matches('/');
868 (!user.is_empty() && !user.contains('/')).then(|| dav::absolute(&base, &format!("{}/files/{user}/", &href[..at])))
869 });
870 match root {
871 Some(root) if dav::origin(&root).is_none() || dav::origin(&root) != dav::origin(&base) => Err(SyncError::Elsewhere(root)),
872 root => Ok(root),
873 }
874}
875
876pub fn places(folder: &Path, nextcloud: &[String]) -> Vec<String> {
885 let text = folder.to_string_lossy().replace('\\', "/");
886 let text = text.trim_end_matches('/');
887 let mut out: Vec<String> = Vec::new();
888 for storage in ["/storage/emulated/0", "/sdcard", "/storage/self/primary", "/mnt/sdcard", "/mnt/user/0/primary"] {
889 let Some(rest) = text.strip_prefix(storage).and_then(|r| r.strip_prefix('/')) else { continue };
890 match rest.strip_prefix("Android/media/com.nextcloud.client/nextcloud/").and_then(|r| r.split_once('/')) {
892 Some((_, inside)) => out.push(inside.to_string()),
893 None if !rest.starts_with("Android/") => out.push(rest.to_string()),
894 None => {}
895 }
896 }
897 for config in nextcloud {
898 out.extend(nextcloud_places(text, config));
899 }
900 let names: Vec<&str> = text.rsplit('/').filter(|n| !n.is_empty()).take(3).collect();
901 for n in 1..=names.len() {
902 out.push(names[..n].iter().rev().copied().collect::<Vec<_>>().join("/"));
903 }
904 let mut seen = BTreeSet::new();
905 out.into_iter().map(|p| p.trim_matches('/').to_string()).filter(|p| !p.is_empty() && seen.insert(p.clone())).collect()
906}
907
908fn nextcloud_places(folder: &str, config: &str) -> Vec<String> {
912 let mut local: BTreeMap<&str, &str> = BTreeMap::new();
913 let mut target: BTreeMap<&str, &str> = BTreeMap::new();
914 for line in config.lines() {
915 let Some((key, value)) = line.split_once('=') else { continue };
916 if let Some(prefix) = key.trim().strip_suffix("localPath") {
917 local.insert(prefix, value.trim());
918 } else if let Some(prefix) = key.trim().strip_suffix("targetPath") {
919 target.insert(prefix, value.trim());
920 }
921 }
922 local
923 .into_iter()
924 .filter_map(|(prefix, root)| {
925 let root = root.trim_end_matches('/');
926 let rest = folder.strip_prefix(root).filter(|r| !root.is_empty() && (r.is_empty() || r.starts_with('/')))?;
927 Some(format!("{}/{}", target.get(prefix)?.trim_matches('/'), rest.trim_matches('/')))
928 })
929 .collect()
930}
931
932pub fn find(memory: &Path, folder: &Path, logins: &[Login], places: &[String], now: i64) -> State {
938 find_with(memory, folder, logins, places, now, LIMITS)
939}
940
941pub(crate) fn find_with(memory: &Path, folder: &Path, logins: &[Login], places: &[String], now: i64, limits: Limits) -> State {
942 let mut state = State::load(memory);
943 if state.mode == MIRROR {
945 return state;
946 }
947 if state.folder != shown(folder) {
948 let _ = std::fs::remove_dir_all(cache_of(memory));
950 state = State { on: state.on, given: state.given.clone(), asked: state.asked, send: state.send, folder: shown(folder), ..State::default() };
951 }
952 state.looked = now;
953 state.confirmed = 0;
954 let Some(seal) = seal_of(folder) else {
955 state.said = "no-seal".into();
956 let _ = state.save(memory);
957 return state;
958 };
959 let given = state.given.trim().to_string();
960 let mut said: Vec<String> = Vec::new();
961 let mut looked_on: Vec<String> = Vec::new();
962 for login in logins {
963 let host = host_of(&login.url);
964 if given.contains("://") && host_of(&given) != host {
966 continue;
967 }
968 let server = match Server::new(login, limits) {
969 Ok(server) => server,
970 Err(e) => {
971 said.push(code(&e, &login.account));
972 continue;
973 }
974 };
975 let urls: Vec<String> = if given.contains("://") {
976 vec![format!("{}/", given.trim_end_matches('/'))]
977 } else {
978 let root = match files_root(&server, login) {
979 Ok(Some(root)) => root,
980 Ok(None) => continue,
981 Err(e) => {
982 said.push(code(&e, &host));
983 continue;
984 }
985 };
986 let wanted: Vec<String> = if given.is_empty() { places.to_vec() } else { vec![given.trim_matches('/').to_string()] };
987 wanted.iter().filter(|p| !p.is_empty()).map(|p| format!("{root}{}/", encode_path(p))).collect()
988 };
989 if !looked_on.contains(&host) {
990 looked_on.push(host.clone());
991 }
992 for url in urls {
993 match server.small(&format!("{url}seal.toml")) {
994 Ok(Some((bytes, etag))) if bytes == seal => {
995 confirm(&mut state, memory, login, &url, &seal, etag, now);
996 let _ = state.save(memory);
997 return state;
998 }
999 Ok(Some(_)) => said.push(format!("seal-differs:{host}")),
1000 Ok(None) => {}
1001 Err(e) => {
1002 said.push(code(&e, &host));
1003 if e.is_lasting() || matches!(e, SyncError::Network(_)) {
1005 break;
1006 }
1007 }
1008 }
1009 }
1010 }
1011 let rank = |code: &String| ["seal-differs:", "login:", "no-password:", "tls:", "network:", "server:", "disk:"].iter().position(|p| code.starts_with(p)).unwrap_or(9);
1013 said.sort_by_key(rank);
1014 state.said = match said.into_iter().next() {
1015 Some(why) => why,
1016 None if looked_on.is_empty() && given.contains("://") => format!("no-account-for:{}", host_of(&given)),
1017 None if looked_on.is_empty() => "no-account".into(),
1018 None => format!("not-found:{}", looked_on.join(", ")),
1019 };
1020 let _ = state.save(memory);
1021 state
1022}
1023
1024fn confirm(state: &mut State, memory: &Path, login: &Login, url: &str, seal: &[u8], etag: Option<String>, now: i64) {
1027 let cache = cache_of(memory);
1028 if seal_of(&cache).as_deref() != Some(seal) {
1029 let _ = std::fs::remove_dir_all(&cache);
1030 state.files.clear();
1031 state.folders.clear();
1032 }
1033 if state.url != url {
1034 state.files.clear();
1035 state.folders.clear();
1036 }
1037 let _ = crate::share::write_atomically(&cache.join("seal.toml"), seal);
1038 state.account = login.account.clone();
1039 state.url = url.to_string();
1040 state.confirmed = now;
1041 state.sealed = now;
1042 state.said.clear();
1043 if let Some(etag) = etag {
1045 state.files.insert("seal.toml".into(), Fetched { etag, size: seal.len() as u64 });
1046 }
1047}
1048
1049#[derive(Debug, Clone, Copy, Default)]
1053pub struct Pace {
1054 pub phone: bool,
1056 pub shown: bool,
1058 pub in_use: bool,
1060 pub urgent: bool,
1062 pub step: i64,
1064}
1065
1066pub fn due(state: &State, now: i64, pace: Pace) -> bool {
1076 let since = now - state.tried;
1077 if state.tried < state.confirmed || since < 0 {
1079 return true;
1080 }
1081 if pace.urgent {
1082 return since >= 20;
1083 }
1084 if !pace.phone || pace.shown || pace.in_use {
1085 return since >= 60;
1086 }
1087 since >= pace.step - 30
1088}
1089
1090fn round_cap() -> u64 {
1092 #[cfg(test)]
1093 if let Some(cap) = tests::ROUND_CAP.with(std::cell::Cell::get) {
1094 return cap;
1095 }
1096 LARGEST
1097}
1098
1099#[derive(Debug, Clone, Copy)]
1101pub struct AlarmTimes {
1102 pub alone: Duration,
1104 pub most: Duration,
1106}
1107
1108pub const ALARM_TIMES: AlarmTimes = AlarmTimes { alone: Duration::from_secs(5), most: Duration::from_secs(25) };
1110
1111#[derive(Debug, Clone, PartialEq, Eq)]
1113pub enum AlarmNews {
1114 Pulled,
1116 SyncApp { heard: BTreeSet<String> },
1124}
1125
1126#[allow(clippy::too_many_arguments)]
1140pub fn alarm_news(memory: &Path, folder: &Path, own: &str, key: [u8; 32], login: Option<Login>, lock: &'static Mutex<()>, since: i64, clock: std::sync::Arc<dyn Fn() -> i64 + Send + Sync>, ask: &mut dyn FnMut() -> Duration) -> AlarmNews {
1141 alarm_news_with(memory, folder, own, key, login, lock, since, ALARM_TIMES, clock, ask, LIMITS)
1142}
1143
1144#[allow(clippy::too_many_arguments)]
1145pub(crate) fn alarm_news_with(memory: &Path, folder: &Path, own: &str, key: [u8; 32], login: Option<Login>, lock: &'static Mutex<()>, since: i64, times: AlarmTimes, clock: std::sync::Arc<dyn Fn() -> i64 + Send + Sync>, ask: &mut dyn FnMut() -> Duration, limits: Limits) -> AlarmNews {
1146 let begun = Instant::now();
1147 let until = begun + times.most;
1148 let before = said_by_each(folder, &key, own);
1150 let mut pulling = {
1151 let (memory, folder, own, clock) = (memory.to_path_buf(), folder.to_path_buf(), own.to_string(), std::sync::Arc::clone(&clock));
1152 Some(std::thread::spawn(move || pull_since_with(&memory, &folder, &own, login.as_ref(), lock, since, begun, until, &*clock, limits)))
1153 };
1154 let mut went = None;
1155 let mut asked: Option<Instant> = None;
1156 loop {
1157 if went.is_none() && pulling.as_ref().is_some_and(std::thread::JoinHandle::is_finished) {
1158 went = pulling.take().map(|pull| pull.join().unwrap_or(false));
1159 }
1160 if went == Some(true) {
1161 return AlarmNews::Pulled;
1162 }
1163 let now = Instant::now();
1164 if asked.is_none() && (went == Some(false) || now >= begun + times.alone) {
1166 asked = Some(now + ask());
1167 }
1168 if asked.is_some_and(|end| now >= end) || now >= until {
1169 if through_since(memory, folder, since, begun, clock()) {
1171 return AlarmNews::Pulled;
1172 }
1173 return AlarmNews::SyncApp { heard: heard_since(&before, &said_by_each(folder, &key, own)) };
1174 }
1175 std::thread::sleep(Duration::from_millis(50));
1176 }
1177}
1178
1179fn through_since(memory: &Path, folder: &Path, since: i64, begun: Instant, now: i64) -> bool {
1185 let state = State::load(memory);
1186 if !state.pulls_well(folder) {
1187 return false;
1188 }
1189 match PULLED.lock().ok().and_then(|all| all.get(memory).copied()) {
1190 Some((at, last)) if last == state.last => at >= begun,
1191 _ => state.went_through_since(folder, since, now),
1192 }
1193}
1194
1195#[allow(clippy::too_many_arguments)]
1200pub(crate) fn pull_since_with(memory: &Path, folder: &Path, own: &str, login: Option<&Login>, lock: &Mutex<()>, since: i64, begun: Instant, until: Instant, clock: &dyn Fn() -> i64, limits: Limits) -> bool {
1201 let through = || through_since(memory, folder, since, begun, clock());
1202 let _one = loop {
1203 match lock.try_lock() {
1204 Ok(held) => break held,
1205 Err(std::sync::TryLockError::Poisoned(held)) => break held.into_inner(),
1206 Err(std::sync::TryLockError::WouldBlock) if Instant::now() < until => std::thread::sleep(Duration::from_millis(50)),
1207 Err(std::sync::TryLockError::WouldBlock) => return through(),
1208 }
1209 };
1210 if through() {
1211 return true;
1212 }
1213 let state = State::load(memory);
1217 if (!state.said.is_empty() && (since..=clock()).contains(&state.tried)) || Instant::now() >= until {
1218 return false;
1219 }
1220 let Some(login) = login else { return false };
1221 let _ = pull_with(memory, folder, own, login, clock(), limits);
1222 through()
1223}
1224
1225type Said = BTreeMap<String, (Option<(i64, i64, i64, i64, i64)>, Option<(i64, i64)>)>;
1230
1231pub(crate) fn said_by_each(folder: &Path, key: &[u8; 32], own: &str) -> Said {
1232 let mut out: Said = BTreeMap::new();
1233 for entry in crate::devices::all(folder, key).0.into_iter().filter(|e| e.id != own) {
1234 out.entry(entry.id.clone()).or_default().0 = Some(crate::devices::written(&entry));
1235 }
1236 for claim in crate::lease::claims(folder, key, "health").into_iter().filter(|c| c.computer != own) {
1237 out.entry(claim.computer.clone()).or_default().1 = Some((claim.renewed, claim.until));
1238 }
1239 out
1240}
1241
1242fn heard_since(before: &Said, now: &Said) -> BTreeSet<String> {
1247 now.iter()
1248 .filter(|(id, (entry, claim))| {
1249 let (was_entry, was_claim) = before.get(*id).copied().unwrap_or_default();
1250 entry.is_some_and(|e| was_entry.is_none_or(|w| e > w)) || claim.is_some_and(|c| was_claim.is_none_or(|w| c > w))
1251 })
1252 .map(|(id, _)| id.clone())
1253 .collect()
1254}
1255
1256#[derive(Debug, Default, Clone, PartialEq, Eq)]
1258pub struct Pulled {
1259 pub listed: usize,
1261 pub fetched: usize,
1263 pub sent: usize,
1265 pub same: usize,
1267 pub longer: usize,
1270 pub differ: Vec<String>,
1273 pub paced: usize,
1275 pub held: usize,
1278 pub failure: Option<Failure>,
1280 pub bytes: u64,
1282 pub left: usize,
1284 pub problem: Option<String>,
1286}
1287
1288enum Stop {
1290 Failed(SyncError),
1291 Unconfirmed(String),
1292}
1293
1294impl From<SyncError> for Stop {
1295 fn from(e: SyncError) -> Stop {
1296 Stop::Failed(e)
1297 }
1298}
1299
1300pub fn pull(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64) -> Pulled {
1305 pull_with(memory, folder, own, login, now, LIMITS)
1306}
1307
1308pub(crate) fn pull_with(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64, limits: Limits) -> Pulled {
1309 let mut state = State::load(memory);
1310 let mut pulled = Pulled::default();
1311 if state.mode == MIRROR {
1313 pulled.problem = Some("mirror".into());
1314 return pulled;
1315 }
1316 if !state.confirmed_for(folder) || state.account != login.account {
1317 pulled.problem = Some("not-confirmed".into());
1318 return pulled;
1319 }
1320 let cache = cache_of(memory);
1321 if !state.fetching() {
1323 tidy_cache(&state, folder, memory, own);
1324 pulled.problem = Some("off".into());
1325 return pulled;
1326 }
1327 if !usable(folder, memory) {
1330 let _ = std::fs::remove_dir_all(&cache);
1331 state = State { on: state.on, given: state.given.clone(), asked: state.asked, send: state.send, folder: state.folder.clone(), said: "no-seal".into(), ..State::default() };
1332 let _ = state.save(memory);
1333 pulled.problem = Some(state.said);
1334 return pulled;
1335 }
1336 let loaded_left = state.left.clone();
1337 state.tried = now;
1338 let begun = Instant::now();
1339 let host = state.host();
1340 let outcome = Server::new(login, limits).map_err(Stop::Failed).and_then(|server| Puller { server: &server, folder, into: &cache, own, state: &mut state, pulled: &mut pulled, now, own_there: BTreeMap::new(), covered: BTreeSet::new(), lost: false, blobs: Vec::new(), blobs_listed: None }.run());
1341 match outcome {
1342 Ok(()) => {
1343 state.last = now;
1344 state.said.clear();
1345 }
1346 Err(Stop::Unconfirmed(why)) => {
1347 state.confirmed = 0;
1348 state.said = why;
1349 }
1350 Err(Stop::Failed(e)) => state.said = code(&e, &host),
1351 }
1352 tidy_cache(&state, folder, memory, own);
1353 if state.said.is_empty() {
1354 if let Ok(mut all) = PULLED.lock()
1356 && all.get(memory).is_none_or(|(at, _)| *at < begun)
1357 {
1358 all.insert(memory.to_path_buf(), (begun, state.last));
1359 }
1360 }
1361 let _ = state.save_pulled(memory, now + begun.elapsed().as_secs() as i64 + 1, &loaded_left);
1362 pulled.problem = (!state.said.is_empty()).then(|| state.said.clone());
1363 pulled
1364}
1365
1366static PULLED: Mutex<BTreeMap<PathBuf, (Instant, i64)>> = Mutex::new(BTreeMap::new());
1371
1372pub fn tidy(memory: &Path, folder: &Path, own: &str) {
1381 let state = State::load(memory);
1382 if state.folder == shown(folder) && state.mode != MIRROR {
1383 tidy_cache(&state, folder, memory, own);
1384 }
1385}
1386
1387fn tidy_cache(state: &State, folder: &Path, memory: &Path, own: &str) {
1388 let cache = &cache_of(memory);
1389 for dir in [cache.to_path_buf(), cache.join("devices"), cache.join("blobs")].into_iter().chain(std::fs::read_dir(cache.join("leases")).into_iter().flatten().filter_map(Result::ok).map(|e| e.path())) {
1391 crate::share::clean_leftovers(&dir);
1392 }
1393 for relative in cached_files(cache) {
1394 let (here, fetched) = (folder.join(&relative), cache.join(&relative));
1395 let name = relative.rsplit('/').next().unwrap_or_default();
1396 let gone = match kind_of(&relative) {
1397 None => continue,
1398 Some((_, id)) if id == own => true,
1399 Some((Kind::Round, _)) => std::fs::metadata(&here).is_ok_and(|h| std::fs::metadata(&fetched).is_ok_and(|f| h.len() >= f.len())),
1400 Some(_) => !state.fetching() || std::fs::read(&here).ok().is_some_and(|h| std::fs::read(&fetched).ok().is_some_and(|f| f == h)),
1401 };
1402 if gone && !name.starts_with('.') {
1403 let _ = std::fs::remove_file(&fetched);
1404 }
1405 }
1406 let wanted = crate::blobs::Wanted::load(memory);
1414 let mut aged: Vec<String> = Vec::new();
1415 for entry in std::fs::read_dir(cache.join("blobs")).into_iter().flatten().filter_map(Result::ok) {
1416 let name = entry.file_name().to_string_lossy().to_string();
1417 let Ok(meta) = entry.metadata() else { continue };
1418 if name.starts_with('.') || !meta.is_file() {
1419 continue;
1420 }
1421 let brought = std::fs::metadata(folder.join("blobs").join(&name)).is_ok_and(|here| here.len() == meta.len());
1422 let old = meta.modified().ok().and_then(|t| t.elapsed().ok()).is_some_and(|age| age.as_secs() >= WANTED_KEPT_DAYS * 86_400);
1423 if !state.fetching() || brought || old || !wanted.blobs.contains_key(&name) {
1424 let _ = std::fs::remove_file(entry.path());
1425 if old {
1426 aged.push(name);
1427 }
1428 }
1429 }
1430 if !aged.is_empty() {
1431 let _ = crate::blobs::Wanted::change(memory, |wanted| wanted.blobs.retain(|name, _| !aged.contains(name)));
1432 }
1433}
1434
1435#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1439pub(crate) enum Kind {
1440 Round,
1441 Seen,
1442 Device,
1443 Claim,
1444 Texts,
1446}
1447
1448pub(crate) fn is_id(id: &str) -> bool {
1449 id.len() == 36 && uuid::Uuid::parse_str(id).is_ok()
1450}
1451
1452pub(crate) fn kind_of(relative: &str) -> Option<(Kind, &str)> {
1453 let parts: Vec<&str> = relative.split('/').collect();
1454 match parts.as_slice() {
1455 [name] => {
1456 if let Some((id, round)) = name.strip_suffix(".jsonl").and_then(|stem| stem.rsplit_once('-')) {
1457 return (is_id(id) && round.parse::<u32>().is_ok()).then_some((Kind::Round, id));
1458 }
1459 name.strip_suffix(".toml").filter(|id| is_id(id)).map(|id| (Kind::Seen, id))
1460 }
1461 ["devices", name] => name.strip_suffix(".device").filter(|id| is_id(id)).map(|id| (Kind::Device, id)),
1462 ["leases", _, name] => name.strip_suffix(".lease").filter(|id| !id.is_empty() && !id.starts_with('.')).map(|id| (Kind::Claim, id)),
1463 ["texts", "send", name] => name.strip_suffix(".jsonl").filter(|id| is_id(id)).map(|id| (Kind::Texts, id)),
1464 _ => None,
1465 }
1466}
1467
1468fn cached_files(cache: &Path) -> Vec<String> {
1470 let names = |dir: &Path| -> Vec<String> { std::fs::read_dir(dir).into_iter().flatten().filter_map(Result::ok).filter(|e| e.file_type().is_ok_and(|t| t.is_file())).map(|e| e.file_name().to_string_lossy().to_string()).collect() };
1471 let mut out: Vec<String> = names(cache);
1472 out.extend(names(&cache.join("devices")).into_iter().map(|n| format!("devices/{n}")));
1473 for part in std::fs::read_dir(cache.join("leases")).into_iter().flatten().filter_map(Result::ok).filter(|e| e.file_type().is_ok_and(|t| t.is_dir())) {
1474 let part = part.file_name().to_string_lossy().to_string();
1475 out.extend(names(&cache.join("leases").join(&part)).into_iter().map(|n| format!("leases/{part}/{n}")));
1476 }
1477 out.extend(names(&cache.join("texts").join("send")).into_iter().map(|n| format!("texts/send/{n}")));
1478 out
1479}
1480
1481const WANTED_FILES: usize = 16;
1485const WANTED_BYTES: u64 = 16 << 20;
1487pub const SMALL_WANTED: u64 = 1 << 20;
1490const WANTED_KEPT_DAYS: u64 = 3;
1493const WANTED_AGAIN: i64 = 5 * 60;
1496
1497pub fn fetch_wanted(memory: &Path, folder: &Path, login: &Login, key: &[u8; 32], now: i64, frugal: bool) -> Pulled {
1512 fetch_wanted_with(memory, folder, login, key, now, frugal, LIMITS)
1513}
1514
1515pub fn wanted_due(memory: &Path, folder: &Path, now: i64) -> bool {
1518 crate::blobs::Wanted::path(memory).exists() && !wanted_now(&crate::blobs::Wanted::load(memory), folder, &cache_of(memory).join("blobs"), now).is_empty()
1519}
1520
1521fn wanted_now<'a>(wanted: &'a crate::blobs::Wanted, folder: &Path, into: &Path, now: i64) -> Vec<(&'a String, &'a crate::blobs::Want)> {
1524 let here = |dir: &Path, name: &str| std::fs::metadata(dir.join(name)).is_ok_and(|m| m.len() > 0);
1525 wanted.blobs.iter().filter(|(name, want)| want.again <= now && !here(&folder.join("blobs"), name) && !here(into, name)).collect()
1526}
1527
1528#[allow(clippy::too_many_arguments)]
1529pub(crate) fn fetch_wanted_with(memory: &Path, folder: &Path, login: &Login, key: &[u8; 32], now: i64, frugal: bool, limits: Limits) -> Pulled {
1530 let mut pulled = Pulled::default();
1531 let state = State::load(memory);
1532 if state.mode == MIRROR || !state.fetching() || !state.confirmed_for(folder) || state.account != login.account || !usable(folder, memory) {
1533 pulled.problem = Some("not-confirmed".into());
1534 return pulled;
1535 }
1536 let into = cache_of(memory).join("blobs");
1537 sioul_core::filelock::with_lock(&memory.with_file_name("wanted.step"), || {
1538 let wanted = crate::blobs::Wanted::load(memory);
1539 let mut due = wanted_now(&wanted, folder, &into, now);
1540 if due.is_empty() {
1541 return pulled;
1542 }
1543 due.sort_by_key(|(name, want)| (want.size, want.since, *name));
1544 let mut tried: BTreeMap<String, (u32, i64)> = BTreeMap::new();
1545 let mut large_at = wanted.large_at;
1546 let (mut count, mut budget) = (0, WANTED_BYTES);
1547 let outcome = Server::new(login, limits).and_then(|server| {
1548 for (name, want) in due {
1549 if count >= WANTED_FILES || (count > 0 && want.size > budget) {
1550 break;
1551 }
1552 let large = want.size >= SMALL_WANTED;
1553 if frugal && large && now - large_at < PACE {
1554 pulled.paced += 1;
1555 continue;
1556 }
1557 if frugal && large {
1558 large_at = now;
1559 }
1560 count += 1;
1561 budget = budget.saturating_sub(want.size);
1562 let path = into.join(name);
1563 let failed = |soon: bool| {
1564 let tries = want.tries + 1;
1565 (tries, now + if soon { backoff(tries).min(WANTED_AGAIN) } else { backoff(tries) })
1566 };
1567 match server.download(&format!("{}{}", state.url, encode_path(&format!("blobs/{name}"))), &path, BLOB_LIMIT) {
1568 Ok(Some(size)) => match crate::blobs::check(&path, key, &want.hash) {
1569 Ok(()) => {
1570 pulled.fetched += 1;
1571 pulled.bytes += size;
1572 tried.insert(name.clone(), (0, 0));
1573 }
1574 Err(fault) => {
1576 let _ = std::fs::remove_file(&path);
1577 pulled.held += 1;
1578 tried.insert(name.clone(), failed(!matches!(fault, crate::blobs::Fault::Broken)));
1579 }
1580 },
1581 Ok(None) => {
1583 tried.insert(name.clone(), failed(true));
1584 }
1585 Err(e) => {
1586 tried.insert(name.clone(), failed(true));
1587 return Err(e);
1588 }
1589 }
1590 }
1591 Ok(())
1592 });
1593 if let Err(e) = outcome {
1594 pulled.problem = Some(code(&e, &state.host()));
1595 }
1596 let _ = crate::blobs::Wanted::change(memory, |wanted| {
1597 for (name, (tries, again)) in tried {
1598 if let Some(want) = wanted.blobs.get_mut(&name) {
1599 (want.tries, want.again) = (tries, again);
1600 }
1601 }
1602 wanted.large_at = wanted.large_at.max(large_at);
1603 });
1604 pulled
1605 })
1606}
1607
1608struct Puller<'a> {
1610 server: &'a Server,
1611 folder: &'a Path,
1613 into: &'a Path,
1614 own: &'a str,
1615 state: &'a mut State,
1616 pulled: &'a mut Pulled,
1617 now: i64,
1618 own_there: BTreeMap<String, String>,
1620 covered: BTreeSet<String>,
1622 lost: bool,
1624 blobs: Vec<(String, Item)>,
1627 blobs_listed: Option<String>,
1629}
1630
1631impl Puller<'_> {
1632 fn url(&self, relative: &str) -> String {
1633 format!("{}{}", self.state.url, encode_path(relative))
1634 }
1635
1636 fn unchanged(&self, relative: &str, etag: &str) -> bool {
1638 !self.lost && !etag.is_empty() && self.state.folders.get(relative).is_some_and(|l| l.etag == etag && self.now - l.at < LIST_AGAIN)
1639 }
1640
1641 fn looked(&mut self, relative: &str, etag: &str) {
1643 self.state.folders.insert(relative.to_string(), Looked { etag: etag.to_string(), at: self.now });
1644 }
1645
1646 fn run(&mut self) -> Result<(), Stop> {
1647 let root = self.server.list(&self.state.url).map_err(|e| match e {
1648 SyncError::NotFound(_) => Stop::Unconfirmed(format!("seal-gone:{}", self.state.host())),
1649 e => Stop::Failed(e),
1650 })?;
1651 self.pulled.listed += root.items.len();
1652 self.seal(&root)?;
1654 self.lost = self.mirror() && own_latest(self.folder, self.own).is_none();
1657 if self.unchanged("", &root.etag) {
1658 let (folder, into) = (self.folder, self.into);
1660 let here = |name: &str| [folder.join(name), into.join(name)].iter().filter_map(|path| std::fs::metadata(path).ok()).map(|m| m.len()).max().unwrap_or(0);
1661 self.state.left.retain(|name, size| here(name) < *size);
1662 return Ok(());
1663 }
1664 self.state.left.clear();
1666 self.own_listed("", &root);
1667 for (name, item) in root.items.iter().filter(|(_, item)| !item.dir) {
1668 match kind_of(name) {
1669 Some((Kind::Round, id)) if id != self.own => self.round(name, item)?,
1670 Some((Kind::Seen, id)) if id != self.own => self.small(name, item)?,
1671 Some(_) if self.mirror() => self.restore(name, item)?,
1672 _ => {}
1673 }
1674 }
1675 if let Some(devices) = root.items.get("devices").filter(|item| item.dir)
1676 && !self.unchanged("devices/", &devices.etag)
1677 {
1678 let listing = self.server.list(&self.url("devices/"))?;
1679 self.pulled.listed += listing.items.len();
1680 self.own_listed("devices/", &listing);
1681 if let Some(item) = listing.items.get(&format!("{}.device", self.own)).filter(|item| !item.dir) {
1683 self.state.own_entry = Some((item.size, self.now));
1684 }
1685 for (name, item) in listing.items.iter().filter(|(_, item)| !item.dir) {
1686 let relative = format!("devices/{name}");
1687 match kind_of(&relative) {
1688 Some((_, id)) if id != self.own => self.small(&relative, item)?,
1689 Some(_) if self.mirror() => self.restore(&relative, item)?,
1690 _ => {}
1691 }
1692 }
1693 self.forget_gone("devices/", &listing);
1694 self.looked("devices/", &listing.etag);
1695 }
1696 if let Some(leases) = root.items.get("leases").filter(|item| item.dir)
1697 && !self.unchanged("leases/", &leases.etag)
1698 {
1699 let parts = self.server.list(&self.url("leases/"))?;
1700 self.pulled.listed += parts.items.len();
1701 for (part, item) in parts.items.iter().filter(|(_, item)| item.dir) {
1702 let dir = format!("leases/{part}/");
1703 if self.unchanged(&dir, &item.etag) {
1704 continue;
1705 }
1706 let listing = self.server.list(&self.url(&dir))?;
1707 self.pulled.listed += listing.items.len();
1708 self.own_listed(&dir, &listing);
1709 for (name, item) in listing.items.iter().filter(|(_, item)| !item.dir) {
1710 let relative = format!("{dir}{name}");
1711 match kind_of(&relative) {
1712 Some((_, id)) if id != self.own => self.small(&relative, item)?,
1713 Some(_) if self.mirror() => self.restore(&relative, item)?,
1714 _ => {}
1715 }
1716 }
1717 self.forget_gone(&dir, &listing);
1718 self.looked(&dir, &listing.etag);
1719 }
1720 self.looked("leases/", &parts.etag);
1721 }
1722 if let Some(texts) = root.items.get("texts").filter(|item| item.dir)
1725 && !self.unchanged("texts/", &texts.etag)
1726 {
1727 match self.server.list(&self.url("texts/send/")) {
1728 Ok(listing) => {
1729 self.pulled.listed += listing.items.len();
1730 self.own_listed("texts/send/", &listing);
1731 for (name, item) in listing.items.iter().filter(|(_, item)| !item.dir) {
1732 let relative = format!("texts/send/{name}");
1733 if let Some((Kind::Texts, id)) = kind_of(&relative)
1734 && id != self.own
1735 {
1736 self.small(&relative, item)?;
1737 }
1738 }
1739 self.forget_gone("texts/send/", &listing);
1740 self.looked("texts/", &texts.etag);
1741 }
1742 Err(SyncError::NotFound(_)) => self.looked("texts/", &texts.etag),
1743 Err(e) => return Err(Stop::Failed(e)),
1744 }
1745 }
1746 if self.mirror()
1748 && let Some(blobs) = root.items.get("blobs").filter(|item| item.dir)
1749 && !self.unchanged("blobs/", &blobs.etag)
1750 {
1751 let listing = self.server.list(&self.url("blobs/"))?;
1752 self.pulled.listed += listing.items.len();
1753 self.own_listed("blobs/", &listing);
1754 self.blobs = listing.items.iter().filter(|(name, item)| !item.dir && !name.starts_with('.')).map(|(name, item)| (format!("blobs/{name}"), item.clone())).collect();
1757 self.blobs_listed = Some(listing.etag.clone());
1758 }
1759 self.forget_gone("", &root);
1760 self.looked("", &root.etag);
1761 Ok(())
1762 }
1763
1764 fn mirror(&self) -> bool {
1766 self.into == self.folder
1767 }
1768
1769 fn own_listed(&mut self, dir: &str, listing: &Listing) {
1772 self.covered.insert(dir.to_string());
1773 for (name, item) in listing.items.iter().filter(|(_, item)| !item.dir) {
1774 let relative = format!("{dir}{name}");
1775 let own = match kind_of(&relative) {
1776 Some((_, id)) => id == self.own,
1777 None => dir == "blobs/" && self.state.sent.contains_key(&relative),
1778 };
1779 if own {
1780 self.own_there.insert(relative, item.etag.clone());
1781 }
1782 }
1783 }
1784
1785 fn seal(&mut self, root: &Listing) -> Result<(), Stop> {
1788 let host = self.state.host();
1789 let Some(item) = root.items.get("seal.toml").filter(|item| !item.dir) else { return Err(Stop::Unconfirmed(format!("seal-gone:{host}"))) };
1790 let known = Fetched { etag: item.etag.clone(), size: item.size };
1791 if !item.etag.is_empty() && self.state.files.get("seal.toml") == Some(&known) && self.now - self.state.sealed < SEAL_AGAIN {
1792 return Ok(());
1793 }
1794 let here = seal_of(self.folder).ok_or_else(|| Stop::Unconfirmed("no-seal".into()))?;
1795 match self.server.small(&self.url("seal.toml"))? {
1796 Some((bytes, _)) if bytes == here => {
1797 self.state.sealed = self.now;
1798 self.state.files.insert("seal.toml".into(), known);
1799 Ok(())
1800 }
1801 Some(_) => Err(Stop::Unconfirmed(format!("seal-differs:{host}"))),
1802 None => Err(Stop::Unconfirmed(format!("seal-gone:{host}"))),
1803 }
1804 }
1805
1806 fn forget_gone(&mut self, dir: &str, listing: &Listing) {
1810 let gone: Vec<String> = self.state.files.keys().filter(|path| path.strip_prefix(dir).is_some_and(|name| !name.contains('/') && !listing.items.contains_key(name))).cloned().collect();
1811 for path in gone {
1812 self.state.files.remove(&path);
1813 let round = matches!(kind_of(&path), Some((Kind::Round, _)));
1816 if path != "seal.toml" && (self.into != self.folder || round) {
1817 let _ = std::fs::remove_file(self.into.join(&path));
1818 }
1819 }
1820 }
1821
1822 fn small(&mut self, relative: &str, item: &Item) -> Result<(), Stop> {
1826 let known = Fetched { etag: item.etag.clone(), size: item.size };
1827 if !item.etag.is_empty() && self.state.files.get(relative) == Some(&known) && (!self.mirror() || self.folder.join(relative).exists()) {
1829 return Ok(());
1830 }
1831 let cap = if matches!(kind_of(relative), Some((Kind::Texts, _))) { TEXTS_LARGEST } else { crate::share::SMALL_FILE };
1832 if item.size > cap {
1833 return Ok(());
1834 }
1835 let Some((bytes, _)) = self.server.small_up_to(&self.url(relative), cap)? else {
1836 self.state.files.remove(relative);
1837 return Ok(());
1838 };
1839 self.pulled.bytes += bytes.len() as u64;
1840 let (here, into) = (self.folder.join(relative), self.into.join(relative));
1841 if std::fs::read(&here).ok().as_deref() == Some(&bytes[..]) {
1842 if into != here {
1844 let _ = std::fs::remove_file(&into);
1845 }
1846 } else {
1847 crate::share::write_atomically(&into, &bytes).map_err(|e| Stop::Failed(SyncError::Disk(e)))?;
1848 self.pulled.fetched += 1;
1849 }
1850 self.state.files.insert(relative.to_string(), known);
1851 Ok(())
1852 }
1853
1854 fn restore(&mut self, relative: &str, item: &Item) -> Result<(), Stop> {
1859 let here = self.folder.join(relative);
1860 if here.exists() || (self.state.sent.contains_key(relative) && !self.lost) || item.size > LARGEST {
1861 return Ok(());
1862 }
1863 let Some(got) = self.server.tail(&self.url(relative), 0)? else { return Ok(()) };
1864 if let Some(e) = got.error {
1865 return Err(Stop::Failed(e));
1866 }
1867 let keep = if matches!(kind_of(relative), Some((Kind::Round, _))) { got.bytes.iter().rposition(|b| *b == b'\n').map_or(0, |at| at + 1) } else { got.bytes.len() };
1869 crate::share::write_atomically(&here, &got.bytes[..keep]).map_err(|e| Stop::Failed(SyncError::Disk(e)))?;
1870 let modified = std::fs::metadata(&here).map(|m| crate::share::modified_ns(&m)).unwrap_or(0);
1871 let etag = if keep == got.bytes.len() { item.etag.clone() } else { String::new() };
1872 self.state.sent.insert(relative.to_string(), Sent { size: keep as u64, modified, etag });
1873 self.pulled.fetched += 1;
1874 self.pulled.bytes += keep as u64;
1875 Ok(())
1876 }
1877
1878 fn fetch_blobs(&mut self) -> Result<(), Stop> {
1882 for (relative, item) in std::mem::take(&mut self.blobs) {
1883 if let Err(stop) = self.blob(&relative, &item) {
1884 self.state.folders.remove("blobs/");
1885 self.state.folders.remove("");
1886 self.blobs_listed = None;
1887 return Err(stop);
1888 }
1889 }
1890 if let Some(etag) = self.blobs_listed.take() {
1891 self.looked("blobs/", &etag);
1892 }
1893 Ok(())
1894 }
1895
1896 fn blob(&mut self, relative: &str, item: &Item) -> Result<(), Stop> {
1899 let here = self.folder.join(relative);
1900 if self.state.sent.contains_key(relative) || item.size > BLOB_LIMIT {
1901 return Ok(());
1902 }
1903 let known = Fetched { etag: item.etag.clone(), size: item.size };
1904 if std::fs::metadata(&here).is_ok_and(|m| m.len() == item.size) {
1905 self.state.files.insert(relative.to_string(), known);
1906 return Ok(());
1907 }
1908 if let Some(size) = self.server.download(&self.url(relative), &here, BLOB_LIMIT)? {
1909 self.pulled.fetched += 1;
1910 self.pulled.bytes += size;
1911 self.state.files.insert(relative.to_string(), known);
1912 }
1913 Ok(())
1914 }
1915
1916 fn push(&mut self) -> Result<(), Stop> {
1924 for relative in own_files(self.folder, self.own, self.state) {
1925 let Ok(meta) = std::fs::metadata(self.folder.join(&relative)) else { continue };
1926 let (size, modified) = (meta.len(), crate::share::modified_ns(&meta));
1927 let sent = self.state.sent.get(&relative).cloned();
1928 let covered = self.covered.contains(dir_of(&relative));
1929 let there = self.own_there.get(&relative).cloned();
1930 let same_here = sent.as_ref().is_some_and(|s| s.size == size && s.modified == modified);
1931 if same_here
1933 && let (Some(sent), Some(etag)) = (&sent, &there)
1934 && sent.etag.is_empty()
1935 {
1936 self.state.sent.insert(relative.clone(), Sent { etag: etag.clone(), ..sent.clone() });
1937 continue;
1938 }
1939 let same_there = !covered || there.as_deref() == sent.as_ref().map(|s| s.etag.as_str());
1940 if same_here && same_there {
1941 continue;
1942 }
1943 let seen = if covered { there } else { sent.map(|s| s.etag).filter(|e| !e.is_empty()) };
1944 self.send(&relative, size, modified, seen)?;
1945 }
1946 if let Some(latest) = own_latest(self.folder, self.own)
1947 && self.covered.contains("")
1948 {
1949 let gone: Vec<(String, String)> = self
1950 .own_there
1951 .iter()
1952 .filter(|(relative, _)| matches!(kind_of(relative), Some((Kind::Round, _))) && self.state.sent.contains_key(*relative) && !self.folder.join(relative).exists())
1953 .filter(|(relative, _)| relative.trim_end_matches(".jsonl").rsplit('-').next().and_then(|n| n.parse::<u32>().ok()).is_some_and(|round| round < latest))
1954 .map(|(relative, etag)| (relative.clone(), etag.clone()))
1955 .collect();
1956 for (relative, etag) in gone {
1957 self.take_out(&relative, Some(etag))?;
1958 }
1959 }
1960 let swept: Vec<(String, String)> = self.state.sent.iter().filter(|(relative, _)| relative.starts_with("blobs/") && !self.folder.join(relative).exists()).map(|(relative, sent)| (relative.clone(), sent.etag.clone())).collect();
1961 for (relative, etag) in swept {
1962 self.take_out(&relative, Some(etag).filter(|e| !e.is_empty()))?;
1963 }
1964 Ok(())
1965 }
1966
1967 fn send(&mut self, relative: &str, size: u64, modified: u64, seen: Option<String>) -> Result<(), Stop> {
1974 let blob = relative.starts_with("blobs/");
1975 let bytes = std::fs::read(self.folder.join(relative)).map_err(|e| Stop::Failed(SyncError::Disk(format!("{relative}: {e}"))))?;
1976 let url = self.url(relative);
1977 let mut seen = if blob { None } else { seen };
1978 let sha1 = sha1_hex(&bytes);
1979 for attempt in 0..3 {
1980 let (status, etag) = self.server.put_file(&url, bytes.clone(), seen.as_deref(), modified / 1_000_000_000, &sha1).map_err(|failed| Stop::Failed(failed.error))?;
1982 match status {
1983 200 | 201 | 204 => {
1984 self.state.sent.insert(relative.to_string(), Sent { size, modified, etag: etag.unwrap_or_default() });
1985 self.pulled.sent += 1;
1986 self.pulled.bytes += bytes.len() as u64;
1987 return Ok(());
1988 }
1989 412 if blob => {
1990 self.state.sent.insert(relative.to_string(), Sent { size, modified, etag: String::new() });
1991 return Ok(());
1992 }
1993 412 => seen = self.server.etag(&url)?,
1994 409 if attempt == 0 => self.make_folders(relative)?,
1995 507 => return Err(Stop::Failed(SyncError::Server("quota".into()))),
1996 status => return Err(Stop::Failed(SyncError::Server(format!("PUT: {status}")))),
1997 }
1998 }
1999 Err(Stop::Failed(SyncError::Server("PUT: 412".into())))
2000 }
2001
2002 fn make_folders(&mut self, relative: &str) -> Result<(), Stop> {
2004 let mut at = String::new();
2005 let names: Vec<&str> = relative.split('/').collect();
2006 for name in &names[..names.len().saturating_sub(1)] {
2007 at = format!("{at}{name}/");
2008 match self.server.mkcol(&self.url(&at))? {
2009 201 | 405 => {}
2010 status => return Err(Stop::Failed(SyncError::Server(format!("MKCOL: {status}")))),
2011 }
2012 }
2013 Ok(())
2014 }
2015
2016 fn take_out(&mut self, relative: &str, seen: Option<String>) -> Result<(), Stop> {
2020 match self.server.delete(&self.url(relative), seen.as_deref())? {
2021 200 | 204 | 404 | 412 => {
2022 self.state.sent.remove(relative);
2023 Ok(())
2024 }
2025 status => Err(Stop::Failed(SyncError::Server(format!("DELETE: {status}")))),
2026 }
2027 }
2028
2029 fn round(&mut self, name: &str, item: &Item) -> Result<(), Stop> {
2035 let (here, into) = (self.folder.join(name), self.into.join(name));
2036 let size = |path: &Path| std::fs::metadata(path).map_or(0, |m| m.len());
2037 let longest = size(&here).max(size(&into));
2038 let known = Fetched { etag: item.etag.clone(), size: item.size };
2039 if item.size <= longest {
2040 self.state.files.insert(name.to_string(), known);
2042 return Ok(());
2043 }
2044 if item.size > round_cap() {
2045 self.state.left.insert(name.to_string(), item.size);
2047 self.pulled.left += 1;
2048 return Ok(());
2049 }
2050 let base = if size(&into) >= size(&here) { into.clone() } else { here.clone() };
2051 let (end, last) = whole_end(&base);
2052 let from = if end - last > OVERLAP { end.saturating_sub(1) } else { last };
2053 let overlap = read_range(&base, from, end).unwrap_or_default();
2054 let url = self.url(name);
2055 let Some(mut got) = self.server.tail(&url, from)? else { return Ok(()) };
2056 let joins = got.start == from && from > 0;
2057 if joins && !got.bytes.starts_with(&overlap) {
2058 if got.error.is_some() && overlap.starts_with(&got.bytes) {
2059 return Err(Stop::Failed(got.error.take().unwrap_or(SyncError::Network(String::new()))));
2061 }
2062 match self.server.tail(&url, 0)? {
2064 Some(whole) => got = whole,
2065 None => return Ok(()),
2066 }
2067 } else if !joins && got.start != 0 {
2068 return Err(Stop::Failed(SyncError::Server(format!("GET: from {}", got.start))));
2069 }
2070 let prefix = if got.start == from && from > 0 { from } else { 0 };
2071 self.pulled.bytes += got.bytes.len() as u64;
2072 let whole = got.bytes.iter().rposition(|b| *b == b'\n').map_or(0, |at| at + 1);
2074 if prefix + whole as u64 > longest {
2075 let temporary = crate::share::temporary(&into);
2076 let written = (|| -> std::io::Result<()> {
2077 if let Some(parent) = into.parent() {
2078 crate::share::private_dirs(parent)?;
2079 }
2080 let mut out = crate::share::new_private(&temporary)?;
2081 if prefix > 0 {
2082 std::io::copy(&mut std::fs::File::open(&base)?.take(prefix), &mut out)?;
2083 }
2084 out.write_all(&got.bytes[..whole])?;
2085 out.sync_all()?;
2086 std::fs::rename(&temporary, &into)
2087 })();
2088 if let Err(e) = written {
2089 let _ = std::fs::remove_file(&temporary);
2090 return Err(Stop::Failed(SyncError::Disk(format!("{name}: {e}"))));
2091 }
2092 self.pulled.fetched += 1;
2093 }
2094 match got.error {
2095 Some(e) => Err(Stop::Failed(e)),
2096 None => {
2097 self.state.files.insert(name.to_string(), known);
2098 Ok(())
2099 }
2100 }
2101 }
2102}
2103
2104fn whole_end(path: &Path) -> (u64, u64) {
2106 let Ok(mut file) = std::fs::File::open(path) else { return (0, 0) };
2107 let size = file.metadata().map_or(0, |m| m.len());
2108 let Some(end) = newline_before(&mut file, size).map(|at| at + 1) else { return (0, 0) };
2109 let last = newline_before(&mut file, end - 1).map_or(0, |at| at + 1);
2110 (end, last)
2111}
2112
2113fn newline_before(file: &mut std::fs::File, before: u64) -> Option<u64> {
2115 let mut end = before;
2116 while end > 0 {
2117 let from = end.saturating_sub(64 << 10);
2118 let mut piece = vec![0u8; usize::try_from(end - from).ok()?];
2119 file.seek(SeekFrom::Start(from)).ok()?;
2120 file.read_exact(&mut piece).ok()?;
2121 if let Some(at) = piece.iter().rposition(|b| *b == b'\n') {
2122 return Some(from + at as u64);
2123 }
2124 end = from;
2125 }
2126 None
2127}
2128
2129fn read_range(path: &Path, from: u64, to: u64) -> Option<Vec<u8>> {
2130 let mut file = std::fs::File::open(path).ok()?;
2131 file.seek(SeekFrom::Start(from)).ok()?;
2132 let mut bytes = vec![0u8; usize::try_from(to.saturating_sub(from)).ok()?];
2133 file.read_exact(&mut bytes).ok()?;
2134 Some(bytes)
2135}
2136
2137pub const MIRROR: &str = "mirror";
2144
2145pub fn mirror_of(memory: &Path) -> PathBuf {
2147 memory.with_file_name("mirror")
2148}
2149
2150const BLOB_LIMIT: u64 = crate::blobs::LARGEST + (1 << 20);
2152
2153pub struct Opened {
2157 pub url: String,
2158 root: String,
2159 pub place: String,
2160 pub seal: Option<Vec<u8>>,
2161}
2162
2163pub fn open(login: &Login, place: &str) -> Result<Opened, String> {
2166 open_with(login, place, LIMITS)
2167}
2168
2169pub(crate) fn open_with(login: &Login, place: &str, limits: Limits) -> Result<Opened, String> {
2170 let host = host_of(&login.url);
2171 let place = place.trim().trim_matches('/').to_string();
2172 if place.is_empty() {
2173 return Err("no-place".into());
2174 }
2175 let server = Server::new(login, limits).map_err(|e| code(&e, &login.account))?;
2176 let root = files_root(&server, login).map_err(|e| code(&e, &host))?.ok_or_else(|| format!("no-files:{host}"))?;
2177 let url = format!("{root}{}/", encode_path(&place));
2178 let seal = server.small(&format!("{url}seal.toml")).map_err(|e| code(&e, &host))?.map(|(bytes, _)| bytes);
2179 Ok(Opened { url, root, place, seal })
2180}
2181
2182pub fn create(login: &Login, opened: &Opened, folder: &Path) -> Result<(), String> {
2186 create_with(login, opened, folder, LIMITS)
2187}
2188
2189pub(crate) fn create_with(login: &Login, opened: &Opened, folder: &Path, limits: Limits) -> Result<(), String> {
2190 let host = host_of(&login.url);
2191 let failed = |e: SyncError| code(&e, &host);
2192 let server = Server::new(login, limits).map_err(failed)?;
2193 let mut at = opened.root.clone();
2194 for name in opened.place.split('/').filter(|n| !n.is_empty()) {
2195 at = format!("{at}{}/", encode_path(name));
2196 if !matches!(server.mkcol(&at).map_err(failed)?, 201 | 405) {
2197 return Err(format!("server:{host}"));
2198 }
2199 }
2200 let seal = seal_of(folder).ok_or_else(|| "no-seal".to_string())?;
2201 match server.put(&format!("{}seal.toml", opened.url), seal, None).map_err(failed)?.0 {
2202 200 | 201 | 204 => Ok(()),
2203 412 => Err("sealed-meanwhile".into()),
2204 _ => Err(format!("server:{host}")),
2205 }
2206}
2207
2208pub fn begin(memory: &Path, folder: &Path, login: &Login, opened: &Opened, now: i64) -> Result<(), String> {
2211 let _ = std::fs::remove_dir_all(cache_of(memory));
2212 let state = State { mode: MIRROR.into(), place: opened.place.clone(), folder: shown(folder), account: login.account.clone(), url: opened.url.clone(), confirmed: now, sealed: now, ..State::default() };
2213 let path = state_path(memory);
2214 sioul_core::filelock::with_lock(&path, || state.write(memory))
2215}
2216
2217pub fn step(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64, whole: bool) -> Pulled {
2228 step_with(memory, folder, own, login, now, whole, LIMITS)
2229}
2230
2231pub(crate) fn step_with(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64, whole: bool, limits: Limits) -> Pulled {
2232 sioul_core::filelock::with_lock(&memory.with_file_name("mirror.step"), || {
2233 let mut state = State::load(memory);
2234 let mut pulled = Pulled::default();
2235 if state.mode != MIRROR || !state.confirmed_for(folder) || state.account != login.account {
2236 pulled.problem = Some("not-confirmed".into());
2237 return pulled;
2238 }
2239 let host = state.host();
2240 if whole {
2241 state.tried = now;
2242 }
2243 let outcome = Server::new(login, limits).map_err(Stop::Failed).and_then(|server| {
2244 let mut puller = Puller { server: &server, folder, into: folder, own, state: &mut state, pulled: &mut pulled, now, own_there: BTreeMap::new(), covered: BTreeSet::new(), lost: false, blobs: Vec::new(), blobs_listed: None };
2245 if whole {
2246 puller.run()?;
2247 }
2248 let pushed = puller.push();
2252 let fetched = if whole { puller.fetch_blobs() } else { Ok(()) };
2253 pushed.and(fetched)
2254 });
2255 match outcome {
2256 Ok(()) => {
2257 if whole {
2258 state.last = now;
2259 }
2260 state.pushed = now;
2261 state.said.clear();
2262 }
2263 Err(Stop::Unconfirmed(why)) => {
2264 state.confirmed = 0;
2265 state.said = why;
2266 }
2267 Err(Stop::Failed(e)) => state.said = code(&e, &host),
2268 }
2269 let _ = state.save(memory);
2270 pulled.problem = (!state.said.is_empty()).then(|| state.said.clone());
2271 pulled
2272 })
2273}
2274
2275fn dir_of(relative: &str) -> &str {
2277 relative.rfind('/').map_or("", |at| &relative[..=at])
2278}
2279
2280fn own_files(folder: &Path, own: &str, state: &State) -> Vec<String> {
2288 let mut out = own_plain(folder, own);
2289 out.extend(file_names(&folder.join("blobs")).into_iter().map(|n| format!("blobs/{n}")).filter(|r| !state.files.contains_key(r)));
2290 out
2291}
2292
2293fn file_names(dir: &Path) -> Vec<String> {
2295 std::fs::read_dir(dir).into_iter().flatten().filter_map(Result::ok).filter(|e| e.file_type().is_ok_and(|t| t.is_file())).map(|e| e.file_name().to_string_lossy().to_string()).filter(|n| !n.starts_with('.')).collect()
2296}
2297
2298fn own_plain(folder: &Path, own: &str) -> Vec<String> {
2301 let names = file_names;
2302 let mut out: Vec<String> = Vec::new();
2303 let mut records: Vec<(u8, u32, String)> = names(folder)
2304 .into_iter()
2305 .filter_map(|name| match kind_of(&name) {
2306 Some((Kind::Round, id)) if id == own => Some((0, name.trim_end_matches(".jsonl").rsplit('-').next().and_then(|n| n.parse().ok()).unwrap_or(0), name)),
2307 Some((Kind::Seen, id)) if id == own => Some((1, 0, name)),
2308 _ => None,
2309 })
2310 .collect();
2311 records.sort();
2312 out.extend(records.into_iter().map(|(_, _, name)| name));
2313 let mut parts: Vec<String> = std::fs::read_dir(folder.join("leases")).into_iter().flatten().filter_map(Result::ok).filter(|e| e.file_type().is_ok_and(|t| t.is_dir())).map(|e| e.file_name().to_string_lossy().to_string()).collect();
2314 parts.sort();
2315 out.extend(parts.into_iter().map(|part| format!("leases/{part}/{own}.lease")).filter(|claim| folder.join(claim).is_file()));
2316 let entry = format!("devices/{own}.device");
2317 if folder.join(&entry).is_file() {
2318 out.push(entry);
2319 }
2320 out
2321}
2322
2323fn own_latest(folder: &Path, own: &str) -> Option<u32> {
2325 std::fs::read_dir(folder)
2326 .into_iter()
2327 .flatten()
2328 .filter_map(Result::ok)
2329 .filter_map(|e| {
2330 let name = e.file_name().to_string_lossy().to_string();
2331 match kind_of(&name) {
2332 Some((Kind::Round, id)) if id == own => name.trim_end_matches(".jsonl").rsplit('-').next().and_then(|n| n.parse().ok()),
2333 _ => None,
2334 }
2335 })
2336 .max()
2337}
2338
2339const CHECK_AGAIN: i64 = 6 * 3600;
2345pub const LARGE_FILE: u64 = 256 << 10;
2348const PACE: i64 = 10 * 60;
2349pub const SLOW: u64 = 64 << 10;
2352
2353pub fn backoff(tries: u32) -> i64 {
2356 [60, 120, 300, 900, 1800, 3600][tries.saturating_sub(1).min(5) as usize]
2357}
2358
2359#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2363enum Lane {
2364 Main,
2366 Urgent,
2369}
2370
2371impl Lane {
2372 fn memory(self, memory: &Path) -> PathBuf {
2373 memory.with_file_name(match self {
2374 Lane::Main => "sending.toml",
2375 Lane::Urgent => "sending-urgent.toml",
2376 })
2377 }
2378
2379 fn lock(self, memory: &Path) -> PathBuf {
2380 memory.with_file_name(match self {
2381 Lane::Main => "send.step",
2382 Lane::Urgent => "send.urgent",
2383 })
2384 }
2385}
2386
2387#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
2392pub struct Sending {
2393 #[serde(default, skip_serializing_if = "String::is_empty")]
2395 pub url: String,
2396 #[serde(default)]
2398 pub last: i64,
2399 #[serde(default)]
2400 pub tried: i64,
2401 #[serde(default)]
2403 pub checked: i64,
2404 #[serde(default, skip_serializing_if = "String::is_empty")]
2406 pub said: String,
2407 #[serde(default, skip_serializing_if = "Option::is_none")]
2409 pub failure: Option<Failure>,
2410 #[serde(default)]
2414 pub tries: u32,
2415 #[serde(default)]
2416 pub retry_at: i64,
2417 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
2420 pub held: BTreeMap<String, Held>,
2421 #[serde(default)]
2424 pub rate: u64,
2425 #[serde(default)]
2427 pub large_at: i64,
2428 #[serde(default)]
2430 pub again: Again,
2431 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
2433 pub sent: BTreeMap<String, Sent>,
2434}
2435
2436#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
2438pub struct Failure {
2439 #[serde(default)]
2442 pub kind: String,
2443 #[serde(default)]
2445 pub file: String,
2446 #[serde(default)]
2447 pub seconds: u64,
2448 #[serde(default)]
2449 pub sent: u64,
2450 #[serde(default)]
2451 pub total: u64,
2452 #[serde(default, skip_serializing_if = "String::is_empty")]
2454 pub words: String,
2455 #[serde(default)]
2457 pub at: i64,
2458}
2459
2460#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
2462pub struct Held {
2463 #[serde(default)]
2464 pub tries: u32,
2465 #[serde(default)]
2466 pub until: i64,
2467}
2468
2469#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
2473pub struct Again {
2474 #[serde(default)]
2475 pub at: i64,
2476 #[serde(default)]
2477 pub sent: usize,
2478 #[serde(default)]
2479 pub same: usize,
2480 #[serde(default, skip_serializing_if = "String::is_empty")]
2481 pub said: String,
2482}
2483
2484#[cfg(test)]
2485fn sending_path(memory: &Path) -> PathBuf {
2486 Lane::Main.memory(memory)
2487}
2488
2489impl Sending {
2490 pub fn load(memory: &Path) -> Sending {
2491 Sending::load_lane(memory, Lane::Main)
2492 }
2493
2494 pub fn load_urgent(memory: &Path) -> Sending {
2496 Sending::load_lane(memory, Lane::Urgent)
2497 }
2498
2499 fn load_lane(memory: &Path, lane: Lane) -> Sending {
2500 std::fs::read_to_string(lane.memory(memory)).ok().and_then(|text| toml::from_str(&text).ok()).unwrap_or_default()
2501 }
2502
2503 fn save(&self, memory: &Path, lane: Lane) -> Result<(), String> {
2504 let text = toml::to_string(self).map_err(|e| e.to_string())?;
2505 crate::share::write_atomically(&lane.memory(memory), format!("# Sioul: this device's files sent to the sharing folder's server beside the sync app (docs/database.md). Never shared.\n{text}").as_bytes())
2506 }
2507
2508 pub fn note_again(memory: &Path, now: i64, why: &str) {
2511 let mut sending = Sending::load(memory);
2512 sending.again = Again { at: now, sent: 0, same: 0, said: why.to_string() };
2513 let _ = sending.save(memory, Lane::Main);
2514 }
2515
2516 pub fn slow(&self) -> bool {
2518 self.rate > 0 && self.rate < SLOW
2519 }
2520
2521 pub fn next_try(&self) -> i64 {
2524 self.held.values().map(|h| h.until).chain([self.retry_at]).filter(|at| *at > 0).min().unwrap_or(0)
2525 }
2526}
2527
2528#[allow(clippy::too_many_arguments)]
2546pub fn send(memory: &Path, folder: &Path, own: &str, login: &Login, sealed: &BTreeMap<String, u64>, now: i64, again: bool, frugal: bool) -> Pulled {
2547 send_lane(Lane::Main, memory, folder, own, login, sealed, now, again, frugal, SEND_LIMITS)
2548}
2549
2550pub fn send_urgent(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64) -> Pulled {
2555 send_lane(Lane::Urgent, memory, folder, own, login, &BTreeMap::new(), now, false, false, SEND_LIMITS)
2556}
2557
2558#[cfg(test)]
2559pub(crate) fn send_urgent_with(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64, limits: Limits) -> Pulled {
2560 send_lane(Lane::Urgent, memory, folder, own, login, &BTreeMap::new(), now, false, false, limits)
2561}
2562
2563#[cfg(test)]
2564#[allow(clippy::too_many_arguments)]
2565pub(crate) fn send_with(memory: &Path, folder: &Path, own: &str, login: &Login, sealed: &BTreeMap<String, u64>, now: i64, again: bool, limits: Limits) -> Pulled {
2566 send_lane(Lane::Main, memory, folder, own, login, sealed, now, again, false, limits)
2567}
2568
2569#[allow(clippy::too_many_arguments)]
2570fn send_lane(lane: Lane, memory: &Path, folder: &Path, own: &str, login: &Login, sealed: &BTreeMap<String, u64>, now: i64, again: bool, frugal: bool, limits: Limits) -> Pulled {
2571 sioul_core::filelock::with_lock(&lane.lock(memory), || {
2572 let state = State::load(memory);
2573 let mut pulled = Pulled::default();
2574 let refused = if state.mode == MIRROR && lane == Lane::Main {
2575 Some("mirror")
2576 } else if !state.confirmed_for(folder) || state.account != login.account {
2577 Some("not-confirmed")
2578 } else if state.mode != MIRROR && !state.sending() {
2579 Some("off")
2580 } else {
2581 None
2582 };
2583 if let Some(why) = refused {
2584 if again {
2585 Sending::note_again(memory, now, why);
2586 }
2587 pulled.problem = Some(why.into());
2588 return pulled;
2589 }
2590 let mut sending = Sending::load_lane(memory, lane);
2591 if sending.url != state.url {
2592 sending = Sending { url: state.url.clone(), again: sending.again.clone(), ..Sending::default() };
2593 }
2594 let entry = format!("devices/{own}.device");
2600 if lane == Lane::Main
2601 && let Some((size, at)) = state.own_entry
2602 && at > sending.last
2603 && sending.sent.get(&entry).is_some_and(|sent| sent.size != size)
2604 {
2605 sending.sent.remove(&entry);
2606 }
2607 let host = state.host();
2608 if !again && now < sending.retry_at {
2610 pulled.problem = Some(format!("waiting:{host}"));
2611 return pulled;
2612 }
2613 sending.tried = now;
2614 let check = again || sending.checked == 0 || now - sending.checked >= CHECK_AGAIN;
2615 let frugal = frugal || sending.slow();
2616 let files: Vec<String> = match lane {
2617 Lane::Urgent => own_urgent(folder, own),
2618 Lane::Main => {
2619 let mut files = own_plain(folder, own);
2620 let mut names: Vec<&String> = sealed.keys().collect();
2621 names.sort();
2622 files.extend(names.into_iter().map(|name| format!("blobs/{name}")).filter(|relative| folder.join(relative).is_file()));
2623 files
2624 }
2625 };
2626 let outcome = Server::new(login, limits).map_err(Stop::Failed).and_then(|server| Sender { server: &server, folder, own, url: &state.url, sealed, sending: &mut sending, pulled: &mut pulled, now, frugal, lane }.run(&files, check, again));
2627 match outcome {
2628 Ok(()) => {
2629 sending.tries = 0;
2630 sending.retry_at = 0;
2631 if sending.held.is_empty() {
2632 sending.last = now;
2633 sending.said.clear();
2634 sending.failure = None;
2635 } else if let Some(failure) = &sending.failure {
2636 sending.said = format!("network:{host}");
2637 pulled.failure = Some(failure.clone());
2638 }
2639 if check {
2640 sending.checked = now;
2641 }
2642 }
2643 Err(Stop::Unconfirmed(why)) => sending.said = why,
2644 Err(Stop::Failed(e)) => {
2645 sending.tries += 1;
2647 sending.retry_at = now + backoff(sending.tries);
2648 sending.said = code(&e, &host);
2649 if sending.failure.as_ref().is_none_or(|f| f.at != now) {
2650 sending.failure = Some(Failure { kind: kind_of_error(&e).into(), file: String::new(), words: e.to_string(), at: now, ..Failure::default() });
2651 }
2652 pulled.failure = sending.failure.clone();
2653 }
2654 }
2655 if again {
2656 sending.again = Again { at: now, sent: pulled.sent, same: pulled.same, said: sending.said.clone() };
2657 }
2658 let _ = sending.save(memory, lane);
2659 pulled.problem = (!sending.said.is_empty()).then(|| sending.said.clone());
2660 pulled
2661 })
2662}
2663
2664fn kind_of_error(e: &SyncError) -> &'static str {
2666 match e {
2667 SyncError::Login(_) | SyncError::AppPassword(_) => "login",
2668 SyncError::Tls(_) => "tls",
2669 SyncError::Server(d) if d == "quota" => "quota",
2670 SyncError::Server(_) => "server",
2671 SyncError::Disk(_) => "disk",
2672 SyncError::Network(d) if d == "out of time" => "out-of-time",
2673 _ => "unreachable",
2674 }
2675}
2676
2677pub fn again_mirror(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64) -> Pulled {
2683 again_mirror_with(memory, folder, own, login, now, LIMITS)
2684}
2685
2686pub(crate) fn again_mirror_with(memory: &Path, folder: &Path, own: &str, login: &Login, now: i64, limits: Limits) -> Pulled {
2687 let forgot = sioul_core::filelock::with_lock(&memory.with_file_name("mirror.step"), || {
2688 let mut state = State::load(memory);
2689 if state.mode != MIRROR {
2690 return false;
2691 }
2692 state.sent.retain(|relative, _| relative.starts_with("blobs/"));
2693 state.folders.clear();
2694 state.save(memory).is_ok()
2695 });
2696 let pulled = if forgot { step_with(memory, folder, own, login, now, true, limits) } else { Pulled { problem: Some("not-confirmed".into()), ..Pulled::default() } };
2697 let mut sending = Sending::load(memory);
2698 sending.again = Again { at: now, sent: pulled.sent, same: 0, said: pulled.problem.clone().unwrap_or_default() };
2699 let _ = sending.save(memory, Lane::Main);
2700 pulled
2701}
2702
2703pub fn place_texts(folder: &Path, own: &str, source: &Path) -> Result<bool, String> {
2710 let target = texts_file(folder, own);
2711 let Ok(bytes) = std::fs::read(source) else {
2712 return Ok(std::fs::remove_file(&target).is_ok());
2713 };
2714 if std::fs::read(&target).ok().as_deref() == Some(&bytes[..]) {
2715 return Ok(false);
2716 }
2717 if let Some(parent) = target.parent() {
2718 crate::share::private_dirs(parent).map_err(|e| format!("{}: {e}", parent.display()))?;
2719 }
2720 crate::share::write_atomically(&target, &bytes).map(|()| true)
2721}
2722
2723pub fn texts_file(folder: &Path, device: &str) -> PathBuf {
2725 folder.join("texts").join("send").join(format!("{device}.jsonl"))
2726}
2727
2728fn own_urgent(folder: &Path, own: &str) -> Vec<String> {
2730 let relative = format!("texts/send/{own}.jsonl");
2731 if folder.join(&relative).is_file() { vec![relative] } else { Vec::new() }
2732}
2733
2734fn file_kind(relative: &str) -> &'static str {
2736 match kind_of(relative) {
2737 Some((Kind::Round, _)) => "records",
2738 Some((Kind::Seen, _)) => "notes",
2739 Some((Kind::Device, _)) => "entry",
2740 Some((Kind::Claim, _)) => "claim",
2741 Some((Kind::Texts, _)) => "texts",
2742 None => "sealed",
2743 }
2744}
2745
2746#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2749enum Went {
2750 Done,
2751 Left,
2752 Held,
2753}
2754
2755struct Sender<'a> {
2757 server: &'a Server,
2758 folder: &'a Path,
2759 own: &'a str,
2760 url: &'a str,
2762 sealed: &'a BTreeMap<String, u64>,
2763 sending: &'a mut Sending,
2764 pulled: &'a mut Pulled,
2765 now: i64,
2766 frugal: bool,
2767 lane: Lane,
2768}
2769
2770impl Sender<'_> {
2771 fn url(&self, relative: &str) -> String {
2772 format!("{}{}", self.url, encode_path(relative))
2773 }
2774
2775 fn run(&mut self, files: &[String], check: bool, again: bool) -> Result<(), Stop> {
2780 let listed = if check { Some(self.list_own()?) } else { None };
2781 let mut records_held = false;
2782 for relative in files {
2783 let records = matches!(kind_of(relative), Some((Kind::Round, _)));
2784 if records && records_held {
2785 continue;
2786 }
2787 let there = listed.as_ref().map(|listed| listed.get(relative).cloned());
2788 let went = self.one(relative, there, again)?;
2789 records_held |= records && went == Went::Held;
2790 }
2791 if self.lane == Lane::Main {
2792 self.take_out_gone()?;
2793 }
2794 Ok(())
2795 }
2796
2797 fn list_own(&mut self) -> Result<BTreeMap<String, Item>, Stop> {
2801 let mut out = BTreeMap::new();
2802 let root = self.server.list(self.url).map_err(|e| match e {
2803 SyncError::NotFound(_) => Stop::Unconfirmed(format!("seal-gone:{}", host_of(self.url))),
2804 e => Stop::Failed(e),
2805 })?;
2806 self.pulled.listed += root.items.len();
2807 let mut dirs: Vec<String> = Vec::new();
2808 for (name, item) in &root.items {
2809 if item.dir {
2810 if name == "devices" || (name == "blobs" && !self.sealed.is_empty()) {
2811 dirs.push(format!("{name}/"));
2812 }
2813 if name == "texts" {
2814 dirs.push("texts/send/".into());
2815 }
2816 if name == "leases" {
2817 match self.server.list(&self.url("leases/")) {
2818 Ok(parts) => dirs.extend(parts.items.iter().filter(|(_, item)| item.dir).map(|(part, _)| format!("leases/{part}/"))),
2819 Err(SyncError::NotFound(_)) => {}
2820 Err(e) => return Err(Stop::Failed(e)),
2821 }
2822 }
2823 } else {
2824 out.insert(name.clone(), item.clone());
2825 }
2826 }
2827 for dir in dirs {
2828 match self.server.list(&self.url(&dir)) {
2829 Ok(listing) => {
2830 self.pulled.listed += listing.items.len();
2831 out.extend(listing.items.into_iter().filter(|(_, item)| !item.dir).map(|(name, item)| (format!("{dir}{name}"), item)));
2832 }
2833 Err(SyncError::NotFound(_)) => {}
2834 Err(e) => return Err(Stop::Failed(e)),
2835 }
2836 }
2837 Ok(out)
2838 }
2839
2840 fn same(&self, relative: &str, there: &Item, bytes: &[u8], sha1: &str) -> Result<bool, Stop> {
2845 if there.dir || there.size != bytes.len() as u64 {
2846 return Ok(false);
2847 }
2848 if relative.starts_with("blobs/") {
2849 return Ok(true);
2850 }
2851 if let Some(theirs) = sha1_of(&there.checksum) {
2852 return Ok(theirs == sha1);
2853 }
2854 if matches!(kind_of(relative), Some((Kind::Round, _))) {
2855 return Ok(true);
2856 }
2857 Ok(self.server.small_up_to(&self.url(relative), TEXTS_LARGEST)?.is_some_and(|(theirs, _)| theirs == bytes))
2858 }
2859
2860 fn one(&mut self, relative: &str, listed: Option<Option<Item>>, again: bool) -> Result<Went, Stop> {
2867 let path = self.folder.join(relative);
2868 let Ok(meta) = std::fs::metadata(&path) else { return Ok(Went::Left) };
2869 let (size, modified) = (meta.len(), crate::share::modified_ns(&meta));
2870 let blob = relative.strip_prefix("blobs/");
2871 if blob.is_some_and(|name| self.sealed.get(name) != Some(&size)) {
2873 return Ok(Went::Left);
2874 }
2875 let records = matches!(kind_of(relative), Some((Kind::Round, _)));
2876 let sent = self.sending.sent.get(relative).cloned();
2877 let same_here = sent.as_ref().is_some_and(|s| s.size == size && s.modified == modified);
2878 match &listed {
2879 None if same_here => return Ok(Went::Left),
2880 Some(there) if same_here && !again && there.as_ref().is_some_and(|t| t.etag == sent.as_ref().map_or("", |s| s.etag.as_str())) => return Ok(Went::Left),
2881 _ => {}
2882 }
2883 if !again && self.sending.held.get(relative).is_some_and(|held| self.now < held.until) {
2885 return Ok(Went::Held);
2886 }
2887 if records && sent.as_ref().is_some_and(|s| size < s.size) {
2890 self.pulled.longer += 1;
2891 return Ok(Went::Left);
2892 }
2893 let large = size > LARGE_FILE;
2895 if large && self.frugal && !again && self.now - self.sending.large_at < PACE {
2896 self.pulled.paced += 1;
2897 return Ok(if records { Went::Held } else { Went::Left });
2898 }
2899 let bytes = std::fs::read(&path).map_err(|e| Stop::Failed(SyncError::Disk(format!("{relative}: {e}"))))?;
2900 let size = bytes.len() as u64;
2902 let sha1 = sha1_hex(&bytes);
2903 let url = self.url(relative);
2904 let mut there: Option<Option<Item>> = listed;
2906 if there.is_none() && sent.as_ref().is_none_or(|s| s.etag.is_empty()) {
2907 there = Some(self.server.stat(&url)?);
2908 }
2909 let modified_s = modified / 1_000_000_000;
2910 for attempt in 0..3 {
2911 if let Some(Some(item)) = &there {
2912 if self.same(relative, item, &bytes, &sha1)? {
2913 self.sending.sent.insert(relative.to_string(), Sent { size, modified, etag: item.etag.clone() });
2914 self.sending.held.remove(relative);
2915 self.pulled.same += 1;
2916 return Ok(Went::Done);
2917 }
2918 if self.pulled.differ.len() < 8 {
2920 self.pulled.differ.push(format!("{} {size} here, {} there", file_kind(relative), item.size));
2921 }
2922 if records && item.size > size {
2926 self.sending.sent.insert(relative.to_string(), Sent { size: item.size, modified, etag: item.etag.clone() });
2927 self.pulled.longer += 1;
2928 return Ok(Went::Left);
2929 }
2930 }
2931 let seen: Option<String> = match &there {
2932 Some(Some(item)) => Some(item.etag.clone()),
2933 Some(None) => None,
2934 None => sent.as_ref().map(|s| s.etag.clone()),
2935 };
2936 if large {
2937 self.sending.large_at = self.now;
2938 }
2939 let started = Instant::now();
2940 let (status, etag) = match self.server.put_file(&url, bytes.clone(), seen.as_deref(), modified_s, &sha1) {
2941 Ok(answer) => answer,
2942 Err(failed) => return self.failed(relative, failed),
2943 };
2944 match status {
2945 200 | 201 | 204 => {
2946 self.sending.sent.insert(relative.to_string(), Sent { size, modified, etag: etag.unwrap_or_default() });
2947 self.sending.held.remove(relative);
2948 self.pulled.sent += 1;
2949 self.pulled.bytes += size;
2950 if size >= 64 << 10 {
2951 self.sending.rate = size * 1000 / (started.elapsed().as_millis() as u64).max(1);
2952 }
2953 return Ok(Went::Done);
2954 }
2955 412 | 404 => there = Some(self.server.stat(&url)?),
2957 409 if attempt == 0 => self.make_folders(relative)?,
2958 507 => return Err(Stop::Failed(SyncError::Server("quota".into()))),
2959 status => return Err(Stop::Failed(SyncError::Server(format!("PUT: {status}")))),
2960 }
2961 }
2962 Err(Stop::Failed(SyncError::Server("PUT: 412".into())))
2963 }
2964
2965 fn failed(&mut self, relative: &str, failed: PutFailed) -> Result<Went, Stop> {
2970 self.sending.failure = Some(Failure { kind: failed.kind.into(), file: file_kind(relative).into(), seconds: failed.seconds, sent: failed.sent, total: failed.total, words: failed.words.clone(), at: self.now });
2971 if failed.sent > 0 && failed.seconds > 0 {
2972 self.sending.rate = failed.sent / failed.seconds;
2973 }
2974 if matches!(failed.kind, "timeout" | "stalled" | "network") && failed.sent > 0 {
2975 let held = self.sending.held.entry(relative.to_string()).or_default();
2976 held.tries += 1;
2977 held.until = self.now + backoff(held.tries);
2978 self.pulled.held += 1;
2979 return Ok(Went::Held);
2980 }
2981 Err(Stop::Failed(failed.error))
2982 }
2983
2984 fn make_folders(&mut self, relative: &str) -> Result<(), Stop> {
2986 let mut at = String::new();
2987 let names: Vec<&str> = relative.split('/').collect();
2988 for name in &names[..names.len().saturating_sub(1)] {
2989 at = format!("{at}{name}/");
2990 match self.server.mkcol(&self.url(&at))? {
2991 201 | 405 => {}
2992 status => return Err(Stop::Failed(SyncError::Server(format!("MKCOL: {status}")))),
2993 }
2994 }
2995 Ok(())
2996 }
2997
2998 fn take_out_gone(&mut self) -> Result<(), Stop> {
3005 let latest = own_latest(self.folder, self.own);
3006 let gone: Vec<(String, String)> = self.sending.sent.iter().filter(|(relative, _)| !self.folder.join(relative).exists()).map(|(relative, sent)| (relative.clone(), sent.etag.clone())).collect();
3007 for (relative, etag) in gone {
3008 let round = relative.trim_end_matches(".jsonl").rsplit('-').next().and_then(|n| n.parse::<u32>().ok());
3009 let past = matches!(kind_of(&relative), Some((Kind::Round, _))) && round.zip(latest).is_some_and(|(round, latest)| round < latest);
3010 let swept = relative.strip_prefix("blobs/").is_some_and(|name| !self.sealed.contains_key(name));
3011 if past || swept {
3012 match self.server.delete(&self.url(&relative), Some(&etag).filter(|e| !e.is_empty()).map(String::as_str))? {
3013 200 | 204 | 404 | 412 => {}
3014 status => return Err(Stop::Failed(SyncError::Server(format!("DELETE: {status}")))),
3015 }
3016 }
3017 if past || swept || !matches!(kind_of(&relative), Some((Kind::Round, _))) {
3018 self.sending.sent.remove(&relative);
3019 self.sending.held.remove(&relative);
3020 }
3021 }
3022 Ok(())
3023 }
3024}
3025
3026#[cfg(test)]
3027pub(crate) mod fake {
3028 use std::collections::hash_map::DefaultHasher;
3034 use std::hash::{Hash, Hasher};
3035 use std::io::{BufRead, BufReader, Read, Write};
3036 use std::net::{TcpListener, TcpStream};
3037 use std::path::{Path, PathBuf};
3038 use std::sync::{Arc, Mutex};
3039 use std::time::Duration;
3040
3041 pub(crate) const USER: &str = "jane";
3042 pub(crate) const PASSWORD: &str = "correct horse battery staple";
3043 const FILES: &str = "/remote.php/dav/files/jane";
3044
3045 #[derive(Debug, Clone, Copy)]
3047 pub(crate) enum Fault {
3048 Status(u16),
3050 Late(Duration),
3052 Cut(usize),
3054 Drop,
3056 }
3057
3058 pub(crate) struct Request {
3060 pub method: String,
3061 pub path: String,
3062 headers: Vec<(String, String)>,
3063 pub body: Vec<u8>,
3064 }
3065
3066 impl Request {
3067 pub fn header(&self, name: &str) -> Option<&str> {
3068 self.headers.iter().find(|(n, _)| n.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())
3069 }
3070 }
3071
3072 type Faults = Vec<Box<dyn FnMut(&Request) -> Option<Fault> + Send>>;
3073 pub(crate) type Reply = (u16, Vec<(&'static str, String)>, Vec<u8>);
3074
3075 pub(crate) struct Fake {
3076 pub(crate) files: PathBuf,
3078 pub(crate) base: String,
3079 sums: Mutex<std::collections::BTreeMap<PathBuf, (String, String)>>,
3083 pub(crate) throttle: Mutex<Option<u64>>,
3085 pub(crate) received: std::sync::atomic::AtomicU64,
3087 faults: Mutex<Faults>,
3088 seen: Mutex<Vec<String>>,
3089 pub(crate) turn: Mutex<()>,
3091 }
3092
3093 impl Fake {
3094 pub(crate) fn start(files: &Path) -> Arc<Fake> {
3095 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
3096 let base = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
3097 let fake = Arc::new(Fake {
3098 files: files.to_path_buf(),
3099 base,
3100 sums: Mutex::new(std::collections::BTreeMap::new()),
3101 throttle: Mutex::new(None),
3102 received: std::sync::atomic::AtomicU64::new(0),
3103 faults: Mutex::new(Vec::new()),
3104 seen: Mutex::new(Vec::new()),
3105 turn: Mutex::new(()),
3106 });
3107 let serving = Arc::clone(&fake);
3108 std::thread::spawn(move || {
3109 for stream in listener.incoming().filter_map(Result::ok) {
3110 let fake = Arc::clone(&serving);
3111 std::thread::spawn(move || fake.exchange(stream));
3112 }
3113 });
3114 fake
3115 }
3116
3117 pub(crate) fn login(&self) -> super::Login {
3119 super::Login { account: "cloud".into(), url: format!("{}/remote.php/dav/calendars/{USER}/", self.base), user: USER.into(), password: Some(PASSWORD.into()) }
3120 }
3121
3122 pub(crate) fn folder_url(&self, path: &str) -> String {
3124 format!("{}{FILES}/{}/", self.base, path.trim_matches('/'))
3125 }
3126
3127 pub(crate) fn fault(&self, fault: impl FnMut(&Request) -> Option<Fault> + Send + 'static) {
3129 self.faults.lock().unwrap().push(Box::new(fault));
3130 }
3131
3132 pub(crate) fn clear_faults(&self) {
3133 self.faults.lock().unwrap().clear();
3134 }
3135
3136 pub(crate) fn seen(&self) -> Vec<String> {
3138 self.seen.lock().unwrap().clone()
3139 }
3140
3141 pub(crate) fn forget_seen(&self) {
3142 self.seen.lock().unwrap().clear();
3143 }
3144
3145 fn exchange(&self, mut stream: TcpStream) {
3146 let throttle = *self.throttle.lock().unwrap();
3147 let Some(request) = read(&stream, throttle, &self.received) else { return };
3148 let fault = self.faults.lock().unwrap().iter_mut().find_map(|f| f(&request));
3149 let (status, headers, body) = match fault {
3150 Some(Fault::Drop) => {
3151 self.note(&request, "dropped");
3152 return;
3153 }
3154 Some(Fault::Status(status)) => (status, Vec::new(), Vec::new()),
3155 Some(Fault::Late(wait)) => {
3156 std::thread::sleep(wait);
3157 self.answer(&request)
3158 }
3159 _ => self.answer(&request),
3160 };
3161 self.note(&request, &status.to_string());
3162 let mut head = format!("HTTP/1.1 {status} Fake\r\nContent-Length: {}\r\nConnection: close\r\n", body.len());
3163 for (name, value) in &headers {
3164 head.push_str(&format!("{name}: {value}\r\n"));
3165 }
3166 head.push_str("\r\n");
3167 if stream.write_all(head.as_bytes()).is_err() {
3168 return;
3169 }
3170 let sent = match fault {
3171 Some(Fault::Cut(n)) => &body[..n.min(body.len())],
3172 _ => &body[..],
3173 };
3174 let _ = stream.write_all(sent);
3175 let _ = stream.flush();
3176 }
3177
3178 fn sum_of(&self, path: &Path) -> Option<String> {
3180 let sums = self.sums.lock().unwrap();
3181 sums.get(path).filter(|(etag, _)| *etag == etag_of(path)).map(|(_, sum)| sum.clone())
3182 }
3183
3184 fn propfind(&self, request: &Request, relative: &str, path: &Path) -> Reply {
3185 let Ok(meta) = std::fs::metadata(path) else { return (404, Vec::new(), Vec::new()) };
3186 let mut out = entry(&href(relative, meta.is_dir()), meta.is_dir(), &etag_of(path), meta.len(), self.sum_of(path));
3187 if meta.is_dir() && request.header("Depth") != Some("0") {
3188 let mut children: Vec<_> = std::fs::read_dir(path).into_iter().flatten().filter_map(Result::ok).collect();
3189 children.sort_by_key(|e| e.file_name());
3190 for child in children {
3191 let name = child.file_name().to_string_lossy().to_string();
3192 let inside = if relative.is_empty() { name } else { format!("{relative}/{name}") };
3193 let dir = child.file_type().is_ok_and(|t| t.is_dir());
3194 out += &entry(&href(&inside, dir), dir, &etag_of(&child.path()), child.metadata().map_or(0, |m| m.len()), self.sum_of(&child.path()));
3195 }
3196 }
3197 (207, vec![("Content-Type", "application/xml; charset=utf-8".into())], multistatus(&out))
3198 }
3199
3200 fn put(&self, request: &Request, path: &Path) -> Reply {
3204 let reply = put(request, path);
3205 if matches!(reply.0, 201 | 204) {
3206 if let Some(seconds) = request.header("X-OC-Mtime").and_then(|t| t.parse::<u64>().ok())
3207 && let Ok(file) = std::fs::File::options().write(true).open(path)
3208 {
3209 let _ = file.set_modified(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(seconds));
3210 }
3211 let mut sums = self.sums.lock().unwrap();
3212 match request.header("OC-Checksum") {
3213 Some(sum) => sums.insert(path.to_path_buf(), (etag_of(path), sum.to_string())),
3214 None => sums.remove(path),
3215 };
3216 }
3217 reply
3218 }
3219
3220 fn note(&self, request: &Request, what: &str) {
3221 self.seen.lock().unwrap().push(format!("{} {} {what}", request.method, request.path));
3222 }
3223
3224 pub(crate) fn answer(&self, request: &Request) -> Reply {
3225 let _turn = self.turn.lock().unwrap();
3226 let expected = format!("Basic {}", sioul_core::lines::base64_encode(format!("{USER}:{PASSWORD}").as_bytes()));
3227 if request.header("Authorization") != Some(expected.as_str()) {
3228 return (401, Vec::new(), Vec::new());
3229 }
3230 if request.method == "PROPFIND" && request.path == "/remote.php/dav" {
3231 let principal = format!("<d:response><d:href>/remote.php/dav/</d:href><d:propstat><d:prop><d:current-user-principal><d:href>/remote.php/dav/principals/users/{USER}/</d:href></d:current-user-principal></d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response>");
3232 return (207, vec![("Content-Type", "application/xml".into())], multistatus(&principal));
3233 }
3234 let Some(relative) = request.path.strip_prefix(FILES).map(|r| r.trim_start_matches('/').to_string()) else { return (404, Vec::new(), Vec::new()) };
3235 let path = if relative.is_empty() { self.files.clone() } else { self.files.join(&relative) };
3236 match request.method.as_str() {
3237 "PROPFIND" => self.propfind(request, &relative, &path),
3238 "GET" => get(request, &path),
3239 "PUT" => self.put(request, &path),
3240 "MKCOL" => mkcol(&path),
3241 "DELETE" => delete(request, &path),
3242 _ => (405, Vec::new(), Vec::new()),
3243 }
3244 }
3245 }
3246
3247 fn read(stream: &TcpStream, throttle: Option<u64>, received: &std::sync::atomic::AtomicU64) -> Option<Request> {
3249 let mut reader = BufReader::new(stream);
3250 let mut line = String::new();
3251 reader.read_line(&mut line).ok()?;
3252 let mut words = line.split_whitespace();
3253 let (method, raw) = (words.next()?.to_string(), words.next()?.to_string());
3254 let mut headers: Vec<(String, String)> = Vec::new();
3255 loop {
3256 let mut line = String::new();
3257 reader.read_line(&mut line).ok()?;
3258 let Some((name, value)) = line.trim_end().split_once(':') else { break };
3259 headers.push((name.trim().to_string(), value.trim().to_string()));
3260 }
3261 let length = headers.iter().find(|(n, _)| n.eq_ignore_ascii_case("content-length")).and_then(|(_, v)| v.parse().ok()).unwrap_or(0);
3262 let mut body = vec![0; length];
3263 let piece = throttle.map_or(length.max(1), |rate| usize::try_from(rate / 20).unwrap_or(usize::MAX).max(1));
3264 let mut at = 0;
3265 while at < length {
3266 let end = (at + piece).min(length);
3267 let read = reader.read_exact(&mut body[at..end]);
3268 if method == "PUT" {
3269 received.fetch_add((end - at) as u64, std::sync::atomic::Ordering::Relaxed);
3270 }
3271 read.ok()?;
3272 at = end;
3273 if throttle.is_some() && at < length {
3274 std::thread::sleep(Duration::from_millis(50));
3275 }
3276 }
3277 Some(Request { method, path: crate::dav::path_key(&raw), headers, body })
3278 }
3279
3280 pub(crate) fn multistatus(responses: &str) -> Vec<u8> {
3281 format!(r#"<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">{responses}</d:multistatus>"#).into_bytes()
3282 }
3283
3284 pub(crate) fn etag_of(path: &Path) -> String {
3286 let mut hasher = DefaultHasher::new();
3287 if path.is_dir() {
3288 let mut names: Vec<_> = std::fs::read_dir(path).into_iter().flatten().filter_map(Result::ok).map(|e| e.file_name()).collect();
3289 names.sort();
3290 for name in names {
3291 name.hash(&mut hasher);
3292 etag_of(&path.join(&name)).hash(&mut hasher);
3293 }
3294 } else {
3295 std::fs::read(path).unwrap_or_default().hash(&mut hasher);
3296 GENERATIONS.lock().unwrap().get(path).copied().unwrap_or(0).hash(&mut hasher);
3297 }
3298 format!("{:016x}", hasher.finish())
3299 }
3300
3301 static GENERATIONS: Mutex<std::collections::BTreeMap<PathBuf, u64>> = Mutex::new(std::collections::BTreeMap::new());
3304
3305 pub(crate) fn bump(path: &Path) {
3307 *GENERATIONS.lock().unwrap().entry(path.to_path_buf()).or_insert(0) += 1;
3308 }
3309
3310 fn entry(href: &str, dir: bool, etag: &str, size: u64, sum: Option<String>) -> String {
3311 let kind = if dir { "<d:resourcetype><d:collection/></d:resourcetype>" } else { "<d:resourcetype/>" };
3312 let length = if dir { String::new() } else { format!("<d:getcontentlength>{size}</d:getcontentlength>") };
3313 let sums = sum.map(|sum| format!("<oc:checksums><oc:checksum>{sum}</oc:checksum></oc:checksums>")).unwrap_or_default();
3314 format!("<d:response><d:href>{href}</d:href><d:propstat><d:prop>{kind}<d:getetag>"{etag}"</d:getetag>{length}{sums}</d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response>")
3315 }
3316
3317 pub(crate) fn href(relative: &str, dir: bool) -> String {
3318 let path = super::encode_path(relative);
3319 match (relative.is_empty(), dir) {
3320 (true, _) => format!("{FILES}/"),
3321 (false, true) => format!("{FILES}/{path}/"),
3322 (false, false) => format!("{FILES}/{path}"),
3323 }
3324 }
3325
3326 fn put(request: &Request, path: &Path) -> Reply {
3329 if !path.parent().is_some_and(Path::is_dir) {
3330 return (409, Vec::new(), Vec::new());
3331 }
3332 let reply = put_once(request, path);
3333 if matches!(reply.0, 201 | 204) {
3334 bump(path);
3335 return (reply.0, vec![("ETag", format!("\"{}\"", etag_of(path)))], Vec::new());
3336 }
3337 reply
3338 }
3339
3340 fn put_once(request: &Request, path: &Path) -> Reply {
3341 let current = path.is_file().then(|| format!("\"{}\"", etag_of(path)));
3342 let refused = match (request.header("If-None-Match"), request.header("If-Match")) {
3343 (Some("*"), _) => current.is_some(),
3344 (_, Some(wanted)) => current.as_deref() != Some(wanted),
3345 _ => false,
3346 };
3347 if refused {
3348 return (412, Vec::new(), Vec::new());
3349 }
3350 std::fs::write(path, &request.body).unwrap();
3351 (if current.is_some() { 204 } else { 201 }, vec![("ETag", format!("\"{}\"", etag_of(path)))], Vec::new())
3352 }
3353
3354 fn mkcol(path: &Path) -> Reply {
3355 if path.exists() {
3356 return (405, Vec::new(), Vec::new());
3357 }
3358 if !path.parent().is_some_and(Path::is_dir) {
3359 return (409, Vec::new(), Vec::new());
3360 }
3361 std::fs::create_dir(path).unwrap();
3362 (201, Vec::new(), Vec::new())
3363 }
3364
3365 fn delete(request: &Request, path: &Path) -> Reply {
3366 if !path.exists() {
3367 return (404, Vec::new(), Vec::new());
3368 }
3369 if request.header("If-Match").is_some_and(|wanted| format!("\"{}\"", etag_of(path)) != wanted) {
3370 return (412, Vec::new(), Vec::new());
3371 }
3372 if path.is_dir() {
3373 std::fs::remove_dir_all(path).unwrap();
3374 } else {
3375 std::fs::remove_file(path).unwrap();
3376 }
3377 (204, Vec::new(), Vec::new())
3378 }
3379
3380 fn get(request: &Request, path: &Path) -> Reply {
3381 let Ok(bytes) = std::fs::read(path) else { return (404, Vec::new(), Vec::new()) };
3382 let etag = format!("\"{}\"", etag_of(path));
3383 let from = request.header("Range").and_then(|r| r.strip_prefix("bytes=")).and_then(|r| r.strip_suffix('-')).and_then(|s| s.parse::<usize>().ok());
3384 match from {
3385 Some(from) if from >= bytes.len() => (416, vec![("Content-Range", format!("bytes */{}", bytes.len()))], Vec::new()),
3386 Some(from) => (206, vec![("ETag", etag), ("Content-Range", format!("bytes {from}-{}/{}", bytes.len() - 1, bytes.len()))], bytes[from..].to_vec()),
3387 None => (200, vec![("ETag", etag)], bytes),
3388 }
3389 }
3390}
3391
3392#[cfg(test)]
3393mod tests {
3394 use super::fake::{Fake, Fault, PASSWORD};
3395 use super::*;
3396 use crate::devices::{COMPUTER, Entry, PHONE};
3397 use crate::share::{Outcome, Roots, Sharing, Store};
3398 use sioul_core::config::Config;
3399 use sioul_core::everywhere::{Press, Switch, change_own};
3400 use sioul_core::health::{Doubt, Peer, doubts_now};
3401 use std::sync::Arc;
3402
3403 const KEY: [u8; 32] = [11u8; 32];
3404 const SEAL: &str = "# Sioul: how your devices' shared records are sealed (docs/database.md).\nversion = 1\nsalt = \"AAAAAAAAAAAAAAAAAAAAAA==\"\nmemory_kib = 65536\npasses = 3\ncheck = \"one\"\n";
3405 const OTHER_SEAL: &str = "# Sioul: how your devices' shared records are sealed (docs/database.md).\nversion = 1\nsalt = \"BBBBBBBBBBBBBBBBBBBBBB==\"\nmemory_kib = 65536\npasses = 3\ncheck = \"two\"\n";
3406 const TEST: Limits = Limits { wait: Duration::from_millis(700), small: Duration::from_secs(3), large: Duration::from_secs(5), pull: Duration::from_secs(30), blob: Duration::from_secs(5), floor: 1 << 30, stall: Duration::from_millis(700) };
3408 thread_local! {
3409 pub(super) static ROUND_CAP: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
3411 }
3412 const DUE: i64 = 1_800_000_000;
3413
3414 fn scratch(name: &str) -> PathBuf {
3415 let dir = std::env::temp_dir().join(format!("sioul-remote-{name}-{}", std::process::id()));
3416 let _ = std::fs::remove_dir_all(&dir);
3417 std::fs::create_dir_all(&dir).unwrap();
3418 dir
3419 }
3420
3421 #[test]
3424 fn the_files_stay_on_the_account_s_server() {
3425 use crate::dav::stand_in::{Reply, Request, Server as StandIn, serve};
3426 struct Noted(std::sync::Mutex<Vec<String>>);
3428 impl StandIn for Noted {
3429 fn misbehaves(&self, _: &Request) -> Option<crate::dav::stand_in::Fault> {
3430 None
3431 }
3432 fn answer(&self, request: &Request) -> Reply {
3433 self.0.lock().unwrap().push(format!("{} {} {}", request.method, request.path, request.header("Authorization").unwrap_or("")));
3434 Reply::new(404, "")
3435 }
3436 }
3437 struct Pointing(String);
3439 impl StandIn for Pointing {
3440 fn misbehaves(&self, _: &Request) -> Option<crate::dav::stand_in::Fault> {
3441 None
3442 }
3443 fn answer(&self, _: &Request) -> Reply {
3444 Reply::new(207, format!(r#"<?xml version="1.0" encoding="utf-8"?><d:multistatus xmlns:d="DAV:"><d:response><d:href>/remote.php/dav/</d:href><d:propstat><d:prop><d:current-user-principal><d:href>{}</d:href></d:current-user-principal></d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response></d:multistatus>"#, self.0))
3445 }
3446 }
3447 let noted = Arc::new(Noted(std::sync::Mutex::new(Vec::new())));
3448 let there = serve(noted.clone());
3449 let here = serve(Arc::new(Pointing(format!("{there}/remote.php/dav/principals/users/jane/"))));
3450 let login = Login { account: "cloud".into(), url: format!("{here}/remote.php/dav/calendars/jane/"), user: "jane".into(), password: Some("secret".into()) };
3451 let found = files_root(&Server::new(&login, TEST).unwrap(), &login);
3452 assert!(matches!(&found, Err(SyncError::Elsewhere(root)) if root.starts_with(&there)), "{found:?}");
3453 assert!(noted.0.lock().unwrap().is_empty());
3454 let own = serve(Arc::new(Pointing("/remote.php/dav/principals/users/jane/".into())));
3456 let login = Login { url: format!("{own}/remote.php/dav/calendars/jane/"), ..login };
3457 assert_eq!(files_root(&Server::new(&login, TEST).unwrap(), &login).unwrap(), Some(format!("{own}/remote.php/dav/files/jane/")));
3458 }
3459
3460 fn put(dir: &Path, relative: &str, text: &str) {
3461 let path = dir.join(relative);
3462 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
3463 std::fs::write(path, text).unwrap();
3464 }
3465
3466 fn walk(dir: &Path) -> BTreeMap<String, Vec<u8>> {
3468 let mut out = BTreeMap::new();
3469 let mut stack = vec![dir.to_path_buf()];
3470 while let Some(at) = stack.pop() {
3471 for entry in std::fs::read_dir(&at).into_iter().flatten().filter_map(Result::ok) {
3472 let path = entry.path();
3473 if entry.file_name().to_string_lossy().starts_with('.') {
3474 continue;
3475 }
3476 if path.is_dir() {
3477 stack.push(path);
3478 } else {
3479 let relative = path.strip_prefix(dir).unwrap().to_string_lossy().replace('\\', "/");
3480 out.insert(relative, std::fs::read(&path).unwrap());
3481 }
3482 }
3483 }
3484 out
3485 }
3486
3487 fn copy_dir(from: &Path, to: &Path) {
3488 for (relative, bytes) in walk(from) {
3489 put_bytes(to, &relative, &bytes);
3490 }
3491 }
3492
3493 fn put_bytes(dir: &Path, relative: &str, bytes: &[u8]) {
3494 let path = dir.join(relative);
3495 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
3496 std::fs::write(path, bytes).unwrap();
3497 }
3498
3499 struct Device {
3501 roots: Roots,
3502 id: String,
3503 memory: PathBuf,
3504 kind: &'static str,
3505 }
3506
3507 impl Device {
3508 fn new(base: &Path, name: &str, kind: &'static str) -> Device {
3509 let root = base.join(name);
3510 let roots = Roots { config: root.join("config"), data: root.join("data"), state: root.join("state") };
3511 for dir in [&roots.config, &roots.data, &roots.state] {
3512 std::fs::create_dir_all(dir).unwrap();
3513 }
3514 let memory = roots.state.join("share").join("memory.json");
3515 Device { roots, id: uuid::Uuid::new_v4().to_string(), memory, kind }
3516 }
3517
3518 fn stores(&self) -> Vec<Store> {
3519 crate::share::stores(&Config::default(), &self.roots)
3520 }
3521
3522 fn exchange(&self, folder: &Path, now: i64) -> Outcome {
3524 let outcome = crate::share::exchange(&Sharing { folder, computer: &self.id, key: &KEY, memory: &self.memory, files: false, hurry: None }, &self.stores(), now * 1000).unwrap();
3525 crate::devices::change(&crate::devices::own_path(&self.roots.state), Some((folder, &KEY)), |e| e.exported(now * 1000, outcome.wrote, now)).unwrap();
3526 outcome
3527 }
3528
3529 fn session(&self, folder: &Path, f: impl FnOnce(&mut Entry)) {
3530 let (kind, id) = (self.kind, self.id.clone());
3531 crate::devices::change(&crate::devices::own_path(&self.roots.state), Some((folder, &KEY)), |e| {
3532 e.id = id;
3533 e.kind = kind.into();
3534 e.doses = true;
3535 f(e);
3536 })
3537 .unwrap();
3538 }
3539
3540 fn switch(&self) -> PathBuf {
3541 self.roots.state.join(sioul_core::everywhere::SWITCH_FILE)
3542 }
3543
3544 fn press(&self, on: bool, at: i64) {
3546 change_own(&self.switch(), &self.id, |s| {
3547 s.press(&self.id, on, 0, at * 1000);
3548 true
3549 })
3550 .unwrap();
3551 }
3552
3553 fn dnd(&self) -> Option<Press> {
3554 Switch::read(&self.switch()).unwrap().latest()
3555 }
3556
3557 fn write(&self, relative: &str, text: &str) {
3558 put(&self.roots.state, relative, text);
3559 }
3560
3561 fn read(&self, relative: &str) -> String {
3562 std::fs::read_to_string(self.roots.state.join(relative)).unwrap_or_default()
3563 }
3564
3565 fn sees(&self, folder: &Path, other: &Device, now: i64) -> Peer {
3567 let entry = crate::devices::all(folder, &KEY).0.into_iter().find(|e| e.id == other.id).expect("its entry");
3568 let heard = crate::share::heard(&self.memory, &self.id);
3569 Peer { id: other.id.clone(), name: entry.name.clone(), said: Some(entry.said()), complete: entry.wrote.is_none_or(|wrote| heard.complete(&other.id, wrote)), seen: now, heard: entry.exported, ..Peer::default() }
3570 }
3571
3572 fn kept(&self) -> BTreeMap<String, Vec<u8>> {
3574 let mut out = BTreeMap::new();
3575 for (root, dir) in [("config", &self.roots.config), ("data", &self.roots.data), ("state", &self.roots.state)] {
3576 for (relative, bytes) in walk(dir) {
3577 if !relative.starts_with("share/") {
3578 out.insert(format!("{root}/{relative}"), bytes);
3579 }
3580 }
3581 }
3582 out
3583 }
3584 }
3585
3586 struct World {
3591 base: PathBuf,
3592 fake: Arc<Fake>,
3593 server: PathBuf,
3594 phone_folder: PathBuf,
3595 desk: Device,
3596 phone: Device,
3597 }
3598
3599 impl World {
3600 fn new(name: &str) -> World {
3601 let base = scratch(name);
3602 let files = base.join("cloud");
3603 let server = files.join("Documents").join("Sioul");
3604 let phone_folder = base.join("phone").join("storage").join("Documents").join("Sioul");
3605 put(&server, "seal.toml", SEAL);
3606 put(&phone_folder, "seal.toml", SEAL);
3607 let fake = Fake::start(&files);
3608 let desk = Device::new(&base, "desk", COMPUTER);
3609 let phone = Device::new(&base, "phone-state", PHONE);
3610 let world = World { base, fake, server, phone_folder, desk, phone };
3611 world.desk.session(&world.server, |e| e.start(DUE - 7_200));
3612 world.desk.exchange(&world.server, DUE - 7_100);
3613 world.down();
3614 world.phone.session(&world.phone_folder, |e| e.start(DUE - 7_050));
3615 world.phone.exchange(&world.phone_folder, DUE - 7_000);
3616 world.up();
3617 world.desk.exchange(&world.server, DUE - 6_900);
3618 world.down();
3619 world
3620 }
3621
3622 fn up(&self) {
3624 let _turn = self.fake.turn.lock().unwrap();
3625 for (relative, bytes) in walk(&self.phone_folder).into_iter().filter(|(r, _)| r.contains(&self.phone.id)) {
3626 put_bytes(&self.server, &relative, &bytes);
3627 }
3628 }
3629
3630 fn down(&self) {
3632 for (relative, bytes) in walk(&self.server).into_iter().filter(|(r, _)| !r.contains(&self.phone.id)) {
3633 put_bytes(&self.phone_folder, &relative, &bytes);
3634 }
3635 }
3636
3637 fn find(&self, now: i64) -> State {
3638 find_with(&self.phone.memory, &self.phone_folder, &[self.fake.login()], &places(&self.phone_folder, &[]), now, TEST)
3639 }
3640
3641 fn pull(&self, now: i64) -> Pulled {
3642 pull_with(&self.phone.memory, &self.phone_folder, &self.phone.id, &self.fake.login(), now, TEST)
3643 }
3644
3645 fn cache(&self) -> PathBuf {
3646 cache_of(&self.phone.memory)
3647 }
3648
3649 fn desk_round(&self) -> String {
3651 let mut names: Vec<String> = walk(&self.server).into_keys().filter(|n| n.starts_with(&self.desk.id) && n.ends_with(".jsonl")).collect();
3652 names.sort_by_key(|n| n.trim_end_matches(".jsonl").rsplit('-').next().unwrap().parse::<u32>().unwrap());
3653 names.pop().unwrap()
3654 }
3655
3656 fn snapshot(&self) -> PathBuf {
3658 let aside = self.base.join("aside");
3659 let _ = std::fs::remove_dir_all(&aside);
3660 copy_dir(&self.base.join("phone-state"), &aside.join("phone-state"));
3661 copy_dir(&self.phone_folder, &aside.join("folder"));
3662 aside
3663 }
3664
3665 fn restore(&self, aside: &Path) {
3666 let _ = std::fs::remove_dir_all(self.base.join("phone-state"));
3667 let _ = std::fs::remove_dir_all(&self.phone_folder);
3668 copy_dir(&aside.join("phone-state"), &self.base.join("phone-state"));
3669 copy_dir(&aside.join("folder"), &self.phone_folder);
3670 }
3671
3672 fn others_here(&self) -> BTreeMap<String, Vec<u8>> {
3674 walk(&self.phone_folder).into_iter().filter(|(r, _)| !r.contains(&self.phone.id)).collect()
3675 }
3676 }
3677
3678 impl Drop for World {
3679 fn drop(&mut self) {
3680 let _ = std::fs::remove_dir_all(&self.base);
3681 }
3682 }
3683
3684 fn once(matches: impl Fn(&super::fake::Request) -> bool + Send + 'static, fault: Fault) -> impl FnMut(&super::fake::Request) -> Option<Fault> + Send + 'static {
3686 let mut done = false;
3687 move |request| {
3688 if done || !matches(request) {
3689 return None;
3690 }
3691 done = true;
3692 Some(fault)
3693 }
3694 }
3695
3696 #[test]
3697 fn a_stale_folder_is_caught_up_from_the_server() {
3698 let w = World::new("stale");
3699 w.desk.press(true, DUE + 30);
3703 w.desk.exchange(&w.server, DUE + 150);
3704 w.up();
3705 let alone = w.phone.exchange(&w.phone_folder, DUE + 170);
3706 assert!(alone.problems.is_empty(), "{:?}", alone.problems);
3707 assert!(w.phone.dnd().is_none_or(|p| !p.on), "nothing came yet");
3708 let stale = w.phone.sees(&w.phone_folder, &w.desk, DUE + 180);
3709 assert!(matches!(doubts_now(DUE, DUE + 180, None, &[stale]).as_slice(), [Doubt::Working { .. }]), "the desk's news has not come: a doubt");
3710 let state = w.find(DUE + 180);
3712 assert!(state.confirmed_for(&w.phone_folder), "{state:?}");
3713 assert_eq!(state.url, w.fake.folder_url("Documents/Sioul"));
3714 let before = w.others_here();
3715 let pulled = w.pull(DUE + 181);
3716 assert!(pulled.problem.is_none() && pulled.fetched >= 2, "{pulled:?}");
3717 assert_eq!(w.others_here(), before, "nothing written into the synced folder");
3718 let outcome = w.phone.exchange(&w.phone_folder, DUE + 182);
3719 assert!(outcome.problems.is_empty(), "{:?}", outcome.problems);
3720 let press = w.phone.dnd().expect("the desk's press");
3721 assert!(press.on && press.from == w.desk.id, "{press:?}");
3722 let fresh = w.phone.sees(&w.phone_folder, &w.desk, DUE + 183);
3724 assert!(doubts_now(DUE, DUE + 183, None, std::slice::from_ref(&fresh)).is_empty(), "{fresh:?}");
3725 let direct = crate::devices::all(&w.server, &KEY).0.into_iter().find(|e| e.id == w.desk.id).unwrap();
3726 assert_eq!(fresh.said, Some(direct.said()), "the same as the server's folder says");
3727 assert!(crate::share::others(&w.phone_folder, &w.phone.id).iter().any(|o| o.id == w.desk.id && o.heard >= DUE + 150));
3728 for line in w.fake.seen() {
3730 assert!(line.starts_with("PROPFIND ") || line.starts_with("GET "), "{line}");
3731 assert!(!(line.starts_with("GET ") && line.contains(&w.phone.id)), "{line}");
3732 }
3733 let kept = std::fs::read_to_string(state_path(&w.phone.memory)).unwrap();
3734 assert!(!kept.contains(PASSWORD) && !kept.contains("Basic"), "{kept}");
3735 assert!(walk(&w.cache()).keys().all(|name| !name.contains(&w.phone.id)), "never the phone's own");
3736 }
3737
3738 #[test]
3739 fn with_the_folder_fresh_the_backup_changes_nothing() {
3740 let w = World::new("fresh");
3741 w.desk.press(true, DUE + 30);
3742 w.desk.exchange(&w.server, DUE + 150);
3743 crate::lease::renew(&w.server, &KEY, "health", &w.desk.id, DUE + 150, DUE + 150, false, crate::lease::Rule::FollowsYou, crate::share::written(&w.desk.memory, &w.desk.id)).unwrap();
3744 w.up();
3745 w.down();
3746 let aside = w.snapshot();
3748 let without = w.phone.exchange(&w.phone_folder, DUE + 200);
3749 let kept_without = w.phone.kept();
3750 w.restore(&aside);
3752 assert!(w.find(DUE + 190).confirmed_for(&w.phone_folder));
3753 w.fake.forget_seen();
3754 let pulled = w.pull(DUE + 195);
3755 assert!(pulled.problem.is_none(), "{pulled:?}");
3756 assert_eq!(pulled.fetched, 0, "nothing the folder holds is kept twice");
3757 assert!(w.fake.seen().iter().all(|line| !(line.starts_with("GET ") && line.contains(".jsonl"))), "no records fetched: {:?}", w.fake.seen());
3758 assert_eq!(walk(&w.cache()).into_keys().collect::<Vec<_>>(), ["seal.toml"]);
3759 let with = w.phone.exchange(&w.phone_folder, DUE + 200);
3760 assert_eq!((with.received, with.sent, &with.problems), (without.received, without.sent, &without.problems));
3761 assert_eq!(w.phone.kept(), kept_without, "the same files, byte for byte");
3762 w.fake.forget_seen();
3764 w.pull(DUE + 260);
3765 assert!(w.fake.seen().iter().all(|line| line.starts_with("PROPFIND ")), "{:?}", w.fake.seen());
3766 }
3767
3768 #[test]
3769 fn a_half_downloaded_file_is_never_read_as_whole() {
3770 let w = World::new("half");
3771 let first = "\"first@1800000000\" = 1800000000\n\"second@1800000000\" = 1800000001\n";
3773 w.desk.write("health-state.toml", &format!("[taken]\n{first}"));
3774 w.desk.exchange(&w.server, DUE + 50);
3775 w.down();
3776 let marks: String = (0..40).map(|n| format!("\"dose-{n}@1800000000\" = {}\n", DUE + n)).collect();
3778 w.desk.write("health-state.toml", &format!("[taken]\n{first}{marks}"));
3779 w.desk.exchange(&w.server, DUE + 100);
3780 assert!(w.find(DUE + 110).confirmed_for(&w.phone_folder));
3781 let round = w.desk_round();
3782 let whole = std::fs::read(w.server.join(&round)).unwrap();
3783 let (end, last) = whole_end(&w.phone_folder.join(&round));
3784 assert!(end > 0 && whole.len() as u64 > end, "the phone's copy is behind");
3785 let cut = (whole.len() - last as usize) / 2;
3787 let name = round.clone();
3788 w.fake.fault(once(move |r| r.method == "GET" && r.path.ends_with(&name), Fault::Cut(cut)));
3789 let pulled = w.pull(DUE + 120);
3790 assert!(pulled.problem.as_deref().is_some_and(|p| p.starts_with("network:")), "{pulled:?}");
3791 assert!(w.fake.seen().iter().any(|l| l.starts_with("GET ") && l.contains(&round) && l.ends_with(" 206")), "asked from where the phone's copy ends: {:?}", w.fake.seen());
3792 let cached = w.cache().join(&round);
3793 let arrived = std::fs::read(&cached).unwrap_or_default();
3794 assert!(arrived.is_empty() || (arrived.ends_with(b"\n") && whole.starts_with(&arrived) && arrived.len() < whole.len()), "whole lines only: {} of {}", arrived.len(), whole.len());
3795 let first = w.phone.exchange(&w.phone_folder, DUE + 130);
3796 assert!(first.problems.iter().all(|p| !p.starts_with("share-other")), "{:?}", first.problems);
3797 let entry = format!("devices/{}.device", w.desk.id);
3799 let name = entry.clone();
3800 w.fake.fault(once(move |r| r.method == "GET" && r.path.ends_with(&name), Fault::Cut(40)));
3801 let pulled = w.pull(DUE + 140);
3802 assert!(pulled.problem.as_deref().is_some_and(|p| p.starts_with("network:")), "{pulled:?}");
3803 assert_eq!(std::fs::read(&cached).unwrap(), whole, "the records whole now");
3804 assert!(!w.cache().join(&entry).exists(), "an entry cut is not kept");
3805 let (entries, unread) = crate::devices::all(&w.phone_folder, &KEY);
3806 assert!(entries.iter().any(|e| e.id == w.desk.id) && !unread.contains(&w.desk.id), "the folder's copy read meanwhile");
3807 let second = w.phone.exchange(&w.phone_folder, DUE + 150);
3808 assert!(second.problems.iter().all(|p| !p.starts_with("share-other")), "{:?}", second.problems);
3809 assert!(w.pull(DUE + 160).problem.is_none());
3810 assert!(w.cache().join(&entry).exists());
3811 let record = w.phone.read("health-state.toml");
3813 for n in 0..40 {
3814 assert!(record.contains(&format!("dose-{n}@")), "dose-{n}:\n{record}");
3815 }
3816 let heard = crate::share::heard(&w.phone.memory, &w.phone.id);
3817 assert!(heard.complete(&w.desk.id, crate::share::written(&w.desk.memory, &w.desk.id).unwrap()), "{heard:?}");
3818 assert!(!heard.broken.contains_key(&w.desk.id), "no line lost nor read twice: {heard:?}");
3819 }
3820
3821 #[test]
3822 fn a_seal_that_differs_is_never_used() {
3823 let w = World::new("seal");
3824 put(&w.server, "seal.toml", OTHER_SEAL);
3826 let state = w.find(DUE);
3827 assert!(!state.confirmed_for(&w.phone_folder) && state.said.starts_with("seal-differs:"), "{state:?}");
3828 assert!(attach(&w.phone_folder, &w.phone.memory).is_none());
3829 assert_eq!(w.pull(DUE + 1).problem.as_deref(), Some("not-confirmed"));
3830 assert!(w.fake.seen().iter().filter(|l| l.starts_with("GET ")).all(|l| l.ends_with("seal.toml 200") || l.ends_with("seal.toml 404")), "{:?}", w.fake.seen());
3831 put(&w.server, "seal.toml", SEAL);
3833 State::choose(&w.phone.memory, |s| s.asked = DUE + 2).unwrap();
3834 assert!(w.find(DUE + 3).confirmed_for(&w.phone_folder));
3835 assert!(w.pull(DUE + 4).problem.is_none());
3836 put(&w.server, "seal.toml", OTHER_SEAL);
3837 w.desk.press(true, DUE + 5);
3838 w.desk.exchange(&w.server, DUE + 6);
3839 w.fake.forget_seen();
3840 let pulled = w.pull(DUE + 7);
3841 assert!(pulled.problem.as_deref().is_some_and(|p| p.starts_with("seal-differs:")), "{pulled:?}");
3842 assert!(w.fake.seen().iter().filter(|l| l.starts_with("GET ")).all(|l| l.contains("seal.toml")), "{:?}", w.fake.seen());
3843 assert!(!State::load(&w.phone.memory).confirmed_for(&w.phone_folder));
3844 put(&w.server, "seal.toml", SEAL);
3846 State::choose(&w.phone.memory, |s| s.asked = DUE + 8).unwrap();
3847 assert!(w.find(DUE + 9).confirmed_for(&w.phone_folder));
3848 assert!(w.pull(DUE + 10).problem.is_none());
3849 assert!(attach(&w.phone_folder, &w.phone.memory).is_some());
3850 put(&w.phone_folder, "seal.toml", OTHER_SEAL);
3851 assert!(attach(&w.phone_folder, &w.phone.memory).is_none(), "another sharing's");
3852 assert_eq!(w.pull(DUE + 11).problem.as_deref(), Some("no-seal"));
3853 assert!(!w.cache().exists());
3854 }
3855
3856 #[test]
3857 fn server_errors_and_timeouts_leave_the_exchange_as_before() {
3858 let w = World::new("errors");
3859 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
3860 w.desk.press(true, DUE + 30);
3861 w.desk.exchange(&w.server, DUE + 60);
3862 w.up();
3863 for (n, fault) in [Fault::Status(500), Fault::Status(503), Fault::Late(Duration::from_secs(2)), Fault::Drop, Fault::Cut(20)].into_iter().enumerate() {
3864 w.fake.clear_faults();
3865 w.fake.fault(move |_| Some(fault));
3866 let pulled = w.pull(DUE + 100 + n as i64);
3867 let said = pulled.problem.clone().unwrap_or_default();
3868 assert!(said.starts_with("server:") || said.starts_with("network:"), "{fault:?}: {pulled:?}");
3869 assert_eq!(walk(&w.cache()).into_keys().collect::<Vec<_>>(), ["seal.toml"], "{fault:?}: nothing kept");
3870 assert_eq!(State::load(&w.phone.memory).said, said);
3871 }
3872 w.fake.clear_faults();
3873 let aside = w.snapshot();
3875 let with = w.phone.exchange(&w.phone_folder, DUE + 200);
3876 let kept_with = w.phone.kept();
3877 w.restore(&aside);
3878 forget(&w.phone.memory, &w.phone_folder);
3879 let without = w.phone.exchange(&w.phone_folder, DUE + 200);
3880 assert_eq!((with.received, &with.problems), (without.received, &without.problems));
3881 assert_eq!(w.phone.kept(), kept_with);
3882 }
3883
3884 #[test]
3885 fn what_was_fetched_never_wins_over_a_newer_synced_copy() {
3886 let w = World::new("older");
3887 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
3888 let renew = |now: i64| crate::lease::renew(&w.server, &KEY, "health", &w.desk.id, now, now, false, crate::lease::Rule::FollowsYou, crate::share::written(&w.desk.memory, &w.desk.id)).unwrap();
3889 w.desk.press(true, DUE + 30);
3890 w.desk.exchange(&w.server, DUE + 60);
3891 renew(DUE + 60);
3892 assert!(w.pull(DUE + 70).problem.is_none());
3893 let round = w.desk_round();
3894 assert!(w.cache().join(&round).exists() && w.cache().join(format!("devices/{}.device", w.desk.id)).exists());
3895 w.desk.press(false, DUE + 1_000);
3897 w.desk.exchange(&w.server, DUE + 1_100);
3898 renew(DUE + 1_100);
3899 w.up();
3900 w.down();
3901 let entry = crate::devices::all(&w.phone_folder, &KEY).0.into_iter().find(|e| e.id == w.desk.id).unwrap();
3902 assert_eq!(entry.exported, DUE + 1_100, "the folder's entry");
3903 let claim = crate::lease::claims(&w.phone_folder, &KEY, "health").into_iter().find(|c| c.computer == w.desk.id).unwrap();
3904 assert_eq!(claim.renewed, DUE + 1_100, "the folder's claim");
3905 let heard = crate::share::others(&w.phone_folder, &w.phone.id).into_iter().find(|o| o.id == w.desk.id).unwrap().heard;
3906 assert_eq!(heard, DUE + 1_100, "the folder's notes");
3907 w.phone.exchange(&w.phone_folder, DUE + 1_200);
3908 assert!(!w.phone.dnd().unwrap().on, "the later press, from the folder's longer records");
3909 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
3911 assert!(!w.cache().join(&round).exists());
3912 State::choose(&w.phone.memory, |s| s.on = Some(false)).unwrap();
3914 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
3915 assert_eq!(walk(&w.cache()).into_keys().collect::<Vec<_>>(), ["seal.toml"]);
3916 w.fake.forget_seen();
3917 assert_eq!(w.pull(DUE + 1_300).problem.as_deref(), Some("off"));
3918 assert!(w.fake.seen().is_empty(), "switched off: not a request");
3919 }
3920
3921 #[test]
3925 fn doses_are_never_known_wrongly_with_a_stuck_sync_app_and_the_backup() {
3926 let w = World::new("stuck");
3927 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
3928 let mut marked: Vec<(String, i64)> = Vec::new();
3929 for minute in 0..90 {
3930 let now = DUE + minute * 60;
3931 if minute % 10 == 0 {
3932 marked.push((format!("dose@{minute}"), minute));
3933 let record: String = marked.iter().map(|(d, m)| format!("\"{d}\" = {m}\n")).collect();
3934 w.desk.write("health-state.toml", &format!("[taken]\n{record}"));
3935 }
3936 w.desk.exchange(&w.server, now);
3937 crate::lease::renew(&w.server, &KEY, "health", &w.desk.id, now, now, false, crate::lease::Rule::FollowsYou, crate::share::written(&w.desk.memory, &w.desk.id)).unwrap();
3938 w.up();
3939 let _ = w.pull(now + 20);
3940 w.phone.exchange(&w.phone_folder, now + 30);
3941 if let Some(claim) = crate::lease::claims(&w.phone_folder, &KEY, "health").into_iter().find(|c| c.computer == w.desk.id)
3942 && let Some(wrote) = claim.wrote
3943 && crate::share::heard(&w.phone.memory, &w.phone.id).complete(&w.desk.id, wrote)
3944 {
3945 let record = w.phone.read("health-state.toml");
3946 let claimed = (claim.renewed - DUE) / 60;
3947 for (dose, at) in marked.iter().filter(|(_, m)| *m < claimed) {
3948 assert!(record.contains(dose.as_str()), "minute {minute}: {dose} ({at}) taken as known, not there:\n{record}");
3949 }
3950 }
3951 }
3952 let record = w.phone.read("health-state.toml");
3953 for (dose, _) in &marked {
3954 assert!(record.contains(dose.as_str()), "{dose} never came:\n{record}");
3955 }
3956 let synced = std::fs::read(w.phone_folder.join(w.desk_round())).map(|b| b.len()).unwrap_or(0);
3958 assert!(synced < std::fs::read(w.server.join(w.desk_round())).unwrap().len());
3959 }
3960
3961 const QUICK: AlarmTimes = AlarmTimes { alone: Duration::from_secs(2), most: Duration::from_secs(5) };
3963
3964 #[test]
3973 fn a_doses_alarm_on_its_pull_knows_what_it_knew_with_the_sync_app_asked() {
3974 static ONE_PULL: Mutex<()> = Mutex::new(());
3975 let w = World::new("alarm-alone");
3976 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
3977 assert!(w.pull(DUE).problem.is_none());
3978 let mut marked: Vec<(String, i64)> = Vec::new();
3979 for minute in 0..16 {
3980 let now = DUE + minute * 60;
3981 if minute % 3 == 0 {
3982 marked.push((format!("dose@{minute}"), now));
3983 let record: String = marked.iter().map(|(d, at)| format!("\"{d}\" = {at}\n")).collect();
3984 w.desk.write("health-state.toml", &format!("[taken]\n{record}"));
3985 }
3986 w.desk.exchange(&w.server, now);
3987 w.up();
3988 let aside = w.snapshot();
3990 w.down();
3991 let _ = w.pull(now + 20);
3992 w.phone.exchange(&w.phone_folder, now + 25);
3993 let asked = (w.phone.read("health-state.toml"), w.phone.sees(&w.phone_folder, &w.desk, now + 30));
3994 w.restore(&aside);
3996 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder), "minute {minute}");
3997 let mut ask = || -> Duration { panic!("minute {minute}: the pull went through, the sync app is not asked") };
3998 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, now + 20, QUICK, Arc::new(move || now + 20), &mut ask, TEST);
3999 assert_eq!(came, AlarmNews::Pulled, "minute {minute}");
4000 w.phone.exchange(&w.phone_folder, now + 25);
4001 let alone = (w.phone.read("health-state.toml"), w.phone.sees(&w.phone_folder, &w.desk, now + 30));
4002 assert_eq!(alone.0, asked.0, "minute {minute}: the same answers here");
4003 for (dose, due) in &marked {
4004 let doubted = |peer: &Peer| !doubts_now(*due, now + 30, None, std::slice::from_ref(peer)).is_empty();
4005 assert_eq!(doubted(&alone.1), doubted(&asked.1), "minute {minute}, {dose}: the same doubt");
4006 assert!(alone.0.contains(dose.as_str()) || doubted(&alone.1), "minute {minute}: {dose}, marked on the desk, known not taken");
4007 }
4008 }
4009 assert!(std::fs::read(w.phone_folder.join(w.desk_round())).map_or(0, |b| b.len()) < std::fs::read(w.server.join(w.desk_round())).unwrap().len());
4011 }
4012
4013 #[test]
4022 fn a_doses_alarm_whose_pull_fails_asks_the_sync_app_and_never_knows_wrongly() {
4023 static ONE_PULL: Mutex<()> = Mutex::new(());
4024 let w = World::new("alarm-fails");
4025 assert!(w.find(DUE - 3_000).confirmed_for(&w.phone_folder));
4026 w.desk.exchange(&w.server, DUE - 3_600);
4027 w.desk.session(&w.server, |e| e.close(DUE - 3_590));
4028 assert!(w.pull(DUE - 3_000).problem.is_none());
4029 w.phone.exchange(&w.phone_folder, DUE - 2_990);
4030 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder), "the pull works");
4031 assert!(doubts_now(DUE, DUE + 10, None, &[w.phone.sees(&w.phone_folder, &w.desk, DUE + 10)]).is_empty(), "the desk closed before the dose: known");
4032 w.desk.session(&w.server, |e| e.start(DUE - 60));
4033 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4034 w.desk.exchange(&w.server, DUE + 30);
4035 w.fake.fault(|_| Some(Fault::Status(503)));
4037 let mut asked = 0;
4038 let mut ask = || {
4039 asked += 1;
4040 w.down();
4041 Duration::from_millis(200)
4042 };
4043 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4044 assert!(matches!(came, AlarmNews::SyncApp { ref heard } if heard.contains(&w.desk.id)) && asked == 1, "the pull failed: the sync app asked, once, the desk's entry brought: {came:?}");
4045 w.phone.exchange(&w.phone_folder, DUE + 65);
4046 let taken = w.phone.read("health-state.toml").contains("dose@0");
4047 let doubts = doubts_now(DUE, DUE + 70, None, &[w.phone.sees(&w.phone_folder, &w.desk, DUE + 70)]);
4048 assert!(taken || !doubts.is_empty(), "the dose marked on the desk is never known not taken: {doubts:?}");
4049 assert!(taken, "the sync app brought it");
4050 assert!(!State::load(&w.phone.memory).pulls_well(&w.phone_folder), "the next alarm asks the sync app first, as before");
4051 w.fake.clear_faults();
4053 assert!(w.pull(DUE + 180).problem.is_none());
4054 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder));
4055 }
4056
4057 #[test]
4062 fn two_doses_at_once_the_second_alarm_waits_for_the_first_ones_pull() {
4063 static ONE_PULL: Mutex<()> = Mutex::new(());
4064 let w = World::new("alarm-twice");
4065 assert!(w.find(DUE - 3_000).confirmed_for(&w.phone_folder));
4066 w.desk.session(&w.server, |e| e.close(DUE - 3_590));
4067 assert!(w.pull(DUE - 3_000).problem.is_none());
4068 w.phone.exchange(&w.phone_folder, DUE - 2_990);
4069 w.desk.session(&w.server, |e| e.start(DUE - 60));
4071 w.desk.write("health-state.toml", "[taken]\n\"a@0\" = 1\n\"b@0\" = 2\n");
4072 w.desk.exchange(&w.server, DUE + 10);
4073 w.fake.fault(once(|r| r.method == "PROPFIND", Fault::Late(Duration::from_millis(450))));
4075 let outcomes: Vec<(AlarmNews, Duration)> = std::thread::scope(|scope| {
4076 let alarm = |pause: u64| {
4077 let w = &w;
4078 scope.spawn(move || {
4079 std::thread::sleep(Duration::from_millis(pause));
4080 let begun = Instant::now();
4081 let mut ask = || -> Duration { panic!("the pull went through: the sync app is not asked") };
4082 (alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST), begun.elapsed())
4083 })
4084 };
4085 let (first, second) = (alarm(0), alarm(150));
4086 vec![first.join().unwrap(), second.join().unwrap()]
4087 });
4088 assert!(outcomes.iter().all(|(came, _)| *came == AlarmNews::Pulled), "{outcomes:?}");
4089 assert!(outcomes[1].1 >= Duration::from_millis(250), "the second waited for the first one's pull: {outcomes:?}");
4090 w.phone.exchange(&w.phone_folder, DUE + 70);
4091 let record = w.phone.read("health-state.toml");
4092 assert!(record.contains("a@0") && record.contains("b@0"), "{record}");
4093 let seen = w.phone.sees(&w.phone_folder, &w.desk, DUE + 70);
4094 assert!(seen.said.as_ref().is_some_and(|s| s.working), "the desk known in use, from the pull: {seen:?}");
4095 }
4096
4097 #[test]
4102 fn an_alarm_while_the_windows_pull_runs_pulls_again_after_it() {
4103 static ONE_PULL: Mutex<()> = Mutex::new(());
4104 let w = World::new("alarm-window");
4105 assert!(w.find(DUE - 3_000).confirmed_for(&w.phone_folder));
4106 w.desk.session(&w.server, |e| e.close(DUE - 3_590));
4107 assert!(w.pull(DUE - 3_000).problem.is_none());
4108 w.phone.exchange(&w.phone_folder, DUE - 2_990);
4109 w.desk.session(&w.server, |e| e.start(DUE - 60));
4110 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4111 w.desk.exchange(&w.server, DUE + 10);
4112 w.fake.fault(once(|r| r.method == "PROPFIND", Fault::Late(Duration::from_millis(450))));
4113 let came = std::thread::scope(|scope| {
4114 let window = scope.spawn(|| {
4116 let _one = ONE_PULL.lock().unwrap();
4117 pull_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE + 50, TEST)
4118 });
4119 std::thread::sleep(Duration::from_millis(200));
4120 let mut ask = || -> Duration { panic!("a pull went through: the sync app is not asked") };
4121 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 61), &mut ask, TEST);
4122 assert!(window.join().unwrap().problem.is_none());
4123 came
4124 });
4125 assert_eq!(came, AlarmNews::Pulled);
4126 assert_eq!(State::load(&w.phone.memory).last, DUE + 61, "pulled again, after the window's");
4127 w.phone.exchange(&w.phone_folder, DUE + 70);
4128 assert!(w.phone.read("health-state.toml").contains("dose@0"));
4129 }
4130
4131 #[test]
4136 fn an_alarm_whose_pull_does_not_go_through_in_time_asks_the_sync_app() {
4137 static ONE_PULL: Mutex<()> = Mutex::new(());
4138 let w = World::new("alarm-slow");
4139 assert!(w.find(DUE - 3_000).confirmed_for(&w.phone_folder));
4140 w.desk.session(&w.server, |e| e.close(DUE - 3_590));
4141 assert!(w.pull(DUE - 3_000).problem.is_none());
4142 w.phone.exchange(&w.phone_folder, DUE - 2_990);
4143 w.desk.session(&w.server, |e| e.start(DUE - 60));
4144 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4145 w.desk.exchange(&w.server, DUE + 10);
4146 w.fake.fault(|_| Some(Fault::Late(Duration::from_secs(6))));
4148 let mut asked = 0;
4149 let mut ask = || {
4150 asked += 1;
4151 w.down();
4152 Duration::from_millis(300)
4153 };
4154 let begun = Instant::now();
4155 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4156 let waited = begun.elapsed();
4157 assert!(matches!(came, AlarmNews::SyncApp { .. }) && asked == 1, "not through in time: the sync app asked, once: {came:?}");
4158 assert!(waited < QUICK.most, "never past the alarm's wait: {waited:?}");
4159 w.phone.exchange(&w.phone_folder, DUE + 65);
4160 assert!(w.phone.read("health-state.toml").contains("dose@0"), "what the sync app brought is read");
4161 w.fake.clear_faults();
4162 }
4163
4164 fn alarm_scene(name: &str, pulled_at: i64, opens: bool) -> World {
4169 let w = World::new(name);
4170 assert!(w.find(DUE - 3_000).confirmed_for(&w.phone_folder));
4171 w.desk.exchange(&w.server, DUE - 3_600);
4172 w.desk.session(&w.server, |e| e.close(DUE - 3_590));
4173 assert!(w.pull(pulled_at).problem.is_none());
4174 w.phone.exchange(&w.phone_folder, DUE - 2_990);
4175 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder));
4176 if opens {
4177 w.desk.session(&w.server, |e| e.start(DUE - 60));
4178 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4179 w.desk.exchange(&w.server, DUE + 30);
4180 }
4181 w
4182 }
4183
4184 #[test]
4190 fn an_alarm_never_takes_a_pull_from_before_the_clock_was_set_back() {
4191 static ONE_PULL: Mutex<()> = Mutex::new(());
4192 let w = alarm_scene("alarm-clock", DUE + 600, true);
4193 w.fake.fault(|_| Some(Fault::Status(503)));
4194 let mut asked = 0;
4195 let mut ask = || {
4196 asked += 1;
4197 Duration::from_millis(200)
4198 };
4199 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4200 assert_eq!((&came, asked), (&AlarmNews::SyncApp { heard: BTreeSet::new() }, 1), "the old pull not taken: pulled again, refused; the sync app asked, nothing came");
4201 w.phone.exchange(&w.phone_folder, DUE + 65);
4202 assert!(!w.phone.read("health-state.toml").contains("dose@0"));
4203 let peer = w.phone.sees(&w.phone_folder, &w.desk, DUE + 70);
4204 assert!(doubts_now(DUE, DUE + 70, None, std::slice::from_ref(&peer)).is_empty(), "what was read before the alarm says the desk closed");
4205 assert!(!sioul_core::health::doubts_unread(DUE, DUE + 70, None, std::slice::from_ref(&peer), DUE + 60, &BTreeSet::new()).is_empty(), "nothing read since the alarm: the doubt said, never known");
4206 let state = State::load(&w.phone.memory);
4207 assert!(!state.pulls_well(&w.phone_folder), "the refused pull is the later one, whatever the clock said: {state:?}");
4208 let pace = Pace { phone: true, shown: false, in_use: false, urgent: false, step: 5 * 60 };
4209 assert!(due(&State { tried: DUE + 600, ..state }, DUE + 120, pace), "tried \"later\" than now: due at once");
4210 w.fake.clear_faults();
4211 }
4212
4213 #[test]
4220 fn an_alarm_that_reads_no_news_says_the_doubt_and_one_that_pulls_does_not() {
4221 static ONE_PULL: Mutex<()> = Mutex::new(());
4222 let w = alarm_scene("alarm-nothing", DUE - 3_000, true);
4223 w.fake.fault(|_| Some(Fault::Status(503)));
4224 let mut ask = || Duration::from_millis(200);
4225 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4226 assert_eq!(came, AlarmNews::SyncApp { heard: BTreeSet::new() });
4227 w.phone.exchange(&w.phone_folder, DUE + 65);
4228 let peer = w.phone.sees(&w.phone_folder, &w.desk, DUE + 70);
4229 assert!(!w.phone.read("health-state.toml").contains("dose@0"));
4230 let doubts = sioul_core::health::doubts_unread(DUE, DUE + 70, None, std::slice::from_ref(&peer), DUE + 60, &BTreeSet::new());
4231 assert!(matches!(doubts.as_slice(), [Doubt::Unheard { closed: true, until, .. }] if *until == DUE - 3_590), "the desk as last heard, closed: {doubts:?}");
4232 w.fake.clear_faults();
4233 static ANOTHER: Mutex<()> = Mutex::new(());
4235 let quiet = alarm_scene("alarm-pulled", DUE - 3_000, false);
4236 let mut never = || -> Duration { panic!("the pull went through: the sync app is not asked") };
4237 let came = alarm_news_with(&quiet.phone.memory, &quiet.phone_folder, &quiet.phone.id, KEY, Some(quiet.fake.login()), &ANOTHER, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut never, TEST);
4238 assert_eq!(came, AlarmNews::Pulled);
4239 quiet.phone.exchange(&quiet.phone_folder, DUE + 65);
4240 assert!(doubts_now(DUE, DUE + 70, None, &[quiet.phone.sees(&quiet.phone_folder, &quiet.desk, DUE + 70)]).is_empty(), "pulled since the alarm, the desk closed: known");
4241 }
4242
4243 fn alarm_verdict(w: &World, came: &AlarmNews, since: i64, at: i64) -> (bool, bool, Vec<Doubt>) {
4247 w.phone.exchange(&w.phone_folder, at);
4248 let taken = w.phone.read("health-state.toml").contains("dose@0");
4249 let peer = w.phone.sees(&w.phone_folder, &w.desk, at + 5);
4250 let doubts = match came {
4251 AlarmNews::SyncApp { heard } => sioul_core::health::doubts_unread(DUE, at + 5, None, std::slice::from_ref(&peer), since, heard),
4252 AlarmNews::Pulled => doubts_now(DUE, at + 5, None, std::slice::from_ref(&peer)),
4253 };
4254 (!taken && doubts.is_empty(), taken, doubts)
4255 }
4256
4257 #[test]
4263 fn an_alarm_doubts_each_device_whose_own_entry_did_not_come() {
4264 static ONE_PULL: Mutex<()> = Mutex::new(());
4265 let w = alarm_scene("alarm-unrelated", DUE - 3_000, true);
4266 w.fake.fault(|_| Some(Fault::Status(503)));
4267 let notes = format!("{}.toml", w.desk.id);
4268 let mut ask = || {
4269 std::fs::copy(w.server.join(¬es), w.phone_folder.join(¬es)).unwrap();
4271 std::fs::File::options().write(true).open(w.phone_folder.join(¬es)).unwrap().set_modified(std::time::SystemTime::now()).unwrap();
4272 Duration::from_millis(200)
4273 };
4274 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4275 w.fake.clear_faults();
4276 assert_eq!(came, AlarmNews::SyncApp { heard: BTreeSet::new() }, "the notes are no news of the desk's session");
4277 let (wrong, taken, doubts) = alarm_verdict(&w, &came, DUE + 60, DUE + 65);
4278 assert!(!wrong && !taken && matches!(doubts.as_slice(), [Doubt::Unheard { closed: true, .. }]), "{doubts:?}");
4279 static AGAIN: Mutex<()> = Mutex::new(());
4281 let other = alarm_scene("alarm-entry-came", DUE - 3_000, true);
4282 other.fake.fault(|_| Some(Fault::Status(503)));
4283 let entry = format!("devices/{}.device", other.desk.id);
4284 let mut ask = || {
4285 std::fs::copy(other.server.join(&entry), other.phone_folder.join(&entry)).unwrap();
4286 Duration::from_millis(200)
4287 };
4288 let came = alarm_news_with(&other.phone.memory, &other.phone_folder, &other.phone.id, KEY, Some(other.fake.login()), &AGAIN, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4289 other.fake.clear_faults();
4290 assert!(matches!(came, AlarmNews::SyncApp { ref heard } if heard.contains(&other.desk.id)), "{came:?}");
4291 let (wrong, _, doubts) = alarm_verdict(&other, &came, DUE + 60, DUE + 65);
4292 assert!(!wrong && matches!(doubts.as_slice(), [Doubt::Working { .. }] | [Doubt::Coming { .. }]), "the desk in use, as its entry says: {doubts:?}");
4293 }
4294
4295 #[test]
4299 fn a_clock_set_back_never_hides_the_alarms_own_pull() {
4300 static ONE_PULL: Mutex<()> = Mutex::new(());
4301 let w = alarm_scene("alarm-forward-back", DUE - 3_000, false);
4302 assert!(w.pull(DUE + 600).problem.is_none());
4303 let mut never = || -> Duration { panic!("the alarm's own pull went through: the sync app is not asked") };
4304 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut never, TEST);
4305 assert_eq!(came, AlarmNews::Pulled);
4306 assert_eq!(State::load(&w.phone.memory).last, DUE + 60, "the time from before the set-back gives way");
4307 let (_, _, doubts) = alarm_verdict(&w, &came, DUE + 60, DUE + 65);
4308 assert!(doubts.is_empty(), "{doubts:?}");
4309 }
4310
4311 #[test]
4315 fn a_round_found_again_grown_stays_left_in_the_merge() {
4316 let w = World::new("left-regrown");
4317 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4318 assert!(w.pull(DUE + 10).problem.is_none());
4319 let disk = State::load(&w.phone.memory);
4320 State { left: BTreeMap::new(), tried: DUE + 20, last: DUE + 20, ..disk.clone() }.write(&w.phone.memory).unwrap();
4322 let loaded: BTreeMap<String, u64> = [("r-1.jsonl".to_string(), 100)].into();
4324 let ours = State { tried: DUE + 21, last: DUE + 21, left: [("r-1.jsonl".to_string(), 200)].into(), ..disk };
4325 ours.save_pulled(&w.phone.memory, DUE + 25, &loaded).unwrap();
4326 let merged = State::load(&w.phone.memory);
4327 assert_eq!(merged.left.get("r-1.jsonl"), Some(&200), "{:?}", merged.left);
4328 assert!(!merged.pulls_well(&w.phone_folder));
4329 }
4330
4331 #[test]
4337 fn only_a_newer_entry_or_health_claim_is_news_of_a_device() {
4338 static ONE_PULL: Mutex<()> = Mutex::new(());
4339 let w = alarm_scene("alarm-notices", DUE - 3_000, true);
4340 crate::lease::renew(&w.server, &KEY, "notices", &w.desk.id, DUE + 20, DUE + 20, false, crate::lease::Rule::FollowsYou, crate::share::written(&w.desk.memory, &w.desk.id)).unwrap();
4341 w.fake.fault(|_| Some(Fault::Status(503)));
4342 let lease = format!("leases/notices/{}.lease", w.desk.id);
4343 let mut ask = || {
4344 let to = w.phone_folder.join(&lease);
4345 std::fs::create_dir_all(to.parent().unwrap()).unwrap();
4346 std::fs::copy(w.server.join(&lease), &to).unwrap();
4347 Duration::from_millis(200)
4348 };
4349 let came = alarm_news_with(&w.phone.memory, &w.phone_folder, &w.phone.id, KEY, Some(w.fake.login()), &ONE_PULL, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4350 w.fake.clear_faults();
4351 assert_eq!(came, AlarmNews::SyncApp { heard: BTreeSet::new() }, "a notices claim is no news of the session");
4352 let (wrong, _, doubts) = alarm_verdict(&w, &came, DUE + 60, DUE + 65);
4353 assert!(!wrong && !doubts.is_empty(), "{doubts:?}");
4354 static AGAIN: Mutex<()> = Mutex::new(());
4356 let touched = alarm_scene("alarm-touched", DUE - 3_000, true);
4357 touched.fake.fault(|_| Some(Fault::Status(503)));
4358 let entry = touched.phone_folder.join(format!("devices/{}.device", touched.desk.id));
4359 let mut ask = || {
4360 std::fs::File::options().write(true).open(&entry).unwrap().set_modified(std::time::SystemTime::now()).unwrap();
4361 Duration::from_millis(200)
4362 };
4363 let came = alarm_news_with(&touched.phone.memory, &touched.phone_folder, &touched.phone.id, KEY, Some(touched.fake.login()), &AGAIN, DUE + 60, QUICK, Arc::new(|| DUE + 60), &mut ask, TEST);
4364 touched.fake.clear_faults();
4365 assert_eq!(came, AlarmNews::SyncApp { heard: BTreeSet::new() }, "the same entry at a new time is no news");
4366 let (wrong, _, doubts) = alarm_verdict(&touched, &came, DUE + 60, DUE + 65);
4367 assert!(!wrong && !doubts.is_empty(), "{doubts:?}");
4368 }
4369
4370 #[test]
4376 fn two_pulls_at_once_keep_what_each_found() {
4377 let w = World::new("two-pulls");
4378 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4379 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4380 w.desk.exchange(&w.server, DUE + 10);
4381 w.fake.fault(once(|r| r.method == "PROPFIND", Fault::Late(Duration::from_millis(450))));
4382 std::thread::scope(|scope| {
4383 let slow = scope.spawn(|| pull_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE + 50, TEST));
4384 std::thread::sleep(Duration::from_millis(100));
4385 ROUND_CAP.with(|cap| cap.set(Some(64)));
4387 let quick = pull_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE + 51, TEST);
4389 ROUND_CAP.with(|cap| cap.set(None));
4390 assert!(quick.problem.is_none() && quick.left >= 1, "{quick:?}");
4391 let slow = slow.join().unwrap();
4392 assert!(slow.problem.is_none(), "{slow:?}");
4393 });
4394 let state = State::load(&w.phone.memory);
4395 assert_eq!((state.tried, state.last), (DUE + 51, DUE + 51), "the later begun says when");
4396 assert!(!state.left.is_empty() && !state.pulls_well(&w.phone_folder), "found left by the other, never erased by the slower one's save: {:?}", state.left);
4397 assert!(w.pull(DUE + 120).problem.is_none());
4399 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder));
4400 }
4401
4402 #[test]
4407 fn a_round_too_large_for_the_pull_is_said_and_left_to_the_sync_app() {
4408 let w = World::new("too-large");
4409 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4410 w.desk.write("health-state.toml", "[taken]\n\"dose@0\" = 1\n");
4411 w.desk.exchange(&w.server, DUE + 10);
4412 ROUND_CAP.with(|cap| cap.set(Some(64)));
4413 let pulled = w.pull(DUE + 20);
4414 assert!(pulled.problem.is_none() && pulled.left >= 1, "{pulled:?}");
4415 let state = State::load(&w.phone.memory);
4416 assert!(!state.left.is_empty() && !state.pulls_well(&w.phone_folder) && !state.went_through_since(&w.phone_folder, DUE + 20, DUE + 20), "{:?}", state.left);
4417 w.down();
4419 let pulled = w.pull(DUE + 100);
4420 ROUND_CAP.with(|cap| cap.set(None));
4421 assert_eq!(pulled.left, 0, "{pulled:?}");
4422 assert!(State::load(&w.phone.memory).pulls_well(&w.phone_folder));
4423 }
4424
4425 #[test]
4431 fn the_phones_entry_put_back_older_on_the_server_is_sent_again() {
4432 let w = World::new("entry-back");
4433 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4434 let entry = format!("devices/{}.device", w.phone.id);
4435 let older = std::fs::read(w.phone_folder.join(&entry)).unwrap();
4436 w.phone.session(&w.phone_folder, |e| e.close(DUE + 10));
4437 assert!(phone_sends(&w, DUE + 20, false).problem.is_none());
4438 let current = std::fs::read(w.phone_folder.join(&entry)).unwrap();
4439 assert_eq!(std::fs::read(w.server.join(&entry)).unwrap(), current);
4440 assert_ne!(older.len(), current.len());
4441 std::fs::write(w.server.join(&entry), &older).unwrap();
4443 w.desk.exchange(&w.server, DUE + 30);
4444 assert!(w.pull(DUE + 40).problem.is_none());
4445 assert_eq!(State::load(&w.phone.memory).own_entry.map(|(size, _)| size), Some(older.len() as u64));
4446 assert!(phone_sends(&w, DUE + 50, false).problem.is_none());
4447 assert_eq!(std::fs::read(w.server.join(&entry)).unwrap(), current, "sent over the copy put back");
4448 }
4449
4450 #[test]
4456 fn found_at_night_the_first_pull_comes_at_once() {
4457 let w = World::new("night");
4458 w.desk.press(true, DUE + 30);
4459 w.desk.exchange(&w.server, DUE + 60);
4460 w.up();
4461 let in_use = |now: i64| crate::devices::all(&w.phone_folder, &KEY).0.iter().any(|e| e.id == w.desk.id && e.working && !e.left && now - e.exported <= sioul_core::health::FRESH + sioul_core::health::SKEW);
4462 let night = |now: i64| Pace { phone: true, shown: false, in_use: in_use(now), urgent: false, step: 15 * 60 };
4463 assert!(!in_use(DUE + 70), "the stale folder tells the desk not in use");
4464 let state = w.find(DUE + 70);
4465 assert!(state.confirmed_for(&w.phone_folder));
4466 attach(&w.phone_folder, &w.phone.memory);
4467 assert!(due(&State::load(&w.phone.memory), DUE + 80, night(DUE + 80)), "found: pulled at once, whatever the hour");
4468 let pulled = w.pull(DUE + 80);
4469 assert!(pulled.problem.is_none() && pulled.listed > 0 && pulled.fetched > 0, "{pulled:?}");
4470 assert!(in_use(DUE + 90), "the desk known in use, from the server");
4471 w.phone.exchange(&w.phone_folder, DUE + 90);
4472 assert!(w.phone.dnd().is_some_and(|p| p.on), "its do-not-disturb too");
4473 let state = State::load(&w.phone.memory);
4474 assert!(!due(&state, DUE + 100, night(DUE + 100)), "not twice in a minute");
4475 w.desk.exchange(&w.server, DUE + 130);
4476 assert!(w.pull(DUE + 140).problem.is_none(), "once a minute while the desk is in use, at night too");
4477 w.desk.session(&w.server, |e| e.close(DUE + 150));
4479 assert!(w.pull(DUE + 200).problem.is_none());
4480 let state = State::load(&w.phone.memory);
4481 assert!(!in_use(DUE + 260) && !due(&state, DUE + 260, night(DUE + 260)));
4482 assert!(due(&state, DUE + 200 + 15 * 60, night(DUE + 200 + 15 * 60)), "a device opened at night is noticed within the step's fifteen minutes");
4483 }
4484
4485 #[test]
4486 fn the_pace_of_pulls() {
4487 let state = State { confirmed: 1_000, tried: 1_000, ..State::default() };
4488 let pace = |phone, shown, in_use, urgent| Pace { phone, shown, in_use, urgent, step: 5 * 60 };
4489 assert!(due(&state, 1_060, pace(false, false, false, false)) && !due(&state, 1_059, pace(false, false, false, false)));
4491 assert!(due(&state, 1_060, pace(true, true, false, false)));
4492 assert!(due(&state, 1_060, pace(true, false, true, false)));
4493 assert!(!due(&state, 1_120, pace(true, false, false, false)) && due(&state, 1_270, pace(true, false, false, false)));
4495 assert!(due(&state, 1_020, pace(true, false, false, true)) && !due(&state, 1_010, pace(true, false, false, true)));
4497 assert!(due(&State { confirmed: 1_005, ..state }, 1_006, pace(true, false, false, false)));
4499 }
4500
4501 #[test]
4504 fn the_pull_alone_is_trusted_only_while_it_goes_through() {
4505 let folder = Path::new("/storage/emulated/0/Documents/Sioul");
4506 let well = State { folder: shown(folder), url: "https://cloud.example.org/remote.php/dav/files/jane/Documents/Sioul/".into(), confirmed: 1_000, last: 1_200, tried: 1_200, ..State::default() };
4507 assert!(well.pulls_well(folder));
4508 assert!(!State { on: Some(false), ..well.clone() }.pulls_well(folder), "switched off here");
4509 assert!(State { on: Some(true), ..well.clone() }.pulls_well(folder));
4510 assert!(!State { confirmed: 0, ..well.clone() }.pulls_well(folder), "not confirmed");
4511 assert!(!State { last: 0, ..well.clone() }.pulls_well(folder), "never pulled yet");
4512 assert!(!State { said: "network:cloud.example.org".into(), ..well.clone() }.pulls_well(folder), "the last pull failed");
4513 assert!(!State { said: "not-found:cloud.example.org".into(), ..well.clone() }.pulls_well(folder), "looked for again, not found");
4514 assert!(!State { mode: MIRROR.into(), ..well.clone() }.pulls_well(folder), "Sioul keeping the folder: no sync app to ask anyway");
4515 assert!(!well.pulls_well(Path::new("/storage/emulated/0/Documents/Other")), "another folder");
4516 }
4517
4518 #[test]
4524 fn a_steps_requests_go_through_one_connection() {
4525 use std::io::{BufRead, BufReader, Write};
4526 use std::sync::atomic::{AtomicUsize, Ordering};
4527 let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
4528 let base = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
4529 let (connections, requests) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
4530 let (counted, served) = (Arc::clone(&connections), Arc::clone(&requests));
4531 std::thread::spawn(move || {
4533 for stream in listener.incoming().filter_map(Result::ok) {
4534 counted.fetch_add(1, Ordering::SeqCst);
4535 let served = Arc::clone(&served);
4536 std::thread::spawn(move || {
4537 let mut reader = BufReader::new(stream.try_clone().unwrap());
4538 let mut stream = stream;
4539 loop {
4540 let mut line = String::new();
4541 if reader.read_line(&mut line).unwrap_or(0) == 0 {
4542 return;
4543 }
4544 let mut length = 0;
4545 loop {
4546 let mut header = String::new();
4547 reader.read_line(&mut header).unwrap();
4548 if header.trim().is_empty() {
4549 break;
4550 }
4551 if let Some((name, value)) = header.split_once(':')
4552 && name.eq_ignore_ascii_case("content-length")
4553 {
4554 length = value.trim().parse().unwrap();
4555 }
4556 }
4557 let mut body = vec![0; length];
4558 reader.read_exact(&mut body).unwrap();
4559 served.fetch_add(1, Ordering::SeqCst);
4560 let (status, answer) = match line.split_whitespace().next() {
4561 Some("PROPFIND") => ("207 Multi-Status", super::fake::multistatus("")),
4562 Some("GET") => ("200 OK", b"version = 1\n".to_vec()),
4563 _ => ("201 Created", Vec::new()),
4564 };
4565 let head = format!("HTTP/1.1 {status}\r\nContent-Length: {}\r\nETag: \"e\"\r\n\r\n", answer.len());
4566 if stream.write_all(head.as_bytes()).and_then(|()| stream.write_all(&answer)).is_err() {
4567 return;
4568 }
4569 }
4570 });
4571 }
4572 });
4573 let login = Login { account: "cloud".into(), url: format!("{base}/remote.php/dav/"), user: "jane".into(), password: Some(PASSWORD.into()) };
4574 let folder = format!("{base}/remote.php/dav/files/jane/Documents/Sioul/");
4575 let pull = Server::new(&login, TEST).unwrap();
4577 pull.list(&folder).unwrap();
4578 assert!(pull.small(&format!("{folder}seal.toml")).unwrap().is_some());
4579 let send = Server::new(&login, Limits { wait: Duration::from_secs(2), stall: Duration::from_secs(3), ..TEST }).unwrap();
4581 assert!(send.stat(&format!("{folder}devices/")).unwrap().is_none_or(|item| item.etag.is_empty()));
4582 let (status, _) = send.put_file(&format!("{folder}phone.toml"), b"read = 1\n".to_vec(), None, 1, &sha1_hex(b"read = 1\n")).unwrap();
4583 assert_eq!(status, 201);
4584 let again = Server::new(&login, TEST).unwrap();
4586 again.list(&folder).unwrap();
4587 assert_eq!(requests.load(Ordering::SeqCst), 5);
4588 assert_eq!(connections.load(Ordering::SeqCst), 1, "one connection for the whole step");
4589 }
4590
4591 #[test]
4592 fn where_the_folder_may_be_on_the_server() {
4593 let found = |folder: &str, configs: &[&str]| places(Path::new(folder), &configs.iter().map(|c| c.to_string()).collect::<Vec<_>>());
4594 assert_eq!(found("/storage/emulated/0/Documents/Sioul", &[]), ["Documents/Sioul", "Sioul", "0/Documents/Sioul"]);
4596 assert_eq!(found("/sdcard/Documents/Sioul/", &[])[0], "Documents/Sioul");
4597 assert_eq!(found("/storage/emulated/0/Android/media/com.nextcloud.client/nextcloud/jane@murena.io/Shared/Sioul", &[])[0], "Shared/Sioul");
4599 let cfg = "[Accounts]\n0\\Folders\\1\\localPath=/home/me/Nextcloud/\n0\\Folders\\1\\targetPath=/\n0\\Folders\\2\\localPath=/home/me/Work\n0\\Folders\\2\\targetPath=/Projects/Work\n0\\url=https://cloud.example.org\n";
4601 assert_eq!(found("/home/me/Nextcloud/Documents/Sioul", &[cfg])[0], "Documents/Sioul");
4602 assert_eq!(found("/home/me/Work/Sioul", &[cfg])[0], "Projects/Work/Sioul");
4603 assert_eq!(found("/home/me/Workshop/Sioul", &[cfg]), ["Sioul", "Workshop/Sioul", "me/Workshop/Sioul"], "not inside /home/me/Work");
4604 assert_eq!(encode_path("Mes documents/Sioul"), "Mes%20documents/Sioul");
4606 assert_eq!(host_of("https://jane@murena.io/remote.php/dav/"), "murena.io");
4607 assert_eq!(kind_of("devices/0f8fad5b-d9cb-469f-a165-70867728950e.device").map(|k| k.0), Some(Kind::Device));
4608 assert_eq!(kind_of("0f8fad5b-d9cb-469f-a165-70867728950e-12.jsonl").map(|k| k.0), Some(Kind::Round));
4609 assert_eq!(kind_of("0f8fad5b-d9cb-469f-a165-70867728950e-12 (conflicted copy).jsonl"), None);
4610 assert_eq!(kind_of("leases/health/desktop.lease").map(|k| k.0), Some(Kind::Claim));
4611 }
4612
4613 #[test]
4614 fn found_on_the_account_given_by_hand_or_not_at_all() {
4615 let w = World::new("given");
4616 State::choose(&w.phone.memory, |s| s.given = "Documents/Sioul".into()).unwrap();
4618 let state = find_with(&w.phone.memory, &w.phone_folder, &[w.fake.login()], &[], DUE, TEST);
4619 assert!(state.confirmed_for(&w.phone_folder), "{state:?}");
4620 State::choose(&w.phone.memory, |s| s.given = w.fake.folder_url("Documents/Sioul")).unwrap();
4622 assert!(find_with(&w.phone.memory, &w.phone_folder, &[w.fake.login()], &[], DUE + 1, TEST).confirmed_for(&w.phone_folder));
4623 State::choose(&w.phone.memory, |s| s.given = "Elsewhere".into()).unwrap();
4625 let state = find_with(&w.phone.memory, &w.phone_folder, &[w.fake.login()], &places(&w.phone_folder, &[]), DUE + 2, TEST);
4626 assert!(state.said.starts_with("not-found:") && !state.confirmed_for(&w.phone_folder), "{state:?}");
4627 State::choose(&w.phone.memory, |s| s.given.clear()).unwrap();
4628 let wrong = Login { password: Some("wrong".into()), ..w.fake.login() };
4630 assert!(find_with(&w.phone.memory, &w.phone_folder, &[wrong], &places(&w.phone_folder, &[]), DUE + 3, TEST).said.starts_with("login:"));
4631 let none = Login { password: None, ..w.fake.login() };
4632 assert_eq!(find_with(&w.phone.memory, &w.phone_folder, &[none], &[], DUE + 4, TEST).said, "no-password:cloud");
4633 assert_eq!(find_with(&w.phone.memory, &w.phone_folder, &[], &[], DUE + 5, TEST).said, "no-account");
4634 let other = Login { url: format!("{}/dav/calendars/jane/", w.fake.base), ..w.fake.login() };
4636 assert_eq!(find_with(&w.phone.memory, &w.phone_folder, &[other], &places(&w.phone_folder, &[]), DUE + 6, TEST).said, "no-account");
4637 let state = State::load(&w.phone.memory);
4639 assert!(!state.look_due(&w.phone_folder, DUE + 7) && state.look_due(&w.phone_folder, DUE + 6 + LOOK_AGAIN));
4640 State::choose(&w.phone.memory, |s| s.asked = DUE + 8).unwrap();
4641 assert!(State::load(&w.phone.memory).look_due(&w.phone_folder, DUE + 9));
4642 }
4643
4644 #[test]
4645 fn your_choices_are_never_written_over_by_a_pull() {
4646 let w = World::new("choices");
4647 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4648 let held = State::load(&w.phone.memory);
4650 State::choose(&w.phone.memory, |s| s.on = Some(false)).unwrap();
4651 held.save(&w.phone.memory).unwrap();
4652 assert_eq!(State::load(&w.phone.memory).on, Some(false));
4653 State::choose(&w.phone.memory, |s| s.on = None).unwrap();
4654 assert!(State::load(&w.phone.memory).fetching());
4655 }
4656
4657 struct Pair {
4662 base: PathBuf,
4663 fake: Arc<Fake>,
4664 desk: Device,
4665 desk_folder: PathBuf,
4666 phone: Device,
4667 phone_folder: PathBuf,
4668 }
4669
4670 impl Pair {
4671 fn new(name: &str) -> Pair {
4672 let base = scratch(name);
4673 let files = base.join("cloud");
4674 std::fs::create_dir_all(&files).unwrap();
4675 let fake = Fake::start(&files);
4676 let desk = Device::new(&base, "desk", COMPUTER);
4677 let phone = Device::new(&base, "phone", PHONE);
4678 let (desk_folder, phone_folder) = (mirror_of(&desk.memory), mirror_of(&phone.memory));
4679 let login = fake.login();
4680 let opened = open_with(&login, "Documents/Sioul", TEST).unwrap();
4682 assert!(opened.seal.is_none(), "nothing there yet");
4683 put(&desk_folder, "seal.toml", SEAL);
4684 create_with(&login, &opened, &desk_folder, TEST).unwrap();
4685 begin(&desk.memory, &desk_folder, &login, &opened, DUE - 7_300).unwrap();
4686 assert_eq!(create_with(&login, &opened, &desk_folder, TEST), Err("sealed-meanwhile".to_string()));
4688 let joined = open_with(&login, "/Documents/Sioul/", TEST).unwrap();
4690 assert_eq!(joined.seal.as_deref(), Some(SEAL.as_bytes()));
4691 put_bytes(&phone_folder, "seal.toml", joined.seal.as_deref().unwrap());
4692 begin(&phone.memory, &phone_folder, &login, &joined, DUE - 7_250).unwrap();
4693 let pair = Pair { base, fake, desk, desk_folder, phone, phone_folder };
4694 pair.desk.session(&pair.desk_folder, |e| e.start(DUE - 7_200));
4695 pair.phone.session(&pair.phone_folder, |e| e.start(DUE - 7_200));
4696 pair.turn(true, DUE - 7_100);
4697 pair.turn(false, DUE - 7_090);
4698 pair.turn(true, DUE - 7_080);
4699 pair
4700 }
4701
4702 fn device(&self, desk: bool) -> (&Device, &Path) {
4703 if desk { (&self.desk, &self.desk_folder) } else { (&self.phone, &self.phone_folder) }
4704 }
4705
4706 fn step(&self, desk: bool, now: i64, whole: bool) -> Pulled {
4707 let (device, folder) = self.device(desk);
4708 step_with(&device.memory, folder, &device.id, &self.fake.login(), now, whole, TEST)
4709 }
4710
4711 fn turn(&self, desk: bool, now: i64) -> Outcome {
4713 let before = self.step(desk, now, true);
4714 assert!(before.problem.is_none(), "{before:?}");
4715 let outcome = self.device(desk).0.exchange(self.device(desk).1, now);
4716 let after = self.step(desk, now, false);
4717 assert!(after.problem.is_none(), "{after:?}");
4718 outcome
4719 }
4720 }
4721
4722 impl Drop for Pair {
4723 fn drop(&mut self) {
4724 let _ = std::fs::remove_dir_all(&self.base);
4725 }
4726 }
4727
4728 #[test]
4729 fn two_devices_without_a_sync_app_share_through_the_server_alone() {
4730 let p = Pair::new("mirror");
4731 p.desk.press(true, DUE + 10);
4733 p.turn(true, DUE + 20);
4734 p.turn(false, DUE + 30);
4735 let press = p.phone.dnd().expect("the desk's press");
4736 assert!(press.on && press.from == p.desk.id, "{press:?}");
4737 p.phone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n");
4739 p.turn(false, DUE + 40);
4740 p.turn(true, DUE + 50);
4741 assert!(p.desk.read("health-state.toml").contains("levo@1800000000"));
4742 p.phone.press(false, DUE + 60);
4744 p.turn(false, DUE + 70);
4745 p.turn(true, DUE + 80);
4746 assert!(!p.desk.dnd().unwrap().on);
4747 let phone = p.desk.sees(&p.desk_folder, &p.phone, DUE + 90);
4749 assert!(phone.complete && phone.said.as_ref().is_some_and(|s| s.working && s.exported >= DUE + 70), "{phone:?}");
4750 assert!(doubts_now(DUE - 100, DUE + 90, None, &[phone]).is_empty(), "a dose due before its last export is known");
4751 assert_eq!(pull_with(&p.phone.memory, &p.phone_folder, &p.phone.id, &p.fake.login(), DUE + 95, TEST).problem.as_deref(), Some("mirror"));
4753 assert!(attach(&p.phone_folder, &p.phone.memory).is_none());
4754 }
4755
4756 #[test]
4757 fn a_device_never_writes_another_devices_file() {
4758 let p = Pair::new("own");
4759 for minute in 0..6 {
4760 for desk in [true, false] {
4761 let device = p.device(desk).0;
4762 if minute % 2 == 0 {
4763 device.press(minute % 4 == 0, DUE + minute * 60 + 1);
4764 }
4765 p.fake.forget_seen();
4766 p.turn(desk, DUE + minute * 60 + if desk { 0 } else { 30 });
4767 for line in p.fake.seen() {
4768 let path = line.split(' ').nth(1).unwrap_or_default();
4769 if line.starts_with("PUT ") || line.starts_with("DELETE ") {
4770 assert!(path.contains(&device.id) || path.contains("/blobs/"), "{line}");
4771 } else {
4772 assert!(line.starts_with("GET ") || line.starts_with("PROPFIND ") || line.starts_with("MKCOL "), "{line}");
4773 }
4774 }
4775 }
4776 }
4777 }
4778
4779 #[test]
4780 fn a_refused_or_failed_upload_is_never_lost() {
4781 let p = Pair::new("refused");
4782 p.phone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n");
4783 for (n, fault) in [Fault::Status(412), Fault::Status(500), Fault::Drop].into_iter().enumerate() {
4784 p.fake.clear_faults();
4785 let id = p.phone.id.clone();
4786 p.fake.fault(move |r| (r.method == "PUT" && r.path.contains(&id) && r.path.ends_with(".jsonl")).then_some(fault));
4787 let now = DUE + 40 + n as i64 * 10;
4788 let _ = p.step(false, now, true);
4790 p.phone.exchange(&p.phone_folder, now);
4791 let pushed = p.step(false, now, false);
4792 assert!(pushed.problem.is_some(), "{fault:?}: {pushed:?}");
4793 assert!(p.phone.read("health-state.toml").contains("levo@"), "nothing lost here");
4794 p.turn(true, now + 5);
4795 assert!(!p.desk.read("health-state.toml").contains("levo@"), "{fault:?}: not there yet");
4796 }
4797 p.fake.clear_faults();
4799 p.turn(false, DUE + 100);
4800 p.turn(true, DUE + 110);
4801 assert!(p.desk.read("health-state.toml").contains("levo@1800000000"));
4802 let heard = crate::share::heard(&p.desk.memory, &p.desk.id);
4803 assert!(!heard.broken.contains_key(&p.phone.id), "{heard:?}");
4804 }
4805
4806 #[test]
4807 fn its_copy_lost_a_device_goes_on_after_its_last_record() {
4808 let p = Pair::new("lost");
4809 p.phone.write("health-state.toml", "[taken]\n\"one@1800000000\" = 1800000010\n");
4810 p.turn(false, DUE + 20);
4811 p.turn(true, DUE + 25);
4812 for entry in std::fs::read_dir(&p.phone_folder).unwrap().filter_map(Result::ok) {
4814 if entry.file_name() != "seal.toml" {
4815 let _ = std::fs::remove_dir_all(entry.path());
4816 let _ = std::fs::remove_file(entry.path());
4817 }
4818 }
4819 p.phone.write("health-state.toml", "[taken]\n\"one@1800000000\" = 1800000010\n\"two@1800000000\" = 1800000030\n");
4820 let outcome = p.turn(false, DUE + 40);
4821 assert!(outcome.problems.iter().all(|p| !p.starts_with("share-own")), "{:?}", outcome.problems);
4822 p.turn(true, DUE + 50);
4823 let record = p.desk.read("health-state.toml");
4824 assert!(record.contains("one@") && record.contains("two@"), "{record}");
4825 let heard = crate::share::heard(&p.desk.memory, &p.desk.id);
4826 assert!(!heard.broken.contains_key(&p.phone.id), "no gap nor cut on the desk: {heard:?}");
4827 assert!(p.phone_folder.join(format!("devices/{}.device", p.desk.id)).exists());
4829 }
4830
4831 #[test]
4832 fn sealed_files_travel_whole_and_go_when_swept() {
4833 let p = Pair::new("blobs");
4834 let source = p.base.join("note.md");
4835 std::fs::write(&source, "a note\n".repeat(1000)).unwrap();
4836 let (hash, _) = crate::blobs::hash_file(&source).unwrap();
4837 crate::blobs::put(&p.desk_folder, &KEY, &source, &hash).unwrap();
4838 let name = crate::blobs::name(&KEY, &hash);
4839 p.turn(true, DUE + 10);
4840 p.turn(false, DUE + 20);
4841 let sealed = std::fs::read(p.desk_folder.join("blobs").join(&name)).unwrap();
4842 assert_eq!(std::fs::read(p.phone_folder.join("blobs").join(&name)).unwrap(), sealed);
4843 p.fake.forget_seen();
4845 p.turn(false, DUE + 30);
4846 assert!(p.fake.seen().iter().all(|l| !(l.starts_with("PUT ") && l.contains("/blobs/"))), "{:?}", p.fake.seen());
4847 std::fs::remove_file(p.desk_folder.join("blobs").join(&name)).unwrap();
4849 p.turn(true, DUE + 40);
4850 assert!(!p.fake.files.join("Documents/Sioul/blobs").join(&name).exists());
4851 }
4852
4853 #[test]
4858 fn a_slow_or_failing_sealed_file_never_holds_back_the_records() {
4859 let p = Pair::new("sealed-last");
4860 let made = |seed: u32| {
4861 let mut x = seed.wrapping_mul(2_654_435_761) | 1;
4862 (0..600_000)
4863 .map(|_| {
4864 x ^= x << 13;
4865 x ^= x >> 17;
4866 x ^= x << 5;
4867 x as u8
4868 })
4869 .collect::<Vec<u8>>()
4870 };
4871 let table = p.desk.roots.data.join("spam").join("table.bin");
4872 std::fs::create_dir_all(table.parent().unwrap()).unwrap();
4873 std::fs::write(&table, made(1)).unwrap();
4874 let mut taken = String::from("[taken]\n");
4875 for (n, fault) in [Fault::Status(500), Fault::Drop, Fault::Late(Duration::from_secs(6))].into_iter().enumerate() {
4877 p.fake.clear_faults();
4878 p.fake.fault(move |r| (r.method == "PUT" && r.path.contains("/blobs/")).then_some(fault));
4879 let now = DUE + 100 + n as i64 * 60;
4880 taken.push_str(&format!("\"levo@{}\" = {}\n", 1_800_000_000 + n, 1_800_000_040 + n));
4882 p.desk.write("health-state.toml", &taken);
4883 let _ = p.step(true, now, true);
4884 p.desk.exchange(&p.desk_folder, now);
4885 let pushed = p.step(true, now, false);
4886 assert!(pushed.problem.is_some(), "{fault:?}: {pushed:?}");
4887 p.turn(false, now + 30);
4889 assert!(p.phone.read("health-state.toml").contains(&format!("levo@{}", 1_800_000_000 + n)), "{fault:?}");
4890 }
4891 p.fake.clear_faults();
4893 p.turn(true, DUE + 400);
4894 p.turn(false, DUE + 410);
4895 let phone_table = p.phone.roots.data.join("spam").join("table.bin");
4896 assert_eq!(std::fs::read(&phone_table).unwrap(), std::fs::read(&table).unwrap());
4897 std::fs::write(&table, made(2)).unwrap();
4899 p.turn(true, DUE + 500);
4900 p.phone.write("health-state.toml", &format!("{taken}\"two@1800000000\" = 1800000090\n"));
4901 p.phone.exchange(&p.phone_folder, DUE + 505);
4902 p.fake.fault(|r| (r.method == "GET" && r.path.contains("/blobs/")).then_some(Fault::Late(Duration::from_secs(2))));
4903 let stepped = p.step(false, DUE + 510, true);
4904 assert!(stepped.problem.is_some(), "{stepped:?}");
4905 p.fake.clear_faults();
4906 p.turn(true, DUE + 520);
4907 assert!(p.desk.read("health-state.toml").contains("two@1800000000"));
4908 p.turn(false, DUE + 530);
4910 assert_eq!(std::fs::read(&phone_table).unwrap(), std::fs::read(&table).unwrap());
4911 }
4912
4913 #[test]
4916 fn a_device_with_a_sync_app_and_one_without_share_one_folder() {
4917 let w = World::new("mixed");
4918 let lone = Device::new(&w.base, "lone", PHONE);
4919 let folder = mirror_of(&lone.memory);
4920 let login = w.fake.login();
4921 let opened = open_with(&login, "Documents/Sioul", TEST).unwrap();
4922 put_bytes(&folder, "seal.toml", opened.seal.as_deref().expect("sealed by the desk"));
4923 begin(&lone.memory, &folder, &login, &opened, DUE).unwrap();
4924 lone.session(&folder, |e| e.start(DUE));
4925 let turn = |now: i64| {
4926 assert!(step_with(&lone.memory, &folder, &lone.id, &login, now, true, TEST).problem.is_none());
4927 lone.exchange(&folder, now);
4928 assert!(step_with(&lone.memory, &folder, &lone.id, &login, now, false, TEST).problem.is_none());
4929 };
4930 turn(DUE + 10);
4931 w.desk.press(true, DUE + 20);
4933 w.desk.exchange(&w.server, DUE + 30);
4934 turn(DUE + 40);
4935 assert!(lone.dnd().is_some_and(|p| p.on && p.from == w.desk.id));
4936 lone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000050\n");
4938 turn(DUE + 50);
4939 w.desk.exchange(&w.server, DUE + 60);
4940 assert!(w.desk.read("health-state.toml").contains("levo@1800000000"));
4941 let seen = w.desk.sees(&w.server, &lone, DUE + 70);
4942 assert!(seen.complete && seen.said.is_some_and(|s| s.working), "the desk knows how the phone is");
4943 }
4944
4945 #[test]
4946 fn keeping_the_folder_is_inert_until_set_up() {
4947 let w = World::new("inert");
4948 w.fake.forget_seen();
4949 let stepped = step_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE, true, TEST);
4950 assert_eq!(stepped.problem.as_deref(), Some("not-confirmed"));
4951 assert!(w.fake.seen().is_empty(), "not a request");
4952 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
4954 w.fake.forget_seen();
4955 assert_eq!(step_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE + 1, false, TEST).problem.as_deref(), Some("not-confirmed"));
4956 assert!(w.fake.seen().is_empty());
4957 }
4958
4959 fn mtime(path: &Path) -> Option<SystemTime> {
4962 std::fs::metadata(path).and_then(|m| m.modified()).ok()
4963 }
4964
4965 fn set_mtime(path: &Path, at: SystemTime) {
4966 if let Ok(file) = std::fs::File::options().write(true).open(path) {
4967 let _ = file.set_modified(at);
4968 }
4969 }
4970
4971 fn sync_up(local: &Path, server: &Path) {
4973 std::fs::create_dir_all(server.parent().unwrap()).unwrap();
4974 std::fs::write(server, std::fs::read(local).unwrap()).unwrap();
4975 fake::bump(server);
4976 if let Some(at) = mtime(local) {
4977 set_mtime(server, at);
4978 }
4979 }
4980
4981 fn sync_down(server: &Path, local: &Path) {
4983 std::fs::create_dir_all(local.parent().unwrap()).unwrap();
4984 let temporary = local.with_file_name(format!(".{}.download", local.file_name().unwrap().to_string_lossy()));
4985 std::fs::write(&temporary, std::fs::read(server).unwrap()).unwrap();
4986 if let Some(at) = mtime(server) {
4987 set_mtime(&temporary, at);
4988 }
4989 std::fs::rename(&temporary, local).unwrap();
4990 }
4991
4992 fn folder_of(relative: &str) -> &str {
4993 relative.rfind('/').map_or("", |at| &relative[..at])
4994 }
4995
4996 #[derive(Default)]
5007 struct EDrive {
5008 known: BTreeMap<String, (String, Option<SystemTime>)>,
5009 dirs: BTreeMap<String, Option<SystemTime>>,
5010 refused: BTreeMap<String, u32>,
5011 uploads: usize,
5012 refusals: usize,
5013 downloads_over_newer: usize,
5014 }
5015
5016 impl EDrive {
5017 fn scan(&mut self, fake: &Fake, server: &Path, phone: &Path) {
5018 let _turn = fake.turn.lock().unwrap();
5019 let (there, here) = (walk(server), walk(phone));
5020 let mut downloads: BTreeSet<String> = BTreeSet::new();
5021 for (relative, bytes) in there.iter().filter(|(r, _)| *r != "seal.toml") {
5022 let etag = fake::etag_of(&server.join(relative));
5023 if self.known.get(relative).is_some_and(|(known, _)| *known == etag) {
5024 continue;
5025 }
5026 if here.get(relative).map(Vec::len) == Some(bytes.len()) {
5027 let time = self.known.get(relative).and_then(|(_, t)| *t);
5028 self.known.insert(relative.clone(), (etag, time));
5029 } else {
5030 downloads.insert(relative.clone());
5031 }
5032 }
5033 let mut uploads: Vec<String> = Vec::new();
5034 for relative in here.keys().filter(|r| *r != "seal.toml") {
5035 let dir = folder_of(relative).to_string();
5036 let looked = self.dirs.get(&dir).copied().flatten() != mtime(&phone.join(&dir));
5037 let later = self.known.get(relative).is_none_or(|(_, t)| mtime(&phone.join(relative)) > *t);
5038 if looked && later {
5039 downloads.remove(relative);
5040 uploads.push(relative.clone());
5041 }
5042 }
5043 for relative in downloads {
5044 if here.get(&relative).is_some_and(|mine| there[&relative].len() < mine.len()) {
5045 self.downloads_over_newer += 1;
5046 }
5047 sync_down(&server.join(&relative), &phone.join(&relative));
5048 self.known.insert(relative.clone(), (fake::etag_of(&server.join(&relative)), mtime(&phone.join(&relative))));
5049 }
5050 for relative in uploads {
5051 self.upload(server, phone, &relative);
5052 }
5053 self.dirs = here.keys().map(|r| folder_of(r).to_string()).chain([String::new()]).map(|d| (d.clone(), mtime(&phone.join(&d)))).collect();
5054 }
5055
5056 fn upload(&mut self, server: &Path, phone: &Path, relative: &str) {
5058 if self.refused.get(relative).is_some_and(|n| *n >= 4) {
5059 return;
5060 }
5061 let there = server.join(relative);
5062 if let Some((known, _)) = self.known.get(relative)
5063 && there.is_file()
5064 && fake::etag_of(&there) != *known
5065 {
5066 self.refusals += 1;
5067 *self.refused.entry(relative.to_string()).or_insert(0) += 1;
5068 return;
5069 }
5070 sync_up(&phone.join(relative), &there);
5071 self.uploads += 1;
5072 self.known.insert(relative.to_string(), (fake::etag_of(&there), mtime(&phone.join(relative))));
5073 }
5074
5075 fn closed(&mut self, fake: &Fake, server: &Path, phone: &Path, relative: &str) {
5077 let _turn = fake.turn.lock().unwrap();
5078 if phone.join(relative).is_file() {
5079 self.upload(server, phone, relative);
5080 }
5081 }
5082 }
5083
5084 #[derive(Default)]
5091 struct Client {
5092 journal: BTreeMap<String, (String, u64, Option<SystemTime>)>,
5093 conflicts: usize,
5094 uploads: usize,
5095 }
5096
5097 impl Client {
5098 fn stamp(path: &Path) -> Option<(u64, Option<SystemTime>)> {
5099 std::fs::metadata(path).ok().filter(|m| m.is_file()).map(|m| (m.len(), m.modified().ok()))
5100 }
5101
5102 fn sync(&mut self, fake: &Fake, server: &Path, local: &Path) {
5103 let _turn = fake.turn.lock().unwrap();
5104 let names: BTreeSet<String> = walk(server).into_keys().chain(walk(local).into_keys()).filter(|r| r != "seal.toml" && !r.contains("(conflicted copy")).collect();
5105 for relative in names {
5106 let (there, here) = (server.join(&relative), local.join(&relative));
5107 let etag = there.is_file().then(|| fake::etag_of(&there));
5108 let stamp = Self::stamp(&here);
5109 let known = self.journal.get(&relative).cloned();
5110 let local_changed = known.as_ref().map(|(_, size, at)| Some((*size, *at))) != Some(stamp);
5111 let remote_changed = known.as_ref().map(|(e, ..)| e.clone()) != etag;
5112 let remember = |journal: &mut BTreeMap<String, (String, u64, Option<SystemTime>)>| {
5113 let (size, at) = Self::stamp(&here).unwrap_or((0, None));
5114 journal.insert(relative.clone(), (fake::etag_of(&there), size, at));
5115 };
5116 match (stamp.is_some(), etag.is_some()) {
5117 (true, true) if !local_changed && !remote_changed => {}
5118 (true, true) if !remote_changed => {
5119 sync_up(&here, &there);
5120 self.uploads += 1;
5121 remember(&mut self.journal);
5122 }
5123 (true, true) if !local_changed => {
5124 sync_down(&there, &here);
5125 remember(&mut self.journal);
5126 }
5127 (true, true) => {
5128 if std::fs::read(&here).ok() != std::fs::read(&there).ok() {
5129 let name = here.file_name().unwrap().to_string_lossy().to_string();
5130 let (stem, extension) = name.rsplit_once('.').unwrap_or((&name, ""));
5131 std::fs::rename(&here, here.with_file_name(format!("{stem} (conflicted copy 20261008 104200).{extension}"))).unwrap();
5132 sync_down(&there, &here);
5133 self.conflicts += 1;
5134 }
5135 remember(&mut self.journal);
5136 }
5137 (true, false) if known.is_some() && !local_changed => {
5138 let _ = std::fs::remove_file(&here);
5139 self.journal.remove(&relative);
5140 }
5141 (true, false) => {
5142 sync_up(&here, &there);
5143 self.uploads += 1;
5144 remember(&mut self.journal);
5145 }
5146 (false, true) if known.is_some() && !remote_changed => {
5147 let _ = std::fs::remove_file(&there);
5148 self.journal.remove(&relative);
5149 }
5150 (false, true) => {
5151 sync_down(&there, &here);
5152 remember(&mut self.journal);
5153 }
5154 (false, false) => {
5155 self.journal.remove(&relative);
5156 }
5157 }
5158 }
5159 }
5160 }
5161
5162 fn records_there(server: &Path, id: &str) -> BTreeMap<String, usize> {
5164 walk(server).into_iter().filter(|(r, _)| r.starts_with(id) && r.ends_with(".jsonl")).map(|(r, b)| (r, b.len())).collect()
5165 }
5166
5167 fn never_shorter(before: &mut BTreeMap<String, usize>, now: BTreeMap<String, usize>, at: &str) {
5168 for (name, size) in &now {
5169 assert!(before.get(name).is_none_or(|was| size >= was), "{at}: {name} shorter there: {size} after {:?}", before.get(name));
5170 }
5171 before.extend(now);
5172 }
5173
5174 fn phone_sends(w: &World, now: i64, again: bool) -> Pulled {
5176 send_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), &BTreeMap::new(), now, again, TEST)
5177 }
5178
5179 #[test]
5184 fn beside_a_sync_app_only_this_devices_files_go_up_and_never_twice() {
5185 let w = World::new("beside");
5186 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
5187 w.phone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n");
5189 w.phone.exchange(&w.phone_folder, DUE + 40);
5190 w.fake.forget_seen();
5191 let sent = phone_sends(&w, DUE + 41, false);
5192 assert!(sent.problem.is_none() && sent.sent >= 2, "{sent:?}");
5193 let mine = |line: &String| line.split(' ').nth(1).is_some_and(|path| path.contains(&w.phone.id));
5194 for line in w.fake.seen() {
5195 if line.starts_with("PUT ") || line.starts_with("DELETE ") {
5196 assert!(mine(&line), "never another device's file: {line}");
5197 }
5198 }
5199 for (relative, bytes) in walk(&w.phone_folder).into_iter().filter(|(r, _)| r.contains(&w.phone.id)) {
5200 assert_eq!(std::fs::read(w.server.join(&relative)).ok(), Some(bytes), "{relative} there as here");
5201 }
5202 let kept = std::fs::read_to_string(sending_path(&w.phone.memory)).unwrap();
5203 assert!(!kept.contains(PASSWORD) && !kept.contains("Basic"), "{kept}");
5204 w.desk.exchange(&w.server, DUE + 45);
5205 assert!(w.desk.read("health-state.toml").contains("levo@1800000000"), "the desk has it, no sync app between");
5206 w.up();
5208 for relative in walk(&w.phone_folder).into_keys().filter(|r| r.contains(&w.phone.id)) {
5209 fake::bump(&w.server.join(relative));
5210 }
5211 w.fake.forget_seen();
5212 let again = phone_sends(&w, DUE + 50, false);
5213 assert_eq!((again.sent, &again.problem), (0, &None));
5214 assert!(w.fake.seen().is_empty(), "{:?}", w.fake.seen());
5215 w.phone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n\"iron@1800000000\" = 1800000060\n");
5217 w.phone.exchange(&w.phone_folder, DUE + 60);
5218 {
5219 let _turn = w.fake.turn.lock().unwrap();
5220 for relative in walk(&w.phone_folder).into_keys().filter(|r| r.contains(&w.phone.id)) {
5221 sync_up(&w.phone_folder.join(&relative), &w.server.join(&relative));
5222 }
5223 }
5224 w.fake.forget_seen();
5225 let raced = phone_sends(&w, DUE + 61, false);
5226 assert!(raced.problem.is_none() && raced.sent == 0 && raced.same >= 2, "{raced:?} {:?}", w.fake.seen());
5227 assert!(w.fake.seen().iter().all(|l| !(l.starts_with("PUT ") && !l.ends_with(" 412"))), "{:?}", w.fake.seen());
5228 let round = walk(&w.phone_folder).into_keys().find(|r| r.starts_with(&w.phone.id) && r.ends_with(".jsonl")).unwrap();
5230 let longer = std::fs::read(w.server.join(&round)).unwrap();
5231 let first = longer.iter().position(|b| *b == b'\n').unwrap() + 1;
5232 std::fs::write(w.phone_folder.join(&round), &longer[..first]).unwrap();
5233 w.fake.forget_seen();
5234 let cut = phone_sends(&w, DUE + 70, false);
5235 assert!(cut.problem.is_none(), "{cut:?}");
5236 assert_eq!(std::fs::read(w.server.join(&round)).unwrap(), longer, "never shorter there");
5237 assert!(w.fake.seen().iter().all(|l| !(l.starts_with("PUT ") && l.contains(&round))), "{:?}", w.fake.seen());
5238 assert_eq!(phone_sends(&w, DUE + 72, true).longer, 1);
5240 let mut appended = longer[..first].to_vec();
5241 appended.extend_from_slice(&longer[first..first + (longer.len() - first) / 2]);
5242 appended.push(b'\n');
5243 std::fs::write(w.phone_folder.join(&round), &appended).unwrap();
5244 w.fake.forget_seen();
5245 assert!(phone_sends(&w, DUE + 75, false).problem.is_none());
5246 assert_eq!(std::fs::read(w.server.join(&round)).unwrap(), longer, "an append here, still shorter, never sent over the longer copy");
5247 assert!(w.fake.seen().iter().all(|l| !(l.starts_with("PUT ") && l.contains(&round))), "{:?}", w.fake.seen());
5248 State::choose(&w.phone.memory, |s| s.send = Some(false)).unwrap();
5250 w.fake.forget_seen();
5251 assert_eq!(phone_sends(&w, DUE + 80, true).problem.as_deref(), Some("off"));
5252 assert!(w.fake.seen().is_empty());
5253 assert!(State::load(&w.phone.memory).fetching());
5254 }
5255
5256 #[test]
5258 fn sealed_files_go_up_whole_and_only_this_devices() {
5259 let w = World::new("beside-blobs");
5260 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
5261 let source = w.base.join("note.md");
5262 std::fs::write(&source, "a note\n".repeat(500)).unwrap();
5263 let (hash, _) = crate::blobs::hash_file(&source).unwrap();
5264 crate::blobs::put(&w.phone_folder, &KEY, &source, &hash).unwrap();
5265 let name = crate::blobs::name(&KEY, &hash);
5266 let size = std::fs::metadata(w.phone_folder.join("blobs").join(&name)).unwrap().len();
5267 put(&w.phone_folder, "blobs/0123456789abcdef", "another device's");
5269 let half: BTreeMap<String, u64> = [(name.clone(), size + 1)].into();
5271 send_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), &half, DUE + 10, false, TEST);
5272 assert!(!w.server.join("blobs").join(&name).exists());
5273 let whole: BTreeMap<String, u64> = [(name.clone(), size)].into();
5274 let sent = send_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), &whole, DUE + 20, false, TEST);
5275 assert!(sent.problem.is_none(), "{sent:?}");
5276 assert_eq!(std::fs::read(w.server.join("blobs").join(&name)).unwrap(), std::fs::read(w.phone_folder.join("blobs").join(&name)).unwrap());
5277 assert!(!w.server.join("blobs/0123456789abcdef").exists(), "never another device's");
5278 std::fs::remove_file(w.phone_folder.join("blobs").join(&name)).unwrap();
5280 send_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), &BTreeMap::new(), DUE + 30, false, TEST);
5281 assert!(!w.server.join("blobs").join(&name).exists());
5282 }
5283
5284 #[test]
5287 fn send_everything_again_after_the_server_lost_this_devices_files() {
5288 let w = World::new("again");
5289 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
5290 w.phone.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n");
5291 w.phone.exchange(&w.phone_folder, DUE + 40);
5292 assert!(phone_sends(&w, DUE + 41, false).problem.is_none());
5293 let own: Vec<String> = walk(&w.phone_folder).into_keys().filter(|r| r.contains(&w.phone.id)).collect();
5294 assert!(own.len() >= 2, "{own:?}");
5295 for relative in &own {
5297 std::fs::remove_file(w.server.join(relative)).unwrap();
5298 }
5299 assert!(crate::devices::all(&w.server, &KEY).0.iter().all(|e| e.id != w.phone.id));
5300 w.fake.forget_seen();
5302 assert_eq!(phone_sends(&w, DUE + 50, false).sent, 0);
5303 assert!(w.fake.seen().is_empty());
5304 let again = phone_sends(&w, DUE + 60, true);
5306 assert!(again.problem.is_none() && again.sent == own.len(), "{again:?}");
5307 for relative in &own {
5308 assert_eq!(std::fs::read(w.server.join(relative)).ok(), std::fs::read(w.phone_folder.join(relative)).ok(), "{relative}");
5309 }
5310 assert_eq!(Sending::load(&w.phone.memory).again, Again { at: DUE + 60, sent: own.len(), same: 0, said: String::new() });
5311 w.desk.exchange(&w.server, DUE + 70);
5312 assert!(w.desk.read("health-state.toml").contains("levo@1800000000"));
5313 assert!(crate::devices::all(&w.server, &KEY).0.iter().any(|e| e.id == w.phone.id), "its entry back");
5314 w.fake.forget_seen();
5316 let twice = phone_sends(&w, DUE + 80, true);
5317 assert!(twice.problem.is_none() && twice.sent == 0 && twice.same == own.len(), "{twice:?}");
5318 assert!(w.fake.seen().iter().all(|l| !l.starts_with("PUT ")), "{:?}", w.fake.seen());
5319 w.fake.fault(|_| Some(Fault::Status(503)));
5321 let failed = phone_sends(&w, DUE + 90, true);
5322 assert!(failed.problem.as_deref().is_some_and(|p| p.starts_with("server:")), "{failed:?}");
5323 assert!(Sending::load(&w.phone.memory).again.said.starts_with("server:"));
5324 }
5325
5326 #[test]
5333 fn doses_are_never_known_wrongly_with_edrive_sending_too() {
5334 let w = World::new("edrive-sends");
5335 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
5336 let mut edrive = EDrive::default();
5337 edrive.scan(&w.fake, &w.server, &w.phone_folder);
5338 let mut marked: Vec<(String, i64)> = Vec::new();
5339 let mut sizes = BTreeMap::new();
5340 for minute in 0..150 {
5341 let now = DUE + minute * 60;
5342 if (minute % 10 == 0 || [17, 19, 77, 79].contains(&minute)) && minute < 130 {
5344 marked.push((format!("dose@{minute}"), now));
5345 let record: String = marked.iter().map(|(d, at)| format!("\"{d}\" = {at}\n")).collect();
5346 w.phone.write("health-state.toml", &format!("[taken]\n{record}"));
5347 }
5348 w.phone.exchange(&w.phone_folder, now);
5349 let round = walk(&w.phone_folder).into_keys().filter(|r| r.starts_with(&w.phone.id) && r.ends_with(".jsonl")).max_by_key(|r| r.trim_end_matches(".jsonl").rsplit('-').next().unwrap().parse::<u32>().unwrap()).unwrap();
5350 match minute % 6 {
5351 1 => edrive.scan(&w.fake, &w.server, &w.phone_folder),
5353 3 => edrive.closed(&w.fake, &w.server, &w.phone_folder, &round),
5355 _ => {}
5356 }
5357 let sent = phone_sends(&w, now + 1, false);
5358 assert!(sent.problem.is_none(), "minute {minute}: {sent:?}");
5359 if minute % 4 == 0 {
5361 edrive.scan(&w.fake, &w.server, &w.phone_folder);
5362 }
5363 never_shorter(&mut sizes, records_there(&w.server, &w.phone.id), &format!("minute {minute}"));
5364 let outcome = w.desk.exchange(&w.server, now + 20);
5365 assert!(outcome.problems.iter().all(|p| !p.starts_with("share-other-cut") && !p.starts_with("share-other-gap")), "minute {minute}: {:?}", outcome.problems);
5366 if let Some(entry) = crate::devices::all(&w.server, &KEY).0.into_iter().find(|e| e.id == w.phone.id)
5367 && let Some(wrote) = entry.wrote
5368 && crate::share::heard(&w.desk.memory, &w.desk.id).complete(&w.phone.id, wrote)
5369 {
5370 let record = w.desk.read("health-state.toml");
5371 for (dose, at) in marked.iter().filter(|(_, at)| *at <= entry.exported) {
5372 assert!(record.contains(dose.as_str()), "minute {minute}: the desk takes {dose} ({at}) as known, its record lacks it:\n{record}");
5373 }
5374 }
5375 }
5376 let record = w.desk.read("health-state.toml");
5377 for (dose, _) in &marked {
5378 assert!(record.contains(dose.as_str()), "{dose} never came:\n{record}");
5379 }
5380 assert!(walk(&w.server).keys().all(|r| !r.contains("conflicted")));
5381 eprintln!("eDrive beside Sioul: {} uploads, {} refused, {} brought down over a newer copy", edrive.uploads, edrive.refusals, edrive.downloads_over_newer);
5382 assert!(edrive.downloads_over_newer > 0 && edrive.refusals > 0, "the races were run");
5383 }
5384
5385 #[test]
5391 fn doses_are_never_known_wrongly_with_the_nextcloud_client_sending_too() {
5392 let w = World::new("client-sends");
5393 let desk_folder = w.base.join("desk-folder").join("Sioul");
5394 copy_dir(&w.server, &desk_folder);
5395 State::choose(&w.desk.memory, |s| s.given = "Documents/Sioul".into()).unwrap();
5396 assert!(find_with(&w.desk.memory, &desk_folder, &[w.fake.login()], &[], DUE, TEST).confirmed_for(&desk_folder));
5397 let mut client = Client::default();
5398 client.sync(&w.fake, &w.server, &desk_folder);
5399 let phone_down = || {
5400 let _turn = w.fake.turn.lock().unwrap();
5401 for (relative, bytes) in walk(&w.server).into_iter().filter(|(r, _)| !r.contains(&w.phone.id)) {
5402 put_bytes(&w.phone_folder, &relative, &bytes);
5403 }
5404 };
5405 let phone_up = || {
5406 let _turn = w.fake.turn.lock().unwrap();
5407 for relative in walk(&w.phone_folder).into_keys().filter(|r| r.contains(&w.phone.id)) {
5408 sync_up(&w.phone_folder.join(&relative), &w.server.join(&relative));
5409 }
5410 };
5411 let mut marked: Vec<(String, i64)> = Vec::new();
5412 let mut sizes = BTreeMap::new();
5413 let mut cuts = 0;
5414 for minute in 0..150 {
5415 let now = DUE + minute * 60;
5416 if (minute % 10 == 0 || [21, 71].contains(&minute)) && minute < 130 {
5418 marked.push((format!("dose@{minute}"), now));
5419 let record: String = marked.iter().map(|(d, at)| format!("\"{d}\" = {at}\n")).collect();
5420 w.desk.write("health-state.toml", &format!("[taken]\n{record}"));
5421 }
5422 let outcome = w.desk.exchange(&desk_folder, now);
5423 cuts += outcome.problems.iter().filter(|p| *p == "share-own-cut").count();
5424 if minute % 5 == 1 {
5426 client.sync(&w.fake, &w.server, &desk_folder);
5427 }
5428 let sent = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), now + 1, false, TEST);
5429 assert!(sent.problem.is_none(), "minute {minute}: {sent:?}");
5430 if minute % 3 == 0 {
5431 client.sync(&w.fake, &w.server, &desk_folder);
5432 }
5433 never_shorter(&mut sizes, records_there(&w.server, &w.desk.id), &format!("minute {minute}"));
5434 phone_down();
5435 let read = w.phone.exchange(&w.phone_folder, now + 30);
5436 assert!(read.problems.iter().all(|p| !p.starts_with("share-other-cut") && !p.starts_with("share-other-gap")), "minute {minute}: {:?}", read.problems);
5437 phone_up();
5438 if let Some(entry) = crate::devices::all(&w.phone_folder, &KEY).0.into_iter().find(|e| e.id == w.desk.id)
5439 && let Some(wrote) = entry.wrote
5440 && crate::share::heard(&w.phone.memory, &w.phone.id).complete(&w.desk.id, wrote)
5441 {
5442 let record = w.phone.read("health-state.toml");
5443 for (dose, at) in marked.iter().filter(|(_, at)| *at <= entry.exported) {
5444 assert!(record.contains(dose.as_str()), "minute {minute}: the phone takes {dose} ({at}) as known, its record lacks it:\n{record}");
5445 }
5446 }
5447 }
5448 let record = w.phone.read("health-state.toml");
5449 for (dose, _) in &marked {
5450 assert!(record.contains(dose.as_str()), "{dose} never came:\n{record}");
5451 }
5452 assert!(walk(&w.server).keys().all(|r| !r.contains("conflicted")), "a conflicted copy never goes up");
5453 eprintln!("Nextcloud client beside Sioul: {} uploads, {} conflicted copies here, {} rounds started again", client.uploads, client.conflicts, cuts);
5454 assert!(client.conflicts > 0 && cuts > 0, "the races were run");
5455 }
5456
5457 fn desk_beside(w: &World) -> PathBuf {
5462 let desk_folder = w.base.join("desk-folder").join("Sioul");
5463 copy_dir(&w.server, &desk_folder);
5464 State::choose(&w.desk.memory, |s| s.given = "Documents/Sioul".into()).unwrap();
5465 assert!(find_with(&w.desk.memory, &desk_folder, &[w.fake.login()], &[], DUE, TEST).confirmed_for(&desk_folder));
5466 desk_folder
5467 }
5468
5469 fn a_desk_holding_much(w: &World, folder: &Path, at: i64) {
5471 let many: String = (0..8000).map(|n| format!("sender{n}@example.org\n")).collect();
5472 put(&w.desk.roots.config, "known-senders.txt", &many);
5473 w.desk.exchange(folder, at);
5474 }
5475
5476 fn desk_records(folder: &Path, id: &str) -> Vec<(String, u64)> {
5477 let mut out: Vec<(String, u64)> = walk(folder).into_iter().filter(|(r, _)| r.starts_with(id) && r.ends_with(".jsonl")).map(|(r, b)| (r, b.len() as u64)).collect();
5478 out.sort_by_key(|(r, _)| r.trim_end_matches(".jsonl").rsplit('-').next().unwrap().parse::<u32>().unwrap());
5479 out
5480 }
5481
5482 #[test]
5488 fn a_large_file_goes_through_a_slow_uplink() {
5489 let w = World::new("slow-uplink");
5490 let desk_folder = desk_beside(&w);
5491 a_desk_holding_much(&w, &desk_folder, DUE);
5492 let (round, size) = desk_records(&desk_folder, &w.desk.id).into_iter().find(|(_, s)| *s > 1 << 20).unwrap();
5494 assert!(size > 1 << 20, "a large records file: {size}");
5495 *w.fake.throttle.lock().unwrap() = Some(512 << 10);
5497 let flat = Limits { large: Duration::from_secs(1), floor: 1 << 30, ..TEST };
5498 let tried = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 10, false, flat);
5499 let failure = tried.failure.clone().expect("said in detail");
5500 assert!(failure.kind == "timeout" && failure.file == "records" && failure.sent > 0 && failure.total == size && failure.seconds >= 1, "{failure:?}");
5501 assert!(tried.held == 1, "{tried:?}");
5502 let sending = Sending::load(&w.desk.memory);
5503 assert_eq!(sending.failure.as_ref().map(|f| f.kind.as_str()), Some("timeout"));
5504 assert!(sending.held.get(&round).is_some_and(|h| h.tries == 1 && h.until == DUE + 10 + backoff(1)), "{:?}", sending.held);
5505 assert!(sending.rate > 0, "its rate measured from what went");
5506 let spent = w.fake.received.load(std::sync::atomic::Ordering::Relaxed);
5508 let early = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 40, false, flat);
5509 assert!(early.held == 0 && w.fake.received.load(std::sync::atomic::Ordering::Relaxed) - spent < size / 2, "{early:?}");
5510 assert!(std::fs::metadata(w.server.join(&round)).map_or(0, |m| m.len()) < size);
5511 let scaled = Limits { large: Duration::from_secs(1), floor: 128 << 10, stall: Duration::from_secs(3), ..TEST };
5514 let went = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 10 + backoff(1), false, scaled);
5515 assert!(went.problem.is_none() && went.held == 0, "{went:?}");
5516 assert_eq!(std::fs::read(w.server.join(&round)).unwrap(), std::fs::read(desk_folder.join(&round)).unwrap());
5517 let sending = Sending::load(&w.desk.memory);
5518 assert!(sending.held.is_empty() && sending.failure.is_none() && sending.last == DUE + 10 + backoff(1), "{sending:?}");
5519 }
5520
5521 #[test]
5523 fn a_failed_send_waits_for_its_backoff() {
5524 let w = World::new("backoff");
5525 let desk_folder = desk_beside(&w);
5526 w.desk.write("health-state.toml", "[taken]\n\"levo@1800000000\" = 1800000040\n");
5527 w.desk.exchange(&desk_folder, DUE + 40);
5528 w.fake.fault(|r| (r.method == "PUT").then_some(Fault::Status(503)));
5529 let failed = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 41, false, TEST);
5530 assert!(failed.problem.as_deref().is_some_and(|p| p.starts_with("server:")), "{failed:?}");
5531 assert_eq!(Sending::load(&w.desk.memory).retry_at, DUE + 41 + backoff(1));
5532 w.fake.forget_seen();
5533 let early = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 41 + backoff(1) - 1, false, TEST);
5534 assert!(early.problem.as_deref().is_some_and(|p| p.starts_with("waiting:")) && w.fake.seen().is_empty(), "{early:?} {:?}", w.fake.seen());
5535 let again = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 41 + backoff(1), false, TEST);
5537 assert!(again.problem.is_some());
5538 assert_eq!(Sending::load(&w.desk.memory).retry_at, DUE + 41 + backoff(1) + backoff(2));
5539 assert_eq!((backoff(1), backoff(2), backoff(3), backoff(9)), (60, 120, 300, 3600));
5540 w.fake.clear_faults();
5542 let at = DUE + 41 + backoff(1) + backoff(2);
5543 assert!(send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), at, false, TEST).problem.is_none());
5544 let sending = Sending::load(&w.desk.memory);
5545 assert!(sending.tries == 0 && sending.retry_at == 0 && sending.failure.is_none());
5546 }
5547
5548 #[test]
5552 fn a_text_to_send_reaches_the_phone_while_the_records_crawl() {
5553 let w = World::new("texts-lane");
5554 let desk_folder = desk_beside(&w);
5555 assert!(w.find(DUE).confirmed_for(&w.phone_folder));
5556 a_desk_holding_much(&w, &desk_folder, DUE);
5557 *w.fake.throttle.lock().unwrap() = Some(256 << 10);
5559 let slow = Limits { large: Duration::from_secs(2), floor: 64 << 10, stall: Duration::from_secs(5), pull: Duration::from_secs(120), ..TEST };
5560 let crawling = {
5561 let (memory, folder, id, login) = (w.desk.memory.clone(), desk_folder.clone(), w.desk.id.clone(), w.fake.login());
5562 std::thread::spawn(move || send_with(&memory, &folder, &id, &login, &BTreeMap::new(), DUE + 10, false, slow))
5563 };
5564 std::thread::sleep(Duration::from_millis(500));
5565 let source = w.base.join("desk-texts-send.jsonl");
5567 std::fs::write(&source, "a sealed request line\n").unwrap();
5568 let written = Instant::now();
5569 assert!(place_texts(&desk_folder, &w.desk.id, &source).unwrap());
5570 let urgent = send_urgent_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), DUE + 11, slow);
5571 assert!(urgent.problem.is_none() && urgent.sent == 1, "{urgent:?}");
5572 let pulled = pull_with(&w.phone.memory, &w.phone_folder, &w.phone.id, &w.fake.login(), DUE + 20, TEST);
5574 assert!(pulled.problem.is_none(), "{pulled:?}");
5575 assert_eq!(std::fs::read(texts_file(&cache_of(&w.phone.memory), &w.desk.id)).unwrap(), b"a sealed request line\n");
5576 assert!(!crawling.is_finished(), "the records still going up");
5577 assert!(written.elapsed() < Duration::from_secs(10), "{:?}", written.elapsed());
5578 let crawled = crawling.join().unwrap();
5579 assert!(crawled.problem.is_none(), "{crawled:?}");
5580 assert!(!own_plain(&desk_folder, &w.desk.id).iter().any(|r| r.starts_with("texts/")));
5582 assert!(!place_texts(&desk_folder, &w.desk.id, &source).unwrap(), "unchanged: nothing to place");
5583 std::fs::remove_file(&source).unwrap();
5584 assert!(place_texts(&desk_folder, &w.desk.id, &source).unwrap() && !texts_file(&desk_folder, &w.desk.id).exists());
5585 }
5586
5587 #[test]
5591 fn bytes_an_hour_of_a_computer_writing_each_minute() {
5592 let w = World::new("an-hour");
5593 let desk_folder = desk_beside(&w);
5594 a_desk_holding_much(&w, &desk_folder, DUE);
5595 assert!(send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), DUE + 1, false, TEST).problem.is_none());
5596 let start = w.fake.received.load(std::sync::atomic::Ordering::Relaxed);
5597 let mut minutes: String = String::new();
5598 for minute in 1..=60 {
5599 let now = DUE + minute * 60;
5600 minutes.push_str(&format!("\"dose@{minute}\" = {now}\n"));
5601 w.desk.write("health-state.toml", &format!("[taken]\n{minutes}"));
5602 w.desk.exchange(&desk_folder, now);
5603 crate::lease::renew(&desk_folder, &KEY, "health", &w.desk.id, now, now, false, crate::lease::Rule::FollowsYou, crate::share::written(&w.desk.memory, &w.desk.id)).unwrap();
5604 let sent = send_with(&w.desk.memory, &desk_folder, &w.desk.id, &w.fake.login(), &BTreeMap::new(), now + 1, false, TEST);
5605 assert!(sent.problem.is_none(), "minute {minute}: {sent:?}");
5606 }
5607 let hour = w.fake.received.load(std::sync::atomic::Ordering::Relaxed) - start;
5608 let largest = desk_records(&desk_folder, &w.desk.id).last().map_or(0, |(_, s)| *s);
5609 eprintln!("A computer writing each minute, sent beside its sync app: {hour} bytes in an hour; its current records' file {largest} bytes");
5610 assert!(hour < 3 << 20, "{hour} bytes an hour");
5611 assert!(largest < 64 << 10, "{largest}");
5612 w.phone.exchange(&w.server, DUE + 3_700);
5614 assert!(w.phone.read("health-state.toml").contains("dose@60"));
5615 }
5616
5617 impl Device {
5620 fn exchange_notes(&self, folder: &Path, notes: &Path, now: i64) -> Outcome {
5622 let config = Config { notes_root: Some(notes.display().to_string()), ..Config::default() };
5623 let stores = crate::share::stores_of(&config, &self.roots, &|part| part == "notes" || crate::share::shared_by_default(part, &config));
5624 crate::share::exchange(&Sharing { folder, computer: &self.id, key: &KEY, memory: &self.memory, files: true, hurry: None }, &stores, now * 1000).unwrap()
5625 }
5626 }
5627
5628 fn notes_world(name: &str) -> (World, PathBuf, PathBuf) {
5632 let w = World::new(name);
5633 let (desk_notes, phone_notes) = (w.base.join("desk-notes"), w.base.join("phone-notes"));
5634 for dir in [&desk_notes, &phone_notes] {
5635 std::fs::create_dir_all(dir).unwrap();
5636 }
5637 w.desk.exchange_notes(&w.server, &desk_notes, DUE + 10);
5638 w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 20);
5639 w.up();
5640 assert!(w.find(DUE + 30).confirmed_for(&w.phone_folder));
5641 (w, desk_notes, phone_notes)
5642 }
5643
5644 fn desk_writes(w: &World, desk_notes: &Path, file: &str, text: &str, now: i64) -> String {
5646 put(desk_notes, file, text);
5647 w.desk.exchange_notes(&w.server, desk_notes, now);
5648 let name = crate::blobs::name(&KEY, &crate::blobs::hash_file(&desk_notes.join(file)).unwrap().0);
5649 assert!(w.server.join("blobs").join(&name).exists());
5650 name
5651 }
5652
5653 fn fetch(w: &World, now: i64, frugal: bool) -> Pulled {
5654 fetch_wanted_with(&w.phone.memory, &w.phone_folder, &w.fake.login(), &KEY, now, frugal, TEST)
5655 }
5656
5657 fn blob_gets(w: &World) -> usize {
5658 w.fake.seen().iter().filter(|l| l.starts_with("GET ") && l.contains("/blobs/")).count()
5659 }
5660
5661 #[test]
5668 fn a_note_waited_for_is_fetched_by_its_name_and_written_within_one_step() {
5669 let (w, desk_notes, phone_notes) = notes_world("wanted");
5670 let name = desk_writes(&w, &desk_notes, "lease.md", "The lease, signed.\n", DUE + 60);
5671 w.fake.forget_seen();
5672 let pulled = w.pull(DUE + 70);
5674 assert!(pulled.problem.is_none(), "{pulled:?}");
5675 let first = w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 71);
5676 assert_eq!(first.waits.get("no-blob"), Some(&1), "{first:?}");
5677 assert_eq!((first.received, first.pending), (0, 1));
5678 assert!(!phone_notes.join("lease.md").exists());
5679 let wanted = crate::blobs::Wanted::load(&w.phone.memory);
5680 assert_eq!(wanted.blobs.keys().cloned().collect::<Vec<_>>(), [name.clone()]);
5681 assert!(wanted.blobs[&name].size == 19 && wanted.blobs[&name].since == DUE + 71, "{wanted:?}");
5682 assert!(wanted_due(&w.phone.memory, &w.phone_folder, DUE + 71));
5683 let fetched = fetch(&w, DUE + 72, false);
5685 assert_eq!((fetched.fetched, fetched.problem.as_deref()), (1, None), "{fetched:?}");
5686 assert!(w.cache().join("blobs").join(&name).exists());
5687 assert!(!w.phone_folder.join("blobs").join(&name).exists(), "never written into the synced folder");
5688 assert!(!wanted_due(&w.phone.memory, &w.phone_folder, DUE + 72), "fetched: no longer due");
5689 let second = w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 73);
5690 assert_eq!((second.received, second.pending), (1, 0), "{second:?}");
5691 assert!(second.waits.is_empty() && second.problems.is_empty(), "{second:?}");
5692 assert_eq!(std::fs::read_to_string(phone_notes.join("lease.md")).unwrap(), "The lease, signed.\n");
5693 assert!(!crate::blobs::Wanted::path(&w.phone.memory).exists(), "nothing waits");
5694 assert_eq!(blob_gets(&w), 1, "{:?}", w.fake.seen());
5696 for line in w.fake.seen() {
5697 assert!(!(line.starts_with("PROPFIND") && line.contains("/blobs")), "{line}");
5698 assert!(line.starts_with("PROPFIND ") || line.starts_with("GET "), "{line}");
5699 }
5700 w.fake.forget_seen();
5702 assert_eq!(fetch(&w, DUE + 74, false), Pulled::default());
5703 assert!(w.fake.seen().is_empty());
5704 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
5706 assert!(!w.cache().join("blobs").join(&name).exists());
5707 }
5708
5709 #[test]
5714 fn a_damaged_copy_fetched_is_refused_and_fetched_again() {
5715 let (w, desk_notes, phone_notes) = notes_world("wanted-damaged");
5716 let name = desk_writes(&w, &desk_notes, "lease.md", "The lease, signed.\n", DUE + 60);
5717 assert!(w.pull(DUE + 70).problem.is_none());
5718 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 71).waits.get("no-blob"), Some(&1));
5719 let whole = std::fs::read(w.server.join("blobs").join(&name)).unwrap();
5721 let mut damaged = whole.clone();
5722 let middle = damaged.len() / 2;
5723 damaged[middle] ^= 0x10;
5724 put_bytes(&w.cache(), &format!("blobs/{name}"), &damaged);
5725 let outcome = w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 72);
5726 assert_eq!(outcome.waits.get("no-blob"), Some(&1), "{outcome:?}");
5727 assert!(outcome.problems.is_empty(), "the copy fetched is not the folder's: nothing said damaged, {outcome:?}");
5728 assert!(!w.cache().join("blobs").join(&name).exists());
5729 assert!(!phone_notes.join("lease.md").exists());
5730 std::fs::write(w.server.join("blobs").join(&name), &damaged).unwrap();
5732 let fetched = fetch(&w, DUE + 73, false);
5733 assert_eq!((fetched.fetched, fetched.held), (0, 1), "{fetched:?}");
5734 assert!(!w.cache().join("blobs").join(&name).exists());
5735 assert_eq!(crate::blobs::Wanted::load(&w.phone.memory).blobs[&name].again, DUE + 73 + 60);
5736 w.fake.forget_seen();
5737 assert_eq!(fetch(&w, DUE + 100, false).fetched, 0);
5738 assert_eq!(blob_gets(&w), 0, "within its wait: not asked");
5739 std::fs::write(w.server.join("blobs").join(&name), &whole).unwrap();
5741 assert_eq!(fetch(&w, DUE + 140, false).fetched, 1);
5742 let outcome = w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 141);
5743 assert_eq!(outcome.received, 1, "{outcome:?}");
5744 assert_eq!(std::fs::read_to_string(phone_notes.join("lease.md")).unwrap(), "The lease, signed.\n");
5745 }
5746
5747 #[test]
5751 fn a_sealed_file_not_on_the_server_yet_is_asked_again_within_minutes() {
5752 let (w, desk_notes, phone_notes) = notes_world("wanted-late");
5753 let name = desk_writes(&w, &desk_notes, "lease.md", "The lease, signed.\n", DUE + 60);
5754 let aside = w.base.join("aside-blob");
5755 std::fs::rename(w.server.join("blobs").join(&name), &aside).unwrap();
5756 assert!(w.pull(DUE + 70).problem.is_none());
5757 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 71).waits.get("no-blob"), Some(&1));
5758 let mut at = DUE + 72;
5759 for _ in 0..8 {
5760 let fetched = fetch(&w, at, false);
5761 assert!(fetched.fetched == 0 && fetched.problem.is_none(), "{fetched:?}");
5762 let again = crate::blobs::Wanted::load(&w.phone.memory).blobs[&name].again;
5763 assert!(again > at && again <= at + WANTED_AGAIN, "{again} after {at}");
5764 at = again;
5765 }
5766 std::fs::rename(&aside, w.server.join("blobs").join(&name)).unwrap();
5767 assert_eq!(fetch(&w, at, false).fetched, 1);
5768 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, at + 1).received, 1);
5769 }
5770
5771 #[test]
5775 fn a_sealed_file_fetched_then_brought_by_the_sync_app_is_kept_once() {
5776 let (w, desk_notes, phone_notes) = notes_world("wanted-brought");
5777 let name = desk_writes(&w, &desk_notes, "lease.md", "The lease, signed.\n", DUE + 60);
5778 assert!(w.pull(DUE + 70).problem.is_none());
5779 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 71).waits.get("no-blob"), Some(&1));
5780 assert_eq!(fetch(&w, DUE + 72, false).fetched, 1);
5781 w.down();
5783 let (fetched, brought) = (std::fs::read(w.cache().join("blobs").join(&name)).unwrap(), std::fs::read(w.phone_folder.join("blobs").join(&name)).unwrap());
5784 assert_eq!(fetched, brought, "the same file");
5785 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
5786 assert!(!w.cache().join("blobs").join(&name).exists(), "the synced folder holds it: the copy fetched goes");
5787 assert!(crate::blobs::Wanted::path(&w.phone.memory).exists(), "the note still waits");
5788 let outcome = w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 73);
5789 assert_eq!(outcome.received, 1, "{outcome:?}");
5790 assert!(!crate::blobs::Wanted::path(&w.phone.memory).exists());
5791 }
5792
5793 #[test]
5798 fn the_sealed_files_fetched_are_pruned() {
5799 let (w, desk_notes, phone_notes) = notes_world("wanted-pruned");
5800 let name = desk_writes(&w, &desk_notes, "lease.md", "The lease, signed.\n", DUE + 60);
5801 assert!(w.pull(DUE + 70).problem.is_none());
5802 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 71).waits.get("no-blob"), Some(&1));
5803 assert_eq!(fetch(&w, DUE + 72, false).fetched, 1);
5804 let blobs = w.cache().join("blobs");
5805 put(&blobs, "0123abcd", "a copy no note waits for");
5806 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
5807 assert!(blobs.join(&name).exists(), "waited for: kept");
5808 assert!(!blobs.join("0123abcd").exists());
5809 set_mtime(&blobs.join(&name), SystemTime::now() - Duration::from_secs(4 * 86_400));
5810 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
5811 assert!(!blobs.join(&name).exists(), "past a few days");
5812 assert!(!crate::blobs::Wanted::path(&w.phone.memory).exists(), "no longer waited for");
5813 w.fake.forget_seen();
5814 assert_eq!(fetch(&w, DUE + 80, false).fetched, 0);
5815 assert_eq!(blob_gets(&w), 0);
5816 assert_eq!(w.phone.exchange_notes(&w.phone_folder, &phone_notes, DUE + 81).waits.get("no-blob"), Some(&1));
5818 assert_eq!(fetch(&w, DUE + 82, false).fetched, 1);
5819 State::choose(&w.phone.memory, |s| s.on = Some(false)).unwrap();
5821 tidy(&w.phone.memory, &w.phone_folder, &w.phone.id);
5822 assert!(!blobs.join(&name).exists());
5823 assert_eq!(fetch(&w, DUE + 83, false).problem.as_deref(), Some("not-confirmed"));
5824 }
5825
5826 #[test]
5830 fn sealed_files_waited_for_come_bounded_and_paced() {
5831 let (w, _, _) = notes_world("wanted-paced");
5832 let made = |seed: u32, size: usize| {
5833 let mut x = seed.wrapping_mul(2_654_435_761) | 1;
5834 (0..size)
5835 .map(|_| {
5836 x ^= x << 13;
5837 x ^= x >> 17;
5838 x ^= x << 5;
5839 x as u8
5840 })
5841 .collect::<Vec<u8>>()
5842 };
5843 let sealed = |seed: u32, size: usize| {
5845 let source = w.base.join(format!("source-{seed}"));
5846 std::fs::write(&source, made(seed, size)).unwrap();
5847 let (hash, size) = crate::blobs::hash_file(&source).unwrap();
5848 crate::blobs::put(&w.server, &KEY, &source, &hash).unwrap();
5849 let name = crate::blobs::name(&KEY, &hash);
5850 crate::blobs::Wanted::change(&w.phone.memory, |wanted| {
5851 wanted.blobs.insert(name.clone(), crate::blobs::Want { hash, size, since: DUE, ..Default::default() });
5852 })
5853 .unwrap();
5854 name
5855 };
5856 let papers = [sealed(1, 1_500_000), sealed(2, 1_200_000)];
5857 let note = sealed(3, 3_000);
5858 let first = fetch(&w, DUE + 100, true);
5860 assert_eq!((first.fetched, first.paced), (2, 1), "{first:?}");
5861 let blobs = w.cache().join("blobs");
5862 assert!(blobs.join(¬e).exists() && blobs.join(&papers[1]).exists() && !blobs.join(&papers[0]).exists());
5863 w.fake.forget_seen();
5864 assert_eq!((fetch(&w, DUE + 400, true).fetched, blob_gets(&w)), (0, 0), "paced: not asked");
5865 assert_eq!(fetch(&w, DUE + 100 + PACE, true).fetched, 1);
5866 assert!(blobs.join(&papers[0]).exists());
5867 let notes: Vec<String> = (10..10 + WANTED_FILES as u32 + 2).map(|seed| sealed(seed, 500)).collect();
5869 assert_eq!(fetch(&w, DUE + 2_000, true).fetched, WANTED_FILES);
5870 assert_eq!(fetch(&w, DUE + 2_001, true).fetched, 2);
5871 assert!(notes.iter().all(|n| blobs.join(n).exists()));
5872 let large: Vec<String> = (40..43).map(|seed| sealed(seed, 7 << 20)).collect();
5874 assert_eq!(fetch(&w, DUE + 3_000, false).fetched, 2, "14 MiB, then the third past 16");
5875 assert_eq!(fetch(&w, DUE + 3_001, false).fetched, 1);
5876 assert!(large.iter().all(|n| blobs.join(n).exists()));
5877 }
5878}