//! Parallel filesystem walk for the full indexing run.
//!
//! One shared queue of directories, N worker threads. The important part is
//! *what* a worker keeps together: it reads a directory **and** does that
//! directory's per-file work — stat, classify, hash — before moving on.
//!
//! That grouping is the whole design. On SMB, one `QUERY_DIRECTORY` returns
//! size, mtime and attributes for every entry in a directory, and the cifs
//! client primes its inode cache from the reply — but only for `actimeo`,
//! one second by default, and end-user mount options are not ours to set. A
//! `stat` issued right after the directory read is therefore free, while the
//! same `stat` a few seconds later is a full network round trip. Parallelising
//! the directory reads alone — what a general-purpose parallel walker does —
//! hands entries to a consumer that stats them well outside that window, so it
//! throws the cache away and lands *slower* than a serial walk while burning
//! more CPU.
//!
//! The second property this buys: every path below a root is canonical by
//! construction. Roots are canonicalized once at seed time and directories are
//! only ever reached by joining names onto them, so per-file `canonicalize`
//! calls — roughly one `readlink` per path component, per file — disappear.
//! Symlinks are the one exception and are resolved where they are found.
//!
//! Deliberately std-only (`std::thread` + `std::sync::mpsc`), matching the
//! house style set out in [`crate::watcher`].

use std::collections::HashSet;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{mpsc, Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use std::time::UNIX_EPOCH;

use sha2::{Digest, Sha256};

use crate::config::{Config, IgnoreSet};
use crate::extract::Registry;
use crate::file_handling::{
    classify_by_mtime, classify_for_indexing, path_to_db_string, prepare_file_record,
    warn_if_unrepresentable, DirRows, FileIndexAction, OwnedNewFile, UnreadableDirs,
};
use crate::indexing::should_abort;

/// Files one worker takes for itself before handing the rest to the pool.
///
/// Without this split a single very wide directory — a Photos or Downloads
/// folder, a scanned-document share — would be walked by exactly one thread.
const FILES_PER_JOB: usize = 128;

/// Bounded hand-off to the DB writer. Deep enough that workers don't stall
/// while a transaction commits, shallow enough to bound memory.
const CHANNEL_CAP: usize = 4096;

/// Worker threads for a root on local storage. The work is latency-bound
/// rather than CPU-bound, but a local disk needs little queueing and deep
/// parallelism just adds seeks.
const LOCAL_THREADS: usize = 4;

/// Worker threads for a root on a network filesystem. Every uncached
/// metadata operation is a round trip, so throughput is round-trips-in-flight
/// divided by latency; threads are how we raise the numerator.
const NETWORK_THREADS: usize = 16;

/// One file the walk found, with everything the DB writer needs.
#[derive(Debug)]
pub struct WalkedFile {
    /// Canonical path, and the `files.path` key.
    pub path: String,
    pub action: FileIndexAction,
    /// `None` when there is nothing to write: the file was unchanged, or its
    /// record could not be built. Never a reason to drop [`WalkedFile::path`].
    pub record: Option<OwnedNewFile>,
    /// 128-bit truncated SHA-256 of [`WalkedFile::path`], for the writer's
    /// duplicate-visit set. Computed here so the cost lands on the walk pool
    /// rather than on the single writer thread.
    pub digest: u128,
    /// True when this file was reached by resolving a symlink, so the
    /// directory it was found in is *not* the directory its row belongs to.
    ///
    /// Such a row is invisible to the reconciliation of its real parent —
    /// which may be a directory no walk ever visits — so the caller must
    /// exempt it from the vanished-directory sweep.
    pub aliased: bool,
}

impl WalkedFile {
    /// Seen, but with nothing to write. Distinct from not being emitted at
    /// all: the row stays.
    fn skipped(path: String, digest: u128, aliased: bool) -> Self {
        WalkedFile {
            path,
            action: FileIndexAction::Skip,
            record: None,
            digest,
            aliased,
        }
    }
}

/// What the walk emits. Files as they are classified, plus the per-directory
/// verdict on which index rows no longer have a file behind them.
#[derive(Debug)]
pub enum WalkEvent {
    File(WalkedFile),
    /// Paths whose row should be deleted: present in one directory's index
    /// rows, absent from that directory's listing. Emitted once per directory
    /// read, and only for directories that were read successfully.
    Stale(Vec<String>),
}

/// Work waiting for a thread.
enum Job {
    /// Read this directory and process its files. Carries the directory's
    /// index rows, fetched by the prefetcher before the job became runnable.
    Dir(PathBuf, Arc<DirRows>),
    /// Process this slice of one directory's files, split off because the
    /// directory was too wide for one worker to be worth serialising on.
    /// Shares its directory's rows.
    Files(Vec<PathBuf>, Arc<DirRows>),
    /// A resolved symlink target, with the stored mtime for its own path.
    /// Classified against that rather than against any directory's rows.
    Alias(PathBuf, Option<u64>),
}

/// Directories fetched but not yet taken by a worker. Each holds its rows
/// live, so without a cap the prefetcher would run ahead of the pool and
/// materialise rows for thousands of directories at once — reintroducing in
/// one structure the memory this design exists to remove.
const PREFETCH_AHEAD: usize = 64;

#[derive(Default)]
struct Queue {
    /// LIFO. A directory's children are processed close in time to the read
    /// that discovered them, which is what the attribute cache rewards, and it
    /// keeps the live frontier depth-first-ish instead of holding an entire
    /// breadth-first level in memory.
    jobs: Vec<Job>,
    /// Directories discovered but not yet given their rows, and symlink
    /// targets awaiting an mtime lookup. The prefetcher drains both.
    needs_rows: Vec<PathBuf>,
    needs_alias: Vec<PathBuf>,
    /// Set while the prefetcher is mid-query, holding work that is in neither
    /// list. Part of the end-of-walk proof: see [`Shared::take`].
    prefetching: bool,
    /// Fetched directories sitting in `jobs`, i.e. the ones actually holding
    /// rows. What [`PREFETCH_AHEAD`] bounds.
    ///
    /// Counted rather than derived from `jobs.len()`: that also holds
    /// `Job::Files` chunks, which carry no rows of their own, and a wide
    /// directory pushes many of them. Bounding on the total would let file
    /// chunks starve directory prefetching and leave the pool waiting on
    /// rows that were never fetched.
    dirs_ready: usize,
    /// Workers currently holding a job — that is, workers that may still push
    /// more. The walk is over when this is zero and `jobs` is empty.
    active: usize,
    /// Canonical directories already queued. Collapses overlapping roots and
    /// makes symlink cycles impossible: a cycle must revisit a canonical path,
    /// and every directory pushed here is canonical.
    ///
    /// Also the record of which directories the walk reached, which the
    /// caller's vanished-directory sweep reads once the walk has finished.
    seen_dirs: HashSet<PathBuf>,
    done: bool,
}

impl Queue {
    /// Whether any stage still holds work. The prefetch stage is invisible to
    /// a `jobs`/`active` test, so it has to be named here explicitly.
    fn idle(&self) -> bool {
        self.jobs.is_empty()
            && self.needs_rows.is_empty()
            && self.needs_alias.is_empty()
            && !self.prefetching
            && self.active == 0
    }
}

struct Shared {
    queue: Mutex<Queue>,
    idle: Condvar,
    /// Workers currently processing (not parked waiting for work). Purely
    /// observational, for progress display: two relaxed atomic ops per
    /// *job* (a directory read plus up to [`FILES_PER_JOB`] files), so it
    /// costs nothing the queue mutex didn't already.
    stats: WorkerStats,
}

/// Decrements the busy count however the worker leaves its job — including
/// early returns on stop and panics.
pub(crate) struct BusyGuard<'a>(&'a AtomicUsize);

