Repository navigation
Subscription partitions with a routing inbox - #936
Merged
Merged
Conversation
A subscriber with a #[Partition] method is split into one subscription per partition key. The main subscription resolves the key of every message once and writes the message index into the inbox of that partition, creating the partition on the fly. Partitions only process the messages in their inbox and are only claimed by a worker if the inbox has work, so idle partitions cost nothing and a changed or non-deterministic key can not lose messages. Engine command ids now also match the subscriber id, so commands for a partitioned subscriber apply to all its partitions, and a single partition can be targeted by its own id.
|
Hello 👋 here is the most recent benchmark result:
This comment gets update everytime a new commit comes in! |
A partition now loads exactly the messages in its inbox through the new IndexCriterion instead of the whole index range between its first and last entry, which was as expensive as the stream was long for a key per aggregate. The run snapshot joins the inbox, so finding pending partitions costs as much as there is work, not as many partitions as there are. The router looks up only the partitions of the routed keys instead of loading all of them, and the detach check only claims main subscriptions, their partitions follow. Both scanned every partition on every run. Adds a benchmark with one partition per aggregate.
Run and Boot take subscriptionsPerClaim (default 1) and partitionsPerClaim (default 100). A group is claimed and processed in one transaction, and partitions and other subscriptions are never mixed. Grouped subscriptions read the stream once from the lowest position, a grouped partition reads only its inbox and gets the full message limit. The inbox API is batch based now: a whole group is claimed with one query, its inboxes are read with one query and cleared with one, and routed messages are inserted with multi row inserts. While catching up, the router loads all partitions of the subscriber once instead of looking them up one by one.
The subscription store gets addAll() and updateAll(), and the manager uses them when a flush has more than one subscription, e.g. a group of partitions or a router creating new partitions while catching up. addAll() is a multi row insert. updateAll() sets every column with a CASE over the id, which is the only form of a batch update that PostgreSQL, MySQL, MariaDB and SQLite share. PostgreSQL can not infer the type of the parameters in a CASE, so they are cast there. Since MySQL and MariaDB only count changed rows, a short affected row count is checked against the existing ids before a SubscriptionNotFound is thrown.
A batch often keeps a database transaction open from begin to flush. When several partitions or subscriptions of a group were processed message by message, their transactions on the shared connection were nested, and the rollback of a failing batch hit the innermost transaction of another batch. That batch lost its writes while its position moved on, and the failing one was committed. Subscribers with batching now run one after another within a group, the others are still processed together and the writes of the group are still flushed at once.
A group reads the stream from its lowest position and the message limit applies to the whole group. A subscription far behind, e.g. one that is rebuilt, used up the limit before the others reached their new events, so they waited until the rebuild had caught up. Subscriptions are now sorted by position and only grouped while they are at most the message limit apart (1000 without a limit). Within that gap there are never more messages than the limit, so every member of a group is reached in the same run. Without grouping the order of the snapshot is kept as before.
A partition that failed left the run before its inbox was cleaned up, so the messages it had processed before the error stayed in the inbox. They were only removed with the next successful run, and never if the partition stayed failed. Now they are removed up to its position, which is the last processed message, or the position before a rolled back batch.
Subscriptions loaded in bulk will use native lazy objects for their heavy columns, which need PHP 8.4.
Every run loaded all errored subscriptions with their full error context just to claim each of them and find out that the retry was not due yet. With a thousand failed partitions that took more than 128 MB and the worker died on every run. The store gets findLazy(), which loads the error context and the cleanup tasks with native lazy ghosts on first access. Saving such a subscription reads them, so nothing gets lost. find() stays eager, so the public subscriptions() and anything comparing or dumping subscriptions is not affected. The retry listener now checks on that snapshot whether a retry is due and only claims those subscriptions.
The store had grown batch and lazy variants of most methods. Now there is one method per operation: claim() takes a list of ids, add(), update() and remove() take several subscriptions, and find() may load the error context and the cleanup tasks lazily. The Doctrine store only creates lazy ghosts when one of them is set, so plain subscriptions stay plain objects. The inbox moved into the optional PartitionInboxStore interface with just addToInbox() and inbox(). The store keeps the inbox of a partition to the messages after its position when the partition is updated or removed, so the engine no longer clears it separately. A store without it throws PartitionsNotSupported once a partitioned subscriber is processed.
loadIndices() moved out of MessageLoader into the optional IndexMessageLoader, which the built-in loaders implement. Without it, or when the event store does not know the IndexCriterion, a partition reads the range of its inbox and skips the other messages. A custom message loader or event store therefore needs nothing new. The run only filters for pending partitions when the subscription store has an inbox, so other stores never see that criterion. The router adds to the inbox before it creates a partition, so a store without an inbox fails without leaving a partition behind. Internally, forEachClaimedGroup() takes a ClaimGroups object instead of three loose numbers, and the retry filter only exists on forEachClaimed().
When the database of a projection is down, every partition with work failed, used up its retries and ended up failed, and a subscriber with an OnFailed method even skipped a message in each of them. A new listener counts the failing partitions of a subscriber during a run. If #[Partition(maxConsecutiveFailures)] of them (default 5) fail in a row without a success in between, the main subscription goes into the error status, the partitions that failed are set back and the others of the run skip their messages via the new OnHandleMessage::$skip. While the main subscription is in error, its partitions are not processed at all, so they continue where they were once it is retried. Routing no longer resets the retries of the main subscription, otherwise it would never fail during a long outage. The listener resets them once partitions succeed again.
A subscribe method can ask for the partition key with a parameter marked #[PartitionKey], e.g. to write into the database of a tenant, without computing the key from the message again. Without partitions the value is null. Only the key is handed out, the subscription itself stays inside the engine so a handler can not change its state.
A group of partitions loaded the indices of all its inboxes with one IN list. With 100 partitions and a limit of 1000 that are up to 100k parameters, PostgreSQL allows 65535, so the whole group went into error and the circuit breaker stopped the subscriber. The indices are now loaded in chunks of 1000.
If a routed message was removed from the store before its partition processed it, the partition never loaded it, its position stayed in front of it and it was claimed on every run. When a partition processed every loaded message, it now moves to the end of its inbox and logs the skipped range.
Every boot run claimed all booting partitions, also the ones without work, and looked up the main subscription once per partition. With one partition per aggregate the boot grew quadratically (4000 aggregates: 27k queries, 25s). Boot now only claims partitions with messages in their inbox, like run does. Once a main subscription finished booting, its remaining booting partitions are processed one more time to activate them. The runner loads the main subscriptions of a group with one query. The same boot now needs 1.2k queries and 4s.
The router looked up every new key on its own and after 50 lookups loaded all partitions of the subscriber. The cache only lives for one run, so with a small message limit and many keys every run loaded every partition. The router now remembers the partitions it does not know yet and creates the missing ones at the end of the run, looked up with one query per 500 ids. This also happens when the run fails, so routed messages always have their partition. Catching up 8000 aggregates with a limit of 100 went from 22s to 11s on PostgreSQL.
The snapshot of a run was taken before the main subscriptions routed, so the partitions only processed the new messages in the next run. A single run, like the one of RunSubscriptionEngineRepository after a save, left a partitioned projection behind. Run and boot now claim the main subscriptions first and find the partitions with pending messages afterwards. This costs no extra query, the store already looked them up separately. A run is no longer unfinished just because it routed messages.
If another worker held the main subscription, the breaker could not stop it but still set back the failed partitions. They started again right away, without backoff and without using up retries. Now the partitions are only set back if the main subscription is stopped, by this or another worker. A partition without retries left called its #[OnFailed] method before the breaker decided, which could skip the message during a systemic failure. That transition is now held back until the end of the run and dropped if the breaker trips. The failed handling moved from the FailSubscriber into the FailedMessageProcessor, so the breaker can use it without depending on another listener. The breaker is renamed to PartitionCircuitBreakerSubscriber.
Partition keys come from the messages. The default collations of MySQL and MariaDB treat keys that only differ in case as equal, so a second partition could not be created and the run failed with SubscriptionAlreadyExists. The id, subscriber id and partition key columns and the subscription id of the inbox now use utf8mb4_bin there.
…sage A listener could skip a message with OnHandleMessage::$skip, but the next message that was processed moved the position past it and the skipped message was lost. A batch even moved to the last index of the run on its flush. OnHandleMessage::$stop replaces it: the subscription is not processed any further in this run, its position stays in front of the message and a batch is only flushed up to the last message it handled.
Only active, paused and finished partitions were detached. Booting ones stayed booting and partitions in error kept being retried although their subscriber was gone.
A custom store that does not know the partitions criteria returns the main subscription itself. The listener removed it again and dispatched the removed event for it, which ended in an endless recursion.
The inbox query numbered every message of a partition with ROW_NUMBER() and filtered by the limit afterwards, so a partition with a large backlog was read completely on every run. With a backlog of one million messages one call took 2.7s on MySQL, 1.9s on MariaDB and 148ms on PostgreSQL. PostgreSQL now uses a lateral join with a limit per partition, the other databases a union of one limited query per partition, MySQL does not limit a lateral join early enough and MariaDB and SQLite do not support it. One call now takes 1 to 3ms, no matter how large the backlog is.
Every partition emitted into its own subscription_{subscriber}#{key} stream. Readers of the subscriber stream missed
these events, removing a subscriber deleted one stream per partition, and with a long partition key the stream name
did not fit into the stream column anymore. The emitted stream belongs to the subscriber like setup and cleanup, so
partitions now share it and it is removed once with the main subscription.
Instead of marking a method that returns the partition key, #[Partition] is now set on the subscriber class and takes a partitioner, e.g. #[Partition(new StreamPartitioner())]. Partitioners are small reusable classes implementing the Partitioner interface, StreamPartitioner creates one partition per stream. The partitioner is read from the attribute when it is needed, so it is not part of the cached metadata.
BucketPartitioner spreads the keys of another partitioner over a fixed number of partitions, so the number of partitions no longer grows with the aggregates. TagPartitioner uses the tag with a prefix as key, for the tags of the dynamic consistency boundary. HashedPartitioner hashes the key of another partitioner, so it is always short and any characters are fine.
DavidBadura
marked this pull request as ready for review
October 15, 2026 11:40
The inbox of a partition was cleared with a range delete up to its position. The range delete also locks the next row, often a message the main subscription has just routed, and MariaDB with snapshot isolation, the default since 11.6, fails on that row with "Record has changed since last read". The processed rows are now read first and deleted by their key.
DavidBadura
changed the title
The partitioner was created from the attribute with reflection when it was needed, because a closure could not be cached with the metadata. Without closures the partitioner is a plain object and is stored in the metadata.
The status command loaded every partition to show a summary, and run and boot loaded them to collect the subscriber ids. With one partition per aggregate that can be millions of subscriptions. SubscriptionEngineCriteria can now filter by partitions and status. The status command loads the main subscriptions and only the partitions that need attention, like the ones in error, all of them only with --partitions. Run and boot only load the main subscriptions.
SubscriptionStore and SubscriptionEngine get a stats() method that counts the subscriptions matching a criteria per status, without loading them. The status command uses it to show the number of partitions of a subscriber and how many of them are not active again, it only loads the partitions to list them with --partitions.
Commands like pause, reactivate, remove, teardown, setup and refresh claimed with SKIP LOCKED, so a subscription a worker was processing at that moment was silently left out, e.g. a partition stayed active after pausing its subscriber. claim() now takes $wait, these commands and the circuit breaker wait for the lock instead. Detaching and refreshing no longer update subscriptions without a lock. Processing still skips locked subscriptions.
Since the router creates the partitions at the end of the run, the main subscription can already be in error by then. The partitions copied its status without an error, so nothing ever processed or retried them and their inbox stayed unprocessed after the main subscription recovered. A new partition now starts with the status the main subscription had before its error.
A listener of OnHandleMessageSuccess, like a batch flush, ran outside the error handling of the message processor. Its exception failed the whole claimed group: every partition of the group went into error, members that had already failed lost their error and were never retried, and the circuit breaker counted every member. The exception is now an error of that one subscription, and a failing group no longer overwrites the error of a member.
A partition without retries left was put into error while its failure was held back, and only failed at the end of the run. If the run did not get there, the partition stayed in error forever without retries left. Now it stays active, only the position of the messages handled before is stored. If the run does not get to its end, the message is processed again in the next run. A message another worker processed in the meantime is not failed afterwards.
If the breaker tripped but the main subscription was not running, e.g. paused, it could not be stopped and the held back failures still ran #[OnFailed], which could skip the messages during a systemic failure. They are now dropped whenever the breaker trips, the partitions stay active and try again in the next run.
A listener stopped a subscription through OnHandleMessage::$stop, then neither OnHandleMessageSuccess nor OnHandleMessageError followed. Paired listeners, like one that pushes a message context or opens a span, were left open. The stop is now decided in the new OnBeforeHandleMessage event, OnHandleMessage is only dispatched if the message is processed, so it is always followed by a success or an error.
Looking up main subscriptions by id also matched the subscriber id, so every partition of the subscriber was read, once per partition group, in the circuit breaker and in the run snapshot. Main subscriptions are now found by their primary key, and an index on the partition key finds them without ids. With one million partitions on PostgreSQL the run snapshot of the main subscriptions went from 133ms to 0.05ms.
A partition reads at most 1000 messages of its inbox at a time, so a boot without a limit left partitions with a larger inbox booting, and run never processes booting partitions. Without a limit the boot now repeats the partitions until their inboxes are processed, like it processes the whole stream for other subscriptions.
The end of boot loaded all booting partitions to check whether there are any, and run looked for active partitions of finished run once subscribers on every run. Both now count them first and only claim if there are any. The index on the subscriber id also covers the status, for these counts and for finding idle partitions. The docs mention that commands like pause or remove load all partitions of a subscriber at once.
new TagPartitioner('course') now matches course:{id} only, the same tags
#[EventTag(prefix: 'course')] creates. Before, 'course' also matched
courseware:7 and an empty prefix matched every tag.
A subscriber with the id foo#x next to the partitioned subscriber foo made the creation of the partition x fail at flush, again in every run. The subscriber repository now rejects such ids right away.
Instead of checking the ids of all subscribers against each other in the repository, #[Subscriber] now rejects any id with #. A subscriber id can no longer collide with the id of a partition, also with a custom subscriber repository.
PartitionInboxStore gets inboxStats(), which counts the messages after the position of the matching partitions and in how many partitions they wait. The stats of the engine carry it if the store supports partitions, and the status command shows it next to the number of partitions.
DanielBadura
approved these changes
Oct 16, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Alternative to #935. Instead of every partition reading the whole stream and resolving the partition key again, the main subscription resolves the key once per message and writes the message index into the inbox of that partition (new table
subscriptions_inbox). A partition only processes the indices in its inbox, reads only those messages, and is only claimed when its inbox has work.This fixes the main problem of #935: if the partitioner changes (or isn't deterministic) while a partition lags behind, #935 silently skips messages. Here a routed message always stays in its partition. It also makes idle partitions cheap, so fine grained keys like one per aggregate become realistic, and a failing partition isolates exactly that key with the existing error/retry handling.
Other differences to #935:
SKIP LOCKEDover many small partitions distributes the work alreadytypecolumn, a partition is a subscription with apartition_key#[Partition]is set on the subscriber class with a partitioner, e.g.#[Partition(new StreamPartitioner())], custom ones implementPartitioner. Built in areStreamPartitioner,TagPartitioner(DCB tags,new TagPartitioner('course')matchescourse:{id}like#[EventTag(prefix: 'course')]),BucketPartitioner(a fixed number of partitions) andHashedPartitioner. Subscribe methods get the key with#[PartitionKey]#is reserved for the partition ids,#[Subscriber]rejects it in every subscriber id so the two can never collideSubscriptionCriteria::idsmatches the subscription id or the subscriber id, so a single partition can be targeted by its id and the console commands pass subscriber ids instead of every partition idRun and boot claim the main subscriptions first and the partitions with pending messages afterwards, so routed messages are processed in the same run (also with
RunSubscriptionEngineRepository). The router looks up the partitions of a run with one query per 500 keys and creates the missing ones at the end of the run. Boot only claims partitions with work and activates the remaining ones once their main subscription finished booting, so it stays linear. Without a limit it repeats the partitions until their inboxes are empty. Main subscriptions are found by their primary key or an index on(partition_key, status), without reading their partitions. A partition loads its inbox messages via a newIndexCriterion, in chunks of 1000 indices because of the parameter limits. The inbox query reads at mostlimitentries per partition, with a lateral join on PostgreSQL and a union of limited queries on the other databases, so a large backlog doesn't slow down every run.PartitionedSubscriptionEngineBenchuses one partition per aggregate, every aggregate with two events that are as far apart in the stream as possible. PostgreSQL, catch up processes everything from scratch, incremental is one run after 10 new events:Partitions are claimed in groups of 100 (
partitionsPerClaim), the limit applies to every partition of a group. Normal subscriptions can be grouped too withsubscriptionsPerClaim, then they read the stream once. Only subscriptions at mostlimitapart (1000 without a limit) are grouped. The default stays 1, because a group holds all its locks and spreads worse over several workers. Subscribers with batching are processed one after another within a group, overlapping batches on a shared connection would roll back the wrong one.By default partitions are kept. With
#[Partition(removeIdleAfter: 3600)]the main subscription removes active partitions with an empty inbox that weren't saved for that long, before it routes and while it holds its lock. The next message of the key creates the partition again.A circuit breaker (
PartitionCircuitBreakerSubscriber) stops a partitioned subscriber on a systemic failure: ifmaxConsecutiveFailurespartitions (default 5) fail in a row, the main subscription goes into error and the failed partitions are set back. A partition without retries left only fails and calls#[OnFailed]at the end of the run if the breaker didn't trip, until then nothing about the failure is stored. Whenever the breaker trips these held back failures are dropped, the partitions stay active and try again in the next run. The failed handling moved into an internalFailedMessageProcessorfor that. A main subscription in error stops the whole subscriber, also when the routing fails. Listeners can stop a subscription for the rest of a run with the newOnBeforeHandleMessage::$stop, soOnHandleMessageis always followed by a success or an error.Other details:
utf8mb4_bin, keys that only differ in case are different partitionssubscription_<subscriber id>stream of their subscriber, it's removed once with the main subscriptionOnHandleMessageSuccesslistener, like a batch flush, is an error of that one subscription and doesn't fail the whole claimed groupclaim(..., wait: true)) instead of silently leaving them out, processing still usesSKIP LOCKEDThe
SubscriptionStoreinterface stays small:claim()takes a list of ids,add(),update()andremove()are variadic. The inbox lives in the optionalPartitionInboxStoreinterface withaddToInbox(),inbox(),inboxStats()andremoveIdlePartitions(), the store clears the inbox of a partition up to its position on update. A custom store only needs it to support partitions. Loading only the inbox messages works the same way: the built-in loaders implement the optionalIndexMessageLoader, a custom loader without it, or an event store withoutIndexCriterion, falls back to reading the range of the inbox.This PR raises the minimum PHP version to 8.4 (own commit).
find()of the Doctrine store loads error context and cleanup tasks with native lazy ghosts on first access, and the retry listener only claims subscriptions whose retry is due. An idle run with 1000 failed partitions went from out of memory to 13ms.ConcurrentWorkerTestruns two real worker processes against PostgreSQL, MySQL or MariaDB while new events keep coming in, every message has to be processed exactly once and in order.SubscriptionStoreandSubscriptionEngineget astats()method that counts the matching subscriptions per status. If the store supports partitions, the stats of the engine also carry how many messages wait in the inboxes and in how many partitions (inboxStats()). The status command uses it for the partitions of a subscriber, e.g.1204 (error: 2), inbox: 3500 in 120, and only loads them with--partitions.Known gap: pause, reactivate, remove and refresh still load all partitions of a subscriber at once. The docs recommend keeping their number bounded with
BucketPartitionerorremoveIdleAfter.