diff --git a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityDoHResolver.swift b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityDoHResolver.swift index ffe95664..a870d955 100644 --- a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityDoHResolver.swift +++ b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityDoHResolver.swift @@ -90,12 +90,7 @@ struct ConnectivityDoHResolver: Sendable { return nil } defer { group.cancelAll() } - var found: [String] = [] - for _ in 0..<2 { - guard let next = await group.next(), let family = next else { break } - found += family - } - return found + return await Self.collectAddresses(from: group) } guard !addresses.isEmpty else { throw Failure.invalidResponse } // Stable family order makes equal-time candidate ties reproducible. @@ -103,6 +98,21 @@ struct ConnectivityDoHResolver: Sendable { + addresses.filter { ConnectivityAddress.family($0) == AF_INET6 } } + /// A nil event is the shared deadline; keep families received before it. + static func collectAddresses(from results: Results) async rethrows + -> [String] where Results.Element == [String]? + { + var found: [String] = [] + var families = 0 + for try await family in results { + guard let family else { break } + found += family + families += 1 + if families == 2 { break } + } + return found + } + private func answers(host: String, endpoint: String, type: Int, session: URLSession) async -> [String] { diff --git a/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityDoHTests.swift b/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityDoHTests.swift index 6491986a..bdd17007 100644 --- a/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityDoHTests.swift +++ b/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityDoHTests.swift @@ -3,6 +3,7 @@ import Testing @testable import ShadowsocksX_NG2 +@Suite(.timeLimit(.minutes(1))) struct ConnectivityDoHTests { @Test func fallsBackToCloudflareAndAcceptsOnlyMatchingIPRecords() async { let resolver = ConnectivityDoHResolver(sessionConfiguration: { @@ -18,17 +19,42 @@ struct ConnectivityDoHTests { #expect(!result.trace.map(\.description).joined().contains("target.invalid")) } - @Test func familyThatExceedsDeadlineDoesNotDiscardOtherFamily() async throws { + @Test func familyThatExceedsDeadlineDoesNotDiscardOtherFamily() async { + let events = AsyncStream<[String]?> { continuation in + continuation.yield(["192.0.2.1"]) + continuation.yield(nil) + continuation.yield(["2001:db8::1"]) + continuation.finish() + } + let addresses = await ConnectivityDoHResolver.collectAddresses(from: events) + #expect(addresses == ["192.0.2.1"]) + } + + @Test func completedFamiliesDoNotWaitForDeadline() async { + let events = AsyncStream<[String]?> { continuation in + continuation.yield(["192.0.2.1"]) + continuation.yield(["2001:db8::1"]) + } + let addresses = await ConnectivityDoHResolver.collectAddresses(from: events) + #expect(addresses == ["192.0.2.1", "2001:db8::1"]) + } + + @Test func pendingFamiliesAreCancelledAtDeadline() async throws { let resolver = ConnectivityDoHResolver(sessionConfiguration: { let configuration = URLSessionConfiguration.ephemeral - configuration.protocolClasses = [SlowIPv6Protocol.self] + configuration.protocolClasses = [PendingAnswersProtocol.self] return configuration }) let started = ContinuousClock.now - let addresses = try await resolver.query( - host: "target.invalid", endpoint: "https://resolver.invalid/resolve", proxyPort: 1234, - deadline: .now.advanced(by: .milliseconds(100))) - #expect(addresses == ["192.0.2.1"]) + do { + _ = try await resolver.query( + host: "target.invalid", endpoint: "https://resolver.invalid/resolve", proxyPort: 1234, + deadline: .now.advanced(by: .milliseconds(100))) + Issue.record("Pending requests unexpectedly resolved") + } catch { + #expect(error is ConnectivityDoHResolver.Failure) + } + #expect(ContinuousClock.now - started >= .milliseconds(100)) #expect(ContinuousClock.now - started < .seconds(1)) } } @@ -57,11 +83,6 @@ private class DNSAnswersProtocol: URLProtocol, @unchecked Sendable { override func stopLoading() {} } -private final class SlowIPv6Protocol: DNSAnswersProtocol, @unchecked Sendable { - override func startLoading() { - let type = request.url.flatMap { URLComponents(url: $0, resolvingAgainstBaseURL: false) }? - .queryItems?.first(where: { $0.name == "type" })?.value - if type == "28" { return } - super.startLoading() - } +private final class PendingAnswersProtocol: DNSAnswersProtocol, @unchecked Sendable { + override func startLoading() {} } diff --git a/ShadowsocksX-NG2/Tests/Settings/LANListeningWorkflowTests.swift b/ShadowsocksX-NG2/Tests/Settings/LANListeningWorkflowTests.swift index faa8bf8d..162d7736 100644 --- a/ShadowsocksX-NG2/Tests/Settings/LANListeningWorkflowTests.swift +++ b/ShadowsocksX-NG2/Tests/Settings/LANListeningWorkflowTests.swift @@ -3,6 +3,7 @@ import Testing @testable import ShadowsocksX_NG2 +@Suite(.timeLimit(.minutes(1))) @MainActor struct LANListeningWorkflowTests { @Test func disabledAuthenticationDoesNotGateLANSaveOnCredentialShape() async throws { @@ -16,7 +17,7 @@ struct LANListeningWorkflowTests { draft.username = "" draft.password = "" workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(workflow.canSaveLANListening(draft)) #expect(await workflow.saveLANListening(draft) == .persisted) } @@ -31,7 +32,7 @@ struct LANListeningWorkflowTests { draft.username = "user:name" draft.password = "ok" workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(!workflow.canSaveLANListening(draft)) } @@ -46,7 +47,7 @@ struct LANListeningWorkflowTests { var draft = workflow.beginPortSettingsEditing() draft.socksPort = 13001 workflow.refreshPortEditorOccupancy(for: draft) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft, port: .socks) #expect(workflow.canSavePortSettings(draft)) #expect(await workflow.savePortSettings(draft) == .persisted) #expect(committing.committedSettings.listen.lan.username == "") @@ -60,7 +61,7 @@ struct LANListeningWorkflowTests { var draft = try workflow.beginLANListeningEditing() draft.isEnabled = true workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(!workflow.canSaveLANListening(draft)) draft.isEnabled = false @@ -73,7 +74,7 @@ struct LANListeningWorkflowTests { #expect(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy == nil) #expect(!workflow.canSaveLANListening(draft)) workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(probe.callCount == count + 2) #expect(!workflow.canSaveLANListening(draft)) } @@ -139,7 +140,7 @@ struct LANListeningWorkflowTests { var draft = workflow.beginPortSettingsEditing() draft.socksPort = 13001 workflow.refreshPortEditorOccupancy(for: draft) - await waitUntil(workflow.portFieldState(for: .http, editorDraft: draft).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft, port: .http) #expect(await workflow.savePortSettings(draft) == .persisted) #expect(committing.committedSettings.listen.socksPort == 13001) } @@ -174,9 +175,10 @@ struct LANListeningWorkflowTests { draft.isEnabled = true draft.mode = .allIPv4Interfaces let submission = Task { await workflow.saveLANListening(draft) } - await waitUntil(probe.entered > 0) + await probe.waitUntilEntered() #expect(workflow.isCommitting) #expect(await workflow.saveProxyExceptions("example.com") == .rejected(.inProgress)) + probe.passThrough(.free) probe.releaseGatedCalls() #expect(await submission.value == .persisted) #expect(!workflow.isCommitting) @@ -195,7 +197,7 @@ struct LANListeningWorkflowTests { draft.mode = mode draft.ports.socksPort = 13001 workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .http, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .http) var expected = RuntimeListenFacts(listen: committing.committedSettings.listen) expected.lan.isEnabled = true @@ -225,7 +227,7 @@ struct LANListeningWorkflowTests { draft.isEnabled = true draft.mode = .allIPv6Interfaces workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(workflow.portFieldState(for: .socks, editorDraft: draft.ports).canSuggestFreePort) #expect( await workflow.suggestFreePort(for: .socks, from: draft.ports) @@ -252,7 +254,7 @@ struct LANListeningWorkflowTests { #expect(workflow.portFieldState(for: .socks, editorDraft: draft).issues.count == 1) #expect(workflow.portFieldState(for: .http, editorDraft: draft).issues.isEmpty) workflow.refreshPortEditorOccupancy(for: draft) - await waitUntil(workflow.portFieldState(for: .http, editorDraft: draft).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft, port: .http) #expect(!workflow.canSavePortSettings(draft)) } @@ -270,7 +272,7 @@ struct LANListeningWorkflowTests { draft.isEnabled = true draft.mode = .allIPv4Interfaces workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(workflow.portFieldState(for: .socks, editorDraft: draft.ports).isRuntimePortException) #expect(await workflow.saveLANListening(draft) == .persisted) } @@ -287,7 +289,7 @@ struct LANListeningWorkflowTests { var draft = try workflow.beginLANListeningEditing() draft.ports.socksPort = 13001 workflow.refreshPortEditorOccupancy(for: draft.ports) - await waitUntil(workflow.portFieldState(for: .socks, editorDraft: draft.ports).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft.ports, port: .socks) #expect(!workflow.portFieldState(for: .socks, editorDraft: draft.ports).isRuntimePortException) #expect(await workflow.saveLANListening(draft) == .rejected(.occupied([.socks]))) } @@ -298,7 +300,7 @@ struct LANListeningWorkflowTests { let workflow = SettingsWorkflow(committing: committing, occupancyProbe: FakeOccupancyProbe()) let draft = workflow.beginPortSettingsEditing() workflow.refreshPortEditorOccupancy(for: draft) - await waitUntil(workflow.portFieldState(for: .http, editorDraft: draft).occupancy != nil) + await waitForPortEditorOccupancy(workflow, draft: draft, port: .http) #expect(workflow.portFieldState(for: .socks, editorDraft: draft).issues.isEmpty) #expect(workflow.portFieldState(for: .http, editorDraft: draft).issues.isEmpty) #expect(!workflow.canSavePortSettings(draft)) diff --git a/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift b/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift index 11cdbe7e..83827b60 100644 --- a/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift +++ b/ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift @@ -1,3 +1,4 @@ +import Combine import Foundation import XCTest @@ -18,6 +19,25 @@ func waitUntil( XCTAssertTrue(condition(), file: file, line: line) } +/// 等待工作流发布探测结果;发布前订阅,避免依赖轮询与线程调度耗时。 +/// 调用方用测试框架的 timeLimit 保护永不发布的回归。 +@MainActor +func waitForPortEditorOccupancy( + _ workflow: SettingsWorkflow, draft: SettingsPortDraft, port: SettingsPortID +) async { + let (events, continuation) = AsyncStream.makeStream(bufferingPolicy: .bufferingNewest(1)) + let observation = workflow.objectWillChange.sink { continuation.yield(()) } + defer { + observation.cancel() + continuation.finish() + } + continuation.yield(()) + for await _ in events + where workflow.portFieldState(for: port, editorDraft: draft).occupancy != nil { + return + } +} + /// 设置工作流测试的写入缝替身:记录提交调用,可注入失败; /// 不触碰真实偏好文件、钥匙串或运行时。 @MainActor @@ -174,6 +194,8 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { private let gate = DispatchGroup() private var isGated = true private var enteredCount = 0 + private let entry = AsyncGate() + private var released = false private var gatedAnswer: PortOccupancy private var passThroughAnswer: PortOccupancy @@ -190,6 +212,10 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { return enteredCount } + func waitUntilEntered() async { + await entry.wait() + } + /// 之后的调用立即返回 `answer`;已进入且被扣住的调用仍持有各自应答。 func passThrough(_ answer: PortOccupancy) { lock.lock() @@ -201,10 +227,11 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { /// 放行全部被扣住的调用(广播门闩)。 func releaseGatedCalls() { lock.lock() - let wasGated = isGated + let shouldRelease = !released + released = true isGated = false lock.unlock() - if wasGated { + if shouldRelease { gate.leave() } } @@ -215,6 +242,7 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable { let gated = isGated let answer = gated ? gatedAnswer : passThroughAnswer lock.unlock() + entry.release() if gated { gate.wait() }