impl Drop for BusyGuard<'_> {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::Relaxed);
    }
}

/// Lock-free view of a worker pool's activity for progress displays.
///
/// Shared by the walk and by [`crate::content`]: a root runs one pool and then
/// the other, and the progress line has to report whichever is live — a pool
/// whose threads have exited reads as busy 0, which is only the truth while
/// that pool is the one running.
#[derive(Clone)]
pub struct WorkerStats {
    busy: Arc<AtomicUsize>,
    total: usize,
}

impl WorkerStats {
    pub(crate) fn new(total: usize) -> Self {
        WorkerStats {
            busy: Arc::new(AtomicUsize::new(0)),
            total,
        }
    }

    /// Count the calling thread as busy until the returned guard drops.
    pub(crate) fn enter(&self) -> BusyGuard<'_> {
        self.busy.fetch_add(1, Ordering::Relaxed);
        BusyGuard(&self.busy)
    }

    /// Workers doing work right now (the rest are parked).
    pub fn active(&self) -> usize {
        self.busy.load(Ordering::Relaxed).min(self.total)
    }

    pub fn total(&self) -> usize {
        self.total
    }
}

impl Shared {
    /// Claim a job, blocking while other workers are still running.
    ///
    /// Returns `None` only when the queue is empty *and* no worker holds a
    /// job — at that instant nobody is left who could push more, so the walk
    /// is provably finished.
    fn take(&self) -> Option<(Job, ActiveJob<'_>)> {
        let mut q = self.queue.lock().unwrap();
        loop {
            if q.done {
                return None;
            }
            if let Some(job) = q.jobs.pop() {
                q.active += 1;
                if matches!(job, Job::Dir(..)) {
                    q.dirs_ready -= 1;
                }
                // The prefetcher may have been parked behind PREFETCH_AHEAD.
                self.idle.notify_all();
                return Some((
                    job,
                    ActiveJob {
                        shared: self,
                        finished: false,
                    },
                ));
            }
            // Nothing runnable. Only "nobody anywhere holds work" proves the
            // walk is over — a directory sitting in the prefetch stage still
            // becomes a job, and declaring the walk finished with one parked
            // there would hand reconciliation a partial file set.
            if q.idle() {
                q.done = true;
                self.idle.notify_all();
                return None;
            }
            q = self.idle.wait(q).unwrap();
        }
    }

    /// Claim one unit of prefetch work, or `None` once the walk is over.
    ///
    /// Parks while the runnable queue is already `PREFETCH_AHEAD` deep, so
    /// fetched-but-unclaimed rows stay bounded.
    fn take_prefetch(&self) -> Option<PrefetchWork> {
        let mut q = self.queue.lock().unwrap();
        loop {
            if q.done {
                return None;
            }
            // Aliases are never throttled: they carry a single mtime, not a
            // directory's rows, so they cost nothing to hold.
            if let Some(path) = q.needs_alias.pop() {
                q.prefetching = true;
                return Some(PrefetchWork::Alias(path));
            }
            if q.dirs_ready < PREFETCH_AHEAD {
                if let Some(dir) = q.needs_rows.pop() {
                    q.prefetching = true;
                    return Some(PrefetchWork::Dir(dir));
                }
            }
            if q.idle() {
                q.done = true;
                self.idle.notify_all();
                return None;
            }
            q = self.idle.wait(q).unwrap();
        }
    }

    /// Publish a prefetched job and clear the in-flight flag together, under
    /// one lock — the same indivisibility `publish` relies on, for the same
    /// reason.
    fn finish_prefetch(&self, job: Job) {
        let mut q = self.queue.lock().unwrap();
        if matches!(job, Job::Dir(..)) {
            q.dirs_ready += 1;
        }
        q.jobs.push(job);
        q.prefetching = false;
        self.idle.notify_all();
    }

    /// Give the prefetch slot back without producing a job (the query failed).
    fn abandon_prefetch(&self) {
        let mut q = self.queue.lock().unwrap();
        q.prefetching = false;
        self.idle.notify_all();
    }
}

/// One unit of work for the prefetcher.
enum PrefetchWork {
    Dir(PathBuf),
    Alias(PathBuf),
}

/// What a worker discovered while reading a directory.
///
/// Distinct from [`Job`] because most of it is not yet runnable: a newly
/// discovered directory has no rows, and a symlink target has no stored
/// mtime, until the prefetcher supplies them.
enum Found {
    /// A subdirectory. Needs its rows before a worker can classify inside it.
    Dir(PathBuf),
    /// A resolved symlink target. Needs an exact-path mtime lookup.
    Alias(PathBuf),
    /// Overflow files from the directory just read, which already has rows.
    Files(Vec<PathBuf>, Arc<DirRows>),
}

impl Shared {
    /// Push discovered work and give the job slot back, under a single lock
    /// acquisition. Doing both together is what makes the idle test in
    /// [`Shared::take`] an end-of-walk proof rather than a race: a worker
    /// that has popped the last job but not yet published its children must
    /// never look idle.
    fn publish(&self, found: Vec<Found>) {
        let mut q = self.queue.lock().unwrap();
        for item in found {
            match item {
                Found::Dir(dir) => {
                    if q.seen_dirs.insert(dir.clone()) {
                        q.needs_rows.push(dir);
                    }
                }
                Found::Alias(path) => q.needs_alias.push(path),
                Found::Files(files, rows) => q.jobs.push(Job::Files(files, rows)),
            }
        }
        q.active -= 1;
        self.idle.notify_all();
    }

    fn shutdown(&self) {
        let mut q = self.queue.lock().unwrap();
        q.done = true;
        self.idle.notify_all();
    }
}

/// Hands the job slot back even if the worker panics or returns early. A
/// stranded count would leave every other worker waiting on a number that
/// never reaches zero.
struct ActiveJob<'a> {
    shared: &'a Shared,
    finished: bool,
}

impl ActiveJob<'_> {
    fn finish(mut self, found: Vec<Found>) {
        self.shared.publish(found);
        self.finished = true;
    }
}

impl Drop for ActiveJob<'_> {
    fn drop(&mut self) {
        if !self.finished {
            self.shared.publish(Vec::new());
        }
    }
}

