TestTransport detaches Doctrine entities

[gndk]

I'm using a command/event bus architecture and followed your suggestion here to refactor my code to have a specific sync transport instead of using the default one. Now I am able to use the library as intended, which is great.

However, switching the transport in test environment from 'sync://' to 'test://?intercept=false&catch_exceptions=false', breaks a lot of my tests, mostly due to Doctrine\DBAL\Exception\UniqueConstraintViolationException and Doctrine\ORM\ORMInvalidArgumentException (related to ManyToOne etc cascades).

The breaking tests don't even use the messenger. It breaks when creating fixtures at the beginning of the test.

The DB is rolled back after each test with DAMADoctrineTestBundle.

I have custom repositories for my entities, which have all have a save() method like this:

// $this->getCartManager() is the EntityManager for this class

public function save(Cart $cart): void
{
    $this->getCartManager()->persist($cart);
    $this->getCartManager()->flush();
}

I create fixtures individually in each test with methods from my BaseIntegrationTest like this (simplified):

protected static function createCart(): Cart
{
    $cart = Cart::create(CartId::create());

    self::cartRepository()->save($cart);

    return $cart;
}

protected static function createCartItem(Cart $cart, int $quantity): CartItem
{
    $cartItem = CartItem::create(
        CartItemId::create(),
        $cart,
        $quantity,
    );

    self::cartRepository()->save($cart);

    return $cartItem;
}

protected static function cartRepository(): CartRepository
{
    return self::getService(CartRepository::class);
}

And use the fixtures in a test like this:

$cart = self::createCart();
$cartItem = self::createCartItem($cart, 1);
self::assertSomething(...);

The exceptions happen on the $cartItem = self::createCartItem($cart, 1); line, more specifically the second call to self::cartRepository()->save($cart);.

Doctrine\DBAL\Exception\UniqueConstraintViolationException : An exception occurred while executing a query: SQLSTATE[23505]: Unique violation: 7 ERROR: duplicate key value violates unique constraint "cart_pkey" DETAIL: Key (id)=(10000000-0000-0000-0000-000000000005) already exists.

Changing my repository save() method to use merge() instead of persist() solves the exception in some cases, but creates some other problems. Also, its deprecated.

I don't know if I need to change something in my code, or if this is a bug. Looking through the issues and PRs, this sounds like a similar issue: https://github.com/zenstruck/messenger-test/issues/9#issue-891484149

(I believe) When the TestTransport's are created and added to a static array, they, and all their deps are locked. When these dependencies are used in subsequent kernel's (after rebooting), they are not the same instances.

The discussion on doctrine/orm deprecating merge() also mentions message queues, but I don't know if its relevant in this case: https://github.com/doctrine/orm/issues/8461#issuecomment-809322698

If you need more information I can try to create a reproducer repo.

[kbond]

The breaking tests don't even use the messenger. It breaks when creating fixtures at the beginning of the test.

This is strange. So the creation of fixtures and the tests themselves don't use messenger? Do the tests pass when running them individually?

[gndk]

Yeah its really weird.

No, the fixtures don't use messenger. The tests do (indirectly, they are WebTestCases), but only later in the test (after the exception).

No, they also fail when running them individually.

