Projections¶

Projections build read models from event streams. waku provides two types:

• Inline projections run synchronously during append_to_stream, guaranteeing immediate consistency between writes and reads.
• Catch-up projections process events asynchronously in a background loop, eventually catching up with the event store.

graph LR
    subgraph Inline
        direction LR
        W[append_to_stream] --> P1[Projection]
        P1 --> RM1[Read Model]
    end
    subgraph Catch-Up
        direction LR
        ES[Event Store] -->|poll| R[Runner]
        R -->|batch| P2[Projection]
        P2 --> RM2[Read Model]
        P2 -->|checkpoint| CP[(Checkpoint Store)]
    end

Inline Projections¶

Implement IProjection to create an inline projection. It runs inside the same scope as append_to_stream, so the read model is always consistent with the write model.

Every projection must define a projection_name class attribute.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from collections.abc import Sequence

from waku.eventsourcing.contracts.event import StoredEvent
from waku.eventsourcing.contracts.stream import StreamId
from waku.eventsourcing.projection.interfaces import IProjection

from app.events import MoneyDeposited, MoneyWithdrawn


class AccountBalanceProjection(IProjection):
    projection_name = 'account_balance'

    def __init__(self) -> None:
        self.balances: dict[StreamId, int] = {}

    async def project(self, events: Sequence[StoredEvent], /) -> None:
        for event in events:
            match event.data:
                case MoneyDeposited(amount=amount):
                    self.balances[event.stream_id] = self.balances.get(event.stream_id, 0) + amount
                case MoneyWithdrawn(amount=amount):
                    self.balances[event.stream_id] = self.balances.get(event.stream_id, 0) - amount

Register inline projections through bind_aggregate() (or bind_decider()):

EventSourcingExtension().bind_aggregate(
    repository=BankAccountRepository,
    event_types=[AccountOpened, MoneyDeposited, MoneyWithdrawn],
    projections=[AccountBalanceProjection],
)

Catch-Up Projections¶

Implement ICatchUpProjection for projections that run asynchronously in a background process. Catch-up projections poll the event store, process events in batches, and checkpoint their progress.

Error handling is configured per-projection via bind_catch_up_projection() (defaults to ErrorPolicy.STOP with no retries — see Error Policies).

Each catch-up projection also has two optional hooks:

• teardown() — called during rebuilds to clean up existing state
• on_skip(events, error) — called when a batch is skipped due to errors

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from collections.abc import Sequence
from dataclasses import dataclass

from waku.eventsourcing.contracts.event import StoredEvent
from waku.eventsourcing.projection.interfaces import ICatchUpProjection

from app.events import AccountOpened, MoneyDeposited, MoneyWithdrawn


@dataclass
class AccountSummary:
    owner: str = ''
    balance: int = 0
    transaction_count: int = 0


class AccountSummaryProjection(ICatchUpProjection):
    projection_name = 'account_summary'

    def __init__(self) -> None:
        self.summaries: dict[str, AccountSummary] = {}

    async def project(self, events: Sequence[StoredEvent], /) -> None:
        for event in events:
            stream_key = event.stream_id.stream_key
            summary = self.summaries.setdefault(stream_key, AccountSummary())

            match event.data:
                case AccountOpened(owner=owner):
                    summary.owner = owner
                case MoneyDeposited(amount=amount):
                    summary.balance += amount
                    summary.transaction_count += 1
                case MoneyWithdrawn(amount=amount):
                    summary.balance -= amount
                    summary.transaction_count += 1

    async def teardown(self) -> None:
        self.summaries.clear()

Register catch-up projections via bind_catch_up_projection():

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from waku.eventsourcing.projection.interfaces import ErrorPolicy

(
    EventSourcingExtension()
    .bind_aggregate(
        repository=BankAccountRepository,
        event_types=[AccountOpened, MoneyDeposited, MoneyWithdrawn],
    )
    .bind_catch_up_projection(
        AccountSummaryProjection,
        error_policy=ErrorPolicy.SKIP,
        max_retry_attempts=3,
    )
)

Event Type Filtering¶

Catch-up projections can declare which event types they handle via the event_types class variable. When set, the projection only receives events of those types — the event store filters at the query level, avoiding unnecessary reads.

class OrderSummaryProjection(ICatchUpProjection):
    projection_name = 'order_summary'
    event_types = [OrderPlaced, OrderShipped]
    ...

When event_types is None (the default), all events are delivered.

Error Policies¶

Both policies retry first when max_retry_attempts > 0. The policy only applies after retries are exhausted.

CatchUpProjectionRunner¶

CatchUpProjectionRunner polls the event store and dispatches event batches to registered catch-up projections. Each projection runs in its own concurrent task.

Use the create() classmethod to build a runner from a DI container:

runner = await CatchUpProjectionRunner.create(
    container,
    lock=InMemoryProjectionLock(),
    projections=[AccountSummaryProjection],  # optional filter; None = all registered
    polling=PollingConfig(),                 # optional; defaults to sensible values
)
await runner.run()

create() resolves the CatchUpProjectionRegistry from the container to discover bindings. When projections is provided, only those projection classes are included.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from waku.di import AsyncContainer
from waku.eventsourcing.projection.lock.in_memory import InMemoryProjectionLock
from waku.eventsourcing.projection.runner import CatchUpProjectionRunner


async def run_projections(container: AsyncContainer) -> None:
    runner = await CatchUpProjectionRunner.create(
        container=container,
        lock=InMemoryProjectionLock(),
    )
    await runner.run()

The runner listens for SIGTERM and SIGINT to shut down. Call request_shutdown() for programmatic shutdown when running inside another async context.

Use rebuild(projection_name) to reprocess all events from the beginning. This calls teardown() on the projection, resets the checkpoint to -1 (nothing processed), and replays every event through the projection.

Configuration¶

All per-projection behavior — batch size, error handling, retry, and gap detection — is configured through bind_catch_up_projection():

The runner's polling interval is configured globally via PollingConfig (passed to the runner constructor, defaults to sensible values if omitted):

Gap Detection¶

When multiple writers append to the event store concurrently, a projection may read events with non-contiguous global positions — a gap appears when a concurrent transaction has not yet committed. Advancing the checkpoint past a gap would permanently skip that event.

Enable gap detection per-projection via bind_catch_up_projection():

(
    EventSourcingExtension()
    .bind_aggregate(...)
    .bind_catch_up_projection(
        AccountSummaryProjection,
        gap_detection_enabled=True,
        gap_timeout_seconds=10.0,
    )
)

When active, the processor queries committed positions from the event store and only advances the checkpoint to the last contiguous position. Gaps are tracked with a timeout — if a gap persists beyond gap_timeout_seconds, it is assumed permanent (e.g., a rolled-back transaction) and skipped.

Distributed Locking¶

IProjectionLock ensures only one instance of each catch-up projection runs at a time across multiple processes. This prevents duplicate processing and checkpoint conflicts.

class IProjectionLock(abc.ABC):
    @contextlib.asynccontextmanager
    async def acquire(self, projection_name: str) -> AsyncIterator[bool]:
        """Yields True if the lock was acquired, False if held by another instance."""

Choosing a Lock¶

graph TD
    Q1{Single process?}
    Q1 -->|Yes| IM[InMemoryProjectionLock]
    Q1 -->|No| Q2{Using PgBouncer<br/>in transaction mode?}
    Q2 -->|Yes| LB[PostgresLeaseProjectionLock]
    Q2 -->|No| Q3{Long-running projections<br/>with many connections?}
    Q3 -->|Yes| LB
    Q3 -->|No| ADV[PostgresAdvisoryProjectionLock]

InMemoryProjectionLock¶

Always acquires the lock immediately. Tracks held lock names for testing. Use this for single-process deployments and in tests.

PostgresLeaseProjectionLock¶

Uses a database table with TTL-based leases for multi-process coordination. A background heartbeat task renews the lease periodically. If the heartbeat detects the lease was stolen (e.g., by another instance after TTL expiry), it cancels the projection task.

Configured via LeaseConfig:

With the defaults, the lease renews every 10 seconds and expires after 30 seconds of silence.

PostgresAdvisoryProjectionLock¶

Uses PostgreSQL advisory locks via pg_try_advisory_lock(hashtext(name)). The lock is session-bound — it holds a database connection for the entire duration.

Checkpoint Store¶

ICheckpointStore tracks each catch-up projection's last processed position so it resumes from where it left off after restarts.

class ICheckpointStore(abc.ABC):
    async def load(self, projection_name: str, /) -> Checkpoint | None: ...
    async def save(self, checkpoint: Checkpoint, /) -> None: ...

The Checkpoint dataclass carries the projection name, last processed global position, and timestamp.

Built-in implementations:

• InMemoryCheckpointStore — dictionary-backed, suitable for single-process deployments and testing
• SqlAlchemyCheckpointStore — PostgreSQL-backed via SQLAlchemy async session

Configure the checkpoint store through EventSourcingConfig:

1
2
3
4
5
6

from waku.eventsourcing import EventSourcingConfig
from waku.eventsourcing.projection.sqlalchemy import make_sqlalchemy_checkpoint_store

es_config = EventSourcingConfig(
    checkpoint_store=make_sqlalchemy_checkpoint_store(checkpoints_table),
)

make_sqlalchemy_checkpoint_store() works the same way as make_sqlalchemy_event_store() — it returns a factory that Dishka uses to construct the store with its AsyncSession dependency injected automatically.

Table Schema Reference¶

es_checkpoints¶

Bind with bind_checkpoint_tables(metadata) from waku.eventsourcing.projection.sqlalchemy.

es_projection_leases¶

Only required when using PostgresLeaseProjectionLock.

Bind with bind_lease_tables(metadata) from waku.eventsourcing.projection.lock.sqlalchemy.

The Event Store as Outbox¶

In event sourcing, the event store is already a durable, ordered log of everything that happened — it naturally serves as a transactional outbox.

The publisher.publish() call in command handlers is an in-process convenience — it dispatches mediator notifications to other handlers within the same process. It is not a reliability mechanism. If the process crashes after saving events but before publishing, the in-process notifications are lost.

For reliable cross-service messaging (e.g., publishing domain events to Kafka, RabbitMQ, or another service), write a catch-up projection that reads from the event store and publishes to your message broker:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from collections.abc import Sequence

from waku.eventsourcing.contracts.event import StoredEvent
from waku.eventsourcing.projection.interfaces import ICatchUpProjection


class OrderEventPublisher(ICatchUpProjection):
    projection_name = 'order_event_publisher'

    def __init__(self, broker: MessageBroker) -> None:
        self._broker = broker

    async def project(self, events: Sequence[StoredEvent], /) -> None:
        for event in events:
            await self._broker.publish(
                topic='orders',
                key=str(event.stream_id),
                value=event.data,
            )

The same at-least-once semantics apply.

Further reading¶

• Snapshots — optimize loading for long-lived aggregates
• Schema Evolution — handling evolved events in projections
• Testing — in-memory stores and projection wait utilities