diff --git a/ShadowsocksX-NG2/Agent/main.swift b/ShadowsocksX-NG2/Agent/main.swift index 5fe2031d..6c0749d5 100644 --- a/ShadowsocksX-NG2/Agent/main.swift +++ b/ShadowsocksX-NG2/Agent/main.swift @@ -87,7 +87,7 @@ private enum SupervisionOutcome { case stopped, restart } private struct SupervisionState { var current: RuntimeSupervision var processFiles: Set - var reloadPending = Set() + var reloadPending = RuntimeReloadState() var healthy = false var receiptPublished = false var readinessDeadline: Date @@ -201,7 +201,7 @@ private func applyInPlaceReload( _ next: RuntimeSupervision, child: Process, state: inout SupervisionState ) -> SupervisionOutcome? { let needsDeferral = next.acl != nil && !state.healthy - state.reloadPending.formUnion(next.reloadSignals(from: state.current)) + state.reloadPending.enqueue(next, from: state.current) guard selectDeployment(next), needsDeferral || sendReloadSignals(&state.reloadPending, to: child.processIdentifier) else { @@ -217,10 +217,8 @@ private func applyInPlaceReload( return nil } -private func sendReloadSignals(_ signals: inout Set, to processID: Int32) -> Bool { - guard signals.sorted().allSatisfy({ kill(processID, $0) == 0 }) else { return false } - signals.removeAll() - return true +private func sendReloadSignals(_ signals: inout RuntimeReloadState, to processID: Int32) -> Bool { + signals.flush { kill(processID, $0) == 0 } } /// 停止超时不得进入启动分支。保留 pid 文件供下次 Agent 拒绝重叠启动。 diff --git a/ShadowsocksX-NG2/Domain/Runtime/RuntimeReloadState.swift b/ShadowsocksX-NG2/Domain/Runtime/RuntimeReloadState.swift new file mode 100644 index 00000000..7820c66f --- /dev/null +++ b/ShadowsocksX-NG2/Domain/Runtime/RuntimeReloadState.swift @@ -0,0 +1,16 @@ +import Foundation + +/// 启动期间累积重载信号,只有全部投递成功后才清空。 +struct RuntimeReloadState { + private var pending = Set() + + mutating func enqueue(_ next: RuntimeSupervision, from current: RuntimeSupervision) { + pending.formUnion(next.reloadSignals(from: current)) + } + + mutating func flush(send: (Int32) -> Bool) -> Bool { + guard pending.sorted().allSatisfy(send) else { return false } + pending.removeAll() + return true + } +} diff --git a/ShadowsocksX-NG2/Tests/Agent/AgentLifecycleTests+ChangeProtocol.swift b/ShadowsocksX-NG2/Tests/Agent/AgentLifecycleTests+ChangeProtocol.swift index da4f1013..41228c73 100644 --- a/ShadowsocksX-NG2/Tests/Agent/AgentLifecycleTests+ChangeProtocol.swift +++ b/ShadowsocksX-NG2/Tests/Agent/AgentLifecycleTests+ChangeProtocol.swift @@ -51,117 +51,6 @@ final class AgentLifecycleChangeProtocolTests: AgentLifecycleTests { "两次停止各转发一次 SIGTERM") } - private func installGatedListeningStub() throws { - // 同一 PID 在显式放行后才建立监听,避免通过延时猜测启动阶段。 - let script = #""" - #!/usr/bin/python3 - import json, os, pathlib, select, signal, socket, sys, time - root = pathlib.Path(os.environ["SSXNG_RUNTIME_DIR"]) - state = pathlib.Path(os.environ["SSLOCAL_STUB_STATE"]) - def record(value): - with state.open("a") as output: - output.write(value + "\n") - def stopped(*args): - record("SIGTERM") - sys.exit(0) - signal.signal(signal.SIGTERM, stopped) - signal.signal(signal.SIGUSR1, lambda *args: record("SIGUSR1")) - signal.signal(signal.SIGUSR2, lambda *args: record("SIGUSR2")) - (pathlib.Path(os.environ["SSLOCAL_STUB_PID_DIR"]) / str(os.getpid())).touch() - record("invoked: gated listener") - with open(sys.argv[2]) as source: - configuration = json.load(source) - while not (root / "release-listeners").exists(): - time.sleep(0.01) - listeners = [] - for local in configuration["locals"]: - listener = socket.socket() - listener.bind((local["local_address"], local["local_port"])) - listener.listen(128) - listeners.append(listener) - probe_count = 0 - while True: - ready = select.select(listeners, [], [], 0.01)[0] - while ready: - for listener in ready: - connection, _ = listener.accept() - connection.close() - probe_count += 1 - ready = select.select(listeners, [], [], 0)[0] - request = root / "probe-count-request" - if request.exists(): - token = request.read_text() - request.unlink() - response = root / ("probe-count-" + token) - temporary = response.with_suffix(".tmp") - temporary.write_text(str(probe_count)) - temporary.replace(response) - """# - try script.write( - to: workDir.appendingPathComponent("stub-sslocal.sh"), atomically: false, encoding: .utf8) - } - - private func probeCount(_ token: Int) throws -> Int { - let response = workDir.appendingPathComponent("probe-count-\(token)") - try AtomicFileWriter.write( - Data(String(token).utf8), to: workDir.appendingPathComponent("probe-count-request")) - XCTAssertTrue(try waitUntil { FileManager.default.fileExists(atPath: response.path) }) - return try XCTUnwrap(Int(String(contentsOf: response, encoding: .utf8))) - } - - func testDeferredACLSignalSurvivesSubsequentServerOnlyPublication() throws { - let runtimeStore = RuntimeFileStore(fileURL: contractURL) - let socks = try TCPListenerFixture() - let http = try TCPListenerFixture() - let listen = ProxyListeningConfiguration(socksPort: socks.port, httpPort: http.port) - socks.closeListener() - http.closeListener() - let previous = SslocalRuntimeDocument( - servers: [], listen: listen, acl: .global(at: runtimeStore.aclFileURL)) - try writeContract(previous) - try installGatedListeningStub() - let wrapper = try launchWrapper(behavior: "run") - XCTAssertTrue(try waitUntil { self.stateLog().contains("invoked:") }) - let next = previous.replacingACL(.direct(at: runtimeStore.aclFileURL)) - try writeContract(next) - XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0) - let deferred = "reload deferred until listeners are ready" - XCTAssertTrue(try waitUntil { self.agentLog().components(separatedBy: deferred).count == 2 }) - let final = SslocalRuntimeDocument( - servers: ProxyRuntimeFixture.makeDocument().servers, listen: listen, acl: next.aclRuntime) - try writeContract(final) - XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0) - XCTAssertTrue(try waitUntil { self.agentLog().components(separatedBy: deferred).count == 3 }) - XCTAssertFalse(stateLog().contains("SIGUSR1")) - XCTAssertFalse(stateLog().contains("SIGUSR2")) - try Data().write(to: workDir.appendingPathComponent("release-listeners")) - XCTAssertTrue( - try waitUntil { - self.stateLog().contains("SIGUSR1") && self.stateLog().contains("SIGUSR2") - && runtimeStore.readRuntimeReceipt()?.deploymentID - == runtimeStore.receiptIdentity(for: final) - }) - let probesBeforeReload = try probeCount(1) - let aclOnly = final.replacingACL(.global(at: runtimeStore.aclFileURL)) - try writeContract(aclOnly) - XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0) - XCTAssertTrue( - try waitUntil { - self.stateLog().components(separatedBy: "SIGUSR2").count == 3 - && runtimeStore.readRuntimeReceipt()?.deploymentID - == runtimeStore.receiptIdentity(for: aclOnly) - }) - XCTAssertEqual( - try probeCount(2), probesBeforeReload, "健康实例热重载前后均不得重新连接监听端口") - XCTAssertEqual( - stateLog().components(separatedBy: "SIGUSR1").count - 1, 1, - "仅 ACL 变化不得重新加载服务器配置") - XCTAssertEqual(stateLog().components(separatedBy: "invoked:").count - 1, 1) - XCTAssertFalse(stateLog().contains("SIGTERM")) - kill(wrapper.processIdentifier, SIGTERM) - XCTAssertEqual(try waitForExit(wrapper), 0) - } - func testACLReloadIsDeferredWithoutRestartUntilListenersAreReady() throws { let runtimeStore = RuntimeFileStore(fileURL: contractURL) let previous = ProxyRuntimeFixture.makeDocument() diff --git a/ShadowsocksX-NG2/Tests/Runtime/RuntimeReloadStateTests.swift b/ShadowsocksX-NG2/Tests/Runtime/RuntimeReloadStateTests.swift new file mode 100644 index 00000000..87c6df8c --- /dev/null +++ b/ShadowsocksX-NG2/Tests/Runtime/RuntimeReloadStateTests.swift @@ -0,0 +1,52 @@ +import Darwin +import Foundation +import Testing + +@testable import ShadowsocksX_NG2 + +struct RuntimeReloadStateTests { + @Test func serverPublicationPreservesDeferredACLReloadAndFlushClearsSignals() throws { + let runtime = ProxyRuntimeFixture.makeTemporaryRuntime() + defer { try? FileManager.default.removeItem(at: runtime.directory) } + let store = RuntimeFileStore(fileURL: runtime.contract) + let initial = ProxyRuntimeFixture.makeDocument(localPort: 1086).replacingACL( + .global(at: store.aclFileURL)) + try store.write(initial) + let previous = try #require(RuntimePublication.load(at: store.supervisionFileURL)) + let direct = initial.replacingACL(.direct(at: store.aclFileURL)) + try store.write(direct) + let aclOnly = try #require(RuntimePublication.load(at: store.supervisionFileURL)) + try store.write( + ProxyRuntimeFixture.makeDocument(serverAddress: "198.51.100.9", localPort: 1086) + .replacingACL(direct.aclRuntime)) + let serverOnly = try #require(RuntimePublication.load(at: store.supervisionFileURL)) + + var state = RuntimeReloadState() + state.enqueue(aclOnly, from: previous) + state.enqueue(serverOnly, from: aclOnly) + state.enqueue(serverOnly, from: serverOnly) + var delivered: [Int32] = [] + #expect( + state.flush { + delivered.append($0) + return true + }) + #expect(delivered == [SIGUSR1, SIGUSR2].sorted()) + delivered.removeAll() + #expect( + state.flush { + delivered.append($0) + return true + }) + #expect(delivered.isEmpty) + + state.enqueue(aclOnly, from: serverOnly) + #expect(!state.flush { _ in false }) + #expect( + state.flush { + delivered.append($0) + return true + }) + #expect(delivered == [SIGUSR1]) + } +} diff --git a/ShadowsocksX-NG2/project.yml b/ShadowsocksX-NG2/project.yml index c446dcd5..990f7882 100644 --- a/ShadowsocksX-NG2/project.yml +++ b/ShadowsocksX-NG2/project.yml @@ -220,6 +220,7 @@ targets: - path: Domain/Runtime/RuntimePaths.swift - path: Domain/Runtime/AtomicFileWriter.swift - path: Domain/Runtime/RuntimeSupervision.swift + - path: Domain/Runtime/RuntimeReloadState.swift - path: Domain/Runtime/RuntimeLog.swift - path: Domain/Runtime/SystemLog.swift - path: Domain/Runtime/LogLine.swift