SubscriptionEngine endless loop with in memory stores

[ergosarapu]

Setting both, event_store and subscription store to 'in_memory' causes the SubscriptionEngine to loop forever when producing events from the Processor. This is especially blocking problem when not running tests against real database.

// config/packages/patchlevel_event_sourcing.yaml

when@test:
    patchlevel_event_sourcing:
        store:
            type: in_memory
        subscription:
            catch_up: true
            throw_on_error: true
            run_after_aggregate_save: true
            store:
                type: in_memory

// src/Hotel/Application/Processor/AutoCheckinProcessor.php

#[Processor('auto_checkin')]
class AutoCheckinProcessor
{
    /**
     * @param Repository<Hotel> $hotelRepository
     */
    public function __construct(private readonly Repository $hotelRepository)
    {
    }

    #[Subscribe(HotelCreated::class)]
    public function onHotelCreated(HotelCreated $event): void
    {
        $hotel = $this->hotelRepository->load($event->hotelId);
        $hotel->checkIn('First Guest');
        $this->hotelRepository->save($hotel);
    }

    #[Subscribe(GuestIsCheckedIn::class)]
    public function onGuestIsCheckedIn(GuestIsCheckedIn $event): void
    {
        // Never reaches here!
        $hotel = $this->hotelRepository->load($event->hotelId);
        $hotel->checkOut('First Guest');
        $this->hotelRepository->save($hotel);
    }
}

// tests/InMemoryTest.php

class InMemoryTest extends KernelTestCase
{

    protected function setUp(): void
    {
        parent::setUp();
        $subscriptionEngine = static::getContainer()->get(SubscriptionEngine::class);
        $subscriptionEngine->setup();
        $subscriptionEngine->boot();
    }
    public function testInMemoryFunctionality()
    {
        $repository = $this->getContainer()->get('test.event_sourcing.hotel.repository');
        $hotel = Hotel::create(Uuid::generate(), 'Test Hotel');
        $repository->save($hotel);
        $this->assertTrue(true);
    }
}

Log output:

[2025-11-18T13:34:48.948601+02:00] event_sourcing.INFO: Subscription Engine: Start to setup. [] []
[2025-11-18T13:34:48.954379+02:00] event_sourcing.INFO: Subscription Engine: New Subscriber "auto_checkin" was found and added to the subscription store. [] []
[2025-11-18T13:34:48.954510+02:00] event_sourcing.INFO: Subscription Engine: No subscriptions to setup, finish setup. [] []
[2025-11-18T13:34:48.954555+02:00] event_sourcing.INFO: Subscription Engine: Start booting. [] []
[2025-11-18T13:34:48.954572+02:00] event_sourcing.INFO: Subscription Engine: No subscriptions in booting status, finish booting. [] []
[2025-11-18T13:34:48.959266+02:00] event_sourcing.DEBUG: Repository: Aggregate "hotel" with the id "019a96be-f77e-7063-a1d6-943da2d4f06b" saved. [] []
[2025-11-18T13:34:48.959296+02:00] event_sourcing.INFO: Subscription Engine: Start processing. [] []
[2025-11-18T13:34:48.959320+02:00] event_sourcing.DEBUG: Subscription Engine: Event stream is processed from position 0. [] []
[2025-11-18T13:34:48.959629+02:00] event_sourcing.DEBUG: Repository: Aggregate "hotel" with the id "019a96be-f77e-7063-a1d6-943da2d4f06b" loaded from store. [] []
[2025-11-18T13:34:48.959669+02:00] event_sourcing.DEBUG: Repository: Aggregate "hotel" with the id "019a96be-f77e-7063-a1d6-943da2d4f06b" saved. [] []
[2025-11-18T13:34:48.959714+02:00] event_sourcing.DEBUG: Subscription Engine: Subscriber "Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessor" for "auto_checkin" processed the event "App\Hotel\Domain\Event\HotelCreated". [] []
[2025-11-18T13:34:48.959728+02:00] event_sourcing.DEBUG: Subscription Engine: Current event stream position: 1 [] []
[2025-11-18T13:34:48.959739+02:00] event_sourcing.INFO: Subscription Engine: End of stream on position "1" has been reached, finish processing. [] []
[2025-11-18T13:34:48.959754+02:00] event_sourcing.INFO: Subscription Engine: Start processing. [] []
[2025-11-18T13:34:48.959769+02:00] event_sourcing.DEBUG: Subscription Engine: Event stream is processed from position 1. [] []
[2025-11-18T13:34:48.959782+02:00] event_sourcing.DEBUG: Subscription Engine: Subscription "auto_checkin" is farther than the current position (1 > 1), continue processing. [] []
[2025-11-18T13:34:48.959794+02:00] event_sourcing.DEBUG: Subscription Engine: Current event stream position: 1 [] []
[2025-11-18T13:34:48.959804+02:00] event_sourcing.INFO: Subscription Engine: End of stream on position "1" has been reached, finish processing. [] []
[2025-11-18T13:34:48.959813+02:00] event_sourcing.INFO: Subscription Engine: Start processing. [] []
[2025-11-18T13:34:48.959826+02:00] event_sourcing.DEBUG: Subscription Engine: Event stream is processed from position 1. [] []
[2025-11-18T13:34:48.959836+02:00] event_sourcing.DEBUG: Subscription Engine: Subscription "auto_checkin" is farther than the current position (1 > 1), continue processing. [] []
...

[DanielBadura]

Thanks for reporting! I'll inverstigate why this is happening.

[DanielBadura]

This should be fixed with https://github.com/patchlevel/event-sourcing-bundle/releases/tag/3.12.1
Could you test it out?

[ergosarapu]

Thanks for quick update.

The issue is still present with following set of packages:

patchlevel/event-sourcing          3.13.0
patchlevel/event-sourcing-bundle   3.12.1

I think the root cause is the indexing (0 vs 1 based) used in InMemoryStore. Therefore I created a pull request: patchlevel/event-sourcing#790

Could you please review this? It fixes the endless loop.

[DavidBadura]

The fix was released with version 1.13.1.

[ergosarapu]

Confirming the fix is working. Thanks for a great library!