From 642934ff097d697c992b0f74e24fe2cb3d6d0299 Mon Sep 17 00:00:00 2001 From: David Badura Date: Sat, 3 Oct 2026 21:31:42 +0200 Subject: [PATCH 1/3] Add checkpoints for decision models A projection that has to load a long history can now mark an apply method with #[Checkpoint]. Its event has to contain the whole state the projection needs, so everything before the last of these events is skipped. Before reading, the builders look up the last checkpoint event of every projection with the projection's own tags and read from there. Each projection only gets events from its own checkpoint onwards, so other projections in the same decision model still get their whole history. Nothing is archived and the append condition stays the same. This is deliberately separate from #[SplitStream]: that one cuts off a whole stream for aggregates, a checkpoint only affects the projection declaring it. --- docs/dynamic-consistency-boundary.md | 61 +++++++ src/Attribute/Checkpoint.php | 18 ++ .../StoreDecisionModelBuilder.php | 7 +- src/Projection/ApplyMethodDetectionError.php | 10 ++ src/Projection/BasicProjection.php | 23 ++- src/Projection/CheckpointPositions.php | 68 ++++++++ src/Projection/CheckpointProvider.php | 17 ++ src/Projection/CompositeProjection.php | 14 +- src/Projection/StoreProjectionBuilder.php | 7 +- .../StoreDecisionModelBuilderTest.php | 161 ++++++++++++++++++ tests/Unit/Fixture/VisitsProjection.php | 42 +++++ .../ApplyMethodDetectionErrorTest.php | 11 ++ tests/Unit/Projection/BasicProjectionTest.php | 45 +++++ .../Projection/CompositeProjectionTest.php | 24 +++ .../Projection/StoreProjectionBuilderTest.php | 19 +++ 15 files changed, 521 insertions(+), 6 deletions(-) create mode 100644 src/Attribute/Checkpoint.php create mode 100644 src/Projection/CheckpointPositions.php create mode 100644 src/Projection/CheckpointProvider.php create mode 100644 tests/Unit/Fixture/VisitsProjection.php diff --git a/docs/dynamic-consistency-boundary.md b/docs/dynamic-consistency-boundary.md index 53a1ab436..3470379ac 100644 --- a/docs/dynamic-consistency-boundary.md +++ b/docs/dynamic-consistency-boundary.md @@ -299,6 +299,67 @@ final class GuestAlreadyCheckedIn extends BasicProjection } } ``` +### Checkpoints + +Over time, a projection may have to load a lot of events. +If an event contains the whole state the projection needs, you can mark its apply method with `#[Checkpoint]`. +The decision model then only loads events from the last of these events onwards, +looked up with the tags of the projection. + +```php +use Patchlevel\EventSourcing\Attribute\Apply; +use Patchlevel\EventSourcing\Attribute\Checkpoint; +use Patchlevel\EventSourcing\Identifier\Uuid; +use Patchlevel\EventSourcing\Projection\BasicProjection; + +final class NumberOfGuestsInHotel extends BasicProjection +{ + public function __construct( + private readonly Uuid $hotelId, + ) { + } + + public function initialState(): int + { + return 0; + } + + /** @return list */ + protected function tagFilter(): array + { + return ["hotel:{$this->hotelId->toString()}"]; + } + + #[Apply] + #[Checkpoint] + public function applyGuestsCounted(int $state, GuestsCounted $event): int + { + return $event->numberOfGuests; + } + + #[Apply] + public function applyGuestIsCheckedIn(int $state, GuestIsCheckedIn $event): int + { + return $state + 1; + } + + #[Apply] + public function applyGuestIsCheckedOut(int $state, GuestIsCheckedOut $event): int + { + return $state - 1; + } +} +``` +:::note +A checkpoint only affects the projection that declares it. +Other projections in the same decision model and subscriptions still get all events they need. +Nothing is archived and the append condition is the same as without a checkpoint. +::: + +:::tip +If a projection only ever needs its last event, you can override `lastEventIsEnough()` to return `true` instead. +::: + ## Define handlers We’ll implement three command handlers corresponding to our commands. diff --git a/src/Attribute/Checkpoint.php b/src/Attribute/Checkpoint.php new file mode 100644 index 000000000..cece77208 --- /dev/null +++ b/src/Attribute/Checkpoint.php @@ -0,0 +1,18 @@ +store, $projections); + $projection = new CompositeProjection($projections, $from); $query = $projection->query(); - $stream = $this->store->query($query); + $stream = $this->store->query($query, $from === [] ? 0 : min($from)); $state = $projection->initialState(); diff --git a/src/Projection/ApplyMethodDetectionError.php b/src/Projection/ApplyMethodDetectionError.php index 76512951f..7518b1120 100644 --- a/src/Projection/ApplyMethodDetectionError.php +++ b/src/Projection/ApplyMethodDetectionError.php @@ -80,4 +80,14 @@ public static function argumentTypeIsMissing(string $methodName): self ), ); } + + public static function checkpointWithoutApply(string $methodName): self + { + return new self( + sprintf( + 'The method "%s" has a #[Checkpoint] attribute, but no #[Apply] attribute.', + $methodName, + ), + ); + } } diff --git a/src/Projection/BasicProjection.php b/src/Projection/BasicProjection.php index 8a9d0b5c8..2fa26cd53 100644 --- a/src/Projection/BasicProjection.php +++ b/src/Projection/BasicProjection.php @@ -5,6 +5,7 @@ namespace Patchlevel\EventSourcing\Projection; use Patchlevel\EventSourcing\Attribute\Apply; +use Patchlevel\EventSourcing\Attribute\Checkpoint; use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Store\SubQuery; use ReflectionClass; @@ -24,7 +25,7 @@ * @template S = mixed * @implements Projection */ -abstract class BasicProjection implements Projection, SubQueryProvider +abstract class BasicProjection implements Projection, SubQueryProvider, CheckpointProvider { /** @var array|null $applyMethods */ private array|null $applyMethods = null; @@ -67,6 +68,22 @@ public function subQuery(): SubQuery return $this->subQuery; } + /** @return list */ + public function checkpointEvents(): array + { + $checkpointEvents = []; + + foreach ($this->applyMethods() as $eventClass => $methodName) { + if ((new ReflectionMethod($this, $methodName))->getAttributes(Checkpoint::class) === []) { + continue; + } + + $checkpointEvents[] = $eventClass; + } + + return $checkpointEvents; + } + /** @return list */ protected function eventTypeFilter(): array { @@ -88,6 +105,10 @@ private function applyMethods(): array $attributes = $method->getAttributes(Apply::class); if ($attributes === []) { + if ($method->getAttributes(Checkpoint::class) !== []) { + throw ApplyMethodDetectionError::checkpointWithoutApply($method->getName()); + } + continue; } diff --git a/src/Projection/CheckpointPositions.php b/src/Projection/CheckpointPositions.php new file mode 100644 index 000000000..e358f8243 --- /dev/null +++ b/src/Projection/CheckpointPositions.php @@ -0,0 +1,68 @@ + $projections + * + * @return array + */ + public static function resolve(AppendStore $store, array $projections): array + { + $positions = array_map(static fn (): int => 0, $projections); + $lookups = []; + + foreach ($projections as $name => $projection) { + if (!$projection instanceof CheckpointProvider || !$projection instanceof SubQueryProvider) { + continue; + } + + $checkpointEvents = $projection->checkpointEvents(); + + if ($checkpointEvents === []) { + continue; + } + + $subQuery = $projection->subQuery(); + + $lookups[$name] = new SubQuery( + $subQuery->tags, + $checkpointEvents, + $subQuery->streamName, + true, + ); + } + + if ($lookups === []) { + return $positions; + } + + foreach ($store->query(new Query(...array_values($lookups))) as $index => $message) { + foreach ($lookups as $name => $lookup) { + if (!$lookup->match($message)) { + continue; + } + + $positions[$name] = max($positions[$name], $index); + } + } + + return $positions; + } +} diff --git a/src/Projection/CheckpointProvider.php b/src/Projection/CheckpointProvider.php new file mode 100644 index 000000000..6fbfbd4a4 --- /dev/null +++ b/src/Projection/CheckpointProvider.php @@ -0,0 +1,17 @@ + + */ + public function checkpointEvents(): array; +} diff --git a/src/Projection/CompositeProjection.php b/src/Projection/CompositeProjection.php index 4271aea8c..409880ceb 100644 --- a/src/Projection/CompositeProjection.php +++ b/src/Projection/CompositeProjection.php @@ -5,6 +5,7 @@ namespace Patchlevel\EventSourcing\Projection; use Patchlevel\EventSourcing\Message\Message; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Query; use function array_map; @@ -12,9 +13,13 @@ /** @experimental */ final class CompositeProjection { - /** @param array $projections */ + /** + * @param array $projections + * @param array $from messages before this index are not applied to the projection + */ public function __construct( private readonly array $projections, + private readonly array $from = [], ) { } @@ -51,6 +56,13 @@ public function initialState(): array public function apply(mixed $state, Message $message): mixed { foreach ($this->projections as $name => $projection) { + if ( + ($this->from[$name] ?? 0) > 0 + && $message->header(IndexHeader::class)->index < $this->from[$name] + ) { + continue; + } + $state[$name] = $projection->apply($state[$name], $message); } diff --git a/src/Projection/StoreProjectionBuilder.php b/src/Projection/StoreProjectionBuilder.php index 852ee0d08..3bc59480e 100644 --- a/src/Projection/StoreProjectionBuilder.php +++ b/src/Projection/StoreProjectionBuilder.php @@ -6,6 +6,8 @@ use Patchlevel\EventSourcing\Store\AppendStore; +use function min; + /** @experimental */ final class StoreProjectionBuilder implements ProjectionBuilder { @@ -22,10 +24,11 @@ public function __construct( public function build( array $projections, ): array { - $projection = new CompositeProjection($projections); + $from = CheckpointPositions::resolve($this->store, $projections); + $projection = new CompositeProjection($projections, $from); $query = $projection->query(); - $stream = $this->store->query($query); + $stream = $this->store->query($query, $from === [] ? 0 : min($from)); $state = $projection->initialState(); diff --git a/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php b/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php index 5de35fa13..75d3f6429 100644 --- a/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php +++ b/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php @@ -4,11 +4,14 @@ namespace Patchlevel\EventSourcing\Tests\Unit\DecisionModel; +use Patchlevel\EventSourcing\Attribute\Apply; use Patchlevel\EventSourcing\DecisionModel\StoreDecisionModelBuilder; use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Message\Stream; +use Patchlevel\EventSourcing\Projection\BasicProjection; use Patchlevel\EventSourcing\Store\AppendCondition; use Patchlevel\EventSourcing\Store\AppendStore; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\InMemoryStore; use Patchlevel\EventSourcing\Store\Query; @@ -18,6 +21,9 @@ use Patchlevel\EventSourcing\Tests\Unit\Fixture\LastEmailProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; @@ -115,4 +121,159 @@ public function testEmptyStream(): void self::assertEquals(new AppendCondition($expectedQuery, 0), $state->appendCondition); } + + public function testCheckpoint(): void + { + $store = new InMemoryStore(); + $store->save( + $this->visited('1', ['profile-1']), + $this->visited('1', ['profile-1']), + $this->split(10, ['profile-1']), + $this->visited('1', ['profile-1']), + $this->split(99, ['profile-2']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 12], $state->state); + self::assertSame(6, $state->appendCondition->after); + } + + public function testCheckpointReadsFromLastCheckpointEvent(): void + { + $splitMessage = $this->split(10, ['profile-1'])->withHeader(new IndexHeader(3)); + + $expectedQuery = new Query( + new SubQuery(['profile-1'], [ProfileVisited::class, SplittingEvent::class]), + ); + + $store = new class ($splitMessage) implements AppendStore { + /** @var list */ + public array $calls = []; + + public function __construct( + private readonly Message $splitMessage, + ) { + } + + /** @param iterable $messages */ + public function append(iterable $messages, AppendCondition|null $appendCondition = null): void + { + } + + public function query(Query $query, int $from = 0): Stream + { + $this->calls[] = [$query, $from]; + + return new Stream([3 => $this->splitMessage]); + } + }; + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 10], $state->state); + self::assertEquals(new AppendCondition($expectedQuery, 3), $state->appendCondition); + self::assertEquals( + [ + [new Query(new SubQuery(['profile-1'], [SplittingEvent::class], null, true)), 0], + [$expectedQuery, 3], + ], + $store->calls, + ); + } + + public function testCheckpointWithoutCheckpointEvent(): void + { + $store = new InMemoryStore(); + $store->save( + $this->visited('1', ['profile-1']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 2], $state->state); + } + + public function testCheckpointWithProjectionWithoutCheckpoint(): void + { + $store = new InMemoryStore(); + $store->save( + Message::create(new ProfileCreated(ProfileId::fromString('1'), Email::fromString('info@patchlevel.de'))) + ->withHeader(new TagsHeader(['profile-1'])), + $this->visited('1', ['profile-1']), + $this->split(10, ['profile-1']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'created' => new IncrementProjection(0, ['profile-1']), + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['created' => 1, 'visits' => 11], $state->state); + } + + public function testSplitStreamAloneIsNoCheckpoint(): void + { + $store = new InMemoryStore(); + $store->save( + $this->split(10, ['profile-1']), + $this->split(20, ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'splits' => new class extends BasicProjection { + public function initialState(): int + { + return 0; + } + + #[Apply] + public function applySplittingEvent(int $state, SplittingEvent $event): int + { + return $state + 1; + } + + /** @return list */ + protected function tagFilter(): array + { + return ['profile-1']; + } + }, + ]); + + self::assertSame(['splits' => 2], $state->state); + } + + /** @param list $tags */ + private function visited(string $id, array $tags): Message + { + return Message::create(new ProfileVisited(ProfileId::fromString($id))) + ->withHeader(new TagsHeader($tags)); + } + + /** @param list $tags */ + private function split(int $visits, array $tags): Message + { + return Message::create(new SplittingEvent(Email::fromString('info@patchlevel.de'), $visits)) + ->withHeader(new TagsHeader($tags)); + } } diff --git a/tests/Unit/Fixture/VisitsProjection.php b/tests/Unit/Fixture/VisitsProjection.php new file mode 100644 index 000000000..a188bded7 --- /dev/null +++ b/tests/Unit/Fixture/VisitsProjection.php @@ -0,0 +1,42 @@ + $tags */ + public function __construct( + private readonly array $tags, + ) { + } + + public function initialState(): int + { + return 0; + } + + #[Apply] + public function applyProfileVisited(int $state, ProfileVisited $event): int + { + return $state + 1; + } + + #[Apply] + #[Checkpoint] + public function applySplittingEvent(int $state, SplittingEvent $event): int + { + return $event->visits; + } + + /** @return list */ + protected function tagFilter(): array + { + return $this->tags; + } +} diff --git a/tests/Unit/Projection/ApplyMethodDetectionErrorTest.php b/tests/Unit/Projection/ApplyMethodDetectionErrorTest.php index 2ef4dfcb9..f91eea130 100644 --- a/tests/Unit/Projection/ApplyMethodDetectionErrorTest.php +++ b/tests/Unit/Projection/ApplyMethodDetectionErrorTest.php @@ -89,4 +89,15 @@ public function testArgumentTypeIsMissing(): void ); self::assertSame(0, $exception->getCode()); } + + public function testCheckpointWithoutApply(): void + { + $exception = ApplyMethodDetectionError::checkpointWithoutApply('applyEvent'); + + self::assertSame( + 'The method "applyEvent" has a #[Checkpoint] attribute, but no #[Apply] attribute.', + $exception->getMessage(), + ); + self::assertSame(0, $exception->getCode()); + } } diff --git a/tests/Unit/Projection/BasicProjectionTest.php b/tests/Unit/Projection/BasicProjectionTest.php index 12e41ca46..a9dc11b5d 100644 --- a/tests/Unit/Projection/BasicProjectionTest.php +++ b/tests/Unit/Projection/BasicProjectionTest.php @@ -6,15 +6,19 @@ use Countable; use Patchlevel\EventSourcing\Attribute\Apply; +use Patchlevel\EventSourcing\Attribute\Checkpoint; use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Projection\ApplyMethodDetectionError; use Patchlevel\EventSourcing\Projection\BasicProjection; use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\SubQuery; use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\IncrementProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; use Stringable; @@ -53,6 +57,47 @@ public function applyProfileCreated(int $state, ProfileCreated $event): int self::assertSame($projection->subQuery(), $projection->subQuery()); } + public function testCheckpointEvents(): void + { + $projection = new VisitsProjection([]); + + self::assertSame([SplittingEvent::class], $projection->checkpointEvents()); + } + + public function testNoCheckpointEvents(): void + { + $projection = new IncrementProjection(0); + + self::assertSame([], $projection->checkpointEvents()); + } + + public function testCheckpointWithoutApply(): void + { + $projection = new class extends BasicProjection { + public function initialState(): int + { + return 0; + } + + /** @return list */ + protected function tagFilter(): array + { + return []; + } + + #[Checkpoint] + public function applyProfileCreated(int $state, ProfileCreated $event): int + { + return $state; + } + }; + + $this->expectException(ApplyMethodDetectionError::class); + $this->expectExceptionMessage('has a #[Checkpoint] attribute, but no #[Apply] attribute'); + + $projection->subQuery(); + } + public function testApplyWithNonMatchingMessage(): void { $projection = new class extends BasicProjection { diff --git a/tests/Unit/Projection/CompositeProjectionTest.php b/tests/Unit/Projection/CompositeProjectionTest.php index 4d965d3f0..c5d83fd05 100644 --- a/tests/Unit/Projection/CompositeProjectionTest.php +++ b/tests/Unit/Projection/CompositeProjectionTest.php @@ -7,6 +7,7 @@ use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Projection\CompositeProjection; use Patchlevel\EventSourcing\Projection\Projection; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Header\StreamNameHeader; use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\Query; @@ -77,6 +78,29 @@ public function testApplyDelegatesToEachProjection(): void self::assertSame(11, $newState['y']); } + public function testApplySkipsMessagesBeforeFrom(): void + { + $composite = new CompositeProjection( + [ + 'x' => new IncrementProjection(0, ['match']), + 'y' => new IncrementProjection(0, ['match']), + ], + ['y' => 2], + ); + + $message = Message::create(new ProfileCreated( + ProfileId::fromString('test'), + Email::fromString('foo@example.com'), + )) + ->withHeader(new TagsHeader(['match'])); + + $state = $composite->initialState(); + $state = $composite->apply($state, $message->withHeader(new IndexHeader(1))); + $state = $composite->apply($state, $message->withHeader(new IndexHeader(2))); + + self::assertSame(['x' => 2, 'y' => 1], $state); + } + public function testApplySkipsNonMatchingProjection(): void { $composite = new CompositeProjection([ diff --git a/tests/Unit/Projection/StoreProjectionBuilderTest.php b/tests/Unit/Projection/StoreProjectionBuilderTest.php index 509e41465..07b071a84 100644 --- a/tests/Unit/Projection/StoreProjectionBuilderTest.php +++ b/tests/Unit/Projection/StoreProjectionBuilderTest.php @@ -10,10 +10,14 @@ use Patchlevel\EventSourcing\Projection\StoreProjectionBuilder; use Patchlevel\EventSourcing\Store\AppendStore; use Patchlevel\EventSourcing\Store\Header\TagsHeader; +use Patchlevel\EventSourcing\Store\InMemoryStore; use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email; use Patchlevel\EventSourcing\Tests\Unit\Fixture\IncrementProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; @@ -42,4 +46,19 @@ public function testBuild(): void self::assertSame(['count' => 2], $builder->build($projections)); } + + public function testBuildWithCheckpoint(): void + { + $store = new InMemoryStore(); + $store->save( + Message::create(new ProfileVisited(ProfileId::fromString('1')))->withHeader(new TagsHeader(['match'])), + Message::create(new SplittingEvent(Email::fromString('foo@bar.com'), 5)) + ->withHeader(new TagsHeader(['match'])), + Message::create(new ProfileVisited(ProfileId::fromString('1')))->withHeader(new TagsHeader(['match'])), + ); + + $builder = new StoreProjectionBuilder($store); + + self::assertSame(['visits' => 6], $builder->build(['visits' => new VisitsProjection(['match'])])); + } } From 4e46c214e0dec334347f3defc0187d85a5eb0a78 Mon Sep 17 00:00:00 2001 From: David Badura Date: Sat, 3 Oct 2026 23:01:10 +0200 Subject: [PATCH 2/3] Add benchmarks for the dynamic consistency boundary The existing taggable store benchmarks only cover aggregates and plain appends. Decisions are different: their cost depends on the size of the whole table and on how selective the tags are, not just on the matching events. The new benchmark fills the store with 1.000 courses, 1.000 invoices and two courses with a long history, and measures building small, composite and long decision models, checkpoints, only-last-event projections and conditional appends against that data. --- .../Events/CourseCapacityChanged.php | 19 ++ .../Events/CourseDefined.php | 19 ++ .../Events/InvoiceCreated.php | 18 ++ .../Events/StudentSubscribedToCourse.php | 20 ++ .../Events/SubscriptionsCounted.php | 19 ++ .../Projection/CourseCapacity.php | 41 ++++ .../Projection/CourseExists.php | 34 +++ .../Projection/NextInvoiceNumber.php | 34 +++ .../NumberOfCourseSubscriptions.php | 34 +++ ...berOfCourseSubscriptionsWithCheckpoint.php | 43 ++++ .../NumberOfStudentSubscriptions.php | 34 +++ .../Projection/StudentAlreadySubscribed.php | 35 ++++ .../DynamicConsistencyBoundaryBench.php | 196 ++++++++++++++++++ 13 files changed, 546 insertions(+) create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Events/CourseCapacityChanged.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Events/CourseDefined.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Events/InvoiceCreated.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Events/StudentSubscribedToCourse.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Events/SubscriptionsCounted.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/CourseCapacity.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/CourseExists.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/NextInvoiceNumber.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptions.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptionsWithCheckpoint.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfStudentSubscriptions.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundary/Projection/StudentAlreadySubscribed.php create mode 100644 tests/Benchmark/DynamicConsistencyBoundaryBench.php diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Events/CourseCapacityChanged.php b/tests/Benchmark/DynamicConsistencyBoundary/Events/CourseCapacityChanged.php new file mode 100644 index 000000000..146f8b567 --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Events/CourseCapacityChanged.php @@ -0,0 +1,19 @@ +capacity; + } + + #[Apply] + public function applyCourseCapacityChanged(int $state, CourseCapacityChanged $event): int + { + return $event->capacity; + } + + /** @return list */ + protected function tagFilter(): array + { + return ["course:{$this->courseId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/CourseExists.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/CourseExists.php new file mode 100644 index 000000000..75863a24b --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/CourseExists.php @@ -0,0 +1,34 @@ + */ + protected function tagFilter(): array + { + return ["course:{$this->courseId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/NextInvoiceNumber.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NextInvoiceNumber.php new file mode 100644 index 000000000..b674ff1d6 --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NextInvoiceNumber.php @@ -0,0 +1,34 @@ +invoiceNumber + 1; + } + + /** @return list */ + protected function tagFilter(): array + { + return []; + } + + protected function lastEventIsEnough(): bool + { + return true; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptions.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptions.php new file mode 100644 index 000000000..a9104a4a2 --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptions.php @@ -0,0 +1,34 @@ + */ + protected function tagFilter(): array + { + return ["course:{$this->courseId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptionsWithCheckpoint.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptionsWithCheckpoint.php new file mode 100644 index 000000000..7b0556972 --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfCourseSubscriptionsWithCheckpoint.php @@ -0,0 +1,43 @@ +subscriptions; + } + + /** @return list */ + protected function tagFilter(): array + { + return ["course:{$this->courseId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfStudentSubscriptions.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfStudentSubscriptions.php new file mode 100644 index 000000000..0d3904233 --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/NumberOfStudentSubscriptions.php @@ -0,0 +1,34 @@ + */ + protected function tagFilter(): array + { + return ["student:{$this->studentId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundary/Projection/StudentAlreadySubscribed.php b/tests/Benchmark/DynamicConsistencyBoundary/Projection/StudentAlreadySubscribed.php new file mode 100644 index 000000000..b4d0bcd4d --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundary/Projection/StudentAlreadySubscribed.php @@ -0,0 +1,35 @@ + */ + protected function tagFilter(): array + { + return ["course:{$this->courseId}", "student:{$this->studentId}"]; + } +} diff --git a/tests/Benchmark/DynamicConsistencyBoundaryBench.php b/tests/Benchmark/DynamicConsistencyBoundaryBench.php new file mode 100644 index 000000000..b1f7bc2ff --- /dev/null +++ b/tests/Benchmark/DynamicConsistencyBoundaryBench.php @@ -0,0 +1,196 @@ +store = new TaggableDoctrineDbalStore( + $connection, + DefaultEventSerializer::createFromPaths([__DIR__ . '/DynamicConsistencyBoundary/Events']), + (new AttributeEventRegistryFactory())->create([__DIR__ . '/DynamicConsistencyBoundary/Events']), + ); + + $this->decisionModelBuilder = new StoreDecisionModelBuilder($this->store); + $this->eventAppender = new StoreEventAppender($this->store); + + $schemaDirector = new DoctrineSchemaDirector( + $connection, + $this->store, + ); + + $schemaDirector->create(); + + $events = []; + + for ($course = 1; $course <= self::COURSES; $course++) { + $events[] = new CourseDefined(sprintf('c%d', $course), 20); + + for ($i = 0; $i < self::SUBSCRIPTIONS_PER_COURSE; $i++) { + $events[] = new StudentSubscribedToCourse( + sprintf('c%d', $course), + sprintf('s%d', ($course * self::SUBSCRIPTIONS_PER_COURSE + $i) % self::STUDENTS), + ); + } + } + + for ($invoice = 1; $invoice <= self::INVOICES; $invoice++) { + $events[] = new InvoiceCreated($invoice); + } + + $events[] = new CourseDefined('long', self::LONG_HISTORY); + $events[] = new CourseDefined('checkpoint', self::LONG_HISTORY); + + for ($i = 1; $i <= self::LONG_HISTORY; $i++) { + $events[] = new StudentSubscribedToCourse('long', sprintf('s%d', $i % self::STUDENTS)); + $events[] = new StudentSubscribedToCourse('checkpoint', sprintf('s%d', $i % self::STUDENTS)); + + if ($i % self::CHECKPOINT_EVERY !== 0) { + continue; + } + + $events[] = new SubscriptionsCounted('checkpoint', $i); + } + + $tagExtractor = new AttributeEventTagExtractor(); + + foreach (array_chunk($events, 1_000) as $chunk) { + $messages = []; + + foreach ($chunk as $event) { + $messages[] = Message::create($event)->withHeader(new TagsHeader($tagExtractor->extract($event))); + } + + $this->store->save(...$messages); + } + } + + #[Bench\Revs(10)] + public function benchDecisionModelSmall(): void + { + $this->decisionModelBuilder->build([ + 'courseExists' => new CourseExists('c500'), + ]); + } + + #[Bench\Revs(10)] + public function benchDecisionModelComposite(): void + { + $this->decisionModelBuilder->build([ + 'courseExists' => new CourseExists('c500'), + 'courseCapacity' => new CourseCapacity('c500'), + 'numberOfCourseSubscriptions' => new NumberOfCourseSubscriptions('c500'), + 'numberOfStudentSubscriptions' => new NumberOfStudentSubscriptions('s42'), + 'studentAlreadySubscribed' => new StudentAlreadySubscribed('c500', 's42'), + ]); + } + + #[Bench\Revs(10)] + public function benchDecisionModel10000Events(): void + { + $this->decisionModelBuilder->build([ + 'numberOfCourseSubscriptions' => new NumberOfCourseSubscriptions('long'), + ]); + } + + #[Bench\Revs(10)] + public function benchDecisionModel10000EventsWithCheckpoint(): void + { + $this->decisionModelBuilder->build([ + 'numberOfCourseSubscriptions' => new NumberOfCourseSubscriptionsWithCheckpoint('checkpoint'), + ]); + } + + #[Bench\Revs(10)] + public function benchDecisionModelOnlyLastEvent(): void + { + $this->decisionModelBuilder->build([ + 'nextInvoiceNumber' => new NextInvoiceNumber(), + ]); + } + + #[Bench\Revs(10)] + public function benchDecideAndAppend(): void + { + $studentId = sprintf('new%d', ++$this->newStudent); + + $decisionModel = $this->decisionModelBuilder->build([ + 'courseExists' => new CourseExists('c500'), + 'courseCapacity' => new CourseCapacity('c500'), + 'numberOfCourseSubscriptions' => new NumberOfCourseSubscriptions('c500'), + 'numberOfStudentSubscriptions' => new NumberOfStudentSubscriptions($studentId), + 'studentAlreadySubscribed' => new StudentAlreadySubscribed('c500', $studentId), + ]); + + $this->eventAppender->append( + [new StudentSubscribedToCourse('c500', $studentId)], + $decisionModel->appendCondition, + ); + } + + #[Bench\Revs(10)] + public function benchAppendWithCondition(): void + { + $courseId = sprintf('new%d', ++$this->newStudent); + + $this->eventAppender->append( + [new CourseDefined($courseId, 20)], + new AppendCondition(new Query(new SubQuery(['course:' . $courseId]))), + ); + } +} From c133b6940fc3c8d6f27ac1bcaf7b0e38b185a5a1 Mon Sep 17 00:00:00 2001 From: David Badura Date: Mon, 5 Oct 2026 10:41:20 +0200 Subject: [PATCH 3/3] Use the GIN index middleware for test and benchmark connections --- tests/DbalManager.php | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/tests/DbalManager.php b/tests/DbalManager.php index 240e5c21d..d8b161245 100644 --- a/tests/DbalManager.php +++ b/tests/DbalManager.php @@ -4,12 +4,13 @@ namespace Patchlevel\EventSourcing\Tests; +use Doctrine\DBAL\Configuration; use Doctrine\DBAL\Connection; -use Doctrine\DBAL\Driver\AbstractSQLiteDriver; use Doctrine\DBAL\DriverManager; use Doctrine\DBAL\Platforms\PostgreSQLPlatform; use Doctrine\DBAL\Tools\DsnParser; use Patchlevel\EventSourcing\Console\DoctrineHelper; +use Patchlevel\EventSourcing\Store\Dbal\PostgreSQLPlatformMiddleware; use RuntimeException; use function getenv; @@ -34,9 +35,12 @@ public static function createConnection(string $dbName = self::DEFAULT_DB_NAME): $connectionParams['dbname'] = $dbName; } - $connection = DriverManager::getConnection($connectionParams); + $connection = DriverManager::getConnection( + $connectionParams, + (new Configuration())->setMiddlewares([new PostgreSQLPlatformMiddleware()]), + ); - if ($connection->getDriver() instanceof AbstractSQLiteDriver) { + if (in_array($connectionParams['driver'] ?? null, ['pdo_sqlite', 'sqlite3'], true)) { return $connection; }