struct Ctx {
    follow_symlinks: bool,
    include_hidden: bool,
    ignore: IgnoreSet,
    config: Config,
    /// Lets a worker finish small text files outright: the head it reads to
    /// hash them is already their entire contents, so an extractor that works
    /// from bytes saves the content pass an open/read/close per file.
    registry: Arc<Registry>,
    unreadable: UnreadableDirs,
    stop_flag: Arc<AtomicBool>,
    suspend_flag: Arc<AtomicBool>,
}

/// Read one directory, apply the hidden/ignore rules, and split the result:
/// subdirectories and overflow file chunks go to `found` for the pool, the
/// remaining files come back for this worker to handle immediately.
///
/// Also reconciles the directory against its index rows. This is the right
/// place for it and the only cheap one: the *complete* filtered listing
/// exists here, before the directory is split across workers, so the diff
/// needs no per-directory completion count. `stale` receives the paths whose
/// row has no file behind it any more.
///
/// A directory that cannot be read returns before reconciling, so nothing
/// under it is ever deleted — an unreadable directory must not read as an
/// empty one.
fn read_directory(
    dir: &Path,
    rows: &Arc<DirRows>,
    ctx: &Ctx,
    found: &mut Vec<Found>,
    stale: &mut Vec<String>,
) -> Vec<PathBuf> {
    let entries = match fs::read_dir(dir) {
        Ok(entries) => entries,
        Err(e) => {
            // Not the same as "this directory is empty": see UnreadableDirs.
            crate::log_warn!("cannot read {}: {}", dir.display(), e);
            ctx.unreadable.record(dir.to_path_buf());
            return Vec::new();
        }
    };
    // Names surviving the filters, for the diff below. Only meaningful
    // because every `continue` in the loop is a genuine "not indexable",
    // matching what would have been absent from the old global seen set.
    let mut present: HashSet<String> = HashSet::new();
    let mut unreadable_entry = false;

    let mut files = Vec::new();
    for entry in entries {
        let entry = match entry {
            Ok(entry) => entry,
            Err(e) => {
                crate::log_warn!("cannot read an entry of {}: {}", dir.display(), e);
                ctx.unreadable.record(dir.to_path_buf());
                // The listing is now incomplete, so it cannot be used to
                // decide what is missing: an entry we failed to read would
                // look identical to one that was deleted.
                unreadable_entry = true;
                continue;
            }
        };

        let name = entry.file_name();
        let name = name.to_string_lossy();
        // `entry.metadata()` is only consulted on Windows, where it is free —
        // the attributes came back with the directory read. On Unix the
        // closure is never called, so this stays at zero extra syscalls.
        if !ctx.include_hidden && crate::platform::entry_is_hidden(&name, || entry.metadata().ok())
        {
            continue;
        }
        if ctx.ignore.matches_component(&name) {
            continue;
        }
        let path = entry.path();
        if ctx.ignore.matches_path_pattern(&path) {
            continue;
        }

        // `file_type` is the cached `d_type` from the directory read, so
        // splitting directories from files here is free.
        match entry.file_type() {
            // Directories hold no `files` row, so they are deliberately not
            // marked present: a name that was a file last run and is a
            // directory now *should* lose its row.
            Ok(ft) if ft.is_dir() => found.push(Found::Dir(path)),
            Ok(ft) if ft.is_symlink() => {
                // With links off there is nothing here to index — targets
                // included. Returning before `canonicalize` also drops a
                // readlink chain and a stat per symlink, so honouring the
                // setting costs fewer syscalls than ignoring it, not more.
                //
                // Both kinds have to be gated together, or the two walkers
                // disagree: `filtered_walk` (which the watcher and the
                // incremental path use) passes `follow_links` straight to
                // walkdir and follows neither. A file target followed here but
                // not there is indexed by every full run and never updated
                // between them — and, if it resolves outside every configured
                // root, indexed despite living somewhere the user never asked
                // us to look.
                if !ctx.follow_symlinks {
                    continue;
                }
                // Resolve aliases where they are found. The target's canonical
                // path is what the index stores, and pushing only canonical
                // directories is what keeps `seen_dirs` able to break cycles.
                //
                // Normalized like the roots (walk_parallel), or on Windows the
                // target keeps `canonicalize`'s `\\?\` prefix: every path below
                // it would be spelled differently from the plainly-spelled
                // roots, so full-path ignore patterns would never match under a
                // followed junction and `seen_dirs` could not dedup against an
                // overlapping root.
                if let Ok(target) = path.canonicalize() {
                    let target = PathBuf::from(path_to_db_string(&target));
                    match fs::metadata(&target) {
                        Ok(m) if m.is_dir() => found.push(Found::Dir(target)),
                        // The row for a resolved target belongs to the
                        // target's own directory, not this one, so it is not
                        // marked present here and cannot be classified
                        // against these rows.
                        Ok(_) => found.push(Found::Alias(target)),
                        Err(_) => {}
                    }
                }
            }
            Ok(_) => {
                present.insert(name.into_owned());
                files.push(path);
            }
            // Type unknown: the entry exists but we could not classify it.
            // Mark it present so an existing row survives — seen, not deleted.
            Err(_) => {
                present.insert(name.into_owned());
            }
        }
    }

    if !unreadable_entry {
        for name in rows.keys() {
            if !present.contains(name.as_str()) {
                // Rebuild the stored path the way `prepare` does, by joining
                // onto the canonical directory, so separators and roots match
                // the `files.path` spelling exactly.
                stale.push(path_to_db_string(&dir.join(name)));
            }
        }
    }

    // Spread a wide directory across the pool, keeping the tail for
    // ourselves so the entries the read just warmed are handled now. Each
    // chunk shares this directory's rows: they are the same directory, and
    // classifying them against anything else would read every file as new.
    while files.len() > FILES_PER_JOB {
        let chunk = files.split_off(files.len() - FILES_PER_JOB);
        found.push(Found::Files(chunk, rows.clone()));
    }
    files
}

/// How a file's stored mtime is to be found.
enum Known<'a> {
    /// By name within the directory being walked — the ordinary case.
    InDir(&'a DirRows),
    /// Already resolved by exact path, for a symlink target whose row lives
    /// under a different parent.
    Exact(Option<u64>),
}

/// 128-bit truncated SHA-256 of a path, for the writer's duplicate-visit set.
///
/// Truncated rather than full: 16 bytes is ~4e-26 collision probability at
/// 7M paths, where 8 bytes would be ~1e-6 — and a collision here silently
/// drops a real file from the index. Cryptographic rather than fast because
/// filenames on a shared volume are attacker-supplied, so a cheap hash would
/// let a chosen pair hide one of the two files.
pub fn path_digest(path: &str) -> u128 {
    let digest = Sha256::digest(path.as_bytes());
    let mut bytes = [0u8; 16];
    bytes.copy_from_slice(&digest[..16]);
    u128::from_be_bytes(bytes)
}

/// One `stat`, then classify; only files that are actually going to be
/// written get opened, and small text files are finished outright.
fn prepare(path: PathBuf, known: Known<'_>, ctx: &Ctx) -> WalkedFile {
    let db_path = path_to_db_string(&path);
    let digest = path_digest(&db_path);
    let aliased = matches!(known, Known::Exact(_));

    // A name that is not valid UTF-8 cannot be stored in `files.path` and read
    // back as the same file, so there is nothing to hash or text-index. `Skip`
    // rather than an early return with no entry: the caller reads a missing
    // path as "deleted", and this file was seen, not removed.
    if warn_if_unrepresentable(&path) {
        return WalkedFile::skipped(db_path, digest, aliased);
    }

    let Ok(meta) = fs::metadata(&path) else {
        // Seen but unreadable. Emitting it anyway keeps its index row alive:
        // a transient stat failure must not read as "deleted".
        return WalkedFile::skipped(db_path, digest, aliased);
    };
    let Some(mtime) = meta
        .modified()
        .ok()
        .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
    else {
        return WalkedFile::skipped(db_path, digest, aliased);
    };

    let action = match known {
        Known::InDir(rows) => {
            // The name is what these rows are keyed by; it is the last
            // component of the same path `db_path` was built from.
            let name = path
                .file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_default();
            classify_for_indexing(&name, mtime, rows)
        }
        Known::Exact(stored) => classify_by_mtime(stored, mtime),
    };
    let record = match action {
        // Unchanged: never opened, never hashed. This is nearly every file on
        // a re-index, and it is the case that has to stay at one syscall.
        FileIndexAction::Skip => None,
        // `prepare_file_record` gates on `is_file()`, which is what keeps us
        // from opening a FIFO — that would block forever, uninterruptibly.
        _ => prepare_file_record(&db_path, &meta, &ctx.config, &ctx.registry),
    };

    WalkedFile {
        path: db_path,
        action,
        record,
        digest,
        aliased,
    }
}

fn worker(shared: &Shared, ctx: &Ctx, tx: &mpsc::SyncSender<WalkEvent>) {
    while let Some((job, slot)) = shared.take() {
        let _busy = shared.stats.enter();
        if should_abort(&ctx.stop_flag, &ctx.suspend_flag) {
            shared.shutdown();
            return;
        }

        let mut found = Vec::new();
        let mut stale = Vec::new();
        // An alias is a single file with its mtime already resolved; the
        // other two variants are a directory's worth classified against
        // that directory's rows.
        let (files, rows) = match job {
            Job::Dir(dir, rows) => {
                let files = read_directory(&dir, &rows, ctx, &mut found, &mut stale);
                (files, rows)
            }
            Job::Files(files, rows) => (files, rows),
            Job::Alias(path, stored) => {
                slot.finish(found);
                if tx
                    .send(WalkEvent::File(prepare(path, Known::Exact(stored), ctx)))
                    .is_err()
                {
                    shared.shutdown();
                    return;
                }
                continue;
            }
        };

        // Hand the subdirectories over before doing our own per-file work, so
        // the rest of the pool never idles waiting behind one worker. This
        // also confines the job slot to `read_directory`.
        slot.finish(found);

        if !stale.is_empty() && tx.send(WalkEvent::Stale(stale)).is_err() {
            shared.shutdown();
            return;
        }

        for path in files {
            if should_abort(&ctx.stop_flag, &ctx.suspend_flag) {
                shared.shutdown();
                return;
            }
            if tx
                .send(WalkEvent::File(prepare(path, Known::InDir(&rows), ctx)))
                .is_err()
            {
                // Receiver gone: the run was stopped or failed. Not an error.
                shared.shutdown();
                return;
            }
        }
    }
}

/// Serves the pool's directory-row and symlink-mtime lookups from one
/// read-only connection.
///
/// One per walk. The alternative — a connection per worker — would multiply
/// SQLite's page cache by the pool size; see [`PRAGMAS_WALK_READER`]. Every
/// query here is a single index lookup, so one thread stays far ahead of a
/// pool bound by `stat` latency.
///
/// A failed query is not fatal: the job is abandoned rather than retried, and
/// the directory it was for simply goes unwalked, which reconciliation reads
/// as "not seen" and therefore deletes nothing.
fn prefetcher(shared: &Shared, db_path: &str) {
    let conn = match crate::db::open::open_walk_reader(db_path) {
        Ok(conn) => conn,
        Err(e) => {
            // Without rows nothing can be classified, so stopping the walk is
            // the honest outcome: continuing would treat every file as new
            // and every row as stale.
            crate::log_warn!("walk reader: {}", e);
            shared.shutdown();
            return;
        }
    };

    while let Some(work) = shared.take_prefetch() {
        match work {
            PrefetchWork::Dir(dir) => {
                match crate::db::repo::dir_rows(&conn, &path_to_db_string(&dir)) {
                    Ok(rows) => shared.finish_prefetch(Job::Dir(dir, Arc::new(rows))),
                    Err(e) => {
                        crate::log_warn!("{}", e);
                        shared.abandon_prefetch();
                    }
                }
            }
            PrefetchWork::Alias(path) => {
                match crate::db::repo::mtime_for_path(&conn, &path_to_db_string(&path)) {
                    Ok(stored) => shared.finish_prefetch(Job::Alias(path, stored)),
                    Err(e) => {
                        crate::log_warn!("{}", e);
                        shared.abandon_prefetch();
                    }
                }
            }
        }
    }
}

/// A running parallel walk. Iterating it drains finished files; dropping it
/// stops the workers and joins them.
pub struct ParallelWalk {
    rx: Option<mpsc::Receiver<WalkEvent>>,
    handles: Vec<JoinHandle<()>>,
    /// Joined by [`ParallelWalk::finish`] alongside the workers. Held
    /// separately only so a failure to open its connection is attributable.
    prefetch: Option<JoinHandle<()>>,
    shared: Arc<Shared>,
    ctx: Arc<Ctx>,
}

impl ParallelWalk {
    /// Directories that could not be read. Only final once the iterator has
    /// ended, because the channel closes when the last worker exits.
    pub fn unreadable(&self) -> &UnreadableDirs {
        &self.ctx.unreadable
    }

    /// Every canonical directory the walk queued, in `files.parent` spelling.
    ///
    /// The caller's vanished-directory sweep needs this: a directory deleted
    /// wholesale is never read, so nothing reconciles the rows beneath it,
    /// and "was this parent reached at all" is the only way to find them.
    ///
    /// Only meaningful once the walk has finished.
    pub fn seen_dirs(&self) -> HashSet<String> {
        self.shared
            .queue
            .lock()
            .unwrap()
            .seen_dirs
            .iter()
            .map(|d| path_to_db_string(d))
            .collect()
    }

    /// A cheap, cloneable handle for reading worker activity while the
    /// walk's iterator is mutably borrowed by a `for` loop.
    ///
    /// Meaningful only while the walk is running: once the workers exit, the
    /// pool size stays but the busy count is permanently zero.
    pub fn worker_stats(&self) -> WorkerStats {
        self.shared.stats.clone()
    }

    /// Join the workers and report whether every one of them finished
    /// cleanly.
    ///
    /// The caller needs this because a dead worker and a finished worker look
    /// identical from the receiving end: both close the channel, so iteration
    /// simply ends. Treating a panicked walk as a completed one would hand
    /// stale cleanup a partial file set and delete everything the dead workers
    /// never reached.
    pub fn finish(&mut self) -> bool {
        // Dropping the receiver first releases any worker parked in `send`.
        self.rx = None;
        let mut clean = true;
        for handle in self.handles.drain(..) {
            if handle.join().is_err() {
                clean = false;
            }
        }
        // After the workers, so a prefetcher parked waiting for the pool to
        // drain below PREFETCH_AHEAD is already free to observe `done`.
        if let Some(handle) = self.prefetch.take() {
            // `shutdown` is what releases it; without that this would block
            // until the queue emptied on its own.
            self.shared.shutdown();
            if handle.join().is_err() {
                clean = false;
            }
        }
        clean
    }
}

/// Result of a non-blocking pull from a producer pool.
///
/// Shared by every pass the writer loop multiplexes — the walk here and the
/// content pass in [`crate::content`] — so that draining one root's work reads
/// identically whichever pass it came from.
pub enum TryNext<T> {
    Item(T),
    /// Nothing ready right now; the pass is still running.
    Empty,
    /// The pass has ended (all workers exited, for any reason).
    Finished,
}

/// Translate a non-blocking channel pull into [`TryNext`]. `None` is a
/// receiver the owner already dropped, which reads as finished.
pub(crate) fn try_recv_next<T>(rx: Option<&mpsc::Receiver<T>>) -> TryNext<T> {
    match rx {
        None => TryNext::Finished,
        Some(rx) => match rx.try_recv() {
            Ok(item) => TryNext::Item(item),
            Err(mpsc::TryRecvError::Empty) => TryNext::Empty,
            Err(mpsc::TryRecvError::Disconnected) => TryNext::Finished,
        },
    }
}

impl ParallelWalk {
    /// Non-blocking variant of `next`, for callers multiplexing several
    /// walks (the per-root writer loop).
    pub fn try_next(&mut self) -> TryNext<WalkEvent> {
        try_recv_next(self.rx.as_ref())
    }
}

impl Iterator for ParallelWalk {
    type Item = WalkEvent;

    fn next(&mut self) -> Option<WalkEvent> {
        self.rx.as_ref()?.recv().ok()
    }
}

impl Drop for ParallelWalk {
    fn drop(&mut self) {
        self.shared.shutdown();
        // No-op if the caller already called `finish`.
        self.finish();
    }
}

/// Walk `roots` in parallel, yielding every indexable file exactly once per
/// canonical path.
///
/// `workers` is explicit so callers can honour per-root overrides; use
/// [`thread_count_for`] for the storage-appropriate default. Clamped to
/// 1..=64.
/// `db_path` is opened read-only by this walk's row prefetcher; the walk
/// itself never writes.
#[allow(clippy::too_many_arguments)]
pub fn walk_indexable_files(
    roots: &[String],
    follow_symlinks: bool,
    include_hidden: bool,
    ignore: IgnoreSet,
    db_path: &str,
    config: Config,
    registry: Arc<Registry>,
    stop_flag: Arc<AtomicBool>,
    suspend_flag: Arc<AtomicBool>,
    workers: usize,
) -> ParallelWalk {
    let mut queue = Queue::default();
    let mut unresolvable: Vec<PathBuf> = Vec::new();
    for root in roots {
        // Canonicalize here, not just at the caller, so "everything below a
        // root is already canonical" holds however this is called. Without it
        // a non-canonical root would spell every path below it differently
        // from the stored rows: every file would look new *and* every stored
        // row would look stale.
        //
        // Roots themselves are never filtered — the user chose them, so a
        // hidden or ignore-matching root still gets walked.
        match fs::canonicalize(root) {
            Ok(dir) => {
                let dir = PathBuf::from(path_to_db_string(&dir));
                if queue.seen_dirs.insert(dir.clone()) {
                    // Through the prefetcher like any other directory: a root
                    // needs its rows before anything inside it can be
                    // classified.
                    queue.needs_rows.push(dir);
                }
            }
            Err(e) => {
                crate::log_warn!("cannot resolve indexing root {}: {}", root, e);
                // An unmounted or renamed root yields nothing, which is
                // indistinguishable from "all its files were deleted" unless
                // we say so. Recorded here so stale cleanup leaves it alone.
                unresolvable.push(PathBuf::from(root));
            }
        }
    }

    let threads = workers.clamp(1, 64);
    let shared = Arc::new(Shared {
        queue: Mutex::new(queue),
        idle: Condvar::new(),
        stats: WorkerStats::new(threads),
    });
    let ctx = Arc::new(Ctx {
        follow_symlinks,
        include_hidden,
        ignore,
        config,
        registry,
        unreadable: UnreadableDirs::default(),
        stop_flag,
        suspend_flag,
    });

    for root in unresolvable {
        ctx.unreadable.record(root);
    }

    let (tx, rx) = mpsc::sync_channel(CHANNEL_CAP);
    let handles = (0..threads)
        .map(|_| {
            let (shared, ctx, tx) = (shared.clone(), ctx.clone(), tx.clone());
            thread::spawn(move || {
                // Walker threads are the bulk of a run's CPU and I/O; the
                // foreground must stay ahead of them.
                crate::platform::set_background_priority();
                worker(&shared, &ctx, &tx)
            })
        })
        .collect();
    // The workers must hold the only senders, or `recv` never reports the end
    // of the walk and phase 1 hangs forever. The prefetcher deliberately holds
    // none: it produces jobs, not files.
    drop(tx);

    let prefetch = {
        let (shared, db_path) = (shared.clone(), db_path.to_string());
        thread::spawn(move || {
            crate::platform::set_background_priority();
            prefetcher(&shared, &db_path)
        })
    };

    ParallelWalk {
        rx: Some(rx),
        handles,
        prefetch: Some(prefetch),
        shared,
        ctx,
    }
}

