proton-drive-linux/crates/pdfs-fuse/src/mount.rs

//! Mount lifecycle and background-service orchestration.

use super::*;

/// Why a [`mount`] call returned. Lets the daemon decide whether to exit (clean
/// shutdown) or remount (the mount went away under it).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MountOutcome {
    /// The daemon was asked to stop (SIGTERM/SIGINT) and we lazily unmounted
    /// ourselves. The caller should exit cleanly.
    Shutdown,
    /// 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
/// daemon killed before it could unmount leaves behind.
///
/// The kernel answers every operation on such a path with `ENOTCONN`, which the
/// ordinary existence checks (`is_dir`, `exists`) report as plain `false`: the
/// path looks *absent* rather than broken. Callers that would otherwise treat
/// that as "nothing to mount here" use this to tell the two apart.
pub(crate) fn is_stale_mount(path: &Path) -> bool {
    matches!(
        std::fs::metadata(path).map_err(|e| e.raw_os_error()),
        Err(Some(libc::ENOTCONN))
    )
}

/// A secondary (on-demand sync folder) FUSE session and the exact liveness flag
/// published with its forked inode state.
///
/// Keeping these together makes every teardown mark the fork unroutable before
/// unmounting, including paths that can block on an in-flight FUSE request.
pub(super) struct SecondaryMount {
    session: BackgroundSession,
    conn: Option<u32>,
    session_live: Arc<AtomicBool>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SecondaryInsertRejection {
    Closed,
    Duplicate,
}

/// Secondary sessions and the daemon-wide gate controlling whether newly
/// spawned sessions may become visible. The gate and map share one lock so
/// shutdown cannot race an insertion between checking and publishing.
pub(super) struct SecondaryMountRegistry<T> {
    accepting: bool,
    entries: HashMap<i64, T>,
}

impl<T> Default for SecondaryMountRegistry<T> {
    fn default() -> Self {
        Self {
            accepting: false,
            entries: HashMap::new(),
        }
    }
}

impl<T> SecondaryMountRegistry<T> {
    pub(super) fn open(&mut self) {
        self.accepting = true;
    }

    pub(super) fn is_accepting(&self) -> bool {
        self.accepting
    }

    pub(super) fn close(&mut self) {
        self.accepting = false;
    }

    pub(super) fn insert(
        &mut self,
        id: i64,
        mount: T,
    ) -> Result<(), (SecondaryInsertRejection, T)> {
        if !self.accepting {
            return Err((SecondaryInsertRejection::Closed, mount));
        }
        if self.entries.contains_key(&id) {
            return Err((SecondaryInsertRejection::Duplicate, mount));
        }
        self.entries.insert(id, mount);
        Ok(())
    }

    pub(super) fn remove(&mut self, id: &i64) -> Option<T> {
        self.entries.remove(id)
    }

    pub(super) fn contains_key(&self, id: &i64) -> bool {
        self.entries.contains_key(id)
    }

    pub(super) fn close_and_drain(&mut self) -> Vec<(i64, T)> {
        self.close();
        self.entries.drain().collect()
    }
}

impl SecondaryMount {
    pub(super) fn new(
        session: BackgroundSession,
        conn: Option<u32>,
        session_live: Arc<AtomicBool>,
    ) -> Self {
        Self {
            session,
            conn,
            session_live,
        }
    }

