fix(fuse): keep an upload pending until its bytes are cached (B119)

The drain retired an upload's op first, and only then read the node back
and linked the staged blob into the content cache. For that one round
trip, the status reported an empty queue for a file that wasn't cached
yet, so anything that waited for an empty queue before reading the file
could still need the network. The account run of 2026-10-03 hit it over
a slow link: "regression B119" failed in the same second the upload
landed.

`Core::landing_uploads` now counts an upload from just before its op is
retired until its bytes are cached. The status adds it to
`pending_uploads`, and reads the queue before the count, so it always
sees the upload in one of them. A drop guard uncounts it on every exit
path.

The op still goes first. An op restored after a crash would re-send a
revision this daemon hasn't recorded as its own yet, and
`revision_conflict` would keep it as a conflict copy of the user's file.

crates/pdfs-fuse/src/control.rs

@@ -315,6 +315,9 @@ impl Drop for RefreshGuard {
 pub(crate) fn status_response(core: &Core, username: &str, mountpoint: &Path) -> CtlResponse {
     let pins = core.cache.list_pins();
     let queued = core.db.pending_op_counts().unwrap_or_default();
+    // Read after the queue: the drain counts an upload as landing before it
+    // retires the op, so one of the two always has it (B119).
+    let landing = core.landing_uploads.load(Ordering::SeqCst);
     let staging = core.cache.staging_usage();
     CtlResponse::Status {
         parked_uploads: queued.parked.max(0) as u64,
@@ -331,7 +334,7 @@ pub(crate) fn status_response(core: &Core, username: &str, mountpoint: &Path) ->
         budget: core.cache.budget(),
         pins,
         online: core.online.load(Ordering::Relaxed),
-        pending_uploads: queued.uploads.max(0) as u64,
+        pending_uploads: queued.uploads.max(0) as u64 + landing,
         pending_changes: queued.changes.max(0) as u64,
     }
 }

crates/pdfs-fuse/src/drain.rs

@@ -24,7 +24,7 @@ use std::collections::HashMap;
 use std::fs::File;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
 use parking_lot::Mutex;
@@ -113,6 +113,23 @@ impl Drop for UploadCancel {
     }
 }
 
+/// One upload counted in [`Core::landing_uploads`] from its op's retirement
+/// until its bytes are cached, and uncounted when dropped, early return or not.
+struct LandingUpload(Arc<AtomicU64>);
+
+impl LandingUpload {
+    fn begin(count: &Arc<AtomicU64>) -> Self {
+        count.fetch_add(1, Ordering::SeqCst);
+        Self(count.clone())
+    }
+}
+
+impl Drop for LandingUpload {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
 /// The debounce arithmetic of [`Core::revision_debounce`], without the map:
 /// the last measured upload time for this node, bounded on both sides.
 ///
@@ -1635,6 +1652,14 @@ impl Core {
         // is what the next write's debounce is scaled from.
         self.record_upload_time(&uid, took);
 
+        // Still pending to whoever asks until the bytes are cached below, and
+        // counted before the op goes so the status never reads empty in between.
+        // An empty queue is how a script or the app knows a written file reads
+        // without the network; retiring the op alone left the read-back's round
+        // trip in which it did not (B119). The op itself has to go first: one
+        // restored after a crash would re-send a revision this daemon has not
+        // yet recorded as its own, and come back as a conflict copy.
+        let _landing = LandingUpload::begin(&self.landing_uploads);
         // Retire the op before dropping the blob: a crash between the two leaves
         // an orphaned file (harmless), whereas the reverse would leave a queued
         // op pointing at nothing.
@@ -2337,6 +2362,22 @@ mod tests {
         assert!(keeps_landed_upload(false, 4 * GIB, 0));
     }
 
+    #[test]
+    fn a_landing_upload_counts_until_its_guard_goes() {
+        let count = Arc::new(AtomicU64::new(0));
+        let first = LandingUpload::begin(&count);
+        let landed = || -> Result<(), ()> {
+            let _second = LandingUpload::begin(&count);
+            assert_eq!(count.load(Ordering::SeqCst), 2);
+            Err(())
+        };
+        // Uncounted on an early return too, or the queue never reads empty.
+        assert!(landed().is_err());
+        assert_eq!(count.load(Ordering::SeqCst), 1);
+        drop(first);
+        assert_eq!(count.load(Ordering::SeqCst), 0);
+    }
+
     fn baseline(mtime: i64, size: u64, rev: Option<&str>) -> Baseline {
         Baseline {
             mtime,

crates/pdfs-fuse/src/lib.rs

@@ -575,6 +575,12 @@ struct Core {
     /// replaces. Read by the upload's own reader ([`CountingReader`]), which is
     /// the one thing the SDK calls often enough to notice.
     upload_cancel: Arc<Mutex<HashMap<NodeUid, Arc<AtomicBool>>>>,
+    /// Uploads whose op is retired but whose bytes the content cache has not
+    /// adopted yet, because the revision just sealed is still being read back.
+    /// The status counts them as pending, so an empty queue means every file
+    /// written through the mount reads without the network (`docs/BUGS.md`
+    /// B119).
+    landing_uploads: Arc<AtomicU64>,
     /// True while a background refresh of the photos timeline (resp. the trash) is
     /// already running, so a burst of page requests against a stale listing kicks
     /// off one refresh rather than one per request.
@@ -8624,6 +8630,24 @@ mod tests {
         assert_before(upload, "self.db.op_exists(op.id)", "File::open(&blob)");
     }
 
+    #[test]
+    fn a_landed_upload_stays_pending_until_its_bytes_are_cached() {
+        let drain = function_source(include_str!("drain.rs"), "pub(crate) fn drain_revision(");
+        assert_before(
+            drain,
+            "let _landing = LandingUpload::begin(&self.landing_uploads)",
+            "self.db.delete_op(op.id)",
+        );
+        assert_before(drain, "self.db.delete_op(op.id)", ".store_file(&uid,");
+        let status = function_source(include_str!("control.rs"), "fn status_response(");
+        assert_before(
+            status,
+            "core.db.pending_op_counts()",
+            "core.landing_uploads.load(",
+        );
+        assert!(status.contains("queued.uploads.max(0) as u64 + landing"));
+    }
+
     #[test]
     fn replacement_rename_authorizes_source_and_both_parents() {
         use std::cell::RefCell;

crates/pdfs-fuse/src/mount.rs

@@ -391,6 +391,7 @@ pub fn mount(
         shutdown: Arc::new(crate::shutdown::Shutdown::default()),
         upload_times: Arc::new(Mutex::new(HashMap::new())),
         upload_cancel: Arc::new(Mutex::new(HashMap::new())),
+        landing_uploads: Arc::new(AtomicU64::new(0)),
         timeline_refreshing: Arc::new(AtomicBool::new(false)),
         timeline_progress: Arc::new(Mutex::new(None)),
         similar_hashing: Arc::new(AtomicBool::new(false)),