Event store #421

jlabedo · 2024-12-13T18:24:28Z

Why is this change proposed?

Event store with resiliency, scalability and availability

Description of Changes

TODO

Pull Request Contribution Terms

I have read and agree to the contribution terms outlined in CONTRIBUTING.

lifinsky · 2024-12-13T21:52:37Z

packages/EventSourcingV2/src/EventStore/SQL/PostgresEventStore.php

+    before_transaction_id < pg_current_xact_id())
+ORDER BY name
+FOR SHARE
+SQL);


heredoc adjusts indentation based on the position of the enclosing element, so you can use normal indentation.

lifinsky · 2024-12-13T22:10:25Z

packages/EventSourcingV2/tests/EventStore/Fixtures/PostgresEventCounterProjector.php

+            ON CONFLICT (event_type) DO
+                UPDATE SET counter =  {$this->tableName}.counter + 1
+                WHERE  {$this->tableName}.event_type = ?
+        SQL);


Missing indent for SQL

lifinsky · 2024-12-13T22:22:08Z

packages/EventSourcingV2/composer.json

+        "phpunit/phpunit": "^9.5",
+        "phpstan/phpstan": "^1.8",
+        "psr/container": "^2.0",
+        "symfony/event-dispatcher": "^7.2",


Why from 7.2?

lifinsky · 2024-12-13T22:25:54Z

packages/EventSourcingV2/src/EventStore/SQL/PostgresEventStore.php

+    }
+
+    public function schemaUp(): void
+    {


How to migrate from Prooph event store?

lifinsky · 2024-12-13T22:44:39Z

packages/EventSourcingV2/src/Ecotone/EcotoneAsynchronousProjectionRunner.php

+            foreach ($eventPage->events as $event) {
+                $projector->project($event);
+            }
+            $this->eventStore->ack($eventPage);


Ack does not require a commit and opening a new transaction? And if it is in one transaction, then it will be very large in the number of statements. If the transaction is not used for projection run, it may lost ack changes.

Assumption is, it's wrapped in Ecotone's Dbal transaction. Right @jlabedo ?

Yes for now it's wrapped in Ecotone's Dbal transaction. But @lifinsky is right, if we are doing the catchup here (full reset) then it will be very large in the number of statements.
I see two options:

We let the asynchronous runner wrapped in Ecotone's transaction but we need to first catchup the projection (projection:init command) before the async projection is triggered.

We disable Ecotone's transaction on this endpoint an we manually manage small transactions there

dgafka · 2024-12-14T11:02:02Z

packages/EventSourcingV2/src/Ecotone/StandaloneAggregate/WithEventSourcingAttributes.php

+    }
+
+    #[MutatingEvents]
+    public function mutatingEvents(): array


What is this for?

dgafka · 2024-12-14T11:04:00Z

packages/EventSourcingV2/src/Ecotone/StandaloneAggregate/WithEventSourcingAttributes.php

+    protected function apply(object $event): void
+    {
+        $this->mutatingEvents[] = $event;
+        $this->applyHandler($event);


At final stage the apply won't happen here, and will be part of Ecotone's Aggregate abstraction?

dgafka · 2024-12-14T11:26:48Z

packages/EventSourcingV2/src/EventStore/Projection/InlineProjectionManager.php

+     */
+    public function runProjectionsWith(array $events): void;
+
+    public function addProjection(string $projectorName, string $state = "catchup"): void;


Can $state be enum for more visibility where each state is used?

Ecotone is now locked on 8.1, so we are free to go

dgafka · 2024-12-14T11:32:20Z

packages/EventSourcingV2/src/EventStore/SQL/MysqlEventStore.php

+                $this->connection->prepare("UPDATE {$this->streamTableName} SET version = ? WHERE stream_id = ?")->execute([$version, $eventStreamId->streamId]);
+            }
+
+            $this->runProjectionsWith($persistedEvents);


This method right now just fetches all the projections and tries to trigger them.
Ecotone can deliver the bindings between interested events and projections, so we can just trigger the related ones.

Instead of duplicacting this logic, we can instead inject here a service, that we will reuse between Postgres, Mysql and In Memory, and just feed it from Events from Event Store.

Wdyt?

dgafka · 2024-12-14T11:52:38Z

packages/EventSourcingV2/src/EventStore/StreamEventId.php

+
+use Stringable;
+
+readonly class StreamEventId


The name is a bit confusing, I thought it's a wrapper around stream id, but also is a version in the stream.

I would say StreamPosition would be more descriptive here, as then it make sense that it contains of stream id and the version.
Wdyt?

dgafka · 2024-12-14T15:41:00Z

packages/EventSourcingV2/src/EventStore/SQL/PostgresEventStore.php

