Possible Memory Leak when observing shared state publisher in child of navigation stack

[wshamp]

The beta discussion has been closed so I am creating a new discussion here. It seems that when you are observing the publisher of a @shared property in a child of a navigation stack it sometimes causes the effect to be retained and you get the purple warning.

Steps to reproduce:

1. Have a root feature that is driven by an enum Reducer for multiple app states like launching or logged in, and also has a @shared property
2. In one of the states you have a navigation stack that has pushed a child that observes the @shared property from root
3. Root also observes an async stream from a client dependency that returns a Result with the same value type as the shared property ( think array of strings). When you get a success you mutate the @shared property with the result. This should update the child in the nav stack. This all works as intended up till now
4. The issue happens if you need to reset the root state and reobserve the Async Stream. There are many cases where this may happen ( like a user changed for example or logged out). In this case the we get the memory leak as the effect still sends an action in the child even though the child should have been deallocated and the effect canceled. Here is an example:

This is kind of verbose so here is a link to a project example
TCASharedStateMemoryLeak.zip

@Reducer
enum RootMode {
    case launching
    case parent(ParentFeature)
}
@Reducer
struct RootFeature {
    @Dependency(\.continuousClock) var clock
    @Dependency(StringsClient.self) var stringsClient
    @ObservableState
    struct State {
        var mode: RootMode.State? = .launching
        var elapsed: TimeInterval = 0.0
        @Shared(.strings) var strings = [String]()
    }
    enum Action {
        case mode(RootMode.Action)
        case timerTicked
        case stringsChangedFromAPI(Result<[String], Error>)
        case onTask
    }
    private enum CancelID {
        case timer
        case stringsObservation
    }
    var body : some ReducerOf<Self> {
        Reduce { state, action in
            switch action {
            case .onTask:
                var path = StackState<ParentFeature.Path.State>()
                path.append(.child(ChildFeature.State()))
                state.mode = .parent(ParentFeature.State(path: path))
                return .merge (
                    .run { send in
                        for await _ in self.clock.timer(interval: .seconds(1.0)) {
                            await send(.timerTicked)
                        }
                    }.cancellable(id: CancelID.timer, cancelInFlight: true),
                    .run { send in
                        
                        for try await strings in await stringsClient.observeStringsChange() {
                            
                            await send(.stringsChangedFromAPI(.success(strings)))
                        }
                    } catch: { error, send in
                        await send(.stringsChangedFromAPI(.failure(error)))
                    }
                        .cancellable(id: CancelID.stringsObservation, cancelInFlight: true))
            case .timerTicked:
                state.elapsed += 1.0
                print(state.elapsed)
                if state.elapsed == 5.0 {
                    state.mode = .parent(ParentFeature.State())
                    return .merge(
                        .cancel(id: CancelID.stringsObservation),
                        .run { send in
                            
                            for try await strings in await stringsClient.observeStringsChange() {
                                
                                await send(.stringsChangedFromAPI(.success(strings)))
                            }
                        } catch: { error, send in
                            await send(.stringsChangedFromAPI(.failure(error)))
                        }
                            .cancellable(id: CancelID.stringsObservation, cancelInFlight: true))
                } else {
                    return .none
                }
            case let .stringsChangedFromAPI(.success(newStrings)):
                return .run { @MainActor [strings = state.$strings, newStrings] _ in
                    strings.wrappedValue = newStrings
                }
            case .stringsChangedFromAPI:
                return .none
            case .mode:
                return .none
            }
        }
        .ifLet(\.mode, action: \.mode) {
            RootMode.body
        }
    }
}
struct RootView: View {
    @Bindable var store: StoreOf<RootFeature> = Store(initialState: RootFeature.State()) {
        RootFeature()
    }
    var body: some View {
        ZStack {
            if let store = store.scope(state: \.mode, action: \.mode) {
                switch store.case {
                case .launching:
                    Text("Launching")
                case let .parent(store):
                    ParentView(store: store)
                }
            }
        }
        .task {store.send(.onTask)}
    }
}
@Reducer
struct ParentFeature {
    @Reducer
    enum Path {
        case child(ChildFeature)
    }
    @ObservableState
    struct State {
        var path = StackState<Path.State>()
    }
    enum Action {
        case path(StackAction<Path.State, Path.Action>)
    }
    var body: some ReducerOf<Self> {
        Reduce { state, action in
            switch action {
            case .path:
                return .none
            }
        }
        .forEach(\.path, action: \.path)
    }
}
struct ParentView: View {
    @Bindable var store: StoreOf<ParentFeature>
    var body: some View {
        NavigationStack(path: $store.scope(state: \.path, action: \.path)) {
            Text("Parent")
        } destination: { store in
            switch store.case {
            case let .child(store):
                ChildView(store: store)
            }
        }
    }
}
@Reducer
struct ChildFeature {
    @ObservableState
    struct State {
        @Shared(.strings) var strings = [String]()
    }
    enum Action {
        case onTask
        case stringsChanged
    }
    private enum CancelID {
        case stringsChanged
    }
    var body: some ReducerOf<Self> {
        Reduce { state, action in
            switch action {
            case .onTask:
                return .publisher { [strings = state.$strings] in
                    strings.publisher.map{_ in Action.stringsChanged}.receive(on: DispatchQueue.main)
                }
                .cancellable(id: CancelID.stringsChanged, cancelInFlight: true)
            case .stringsChanged:
                print("Child Strings Changed")
                return .none
            }
        }
    }
}
struct ChildView: View {
    @Bindable var store: StoreOf<ChildFeature>
    var body: some View {
        Text("Child")
        .onAppear { store.send(.onTask) }
    }
}
extension PersistenceReaderKey where Self == InMemoryKey<[String]> {
  static var strings: Self {
      inMemory("strings")
  }
}
struct StringsClient: DependencyKey {
    var observeStringsChange: @Sendable () async -> AsyncThrowingStream<[String], Error>
    static var liveValue: StringsClient {
        return StringsClient (
            observeStringsChange: {
                return AsyncThrowingStream([String].self ) { continuation in
                    continuation.yield(["New String"])
                }
            }
        )
    }
}