I started working on a reproducer, but no luck so far :(

[gndk]

Still haven't managed to reproduce it in a Symfony skeleton application. Seems to be caused by some kind of side-effect.

But I found a solution to fix at least some of the failing tests by refactoring my fixtures a bit.

I'm now loading the object fresh from the repository before returning it.

So changing this:

protected static function createCart(): Cart
{
    $cart = Cart::create(CartId::create());

    self::cartRepository()->save($cart);

    return $cart;
}

to this:

protected static function createCart(): Cart
{
    $cart = Cart::create(CartId::create());

    self::cartRepository()->save($cart);

    return self::cartRepository()->get($cart->getId());
}

Still stumped why simply changing the transport from sync:// to test:// causes this, even when not using the messenger and not using the InteractsWithMessenger trait...

Will try some more refactoring and investigating tomorrow.

[kbond]

What's going on in that cartRepository() method? Is it somehow rebooting the kernel? Still, doesn't explain why test:// causes the problem. I do keep some static state in the TestTransport but as changed in #11, just some config and the messages now (no services).

Grasping at straws but do you have any doctrine event listeners that fire commands?

[gndk]

I have a small update. Found something, but haven't solved it yet.

On one of the failed tests that I couldn't solve otherwise, I added TestTransport::resetAll(); just before the line that throws an exception. Basically a shot in the dark.

This gave me a new error:

Undefined array key "sync"
 /srv/share/vendor/sentry/sentry/src/ErrorHandler.php:305
 /srv/share/vendor/zenstruck/messenger-test/src/Transport/TestTransport.php:277
 /srv/share/vendor/zenstruck/messenger-test/src/Transport/TestTransport.php:252
 /srv/share/vendor/symfony/messenger/Middleware/SendMessageMiddleware.php:67
 /srv/share/vendor/symfony/messenger/Middleware/FailedMessageProcessingMiddleware.php:34
 /srv/share/vendor/symfony/messenger/Middleware/DispatchAfterCurrentBusMiddleware.php:68
 /srv/share/vendor/symfony/messenger/Middleware/RejectRedeliveredMessageMiddleware.php:41
 /srv/share/vendor/symfony/messenger/Middleware/AddBusNameStampMiddleware.php:37
 /srv/share/vendor/symfony/messenger/Middleware/TraceableMiddleware.php:43
 /srv/share/vendor/symfony/messenger/MessageBus.php:73
 /srv/share/src/Domain/Messenger/MessengerEventBus.php:18
 /srv/share/src/Infrastructure/Doctrine/ORM/Repository/SimpleTagConfigurationRepository.php:106
 /srv/share/tests/Integration/BaseIntegrationTest.php:552
 /srv/share/tests/Integration/Shop/Checkout/Cart/Item/QuantityActionTest.php:57

sync is the name of the transport. This is my messenger.yaml config:

framework:
  messenger:
    reset_on_message: true
    failure_transport: failed
    default_bus: command.bus
    buses:
      command.bus:
        middleware:
          - doctrine_transaction
      event.bus:
        default_middleware: allow_no_handlers

    transports:
      sync: 'sync://'
      async: '%env(MESSENGER_ASYNC_TRANSPORT_DSN)%?queue_name=async'
      failed: '%env(MESSENGER_FAILED_TRANSPORT_DSN)%?queue_name=failed'

    routing:
      'App\Domain\Messenger\SyncMessage': sync
      'App\Domain\Messenger\AsyncMessage': async

And test/messenger.yaml:

framework:
  messenger:
    transports:
      sync: 'test://?intercept=false&catch_exceptions=false'
      async: 'test://?intercept=false&catch_exceptions=false'
      failed: 'test://?intercept=false&catch_exceptions=false'

Whats different about this one, is that the save() method dispatches an event to the event bus:

public function save(SimpleTagConfiguration $simpleTagConfiguration): void
{
    $this->getManager()->persist($simpleTagConfiguration);
    $this->getManager()->flush();

    $this->eventBus->dispatch(new SimpleTagConfigurationUpdated($simpleTagConfiguration->getId()));
}

That event implements the SyncMessage interface, which routes it to the sync transport:

<?php

declare(strict_types=1);

namespace App\Domain\Shop\ProductConfiguration\SimpleTag\Event;

use App\Domain\Messenger\Event;
use App\Domain\Messenger\SyncMessage;
use App\Domain\Shop\ProductConfiguration\SimpleTag\Model\SimpleTagConfigurationId;

final class SimpleTagConfigurationUpdated implements Event, SyncMessage
{
    private readonly string $simpleTagConfigurationId;

    public function __construct(SimpleTagConfigurationId $simpleTagConfigurationId)
    {
        $this->simpleTagConfigurationId = $simpleTagConfigurationId->toString();
    }

    public function getSimpleTagConfigurationId(): SimpleTagConfigurationId
    {
        return SimpleTagConfigurationId::fromString($this->simpleTagConfigurationId);
    }
}

This event has two listeners, but I deleted them and the test still fails. So its not related to that.

Removing the $this->eventBus->dispatch(...) line in the repository save() method makes the test pass.

This test also fails on fixture creation, like mentioned in an earlier comment, which in this case includes dispatching this event.

I tried adding the InteractsWithMessenger trait in this test, but it doesnt make a difference. Test still fails and TestTransport::resetAll() throws the same error.

Now that I narrowed it down, I'll try to reproduce this.

[kbond]

On one of the failed tests that I couldn't solve otherwise, I added TestTransport::resetAll(); just before the line that throws an exception. Basically a shot in the dark.

This makes sense, resetAll() clears the static state.

[gndk]

Oh yeah, you're right, makes sense.

But it lead me to the right spot. Managed to reproduce it :). Just checking a few last things.

