Removing cancellable from set crashes app

Question

I have a publisher, pipelinePublisher, which runs a combine pipeline of various operations, some of which send a state update to a statePublisher passed in as an argument. pipelinePublisher gets removed on completion of its Combine pipeline:

func myFunction(_ request: MyRequest) -> PassthroughSubject<State, Never> {
    let statePublisher = PassthroughSubject<State, Never>()
    let presentationSubject = CurrentValueSubject<MyRequest, Error>(request)

    var pipelinePublisher: AnyCancellable!

    pipelinePublisher = presentationSubject
      .eraseToAnyPublisher()
      .checkSomething(returningStateTo: statePublisher)
      // a few more operators here...
      .sink(
        receiveCompletion: { [weak self] _ in
          self?.cancellables.remove(pipelinePublisher) // Crash happens here
        },
        receiveValue: { _ in }
      )

    pipelinePublisher.store(in: &cancellables)

    return statePublisher
      .receive(on: RunLoop.main)
      .eraseToAnyPublisher()
  }

However, very occasionally when I call the function multiple times in very quick succession, the app crashes on the line self?.cancellables.remove(pipelinePublisher). This usually brings up one of two possible stack traces. This first is this:

2022-12-21 15:24:51.926131+0000 MyApp[23082:12933690] -[_NSCoreDataTaggedObjectID member:]: unrecognized selector sent to instance 0x8000000000000000
2022-12-21 15:24:51.931941+0000 MyApp[23082:12933690] *** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[_NSCoreDataTaggedObjectID member:]: unrecognized selector sent to instance 0x8000000000000000'
*** First throw call stack:
(
    0   CoreFoundation                      0x000000018040e7c8 __exceptionPreprocess + 172
    1   libobjc.A.dylib                     0x0000000180051144 objc_exception_throw + 56
    2   CoreFoundation                      0x000000018041d47c +[NSObject(NSObject) instanceMethodSignatureForSelector:] + 0
    3   CoreFoundation                      0x00000001804126c8 ___forwarding___ + 1308
    4   CoreFoundation                      0x0000000180414b4c _CF_forwarding_prep_0 + 92
    5   libswiftCore.dylib                  0x000000018be6ee68 $sSh8_VariantV6removeyxSgxF + 160
    6   MyApp               0x00000001026c6080 $s12MyApp0A0C17myFunctiony7Combine12AnyPublisherVyAA12StateOs5NeverOGAA19MyRequestVFyAE11SubscribersO10CompletionOy_s5Error_pGcfU_ + 440
    7   Combine                             0x000000019baa2a70 $s7Combine11SubscribersO4SinkC7receive10completionyAC10CompletionOy_q_G_tF + 364
    8   Combine                             0x000000019baa2f28 $s7Combine11SubscribersO4SinkCy_xq_GAA10SubscriberA2aGP7receive10completionyAC10CompletionOy_7FailureQzG_tFTW + 20
    9   Combine                             0x000000019bb541cc $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tF + 1516
    10  Combine                             0x000000019bb55328 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLCy_xq__qd__GAA10SubscriberA2aJP7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tFTW + 20
    11  Combine                             0x000000019bb53474 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC12receiveInner10completion_yAA11SubscribersO10CompletionOy_7FailureQzG_SitF + 1668
    12  Combine                             0x000000019bb52de4 $s7Combine10PublishersO7FlatMapV5Outer33_E91C3F00A6DFAAFEA2009FAF507AE039LLC4SideV7receive10completionyAA11SubscribersO10CompletionOy_7FailureQzG_tF + 20
    13  Combine                             0x000000019bac45ec $s7Combine6FutureC7Conduit33_3AE68DE9BADC00342FC052FEBC7D3BA6LLC7fulfillyys6ResultOyxq_GF + 1056
    14  Combine                             0x000000019bac4960 $s7Combine6FutureC7Conduit33_3AE68DE9BADC00342FC052FEBC7D3BA6LLC6finish10completionyAA11SubscribersO10CompletionOy_q_G_tF + 336
    15  Combine                             0x000000019bac2de4 $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GFyAA11ConduitBaseCyxq_GXEfU0_ + 156
    16  Combine                             0x000000019bac6b28 $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GFyAA11ConduitBaseCyxq_GXEfU0_TA + 16
    17  Combine                             0x000000019bae5140 $s7Combine11ConduitListO7forEachyyyAA0B4BaseCyxq_GKXEKF + 212
    18  Combine                             0x000000019bac2bfc $s7Combine6FutureC7promise33_3AE68DE9BADC00342FC052FEBC7D3BA6LLyys6ResultOyxq_GF + 716
    19  Combine                             0x000000019bac6b08 $s7Combine6FutureCyACyxq_Gyys6ResultOyxq_GcccfcyAGcfU_TA + 20
    20  MyApp               0x0000000102541dd8 $s7Combine6FutureC12MyApps5Error_pRs_rlE9operationACyxsAE_pGxyYaKc_tcfcyys6ResultOyxsAE_pGccfU_yyYaYbcfU_TY2_ + 212
    21  MyApp               0x0000000102542705 $s7Combine6FutureC12MyApps5Error_pRs_rlE9operationACyxsAE_pGxyYaKc_tcfcyys6ResultOyxsAE_pGccfU_yyYaYbcfU_TATQ0_ + 1
    22  MyApp               0x000000010242f1a1 $sxIeghHr_xs5Error_pIegHrzo_s8SendableRzs5NeverORs_r0_lTRTQ0_ + 1
    23  MyApp               0x000000010242f749 $sxIeghHr_xs5Error_pIegHrzo_s8SendableRzs5NeverORs_r0_lTRTA.24TQ0_ + 1
    24  libswift_Concurrency.dylib          0x00000001b03bedcd _ZL23completeTaskWithClosurePN5swift12AsyncContextEPNS_10SwiftErrorE + 1
)
libc++abi: terminating with uncaught exception of type NSException

The second is in the same place but with the error: *** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[__NSCFNumber member:]: unrecognized selector sent to instance 0x8000000000000000'

What is causing this? I've tried making pipelinePublisher optional and having a check that it exists before it gets removed but it does actually exist and still crashes. I can't figure this out, please help!

EDIT

I am calling myFunction(_:) by using the method foo which gets a publisher.

static func publisher(forParam: String) -> AnyPublisher<State, Never> {
    return Future {
      // Do some stuff here
      return objects
    }
    .flatMap { objects in
      let request = MyRequest(objects)
      return shared.myFunction(request)
    }
    .eraseToAnyPublisher()
  }

static func foo(
    param: String,
    handler: ((State) -> Void)? = nil
  ) {
    var cancellable: AnyCancellable!
    cancellable = publisher(forParam: param)
    .sink(
      receiveCompletion: { _ in
        self.shared.fooItems.cancellables.remove(cancellable) // sometimes crashes here too with the exact same crash!
      }, receiveValue: { state in
        handler?(state)
      }
    )

    cancellable.store(in: &shared.fooItems.cancellables)
  }

A few things to note:

• Foo has to use completion block as part of an API.
• publisher(forParam:) and therefore myFunction(_:) are only called from foo. However foo is called from many places.
• myFunction’s pipelinePublisher cancels by an error that’s thrown at the same time as sending a state and a completion through to foo.
• state can be sent to foo without a follow up completion too.

As you can see above, foo also sometimes gets the same error when removing the cancellable.

Answer 1

With help from people in the comments of my question, it looks like I had a race condition possibly caused by threads interacting with each other. I was removing from a Set, but Sets aren't thread safe. I solved this by changing the code to use Subscribers.Sink:

func myFunction(_ request: MyRequest) -> PassthroughSubject<State, Never> {
    let statePublisher = PassthroughSubject<State, Never>()
    let presentationSubject = CurrentValueSubject<MyRequest, Error>(request)

    var pipelinePublisher: AnyCancellable!

    pipelinePublisher = presentationSubject
      .eraseToAnyPublisher()
      .checkSomething(returningStateTo: statePublisher)
      // a few more operators here...
      .subscribe(Subscribers.Sink(
        receiveCompletion: { _ in },
        receiveValue: { _ in }
      ))

    pipelinePublisher.store(in: &cancellables)

    return statePublisher
      .receive(on: RunLoop.main)
      .eraseToAnyPublisher()
  }

This fixed the problem. Combine handles the clean up of the subscription and the subscriber when the pipeline returns.

This was taken from the answer given here: With Combine, how to deallocate the Subscription after a network request