diff --git a/docs/guide/eventsourcingdb/event-observer.md b/docs/guide/eventsourcingdb/event-observer.md index 897a7f4..ec19c5f 100644 --- a/docs/guide/eventsourcingdb/event-observer.md +++ b/docs/guide/eventsourcingdb/event-observer.md @@ -11,7 +11,7 @@ next: # Event Observer -The `initEventObserver` function starts a background event observation loop that continuously listens for new events from EventSourcingDB. Observers automatically reconnect with exponential backoff on failure, making them ideal for building read-side projections and reactive event handlers. +The `initEventObserver` function starts a background event observation loop that continuously listens for new events from EventSourcingDB. Observers automatically reconnect with exponential backoff on connection failure, and retry handler failures in-place without tearing down the stream — making them ideal for building read-side projections and reactive event handlers. For full details on observing events, including resuming after connection loss and observing from the last event of a given type, see the [Observing Events](https://docs.eventsourcingdb.io/getting-started/observing-events/) section in the EventSourcingDB documentation. @@ -61,14 +61,17 @@ initEventObserver({ ## EventObserver Options -| Option | Type | Default | Description | -| ----------------- | ------------------------ | ------------ | ---------------------------------------------------------- | -| `subject` | `string` | _(required)_ | The subject to observe events for | -| `recursive` | `boolean` | `false` | Whether to observe events recursively for all sub-subjects | -| `lowerBound` | `Bound` | `undefined` | The starting position for observation | -| `fromLatestEvent` | `ObserveFromLatestEvent` | `undefined` | Start observation from a specific latest event | -| `eventHandler` | `(event: Event) => void` | _(required)_ | Handler function called for each observed event | -| `retryOptions` | `RetryOptions` | see below | Options for retry behavior on connection failure | +| Option | Type | Default | Description | +| ------------------------ | -------------------------------------- | ------------ | ----------------------------------------------------------------- | +| `subject` | `string` | _(required)_ | The subject to observe events for | +| `recursive` | `boolean` | `false` | Whether to observe events recursively for all sub-subjects | +| `lowerBound` | `Bound` | `undefined` | The starting position for observation | +| `fromLatestEvent` | `ObserveFromLatestEvent` | `undefined` | Start observation from a specific latest event | +| `eventHandler` | `(event: Event) => void` | _(required)_ | Handler function called for each observed event | +| `retryOptions` | `RetryOptions` | see below | **Deprecated.** Alias for `connectionRetryOptions` | +| `connectionRetryOptions` | `RetryOptions` | see below | Retry options when the observe stream / connection fails | +| `handlerRetryOptions` | `RetryOptions` | see below | Retry options when `eventHandler` throws (in-place, no reconnect) | +| `onHandlerError` | `(error: Error, event: Event) => void` | `undefined` | Called after handler retries are exhausted; event is then skipped | ### Bound @@ -106,16 +109,57 @@ The `fromLatestEvent` option starts observation from the latest event matching s ## Retry Options +`connectionRetryOptions` and `handlerRetryOptions` share the same shape (defaults apply independently to each): + | Option | Type | Default | Description | | --------------------- | -------- | ------- | ---------------------------------------------------- | -| `maxRetries` | `number` | `3` | Maximum number of retry attempts before giving up | +| `maxRetries` | `number` | `3` | Maximum number of retries after the initial attempt | | `initialRetryDelayMs` | `number` | `3000` | Initial delay in milliseconds before the first retry | -The observer uses **exponential backoff with jitter** for retries: +Both paths use **exponential backoff with jitter**: + +- Base delay doubles with each attempt: `initialDelayMs * 2^attempt` +- Random jitter of 0-30% is added to avoid thundering-herd effects + +**Connection retries** (`connectionRetryOptions`, or deprecated `retryOptions`): + +- Used when the observe stream fails or disconnects +- After the initial attempt plus `maxRetries` reconnect attempts fail, a critical error is logged and the observer stops (`maxRetries + 1` failed attempts total) +- The connection retry counter resets when events start flowing again + +**Handler retries** (`handlerRetryOptions`): + +- Used when `eventHandler` throws; retries happen in-place without reconnecting +- After the initial attempt plus `maxRetries` retries fail for an event, `onHandlerError` is called (if provided) or a critical log is emitted (`maxRetries + 1` failed attempts total) +- The event is then **skipped** (lower bound advanced) so subsequent events keep being processed + +```typescript +import { getLogger } from "@nimbus-cqrs/core"; +import { initEventObserver } from "@nimbus-cqrs/eventsourcingdb"; -- Base delay doubles with each attempt: `initialDelayMs * 2^attempt` -- Random jitter of 0-30% is added to avoid thundering-herd effects -- After `maxRetries` consecutive failures, a critical error is logged and the observer stops +initEventObserver({ + subject: "/users", + recursive: true, + eventHandler: async (event) => { + // ... + }, + connectionRetryOptions: { + maxRetries: 5, + initialRetryDelayMs: 1000, + }, + handlerRetryOptions: { + maxRetries: 3, + initialRetryDelayMs: 500, + }, + onHandlerError: (error, event) => { + getLogger().error({ + category: "EventObserver", + message: `Skipping event after handler retries: ${event.id}`, + error, + }); + }, +}); +``` ## Building Projections @@ -189,7 +233,7 @@ await setupEventSourcingDBClient({ ## Automatic Position Tracking -The observer automatically tracks its position in the event stream. After each successfully handled event, the `lowerBound` is updated so that reconnections resume from the last processed event rather than replaying the entire stream. +The observer automatically tracks its position in the event stream. After each event is handled successfully **or skipped** after exhausted handler retries, the `lowerBound` is updated so that reconnections resume from that position rather than replaying the entire stream. ## OpenTelemetry Tracing diff --git a/examples/eventsourcing-demo/src/eventsourcingdb.ts b/examples/eventsourcing-demo/src/eventsourcingdb.ts index e72deac..72ebdef 100644 --- a/examples/eventsourcing-demo/src/eventsourcingdb.ts +++ b/examples/eventsourcing-demo/src/eventsourcingdb.ts @@ -1,5 +1,7 @@ +import { getLogger } from '@nimbus-cqrs/core'; import { setupEventSourcingDBClient } from '@nimbus-cqrs/eventsourcingdb'; import { getEnv } from '@nimbus-cqrs/utils'; +import type { Event as EventSourcingDBEvent } from 'eventsourcingdb'; import { getContactProjectionLowerBound, projectContacts, @@ -47,6 +49,52 @@ export const initEventSourcingDB = async () => { eventHandler: projectContacts, lowerBound: await getContactProjectionLowerBound() as any, }, + + // Example: handling observer handler failures. + // + // The observer retries the handler based on + // handlerRetryOptions, then calls onHandlerError. + // Decide per use case what to do when an event + // still cannot be handled (e.g. dead-letter / alert). + { + subject: '/users', + recursive: true, + handlerRetryOptions: { + maxRetries: 2, + initialRetryDelayMs: 1000, + }, + onHandlerError: (_error, event) => { + getLogger().info({ + category: 'DLQ', + message: `DLQ received event: ${event.id}`, + }); + }, + eventHandler: async ( + eventSourcingDBEvent: EventSourcingDBEvent, + ) => { + await new Promise((resolve) => + setTimeout(resolve, 1000) + ); + + getLogger().debug({ + category: 'FaultyEventHandler', + message: + `Got event: "${eventSourcingDBEvent.id}" (${eventSourcingDBEvent.type}) for subject: "${eventSourcingDBEvent.subject}"`, + }); + + if (eventSourcingDBEvent.id === '1') { + getLogger().debug({ + category: 'FaultyEventHandler', + message: + `Will demo faulty behavior by throwing an error.`, + }); + + throw new Error( + 'Faulty behavior demo: throwing an error', + ); + } + }, + }, ], }, ); diff --git a/packages/eventsourcingdb/README.md b/packages/eventsourcingdb/README.md index cab71b1..83cc8e5 100644 --- a/packages/eventsourcingdb/README.md +++ b/packages/eventsourcingdb/README.md @@ -7,9 +7,9 @@ Integration between Nimbus and [EventSourcingDB](https://www.eventsourcingdb.io/). The package wraps the official `eventsourcingdb` client with: -- a **singleton setup** that pings the server, verifies the API token and registers long-running observers in one call, -- typed **`writeEvents` / `readEvents`** helpers that translate between Nimbus events and EventSourcingDB events while preserving correlation IDs, data schemas and W3C trace context (`traceparent` / `tracestate`), -- resilient **event observers** with exponential-backoff retries, jitter, position tracking across reconnects and OpenTelemetry span linking back to the original writer. +- a **singleton setup** that pings the server, verifies the API token and registers long-running observers in one call, +- typed **`writeEvents` / `readEvents`** helpers that translate between Nimbus events and EventSourcingDB events while preserving correlation IDs, data schemas and W3C trace context (`traceparent` / `tracestate`), +- resilient **event observers** with exponential-backoff retries, jitter, position tracking across reconnects and OpenTelemetry span linking back to the original writer. Refer to the [Nimbus main repository](https://github.com/overlap-dev/Nimbus) or the [Nimbus documentation](https://nimbus.overlap.at) for more information about the Nimbus framework. @@ -122,7 +122,7 @@ await writeEvents( type: "isSubjectPristine", payload: { subject: "/todos/todo-1" }, }, - ] + ], ); ``` @@ -146,11 +146,17 @@ Pass an `AbortSignal` as the third argument if you need to cancel a long-running ## Event observers -An `EventObserver` is a long-running consumer attached to a subject. `initEventObserver` (or the `eventObservers` array on `setupEventSourcingDBClient`) starts it in the background and keeps it alive: on connection failures it retries with exponential backoff plus jitter (defaults: 3 retries, 3000 ms initial delay) and on every successful event it advances its lower bound, so a reconnection resumes from exactly where it left off — no replays, no gaps. +An `EventObserver` is a long-running consumer attached to a subject. `initEventObserver` (or the `eventObservers` array on `setupEventSourcingDBClient`) starts it in the background and keeps it alive. Connection and handler failures are retried separately with exponential backoff plus jitter (defaults: 3 retries, 3000 ms initial delay for each): + +- **Connection failures** reconnect the observe stream (`connectionRetryOptions`; deprecated alias: `retryOptions`). +- **Handler failures** are retried in-place without reconnecting (`handlerRetryOptions`). After handler retries are exhausted the event is skipped (optional `onHandlerError`) and observation continues. + +On every handled or skipped event the observer advances its lower bound so a reconnection resumes from exactly where it left off — no replays, no gaps. Each event handler runs inside an OpenTelemetry span. If the source event carries a `traceparent`, that span is linked back to the writer's trace, giving you end-to-end visibility from the command that produced the event to every subscriber that reacted to it. ```typescript +import { getLogger } from "@nimbus-cqrs/core"; import { initEventObserver } from "@nimbus-cqrs/eventsourcingdb"; initEventObserver({ @@ -166,10 +172,21 @@ initEventObserver({ // ...update a read model, send a notification, ... } }, - retryOptions: { + connectionRetryOptions: { maxRetries: 5, initialRetryDelayMs: 1000, }, + handlerRetryOptions: { + maxRetries: 3, + initialRetryDelayMs: 1000, + }, + onHandlerError: (error, event) => { + getLogger().error({ + category: "EventObserver", + message: `Skipping event after handler retries: ${event.id}`, + error, + }); + }, }); ``` @@ -201,9 +218,9 @@ import { } from "@nimbus-cqrs/eventsourcingdb"; ``` -- `nimbusEventToEventSourcingDBEventCandidate(event)` — manual conversion, e.g. when batching with the raw client. -- `eventSourcingDBEventToNimbusEvent(event)` — typed conversion when reading or observing. -- `isEventData(value)` — type guard to detect the Nimbus envelope on arbitrary stored data. +- `nimbusEventToEventSourcingDBEventCandidate(event)` — manual conversion, e.g. when batching with the raw client. +- `eventSourcingDBEventToNimbusEvent(event)` — typed conversion when reading or observing. +- `isEventData(value)` — type guard to detect the Nimbus envelope on arbitrary stored data. # License diff --git a/packages/eventsourcingdb/src/lib/client.ts b/packages/eventsourcingdb/src/lib/client.ts index 3e0886c..3247860 100644 --- a/packages/eventsourcingdb/src/lib/client.ts +++ b/packages/eventsourcingdb/src/lib/client.ts @@ -64,7 +64,11 @@ export type SetupEventSourcingDBClientInput = { * eventHandler: async (event: Event) => { * console.log('Received event:', event); * }, - * retryOptions: { + * connectionRetryOptions: { + * maxRetries: 3, + * initialRetryDelayMs: 3000, + * }, + * handlerRetryOptions: { * maxRetries: 3, * initialRetryDelayMs: 3000, * }, diff --git a/packages/eventsourcingdb/src/lib/eventObserver.ts b/packages/eventsourcingdb/src/lib/eventObserver.ts index 0028949..a4a5849 100644 --- a/packages/eventsourcingdb/src/lib/eventObserver.ts +++ b/packages/eventsourcingdb/src/lib/eventObserver.ts @@ -16,8 +16,8 @@ type ObserveFromLatestEvent = { export type RetryOptions = { /** - * The maximum number of retry attempts before giving up. - * Defaults to 3. + * The maximum number of retries after the initial attempt + * before giving up. Defaults to 3 (4 failed attempts total). */ maxRetries: number; /** @@ -28,6 +28,11 @@ export type RetryOptions = { initialRetryDelayMs: number; }; +const DEFAULT_RETRY_OPTIONS: RetryOptions = { + maxRetries: 3, + initialRetryDelayMs: 3000, +}; + /** * An event observer defines a handler function which will be applied to each event * and the options to observe the events according to the EventSourcingDB API. @@ -62,11 +67,36 @@ export type EventObserver = { */ eventHandler: (event: EventSourcingDBEvent) => Promise | void; /** - * Options for retry behavior when the connection fails. - * Uses exponential backoff with jitter between retries. + * Options for retry behavior when the observe stream / connection fails. + * Uses exponential backoff with jitter between reconnects. + * Defaults to { maxRetries: 3, initialRetryDelayMs: 3000 }. + */ + connectionRetryOptions?: RetryOptions; + /** + * Options for retry behavior when {@link eventHandler} throws. + * Handler retries happen in-place without reconnecting the stream. * Defaults to { maxRetries: 3, initialRetryDelayMs: 3000 }. */ + handlerRetryOptions?: RetryOptions; + /** + * @deprecated Use {@link connectionRetryOptions} instead. + * + * Options for retry behavior when the observe stream / connection fails. + * Ignored when {@link connectionRetryOptions} is set. + */ retryOptions?: RetryOptions; + /** + * Called when {@link eventHandler} keeps failing after all handler + * retries are exhausted. The event is then skipped and observation + * continues. When omitted, a critical log entry is emitted instead. + * + * @param error - The last error thrown by the handler. + * @param event - The event that could not be handled. + */ + onHandlerError?: ( + error: Error, + event: EventSourcingDBEvent, + ) => Promise | void; }; /** @@ -97,6 +127,36 @@ export const calculateBackoffDelay = ( return Math.floor(baseDelay + jitter); }; +/** + * Resolves connection retry options, preferring + * {@link EventObserver.connectionRetryOptions} over the deprecated + * {@link EventObserver.retryOptions}. + */ +const resolveConnectionRetryOptions = ( + eventObserver: EventObserver, +): RetryOptions => ({ + maxRetries: eventObserver.connectionRetryOptions?.maxRetries ?? + eventObserver.retryOptions?.maxRetries ?? + DEFAULT_RETRY_OPTIONS.maxRetries, + initialRetryDelayMs: + eventObserver.connectionRetryOptions?.initialRetryDelayMs ?? + eventObserver.retryOptions?.initialRetryDelayMs ?? + DEFAULT_RETRY_OPTIONS.initialRetryDelayMs, +}); + +/** + * Resolves handler retry options from {@link EventObserver.handlerRetryOptions}. + */ +const resolveHandlerRetryOptions = ( + eventObserver: EventObserver, +): RetryOptions => ({ + maxRetries: eventObserver.handlerRetryOptions?.maxRetries ?? + DEFAULT_RETRY_OPTIONS.maxRetries, + initialRetryDelayMs: + eventObserver.handlerRetryOptions?.initialRetryDelayMs ?? + DEFAULT_RETRY_OPTIONS.initialRetryDelayMs, +}); + /** * Logs an informational message when an event observer connects or * reconnects to EventSourcingDB. When {@link retryCount} is greater @@ -121,10 +181,10 @@ const logObserverConnection = ( }; /** - * Handles an observer error by logging it and waiting with exponential - * backoff before the next retry attempt. When the maximum number of - * retries is exceeded a critical log entry is emitted and no further - * retries are attempted. + * Handles a connection / stream error by logging it and waiting with + * exponential backoff before the next reconnect attempt. When the + * maximum number of retries is exceeded a critical log entry is + * emitted and no further retries are attempted. * * @param error - The error that caused the observer to disconnect. * @param subject - The observed subject. @@ -135,7 +195,7 @@ const logObserverConnection = ( * @returns `true` if the observer should retry, `false` if retries * are exhausted. */ -const handleObserverError = async ( +const handleConnectionError = async ( error: unknown, subject: string, retryCount: number, @@ -169,38 +229,119 @@ const handleObserverError = async ( return true; }; +/** + * Invokes the event handler with in-place retries. Does not reconnect + * the observe stream. When all retries are exhausted, calls + * {@link EventObserver.onHandlerError} or logs critically, then + * returns without throwing so the observer can skip the event. + */ +const handleEventWithRetry = async ( + eventObserver: EventObserver, + event: EventSourcingDBEvent, + handlerRetryOptions: RetryOptions, +): Promise => { + const { maxRetries, initialRetryDelayMs } = handlerRetryOptions; + let attempt = 0; + + while (attempt <= maxRetries) { + try { + await eventObserver.eventHandler(event); + return; + } catch (error: unknown) { + attempt++; + + const handlerError = error instanceof Error + ? error + : new Error(String(error)); + + if (attempt > maxRetries) { + if (eventObserver.onHandlerError) { + try { + await eventObserver.onHandlerError( + handlerError, + event, + ); + } catch (err: unknown) { + // Isolate user callback failures so the poison + // event is still skipped and observation continues. + const callbackError = err instanceof Error + ? err + : new Error(String(err)); + + getLogger().critical({ + category: 'Nimbus', + message: + `onHandlerError failed for event "${event.id}" (${event.type}) for subject "${eventObserver.subject}". Skipping event.`, + error: callbackError, + }); + } + } else { + getLogger().critical({ + category: 'Nimbus', + message: + `Failed to handle event "${event.id}" (${event.type}) for subject "${eventObserver.subject}" after ${maxRetries} ${ + maxRetries === 1 ? 'retry' : 'retries' + }. Skipping event.`, + error: handlerError, + }); + } + return; + } + + const backoffDelay = calculateBackoffDelay( + initialRetryDelayMs, + attempt - 1, + ); + + getLogger().error({ + category: 'Nimbus', + message: + `Error handling event "${event.id}" (${event.type}) for subject "${eventObserver.subject}" (retry ${attempt}/${maxRetries}), retrying in ${backoffDelay}ms`, + error: handlerError, + }); + + await delay(backoffDelay); + } + } +}; + /** * Starts observing events for the given {@link EventObserver} with - * automatic reconnection on failure. + * automatic reconnection on stream failure and separate in-place + * retries for handler failures. * - * On each connection attempt the EventSourcingDB server is pinged - * first. Events are then consumed from the stream and each one is - * passed to the observer's event handler inside an OpenTelemetry - * span. If the event carries a `traceparent`, the span is linked to - * the original writer's trace for end-to-end distributed tracing. + * Events are consumed from the stream and each one is passed to the + * observer's event handler inside an OpenTelemetry span. If the event + * carries a `traceparent`, the span is linked to the original writer's + * trace for end-to-end distributed tracing. * - * After every successfully handled event the lower bound is advanced - * so that a reconnection resumes from the last processed position. - * - * When the connection drops, exponential backoff with jitter is - * applied up to the configured maximum number of retries. + * Handler failures are retried without reconnecting. After handler + * retries are exhausted the event is skipped (lower bound advanced) + * so observation can continue. When the connection drops, exponential + * backoff with jitter is applied up to the configured maximum number + * of connection retries. * * @param eventObserver - The event observer configuration. + * @returns A promise that resolves when the observe stream ends or + * connection retries are exhausted. */ -const observeWithRetry = async ( +export const observeWithRetry = async ( eventObserver: EventObserver, ): Promise => { - const eventSourcingDBClient = getEventSourcingDBClient(); - - const maxRetries = eventObserver.retryOptions?.maxRetries ?? 3; - const initialRetryDelayMs = - eventObserver.retryOptions?.initialRetryDelayMs ?? 3000; + const connectionRetryOptions = resolveConnectionRetryOptions( + eventObserver, + ); + const handlerRetryOptions = resolveHandlerRetryOptions(eventObserver); - let retryCount = 0; + let connectionRetryCount = 0; let lastProcessedEventId: string | undefined; while (true) { try { + // Resolve the client on every attempt so a re-initialized + // singleton is picked up after reconnecting to EventSourcingDB. + const eventSourcingDBClient = getEventSourcingDBClient(); + // Once we have a concrete position, use it as lower bound and // drop fromLatestEvent; otherwise fall back to the original options. const lowerBound: Bound | undefined = lastProcessedEventId @@ -211,7 +352,7 @@ const observeWithRetry = async ( ? undefined : eventObserver.fromLatestEvent; - logObserverConnection(eventObserver.subject, retryCount, { + logObserverConnection(eventObserver.subject, connectionRetryCount, { recursive: eventObserver.recursive ?? false, lowerBound, fromLatestEvent, @@ -227,8 +368,8 @@ const observeWithRetry = async ( }, ) ) { - // Reset the retry count as soon as we successfully receive an event - retryCount = 0; + // Stream is healthy once events flow again. + connectionRetryCount = 0; const traceContext: TraceContext | undefined = event.traceparent ? { @@ -240,26 +381,31 @@ const observeWithRetry = async ( await withSpan( 'observeEvent', async () => { - await eventObserver.eventHandler(event); + await handleEventWithRetry( + eventObserver, + event, + handlerRetryOptions, + ); }, traceContext, ); - // Track last processed position so retries resume from here + // Advance after success or skip so poison events do not block + // the stream and reconnects resume past the last attempted event. lastProcessedEventId = event.id; } // If the loop completes normally (stream ended), we're done return; } catch (error) { - retryCount++; + connectionRetryCount++; - const shouldRetry = await handleObserverError( + const shouldRetry = await handleConnectionError( error, eventObserver.subject, - retryCount, - maxRetries, - initialRetryDelayMs, + connectionRetryCount, + connectionRetryOptions.maxRetries, + connectionRetryOptions.initialRetryDelayMs, ); if (!shouldRetry) { @@ -272,8 +418,9 @@ const observeWithRetry = async ( /** * Initializes an event observer by starting the observation loop in * the background (non-blocking). The observer will keep running and - * reconnecting according to its retry options until the stream ends - * or retries are exhausted. + * reconnecting according to its connection retry options until the + * stream ends or connection retries are exhausted. Handler failures + * are retried separately and do not stop the observer. * * @param eventObserver - The event observer configuration. */ diff --git a/packages/eventsourcingdb/src/lib/integration.test.ts b/packages/eventsourcingdb/src/lib/integration.test.ts index 6976878..7450eef 100644 --- a/packages/eventsourcingdb/src/lib/integration.test.ts +++ b/packages/eventsourcingdb/src/lib/integration.test.ts @@ -1,12 +1,37 @@ import { createEvent, type Event } from '@nimbus-cqrs/core'; -import { assertEquals } from '@std/assert'; +import { assert, assertEquals } from '@std/assert'; import { Container, type Event as EventSourcingDBEvent } from 'eventsourcingdb'; import { setupEventSourcingDBClient } from './client.ts'; import { eventSourcingDBEventToNimbusEvent } from './eventMapping.ts'; -import { initEventObserver } from './eventObserver.ts'; +import { initEventObserver, observeWithRetry } from './eventObserver.ts'; import { readEvents } from './readEvents.ts'; import { writeEvents } from './writeEvents.ts'; +const withTimeout = ( + promise: Promise, + ms: number, + message: string, +): Promise => { + let timer: ReturnType | undefined; + + return new Promise((resolve, reject) => { + timer = setTimeout(() => { + reject(new Error(message)); + }, ms); + + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); +}; + // --------------------------------------------------------------------------- // Shared container lifecycle // --------------------------------------------------------------------------- @@ -200,28 +225,17 @@ Deno.test({ resolveReceived(); } }, - retryOptions: { + connectionRetryOptions: { maxRetries: 1, initialRetryDelayMs: 100, }, }); - // Wait for the handler to collect both events - // (with a safety timeout) - await Promise.race([ + await withTimeout( allReceived, - new Promise((_, reject) => - setTimeout( - () => - reject( - new Error( - 'Observer did not receive events within 5 s', - ), - ), - 5000, - ) - ), - ]); + 5000, + 'Observer did not receive events within 5 s', + ); assertEquals(received.length, 2); assertEquals( @@ -234,8 +248,289 @@ Deno.test({ ); }, ); + + await t.step( + 'initEventObserver retries handler failures in-place', + async () => { + await writeEvents([ + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.handler-retry', + subject: '/observer/handler-retry', + data: { index: 0 }, + }), + ]); + + let handlerCalls = 0; + let resolveDone: () => void; + const done = new Promise((resolve) => { + resolveDone = resolve; + }); + + initEventObserver({ + subject: '/observer/handler-retry', + recursive: false, + eventHandler: () => { + handlerCalls++; + if (handlerCalls < 3) { + throw new Error('transient handler failure'); + } + resolveDone(); + }, + handlerRetryOptions: { + maxRetries: 3, + initialRetryDelayMs: 10, + }, + }); + + await withTimeout( + done, + 5000, + 'Handler did not succeed within 5 s', + ); + + assertEquals(handlerCalls, 3); + }, + ); + + await t.step( + 'initEventObserver skips poison events and continues observing', + async () => { + await writeEvents([ + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.poison', + subject: '/observer/poison', + data: { index: 0 }, + }), + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.after-poison', + subject: '/observer/poison', + data: { index: 1 }, + }), + ]); + + const handledTypes: string[] = []; + const onHandlerErrorTypes: string[] = []; + let resolveDone: () => void; + const done = new Promise((resolve) => { + resolveDone = resolve; + }); + + initEventObserver({ + subject: '/observer/poison', + recursive: false, + eventHandler: (event) => { + if (event.type === 'at.test.nimbus.poison') { + throw new Error('poison event'); + } + handledTypes.push(event.type); + resolveDone(); + }, + handlerRetryOptions: { + maxRetries: 2, + initialRetryDelayMs: 10, + }, + onHandlerError: (_error, event) => { + onHandlerErrorTypes.push(event.type); + }, + }); + + await withTimeout( + done, + 5000, + 'Observer did not continue past poison event within 5 s', + ); + + assertEquals(onHandlerErrorTypes, [ + 'at.test.nimbus.poison', + ]); + assertEquals(handledTypes, [ + 'at.test.nimbus.after-poison', + ]); + }, + ); + + await t.step( + 'initEventObserver skips event when onHandlerError throws', + async () => { + await writeEvents([ + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.on-error-throw', + subject: '/observer/on-handler-error-throw', + data: { index: 0 }, + }), + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.after-on-error-throw', + subject: '/observer/on-handler-error-throw', + data: { index: 1 }, + }), + ]); + + const handledTypes: string[] = []; + let onHandlerErrorCalls = 0; + let resolveDone: () => void; + const done = new Promise((resolve) => { + resolveDone = resolve; + }); + + initEventObserver({ + subject: '/observer/on-handler-error-throw', + recursive: false, + eventHandler: (event) => { + if ( + event.type === 'at.test.nimbus.on-error-throw' + ) { + throw new Error('poison event'); + } + handledTypes.push(event.type); + resolveDone(); + }, + handlerRetryOptions: { + maxRetries: 1, + initialRetryDelayMs: 10, + }, + onHandlerError: () => { + onHandlerErrorCalls++; + throw new Error('onHandlerError failed'); + }, + }); + + await withTimeout( + done, + 5000, + 'Observer did not continue after onHandlerError threw within 5 s', + ); + + assertEquals(onHandlerErrorCalls, 1); + assertEquals(handledTypes, [ + 'at.test.nimbus.after-on-error-throw', + ]); + }, + ); + + // ----------------------------------------------------------------- + // Connection retries (stop/restart the real EventSourcingDB) + // ----------------------------------------------------------------- + + await t.step( + 'observeWithRetry reconnects after EventSourcingDB comes back', + async () => { + // Stop first so the observer begins in a failing + // connection state, then bring EventSourcingDB back. + await container.stop(); + + const received: EventSourcingDBEvent[] = []; + let resolveReceived: () => void; + const eventReceived = new Promise((resolve) => { + resolveReceived = resolve; + }); + + initEventObserver({ + subject: '/observer/reconnect', + recursive: false, + eventHandler: (event) => { + received.push(event); + resolveReceived(); + }, + connectionRetryOptions: { + maxRetries: 20, + initialRetryDelayMs: 100, + }, + }); + + // Let the observer attempt (and fail) at least once. + await new Promise((resolve) => setTimeout(resolve, 250)); + + await container.start(); + await setupEventSourcingDBClient({ + url: container.getBaseUrl(), + apiToken: container.getApiToken(), + }); + + await writeEvents([ + createEvent({ + source: 'https://nimbus.test', + type: 'at.test.nimbus.reconnect', + subject: '/observer/reconnect', + data: { index: 0 }, + }), + ]); + + await withTimeout( + eventReceived, + 15000, + 'Observer did not reconnect and receive an event within 15 s', + ); + + assertEquals(received.length, 1); + assertEquals( + received[0].type, + 'at.test.nimbus.reconnect', + ); + }, + ); + + await t.step( + 'observeWithRetry stops after connection retries are exhausted', + async () => { + await container.stop(); + + const startedAt = Date.now(); + + await withTimeout( + observeWithRetry({ + subject: '/observer/connection-exhausted', + recursive: false, + eventHandler: () => {}, + connectionRetryOptions: { + maxRetries: 2, + initialRetryDelayMs: 50, + }, + }), + 10000, + 'Observer did not stop after exhausting connection retries', + ); + + const elapsedMs = Date.now() - startedAt; + + // Initial attempt + 2 retries with backoff; should finish + // quickly and must not hang until the timeout. + assert( + elapsedMs < 10000, + `Expected exhaustion within 10 s, took ${elapsedMs}ms`, + ); + }, + ); + + await t.step( + 'observeWithRetry uses deprecated retryOptions for connection retries', + async () => { + // Container is already stopped from the previous step. + assert(!container.isRunning()); + + await withTimeout( + observeWithRetry({ + subject: '/observer/deprecated-retry-options', + recursive: false, + eventHandler: () => {}, + retryOptions: { + maxRetries: 1, + initialRetryDelayMs: 50, + }, + }), + 10000, + 'Observer did not stop when using deprecated retryOptions', + ); + }, + ); } finally { - await container.stop(); + if (container.isRunning()) { + await container.stop(); + } } }, });