    pub(super) fn teardown(self) -> std::io::Result<()> {
        teardown_session(&self.session_live, || {
            umount_session_unblocked(self.session, self.conn)
        })
    }
}

/// The kernel's id for the FUSE connection backing the mount at `mountpoint` —
/// the directory name under `/sys/fs/fuse/connections`, which is the minor
/// number of the mountpoint's device. Must be read *while still mounted*; after
/// unmount the path resolves to the underlying directory on another device.
/// `None` when the path can't be stat'd.
pub(super) fn fuse_connection_id(mountpoint: &Path) -> Option<u32> {
    use std::os::unix::fs::MetadataExt;
    let dev = std::fs::metadata(mountpoint).ok()?.dev();
    Some(libc::minor(dev))
}

/// Force the kernel to abort FUSE connection `id`, erroring every in-flight
/// request.
///
/// On a stop signal we unmount lazily (`MNT_DETACH`) so the call succeeds even
/// mid-transfer — but detach only removes the mountpoint; it does *not* end the
/// connection while a request is still in flight. fuser's session loop then
/// blocks on `/dev/fuse` waiting for a next request that never comes, so `join`
/// hangs — long enough during a transfer that systemd's stop timeout SIGKILLs
/// the daemon mid-unmount, stranding the on-demand mounts as dead endpoints.
/// Writing the connection's `abort` file makes the pending reads fail with
/// `ENODEV`, so the loop returns and `join` completes at once. Best-effort:
/// there is nothing more to do on the shutdown path if it fails.
pub(super) fn abort_fuse_connection(id: u32) {
    let path = format!("/sys/fs/fuse/connections/{id}/abort");
    if std::fs::write(&path, b"1").is_ok() {
        info!(id, "aborted FUSE connection to unblock unmount");
    }
}

/// Unmount a background session that may be mid-transfer without wedging: abort
/// its connection first (so the session loop exits promptly), then lazily
/// unmount and join. `conn` is the id captured at mount time; `None` skips the
/// abort and just unmounts (a healthy idle mount joins on its own `Destroy`).
pub(super) fn umount_session_unblocked(
    session: BackgroundSession,
    conn: Option<u32>,
) -> std::io::Result<()> {
    if let Some(id) = conn {
        abort_fuse_connection(id);
    }
    session.umount_and_join()
}

/// Best-effort teardown of a stale mount left behind by a crashed daemon. A
/// previous run that died without unmounting leaves the kernel mount in place,
/// so the fresh `Session::new` below would fail with EBUSY ("Device or resource
/// busy"). `fusermount3 -u -z` is the lazy (detach) unmount, which succeeds even
/// when the old mount is still busy. Swallow all output/errors: if there is no
/// stale mount this is simply a no-op.
pub(super) fn clear_stale_mount(mountpoint: &Path) {
    for bin in ["fusermount3", "fusermount"] {
        let ok = std::process::Command::new(bin)
            .arg("-u")
            .arg("-z")
            .arg(mountpoint)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .map(|s| s.success())
            .unwrap_or(false);
        if ok {
            info!(mountpoint = %mountpoint.display(), "cleared stale mount before remount");
            return;
        }
    }
}

/// Make sure `mountpoint` is a directory a mount can go on, clearing a dead
/// mount a killed daemon left there first.
///
/// A dead mount answers `ENOTCONN`, which `create_dir_all` takes for "absent";
/// it then tries `mkdir` and fails with `EEXIST`. The daemon retried that error
/// forever and never got as far as [`check_mountpoint`], which would have
/// cleared the mount (`docs/BUGS.md` B279).
pub fn prepare_mountpoint(mountpoint: &Path) -> std::io::Result<()> {
    if is_stale_mount(mountpoint) {
        clear_stale_mount(mountpoint);
    }
    std::fs::create_dir_all(mountpoint)
}

/// Refuse to mount My files over a folder that already has files in it: FUSE
/// would hide them, unsent and out of sight, until the service stops
/// (`docs/BUGS.md` B277). Runs before the root fetch and before any worker
/// starts, so a refusal costs nothing to retry.
///
/// A dead mount is cleared first, since it cannot be listed. One of this
/// client's own live mounts is left to [`spawn_session`], which clears it as it
/// always has; listing it would show Drive's files, not the folder's.
pub(super) fn check_mountpoint(mountpoint: &Path) -> std::io::Result<()> {
    if is_stale_mount(mountpoint) {
        clear_stale_mount(mountpoint);
    }
    if pdfs_core::mountpoint::mounted_fs_type(mountpoint).as_deref()
        == Some(pdfs_core::mountpoint::OWN_FS_TYPE)
    {
        return Ok(());
    }
    match pdfs_core::mountpoint::not_empty(mountpoint)? {
        Some(found) => Err(std::io::Error::other(found)),
        None => Ok(()),
    }
}

/// `sync_state` key holding the uid of the My Files root, so a later run can
/// recover it from `nodes` without the API.
const ROOT_UID_KEY: &str = "root_uid";

/// The My Files root, and whether we got it from the API (`true`) or from the
/// cache because the API was unreachable (`false`).
///
/// A successful fetch also records the root's uid, which is what makes the
/// fallback possible on a later run: `nodes` is keyed by uid, so without this we
/// would have the root's row on disk and no way to tell which one it is.
///
/// Failing to fetch is only fatal on a genuinely cold start — no cached root
/// means an empty tree, and mounting that would show the user an empty Drive
/// rather than an honest error.
fn fetch_or_recall_root(
    drive: &dyn DriveApi,
    rt: &tokio::runtime::Handle,
    db: &Db,
) -> std::io::Result<(Node, Start)> {
    let err = match rt.block_on(drive.get_my_files_folder()) {
        Ok(root) => {
            if let Err(e) = db.set_state_str(ROOT_UID_KEY, &root.uid.to_string()) {
                warn!(error = %e, "persist root uid failed");
            }
            return Ok((root, Start::Online));
        }
        Err(e) => e,
    };
    // Still mounted from the cache, so cached files stay readable and changes
    // queue for the daemon the next login starts.
    let start = if pdfs_core::error::session_revoked(&err) {
        Start::SessionExpired
    } 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()
        .flatten()
        .and_then(|uid| db.node_by_uid(&uid).ok().flatten());
    match cached {
        Some(root) if start == Start::SessionExpired => {
            warn!(error = %err,
                  "Proton refused the session; mounting from cache until `pdfs login`");
            Ok((root, start))
        }
        Some(root) => {
            warn!(error = %err, "fetch My Files root failed; mounting from cache (offline)");
            Ok((root, start))
        }
        None if start == Start::SessionExpired => Err(std::io::Error::other(format!(
            "Proton refused the session ({err}); run `pdfs login`"
        ))),
        None => Err(std::io::Error::other(format!("fetch My Files root: {err}"))),
    }
}

/// How a mount comes up, from what fetching its root said.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Start {
    Online,
    /// Proton could not be reached: serve the cache and probe for the link.
    Offline,
    /// Proton refused the session: serve the cache until a new login.
    SessionExpired,
}

