parallel processing subscription engine

Author: DavidBadura

docs/UPGRADE-4.0.md

@@ -373,6 +373,85 @@ final class MigrationSubscriber
 }
 ```
 
+### Parallel subscription processing
+
+The subscription engine now processes one subscription at a time instead of driving a single shared
+stream across all matching subscriptions. Each subscription is claimed individually with
+`FOR UPDATE SKIP LOCKED`, read from its own position with its own event filter, processed and
+committed in its own short transaction. Several `subscription:run` workers can now process different
+subscriptions in true parallel: a worker that finds a subscription locked by another worker simply
+skips to the next one.
+
+This is mostly transparent, but a few contracts changed.
+
+#### SubscriptionStore
+
+`claim()` and `inLock()` are now mandatory parts of the `SubscriptionStore` interface, and the
+separate `LockableSubscriptionStore` interface has been removed. Every custom store has to implement
+both:
+
+```php
+use Closure;
+use Patchlevel\EventSourcing\Subscription\Store\SubscriptionCriteria;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+interface SubscriptionStore
+{
+    // ... get/find/add/update/remove ...
+
+    /**
+     * Claim exactly one subscription via a row lock (SKIP LOCKED). Return null if the row is held
+     * by another worker or no longer matches the criteria.
+     */
+    public function claim(string $id, SubscriptionCriteria $criteria): Subscription|null;
+
+    /**
+     * @param Closure():T $closure
+     *
+     * @return T
+     *
+     * @template T
+     */
+    public function inLock(Closure $closure): mixed;
+}
+```
+
+`find()` no longer locks the matched rows: it is now a plain, unlocked snapshot read. The locking
+happens per subscription inside `claim()`.
+
+#### Message limit is now per subscription
+
+For `Run` and `Boot`, the `limit` used to cap the total number of messages across the shared stream.
+Because there is no shared stream anymore, `limit` now caps the messages **per subscription**. One
+`subscription:run` pass therefore processes up to `limit × number of subscriptions` messages. The
+CLI default of `message-limit=100` now means "100 per subscription". It also defines the
+checkpoint/lock-hold granularity: a unit of at most `limit` messages commits atomically and releases
+the lock afterwards.
+
+#### Error isolation
+
+An error in one subscription no longer aborts the whole run. Transient errors
+(`Doctrine\DBAL\Exception\RetryableException`, which now includes `TransactionCommitNotPossible`)
+are logged and retried on the next pass without landing in `result.errors`. Any other error locks
+that single subscription into status `Error` and surfaces in `result.errors`, while the remaining
+subscriptions keep processing.
+
+#### OnProcessingFinished replaced by OnSubscriptionProcessed
+
+The `OnProcessingFinished` event was a single-stream concept (one shared last index) and has been
+removed. The runner now dispatches `OnSubscriptionProcessed($subscription, $lastIndex)` per
+subscription, which is where batches are flushed with that subscription's own last index. The
+run-wide completion hook is the already existing `OnResult` event.
+
+If you dispatched or listened to `OnProcessingFinished`, switch to `OnSubscriptionProcessed`
+(per subscription) or `OnResult` (per command).
+
+#### No global ordering across subscriptions
+
+Subscriptions are processed independently, so there is no global ordering of messages across
+different subscriptions anymore (the per-subscription order is of course preserved). This was never
+guaranteed before either.
+
 ## Store
 
 ### StreamStore

docs/subscription.md

@@ -873,6 +873,14 @@ and keeping all subscriptions up to date.
 It also takes care that new subscribers are booted and old ones are removed again.
 If something breaks, the subscription engine marks the individual subscriptions as faulty and retries them.
 
+Each subscription is processed independently: the engine claims one subscription at a time with a
+row lock, reads its own substream from its own position, processes it and
+commits in a short transaction before moving on to the next one. This has two consequences:
+
+* An error in one subscription no aborts the whole run, the other subscriptions keep going.
+* Multiple workers can run the engine in parallel and share the work automatically (see
+  [Parallel processing](#parallel-processing)).
+
 :::tip
 The Subscription Engine was inspired by the following two blog posts:
 
@@ -1336,7 +1344,7 @@ You can register your own listeners to hook into the engine, for example for log
 | `OnHandleMessage`        | A message is about to be passed to a subscriber                      |
 | `OnHandleMessageSuccess` | A message was successfully handled by a subscriber                   |
 | `OnHandleMessageError`   | An error occurred while a subscriber was handling a message          |
-| `OnProcessingFinished`   | The engine finished processing the stream (ended or limit reached)   |
+| `OnSubscriptionProcessed`| A subscription finished its stream (ended or limit reached)          |
 | `OnResult`               | The engine finished the command and returns the result               |
 
 ```php
