Refresh Mechanism

DirtyFlag

A DirtyFlag is an entity indicating a dirty partition, which is a partition in a summary table that needs to be refreshed. The dirty flags are written in the same transaction as the changes to the source entities, so if the flush fails, the flags are not saved.

A DirtyFlag can have null partition information. In this case, it means there are new persisted entities that need to be summarized. New entities need to be handled differently because they are not guaranteed to have an ID before flush.

The creation time of a dirty flag is the time of the earliest change not yet reflected in its partition. The debouncing of the refresh, the status command, and CubeCell::getValidAsOf() depend on it. The time columns of the engine tables, rekalogika_summary_dirty.created, rekalogika_summary_refresh_log.refreshed, rekalogika_summary_properties.scheduled_refresh, and rekalogika_summary_properties.source_checked, are TIMESTAMP WITH TIME ZONE, so the times written by the application and by the database triggers are absolute, regardless of the time zone of PHP and of the database session.

SummaryProperties

Stores the state of the refresh of a summary class, in the rekalogika_summary_properties table:

• lastId: the latest partitioning key that has been summarized. The new records are the source records with a greater key.
• sourceChecked: the time just before the refresh last read the latest key of the source, to update lastId, or to find that there are no new records. Every source record with a key greater than lastId was inserted after this time.
• scheduledRefresh: the time the refresh agent is scheduled to run next, see Scheduling.

SourceEntityListener

Detects the changes to the source entities, and to the entities the summaries read through an association. The user-facing behavior is described in Change Detection.

On Doctrine's onFlush event, the listener goes through the scheduled insertions, updates, and deletions of the unit of work:

• For a deleted source entity, it creates a DirtyFlag for the partition of the entity.
• For an updated source entity, it checks the change set against the involved properties of the summaries. If a summary is affected, it creates a DirtyFlag for the partition of the entity. If the partitioning key is modified, it also creates one for the partition of the original value.
• For an inserted source entity, it creates a DirtyFlag with null partition information, because the entity might not have an ID yet.
• For an updated or deleted entity that a summary reads through an association, it records the entity, grouped by the summary and the reverse path from the source entity. Then ReferringSourcePartitionsQuery runs one query for each group, to find the lowest partitions of the source entities referring to the recorded entities, and the listener creates a DirtyFlag for each of them. The queries run before the flush is executed, so the references to the deleted entities are still in the database. They do not apply the filter of the summary, because the change can move a source entity out of the filter. These dependencies are resolved from the Doctrine association mappings by AssociationDependencyResolver, which only follows to-one associations. A summary with trackAssociations: false has none.

The flags are persisted in the same onFlush event, so they are flushed with the rest of the entities.

On the postPersist event, after the insert query of a new source entity is executed, its key is known. The listener then inserts the DirtyFlag of its partition using DBAL, in the same transaction. Unlike the flag with null partition information, this one does not depend on the latest summarized key. It covers an entity whose transaction commits after a refresh has already moved the latest summarized key past the key of the entity, e.g. when concurrent transactions commit out of ID order.

At the end of a flush, the listener will dispatch a NewDirtyFlagEvent event for each of the created DirtyFlag objects, and also a DirtySummaryEvent for each of the summary tables that need to be refreshed.

The changes made outside the ORM are not seen by the listener. They can be reported using SummaryManager::markDirty() and markDirtyRange(), which insert the dirty flags using DBAL and dispatch the same events, or recorded by the PostgreSQL triggers generated by TriggerSqlGenerator, which insert the dirty flags in the database without dispatching any event.

DirtySummaryEventListener

Listens on the DirtySummaryEvent event. It calls RefreshAgentRunner::refresh() to run the refresh agent for the summary table, if one is not already running.

Refresh Agent

The refresh agent runs in the background, in a separate process from the one that modified the source entities. It is dispatched by RefreshAgentRunner as a RefreshAgentStartCommand to be executed by RefreshAgent.

Locking

Only one refresh agent can run for a summary class at a time. This is enforced by RefreshAgentLock. The lock is always released after the agent finishes, even if the refresh fails. The manual refresh commands, the truncate command, and SummaryManager::refresh() also wait for and hold the same lock, because concurrent refreshes of the same partition produce duplicate rows.

There are two implementations of RefreshAgentLock:

• PostgreSqlRefreshAgentLock, the default. It uses PostgreSQL session-level advisory locks on the connection of the summary, so it does not need any other infrastructure. The lock is released when the database session ends, e.g. when the process dies. If the refresh fails and leaves the connection unusable, e.g. in an aborted transaction, the lock is released by closing the connection.
• SymfonyRefreshAgentLock, used if refresh_lock.factory is configured. It uses a LockFactory from symfony/lock, so any of its stores can be used. With an expiring store, the lock expires after refresh_lock.ttl seconds, 300 by default. The lock is extended at the start of every step of a refresh, so the TTL only needs to cover a single step, not the entire refresh. The store should not use the same database connection as the summary: if the refresh fails in an aborted transaction, the lock cannot be released.

