diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 05f94f5..91a74e3 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -837,6 +837,19 @@ impl Db { /// drain replaces it with the uid the server assigns. pub const LOCAL_VOLUME: &str = "local"; + /// Whether op `id` is still queued. A drain worker that claimed it can find + /// it gone: the node was deleted, or a newer write superseded it, while the + /// worker was talking to the server. + pub fn op_exists(&self, id: i64) -> Result { + let conn = self.read(); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM pending_op WHERE id = ?1", + params![id], + |row| row.get(0), + )?; + Ok(count > 0) + } + /// Drop a queued op, once its upload has actually landed. pub fn delete_op(&self, id: i64) -> Result<()> { let conn = self.conn.lock(); diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index f86b995..ab46761 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -5209,3 +5209,28 @@ fn retry_all_touches_only_ops_that_have_failed() { assert_eq!(due(debounced), 7_000); assert_eq!(due(parked), ops::PARK_UNTIL); } + +#[test] +fn a_queued_op_stops_existing_once_the_node_is_deleted() { + let db = Db::open_in_memory().unwrap(); + let (id, _) = db + .enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("journal").to_string(), + parent_uid: None, + name: None, + blob_path: Some("/staging/journal".to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + assert!(db.op_exists(id).unwrap()); + + let blobs = db.delete_ops_for_uid(&uid("journal").to_string()).unwrap(); + assert_eq!(blobs, vec!["/staging/journal"]); + assert!(!db.op_exists(id).unwrap()); +} diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 655a1db..d588552 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -1486,6 +1486,16 @@ impl Core { let uid = parse_node_uid(&meta.uid).ok_or("staged write has an unparseable uid")?; if let Some(reason) = self.revision_conflict(&uid, &meta)? { + // Trashing the file from the mount drops its queued write, but not + // from under a worker already holding it. That worker then finds the + // node trashed — by us — and would upload the withdrawn bytes as a + // conflict copy (a `recovered-…` file in the root when the tree has + // already forgotten the node). A write nobody queues any more is + // not a conflict. + if !self.db.op_exists(op.id)? { + debug!(%uid, reason, "queued write withdrawn while it drained; dropping it"); + return Ok(()); + } return self.keep_as_conflict_copy(op, &blob, &meta, &uid, &reason); } @@ -1522,6 +1532,12 @@ impl Core { // this is on the wire can stop it rather than paying for a revision the // op it queued will immediately replace. let cancel = self.begin_cancellable_upload(&uid); + // An unlink that withdrew this op before the flag above existed had no + // upload to cancel; the row is the record that it happened. + if !self.db.op_exists(op.id)? { + debug!(%uid, "queued write withdrawn before it went out; dropping it"); + return Ok(()); + } let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag()); let started = Instant::now(); let sent = self.rt.block_on(self.client.upload_new_revision_from( diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index bbabe15..2cf2268 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -1743,12 +1743,18 @@ impl ProtonFs { if let Some(uid) = unlinked_uid && release_can_discard_unlinked(&self.core.db, &uid) { + // Ops withdrawn at unlink left the bytes this handle read behind, + // owned by no op any more; they go with the last handle. + let withdrawn = self.core.pending_blob(&uid); if let Err(error) = self.core.discard_queued_ops(&uid) { warn!(%uid, ?error, "queued-op cleanup failed; retaining unlinked state"); } else { if let Err(error) = self.core.db.delete_node(&uid) { warn!(%uid, %error, "deleting released unlinked node state failed"); } + if let Some(blob) = withdrawn { + self.core.cache.discard_staged(&blob); + } self.core.cache.evict(&uid); self.core.evict_reader(&uid); } @@ -2129,7 +2135,12 @@ impl ProtonFs { // it just means its queued creation is no longer wanted. This works // offline, which the remote path below cannot (offline.md Phase 3b). if is_local_uid(&uid) { - if !open_now && let Err(error) = self.core.discard_queued_ops(&uid) { + let dropped = if open_now { + self.core.withdraw_queued_ops(&uid) + } else { + self.core.discard_queued_ops(&uid) + }; + if let Err(error) = dropped { reply.error(error); return; } @@ -2168,7 +2179,13 @@ impl ProtonFs { // Otherwise an EIO response still made the path disappear locally. self.core.hidden.lock().insert(uid.clone()); self.core.state().forget_or_unlink(&uid); - if !open_now { + if open_now { + // The open handle keeps the staged bytes; the drain must not send + // them to a node that is already in the trash (docs/BUGS.md B106). + if let Err(error) = self.core.withdraw_queued_ops(&uid) { + warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed"); + } + } else { if let Err(error) = self.core.discard_queued_ops(&uid) { warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed"); } diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index b01a139..98006a8 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -3539,6 +3539,30 @@ impl Core { Ok(()) } + /// Drop every op queued against a node that just left the namespace while + /// a handle still has it open, but keep the bytes that handle reads. + /// + /// The ops cannot wait for the final close: the node is already trashed, and + /// a revision the drain sends in the meantime lands on it after the trash — + /// an editor's save loop then found the name it was renaming onto still + /// taken (`docs/BUGS.md` B106). The pending entry and its staged blob are + /// what the open handle reads, so they stay until the final close drops + /// them. + fn withdraw_queued_ops(&self, uid: &NodeUid) -> Result<(), Errno> { + self.cancel_upload(uid); + let blobs = self.db.delete_ops_for_uid(&uid.to_string()).map_err(|e| { + error!(%uid, error = %e, "withdrawing queued ops failed"); + Errno::EIO + })?; + let read_from = self.pending_blob(uid); + let blobs = blobs + .into_iter() + .filter(|blob| read_from.as_deref() != Some(Path::new(blob))) + .collect(); + self.release_dropped_below(blobs); + Ok(()) + } + /// Let go of what the ops just dropped for `uid` held: their pending /// entries, any upload still reading their bytes, and the bytes. /// @@ -3546,9 +3570,15 @@ impl Core { /// (`docs/BUGS.md` B102), so the entries to forget are found by blob, not /// only by `uid`. fn release_dropped_ops(&self, uid: &NodeUid, blobs: Vec) { + self.pending.lock().remove(uid); + self.release_dropped_below(blobs); + } + + /// [`Core::release_dropped_ops`] without touching the node's own pending + /// entry. + fn release_dropped_below(&self, blobs: Vec) { let below: Vec = { let mut pending = self.pending.lock(); - pending.remove(uid); let below: Vec = pending .iter() .filter(|(_, p)| blobs.iter().any(|b| p.path == Path::new(b))) @@ -4420,11 +4450,21 @@ impl Core { /// A node whose own creation is still queued has never reached the server, so /// dropping its queued ops is the whole removal; nothing goes to the wire and /// it works offline. + /// + /// A handle still open on the replaced node keeps reading its bytes, as + /// `rename(2)` promises: the inode stays as an unlinked entry and its + /// staged bytes, cache and reader go at the final close, exactly as for an + /// `unlink` under an open handle. Its queued ops go now, or the drain would + /// upload to the node after it was trashed (`docs/BUGS.md` B106). fn remove_replaced(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> { if is_local_uid(uid) { - self.discard_queued_ops(uid)?; + if self.is_open_anywhere(uid) { + self.withdraw_queued_ops(uid)?; + } else { + self.discard_queued_ops(uid)?; + } self.for_each_state(|st| { - st.forget(uid); + st.forget_or_unlink(uid); }); debug!(%uid, name, "replaced a node whose create was still queued"); return Ok(()); @@ -4439,17 +4479,26 @@ impl Core { self.log_activity(ActivityKind::Trash, name, e.to_string(), false); return Err(Errno::EIO); } - if let Err(error) = self.discard_queued_ops(uid) { - error!(%uid, "remote replacement landed but queued-op cleanup failed"); - return Err(error); - } // The node was trashed on the server; withdraw it from every inode // space that had it, not only the mount the rename came through. + let open_now = self.is_open_anywhere(uid); self.for_each_state(|st| { - st.forget(uid); + st.forget_or_unlink(uid); }); - self.cache.evict(uid); - self.evict_reader(uid); + self.hidden.lock().insert(uid.clone()); + if open_now { + if let Err(error) = self.withdraw_queued_ops(uid) { + error!(%uid, "remote replacement landed but queued-op cleanup failed"); + return Err(error); + } + } else { + if let Err(error) = self.discard_queued_ops(uid) { + error!(%uid, "remote replacement landed but queued-op cleanup failed"); + return Err(error); + } + self.cache.evict(uid); + self.evict_reader(uid); + } self.invalidate_trash(); // The node is recoverable from the trash, but only if the user knows it // went there — a rename is not an operation anyone expects to trash @@ -4463,6 +4512,19 @@ impl Core { Ok(()) } + /// Whether any mount holds an open handle on `uid`. + fn is_open_anywhere(&self, uid: &NodeUid) -> bool { + let mut open = false; + self.for_each_state(|st| { + open |= st + .by_uid + .get(uid) + .and_then(|ino| st.entries.get(ino)) + .is_some_and(|entry| entry.open_count > 0); + }); + open + } + /// Put back the node [`Core::remove_replaced`] trashed, after the rename it /// was clearing the way for failed anyway. /// @@ -7946,6 +8008,7 @@ mod tests { ("fn enqueue_staged_write(", "attach_blob_to_create"), ("fn queue_trash(", "replace_ops_with_trash"), ("fn discard_queued_ops(", "delete_ops_for_uid"), + ("fn withdraw_queued_ops(", "delete_ops_for_uid"), ] { let function = function_source(source, signature); assert_before(function, "cancel_upload", mutation); @@ -8006,6 +8069,84 @@ mod tests { assert_before(folder, "require_uid_writable(parent_uid)", ".create_folder"); } + #[test] + fn a_replaced_node_survives_its_open_handles() { + // A rename over a file someone is reading must leave that reader on the + // replaced bytes; forgetting the inode outright made its next read EIO. + let source = include_str!("lib.rs"); + let replaced = function_source(source, "fn remove_replaced("); + assert!(replaced.contains("forget_or_unlink(uid)")); + assert!(!replaced.contains("st.forget(uid)")); + assert_before( + replaced, + "if self.is_open_anywhere(uid)", + "self.discard_queued_ops(uid)", + ); + assert_before(replaced, "if open_now", "self.cache.evict(uid)"); + // ...but its queued ops go at once, open or not: a revision drained + // after the trash landed on the trashed node and kept its name taken. + assert_eq!(replaced.matches("self.withdraw_queued_ops(uid)").count(), 2); + } + + #[test] + fn a_remote_read_gives_up_at_its_deadline() { + // The SDK bounds requests, not reads; a stalled link held one `read` + // for 34 minutes. + let remote = function_source(include_str!("reads.rs"), "fn read_range_remote("); + assert_before( + remote, + "let fetch = async", + "tokio::time::timeout(READ_FETCH_TIMEOUT, fetch)", + ); + // Built inside the runtime: a FUSE worker has no reactor, and a timer + // built there panicked every remote read. + assert!(remote.contains("async { tokio::time::timeout(READ_FETCH_TIMEOUT, fetch).await }")); + } + + #[test] + fn a_deadline_built_under_block_on_needs_no_ambient_runtime() { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .unwrap(); + let answered = std::thread::spawn(move || { + rt.block_on(async { tokio::time::timeout(Duration::from_secs(5), async { 7 }).await }) + }) + .join() + .expect("a worker thread outside the runtime can wait on the deadline"); + assert_eq!(answered.unwrap(), 7); + } + + #[test] + fn a_trashed_open_node_sends_nothing_more() { + // Trashing a file under an open handle withdraws its queued upload at + // once; the bytes the handle reads go with the last close. + let withdraw = function_source(include_str!("lib.rs"), "fn withdraw_queued_ops("); + assert_before(withdraw, "self.pending_blob(uid)", "release_dropped_below"); + assert!(!withdraw.contains("self.pending.lock().remove(uid)")); + + let filesystem = include_str!("filesystem.rs"); + let trash = function_source(filesystem, "fn trash_child("); + assert_eq!( + trash.matches("self.core.withdraw_queued_ops(&uid)").count(), + 2 + ); + let release = function_source(filesystem, "fn finish_release("); + assert_before( + release, + "let withdrawn = self.core.pending_blob(&uid)", + "discard_queued_ops", + ); + assert_before(release, "discard_queued_ops", "discard_staged(&blob)"); + + // A drain worker that claimed the op before it was withdrawn checks + // again before any byte leaves. + let drain = function_source(include_str!("drain.rs"), "pub(crate) fn drain_revision("); + let upload = &drain[drain.find("begin_cancellable_upload").unwrap()..]; + assert_before(upload, "self.db.op_exists(op.id)", "File::open(&blob)"); + } + #[test] fn replacement_rename_authorizes_source_and_both_parents() { use std::cell::RefCell; diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs index e93382f..0a3313d 100644 --- a/crates/pdfs-fuse/src/reads.rs +++ b/crates/pdfs-fuse/src/reads.rs @@ -36,6 +36,15 @@ pub(super) const REPAIR_MAX: u64 = 512 * 1024 * 1024; /// demand read in the SDK's flat block semaphore. pub(super) const PREFETCH_BUDGET: usize = 8; +/// How long one read may wait on the network before it fails with `EIO`. +/// +/// The SDK bounds each request, not a read: a block that times out is retried, +/// and a read can need several blocks, a block table and a repair. On a link +/// that stalled without dropping, one `read` held a FUSE worker for 34 minutes +/// and the application behind it with it (`docs/BUGS.md` B107). An error the +/// caller can retry beats a process stuck in `D` state. +pub(super) const READ_FETCH_TIMEOUT: Duration = Duration::from_secs(120); + /// Cap on tracked sequential streams. Only reads in progress matter, so the map /// is cleared wholesale rather than aged — losing the ramp costs one read. const MAX_PREFETCH_STREAMS: usize = 256; @@ -939,7 +948,7 @@ impl Core { } if !misses.is_empty() { - let fetched = self.rt.block_on(async { + let fetch = async { // Resolve the file's keys and block table once, then read every // missing block through the shared reader. Previously each block // called `download_range`, which redid that resolution per block. @@ -960,7 +969,22 @@ impl Core { out.push(joined.map_err(|_| Errno::EIO)??); } Ok::<_, Errno>(out) - })?; + }; + // Dropping the fetch on the deadline aborts the block tasks with it; + // blocks that already landed stay cached for the retry. The timer is + // built inside the runtime: a FUSE worker has no reactor of its own, + // and building it out here panicked every remote read. + let fetched = self + .rt + .block_on(async { tokio::time::timeout(READ_FETCH_TIMEOUT, fetch).await }) + .unwrap_or_else(|_| { + warn!( + %uid, offset, len, secs = READ_FETCH_TIMEOUT.as_secs(), + "read got no answer from the network in time; failing it" + ); + self.evict_reader(uid); + Err(Errno::EIO) + })?; for (bidx, bytes) in fetched { blocks[(bidx - first) as usize] = Some(bytes); } diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index cfa5193..52eef56 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -73,6 +73,9 @@ enum Applied { Downloaded, /// A conflict copy was kept. Conflict, + /// The local file to upload was deleted after the scan planned it. There is + /// nothing left to send, and the next pass sees the deletion (B108). + Vanished, } impl Core { @@ -1147,13 +1150,10 @@ impl Core { Ok(Applied::Dir(rel.clone(), uid)) } Pending::UploadNew { rel, parent } => { - self.upload_new(folder_id, local_root, rel, parent).await?; - Ok(Applied::Uploaded) + self.upload_new(folder_id, local_root, rel, parent).await } Pending::UploadRevision { rel, uid } => { - self.upload_revision(folder_id, local_root, rel, uid) - .await?; - Ok(Applied::Uploaded) + self.upload_revision(folder_id, local_root, rel, uid).await } Pending::Download { rel, @@ -1217,10 +1217,12 @@ impl Core { local_root: &Path, rel: &str, parent: &NodeUid, - ) -> Result<(), String> { + ) -> Result { let path = local_root.join(rel_to_path(rel)); let name = base_name(rel); - let file = std::fs::File::open(&path).map_err(|e| format!("open {rel}: {e}"))?; + let Some(file) = open_to_upload(&path, rel)? else { + return Ok(Applied::Vanished); + }; let meta = file.metadata().map_err(|e| format!("stat {rel}: {e}"))?; let mtime = system_mtime(&meta); // Taken from the descriptor being streamed, so it names the content that @@ -1261,7 +1263,8 @@ impl Core { self.invalidate_children_of(parent); warn_if_torn(rel, &path, streamed); self.record_file_baseline(folder_id, rel, Some(streamed), &uid) - .await + .await?; + Ok(Applied::Uploaded) } /// Upload a changed local file as a new revision of an existing remote node. @@ -1271,9 +1274,11 @@ impl Core { local_root: &Path, rel: &str, uid: &NodeUid, - ) -> Result<(), String> { + ) -> Result { let path = local_root.join(rel_to_path(rel)); - let file = std::fs::File::open(&path).map_err(|e| format!("open {rel}: {e}"))?; + let Some(file) = open_to_upload(&path, rel)? else { + return Ok(Applied::Vanished); + }; let meta = file.metadata().map_err(|e| format!("stat {rel}: {e}"))?; let mtime = system_mtime(&meta); let streamed = LocalSig::from(&meta); @@ -1295,7 +1300,8 @@ impl Core { .map_err(|e| format!("upload revision {rel}: {e}"))?; warn_if_torn(rel, &path, streamed); self.record_file_baseline(folder_id, rel, Some(streamed), uid) - .await + .await?; + Ok(Applied::Uploaded) } /// Download a remote file to its local path (atomically via a temp file), @@ -1445,6 +1451,23 @@ impl Core { } } +/// Open a local file for upload, or `None` when it is gone. +/// +/// A scan plans uploads from a listing, and the file can be deleted before its +/// upload starts — a build or a test tree cleaned up mid-pass. That is not a +/// failure: the next pass sees the deletion. Counting it as one put the folder +/// in `error` for a file nobody wanted any more (B108). +fn open_to_upload(path: &Path, rel: &str) -> Result, String> { + match std::fs::File::open(path) { + Ok(file) => Ok(Some(file)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + debug!(rel, "sync: file deleted before its upload; skipping"); + Ok(None) + } + Err(e) => Err(format!("open {rel}: {e}")), + } +} + /// Say so when a file changed under an upload that was streaming it. /// /// Purely diagnostic — the correction is already in place, because the baseline @@ -2239,6 +2262,27 @@ mod tests { assert_eq!(remote_sig(&e), None); } + #[test] + fn a_file_deleted_before_its_upload_is_skipped_not_failed() { + let root = std::env::temp_dir().join(format!("pdfs-b108-{}", std::process::id())); + std::fs::create_dir_all(&root).unwrap(); + let gone = root.join("gone.txt"); + assert!(open_to_upload(&gone, "gone.txt").unwrap().is_none()); + + std::fs::write(&gone, b"x").unwrap(); + assert!(open_to_upload(&gone, "gone.txt").unwrap().is_some()); + + // Other open failures still count as errors. + let err = open_to_upload(&gone.join("child"), "gone.txt/child"); + assert!(err.unwrap_err().starts_with("open gone.txt/child:")); + let _ = std::fs::remove_dir_all(&root); + + let mut o = Outcome::default(); + o.record(&Applied::Vanished); + assert!(o.is_empty()); + assert_eq!(o.errors, 0); + } + #[test] fn outcome_summarises_only_what_it_moved() { let mut o = Outcome::default(); diff --git a/crates/pdfs-fuse/src/sync/model.rs b/crates/pdfs-fuse/src/sync/model.rs index 01f9df2..afb55e3 100644 --- a/crates/pdfs-fuse/src/sync/model.rs +++ b/crates/pdfs-fuse/src/sync/model.rs @@ -108,6 +108,7 @@ impl Outcome { Applied::Uploaded => self.uploaded += 1, Applied::Downloaded => self.downloaded += 1, Applied::Conflict => self.conflicts += 1, + Applied::Vanished => {} } } diff --git a/docs/BUGS.md b/docs/BUGS.md index bd70ba9..0e45276 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -12,6 +12,134 @@ Conventions: --- +## B108 — A file deleted during a mirror pass puts the sync folder in `error` + +**Status:** Fixed (unverified). +**Found:** 2026-09-29, by the account run of `scripts/fuse-acceptance.sh --quick`. It stopped with +`sync folder … entered error` before the managed-move phase. At the same time the journal +showed `sync: step failed; continuing error=open …/workloads/tree.tar: No such file or directory` +for eleven files. The contract suite had removed its `workloads` tree while the mirror pass +was still uploading it. + +**Where:** `crates/pdfs-fuse/src/sync.rs` (`upload_new`, `upload_revision`, `open_to_upload`). + +**Cause.** A pass plans its uploads from a scan and runs them later. When a planned file was +deleted in between, opening it failed with `NotFound`. That counted as a failed step, so the pass +ended in `error` and logged "item(s) failed; will retry" for files nobody wanted any more. + +**Fix.** An upload whose local file is gone when it starts is skipped (`Applied::Vanished`) and +does not count as an error. The next pass sees the deletion and handles it like any other. +Other open failures are still errors. + +**Also open from the same run.** "application workloads" on My files failed with +`database disk image is malformed` at the `SELECT` after the insert. The daemon logged nothing +for it. The case now reports the step that failed, the database and journal files beside it, +and which pages differ from a local replay of the same statements. + +## B107 — One read on a stalled connection hangs the reading program for half an hour + +**Status:** Fixed (unverified). +**Found:** 2026-09-29, by the acceptance case "application workloads" on My files, which hit its +180 s timeout. The journal showed `a fuse worker has held the same job for a long time +worker=pdfs-fuse-10 job=read` from `age_secs=124` up to `age_secs=2059`. Event polls in the +same window failed with `HTTP transport error` after 91-103 s. + +**Where:** `crates/pdfs-fuse/src/reads.rs` (`Core::read_range_remote`, `READ_FETCH_TIMEOUT`). + +**Cause.** The SDK bounds each request (30 s for the API, 300 s for a storage block) and retries +failed ones, but nothing bounds a whole read. A read can need a block table, several blocks and a +repair, each with its own retries. On a connection that stalled without dropping, one `read` +held its FUSE worker for about 34 minutes, and the program that issued it could not be +interrupted for that time. + +**Fix.** The network part of a read now has a 120 s deadline. When it passes, the read fails +with `EIO`, the file's revision reader is dropped so the next read starts fresh, and blocks that +already arrived stay cached for the retry. A single read is at most a few 4 MiB blocks, so this +only fails a read on a link slower than about 70 KB/s. A block repair that has to download a +large file whole can also hit the deadline on a slow link; pinning the file avoids that. + +The first version of this fix built the timer on the FUSE worker thread, outside the runtime. +Every read that went to the network panicked with `there is no reactor running`, and the kernel +got `EIO`. The next account run failed 11 cases on My files this way. The timer is now built +inside `block_on`, and a unit test runs that pattern from a thread without a runtime. + +--- + +## B106 — Saving over a file that was just read fails with "already exists" + +**Status:** Fixed (unverified). +**Found:** 2026-09-29, by the acceptance case "application workloads" on an on-demand folder, +after installing the B104 fix: `OSError: [Errno 5]` from `os.replace` in the editor-save loop, +at the third save. The journal had `rename failed … AlreadyExists (http 422): A file or folder +with name already exists` from `rename_node`. + +**Where:** `crates/pdfs-fuse/src/lib.rs` (`Core::remove_replaced`, +`Core::withdraw_queued_ops`), `crates/pdfs-fuse/src/filesystem.rs` (`trash_child`, +`finish_release`), `crates/pdfs-fuse/src/drain.rs` (`Core::drain_revision`). + +**Cause.** A regression from B104. The save loop writes a temporary file, renames it over the +document, and reads the document back. `close()` returns before the daemon sees the release, so +at the next save the replaced file still counts as open. B104 then left its queued upload in +the queue until the final close, and the drain could send it after the file had been trashed. +The next rename onto the name was refused. `unlink` of an open file had kept its queued upload +the same way. + +**Fix.** A file trashed or replaced while open now loses its queued ops at once, and an upload +already under way is cancelled. Only the staged bytes stay, because the open handle reads them, +and they are deleted at the final close. A drain worker that claimed the op before it was +withdrawn checks that the op still exists before it sends any bytes. + +--- + +## B105 — A file deleted while its upload is in flight comes back as a conflict copy + +**Status:** Fixed (unverified). +**Found:** 2026-09-29, in the daemon journal of an acceptance run on the account. During the +SQLite workload two `queued write conflicts; keeping a conflict copy` warnings named files called +`recovered-XrPUSW5W` and `recovered-JbZM_pWK`, with the reason `the file was trashed remotely`. +Each was followed by `pending upload failed; will retry` and `No such file or directory`. + +**Where:** `crates/pdfs-fuse/src/drain.rs` (`Core::drain_revision`), +`crates/pdfs-core/src/db/ops.rs` (`Db::op_exists`). + +**Cause.** SQLite writes, syncs and deletes its journal file within milliseconds. The sync +queues an upload of the journal. The delete trashes the node and drops that queued upload, but a +drain worker that already holds the op carries on. It then asks the server about the node, +finds it trashed, and treats the trash as another device's change: it tries to upload the bytes +as a conflict copy. The tree has already forgotten the node, so the copy is named `recovered-…` +and placed in the Drive root. Here the blob was already gone and the attempt failed, but with +the blob still on disk the copy would land. + +**Fix.** Before keeping a conflict copy, the drain checks that its op is still queued. When the +op is gone, the write was withdrawn by the delete, or superseded, while the worker held it, and +it is dropped. + +--- + +## B104 — Reading a file that was replaced by a rename fails with `EIO` + +**Status:** Fixed (unverified). +**Found:** 2026-09-29, by the acceptance case "rename patterns" on My files and on an on-demand +folder: `OSError: [Errno 5] Input/output error` at +`os.pread(reader, 64, 0)` after `os.replace(replacement, target)` with `reader` open on `target`. +The mirror folder, a local disk, passed. + +**Where:** `crates/pdfs-fuse/src/lib.rs` (`Core::remove_replaced`). + +**Cause.** A rename over an existing file first trashes that file, because Drive has no atomic +replace. That step forgot the replaced inode outright and evicted its cache and reader, even +when a handle was still open on it. The next read through that handle found no inode and +failed. `rename(2)`, like `unlink(2)`, keeps a replaced file readable through the handles +already open on it. The unlink path (`trash_child`) already did this, but the rename path +did not. + +**Fix.** `remove_replaced` now handles an open victim as `unlink` does. It keeps the inode as +an unlinked entry, and leaves its cache and reader to the final close. That applies both to a +node the server knows and to one whose create is still queued. Its queued ops at first also +waited for the final close; B106 moved them back to the trash. + +--- + ## B103 — One stuck trash refresh freezes the trash listing until the daemon restarts **Status:** Fixed (unverified). The hang itself has not been reproduced on purpose. diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 08cd8ce..9b01999 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -31,6 +31,21 @@ scratch (user data in `staging/` and `recovery/` is never touched by this). - **Removing a folder also cancels the pending uploads of the files inside it.** Before, an edit that had not reached Drive yet kept retrying against the trash forever and kept its staged copy on disk. `pdfs rm` now also drops what was queued for what it removes (B102). +- **A program reading a file keeps its data when another program replaces the file.** Saving + over a file by renaming a new copy onto it (what most editors and `os.replace` do) made + reads through handles already open on the old file fail with `Input/output error` (B104). +- **Short-lived files no longer come back as conflict copies.** A file deleted right after it + was written, such as a SQLite journal, could be uploaded as a `recovered-…` conflict copy in + the Drive root (B105). +- **Editors can save over a file several times in a row.** The fix for B104 could leave the + upload of a replaced file queued, and the next save onto the same name failed with + `Input/output error` (B106). +- **Deleting files during a mirror pass no longer marks the folder as failed.** A file removed + after the pass planned its upload was counted as an error, so the folder showed `error` until + a later pass (B108). +- **A read no longer hangs when the connection stalls.** A read that gets no answer from Drive + within two minutes now fails with `Input/output error` instead of blocking the program for + as long as half an hour (B107). - **The trash no longer freezes after a server call hangs.** Until the daemon restarted, the trash kept showing the same entries, and a folder just deleted could not be found there to delete permanently (B103). diff --git a/scripts/fuse-acceptance.py b/scripts/fuse-acceptance.py index 2de3c6d..93654b0 100755 --- a/scripts/fuse-acceptance.py +++ b/scripts/fuse-acceptance.py @@ -35,6 +35,7 @@ import sys import tarfile import tempfile import time +import traceback import uuid import xml.etree.ElementTree as ElementTree @@ -117,6 +118,26 @@ def check(condition: bool, message: str) -> None: raise AssertionError(message) +def failure_detail(error: BaseException) -> str: + """The error, and for anything but a failed check the script line that hit it. + + A bare `OSError: [Errno 5]` from a case with forty filesystem calls says + nothing about which one the mount refused. + """ + detail = f"{type(error).__name__}: {error}" + if isinstance(error, AssertionError): + return detail + frames = [ + frame + for frame in traceback.extract_tb(error.__traceback__) + if frame.filename == __file__ and frame.name not in {"check", "run"} + ] + if frames: + frame = frames[-1] + detail += f" (at {frame.name}:{frame.lineno}: {frame.line})" + return detail + + def expect_errno(expected: set[int], operation, description: str) -> None: try: operation() @@ -907,6 +928,59 @@ def test_concurrency(ctx: Context) -> None: ctx.record("shared.digest", hashlib.sha256(read(shared)).hexdigest()) +def _sqlite_create(connection: sqlite3.Connection) -> None: + connection.execute("CREATE TABLE rows (id INTEGER PRIMARY KEY, value TEXT)") + + +def _sqlite_insert(connection: sqlite3.Connection) -> None: + with connection: + connection.executemany( + "INSERT INTO rows (value) VALUES (?)", [(f"row-{index}",) for index in range(500)] + ) + + +def _sqlite_diagnosis(database: Path) -> str: + """What a malformed database on the mount looks like next to a local replay. + + The same statements from the same library write the same bytes, so the + pages that differ, and whether a stray journal sat next to the file, say + whether the mount lost a write, served a stale read or let a rollback in. + """ + notes = [] + try: + entries = sorted( + f"{entry.name}={entry.stat(follow_symlinks=False).st_size}" + for entry in os.scandir(database.parent) + if entry.name.startswith(database.name) + ) + notes.append(f"files [{', '.join(entries)}]") + except OSError as error: + notes.append(f"listing failed: {error}") + try: + actual = database.read_bytes() + except OSError as error: + return "; ".join(notes + [f"reading the database failed: {error}"]) + with tempfile.TemporaryDirectory(prefix="pdfs-sqlite-reference-") as scratch: + reference_path = Path(scratch) / database.name + reference = sqlite3.connect(reference_path) + try: + _sqlite_create(reference) + _sqlite_insert(reference) + finally: + reference.close() + expected = reference_path.read_bytes() + page = int.from_bytes(expected[16:18], "big") or 65536 + notes.append(f"size {len(actual)} vs reference {len(expected)}") + differing = [] + for offset in range(0, max(len(actual), len(expected)), page): + ours, theirs = actual[offset : offset + page], expected[offset : offset + page] + if ours != theirs: + kind = "zeros" if ours.strip(b"\0") == b"" else "differs" + differing.append(f"{offset // page + 1}:{kind}") + notes.append(f"pages differing [{', '.join(differing[:16]) or 'none'}]") + return "; ".join(notes) + + def test_application_workloads(ctx: Context) -> None: """Real tools, because they combine syscalls in ways a matrix will not. @@ -954,14 +1028,20 @@ def test_application_workloads(ctx: Context) -> None: # sqlite exercises locking, journal creation and unlink, and fsync ordering. database = workloads / "workload.db" connection = sqlite3.connect(database) + step = "open" try: - connection.execute("CREATE TABLE rows (id INTEGER PRIMARY KEY, value TEXT)") - with connection: - connection.executemany( - "INSERT INTO rows (value) VALUES (?)", [(f"row-{index}",) for index in range(500)] - ) + step = "create table" + _sqlite_create(connection) + step = "insert" + _sqlite_insert(connection) + step = "checkpoint" connection.execute("PRAGMA wal_checkpoint(TRUNCATE)") + step = "count" count = connection.execute("SELECT COUNT(*) FROM rows").fetchone()[0] + except sqlite3.DatabaseError as error: + raise AssertionError( + f"sqlite failed at {step}: {error}; {_sqlite_diagnosis(database)}" + ) from error finally: connection.close() check(count == 500, f"sqlite lost rows: {count}") @@ -1972,7 +2052,7 @@ def run_contract( raise except BaseException as error: # noqa: BLE001 - reported, then continued elapsed = time.monotonic() - started - detail = f"{type(error).__name__}: {error}" + detail = failure_detail(error) print(f" FAIL {detail}") run.results.append(Result(case.name, label, "fail", elapsed, detail)) if fail_fast: @@ -2148,6 +2228,32 @@ class JournalWatch: # -------------------------------------------------------------------------- +# Queue ops that were already there when the run started, by id. They belong +# to the user, or to an earlier run that died, and may never drain: a run that +# waited for an empty queue would time out on every sync case. Waits skip them. +PREEXISTING_OPS: set[int] = set() + + +def remember_preexisting_ops(queue: list[dict]) -> None: + if PREEXISTING_OPS: + return + PREEXISTING_OPS.update(item["id"] for item in queue) + for item in queue: + print( + f"NOTE: queued {item['kind']} #{item['id']} for {item['path']} predates this run; " + "queue waits ignore it" + ) + + +def queue_settled(mount: dict, queue) -> bool: + """Whether everything this run queued has drained. `queue` lists the ops.""" + if mount.get("pending_uploads", 0) == 0 and mount.get("pending_changes", 0) == 0: + return True + if not PREEXISTING_OPS: + return False + return all(item["id"] in PREEXISTING_OPS for item in queue()) + + class Daemon: """The `pdfs` CLI, plus the unit control the durability drill needs.""" @@ -2166,6 +2272,8 @@ class Daemon: daemon.status() except Exception: return None + with contextlib.suppress(Exception): + remember_preexisting_ops(daemon.queue()) return daemon def command(self, *args: str, json_output: bool = False, stdin: str | None = None) -> str: @@ -2194,7 +2302,7 @@ class Daemon: last = None while time.monotonic() < deadline: last = self.status().get("mount") or {} - if last.get("pending_uploads", 0) == 0 and last.get("pending_changes", 0) == 0: + if queue_settled(last, self.queue): return time.sleep(1) raise TimeoutError(f"daemon mutation queue did not drain: {last}") @@ -2472,7 +2580,17 @@ class Janitor: uid cannot. The node has to show up in the trash first, which is also what proves the trash step before it worked, and it has to be gone from a fresh listing afterwards. + + The server's own answer comes first, though: it only deletes a node + that is in the trash, and it answers per uid. A listing is a whole-trash + enumeration, which on a large trash can time out on every attempt and + would otherwise keep a folder the server would delete on request. """ + with contextlib.suppress(Exception): + reply = self.daemon.command("delete-forever", uid, stdin="yes\n") + if re.search(r"\bpermanently deleted 1 item", reply): + print(f"[cleanup] permanently deleted {name}") + return True deadline = time.monotonic() + self.daemon.timeout attempted = False last_error = "" @@ -2645,7 +2763,10 @@ class ManagedSyncPair: 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: + if queue_settled( + last, + lambda: json.loads(self.command("sync", "queue", json_output=True))["items"], + ): return time.sleep(1) raise TimeoutError(f"daemon mutation queue did not drain: {last}") @@ -3158,7 +3279,7 @@ def run_move_contract( except KeyboardInterrupt: raise except BaseException as error: # noqa: BLE001 - reported, then continued - detail = f"{type(error).__name__}: {error}" + detail = failure_detail(error) print(f" FAIL {detail}") run.results.append(Result(case.name, label, "fail", time.monotonic() - started, detail)) if fail_fast: