Skip to main content

max / alloy

run: bound the wait for a reaped command's last output Draining to EOF after reaping the child assumed the wait was bounded because the child was dead. A pipe reaches EOF only when every holder of the write end closes it, and descendants inherit that end: bootc install to-disk drives podman, ostree and bootupd, so a grandchild outliving its parent holds the pipe open. The drain then blocked the render thread for as long as the grandchild lived. A probe stage running sh -c "sleep 30 & echo parent-done" froze a single Sequence::poll for 30.0s. On the run screen that is a stopped frame during a disk write, which reads as a hung install. Drain a slice at a time across ticks instead, with the wait ending on EOF or on half a second of silence, whichever comes first. The grace runs from the last line rather than from the exit, so a command still producing output is never cut off, which is what keeps a resolver seeing its command's complete output.
Author: Max Johnson <me@maxj.phd> · 2026-07-22 21:49 UTC
Signed with PGP, not checked
Commit: 037decbdf3f7a34e0917329313854fc41437a7f0
Parent: 4bffd6f
1 file changed, +194 insertions, -60 deletions
M crates/alloy/src/run.rs +194 -60
@@ -9,15 +9,17 @@
9 9 //!
10 10 //! The shell's loop is synchronous by design and this does not change that. A
11 11 //! reader thread per stream pushes lines into a channel; the view drains the
12 - //! channel on the shell's tick. Nothing here blocks for longer than it takes to
13 - //! empty a queue, so the frame keeps rendering at the tick rate whether the
14 - //! child is talkative or silent for a minute.
12 + //! channel on the shell's tick. Nothing here waits on the child, and the one
13 + //! place that waits on its output waits for [`DRAIN_SLICE`] at a time, so the
14 + //! frame keeps rendering at the tick rate whether the child is talkative or
15 + //! silent for a minute.
15 16
16 17 use std::collections::VecDeque;
17 18 use std::io::{BufRead, BufReader};
18 19 use std::process::{Child, ExitStatus};
19 - use std::sync::mpsc::{Receiver, TryRecvError, channel};
20 + use std::sync::mpsc::{Receiver, RecvTimeoutError, TryRecvError, channel};
20 21 use std::thread;
22 + use std::time::{Duration, Instant};
21 23
22 24 use crate::cli::{CommandLog, Invocation};
23 25
@@ -28,6 +30,20 @@
28 30 /// the interesting lines during a failure are the last ones.
29 31 const SCROLLBACK: usize = 500;
30 32
33 + /// How long one [`poll`](Sequence::poll) may wait for the output a reaped
34 + /// command left in the pipe.
35 + ///
36 + /// Long enough that the usual case costs a single tick: the pump thread is
37 + /// sitting in a blocking read, so lines already written arrive in well under a
38 + /// millisecond. Short enough to be invisible against the shell's tick.
39 + const DRAIN_SLICE: Duration = Duration::from_millis(50);
40 +
41 + /// How long the sequence keeps waiting after a reaped command has gone quiet.
42 + ///
43 + /// Measured from the last line rather than from the exit, so a command that is
44 + /// still producing is never cut off. Only silence ends the wait.
45 + const DRAIN_GRACE: Duration = Duration::from_millis(500);
46 +
31 47 /// A command, and optionally what its output decides.
32 48 ///
33 49 /// The installer needs this because two of its arguments cannot be known when
@@ -104,18 +120,53 @@
104 120 /// terminal, and a progress line on stdout followed by its error on stderr
105 121 /// is only legible in that order.
106 122 lines: Receiver<String>,
123 + /// Set once the child has been reaped, `None` while it runs.
124 + exited: Option<Exited>,
125 + }
126 +
127 + /// A reaped child, and how long the sequence will go on collecting the output
128 + /// it left behind in the pipe.
129 + struct Exited {
130 + status: ExitStatus,
131 + /// When to stop waiting. Pushed back by every line that arrives, so this
132 + /// is a deadline on silence rather than on the command.
133 + quiet_by: Instant,
134 + }
135 +
136 + /// What a drain found.
137 + struct Drained {
138 + /// Every line has been forwarded: both pump threads reached EOF and
139 + /// dropped their senders.
140 + complete: bool,
141 + /// How many lines this drain took, which is what extends the grace.
142 + lines: usize,
107 143 }
108 144
109 145 impl Running {
110 146 /// Drain whatever has arrived, without waiting for more.
111 - fn drain(&mut self, into: &mut Vec<String>) {
147 + fn drain(&mut self, into: &mut Vec<String>) -> Drained {
148 + let mut lines = 0;
112 149 loop {
113 150 match self.lines.try_recv() {
114 - Ok(line) => self.take(line, into),
151 + Ok(line) => {
152 + self.take(line, into);
153 + lines += 1;
154 + }
155 + Err(TryRecvError::Empty) => {
156 + return Drained {
157 + complete: false,
158 + lines,
159 + };
160 + }
115 161 // Disconnected means both readers are done, which is a fact
116 162 // about the pipes rather than about the process: the child may
117 163 // still be exiting. `finished` is what decides that.
118 - Err(TryRecvError::Empty | TryRecvError::Disconnected) => return,
164 + Err(TryRecvError::Disconnected) => {
165 + return Drained {
166 + complete: true,
167 + lines,
168 + };
169 + }
119 170 }
120 171 }
121 172 }
@@ -134,22 +185,53 @@
134 185 self.captured.join("\n")
135 186 }
136 187
137 - /// Drain everything the command will ever produce, blocking until the
138 - /// readers are done.
188 + /// Drain for up to `slice`, stopping early once every line has arrived.
139 189 ///
140 - /// Only correct once the child has exited, and only safe then: reaping the
141 - /// child is not the same event as its output arriving. `try_wait` can reap
142 - /// a process whose last lines are still sitting in the pipe, unread by the
143 - /// pump threads, and a `try_recv` at that moment reports `Empty` and takes
144 - /// the lines to the grave with the `Running` that gets dropped next.
190 + /// For the window after the child has been reaped. Reaping is not the same
191 + /// event as the output arriving: `try_wait` can reap a process whose last
192 + /// lines are still sitting in the pipe, unread by the pump threads, and a
193 + /// `try_recv` at that moment reports `Empty` and takes the lines to the
194 + /// grave with the `Running` that gets dropped next. So this waits, which a
195 + /// drain during the run must never do.
145 196 ///
146 - /// Blocking to `Disconnected` waits for the one event that means all
147 - /// output has been forwarded: both pump threads reaching EOF and dropping
148 - /// their senders. The wait is bounded because the child is already dead,
149 - /// so both pipes are closed and EOF is already on its way.
150 - fn drain_to_end(&mut self, into: &mut Vec<String>) {
151 - while let Ok(line) = self.lines.recv() {
152 - self.take(line, into);
197 + /// `Disconnected` is the one event that means all output has been
198 + /// forwarded, but waiting for it outright is what the caller must not do.
199 + /// A pipe reaches EOF when every holder of the write end closes it, and the
200 + /// child's descendants inherit that end: `bootc install to-disk` drives
201 + /// podman, ostree and bootupd, so a grandchild outliving its parent holds
202 + /// the pipe open with the parent already reaped. Blocking there froze the
203 + /// frame for as long as the grandchild lived, which on the run screen is
204 + /// indistinguishable from a hung install. A slice at a time, across ticks,
205 + /// costs the same in the ordinary case and cannot freeze in that one.
206 + fn drain_briefly(&mut self, into: &mut Vec<String>, slice: Duration) -> Drained {
207 + let until = Instant::now() + slice;
208 + let mut lines = 0;
209 + loop {
210 + let left = until.saturating_duration_since(Instant::now());
211 + if left.is_zero() {
212 + return Drained {
213 + complete: false,
214 + lines,
215 + };
216 + }
217 + match self.lines.recv_timeout(left) {
218 + Ok(line) => {
219 + self.take(line, into);
220 + lines += 1;
221 + }
222 + Err(RecvTimeoutError::Timeout) => {
223 + return Drained {
224 + complete: false,
225 + lines,
226 + };
227 + }
228 + Err(RecvTimeoutError::Disconnected) => {
229 + return Drained {
230 + complete: true,
231 + lines,
232 + };
233 + }
234 + }
153 235 }
154 236 }
155 237
@@ -237,51 +319,68 @@
237 319 }
238 320
239 321 if let Some(running) = &mut self.current {
240 - let mut lines = std::mem::take(&mut self.output);
241 - running.drain(&mut lines);
242 - self.output = lines;
322 + if running.exited.is_none() {
323 + let mut lines = std::mem::take(&mut self.output);
324 + running.drain(&mut lines);
325 + self.output = lines;
243 326
244 - match running.finished() {
245 - None => return,
246 - Some(Err(err)) => {
247 - self.fail(format!("could not wait for the command: {err}"));
248 - return;
249 - }
250 - Some(Ok(status)) => {
251 - // The pipes can still hold output written just before exit,
252 - // so drain once more after reaping. Without this the last
253 - // lines of a failing command — the ones that say why — are
254 - // the ones that get lost, and a resolver reading this
255 - // command's output parses whatever happened to arrive in
256 - // time. Blocking, because a non-blocking drain here is a
257 - // race against the pump threads that the pumps often lose.
258 - let mut lines = std::mem::take(&mut self.output);
259 - let mut finished = self.current.take().expect("current was Some");
260 - finished.drain_to_end(&mut lines);
261 - self.output = lines;
262 - self.done_count += 1;
263 -
264 - if !status.success() {
265 - self.fail(format!("command exited with {status}"));
327 + match running.finished() {
328 + None => return,
329 + Some(Err(err)) => {
330 + self.fail(format!("could not wait for the command: {err}"));
266 331 return;
267 332 }
333 + Some(Ok(status)) => {
334 + running.exited = Some(Exited {
335 + status,
336 + quiet_by: Instant::now() + DRAIN_GRACE,
337 + });
338 + }
339 + }
340 + }
268 341
269 - if let Some(resolve) = finished.then.take() {
270 - let captured = finished.captured();
271 - match resolve(&captured) {
272 - // Pushed to the front: what a discovery decides
273 - // runs before whatever was queued behind it.
274 - Ok(stages) => {
275 - for stage in stages.into_iter().rev() {
276 - self.queue.push_front(stage);
277 - }
278 - }
279 - Err(message) => {
280 - self.fail(message);
281 - return;
282 - }
342 + // Reaped, so what is left is whatever was written just before exit
343 + // and is still in flight. Without collecting it, the lines that get
344 + // lost are the last ones a failing command wrote, which are the
345 + // ones that say why, and a resolver reading this command's output
346 + // parses whatever happened to arrive in time.
347 + let mut lines = std::mem::take(&mut self.output);
348 + let drained = running.drain_briefly(&mut lines, DRAIN_SLICE);
349 + self.output = lines;
350 +
351 + let exited = running.exited.as_mut().expect("reaped");
352 + if drained.lines > 0 {
353 + exited.quiet_by = Instant::now() + DRAIN_GRACE;
354 + }
355 + // Not done, but still producing or recently so: come back next tick
356 + // rather than holding the frame here.
357 + if !drained.complete && Instant::now() < exited.quiet_by {
358 + return;
359 + }
360 +
361 + let mut finished = self.current.take().expect("current was Some");
362 + let status = finished.exited.take().expect("reaped").status;
363 + self.done_count += 1;
364 +
365 + if !status.success() {
366 + self.fail(format!("command exited with {status}"));
367 + return;
368 + }
369 +
370 + if let Some(resolve) = finished.then.take() {
371 + let captured = finished.captured();
372 + match resolve(&captured) {
373 + // Pushed to the front: what a discovery decides runs before
374 + // whatever was queued behind it.
375 + Ok(stages) => {
376 + for stage in stages.into_iter().rev() {
377 + self.queue.push_front(stage);
283 378 }
284 379 }
380 + Err(message) => {
381 + self.fail(message);
382 + return;
383 + }
285 384 }
286 385 }
287 386 }
@@ -339,6 +438,7 @@
339 438 captured: Vec::new(),
340 439 then: None,
341 440 lines: receiver,
441 + exited: None,
342 442 })
343 443 }
344 444
@@ -438,6 +538,40 @@
438 538 assert_eq!(sequence.output(), ["a", "b", "c"]);
439 539 }
440 540
541 + // A grandchild that outlives its parent holds the write end of the pipe,
542 + // so EOF does not arrive when the parent is reaped. Waiting for it froze
543 + // the render thread for as long as the grandchild lived: on the run screen
544 + // the frame stops and the install looks hung, which is the one screen where
545 + // the user's alternative is power-cycling a half-written disk. `bootc
546 + // install to-disk` drives podman, ostree and bootupd, so this is the
547 + // ordinary case rather than a contrived one.
548 + //
549 + // The sleep is far longer than the sequence should ever take, so a
550 + // regression here shows up as elapsed time rather than as a wrong value.
551 + #[test]
552 + fn a_grandchild_holding_the_pipe_does_not_stall_the_sequence() {
553 + let mut log = CommandLog::new();
554 + let mut sequence = Sequence::new(vec![Stage::Run(
555 + Invocation::new("sh").args(["-c", "sleep 30 & echo parent-done"]),
556 + )]);
557 +
558 + let started = Instant::now();
559 + drive(&mut sequence, &mut log);
560 + let elapsed = started.elapsed();
561 +
562 + assert_eq!(sequence.outcome(), Some(&Ok(())));
563 + // The parent's own output is still complete.
564 + assert!(
565 + sequence.output().iter().any(|l| l == "parent-done"),
566 + "{:?}",
567 + sequence.output()
568 + );
569 + assert!(
570 + elapsed < Duration::from_secs(5),
571 + "the sequence waited on the grandchild for {elapsed:?}"
572 + );
573 + }
574 +
441 575 #[test]
442 576 fn a_command_that_does_not_exist_fails_the_sequence() {
443 577 let mut log = CommandLog::new();