fix(runtime): allow initial balancer startup

Initial multi-server checks can consume the entire five-second listener
budget before sslocal binds its inbound ports. Allow 15 seconds for
listeners and 30 seconds for GUI startup health confirmation.

Include endpoint readiness diagnostics and a delayed-start regression.
This commit is contained in:
Qiu Yuzhou
2026-10-30 08:52:36 +08:00
parent 8d8fe2209d
commit 3e64743237
6 changed files with 87 additions and 30 deletions
+17 -13
View File
@@ -2,31 +2,35 @@ import Foundation
// MARK: - 监听建立判定
/// 监听判定时限:sslocal 拉起后 5 秒内应完成本地监听绑定。
let listenEstablishmentDeadline: TimeInterval = 5
/// 覆盖插件启动、上游初始测速(默认超时 5 秒)与调度余量;GUI 健康窗另留拉起时间。
let listenEstablishmentDeadline: TimeInterval = 15
/// 单次探测超时与轮询间隔;正常启动百毫秒级完成,超时路径总耗时不超时限太多。
let listenProbeTimeout: TimeInterval = 0.25
let listenProbeInterval: TimeInterval = 0.1
func listenersAreEstablished(
/// 返回首个未通过的检查;nil 表示所有入站已就绪。轮询只在期限到达时输出失败。
func listenerReadinessFailure(
document: RuntimeSupervision,
child: Process
) -> Bool {
guard child.isRunning else { return false }
) -> String? {
guard child.isRunning else { return "check=process outcome=exited" }
for local in document.locals {
guard
EndpointHealthProbe.probe(
host: local.probeHost,
port: local.localPort,
timeout: listenProbeTimeout) == .reachable
else { return false }
let endpoint = "endpoint=\(local.inboundProtocol) \(local.localAddress):\(local.localPort)"
let connection = "\(endpoint) probeHost=\(local.probeHost) check=connect"
switch EndpointHealthProbe.probe(
host: local.probeHost, port: local.localPort, timeout: listenProbeTimeout)
{
case .reachable: break
case .refused(let detail): return "\(connection) outcome=refused detail=\(detail)"
case .timedOut: return "\(connection) outcome=timedOut"
}
if document.acl != nil,
!RuntimeSocketOwnershipProbe.process(
child.processIdentifier, ownsTCPListenerOn: local.localPort)
{
return false
return "\(endpoint) check=ownership outcome=unconfirmed"
}
}
return child.isRunning
return child.isRunning ? nil : "check=process outcome=exited"
}
+5 -4
View File
@@ -134,7 +134,8 @@ private func establishChildReadiness(
_ child: Process, state: inout SupervisionState
) -> SupervisionOutcome? {
guard !state.healthy else { return nil }
if listenersAreEstablished(document: state.current, child: child) {
let readinessFailure = listenerReadinessFailure(document: state.current, child: child)
guard let readinessFailure else {
guard sendReloadSignals(&state.reloadPending, to: child.processIdentifier) else {
recordDeploymentFailure(state.current, reason: "Deferred reload failed")
_ = stopChild(child)
@@ -144,11 +145,11 @@ private func establishChildReadiness(
return nil
}
guard Date() >= state.readinessDeadline else { return nil }
let elapsed = listenEstablishmentDeadline - state.readinessDeadline.timeIntervalSinceNow
let timing = String(format: "elapsed=%.3fs deadline=%.3fs", elapsed, listenEstablishmentDeadline)
RuntimeLog.emit(
.listenNotEstablished(
detail: state.current.locals.map {
"\($0.inboundProtocol) \($0.localAddress):\($0.localPort)"
}.joined(separator: ", ")))
detail: "pid=\(child.processIdentifier) \(timing) \(readinessFailure)"))
recordDeploymentFailure(state.current, reason: "Listener establishment failed")
_ = stopChild(child)
return .stopped
@@ -147,7 +147,7 @@ final class ProxyRuntimeController {
firewallChecker: FirewallStatusChecking = SocketFilterFirewallChecker(),
firewallExecutableURLs: [URL]? = nil,
firewallPollIntervalNanoseconds: UInt64 = 2_000_000_000,
launchHealthTimeoutSeconds: TimeInterval = 15,
launchHealthTimeoutSeconds: TimeInterval = 30,
launchHealthRetryDelay: @escaping () async throws -> Void = {
try await Task.sleep(for: .milliseconds(200))
},
@@ -121,7 +121,7 @@ class ProxyRuntimeControllerTests: XCTestCase {
firewallChecker: FirewallStatusChecking = ProxyRuntimeFixture.FakeFirewallChecker(),
firewallExecutableURLs: [URL] = [URL(fileURLWithPath: "/bundle/Helpers/sslocal")],
firewallPollIntervalNanoseconds: UInt64 = 1_000_000,
// 生产默认 15 秒健康窗;FakeProbe 结果确定,短窗口走同一超时呈现路径。
// 生产默认 30 秒健康窗;FakeProbe 结果确定,短窗口走同一超时呈现路径。
launchHealthTimeoutSeconds: TimeInterval = 0.05,
launchHealthRetryDelay: @escaping () async throws -> Void = {
try await Task.sleep(for: .milliseconds(200))
@@ -117,6 +117,11 @@ final class AgentLifecycleStaleListenerTests: AgentLifecycleTests {
XCTAssertNil(
runtimeStore.readRuntimeReceipt(),
"候选没有绑定端口时,旧监听和 server-only reload 都不能发布直连回执")
XCTAssertTrue(
try waitUntil(timeout: 20) { self.agentLog().contains("check=ownership") },
"超时日志应区分端口可连接但不属于当前子进程:\(agentLog())")
XCTAssertEqual(try waitForExit(wrapper), 0)
XCTAssertTrue(agentLog().contains("endpoint=socks 127.0.0.1:\(staleSocksListener.port)"))
}
}
@@ -124,20 +129,29 @@ final class AgentLifecycleListenDeadlineTests: AgentLifecycleTests {
// MARK: 监听建立判定
func testListenDeadlinePersistsFailureAndStopsChild() throws {
// stub sslocal 不监听任何端点:超时后应记 error 日志点名端点,且监管继续。
// stub sslocal 不监听任何端点:超时日志应包含失败检查与等待耗时。
try writeContract(ProxyRuntimeFixture.makeDocument())
let wrapper = try launchWrapper(behavior: "run")
let agentLogURL = workDir.appendingPathComponent("agent.log")
XCTAssertTrue(
try waitUntil {
try waitUntil(timeout: 20) {
guard let log = try? String(contentsOf: agentLogURL, encoding: .utf8) else {
return false
}
return log.contains("listen not established within deadline")
},
"5 秒未建立监听应以 error 日志点名(D10),实际 agent.log:\(agentLog())")
"15 秒未建立监听应以 error 日志点名,实际 agent.log:\(agentLog())")
XCTAssertEqual(try waitForExit(wrapper), 0)
let log = try XCTUnwrap(
agentLog().components(separatedBy: "\n").first {
$0.contains("listen not established within deadline")
})
XCTAssertTrue(log.contains("endpoint=socks 127.0.0.1:"), log)
XCTAssertTrue(log.contains("check=connect outcome=refused"), log)
XCTAssertTrue(log.contains("pid="), log)
XCTAssertTrue(log.contains("elapsed="), log)
XCTAssertTrue(log.contains("deadline=15.000s"), log)
XCTAssertNotNil(RuntimeFileStore(fileURL: contractURL).readFailure())
XCTAssertTrue(stateLog().contains("SIGTERM"))
}
@@ -145,7 +159,7 @@ final class AgentLifecycleListenDeadlineTests: AgentLifecycleTests {
private final class TCPListenerFixture {
let port: Int
private var descriptor: Int32
private let source: DispatchSourceRead
init() throws {
let socketDescriptor = socket(AF_INET, SOCK_STREAM, 0)
@@ -179,13 +193,25 @@ private final class TCPListenerFixture {
throw POSIXError(.EINVAL)
}
descriptor = socketDescriptor
guard fcntl(socketDescriptor, F_SETFL, O_NONBLOCK) == 0 else {
close(socketDescriptor)
throw POSIXError(.EINVAL)
}
port = Int(CFSwapInt16BigToHost(boundAddress.sin_port))
source = DispatchSource.makeReadSource(fileDescriptor: socketDescriptor, queue: .global())
source.setEventHandler {
while true {
let connection = accept(socketDescriptor, nil, nil)
guard connection >= 0 else { return }
close(connection)
}
}
source.setCancelHandler { close(socketDescriptor) }
source.resume()
}
func closeListener() {
if descriptor >= 0 { close(descriptor) }
descriptor = -1
source.cancel()
}
deinit {
@@ -78,7 +78,9 @@ class RealSslocalSmokeTests: XCTestCase {
}
/// 写入给定契约并拉起 wrapper(契约内容由调用方构造:无插件与插件场景各异)。
func launchWrapper(_ document: SslocalRuntimeDocument, publish: Bool = true) throws -> Process {
func launchWrapper(
_ document: SslocalRuntimeDocument, publish: Bool = true, startupDelay: TimeInterval = 0
) throws -> Process {
if publish { try RuntimeFileStore(fileURL: contractURL).write(document) }
let wrapperURL = AppArtifact.bundleURL.appendingPathComponent(
@@ -87,11 +89,13 @@ class RealSslocalSmokeTests: XCTestCase {
process.executableURL = wrapperURL
var environment = ProcessInfo.processInfo.environment
environment["SSXNG_CONTRACT_PATH"] = contractURL.path
if usesLocalDNS {
// exec preserves the PID supervised by the real wrapper. Only DNS is test-owned.
if usesLocalDNS || startupDelay > 0 {
// exec preserves the PID supervised by the real wrapper, including delayed startup.
let launcher = workDir.appendingPathComponent("sslocal-with-local-dns")
let executable = try sslocalURL.path.replacingOccurrences(of: "'", with: "'\"'\"'")
let script = "#!/bin/sh\nexec '\(executable)' --dns udp://127.0.0.1:\(dns.port) \"$@\"\n"
let dnsArguments = usesLocalDNS ? " --dns udp://127.0.0.1:\(dns.port)" : ""
let delay = startupDelay > 0 ? "sleep \(startupDelay)\n" : ""
let script = "#!/bin/sh\n\(delay)exec '\(executable)'\(dnsArguments) \"$@\"\n"
try script.write(to: launcher, atomically: true, encoding: .utf8)
try FileManager.default.setAttributes([.posixPermissions: 0o700], ofItemAtPath: launcher.path)
environment["SSXNG_SSLOCAL_PATH"] = launcher.path
@@ -192,6 +196,28 @@ class RealSslocalSmokeTests: XCTestCase {
}
}
final class RealSslocalDelayedStartupTests: RealSslocalSmokeTests {
func testListenersBecomingReadyAfterFiveSecondsRemainRunning() throws {
let ports = try grabThreeListenPorts(excluding: [])
let document = SslocalRuntimeDocument(
servers: [], listen: ProxyListeningConfiguration(socksPort: ports[0], httpPort: ports[1]),
acl: .direct(at: aclFileURL))
let store = RuntimeFileStore(fileURL: contractURL)
let started = Date()
let wrapper = try launchWrapper(document, startupDelay: 6)
XCTAssertTrue(
try waitForCondition(timeout: 12) {
store.readRuntimeReceipt() != nil || store.readFailure() != nil
})
XCTAssertNil(store.readFailure(), "延迟启动不能被旧的 5 秒期限终止")
_ = try XCTUnwrap(store.readRuntimeReceipt())
XCTAssertGreaterThanOrEqual(Date().timeIntervalSince(started), 6)
try assertReceiptOwnsListeners(document: document)
XCTAssertTrue(wrapper.isRunning)
stopWrapperAndAssertCleanExit(wrapper, description: "delayed startup wrapper exits")
}
}
final class RealSslocalListenerSmokeTests: RealSslocalSmokeTests {
func testRealSslocalBindsSOCKSAndHTTPPortsAndCompletesHandshake() throws {
var ports = Set<Int>()