From 55eddf911b271b71558dce0ae5b9f9eebe1e372a Mon Sep 17 00:00:00 2001 From: Sergii Lifinskyi Date: Mon, 14 Sep 2026 17:41:10 +0300 Subject: [PATCH 1/2] Fix Prooph projection manager reuse after reconnect --- .../Prooph/LazyProophProjectionManager.php | 19 +++++---- .../SynchronousEventDrivenProjectionTest.php | 42 +++++++++++++++++++ 2 files changed, 54 insertions(+), 7 deletions(-) diff --git a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php index 86bc28240..2cc253cef 100644 --- a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php +++ b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php @@ -15,6 +15,7 @@ use Ecotone\Messaging\MessageHeaders; use Ecotone\Messaging\Support\InvalidArgumentException; use Prooph\Common\Messaging\Message; +use Prooph\EventStore\EventStore; use Prooph\EventStore\Exception\ProjectionNotFound; use Prooph\EventStore\Exception\RuntimeException; use Prooph\EventStore\Pdo\Projection\MariaDbProjectionManager; @@ -36,8 +37,10 @@ */ class LazyProophProjectionManager implements ProjectionManager { - /** @var LazyProophProjectionManager[] */ + /** @var ProjectionManager[] */ private array $lazyInitializedProjectionManager = []; + /** @var EventStore[] */ + private array $initializedEventStores = []; /** * @param array $projectionSetupConfigurations @@ -54,18 +57,20 @@ public function __construct( private function getProjectionManager(): ProjectionManager { $context = $this->lazyProophEventStore->getContextName(); - if (isset($this->lazyInitializedProjectionManager[$context])) { + $eventStore = $this->getLazyProophEventStore(); + $innerEventStore = $eventStore->getEventStore(); + + if (isset($this->lazyInitializedProjectionManager[$context]) && $this->initializedEventStores[$context] === $innerEventStore) { return $this->lazyInitializedProjectionManager[$context]; } - $eventStore = $this->getLazyProophEventStore(); - $this->lazyInitializedProjectionManager[$context] = match ($eventStore->getEventStoreType()) { - LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), + LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), + LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), + LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), LazyProophEventStore::EVENT_STORE_TYPE_IN_MEMORY => $this->eventSourcingConfiguration->getInMemoryProjectionManager() }; + $this->initializedEventStores[$context] = $innerEventStore; return $this->lazyInitializedProjectionManager[$context]; } diff --git a/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php b/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php index c70391963..72cfcd081 100644 --- a/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php +++ b/packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php @@ -64,6 +64,48 @@ public function test_building_synchronous_event_driven_projection(): void self::assertEquals([['ticket_id' => '124', 'ticket_type' => 'info']], $ecotone->sendQueryWithRouting('getInProgressTickets')); } + public function test_synchronous_projection_uses_the_current_transaction_after_reconnecting(): void + { + $connection = $this->getConnection(); + $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore( + containerOrAvailableServices: [new InProgressTicketList($connection), new TicketEventConverter(), DbalConnectionFactory::class => $this->getConnectionFactory()], + configuration: ServiceConfiguration::createWithDefaults() + ->withEnvironment('prod') + ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::EVENT_SOURCING_PACKAGE])) + ->withNamespaces([ + 'Test\\Ecotone\\EventSourcing\\Fixture\\Ticket', + 'Test\\Ecotone\\EventSourcing\\Fixture\\TicketWithSynchronousEventDrivenProjection', + ]) + ->withExtensionObjects([ + EventSourcingConfiguration::createWithDefaults(), + ]), + pathToRootCatalog: __DIR__ . '/../../', + runForProductionEventStore: true + ); + + $ecotone->initializeProjection(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION); + $ecotone->sendCommand(new RegisterTicket('123', 'Johnny', 'alert')); + self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets')); + + $connection->close(); + $connection->beginTransaction(); + + try { + $ecotone->sendCommand(new CloseTicket('123')); + + self::assertTrue($connection->isTransactionActive()); + self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets')); + } finally { + $connection->rollBack(); + } + + self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets')); + + $ecotone->sendCommand(new CloseTicket('123')); + + self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets')); + } + public function test_synchronous_event_driven_projection_should_be_called_before_standard_event_handlers(): void { $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore( From 039b0005877533055afe29e0bfc96e0561de86cc Mon Sep 17 00:00:00 2001 From: Sergii Lifinskyi Date: Mon, 14 Sep 2026 18:11:25 +0300 Subject: [PATCH 2/2] Skip reconnect tracking for in-memory projection manager --- .../src/Prooph/LazyProophProjectionManager.php | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php index 2cc253cef..6bfa6fe15 100644 --- a/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php +++ b/packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php @@ -56,6 +56,10 @@ public function __construct( private function getProjectionManager(): ProjectionManager { + if ($this->eventSourcingConfiguration->isInMemory()) { + return $this->eventSourcingConfiguration->getInMemoryProjectionManager(); + } + $context = $this->lazyProophEventStore->getContextName(); $eventStore = $this->getLazyProophEventStore(); $innerEventStore = $eventStore->getEventStore(); @@ -68,7 +72,6 @@ private function getProjectionManager(): ProjectionManager LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()), - LazyProophEventStore::EVENT_STORE_TYPE_IN_MEMORY => $this->eventSourcingConfiguration->getInMemoryProjectionManager() }; $this->initializedEventStores[$context] = $innerEventStore;