test(agent): replace timed reload integration test

ShadowsocksX-NG2/Agent/main.swift

@@ -87,7 +87,7 @@ private enum SupervisionOutcome { case stopped, restart }
 private struct SupervisionState {
   var current: RuntimeSupervision
   var processFiles: Set<String>
-  var reloadPending = Set<Int32>()
+  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<Int32>, 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 拒绝重叠启动。

ShadowsocksX-NG2/Domain/Runtime/RuntimeReloadState.swift

@@ -0,0 +1,16 @@
+import Foundation
+
+/// 启动期间累积重载信号，只有全部投递成功后才清空。
+struct RuntimeReloadState {
+  private var pending = Set<Int32>()
+
+  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
+  }
+}

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()

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])
+  }
+}

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