/// Per-mount settings resolved from [`pdfs_core::config::AppConfig`] by the
/// caller, so `mount` itself never reads config or the environment.
pub struct MountOptions {
    /// Proton account the session belongs to, for display and per-user paths.
    pub username: String,
    /// What the background conflict sweep may do. See `docs/BUGS.md` B71.
    pub sweep_mode: SweepMode,
    /// Upload and download caps in bytes per second (`0` = no cap).
    pub upload_limit: u64,
    pub download_limit: u64,
    /// Namespace changes through the mount are queued, not sent in the
    /// syscall. See [`pdfs_core::config::AppConfig::local_first`].
    pub local_first: bool,
}

/// Spawn one FUSE session rooted at an arbitrary remote node.
///
/// Primary My Files and secondary on-demand device locations use this same
/// construction path so mount options, stale-mount recovery, registry
/// liveness, and notifier ownership cannot drift apart. Each caller owns its
/// state/path registration and performs it exactly once before calling here.
pub(super) fn spawn_session(
    core: &Core,
    mountpoint: &Path,
    root: Node,
) -> std::io::Result<BackgroundSession> {
    let mut config = Config::default();
    // `n_threads` stays at fuser's default of 1 (with `clone_fd` off) on
    // purpose, for now. A concurrent dispatch loop is worth having, but
    // `release` currently relies on being serialised against the `open`+`read`
    // an application issues right after `close(2)` — see the comment on
    // `Filesystem::release`. Raising it before that ordering is made explicit
    // (a per-node staging-in-flight barrier) reintroduces the acceptance
    // suite's "concurrent file 1 mismatch". Kernel-side caps that do *not*
    // depend on that ordering are negotiated in `ProtonFs::init`.
    config.mount_options = vec![
        MountOption::FSName("protondrive".to_string()),
        MountOption::Subtype("protondrive".to_string()),
        MountOption::DefaultPermissions,
    ];

    clear_stale_mount(mountpoint);
    info!(mountpoint = %mountpoint.display(), "mounting Proton Drive location");
    let fs = ProtonFs::new(core.clone(), root);
    if core.primary {
        // Only My Files rehydrates the daemon-wide node cache, once. A
        // secondary mount shares the tree it fills.
        core.hydrate();
    }
    let session = Session::new(fs, mountpoint, &config)?.spawn()?;
    let _ = core
        .notifier
        .set(KernelChannel::new(session.notifier(), core.root_ino()));
    core.session_live.store(true, Ordering::Release);
    Ok(session)
}

/// Make a session unroutable before beginning any teardown that may block.
fn teardown_session<T>(session_live: &AtomicBool, teardown: impl FnOnce() -> T) -> T {
    session_live.store(false, Ordering::Release);
    teardown()
}

impl Core {
    /// Resolve the primary share id away from startup and publish it only if the
    /// projected root has not changed while the request was in flight.
    pub(crate) fn repair_primary_share_id(&self, root_uid: &NodeUid) {
        match self.rt.block_on(self.client.context_share_id(root_uid)) {
            Ok(share_id) => {
                match self
                    .db
                    .mount_repair_my_files_share_id(&root_uid.to_string(), &share_id.to_string())
                {
                    Ok(true) => debug!(%root_uid, "repaired My Files share id"),
                    Ok(false) => {
                        debug!(%root_uid, "discarded stale My Files share-id result");
                    }
                    Err(error) => {
                        warn!(%root_uid, error = %error, "persist My Files share id failed");
                    }
                }
            }
            Err(error) => {
                debug!(%root_uid, error = %error, "resolve My Files share id failed");
            }
        }
    }
}

/// Mount the filesystem at `mountpoint` and block until it is unmounted or the
/// daemon is asked to stop.
///
/// Resolves the My Files root up front — from the API, or from the cached tree
/// when the API is unreachable (`fetch_or_recall_root`) — then spawns the
/// Phase 2 event-sync task, the
/// Phase 4 control socket, and runs the FUSE session on its own thread while
/// this thread waits for either a stop signal (SIGTERM/SIGINT) or the kernel
/// mount ending on its own. On a stop signal we lazily unmount ourselves
/// (`umount_and_join`, the MNT_DETACH path that succeeds even while downloads
/// are in flight), so `systemctl --user stop` is always a clean teardown.
/// `rt` must be a handle to a *running* multi-threaded runtime.
pub fn mount(
    client: ProtonDriveClient,
    rt: tokio::runtime::Handle,
    mountpoint: &Path,
    cache: ContentCache,
    control_socket: &Path,
    db: Arc<Db>,
    options: MountOptions,
) -> std::io::Result<MountOutcome> {
    let drive = Arc::new(client.clone());
    mount_with(
        client,
        drive,
        rt,
        mountpoint,
        cache,
        control_socket,
        db,
        options,
        Host::Daemon,
    )
}

