From 3a10b7524c50d2b33cb5b62ffaeb73ca3fdf0a99 Mon Sep 17 00:00:00 2001 From: Qiu Yuzhou Date: Sun, 25 Oct 2026 22:30:19 +0800 Subject: [PATCH] refactor: consolidate subscription update execution ownership --- .../CatalogWorkflow+AutomaticUpdates.swift | 86 ++++++++----- .../CatalogWorkflow+Subscriptions.swift | 3 - .../CatalogWorkflow/CatalogWorkflow.swift | 20 +-- .../Tests/Architecture/SkeletonTests.swift | 2 +- ...ionAutoUpdateWorkflowTests+Lifecycle.swift | 116 ++++++++++++++++++ ...riptionAutoUpdateWorkflowTests+Races.swift | 34 +++-- .../SubscriptionAutoUpdateWorkflowTests.swift | 13 +- 7 files changed, 208 insertions(+), 66 deletions(-) create mode 100644 ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Lifecycle.swift diff --git a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+AutomaticUpdates.swift b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+AutomaticUpdates.swift index 5c9b97d4..59ec5561 100644 --- a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+AutomaticUpdates.swift +++ b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+AutomaticUpdates.swift @@ -1,67 +1,91 @@ import Foundation extension CatalogWorkflow { + enum AutomaticUpdateState { + case stopped, idle + case waiting(id: UUID, task: Task) + case updating(id: UUID, task: Task) + + var task: Task? { + 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 = [] - 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) } } diff --git a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+Subscriptions.swift b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+Subscriptions.swift index 9e8ef0c5..6bd3012a 100644 --- a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+Subscriptions.swift +++ b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow+Subscriptions.swift @@ -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) } /// 删除订阅：递归清除订阅源、固定分组与远端成员并清理凭据； diff --git a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow.swift b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow.swift index 45ec7abe..b9d6269e 100644 --- a/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow.swift +++ b/ShadowsocksX-NG2/App/Application/CatalogWorkflow/CatalogWorkflow.swift @@ -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 = [] /// 当前会话的刷新结果；不从持久化状态重建，供各入口共用结束时间。 @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)] = [:] - var automaticUpdatesRunning = false - var automaticUpdateTimer: Task? - var automaticUpdateBatch: Task? - var checkingDueSubscriptions = false + var subscriptionRequests: [NodeID: (token: UUID, task: Task)] = [:] { + 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 ) { diff --git a/ShadowsocksX-NG2/Tests/Architecture/SkeletonTests.swift b/ShadowsocksX-NG2/Tests/Architecture/SkeletonTests.swift index 1f07dec4..9ba2ef8b 100644 --- a/ShadowsocksX-NG2/Tests/Architecture/SkeletonTests.swift +++ b/ShadowsocksX-NG2/Tests/Architecture/SkeletonTests.swift @@ -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 diff --git a/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Lifecycle.swift b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Lifecycle.swift new file mode 100644 index 00000000..3b390066 --- /dev/null +++ b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Lifecycle.swift @@ -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) + } + } + +} diff --git a/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Races.swift b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Races.swift index 0464a96e..f7ac1f0f 100644 --- a/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Races.swift +++ b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests+Races.swift @@ -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) } diff --git a/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests.swift b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests.swift index 47d974a7..f1d49fbd 100644 --- a/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests.swift +++ b/ShadowsocksX-NG2/Tests/Subscription/SubscriptionAutoUpdateWorkflowTests.swift @@ -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)