config/packages/rekalogika_analytics.yaml

rekalogika_analytics:
    refresh_lock:
        # The service ID of a symfony/lock LockFactory. If null, the default
        # PostgreSQL advisory lock is used.
        factory: lock.factory
        # Used with factory. Null disables expiration, if the store supports
        # it.
        ttl: 300

The refresh agent and the locks log their activities to the rekalogika.analytics Monolog channel.

Debouncing Source Changes

Refreshing a partition every time its source records change would be wasteful if the records are changed frequently. Instead, the refresh of the lowest-level partitions is debounced according to the RefreshAgentStrategy:

• minimumAge: a dirty partition is not refreshed until its oldest dirty flag is at least this old. Defaults to 10 minutes.
• minimumIdleDelay: a dirty partition is not refreshed until there have been no new dirty flags for it for this long. Defaults to 5 minutes.
• maximumAge: once the oldest dirty flag of a partition is this old, the minimumIdleDelay requirement is waived, so that a partition that is continuously receiving changes is not postponed indefinitely. Defaults to 30 minutes.

Multiple changes to the same partition are coalesced into a single refresh. In a busy table, a partition receiving continuous changes is refreshed at most once every maximumAge, and its summary lags behind the source by at most the same amount.

New records are recorded as dirty flags without partition information, because they might not have an ID before the flush. The refresh agent converts them to dirty flags of the partitions of the new records. The converted flags inherit the creation time of the original flags, so the debouncing is based on the time the records were created, not the time the agent converts them.

To find the new records, the agent reads the latest key of the source, bypassing any cache, and records the time just before it in sourceChecked. New records without any dirty flag, e.g. inserted using native SQL without the database triggers, are converted to dirty flags with the time of the previous check of the source, because they were inserted after it. If the source has never been checked, their time is unknown, and the earliest possible time is used, so they are refreshed without waiting for minimumAge.

Scheduling

The refresh agent runs only when there is something to refresh:

• After source changes are flushed, RefreshAgentRunner schedules the agent to run when the new dirty flags will satisfy the strategy, unless the agent is already scheduled to run before that.
• Before exiting, the agent determines the earliest time one of the remaining dirty partitions will be ready to be refreshed, and schedules itself to run again at that time, unless it is already scheduled to run before that. Therefore, a refresh is never lost even if no more changes are made to the source.
• If the agent cannot acquire the lock because another refresh is running, it schedules itself to run again in 10 seconds. The other refresh might have missed the dirty flags created after it started.
• If the run fails, the agent schedules another run 5 minutes later, then rethrows the exception. Otherwise, the remaining dirty flags would wait until the next change to the source. Messenger's retry strategy also applies to the failed message.

The time of the next scheduled run is stored in SummaryProperties. This keeps the number of pending runs per summary class small, regardless of how often the source changes. A scheduled run that is lost, for example because the worker crashed, does not stop the refresh permanently: once the scheduled time has passed, the next change to the source schedules a new run, and the sweeper dispatches a new run even if there are no more changes.

Sweeper

The refresh agent is dispatched after a flush, or a call to SummaryManager::markDirty(), creates dirty flags. That does not happen if the dirty flags are created in another way, e.g. by the database triggers. And a dispatched message can be lost, or end up in the failure transport. In these cases, the dirty flags would wait until the next change to the same summary.

RefreshAgentSweeper dispatches the refresh agent for every summary that has dirty flags but no pending run. A run is considered pending if it is scheduled in the future, or if it is overdue by less than refresh.sweep_grace, 300 seconds by default, because the message might still be in the queue. Dispatching a duplicate run is harmless: it either finds the lock held and reschedules itself, or refreshes only the partitions that are ready.

The sweeper runs from the rekalogika:analytics:sweep command, or from a RefreshAgentSweepCommand message, e.g. sent periodically using Symfony Scheduler.

Propagating to Upper Partitions

After a partition is refreshed, its parent partition is marked as dirty, if the parent partition is used by queries. Queries use an upper partition only if it is complete, i.e. it is entirely before the latest summarized record. Therefore, appending new records only causes an upper partition refresh once, when the upper partition becomes complete. Most upper partition refreshes are caused by changes to older records.

Unlike the lowest-level partitions, the dirty upper partitions are caused by the refresh itself, not by changes in the source, so they are not subject to the RefreshAgentStrategy debouncing. Instead, they are debounced using a cooldown period, upperPartitionCooldown, defaulting to 5 minutes:

• An upper partition is refreshed immediately after its children, in the same run, unless it has been refreshed within the cooldown period.
• If it has been refreshed within the cooldown period, its dirty flag is kept, and the agent schedules itself to refresh it when the cooldown period ends.