/// What a mount runs in.
pub(crate) enum Host {
    /// The daemon: stopped by SIGTERM or SIGINT, and indexing the user's home
    /// for the launcher prompt.
    Daemon,
    /// One of several simulated daemons in a test process (`sim`): stopped by
    /// a message on this channel, and with no home to index.
    #[cfg(test)]
    Simulation(std::sync::mpsc::Receiver<()>),
}

/// [`mount`], reaching Drive through `drive` and hosted by `host`; `client`
/// serves only what [`DriveApi`] does not cover.
#[allow(clippy::too_many_arguments)]
pub(crate) fn mount_with(
    client: ProtonDriveClient,
    drive: Arc<dyn DriveApi>,
    rt: tokio::runtime::Handle,
    mountpoint: &Path,
    cache: ContentCache,
    control_socket: &Path,
    db: Arc<Db>,
    options: MountOptions,
    host: Host,
) -> std::io::Result<MountOutcome> {
    let MountOptions {
        username,
        sweep_mode,
        upload_limit,
        download_limit,
        local_first,
    } = options;
    // Start the uptime clock here rather than at the first `pdfs diagnostics`,
    // so the age it reports is the daemon's own.
    super::diagnostics::start_clock();
    check_mountpoint(mountpoint)?;
    let (root, start) = fetch_or_recall_root(drive.as_ref(), &rt, &db)?;
    let online = start == Start::Online;
    let scope = root.tree_event_scope_id();
    db.mount_upsert_my_files(&mountpoint.to_string_lossy(), &root.uid.to_string(), None)
        .map_err(|error| std::io::Error::other(format!("project My Files mount: {error}")))?;
    let share_access = db
        .all_share_access()
        .map_err(|error| std::io::Error::other(format!("load shared access: {error}")))?;

    // The folder-sync engine runs on its own thread and is
    // nudged over this channel; the sender lives in Core so control-socket
    // handlers can trigger reconciles.
    let (sync_tx, sync_rx) = std::sync::mpsc::channel::<sync::SyncMsg>();

    let paused_until = pause::load_paused_until(&db);
    let core = Core {
        drive,
        client,
        rt: rt.clone(),
        maintenance: Arc::new(Mutex::new(Default::default())),
        primary_root_uid: root.uid.clone(),
        primary: true,
        state: Arc::new(Mutex::new(State::new(db.clone(), share_access))),
        root: Arc::new(OnceLock::new()),
        cache: Arc::new(cache),
        readers: Arc::new(Mutex::new(HashMap::new())),
        block_ring: Arc::new(Mutex::new(BlockRing::default())),
        block_flight: Arc::new(BlockFlight::default()),
        content_repairs: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
        prefetch: Arc::new(Prefetch::default()),
        prefetch_budget: Arc::new(tokio::sync::Semaphore::new(PREFETCH_BUDGET)),
        workers: Arc::new(Workers::new(FUSE_WORKERS)?),
        db,
        shared_publication: Arc::new(Mutex::new(())),
        shared_generation: Arc::new(AtomicU64::new(0)),
        shared_refresh_deadlines: Arc::new(Mutex::new(SharedRefreshDeadlines::default())),
        online: Arc::new(AtomicBool::new(online)),
        link: Arc::new(link::Link::starting(start == Start::SessionExpired)),
        sync_paused_until: Arc::new(AtomicI64::new(paused_until)),
        pending: Arc::new(Mutex::new(HashMap::new())),
        hidden: Arc::new(Mutex::new(HashSet::new())),
        creating: Arc::new(Mutex::new(HashMap::new())),
        sealed_unread: Arc::new(Mutex::new(HashMap::new())),
        landed: Arc::new(Mutex::new(HashMap::new())),
        unadopted: Arc::new(Mutex::new(HashMap::new())),
        drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
        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)),
        albums_refreshing: Arc::new(AtomicBool::new(false)),
        trash_refreshing: Arc::new(AtomicBool::new(false)),
        trash_progress: Arc::new(tokio::sync::Notify::new()),
        conflict_notified: Arc::new(Mutex::new(HashSet::new())),
        sweep_mode,
        local_first,
        self_changes: Arc::new(Mutex::new(HashMap::new())),
        thumb_gen: Arc::new(Mutex::new(HashSet::new())),
        thumb_gen_budget: Arc::new(tokio::sync::Semaphore::new(
            super::photos::THUMB_GEN_CONCURRENCY,
        )),
        file_thumb_generation: Arc::new(AtomicU64::new(0)),
        file_thumb_cancel: Arc::new(tokio::sync::Notify::new()),
        thumbnail_build: Arc::new(Mutex::new(Default::default())),
        thumbnail_build_cancelled: Arc::new(AtomicBool::new(false)),
        thumbnail_build_cancel: Arc::new(tokio::sync::Notify::new()),
        thumbnail_misses: Arc::new(Mutex::new(Default::default())),
        quota: Arc::new(Mutex::new(None)),
        quota_refreshing: Arc::new(AtomicBool::new(false)),
        size_upgrades: Arc::new(Mutex::new(HashMap::new())),
        size_waiters: Arc::new(Default::default()),
        notifier: Arc::new(OnceLock::new()),
        session_live: Arc::new(AtomicBool::new(false)),
        transfers: TransferRegistry::new(),
        events: Arc::new(Default::default()),
        indexing: Arc::new(AtomicBool::new(false)),
        sync_progress: Arc::new(Mutex::new(HashMap::new())),
        sync_engine: Arc::default(),
        sync_tx,
        mounts: Arc::new(Mutex::new(SecondaryMountRegistry::default())),
        sync_locks: Arc::new(Mutex::new(HashMap::new())),
        sync_unscanned: Arc::new(Mutex::new(HashMap::new())),
        states: Arc::new(StateRegistry::default()),
    };
    // Before anything can queue work against it: the drain thread below reaches
    // every mount's session through this registry.
    core.register_state(mountpoint);
    core.transfers.set_limits(upload_limit, download_limit);

    // Writes queued by a previous run (or left behind by a crash) are still owed
    // an upload, and reads must be served from their staged blobs until they land.
    core.hydrate_pending();
    // Then the writes that were fsynced but never closed, which the cache moved
    // aside at open. After `hydrate_pending`, so a recovered partial write can
    // see an already-queued write to the same node.
    core.recover_fsynced_writes();
    // And finally the staged blobs neither of those accounts for: bytes a
    // failed release left behind with nothing pointing at them.
    core.reconcile_staging();
    // Several drain workers over one queue. One thread meant a single large
    // upload held every queued rename, trash and small write behind it for as
    // long as it ran; the claim column (`Db::claim_next_due_op`) keeps the
    // workers off each other's nodes, which is the only ordering the queue
    // actually needs. Worker 0 additionally runs the queue's idle chores.
    //
    // Every long-lived thread started from here on is kept in `workers` and
    // joined at the end of this function. Leaving them running turned an
    // in-process remount into a second generation of drain workers, sync engines
    // and sweeps all operating on a mount that no longer exists (`docs/BUGS.md` B44).
    let mut workers: Vec<std::thread::JoinHandle<()>> = Vec::new();
    for worker in 0..DRAIN_WORKERS {
        let drain_core = core.clone();
        let spawned = std::thread::Builder::new()
            .name(format!("pdfs-drain-{worker}"))
            .spawn(move || drain_core.run_pending_drain(worker == 0));
        keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
    }

    // Start the folder-sync engine. It watches every mirror folder, polls the
    // remotes, and reconciles on its own thread — never in front of a FUSE call.
    workers.extend(sync::spawn(core.clone(), sync_rx));
    // A timed pause persisted by an earlier run still has to end on time.
    if pause::paused_at(paused_until, now_secs()) && paused_until != pause::PAUSED_INDEFINITELY {
        core.arm_pause_timer(paused_until);
    }

    // Reconcile leftover `(sync-conflict …)` copies: drop the ones that turned
    // out identical to the live file, surface the ones that genuinely diverge.
    // Its own thread — it enumerates and trashes, and must never block a FUSE call.
    // `Off` skips the thread entirely rather than starting an idle one, so the
    // setting is verifiable from the outside (`ls /proc/<pid>/task/*/comm`).
    if sweep_mode == SweepMode::Off {
        info!("conflict sweep disabled by configuration");
    } else {
        info!(mode = ?sweep_mode, "conflict sweep enabled");
        let sweep_core = core.clone();
        let spawned = std::thread::Builder::new()
            .name("pdfs-conflict-sweep".into())
            .spawn(move || sweep_core.run_conflict_sweep_loop());
        keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
    }

    // Watch the link for the life of the mount: a mount from the cache waits for
    // the network to come back, and an online one can lose it at any moment and
    // must notice it coming back just the same.
    {
        let probe_core = core.clone();
        let spawned = std::thread::Builder::new()
            .name("pdfs-online-probe".into())
            .spawn(move || probe_core.run_online_probe());
        keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
    }

    // Keep the launcher prompt's "This computer" index fresh. Its own thread:
    // the walk is I/O-heavy and must never sit in front of a FUSE callback.
    if matches!(host, Host::Daemon) {
        let db = core.db.clone();
        let indexing = core.indexing.clone();
        let transfers = core.transfers.clone();
        let mountpoint = mountpoint.to_path_buf();
        let shutdown = core.shutdown.clone();
        let spawned = std::thread::Builder::new()
            .name("pdfs-localindex".into())
            .spawn(move || run_local_index(db, indexing, transfers, mountpoint, shutdown));
        keep_or_abandon(&core, control_socket, &mut workers, spawned)?;
    }

    // Bind the control socket before the FUSE session takes over the thread. A
    // stale socket file from a previous run would block the bind, so clear it.
    let _ = std::fs::remove_file(control_socket);
    let old_umask = unsafe { libc::umask(0o77) };
    let listener_res = UnixListener::bind(control_socket);
    unsafe { libc::umask(old_umask) };
    let listener = match listener_res {
        Ok(listener) => listener,
        Err(error) => {
            abandon_workers(&core, control_socket, workers);
            return Err(error);
        }
    };
    // Owner-only before anything can connect: a peer on this socket commands the
    // daemon's authenticated session without a credential of its own (B6).
    if let Err(e) = pdfs_core::config::restrict_socket(control_socket) {
        error!(error = %e, "cannot restrict control socket permissions; refusing to serve");
        abandon_workers(&core, control_socket, workers);
        return Err(std::io::Error::other(format!(
            "control socket permissions: {e}"
        )));
    }
    // Do not accept control requests until `spawn_session` has transitioned the
    // primary registration to live. The listener may queue a connection during
    // this short window, but its handler cannot observe `mounted = false`.
    let bg = match spawn_session(&core, mountpoint, root) {
        Ok(session) => session,
        Err(error) => {
            abandon_workers(&core, control_socket, workers);
            return Err(error);
        }
    };
    core.mounts.lock().open();
    // The connection id, captured now while the mount is live, so a stop signal
    // mid-transfer can abort it rather than block `join` (see `abort_fuse_connection`).
    let main_conn = fuse_connection_id(mountpoint);
    {
        let control_core = core.clone();
        let username = username.clone();
        let mountpoint = mountpoint.to_path_buf();
        match std::thread::Builder::new()
            .name("pdfs-control".into())
            .spawn(move || run_control_socket(control_core, username, mountpoint, listener))
        {
            Ok(handle) => workers.push(handle),
            Err(error) => {
                // A failure here abandons the mount, so the background threads
                // started above have to be told to stop with it.
                abandon_workers(&core, control_socket, workers);
                let secondaries = core.mounts.lock().close_and_drain();
                for (id, mount) in secondaries {
                    if let Err(unmount_error) = mount.teardown() {
                        warn!(id, error = %unmount_error, "secondary teardown after control startup failure failed");
                    }
                }
                if let Err(unmount_error) = teardown_session(&core.session_live, || {
                    umount_session_unblocked(bg, main_conn)
                }) {
                    warn!(error = %unmount_error, "unmount after control startup failure failed");
                }
                let _ = std::fs::remove_file(control_socket);
                return Err(error);
            }
        }
    }
    // Publishes what changes without one place to publish it from (status,
    // transfers, sync progress) to the front-ends subscribed on that socket.
    {
        let sampler_core = core.clone();
        let username = username.clone();
        let mountpoint = mountpoint.to_path_buf();
        match std::thread::Builder::new()
            .name("pdfs-events".into())
            .spawn(move || events::run_event_sampler(sampler_core, username, mountpoint))
        {
            Ok(handle) => workers.push(handle),
            // Not fatal: front-ends still get every published change, only
            // not the sampled ones.
            Err(error) => warn!(error = %error, "could not start the event sampler"),
        }
    }
    // The supervisor watches the pool and answers the systemd watchdog. Started
    // after the listener, because its liveness probe is a round trip over that
    // socket.
    {
        let supervisor_core = core.clone();
        let socket = control_socket.to_path_buf();
        match std::thread::Builder::new()
            .name("pdfs-supervisor".into())
            .spawn(move || supervisor::run(supervisor_core, socket))
        {
            Ok(handle) => workers.push(handle),
            // Not fatal: the mount works without a supervisor, it just goes back
            // to being undiagnosable if it wedges.
            Err(error) => warn!(error = %error, "could not start the daemon supervisor"),
        }
    }
    // The mount is live and the control socket answers: this is the moment the
    // unit is actually usable, which is what `Type=notify` reports. Whatever
    // failure an earlier attempt recorded for the front ends is over.
    systemd::ready();
    if let Some(state_dir) = control_socket.parent() {
        pdfs_core::mountfail::clear(state_dir);
    }

    // Re-establish on-demand mounts only after the primary session is live.
    // Keep the worker so shutdown can close publication, wait for any in-flight
    // fetch to observe that closure, and then drain a stable registry.
    let restore_worker = {
        let restore_core = core.clone();
        match std::thread::Builder::new()
            .name("pdfs-restore-ondemand".into())
            .spawn(move || restore_core.restore_ondemand_mounts())
        {
            Ok(worker) => Some(worker),
            Err(error) => {
                warn!(%error, "start on-demand restore worker failed");
                None
            }
        }
    };
    rt.spawn(run_event_sync(core.drive.clone(), scope, core.clone()));
    // The photos volume has its own event stream; without it the gallery only
    // learns about a phone-side deletion when the timeline goes stale.
    rt.spawn(run_photos_event_sync(core.clone()));
    // Not in a simulation: `client` there reaches no API, and the thread, still
    // waiting on it after the stop, held the database a test opens next.
    if online && matches!(host, Host::Daemon) {
        let repair_core = core.clone();
        let root_uid = core.primary_root_uid.clone();
        if let Err(error) = std::thread::Builder::new()
            .name("pdfs-share-id".into())
            .spawn(move || repair_core.repair_primary_share_id(&root_uid))
        {
            warn!(%error, "start My Files share-id repair failed");
        }
    }

    // Stop signals (SIGTERM from `systemctl --user stop`, SIGINT from Ctrl-C)
    // are delivered onto the async runtime; bridge them onto a sync channel so
    // the loop below can react without blocking a worker thread. A bounded
    // channel of 1 is enough — we only need to know that *a* stop arrived.
    let sig_rx = match host {
        Host::Daemon => signal_stop(&rt),
        #[cfg(test)]
        Host::Simulation(rx) => rx,
    };

    // Wait for whichever happens first: a stop signal (→ we unmount ourselves
    // via the lazy MNT_DETACH path, clean even mid-download), or the kernel
    // mount ending on its own (→ the session thread finishes). Poll instead of
    // blocking on `join` so we can also notice the signal.
    let outcome = loop {
        match sig_rx.recv_timeout(Duration::from_millis(500)) {
            Ok(()) => {
                info!("stop requested; unmounting");
                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::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");
                    core.mounts.lock().close();
                    if let Err(e) = teardown_session(&core.session_live, || bg.join()) {
                        warn!(error = %e, "session join failed");
                    }
                    break MountOutcome::Unmounted;
                }
            }
            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                // Signal task gone (failed to install); fall back to join.
                core.mounts.lock().close();
                let _ = teardown_session(&core.session_live, || bg.join());
                break MountOutcome::Unmounted;
            }
        }
    };

    // Unmount every on-demand sync folder too, or the kernel mounts linger as
    // stale and the next start fails with EBUSY.
    if let Some(worker) = restore_worker
        && worker.join().is_err()
    {
        warn!("on-demand restore worker panicked");
    }
    let secondaries = core.mounts.lock().close_and_drain();
    for (id, mount) in secondaries {
        if let Err(e) = mount.teardown() {
            warn!(id, error = %e, "unmount on-demand folder failed");
        }
    }
    core.queue_unreleased_writes();

    // Last, because the workers may legitimately be finishing an upload, and
    // nothing they can still do is harmful once the mounts are down.
    stop_workers(&core, control_socket, workers);
    Ok(outcome)
}

