From 7050e3f4d0517859b70c3b03ae12ae2b8478f47f Mon Sep 17 00:00:00 2001 From: Graeme Arthur Date: Sat, 22 Aug 2026 16:55:08 +1000 Subject: [PATCH 1/2] Wait on conditions rather than sleeps in the command centre tests Twelve tests slept for a fixed period and then asserted what an AsyncStream subscriber had collected. The sleep was sizing how long delivery was expected to take, which is a guess in both directions: too short and a loaded runner fails a test that is merely late, too long and every run pays for the worst case. Several asserted exact contents, so late delivery failed as wrongness. The two coalescing tests were a different shape: the sleep sized an overlap window, holding that the first command was still in flight when the duplicate arrived. TestGate holds the runner open until the test lets it go. The recording centre logs the duplicate submission before it coalesces, so that half is observable too; the plain centre offers nothing equivalent, and its remaining window is between the second task signalling and its call reaching the actor, with the first command held. The two cleanup tests assert an absence, which polling cannot express. They await the cancelled task so termination is done rather than assumed, and a live witness subscriber observes the broadcast. Verified by mutation: dropping the output broadcast fails these tests rather than passing vacuously. Co-Authored-By: Claude Opus 5 --- .../SerialBrewCommandCenterOutputTests.swift | 23 ++++++-- .../SerialBrewCommandCenterTests.swift | 53 ++++++++++++++----- 2 files changed, 59 insertions(+), 17 deletions(-) diff --git a/Tests/BrewCLITests/SerialBrewCommandCenterOutputTests.swift b/Tests/BrewCLITests/SerialBrewCommandCenterOutputTests.swift index 74c1ca12..947d1e59 100644 --- a/Tests/BrewCLITests/SerialBrewCommandCenterOutputTests.swift +++ b/Tests/BrewCLITests/SerialBrewCommandCenterOutputTests.swift @@ -61,7 +61,7 @@ struct SerialBrewCommandCenterOutputTests { defer { collect.cancel() } try await center.capture(command("go"), id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await collector.events.count == 4 } let lines = await collector.events.filter { $0.0 == id }.map(\.1) #expect(lines.map(\.text) == ["one", "two", "warn", "three"]) @@ -85,7 +85,7 @@ struct SerialBrewCommandCenterOutputTests { try await center.capture(command("a"), id: idA) try await center.capture(command("b"), id: idB) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await collector.events.count == 3 } let events = await collector.events #expect(events.filter { $0.0 == idA }.map(\.1.text) == ["a1", "a2"]) @@ -115,7 +115,11 @@ struct SerialBrewCommandCenterOutputTests { } try await center.capture(command("go"), id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { + let countA = await collectorA.events.count + let countB = await collectorB.events.count + return countA == 2 && countB == 2 + } let textsA = await collectorA.events.map(\.1.text) let textsB = await collectorB.events.map(\.1.text) @@ -134,10 +138,19 @@ struct SerialBrewCommandCenterOutputTests { } } collect.cancel() - try await Task.sleep(for: .milliseconds(20)) + await collect.value + + let witnessStream = await center.allOutputChanges() + let witness = AllOutputCollector() + let observe = Task { + for await pair in witnessStream { + await witness.append(id: pair.0, line: pair.1) + } + } + defer { observe.cancel() } try await center.capture(command("go"), id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await witness.events.count == 1 } let events = await collector.events #expect(events.isEmpty) diff --git a/Tests/BrewCLITests/SerialBrewCommandCenterTests.swift b/Tests/BrewCLITests/SerialBrewCommandCenterTests.swift index 10c5ea4e..3f4e81b1 100644 --- a/Tests/BrewCLITests/SerialBrewCommandCenterTests.swift +++ b/Tests/BrewCLITests/SerialBrewCommandCenterTests.swift @@ -68,16 +68,24 @@ struct SerialBrewCommandCenterTests { @Test func `duplicate id coalesces to a single command body`() async throws { let counter = InvocationCounter() + let gate = TestGate() let center = makeCenter(runner: ClosureRunner { _ in await counter.increment() - try await Task.sleep(for: .milliseconds(30)) + await gate.wait() return successOutput }) let id = BrewOperationID(kind: .formula, name: "git") let first = Task { try await center.perform(noopCommand, id: id) } - try await Task.sleep(for: .milliseconds(5)) - let second = Task { try await center.perform(noopCommand, id: id) } + try await waitUntil { await counter.value == 1 } + + let submitting = InvocationCounter() + let second = Task { + await submitting.increment() + try await center.perform(noopCommand, id: id) + } + try await waitUntil { await submitting.value == 1 } + await gate.open() try await first.value try await second.value @@ -157,7 +165,7 @@ struct SerialBrewCommandCenterTests { defer { collect.cancel() } try await center.perform(noopCommand, id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await collector.phases.count >= 3 } let values = await collector.phases #expect(values.count >= 3) #expect(values.first == .idle) @@ -190,7 +198,11 @@ struct SerialBrewCommandCenterTests { } try await center.perform(noopCommand, id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { + let countA = await collectorA.phases.count + let countB = await collectorB.phases.count + return countA >= 3 && countB >= 3 + } let countA = await collectorA.phases.count let countB = await collectorB.phases.count #expect(countA >= 3) @@ -199,10 +211,11 @@ struct SerialBrewCommandCenterTests { @Test func `recording wrapper logs each submit while duplicate id coalesces body`() async throws { let counter = InvocationCounter() + let gate = TestGate() let ctx = BrewCommandExecutionContext( commandRunner: ClosureRunner { _ in await counter.increment() - try await Task.sleep(for: .milliseconds(30)) + await gate.wait() return successOutput }, locator: BrewExecutableLocator(overrideURL: InstalledPackagesTestSupport.fakeBrewExecutableURL), @@ -211,8 +224,10 @@ struct SerialBrewCommandCenterTests { let id = BrewOperationID(kind: .formula, name: "git") let first = Task { try await center.perform(noopCommand, id: id) } - try await Task.sleep(for: .milliseconds(5)) + try await waitUntil { await counter.value == 1 } let second = Task { try await center.perform(noopCommand, id: id) } + try await waitUntil { await center.recordedSubmitEntries.count == 2 } + await gate.open() try await first.value try await second.value @@ -240,7 +255,7 @@ struct SerialBrewAllPhaseStreamTests { try await center.perform(noopCommand, id: idA) try await center.perform(noopCommand, id: idB) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await collector.events.count >= 4 } let events = await collector.events let eventsForA = events.filter { $0.0 == idA }.map(\.1) let eventsForB = events.filter { $0.0 == idB }.map(\.1) @@ -279,7 +294,11 @@ struct SerialBrewAllPhaseStreamTests { } try await center.perform(noopCommand, id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { + let countA = await collectorA.events.count + let countB = await collectorB.events.count + return countA >= 2 && countB >= 2 + } let eventsA = await collectorA.events let eventsB = await collectorB.events #expect(eventsA.count == eventsB.count) @@ -301,10 +320,20 @@ struct SerialBrewAllPhaseStreamTests { } } collect.cancel() - try await Task.sleep(for: .milliseconds(20)) + await collect.value + + let witnessStream = await center.allPhaseChanges() + let witness = AllPhaseStreamCollector() + let observe = Task { + for await pair in witnessStream { + await witness.append(id: pair.0, phase: pair.1) + } + } try await center.perform(noopCommand, id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await witness.events.count >= 2 } + observe.cancel() + let eventsAfterCancel = await collector.events #expect(eventsAfterCancel.isEmpty) @@ -318,7 +347,7 @@ struct SerialBrewAllPhaseStreamTests { defer { collect2.cancel() } try await center.perform(noopCommand, id: id) - try await Task.sleep(for: .milliseconds(80)) + try await waitUntil { await collector2.events.count >= 2 } let eventsFresh = await collector2.events #expect(eventsFresh.count >= 2) #expect(eventsFresh.first?.0 == id) From a77ea21a6f85d6ee9c35f1765cddd07bdcdad90b Mon Sep 17 00:00:00 2001 From: Graeme Arthur Date: Wed, 2 Sep 2026 23:35:54 +1000 Subject: [PATCH 2/2] Read the terminal dry before dropping the last replica Darwin discards whatever is still queued on a pty when the last replica descriptor closes. Writing three bytes into the replica and closing it leaves the primary with nothing to read, so this is the device, not anything about the child or the spawn. The previous fix moved that last close into this process: the parent now holds the replica for the whole run and closes it once the child has exited. That put the destructive close immediately after a loop that can exit with bytes still queued. A poll times out, the child writes and exits before isRunning is read, the loop breaks without polling again, and closeReplica destroys the three bytes. An empty transcript, an exit status of 0 and a clean end of input follow, which is the failure CI reported twice. Reading the terminal dry once the child is reaped closes the window. Nothing further can arrive after waitpid returns, so what is queued at that point is all there will ever be. Stalling the gap between the poll timeout and the isRunning check reproduces the failure on demand: 2ms loses output on a quarter of runs and 60ms on all of them, both with the reported signature. With the drain in place every stall passes. The runner is unaffected. Its drain is parked on the primary from before the spawn, so it is woken while the child still holds the replica open. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_015NMRuMKDdQZg32DHwxdMDP --- Tests/BrewCLITests/PseudoTerminalTests.swift | 21 ++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/Tests/BrewCLITests/PseudoTerminalTests.swift b/Tests/BrewCLITests/PseudoTerminalTests.swift index 68cbdcbb..630de55e 100644 --- a/Tests/BrewCLITests/PseudoTerminalTests.swift +++ b/Tests/BrewCLITests/PseudoTerminalTests.swift @@ -127,6 +127,7 @@ private extension PseudoTerminalTests { } } process.waitUntilExit() + drainPending(terminal, into: &data) terminal.closeReplica() drainToEndOfInput(terminal, into: &data) @@ -135,6 +136,26 @@ private extension PseudoTerminalTests { return UTF8StreamDecoder.lossyString(data) } + /// Darwin discards whatever is still queued on the terminal when the last replica descriptor closes, + /// so the child's bytes have to be read before ``PseudoTerminal/closeReplica()``. The child has been + /// reaped by this point, so what is queued now is all there will ever be. + func drainPending(_ terminal: PseudoTerminal, into data: inout Data) { + pending: while true { + switch terminal.read(timeout: .zero) { + case let .data(chunk): + data.append(chunk) + case .timedOut: + break pending + case .endOfInput: + Issue.record("end of input arrived while this process still held the replica open") + break pending + case let .failed(code): + Issue.record("reading the terminal failed: \(PseudoTerminal.describe(errno: code))") + break pending + } + } + } + func drainToEndOfInput(_ terminal: PseudoTerminal, into data: inout Data) { let deadline = Date().addingTimeInterval(10) loop: while true {