diff --git a/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.pbxproj b/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.pbxproj index 9513626..7277d50 100644 --- a/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.pbxproj +++ b/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.pbxproj @@ -7,7 +7,7 @@ objects = { /* Begin PBXBuildFile section */ - A139516F2FAC867A00859397 /* EffectView in Frameworks */ = {isa = PBXBuildFile; productRef = A139516E2FAC867A00859397 /* EffectView */; }; + A1C326A62FD21F7C00393BD8 /* EffectComponents in Frameworks */ = {isa = PBXBuildFile; productRef = A1C326A52FD21F7C00393BD8 /* EffectComponents */; }; /* End PBXBuildFile section */ /* Begin PBXContainerItemProxy section */ @@ -46,7 +46,7 @@ isa = PBXFrameworksBuildPhase; buildActionMask = 2147483647; files = ( - A139516F2FAC867A00859397 /* EffectView in Frameworks */, + A1C326A62FD21F7C00393BD8 /* EffectComponents in Frameworks */, ); runOnlyForDeploymentPostprocessing = 0; }; @@ -113,7 +113,7 @@ ); name = EffectViewExample; packageProductDependencies = ( - A139516E2FAC867A00859397 /* EffectView */, + A1C326A52FD21F7C00393BD8 /* EffectComponents */, ); productName = SimpleEffectView; productReference = A13E260F2FA6014E00C324E5 /* EffectViewExample.app */; @@ -192,7 +192,7 @@ mainGroup = A13E26062FA6014E00C324E5; minimizedProjectReferenceProxies = 1; packageReferences = ( - A139516D2FAC867A00859397 /* XCLocalSwiftPackageReference "../../../EffectView" */, + A1C326A42FD21F7C00393BD8 /* XCLocalSwiftPackageReference "../../../EffectComponents" */, ); preferredProjectObjectVersion = 77; productRefGroup = A13E26102FA6014E00C324E5 /* Products */; @@ -321,7 +321,7 @@ GCC_WARN_UNINITIALIZED_AUTOS = YES_AGGRESSIVE; GCC_WARN_UNUSED_FUNCTION = YES; GCC_WARN_UNUSED_VARIABLE = YES; - IPHONEOS_DEPLOYMENT_TARGET = 26.4; + IPHONEOS_DEPLOYMENT_TARGET = 18.6; LOCALIZATION_PREFERS_STRING_CATALOGS = YES; MTL_ENABLE_DEBUG_INFO = INCLUDE_SOURCE; MTL_FAST_MATH = YES; @@ -381,7 +381,7 @@ GCC_WARN_UNINITIALIZED_AUTOS = YES_AGGRESSIVE; GCC_WARN_UNUSED_FUNCTION = YES; GCC_WARN_UNUSED_VARIABLE = YES; - IPHONEOS_DEPLOYMENT_TARGET = 26.4; + IPHONEOS_DEPLOYMENT_TARGET = 18.6; LOCALIZATION_PREFERS_STRING_CATALOGS = YES; MTL_ENABLE_DEBUG_INFO = NO; MTL_FAST_MATH = YES; @@ -408,7 +408,7 @@ INFOPLIST_KEY_UILaunchScreen_Generation = YES; INFOPLIST_KEY_UISupportedInterfaceOrientations_iPad = "UIInterfaceOrientationPortrait UIInterfaceOrientationPortraitUpsideDown UIInterfaceOrientationLandscapeLeft UIInterfaceOrientationLandscapeRight"; INFOPLIST_KEY_UISupportedInterfaceOrientations_iPhone = "UIInterfaceOrientationPortrait UIInterfaceOrientationLandscapeLeft UIInterfaceOrientationLandscapeRight"; - IPHONEOS_DEPLOYMENT_TARGET = 15.6; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; LD_RUNPATH_SEARCH_PATHS = ( "$(inherited)", "@executable_path/Frameworks", @@ -440,7 +440,7 @@ INFOPLIST_KEY_UILaunchScreen_Generation = YES; INFOPLIST_KEY_UISupportedInterfaceOrientations_iPad = "UIInterfaceOrientationPortrait UIInterfaceOrientationPortraitUpsideDown UIInterfaceOrientationLandscapeLeft UIInterfaceOrientationLandscapeRight"; INFOPLIST_KEY_UISupportedInterfaceOrientations_iPhone = "UIInterfaceOrientationPortrait UIInterfaceOrientationLandscapeLeft UIInterfaceOrientationLandscapeRight"; - IPHONEOS_DEPLOYMENT_TARGET = 15.6; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; LD_RUNPATH_SEARCH_PATHS = ( "$(inherited)", "@executable_path/Frameworks", @@ -465,7 +465,7 @@ CURRENT_PROJECT_VERSION = 1; DEVELOPMENT_TEAM = JQD365KZ66; GENERATE_INFOPLIST_FILE = YES; - IPHONEOS_DEPLOYMENT_TARGET = 26.4; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; MARKETING_VERSION = 1.0; PRODUCT_BUNDLE_IDENTIFIER = com.andreas.grosam.home.SimpleEffectViewTests; PRODUCT_NAME = "$(TARGET_NAME)"; @@ -486,7 +486,7 @@ CURRENT_PROJECT_VERSION = 1; DEVELOPMENT_TEAM = JQD365KZ66; GENERATE_INFOPLIST_FILE = YES; - IPHONEOS_DEPLOYMENT_TARGET = 26.4; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; MARKETING_VERSION = 1.0; PRODUCT_BUNDLE_IDENTIFIER = com.andreas.grosam.home.SimpleEffectViewTests; PRODUCT_NAME = "$(TARGET_NAME)"; @@ -506,6 +506,7 @@ CURRENT_PROJECT_VERSION = 1; DEVELOPMENT_TEAM = JQD365KZ66; GENERATE_INFOPLIST_FILE = YES; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; MARKETING_VERSION = 1.0; PRODUCT_BUNDLE_IDENTIFIER = com.andreas.grosam.home.SimpleEffectViewUITests; PRODUCT_NAME = "$(TARGET_NAME)"; @@ -525,6 +526,7 @@ CURRENT_PROJECT_VERSION = 1; DEVELOPMENT_TEAM = JQD365KZ66; GENERATE_INFOPLIST_FILE = YES; + IPHONEOS_DEPLOYMENT_TARGET = 26.0; MARKETING_VERSION = 1.0; PRODUCT_BUNDLE_IDENTIFIER = com.andreas.grosam.home.SimpleEffectViewUITests; PRODUCT_NAME = "$(TARGET_NAME)"; @@ -579,16 +581,16 @@ /* End XCConfigurationList section */ /* Begin XCLocalSwiftPackageReference section */ - A139516D2FAC867A00859397 /* XCLocalSwiftPackageReference "../../../EffectView" */ = { + A1C326A42FD21F7C00393BD8 /* XCLocalSwiftPackageReference "../../../EffectComponents" */ = { isa = XCLocalSwiftPackageReference; - relativePath = ../../../EffectView; + relativePath = ../../../EffectComponents; }; /* End XCLocalSwiftPackageReference section */ /* Begin XCSwiftPackageProductDependency section */ - A139516E2FAC867A00859397 /* EffectView */ = { + A1C326A52FD21F7C00393BD8 /* EffectComponents */ = { isa = XCSwiftPackageProductDependency; - productName = EffectView; + productName = EffectComponents; }; /* End XCSwiftPackageProductDependency section */ }; diff --git a/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved b/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved index aceb888..3bd69f8 100644 --- a/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved +++ b/Examples/EffectViewExample/EffectViewExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved @@ -1,5 +1,5 @@ { - "originHash" : "b971c143a10f44d540b259eca77415d2cc7af5727022d1fbfa7ed14ee8bb1f80", + "originHash" : "ff13a08dcd23e75e1fece101d3795b12bf5a727e772b108379ed732c086f5a3c", "pins" : [ { "identity" : "swift-mutex", diff --git a/Examples/EffectViewExample/EffectViewExample/Counter.swift b/Examples/EffectViewExample/EffectViewExample/Counter.swift index 417e655..73ece66 100644 --- a/Examples/EffectViewExample/EffectViewExample/Counter.swift +++ b/Examples/EffectViewExample/EffectViewExample/Counter.swift @@ -13,7 +13,7 @@ extension EnvironmentValues { } // MARK: - Transducer -extension Counter.Transducer: EffectView::Transducer { +extension Counter.Transducer: Transducer { struct State { var counter = 0 @@ -40,12 +40,12 @@ extension Counter.Transducer: EffectView::Transducer { switch event { case .start: state.counter = 0 - return run(id: "Counter") { input, env in + return task(id: "Counter") { input, env in while true { do { try await Task.sleep(nanoseconds: 1_000_000_000) // 1 sec print("tick") - input(.tick) + try? input(.tick) } catch {} // most likely, the counter task has been cancelled; ignore it. } } @@ -61,6 +61,7 @@ extension Counter.Transducer: EffectView::Transducer { // MARK: - Views extension Counter.Views { + @MainActor struct ContentView: View { var body: some View { EnvReader(\.counterViewEnv) { @@ -69,6 +70,7 @@ extension Counter.Views { } } + @MainActor struct CounterView: View { typealias Transducer = Counter.Transducer typealias Env = Transducer.Env @@ -82,12 +84,12 @@ extension Counter.Views { of: Transducer.self, state: $state, initialEnv: env, - ) { state, send in + ) { state, input in VStack { Text("\(state.counter)") .font(Font.largeTitle.monospacedDigit()) - Button("Start") { send(.start) } - Button("Stop") { send(.stop) } + Button("Start") { try? input(.start) } + Button("Stop") { try? input(.stop) } } } .id(env.id) // restart the EffectView when the env changes diff --git a/Examples/EffectViewExample/EffectViewExample/Movies.swift b/Examples/EffectViewExample/EffectViewExample/Movies.swift index 3b8ba08..baaead9 100644 --- a/Examples/EffectViewExample/EffectViewExample/Movies.swift +++ b/Examples/EffectViewExample/EffectViewExample/Movies.swift @@ -133,24 +133,24 @@ extension Movies.Transducer: Transducer { static func loadMovies() -> Effect { - run(id: "load") { input, env in + task(id: "load") { input, env in do { let movies = try await env.movieFetch() - input(.loaded(movies)) + try input(.loaded(movies)) } catch { - input(.loadFailed(error)) + try input(.loadFailed(error)) } } } static func refreshMovies() -> Effect { // Note: a refresh action - run(id: "refresh") { input, env in + task(id: "refresh") { input, env in do { let movies = try await env.movieFetch() - input(.loaded(movies)) + try input(.loaded(movies)) } catch { - input(.loadFailed(error)) + try input(.loadFailed(error)) } } } @@ -215,7 +215,7 @@ extension Movies.Views { isPresented: .constant(state.error != nil), presenting: state.error ) { _ in - Button("OK") { input(.dismiss) } + Button("OK") { try? input(.dismiss) } } message: { error in Text(error.localizedDescription) } diff --git a/Examples/EffectViewExample/EffectViewExample/RemoteCounter.swift b/Examples/EffectViewExample/EffectViewExample/RemoteCounter.swift index 4f78362..9a3174e 100644 --- a/Examples/EffectViewExample/EffectViewExample/RemoteCounter.swift +++ b/Examples/EffectViewExample/EffectViewExample/RemoteCounter.swift @@ -17,7 +17,8 @@ extension RemoteCounter { /// Private state, a public read-only projection, and an event-driven mutation API. /// Observers must use `withObservationTracking` (or wrap it) — the store itself /// does not publish a stream. - @Observable @MainActor + @Observable + @MainActor final class CounterStore: Sendable { static let shared: CounterStore = .init() @@ -91,19 +92,19 @@ extension RemoteCounter.Transducer: Transducer { case .incrementTapped: print("incrementTapped") - return run { input, env in + return task { input, env in await env.store.send(.increment) } case .decrementTapped: print("decrementTapped") - return run { _, env in + return task { _, env in await env.store.send(.decrement) } case .resetTapped: print("resetTapped") - return run { _, env in + return task { _, env in await env.store.send(.reset) } } @@ -146,9 +147,9 @@ extension RemoteCounter.Views { Text(deltaLabel(state.lastDelta)) .foregroundStyle(.secondary) HStack(spacing: 24) { - Button("−") { send(.decrementTapped) } - Button("+") { send(.incrementTapped) } - Button("Reset") { send(.resetTapped) } + Button("−") { try? send(.decrementTapped) } + Button("+") { try? send(.incrementTapped) } + Button("Reset") { try? send(.resetTapped) } } .buttonStyle(.bordered) } @@ -174,3 +175,4 @@ extension RemoteCounter.Views { Text("RemoteCounter not available on this OS version") } } + diff --git a/Sources/EffectComponents/Transducer/TaskManager.swift b/Sources/EffectComponents/Transducer/TaskManager.swift index f150554..0cbf128 100644 --- a/Sources/EffectComponents/Transducer/TaskManager.swift +++ b/Sources/EffectComponents/Transducer/TaskManager.swift @@ -89,24 +89,15 @@ final class TaskManager { self.state = .cancelling(error: error) // Break the send/taskManager retain cycle once the runtime has irreversibly failed. systemErrorCallback = nil - - for taskKey in Array(tasks.keys) { - tasks[taskKey]?.cancel(with: error) + + for key in tasks.keys { + tasks[key]!.cancel(with: error) } if tasks.isEmpty { state = .cancelled(error: error) } } - private var latchedShutdownError: any Swift.Error { - switch state { - case .active: - return RuntimeError.cancelled - case .cancelling(let error), .cancelled(let error): - return error ?? RuntimeError.cancelled - } - } - /// Cancels the tracked task for `identifier`, if one exists. /// /// All waiters currently attached to that task are resumed with @@ -169,88 +160,58 @@ final class TaskManager { priority: TaskPriority? = nil, isolatedOperation: @escaping (isolated any Actor) async throws -> Output? ) { - // TODO: check if this should be better a precondition guard case .active = state else { + // TODO: check if this should be better a precondition if let continuation { continuation.resume(throwing: latchedShutdownError) } return } - switch option { - case .switchToLatest: - var continuations: [Continuation] = [] - if let taskIdentifier = identifier { - continuations = cancelForReplacement(identifier: taskIdentifier) - } - if let continuation { - continuations.append(continuation) - } + // TODO: we should better use an enum for the valid variants, instead tuple (option, identifier, continuation) + switch (option, identifier, continuation) { + case (.switchToLatest, .some(let identifier), _): + replaceNewTask( + identifier: identifier, + continuation: continuation, + priority: priority, + isolatedOperation: isolatedOperation + ) + + case (.subscribe, .some(let identifier), .some(let continuation)): + addNewTaskOrSubscribe( + identifier: identifier, + continuation: continuation, + priority: priority, + isolatedOperation: isolatedOperation + ) + case (.subscribe, .none, _), (.switchToLatest, .none, _): addNewTask( identifier: identifier, - continuations: continuations, + continuation: continuation, priority: priority, isolatedOperation: isolatedOperation ) - - case .subscribe: - if let identifier, - let continuation, - var taskValue = tasks[TaskKey(identifier)] { - // add subscriber (aka waiter) to the existing tracked task, even if it - // has already been cancelled but has not completed yet. - taskValue.subscribe(continuation: continuation) - tasks[TaskKey(identifier)] = taskValue - } else { - let continuations = continuation.map { [$0] } ?? [] - addNewTask( - identifier: identifier, - continuations: continuations, - priority: priority, - isolatedOperation: isolatedOperation - ) - } + case (.subscribe, .some, .none): + preconditionFailure("subcribing to a task without a continuation is not supported") } } - - /// Cancels the current task for replacement and returns its waiter set. - /// - /// `.switchToLatest` replaces the task instance but preserves the waiter set - /// by moving those continuations onto the replacement task. - @discardableResult - private func cancelForReplacement(identifier: TaskIdentifier) -> [Continuation] { - let taskKey = TaskKey(identifier) - guard var taskValue = tasks[taskKey] else { - return [] + + private var latchedShutdownError: any Swift.Error { + switch state { + case .active: + return RuntimeError.cancelled + case .cancelling(let error), .cancelled(let error): + return error ?? RuntimeError.cancelled } - // Keep the waiter set; `.switchToLatest` replaces the task instance, not the waiters. - let continuations = taskValue.cancelForReplacement() - tasks[taskKey] = taskValue - #if DEBUG - print("EffectManager cancelled task for replacement: \(identifier)-\(taskValue.id)") - #endif - return continuations } - - /// Inserts a fresh tracked task under `identifier`. - /// - /// The new task captures `systemActor`, runs `isolatedOperation`, and then - /// either resumes the attached waiters with the operation result or begins - /// manager cancellation when the operation fails with a non-task-cancellation - /// error. - private func addNewTask( + + private func makeTask( systemActor: isolated any Actor = #isolation, - identifier: TaskIdentifier?, - continuations: [Continuation], + taskKey: TaskKey, + id taskId: Int, priority: TaskPriority?, isolatedOperation: @escaping (isolated any Actor) async throws -> Output? - ) { - let taskKey: TaskKey - let id = taskId - if let identifier = identifier { - taskKey = TaskKey(identifier) - } else { - taskKey = TaskKey.makeAnon(with: id) - } + ) -> Task { // CAUTION: `systemActor` is captured *strongly*!. In cases, where the // systemActor keeps a strong reference to `self`, self will never be // deallocated before all tasks are finished, because the captured @@ -259,7 +220,13 @@ final class TaskManager { // Actor" based on Swift Actors. That is, a proper implementation of an // "FSM Effect Actor" should always have a `cancel()` method which cancels // all running tasks and additionally prevents enqueueing new ones. - let task = Task(name: taskKey.string, priority: priority) { [weak self] in + + #if DEBUG + let taskName = taskKey.string + #else + let taskName: String? = nil + #endif + let task = Task(name: taskName, priority: priority) { [weak self] in _ = systemActor let result: Result do { @@ -271,36 +238,152 @@ final class TaskManager { switch result { case .failure(let error): if error is CancellationError && Task.isCancelled { - self?.finish(taskKey: taskKey, id: id, result: result) + self?.finish(taskKey: taskKey, id: taskId, result: result) } else { if let systemErrorCallback = self?.systemErrorCallback { await systemErrorCallback(error) } else { self?.cancel(with: error) } - self?.complete(taskKey: taskKey, id: id) + self?.complete(taskKey: taskKey, id: taskId) } case .success: - self?.finish(taskKey: taskKey, id: id, result: result) + self?.finish(taskKey: taskKey, id: taskId, result: result) } } - + return task + } + + /// Inserts a new tracked task under `identifier`. If a task already exists at the given + /// identifier, it cancels the previous one and all associated waiters. + /// + /// The new task captures `systemActor`, runs `isolatedOperation`, and then + /// either resumes the attached waiters with the operation result or begins + /// manager cancellation when the operation fails with a non-task-cancellation + /// error. + private func addNewTask( + systemActor: isolated any Actor = #isolation, + identifier: TaskIdentifier?, + continuation: Continuation?, + priority: TaskPriority?, + isolatedOperation: @escaping (isolated any Actor) async throws -> Output? + ) { + let taskKey: TaskKey + let id = taskId + if let identifier = identifier { + taskKey = TaskKey(identifier) + } else { + taskKey = TaskKey.makeAnon(with: id) + } + if var taskValue = tasks[taskKey] { + taskValue.cancel() + } + let task = makeTask( + taskKey: taskKey, + id: taskId, + priority: priority, + isolatedOperation: isolatedOperation + ) + let continuations = continuation != nil ? [continuation!] : [] let taskValue = TaskValue(id: id, task: task, continuations: continuations) - taskId += 1 tasks[taskKey] = taskValue - + taskId += 1 #if DEBUG print("EffectManager added Task: \(taskKey)-\(taskValue.id)") #endif } + + /// If a task already exists at the given identifier, adds it as a subscriber and the closure + /// `isolatedOperation` will be discarded. Otherwise a new Task will be added. + /// + /// The new task captures `systemActor`, runs `isolatedOperation`, and then + /// either resumes the attached waiters with the operation result or begins + /// manager cancellation when the operation fails with a non-task-cancellation + /// error. + // TODO: elevate priority when this task will be added as a subscriber. + private func addNewTaskOrSubscribe( + systemActor: isolated any Actor = #isolation, + identifier: TaskIdentifier, + continuation: Continuation, + priority: TaskPriority?, + isolatedOperation: @escaping (isolated any Actor) async throws -> Output? + ) { + let taskKey = TaskKey(identifier) + if var taskValue = tasks[taskKey] { + taskValue.continuations.append(continuation) + tasks[taskKey] = taskValue + #if DEBUG + print("EffectManager added subscriber to Task: \(taskKey)-\(taskValue.id)") + #endif + } else { + addNewTask( + identifier: identifier, + continuation: continuation, + priority: priority, + isolatedOperation: isolatedOperation + ) + } + } + + /// Replaces a the task under `identifier`. Existing continuations will be kept. + /// + /// If no previous task exist with this identifier, a new task well be added. + /// + /// The new task captures `systemActor`, runs `isolatedOperation`, and then + /// either resumes the attached waiters with the operation result or begins + /// manager cancellation when the operation fails with a non-task-cancellation + /// error. + private func replaceNewTask( + systemActor: isolated any Actor = #isolation, + identifier: TaskIdentifier, + continuation: Continuation?, + priority: TaskPriority?, + isolatedOperation: @escaping (isolated any Actor) async throws -> Output? + ) { + let id = taskId + let taskKey = TaskKey(identifier) + + if var taskValue = tasks.removeValue(forKey: taskKey) { + taskValue.task.cancel() + + let task = makeTask( + taskKey: taskKey, + id: id, + priority: priority, + isolatedOperation: isolatedOperation + ) + taskValue.task = task + taskValue.id = id + if let continuation { + taskValue.continuations.append(continuation) + } + tasks[taskKey] = taskValue + taskId += 1 + #if DEBUG + print("EffectManager replaced Task: \(taskKey)-\(taskValue.id)") + #endif + } else { + addNewTask( + identifier: identifier, + continuation: continuation, + priority: priority, + isolatedOperation: isolatedOperation + ) + } + } /// Resumes all waiters for the matching task and removes it from tracking. private func finish(taskKey: TaskKey, id: Int, result: Result) { if var taskValue = tasks[taskKey], taskValue.id == id { taskValue.resume(with: result) - tasks[taskKey] = taskValue + tasks[taskKey] = nil + if tasks.isEmpty, case .cancelling(let error) = state { + state = .cancelled(error: error) + } + #if DEBUG + print("EffectManager task completed: \(taskKey.string)-\(id)") + #endif } - complete(taskKey: taskKey, id: id) } /// Removes the tracked task if `id` still matches the current entry. @@ -348,8 +431,8 @@ extension TaskManager { /// The mutable tracked value for one logical task entry. struct TaskValue { - let id: Int // unique task id - let task: Task + var id: Int // unique task id + var task: Task var continuations: [Continuation] init(id: Int, task: Task, continuations: [Continuation]) { @@ -385,14 +468,6 @@ extension TaskManager { mutating func subscribe(continuation: Continuation) { continuations.append(continuation) } - - /// Cancels the task for `.switchToLatest` while preserving its waiters. - mutating func cancelForReplacement() -> [Continuation] { - task.cancel() - let continuations = continuations - self.continuations = [] - return continuations - } } } diff --git a/Sources/EffectComponents/Transducer/Transducer.Effects.swift b/Sources/EffectComponents/Transducer/Transducer.Effects.swift index 61f27a3..95d9d79 100644 --- a/Sources/EffectComponents/Transducer/Transducer.Effects.swift +++ b/Sources/EffectComponents/Transducer/Transducer.Effects.swift @@ -15,10 +15,6 @@ extension Transducer where Effect == TransducerEffect { /// cancelled by the runtime, late failures from that obsolete work no longer /// participate in global error escalation. /// - /// Prefer ``run(id:priority:option:operation:)`` for fire-and-forget tasks and - /// ``request(id:priority:option:operation:)`` for perform-driven tasks rather than - /// constructing `.task` directly. - /// /// - Parameters: /// - id: An optional identifier used to track and cancel the task. Pass `nil` for /// anonymous tasks that run to completion without cancellation support. @@ -31,15 +27,19 @@ extension Transducer where Effect == TransducerEffect { /// - operation: The async work to perform. Returns an optional `Output` value /// forwarded to any caller suspended on ``Input/request(_:)``. /// + /// When the `operation` throws, the effect engine treats this as a system error. This typically + /// indicates that the input channel failed (for example, its host actor was deallocated or an + /// internal buffer overflow occurred). A thrown error places the transducer into a failure mode: + /// the runtime stops processing further events, cancels all running tasks associated with the + /// transducer, and begins propagating the failure to any event senders and waiters. Prefer to + /// design operations that do not throw under normal circumstances; only system-level failures + /// should surface as thrown errors from `operation`. + /// + /// - Throws: A system error indicating that the transducer can no longer operate correctly. Upon + /// throw, the transducer transitions to a failure mode, stops processing new events, and cancels + /// all running tasks. + /// /// - Returns: The effect. - // TODO: Explain clearly what it means when an operation throws. - // Usually, operations should not fail, but in some cases, the operation may use - // an input to send events back to the system and *this* input can fail due to a "system error". System - // errors are critical errors - that is, it might mean, the actor is deallocated, - // or a potential event buffer overflowed, or some other system error occurred. - // That means the transducer is no longer guaranteed to perform correctly. The - // best course of action is to tear down the transducer and actor, and forward - // the error to event senders and waiters. @inline(__always) public static func task( id: TaskIdentifier? = nil, @@ -50,6 +50,60 @@ extension Transducer where Effect == TransducerEffect { .init(._task(id: id, priority: priority, option: option, operation: operation)) } + /// Returns an effect which when invoked starts an async throwing operation isolated to a global actor + /// tracked by the effect engine. + /// + /// The `operation` closure receives an ``Input`` handle for dispatching events and + /// the captured `Env` for dependencies. Named tasks are automatically cancelled when + /// the view disappears, or when ``cancel(_:)`` is returned from `update` with the + /// same identifier. + /// + /// - Important: Managed cancellation takes precedence over racing task failures. + /// If the runtime cancels a tracked task and the operation concurrently throws, + /// the effect engine may classify that outcome as cancellation rather than as a + /// system error. This is intentional: once a task has been superseded or + /// cancelled by the runtime, late failures from that obsolete work no longer + /// participate in global error escalation. + /// + /// - Parameters: + /// - id: An optional identifier used to track and cancel the task. Pass `nil` for + /// anonymous tasks that run to completion without cancellation support. + /// - priority: The `TaskPriority` for the launched task. Pass `nil` to inherit + /// the current task's priority. + /// - option: Defines how overlapping waiters for the same `id` are handled. + /// `.subscribe` keeps the running task and attaches the new waiter to it. + /// `.switchToLatest` cancels the running task, starts a fresh one, and moves + /// all current waiters for that identifier onto the replacement task. + /// - operation: The async work to perform. Returns an optional `Output` value + /// forwarded to any caller suspended on ``Input/request(_:)``. + /// + /// When the `operation` throws, the effect engine treats this as a system error. This typically + /// indicates that the input channel failed (for example, its host actor was deallocated or an + /// internal buffer overflow occurred). A thrown error places the transducer into a failure mode: + /// the runtime stops processing further events, cancels all running tasks associated with the + /// transducer, and begins propagating the failure to any event senders and waiters. Prefer to + /// design operations that do not throw under normal circumstances; only system-level failures + /// should surface as thrown errors from `operation`. + /// + /// - Throws: A system error indicating that the transducer can no longer operate correctly. Upon + /// throw, the transducer transitions to a failure mode, stops processing new events, and cancels + /// all running tasks. + /// + /// - Returns: The effect. + @inline(__always) + public static func run( + id: TaskIdentifier? = nil, + priority: TaskPriority? = nil, + option: TaskExecutionOption = .switchToLatest, + operation: @escaping @Sendable @isolated(any) (any TransducerInput & Sendable, Env) async throws -> Void + ) -> Effect where Env: Sendable { + .init(._task(id: id, priority: priority, option: option, operation: { input, env in + try await operation(input, env) + return nil + })) + } + + /// Returns an effect which when invoked starts an async throwing operation isolated to the system actor /// tracked by the effect engine. /// @@ -65,10 +119,6 @@ extension Transducer where Effect == TransducerEffect { /// cancelled by the runtime, late failures from that obsolete work no longer /// participate in global error escalation. /// - /// Prefer ``run(id:priority:option:operation:)`` for fire-and-forget tasks and - /// ``request(id:priority:option:operation:)`` for perform-driven tasks rather than - /// constructing `.task` directly. - /// /// - Parameters: /// - id: An optional identifier used to track and cancel the task. Pass `nil` for /// anonymous tasks that run to completion without cancellation support. @@ -81,6 +131,18 @@ extension Transducer where Effect == TransducerEffect { /// - isolatedOperation: The async work to perform. Returns an optional `Output` value /// forwarded to any caller suspended on ``Input/request(_:)``. /// + /// When the `operation` throws, the effect engine treats this as a system error. This typically + /// indicates that the input channel failed (for example, its host actor was deallocated or an + /// internal buffer overflow occurred). A thrown error places the transducer into a failure mode: + /// the runtime stops processing further events, cancels all running tasks associated with the + /// transducer, and begins propagating the failure to any event senders and waiters. Prefer to + /// design operations that do not throw under normal circumstances; only system-level failures + /// should surface as thrown errors from `operation`. + /// + /// - Throws: A system error indicating that the transducer can no longer operate correctly. Upon + /// throw, the transducer transitions to a failure mode, stops processing new events, and cancels + /// all running tasks. + /// /// - Returns: An effect. @inline(__always) public static func task( @@ -92,6 +154,63 @@ extension Transducer where Effect == TransducerEffect { .init(._taskIsolated(id: id, priority: priority, option: option, isolatedOperation: isolatedOperation)) } + /// Returns an effect which when invoked starts an async throwing operation isolated to the system actor + /// tracked by the effect engine. + /// + /// The `isolatedOperation` closure receives an ``Input`` handle for dispatching events and + /// the captured `Env` for dependencies. Named tasks are automatically cancelled when + /// the view disappears, or when ``cancel(_:)`` is returned from `update` with the + /// same identifier. + /// + /// - Important: Managed cancellation takes precedence over racing task failures. + /// If the runtime cancels a tracked task and the operation concurrently throws, + /// the effect engine may classify that outcome as cancellation rather than as a + /// system error. This is intentional: once a task has been superseded or + /// cancelled by the runtime, late failures from that obsolete work no longer + /// participate in global error escalation. + /// + /// - Parameters: + /// - id: An optional identifier used to track and cancel the task. Pass `nil` for + /// anonymous tasks that run to completion without cancellation support. + /// - priority: The `TaskPriority` for the launched task. Pass `nil` to inherit + /// the current task's priority. + /// - option: Defines how overlapping waiters for the same `id` are handled. + /// `.subscribe` keeps the running task and attaches the new waiter to it. + /// `.switchToLatest` cancels the running task, starts a fresh one, and moves + /// all current waiters for that identifier onto the replacement task. + /// - isolatedOperation: The async work to perform. Returns an optional `Output` value + /// forwarded to any caller suspended on ``Input/request(_:)``. + /// + /// When the `operation` throws, the effect engine treats this as a system error. This typically + /// indicates that the input channel failed (for example, its host actor was deallocated or an + /// internal buffer overflow occurred). A thrown error places the transducer into a failure mode: + /// the runtime stops processing further events, cancels all running tasks associated with the + /// transducer, and begins propagating the failure to any event senders and waiters. Prefer to + /// design operations that do not throw under normal circumstances; only system-level failures + /// should surface as thrown errors from `operation`. + /// + /// - Throws: A system error indicating that the transducer can no longer operate correctly. Upon + /// throw, the transducer transitions to a failure mode, stops processing new events, and cancels + /// all running tasks. + /// + /// - Returns: An effect. + @inline(__always) + public static func run( + id: TaskIdentifier? = nil, + priority: TaskPriority? = nil, + option: TaskExecutionOption = .switchToLatest, + isolatedOperation: sending @escaping (any TransducerInput, Env, isolated any Actor) async throws -> Void + ) -> Effect { + .init(._taskIsolated(id: id, priority: priority, option: option, isolatedOperation: { input, env, isolation in + try await isolatedOperation(input, env, isolation) + return nil + })) + } + +} + +extension Transducer where Effect == TransducerEffect { + /// Return an effect which when invoked executes a synchronous step that may produce the next /// event to process immediately. /// @@ -140,7 +259,7 @@ extension Transducer where Effect == TransducerEffect { return nil })) } - + /// Return an effect which when invoked executes an async step on a user specified global actor /// that may produce the next event to process immediately. /// @@ -166,7 +285,7 @@ extension Transducer where Effect == TransducerEffect { ) -> Effect { .init(._actionAsync(action)) } - + /// Return an effect which when invoked executes an async step on a user specified global actor. /// /// The `action` closure receives `Env`. It runs synchronously on the system actor before any @@ -187,7 +306,7 @@ extension Transducer where Effect == TransducerEffect { return nil })) } - + /// Return an effect which when invoked executes an async step on the system actor /// that may produce the next event to process immediately. /// @@ -234,7 +353,10 @@ extension Transducer where Effect == TransducerEffect { return nil })) } + +} +extension Transducer where Effect == TransducerEffect { /// Returns an effect which, when invoked, feeds an event back into `update` immediately /// in the current synchronous turn. @@ -248,7 +370,7 @@ extension Transducer where Effect == TransducerEffect { public static func send(_ event: Event) -> Effect { .init(._event(event)) } - + /// Returns an effect which when invoked feeds `event` back into `update` immediately, in /// the current synchronous turn. /// @@ -299,9 +421,10 @@ extension Transducer where Effect == TransducerEffect { } } - +// deprecated +#if false extension Transducer where Effect == TransducerEffect { - + /// Returns an effect which, when invoked, starts a fire-and-forget async task that communicates /// back through events. /// @@ -391,8 +514,7 @@ extension Transducer where Effect == TransducerEffect { return nil }) } - - + /// Starts an async task whose result is returned to the caller of ``Input/request(_:)``. /// /// The `operation` closure performs its work, drives the FSM to a completion event @@ -429,3 +551,5 @@ extension Transducer where Effect == TransducerEffect { } } + +#endif diff --git a/Sources/EffectComponents/Transducer/Transducer.swift b/Sources/EffectComponents/Transducer/Transducer.swift index 382e7a5..f380410 100644 --- a/Sources/EffectComponents/Transducer/Transducer.swift +++ b/Sources/EffectComponents/Transducer/Transducer.swift @@ -115,7 +115,7 @@ import Foundation /// // time the request settles /// } /// ``` -public protocol Transducer: SendableMetatype { +public protocol Transducer { /// Mutable feature state owned by the host runtime. associatedtype State @@ -496,6 +496,24 @@ extension Transducer where Effect == TransducerEffect { } ) return (nil, nil) + + case ._taskNonsending(id: let identifier, priority: let priority, option: let option, nonsendingOperation: let nonsendingOperation): + guard let input else { + preconditionFailure("No Input value given when creating a task") + } + try taskManager.checkCancellation() + taskManager.addTask( + with: identifier, + option: option, + continuation: continuation, + priority: priority, + isolatedOperation: { isolated in + _ = systemActor + return try await nonsendingOperation(input, env) + } + ) + return (nil, nil) + case ._event(event: let event): return (event, continuation) diff --git a/Sources/EffectComponents/Transducer/TransducerEffect.swift b/Sources/EffectComponents/Transducer/TransducerEffect.swift index c2be5e5..9c7fa21 100644 --- a/Sources/EffectComponents/Transducer/TransducerEffect.swift +++ b/Sources/EffectComponents/Transducer/TransducerEffect.swift @@ -69,6 +69,13 @@ enum EffectType { isolatedOperation: (any TransducerInput, Env, isolated any Actor) async throws -> Output? ) + case _taskNonsending( + id: TaskIdentifier?, + priority: TaskPriority?, + option: TaskExecutionOption, + nonsendingOperation: nonisolated(nonsending) (any TransducerInput, Env) async throws -> Output? + ) + case _actionSync( (Env) -> Event? ) diff --git a/Sources/EffectComponents/Transducer/TransducerInput.swift b/Sources/EffectComponents/Transducer/TransducerInput.swift index 447c8f7..bba793e 100644 --- a/Sources/EffectComponents/Transducer/TransducerInput.swift +++ b/Sources/EffectComponents/Transducer/TransducerInput.swift @@ -19,7 +19,7 @@ /// Equal task identifiers therefore mean more than "same cancellation key": they /// declare the same logical in-flight work. Overlapping waiters for one identifier /// must converge to one current result or one current error. -public protocol TransducerInput: SendableMetatype { +public protocol TransducerInput { associatedtype Event associatedtype Output diff --git a/Tests/EffectComponents/EffectViewInputTests.swift b/Tests/EffectComponents/EffectViewInputTests.swift index 49ece3c..945c5f0 100644 --- a/Tests/EffectComponents/EffectViewInputTests.swift +++ b/Tests/EffectComponents/EffectViewInputTests.swift @@ -172,7 +172,7 @@ struct EffectViewInputTests { static func update(_ state: inout State, event: Event) -> Effect? { switch event { case .load: - return request(id: "load") { input, _ in + return task(id: "load") { input, _ in do { try await Task.sleep(for: .milliseconds(1)) } catch { @@ -276,4 +276,4 @@ struct EffectViewInputTests { } } -#endif \ No newline at end of file +#endif diff --git a/Tests/EffectComponents/TaskManagerTests.swift b/Tests/EffectComponents/TaskManagerTests.swift index e120783..be91165 100644 --- a/Tests/EffectComponents/TaskManagerTests.swift +++ b/Tests/EffectComponents/TaskManagerTests.swift @@ -2,11 +2,12 @@ import Foundation import Testing @testable import EffectComponents -private enum TaskManagerFailure: Error, Equatable { +private enum MyError: Error, Equatable { case boom case later } + private enum TaskManagerStateSnapshot: Equatable, Sendable { case active case cancellingNoError @@ -17,29 +18,56 @@ private enum TaskManagerStateSnapshot: Equatable, Sendable { case cancelledOther } -private actor TaskManagerHarness { - let taskManager = TaskManager() +private actor TaskManagerHarness { + typealias OperationFunc = nonisolated(nonsending) () async throws -> Output? + typealias Continuation = TaskManager.Continuation + + let taskManager = TaskManager() + + func addTask( + with identifier: TaskIdentifier? = nil, + option: TaskExecutionOption, + continuation: CheckedContinuation?, + priority: TaskPriority? = nil, + operation: sending @escaping OperationFunc + ) { + taskManager.addTask( + with: identifier, + option: option, + continuation: continuation, + priority: priority, + isolatedOperation: { actor in + self.assertIsolated() + return try await operation() + } + ) + } func request( identifier: TaskIdentifier, - started: Expectation?, - cancelled: Expectation? - ) async throws -> String? { - try await withCheckedThrowingContinuation { continuation in - taskManager.addTask( - systemActor: self, + option: TaskExecutionOption = .subscribe, + started: Expectation? = nil, + finished: Expectation? = nil, + failure: Expectation? = nil, + cancelled: Expectation? = nil, + operation: sending @escaping OperationFunc + ) async throws -> Output? { + try await withCheckedThrowingContinuation { (continuation: Continuation) in + addTask( with: identifier, - option: .subscribe, + option: option, continuation: continuation, - isolatedOperation: { _ in - started?.fulfill() + operation: { do { - while true { - try await Task.sleep(nanoseconds: 50_000_000) - } + defer { finished?.fulfill() } + started?.fulfill() + return try await operation() } catch is CancellationError { cancelled?.fulfill() throw CancellationError() + } catch { + failure?.fulfill() + throw error } } ) @@ -57,7 +85,7 @@ private actor TaskManagerHarness { func failTrackedTask( identifier: TaskIdentifier, started: Expectation? - ) async throws -> String? { + ) async throws -> Output? { try await withCheckedThrowingContinuation { continuation in taskManager.addTask( systemActor: self, @@ -66,7 +94,7 @@ private actor TaskManagerHarness { continuation: continuation, isolatedOperation: { _ in started?.fulfill() - throw TaskManagerFailure.boom + throw MyError.boom } ) } @@ -84,7 +112,7 @@ private actor TaskManagerHarness { switch error { case nil: return .cancellingNoError - case let error as TaskManagerFailure where error == .boom: + case let error as MyError where error == .boom: return .cancellingBoom default: return .cancellingOther @@ -93,7 +121,7 @@ private actor TaskManagerHarness { switch error { case nil: return .cancelledNoError - case let error as TaskManagerFailure where error == .boom: + case let error as MyError where error == .boom: return .cancelledBoom default: return .cancelledOther @@ -104,9 +132,9 @@ private actor TaskManagerHarness { @Suite("Task manager") struct TaskManagerTests { - + @Test func cancelWithSystemErrorCancelsTrackedTasksAndRejectsNewAdds() async throws { - let harness = TaskManagerHarness() + let harness = TaskManagerHarness() let started = Expectation() let cancelled = Expectation() let timeout: UInt64 = 5_000_000_000 @@ -116,65 +144,49 @@ struct TaskManagerTests { identifier: "tracked", started: started, cancelled: cancelled - ) + ) { + while true { + try await Task.sleep(nanoseconds: 50_000_000) + } + } } try await started.await(nanoseconds: timeout) - - await harness.cancelWithError(TaskManagerFailure.boom) - - do { + await harness.cancelWithError(MyError.boom) + + await #expect(throws: MyError.boom, "Expected latched system error") { try await harness.checkCancellation() - Issue.record("Expected latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected latched error: \(error)") } - - do { + await #expect(throws: MyError.boom) { _ = try await waiter.value - Issue.record("Expected active waiter to receive the latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected waiter error: \(error)") } try await cancelled.await(nanoseconds: timeout) - - do { + await #expect(throws: MyError.boom, "Expected new waiter to be rejected after system error") { _ = try await harness.request( - identifier: "tracked", - started: nil, - cancelled: nil - ) - Issue.record("Expected new waiter to be rejected after system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected rejected waiter error: \(error)") + identifier: "tracked" + ) { + Issue.record("Unexpected execution of task") + while true { + try await Task.sleep(nanoseconds: 50_000_000) + } + } } } @Test func cancelWithSystemErrorKeepsFirstError() async throws { - let harness = TaskManagerHarness() + let harness = TaskManagerHarness() - await harness.cancelWithError(TaskManagerFailure.boom) - await harness.cancelWithError(TaskManagerFailure.later) - - do { + await harness.cancelWithError(MyError.boom) + await harness.cancelWithError(MyError.later) + + await #expect(throws: MyError.boom, "Expected first latched system error") { try await harness.checkCancellation() - Issue.record("Expected first latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected error: \(error)") } } @Test func thrownTaskErrorLatchesSystemErrorAndCancelsItsWaiters() async throws { - let harness = TaskManagerHarness() + let harness = TaskManagerHarness() let started = Expectation() let timeout: UInt64 = 5_000_000_000 @@ -183,28 +195,17 @@ struct TaskManagerTests { } try await started.await(nanoseconds: timeout) - - do { + + await #expect(throws: MyError.boom, "Expected thrown task waiter to receive the latched system error") { _ = try await waiter.value - Issue.record("Expected thrown task waiter to receive the latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected waiter error: \(error)") } - - do { + await #expect(throws: MyError.boom, "Expected latched system error") { try await harness.checkCancellation() - Issue.record("Expected latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected latched error: \(error)") } } @Test func cancelWithoutErrorCancelsActiveWaitersButRejectsFutureOnesAsRuntimeUnavailable() async throws { - let harness = TaskManagerHarness() + let harness = TaskManagerHarness() let started = Expectation() let cancelled = Expectation() let timeout: UInt64 = 5_000_000_000 @@ -214,57 +215,44 @@ struct TaskManagerTests { identifier: "tracked", started: started, cancelled: cancelled - ) + ) { + while true { + try await Task.sleep(nanoseconds: 50_000_000) + } + } } try await started.await(nanoseconds: timeout) - await harness.cancelWithoutError() - do { + await #expect(throws: RuntimeError.cancelled, "Expected runtime unavailable cancellation") { try await harness.checkCancellation() - Issue.record("Expected runtime unavailable cancellation") - } catch let error as RuntimeError { - #expect(error == .cancelled) - } catch { - Issue.record("Unexpected latched error: \(error)") } - - do { + await #expect(throws: CancellationError.self, "Expected active waiter to receive CancellationError") { _ = try await waiter.value - Issue.record("Expected active waiter to receive CancellationError") - } catch is CancellationError { - /* expected */ - } catch { - Issue.record("Unexpected waiter error: \(error)") } - try await cancelled.await(nanoseconds: timeout) - - do { + await #expect(throws: RuntimeError.cancelled, "Expected future waiter to be rejected as runtime unavailable") { _ = try await harness.request( identifier: "tracked", started: nil, cancelled: nil - ) - Issue.record("Expected future waiter to be rejected as runtime unavailable") - } catch let error as RuntimeError { - #expect(error == .cancelled) - } catch { - Issue.record("Unexpected rejected waiter error: \(error)") + ) { + while true { + try await Task.sleep(nanoseconds: 50_000_000) + } + } } } @Test func cancelTransitionsToCancelledStateImmediatelyWhenNoTasksAreTracked() async { - let harness = TaskManagerHarness() - - await harness.cancelWithError(TaskManagerFailure.boom) - + let harness = TaskManagerHarness() + await harness.cancelWithError(MyError.boom) #expect(await harness.stateSnapshot() == .cancelledBoom) } @Test func cancelTransitionsFromCancellingToCancelledAfterTrackedTaskDrains() async throws { - let harness = TaskManagerHarness() + let harness = TaskManagerHarness() let started = Expectation() let cancelled = Expectation() let timeout: UInt64 = 5_000_000_000 @@ -274,26 +262,21 @@ struct TaskManagerTests { identifier: "tracked", started: started, cancelled: cancelled - ) + ) { + while true { + try await Task.sleep(nanoseconds: 50_000_000) + } + } } try await started.await(nanoseconds: timeout) - - await harness.cancelWithError(TaskManagerFailure.boom) - + await harness.cancelWithError(MyError.boom) #expect(await harness.stateSnapshot() == .cancellingBoom) - do { + await #expect(throws: MyError.boom, "Expected active waiter to receive the latched system error") { _ = try await waiter.value - Issue.record("Expected active waiter to receive the latched system error") - } catch let error as TaskManagerFailure { - #expect(error == .boom) - } catch { - Issue.record("Unexpected waiter error: \(error)") } - try await cancelled.await(nanoseconds: timeout) - #expect(await harness.stateSnapshot() == .cancelledBoom) } } diff --git a/Tests/EffectComponents/TaskSubscriptionTests.swift b/Tests/EffectComponents/TaskSubscriptionTests.swift index c71ddcf..a0bfbc1 100644 --- a/Tests/EffectComponents/TaskSubscriptionTests.swift +++ b/Tests/EffectComponents/TaskSubscriptionTests.swift @@ -67,7 +67,7 @@ extension TaskSubscriptionTests { static func update(_ state: inout State, event: Event) -> Effect? { switch event { case .load: - return request(id: "shared-load", option: .subscribe) { _, env in + return task(id: "shared-load", option: .subscribe) { _, env in await env.counter.increment() env.started.fulfill() do { @@ -232,7 +232,7 @@ extension TaskSubscriptionTests { static func update(_ state: inout State, event: Event) -> Effect? { switch event { case .load: - return request(id: "shared-load", option: .subscribe) { _, env in + return task(id: "shared-load", option: .subscribe) { _, env in let count = await env.counter.increment() return "output-\(count)" } @@ -283,7 +283,7 @@ extension TaskSubscriptionTests { static func update(_ state: inout State, event: Event) -> Effect? { switch event { case .load: - return request(id: "shared-load", option: .subscribe) { _, env in + return task(id: "shared-load", option: .subscribe) { _, env in let invocation = await env.counter.increment() if invocation > 1 { return "fresh-output-\(invocation)" @@ -369,10 +369,10 @@ extension TaskSubscriptionTests { @Test func subscribeAttachesToCancelledTrackedTaskAndCancelsWaitersOnLateFailure() async throws { struct WorkerEnv: Sendable { - let counter: InvocationCounter - let started: Expectation - let cancelled: Expectation - let release: Expectation + let counter: InvocationCounter = InvocationCounter() + let startedExpectation: Expectation = .init() + let cancelledExpectation: Expectation = .init() + let releaseExpectation: Expectation = .init() let timeout: UInt64 } @@ -394,13 +394,13 @@ extension TaskSubscriptionTests { if invocation > 1 { return "fresh-output-\(invocation)" } - env.started.fulfill() + env.startedExpectation.fulfill() do { try await Task.sleep(nanoseconds: env.timeout) return "stale-output" } catch is CancellationError { - env.cancelled.fulfill() - try? await env.release.await(nanoseconds: env.timeout) + env.cancelledExpectation.fulfill() + try? await env.releaseExpectation.await(nanoseconds: env.timeout) throw LateFailure.boom } } @@ -412,25 +412,16 @@ extension TaskSubscriptionTests { static func output(state: State, event: Event) -> String { "" } } - let counter = InvocationCounter() - let startedExpectation = Expectation() - let cancelledExpectation = Expectation() - let releaseExpectation = Expectation() let timeout: UInt64 = 5_000_000_000 var capturedInput: EffectViewInput? + let env = WorkerEnv(timeout: timeout) let secondRequestEnqueuedExpectation = Expectation() try await testView(initialState: T.State()) { binding in EffectView( of: T.self, state: binding, - initialEnv: WorkerEnv( - counter: counter, - started: startedExpectation, - cancelled: cancelledExpectation, - release: releaseExpectation, - timeout: timeout - ) + initialEnv: env ) { _, input in Color.clear.onAppear { capturedInput = input } } @@ -438,10 +429,10 @@ extension TaskSubscriptionTests { guard let input = capturedInput else { Issue.record("Input not captured"); return } let firstWaiter = Task { try await input.request(.load) } - try await startedExpectation.await(nanoseconds: timeout) + try await env.startedExpectation.await(nanoseconds: timeout) try await input.send(.stop) - try await cancelledExpectation.await(nanoseconds: timeout) + try await env.cancelledExpectation.await(nanoseconds: timeout) do { _ = try await firstWaiter.value @@ -458,7 +449,7 @@ extension TaskSubscriptionTests { enqueued: secondRequestEnqueuedExpectation ) try await secondRequestEnqueuedExpectation.await(nanoseconds: timeout) - releaseExpectation.fulfill() + env.releaseExpectation.fulfill() do { _ = try await secondWaiter.value @@ -469,17 +460,17 @@ extension TaskSubscriptionTests { Issue.record("Unexpected second waiter error: \(error)") } - #expect(await counter.count == 1, "subscribe should attach to the cancelled tracked task instead of starting fresh work") + #expect(await env.counter.count == 1, "subscribe should attach to the cancelled tracked task instead of starting fresh work") } } @Test func switchToLatestRestartsTaskAndReturnsReplacementResultToAllWaiters() async throws { struct WorkerEnv: Sendable { - let counter: InvocationCounter - let firstStarted: Expectation - let firstCancelled: Expectation - let secondStarted: Expectation - let secondRelease: Expectation + let counter: InvocationCounter = InvocationCounter() + let firstStartedExpectation: Expectation = .init() + let firstCancelledExpectation: Expectation = .init() + let secondStartedExpectation: Expectation = .init() + let secondReleaseExpectation: Expectation = .init() let timeout: UInt64 } @@ -495,34 +486,34 @@ extension TaskSubscriptionTests { return task(id: "replaceable", option: .switchToLatest) { _, env in let invocation = await env.counter.increment() if invocation == 1 { - env.firstStarted.fulfill() + env.firstStartedExpectation.fulfill() do { try await Task.sleep(nanoseconds: env.timeout) return "stale-first-output" } catch is CancellationError { - env.firstCancelled.fulfill() + env.firstCancelledExpectation.fulfill() throw CancellationError() } } - env.secondStarted.fulfill() - try? await env.secondRelease.await(nanoseconds: env.timeout) + env.secondStartedExpectation.fulfill() + try? await env.secondReleaseExpectation.await(nanoseconds: env.timeout) return "replacement-output" } case .second: return task(id: "replaceable", option: .switchToLatest) { _, env in let invocation = await env.counter.increment() if invocation == 1 { - env.firstStarted.fulfill() + env.firstStartedExpectation.fulfill() do { try await Task.sleep(nanoseconds: env.timeout) return "stale-first-output" } catch is CancellationError { - env.firstCancelled.fulfill() + env.firstCancelledExpectation.fulfill() throw CancellationError() } } - env.secondStarted.fulfill() - try? await env.secondRelease.await(nanoseconds: env.timeout) + env.secondStartedExpectation.fulfill() + try? await env.secondReleaseExpectation.await(nanoseconds: env.timeout) return "replacement-output" } } @@ -531,56 +522,56 @@ extension TaskSubscriptionTests { static func output(state: State, event: Event) -> String { "" } } - let counter = InvocationCounter() - let firstStartedExpectation = Expectation() - let firstCancelledExpectation = Expectation() - let secondStartedExpectation = Expectation() - let secondReleaseExpectation = Expectation() - let timeout: UInt64 = 5_000_000_000 + let timeout: UInt64 = 500_000_000_000 + let env = WorkerEnv(timeout: timeout) + var capturedInput: EffectViewInput? try await testView(initialState: T.State()) { binding in EffectView( of: T.self, state: binding, - initialEnv: WorkerEnv( - counter: counter, - firstStarted: firstStartedExpectation, - firstCancelled: firstCancelledExpectation, - secondStarted: secondStartedExpectation, - secondRelease: secondReleaseExpectation, - timeout: timeout - ) + initialEnv: env ) { _, input in Color.clear.onAppear { capturedInput = input } } } expect: { guard let input = capturedInput else { Issue.record("Input not captured"); return } - let firstWaiter = Task { try await input.request(.first) } - try await firstStartedExpectation.await(nanoseconds: timeout) + let firstWaiter = Task { + do { + return try await input.request(.first) + } catch { + print(error.localizedDescription) + throw error + } + } + try await env.firstStartedExpectation.await(nanoseconds: timeout) let secondWaiter = Task { try await input.request(.second) } - try await firstCancelledExpectation.await(nanoseconds: timeout) - try await secondStartedExpectation.await(nanoseconds: timeout) - secondReleaseExpectation.fulfill() + try await env.firstCancelledExpectation.await(nanoseconds: timeout) + try await env.secondStartedExpectation.await(nanoseconds: timeout) + env.secondReleaseExpectation.fulfill() - let firstOutput = try await firstWaiter.value + await #expect(throws: Never.self, "first waiter requires to have a value") { + let firstOutput = try await firstWaiter.value + #expect(firstOutput == "replacement-output") + } + // let firstOutput = try await firstWaiter.value let secondOutput = try await secondWaiter.value - #expect(firstOutput == "replacement-output") #expect(secondOutput == "replacement-output") - #expect(await counter.count == 2, "switchToLatest should restart the active task") + #expect(await env.counter.count == 2, "switchToLatest should restart the active task") } } @Test func switchToLatestReplacementFailureCancelsAllWaiters() async throws { struct WorkerEnv: Sendable { - let counter: InvocationCounter - let firstStarted: Expectation - let firstCancelled: Expectation - let secondStarted: Expectation - let secondRelease: Expectation + let counter: InvocationCounter = InvocationCounter() + let firstStartedExpectation: Expectation = .init() + let firstCancelledExpectation: Expectation = .init() + let secondStartedExpectation: Expectation = .init() + let secondReleaseExpectation: Expectation = .init() let timeout: UInt64 } @@ -600,34 +591,34 @@ extension TaskSubscriptionTests { return task(id: "replaceable", option: .switchToLatest) { _, env in let invocation = await env.counter.increment() if invocation == 1 { - env.firstStarted.fulfill() + env.firstStartedExpectation.fulfill() do { try await Task.sleep(nanoseconds: env.timeout) return "stale-first-output" } catch is CancellationError { - env.firstCancelled.fulfill() + env.firstCancelledExpectation.fulfill() throw CancellationError() } } - env.secondStarted.fulfill() - try? await env.secondRelease.await(nanoseconds: env.timeout) + env.secondStartedExpectation.fulfill() + try? await env.secondReleaseExpectation.await(nanoseconds: env.timeout) throw ReplacementFailure.boom } case .second: return task(id: "replaceable", option: .switchToLatest) { _, env in let invocation = await env.counter.increment() if invocation == 1 { - env.firstStarted.fulfill() + env.firstStartedExpectation.fulfill() do { try await Task.sleep(nanoseconds: env.timeout) return "stale-first-output" } catch is CancellationError { - env.firstCancelled.fulfill() + env.firstCancelledExpectation.fulfill() throw CancellationError() } } - env.secondStarted.fulfill() - try? await env.secondRelease.await(nanoseconds: env.timeout) + env.secondStartedExpectation.fulfill() + try? await env.secondReleaseExpectation.await(nanoseconds: env.timeout) throw ReplacementFailure.boom } } @@ -636,26 +627,16 @@ extension TaskSubscriptionTests { static func output(state: State, event: Event) -> String { "" } } - let counter = InvocationCounter() - let firstStartedExpectation = Expectation() - let firstCancelledExpectation = Expectation() - let secondStartedExpectation = Expectation() - let secondReleaseExpectation = Expectation() let timeout: UInt64 = 5_000_000_000 var capturedInput: EffectViewInput? + + let env = WorkerEnv(timeout: timeout) try await testView(initialState: T.State()) { binding in EffectView( of: T.self, state: binding, - initialEnv: WorkerEnv( - counter: counter, - firstStarted: firstStartedExpectation, - firstCancelled: firstCancelledExpectation, - secondStarted: secondStartedExpectation, - secondRelease: secondReleaseExpectation, - timeout: timeout - ) + initialEnv: env ) { _, input in Color.clear.onAppear { capturedInput = input } } @@ -663,12 +644,12 @@ extension TaskSubscriptionTests { guard let input = capturedInput else { Issue.record("Input not captured"); return } let firstWaiter = Task { try await input.request(.first) } - try await firstStartedExpectation.await(nanoseconds: timeout) + try await env.firstStartedExpectation.await(nanoseconds: timeout) let secondWaiter = Task { try await input.request(.second) } - try await firstCancelledExpectation.await(nanoseconds: timeout) - try await secondStartedExpectation.await(nanoseconds: timeout) - secondReleaseExpectation.fulfill() + try await env.firstCancelledExpectation.await(nanoseconds: timeout) + try await env.secondStartedExpectation.await(nanoseconds: timeout) + env.secondReleaseExpectation.fulfill() do { _ = try await firstWaiter.value @@ -688,7 +669,7 @@ extension TaskSubscriptionTests { Issue.record("Unexpected second waiter error: \(error)") } - #expect(await counter.count == 2, "switchToLatest should restart the active task") + #expect(await env.counter.count == 2, "switchToLatest should restart the active task") } } @@ -702,7 +683,7 @@ extension TaskSubscriptionTests { static func update(_ state: inout State, event: Event) -> Effect? { switch event { case .load: - return request(id: nil, option: .subscribe) { _, _ in + return task(id: nil, option: .subscribe) { _, _ in "anonymous-output" } }