Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fa9a2042bf | ||
|
|
e48c3f6840 | ||
|
|
7751701208 |
@@ -27,5 +27,4 @@ ICCery.xcodeproj/
|
|||||||
Release/
|
Release/
|
||||||
notarization/
|
notarization/
|
||||||
build/
|
build/
|
||||||
docs/megaplans/*
|
|
||||||
docs/megaplans
|
docs/megaplans
|
||||||
|
|||||||
@@ -145,9 +145,7 @@ public actor ProcessManager {
|
|||||||
)
|
)
|
||||||
let process = prepared.process
|
let process = prepared.process
|
||||||
|
|
||||||
AppLogger(category: "process").debug(
|
logSpawn(id: id, binary: binary, arguments: arguments, captured: false)
|
||||||
"spawn \(id): \(binary.path) \(LogSanitizer.sanitizeArgs(arguments))"
|
|
||||||
)
|
|
||||||
|
|
||||||
children[id] = RunningChild(
|
children[id] = RunningChild(
|
||||||
process: process,
|
process: process,
|
||||||
@@ -214,9 +212,7 @@ public actor ProcessManager {
|
|||||||
let stdoutPipe = prepared.stdoutPipe
|
let stdoutPipe = prepared.stdoutPipe
|
||||||
let stderrPipe = prepared.stderrPipe
|
let stderrPipe = prepared.stderrPipe
|
||||||
|
|
||||||
AppLogger(category: "process").debug(
|
logSpawn(id: id, binary: binary, arguments: arguments, captured: true)
|
||||||
"spawn(captured) \(id): \(binary.path) \(LogSanitizer.sanitizeArgs(arguments))"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Register and set up the termination hand-off before run() so
|
// Register and set up the termination hand-off before run() so
|
||||||
// a very fast exit is never missed (#50, #52).
|
// a very fast exit is never missed (#50, #52).
|
||||||
@@ -466,6 +462,18 @@ public actor ProcessManager {
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private nonisolated func logSpawn(
|
||||||
|
id: String,
|
||||||
|
binary: URL,
|
||||||
|
arguments: [String],
|
||||||
|
captured: Bool
|
||||||
|
) {
|
||||||
|
let prefix = captured ? "spawn(captured)" : "spawn"
|
||||||
|
AppLogger(category: "process").debug(
|
||||||
|
"\(prefix) \(id): \(binary.path) \(LogSanitizer.sanitizeArgs(arguments))"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
/// `terminationHandler` can lose a fast-exit race on a loaded host;
|
/// `terminationHandler` can lose a fast-exit race on a loaded host;
|
||||||
/// `waitUntilExit` on a detached thread is the fallback (#50, #52).
|
/// `waitUntilExit` on a detached thread is the fallback (#50, #52).
|
||||||
/// The handler is attached before `run()`; the wait thread starts
|
/// The handler is attached before `run()`; the wait thread starts
|
||||||
|
|||||||
@@ -58,6 +58,59 @@ struct ProcessManagerTests {
|
|||||||
func finish() -> Bool { lock.lock(); defer { lock.unlock() }; if finished { return false }; finished = true; return true }
|
func finish() -> Bool { lock.lock(); defer { lock.unlock() }; if finished { return false }; finished = true; return true }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Subscribes synchronously (registration happens inside `events()`)
|
||||||
|
/// then records every event for `id` until the task is cancelled.
|
||||||
|
/// Unlike `collect`, observation continues past `.exit` so tests can
|
||||||
|
/// prove exactly-once exit emission.
|
||||||
|
private func observe(
|
||||||
|
_ manager: ProcessManager,
|
||||||
|
id: String,
|
||||||
|
into box: Box
|
||||||
|
) -> Task<Void, Never> {
|
||||||
|
let stream = manager.events()
|
||||||
|
return Task {
|
||||||
|
for await event in stream {
|
||||||
|
guard event.id == id else { continue }
|
||||||
|
box.append(event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func exitCount(in box: Box) -> Int {
|
||||||
|
box.events.filter { if case .exit = $0 { return true }; return false }.count
|
||||||
|
}
|
||||||
|
|
||||||
|
private func waitForExit(in box: Box, timeout: TimeInterval = 10) async -> Bool {
|
||||||
|
let deadline = Date().addingTimeInterval(timeout)
|
||||||
|
while Date() < deadline {
|
||||||
|
if exitCount(in: box) > 0 { return true }
|
||||||
|
try? await Task.sleep(for: .milliseconds(10))
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
private func waitForFile(_ url: URL, timeout: TimeInterval = 5) async -> Bool {
|
||||||
|
let deadline = Date().addingTimeInterval(timeout)
|
||||||
|
while Date() < deadline {
|
||||||
|
if FileManager.default.fileExists(atPath: url.path) { return true }
|
||||||
|
try? await Task.sleep(for: .milliseconds(10))
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
private func waitForRunning(
|
||||||
|
_ manager: ProcessManager,
|
||||||
|
id: String,
|
||||||
|
timeout: TimeInterval = 5
|
||||||
|
) async -> Bool {
|
||||||
|
let deadline = Date().addingTimeInterval(timeout)
|
||||||
|
while Date() < deadline {
|
||||||
|
if await manager.isRunning(id) { return true }
|
||||||
|
try? await Task.sleep(for: .milliseconds(10))
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// MARK: - Tests
|
// MARK: - Tests
|
||||||
|
|
||||||
@Test func streamsStdoutAndEmitsExit() async throws {
|
@Test func streamsStdoutAndEmitsExit() async throws {
|
||||||
@@ -214,6 +267,145 @@ struct ProcessManagerTests {
|
|||||||
try await pm.sendStdin(id: "nope", text: "d\n")
|
try await pm.sendStdin(id: "nope", text: "d\n")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test func explicitPartialFlushEmitsRowColorsJSON() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let marker = Self.fixtureDir
|
||||||
|
.appendingPathComponent("partial-row-ready-\(UUID().uuidString)")
|
||||||
|
let bin = try script(
|
||||||
|
"partial-row.sh",
|
||||||
|
"#!/bin/sh\nprintf 'ROW_COLORS_JSON: {\"row\":9}'\ntouch \"$1\"\nsleep 30\n"
|
||||||
|
)
|
||||||
|
let box = Box()
|
||||||
|
let observer = observe(pm, id: "t11", into: box)
|
||||||
|
try await pm.runStreaming(id: "t11", binary: bin, arguments: [marker.path])
|
||||||
|
#expect(await waitForFile(marker))
|
||||||
|
// Retry the flush so the pipe-ingest task can win the actor race
|
||||||
|
// on a loaded host; the first successful flush emits the row.
|
||||||
|
var flushed = false
|
||||||
|
for _ in 0..<50 {
|
||||||
|
await pm.flushPartialLine(id: "t11")
|
||||||
|
if box.events.contains(where: { if case .jsonRow = $0 { return true }; return false }) {
|
||||||
|
flushed = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
try await Task.sleep(for: .milliseconds(20))
|
||||||
|
}
|
||||||
|
#expect(flushed)
|
||||||
|
await pm.kill(id: "t11")
|
||||||
|
#expect(await waitForExit(in: box))
|
||||||
|
observer.cancel()
|
||||||
|
let events = box.events
|
||||||
|
let rows = events.compactMap { e -> String? in
|
||||||
|
if case .jsonRow(_, let d) = e { return String(decoding: d, as: UTF8.self) }
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
#expect(rows == ["{\"row\":9}"])
|
||||||
|
// Prefixed tails must not leak into stdout, even via finalize.
|
||||||
|
#expect(!events.contains(.stdout(id: "t11", line: "ROW_COLORS_JSON: {\"row\":9}")))
|
||||||
|
#expect(exitCount(in: box) == 1)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test func unterminatedRowTailFinalizesAsJSONRow() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let bin = try script(
|
||||||
|
"row-tail.sh",
|
||||||
|
"#!/bin/sh\nprintf 'ROW_COLORS_JSON: {\"row\":42}'\n"
|
||||||
|
)
|
||||||
|
let box = Box()
|
||||||
|
let observer = observe(pm, id: "t12", into: box)
|
||||||
|
try await pm.runStreaming(id: "t12", binary: bin, arguments: [])
|
||||||
|
#expect(await waitForExit(in: box))
|
||||||
|
observer.cancel()
|
||||||
|
let events = box.events
|
||||||
|
let rows = events.compactMap { e -> String? in
|
||||||
|
if case .jsonRow(_, let d) = e { return String(decoding: d, as: UTF8.self) }
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
#expect(rows == ["{\"row\":42}"])
|
||||||
|
#expect(!events.contains(.stdout(id: "t12", line: "ROW_COLORS_JSON: {\"row\":42}")))
|
||||||
|
let rowIndex = events.firstIndex {
|
||||||
|
if case .jsonRow = $0 { return true }; return false
|
||||||
|
}
|
||||||
|
let exitIndexes = events.indices.filter {
|
||||||
|
if case .exit = events[$0] { return true }; return false
|
||||||
|
}
|
||||||
|
#expect(exitIndexes.count == 1)
|
||||||
|
if let rowIndex, let exitIndex = exitIndexes.first {
|
||||||
|
#expect(rowIndex < exitIndex)
|
||||||
|
} else {
|
||||||
|
Issue.record("expected a jsonRow before the exit event")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test func fastStreamingExitEmitsExactlyOneExit() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let bin = try script("fast-stream.sh", "#!/bin/sh\nexit 0\n")
|
||||||
|
let box = Box()
|
||||||
|
let observer = observe(pm, id: "t13", into: box)
|
||||||
|
try await pm.runStreaming(id: "t13", binary: bin, arguments: [])
|
||||||
|
#expect(await waitForExit(in: box))
|
||||||
|
// The grace window must outlast the 2 s finalize watchdog so a
|
||||||
|
// duplicate emission from it would be observed.
|
||||||
|
try await Task.sleep(for: .milliseconds(2500))
|
||||||
|
observer.cancel()
|
||||||
|
#expect(box.events == [.exit(id: "t13", code: 0)])
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test func fastCapturedExitEmitsExactlyOneExit() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let bin = try script("fast-cap.sh", "#!/bin/sh\nexit 7\n")
|
||||||
|
let box = Box()
|
||||||
|
let observer = observe(pm, id: "t14", into: box)
|
||||||
|
let result = try await pm.runCaptured(id: "t14", binary: bin, arguments: [])
|
||||||
|
#expect(result.exitCode == 7)
|
||||||
|
// Both the termination handler and the waitUntilExit watchdog
|
||||||
|
// resume the same box; give the slower path time to fire.
|
||||||
|
try await Task.sleep(for: .milliseconds(500))
|
||||||
|
observer.cancel()
|
||||||
|
#expect(box.events == [.exit(id: "t14", code: 7)])
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test func capturedRunSetsArgyllNotInteractive() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let bin = try script(
|
||||||
|
"cap-env.sh",
|
||||||
|
"#!/bin/sh\necho \"ANI=$ARGYLL_NOT_INTERACTIVE\"\n"
|
||||||
|
)
|
||||||
|
let result = try await pm.runCaptured(id: "t15", binary: bin, arguments: [])
|
||||||
|
#expect(result.stdout == "ANI=1\n")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test func killAllTerminatesStreamingAndCapturedChildren() async throws {
|
||||||
|
let pm = ProcessManager()
|
||||||
|
let marker = Self.fixtureDir
|
||||||
|
.appendingPathComponent("mixed-cap-ready-\(UUID().uuidString)")
|
||||||
|
let slowBin = try script("mixed-slow.sh", "#!/bin/sh\nsleep 30\n")
|
||||||
|
let capBin = try script("mixed-cap.sh", "#!/bin/sh\ntouch \"$1\"\nsleep 30\n")
|
||||||
|
let streamBox = Box()
|
||||||
|
let capBox = Box()
|
||||||
|
let streamObserver = observe(pm, id: "t16", into: streamBox)
|
||||||
|
let capObserver = observe(pm, id: "t17", into: capBox)
|
||||||
|
try await pm.runStreaming(id: "t16", binary: slowBin, arguments: [])
|
||||||
|
let capTask = Task {
|
||||||
|
try await pm.runCaptured(id: "t17", binary: capBin, arguments: [marker.path])
|
||||||
|
}
|
||||||
|
#expect(await waitForFile(marker))
|
||||||
|
#expect(await waitForRunning(pm, id: "t16"))
|
||||||
|
#expect(await waitForRunning(pm, id: "t17"))
|
||||||
|
#expect(await pm.killAll() == 2)
|
||||||
|
_ = try await capTask.value
|
||||||
|
#expect(await waitForExit(in: streamBox))
|
||||||
|
#expect(await waitForExit(in: capBox))
|
||||||
|
// Grace window outlasts the streaming finalize watchdog.
|
||||||
|
try await Task.sleep(for: .milliseconds(2500))
|
||||||
|
streamObserver.cancel()
|
||||||
|
capObserver.cancel()
|
||||||
|
#expect(!(await pm.isRunning("t16")))
|
||||||
|
#expect(!(await pm.isRunning("t17")))
|
||||||
|
#expect(exitCount(in: streamBox) == 1)
|
||||||
|
#expect(exitCount(in: capBox) == 1)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Suite("ProcessLineDecoder")
|
@Suite("ProcessLineDecoder")
|
||||||
|
|||||||
Reference in New Issue
Block a user