mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
release(1.8.2)
This commit is contained in:
1 parent
326a9b4e92
commit
078774f6a1
11 files changed
+2106
-253
No files matched your search
@@ -104,6 +104,31 @@ fn read_request_line(reader: &mut impl BufRead) -> std::io::Result<Option<String
|
||||
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
|
||||
}
|
||||
|
||||
/// Turn a CLI-supplied path into the mount that owns it and the path relative
|
||||
/// to *that* mount's root.
|
||||
///
|
||||
/// [`rel_to_mount`] is the older, primary-only reading, and it is what every
|
||||
/// mutating request still uses. It rejects anything outside the primary
|
||||
/// mountpoint, which leaves a secondary on-demand mount — a synced folder
|
||||
/// switched out of mirror mode — with no way to be named at all: `pdfs ls` and
|
||||
/// `pdfs refresh` both answered "is not under the mountpoint", so a folder
|
||||
/// serving a stale listing had no manual escape hatch (`docs/BUGS.md` B86).
|
||||
///
|
||||
/// A relative path keeps the old meaning, relative to the primary mount, since
|
||||
/// there is nothing in it to choose a mount by.
|
||||
fn route_to_mount(core: &Core, mountpoint: &Path, path: &str) -> CoreResult<(Core, PathBuf)> {
|
||||
let p = Path::new(path);
|
||||
if !p.is_absolute() {
|
||||
return Ok((core.clone(), p.to_path_buf()));
|
||||
}
|
||||
if let Some(routed) = core.rooted_at(p) {
|
||||
return Ok(routed);
|
||||
}
|
||||
// Fall back to the primary reading so the error text stays the familiar one
|
||||
// when the path is simply outside everything this daemon has mounted.
|
||||
rel_to_mount(mountpoint, path).map(|rel| (core.clone(), rel))
|
||||
}
|
||||
|
||||
/// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path
|
||||
/// must live under `mountpoint`; a relative path is taken as already relative to
|
||||
/// the mount root.
|
||||
@@ -193,8 +218,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
Ok(CtlRequest::ListPins) => CtlResponse::Pins {
|
||||
pins: core.cache.list_pins(),
|
||||
},
|
||||
Ok(CtlRequest::ListDir { path }) => match rel_to_mount(mountpoint, &path) {
|
||||
Ok(rel) => match core.list_dir(&rel) {
|
||||
Ok(CtlRequest::ListDir { path }) => match route_to_mount(core, mountpoint, &path) {
|
||||
Ok((core, rel)) => match core.list_dir(&rel) {
|
||||
Ok(entries) => CtlResponse::Entries { entries },
|
||||
Err(e) => CtlResponse::error(e),
|
||||
},
|
||||
@@ -202,9 +227,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
},
|
||||
Ok(CtlRequest::Refresh { scope }) => {
|
||||
let result = match &scope {
|
||||
RefreshScope::Dir { path } => {
|
||||
rel_to_mount(mountpoint, path).and_then(|rel| core.refresh_dir(&rel))
|
||||
}
|
||||
RefreshScope::Dir { path } => route_to_mount(core, mountpoint, path)
|
||||
.and_then(|(core, rel)| core.refresh_dir(&rel)),
|
||||
RefreshScope::Trash => {
|
||||
core.invalidate_trash();
|
||||
Ok(())
|
||||
|
||||
+202
-14
@@ -55,7 +55,7 @@ use fuser::{
|
||||
};
|
||||
use futures::StreamExt as _;
|
||||
use pdfs_core::batch;
|
||||
use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite, block_len};
|
||||
use pdfs_core::cache::{BLOCK_SIZE, Baseline, BlockGeometry, BlockSpan, ContentCache, StagedWrite};
|
||||
use pdfs_core::config::{AppDirs, SweepMode};
|
||||
use pdfs_core::control::{
|
||||
ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
|
||||
@@ -715,6 +715,26 @@ struct MountedState {
|
||||
/// [`Core::notifier`], still empty until its session is spawned.
|
||||
notifier: Arc<OnceLock<Notifier>>,
|
||||
session_live: Arc<AtomicBool>,
|
||||
/// This mount's in-flight size upgrades, keyed by *its* folder inodes — so
|
||||
/// a `Core` rebuilt onto this inode space by [`Core::rooted_at`] shares the
|
||||
/// batch the mount's own session is already running rather than starting a
|
||||
/// second one against inode numbers that mean nothing to it.
|
||||
size_upgrades: Arc<Mutex<HashMap<u64, Arc<SizeUpgrade>>>>,
|
||||
}
|
||||
|
||||
/// The per-mount half of a [`Core`] — every field [`Core::fork_state`] gives a
|
||||
/// fork a fresh copy of — recovered from the registry.
|
||||
///
|
||||
/// This is what lets a control request naming a path under a secondary mount be
|
||||
/// answered *in that mount's inode space* instead of being rejected for not
|
||||
/// being under the primary mountpoint (`docs/BUGS.md` B86). See
|
||||
/// [`Core::rooted_at`].
|
||||
struct MountParts {
|
||||
mountpoint: PathBuf,
|
||||
state: Arc<Mutex<State>>,
|
||||
notifier: Arc<OnceLock<Notifier>>,
|
||||
session_live: Arc<AtomicBool>,
|
||||
size_upgrades: Arc<Mutex<HashMap<u64, Arc<SizeUpgrade>>>>,
|
||||
}
|
||||
|
||||
/// Every mounted inode space in the daemon, shared by the primary `Core` and
|
||||
@@ -737,6 +757,7 @@ impl StateRegistry {
|
||||
state: &Arc<Mutex<State>>,
|
||||
notifier: Arc<OnceLock<Notifier>>,
|
||||
session_live: Arc<AtomicBool>,
|
||||
size_upgrades: Arc<Mutex<HashMap<u64, Arc<SizeUpgrade>>>>,
|
||||
) {
|
||||
let mut states = self.0.lock();
|
||||
states.retain(|m| m.state.strong_count() > 0);
|
||||
@@ -751,6 +772,7 @@ impl StateRegistry {
|
||||
state: Arc::downgrade(state),
|
||||
notifier,
|
||||
session_live,
|
||||
size_upgrades,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -777,9 +799,9 @@ impl StateRegistry {
|
||||
///
|
||||
/// On-demand roots can be nested below another location. Selecting by the
|
||||
/// longest component prefix ensures the nested session wins rather than the
|
||||
/// broader primary mount. Callers currently need the selected mount only;
|
||||
/// a future path router can derive a relative suffix with
|
||||
/// `path.strip_prefix(mountpoint)` without storing a second path value here.
|
||||
/// broader primary mount. Callers that need the relative suffix derive it
|
||||
/// with `path.strip_prefix(mountpoint)`; [`StateRegistry::covering_parts`]
|
||||
/// is the variant that also hands back enough to *serve* a request there.
|
||||
fn covering(&self, path: &Path) -> Option<LiveMount> {
|
||||
self.live()
|
||||
.into_iter()
|
||||
@@ -789,6 +811,48 @@ impl StateRegistry {
|
||||
.max_by_key(|(mountpoint, _, _, _)| mountpoint.components().count())
|
||||
}
|
||||
|
||||
/// [`StateRegistry::register`] with an empty size-upgrade map, for the
|
||||
/// tests that exercise the reap-and-route rules and nothing else.
|
||||
#[cfg(test)]
|
||||
fn register_bare(
|
||||
&self,
|
||||
mountpoint: &Path,
|
||||
state: &Arc<Mutex<State>>,
|
||||
notifier: Arc<OnceLock<Notifier>>,
|
||||
session_live: Arc<AtomicBool>,
|
||||
) {
|
||||
self.register(
|
||||
mountpoint,
|
||||
state,
|
||||
notifier,
|
||||
session_live,
|
||||
Arc::new(Mutex::new(HashMap::new())),
|
||||
);
|
||||
}
|
||||
|
||||
/// [`StateRegistry::covering`], but returning every per-mount field a
|
||||
/// [`Core`] needs to be re-rooted onto that inode space.
|
||||
///
|
||||
/// Kept separate from `covering` so the common callers — which only ask
|
||||
/// *whether* a path is mounted — keep the cheaper tuple.
|
||||
fn covering_parts(&self, path: &Path) -> Option<MountParts> {
|
||||
let mut states = self.0.lock();
|
||||
states.retain(|m| m.state.strong_count() > 0);
|
||||
states
|
||||
.iter()
|
||||
.filter(|m| m.session_live.load(Ordering::Acquire) && path.starts_with(&m.mountpoint))
|
||||
.max_by_key(|m| m.mountpoint.components().count())
|
||||
.and_then(|m| {
|
||||
Some(MountParts {
|
||||
mountpoint: m.mountpoint.clone(),
|
||||
state: m.state.upgrade()?,
|
||||
notifier: m.notifier.clone(),
|
||||
session_live: m.session_live.clone(),
|
||||
size_upgrades: m.size_upgrades.clone(),
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
fn is_mounted_at(&self, path: &Path) -> bool {
|
||||
self.covering(path).is_some_and(|(mountpoint, _, _, live)| {
|
||||
mountpoint == path && live.load(Ordering::Acquire)
|
||||
@@ -908,9 +972,41 @@ impl Core {
|
||||
&self.state,
|
||||
self.notifier.clone(),
|
||||
self.session_live.clone(),
|
||||
self.size_upgrades.clone(),
|
||||
);
|
||||
}
|
||||
|
||||
/// This `Core`, re-rooted onto whichever live mount most specifically covers
|
||||
/// the absolute path `abs`, together with `abs` relative to that mount's
|
||||
/// root.
|
||||
///
|
||||
/// Everything a `Core` holds is per-daemon except the five fields
|
||||
/// [`Core::fork_state`] replaces, so swapping exactly those turns the
|
||||
/// primary `Core` into the fork that owns the path — sharing the fork's
|
||||
/// inode space, notification channel and size-upgrade batches rather than
|
||||
/// shadowing them. Without this, every path-addressed control request is
|
||||
/// answered against the primary inode space, so anything under a secondary
|
||||
/// on-demand mount is rejected as "not under the mountpoint" and the user
|
||||
/// has no way to re-enumerate a folder they can see (`docs/BUGS.md` B86).
|
||||
///
|
||||
/// `None` when no live mount covers `abs`.
|
||||
fn rooted_at(&self, abs: &Path) -> Option<(Core, PathBuf)> {
|
||||
let parts = self.states.covering_parts(abs)?;
|
||||
let rel = abs.strip_prefix(&parts.mountpoint).ok()?.to_path_buf();
|
||||
let mut core = self.clone();
|
||||
if !Arc::ptr_eq(&parts.state, &self.state) {
|
||||
// A different inode space, so adopt its per-mount half wholesale.
|
||||
// `primary` is a property of the mount, and the only mount this
|
||||
// `Core` can be the primary of is the one it was built for.
|
||||
core.primary = false;
|
||||
core.state = parts.state;
|
||||
core.notifier = parts.notifier;
|
||||
core.session_live = parts.session_live;
|
||||
core.size_upgrades = parts.size_upgrades;
|
||||
}
|
||||
Some((core, rel))
|
||||
}
|
||||
|
||||
/// Run `apply` against every live inode space — this daemon's primary mount
|
||||
/// and each on-demand fork — one lock at a time.
|
||||
///
|
||||
@@ -3934,6 +4030,33 @@ impl Core {
|
||||
}
|
||||
}
|
||||
|
||||
/// Drop the cached child listing of the folder `parent`, everywhere it is
|
||||
/// held — naming the folder by uid rather than by inode.
|
||||
///
|
||||
/// [`Core::invalidate_parent_listing`] is the path-based sibling, and it
|
||||
/// only reaches *this* mount's inode space. A remote mutation made by the
|
||||
/// sync engine needs more than that, for two reasons. The folder it changed
|
||||
/// is usually not resident in any inode space at the time — a mirror folder
|
||||
/// has no FUSE session at all — so there is no inode to name. And the stale
|
||||
/// listing that outlives the pass is not a hot-cache map but the `listed`
|
||||
/// flag on the folder's DB row, which survives a daemon restart and is what
|
||||
/// a later `ensure_children` trusts.
|
||||
///
|
||||
/// So the flag is cleared unconditionally, and the resident inode spaces are
|
||||
/// invalidated on top of it. That is what makes a mirror folder switched to
|
||||
/// on-demand re-enumerate instead of serving a snapshot that predates the
|
||||
/// engine's own uploads — `docs/BUGS.md` B86.
|
||||
pub(crate) fn invalidate_children_of(&self, parent: &NodeUid) {
|
||||
if let Err(e) = self.db.set_listed(parent, false) {
|
||||
warn!(uid = %parent, error = ?e, "could not clear cached listing");
|
||||
}
|
||||
self.for_each_state(|st| {
|
||||
if let Some(&ino) = st.by_uid.get(parent) {
|
||||
st.invalidate_listing(ino);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Rename a file or folder to `new_name`. `rel` is mountpoint-relative.
|
||||
/// Mirrors the FUSE `rename` write path: rename on the remote, forget the
|
||||
/// node so it re-interns under its new name, and drop the parent listing so
|
||||
@@ -5853,7 +5976,7 @@ mod tests {
|
||||
("/mnt/fork-a", &fork_a),
|
||||
("/mnt/fork-b", &fork_b),
|
||||
] {
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new(mountpoint),
|
||||
state,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -5904,19 +6027,19 @@ mod tests {
|
||||
let nested_live = session_flag(false);
|
||||
let sibling_live = session_flag(true);
|
||||
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
primary_path,
|
||||
&primary,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
primary_live.clone(),
|
||||
);
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
nested_path,
|
||||
&nested,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
nested_live.clone(),
|
||||
);
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
sibling_path,
|
||||
&sibling,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -5994,6 +6117,71 @@ mod tests {
|
||||
assert_eq!(registry.live().len(), 2);
|
||||
}
|
||||
|
||||
/// The routing half of `docs/BUGS.md` B86: a path under a secondary mount
|
||||
/// has to resolve to *that* mount's inode space and a suffix relative to
|
||||
/// *its* root, or `pdfs ls` and `pdfs refresh` can only ever name the
|
||||
/// primary mount and a stale on-demand folder has no escape hatch.
|
||||
#[test]
|
||||
fn covering_parts_route_a_path_to_the_mount_that_owns_it() {
|
||||
let registry = StateRegistry::default();
|
||||
let (primary, _primary_dir) = rooted_state("primary-volume", "primary");
|
||||
let (nested, _nested_dir) = rooted_state("nested-volume", "nested");
|
||||
let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary));
|
||||
let nested = std::sync::Arc::new(parking_lot::Mutex::new(nested));
|
||||
let primary_path = std::path::Path::new("/home/me/ProtonDrive");
|
||||
let nested_path = std::path::Path::new("/home/me/ProtonDrive/Device");
|
||||
let primary_upgrades = std::sync::Arc::new(parking_lot::Mutex::new(HashMap::new()));
|
||||
let nested_upgrades = std::sync::Arc::new(parking_lot::Mutex::new(HashMap::new()));
|
||||
|
||||
registry.register(
|
||||
primary_path,
|
||||
&primary,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
session_flag(true),
|
||||
primary_upgrades.clone(),
|
||||
);
|
||||
registry.register(
|
||||
nested_path,
|
||||
&nested,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
session_flag(true),
|
||||
nested_upgrades.clone(),
|
||||
);
|
||||
|
||||
let deep = nested_path.join("sub/file.txt");
|
||||
let parts = registry.covering_parts(&deep).expect("nested mount covers");
|
||||
assert_eq!(parts.mountpoint, nested_path, "the nested mount wins");
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(&parts.state, &nested),
|
||||
"the request must be answered in the nested inode space"
|
||||
);
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(&parts.size_upgrades, &nested_upgrades),
|
||||
"and share that mount's in-flight size upgrades, not the primary's"
|
||||
);
|
||||
assert_eq!(
|
||||
deep.strip_prefix(&parts.mountpoint).unwrap(),
|
||||
std::path::Path::new("sub/file.txt"),
|
||||
"the suffix is relative to the mount that owns it, not the primary"
|
||||
);
|
||||
|
||||
let shallow = primary_path.join("Documents");
|
||||
let parts = registry
|
||||
.covering_parts(&shallow)
|
||||
.expect("primary mount covers");
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(&parts.state, &primary),
|
||||
"a path outside the nested root still routes to the primary"
|
||||
);
|
||||
|
||||
assert!(
|
||||
registry
|
||||
.covering_parts(std::path::Path::new("/tmp/outside"))
|
||||
.is_none(),
|
||||
"a path under no mount routes nowhere, and the caller reports that"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn state_registry_resolves_uids_from_any_resident_mount() {
|
||||
let registry = StateRegistry::default();
|
||||
@@ -6007,13 +6195,13 @@ mod tests {
|
||||
fork.children.insert(super::ROOT_INO, vec![fork_ino]);
|
||||
let primary = std::sync::Arc::new(parking_lot::Mutex::new(primary));
|
||||
let fork = std::sync::Arc::new(parking_lot::Mutex::new(fork));
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/primary"),
|
||||
&primary,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
session_flag(true),
|
||||
);
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/device"),
|
||||
&fork,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -6041,7 +6229,7 @@ mod tests {
|
||||
let child_ino = state.intern(super::ROOT_INO, child_node);
|
||||
state.children.insert(super::ROOT_INO, vec![child_ino]);
|
||||
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/device"),
|
||||
&state,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -6104,7 +6292,7 @@ mod tests {
|
||||
);
|
||||
|
||||
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/device"),
|
||||
&state,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -6372,7 +6560,7 @@ mod tests {
|
||||
state.children.insert(revoked_ino, vec![child_ino]);
|
||||
|
||||
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/device"),
|
||||
&state,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
@@ -6426,7 +6614,7 @@ mod tests {
|
||||
state.children.insert(cycle_b, vec![cycle_a]);
|
||||
|
||||
let state = std::sync::Arc::new(parking_lot::Mutex::new(state));
|
||||
registry.register(
|
||||
registry.register_bare(
|
||||
std::path::Path::new("/mnt/device"),
|
||||
&state,
|
||||
std::sync::Arc::new(std::sync::OnceLock::new()),
|
||||
|
||||
+157
-88
@@ -102,22 +102,38 @@ static NEXT_OPEN_ID: AtomicU64 = AtomicU64::new(0);
|
||||
/// Validated by the same `(mtime, size)` pair the on-disk caches use: a node
|
||||
/// whose tag no longer matches has its blocks dropped rather than served, so a
|
||||
/// new revision can never be stitched together from the old one.
|
||||
/// A ring entry's identity: the node, and the plaintext range the block covers.
|
||||
///
|
||||
/// The range, rather than the block index, because the index alone stopped
|
||||
/// naming one range when the geometry started coming from the revision — a
|
||||
/// block cached under the [`BlockGeometry::uniform`] fallback and the block at
|
||||
/// the same index under the real geometry are different bytes
|
||||
/// (`docs/BUGS.md` B85). A range is the same identity under either plan, so a
|
||||
/// re-planned read misses rather than being served the wrong bytes.
|
||||
type RingKey = (NodeUid, BlockSpan);
|
||||
|
||||
#[derive(Default)]
|
||||
pub(super) struct BlockRing {
|
||||
blocks: HashMap<(NodeUid, u64), Arc<Vec<u8>>>,
|
||||
blocks: HashMap<RingKey, Arc<Vec<u8>>>,
|
||||
/// Keys least-recently-used first, for eviction.
|
||||
order: VecDeque<(NodeUid, u64)>,
|
||||
order: VecDeque<RingKey>,
|
||||
/// Per-node validity tag, `(mtime, size)`.
|
||||
tags: HashMap<NodeUid, (i64, u64)>,
|
||||
bytes: u64,
|
||||
}
|
||||
|
||||
impl BlockRing {
|
||||
fn get(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64) -> Option<Arc<Vec<u8>>> {
|
||||
fn get(
|
||||
&mut self,
|
||||
uid: &NodeUid,
|
||||
mtime: i64,
|
||||
size: u64,
|
||||
span: BlockSpan,
|
||||
) -> Option<Arc<Vec<u8>>> {
|
||||
if self.tags.get(uid) != Some(&(mtime, size)) {
|
||||
return None;
|
||||
}
|
||||
let key = (uid.clone(), idx);
|
||||
let key = (uid.clone(), span);
|
||||
let hit = self.blocks.get(&key).cloned()?;
|
||||
// Move to the young end: a block being read repeatedly (the whole reason
|
||||
// this exists) must not age out under a sequential read passing through.
|
||||
@@ -128,14 +144,21 @@ impl BlockRing {
|
||||
Some(hit)
|
||||
}
|
||||
|
||||
fn insert(&mut self, uid: &NodeUid, mtime: i64, size: u64, idx: u64, bytes: Arc<Vec<u8>>) {
|
||||
fn insert(
|
||||
&mut self,
|
||||
uid: &NodeUid,
|
||||
mtime: i64,
|
||||
size: u64,
|
||||
span: BlockSpan,
|
||||
bytes: Arc<Vec<u8>>,
|
||||
) {
|
||||
if self.tags.get(uid) != Some(&(mtime, size)) {
|
||||
// Revision changed under us (or first sight): anything held for this
|
||||
// node describes the old one.
|
||||
self.drop_node(uid);
|
||||
self.tags.insert(uid.clone(), (mtime, size));
|
||||
}
|
||||
let key = (uid.clone(), idx);
|
||||
let key = (uid.clone(), span);
|
||||
if self.blocks.contains_key(&key) {
|
||||
return;
|
||||
}
|
||||
@@ -176,8 +199,10 @@ impl BlockRing {
|
||||
/// The result of fetching one block, shared between everyone who wanted it.
|
||||
type BlockResult = Result<Arc<Vec<u8>>, Errno>;
|
||||
|
||||
/// Identifies a block *of a revision*: node, validity tag, index.
|
||||
type BlockKey = (NodeUid, i64, u64, u64);
|
||||
/// Identifies a block *of a revision*: node, validity tag, and the plaintext
|
||||
/// range the block covers — see [`RingKey`] for why it is the range and not the
|
||||
/// index.
|
||||
type BlockKey = (NodeUid, i64, u64, BlockSpan);
|
||||
|
||||
/// A claim on a block key: who owns the fetch, and where its result is published.
|
||||
type Claim = (u64, tokio::sync::watch::Receiver<Option<BlockResult>>);
|
||||
@@ -417,11 +442,30 @@ impl Core {
|
||||
self.prefetch.streams.lock().remove(uid);
|
||||
}
|
||||
|
||||
/// How this revision's plaintext is divided into content blocks.
|
||||
///
|
||||
/// The revision's own geometry when a previous read has learned and recorded
|
||||
/// it, and the [`BLOCK_SIZE`] assumption when it has not — which is every
|
||||
/// first read, because learning it means opening a `RevisionReader`, and
|
||||
/// opening one per read is precisely the per-file key derivation the block
|
||||
/// cache exists to avoid (bugs.md B12, B85).
|
||||
///
|
||||
/// Being wrong here is not a correctness problem: `read_at` plans over the
|
||||
/// real sizes and clamps to what was asked for either way. It costs a
|
||||
/// straddling fetch and a cached block the next read will not ask for, and
|
||||
/// both settle once the geometry is known.
|
||||
fn block_geometry(&self, uid: &NodeUid, mtime: i64, fsize: u64) -> BlockGeometry {
|
||||
self.cache
|
||||
.block_geometry(uid, mtime, fsize)
|
||||
.unwrap_or_else(|| BlockGeometry::uniform(fsize))
|
||||
}
|
||||
|
||||
/// Serve bytes `[offset, offset + len)` of `uid`'s active revision, hitting
|
||||
/// the on-disk caches before the network: a whole-file blob (pinned files)
|
||||
/// first, then the block cache — fetching only the [`BLOCK_SIZE`]-aligned
|
||||
/// blocks that overlap the request and caching each. `mtime`/`fsize` validate
|
||||
/// both caches. Network I/O runs without any lock held.
|
||||
/// first, then the block cache — fetching only the blocks that overlap the
|
||||
/// request, on the revision's own boundaries ([`Core::block_geometry`]), and
|
||||
/// caching each. `mtime`/`fsize` validate both caches. Network I/O runs
|
||||
/// without any lock held.
|
||||
pub(super) fn read_range(
|
||||
&self,
|
||||
uid: &NodeUid,
|
||||
@@ -500,7 +544,7 @@ impl Core {
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Fetch block `bidx`, joining a fetch already in flight for it rather than
|
||||
/// Fetch the block at `span`, joining a fetch already in flight rather than
|
||||
/// issuing a second one, and publish the result to everyone who joined.
|
||||
///
|
||||
/// See [`BlockFlight`]. A follower whose leader disappears (a panic, or a
|
||||
@@ -512,10 +556,10 @@ impl Core {
|
||||
uid: &NodeUid,
|
||||
mtime: i64,
|
||||
fsize: u64,
|
||||
bidx: u64,
|
||||
span: BlockSpan,
|
||||
cache_blocks: bool,
|
||||
) -> BlockResult {
|
||||
let key: BlockKey = (uid.clone(), mtime, fsize, bidx);
|
||||
let key: BlockKey = (uid.clone(), mtime, fsize, span);
|
||||
let claim = {
|
||||
let mut inflight = self.block_flight.inflight.lock();
|
||||
match inflight.get(&key) {
|
||||
@@ -547,12 +591,12 @@ impl Core {
|
||||
}
|
||||
}
|
||||
return self
|
||||
.read_block(reader, uid, mtime, fsize, bidx, cache_blocks)
|
||||
.read_block(reader, uid, mtime, fsize, span, cache_blocks)
|
||||
.await;
|
||||
}
|
||||
};
|
||||
let result = self
|
||||
.read_block(reader, uid, mtime, fsize, bidx, cache_blocks)
|
||||
.read_block(reader, uid, mtime, fsize, span, cache_blocks)
|
||||
.await;
|
||||
// Retire the claim before publishing, so a racer that wakes and re-enters
|
||||
// finds no stale entry to join.
|
||||
@@ -578,11 +622,9 @@ impl Core {
|
||||
uid: &NodeUid,
|
||||
mtime: i64,
|
||||
fsize: u64,
|
||||
bidx: u64,
|
||||
span: BlockSpan,
|
||||
cache_blocks: bool,
|
||||
) -> BlockResult {
|
||||
let bstart = bidx * BLOCK_SIZE;
|
||||
let blen = block_len(fsize, bidx);
|
||||
// The reader derives every block boundary from the revision's recorded
|
||||
// block sizes. If they do not add up to the size the node reports, this
|
||||
// file's boundaries are wrong, and the failure is not confined to short
|
||||
@@ -591,25 +633,33 @@ impl Core {
|
||||
// kind takes the file off the range path entirely (bugs.md B84).
|
||||
let bytes = if reader.size() != fsize {
|
||||
warn!(
|
||||
%uid, bidx, fsize,
|
||||
%uid, bidx = span.idx, fsize,
|
||||
reader_size = reader.size(),
|
||||
reader_blocks = reader.block_sizes().len(),
|
||||
"the revision's block table disagrees with the node size"
|
||||
);
|
||||
self.repair_block(uid, mtime, fsize, bidx).await?
|
||||
self.repair_block(uid, mtime, fsize, span).await?
|
||||
} else {
|
||||
let bytes = reader.read_at(bstart, blen).await.map_err(|e| {
|
||||
warn!(%uid, bstart, blen, error = %e, "block read failed");
|
||||
// Learned here rather than on a path of its own: this is where a
|
||||
// reader is open anyway, and recording the real geometry is what
|
||||
// lets the *next* read plan its blocks on the revision's own
|
||||
// boundaries instead of the 4 MiB fallback (bugs.md B85).
|
||||
if cache_blocks {
|
||||
self.cache
|
||||
.store_block_geometry(uid, mtime, fsize, reader.block_sizes());
|
||||
}
|
||||
let bytes = reader.read_at(span.start, span.len).await.map_err(|e| {
|
||||
warn!(%uid, start = span.start, len = span.len, error = %e, "block read failed");
|
||||
Errno::EIO
|
||||
})?;
|
||||
if bytes.len() as u64 == blen {
|
||||
if bytes.len() as u64 == span.len {
|
||||
bytes
|
||||
} else {
|
||||
warn!(
|
||||
%uid, bidx, blen, got = bytes.len(), fsize,
|
||||
"block came back shorter than the file's size implies"
|
||||
%uid, bidx = span.idx, blen = span.len, got = bytes.len(), fsize,
|
||||
"block came back shorter than the range asked for"
|
||||
);
|
||||
self.repair_block(uid, mtime, fsize, bidx).await?
|
||||
self.repair_block(uid, mtime, fsize, span).await?
|
||||
}
|
||||
};
|
||||
let bytes = Arc::new(bytes);
|
||||
@@ -618,16 +668,16 @@ impl Core {
|
||||
let uid = uid.clone();
|
||||
let bytes = bytes.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let _ = cache.store_block(&uid, mtime, fsize, bidx, &bytes);
|
||||
let _ = cache.store_block(&uid, mtime, fsize, span, &bytes);
|
||||
});
|
||||
}
|
||||
self.block_ring
|
||||
.lock()
|
||||
.insert(uid, mtime, fsize, bidx, bytes.clone());
|
||||
.insert(uid, mtime, fsize, span, bytes.clone());
|
||||
Ok(bytes)
|
||||
}
|
||||
|
||||
/// Recover block `bidx` of a file whose per-block reads cannot be trusted,
|
||||
/// Recover the block at `span` of a file whose per-block reads cannot be
|
||||
/// by fetching the whole revision through the manifest-verified download and
|
||||
/// serving the block out of that.
|
||||
///
|
||||
@@ -648,10 +698,9 @@ impl Core {
|
||||
uid: &NodeUid,
|
||||
mtime: i64,
|
||||
fsize: u64,
|
||||
bidx: u64,
|
||||
span: BlockSpan,
|
||||
) -> Result<Vec<u8>, Errno> {
|
||||
let blen = block_len(fsize, bidx);
|
||||
let bstart = bidx * BLOCK_SIZE;
|
||||
let (bstart, blen) = (span.start, span.len);
|
||||
if fsize > REPAIR_MAX {
|
||||
warn!(
|
||||
%uid, fsize,
|
||||
@@ -683,11 +732,11 @@ impl Core {
|
||||
}
|
||||
match self.cache.read_range(uid, mtime, fsize, bstart, blen) {
|
||||
Some(bytes) if bytes.len() as u64 == blen => {
|
||||
info!(%uid, bidx, "served a short block from a repaired whole-file download");
|
||||
info!(%uid, bidx = span.idx, "served a short block from a repaired whole-file download");
|
||||
Ok(bytes)
|
||||
}
|
||||
_ => {
|
||||
error!(%uid, bidx, fsize, "repairing a short block did not produce the block");
|
||||
error!(%uid, bidx = span.idx, fsize, "repairing a short block did not produce the block");
|
||||
Err(Errno::EIO)
|
||||
}
|
||||
}
|
||||
@@ -776,17 +825,19 @@ impl Core {
|
||||
if depth == 0 {
|
||||
return;
|
||||
}
|
||||
let last_block = fsize.saturating_sub(1) / BLOCK_SIZE;
|
||||
let wanted: Vec<u64> = ((last + 1)..=(last + depth).min(last_block))
|
||||
.filter(|&bidx| {
|
||||
let geometry = self.block_geometry(uid, mtime, fsize);
|
||||
let last_block = geometry.blocks().saturating_sub(1);
|
||||
let wanted: Vec<BlockSpan> = ((last + 1)..=(last + depth).min(last_block))
|
||||
.filter_map(|bidx| geometry.span(bidx))
|
||||
.filter(|&span| {
|
||||
// Blocks already in memory need nothing; blocks already on disk
|
||||
// are cheap enough to serve that spending a round-trip to move
|
||||
// them into the ring would be the wrong trade.
|
||||
self.block_ring
|
||||
.lock()
|
||||
.get(uid, mtime, fsize, bidx)
|
||||
.get(uid, mtime, fsize, span)
|
||||
.is_none()
|
||||
&& !(cache_blocks && self.cache.has_block(uid, mtime, fsize, bidx))
|
||||
&& !(cache_blocks && self.cache.has_block(uid, mtime, fsize, span))
|
||||
})
|
||||
.collect();
|
||||
if wanted.is_empty() {
|
||||
@@ -803,7 +854,7 @@ impl Core {
|
||||
// round-trips, and the reader catches up with the window before it
|
||||
// finishes.
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for bidx in wanted {
|
||||
for span in wanted {
|
||||
let Ok(permit) = core.prefetch_budget.clone().try_acquire_owned() else {
|
||||
break; // the budget is spent; the demand path can have it
|
||||
};
|
||||
@@ -811,10 +862,10 @@ impl Core {
|
||||
set.spawn(async move {
|
||||
let _permit = permit;
|
||||
if let Err(error) = core
|
||||
.fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks)
|
||||
.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks)
|
||||
.await
|
||||
{
|
||||
debug!(%uid, bidx, ?error, "prefetch failed");
|
||||
debug!(%uid, bidx = span.idx, ?error, "prefetch failed");
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -844,8 +895,13 @@ impl Core {
|
||||
}
|
||||
let end = offset.saturating_add(len).min(fsize);
|
||||
let mut out = Vec::with_capacity((end - offset) as usize);
|
||||
let first = offset / BLOCK_SIZE;
|
||||
let last = (end - 1) / BLOCK_SIZE;
|
||||
// On the revision's own boundaries where they are known, and on the
|
||||
// 4 MiB fallback where they are not (bugs.md B85).
|
||||
let wanted = self.block_geometry(uid, mtime, fsize).spans(offset, end);
|
||||
let (Some(&head), Some(&tail)) = (wanted.first(), wanted.last()) else {
|
||||
return Ok(out);
|
||||
};
|
||||
let (first, last) = (head.idx, tail.idx);
|
||||
|
||||
// Collect the blocks overlapping the request, serving any already cached
|
||||
// and fetching the rest concurrently. A multi-block read (e.g. a media
|
||||
@@ -853,31 +909,31 @@ impl Core {
|
||||
// would otherwise stall on each block round-trip in turn; downloading the
|
||||
// misses in parallel saturates the connection and bounds latency at the
|
||||
// slowest single block instead of their sum.
|
||||
let mut blocks: Vec<Option<Arc<Vec<u8>>>> = Vec::with_capacity((last - first + 1) as usize);
|
||||
let mut misses: Vec<u64> = Vec::new();
|
||||
for bidx in first..=last {
|
||||
let mut blocks: Vec<Option<Arc<Vec<u8>>>> = Vec::with_capacity(wanted.len());
|
||||
let mut misses: Vec<BlockSpan> = Vec::new();
|
||||
for &span in &wanted {
|
||||
// The in-memory ring comes first on both paths: a streaming read has
|
||||
// no other cache at all, and a cached one would otherwise pay two
|
||||
// opens, a JSON parse and a 4 MiB read+alloc for every 128 KiB the
|
||||
// kernel asks for (audit T3).
|
||||
let hit = self.block_ring.lock().get(uid, mtime, fsize, bidx);
|
||||
let hit = self.block_ring.lock().get(uid, mtime, fsize, span);
|
||||
let hit = hit.or_else(|| {
|
||||
if !cache_blocks {
|
||||
return None;
|
||||
}
|
||||
let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?);
|
||||
let bytes = Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?);
|
||||
// Promote it, so the other ~31 reads of this block are answered
|
||||
// from memory.
|
||||
self.block_ring
|
||||
.lock()
|
||||
.insert(uid, mtime, fsize, bidx, bytes.clone());
|
||||
.insert(uid, mtime, fsize, span, bytes.clone());
|
||||
Some(bytes)
|
||||
});
|
||||
match hit {
|
||||
Some(b) => blocks.push(Some(b)),
|
||||
None => {
|
||||
blocks.push(None);
|
||||
misses.push(bidx);
|
||||
misses.push(span);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -890,12 +946,12 @@ impl Core {
|
||||
let reader = self.revision_reader(uid, mtime, fsize).await?;
|
||||
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for &bidx in &misses {
|
||||
for &span in &misses {
|
||||
let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone());
|
||||
set.spawn(async move {
|
||||
core.fetch_block(&reader, &uid, mtime, fsize, bidx, cache_blocks)
|
||||
core.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks)
|
||||
.await
|
||||
.map(|bytes| (bidx, bytes))
|
||||
.map(|bytes| (span.idx, bytes))
|
||||
});
|
||||
}
|
||||
let mut out = Vec::with_capacity(misses.len());
|
||||
@@ -913,18 +969,20 @@ impl Core {
|
||||
// ahead of a sequential reader instead of stalling once it catches up.
|
||||
self.prefetch_ahead(uid, mtime, fsize, first, last, cache_blocks);
|
||||
|
||||
for (i, block) in blocks.into_iter().enumerate() {
|
||||
let bidx = first + i as u64;
|
||||
let bstart = bidx * BLOCK_SIZE;
|
||||
for (span, block) in wanted.into_iter().zip(blocks) {
|
||||
let bstart = span.start;
|
||||
// Every slot is populated: cache hits up front, misses by the fetch above.
|
||||
let block = block.expect("block fetched or cached");
|
||||
// Both sources guarantee the length the file's size implies — the
|
||||
// cache validates it on read, the network path in `read_block` — so
|
||||
// this refuses rather than quietly contributing fewer bytes than the
|
||||
// range covers, which is what turned a bad block into a truncated
|
||||
// file (bugs.md B84).
|
||||
if block.len() as u64 != block_len(fsize, bidx) {
|
||||
error!(%uid, bidx, len = block.len(), fsize, "assembling a read hit a short block");
|
||||
// Both sources guarantee the length the geometry says the block has
|
||||
// — the cache validates it on read, the network path in `read_block`
|
||||
// — so this refuses rather than quietly contributing fewer bytes
|
||||
// than the range covers, which is what turned a bad block into a
|
||||
// truncated file (bugs.md B84).
|
||||
if block.len() as u64 != span.len {
|
||||
error!(
|
||||
%uid, bidx = span.idx, len = block.len(), expected = span.len, fsize,
|
||||
"assembling a read hit a short block"
|
||||
);
|
||||
return Err(Errno::EIO);
|
||||
}
|
||||
let s = (offset.max(bstart) - bstart) as usize;
|
||||
@@ -1016,13 +1074,13 @@ impl Core {
|
||||
return None;
|
||||
}
|
||||
let mut out = Vec::with_capacity((end - offset) as usize);
|
||||
for bidx in (offset / BLOCK_SIZE)..=((end - 1) / BLOCK_SIZE) {
|
||||
let ring = self.block_ring.lock().get(uid, mtime, fsize, bidx);
|
||||
for span in self.block_geometry(uid, mtime, fsize).spans(offset, end) {
|
||||
let ring = self.block_ring.lock().get(uid, mtime, fsize, span);
|
||||
let block = match ring {
|
||||
Some(bytes) => bytes,
|
||||
None => Arc::new(self.cache.cached_block(uid, mtime, fsize, bidx)?),
|
||||
None => Arc::new(self.cache.cached_block(uid, mtime, fsize, span)?),
|
||||
};
|
||||
let bstart = bidx * BLOCK_SIZE;
|
||||
let bstart = span.start;
|
||||
let s = (offset.max(bstart) - bstart) as usize;
|
||||
let e = (end.min(bstart + block.len() as u64) - bstart) as usize;
|
||||
if s >= e {
|
||||
@@ -1142,31 +1200,42 @@ mod block_ring_tests {
|
||||
RING_BYTES / BLOCK_SIZE
|
||||
}
|
||||
|
||||
/// Block `idx` under the uniform geometry — the shape these tests are
|
||||
/// written against, since what they exercise is tagging and eviction rather
|
||||
/// than block layout.
|
||||
fn span(idx: u64) -> BlockSpan {
|
||||
BlockSpan {
|
||||
idx,
|
||||
start: idx * BLOCK_SIZE,
|
||||
len: BLOCK_SIZE,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stored_block_is_served_back() {
|
||||
let mut ring = BlockRing::default();
|
||||
let node = uid("a");
|
||||
ring.insert(&node, 7, 100, 3, block(0xab));
|
||||
assert_eq!(ring.get(&node, 7, 100, 3).map(|b| b[0]), Some(0xab));
|
||||
ring.insert(&node, 7, 100, span(3), block(0xab));
|
||||
assert_eq!(ring.get(&node, 7, 100, span(3)).map(|b| b[0]), Some(0xab));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_block_of_another_revision_is_never_served() {
|
||||
let mut ring = BlockRing::default();
|
||||
let node = uid("a");
|
||||
ring.insert(&node, 7, 100, 0, block(1));
|
||||
ring.insert(&node, 7, 100, span(0), block(1));
|
||||
// Same node, newer revision: the tag moved, so the old bytes go.
|
||||
assert!(ring.get(&node, 8, 100, 0).is_none());
|
||||
assert!(ring.get(&node, 7, 101, 0).is_none());
|
||||
assert!(ring.get(&node, 8, 100, span(0)).is_none());
|
||||
assert!(ring.get(&node, 7, 101, span(0)).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_new_revision_drops_what_the_old_one_left() {
|
||||
let mut ring = BlockRing::default();
|
||||
let node = uid("a");
|
||||
ring.insert(&node, 7, 100, 0, block(1));
|
||||
ring.insert(&node, 8, 100, 0, block(2));
|
||||
assert_eq!(ring.get(&node, 8, 100, 0).map(|b| b[0]), Some(2));
|
||||
ring.insert(&node, 7, 100, span(0), block(1));
|
||||
ring.insert(&node, 8, 100, span(0), block(2));
|
||||
assert_eq!(ring.get(&node, 8, 100, span(0)).map(|b| b[0]), Some(2));
|
||||
assert_eq!(ring.bytes, BLOCK_SIZE);
|
||||
}
|
||||
|
||||
@@ -1175,13 +1244,13 @@ mod block_ring_tests {
|
||||
let mut ring = BlockRing::default();
|
||||
let node = uid("a");
|
||||
for idx in 0..capacity() {
|
||||
ring.insert(&node, 1, u64::MAX, idx, block(idx as u8));
|
||||
ring.insert(&node, 1, u64::MAX, span(idx), block(idx as u8));
|
||||
}
|
||||
// Re-reading block 0 makes block 1 the oldest instead.
|
||||
assert!(ring.get(&node, 1, u64::MAX, 0).is_some());
|
||||
ring.insert(&node, 1, u64::MAX, capacity(), block(0xff));
|
||||
assert!(ring.get(&node, 1, u64::MAX, 0).is_some());
|
||||
assert!(ring.get(&node, 1, u64::MAX, 1).is_none());
|
||||
assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some());
|
||||
ring.insert(&node, 1, u64::MAX, span(capacity()), block(0xff));
|
||||
assert!(ring.get(&node, 1, u64::MAX, span(0)).is_some());
|
||||
assert!(ring.get(&node, 1, u64::MAX, span(1)).is_none());
|
||||
assert_eq!(ring.bytes, RING_BYTES);
|
||||
}
|
||||
|
||||
@@ -1189,9 +1258,9 @@ mod block_ring_tests {
|
||||
fn a_node_whose_last_block_ages_out_keeps_no_tag() {
|
||||
let mut ring = BlockRing::default();
|
||||
let (old, new) = (uid("old"), uid("new"));
|
||||
ring.insert(&old, 1, u64::MAX, 0, block(1));
|
||||
ring.insert(&old, 1, u64::MAX, span(0), block(1));
|
||||
for idx in 0..capacity() {
|
||||
ring.insert(&new, 1, u64::MAX, idx, block(2));
|
||||
ring.insert(&new, 1, u64::MAX, span(idx), block(2));
|
||||
}
|
||||
assert!(!ring.tags.contains_key(&old));
|
||||
assert!(ring.tags.contains_key(&new));
|
||||
@@ -1201,11 +1270,11 @@ mod block_ring_tests {
|
||||
fn dropping_a_node_frees_only_its_own_bytes() {
|
||||
let mut ring = BlockRing::default();
|
||||
let (a, b) = (uid("a"), uid("b"));
|
||||
ring.insert(&a, 1, 100, 0, block(1));
|
||||
ring.insert(&b, 1, 100, 0, block(2));
|
||||
ring.insert(&a, 1, 100, span(0), block(1));
|
||||
ring.insert(&b, 1, 100, span(0), block(2));
|
||||
ring.drop_node(&a);
|
||||
assert!(ring.get(&a, 1, 100, 0).is_none());
|
||||
assert_eq!(ring.get(&b, 1, 100, 0).map(|x| x[0]), Some(2));
|
||||
assert!(ring.get(&a, 1, 100, span(0)).is_none());
|
||||
assert_eq!(ring.get(&b, 1, 100, span(0)).map(|x| x[0]), Some(2));
|
||||
assert_eq!(ring.bytes, BLOCK_SIZE);
|
||||
}
|
||||
}
|
||||
|
||||
+364
-64
@@ -525,6 +525,9 @@ impl Core {
|
||||
continue;
|
||||
}
|
||||
trashed.push(rel.clone());
|
||||
if let Some(parent) = parent_uid_of(rel, &remote_dirs, &baseline) {
|
||||
self.invalidate_children_of(&parent);
|
||||
}
|
||||
let _ = self.db.sync_entry_remove(folder_id, rel);
|
||||
self.log_activity(
|
||||
ActivityKind::Trash,
|
||||
@@ -777,6 +780,9 @@ impl Core {
|
||||
outcome.errors += 1;
|
||||
continue;
|
||||
}
|
||||
if let Some(parent) = parent_uid_of(rel, &remote_dirs, &baseline) {
|
||||
self.invalidate_children_of(&parent);
|
||||
}
|
||||
let _ = self.db.sync_entry_remove(folder_id, rel);
|
||||
self.log_activity(
|
||||
ActivityKind::Trash,
|
||||
@@ -832,6 +838,9 @@ impl Core {
|
||||
outcome.errors += 1;
|
||||
continue;
|
||||
}
|
||||
if let Some(parent) = parent_uid_of(&rel, &remote_dirs, &baseline) {
|
||||
self.invalidate_children_of(&parent);
|
||||
}
|
||||
let _ = self.db.sync_entry_remove(folder_id, &rel);
|
||||
self.log_activity(
|
||||
ActivityKind::Trash,
|
||||
@@ -857,70 +866,9 @@ impl Core {
|
||||
writing: &HashSet<PathBuf>,
|
||||
out: &mut HashMap<String, LocalItem>,
|
||||
) -> Result<(), String> {
|
||||
let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?;
|
||||
for entry in entries {
|
||||
let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?;
|
||||
let path = entry.path();
|
||||
let meta = std::fs::symlink_metadata(&path)
|
||||
.map_err(|e| format!("stat {}: {e}", path.display()))?;
|
||||
let stripped = path.strip_prefix(root).map_err(|e| {
|
||||
format!(
|
||||
"local path {} escaped sync root {}: {e}",
|
||||
path.display(),
|
||||
root.display()
|
||||
)
|
||||
})?;
|
||||
let rel = stripped.to_str().ok_or_else(|| {
|
||||
format!(
|
||||
"local path {} is not valid UTF-8; refusing a destructive sync pass",
|
||||
path.display()
|
||||
)
|
||||
})?;
|
||||
// Ignore our own in-flight download temp files.
|
||||
if rel.contains(".pdfs-tmp-") {
|
||||
continue;
|
||||
}
|
||||
// Ignored paths are dropped here as well as from the classification
|
||||
// union — not for correctness (the union filter is what makes this
|
||||
// safe) but so an ignored `node_modules/` is never descended into or
|
||||
// counted against scan progress.
|
||||
if rules.is_ignored(rel, meta.is_dir()) {
|
||||
continue;
|
||||
}
|
||||
if meta.is_dir() {
|
||||
out.insert(
|
||||
rel.to_string(),
|
||||
LocalItem {
|
||||
is_dir: true,
|
||||
mtime: 0,
|
||||
mtime_ns: None,
|
||||
size: 0,
|
||||
open_for_write: false,
|
||||
},
|
||||
);
|
||||
self.progress_scanned(folder_id, base_name(rel));
|
||||
self.walk_local(folder_id, root, &path, rules, writing, out)?;
|
||||
} else if meta.is_file() {
|
||||
// Check whether any process holds this file open for writing.
|
||||
// The canonical path is used because /proc/*/fd links resolve
|
||||
// to canonical targets.
|
||||
let ofw = std::fs::canonicalize(&path)
|
||||
.map(|canon| writing.contains(&canon))
|
||||
.unwrap_or(false);
|
||||
out.insert(
|
||||
rel.to_string(),
|
||||
LocalItem {
|
||||
is_dir: false,
|
||||
mtime: system_mtime(&meta),
|
||||
mtime_ns: system_mtime_ns(&meta),
|
||||
size: meta.len() as i64,
|
||||
open_for_write: ofw,
|
||||
},
|
||||
);
|
||||
self.progress_scanned(folder_id, base_name(rel));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
walk_local_tree(root, dir, rules, writing, out, &mut |rel| {
|
||||
self.progress_scanned(folder_id, base_name(rel))
|
||||
})
|
||||
}
|
||||
|
||||
/// Map every remote folder under `folder` to its uid by rel path, ignoring
|
||||
@@ -1186,6 +1134,9 @@ impl Core {
|
||||
.create_folder(parent, base_name(rel), Some(now_secs()))
|
||||
.await
|
||||
.map_err(|e| format!("create remote folder {rel}: {e}"))?;
|
||||
// The parent has gained a child that no cached listing of it
|
||||
// knows about (B86).
|
||||
self.invalidate_children_of(parent);
|
||||
self.baseline_dir(folder_id, rel, &uid)?;
|
||||
Ok(Applied::Dir(rel.clone(), uid))
|
||||
}
|
||||
@@ -1295,6 +1246,10 @@ impl Core {
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("upload {rel}: {e}"))?;
|
||||
// The parent has gained a child that no cached listing of it knows
|
||||
// about. Without this, a folder later switched to on-demand serves a
|
||||
// snapshot that predates this upload (B86).
|
||||
self.invalidate_children_of(parent);
|
||||
warn_if_torn(rel, &path, streamed);
|
||||
self.record_file_baseline(folder_id, rel, Some(streamed), &uid)
|
||||
.await
|
||||
@@ -1546,6 +1501,95 @@ fn removed_locally(result: std::io::Result<()>) -> std::io::Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
/// The body of [`Core::walk_local`], without a `Core`: recursively collect the
|
||||
/// local tree under `dir` into `out`, reporting each entry to `scanned`.
|
||||
///
|
||||
/// **Every failure here is fatal to the pass, deliberately.** A `read_dir` that
|
||||
/// fails, an entry that cannot be stat-ed, a name that is not UTF-8 — each of
|
||||
/// them would, if skipped, leave `out` describing a *smaller* tree than the one
|
||||
/// on disk. Reconciliation reads a path present in the baseline and absent from
|
||||
/// the scan as "the user deleted this", and propagates that to Drive. So an
|
||||
/// unreadable subdirectory would trash its remote counterpart, and the deeper
|
||||
/// the failure, the more it takes with it (`docs/BUGS.md` B55). Refusing the
|
||||
/// whole pass keeps the previous baseline and leaves the remote alone; the next
|
||||
/// pass retries once the transient condition clears.
|
||||
///
|
||||
/// Split out from the method so exactly that can be tested against a real tree
|
||||
/// with a real unreadable directory in it, which needs no `Core` at all.
|
||||
fn walk_local_tree(
|
||||
root: &Path,
|
||||
dir: &Path,
|
||||
rules: &IgnoreRules,
|
||||
writing: &HashSet<PathBuf>,
|
||||
out: &mut HashMap<String, LocalItem>,
|
||||
scanned: &mut dyn FnMut(&str),
|
||||
) -> Result<(), String> {
|
||||
let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?;
|
||||
for entry in entries {
|
||||
let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?;
|
||||
let path = entry.path();
|
||||
let meta = std::fs::symlink_metadata(&path)
|
||||
.map_err(|e| format!("stat {}: {e}", path.display()))?;
|
||||
let stripped = path.strip_prefix(root).map_err(|e| {
|
||||
format!(
|
||||
"local path {} escaped sync root {}: {e}",
|
||||
path.display(),
|
||||
root.display()
|
||||
)
|
||||
})?;
|
||||
let rel = stripped.to_str().ok_or_else(|| {
|
||||
format!(
|
||||
"local path {} is not valid UTF-8; refusing a destructive sync pass",
|
||||
path.display()
|
||||
)
|
||||
})?;
|
||||
// Ignore our own in-flight download temp files.
|
||||
if rel.contains(".pdfs-tmp-") {
|
||||
continue;
|
||||
}
|
||||
// Ignored paths are dropped here as well as from the classification
|
||||
// union — not for correctness (the union filter is what makes this
|
||||
// safe) but so an ignored `node_modules/` is never descended into or
|
||||
// counted against scan progress.
|
||||
if rules.is_ignored(rel, meta.is_dir()) {
|
||||
continue;
|
||||
}
|
||||
if meta.is_dir() {
|
||||
out.insert(
|
||||
rel.to_string(),
|
||||
LocalItem {
|
||||
is_dir: true,
|
||||
mtime: 0,
|
||||
mtime_ns: None,
|
||||
size: 0,
|
||||
open_for_write: false,
|
||||
},
|
||||
);
|
||||
scanned(rel);
|
||||
walk_local_tree(root, &path, rules, writing, out, scanned)?;
|
||||
} else if meta.is_file() {
|
||||
// Check whether any process holds this file open for writing.
|
||||
// The canonical path is used because /proc/*/fd links resolve
|
||||
// to canonical targets.
|
||||
let ofw = std::fs::canonicalize(&path)
|
||||
.map(|canon| writing.contains(&canon))
|
||||
.unwrap_or(false);
|
||||
out.insert(
|
||||
rel.to_string(),
|
||||
LocalItem {
|
||||
is_dir: false,
|
||||
mtime: system_mtime(&meta),
|
||||
mtime_ns: system_mtime_ns(&meta),
|
||||
size: meta.len() as i64,
|
||||
open_for_write: ofw,
|
||||
},
|
||||
);
|
||||
scanned(rel);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Publish a conflict copy without overwriting an earlier conflict bearing the
|
||||
/// same timestamp. The original is removed only after the new copy is durable.
|
||||
fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result<PathBuf> {
|
||||
@@ -1579,6 +1623,26 @@ fn preserve_conflict_copy(path: &Path, stamp: i64) -> std::io::Result<PathBuf> {
|
||||
unreachable!("u32 conflict suffix space exhausted")
|
||||
}
|
||||
|
||||
/// The remote uid of `rel`'s parent folder: from the pass's folder-uid map
|
||||
/// first, and from the baseline second for a parent the map never covered.
|
||||
///
|
||||
/// Used to name the folder whose cached listing a remote mutation has just
|
||||
/// invalidated (B86). `None` is not an error — it only means this pass cannot
|
||||
/// say which folder to invalidate, and the caller skips it.
|
||||
fn parent_uid_of(
|
||||
rel: &str,
|
||||
remote_dirs: &HashMap<String, NodeUid>,
|
||||
baseline: &HashMap<String, StoredSyncEntry>,
|
||||
) -> Option<NodeUid> {
|
||||
let parent = parent_rel(rel);
|
||||
remote_dirs.get(parent).cloned().or_else(|| {
|
||||
baseline
|
||||
.get(parent)
|
||||
.and_then(|e| e.remote_uid.as_deref())
|
||||
.and_then(parse_uid)
|
||||
})
|
||||
}
|
||||
|
||||
fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
|
||||
match path.parent() {
|
||||
Some(parent) => std::fs::File::open(parent)?.sync_all(),
|
||||
@@ -1925,6 +1989,44 @@ mod tests {
|
||||
assert!(guard_local_wipe(&base, &fresh).is_err());
|
||||
}
|
||||
|
||||
/// B54's remaining case, end to end: a folder whose local root is still
|
||||
/// *readable* but is no longer the tree it was — an empty directory left
|
||||
/// where a mount used to be, or a different device mounted over it.
|
||||
///
|
||||
/// Nothing fails here. The scan succeeds and honestly reports an empty
|
||||
/// directory, so the only thing standing between the user and a trashed
|
||||
/// remote folder is the guard, and it has to fire for a one-file baseline
|
||||
/// exactly as it does for a large one.
|
||||
#[test]
|
||||
fn a_replaced_mountpoint_trips_the_guard_at_every_baseline_size() {
|
||||
let root = sync_test_dir("wipe-replaced-mount");
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut local: HashMap<String, LocalItem> = HashMap::new();
|
||||
let mut progress = no_progress();
|
||||
walk_local_tree(&root, &root, &rules, &writing, &mut local, &mut progress).unwrap();
|
||||
assert!(local.is_empty(), "the replacement root is genuinely empty");
|
||||
|
||||
for size in 1..=3 {
|
||||
let baseline: HashMap<String, ()> =
|
||||
(0..size).map(|i| (format!("file{i}.txt"), ())).collect();
|
||||
assert!(
|
||||
guard_local_wipe(&filter_baseline(&baseline, &rules), &local).is_err(),
|
||||
"a {size}-path baseline against an empty root is not a deletion"
|
||||
);
|
||||
}
|
||||
|
||||
// And the guard steps aside the moment the real tree is back, so a
|
||||
// remounted folder resumes rather than staying wedged.
|
||||
std::fs::write(root.join("file0.txt"), b"back").unwrap();
|
||||
let mut local: HashMap<String, LocalItem> = HashMap::new();
|
||||
walk_local_tree(&root, &root, &rules, &writing, &mut local, &mut progress).unwrap();
|
||||
let baseline: HashMap<String, ()> = [("file0.txt".to_string(), ())].into_iter().collect();
|
||||
assert!(guard_local_wipe(&filter_baseline(&baseline, &rules), &local).is_ok());
|
||||
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
}
|
||||
|
||||
/// A restored folder (features.md 5.2) starts with an empty baseline against
|
||||
/// an empty local directory and a full remote. That must reconcile as
|
||||
/// "download everything" rather than tripping the wipe guard — otherwise
|
||||
@@ -2126,6 +2228,204 @@ mod tests {
|
||||
std::fs::remove_dir_all(dir).unwrap();
|
||||
}
|
||||
|
||||
/// B49's remaining cases. Preservation is the step that makes a download
|
||||
/// safe to publish, so every way it can fail has to leave the local file
|
||||
/// where it was — the download is aborted, not the file replaced.
|
||||
#[test]
|
||||
fn conflict_preservation_failures_all_keep_the_local_file() {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
// An unwritable directory: the conflict copy cannot be created at all.
|
||||
let dir = sync_test_dir("conflict-eacces");
|
||||
let source = dir.join("notes.txt");
|
||||
std::fs::write(&source, b"the only copy").unwrap();
|
||||
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap();
|
||||
let denied = preserve_conflict_copy(&source, 1700);
|
||||
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||
assert!(denied.is_err(), "a read-only directory must fail the copy");
|
||||
assert_eq!(
|
||||
std::fs::read(&source).unwrap(),
|
||||
b"the only copy",
|
||||
"the local file survives byte-identical"
|
||||
);
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
|
||||
// A source that vanished between planning and preservation. The pass
|
||||
// must report it rather than publish the download over the gap.
|
||||
let dir = sync_test_dir("conflict-enoent");
|
||||
assert!(preserve_conflict_copy(&dir.join("gone.txt"), 1700).is_err());
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
|
||||
// A write error partway through the copy — induced by making the
|
||||
// destination name a directory, so `create_new` fails as something
|
||||
// other than `AlreadyExists`.
|
||||
let dir = sync_test_dir("conflict-eisdir");
|
||||
let source = dir.join("notes.txt");
|
||||
std::fs::write(&source, b"still the only copy").unwrap();
|
||||
std::fs::create_dir(conflict_path(&source, 1700)).unwrap();
|
||||
// Suffix 0 is taken by a directory; `create_new` on it reports
|
||||
// `AlreadyExists`, so preservation moves on to suffix 1 and succeeds.
|
||||
// What must not happen is the source being removed without a copy.
|
||||
let preserved = preserve_conflict_copy(&source, 1700).unwrap();
|
||||
assert!(preserved.is_file());
|
||||
assert_eq!(std::fs::read(&preserved).unwrap(), b"still the only copy");
|
||||
assert!(!source.exists());
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
}
|
||||
|
||||
/// A conflict copy keeps the bytes *and* the metadata that identifies them,
|
||||
/// so the next pass uploads the user's version rather than a file that looks
|
||||
/// freshly written.
|
||||
#[test]
|
||||
fn a_conflict_copy_is_byte_and_mtime_identical() {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
let dir = sync_test_dir("conflict-identity");
|
||||
let source = dir.join("notes.txt");
|
||||
std::fs::write(&source, b"local edit that must survive").unwrap();
|
||||
let before = std::fs::metadata(&source).unwrap();
|
||||
|
||||
let preserved = preserve_conflict_copy(&source, 1700).unwrap();
|
||||
let after = std::fs::metadata(&preserved).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
std::fs::read(&preserved).unwrap(),
|
||||
b"local edit that must survive"
|
||||
);
|
||||
assert_eq!(after.len(), before.len());
|
||||
assert_eq!(after.modified().unwrap(), before.modified().unwrap());
|
||||
assert_eq!(after.permissions().mode(), before.permissions().mode());
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
}
|
||||
|
||||
// ---- B55: an incomplete local scan is never a set of deletions ---------
|
||||
|
||||
fn no_progress() -> impl FnMut(&str) {
|
||||
|_: &str| {}
|
||||
}
|
||||
|
||||
/// A directory the pass cannot read must fail the pass. Omitting it would
|
||||
/// hand reconciliation a baseline-minus-subtree, which it reads as the user
|
||||
/// having deleted that subtree and propagates to Drive (B55).
|
||||
#[test]
|
||||
fn an_unreadable_subdirectory_fails_the_scan_instead_of_shrinking_it() {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
let root = sync_test_dir("scan-eacces");
|
||||
std::fs::write(root.join("keep.txt"), b"a").unwrap();
|
||||
let locked = root.join("locked");
|
||||
std::fs::create_dir(&locked).unwrap();
|
||||
std::fs::write(locked.join("inner.txt"), b"b").unwrap();
|
||||
std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut out = HashMap::new();
|
||||
let mut progress = no_progress();
|
||||
let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress);
|
||||
|
||||
std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
|
||||
let error = result.expect_err("an unreadable directory must abort the pass");
|
||||
assert!(
|
||||
error.contains("locked"),
|
||||
"the error should name the subtree it could not read: {error}"
|
||||
);
|
||||
}
|
||||
|
||||
/// The same rule one level up: if the sync root itself cannot be read, the
|
||||
/// pass sees nothing, and "nothing" must never mean "delete everything".
|
||||
#[test]
|
||||
fn an_unreadable_root_fails_the_scan() {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
let root = sync_test_dir("scan-root-eacces");
|
||||
std::fs::write(root.join("keep.txt"), b"a").unwrap();
|
||||
std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut out = HashMap::new();
|
||||
let mut progress = no_progress();
|
||||
let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress);
|
||||
|
||||
std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)).unwrap();
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
|
||||
assert!(result.is_err());
|
||||
assert!(out.is_empty());
|
||||
}
|
||||
|
||||
/// A root that is gone entirely — an unmounted device folder, a removed
|
||||
/// directory — is the sharpest version of the same case: every baseline path
|
||||
/// is "missing", and the scan must refuse rather than report an empty tree.
|
||||
#[test]
|
||||
fn an_absent_root_fails_the_scan() {
|
||||
let root = sync_test_dir("scan-absent");
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut out = HashMap::new();
|
||||
let mut progress = no_progress();
|
||||
assert!(walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress).is_err());
|
||||
assert!(out.is_empty());
|
||||
}
|
||||
|
||||
/// A name that is not valid UTF-8 cannot be keyed into the scan, so it
|
||||
/// cannot be compared against the baseline either. Refusing the pass is the
|
||||
/// only safe reading: skipping it would classify the file as deleted.
|
||||
#[test]
|
||||
fn a_non_utf8_name_fails_the_scan() {
|
||||
use std::ffi::OsStr;
|
||||
use std::os::unix::ffi::OsStrExt as _;
|
||||
|
||||
let root = sync_test_dir("scan-non-utf8");
|
||||
std::fs::write(root.join(OsStr::from_bytes(b"bad\xffname")), b"x").unwrap();
|
||||
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut out = HashMap::new();
|
||||
let mut progress = no_progress();
|
||||
let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut progress);
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
|
||||
let error = result.expect_err("a name the scan cannot key must abort the pass");
|
||||
assert!(error.contains("UTF-8"), "{error}");
|
||||
}
|
||||
|
||||
/// The complement: a scan that *can* read everything reports everything, so
|
||||
/// the strictness above does not come at the cost of a working pass. A
|
||||
/// symlink is skipped by design rather than by failure, and a dangling one
|
||||
/// stats fine through `symlink_metadata`.
|
||||
#[test]
|
||||
fn a_readable_tree_scans_completely() {
|
||||
let root = sync_test_dir("scan-complete");
|
||||
std::fs::write(root.join("a.txt"), b"aa").unwrap();
|
||||
std::fs::create_dir(root.join("sub")).unwrap();
|
||||
std::fs::write(root.join("sub/b.txt"), b"bbb").unwrap();
|
||||
std::os::unix::fs::symlink("nowhere", root.join("dangling")).unwrap();
|
||||
|
||||
let rules = rules_for(&[]);
|
||||
let writing = HashSet::new();
|
||||
let mut out = HashMap::new();
|
||||
let mut seen: Vec<String> = Vec::new();
|
||||
let result = walk_local_tree(&root, &root, &rules, &writing, &mut out, &mut |rel| {
|
||||
seen.push(rel.to_string())
|
||||
});
|
||||
std::fs::remove_dir_all(&root).unwrap();
|
||||
|
||||
result.unwrap();
|
||||
let mut keys: Vec<&str> = out.keys().map(String::as_str).collect();
|
||||
keys.sort_unstable();
|
||||
assert_eq!(keys, vec!["a.txt", "sub", "sub/b.txt"]);
|
||||
assert_eq!(out["a.txt"].size, 2);
|
||||
assert_eq!(out["sub/b.txt"].size, 3);
|
||||
assert!(out["sub"].is_dir);
|
||||
assert_eq!(seen.len(), 3, "every reported entry ticks scan progress");
|
||||
}
|
||||
|
||||
fn sync_test_dir(label: &str) -> PathBuf {
|
||||
let nonce = SystemTime::now()
|
||||
.duration_since(SystemTime::UNIX_EPOCH)
|
||||
|
||||
Reference in new issue
Block a user