mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
relocate() called forget(), dropping the node's by_uid mapping, so the re-enumeration assigned a fresh inode. The kernel had already carried the renamed dentry onto the old one and holds it for the entry TTL, so lookup/getattr/opendir resolved to an inode entries no longer held and returned ENOENT — while readdir walked children and reported the new inode. A renamed directory listed in its parent but could not be entered until the TTL expired. Rewrite the node in place and invalidate both parents' listings instead. Also stops the moved node's DB row being deleted and recreated on every move, which the B1 data-loss fix worked around
1883 lines
78 KiB
Rust
1883 lines
78 KiB
Rust
//! On-disk content cache and pin registry.
|
|
//!
|
|
//! Files-On-Demand keeps file *content* off local disk by default: a `read`
|
|
//! hydrates the requested byte range straight from the remote. Two things make
|
|
//! that cache-worthy:
|
|
//!
|
|
//! * **Pinning** — the user marks a file "always keep on this device". Its full
|
|
//! plaintext is downloaded once and stored under [`AppDirs::cache_dir`]; every
|
|
//! later `read` is served from disk with no network.
|
|
//! * **Validation** — Proton does not hand us a stable revision id on the public
|
|
//! [`Node`], so a cached blob is validated against the node's
|
|
//! `(modification_time, plaintext_size)` pair. A new revision changes at least
|
|
//! the mtime, so a stale blob is detected and ignored (and the event poller
|
|
//! evicts it eagerly anyway).
|
|
//!
|
|
//! [`AppDirs::cache_dir`]: crate::config::AppDirs::cache_dir
|
|
//! [`Node`]: proton_drive_rs::Node
|
|
|
|
use std::collections::{BTreeMap, HashSet};
|
|
use std::io::{Read, Seek, SeekFrom};
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::time::{SystemTime, UNIX_EPOCH};
|
|
|
|
use proton_drive_rs::proton_sdk::ids::NodeUid;
|
|
use serde::{Deserialize, Serialize};
|
|
use sha2::{Digest, Sha256};
|
|
|
|
use crate::db::{CacheEntryInput, Db};
|
|
use crate::error::Result;
|
|
|
|
/// `cache_entries.kind` tag for whole-file content blobs.
|
|
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
|
|
/// still order correctly — a coarse second-granularity clock would make LRU
|
|
/// eviction order indeterminate under bursty access.
|
|
fn now_millis() -> i64 {
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.map(|d| d.as_millis() as i64)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// Block size for the on-demand block cache. Matches the SDK content block size
|
|
/// (`DEFAULT_BLOCK_SIZE`, 4 MiB) so each cached block maps to exactly one
|
|
/// `download_range` fetch with no straddling.
|
|
pub const BLOCK_SIZE: u64 = 1 << 22;
|
|
|
|
/// How many eviction candidates to pull per batch when a pool is over budget.
|
|
/// Large enough that a normal overshoot (a store or two past the cap) is settled
|
|
/// in one query, small enough that a badly-over-budget cache does not read its
|
|
/// whole index to drop the oldest few.
|
|
const EVICT_BATCH: usize = 64;
|
|
|
|
/// Validity tag stored alongside a cached blob. A blob is fresh only if both
|
|
/// fields still match the node's current metadata.
|
|
#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
|
|
struct Meta {
|
|
/// Node modification time, epoch seconds.
|
|
mtime: i64,
|
|
/// Plaintext size in bytes.
|
|
size: u64,
|
|
}
|
|
|
|
/// One pin, as carried over the control socket and listed in `status`.
|
|
#[derive(Serialize, Deserialize, Clone, Debug)]
|
|
pub struct Pin {
|
|
/// Node uid in `volume~link` display form.
|
|
pub uid: String,
|
|
/// Last path the node was pinned under, for display in `status`. Advisory
|
|
/// only — the uid is the identity.
|
|
pub path: String,
|
|
/// A folder pin whose whole subtree is kept on disk. `false` for a single
|
|
/// file. Defaulted so a legacy `pins.json` (file pins only) still parses.
|
|
#[serde(default)]
|
|
pub recursive: bool,
|
|
}
|
|
|
|
/// The legacy JSON pin registry, kept only to import a pre-P5 `pins.json` into
|
|
/// the DB once on open. Live pins are owned by the `pins` table.
|
|
#[derive(Serialize, Deserialize, Default)]
|
|
struct PinFile {
|
|
pins: BTreeMap<String, Pin>,
|
|
}
|
|
|
|
/// What a staged file actually contains, written beside it as `<name>.json`
|
|
/// (offline.md Phase 2).
|
|
///
|
|
/// A staged file is **not** necessarily valid whole-file content, and that is
|
|
/// the whole reason this exists. A partial overwrite commits by filling the
|
|
/// untouched regions from the remote base — which is exactly what fails when the
|
|
/// network is down, so what lands in staging is the authored bytes with *zeros*
|
|
/// 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.
|
|
#[derive(Serialize, Deserialize, Debug, Clone)]
|
|
pub struct StagedWrite {
|
|
/// Node the write targeted, in `volume~link` display form.
|
|
pub uid: String,
|
|
/// Length of the intended new content.
|
|
pub len: u64,
|
|
/// Size and mtime of the revision the file's *untouched* ranges came from,
|
|
/// i.e. what a gap-fill reads. Meaningless once `complete` is true.
|
|
pub base_size: u64,
|
|
pub base_mtime: i64,
|
|
/// Locally authored `[start, end)` ranges. Everything else in the file is a
|
|
/// zero-filled hole, not content.
|
|
pub authored: Vec<(u64, u64)>,
|
|
/// True when `authored` covers the whole file, i.e. the staged bytes are the
|
|
/// complete new content and can be uploaded directly.
|
|
pub complete: bool,
|
|
/// The remote revision this change was made against, if it is known.
|
|
///
|
|
/// Distinct from `base_size`/`base_mtime`, which describe wherever the
|
|
/// untouched bytes are to be read from — for a write that supersedes an
|
|
/// earlier queued one, that is the *previous staged blob*, not the remote.
|
|
/// This instead always names the server's revision, carried across
|
|
/// supersedes, so the drain can tell whether the remote moved on under a
|
|
/// queued change and keep a conflict copy rather than clobber it.
|
|
///
|
|
/// `None` on a sidecar written before this field existed, and for a node
|
|
/// that has never existed remotely; both mean "do not conflict-check".
|
|
#[serde(default)]
|
|
pub based_on: Option<Baseline>,
|
|
}
|
|
|
|
/// The identity of a remote revision, as far as we can observe it: a file whose
|
|
/// size and mtime both still match the ones a queued change was made against has
|
|
/// not been rewritten by anyone else in the meantime.
|
|
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct Baseline {
|
|
pub mtime: i64,
|
|
pub size: u64,
|
|
}
|
|
|
|
/// Content cache rooted at a directory, with a sibling pin-registry file.
|
|
pub struct ContentCache {
|
|
/// Directory holding `<key>` blobs and `<key>.meta` tags.
|
|
content_dir: PathBuf,
|
|
/// Subdirectory holding cached thumbnails (`<key>.t<n>` + `.meta`). Kept out
|
|
/// of `content_dir` so the budget scan never sees thumbnail files.
|
|
thumb_dir: PathBuf,
|
|
/// Subdirectory holding on-demand block-cache files (`<key>.b<idx>` +
|
|
/// `.meta`). Kept out of `content_dir` so the whole-file budget scan never
|
|
/// sees them; blocks carry their own LRU budget.
|
|
block_dir: PathBuf,
|
|
/// Subdirectory for write-handle scratch files (disk-backed write buffers).
|
|
/// Emptied on open so a crashed run leaves no orphans.
|
|
scratch_dir: PathBuf,
|
|
/// Subdirectory holding the bytes of writes that have not been uploaded yet
|
|
/// (offline.md Phase 2/3) — every released write passes through here, since
|
|
/// the upload is a queued op performed later. Unlike `scratch_dir` this is
|
|
/// **never** emptied on open: these are the only copy of content the user
|
|
/// authored, and the whole point is that they outlive the daemon.
|
|
staging_dir: PathBuf,
|
|
/// Subdirectory holding scratch files rescued from an unclean shutdown —
|
|
/// writes an application had `fsync`ed but not yet closed. Populated only at
|
|
/// open, drained by the daemon into `staging_dir` once it can address the
|
|
/// nodes they belong to. Like `staging_dir` and unlike `scratch_dir`, it is
|
|
/// never emptied blindly.
|
|
recovery_dir: PathBuf,
|
|
/// JSON pin registry path.
|
|
pins_path: PathBuf,
|
|
/// Soft cap on total blob bytes. Exceeded only transiently: a `store`
|
|
/// evicts least-recently-used *unpinned* blobs back under the cap. `0`
|
|
/// disables the cap. Pinned blobs are never evicted, so pins alone may push
|
|
/// the cache over budget. Atomic so the daemon can retune it at runtime
|
|
/// (a Settings-page change) via [`set_budget`](Self::set_budget) without
|
|
/// taking a lock on every cache read.
|
|
max_bytes: AtomicU64,
|
|
/// Running byte totals for the two budgeted pools, so the overwhelmingly
|
|
/// common "still under budget" answer costs an atomic load instead of a
|
|
/// database query.
|
|
///
|
|
/// These are a *fast path*, not the source of truth — the `cache_entries`
|
|
/// index is. They are seeded from it at open (right after `reconcile`, so
|
|
/// they describe what is actually on disk), advanced on each store, and
|
|
/// **re-seeded from the index after every eviction pass**, which is what
|
|
/// keeps a long-running daemon from drifting: the only operations that
|
|
/// change a total without going through here are external file deletions,
|
|
/// 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).
|
|
db: Arc<Db>,
|
|
}
|
|
|
|
impl ContentCache {
|
|
/// Open (and create) a cache under `content_dir`, with the pin registry at
|
|
/// `pins_path` and a `max_bytes` size cap (`0` = unlimited). Both parent
|
|
/// directories are created if missing. `db` is the shared metadata database
|
|
/// whose `cache_entries` table backs LRU eviction; the on-disk cache is
|
|
/// reconciled into it on open.
|
|
pub fn open(
|
|
content_dir: PathBuf,
|
|
pins_path: PathBuf,
|
|
max_bytes: u64,
|
|
db: Arc<Db>,
|
|
) -> Result<Self> {
|
|
std::fs::create_dir_all(&content_dir)?;
|
|
let thumb_dir = content_dir.join("thumbs");
|
|
std::fs::create_dir_all(&thumb_dir)?;
|
|
let block_dir = content_dir.join("blocks");
|
|
std::fs::create_dir_all(&block_dir)?;
|
|
// Scratch holds disk-backed write buffers. A previous run's leftovers are
|
|
// worthless *except* where the application asked for durability: a handle
|
|
// that was `fsync`ed but never closed left a sidecar behind, and POSIX
|
|
// says those bytes are on stable storage. Rescue those, discard the rest.
|
|
let scratch_dir = content_dir.join("scratch");
|
|
let recovery_dir = content_dir.join("recovery");
|
|
std::fs::create_dir_all(&recovery_dir)?;
|
|
Self::rescue_scratch(&scratch_dir, &recovery_dir);
|
|
let _ = std::fs::remove_dir_all(&scratch_dir);
|
|
std::fs::create_dir_all(&scratch_dir)?;
|
|
// Staging, by contrast, is deliberately preserved across runs.
|
|
let staging_dir = content_dir.join("staging");
|
|
std::fs::create_dir_all(&staging_dir)?;
|
|
if let Some(parent) = pins_path.parent() {
|
|
std::fs::create_dir_all(parent)?;
|
|
}
|
|
let cache = Self {
|
|
content_dir,
|
|
thumb_dir,
|
|
block_dir,
|
|
scratch_dir,
|
|
staging_dir,
|
|
recovery_dir,
|
|
pins_path,
|
|
max_bytes: AtomicU64::new(max_bytes),
|
|
blob_bytes: AtomicU64::new(0),
|
|
block_bytes: AtomicU64::new(0),
|
|
thumb_bytes: AtomicU64::new(0),
|
|
db,
|
|
};
|
|
cache.reconcile()?;
|
|
// After reconcile, so the totals describe the index that now describes
|
|
// the disk.
|
|
cache.reseed_totals();
|
|
cache.import_legacy_pins()?;
|
|
Ok(cache)
|
|
}
|
|
|
|
/// One-time migration of a pre-P5 `pins.json` into the DB `pins` table, then
|
|
/// delete the file so it never re-imports. Legacy pins were all whole-file
|
|
/// (non-recursive). Absent file → nothing to do.
|
|
fn import_legacy_pins(&self) -> Result<()> {
|
|
let Ok(bytes) = std::fs::read(&self.pins_path) else {
|
|
return Ok(());
|
|
};
|
|
if let Ok(file) = serde_json::from_slice::<PinFile>(&bytes) {
|
|
for (uid, pin) in file.pins {
|
|
self.db.pin_add(&uid, &pin.path, false)?;
|
|
}
|
|
}
|
|
let _ = std::fs::remove_file(&self.pins_path);
|
|
Ok(())
|
|
}
|
|
|
|
/// Rebuild the `cache_entries` LRU index from the on-disk cache. Called once
|
|
/// on open: the index is cleared, then every blob (`content_dir`) and block
|
|
/// (`block_dir`) file is re-registered with its on-disk size and last-modified
|
|
/// time as the initial LRU key. This makes the DB authoritative even after a
|
|
/// crash or an external file deletion, and picks up caches written by builds
|
|
/// predating the index. In-run accesses then refine the ordering.
|
|
fn reconcile(&self) -> Result<()> {
|
|
// Both directories are walked before anything is written: the index is
|
|
// replaced in a single transaction (see [`Db::cache_rebuild`]), so the
|
|
// rows have to be in hand first. Names are owned for the same reason —
|
|
// `DirEntry::file_name` does not outlive the iteration.
|
|
let mut names: Vec<(String, &str, u64, i64)> = Vec::new();
|
|
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;
|
|
};
|
|
for entry in rd.flatten() {
|
|
let name = entry.file_name();
|
|
let Some(name) = name.to_str() else { continue };
|
|
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 };
|
|
if !meta.is_file() {
|
|
continue;
|
|
}
|
|
let at = meta
|
|
.modified()
|
|
.ok()
|
|
.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
|
|
.map(|d| d.as_millis() as i64)
|
|
.unwrap_or(0);
|
|
names.push((name.to_string(), kind, meta.len(), at));
|
|
}
|
|
}
|
|
let entries: Vec<CacheEntryInput<'_>> = names
|
|
.iter()
|
|
.map(|(key, kind, size, at)| CacheEntryInput {
|
|
key,
|
|
kind,
|
|
size: *size,
|
|
last_accessed: *at,
|
|
})
|
|
.collect();
|
|
self.db.cache_rebuild(&entries)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Filesystem-safe, fixed-length cache key for a uid display string: the
|
|
/// SHA-256 of `s`, hex-encoded (64 chars). A plain hex of the uid would
|
|
/// *double* its length, and Proton's `volume~link` ids are long enough
|
|
/// (each half can be 60+ chars) that the doubled key overflows `NAME_MAX`
|
|
/// (255) — `store` then fails with ENAMETOOLONG. Hashing bounds the key at
|
|
/// 64 chars; reversibility is not needed and collisions are negligible.
|
|
fn key_str(s: &str) -> String {
|
|
let digest = Sha256::digest(s.as_bytes());
|
|
let mut out = String::with_capacity(digest.len() * 2);
|
|
for b in digest {
|
|
out.push(char::from_digit((b >> 4) as u32, 16).unwrap());
|
|
out.push(char::from_digit((b & 0xf) as u32, 16).unwrap());
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Filesystem-safe cache key for `uid`.
|
|
fn key(uid: &NodeUid) -> String {
|
|
Self::key_str(&uid.to_string())
|
|
}
|
|
|
|
fn blob_path(&self, uid: &NodeUid) -> PathBuf {
|
|
self.content_dir.join(Self::key(uid))
|
|
}
|
|
|
|
fn meta_path(&self, uid: &NodeUid) -> PathBuf {
|
|
self.content_dir.join(format!("{}.meta", Self::key(uid)))
|
|
}
|
|
|
|
/// The validated blob path for `uid`, or `None` if absent or stale against
|
|
/// `(mtime, size)`.
|
|
fn valid_blob(&self, uid: &NodeUid, mtime: i64, size: u64) -> Option<PathBuf> {
|
|
let want = Meta { mtime, size };
|
|
let meta: Meta = serde_json::from_slice(&std::fs::read(self.meta_path(uid)).ok()?).ok()?;
|
|
if meta != want {
|
|
return None;
|
|
}
|
|
let blob = self.blob_path(uid);
|
|
// Guard against a torn write: the blob must match the recorded size.
|
|
match std::fs::metadata(&blob) {
|
|
Ok(m) if m.len() == size => Some(blob),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Whether a fresh cached blob exists for `uid`.
|
|
pub fn is_cached(&self, uid: &NodeUid, mtime: i64, size: u64) -> bool {
|
|
self.valid_blob(uid, mtime, size).is_some()
|
|
}
|
|
|
|
/// Serve `len` bytes from `offset` out of the cached blob, or `None` on a
|
|
/// miss/stale entry. Reads only the requested window off disk.
|
|
pub fn read_range(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
size: u64,
|
|
offset: u64,
|
|
len: u64,
|
|
) -> Option<Vec<u8>> {
|
|
let blob = self.valid_blob(uid, mtime, size)?;
|
|
// Record the access for LRU in the cache index. Best effort — a failed
|
|
// update only makes eviction order slightly less accurate.
|
|
let _ = self.db.cache_accessed(&Self::key(uid), now_millis());
|
|
let mut f = std::fs::File::open(&blob).ok()?;
|
|
if offset >= size {
|
|
return Some(Vec::new());
|
|
}
|
|
let end = offset.saturating_add(len).min(size);
|
|
let want = (end - offset) as usize;
|
|
f.seek(SeekFrom::Start(offset)).ok()?;
|
|
let mut buf = vec![0u8; want];
|
|
f.read_exact(&mut buf).ok()?;
|
|
Some(buf)
|
|
}
|
|
|
|
/// Store `bytes` as the cached content for `uid`, tagged with `(mtime,
|
|
/// size)`. The blob is written to a temp file then renamed so a concurrent
|
|
/// reader never sees a partial blob; the meta tag is written last so a
|
|
/// crash mid-store fails validation rather than serving truncated data.
|
|
pub fn store(&self, uid: &NodeUid, mtime: i64, size: u64, bytes: &[u8]) -> Result<()> {
|
|
let blob = self.blob_path(uid);
|
|
// `with_extension` is safe *here* only because a blob path is a bare
|
|
// 64-char hex key with no `.`, so this appends rather than replaces.
|
|
// `store_thumbnail` cannot use the same idiom — see the note there.
|
|
let tmp = blob.with_extension("tmp");
|
|
std::fs::write(&tmp, bytes)?;
|
|
std::fs::rename(&tmp, &blob)?;
|
|
let meta = serde_json::to_vec(&Meta { mtime, size })?;
|
|
std::fs::write(self.meta_path(uid), meta)?;
|
|
// Register the blob in the LRU index before enforcing the budget so the
|
|
// newest entry is weighed (and ordered most-recent) like any other.
|
|
self.db
|
|
.cache_touch(&Self::key(uid), KIND_BLOB, bytes.len() as u64, now_millis())?;
|
|
self.blob_bytes
|
|
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
|
|
self.enforce_budget();
|
|
Ok(())
|
|
}
|
|
|
|
/// Adopt an existing on-disk file as `uid`'s cached content, without reading
|
|
/// it into memory.
|
|
///
|
|
/// [`Self::store`] takes the bytes as a slice, which means a caller holding a
|
|
/// file has to load all of it first — fine for a thumbnail, an OOM for a
|
|
/// pinned video. This takes the path instead and hardlinks it into place,
|
|
/// falling back to a copy when `src` lives on another filesystem.
|
|
///
|
|
/// The link makes the cache blob a second name for the same inode, so `src`
|
|
/// must not be rewritten in place afterwards — unlinking it (the staging
|
|
/// case, which is what this exists for) is fine and leaves the blob intact.
|
|
pub fn store_file(&self, uid: &NodeUid, mtime: i64, size: u64, src: &Path) -> Result<()> {
|
|
let blob = self.blob_path(uid);
|
|
let tmp = blob.with_extension("tmp");
|
|
// A stale tmp from an interrupted run would fail the link with EEXIST.
|
|
let _ = std::fs::remove_file(&tmp);
|
|
if std::fs::hard_link(src, &tmp).is_err() {
|
|
std::fs::copy(src, &tmp)?;
|
|
}
|
|
let bytes = std::fs::metadata(&tmp)?.len();
|
|
std::fs::rename(&tmp, &blob)?;
|
|
let meta = serde_json::to_vec(&Meta { mtime, size })?;
|
|
std::fs::write(self.meta_path(uid), meta)?;
|
|
self.db
|
|
.cache_touch(&Self::key(uid), KIND_BLOB, bytes, now_millis())?;
|
|
self.blob_bytes.fetch_add(bytes, Ordering::Relaxed);
|
|
self.enforce_budget();
|
|
Ok(())
|
|
}
|
|
|
|
/// Cache-index key for block `idx` of `uid` — the block file's name, which
|
|
/// is also its `cache_entries.cache_key`.
|
|
fn block_key(&self, uid: &NodeUid, idx: u64) -> String {
|
|
format!("{}.b{idx}", Self::key(uid))
|
|
}
|
|
|
|
fn block_blob(&self, uid: &NodeUid, idx: u64) -> PathBuf {
|
|
self.block_dir.join(self.block_key(uid, idx))
|
|
}
|
|
|
|
fn block_meta(&self, uid: &NodeUid, idx: u64) -> PathBuf {
|
|
self.block_dir
|
|
.join(format!("{}.b{idx}.meta", Self::key(uid)))
|
|
}
|
|
|
|
/// Serve cached block `idx` (a [`BLOCK_SIZE`]-aligned chunk) of `uid`, or
|
|
/// `None` on miss/stale. Validated against `(mtime, size)` like a whole-file
|
|
/// blob, so a new revision (which bumps the mtime) is detected. Bumps the
|
|
/// block's mtime for LRU, best effort.
|
|
pub fn cached_block(&self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option<Vec<u8>> {
|
|
let want = Meta { mtime, size };
|
|
let meta: Meta =
|
|
serde_json::from_slice(&std::fs::read(self.block_meta(uid, idx)).ok()?).ok()?;
|
|
if meta != want {
|
|
return None;
|
|
}
|
|
let mut f = std::fs::File::open(self.block_blob(uid, idx)).ok()?;
|
|
// LRU touch in the cache index (best effort).
|
|
let _ = self
|
|
.db
|
|
.cache_accessed(&self.block_key(uid, idx), now_millis());
|
|
let mut buf = Vec::new();
|
|
f.read_to_end(&mut buf).ok()?;
|
|
Some(buf)
|
|
}
|
|
|
|
/// Store `bytes` as cached block `idx` of `uid`, tagged `(mtime, size)`.
|
|
/// Temp-file-then-rename like [`store`](Self::store); meta written last so a
|
|
/// crash mid-store fails validation. Enforces the block-cache LRU budget.
|
|
pub fn store_block(
|
|
&self,
|
|
uid: &NodeUid,
|
|
mtime: i64,
|
|
size: u64,
|
|
idx: u64,
|
|
bytes: &[u8],
|
|
) -> Result<()> {
|
|
let blob = self.block_blob(uid, idx);
|
|
let tmp = self
|
|
.block_dir
|
|
.join(format!("{}.b{idx}.tmp", Self::key(uid)));
|
|
std::fs::write(&tmp, bytes)?;
|
|
std::fs::rename(&tmp, &blob)?;
|
|
std::fs::write(
|
|
self.block_meta(uid, idx),
|
|
serde_json::to_vec(&Meta { mtime, size })?,
|
|
)?;
|
|
self.db.cache_touch(
|
|
&self.block_key(uid, idx),
|
|
KIND_BLOCK,
|
|
bytes.len() as u64,
|
|
now_millis(),
|
|
)?;
|
|
self.block_bytes
|
|
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
|
|
self.enforce_block_budget();
|
|
Ok(())
|
|
}
|
|
|
|
/// Re-read every pool total from the cache index, which is authoritative.
|
|
fn reseed_totals(&self) {
|
|
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 {
|
|
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.
|
|
fn enforce_block_budget(&self) {
|
|
self.enforce_pool(KIND_BLOCK, &self.block_dir, None);
|
|
}
|
|
|
|
/// Evict least-recently-used entries of one pool until it fits the cap.
|
|
///
|
|
/// Runs on every store, so the under-budget case is the one that matters:
|
|
/// it is an atomic load and nothing else. Only once the running total says
|
|
/// we are over does this touch the database, and then it works in bounded
|
|
/// batches rather than reading the whole index — a cache that is 10 GB over
|
|
/// budget should not materialize every row to drop the oldest few.
|
|
///
|
|
/// `exempt` names keys that count toward the total but are never victims
|
|
/// (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.pool_cap(kind);
|
|
if cap == 0 {
|
|
return;
|
|
}
|
|
let total = self.pool(kind);
|
|
if total.load(Ordering::Relaxed) <= cap {
|
|
return; // the common case: one atomic load, no query
|
|
}
|
|
|
|
// Over budget by the running count — now do the accurate work.
|
|
let mut running = match self.db.cache_total_bytes(kind) {
|
|
Ok(n) => n,
|
|
Err(_) => return,
|
|
};
|
|
let mut skipped = 0usize;
|
|
while running > cap {
|
|
// Fetch a batch past whatever we have already skipped as exempt.
|
|
let want = skipped + EVICT_BATCH;
|
|
let Ok(batch) = self.db.cache_eviction_candidates(kind, want) else {
|
|
break;
|
|
};
|
|
if batch.len() <= skipped {
|
|
break; // nothing new to consider
|
|
}
|
|
let mut evicted_any = false;
|
|
for (key, size) in batch.into_iter().skip(skipped) {
|
|
if running <= cap {
|
|
break;
|
|
}
|
|
if exempt.is_some_and(|set| set.contains(&key)) {
|
|
skipped += 1; // counts toward the total, never a victim
|
|
continue;
|
|
}
|
|
let _ = std::fs::remove_file(dir.join(&key));
|
|
let _ = std::fs::remove_file(dir.join(format!("{key}.meta")));
|
|
let _ = self.db.cache_remove(&key);
|
|
running = running.saturating_sub(size);
|
|
evicted_any = true;
|
|
}
|
|
if !evicted_any {
|
|
break; // every remaining candidate is exempt
|
|
}
|
|
}
|
|
// The index is the truth; realign the fast path with it.
|
|
self.reseed_totals();
|
|
}
|
|
|
|
/// Create a fresh, empty read-write scratch file for a disk-backed write
|
|
/// handle. Returns the open file and its path (for cleanup on release).
|
|
pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> {
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
static N: AtomicU64 = AtomicU64::new(0);
|
|
let path = self.scratch_dir.join(format!(
|
|
"w-{}-{}",
|
|
std::process::id(),
|
|
N.fetch_add(1, Ordering::Relaxed)
|
|
));
|
|
let file = std::fs::OpenOptions::new()
|
|
.create(true)
|
|
.truncate(true)
|
|
.read(true)
|
|
.write(true)
|
|
.open(&path)?;
|
|
Ok((file, path))
|
|
}
|
|
|
|
/// The sidecar path for a scratch file, holding the [`StagedWrite`] that
|
|
/// describes it. Written by `fsync` and removed when the write is released;
|
|
/// its presence at open is what marks a scratch file as durable rather than
|
|
/// as a crashed run's rubbish.
|
|
///
|
|
/// Safe as `with_extension` only because scratch names are dot-free (see
|
|
/// [`Self::create_scratch`]), so this appends rather than replaces.
|
|
pub fn scratch_sidecar(scratch: &Path) -> PathBuf {
|
|
scratch.with_extension("json")
|
|
}
|
|
|
|
/// Record that the write buffered in `scratch` is durable, so a crash before
|
|
/// `close(2)` does not lose it.
|
|
///
|
|
/// Called from `fsync(2)`, whose contract is that the bytes survive a crash.
|
|
/// The scratch file itself does survive — it is a real file — but nothing
|
|
/// else knows it holds anything worth keeping, and open() clears the scratch
|
|
/// directory. The sidecar is that knowledge: uid, logical length, and which
|
|
/// ranges are authored rather than holes, which is exactly what recovery
|
|
/// needs to hand the blob to the drain.
|
|
///
|
|
/// Written whole to a temp name and renamed, so a crash mid-write leaves the
|
|
/// previous sidecar rather than a truncated one. The caller is expected to
|
|
/// have synced `scratch` first: a sidecar promising bytes that never reached
|
|
/// the disk is worse than no sidecar.
|
|
pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> {
|
|
let side = Self::scratch_sidecar(scratch);
|
|
let tmp = side.with_extension("json.tmp");
|
|
std::fs::write(&tmp, serde_json::to_vec(meta)?)?;
|
|
std::fs::rename(&tmp, &side)?;
|
|
// The rename is what publishes the sidecar; fsync the directory so the
|
|
// rename itself survives, not just the bytes it points at.
|
|
if let Some(dir) = side.parent()
|
|
&& let Ok(d) = std::fs::File::open(dir)
|
|
{
|
|
let _ = d.sync_all();
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Drop a scratch file's durability sidecar, once the write it described has
|
|
/// been staged (or turned out not to be dirty at all). Best effort: a
|
|
/// leftover sidecar whose blob is gone is ignored by recovery anyway.
|
|
pub fn clear_scratch_durable(&self, scratch: &Path) {
|
|
let _ = std::fs::remove_file(Self::scratch_sidecar(scratch));
|
|
}
|
|
|
|
/// Move every scratch file that carries a readable sidecar into `recovery`,
|
|
/// before the caller empties the scratch directory.
|
|
///
|
|
/// Blob first, then sidecar: a blob in recovery without its sidecar is
|
|
/// dropped on the next open (indistinguishable from rubbish), whereas a
|
|
/// sidecar whose blob never arrived would describe nothing. Neither order is
|
|
/// lossless if we crash mid-rescue, but this one fails toward "unrecovered
|
|
/// bytes on disk" rather than "op queued against missing content".
|
|
fn rescue_scratch(scratch_dir: &Path, recovery_dir: &Path) {
|
|
let Ok(entries) = std::fs::read_dir(scratch_dir) else {
|
|
return;
|
|
};
|
|
for entry in entries.flatten() {
|
|
let blob = entry.path();
|
|
if blob.extension().is_some_and(|e| e == "json") {
|
|
continue;
|
|
}
|
|
let side = Self::scratch_sidecar(&blob);
|
|
// Parsed, not merely present: an unreadable sidecar cannot drive a
|
|
// recovery, so the blob is rubbish like any other leftover.
|
|
if std::fs::read(&side)
|
|
.ok()
|
|
.and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
|
|
.is_none()
|
|
{
|
|
continue;
|
|
}
|
|
let Some(name) = blob.file_name() else {
|
|
continue;
|
|
};
|
|
if std::fs::rename(&blob, recovery_dir.join(name)).is_err() {
|
|
continue;
|
|
}
|
|
let Some(side_name) = side.file_name() else {
|
|
continue;
|
|
};
|
|
let _ = std::fs::rename(&side, recovery_dir.join(side_name));
|
|
}
|
|
}
|
|
|
|
/// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the
|
|
/// daemon to hand to the upload queue. A blob whose sidecar did not survive
|
|
/// the rescue is deleted here — nothing can address it.
|
|
pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> {
|
|
let Ok(entries) = std::fs::read_dir(&self.recovery_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;
|
|
}
|
|
match std::fs::read(Self::scratch_sidecar(&blob))
|
|
.ok()
|
|
.and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
|
|
{
|
|
Some(meta) => out.push((blob, meta)),
|
|
None => {
|
|
let _ = std::fs::remove_file(&blob);
|
|
}
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Retire a recovered write. `stage_write` has already moved the blob out,
|
|
/// so this is the sidecar left behind — and the blob too if the caller could
|
|
/// not use it and said so by leaving it in place.
|
|
pub fn discard_recovered(&self, blob: &Path) {
|
|
let _ = std::fs::remove_file(Self::scratch_sidecar(blob));
|
|
let _ = std::fs::remove_file(blob);
|
|
}
|
|
|
|
/// Move a released scratch file into the staging directory with a
|
|
/// [`StagedWrite`] sidecar, and return where it landed.
|
|
///
|
|
/// This is what makes a write survive its upload: the caller is releasing a
|
|
/// write handle and would otherwise delete the file, so until the bytes are
|
|
/// on the remote, staging holds the only copy. Every dirty handle goes
|
|
/// through here — the upload is a queued op performed later (offline.md
|
|
/// Phase 3), and a staged file is also what a human can recover from if the
|
|
/// queue never drains.
|
|
///
|
|
/// Falls back to a copy when the rename crosses a filesystem boundary; a
|
|
/// failure here means we could not save the bytes at all, so it is reported
|
|
/// rather than swallowed.
|
|
pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result<PathBuf> {
|
|
static N: AtomicU64 = AtomicU64::new(0);
|
|
// The uid goes in the name so a staged file can be tied back to its node
|
|
// without a database; `/` in a uid would otherwise open a path.
|
|
let safe_uid = meta.uid.replace(['/', '\\'], "_");
|
|
let path = self.staging_dir.join(format!(
|
|
"{}-{}-{}",
|
|
safe_uid,
|
|
now_millis(),
|
|
N.fetch_add(1, Ordering::Relaxed)
|
|
));
|
|
if let Err(e) = std::fs::rename(scratch, &path) {
|
|
// Cross-device: rename cannot move it, so copy and drop the original.
|
|
let _ = e;
|
|
std::fs::copy(scratch, &path)?;
|
|
let _ = std::fs::remove_file(scratch);
|
|
}
|
|
// Sidecar last: a staged file without one is still recoverable bytes,
|
|
// while a sidecar without a file would describe nothing.
|
|
std::fs::write(
|
|
path.with_extension("json"),
|
|
serde_json::to_vec_pretty(meta)?,
|
|
)?;
|
|
Ok(path)
|
|
}
|
|
|
|
/// Drop a staged blob and its sidecar, once its bytes are safely on the
|
|
/// remote (or the op that owned them has been superseded). Best effort:
|
|
/// leftovers cost disk, while failing here would strand a drained op.
|
|
pub fn discard_staged(&self, path: &Path) {
|
|
let _ = std::fs::remove_file(path);
|
|
let _ = std::fs::remove_file(path.with_extension("json"));
|
|
}
|
|
|
|
/// Remove any cached blob + meta for `uid`, including its thumbnails (best
|
|
/// effort; absence is fine). Thumbnails are tied to the revision, so a
|
|
/// content eviction must drop them too.
|
|
pub fn evict(&self, uid: &NodeUid) {
|
|
let _ = std::fs::remove_file(self.blob_path(uid));
|
|
let _ = std::fs::remove_file(self.meta_path(uid));
|
|
for ttype in [1, 2] {
|
|
let _ = std::fs::remove_file(self.thumb_blob(uid, ttype));
|
|
let _ = std::fs::remove_file(self.thumb_meta(uid, ttype));
|
|
}
|
|
// Drop every cached block (and its meta/tmp) for this uid. The block
|
|
// dir is still scanned here because eviction targets one uid's files by
|
|
// name prefix, not by LRU order; the index rows are removed in one shot
|
|
// by `cache_remove_all` below.
|
|
let prefix = format!("{}.b", Self::key(uid));
|
|
if let Ok(rd) = std::fs::read_dir(&self.block_dir) {
|
|
for entry in rd.flatten() {
|
|
if entry
|
|
.file_name()
|
|
.to_str()
|
|
.is_some_and(|n| n.starts_with(&prefix))
|
|
{
|
|
let _ = std::fs::remove_file(entry.path());
|
|
}
|
|
}
|
|
}
|
|
// Forget the blob and all of this uid's block rows in the LRU index.
|
|
let _ = self.db.cache_remove_all(&Self::key(uid));
|
|
// Rows just left the index; the running totals must follow.
|
|
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(self.thumb_key(uid, ttype))
|
|
}
|
|
|
|
fn thumb_meta(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
|
|
self.thumb_dir
|
|
.join(format!("{}.t{ttype}.meta", Self::key(uid)))
|
|
}
|
|
|
|
/// Serve a cached thumbnail of `ttype` for `uid`, or `None` on miss/stale.
|
|
/// Validated by node `mtime` alone: a new revision (which is what changes the
|
|
/// thumbnail) bumps the mtime, so a stale thumbnail is detected. Thumbnails
|
|
/// are small, so the whole blob is read at once.
|
|
pub fn read_thumbnail(&self, uid: &NodeUid, ttype: i32, mtime: i64) -> Option<Vec<u8>> {
|
|
let recorded: i64 =
|
|
serde_json::from_slice(&std::fs::read(self.thumb_meta(uid, ttype)).ok()?).ok()?;
|
|
if recorded != mtime {
|
|
return None;
|
|
}
|
|
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
|
|
/// `(mtime, size)`; `None` on miss/stale. Lets a front-end open the blob
|
|
/// directly instead of streaming its bytes back over the control socket.
|
|
pub fn cached_content_path(&self, uid: &NodeUid, mtime: i64, size: u64) -> Option<PathBuf> {
|
|
self.valid_blob(uid, mtime, size)
|
|
}
|
|
|
|
/// On-disk path where `uid`'s content blob lives once stored (the file may
|
|
/// not exist yet, or may be stale — pair with [`store`](Self::store)).
|
|
pub fn content_path(&self, uid: &NodeUid) -> PathBuf {
|
|
self.blob_path(uid)
|
|
}
|
|
|
|
/// On-disk path of the cached `ttype` thumbnail for `uid`, fresh against
|
|
/// `tag` (the validity tag last passed to [`store_thumbnail`](Self::store_thumbnail));
|
|
/// `None` on miss/stale. For Drive files the tag is the node mtime; for
|
|
/// photos it is the capture time.
|
|
pub fn cached_thumbnail_path(&self, uid: &NodeUid, ttype: i32, tag: i64) -> Option<PathBuf> {
|
|
let recorded: i64 =
|
|
serde_json::from_slice(&std::fs::read(self.thumb_meta(uid, ttype)).ok()?).ok()?;
|
|
if recorded != tag {
|
|
return None;
|
|
}
|
|
let blob = self.thumb_blob(uid, ttype);
|
|
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.
|
|
pub fn thumbnail_path(&self, uid: &NodeUid, ttype: i32) -> PathBuf {
|
|
self.thumb_blob(uid, ttype)
|
|
}
|
|
|
|
/// Cache `bytes` as the `ttype` thumbnail for `uid`, tagged with `mtime`.
|
|
/// Blob written to a temp file then renamed; the meta tag is written last so
|
|
/// a crash mid-store fails validation rather than serving a torn thumbnail.
|
|
///
|
|
/// The temp name carries `ttype`, and must. `Path::with_extension` would
|
|
/// *replace* the `.t{ttype}` suffix rather than append to it, giving every
|
|
/// thumbnail type of one node the same staging file — so a type-1 and a
|
|
/// type-2 store racing (the gallery caches type 1 from a control-socket
|
|
/// thread while `getxattr` caches type 2 on the FUSE dispatch loop) would
|
|
/// publish one type's bytes under the other's name. See
|
|
/// `concurrent_thumbnail_types_do_not_share_a_temp_file`.
|
|
pub fn store_thumbnail(
|
|
&self,
|
|
uid: &NodeUid,
|
|
ttype: i32,
|
|
mtime: i64,
|
|
bytes: &[u8],
|
|
) -> Result<()> {
|
|
let blob = self.thumb_blob(uid, ttype);
|
|
let tmp = self
|
|
.thumb_dir
|
|
.join(format!("{}.t{ttype}.tmp", Self::key(uid)));
|
|
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.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
|
|
// single store.
|
|
return;
|
|
}
|
|
// Pinned blobs (by cache key) are exempt from eviction. Resolves direct
|
|
// pins plus every descendant of a recursively-pinned folder, hashed into
|
|
// cache keys to match on-disk filenames.
|
|
let pinned: HashSet<String> = self
|
|
.db
|
|
.pinned_uids()
|
|
.unwrap_or_default()
|
|
.iter()
|
|
.map(|uid| Self::key_str(uid))
|
|
.collect();
|
|
let content_dir = self.content_dir.clone();
|
|
self.enforce_pool(KIND_BLOB, &content_dir, Some(&pinned));
|
|
}
|
|
|
|
/// Whether `uid` is pinned — directly, or because an ancestor folder is
|
|
/// pinned recursively (resolved in the DB against the node tree). Independent
|
|
/// of whether its blob is currently cached. Best effort: a DB error reads as
|
|
/// "not pinned" rather than failing the calling FUSE op.
|
|
pub fn is_pinned(&self, uid: &NodeUid) -> bool {
|
|
self.db.is_pinned(&uid.to_string()).unwrap_or(false)
|
|
}
|
|
|
|
/// Record `uid` as pinned under `path`. `recursive` marks a folder pin whose
|
|
/// whole subtree is kept. The caller is responsible for having cached the
|
|
/// content (a file via [`store`](Self::store); a folder's descendants by the
|
|
/// daemon walking the subtree).
|
|
pub fn add_pin(&self, uid: &NodeUid, path: &Path, recursive: bool) -> Result<()> {
|
|
self.db
|
|
.pin_add(&uid.to_string(), &path.display().to_string(), recursive)
|
|
}
|
|
|
|
/// Drop `uid` from the pin registry and evict its blob. No-op if not pinned.
|
|
/// A recursively-pinned folder's descendants are evicted by the daemon
|
|
/// (`Core::unpin`), which knows the node tree.
|
|
pub fn remove_pin(&self, uid: &NodeUid) -> Result<()> {
|
|
self.db.pin_remove(&uid.to_string())?;
|
|
self.evict(uid);
|
|
Ok(())
|
|
}
|
|
|
|
/// All pins (files and recursive folders), ordered by uid.
|
|
pub fn list_pins(&self) -> Vec<Pin> {
|
|
self.db
|
|
.pin_list()
|
|
.unwrap_or_default()
|
|
.into_iter()
|
|
.map(|(uid, path, recursive)| Pin {
|
|
uid,
|
|
path,
|
|
recursive,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// 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 {
|
|
[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). 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()
|
|
}
|
|
|
|
/// The current soft byte cap. A plain atomic load; the value can change
|
|
/// under [`set_budget`](Self::set_budget) while the daemon runs.
|
|
fn cap(&self) -> u64 {
|
|
self.max_bytes.load(Ordering::Relaxed)
|
|
}
|
|
|
|
/// Retune the soft byte cap (`0` = unlimited) at runtime and immediately
|
|
/// enforce it: blobs and blocks are LRU-evicted back under the new cap (a
|
|
/// lower cap frees space now; a higher one is a no-op until the next store).
|
|
/// Called by the daemon when the Settings page changes the cache budget.
|
|
pub fn set_budget(&self, bytes: u64) {
|
|
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,
|
|
/// keeping pinned files intact. Pinned files keep their whole-file blobs
|
|
/// (and are never served from blocks), so dropping all blocks is safe.
|
|
/// Returns the number of bytes freed, for a user-facing confirmation.
|
|
pub fn clear_unpinned(&self) -> u64 {
|
|
// Pinned blobs are exempt; resolve them to cache keys to match filenames.
|
|
let pinned: HashSet<String> = self
|
|
.db
|
|
.pinned_uids()
|
|
.unwrap_or_default()
|
|
.iter()
|
|
.map(|uid| Self::key_str(uid))
|
|
.collect();
|
|
let mut freed = 0u64;
|
|
// Unpinned whole-file blobs.
|
|
if let Ok(entries) = self.db.cache_entries_by_kind(KIND_BLOB) {
|
|
for (key, size) in entries {
|
|
if pinned.contains(&key) {
|
|
continue;
|
|
}
|
|
let _ = std::fs::remove_file(self.content_dir.join(&key));
|
|
let _ = std::fs::remove_file(self.content_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();
|
|
freed
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
|
|
|
|
fn uid(link: &str) -> NodeUid {
|
|
NodeUid::new(VolumeId::from("vol"), LinkId::from(link))
|
|
}
|
|
|
|
/// A unique temp directory removed on drop; avoids a dev-dependency.
|
|
struct TempDir(PathBuf);
|
|
impl TempDir {
|
|
fn new() -> Self {
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
static N: AtomicU64 = AtomicU64::new(0);
|
|
let p = std::env::temp_dir().join(format!(
|
|
"pdfs-cache-{}-{}",
|
|
std::process::id(),
|
|
N.fetch_add(1, Ordering::Relaxed)
|
|
));
|
|
std::fs::create_dir_all(&p).unwrap();
|
|
TempDir(p)
|
|
}
|
|
fn path(&self) -> &Path {
|
|
&self.0
|
|
}
|
|
}
|
|
impl Drop for TempDir {
|
|
fn drop(&mut self) {
|
|
let _ = std::fs::remove_dir_all(&self.0);
|
|
}
|
|
}
|
|
|
|
fn cache() -> (ContentCache, TempDir) {
|
|
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());
|
|
let c = ContentCache::open(
|
|
dir.path().join("content"),
|
|
dir.path().join("pins.json"),
|
|
max_bytes,
|
|
db,
|
|
)
|
|
.unwrap();
|
|
(c, dir)
|
|
}
|
|
|
|
#[test]
|
|
fn store_then_read_range() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
let data = b"hello world";
|
|
c.store(&u, 100, data.len() as u64, data).unwrap();
|
|
assert!(c.is_cached(&u, 100, data.len() as u64));
|
|
assert_eq!(
|
|
c.read_range(&u, 100, data.len() as u64, 6, 5).unwrap(),
|
|
b"world"
|
|
);
|
|
// Offset past EOF yields an empty slice, not a miss.
|
|
assert_eq!(
|
|
c.read_range(&u, 100, data.len() as u64, 100, 5).unwrap(),
|
|
b""
|
|
);
|
|
}
|
|
|
|
/// The drain promotes a staged upload into the cache by path and then
|
|
/// unlinks the staging name. The cached copy has to survive that, and the
|
|
/// promote has to charge the LRU index like [`ContentCache::store`] does.
|
|
#[test]
|
|
fn store_file_adopts_blob_and_survives_source_unlink() {
|
|
let (c, d) = cache();
|
|
let u = uid("a");
|
|
let data = b"staged contents";
|
|
let src = d.path().join("staged-blob");
|
|
std::fs::write(&src, data).unwrap();
|
|
|
|
c.store_file(&u, 100, data.len() as u64, &src).unwrap();
|
|
std::fs::remove_file(&src).unwrap();
|
|
|
|
assert!(c.is_cached(&u, 100, data.len() as u64));
|
|
assert_eq!(
|
|
c.read_range(&u, 100, data.len() as u64, 7, 8).unwrap(),
|
|
b"contents"
|
|
);
|
|
assert_eq!(c.usage(), data.len() as u64);
|
|
}
|
|
|
|
/// A cross-filesystem staging dir cannot be hardlinked from; the copy
|
|
/// blob and its metadata must both be replaced wholesale — a stale `.meta`
|
|
/// left beside new content would validate a read of the wrong bytes. The
|
|
/// link goes to a temp name and is renamed over the blob for exactly this,
|
|
/// so re-storing over an existing entry is the case worth pinning.
|
|
#[test]
|
|
fn store_file_replaces_an_existing_blob() {
|
|
let (c, d) = cache();
|
|
let u = uid("a");
|
|
c.store(&u, 100, 3, b"old").unwrap();
|
|
|
|
let src = d.path().join("new-blob");
|
|
std::fs::write(&src, b"newer").unwrap();
|
|
c.store_file(&u, 200, 5, &src).unwrap();
|
|
|
|
assert!(!c.is_cached(&u, 100, 3), "stale meta must not validate");
|
|
assert_eq!(c.read_range(&u, 200, 5, 0, 5).unwrap(), b"newer");
|
|
}
|
|
|
|
#[test]
|
|
fn reconcile_rebuilds_index_from_disk() {
|
|
let dir = TempDir::new();
|
|
let content = dir.path().join("content");
|
|
let pins = dir.path().join("pins.json");
|
|
|
|
// First session: unlimited cap, store three blobs onto disk.
|
|
let db1 = Arc::new(Db::open_in_memory().unwrap());
|
|
{
|
|
let c = ContentCache::open(content.clone(), pins.clone(), 0, db1).unwrap();
|
|
c.store(&uid("a"), 1, 4, b"aaaa").unwrap();
|
|
c.store(&uid("b"), 1, 4, b"bbbb").unwrap();
|
|
c.store(&uid("c"), 1, 4, b"cccc").unwrap();
|
|
}
|
|
|
|
// Second session with a *fresh* index (as if predating P4, or after a
|
|
// crash) but the same on-disk cache: reconcile repopulates from disk.
|
|
let db2 = Arc::new(Db::open_in_memory().unwrap());
|
|
assert!(db2.cache_entries_by_kind("blob").unwrap().is_empty());
|
|
let c = ContentCache::open(content, pins, 8, db2.clone()).unwrap();
|
|
assert_eq!(db2.cache_entries_by_kind("blob").unwrap().len(), 3);
|
|
|
|
// With the index rebuilt, a further store enforces the budget against the
|
|
// reconciled entries instead of leaking them.
|
|
c.store(&uid("d"), 1, 4, b"dddd").unwrap();
|
|
let total: u64 = db2
|
|
.cache_entries_by_kind("blob")
|
|
.unwrap()
|
|
.iter()
|
|
.map(|(_, s)| s)
|
|
.sum();
|
|
assert!(total <= 8, "budget enforced after reconcile, total={total}");
|
|
}
|
|
|
|
#[test]
|
|
fn stale_metadata_is_a_miss() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
c.store(&u, 100, 3, b"abc").unwrap();
|
|
// A newer mtime (new revision) invalidates the cached blob.
|
|
assert!(!c.is_cached(&u, 101, 3));
|
|
assert!(c.read_range(&u, 101, 3, 0, 3).is_none());
|
|
// A size mismatch also invalidates it.
|
|
assert!(c.read_range(&u, 100, 4, 0, 3).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_pins_json_imported_then_removed() {
|
|
let dir = TempDir::new();
|
|
let pins = dir.path().join("pins.json");
|
|
// Hand-write a pre-P5 pins.json (file pins only, no `recursive` field).
|
|
std::fs::write(
|
|
&pins,
|
|
r#"{"pins":{"vol~a":{"uid":"vol~a","path":"docs/a.txt"}}}"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let db = Arc::new(Db::open_in_memory().unwrap());
|
|
let c = ContentCache::open(dir.path().join("content"), pins.clone(), 0, db).unwrap();
|
|
|
|
// Imported into the DB as a non-recursive pin...
|
|
assert!(c.is_pinned(&uid("a")));
|
|
let listed = c.list_pins();
|
|
assert_eq!(listed.len(), 1);
|
|
assert_eq!(listed[0].path, "docs/a.txt");
|
|
assert!(!listed[0].recursive);
|
|
// ...and the legacy file is gone so it never re-imports.
|
|
assert!(!pins.exists());
|
|
}
|
|
|
|
#[test]
|
|
fn pin_lifecycle_evicts_on_unpin() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
c.store(&u, 100, 3, b"abc").unwrap();
|
|
c.add_pin(&u, Path::new("docs/a.txt"), false).unwrap();
|
|
assert!(c.is_pinned(&u));
|
|
assert_eq!(c.list_pins().len(), 1);
|
|
assert_eq!(c.list_pins()[0].path, "docs/a.txt");
|
|
|
|
c.remove_pin(&u).unwrap();
|
|
assert!(!c.is_pinned(&u));
|
|
assert!(c.list_pins().is_empty());
|
|
// Unpin evicts the cached content.
|
|
assert!(!c.is_cached(&u, 100, 3));
|
|
}
|
|
|
|
/// The under-budget path — which is every store until the cache fills — must
|
|
/// not touch the database at all. It is on the cold-read hot path: one call
|
|
/// per cached 4 MiB block, under the connection lock every FUSE metadata
|
|
/// call also needs.
|
|
#[test]
|
|
fn budget_check_is_free_when_under_budget() {
|
|
use std::time::Instant;
|
|
|
|
// Cap far above what we store, so every store stays under budget.
|
|
let (c, _d) = cache_capped(1 << 30);
|
|
let payload = vec![0u8; 4096];
|
|
|
|
// Cost of a store against a near-empty index...
|
|
let t = Instant::now();
|
|
for i in 0..500u64 {
|
|
c.store_block(&uid("a"), 1, 1 << 30, i, &payload).unwrap();
|
|
}
|
|
let empty = t.elapsed() / 500;
|
|
|
|
// ...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!(
|
|
full < empty * 3,
|
|
"under-budget store should not scale with the index: \
|
|
{empty:?} at ~0 entries vs {full:?} at 2500"
|
|
);
|
|
}
|
|
|
|
/// The running totals are a fast path, not a second source of truth: after
|
|
/// an eviction pass they are realigned with the index, so repeated
|
|
/// over-budget stores cannot drift them.
|
|
#[test]
|
|
fn running_totals_stay_aligned_with_the_index() {
|
|
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));
|
|
}
|
|
let indexed: u64 =
|
|
c.db.cache_entries_by_kind("block")
|
|
.unwrap()
|
|
.iter()
|
|
.map(|(_, s)| s)
|
|
.sum();
|
|
assert_eq!(
|
|
c.block_bytes.load(Ordering::Relaxed),
|
|
indexed,
|
|
"running total matches the index after repeated eviction"
|
|
);
|
|
assert!(indexed <= 8, "and the cap is actually held: {indexed}");
|
|
}
|
|
|
|
#[test]
|
|
fn budget_evicts_lru_unpinned() {
|
|
// 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();
|
|
// Distinct mtimes so LRU order is deterministic (oldest = a).
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&b, 1, 4, b"bbbb").unwrap();
|
|
// Touch `a` so `b` becomes the least-recently-used.
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
assert!(c.read_range(&a, 1, 4, 0, 4).is_some());
|
|
|
|
// Third store pushes over budget → evicts LRU unpinned (`b`).
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&d, 1, 4, b"dddd").unwrap();
|
|
assert!(c.is_cached(&a, 1, 4), "recently-read blob survives");
|
|
assert!(!c.is_cached(&b, 1, 4), "LRU blob evicted");
|
|
assert!(c.is_cached(&d, 1, 4), "newest blob survives");
|
|
}
|
|
|
|
#[test]
|
|
fn budget_never_evicts_pinned() {
|
|
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();
|
|
c.add_pin(&a, Path::new("a"), false).unwrap();
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&b, 1, 4, b"bbbb").unwrap();
|
|
// Over budget: `a` is pinned (oldest) so `b` must be the victim.
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&d, 1, 4, b"dddd").unwrap();
|
|
assert!(c.is_cached(&a, 1, 4), "pinned blob never evicted");
|
|
assert!(!c.is_cached(&b, 1, 4), "unpinned LRU blob evicted");
|
|
assert!(c.is_cached(&d, 1, 4));
|
|
}
|
|
|
|
#[test]
|
|
fn clear_unpinned_keeps_pinned() {
|
|
let (c, _d) = cache();
|
|
let (a, b) = (uid("a"), uid("b"));
|
|
c.store(&a, 1, 4, b"aaaa").unwrap();
|
|
c.add_pin(&a, Path::new("a"), false).unwrap();
|
|
c.store(&b, 1, 4, b"bbbb").unwrap();
|
|
// A stray on-demand block of the unpinned file is purged too.
|
|
c.store_block(&b, 1, 8, 0, b"bbbb").unwrap();
|
|
|
|
let freed = c.clear_unpinned();
|
|
assert_eq!(freed, 8, "one 4-byte blob + one 4-byte block freed");
|
|
assert!(c.is_cached(&a, 1, 4), "pinned blob survives purge");
|
|
assert!(!c.is_cached(&b, 1, 4), "unpinned blob purged");
|
|
assert!(c.cached_block(&b, 1, 8, 0).is_none(), "block purged");
|
|
}
|
|
|
|
#[test]
|
|
fn set_budget_evicts_immediately() {
|
|
// Start unlimited so three blobs all fit, then tighten the cap.
|
|
let (c, _d) = cache();
|
|
let (a, b, d) = (uid("a"), uid("b"), uid("d"));
|
|
c.store(&a, 1, 4, b"aaaa").unwrap();
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&b, 1, 4, b"bbbb").unwrap();
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store(&d, 1, 4, b"dddd").unwrap();
|
|
|
|
// 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));
|
|
// `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
|
|
/// a staging file. They are stored from different subsystems — the Photos
|
|
/// gallery caches type 1 from a control-socket thread while `getxattr`
|
|
/// caches type 2 on the FUSE dispatch loop — so the two genuinely overlap.
|
|
///
|
|
/// The failure is silent: the loser's rename either publishes the *other*
|
|
/// type's bytes under this type's name, or vanishes with ENOENT. Both are
|
|
/// reported here, because a fix that only removed the error would leave the
|
|
/// corruption.
|
|
#[test]
|
|
fn concurrent_thumbnail_types_do_not_share_a_temp_file() {
|
|
use std::sync::Arc as StdArc;
|
|
|
|
let (c, _d) = cache();
|
|
let c = StdArc::new(c);
|
|
let u = uid("a");
|
|
// Big enough that the write is not a single atomic-looking syscall, and
|
|
// distinguishable by content and length.
|
|
let one = vec![0xA1u8; 96 * 1024];
|
|
let two = vec![0xB2u8; 64 * 1024];
|
|
|
|
let mut handles = Vec::new();
|
|
for (ttype, payload) in [(1i32, one.clone()), (2i32, two.clone())] {
|
|
let c = c.clone();
|
|
let u = u.clone();
|
|
handles.push(std::thread::spawn(move || {
|
|
let mut errors = 0usize;
|
|
for _ in 0..400 {
|
|
if c.store_thumbnail(&u, ttype, 100, &payload).is_err() {
|
|
errors += 1;
|
|
}
|
|
}
|
|
errors
|
|
}));
|
|
}
|
|
|
|
let mut store_errors = 0usize;
|
|
for h in handles {
|
|
store_errors += h.join().expect("store thread panicked");
|
|
}
|
|
|
|
// Whatever is on disk for each type must be that type's bytes.
|
|
if let Some(got) = c.read_thumbnail(&u, 1, 100) {
|
|
assert_eq!(got, one, "type 1 served type 2's bytes");
|
|
}
|
|
if let Some(got) = c.read_thumbnail(&u, 2, 100) {
|
|
assert_eq!(got, two, "type 2 served type 1's bytes");
|
|
}
|
|
assert_eq!(
|
|
store_errors, 0,
|
|
"a store lost its staging file to the other type"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn thumbnail_store_read_and_evict() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
c.store_thumbnail(&u, 1, 100, b"thumb1").unwrap();
|
|
c.store_thumbnail(&u, 2, 100, b"preview2").unwrap();
|
|
assert_eq!(c.read_thumbnail(&u, 1, 100).unwrap(), b"thumb1");
|
|
assert_eq!(c.read_thumbnail(&u, 2, 100).unwrap(), b"preview2");
|
|
// A newer revision (mtime bump) invalidates the cached thumbnail.
|
|
assert!(c.read_thumbnail(&u, 1, 101).is_none());
|
|
// Content eviction drops thumbnails too.
|
|
c.evict(&u);
|
|
assert!(c.read_thumbnail(&u, 1, 100).is_none());
|
|
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();
|
|
let u = uid("a");
|
|
c.store(&u, 1, 3, b"abc").unwrap();
|
|
c.evict(&u);
|
|
assert!(!c.is_cached(&u, 1, 3));
|
|
}
|
|
|
|
#[test]
|
|
fn block_store_then_read() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
c.store_block(&u, 100, 4096, 0, b"block-zero").unwrap();
|
|
assert_eq!(c.cached_block(&u, 100, 4096, 0).unwrap(), b"block-zero");
|
|
// A different index is a separate cache entry, absent here.
|
|
assert!(c.cached_block(&u, 100, 4096, 1).is_none());
|
|
// A new revision (mtime/size bump) invalidates the block.
|
|
assert!(c.cached_block(&u, 101, 4096, 0).is_none());
|
|
assert!(c.cached_block(&u, 100, 5000, 0).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn evict_drops_blocks() {
|
|
let (c, _d) = cache();
|
|
let u = uid("a");
|
|
c.store_block(&u, 1, 8, 0, b"aaaa").unwrap();
|
|
c.store_block(&u, 1, 8, 1, b"bbbb").unwrap();
|
|
c.evict(&u);
|
|
assert!(c.cached_block(&u, 1, 8, 0).is_none());
|
|
assert!(c.cached_block(&u, 1, 8, 1).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn block_budget_evicts_lru() {
|
|
// 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));
|
|
c.store_block(&u, 1, 12, 1, b"bbbb").unwrap();
|
|
// Touch block 0 so block 1 is the least-recently-used.
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
assert!(c.cached_block(&u, 1, 12, 0).is_some());
|
|
std::thread::sleep(std::time::Duration::from_millis(10));
|
|
c.store_block(&u, 1, 12, 2, b"cccc").unwrap();
|
|
assert!(
|
|
c.cached_block(&u, 1, 12, 0).is_some(),
|
|
"recently-read survives"
|
|
);
|
|
assert!(c.cached_block(&u, 1, 12, 1).is_none(), "LRU block evicted");
|
|
assert!(c.cached_block(&u, 1, 12, 2).is_some(), "newest survives");
|
|
}
|
|
|
|
/// A failed upload must never cost the user their bytes (offline.md Phase 2):
|
|
/// the scratch file moves to staging intact, and staging survives a reopen
|
|
/// the way scratch deliberately does not.
|
|
#[test]
|
|
fn failed_write_is_staged_and_survives_reopen() {
|
|
use std::io::Write as _;
|
|
let dir = TempDir::new();
|
|
let db = Arc::new(Db::open_in_memory().unwrap());
|
|
let content = dir.path().join("content");
|
|
let c = ContentCache::open(content.clone(), dir.path().join("pins.json"), 0, db.clone())
|
|
.unwrap();
|
|
|
|
let (mut f, scratch) = c.create_scratch().unwrap();
|
|
f.write_all(b"unsent work").unwrap();
|
|
drop(f);
|
|
|
|
// A partial overwrite: only the first 11 bytes are real, so the staged
|
|
// file must not be mistaken for uploadable whole-file content.
|
|
let meta = StagedWrite {
|
|
uid: uid("a").to_string(),
|
|
len: 40,
|
|
base_size: 40,
|
|
base_mtime: 100,
|
|
authored: vec![(0, 11)],
|
|
complete: false,
|
|
based_on: Some(Baseline {
|
|
mtime: 100,
|
|
size: 40,
|
|
}),
|
|
};
|
|
let staged = c.stage_write(&meta, &scratch).unwrap();
|
|
assert!(!scratch.exists(), "scratch is moved, not copied");
|
|
assert_eq!(std::fs::read(&staged).unwrap(), b"unsent work");
|
|
|
|
let sidecar: StagedWrite =
|
|
serde_json::from_slice(&std::fs::read(staged.with_extension("json")).unwrap()).unwrap();
|
|
assert_eq!(sidecar.authored, vec![(0, 11)]);
|
|
assert!(
|
|
!sidecar.complete,
|
|
"gaps are zeros, not content: uploading this as-is would corrupt the file"
|
|
);
|
|
|
|
// Reopening wipes scratch (worthless leftovers) but must keep staging.
|
|
drop(c);
|
|
let _c2 = ContentCache::open(content, dir.path().join("pins.json"), 0, db).unwrap();
|
|
assert_eq!(
|
|
std::fs::read(&staged).unwrap(),
|
|
b"unsent work",
|
|
"staged bytes outlive the daemon that failed to upload them"
|
|
);
|
|
}
|
|
|
|
/// A2: `fsync(2)` promises the bytes survive a crash, but queueing happens
|
|
/// at `release`. A scratch file marked durable must therefore outlive the
|
|
/// open that clears the scratch directory — and an unmarked one must not,
|
|
/// since that is a crashed run's rubbish.
|
|
#[test]
|
|
fn fsynced_scratch_survives_reopen_and_unmarked_scratch_does_not() {
|
|
use std::io::Write as _;
|
|
let dir = TempDir::new();
|
|
let db = Arc::new(Db::open_in_memory().unwrap());
|
|
let content = dir.path().join("content");
|
|
let pins = dir.path().join("pins.json");
|
|
let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
|
|
|
|
let (mut synced, synced_path) = c.create_scratch().unwrap();
|
|
synced.write_all(b"fsynced bytes").unwrap();
|
|
synced.sync_all().unwrap();
|
|
let meta = StagedWrite {
|
|
uid: uid("a").to_string(),
|
|
len: 13,
|
|
base_size: 0,
|
|
base_mtime: 100,
|
|
authored: vec![(0, 13)],
|
|
complete: true,
|
|
based_on: None,
|
|
};
|
|
c.mark_scratch_durable(&synced_path, &meta).unwrap();
|
|
|
|
// A second handle that was written but never fsynced: no promise was
|
|
// made about it, and it is indistinguishable from a torn buffer.
|
|
let (mut loose, loose_path) = c.create_scratch().unwrap();
|
|
loose.write_all(b"never fsynced").unwrap();
|
|
|
|
// Crash: no release, no staging, just a restart.
|
|
drop(c);
|
|
let c2 = ContentCache::open(content, pins, 0, db).unwrap();
|
|
|
|
assert!(!synced_path.exists(), "scratch dir is still cleared");
|
|
assert!(!loose_path.exists());
|
|
|
|
let recovered = c2.recovered_writes();
|
|
assert_eq!(recovered.len(), 1, "only the fsynced write is recoverable");
|
|
let (blob, got) = &recovered[0];
|
|
assert_eq!(std::fs::read(blob).unwrap(), b"fsynced bytes");
|
|
assert_eq!(got.uid, uid("a").to_string());
|
|
assert_eq!(got.authored, vec![(0, 13)]);
|
|
|
|
// Staging it is what a restart does with it; the sidecar goes after.
|
|
let staged = c2.stage_write(got, blob).unwrap();
|
|
c2.discard_recovered(blob);
|
|
assert_eq!(std::fs::read(&staged).unwrap(), b"fsynced bytes");
|
|
assert!(c2.recovered_writes().is_empty(), "recovery does not repeat");
|
|
}
|
|
|
|
/// Releasing a write hands the bytes to staging, so the sidecar an earlier
|
|
/// `fsync` left must not offer recovery a second, stale copy of it.
|
|
#[test]
|
|
fn cleared_durability_marker_stops_recovery() {
|
|
use std::io::Write as _;
|
|
let dir = TempDir::new();
|
|
let db = Arc::new(Db::open_in_memory().unwrap());
|
|
let content = dir.path().join("content");
|
|
let pins = dir.path().join("pins.json");
|
|
let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
|
|
|
|
let (mut f, path) = c.create_scratch().unwrap();
|
|
f.write_all(b"released").unwrap();
|
|
let meta = StagedWrite {
|
|
uid: uid("a").to_string(),
|
|
len: 8,
|
|
base_size: 0,
|
|
base_mtime: 100,
|
|
authored: vec![(0, 8)],
|
|
complete: true,
|
|
based_on: None,
|
|
};
|
|
c.mark_scratch_durable(&path, &meta).unwrap();
|
|
c.clear_scratch_durable(&path);
|
|
|
|
drop(c);
|
|
let c2 = ContentCache::open(content, pins, 0, db).unwrap();
|
|
assert!(c2.recovered_writes().is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn scratch_file_is_writable_and_isolated() {
|
|
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
|
|
let (c, _d) = cache();
|
|
let (mut f1, p1) = c.create_scratch().unwrap();
|
|
let (_f2, p2) = c.create_scratch().unwrap();
|
|
assert_ne!(p1, p2, "each scratch file is unique");
|
|
f1.write_all(b"scratch").unwrap();
|
|
f1.seek(SeekFrom::Start(0)).unwrap();
|
|
let mut s = String::new();
|
|
f1.read_to_string(&mut s).unwrap();
|
|
assert_eq!(s, "scratch");
|
|
}
|
|
}
|