Compare commits

..
Author SHA1 Message Date
nvrl 0d8a474442 version bump 2026-07-22 15:11:33 +02:00
nvrl 8be0f04b61 fixed fuse bugs from BUGS.md
- Durable scratch/staging publication and preservation on failure.
      - Transactional pending-operation supersession.
      - Conflict copies cannot overwrite older copies or lose the source.
      - Single-entry wipe protection and incomplete-scan aborts.
      - Failed local deletion retains its baseline.
      - Downloads are fsynced before baseline settlement.
      - Future database schemas are refused.
      - Event cursors advance only after successful invalidation and persistence.
      - IPC frame, timeout, and concurrency limits.
      - Atomic config saves; malformed configs are preserved.
      - Sensitive directories require correct ownership, real directories, and 0700.
      - Debian package now includes service and tray autostart files.
      - Release workflow enforces versions, tests, lint, formatting, and offline acceptance.
      - CI now runs offline filesystem acceptance.
2026-07-22 14:49:06 +02:00
nvrl a76f36bdff fixed fs bugs + compleance test for fuse fs 2026-07-22 14:24:40 +02:00
nvrl d0b49d3c8c fixed a truncate defect 2026-07-22 11:50:58 +02:00
nvrl 25f578161d added more comprehensive testing 2026-07-22 11:43:17 +02:00
nvrl ea56818a80 prompt refactor 2026-07-22 11:27:57 +02:00
nvrl 39ab4aae48 prompt refactor 2026-07-22 09:50:42 +02:00
nvrl 6cba77d3cb Merge pull request #15 from narrrl/fix/ioctl-handler
Fix/ioctl handler
2026-07-22 03:07:30 +02:00
nvrl 714b3048dc Merge pull request #14 from narrrl/fix/ioctl-handler
filesystem fixes, critical + feature implementations
2026-07-22 02:30:57 +02:00
32 changed files with 3659 additions and 438 deletions

No files matched your search