/// A channel that receives once SIGTERM or SIGINT arrives.
fn signal_stop(rt: &tokio::runtime::Handle) -> std::sync::mpsc::Receiver<()> {
    let (sig_tx, sig_rx) = std::sync::mpsc::sync_channel::<()>(1);
    rt.spawn(async move {
        let mut sigterm =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
                Ok(s) => s,
                Err(e) => {
                    warn!(error = %e, "install SIGTERM handler failed");
                    return;
                }
            };
        let mut sigint =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) {
                Ok(s) => s,
                Err(e) => {
                    warn!(error = %e, "install SIGINT handler failed");
                    return;
                }
            };
        tokio::select! {
            _ = sigterm.recv() => info!("received SIGTERM"),
            _ = sigint.recv() => info!("received SIGINT"),
        }
        let _ = sig_tx.try_send(());
    });
    sig_rx
}

/// How long teardown waits for the background workers and then the FUSE worker
/// pool before proceeding without them. Long enough for an in-flight block to
/// finish, short enough that `systemctl --user stop` still returns inside
/// `TimeoutStopSec`.
const WORKER_JOIN_DEADLINE: Duration = Duration::from_secs(10);

/// Signal every background loop, wake the ones that are blocked, and join them.
///
/// The mount used to return with its drain workers, sync engine, sweep, online
/// probe, indexer and control listener all still running on `Core` clones. The
/// process usually exited straight afterwards and hid it; an in-process remount
/// did not, and started a second full set on top of the first (`docs/BUGS.md` B44).
///
/// Two of the waits need waking rather than merely flagging: the drain workers
/// block on their own condvar (`wake_drain`), and the control listener blocks
/// inside `accept`, which only returns when a connection arrives — so it gets
/// one, from us, after the flag is already set. A join that takes a while is
/// expected and correct: a worker part-way through an upload finishes it rather
/// than abandoning the user's bytes mid-flight. Up to [`WORKER_JOIN_DEADLINE`]:
/// past it, systemd kills the daemon anyway, and a queued op is sent again.
fn stop_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
    // Say so before the joins: a teardown that takes a while must not be read as
    // a missed watchdog ping and turned into a kill.
    systemd::stopping();
    halt_workers(core, control_socket, workers);
}

