Tweak the way that effects work so that they can wait for the completion of child effects sent with send

[jchitel]

There is something unexpected that I recently discovered about the EffectTask.run { send in ... } API.

I originally noticed that send() can be awaited, which excited me because I thought that meant that effects can wait for those actions to complete. However, when I read through the source code, I discovered that the Send type is only awaitable because it has @MainActor. The actual send function that the Store provides to effects is synchronous, and internally it holds onto a reference to any task returned from the child effect, but doesn't await the task.

After further review of the source code (btw, I think the source code needs slightly more comments to explain some of the less obvious aspects of the implementation), I can't actually think of any logical reason why the send function can't await the task.

I'll try to illustrate my point with an example:

struct MyReducer: Reducer {
    struct State: Equatable {
        var done: Bool
    }

    enum Action: Equatable {
        case first
        case second
        case third
    }

    func reduce(state: inout State, action: Action) -> EffectTask<Action> {
        switch action {
        case .first: return .run { send in
            print("Started first effect")
            await Task.sleep(1)
            print("Sending second action")
            await send(.second)
            print("Sent second action")
            await Task.sleep(2)
            print("Finishing first effect")
        }
        case .second: return .run { send in
            print("Started second effect")
            await Task.sleep(3)
            print("Sending third action")
            await send(.third)
            print("Sent third action")
            await Task.sleep(4)
            print("Finishing second effect")
        }
        case .third:
            print("Executing third action")
            state.done = true
            return .none
        }
    }
}

On sending .first, I would want the order of prints to go like this:

Started first effect
-- 1 second
Sending second action
  Started second effect
  -- 3 seconds
  Sending third action
    Executing third action
  Sent third action
  -- 4 seconds
  Finishing second effect
Sent second action
-- 2 seconds
Finishing first effect

Based on my understanding of the source code, however, I believe it would go like this:

Started first effect
-- 1 second
Sending second action
Sent second action
  Started second effect -- first effect is sleeping for 2 seconds
-- 2 seconds
Finishing first effect
  -- 1 second (3 minus 2)
  Sending third action
    Executing third action -- non-effectful actions execute synchronously
  Sent third action
  -- 4 seconds
  Finishing second effect

Basically, if an effect sends an action that returns another effect, there is no way for the parent effect to wait on the child effect, but I don't see a limitation necessarily preventing this. Since effects are executed in a task within the main actor, they won't start executing until after the synchronous parts of store.send() are done. Thus, any child effects will always be new "top-level" actions, so there won't be any logical difference between the first effect and the second effect from the perspective of store.send().

The way that I see it, the following code changes would have to be made:

.run effect handler in store.send()

case let .run(priority, operation):
  tasks.wrappedValue.append(
    Task(priority: priority) {
      await operation(
        Send {
          if let task = self.send($0, originatingFrom: action) {
            tasks.wrappedValue.append(task)
+          await task // await task so the effect doesn't resume until the child effect is complete
          }
        }
      )
    }
  )

Returned task at the bottom of store.send()

return Task {
  await withTaskCancellationHandler {
+   await withTaskGroup(of: Void.self) { group in
      var index = tasks.wrappedValue.startIndex
      while index < tasks.wrappedValue.endIndex {
        defer { index += 1 }
-       await tasks.wrappedValue[index].value
+       group.addTask(tasks.wrappedValue[index]) // wait for tasks in parallel instead of sequentially
      }
+   }
  } onCancel: {
    var index = tasks.wrappedValue.startIndex
    while index < tasks.wrappedValue.endIndex {
      defer { index += 1 }
      tasks.wrappedValue[index].cancel()
    }
  }
}

This latter change is probably insufficient because it doesn't properly handle awaiting all child tasks, but there are certainly ways to handle that. Perhaps instead of returning Void, each task could return a [Task] with all child tasks. You'd need to introduce another type because the types of those tasks would be recursive. The aggregate task would create a task group where each task within the group waits for an effect task, then adds all returned tasks to the group, and so on until everything has been awaited.

You'd also need to introduce something like ViewStoreTask as a return type of Send that would make awaiting the action optional.

This sounds perfectly doable to me. Please let me know if there is anything I'm missing here. Thanks!

[mbrandonw]

Hi @jchitel, thanks for starting this discussion. The topic has come up once or twice before.

I don't think the behavior of send(action) should always be changed to suspending for the duration of action's effects. In fact, I kinda think that may be a minority use case. Just consider a timer:

return .run { send in 
  for await _ in self.clock.timer(interval: .seconds(1)) {
    await send(.timerTicked)
  }
}

It'd be a little strange if the timer was essentially paused just because timerTick was doing a lot of async work.

But, having said that, I do think there are probably some really interesting and powerful use cases for tying the suspension of send to the action's effect's lifetime. We would probably just want a different syntax for it.