[gndk]

Reproducer should be ok now, I tried to keep it as minimal as possible. Its based on Symfony skeleton and requires Docker for the DB. I added the steps in the readme: https://github.com/gndk/messenger-test-detach

[kbond]

Great, thank you. I'll play with your reproducer later today and try and find the root cause.

[gndk]

Just FYI I simplified the reproducer a bit more to reduce possible side effects.

Also added a second test case. The exceptions only occur if the $repository->save() method is used more than one time.

[kbond]

Found the culprit: https://github.com/symfony/symfony/blob/fe63f83daab08ecd1d44c33cb7e007014777cd78/src/Symfony/Bridge/Doctrine/Messenger/DoctrineClearEntityManagerWorkerSubscriber.php

This event subscriber clears the entity managers after the worker handles a message. TestTransport uses the worker to process messages (the sync transport does not - which is why it works).

In your reproducer, this solves the issue:

public function testSaveTwice(): void
{
+    $subscriber = self::getContainer()->get('doctrine.orm.messenger.event_subscriber.doctrine_clear_entity_manager');
+    self::getContainer()->get('event_dispatcher')->removeSubscriber($subscriber);

    $user = User::create(UserId::create());

    $this->userRepository()->save($user);

    self::assertNull($user->getName());

    $user->updateName('Some Name');

    $this->userRepository()->save($user);

    self::assertSame('Some Name', $user->getName());
}

Still thinking of how this should be handled. There doesn't look to be a way to disable this feature.

[gndk]

Great 🎉

I added the provisional fix in my BaseIntegrationTest setUp() method and can confirm that it fixes almost all of the failing test cases 👍 .

The remaining ones are structured a bit different:

1. Create fixtures (like in the reproducer test case)
2. Make a POST request using the WebTestCase KernelBrowser
3. Controller dispatches a command
4. Handler calls $repository->save() twice. Second call throws Doctrine\DBAL\Exception\UniqueConstraintViolationException

I'm will try to find out why this happens and add an additional test for it in the reproducer if I find something. Maybe the fix just needs to be applied in a different way.

[gndk]

I managed to reproduce it and added it to the reproducer repo with a new test.

It was actually a bit more complicated than I first thought:

1. KernelBrowser POST request to /usercreate/{userId}
2. UserCreateController dispatches CreateUser command
3. CreateUserHandler creates entity and persists to DB with $repository->save()
4. KernelBrowser POST request to /userupdate/{userId}
5. UserUpdateController dispatches UpdateUsercommand
6. UpdateUserHandler retrieves the entity created in step 3 with $repository->get() and calls $repository->save() on it twice . Second call throws Doctrine\DBAL\Exception\UniqueConstraintViolationException.

This reproducer case requires catch_exceptions=false to see the exception. Otherwise the test just fails because an assert fails.

[kbond]

Ok, this is because when making a request, you get a new container that doesn't have the DoctrineClearEntityManagerWorkerSubscriber removed from the event dispatcher.

This is not a final solution but shows what needs to be done in the reproducer to make it work:

1. In Kernel.php:

   class Kernel extends BaseKernel implements CompilerPassInterface
   {
       use MicroKernelTrait;
   
       public function process(ContainerBuilder $container): void
       {
           $container->removeDefinition('doctrine.orm.messenger.event_subscriber.doctrine_clear_entity_manager');
       }
   }

2. In UserTest::setUp(), comment out $subscriber = self::getContainer()->get('doctrine.orm.messenger.event_subscriber.doctrine_clear_entity_manager');

[kbond]

44 is a possible solution within this library. Before processing, it removes DoctrineClearEntityManagerWorkerSubscriber (if enabled), then adds it back after processing.

