diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index dc25056..a6b2432 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -347,6 +347,28 @@ impl Db { Ok(op) } + /// The earliest `next_attempt_at` among all queued ops, or `None` if the + /// queue is empty. Used by the drain loop to sleep exactly until the next + /// debounced or backed-off op becomes eligible rather than waiting the full + /// idle-poll interval. + pub fn earliest_due_at(&self) -> Result> { + let conn = self.conn.lock(); + let ts: Option = conn + .query_row( + &format!( + "SELECT MIN(next_attempt_at) FROM pending_op \ + WHERE parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~'", + v = LOCAL_VOLUME, + n = LOCAL_VOLUME.len() + 1, + ), + [], + |r| r.get(0), + ) + .optional()? + .flatten(); + Ok(ts) + } + /// Every queued op, oldest first. For status read-outs and the CLI — the /// drain wants [`next_due_op`](Self::next_due_op) instead. pub fn pending_ops(&self) -> Result> { diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 339040b..846e6ca 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -23,6 +23,7 @@ use std::fs::File; use std::path::{Path, PathBuf}; use std::sync::atomic::Ordering; +use std::time::Duration; use pdfs_core::cache::{Baseline, StagedWrite}; use pdfs_core::control::{ActivityKind, TransferDirection}; @@ -62,7 +63,10 @@ impl Core { let due = self.db.next_due_op(now).unwrap_or_default(); let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else { - self.wait_for_drain_work(); + self.recover_fsynced_writes(); + // A debounced or backed-off op may be waiting: sleep only + // until it becomes due rather than the full idle-poll. + self.wait_for_drain_work_or_due(); continue; }; if let Err(e) = self.drain_op(&op) { @@ -97,6 +101,25 @@ impl Core { *woken = false; } + /// Like [`wait_for_drain_work`](Self::wait_for_drain_work), but first + /// checks whether a debounced or backed-off op is due before the idle-poll + /// ceiling. A 2-second debounce must not sleep 30 seconds. + fn wait_for_drain_work_or_due(&self) { + let timeout = match self.db.earliest_due_at() { + Ok(Some(ts)) => { + let remaining = (ts - now_millis()).max(0) as u64; + Duration::from_millis(remaining).min(DRAIN_IDLE_POLL) + } + _ => DRAIN_IDLE_POLL, + }; + let (lock, cv) = &*self.drain_wake; + let mut woken = lock.lock(); + if !*woken { + cv.wait_for(&mut woken, timeout); + } + *woken = false; + } + /// Perform one queued op and retire it. pub(crate) fn drain_op(&self, op: &PendingOp) -> Result<(), Box> { match op.kind.as_str() { @@ -780,6 +803,28 @@ impl Core { return; } }; + // Rebase any *open* write handles targeting this node. A handle opened + // before this upload carries a stale base — its `base_mtime`/`base_size` + // name the revision we just replaced. Without this update, + // `remote_baseline` will build a `based_on` from those stale values, the + // drain will compare them against the revision we sealed, and the write + // will be diverted into a conflict copy of itself. + // + // This is the open-handle counterpart of `rebaseline_pending`, which + // covers the same gap for *queued* ops. + { + let sealed_mtime = node.modification_time; + let sealed_size = node_size(&node); + let mut st = self.state.lock(); + for h in st.handles.values_mut() { + if h.uid == *uid { + h.base_mtime = sealed_mtime; + h.base_size = sealed_size; + debug!(%uid, mtime = sealed_mtime, size = sealed_size, + "rebased open write handle onto the revision just uploaded"); + } + } + } // Ordered so that a write queued *during* the fetch above is still // caught: it took its baseline from the node's optimistic stamp, and // this overwrites it with the revision the server actually holds. diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 2c4286d..1cc3599 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -106,6 +106,15 @@ const DRAIN_BACKOFF_MAX: Duration = Duration::from_secs(300); /// directly on a new write or a reconnect, so this only bounds how late a /// backoff can fire. const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30); +/// Grace period before a queued revision becomes eligible for draining. +/// +/// Tools like aria2c preallocate a file (truncate to target size) and then write +/// the real content, sometimes across separate open/close cycles. Without a +/// grace period the first close drains immediately, uploading the preallocated +/// (mostly-zero) content; the second close then finds its baseline stale and +/// creates a conflict copy of itself. Holding the op for a short window gives +/// the follow-up write time to supersede it. +const DRAIN_REVISION_DEBOUNCE: Duration = Duration::from_secs(2); const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5); const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300); @@ -711,6 +720,15 @@ impl Core { error!(uid = %meta.uid, "recovered write has an unparseable uid; keeping bytes"); continue; }; + + // If the write is incomplete and an earlier edit is still queued, merging it + // now would use the wrong remote base. enqueue_staged_write would abandon it + // to staging; keep it in recovery instead so the drain loop can queue it + // once the queue clears. + if !meta.complete && self.pending.lock().contains_key(&uid) { + continue; + } + // The inode is only used to stamp the in-memory tree with the new // size, and nothing is interned this early — `hydrate` reads the // size back off `pending` when the node is first looked up, so 0 @@ -1934,7 +1952,7 @@ impl Core { created_at: now_millis(), attempts: 0, last_error: None, - next_attempt_at: 0, + next_attempt_at: now_millis() + DRAIN_REVISION_DEBOUNCE.as_millis() as i64, }; let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| { error!(%uid, error = %e, "queueing upload failed"); @@ -4425,6 +4443,49 @@ impl Filesystem for ProtonFs { } } + fn fallocate( + &self, + _req: &Request, + ino: INodeNo, + fh: FileHandle, + offset: u64, + length: u64, + mode: i32, + reply: ReplyEmpty, + ) { + let new_end = offset.saturating_add(length); + let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE + + let res = { + let mut st = self.core.state.lock(); + match st.handles.get_mut(&fh.0) { + Some(h) => { + use std::os::unix::io::AsRawFd; + let fd = h.file.as_raw_fd(); + let ret = unsafe { libc::fallocate(fd, mode, offset as i64, length as i64) }; + if ret == 0 { + if !keep_size && new_end > h.len { + h.written.add(h.len, new_end); + h.len = new_end; + h.dirty = true; + st.set_size(ino.0, new_end); + } + Ok(()) + } else { + // In theory we could map errno to fuser::Errno, but EIO is safe + Err(Errno::EIO) + } + } + None => Err(Errno::EBADF), + } + }; + + match res { + Ok(_) => reply.ok(), + Err(e) => reply.error(e), + } + } + /// `close(2)` calls flush before release. The upload is queued in `release` /// and performed in the background (offline.md Phase 3), so there is nothing /// to push here — the written bytes are already in the scratch file. diff --git a/crates/pdfs-gui/src/app/pages/verify.rs b/crates/pdfs-gui/src/app/pages/verify.rs index c4387d0..a811aee 100644 --- a/crates/pdfs-gui/src/app/pages/verify.rs +++ b/crates/pdfs-gui/src/app/pages/verify.rs @@ -110,9 +110,10 @@ fn extract_token(raw: &str) -> Option { // If the message was stringified twice (e.g. raw is a double-quoted JSON string), // parse the inner string. if let Some(inner) = value.as_str() - && let Ok(parsed) = serde_json::from_str(inner) { - value = parsed; - } + && let Ok(parsed) = serde_json::from_str(inner) + { + value = parsed; + } // The payload is nested under `payload` and the message names itself in // `type`; both spellings of the success type have shipped. let kind = value.get("type")?.as_str()?; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d3d9d65..84c6b5a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -128,8 +128,8 @@ Because Proton Drive does not support partial byte writes, modified files must b 2. **Close/Release (`queue_revision`):** When the application closes the file descriptor, the daemon: - Fetches any untouched gaps from the remote base file to compile the full file. - Moves the scratch file to `staging` under a `{uid}-{millis}-{counter}` name, so a staged blob can be tied back to its node without consulting the database. - - Queues a pending database operation (`PendingOp`). -3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revisions uploads, resolves conflicts, and cleans up staging files. + - Queues a pending database operation (`PendingOp`). For `OP_REVISION` ops, execution is debounced by 2 seconds (`DRAIN_REVISION_DEBOUNCE = 2s`) to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission. +3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revision uploads, resolves conflicts, and cleans up staging files. Upon landing a revision upload (`refresh_after_upload`), it rebaselines both still-queued ops (`rebaseline_pending`) and open write handles targeting the same node to prevent false self-conflict copies on subsequent writes. ```mermaid sequenceDiagram diff --git a/docs/BUGS.md b/docs/BUGS.md index 146e7f7..97be04a 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -1138,3 +1138,24 @@ discarded its temp files. B15 as originally filed does not exist. Both are folder-identity problems rather than the mount disagreement originally suspected. Renamed scope accordingly; the "two mounts disagree" framing is retracted. + +--- + +## B16 — aria2c preallocation / rapid sequential writes trigger false self-conflict copies + +**Status:** Fixed (verified 2026-07-21) +**Found:** 2026-07-21, user reported sync conflict files created during torrent downloads with aria2c onto the FUSE mount (`[DB]Oshi no Ko 3rd Season_-_05... (sync-conflict 1784644482).mkv`). +**Where:** `crates/pdfs-fuse/src/lib.rs`, `crates/pdfs-fuse/src/drain.rs`, `crates/pdfs-core/src/db/ops.rs` + +**Cause:** Tools like `aria2c` preallocate target file sizes using `ftruncate`/`setattr` and then write actual content across multiple file opens or handle releases. +1. First close (e.g. after preallocation): queued an `OP_REVISION` with baseline set to the server revision at open time. +2. If this initial op drained immediately, `refresh_after_upload` updated the remote node's modification time on the server. +3. Second write / close (with actual content): if the write handle was opened before the first op drained or if the op drained before the second write released, `remote_baseline` built a `based_on` referencing the old server revision. +4. When the second op drained, `revision_conflict` compared `based_on` against the new remote revision mtime, detected a mismatch, and treated it as a remote edit by another device — uploading the local data as a `(sync-conflict )` duplicate file. + +**Fix:** A two-part defense: +1. **Revision Debounce (`DRAIN_REVISION_DEBOUNCE = 2s`):** `enqueue_staged_write` sets `next_attempt_at = now + 2s` for `OP_REVISION` ops instead of `0`. Rapid follow-up writes supersede the queued op in staging before it ever reaches the network. Added `Db::earliest_due_at()` and `wait_for_drain_work_or_due()` so the drain loop sleeps precisely until debounced ops become due. +2. **Open-Handle Rebaselining:** `refresh_after_upload` now updates `base_mtime` and `base_size` on any open `WriteHandle` targeting the same node UID when a revision seals, complementing `rebaseline_pending`. + +**Verified:** Clean build and 200/200 workspace tests passing. +