+3
View File
@@ -44,6 +44,9 @@ jobs:
- name: Unit tests - name: Unit tests
run: cargo test --workspace --locked run: cargo test --workspace --locked
- name: Account-free filesystem acceptance
run: scripts/fuse-acceptance.sh --offline-only
- name: Acceptance runner safety guard - name: Acceptance runner safety guard
run: | run: |
if scripts/fuse-acceptance.sh /tmp >acceptance.out 2>&1; then if scripts/fuse-acceptance.sh /tmp >acceptance.out 2>&1; then
+23 -1
View File
@@ -35,11 +35,29 @@ jobs:
- name: Set up Rust - name: Set up Rust
uses: actions-rust-lang/setup-rust-toolchain@v1 uses: actions-rust-lang/setup-rust-toolchain@v1
with: with:
toolchain: stable toolchain: '1.96'
components: rustfmt, clippy
- name: Cache Cargo target - name: Cache Cargo target
uses: swatinem/rust-cache@v2 uses: swatinem/rust-cache@v2
- name: Verify tag and manifest versions
if: startsWith(github.ref, 'refs/tags/')
shell: bash
run: |
tag_version="${GITHUB_REF_NAME#v}"
manifest_version="$(sed -n 's/^version = "\([^"]*\)"/\1/p' Cargo.toml | head -n1)"
pkgbuild_version="$(sed -n 's/^pkgver=//p' packaging/PKGBUILD | head -n1)"
test "$tag_version" = "$manifest_version"
test "$tag_version" = "$pkgbuild_version"
- name: Release quality gates
run: |
cargo fmt --all -- --check
cargo clippy --workspace --all-targets -- -D warnings
cargo test --workspace --locked
scripts/fuse-acceptance.sh --offline-only
- name: Build release binaries - name: Build release binaries
run: | run: |
cargo build --release --locked cargo build --release --locked
@@ -76,6 +94,8 @@ jobs:
mkdir -p deb/usr/share/applications mkdir -p deb/usr/share/applications
mkdir -p deb/usr/share/icons/hicolor/scalable/apps mkdir -p deb/usr/share/icons/hicolor/scalable/apps
mkdir -p deb/usr/share/licenses/proton-drive-linux mkdir -p deb/usr/share/licenses/proton-drive-linux
mkdir -p deb/usr/lib/systemd/user
mkdir -p deb/etc/xdg/autostart
# Copy binaries # Copy binaries
cp target/release/pdfs deb/usr/bin/ cp target/release/pdfs deb/usr/bin/
@@ -85,7 +105,9 @@ jobs:
# Copy desktop integration files # Copy desktop integration files
cp packaging/io.narl.proton-drive-linux.desktop deb/usr/share/applications/ cp packaging/io.narl.proton-drive-linux.desktop deb/usr/share/applications/
cp packaging/io.narl.proton-drive-linux-tray.desktop deb/etc/xdg/autostart/
cp packaging/io.narl.proton-drive-linux.svg deb/usr/share/icons/hicolor/scalable/apps/ cp packaging/io.narl.proton-drive-linux.svg deb/usr/share/icons/hicolor/scalable/apps/
cp packaging/proton-drive.service deb/usr/lib/systemd/user/
cp LICENSE deb/usr/share/licenses/proton-drive-linux/ cp LICENSE deb/usr/share/licenses/proton-drive-linux/
# Calculate Installed-Size in KB for the control file # Calculate Installed-Size in KB for the control file
Generated
+5 -4
View File
@@ -3518,7 +3518,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
[[package]] [[package]]
name = "pdfs-cli" name = "pdfs-cli"
version = "0.7.1" version = "1.0.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"clap 4.6.2", "clap 4.6.2",
@@ -3533,11 +3533,12 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-core" name = "pdfs-core"
version = "0.7.1" version = "1.0.0"
dependencies = [ dependencies = [
"directories", "directories",
"ignore", "ignore",
"keyring", "keyring",
"libc",
"parking_lot", "parking_lot",
"proton-drive-rs", "proton-drive-rs",
"proton-sdk", "proton-sdk",
@@ -3552,7 +3553,7 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-fuse" name = "pdfs-fuse"
version = "0.7.1" version = "1.0.0"
dependencies = [ dependencies = [
"fuser", "fuser",
"image", "image",
@@ -3568,7 +3569,7 @@ dependencies = [
[[package]] [[package]]
name = "pdfs-gui" name = "pdfs-gui"
version = "0.7.1" version = "1.0.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-channel", "async-channel",
+1 -1
View File
@@ -3,7 +3,7 @@ resolver = "2"
members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"] members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
[workspace.package] [workspace.package]
version = "0.7.1" version = "1.0.0"
edition = "2024" edition = "2024"
license = "MIT" license = "MIT"
authors = ["Nils Pukropp <contact@narl.io>"] authors = ["Nils Pukropp <contact@narl.io>"]
+1
View File
@@ -21,3 +21,4 @@ sha2.workspace = true
rusqlite.workspace = true rusqlite.workspace = true
keyring.workspace = true keyring.workspace = true
parking_lot.workspace = true parking_lot.workspace = true
libc.workspace = true
+122 -33
View File
@@ -241,7 +241,9 @@ impl ContentCache {
let recovery_dir = content_dir.join("recovery"); let recovery_dir = content_dir.join("recovery");
std::fs::create_dir_all(&recovery_dir)?; std::fs::create_dir_all(&recovery_dir)?;
Self::rescue_scratch(&scratch_dir, &recovery_dir); Self::rescue_scratch(&scratch_dir, &recovery_dir);
let _ = std::fs::remove_dir_all(&scratch_dir); // A failed rescue may leave the only durable copy here. Never clear the
// directory wholesale; discard only blobs that carry no fsync marker.
Self::discard_unmarked_scratch(&scratch_dir);
std::fs::create_dir_all(&scratch_dir)?; std::fs::create_dir_all(&scratch_dir)?;
// Staging, by contrast, is deliberately preserved across runs. // Staging, by contrast, is deliberately preserved across runs.
let staging_dir = content_dir.join("staging"); let staging_dir = content_dir.join("staging");
@@ -669,18 +671,24 @@ impl ContentCache {
pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> { pub fn create_scratch(&self) -> Result<(std::fs::File, PathBuf)> {
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
static N: AtomicU64 = AtomicU64::new(0); static N: AtomicU64 = AtomicU64::new(0);
let path = self.scratch_dir.join(format!( loop {
"w-{}-{}", let path = self.scratch_dir.join(format!(
std::process::id(), "w-{}-{}-{}",
N.fetch_add(1, Ordering::Relaxed) std::process::id(),
)); now_millis(),
let file = std::fs::OpenOptions::new() N.fetch_add(1, Ordering::Relaxed)
.create(true) ));
.truncate(true) match std::fs::OpenOptions::new()
.read(true) .create_new(true)
.write(true) .read(true)
.open(&path)?; .write(true)
Ok((file, path)) .open(&path)
{
Ok(file) => return Ok((file, path)),
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(e) => return Err(e.into()),
}
}
} }
/// The sidecar path for a scratch file, holding the [`StagedWrite`] that /// The sidecar path for a scratch file, holding the [`StagedWrite`] that
@@ -709,16 +717,23 @@ impl ContentCache {
/// have synced `scratch` first: a sidecar promising bytes that never reached /// have synced `scratch` first: a sidecar promising bytes that never reached
/// the disk is worse than no sidecar. /// the disk is worse than no sidecar.
pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> { pub fn mark_scratch_durable(&self, scratch: &Path, meta: &StagedWrite) -> Result<()> {
use std::io::Write as _;
let side = Self::scratch_sidecar(scratch); let side = Self::scratch_sidecar(scratch);
let tmp = side.with_extension("json.tmp"); let tmp = side.with_extension("json.tmp");
std::fs::write(&tmp, serde_json::to_vec(meta)?)?; let mut file = std::fs::OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
file.write_all(&serde_json::to_vec(meta)?)?;
file.sync_all()?;
drop(file);
std::fs::rename(&tmp, &side)?; std::fs::rename(&tmp, &side)?;
// The rename is what publishes the sidecar; fsync the directory so the // The rename is what publishes the sidecar; fsync the directory so the
// rename itself survives, not just the bytes it points at. // rename itself survives, not just the bytes it points at.
if let Some(dir) = side.parent() if let Some(dir) = side.parent() {
&& let Ok(d) = std::fs::File::open(dir) std::fs::File::open(dir)?.sync_all()?;
{
let _ = d.sync_all();
} }
Ok(()) Ok(())
} }
@@ -796,6 +811,21 @@ impl ContentCache {
} }
} }
fn discard_unmarked_scratch(scratch_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;
}
if !Self::scratch_sidecar(&blob).exists() {
let _ = std::fs::remove_file(blob);
}
}
}
/// Emergency eviction of unpinned blobs when local disk runs out of space (ENOSPC). /// Emergency eviction of unpinned blobs when local disk runs out of space (ENOSPC).
pub fn emergency_evict(&self) { pub fn emergency_evict(&self) {
let pinned: HashSet<String> = self let pinned: HashSet<String> = self
@@ -826,7 +856,8 @@ impl ContentCache {
/// Writes rescued from an unclean shutdown, as `(blob, meta)` pairs for the /// 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 /// daemon to hand to the upload queue. A blob whose sidecar did not survive
/// the rescue is deleted here — nothing can address it. /// is not actionable automatically, but is retained for manual recovery:
/// unexplained bytes are never evidence that user data is disposable.
pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> { pub fn recovered_writes(&self) -> Vec<(PathBuf, StagedWrite)> {
let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else { let Ok(entries) = std::fs::read_dir(&self.recovery_dir) else {
return Vec::new(); return Vec::new();
@@ -837,14 +868,11 @@ impl ContentCache {
if blob.extension().is_some_and(|e| e == "json") { if blob.extension().is_some_and(|e| e == "json") {
continue; continue;
} }
match std::fs::read(Self::scratch_sidecar(&blob)) if let Some(meta) = std::fs::read(Self::scratch_sidecar(&blob))
.ok() .ok()
.and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok()) .and_then(|b| serde_json::from_slice::<StagedWrite>(&b).ok())
{ {
Some(meta) => out.push((blob, meta)), out.push((blob, meta));
None => {
let _ = std::fs::remove_file(&blob);
}
} }
} }
out out
@@ -872,6 +900,8 @@ impl ContentCache {
/// failure here means we could not save the bytes at all, so it is reported /// failure here means we could not save the bytes at all, so it is reported
/// rather than swallowed. /// rather than swallowed.
pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result<PathBuf> { pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result<PathBuf> {
use std::io::Write as _;
static N: AtomicU64 = AtomicU64::new(0); static N: AtomicU64 = AtomicU64::new(0);
// The uid goes in the name so a staged file can be tied back to its node // 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. // without a database; `/` in a uid would otherwise open a path.
@@ -882,18 +912,32 @@ impl ContentCache {
now_millis(), now_millis(),
N.fetch_add(1, Ordering::Relaxed) N.fetch_add(1, Ordering::Relaxed)
)); ));
if let Err(e) = std::fs::rename(scratch, &path) { let copied = if let Err(e) = std::fs::rename(scratch, &path) {
// Cross-device: rename cannot move it, so copy and drop the original. if e.kind() != std::io::ErrorKind::CrossesDevices {
let _ = e; return Err(e.into());
}
std::fs::copy(scratch, &path)?; std::fs::copy(scratch, &path)?;
let _ = std::fs::remove_file(scratch); true
} } else {
false
};
std::fs::File::open(&path)?.sync_all()?;
// Sidecar last: a staged file without one is still recoverable bytes, // Sidecar last: a staged file without one is still recoverable bytes,
// while a sidecar without a file would describe nothing. // while a sidecar without a file would describe nothing.
std::fs::write( let side = path.with_extension("json");
path.with_extension("json"), let side_tmp = path.with_extension("json.tmp");
serde_json::to_vec_pretty(meta)?, let mut side_file = std::fs::OpenOptions::new()
)?; .write(true)
.create_new(true)
.open(&side_tmp)?;
side_file.write_all(&serde_json::to_vec_pretty(meta)?)?;
side_file.sync_all()?;
drop(side_file);
std::fs::rename(&side_tmp, &side)?;
std::fs::File::open(&self.staging_dir)?.sync_all()?;
if copied {
std::fs::remove_file(scratch)?;
}
Ok(path) Ok(path)
} }
@@ -1906,6 +1950,51 @@ mod tests {
assert!(c2.recovered_writes().is_empty(), "recovery does not repeat"); assert!(c2.recovered_writes().is_empty(), "recovery does not repeat");
} }
#[test]
fn recovery_keeps_an_orphan_blob_when_its_sidecar_is_missing() {
let (c, dir) = cache();
let orphan = dir.path().join("content/recovery/orphaned-user-bytes");
std::fs::write(&orphan, b"retain me").unwrap();
assert!(c.recovered_writes().is_empty());
assert_eq!(
std::fs::read(&orphan).unwrap(),
b"retain me",
"metadata loss must not trigger deletion of unexplained content"
);
}
#[test]
fn failed_scratch_rescue_does_not_delete_the_durable_source() {
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 file, scratch) = c.create_scratch().unwrap();
file.write_all(b"durable source").unwrap();
file.sync_all().unwrap();
let meta = StagedWrite {
uid: uid("rescue").to_string(),
len: 14,
base_size: 0,
base_mtime: 0,
authored: vec![(0, 14)],
complete: true,
based_on: None,
};
c.mark_scratch_durable(&scratch, &meta).unwrap();
let blocking_destination = content.join("recovery").join(scratch.file_name().unwrap());
std::fs::create_dir(&blocking_destination).unwrap();
drop(file);
drop(c);
let _c2 = ContentCache::open(content, pins, 0, db).unwrap();
assert_eq!(std::fs::read(&scratch).unwrap(), b"durable source");
assert!(ContentCache::scratch_sidecar(&scratch).exists());
}
/// Releasing a write hands the bytes to staging, so the sidecar an earlier /// 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. /// `fsync` left must not offer recovery a second, stale copy of it.
#[test] #[test]
+95 -24
View File
@@ -105,11 +105,21 @@ impl AppDirs {
/// Load config from disk, creating default if missing. /// Load config from disk, creating default if missing.
pub fn load_config(&self) -> AppConfig { pub fn load_config(&self) -> AppConfig {
let path = self.config_path(); let path = self.config_path();
if path.exists() if path.exists() {
&& let Ok(content) = std::fs::read_to_string(&path) match std::fs::read_to_string(&path)
&& let Ok(config) = serde_json::from_str::<AppConfig>(&content) .map_err(Error::from)
{ .and_then(|content| {
return config; serde_json::from_str::<AppConfig>(&content).map_err(Error::from)
}) {
Ok(config) => return config,
Err(e) => {
// Preserve the original bytes for diagnosis/recovery. In
// particular, never replace a truncated config with defaults
// and silently lose mount, device, or ignore settings.
tracing::error!(path = %path.display(), error = %e, "config is unreadable; using in-memory defaults without overwriting it");
return AppConfig::default();
}
}
} }
let default_config = AppConfig::default(); let default_config = AppConfig::default();
@@ -125,10 +135,26 @@ impl AppDirs {
/// cache budget or mountpoint; the daemon re-reads the file on its next mount. /// cache budget or mountpoint; the daemon re-reads the file on its next mount.
pub fn save_config(&self, config: &AppConfig) -> Result<()> { pub fn save_config(&self, config: &AppConfig) -> Result<()> {
let path = self.config_path(); let path = self.config_path();
if let Some(parent) = path.parent() { let parent = path
std::fs::create_dir_all(parent)?; .parent()
} .ok_or_else(|| Error::Other("configuration path has no parent".into()))?;
std::fs::write(&path, serde_json::to_string_pretty(config)?)?; std::fs::create_dir_all(parent)?;
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or(0);
let tmp = path.with_extension(format!("json.tmp-{}-{nonce}", std::process::id()));
use std::os::unix::fs::OpenOptionsExt as _;
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&tmp)?;
use std::io::Write as _;
file.write_all(&serde_json::to_vec_pretty(config)?)?;
file.sync_all()?;
std::fs::rename(&tmp, &path)?;
std::fs::File::open(parent)?.sync_all()?;
Ok(()) Ok(())
} }
@@ -211,24 +237,41 @@ impl AppDirs {
pub fn ensure(&self) -> Result<()> { pub fn ensure(&self) -> Result<()> {
for dir in [self.state_dir(), self.cache_dir()] { for dir in [self.state_dir(), self.cache_dir()] {
std::fs::create_dir_all(&dir)?; std::fs::create_dir_all(&dir)?;
restrict_dir(&dir); restrict_dir(&dir)?;
} }
let config_dir = self.dirs.config_dir().to_path_buf(); let config_dir = self.dirs.config_dir().to_path_buf();
if std::fs::create_dir_all(&config_dir).is_ok() { std::fs::create_dir_all(&config_dir)?;
restrict_dir(&config_dir); restrict_dir(&config_dir)?;
}
Ok(()) Ok(())
} }
} }
/// Make `dir` owner-only (`0700`). Best effort: a mode we cannot set (an /// Make `dir` owner-only (`0700`). These directories contain decrypted bytes,
/// exotic filesystem, a directory we do not own) is not worth refusing to /// plaintext names, and an authenticated control socket, so failure is fatal.
/// start over, and the caller has no better answer than continuing. fn restrict_dir(dir: &Path) -> Result<()> {
fn restrict_dir(dir: &Path) { use std::os::unix::fs::{MetadataExt as _, PermissionsExt};
use std::os::unix::fs::PermissionsExt; let before = std::fs::symlink_metadata(dir)?;
if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) { if before.file_type().is_symlink() || !before.is_dir() {
tracing::warn!(path = %dir.display(), error = %e, "could not restrict directory to 0700"); return Err(Error::Other(format!(
"sensitive path is not a real directory: {}",
dir.display()
)));
} }
if before.uid() != unsafe { libc::geteuid() } {
return Err(Error::Other(format!(
"sensitive directory is not owned by the current user: {}",
dir.display()
)));
}
std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
let after = std::fs::symlink_metadata(dir)?;
if after.permissions().mode() & 0o777 != 0o700 {
return Err(Error::Other(format!(
"sensitive directory did not become owner-only: {}",
dir.display()
)));
}
Ok(())
} }
/// Make a just-bound Unix socket owner-only (`0600`). /// Make a just-bound Unix socket owner-only (`0600`).
@@ -263,29 +306,57 @@ mod tests {
/// and the control socket that commands the daemon's session. /// and the control socket that commands the daemon's session.
#[test] #[test]
fn restrict_dir_makes_a_directory_owner_only() { fn restrict_dir_makes_a_directory_owner_only() {
let dir = std::env::temp_dir().join(format!("pdfs-cfg-{}", std::process::id())); let dir = std::env::temp_dir().join(format!(
"pdfs-cfg-{}-{:?}",
std::process::id(),
std::thread::current().id()
));
let _ = std::fs::remove_dir_all(&dir); let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap(); std::fs::create_dir_all(&dir).unwrap();
// Start deliberately world-readable, as `create_dir_all` under a typical // Start deliberately world-readable, as `create_dir_all` under a typical
// umask would leave it. // umask would leave it.
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap(); std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)).unwrap();
restrict_dir(&dir); restrict_dir(&dir).unwrap();
assert_eq!(mode_of(&dir), 0o700, "group and other must have no access"); assert_eq!(mode_of(&dir), 0o700, "group and other must have no access");
let _ = std::fs::remove_dir_all(&dir); let _ = std::fs::remove_dir_all(&dir);
} }
#[test]
fn restrict_dir_refuses_a_symlink() {
let root = std::env::temp_dir().join(format!(
"pdfs-cfg-link-{}-{:?}",
std::process::id(),
std::thread::current().id()
));
let target = root.join("target");
let link = root.join("state");
std::fs::create_dir_all(&target).unwrap();
std::os::unix::fs::symlink(&target, &link).unwrap();
assert!(restrict_dir(&link).is_err());
let _ = std::fs::remove_dir_all(&root);
}
/// Connecting to the control socket confers the daemon's authenticated /// Connecting to the control socket confers the daemon's authenticated
/// session with no credential, so it is an authority boundary rather than /// session with no credential, so it is an authority boundary rather than
/// merely private data. /// merely private data.
#[test] #[test]
fn restrict_socket_makes_a_socket_owner_only() { fn restrict_socket_makes_a_socket_owner_only() {
let dir = std::env::temp_dir().join(format!("pdfs-sock-{}", std::process::id())); let dir = std::env::temp_dir().join(format!(
"pdfs-sock-{}-{:?}",
std::process::id(),
std::thread::current().id()
));
let _ = std::fs::remove_dir_all(&dir); let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap(); std::fs::create_dir_all(&dir).unwrap();
let sock = dir.join("control.sock"); let sock = dir.join("control.sock");
let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap(); // chmod semantics are identical for a socket path and a regular file;
// use a file so this regression also runs in sandboxes that prohibit
// creating Unix sockets.
let listener = std::fs::File::create(&sock).unwrap();
restrict_socket(&sock).unwrap(); restrict_socket(&sock).unwrap();
+327
View File
@@ -101,6 +101,19 @@ pub enum Request {
/// Independent of [`Request::Search`] so a front-end can fire both at once and /// Independent of [`Request::Search`] so a front-end can fire both at once and
/// render whichever lands first. Replies with [`Response::LocalResults`]. /// render whichever lands first. Replies with [`Response::LocalResults`].
SearchLocal { query: String, limit: usize }, SearchLocal { query: String, limit: usize },
/// Search both the Drive metadata index and the local-file index in one
/// round-trip. This is the preferred prompt API; the two legacy search
/// requests remain available for older clients. `limit` applies to each
/// source independently. `filters` is applied before limiting, so selecting
/// (for example) images cannot hide valid matches below an unfiltered top
/// `limit`. Older clients omit it and therefore search both sources and all
/// kinds. Replies with [`Response::SearchResultsV2`].
SearchV2 {
query: String,
limit: usize,
#[serde(default)]
filters: SearchFilters,
},
/// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a /// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a
/// single path component (no separators). Replies with [`Response::Ok`]. /// single path component (no separators). Replies with [`Response::Ok`].
Rename { path: String, new_name: String }, Rename { path: String, new_name: String },
@@ -665,6 +678,15 @@ pub struct SearchHit {
pub pinned: bool, pub pinned: bool,
/// Node uid in `volume~link` form. /// Node uid in `volume~link` form.
pub uid: String, pub uid: String,
/// Absolute path through the most specific active on-demand mount, when the
/// node is covered by one. Clients should prefer this for folders and
/// streamable media, and fall back to `mountpoint/path` when absent.
#[serde(default)]
pub mounted_path: Option<String>,
/// Relevance assigned by the daemon. Higher scores sort first. Legacy
/// daemons omit it, in which case clients retain the daemon's result order.
#[serde(default)]
pub score: i64,
} }
/// One hit in a [`Request::SearchLocal`] result: a file on this machine, outside /// One hit in a [`Request::SearchLocal`] result: a file on this machine, outside
@@ -679,6 +701,107 @@ pub struct LocalHit {
pub size: u64, pub size: u64,
/// Modification time, epoch seconds. /// Modification time, epoch seconds.
pub modified: i64, pub modified: i64,
/// Relevance assigned by the daemon. Higher scores sort first. Legacy
/// daemons omit it, in which case clients retain the daemon's result order.
#[serde(default)]
pub score: i64,
}
/// A searchable corpus in [`Request::SearchV2`]. Kept explicit on the wire so
/// future sources can be added without multiplying request variants.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum SearchSource {
Drive,
Local,
}
/// Coarse result kind used by the prompt's facets. Classification belongs in
/// core/the daemon and must happen before the per-source limit is applied.
#[derive(Serialize, Deserialize, Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
pub enum SearchKind {
#[default]
All,
Folders,
Documents,
Images,
Media,
}
impl SearchKind {
/// Classify a result before the caller applies its final limit. This stays
/// beside the wire enum so every front end and both search corpora agree.
pub fn accepts(self, name: &str, is_dir: bool) -> bool {
if self == Self::All {
return true;
}
if self == Self::Folders {
return is_dir;
}
if is_dir {
return false;
}
let extension = std::path::Path::new(name)
.extension()
.and_then(|value| value.to_str())
.unwrap_or_default()
.to_ascii_lowercase();
match self {
Self::Documents => matches!(
extension.as_str(),
"pdf"
| "doc"
| "docx"
| "odt"
| "rtf"
| "txt"
| "md"
| "xls"
| "xlsx"
| "ods"
| "csv"
| "ppt"
| "pptx"
| "odp"
| "epub"
),
Self::Images => matches!(
extension.as_str(),
"png" | "jpg" | "jpeg" | "webp" | "gif" | "bmp" | "svg" | "avif" | "heic" | "tiff"
),
Self::Media => {
PhotoKind::classify(Some(name), None) == PhotoKind::Video
|| matches!(
extension.as_str(),
"mp3" | "flac" | "wav" | "ogg" | "opus" | "m4a" | "aac"
)
}
Self::All | Self::Folders => unreachable!(),
}
}
}
/// Optional constraints for a unified search. The default deliberately means
/// the original SearchV2 behaviour, preserving old JSON requests.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct SearchFilters {
/// Sources to query. An explicit empty list searches neither source.
#[serde(default = "default_search_sources")]
pub sources: Vec<SearchSource>,
#[serde(default)]
pub kind: SearchKind,
}
impl Default for SearchFilters {
fn default() -> Self {
Self {
sources: default_search_sources(),
kind: SearchKind::All,
}
}
}
fn default_search_sources() -> Vec<SearchSource> {
vec![SearchSource::Drive, SearchSource::Local]
} }
/// What kind of media a timeline entry is, so the Photos page can split into /// What kind of media a timeline entry is, so the Photos page can split into
@@ -947,6 +1070,14 @@ pub enum Response {
/// is true while a scan of the machine is still running, so a front-end can /// is true while a scan of the machine is still running, so a front-end can
/// say "still indexing" instead of "no matches" on a cold first launch. /// say "still indexing" instead of "no matches" on a cold first launch.
LocalResults { hits: Vec<LocalHit>, indexing: bool }, LocalResults { hits: Vec<LocalHit>, indexing: bool },
/// Drive and local-file search results returned together (reply to
/// [`Request::SearchV2`]). `local_indexing` distinguishes a genuinely empty
/// local result set from one observed while the initial scan is running.
SearchResultsV2 {
drive_hits: Vec<SearchHit>,
local_hits: Vec<LocalHit>,
local_indexing: bool,
},
/// A snapshot of what the daemon is working on (reply to /// A snapshot of what the daemon is working on (reply to
/// [`Request::GetQueueStatus`]): `items` are byte-moving transfers, `jobs` /// [`Request::GetQueueStatus`]): `items` are byte-moving transfers, `jobs`
/// the longer non-transfer work around them (scans, folder skeletons, the /// the longer non-transfer work around them (scans, folder skeletons, the
@@ -1104,6 +1235,58 @@ mod tests {
assert_eq!(PhotoKind::classify(Some("x.jpg"), Some("video/mp4")), Photo); assert_eq!(PhotoKind::classify(Some("x.jpg"), Some("video/mp4")), Photo);
} }
/// Keep every extension accepted by the shared classifier covered. Drive
/// activation uses this classification before it knows whether opening a
/// file should stream through FUSE or materialise a complete cache entry;
/// silently dropping one of these formats would therefore regress large
/// videos back to the expensive whole-file path.
#[test]
fn photo_kind_recognises_every_supported_video_extension() {
for ext in VIDEO_EXTS {
let lower = format!("movie.{ext}");
let upper = format!("MOVIE.{}", ext.to_ascii_uppercase());
assert_eq!(
PhotoKind::classify(Some(&lower), None),
PhotoKind::Video,
"lower-case .{ext}"
);
assert_eq!(
PhotoKind::classify(Some(&upper), None),
PhotoKind::Video,
"upper-case .{ext}"
);
}
}
#[test]
fn photo_kind_uses_video_mime_only_when_extension_is_not_authoritative() {
use PhotoKind::*;
assert_eq!(
PhotoKind::classify(Some("extensionless"), Some("video/mp4; codecs=avc1")),
Video
);
assert_eq!(
PhotoKind::classify(Some("clip.unknown"), Some("video/x-matroska")),
Video
);
// Audio is not a video and must not accidentally take the video
// streaming policy merely because both are time-based media.
assert_eq!(
PhotoKind::classify(Some("recording"), Some("audio/flac")),
Photo
);
// A known still/raw suffix remains authoritative over server metadata.
assert_eq!(
PhotoKind::classify(Some("poster.AVIF"), Some("video/mp4")),
Photo
);
assert_eq!(
PhotoKind::classify(Some("negative.NEF"), Some("video/mp4")),
Raw
);
}
#[test] #[test]
fn photo_kind_i64_round_trips() { fn photo_kind_i64_round_trips() {
for k in [PhotoKind::Photo, PhotoKind::Video, PhotoKind::Raw] { for k in [PhotoKind::Photo, PhotoKind::Video, PhotoKind::Raw] {
@@ -1167,6 +1350,150 @@ mod tests {
} }
} }
#[test]
fn unified_search_request_and_response_roundtrip() {
let request = Request::SearchV2 {
query: "quarterly report".into(),
limit: 37,
filters: SearchFilters::default(),
};
let line = serde_json::to_string(&request).unwrap();
assert_eq!(
line,
r#"{"SearchV2":{"query":"quarterly report","limit":37,"filters":{"sources":["Drive","Local"],"kind":"All"}}}"#
);
let decoded: Request = serde_json::from_str(&line).unwrap();
assert_eq!(line, serde_json::to_string(&decoded).unwrap());
let response = Response::SearchResultsV2 {
drive_hits: vec![SearchHit {
name: "report.pdf".into(),
path: "Work/report.pdf".into(),
is_dir: false,
size: 12,
modified: 123,
pinned: true,
uid: "vol~link".into(),
mounted_path: Some("/home/me/Videos/report.pdf".into()),
score: 985,
}],
local_hits: vec![LocalHit {
name: "notes.txt".into(),
path: "/home/me/notes.txt".into(),
is_dir: false,
size: 7,
modified: 456,
score: 720,
}],
local_indexing: true,
};
let line = serde_json::to_string(&response).unwrap();
assert!(!line.contains('\n'));
let decoded: Response = serde_json::from_str(&line).unwrap();
match decoded {
Response::SearchResultsV2 {
drive_hits,
local_hits,
local_indexing,
} => {
assert_eq!(drive_hits.len(), 1);
assert_eq!(drive_hits[0].uid, "vol~link");
assert_eq!(drive_hits[0].score, 985);
assert_eq!(local_hits.len(), 1);
assert_eq!(local_hits[0].path, "/home/me/notes.txt");
assert_eq!(local_hits[0].score, 720);
assert!(local_indexing);
}
other => panic!("expected unified search results, got {other:?}"),
}
}
#[test]
fn search_v2_legacy_wire_form_gets_compatible_defaults() {
let decoded: Request =
serde_json::from_str(r#"{"SearchV2":{"query":"old client","limit":20}}"#).unwrap();
match decoded {
Request::SearchV2 {
query,
limit,
filters,
} => {
assert_eq!(query, "old client");
assert_eq!(limit, 20);
assert_eq!(filters, SearchFilters::default());
}
other => panic!("expected SearchV2, got {other:?}"),
}
// New daemons add scores, but old-daemon replies remain readable and
// signal that their existing order should be retained with score 0.
let decoded: Response = serde_json::from_str(
r#"{"SearchResultsV2":{"drive_hits":[{"name":"a","path":"a","is_dir":false,"size":1,"modified":2,"pinned":false,"uid":"v~a"}],"local_hits":[{"name":"b","path":"/b","is_dir":false,"size":3,"modified":4}],"local_indexing":false}}"#,
)
.unwrap();
match decoded {
Response::SearchResultsV2 {
drive_hits,
local_hits,
..
} => {
assert_eq!(drive_hits[0].score, 0);
assert_eq!(local_hits[0].score, 0);
}
other => panic!("expected SearchResultsV2, got {other:?}"),
}
}
#[test]
fn search_v2_filters_roundtrip() {
let request = Request::SearchV2 {
query: "holiday".into(),
limit: 50,
filters: SearchFilters {
sources: vec![SearchSource::Drive],
kind: SearchKind::Images,
},
};
let line = serde_json::to_string(&request).unwrap();
let decoded: Request = serde_json::from_str(&line).unwrap();
assert!(matches!(
decoded,
Request::SearchV2 {
filters: SearchFilters { sources, kind: SearchKind::Images },
..
} if sources == vec![SearchSource::Drive]
));
}
#[test]
fn search_kind_classifies_before_results_are_limited() {
assert!(SearchKind::Folders.accepts("movie.mp4", true));
assert!(!SearchKind::Folders.accepts("movie.mp4", false));
assert!(SearchKind::Documents.accepts("REPORT.PDF", false));
assert!(SearchKind::Images.accepts("scan.HEIC", false));
assert!(SearchKind::Media.accepts("film.mkv", false));
assert!(SearchKind::Media.accepts("recording.flac", false));
assert!(!SearchKind::Media.accepts("poster.jpg", false));
assert!(SearchKind::All.accepts("anything", true));
}
#[test]
fn legacy_search_wire_forms_still_decode() {
let drive: Request =
serde_json::from_str(r#"{"Search":{"query":"old","limit":20}}"#).unwrap();
assert!(matches!(
drive,
Request::Search { query, limit } if query == "old" && limit == 20
));
let local: Request =
serde_json::from_str(r#"{"SearchLocal":{"query":"old","limit":10}}"#).unwrap();
assert!(matches!(
local,
Request::SearchLocal { query, limit } if query == "old" && limit == 10
));
}
/// The trash requests carry uids rather than paths; they must survive the same /// The trash requests carry uids rather than paths; they must survive the same
/// line-delimited round-trip, since a mangled uid would restore or destroy the /// line-delimited round-trip, since a mangled uid would restore or destroy the
/// wrong node. /// wrong node.
+53
View File
@@ -7,6 +7,7 @@ use super::Db;
use crate::Result; use crate::Result;
use crate::localindex::LocalEntry; use crate::localindex::LocalEntry;
use super::nodes::candidate_trigrams;
use super::utils::{TRIGRAM_MIN, like_escape}; use super::utils::{TRIGRAM_MIN, like_escape};
/// One hit from [`Db::search_local`]: an indexed file on the machine itself, not /// One hit from [`Db::search_local`]: an indexed file on the machine itself, not
@@ -147,6 +148,58 @@ impl Db {
} }
Ok(out) Ok(out)
} }
/// Bounded candidates for the fuzzy scorer, including matches in the full
/// local path. ORed trigrams admit misspellings while the FTS index prevents
/// this from becoming an unbounded filesystem-table scan.
pub fn search_local_candidates(&self, query: &str, limit: usize) -> Result<Vec<LocalFileHit>> {
let query = query.trim();
if query.is_empty() || limit == 0 {
return Ok(Vec::new());
}
let conn = self.conn.lock();
let terms = candidate_trigrams(query);
let lane_limit = limit.div_ceil(2);
let mut rows: Vec<LocalFileHit> = if terms.is_empty() {
let pat = format!("%{}%", like_escape(query));
let mut stmt = conn.prepare(
"SELECT path, name, is_dir, size, mtime FROM local_files
WHERE name LIKE ?1 ESCAPE '\\' OR path LIKE ?1 ESCAPE '\\'
ORDER BY mtime DESC LIMIT ?2",
)?;
stmt.query_map(params![pat, limit as i64], local_hit)?
.collect::<std::result::Result<_, _>>()?
} else {
let mut stmt = conn.prepare(
"SELECT f.path, f.name, f.is_dir, f.size, f.mtime
FROM local_fts x JOIN local_files f ON f.id = x.rowid
WHERE local_fts MATCH ?1
ORDER BY x.rank LIMIT ?2",
)?;
stmt.query_map(params![terms.join(" OR "), lane_limit as i64], local_hit)?
.collect::<std::result::Result<_, _>>()?
};
if !terms.is_empty()
&& let Some(first) = query.chars().next()
{
let prefix = format!("{}%", like_escape(&first.to_string()));
let mut stmt = conn.prepare(
"SELECT path, name, is_dir, size, mtime FROM local_files
WHERE name COLLATE NOCASE LIKE ?1 ESCAPE '\\'
ORDER BY name COLLATE NOCASE LIMIT ?2",
)?;
let fallback = stmt
.query_map(params![prefix, lane_limit as i64], local_hit)?
.collect::<std::result::Result<Vec<_>, _>>()?;
for hit in fallback {
if !rows.iter().any(|existing| existing.path == hit.path) {
rows.push(hit);
}
}
rows.truncate(limit);
}
Ok(rows)
}
} }
/// Row → [`LocalFileHit`]. Every local-search query selects the same /// Row → [`LocalFileHit`]. Every local-search query selects the same
+44 -2
View File
@@ -9,7 +9,7 @@ use super::Db;
use crate::Result; use crate::Result;
/// Current schema version. Bump on every forward migration added below. /// Current schema version. Bump on every forward migration added below.
pub(super) const SCHEMA_VERSION: i64 = 15; pub(super) const SCHEMA_VERSION: i64 = 16;
impl Db { impl Db {
pub(super) fn migrate(&self) -> Result<()> { pub(super) fn migrate(&self) -> Result<()> {
@@ -31,7 +31,12 @@ impl Db {
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
.unwrap_or(0); .unwrap_or(0);
if current >= SCHEMA_VERSION { if current > SCHEMA_VERSION {
return Err(crate::Error::Other(format!(
"database schema {current} is newer than this build supports ({SCHEMA_VERSION}); refusing to open it to avoid corrupting user data"
)));
}
if current == SCHEMA_VERSION {
return Ok(()); return Ok(());
} }
@@ -81,6 +86,9 @@ impl Db {
if current < 15 { if current < 15 {
tx.execute_batch(MIGRATION_V15)?; tx.execute_batch(MIGRATION_V15)?;
} }
if current < 16 {
tx.execute_batch(MIGRATION_V16)?;
}
tx.execute( tx.execute(
"INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
ON CONFLICT(key) DO UPDATE SET value = excluded.value", ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -369,3 +377,37 @@ CREATE VIRTUAL TABLE nodes_fts USING fts5(
INSERT INTO nodes_fts (rowid, name) INSERT INTO nodes_fts (rowid, name)
SELECT rowid, name FROM nodes WHERE trashed = 0; SELECT rowid, name FROM nodes WHERE trashed = 0;
"; ";
/// Schema v16: fuzzy search needs a bounded candidate set that includes parent
/// paths. Keeping the path in the trigram indexes lets SQLite narrow both name
/// and path candidates without walking either base table for every keystroke.
/// Node paths are initially backfilled below and subsequently maintained by the
/// node write path (including descendants after a folder move/rename).
const MIGRATION_V16: &str = "
DROP TABLE IF EXISTS nodes_fts;
CREATE VIRTUAL TABLE nodes_fts USING fts5(
name, path, tokenize='trigram'
);
CREATE INDEX IF NOT EXISTS idx_nodes_name_nocase ON nodes(name COLLATE NOCASE);
INSERT INTO nodes_fts (rowid, name, path)
WITH RECURSIVE paths(rowid, uid, path) AS (
SELECT n.rowid, n.uid, '' FROM nodes n
WHERE n.parent_uid IS NULL
OR NOT EXISTS (SELECT 1 FROM nodes p WHERE p.uid = n.parent_uid)
UNION ALL
SELECT n.rowid, n.uid,
CASE WHEN paths.path = '' THEN n.name ELSE paths.path || '/' || n.name END
FROM nodes n JOIN paths ON n.parent_uid = paths.uid
)
SELECT n.rowid, n.name, paths.path
FROM nodes n JOIN paths ON paths.rowid = n.rowid
WHERE n.trashed = 0;
DROP TABLE IF EXISTS local_fts;
CREATE VIRTUAL TABLE local_fts USING fts5(
name, path, content='local_files', content_rowid='id', tokenize='trigram'
);
CREATE INDEX IF NOT EXISTS idx_local_files_name_nocase
ON local_files(name COLLATE NOCASE);
INSERT INTO local_fts(local_fts) VALUES('rebuild');
";
+148 -5
View File
@@ -81,12 +81,42 @@ impl Db {
params![uid], params![uid],
|row| row.get(0), |row| row.get(0),
)?; )?;
tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?; // A folder rename/move changes every descendant's searchable path.
if !node.trashed { // Refresh the affected subtree, while retaining efficient rowid
tx.execute( // deletes for the overwhelmingly common one-file case.
"INSERT INTO nodes_fts (rowid, name) VALUES (?1, ?2)", let mut affected = vec![rowid];
params![rowid, node.name], if node.is_folder() {
let mut stmt = tx.prepare(
"WITH RECURSIVE descendants(rowid, uid) AS (
SELECT rowid, uid FROM nodes WHERE parent_uid = ?1
UNION ALL
SELECT n.rowid, n.uid FROM nodes n
JOIN descendants d ON n.parent_uid = d.uid
) SELECT rowid FROM descendants",
)?; )?;
for descendant in stmt.query_map(params![uid], |row| row.get(0))? {
affected.push(descendant?);
}
}
for affected_rowid in affected {
tx.execute(
"DELETE FROM nodes_fts WHERE rowid = ?1",
params![affected_rowid],
)?;
let indexed: Option<(String, String)> = tx
.query_row(
"SELECT name, uid FROM nodes WHERE rowid = ?1 AND trashed = 0",
params![affected_rowid],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((name, indexed_uid)) = indexed {
let path = path_of(&tx, &indexed_uid)?;
tx.execute(
"INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
params![affected_rowid, name, path],
)?;
}
} }
} }
tx.commit()?; tx.commit()?;
@@ -130,6 +160,40 @@ impl Db {
Ok(count > 0) Ok(count > 0)
} }
/// Resolve `uid` relative to an ancestor node.
///
/// This is used when a remote subtree has its own local sync mount: the
/// sync folder stores the ancestor UID, while search results identify the
/// selected descendant. Returning `None` for an unrelated or incomplete
/// chain keeps callers from accidentally joining a Drive-wide path onto the
/// wrong mountpoint. The ancestor itself resolves to the empty path.
pub fn path_relative_to(&self, ancestor_uid: &str, uid: &str) -> Result<Option<String>> {
let conn = self.conn.lock();
let mut stmt = conn.prepare(
"WITH RECURSIVE anc(uid, parent_uid, name, depth) AS (
SELECT uid, parent_uid, name, 0 FROM nodes WHERE uid = ?1
UNION ALL
SELECT n.uid, n.parent_uid, n.name, anc.depth + 1
FROM nodes n JOIN anc ON n.uid = anc.parent_uid
WHERE anc.depth < 1024
)
SELECT uid, name FROM anc ORDER BY depth",
)?;
let rows = stmt.query_map(params![uid], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?;
let mut parts = Vec::new();
for row in rows {
let (current_uid, name) = row?;
if current_uid == ancestor_uid {
parts.reverse();
return Ok(Some(parts.join("/")));
}
parts.push(name);
}
Ok(None)
}
/// Full-text search over node names, newest schema's trigram index giving /// Full-text search over node names, newest schema's trigram index giving
/// substring (not just prefix) matches. Returns up to `limit` non-trashed /// substring (not just prefix) matches. Returns up to `limit` non-trashed
/// hits, each with its mountpoint-relative path resolved. Queries shorter /// hits, each with its mountpoint-relative path resolved. Queries shorter
@@ -175,6 +239,68 @@ impl Db {
Ok(hits) Ok(hits)
} }
/// Return a bounded, deliberately broad candidate pool for fuzzy ranking.
/// Unlike [`search`](Self::search), trigram terms are ORed across both the
/// basename and parent path, so a typo can still share enough trigrams to
/// enter the pool. Final relevance ordering belongs to the caller.
pub fn search_candidates(&self, query: &str, limit: usize) -> Result<Vec<SearchHit>> {
let query = query.trim();
if query.is_empty() || limit == 0 {
return Ok(Vec::new());
}
let conn = self.conn.lock();
let terms = candidate_trigrams(query);
let lane_limit = limit.div_ceil(2);
let mut rows: Vec<(String, String)> = if terms.is_empty() {
let pat = format!("%{}%", like_escape(query));
let mut stmt = conn.prepare(
"SELECT node_json, uid FROM nodes
WHERE name LIKE ?1 ESCAPE '\\' AND trashed = 0 AND node_json IS NOT NULL
ORDER BY name LIMIT ?2",
)?;
collect_pairs(stmt.query_map(params![pat, limit as i64], pair)?)?
} else {
let expression = terms.join(" OR ");
let mut stmt = conn.prepare(
"SELECT n.node_json, n.uid
FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid
WHERE nodes_fts MATCH ?1 AND n.trashed = 0 AND n.node_json IS NOT NULL
ORDER BY f.rank LIMIT ?2",
)?;
collect_pairs(stmt.query_map(params![expression, lane_limit as i64], pair)?)?
};
// A short substitution can destroy every trigram (`vedio` vs `video`).
// Feed the scorer an additional indexed same-initial lane so it can
// recover those candidates without scanning the full node table.
if !terms.is_empty()
&& let Some(first) = query.chars().next()
{
let prefix = format!("{}%", like_escape(&first.to_string()));
let mut stmt = conn.prepare(
"SELECT node_json, uid FROM nodes
WHERE name COLLATE NOCASE LIKE ?1 ESCAPE '\\'
AND trashed = 0 AND node_json IS NOT NULL
ORDER BY name COLLATE NOCASE LIMIT ?2",
)?;
let fallback =
collect_pairs(stmt.query_map(params![prefix, lane_limit as i64], pair)?)?;
for row in fallback {
if !rows.iter().any(|(_, uid)| uid == &row.1) {
rows.push(row);
}
}
rows.truncate(limit);
}
rows.into_iter()
.map(|(json, uid)| {
Ok(SearchHit {
node: serde_json::from_str(&json)?,
path: path_of(&conn, &uid)?,
})
})
.collect()
}
/// Mark (or unmark) a folder's child listing as complete. A listed folder /// Mark (or unmark) a folder's child listing as complete. A listed folder
/// rehydrates its `children` map on mount even when empty; an unlisted one /// rehydrates its `children` map on mount even when empty; an unlisted one
/// re-enumerates from the remote on next access. /// re-enumerates from the remote on next access.
@@ -261,6 +387,23 @@ impl Db {
// in sync on every store/read/evict, so it is authoritative for eviction. // in sync on every store/read/evict, so it is authoritative for eviction.
} }
/// Unique quoted character trigrams suitable for an FTS5 OR expression.
pub(super) fn candidate_trigrams(query: &str) -> Vec<String> {
let chars: Vec<char> = query.to_lowercase().chars().collect();
let mut terms = Vec::new();
for window in chars.windows(TRIGRAM_MIN) {
if window.iter().all(|c| c.is_whitespace()) {
continue;
}
let raw: String = window.iter().collect();
let quoted = format!("\"{}\"", raw.replace('"', "\"\""));
if !terms.contains(&quoted) {
terms.push(quoted);
}
}
terms
}
/// Effective plaintext size of a node for the indexed `size` column: the /// Effective plaintext size of a node for the indexed `size` column: the
/// claimed size when known, else the on-storage size; folders are 0. /// claimed size when known, else the on-storage size; folders are 0.
pub(super) fn node_size(node: &Node) -> i64 { pub(super) fn node_size(node: &Node) -> i64 {
+8 -5
View File
@@ -118,9 +118,10 @@ pub struct AttachedBlob {
impl Db { impl Db {
pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> { pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
let conn = self.conn.lock(); let mut conn = self.conn.lock();
let tx = conn.transaction()?;
let superseded: Option<String> = if op_supersedes(&op.kind) { let superseded: Option<String> = if op_supersedes(&op.kind) {
let blob: Option<String> = conn let blob: Option<String> = tx
.query_row( .query_row(
"SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2", "SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2",
params![op.uid, op.kind], params![op.uid, op.kind],
@@ -128,7 +129,7 @@ impl Db {
) )
.optional()? .optional()?
.flatten(); .flatten();
conn.execute( tx.execute(
"DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2", "DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2",
params![op.uid, op.kind], params![op.uid, op.kind],
)?; )?;
@@ -136,7 +137,7 @@ impl Db {
} else { } else {
None None
}; };
conn.execute( tx.execute(
"INSERT INTO pending_op "INSERT INTO pending_op
(kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at) (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
@@ -151,7 +152,9 @@ impl Db {
op.next_attempt_at op.next_attempt_at
], ],
)?; )?;
Ok((conn.last_insert_rowid(), superseded)) let id = tx.last_insert_rowid();
tx.commit()?;
Ok((id, superseded))
} }
/// Point a queued create at the bytes that were just written to it, returning /// Point a queued create at the bytes that were just written to it, returning
+248
View File
@@ -5,6 +5,7 @@ use crate::control::{ActivityEntry, ActivityKind};
use crate::localindex::LocalEntry; use crate::localindex::LocalEntry;
use proton_drive_rs::proton_sdk::ids::NodeUid; use proton_drive_rs::proton_sdk::ids::NodeUid;
use proton_drive_rs::{Node, NodeKind}; use proton_drive_rs::{Node, NodeKind};
use serde_json::json;
/// `Db::open` applies the pragmas that `open_in_memory` skips, so the WAL /// `Db::open` applies the pragmas that `open_in_memory` skips, so the WAL
/// settings can only be checked against a real file. /// settings can only be checked against a real file.
@@ -542,6 +543,95 @@ fn search_short_query_like_fallback() {
assert_eq!(hits[0].node.name, "ab.txt"); assert_eq!(hits[0].node.name, "ab.txt");
} }
#[test]
fn fuzzy_candidates_include_typo_and_parent_path_matches() {
let db = Db::open_in_memory().unwrap();
db.upsert_node(&folder("root", None, "My Files")).unwrap();
db.upsert_node(&folder("projects", Some("root"), "Projects"))
.unwrap();
db.upsert_node(&file("report", "projects", "quarterly-report.pdf", 1))
.unwrap();
db.upsert_node(&file("video", "projects", "video.mp4", 1))
.unwrap();
let typo = db.search_candidates("quartrly", 20).unwrap();
assert!(
typo.iter()
.any(|hit| hit.node.name == "quarterly-report.pdf")
);
assert!(
db.search_candidates("vedio", 20)
.unwrap()
.iter()
.any(|hit| hit.node.name == "video.mp4")
);
let path = db.search_candidates("projects", 20).unwrap();
assert!(
path.iter()
.any(|hit| hit.node.name == "quarterly-report.pdf")
);
}
#[test]
fn renaming_parent_refreshes_descendant_search_paths() {
let db = Db::open_in_memory().unwrap();
db.upsert_node(&folder("root", None, "My Files")).unwrap();
db.upsert_node(&folder("folder", Some("root"), "zzbeforezz"))
.unwrap();
db.upsert_node(&file("child", "folder", "plain.txt", 1))
.unwrap();
assert!(
db.search_candidates("zzbeforezz", 20)
.unwrap()
.iter()
.any(|h| h.node.name == "plain.txt")
);
db.upsert_node(&folder("folder", Some("root"), "yyafteryy"))
.unwrap();
assert!(
!db.search_candidates("zzbeforezz", 20)
.unwrap()
.iter()
.any(|h| h.node.name == "plain.txt")
);
assert!(
db.search_candidates("yyafteryy", 20)
.unwrap()
.iter()
.any(|h| h.node.name == "plain.txt")
);
}
#[test]
fn local_fuzzy_candidates_include_typo_and_parent_path() {
let db = Db::open_in_memory().unwrap();
let generation = db.local_begin_scan().unwrap();
db.local_upsert_batch(
generation,
&[
local(
"/home/u/Projects/quarterly-report.pdf",
"quarterly-report.pdf",
false,
),
local("/home/u/Videos/video.mp4", "video.mp4", false),
],
)
.unwrap();
db.local_finish_scan(generation, 1).unwrap();
assert_eq!(db.search_local_candidates("quartrly", 20).unwrap().len(), 1);
assert_eq!(db.search_local_candidates("projects", 20).unwrap().len(), 1);
assert!(
db.search_local_candidates("vedio", 20)
.unwrap()
.iter()
.any(|hit| hit.name == "video.mp4")
);
}
#[test] #[test]
fn search_drops_fts_row_on_delete_and_trash() { fn search_drops_fts_row_on_delete_and_trash() {
let db = Db::open_in_memory().unwrap(); let db = Db::open_in_memory().unwrap();
@@ -694,6 +784,41 @@ fn opens_and_migrates() {
assert_eq!(version, SCHEMA_VERSION.to_string()); assert_eq!(version, SCHEMA_VERSION.to_string());
} }
#[test]
fn refuses_a_schema_newer_than_the_running_build() {
let path = std::env::temp_dir().join(format!(
"pdfs-db-future-{}-{}.db",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let conn = rusqlite::Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT);
INSERT INTO sync_state VALUES ('schema_version', '9999');",
)
.unwrap();
drop(conn);
let err = Db::open(&path)
.err()
.expect("future schema must fail closed");
assert!(err.to_string().contains("newer than this build supports"));
let conn = rusqlite::Connection::open(&path).unwrap();
let version: String = conn
.query_row(
"SELECT value FROM sync_state WHERE key = 'schema_version'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(version, "9999", "failed open must not rewrite the database");
drop(conn);
let _ = std::fs::remove_file(path);
}
/// A queued write whose baseline is restamped must keep everything else. /// A queued write whose baseline is restamped must keep everything else.
/// ///
/// The restamp happens after our *own* upload seals a new revision under a /// The restamp happens after our *own* upload seals a new revision under a
@@ -777,6 +902,39 @@ fn a_second_write_supersedes_the_first_pending_op() {
assert_eq!(db.pending_op_counts().unwrap().uploads, 1); assert_eq!(db.pending_op_counts().unwrap().uploads, 1);
} }
#[test]
fn failed_superseding_insert_keeps_the_old_pending_op() {
let db = Db::open_in_memory().unwrap();
let op = |blob: &str| PendingOp {
id: 0,
kind: OP_REVISION.to_string(),
uid: uid("atomic").to_string(),
parent_uid: None,
name: None,
blob_path: Some(blob.to_string()),
meta_json: Some("{}".to_string()),
created_at: 1,
attempts: 0,
last_error: None,
next_attempt_at: 0,
};
db.enqueue_op(&op("/staging/original")).unwrap();
db.with_conn(|conn| {
conn.execute_batch(
"CREATE TRIGGER reject_replacement BEFORE INSERT ON pending_op
WHEN NEW.blob_path = '/staging/replacement'
BEGIN SELECT RAISE(ABORT, 'injected insert failure'); END;",
)?;
Ok(())
})
.unwrap();
assert!(db.enqueue_op(&op("/staging/replacement")).is_err());
let ops = db.pending_ops().unwrap();
assert_eq!(ops.len(), 1);
assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/original"));
}
/// Deleting a folder that was created offline must take the ops queued /// Deleting a folder that was created offline must take the ops queued
/// underneath it with it: they name a placeholder parent that will now never /// underneath it with it: they name a placeholder parent that will now never
/// become real, so nothing could ever drain them and nothing is left to /// become real, so nothing could ever drain them and nothing is left to
@@ -1092,6 +1250,96 @@ fn pending_mode_is_queued_until_the_mode_is_reached() {
assert_eq!(folder.pending_mode, None); assert_eq!(folder.pending_mode, None);
} }
#[test]
fn node_path_resolves_relative_to_a_sync_root_uid() {
let db = Db::open_in_memory().unwrap();
let root = node_from(serde_json::Value::Null, "root", "My Files", json!("Folder"));
let videos = node_from(json!(uid("root")), "videos", "Videos", json!("Folder"));
let anime = node_from(json!(uid("videos")), "anime", "anime", json!("Folder"));
let show = node_from(json!(uid("anime")), "show", "Oshi no Ko", json!("Folder"));
let episode = node_from(
json!(uid("show")),
"episode",
"episode 01.mkv",
json!({"File": {
"media_type": "video/x-matroska",
"total_size_on_storage": 42,
"claimed_size": 42,
"claimed_modification_time": null
}}),
);
db.upsert_nodes(&[root, videos, anime, show, episode])
.unwrap();
assert_eq!(
db.path_relative_to(&uid("videos").to_string(), &uid("episode").to_string())
.unwrap()
.as_deref(),
Some("anime/Oshi no Ko/episode 01.mkv")
);
assert_eq!(
db.path_relative_to(&uid("videos").to_string(), &uid("videos").to_string())
.unwrap()
.as_deref(),
Some("")
);
assert_eq!(
db.path_relative_to(&uid("anime").to_string(), &uid("episode").to_string())
.unwrap()
.as_deref(),
Some("Oshi no Ko/episode 01.mkv")
);
}
#[test]
fn node_path_rejects_unrelated_missing_and_broken_ancestor_chains() {
let db = Db::open_in_memory().unwrap();
let root = node_from(serde_json::Value::Null, "root", "My Files", json!("Folder"));
let videos = node_from(json!(uid("root")), "videos", "Videos", json!("Folder"));
let episode = node_from(
json!(uid("videos")),
"episode",
"episode.mkv",
json!({"File": {
"media_type": "video/x-matroska",
"total_size_on_storage": 42,
"claimed_size": 42,
"claimed_modification_time": null
}}),
);
let unrelated = node_from(
json!(uid("root")),
"documents",
"Documents",
json!("Folder"),
);
let orphan = node_from(
json!(uid("missing-parent")),
"orphan",
"orphan.mkv",
json!({"File": {
"media_type": "video/x-matroska",
"total_size_on_storage": 42,
"claimed_size": 42,
"claimed_modification_time": null
}}),
);
db.upsert_nodes(&[root, videos, episode, unrelated, orphan])
.unwrap();
for (ancestor, descendant) in [
("documents", "episode"),
("videos", "missing-node"),
("videos", "orphan"),
] {
assert_eq!(
db.path_relative_to(&uid(ancestor).to_string(), &uid(descendant).to_string())
.unwrap(),
None
);
}
}
#[test] #[test]
fn cache_index_touch_access_and_lru_order() { fn cache_index_touch_access_and_lru_order() {
let db = Db::open_in_memory().unwrap(); let db = Db::open_in_memory().unwrap();
+1
View File
@@ -10,6 +10,7 @@ pub mod db;
pub mod error; pub mod error;
pub mod localindex; pub mod localindex;
pub mod profile; pub mod profile;
pub mod search;
pub mod service; pub mod service;
pub mod shell; pub mod shell;
pub mod syncignore; pub mod syncignore;
+214
View File
@@ -0,0 +1,214 @@
//! Shared, deterministic search-result relevance scoring.
//!
//! The scorer deliberately operates on a small candidate set rather than
//! replacing an index. Callers should use SQLite (or another index) to find
//! candidates, then use this module to put local and remote results on the
//! same relevance scale.
/// Scores a candidate against a user query.
///
/// Every query term must match either `basename` or `parent_path`. Basename
/// matches are weighted twice as strongly as path matches. The returned score
/// is stable and only meaningful when compared with scores produced by this
/// function.
#[must_use]
pub fn relevance_score(query: &str, basename: &str, parent_path: &str) -> Option<i64> {
let query = normalize(query);
if query.is_empty() {
return Some(0);
}
let name = normalize(basename);
let path = normalize(parent_path);
let query_terms: Vec<_> = query.split_whitespace().collect();
let name_words: Vec<_> = name.split_whitespace().collect();
let path_words: Vec<_> = path.split_whitespace().collect();
let mut total = 0_i64;
for term in query_terms {
let name_score = field_term_score(term, &name, &name_words).map(|score| score * 2);
let path_score = field_term_score(term, &path, &path_words);
total += name_score.into_iter().chain(path_score).max()?;
}
// Whole-field bonuses make ordering intuitive without changing whether a
// candidate matches. They dominate small differences between term scores.
if name == query {
total += 20_000;
} else if name.starts_with(&query) {
total += 10_000;
} else if name.contains(&query) {
total += 4_000;
}
Some(total)
}
fn normalize(value: &str) -> String {
let mut normalized = String::with_capacity(value.len());
let mut last_was_space = true;
for character in value.chars().flat_map(char::to_lowercase) {
if character.is_alphanumeric() {
normalized.push(character);
last_was_space = false;
} else if !last_was_space {
normalized.push(' ');
last_was_space = true;
}
}
if last_was_space {
normalized.pop();
}
normalized
}
fn field_term_score(term: &str, field: &str, words: &[&str]) -> Option<i64> {
if field == term {
return Some(10_000);
}
if field.starts_with(term) {
return Some(8_500);
}
if words.contains(&term) {
return Some(8_000);
}
if words.iter().any(|word| word.starts_with(term)) {
return Some(7_000);
}
if field.contains(term) {
return Some(6_000);
}
// Subsequence matching supports useful abbreviations such as "qtrrpt".
let subsequence = std::iter::once(field)
.chain(words.iter().copied())
.filter_map(|word| subsequence_gap(term, word))
.min();
if let Some(gap) = subsequence {
return Some(4_500 - i64::try_from(gap.min(500)).unwrap_or(500));
}
// Typo matching is deliberately restricted to substantial terms to avoid
// noisy matches for one- and two-letter searches.
if term.chars().count() >= 4
&& words
.iter()
.any(|word| bounded_damerau_levenshtein(term, word, typo_limit(term)))
{
return Some(3_500);
}
None
}
fn subsequence_gap(needle: &str, haystack: &str) -> Option<usize> {
let needle_length = needle.chars().count();
let mut needle = needle.chars();
let mut wanted = needle.next()?;
let mut first = None;
for (index, character) in haystack.chars().enumerate() {
if character == wanted {
first.get_or_insert(index);
match needle.next() {
Some(next) => wanted = next,
None => return Some(index + 1 - first.unwrap_or(index) - needle_length),
}
}
}
None
}
fn typo_limit(term: &str) -> usize {
usize::from(term.chars().count() >= 5) + 1
}
/// A bounded optimal-string-alignment distance check. This treats a single
/// adjacent transposition (for example, `vidoe`) as one edit.
fn bounded_damerau_levenshtein(left: &str, right: &str, limit: usize) -> bool {
let left: Vec<_> = left.chars().collect();
let right: Vec<_> = right.chars().collect();
if left.len().abs_diff(right.len()) > limit || left.len().max(right.len()) > 64 {
return false;
}
let mut distances = vec![vec![0_usize; right.len() + 1]; left.len() + 1];
for (index, row) in distances.iter_mut().enumerate() {
row[0] = index;
}
for (index, distance) in distances[0].iter_mut().enumerate() {
*distance = index;
}
for i in 1..=left.len() {
for j in 1..=right.len() {
let cost = usize::from(left[i - 1] != right[j - 1]);
distances[i][j] = (distances[i - 1][j] + 1)
.min(distances[i][j - 1] + 1)
.min(distances[i - 1][j - 1] + cost);
if i > 1 && j > 1 && left[i - 1] == right[j - 2] && left[i - 2] == right[j - 1] {
distances[i][j] = distances[i][j].min(distances[i - 2][j - 2] + 1);
}
}
}
distances[left.len()][right.len()] <= limit
}
#[cfg(test)]
mod tests {
use super::relevance_score;
#[test]
fn exact_name_beats_prefix_and_substring() {
let exact = relevance_score("report", "report", "work").unwrap();
let prefix = relevance_score("report", "report final", "work").unwrap();
let substring = relevance_score("report", "annual report", "work").unwrap();
assert!(exact > prefix);
assert!(prefix > substring);
}
#[test]
fn basename_is_weighted_above_parent_path() {
let name = relevance_score("budget", "budget notes", "archive").unwrap();
let path = relevance_score("budget", "notes", "work/budget").unwrap();
assert!(name > path);
}
#[test]
fn matches_typos_and_adjacent_transpositions() {
assert!(relevance_score("vedio", "video.mp4", "media").is_some());
assert!(relevance_score("vidoe", "video.mp4", "media").is_some());
assert!(relevance_score("presentaton", "presentation.pdf", "work").is_some());
}
#[test]
fn separators_form_searchable_word_boundaries() {
assert!(relevance_score("final draft", "Final_Draft-v2.pdf", "work").is_some());
assert!(relevance_score("quarter report", "quarterly-report.pdf", "work").is_some());
}
#[test]
fn unicode_case_folding_preserves_accented_matches() {
assert!(relevance_score("CAFÉ", "Café Photos", "Trips").is_some());
assert!(relevance_score("ångström", "ÅNGSTRÖM notes", "Research").is_some());
}
#[test]
fn matches_terms_across_name_and_path() {
assert!(relevance_score("tax invoice", "invoice-1042.pdf", "archive/tax/2025").is_some());
}
#[test]
fn ordered_subsequence_supports_abbreviations() {
assert!(relevance_score("qtrrpt", "quarter-report.pdf", "finance").is_some());
}
#[test]
fn rejects_unrelated_candidates_and_short_typos() {
assert_eq!(relevance_score("video", "holiday.jpg", "photos"), None);
assert_eq!(relevance_score("xy", "xz.txt", "misc"), None);
}
}
+20 -12
View File
@@ -126,12 +126,16 @@ pub(super) async fn run_event_sync(
match client.enumerate_events(&scope, None).await { match client.enumerate_events(&scope, None).await {
Ok(events) => { Ok(events) => {
let head = events.last().map(|e| e.id().clone()); let head = events.last().map(|e| e.id().clone());
if let Some(c) = &head let Some(c) = &head else {
&& let Err(e) = db.set_event_cursor(c.as_str()) break None;
{ };
warn!(error = %e, "persist seed cursor failed"); if let Err(e) = db.set_event_cursor(c.as_str()) {
warn!(error = %e, ?delay, "persist seed cursor failed; retrying");
tokio::time::sleep(delay).await;
delay = (delay * 2).min(ONLINE_PROBE_MAX);
continue;
} }
break head; break Some(c.clone());
} }
Err(e) => { Err(e) => {
warn!(error = %e, ?delay, "seed event cursor failed; retrying"); warn!(error = %e, ?delay, "seed event cursor failed; retrying");
@@ -168,15 +172,19 @@ pub(super) async fn run_event_sync(
// the life of the daemon (SDK plan #9). `apply_event` only touches // the life of the daemon (SDK plan #9). `apply_event` only touches
// our FUSE state, so nothing else does this. // our FUSE state, so nothing else does this.
if let Err(e) = client.invalidate_caches_for_event(event).await { if let Err(e) = client.invalidate_caches_for_event(event).await {
warn!(error = %e, "sdk cache invalidation for event failed"); warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry");
break;
} }
apply_event(&state, &content, &pending, &notifier, event); apply_event(&state, &content, &pending, &notifier, event);
} let applied = event.id().clone();
cursor = events.last().map(|e| e.id().clone()); if let Err(e) = db.set_event_cursor(applied.as_str()) {
if let Some(c) = &cursor warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry");
&& let Err(e) = db.set_event_cursor(c.as_str()) break;
{ }
warn!(error = %e, "persist event cursor failed"); // Advance only after this exact event is durably acknowledged. If a
// later event fails, polling resumes from here and replays nothing
// that was not already made visible locally.
cursor = Some(applied);
} }
} }
} }
+133 -7
View File
@@ -12,8 +12,8 @@
use std::io::{BufRead, BufReader, Write as _}; use std::io::{BufRead, BufReader, Write as _};
use std::os::unix::net::{UnixListener, UnixStream}; use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering; use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Instant; use std::time::{Duration, Instant};
use pdfs_core::config::AppDirs; use pdfs_core::config::AppDirs;
use pdfs_core::control::{ use pdfs_core::control::{
@@ -27,6 +27,67 @@ use tracing::{info, warn};
use super::transfers::CountingReader; use super::transfers::CountingReader;
use super::{Core, count_noun, human_bytes, human_duration, parse_uid}; use super::{Core, count_noun, human_bytes, human_duration, parse_uid};
const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024;
const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_CONTROL_HANDLERS: usize = 64;
static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0);
struct ControlHandlerPermit;
impl ControlHandlerPermit {
fn acquire() -> Option<Self> {
ACTIVE_CONTROL_HANDLERS
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| {
(active < MAX_CONTROL_HANDLERS).then_some(active + 1)
})
.ok()
.map(|_| Self)
}
}
impl Drop for ControlHandlerPermit {
fn drop(&mut self) {
ACTIVE_CONTROL_HANDLERS.fetch_sub(1, Ordering::AcqRel);
}
}
fn read_request_line(reader: &mut impl BufRead) -> std::io::Result<Option<String>> {
let mut bytes = Vec::new();
loop {
let available = reader.fill_buf()?;
if available.is_empty() {
break;
}
let take = available
.iter()
.position(|byte| *byte == b'\n')
.map_or(available.len(), |index| index + 1);
if bytes.len().saturating_add(take) > MAX_CONTROL_REQUEST_BYTES as usize {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"control request exceeds size limit",
));
}
bytes.extend_from_slice(&available[..take]);
reader.consume(take);
if bytes.ends_with(b"\n") {
break;
}
}
if bytes.is_empty() {
return Ok(None);
}
if bytes.len() as u64 > MAX_CONTROL_REQUEST_BYTES || !bytes.ends_with(b"\n") {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"control request exceeds limit or is not newline terminated",
));
}
String::from_utf8(bytes)
.map(Some)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
}
/// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path /// 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 /// must live under `mountpoint`; a relative path is taken as already relative to
/// the mount root. /// the mount root.
@@ -44,6 +105,10 @@ fn rel_to_mount(mountpoint: &Path, path: &str) -> CoreResult<PathBuf> {
/// Handle one control-socket connection: read a single JSON request line, /// Handle one control-socket connection: read a single JSON request line,
/// dispatch it against `core`, and write back a JSON response line. /// dispatch it against `core`, and write back a JSON response line.
fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) { fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) {
if let Err(e) = stream.set_read_timeout(Some(CONTROL_READ_TIMEOUT)) {
warn!(error = %e, "control: could not set request timeout");
return;
}
let mut reader = BufReader::new(match stream.try_clone() { let mut reader = BufReader::new(match stream.try_clone() {
Ok(s) => s, Ok(s) => s,
Err(e) => { Err(e) => {
@@ -51,10 +116,14 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
return; return;
} }
}); });
let mut line = String::new(); let line = match read_request_line(&mut reader) {
if reader.read_line(&mut line).is_err() || line.trim().is_empty() { Ok(Some(line)) if !line.trim().is_empty() => line,
return; Ok(_) => return,
} Err(e) => {
warn!(error = %e, "control: rejected request");
return;
}
};
let response = match serde_json::from_str::<CtlRequest>(line.trim()) { let response = match serde_json::from_str::<CtlRequest>(line.trim()) {
Ok(CtlRequest::Status) => { Ok(CtlRequest::Status) => {
let pins = core.cache.list_pins(); let pins = core.cache.list_pins();
@@ -241,6 +310,18 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
}, },
Err(e) => CtlResponse::error(e), Err(e) => CtlResponse::error(e),
}, },
Ok(CtlRequest::SearchV2 {
query,
limit,
filters,
}) => match core.search_v2(&query, limit, &filters) {
Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 {
drive_hits,
local_hits,
local_indexing: core.indexing.load(Ordering::Relaxed),
},
Err(e) => CtlResponse::error(e),
},
Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) { Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) {
Ok(rel) => match core.rename(&rel, &new_name) { Ok(rel) => match core.rename(&rel, &new_name) {
Ok(name) => { Ok(name) => {
@@ -780,12 +861,22 @@ pub(crate) fn run_control_socket(
for conn in listener.incoming() { for conn in listener.incoming() {
match conn { match conn {
Ok(stream) => { Ok(stream) => {
let Some(permit) = ControlHandlerPermit::acquire() else {
warn!(
limit = MAX_CONTROL_HANDLERS,
"control: rejecting connection at handler limit"
);
continue;
};
let core = core.clone(); let core = core.clone();
let username = username.clone(); let username = username.clone();
let mountpoint = mountpoint.clone(); let mountpoint = mountpoint.clone();
if let Err(e) = std::thread::Builder::new() if let Err(e) = std::thread::Builder::new()
.name("pdfs-control".into()) .name("pdfs-control".into())
.spawn(move || handle_control_conn(&core, &username, &mountpoint, stream)) .spawn(move || {
let _permit = permit;
handle_control_conn(&core, &username, &mountpoint, stream);
})
{ {
warn!(error = %e, "control: spawn handler failed"); warn!(error = %e, "control: spawn handler failed");
} }
@@ -796,3 +887,38 @@ pub(crate) fn run_control_socket(
} }
} }
} }
#[cfg(test)]
mod request_limit_tests {
use super::*;
use std::io::Cursor;
#[test]
fn accepts_one_bounded_newline_terminated_request() {
let mut input = Cursor::new(b"{\"Status\":null}\n".to_vec());
assert_eq!(
read_request_line(&mut input).unwrap().as_deref(),
Some("{\"Status\":null}\n")
);
}
#[test]
fn rejects_a_request_without_a_terminating_newline() {
let mut input = Cursor::new(b"{\"Status\":null}".to_vec());
assert_eq!(
read_request_line(&mut input).unwrap_err().kind(),
std::io::ErrorKind::InvalidData
);
}
#[test]
fn rejects_a_request_larger_than_the_limit() {
let mut bytes = vec![b'x'; MAX_CONTROL_REQUEST_BYTES as usize + 1];
bytes.push(b'\n');
let mut input = Cursor::new(bytes);
assert_eq!(
read_request_line(&mut input).unwrap_err().kind(),
std::io::ErrorKind::InvalidData
);
}
}
+97 -32
View File
@@ -5,7 +5,13 @@ use super::*;
impl Filesystem for ProtonFs { impl Filesystem for ProtonFs {
fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) { fn lookup(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
let parent = parent.0; let parent = parent.0;
let name = name.to_string_lossy().into_owned(); let name = match fuse_name(name) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
// A folder that has not been listed yet is enumerated from the remote, // A folder that has not been listed yet is enumerated from the remote,
// so serve it from a worker rather than stalling the dispatch loop. A // so serve it from a worker rather than stalling the dispatch loop. A
// listed folder — the common case — is a map hit, and answering it // listed folder — the common case — is a map hit, and answering it
@@ -307,7 +313,13 @@ impl Filesystem for ProtonFs {
reply.error(e); reply.error(e);
return; return;
} }
let name = name.to_string_lossy().into_owned(); let name = match fuse_name(name) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
let parent_uid = { let parent_uid = {
let st = self.core.state.lock(); let st = self.core.state.lock();
match st.entries.get(&parent) { match st.entries.get(&parent) {
@@ -761,7 +773,13 @@ impl Filesystem for ProtonFs {
} }
fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
let name_str = name.to_string_lossy(); let name_str = match fuse_name(name) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
let st = self.core.state.lock(); let st = self.core.state.lock();
if let Some(entry) = st.entries.get(&ino) if let Some(entry) = st.entries.get(&ino)
@@ -775,7 +793,13 @@ impl Filesystem for ProtonFs {
} }
fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
let name_str = name.to_string_lossy(); let name_str = match fuse_name(name) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) {
{ {
let st = self.core.state.lock(); let st = self.core.state.lock();
@@ -817,7 +841,13 @@ impl Filesystem for ProtonFs {
reply.error(e); reply.error(e);
return; return;
} }
let name = name.to_string_lossy().into_owned(); let name = match fuse_name(name) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
let parent_uid = { let parent_uid = {
let st = self.core.state.lock(); let st = self.core.state.lock();
match st.entries.get(&parent) { match st.entries.get(&parent) {
@@ -899,8 +929,20 @@ impl Filesystem for ProtonFs {
) { ) {
let parent = parent.0; let parent = parent.0;
let newparent = newparent.0; let newparent = newparent.0;
let name = name.to_string_lossy().into_owned(); let name = match fuse_name(name) {
let newname = newname.to_string_lossy().into_owned(); Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
let newname = match fuse_name(newname) {
Ok(name) => name,
Err(error) => {
reply.error(error);
return;
}
};
let (ino, uid) = match self.core.lookup_child(parent, &name) { let (ino, uid) = match self.core.lookup_child(parent, &name) {
Ok(x) => x, Ok(x) => x,
Err(e) => { Err(e) => {
@@ -1050,12 +1092,31 @@ impl Filesystem for ProtonFs {
// take the synchronous path, so a genuine API refusal (permissions, a // take the synchronous path, so a genuine API refusal (permissions, a
// name clash) surfaces to the caller instead of becoming a queued op // name clash) surfaces to the caller instead of becoming a queued op
// that can only ever conflict. // that can only ever conflict.
if !self.core.online.load(Ordering::Relaxed) || is_local_uid(&new_parent_uid) { if rename_needs_queue(
self.core.online.load(Ordering::Relaxed),
is_local_uid(&new_parent_uid),
newparent != parent,
newname != name,
) {
match self match self
.core .core
.queue_rename(ino, &uid, newparent, &new_parent_uid, &newname) .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname)
{ {
Ok(()) => reply.ok(), Ok(()) => {
// Send the rename response first. Kernel-side rename cache
// updates happen while processing that response; notifying
// before it lets those updates overwrite our invalidation
// and retain an empty destination readdir page.
reply.ok();
if let Some(notifier) = self.core.notifier.get() {
let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name));
let _ = notifier.inval_entry(INodeNo(newparent), OsStr::new(&newname));
let _ = notifier.inval_inode(INodeNo(parent), 0, 0);
if newparent != parent {
let _ = notifier.inval_inode(INodeNo(newparent), 0, 0);
}
}
}
Err(e) => { Err(e) => {
self.core.restore_replaced(victim.as_ref(), &newname); self.core.restore_replaced(victim.as_ref(), &newname);
reply.error(e); reply.error(e);
@@ -1063,54 +1124,57 @@ impl Filesystem for ProtonFs {
} }
return; return;
} }
// Move first if the parent changed, then rename if the name changed. // Rename first if both halves change. Moving first makes the encrypted
// name requirements stale and repeatedly failed with InvalidRequirements.
// A failure past this point has already trashed any node the rename was // A failure past this point has already trashed any node the rename was
// replacing, so put it back rather than leaving the caller with neither // replacing, so put it back rather than leaving the caller with neither
// the source moved nor the destination intact. // the source moved nor the destination intact.
if newparent != parent if newname != name
&& let Err(e) = self && let Err(e) = self
.core .core
.rt .rt
.block_on(self.core.client.move_node(&uid, &new_parent_uid)) .block_on(self.core.client.rename_node(&uid, &newname, None))
{ {
error!(%uid, error = %e, "move failed"); error!(%uid, error = %e, "rename failed");
self.core.restore_replaced(victim.as_ref(), &newname); self.core.restore_replaced(victim.as_ref(), &newname);
reply.error(Errno::EIO); reply.error(Errno::EIO);
return; return;
} }
if newname != name { if newparent != parent {
// A move changes the node's encrypted name requirements. Proton's
// link-details read can remain briefly stale immediately after the
// move, making the following rename fail with InvalidRequirements
// ("out of date"). Re-read/retry that narrow eventual-consistency
// error; other refusals are stable and must surface immediately.
let mut attempts = 0u32; let mut attempts = 0u32;
let renamed = loop { let moved = loop {
match self match self
.core .core
.rt .rt
.block_on(self.core.client.rename_node(&uid, &newname, None)) .block_on(self.core.client.move_node(&uid, &new_parent_uid))
{ {
Ok(()) => break Ok(()), Ok(()) => break Ok(()),
Err(e) Err(e)
if newparent != parent if newname != name
&& api_code(&e) == Some(ResponseCode::InvalidRequirements) && api_code(&e) == Some(ResponseCode::InvalidRequirements)
&& attempts < 4 => && attempts < 20 =>
{ {
attempts += 1; attempts += 1;
std::thread::sleep(Duration::from_millis(100 * (1 << attempts))); // Requirement propagation routinely takes longer than
// three seconds. Keep a bounded ten-second window; a
// successful POSIX reply must mean both remote halves
// landed, not merely that the second half was queued.
std::thread::sleep(Duration::from_millis(500));
} }
Err(e) => break Err(e), Err(e) => break Err(e),
} }
}; };
if let Err(e) = renamed { if let Err(e) = moved {
error!(%uid, attempts, error = %e, "rename after move failed"); error!(%uid, attempts, error = %e, "move after rename failed");
// The move half already landed. Keep our inode model honest // The rename half landed in the source directory.
// even though the combined POSIX operation has to return EIO. let mut state = self.core.state.lock();
self.core let old_parent_uid = state
.state .entries
.lock() .get(&parent)
.relocate(ino, parent, newparent, &new_parent_uid, &name); .map(|entry| entry.uid.clone())
.unwrap_or_else(|| uid.clone());
state.rename_in_place(ino, parent, &old_parent_uid, &newname);
drop(state);
self.core.restore_replaced(victim.as_ref(), &newname); self.core.restore_replaced(victim.as_ref(), &newname);
reply.error(Errno::EIO); reply.error(Errno::EIO);
return; return;
@@ -1385,6 +1449,7 @@ impl ProtonFs {
} }
// Only remove the namespace entry after Drive accepted the mutation. // Only remove the namespace entry after Drive accepted the mutation.
// Otherwise an EIO response still made the path disappear locally. // Otherwise an EIO response still made the path disappear locally.
self.core.hidden.lock().insert(uid.clone());
self.core.state.lock().forget_or_unlink(&uid); self.core.state.lock().forget_or_unlink(&uid);
if !open_now { if !open_now {
self.core.discard_queued_ops(&uid); self.core.discard_queued_ops(&uid);
+434 -20
View File
@@ -35,6 +35,8 @@ use parking_lot::{Condvar, Mutex};
use std::collections::{HashMap, HashSet, VecDeque}; use std::collections::{HashMap, HashSet, VecDeque};
use std::ffi::OsStr; use std::ffi::OsStr;
use std::fs::File; use std::fs::File;
use std::io::Write;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::FileExt; use std::os::unix::fs::FileExt;
use std::os::unix::net::UnixListener; use std::os::unix::net::UnixListener;
use std::path::{Component, Path, PathBuf}; use std::path::{Component, Path, PathBuf};
@@ -54,13 +56,15 @@ use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite};
use pdfs_core::config::AppDirs; use pdfs_core::config::AppDirs;
use pdfs_core::control::{ use pdfs_core::control::{
ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo, ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
SearchHit, SyncFolderInfo, SyncPhase, SyncProgress, TransferDirection, SearchFilters, SearchHit, SearchSource, SyncFolderInfo, SyncPhase, SyncProgress,
TransferDirection,
}; };
use pdfs_core::db::{ use pdfs_core::db::{
Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, StoredNode, Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, StoredNode,
StoredSyncFolder, StoredTrash, StoredSyncFolder, StoredTrash,
}; };
use pdfs_core::localindex; use pdfs_core::localindex;
use pdfs_core::search::relevance_score;
use pdfs_core::{CoreError, CoreResult}; use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::api::ResponseCode; use proton_drive_rs::proton_sdk::api::ResponseCode;
use proton_drive_rs::proton_sdk::error::ProtonError; use proton_drive_rs::proton_sdk::error::ProtonError;
@@ -222,6 +226,9 @@ struct Core {
/// staged blob *is* the file's content and the remote still holds the old /// staged blob *is* the file's content and the remote still holds the old
/// revision; and the drain worker, which performs the uploads. /// revision; and the drain worker, which performs the uploads.
pending: Arc<Mutex<HashMap<NodeUid, PendingRevision>>>, pending: Arc<Mutex<HashMap<NodeUid, PendingRevision>>>,
/// Nodes removed through this daemon which a briefly stale remote listing
/// may still return. Uids are not reused; an explicit restore clears one.
hidden: Arc<Mutex<HashSet<NodeUid>>>,
/// Nudges the drain worker: set true and notify to have it re-examine the /// Nudges the drain worker: set true and notify to have it re-examine the
/// queue instead of waiting out its backoff. /// queue instead of waiting out its backoff.
drain_wake: Arc<(Mutex<bool>, Condvar)>, drain_wake: Arc<(Mutex<bool>, Condvar)>,
@@ -651,6 +658,7 @@ impl Core {
// Before the lock: a DB row carries the size the server last // Before the lock: a DB row carries the size the server last
// sealed, which a queued write is ahead of (B11). // sealed, which a queued write is ahead of (B11).
self.stamp_pending_sizes(&mut nodes); self.stamp_pending_sizes(&mut nodes);
let hidden = self.hidden.lock().clone();
let mut st = self.state.lock(); let mut st = self.state.lock();
if st.children.contains_key(&ino) { if st.children.contains_key(&ino) {
return Ok(()); return Ok(());
@@ -658,7 +666,7 @@ impl Core {
let mut child_inos = Vec::with_capacity(nodes.len()); let mut child_inos = Vec::with_capacity(nodes.len());
let mut needs_size = Vec::new(); let mut needs_size = Vec::new();
for node in nodes { for node in nodes {
if node.trashed || node.uid == folder_uid { if !node_visible(&node, &folder_uid, &hidden) {
continue; continue;
} }
if matches!( if matches!(
@@ -709,6 +717,7 @@ impl Core {
// Same as the DB path above: the remote's size for a file with a write // Same as the DB path above: the remote's size for a file with a write
// still queued is the pre-write one (B11). // still queued is the pre-write one (B11).
self.stamp_pending_sizes(&mut nodes); self.stamp_pending_sizes(&mut nodes);
let hidden = self.hidden.lock().clone();
let mut st = self.state.lock(); let mut st = self.state.lock();
// Lost the race? Another thread already populated it. // Lost the race? Another thread already populated it.
@@ -718,7 +727,7 @@ impl Core {
let mut child_inos = Vec::with_capacity(nodes.len()); let mut child_inos = Vec::with_capacity(nodes.len());
let filtered_nodes: Vec<Node> = nodes let filtered_nodes: Vec<Node> = nodes
.into_iter() .into_iter()
.filter(|node| !node.trashed && node.uid != folder_uid) .filter(|node| node_visible(node, &folder_uid, &hidden))
.collect(); .collect();
// Files whose real size the cheap enumeration could not read. Collected // Files whose real size the cheap enumeration could not read. Collected
// before interning so the upgrade below has the uids without re-walking. // before interning so the upgrade below has the uids without re-walking.
@@ -1297,6 +1306,21 @@ impl Core {
if size == base_size { if size == base_size {
return Ok(()); return Ok(());
} }
// A close immediately followed by truncate is common (`write`, close,
// `truncate(2)`) and usually beats the write-back debounce. In that
// case the remote is *not* our base: the complete staged revision is.
// Compose from those local bytes so the truncate can supersede the
// queued upload without fetching stale remote content or returning EIO.
let pending_base = self.pending.lock().get(&uid).cloned();
if pending_base
.as_ref()
.is_some_and(|pending| !pending.meta.complete)
{
// An incomplete pending blob still has holes referring to the
// remote revision. Stacking another incomplete transform on it is
// not representable safely; preserve the existing refusal.
return Err(Errno::EIO);
}
let (authored, complete) = if size == 0 { let (authored, complete) = if size == 0 {
// An empty file has no content to be missing. // An empty file has no content to be missing.
(Vec::new(), true) (Vec::new(), true)
@@ -1313,19 +1337,45 @@ impl Core {
error!(%uid, error = %e, "create scratch file for truncate failed"); error!(%uid, error = %e, "create scratch file for truncate failed");
Errno::EIO Errno::EIO
})?; })?;
if let Some(pending) = &pending_base
&& let Err(e) = copy_pending_for_truncate(pending, &file)
{
error!(%uid, source = %pending.path.display(), error = %e,
"copy pending revision for truncate failed");
let _ = std::fs::remove_file(&path);
return Err(Errno::EIO);
}
file.set_len(size).map_err(|e| { file.set_len(size).map_err(|e| {
error!(%uid, error = %e, "resize scratch file for truncate failed"); error!(%uid, error = %e, "resize scratch file for truncate failed");
let _ = std::fs::remove_file(&path); let _ = std::fs::remove_file(&path);
Errno::EIO Errno::EIO
})?; })?;
let meta = StagedWrite { let meta = match pending_base {
uid: uid.to_string(), Some(pending) => StagedWrite {
len: size, uid: uid.to_string(),
base_size, len: size,
base_mtime, base_size: pending.meta.len,
authored, base_mtime: pending.meta.base_mtime,
complete, authored: if size == 0 {
based_on: self.remote_baseline(&uid, base_mtime, base_size), Vec::new()
} else {
vec![(0, size)]
},
complete: true,
// Both local revisions descend from the same last-observed
// remote revision. Never replace this with the optimistic
// mtime/size currently published by State.
based_on: pending.meta.based_on,
},
None => StagedWrite {
uid: uid.to_string(),
len: size,
base_size,
base_mtime,
authored,
complete,
based_on: self.remote_baseline(&uid, base_mtime, base_size),
},
}; };
self.enqueue_staged_write(&uid, ino, &path, meta)?; self.enqueue_staged_write(&uid, ino, &path, meta)?;
debug!(%uid, size, complete, "queued truncate"); debug!(%uid, size, complete, "queued truncate");
@@ -1444,6 +1494,7 @@ impl Core {
error!(%uid, error = %e, "queueing trash failed"); error!(%uid, error = %e, "queueing trash failed");
Errno::EIO Errno::EIO
})?; })?;
self.hidden.lock().insert(uid.clone());
self.state.lock().forget(uid); self.state.lock().forget(uid);
self.cache.evict(uid); self.cache.evict(uid);
self.evict_reader(uid); self.evict_reader(uid);
@@ -1497,8 +1548,32 @@ impl Core {
); );
} }
Err(e) => { Err(e) => {
error!(%uid, error = %e, "staging write failed; bytes lost"); // `src` is the only known copy of bytes the kernel already
let _ = std::fs::remove_file(src); // accepted. A failure to publish them into staging must never
// turn into permission to delete them: leave the scratch file
// in place so startup recovery or a human can still salvage it.
error!(
%uid,
error = %e,
scratch = %src.display(),
"staging write failed; retaining the only scratch copy"
);
let name = {
let st = self.state.lock();
st.entries
.get(&ino)
.map(|entry| entry.node.name.clone())
.unwrap_or_default()
};
self.log_activity(
ActivityKind::Upload,
&name,
format!(
"write not queued and could not be staged; recovery copy retained at {}",
src.display()
),
false,
);
} }
} }
} }
@@ -1524,6 +1599,27 @@ impl Core {
Ok(out.into_inner()) Ok(out.into_inner())
} }
/// Download a whole file directly into `out`, without retaining its
/// plaintext in memory. This is the disk-backed twin of
/// [`Core::download_file_tracked`], and preserves the same transfer
/// accounting while allowing callers that only need an on-disk file to
/// keep memory use independent of file size.
fn download_file_tracked_to<W: Write>(
&self,
uid: &NodeUid,
name: &str,
total: u64,
out: W,
) -> std::result::Result<W, ProtonError> {
let guard = self
.transfers
.begin(name, uid.to_string(), TransferDirection::Download, total);
let mut out = CountingWriter::new(out, &guard);
self.rt
.block_on(self.client.download_file_to(uid, &mut out))?;
Ok(out.into_inner())
}
/// Like [`download_file_tracked`] for a photo, streaming through the photos /// Like [`download_file_tracked`] for a photo, streaming through the photos
/// client's [`download_photo_to`]. /// client's [`download_photo_to`].
/// ///
@@ -1760,6 +1856,24 @@ impl Core {
.collect()) .collect())
} }
/// Return an absolute path through the most specific live on-demand mount
/// covering `uid`. A node may be below several configured roots, so prefer
/// the root leaving the shortest descendant path.
fn mounted_search_path(&self, uid: &str) -> Option<String> {
let live_mounts = self.mounts.lock();
let folders = self.db.sync_folder_list().ok()?;
folders
.into_iter()
.filter(|folder| folder.mode == "ondemand" && live_mounts.contains_key(&folder.id))
.filter_map(|folder| {
let relative = self.db.path_relative_to(&folder.remote_uid, uid).ok()??;
let depth = relative.split('/').count();
Some((depth, PathBuf::from(folder.local_path).join(relative)))
})
.min_by_key(|(depth, _)| *depth)
.map(|(_, path)| path.to_string_lossy().into_owned())
}
/// Full-text search node names against the local SQLite index, mapping each /// Full-text search node names against the local SQLite index, mapping each
/// DB hit to the wire [`SearchHit`] (resolving live pin state from the cache, /// DB hit to the wire [`SearchHit`] (resolving live pin state from the cache,
/// which the DB doesn't track). Pure local lookup — never hits the network. /// which the DB doesn't track). Pure local lookup — never hits the network.
@@ -1778,6 +1892,8 @@ impl Core {
modified: h.node.modification_time, modified: h.node.modification_time,
pinned: self.cache.is_pinned(&h.node.uid), pinned: self.cache.is_pinned(&h.node.uid),
uid: h.node.uid.to_string(), uid: h.node.uid.to_string(),
mounted_path: self.mounted_search_path(&h.node.uid.to_string()),
score: 0,
}) })
.collect()) .collect())
} }
@@ -1798,10 +1914,96 @@ impl Core {
is_dir: h.is_dir, is_dir: h.is_dir,
size: h.size.max(0) as u64, size: h.size.max(0) as u64,
modified: h.mtime, modified: h.mtime,
score: 0,
}) })
.collect()) .collect())
} }
/// Search both prompt sources with identical query and per-source limits.
/// Keeping the composition in the daemon core gives every control surface
/// the same semantics while preserving the legacy source-specific methods.
fn search_v2(
&self,
query: &str,
limit: usize,
filters: &SearchFilters,
) -> CoreResult<(Vec<SearchHit>, Vec<LocalHit>)> {
if query.trim().is_empty() || limit == 0 {
return Ok((Vec::new(), Vec::new()));
}
// Candidate generation is intentionally broader than the final result
// cap: filtering and fuzzy rejection happen below, so limiting at the
// SQLite boundary to `limit` would recreate the old empty-filter bug.
let candidate_limit = limit.saturating_mul(10).clamp(100, 1_000);
let mut drive_hits = if filters.sources.contains(&SearchSource::Drive) {
self.db
.search_candidates(query, candidate_limit)
.map_err(|e| CoreError::from_api(&e, "search candidates"))?
.into_iter()
.filter(|hit| filters.kind.accepts(&hit.node.name, hit.node.is_folder()))
.filter_map(|hit| {
let parent = Path::new(&hit.path)
.parent()
.map_or_else(String::new, |path| path.display().to_string());
let pinned = self.cache.is_pinned(&hit.node.uid);
let score = relevance_score(query, &hit.node.name, &parent)?
+ if pinned { 250 } else { 0 };
Some(SearchHit {
name: hit.node.name.clone(),
path: hit.path,
is_dir: hit.node.is_folder(),
size: node_size(&hit.node),
modified: hit.node.modification_time,
pinned,
uid: hit.node.uid.to_string(),
mounted_path: self.mounted_search_path(&hit.node.uid.to_string()),
score,
})
})
.collect()
} else {
Vec::new()
};
let mut local_hits = if filters.sources.contains(&SearchSource::Local) {
self.db
.search_local_candidates(query, candidate_limit)
.map_err(|e| CoreError::from_api(&e, "local search candidates"))?
.into_iter()
.filter(|hit| filters.kind.accepts(&hit.name, hit.is_dir))
.filter_map(|hit| {
let parent = Path::new(&hit.path)
.parent()
.map_or_else(String::new, |path| path.display().to_string());
Some(LocalHit {
score: relevance_score(query, &hit.name, &parent)?,
name: hit.name,
path: hit.path,
is_dir: hit.is_dir,
size: hit.size.max(0) as u64,
modified: hit.mtime,
})
})
.collect()
} else {
Vec::new()
};
drive_hits.sort_by(|a, b| {
b.score
.cmp(&a.score)
.then_with(|| a.name.cmp(&b.name))
.then_with(|| a.path.cmp(&b.path))
});
local_hits.sort_by(|a, b| {
b.score
.cmp(&a.score)
.then_with(|| a.name.cmp(&b.name))
.then_with(|| a.path.cmp(&b.path))
});
drive_hits.truncate(limit);
local_hits.truncate(limit);
Ok((drive_hits, local_hits))
}
/// A page of the photos timeline (newest first), sliced out of the DB. /// A page of the photos timeline (newest first), sliced out of the DB.
/// `Ok(None)` when the account has no photos volume. /// `Ok(None)` when the account has no photos volume.
/// ///
@@ -1847,12 +2049,43 @@ impl Core {
if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) { if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
return Ok(p); return Ok(p);
} }
let bytes = self
.download_file_tracked(&uid, &name, size) // Keep the temporary file beside the final cache blob. `store_file`
.map_err(|e| CoreError::from_api(&e, "download"))?; // can then adopt it with a hard link and atomic rename, while the SDK
self.cache // streams decrypted blocks to disk instead of building a file-sized
.store(&uid, mtime, size, &bytes) // `Vec<u8>`. A unique name also keeps concurrent opens of the same node
.map_err(|e| CoreError::internal(format!("cache store: {e}")))?; // from sharing a partially written staging file.
static OPEN_TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
let seq = OPEN_TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
let tmp = self.cache.content_path(&uid).with_extension(format!(
"open-{}-{}-{seq}.tmp",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
));
let file = File::options()
.write(true)
.create_new(true)
.open(&tmp)
.map_err(|e| CoreError::internal(format!("create download temp: {e}")))?;
let file = match self.download_file_tracked_to(&uid, &name, size, file) {
Ok(file) => file,
Err(e) => {
let _ = std::fs::remove_file(&tmp);
return Err(CoreError::from_api(&e, "download"));
}
};
if let Err(e) = file.sync_all() {
let _ = std::fs::remove_file(&tmp);
return Err(CoreError::internal(format!("sync download temp: {e}")));
}
if let Err(e) = self.cache.store_file(&uid, mtime, size, &tmp) {
let _ = std::fs::remove_file(&tmp);
return Err(CoreError::internal(format!("cache store: {e}")));
}
let _ = std::fs::remove_file(&tmp);
Ok(self.cache.content_path(&uid)) Ok(self.cache.content_path(&uid))
} }
@@ -2132,6 +2365,12 @@ impl Core {
self.rt self.rt
.block_on(self.client.restore_nodes(&parsed)) .block_on(self.client.restore_nodes(&parsed))
.map_err(|e| CoreError::from_api(&e, "restore"))?; .map_err(|e| CoreError::from_api(&e, "restore"))?;
{
let mut hidden = self.hidden.lock();
for uid in &parsed {
hidden.remove(uid);
}
}
{ {
let mut st = self.state.lock(); let mut st = self.state.lock();
for parent in parents { for parent in parents {
@@ -2700,6 +2939,59 @@ fn count_noun(n: usize, one: &str, many: &str) -> String {
format!("{n} {}", if n == 1 { one } else { many }) format!("{n} {}", if n == 1 { one } else { many })
} }
/// Seed a truncate scratch file from a complete queued revision.
///
/// `File` writes use an explicit offset so this is independent of either
/// descriptor's cursor and remains correct when the destination came from the
/// cache's scratch allocator.
fn copy_pending_for_truncate(pending: &PendingRevision, destination: &File) -> std::io::Result<()> {
let source = File::open(&pending.path)?;
destination.set_len(0)?;
let mut offset = 0u64;
let mut buffer = vec![0u8; 1024 * 1024];
loop {
let count = source.read_at(&mut buffer, offset)?;
if count == 0 {
break;
}
destination.write_all_at(&buffer[..count], offset)?;
offset += count as u64;
}
destination.set_len(offset)
}
/// Whether a rename must be represented as a durable desired end state.
/// Offline operations and operations targeting a not-yet-uploaded directory
/// cannot use the API immediately. Online remote-to-remote operations stay
/// synchronous so a successful kernel reply means the remote namespace has
/// already reached the same end state.
fn rename_needs_queue(
online: bool,
destination_is_local: bool,
_parent_changed: bool,
_name_changed: bool,
) -> bool {
!online || destination_is_local
}
/// Convert one kernel pathname component into the UTF-8 name accepted by Drive.
/// Linux filesystems conventionally cap a component at `NAME_MAX` (255 bytes),
/// but the FUSE kernel path can pass a longer component through to userspace.
fn fuse_name(name: &OsStr) -> Result<String, Errno> {
let bytes = name.as_bytes();
if bytes.len() > 255 {
return Err(Errno::ENAMETOOLONG);
}
if bytes.is_empty() || bytes == b"." || bytes == b".." {
return Err(Errno::EINVAL);
}
name.to_str().map(str::to_owned).ok_or(Errno::EILSEQ)
}
fn node_visible(node: &Node, folder_uid: &NodeUid, hidden: &HashSet<NodeUid>) -> bool {
!node.trashed && node.uid != *folder_uid && !hidden.contains(&node.uid)
}
/// Bytes rendered with a binary unit and one decimal place (e.g. `"1.2 GB"`), /// Bytes rendered with a binary unit and one decimal place (e.g. `"1.2 GB"`),
/// for the activity log. Uses 1024-based steps but the shorter SI labels. /// for the activity log. Uses 1024-based steps but the shorter SI labels.
fn human_bytes(bytes: u64) -> String { fn human_bytes(bytes: u64) -> String {
@@ -3063,7 +3355,11 @@ mod pending_size_tests {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{Intervals, conflict_name, is_stale_mount}; use super::{
Intervals, PendingRevision, conflict_name, copy_pending_for_truncate, fuse_name,
is_stale_mount, node_visible, rename_needs_queue,
};
use pdfs_core::cache::{Baseline, StagedWrite};
/// The predicate must answer *only* for a dead FUSE connection. A healthy /// The predicate must answer *only* for a dead FUSE connection. A healthy
/// directory and an absent path are both "not stale" — widening it to any /// directory and an absent path are both "not stale" — widening it to any
@@ -3170,6 +3466,124 @@ mod tests {
assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]); assert_eq!(segs(&iv, 0, 30), vec![(0, 30, true)]);
} }
fn pending_revision(dir: &TestDir, bytes: &[u8], complete: bool) -> PendingRevision {
let path = dir.0.join("pending");
std::fs::write(&path, bytes).unwrap();
PendingRevision {
path,
meta: StagedWrite {
uid: "vol~file".into(),
len: bytes.len() as u64,
base_size: 3,
base_mtime: 42,
authored: if complete {
vec![(0, bytes.len() as u64)]
} else {
vec![(3, bytes.len() as u64)]
},
complete,
based_on: Some(Baseline {
mtime: 42,
size: 3,
hash: None,
}),
},
}
}
#[test]
fn complete_pending_revision_can_seed_immediate_shrink() {
let (_, dir) = state_test_helper();
let pending = pending_revision(&dir, b"0123456789", true);
let destination_path = dir.0.join("shrink");
let destination = std::fs::File::create(&destination_path).unwrap();
copy_pending_for_truncate(&pending, &destination).unwrap();
destination.set_len(4).unwrap();
assert_eq!(std::fs::read(destination_path).unwrap(), b"0123");
}
#[test]
fn complete_pending_revision_can_seed_immediate_growth_and_zero() {
let (_, dir) = state_test_helper();
let pending = pending_revision(&dir, b"0123456789", true);
let destination_path = dir.0.join("grow");
let destination = std::fs::File::create(&destination_path).unwrap();
copy_pending_for_truncate(&pending, &destination).unwrap();
destination.set_len(14).unwrap();
assert_eq!(
std::fs::read(&destination_path).unwrap(),
b"0123456789\0\0\0\0"
);
destination.set_len(0).unwrap();
assert!(std::fs::read(destination_path).unwrap().is_empty());
}
#[test]
fn incomplete_pending_revision_is_not_a_safe_truncate_base() {
let (_, dir) = state_test_helper();
let pending = pending_revision(&dir, b"\0\0\0authored", false);
assert!(!pending.meta.complete);
assert_ne!(pending.meta.authored, vec![(0, pending.meta.len)]);
}
#[test]
fn online_combined_cross_directory_rename_stays_synchronous() {
assert!(!rename_needs_queue(true, false, true, true));
assert!(rename_needs_queue(false, false, false, true));
assert!(rename_needs_queue(true, true, true, false));
}
#[test]
fn one_step_online_namespace_changes_stay_synchronous() {
assert!(!rename_needs_queue(true, false, false, true));
assert!(!rename_needs_queue(true, false, true, false));
assert!(!rename_needs_queue(true, false, false, false));
}
#[test]
fn fuse_names_enforce_linux_component_limits() {
use std::ffi::OsStr;
assert_eq!(
fuse_name(OsStr::new("x".repeat(255).as_str()))
.unwrap()
.len(),
255
);
assert_eq!(
fuse_name(OsStr::new("x".repeat(256).as_str()))
.expect_err("256-byte component must be refused")
.code(),
libc::ENAMETOOLONG
);
assert_eq!(
fuse_name(OsStr::new("."))
.expect_err("dot is not an ordinary component")
.code(),
libc::EINVAL
);
assert_eq!(
fuse_name(OsStr::new("unicodé-文件")).unwrap(),
"unicodé-文件"
);
}
#[test]
fn local_trash_tombstones_hide_stale_remote_entries() {
let folder_uid = NodeUid::new(VolumeId::from("vol"), LinkId::from("folder"));
let node = node_helper("child", "folder", "child", false);
assert!(node_visible(
&node,
&folder_uid,
&std::collections::HashSet::new()
));
assert!(!node_visible(
&node,
&folder_uid,
&std::collections::HashSet::from([node.uid.clone()])
));
}
use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId}; use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
use proton_drive_rs::{Node, NodeKind}; use proton_drive_rs::{Node, NodeKind};
+1
View File
@@ -192,6 +192,7 @@ pub fn mount(
db, db,
online: Arc::new(AtomicBool::new(online)), online: Arc::new(AtomicBool::new(online)),
pending: Arc::new(Mutex::new(HashMap::new())), pending: Arc::new(Mutex::new(HashMap::new())),
hidden: Arc::new(Mutex::new(HashSet::new())),
drain_wake: Arc::new((Mutex::new(false), Condvar::new())), drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
timeline_refreshing: Arc::new(AtomicBool::new(false)), timeline_refreshing: Arc::new(AtomicBool::new(false)),
trash_refreshing: Arc::new(AtomicBool::new(false)), trash_refreshing: Arc::new(AtomicBool::new(false)),
+11 -4
View File
@@ -219,10 +219,12 @@ impl State {
/// Check if a directory inode has any child nodes in memory or in the database. /// Check if a directory inode has any child nodes in memory or in the database.
pub(crate) fn has_children(&self, parent: u64) -> bool { pub(crate) fn has_children(&self, parent: u64) -> bool {
if let Some(kids) = self.children.get(&parent) // A resident listing is authoritative even when empty. Falling through
&& !kids.is_empty() // from `Some([])` to SQLite lets an obsolete child row contradict the
{ // listing `ensure_children` just established, so readdir says empty but
return true; // rmdir returns ENOTEMPTY forever.
if let Some(kids) = self.children.get(&parent) {
return !kids.is_empty();
} }
if let Some(entry) = self.entries.get(&parent) if let Some(entry) = self.entries.get(&parent)
&& let Ok(has_kids) = self.db.has_children(&entry.uid) && let Ok(has_kids) = self.db.has_children(&entry.uid)
@@ -607,6 +609,11 @@ mod tests {
st.db st.db
.upsert_node(&node("file2_uid", "dir_uid", "file2.txt", false)) .upsert_node(&node("file2_uid", "dir_uid", "file2.txt", false))
.unwrap(); .unwrap();
assert!(
!st.has_children(folder),
"a resident empty listing overrides stale DB children"
);
st.children.remove(&folder);
assert!(st.has_children(folder), "has child in db"); assert!(st.has_children(folder), "has child in db");
} }
} }
+146 -29
View File
@@ -55,9 +55,11 @@ use local_scan::is_write_mode;
use local_scan::open_for_write_set; use local_scan::open_for_write_set;
use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem}; use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem};
pub(crate) use planner::base_name; pub(crate) use planner::base_name;
#[cfg(test)]
use planner::conflict_path;
use planner::{ use planner::{
FilePlan, classification_order, conflict_path, filter_baseline, guard_local_wipe, join_rel, FilePlan, classification_order, conflict_path_with_suffix, filter_baseline, guard_local_wipe,
parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size, join_rel, parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size,
}; };
/// What a [`Pending`] op did, so the reconcile can update shared state on the /// What a [`Pending`] op did, so the reconcile can update shared state on the
/// engine thread (never inside a task). /// engine thread (never inside a task).
@@ -711,7 +713,17 @@ impl Core {
} }
}, },
FilePlan::DeleteLocal => { FilePlan::DeleteLocal => {
let _ = std::fs::remove_file(local_root.join(rel_to_path(rel))); if let Err(e) = std::fs::remove_file(local_root.join(rel_to_path(rel))) {
warn!(rel, error = %e, "sync: remove local file failed");
self.log_activity(
ActivityKind::Trash,
base_name(rel),
e.to_string(),
false,
);
outcome.errors += 1;
continue;
}
let _ = self.db.sync_entry_remove(folder_id, rel); let _ = self.db.sync_entry_remove(folder_id, rel);
self.log_activity(ActivityKind::Trash, base_name(rel), "removed locally", true); self.log_activity(ActivityKind::Trash, base_name(rel), "removed locally", true);
outcome.deleted += 1; outcome.deleted += 1;
@@ -758,7 +770,12 @@ impl Core {
// Deferred folder deletions, deepest first. // Deferred folder deletions, deepest first.
delete_local_dirs.sort_by_key(|p| std::cmp::Reverse(p.matches('/').count())); delete_local_dirs.sort_by_key(|p| std::cmp::Reverse(p.matches('/').count()));
for rel in delete_local_dirs { for rel in delete_local_dirs {
let _ = std::fs::remove_dir_all(local_root.join(rel_to_path(&rel))); if let Err(e) = std::fs::remove_dir_all(local_root.join(rel_to_path(&rel))) {
warn!(rel, error = %e, "sync: remove local folder failed");
self.log_activity(ActivityKind::Trash, base_name(&rel), e.to_string(), false);
outcome.errors += 1;
continue;
}
let _ = self.db.sync_entry_remove(folder_id, &rel); let _ = self.db.sync_entry_remove(folder_id, &rel);
self.log_activity( self.log_activity(
ActivityKind::Trash, ActivityKind::Trash,
@@ -802,18 +819,24 @@ impl Core {
out: &mut HashMap<String, LocalItem>, out: &mut HashMap<String, LocalItem>,
) -> Result<(), String> { ) -> Result<(), String> {
let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?; let entries = std::fs::read_dir(dir).map_err(|e| format!("read {}: {e}", dir.display()))?;
for entry in entries.filter_map(|e| e.ok()) { for entry in entries {
let entry = entry.map_err(|e| format!("read entry in {}: {e}", dir.display()))?;
let path = entry.path(); let path = entry.path();
let meta = match std::fs::symlink_metadata(&path) { let meta = std::fs::symlink_metadata(&path)
Ok(m) => m, .map_err(|e| format!("stat {}: {e}", path.display()))?;
Err(_) => continue, let stripped = path.strip_prefix(root).map_err(|e| {
}; format!(
let Ok(stripped) = path.strip_prefix(root) else { "local path {} escaped sync root {}: {e}",
continue; path.display(),
}; root.display()
let Some(rel) = stripped.to_str() else { )
continue; })?;
}; 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. // Ignore our own in-flight download temp files.
if rel.contains(".pdfs-tmp-") { if rel.contains(".pdfs-tmp-") {
continue; continue;
@@ -1153,12 +1176,9 @@ impl Core {
// pass), then take the remote version as the shared truth. // pass), then take the remote version as the shared truth.
let path = local_root.join(rel_to_path(rel)); let path = local_root.join(rel_to_path(rel));
if path.exists() { if path.exists() {
let conflict = conflict_path(&path, now_secs()); preserve_conflict_copy(&path, now_secs())
if let Err(e) = std::fs::rename(&path, &conflict) { .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
warn!(rel, error = %e, "sync: could not set aside conflict copy"); info!(rel, "sync: kept local changes as a conflict copy");
} else {
info!(rel, "sync: kept local changes as a conflict copy");
}
} }
self.download_file(folder_id, local_root, rel, uid, *mtime, *size) self.download_file(folder_id, local_root, rel, uid, *mtime, *size)
.await?; .await?;
@@ -1171,8 +1191,7 @@ impl Core {
// evicted. Push the conflict copy now, while it still exists, and // evicted. Push the conflict copy now, while it still exists, and
// leave the remote file holding the original path. // leave the remote file holding the original path.
let path = local_root.join(rel_to_path(rel)); let path = local_root.join(rel_to_path(rel));
let conflict = conflict_path(&path, now_secs()); let conflict = preserve_conflict_copy(&path, now_secs())
std::fs::rename(&path, &conflict)
.map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?; .map_err(|e| format!("set aside conflict copy for {rel}: {e}"))?;
let name = conflict let name = conflict
.file_name() .file_name()
@@ -1297,13 +1316,24 @@ impl Core {
let _ = std::fs::remove_file(&tmp); let _ = std::fs::remove_file(&tmp);
format!("download {rel}: {e}") format!("download {rel}: {e}")
})?; })?;
out.into_inner().flush().ok(); let mut out = out.into_inner();
out.flush().map_err(|e| format!("flush tmp {rel}: {e}"))?;
out.sync_all().map_err(|e| format!("sync tmp {rel}: {e}"))?;
} }
std::fs::rename(&tmp, &path).map_err(|e| format!("place {rel}: {e}"))?; std::fs::rename(&tmp, &path).map_err(|e| format!("place {rel}: {e}"))?;
// Match local mtime to the remote's so neither side looks "changed" next pass. // Match local mtime to the remote's so neither side looks "changed" next pass.
if let Ok(f) = std::fs::File::options().write(true).open(&path) { let f = std::fs::File::options()
let _ = .write(true)
f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.max(0) as u64)); .open(&path)
.map_err(|e| format!("open placed file {rel}: {e}"))?;
f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.max(0) as u64))
.map_err(|e| format!("set mtime for {rel}: {e}"))?;
f.sync_all()
.map_err(|e| format!("sync placed file {rel}: {e}"))?;
if let Some(parent) = path.parent() {
std::fs::File::open(parent)
.and_then(|dir| dir.sync_all())
.map_err(|e| format!("sync directory for {rel}: {e}"))?;
} }
self.record_file_baseline(folder_id, rel, &path, uid).await self.record_file_baseline(folder_id, rel, &path, uid).await
} }
@@ -1350,6 +1380,46 @@ impl Core {
} }
} }
/// 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> {
let metadata = std::fs::metadata(path)?;
for suffix in 0..=u32::MAX {
let conflict = conflict_path_with_suffix(path, stamp, suffix);
let mut source = std::fs::File::open(path)?;
let mut target = match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&conflict)
{
Ok(file) => file,
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(e) => return Err(e),
};
if let Err(e) = std::io::copy(&mut source, &mut target).and_then(|_| target.sync_all()) {
let _ = std::fs::remove_file(&conflict);
return Err(e);
}
target.set_permissions(metadata.permissions())?;
if let Ok(modified) = metadata.modified() {
target.set_modified(modified)?;
}
target.sync_all()?;
sync_parent_dir(&conflict)?;
std::fs::remove_file(path)?;
sync_parent_dir(&conflict)?;
return Ok(conflict);
}
unreachable!("u32 conflict suffix space exhausted")
}
fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
match path.parent() {
Some(parent) => std::fs::File::open(parent)?.sync_all(),
None => Ok(()),
}
}
// ---- helpers -------------------------------------------------------------- // ---- helpers --------------------------------------------------------------
/// A remote node's `(mtime, size)` change signature. Size prefers the plaintext /// A remote node's `(mtime, size)` change signature. Size prefers the plaintext
@@ -1609,9 +1679,10 @@ mod tests {
guard_local_wipe(&base, &one).is_ok(), guard_local_wipe(&base, &one).is_ok(),
"one survivor means the tree is readable; the rest are real deletions" "one survivor means the tree is readable; the rest are real deletions"
); );
// A one-entry baseline is a genuine single delete, not a wipe to detect. // A one-entry folder can disappear because its mount or parent became
// unavailable just as easily as a larger tree can.
let single: HashMap<String, ()> = [("a.txt".to_string(), ())].into_iter().collect(); let single: HashMap<String, ()> = [("a.txt".to_string(), ())].into_iter().collect();
assert!(guard_local_wipe(&single, &none).is_ok()); assert!(guard_local_wipe(&single, &none).is_err());
// Nothing synced yet: a first pass over an empty folder is not a wipe. // Nothing synced yet: a first pass over an empty folder is not a wipe.
assert!(guard_local_wipe(&none, &none).is_ok()); assert!(guard_local_wipe(&none, &none).is_ok());
// Local paths that are not in the baseline are new uploads, and do not // Local paths that are not in the baseline are new uploads, and do not
@@ -1781,6 +1852,52 @@ mod tests {
assert_eq!(no_ext, PathBuf::from("/home/me/README (sync-conflict 42)")); assert_eq!(no_ext, PathBuf::from("/home/me/README (sync-conflict 42)"));
} }
#[test]
fn conflict_preservation_never_replaces_an_existing_copy() {
let dir = sync_test_dir("conflict-collision");
let source = dir.join("notes.txt");
let existing = conflict_path(&source, 1700);
std::fs::write(&source, b"new local edit").unwrap();
std::fs::write(&existing, b"older conflict").unwrap();
let preserved = preserve_conflict_copy(&source, 1700).unwrap();
assert_eq!(std::fs::read(&existing).unwrap(), b"older conflict");
assert_eq!(std::fs::read(&preserved).unwrap(), b"new local edit");
assert_eq!(
preserved.file_name().unwrap(),
"notes (sync-conflict 1700-1).txt"
);
assert!(!source.exists());
std::fs::remove_dir_all(dir).unwrap();
}
#[test]
fn conflict_preservation_failure_keeps_the_source() {
let dir = sync_test_dir("conflict-failure");
// A directory cannot be copied through the regular-file preservation
// path, deterministically forcing publication to fail after the source
// has been discovered.
let source = dir.join("not-a-file.txt");
std::fs::create_dir(&source).unwrap();
assert!(preserve_conflict_copy(&source, 1700).is_err());
assert!(source.is_dir(), "failure must not remove the source");
assert!(!conflict_path(&source, 1700).exists());
std::fs::remove_dir_all(dir).unwrap();
}
fn sync_test_dir(label: &str) -> PathBuf {
let nonce = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap()
.as_nanos();
let path =
std::env::temp_dir().join(format!("pdfs-sync-{label}-{}-{nonce}", std::process::id()));
std::fs::create_dir(&path).unwrap();
path
}
// ---- open-for-write detection tests ----------------------------------- // ---- open-for-write detection tests -----------------------------------
#[test] #[test]
+15 -3
View File
@@ -99,7 +99,7 @@ pub(super) fn guard_local_wipe<B, L>(
baseline: &HashMap<String, B>, baseline: &HashMap<String, B>,
local: &HashMap<String, L>, local: &HashMap<String, L>,
) -> Result<(), String> { ) -> Result<(), String> {
if baseline.len() >= 2 && baseline.keys().all(|rel| !local.contains_key(rel)) { if !baseline.is_empty() && baseline.keys().all(|rel| !local.contains_key(rel)) {
return Err(format!( return Err(format!(
"every one of the {} synced paths is missing locally; refusing to trash \ "every one of the {} synced paths is missing locally; refusing to trash \
them on Drive. Check that the folder is mounted and readable.", them on Drive. Check that the folder is mounted and readable.",
@@ -162,12 +162,24 @@ pub(super) fn rel_to_path(rel: &str) -> PathBuf {
} }
/// The name for a conflict copy of `path`, e.g. `notes (sync-conflict 1700000000).txt`. /// The name for a conflict copy of `path`, e.g. `notes (sync-conflict 1700000000).txt`.
#[cfg(test)]
pub(super) fn conflict_path(path: &Path, stamp: i64) -> PathBuf { pub(super) fn conflict_path(path: &Path, stamp: i64) -> PathBuf {
conflict_path_with_suffix(path, stamp, 0)
}
/// As [`conflict_path`], with a deterministic suffix used when that name already
/// exists. This keeps conflict preservation from ever replacing an older copy.
pub(super) fn conflict_path_with_suffix(path: &Path, stamp: i64, suffix: u32) -> PathBuf {
let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file"); let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file");
let ext = path.extension().and_then(|s| s.to_str()); let ext = path.extension().and_then(|s| s.to_str());
let suffix = if suffix == 0 {
String::new()
} else {
format!("-{suffix}")
};
let name = match ext { let name = match ext {
Some(ext) => format!("{stem} (sync-conflict {stamp}).{ext}"), Some(ext) => format!("{stem} (sync-conflict {stamp}{suffix}).{ext}"),
None => format!("{stem} (sync-conflict {stamp})"), None => format!("{stem} (sync-conflict {stamp}{suffix})"),
}; };
match path.parent() { match path.parent() {
Some(dir) => dir.join(name), Some(dir) => dir.join(name),
+89
View File
@@ -0,0 +1,89 @@
//! Shared Drive-result activation policy for the GTK browser and quick prompt.
//!
//! Media should be handed to its mounted path so players can issue range reads
//! through FUSE. Other files retain the materialize-then-open behaviour because
//! many desktop applications expect a stable ordinary file.
use std::path::{Path, PathBuf};
use pdfs_core::control::PhotoKind;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DriveActivation {
Folder,
MountedMedia,
Materialize,
}
pub(crate) fn drive_activation(name: &str, is_dir: bool) -> DriveActivation {
if is_dir {
DriveActivation::Folder
} else if is_streamable_media(name) {
DriveActivation::MountedMedia
} else {
DriveActivation::Materialize
}
}
pub(crate) fn mounted_path(mountpoint: &Path, relative_path: &str) -> PathBuf {
mountpoint.join(relative_path)
}
fn is_streamable_media(name: &str) -> bool {
if PhotoKind::classify(Some(name), None) == PhotoKind::Video {
return true;
}
let extension = Path::new(name)
.extension()
.and_then(|value| value.to_str())
.unwrap_or_default();
matches!(
extension.to_ascii_lowercase().as_str(),
"mp3" | "flac" | "wav" | "ogg" | "opus" | "m4a" | "aac"
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn folders_use_the_mount_regardless_of_extension() {
assert_eq!(
drive_activation("recordings.mp4", true),
DriveActivation::Folder
);
}
#[test]
fn media_extensions_use_fuse_path_case_insensitively() {
for name in ["movie.MKV", "voice.opus", "song.FLAC", "clip.m4v"] {
assert_eq!(
drive_activation(name, false),
DriveActivation::MountedMedia,
"{name}"
);
}
}
#[test]
fn ordinary_files_are_materialized() {
for name in ["report.pdf", "notes.txt", "archive.zip", "no-extension"] {
assert_eq!(
drive_activation(name, false),
DriveActivation::Materialize,
"{name}"
);
}
}
#[test]
fn mounted_paths_preserve_relative_components() {
assert_eq!(
mounted_path(Path::new("/mnt/drive"), "Videos/a movie.mkv"),
Path::new("/mnt/drive/Videos/a movie.mkv")
);
}
}
+2
View File
@@ -1,3 +1,5 @@
#[path = "../activation.rs"]
pub(crate) mod activation;
pub(crate) mod pages; pub(crate) mod pages;
pub(crate) mod widgets; pub(crate) mod widgets;
+14 -15
View File
@@ -1,3 +1,4 @@
use crate::activation::{DriveActivation, drive_activation, mounted_path};
use crate::*; use crate::*;
pub(crate) struct BrowserState { pub(crate) struct BrowserState {
@@ -493,9 +494,9 @@ pub(crate) fn show_context_menu(ui: &Rc<Ui>, entry: &DirEntry, anchor: &gtk4::Bo
popover.set_parent(anchor); popover.set_parent(anchor);
popover.connect_closed(|p| p.unparent()); popover.connect_closed(|p| p.unparent());
// Videos lead with Play (stream from the mount, no download); the plain Open // Media leads with Play (stream from the mount, no download); the plain Open
// below still downloads a local copy for anyone who wants one. // below still downloads a local copy for anyone who wants one.
if is_video_entry(entry) { if is_streamable_media_entry(entry) {
let play = menu_item("Play (stream)", "media-playback-start-symbolic"); let play = menu_item("Play (stream)", "media-playback-start-symbolic");
let ui_play = ui.clone(); let ui_play = ui.clone();
let entry_play = entry.clone(); let entry_play = entry.clone();
@@ -514,8 +515,8 @@ pub(crate) fn show_context_menu(ui: &Rc<Ui>, entry: &DirEntry, anchor: &gtk4::Bo
open.connect_clicked(move |_| { open.connect_clicked(move |_| {
pop.popdown(); pop.popdown();
// Open always means "download a local copy and hand off", even for a // Open always means "download a local copy and hand off", even for a
// video — `activate_entry` would otherwise stream it. // media file — `activate_entry` would otherwise stream it.
if is_video_entry(&entry_open) { if is_streamable_media_entry(&entry_open) {
download_and_open(&ui_open, &entry_open); download_and_open(&ui_open, &entry_open);
} else { } else {
activate_entry(&ui_open, &entry_open); activate_entry(&ui_open, &entry_open);
@@ -610,14 +611,12 @@ pub(crate) fn entry_at(model: Option<&impl IsA<gio::ListModel>>, pos: u32) -> Op
Some(entry) Some(entry)
} }
/// Whether an entry's name marks it as a video, reusing the same classifier the /// Whether an entry can be opened through FUSE instead of fully materialized.
/// Photos page splits on — so the file browser and the gallery agree on what a pub(crate) fn is_streamable_media_entry(entry: &DirEntry) -> bool {
/// video is. drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia
pub(crate) fn is_video_entry(entry: &DirEntry) -> bool {
!entry.is_dir && PhotoKind::classify(Some(&entry.name), None) == PhotoKind::Video
} }
/// Stream a video straight from the mount, no download. Drive folders *are* part /// Stream media straight from the mount, no download. Drive folders *are* part
/// of the FUSE mount, so a player pointed at `<mountpoint>/<rel>` reads the file /// of the FUSE mount, so a player pointed at `<mountpoint>/<rel>` reads the file
/// through [`Core::read_range_remote`] — 4 MB blocks fetched on demand as it /// through [`Core::read_range_remote`] — 4 MB blocks fetched on demand as it
/// seeks and buffers — instead of waiting for the whole file to land like /// seeks and buffers — instead of waiting for the whole file to land like
@@ -626,11 +625,11 @@ pub(crate) fn is_video_entry(entry: &DirEntry) -> bool {
pub(crate) fn stream_entry(ui: &Rc<Ui>, entry: &DirEntry) { pub(crate) fn stream_entry(ui: &Rc<Ui>, entry: &DirEntry) {
let rel = entry_rel(ui, entry); let rel = entry_rel(ui, entry);
let mountpoint = ui.dirs.resolved_mountpoint(&ui.dirs.load_config()); let mountpoint = ui.dirs.resolved_mountpoint(&ui.dirs.load_config());
let abs = mountpoint.join(&rel); let abs = mounted_path(&mountpoint, &rel);
let Some(path) = abs.to_str() else { let Some(path) = abs.to_str() else {
toast_error( toast_error(
ui, ui,
"Couldn't play video", "Couldn't play media",
"The file path isn't valid UTF-8.", "The file path isn't valid UTF-8.",
); );
return; return;
@@ -639,7 +638,7 @@ pub(crate) fn stream_entry(ui: &Rc<Ui>, entry: &DirEntry) {
play_external(path); play_external(path);
} }
/// Open an entry the Nautilus way: folders descend, videos stream from the mount, /// Open an entry the Nautilus way: folders descend, media streams from the mount,
/// other files download-and-open. /// other files download-and-open.
pub(crate) fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) { pub(crate) fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) {
let rel = entry_rel(ui, entry); let rel = entry_rel(ui, entry);
@@ -651,8 +650,8 @@ pub(crate) fn activate_entry(ui: &Rc<Ui>, entry: &DirEntry) {
} }
*ui.browser.path.borrow_mut() = rel; *ui.browser.path.borrow_mut() = rel;
load_browser(ui); load_browser(ui);
} else if is_video_entry(entry) { } else if drive_activation(&entry.name, entry.is_dir) == DriveActivation::MountedMedia {
// A video streams rather than downloads: that is exactly the "play it, // Media streams rather than downloads: that is exactly the "play it,
// don't fetch the whole thing" behaviour this is for. // don't fetch the whole thing" behaviour this is for.
stream_entry(ui, entry); stream_entry(ui, entry);
} else { } else {
+262 -83
View File
@@ -1,11 +1,11 @@
//! `pdfs-prompt` — the launcher: one search field over both Proton Drive and the //! `pdfs-prompt` — the launcher: one search field over both Proton Drive and the
//! files on this machine, styled after Google Drive's search overlay. //! files on this machine, styled after Google Drive's search overlay.
//! //!
//! Bind it to a system shortcut (e.g. in Hyprland) for a quick HUD search. It is //! Bind it to a system shortcut (e.g. in Hyprland) for a quick HUD search. The
//! a short-lived process: it must draw *now*, so nothing here blocks the main //! application is single-instance and keeps its window alive between summons,
//! loop. Every daemon round-trip runs on a worker thread and lands back through //! so repeat activations only reset and present the existing widget tree. Every
//! an async channel; the two searches (Drive and local) are fired in parallel and //! daemon round-trip runs on a worker thread and lands back through an async
//! each section renders as its own reply arrives. //! channel. Drive and local lookup share one request so a result set paints once.
use std::cell::{Cell, RefCell}; use std::cell::{Cell, RefCell};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
@@ -17,7 +17,12 @@ use adw::prelude::*;
use gtk4::{gio, glib}; use gtk4::{gio, glib};
use pdfs_core::config::AppDirs; use pdfs_core::config::AppDirs;
use pdfs_core::control::{LocalHit, Request, Response, SearchHit, send}; use pdfs_core::control::{
LocalHit, Request, Response, SearchFilters, SearchHit, SearchKind, SearchSource, send,
};
mod activation;
use activation::{DriveActivation, drive_activation, mounted_path};
const APP_ID: &str = "io.narl.proton-drive-linux-prompt"; const APP_ID: &str = "io.narl.proton-drive-linux-prompt";
@@ -60,6 +65,16 @@ impl Filter {
Filter::Media => !is_dir && is_media(name), Filter::Media => !is_dir && is_media(name),
} }
} }
fn search_kind(self) -> SearchKind {
match self {
Self::All => SearchKind::All,
Self::Folders => SearchKind::Folders,
Self::Documents => SearchKind::Documents,
Self::Images => SearchKind::Images,
Self::Media => SearchKind::Media,
}
}
} }
fn extension(name: &str) -> String { fn extension(name: &str) -> String {
@@ -174,6 +189,13 @@ impl Hit {
matches!(self, Hit::Drive(h) if h.pinned) matches!(self, Hit::Drive(h) if h.pinned)
} }
fn score(&self) -> i64 {
match self {
Self::Drive(hit) => hit.score,
Self::Local(hit) => hit.score,
}
}
/// Stable identity used to preserve the keyboard cursor across re-renders. /// Stable identity used to preserve the keyboard cursor across re-renders.
/// The bool keeps a Drive and a local hit at the same path distinct. /// The bool keeps a Drive and a local hit at the same path distinct.
fn key(&self) -> (bool, String) { fn key(&self) -> (bool, String) {
@@ -184,6 +206,15 @@ impl Hit {
} }
} }
fn rank_hits(hits: &mut [Hit]) {
hits.sort_by(|a, b| {
b.score()
.cmp(&a.score())
.then_with(|| a.name().cmp(b.name()))
.then_with(|| a.key().cmp(&b.key()))
});
}
/// Everything the window needs to render and act, in one place so the many /// Everything the window needs to render and act, in one place so the many
/// callbacks below capture a single `Rc` instead of a dozen clones each. /// callbacks below capture a single `Rc` instead of a dozen clones each.
struct Ui { struct Ui {
@@ -195,6 +226,7 @@ struct Ui {
stack: gtk4::Stack, stack: gtk4::Stack,
scroller: gtk4::ScrolledWindow, scroller: gtk4::ScrolledWindow,
results: gtk4::Box, results: gtk4::Box,
search_section: Section,
drive_section: Section, drive_section: Section,
local_section: Section, local_section: Section,
placeholder: adw::StatusPage, placeholder: adw::StatusPage,
@@ -217,8 +249,7 @@ struct Ui {
/// Monotonic query id. A reply whose id is stale (the user typed again while /// Monotonic query id. A reply whose id is stale (the user typed again while
/// it was in flight) is dropped instead of overwriting fresher results. /// it was in flight) is dropped instead of overwriting fresher results.
query_id: Cell<u64>, query_id: Cell<u64>,
/// In-flight searches for the current query id, so the spinner stops only /// In-flight requests for the current query id.
/// once both sections have answered.
pending: Cell<u8>, pending: Cell<u8>,
/// True once a Drive open is running: the window is about to close, so /// True once a Drive open is running: the window is about to close, so
/// further input is ignored. /// further input is ignored.
@@ -229,6 +260,10 @@ struct Ui {
/// against the selected row once the fresh results settle, rather than being /// against the selected row once the fresh results settle, rather than being
/// dropped — so a plain "type, Enter" always opens something. /// dropped — so a plain "type, Enter" always opens something.
open_pending: Cell<bool>, open_pending: Cell<bool>,
/// Prevent lifecycle-driven entry resets from also scheduling the normal
/// debounced empty-query request; activation performs one explicit daemon
/// bootstrap instead.
suppress_entry_change: Cell<bool>,
/// The window, so a deferred open (above) can reach it from `render`. /// The window, so a deferred open (above) can reach it from `render`.
window: RefCell<Option<adw::ApplicationWindow>>, window: RefCell<Option<adw::ApplicationWindow>>,
} }
@@ -301,17 +336,33 @@ fn main() -> glib::ExitCode {
.init(); .init();
let app = adw::Application::builder().application_id(APP_ID).build(); let app = adw::Application::builder().application_id(APP_ID).build();
// GtkApplication normally remains active while it owns a window, including
// when that window is hidden. Hold it explicitly as well: residency is a
// product requirement here, not an incidental window-lifetime side effect.
let _resident = app.hold();
app.connect_startup(|_| load_css()); app.connect_startup(|_| load_css());
app.connect_activate(build_window); let prompt: Rc<RefCell<Option<Rc<Ui>>>> = Rc::new(RefCell::new(None));
app.connect_activate(move |app| {
let ui = if let Some(ui) = prompt.borrow().clone() {
ui
} else {
let Some(ui) = build_window(app) else {
return;
};
*prompt.borrow_mut() = Some(ui.clone());
ui
};
ui.activate();
});
app.run() app.run()
} }
fn build_window(app: &adw::Application) { fn build_window(app: &adw::Application) -> Option<Rc<Ui>> {
let dirs = match AppDirs::new() { let dirs = match AppDirs::new() {
Ok(d) => d, Ok(d) => d,
Err(e) => { Err(e) => {
tracing::error!("cannot resolve app dirs: {e}"); tracing::error!("cannot resolve app dirs: {e}");
return; return None;
} }
}; };
@@ -362,8 +413,10 @@ fn build_window(app: &adw::Application) {
let results = gtk4::Box::new(gtk4::Orientation::Vertical, 0); let results = gtk4::Box::new(gtk4::Orientation::Vertical, 0);
results.add_css_class("results"); results.add_css_class("results");
let search_section = Section::new("Best matches", "system-search-symbolic");
let drive_section = Section::new("Proton Drive", "folder-remote-symbolic"); let drive_section = Section::new("Proton Drive", "folder-remote-symbolic");
let local_section = Section::new("This computer", "drive-harddisk-symbolic"); let local_section = Section::new("This computer", "drive-harddisk-symbolic");
search_section.attach(&results);
drive_section.attach(&results); drive_section.attach(&results);
local_section.attach(&results); local_section.attach(&results);
@@ -430,6 +483,7 @@ fn build_window(app: &adw::Application) {
stack, stack,
scroller, scroller,
results, results,
search_section,
drive_section, drive_section,
local_section, local_section,
placeholder, placeholder,
@@ -446,6 +500,7 @@ fn build_window(app: &adw::Application) {
opening: Cell::new(false), opening: Cell::new(false),
indexing: Cell::new(false), indexing: Cell::new(false),
open_pending: Cell::new(false), open_pending: Cell::new(false),
suppress_entry_change: Cell::new(false),
window: RefCell::new(None), window: RefCell::new(None),
}); });
*ui.window.borrow_mut() = Some(window.clone()); *ui.window.borrow_mut() = Some(window.clone());
@@ -477,15 +532,17 @@ fn build_window(app: &adw::Application) {
(&ui.local_section.list, true), (&ui.local_section.list, true),
] { ] {
let ui_row = ui.clone(); let ui_row = ui.clone();
let window_row = window.clone();
list.connect_row_activated(move |_, row| { list.connect_row_activated(move |_, row| {
let offset = if is_local { ui_row.drive_count() } else { 0 }; let offset = if is_local { ui_row.drive_count() } else { 0 };
ui_row.open(offset + row.index() as usize, &window_row); ui_row.open(offset + row.index() as usize);
}); });
} }
let ui_row = ui.clone();
ui.search_section
.list
.connect_row_activated(move |_, row| ui_row.open(row.index() as usize));
let ui_key = ui.clone(); let ui_key = ui.clone();
let window_key = window.clone();
let keys = gtk4::EventControllerKey::new(); let keys = gtk4::EventControllerKey::new();
keys.connect_key_pressed(move |_, key, _, state| { keys.connect_key_pressed(move |_, key, _, state| {
if ui_key.opening.get() { if ui_key.opening.get() {
@@ -496,7 +553,7 @@ fn build_window(app: &adw::Application) {
// Spotlight-style: first Escape clears a non-empty query, a // Spotlight-style: first Escape clears a non-empty query, a
// second (or an already-empty box) dismisses the launcher. // second (or an already-empty box) dismisses the launcher.
if ui_key.entry.text().is_empty() { if ui_key.entry.text().is_empty() {
window_key.close(); ui_key.dismiss();
} else { } else {
ui_key.entry.set_text(""); ui_key.entry.set_text("");
} }
@@ -532,19 +589,41 @@ fn build_window(app: &adw::Application) {
// focused entry is never reached. Enter with focus in the list is covered by // focused entry is never reached. Enter with focus in the list is covered by
// `row_activated` above. // `row_activated` above.
let ui_activate = ui.clone(); let ui_activate = ui.clone();
let window_activate = window.clone();
entry.connect_activate(move |_| { entry.connect_activate(move |_| {
if ui_activate.opening.get() { if ui_activate.opening.get() {
return; return;
} }
// Fall back to the top row when nothing is explicitly selected (e.g. // Fall back to the top row when nothing is explicitly selected (e.g.
// results haven't rendered yet); `open` guards the rest. // results haven't rendered yet); `open` guards the rest.
ui_activate.open(ui_activate.cursor.get().unwrap_or(0), &window_activate); ui_activate.open(ui_activate.cursor.get().unwrap_or(0));
}); });
// Some GTK/input-method combinations consume the physical Return binding in
// GtkText without emitting Entry::activate. Capture only Return ahead of the
// text widget and leave every other key to normal GTK/IME handling. Stopping
// propagation also prevents the activate signal above from opening twice;
// that signal remains the programmatic/accessibility activation path.
let ui_return = ui.clone();
let return_key = gtk4::EventControllerKey::new();
return_key.set_propagation_phase(gtk4::PropagationPhase::Capture);
return_key.connect_key_pressed(move |_, key, _, _| {
if matches!(key, gtk4::gdk::Key::Return | gtk4::gdk::Key::KP_Enter) {
if !ui_return.opening.get() {
ui_return.open(ui_return.cursor.get().unwrap_or(0));
}
glib::Propagation::Stop
} else {
glib::Propagation::Proceed
}
});
entry.add_controller(return_key);
let debounce: Rc<RefCell<Option<glib::SourceId>>> = Rc::new(RefCell::new(None)); let debounce: Rc<RefCell<Option<glib::SourceId>>> = Rc::new(RefCell::new(None));
let ui_changed = ui.clone(); let ui_changed = ui.clone();
entry.connect_changed(move |entry| { entry.connect_changed(move |entry| {
if ui_changed.suppress_entry_change.get() {
return;
}
// A fresh keystroke supersedes any Enter that was waiting on results: // A fresh keystroke supersedes any Enter that was waiting on results:
// the user is still refining the query, not asking to open yet. // the user is still refining the query, not asking to open yet.
ui_changed.open_pending.set(false); ui_changed.open_pending.set(false);
@@ -564,22 +643,59 @@ fn build_window(app: &adw::Application) {
let ui_retry = ui.clone(); let ui_retry = ui.clone();
retry.connect_clicked(move |_| ui_retry.connect()); retry.connect_clicked(move |_| ui_retry.connect());
// A launcher is modal by convention: losing focus dismisses it. // Keep the widget tree alive when the compositor asks the window to close.
window.connect_is_active_notify(|w| { // This also avoids running GTK's destruction path while async replies still
if !w.is_active() { // hold references to prompt widgets.
w.close(); let ui_close = ui.clone();
} window.connect_close_request(move |_| {
ui_close.dismiss();
glib::Propagation::Stop
}); });
let entry_map = entry.clone(); let entry_map = entry.clone();
window.connect_map(move |_| { window.connect_map(move |_| {
entry_map.grab_focus(); entry_map.grab_focus();
}); });
ui.connect(); Some(ui)
window.present();
} }
impl Ui { impl Ui {
/// Reset and present the resident prompt in response to GApplication
/// activation. A second invocation while a file is materialising merely
/// raises the progress window; it must not make the in-flight open mutable.
fn activate(self: &Rc<Self>) {
let Some(window) = self.window.borrow().clone() else {
return;
};
if !self.opening.get() {
// Supersede replies from the previous showing before clearing the
// query. The daemon work may finish, but it can no longer repaint
// the newly activated prompt with stale rows.
self.query_id.set(self.query_id.get() + 1);
self.pending.set(0);
self.open_pending.set(false);
self.spinner.stop();
self.spinner.set_visible(false);
self.suppress_entry_change.set(true);
self.entry.set_text("");
self.suppress_entry_change.set(false);
self.set_filter(Filter::All);
self.connect();
}
window.present();
self.entry.grab_focus();
}
/// Dismiss without destroying the application window. GTK continues to
/// associate the hidden window with the single application instance, so a
/// later desktop shortcut invocation is a cheap `activate` round-trip.
fn dismiss(&self) {
self.open_pending.set(false);
if let Some(window) = self.window.borrow().as_ref() {
window.set_visible(false);
}
}
/// Ask the daemon for status and, if it answers, load the suggested (pinned) /// Ask the daemon for status and, if it answers, load the suggested (pinned)
/// files. Runs off-thread: a dead daemon must not freeze the first frame. /// files. Runs off-thread: a dead daemon must not freeze the first frame.
fn connect(self: &Rc<Self>) { fn connect(self: &Rc<Self>) {
@@ -640,6 +756,8 @@ impl Ui {
modified: 0, modified: 0,
pinned: true, pinned: true,
uid: pin.uid, uid: pin.uid,
mounted_path: None,
score: 0,
}) })
.collect(), .collect(),
_ => Vec::new(), _ => Vec::new(),
@@ -650,32 +768,41 @@ impl Ui {
}); });
} }
/// Fire both searches for `query` in parallel. An empty query falls back to /// Search Drive and local metadata in one daemon round-trip. An empty query
/// the suggestions view. /// falls back to the suggestions view.
fn search(self: &Rc<Self>, query: &str) { fn search(self: &Rc<Self>, query: &str) {
if query.is_empty() { if query.is_empty() {
self.load_suggestions(); self.load_suggestions();
return; return;
} }
let id = self.begin_query(2); let id = self.begin_query(1);
let search = spawn_request(
let drive = spawn_request(
self.socket.clone(), self.socket.clone(),
Request::Search { Request::SearchV2 {
query: query.to_string(), query: query.to_string(),
limit: SEARCH_LIMIT, limit: SEARCH_LIMIT,
filters: SearchFilters {
sources: vec![SearchSource::Drive, SearchSource::Local],
kind: self.filter.get().search_kind(),
},
}, },
); );
let ui = self.clone(); let ui = self.clone();
glib::spawn_future_local(async move { glib::spawn_future_local(async move {
let reply = drive.recv().await; let reply = search.recv().await;
if !ui.finish_query(id) { if !ui.finish_query(id) {
return; return;
} }
match reply { match reply {
Ok(Ok(Response::SearchResults { hits })) => { Ok(Ok(Response::SearchResultsV2 {
*ui.drive_hits.borrow_mut() = hits; drive_hits,
local_hits,
local_indexing,
})) => {
*ui.drive_hits.borrow_mut() = drive_hits;
*ui.local_hits.borrow_mut() = local_hits;
ui.indexing.set(local_indexing);
ui.render(); ui.render();
} }
// Transport failure: the daemon is gone. Don't paint "No // Transport failure: the daemon is gone. Don't paint "No
@@ -683,32 +810,6 @@ impl Ui {
Ok(Err(_)) | Err(_) => ui.go_offline(), Ok(Err(_)) | Err(_) => ui.go_offline(),
_ => { _ => {
ui.drive_hits.borrow_mut().clear(); ui.drive_hits.borrow_mut().clear();
ui.render();
}
}
});
let local = spawn_request(
self.socket.clone(),
Request::SearchLocal {
query: query.to_string(),
limit: SEARCH_LIMIT,
},
);
let ui = self.clone();
glib::spawn_future_local(async move {
let reply = local.recv().await;
if !ui.finish_query(id) {
return;
}
match reply {
Ok(Ok(Response::LocalResults { hits, indexing })) => {
ui.indexing.set(indexing);
*ui.local_hits.borrow_mut() = hits;
ui.render();
}
Ok(Err(_)) | Err(_) => ui.go_offline(),
_ => {
ui.local_hits.borrow_mut().clear(); ui.local_hits.borrow_mut().clear();
ui.render(); ui.render();
} }
@@ -751,7 +852,12 @@ impl Ui {
for (chip_filter, chip) in self.chips.borrow().iter() { for (chip_filter, chip) in self.chips.borrow().iter() {
chip.set_active(*chip_filter == filter); chip.set_active(*chip_filter == filter);
} }
self.render(); let query = self.entry.text().trim().to_string();
if query.is_empty() {
self.render();
} else {
self.search(&query);
}
} }
fn cycle_filter(self: &Rc<Self>, backwards: bool) { fn cycle_filter(self: &Rc<Self>, backwards: bool) {
@@ -776,9 +882,8 @@ impl Ui {
.count() .count()
} }
/// Rebuild both sections from the last replies under the active chip, then /// Rebuild suggestions or the globally-ranked search section, preserving the
/// restore the cursor to the top. Every state change funnels through here, so /// selected identity when a result survives the update.
/// the list, the placeholder, and the footer can never disagree.
fn render(self: &Rc<Self>) { fn render(self: &Rc<Self>) {
let filter = self.filter.get(); let filter = self.filter.get();
@@ -808,17 +913,24 @@ impl Ui {
let query = self.entry.text().trim().to_string(); let query = self.entry.text().trim().to_string();
let searching = !query.is_empty(); let searching = !query.is_empty();
let drive_title = if searching {
"Proton Drive"
} else {
"Pinned in Proton Drive"
};
self.drive_section.set_rows(&drive, Some(drive_title));
self.local_section.set_rows(&local, None);
let total = drive.len() + local.len(); let total = drive.len() + local.len();
let mut visible = drive; let mut visible = drive;
visible.extend(local); visible.extend(local);
if searching {
rank_hits(&mut visible);
self.search_section.set_rows(&visible, Some("Best matches"));
self.drive_section.set_rows(&[], None);
self.local_section.set_rows(&[], None);
} else {
self.search_section.set_rows(&[], None);
let drive_count = visible
.iter()
.take_while(|hit| matches!(hit, Hit::Drive(_)))
.count();
self.drive_section
.set_rows(&visible[..drive_count], Some("Pinned in Proton Drive"));
self.local_section.set_rows(&visible[drive_count..], None);
}
// Find where the previously-selected row landed in the rebuilt list. // Find where the previously-selected row landed in the rebuilt list.
let restored = selected_key.and_then(|key| visible.iter().position(|hit| hit.key() == key)); let restored = selected_key.and_then(|key| visible.iter().position(|hit| hit.key() == key));
@@ -854,8 +966,8 @@ impl Ui {
// so it lands on the final list, not a half-populated one. // so it lands on the final list, not a half-populated one.
if self.open_pending.get() && self.pending.get() == 0 { if self.open_pending.get() && self.pending.get() == 0 {
self.open_pending.set(false); self.open_pending.set(false);
if let (Some(index), Some(window)) = (self.cursor.get(), self.window.borrow().clone()) { if let Some(index) = self.cursor.get() {
self.open(index, &window); self.open(index);
} }
} }
} }
@@ -892,6 +1004,19 @@ impl Ui {
/// section owns it, clear the other section's selection, and scroll it into /// section owns it, clear the other section's selection, and scroll it into
/// view. Focus stays in the entry so typing never breaks. /// view. Focus stays in the entry so typing never breaks.
fn select(self: &Rc<Self>, index: usize) { fn select(self: &Rc<Self>, index: usize) {
if !self.rendered_query.borrow().is_empty() {
self.drive_section.list.unselect_all();
self.local_section.list.unselect_all();
let Some(row) = self.search_section.list.row_at_index(index as i32) else {
return;
};
self.search_section.list.select_row(Some(&row));
self.cursor.set(Some(index));
self.scroll_into_view(&row);
return;
}
self.search_section.list.unselect_all();
let drive_count = self.drive_count(); let drive_count = self.drive_count();
let (list, other, local_index) = if index < drive_count { let (list, other, local_index) = if index < drive_count {
( (
@@ -938,7 +1063,7 @@ impl Ui {
/// straight to the desktop; a Drive hit must first be hydrated into the cache /// straight to the desktop; a Drive hit must first be hydrated into the cache
/// by the daemon (`OpenFile`), which can take a moment — the window stays up, /// by the daemon (`OpenFile`), which can take a moment — the window stays up,
/// with the spinner running, until the path comes back. /// with the spinner running, until the path comes back.
fn open(self: &Rc<Self>, index: usize, window: &adw::ApplicationWindow) { fn open(self: &Rc<Self>, index: usize) {
if self.opening.get() { if self.opening.get() {
return; return;
} }
@@ -957,15 +1082,36 @@ impl Ui {
match hit { match hit {
Hit::Local(local) => { Hit::Local(local) => {
xdg_open(Path::new(&local.path)); xdg_open(Path::new(&local.path));
window.close(); self.dismiss();
}
// A Drive folder exists in the mount — open it in the file manager
// rather than downloading anything.
Hit::Drive(drive) if drive.is_dir => {
xdg_open(&self.mountpoint.borrow().join(&drive.path));
window.close();
} }
Hit::Drive(drive) => { Hit::Drive(drive) => {
// Empty-query rows come from ListPins, whose `recursive` flag is
// pin policy rather than reliable node kind metadata. Every pin
// already has a mounted path, so activate that path directly;
// this handles non-recursively pinned folders without mistaking
// them for files and sending an invalid OpenFile request.
if self.rendered_query.borrow().is_empty() {
let path = drive.mounted_path.as_deref().map_or_else(
|| mounted_path(&self.mountpoint.borrow(), &drive.path),
PathBuf::from,
);
xdg_open(&path);
self.dismiss();
return;
}
match drive_activation(&drive.name, drive.is_dir) {
DriveActivation::Folder | DriveActivation::MountedMedia => {
let path = drive.mounted_path.as_deref().map_or_else(
|| mounted_path(&self.mountpoint.borrow(), &drive.path),
PathBuf::from,
);
xdg_open(&path);
self.dismiss();
return;
}
DriveActivation::Materialize => {}
}
self.opening.set(true); self.opening.set(true);
self.spinner.set_visible(true); self.spinner.set_visible(true);
self.spinner.start(); self.spinner.start();
@@ -978,7 +1124,6 @@ impl Ui {
}, },
); );
let ui = self.clone(); let ui = self.clone();
let window = window.clone();
glib::spawn_future_local(async move { glib::spawn_future_local(async move {
let reply = rx.recv().await; let reply = rx.recv().await;
ui.spinner.stop(); ui.spinner.stop();
@@ -987,7 +1132,7 @@ impl Ui {
match reply { match reply {
Ok(Ok(Response::FilePath { path })) => { Ok(Ok(Response::FilePath { path })) => {
xdg_open(Path::new(&path)); xdg_open(Path::new(&path));
window.close(); ui.dismiss();
} }
Ok(Ok(Response::Error { message, .. })) => { Ok(Ok(Response::Error { message, .. })) => {
ui.hint.set_label(&format!("Could not open: {message}")); ui.hint.set_label(&format!("Could not open: {message}"));
@@ -1288,3 +1433,37 @@ fn load_css() {
); );
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ranked_hits_merge_sources_on_one_score_scale() {
let mut hits = vec![
Hit::Drive(SearchHit {
name: "alpha.pdf".into(),
path: "Drive/alpha.pdf".into(),
is_dir: false,
size: 1,
modified: 1,
pinned: false,
uid: "v~a".into(),
mounted_path: None,
score: 100,
}),
Hit::Local(LocalHit {
name: "better.pdf".into(),
path: "/home/me/better.pdf".into(),
is_dir: false,
size: 1,
modified: 1,
score: 300,
}),
];
rank_hits(&mut hits);
assert!(matches!(&hits[0], Hit::Local(hit) if hit.name == "better.pdf"));
assert!(matches!(&hits[1], Hit::Drive(hit) if hit.name == "alpha.pdf"));
}
}
+493 -12
View File
@@ -1206,7 +1206,8 @@ retracted.
## B19 — Remote Folder Trashing on Mode Switch Failure (CRIT-02) ## B19 — Remote Folder Trashing on Mode Switch Failure (CRIT-02)
**Status:** Fixed in code (unverified on a live mount, 2026-07-22) **Status:** Fixed; all four managed-live mode pairs verified 2026-07-22,
fault-injected mount failure still pending
**Found:** 2026-07-21, multi-agent sync engine audit (`audit_bugs.md` CRIT-02) **Found:** 2026-07-21, multi-agent sync engine audit (`audit_bugs.md` CRIT-02)
**Where:** `crates/pdfs-fuse/src/devices.rs`, `apply_sync_folder_mode` **Where:** `crates/pdfs-fuse/src/devices.rs`, `apply_sync_folder_mode`
@@ -1220,8 +1221,9 @@ user deletion, evict the underlying local mirror while holding the sync-folder
lock, and only then mount FUSE. A mount failure leaves an inert `ondemand` row; lock, and only then mount FUSE. A mount failure leaves an inert `ondemand` row;
switching back to `mirror` clears the baseline and restores the local copy. switching back to `mirror` clears the baseline and restores the local copy.
**Verified:** `cargo clippy -p pdfs-fuse --all-targets -- -D warnings` and the **Verified:** `cargo clippy -p pdfs-fuse --all-targets -- -D warnings`, the
`pdfs-fuse` unit suite pass. A fault-injected/live mode-switch test is still needed. `pdfs-fuse` unit suite, and the complete managed-live matrix passed against the
1.0.0 release binary. A fault-injected mount-failure test is still needed.
--- ---
@@ -1587,16 +1589,23 @@ counts to baseline and observe no post-unmount DB work.
--- ---
## B45 — Truncate After a Queued Rewrite Replaces the Prefix With Zeros ## B45 — Truncate After a Queued Rewrite Corrupts or Returns EIO
**Status:** Fixed and live-verified (2026-07-22) **Status:** Fixed and live-verified (2026-07-22)
**Found by:** `scripts/fuse-acceptance.sh` **Found by:** `scripts/fuse-acceptance.sh`
**Where:** `crates/pdfs-fuse/src/filesystem.rs`, write-open scratch setup **Where:** `crates/pdfs-fuse/src/filesystem.rs`, write-open scratch setup;
`crates/pdfs-fuse/src/lib.rs`, `Core::queue_truncate`
**Repro:** Write a file, immediately replace its contents, then run **Repro:** Write a file, immediately replace its contents, then run
`truncate -s 4 file` before the queued replacement drains. Reading it returned `truncate -s 4 file` before the queued replacement drains. Reading it returned
four NUL bytes instead of the first four replacement bytes. four NUL bytes instead of the first four replacement bytes.
The expanded managed suite found the path-based variant: create and close a
10-byte file, then immediately call `truncate(path, 4)`. The close's complete
revision was still inside the two-second drain debounce. `queue_truncate`
described the shrink as an incomplete edit over the remote base, noticed the
pending edit, preserved an orphan staging file, and returned `EIO`.
**Cause:** A new write handle always treated the server revision as its base. **Cause:** A new write handle always treated the server revision as its base.
When a newer complete revision was still queued locally, the handle's scratch When a newer complete revision was still queued locally, the handle's scratch
file began sparse and zero-filled. Shrinking it preserved those scratch zeros, file began sparse and zero-filled. Shrinking it preserved those scratch zeros,
@@ -1604,8 +1613,10 @@ not the prefix of the locally authoritative queued revision.
**Fix:** A write opened over a complete queued revision seeds its scratch file **Fix:** A write opened over a complete queued revision seeds its scratch file
from that staged blob and marks the copied range authored. Stacking a write over from that staged blob and marks the copied range authored. Stacking a write over
an incomplete queued revision is refused rather than guessed. Live acceptance an incomplete queued revision is refused rather than guessed. Path-based
verified the exact rewrite-then-truncate sequence returns `b"a mu"`. truncate now composes the same way: it copies a complete pending blob, applies
the new length, marks the result complete, and inherits the last real remote
baseline. Live acceptance verified shrink, zero-filled growth, and sparse I/O.
--- ---
@@ -1624,8 +1635,478 @@ date"), surfaced as `EIO`.
link-details read can briefly observe the pre-move state, so the new name is link-details read can briefly observe the pre-move state, so the new name is
signed against stale requirements. signed against stale requirements.
**Fix:** Retry only `InvalidRequirements` from the rename-after-move step with a **Fix history:** A bounded retry of `InvalidRequirements` reduced the window but
short bounded backoff. Stable API failures are not retried. If every retry did not close it; the managed matrix reproduced the same failure after all four
fails, adopt the already-completed move in inode state before returning `EIO`, retries. Combined cross-directory move+rename now enters the durable rename
so the local model never claims the node remains in its old parent. The full queue as a desired end state. Its drain re-fetches the node on every attempt,
live suite subsequently passed the replacement and combined move/rename cases. skips either half that already landed, renames before moving, and resolves name
collisions without losing the source. Simple rename-only and move-only calls
remain synchronous. The managed live suite subsequently passed the replacement
and combined move/rename cases.
---
## B47 — FUSE Accepts Path Components Longer Than NAME_MAX
**Status:** Fixed and managed-live verified (2026-07-22)
**Found:** 2026-07-22, managed FUSE acceptance suite
**Where:** `crates/pdfs-fuse/src/filesystem.rs`, name-taking callbacks
**Repro:** `open(<mount>/<256 ASCII bytes>, O_CREAT)` succeeded. A conventional
Linux filesystem must reject a pathname component longer than 255 bytes with
`ENAMETOOLONG`.
**Cause:** Every callback converted `OsStr` with `to_string_lossy` and passed it
straight to Drive. This imposed neither Linux's component limit nor Drive's
UTF-8 requirement, and could silently change non-UTF-8 names.
**Fix:** Route lookup, create, mkdir, unlink, rmdir, and both rename components
through one validator. It enforces the 255-byte limit, rejects reserved/empty
components, rejects invalid UTF-8 with `EILSEQ`, and preserves valid Unicode.
The managed live suite is the verification gate.
---
## T1 — Managed Acceptance Raced Daemon Startup
**Status:** Fixed (2026-07-22)
**Found:** repeated install/restart/acceptance loop
**Where:** `scripts/fuse-acceptance.py`, managed setup
**Repro:** Run `systemctl --user restart proton-drive.service` and immediately
start `--managed-live`. The offline reference completed before the daemon had
recreated its control socket, so the first `pdfs --json sync list` failed with
`ENOENT` and setup aborted.
**Fix:** Managed validation polls the supported `sync list` API until the daemon
is ready or `PDFS_ACCEPTANCE_SYNC_TIMEOUT` expires. This is tracked as a test
harness defect rather than an application bug.
---
## T2 — Acceptance Cleanup Raced Queued Namespace Replay
**Status:** Fixed (2026-07-22)
**Found:** managed on-demand/on-demand matrix after every functional test passed
**Where:** `scripts/fuse-acceptance.py`, per-contract cleanup
**Repro:** A combined move+rename was correctly accepted into the durable queue.
Cleanup started recursively deleting its test tree while the rename drain was
still landing. The moved entry appeared in `dir-b` after `rmtree` enumerated it
but before `rmdir(dir-b)`, producing `ENOTEMPTY`.
**Fix:** Before cleanup, poll `pdfs --json status` until both pending counters
are zero. Retry `rmtree` only for `ENOTEMPTY` inside the uniquely owned test
root, then wait for the queued deletions before changing modes. Other errors are
still immediate failures.
---
## T3 — Mode-Matrix Preservation Check Raced Mirror Restoration
**Status:** Fixed (2026-07-22)
**Found:** managed on-demand/on-demand → on-demand/mirror transition
**Where:** `scripts/fuse-acceptance.py`, managed mode switching
**Repro:** Switch a populated managed folder from `ondemand` to `mirror`, wait
until `sync list` reports mirror/idle, and immediately read a preservation file.
The file can still be absent even though the subsequent restore pass downloads
it correctly.
**Cause:** The application commits the new mode before scheduling reconciliation
and leaves the previous idle state and sync timestamp in the row. Those values
describe the completed on-demand state, not a completed mirror restoration.
**Fix:** The harness records every transition to mirror and explicitly requests
and waits for a newer completed sync pass before validating local bytes. This is
tracked as a harness synchronization defect; transient absence before that pass
is expected asynchronous behavior, not data loss.
---
## T4 — Successful Managed Cleanup Left Its Own Sentinel Behind
**Status:** Fixed (2026-07-22)
**Found:** second consecutive managed acceptance run
**Where:** `scripts/fuse-acceptance.py`, managed cleanup
**Repro:** Complete the mode matrix in mirror/mirror, then immediately start a
new managed run. Registration cleanup removed the sync-folder records and remote
test folders, but the next run's empty-directory precheck found the harness's
128 KiB preservation sentinel in each local directory.
**Cause:** Removing a mirror registration intentionally preserves the local
copy. The harness treated unregistering as if it also emptied local storage.
**Fix:** After successful unregister, cleanup deletes only the sentinel whose
contents match the per-run bytes generated by this harness. Unknown files remain
untouched and continue to fail the strict precheck.
---
## B48 — Successful Unlink Can Reappear From a Stale Remote Listing
**Status:** Fixed (unverified)
**Found:** 2026-07-22, managed acceptance cleanup after all on-demand I/O passed
**Where:** `crates/pdfs-fuse/src/lib.rs`, child enumeration and trash paths
**Repro:** Unlink every child of a directory and immediately remove the
directory. `unlink` returned success, but a parent invalidation followed by an
eventually consistent remote enumeration returned the just-trashed child again.
`rmdir` then returned `ENOTEMPTY`; repeated recursive removal could reproduce it
for the entire acceptance timeout.
**Cause:** Three stale-data paths combined. A queued cross-directory rename
updated `State` but did not invalidate the kernel's cached empty destination
listing, so recursive walkers never saw or unlinked the moved child. Removing a node recorded no
authoritative local tombstone, so an eventually consistent Drive listing could
intern it again. Separately, `State::has_children` treated a resident empty
listing as inconclusive and fell back to obsolete SQLite child rows. Thus
`readdir` could report empty while `rmdir` returned `ENOTEMPTY` forever.
**Fix:** Every successful online or queued trash records its uid in a
session-shared hidden set. Both persisted and remote child enumeration filter
that set. Secondary on-demand mounts share it, and an explicit restore removes
the uid so restored content can appear normally. `has_children` now treats any
resident listing—including an empty one—as authoritative and consults SQLite
only when the listing is absent. After replying successfully to a queued rename,
the FUSE handler invalidates both exact dentries and both directory inodes; the
ordering matters because an invalidation sent before the rename reply is
overwritten by the kernel's response processing. Uids are immutable and never
reused, making a session-long tombstone safe.
---
## B49 — Failed Conflict Preservation Still Overwrites the Local File (CRIT-05)
**Status:** Fixed in code with deterministic regressions; live/fault verification pending
**Found:** 2026-07-22, 1.0 data-safety audit
**Where:** `crates/pdfs-fuse/src/sync.rs`, download/conflict handling
**Cause:** If renaming a concurrently changed local file to a conflict copy
fails, sync only logs the error and continues publishing the remote download
over the local version. Second-resolution conflict names can also collide.
**Required fix/test:** Abort download publication unless preservation succeeds;
use collision-resistant names. Test existing names, permissions, `ENOSPC`,
`EIO`, and cross-filesystem behavior. This is B39's failure path; B25 remains
the separate weak-baseline problem.
---
## B50 — Failed Orphan Staging Deletes the Sole Scratch Copy (CRIT-06)
**Status:** Fixed in code; injected filesystem-failure verification pending
**Found:** 2026-07-22, 1.0 data-safety audit
**Where:** `crates/pdfs-fuse/src/lib.rs`, `Core::stage_orphaned_write`
**Cause:** If preserving an orphaned scratch write into staging fails, the
failure path removes the original even though no other durable copy exists.
**Required fix/test:** Never unlink the source until a durable replacement and
metadata are committed. Inject `ENOSPC`, `EIO`, `EROFS`, permission, and
cross-device failures and assert a byte-identical copy remains discoverable.
Related: B23, B27, and B28.
---
## B51 — Scratch Sidecar Does Not Establish a Durable Recovery Record (CRIT-07)
**Status:** Fixed in code; power-loss verification pending
**Found:** 2026-07-22, 1.0 crash-consistency audit
**Where:** `crates/pdfs-core/src/cache.rs`, scratch durability markers
**Cause:** A temporary sidecar is renamed without syncing the sidecar first or
the directory afterward. Power loss after successful user `fsync` can preserve
bytes but lose authored-range metadata; recovery may discard the write or treat
a sparse partial write as complete and replace untouched ranges with zeros.
**Required fix/test:** Sync data, write and sync temporary metadata, rename it,
then sync the directory before acknowledging durability. Crash-test every
boundary and corrupt/partial sidecars. B20 and B23 do not cover this invariant.
---
## B52 — Staged-Write Publication Is Not Crash-Safe (CRIT-08)
**Status:** Fixed in code; cross-filesystem and power-loss verification pending
**Found:** 2026-07-22, 1.0 crash-consistency audit
**Where:** `crates/pdfs-core/src/cache.rs`, staged-write publication
**Cause:** Rename/copy and direct sidecar writes lack a complete sync protocol.
The cross-filesystem fallback can remove its source after an unsynced copy,
leaving no complete data/metadata pair after a crash.
**Required fix/test:** Publish synced temporary data and metadata via atomic
renames and directory syncs; delete the source only after verification. Inject
failure after every write, sync, rename, DB enqueue, and deletion. Related: B50
and B51.
---
## B53 — Superseding a Pending Operation Is Not Transactional (CRIT-09)
**Status:** Fixed with rollback regression; crash/ENOSPC verification pending
**Found:** 2026-07-22, 1.0 SQLite durability audit
**Where:** `crates/pdfs-core/src/db/ops.rs`, `Db::enqueue_op`
**Cause:** The old durable row is deleted before its replacement is inserted,
outside a transaction. A crash, constraint error, full disk, or I/O failure can
erase acknowledged upload work and orphan its blob.
**Required fix/test:** Select and replace in one transaction; retain the old
blob until commit and make cleanup recoverable. Inject insert/commit failures
and prove that old or new operation—never neither—remains queued. B27 concerns
a retained failing row; this issue erases the row.
---
## B54 — Total-Wipe Guard Does Not Protect a One-Entry Baseline (CRIT-10)
**Status:** Fixed with one-entry regression; live mode-matrix verification pending
**Found:** 2026-07-22, 1.0 mirror-sync audit
**Where:** `crates/pdfs-fuse/src/sync/planner.rs`
**Cause:** The safeguard activates only when the baseline has at least two
paths. A one-file local folder that is missing, unreadable, or wrongly mounted
can therefore trash its sole remote file.
**Required fix/test:** Protect every non-empty baseline and require an
independently complete scan plus an explicit deliberate-delete signal for a
total wipe. Test absent roots, replaced mountpoints, permissions, and one-file
folders in every mode combination.
---
## B55 — Incomplete Local Scans Are Interpreted as Deletions (CRIT-11)
**Status:** Fixed in code; injected iterator/stat failure tests pending
**Found:** 2026-07-22, 1.0 mirror-sync audit
**Where:** `crates/pdfs-fuse/src/sync.rs`, local scan/reconciliation
**Cause:** Directory-entry and metadata errors omit paths instead of marking the
snapshot incomplete. Reconciliation can treat omissions as intentional local
deletions and propagate them remotely.
**Required fix/test:** Carry completeness/errors into planning; an incomplete
subtree must make the pass non-destructive and retain its prior baseline. Test
`readdir`, `stat`, permission, transient-I/O, and disappearing-entry failures.
B42 covers the inverse local-delete failure.
---
## B56 — Database Schema Version Parsing Fails Open (HIGH-13)
**Status:** Partly fixed — future schemas are refused; malformed-version handling remains open
**Found:** 2026-07-22, 1.0 database audit
**Where:** `crates/pdfs-core/src/db/migrations.rs`, `Db::migrate`
**Cause:** Missing, malformed, or non-numeric version values become zero, while
all future versions are accepted. Corruption can replay migrations, and an
older binary can open a newer incompatible database.
**Required fix/test:** Distinguish a new DB from corrupt metadata; reject
malformed/future versions without mutation and validate required schema. Test
empty, malformed, negative, future, partial/interrupted migration, and upgrade
fixtures from every released schema.
---
## B57 — Event Cursor Can Advance Past Unapplied State (HIGH-14)
**Status:** Partly fixed — cursor now advances per successfully invalidated event; DB/apply fault tests remain
**Found:** 2026-07-22, 1.0 event-recovery audit
**Where:** `crates/pdfs-fuse/src/background.rs`, `run_event_sync`
**Cause:** Cache invalidation failures are ignored while the batch cursor still
advances. Cursor-persistence failure also leaves the in-memory cursor moving;
first-seed persistence failure can later reseed at a newer head. Applying state
and committing its cursor have no atomic relationship.
**Required fix/test:** Apply idempotently and persist derived DB state plus the
cursor atomically, or advance only through the last durable event. Retry failed
prerequisites. Crash/fault-test every event and cursor boundary, first seed,
duplicate replay, malformed events, and database-full conditions.
---
## B58 — Control Socket Has Unbounded Frames and Connections (HIGH-15)
**Status:** Fixed in code with frame-limit regressions; connection-flood verification pending
**Found:** 2026-07-22, 1.0 IPC audit
**Where:** `crates/pdfs-fuse/src/control.rs`
**Cause:** The server uses unbounded `read_line`, a new thread per connection,
and no concurrency or idle limit. A local process can exhaust memory, file
descriptors, and threads, blocking filesystem control and safe shutdown.
**Required fix/test:** Bound frames, clients, and idle lifetime. Test slow/no-
newline clients, oversized/truncated JSON, floods, and recovery. B6/B35 cover
socket authority and permissions, not resource bounds.
---
## B59 — Private State Permission Enforcement Fails Open (HIGH-16)
**Status:** Fixed in code with symlink regression; wrong-owner integration verification pending
**Found:** 2026-07-22, 1.0 local-privacy audit
**Where:** `crates/pdfs-core/src/config.rs`, `AppDirs::ensure`
**Cause:** chmod/creation failures are warned about or ignored while decrypted
content, plaintext metadata, and the authenticated socket may remain accessible
to other local users.
**Required fix/test:** Before loading credentials or serving data, verify every
sensitive directory is owned by the effective user, is not a symlink, and is
owner-only. Test wrong owner, symlink substitution, read-only filesystems,
permissive restored directories, and chmod failure. This tightens B6/B35.
---
## B60 — Config Writes Are Non-Atomic and Parse Errors Are Overwritten (HIGH-17)
**Status:** Fixed in code; fault/concurrent-save verification pending
**Found:** 2026-07-22, 1.0 configuration audit
**Where:** `crates/pdfs-core/src/config.rs`, `load_config` / `save_config`
**Cause:** Save truncates in place without sync. Load treats read/parse failure
as absence and best-effort overwrites with defaults, silently losing mountpoint,
ignore policy, budget, and explicit device adoption.
**Required fix/test:** Atomically publish a restricted, synced temporary file
and sync its directory. Preserve malformed input and report an actionable error.
Test invalid/truncated JSON, permissions, `ENOSPC`, crash points, and concurrent
saves.
---
## B61 — Rotated Single-Use Credentials Can Be Lost (HIGH-18)
**Status:** Open — 1.0 authentication/recovery blocker
**Found:** 2026-07-22, 1.0 authentication audit
**Where:** `crates/pdfs-core/src/auth.rs`, refresh callback
**Cause:** Failure to serialize or persist rotated credentials is only logged
(and some setup failures are silent). The daemon continues in memory while the
keyring retains an invalid single-use refresh token, so restart loses login.
**Required fix/test:** Expose unhealthy persistence, retry with bounded backoff,
notify the user, and define a safe re-login/shutdown path. Test locked/full/
unavailable keyrings, callback races, repeated rotations, death, and restart.
---
## B62 — Debian Artifact Omits Required Service and Autostart Units (HIGH-19)
**Status:** Fixed in workflow; clean-install VM verification pending
**Found:** 2026-07-22, 1.0 packaging audit
**Where:** `.github/workflows/release.yml`; `packaging/`
**Cause:** The generated package omits `proton-drive.service` and the tray
autostart entry even though normal login/GUI flows assume that lifecycle.
**Required fix/test:** Package all units/resources with correct paths, modes,
and hooks. In clean supported-distribution VMs test install, login, reboot,
daemon/tray/FUSE, upgrade, rollback, and uninstall without deleting user state.
---
## B63 — Release Version Identity Is Inconsistent and Unenforced (HIGH-20)
**Status:** Partly fixed — tag/workspace/PKGBUILD equality is enforced; protocol identity remains
**Found:** 2026-07-22, 1.0 release audit
**Where:** manifests, client identity constants, tags, release workflow
**Cause:** Workspace/binary versions, protocol identity strings, and tags can
disagree; the workflow trusts any `v*` tag without checking package metadata or
installed `--version` output.
**Required fix/test:** Establish one version source and fail CI on every tag,
manifest, package, and binary mismatch. Document any intentionally independent
protocol identity.
---
## B64 — Stable Publishing Lacks Data-Safety and Recovery Gates (CRIT-12)
**Status:** Partly fixed — automated gates and managed-live matrix pass; recovery approval remains open
**Found:** 2026-07-22, 1.0 release audit
**Where:** CI and `.github/workflows/release.yml`
**Cause:** A tag can publish without required format/lint/tests, offline and
dedicated-account live acceptance, migrations, or crash/recovery drills. Fixes
still marked unverified can ship as stable.
**Required fix/test:** Publish immutable RC artifacts only after all workspace
checks, full managed mode matrix, migration/power-loss recovery, clean install/
upgrade tests, and explicit sign-off for every critical/high integrity issue.
---
## B65 — Release Artifacts Lack Supply-Chain Verification (MED-11)
**Status:** Open — 1.0 release/security blocker
**Found:** 2026-07-22, 1.0 supply-chain audit
**Where:** CI and release workflows
**Cause:** Releases do not enforce advisory/license policy or publish signed
provenance, SBOMs, and verifiable checksums for exact distributed artifacts.
**Required fix/test:** Add locked audit/license gates, SBOM and provenance,
checksums and signatures, documented verification, and an independent clean-job
verification/install test.
---
## B66 — Local-Only Pending Data Is Not Protected at Shutdown (HIGH-21)
**Status:** Open — 1.0 data-safety/UX blocker
**Found:** 2026-07-22, 1.0 recovery audit
**Where:** shutdown, CLI/GUI status, logout/unregister workflows
**Cause:** Undrained writes may exist only on one machine, but there is no flush
command with a completion contract, prominent pending byte/count health, or
shutdown/logout warning/inhibition.
**Required fix/test:** Add `pdfs sync flush`, expose pending operations/bytes
and last errors, and protect destructive lifecycle actions. Test offline stop,
logout, unregister, reboot, retry exhaustion, and drain. Credential removal must
never delete staged bytes.
---
## B67 — Fresh-State Restore Can Silently Select a New Device (HIGH-22)
**Status:** Open — 1.0 recovery blocker
**Found:** 2026-07-22, 1.0 disaster-recovery audit
**Where:** device discovery/adoption; `docs/RECOVERY.md`
**Cause:** Restore depends on hostname and folder basename. After local-state
loss, a mismatch can silently create/select another device rather than attach
the intended remote roots, making recovery appear complete while data is absent.
**Required fix/test:** Require explicit adoption/confirmation on ambiguity and
show old/proposed device and root IDs. Test state loss, hostname change,
duplicates, renamed folders, multiple devices, and byte-identical restoration.
---
## B68 — Profile Backup Cannot Be Written to the Device Root
**Status:** Open — live-reproduced 2026-07-22
**Found:** complete managed-live mode matrix against the 1.0.0 release binary
**Where:** `crates/pdfs-fuse/src/profile.rs`, profile backup destination
**Repro:** Every sync-folder registration or mode transition attempted the
profile backup and received `NotEnoughPermissions` (HTTP 422): `Cannot create
file at the root of a device`. The filesystem matrix itself passed, but profile
backup never became durable remotely.
**Impact:** The recovery feature appears active but does not preserve the device
profile, so a replacement machine still lacks modes, pins, mount settings, and
the explicit remote-folder mapping. Repeated retries also add noisy warnings.
**Required fix/test:** Store the profile below a writable application folder
rather than directly at the device root, reuse that folder by stable identity,
and make backup health visible. Verify upload, replacement, restart restore, and
byte-identical fresh-state recovery on a dedicated account.
+43 -9
View File
@@ -9,9 +9,21 @@ cargo test --workspace
cargo clippy --workspace --all-targets -- -D warnings cargo clippy --workspace --all-targets -- -D warnings
``` ```
## Live FUSE acceptance suite ## Filesystem API and live FUSE acceptance suite
Kernel behavior and remote convergence need a real mount. The live suite runs The account-free suite exercises the filesystem syscall contract against a
temporary local directory. This validates the runner and covers creation/open
flags, positioned and vectored I/O, truncation and sparse files, `mmap`,
`sendfile`, durability barriers, namespace/error semantics, open-file lifetime,
names and enumeration, and concurrent I/O. It never reads application state or
contacts Proton:
```console
scripts/fuse-acceptance.sh --offline-only
```
Kernel/FUSE behavior and remote convergence need a real mount. Live testing is
explicitly opt-in and always runs the account-free contract first. It runs
destructive POSIX operations only below a fresh `pdfs-acceptance-*` directory. destructive POSIX operations only below a fresh `pdfs-acceptance-*` directory.
It refuses paths that are not FUSE mounts and removes its directory on success, It refuses paths that are not FUSE mounts and removes its directory on success,
failure, or interruption. failure, or interruption.
@@ -19,16 +31,39 @@ failure, or interruption.
Use a dedicated test account without irreplaceable data: Use a dedicated test account without irreplaceable data:
```console ```console
scripts/fuse-acceptance.sh /mnt/on-demand-testmount scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount
``` ```
### Automated mode matrix
The managed live runner accepts two existing, empty, unmounted directories. It
registers both as new sync folders, waits for initial synchronization, then
tests all four pairs in order: on-demand/on-demand, on-demand/mirror,
mirror/on-demand, and mirror/mirror. Every transition must become idle, and an
on-demand folder must actually appear as a mount before its contract runs.
```console
mkdir -p /mnt/pdfs-test-a /mnt/pdfs-test-b
scripts/fuse-acceptance.sh --managed-live /mnt/pdfs-test-a /mnt/pdfs-test-b
```
This creates remote folders and permanently deletes those test remotes during
cleanup. It refuses non-empty paths, existing mounts, and paths already present
in `pdfs sync list`. Use `PDFS_ACCEPTANCE_SYNC_TIMEOUT` to extend transition
timeouts and `PDFS_ACCEPTANCE_PDFS` to select a non-installed CLI binary, for
example `target/debug/pdfs`.
During development, `PDFS_ACCEPTANCE_ONLY` runs tests whose descriptive name
contains the supplied text (for example `PDFS_ACCEPTANCE_ONLY=namespace`). A
release acceptance run must leave it unset so the complete contract executes.
Passing more paths runs the same filesystem suite independently against each. Passing more paths runs the same filesystem suite independently against each.
The first path must be the FUSE mount under test; later paths may be another The first path must be the FUSE mount under test; later paths may be another
FUSE mount or a normal local mirror filesystem: FUSE mount or a normal local mirror filesystem:
```console ```console
PDFS_ACCEPTANCE_SYNC_TIMEOUT=180 \ PDFS_ACCEPTANCE_SYNC_TIMEOUT=180 \
scripts/fuse-acceptance.sh /mnt/on-demand-testmount /mnt/testmount scripts/fuse-acceptance.sh --live /mnt/on-demand-testmount /mnt/testmount
``` ```
If two paths really are views of the **same remote folder**, enable an If two paths really are views of the **same remote folder**, enable an
@@ -36,16 +71,15 @@ additional byte-for-byte convergence check explicitly:
```console ```console
PDFS_ACCEPTANCE_CONVERGENCE=1 \ PDFS_ACCEPTANCE_CONVERGENCE=1 \
scripts/fuse-acceptance.sh /mnt/first-view /mnt/second-view scripts/fuse-acceptance.sh --live /mnt/first-view /mnt/second-view
``` ```
Do not enable that flag merely because two sync folders belong to the same Do not enable that flag merely because two sync folders belong to the same
account. Their configured remote UIDs must be identical. account. Their configured remote UIDs must be identical.
The suite covers create/read/rewrite/truncate, multi-megabyte binary content, Normally the live suite applies the same API contract independently to each
unaligned random writes, rename replacement, cross-directory moves, important supplied mount. In convergence mode it exercises the primary once and verifies
POSIX error cases, open-then-unlink behavior, concurrent writers, directory that the resulting bytes appear through every secondary view. A pass is
enumeration, and optional cross-mount byte-for-byte convergence. A pass is
strong evidence for those paths, not a blanket data-loss guarantee. Before a strong evidence for those paths, not a blanket data-loss guarantee. Before a
release, also run the recovery drills in `docs/RECOVERY.md` and inspect release, also run the recovery drills in `docs/RECOVERY.md` and inspect
`docs/BUGS.md` for live-verification items. `docs/BUGS.md` for live-verification items.
+1 -1
View File
@@ -9,7 +9,7 @@
# Build + install: cd packaging && makepkg -fi # Build + install: cd packaging && makepkg -fi
pkgname=proton-drive-linux pkgname=proton-drive-linux
pkgver=0.7.1 pkgver=1.0.0
pkgrel=1 pkgrel=1
pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
arch=('x86_64') arch=('x86_64')
+572
View File
@@ -0,0 +1,572 @@
#!/usr/bin/env python3
"""Account-free filesystem API contract plus opt-in live FUSE acceptance."""
from __future__ import annotations
import argparse
import concurrent.futures
import errno
import hashlib
import mmap
import os
from pathlib import Path
import shutil
import stat
import subprocess
import tempfile
import time
import uuid
def check(condition: bool, message: str) -> None:
if not condition:
raise AssertionError(message)
def expect_errno(expected: set[int], operation, description: str) -> None:
try:
operation()
except OSError as error:
check(error.errno in expected, f"{description}: errno {error.errno}, expected {expected}")
else:
raise AssertionError(f"{description}: unexpectedly succeeded")
def read(path: Path) -> bytes:
with path.open("rb", buffering=0) as file:
return file.read()
def write_durable(path: Path, data: bytes) -> None:
fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
try:
view = memoryview(data)
while view:
written = os.write(fd, view)
check(written > 0, "write made no progress")
view = view[written:]
os.fsync(fd)
finally:
os.close(fd)
def test_create_flags(root: Path) -> None:
path = root / "flags"
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)
os.write(fd, b"abcdef")
os.close(fd)
expect_errno({errno.EEXIST}, lambda: os.open(path, os.O_CREAT | os.O_EXCL), "O_EXCL")
fd = os.open(path, os.O_WRONLY | os.O_APPEND)
os.lseek(fd, 0, os.SEEK_SET)
os.write(fd, b"++")
os.close(fd)
check(read(path) == b"abcdef++", "O_APPEND ignored")
fd = os.open(path, os.O_WRONLY | os.O_TRUNC)
os.close(fd)
check(path.stat().st_size == 0, "O_TRUNC did not truncate")
def test_positioned_and_vectored_io(root: Path) -> None:
path = root / "positioned.bin"
base = bytes(range(256)) * 32768
write_durable(path, base)
fd = os.open(path, os.O_RDWR)
try:
check(os.pread(fd, 19, 4093) == base[4093:4112], "pread returned wrong range")
check(os.pwrite(fd, b"boundary-write", 4093) == 14, "short pwrite")
if hasattr(os, "pwritev"):
check(os.pwritev(fd, [b"vector-", b"write"], 65531) == 12, "short pwritev")
if hasattr(os, "preadv"):
chunks = [bytearray(7), bytearray(5)]
check(os.preadv(fd, chunks, 65531) == 12, "short preadv")
check(b"".join(chunks) == b"vector-write", "preadv data mismatch")
os.fdatasync(fd)
finally:
os.close(fd)
expected = bytearray(base)
expected[4093:4107] = b"boundary-write"
expected[65531:65543] = b"vector-write"
got = read(path)
check(got == expected, "positioned write damaged surrounding bytes")
print(f" sha256 {hashlib.sha256(got).hexdigest()}")
def test_resize_and_sparse_io(root: Path) -> None:
path = root / "resize"
write_durable(path, b"0123456789")
os.truncate(path, 4)
check(read(path) == b"0123", "shrink did not preserve prefix")
os.truncate(path, 8193)
data = read(path)
check(data[:4] == b"0123" and data[4:] == bytes(8189), "grown range is not zero-filled")
sparse = root / "sparse"
fd = os.open(sparse, os.O_CREAT | os.O_RDWR, 0o600)
try:
os.pwrite(fd, b"head", 0)
os.pwrite(fd, b"tail", 8 * 1024 * 1024 + 17)
os.fsync(fd)
finally:
os.close(fd)
check(sparse.stat().st_size == 8 * 1024 * 1024 + 21, "sparse size mismatch")
fd = os.open(sparse, os.O_RDONLY)
try:
check(os.pread(fd, 8, 4) == bytes(8), "sparse hole is not zero-filled")
check(os.pread(fd, 4, 8 * 1024 * 1024 + 17) == b"tail", "sparse tail missing")
check(os.pread(fd, 1, sparse.stat().st_size + 4096) == b"", "read beyond EOF not empty")
finally:
os.close(fd)
def test_mmap_and_copy_paths(root: Path) -> None:
source = root / "mapped"
data = bytearray((b"mmap-and-sendfile\0" * 65536)[:1024 * 1024])
write_durable(source, data)
fd = os.open(source, os.O_RDWR)
try:
with mmap.mmap(fd, len(data), access=mmap.ACCESS_WRITE) as mapping:
mapping[4091:4107] = b"mapped-boundary!"
mapping.flush()
os.fsync(fd)
finally:
os.close(fd)
data[4091:4107] = b"mapped-boundary!"
check(read(source) == data, "shared mmap write/read mismatch")
target = root / "sendfile-copy"
src = os.open(source, os.O_RDONLY)
dst = os.open(target, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
try:
offset = 0
while offset < len(data):
count = os.sendfile(dst, src, offset, len(data) - offset)
check(count > 0, "sendfile made no progress")
offset += count
os.fsync(dst)
finally:
os.close(src)
os.close(dst)
check(read(target) == data, "sendfile copy mismatch")
def test_namespace_and_errors(root: Path) -> None:
a, b = root / "dir-a", root / "dir-b"
a.mkdir(); b.mkdir()
(a / "nested").mkdir()
write_durable(a / "nested" / "child", b"child")
write_durable(root / "source", b"source")
write_durable(root / "victim", b"victim")
os.replace(root / "source", root / "victim")
check(read(root / "victim") == b"source" and not (root / "source").exists(), "replace failed")
os.rename(root / "victim", b / "moved")
check(read(b / "moved") == b"source", "cross-directory move failed")
check("moved" in os.listdir(b), "cross-directory move missing from readdir")
os.rename(b / "moved", b / "moved")
check(read(b / "moved") == b"source", "same-name rename changed data")
expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.rmdir(a), "rmdir(non-empty)")
expect_errno({errno.ENOTDIR}, lambda: os.rmdir(b / "moved"), "rmdir(file)")
expect_errno({errno.EISDIR, errno.EPERM}, lambda: os.unlink(a), "unlink(directory)")
expect_errno({errno.EINVAL}, lambda: os.rename(a, a / "nested" / "cycle"), "directory cycle")
expect_errno({errno.ENOENT}, lambda: os.unlink(root / "absent"), "unlink(absent)")
expect_errno({errno.ENOENT}, lambda: os.rename(root / "absent", root / "new"), "rename(absent)")
empty_src, empty_dst = root / "empty-src", root / "empty-dst"
empty_src.mkdir(); empty_dst.mkdir()
os.replace(empty_src, empty_dst)
check(empty_dst.is_dir() and not empty_src.exists(), "empty directory replacement failed")
nonempty = root / "nonempty-dst"
nonempty.mkdir(); write_durable(nonempty / "keep", b"keep")
replacement = root / "replacement-dir"; replacement.mkdir()
expect_errno({errno.ENOTEMPTY, errno.EEXIST}, lambda: os.replace(replacement, nonempty), "replace non-empty dir")
check(read(nonempty / "keep") == b"keep", "failed replacement damaged destination")
def test_open_lifetime(root: Path) -> None:
path = root / "open-unlink"
data = b"held-open\0" * 4096
write_durable(path, data)
fd = os.open(path, os.O_RDWR)
os.unlink(path)
check(not path.exists(), "unlinked name remains visible")
check(os.pread(fd, len(data), 0) == data, "open file lost data after unlink")
os.pwrite(fd, b"still-open", 0)
check(os.pread(fd, 10, 0) == b"still-open", "unlinked open file is not writable")
os.close(fd)
old, new = root / "open-rename-old", root / "open-rename-new"
write_durable(old, b"before")
fd = os.open(old, os.O_RDWR)
os.rename(old, new)
os.pwrite(fd, b"after!", 0)
os.fsync(fd); os.close(fd)
check(not old.exists() and read(new) == b"after!", "open handle did not follow rename")
def test_names_and_enumeration(root: Path) -> None:
names = ["space name.txt", "unicodé-文件", ".hidden", "trailing.dot.", "case", "CASE"]
for index, name in enumerate(names):
write_durable(root / name, f"name-{index}".encode())
listed = set(os.listdir(root))
check(set(names) <= listed, "directory enumeration omitted valid names")
for name in listed:
os.lstat(root / name)
expect_errno({errno.ENAMETOOLONG}, lambda: os.open(root / ("x" * 256), os.O_CREAT), "overlong name")
def test_concurrency(root: Path) -> None:
def independent(index: int) -> None:
data = bytes([index]) * (131072 + index)
path = root / f"concurrent-{index}"
write_durable(path, data)
check(read(path) == data, f"concurrent file {index} mismatch")
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
list(pool.map(independent, range(1, 17)))
shared = root / "concurrent-ranges"
extent = 64 * 1024
write_durable(shared, bytes(extent * 8))
fd = os.open(shared, os.O_RDWR)
try:
def positioned(index: int) -> None:
payload = bytes([index + 1]) * extent
check(os.pwrite(fd, payload, index * extent) == extent, f"short concurrent pwrite {index}")
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
list(pool.map(positioned, range(8)))
os.fsync(fd)
finally:
os.close(fd)
expected = b"".join(bytes([i + 1]) * extent for i in range(8))
check(read(shared) == expected, "concurrent disjoint writes overlapped or vanished")
TESTS = [
("creation and open flags", test_create_flags),
("positioned and vectored I/O", test_positioned_and_vectored_io),
("truncate, growth, sparse ranges, and EOF", test_resize_and_sparse_io),
("mmap and zero-copy copy paths", test_mmap_and_copy_paths),
("namespace operations and POSIX errors", test_namespace_and_errors),
("open-handle lifetime across unlink and rename", test_open_lifetime),
("names, lookup, stat, and enumeration", test_names_and_enumeration),
("independent and shared-file concurrency", test_concurrency),
]
def run_contract(parent: Path, label: str) -> tuple[Path, str]:
root = parent / f"pdfs-acceptance-{uuid.uuid4().hex}"
root.mkdir()
print(f"[target] {label}: {parent}")
try:
selected = os.environ.get("PDFS_ACCEPTANCE_ONLY")
tests = TESTS if not selected else [test for test in TESTS if selected.lower() in test[0].lower()]
check(bool(tests), f"PDFS_ACCEPTANCE_ONLY={selected!r} matched no tests")
for name, operation in tests:
print(f" [test] {name}")
operation(root)
digest_path = root / "positioned.bin"
digest = hashlib.sha256(read(digest_path)).hexdigest() if digest_path.exists() else ""
return root, digest
except BaseException:
shutil.rmtree(root, ignore_errors=True)
raise
def wait_for_copy(root: Path, relative: Path, digest: str, timeout: int) -> None:
deadline = time.monotonic() + timeout
target = root / relative
while time.monotonic() < deadline:
try:
if target.is_file() and hashlib.sha256(read(target)).hexdigest() == digest:
return
except OSError:
pass
time.sleep(2)
raise AssertionError(f"{target} did not converge byte-for-byte within {timeout}s")
class ManagedSyncPair:
"""Two sync registrations created by this run and removed on exit."""
def __init__(self, paths: list[Path], timeout: int) -> None:
self.paths = [path.resolve() for path in paths]
self.timeout = timeout
self.pdfs = os.environ.get("PDFS_ACCEPTANCE_PDFS", "pdfs")
self.ids: dict[Path, int] = {}
self.sentinels = {
path: hashlib.sha256(f"pdfs-preservation:{path}:{uuid.uuid4()}".encode()).digest()
* 4096
for path in self.paths
}
def command(self, *args: str, json_output: bool = False) -> str:
command = [self.pdfs]
if json_output:
command.append("--json")
command.extend(args)
result = subprocess.run(command, text=True, capture_output=True)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip()
raise RuntimeError(f"{' '.join(command)} failed: {detail}")
return result.stdout
def folders(self) -> list[dict]:
import json
value = json.loads(self.command("sync", "list", json_output=True))
return value["items"]
def wait_for_queue(self) -> None:
import json
deadline = time.monotonic() + self.timeout
last = None
while time.monotonic() < deadline:
value = json.loads(self.command("status", json_output=True))
last = value.get("mount") or {}
if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0:
return
time.sleep(1)
raise TimeoutError(f"daemon mutation queue did not drain: {last}")
def remove_tree(self, root: Path) -> None:
deadline = time.monotonic() + self.timeout
while True:
try:
shutil.rmtree(root)
return
except FileNotFoundError:
return
except OSError as error:
if error.errno != errno.ENOTEMPTY or time.monotonic() >= deadline:
survivors = []
if root.exists():
for directory, dirs, files in os.walk(root):
survivors.extend(str(Path(directory) / name) for name in dirs + files)
raise OSError(
error.errno,
f"{error.strerror}; surviving test entries: {survivors[:50]}",
error.filename,
) from error
# A queued namespace operation can land between rmtree's
# enumeration and rmdir. Re-walk the test-owned tree.
time.sleep(0.25)
def validate(self) -> None:
if shutil.which(self.pdfs) is None and not Path(self.pdfs).is_file():
raise RuntimeError(
f"cannot find {self.pdfs!r}; set PDFS_ACCEPTANCE_PDFS to the CLI binary"
)
if self.paths[0] == self.paths[1]:
raise ValueError("managed live paths must be different directories")
deadline = time.monotonic() + self.timeout
last_error = None
while True:
try:
current = self.folders()
break
except Exception as error:
last_error = error
if time.monotonic() >= deadline:
raise TimeoutError(
f"pdfs daemon did not become ready within {self.timeout}s: {last_error}"
) from error
time.sleep(1)
registered = {Path(item["local_path"]).resolve() for item in current}
for path in self.paths:
check(path.is_dir(), f"managed path is not a directory: {path}")
check(not any(path.iterdir()), f"managed path is not empty: {path}")
check(path not in registered, f"managed path is already registered: {path}")
check(not is_mountpoint(path), f"managed path is already a mount: {path}")
def wait_for(self, path: Path, *, mode: str | None = None) -> dict:
deadline = time.monotonic() + self.timeout
last = None
while time.monotonic() < deadline:
for item in self.folders():
if Path(item["local_path"]).resolve() != path:
continue
last = item
if (
item["state"] == "idle"
and item.get("pending_mode") is None
and (mode is None or item["mode"] == mode)
):
return item
if item["state"] in {"error", "conflict"}:
raise RuntimeError(f"sync folder {path} entered {item['state']}: {item}")
time.sleep(2)
raise TimeoutError(f"sync folder {path} did not become idle in mode {mode}: {last}")
def create(self) -> None:
self.validate()
for path in self.paths:
print(f"[setup] registering empty sync folder {path}")
self.command("sync", "add", str(path))
item = self.wait_for(path, mode="mirror")
self.ids[path] = int(item["id"])
for path in self.paths:
write_durable(path / "pdfs-mode-preservation.bin", self.sentinels[path])
self.force_sync(path)
def force_sync(self, path: Path) -> None:
before = int(self.wait_for(path)["last_sync"])
# last_sync has one-second resolution. Ensure this requested pass cannot
# finish with the same timestamp and look indistinguishable from no pass.
while int(time.time()) <= before:
time.sleep(0.1)
self.command("sync", "now", str(self.ids[path]))
deadline = time.monotonic() + self.timeout
last = None
while time.monotonic() < deadline:
last = next(
(item for item in self.folders() if Path(item["local_path"]).resolve() == path),
None,
)
if last and last["state"] == "idle" and int(last["last_sync"]) > before:
return
if last and last["state"] in {"error", "conflict"}:
raise RuntimeError(f"forced sync for {path} entered {last['state']}: {last}")
time.sleep(1)
raise TimeoutError(f"forced sync for {path} did not complete: {last}")
def set_modes(self, modes: tuple[str, str]) -> None:
changed_to_mirror: set[Path] = set()
for path, mode in zip(self.paths, modes, strict=True):
current = self.wait_for(path)
if current["mode"] != mode:
print(f"[setup] switching {path} to {mode}")
self.command("sync", "mode", str(self.ids[path]), mode)
if mode == "mirror":
changed_to_mirror.add(path)
for path, mode in zip(self.paths, modes, strict=True):
self.wait_for(path, mode=mode)
# The mode row flips before the asynchronous restore pass starts,
# and its prior idle/last_sync values remain visible meanwhile.
# Demand a completed pass before inspecting restored local bytes.
if path in changed_to_mirror:
self.force_sync(path)
mounted = is_mountpoint(path)
check(mounted == (mode == "ondemand"), f"{path}: mode is {mode}, mounted={mounted}")
check(
read(path / "pdfs-mode-preservation.bin") == self.sentinels[path],
f"{path}: preservation sentinel changed or vanished after switch to {mode}",
)
def cleanup(self) -> None:
# An async add can create its row and then time out before create() has
# recorded the id. validate() proved these exact paths were unregistered
# at entry, so rows now at those paths belong to this run.
try:
for item in self.folders():
path = Path(item["local_path"]).resolve()
if path in self.paths:
self.ids[path] = int(item["id"])
except Exception as error:
print(f"WARNING: could not rediscover managed sync folders: {error}")
for path, folder_id in reversed(list(self.ids.items())):
try:
print(f"[cleanup] removing test sync folder {path}")
self.command("sync", "rm", str(folder_id), "--delete-remote")
except Exception as error:
print(f"WARNING: cleanup failed for sync folder {folder_id}: {error}")
continue
# Unregistering a mirror intentionally preserves its local copy.
# Remove only the sentinel owned by this harness so the next run's
# strict empty-directory precondition remains meaningful. Unknown
# survivors are left untouched and will fail validate() next time.
sentinel = path / "pdfs-mode-preservation.bin"
try:
if sentinel.exists() and read(sentinel) == self.sentinels[path]:
sentinel.unlink()
except OSError as error:
print(f"WARNING: could not remove managed sentinel {sentinel}: {error}")
def is_mountpoint(path: Path) -> bool:
"""True for a distinct mount, including FUSE mounts over an existing dir."""
return os.path.ismount(path)
def run_managed_matrix(paths: list[Path]) -> None:
timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "180"))
pair = ManagedSyncPair(paths, timeout)
roots: list[Path] = []
try:
pair.create()
matrices = [
("ondemand", "ondemand"),
("ondemand", "mirror"),
("mirror", "ondemand"),
("mirror", "mirror"),
]
for modes in matrices:
print(f"[matrix] first={modes[0]}, second={modes[1]}")
pair.set_modes(modes)
for path, mode in zip(pair.paths, modes, strict=True):
root, _ = run_contract(path, f"managed {mode}")
roots.append(root)
pair.wait_for_queue()
pair.remove_tree(root)
roots.remove(root)
pair.wait_for_queue()
# Mirror changes are asynchronous; force and await a pass before
# changing its mode so authored bytes/deletions cannot be lost.
if mode == "mirror":
pair.force_sync(path)
finally:
for root in roots:
shutil.rmtree(root, ignore_errors=True)
pair.cleanup()
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
live_mode = parser.add_mutually_exclusive_group()
live_mode.add_argument("--live", nargs="+", type=Path, metavar="MOUNTPOINT")
live_mode.add_argument(
"--managed-live",
nargs=2,
type=Path,
metavar=("EMPTY_DIR_A", "EMPTY_DIR_B"),
)
args = parser.parse_args()
with tempfile.TemporaryDirectory(prefix="pdfs-api-reference-") as directory:
reference, _ = run_contract(Path(directory), "local filesystem API reference (no account)")
shutil.rmtree(reference)
print("[pass] account-free filesystem API contract")
if args.managed_live:
run_managed_matrix(args.managed_live)
print("PASS: offline API and managed live mode-matrix suites completed")
return
if not args.live:
print("PASS: offline acceptance suite completed; live FUSE test was not requested")
return
completed: list[tuple[Path, Path, str]] = []
convergence = os.environ.get("PDFS_ACCEPTANCE_CONVERGENCE", "0") == "1"
try:
# Views of one remote folder must not independently create the same
# fixtures. Exercise the primary, then observe that exact tree through
# every secondary. Without convergence mode, each mount is independent
# and receives the full contract.
targets = args.live[:1] if convergence and len(args.live) > 1 else args.live
for mountpoint in targets:
root, digest = run_contract(mountpoint.resolve(), "live FUSE")
completed.append((mountpoint.resolve(), root, digest))
if convergence and len(args.live) > 1:
print(" [test] cross-mount remote convergence")
timeout = int(os.environ.get("PDFS_ACCEPTANCE_SYNC_TIMEOUT", "120"))
source_root, digest = completed[0][1], completed[0][2]
relative = source_root.relative_to(completed[0][0]) / "positioned.bin"
for mountpoint in args.live[1:]:
wait_for_copy(mountpoint.resolve(), relative, digest, timeout)
finally:
for _, root, _ in completed:
shutil.rmtree(root, ignore_errors=True)
print("PASS: offline API and live FUSE acceptance suites completed")
if __name__ == "__main__":
main()
+33 -136
View File
@@ -1,150 +1,47 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# Destructive only inside a uniquely named directory created by this script. # Filesystem syscall contract, with optional tests against mounted Drive views.
set -Eeuo pipefail set -Eeuo pipefail
usage() { usage() {
echo "usage: $0 MOUNTPOINT [MOUNTPOINT ...]" >&2 cat >&2 <<'EOF'
echo "example: $0 /mnt/on-demand-testmount /mnt/testmount" >&2 usage: scripts/fuse-acceptance.sh [--offline-only]
scripts/fuse-acceptance.sh --live MOUNTPOINT [MOUNTPOINT ...]
scripts/fuse-acceptance.sh --managed-live EMPTY_DIR EMPTY_DIR
scripts/fuse-acceptance.sh MOUNTPOINT [MOUNTPOINT ...] # compatibility
The local reference suite always runs first and requires no account. --live then
runs the same contract on each mount. Set PDFS_ACCEPTANCE_CONVERGENCE=1 only
when all live mountpoints show the same remote folder.
EOF
exit 2 exit 2
} }
(( $# > 0 )) || usage
command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; } command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; }
command -v findmnt >/dev/null || { echo "findmnt is required" >&2; exit 2; } script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
mounts=("$@") case "${1:-}" in
run_id="pdfs-acceptance-$(date +%Y%m%dT%H%M%S)-$$" -h|--help) usage ;;
roots=() --offline-only|"")
(( $# <= 1 )) || usage
exec python3 -u "$script_dir/fuse-acceptance.py"
;;
--live) shift ; (( $# > 0 )) || usage ;;
--managed-live)
shift
(( $# == 2 )) || usage
exec python3 -u "$script_dir/fuse-acceptance.py" --managed-live "$@"
;;
--*) usage ;;
esac
cleanup() { command -v findmnt >/dev/null || { echo "findmnt is required for --live" >&2; exit 2; }
local root for mountpoint in "$@"; do
for root in "${roots[@]}"; do
[[ -n "$root" && "$root" == */pdfs-acceptance-* ]] || continue
rm -rf -- "$root" 2>/dev/null || true
done
}
trap cleanup EXIT INT TERM
for index in "${!mounts[@]}"; do
mountpoint="${mounts[$index]}"
[[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; } [[ -d "$mountpoint" ]] || { echo "not a directory: $mountpoint" >&2; exit 2; }
fs_type="$(findmnt -T "$mountpoint" -n -o FSTYPE | head -n 1)"
if (( index == 0 )) && [[ "$fs_type" != fuse* ]]; then
echo "refusing primary non-FUSE path $mountpoint (type: ${fs_type:-unknown})" >&2
exit 2
fi
roots+=("${mountpoint%/}/$run_id")
done done
fstype="$(findmnt -T "$1" -n -o FSTYPE | head -n 1)"
step() { printf '[test] %s\n' "$1"; } [[ "$fstype" == fuse* ]] || {
assert_text() { echo "refusing primary non-FUSE path $1 (type: ${fstype:-unknown})" >&2
python3 - "$1" "$2" <<'PY' exit 2
import sys
p, expected = sys.argv[1], sys.argv[2].encode()
with open(p, "rb") as f: got = f.read()
if got != expected:
raise AssertionError(
f"{p}: expected {expected!r} ({len(expected)} bytes), "
f"got {got!r} ({len(got)} bytes)"
)
PY
} }
run_suite() { exec python3 -u "$script_dir/fuse-acceptance.py" --live "$@"
local primary="$1"
printf '[mount] testing %s\n' "${primary%/*}"
mkdir -- "$primary"
step "create, read, overwrite, truncate"
printf 'alpha\n' > "$primary/plain.txt"
assert_text "$primary/plain.txt" $'alpha\n'
printf 'a much longer replacement\n' > "$primary/plain.txt"
truncate -s 4 "$primary/plain.txt"
assert_text "$primary/plain.txt" "a mu"
step "binary round-trip and random-access writes"
python3 - "$primary/binary.dat" <<'PY'
import hashlib, sys
p = sys.argv[1]
data = bytes(range(256)) * 16384
with open(p, "wb") as f: f.write(data)
with open(p, "r+b", buffering=0) as f:
f.seek(4093); f.write(b"boundary-write")
with open(p, "rb") as f: got = f.read()
expected = data[:4093] + b"boundary-write" + data[4107:]
assert got == expected
print(hashlib.sha256(got).hexdigest())
PY
step "rename, move, and replacement semantics"
mkdir -- "$primary/dir-a" "$primary/dir-b"
printf 'source' > "$primary/source"
printf 'victim' > "$primary/victim"
mv -- "$primary/source" "$primary/victim"
assert_text "$primary/victim" "source"
[[ ! -e "$primary/source" ]]
mv -- "$primary/victim" "$primary/dir-a/moved"
assert_text "$primary/dir-a/moved" "source"
step "POSIX error behavior"
if rmdir -- "$primary/dir-a/moved" 2>/dev/null; then echo "rmdir removed a file" >&2; exit 1; fi
if rm -- "$primary/dir-a" 2>/dev/null; then echo "rm removed a directory" >&2; exit 1; fi
if mv -- "$primary/dir-a" "$primary/dir-a/child" 2>/dev/null; then
echo "moving a directory below itself succeeded" >&2; exit 1
fi
step "open file remains readable after unlink"
python3 - "$primary/open-unlink" <<'PY'
import os, sys
p = sys.argv[1]; data = b"held-open\0" * 4096
with open(p, "wb") as f: f.write(data)
f = open(p, "rb", buffering=0); os.unlink(p)
assert not os.path.exists(p) and f.read() == data
f.close()
PY
step "concurrent independent writers"
for i in $(seq 1 12); do
(python3 - "$primary/concurrent-$i" "$i" <<'PY'
import sys
p, seed = sys.argv[1], int(sys.argv[2]); data = bytes([seed]) * (131072 + seed)
with open(p, "wb") as f: f.write(data)
with open(p, "rb") as f: assert f.read() == data
PY
) &
done
wait
step "directory enumeration agrees with stat"
python3 - "$primary" <<'PY'
import os, sys
for name in os.listdir(sys.argv[1]): os.lstat(os.path.join(sys.argv[1], name))
PY
}
for root in "${roots[@]}"; do
run_suite "$root"
done
if [[ "${PDFS_ACCEPTANCE_CONVERGENCE:-0}" == 1 ]] && (( ${#roots[@]} > 1 )); then
step "cross-mount convergence"
primary="${roots[0]}"
expected="$(sha256sum "$primary/binary.dat" | cut -d' ' -f1)"
for secondary in "${roots[@]:1}"; do
deadline=$((SECONDS + ${PDFS_ACCEPTANCE_SYNC_TIMEOUT:-120}))
while (( SECONDS < deadline )); do
if [[ -f "$secondary/binary.dat" ]] &&
[[ "$(sha256sum "$secondary/binary.dat" 2>/dev/null | cut -d' ' -f1)" == "$expected" ]]; then break; fi
sleep 2
done
[[ -f "$secondary/binary.dat" ]] || { echo "did not appear on $secondary" >&2; exit 1; }
[[ "$(sha256sum "$secondary/binary.dat" | cut -d' ' -f1)" == "$expected" ]] || {
echo "content mismatch on $secondary" >&2; exit 1
}
done
fi
step "cleanup"
cleanup
roots=()
echo "PASS: FUSE acceptance suite completed ($run_id)"