From 09e4d2e097a7c43deff3a40a26035310c182811e Mon Sep 17 00:00:00 2001 From: Qiu Yuzhou Date: Fri, 23 Oct 2026 03:46:19 +0800 Subject: [PATCH] fix(tests): avoid blocking async test workers --- ...oxyRuntimeController+SystemProxyGate.swift | 4 +- .../ProxyRuntime/SystemEndpointProbe.swift | 4 +- .../Services/FirewallStatusChecker.swift | 8 +- .../SettingsWorkflow/ListenerOccupancy.swift | 21 +++-- .../Domain/Rules/BuiltinRuleCatalog.swift | 4 +- .../Domain/Runtime/PortOccupancyProbe.swift | 4 +- .../Domain/Runtime/ProxyPortSemantics.swift | 9 ++ .../ProxyPortSemanticsTests.swift | 4 +- .../ProxyRuntimeControllerModeTests.swift | 8 +- .../Tests/Rules/BuiltinRuleCatalogTests.swift | 15 ++-- .../Rules/RuleEnablementControllerTests.swift | 8 +- .../Runtime/FirewallStatusCheckerTests.swift | 13 +-- .../ListenerOccupancyLifecycleTests.swift | 4 +- .../Tests/Support/BoundedTestWait.swift | 70 ++++++++++++--- .../ProxyRuntimeBlockingFixtures.swift | 86 +++++++++++++++++++ .../Tests/Support/ProxyRuntimeFixture.swift | 83 +++--------------- .../Tests/Support/RuleRuntimeDelayGate.swift | 6 +- .../Support/SettingsWorkflowTestSupport.swift | 31 +++++-- .../TestSupport/BoundedTestWaitTests.swift | 19 ++-- 19 files changed, 252 insertions(+), 149 deletions(-) create mode 100644 ShadowsocksX-NG2/Tests/Support/ProxyRuntimeBlockingFixtures.swift diff --git a/ShadowsocksX-NG2/App/Application/ProxyRuntime/ProxyRuntimeController+SystemProxyGate.swift b/ShadowsocksX-NG2/App/Application/ProxyRuntime/ProxyRuntimeController+SystemProxyGate.swift index 5eab85b3..63e68e4e 100644 --- a/ShadowsocksX-NG2/App/Application/ProxyRuntime/ProxyRuntimeController+SystemProxyGate.swift +++ b/ShadowsocksX-NG2/App/Application/ProxyRuntime/ProxyRuntimeController+SystemProxyGate.swift @@ -60,7 +60,7 @@ extension ProxyRuntimeController { private func firewallStatus(for executableURL: URL) async -> FirewallBlockStatus { let checker = firewallChecker return await Task.detached(priority: .utility) { - checker.status(for: executableURL) + await checker.status(for: executableURL) }.value } @@ -69,7 +69,7 @@ extension ProxyRuntimeController { ) async -> EndpointHealthProbe.Outcome { let probe = probe return await Task.detached(priority: .utility) { - probe.probe(host: host, port: port, timeout: timeout) + await probe.probe(host: host, port: port, timeout: timeout) }.value } diff --git a/ShadowsocksX-NG2/App/Application/ProxyRuntime/SystemEndpointProbe.swift b/ShadowsocksX-NG2/App/Application/ProxyRuntime/SystemEndpointProbe.swift index 7f3eb5b5..9c93cf87 100644 --- a/ShadowsocksX-NG2/App/Application/ProxyRuntime/SystemEndpointProbe.swift +++ b/ShadowsocksX-NG2/App/Application/ProxyRuntime/SystemEndpointProbe.swift @@ -2,11 +2,11 @@ import Foundation /// 端点探测缝：注入以便控制器单测（真实探测走 EndpointHealthProbe）。 protocol EndpointProbing: Sendable { - func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome + func probe(host: String, port: Int, timeout: TimeInterval) async -> EndpointHealthProbe.Outcome } struct SystemEndpointProbe: EndpointProbing { - func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome { + func probe(host: String, port: Int, timeout: TimeInterval) async -> EndpointHealthProbe.Outcome { EndpointHealthProbe.probe(host: host, port: port, timeout: timeout) } } diff --git a/ShadowsocksX-NG2/App/Application/Services/FirewallStatusChecker.swift b/ShadowsocksX-NG2/App/Application/Services/FirewallStatusChecker.swift index 7367c2b2..434c4a38 100644 --- a/ShadowsocksX-NG2/App/Application/Services/FirewallStatusChecker.swift +++ b/ShadowsocksX-NG2/App/Application/Services/FirewallStatusChecker.swift @@ -7,7 +7,7 @@ enum FirewallBlockStatus: Equatable, Sendable { } protocol FirewallStatusChecking: Sendable { - func status(for executableURL: URL) -> FirewallBlockStatus + func status(for executableURL: URL) async -> FirewallBlockStatus } /// `socketfilterfw --getappblocked` 是应用防火墙拒绝态的只读系统边界。 @@ -23,7 +23,11 @@ struct SocketFilterFirewallChecker: FirewallStatusChecking, @unchecked Sendable self.runCommand = runCommand } - func status(for executableURL: URL) -> FirewallBlockStatus { + func status(for executableURL: URL) async -> FirewallBlockStatus { + statusSynchronously(for: executableURL) + } + + private func statusSynchronously(for executableURL: URL) -> FirewallBlockStatus { guard let output = try? runCommand( Self.toolURL, ["--getappblocked", executableURL.path]) diff --git a/ShadowsocksX-NG2/App/Application/SettingsWorkflow/ListenerOccupancy.swift b/ShadowsocksX-NG2/App/Application/SettingsWorkflow/ListenerOccupancy.swift index 07250047..65430999 100644 --- a/ShadowsocksX-NG2/App/Application/SettingsWorkflow/ListenerOccupancy.swift +++ b/ShadowsocksX-NG2/App/Application/SettingsWorkflow/ListenerOccupancy.swift @@ -154,11 +154,11 @@ final class ListenerOccupancy: ObservableObject { let excluded = Set(ProxyEndpointKind.allCases.map { context.proposedListen.port(for: $0) }) let probe = probe let candidate = await Task.detached(priority: .utility) { - PortSuggestion.firstFree(excluding: excluded) { port in + await PortSuggestion.firstFreeAsync(excluding: excluded) { port in let request = PortOccupancyProbeRequest( endpoint: endpoint, listen: context.proposedListen.replacingPort(port, for: endpoint), port: port) - if case .free = probe.occupancy(for: request) { return true } + if case .free = await probe.occupancy(for: request) { return true } return false } }.value @@ -202,15 +202,14 @@ final class ListenerOccupancy: ObservableObject { -> [SettingsPortID: PortOccupancy] { await Task.detached(priority: .utility) { - Dictionary( - uniqueKeysWithValues: SettingsPortID.allCases.map { id in - let endpoint = SettingsPortAdapter.endpoint(for: id, group: context.draft.group) - return ( - id, - probe.occupancy( - for: PortOccupancyProbeRequest(endpoint: endpoint, listen: context.proposedListen)) - ) - }) + var results: [SettingsPortID: PortOccupancy] = [:] + for id in SettingsPortID.allCases { + let endpoint = SettingsPortAdapter.endpoint(for: id, group: context.draft.group) + let request = PortOccupancyProbeRequest( + endpoint: endpoint, listen: context.proposedListen) + results[id] = await probe.occupancy(for: request) + } + return results }.value } } diff --git a/ShadowsocksX-NG2/Domain/Rules/BuiltinRuleCatalog.swift b/ShadowsocksX-NG2/Domain/Rules/BuiltinRuleCatalog.swift index 6b58c749..87577fa7 100644 --- a/ShadowsocksX-NG2/Domain/Rules/BuiltinRuleCatalog.swift +++ b/ShadowsocksX-NG2/Domain/Rules/BuiltinRuleCatalog.swift @@ -52,7 +52,7 @@ struct BuiltinRuleCatalog { /// App-process source facts. Concurrent consumers share one attempt per source; /// successful bundled snapshots never expire, failed attempts require explicit retry. actor BuiltinRuleSnapshots { - typealias Loader = @Sendable (RulesSource) throws -> RuleSnapshot + typealias Loader = @Sendable (RulesSource) async throws -> RuleSnapshot private struct Attempt { let id: UUID let task: Task @@ -81,7 +81,7 @@ actor BuiltinRuleSnapshots { attempt = Attempt( id: UUID(), task: Task.detached(priority: .userInitiated) { - let snapshot = try loader(source) + let snapshot = try await loader(source) guard let kind = source.sourceKind, kind != .custom else { throw RuleSnapshotError.missing } diff --git a/ShadowsocksX-NG2/Domain/Runtime/PortOccupancyProbe.swift b/ShadowsocksX-NG2/Domain/Runtime/PortOccupancyProbe.swift index c551a389..bb504299 100644 --- a/ShadowsocksX-NG2/Domain/Runtime/PortOccupancyProbe.swift +++ b/ShadowsocksX-NG2/Domain/Runtime/PortOccupancyProbe.swift @@ -148,7 +148,7 @@ struct PortOccupancyProbeRequest: Equatable, Sendable { /// 端口占用探测缝：设置编辑期的即时占用校验。激活是否成功 /// 仍以 runtime 实际绑定为准（健康门禁），本探测不构成权威判定。 protocol PortOccupancyProbing: Sendable { - func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy + func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy } /// 系统实现：对 `bindAddress:port` 做一次不带复用选项的 TCP bind + listen， @@ -156,7 +156,7 @@ protocol PortOccupancyProbing: Sendable { /// 允许共存，到 listen 阶段才报告冲突，因此两步都要覆盖。占用时尽力以 /// `lsof` 解析监听进程名。探测不设置 SO_REUSEADDR/SO_REUSEPORT，避免假空闲。 struct SystemPortOccupancyProbe: PortOccupancyProbing { - func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy { + func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy { occupancy(port: request.port, listenerMode: request.listenerMode) } diff --git a/ShadowsocksX-NG2/Domain/Runtime/ProxyPortSemantics.swift b/ShadowsocksX-NG2/Domain/Runtime/ProxyPortSemantics.swift index 441ee1d8..4569df22 100644 --- a/ShadowsocksX-NG2/Domain/Runtime/ProxyPortSemantics.swift +++ b/ShadowsocksX-NG2/Domain/Runtime/ProxyPortSemantics.swift @@ -39,6 +39,15 @@ enum PortSuggestion { } return nil } + + static func firstFreeAsync( + excluding excludedPorts: Set, isFree: (Int) async -> Bool + ) async -> Int? { + for candidate in range where !excludedPorts.contains(candidate) { + if await isFree(candidate) { return candidate } + } + return nil + } } extension ProxyListeningConfiguration { diff --git a/ShadowsocksX-NG2/Tests/ProxyControl/ProxyPortSemanticsTests.swift b/ShadowsocksX-NG2/Tests/ProxyControl/ProxyPortSemanticsTests.swift index 74c317c3..861cd0bb 100644 --- a/ShadowsocksX-NG2/Tests/ProxyControl/ProxyPortSemanticsTests.swift +++ b/ShadowsocksX-NG2/Tests/ProxyControl/ProxyPortSemanticsTests.swift @@ -105,14 +105,14 @@ final class ProxyPortSemanticsTests: XCTestCase { } } - func testBindProbeReportsUnknownWhenBindIsNotPermitted() throws { + func testBindProbeReportsUnknownWhenBindIsNotPermitted() async throws { try XCTSkipIf(getuid() == 0, "root 可绑定特权端口，无法用 EACCES 构造 .unknown") // 特权端口：非 root 进程 bind 失败（EACCES）≠ 占用，也不得冒充空闲。 let request = PortOccupancyProbeRequest( endpoint: .socks, listen: RuntimeListenFacts(listen: ProxyListeningConfiguration()), port: 80) - let outcome = SystemPortOccupancyProbe().occupancy(for: request) + let outcome = await SystemPortOccupancyProbe().occupancy(for: request) guard case .unknown = outcome else { return XCTFail("绑定失败时必须报 .unknown，不得冒充空闲或占用，实际 \(outcome)") diff --git a/ShadowsocksX-NG2/Tests/ProxyRuntime/ProxyRuntimeControllerModeTests.swift b/ShadowsocksX-NG2/Tests/ProxyRuntime/ProxyRuntimeControllerModeTests.swift index 7fa2864b..f160c3a0 100644 --- a/ShadowsocksX-NG2/Tests/ProxyRuntime/ProxyRuntimeControllerModeTests.swift +++ b/ShadowsocksX-NG2/Tests/ProxyRuntime/ProxyRuntimeControllerModeTests.swift @@ -72,12 +72,12 @@ extension ProxyRuntimeControllerModeTests { func testAgentOffDuringModeRuleDerivationSupersedesDeployment() async throws { let seeded = try makeSeededCatalog() let started = expectation(description: "rule snapshot load paused") - let release = DispatchSemaphore(value: 0) - defer { release.signal() } + let release = TestAsyncGate("rule snapshot load release") + defer { release.release() } let loader = BuiltinRuleSnapshots(loader: { source -> RuleSnapshot in if source == .geolocationCN { started.fulfill() - release.wait() + _ = await release.wait() } return rulesFixture(source) }) @@ -94,7 +94,7 @@ extension ProxyRuntimeControllerModeTests { await controller.setAgentEnabled(false) XCTAssertEqual(controller.state, .off) let registrationsAfterStop = agent.registerCount - release.signal() + release.release() await modeSwitch.value XCTAssertEqual(agent.registerCount, registrationsAfterStop, "派生被取代后不得部署") diff --git a/ShadowsocksX-NG2/Tests/Rules/BuiltinRuleCatalogTests.swift b/ShadowsocksX-NG2/Tests/Rules/BuiltinRuleCatalogTests.swift index 98a8f6e3..9f4206fd 100644 --- a/ShadowsocksX-NG2/Tests/Rules/BuiltinRuleCatalogTests.swift +++ b/ShadowsocksX-NG2/Tests/Rules/BuiltinRuleCatalogTests.swift @@ -9,11 +9,12 @@ final class BuiltinRuleCatalogTests: XCTestCase { func testConcurrentConsumersShareOneSourceLoad() async throws { let started = expectation(description: "source read") let loader = SharedRuleSourceLoader(started: started) - let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) }) + defer { loader.release.release() } + let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) }) async let first = snapshots.load(.gfwlist) await fulfillment(of: [started], timeout: 3) async let second = snapshots.load(.gfwlist) - loader.release.signal() + loader.release.release() let (one, two) = try await (first, second) XCTAssertEqual(one, two) _ = try await snapshots.load(.gfwlist) @@ -22,7 +23,7 @@ final class BuiltinRuleCatalogTests: XCTestCase { func testRetryReadsOnlyFailedSourcesAndSuccessNeverExpires() async throws { let loader = SharedRuleSourceLoader() - let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) }) + let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) }) let first = await snapshots.browsingSources() XCTAssertThrowsError(try first[.chinaIPv4]?.get()) _ = await snapshots.browsingSources() @@ -37,7 +38,7 @@ final class BuiltinRuleCatalogTests: XCTestCase { @MainActor func testFirstBrowsingRetainsRuntimeSourceFailureUntilExplicitRetry() async throws { let loader = SharedRuleSourceLoader() - let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) }) + let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) }) do { _ = try await snapshots.load(.chinaIPv4) XCTFail("The first runtime source request should fail") @@ -187,17 +188,17 @@ private final class SharedRuleSourceLoader: @unchecked Sendable { private let lock = NSLock() private var counts: [RulesSource: Int] = [:] private let started: XCTestExpectation? - let release = DispatchSemaphore(value: 0) + let release = TestAsyncGate("shared rule source release") init(started: XCTestExpectation? = nil) { self.started = started } func count(_ source: RulesSource) -> Int { lock.withLock { counts[source, default: 0] } } - func load(_ source: RulesSource) throws -> RuleSnapshot { + func load(_ source: RulesSource) async throws -> RuleSnapshot { let count = lock.withLock { counts[source, default: 0] += 1 return counts[source, default: 0] } if let started { started.fulfill() - release.wait() + _ = await release.wait() } else if source == .chinaIPv4 && count == 1 { throw RuleSnapshotError.missing } diff --git a/ShadowsocksX-NG2/Tests/Rules/RuleEnablementControllerTests.swift b/ShadowsocksX-NG2/Tests/Rules/RuleEnablementControllerTests.swift index 76924932..275cccf6 100644 --- a/ShadowsocksX-NG2/Tests/Rules/RuleEnablementControllerTests.swift +++ b/ShadowsocksX-NG2/Tests/Rules/RuleEnablementControllerTests.swift @@ -124,7 +124,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests { store: store, settings: ProxyRuntimeSettings( listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true), - proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) })) + proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) })) try await controller.activate(seeded.server) let runtimeStore = RuntimeFileStore(fileURL: runtime.contract) let previous = try XCTUnwrap(runtimeStore.loadDocument()) @@ -164,7 +164,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests { store: store, settings: ProxyRuntimeSettings( listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true), - proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) })) + proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) })) try await controller.activate(seeded.server) let modeSwitch = Task { await controller.setProxyMode(.rule) } try await withTestTask( @@ -187,7 +187,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests { let started = expectation(description: "background source preparation") let loader = PausedRuntimeRuleSource(started: started) defer { loader.resume() } - let snapshots = BuiltinRuleSnapshots(loader: { loader.load($0) }) + let snapshots = BuiltinRuleSnapshots(loader: { await loader.load($0) }) let settings = ProxyRuntimeSettings( listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true) let controller = makeRuleEnablementController( @@ -237,7 +237,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests { store: store, settings: ProxyRuntimeSettings( listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true), - proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) })) + proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) })) try await controller.activate(seeded.server) let old = Task { await controller.setProxyMode(.rule) } try await withTestTask( diff --git a/ShadowsocksX-NG2/Tests/Runtime/FirewallStatusCheckerTests.swift b/ShadowsocksX-NG2/Tests/Runtime/FirewallStatusCheckerTests.swift index 7e06cf34..658402e4 100644 --- a/ShadowsocksX-NG2/Tests/Runtime/FirewallStatusCheckerTests.swift +++ b/ShadowsocksX-NG2/Tests/Runtime/FirewallStatusCheckerTests.swift @@ -3,7 +3,7 @@ import XCTest @testable import ShadowsocksX_NG2 final class FirewallStatusCheckerTests: XCTestCase { - func testGetAppBlockedOutputMapsToBlockedStatus() { + func testGetAppBlockedOutputMapsToBlockedStatus() async { var capturedExecutable: URL? var capturedArguments: [String] = [] let checker = SocketFilterFirewallChecker { executable, arguments in @@ -13,14 +13,15 @@ final class FirewallStatusCheckerTests: XCTestCase { } let target = URL(fileURLWithPath: "/bundle/Helpers/sslocal") - XCTAssertEqual(checker.status(for: target), .blocked) + let blocked = await checker.status(for: target) + XCTAssertEqual(blocked, .blocked) XCTAssertEqual( capturedExecutable?.path, "/usr/libexec/ApplicationFirewall/socketfilterfw") XCTAssertEqual(capturedArguments, ["--getappblocked", target.path]) } - func testPermittedAndUnknownResultsDegradeWithoutFalseBlock() { + func testPermittedAndUnknownResultsDegradeWithoutFalseBlock() async { let permitted = SocketFilterFirewallChecker { _, _ in "Incoming connection to /bundle/Helpers/sslocal is permitted.\n" } @@ -29,7 +30,9 @@ final class FirewallStatusCheckerTests: XCTestCase { } let target = URL(fileURLWithPath: "/bundle/Helpers/sslocal") - XCTAssertEqual(permitted.status(for: target), .permitted) - XCTAssertEqual(unavailable.status(for: target), .unknown) + let permittedStatus = await permitted.status(for: target) + let unavailableStatus = await unavailable.status(for: target) + XCTAssertEqual(permittedStatus, .permitted) + XCTAssertEqual(unavailableStatus, .unknown) } } diff --git a/ShadowsocksX-NG2/Tests/Settings/ListenerOccupancyLifecycleTests.swift b/ShadowsocksX-NG2/Tests/Settings/ListenerOccupancyLifecycleTests.swift index 0bf8eb3f..3f951fd3 100644 --- a/ShadowsocksX-NG2/Tests/Settings/ListenerOccupancyLifecycleTests.swift +++ b/ShadowsocksX-NG2/Tests/Settings/ListenerOccupancyLifecycleTests.swift @@ -155,7 +155,7 @@ struct ListenerOccupancyLifecycleTests { private final class SuggestionProbe: PortOccupancyProbing, Sendable { let gate = GatedOccupancyProbe(gatedAnswer: .free, passThroughAnswer: .free) - func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy { - request.port >= PortSuggestion.range.lowerBound ? gate.occupancy(for: request) : .free + func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy { + request.port >= PortSuggestion.range.lowerBound ? await gate.occupancy(for: request) : .free } } diff --git a/ShadowsocksX-NG2/Tests/Support/BoundedTestWait.swift b/ShadowsocksX-NG2/Tests/Support/BoundedTestWait.swift index 9d504ad9..3466235b 100644 --- a/ShadowsocksX-NG2/Tests/Support/BoundedTestWait.swift +++ b/ShadowsocksX-NG2/Tests/Support/BoundedTestWait.swift @@ -46,11 +46,16 @@ func withTestTask( await finish() } -/// Broadcast release is remembered, including when cleanup runs before a worker enters. -final class TestBlockingGate: @unchecked Sendable { - private let group = DispatchGroup() +/// Broadcast async release is remembered, and timed-out waiters do not occupy a worker thread. +final class TestAsyncGate: @unchecked Sendable { + private struct Waiter { + let continuation: CheckedContinuation + var timeoutTask: Task? + } + private let lock = NSLock() private var released = false + private var waiters: [UUID: Waiter] = [:] private let onTimeout: @Sendable () -> Void init( @@ -61,7 +66,6 @@ final class TestBlockingGate: @unchecked Sendable { if let onTimeout { self.onTimeout = onTimeout } else if let test = Test.current { - // Detached workers have no current test; retain the creating test and source location. let message = Comment( rawValue: "Timed out waiting for \(description) (created by \(test.name))") let location = SourceLocation( @@ -70,22 +74,64 @@ final class TestBlockingGate: @unchecked Sendable { } else { self.onTimeout = { XCTFail("Timed out waiting for \(description)", file: file, line: line) } } - group.enter() } - func wait(timeout: TimeInterval = 10) -> Bool { - guard group.wait(timeout: .now() + timeout) == .success else { - onTimeout() - return false + func wait(timeout: Duration = .seconds(10)) async -> Bool { + let id = UUID() + return await withCheckedContinuation { continuation in + lock.lock() + guard !released else { + lock.unlock() + continuation.resume(returning: true) + return + } + waiters[id] = Waiter(continuation: continuation) + lock.unlock() + + let timeoutTask = Task { [weak self] in + do { + try await Task.sleep(for: timeout) + self?.expire(id) + } catch { + // Release cancels the timer after resuming the waiter. + } + } + + lock.lock() + if var waiter = waiters[id] { + waiter.timeoutTask = timeoutTask + waiters[id] = waiter + lock.unlock() + } else { + lock.unlock() + timeoutTask.cancel() + } } - return true } func release() { lock.lock() - let shouldRelease = !released + guard !released else { + lock.unlock() + return + } released = true + let pending = Array(waiters.values) + waiters.removeAll() lock.unlock() - if shouldRelease { group.leave() } + + for waiter in pending { + waiter.timeoutTask?.cancel() + waiter.continuation.resume(returning: true) + } + } + + private func expire(_ id: UUID) { + lock.lock() + let waiter = waiters.removeValue(forKey: id) + lock.unlock() + guard let waiter else { return } + onTimeout() + waiter.continuation.resume(returning: false) } } diff --git a/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeBlockingFixtures.swift b/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeBlockingFixtures.swift new file mode 100644 index 00000000..945ebcb3 --- /dev/null +++ b/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeBlockingFixtures.swift @@ -0,0 +1,86 @@ +import Foundation +import XCTest + +@testable import ShadowsocksX_NG2 + +extension ProxyRuntimeFixture { + /// 可暂停探测替身：arm 后的下一次探测异步等待测试放行，在启动健康门的 + /// await 窗口内确定性插入交错命令，不占用线程池工作线程。 + final class BlockingProbe: EndpointProbing, @unchecked Sendable { + private let lock = NSLock() + private var gate: TestAsyncGate? + private var started: XCTestExpectation? + + func arm(started: XCTestExpectation) { + lock.lock() + defer { lock.unlock() } + precondition(gate == nil, "上一次 arm 尚未 release") + gate = TestAsyncGate("runtime fixture release") + self.started = started + } + + func release() { + lock.lock() + let gate = self.gate + self.gate = nil + self.started = nil + lock.unlock() + gate?.release() + } + + func probe(host: String, port: Int, timeout: TimeInterval) async + -> EndpointHealthProbe.Outcome + { + let (gate, started) = armedGate() + if let gate { + started?.fulfill() + _ = await gate.wait() + } + return .reachable + } + + private func armedGate() -> (TestAsyncGate?, XCTestExpectation?) { + lock.lock() + let gate = self.gate + let started = self.started + lock.unlock() + return (gate, started) + } + } + + /// 按调用序返回预设结果，并可在指定调用处异步等待测试放行。 + final class BlockingFirewallChecker: FirewallStatusChecking, @unchecked Sendable { + private let lock = NSLock() + private var callCount = 0 + private let outcomes: [FirewallBlockStatus] + private let blockOnCall: Int + private let started: XCTestExpectation + private let gate = TestAsyncGate("runtime fixture release") + + init(outcomes: [FirewallBlockStatus], blockOnCall: Int, started: XCTestExpectation) { + precondition(!outcomes.isEmpty) + self.outcomes = outcomes + self.blockOnCall = blockOnCall + self.started = started + } + + func release() { gate.release() } + + func status(for executableURL: URL) async -> FirewallBlockStatus { + let call = nextCall() + if call == blockOnCall { + started.fulfill() + _ = await gate.wait() + } + return outcomes[min(call, outcomes.count) - 1] + } + + private func nextCall() -> Int { + lock.lock() + callCount += 1 + let call = callCount + lock.unlock() + return call + } + } +} diff --git a/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeFixture.swift b/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeFixture.swift index 0a545b1f..f65ba2a6 100644 --- a/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeFixture.swift +++ b/ShadowsocksX-NG2/Tests/Support/ProxyRuntimeFixture.swift @@ -206,7 +206,13 @@ enum ProxyRuntimeFixture { return requestedPorts } - func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome { + func probe(host: String, port: Int, timeout: TimeInterval) async + -> EndpointHealthProbe.Outcome + { + recordProbe(port: port) + } + + private func recordProbe(port: Int) -> EndpointHealthProbe.Outcome { lock.lock() defer { lock.unlock() } callCount += 1 @@ -239,7 +245,11 @@ enum ProxyRuntimeFixture { self.outcomes = outcomes } - func status(for executableURL: URL) -> FirewallBlockStatus { + func status(for executableURL: URL) async -> FirewallBlockStatus { + recordStatus(for: executableURL) + } + + private func recordStatus(for executableURL: URL) -> FirewallBlockStatus { lock.lock() defer { lock.unlock() } recordedURLs.append(executableURL) @@ -249,75 +259,6 @@ enum ProxyRuntimeFixture { } } - /// 可暂停探测替身：arm 后的下一次探测挂起在信号量上，直到测试放行—— - /// 在启动健康门的 await 窗口内确定性插入交错命令（探测运行在 detached - /// 线程，不占主 actor）。放行后恢复直通。 - final class BlockingProbe: EndpointProbing, @unchecked Sendable { - private let lock = NSLock() - private var gate: TestBlockingGate? - private var started: XCTestExpectation? - - func arm(started: XCTestExpectation) { - lock.lock() - defer { lock.unlock() } - precondition(gate == nil, "上一次 arm 尚未 release") - gate = TestBlockingGate("runtime fixture release") - self.started = started - } - - func release() { - lock.lock() - let gate = self.gate - self.gate = nil - self.started = nil - lock.unlock() - gate?.release() - } - - func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome { - lock.lock() - let gate = self.gate - let started = self.started - lock.unlock() - if let gate { - started?.fulfill() - _ = gate.wait() - } - return .reachable - } - } - - /// 按调用序返回预设结果、并可在第 N 次调用处挂起的防火墙检查替身： - /// 钉住「轮询 await 恢复后代际已推进时不得再写状态」的回归。 - final class BlockingFirewallChecker: FirewallStatusChecking, @unchecked Sendable { - private let lock = NSLock() - private var callCount = 0 - private let outcomes: [FirewallBlockStatus] - private let blockOnCall: Int - private let started: XCTestExpectation - private let gate = TestBlockingGate("runtime fixture release") - - init(outcomes: [FirewallBlockStatus], blockOnCall: Int, started: XCTestExpectation) { - precondition(!outcomes.isEmpty) - self.outcomes = outcomes - self.blockOnCall = blockOnCall - self.started = started - } - - func release() { gate.release() } - - func status(for executableURL: URL) -> FirewallBlockStatus { - lock.lock() - callCount += 1 - let call = callCount - lock.unlock() - if call == blockOnCall { - started.fulfill() - _ = gate.wait() - } - return outcomes[min(call, outcomes.count) - 1] - } - } } /// 每个夹具使用独立 defaults 域，最后一个引用释放时清理持久值。 diff --git a/ShadowsocksX-NG2/Tests/Support/RuleRuntimeDelayGate.swift b/ShadowsocksX-NG2/Tests/Support/RuleRuntimeDelayGate.swift index db2a6778..e6400152 100644 --- a/ShadowsocksX-NG2/Tests/Support/RuleRuntimeDelayGate.swift +++ b/ShadowsocksX-NG2/Tests/Support/RuleRuntimeDelayGate.swift @@ -71,13 +71,13 @@ func withRuleApplicationTask( final class PausedRuntimeRuleSource: @unchecked Sendable { let started: XCTestExpectation - private let release = TestBlockingGate("rule source release") + private let release = TestAsyncGate("rule source release") init(started: XCTestExpectation) { self.started = started } func resume() { release.release() } - func load(_ source: RulesSource) -> RuleSnapshot { + func load(_ source: RulesSource) async -> RuleSnapshot { if source == .geolocationCN { started.fulfill() - _ = release.wait() + _ = await release.wait() } return rulesFixture(source) } diff --git a/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift b/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift index 06a57b94..fbd61c68 100644 --- a/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift +++ b/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift @@ -163,7 +163,11 @@ final class FakeOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { answerOverride = answer } - func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy { + func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy { + occupancySynchronously(for: request) + } + + private func occupancySynchronously(for request: PortOccupancyProbeRequest) -> PortOccupancy { lock.lock() defer { lock.unlock() } requestedPorts.append(request.port) @@ -199,8 +203,14 @@ final class FakeOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { /// 被扣住，`passThrough` 之后的调用立即返回新应答。用于把「旧一轮」结果钉在 /// 新一轮之后落盘，验证过期结果不得覆盖。 final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { + private struct Invocation { + let isFirstEntry: Bool + let isGated: Bool + let answer: PortOccupancy + } + private let lock = NSLock() - private let gate = TestBlockingGate("occupancy probe release") + private let gate = TestAsyncGate("occupancy probe release") private var isGated = true private var enteredCount = 0 private let entry = XCTestExpectation(description: "occupancy probe entered") @@ -239,17 +249,22 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { gate.release() } - func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy { + func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy { + let invocation = record(request) + if invocation.isFirstEntry { entry.fulfill() } + if invocation.isGated, !(await gate.wait()) { + return .unknown(detail: "Test fixture release timed out") + } + return invocation.answer + } + + private func record(_ request: PortOccupancyProbeRequest) -> Invocation { lock.lock() enteredCount += 1 let firstEntry = enteredCount == 1 let gated = isGated let answer = gated ? gatedAnswer : passThroughAnswer lock.unlock() - if firstEntry { entry.fulfill() } - if gated, !gate.wait() { - return .unknown(detail: "Test fixture release timed out") - } - return answer + return Invocation(isFirstEntry: firstEntry, isGated: gated, answer: answer) } } diff --git a/ShadowsocksX-NG2/Tests/TestSupport/BoundedTestWaitTests.swift b/ShadowsocksX-NG2/Tests/TestSupport/BoundedTestWaitTests.swift index c95df1d1..26a50eec 100644 --- a/ShadowsocksX-NG2/Tests/TestSupport/BoundedTestWaitTests.swift +++ b/ShadowsocksX-NG2/Tests/TestSupport/BoundedTestWaitTests.swift @@ -40,27 +40,26 @@ struct BoundedTestWaitTests { await #expect(throws: CancellationError.self) { try await task.value } } - @Test func blockingGateTimesOutInsteadOfRetainingAWorker() async { - let gate = TestBlockingGate(onTimeout: {}) - let worker = Task.detached { gate.wait(timeout: 0.02) } - #expect(await worker.value == false) + @Test func asyncGateTimesOutWithoutBlockingAWorker() async { + let gate = TestAsyncGate(onTimeout: {}) + #expect(await gate.wait(timeout: .milliseconds(20)) == false) } - @Test func blockingGateRemembersEarlyAndRepeatedRelease() { - let gate = TestBlockingGate() + @Test func asyncGateRemembersEarlyAndRepeatedRelease() async { + let gate = TestAsyncGate() gate.release() gate.release() - #expect(gate.wait(timeout: 0)) + #expect(await gate.wait(timeout: .zero)) } - @Test func blockingGateReleasesEveryWorker() async throws { - let gate = TestBlockingGate() + @Test func asyncGateReleasesEveryWorker() async throws { + let gate = TestAsyncGate() let entered = XCTestExpectation(description: "workers entered") entered.expectedFulfillmentCount = 4 let workers = (0..<4).map { _ in Task.detached { entered.fulfill() - return gate.wait() + return await gate.wait() } } try await requireTestPhase([entered])