I think we can even borrow the syntax from viewStore.send, which returns a ViewStoreTask that represents the lifetime of the effects that are started from the action:

await viewStore.send(action).finish()

So, maybe we could have Effect's send return a kind of "task" to represent the effect lifetime, and that would allow you to do things like:

return .run { send in 
  // Suspend until the effects from .started fully complete.
  await send(
    .authorizationResponse(self.speechClient.requestAuthorization())
  )
  .finish()
  // ...
}

Heck... it might even let you cancel the actions later too 🤔:

return .run { send in 
  // Suspend until the effects from .started fully complete.
  let task = await send(
    .authorizationResponse(self.speechClient.requestAuthorization())
  )

  // Cancel any effects from the auth response if they haven't finished in 2 seconds:
  try await self.clock.sleep(for: .seconds(2))
  task.cancel()
}

Lots to think about! We don't personally have time to explore this right now, but if you want to start a branch with some explorations we'd love to take a look. And especially so if you have some interesting use cases for the send lifetime behavior you are describing.

[mbrandonw]

Oh, and one important note to make.

All of the sends in the library are synchronous. This includes Store.send, ViewStore.send, and Effect.run { send ... }. (Actually, an exception is TestStore.send, but that's only because we do extra work to yield to executing effects due to lack of proper async testing tools.)

And the reason for this is an important part of the architecture. When actions are sent we want to immediately and synchronously perform the state mutations, and then separately start up the effects. This allows you to do things like:

withAnimation {
  viewStore.send(.action)
}

…and it just works. If we did not process state mutations synchronously, then things like that would not be possible.

[jchitel]

Excellent! I'm glad I wasn't missing something, it seems that it's just a feature that hasn't been requested yet.

Something like ViewStoreTask was exactly what I had in mind. I will definitely make a branch and try this out.

And yes, in my explorations of the source code, I did see that send is synchronous. That's something I'll definitely keep in mind.

[kabiroberai]

I'm playing around with this atm and the implementation is mostly simple except for one fundamental issue: Effects don't have to emit actions, and they can be observed by things that aren't stores. Consider the following example:

enum Action { case doSomething, doOtherThing }
let effect = EffectTask<Action>.run { send in
  await send(.doSomething).finish()
  await send(.doOtherThing)
}
effect.sink { nextAction in print(nextAction) }

In the above case there's no semantic meaning to waiting for doSomething to "finish" because there's no Store processing the action. We end up with this issue any time the Effect is treated as a Publisher since the flow of information is/should be one-way, and so send(...).finish() only makes sense in a world where the EffectTask.run case is directly being consumed by a Store (since it can then define the meaning of a sent action "finishing").

With reference to the TCA codebase, the exact issue is that we can't get any info back from the subscriber here:

swift-composable-architecture/Sources/ComposableArchitecture/Effects/Publisher.swift

Line 26 in b32b1ac

let send = Send { subscriber.send($0) }

whereas we have a lot more information in the Store case here

swift-composable-architecture/Sources/ComposableArchitecture/Store.swift

Line 423 in b32b1ac

if let task = self.send($0, originatingFrom: action) {

The only solutions I can think of rn are 1) removing Publisher support entirely, which would break tons of stuff, or 2) performing some sort of reference type spooky action at a distance such that if the action does eventually trickle up to a Store, it can populate a completion field to communicate back. Neither seems ideal; wondering if anyone else has better ideas.

[mbrandonw]

Hey @kabiroberai, thanks for looking into this! This exact topic just came up today on a Slack of someone wanting to do this.

I haven't chatted about this with Stephen yet, but my first reaction (and I could be wrong) is that it may not be a big deal for this functionality to only work in effects tied to stores. We honestly don't see a use case for Effect outside of TCA, and it may be fine to document it as such.

Do you have any scratch work you could share?

[kabiroberai]

Just put up a PR! See #1895.

[dvosscricut]

This could work if you really need the wait for the prior effect to finish before doing something else functionality:

Effect.run { send in 
  var cancellable: AnyCancellable?
  await withCheckedContinuation { continuation in
      let publisher = _EffectPublisher(effectWeWantToWaitForToFinishBeforeProceeding)
      cancellable = publisher
          .sink(
              receiveCompletion: { _ in
                  continuation.resume()
              },
              receiveValue: { action in
                  // You can also use an AsyncStream here to enforce sequential actions and observe it elsewhere in the reducer but just as a quick example using Task (not guaranteed sequential but it kinda works for example purposes).
                  Task {
                      await send(action)
                  }
              }
          )
  }
  _ = cancellable // No op to remove warning about cancellable being written to but never read. Cancellable var is needed to keep publisher alive and prevent from terminating.

  await send(theEffectWeWantToHappenSequentiallyAfterThePriorOneFinishes)
}