feat(runtime): hot reload ACL rules with signals

This commit is contained in:
Qiu Yuzhou
2026-10-20 23:54:05 +08:00
parent 8508080480
commit 2c75ee67ec
17 changed files with 489 additions and 60 deletions
+19 -14
View File
@@ -89,9 +89,9 @@ private func superviseChild(
flags: SignalFlags
) -> SupervisionOutcome {
var current = document
// SIGUSR1 没有应用确认,旧进程曾用过的引用保守保留到该进程退出。
// 重载信号没有应用确认,旧进程曾用过的引用保守保留到该进程退出。
var processFiles = document.retainedFiles
var reloadPending = false
var reloadPending = Set<Int32>()
var healthy = false
var receiptPublished = false
var readinessDeadline = Date().addingTimeInterval(listenEstablishmentDeadline)
@@ -113,13 +113,10 @@ private func superviseChild(
}
if !healthy {
if listenersAreEstablished(document: current, child: child) {
if reloadPending {
guard kill(child.processIdentifier, SIGUSR1) == 0 else {
recordDeploymentFailure(current, reason: "Deferred reload failed")
_ = stopChild(child)
return .stopped
}
reloadPending = false
guard sendReloadSignals(&reloadPending, to: child.processIdentifier) else {
recordDeploymentFailure(current, reason: "Deferred reload failed")
_ = stopChild(child)
return .stopped
}
healthy = true
} else if Date() >= readinessDeadline {
@@ -165,25 +162,33 @@ private func superviseChild(
}
return .restart
}
let needsDeferral = next.acl != nil && !listenersAreEstablished(document: next, child: child)
guard selectDeployment(next), needsDeferral || kill(child.processIdentifier, SIGUSR1) == 0
// 无需重启的更新沿用本子进程已建立的监听事实。
let needsDeferral = next.acl != nil && !healthy
reloadPending.formUnion(next.reloadSignals(from: current))
guard selectDeployment(next),
needsDeferral
|| sendReloadSignals(&reloadPending, to: child.processIdentifier)
else {
recordDeploymentFailure(next, reason: "Server reload failed")
recordDeploymentFailure(next, reason: "Reload failed")
_ = stopChild(child)
return .stopped
}
RuntimeLog.emit(needsDeferral ? .reloadDeferred : .reloadForwarded)
reloadPending = needsDeferral
processFiles.formUnion(next.retainedFiles)
current = next
receiptPublished = false
healthy = false
readinessDeadline = Date().addingTimeInterval(listenEstablishmentDeadline)
}
_ = flags.wait(timeout: .now() + listenProbeInterval)
}
}
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
}
/// 停止超时不得进入启动分支。保留 pid 文件供下次 Agent 拒绝重叠启动。
private func stopChild(_ child: Process) -> Bool {
RuntimeLog.emit(.sslocalStopRequested)
@@ -54,6 +54,7 @@ extension ProxyRuntimeController {
private func convergencePlanIsEmpty(
_ document: SslocalRuntimeDocument, preparedContract: PreparedRuntimeContract?
) -> Bool {
guard runtimeReloadProcesses.isEmpty else { return false }
switch state {
case .running, .firewallBlocked:
break
@@ -197,11 +198,12 @@ extension ProxyRuntimeController {
return .unchanged
}
if policy == .modeTransition, unchanged,
runtimeFileStore.readFailure() == nil, alreadyRunning
runtimeFileStore.readFailure() == nil, alreadyRunning,
aclReloadProcessID(for: next) != nil
{
guard settings.systemProxyEnabled else { return .unchanged }
state = .starting
return await presentLaunchHealth(next) ? .applied : .failed
await systemProxyObserver.convergeSystemProxy()
return .applied
}
if policy == .savedRules {
modeChangeGeneration += 1
@@ -217,6 +219,20 @@ extension ProxyRuntimeController {
case savedRules
}
/// 仅 ACL 变化且原有子进程仍健康时,继续沿用监听事实。
private func aclReloadProcessID(for document: SslocalRuntimeDocument) -> Int32? {
switch state {
case .running, .firewallBlocked: break
default: return nil
}
guard agent.status == .registered, case .running = wrapperState(),
let identity = document.replacingACL(nil).intentSHA256,
runtimeObservation?.configurationIntentSHA256 == identity
else { return nil }
return effectiveRuntimeListenerProcessID
?? runtimeReloadProcesses.values.first(where: { processIsAlive($0) })
}
/// 共用执行脊柱;票据不越过意图 seam。
private func apply(
_ document: SslocalRuntimeDocument,
@@ -235,8 +251,12 @@ extension ProxyRuntimeController {
return .unchanged
}
guard convergenceIsCurrent(ticket) else { return .superseded }
let reloadingProcessID = aclReloadProcessID(for: document)
let reloadID = UUID()
if let reloadingProcessID { runtimeReloadProcesses[reloadID] = reloadingProcessID }
defer { runtimeReloadProcesses.removeValue(forKey: reloadID) }
runtimeObservation = RuntimeDeploymentFacts(document)
state = .starting
if reloadingProcessID == nil { state = .starting }
let execution = await execute(.run(document), document: document, preparedContract: contract)
ticket.flow = execution.flow
guard convergenceIsCurrent(ticket) else { return .superseded }
@@ -250,7 +270,8 @@ extension ProxyRuntimeController {
document,
requiresReceipt: policy != .deployment || document.aclRuntime != nil,
convergeProxyOnSuccess: policy == .deployment,
preserveProxyOnFailure: false, preparedContract: contract)
preserveProxyOnFailure: false, preparedContract: contract,
reloadingProcessID: reloadingProcessID)
guard convergenceIsCurrent(ticket) else { return .superseded }
guard healthy else {
await failDeployment(ticket: &ticket)
@@ -29,7 +29,8 @@ extension ProxyRuntimeController {
requiresReceipt: Bool = false,
convergeProxyOnSuccess: Bool = true,
preserveProxyOnFailure: Bool = false,
preparedContract: PreparedRuntimeContract? = nil
preparedContract: PreparedRuntimeContract? = nil,
reloadingProcessID: Int32? = nil
) async -> Bool {
guard preparedContract == nil || preparedContract?.document == document else {
return await reportRuntimeFileFailure(preserveProxyOnFailure: preserveProxyOnFailure)
@@ -49,7 +50,8 @@ extension ProxyRuntimeController {
guard generation == flowGeneration, preparation == runtimePreparationGeneration else {
return false
}
let attempt = await launchHealthAttempt(facts, expectedDigest: expectedDigest)
let attempt = await launchHealthAttempt(
facts, expectedDigest: expectedDigest, reloadingProcessID: reloadingProcessID)
guard generation == flowGeneration, preparation == runtimePreparationGeneration else {
return false
}
@@ -78,7 +80,7 @@ extension ProxyRuntimeController {
private func launchHealthAttempt(
_ document: RuntimeDeploymentFacts,
expectedDigest: String?
expectedDigest: String?, reloadingProcessID: Int32? = nil
) async -> LaunchHealthAttempt {
if runtimeFileStore.blocksDeployment(document.intentSHA256) { return .agentLost }
var receiptForHealth: RuntimeDeploymentReceipt?
@@ -94,8 +96,11 @@ extension ProxyRuntimeController {
receiptForHealth = receipt
}
if let endpointFailure = await unhealthyLocalEndpoint(in: document) {
return .retry(endpointFailure)
if reloadingProcessID == nil || receiptForHealth?.processID != reloadingProcessID {
if reloadingProcessID != nil { state = .starting }
if let endpointFailure = await unhealthyLocalEndpoint(in: document) {
return .retry(endpointFailure)
}
}
if let receiptForHealth {
@@ -224,7 +229,7 @@ extension ProxyRuntimeController {
else { return }
systemProxyObserver.refreshSystemProxyApproval()
systemProxyObserver.updateSystemProxyActions()
guard settings.systemProxyEnabled, state != .starting,
guard settings.systemProxyEnabled, state != .starting, runtimeReloadProcesses.isEmpty,
!systemProxyObserver.systemProxyStartupInProgress, settings.agentEnabled,
let document = runtimeObservation
else { continue }
@@ -108,6 +108,7 @@ final class ProxyRuntimeController {
var flowGeneration = 0
var modeChangeGeneration = 0
var runtimeObservation: RuntimeDeploymentFacts?
var runtimeReloadProcesses: [UUID: Int32] = [:]
var firewallObservationTask: Task<Void, Never>?
var proxyMode: ProxyMode { didSet { factChanges.markChanged() } }
@@ -71,6 +71,16 @@ struct RuntimeFileStore {
contract.document, old: old, create: { try create($0, kind: $1, suffix: $2) })
let encoded = authentication.document
let configurationData = try encoded.jsonData()
let configurationID =
old.flatMap { previous -> String? in
guard let identity = previous.configurationID,
publishedIntent(previous)?.configurationSHA256
== ProxyACLDocument.digest(configurationData),
(try? Data(contentsOf: directoryURL.appendingPathComponent(previous.configuration)))
== configurationData
else { return nil }
return identity
} ?? UUID().uuidString
let configuration = try create(configurationData, kind: "config", suffix: "sslocal.json")
let intent = PublishedRuntimeIntent(
sha256: contract.sha256, configurationSHA256: ProxyACLDocument.digest(configurationData),
@@ -79,7 +89,8 @@ struct RuntimeFileStore {
JSONEncoder().encode(intent), kind: "config", suffix: "intent.json")
let snapshot = RuntimeSupervision(
version: 2, generation: generation, deploymentID: UUID().uuidString,
configuration: configuration, intentMetadata: intentMetadata,
configuration: configuration, configurationID: configurationID,
intentMetadata: intentMetadata,
locals: encoded.locals.map {
.init(
inboundProtocol: $0.inboundProtocol, localAddress: $0.localAddress,
@@ -28,6 +28,8 @@ struct RuntimeSupervision: Codable, Equatable, Sendable {
let generation: UInt64
let deploymentID: String
let configuration: String
/// 配置字节未变时复用的随机身份;旧监管资料缺失时保守重载配置。
let configurationID: String?
/// GUI 私有意图摘要的同代文件;Agent 只保留引用,不读取正文。
let intentMetadata: String
let locals: [Endpoint]
@@ -44,7 +46,16 @@ struct RuntimeSupervision: Codable, Equatable, Sendable {
func requiresRestart(from old: Self) -> Bool {
locals != old.locals || ipv6Only != old.ipv6Only
|| authenticationID != old.authenticationID || acl != old.acl
|| authenticationID != old.authenticationID
}
func reloadSignals(from old: Self) -> Set<Int32> {
var signals = Set<Int32>()
if configurationID == nil || configurationID != old.configurationID {
signals.insert(SIGUSR1)
}
if acl != old.acl { signals.insert(SIGUSR2) }
return signals
}
var hasValidStructure: Bool {
@@ -53,6 +64,7 @@ struct RuntimeSupervision: Codable, Equatable, Sendable {
RuntimePublication.generation(of: configuration) == generation,
RuntimePublication.generation(of: intentMetadata) == generation,
configuration != intentMetadata,
configurationID.map({ UUID(uuidString: $0) != nil }) ?? true,
authenticationFiles.allSatisfy({ $0.hasPrefix("auth-") && $0.hasSuffix(".json") }),
aclVariants.values.allSatisfy({ $0.file.hasPrefix("acl-") && $0.file.hasSuffix(".ini") }),
authenticationID.map({ UUID(uuidString: $0) != nil }) ?? true,
@@ -317,6 +317,7 @@ struct RuntimeDeploymentFacts: Equatable, Sendable {
let hasServers: Bool
let aclSummary: String?
let intentSHA256: String?
let configurationIntentSHA256: String?
init(_ document: SslocalRuntimeDocument) {
locals = document.locals.map {
@@ -328,5 +329,6 @@ struct RuntimeDeploymentFacts: Equatable, Sendable {
hasServers = !document.servers.isEmpty
aclSummary = document.aclRuntime?.summary
intentSHA256 = document.intentSHA256
configurationIntentSHA256 = document.replacingACL(nil).intentSHA256
}
}
@@ -51,45 +51,139 @@ final class AgentLifecycleChangeProtocolTests: AgentLifecycleTests {
"两次停止各转发一次 SIGTERM")
}
func testSIGUSR1WithACLChangeRestartsSslocalWithoutUnreadyReceipt() throws {
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)
}
func testDeferredACLSignalSurvivesSubsequentServerOnlyPublication() throws {
let runtimeStore = RuntimeFileStore(fileURL: contractURL)
let previous = ProxyRuntimeFixture.makeDocument()
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 {
runtimeStore.readRuntimeReceipt()?.deploymentID
== runtimeStore.receiptIdentity(for: previous)
}, "首次实例启动后应发布运行回执")
let previousProcessID = try XCTUnwrap(runtimeStore.readRuntimeReceipt()?.processID)
self.stateLog().contains("SIGUSR1") && self.stateLog().contains("SIGUSR2")
&& runtimeStore.readRuntimeReceipt()?.deploymentID
== runtimeStore.receiptIdentity(for: final)
})
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)))
}
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()
.replacingACL(.global(at: runtimeStore.aclFileURL))
try writeContract(previous)
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, "SIGUSR1 应送达 wrapper")
XCTAssertTrue(
try waitUntil {
return self.agentLog().contains("reload: listen change, restarting sslocal")
&& self.stateLog().components(separatedBy: "invoked:").count >= 3
},
"ACL 摘要变化应完整重启 sslocal;agent: \(agentLog())"
)
// 再发送 server-only reload,让就绪检查完成后给出明确的拒绝信号。
let serverReload = ProxyRuntimeFixture.makeDocument(serverAddress: "198.51.100.9")
.replacingACL(.direct(at: runtimeStore.aclFileURL))
try writeContract(serverReload)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
XCTAssertTrue(
try waitUntil { self.agentLog().contains("reload deferred until listeners are ready") },
"未绑定监听的重启实例应拒绝发布回执并推迟重载")
XCTAssertFalse(stateLog().contains("SIGUSR1"), "未就绪时不能转发 server-only reload")
XCTAssertNil(
runtimeStore.readRuntimeReceipt(),
"stub 未绑定代理端口时,不得为直连实例发布回执(旧 pid=\(previousProcessID))")
try waitUntil { self.agentLog().contains("reload deferred until listeners are ready") })
XCTAssertFalse(stateLog().contains("SIGUSR1"))
XCTAssertFalse(stateLog().contains("SIGUSR2"))
XCTAssertFalse(stateLog().contains("SIGTERM"))
XCTAssertEqual(stateLog().components(separatedBy: "invoked:").count - 1, 1)
XCTAssertNil(runtimeStore.readRuntimeReceipt())
kill(wrapper.processIdentifier, SIGTERM)
XCTAssertEqual(try waitForExit(wrapper), 0)
}
}
final class AgentLifecycleStaleListenerTests: AgentLifecycleTests {
@@ -161,7 +255,7 @@ final class AgentLifecycleListenDeadlineTests: AgentLifecycleTests {
private final class TCPListenerFixture {
let port: Int
private let descriptor: Int32
private var descriptor: Int32
init() throws {
let socketDescriptor = socket(AF_INET, SOCK_STREAM, 0)
@@ -199,7 +293,12 @@ private final class TCPListenerFixture {
port = Int(CFSwapInt16BigToHost(boundAddress.sin_port))
}
func closeListener() {
if descriptor >= 0 { close(descriptor) }
descriptor = -1
}
deinit {
close(descriptor)
closeListener()
}
}
@@ -53,6 +53,7 @@ class AgentLifecycleTests: XCTestCase {
trap 'rm -f "$child_pid_file"' EXIT
trap 'echo "SIGTERM" >> "$SSLOCAL_STUB_STATE"; exit 0' TERM
trap 'echo "SIGUSR1" >> "$SSLOCAL_STUB_STATE"' USR1
trap 'echo "SIGUSR2" >> "$SSLOCAL_STUB_STATE"' USR2
for other in "$SSLOCAL_STUB_PID_DIR"/*; do
if [ -f "$other" ] && kill -0 "${other##*/}" 2>/dev/null; then
echo "OVERLAPPING CHILD" >> "$SSLOCAL_STUB_STATE"
@@ -8,6 +8,93 @@ final class ProxyRuntimeModeCommandTests: ProxyRuntimeControllerTests {
BuiltinRuleSnapshots(loader: ProxyRuntimeFixture.controlFlowRuleSnapshot)
}
func testHealthyModeSwitchKeepsRunningWithoutProbingPorts() async throws {
let seeded = try makeSeededCatalog()
let probe = ProxyRuntimeFixture.FakeProbe.reachable()
let controller = makeController(probe: probe)
try await controller.activate(seeded.server)
try Data("42".utf8).write(to: runtime.pidFile)
let callsBefore = probe.callCount
let runtimeStore = RuntimeFileStore(fileURL: runtime.contract)
var statesAtReload: [ProxyRuntimeController.AgentRunState] = []
signals.reloadReceipt = {
statesAtReload.append(controller.state)
try? runtimeStore.writePublishedRuntimeReceipt(processID: 42)
}
await controller.setProxyMode(.direct)
await controller.setProxyMode(.global)
await controller.setProxyMode(.rule)
XCTAssertEqual(statesAtReload, [.running, .running, .running])
XCTAssertEqual(controller.state, .running)
XCTAssertEqual(probe.callCount, callsBefore, "热重载不得重新探测未变化的监听端口")
}
func testACLReloadWithReplacementChildStillProbesListeners() async throws {
let seeded = try makeSeededCatalog()
let probe = ProxyRuntimeFixture.FakeProbe.reachable()
let controller = makeController(probe: probe, processIsAlive: { [42, 43].contains($0) })
try await controller.activate(seeded.server)
try Data("42".utf8).write(to: runtime.pidFile)
let callsBefore = probe.callCount
let runtimeStore = RuntimeFileStore(fileURL: runtime.contract)
signals.reloadReceipt = { try? runtimeStore.writePublishedRuntimeReceipt(processID: 43) }
await controller.setProxyMode(.direct)
XCTAssertEqual(controller.state, .running)
XCTAssertGreaterThan(probe.callCount, callsBefore, "新子进程必须重新验证监听")
}
func testACLReloadRequiresCurrentReceiptWithoutProbingUnchangedListeners() async throws {
let seeded = try makeSeededCatalog()
let probe = ProxyRuntimeFixture.FakeProbe.reachable()
let controller = makeController(probe: probe)
try await controller.activate(seeded.server)
try Data("42".utf8).write(to: runtime.pidFile)
let callsBefore = probe.callCount
signals.reloadReceipt = {}
await controller.setProxyMode(.direct)
XCTAssertEqual(controller.state, .serviceFailed(.agent))
XCTAssertEqual(probe.callCount, callsBefore)
XCTAssertNotNil(RuntimeFileStore(fileURL: runtime.contract).readFailure())
}
func testNewModeSupersedesPendingACLReloadWithoutStartingOrProbing() async throws {
let seeded = try makeSeededCatalog()
let probe = ProxyRuntimeFixture.FakeProbe.reachable()
let waiting = expectation(description: "waiting for reload receipt")
let gate = RuleRuntimeDelayGate(started: waiting)
let controller = makeController(
probe: probe, launchHealthTimeoutSeconds: 5,
launchHealthRetryDelay: { try await gate.wait() })
try await controller.activate(seeded.server)
try Data("42".utf8).write(to: runtime.pidFile)
let callsBefore = probe.callCount
signals.reloadReceipt = {}
let pending = Task { await controller.setProxyMode(.direct) }
try await withTestTask(pending, release: { gate.resume() }) {
try await requireTestPhase([waiting])
XCTAssertEqual(controller.state, .running)
let store = RuntimeFileStore(fileURL: runtime.contract)
var stateAtReload: ProxyRuntimeController.AgentRunState?
self.signals.reloadReceipt = {
stateAtReload = controller.state
try? store.writePublishedRuntimeReceipt(processID: 42)
}
await controller.setProxyMode(.global)
XCTAssertEqual(stateAtReload, .running)
XCTAssertEqual(controller.state, .running)
XCTAssertEqual(probe.callCount, callsBefore)
}
XCTAssertTrue(controller.runtimeReloadProcesses.isEmpty)
XCTAssertEqual(controller.proxyMode, .global)
XCTAssertEqual(controller.state, .running)
}
func testSetProxyModePersistsTheChoiceAndARestoreRestoresIt() async throws {
let listeningStore = InMemoryProxyListeningStore()
let controller = makeController(
@@ -116,12 +116,13 @@ struct RuntimeConvergenceTests {
}
@Test(arguments: [false, true])
func unchangedModeACLOnlyChecksHealthWhenSystemProxyIsEnabled(
func unchangedModeACLDoesNotReprobeHealthyListeners(
systemProxyEnabled: Bool
) async throws {
let fixture = try Fixture(systemProxyEnabled: systemProxyEnabled)
defer { fixture.cleanUp() }
try await fixture.controller.activate(fixture.server)
try Data("42".utf8).write(to: fixture.runtime.pidFile)
let bytes = try Data(contentsOf: fixture.runtimeStore.fileURL)
let registers = fixture.agent.registerCount
let unregisters = fixture.agent.unregisterCount
@@ -138,7 +139,7 @@ struct RuntimeConvergenceTests {
#expect(fixture.agent.registerCount == registers)
#expect(fixture.agent.unregisterCount == unregisters)
#expect(fixture.signals.signalsSent.filter { $0.signal != 0 }.count == signals)
#expect(fixture.probe.ports.count == probes + (systemProxyEnabled ? 2 : 0))
#expect(fixture.probe.ports.count == probes)
}
@MainActor
@@ -1,9 +1,49 @@
import Darwin
import Foundation
import Testing
@testable import ShadowsocksX_NG2
struct RuntimePublicationTests {
@Test func aclAndServerChangesDoNotRequireRestart() throws {
let runtime = ProxyRuntimeFixture.makeTemporaryRuntime()
defer { try? FileManager.default.removeItem(at: runtime.directory) }
let store = RuntimeFileStore(fileURL: runtime.contract)
try store.write(
ProxyRuntimeFixture.makeDocument().replacingACL(.global(at: store.aclFileURL)))
let previous = try #require(RuntimePublication.load(at: store.supervisionFileURL))
try store.write(
ProxyRuntimeFixture.makeDocument(serverAddress: "198.51.100.9")
.replacingACL(.direct(at: store.aclFileURL)))
let next = try #require(RuntimePublication.load(at: store.supervisionFileURL))
#expect(next.acl != previous.acl)
#expect(next.configurationID != previous.configurationID)
#expect(next.reloadSignals(from: previous) == [SIGUSR1, SIGUSR2])
#expect(!next.requiresRestart(from: previous))
}
@Test func configurationIdentitySurvivesACLOnlyPublicationAndLegacyReloadIsConservative() throws {
let runtime = ProxyRuntimeFixture.makeTemporaryRuntime()
defer { try? FileManager.default.removeItem(at: runtime.directory) }
let store = RuntimeFileStore(fileURL: runtime.contract)
let source = ProxyRuntimeFixture.makeDocument().replacingACL(.global(at: store.aclFileURL))
try store.write(source)
let previous = try #require(RuntimePublication.load(at: store.supervisionFileURL))
try store.write(source.replacingACL(.direct(at: store.aclFileURL)))
let next = try #require(RuntimePublication.load(at: store.supervisionFileURL))
#expect(previous.configurationID != nil)
#expect(next.configurationID == previous.configurationID)
#expect(next.configuration != previous.configuration)
#expect(next.reloadSignals(from: previous) == [SIGUSR2])
let bytes = try JSONEncoder().encode(next)
var object = try #require(JSONSerialization.jsonObject(with: bytes) as? [String: Any])
object.removeValue(forKey: "configurationID")
let legacy = try JSONDecoder().decode(
RuntimeSupervision.self, from: JSONSerialization.data(withJSONObject: object))
#expect(legacy.hasValidStructure)
#expect(legacy.reloadSignals(from: legacy) == [SIGUSR1])
}
@Test func publicationDoesNotSelectActiveLinks() throws {
let runtime = ProxyRuntimeFixture.makeTemporaryRuntime()
defer { try? FileManager.default.removeItem(at: runtime.directory) }
@@ -218,6 +258,7 @@ extension RuntimePublicationTests {
let bytes = try Data(contentsOf: store.supervisionFileURL)
for (key, value) in [
("version", 1 as Any), ("version", 99 as Any), ("configuration", "../outside.json" as Any),
("configurationID", "invalid-identity" as Any),
] {
var object = try #require(JSONSerialization.jsonObject(with: bytes) as? [String: Any])
object[key] = value
@@ -40,7 +40,7 @@ final class RuntimePublicationSmokeTests: RealSslocalSmokeTests {
stopWrapperAndAssertCleanExit(wrapper, description: "reload wrapper exits")
}
func testLargeACLReuseAndSingleInstanceModeHandoff() throws {
func testLargeACLReuseAndModeReloadWithoutReplacingChild() throws {
let echo = try LoopbackEchoServer()
let proxy = try ConnectionCountingServer()
let ports = try grabThreeListenPorts(excluding: [echo.port, proxy.port])
@@ -69,8 +69,13 @@ final class RuntimePublicationSmokeTests: RealSslocalSmokeTests {
try store.write(direct)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
try assertReceiptOwnsListeners(document: direct)
XCTAssertNotEqual(store.readRuntimeReceipt()?.processID, oldPID)
XCTAssertNotEqual(kill(oldPID, 0), 0, "旧子进程退出后新监听才就绪")
XCTAssertEqual(store.readRuntimeReceipt()?.processID, oldPID)
// 回执不证明上游已应用 ACL;实际路由收敛后再验证两个入口。
XCTAssertTrue(
try waitForCondition(timeout: 8) {
self.performSocksConnectReply(
socksPort: listen.socksPort, targetHost: "127.0.0.1", targetPort: echo.port) == 0
})
let payload = Array("direct\n".utf8)
XCTAssertEqual(
try performDirectSocksEcho(
@@ -81,6 +86,14 @@ final class RuntimePublicationSmokeTests: RealSslocalSmokeTests {
try store.write(source)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
try assertReceiptOwnsListeners(document: source)
XCTAssertEqual(store.readRuntimeReceipt()?.processID, oldPID)
let before = proxy.connectionCount
XCTAssertTrue(
try waitForCondition(timeout: 8) {
self.performHTTPConnect(
httpPort: listen.httpPort, targetHost: "127.0.0.1", targetPort: echo.port)
return proxy.connectionCount > before
})
try assertPublicTargetsProxy(listen: listen, fakeSS: proxy)
XCTAssertEqual(RuntimePublication.load(at: store.supervisionFileURL)?.acl, acl)
XCTAssertEqual(
@@ -91,6 +104,113 @@ final class RuntimePublicationSmokeTests: RealSslocalSmokeTests {
stopWrapperAndAssertCleanExit(wrapper, description: "mode wrapper exits")
}
func testCombinedServerAndRuleReloadAndDefaultActionChangesKeepChild() throws {
let echo = try LoopbackEchoServer()
let firstExit = try ConnectionCountingServer()
let secondExit = try ConnectionCountingServer()
let ports = try grabThreeListenPorts(excluding: [echo.port, firstExit.port, secondExit.port])
let listen = ProxyListeningConfiguration(socksPort: ports[0], httpPort: ports[1])
func document(_ port: Int, _ acl: ProxyACLDocument) -> SslocalRuntimeDocument {
SslocalRuntimeDocument(
servers: [
.init(
id: "rules", server: "127.0.0.1", serverPort: port,
password: "smoke", method: "aes-256-gcm", plugin: nil, pluginOpts: nil)
], listen: listen, acl: acl)
}
let source = document(
firstExit.port, ruleProbeACL(defaultAction: .directWhenUnmatched, rules: []))
let wrapper = try launchWrapper(source)
let store = RuntimeFileStore(fileURL: contractURL)
try assertReceiptOwnsListeners(document: source)
let pid = try XCTUnwrap(store.readRuntimeReceipt()?.processID)
let custom = document(
secondExit.port,
ruleProbeACL(
defaultAction: .directWhenUnmatched,
rules: [ProxyRule(action: .proxy, match: try RuleMatch(domainSuffix: "blocked.example"))]))
try store.write(custom)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
try assertReceiptOwnsListeners(document: custom)
XCTAssertTrue(
try waitForCondition(timeout: 8) {
self.performHTTPConnect(
httpPort: listen.httpPort, targetHost: "blocked.example", targetPort: echo.port)
return secondExit.connectionCount > 0
})
XCTAssertEqual(store.readRuntimeReceipt()?.processID, pid)
let removed = document(
secondExit.port, ruleProbeACL(defaultAction: .directWhenUnmatched, rules: []))
try store.write(removed)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
try assertReceiptOwnsListeners(document: removed)
XCTAssertTrue(
try waitForCondition(timeout: 8) {
self.performSocksConnectReply(
socksPort: listen.socksPort, targetHost: "blocked.example", targetPort: echo.port) == 0
})
let payload = Array("reloaded rule\n".utf8)
XCTAssertEqual(
try performDirectHTTPEcho(
httpPort: listen.httpPort, targetHost: "blocked.example", targetPort: echo.port,
payload: payload), payload)
for (acl, proxiesDomain) in [
(ruleProbeACL(defaultAction: .proxyWhenUnmatched, rules: []), true),
(ProxyACLDocument.global(at: aclFileURL), false),
] {
let next = document(secondExit.port, acl)
let before = secondExit.connectionCount
try store.write(next)
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
try assertReceiptOwnsListeners(document: next)
XCTAssertTrue(
try waitForCondition(timeout: 8) {
if proxiesDomain {
self.performHTTPConnect(
httpPort: listen.httpPort, targetHost: "blocked.example", targetPort: echo.port)
return secondExit.connectionCount > before
}
return self.performSocksConnectReply(
socksPort: listen.socksPort, targetHost: "blocked.example", targetPort: echo.port) == 0
})
XCTAssertEqual(store.readRuntimeReceipt()?.processID, pid)
}
stopWrapperAndAssertCleanExit(wrapper, description: "rule reload wrapper exits")
}
func testRejectedACLReloadKeepsPreviousRulesWithoutReplacingChild() throws {
let echo = try LoopbackEchoServer()
let ports = try grabThreeListenPorts(excluding: [echo.port])
let listen = ProxyListeningConfiguration(socksPort: ports[0], httpPort: ports[1])
let source = SslocalRuntimeDocument(
servers: [], listen: listen, acl: .direct(at: aclFileURL))
let wrapper = try launchWrapper(source)
let store = RuntimeFileStore(fileURL: contractURL)
try assertReceiptOwnsListeners(document: source)
let pid = try XCTUnwrap(store.readRuntimeReceipt()?.processID)
let next = source.replacingACL(
ruleProbeACL(defaultAction: .proxyWhenUnmatched, rules: []))
try store.write(next)
let snapshot = try XCTUnwrap(RuntimePublication.load(at: store.supervisionFileURL))
let acl = try XCTUnwrap(snapshot.acl)
// 模拟发布后磁盘损坏;上游解析拒绝,Agent 回执无法识别这种内部失败。
try AtomicFileWriter.write(Data("[\n".utf8), to: workDir.appendingPathComponent(acl.file))
XCTAssertEqual(kill(wrapper.processIdentifier, SIGUSR1), 0)
XCTAssertTrue(
try waitForCondition(timeout: 8) {
let log = try String(contentsOf: self.workDir.appendingPathComponent("agent.log"))
return log.contains("ACL reload failed") && log.contains("keeping the previous rules")
})
try assertReceiptOwnsListeners(document: next)
XCTAssertEqual(store.readRuntimeReceipt()?.processID, pid)
XCTAssertNil(store.readFailure())
let payload = Array("previous rules\n".utf8)
XCTAssertEqual(
try performDirectSocksEcho(
socksPort: listen.socksPort, targetPort: echo.port, payload: payload), payload)
stopWrapperAndAssertCleanExit(wrapper, description: "rejected ACL wrapper exits")
}
func testHealthyCrashReplaysCurrentDeploymentAndRecoveryFailureDoesNotLoop() throws {
let ports = try grabThreeListenPorts(excluding: [])
let document = SslocalRuntimeDocument(
+2
View File
@@ -2,6 +2,8 @@
**Status**: accepted; system-proxy rollback clause superseded by ADR-0019. Runtime and mode rollback remain.
**ACL reload superseded (2026-10-20)**: [ADR-0048](0048-reload-sslocal-acl-by-signal.md) replaces restart-on-ACL-change with `SIGUSR2` reload. Listener and authentication changes still require restart.
## Context
Global mode previously projected only the system SOCKS endpoint and relied on sslocal's default proxy-everything behaviour. That left local-network targets without a product-owned direct guarantee: LAN-sharing clients, IPv6 destinations whose system-proxy exceptions macOS does not reliably honour, and any request that reaches the SOCKS/HTTP inbounds still needed a defined routing policy. The ACL work (spec #59) requires global mode to mean "public targets default through Shadowsocks, local targets stay direct", using only the fixed local safety rules.
@@ -8,6 +8,8 @@
**Supersedes in part**: ADR-0009 and ADR-0010's "ACL path, content, digest, and summary are runtime identity" — content is no longer carried in the contract, and the ACL path is a stable link that does not change across mode switches. The restart-on-change rule itself stands.
**ACL reload superseded (2026-10-20)**: [ADR-0048](0048-reload-sslocal-acl-by-signal.md) replaces restart-on-ACL-change with `SIGUSR2` reload. Listener and authentication changes still require restart.
## Context
sslocal's routing policy is one of four ACL shapes — `direct`, `global`, `rule-proxy-default`, `rule-direct-default` — but all four were written into a single `sslocal-active.acl`, overwritten on every proxy mode switch and on every deploy even when the ACL body had not changed. The contract JSON also embedded the full ACL body (`x_shadowsocksx_ng_acl.content`), so a rule-mode switch paid roughly the same large write twice. Upstream sslocal is gaining signal-triggered ACL reload; mode switches should be cheap now and restart-free later.
@@ -2,7 +2,7 @@
**Status**:accepted(2026-10-19)。已实施;验证结果见对应实现提交。
后续修订:GUI 与上游配置的单向输出及私有意图比较见 [ADR-0043](0043-one-way-sslocal-configuration-output.md)。
后续修订:GUI 与上游配置的单向输出及私有意图比较见 [ADR-0043](0043-one-way-sslocal-configuration-output.md)。ACL 变化改用信号热重载见 [ADR-0048](0048-reload-sslocal-acl-by-signal.md),取代下文 ACL 变化必须重启的规定。
## 职责
@@ -0,0 +1,19 @@
# 通过信号热重载 sslocal ACL
**状态**:已接受并实施(2026-10-20)。
使用已固定的 sslocal fork v1.26.0 的 `SIGUSR2` 能力,让规则修改、规则默认动作变化,以及规则/全局/直连模式切换产生的 ACL 更新在原子切换稳定 ACL 链接后热重载,减少子进程重启造成的中断。GUI 继续向 Agent 发送 `SIGUSR1` 通知新部署,Agent 根据变化向 sslocal 发送相应信号。监听、IPv6 监听设置或认证变化仍按原协议停止旧子进程、确认退出后重新启动。
监管资料通过随机 `configurationID` 表示 sslocal 配置字节的身份,GUI 在旧配置字节及其私有完整性摘要均与本次输出一致时复用该身份。Agent 只比较身份,配置变化时发送 `SIGUSR1`,ACL 变化时发送 `SIGUSR2`;仅切换代理模式或更新 ACL 不重载服务器配置。身份不包含服务器或凭据摘要,Agent 仍不读取秘密配置。第 2 版监管资料允许缺失此字段,遇到旧资料时保守发送 `SIGUSR1`。服务器与 ACL 同时变化时分别发送 `SIGUSR1` 和 `SIGUSR2`,接受两者先后生效的短暂窗口,不为组合变化单独重启,也不承诺原子应用。上游 `SIGUSR1` 仅刷新服务器,不能代替 ACL 重载。
Agent 对已经健康、无需重启的同一子进程沿用已建立的监听事实,不在热重载前后重复连接端口,也不因信号重载将其降为未就绪。首次启动或重启仍验证新子进程的监听归属。启动监听尚未就绪时合并保留待发信号,监听就绪后发送,防止后续仅服务器更新覆盖之前待发的 ACL 更新。正式 Agent 部署在所有代理模式下始终生成 ACL,不为模型允许的 ACL 空值增加产品转换场景。
GUI 以仅保存在内存中的非 ACL 意图摘要识别纯 ACL 更新;前提是已处于健康运行态、Agent 已注册、wrapper 与当前回执对应的子进程仍存活。更新期间保持运行状态,不呈现“正在启动代理”,等待当前部署回执但不重复探测未变监听。回执对应子进程 PID 变化时恢复启动状态与完整端口检查;缺少回执、持久失败或进程消失仍报告失败。后台健康轮询避开正在执行的热重载,周期性健康检查在更新完成后继续运行。
沿用现有回执的证据范围:它只记录 Agent 操作及监听证据,不代表 sslocal 已确认应用新规则。上游在 ACL 读取或解析失败时保留旧规则并记录日志;信号发送成功不能排除这种情况。本次接受此限制,不增加上游应用确认协议或根据日志推断部署成功。链接切换、信号投递或监听检查等可检测的失败继续遵循 ADR-0037,停止服务、暂停系统代理并保留最新意图,不自动恢复旧部署。
本决定部分替代 ADR-0010、ADR-0011、ADR-0038 中 ACL 变化要求重启的规定;稳定链接的 Agent 独占切换、不可变文件、代次与锁、子进程引用的保守保留及失败屏障继续适用。当前实现仍以代码为准。
上游依据:[信号处理](https://github.com/qiuyuzhou/shadowsocks-rust/blob/v1.26.0/src/service/local.rs#L1078-L1165)、[ACL 重新加载及原子替换](https://github.com/qiuyuzhou/shadowsocks-rust/blob/v1.26.0/crates/shadowsocks-service/src/acl/mod.rs#L687-L738)。
验证通过:1150 项 hostless 测试、19 项真实进程 smoke、Lint 与 Debug 构建。定向覆盖模式切换保持运行状态且不重复探测端口、回执缺失与子进程替换、连续模式切换、配置身份复用与旧监管资料保守重载、Agent 热重载前后实际 TCP 连接计数不增加、启动期间待发信号合并、组合服务器与 ACL 更新、规则增删与默认动作变化、模式切换时子进程 PID 保持不变且实际路由变化,以及上游 ACL 解析失败保留旧规则的回执边界。未执行原生界面与真实系统代理验收。