Articles · Projections & Read Models ·031

Intra-Projection Sharding: Parallel Lanes for One Hot Projection

Runnable sample on GitHub

Sample: samples/031-intra-projection-sharding · Prerequisites: 007 — Projections, 008 — Projection Operations

Overview

Article 008 makes a deliberate promise: a projection has exactly one active worker. When you scale out, the per-checkpoint advisory lock (AddRelayProjectionCoordinationPostgres()) ensures only one instance owns a given projection at a time, so its checkpoint advances monotonically and events apply in order. That single-owner model is what keeps a read model correct.

It is also a throughput ceiling. A projection that one worker can’t keep up with — a hot projection — has no relief valve in the single-owner model: adding instances just changes which one box does all the work. Scaling out across projections (one worker each) doesn’t help when the bottleneck is a single projection.

Intra-projection sharding lifts that ceiling for one projection without giving up ordering. Split the projection into a fixed number of lanes, assign each event to a lane by a stable hash of its aggregate id, and run one worker per lane. Per-aggregate order is preserved (an aggregate’s events never split across lanes); different aggregates fan out and run concurrently. Each lane has its own checkpoint, so the existing per-checkpoint advisory lock coordinates one worker per lane — the single-owner model, applied per lane instead of per projection.

The mechanism is a pure, DB-free primitive: ProjectionPartitioner (an IProjectionPartitioner).

Why this exists

A projection’s throughput is capped by one worker (article 008). That is the right default — it makes ordering and checkpointing trivial — and for almost every projection it is plenty. But one or two projections in a busy system are hot: the order read model, the per-customer ledger, the search-index feeder. Those projections fall behind under load, and the single-owner model gives you nowhere to go: you can’t add a second worker to the same projection without two workers racing the same checkpoint and reapplying events out of order.

The naive fixes are wrong:

  • Drop the lock and run N workers on one checkpoint. They race: two workers read the same batch, both advance the checkpoint, events apply twice and out of order. The read model corrupts.
  • Split by event type. Order matters per aggregate, not per type; splitting OrderPlaced from OrderShipped reorders an order’s lifecycle.
  • Just make the projection faster. Sometimes you can (batch the writes, drop an index). When you can’t, you need parallelism — and parallelism that keeps per-aggregate order.

Sharding by aggregate id is the correct parallelism: it preserves the only ordering guarantee that matters (within an aggregate) while letting unrelated aggregates proceed in parallel.

When to use this

  • One projection is the bottleneck. Its lag grows under load while the rest of the system keeps up. Sharding gives that projection N workers without touching the others.
  • Ordering is per-aggregate, not global. Your projection only needs events for a given aggregate applied in order (the common case) — not a single global order across all aggregates.
  • You can pick a lane count up front. A fixed, generous lane count (powers of two are convenient) sized for your peak fan-out.

When not to use this

  • The projection isn’t hot. A projection that keeps up with one worker should stay on one worker. Sharding adds N checkpoints, N locks, and N workers’ worth of overhead and operational surface — pay it only where the throughput ceiling actually bites.
  • You need global ordering. If the projection must apply all events in one total order (e.g. a global running sequence number across aggregates), lanes break it — different lanes advance independently. Keep it single-owner.
  • The bottleneck is the database, not the worker. If one worker already saturates the read store’s write capacity, more lanes just contend on the same database. Fix the write path first; sharding multiplies a worker that wasn’t the limiting resource.
  • You’d shard every projection. That’s just running more instances; the framework already spreads distinct projections across instances. Sharding is a targeted tool for one hot projection.

The cost: a fixed lane count you must size in advance (changing it reshuffles aggregate→lane ownership unless you use consistent hashing), ordering guaranteed only within a lane, and the operational reality of N checkpoints and N workers to monitor instead of one.

Concepts

Lane assignment by aggregate id. ProjectionPartitioner.PartitionFor(EventData @event) hashes the event’s AggregateId and returns a lane in [0, PartitionCount). Because the lane derives from the aggregate id (not the event id or position), every event of an aggregate maps to the same lane — so one worker applies that aggregate’s events in order. The hash is stable (FNV-1a over the aggregate id’s bytes, not string.GetHashCode(), which is randomized per process), so the same aggregate lands in the same lane in every process and after every restart — the property article 023 calls out as essential for keyed consumers.

Per-lane checkpoints. CheckpointName(projectionName, partition) returns {projection}#p{n} — e.g. orders#p0, orders#p3. Each lane is, to the rest of the system, a normal projection checkpoint with its own name. That is what lets the existing machinery treat each lane independently: each lane advances its own checkpoint, dead-letters into its own scope, and rebuilds on its own.

One worker per lane, coordinated by the existing lock. A sharded host doesn’t need new coordination infrastructure. It runs one projection worker per lane, each contending for the per-checkpoint advisory lock on its lane’s checkpoint name (AddRelayProjectionCoordinationPostgres(), article 008/023). Because lane checkpoints are distinct names, the locks are distinct: instance A can own orders#p0 while instance B owns orders#p3. The single-owner guarantee now holds per lane — exactly one worker advances each lane’s checkpoint — so the read model stays correct, just N-way parallel.

Composing a partitioner. ProjectionPartitioner is a thin shell over an IPartitioner (Nuvora.Nexus.Relay.Core.Partitioning). The default is the FNV-1a hash-modulo DefaultPartitioner. Pass a ConsistentHashPartitioner when you expect to resize the lane count: hash-modulo remaps almost every aggregate when PartitionCount changes (reshuffling ordering across the change), whereas a consistent-hash ring moves only ≈1/N of aggregates — far less rebalancing churn, at a little load skew.

Architecture

flowchart TD
    ES[(Event store)] -->|catch-up read| H[Sharded projection host]
    H -->|PartitionFor aggregate id| PP{ProjectionPartitioner<br/>stable hash mod N}

    PP -->|lane 0| W0[Worker · checkpoint 'orders#p0']
    PP -->|lane 1| W1[Worker · checkpoint 'orders#p1']
    PP -->|lane 2| W2[Worker · checkpoint 'orders#p2']

    W0 -->|pg_advisory_lock 'orders#p0'| L0[(one owner per lane)]
    W1 -->|pg_advisory_lock 'orders#p1'| L1[(one owner per lane)]
    W2 -->|pg_advisory_lock 'orders#p2'| L2[(one owner per lane)]

    W0 --> RM[(Read model)]
    W1 --> RM
    W2 --> RM

    A1[aggregate 'order-42' all events] -.always same lane.-> W1

Every event of order-42 always reaches the same lane (W1 here), so its order is preserved; unrelated aggregates spread across lanes 0/1/2 and apply concurrently. Each lane’s worker holds its own advisory lock, so the single-owner-per-checkpoint guarantee of article 008 now holds per lane.

Building it step by step

The sample is samples/031-intra-projection-sharding. It is pure — no host, no database — because the sharding logic (which lane, which checkpoint) is the part worth proving in isolation.

1. Construct a partitioner with a fixed lane count

var partitioner = new ProjectionPartitioner(partitionCount: 8);   // 8 lanes for one hot projection

PartitionCount is the parallelism dial. Size it up front for peak fan-out and leave it (see Production considerations on resizing).

2. Assign an event to its lane

int lane = partitioner.PartitionFor(@event);   // in [0, 8); stable per aggregate id

Every event of an aggregate returns the same lane — that is what preserves per-aggregate order while letting different aggregates run in parallel.

3. Name each lane’s checkpoint

string checkpoint = partitioner.CheckpointName("orders", lane);   // e.g. "orders#p3"

A sharded host runs one worker per lane, each driving the projection but checkpointing under its lane’s name — so the per-checkpoint advisory lock (article 008) gives each lane exactly one active worker.

4. (Optional) compose a consistent-hash partitioner for cheap resizing

var partitioner = new ProjectionPartitioner(8, new ConsistentHashPartitioner());

Now changing the lane count later moves only ≈1/N of aggregates instead of nearly all of them.

Complete source code

File Shows DB?
Projections.Sharding.Tests/ProjectionShardingTests.cs lanes in range, per-aggregate stability, fan-out, distinct checkpoints, consistent-hash composition, guards No
Projections.Sharding.Tests/Projections.Sharding.Tests.csproj references Core + EventStore + Projections (pure, no host) No

Running the example

dotnet test samples/Relay.Samples.slnx

The sharding tests need only the .NET 10 SDK — there is no database, no broker, and no Docker, because the partitioner is a pure primitive.

Testing

The sample proves the properties a sharded host depends on, all without infrastructure:

Per-aggregate order is preserved. Every event of one aggregate must land on the same lane — otherwise two workers would apply an aggregate’s events and reorder them:

var partitioner = new ProjectionPartitioner(partitionCount: 8);
var aggregateId = Guid.NewGuid();

var first = partitioner.PartitionFor(EventFor(aggregateId));
var second = partitioner.PartitionFor(EventFor(aggregateId)); // different event, same aggregate
second.Should().Be(first, "every event of an aggregate must share a lane to preserve per-aggregate order");

Different aggregates fan out. Parallelism is real only if distinct aggregates spread across lanes rather than collapsing onto one:

var lanesUsed = Enumerable.Range(0, 500)
    .Select(_ => partitioner.PartitionFor(EventFor(Guid.NewGuid())))
    .Distinct();
lanesUsed.Should().HaveCount(8, "a stable hash of distinct aggregate ids should reach every lane");

Each lane checkpoints distinctly. Distinct names are what make distinct advisory locks — one owner per lane:

partitioner.CheckpointName("orders", 3).Should().Be("orders#p3");
Enumerable.Range(0, 4).Select(p => partitioner.CheckpointName("orders", p))
    .Should().OnlyHaveUniqueItems();

It is tested as a pure primitive (no host, no database) deliberately — the same split article 023 uses for the partitioner: push the logic into a fast, deterministic primitive and keep the irreducible database interaction (the advisory lock, the checkpoint commit) in the integration tests it already has.

Production considerations

  • Fix the lane count up front. With the default hash-modulo partitioner, changing PartitionCount reshuffles aggregate→lane ownership, which reorders the in-flight rebalance and forces a coordinated rebuild. Size lanes for peak fan-out and leave them — or use ConsistentHashPartitioner so a resize moves only ≈1/N of aggregates.
  • Ordering holds only within a lane. A sharded projection no longer applies events in one global order — only per aggregate. If anything downstream assumes a single total order across aggregates, sharding breaks it. Confirm your read model only needs per-aggregate order before sharding.
  • Rebalancing is a rebuild, not a live hand-off. This is fixed-lane sharding: there is no live ownership hand-off when the lane count changes. To resize, you rebuild (article 008) under the new lane count. Consistent hashing shrinks how much must move, but it is still an operational event — plan it.
  • One worker per lane still needs the advisory lock. The correctness of sharding rests entirely on each lane’s checkpoint having exactly one active worker. Run AddRelayProjectionCoordinationPostgres() — without it, two instances can both own orders#p3 and you’ve reintroduced the race sharding was meant to avoid.
  • Watch per-lane lag, not just aggregate lag. A hot aggregate can make one lane lag while the others are idle (load skew). Monitor each lane’s checkpoint lag; persistent single-lane skew is a signal to raise the lane count or revisit the key.
  • Keep the projection idempotent (article 008). Sharding doesn’t change the rebuild contract: each lane can be reset and replayed independently, so the projection must be idempotent and side-effect-free, exactly as 008 requires.

Common mistakes

  • Sharding by event type instead of aggregate id. It looks like parallelism, but it reorders an aggregate’s lifecycle (OrderPlaced in one lane, OrderShipped in another). Shard by aggregate id; order is per aggregate.
  • Dropping the per-checkpoint lock to “go faster”. N workers on N lanes is correct because each lane has one owner. Run two workers on one lane (no lock) and that lane’s checkpoint races — the corruption sharding was built to prevent.
  • Using string.GetHashCode() to pick the lane. It is randomized per process, so the same aggregate lands on different lanes in different instances — shredding per-aggregate order the moment you scale to two boxes. ProjectionPartitioner uses a stable FNV-1a hash for exactly this reason.
  • Resizing lanes casually under hash-modulo. Bumping PartitionCount from 8 to 16 remaps almost every aggregate. Without consistent hashing and a planned rebuild, you reorder events across the change.
  • Sharding a projection that isn’t hot. You’ve added N checkpoints, N locks, and N workers to monitor for a projection one worker handled fine. Shard the bottleneck, not everything.

Tradeoffs

Benefits. A single hot projection scales past one worker without losing per-aggregate order; the parallelism reuses the existing per-checkpoint advisory lock (no new coordination infrastructure); each lane checkpoints, dead-letters, and rebuilds independently; and the assignment is a pure, stable primitive that is trivial to test and identical across every node.

Costs. A fixed lane count you must size in advance and can only change via a planned rebuild (less painful with consistent hashing); ordering guaranteed only within a lane, not globally; N checkpoints and N workers to operate and monitor instead of one; and load skew when a few aggregates are far hotter than the rest.

Alternatives

  • Stay single-owner and optimize the projection. Batch the writes, drop an unused index, denormalize less. Always try this first — it has none of sharding’s operational cost. Sharding is for when the work genuinely exceeds one worker.
  • Split into multiple distinct projections. If the “hot” projection is really several read models bundled together, split them; the framework already runs distinct projections on different instances. This is cleaner than sharding when the work decomposes by read model — sharding is for when one indivisible read model is the bottleneck.
  • A dedicated stream processor (Kafka Streams, Flink). Purpose-built partitioned, ordered consumers with dynamic rebalancing. Worth it at very large scale or when you already run the platform; heavy new infrastructure for a Relay service that can shard one projection over the database it already has.
  • Dynamic repartitioning with live ownership hand-off. Move lane ownership between instances without a rebuild. Far more complex (ownership transfer, in-flight draining); fixed lanes plus consistent hashing covers the vast majority of cases at a fraction of the complexity.

Lessons from production systems

  • The hot projection is almost always one or two, not all of them. Teams that reach for “scale every projection” pay broad overhead; the ones that profile, find the single bottleneck, and shard just that one get the throughput without the operational sprawl.
  • Per-aggregate ordering is the line you must not cross. Every sharding incident traces back to a split that reordered an aggregate (sharding by type, by event id, by a per-process hash). Shard by a stable hash of the aggregate id and the ordering bugs disappear.
  • Resizing lanes is the part people underestimate. “We’ll just add lanes later” turns into a full rebuild under hash-modulo. Teams that picked a generous fixed count up front — or opted into consistent hashing — avoided the surprise; the rest learned it the hard way.

Should you use this?

If you… Then…
have one projection whose lag grows under load yes — shard it into lanes
only need per-aggregate ordering (the common case) yes — lane-by-aggregate-id preserves it
need a single global order across all aggregates no — keep it single-owner
are bottlenecked on the read database, not the worker fix the write path first; sharding won’t help
expect to resize the lane count yes, but compose a ConsistentHashPartitioner and plan the rebuild
would shard every projection no — that’s just more instances; the framework already spreads projections

Next steps

  • 008 — Projection Operations: the single-owner model, per-checkpoint advisory lock, dead-lettering, and rebuilds that each lane inherits.
  • 023 — Distributed Coordination: the partitioner, leader election, and advisory locks this builds on — and why a stable hash (not GetHashCode()) is non-negotiable for keyed consumers.