diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index d29c27a..8ea4a50 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -1336,6 +1336,11 @@ fn cmd_status() -> Result<()> { online, pending_uploads, pending_changes, + parked_uploads, + failing_ops, + failing_error, + staged_bytes, + staged_oldest_secs, .. }) => { let state = if online { "" } else { ", offline" }; @@ -1344,6 +1349,25 @@ fn cmd_status() -> Result<()> { None => String::new(), }; println!("Mounted at {mountpoint} ({pinned} pinned{state}{queued})"); + // Everything below is a queue that is not making progress on its + // own. Printed only when it exists, because in the ordinary case + // there is nothing to say. + if staged_bytes > 0 { + println!( + "Staged {} not yet uploaded (oldest {})", + human_bytes(staged_bytes), + human_age(staged_oldest_secs), + ); + } + if parked_uploads > 0 { + println!("Parked {parked_uploads} upload(s) waiting for a final filename"); + } + if failing_ops > 0 { + println!( + "Stuck {failing_ops} queued operation(s) keep failing: {}", + failing_error.as_deref().unwrap_or("no error recorded"), + ); + } } Ok(other) => println!("Mount: unexpected response {other:?}"), Err(_) => println!("Mount: not running."), @@ -1886,6 +1910,18 @@ fn path_arg(path: &Path) -> Result { .ok_or_else(|| anyhow!("path is not valid UTF-8")) } +/// Format an age in seconds for a report line, coarsely — the point of the +/// number is whether it is minutes or weeks. +fn human_age(secs: u64) -> String { + const UNITS: [(&str, u64); 4] = [("d", 86_400), ("h", 3_600), ("m", 60), ("s", 1)]; + for (unit, scale) in UNITS { + if secs >= scale { + return format!("{}{unit}", secs / scale); + } + } + "just now".to_string() +} + /// Format a byte count for a report line. fn human_bytes(bytes: u64) -> String { const UNITS: [(&str, f64); 4] = [ @@ -1925,6 +1961,32 @@ fn cmd_cache(action: CacheCmd) -> Result<()> { } } +/// Ask for a deep integrity check and wait for it to finish. +/// +/// The daemon answers the request immediately and runs the check on a thread — +/// it holds the single database connection for as long as it takes, which would +/// otherwise stall the mount — so a caller that needs the verdict polls for it. +/// `pdfs doctor` is that caller: it is a diagnostic the user is watching, so +/// waiting is right, and the bound is generous enough for a large database +/// without hanging forever if the daemon never finishes. +fn await_integrity_check() -> Result { + const POLL: std::time::Duration = std::time::Duration::from_secs(1); + const GIVE_UP_AFTER: std::time::Duration = std::time::Duration::from_secs(120); + + let started = std::time::Instant::now(); + loop { + let response = control_request(CtlRequest::CacheInspect { deep: true })?; + match &response { + CtlResponse::CacheReport { + integrity_running: true, + integrity_checked: false, + .. + } if started.elapsed() < GIVE_UP_AFTER => std::thread::sleep(POLL), + _ => return Ok(response), + } + } +} + fn cmd_cache_inspect(deep: bool) -> Result<()> { let response = control_request(CtlRequest::CacheInspect { deep })?; if emit_json(&response)? { @@ -1940,6 +2002,7 @@ fn cmd_cache_inspect(deep: bool) -> Result<()> { cache_budget, integrity_problems, integrity_checked, + integrity_running, } => { println!( "Database schema v{schema_version}, {}", @@ -1972,6 +2035,11 @@ fn cmd_cache_inspect(deep: bool) -> Result<()> { println!(" {problem}"); } } + } else if integrity_running { + // The daemon runs the check on a thread rather than holding its + // only database connection for the length of it, so the answer + // belongs to a later request. + println!("\nIntegrity checking in the background — run this again for the result"); } else { println!("\nIntegrity not checked (pass --deep)"); } @@ -2209,14 +2277,24 @@ fn cmd_diagnose() -> Result<()> { format!("{pending} waiting to upload") }, ); - match control_request(CtlRequest::CacheInspect { deep: true }) { + match await_integrity_check() { Ok(CtlResponse::CacheReport { - integrity_problems, .. + integrity_problems, + integrity_checked: true, + .. }) if integrity_problems.is_empty() => report.finding( DiagnoseLevel::Ok, " database integrity", "SQLite integrity_check passed", ), + Ok(CtlResponse::CacheReport { + integrity_checked: false, + .. + }) => report.finding( + DiagnoseLevel::Warn, + " database integrity", + "still checking — run `pdfs cache inspect --deep` for the result", + ), Ok(CtlResponse::CacheReport { integrity_problems, .. }) => report.finding( diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 35d5331..e9e92fa 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -21,7 +21,7 @@ use std::fmt; use std::io::{Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use proton_drive_rs::proton_sdk::ids::NodeUid; @@ -146,6 +146,19 @@ struct PinFile { /// in the gaps. Uploading that as-is would silently corrupt the file. /// `authored` says which ranges are real; `complete` says whether the file can /// be uploaded as it stands. +/// What `staging/` currently holds, from [`ContentCache::staging_usage`]. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct StagingUsage { + /// Bytes across every staged blob. + pub bytes: u64, + /// Staged blobs, not counting their sidecars. + pub files: u64, + /// Age of the oldest staged blob, in seconds. Zero when there are none — + /// what makes a figure here interesting is it being *large*, which is the + /// shape of a write that is never going to drain by itself. + pub oldest_secs: u64, +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct StagedWrite { /// Node the write targeted, in `volume~link` display form. @@ -287,6 +300,14 @@ pub struct ContentCache { /// once [`TOUCH_FLUSH_AT`] have accumulated, and always before an eviction /// pass reads the ordering these describe. touches: Mutex>, + /// Whether `recovery/` may hold a write the daemon has not queued yet. + /// + /// The drain asks for recovered writes on every idle pass — a `read_dir` of + /// a directory that is empty in every run that did not crash. The directory + /// gains entries in exactly two places (the rescue at open, and the release + /// path that parks a still-authoritative scratch file), both of which set + /// this; a scan that finds nothing actionable clears it. + recovery_pending: AtomicBool, } impl ContentCache { @@ -350,6 +371,8 @@ impl ContentCache { thumb_bytes: AtomicU64::new(0), db, touches: Mutex::new(HashMap::new()), + // The rescue above has already run, so start by looking once. + recovery_pending: AtomicBool::new(true), }; cache.reconcile()?; // After reconcile, so the totals describe the index that now describes @@ -1039,7 +1062,14 @@ impl ContentCache { /// daemon to hand to the upload queue. A blob whose sidecar did not survive /// is not actionable automatically, but is retained for manual recovery: /// unexplained bytes are never evidence that user data is disposable. + /// + /// Asked for on every idle drain pass, so the usual answer — a directory + /// that has been empty since the last clean shutdown — costs an atomic load + /// rather than a `read_dir` (see [`recovery_pending`](Self::recovery_pending)). pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> { + if !self.recovery_pending.load(Ordering::Relaxed) { + return Vec::new(); + } let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else { return Vec::new(); }; @@ -1056,6 +1086,39 @@ impl ContentCache { out.push((blob, meta)); } } + // Blobs whose sidecar did not survive stay here for a human and are + // never actionable, so "nothing to queue" is the condition that stops + // the polling — not "the directory is empty". + if out.is_empty() { + self.recovery_pending.store(false, Ordering::Relaxed); + } + out + } + + /// Every blob in `staging/`, paired with its sidecar when it has a readable + /// one. + /// + /// For the startup reconcile: staging holds the only copy of writes the + /// kernel has already been told succeeded, and three failure paths can + /// leave one there with no `pending_op` row pointing at it. Nothing walked + /// this directory, so those bytes were invisible to the daemon and to the + /// user both — `hydrate_pending` walks the op table and + /// [`recovered_writes`](Self::recovered_writes) walks `recovery/`. + pub fn staged_writes(&self) -> Vec<(PathBuf, Option)> { + let Ok(entries) = std::fs::read_dir(&self.staging_dir) else { + return Vec::new(); + }; + let mut out = Vec::new(); + for entry in entries.flatten() { + let blob = entry.path(); + if blob.extension().is_some_and(|e| e == "json") { + continue; + } + let meta = std::fs::read(Self::scratch_sidecar(&blob)) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()); + out.push((blob, meta)); + } out } @@ -1226,6 +1289,10 @@ impl ContentCache { if std::fs::rename(scratch, &recovery).is_ok() { let _ = std::fs::File::open(&self.recovery_dir) .and_then(|dir| dir.sync_all()); + // These bytes are unaccompanied, but re-arm the + // scan anyway: the flag exists to skip a directory + // nothing has written to, and something just did. + self.recovery_pending.store(true, Ordering::Relaxed); return Err(retained( &recovery, Error::Other(format!( @@ -1504,6 +1571,44 @@ impl ContentCache { .sum() } + /// Bytes and file count held in `staging/`, with the age of the oldest + /// entry in seconds. + /// + /// Staging is not part of [`usage`](Self::usage) and is not subject to the + /// budget — it holds the *only* copy of writes that have not been uploaded, + /// so nothing here may ever be evicted to reclaim space. That is precisely + /// why it has to be reportable: a parked transient create (three abandoned + /// 4 GiB downloads, say) otherwise occupies disk with nothing in any UI + /// accounting for it. + /// + /// A `read_dir` per call, which is fine for the callers it has: a status + /// request, polled every couple of seconds against a directory that is + /// empty whenever the queue is drained. + pub fn staging_usage(&self) -> StagingUsage { + let Ok(entries) = std::fs::read_dir(&self.staging_dir) else { + return StagingUsage::default(); + }; + let mut usage = StagingUsage::default(); + let now = std::time::SystemTime::now(); + for entry in entries.flatten() { + // Sidecars describe the blob beside them; counting both would double + // the file count and report metadata as user bytes. + if entry.path().extension().is_some_and(|e| e == "json") { + continue; + } + let Ok(meta) = entry.metadata() else { continue }; + usage.files += 1; + usage.bytes += meta.len(); + if let Ok(age) = meta.modified().and_then(|m| { + now.duration_since(m) + .map_err(|_| std::io::Error::other("modified in the future")) + }) { + usage.oldest_secs = usage.oldest_secs.max(age.as_secs()); + } + } + usage + } + /// Configured soft byte cap (`0` = unlimited), for display alongside /// [`usage`](Self::usage). This is the whole-cache figure; internally it is /// divided between the pools (see [`SPLIT_BLOB`]), but the user set one diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index fdb2e0e..f751745 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -1193,6 +1193,28 @@ pub enum Response { /// calling a queued `mkdir` an upload is a lie. #[serde(default)] pending_changes: u64, + /// Queued writes held back because the file still wears a transient + /// name (a browser's `*.crdownload`, an editor's `*.swp`). They are not + /// waiting on the network and will not drain until the finalising + /// rename, so they are counted apart from `pending_uploads`. + #[serde(default)] + parked_uploads: u64, + /// Queued ops that have failed enough times to be considered stuck. + /// They keep retrying forever — correctly, since their staged bytes are + /// the only copy — which is why they have to be visible. + #[serde(default)] + failing_ops: u64, + /// The most recent error from a stuck op, when there is one. + #[serde(default)] + failing_error: Option, + /// Bytes held in `staging/`: writes accepted from the kernel whose only + /// copy is on local disk. Never evictable, never counted against the + /// cache budget — hence reported separately. + #[serde(default)] + staged_bytes: u64, + /// Age of the oldest staged write, in seconds. + #[serde(default)] + staged_oldest_secs: u64, }, /// A human-readable success message. Ok { message: String }, @@ -1217,6 +1239,12 @@ pub enum Response { /// Whether the integrity check actually ran, so an empty problem list /// is not mistaken for a clean bill of health it never gave. integrity_checked: bool, + /// Whether a deep check is running right now. It reads every page of + /// the database under the daemon's only connection, so it is answered + /// asynchronously: this says "ask again shortly", which is a different + /// thing to say than either of the two above. + #[serde(default)] + integrity_running: bool, }, /// The pin registry. Pins { pins: Vec }, diff --git a/crates/pdfs-core/src/db/local.rs b/crates/pdfs-core/src/db/local.rs index fbbb6a0..2e2419e 100644 --- a/crates/pdfs-core/src/db/local.rs +++ b/crates/pdfs-core/src/db/local.rs @@ -41,32 +41,43 @@ impl Db { Ok(next) } - /// Write one batch of walked entries under scan generation `generation`. The - /// FTS index is *not* touched here — it is rebuilt once in - /// [`local_finish_scan`](Self::local_finish_scan). + /// Write one batch of walked entries under scan generation `generation`, + /// keeping the FTS index in step with them. + /// + /// The index is maintained incrementally rather than rebuilt at the end of + /// the scan, because a rebuild reads and re-tokenises every row in the table + /// — trigrams over a few hundred thousand paths — inside one transaction on + /// the single shared connection, which stalls the whole mount for as long as + /// it takes. Incremental maintenance is possible here for a reason specific + /// to this table: `local_fts` covers `name` and `path`, `path` is the + /// table's key, and `name` is derived from it, so an existing row's indexed + /// content can never change. Only insertions and deletions touch the index; + /// a re-scan that finds the same files does no index work at all. + /// + /// Hence the update-then-insert shape below rather than an upsert: the + /// distinction between "already indexed" and "new row" is exactly what + /// decides whether the index needs a write. pub fn local_upsert_batch(&self, generation: i64, entries: &[LocalEntry]) -> Result<()> { let mut conn = self.conn.lock(); let tx = conn.transaction()?; { - let mut stmt = tx.prepare_cached( - "INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen) - VALUES (?1, ?2, ?3, ?4, ?5, ?6) - ON CONFLICT(path) DO UPDATE SET - name = excluded.name, - is_dir = excluded.is_dir, - size = excluded.size, - mtime = excluded.mtime, - scan_gen = excluded.scan_gen", + let mut update = tx.prepare_cached( + "UPDATE local_files + SET name = ?2, is_dir = ?3, size = ?4, mtime = ?5, scan_gen = ?6 + WHERE path = ?1", )?; + let mut insert = tx.prepare_cached( + "INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + )?; + let mut index = + tx.prepare_cached("INSERT INTO local_fts (rowid, name, path) VALUES (?1, ?2, ?3)")?; for e in entries { - stmt.execute(params![ - e.path, - e.name, - e.is_dir as i64, - e.size, - e.mtime, - generation - ])?; + let row = params![e.path, e.name, e.is_dir as i64, e.size, e.mtime, generation]; + if update.execute(row)? == 0 { + insert.execute(row)?; + index.execute(params![tx.last_insert_rowid(), e.name, e.path])?; + } } } tx.commit()?; @@ -74,16 +85,25 @@ impl Db { } /// Close scan generation `generation`: drop every row an older scan wrote - /// (those paths are gone from disk), rebuild the FTS index over what remains, - /// and stamp the completion time. Returns the number of indexed entries. + /// (those paths are gone from disk), drop their FTS entries, and stamp the + /// completion time. Returns the number of indexed entries. + /// + /// The FTS deletes name the old values explicitly, which is how an + /// external-content FTS5 table is told to retract a row — see + /// [`local_upsert_batch`](Self::local_upsert_batch) for why this is done + /// per-row instead of by rebuilding the index. pub fn local_finish_scan(&self, generation: i64, finished_at: i64) -> Result { let mut conn = self.conn.lock(); let tx = conn.transaction()?; + tx.execute( + "INSERT INTO local_fts (local_fts, rowid, name, path) + SELECT 'delete', id, name, path FROM local_files WHERE scan_gen != ?1", + params![generation], + )?; tx.execute( "DELETE FROM local_files WHERE scan_gen != ?1", params![generation], )?; - tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?; tx.execute( "INSERT INTO sync_state (key, value) VALUES ('local_indexed_at', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 5ec5f3e..116b601 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/crates/pdfs-core/src/db/migrations.rs @@ -9,7 +9,7 @@ use super::Db; use crate::Result; /// Current schema version. Bump on every forward migration added below. -pub(super) const SCHEMA_VERSION: i64 = 22; +pub(super) const SCHEMA_VERSION: i64 = 24; impl Db { pub(super) fn migrate(&self) -> Result<()> { @@ -131,6 +131,24 @@ impl Db { // order by `id`: it has only ever cost writes. tx.execute_batch("DROP INDEX IF EXISTS activity_time;")?; } + if current < 23 { + // One last full rebuild of the local index, because from here on the + // scan maintains it incrementally (`local_upsert_batch`) and so can + // no longer repair a `local_fts` that drifted from `local_files`. + // Whatever it costs, it costs once, at startup, before the mount + // serves anything. + let present: bool = tx.query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'local_fts'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + if present { + tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?; + } + } + if current < 24 { + tx.execute_batch(MIGRATION_V24)?; + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -633,3 +651,22 @@ const MIGRATION_V22: [(&str, &str); 3] = [ "CREATE INDEX IF NOT EXISTS idx_nodes_parent_live ON nodes(parent_uid) WHERE trashed = 0;", ), ]; + +/// V24: the revisions this daemon sealed itself, so restarting does not reopen +/// the conflict hole B70 layer B closed. +/// +/// `Core::own_sealed_revs` held these in memory: a queued write whose baseline +/// names an earlier revision is a conflict *unless* the remote sits at a +/// revision we sealed, in which case it is one writer stalling and resuming and +/// must supersede rather than fork a `(sync-conflict)` copy. Losing the map at +/// shutdown meant the first drain after a restart could not tell the two apart. +/// +/// `sealed_at` is what bounds the table: entries older than the window in which +/// a stall→resume is plausible are pruned on write. +const MIGRATION_V24: &str = " +CREATE TABLE IF NOT EXISTS own_sealed_rev ( + uid TEXT PRIMARY KEY, + revision_id TEXT NOT NULL, + sealed_at INTEGER NOT NULL +); +"; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index b16f743..e5a4099 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -18,8 +18,10 @@ use parking_lot::Mutex; use rusqlite::Connection; -use crate::Result; -use std::path::Path; +use crate::{Error, Result}; +use std::fs::File; +use std::os::fd::AsRawFd; +use std::path::{Path, PathBuf}; mod activity; mod albums; @@ -86,12 +88,50 @@ const WAL_AUTOCHECKPOINT_PAGES: i64 = 500; /// the inner `Mutex`. pub struct Db { conn: Mutex, + /// The single-writer lock, held open for as long as this handle lives. + /// + /// Never read; dropping the `File` is what releases the `flock`, so it has + /// to be owned here rather than by `open`'s stack frame. `None` for + /// in-memory databases, which nothing else can reach. + _single_writer: Option, } impl Db { /// Open (creating if absent) the database at `path`, enable WAL, and bring /// the schema up to [`SCHEMA_VERSION`]. + /// + /// Takes the single-instance lock first (see + /// [`acquire_single_writer_lock`]), and rebuilds from empty if what is on + /// disk turns out not to be a readable database (see [`is_corrupt`]). pub fn open(path: &Path) -> Result { + let single_writer = acquire_single_writer_lock(path)?; + let conn = match Self::open_configured(path) { + Ok(conn) => conn, + Err(e) if is_corrupt(&e) => { + // Every byte in here is derived from the account or from the + // content cache directory, so the recovery is to start over — + // loudly, and keeping the damaged file for a post-mortem rather + // than deleting evidence. + tracing::error!( + db = %path.display(), + error = %e, + "cache database is corrupt — moving it aside and rebuilding \ + (metadata will be re-fetched; no user data is stored here)" + ); + quarantine_corrupt_db(path)?; + Self::open_configured(path)? + } + Err(e) => return Err(e), + }; + Ok(Self { + conn: Mutex::new(conn), + _single_writer: single_writer, + }) + } + + /// Open the file, apply every PRAGMA, and migrate. Split out of + /// [`open`](Self::open) so the corruption path can retry it verbatim. + fn open_configured(path: &Path) -> Result { let conn = Connection::open(path)?; // WAL: readers never block the single writer. NORMAL sync is the // standard durability/throughput tradeoff for WAL. @@ -137,9 +177,17 @@ impl Db { let db = Self { conn: Mutex::new(conn), + _single_writer: None, }; db.migrate()?; - Ok(db) + Ok(db.into_conn()) + } + + /// Unwrap the connection back out of a temporary handle. Only + /// [`open_configured`](Self::open_configured) uses it, to run `migrate` + /// (which is written against `&self`) before the real handle exists. + fn into_conn(self) -> Connection { + self.conn.into_inner() } /// Open an in-memory database. For tests. @@ -152,6 +200,7 @@ impl Db { conn.pragma_update(None, "foreign_keys", "ON")?; let db = Self { conn: Mutex::new(conn), + _single_writer: None, }; db.migrate()?; Ok(db) @@ -165,5 +214,102 @@ impl Db { } } +/// Whether `path` names the in-memory database rather than a file. Both +/// spellings SQLite accepts are checked, because neither has a lock file or a +/// corruption story. +fn is_in_memory(path: &Path) -> bool { + let p = path.as_os_str(); + p == ":memory:" || p.is_empty() +} + +/// Take the single-instance write lock for the database at `path`. +/// +/// The daemon is the only writer by design — the front-ends go through the +/// control socket — but nothing stopped a hand-run `pdfs mount` from starting +/// alongside the systemd unit, at which point two processes own the same inode +/// space, the same content cache, and the same drain queue. SQLite's own +/// locking makes each *statement* safe and does nothing about that. +/// +/// `flock` on a sibling `.lock` file rather than on the database itself: +/// SQLite uses POSIX record locks, so the two never interact, and a lock file +/// left behind carries no state — the lock lives in the kernel and dies with +/// the process, however it exits. +fn acquire_single_writer_lock(path: &Path) -> Result> { + if is_in_memory(path) { + return Ok(None); + } + let lock_path = lock_path_for(path); + let file = std::fs::OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&lock_path)?; + // SAFETY: `file` owns the descriptor and outlives the call. + let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; + if rc != 0 { + let e = std::io::Error::last_os_error(); + if e.raw_os_error() == Some(libc::EWOULDBLOCK) { + return Err(Error::Other(format!( + "another Proton Drive daemon is already using {} — \ + stop it first (`systemctl --user stop proton-drive.service`)", + path.display() + ))); + } + return Err(Error::Io(e)); + } + Ok(Some(file)) +} + +fn lock_path_for(path: &Path) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(".lock"); + PathBuf::from(name) +} + +/// Whether `e` says the file is not a usable database, as opposed to any other +/// failure. These are the only two conditions rebuilding fixes; a disk error, a +/// permission problem or a newer-than-supported schema must all still fail. +fn is_corrupt(e: &Error) -> bool { + matches!( + e, + Error::Db(rusqlite::Error::SqliteFailure(inner, _)) + if inner.code == rusqlite::ErrorCode::DatabaseCorrupt + || inner.code == rusqlite::ErrorCode::NotADatabase + ) +} + +/// Move a damaged database (and its WAL sidecars) out of the way so the next +/// open starts from empty. +/// +/// Renamed rather than deleted: this is a cache, but it is also the only +/// evidence of what went wrong, and it costs one file the user can remove. +fn quarantine_corrupt_db(path: &Path) -> Result<()> { + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + for suffix in ["", "-wal", "-shm"] { + let mut from = path.as_os_str().to_owned(); + from.push(suffix); + let from = PathBuf::from(from); + if !from.exists() { + continue; + } + let mut to = path.as_os_str().to_owned(); + to.push(format!(".corrupt-{stamp}{suffix}")); + let to = PathBuf::from(to); + // A failure to rename the sidecars is survivable — SQLite discards a + // WAL whose database header no longer matches — but failing to move the + // database itself is not, because the retry would hit the same file. + match std::fs::rename(&from, &to) { + Ok(()) => tracing::warn!(from = %from.display(), to = %to.display(), "quarantined"), + Err(e) if suffix.is_empty() => return Err(Error::Io(e)), + Err(e) => tracing::warn!(file = %from.display(), error = %e, "could not quarantine"), + } + } + Ok(()) +} + #[cfg(test)] mod tests; diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index e02139a..7b23efc 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -119,15 +119,32 @@ pub struct PendingOp { } /// How much work the queue owes the server, by whether it carries bytes. -#[derive(Debug, Clone, Copy, Default)] +#[derive(Debug, Clone, Default)] pub struct PendingCounts { /// Queued `create`/`revision` ops: files whose content is not on the remote. pub uploads: i64, /// Queued `mkdir`/`rename`/`trash` ops: metadata the remote has not been /// told about. pub changes: i64, + /// Ops parked at [`PARK_UNTIL`]: a transient file's bytes, held back until + /// the rename that gives it its finished name. They are queued but will + /// never drain on their own, so counting them as "waiting to upload" would + /// have the front-end promise progress that is not coming. + pub parked: i64, + /// Ops that have failed at least [`FAILING_ATTEMPTS`] times. A permanently + /// failing op never wedges the queue — the backoff sees to that — which is + /// exactly why it needs surfacing: it retries forever, invisibly. + pub failing: i64, + /// The most recent error text from a failing op, for the front-end to show + /// instead of the bare count. + pub last_error: Option, } +/// Failed attempts after which an op is reported as failing rather than merely +/// retrying. Six is where the backoff has reached its ceiling, so an op past it +/// is retrying at the slowest rate it ever will and has been for minutes. +pub const FAILING_ATTEMPTS: i64 = 6; + /// The outcome of folding freshly written bytes into a queued create. #[derive(Debug, Clone)] pub struct AttachedBlob { @@ -564,7 +581,33 @@ impl Db { params![OP_REVISION, OP_CREATE], |r| r.get(0), )?; - Ok(PendingCounts { uploads, changes }) + let parked = conn.query_row( + "SELECT COUNT(*) FROM pending_op WHERE next_attempt_at >= ?1", + params![PARK_UNTIL], + |r| r.get(0), + )?; + let failing = conn.query_row( + "SELECT COUNT(*) FROM pending_op WHERE attempts >= ?1", + params![FAILING_ATTEMPTS], + |r| r.get(0), + )?; + let last_error = conn + .query_row( + "SELECT last_error FROM pending_op + WHERE attempts >= ?1 AND last_error IS NOT NULL + ORDER BY attempts DESC LIMIT 1", + params![FAILING_ATTEMPTS], + |r| r.get::<_, Option>(0), + ) + .optional()? + .flatten(); + Ok(PendingCounts { + uploads, + changes, + parked, + failing, + last_error, + }) } /// Nothing bearing this volume may be handed to the API — it would 404. The @@ -609,4 +652,78 @@ impl Db { )?; Ok(()) } + + /// Every staged blob the queue still refers to. + /// + /// The startup staging reconcile subtracts this from what is actually in + /// `staging/`: the difference is bytes the kernel was told were written and + /// that nothing is going to upload. + pub fn op_blob_paths(&self) -> Result> { + let conn = self.conn.lock(); + let mut stmt = + conn.prepare("SELECT blob_path FROM pending_op WHERE blob_path IS NOT NULL")?; + let rows = stmt.query_map([], |r| r.get::<_, String>(0))?; + let mut out = std::collections::HashSet::new(); + for row in rows { + out.insert(row?); + } + Ok(out) + } + + /// Record that this daemon sealed `revision_id` on `uid`, and prune entries + /// past [`OWN_SEALED_TTL_MS`]. + /// + /// One row per node, overwritten as the node advances. See + /// [`own_sealed_rev`](Self::own_sealed_rev) for what reads it. + pub fn set_own_sealed_rev(&self, uid: &str, revision_id: &str, now_ms: i64) -> Result<()> { + let conn = self.conn.lock(); + conn.execute( + "INSERT INTO own_sealed_rev (uid, revision_id, sealed_at) + VALUES (?1, ?2, ?3) + ON CONFLICT(uid) DO UPDATE SET + revision_id = excluded.revision_id, + sealed_at = excluded.sealed_at", + params![uid, revision_id, now_ms], + )?; + conn.execute( + "DELETE FROM own_sealed_rev WHERE sealed_at < ?1", + params![now_ms - OWN_SEALED_TTL_MS], + )?; + Ok(()) + } + + /// The last revision of `uid` this daemon sealed itself, if it is still + /// within [`OWN_SEALED_TTL_MS`]. + /// + /// The drain asks before treating a moved-on remote as a conflict: a + /// revision we sealed means one writer stalled and resumed, which + /// supersedes rather than forking a `(sync-conflict)` copy (docs/BUGS.md + /// B70 layer B). + pub fn own_sealed_rev(&self, uid: &str, now_ms: i64) -> Result> { + let conn = self.conn.lock(); + Ok(conn + .query_row( + "SELECT revision_id FROM own_sealed_rev + WHERE uid = ?1 AND sealed_at >= ?2", + params![uid, now_ms - OWN_SEALED_TTL_MS], + |r| r.get(0), + ) + .optional()?) + } + + /// Forget what we sealed on `uid` — it has been trashed, so there is no + /// revision left to chain onto. + pub fn clear_own_sealed_rev(&self, uid: &str) -> Result<()> { + let conn = self.conn.lock(); + conn.execute("DELETE FROM own_sealed_rev WHERE uid = ?1", params![uid])?; + Ok(()) + } } + +/// How long a revision this daemon sealed stays recognisable as its own. +/// +/// Bounds a table that would otherwise grow with every write for the life of +/// the install. A week is far longer than any stall→resume — a browser download +/// that pauses and continues, an editor that reopens its handle — and short +/// enough that the table tracks recent work rather than accumulating history. +pub const OWN_SEALED_TTL_MS: i64 = 7 * 24 * 60 * 60 * 1000; diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 23c85e5..38ac293 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -867,6 +867,67 @@ fn shared_root_pinned_name_survives_database_reopen() { let _ = std::fs::remove_file(path); } +/// Two writers on one database is the failure mode the single-instance lock +/// exists to prevent — a hand-run `pdfs mount` next to the systemd unit gives +/// two daemons the same inode space, content cache, and drain queue. The second +/// open has to fail, and it has to stop failing once the first handle is gone. +#[test] +fn a_second_writer_is_refused_while_the_first_holds_the_database() { + let path = std::env::temp_dir().join(format!( + "pdfs-single-writer-{}-{}.db", + std::process::id(), + now_test_id() + )); + let first = Db::open(&path).unwrap(); + let second = Db::open(&path); + assert!(second.is_err(), "a second writer was allowed in"); + + drop(first); + // The lock lives in the kernel and dies with the descriptor, so nothing has + // to be cleaned up for the next daemon to start. + Db::open(&path).expect("reopen after the first handle is dropped"); + + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(path.with_extension("db.lock")); +} + +/// `cache.db` is derived state — every row can be re-fetched or re-scanned — so +/// a file that is not a database at all must not stop the daemon starting. It +/// is moved aside (not deleted: it is the only evidence) and rebuilt. +#[test] +fn a_corrupt_database_is_moved_aside_and_rebuilt() { + let path = std::env::temp_dir().join(format!( + "pdfs-corrupt-{}-{}.db", + std::process::id(), + now_test_id() + )); + std::fs::write(&path, b"this is not an SQLite database").unwrap(); + + let db = Db::open(&path).unwrap(); + db.set_state_str("shared_with_me_name", "rebuilt").unwrap(); + assert_eq!( + db.state_str("shared_with_me_name").unwrap().as_deref(), + Some("rebuilt") + ); + drop(db); + + let dir = path.parent().unwrap(); + let stem = path.file_name().unwrap().to_string_lossy().to_string(); + let quarantined: Vec<_> = std::fs::read_dir(dir) + .unwrap() + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().to_string()) + .filter(|name| name.starts_with(&format!("{stem}.corrupt-"))) + .collect(); + assert_eq!(quarantined.len(), 1, "the damaged file was not kept"); + + for name in quarantined { + let _ = std::fs::remove_file(dir.join(name)); + } + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(path.with_extension("db.lock")); +} + fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry { LocalEntry { path: path.into(), @@ -911,6 +972,25 @@ fn local_scan_indexes_then_prunes_stale_paths() { assert_eq!(db.search_local("notes", 10).unwrap().len(), 1); } +/// The index is maintained incrementally rather than rebuilt at the end of a +/// scan, so the case that used to be covered for free — a file the scan sees +/// again — needs its own test: a re-scan must neither lose the entry nor +/// duplicate it. +#[test] +fn rescanning_an_unchanged_file_keeps_exactly_one_index_entry() { + let db = Db::open_in_memory().unwrap(); + for (generation, mtime) in [(1_000, 5), (2_000, 9)] { + let g = db.local_begin_scan().unwrap(); + let mut e = local("/home/u/docs/report.pdf", "report.pdf", false); + e.mtime = mtime; + db.local_upsert_batch(g, &[e]).unwrap(); + assert_eq!(db.local_finish_scan(g, generation).unwrap(), 1); + let hits = db.search_local("report", 10).unwrap(); + assert_eq!(hits.len(), 1, "generation {generation}"); + assert_eq!(hits[0].mtime, mtime); + } +} + /// Queries below the trigram minimum still match, via the `LIKE` fallback. #[test] fn local_search_short_query_like_fallback() { diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index ec5d12e..bda2618 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -148,7 +148,13 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U Ok(CtlRequest::Status) => { let pins = core.cache.list_pins(); let queued = core.db.pending_op_counts().unwrap_or_default(); + let staging = core.cache.staging_usage(); CtlResponse::Status { + parked_uploads: queued.parked.max(0) as u64, + failing_ops: queued.failing.max(0) as u64, + failing_error: queued.last_error.clone(), + staged_bytes: staging.bytes, + staged_oldest_secs: staging.oldest_secs, username: username.to_string(), mountpoint: mountpoint.display().to_string(), pinned: pins.len(), @@ -562,19 +568,20 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U } Ok(CtlRequest::CacheInspect { deep }) => match core.db.stats() { Ok(stats) => { - // The deep check reads every page, so it runs only on request; - // `integrity_checked` tells the caller which answer it is - // holding, since "no problems found" and "never looked" are - // both an empty list. - let (integrity_problems, integrity_checked) = if deep { - match core.db.integrity_check() { - Ok(problems) => (problems, true), - Err(e) => (vec![format!("integrity check failed: {e:?}")], true), - } + // The deep check reads every page — tens of seconds on a real + // database, all of it holding the connection every FUSE handler + // needs — so it runs on a thread and this request returns + // whatever the last completed check found. `integrity_checked` + // tells the caller which answer it is holding, since "no + // problems found" and "never looked" are both an empty list; + // `integrity_running` says a fresh answer is on its way. + let (integrity_problems, integrity_checked, integrity_running) = if deep { + core.start_or_read_integrity_check() } else { - (Vec::new(), false) + (Vec::new(), false, false) }; CtlResponse::CacheReport { + integrity_running, schema_version: stats.schema_version, db_bytes: stats.total_bytes(), db_reclaimable_bytes: stats.reclaimable_bytes(), @@ -587,17 +594,16 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U } Err(e) => CtlResponse::error(CoreError::internal(format!("read db stats: {e:?}"))), }, - Ok(CtlRequest::CacheVacuum) => match core.db.vacuum() { - Ok(outcome) => CtlResponse::Ok { - message: format!( - "vacuumed: {:.1} MiB freed ({:.1} MiB → {:.1} MiB), {} WAL frame(s) checkpointed", - outcome.freed_bytes() as f64 / 1_048_576.0, - outcome.before_bytes as f64 / 1_048_576.0, - outcome.after_bytes as f64 / 1_048_576.0, - outcome.wal_frames_checkpointed, - ), + // Ack and vacuum on a thread. `VACUUM` rewrites the whole database + // under the connection mutex — minutes on a large one — and the + // request's own socket timeout is 10 s, so running it inline both froze + // the mount and lost its own answer. Progress shows up as a job in + // `GetQueueStatus`; the outcome goes to the log. + Ok(CtlRequest::CacheVacuum) => match core.start_vacuum() { + Ok(()) => CtlResponse::Ok { + message: "compacting the cache database in the background".into(), }, - Err(e) => CtlResponse::error(CoreError::internal(format!("vacuum: {e:?}"))), + Err(e) => CtlResponse::error(e), }, Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers { items: core.transfers.snapshot(), diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 3a9d3e1..5217726 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -355,6 +355,9 @@ impl Core { } self.db.complete_trash_op(op.id, &uid)?; self.own_sealed_revs.lock().remove(&uid); + if let Err(e) = self.db.clear_own_sealed_rev(&uid.to_string()) { + debug!(%uid, error = %e, "clearing the sealed-revision record failed"); + } self.invalidate_trash(); self.log_activity(ActivityKind::Trash, &name, "trashed", true); info!(%uid, name, "pending trash landed"); @@ -629,7 +632,16 @@ impl Core { // instead of forking — so the finished bytes win the name rather than // being exiled to a `(sync-conflict)` copy. let remote_rev = node_revision_id(&node); - let own_rev = self.own_sealed_revs.lock().get(uid).cloned(); + // In memory first, then the table it is written through to: the map is + // this process's cache of it, and after a restart only the table has + // the answer (which is the whole reason it is persisted). + let own_rev = match self.own_sealed_revs.lock().get(uid).cloned() { + Some(rev) => Some(rev), + None => self + .db + .own_sealed_rev(&uid.to_string(), now_millis()) + .unwrap_or_default(), + }; if is_own_self_supersede(meta.complete, remote_rev.as_deref(), own_rev.as_deref()) { debug!(%uid, ?remote_rev, "queued write chains onto our own sealed revision; not a conflict"); @@ -985,7 +997,16 @@ impl Core { // change and does not fork (B70 layer B; consulted in // `revision_conflict`). if let Some(rev) = sealed_rev.clone() { - self.own_sealed_revs.lock().insert(uid.clone(), rev); + self.own_sealed_revs.lock().insert(uid.clone(), rev.clone()); + // Write through: losing this at shutdown reopened the fork + // window B70 layer B closed, for the first drain after a + // restart. + if let Err(e) = self + .db + .set_own_sealed_rev(&uid.to_string(), &rev, now_millis()) + { + warn!(%uid, error = %e, "persisting our sealed revision failed"); + } } self.for_each_state(|st| { for aw in st.active_writes.values_mut() { diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 430a3b8..69f1204 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -425,6 +425,14 @@ struct Core { /// storage moves slowly enough that a minute-old number is a better answer /// than either a stall or the zeroes the default implementation returns. quota: Arc>>, + /// Database maintenance that has been moved off the request thread. + /// + /// `VACUUM` and a deep `integrity_check` both hold the single connection for + /// tens of seconds against a real install, which stops every FUSE handler + /// that needs the database — so the handler acknowledges the request and a + /// thread does the work, exactly as the other long control requests already + /// do. See [`Maintenance`]. + maintenance: Arc>, /// Folders whose listing was enumerated cheaply and is having its file sizes /// filled in right now, so a burst of `stat`s over a fresh listing starts one /// upgrade rather than one per entry. See [`Core::spawn_size_upgrade`]. @@ -510,6 +518,21 @@ struct Core { states: Arc, } +/// Background database maintenance: what is running, and what the last deep +/// integrity check found. +/// +/// One at a time — both operations are whole-database and there is nothing to +/// gain from overlapping them — and the result of a check outlives the request +/// that asked for it, because the request is answered before the check finishes. +#[derive(Default)] +struct Maintenance { + /// A vacuum or integrity check is running right now. + running: bool, + /// Findings of the most recent completed deep check. `None` means none has + /// finished this run, which is *not* the same as "found nothing". + integrity: Option>, +} + /// How long a change this daemon made stays recognisable as its own. Generous /// against a feed that can lag, but finite: past it, an event for the node is /// applied like any other. See [`Core::self_changes`]. @@ -695,6 +718,82 @@ enum SwitchBlocked { } impl Core { + /// Start a background `VACUUM`, or refuse because maintenance is already + /// running. + /// + /// Returns as soon as the thread is spawned; the caller acknowledges the + /// request. The outcome is a log line and a job that disappears from + /// `GetQueueStatus` when it is done — a vacuum has nothing to report that a + /// follow-up `CacheInspect` cannot show. + fn start_vacuum(&self) -> CoreResult<()> { + { + let mut m = self.maintenance.lock(); + if m.running { + return Err(CoreError::conflict( + "database maintenance is already running", + )); + } + m.running = true; + } + let core = self.clone(); + std::thread::spawn(move || { + let job = core.transfers.begin_job("Compacting the cache database"); + job.detail("Rewriting the database file"); + let started = Instant::now(); + match core.db.vacuum() { + Ok(outcome) => info!( + freed_bytes = outcome.freed_bytes(), + before_bytes = outcome.before_bytes, + after_bytes = outcome.after_bytes, + wal_frames = outcome.wal_frames_checkpointed, + elapsed_ms = started.elapsed().as_millis() as u64, + "cache database vacuumed" + ), + Err(e) => warn!(error = %e, "vacuum failed"), + } + core.maintenance.lock().running = false; + }); + Ok(()) + } + + /// Answer a deep `CacheInspect`: hand back the last completed integrity + /// check, and start one if none is running. + /// + /// Returns `(problems, checked, running)`. A first request gets + /// `(_, false, true)` and the answer lands on a later one — the check reads + /// every page of the database under the connection mutex, so the choice is + /// between answering late and freezing the mount until it finishes. + fn start_or_read_integrity_check(&self) -> (Vec, bool, bool) { + let mut m = self.maintenance.lock(); + if let Some(problems) = m.integrity.clone() { + return (problems, true, m.running); + } + if m.running { + return (Vec::new(), false, true); + } + m.running = true; + drop(m); + + let core = self.clone(); + std::thread::spawn(move || { + let job = core.transfers.begin_job("Checking the cache database"); + job.detail("Verifying every page"); + let problems = match core.db.integrity_check() { + Ok(problems) => problems, + Err(e) => vec![format!("integrity check failed: {e:?}")], + }; + if problems.is_empty() { + info!("cache database integrity check found no problems"); + } else { + warn!(problems = problems.len(), "cache database integrity check"); + } + let mut m = core.maintenance.lock(); + m.integrity = Some(problems); + m.running = false; + }); + (Vec::new(), false, true) + } + /// Register an inode space so background work can reach it. Called once for /// the primary mount and once per on-demand fork. fn register_state(&self, mountpoint: &Path) { @@ -1218,6 +1317,100 @@ impl Core { } } + /// Re-attach staged blobs that no queued op refers to. + /// + /// `staging/` holds the only copy of writes the kernel has already been + /// told succeeded. Three release-path failures put bytes there and return + /// `EIO` without leaving a `pending_op` row behind — an access change + /// mid-release, a write overlapping an undrained incomplete edit, a create + /// that drained out from under its own handle. Until now nothing ever + /// walked the directory again: `hydrate_pending` reads the op table and + /// [`recover_fsynced_writes`](Self::recover_fsynced_writes) reads + /// `recovery/`, so an orphaned staged blob was invisible to the daemon and + /// to the user alike, and stayed that way until someone found the file. + /// + /// Runs once at mount, after `hydrate_pending`, so `pending` already + /// describes what the queue owns. An orphan whose sidecar names a real node + /// is queued as an ordinary revision — the condition that made it fail is + /// gone by definition, since the op it collided with has since drained. + /// Anything that cannot be addressed (no sidecar, an unparseable uid, a + /// `local~` placeholder whose create is gone) is left in place and named in + /// the log: unexplained bytes are never evidence that user data is + /// disposable. + fn reconcile_staging(&self) { + let staged = self.cache.staged_writes(); + if staged.is_empty() { + return; + } + let claimed = match self.db.op_blob_paths() { + Ok(paths) => paths, + Err(e) => { + error!(error = %e, "cannot read queued blob paths; skipping staging reconcile"); + return; + } + }; + let mut queued = 0usize; + let mut stranded = 0usize; + for (blob, meta) in staged { + if claimed.contains(&blob.to_string_lossy().into_owned()) { + continue; + } + let addressable = meta.as_ref().and_then(|m| parse_node_uid(&m.uid)); + let Some((meta, uid)) = meta.zip(addressable).filter(|(_, uid)| !is_local_uid(uid)) + else { + stranded += 1; + warn!(blob = %blob.display(), + "staged bytes belong to no queued upload and name no node; kept for recovery"); + continue; + }; + if self.pending.lock().contains_key(&uid) { + // A newer write for the same node is queued and carries its own + // blob. These bytes are older; keeping them is right, uploading + // them over the newer ones is not. + stranded += 1; + warn!(%uid, blob = %blob.display(), + "staged bytes are superseded by a queued write; kept for recovery"); + continue; + } + let op = PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid.to_string(), + parent_uid: None, + name: None, + blob_path: Some(blob.to_string_lossy().into_owned()), + meta_json: Some(serde_json::to_string(&meta).unwrap_or_default()), + created_at: now_millis(), + attempts: 0, + last_error: None, + next_attempt_at: now_millis(), + }; + match self.db.enqueue_op(&op) { + Ok((_, superseded)) => { + if let Some(old) = superseded { + self.cache.discard_staged(Path::new(&old)); + } + self.pending.lock().insert( + uid.clone(), + PendingRevision { + path: blob.clone(), + meta, + }, + ); + queued += 1; + } + Err(e) => { + stranded += 1; + error!(%uid, blob = %blob.display(), error = %e, + "cannot queue an orphaned staged write; bytes kept"); + } + } + } + if queued > 0 || stranded > 0 { + info!(queued, stranded, "reconciled orphaned staged writes"); + } + } + /// Queue the writes that an unclean shutdown caught between `fsync(2)` and /// `close(2)`. /// diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index ea35d90..64a2eed 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -357,6 +357,7 @@ pub fn mount( let core = Core { client: client.clone(), rt: rt.clone(), + maintenance: Arc::new(Mutex::new(Default::default())), primary_root_uid: root.uid.clone(), primary: true, state: Arc::new(Mutex::new(State { @@ -416,6 +417,9 @@ pub fn mount( // aside at open. After `hydrate_pending`, so a recovered partial write can // see an already-queued write to the same node. core.recover_fsynced_writes(); + // And finally the staged blobs neither of those accounts for: bytes a + // failed release left behind with nothing pointing at them. + core.reconcile_staging(); { let core = core.clone(); std::thread::Builder::new()