fix(eventsourcingdb): separate handler retries from observer reconnects

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
 

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',
+                            );
+                        }
+                    },
+                },
             ],
         },
     );

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<TEvent>(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<TEvent>(event)` — typed conversion when reading or observing.
+- `isEventData(value)` — type guard to detect the Nimbus envelope on arbitrary stored data.
 
 # License
 

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,
  *             },