/// Stop the threads a mount attempt started before it failed.
///
/// The early returns used to leave them running: every retry started another
/// set beside the last one's, and their `Core` kept the database open, so the
/// next attempt could fail on its lock instead of the real error
/// (`docs/BUGS.md` B264). Unlike [`stop_workers`] this does not tell systemd
/// the daemon is stopping, because it is about to try again, and a manager
/// told so would wait for the process to exit and then kill it.
fn abandon_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
    halt_workers(core, control_socket, workers);
}

/// Keep a worker the mount just spawned, or, if the spawn failed, stop the
/// ones already running and hand the error back.
fn keep_or_abandon(
    core: &Core,
    control_socket: &Path,
    workers: &mut Vec<std::thread::JoinHandle<()>>,
    spawned: std::io::Result<std::thread::JoinHandle<()>>,
) -> std::io::Result<()> {
    match spawned {
        Ok(handle) => {
            workers.push(handle);
            Ok(())
        }
        Err(error) => {
            abandon_workers(core, control_socket, std::mem::take(workers));
            Err(error)
        }
    }
}

/// Signal every worker to stop, then join them up to [`WORKER_JOIN_DEADLINE`].
fn halt_workers(core: &Core, control_socket: &Path, workers: Vec<std::thread::JoinHandle<()>>) {
    core.shutdown.stop();
    core.wake_drain();
    let _ = core.sync_tx.send(sync::SyncMsg::Stop);
    // Connect while the socket still exists — that is the whole point of the
    // poke — then unlink it so nothing else can arrive behind us. The listener
    // re-checks the flag after every accept, so even a connection that races the
    // unlink cannot put it back to sleep.
    let _ = std::os::unix::net::UnixStream::connect(control_socket);
    let _ = std::fs::remove_file(control_socket);
    let deadline = Instant::now() + WORKER_JOIN_DEADLINE;
    let running = join_until(workers, deadline);
    if !running.is_empty() {
        warn!(
            running = running.join(","),
            "background workers did not finish in time; shutting down around them"
        );
    }
    // The FUSE worker pool last, and *before* the caller drops the tokio
    // runtime: a worker still inside a job when the runtime goes away panics on
    // its next timer with "A Tokio 1.x context was found, but it is being
    // shutdown" (seen 2026-09-18 09:58:35). Bounded, so a worker wedged on the
    // network cannot hold the stop the user asked for.
    core.workers
        .stop_and_join(deadline.saturating_duration_since(Instant::now()));
    debug!("background workers stopped");
}