I tested on your reproducer and it works. @gndk, can you test it on your main app?

[gndk]

I tested your remove-subscribers branch on my main app. Unfortunately some errors remain after switching the transport. They all seem to have in common that there is a doctrine association going on in the entity that gets saved, with the exception being thrown for violating the unique constraint of the owning side primary key.

These are different errors than those with 1.4.1 + adding the fix in the setUp method of my BaseIntegrationTest. If I do that, I only get the errors that are related to the POST requests.

I don't get the errors related to the POST requests with the remove-subscribers branch.

I guess the best way forward would be to try to reproduce the new errors caused by the remove-subscribers branch. WDYT?

[kbond]

I guess the best way forward would be to try to reproduce the new errors caused by the remove-subscribers branch. WDYT?

I think so, yes. In your reproducer if possible.

[gndk]

Sorry, didn't have time to look into this more until now.

Tried for 2 hours to reproduce the new errors, but no luck so far.

Got frustrated and commented out //$manager->clear here: https://github.com/symfony/symfony/blob/fe63f83daab08ecd1d44c33cb7e007014777cd78/src/Symfony/Bridge/Doctrine/Messenger/DoctrineClearEntityManagerWorkerSubscriber.php#L54

This makes all tests pass. Obviously not a solution. It made me look at your fix and I played around with it.

The problem is the re-adding of the subscriber here: https://github.com/zenstruck/messenger-test/pull/44/files#diff-2a70c8135a1d976dab248ce14f954b1448722d8efdea6c123a6706399341718eR193

