diff --git a/builtin/runtime/fs.lua b/builtin/runtime/fs.lua index 35b25a2..0cf9eb6 100644 --- a/builtin/runtime/fs.lua +++ b/builtin/runtime/fs.lua @@ -106,6 +106,7 @@ end local READ_DIR_OPTS = { supersede = true, tolerant = true } local STAT_OPTS = { supersede = true } +local WALK_TREE_OPTS = { supersede = true } -- Two result shapes, chosen by `opts.tolerant` (dired Q#DR6): -- @@ -130,6 +131,22 @@ function fs.read_dir(path, opts) return build_handle(id) end +-- walk_tree(base [, opts]) -> handle; await -> { , ... } where +-- each entry is the read_dir shape but `name` is a BASE-RELATIVE path +-- ("sub/dir/file.txt") and the listing covers the whole tree as ONE +-- job (issue #233 D3). Symlinks are recorded, never traversed; an +-- unreadable subdirectory is skipped with its subtree; only the root +-- failing to open fails the walk. Directory entries are included +-- (kind "dir") --- consumers that only want files filter on kind. +function fs.walk_tree(base, opts) + if type(base) ~= "string" then + error("pmacs.fs.walk_tree: base must be a string, got " .. type(base)) + end + local key = read_opts(opts, "pmacs.fs.walk_tree", WALK_TREE_OPTS) + local id = async_mod._dispatch_fs_walk_tree(base, key) + return build_handle(id) +end + function fs.stat(path, opts) if type(path) ~= "string" then error("pmacs.fs.stat: path must be a string, got " .. type(path)) diff --git a/builtin/runtime/lsp.lua b/builtin/runtime/lsp.lua index 2ef706c..95104ab 100644 --- a/builtin/runtime/lsp.lua +++ b/builtin/runtime/lsp.lua @@ -1910,22 +1910,33 @@ local function repull_for_attachments(sid, request_fn) end end --- T M4.5 — workspace file watching (workspace/didChangeWatchedFiles). +-- T M4.5 / issue #233 D3 — workspace file watching +-- (workspace/didChangeWatchedFiles). -- -- Servers register watchers dynamically via client/registerCapability. --- pmacs has no kernel file-watch, so each registration runs a polling --- snapshot-diff coroutine: walk the base dir into a { relpath = sig } --- map and, every tick, diff against the previous map to emit per-file --- created/changed/deleted FileEvents (filtered by the glob and the --- WatchKind bitmask), batched into one notification. Coarser than an --- inotify bridge but accurate; a watcher self-cancels when the server --- dies or the capability is unregistered. +-- pmacs has no kernel file-watch, so watching is a polling +-- snapshot-diff — but scheduled, not slept (D3 framing, approved +-- 2026-08-11): one SCAN GROUP per (server, base) owns a retained +-- { relpath = sig } snapshot, and a single `process.after-tick` +-- subscription drives every group's cadence off +-- `pmacs.editor.monotonic_ms` (autosave's Q#AS2 idiom). Waiting +-- allocates no job and holds no pool thread; a due group runs ONE +-- `pmacs.fs.walk_tree` job for the whole tree, diffs in Lua, and +-- routes per-file created/changed/deleted FileEvents through each +-- member watcher's glob and WatchKind mask, deduped into one +-- notification. Quiet scans back the interval off to a cap; any +-- change resets it. Groups retire when their last member leaves or +-- their server dies, cancelling an in-flight walk cooperatively. local FILE_WATCH_INTERVAL_MS = 250 +local FILE_WATCH_BACKOFF_CAP_MS = 4000 -- file_watchers[tostring(sid)][registrationId] = list of watch records --- ({ cancelled = bool, form = "relative"|"absolute", _sleep = handle? }), --- one per glob watcher. +-- ({ cancelled, form = "relative"|"absolute", kind_mask, match_subject, +-- baseline_epoch, group }), one per glob watcher. `baseline_epoch` +-- is nil until the first snapshot whose WALK STARTED after the record +-- joined its group — the registration-epoch rule that keeps a joiner +-- from receiving events for files that predate it. local file_watchers = {} -- WatchKind is a bitmask (Create=1, Change=2, Delete=4); test it @@ -2029,108 +2040,328 @@ local function glob_matcher(glob) end end --- Recursively list files under `base` → { relpath = sig }. `sig` --- folds size+mtime+kind so a content/metadata change flips it. --- Symlinks are recorded, not traversed (loop-safe). Awaits fs --- primitives, so call from inside an async coroutine. -local function scan_tree(base, matches) - local out = {} - local function walk(dir, rel_prefix) - local ok, entries = pcall(function() - return pmacs.fs.read_dir(dir):await() - end) - if not ok or not entries then return end - for _, e in ipairs(entries) do - local rel = (rel_prefix == "") and e.name or (rel_prefix .. "/" .. e.name) - if e.kind == "dir" then - walk(dir .. "/" .. e.name, rel) - elseif matches(rel) then - out[rel] = table.concat({ - tostring(e.size), tostring(e.mtime), - tostring(e.mtime_nsec), tostring(e.kind), - }, "|") - end - end - end - walk(base, "") - return out -end - local FC_CREATED, FC_CHANGED, FC_DELETED = 1, 2, 3 -local function start_file_watcher(sid, base, glob, kind_mask, record) - -- Per LSP, a plain-string glob matches the file's ABSOLUTE path, - -- while a RelativePattern's pattern is relative to its base — the - -- record's `form` (from resolve_watcher) picks the match subject. - -- scan_tree always walks in relative terms; only the string handed - -- to the matcher changes. - local match_glob = glob_matcher(glob) - local matches = match_glob - if record.form == "absolute" then - matches = function(rel) - return match_glob(base .. "/" .. rel) +-- Join a base-relative path under its base. The filesystem root is +-- special-cased the way `dired`'s handler already spells it (the +-- `(dir == "/") and "" or dir` idiom): a naive `base .. "/" .. rel` +-- at `/` yields `//path` — the implementation-defined POSIX spelling +-- — and `file:////path` once a URI wraps it. +local function join_under(base, rel) + if base == "/" then return "/" .. rel end + return base .. "/" .. rel +end + +-- Build a record's match predicate over a base-RELATIVE path. Per LSP +-- (and #234's P1 review), a plain-string glob matches the file's +-- ABSOLUTE path while a RelativePattern's pattern is relative to its +-- base — the record's `form` picks the subject; the walk itself is +-- always relative. +local function make_matcher(base, pat, form) + local match_glob = glob_matcher(pat) + if form == "absolute" then + return function(rel) + return match_glob(join_under(base, rel)) end end - pmacs.async(function() - local prev = scan_tree(base, matches) - while not record.cancelled and server_is_live(sid) do - local sh = pmacs.workers.sleep(FILE_WATCH_INTERVAL_MS) - record._sleep = sh - pcall(function() sh:await() end) - record._sleep = nil - if record.cancelled or not server_is_live(sid) then break end + return match_glob +end - local cur = scan_tree(base, matches) - -- The seam that makes the recheck below WITNESSABLE. `scan_tree` - -- suspends on `read_dir` once per directory, and the race is a - -- cancel arriving during one of those suspensions --- which no - -- arrangement of real timing can be made to happen on demand. - -- Same reason `git.lua` exposes `_deliver_status`: the contract is - -- about an interleaving the caller does not choose. Unset in - -- production, so this costs one nil test per tick. - -- `cur` is handed over so a test can cancel on THE SCAN THAT - -- OBSERVED a given change. Cancelling on any other scan is not a - -- witness: the loop would break at the post-sleep check on the - -- next iteration and emit nothing anyway, so the assertion would - -- pass with the recheck below deleted. - if pmacs.lsp._after_scan_for_tests then - pcall(pmacs.lsp._after_scan_for_tests, record, cur) +-- The event URI for one change, as `finish_group_scan` emits it. A +-- named function rather than an inline concat so the root-boundary +-- witness can drive the PRODUCTION construction at base "/" — a base +-- no fixture can walk for real. +local function watch_change_uri(base, rel) + return file_uri_for(join_under(base, rel)) +end + +-- Exposed for the root-boundary witness (the `_deliver_status` +-- pattern): the exact functions the watcher matches subjects and +-- builds URIs with. Test-only by convention; production never reads +-- them back. +pmacs.lsp._watch_matcher_for_tests = make_matcher +pmacs.lsp._watch_change_uri_for_tests = watch_change_uri + +-- scan_groups[skey .. "\0" .. base] = one scan group per +-- (server, base): the shared snapshot, the members it serves, and the +-- D3 state machine — single-flight (`in_flight`), completion-advanced +-- deadlines (`next_scan_at`), a scan `generation` that rejects stale +-- completions, `rescan_queued` for joins landing mid-walk, the +-- backoff `interval`, and the `failure_reported` dedup latch. +local scan_groups = {} + +local function group_key(skey, base) + return skey .. "\0" .. base +end + +-- Retire a group: forget it and cancel any in-flight walk +-- cooperatively. The walk's completion is then rejected by the +-- identity/generation check in `finish_group_scan`, so a retired +-- group's state can never be written again. +local function retire_group(group) + scan_groups[group.key] = nil + if group.in_flight and group.in_flight.handle then + pcall(function() + group.in_flight.handle:cancel() + end) + end +end + +-- One scan's completion. `scan_members` is the membership captured at +-- scan START — a watcher joining mid-walk is not in it and waits for +-- its queued baseline scan. Three disjoint arms (framing round 3): +-- stale/retired, live non-success, success. +local function finish_group_scan(group, gen, scan_members, ok, result) + -- Stale/retired arm: the group was retired (or superseded under the + -- same key) while the walk was parked. Nothing here may touch a + -- successor's state — #234's P2 recheck, applied at group scope. + if scan_groups[group.key] ~= group then return end + if not group.in_flight or group.in_flight.generation ~= gen then return end + group.in_flight = nil + local now = pmacs.editor.monotonic_ms() + + if not ok then + -- Live non-success arm: no snapshot, no epoch, no emit; the prior + -- snapshot and backoff interval survive. A queued join still gets + -- its immediate baseline attempt; otherwise reschedule normally. + -- `Handle:await()` raises { tag = "cancelled" } (R45) for the + -- intentional `workers.cancel-at-point` outcome, which stays + -- quiet; a failure is reported once per distinct message until a + -- success clears the latch. + local tag = type(result) == "table" and result.tag or nil + if tag ~= "cancelled" then + local msg = type(result) == "table" + and tostring(result.message or result.tag) + or tostring(result) + if group.failure_reported ~= msg then + group.failure_reported = msg + local report = "lsp: file watch scan failed for " .. group.base .. ": " .. msg + if pmacs.error then pcall(pmacs.error, report) end + pcall(pmacs.editor.set_status, report) end - -- RECHECKED AFTER THE SCAN, not only after the sleep (review P2). - -- The coroutine is suspended for most of a tick with `_sleep` - -- already cleared, so a cancel landing there sets `cancelled` and - -- has no sleep to interrupt. Without this line the resumed scan - -- runs on to `did_change_watched_files` below and a SUPERSEDED - -- watcher emits one last batch under its OLD pattern. One batch is - -- enough: it is a wrong-pattern notification the server acts on. - if record.cancelled or not server_is_live(sid) then break end - local changes = {} - for rel, sig in pairs(cur) do - local was = prev[rel] - if was == nil then - if kind_has(kind_mask, 1) then - changes[#changes + 1] = - { uri = file_uri_for(base .. "/" .. rel), type = FC_CREATED } + end + if group.rescan_queued then + group.rescan_queued = false + group.next_scan_at = now + else + group.next_scan_at = now + group.interval + end + return + end + + -- Success arm. + group.failure_reported = nil + local cur = {} + for _, e in ipairs(result) do + if e.kind ~= "dir" then + cur[e.name] = table.concat({ + tostring(e.size), tostring(e.mtime), + tostring(e.mtime_nsec), tostring(e.kind), + }, "|") + end + end + + -- The seam that makes cancel-during-scan WITNESSABLE (#234's P2 + -- device, per member): the race is a cancel landing while the walk + -- job is out, which no arrangement of real timing produces on + -- demand. Unset in production. Handed the snapshot so a test can + -- cancel on THE SCAN THAT OBSERVED a given change; the delivery + -- loop below rechecks `cancelled` per member. + if pmacs.lsp._after_scan_for_tests then + for _, m in ipairs(scan_members) do + pcall(pmacs.lsp._after_scan_for_tests, m, cur) + end + end + + local changes = {} + local prev = group.snapshot + if prev then + -- Diff once; route per member. `bit` is the WatchKind the event + -- needs (Create=1, Change=2, Delete=4). + local diff = {} + for rel, sig in pairs(cur) do + local was = prev[rel] + if was == nil then + diff[#diff + 1] = { rel = rel, type = FC_CREATED, bit = 1 } + elseif was ~= sig then + diff[#diff + 1] = { rel = rel, type = FC_CHANGED, bit = 2 } + end + end + for rel in pairs(prev) do + if cur[rel] == nil then + diff[#diff + 1] = { rel = rel, type = FC_DELETED, bit = 4 } + end + end + -- A member delivers only with a baseline (the registration-epoch + -- rule: no events for files that predate the join) and only while + -- uncancelled. Changes are deduped by (path, type) into the + -- server's single notification. + local emitted = {} + for _, m in ipairs(scan_members) do + if not m.cancelled and m.baseline_epoch ~= nil then + for _, d in ipairs(diff) do + if kind_has(m.kind_mask, d.bit) and m.match_subject(d.rel) then + local key = d.type .. " " .. d.rel + if not emitted[key] then + emitted[key] = true + changes[#changes + 1] = { + uri = watch_change_uri(group.base, d.rel), + type = d.type, + } + end end - elseif was ~= sig and kind_has(kind_mask, 2) then - changes[#changes + 1] = - { uri = file_uri_for(base .. "/" .. rel), type = FC_CHANGED } end end - for rel in pairs(prev) do - if cur[rel] == nil and kind_has(kind_mask, 4) then - changes[#changes + 1] = - { uri = file_uri_for(base .. "/" .. rel), type = FC_DELETED } - end + end + end + + group.snapshot = cur + group.epoch = group.epoch + 1 + -- Baseline assignment: exactly the members captured at scan start + -- with no baseline yet — for them this walk is the first one that + -- STARTED after their join. + for _, m in ipairs(scan_members) do + if not m.cancelled and m.baseline_epoch == nil then + m.baseline_epoch = group.epoch + end + end + + if #changes > 0 then + pcall(pmacs.lsp.did_change_watched_files, group.sid, changes) + group.interval = FILE_WATCH_INTERVAL_MS + else + -- A quiet scan backs off toward the cap; waiting costs nothing + -- under the after-tick cadence, so this bounds scan frequency, + -- not sleep-job length (there are no sleeps). + group.interval = math.min(group.interval * 2, FILE_WATCH_BACKOFF_CAP_MS) + end + + -- Sweep members cancelled outside cancel_watch_records (the seam, + -- or a supersede that raced the walk); an empty group retires. + local live = {} + for _, m in ipairs(group.members) do + if not m.cancelled then live[#live + 1] = m end + end + group.members = live + if #group.members == 0 then + retire_group(group) + return + end + + if group.rescan_queued then + group.rescan_queued = false + group.next_scan_at = now + else + group.next_scan_at = now + group.interval + end +end + +-- Start one walk for `group`. Single-flight is the caller's contract +-- (the tick skips in-flight groups; joins queue instead) — this only +-- stamps the generation and captures the delivery membership. +local function start_group_scan(group) + group.generation = group.generation + 1 + local gen = group.generation + local scan_members = {} + for i, m in ipairs(group.members) do + scan_members[i] = m + end + local handle = pmacs.fs.walk_tree(group.base) + group.in_flight = { + generation = gen, + started_at = pmacs.editor.monotonic_ms(), + handle = handle, + } + pmacs.async(function() + local ok, result = pcall(function() + return handle:await() + end) + finish_group_scan(group, gen, scan_members, ok, result) + end) +end + +-- The cadence (D3 framing; autosave's Q#AS2 idiom). One after-tick +-- subscription drives every group: a clock read and a compare per +-- frame, no job and no pool thread while waiting. Installed once and +-- guarded rather than removed, because `pmacs.hook.remove` does not +-- exist (the P3 gap). +local watch_tick_installed = false +local function ensure_watch_tick() + if watch_tick_installed then return end + watch_tick_installed = true + pmacs.hook.add("process.after-tick", function() + if next(scan_groups) == nil then return end + local now = pmacs.editor.monotonic_ms() + for _, group in pairs(scan_groups) do + if not server_is_live(group.sid) then + retire_group(group) + elseif not group.in_flight and now >= group.next_scan_at then + start_group_scan(group) end - if #changes > 0 then - pcall(pmacs.lsp.did_change_watched_files, sid, changes) - end - prev = cur end end) end +-- Join `record` to its (server, base) group, creating the group on +-- first use. Joins WAKE the group (framing round 2): the deadline +-- pulls to now, or — mid-walk — exactly one immediate follow-up scan +-- is queued, so a joiner's baseline is at most the current walk's +-- remainder plus one walk away, never a backoff cap. The +-- join-triggered scan does not reset the backoff curve; only observed +-- changes do. +local function join_group(sid, skey, base, record) + ensure_watch_tick() + local key = group_key(skey, base) + local group = scan_groups[key] + if not group then + group = { + key = key, + sid = sid, + skey = skey, + base = base, + members = {}, + snapshot = nil, + epoch = 0, + generation = 0, + in_flight = nil, + rescan_queued = false, + next_scan_at = 0, + interval = FILE_WATCH_INTERVAL_MS, + failure_reported = nil, + } + scan_groups[key] = group + end + group.members[#group.members + 1] = record + record.group = group + if group.in_flight then + group.rescan_queued = true + else + group.next_scan_at = 0 + end +end + +-- The workspace directory the registering server actually serves +-- (Q#D3-3): its spec `root_uri` — verbatim, nil when the server never +-- asked for a root — then its `cwd`. Configured roots and custom +-- resolvers (texlab's Q#LX2) are already folded into the spec, which +-- is why this must NOT be a fresh `pmacs.project.detect`. +local function server_workspace_dir(sid) + local skey = tostring(sid) + local ok, rows = pcall(pmacs.lsp.list) + if not ok then return nil end + for _, row in ipairs(rows) do + if tostring(row.id) == skey then + if row.root_uri then + local p = pmacs.lsp.path_for_uri(row.root_uri) + if p then return p end + end + if row.cwd then return row.cwd end + return nil + end + end + return nil +end + -- Resolve a GlobPattern (string | { baseUri, pattern }) to -- (base_dir, pattern, form). The form must travel with the pair: a -- RelativePattern's pattern is relative to its baseUri, and dropping @@ -2152,23 +2383,52 @@ local function resolve_watcher(sid, gp) return pmacs.lsp.path_for_uri(gp.baseUri), gp.pattern or "**", "relative" end if type(gp) == "string" then + local form = (gp:sub(1, 1) == "/") and "absolute" or "relative" + -- Q#D3-3: the server's own workspace first. The + -- attachment-directory guess remains only for a server with + -- neither root_uri nor cwd — and even there it must be + -- DETERMINISTIC (review blocker): `pairs` order is hash order, so + -- the fallback takes the lexicographically smallest attachment + -- directory rather than whichever record iteration yields first. + -- REACHABLE, not defensive (review round 2 corrected round 1's + -- "unreachable" claim here): `pmacs.lsp.spawn` accepts a spec + -- with neither field, and for a MARKERLESS file `ensure_server` + -- adopts such a live server because its `root_uri` and the + -- attach's `key_uri` are both nil — after which any number of + -- buffers in different directories can attach to it. + local dir = server_workspace_dir(sid) + if dir then return dir, gp, form end + local fallback = nil for _, rec in pairs(attachments) do if rec.server == sid and rec.uri then local p = pmacs.lsp.path_for_uri(rec.uri) - local dir = p and p:match("^(.*)/[^/]*$") - if dir then - return dir, gp, (gp:sub(1, 1) == "/") and "absolute" or "relative" - end + local d = p and p:match("^(.*)/[^/]*$") + -- A file at the filesystem root leaves the capture empty. + if d == "" then d = "/" end + if d and (fallback == nil or d < fallback) then fallback = d end end end + if fallback then return fallback, gp, form end end return nil, nil, nil end +-- Cancel records and leave their groups; a group losing its last +-- member retires (its in-flight walk cancelled cooperatively). local function cancel_watch_records(recs) for _, r in ipairs(recs or {}) do r.cancelled = true - if r._sleep then pcall(function() r._sleep:cancel() end) end + local group = r.group + if group then + r.group = nil + for i, m in ipairs(group.members) do + if m == r then + table.remove(group.members, i) + break + end + end + if #group.members == 0 then retire_group(group) end + end end end @@ -2177,20 +2437,29 @@ local function register_file_watchers(sid, registrations) file_watchers[skey] = file_watchers[skey] or {} for _, reg in ipairs(registrations or {}) do if reg.method == "workspace/didChangeWatchedFiles" then - -- Re-registering a live id supersedes it (rust-analyzer does - -- this): cancel the outgoing records first, because the table - -- write below drops the only reference to them and an - -- uncancelled record polls until the server dies. - cancel_watch_records(file_watchers[skey][reg.id]) + local outgoing = file_watchers[skey][reg.id] local recs = {} for _, w in ipairs((reg.registerOptions or {}).watchers or {}) do local base, pat, form = resolve_watcher(sid, w.globPattern) if base and pat then - local r = { cancelled = false, form = form } + local r = { + cancelled = false, + form = form, + kind_mask = w.kind or 7, + match_subject = make_matcher(base, pat, form), + baseline_epoch = nil, + } recs[#recs + 1] = r - start_file_watcher(sid, base, pat, w.kind or 7, r) + join_group(sid, skey, base, r) end end + -- Re-registering a live id supersedes it (rust-analyzer does + -- this, #234's D2). The successors joined FIRST, so a same-base + -- group stays alive across the hand-over — its in-flight walk + -- is not torn down, and the joiners already queued their + -- baseline. An id whose successors watch a different base still + -- retires the old group when its last member leaves here. + cancel_watch_records(outgoing) file_watchers[skey][reg.id] = recs end end @@ -2225,7 +2494,7 @@ end -- family for every attached document so the matching store -- (`pmacs.inlay_hint` / `pmacs.semantic_tokens`) stays fresh. -- * `client/registerCapability` / `client/unregisterCapability` — --- start/stop the file-watch coroutines for any +-- join/leave the file-watch scan groups for any -- `workspace/didChangeWatchedFiles` registration; reply `null`. -- -- Only servers in `attachments` are drained, so a test (or package) diff --git a/docs/active-work.md b/docs/active-work.md index 2166600..cbd139b 100644 --- a/docs/active-work.md +++ b/docs/active-work.md @@ -250,9 +250,58 @@ the same day (`b867f64`), refreshed and re-gated on the merged base. **D3 — the polling cost — is the remainder, and the user has ruled it is next (2026-08-11).** **Branch `lsp-file-watch-d3`** (base `githubsucks/main` @ `add0ba1`; the remote ref is authoritative), with -**framing `docs/lsp-file-watch-d3-framing.md`, revision 4, DRAFT — -review corrections absorbed; awaiting the four user rulings**, committed -at the branch's first commit so it is portable during review. **Review +**framing `docs/lsp-file-watch-d3-framing.md`, revision 4, APPROVED +2026-08-11 with the four rulings adopted as proposed** (honest ⋯N bar; +no exclusions; server root then cwd then attachment fallback; +constants). **IMPLEMENTED on the branch**: `pmacs.fs.walk_tree` (one +cancellable job per scan, eight Rust unit tests), the group scheduler +in `lsp.lua` (after-tick cadence, single-flight state machine with +the round-3 non-success partition, registration epochs, backoff, +retirement), and eighteen acceptance tests — the six #234 tests +byte-unchanged plus twelve witnesses, each mutation-verified. Two +implementation-time facts worth keeping: + +- **Retirement is deliberately double-enforced** (the unregister path + and the post-scan sweep), and the mutation pass proved it: biting + either copy alone is masked by the other; only biting both goes red. + The sweep exists for seam-cancelled members, the unregister path for + idle groups whose next scan may be seconds away. +- **A second pre-commit round found three more** (implementation + review, not framing): mid-walk cancellation was unwitnessed — both + Rust cancel tests pre-cancelled and the acceptance test cancelled a + QUEUED walk, so deleting the internal polls left everything green + (a `cfg(test)` entry hook now cancels at an exact entry boundary + and asserts the walk stopped NEAR it, and the retirement witness + holds a walk in flight across the unregister and asserts the job + settles cancelled); the "unreachable fallback" claim was WRONG — a + manual `pmacs.lsp.spawn` may omit both `cwd` and `root_uri`, and + `ensure_server` adopts such a server for markerless files (nil == + nil), so the deterministic minimum now has a five-directory + through-the-server witness (five, because with two the build's hash + order coincided with the lexicographic answer and the first-pairs + bite survived); and the root-boundary joins gained a witness + through the exported production matcher/URI functions, since no + fixture can walk `/` for real. +- **A pre-commit review round found four blockers**, fixed before + anything was committed: empty-tree cancellation (the entry loops + never run, so a pre-cancelled walk returned empty SUCCESS — the + deletion-storm shape the non-success arm exists to prevent); the + defensive attachment fallback was still `pairs`-order + nondeterministic (now lexicographic-minimum, with its + unreachability-through-production-spawning recorded at the site); a + filesystem-root base joined as `//path` (now `join_under`, dired's + idiom); and two test probes defaulted on error, so a broken pair + could compare equal and lie green (every probe now `expect`s). +- **A stray marker high in the tree re-roots every markerless fixture + under it.** An empty `/tmp/.git` (leftover from the #233 + investigation, since removed by the user) made project detection + root tempdir fixtures at `/tmp`, which under Q#D3-3 the watcher then + faithfully watched. Any machine can grow one; a markerless-fixture + red that looks like a watcher bug may be an ancestor marker. + +Framing and lane were committed +at the branch's first commit so the document stayed portable during +review. **Review round 3 (2026-08-11) found the live group's non-success transition missing**: every job is user-cancellable and `Handle:await()` raises on cancel/failure, so an uncaught result could leave `in_flight` set forever. @@ -285,10 +334,10 @@ became **none** (any unconditional skip deviates from the registered glob contract, and `walk_tree` removes the job-count economics that motivated it), and the cost arithmetic was corrected to this checkout (1,326 jobs/tick for rust-analyzer's six watchers; revised -steady state is zero jobs at idle). Four open rulings (Q#D3-1..4: -the acceptance bar, exclusions, the scan root, knobs vs constants) -block implementation. What was known before framing, verified while -framing D1/D2: +steady state is zero jobs at idle). The four rulings (Q#D3-1..4) +were ADOPTED AS PROPOSED at the 2026-08-11 approval and are recorded +in the framing's status block. What was known before framing, +verified while framing D1/D2: - After #234 the watcher is *correct* but still walks: `walk` recurses unconditionally and `matches` gates only recording, so rust-analyzer diff --git a/docs/lsp-file-watch-d3-framing.md b/docs/lsp-file-watch-d3-framing.md index b23a8d2..66d584c 100644 --- a/docs/lsp-file-watch-d3-framing.md +++ b/docs/lsp-file-watch-d3-framing.md @@ -1,8 +1,12 @@ # LSP file watcher D3 — the polling cost — framing -**Status: revision 4 — DRAFT, review corrections absorbed; awaiting -the user rulings Q#D3-1..4. No implementation may begin from this -document.** +**Status: revision 4 — APPROVED 2026-08-11.** The user's own review +pass (round 3, absorbed below) closed the state machine; on the +soundness confirmation the four rulings were adopted as proposed: +**Q#D3-1** the honest bar — absent at idle, one attributable job per +concurrently due group; **Q#D3-2** no exclusions by default; +**Q#D3-3** server `root_uri` → `cwd` → attachment fallback; +**Q#D3-4** constants, no config keys. Continues issue #233, which stays open until this lane closes it. D1 and D2 — matching correctness and the re-registration leak — merged as @@ -374,7 +378,7 @@ jobs at idle**, with one `walk_tree` job per group for the few milliseconds each scan actually runs — at most every 250 ms under activity and every 4 s at rest, immediately once at registration. -## Open rulings — each blocks implementation +## The rulings — adopted as proposed at approval (2026-08-11) - **Q#D3-1 — the acceptance bar, stated accurately (round 2).** At idle the indicator is **absent** (no running job exists — @@ -382,22 +386,23 @@ activity and every 4 s at rest, immediately once at registration. shows **one attributable job per concurrently due group** — `⋯N` when N (server, base) groups are due on the same frame, each named for its root; a typical single-project session has one group. - Alternative if `⋯1` must be guaranteed: a global scan queue - serializing walks across groups, at the cost of coupling one - server's scan latency to another's tree size. Which bar? -- **Q#D3-2 — exclusions.** Proposed: none by default, with opt-in - exclusion as a documented contract trade (option b) if a user - asks. Confirm, or rule for one of (b)/(c)/(d) above. -- **Q#D3-3 — the scan root.** Proposed: server `root_uri` → server - `cwd` → attached-file directory. This widens the watched tree for - servers with a real root (today it is one attached file's - directory, chosen by hash order) — a behavioural change to a path - real servers exercise. Confirm the order, or rule otherwise. -- **Q#D3-4 — interval, cap, and backoff curve: constants or config - keys.** The D1/D2 framing refused a knob for a defect; with D3 the - cadence becomes a designed mechanism, so keys are defensible — but - more registry surface is coherence cost. Proposed: constants until - someone asks. + **Adopted.** The alternative — a global scan queue guaranteeing + `⋯1` at the cost of coupling one server's scan latency to + another's tree size — was declined. +- **Q#D3-2 — exclusions.** **Adopted: none by default**, with opt-in + exclusion as a documented contract trade (option b) if a user ever + asks. +- **Q#D3-3 — the scan root.** **Adopted: server `root_uri` → server + `cwd` → attached-file directory** (the fallback itself made + deterministic in implementation review: lexicographically smallest + attachment directory). This widens the watched tree for servers + with a real root — a behavioural change to a path real servers + exercise, accepted as such. +- **Q#D3-4 — interval, cap, and backoff curve.** **Adopted: + constants until someone asks.** The D1/D2 framing refused a knob + for a defect; with D3 the cadence is a designed mechanism, so keys + would be defensible — but more registry surface is coherence cost + nobody has yet paid for a reason. ## Verification sketch diff --git a/src/async_runtime.rs b/src/async_runtime.rs index 284d2ee..9310ce3 100644 --- a/src/async_runtime.rs +++ b/src/async_runtime.rs @@ -317,6 +317,10 @@ pub enum JobKind { Parse, /// `dispatch_fs_read_dir` --- directory enumeration ([T M8.1]). FsReadDir, + /// `dispatch_fs_walk_tree` --- one whole-tree enumeration as a + /// single job (issue #233 D3: the watcher's per-directory job + /// storm collapsed into one purpose per scan). + FsWalkTree, /// `dispatch_fs_stat` --- single-path metadata ([T M8.1]). FsStat, /// `dispatch_fs_rename` --- atomic rename ([T M8.1]). @@ -355,6 +359,7 @@ impl JobKind { JobKind::Grep => "grep", JobKind::Parse => "parse", JobKind::FsReadDir => "fs_read_dir", + JobKind::FsWalkTree => "fs_walk_tree", JobKind::FsStat => "fs_stat", JobKind::FsRename => "fs_rename", JobKind::FsChmod => "fs_chmod", @@ -1185,6 +1190,29 @@ impl AsyncRuntime { id } + /// Dispatch a `walk_tree(base)` job: the whole recursive tree as + /// ONE job, entry names base-relative (issue #233 D3). The reply + /// reuses [`ReplyKind::ReadDir`] --- the payload shape is + /// identical and the Lua boundary needs no second conversion; + /// [`JobKind::FsWalkTree`] still separates the two for the + /// `*workers*` label and the purpose string. Cancellation and + /// tolerance are [`crate::fs::walk_tree_blocking`]'s contract. + pub fn dispatch_fs_walk_tree(&self, base: PathBuf, supersede: Option<&str>) -> JobId { + let (id, cancel) = self.allocate(JobSpec { + kind: JobKind::FsWalkTree, + supersede, + stream: None, + resource: None, + purpose: format!("walk_tree {}", base.display()), + }); + let bus = self.workers.clone(); + self.pool.dispatch(move |_pool| { + let kind = run_fs_walk_tree(&cancel, &base); + let _ = bus.send(ASYNC_REPLY_TOPIC, &WorkerReply { job_id: id, kind }); + }); + id + } + /// Dispatch a `stat(path)` job. Returns one [`FsDirEntry`] of /// metadata for `path`. T M8.1. pub fn dispatch_fs_stat(&self, path: PathBuf, supersede: Option<&str>) -> JobId { @@ -1759,6 +1787,16 @@ fn run_fs_read_dir( } } +fn run_fs_walk_tree(cancel: &CancellationToken, base: &Path) -> ReplyKind { + match crate::fs::walk_tree_blocking(base, cancel) { + Ok(listing) => ReplyKind::ReadDir(listing), + Err(FsError::Cancelled) => ReplyKind::Cancelled, + Err(e @ (FsError::Io { .. } | FsError::NonUtf8Path { .. })) => { + ReplyKind::Error(e.to_string()) + } + } +} + fn run_fs_stat(cancel: &CancellationToken, path: &Path) -> ReplyKind { match stat_blocking(path, cancel) { Ok(entry) => ReplyKind::Stat(entry), diff --git a/src/bin/pmacs_fake_lsp.rs b/src/bin/pmacs_fake_lsp.rs index c7393a3..90b1d3c 100644 --- a/src/bin/pmacs_fake_lsp.rs +++ b/src/bin/pmacs_fake_lsp.rs @@ -88,6 +88,8 @@ fn main() { let mut stdout = io::stdout().lock(); let mut crashed_after_init = false; let mut open_docs: HashMap = HashMap::new(); + // `filewatchjoin` / `filewatchretire` mid-session triggers. + let mut didchange_count: u32 = 0; // `fullonly` observability: counts /full responses (rid-1, rid-2…). let mut full_count: u32 = 0; loop { @@ -354,6 +356,58 @@ fn main() { }); write_frame(&mut stdout, &req); } + // Issue #233 D3: `filewatchjoin` registers one watcher at + // initialized and a SECOND one — same base, different + // pattern — on the first didChange it receives. The + // didChange is the test's trigger for a mid-session join, + // which is what the join-wakes / registration-epoch / + // queued-baseline witnesses need and no + // at-initialized-only mode can produce. + ("initialized", _) if mode == "filewatchjoin" => { + let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default(); + let req = serde_json::json!({ + "jsonrpc": "2.0", + "id": 9310, + "method": "client/registerCapability", + "params": { "registrations": [{ + "id": "watch-j1", + "method": "workspace/didChangeWatchedFiles", + "registerOptions": { "watchers": [{ + "globPattern": { + "baseUri": format!("file://{base}"), + "pattern": "**/*.aaa" + }, + "kind": 7 + }] } + }] } + }); + write_frame(&mut stdout, &req); + } + // Issue #233 D3: `filewatchretire` registers at + // initialized, UNREGISTERS on the first didChange (the + // group's last member leaves — retirement), and + // re-registers on the second (a fresh group whose + // baseline folds whatever happened while unwatched). + ("initialized", _) if mode == "filewatchretire" => { + let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default(); + let req = serde_json::json!({ + "jsonrpc": "2.0", + "id": 9320, + "method": "client/registerCapability", + "params": { "registrations": [{ + "id": "watch-r1", + "method": "workspace/didChangeWatchedFiles", + "registerOptions": { "watchers": [{ + "globPattern": { + "baseUri": format!("file://{base}"), + "pattern": "**/*.txt" + }, + "kind": 7 + }] } + }] } + }); + write_frame(&mut stdout, &req); + } // Issue #233 review P1 guard: `filewatchbare` registers a // BARE STRING with no base and no leading `/` — `*.txt`. // The string arm and the `filewatchflat` arm below carry the @@ -563,6 +617,63 @@ fn main() { } } ("textDocument/didOpen" | "textDocument/didChange", _) => { + if method == "textDocument/didChange" { + didchange_count += 1; + let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default(); + // `filewatchjoin` / `filewatchretire`: a didChange + // is the test's mid-session trigger; see the + // `initialized` arms above. + if mode == "filewatchjoin" && didchange_count == 1 { + let req = serde_json::json!({ + "jsonrpc": "2.0", + "id": 9311, + "method": "client/registerCapability", + "params": { "registrations": [{ + "id": "watch-j2", + "method": "workspace/didChangeWatchedFiles", + "registerOptions": { "watchers": [{ + "globPattern": { + "baseUri": format!("file://{base}"), + "pattern": "**/*.bbb" + }, + "kind": 7 + }] } + }] } + }); + write_frame(&mut stdout, &req); + } + if mode == "filewatchretire" && didchange_count == 1 { + let req = serde_json::json!({ + "jsonrpc": "2.0", + "id": 9321, + "method": "client/unregisterCapability", + "params": { "unregisterations": [{ + "id": "watch-r1", + "method": "workspace/didChangeWatchedFiles" + }] } + }); + write_frame(&mut stdout, &req); + } + if mode == "filewatchretire" && didchange_count == 2 { + let req = serde_json::json!({ + "jsonrpc": "2.0", + "id": 9322, + "method": "client/registerCapability", + "params": { "registrations": [{ + "id": "watch-r2", + "method": "workspace/didChangeWatchedFiles", + "registerOptions": { "watchers": [{ + "globPattern": { + "baseUri": format!("file://{base}"), + "pattern": "**/*.txt" + }, + "kind": 7 + }] } + }] } + }); + write_frame(&mut stdout, &req); + } + } let uri = params .get("textDocument") .and_then(|t| t.get("uri")) diff --git a/src/fs.rs b/src/fs.rs index 0e05cbc..6df4f7c 100644 --- a/src/fs.rs +++ b/src/fs.rs @@ -377,6 +377,144 @@ pub fn read_dir_blocking( }) } +/// Recursively walk `base` into ONE flat listing whose entry names are +/// base-RELATIVE paths (`sub/dir/file.txt`). Issue #233 D3: the LSP +/// file-watcher scan used to be one `read_dir` job per directory per +/// tick; this is the whole tree as one job. +/// +/// Contract, in the D3 framing's terms: +/// - **Symlinks are recorded, never traversed.** `lstat` kind +/// `Symlink` regardless of target, so a link cycle cannot loop the +/// walk. A dangling or unreadable link target leaves the entry in +/// the listing with `symlink_target = None`, the same tolerance +/// [`read_dir_blocking`] applies. +/// - **Cancellation is cooperative and prompt**: the token is polled +/// once per directory and every [`READDIR_CANCEL_POLL_EVERY`] +/// entries within one --- the cadence [`read_dir_blocking`] set. +/// - **The root failing to open fails the walk** (the caller's +/// live-failure arm owns it); an unreadable SUBDIRECTORY is skipped +/// with its whole subtree, which is the Lua `scan_tree` `pcall` +/// behaviour this primitive replaces. A non-UTF-8 entry name is +/// skipped the same way: such a path cannot become a watcher URI, +/// and one weird name must not take the walk down. +/// +/// Directory entries appear in the listing (kind `dir`) so a consumer +/// can see structure; the watcher filters them out when building +/// signatures, as its Lua walk always did. +pub fn walk_tree_blocking( + base: &Path, + cancel: &CancellationToken, +) -> Result { + // Checked BEFORE opening and again before returning, not only + // inside the entry loops: an empty tree never enters a loop, so a + // pre-cancelled queued walk would otherwise return an empty + // SUCCESS — which the scheduler's success arm would commit as a + // snapshot and diff into a deletion storm, the exact outcome the + // live non-success arm exists to prevent. (And a missing root + // must report Cancelled, not its Io error, for the same reason.) + if cancel.is_cancelled() { + return Err(FsError::Cancelled); + } + let root = std::fs::read_dir(base).map_err(|source| FsError::Io { + path: base.display().to_string(), + source, + })?; + let mut out: Vec = Vec::new(); + // Subdirectories discovered but not yet walked, with their + // base-relative prefixes. LIFO order --- traversal order is not + // part of the contract; the consumer diffs a map. + let mut pending: Vec<(std::path::PathBuf, String)> = Vec::new(); + walk_one_dir(root, "", cancel, &mut out, &mut pending)?; + while let Some((dir, prefix)) = pending.pop() { + if cancel.is_cancelled() { + return Err(FsError::Cancelled); + } + // Subtree skip on an unreadable directory: scan_tree's pcall. + let Ok(iter) = std::fs::read_dir(&dir) else { + continue; + }; + walk_one_dir(iter, &prefix, cancel, &mut out, &mut pending)?; + } + if cancel.is_cancelled() { + return Err(FsError::Cancelled); + } + Ok(FsDirListing { + entries: out, + errors: None, + }) +} + +#[cfg(test)] +thread_local! { + /// Test-only seam: invoked after every entry [`walk_one_dir`] + /// records. A MID-walk cancellation is an interleaving no real + /// timing produces on demand (the `_after_scan_for_tests` + /// justification, at this layer): the hook lets a test flip the + /// cancel token at an exact entry boundary and assert the walk + /// stopped NEAR it --- which is what discriminates the internal + /// polls from the entry/exit checks alone. `None` outside the one + /// test that arms it. + static WALK_ENTRY_HOOK: std::cell::RefCell>> = + const { std::cell::RefCell::new(None) }; +} + +/// One directory's worth of [`walk_tree_blocking`]: record every +/// representable entry under its base-relative name and queue child +/// directories. Only cancellation propagates as an error --- every +/// per-entry failure is a skip, per the walk's tolerance contract. +fn walk_one_dir( + iter: std::fs::ReadDir, + prefix: &str, + cancel: &CancellationToken, + out: &mut Vec, + pending: &mut Vec<(std::path::PathBuf, String)>, +) -> Result<(), FsError> { + for (i, entry_result) in iter.enumerate() { + if i % READDIR_CANCEL_POLL_EVERY == 0 && cancel.is_cancelled() { + return Err(FsError::Cancelled); + } + let Ok(entry) = entry_result else { continue }; + let Some(name) = entry.file_name().to_str().map(ToOwned::to_owned) else { + continue; + }; + let rel = if prefix.is_empty() { + name + } else { + format!("{prefix}/{name}") + }; + let entry_path = entry.path(); + let Ok(metadata) = std::fs::symlink_metadata(&entry_path) else { + continue; + }; + let kind = classify(&metadata); + if matches!(kind, FsEntryKind::Dir) { + pending.push((entry_path.clone(), rel.clone())); + } + let mut symlink_target = None; + if matches!(kind, FsEntryKind::Symlink) + && let Ok(target) = std::fs::read_link(&entry_path) + { + symlink_target = target.to_str().map(ToOwned::to_owned); + } + out.push(FsDirEntry { + name: rel, + kind, + size: metadata.len(), + mtime_secs: mtime_to_unix_secs(&metadata), + mtime_nsec: mtime_to_unix_nsec(&metadata), + mode: mode_bits(&metadata), + symlink_target, + }); + #[cfg(test)] + WALK_ENTRY_HOOK.with(|h| { + if let Some(hook) = h.borrow_mut().as_mut() { + hook(); + } + }); + } + Ok(()) +} + /// Route one per-entry failure: append it to the tolerant channel, or /// propagate it when the caller asked for the fatal contract. /// @@ -930,4 +1068,185 @@ mod tests { other => panic!("expected NonUtf8Path, got {other:?}"), } } + + #[test] + fn walk_tree_flattens_with_relative_names() { + let td = tempfile::tempdir().expect("tempdir"); + std::fs::write(td.path().join("a.txt"), b"hello").expect("write"); + std::fs::create_dir(td.path().join("sub")).expect("mkdir"); + std::fs::write(td.path().join("sub/b.txt"), b"xy").expect("write"); + std::fs::create_dir(td.path().join("sub/deep")).expect("mkdir"); + std::fs::write(td.path().join("sub/deep/c.txt"), b"z").expect("write"); + let listing = walk_tree_blocking(td.path(), &token()).expect("walk"); + assert!(listing.errors.is_none()); + let mut names: Vec<&str> = listing.entries.iter().map(|e| e.name.as_str()).collect(); + names.sort_unstable(); + assert_eq!( + names, + vec!["a.txt", "sub", "sub/b.txt", "sub/deep", "sub/deep/c.txt"] + ); + let c = listing + .entries + .iter() + .find(|e| e.name == "sub/deep/c.txt") + .unwrap(); + assert_eq!(c.kind, FsEntryKind::File); + assert_eq!(c.size, 1); + let sub = listing.entries.iter().find(|e| e.name == "sub").unwrap(); + assert_eq!(sub.kind, FsEntryKind::Dir); + } + + #[test] + fn walk_tree_records_symlink_without_traversing() { + let td = tempfile::tempdir().expect("tempdir"); + std::fs::write(td.path().join("real.txt"), b"x").expect("write"); + // A link back to the root: traversal would loop forever, so + // completing at all is half the witness. + symlink(td.path(), td.path().join("loop")).expect("symlink"); + let listing = walk_tree_blocking(td.path(), &token()).expect("walk"); + let link = listing.entries.iter().find(|e| e.name == "loop").unwrap(); + assert_eq!(link.kind, FsEntryKind::Symlink); + assert!( + !listing.entries.iter().any(|e| e.name.starts_with("loop/")), + "a symlinked directory must be recorded, never entered" + ); + } + + #[test] + fn walk_tree_skips_unreadable_subdirectory_and_keeps_the_rest() { + use std::os::unix::fs::PermissionsExt; + let td = tempfile::tempdir().expect("tempdir"); + std::fs::write(td.path().join("a.txt"), b"x").expect("write"); + std::fs::create_dir(td.path().join("locked")).expect("mkdir"); + std::fs::write(td.path().join("locked/hidden.txt"), b"x").expect("write"); + std::fs::set_permissions( + td.path().join("locked"), + std::fs::Permissions::from_mode(0o000), + ) + .expect("chmod 000"); + let result = walk_tree_blocking(td.path(), &token()); + // Restore before asserting so the tempdir can be removed even + // if an assertion fails. + std::fs::set_permissions( + td.path().join("locked"), + std::fs::Permissions::from_mode(0o755), + ) + .expect("chmod back"); + let listing = result.expect("walk must survive an unreadable subdirectory"); + let names: Vec<&str> = listing.entries.iter().map(|e| e.name.as_str()).collect(); + assert!(names.contains(&"a.txt")); + assert!( + names.contains(&"locked"), + "the dir entry itself is representable" + ); + assert!( + !names.contains(&"locked/hidden.txt"), + "the unreadable subtree is skipped, matching scan_tree's pcall" + ); + } + + #[test] + fn walk_tree_polls_cancellation_token() { + let td = tempfile::tempdir().expect("tempdir"); + std::fs::write(td.path().join("a.txt"), b"x").expect("write"); + let cancel = token(); + cancel.cancel(); + let err = walk_tree_blocking(td.path(), &cancel).expect_err("must observe cancel"); + assert!(matches!(err, FsError::Cancelled)); + } + + /// Review blocker: the entry loops never run on an EMPTY tree, so + /// without the entry/exit checks a pre-cancelled queued walk + /// returned an empty SUCCESS — which the scheduler would commit as + /// a snapshot and diff into a deletion storm. A missing root must + /// likewise report Cancelled, not its Io error. + #[test] + fn walk_tree_pre_cancelled_is_cancelled_even_with_no_entries() { + let empty = tempfile::tempdir().expect("tempdir"); + let cancel = token(); + cancel.cancel(); + let err = walk_tree_blocking(empty.path(), &cancel) + .expect_err("empty tree must still observe cancel"); + assert!(matches!(err, FsError::Cancelled)); + + let err = walk_tree_blocking(&empty.path().join("absent"), &cancel) + .expect_err("missing root must still observe cancel"); + assert!( + matches!(err, FsError::Cancelled), + "cancellation outranks the root error: got {err:?}" + ); + } + + /// Review blocker: both prior cancellation tests pre-cancelled, so + /// deleting the INTERNAL polls (per directory, and every + /// [`READDIR_CANCEL_POLL_EVERY`] entries) left every test green — + /// the walk ran to completion and only the exit check fired. This + /// cancels MID-walk through the test hook and asserts the walk + /// stopped near the cancellation point, not at the end. + #[test] + fn walk_tree_observes_cancellation_mid_walk() { + let td = tempfile::tempdir().expect("tempdir"); + for d in 0..3 { + let dir = td.path().join(format!("d{d}")); + std::fs::create_dir(&dir).expect("mkdir"); + for i in 0..41 { + std::fs::write(dir.join(format!("f{i:02}.txt")), b"x").expect("write"); + } + } + let cancel = token(); + let seen = std::rc::Rc::new(std::cell::Cell::new(0usize)); + { + let seen = seen.clone(); + let cancel = cancel.clone(); + WALK_ENTRY_HOOK.with(|h| { + *h.borrow_mut() = Some(Box::new(move || { + let n = seen.get() + 1; + seen.set(n); + if n == 5 { + cancel.cancel(); + } + })); + }); + } + let result = walk_tree_blocking(td.path(), &cancel); + WALK_ENTRY_HOOK.with(|h| *h.borrow_mut() = None); + assert!( + matches!(result, Err(FsError::Cancelled)), + "mid-walk cancel must surface as Cancelled" + ); + // 126 entries total (3 dirs + 123 files). The next entry poll + // after the cancel at entry 5 is at most one + // READDIR_CANCEL_POLL_EVERY stride away. + assert!( + seen.get() < 60, + "the walk must stop near the mid-walk cancel, not run the \ + whole tree ({} of 126 entries processed)", + seen.get() + ); + } + + #[test] + fn walk_tree_fails_when_the_root_cannot_open() { + let td = tempfile::tempdir().expect("tempdir"); + let err = walk_tree_blocking(&td.path().join("absent"), &token()) + .expect_err("missing root must fail the walk"); + assert!(matches!(err, FsError::Io { .. })); + } + + #[test] + fn walk_tree_matches_read_dir_on_a_flat_directory() { + let td = tempfile::tempdir().expect("tempdir"); + std::fs::write(td.path().join("a.txt"), b"hello").expect("write"); + std::fs::write(td.path().join("b.md"), b"xy").expect("write"); + let mut walked = walk_tree_blocking(td.path(), &token()) + .expect("walk") + .entries; + let mut listed = read_dir_fatal(td.path(), &token()).expect("read_dir"); + walked.sort_by(|a, b| a.name.cmp(&b.name)); + listed.sort_by(|a, b| a.name.cmp(&b.name)); + // At depth zero the relative name IS the basename, so the two + // primitives must agree entry-for-entry --- the signature + // parity the D3 framing requires. + assert_eq!(walked, listed); + } } diff --git a/src/lua_bindings/mod.rs b/src/lua_bindings/mod.rs index 0a9fa59..f957f44 100644 --- a/src/lua_bindings/mod.rs +++ b/src/lua_bindings/mod.rs @@ -7364,6 +7364,16 @@ pub fn install_async( )?; } + { + let rt = runtime.clone(); + async_mod.set( + "_dispatch_fs_walk_tree", + lua.create_function(move |_, (base, key): (String, Option)| { + Ok(rt.dispatch_fs_walk_tree(std::path::PathBuf::from(base), key.as_deref())) + })?, + )?; + } + { let rt = runtime.clone(); async_mod.set( diff --git a/tests/m4_acceptance.rs b/tests/m4_acceptance.rs index db79602..093d0b6 100644 --- a/tests/m4_acceptance.rs +++ b/tests/m4_acceptance.rs @@ -5674,6 +5674,1003 @@ fn m4_24_bare_string_glob_stays_relative() { ); } +// --------------------------------------------------------------------- +// Issue #233 D3 — the group-scheduler witnesses. Shared scaffold: spawn +// the fake in `mode` with PMACS_FAKE_LSP_WATCH_BASE at `watch_base` +// (the tempdir root unless a test passes a subdirectory), open `a.rs` +// at the tempdir root, and wait for initialization. +// --------------------------------------------------------------------- + +fn d3_scaffold( + mode: &str, + watch_sub: Option<&str>, +) -> ( + tempfile::TempDir, + pmacs::editor::EditorState, + std::path::PathBuf, +) { + use pmacs::editor::EditorState; + + let dir = tempfile::tempdir().expect("tempdir"); + let base = dir.path().to_path_buf(); + let watch_base = match watch_sub { + Some(sub) => { + let p = base.join(sub); + std::fs::create_dir_all(&p).expect("mkdir watch base"); + p + } + None => base.clone(), + }; + let a_path = base.join("a.rs"); + std::fs::write(&a_path, b"fn a() {}\n").expect("write a"); + + let mut state = EditorState::new_with_roots(&crate::iso::roots()); + let fake = fake_lsp_path(); + state + .lua_host + .lua() + .load(format!( + "pmacs.lsp.config.rust = {{ command = '{fake}', + env = {{ PMACS_FAKE_LSP_MODE = '{mode}', + PMACS_FAKE_LSP_WATCH_BASE = '{}' }} }}", + watch_base.display() + )) + .exec() + .expect("override rust config"); + state + .lua_host + .lua() + .load(format!("pmacs.buffer.find_or_open('{}')", a_path.display())) + .exec() + .expect("open a.rs"); + assert!( + pump_lua_flag( + &mut state, + "(function() for _,r in ipairs(pmacs.lsp.list()) do \ + if r.state and r.state.kind=='initialized' then return true end \ + end return false end)()", + 5, + ), + "fake never initialized" + ); + (dir, state, watch_base) +} + +/// Pump every tick for `ms` milliseconds. +fn d3_pump(state: &mut pmacs::editor::EditorState, ms: u64) { + let deadline = Instant::now() + Duration::from_millis(ms); + while Instant::now() < deadline { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// Install the scan-time collector on the group seam. Multi-member +/// scans fire the seam once per member back-to-back, so calls within +/// 40 ms collapse to one recorded scan. +fn d3_install_scan_collector(state: &mut pmacs::editor::EditorState) { + state + .lua_host + .lua() + .load( + "_G.__d3_scans = {} + pmacs.lsp._after_scan_for_tests = function(_, _) + local t = pmacs.editor.monotonic_ms() + local s = _G.__d3_scans + if #s == 0 or t - s[#s] > 40 then s[#s + 1] = t end + end", + ) + .exec() + .expect("install scan collector"); +} + +fn d3_scan_times(state: &pmacs::editor::EditorState) -> Vec { + // `expect`, never a default: a broken probe must be a red test, + // not a green lie (the first idle probe was exactly that). + state + .lua_host + .lua() + .load("return _G.__d3_scans") + .eval::>() + .expect("scan-times probe must not error") +} + +/// Count `fs_walk_tree` jobs currently visible in the workers snapshot +/// (active plus the 64-entry completed ring). +fn d3_walk_job_count(state: &pmacs::editor::EditorState) -> i64 { + state + .lua_host + .lua() + .load( + "(function() + local n = 0 + local snap = pmacs.workers.snapshot() + for _, r in ipairs(snap.active) do + if r.kind == 'fs_walk_tree' then n = n + 1 end + end + for _, r in ipairs(snap.completed) do + if r.kind == 'fs_walk_tree' then n = n + 1 end + end + return n + end)()", + ) + .eval() + .expect("walk-count probe must not error") +} + +/// Pump until at least one `fs_walk_tree` job is active (typically +/// queued behind a saturated pool). Panics on timeout. +fn d3_wait_for_active_walk(state: &mut pmacs::editor::EditorState) { + let deadline = Instant::now() + Duration::from_secs(4); + loop { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let active: i64 = state + .lua_host + .lua() + .load( + "(function() + local n = 0 + for _, r in ipairs(pmacs.workers.snapshot().active) do + if r.kind == 'fs_walk_tree' then n = n + 1 end + end + return n + end)()", + ) + .eval() + .expect("active-walk probe must not error"); + if active >= 1 { + return; + } + assert!( + Instant::now() < deadline, + "no walk dispatched while waiting" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// Pump until some `fs_walk_tree` job is observed cancelled — either +/// `cancel_requested` while active or a `cancelled` completion in the +/// ring. Panics on timeout. +fn d3_wait_for_walk_cancelled(state: &mut pmacs::editor::EditorState) { + let deadline = Instant::now() + Duration::from_secs(4); + loop { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let cancelled: bool = state + .lua_host + .lua() + .load( + "(function() + local snap = pmacs.workers.snapshot() + for _, r in ipairs(snap.active) do + if r.kind == 'fs_walk_tree' and r.cancel_requested then + return true + end + end + for _, r in ipairs(snap.completed) do + if r.kind == 'fs_walk_tree' and r.status == 'cancelled' then + return true + end + end + return false + end)()", + ) + .eval() + .expect("cancel-observation probe must not error"); + if cancelled { + return; + } + assert!( + Instant::now() < deadline, + "the in-flight walk was never cancelled" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// D3 idle witness. With a watcher registered and no file activity, +/// the activity indicator must settle to ABSENT between scans — +/// `activity_summary`'s `None`-at-zero contract — and no `sleep` +/// purpose may ever appear: the old design ran one pool-thread-holding +/// `sleep 250ms` job per watcher per tick, forever, and this witness +/// is unwritable under it. +#[test] +fn m4_24_d3_idle_is_absent_and_never_sleeps() { + let (_dir, mut state, _watch) = d3_scaffold("filewatch", None); + // Let the baseline land and the backoff start stretching. + d3_pump(&mut state, 1200); + + let mut absent_seen = 0u32; + let mut samples = 0u32; + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let s: String = state + .lua_host + .lua() + .load( + "(function() + local s = pmacs._async._activity_summary() + if s == nil then return '' end + return tostring(s.purpose) + end)()", + ) + .eval() + .expect("activity summary probe must not error"); + samples += 1; + if s.is_empty() { + absent_seen += 1; + } + assert!( + !s.contains("sleep"), + "the watcher must not run sleep jobs; activity showed {s:?}" + ); + std::thread::sleep(Duration::from_millis(20)); + } + assert!( + absent_seen > 0, + "activity never settled to absent across {samples} samples — \ + something is running continuously at idle" + ); +} + +/// D3 scan-cost witness. One scan is ONE `fs_walk_tree` job; the +/// per-directory `read_dir` storm is gone. The fixture holds twelve +/// subdirectories, so under the old design a single scan would allocate +/// twelve `read_dir` jobs — here the watcher may allocate none at all. +#[test] +fn m4_24_d3_one_walk_job_per_scan_not_per_directory() { + let dir = tempfile::tempdir().expect("tempdir"); + for i in 0..12 { + let d = dir.path().join(format!("sub{i}")); + std::fs::create_dir(&d).expect("mkdir"); + std::fs::write(d.join("f.txt"), b"x").expect("write"); + } + // Scaffold by hand so the subdirectories exist before the baseline. + let base = dir.path().to_path_buf(); + let a_path = base.join("a.rs"); + std::fs::write(&a_path, b"fn a() {}\n").expect("write a"); + let mut state = pmacs::editor::EditorState::new_with_roots(&crate::iso::roots()); + let fake = fake_lsp_path(); + state + .lua_host + .lua() + .load(format!( + "pmacs.lsp.config.rust = {{ command = '{fake}', + env = {{ PMACS_FAKE_LSP_MODE = 'filewatch', + PMACS_FAKE_LSP_WATCH_BASE = '{}' }} }}", + base.display() + )) + .exec() + .expect("override rust config"); + state + .lua_host + .lua() + .load(format!("pmacs.buffer.find_or_open('{}')", a_path.display())) + .exec() + .expect("open a.rs"); + assert!( + pump_lua_flag( + &mut state, + "(function() for _,r in ipairs(pmacs.lsp.list()) do \ + if r.state and r.state.kind=='initialized' then return true end \ + end return false end)()", + 5, + ), + "fake never initialized" + ); + d3_pump(&mut state, 1500); + + let (walks, watcher_read_dirs): (i64, i64) = state + .lua_host + .lua() + .load(format!( + "(function() + local walks, rds = 0, 0 + local snap = pmacs.workers.snapshot() + local function scan(rows) + for _, r in ipairs(rows) do + if r.kind == 'fs_walk_tree' then walks = walks + 1 end + if r.kind == 'fs_read_dir' + and r.purpose:find('{}', 1, true) then + rds = rds + 1 + end + end + end + scan(snap.active) + scan(snap.completed) + return walks, rds + end)()", + base.display() + )) + .eval() + .expect("count jobs"); + assert!( + walks >= 1, + "at least one walk_tree job must have run; saw {walks}" + ); + assert_eq!( + watcher_read_dirs, 0, + "the watcher must not allocate per-directory read_dir jobs" + ); +} + +/// D3 join-wakes + registration-epoch witness. `filewatchjoin` +/// registers `**/*.aaa` at initialized and `**/*.bbb` on the first +/// didChange. With the group backed off, a file matching the SECOND +/// pattern is created, then the join is triggered: the group must scan +/// immediately (never a backoff cap away), and the joiner must NOT +/// receive a CREATED for the pre-join file — it folds into its +/// baseline, exactly as the initial scan folds pre-existing files. A +/// file created after the baseline settles IS reported. +#[test] +fn m4_24_d3_join_wakes_a_backed_off_group_and_epochs_gate_delivery() { + let (_dir, mut state, watch) = d3_scaffold("filewatchjoin", None); + let received = watch.join(".received"); + d3_install_scan_collector(&mut state); + + // Back off: pump quietly until the last two scans are ≥ 1200 ms + // apart (the curve has left the 250 ms floor well behind). + let deadline = Instant::now() + Duration::from_secs(12); + loop { + d3_pump(&mut state, 120); + let t = d3_scan_times(&state); + if t.len() >= 3 && t[t.len() - 1] - t[t.len() - 2] >= 1200.0 { + break; + } + assert!( + Instant::now() < deadline, + "group never backed off; scan times: {t:?}" + ); + } + + // The discriminating file: matches only the still-unregistered + // second watcher, created BEFORE the join. + std::fs::write(watch.join("pre.bbb"), b"early\n").expect("write pre.bbb"); + let edit_at: f64 = state + .lua_host + .lua() + .load( + "pmacs.window.buffer():insert(0, '-') + pmacs.hook.run('buffer.after-edit') + return pmacs.editor.monotonic_ms()", + ) + .eval() + .expect("edit to trigger join"); + + // The join must wake the backed-off group promptly. + let deadline = Instant::now() + Duration::from_secs(3); + let woke = loop { + d3_pump(&mut state, 40); + let t = d3_scan_times(&state); + if let Some(last) = t.last() + && *last > edit_at + { + break *last - edit_at; + } + if Instant::now() >= deadline { + break f64::INFINITY; + } + }; + assert!( + woke < 1000.0, + "join did not wake the backed-off group (first post-join scan \ + {woke} ms after the edit)" + ); + + // Baseline correctness: a post-baseline file is reported to the + // joiner; the pre-join file never is. + d3_pump(&mut state, 300); + let post_uri = format!("file://{}", watch.join("post.bbb").display()); + std::fs::write(watch.join("post.bbb"), b"late\n").expect("write post.bbb"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {post_uri}"), 10), + "CREATED for post.bbb never reached the joined watcher; \ + .received = {:?}", + std::fs::read_to_string(&received).unwrap_or_default() + ); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("pre.bbb"), + "a file predating the join must fold into the joiner's \ + baseline, never appear as CREATED" + ); +} + +/// D3 queued-baseline witness. A join landing while a walk is IN +/// FLIGHT queues exactly one immediate follow-up scan, and the +/// joiner's baseline is that follow-up. The in-flight state is made +/// real by saturating the worker pool with sleeps so the walk sits +/// queued; the join and a `.bbb` file both land in that window, and +/// the file must fold into the baseline rather than surface as +/// CREATED. A later `.bbb` file is reported — which also proves the +/// follow-up actually ran (without it the joiner would have no +/// baseline and nothing would ever be delivered). +#[test] +fn m4_24_d3_join_mid_walk_queues_one_immediate_baseline() { + let (_dir, mut state, watch) = d3_scaffold("filewatchjoin", None); + let received = watch.join(".received"); + d3_pump(&mut state, 900); + + // Reset the cadence to the floor so the next scan dispatches + // quickly once the pool is saturated: an .aaa event is observed by + // the FIRST watcher. + let trig_uri = format!("file://{}", watch.join("trig.aaa").display()); + std::fs::write(watch.join("trig.aaa"), b"t\n").expect("write trig.aaa"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {trig_uri}"), 8), + "watcher-1 never reported trig.aaa" + ); + + // Saturate the pool; the next due walk queues behind the sleeps. + state + .lua_host + .lua() + .load(format!( + "for _ = 1, {} do pmacs.workers.sleep(1200) end", + std::thread::available_parallelism().map_or(8, std::num::NonZeroUsize::get) + 4 + )) + .exec() + .expect("saturate pool"); + // Wait for the due scan to dispatch (it sits queued, in-flight + // from the group's point of view). + d3_wait_for_active_walk(&mut state); + + // Join lands mid-walk; the discriminating file lands in the same + // window. + state + .lua_host + .lua() + .load( + "pmacs.window.buffer():insert(0, '-') + pmacs.hook.run('buffer.after-edit')", + ) + .exec() + .expect("edit to trigger join"); + std::fs::write(watch.join("mid.bbb"), b"m\n").expect("write mid.bbb"); + + // Drain: sleeps expire, the queued walk runs, the follow-up + // baselines the joiner, and a later file is delivered to it. + d3_pump(&mut state, 1600); + let late_uri = format!("file://{}", watch.join("late.bbb").display()); + std::fs::write(watch.join("late.bbb"), b"l\n").expect("write late.bbb"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {late_uri}"), 10), + "the joiner never received events — the queued baseline scan \ + did not run; .received = {:?}", + std::fs::read_to_string(&received).unwrap_or_default() + ); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("mid.bbb"), + "a file from the join window must fold into the queued \ + baseline, never appear as CREATED" + ); +} + +/// D3 single-flight witness. With completion delivery withheld (the +/// async tick is the delivery path; only the process/LSP ticks run), +/// the group's walk stays in flight while several intervals elapse — +/// and the scheduler must NOT dispatch a second walk. Resuming +/// delivery resumes the cadence. +#[test] +fn m4_24_d3_no_overlap_when_a_walk_outlives_its_interval() { + let (_dir, mut state, _watch) = d3_scaffold("filewatch", None); + d3_pump(&mut state, 900); + + let before = d3_walk_job_count(&state); + assert!(before >= 1, "no walks during warm-up?"); + + // Withhold completions for ~1.3 s (≥ 2 intervals even after one + // backoff step): deadlines pass, the in-flight walk cannot settle, + // and at most ONE new dispatch (the scan due at stop time) may + // appear. + let deadline = Instant::now() + Duration::from_millis(1300); + while Instant::now() < deadline { + state.tick_processes(); + state.tick_lsp(); + std::thread::sleep(Duration::from_millis(10)); + } + let during = d3_walk_job_count(&state); + assert!( + during - before <= 1, + "overlapping walks dispatched while one was in flight: \ + {before} -> {during}" + ); + + // Resume delivery: the cadence recovers. + let deadline = Instant::now() + Duration::from_secs(6); + loop { + d3_pump(&mut state, 150); + if d3_walk_job_count(&state) > during { + break; + } + assert!( + Instant::now() < deadline, + "cadence never resumed after completions were delivered" + ); + } +} + +/// D3 retirement witness. `filewatchretire` unregisters its only +/// watcher on the first didChange — the group's last member leaves, so +/// the group must retire: no further walks are dispatched, and a file +/// created while unwatched is NEVER reported. The second didChange +/// re-registers: a FRESH group whose baseline folds the unwatched-era +/// file; a file created after that is reported. +#[test] +fn m4_24_d3_retirement_stops_scans_and_a_fresh_group_rebaselines() { + let (_dir, mut state, watch) = d3_scaffold("filewatchretire", None); + let received = watch.join(".received"); + d3_pump(&mut state, 900); + + // Watched: f1 is reported. + let f1_uri = format!("file://{}", watch.join("f1.txt").display()); + std::fs::write(watch.join("f1.txt"), b"1\n").expect("write f1"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {f1_uri}"), 8), + "watched file never reported before retirement" + ); + + // Hold a walk genuinely in flight across the unregister (review + // round 2): saturate the pool so the next due walk sits queued, + // then let the unregister land — retirement must CANCEL that job, + // not orphan it. + state + .lua_host + .lua() + .load(format!( + "for _ = 1, {} do pmacs.workers.sleep(1200) end", + std::thread::available_parallelism().map_or(8, std::num::NonZeroUsize::get) + 4 + )) + .exec() + .expect("saturate pool"); + d3_wait_for_active_walk(&mut state); + + // First edit → unregister → retirement, with the walk in flight. + state + .lua_host + .lua() + .load( + "pmacs.window.buffer():insert(0, '-') + pmacs.hook.run('buffer.after-edit')", + ) + .exec() + .expect("edit 1"); + d3_wait_for_walk_cancelled(&mut state); + // Drain the sleeps, then confirm scanning has stopped. + d3_pump(&mut state, 1400); + let after_retire = d3_walk_job_count(&state); + d3_pump(&mut state, 1500); + assert_eq!( + d3_walk_job_count(&state), + after_retire, + "walks kept dispatching after the group's last member left" + ); + + // Unwatched: f2 exists but must never be reported. + std::fs::write(watch.join("f2.txt"), b"2\n").expect("write f2"); + d3_pump(&mut state, 400); + + // Second edit → re-register → fresh group, fresh baseline. + state + .lua_host + .lua() + .load( + "pmacs.window.buffer():insert(0, '-') + pmacs.hook.run('buffer.after-edit')", + ) + .exec() + .expect("edit 2"); + d3_pump(&mut state, 600); + let f3_uri = format!("file://{}", watch.join("f3.txt").display()); + std::fs::write(watch.join("f3.txt"), b"3\n").expect("write f3"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {f3_uri}"), 10), + "the re-registered watcher never reported a fresh file" + ); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("f2.txt"), + "a file created while unwatched must fold into the fresh \ + group's baseline, never appear as CREATED" + ); +} + +/// D3 live-cancel witness. A walk cancelled while its group is still +/// live (the `workers.cancel-at-point` outcome, driven here through +/// `pmacs.async._cancel` on the queued job) must commit nothing: the +/// previous snapshot survives, so a change the cancelled scan would +/// have observed is still reported by the NEXT scan — and no spurious +/// events appear for files already in the baseline. +#[test] +fn m4_24_d3_live_cancel_preserves_snapshot_and_cadence() { + let (_dir, mut state, watch) = d3_scaffold("filewatch", None); + let received = watch.join(".received"); + // bar.txt is in the baseline; it must never produce an event. + std::fs::write(watch.join("bar.txt"), b"b\n").expect("write bar"); + d3_pump(&mut state, 900); + + // Saturate the pool, then create the change and cancel the walk + // that would observe it while it is still queued. + state + .lua_host + .lua() + .load(format!( + "for _ = 1, {} do pmacs.workers.sleep(1200) end", + std::thread::available_parallelism().map_or(8, std::num::NonZeroUsize::get) + 4 + )) + .exec() + .expect("saturate pool"); + std::fs::write(watch.join("foo.txt"), b"f\n").expect("write foo"); + let deadline = Instant::now() + Duration::from_secs(4); + let cancelled: bool = loop { + state.tick_processes(); + state.tick_lsp(); + state.tick_async(); + let did: bool = state + .lua_host + .lua() + .load( + "(function() + for _, r in ipairs(pmacs.workers.snapshot().active) do + if r.kind == 'fs_walk_tree' and not r.cancel_requested then + pmacs._async._cancel(r.id) + return true + end + end + return false + end)()", + ) + .eval() + .expect("cancel probe must not error"); + if did { + break true; + } + if Instant::now() >= deadline { + break false; + } + std::thread::sleep(Duration::from_millis(10)); + }; + assert!(cancelled, "never caught a walk to cancel"); + + // The cancelled scan commits nothing; the next scheduled scan + // reports foo.txt off the PRESERVED snapshot. + let foo_uri = format!("file://{}", watch.join("foo.txt").display()); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {foo_uri}"), 10), + "the change was lost after a live cancel — the prior snapshot \ + did not survive; .received = {:?}", + std::fs::read_to_string(&received).unwrap_or_default() + ); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("bar.txt"), + "a cancelled scan must not commit: baseline files must never \ + produce spurious events" + ); +} + +/// D3 live-failure witness. The watch base is a subdirectory; deleting +/// it makes every walk fail at the root. The failure must be reported +/// ONCE per distinct error (`pmacs.error` is nil in production — the +/// test installs a counter), the group must keep its schedule, and the +/// PRIOR snapshot must survive: recreating the directory yields both a +/// DELETED for the old file and a CREATED for the new one, which is +/// only possible off the retained pre-failure snapshot. +#[test] +fn m4_24_d3_live_failure_reports_once_and_preserves_snapshot() { + let (_dir, mut state, watch) = d3_scaffold("filewatch", Some("watched")); + let received = watch.join(".received"); + std::fs::write(watch.join("foo.txt"), b"f\n").expect("write foo"); + state + .lua_host + .lua() + .load( + "_G.__d3_failures = 0 + pmacs.error = function(msg) + if tostring(msg):find('file watch scan failed', 1, true) then + _G.__d3_failures = _G.__d3_failures + 1 + end + end", + ) + .exec() + .expect("install failure counter"); + d3_pump(&mut state, 900); + + std::fs::remove_dir_all(&watch).expect("remove watch base"); + d3_pump(&mut state, 1800); + let failures: i64 = state + .lua_host + .lua() + .load("return _G.__d3_failures") + .eval() + .expect("read counter"); + assert_eq!( + failures, 1, + "a persistent walk failure must be reported exactly once, not \ + per attempt" + ); + + // Recovery: the retained snapshot yields DELETED foo + CREATED bar. + std::fs::create_dir_all(&watch).expect("recreate watch base"); + std::fs::write(watch.join("bar.txt"), b"b\n").expect("write bar"); + let foo_uri = format!("file://{}", watch.join("foo.txt").display()); + let bar_uri = format!("file://{}", watch.join("bar.txt").display()); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {bar_uri}"), 10), + "the group did not resume scanning after the failure cleared" + ); + assert!( + pump_until_file_contains(&mut state, &received, &format!("3 {foo_uri}"), 6), + "DELETED for the pre-failure file never arrived — the prior \ + snapshot was not preserved across the failures" + ); + let failures_after: i64 = state + .lua_host + .lua() + .load("return _G.__d3_failures") + .eval() + .expect("read counter"); + assert_eq!(failures_after, 1, "recovery must not re-report the failure"); +} + +/// D3 backoff witness. Quiet scans stretch the gap between scans; +/// one observed change snaps it back to the floor. Asserted on scan +/// timestamps from the group seam — there are no sleep purposes to +/// read, because there are no sleeps. +#[test] +fn m4_24_d3_backoff_lengthens_quiet_gaps_and_a_change_resets() { + let (_dir, mut state, watch) = d3_scaffold("filewatch", None); + let received = watch.join(".received"); + d3_install_scan_collector(&mut state); + + // Quiet: collect at least five scans (baseline + four backing off). + let deadline = Instant::now() + Duration::from_secs(12); + loop { + d3_pump(&mut state, 120); + let t = d3_scan_times(&state); + if t.len() >= 5 { + break; + } + assert!(Instant::now() < deadline, "too few scans: {t:?}"); + } + let t = d3_scan_times(&state); + let first_gap = t[1] - t[0]; + let last_gap = t[t.len() - 1] - t[t.len() - 2]; + assert!( + last_gap > first_gap * 2.5, + "quiet gaps must lengthen: first {first_gap} ms, last {last_gap} ms" + ); + + // One change resets the curve. + let z_uri = format!("file://{}", watch.join("z.txt").display()); + std::fs::write(watch.join("z.txt"), b"z\n").expect("write z"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {z_uri}"), 10), + "change never observed" + ); + let observed_at: f64 = *d3_scan_times(&state).last().expect("scan times"); + d3_pump(&mut state, 900); + let t = d3_scan_times(&state); + let post_gap = t + .windows(2) + .find(|w| w[0] >= observed_at) + .map(|w| w[1] - w[0]); + let post_gap = post_gap.expect("no scan followed the change"); + assert!( + post_gap < last_gap / 2.0, + "an observed change must reset the cadence: pre-change gap \ + {last_gap} ms, post-change gap {post_gap} ms" + ); +} + +/// D3 root-boundary witness (review round 2). Both join sites +/// special-case the filesystem root: a naive `base .. "/" .. rel` at +/// `/` yields `//path` — the implementation-defined POSIX spelling — +/// and `file:////path` once a URI wraps it. No fixture can walk `/` +/// for real, so this drives the PRODUCTION matcher and URI builder +/// (exported for tests, the `_deliver_status` pattern) at base `/`: +/// reverting either `join_under` call to concatenation makes the +/// anchored absolute glob refuse `//hit-1.txt` and the URI grow a +/// fourth slash. +#[test] +fn m4_24_d3_root_base_joins_are_root_aware() { + let state = pmacs::editor::EditorState::new_with_roots(&crate::iso::roots()); + let (matched, uri): (bool, String) = state + .lua_host + .lua() + .load( + "(function() + local m = pmacs.lsp._watch_matcher_for_tests( + '/', '/hit-*.txt', 'absolute') + local uri = pmacs.lsp._watch_change_uri_for_tests( + '/', 'hit-1.txt') + return m('hit-1.txt'), uri + end)()", + ) + .eval() + .expect("root-boundary probes must not error"); + assert!( + matched, + "an absolute glob at base `/` must see the subject `/hit-1.txt`, \ + not `//hit-1.txt`" + ); + assert_eq!( + uri, "file:///hit-1.txt", + "an event URI at base `/` must not grow a fourth slash" + ); +} + +/// D3 fallback-determinism witness (review round 2, which also +/// corrected round 1's "unreachable" claim). A manually spawned +/// server may omit BOTH `cwd` and `root_uri`, and for markerless +/// files `ensure_server` adopts it (`root_uri` and `key_uri` both +/// nil) — after which buffers in different directories attach to it. +/// Its bare-string registration must then base at the +/// lexicographically SMALLEST attachment directory, not whichever +/// record `pairs` yields first. `beta` is opened before `alpha` so +/// insertion order disagrees with the lexicographic answer. +#[test] +fn m4_24_d3_fallback_base_is_the_smallest_attachment_dir() { + use pmacs::editor::EditorState; + + let dir = tempfile::tempdir().expect("tempdir"); + let base = dir.path().to_path_buf(); + // Five sibling directories, the lexicographic minimum opened LAST + // — so neither insertion order nor (very probably) hash order + // coincides with the contract's answer. + let alpha = base.join("alpha"); + let beta = base.join("beta"); + for name in ["beta", "gamma", "delta", "zeta", "alpha"] { + let d = base.join(name); + std::fs::create_dir(&d).expect("mkdir"); + std::fs::write(d.join(format!("{name}.rs")), b"fn x() {}\n").expect("write"); + } + let received = base.join(".received"); + + let mut state = EditorState::new_with_roots(&crate::iso::roots()); + let fake = fake_lsp_path(); + state + .lua_host + .lua() + .load(format!( + "pmacs.lsp.config.rust = {{ command = '{fake}' }} + pmacs.lsp.spawn {{ + label = 'manual', language_id = 'rust', command = '{fake}', + env = {{ PMACS_FAKE_LSP_MODE = 'filewatchbare', + PMACS_FAKE_LSP_WATCH_BASE = '{}' }} }}", + base.display() + )) + .exec() + .expect("manual spawn without cwd or root_uri"); + for name in ["beta", "gamma", "delta", "zeta", "alpha"] { + state + .lua_host + .lua() + .load(format!( + "pmacs.buffer.find_or_open('{}')", + base.join(name).join(format!("{name}.rs")).display() + )) + .exec() + .expect("open markerless file"); + } + assert!( + pump_lua_flag( + &mut state, + "(function() for _,r in ipairs(pmacs.lsp.list()) do \ + if r.state and r.state.kind=='initialized' then return true end \ + end return false end)()", + 5, + ), + "fake never initialized" + ); + d3_pump(&mut state, 900); + + // The smaller directory is watched... + let hit_uri = format!("file://{}", alpha.join("hit.txt").display()); + std::fs::write(alpha.join("hit.txt"), b"h\n").expect("write hit"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {hit_uri}"), 10), + "the fallback base must be the lexicographically smallest \ + attachment directory; .received = {:?}", + std::fs::read_to_string(&received).unwrap_or_default() + ); + // ...and the larger one is not. + std::fs::write(beta.join("miss.txt"), b"m\n").expect("write miss"); + d3_pump(&mut state, 900); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("miss.txt"), + "a sibling attachment directory above the chosen base must not \ + be watched" + ); +} + +/// D3 root witness (Q#D3-3). A server with a CONFIGURED root watches +/// that root: the bare-string `*.txt` watcher must match at the +/// configured workspace's top level even though the attached file +/// lives in a subdirectory — under the old attachment-directory +/// guessing, the base would have been `ws/src` and the top-level file +/// unreachable. The subdirectory file pins the other half: a +/// base-level pattern does not match into subdirectories. +#[test] +fn m4_24_d3_configured_root_is_the_bare_string_base() { + use pmacs::editor::EditorState; + + let dir = tempfile::tempdir().expect("tempdir"); + let ws = dir.path().join("ws"); + std::fs::create_dir_all(ws.join("src")).expect("mkdir ws/src"); + let a_path = ws.join("src/a.rs"); + std::fs::write(&a_path, b"fn a() {}\n").expect("write a"); + let received = ws.join(".received"); + + let mut state = EditorState::new_with_roots(&crate::iso::roots()); + let fake = fake_lsp_path(); + state + .lua_host + .lua() + .load(format!( + "pmacs.lsp.config.rust = {{ command = '{fake}', root = '{}', + env = {{ PMACS_FAKE_LSP_MODE = 'filewatchbare', + PMACS_FAKE_LSP_WATCH_BASE = '{}' }} }}", + ws.display(), + ws.display() + )) + .exec() + .expect("override rust config"); + state + .lua_host + .lua() + .load(format!("pmacs.buffer.find_or_open('{}')", a_path.display())) + .exec() + .expect("open a.rs"); + assert!( + pump_lua_flag( + &mut state, + "(function() for _,r in ipairs(pmacs.lsp.list()) do \ + if r.state and r.state.kind=='initialized' then return true end \ + end return false end)()", + 5, + ), + "fake never initialized" + ); + d3_pump(&mut state, 900); + + // Created before the top-level hit so the negative assertion after + // the positive one is race-free (both land in the same or an + // earlier scan). + std::fs::write(ws.join("src/nested.txt"), b"n\n").expect("write nested"); + let hit_uri = format!("file://{}", ws.join("root_hit.txt").display()); + std::fs::write(ws.join("root_hit.txt"), b"r\n").expect("write hit"); + assert!( + pump_until_file_contains(&mut state, &received, &format!("1 {hit_uri}"), 10), + "a bare-string watcher did not watch the CONFIGURED root; \ + .received = {:?}", + std::fs::read_to_string(&received).unwrap_or_default() + ); + assert!( + !std::fs::read_to_string(&received) + .unwrap_or_default() + .contains("nested.txt"), + "a base-level `*.txt` pattern must not match into \ + subdirectories of the configured root" + ); +} + /// Issue #233 D2 — re-registering a live id supersedes it. The /// `filewatchrereg` fake registers `watch-re` TWICE with no /// unregister between — `**/*.old`, then `**/*.new` — exactly the