Alawn/ShadowsocksX-NG
1
0
Fork 0
mirror of https://github.com/shadowsocks/ShadowsocksX-NG.git synced 2026-10-29 13:43:49 +00:00
Code Releases Activity

refactor: consolidate subscription update execution ownership

Browse Source
This commit is contained in:
Qiu Yuzhou
2026-10-25 22:30:19 +08:00
parent f3ba330cf4
commit 3a10b7524c
7 changed files with 208 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+AutomaticUpdates.swift
+55 -31
View File
@@ -1,67 +1,91 @@
import Foundation
extension CatalogWorkflow {
enum AutomaticUpdateState {
case stopped, idle
case waiting(id: UUID, task: Task<Void, Never>)
case updating(id: UUID, task: Task<Void, Never>)
var task: Task<Void, Never>? {
switch self {
case .stopped, .idle: nil
case .waiting(_, let task), .updating(_, let task): task
}
}
var id: UUID? {
switch self {
case .stopped, .idle: nil
case .waiting(let id, _), .updating(let id, _): id
}
}
}
/// Owned by the GUI process, independently of workspace presentation.
func startSubscriptionAutomaticUpdates() {
automaticUpdatesRunning = true
if case .stopped = automaticUpdateState { automaticUpdateState = .idle }
scheduleAutomaticUpdateCheck()
}
func stopSubscriptionAutomaticUpdates() {
automaticUpdatesRunning = false
automaticUpdateTimer?.cancel()
automaticUpdateTimer = nil
automaticUpdateBatch?.cancel()
let task = automaticUpdateState.task
automaticUpdateState = .stopped
task?.cancel()
for id in Array(subscriptionRequests.keys) { cancelSubscriptionRequest(id) }
}
/// Startup and wake use the persisted deadline directly, without catch-up jitter.
func subscriptionAutomaticUpdatesDidWake() {
guard automaticUpdatesRunning else { return }
scheduleAutomaticUpdateCheck()
}
func refreshDueSubscriptions() async {
guard !checkingDueSubscriptions else { return }
checkingDueSubscriptions = true
defer { checkingDueSubscriptions = false }
private func refreshDueSubscriptions() async {
var attempted: Set<NodeID> = []
while let due = subscriptions.first(where: {
!attempted.contains($0.id) && !refreshingSubscriptionIDs.contains($0.id)
&& $0.autoUpdate.effectiveMode != .off
&& ($0.autoUpdate.nextUpdateAt.map { $0 <= dependencies.subscriptionNow() } ?? false)
}) {
guard !Task.isCancelled else { return }
while !Task.isCancelled,
let due = subscriptions.first(where: {
!attempted.contains($0.id) && subscriptionRequests[$0.id] == nil
&& $0.autoUpdate.effectiveMode != .off
&& ($0.autoUpdate.nextUpdateAt.map { $0 <= dependencies.subscriptionNow() } ?? false)
})
{
attempted.insert(due.id)
await refreshSubscription(due.id)
}
}
func scheduleAutomaticUpdateCheck(includeOverdue: Bool = true) {
automaticUpdateTimer?.cancel()
automaticUpdateTimer = nil
guard automaticUpdatesRunning, automaticUpdateBatch == nil,
switch automaticUpdateState {
case .stopped, .updating: return
case .idle, .waiting:
automaticUpdateState.task?.cancel()
automaticUpdateState = .idle
}
let now = dependencies.subscriptionNow()
guard
let deadline = subscriptions.filter({
$0.autoUpdate.effectiveMode != .off && !refreshingSubscriptionIDs.contains($0.id)
$0.autoUpdate.effectiveMode != .off && subscriptionRequests[$0.id] == nil
}).compactMap(\.autoUpdate.nextUpdateAt).filter({
includeOverdue || $0 > dependencies.subscriptionNow()
includeOverdue || $0 > now
}).min()
else { return }
let delay = max(0, deadline.timeIntervalSince(dependencies.subscriptionNow()))
let delay = max(0, deadline.timeIntervalSince(now))
let wait = dependencies.subscriptionWait
automaticUpdateTimer = Task { [weak self] in
let id = UUID()
let task = Task { [weak self] in
do {
try await wait(delay)
try Task.checkCancellation()
} catch { return }
guard let self, self.automaticUpdatesRunning else { return }
self.automaticUpdateBatch = Task { [weak self] in
guard let self else { return }
await self.refreshDueSubscriptions()
self.automaticUpdateBatch = nil
// An unchanged overdue plan (for example a failed save) is not a retry loop.
self.scheduleAutomaticUpdateCheck(includeOverdue: false)
}
guard let self, case .waiting(let currentID, let task) = self.automaticUpdateState,
currentID == id
else { return }
self.automaticUpdateState = .updating(id: id, task: task)
await self.refreshDueSubscriptions()
guard self.automaticUpdateState.id == id, !Task.isCancelled else { return }
self.automaticUpdateState = .idle
// An unchanged overdue plan (for example a failed save) is not a retry loop.
self.scheduleAutomaticUpdateCheck(includeOverdue: false)
}
automaticUpdateState = .waiting(id: id, task: task)
}
}
ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+Subscriptions.swift
-3
View File
@@ -45,7 +45,6 @@ extension CatalogWorkflow {
else { return }
let token = UUID()
setSubscriptionRefreshCompletion(nil, for: id)
setRefreshInFlight(id, true)
let request = Task { await runSubscriptionRefresh(source, token: token) }
subscriptionRequests[id] = (token, request)
await withTaskCancellationHandler {
@@ -60,7 +59,6 @@ extension CatalogWorkflow {
defer {
if subscriptionRequests[id]?.token == token {
subscriptionRequests[id] = nil
setRefreshInFlight(id, false)
scheduleAutomaticUpdateCheck()
}
}
@@ -115,7 +113,6 @@ extension CatalogWorkflow {
func cancelSubscriptionRequest(_ id: NodeID) {
subscriptionRequests.removeValue(forKey: id)?.task.cancel()
setRefreshInFlight(id, false)
}
/// 删除订阅：递归清除订阅源、固定分组与远端成员并清理凭据；
ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow.swift
+5 -15
View File
@@ -24,7 +24,7 @@ final class CatalogWorkflow: ObservableObject {
@Published private(set) var favoriteIDs: [NodeID]
/// 订阅卡片 projection（非敏感）。
@Published private(set) var subscriptions: [SubscriptionSummary] = []
/// 正在刷新的订阅身份（并发守卫的只读观察面）。
/// 请求登记的只读展示投影；刷新互斥仅由请求表判定。
@Published private(set) var refreshingSubscriptionIDs: Set<NodeID> = []
/// 当前会话的刷新结果；不从持久化状态重建，供各入口共用结束时间。
@Published private(set) var subscriptionRefreshCompletions:
@@ -55,11 +55,10 @@ final class CatalogWorkflow: ObservableObject {
let dependencies: CatalogWorkflowDependencies
/// 启动时发现的 Legacy 快照（导入编排用；秘密值不进 projection）。
var discoveredLegacySnapshot: LegacySnapshot?
var subscriptionRequests: [NodeID: (token: UUID, task: Task<Void, Never>)] = [:]
var automaticUpdatesRunning = false
var automaticUpdateTimer: Task<Void, Never>?
var automaticUpdateBatch: Task<Void, Never>?
var checkingDueSubscriptions = false
var subscriptionRequests: [NodeID: (token: UUID, task: Task<Void, Never>)] = [:] {
didSet { refreshingSubscriptionIDs = Set(subscriptionRequests.keys) }
}
var automaticUpdateState = AutomaticUpdateState.stopped
/// 组合根注入依赖束；workflow 不自行创建任何生产 adapter。
init(dependencies: CatalogWorkflowDependencies) {
@@ -279,15 +278,6 @@ final class CatalogWorkflow: ObservableObject {
legacyImportState = state
}
/// 订阅刷新并发守卫的写入口（集合对外只读）。
func setRefreshInFlight(_ id: NodeID, _ inFlight: Bool) {
if inFlight {
refreshingSubscriptionIDs.insert(id)
} else {
refreshingSubscriptionIDs.remove(id)
}
}
func setSubscriptionRefreshCompletion(
_ completion: SubscriptionRefreshCompletion?, for id: NodeID
) {
ShadowsocksX-NG2/Tests/Architecture/SkeletonTests.swift
+1 -1
View File
@@ -76,7 +76,7 @@ final class CatalogWorkflowArchitectureTests: XCTestCase {
let memberTokens = [
"\\.dependencies\\b", "\\bfileStore\\b", "discoveredLegacySnapshot",
"republishCommittedState", "commitSubscriptionDocument", "publishLegacyImportState",
"setRefreshInFlight", "setSubscriptionRefreshFailure",
"automaticUpdateState", "subscriptionRequests", "setSubscriptionRefreshFailure",
"subscriptionSummaries", "credentialRefs",
]
return (typeTokens + memberTokens).map { token in
ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Lifecycle.swift
+116
View File
@@ -0,0 +1,116 @@
import Foundation
import Testing
@testable import ShadowsocksX_NG2
extension SubscriptionAutoUpdateWorkflowTests {
@Test(arguments: [true, false])
func failedSaveWaitsForWakeOrCommittedChangesBeforeRetry(wake: Bool) async throws {
try await withAutoUpdateFixture { fixture in
let first = try await fixture.create(settings: .init(userOverride: .daily))
fixture.now = fixture.now.addingTimeInterval(1_000)
let second = try await fixture.create(settings: .init(userOverride: .daily))
fixture.now = fixture.now.addingTimeInterval(85_400)
let scheduled = TestAsyncGate("future plan scheduled after failed save")
let workflow = fixture.makeWorkflow(wait: { delay in
if delay > 0 {
scheduled.release()
try await Task.sleep(for: .seconds(delay))
}
})
defer { workflow.stopSubscriptionAutomaticUpdates() }
let before = workflow.subscriptions
let requests = fixture.fetcher.requestCount
try FileManager.default.removeItem(at: fixture.directory)
try Data().write(to: fixture.directory)
workflow.startSubscriptionAutomaticUpdates()
try #require(await scheduled.wait())
#expect(workflow.subscriptions == before)
#expect(workflow.subscriptionRefreshFailure(for: first.id) == .commit)
#expect(fixture.fetcher.requestCount == requests + 1)
#expect(workflow.refreshingSubscriptionIDs.isEmpty)
try FileManager.default.removeItem(at: fixture.directory)
try FileManager.default.createDirectory(
at: fixture.directory, withIntermediateDirectories: true)
if wake {
workflow.subscriptionAutomaticUpdatesDidWake()
} else {
try await workflow.saveSubscription(second.id, urlString: "https://example.com/sub")
}
try await requireTestCondition("overdue plan succeeds after external event") {
workflow.subscriptions.first?.autoUpdate.nextUpdateAt
== fixture.now.addingTimeInterval(86_400)
&& workflow.refreshingSubscriptionIDs.isEmpty
}
#expect(fixture.fetcher.requestCount == requests + 2)
#expect(workflow.subscriptionRefreshFailure(for: first.id) == nil)
}
}
@Test func stoppingCancelsManualRequestsAndPreservesTheirCommittedPlan() async throws {
try await withAutoUpdateFixture { fixture in
let first = try await fixture.create(settings: .init(userOverride: .daily))
let fetcher = GatedFetcher()
let waiting = TestAsyncGate("automatic update waiting")
var waitCount = 0
let workflow = fixture.makeWorkflow(
fetcher: fetcher,
wait: { delay in
waitCount += 1
waiting.release()
try await Task.sleep(for: .seconds(delay))
})
defer { workflow.stopSubscriptionAutomaticUpdates() }
workflow.startSubscriptionAutomaticUpdates()
try #require(await waiting.wait())
let before = workflow.subscriptions
let request = Task { await workflow.refreshSubscription(first.id) }
try await withTestTask(request, release: fetcher.releaseAll) {
try await requireTestCondition("manual refresh entered") { fetcher.didEnterFetch }
#expect(workflow.refreshingSubscriptionIDs == [first.id])
workflow.stopSubscriptionAutomaticUpdates()
workflow.stopSubscriptionAutomaticUpdates()
workflow.subscriptionAutomaticUpdatesDidWake()
#expect(workflow.refreshingSubscriptionIDs.isEmpty)
}
#expect(workflow.subscriptions == before)
#expect(workflow.subscriptionRefreshCompletions.isEmpty)
#expect(workflow.subscriptionRefreshFailure(for: first.id) == nil)
#expect(waitCount == 1)
}
}
@Test func stoppedBatchDoesNotReadTheClockAfterItsRequestReturns() async throws {
try await withAutoUpdateFixture { fixture in
_ = try await fixture.create(settings: .init(userOverride: .daily))
fixture.now = fixture.now.addingTimeInterval(1_000)
_ = try await fixture.create(settings: .init(userOverride: .daily))
fixture.now = fixture.now.addingTimeInterval(85_400)
let fetcher = GatedFetcher()
var clockReads = 0
var workflow: CatalogWorkflow? = makeCatalogWorkflow(
coordinator: fixture.coordinator, subscriptionFetcher: fetcher,
subscriptionNow: {
clockReads += 1
return fixture.now
})
weak var releasedWorkflow = workflow
defer {
workflow?.stopSubscriptionAutomaticUpdates()
fetcher.releaseAll()
}
workflow?.startSubscriptionAutomaticUpdates()
try await requireTestCondition("automatic request entered") { fetcher.didEnterFetch }
let readsBeforeStop = clockReads
workflow?.stopSubscriptionAutomaticUpdates()
workflow = nil
fetcher.releaseAll()
try await requireTestCondition("stopped batch released its workflow") {
releasedWorkflow == nil
}
#expect(clockReads == readsBeforeStop)
}
}
}
ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Races.swift
+23 -11
View File
@@ -80,12 +80,18 @@ extension SubscriptionAutoUpdateWorkflowTests {
fixture.now = fixture.now.addingTimeInterval(100_000)
let fetcher = GatedFetcher()
let workflow = fixture.makeWorkflow(fetcher: fetcher)
let batch = Task { await workflow.refreshDueSubscriptions() }
try await withTestTask(batch, release: fetcher.releaseAll) {
try await requireTestCondition("batch fetch entered") { fetcher.didEnterFetch }
#expect(fetcher.requestCount == 1)
await workflow.refreshSubscription(first.id)
#expect(fetcher.requestCount == 1)
defer {
workflow.stopSubscriptionAutomaticUpdates()
fetcher.releaseAll()
}
workflow.startSubscriptionAutomaticUpdates()
try await requireTestCondition("batch fetch entered") { fetcher.didEnterFetch }
#expect(fetcher.requestCount == 1)
await workflow.refreshSubscription(first.id)
#expect(fetcher.requestCount == 1)
fetcher.releaseAll()
try await requireTestCondition("automatic batch completed") {
fetcher.requestCount == 2 && workflow.refreshingSubscriptionIDs.isEmpty
}
#expect(fetcher.requestCount == 2)
}
@@ -99,11 +105,17 @@ extension SubscriptionAutoUpdateWorkflowTests {
fixture.now = fixture.now.addingTimeInterval(85_400)
let fetcher = GatedFetcher()
let workflow = fixture.makeWorkflow(fetcher: fetcher)
let batch = Task { await workflow.refreshDueSubscriptions() }
try await withTestTask(batch, release: fetcher.releaseAll) {
try await requireTestCondition("first due fetch entered") { fetcher.didEnterFetch }
#expect(fetcher.requestCount == 1)
fixture.now = fixture.now.addingTimeInterval(1_000)
defer {
workflow.stopSubscriptionAutomaticUpdates()
fetcher.releaseAll()
}
workflow.startSubscriptionAutomaticUpdates()
try await requireTestCondition("first due fetch entered") { fetcher.didEnterFetch }
#expect(fetcher.requestCount == 1)
fixture.now = fixture.now.addingTimeInterval(1_000)
fetcher.releaseAll()
try await requireTestCondition("automatic batch completed") {
fetcher.requestCount == 2 && workflow.refreshingSubscriptionIDs.isEmpty
}
#expect(fetcher.requestCount == 2)
}
ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests.swift
+8 -5
View File
@@ -137,13 +137,16 @@ struct SubscriptionAutoUpdateWorkflowTests {
let first = try await fixture.create(settings: .init(userOverride: .daily))
let second = try await fixture.create(settings: .init(userOverride: .daily))
let requests = fixture.fetcher.requestCount
await fixture.workflow.refreshDueSubscriptions()
fixture.workflow.startSubscriptionAutomaticUpdates()
#expect(fixture.fetcher.requestCount == requests)
fixture.now = fixture.now.addingTimeInterval(1_000_000)
fixture.fetcher.setBehavior(.failure(.httpStatus(code: 503)))
await fixture.workflow.refreshDueSubscriptions()
#expect(fixture.fetcher.requestCount == requests + 2)
await fixture.workflow.refreshDueSubscriptions()
fixture.workflow.subscriptionAutomaticUpdatesDidWake()
try await requireTestCondition("all overdue updates completed") {
fixture.fetcher.requestCount == requests + 2
&& fixture.workflow.refreshingSubscriptionIDs.isEmpty
}
fixture.workflow.subscriptionAutomaticUpdatesDidWake()
#expect(fixture.fetcher.requestCount == requests + 2)
#expect(fixture.workflow.subscriptions.map(\.id) == [first.id, second.id])
#expect(
@@ -167,7 +170,7 @@ struct SubscriptionAutoUpdateWorkflowTests {
let requests = fixture.fetcher.requestCount
let reload = fixture.makeWorkflow()
reload.startSubscriptionAutomaticUpdates()
await reload.refreshDueSubscriptions()
reload.subscriptionAutomaticUpdatesDidWake()
reload.stopSubscriptionAutomaticUpdates()
#expect(reload.subscriptions.first?.id == first.id)
#expect(fixture.fetcher.requestCount == requests)