Queries in EventBus.ofType subscription failing with QueryRunnerAlreadyReleasedError

[chladog]

Describe the bug

onVendureBootstrap(): void | Promise<void> {
    this.eventBus.ofType(OrderStateTransitionEvent).pipe(filter(e => e.toState === 'PaymentAuthorized'))
      .subscribe(async e => {
        await this.orderService.transitionToState(e.ctx, e.order.id, 'NextState' as OrderState);
      });
  }

Sometimes results in UnhandledPromiseRejectionWarning: QueryRunnerAlreadyReleasedError: Query runner already released. Cannot run queries anymore

Findings and discussion from Slack:
chladog 14 hours ago
The transaction gets closed when addPaymentToOrder is finished and won't wait for any actions I'm trying to do in eventbus.
I believe the Subject of eventbus is async function.
When I see all this. I believe eventbus should be async. Stream of events is useful when we try to react to and do internal actions - and should be different context from each API request ctx/transaction.
I fixed this temporarily for my case by introducing timeout in eventbus subscription - the transaction will complete and then new transaction is created. I didn't find any other way to start new independent transaction.

Michael Bromley 12 hours ago
Right, it all makes sense now, thanks for digging in.
So, the EventBus is sync in the sense that any subscribers are called synchronously in the same VM tick as when the event is fired. However, the method which fires the event has no knowledge of the subscriber function or when it is complete. So as soon as your subscriber function does something async (like awaiting a DB call) then the outer resolver will continue to execute and complete.
I'm not sure of the best way to solve this yet. Of course, making the EventBus async with a timeout-like operator will work, but I think that's also just asking for horrible-to-debug race conditions (e.g. if the resolver function for some reason is slower to resolve than the timeout).
So one solution might be to transparently remove the reference to the transactional queryRunner inside the EventBus subscription mechanism. The downside there is that, if you are relying on being able to read data that has been created/modified within the transaction, you won't be able to due to the isolation of the transaction.
It's quite a hard problem, I need to think about this some more...

chladog 9 minutes ago
Yep.
The timeout is really just quickfix for current issue as I don't mind transitioning to next state with delay, but for some functionality it might be problem.
If you can fix the main issue here "outer resolver will continue to execute and complete" - so that we are able to join the transaction from within eventbus (without being prematurely terminated) then I still think we want 2 choices when dealing with events - one that would join the causing transaction and another that would be parted from the causing transaction altogether (from my POV latter should be default).

Expected behavior
1.) Possibility to join the causing transaction from within eventbus subscription without being prematurely terminated => e.g. process of eventbus subscription function is needed for correct response.
2.) Possibility to part the actions of eventbus subscription from the causing transaction altogether (not blocking resolver) => e.g. process of eventbus subscription function is not needed for correct response.

Environment (please complete the following information):

• @vendure/core version: v16.1

[michaelbromley]

1.) Possibility to join the causing transaction from within eventbus subscription without being prematurely terminated => e.g. process of eventbus subscription function is needed for correct response.

So this would essentially make the event handler functions blocking with regards to the code path that publishes the event. I am against this idea because (as far as I understand it) it would be highly unexpected for an Observer pattern / pub-sub system to work this way. On top of that, I cannot even think of how to implement something like that in the context of RxJS (which is what the EventBus is built on).

2.) Possibility to part the actions of eventbus subscription from the causing transaction altogether (not blocking resolver) => e.g. process of eventbus subscription function is not needed for correct response.

This is what I mentioned earlier in the chat. If we just remove the internal reference to the transaction EntityManager from the ctx object, this will be the result.
The trade-off as noted is that data updated within the transaction would not be accessible until the transaction completes. However, I believe this would be a relative edge-case and there are work-arounds.

I'll put together some failing tests and then attempt solution 2.

[devtony10]

TypeError: this.eventBus.ofType is not a function. Please what's happening? I can't ping my server from storefront with this error