The upper partitions are processed from the lowest to the highest level, so the propagation reaches the top level in a single run. All the partitions refreshed in a run share the same refresh time, so their cooldown periods end at the same time. Usually, the entire chain of upper partitions is refreshed in the same run, and the delays do not accumulate across levels. Each upper partition lags behind the partitions below it by at most the cooldown period, while each upper partition is refreshed at most once per cooldown period, regardless of how the changes in the source are distributed.

The dirty flag of an upper partition, created when a partition below it is refreshed, inherits the creation time of the earliest dirty flag of that partition, not the time of the refresh. Therefore, the time of every dirty flag is the time of the earliest change not yet reflected in its partition, at every level.

The refresh times of the upper partitions are recorded in RefreshLog. Records older than the cooldown period are removed by the agent.

Worst-Case Lag

The lag is the time between a change being committed to the source table and the change being reflected in the query results. Its theoretical maximum is determined as follows.

Lowest-level partition. A dirty partition is ready to be refreshed at min(latest + minimumIdleDelay, earliest + maximumAge), but not before earliest + minimumAge, where earliest and latest are the creation times of its oldest and newest dirty flags. The dirty flag of a change is never older than the oldest dirty flag of its partition, so the change is refreshed at most max(minimumAge, maximumAge) after it is made. The agent is scheduled to run exactly at that time.

Upper partitions. After a partition is refreshed, its parent is refreshed in the same run, or at most upperPartitionCooldown later if the parent is in its cooldown period. Usually, the entire chain is refreshed in the same run. But in the worst case, every level waits for a full cooldown period: the parent was refreshed just before the child, the grandparent was refreshed, through another child, just before the parent, and so on. It requires frequent changes to many partitions, which is the case in a busy table. Therefore, the upper partitions add at most (L - 1) × upperPartitionCooldown, where L is the number of partition levels.

Combined:

maximum lag = max(minimumAge, maximumAge)
            + (L - 1) × upperPartitionCooldown
            + overhead

With the default strategy:

Partition class	Levels	Maximum lag (excluding overhead)
UuidV7IntegerPartition	4	30 + 3 × 5 = 45 minutes
DefaultIntegerPartition	5	30 + 4 × 5 = 50 minutes
SingleLevelIntegerPartition	1	30 minutes

The upper partition term only applies to changes to older records. New records only affect upper partitions that are not complete yet, and are not used by queries. When an upper partition becomes complete, it has never been refreshed, so it is not in the cooldown period. Therefore, new records are reflected in the query results at most max(minimumAge, maximumAge) plus overhead after they are created.

The overhead is not bounded by the framework:

• The time the agent takes to refresh the partitions, including waiting for another refresh to finish. A run refreshes all the partitions that are ready, so this grows with the number of partitions ready at the same time.
• The delivery latency of Symfony Messenger, e.g. the availability of the workers, the backlog of the queue, and the polling interval of the transport.
• Up to one second, because the timestamps are stored with second precision.
• If the lock is held by another refresh, e.g. the manual rekalogika:analytics:refresh command, up to 10 seconds after it finishes, because the agent retries every 10 seconds while the lock is held.
• If a run fails, up to 5 minutes until the retry, assuming the cause of the failure has gone away by then.
• If the message dispatching the agent is lost, until the sweeper dispatches a new run: up to sweep_grace after the scheduled time, plus the interval at which the sweeper runs.

The lag is unbounded if:

• maximumAge is null while minimumIdleDelay is not. A partition that receives changes more often than minimumIdleDelay is never refreshed.
• No worker is consuming the messages.
• A scheduled run is lost, no more changes are made to the source, and the sweeper is not running. The remaining dirty partitions are refreshed only after the next change, or by the manual refresh command.
• The dirty flags are created by the database triggers, and the sweeper is not running.

The manual rekalogika:analytics:refresh command does not use the debouncing nor the cooldown period, it refreshes all the dirty partitions immediately.

Validity Time

CubeCell::getValidAsOf() returns the time before which every change to the source is known to be reflected in a query result. It is determined by ValidAsOfQuery, once for each query, before querying the summary table. It is the earliest of:

• the time of the query,
• the creation time of the earliest dirty flag of the summary class,
• if the source has records with a key greater than lastId, the sourceChecked time.

The reads are ordered so that a refresh running concurrently can only make the time earlier than necessary, never later:

1. The time of the query. A change made before it is visible to the following reads.
2. lastId and sourceChecked, read together.
3. The latest key of the source. If it is greater than lastId, the records after lastId were inserted after sourceChecked, and may have no dirty flags.
4. The dirty flags. The records up to lastId are either flagged, or already reflected in the summary table, which is read last.

If there are new records, but sourceChecked is null, the time cannot be determined, and the method returns null. The changes that have no dirty flags, other than the new records, are not taken into account.