fix(fuse,core,cli): resume a session another process refreshed

`pdfs unlock` resumes the daemon's session and may refresh it. The
daemon then holds a spent refresh token, and since B202 a refusal of it
read as an expired session although the keyring held a live one.

Before calling a refused session expired, compare the keyring with what
this process last stored. When it holds a newer rotation or another
login, end the mount with `MountOutcome::ResumeStored` and mount again
from the keyring. A rotation another process already wrote counts as
this session's, so the daemon keeps writing its later ones (B203).

crates/pdfs-cli/src/main.rs

@@ -2193,6 +2193,9 @@ fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
                 tracing::info!("daemon stopping");
                 return Ok(());
             }
+            Ok(pdfs_fuse::MountOutcome::ResumeStored) => {
+                tracing::info!("resuming the session stored in the keyring");
+            }
             Ok(pdfs_fuse::MountOutcome::Unmounted) => {
                 tracing::warn!("mount ended externally; remounting in 2s");
                 pdfs_fuse::idle_sleep(std::time::Duration::from_secs(2));

crates/pdfs-core/src/auth.rs

@@ -687,6 +687,28 @@ pub fn persist(session: &ProtonApiSession) -> Result<()> {
     }
 }
 
+/// Whether the keyring holds a session newer than any this process runs on:
+/// another login, or one of this process's sessions rotated by another
+/// process (`pdfs unlock` resumes the same login and may refresh it).
+///
+/// The refresh token this process holds is spent then, and Proton refuses it
+/// as if the session were gone. A daemon asks this before it reports the
+/// session expired, and resumes the stored one instead.
+pub fn stored_session_is_newer() -> bool {
+    // Nothing to compare with, and no keyring to wake up for it.
+    if TOKEN_SINKS.lock().is_empty() {
+        return false;
+    }
+    let Ok(stored) = load() else {
+        return false;
+    };
+    let sinks = TOKEN_SINKS.lock();
+    match sinks.get(&stored.session_id) {
+        Some(sink) => sink.state.lock().stored_refresh != stored.refresh_token,
+        None => !sinks.is_empty(),
+    }
+}
+
 /// Where each live session in this process writes its rotated tokens, by
 /// session id, for [`persist`] to find.
 static TOKEN_SINKS: parking_lot::Mutex<BTreeMap<String, Arc<TokenSink>>> =
@@ -737,14 +759,17 @@ enum Rotation {
 }
 
 /// Put `tokens` into `stored` if it is still this session's blob: the same
