Merge pull request #90941 from NousResearch/bb/installer-drain-bound
fix(installer): the bootstrap installer stops waiting on pipe EOF
This commit is contained in:
brooklyn!
@@ -48,6 +48,9 @@ outputs:
|
||||
installer:
|
||||
description: Run the PowerShell installer tests on a Windows runner.
|
||||
value: ${{ steps.classify.outputs.installer }}
|
||||
rust:
|
||||
description: Run `cargo test` for the Tauri bootstrap installer.
|
||||
value: ${{ steps.classify.outputs.rust }}
|
||||
mcp_catalog:
|
||||
description: Require MCP catalog security review label.
|
||||
value: ${{ steps.classify.outputs.mcp_catalog }}
|
||||
|
||||
@@ -49,6 +49,7 @@ jobs:
|
||||
uv_lock: ${{ steps.classify.outputs.uv_lock }}
|
||||
npm_lock: ${{ steps.classify.outputs.npm_lock }}
|
||||
installer: ${{ steps.classify.outputs.installer }}
|
||||
rust: ${{ steps.classify.outputs.rust }}
|
||||
docker_meta: ${{ steps.classify.outputs.docker_meta }}
|
||||
mcp_catalog: ${{ steps.classify.outputs.mcp_catalog }}
|
||||
ci_review: ${{ steps.classify.outputs.ci_review }}
|
||||
@@ -105,6 +106,14 @@ jobs:
|
||||
if: needs.detect.outputs.installer == 'true'
|
||||
uses: ./.github/workflows/installer-tests.yml
|
||||
|
||||
rust-tests:
|
||||
name: Rust tests
|
||||
needs: detect
|
||||
# Only for PRs that touch a Rust crate. `.rs` is under apps/, so these
|
||||
# changes used to run the TypeScript matrix and nothing that compiles them.
|
||||
if: needs.detect.outputs.rust == 'true'
|
||||
uses: ./.github/workflows/rust-tests.yml
|
||||
|
||||
e2e-desktop:
|
||||
name: Desktop E2E
|
||||
needs: detect
|
||||
@@ -213,6 +222,7 @@ jobs:
|
||||
- lint
|
||||
- js-tests
|
||||
- installer-tests
|
||||
- rust-tests
|
||||
- e2e-desktop
|
||||
- docs-site
|
||||
- history-check
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
# .github/workflows/rust-tests.yml
|
||||
name: Rust tests
|
||||
|
||||
# `cargo test` for the Tauri bootstrap installer (Hermes-Setup). Nothing in CI
|
||||
# compiled this crate before: `.rs` lives under `apps/`, so the change
|
||||
# classifier matched it as `frontend` and ran the TypeScript matrix, which
|
||||
# cannot notice a Rust error. The crate's unit tests existed in the tree and had
|
||||
# never run.
|
||||
#
|
||||
# Linux runner on purpose. The pipe-drain tests in src/powershell.rs need a real
|
||||
# process tree whose grandchild inherits the parent's stdout, and their fixture
|
||||
# is `#[cfg(unix)]`; the Windows half of that same contract is covered by
|
||||
# `-SelfTestPipeDrain` in scripts/desktop-update/windows.ps1 on the Windows
|
||||
# lane. A Windows runner here would compile them out and report green over zero
|
||||
# coverage.
|
||||
|
||||
on:
|
||||
workflow_call:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: rust-tests-${{ github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
bootstrap-installer:
|
||||
name: cargo test (bootstrap installer)
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 30
|
||||
defaults:
|
||||
run:
|
||||
working-directory: apps/bootstrap-installer/src-tauri
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
|
||||
# Tauri links against the system webkit2gtk on Linux, so the crate does
|
||||
# not compile without these even for `cargo test --lib`.
|
||||
- name: Install Tauri system dependencies
|
||||
working-directory: .
|
||||
run: |
|
||||
sudo apt-get update
|
||||
sudo apt-get install --no-install-recommends -y \
|
||||
libwebkit2gtk-4.1-dev \
|
||||
libappindicator3-dev \
|
||||
librsvg2-dev \
|
||||
libxdo-dev \
|
||||
libssl-dev \
|
||||
patchelf
|
||||
|
||||
# Keyed on Cargo.toml, not Cargo.lock: apps/bootstrap-installer/.gitignore
|
||||
# excludes the lockfile (a create-tauri-app scaffold default), so there is
|
||||
# nothing pinned to hash and `--locked` cannot be used. That also means
|
||||
# this crate re-resolves its whole dependency graph on every build, which
|
||||
# is a real gap for a signed installer given the pinning policy in
|
||||
# AGENTS.md — tracking separately rather than widening this PR.
|
||||
#
|
||||
# No restore-keys: a partial hit leaves a stale target dir, and cargo
|
||||
# re-resolves correctly on top of a Cargo.toml-keyed hit anyway.
|
||||
- name: Restore cargo cache
|
||||
uses: actions/cache@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4
|
||||
with:
|
||||
path: |
|
||||
~/.cargo/registry
|
||||
~/.cargo/git
|
||||
apps/bootstrap-installer/src-tauri/target
|
||||
key: cargo-${{ runner.os }}-${{ hashFiles('apps/bootstrap-installer/src-tauri/Cargo.toml') }}
|
||||
|
||||
# --lib only: the integration/bin targets would need a built frontend
|
||||
# (vite dist) that this lane deliberately does not produce.
|
||||
- name: cargo test
|
||||
run: cargo test --lib
|
||||
@@ -8,10 +8,12 @@
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use std::path::Path;
|
||||
use std::process::Stdio;
|
||||
use std::process::{ExitStatus, Stdio};
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
use tokio::process::{Child, Command};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::timeout;
|
||||
|
||||
/// CP1252 mapping for bytes `0x80..=0x9F` (the range that differs from Latin-1).
|
||||
/// Undefined slots keep the C1 control code points, matching Windows-1252
|
||||
@@ -129,6 +131,153 @@ pub struct ScriptResult {
|
||||
/// Cancellation signal — `cancel_tx.send(()).await` aborts the running script.
|
||||
pub type CancelRx = mpsc::Receiver<()>;
|
||||
|
||||
/// How long a child's pipes get to reach EOF AFTER the child itself has exited.
|
||||
///
|
||||
/// This is not a timeout on the child. The clock starts once the process is
|
||||
/// already gone and everything it wrote is sitting in the pipe buffer, so a
|
||||
/// 40-minute `uv pip install` is untouched — the grace only covers the final
|
||||
/// drain.
|
||||
///
|
||||
/// It exists because pipe EOF is not the child's to give. The write end of a
|
||||
/// redirected pipe is handed to the child as an inheritable handle, so every
|
||||
/// descendant spawned without its own redirection holds a duplicate, and the
|
||||
/// read side does not see EOF until the last of them closes it. `hermes update`
|
||||
/// deliberately runs its build steps with stdout inherited, so the tree under a
|
||||
/// child is arbitrarily deep and not something the caller can enumerate. When
|
||||
/// one of those descendants is a resident gateway, the pipe stays open for the
|
||||
/// life of the gateway — and every obligation downstream of the read is
|
||||
/// stranded with it.
|
||||
///
|
||||
/// Same bound `Invoke-HermesStep` grew in `scripts/desktop-update/windows.ps1`
|
||||
/// (#90455), and the same shape as Go's `exec.Cmd.WaitDelay`.
|
||||
pub(crate) const DRAIN_GRACE: Duration = Duration::from_secs(20);
|
||||
|
||||
/// What [`pump_child`] observed.
|
||||
pub(crate) struct PumpOutcome {
|
||||
pub exit_code: Option<i32>,
|
||||
/// The child was killed because the caller cancelled.
|
||||
pub killed: bool,
|
||||
/// The child exited but its pipes never reached EOF within [`DRAIN_GRACE`],
|
||||
/// so the tail of its output was dropped. Callers must say so out loud: a
|
||||
/// silently truncated log is indistinguishable from one that was empty.
|
||||
pub abandoned: bool,
|
||||
}
|
||||
|
||||
/// Stream a child's stdout/stderr line by line, then reap it.
|
||||
///
|
||||
/// The contract that matters: the exit status comes from waiting on the
|
||||
/// *process*, never from pipe EOF. See [`DRAIN_GRACE`] for why those are not
|
||||
/// the same event; callers pass it, tests pass something they can wait out.
|
||||
pub(crate) async fn pump_child<FO, FE>(
|
||||
child: &mut Child,
|
||||
mut on_stdout: FO,
|
||||
mut on_stderr: FE,
|
||||
cancel_rx: &mut Option<CancelRx>,
|
||||
grace: Duration,
|
||||
) -> Result<PumpOutcome>
|
||||
where
|
||||
FO: FnMut(&str),
|
||||
FE: FnMut(&str),
|
||||
{
|
||||
let stdout = child.stdout.take().context("stdout was piped")?;
|
||||
let stderr = child.stderr.take().context("stderr was piped")?;
|
||||
let mut out = BufReader::new(stdout);
|
||||
let mut err = BufReader::new(stderr);
|
||||
let mut out_buf = Vec::new();
|
||||
let mut err_buf = Vec::new();
|
||||
let mut out_done = false;
|
||||
let mut err_done = false;
|
||||
let mut status: Option<ExitStatus> = None;
|
||||
let mut cancelled = false;
|
||||
|
||||
// Phase 1 — the child is alive, so there is no deadline. A slow child is
|
||||
// not a stuck one, and the point of streaming is that a long build keeps
|
||||
// reporting. We leave on whichever lands first: both pipes at EOF (the
|
||||
// clean case), the process exiting (the case that used to hang here), or
|
||||
// cancellation.
|
||||
while !(out_done && err_done) {
|
||||
tokio::select! {
|
||||
line = read_decoded_line(&mut out, &mut out_buf), if !out_done => match line {
|
||||
Ok(Some(l)) => on_stdout(&l),
|
||||
Ok(None) => out_done = true,
|
||||
Err(e) => {
|
||||
tracing::warn!("stdout read error: {e}");
|
||||
out_done = true;
|
||||
}
|
||||
},
|
||||
line = read_decoded_line(&mut err, &mut err_buf), if !err_done => match line {
|
||||
Ok(Some(l)) => on_stderr(&l),
|
||||
Ok(None) => err_done = true,
|
||||
Err(e) => {
|
||||
tracing::warn!("stderr read error: {e}");
|
||||
err_done = true;
|
||||
}
|
||||
},
|
||||
reaped = child.wait() => {
|
||||
status = Some(reaped.context("waiting for child to exit")?);
|
||||
break;
|
||||
}
|
||||
_ = recv_cancel(cancel_rx) => {
|
||||
cancelled = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Kill outside the loop: `child.wait()` above holds the mutable borrow for
|
||||
// as long as the select is in scope.
|
||||
if cancelled {
|
||||
tracing::warn!("cancellation received — killing child");
|
||||
let _ = child.start_kill();
|
||||
}
|
||||
|
||||
// Phase 2 — bounded. Whatever the child already wrote is still worth
|
||||
// keeping, so we keep reading; we just stop caring once a descendant is the
|
||||
// only thing still holding the pipe open. Cancelling does not rescue us
|
||||
// either: `start_kill` kills the child, not the grandchild with the handle.
|
||||
let mut abandoned = false;
|
||||
if !(out_done && err_done) {
|
||||
let drain = async {
|
||||
while !(out_done && err_done) {
|
||||
tokio::select! {
|
||||
line = read_decoded_line(&mut out, &mut out_buf), if !out_done => match line {
|
||||
Ok(Some(l)) => on_stdout(&l),
|
||||
_ => out_done = true,
|
||||
},
|
||||
line = read_decoded_line(&mut err, &mut err_buf), if !err_done => match line {
|
||||
Ok(Some(l)) => on_stderr(&l),
|
||||
_ => err_done = true,
|
||||
},
|
||||
}
|
||||
}
|
||||
};
|
||||
abandoned = timeout(grace, drain).await.is_err();
|
||||
}
|
||||
|
||||
let status = match status {
|
||||
Some(s) => s,
|
||||
// Both pipes reached EOF while the child stayed alive. Reads are done,
|
||||
// but the process is still the authoritative terminal condition — and
|
||||
// it may never exit on its own, so this wait stays cancellable. A bare
|
||||
// `child.wait()` here would strand the caller's cancel channel exactly
|
||||
// when it is the only way out.
|
||||
None => tokio::select! {
|
||||
reaped = child.wait() => reaped.context("waiting for child to exit")?,
|
||||
_ = recv_cancel(cancel_rx) => {
|
||||
tracing::warn!("cancellation received after EOF — killing child");
|
||||
cancelled = true;
|
||||
let _ = child.start_kill();
|
||||
child.wait().await.context("waiting for killed child to exit")?
|
||||
}
|
||||
},
|
||||
};
|
||||
Ok(PumpOutcome {
|
||||
exit_code: status.code(),
|
||||
killed: cancelled,
|
||||
abandoned,
|
||||
})
|
||||
}
|
||||
|
||||
/// Spawns install.ps1 / install.sh with the given args and streams output.
|
||||
///
|
||||
/// `hermes_home_override` propagates to the child as $HERMES_HOME so the
|
||||
@@ -171,88 +320,49 @@ pub async fn run_script(
|
||||
.spawn()
|
||||
.with_context(|| format!("spawning {} via {}", script_path.display(), interpreter_label()))?;
|
||||
|
||||
let stdout = child.stdout.take().expect("stdout was piped");
|
||||
let stderr = child.stderr.take().expect("stderr was piped");
|
||||
|
||||
// Byte-oriented readers + [`decode_console_bytes`]: do NOT use
|
||||
// `BufReader::lines()`, which requires valid UTF-8 and hides localized
|
||||
// PowerShell errors on non-English Windows (#67193).
|
||||
let mut stdout_reader = BufReader::new(stdout);
|
||||
let mut stderr_reader = BufReader::new(stderr);
|
||||
let mut stdout_buf = Vec::new();
|
||||
let mut stderr_buf = Vec::new();
|
||||
|
||||
// PowerShell errors on non-English Windows (#67193). [`pump_child`] owns
|
||||
// that, plus the rule that the exit status comes from the process and not
|
||||
// from pipe EOF.
|
||||
let mut combined_stdout = String::new();
|
||||
let mut combined_stderr = String::new();
|
||||
let mut killed = false;
|
||||
|
||||
// Loop: poll stdout, stderr, cancel, and child exit concurrently.
|
||||
loop {
|
||||
tokio::select! {
|
||||
line = read_decoded_line(&mut stdout_reader, &mut stdout_buf) => {
|
||||
match line {
|
||||
Ok(Some(l)) => {
|
||||
(sink.on_stdout_line)(&l);
|
||||
combined_stdout.push_str(&l);
|
||||
combined_stdout.push('\n');
|
||||
}
|
||||
Ok(None) => {
|
||||
// EOF on stdout — wait for stderr + exit.
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("stdout read error: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
line = read_decoded_line(&mut stderr_reader, &mut stderr_buf) => {
|
||||
match line {
|
||||
Ok(Some(l)) => {
|
||||
(sink.on_stderr_line)(&l);
|
||||
combined_stderr.push_str(&l);
|
||||
combined_stderr.push('\n');
|
||||
}
|
||||
Ok(None) => {
|
||||
// stderr EOF — keep draining stdout.
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("stderr read error: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = recv_cancel(cancel_rx) => {
|
||||
tracing::warn!("cancellation received — killing child");
|
||||
killed = true;
|
||||
// best-effort kill; don't propagate errors
|
||||
let _ = child.start_kill();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
let outcome = pump_child(
|
||||
&mut child,
|
||||
|l| {
|
||||
(sink.on_stdout_line)(l);
|
||||
combined_stdout.push_str(l);
|
||||
combined_stdout.push('\n');
|
||||
},
|
||||
|l| {
|
||||
(sink.on_stderr_line)(l);
|
||||
combined_stderr.push_str(l);
|
||||
combined_stderr.push('\n');
|
||||
},
|
||||
cancel_rx,
|
||||
DRAIN_GRACE,
|
||||
)
|
||||
.await
|
||||
.context("streaming install script output")?;
|
||||
|
||||
// Drain remaining lines after the loop exited.
|
||||
while let Ok(Some(l)) = read_decoded_line(&mut stdout_reader, &mut stdout_buf).await {
|
||||
(sink.on_stdout_line)(&l);
|
||||
combined_stdout.push_str(&l);
|
||||
combined_stdout.push('\n');
|
||||
}
|
||||
while let Ok(Some(l)) = read_decoded_line(&mut stderr_reader, &mut stderr_buf).await {
|
||||
(sink.on_stderr_line)(&l);
|
||||
combined_stderr.push_str(&l);
|
||||
if outcome.abandoned {
|
||||
let note = format!(
|
||||
"install script exited but a surviving descendant still holds its \
|
||||
stdout/stderr; gave up on the last {}s of output (#90455)",
|
||||
DRAIN_GRACE.as_secs()
|
||||
);
|
||||
tracing::warn!("{note}");
|
||||
(sink.on_stderr_line)(¬e);
|
||||
combined_stderr.push_str(¬e);
|
||||
combined_stderr.push('\n');
|
||||
}
|
||||
|
||||
let status = child
|
||||
.wait()
|
||||
.await
|
||||
.context("waiting for install script to exit")?;
|
||||
|
||||
Ok(ScriptResult {
|
||||
stdout: combined_stdout,
|
||||
stderr: combined_stderr,
|
||||
exit_code: status.code(),
|
||||
killed,
|
||||
exit_code: outcome.exit_code,
|
||||
killed: outcome.killed,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -550,4 +660,183 @@ info line
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
|
||||
/// Spawn `sh -c <script>` with both pipes redirected, the way run_script and
|
||||
/// run_streamed do.
|
||||
#[cfg(unix)]
|
||||
fn sh(script: &str) -> Child {
|
||||
Command::new("/bin/sh")
|
||||
.arg("-c")
|
||||
.arg(script)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.expect("spawn /bin/sh")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn pump_collect(child: &mut Child, grace: Duration) -> (PumpOutcome, Vec<String>) {
|
||||
let mut lines = Vec::new();
|
||||
let outcome = pump_child(
|
||||
child,
|
||||
|l| lines.push(l.to_string()),
|
||||
|_| {},
|
||||
&mut None,
|
||||
grace,
|
||||
)
|
||||
.await
|
||||
.expect("pump");
|
||||
(outcome, lines)
|
||||
}
|
||||
|
||||
/// The #90455 deadlock: a child exits, but a descendant it spawned still
|
||||
/// holds the inherited write end of the pipe, so EOF never comes. The pump
|
||||
/// must return on the *process* exiting and abandon the drain.
|
||||
///
|
||||
/// The Windows half of this contract lives in `-SelfTestPipeDrain`
|
||||
/// (scripts/desktop-update/windows.ps1) and runs on the Windows CI lane; the
|
||||
/// pump logic under test here is OS-agnostic, only the fixture is not.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pump_child_returns_when_a_grandchild_still_holds_the_pipe() {
|
||||
// `sleep` inherits stdout and outlives the shell by design. Nothing
|
||||
// redirects it -- redirecting is what would close the handle and stop
|
||||
// the bug from reproducing at all.
|
||||
let mut child = sh("sleep 30 & echo hello; exit 7");
|
||||
let started = std::time::Instant::now();
|
||||
let (outcome, lines) = pump_collect(&mut child, Duration::from_millis(300)).await;
|
||||
let elapsed = started.elapsed();
|
||||
|
||||
assert!(outcome.abandoned, "drain should have been abandoned");
|
||||
assert_eq!(
|
||||
outcome.exit_code,
|
||||
Some(7),
|
||||
"exit code must survive an abandoned drain"
|
||||
);
|
||||
assert_eq!(
|
||||
lines,
|
||||
vec!["hello"],
|
||||
"output written before the pipe was stranded must survive"
|
||||
);
|
||||
assert!(!outcome.killed);
|
||||
// Far below the grandchild's 30s: a pass cannot be it exiting on its own.
|
||||
assert!(elapsed < Duration::from_secs(10), "took {elapsed:?}");
|
||||
}
|
||||
|
||||
/// The other cliff: a child that leaks nothing must not pay the grace. This
|
||||
/// is what fails if the pump ever waits on the deadline unconditionally
|
||||
/// instead of only when a pipe outlives its process.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pump_child_does_not_pay_the_grace_when_pipes_close_cleanly() {
|
||||
let mut child = sh("echo one; echo two >&2; echo three; exit 3");
|
||||
let started = std::time::Instant::now();
|
||||
let (outcome, lines) = pump_collect(&mut child, Duration::from_secs(30)).await;
|
||||
let elapsed = started.elapsed();
|
||||
|
||||
assert!(!outcome.abandoned);
|
||||
assert_eq!(outcome.exit_code, Some(3));
|
||||
assert_eq!(lines, vec!["one", "three"]);
|
||||
assert!(elapsed < Duration::from_secs(10), "took {elapsed:?}");
|
||||
}
|
||||
|
||||
/// A chatty child must stream at pipe speed, not at one buffer per tick.
|
||||
/// The PowerShell port of this pump regressed exactly here: idling after
|
||||
/// every chunk it *did* read metered the drain and backpressured the running
|
||||
/// child. `hermes update` is this shape -- the Electron build alone is
|
||||
/// megabytes.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pump_child_streams_a_flood_without_metering_it() {
|
||||
let mut child = sh("i=0; while [ $i -lt 20000 ]; do echo line$i; i=$((i+1)); done; exit 0");
|
||||
let started = std::time::Instant::now();
|
||||
let (outcome, lines) = pump_collect(&mut child, Duration::from_secs(30)).await;
|
||||
let elapsed = started.elapsed();
|
||||
|
||||
assert_eq!(outcome.exit_code, Some(0));
|
||||
assert!(!outcome.abandoned);
|
||||
assert_eq!(lines.len(), 20000, "every line must survive");
|
||||
assert_eq!(lines.last().unwrap(), "line19999");
|
||||
// Loose on purpose: this catches per-chunk sleeping (which would put
|
||||
// this in the tens of seconds), not small scheduler variance.
|
||||
assert!(elapsed < Duration::from_secs(20), "took {elapsed:?}");
|
||||
}
|
||||
|
||||
/// Cancelling kills the child, but not a grandchild holding the pipe --
|
||||
/// `start_kill` only reaches the child. The bounded drain is what actually
|
||||
/// lets a cancel return, so cancellation and the grace are one mechanism.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pump_child_cancellation_returns_even_with_the_pipe_stranded() {
|
||||
let mut child = sh("sleep 30 & echo working; sleep 30");
|
||||
let (tx, rx) = mpsc::channel(1);
|
||||
let mut cancel = Some(rx);
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let _ = tx.send(()).await;
|
||||
});
|
||||
|
||||
let started = std::time::Instant::now();
|
||||
let mut lines = Vec::new();
|
||||
let outcome = pump_child(
|
||||
&mut child,
|
||||
|l| lines.push(l.to_string()),
|
||||
|_| {},
|
||||
&mut cancel,
|
||||
Duration::from_millis(300),
|
||||
)
|
||||
.await
|
||||
.expect("pump");
|
||||
|
||||
assert!(outcome.killed, "cancellation should report killed");
|
||||
assert!(outcome.abandoned, "the grandchild still holds the pipe");
|
||||
assert_eq!(lines, vec!["working"]);
|
||||
assert!(
|
||||
started.elapsed() < Duration::from_secs(10),
|
||||
"took {:?}",
|
||||
started.elapsed()
|
||||
);
|
||||
}
|
||||
|
||||
/// The reverse topology of the test above: the pipes die and the *process*
|
||||
/// outlives them. Phase 1 leaves on EOF with no exit status, so the final
|
||||
/// wait is the only thing left holding the turn -- and it has to stay
|
||||
/// cancellable. A bare `child.wait()` there strands the cancel channel
|
||||
/// against a child that may never exit on its own.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pump_child_cancellation_returns_after_both_pipes_reach_eof() {
|
||||
// Closes fd 1 and 2, then lingers: both reads hit EOF immediately while
|
||||
// the process stays alive far past any plausible test duration.
|
||||
let mut child = sh("echo bye; exec 1>&- 2>&-; sleep 30");
|
||||
let (tx, rx) = mpsc::channel(1);
|
||||
let mut cancel = Some(rx);
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
let _ = tx.send(()).await;
|
||||
});
|
||||
|
||||
let started = std::time::Instant::now();
|
||||
let mut lines = Vec::new();
|
||||
let outcome = pump_child(
|
||||
&mut child,
|
||||
|l| lines.push(l.to_string()),
|
||||
|_| {},
|
||||
&mut cancel,
|
||||
Duration::from_millis(300),
|
||||
)
|
||||
.await
|
||||
.expect("pump");
|
||||
let elapsed = started.elapsed();
|
||||
|
||||
assert!(outcome.killed, "cancellation should report killed");
|
||||
assert!(
|
||||
!outcome.abandoned,
|
||||
"both pipes reached EOF, so nothing was abandoned"
|
||||
);
|
||||
assert_eq!(lines, vec!["bye"], "output written before EOF must survive");
|
||||
// Far below the child's 30s: a pass cannot be it exiting on its own.
|
||||
assert!(elapsed < Duration::from_secs(10), "took {elapsed:?}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,11 +31,10 @@ use std::time::{Duration, Instant};
|
||||
|
||||
use anyhow::{anyhow, Result};
|
||||
use tauri::{AppHandle, Emitter};
|
||||
use tokio::io::BufReader;
|
||||
use tokio::process::Command;
|
||||
|
||||
use crate::events::{BootstrapEvent, LogStream, StageInfo, StageState};
|
||||
use crate::powershell::read_decoded_line;
|
||||
use crate::powershell::{pump_child, DRAIN_GRACE};
|
||||
|
||||
/// `hermes update` exit code meaning "another hermes process is holding the
|
||||
/// venv shim open / dirty precondition" — see _cmd_update_impl in
|
||||
@@ -825,39 +824,34 @@ async fn run_streamed(
|
||||
.spawn()
|
||||
.map_err(|e| anyhow!("spawning {} {:?}: {e}", program.display(), args))?;
|
||||
|
||||
let stdout = child.stdout.take().expect("stdout piped");
|
||||
let stderr = child.stderr.take().expect("stderr piped");
|
||||
// Same non-UTF-8-safe decode path as powershell::run_script (#67193).
|
||||
let mut out = BufReader::new(stdout);
|
||||
let mut err = BufReader::new(stderr);
|
||||
let mut out_buf = Vec::new();
|
||||
let mut err_buf = Vec::new();
|
||||
|
||||
// Same non-UTF-8-safe decode path as powershell::run_script (#67193), and
|
||||
// the same rule about pipe EOF: `hermes update` is precisely the shape that
|
||||
// leaves resident descendants holding an inherited stdout handle, and every
|
||||
// stage this drives sits downstream of the read.
|
||||
let stage_owned = stage.map(|s| s.to_string());
|
||||
loop {
|
||||
tokio::select! {
|
||||
line = read_decoded_line(&mut out, &mut out_buf) => match line {
|
||||
Ok(Some(l)) => emit_log(app, stage_owned.as_deref(), LogStream::Stdout, &l),
|
||||
Ok(None) => break,
|
||||
Err(e) => { tracing::warn!("stdout read error: {e}"); break; }
|
||||
},
|
||||
line = read_decoded_line(&mut err, &mut err_buf) => match line {
|
||||
Ok(Some(l)) => emit_log(app, stage_owned.as_deref(), LogStream::Stderr, &l),
|
||||
Ok(None) => {}
|
||||
Err(e) => { tracing::warn!("stderr read error: {e}"); }
|
||||
},
|
||||
}
|
||||
}
|
||||
while let Ok(Some(l)) = read_decoded_line(&mut out, &mut out_buf).await {
|
||||
emit_log(app, stage_owned.as_deref(), LogStream::Stdout, &l);
|
||||
}
|
||||
while let Ok(Some(l)) = read_decoded_line(&mut err, &mut err_buf).await {
|
||||
emit_log(app, stage_owned.as_deref(), LogStream::Stderr, &l);
|
||||
let outcome = pump_child(
|
||||
&mut child,
|
||||
|l| emit_log(app, stage_owned.as_deref(), LogStream::Stdout, l),
|
||||
|l| emit_log(app, stage_owned.as_deref(), LogStream::Stderr, l),
|
||||
&mut None,
|
||||
DRAIN_GRACE,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| anyhow!("streaming {} {:?}: {e}", program.display(), args))?;
|
||||
|
||||
if outcome.abandoned {
|
||||
let note = format!(
|
||||
"{} exited but a surviving descendant still holds its stdout/stderr; \
|
||||
gave up on the last {}s of output (#90455)",
|
||||
program.display(),
|
||||
DRAIN_GRACE.as_secs()
|
||||
);
|
||||
tracing::warn!("{note}");
|
||||
emit_log(app, stage_owned.as_deref(), LogStream::Stderr, ¬e);
|
||||
}
|
||||
|
||||
let status = child.wait().await.map_err(|e| anyhow!("waiting for child: {e}"))?;
|
||||
Ok(CmdResult {
|
||||
exit_code: status.code(),
|
||||
exit_code: outcome.exit_code,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -25,6 +25,9 @@ Lanes:
|
||||
must not run it.
|
||||
* ``npm_lock`` — semantic package-lock.json diff PR comment.
|
||||
* ``installer`` — PowerShell installer tests (Windows runner).
|
||||
* ``rust`` — ``cargo test`` for the Tauri bootstrap installer. ``.rs``
|
||||
lives under ``apps/``, so without this lane a Rust change matched ``frontend``
|
||||
and only the TypeScript matrix ran.
|
||||
* ``mcp_catalog`` — bundled MCP catalog / installer review.
|
||||
|
||||
Docker is not a lane — it builds on push-to-main and release only,
|
||||
@@ -106,6 +109,13 @@ _MCP_CATALOG_FILES = {"hermes_cli/mcp_catalog.py"}
|
||||
_INSTALLER_PATHS = ("scripts/tests/",)
|
||||
_INSTALLER_FILES = {"scripts/install.ps1", "scripts/install.cmd"}
|
||||
|
||||
# Rust crates — currently just the Tauri bootstrap installer (Hermes-Setup).
|
||||
# These live under ``apps/``, so before this lane existed a ``.rs`` edit matched
|
||||
# ``frontend`` and nothing more: the TypeScript matrix built, cargo never ran,
|
||||
# and the crate's unit tests had never executed in CI at all.
|
||||
_RUST_PATHS = ("apps/bootstrap-installer/src-tauri/",)
|
||||
_RUST_FILENAMES = {"Cargo.toml", "Cargo.lock"}
|
||||
|
||||
def _is_docs(p: str) -> bool:
|
||||
if p.startswith(("skills/", "optional-skills/")):
|
||||
return False
|
||||
@@ -152,6 +162,14 @@ def _is_installer(p: str) -> bool:
|
||||
return p.startswith(_INSTALLER_PATHS) or p in _INSTALLER_FILES
|
||||
|
||||
|
||||
def _is_rust(p: str) -> bool:
|
||||
return (
|
||||
p.endswith(".rs")
|
||||
or p.startswith(_RUST_PATHS)
|
||||
or os.path.basename(p) in _RUST_FILENAMES
|
||||
)
|
||||
|
||||
|
||||
def _is_ci_review(p: str) -> bool:
|
||||
if p in _CI_REVIEW_FILES or p.startswith(_CI_REVIEW_PATHS):
|
||||
return True
|
||||
@@ -187,6 +205,7 @@ def classify(files: list[str]) -> dict[str, bool]:
|
||||
"uv_lock": any(f in ("pyproject.toml", "uv.lock") for f in files),
|
||||
"npm_lock": npm_lock,
|
||||
"installer": any(_is_installer(f) for f in files),
|
||||
"rust": any(_is_rust(f) for f in files),
|
||||
"mcp_catalog": any(_is_mcp_catalog(f) for f in files),
|
||||
"ci_review": any(_is_ci_review(f) for f in files),
|
||||
"nix": python_prod or frontend or any(_is_nix(f) for f in files)
|
||||
@@ -203,6 +222,7 @@ def classify(files: list[str]) -> dict[str, bool]:
|
||||
ret["uv_lock"] = True
|
||||
ret["npm_lock"] = True
|
||||
ret["installer"] = True
|
||||
ret["rust"] = True
|
||||
ret["nix"] = True
|
||||
ret["ci_review"] = True
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ change could have broken.
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -34,12 +35,13 @@ DEFAULT = {
|
||||
"uv_lock": True,
|
||||
"npm_lock": True,
|
||||
"installer": True,
|
||||
"rust": True,
|
||||
"mcp_catalog": False,
|
||||
"ci_review": True,
|
||||
}
|
||||
|
||||
|
||||
def _lanes(python=False, frontend=False, site=False, scan=False, deps=False, uv_lock=False, npm_lock=False, installer=False, mcp_catalog=False, docker_meta=False, ci_review=False, python_prod=None, nix=None, docker=None) -> dict[str, bool]:
|
||||
def _lanes(python=False, frontend=False, site=False, scan=False, deps=False, uv_lock=False, npm_lock=False, installer=False, rust=False, mcp_catalog=False, docker_meta=False, ci_review=False, python_prod=None, nix=None, docker=None) -> dict[str, bool]:
|
||||
# python_prod tracks python except for tests-only diffs; default it to
|
||||
# python so the majority of cases don't need to spell it out.
|
||||
#
|
||||
@@ -61,6 +63,7 @@ def _lanes(python=False, frontend=False, site=False, scan=False, deps=False, uv_
|
||||
"uv_lock": uv_lock,
|
||||
"npm_lock": npm_lock,
|
||||
"installer": installer,
|
||||
"rust": rust,
|
||||
"mcp_catalog": mcp_catalog,
|
||||
"ci_review": ci_review,
|
||||
}
|
||||
@@ -132,6 +135,26 @@ CASES = {
|
||||
_lanes(python=True, installer=True),
|
||||
),
|
||||
"python source alone → no installer lane": (["run_agent.py"], _lanes(python=True, scan=True)),
|
||||
# `.rs` lives under apps/, so it matches `frontend` too. That lane builds
|
||||
# TypeScript and cannot notice a Rust error — before `rust` existed it was
|
||||
# the ONLY lane a Rust change ran, and the crate's tests never executed.
|
||||
"rust source → rust": (
|
||||
["apps/bootstrap-installer/src-tauri/src/powershell.rs"],
|
||||
_lanes(frontend=True, rust=True),
|
||||
),
|
||||
"cargo lockfile → rust": (
|
||||
["apps/bootstrap-installer/src-tauri/Cargo.lock"],
|
||||
_lanes(frontend=True, rust=True),
|
||||
),
|
||||
# Non-.rs files in the crate still change what cargo builds.
|
||||
"tauri config → rust": (
|
||||
["apps/bootstrap-installer/src-tauri/tauri.conf.json"],
|
||||
_lanes(frontend=True, rust=True),
|
||||
),
|
||||
"ts source alone → no rust lane": (
|
||||
["apps/bootstrap-installer/src/main.tsx"],
|
||||
_lanes(frontend=True),
|
||||
),
|
||||
# Unknown top-level file keeps Python on rather than risk a silent skip.
|
||||
"unknown toplevel → python": (["Makefile"], _lanes(python=True)),
|
||||
"mixed docs+python → python": (["README.md", "agent/x.py"], _lanes(python=True, scan=True)),
|
||||
@@ -218,6 +241,55 @@ def test_classify(files, expected):
|
||||
assert classify(files) == expected
|
||||
|
||||
|
||||
_REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _yaml(rel: str) -> dict:
|
||||
yaml = pytest.importorskip("yaml")
|
||||
return yaml.safe_load((_REPO / rel).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_every_lane_reaches_the_composite_action():
|
||||
"""The action is the one surface every consumer reads, so it must carry all
|
||||
of them — ci.yaml, nix.yml and docker.yml each re-export a different subset.
|
||||
"""
|
||||
lanes = set(classify(["run_agent.py"]))
|
||||
action_outputs = set(_yaml(".github/actions/detect-changes/action.yml")["outputs"])
|
||||
assert lanes - action_outputs == set(), "lane(s) missing from the composite action's outputs"
|
||||
|
||||
|
||||
def test_ci_jobs_only_gate_on_detect_outputs_that_detect_actually_declares():
|
||||
"""An ``if`` that reads an undeclared output resolves to the empty string.
|
||||
|
||||
The lane then reports "skipping" on every PR, forever, and nothing goes red
|
||||
— there is no error for referencing an output a job never declared. That is
|
||||
exactly how the ``rust`` lane shipped dead: the classifier emitted it and
|
||||
the composite action re-exported it, but ci.yaml's ``detect`` job did not,
|
||||
so ``needs.detect.outputs.rust`` was never anything but "".
|
||||
"""
|
||||
ci = _yaml(".github/workflows/ci.yaml")
|
||||
declared = set(ci["jobs"]["detect"]["outputs"])
|
||||
|
||||
referenced: set[str] = set()
|
||||
for job in ci["jobs"].values():
|
||||
for expr in _iter_if_expressions(job):
|
||||
referenced.update(re.findall(r"needs\.detect\.outputs\.(\w+)", expr))
|
||||
|
||||
assert referenced, "found no detect-gated jobs — the walk is broken, not the wiring"
|
||||
assert referenced - declared == set(), "job(s) gate on an output detect never declares"
|
||||
|
||||
|
||||
def _iter_if_expressions(job: object):
|
||||
"""Yield every ``if:`` string in a job, including inside its steps."""
|
||||
if not isinstance(job, dict):
|
||||
return
|
||||
if isinstance(cond := job.get("if"), str):
|
||||
yield cond
|
||||
for step in job.get("steps", []) or []:
|
||||
if isinstance(step, dict) and isinstance(cond := step.get("if"), str):
|
||||
yield cond
|
||||
|
||||
|
||||
def test_ci_review_files_returns_only_sensitive_paths_sorted_and_unique():
|
||||
assert ci_review_files([
|
||||
"apps/desktop/src/app.tsx",
|
||||
|
||||
Reference in New Issue
Block a user