If I comment out this line (and uncomment //$manager->clear again), all tests also pass.

So maybe this re-adding needs to happen later, for example in tearDown()?

I don't know how exactly the Worker works. Maybe it re-adds while there is still a worker running. So maybe track all WorkerRunningEvent and only re-add the subscribers when there is a matching WorkerStoppedEvent. Similar to the listener for WorkerMessageFailedEvent here: https://github.com/zenstruck/messenger-test/pull/44/files#diff-2a70c8135a1d976dab248ce14f954b1448722d8efdea6c123a6706399341718eR172

I will try again to come up with a reproducer later today.

[kbond]

Strange, I wonder if it's adding it back in a way that prevents it from being removed again? I can't see how though... Are you able to inspect the event dispatcher after it's been removed to see if it's actually been removed again?

[gndk]

Are you able to inspect the event dispatcher after it's been removed to see if it's actually been removed again?

How would I do that? Happy to try, but don't know how.

[gndk]

Oh now I understand what you mean, I'll take a look

[kbond]

I was thinking dump($this->dispatcher->getListeners()) right before new Worker() and ensure the DoctrineClearEntityManagerWorkerSubscriber isn't in there.

[kbond]

To clarify, does https://github.com/zenstruck/messenger-test/issues/43#issuecomment-1113371380 fix all your issues (disabling that subscriber globally).

[gndk]

To clarify, does #43 (comment) fix all your issues (disabling that subscriber globally).

Yes, with 1.4.1 it fixes the issue. With fix branch I get a non-existent service error, as expected.

[gndk]

I was thinking dump($this->dispatcher->getListeners()) right before new Worker() and ensure the DoctrineClearEntityManagerWorkerSubscriber isn't in there.

I tried in different spots. At the beginning of process() its there, right before new Worker() its not there, and at the end back again.

[kbond]

Yes, with 1.4.1 it fixes the issue.

Ok, so we confirmed that subscriber is 100% the source of your issue.

I tried in different spots. At the beginning of process() its there, right before new Worker() its not there, and at the end back again.

Ok, that's what I would expect. This is the case on subsequent calls to process() as well? This is what I was wondering.

[gndk]

Ok, that's what I would expect. This is the case on subsequent calls to process() as well? This is what I was wondering.

Right now I'm not calling process() manually, because I'm using test://?intercept=false&catch_exceptions=false for the transport DSN.

I'll try calling it manually in a test and see it this changes something.

To give you some context: in the remaining failing cases there are a lot of events thrown around. Event -> Listener -> Triggers another Event -> Listener -> Triggers another Event. So my guess is that is has to do with the recursion of process().

[gndk]

Removing intercept=false from the DSN for the sync transport makes one of those new tests pass on the fix branch. But now a lot of other tests fail (because of wrong assertions, not exceptions). Guess that is expected, because the exceptions are thrown during processing.

I tried putting unblock() and intercept() around the method calls that cause the exception, but its the same result as with intercept=false.

[kbond]

So my guess is that is has to do with the recursion of process().

Yeah, but I'm not seeing how process could be called and have that subscriber.

Can you throw an exception here to try and see if the backtrace reveals anything?

[gndk]

Can you throw an exception here to try and see if the backtrace reveals anything?

RuntimeException
 /srv/share/vendor/symfony/doctrine-bridge/Messenger/DoctrineClearEntityManagerWorkerSubscriber.php:54
 /srv/share/vendor/symfony/doctrine-bridge/Messenger/DoctrineClearEntityManagerWorkerSubscriber.php:35
 /srv/share/vendor/symfony/event-dispatcher/Debug/WrappedListener.php:111
 /srv/share/vendor/symfony/event-dispatcher/EventDispatcher.php:230
 /srv/share/vendor/symfony/event-dispatcher/EventDispatcher.php:59
 /srv/share/vendor/symfony/event-dispatcher/Debug/TraceableEventDispatcher.php:152
 /srv/share/vendor/symfony/messenger/Worker.php:276
 /srv/share/vendor/symfony/messenger/Worker.php:205
 /srv/share/vendor/symfony/messenger/Worker.php:170
 /srv/share/vendor/symfony/messenger/Worker.php:106
 /srv/share/vendor/zenstruck/messenger-test/src/Transport/TestTransport.php:180
 /srv/share/vendor/zenstruck/messenger-test/src/Transport/TestTransport.php:275
 /srv/share/vendor/symfony/messenger/Middleware/SendMessageMiddleware.php:67
 /srv/share/vendor/symfony/messenger/Middleware/FailedMessageProcessingMiddleware.php:34
 /srv/share/vendor/symfony/messenger/Middleware/DispatchAfterCurrentBusMiddleware.php:68
 /srv/share/vendor/symfony/messenger/Middleware/RejectRedeliveredMessageMiddleware.php:41
 /srv/share/vendor/symfony/messenger/Middleware/AddBusNameStampMiddleware.php:37
 /srv/share/vendor/symfony/messenger/Middleware/TraceableMiddleware.php:43
 /srv/share/vendor/symfony/messenger/MessageBus.php:73
 /srv/share/src/Domain/Messenger/MessengerEventBus.php:18
 /srv/share/src/Infrastructure/Doctrine/ORM/Repository/OrderRepository.php:178
 /srv/share/tests/Integration/BaseIntegrationTest.php:587
 /srv/share/tests/Integration/BaseIntegrationTest.php:605
 /srv/share/tests/Integration/Admin/OrderViewTest.php:19

[kbond]

Ok, so it's clearly not being properly removed or re-added in a way that it can't be removed again on subsequent calls to process().

Right now I'm not calling process() manually, because I'm using test://?intercept=false&catch_exceptions=false for the transport DSN.

With the dump()'s you put in earlier, does it always show as removed when it should be, even when it's called recursively?

[gndk]

With the dump()'s you put in earlier, does it always show as removed when it should be, even when it's called recursively?

I'll try to put in a logging method to track the dump of each call.

[kbond]

Can you try this:

# config/packages/test/messenger.yaml

framework:
    messenger:
        transports:
            sync: test://?intercept=false&catch_exceptions=false

services:
    doctrine.orm.messenger.event_subscriber.doctrine_clear_entity_manager:
        class: stdClass # effectively disables this service in your test env

I think this might be the easiest work-around.

[gndk]

Can you try this

Works as a work-around 👍

I'll still see if I can get it reproduced.

[kbond]

Ok, if you can, that's great and we can diagnose further, but I'm thinking I might just document this workaround if you can't.

Additionally, I think this subscriber should actually be a "service resetter", not a messenger subscriber. Any long running process should clear the em's - not just messenger:consume. If this change is made, it would fix the issue (as this lib doesn't reset services during process).

Anyway, I'm going to propose this to Symfony/Doctrine.

[gndk]

Sounds like a better approach 👍 Thanks for your quick responses and help here! Looking forward to use the lib properly now 🙂