Skip to content

Coordinate Go oracle termination across process launch #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rohenaz merged 1 commit into from
Aug 31, 2026
Merged

Coordinate Go oracle termination across process launch #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion Tests/BSVConformanceTests/GoOracleProtocolTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,27 @@ struct GoOracleProtocolTests {
}
}

@Test("Concurrent stdout and stderr shutdown requests race normal exit safely")
func concurrentShutdownRace() throws {
for iteration in 0..<12 {
let script = try fakeOracle(metadata: validMetadata(), serve: .concurrentOutputExit)
let client = try requireClient(script)
#expect(throws: GoOracleClientError.self) {
try client.request(
id: "race-\(iteration)",
operation: "metadata",
arguments: [:]
)
}
client.close()

let marker = URL(fileURLWithPath: script.path + ".serve-count")
let children = try String(contentsOf: marker, encoding: .utf8)
.split(separator: "\n")
#expect(children.count == 1)
}
}

@Test("Timeout terminates the child within the two-second grace")
func timeout() throws {
let script = try fakeOracle(metadata: validMetadata(), serve: .hang)
Expand Down Expand Up @@ -285,7 +306,7 @@ struct GoOracleProtocolTests {

enum ServeBehavior: String, CaseIterable, Sendable {
case success, sameChild, operationError, unknownID, twoLines, invalidJSON, overlong, hang, noRead, exitSeven
case unknownResponseField, unknownErrorField
case unknownResponseField, unknownErrorField, concurrentOutputExit
}

private func validMetadata() -> GoOracleMetadata {
Expand Down Expand Up @@ -356,6 +377,8 @@ private func fakeOracle(
serve = "printf '%s\\n' '{\"schema\":\"bsv-conformance/1\",\"id\":\"one\",\"ok\":true,\"result\":{},\"unknown\":true}'"
case .unknownErrorField:
serve = "printf '%s\\n' '{\"schema\":\"bsv-conformance/1\",\"id\":\"one\",\"ok\":false,\"error\":{\"category\":\"internal\",\"message\":\"x\",\"unknown\":true}}'"
case .concurrentOutputExit:
serve = "printf '%s\\n' \"$$\" >> \"$0.serve-count\"; (/usr/bin/yes o | /usr/bin/head -c 1048577) & (/usr/bin/yes e | /usr/bin/head -c 1048577 >&2) & wait; exit 0"
}
let metadataCommand: String
if let metadataExitDiagnostic {
Expand Down
153 changes: 139 additions & 14 deletions Tests/BSVConformanceTests/Support/GoOracleClient.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,116 @@ import Glibc
let goOracleSchema = "bsv-conformance/1"
let goOracleMaximumLineBytes = 1 << 20

// Coordinate launch and every termination request through one child lifecycle.
private final class GoOracleProcessLifecycle: @unchecked Sendable {
private enum State {
case starting
case gracefulPending
case running(pid: Int32)
case gracefulRequested(pid: Int32)
case forceRequested(pid: Int32)
case exited
}

private let lock = NSLock()
private weak var process: Process?
private var state = State.starting

init(process: Process) {
self.process = process
}

func markStarted() {
lock.withLock {
let pending: Bool
switch state {
case .starting: pending = false
case .gracefulPending: pending = true
case .running, .gracefulRequested, .forceRequested, .exited: return
}
guard let process else {
state = .exited
return
}
let pid = process.processIdentifier
guard pid > 0,
process.isRunning else {
state = .exited
return
}
if pending {
state = .gracefulRequested(pid: pid)
process.terminate()
} else {
state = .running(pid: pid)
}
}
}

func requestGracefulTermination() {
lock.withLock {
switch state {
case .starting:
state = .gracefulPending
return
case .gracefulPending, .gracefulRequested, .forceRequested, .exited:
return
case .running(let pid):
guard let process,
process.processIdentifier == pid,
process.isRunning else {
state = .exited
return
}
state = .gracefulRequested(pid: pid)
process.terminate()
}
}
}

func requestForceTermination() {
lock.withLock {
let pid: Int32
switch state {
case .running(let value), .gracefulRequested(let value): pid = value
case .starting, .gracefulPending, .forceRequested, .exited: return
}
guard pid > 0,
let process,
process.processIdentifier == pid,
process.isRunning else {
state = .exited
return
}
state = .forceRequested(pid: pid)
_ = kill(pid, SIGKILL)
}
}

func markExited() {
lock.withLock { state = .exited }
}

var isRunning: Bool {
lock.withLock {
let pid: Int32
switch state {
case .running(let value), .gracefulRequested(let value), .forceRequested(let value):
pid = value
case .starting, .gracefulPending, .exited:
return false
}
guard let process,
process.processIdentifier == pid,
process.isRunning else {
state = .exited
return false
}
return true
}
}
}

enum GoOracleJSON: Codable, Equatable, Sendable {
case string(String)
case bool(Bool)
Expand Down Expand Up @@ -269,12 +379,14 @@ final class GoOracleClient: @unchecked Sendable {
private let responseReader: GoOracleLineReader
private let diagnostics: LockedData
private let terminated: DispatchSemaphore
private let lifecycle: GoOracleProcessLifecycle
let metadata: GoOracleMetadata

private init(
configuration: GoOracleConfiguration, metadata: GoOracleMetadata, process: Process,
stdin: FileHandle, stdout: FileHandle, stderr: FileHandle,
responseReader: GoOracleLineReader, diagnostics: LockedData, terminated: DispatchSemaphore
responseReader: GoOracleLineReader, diagnostics: LockedData, terminated: DispatchSemaphore,
lifecycle: GoOracleProcessLifecycle
) {
self.configuration = configuration
self.metadata = metadata
Expand All @@ -285,6 +397,7 @@ final class GoOracleClient: @unchecked Sendable {
self.responseReader = responseReader
self.diagnostics = diagnostics
self.terminated = terminated
self.lifecycle = lifecycle
}

static func connect(configuration: GoOracleConfiguration = .default()) throws -> GoOracleAvailability {
Expand Down Expand Up @@ -380,18 +493,20 @@ final class GoOracleClient: @unchecked Sendable {
let reader = GoOracleLineReader(maximumBytes: goOracleMaximumLineBytes)
let diagnostics = LockedData(maximumBytes: goOracleMaximumLineBytes)
let terminated = DispatchSemaphore(value: 0)
let lifecycle = GoOracleProcessLifecycle(process: process)
outputPipe.fileHandleForReading.readabilityHandler = { handle in
if !reader.append(handle.availableData) { process.terminate() }
if !reader.append(handle.availableData) { lifecycle.requestGracefulTermination() }
}
errorPipe.fileHandleForReading.readabilityHandler = { handle in
if !diagnostics.append(handle.availableData) {
reader.fail(with: .transport("oracle diagnostics exceeded 1 MiB"))
process.terminate()
lifecycle.requestGracefulTermination()
}
}
let outputHandle = outputPipe.fileHandleForReading
let errorHandle = errorPipe.fileHandleForReading
process.terminationHandler = { process in
lifecycle.markExited()
outputHandle.readabilityHandler = nil
errorHandle.readabilityHandler = nil
_ = reader.append(outputHandle.readDataToEndOfFile())
Expand All @@ -404,15 +519,17 @@ final class GoOracleClient: @unchecked Sendable {
}
do { try process.run() }
catch {
lifecycle.markExited()
outputPipe.fileHandleForReading.readabilityHandler = nil
errorPipe.fileHandleForReading.readabilityHandler = nil
throw GoOracleClientError.transport("could not start oracle serve process: \(error)")
}
lifecycle.markStarted()
return GoOracleClient(
configuration: configuration, metadata: metadata, process: process,
stdin: inputPipe.fileHandleForWriting, stdout: outputPipe.fileHandleForReading,
stderr: errorPipe.fileHandleForReading, responseReader: reader,
diagnostics: diagnostics, terminated: terminated
diagnostics: diagnostics, terminated: terminated, lifecycle: lifecycle
)
}

Expand All @@ -424,10 +541,10 @@ final class GoOracleClient: @unchecked Sendable {
}

private func stopProcess(terminateFirst: Bool) {
guard process.isRunning else { return }
if terminateFirst { process.terminate() }
if terminateFirst { lifecycle.requestGracefulTermination() }
guard lifecycle.isRunning else { return }
if terminated.wait(timeout: .now() + 2) == .timedOut {
_ = kill(process.processIdentifier, SIGKILL)
lifecycle.requestForceTermination()
_ = terminated.wait(timeout: .now() + 0.25)
}
}
Expand Down Expand Up @@ -485,9 +602,17 @@ final class GoOracleClient: @unchecked Sendable {
let outputFinished = DispatchSemaphore(value: 0)
let diagnosticsFinished = DispatchSemaphore(value: 0)
let completed = DispatchSemaphore(value: 0)
process.terminationHandler = { _ in completed.signal() }
let lifecycle = GoOracleProcessLifecycle(process: process)
process.terminationHandler = { _ in
lifecycle.markExited()
completed.signal()
}
do { try process.run() }
catch { throw GoOracleClientError.transport("could not start oracle: \(error)") }
catch {
lifecycle.markExited()
throw GoOracleClientError.transport("could not start oracle: \(error)")
}
lifecycle.markStarted()

func drain(
_ handle: FileHandle,
Expand All @@ -500,13 +625,13 @@ final class GoOracleClient: @unchecked Sendable {
do {
while let chunk = try handle.read(upToCount: 64 * 1_024), !chunk.isEmpty {
guard destination.append(chunk) else {
process.terminate()
lifecycle.requestGracefulTermination()
return
}
}
} catch {
reads.set(label: label, error: error)
process.terminate()
lifecycle.requestGracefulTermination()
}
}
}
Expand All @@ -524,13 +649,13 @@ final class GoOracleClient: @unchecked Sendable {
)
if let input {
do { try stdin.fileHandleForWriting.write(contentsOf: input); try stdin.fileHandleForWriting.close() }
catch { process.terminate(); throw GoOracleClientError.transport("could not write request: \(error)") }
catch { lifecycle.requestGracefulTermination(); throw GoOracleClientError.transport("could not write request: \(error)") }
}

if completed.wait(timeout: .now() + configuration.startupDeadline) == .timedOut {
process.terminate()
lifecycle.requestGracefulTermination()
if completed.wait(timeout: .now() + 2) == .timedOut {
_ = kill(process.processIdentifier, SIGKILL)
lifecycle.requestForceTermination()
_ = completed.wait(timeout: .now() + 0.25)
}
_ = outputFinished.wait(timeout: .now() + 2)
Expand Down
Loading