diff --git a/.github/workflows/watch-soak.yml b/.github/workflows/watch-soak.yml new file mode 100644 index 00000000..0ae4bb4c --- /dev/null +++ b/.github/workflows/watch-soak.yml @@ -0,0 +1,194 @@ +name: Watch soak (manual) + +# MANUAL reproduction instrument for the cli_watch flake family (#129 / #318 / #320). +# +# NOT a gate. `workflow_dispatch` is the only trigger, so this workflow publishes +# zero check-runs on any pull-request head: it cannot enter branch protection and +# cannot appear in scripts/verify-pr-checks.mjs's tally. It touches none of the six +# release-surface paths and does not modify release.yml. +# +# It runs CI's exact command -- `cargo test`, deliberately NOT nextest, because the +# process-per-test model nextest uses is a different execution environment and this +# suite's failures are environment-sensitive. ubuntu-latest only: macOS FSEvents +# cannot reproduce this bug class at all (see .devflow/learning/pitfalls.md PF-026 +# and the project-watch-tests-flaky memory -- a green macOS run proves nothing). +# +# Runs every iteration and tallies, rather than aborting on the first red, because +# the quantity of interest is a RATE. "Failed at iteration 3" cannot distinguish +# 1/20 from 20/20, and the before/after control this instrument exists to serve +# (PF-027 resolution 6) needs both numbers. + +on: + workflow_dispatch: + inputs: + iterations: + description: 'How many times to run the cli_watch suite (1-200)' + type: string + default: '20' + filter: + description: 'Optional cargo-test name filter (empty = the whole suite)' + type: string + default: '' + +permissions: + contents: read + +# cancel-in-progress: false is load-bearing. A soak run is a MEASUREMENT; cancelling +# one halfway leaves a partial tally indistinguishable from a clean run with fewer +# iterations. run_id is in the group key so two deliberate dispatches on the same +# ref never evict each other. +concurrency: + group: watch-soak-${{ github.ref }}-${{ github.run_id }} + cancel-in-progress: false + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: 1 + +jobs: + soak: + name: Watch soak (${{ matrix.label }}) + runs-on: ubuntu-latest + # The loop is bounded by `iterations`, but a hung child inside cargo test is not. + timeout-minutes: 120 + strategy: + fail-fast: false + matrix: + include: + # Default features: the same build CI's `rust` job runs. + - label: default + feature: '' + # The #317 probe widens the publish->arm window to 200ms; N iterations turn + # "6/6 green with the probe on" into a rate. + - label: startup-race-probe + feature: 'startup-race-probe' + steps: + - uses: actions/checkout@v7 + + - uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable + + - uses: Swatinem/rust-cache@v2 + with: + # Per-leg key (PF-041): the two legs build different feature sets. + key: watch-soak-${{ matrix.label }} + + - name: Validate the iterations input + env: + ITERATIONS: ${{ inputs.iterations }} + run: | + set -euo pipefail + case "$ITERATIONS" in + ''|*[!0-9]*) + echo "::error::iterations must be a positive integer; got '$ITERATIONS'" + exit 1 + ;; + esac + if [ "$ITERATIONS" -lt 1 ] || [ "$ITERATIONS" -gt 200 ]; then + echo "::error::iterations must be between 1 and 200; got $ITERATIONS" + exit 1 + fi + echo "iterations validated: $ITERATIONS" + + # Compile once so the loop measures the SUITE and not rustc. + - name: Build the test binary once + env: + FEATURE: ${{ matrix.feature }} + run: | + set -euo pipefail + args=(build -p mds-cli --tests) + if [ -n "$FEATURE" ]; then + args+=(--features "$FEATURE") + fi + echo "cargo ${args[*]}" + cargo "${args[@]}" + + - name: Soak + id: soak + env: + ITERATIONS: ${{ inputs.iterations }} + FILTER: ${{ inputs.filter }} + FEATURE: ${{ matrix.feature }} + LABEL: ${{ matrix.label }} + run: | + # -e is DELIBERATELY omitted: a non-zero `cargo test` is the DATA this + # step collects, not an error that should abort it. -u and pipefail stay. + set -uo pipefail + + mkdir -p soak + + args=(test -p mds-cli --test cli_watch) + if [ -n "$FEATURE" ]; then + args+=(--features "$FEATURE") + fi + if [ -n "$FILTER" ]; then + args+=(-- "$FILTER") + fi + echo "command: cargo ${args[*]}" + + pass=0 + fail=0 + failed_iters="" + i=1 + while [ "$i" -le "$ITERATIONS" ]; do + log="soak/iter-$(printf '%03d' "$i").log" + if cargo "${args[@]}" > "$log" 2>&1; then + pass=$((pass + 1)) + rm -f "$log" + printf 'iter %3d PASS\n' "$i" + else + fail=$((fail + 1)) + failed_iters="$failed_iters $i" + printf 'iter %3d FAIL\n' "$i" + echo "::warning title=watch-soak::iteration $i failed on leg $LABEL" + # Surface the discriminator inline (PF-026: the panic's file:line, not + # the test's name, is the diagnosis). + grep -E 'panicked at|\.\.\. FAILED|test result: FAILED' "$log" || true + fi + i=$((i + 1)) + done + + # The summary file is ALWAYS written, so the artifact is never empty (PF-016). + { + echo "leg: $LABEL" + echo "ref: $GITHUB_REF" + echo "sha: $GITHUB_SHA" + echo "command: cargo ${args[*]}" + echo "iterations: $ITERATIONS" + echo "passed: $pass" + echo "failed: $fail" + echo "failed at: $failed_iters" + } > soak/summary.txt + cat soak/summary.txt + + { + echo "### Watch soak - $LABEL" + echo "" + echo "| metric | value |" + echo "| --- | --- |" + echo "| ref | \`$GITHUB_REF\` |" + echo "| sha | \`$GITHUB_SHA\` |" + echo "| command | \`cargo ${args[*]}\` |" + echo "| iterations | $ITERATIONS |" + echo "| passed | $pass |" + echo "| **failed** | **$fail** |" + if [ "$fail" -gt 0 ]; then + echo "| failing iterations |$failed_iters |" + fi + } >> "$GITHUB_STEP_SUMMARY" + + if [ "$fail" -gt 0 ]; then + echo "::error::$fail of $ITERATIONS iterations failed on leg $LABEL" + exit 1 + fi + echo "clean soak: $pass/$ITERATIONS passed on leg $LABEL" + + - name: Upload failing logs and the tally + if: always() + uses: actions/upload-artifact@v7 + with: + name: watch-soak-${{ matrix.label }}-${{ github.run_id }} + path: soak/ + # `error`, not `ignore`: summary.txt is always written, so an empty upload + # means the glob is wrong, not that the soak was clean (PF-016). + if-no-files-found: error + retention-days: 14 diff --git a/CHANGELOG.md b/CHANGELOG.md index e2d49396..7a06b943 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **`mds watch --debounce` is now a quiet period with a hard cap (#379).** + Each content event restarts the window instead of the window expiring at a fixed + offset from the first event, so a save burst longer than the window coalesces into + one rebuild; the window is bounded by `max(10 x window, 1 s)` and 10 000 events so a + file written to continuously still rebuilds and the idle-tick liveness probe cannot + be starved; raw values are clamped to 60 s (`--debounce 18446744073709551615` + previously watched forever without ever rebuilding); `--debounce 0` still means no + coalescing. No new output. Known cost, in both modes: an event that is not the edit + you care about can still extend an open window, because relevance is not re-derived + per message inside it — in directory mode every event also *opens* one (events under + excluded directories are filtered only afterwards), while in file mode the entry's + parent directory is watched non-recursively, so a sibling scratch write by an editor + extends a window a real edit has already opened. Either way `npm install` churn or a + noisy editor can delay a real edit and the idle tick by up to the cap. + ### Fixed - **Warn on duplicate keys in `--vars` JSON files, at every depth, on every `mds watch` rebuild that writes output (#326).** @@ -60,6 +77,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `build-napi` per-leg rust-cache key: adds `key: ${{ matrix.settings.target }}` to the `Swatinem/rust-cache` step so each cross-compile leg's target artifacts stay isolated (PF-041; without the key all four ubuntu legs and both macOS legs restored one shared blob, confirmed live in run 34065573775); `build-python`'s existing `key: matrix.target-matrix.manylinux` (#347) unchanged; spec S20 in `release-auth-probe.spec.mjs` pins both and fails `Version gate` if a key is dropped; spec S3 extended to pin the `-z` CARGO_REG_TOKEN guard in executable code; #345 verified that crates.io `GET /api/v1/me` is `AuthCheck::only_cookie()` (HTTP 403 for any API token) and the only token-accepting read route rejects scoped tokens — non-empty guard is the strongest check available, durable fix tracked in #368; #345 closed won't-fix-as-filed (#345 #352). - Alpine `node:22-alpine` load tests for both musl napi addons gate `publish-crates`: x64 (`linux-x64-musl`) as the last step of `stage-and-verify-napi` (after the staged artifact upload, so the artifact is never suppressed by an x64 failure), arm64 (`linux-arm64-musl`) in a new unguarded `load-test-musl-arm64` job on a native `ubuntu-24.04-arm` runner using the `napi-staged` artifact; both use `scripts/musl-load-probe.cjs` in a `docker run --network none` step with a positive control; `publish-crates` blocks on both via `needs:` AND its `if:` conjunct (PF-047); spec S21 in `release-auth-probe.spec.mjs` pins job existence, runner, guard shape, wiring, step order, and run-block byte-equality (#340); the first CI run surfaced #371 (string compile fails when the base directory is a filesystem root — `node:22-alpine` has no `WORKDIR` so the default container cwd is `/`); the gate now runs the container from `/w` (`docker run -w /w`) and the probe asserts its cwd so a dropped flag fails loudly. - Both musl napi legs (`x86_64-unknown-linux-musl`, `aarch64-unknown-linux-musl`) now cross-compile with `napi build … -x` / cargo-zigbuild 0.23.0: the SHA-pinned `taiki-e/install-action` (v2.85.10, `fallback: none`) installs cargo-zigbuild before `Swatinem/rust-cache` (rust-cache deletes `~/.cargo/bin` on save; napi's detector is presence-only and would `cargo install` an unpinned copy mid-build otherwise); the hand-written zig cc wrappers, fake-zig self-check, and both `CARGO_TARGET_*_MUSL_LINKER` exports are deleted; three new steps assert the pinned version (before and after the build) and the no-op detector reads both musl linker vars inside `[ -z ]` guards to confirm none is set; the readelf gate adds `ALLOWED_NEEDED='libc\.so|libgcc_s\.so\.1'` with a planted `libunwind.so.1` control; `mlugg/setup-zig` SHA-pinned (v2.2.1) in the same step; spec S22 in `release-auth-probe.spec.mjs` pins all of the above (#339). +- manual `watch-soak.yml` Linux soak instrument for the cli_watch flake family (#129 #318 #320); `workflow_dispatch` only, not a gate, not a required context, not release-surface +- `cli_watch` harness: every post-spawn write to a watched path goes through `common::write_atomic` (temp + rename, one FS event instead of the truncate-then-write pair whose 0-byte intermediate was compiled at `--debounce 0`); 45 sites converted by a mechanical rule stated in the file's doc comment, with two `// DELIBERATE:` plain-write exceptions whose subject IS the truncate+write pair (#318). +- `cli_watch` harness: the pipe drain thread is now joinable — `PipeTap::finish`/`finish_text` reap the child and then JOIN the drain, so the final stderr read carries a happens-before edge to the child's last write; 13 of 13 post-kill flush sleeps deleted and `ChildGuard` moved to `tests/common` so `finish` can name it (#320). +- `cli_watch` harness: a piped stdout is drained *before* the readiness wait, not after (`spawn_watch_ready` returns the tap as a third element; `spawn_ready_piped_stdout` hands it to the caller). `mds watch -o -` publishes its startup output before it writes the readiness marker, so an undrained pipe filled and blocked the child while the poller waited for a marker that could never arrive — reproduced locally as a deterministic 10s `READY_TIMEOUT` failure on 512 KiB of stdout (#320). +- `cli_watch` harness: the i16–i20 duplicate-vars-warning family waits for the expected warning count with a bounded `wait_for_stderr_count` before asserting it. In directory mode the warning is emitted after the output write, so sampling stderr the instant the artifact appeared could read one warning short (CI runs 34366009518, 34404318888) (#326 #320). +- `cli_watch`: `watch_readiness_handshake_makes_ctrl_c_exit_deterministic` is a two-arm control (20 iterations) proving the `MDS_TEST_READY` handshake, not luck, is what makes a post-SIGINT `status.success()` deterministic — unsynchronized spawn signalled on the `Watching …` line dies by SIGINT; a spawn signalled after the handshake exits 0 and prints `Stopped watching.` (#129). +- `cli_build`: `watch_bare_filename_from_cwd_succeeds` is synchronised on the readiness handshake and reads `hello.md` once, instead of polling the output artifact for up to 10s; the private `ChildGuard` copy is replaced by `common::ChildGuard` and stderr is drained rather than discarded (#318). ## [0.4.2] — 2026-09-03 diff --git a/README.md b/README.md index 5b594863..6ba0659e 100644 --- a/README.md +++ b/README.md @@ -113,7 +113,12 @@ Build/Watch options: Watch-only options: --clear Clear terminal before each rebuild (only when stderr is a TTY) - --debounce Debounce window in milliseconds (default: 100) + --debounce Quiet period in milliseconds before a rebuild (default: 100). + Each file change restarts the window, so a save burst longer + than MS still coalesces into a single rebuild. The window is + capped at max(10 × MS, 1000) ms, so a file written to + continuously still rebuilds. 0 disables coalescing (every + event rebuilds). Values above 60000 are clamped. --poll-interval Liveness-probe interval in milliseconds (default: 1000). 0 disables self-heal (native events only). Clamped to ≥50ms. The watcher self-heals after a watched dir/root is deleted and diff --git a/crates/mds-cli/src/main.rs b/crates/mds-cli/src/main.rs index db9b32a3..bb846fcb 100644 --- a/crates/mds-cli/src/main.rs +++ b/crates/mds-cli/src/main.rs @@ -220,8 +220,11 @@ enum Commands { /// Clear the terminal before each rebuild (only when stderr is a TTY) #[arg(long)] clear: bool, - /// Debounce window in milliseconds (default 100; use 0 for immediate rebuilds). - /// Controls how long to wait for burst coalescing after the first event. + /// Quiet period in milliseconds before a rebuild (default 100). + /// Each file change restarts the window, so a save burst longer than MS still + /// coalesces into one rebuild; the window is capped at max(10 x MS, 1000) ms so + /// continuous writes still rebuild. Use 0 to disable coalescing. + /// Values above 60000 are clamped. #[arg(long = "debounce", value_name = "MS", default_value = "100")] debounce: u64, /// Self-heal poll interval in milliseconds (default 1000). diff --git a/crates/mds-cli/src/output.rs b/crates/mds-cli/src/output.rs index 74e4a82a..95e64eec 100644 --- a/crates/mds-cli/src/output.rs +++ b/crates/mds-cli/src/output.rs @@ -1371,6 +1371,25 @@ mod tests { assert_eq!(result, PathBuf::from("/out/page.md")); } + /// The `..tmp--` temp files an atomic write leaves in flight must + /// never be collected as sources. The suffix sits AFTER the `.mds`, so + /// `Path::extension()` is the `tmp-…` component and the walker's extension gate + /// rejects it — the same gate the dir-mode watch filter uses. + /// + /// The second half is the non-vacuity control: a name whose `.mds` is genuinely + /// last IS collected, so the first assertion is not passing on an empty walk. + #[test] + fn collect_mds_files_ignores_write_atomic_temp_names() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("t.mds"), "real").unwrap(); + std::fs::write(dir.path().join(".t.mds.tmp-4242-7"), "in flight").unwrap(); + let files = collect_mds_files(dir.path(), 64, None); + assert_eq!(files.len(), 1, "temp file must not be collected: {files:?}"); + // Non-vacuity: the inverted name IS collected. + std::fs::write(dir.path().join(".tmp-4242-8.t.mds"), "wrong shape").unwrap(); + assert_eq!(collect_mds_files(dir.path(), 64, None).len(), 2); + } + #[test] fn is_partial_detects_underscore_prefix() { assert!(is_partial(Path::new("/dir/_partial.mds"))); diff --git a/crates/mds-cli/src/watch.rs b/crates/mds-cli/src/watch.rs index acc66f29..88a86569 100644 --- a/crates/mds-cli/src/watch.rs +++ b/crates/mds-cli/src/watch.rs @@ -29,6 +29,16 @@ //! The tick is scheduled against an absolute deadline (`TickClock`), so a stream of //! filesystem events cannot postpone the backstop indefinitely (#319). //! +//! # Coalescing +//! +//! `--debounce` is a **quiet period**: the first relevant event opens a window and +//! every further content event restarts it, so a save burst longer than the window is +//! still one rebuild. The window is itself bounded — by an absolute cap of +//! `max(10 x window, 1s)` measured from the first event, and by 10 000 drained +//! messages — because the loop does not consult the idle tick while a window is open, +//! so an unbounded window would starve the content backstop as well as the rebuild +//! (#379). +//! //! # Key invariants //! //! - All content output → stdout ONLY when output resolves to stdout. @@ -36,7 +46,9 @@ //! - `--quiet` suppresses status + warnings but NOT compile errors. //! - Exit 0 on clean Ctrl+C; non-zero only on startup failure. //! - Compile errors during watching never terminate the watcher. -//! - All loops have fixed upper bounds (ADR-021 / reliability.md). +//! - All loops have fixed upper bounds (ADR-021 / reliability.md): the idle tick +//! against an absolute deadline, and the debounce window against an absolute cap +//! (window <= cap) and a message bound (<= 10 000 per window). //! - All `.mds` reads go through `compile_to_content` (PF-004). use std::collections::{BTreeSet, HashMap, HashSet}; @@ -592,48 +604,193 @@ impl TickClock { // ── Debounce loop ───────────────────────────────────────────────────────────── -/// Drain the channel for `debounce_ms` milliseconds, collecting all changed paths. +/// Largest accepted `--debounce` window; larger values are clamped to it. /// -/// Returns `(paths, interrupted)`. -/// - `paths`: all file paths seen in notify events during the window. -/// - `interrupted`: true if an Interrupt message was received. +/// `Instant::now() + Duration::from_millis(u64::MAX)` does not overflow on the +/// i64-second monotonic clocks of macOS and Linux: the deadline lands roughly 585 +/// million years out, so an unclamped `--debounce 18446744073709551615` watches +/// forever and silently never rebuilds (observed). 60s is orders of magnitude past +/// any editor save burst. +const MAX_DEBOUNCE_MS: u64 = 60_000; + +/// Absolute cap on one debounce window, as a multiple of the window. +const DEBOUNCE_CAP_FACTOR: u32 = 10; + +/// Floor under the absolute cap. /// -/// The loop is bounded: it ends when `Instant::now() >= deadline` or when -/// `interrupted` is true. -fn drain_debounce(rx: &mpsc::Receiver, debounce_ms: u64) -> (BTreeSet, bool) { - let mut paths = BTreeSet::new(); +/// Matches the default `--poll-interval`: while a window is open the loop never +/// reaches `TickClock::recv_next`, so this floor is also the bound on how late the +/// idle-tick backstop can run under a continuous event stream. +const DEBOUNCE_CAP_FLOOR: Duration = Duration::from_millis(1_000); + +/// Upper bound on the messages one window will drain. +/// +/// The cap bounds the window's DURATION; this bounds its work and its memory. A +/// sender faster than the drain would otherwise grow `paths` without limit inside a +/// single window. Messages left in the channel are not lost: the caller's next +/// event opens a new window and drains them. +const MAX_DEBOUNCE_MESSAGES: usize = 10_000; + +/// Why a debounce window ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum DebounceEnd { + /// `--debounce 0`: coalescing is off; no window was ever opened. + Disabled, + /// A full window passed with no further content event: the intended exit. + Quiet, + /// The absolute cap elapsed while events were still arriving. + Cap, + /// `MAX_DEBOUNCE_MESSAGES` messages were drained in this window. + MessageLimit, + /// Ctrl+C. The caller must stop, not rebuild. + Interrupted, + /// The watcher's sender was dropped. + Disconnected, +} + +/// Result of one debounce window. +struct DebounceOutcome { + /// Every path seen in a content event during the window. + paths: BTreeSet, + /// Why the window ended. + end: DebounceEnd, +} + +impl DebounceOutcome { + fn interrupted(&self) -> bool { + self.end == DebounceEnd::Interrupted + } +} + +/// Convert a raw `--debounce` value (milliseconds) into a quiet-period window. +/// +/// - `0` -> `None`: coalescing disabled, every event rebuilds immediately. +/// - nonzero -> `Some(min(value, MAX_DEBOUNCE_MS))`. +/// +/// Extracted so the clamp contract is verifiable without the watch loop, exactly as +/// [`clamp_poll_interval`] is. +fn clamp_debounce(debounce_ms: u64) -> Option { if debounce_ms == 0 { - return (paths, false); + None + } else { + Some(Duration::from_millis(debounce_ms.min(MAX_DEBOUNCE_MS))) } - let deadline = Instant::now() + Duration::from_millis(debounce_ms); - loop { +} + +/// Absolute bound on one debounce window: `max(10 x window, 1s)`. +/// +/// `window * DEBOUNCE_CAP_FACTOR` cannot overflow `Duration`: [`clamp_debounce`] caps +/// the window at 60s, so the product is at most 600s. +fn debounce_cap(window: Duration) -> Duration { + (window * DEBOUNCE_CAP_FACTOR).max(DEBOUNCE_CAP_FLOOR) +} + +/// Coalesce a burst of filesystem events into one rebuild. +/// +/// # Quiet period, not a fixed window +/// +/// The first relevant event opens a window of `debounce_ms`; every further **content** +/// event restarts it. A window that expired at a fixed offset from the FIRST event +/// split any burst longer than `debounce_ms` across two or three windows and rebuilt +/// once per window, each compile seeing a different intermediate state of the file: +/// visible as three `Recompiled` lines from one ten-write burst on a loaded CI runner. +/// The size of the burst a user can produce is not a property `debounce_ms` can +/// predict; the size of the GAP between saves is. +/// +/// # Why the cap is mandatory +/// +/// An extendable window with no bound is unbounded: a file written to continuously +/// postpones its own rebuild for as long as the writing lasts. Worse, the idle-tick +/// liveness probe is not consulted while a window is open ([`TickClock::recv_next`] is +/// only reached between batches), so an endless stream would starve the content +/// backstop through a door the absolute tick deadline does not cover. The cap +/// (`max(10 x window, 1s)`) bounds both: the rebuild, and the probe behind it. +/// +/// # What does NOT extend +/// +/// `Access` events (inotify reads; see [`is_content_event`]) and watch errors. The +/// compile reads its own sources, so an extending `Access` event would let the watcher +/// hold its own window open. +/// +/// Relevance is deliberately NOT filtered here. An editor's atomic save writes a temp +/// file and renames it; that temp path is in no watch set, and ending the window on it +/// would split the very burst this exists to coalesce. Relevance decides whether to +/// rebuild ([`event_is_relevant`] in file mode, the `.mds`/root filter in dir mode); +/// this decides when. +fn drain_debounce(rx: &mpsc::Receiver, debounce_ms: u64) -> DebounceOutcome { + let mut paths = BTreeSet::new(); + + let Some(window) = clamp_debounce(debounce_ms) else { + // `--debounce 0`: no coalescing. The channel is left untouched, so the next + // event is delivered to the loop as its own batch. + return DebounceOutcome { + paths, + end: DebounceEnd::Disabled, + }; + }; + + let start = Instant::now(); + let hard_cap = start + debounce_cap(window); + let mut deadline = start + window; + let mut messages: usize = 0; + + let end = loop { + // The bound, enforced in release too: a window may be extended by further + // events, never past `start + cap`. Pure arithmetic: a descheduled runner + // cannot trip it, only a defect can. Asserting on MEASURED elapsed time + // instead would panic a shipped watcher whenever `recv_timeout` overshoots. + assert!( + deadline <= hard_cap, + "debounce deadline escaped its cap: a file written to continuously would \ + postpone its own rebuild (and the idle-tick backstop behind it) for as \ + long as the writing lasts" + ); + + if messages >= MAX_DEBOUNCE_MESSAGES { + break DebounceEnd::MessageLimit; + } + let now = Instant::now(); if now >= deadline { - break; + break if deadline == hard_cap { + DebounceEnd::Cap + } else { + DebounceEnd::Quiet + }; } - let remaining = deadline - now; - match rx.recv_timeout(remaining) { - Ok(Msg::Fs(Ok(event))) => { - // Drop Access events (inotify IN_ACCESS/IN_OPEN/IN_CLOSE_NOWRITE) - // — reads must not trigger recompiles; see is_content_event. - if is_content_event(&event.kind) { - for p in event.paths { - paths.insert(p); + + match rx.recv_timeout(deadline - now) { + Ok(msg) => { + messages += 1; + match msg { + Msg::Fs(Ok(event)) => { + // Drop Access events (inotify IN_ACCESS/IN_OPEN/IN_CLOSE_NOWRITE) + // — reads must not trigger recompiles; see is_content_event. + if !is_content_event(&event.kind) { + continue; + } + for p in event.paths { + paths.insert(p); + } + deadline = (Instant::now() + window).min(hard_cap); } + Msg::Fs(Err(e)) => { + eprint_warning(&format!( + "warning: watch error during debounce: {}", + safe_inline(&e) + )); + } + Msg::Interrupt => break DebounceEnd::Interrupted, } } - Ok(Msg::Fs(Err(e))) => { - eprint_warning(&format!( - "warning: watch error during debounce: {}", - safe_inline(&e) - )); - } - Ok(Msg::Interrupt) => return (paths, true), - Err(mpsc::RecvTimeoutError::Timeout) => break, - Err(mpsc::RecvTimeoutError::Disconnected) => break, + // The deadline is the single decision point: re-loop and let the checks + // above classify the exit. + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => break DebounceEnd::Disconnected, } - } - (paths, false) + }; + + DebounceOutcome { paths, end } } // ── Poll-interval clamp (ADR-021) ───────────────────────────────────────────── @@ -908,8 +1065,9 @@ fn handle_fs_event_file( } // Drain the debounce window. - let (_extra_paths, interrupted2) = drain_debounce(rx, debounce_ms); - if interrupted2 { + // The drained paths are discarded: file mode has already decided relevance above + // and rebuilds its single entry regardless of which path moved. + if drain_debounce(rx, debounce_ms).interrupted() { return FileEventAction::Stop; } @@ -1932,11 +2090,11 @@ fn handle_fs_event_dir( } // Drain debounce window. - let (extra, interrupted2) = drain_debounce(rx, ctx.debounce_ms); - changed.extend(extra); - if interrupted2 { + let drained = drain_debounce(rx, ctx.debounce_ms); + if drained.interrupted() { return DirEventOutcome::Stop; } + changed.extend(drained.paths); // Defense-in-depth: ignore events from inside the out-dir subtree. if let OutputBase::Dir(ref od) = ctx.output_base { @@ -3402,6 +3560,355 @@ mod tests { ); } + // ── Debounce window domain (#379) ──────────────────────────────────────── + + /// A minimal content event on `path`, shaped like the ones notify delivers. + fn modify_event(path: &str) -> Msg { + Msg::Fs(Ok(notify::Event { + kind: notify::EventKind::Modify(notify::event::ModifyKind::Any), + paths: vec![PathBuf::from(path)], + attrs: Default::default(), + })) + } + + /// A read event — the kind `is_content_event` drops. + fn access_event(path: &str) -> Msg { + Msg::Fs(Ok(notify::Event { + kind: notify::EventKind::Access(notify::event::AccessKind::Read), + paths: vec![PathBuf::from(path)], + attrs: Default::default(), + })) + } + + /// Run `drain_debounce` on a worker thread and refuse to wait past `bound`. + /// + /// The function under test is meant to be bounded. A mutation that removes the + /// bound would otherwise hang the test binary until the harness's own timeout, + /// which reports as an infrastructure problem rather than as a failed contract. + /// Collecting the result through a `recv_timeout` turns that mutation into a + /// clean, named failure at `bound`. + /// + /// The worker thread is deliberately not joined on the timeout path: it is + /// blocked precisely because the bound it should have honoured is gone, so + /// joining it would reintroduce the hang this exists to prevent. + fn drain_bounded( + rx: mpsc::Receiver, + debounce_ms: u64, + bound: Duration, + why: &str, + ) -> DebounceOutcome { + let (done_tx, done_rx) = mpsc::channel(); + std::thread::spawn(move || { + let outcome = drain_debounce(&rx, debounce_ms); + // The receiver may already have given up; the send failing is fine. + let _ = done_tx.send(outcome); + }); + done_rx.recv_timeout(bound).unwrap_or_else(|e| { + panic!("drain_debounce did not return within {bound:?} ({e:?}): {why}") + }) + } + + /// Every content event restarts the window: a burst longer than the window + /// coalesces into ONE result, not one per window's worth of burst. + /// + /// A window that expired at a fixed offset from the FIRST event splits any burst + /// longer than `debounce_ms`; each piece rebuilds separately, against a different + /// intermediate state of the file. + #[test] + fn debounce_quiet_period_extends_on_content_events() { + let (tx, rx) = mpsc::channel::(); + // The window now outlives the burst, so the test must too: in production the + // notify sender lives as long as the watcher, and a dropped sender means + // "the watcher is gone", not "the burst ended". + let keepalive = tx.clone(); + // 40 events, 5ms apart: a ~200ms burst under a 100ms window. + let sender = std::thread::spawn(move || { + for i in 0..40u32 { + if tx.send(modify_event(&format!("/w/f{i}.mds"))).is_err() { + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + }); + + let t0 = Instant::now(); + let outcome = drain_bounded( + rx, + 100, + Duration::from_secs(3), + "a burst of content events must not postpone the window forever", + ); + let elapsed = t0.elapsed(); + drop(keepalive); + sender.join().expect("sender thread panicked"); + + assert_eq!( + outcome.end, + DebounceEnd::Quiet, + "a 200ms burst under a 100ms window must end quiet, not capped" + ); + assert_eq!( + outcome.paths.len(), + 40, + "every path in the burst must be collected into the one window; got {:?}", + outcome.paths + ); + assert!( + elapsed >= Duration::from_millis(240) && elapsed <= Duration::from_millis(900), + "the window must outlast the burst (>=200ms) and then close one window \ + later (~100ms), so ~300ms; got {elapsed:?}" + ); + } + + /// The cap ends a stream that never goes quiet. + /// + /// Without it an extendable window is unbounded: a file written to continuously + /// postpones its own rebuild — and the idle-tick backstop behind it — for as long + /// as the writing lasts. + #[test] + fn debounce_cap_ends_a_continuous_stream() { + let (tx, rx) = mpsc::channel::(); + // Events every 2ms for ~2s: never a 50ms gap, so the window never goes quiet. + let sender = std::thread::spawn(move || { + // Bounded: at most 1000 iterations regardless of timing. + for _ in 0..1000u32 { + if tx.send(modify_event("/w/hot.mds")).is_err() { + return; + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + + let t0 = Instant::now(); + let outcome = drain_bounded( + rx, + 50, + Duration::from_secs(3), + "an unbounded window never returns while writes continue", + ); + let elapsed = t0.elapsed(); + sender.join().expect("sender thread panicked"); + + assert_eq!( + outcome.end, + DebounceEnd::Cap, + "a continuous stream must end the window at the cap, not quiet" + ); + assert!( + elapsed >= Duration::from_millis(900) && elapsed < Duration::from_millis(1600), + "cap for a 50ms window is max(500ms, 1s) = 1s; got {elapsed:?}" + ); + } + + /// `--debounce 0` opens no window and consumes nothing. + #[test] + fn debounce_zero_is_disabled_and_leaves_the_channel_untouched() { + let (tx, rx) = mpsc::channel::(); + tx.send(modify_event("/w/a.mds")).expect("send failed"); + + let t0 = Instant::now(); + let outcome = drain_debounce(&rx, 0); + let elapsed = t0.elapsed(); + + assert_eq!(outcome.end, DebounceEnd::Disabled); + assert!( + outcome.paths.is_empty(), + "a disabled window must collect nothing" + ); + assert!( + elapsed < Duration::from_millis(50), + "a disabled window must return immediately; got {elapsed:?}" + ); + assert!( + matches!(rx.try_recv(), Ok(Msg::Fs(Ok(_)))), + "the queued event must still be in the channel: with coalescing off the \ + loop delivers it as its own batch" + ); + } + + /// Ctrl+C ends the window at once, however long the window had left. + #[test] + fn debounce_interrupt_returns_immediately() { + let (tx, rx) = mpsc::channel::(); + let sender = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(20)); + let _ = tx.send(Msg::Interrupt); + }); + + let t0 = Instant::now(); + let outcome = drain_bounded( + rx, + 5_000, + Duration::from_secs(2), + "Ctrl+C must not wait out the window", + ); + let elapsed = t0.elapsed(); + sender.join().expect("sender thread panicked"); + + assert_eq!(outcome.end, DebounceEnd::Interrupted); + assert!( + outcome.interrupted(), + "interrupted() must agree with the end reason" + ); + assert!( + elapsed < Duration::from_millis(500), + "an interrupt must end a 5s window immediately; got {elapsed:?}" + ); + } + + /// Reads do not extend the window. + /// + /// The compile reads its own sources, so an extending `Access` event would let the + /// watcher hold its own window open. + #[test] + fn debounce_access_events_do_not_extend() { + let (tx, rx) = mpsc::channel::(); + // Outlive the read stream, so a window that DID extend ends on its own + // elapsed time rather than on the sender being dropped — the failure then + // names the property under test instead of the channel's lifetime. + let keepalive = tx.clone(); + let sender = std::thread::spawn(move || { + if tx.send(modify_event("/w/a.mds")).is_err() { + return; + } + // Bounded: at most 60 iterations (~300ms) regardless of timing. + for _ in 0..60u32 { + if tx.send(access_event("/w/a.mds")).is_err() { + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + }); + + let t0 = Instant::now(); + let outcome = drain_bounded( + rx, + 100, + Duration::from_secs(3), + "a stream of reads must not hold the window open", + ); + let elapsed = t0.elapsed(); + drop(keepalive); + sender.join().expect("sender thread panicked"); + + assert_eq!(outcome.end, DebounceEnd::Quiet); + assert_eq!( + outcome.paths.len(), + 1, + "only the one content event contributes a path; got {:?}", + outcome.paths + ); + assert!( + elapsed < Duration::from_millis(250), + "300ms of reads must not extend a 100ms window past ~100ms; got {elapsed:?}" + ); + } + + /// One window drains a bounded number of messages. + /// + /// The cap bounds the window's duration; this bounds its work and its memory. A + /// sender faster than the drain would otherwise grow `paths` without limit inside + /// a single window. + #[test] + fn debounce_message_limit_bounds_one_window() { + let (tx, rx) = mpsc::channel::(); + // Pre-queued so the drain is never waiting on the sender. + for _ in 0..12_000u32 { + tx.send(modify_event("/w/same.mds")).expect("send failed"); + } + + let t0 = Instant::now(); + let outcome = drain_bounded( + rx, + 100, + Duration::from_secs(5), + "an unbounded message count lets a fast sender own the window", + ); + let elapsed = t0.elapsed(); + drop(tx); + + assert_eq!( + outcome.end, + DebounceEnd::MessageLimit, + "12 000 queued events must hit the message bound, not the quiet period" + ); + assert_eq!( + outcome.paths.len(), + 1, + "all 12 000 events name the same path; got {:?}", + outcome.paths + ); + assert!( + elapsed < Duration::from_secs(2), + "the bound must be reached promptly; got {elapsed:?}" + ); + } + + /// A dropped sender ends the window at once rather than waiting it out. + /// + /// The sender lives as long as the watcher, so a disconnect means the watcher is + /// gone. Sitting out the remaining window there would delay shutdown by up to the + /// cap for no possible gain: no further event can ever arrive. + #[test] + fn debounce_disconnected_ends_the_window_immediately() { + let (tx, rx) = mpsc::channel::(); + // One event already queued, so the drain has something to collect before it + // reaches the disconnect — the exit must not discard it. + tx.send(modify_event("/w/a.mds")).expect("send failed"); + drop(tx); + + let t0 = Instant::now(); + let outcome = drain_debounce(&rx, 5_000); + let elapsed = t0.elapsed(); + + assert_eq!(outcome.end, DebounceEnd::Disconnected); + assert_eq!( + outcome.paths.len(), + 1, + "messages queued before the disconnect must still be collected; got {:?}", + outcome.paths + ); + assert!( + elapsed < Duration::from_millis(500), + "a disconnect must end a 5s window immediately; got {elapsed:?}" + ); + } + + /// The clamp contract, verifiable without the watch loop. + #[test] + fn clamp_debounce_contract() { + assert_eq!( + clamp_debounce(0), + None, + "0 disables coalescing; it does not mean a 0ms window" + ); + assert_eq!(clamp_debounce(100), Some(Duration::from_millis(100))); + assert_eq!( + clamp_debounce(u64::MAX), + Some(Duration::from_secs(60)), + "an unclamped u64::MAX window does not overflow on a monotonic clock — it \ + lands ~585 million years out, so the watcher silently never rebuilds" + ); + } + + /// The cap contract: `max(10 x window, 1s)`. + #[test] + fn debounce_cap_contract() { + assert_eq!( + debounce_cap(Duration::from_millis(10)), + Duration::from_secs(1), + "the floor binds for small windows" + ); + assert_eq!( + debounce_cap(Duration::from_millis(250)), + Duration::from_millis(2_500) + ); + assert_eq!( + debounce_cap(Duration::from_millis(1_000)), + Duration::from_secs(10) + ); + } + // ── Content backstop domain (#321) ─────────────────────────────────────── /// `tracked_set` covers cross-root dependencies, which `known_files` never can. diff --git a/crates/mds-cli/tests/cli_build.rs b/crates/mds-cli/tests/cli_build.rs index 84262524..90b48b14 100644 --- a/crates/mds-cli/tests/cli_build.rs +++ b/crates/mds-cli/tests/cli_build.rs @@ -1,5 +1,8 @@ mod common; -use common::{count_occurrences, dup_vars_file_omitted, dup_vars_file_warning, fixture, mds_bin}; +use common::{ + count_occurrences, dup_vars_file_omitted, dup_vars_file_warning, fixture, mds_bin, + spawn_watch_ready, ChildGuard, +}; #[test] fn build_to_file() { @@ -1303,55 +1306,54 @@ fn check_stdin_resource_limit_exits_3() { /// (build / check / fmt / lint) are above. Watch is the one subcommand that /// resolves parents through its own call sites; this test locks in that startup path. /// -/// Only asserts the INITIAL BUILD (bounded 10-second wait) — no event-timing -/// assertions that would be timing-flaky on Linux CI. +/// Only asserts the INITIAL BUILD — no event-timing assertions that would be +/// timing-flaky on Linux CI. +/// +/// Synchronised on the `MDS_TEST_READY` handshake (#318), not on the artifact. +/// `run_watch_file` publishes the startup output well before it writes the marker, so +/// once [`spawn_watch_ready`] returns, `hello.md` is already on disk and is read +/// **once**, directly. The previous shape polled the output file for up to 10s, which +/// is the defect the issue names: polling turns "the startup compile wrote the file" +/// into "something wrote the file eventually", so a startup path that resolved the +/// bare filename only on a later retry — or a rebuild — still passed. A shorter poll +/// would preserve that; only removing the loop removes it. #[test] fn watch_bare_filename_from_cwd_succeeds() { use std::process::Stdio; - use std::time::{Duration, Instant}; - - // RAII guard — kills + waits the child on drop so the test never leaks processes. - struct ChildGuard(std::process::Child); - impl Drop for ChildGuard { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } - } let dir = tempfile::tempdir().unwrap(); // Use a distinguishable sentinel so "exit 0 + empty file" can't pass. std::fs::write(dir.path().join("hello.mds"), "Hello from watch!\n").unwrap(); let out = dir.path().join("hello.md"); - let _child = ChildGuard( + let (child, tap, stdout_tap) = spawn_watch_ready( mds_bin() .current_dir(dir.path()) .args(["watch", "hello.mds", "--debounce", "0", "-q"]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("failed to spawn mds watch"), - ); - - // Poll until the output file appears and contains the compiled content. - // Bounded to 10 s; the initial compile typically finishes in < 100 ms. - let deadline = Instant::now() + Duration::from_secs(10); - let found = loop { - if let Ok(content) = std::fs::read_to_string(&out) { - if content.contains("Hello from watch!") { - break true; - } - } - if Instant::now() >= deadline { - break false; - } - std::thread::sleep(Duration::from_millis(50)); - }; + .stdout(Stdio::null()), + ); assert!( - found, + stdout_tap.is_none(), + "stdout is null here; a tap would mean the command piped it" + ); + // RAII guard — kills + waits the child on drop so the test never leaks processes. + let _child = ChildGuard(child); + + // stderr is drained rather than discarded: `-q` still lets a compile error + // through, so a failure here names its own cause instead of being silent. + let content = std::fs::read_to_string(&out).unwrap_or_else(|e| { + panic!( + "mds watch from cwd must have written {} before signalling \ + readiness: {e}; stderr:\n{}", + out.display(), + tap.text() + ) + }); + assert!( + content.contains("Hello from watch!"), "mds watch from cwd should complete initial compile and write hello.md \ - containing 'Hello from watch!'" + containing 'Hello from watch!'; got: {content:?}; stderr:\n{}", + tap.text() ); } diff --git a/crates/mds-cli/tests/cli_watch.rs b/crates/mds-cli/tests/cli_watch.rs index 1d936373..f4c8dd85 100644 --- a/crates/mds-cli/tests/cli_watch.rs +++ b/crates/mds-cli/tests/cli_watch.rs @@ -13,6 +13,17 @@ //! a self-heal idle tick, [`STARTUP_WINDOW_TIMEOUT`] for the deliberately //! unsynchronized startup-window tests. See each constant's docs. //! +//! Writes to watched paths go through `common::write_atomic`. The rule is mechanical, +//! so a reviewer can reproduce the set exactly: a write is converted iff it occurs +//! AFTER the `spawn_ready`/`spawn_unsynchronized` call in the same test fn AND targets +//! a path the watcher is watching (the `.mds` source, an imported partial, the +//! `--vars` file, an external dependency). Pre-spawn fixture writes, `.git` markers, +//! `mds.json`, and output files keep `std::fs::write`. Two post-spawn writes are +//! deliberate exceptions and say so inline: `watch_single_status_line_per_rebuild`, +//! whose subject IS the truncate+write pair that `write_atomic` collapses, and +//! `watch_debounce_single_rebuild_from_burst`, which keeps plain writes because they +//! double the event load its coalescing claim has to survive. +//! //! Flakiness mitigations: //! - Assert on output FILE content rather than stderr ordering. //! - Write dependency files BEFORE adding the `@import` that references them. @@ -21,34 +32,16 @@ mod common; use common::{ - dup_vars_file_warning, mds_bin, spawn_watch_ready, spawn_watch_unsynchronized, StderrTap, + dup_vars_file_warning, mds_bin, spawn_watch_ready, spawn_watch_unsynchronized, write_atomic, + ChildGuard, StderrTap, StdoutTap, }; use std::path::Path; -use std::process::{Child, Command, Stdio}; +use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; // ── Helpers ──────────────────────────────────────────────────────────────── -/// RAII guard that kills + waits the child process on drop. -struct ChildGuard(Child); - -impl Drop for ChildGuard { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } -} - -impl ChildGuard { - fn id(&self) -> u32 { - self.0.id() - } - fn wait_status(&mut self) -> std::process::ExitStatus { - self.0.wait().expect("wait failed") - } -} - /// Spawn a watcher, block until it reports readiness, and wrap it in a `ChildGuard`. /// /// Every test that edits files under an `mds watch` must go through this: it returns @@ -58,17 +51,40 @@ impl ChildGuard { /// The returned [`StderrTap`] holds everything the child wrote to stderr, including /// the startup lines printed before the readiness marker. fn spawn_ready(cmd: &mut Command) -> (ChildGuard, StderrTap) { - let (child, tap) = spawn_watch_ready(cmd); + let (child, tap, stdout_tap) = spawn_watch_ready(cmd); + assert!( + stdout_tap.is_none(), + "this spawn piped stdout; use spawn_ready_piped_stdout so the drained stdout \ + is handed back instead of discarded" + ); (ChildGuard(child), tap) } +/// [`spawn_ready`] for a command that set `.stdout(Stdio::piped())`. +/// +/// The stdout pipe is drained by the harness — it has to be, or the child blocks on a +/// full pipe before it can write the readiness marker — so the tap is the only way to +/// read it. Tests must not take `child.0.stdout` themselves; it is already gone. +fn spawn_ready_piped_stdout(cmd: &mut Command) -> (ChildGuard, StderrTap, StdoutTap) { + let (child, tap, stdout_tap) = spawn_watch_ready(cmd); + let stdout_tap = stdout_tap.expect("caller must set .stdout(Stdio::piped())"); + (ChildGuard(child), tap, stdout_tap) +} + /// Spawn a watcher WITHOUT the readiness handshake and wrap it in a `ChildGuard`. /// /// Reserved for the tests that exist precisely to exercise startup: a test that /// synchronises on "startup finished" can never observe anything that happens /// *during* startup. Every other test must use [`spawn_ready`]. fn spawn_unsynchronized(cmd: &mut Command) -> (ChildGuard, StderrTap) { - let (child, tap) = spawn_watch_unsynchronized(cmd); + let (child, tap, stdout_tap) = spawn_watch_unsynchronized(cmd); + assert!( + stdout_tap.is_none(), + "this spawn piped stdout, and the harness has already drained it — the tap \ + would be discarded here. Add a `spawn_unsynchronized_piped_stdout` wrapper \ + mirroring `spawn_ready_piped_stdout` and use that instead; none exists yet \ + because no unsynchronized test pipes stdout." + ); (ChildGuard(child), tap) } @@ -247,7 +263,7 @@ fn watch_edit_entry_updates_output() { ); // Edit the source. - std::fs::write(&src, "---\nname: Bob\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: Bob\n---\nHello {{name}}!\n"); // Wait for rebuild. assert!( @@ -293,11 +309,10 @@ fn watch_edit_imported_dep_updates_entry() { ); // Edit the helper to change the greeting. - std::fs::write( + write_atomic( &helper, "@define greet(name):\nHi there {{name}}!\n@end\n\n@export greet\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out, "Hi there World!", TIMEOUT), @@ -329,7 +344,7 @@ fn watch_compile_error_keeps_watcher_alive() { ); // Introduce a compile error. - std::fs::write(&src, "Hello {{undefined_var_xyz}}!\n").unwrap(); + write_atomic(&src, "Hello {{undefined_var_xyz}}!\n"); // Give the watcher time to attempt rebuild. std::thread::sleep(Duration::from_millis(500)); @@ -342,7 +357,7 @@ fn watch_compile_error_keeps_watcher_alive() { ); // Fix the error — watcher should recover. - std::fs::write(&src, "---\nname: Charlie\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: Charlie\n---\nHello {{name}}!\n"); assert!( wait_for_file_contains(&out, "Hello Charlie!", TIMEOUT), "fixing the error should trigger a successful rebuild" @@ -393,11 +408,10 @@ fn watch_dir_mode_compiles_all_on_startup() { ); // Edit a.mds → only a.md should update. - std::fs::write( - dir.path().join("a.mds"), + write_atomic( + &dir.path().join("a.mds"), "---\nname: A-edited\n---\nFile A: {{name}}\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out_dir.join("a.md"), "File A: A-edited", TIMEOUT), "editing a.mds should update a.md" @@ -442,11 +456,10 @@ fn watch_dir_mode_picks_up_new_files() { ); // Create a new file AFTER the watcher is running. - std::fs::write( - dir.path().join("c.mds"), + write_atomic( + &dir.path().join("c.mds"), "---\nname: C\n---\nNew file {{name}}\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out_dir.join("c.md"), "New file C", TIMEOUT), @@ -539,7 +552,7 @@ fn watch_vars_file_change_triggers_recompile() { ); // Edit the vars file. - std::fs::write(&vars, r#"{"name": "Bob"}"#).unwrap(); + write_atomic(&vars, r#"{"name": "Bob"}"#); assert!( wait_for_file_contains(&out, "Hello Bob!", TIMEOUT), @@ -575,16 +588,17 @@ fn watch_clear_non_tty_no_ansi_escape() { // Edit the source to trigger a rebuild — this is the path that calls // clear_terminal(). On a non-TTY pipe it must be a no-op. - std::fs::write(&src, "---\nname: There\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: There\n---\nHello {{name}}!\n"); assert!( wait_for_file_contains(&out, "Hello There!", TIMEOUT), "rebuild should occur after editing source" ); - // Stop the child and collect everything it wrote to stderr. - let _ = child.0.kill(); - let _ = child.0.wait(); - let stderr_bytes = stderr_tap.bytes(); + // Stop the child and collect everything it wrote to stderr. `finish` reaps the + // child and then JOINS the drain thread, so the snapshot cannot be a truncated + // prefix — this site is where the Linux tearing was first observed. Raw bytes, + // not text: the assertions below hunt for raw ESC sequences. + let stderr_bytes = stderr_tap.finish(&mut child); // AC-F6: the ANSI clear/home sequences emitted by clear_terminal() // (\x1b[2J, \x1b[3J, \x1b[H) must be ABSENT when stderr is not a TTY. @@ -676,7 +690,7 @@ fn watch_set_vars_applied_on_rebuild() { ); // Edit to trigger rebuild — --set should still apply. - std::fs::write(&src, "Greetings {{name}}!\n").unwrap(); + write_atomic(&src, "Greetings {{name}}!\n"); assert!( wait_for_file_contains(&out, "Greetings Alice!", TIMEOUT), "--set name=Alice should persist across rebuilds" @@ -736,8 +750,8 @@ fn watch_stdout_contains_content_when_o_stdout() { let src = dir.path().join("hello.mds"); std::fs::write(&src, "---\nname: World\n---\nHello {{name}}!\n").unwrap(); - // -o - forces stdout output. - let (mut child, _stderr_tap) = spawn_ready( + // -o - forces stdout output. The harness drains the pipe, so poll the tap. + let (child, _stderr_tap, stdout_tap) = spawn_ready_piped_stdout( mds_bin() .args([ "watch", @@ -751,25 +765,13 @@ fn watch_stdout_contains_content_when_o_stdout() { .stdout(Stdio::piped()), ); - // Read from stdout with a timeout. - use std::io::Read as _; + // Bounded by TIMEOUT: at most TIMEOUT / 50ms iterations. let deadline = Instant::now() + TIMEOUT; - let mut buf = String::new(); let mut found = false; - // Give the child time to produce output. while Instant::now() < deadline { - let mut tmp = [0u8; 256]; - if let Some(stdout) = child.0.stdout.as_mut() { - match stdout.read(&mut tmp) { - Ok(0) | Err(_) => {} - Ok(n) => { - buf.push_str(&String::from_utf8_lossy(&tmp[..n])); - if buf.contains("Hello World!") { - found = true; - break; - } - } - } + if stdout_tap.text().contains("Hello World!") { + found = true; + break; } std::thread::sleep(Duration::from_millis(50)); } @@ -849,7 +851,7 @@ fn watch_debounce_final_value_wins_after_rapid_edits() { // Write 10 rapid edits within the debounce window. for i in 1..=10 { - std::fs::write(&src, format!("---\nname: v{i}\n---\nHello {{{{name}}}}!\n")).unwrap(); + write_atomic(&src, format!("---\nname: v{i}\n---\nHello {{{{name}}}}!\n")); // Tiny sleep to ensure filesystem registers the write, but // well within the 200ms debounce window. std::thread::sleep(Duration::from_millis(5)); @@ -935,11 +937,10 @@ fn watch_import_removal_stops_tracking_dep() { // STEP 1 (add direction, already covered by T-I3 but verified here too): // Edit helper — entry output should update because helper is tracked. - std::fs::write( + write_atomic( &helper, "@define greet(name):\nHi {{name}}!\n@end\n\n@export greet\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out, "Hi World!", TIMEOUT), "editing helper while imported should trigger a rebuild" @@ -947,7 +948,7 @@ fn watch_import_removal_stops_tracking_dep() { // STEP 2 (removal direction): rewrite entry to remove the @import. // The entry now produces static output that does NOT reference helper. - std::fs::write(&entry, "Static content\n").unwrap(); + write_atomic(&entry, "Static content\n"); assert!( wait_for_file_contains(&out, "Static content", TIMEOUT), "removing @import should rebuild entry with static content" @@ -958,11 +959,10 @@ fn watch_import_removal_stops_tracking_dep() { // STEP 3: Edit helper again — entry output must NOT change because the dep // was removed from the watch set after the resync in step 2. - std::fs::write( + write_atomic( &helper, "@define greet(name):\nBye {{name}}!\n@end\n\n@export greet\n", - ) - .unwrap(); + ); // Wait long enough for any spurious rebuild to materialize (500ms >> debounce 0). std::thread::sleep(Duration::from_millis(500)); @@ -1028,7 +1028,7 @@ fn watch_dir_mode_vars_change_recompiles_all() { ); // Edit vars.json — BOTH outputs should update. - std::fs::write(&vars, r#"{"greeting": "Goodbye"}"#).unwrap(); + write_atomic(&vars, r#"{"greeting": "Goodbye"}"#); assert!( wait_for_file_contains(&out_dir_path.join("a.md"), "Goodbye from A", TIMEOUT), @@ -1065,7 +1065,7 @@ fn watch_quiet_keeps_errors_visible() { ); // Introduce a compile error (reference an undefined variable with no frontmatter default). - std::fs::write(&src, "Hello {{__undefined_xyz__}}!\n").unwrap(); + write_atomic(&src, "Hello {{__undefined_xyz__}}!\n"); // Give the watcher time to attempt rebuild and emit error. std::thread::sleep(Duration::from_millis(500)); @@ -1093,7 +1093,7 @@ fn watch_quiet_keeps_errors_visible() { ); // Fix the error — watcher should recover. - std::fs::write(&src, "---\nname: Fixed\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: Fixed\n---\nHello {{name}}!\n"); assert!( wait_for_file_contains(&out, "Hello Fixed!", TIMEOUT), "after fixing the compile error, watcher should rebuild successfully" @@ -1149,11 +1149,10 @@ fn watch_ctrl_c_prints_stopped_watching() { "exit code should be 0 after Ctrl+C, got: {status:?}" ); - // Give the reader thread a moment to flush remaining bytes. - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + // The child has already exited; `finish_text` reaps it again (harmless — `wait` + // caches the status) and then joins the drain thread, which is what actually + // guarantees every byte has been copied. + let stderr_str = stderr_tap.finish_text(&mut guard); assert!( stderr_str.contains("Stopped watching."), "stderr should contain 'Stopped watching.' after Ctrl+C, got: {stderr_str:?}" @@ -1162,10 +1161,20 @@ fn watch_ctrl_c_prints_stopped_watching() { // ── AC-P1: Debounce coalesces burst — count rebuild summary lines ────────── -/// Burst of ~10 writes within a 250ms debounce window must produce exactly 1 -/// "Recompiled " line in stderr. 250ms is large enough to be reliable on CI; -/// if the filesystem splits the burst into two windows, the test permits <= 2 -/// rebuilds (documented below) but asserts == 1 as the expected case. +/// A save burst LONGER than the debounce window is still one rebuild (#379). +/// +/// The old shape of this test wrote a burst that fit inside the window and then +/// tolerated a second rebuild, so the property it advertised — one rebuild per burst — +/// was never actually pinned. It failed as `got 3` on loaded CI runners (runs +/// 33996153739, 33976595173, 33753123463), each of the three compiles seeing a +/// different intermediate state of the file. +/// +/// The burst here is deliberately longer than the window: 12 writes, 30ms apart, so at +/// least 330ms against a 250ms window. Under a window that expires at a fixed offset from +/// the FIRST event that is two or three rebuilds; under a quiet period it is one, +/// because no gap between writes ever reaches 250ms. `--poll-interval` is left at its +/// default so the idle-tick liveness probe stays live — a stronger claim than +/// disabling it. #[test] fn watch_debounce_single_rebuild_from_burst() { let dir = tempfile::tempdir().unwrap(); @@ -1173,7 +1182,6 @@ fn watch_debounce_single_rebuild_from_burst() { std::fs::write(&src, "---\nname: v0\n---\nBurst {{name}}!\n").unwrap(); let out = dir.path().join("burst.md"); - // Use a 250ms debounce — large enough to reliably swallow the ~10 × 5ms burst. let (mut child, stderr_tap) = spawn_ready( mds_bin() .args(["watch", src.to_str().unwrap(), "--debounce", "250"]) @@ -1186,45 +1194,144 @@ fn watch_debounce_single_rebuild_from_burst() { "initial compile should produce Burst v0!" ); - // Write 10 rapid edits within the 250ms debounce window. - for i in 1..=10u32 { + // DELIBERATE: this test's subject is the debounce window collapsing a burst of + // truncate+write pairs, so it keeps plain writes — they double the event load + // that `write_atomic` would collapse into one rename. Every other post-spawn write + // in this file goes through `write_atomic`. + let mut stamps: Vec = Vec::with_capacity(12); + for i in 1..=12u32 { std::fs::write(&src, format!("---\nname: v{i}\n---\nBurst {{{{name}}}}!\n")).unwrap(); - std::thread::sleep(Duration::from_millis(5)); + stamps.push(Instant::now()); + std::thread::sleep(Duration::from_millis(30)); } - // Wait for the debounced rebuild to settle (debounce window + generous FSEvent latency). + // Self-diagnosing preconditions, asserted BEFORE the outcome: if the burst this + // process actually produced was not longer than the window, or had a gap wide + // enough to legitimately close it, the outcome assertion below would be measuring + // the scheduler rather than the watcher. + let span = stamps[stamps.len() - 1].duration_since(stamps[0]); + let max_gap = stamps + .windows(2) + .map(|w| w[1].duration_since(w[0])) + .max() + .expect("burst has at least two writes"); assert!( - wait_for_file_contains(&out, "Burst v10!", TIMEOUT), - "after burst, output should reflect final value v10" + span > Duration::from_millis(250), + "precondition: the burst must outlast the 250ms window, else the test proves \ + nothing about extension; span was {span:?}" + ); + assert!( + max_gap < Duration::from_millis(250), + "precondition: no gap between writes may reach the 250ms window, else the \ + window is entitled to close mid-burst; largest gap was {max_gap:?}" ); - // Wait an extra moment to ensure no trailing rebuilds are in-flight. - std::thread::sleep(Duration::from_millis(400)); + // WAIT ONLY — the assertion is the count below, taken from the joined tap. + wait_for_stderr_contains_str(&stderr_tap, "Recompiled ", TIMEOUT); + let stderr = stderr_tap.finish_text(&mut child); - // Kill child and collect all stderr. - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); + assert_eq!( + count_occurrences(&stderr, "Recompiled "), + 1, + "a {span:?} burst with a largest gap of {max_gap:?} must coalesce into exactly \ + one rebuild under a 250ms quiet period; stderr was:\n{stderr}" + ); + assert_eq!( + count_occurrences(&stderr, "Compiled to"), + 1, + "the startup compile is the only 'Compiled to' line; stderr was:\n{stderr}" + ); + // `mds` copies the frontmatter block through verbatim and interpolates the body. + assert_eq!( + std::fs::read_to_string(&out).unwrap(), + "---\nname: v12\n---\nBurst v12!\n", + "the single rebuild must compile the FINAL state of the burst, not an \ + intermediate one; stderr was:\n{stderr}" + ); +} - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); +/// The cap rebuilds a file that is never left alone (#379). +/// +/// A quiet period that can always be extended is unbounded: a writer that never +/// pauses postpones its own rebuild for as long as it keeps writing. `--poll-interval 0` +/// turns the idle-tick liveness probe off, so within this test the cap is the ONLY +/// mechanism that can produce a rebuild while the stream is running — and it is also +/// the reason the probe cannot be starved in the configurations that do enable it, +/// since the loop never reaches `TickClock::recv_next` while a window is open. +#[test] +fn watch_debounce_cap_rebuilds_while_writes_never_stop() { + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("hot.mds"); + std::fs::write(&src, "---\nname: v0\n---\nHot {{name}}!\n").unwrap(); + let out = dir.path().join("hot.md"); - // Count "Recompiled " lines (each rebuild emits exactly one such line). - let rebuild_count = stderr_str.matches("Recompiled ").count(); + // --debounce 200 => cap = max(10 x 200ms, 1s) = 2s. + let (mut child, stderr_tap) = spawn_ready( + mds_bin() + .args([ + "watch", + src.to_str().unwrap(), + "--debounce", + "200", + "--poll-interval", + "0", + ]) + .stdout(Stdio::null()), + ); - // Expected: exactly 1 rebuild from the burst. - // Allow <= 2 as a documented tolerance: on a heavily loaded CI machine the - // 250ms window may occasionally be split by an FSEvent scheduling gap, yielding - // a second rebuild for the tail of the burst. The important property is that - // we do NOT get 10 individual rebuilds. assert!( - rebuild_count >= 1, - "at least one rebuild must have occurred, got 0; stderr: {stderr_str}" + wait_for_file_contains(&out, "Hot v0!", TIMEOUT), + "initial compile should produce Hot v0!" ); + + let writing = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); + let writer_flag = std::sync::Arc::clone(&writing); + let writer_src = src.clone(); + let writer = std::thread::spawn(move || { + let stop_at = Instant::now() + Duration::from_secs(3); + let mut max_gap = Duration::ZERO; + let mut last = Instant::now(); + // Doubly bounded: <= 3s of wall clock AND <= 2000 iterations. + for i in 1..=2_000u32 { + if Instant::now() >= stop_at { + break; + } + write_atomic( + &writer_src, + format!("---\nname: v{i}\n---\nHot {{{{name}}}}!\n"), + ); + let now = Instant::now(); + max_gap = max_gap.max(now.duration_since(last)); + last = now; + std::thread::sleep(Duration::from_millis(5)); + } + writer_flag.store(false, std::sync::atomic::Ordering::SeqCst); + max_gap + }); + + // The cap is 2s; allow the compile that follows it to land inside the bound. + wait_for_stderr_contains_str(&stderr_tap, "Recompiled ", Duration::from_millis(3500)); + let rebuilt_while_writing = writing.load(std::sync::atomic::Ordering::SeqCst); + + let max_gap = writer.join().expect("writer thread panicked"); assert!( - rebuild_count <= 2, - "debounce should coalesce burst into <= 2 rebuilds, got {rebuild_count}; \ - stderr: {stderr_str}" + max_gap < Duration::from_millis(200), + "precondition: no gap in the write stream may reach the 200ms window, else a \ + quiet period could legitimately have ended it; largest gap was {max_gap:?}" + ); + assert!( + rebuilt_while_writing, + "a rebuild must happen WHILE the writes are still arriving — that is what the \ + cap is for; nothing was seen until the stream stopped" + ); + + let stderr = stderr_tap.finish_text(&mut child); + let rebuilds = count_occurrences(&stderr, "Recompiled "); + assert!( + (1..=4).contains(&rebuilds), + "3s of writes under a 200ms window with a 2s cap is one capped rebuild plus \ + the quiet-period rebuild that follows the last write; a fixed 200ms window \ + would give ~15. Got {rebuilds}; stderr was:\n{stderr}" ); } @@ -1323,12 +1430,7 @@ fn watch_startup_no_spurious_recompile() { std::thread::sleep(Duration::from_millis(1500)); // Stop the child and collect all stderr. - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(50)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); // There must be exactly ONE "Compiled to" message (the initial compile). let compiled_count = stderr_str.matches("Compiled to").count(); @@ -1360,7 +1462,7 @@ fn watch_stdout_no_duplicate_write_on_startup() { // Use a distinctive marker so we can count occurrences. std::fs::write(&src, "UNIQUE_MARKER_XYZ\n").unwrap(); - let (mut child, _stderr_tap) = spawn_ready( + let (mut child, _stderr_tap, stdout_tap) = spawn_ready_piped_stdout( mds_bin() .args([ "watch", @@ -1374,37 +1476,12 @@ fn watch_stdout_no_duplicate_write_on_startup() { .stdout(Stdio::piped()), ); - // Drain stdout on a background thread. - let stdout_handle = child.0.stdout.take().expect("piped stdout"); - let stdout_buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); - let stdout_buf_clone = stdout_buf.clone(); - let _reader_thread = std::thread::spawn(move || { - use std::io::Read as _; - let mut handle = stdout_handle; - let mut tmp = [0u8; 512]; - loop { - match handle.read(&mut tmp) { - Ok(0) | Err(_) => break, - Ok(n) => { - stdout_buf_clone - .lock() - .unwrap() - .extend_from_slice(&tmp[..n]); - } - } - } - }); - // Let the watcher run long enough to capture initial compile + any spurious second write. std::thread::sleep(Duration::from_millis(1500)); - // Stop the child and collect all stdout. - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(50)); - - let stdout_bytes = stdout_buf.lock().unwrap().clone(); - let stdout_str = String::from_utf8_lossy(&stdout_bytes); + // Stop the child and collect all stdout. `finish_text` reaps the child and then + // joins the drain, so no flush sleep is needed to make the snapshot complete. + let stdout_str = stdout_tap.finish_text(&mut child); // The marker should appear at least once (the initial compile wrote it). assert!( @@ -1477,12 +1554,7 @@ fn watch_dir_mode_no_spurious_startup_recompile() { std::thread::sleep(Duration::from_millis(1500)); // Stop the child and collect all stderr. - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(50)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); // There must be ZERO "Recompiled" lines — no rebuild without edits. let recompiled_count = stderr_str.matches("Recompiled").count(); @@ -1551,6 +1623,9 @@ fn watch_single_status_line_per_rebuild() { ); // Make ONE real content-changing edit. + // DELIBERATE: this test's subject is coalescing the truncate+write pair at + // --debounce 100, so it keeps the plain write. Every other post-spawn write in + // this file goes through `write_atomic`. std::fs::write(&src, "---\nname: v1\n---\nStatus {{name}}!\n").unwrap(); // Wait for the rebuild to appear in the output. @@ -1563,12 +1638,7 @@ fn watch_single_status_line_per_rebuild() { std::thread::sleep(Duration::from_millis(500)); // Stop the child and collect all stderr. - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(50)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); // Exactly ONE "Recompiled" line (the real edit). let recompiled_count = stderr_str.matches("Recompiled").count(); @@ -1736,11 +1806,10 @@ fn watch_dir_mode_shared_partial_rebuilds_importers() { ); // Edit the partial — both importers must rebuild. - std::fs::write( + write_atomic( &partial, "@define greet(name):\nHi {{name}}!\n@end\n\n@export greet\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out_dir.join("a.md"), "Hi A!", TIMEOUT), @@ -1800,7 +1869,7 @@ fn watch_dir_mode_chain_rebuild() { ); // Edit C — A must update. - std::fs::write(&c, "@define val():\nV2\n@end\n\n@export val\n").unwrap(); + write_atomic(&c, "@define val():\nV2\n@end\n\n@export val\n"); assert!( wait_for_file_contains(&out_dir.join("a.md"), "V2", TIMEOUT), "a.md should update to V2 after editing _c.mds (transitive chain)" @@ -1839,7 +1908,7 @@ fn watch_poll_interval_zero_works() { ); // Verify a real edit also works. - std::fs::write(&src, "---\nname: Poll\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: Poll\n---\nHello {{name}}!\n"); assert!( wait_for_file_contains(&out, "Hello Poll!", TIMEOUT), "--poll-interval 0: edit should still trigger rebuild via native event" @@ -1947,12 +2016,7 @@ fn watch_file_mode_idle_no_recompile_across_ticks() { // Idle for 2.5s (≥2 ticks at 100ms poll-interval — well above the minimum). std::thread::sleep(Duration::from_millis(2500)); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); let recompiled_count = stderr_str.matches("Recompiled").count(); assert_eq!( @@ -2021,12 +2085,7 @@ fn watch_dir_mode_idle_no_recompile_across_ticks() { // Idle for 2.5s (≥2 ticks at 100ms). std::thread::sleep(Duration::from_millis(2500)); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); let recompiled_count = stderr_str.matches("Recompiled").count(); assert_eq!( @@ -2094,7 +2153,7 @@ fn watch_file_mode_parent_dir_delete_recreate_recovers() { // Recreate the parent dir and the source file with new content. std::fs::create_dir(&src_dir).unwrap(); - std::fs::write(&src, "---\nname: After\n---\nEntry {{name}}\n").unwrap(); + write_atomic(&src, "---\nname: After\n---\nEntry {{name}}\n"); // TICK-DEPENDENT: `remove_dir_all(&src_dir)` destroyed the inotify watch on the old // inode, and the recreated dir is a new inode nothing is watching — so the write @@ -2149,11 +2208,10 @@ fn watch_dir_mode_root_delete_recreate_recovers() { // Recreate the root with a brand-new file (init-gap case). std::fs::create_dir(&root).unwrap(); - std::fs::write( - root.join("new.mds"), + write_atomic( + &root.join("new.mds"), "---\nname: N\n---\nNew file {{name}}\n", - ) - .unwrap(); + ); // TICK-DEPENDENT: the recursive watch died with the old root inode, so the create // above is unobservable; the liveness probe's re-arm + reconcile is the only path. @@ -2228,7 +2286,7 @@ fn watch_file_mode_entry_deleted_settles_then_recovers() { ); // Recreate the file with different content. - std::fs::write(&src, "---\nname: Recovered\n---\nHello {{name}}!\n").unwrap(); + write_atomic(&src, "---\nname: Recovered\n---\nHello {{name}}!\n"); // Wait for recompile after recovery. assert!( @@ -2239,12 +2297,7 @@ fn watch_file_mode_entry_deleted_settles_then_recovers() { // Give the watcher a moment to settle after recovery before killing. std::thread::sleep(Duration::from_millis(200)); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); // Sanity: error count across the FULL test run must still be small — rules out a // burst of errors that somehow all arrived in window 1. @@ -2307,18 +2360,14 @@ fn watch_vars_dir_delete_recreate_rearms() { std::thread::sleep(Duration::from_millis(300)); // Now write new vars — the re-armed watcher should catch this event. - std::fs::write(&vars_file, r#"{"greeting": "Goodbye"}"#).unwrap(); + write_atomic(&vars_file, r#"{"greeting": "Goodbye"}"#); // TICK-DEPENDENT: whether the write above is delivered as an event depends on the // probe having already re-armed the recreated vars dir. If it has not, the fallback // is the probe's own `(mtime, size)` comparison — another tick. Either way the // recovery is denominated in ticks, not in event latency. let got = wait_for_file_contains(&out, "Goodbye", TICK_TIMEOUT); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); assert!( got, "watcher must re-arm vars dir watch after delete+recreate and recompile on edit; \ @@ -2394,11 +2443,10 @@ fn watch_dir_mode_cross_root_partial_edit_rebuilds_importer() { ); // Edit the external partial. - std::fs::write( + write_atomic( &partial, "@define greet():\nExternal V2\n@end\n\n@export greet\n", - ) - .unwrap(); + ); // In-root importer output must update. assert!( @@ -2532,7 +2580,7 @@ fn watch_dir_mode_create_missing_partial_heals_importer() { // Now create the previously-missing partial. let partial = dir.path().join("_missing.mds"); - std::fs::write(&partial, "@define val():\nHealed!\n@end\n\n@export val\n").unwrap(); + write_atomic(&partial, "@define val():\nHealed!\n@end\n\n@export val\n"); // The importer should heal and produce output. assert!( @@ -2593,11 +2641,10 @@ fn watch_dir_mode_dual_role_node_edit_and_delete() { // Edit dual.mds — both dual.md and consumer.md should update. // Use a longer content to force a size delta. - std::fs::write( + write_atomic( &dual, "@define greet():\nDual V2 (updated)\n@end\n\n@export greet\n\nStandalone updated content\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains( @@ -2701,12 +2748,7 @@ fn watch_dir_mode_persistent_error_bounded_count() { "watcher must stay alive with persistent syntax error in bad.mds" ); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); assert_eq!( count_w1, count_w2, @@ -2884,11 +2926,10 @@ fn watch_dir_mode_partial_edit_rebuilds_exactly_n_importers() { let independent_before = std::fs::read_to_string(out_dir.join("independent.md")).unwrap(); // Edit the partial with different-length content to force a deterministic (mtime,size) delta. - std::fs::write( + write_atomic( &partial, "@define val():\nV2 updated\n@end\n\n@export val\n", - ) - .unwrap(); + ); // All three importers must update. assert!( @@ -2958,11 +2999,10 @@ fn watch_dir_mode_soak_50_edits_bounded_and_clean_exit() { for i in 1_u32..=50 { // Pad with spaces to ensure each round has a unique byte count. let padding = " ".repeat(i as usize); - std::fs::write( + write_atomic( &partial, format!("@define val():\nSoak V{i}{padding}\n@end\n\n@export val\n"), - ) - .unwrap(); + ); // Wait for this round's rebuild to propagate. let expected = format!("Soak V{i}"); @@ -3065,7 +3105,7 @@ fn watch_file_mode_parent_dir_deleted_bounded_errors_then_recovers() { // Recreate the parent directory and write the file with new content. std::fs::create_dir(&src_dir).unwrap(); - std::fs::write(&src, "V2-recovered\n").unwrap(); + write_atomic(&src, "V2-recovered\n"); // TICK-DEPENDENT: same as AC-W1 — the parent dir was removed, so the watch on it is // gone and the recreated dir is unwatched. Recovery is the vanish→reappear edge in @@ -3078,12 +3118,7 @@ fn watch_file_mode_parent_dir_deleted_bounded_errors_then_recovers() { // Watcher must still be alive after recovery. let still_alive = child.0.try_wait().unwrap().is_none(); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); assert!( still_alive, @@ -3175,12 +3210,7 @@ fn watch_dir_mode_idle_500_files_no_recompile() { // (ADR-021) must emit zero "Recompiled" lines during this window. std::thread::sleep(Duration::from_millis(600)); - let _ = child.0.kill(); - let _ = child.0.wait(); - std::thread::sleep(Duration::from_millis(100)); - - let stderr_bytes = stderr_tap.bytes(); - let stderr_str = String::from_utf8_lossy(&stderr_bytes); + let stderr_str = stderr_tap.finish_text(&mut child); let recompiled_count = stderr_str.matches("Recompiled").count(); assert_eq!( @@ -3455,16 +3485,17 @@ fn watch_esc_in_initial_compile_error_is_sanitized() { // The readiness handshake already implies the initial compile ran to completion: // the error is printed on the startup path, and the marker is only emitted after // it. No sleep needed to "give it time". - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args(["watch", src.to_str().unwrap(), "--debounce", "0"]) .stdout(Stdio::null()), ); - // Kill the watch process (ChildGuard.drop → kill + wait) to close the pipe. - drop(child); - - let stderr_bytes = stderr_tap.bytes(); + // Kill the watch process to close the pipe, then join the drain: `finish` does + // both in that order, so the snapshot is the complete stream rather than whatever + // the drain thread happened to have copied by then. Raw bytes, because assertion + // 2 below hunts for a raw ESC byte. + let stderr_bytes = stderr_tap.finish(&mut child); let stderr_str = String::from_utf8_lossy(&stderr_bytes); // Assertion 1: the initial-compile error was rendered (non-vacuous guard for @@ -3551,7 +3582,7 @@ fn watch_file_mode_edit_during_startup_window_is_not_lost() { ); // Edit now — inside the window under the defective ordering. - std::fs::write(&src, "---\nname: After\n---\nEntry {{name}}\n").unwrap(); + write_atomic(&src, "---\nname: After\n---\nEntry {{name}}\n"); assert!( wait_for_file_contains(&out, "Entry After", STARTUP_WINDOW_TIMEOUT), @@ -3595,7 +3626,7 @@ fn watch_dir_mode_edit_during_startup_window_is_not_lost() { "startup compile should publish 'Dir Before'" ); - std::fs::write(&src, "---\nname: After\n---\nDir {{name}}\n").unwrap(); + write_atomic(&src, "---\nname: After\n---\nDir {{name}}\n"); assert!( wait_for_file_contains(&out, "Dir After", STARTUP_WINDOW_TIMEOUT), @@ -3675,11 +3706,10 @@ fn watch_dir_mode_cross_root_edit_during_startup_window_is_not_lost() { ); // Edit the cross-root partial now — inside the window where nothing is watching it. - std::fs::write( + write_atomic( &partial, "@define greet():\nWindow V2\n@end\n\n@export greet\n", - ) - .unwrap(); + ); // TICK-DEPENDENT: no filesystem event announces this edit, so recovery is the idle // tick's `(mtime, size)` diff against the baseline captured before the first read. @@ -3751,11 +3781,10 @@ fn watch_file_mode_dep_edit_during_startup_window_is_not_lost() { "startup compile should publish 'Dep V1'" ); - std::fs::write( + write_atomic( &partial, "@define greet():\nDep V2\n@end\n\n@export greet\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out, "Dep V2", TICK_TIMEOUT), @@ -3852,7 +3881,7 @@ fn watch_dir_mode_idle_tick_fires_under_event_flood() { std::fs::remove_dir_all(&root).unwrap(); std::thread::sleep(Duration::from_millis(200)); std::fs::create_dir(&root).unwrap(); - std::fs::write(root.join("new.mds"), "---\nname: N\n---\nFlood {{name}}\n").unwrap(); + write_atomic(&root.join("new.mds"), "---\nname: N\n---\nFlood {{name}}\n"); let recovered = wait_for_file_contains(&out_dir.join("new.md"), "Flood N", TICK_TIMEOUT); @@ -4079,6 +4108,155 @@ fn watch_file_mode_ctrl_c_during_startup_compile_terminates() { ); } +/// Bounded wait for a child that has already been signalled. +/// +/// A **bound, not a synchroniser**: a signalled child exits in milliseconds, and one +/// that has not exited by the deadline is the defect the caller is asserting against. +/// `what` names the arm so the panic is self-describing. +#[cfg(unix)] +fn wait_bounded(guard: &mut ChildGuard, timeout: Duration, what: &str) -> std::process::ExitStatus { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if let Some(status) = guard.0.try_wait().expect("try_wait failed") { + return status; + } + std::thread::sleep(Duration::from_millis(1)); + } + panic!("{what}: the process did not exit within {timeout:?} of the signal"); +} + +/// Control for #129 surface 1: the readiness handshake — not luck — is what makes a +/// post-SIGINT `status.success()` deterministic. +/// +/// Two arms, the same signal, opposite verdicts: +/// +/// - **CONTROL.** [`spawn_unsynchronized`], with SIGINT gated on the `Watching …` +/// line. `run_watch_file` prints that line before it even creates the watcher, and +/// therefore long before `ctrlc::set_handler`, so the signal lands in the +/// pre-handler window where the default disposition still applies: death by SIGINT. +/// If this arm ever exits cleanly, the window is no longer being hit and the +/// treatment arm below proves nothing. +/// - **TREATMENT.** [`spawn_ready`], with SIGINT sent the instant the handshake +/// returns. `set_handler` precedes `emit_ready_marker` in `run_watch_file`, so once +/// the marker exists the handler provably does too: exit 0 and `Stopped watching.`. +/// +/// `N = 20` is a live discriminator, not a rate bound — a single clean control exit +/// fails the run. The manual Linux soak workflow is the rate instrument. Every wait +/// here is bounded and none of them is a sleep standing in for a synchroniser. +/// `#[cfg(unix)]` because SIGINT has no Windows analogue; the test compiles out there. +#[test] +#[cfg(unix)] +fn watch_readiness_handshake_makes_ctrl_c_exit_deterministic() { + use std::os::unix::process::ExitStatusExt; + + const N: usize = 20; + /// Imports in the control entry. Same fixture shape as + /// `watch_file_mode_ctrl_c_during_startup_compile_terminates`: enough work that + /// the startup compile is demonstrably still running when the signal lands. + const PARTIALS: usize = 400; + + // Both fixtures are built ONCE. No watcher in this test ever edits a watched file, + // so rebuilding them per iteration would buy nothing but wall clock. + let slow_dir = tempfile::tempdir().unwrap(); + let mut entry = String::new(); + for i in 0..PARTIALS { + let name = format!("_p{i:04}"); + std::fs::write( + slow_dir.path().join(format!("{name}.mds")), + format!("@define v{i}():\nP{i}\n@end\n\n@export v{i}\n"), + ) + .unwrap(); + entry.push_str(&format!("@import \"./{name}.mds\" as p{i}\n")); + } + entry.push_str("done\n"); + let slow_src = slow_dir.path().join("entry.mds"); + std::fs::write(&slow_src, &entry).unwrap(); + + let fast_dir = tempfile::tempdir().unwrap(); + let fast_src = fast_dir.path().join("hello.mds"); + std::fs::write(&fast_src, "---\nname: World\n---\nHello {{name}}!\n").unwrap(); + + for iteration in 0..N { + // ── CONTROL arm: signal delivered before the handler is installed ─────── + let (mut guard, tap) = spawn_unsynchronized( + // No -q: the `Watching …` line is the gate. + mds_bin() + .args(["watch", slow_src.to_str().unwrap(), "--debounce", "0"]) + .stdout(Stdio::null()), + ); + let pid = guard.id(); + + let deadline = Instant::now() + STARTUP_WINDOW_TIMEOUT; + let mut saw_watching = false; + while Instant::now() < deadline { + if tap.text().contains("Watching ") { + saw_watching = true; + break; + } + std::thread::sleep(Duration::from_millis(1)); + } + assert!( + saw_watching, + "control arm, iteration {iteration}: expected the `Watching …` startup \ + line before signalling; stderr:\n{}", + tap.text() + ); + + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGINT); + } + let status = wait_bounded( + &mut guard, + Duration::from_secs(20), + "control arm (SIGINT before the handler is installed)", + ); + assert_eq!( + status.signal(), + Some(libc::SIGINT), + "control arm, iteration {iteration}: SIGINT delivered before \ + `ctrlc::set_handler` runs must terminate the process. A clean exit here \ + means the signal no longer lands in the pre-handler window, and the \ + treatment arm below then proves nothing; got {status:?}" + ); + assert!( + !status.success(), + "control arm, iteration {iteration}: death by signal is not a success \ + status; got {status:?}" + ); + + // ── TREATMENT arm: signal delivered after the readiness handshake ─────── + let (mut guard, tap) = spawn_ready( + mds_bin() + .args(["watch", fast_src.to_str().unwrap(), "--debounce", "0"]) + .stdout(Stdio::null()), + ); + let pid = guard.id(); + + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGINT); + } + let status = wait_bounded( + &mut guard, + Duration::from_secs(5), + "treatment arm (SIGINT after the readiness handshake)", + ); + assert!( + status.success(), + "treatment arm, iteration {iteration}: after the readiness handshake the \ + ctrl-c handler provably exists (`set_handler` precedes \ + `emit_ready_marker` in `run_watch_file`), so SIGINT must exit 0; got \ + {status:?}; stderr:\n{}", + tap.text() + ); + let stderr = tap.finish_text(&mut guard); + assert!( + stderr.contains("Stopped watching."), + "treatment arm, iteration {iteration}: a clean SIGINT exit must also print \ + `Stopped watching.`; stderr:\n{stderr}" + ); + } +} + // ── I8: file-watch mode warns exactly ONCE across two edits (#200) ────────── /// Count non-overlapping occurrences of `needle` in `haystack`. @@ -4110,6 +4288,38 @@ fn wait_for_stderr_contains_str(tap: &StderrTap, needle: &str, timeout: Duration } } +/// Wait until the stderr tap holds at least `n` occurrences of `needle`. +/// +/// Returns the tap's contents as soon as the count is reached. Unlike +/// [`wait_for_stderr_contains_str`], which returns the text on timeout and so lets the +/// caller's assertion report the shortfall as if it were a final answer, this one +/// PANICS on timeout and names the count it actually saw. +/// +/// Why a count and not "contains": a stderr line the watcher emits AFTER the output +/// write has no ordering relationship with the output file the test waited on. +/// Dir-mode emits the duplicate-vars-key warning after the write (watch.rs +/// `handle_fs_event_dir`), so a snapshot taken the instant `wait_for_file_contains` +/// returns can legitimately be one warning short — or, if the previous rebuild's +/// warning has not been sampled yet, one long. Waiting for the expected count first +/// turns the assertion that follows into a genuine over-count check instead of a race. +fn wait_for_stderr_count(tap: &StderrTap, needle: &str, n: usize, timeout: Duration) -> String { + let deadline = Instant::now() + timeout; + // Bounded by `timeout`: at most timeout / 20ms iterations. + loop { + let text = tap.text(); + let seen = count_occurrences(&text, needle); + if seen >= n { + return text; + } + assert!( + Instant::now() < deadline, + "expected at least {n} occurrences of {needle:?} within {timeout:?}; \ + saw {seen}; stderr was:\n{text}" + ); + std::thread::sleep(Duration::from_millis(20)); + } +} + #[test] fn i8_file_watch_duplicate_set_warns_exactly_once_across_two_edits() { // I8: mds watch (file mode) with --set x=1 --set x=2 must print the @@ -4148,14 +4358,14 @@ fn i8_file_watch_duplicate_set_warns_exactly_once_across_two_edits() { ); // Edit 1: trigger a rebuild. - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I8: rebuild after edit 1 must complete" ); // Edit 2: trigger another rebuild. - std::fs::write(&src, "version 3").unwrap(); + write_atomic(&src, "version 3"); assert!( wait_for_file_contains(&out, "version 3", TIMEOUT), "I8: rebuild after edit 2 must complete" @@ -4222,7 +4432,7 @@ fn i9_dir_watch_duplicate_set_warns_exactly_once_at_startup() { ); // Trigger a rebuild to exercise the :1914 path (handle_dir_event). - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I9: rebuild after edit must complete" @@ -4249,7 +4459,11 @@ fn i9_dir_watch_duplicate_set_warns_exactly_once_at_startup() { // re-reported each time (D9). /// I16: mds watch (file mode) with a duplicated top-level key in the vars file -/// warns at STARTUP and on EVERY rebuild. Guards `watch.rs:936`. +/// warns at STARTUP and on EVERY rebuild. Guards the emit in `rebuild_file`. +/// +/// Each count assertion is preceded by a bounded wait for that count, so it reads +/// "never more than N", not "happened to be N when sampled". The warning is written +/// to stderr with no ordering relationship to the output file the test waits on. #[test] fn i16_file_watch_vars_file_duplicate_warns_at_startup_and_on_every_rebuild() { let base = tempfile::tempdir().unwrap(); @@ -4266,7 +4480,7 @@ fn i16_file_watch_vars_file_duplicate_warns_at_startup_and_on_every_rebuild() { let expected = dup_vars_file_warning("x", &vars_file); - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args([ "watch", @@ -4279,7 +4493,7 @@ fn i16_file_watch_vars_file_duplicate_warns_at_startup_and_on_every_rebuild() { .stdout(Stdio::null()), ); - let stderr_after_start = wait_for_stderr_contains_str(&stderr_tap, &expected, TIMEOUT); + let stderr_after_start = wait_for_stderr_count(&stderr_tap, &expected, 1, TIMEOUT); assert_eq!( count_occurrences(&stderr_after_start, &expected), 1, @@ -4288,12 +4502,12 @@ fn i16_file_watch_vars_file_duplicate_warns_at_startup_and_on_every_rebuild() { // Edit 1: trigger a rebuild — ADR-016 reloads the vars file, re-reporting the // duplicate. - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I16: rebuild after edit 1 must complete" ); - let after_edit_1 = stderr_tap.text(); + let after_edit_1 = wait_for_stderr_count(&stderr_tap, &expected, 2, TIMEOUT); assert_eq!( count_occurrences(&after_edit_1, &expected), 2, @@ -4301,25 +4515,33 @@ fn i16_file_watch_vars_file_duplicate_warns_at_startup_and_on_every_rebuild() { ); // Edit 2: trigger another rebuild. - std::fs::write(&src, "version 3").unwrap(); + write_atomic(&src, "version 3"); assert!( wait_for_file_contains(&out, "version 3", TIMEOUT), "I16: rebuild after edit 2 must complete" ); - let after_edit_2 = stderr_tap.text(); + let _ = wait_for_stderr_count(&stderr_tap, &expected, 3, TIMEOUT); + let after_edit_2 = stderr_tap.finish_text(&mut child); assert_eq!( count_occurrences(&after_edit_2, &expected), 3, "I16: a second rebuild must report the duplicate again; stderr:\n{after_edit_2}" ); - - drop(child); } /// I17: mds watch (dir mode) reports the vars-file duplicate exactly once per -/// rebuild: once at startup (proving the `:2196` dedup-baseline second read does -/// NOT double-print), and once more per subsequent rebuild (proving exactly one -/// of `:1793`/`:1919` emits, not both). +/// rebuild: once at startup (proving the dedup-baseline second read in +/// `dir_watch_startup` does NOT double-print), and once more per subsequent rebuild +/// (proving exactly one of `liveness_probe_dir` / `handle_fs_event_dir` emits, not +/// both). +/// +/// Sampling hazard this test has to defend against: dir mode emits the warning AFTER +/// the output write, so `wait_for_file_contains` returning tells you nothing about +/// whether the warning has been written yet. Sampling `stderr_tap.text()` right there +/// is a race in both directions, and CI has shown both — run 34404318888 attempt 1 +/// saw left 1 / right 2 here, while run 34366009518 saw left 3 / right 2. The wait +/// for the expected count has to come first; the exact-count assertion then means +/// "not more than expected" rather than "happened to be sampled at the right moment". #[test] fn i17_dir_watch_vars_file_duplicate_warns_once_per_rebuild() { let base = tempfile::tempdir().unwrap(); @@ -4336,7 +4558,7 @@ fn i17_dir_watch_vars_file_duplicate_warns_once_per_rebuild() { let expected = dup_vars_file_warning("x", &vars_file); - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args([ "watch", @@ -4350,31 +4572,32 @@ fn i17_dir_watch_vars_file_duplicate_warns_once_per_rebuild() { ); // No edits yet: the startup count must be exactly 1, proving the dedup-baseline - // second read at :2196 does not also emit. - let stderr_startup = wait_for_stderr_contains_str(&stderr_tap, &expected, TIMEOUT); + // second read in `dir_watch_startup` does not also emit. + let stderr_startup = wait_for_stderr_count(&stderr_tap, &expected, 1, TIMEOUT); assert_eq!( count_occurrences(&stderr_startup, &expected), 1, "I17: dir-watch startup must emit the vars-file warning exactly once \ - (guards :2196); stderr:\n{stderr_startup}" + (guards the dedup-baseline second read in dir_watch_startup); \ + stderr:\n{stderr_startup}" ); // One rebuild: the count must rise to exactly 2, proving exactly one of // :1793/:1919 fires per rebuild (not both). - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I17: rebuild after edit must complete" ); - let stderr_after_edit = stderr_tap.text(); + let _ = wait_for_stderr_count(&stderr_tap, &expected, 2, TIMEOUT); + let stderr_after_edit = stderr_tap.finish_text(&mut child); assert_eq!( count_occurrences(&stderr_after_edit, &expected), 2, "I17: one rebuild must add exactly one more warning (guards a double-emit \ - between :1793 and :1919); stderr:\n{stderr_after_edit}" + between liveness_probe_dir and handle_fs_event_dir); \ + stderr:\n{stderr_after_edit}" ); - - drop(child); } /// I18 (user decision, positive control first): a vars file that starts clean @@ -4397,7 +4620,7 @@ fn i18_duplicate_introduced_mid_session_is_reported_on_the_next_rebuild() { let expected = dup_vars_file_warning("x", &vars_file); - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args([ "watch", @@ -4424,7 +4647,7 @@ fn i18_duplicate_introduced_mid_session_is_reported_on_the_next_rebuild() { ); // First rebuild, still clean: still no warning. - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I18: first rebuild must complete" @@ -4438,21 +4661,29 @@ fn i18_duplicate_introduced_mid_session_is_reported_on_the_next_rebuild() { ); // Introduce a duplicate mid-session, then trigger the next rebuild. - std::fs::write(&vars_file, r#"{"x": 1, "x": 2}"#).unwrap(); - std::fs::write(&src, "version 3").unwrap(); + // + // Two watched files are written here, yet the expected count below is exactly 1. + // The warning is gated in `rebuild_file` on an OBSERVABLE output-content change — + // the same signal that gates the "Recompiled" line — and this fixture never + // interpolates `x`, so the rebuild the vars-file write triggers produces + // byte-identical output and reports nothing. Only the `version 3` rebuild is + // observable. The atomic writes are what make that exact: a truncate-then-write + // published a 0-byte intermediate, which was itself an observable transition and + // could contribute a second warning. + write_atomic(&vars_file, r#"{"x": 1, "x": 2}"#); + write_atomic(&src, "version 3"); assert!( wait_for_file_contains(&out, "version 3", TIMEOUT), "I18: rebuild after introducing the duplicate must complete" ); - let final_stderr = stderr_tap.text(); + let _ = wait_for_stderr_count(&stderr_tap, &expected, 1, TIMEOUT); + let final_stderr = stderr_tap.finish_text(&mut child); assert_eq!( count_occurrences(&final_stderr, &expected), 1, "I18: the duplicate introduced mid-session must be reported on the next \ rebuild, naming the key; stderr:\n{final_stderr}" ); - - drop(child); } // ── I19-I20: liveness self-heal rebuild and --quiet regressions (#326) ─────── @@ -4482,7 +4713,7 @@ fn i19_dir_watch_liveness_self_heal_rebuild_warns_about_vars_file_duplicate() { let expected = dup_vars_file_warning("x", &vars_file); - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args([ "watch", @@ -4500,7 +4731,7 @@ fn i19_dir_watch_liveness_self_heal_rebuild_warns_about_vars_file_duplicate() { ); // Startup: exactly 1 warning (dir-mode startup, unaffected by this fix). - let startup_stderr = wait_for_stderr_contains_str(&stderr_tap, &expected, TIMEOUT); + let startup_stderr = wait_for_stderr_count(&stderr_tap, &expected, 1, TIMEOUT); assert_eq!( count_occurrences(&startup_stderr, &expected), 1, @@ -4525,11 +4756,10 @@ fn i19_dir_watch_liveness_self_heal_rebuild_warns_about_vars_file_duplicate() { // warning per observable rebuild, which is what the count assertion below // pins. std::fs::create_dir(&root).unwrap(); - std::fs::write( - root.join("new.mds"), + write_atomic( + &root.join("new.mds"), "---\nname: N\n---\nNew file {{name}}\n", - ) - .unwrap(); + ); assert!( wait_for_file_contains(&out_dir.join("new.md"), "New file N", TICK_TIMEOUT), @@ -4539,15 +4769,14 @@ fn i19_dir_watch_liveness_self_heal_rebuild_warns_about_vars_file_duplicate() { // The self-heal recompile must ALSO re-warn about the vars-file duplicate — // proves liveness_probe_dir no longer discards the resolved vars, matching // handle_fs_event_dir's gate (emit iff the rebuild was observable). - let final_stderr = stderr_tap.text(); + let _ = wait_for_stderr_count(&stderr_tap, &expected, 2, TIMEOUT); + let final_stderr = stderr_tap.finish_text(&mut child); assert_eq!( count_occurrences(&final_stderr, &expected), 2, "I19: the liveness self-heal rebuild must warn about the vars-file \ duplicate too, not only at startup; stderr:\n{final_stderr}" ); - - drop(child); } /// I20: `mds watch --quiet` suppresses the vars-file duplicate-key warning on @@ -4573,7 +4802,7 @@ fn i20_watch_quiet_suppresses_vars_file_duplicate_warning_on_every_rebuild() { let expected = dup_vars_file_warning("x", &vars_file); - let (child, stderr_tap) = spawn_ready( + let (mut child, stderr_tap) = spawn_ready( mds_bin() .args([ "watch", @@ -4601,19 +4830,292 @@ fn i20_watch_quiet_suppresses_vars_file_duplicate_warning_on_every_rebuild() { stderr:\n{startup_stderr}" ); - std::fs::write(&src, "version 2").unwrap(); + write_atomic(&src, "version 2"); assert!( wait_for_file_contains(&out, "version 2", TIMEOUT), "I20: rebuild after edit must complete even under --quiet" ); - let after_edit = stderr_tap.text(); + // No count to wait for — the expectation is zero — so this one takes the + // strongest snapshot available instead: `finish_text` joins the drain, so a + // warning the child wrote and the drain had not yet copied would still be here. + let after_edit = stderr_tap.finish_text(&mut child); assert_eq!( count_occurrences(&after_edit, &expected), 0, "I20: --quiet must suppress the vars-file duplicate warning on rebuild \ too; stderr:\n{after_edit}" ); +} + +// ── R1-R3: rename-into-place (atomic write) is a first-class edit (#320) ───── +// +// Editors and `write_atomic` replace a file by writing a sibling temp file and +// renaming it over the target. That is ONE filesystem event on the destination +// (`Modify(Name(RenameMode::To))` under notify 8 / inotify `IN_MOVED_TO`), not the +// truncate-then-write pair `std::fs::write` produces. These three tests pin that the +// watcher treats it as a content edit and that the in-flight temp file is invisible +// to both watch modes. + +/// R1: file mode must rebuild when the watched source is replaced by a rename. +#[test] +fn watch_file_mode_rename_into_place_triggers_rebuild() { + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("t.mds"); + std::fs::write(&src, "version 1").unwrap(); + let out = dir.path().join("t.md"); + + let (child, stderr_tap) = spawn_ready( + mds_bin() + .args(["watch", src.to_str().unwrap(), "--debounce", "0"]) + .stdout(Stdio::null()), + ); + + assert!( + wait_for_file_contains(&out, "version 1", TIMEOUT), + "R1: startup compile must complete" + ); + + write_atomic(&src, "version 2"); + + assert!( + wait_for_file_contains(&out, "version 2", TIMEOUT), + "R1: a rename-into-place edit must trigger a rebuild" + ); + let stderr = wait_for_stderr_contains_str(&stderr_tap, "Recompiled", TIMEOUT); + assert!( + stderr.contains("Recompiled"), + "R1: the rebuild must announce itself; stderr:\n{stderr}" + ); drop(child); } + +/// R2: dir mode must rebuild when a watched source is replaced by a rename. +#[test] +fn watch_dir_mode_rename_into_place_triggers_rebuild() { + let base = tempfile::tempdir().unwrap(); + let src_dir = base.path().join("src"); + let out_dir = base.path().join("out"); + std::fs::create_dir_all(&src_dir).unwrap(); + std::fs::create_dir_all(&out_dir).unwrap(); + + let src = src_dir.join("t.mds"); + std::fs::write(&src, "version 1").unwrap(); + let out = out_dir.join("t.md"); + + let (child, stderr_tap) = spawn_ready( + mds_bin() + .args([ + "watch", + src_dir.to_str().unwrap(), + "--out-dir", + out_dir.to_str().unwrap(), + "--debounce", + "0", + ]) + .stdout(Stdio::null()), + ); + + assert!( + wait_for_file_contains(&out, "version 1", TIMEOUT), + "R2: startup compile must complete" + ); + + write_atomic(&src, "version 2"); + + assert!( + wait_for_file_contains(&out, "version 2", TIMEOUT), + "R2: a rename-into-place edit must trigger a rebuild" + ); + let stderr = wait_for_stderr_contains_str(&stderr_tap, "Recompiled", TIMEOUT); + assert!( + stderr.contains("Recompiled"), + "R2: the rebuild must announce itself; stderr:\n{stderr}" + ); + + drop(child); +} + +/// R3: the temp file an atomic write leaves in flight is never compiled. +/// +/// The `..tmp--` shape puts the suffix AFTER the `.mds`, so +/// `Path::extension()` is not `mds` and both the dir-mode event filter and +/// `collect_mds_files` drop it. Asserting only that absence would be vacuous if the +/// watcher were simply not compiling anything, so the same test writes a REAL second +/// source through `write_atomic` and requires that one to be compiled. +#[test] +fn watch_dir_mode_write_atomic_temp_file_is_never_compiled() { + let base = tempfile::tempdir().unwrap(); + let src_dir = base.path().join("src"); + let out_dir = base.path().join("out"); + std::fs::create_dir_all(&src_dir).unwrap(); + std::fs::create_dir_all(&out_dir).unwrap(); + + let src = src_dir.join("t.mds"); + std::fs::write(&src, "version 1").unwrap(); + + let (child, stderr_tap) = spawn_ready( + mds_bin() + .args([ + "watch", + src_dir.to_str().unwrap(), + "--out-dir", + out_dir.to_str().unwrap(), + "--debounce", + "0", + ]) + .stdout(Stdio::null()), + ); + + assert!( + wait_for_file_contains(&out_dir.join("t.md"), "version 1", TIMEOUT), + "R3: startup compile must complete" + ); + + // Atomic edit of the existing source, then a brand-new source — also atomic. + write_atomic(&src, "version 2"); + assert!( + wait_for_file_contains(&out_dir.join("t.md"), "version 2", TIMEOUT), + "R3: the atomic edit must rebuild t.md" + ); + + // Positive control: a genuine new source written the same way IS compiled, so the + // "temp file produced nothing" assertions below cannot pass vacuously. + write_atomic(&src_dir.join("u.mds"), "brand new"); + assert!( + wait_for_file_contains(&out_dir.join("u.md"), "brand new", TIMEOUT), + "R3 (positive control): a real source created by a rename must be compiled" + ); + + // No output derives from any temp name, in either directory. + for dir in [&out_dir, &src_dir] { + for entry in std::fs::read_dir(dir).unwrap() { + let name = entry.unwrap().file_name().to_string_lossy().into_owned(); + assert!( + !name.contains(".tmp-"), + "R3: no file derived from a write_atomic temp name may survive in {}; \ + found {name}", + dir.display() + ); + } + } + + // And nothing announced compiling one. + let stderr = stderr_tap.text(); + assert!( + !stderr.contains(".tmp-"), + "R3: no status line may mention a write_atomic temp file; stderr:\n{stderr}" + ); + + drop(child); +} + +// ── Stderr capture completeness (#320) ────────────────────────────────────── + +/// The tap must hand back every byte the child wrote, not a prefix of it. +/// +/// `StderrTap::bytes` clones the shared buffer without any happens-before edge to the +/// drain thread's last write. Reaping the child closes its write end and ends the +/// drain loop, but nothing makes the reader observe that the loop has finished, so a +/// snapshot taken right after `kill` + `wait` can be a truncated prefix. The suite hid +/// that behind a `thread::sleep` at every such site. +/// +/// A dir watcher over 500 sources announces `Compiled to` once per file at startup, so +/// the expected count is exact and any lost tail shows up as a shortfall rather than +/// as a vague "looks empty". The `Compiled to` lines are also the positive control: +/// a count of 0 would mean the watcher compiled nothing, not that the tap is sound. +/// +/// macOS has not been observed to lose the tail; the field signature is Linux +/// (`cli_watch.rs:520` in CI runs 32954883014 and 32954876042). The Linux soak is the +/// instrument for this one. +#[test] +fn stderr_tap_finish_captures_every_line_the_child_wrote() { + const FILE_COUNT: usize = 500; + let dir = tempfile::tempdir().unwrap(); + let out_dir = dir.path().join("out"); + std::fs::create_dir(&out_dir).unwrap(); + + for i in 1..=FILE_COUNT { + std::fs::write( + dir.path().join(format!("file_{i:04}.mds")), + format!("drain-{i}\n"), + ) + .unwrap(); + } + + // No -q: the startup compile announces `Compiled to` once per file. + let (mut child, stderr_tap) = spawn_ready( + mds_bin() + .args([ + "watch", + dir.path().to_str().unwrap(), + "--out-dir", + out_dir.to_str().unwrap(), + "--debounce", + "0", + ]) + .stdout(Stdio::null()), + ); + + // Readiness fires only after the whole startup batch, so all FILE_COUNT lines + // have been written by the child by the time this returns. `finish` reaps the + // child and then joins the drain, so what comes back is the complete stream. + let stderr = stderr_tap.finish_text(&mut child); + let announced = count_occurrences(&stderr, "Compiled to"); + assert_eq!( + announced, FILE_COUNT, + "the tap must return every `Compiled to` line the child wrote; got {announced} \ + of {FILE_COUNT}" + ); +} + +// ── R4: readiness must not depend on someone draining stdout (#320) ───────── + +/// A watcher whose stdout is piped but undrained must still signal readiness. +/// +/// `mds watch -o -` publishes the startup output to stdout BEFORE it writes the +/// readiness marker (watch.rs: the marker is emitted after the compile, the arming and +/// the publish). A pipe holds ~64 KiB; once it is full the child blocks in `write`, so +/// if the harness is sitting in the marker poll loop with nothing draining stdout, +/// neither side can move and the spawn helper times out. +/// +/// 256 KiB of body is several pipe buffers on both Linux and macOS, so the block is a +/// certainty, not a matter of timing. +#[test] +fn watch_ready_with_large_piped_stdout_does_not_deadlock() { + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("big.mds"); + // Plain text is valid MDS; short lines keep the compile trivial. + let body: String = std::iter::repeat_n("x".repeat(63) + "\n", 8192).collect(); + assert!( + body.len() > 256 * 1024, + "fixture must exceed several pipe buffers; got {} bytes", + body.len() + ); + std::fs::write(&src, &body).unwrap(); + + let (mut child, _stderr_tap, stdout_tap) = spawn_ready_piped_stdout( + mds_bin() + .args([ + "watch", + src.to_str().unwrap(), + "-o", + "-", + "--debounce", + "0", + "-q", + ]) + .stdout(Stdio::piped()), + ); + + // Readiness returned, so the startup publish got through. Prove the bytes really + // travelled rather than the marker having been written before any output. + let stdout = stdout_tap.finish(&mut child); + assert!( + stdout.len() >= body.len(), + "the whole startup output must reach stdout; got {} bytes of {}", + stdout.len(), + body.len() + ); +} diff --git a/crates/mds-cli/tests/common/mod.rs b/crates/mds-cli/tests/common/mod.rs index 613b73ae..35c38c5f 100644 --- a/crates/mds-cli/tests/common/mod.rs +++ b/crates/mds-cli/tests/common/mod.rs @@ -1,6 +1,7 @@ use std::io::Read; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -70,6 +71,74 @@ pub fn count_occurrences(haystack: &str, needle: &str) -> usize { count } +// ── Atomic file replacement ────────────────────────────────────────────────── + +/// Monotonic counter making every [`write_atomic`] temp name unique within a +/// process; the pid disambiguates across processes. +static WRITE_ATOMIC_SEQ: AtomicU64 = AtomicU64::new(0); + +/// Replace `path`'s contents in ONE filesystem event, the way an editor does: write a +/// fresh temp file in the same directory, then `rename` it over `path`. +/// +/// Why: `std::fs::write` truncates before it writes, so a zero-debounce watcher can +/// compile the 0-byte intermediate — CI run 34366009518 on 2b91850 printed two +/// `Recompiled` lines for one write. notify 8 surfaces the rename as +/// `Modify(Name(RenameMode::To))` on the destination path, which the watcher treats +/// as a content event. +/// +/// The temp name is `..tmp--` — the suffix goes AFTER the name so +/// `Path::extension()` is never `mds`: `collect_mds_files_inner` (output.rs) and the +/// dir-mode event filter (watch.rs) gate on exactly that, so an in-flight temp file +/// is invisible to both. +/// +/// No fsync: `rename` orders the replacement for every live process, which is all a +/// watcher needs. The product's own readiness marker is written the same way. +/// +/// Deliberate non-user: `watch_single_status_line_per_rebuild`, whose subject IS the +/// coalescing of the truncate+write pair. +/// +/// The Windows sharing-violation caveat (a rename over a file another process holds +/// open can fail) is developer-machine only; CI runs this suite on ubuntu. +/// +/// # Panics +/// Panics if `path` has no parent or no file name, or if either filesystem step +/// fails — a test whose edit did not land is a defect, not a slow machine. +#[allow(dead_code)] +pub fn write_atomic(path: &Path, contents: impl AsRef<[u8]>) { + let dir = path + .parent() + .unwrap_or_else(|| panic!("write_atomic: path has no parent: {}", path.display())); + let name = path + .file_name() + .unwrap_or_else(|| panic!("write_atomic: path has no file name: {}", path.display())); + let seq = WRITE_ATOMIC_SEQ.fetch_add(1, Ordering::Relaxed); + let tmp = dir.join(format!( + ".{}.tmp-{}-{}", + name.to_string_lossy(), + std::process::id(), + seq + )); + debug_assert_ne!( + tmp.extension().and_then(|e| e.to_str()), + Some("mds"), + "write_atomic temp name must never end in .mds; it would be collected as a source" + ); + if let Err(e) = std::fs::write(&tmp, contents.as_ref()) { + panic!( + "write_atomic: cannot write temp file {}: {e}", + tmp.display() + ); + } + if let Err(e) = std::fs::rename(&tmp, path) { + let _ = std::fs::remove_file(&tmp); + panic!( + "write_atomic: cannot rename {} -> {}: {e}", + tmp.display(), + path.display() + ); + } +} + // ── Watch readiness handshake ──────────────────────────────────────────────── /// Contents `mds watch` writes to the file named by `MDS_TEST_READY`. @@ -92,29 +161,136 @@ const READY_POLL: Duration = Duration::from_millis(2); /// every source in the tree while the suite runs at full parallelism. const READY_TIMEOUT: Duration = Duration::from_secs(10); -/// Captured stderr of a watcher spawned by [`spawn_watch_ready`] or -/// [`spawn_watch_unsynchronized`]. +/// RAII guard that kills + waits the child on drop. +/// +/// Lives here rather than in `cli_watch.rs` so [`PipeTap::finish`] can take +/// `&mut ChildGuard` and thereby establish "reaped before join" in the type, not in a +/// comment: the drain thread's loop ends at EOF, and EOF arrives only once the child's +/// write end is closed. +#[allow(dead_code)] +pub struct ChildGuard(pub Child); + +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[allow(dead_code)] +impl ChildGuard { + pub fn id(&self) -> u32 { + self.0.id() + } + + /// Reap an already-exiting child. `Child::wait` caches its status, so calling this + /// and then letting `Drop` run is safe. + pub fn wait_status(&mut self) -> std::process::ExitStatus { + self.0.wait().expect("wait failed") + } + + /// Kill (best-effort) and reap. Idempotent — a second call returns the cached + /// status. + pub fn kill_and_wait(&mut self) -> std::process::ExitStatus { + let _ = self.0.kill(); + self.0.wait().expect("wait failed") + } +} + +/// A background-drained capture of one of the child's output pipes. /// /// Holds **exactly** what the child wrote and nothing else — the readiness handshake -/// travels over a file, not this stream. That is load-bearing: tests assert that a -/// compile error reaches stderr through `--quiet` and that no raw ESC byte appears in -/// a diagnostic, and both assertions become unfalsifiable if the harness itself -/// contributes bytes here. +/// travels over a file, not over these streams. That is load-bearing: tests assert +/// that a compile error reaches stderr through `--quiet` and that no raw ESC byte +/// appears in a diagnostic, and both assertions become unfalsifiable if the harness +/// itself contributes bytes here. +/// +/// [`PipeTap::bytes`] stays NON-blocking so the live-poll sites keep working; +/// [`PipeTap::finish`] is the end-of-test read that is guaranteed complete. #[allow(dead_code)] #[derive(Clone)] -pub struct StderrTap(Arc>>); +pub struct PipeTap { + buf: Arc>>, + /// `Option` because `finish` takes the handle out; behind `Arc>` so + /// `PipeTap` stays `Clone`. `Clone` is harness API — it lets a tap be shared with + /// a helper thread — and the mutex is what makes that safe: a clone calling + /// `finish` concurrently blocks on this slot until the drain has been joined, and + /// then observes the fully drained buffer. No call site clones a tap today. + drain: Arc>>>, +} +/// A [`PipeTap`] over the child's stderr. #[allow(dead_code)] -impl StderrTap { - /// Raw bytes written to stderr so far. +pub type StderrTap = PipeTap; + +/// A [`PipeTap`] over the child's stdout. +#[allow(dead_code)] +pub type StdoutTap = PipeTap; + +#[allow(dead_code)] +impl PipeTap { + /// Bytes written so far. + /// + /// Non-blocking, and therefore carries **no** happens-before edge to the child's + /// last write: a snapshot taken right after the child is reaped can be a truncated + /// prefix. Use it only while polling a live child; use [`PipeTap::finish`] for the + /// final read. pub fn bytes(&self) -> Vec { - self.0.lock().expect("stderr tap poisoned").clone() + self.buf.lock().expect("pipe tap poisoned").clone() } - /// Lossy-UTF8 view of [`StderrTap::bytes`]. + /// Lossy-UTF8 view of [`PipeTap::bytes`], with the same caveat. pub fn text(&self) -> String { String::from_utf8_lossy(&self.bytes()).into_owned() } + + /// Stop the child, JOIN the drain thread, and return everything it wrote. + /// + /// Termination is proved, not bounded: the drain loop exits only at EOF, EOF + /// arrives when the child's write end closes, and the child is reaped here first — + /// so the join cannot hang on a live writer. A clone calling `finish` concurrently + /// blocks on the drain slot and then observes a fully drained buffer. + #[must_use] + pub fn finish(self, child: &mut ChildGuard) -> Vec { + child.kill_and_wait(); + { + let mut slot = self.drain.lock().expect("pipe tap drain slot poisoned"); + if let Some(handle) = slot.take() { + handle.join().expect("pipe drain thread panicked"); + } + } + self.bytes() + } + + /// Lossy-UTF8 view of [`PipeTap::finish`]. + #[must_use] + pub fn finish_text(self, child: &mut ChildGuard) -> String { + String::from_utf8_lossy(&self.finish(child)).into_owned() + } +} + +/// Spawn a background thread that drains `reader` into a fresh [`PipeTap`]. +fn tap_reader(reader: R) -> PipeTap { + let buf = Arc::new(Mutex::new(Vec::::new())); + let sink = buf.clone(); + let handle = std::thread::spawn(move || { + let mut reader = reader; + let mut chunk = [0u8; 512]; + // Bounded by EOF: the loop ends when the child's pipe closes. + loop { + match reader.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => sink + .lock() + .expect("pipe tap poisoned") + .extend_from_slice(&chunk[..n]), + } + } + }); + PipeTap { + buf, + drain: Arc::new(Mutex::new(Some(handle))), + } } /// Spawn a `mds watch` command and drain its stderr, WITHOUT waiting for readiness. @@ -125,34 +301,25 @@ impl StderrTap { /// must act *inside* the startup window, so they cannot synchronise on it closing. /// /// stderr is piped and drained on a background thread so the pipe can never fill and -/// block the child. +/// block the child. If the caller also piped stdout, that pipe is drained the same +/// way and the tap is returned as the third element; `Command` inherits stdout by +/// default, so `child.stdout.is_some()` is exactly "the caller asked for a pipe". +/// +/// Draining stdout here rather than in the caller is what keeps the readiness wait +/// sound: `mds watch -o -` publishes its startup output before it writes the marker, +/// so an undrained stdout pipe fills and blocks the child while the poller waits for +/// a marker that can never be written. #[allow(dead_code)] -pub fn spawn_watch_unsynchronized(cmd: &mut Command) -> (Child, StderrTap) { +pub fn spawn_watch_unsynchronized(cmd: &mut Command) -> (Child, StderrTap, Option) { let mut child = cmd .stderr(Stdio::piped()) .spawn() .expect("failed to spawn mds watch"); - let handle = child.stderr.take().expect("stderr must be piped"); - let buf = Arc::new(Mutex::new(Vec::::new())); - let tap = StderrTap(buf.clone()); - - std::thread::spawn(move || { - let mut handle = handle; - let mut chunk = [0u8; 512]; - // Bounded by EOF: the loop ends when the child's stderr closes. - loop { - match handle.read(&mut chunk) { - Ok(0) | Err(_) => break, - Ok(n) => buf - .lock() - .expect("stderr tap poisoned") - .extend_from_slice(&chunk[..n]), - } - } - }); + let tap = tap_reader(child.stderr.take().expect("stderr must be piped")); + let stdout_tap = child.stdout.take().map(tap_reader); - (child, tap) + (child, tap, stdout_tap) } /// Spawn a `mds watch` command and block until the watcher is **fully armed**. @@ -175,12 +342,17 @@ pub fn spawn_watch_unsynchronized(cmd: &mut Command) -> (Child, StderrTap) { /// stderr is still piped and drained on a background thread so the pipe can never /// fill and block the child. Use the returned [`StderrTap`] to inspect it. /// +/// A piped stdout is drained too, and its tap handed back as the third element. That +/// ordering is load-bearing, not a convenience: `mds watch -o -` publishes its startup +/// output before it writes the marker, so leaving stdout undrained would let the child +/// block on a full pipe while this function waits for a marker that can never arrive. +/// /// # Panics /// Panics if the child cannot be spawned, or if readiness is not signalled within /// [`READY_TIMEOUT`] — a watcher that never reports readiness is a defect, not a /// slow machine. #[allow(dead_code)] -pub fn spawn_watch_ready(cmd: &mut Command) -> (Child, StderrTap) { +pub fn spawn_watch_ready(cmd: &mut Command) -> (Child, StderrTap, Option) { // A private directory per spawn: the suite runs at full parallelism, so a shared // path would let one watcher's marker satisfy another's wait. Dropped — and so // deleted — when this function returns, by which point the marker has been read. @@ -191,13 +363,14 @@ pub fn spawn_watch_ready(cmd: &mut Command) -> (Child, StderrTap) { "MDS_TEST_READY must be absolute; mds watch ignores relative values" ); - let (mut child, tap) = spawn_watch_unsynchronized(cmd.env("MDS_TEST_READY", &ready_path)); + let (mut child, tap, stdout_tap) = + spawn_watch_unsynchronized(cmd.env("MDS_TEST_READY", &ready_path)); // Bounded by READY_TIMEOUT: at most READY_TIMEOUT / READY_POLL iterations. let deadline = std::time::Instant::now() + READY_TIMEOUT; loop { if std::fs::read(&ready_path).is_ok_and(|b| b == READY_MARKER.as_bytes()) { - return (child, tap); + return (child, tap, stdout_tap); } // Check liveness before the deadline so a watcher that failed at startup is // reported as "exited", not as "timed out".