test: stabilize parallel readiness and deadline checks

Wait for settings publications instead of a wall-clock polling budget. Separate DoH result aggregation from the real deadline check, and release gated probes after pass-through is enabled.

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<Results: AsyncSequence>(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]
   {

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

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

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<Void>.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()
     }