[mbrandonw]

Hi @wshamp, I looked through the code and I'm a little confused by certain parts. You are starting up the stringsClient.observeStringsChange() subscription with every timer tick. Sure you are marking that effect as cancelInFlight and cancelling it here:

state.mode = .parent(ParentFeature.State())
return .merge(
  .cancel(id: CancelID.stringsObservation)
  .run { send in
    for try await strings in await stringsClient.observeStringsChange() {
      await send(.stringsChangedFromAPI(strings))
    }
  }
    .cancellable(id: CancelID.stringsObservation, cancelInFlight: true)
)

…but that still allows one final tick of the timer to go through, and further due to the re-entrancy problem I described previously, changing the mode with mode = .parent(…) is also not enough to cancel that effect.

However, there seems like a far simpler solution to this problem. Why not just start the stringsClient.observeStringsChange subscription in the onTask alone rather than repeatedly starting and cancelling the subscription? Doing that seems to fix the runtime warning for me:

case .onTask:
  var path = StackState<ParentFeature.Path.State>()
  path.append(.child(ChildFeature.State()))
  state.mode = .parent(ParentFeature.State(path: path))
  return .merge (
    .run { send in
      for await _ in self.clock.timer(interval: .seconds(1.0)) {
        await send(.timerTicked)
      }
    },
    .run { send in
      for try await strings in await stringsClient.observeStringsChange() {
        await send(.stringsChangedFromAPI(.success(strings)))
      }
    }
      .cancellable(id: CancelID.stringsObservation, cancelInFlight: true))
case .timerTicked:
  state.elapsed += 1.0
  print(state.elapsed)
  if state.elapsed == 1.0 {
    state.mode = .parent(ParentFeature.State())
    return .cancel(id: CancelID.stringsObservation)
  } else {
    return .none
  }

[wshamp]

So I am only re-observing when elapsed == 5.0 seconds so it should only happen once. I am just using the timer to simulate an event that would cause the need to reset root. This could be anything like a change in user or some other observed state change. this is a very rudimentary example to remove a lot of business logic from my actual app and narrow in on the specific issue. The reason to re-observe and not just cancel would be if a user change I would need to re-observe by passing a new user id to the api. This is critical that I can re-observe with new parameters when the user changes in my actual app.

So I guess I am confused why resetting mode is not enough to cancel the effect. Am I not using ifLet properly? I assumed that resetting would deallocate the child and the effect would get canceled. I force the observation to the man thread as you suggested and I am forcing the mutation of the @shared into an effect as you suggested earlier with :

return .run { @MainActor [strings = state.$strings, newStrings] _ in
   strings.wrappedValue = newStrings
}

I feel like I am missing something but not sure what.