Merge branch 'fix/diagnose-coverage' into fix/restore-outcome-and-walk-gaps

# Conflicts:
#	docs/BUGS.md
This commit is contained in:
nvrl committed 2026-10-31 19:26:41 +01:00
commit ba414c4446
27 files changed
+2530 -1181

No files matched your search

+122
View File
@@ -0,0 +1,122 @@
//! Log priorities for journald.
//!
//! journald files each line a service writes at the unit's default priority,
//! info, unless the line opens with a syslog `<N>` prefix. Without one every
//! WARN and ERROR the daemon logs is stored as info, and the
//! `journalctl -p warning` that the troubleshooting guide asks for shows
//! nothing (`docs/BUGS.md` B258).
use std::fmt;
use std::os::unix::fs::MetadataExt;
use tracing::{Event, Level, Subscriber};
use tracing_subscriber::fmt::format::Writer;
use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
use tracing_subscriber::registry::LookupSpan;
/// Whether standard output is the journal stream systemd connected this
/// process to.
///
/// `JOURNAL_STREAM` alone does not say so: a terminal started from a user unit
/// inherits it, and every `pdfs` run in that terminal would print the
/// prefixes. systemd sets it to the stream's `device:inode`, so it is compared
/// against what fd 1 actually is.
pub(crate) fn stdout_is_journal() -> bool {
let Some(stream) = std::env::var_os("JOURNAL_STREAM") else {
return false;
};
let Ok(stdout) = std::fs::metadata("/proc/self/fd/1") else {
return false;
};
stream
.to_str()
.is_some_and(|value| names_stream(value, stdout.dev(), stdout.ino()))
}
/// Whether a `JOURNAL_STREAM` value names the file `dev`:`ino`.
fn names_stream(value: &str, dev: u64, ino: u64) -> bool {
value
.split_once(':')
.is_some_and(|(d, i)| d.parse() == Ok(dev) && i.parse() == Ok(ino))
}
/// The syslog priority journald should file a `level` line at.
fn priority(level: &Level) -> u8 {
match *level {
Level::ERROR => 3,
Level::WARN => 4,
Level::INFO => 6,
Level::DEBUG | Level::TRACE => 7,
}
}
/// An event format that opens each line with its `<N>` priority, then writes
/// the line as `F` would.
pub(crate) struct Prioritized<F>(pub(crate) F);
impl<S, N, F> FormatEvent<S, N> for Prioritized<F>
where
S: Subscriber + for<'a> LookupSpan<'a>,
N: for<'a> FormatFields<'a> + 'static,
F: FormatEvent<S, N>,
{
fn format_event(
&self,
ctx: &FmtContext<'_, S, N>,
mut writer: Writer<'_>,
event: &Event<'_>,
) -> fmt::Result {
write!(writer, "<{}>", priority(event.metadata().level()))?;
self.0.format_event(ctx, writer, event)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io;
use std::sync::{Arc, Mutex};
#[derive(Clone)]
struct Buf(Arc<Mutex<Vec<u8>>>);
impl io::Write for Buf {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[test]
fn each_line_opens_with_the_priority_of_its_level() {
let buf = Buf(Arc::new(Mutex::new(Vec::new())));
let writer = buf.clone();
let subscriber = tracing_subscriber::fmt()
.with_max_level(Level::DEBUG)
.with_ansi(false)
.with_writer(move || writer.clone())
.event_format(Prioritized(tracing_subscriber::fmt::format()))
.finish();
tracing::subscriber::with_default(subscriber, || {
tracing::error!("e");
tracing::warn!("w");
tracing::info!("i");
tracing::debug!("d");
});
let out = String::from_utf8(buf.0.lock().unwrap().clone()).unwrap();
let prefixes: Vec<&str> = out.lines().map(|line| &line[..3]).collect();
assert_eq!(prefixes, ["<3>", "<4>", "<6>", "<7>"]);
}
#[test]
fn journal_stream_must_name_the_same_file() {
assert!(names_stream("8:12345", 8, 12345));
assert!(!names_stream("8:12345", 8, 54321));
assert!(!names_stream("9:12345", 8, 12345));
assert!(!names_stream("8", 8, 12345));
assert!(!names_stream("", 8, 12345));
}
}
+269 -18
View File
@@ -20,6 +20,8 @@ use pdfs_core::control::{
use pdfs_core::db::Db;
use pdfs_core::service;
mod journal;
/// `println!` for this binary, shadowing std's, which panics when stdout is
/// closed: `pdfs ls | head` ended in a "Broken pipe" panic once `head` had its
/// lines. This one hands the failure to [`stdout_written`] instead.
@@ -879,15 +881,22 @@ fn cli_error(kind: ErrorKind, message: &str) -> String {
}
fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
// `pgp` logs a benign WARN when it re-serializes a Proton secret-key
// packet with a longer (but equivalent) length encoding than the
// original; drop it to error so normal runs stay quiet.
.unwrap_or_else(|_| "info,pgp=error".into()),
)
.init();
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
// `pgp` logs a benign WARN when it re-serializes a Proton secret-key
// packet with a longer (but equivalent) length encoding than the
// original; drop it to error so normal runs stay quiet.
.unwrap_or_else(|_| "info,pgp=error".into());
if journal::stdout_is_journal() {
// journalctl colours lines by the priority the prefix gives them, so
// the escape codes would only be noise in the stored text.
tracing_subscriber::fmt()
.with_env_filter(filter)
.with_ansi(false)
.event_format(journal::Prioritized(tracing_subscriber::fmt::format()))
.init();
} else {
tracing_subscriber::fmt().with_env_filter(filter).init();
}
let cli = Cli::parse();
JSON_OUTPUT.store(cli.json, std::sync::atomic::Ordering::Relaxed);
@@ -2379,12 +2388,21 @@ where
/// an external unmount triggers a remount; errors back off and retry. This is
/// the entry point for the systemd user service.
fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
let state_dir = AppDirs::new()?.state_dir();
// A failure an earlier daemon recorded is not this one's.
pdfs_core::mountfail::clear(&state_dir);
let mut failure: Option<pdfs_core::mountfail::MountFailure> = None;
loop {
// Wait until a session is stored. The GUI enables this service on login,
// but the service may also start at boot before the user has logged in.
wait_for_session(auth::ready, pdfs_fuse::idle_sleep)?;
match mount_once(mountpoint.clone()) {
let outcome = mount_once(mountpoint.clone());
if outcome.is_ok() {
// The mount came up, so the next failure starts a new run of them.
failure = None;
}
match outcome {
Ok(pdfs_fuse::MountOutcome::Shutdown) => {
tracing::info!("daemon stopping");
return Ok(());
@@ -2399,13 +2417,51 @@ fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
Err(e) => {
// `{:#}` so the whole anyhow chain lands in the log — the top
// context alone ("mount failed") never says *why*.
tracing::error!(error = format!("{e:#}"), "mount failed; retrying in 5s");
pdfs_fuse::idle_sleep(std::time::Duration::from_secs(5));
let error = format!("{e:#}");
let now = unix_now();
let attempts = failure.as_ref().map_or(1, |f| f.attempts.saturating_add(1));
let delay = mount_retry_delay(attempts);
tracing::error!(
error,
attempts,
retry_secs = delay.as_secs(),
"mount failed; retrying"
);
let record = pdfs_core::mountfail::MountFailure {
error: error.clone(),
since: failure.as_ref().map_or(now, |f| f.since),
attempts,
next_attempt: now + delay.as_secs() as i64,
};
if let Err(e) = pdfs_core::mountfail::record(&state_dir, &record) {
tracing::warn!(error = %e, "could not record the mount failure");
}
pdfs_fuse::notify_status(&format!("Mount failing: {error}"));
failure = Some(record);
pdfs_fuse::idle_sleep(delay);
}
}
}
}
/// How long to wait before mount attempt `attempts + 1`, after `attempts`
/// failures in a row.
///
/// Doubles from 5 s to a 2-minute ceiling. A flat 5 s retry sent about four
/// requests to Proton per attempt, tens of thousands a day from an account
/// whose mount fails on a key, which no retry fixes (`docs/BUGS.md` B263). The
/// ceiling stays at minutes because a cold start without a network fails the
/// same way, and must mount soon after the network returns.
fn mount_retry_delay(attempts: u32) -> std::time::Duration {
const BASE_SECS: u64 = 5;
const MAX_SECS: u64 = 120;
let factor = 1u64
.checked_shl(attempts.saturating_sub(1))
.unwrap_or(u64::MAX);
std::time::Duration::from_secs(BASE_SECS.saturating_mul(factor).min(MAX_SECS))
}
/// One mount run: resume the session, mount, and block until the mount ends
/// (clean shutdown or external unmount). Returns the [`MountOutcome`] so the
/// daemon loop can decide whether to exit or remount.
@@ -3100,6 +3156,24 @@ fn human_secs(secs: u64) -> String {
}
}
/// The sync engine's line in `pdfs diagnostics`: whether it runs, and which
/// folder's pass it is on.
fn sync_engine_line(sync: &pdfs_core::control::SyncEngineState) -> String {
let mut line = match (sync.running, sync.pass_folder) {
(false, _) => "STOPPED; synced folders do not sync until the daemon restarts".to_string(),
(true, None) => "idle".to_string(),
(true, Some(folder)) => format!(
"pass on folder {folder} for {}, last progress {} ago",
human_secs(sync.pass_age_secs),
human_secs(sync.pass_idle_secs),
),
};
if sync.panics > 0 {
line.push_str(&format!("; {} pass(es) panicked", sync.panics));
}
line
}
fn cmd_diagnostics() -> Result<()> {
let response = control_request(CtlRequest::Diagnostics)?;
if emit_json(&response)? {
@@ -3135,6 +3209,9 @@ fn cmd_diagnostics() -> Result<()> {
human_secs(worker.age_secs)
);
}
if let Some(sync) = &d.sync {
println!("sync engine: {}", sync_engine_line(sync));
}
if d.inflight.is_empty() {
println!("in-flight requests: none");
} else {
@@ -3373,6 +3450,79 @@ fn cmd_sync_limit(up: Option<u64>, down: Option<u64>) -> Result<()> {
})?)
}
/// How long the oldest write may wait to upload before `diagnose` calls it
/// out. Under a day it is a slow link or a big file; past it, something is
/// holding the queue.
const STAGED_STALE_SECS: u64 = 86_400;
/// `diagnose` findings for the queue and the pause, from the `Status` reply.
/// `paused_for` is how long a timed pause still runs, in seconds.
fn sync_health(
failing_ops: u64,
failing_error: Option<&str>,
staged_bytes: u64,
staged_oldest_secs: u64,
paused: bool,
paused_for: Option<i64>,
) -> Vec<(DiagnoseLevel, &'static str, String)> {
let mut findings = Vec::new();
if failing_ops > 0 {
findings.push((
DiagnoseLevel::Warn,
" failing writes",
match failing_error {
Some(error) => format!("{failing_ops} keep failing; the last error: {error}"),
None => format!("{failing_ops} keep failing"),
},
));
}
if staged_bytes > 0 && staged_oldest_secs >= STAGED_STALE_SECS {
findings.push((
DiagnoseLevel::Warn,
" oldest unsent write",
format!(
"{} old, {} only on this computer",
human_secs(staged_oldest_secs),
human_bytes(staged_bytes)
),
));
}
if paused {
findings.push((
DiagnoseLevel::Warn,
" sync",
match paused_for {
Some(left) if left > 0 => {
format!("paused for another {}", human_secs(left as u64))
}
_ => "paused until resumed (`pdfs sync resume`)".to_string(),
},
));
}
findings
}
/// `diagnose` findings for the synced folders that need someone to look.
fn folder_health(
folders: &[pdfs_core::control::SyncFolderInfo],
) -> Vec<(DiagnoseLevel, String, String)> {
folders
.iter()
.filter_map(|folder| {
let what = match folder.state.as_str() {
"error" => "sync error; `pdfs sync list` and the activity feed say more",
"conflict" => "conflict copies to look at",
_ => return None,
};
Some((
DiagnoseLevel::Warn,
format!(" synced folder [{}]", folder.id),
format!("{}: {what}", folder.local_path),
))
})
.collect()
}
fn unix_now() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@@ -3709,21 +3859,26 @@ fn cmd_diagnose() -> Result<()> {
let socket_exists = socket.exists();
let socket_kind_ok =
std::fs::symlink_metadata(&socket).is_ok_and(|m| m.file_type().is_socket());
// A stale socket file with no listener is the common failure, so the reply
// is what decides this rather than the file's existence.
let status = control_request(CtlRequest::Status);
report.finding(
if socket_exists && !socket_kind_ok {
DiagnoseLevel::Fail
} else if socket_exists {
} else if socket_exists && status.is_ok() {
DiagnoseLevel::Ok
} else {
DiagnoseLevel::Warn
},
" control socket",
socket.display().to_string(),
if socket_exists && socket_kind_ok && status.is_err() {
format!("{} (left over; nothing listens on it)", socket.display())
} else {
socket.display().to_string()
},
);
// A stale socket file with no listener is the common failure, so the reply
// is what decides this rather than the file's existence.
let mut daemon_responding = false;
match control_request(CtlRequest::Status) {
match status {
Ok(CtlResponse::Status {
mountpoint,
online,
@@ -3731,6 +3886,12 @@ fn cmd_diagnose() -> Result<()> {
tokens_unsaved,
pending_uploads,
pending_changes,
failing_ops,
failing_error,
staged_bytes,
staged_oldest_secs,
paused,
paused_until,
version,
..
}) => {
@@ -3805,6 +3966,27 @@ fn cmd_diagnose() -> Result<()> {
format!("{pending} waiting to upload")
},
);
// The rest of the daemon's state that the report used to drop:
// writes that keep failing, a pause, a folder in error. Each of
// them stops changes from reaching Drive while everything above
// reads as fine (`docs/BUGS.md` B268).
for (level, check, detail) in sync_health(
failing_ops,
failing_error.as_deref(),
staged_bytes,
staged_oldest_secs,
paused,
paused_until.map(|until| until - unix_now()),
) {
report.finding(level, check, detail);
}
if let Ok(CtlResponse::SyncFolders { items }) =
control_request(CtlRequest::ListSyncFolders)
{
for (level, check, detail) in folder_health(&items) {
report.finding(level, &check, detail);
}
}
match await_integrity_check() {
Ok(CtlResponse::CacheReport {
integrity_problems,
@@ -3861,7 +4043,23 @@ fn cmd_diagnose() -> Result<()> {
" daemon responding",
format!("stopped or unreachable: {e:#}"),
);
println!(" (start it with `systemctl --user start proton-drive.service`)");
match pdfs_core::mountfail::read(&dirs.state_dir()) {
// A daemon whose mount keeps failing never opens the socket,
// so this is the only place its reason shows (B263).
Some(failure) => report.finding(
DiagnoseLevel::Fail,
" mount",
format!(
"failing for {}, {} attempt(s): {}",
human_secs(unix_now().saturating_sub(failure.since).max(0) as u64),
failure.attempts,
failure.error
),
),
None => println!(
" (start it with `systemctl --user start proton-drive.service`)"
),
}
}
}
@@ -3986,6 +4184,50 @@ mod pause_tests {
mod diagnose_tests {
use super::*;
/// A healthy queue says nothing; writes that keep failing, an old unsent
/// write and a pause each say so (`docs/BUGS.md` B268).
#[test]
fn a_stuck_queue_or_a_pause_is_reported() {
assert!(sync_health(0, None, 0, 0, false, None).is_empty());
assert!(sync_health(0, None, 4096, STAGED_STALE_SECS - 1, false, None).is_empty());
let checks: Vec<&str> = sync_health(
2,
Some("upload a.txt: quota"),
4096,
STAGED_STALE_SECS,
true,
Some(600),
)
.into_iter()
.map(|(level, check, _)| {
assert!(matches!(level, DiagnoseLevel::Warn));
check.trim()
})
.collect();
assert_eq!(checks, ["failing writes", "oldest unsent write", "sync"]);
}
/// Issue #53: a folder that would not sync, and `diagnose` found nothing.
#[test]
fn a_synced_folder_in_error_is_reported() {
let folder = |id: i64, state: &str| pdfs_core::control::SyncFolderInfo {
id,
local_path: format!("/home/u/f{id}"),
remote_uid: "vol~link".into(),
mode: "mirror".into(),
state: state.into(),
last_sync: 0,
pending_mode: None,
progress: None,
paused: false,
};
let findings =
folder_health(&[folder(1, "idle"), folder(2, "error"), folder(3, "syncing")]);
assert_eq!(findings.len(), 1);
assert_eq!(findings[0].1.trim(), "synced folder [2]");
assert!(findings[0].2.starts_with("/home/u/f2: sync error"));
}
#[test]
fn finding_severity_is_counted_centrally() {
let mut report = DiagnoseReport::default();
@@ -4021,6 +4263,15 @@ mod diagnose_tests {
#[cfg(test)]
mod login_wait_tests {
use super::*;
/// The mount retry backs off from 5 s to a ceiling of minutes, rather than
/// asking Proton every 5 s forever (`docs/BUGS.md` B263).
#[test]
fn a_failing_mount_backs_off_to_a_ceiling() {
let secs: Vec<u64> = (1..=8).map(|n| mount_retry_delay(n).as_secs()).collect();
assert_eq!(secs, [5, 10, 20, 40, 80, 120, 120, 120]);
assert_eq!(mount_retry_delay(u32::MAX).as_secs(), 120);
}
use std::time::Duration;
#[test]
+23
View File
@@ -1980,6 +1980,29 @@ pub struct Diagnostics {
pub control_handler_limit: usize,
/// Queued mutations not yet drained.
pub pending_ops: u64,
/// The folder-sync engine. `None` from a daemon that predates the field.
#[serde(default)]
pub sync: Option<SyncEngineState>,
}
/// What the folder-sync engine is doing, for [`Diagnostics`].
///
/// The engine is one thread that runs every synced folder's passes in turn.
/// Dead or stuck, it looked exactly like an idle one: workers idle, nothing
/// queued, nothing in the journal (`docs/BUGS.md` B259).
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct SyncEngineState {
/// Whether the engine thread is running. Once it has ended, nothing syncs
/// until the daemon restarts.
pub running: bool,
/// The synced folder whose pass is running, if one is.
pub pass_folder: Option<i64>,
/// How long that pass has been running.
pub pass_age_secs: u64,
/// How long since that pass last reported progress.
pub pass_idle_secs: u64,
/// Passes that ended in a panic since the daemon started.
pub panics: u64,
}
/// The daemon's reply to a [`Request`].
+52 -16
View File
@@ -160,13 +160,14 @@ impl CoreError {
/// telling the user their connection is down beats telling them the
/// server refused. The SDK's other failures say what went wrong instead
/// (see [`proton_kind`]).
///
/// Callers also hand it this crate's [`Error`] from the cache and the
/// database, which is not a transport error at all. Taking those for
/// "offline" told the user their connection was down when a pin failed on
/// the database (`docs/BUGS.md` B266), so the source chain is searched for
/// either type before falling back.
pub fn from_api(e: &(dyn std::error::Error + 'static), what: &str) -> Self {
let kind = match e.downcast_ref::<proton_sdk::ProtonError>() {
Some(proton) => proton_kind(proton),
// Not the SDK's at all: a timeout around the call, or the socket.
None => ErrorKind::Offline,
};
Self::new(kind, format!("{what}: {e}"))
Self::new(api_kind_of(e), format!("{what}: {e}"))
}
/// Prefix the message while keeping the classification.
@@ -208,19 +209,40 @@ impl From<serde_json::Error> for CoreError {
impl From<Error> for CoreError {
fn from(e: Error) -> Self {
let kind = match &e {
Error::NotLoggedIn
| Error::ReloginRequired
| Error::Locked
| Error::WrongMailboxPassword
| Error::HumanVerificationRequired(_) => ErrorKind::Denied,
Error::Proton(proton) => proton_kind(proton),
_ => ErrorKind::Internal,
};
Self::new(kind, e.to_string())
Self::new(core_kind(&e), e.to_string())
}
}
/// What this crate's [`Error`] means to a front-end.
fn core_kind(e: &Error) -> ErrorKind {
match e {
Error::NotLoggedIn
| Error::ReloginRequired
| Error::Locked
| Error::WrongMailboxPassword
| Error::HumanVerificationRequired(_) => ErrorKind::Denied,
Error::Proton(proton) => proton_kind(proton),
_ => ErrorKind::Internal,
}
}
/// The kind for [`CoreError::from_api`]: the first SDK or crate error in `e`'s
/// source chain decides it.
fn api_kind_of(e: &(dyn std::error::Error + 'static)) -> ErrorKind {
let mut cause = Some(e);
while let Some(err) = cause {
if let Some(proton) = err.downcast_ref::<proton_sdk::ProtonError>() {
return proton_kind(proton);
}
if let Some(core) = err.downcast_ref::<Error>() {
return core_kind(core);
}
cause = err.source();
}
// Not the SDK's at all: a timeout around the call, or the socket.
ErrorKind::Offline
}
/// What a failed SDK call means to a front-end.
///
/// Only a transport failure is the connection. The rest happened on this
@@ -414,6 +436,20 @@ mod tests {
assert_ne!(e.kind, ErrorKind::Offline);
}
/// A pin or a search that fails on the database is not the connection, so
/// it must not say "You're offline" (`docs/BUGS.md` B266). An SDK error
/// wrapped in this crate's type classifies as the SDK error does.
#[test]
fn a_database_error_handed_to_from_api_is_not_offline() {
let db = Error::Other("database is locked".into());
assert_eq!(CoreError::from_api(&db, "pin").kind, ErrorKind::Internal);
let wrapped = Error::Proton(api(C::DoesNotExist));
assert_eq!(
CoreError::from_api(&wrapped, "load node").kind,
ErrorKind::NotFound
);
}
/// `Error` is the plumbing type; converting it must classify as sharply as
/// the direct path, or the two disagree about the same failure.
#[test]
+1
View File
@@ -15,6 +15,7 @@ pub mod kernelkey;
pub mod keycheck;
pub mod localindex;
pub mod menu;
pub mod mountfail;
pub mod mounts;
pub mod opener;
pub mod places;
+75
View File
@@ -0,0 +1,75 @@
//! Why the daemon's mount keeps failing, for the front ends to read.
//!
//! The control socket opens only once a mount has succeeded, so a daemon whose
//! mount fails on every try had no way to say why. The app and the tray showed
//! "Connecting…" for days while the reason sat in the journal (`docs/BUGS.md`
//! B263). The daemon now writes its last failure to a small file in the state
//! directory, next to the control socket, and removes it once a mount is up.
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
/// The file's name inside the state directory.
const FILE_NAME: &str = "mount-failure.json";
/// The last failed mount attempt.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct MountFailure {
/// The error, in English, as the journal has it.
pub error: String,
/// Unix seconds of the first failure in this run of failures.
pub since: i64,
/// Failed attempts in a row.
pub attempts: u32,
/// Unix seconds of the next attempt.
pub next_attempt: i64,
}
/// Where the failure is kept for the state directory `state_dir`.
pub fn path(state_dir: &Path) -> PathBuf {
state_dir.join(FILE_NAME)
}
/// Record `failure`, replacing the one before it. Written to a temporary file
/// and renamed, so a reader never sees half of it.
pub fn record(state_dir: &Path, failure: &MountFailure) -> std::io::Result<()> {
let json = serde_json::to_vec(failure).map_err(std::io::Error::other)?;
let tmp = state_dir.join(format!("{FILE_NAME}.tmp"));
std::fs::write(&tmp, json)?;
std::fs::rename(&tmp, path(state_dir))
}
/// Forget the failure: a mount is up, or a new daemon has started.
pub fn clear(state_dir: &Path) {
let _ = std::fs::remove_file(path(state_dir));
}
/// The recorded failure, if there is one and it can be read.
pub fn read(state_dir: &Path) -> Option<MountFailure> {
let bytes = std::fs::read(path(state_dir)).ok()?;
serde_json::from_slice(&bytes).ok()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_recorded_failure_reads_back_until_cleared() {
let dir = std::env::temp_dir().join(format!("pdfs-mountfail-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
assert_eq!(read(&dir), None);
let failure = MountFailure {
error: "fetch My Files root: missing key".into(),
since: 1_700_000_000,
attempts: 3,
next_attempt: 1_700_000_040,
};
record(&dir, &failure).unwrap();
assert_eq!(read(&dir), Some(failure));
clear(&dir);
assert_eq!(read(&dir), None);
let _ = std::fs::remove_dir_all(&dir);
}
}
+1
View File
@@ -163,6 +163,7 @@ impl Core {
control_handlers: super::control::active_handlers(),
control_handler_limit: super::control::handler_limit(),
pending_ops: self.pending_ops_quickly(),
sync: Some(self.sync_engine.report()),
}
}
+4 -1
View File
@@ -124,7 +124,7 @@ use mount::{
};
use reads::{BlockFlight, BlockRing, PREFETCH_BUDGET, Prefetch, ReaderSlot, STREAM_BYPASS_MIN};
use state::{Entry, Intervals, PendingRevision, State, StateGuard, WriteHandle};
pub use systemd::idle_sleep;
pub use systemd::{idle_sleep, status as notify_status};
use tracing::{debug, error, info, warn};
use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
use r#virtual::{
@@ -728,6 +728,8 @@ struct Core {
/// can say what a pass is doing rather than just "syncing". An entry exists
/// only while that folder's reconcile pass is running.
sync_progress: Arc<Mutex<HashMap<i64, SyncProgress>>>,
/// Whether the folder-sync engine is alive, and which pass it is on.
sync_engine: Arc<sync::EngineHealth>,
/// Channel to the folder-sync engine: nudges it to
/// reconcile a folder, reconcile everything, or re-scan its watch set.
sync_tx: std::sync::mpsc::Sender<sync::SyncMsg>,
@@ -5989,6 +5991,7 @@ impl Core {
if let Some(p) = self.sync_progress.lock().get_mut(&folder_id) {
f(p);
}
self.sync_engine.moved();
}
/// Set the number of items the scan expects to check, from the size of the
+72 -28
View File
@@ -507,6 +507,7 @@ pub(crate) fn mount_with(
events: Arc::new(Default::default()),
indexing: Arc::new(AtomicBool::new(false)),
sync_progress: Arc::new(Mutex::new(HashMap::new())),
sync_engine: Arc::default(),
sync_tx,
mounts: Arc::new(Mutex::new(SecondaryMountRegistry::default())),
sync_locks: Arc::new(Mutex::new(HashMap::new())),
@@ -540,12 +541,11 @@ pub(crate) fn mount_with(
// and sweeps all operating on a mount that no longer exists (`docs/BUGS.md` B44).
let mut workers: Vec<std::thread::JoinHandle<()>> = Vec::new();
for worker in 0..DRAIN_WORKERS {
let core = core.clone();
workers.push(
std::thread::Builder::new()
.name(format!("pdfs-drain-{worker}"))
.spawn(move || core.run_pending_drain(worker == 0))?,
);
let drain_core = core.clone();
let spawned = std::thread::Builder::new()
.name(format!("pdfs-drain-{worker}"))
.spawn(move || drain_core.run_pending_drain(worker == 0));
keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
}
// Start the folder-sync engine. It watches every mirror folder, polls the
@@ -565,24 +565,22 @@ pub(crate) fn mount_with(
info!("conflict sweep disabled by configuration");
} else {
info!(mode = ?sweep_mode, "conflict sweep enabled");
let core = core.clone();
workers.push(
std::thread::Builder::new()
.name("pdfs-conflict-sweep".into())
.spawn(move || core.run_conflict_sweep_loop())?,
);
let sweep_core = core.clone();
let spawned = std::thread::Builder::new()
.name("pdfs-conflict-sweep".into())
.spawn(move || sweep_core.run_conflict_sweep_loop());
keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
}
// Watch the link for the life of the mount: a mount from the cache waits for
// the network to come back, and an online one can lose it at any moment and
// must notice it coming back just the same.
{
let core = core.clone();
workers.push(
std::thread::Builder::new()
.name("pdfs-online-probe".into())
.spawn(move || core.run_online_probe())?,
);
let probe_core = core.clone();
let spawned = std::thread::Builder::new()
.name("pdfs-online-probe".into())
.spawn(move || probe_core.run_online_probe());
keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
}
// Keep the launcher prompt's "This computer" index fresh. Its own thread:
@@ -593,11 +591,10 @@ pub(crate) fn mount_with(
let transfers = core.transfers.clone();
let mountpoint = mountpoint.to_path_buf();
let shutdown = core.shutdown.clone();
workers.push(
std::thread::Builder::new()
.name("pdfs-localindex".into())
.spawn(move || run_local_index(db, indexing, transfers, mountpoint, shutdown))?,
);
let spawned = std::thread::Builder::new()
.name("pdfs-localindex".into())
.spawn(move || run_local_index(db, indexing, transfers, mountpoint, shutdown));
keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
}
// Bind the control socket before the FUSE session takes over the thread. A
@@ -606,12 +603,18 @@ pub(crate) fn mount_with(
let old_umask = unsafe { libc::umask(0o77) };
let listener_res = UnixListener::bind(control_socket);
unsafe { libc::umask(old_umask) };
let listener = listener_res?;
let listener = match listener_res {
Ok(listener) => listener,
Err(error) => {
abandon_workers(&core, control_socket, workers);
return Err(error);
}
};
// Owner-only before anything can connect: a peer on this socket commands the
// daemon's authenticated session without a credential of its own (B6).
if let Err(e) = pdfs_core::config::restrict_socket(control_socket) {
error!(error = %e, "cannot restrict control socket permissions; refusing to serve");
let _ = std::fs::remove_file(control_socket);
abandon_workers(&core, control_socket, workers);
return Err(std::io::Error::other(format!(
"control socket permissions: {e}"
)));
@@ -622,7 +625,7 @@ pub(crate) fn mount_with(
let bg = match spawn_session(&core, mountpoint, root) {
Ok(session) => session,
Err(error) => {
let _ = std::fs::remove_file(control_socket);
abandon_workers(&core, control_socket, workers);
return Err(error);
}
};
@@ -642,7 +645,7 @@ pub(crate) fn mount_with(
Err(error) => {
// A failure here abandons the mount, so the background threads
// started above have to be told to stop with it.
stop_workers(&core, control_socket, workers);
abandon_workers(&core, control_socket, workers);
let secondaries = core.mounts.lock().close_and_drain();
for (id, mount) in secondaries {
if let Err(unmount_error) = mount.teardown() {
@@ -692,8 +695,12 @@ pub(crate) fn mount_with(
}
}
// The mount is live and the control socket answers: this is the moment the
// unit is actually usable, which is what `Type=notify` reports.
// unit is actually usable, which is what `Type=notify` reports. Whatever
// failure an earlier attempt recorded for the front ends is over.
systemd::ready();
if let Some(state_dir) = control_socket.parent() {
pdfs_core::mountfail::clear(state_dir);
}
// Re-establish on-demand mounts only after the primary session is live.
// Keep the worker so shutdown can close publication, wait for any in-flight
@@ -859,6 +866,43 @@ fn stop_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::Jo
// Say so before the joins: a teardown that takes a while must not be read as
// a missed watchdog ping and turned into a kill.
systemd::stopping();
halt_workers(core, control_socket, workers);
}
/// Stop the threads a mount attempt started before it failed.
///
/// The early returns used to leave them running: every retry started another
/// set beside the last one's, and their `Core` kept the database open, so the
/// next attempt could fail on its lock instead of the real error
/// (`docs/BUGS.md` B264). Unlike [`stop_workers`] this does not tell systemd
/// the daemon is stopping, because it is about to try again, and a manager
/// told so would wait for the process to exit and then kill it.
fn abandon_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
halt_workers(core, control_socket, workers);
}
/// Keep a worker the mount just spawned, or, if the spawn failed, stop the
/// ones already running and hand the error back.
fn keep_or_abandon(
core: &Core,
control_socket: &Path,
workers: &mut Vec<std::thread::JoinHandle<()>>,
spawned: std::io::Result<std::thread::JoinHandle<()>>,
) -> std::io::Result<()> {
match spawned {
Ok(handle) => {
workers.push(handle);
Ok(())
}
Err(error) => {
abandon_workers(core, control_socket, std::mem::take(workers));
Err(error)
}
}
}
/// Signal every worker to stop, then join them up to [`WORKER_JOIN_DEADLINE`].
fn halt_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
core.shutdown.stop();
core.wake_drain();
let _ = core.sync_tx.send(sync::SyncMsg::Stop);
+67 -1
View File
@@ -36,6 +36,11 @@ const TICK: Duration = Duration::from_secs(15);
/// would train everyone to ignore the real one.
const STALL_AFTER: Duration = Duration::from_secs(120);
/// How long a sync pass may go without reporting progress before it is called
/// out. Much longer than [`STALL_AFTER`]: one large file in a pass reports
/// nothing until its whole transfer is done.
const SYNC_STALL_AFTER: Duration = Duration::from_secs(900);
/// How often resident size goes into the log.
const RSS_EVERY: Duration = Duration::from_secs(300);
@@ -137,6 +142,26 @@ fn report_stalls(report: &pdfs_core::control::Diagnostics, last_completed: (u64,
"the transfer lane has not completed a job since the last check"
);
}
// The engine is one thread for every synced folder, so a dead or stuck one
// stops all of them while the worker pool looks idle (B259).
if let Some(sync) = &report.sync {
if !sync.running {
stalled = true;
warn!(
"the sync engine is not running; synced folders do not sync until the daemon restarts"
);
} else if let Some(folder) = sync.pass_folder
&& sync.pass_idle_secs >= SYNC_STALL_AFTER.as_secs()
{
stalled = true;
warn!(
folder,
age_secs = sync.pass_age_secs,
idle_secs = sync.pass_idle_secs,
"a sync pass has not reported progress for a long time"
);
}
}
stalled
}
@@ -169,7 +194,7 @@ fn probe(control_socket: &Path) -> bool {
#[cfg(test)]
mod tests {
use super::*;
use pdfs_core::control::{Diagnostics, InflightRequest, WorkerState};
use pdfs_core::control::{Diagnostics, InflightRequest, SyncEngineState, WorkerState};
fn busy_worker(age_secs: u64) -> WorkerState {
WorkerState {
@@ -226,4 +251,45 @@ mod tests {
assert!(!report_stalls(&report, (0, 30)));
assert!(report_stalls(&report, (0, 40)));
}
fn with_sync(sync: SyncEngineState) -> Diagnostics {
Diagnostics {
sync: Some(sync),
..Default::default()
}
}
/// A sync engine that has ended stops every synced folder while the
/// workers look idle, which is what issue #53 looked like (B259).
#[test]
fn a_stopped_sync_engine_is_a_stall() {
assert!(report_stalls(
&with_sync(SyncEngineState::default()),
(0, 0)
));
}
/// A pass that keeps reporting progress is working, however long it runs;
/// one that has gone quiet past the threshold is not.
#[test]
fn only_a_quiet_sync_pass_is_a_stall() {
let busy = SyncEngineState {
running: true,
pass_folder: Some(3),
pass_age_secs: SYNC_STALL_AFTER.as_secs() * 4,
pass_idle_secs: 5,
panics: 0,
};
assert!(!report_stalls(&with_sync(busy.clone()), (0, 0)));
let quiet = SyncEngineState {
pass_idle_secs: SYNC_STALL_AFTER.as_secs(),
..busy
};
assert!(report_stalls(&with_sync(quiet), (0, 0)));
let idle = SyncEngineState {
running: true,
..Default::default()
};
assert!(!report_stalls(&with_sync(idle), (0, 0)));
}
}
+94 -1
View File
@@ -49,7 +49,9 @@ mod local_scan;
mod model;
mod planner;
pub(crate) use engine::SYNC_CONCURRENCY;
pub(crate) use engine::{SyncMsg, spawn};
pub(crate) use engine::{EngineHealth, SyncMsg, spawn};
#[cfg(test)]
use engine::{Running, burst_order, keep_wanted};
#[cfg(test)]
use engine::{classify, event_target, settle_with};
#[cfg(test)]
@@ -67,6 +69,11 @@ use planner::{
plan_file, rel_to_path, remote_sig, remote_subtree_changed, revision_moved, tracked_baseline,
unchanged_remote_size,
};
/// A pass that took at least this long is logged at info even when it changed
/// nothing, so the journal says where a slow engine spends its time.
const SLOW_PASS: Duration = Duration::from_secs(30);
/// What a [`Pending`] op did, so the reconcile can update shared state on the
/// engine thread (never inside a task).
enum Applied {
@@ -154,6 +161,14 @@ impl Core {
// The cost is one small file read against a pass that is about to walk the
// whole tree.
let rules = self.ignore_rules(&local_root);
// A first pass is the one a user waits on after adding or restoring a
// folder, so it is said at info; the poll's passes stay at debug.
if folder.last_sync == 0 {
info!(id = folder.id, path = %folder.local_path, "sync: first pass starting");
} else {
debug!(id = folder.id, "sync: pass starting");
}
let started = Instant::now();
self.progress_begin(folder.id);
let result = if push_only {
self.push_pass(folder.id, &local_root, &remote_root, &rules)
@@ -195,6 +210,18 @@ impl Core {
"idle"
};
let _ = self.db.sync_folder_set_state(folder.id, state, now_secs());
let elapsed = started.elapsed();
if folder.last_sync == 0 || !outcome.is_empty() || elapsed >= SLOW_PASS {
info!(
id = folder.id,
state,
elapsed_ms = elapsed.as_millis() as u64,
summary = %outcome.summary(),
"sync: pass finished"
);
} else {
debug!(id = folder.id, state, "sync: pass finished");
}
self.report_unscanned(folder.id, name, &outcome.unscanned);
// Summarise a pass that actually moved something. A poll over an
// unchanged folder does nothing and says nothing — otherwise the
@@ -2478,6 +2505,72 @@ fn system_mtime_ns(meta: &std::fs::Metadata) -> Option<i64> {
mod tests {
use super::*;
/// A panic in a pass ends that pass and is counted; the engine is not
/// left believing the pass still runs (`docs/BUGS.md` B259).
#[test]
fn a_panicking_pass_is_contained_and_counted() {
let health = EngineHealth::default();
assert!(!health.contain(7, || panic!("a pass that panics")));
let report = health.report();
assert_eq!(report.panics, 1);
assert_eq!(report.pass_folder, None);
assert!(health.contain(7, || {}));
assert_eq!(health.report().panics, 1);
}
/// Folders named by id run before the rest of a `ReconcileAll` in the
/// same burst, so a folder just added does not wait behind a pass over
/// every other folder (`docs/BUGS.md` B260).
#[test]
fn named_folders_run_before_the_rest_of_a_burst() {
let ids: HashSet<i64> = [9, 4].into();
assert_eq!(burst_order(&ids, &[1, 4, 6, 9]), [4, 9, 1, 6]);
assert_eq!(burst_order(&HashSet::new(), &[1, 4]), [1, 4]);
// A folder the list no longer has still gets the pass it asked for.
assert_eq!(burst_order(&[12].into(), &[1]), [12, 1]);
}
/// A folder removed and added again keeps its watch under the new id;
/// a folder that is gone loses its watch (`docs/BUGS.md` B272).
#[test]
fn a_re_added_folder_keeps_its_watch_under_the_new_id() {
let mut have = vec![(PathBuf::from("/a"), 1), (PathBuf::from("/b"), 2)];
let dropped = keep_wanted(&mut have, &[(PathBuf::from("/a"), 7)]);
assert_eq!(have, [(PathBuf::from("/a"), 7)]);
assert_eq!(dropped, [PathBuf::from("/b")]);
}
/// While a pass runs, the report names its folder, which is what tells a
/// long pass apart from an idle engine.
#[test]
fn the_running_pass_is_reported() {
let health = EngineHealth::default();
let mut seen = None;
health.contain(7, || seen = health.report().pass_folder);
assert_eq!(seen, Some(7));
}
/// The engine reads as running only while its thread lives, including
/// when the thread ends in a panic.
#[test]
fn the_engine_reads_as_stopped_once_its_thread_ends() {
let health = Arc::new(EngineHealth::default());
let thread_health = health.clone();
let (started_tx, started_rx) = std::sync::mpsc::channel();
let (end_tx, end_rx) = std::sync::mpsc::channel::<()>();
let thread = std::thread::spawn(move || {
let _running = Running::start(&thread_health);
started_tx.send(()).unwrap();
let _ = end_rx.recv();
panic!("the engine thread panics");
});
started_rx.recv().unwrap();
assert!(health.report().running);
end_tx.send(()).unwrap();
assert!(thread.join().is_err());
assert!(!health.report().running);
}
/// A removal of something already gone has produced the state the sync
/// engine wanted, so its baseline is dropped rather than retried forever.
/// Anything else stays an error and keeps the baseline (`docs/BUGS.md` B42).
+253 -72
View File
@@ -1,5 +1,11 @@
//! Filesystem watcher, polling, and debounce control plane.
use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use pdfs_core::control::SyncEngineState;
use tracing::error;
use super::*;
/// How long the watcher must go quiet before a burst counts as finished.
@@ -23,6 +29,9 @@ const POLL_INTERVAL: Duration = Duration::from_secs(120);
/// unusable done one-at-a-time; this bounds concurrency like the bulk-upload
/// engine does.
pub(crate) const SYNC_CONCURRENCY: usize = 8;
/// How long after a failed watch the engine tries the watches again. Each try
/// walks the folder's whole tree, so not on every burst.
const REWATCH_RETRY: Duration = Duration::from_secs(600);
/// A message to the sync engine's control thread.
pub(crate) enum SyncMsg {
@@ -53,32 +62,159 @@ pub(crate) fn spawn(core: Core, rx: Receiver<SyncMsg>) -> Option<std::thread::Jo
}
}
/// Whether the engine is alive and which pass it is on, for
/// [`Core::diagnostics`] and the supervisor.
///
/// A panic in a pass used to end the engine thread. Every later add, sync-now,
/// file event and poll then went into a channel nobody read, the folder stayed
/// `syncing`, and nothing said so until a restart (`docs/BUGS.md` B259).
#[derive(Default)]
pub(crate) struct EngineHealth {
running: AtomicBool,
pass: Mutex<Option<PassClock>>,
panics: AtomicU64,
}
#[derive(Clone, Copy)]
struct PassClock {
folder: i64,
started: Instant,
moved: Instant,
}
impl EngineHealth {
/// Note that the running pass reported progress.
pub(crate) fn moved(&self) {
if let Some(pass) = self.pass.lock().as_mut() {
pass.moved = Instant::now();
}
}
/// The engine's state now. `try_lock`, like the rest of the diagnostics:
/// a report must not wait on the thing it reports about.
pub(crate) fn report(&self) -> SyncEngineState {
let pass = self.pass.try_lock().and_then(|pass| *pass);
SyncEngineState {
running: self.running.load(Ordering::Relaxed),
pass_folder: pass.map(|p| p.folder),
pass_age_secs: pass.map_or(0, |p| p.started.elapsed().as_secs()),
pass_idle_secs: pass.map_or(0, |p| p.moved.elapsed().as_secs()),
panics: self.panics.load(Ordering::Relaxed),
}
}
/// Run `pass` over `folder` as the current pass. Returns `false` when it
/// panicked: the panic ends that pass, not the engine.
pub(super) fn contain(&self, folder: i64, pass: impl FnOnce()) -> bool {
let now = Instant::now();
*self.pass.lock() = Some(PassClock {
folder,
started: now,
moved: now,
});
let finished = std::panic::catch_unwind(AssertUnwindSafe(pass)).is_ok();
*self.pass.lock() = None;
if !finished {
self.panics.fetch_add(1, Ordering::Relaxed);
}
finished
}
}
/// Marks the engine running for as long as it lives.
pub(super) struct Running<'a>(&'a EngineHealth);
impl<'a> Running<'a> {
pub(super) fn start(health: &'a EngineHealth) -> Self {
health.running.store(true, Ordering::Relaxed);
Self(health)
}
}
impl Drop for Running<'_> {
fn drop(&mut self) {
self.0.running.store(false, Ordering::Relaxed);
if std::thread::panicking() {
error!("sync engine stopped by a panic; nothing syncs until the daemon restarts");
}
}
}
/// One folder's pass, contained: a panic in it sets the folder to `error` and
/// leaves the engine running for the next one.
fn run_pass(core: &Core, folder: &StoredSyncFolder) {
if core
.sync_engine
.contain(folder.id, || core.reconcile_folder(folder))
{
return;
}
core.progress_end(folder.id);
let _ = core
.db
.sync_folder_set_state(folder.id, "error", folder.last_sync);
error!(
id = folder.id,
path = %folder.local_path,
"sync: pass panicked; the folder is set to error and tried again on the next poll"
);
core.log_activity(
ActivityKind::Sync,
format!("sync failed for {}", base_name(&folder.local_path)),
"an internal error stopped the pass; it is tried again on the next poll",
false,
);
}
/// The pass a `Reconcile(id)` asked for. A folder removed since, or one the
/// database cannot read, is said so rather than dropped in silence.
fn run_pass_by_id(core: &Core, id: i64) {
match core.db.sync_folder_get(id) {
Ok(Some(folder)) => run_pass(core, &folder),
Ok(None) => debug!(id, "sync: folder removed before its pass; skipping"),
Err(e) => warn!(id, error = ?e, "sync: cannot read folder; skipping its pass"),
}
}
/// The engine control loop. Single-threaded so reconcile passes never overlap.
fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
let health = core.sync_engine.clone();
let _running = Running::start(&health);
// Paths the watcher currently covers, mapped to their folder id, so an event
// path can be resolved back to the folder it belongs to.
let watched: Arc<Mutex<Vec<(PathBuf, i64)>>> = Arc::new(Mutex::new(Vec::new()));
let tx_events = core.sync_tx.clone();
let watched_events = watched.clone();
let mut watcher =
match notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
let Ok(event) = res else { return };
// Ignore pure metadata/access noise; content changes are what matter.
if matches!(event.kind, notify::EventKind::Access(_)) {
return;
}
let msg = event_target(&watched_events.lock(), &event.paths);
let _ = tx_events.send(msg);
}) {
Ok(w) => w,
let handler = move |res: notify::Result<notify::Event>| {
let event = match res {
Ok(event) => event,
// Running out of inotify watches for a new subfolder, say: that
// subfolder's changes are then found by the poll only (B262).
Err(e) => {
warn!(error = %e, "filesystem watcher unavailable; sync is poll-only");
// Still run the loop so polling and explicit SyncNow work.
poll_only_loop(&core, rx);
warn!(error = %e, "sync: filesystem watcher error");
return;
}
};
// Ignore pure metadata/access noise; content changes are what matter.
if matches!(event.kind, notify::EventKind::Access(_)) {
return;
}
let msg = event_target(&watched_events.lock(), &event.paths);
let _ = tx_events.send(msg);
};
// The local walk skips symlinks, so following them here would only watch
// trees outside the folder, and run passes over it for changes made there.
let config = notify::Config::default().with_follow_symlinks(false);
let mut watcher = match notify::RecommendedWatcher::new(handler, config) {
Ok(w) => w,
Err(e) => {
warn!(error = %e, "filesystem watcher unavailable; sync is poll-only");
// Still run the loop so polling and explicit SyncNow work.
poll_only_loop(&core, rx);
return;
}
};
// Periodic remote poll.
{
@@ -96,8 +232,9 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
.ok();
}
rewatch(&core, &mut watcher, &watched);
reconcile_all(&core);
let mut unwatched = rewatch(&core, &mut watcher, &watched);
let mut last_rewatch = Instant::now();
run_burst(&core, &HashSet::new(), true);
while let Ok(msg) = rx.recv() {
let mut ids: HashSet<i64> = HashSet::new();
@@ -110,21 +247,16 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
return;
}
// A folder whose watch failed is tried again now and then, not only
// when the folder set changes.
if unwatched && last_rewatch.elapsed() >= REWATCH_RETRY {
do_rewatch = true;
}
if do_rewatch {
rewatch(&core, &mut watcher, &watched);
}
if all {
reconcile_all(&core);
} else {
for id in ids {
if core.sync_paused() {
break;
}
if let Ok(Some(folder)) = core.db.sync_folder_get(id) {
core.reconcile_folder(&folder);
}
}
unwatched = rewatch(&core, &mut watcher, &watched);
last_rewatch = Instant::now();
}
run_burst(&core, &ids, all);
}
}
@@ -141,7 +273,7 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
}
});
}
reconcile_all(core);
run_burst(core, &HashSet::new(), true);
while let Ok(msg) = rx.recv() {
let mut ids: HashSet<i64> = HashSet::new();
let mut all = false;
@@ -152,21 +284,59 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
info!("sync engine stopping");
return;
}
if all {
reconcile_all(core);
} else {
for id in ids {
if core.sync_paused() {
break;
}
if let Ok(Some(folder)) = core.db.sync_folder_get(id) {
core.reconcile_folder(&folder);
}
run_burst(core, &ids, all);
}
}
/// Run the passes one burst asked for: the folders named by id, then, when the
/// burst also holds a `ReconcileAll`, every other folder.
///
/// The named folders go first. A poll's `ReconcileAll` in the same burst used
/// to take their place, and a folder just added, which has the highest id,
/// waited behind a full pass over every other folder (`docs/BUGS.md` B260).
///
/// Skipped while sync is paused, and cut short when a pause lands between
/// passes: the resume sends a fresh [`SyncMsg::ReconcileAll`], which walks
/// everything the watcher reported meanwhile.
fn run_burst(core: &Core, ids: &HashSet<i64>, all: bool) {
if core.sync_paused() {
debug!("sync: paused, skipping reconcile");
return;
}
let folders = if all {
match core.db.sync_folder_list() {
Ok(folders) => folders,
Err(e) => {
warn!(error = ?e, "sync: cannot list folders");
Vec::new()
}
}
} else {
Vec::new()
};
let listed: Vec<i64> = folders.iter().map(|f| f.id).collect();
for id in burst_order(ids, &listed) {
if core.sync_paused() {
break;
}
// A folder named by id is read again: it may have been added or
// changed after the list was read.
match folders.iter().find(|f| f.id == id) {
Some(folder) if !ids.contains(&id) => run_pass(core, folder),
_ => run_pass_by_id(core, id),
}
}
}
/// The order of a burst's passes: the folders named by id, lowest first, then
/// the rest of `listed` in its own order.
pub(super) fn burst_order(ids: &HashSet<i64>, listed: &[i64]) -> Vec<i64> {
let mut order: Vec<i64> = ids.iter().copied().collect();
order.sort_unstable();
order.extend(listed.iter().copied().filter(|id| !ids.contains(id)));
order
}
/// Absorb the rest of an event burst, returning once the watcher has been quiet
/// for `quiet` or `cap` has elapsed since the first event — whichever comes
/// first. The caller has already classified the event that opened the burst.
@@ -238,26 +408,6 @@ pub(super) fn classify(
true
}
/// Reconcile every mirror folder in turn. Skipped while sync is paused: the
/// resume sends a fresh [`SyncMsg::ReconcileAll`], which walks everything the
/// watcher reported meanwhile.
fn reconcile_all(core: &Core) {
if core.sync_paused() {
debug!("sync: paused, skipping reconcile");
return;
}
let folders = match core.db.sync_folder_list() {
Ok(f) => f,
Err(e) => {
warn!(error = ?e, "sync: cannot list folders");
return;
}
};
for folder in folders {
core.reconcile_folder(&folder);
}
}
/// The reconcile a watcher event asks for: the one mirror folder all its paths
/// lie in, so an edit in one folder does not walk every other one too. An event
/// that spans folders, or that no watched folder claims (a watch list a moment
@@ -278,7 +428,8 @@ pub(super) fn event_target(watched: &[(PathBuf, i64)], paths: &[PathBuf]) -> Syn
}
/// Bring the filesystem watches in line with the current mirror-folder set:
/// watch newly-added folders, drop removed ones.
/// watch newly-added folders, drop removed ones. Returns whether a folder is
/// left unwatched because its watch failed.
///
/// The watch calls run on a copy, outside the lock: the watcher's event
/// callback takes the same lock, and a backend that hands a watch request to
@@ -287,7 +438,7 @@ fn rewatch(
core: &Core,
watcher: &mut notify::RecommendedWatcher,
watched: &Mutex<Vec<(PathBuf, i64)>>,
) {
) -> bool {
let want: Vec<(PathBuf, i64)> = match core.db.sync_folder_list() {
Ok(f) => f
.into_iter()
@@ -296,30 +447,60 @@ fn rewatch(
.collect(),
Err(e) => {
warn!(error = ?e, "sync: cannot list folders to watch");
return;
return false;
}
};
let mut have = watched.lock().clone();
// Drop watches no longer wanted.
have.retain(|(path, _)| {
if want.iter().any(|(p, _)| p == path) {
true
} else {
let _ = watcher.unwatch(path);
false
}
});
for path in keep_wanted(&mut have, &want) {
let _ = watcher.unwatch(&path);
}
// Add newly wanted watches.
let mut failed = false;
for (path, id) in &want {
if have.iter().any(|(p, _)| p == path) {
continue;
}
match watcher.watch(path, RecursiveMode::Recursive) {
Ok(()) => have.push((path.clone(), *id)),
Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"),
Err(e) => {
// A recursive watch that fails partway keeps the watches it had
// added. Their events would match no folder and run a pass over
// every one, and they hold watches the next folder needs (B262).
let _ = watcher.unwatch(path);
failed = true;
warn!(
path = %path.display(),
error = %e,
"sync: watch failed; the poll finds this folder's changes until a retry \
works. A low fs.inotify.max_user_watches is the usual cause"
);
}
}
}
*watched.lock() = have;
failed
}
/// Keep the watches in `have` that `want` still asks for, and return the paths
/// of the others, to unwatch.
///
/// A path is matched alone, but takes the id `want` gives it: a folder removed
/// and added again in one burst keeps its watch, and its events go to the new
/// row rather than to the removed one (`docs/BUGS.md` B272).
pub(super) fn keep_wanted(have: &mut Vec<(PathBuf, i64)>, want: &[(PathBuf, i64)]) -> Vec<PathBuf> {
let mut dropped = Vec::new();
have.retain_mut(|(path, id)| match want.iter().find(|(p, _)| p == path) {
Some((_, wanted)) => {
*id = *wanted;
true
}
None => {
dropped.push(path.clone());
false
}
});
dropped
}
// ---- reconcile ------------------------------------------------------------
+9 -1
View File
@@ -64,11 +64,19 @@ fn send_abstract(socket: &UnixDatagram, name: &str, message: &str) -> bool {
/// so `systemctl --user start` returns when the mount actually works rather than
/// when the process exists.
pub(crate) fn ready() {
if notify("READY=1\n") {
if notify("READY=1\nSTATUS=Mounted\n") {
debug!("notified systemd: ready");
}
}
/// Set the one-line status `systemctl --user status` shows for the unit, such
/// as why the mount keeps failing (`docs/BUGS.md` B263).
pub fn status(text: &str) {
// A newline would end the `STATUS=` line and start another variable.
let line = text.replace('\n', " ");
notify(&format!("STATUS={line}\n"));
}
/// Tell the manager the daemon is still healthy.
pub(crate) fn watchdog_ping() {
let _ = notify("WATCHDOG=1\n");
+29 -7
View File
@@ -1797,13 +1797,35 @@ pub(crate) fn service_unreachable(
);
return;
}
status(
&ui,
"network-idle-symbolic",
&gettext("Connecting…"),
&gettext("Waiting for the Proton Drive service to start."),
false,
);
// Running but not answering may be a mount that fails on every try,
// which never opens the socket. Its reason is in a file the daemon
// keeps for this (B263); "Connecting…" alone looked like a slow start
// for as long as it lasted.
let state_dir = ui.dirs.state_dir();
let failure = gio::spawn_blocking(move || pdfs_core::mountfail::read(&state_dir))
.await
.ok()
.flatten();
match failure {
Some(failure) => status(
&ui,
"dialog-warning-symbolic",
&gettext("Can't open your Drive"),
// Translators: {error} is an error message, in English.
&gettext_f(
"The Proton Drive service keeps failing to start and tries again on its own. The last error was: {error}",
&[("error", &failure.error)],
),
true,
),
None => status(
&ui,
"network-idle-symbolic",
&gettext("Connecting…"),
&gettext("Waiting for the Proton Drive service to start."),
false,
),
}
glib::timeout_add_local_once(CONNECT_RETRY_INTERVAL, move || {
if ui.stack.visible_child_name().as_deref() == Some(page) {
reload(&ui);
+52 -3
View File
@@ -597,6 +597,19 @@ pub(crate) fn this_computer_subtitle(folders: &[SyncFolderInfo]) -> String {
folders.len() as u64,
&[],
);
// A folder whose first pass has not run is not backed up yet, whatever its
// state column says (`docs/BUGS.md` B261).
let waiting = folders.iter().filter(|f| never_synced(f)).count();
let count = match waiting {
0 => count,
// Translators: {folders} is "N folders backed up"; {n} counts the folders whose first sync has not run yet.
n => ngettext_f(
"{folders} · {n} waiting for its first sync",
"{folders} · {n} waiting for their first sync",
n as u64,
&[("folders", &count)],
),
};
let attention = folders
.iter()
.filter(|f| f.state == "error" || f.state == "conflict")
@@ -613,12 +626,21 @@ pub(crate) fn this_computer_subtitle(folders: &[SyncFolderInfo]) -> String {
}
}
/// Human label for a synced folder's `state` column.
pub(crate) fn sync_state_label(state: &str) -> String {
/// Whether a synced folder is a mirror whose first pass has not finished.
/// Its `state` starts out `idle`, which would read as "up to date".
fn never_synced(folder: &SyncFolderInfo) -> bool {
folder.mode == "mirror" && folder.last_sync == 0 && folder.state == "idle"
}
/// Human label for a synced folder's `state` column. `never_synced` is a
/// mirror folder whose first pass has not finished: its `state` starts out
/// `idle`, and "up to date" would say the opposite of the truth (B261).
pub(crate) fn sync_state_label(state: &str, never_synced: bool) -> String {
match state {
"syncing" => gettext("syncing…"),
"error" => gettext("sync error"),
"conflict" => gettext("needs attention"),
_ if never_synced => gettext("waiting for first sync"),
_ => gettext("up to date"),
}
}
@@ -1274,7 +1296,7 @@ mod tests {
remote_uid: "vol~link".into(),
mode: "mirror".into(),
state: state.into(),
last_sync: 0,
last_sync: 1_700_000_000,
pending_mode: None,
progress: None,
paused: false,
@@ -1286,6 +1308,33 @@ mod tests {
assert_eq!(this_computer_subtitle(&[]), "No folders backed up yet");
}
/// A folder added a moment ago is not backed up yet, and the row must not
/// say it is (`docs/BUGS.md` B261). An on-demand folder has no first sync.
#[test]
fn a_folder_before_its_first_sync_is_not_counted_as_backed_up() {
let new = SyncFolderInfo {
last_sync: 0,
..folder("idle")
};
assert_eq!(
this_computer_subtitle(&[folder("idle"), new.clone()]),
"2 folders backed up · 1 waiting for its first sync"
);
let failed = SyncFolderInfo {
state: "error".into(),
..new.clone()
};
assert_eq!(
this_computer_subtitle(&[new.clone(), failed]),
"2 folders backed up · 1 waiting for its first sync · 1 needs attention"
);
let on_demand = SyncFolderInfo {
mode: "ondemand".into(),
..new
};
assert_eq!(this_computer_subtitle(&[on_demand]), "1 folder backed up");
}
#[test]
fn folders_needing_attention_are_counted_not_hidden() {
// The per-folder rows live on the Sync page, so this row is the only
+22 -1
View File
@@ -1729,7 +1729,10 @@ pub(crate) fn location_subtitle(spec: &MountSpec) -> String {
parts.push(match &spec.progress {
Some(p) => sync_progress_label(p),
None if spec.paused => pgettext("state", "paused"),
None => sync_state_label(&spec.state).to_string(),
None => sync_state_label(
&spec.state,
spec.mode == MountMode::Mirror && spec.last_sync == 0,
),
});
}
// Only the locations that are supposed to have a session report its absence:
@@ -1775,7 +1778,25 @@ mod tests {
MountMode::Mirror,
false,
);
assert_eq!(location_subtitle(&spec), "Synced · waiting for first sync");
}
/// A folder whose first pass has not run is not up to date, though its
/// state column starts out `idle` (`docs/BUGS.md` B261).
#[test]
fn a_folder_reads_up_to_date_only_after_its_first_sync() {
let mut spec = spec(
MountKind::Device { sync_folder_id: 7 },
MountMode::Mirror,
false,
);
assert_eq!(location_subtitle(&spec), "Synced · waiting for first sync");
spec.last_sync = 1_700_000_000;
assert_eq!(location_subtitle(&spec), "Synced · up to date");
// A folder that failed before it ever synced says so.
spec.last_sync = 0;
spec.state = "error".into();
assert_eq!(location_subtitle(&spec), "Synced · sync error");
}
#[test]
+18
View File
@@ -194,6 +194,24 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
),
Phase::Attention,
),
// A daemon whose mount fails on every try never opens the
// socket; the app says why (B263).
Ok(s)
if socket
.parent()
.and_then(pdfs_core::mountfail::read)
.is_some() =>
{
(
// Translators: {username} is the Proton account name. The
// service keeps failing to start; the app shows the error.
gettext_f(
"Signed in as {username} — can't open your Drive",
&[("username", &s.username)],
),
Phase::Attention,
)
}
Ok(s) => (
// Translators: {username} is the Proton account name.
gettext_f(