nvrl/proton-drive-linux
1
0 Fork 0
mirror of https://github.com/narrrl/proton-drive-linux.git synced 2026-11-04 00:47:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

fixed some conflict bugs + ui fix

Browse Source
This commit is contained in:
nvrl committed 2026-07-19 15:36:04 +02:00
1 parent 1d5fdfb6b9
commit 190ea3e9cb
8 files changed
+569 -112
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+3 -2
View File
@@ -3805,7 +3805,7 @@ dependencies = [
[[package]]
name = "proton-drive-rs"
version = "0.1.8"
version = "0.1.9"
dependencies = [
"async-trait",
"base64",
@@ -3830,7 +3830,7 @@ dependencies = [
[[package]]
name = "proton-sdk"
version = "0.1.8"
version = "0.1.9"
dependencies = [
"aes-gcm 0.11.0",
"async-trait",
@@ -3843,6 +3843,7 @@ dependencies = [
"hex",
"hkdf 0.13.0",
"hmac 0.13.0",
"lru",
"num-bigint 0.5.1",
"num-traits",
"pgp",
crates/pdfs-core/src/cache.rs
+296 -67
View File
@@ -34,6 +34,27 @@ use crate::error::Result;
const KIND_BLOB: &str = "blob";
/// `cache_entries.kind` tag for on-demand block-cache chunks.
const KIND_BLOCK: &str = "block";
/// `cache_entries.kind` tag for cached thumbnails.
const KIND_THUMB: &str = "thumb";
/// How the configured budget is divided between the three pools, in percent.
///
/// The alternative — one shared cap across all kinds — was rejected: the pools
/// have genuinely different lifetimes, and sharing lets one starve another. A
/// single streaming read is millions of block bytes and would evict the whole
/// blob cache behind it; a gallery scroll would do the same with thumbnails.
///
/// Whole-file blobs get the bulk: they are what "available offline" means, and
/// they are the only pool a user explicitly asks for (by pinning). Blocks are
/// transient partial reads, cheaply re-fetched. Thumbnails are tiny
/// individually — a few percent buys tens of thousands of them — but there is
/// one per photo, so the pool needs a ceiling, which is what C6 was about.
///
/// Must sum to 100; asserted at compile time below.
const SPLIT_BLOB: u64 = 70;
const SPLIT_BLOCK: u64 = 25;
const SPLIT_THUMB: u64 = 5;
const _: () = assert!(SPLIT_BLOB + SPLIT_BLOCK + SPLIT_THUMB == 100);
/// Current wall-clock time in unix *milliseconds*, for the LRU `last_accessed`
/// column. Milliseconds (not seconds) so two cache events in the same second
@@ -180,6 +201,7 @@ pub struct ContentCache {
/// and the next open reconciles those.
blob_bytes: AtomicU64,
block_bytes: AtomicU64,
thumb_bytes: AtomicU64,
/// Unified metadata DB. Its `cache_entries` table is the LRU index: every
/// store/read/evict updates it, and the budget enforcers query it instead of
/// scanning the cache directories (plan.md P4).
@@ -224,6 +246,7 @@ impl ContentCache {
max_bytes: AtomicU64::new(max_bytes),
blob_bytes: AtomicU64::new(0),
block_bytes: AtomicU64::new(0),
thumb_bytes: AtomicU64::new(0),
db,
};
cache.reconcile()?;
@@ -265,6 +288,7 @@ impl ContentCache {
for (dir, kind) in [
(&self.content_dir, KIND_BLOB),
(&self.block_dir, KIND_BLOCK),
(&self.thumb_dir, KIND_THUMB),
] {
let Ok(rd) = std::fs::read_dir(dir) else {
continue;
@@ -272,7 +296,15 @@ impl ContentCache {
for entry in rd.flatten() {
let name = entry.file_name();
let Some(name) = name.to_str() else { continue };
if name.ends_with(".meta") || name.ends_with(".tmp") {
if name.ends_with(".meta") {
continue;
}
// A `.tmp` is a staging file from a store that never completed
// its rename — a crashed or killed run. Nothing will ever claim
// it, and no pass other than this one looks in these
// directories, so it would sit there forever. Sweep it.
if name.ends_with(".tmp") {
let _ = std::fs::remove_file(entry.path());
continue;
}
let Ok(meta) = entry.metadata() else { continue };
@@ -470,25 +502,45 @@ impl ContentCache {
Ok(())
}
/// Re-read both pool totals from the cache index, which is authoritative.
/// Re-read every pool total from the cache index, which is authoritative.
fn reseed_totals(&self) {
if let Ok(n) = self.db.cache_total_bytes(KIND_BLOB) {
self.blob_bytes.store(n, Ordering::Relaxed);
}
if let Ok(n) = self.db.cache_total_bytes(KIND_BLOCK) {
self.block_bytes.store(n, Ordering::Relaxed);
for kind in [KIND_BLOB, KIND_BLOCK, KIND_THUMB] {
if let Ok(n) = self.db.cache_total_bytes(kind) {
self.pool(kind).store(n, Ordering::Relaxed);
}
}
}
/// The running total for a pool.
fn pool(&self, kind: &str) -> &AtomicU64 {
if kind == KIND_BLOCK {
&self.block_bytes
} else {
&self.blob_bytes
match kind {
KIND_BLOCK => &self.block_bytes,
KIND_THUMB => &self.thumb_bytes,
_ => &self.blob_bytes,
}
}
/// This pool's share of the configured budget (`0` = unlimited, as for the
/// whole cache). See [`SPLIT_BLOB`] for why the budget is divided rather
/// than shared.
fn pool_cap(&self, kind: &str) -> u64 {
let cap = self.cap();
if cap == 0 {
return 0;
}
let share = match kind {
KIND_BLOCK => SPLIT_BLOCK,
KIND_THUMB => SPLIT_THUMB,
_ => SPLIT_BLOB,
};
// Multiply before dividing: `cap / 100 * share` truncates a small budget
// to zero, which `pool_cap`'s callers would read as "unlimited" — the
// opposite of what a small budget asks for.
//
// At least one byte, for the same reason.
(cap.saturating_mul(share) / 100).max(1)
}
/// Evict least-recently-used block-cache files until the block dir fits
/// `max_bytes`. No-op when the cap is disabled (`0`). All blocks are
/// evictable — pinned files are served from whole-file blobs, never blocks.
@@ -508,7 +560,7 @@ impl ContentCache {
/// (pinned blobs). A pass that can find no eligible victim stops rather than
/// looping: a cache held entirely by pins legitimately stays over budget.
fn enforce_pool(&self, kind: &str, dir: &Path, exempt: Option<&HashSet<String>>) {
let cap = self.cap();
let cap = self.pool_cap(kind);
if cap == 0 {
return;
}
@@ -653,8 +705,14 @@ impl ContentCache {
self.reseed_totals();
}
/// Cache-index key for the `ttype` thumbnail of `uid` — the thumbnail file's
/// name, so eviction can remove the file by joining it onto `thumb_dir`.
fn thumb_key(&self, uid: &NodeUid, ttype: i32) -> String {
format!("{}.t{ttype}", Self::key(uid))
}
fn thumb_blob(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
self.thumb_dir.join(format!("{}.t{ttype}", Self::key(uid)))
self.thumb_dir.join(self.thumb_key(uid, ttype))
}
fn thumb_meta(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
@@ -672,7 +730,14 @@ impl ContentCache {
if recorded != mtime {
return None;
}
std::fs::read(self.thumb_blob(uid, ttype)).ok()
let bytes = std::fs::read(self.thumb_blob(uid, ttype)).ok()?;
// Record the access for LRU, as the blob and block readers do. Without
// it the pool evicts in insertion order, so scrolling back up a gallery
// would drop exactly the tiles being looked at.
let _ = self
.db
.cache_accessed(&self.thumb_key(uid, ttype), now_millis());
Some(bytes)
}
/// On-disk path of the cached content blob for `uid`, validated against
@@ -699,7 +764,13 @@ impl ContentCache {
return None;
}
let blob = self.thumb_blob(uid, ttype);
blob.exists().then_some(blob)
if !blob.exists() {
return None;
}
let _ = self
.db
.cache_accessed(&self.thumb_key(uid, ttype), now_millis());
Some(blob)
}
/// On-disk path where `uid`'s `ttype` thumbnail lives once stored.
@@ -732,15 +803,33 @@ impl ContentCache {
std::fs::write(&tmp, bytes)?;
std::fs::rename(&tmp, &blob)?;
std::fs::write(self.thumb_meta(uid, ttype), serde_json::to_vec(&mtime)?)?;
self.db.cache_touch(
&self.thumb_key(uid, ttype),
KIND_THUMB,
bytes.len() as u64,
now_millis(),
)?;
self.thumb_bytes
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
self.enforce_thumb_budget();
Ok(())
}
/// Evict least-recently-used thumbnails until the pool fits its share of the
/// budget. Every thumbnail is evictable, including those of pinned nodes: a
/// thumbnail is derived data that the gallery re-fetches or regenerates, and
/// exempting them would mean a large pinned library could hold the pool
/// permanently over its ceiling — the exact unbounded growth C6 was about.
fn enforce_thumb_budget(&self) {
self.enforce_pool(KIND_THUMB, &self.thumb_dir, None);
}
/// Evict least-recently-used *unpinned* blobs until total blob bytes fit the
/// configured `max_bytes` cap. No-op when the cap is disabled (`0`) or the
/// cache already fits. Pinned blobs are skipped, so a cache held entirely by
/// pins can legitimately stay over budget.
fn enforce_budget(&self) {
let cap = self.cap();
let cap = self.pool_cap(KIND_BLOB);
if cap == 0 || self.blob_bytes.load(Ordering::Relaxed) <= cap {
// The common case, settled without a query — and in particular
// without the recursive pin CTE below, which used to run on every
@@ -801,33 +890,31 @@ impl ContentCache {
.collect()
}
/// Total bytes of cached content blobs (pinned + unpinned), matching what
/// [`enforce_budget`](Self::enforce_budget) weighs against the cap.
/// Thumbnails live in a sibling dir and are not counted. Cheap directory
/// scan, safe to call from a front-end for a usage read-out.
/// Total bytes the cache holds on disk — blobs, blocks and thumbnails
/// together — which is what the configured budget caps and therefore the
/// only honest thing to show next to it.
///
/// This used to scan `content_dir` alone, which excluded both the block dir
/// and the thumb dir: the Settings page could report a few hundred megabytes
/// while the cache held several gigabytes. Now it reads the same running
/// totals the budget enforcers use, so the displayed number and the enforced
/// number cannot disagree — and it costs three atomic loads instead of a
/// directory scan.
///
/// Excludes the `.meta` sidecars (a few dozen bytes each) and the scratch
/// and staging dirs, which hold unuploaded user writes rather than cache and
/// are deliberately not evictable.
pub fn usage(&self) -> u64 {
let Ok(rd) = std::fs::read_dir(&self.content_dir) else {
return 0;
};
let mut total = 0u64;
for entry in rd.flatten() {
let Some(name) = entry.file_name().to_str().map(str::to_owned) else {
continue;
};
if name.ends_with(".meta") || name.ends_with(".tmp") {
continue;
}
if let Ok(meta) = entry.metadata()
&& meta.is_file()
{
total += meta.len();
}
}
total
[KIND_BLOB, KIND_BLOCK, KIND_THUMB]
.iter()
.map(|k| self.pool(k).load(Ordering::Relaxed))
.sum()
}
/// Configured soft byte cap (`0` = unlimited), for display alongside
/// [`usage`](Self::usage).
/// [`usage`](Self::usage). This is the whole-cache figure; internally it is
/// divided between the pools (see [`SPLIT_BLOB`]), but the user set one
/// number and gets that number.
pub fn budget(&self) -> u64 {
self.cap()
}
@@ -846,6 +933,7 @@ impl ContentCache {
self.max_bytes.store(bytes, Ordering::Relaxed);
self.enforce_budget();
self.enforce_block_budget();
self.enforce_thumb_budget();
}
/// Delete every *unpinned* cached blob plus all on-demand block chunks,
@@ -874,13 +962,19 @@ impl ContentCache {
freed = freed.saturating_add(size);
}
}
// Every on-demand block (transient partial reads, re-fetched on demand).
if let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOCK) {
for (key, size) in entries {
let _ = std::fs::remove_file(self.block_dir.join(&key));
let _ = std::fs::remove_file(self.block_dir.join(format!("{key}.meta")));
let _ = self.db.cache_remove(&key);
freed = freed.saturating_add(size);
// Every on-demand block (transient partial reads, re-fetched on demand)
// and every thumbnail (derived, re-fetched or regenerated on demand).
// Both pools are dropped wholesale rather than filtered by pin: neither
// is what "keep this available offline" means, and leaving them behind
// would make "clear cache" free visibly less than it reported.
for (kind, dir) in [(KIND_BLOCK, &self.block_dir), (KIND_THUMB, &self.thumb_dir)] {
if let Ok(entries) = self.db.cache_entries_by_kind(kind) {
for (key, size) in entries {
let _ = std::fs::remove_file(dir.join(&key));
let _ = std::fs::remove_file(dir.join(format!("{key}.meta")));
let _ = self.db.cache_remove(&key);
freed = freed.saturating_add(size);
}
}
}
self.reseed_totals();
@@ -925,6 +1019,28 @@ mod tests {
cache_capped(0)
}
/// A cache whose `kind` pool holds exactly `bytes`.
///
/// The configured budget is a whole-cache figure divided between the pools
/// (C7), while a test that exercises eviction cares about one pool at a
/// time. So the test states the pool size it means and this inverts the
/// split, rather than every such test hard-coding a number that silently
/// changes meaning if the split is ever retuned.
fn cache_with_pool_cap(kind: &str, bytes: u64) -> (ContentCache, TempDir) {
let share = match kind {
KIND_BLOCK => SPLIT_BLOCK,
KIND_THUMB => SPLIT_THUMB,
_ => SPLIT_BLOB,
};
// Round up, so the pool holds at least `bytes` rather than one short.
let (c, d) = cache_capped((bytes * 100).div_ceil(share));
assert!(
c.pool_cap(kind) >= bytes,
"helper must give the {kind} pool at least {bytes} bytes"
);
(c, d)
}
fn cache_capped(max_bytes: u64) -> (ContentCache, TempDir) {
let dir = TempDir::new();
let db = Arc::new(Db::open_in_memory().unwrap());
@@ -1054,23 +1170,33 @@ mod tests {
// Cap far above what we store, so every store stays under budget.
let (c, _d) = cache_capped(1 << 30);
let payload = vec![0u8; 4096];
// Populate enough entries that a full scan would be visibly slower.
for i in 0..2000u64 {
c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
}
// Cost of a store against a near-empty index...
let t = Instant::now();
for i in 2000..2500u64 {
for i in 0..500u64 {
c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
}
let per_store = t.elapsed() / 500;
let empty = t.elapsed() / 500;
println!("B4: store_block while under budget — {per_store:?} each");
// Generous bound: the point is that it does not grow with the 2000
// existing entries, not that any particular microsecond count holds.
// ...and against one with a few thousand entries in it.
for i in 500..2500u64 {
c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
}
let t = Instant::now();
for i in 2500..3000u64 {
c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
}
let full = t.elapsed() / 500;
println!("B4: store_block under budget — {empty:?} at ~0 entries, {full:?} at 2500");
// A ratio, not a wall-clock bound: the claim is that the under-budget
// path does not grow with the index, and only a ratio tests that. An
// absolute threshold measures the machine's load instead, and this test
// runs alongside the rest of the suite.
assert!(
per_store < std::time::Duration::from_millis(1),
"under-budget store should not scan the index; took {per_store:?}"
full < empty * 3,
"under-budget store should not scale with the index: \
{empty:?} at ~0 entries vs {full:?} at 2500"
);
}
@@ -1079,7 +1205,7 @@ mod tests {
/// over-budget stores cannot drift them.
#[test]
fn running_totals_stay_aligned_with_the_index() {
let (c, _d) = cache_capped(8);
let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8);
for i in 0..12u64 {
c.store_block(&uid("a"), 1, 100, i, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(2));
@@ -1101,8 +1227,8 @@ mod tests {
#[test]
fn budget_evicts_lru_unpinned() {
// Cap fits two 4-byte blobs but not three.
let (c, _d) = cache_capped(8);
// Blob pool fits two 4-byte blobs but not three.
let (c, _d) = cache_with_pool_cap(KIND_BLOB, 8);
let (a, b, d) = (uid("a"), uid("b"), uid("d"));
c.store(&a, 1, 4, b"aaaa").unwrap();
@@ -1123,7 +1249,7 @@ mod tests {
#[test]
fn budget_never_evicts_pinned() {
let (c, _d) = cache_capped(8);
let (c, _d) = cache_with_pool_cap(KIND_BLOB, 8);
let (a, b, d) = (uid("a"), uid("b"), uid("d"));
c.store(&a, 1, 4, b"aaaa").unwrap();
@@ -1166,13 +1292,15 @@ mod tests {
std::thread::sleep(std::time::Duration::from_millis(10));
c.store(&d, 1, 4, b"dddd").unwrap();
// Tightening to 8 bytes evicts the LRU unpinned blob(s) now, not on the
// next store.
c.set_budget(8);
// Tightening to a budget whose blob share is 8 bytes evicts the LRU
// unpinned blob(s) now, not on the next store.
c.set_budget((8 * 100u64).div_ceil(SPLIT_BLOB));
assert!(!c.is_cached(&a, 1, 4), "oldest blob evicted on tighten");
assert!(c.is_cached(&b, 1, 4));
assert!(c.is_cached(&d, 1, 4));
assert_eq!(c.budget(), 8);
// `budget()` reports the whole-cache figure that was set, not one
// pool's share of it.
assert_eq!(c.budget(), (8 * 100u64).div_ceil(SPLIT_BLOB));
}
/// **The A2 reproduce.** Thumbnail types 1 and 2 of one node must not share
@@ -1242,6 +1370,107 @@ mod tests {
assert!(c.read_thumbnail(&u, 2, 100).is_none());
}
/// Total bytes of every real file under `root`, recursively — what `du`
/// would report, and what `usage()` claims to be reporting.
fn bytes_on_disk(root: &Path) -> u64 {
let Ok(rd) = std::fs::read_dir(root) else {
return 0;
};
let mut total = 0;
for entry in rd.flatten() {
let Ok(meta) = entry.metadata() else { continue };
if meta.is_dir() {
total += bytes_on_disk(&entry.path());
} else if meta.is_file() {
total += meta.len();
}
}
total
}
/// C6/C7. Two claims the cache makes and does not keep.
///
/// The configured budget is the whole promise of the Settings page: it is
/// the only thing standing between an on-demand filesystem and a full disk.
/// A thumbnail per photo, on a library of tens of thousands, is not a
/// rounding error.
#[test]
fn every_pool_is_budgeted_and_visible() {
const CAP: u64 = 256 * 1024;
let (c, dir) = cache_capped(CAP);
let payload = vec![0u8; 8 * 1024];
// A gallery scroll: one thumbnail per photo, far past the cap.
for i in 0..200u64 {
c.store_thumbnail(&uid(&format!("photo{i}")), 1, 1, &payload)
.unwrap();
}
// Plus enough blob and block traffic to fill both other pools.
for i in 0..100u64 {
c.store(&uid(&format!("file{i}")), 1, payload.len() as u64, &payload)
.unwrap();
c.store_block(&uid("streamed"), 1, 1 << 30, i, &payload)
.unwrap();
}
let on_disk = bytes_on_disk(dir.path());
let reported = c.usage();
// 1. The cap is a cap on the cache, not on one pool of it.
assert!(
on_disk <= CAP + CAP / 8,
"cache holds {on_disk} bytes against a {CAP}-byte budget"
);
// 2. And the number shown to the user is the number on disk. Meta
// sidecars are small and untracked, hence the tolerance.
let drift = on_disk.abs_diff(reported);
assert!(
drift <= on_disk / 8,
"usage() reports {reported} against {on_disk} on disk"
);
}
/// The thumbnail pool must evict by *use*, not by insertion order. A gallery
/// scrolled down and back up re-reads the tiles it already has; under FIFO
/// those are exactly the ones dropped, so every scroll back would re-fetch.
#[test]
fn thumbnails_evict_least_recently_used() {
// Pool fits two 4-byte thumbnails but not three.
let (c, _d) = cache_with_pool_cap(KIND_THUMB, 8);
let (a, b, d) = (uid("a"), uid("b"), uid("d"));
c.store_thumbnail(&a, 1, 1, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_thumbnail(&b, 1, 1, b"bbbb").unwrap();
// Look at `a` again, making `b` the least-recently-used.
std::thread::sleep(std::time::Duration::from_millis(10));
assert!(c.read_thumbnail(&a, 1, 1).is_some());
std::thread::sleep(std::time::Duration::from_millis(10));
c.store_thumbnail(&d, 1, 1, b"dddd").unwrap();
assert!(c.read_thumbnail(&a, 1, 1).is_some(), "recently read, kept");
assert!(c.read_thumbnail(&b, 1, 1).is_none(), "least recently used");
assert!(c.read_thumbnail(&d, 1, 1).is_some(), "just stored");
}
/// Evicting a node drops its thumbnails from the index as well as from disk.
/// A row left behind would keep counting bytes that are gone, and the
/// running total would drift above the truth until the next restart.
#[test]
fn evicting_a_node_clears_its_thumbnail_accounting() {
let (c, _d) = cache();
let u = uid("a");
c.store(&u, 1, 4, b"aaaa").unwrap();
c.store_thumbnail(&u, 1, 1, b"thumb1").unwrap();
c.store_thumbnail(&u, 2, 1, b"preview2").unwrap();
assert!(c.thumb_bytes.load(Ordering::Relaxed) > 0);
c.evict(&u);
assert_eq!(c.thumb_bytes.load(Ordering::Relaxed), 0);
assert!(c.db.cache_entries_by_kind(KIND_THUMB).unwrap().is_empty());
assert_eq!(c.usage(), 0, "nothing cached, nothing reported");
}
#[test]
fn evict_removes_blob() {
let (c, _d) = cache();
@@ -1277,8 +1506,8 @@ mod tests {
#[test]
fn block_budget_evicts_lru() {
// Cap fits two 4-byte blocks but not three.
let (c, _d) = cache_capped(8);
// Block pool fits two 4-byte blocks but not three.
let (c, _d) = cache_with_pool_cap(KIND_BLOCK, 8);
let u = uid("a");
c.store_block(&u, 1, 12, 0, b"aaaa").unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
crates/pdfs-core/src/db/cache.rs
+8 -5
View File
@@ -52,14 +52,17 @@ impl Db {
Ok(())
}
/// Drop the whole-file blob row for `key` plus every block row for the same
/// uid (`<key>.b<idx>`). Called by `ContentCache::evict`, which removes the
/// blob and all of a uid's cached blocks at once.
/// Drop the whole-file blob row for `key` plus every derived row for the
/// same uid — blocks (`<key>.b<idx>`) and thumbnails (`<key>.t<type>`).
/// Called by `ContentCache::evict`, which removes the blob, all of a uid's
/// cached blocks and all of its thumbnails at once; a row left behind here
/// would keep counting bytes that are no longer on disk.
pub fn cache_remove_all(&self, key: &str) -> Result<()> {
let conn = self.conn.lock();
conn.execute(
"DELETE FROM cache_entries WHERE cache_key = ?1 OR cache_key LIKE ?2",
params![key, format!("{key}.b%")],
"DELETE FROM cache_entries
WHERE cache_key = ?1 OR cache_key LIKE ?2 OR cache_key LIKE ?3",
params![key, format!("{key}.b%"), format!("{key}.t%")],
)?;
Ok(())
}
crates/pdfs-core/src/db/mod.rs
+18
View File
@@ -52,6 +52,13 @@ pub use trash::StoredTrash;
/// outliers rather than fighting the normal write path for disk.
const WAL_SIZE_LIMIT: i64 = 64 * 1024 * 1024;
/// How long a statement waits for a database lock before giving up.
///
/// Long enough to outlast any transaction this daemon writes (the largest, a
/// full-index FTS rebuild, is well under a second), short enough that a
/// genuinely stuck holder surfaces as an error rather than an apparent hang.
const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
/// Handle to the unified metadata database.
///
/// Cheap to wrap in an `Arc`; clone the `Arc`, not this. All access goes through
@@ -77,6 +84,17 @@ impl Db {
// forever — a multi-GB file next to a database two orders of magnitude
// smaller. Checkpointing is unaffected; only the file is truncated back.
conn.pragma_update(None, "journal_size_limit", WAL_SIZE_LIMIT)?;
// Wait for a lock rather than failing instantly on SQLITE_BUSY.
//
// This is deliberately explicit and deliberately redundant: rusqlite
// already applies a five-second busy timeout of its own, so this changes
// nothing today. It is here so the value is *ours* — a behaviour the
// daemon depends on should not be an inherited default that a dependency
// bump can silently change. `open_bounds_the_wal_size` pins it.
//
// (Bare SQLite does default to 0, which is where the belief that this
// was unset came from. That default is not what we get.)
conn.busy_timeout(BUSY_TIMEOUT)?;
let db = Self {
conn: Mutex::new(conn),
crates/pdfs-core/src/db/ops.rs
+23
View File
@@ -188,6 +188,29 @@ impl Db {
Ok(Some(AttachedBlob { id, superseded }))
}
/// Replace the sidecar of a queued op, for a baseline that has moved under
/// it.
///
/// Exists for one caller: when *our own* upload seals a new revision while a
/// further write to the same node is already queued, that queued write's
/// [`StagedWrite::based_on`](crate::cache::StagedWrite::based_on) still
/// names the revision we just replaced. Left alone it would read as "another
/// device changed this file" and divert the write into a conflict copy over
/// nothing — a self-conflict. Restamping it is what keeps a chain of writes
/// to one file a chain rather than a pile of copies.
///
/// Deliberately narrow: only the sidecar changes, so the blob, the attempt
/// count and the backoff all survive. Returns false when no such op is
/// queued, which is the ordinary case.
pub fn update_op_meta(&self, uid: &str, kind: &str, meta_json: &str) -> Result<bool> {
let conn = self.conn.lock();
let n = conn.execute(
"UPDATE pending_op SET meta_json = ?3 WHERE uid = ?1 AND kind = ?2",
params![uid, kind, meta_json],
)?;
Ok(n > 0)
}
/// Point a queued create at a new parent and name, for a node renamed or
/// moved before it ever reached the server.
///
crates/pdfs-core/src/db/tests.rs
+111 -12
View File
@@ -26,10 +26,60 @@ fn open_bounds_the_wal_size() {
.unwrap();
assert_eq!(limit, WAL_SIZE_LIMIT);
// Without a busy timeout the default is 0: a lock held by anyone else fails
// the statement instantly instead of waiting for it to clear.
let busy: i64 = conn
.query_row("PRAGMA busy_timeout", [], |r| r.get(0))
.unwrap();
assert_eq!(busy, BUSY_TIMEOUT.as_millis() as i64);
drop(conn);
drop(db);
let _ = std::fs::remove_file(&path);
}
/// A second connection to the same file — a hand-started `pdfs mount` alongside
/// the systemd service, a backup tool, a stray `sqlite3` session — holds a write
/// lock briefly. Our write must wait for it rather than failing on the spot.
///
/// **This passed before the busy timeout was set explicitly**, because rusqlite
/// already applies one. It is kept as a characterisation test, not as evidence
/// of a fix: it states the behaviour the daemon relies on, so that losing it
/// (dropping the pragma, or a dependency changing its default) fails here rather
/// than in front of a user.
#[test]
fn a_write_waits_for_a_competing_writer() {
use std::time::Duration;
let path = std::env::temp_dir().join(format!("pdfs-db-busy-{}.db", std::process::id()));
let _ = std::fs::remove_file(&path);
let db = Db::open(&path).unwrap();
// A second process's connection takes an exclusive write lock and holds it
// for a moment, as any real transaction would.
let holder = {
let path = path.clone();
std::thread::spawn(move || {
let other = rusqlite::Connection::open(&path).unwrap();
other.busy_timeout(Duration::from_secs(5)).unwrap();
other.execute_batch("BEGIN IMMEDIATE").unwrap();
std::thread::sleep(Duration::from_millis(300));
other.execute_batch("COMMIT").unwrap();
})
};
// Let the holder actually acquire the lock before we contend for it.
std::thread::sleep(Duration::from_millis(50));
// This is the assertion: it blocks until the holder commits, rather than
// returning SQLITE_BUSY.
db.set_state_i64("busy_probe", 7).unwrap();
holder.join().unwrap();
assert_eq!(db.state_i64("busy_probe").unwrap(), Some(7));
drop(db);
let _ = std::fs::remove_file(&path);
}
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
fn uid(link: &str) -> NodeUid {
@@ -541,6 +591,56 @@ fn opens_and_migrates() {
assert_eq!(version, SCHEMA_VERSION.to_string());
}
/// A queued write whose baseline is restamped must keep everything else.
///
/// The restamp happens after our *own* upload seals a new revision under a
/// still-queued write. If it took the blob or the retry state with it, the fix
/// for a spurious conflict copy would cost the bytes that conflict copy was
/// there to protect.
#[test]
fn restamping_a_baseline_leaves_the_blob_and_retry_state_alone() {
let db = Db::open_in_memory().unwrap();
let op = PendingOp {
id: 0,
kind: OP_REVISION.to_string(),
uid: uid("a").to_string(),
parent_uid: None,
name: None,
blob_path: Some("/staging/blob".to_string()),
meta_json: Some(r#"{"based_on":"old"}"#.to_string()),
created_at: 1,
attempts: 0,
last_error: None,
next_attempt_at: 0,
};
db.enqueue_op(&op).unwrap();
db.record_op_failure(db.pending_ops().unwrap()[0].id, "offline", 999)
.unwrap();
let updated = db
.update_op_meta(&uid("a").to_string(), OP_REVISION, r#"{"based_on":"new"}"#)
.unwrap();
assert!(updated, "the queued write is there to restamp");
let ops = db.pending_ops().unwrap();
assert_eq!(ops.len(), 1);
assert_eq!(ops[0].meta_json.as_deref(), Some(r#"{"based_on":"new"}"#));
assert_eq!(
ops[0].blob_path.as_deref(),
Some("/staging/blob"),
"the only copy of the user's bytes"
);
assert_eq!(ops[0].attempts, 1, "backoff survives a restamp");
assert_eq!(ops[0].next_attempt_at, 999);
// No queued write for that node is the ordinary case, not an error: most
// uploads are the last one for their file.
assert!(
!db.update_op_meta(&uid("b").to_string(), OP_REVISION, "{}")
.unwrap()
);
}
#[test]
fn a_second_write_supersedes_the_first_pending_op() {
let db = Db::open_in_memory().unwrap();
@@ -1182,9 +1282,11 @@ fn next_due_op_respects_backoff() {
/// under budget" — under the shared connection lock, so it also stalled FUSE
/// metadata calls.
///
/// Asserts the shape: the under-budget check must not scale with cache size.
/// Asserts correctness — the aggregate must agree with summing the rows, and
/// must count the kinds apart — and *reports* the timing without asserting on
/// it. See the note at the measurement for why.
#[test]
fn cache_total_bytes_does_not_scale_with_cache_size() {
fn cache_total_bytes_agrees_with_summing_the_rows() {
use std::time::Instant;
let db = Db::open_in_memory().unwrap();
@@ -1223,17 +1325,14 @@ fn cache_total_bytes_does_not_scale_with_cache_size() {
}
let aggregate = t1.elapsed();
// Reported, not asserted. The SUM is cheaper than reading every row — no
// materialization, no allocation, no sort — but it is still O(rows), so this
// is a constant factor, and an assertion on a constant factor measured
// alongside the rest of the suite tests the machine's load as much as the
// query. The claim that matters — that the *common* path does not touch the
// database at all — is pinned by `budget_check_is_free_when_under_budget`,
// which compares against itself and so is load-independent.
println!("B4: {rounds} budget checks over {N} entries — scan {scan:?}, aggregate {aggregate:?}");
// The SUM is much cheaper — no row materialization, no allocation, no sort —
// but it is still O(rows): summing N values reads N index entries. That is
// why `ContentCache` keeps a running total on top of this and only falls
// back to the query when it has to evict. See
// `budget_check_is_free_when_under_budget`.
assert!(
aggregate * 2 < scan,
"expected the SUM to be materially cheaper than reading every row; \
scan={scan:?} aggregate={aggregate:?}"
);
}
/// Victims still come out least-recently-accessed first, now in bounded batches
crates/pdfs-fuse/src/drain.rs
+90 -21
View File
@@ -24,9 +24,10 @@ use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use pdfs_core::cache::StagedWrite;
use pdfs_core::cache::{Baseline, StagedWrite};
use pdfs_core::control::{ActivityKind, TransferDirection};
use pdfs_core::db::{OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp};
use proton_drive_rs::Node;
use proton_drive_rs::proton_sdk::ids::NodeUid;
use tracing::{debug, error, info, warn};
@@ -511,7 +512,7 @@ impl Core {
let name = place
.as_ref()
.map(|(_, n)| n.clone())
.unwrap_or_else(|| meta.uid.clone());
.unwrap_or_else(|| self.node_name(uid));
// A node the tree has forgotten still has bytes worth keeping, so the
// copy falls back to the root rather than being abandoned.
let Some(parent) = place.map(|(p, _)| p).or_else(|| self.root_uid()) else {
@@ -625,11 +626,38 @@ impl Core {
/// A node's parent uid and name, as the tree currently has them.
pub(crate) fn node_place(&self, uid: &NodeUid) -> Option<(NodeUid, String)> {
let st = self.state.lock();
let ino = st.by_uid.get(uid)?;
let entry = st.entries.get(ino)?;
let parent = entry.node.parent_uid.clone()?;
Some((parent, entry.node.name.clone()))
{
let st = self.state.lock();
if let Some(entry) = st.by_uid.get(uid).and_then(|ino| st.entries.get(ino))
&& let Some(parent) = entry.node.parent_uid.clone()
{
return Some((parent, entry.node.name.clone()));
}
}
// The in-memory tree only holds what has been walked to since the daemon
// started, and the drain routinely runs before anything has walked to
// this file — after a restart, or for a write that arrived by path. The
// DB knows it anyway. Without this fallback a conflict copy was named
// after the node's *uid* and dumped in the root, which is how a file
// called `G88km…==~c2do…== (sync-conflict 1784429627)` appears at the
// top of someone's Drive.
let node = self.db.node_by_uid(&uid.to_string()).ok().flatten()?;
Some((node.parent_uid.clone()?, node.name.clone()))
}
/// The name to show a user for `uid`, for a transfer entry or an activity
/// log line.
///
/// Falls back to something short and human rather than to the uid: a uid is
/// 130 characters of base64 that identifies the file to us and to nobody
/// else, and it has ended up in both the activity log and on real files in
/// the Drive root. The link prefix keeps it traceable without pretending to
/// be a name.
pub(crate) fn node_name(&self, uid: &NodeUid) -> String {
self.node_place(uid).map(|(_, n)| n).unwrap_or_else(|| {
let short: String = uid.link_id.to_string().chars().take(8).collect();
format!("recovered-{short}")
})
}
/// The uid of the My Files root, which every node in the mount descends
@@ -681,14 +709,7 @@ impl Core {
.map_err(|e| self.errno_error(e, "gap-fill from base failed"))?;
}
let name = {
let st = self.state.lock();
st.by_uid
.get(&uid)
.and_then(|ino| st.entries.get(ino))
.map(|e| e.node.name.clone())
.unwrap_or_else(|| meta.uid.clone())
};
let name = self.node_name(&uid);
let guard =
self.transfers
.begin(&name, meta.uid.clone(), TransferDirection::Upload, meta.len);
@@ -736,15 +757,16 @@ impl Core {
/// makes the *next* write to this file look like another device changed it
/// underneath us, and diverts it into a conflict copy over nothing.
///
/// A write queued while this upload was on the wire is handled instead by
/// [`Core::rebaseline_pending`]: its optimistic size/mtime must stay on the
/// node (the same reason `apply_event` refuses to overwrite a node with a
/// queued write), but its *baseline* still has to learn about the revision
/// we just sealed. Skipping both, as this used to, is what produced a file
/// that conflicted with itself and left a full-size duplicate behind.
///
/// Best effort: a failure here costs a spurious conflict copy on the next
/// write, not this upload, which has already landed.
pub(crate) fn refresh_after_upload(&self, uid: &NodeUid) {
// A write queued while this one was on the wire is ahead of the remote
// again, and its optimistic size/mtime must win — the same reason
// `apply_event` refuses to overwrite a node with a queued write.
if self.pending.lock().contains_key(uid) {
return;
}
let node = match self.fetch_node_remote(uid) {
Ok(Some(node)) => node,
// Trashed or deleted under us: the tree will hear it from the event
@@ -757,6 +779,12 @@ impl Core {
return;
}
};
// Ordered so that a write queued *during* the fetch above is still
// caught: it took its baseline from the node's optimistic stamp, and
// this overwrites it with the revision the server actually holds.
if self.rebaseline_pending(uid, &node) {
return;
}
let mut st = self.state.lock();
let Some(parent) = st
.by_uid
@@ -768,4 +796,45 @@ impl Core {
};
st.intern(parent, node);
}
/// Point a still-queued write at the revision this upload just sealed, and
/// report whether there was one.
///
/// The revision we sent *is* the base the queued write will be applied over,
/// but nothing else says so. [`Core::remote_baseline`] carries a baseline
/// across a supersede because the op it inherits from "is the last one that
/// actually observed the remote" — true only until that op drains. Once it
/// has, the inherited baseline names a revision we replaced ourselves, and
/// [`Core::revision_conflict`] reads that as another device having moved the
/// file. It is a self-conflict, and it is expensive: the whole staged blob
/// is re-uploaded as a second file.
///
/// Both copies of the sidecar are updated — the in-memory one the next
/// `release` inherits from, and the persisted one the drain reloads after a
/// restart. Best effort on the DB half; the in-memory half is what the
/// immediate next write reads.
fn rebaseline_pending(&self, uid: &NodeUid, sealed: &Node) -> bool {
let base = Baseline {
mtime: sealed.modification_time,
size: node_size(sealed),
};
let mut pending = self.pending.lock();
let Some(p) = pending.get_mut(uid) else {
return false;
};
p.meta.based_on = Some(base);
match serde_json::to_string(&p.meta) {
Ok(json) => {
if let Err(e) = self.db.update_op_meta(&uid.to_string(), OP_REVISION, &json) {
warn!(%uid, error = %e,
"restamping a queued write's baseline failed; \
it may land as a conflict copy of itself");
}
}
Err(e) => warn!(%uid, error = %e, "serializing a restamped sidecar failed"),
}
debug!(%uid, mtime = base.mtime, size = base.size,
"queued write rebaselined onto the revision just uploaded");
true
}
}
crates/pdfs-fuse/src/workers.rs
+20 -5
View File
@@ -49,11 +49,26 @@ impl Workers {
// `Reply` answers EIO on its own, so the caller of
// the failed op is told; the next job is unaffected.
//
// This only holds because the shared state is behind
// `parking_lot` mutexes: a `std` mutex held at the
// point of the panic would come back poisoned, and
// the worker we just rescued would die on its next
// acquisition — as would every other thread.
// This only holds while no shared state a job
// touches sits behind a *poisoning* lock. A
// `std::sync::Mutex` held at the point of the panic
// comes back poisoned, and the worker we just
// rescued dies on its next acquisition — as does
// every other thread. The rescue would then convert
// one recoverable EIO into a permanently broken
// daemon: strictly worse than not catching at all.
//
// This crate's state is behind `parking_lot`, which
// does not poison. **The requirement crosses the
// crate boundary**, though: a job runs deep into
// `proton-sdk`, whose entity cache is a `std`
// mutex. It recovers the guard rather than
// unwrapping (see `InMemoryCacheRepository::state`);
// its session and HTTP state are
// `tokio::sync::Mutex`, which has no poisoning at
// all. Anything new reached from a job owes the same
// check — an SDK-side `.lock().unwrap()` silently
// invalidates this comment.
Ok(job) => {
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job));
}