@@ -1484,6 +1492,8 @@ $subscriptionEngine->execute(new Boot());
 
 :::tip
 You can limit the number of processed messages with the `limit` parameter: `new Boot(limit: 100)`.
+The limit applies **per subscription**, so one call processes at most `limit` messages for each
+booting subscription.
 :::
 
 ### Run
@@ -1500,6 +1510,34 @@ $subscriptionEngine->execute(new Run());
 
 :::tip
 You can limit the number of processed messages with the `limit` parameter: `new Run(limit: 100)`.
+The limit applies **per subscription**: one `Run` call processes at most `limit` messages for each
+active subscription, so up to `limit × number of subscriptions` messages in total. It also defines
+the checkpoint and lock-hold granularity, a unit of at most `limit` messages commits atomically and
+the lock is released afterwards.
+:::
+
+### Parallel processing
+
+Because every subscription is claimed independently by a worker, 
+you can start the same command in several worker processes at once. 
+Each worker takes the next available subscription while avoiding subscriptions that are already being processed, 
+so the workload is distributed across workers without any static configuration.
+
+```php
+use Patchlevel\EventSourcing\Subscription\Engine\Command\Run;
+use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine;
+
+// run this in as many worker processes as you like, they balance themselves
+/** @var SubscriptionEngine $subscriptionEngine */
+$subscriptionEngine->execute(new Run(limit: 100));
+```
+
+The per-subscription order is always preserved. There is no global ordering across different
+subscriptions, but since subscriptions are independent of each other this does not matter.
+
+:::note
+This requires a database that supports `SKIP LOCKED` (PostgreSQL and MySQL do). SQLite serializes
+writes anyway and simply ignores the clause, so a single worker is used there.
 :::
 
 ### Teardown

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -68,7 +68,7 @@ public function __construct(
         private readonly EventDispatcherInterface $eventDispatcher = new EventDispatcher(),
         iterable $argumentResolvers = [],
     ) {
-        $this->subscriptionManager = new SubscriptionManager($subscriptionStore);
+        $this->subscriptionManager = new SubscriptionManager($subscriptionStore, $this->logger);
 
         if ($retryStrategyRepository instanceof RetryStrategyRepository) {
             $this->retryStrategyRepository = $retryStrategyRepository;
@@ -94,14 +94,19 @@ public function __construct(
             $this->logger,
         );
 
+        $runner = new SubscriptionRunner(
+            $this->messageLoader,
+            $this->subscriptionManager,
+            $this->subscriberRepository,
+            $messageProcessor,
+            $this->eventDispatcher,
+            $this->logger,
+        );
+
         $this->handlers = [
             Boot::class => new BootHandler(
-                $this->messageLoader,
                 $this->subscriptionManager,
-                $this->subscriberRepository,
-                $messageProcessor,
-                $this->eventDispatcher,
-                $this->logger,
+                $runner,
             ),
             Pause::class => new PauseHandler(
                 $this->subscriptionManager,
@@ -125,11 +130,8 @@ public function __construct(
                 $this->logger,
             ),
             Run::class => new RunHandler(
-                $this->messageLoader,
                 $this->subscriptionManager,
-                $messageProcessor,
-                $this->eventDispatcher,
-                $this->logger,
+                $runner,
             ),
             Setup::class => new SetupHandler(
                 $this->messageLoader,

src/Subscription/Engine/Event/OnProcessingFinished.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Subscription\Engine\Error;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+final class OnSubscriptionProcessed
+{
+    /** @var list<Error> */
+    public array $errors = [];
+
+    public function __construct(
+        public readonly Subscription $subscription,
+        public readonly int|null $lastIndex = null,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnSubscriptionProcessed.php

@@ -53,7 +53,7 @@ private function events(array $subscriptions): array
         $eventNames = [];
 
         foreach ($subscriptions as $subscription) {
-            $subscriber =  $this->subscriberRepository->get($subscription->id());
+            $subscriber =  $this->subscriberRepository->get($subscription->subscriberId());
 
             if (!$subscriber instanceof MetadataSubscriberAccessor) {
                 return [];