Publisher crash in 1.1

[yonaskolb]

Summary

I don't have too many details, but we're seeing a crash in our Combine Publisher integration with Apollo in 1.1.0. It's based off this code https://github.com/joel-perry/ApolloCombine

Not really sure what's going on yet, but thought I'd open this issue early.

A colleague shared this image where the errors array was completely filled with the same error, but don't know if the debugger is acting up as I didn't see that 🤷‍♂️

My best guess is that something changed in a breaking way in the multipart subscriptions work, but that's a total guess. #2870

Version

1.1.0

Steps to reproduce the behavior

• make a query through a publisher that fails
• observe crash

Logs

lldb spits out this on `po result`

▿ Result<GraphQLResult<Data>, Error>
  ▿ success : GraphQLResult<Data>
    - data : nil
    ▿ errors : Optional<Array<GraphQLError>>
      ▿ some : 1 element
        ▿ 0 : Unauthorized
          ▿ object : 2 elements
            ▿ 0 : 2 elements
              - key : "message"
              ▿ value : AnyHashable("Unauthorized")
                - value : "Unauthorized"
            ▿ 1 : 2 elements
              - key : "extensions"
              ▿ value : AnyHashable([AnyHashable("code"): AnyHashable("UNAUTHENTICATED")])
                ▿ value : 1 element
                  ▿ 0 : 2 elements
                    ▿ key : AnyHashable("code")
                      - value : "code"
                    ▿ value : AnyHashable("UNAUTHENTICATED")
                      - value : "UNAUTHENTICATED"
    - extensions : nil
    - source : Apollo.GraphQLResult<ShopAPI.FetchOrdersQuery.Data>.Source.server
    - dependentKeys : nil

Anything else?

Our Apollo Publisher code if something sticks out as incorrect when integrating with Apollo 1.1

import Apollo
import Combine
import Foundation

public extension ApolloClientProtocol {
    func fetchPublisher<Query: GraphQLQuery>(query: Query,
                                             cachePolicy: CachePolicy = .default,
                                             contextIdentifier: UUID? = nil,
                                             queue: DispatchQueue = .main) -> Publishers.ApolloFetch<Query> {
        let config = Publishers.ApolloFetchConfiguration(
            client: self,
            query: query,
            cachePolicy: cachePolicy,
            contextIdentifier: contextIdentifier,
            queue: queue
        )
        return Publishers.ApolloFetch(with: config)
    }
    
    // Performs a mutation by sending it to the server.
    ///
    /// - Parameters:
    ///   - mutation: The mutation to perform.
    ///   - publishResultToStore: If `true`, this will publish the result returned from the operation to the cache store. Defaults to `true`.
    ///   - queue: A dispatch queue on which the result handler will be called. Defaults to the main queue.
    /// - Returns: A publisher that delivers results from the perform operation.
    func performPublisher<Mutation: GraphQLMutation>(mutation: Mutation,
                                                     publishResultToStore: Bool = true,
                                                     queue: DispatchQueue = .main) -> Publishers.ApolloPerform<Mutation> {
        let config = Publishers.ApolloPerformConfiguration(client: self, mutation: mutation, publishResultToStore: publishResultToStore, queue: queue)
        return Publishers.ApolloPerform(with: config)
    }
}

public extension Publishers {
    // MARK: Fetch
    
    struct ApolloFetch<Query: GraphQLQuery>: Publisher {
        public typealias Output = GraphQLResult<Query.Data>
        public typealias Failure = Error
        
        private let configuration: ApolloFetchConfiguration<Query>
        
        public init(with configuration: ApolloFetchConfiguration<Query>) {
            self.configuration = configuration
        }
        
        public func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
            let subscription = ApolloFetchSubscription(subscriber: subscriber, configuration: configuration)
            subscriber.receive(subscription: subscription)
        }
    }
    
    struct ApolloFetchConfiguration<Query: GraphQLQuery> {
        let client: ApolloClientProtocol
        let query: Query
        let cachePolicy: CachePolicy
        let contextIdentifier: UUID?
        let queue: DispatchQueue
    }
    
    private final class ApolloFetchSubscription<S: Subscriber, Query: GraphQLQuery>: Subscription
    where S.Failure == Error, S.Input == GraphQLResult<Query.Data> {
        private let configuration: ApolloFetchConfiguration<Query>
        var subscriber: S?
        private var task: Apollo.Cancellable?
        
        init(subscriber: S?, configuration: ApolloFetchConfiguration<Query>) {
            self.subscriber = subscriber
            self.configuration = configuration
        }
        
        func request(_ demand: Subscribers.Demand) {

            task = configuration.client.fetch(query: configuration.query,
                                              cachePolicy: configuration.cachePolicy,
                                              contextIdentifier: configuration.contextIdentifier,
                                              queue: configuration.queue) { [weak self] result in
                switch result {
                case .success(let data):
                    _ = self?.subscriber?.receive(data)
                    
                    if self?.configuration.cachePolicy == .returnCacheDataAndFetch && data.source == .cache {
                        return
                    }
                    self?.subscriber?.receive(completion: .finished)
                    
                case .failure(let error):
                    debugPrint(error)
                    self?.subscriber?.receive(completion: .failure(error))
                }
            }
        }
        
        func cancel() {
            task?.cancel()
            task = nil
            subscriber = nil
        }
    }
    
    // MARK: - Perform
    
    struct ApolloPerform<Mutation: GraphQLMutation>: Publisher {
        public typealias Output = GraphQLResult<Mutation.Data>
        public typealias Failure = Error
        
        private let configuration: ApolloPerformConfiguration<Mutation>
        
        public init(with configuration: ApolloPerformConfiguration<Mutation>) {
            self.configuration = configuration
        }
        
        public func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
            let subscription = ApolloPerformSubscription(subscriber: subscriber, configuration: configuration)
            subscriber.receive(subscription: subscription)
        }
    }
    
    struct ApolloPerformConfiguration<Mutation: GraphQLMutation> {
        let client: ApolloClientProtocol
        let mutation: Mutation
        let publishResultToStore: Bool
        let queue: DispatchQueue
    }
    
    private final class ApolloPerformSubscription<S: Subscriber, Mutation: GraphQLMutation>: Subscription
    where S.Failure == Error, S.Input == GraphQLResult<Mutation.Data> {
        private let configuration: ApolloPerformConfiguration<Mutation>
        private var subscriber: S?
        private var task: Apollo.Cancellable?
        
        init(subscriber: S, configuration: ApolloPerformConfiguration<Mutation>) {
            self.subscriber = subscriber
            self.configuration = configuration
        }
        
        func request(_ demand: Subscribers.Demand) {
            task = configuration.client.perform(
                mutation: configuration.mutation,
                publishResultToStore: configuration.publishResultToStore,
                queue: configuration.queue
            ) { [weak self] result in
                guard let self else { return }
                switch result {
                case .success(let data):
                    if let errorMessage = data.errors?.first?.message {
                        print("[GraphQL Error] \(data.errorString.orElseEmpty)")
                        self.subscriber?.receive(completion: .failure(ShopError.graphqlError(reason: errorMessage)))
                        return
                    }
                    _ = self.subscriber?.receive(data)
                    self.subscriber?.receive(completion: .finished)
                    
                case .failure(let error):
                    self.subscriber?.receive(completion: .failure(error))
                }
            }
        }
        
        func cancel() {
            task?.cancel()
            task = nil
            subscriber = nil
        }
    }
}

