[Feature] Sharded read-side projections with rebalance

[pathosDev]

Size / Priority

• Size: L
• Reference: Lagom Read-Side Processors, Akka Projections.

Rationale

Existing read-side options:

• PersistenceQuery.eventsByTag — a single consumer reads tagged events sequentially. Single-node, single-thread.
• PersistenceQuery.eventsByPersistenceId — per-pid consumer; same single-node limitation.

For high-volume read-models, this doesn't scale. A 10K-event/sec event stream needs:

• N parallel processors (one per pid-shard).
• Cluster-wide distribution — processors run on different nodes.
• Rebalance on node loss — orphan shards picked up by surviving nodes.
• Cursor durability — per-shard cursor persisted; rebalance resumes from cursor.

Sharded projections cover this. Each shard is a PersistentActor that consumes events for its assigned pids; the cluster sharding subsystem distributes shards across nodes; cursor stored alongside shard state.

Reference: what Lagom does

class OrderReadSideProcessor extends ReadSideProcessor[OrderEvent] {
  def aggregateTags: Set[AggregateEventTag[OrderEvent]] = OrderEvent.Tag.allTags

  def buildHandler(): ReadSideHandler[OrderEvent] = readSide.builder[OrderEvent]("orderOffset")
    .setEventHandler[OrderCreated](e => updateReadModel(e.event))
    .setEventHandler[OrderShipped](e => markShipped(e.event))
    .build()
}
// Lagom auto-distributes the projection across the cluster, one slot per shard.

Design sketch — actor-ts equivalent

// src/projection/ShardedProjection.ts (new)

export interface ShardedProjectionSpec<TEvent> {
  /** Name — used for cluster-sharding type name. */
  readonly name: string;
  /** Tag(s) to consume from PersistenceQuery. */
  readonly tags: ReadonlyArray<string>;
  /** Number of shards.  Default: 100. */
  readonly numShards?: number;
  /** Compute shard id from a pid (consistent hashing). */
  readonly shardOf?: (pid: string) => number;
  /** Per-event handler.  Receives one event at a time; called per pid in order. */
  readonly handle: (event: PersistentEvent<TEvent>) => Promise<void>;
  /** Cursor store — where to persist per-shard cursors. */
  readonly cursorStore: ProjectionCursorStore;
}

export interface ProjectionCursorStore {
  load(projectionName: string, shardId: number): Promise<Offset | undefined>;
  save(projectionName: string, shardId: number, cursor: Offset): Promise<void>;
}

// Start the projection:
export function startShardedProjection<TEvent>(
  cluster: Cluster,
  spec: ShardedProjectionSpec<TEvent>,
): Promise<ProjectionHandle>;

Internal design:

1. Cluster sharding creates numShards shard instances of a ProjectionShardActor.
2. Each ProjectionShardActor on preStart:
   • Loads cursor from store.
   • Polls PersistenceQuery.eventsByTag(spec.tags, cursor).
   • Per event: filters to "is this pid in my shard?"; calls spec.handle(event); updates cursor on success.
3. On rebalance: actor stops on losing node; resumes on gaining node from cursor.

class ProjectionShardActor<TEvent> extends PersistentActor<...> {
  private cursor: Offset = offsetStart;

  override async preStart() {
    this.cursor = await this.cursorStore.load(this.spec.name, this.shardId) ?? offsetStart;
    this.scheduler.scheduleAtFixedRateFn(1000, 1000, () => this.poll());
  }

  private async poll(): Promise<void> {
    const events = await this.persistenceQuery.currentEventsByTag(this.spec.tags, this.cursor);
    for (const evt of events) {
      if (this.shardOf(evt.event.persistenceId) !== this.shardId) continue;
      try {
        await this.spec.handle(evt.event);
        this.cursor = evt.offset;
        await this.cursorStore.save(this.spec.name, this.shardId, this.cursor);
      } catch (e) {
        this.log.warn(`projection handle failed`, e);
        return; // retry next tick from same cursor
      }
    }
  }
}

Integration with existing actor-ts subsystems

• Cluster Sharding: provides distribution + rebalance.
• PersistenceQuery: source of events.
• PersistentActor: cursor durability via journal (cursor IS the actor's state).
• PersistenceQuery.AllPersistenceIds ([Feature] PersistenceQuery: AllPersistenceIds live stream + cursor-paginated currentPersistenceIds #156): complementary for fan-out-per-pid projections.

Out of scope / non-goals

• Cross-cluster projection — single cluster.
• Exactly-once handler invocation — at-least-once; handler must be idempotent.
• Streaming aggregations — use existing PersistenceQuery + actor logic; sharded projection is for "process every event, write to read-model".

Open design questions

1. Shard count: fixed at start (sketch) or dynamic. Recommend fixed — matches sharding semantics.
2. Per-shard parallelism: each shard processes events sequentially. To parallelise within a shard, use a router behind the handler. Out of scope; document.
3. Backpressure: slow handler → events back up; next poll skipped if previous still running. Built-in.
4. Cursor persistence frequency: per-event (sketch, slow) vs batch. Recommend batch (every N events or every M seconds), with at-least-once handler implication.

Test plan

1. Single-shard projection — handler called for every event in order.
2. Multi-shard — events distributed by shardOf; each shard sees only its events.
3. Cluster of 3 nodes, 100 shards — shards distributed; each node hosts ~33.
4. Kill a node — orphan shards relocate; cursor preserved; resume from last.
5. Handler failure → retry from cursor; eventual success advances cursor.
6. Stress: 100K events/sec sustained across 100 shards.
7. Cross-runtime parity.

Acceptance criteria

• ShardedProjectionSpec<TEvent> + ProjectionCursorStore.
• startShardedProjection factory.
• Cluster sharding integration.
• Cursor durability + resume on rebalance.
• Metrics: per-shard lag gauge, throughput counter.
• Documentation: "Single-node projection vs ShardedProjection" decision guide.
• Test suite (7 cases).
• CHANGELOG entry under "New: Sharded read-side projections".

Pre-implementation checklist

• Joint review.
• Cursor-persistence frequency policy decided.
• Coordinate with [Feature] PersistenceQuery: AllPersistenceIds live stream + cursor-paginated currentPersistenceIds #156 (AllPersistenceIds) for fan-out projections.