-/// session id, holding the refresh token this session last left in it.
+/// session id, holding the refresh token this session last left in it, or
+/// already holding this very rotation (another process wrote it first).
 fn take_rotation(
     stored: &mut StoredSession,
     session_id: &str,
     stored_refresh: &str,
     tokens: &Tokens,
 ) -> Rotation {
-    if stored.session_id != session_id || stored.refresh_token != stored_refresh {
+    if stored.session_id != session_id
+        || (stored.refresh_token != stored_refresh && stored.refresh_token != tokens.refresh_token)
+    {
         return Rotation::MovedOn;
     }
     stored.access_token = tokens.access_token.clone();
@@ -1011,6 +1036,16 @@ mod tests {
         assert_eq!(stored.refresh_token, "r");
     }
 
+    /// `pdfs unlock` wrote this very rotation first, from the same session.
+    #[test]
+    fn a_rotation_already_in_the_blob_is_taken() {
+        let mut stored = stored_with("r2");
+        assert_eq!(
+            take_rotation(&mut stored, "s", "r", &tokens("a2", "r2")),
+            Rotation::Taken
+        );
+    }
+
     /// The keyring is one store per process, so the tests that use it,
     /// through `keyring-core`'s in-memory store, take turns.
     fn mock_keyring() -> parking_lot::MutexGuard<'static, ()> {
@@ -1046,6 +1081,27 @@ mod tests {
         assert_eq!(now.refresh_token, "fresh");
     }
 
+    /// `pdfs unlock` rotated the daemon's session behind its back: the
+    /// daemon's refresh token is spent, and the keyring holds the live one.
+    #[test]
+    fn a_session_rotated_by_another_process_is_newer_in_the_keyring() {
+        let _turn = mock_keyring();
+        write_stored(&stored_with("r")).unwrap();
+        TOKEN_SINKS
+            .lock()
+            .insert("s".into(), Arc::new(TokenSink::new("s", "r".into())));
+        assert!(!stored_session_is_newer());
+
+        write_stored(&stored_with("r2")).unwrap();
+        assert!(stored_session_is_newer());
+
+        let mut login = stored_with("fresh");
+        login.session_id = "new".into();
+        write_stored(&login).unwrap();
+        assert!(stored_session_is_newer());
+        TOKEN_SINKS.lock().remove("s");
+    }
+
     /// B61: a rotation the keyring refused lived only in memory.
     #[test]
     fn a_rotation_the_keyring_refused_is_written_once_it_takes_it() {

crates/pdfs-fuse/src/link.rs

@@ -60,6 +60,10 @@ pub(crate) struct Link {
     /// that restarts the daemon. Read by front-ends through `Response::Status`,
     /// so they can ask the user to sign in again rather than say "offline".
     pub(crate) session_expired: AtomicBool,
+    /// Proton refused the session, but only because another process rotated
+    /// it (`pdfs unlock`): the keyring holds the live tokens. The mount loop
+    /// ends the mount for the daemon to mount again from the keyring.
+    pub(crate) resume_stored: AtomicBool,
 }
 
 impl Link {
@@ -202,7 +206,19 @@ impl Core {
     /// Flip the mount offline for good: Proton refused the session, and only a
     /// new login brings it back. Changes stay queued for the daemon that login
     /// starts. Only the first caller logs it.
+    ///
+    /// Unless the keyring holds a newer session than the refused one: then
+    /// the mount ends, for the daemon to mount again on that one.
     pub(crate) fn mark_session_expired(&self, what: &str, error: &dyn std::fmt::Display) {
+        if self.link.resume_stored.load(Ordering::Relaxed) {
+            return self.mark_offline(what, error);
+        }
+        if pdfs_core::auth::stored_session_is_newer() {
+            info!(during = what, %error,
+                  "Proton refused a session another process has since refreshed; resuming it");
+            self.link.resume_stored.store(true, Ordering::Relaxed);
+            return self.mark_offline(what, error);
+        }
         if !self.link.session_expired.swap(true, Ordering::Relaxed) {
             warn!(during = what, %error,
                   "Proton refused the session; run `pdfs login` to sign in again");

crates/pdfs-fuse/src/mount.rs

@@ -12,6 +12,10 @@ pub enum MountOutcome {
     /// The kernel mount ended on its own (e.g. an external `fusermount -u`).
     /// The caller may want to remount.
     Unmounted,
+    /// Proton refused the session, and the keyring holds a newer one: another
+    /// process (`pdfs unlock`) refreshed it, or a new login replaced it. We
+    /// unmounted ourselves; the caller should mount again, which resumes it.
+    ResumeStored,
 }
 
 /// Whether `path` is a mountpoint whose FUSE connection is dead — the state a
@@ -227,6 +231,11 @@ fn fetch_or_recall_root(
     } else {
         Start::Offline
     };
+    if start == Start::SessionExpired && pdfs_core::auth::stored_session_is_newer() {
+        return Err(std::io::Error::other(format!(
+            "Proton refused a session another process has since refreshed ({err}); resuming it"
+        )));
+    }
     let cached = db
         .state_str(ROOT_UID_KEY)
         .ok()
@@ -745,6 +754,18 @@ pub(crate) fn mount_with(
                 }
                 break MountOutcome::Shutdown;
             }
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout)
+                if core.link.resume_stored.load(Ordering::Relaxed) =>
+            {
+                info!("unmounting to resume the stored session");
+                core.mounts.lock().close();
+                if let Err(e) = teardown_session(&core.session_live, || {
+                    umount_session_unblocked(bg, main_conn)
+                }) {
+                    warn!(error = %e, "umount_and_join failed");
+                }
+                break MountOutcome::ResumeStored;
+            }
             Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                 if bg.guard.is_finished() {
                     info!("mount ended externally");