[AnthonyMDev]

@calvincestari Any initial thoughts on what could be causing this? If not, lets start putting together a reproduction case!

[calvincestari]

Hmm. I don't have any idea what could be the cause right now, we'll need to get a reproduction case so we can properly debug it.

My best guess is that something changed in a breaking way in the multipart subscriptions work, but that's a total guess. #2870

We had to jump through a few hoops with the interceptor chain to introduce the multipart subscriptions in a way that wasn't breaking for existing users; the description of #2870 describes the implementation details of that. I don't think anything changed about the Cancellable that is returned though requests though. That error makes me think it's maybe something to do with the managedSelf object of the request chain. I highly doubt it's in the MultipartResponseParsingInterceptor as that only reacts to a specific response header value.

[calvincestari]

Hi @yonaskolb, we've got a fix up on branch fix/managed-self-release. The test we've written fails/passes as expected but given that you're using Combine would you mind testing the fix to ensure it resolves the issue for you too. If it does we can get a release out fairly quickly.

[yonaskolb]

Yes, this seems to fix the error in our case, thank you! 🙏

[mariano-mix]

Just coming here because we were getting similar crash but while using the new Swift concurrency APIs. In our case, we had a wrapper around the provided fetch method to return an AsyncThrowingStream looking like this

    AsyncThrowingStream { continuation in
      let request = fetch(...) { response in
        switch response {
        case let .success(result):
          continuation.yield(result)
          if result.isFinalForCachePolicy(cachePolicy) {
            continuation.finish() // This line was crashing
          }
        case let .failure(error):
          continuation.finish(throwing: error)
        }
      }
      continuation.onTermination = { @Sendable _ in request.cancel() }
    }

The app was crashing at continuation.finish() when errors were thrown at call site like this

for try await result in fetch(query: SomeQuery()) {
   if let graphQLError = result.errors?.first {
       throw SomeError // This error was causing `onTermination` getting called
   }
}

The issue was due to NSLock being sent a message after being deallocated. Pointing to this branch alone fixed it.
I was also being able to fix this pointing to the main branch by either holding onto the Cancellable returned by the provided fetch or by reading the continuation termination and only invoking request.cancel() when it matched the cancelled case. In the latter, I tried cancelling the task, verified that the cancel method was called and there was no crash.

Hope this helps someone else too.

[calvincestari]

The issue was due to NSLock being sent a message after being deallocated. Pointing to this branch alone fixed it.
I was also being able to fix this pointing to the main branch by either holding onto the Cancellable returned by the provided fetch or by reading the continuation termination and only invoking request.cancel() when it matched the cancelled case. In the latter, I tried cancelling the task, verified that the cancel method was called and there was no crash.

Hi @mariano-mix 👋🏻 - as you've verified, PR #2943 should resolve that issue too since the onTermination call to cancel() should now become a no-op after the fetch ends normally.

[calvincestari]

Thanks for the confirmation too @yonaskolb. I'll get the PR merged and 1.1.2 out in the next day (hoping to squeeze another fix or two into it too).

[BrentMifsud]

were seeing something similar on our end with RxSwift

[calvincestari]

@BrentMifsud the fix is merged into main. Please try that and let me know if it's resolved for you.

[BrentMifsud]

ill try it out

[BrentMifsud]

can confirm this fixes the issue.

[calvincestari]

OK, thanks for confirming. It'll be included with 1.1.2 soon.

[BrentMifsud]

Just a heads up, one thing I just noticed is that the install CLI option seems to be missing when right clicking on my project after switching to the main branch to test that fix.

Not sure if thats related to any of the changes, or if its just Xcode being Xcode.

Edit:

this is probably a false alarm. I just switched back to 1.1.1 and it's still not there. Probably just an Xcode 14.3 bug.

[calvincestari]

Yup, we're tracking that in #2919 for now.