/// Pick a worker count for these roots.
///
/// A network share wants far more threads than cores, because each worker
/// spends its time blocked on a round trip rather than on the CPU; a local
/// disk wants few. Users cannot be asked to tune this — the indexer runs on
/// machines we do not configure — so it is detected rather than configured.
/// With a mix of roots the higher count wins: over-threading a local disk
/// costs a little, under-threading a share costs everything.
pub fn thread_count_for(roots: &[String]) -> usize {
    let network = roots
        .iter()
        .any(|r| crate::platform::is_network_path(Path::new(r)));
    if network {
        NETWORK_THREADS
    } else {
        LOCAL_THREADS
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tmp_tree(tag: &str) -> PathBuf {
        let mut p = std::env::temp_dir();
        p.push(format!(
            "quicksearch-pwalk-{}-{}-{}",
            tag,
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        fs::create_dir_all(&p).unwrap();
        p
    }

    fn touch(p: &Path) {
        fs::create_dir_all(p.parent().unwrap()).unwrap();
        fs::write(p, b"x").unwrap();
    }

    /// A database seeded with `rows` as already-indexed files.
    ///
    /// Classification data now comes from SQLite rather than from a map the
    /// caller passes in, so these tests build the state they are testing
    /// against the same way the indexer does.
    fn db_with(tag: &str, rows: &[(String, u64)]) -> PathBuf {
        let mut p = std::env::temp_dir();
        p.push(format!(
            "quicksearch-pwalk-db-{}-{}-{}.sqlite",
            tag,
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let conn = crate::db::open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
        for (path, mtime) in rows {
            let as_path = Path::new(path);
            conn.execute(
                "INSERT INTO files (name, path, parent, size, mtime, type, \
                                    basic_state, content_state)
                 VALUES (?1, ?2, ?3, 0, ?4, 0, 1, 3)",
                rusqlite::params![
                    as_path.file_name().unwrap().to_string_lossy(),
                    path,
                    as_path.parent().unwrap().to_string_lossy(),
                    *mtime as i64,
                ],
            )
            .unwrap();
        }
        p
    }

    fn empty_db(tag: &str) -> PathBuf {
        db_with(tag, &[])
    }

    fn walk(root: &Path, db: &Path) -> Vec<WalkedFile> {
        walk_with(root, db, false, false)
    }

    fn walk_with(
        root: &Path,
        db: &Path,
        follow_symlinks: bool,
        include_hidden: bool,
    ) -> Vec<WalkedFile> {
        files_only(walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            follow_symlinks,
            include_hidden,
            IgnoreSet::compile(&[]).unwrap(),
            db.to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        ))
    }

    /// Drop the reconciliation events; most tests are about which files the
    /// walk reports.
    fn files_only(walk: ParallelWalk) -> Vec<WalkedFile> {
        walk.filter_map(|e| match e {
            WalkEvent::File(f) => Some(f),
            WalkEvent::Stale(_) => None,
        })
        .collect()
    }

    /// Paths the walk decided no longer have a file behind them.
    fn stale_only(walk: ParallelWalk) -> Vec<String> {
        walk.filter_map(|e| match e {
            WalkEvent::Stale(paths) => Some(paths),
            WalkEvent::File(_) => None,
        })
        .flatten()
        .collect()
    }

    fn names(files: &[WalkedFile]) -> Vec<String> {
        let mut n: Vec<String> = files
            .iter()
            .map(|f| {
                Path::new(&f.path)
                    .file_name()
                    .unwrap()
                    .to_string_lossy()
                    .into_owned()
            })
            .collect();
        n.sort();
        n
    }

    #[test]
    fn walks_a_nested_tree_exactly_once() {
        let root = tmp_tree("nested");
        touch(&root.join("a.txt"));
        touch(&root.join("sub/b.txt"));
        touch(&root.join("sub/deep/c.txt"));
        touch(&root.join("other/d.txt"));

        let files = walk(&root, &empty_db("nested"));
        assert_eq!(names(&files), vec!["a.txt", "b.txt", "c.txt", "d.txt"]);

        let unique: HashSet<&String> = files.iter().map(|f| &f.path).collect();
        assert_eq!(unique.len(), files.len(), "no path may be yielded twice");
        fs::remove_dir_all(&root).ok();
    }

    /// A name that is not valid UTF-8 survives `stat` but not the round trip
    /// through `files.path`, so it must be skipped before anything tries to
    /// open it by that string. Unix only: on Windows `OsString` comes from
    /// UTF-16 and there is no way to build the case.
    #[cfg(unix)]
    #[test]
    fn a_non_utf8_name_is_skipped_and_never_prepared() {
        use std::ffi::OsStr;
        use std::os::unix::ffi::OsStrExt;

        let root = tmp_tree("nonutf8");
        touch(&root.join("plain.txt"));
        // 0xFF is not valid UTF-8 anywhere in a sequence, so the name only
        // survives `to_string_lossy` as U+FFFD.
        let bad = root.join(OsStr::from_bytes(b"DRH257\xff~X.MP4"));
        touch(&bad);
        assert!(bad.symlink_metadata().is_ok(), "the file really is on disk");

        let files = walk(&root, &empty_db("nonutf8"));

        // Both are yielded, so neither reads as deleted...
        assert_eq!(files.len(), 2, "the bad name is still reported as seen");
        // ...but only the representable one is prepared for insertion, which
        // is what keeps it out of the hasher and out of FTS.
        let prepared: Vec<&WalkedFile> = files.iter().filter(|f| f.record.is_some()).collect();
        assert_eq!(prepared.len(), 1);
        assert!(prepared[0].path.ends_with("plain.txt"));

        let skipped = files.iter().find(|f| f.record.is_none()).unwrap();
        assert!(matches!(skipped.action, FileIndexAction::Skip));
        assert!(
            skipped.path.contains('\u{FFFD}'),
            "stored spelling is the lossy one"
        );

        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn wide_directory_is_split_across_workers_without_loss() {
        // More than FILES_PER_JOB in one flat directory, so the chunking path
        // and the termination protocol both run under real contention.
        let root = tmp_tree("wide");
        let count = FILES_PER_JOB * 4 + 7;
        for i in 0..count {
            touch(&root.join(format!("f{:05}.txt", i)));
        }

        let files = walk(&root, &empty_db("wide"));
        assert_eq!(files.len(), count, "every file is yielded exactly once");
        let unique: HashSet<&String> = files.iter().map(|f| &f.path).collect();
        assert_eq!(unique.len(), count, "and none is yielded twice");
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn terminates_on_an_empty_root() {
        // The "queue empty at t=0" corner: every worker must observe the walk
        // as finished rather than waiting for work that will never arrive.
        let root = tmp_tree("empty");
        assert!(walk(&root, &empty_db("empty")).is_empty());
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn unchanged_files_are_never_opened() {
        // The property the whole SMB story rests on: a re-index of an
        // unchanged tree must cost one stat per file and no file opens.
        let root = tmp_tree("skip");
        touch(&root.join("a.txt"));
        touch(&root.join("sub/b.txt"));

        let first = walk(&root, &empty_db("skip-first"));
        assert_eq!(first.len(), 2);
        assert!(first.iter().all(|f| f.action == FileIndexAction::Insert));

        let indexed: Vec<(String, u64)> = first
            .iter()
            .map(|f| (f.path.clone(), f.record.as_ref().unwrap().mtime))
            .collect();

        let second = walk(&root, &db_with("skip-second", &indexed));
        assert_eq!(
            second.len(),
            2,
            "unchanged files are still reported as seen"
        );
        for f in &second {
            assert_eq!(f.action, FileIndexAction::Skip);
            assert!(f.record.is_none(), "an unchanged file is never hashed");
        }
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn every_seen_file_is_reported_even_when_it_cannot_be_read() {
        // A path missing from the stream gets its index row deleted, so
        // "couldn't process it" must still be reported as seen.
        let root = tmp_tree("unreadable-file");
        touch(&root.join("fine.txt"));
        let bad = root.join("bad.txt");
        touch(&bad);

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&bad, fs::Permissions::from_mode(0o000)).unwrap();

            let files = walk(&root, &empty_db("unreadable-file"));
            fs::set_permissions(&bad, fs::Permissions::from_mode(0o644)).ok();

            assert_eq!(names(&files), vec!["bad.txt", "fine.txt"]);
            let bad_entry = files.iter().find(|f| f.path.ends_with("bad.txt")).unwrap();
            assert!(bad_entry.record.is_none(), "unopenable, so no record");
        }
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn unreadable_directory_is_recorded_not_silently_empty() {
        let root = tmp_tree("unreadable-dir");
        touch(&root.join("visible.txt"));
        let locked = root.join("locked");
        touch(&locked.join("inside.txt"));

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&locked, fs::Permissions::from_mode(0o000)).unwrap();

            let mut w = walk_indexable_files(
                &[root.to_string_lossy().into_owned()],
                false,
                false,
                IgnoreSet::compile(&[]).unwrap(),
                empty_db("unreadable-dir").to_str().unwrap(),
                Config::default(),
                Arc::new(Registry::default_set()),
                Arc::new(AtomicBool::new(false)),
                Arc::new(AtomicBool::new(false)),
                4,
            );
            let files: Vec<WalkedFile> = w
                .by_ref()
                .filter_map(|e| match e {
                    WalkEvent::File(f) => Some(f),
                    WalkEvent::Stale(_) => None,
                })
                .collect();
            let recorded = !w.unreadable().is_empty();
            let covers = w
                .unreadable()
                .covers(locked.join("inside.txt").to_str().unwrap());

            fs::set_permissions(&locked, fs::Permissions::from_mode(0o755)).ok();

            assert_eq!(names(&files), vec!["visible.txt"]);
            assert!(recorded, "the failure must be recorded");
            assert!(covers, "so rows beneath it survive stale cleanup");
        }
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    #[cfg(unix)]
    fn symlink_loop_terminates() {
        // A hand-rolled walker has none of walkdir's cycle detection; the
        // canonical-directory set is what stands in for it.
        let root = tmp_tree("loop");
        touch(&root.join("real.txt"));
        std::os::unix::fs::symlink(&root, root.join("self_link")).unwrap();

        let files = walk_with(&root, &empty_db("loop"), true, false);
        assert_eq!(names(&files), vec!["real.txt"], "the cycle is visited once");
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    #[cfg(unix)]
    fn symlinked_file_resolves_to_its_target_path() {
        // Preserves the stored spelling: resolving links where they are found
        // is what lets the per-file `canonicalize` go away without re-spelling
        // rows on the next run.
        //
        // The walk reaches this file twice — directly, and through the alias —
        // and reports it twice. That is deliberate: the walker dedupes
        // *directories*, while the caller's `seen_paths` dedupes files, which
        // is where a UNIQUE(path) violation would otherwise come from. What
        // matters here is that both routes agree on the canonical path, so
        // that dedup can work at all.
        let root = tmp_tree("symlink-file");
        touch(&root.join("real/target.txt"));
        fs::create_dir_all(root.join("links")).unwrap();
        std::os::unix::fs::symlink(root.join("real/target.txt"), root.join("links/alias.txt"))
            .unwrap();

        let files = walk_with(&root, &empty_db("symlink-file"), true, false);
        let paths: HashSet<&String> = files.iter().map(|f| &f.path).collect();
        assert_eq!(paths.len(), 1, "both routes report one canonical path");

        let canonical = path_to_db_string(&root.join("real/target.txt").canonicalize().unwrap());
        assert_eq!(
            *paths.into_iter().next().unwrap(),
            canonical,
            "the target, not the alias"
        );

        // The alias itself is still reported, so its row is never mistaken for
        // deleted — it is reported under the *target's* path.
        assert_eq!(files.len(), 2, "seen twice, spelled once");
        assert!(files.iter().any(|f| f.aliased), "the link route is marked");
        fs::remove_dir_all(&root).ok();
    }

    /// The counterpart, and the reason both symlink kinds are gated together:
    /// `filtered_walk` — which the watcher and the incremental path use —
    /// passes `follow_links` to walkdir and follows neither kind. A file link
    /// followed only here would be re-indexed by every full run and never
    /// updated between them, and its target may sit outside every root.
    #[test]
    #[cfg(unix)]
    fn a_file_symlink_is_not_followed_when_links_are_off() {
        let root = tmp_tree("symlink-off");
        touch(&root.join("real/target.txt"));
        fs::create_dir_all(root.join("links")).unwrap();
        std::os::unix::fs::symlink(root.join("real/target.txt"), root.join("links/alias.txt"))
            .unwrap();
        // A target outside the walked tree: with links off it must not be
        // reachable at all.
        let outside = tmp_tree("symlink-off-outside");
        touch(&outside.join("elsewhere.txt"));
        std::os::unix::fs::symlink(outside.join("elsewhere.txt"), root.join("links/out.txt"))
            .unwrap();

        let files = walk_with(&root, &empty_db("symlink-off"), false, false);
        assert_eq!(
            names(&files),
            vec!["target.txt"],
            "only the real file, reached directly"
        );
        assert!(!files.iter().any(|f| f.aliased), "nothing was resolved");

        fs::remove_dir_all(&root).ok();
        fs::remove_dir_all(&outside).ok();
    }

    /// Windows counterpart of the symlink tests: `canonicalize` spells a
    /// junction's target `\\?\C:\…`, and the walker must strip that before
    /// storing — otherwise everything beneath the junction is spelled
    /// differently from the plainly-spelled roots, full-path ignore patterns
    /// never match there, and the canonical-directory dedup fails.
    #[test]
    #[cfg(windows)]
    fn a_followed_junction_stores_plain_paths() {
        let root = tmp_tree("junction");
        touch(&root.join("real").join("target.txt"));
        // Junctions need no privileges, unlike symlinks; still, skip cleanly
        // on filesystems where mklink refuses.
        let made = std::process::Command::new("cmd")
            .args([
                "/C",
                "mklink",
                "/J",
                root.join("jlink").to_str().unwrap(),
                root.join("real").to_str().unwrap(),
            ])
            .status()
            .map(|s| s.success())
            .unwrap_or(false);
        if !made {
            fs::remove_dir_all(&root).ok();
            return;
        }

        let files = walk_with(&root, &empty_db("junction"), true, false);
        for f in &files {
            assert!(
                !f.path.starts_with(r"\\?\"),
                "stored path leaked a verbatim prefix: {}",
                f.path
            );
        }
        // The junction resolves to the same canonical directory the walk
        // reaches directly, so the dedup visits it exactly once. A leaked
        // prefix would spell it twice and report the file twice.
        assert_eq!(names(&files), vec!["target.txt"], "visited once");

        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn hidden_and_ignored_entries_are_pruned() {
        let root = tmp_tree("prune");
        touch(&root.join("keep.txt"));
        touch(&root.join("sub/keep2.txt"));
        touch(&root.join("sub/skip.tmp"));
        touch(&root.join(".hidden/inside.txt"));
        touch(&root.join(".dotfile"));
        touch(&root.join("node_modules/dep/index.js"));

        let ignore =
            IgnoreSet::compile(&["*.tmp".to_string(), "node_modules".to_string()]).unwrap();
        let files: Vec<WalkedFile> = files_only(walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            ignore,
            empty_db("prune").to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        ));
        assert_eq!(names(&files), vec!["keep.txt", "keep2.txt"]);

        let files = walk_with(&root, &empty_db("prune-hidden"), false, true);
        assert_eq!(
            names(&files),
            vec![
                ".dotfile",
                "index.js",
                "inside.txt",
                "keep.txt",
                "keep2.txt",
                "skip.tmp"
            ],
            "include_hidden with no ignore patterns keeps everything"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn a_directory_reports_rows_with_no_file_behind_them() {
        // The per-directory diff, at the level it is computed: one listing
        // against one directory's rows, before any splitting.
        let root = tmp_tree("reconcile");
        touch(&root.join("kept.txt"));
        touch(&root.join("sub/nested.txt"));

        let gone = path_to_db_string(&root.join("removed.txt"));
        let gone_nested = path_to_db_string(&root.join("sub/vanished.txt"));
        let kept = path_to_db_string(&root.join("kept.txt"));
        let db = db_with(
            "reconcile",
            &[
                (gone.clone(), 1),
                (gone_nested.clone(), 1),
                (kept.clone(), 1),
            ],
        );

        let mut stale = stale_only(walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            db.to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        ));
        stale.sort();

        let mut want = vec![gone, gone_nested];
        want.sort();
        assert_eq!(
            stale, want,
            "exactly the rows with no file, from both directories"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    #[cfg(unix)]
    fn an_unreadable_directory_reports_nothing_stale() {
        use std::os::unix::fs::PermissionsExt;

        // A failed read leaves an empty listing, which must never be diffed:
        // every row under it would look deleted.
        let root = tmp_tree("reconcile-locked");
        let locked = root.join("locked");
        touch(&locked.join("inside.txt"));

        let db = db_with(
            "reconcile-locked",
            &[(path_to_db_string(&locked.join("inside.txt")), 1)],
        );

        fs::set_permissions(&locked, fs::Permissions::from_mode(0o000)).unwrap();
        let stale = stale_only(walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            db.to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        ));
        fs::set_permissions(&locked, fs::Permissions::from_mode(0o755)).ok();

        assert!(
            stale.is_empty(),
            "an unreadable directory is not an empty one"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn hidden_root_is_still_walked() {
        // Roots are chosen explicitly, so the hidden rule must not silence one.
        let base = tmp_tree("hidden-root");
        let root = base.join(".config");
        touch(&root.join("app.conf"));

        let files = walk(&root, &empty_db("hidden-root"));
        assert_eq!(names(&files), vec!["app.conf"]);
        fs::remove_dir_all(&base).ok();
    }

    #[test]
    fn stop_flag_ends_the_walk_without_hanging() {
        let root = tmp_tree("stop");
        for i in 0..500 {
            touch(&root.join(format!("f{:04}.txt", i)));
        }

        let stop = Arc::new(AtomicBool::new(true));
        let files: Vec<WalkedFile> = files_only(walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            empty_db("stop").to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            stop,
            Arc::new(AtomicBool::new(false)),
            4,
        ));

        assert!(
            files.len() < 500,
            "an already-stopped walk does not run to completion"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn dropping_the_walk_early_does_not_hang() {
        // Workers blocked in `send` must be released by the receiver going
        // away, or `Drop` would join threads that never wake.
        let root = tmp_tree("early-drop");
        for i in 0..2000 {
            touch(&root.join(format!("sub{}/f{}.txt", i % 10, i)));
        }

        let mut w = walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            empty_db("early-drop").to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        );
        assert!(w.next().is_some());
        drop(w); // must return, not deadlock

        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn overlapping_roots_yield_each_file_once() {
        let root = tmp_tree("overlap");
        touch(&root.join("sub/a.txt"));

        let files: Vec<WalkedFile> = files_only(walk_indexable_files(
            &[
                root.to_string_lossy().into_owned(),
                root.join("sub").to_string_lossy().into_owned(),
            ],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            empty_db("overlap").to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        ));

        assert_eq!(files.len(), 1, "the nested root must not double-index");
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn repeated_walks_agree_on_the_result_set() {
        // The termination protocol is racy by nature; run it enough times
        // under contention that a premature exit would show up.
        let root = tmp_tree("repeat");
        for i in 0..40 {
            touch(&root.join(format!("d{}/f{}.txt", i % 7, i)));
        }
        let expected = names(&walk(&root, &empty_db("determinism-base")));
        assert_eq!(expected.len(), 40);

        for run in 0..30 {
            assert_eq!(
                names(&walk(&root, &empty_db(&format!("determinism-{}", run)))),
                expected,
                "run {}",
                run
            );
        }
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn finish_reports_a_clean_walk_and_is_idempotent() {
        // The caller gates stale-row deletion on this: a walk whose workers
        // died yields a partial file set, and "not seen" would otherwise be
        // read as "deleted".
        let root = tmp_tree("finish");
        touch(&root.join("a.txt"));
        touch(&root.join("sub/b.txt"));

        let mut w = walk_indexable_files(
            &[root.to_string_lossy().into_owned()],
            false,
            false,
            IgnoreSet::compile(&[]).unwrap(),
            empty_db("finish").to_str().unwrap(),
            Config::default(),
            Arc::new(Registry::default_set()),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
            4,
        );
        let files: Vec<WalkedFile> = w
            .by_ref()
            .filter_map(|e| match e {
                WalkEvent::File(f) => Some(f),
                WalkEvent::Stale(_) => None,
            })
            .collect();
        assert_eq!(files.len(), 2);
        assert!(w.finish(), "no worker panicked");
        // Drop calls it again; joining an already-drained handle list must be
        // a no-op rather than a panic.
        assert!(w.finish());
        drop(w);

        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn local_temp_dir_is_not_detected_as_network() {
        let root = tmp_tree("fstype");
        assert_eq!(
            thread_count_for(&[root.to_string_lossy().into_owned()]),
            LOCAL_THREADS
        );
        fs::remove_dir_all(&root).ok();
    }
}