/// Join `workers` until `deadline`, and name the ones still running then.
///
/// A stop used to wait for each one however long it took. The sync engine's
/// mirror pass waits on Drive with no deadline, and a stop that waited for it
/// was killed by systemd (`docs/BUGS.md` B165).
fn join_until(workers: Vec<std::thread::JoinHandle<()>>, deadline: Instant) -> Vec<String> {
    let mut running = Vec::new();
    for worker in workers {
        while !worker.is_finished() && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(10));
        }
        if !worker.is_finished() {
            running.push(worker.thread().name().unwrap_or("unnamed").to_string());
        } else if worker.join().is_err() {
            warn!("a background worker panicked before shutdown");
        }
    }
    running
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::{Duration, Instant};

    use super::{SecondaryInsertRejection, SecondaryMountRegistry, join_until, teardown_session};

    #[test]
    fn a_stop_names_the_workers_still_running_at_its_deadline() {
        let (release, wait) = std::sync::mpsc::channel::<()>();
        let done = std::thread::Builder::new()
            .name("done".into())
            .spawn(|| {})
            .unwrap();
        let stuck = std::thread::Builder::new()
            .name("stuck".into())
            .spawn(move || {
                let _ = wait.recv();
            })
            .unwrap();
        let running = join_until(
            vec![done, stuck],
            Instant::now() + Duration::from_millis(100),
        );
        assert_eq!(running, ["stuck"]);
        drop(release);
    }

    #[test]
    fn session_is_unroutable_before_teardown_starts() {
        let live = AtomicBool::new(true);
        teardown_session(&live, || {
            assert!(
                !live.load(Ordering::Acquire),
                "teardown must never run while the registry still reports mounted"
            );
        });
    }

    #[test]
    fn secondary_registry_rejects_closed_and_duplicate_insertions() {
        let mut mounts = SecondaryMountRegistry::default();
        assert_eq!(
            mounts.insert(7, "closed"),
            Err((SecondaryInsertRejection::Closed, "closed"))
        );

        mounts.open();
        assert_eq!(mounts.insert(7, "first"), Ok(()));
        assert_eq!(
            mounts.insert(7, "duplicate"),
            Err((SecondaryInsertRejection::Duplicate, "duplicate"))
        );
        assert!(mounts.contains_key(&7));

        assert_eq!(mounts.close_and_drain(), vec![(7, "first")]);
        assert!(!mounts.is_accepting());
    }
}