+CREATE TABLE IF NOT EXISTS {$this->eventTableName}
+(
+    id             BIGSERIAL PRIMARY KEY,
+    transaction_id XID8    NOT NULL DEFAULT pg_current_xact_id(),


Suggested change

transaction_id XID8 NOT NULL DEFAULT pg_current_xact_id(),

transaction_number XID8 NOT NULL DEFAULT pg_current_xact_id(),

That would make it more clear too, as id does not indicate order by itself (e.g. uuid can be id)

dgafka · 2024-12-14T16:14:01Z

packages/EventSourcingV2/src/EventStore/SQL/PostgresEventStore.php

+
+
+        // Execute missing events
+        $missingEventsLoop = 0;


We already projected events few lines before.

What is the use case for redoing that now?
I do think if something happened between projecting in line #356 and doing that extra commit for inline, then we can simply ignore that, to keep the code more easy to work with.

I don't get your statement.

We are projecting events in order by avoiding gaps on line 356. When there are no more events to catchup, we switch to inline, but events discarded by gaps are still not handled (and won't be handled by inline processing). This is the reason of this code: handling eventually missing events.

By the way, for now there is no lock on the events table, so there is a small period of time where events can be projected in wrong order: Events handled on line 398 can be projected after first "inline" events. It may be ok, but if it is not, we can add a brief lock on events table.

dgafka · 2024-12-14T16:15:30Z

packages/EventSourcingV2/src/EventStore/SQL/PostgresEventStore.php

+        }
+
+        do {
+            $page = $this->readFromSubscription($projectorName);


If EventPage will hold iterator which will call database to fetch the events then we do not even need to recall this. We can rename EventPage to EventStream, as it will hold all ability to fetch all related events for us.

Then we can simply do foreach($eventStream->events() as $event) without while

dgafka · 2024-12-14T16:59:26Z

packages/EventSourcingV2/src/Ecotone/EcotoneAsynchronousProjectionRunner.php

+            }
+            $this->eventStore->ack($eventPage);
+
+            if ($command->until !== null && $eventPage->endPosition->isBefore($command->until)) {


We always want to project till the end, if we can catch something that just happened then why not, less delay for projection.
The other command then will simply skip.

This is the idea of this code: project until there are no more events to handle (You are right I am missing a check to continue while $eventPage->events is not empty)

But the idea behind this is also to force the runner to wait for new events that are still after an interleaved transaction (a gap).
Imagine we start a very long transaction appending events that this projection is not intersested in. Then a second transaction start, appends events related to this projection. The runner starts but won't see any events: there is a gap because of the first transaction still running.
I we stop the runner without waiting, the projection would be stale until another event triggers the projection.
That is the reason why we are stating that this trigger is not complete if it does not reach AT LEAST the position of the last event that triggered the projection.
Does it makes sense ?

dgafka · 2024-12-14T17:00:09Z

packages/EventSourcingV2/src/Ecotone/EcotoneAsynchronousProjectionRunner.php

+            foreach ($eventPage->events as $event) {
+                $projector->project($event);
+            }
+            $this->eventStore->ack($eventPage);


Assumption is, it's wrapped in Ecotone's Dbal transaction. Right @jlabedo ?

jlabedo added 11 commits December 5, 2024 22:39

event store

472f175

the long path to standard repository

89f963e

pure aggregates

8895f31

ecotone projector

b28317e

on the way to async

6a3e11b

async projections

23ed4e0

working tests

17c97da

fix tests

a68e385

add todo

397d73d

add for share

717ac91

save

fb22524

jlabedo marked this pull request as draft December 13, 2024 18:24

lifinsky reviewed Dec 13, 2024

View reviewed changes

dgafka reviewed Dec 14, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Event store #421

Event store #421

jlabedo commented Dec 13, 2024

lifinsky Dec 13, 2024

lifinsky Dec 13, 2024

lifinsky Dec 13, 2024

lifinsky Dec 13, 2024

lifinsky Dec 13, 2024

dgafka Dec 14, 2024

jlabedo Dec 15, 2024

lifinsky Dec 15, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

jlabedo Dec 15, 2024

dgafka Dec 14, 2024

dgafka Dec 14, 2024

jlabedo Dec 15, 2024

dgafka Dec 14, 2024

	transaction_id XID8 NOT NULL DEFAULT pg_current_xact_id(),
	transaction_number XID8 NOT NULL DEFAULT pg_current_xact_id(),

Event store #421

Are you sure you want to change the base?

Event store #421

Conversation

jlabedo commented Dec 13, 2024

Why is this change proposed?

Description of Changes

Pull Request Contribution Terms

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment