Change: Deliver events to worker in batches during restoring phase

Author: snatalenko

src/workers/AbstractWorkerProjection.ts

@@ -28,7 +28,7 @@ export abstract class AbstractWorkerProjection<TView>
 
 		const projectionMethodsToWire = [
 			'project',
-			'_project',
+			'_projectBatch',
 			'ping',
 			'getLastProjectedEvent'
 		] as Extract<keyof T, string>[];
@@ -62,9 +62,10 @@ export abstract class AbstractWorkerProjection<TView>
 		return true;
 	}
 
-	/** @internal Expose protected projection path for worker RPC wiring */
-	public override _project(event: IEvent, meta?: Record<string, any>): Promise<void> {
-		return super._project(event, meta);
+	/** @internal Project restore events in batches to avoid one Comlink roundtrip per event */
+	async _projectBatch(events: IEvent[]): Promise<void> {
+		for (const event of events)
+			await this._project(event);
 	}
 
 	/**

src/workers/WorkerProxyProjection.ts

@@ -1,6 +1,12 @@
 import { Worker, MessageChannel } from 'node:worker_threads';
 import type {
-	IEvent, IEventStorageReader, IEventStore, IExtendableLogger, ILogger, IViewLocker
+	IEvent,
+	IEventSet,
+	IEventStorageReader,
+	IEventStore,
+	IExtendableLogger,
+	ILogger,
+	IViewLocker
 } from '../interfaces/index.ts';
 import type { IProxyProjection, IWorkerProjection, ProxyProjectionParams } from './interfaces/index.ts';
 import * as Comlink from 'comlink';
@@ -18,6 +24,8 @@ export class WorkerProxyProjection<
 	TProjection extends IWorkerProjection<TView> = IWorkerProjection<TView>
 > implements IProxyProjection<TView> {
 
+	static RESTORE_BATCH_SIZE = 5_000;
+
 	#worker?: Worker;
 	readonly #workerInit: Promise<Worker>;
 	readonly #remoteProjection: Comlink.Remote<TProjection>;
@@ -83,7 +91,12 @@ export class WorkerProxyProjection<
 			this.viewLocker.unlock();
 	}
 
-	/** Restore view state from not-yet-projected events */
+	/**
+	 * Restore view state from not-yet-projected events.
+	 *
+	 * Events are projected in batches to reduce worker RPC overhead.
+	 * The batch size can be configured through {@link WorkerProxyProjection.RESTORE_BATCH_SIZE}.
+	 */
 	protected async _restore(eventStore: IEventStorageReader): Promise<void> {
 		if (!this.#worker)
 			await this.#workerInit;
@@ -92,17 +105,25 @@ export class WorkerProxyProjection<
 		const lastEvent = await this.#remoteProjection.getLastProjectedEvent();
 
 		this.#logger?.debug(`retrieving ${lastEvent ? `events after ${describe(lastEvent)}` : 'all events'}...`);
-
 		const eventsIterable = eventStore.getEventsByTypes(this.#messageTypes, { afterEvent: lastEvent });
 
 		let eventsCount = 0;
 		const startTs = Date.now();
+		const batch: IEvent[] = [];
 
 		for await (const event of eventsIterable) {
-			await this._project(event);
+			batch.push(event);
 			eventsCount += 1;
+
+			if (batch.length >= WorkerProxyProjection.RESTORE_BATCH_SIZE) {
+				await this._projectBatch(batch);
+				batch.length = 0;
+			}
 		}
 
+		if (batch.length)
+			await this._projectBatch(batch);
+
 		this.#logger?.info(`view restored from ${eventsCount} event(s) in ${Date.now() - startTs} ms`);
 	}
 
@@ -134,8 +155,8 @@ export class WorkerProxyProjection<
 		return this.#remoteProjection.project(event);
 	}
 
-	protected async _project(event: IEvent): Promise<void> {
-		await this.#remoteProjection._project(event);
+	protected _projectBatch(batch: IEventSet): Promise<void> {
+		return this.remoteProjection._projectBatch(batch);
 	}
 
 	dispose() {

src/workers/interfaces/IWorkerProjection.ts

@@ -1,4 +1,4 @@
-import type { IEvent, IProjection } from '../../interfaces/index.js';
+import type { IEvent, IEventSet, IProjection } from '../../interfaces/index.js';
 
 export interface IWorkerProjection<TView> extends IProjection<TView> {
 
@@ -8,11 +8,11 @@ export interface IWorkerProjection<TView> extends IProjection<TView> {
 	getLastProjectedEvent(): Promise<IEvent | undefined>;
 
 	/**
-	 * Projects an event without waiting for view lock readiness.
+	 * Project restore events in batches to avoid one Comlink roundtrip per event
 	 *
-	 * @internal Expose protected projection path for worker RPC wiring
+	 * @internal
 	 */
-	_project(event: IEvent, meta?: Record<string, any>): Promise<void>;
+	_projectBatch(events: IEventSet): Promise<void>;
 }
 
 export interface IWorkerProjectionType<

tests/integration/workers/AbstractWorkerProjection.test.ts

@@ -211,6 +211,30 @@ describe('AbstractWorkerProjection', () => {
 		expect(await projectionProxy.view.getCounter()).toBe(2);
 	});
 
+	it('restores event store events through worker batches', async () => {
+
+		const restoreBatchSize = WorkerProxyProjection.RESTORE_BATCH_SIZE;
+		WorkerProxyProjection.RESTORE_BATCH_SIZE = 2;
+
+		try {
+			await projectionProxy.restore(createEventStore([
+				{ id: '1', type: 'somethingHappened' },
+				{ id: '2', type: 'somethingHappened' },
+				{ id: '3', type: 'somethingElseHappened' },
+				{ id: '4', type: 'somethingHappened' },
+				{ id: '5', type: 'somethingHappened' },
+				{ id: '6', type: 'somethingHappened' }
+			]));
+		}
+		finally {
+			WorkerProxyProjection.RESTORE_BATCH_SIZE = restoreBatchSize;
+		}
+
+		expect(await projectionProxy.view.getBatchSizes()).toEqual([2, 2, 1]);
+		expect((await projectionProxy.view.getLastEvent())?.id).toBe('6');
+		expect(await projectionProxy.view.getCounter()).toBe(5);
+	});
+
 	it('restores only events after getLastEvent', async () => {
 
 		await projectionProxy.restore(createEventStore([

tests/integration/workers/fixtures/ProjectionFixture.cjs

@@ -33,6 +33,11 @@ class ProjectionFixture extends AbstractWorkerProjection {
 		this.view.increment();
 	}
 
+	async _projectBatch(events) {
+		this.view.recordBatchSize(events.length);
+		await super._projectBatch(events);
+	}
+
 	async slowHappened() {
 		await new Promise(resolve => setTimeout(resolve, 50));
 		this.view.increment();

tests/integration/workers/fixtures/ViewFixture.cjs

@@ -6,6 +6,7 @@ class ViewFixture {
 		lock: 0,
 		unlock: 0
 	};
+	#batchSizes = [];
 	#lastEvent = null;
 	#skipIds = new Set();
 	#readyPromise = Promise.resolve();
@@ -33,6 +34,14 @@ class ViewFixture {
 		return { ...this.#calls };
 	}
 
+	recordBatchSize(size) {
+		this.#batchSizes.push(size);
+	}
+
+	getBatchSizes() {
+		return [...this.#batchSizes];
+	}
+
 	isReady() {
 		return this.ready;
 	}