custom txob error to specify backoff from within handler

Author: dillonstreator

package.json

@@ -25,13 +25,13 @@
     "dist",
     "README.md"
   ],
-  "main": "./dist/cjs/processor.js",
-  "types": "dist/processor.d.ts",
+  "main": "./dist/cjs/index.js",
+  "types": "dist/index.d.ts",
   "exports": {
     ".": {
-      "types": "./dist/processor.d.ts",
-      "import": "./dist/processor.js",
-      "require": "./dist/cjs/processor.js"
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js",
+      "require": "./dist/cjs/index.js"
     },
     "./pg": {
       "types": "./dist/pg/client.d.ts",

src/error.ts

@@ -0,0 +1,53 @@
+/**
+ * TxobError can be thrown by an event handler to indicate that the event processing should be retried.
+ * It allows handlers to specify a custom backoff time via the `backoffUntil` property.
+ *
+ * When multiple handlers throw TxobError with different `backoffUntil` dates, the processor will use
+ * the latest (maximum) backoff time among all handlers and the default backoff calculation.
+ *
+ * @example
+ * ```typescript
+ * // Throw an error with a custom backoff time
+ * throw new TxobError("Rate limit exceeded", {
+ *   backoffUntil: new Date(Date.now() + 60000) // Retry after 1 minute
+ * });
+ *
+ * // Throw an error with a cause
+ * throw new TxobError("Processing failed", {
+ *   cause: originalError,
+ *   backoffUntil: new Date(Date.now() + 30000) // Retry after 30 seconds
+ * });
+ * ```
+ */
+export class TxobError extends Error {
+  /**
+   * Optional date indicating when the event should be retried.
+   * If provided, this backoff time will be considered along with the default backoff calculation,
+   * and the latest (maximum) backoff time will be used.
+   */
+  backoffUntil?: Date;
+
+  constructor(
+    message: string,
+    options: { cause?: Error; backoffUntil?: Date } = {},
+  ) {
+    super(message);
+    this.cause = options.cause;
+    this.backoffUntil = options.backoffUntil;
+  }
+}
+
+/**
+ * ErrorUnprocessableEventHandler can be thrown by an event handler to indicate that the event handler is unprocessable.
+ * It wraps the original error that caused the handler to be unprocessable.
+ * This error will signal the processor to stop processing the event handler and mark the event handler as unprocessable.
+ */
+export class ErrorUnprocessableEventHandler extends Error {
+  error: Error;
+
+  constructor(error: Error) {
+    const message = `unprocessable event handler: ${error.message}`;
+    super(message);
+    this.error = error;
+  }
+}

src/index.ts

@@ -0,0 +1,2 @@
+export * from "./processor.js";
+export * from "./error.js";

src/mongodb/client.ts

@@ -37,7 +37,7 @@ export const createProcessorClient = <EventType extends string>(
       .find(filter)
       .project({ id: 1, errors: 1 })
       .limit(limit)
-      .sort('timestamp', 'asc')
+      .sort("timestamp", "asc")
       .toArray()) as Pick<TxOBEvent<EventType>, "id" | "errors">[];
 
     return events;

src/processor.test.ts

@@ -1,10 +1,6 @@
 import { describe, it, expect, vi, afterEach } from "vitest";
-import {
-  EventProcessor,
-  TxOBEvent,
-  ErrorUnprocessableEventHandler,
-  defaultBackoff,
-} from "./processor.js";
+import { EventProcessor, TxOBEvent, defaultBackoff } from "./processor.js";
+import { TxobError, ErrorUnprocessableEventHandler } from "./error.js";
 import { sleep } from "./sleep.js";
 
 const mockTxClient = {
@@ -763,6 +759,206 @@ describe("EventProcessor - processEvents", () => {
       "error processing event",
     );
   });
+
+  it("should use the latest backoff when multiple TxobErrors have different backoffUntil dates", async () => {
+    const opts = {
+      maxErrors: 5,
+      backoff: vi.fn(() => new Date(now.getTime() + 5000)), // Default backoff: 5 seconds
+      pollingIntervalMs: 10,
+    };
+
+    // Create different backoff times
+    const backoff1 = new Date(now.getTime() + 10000); // 10 seconds
+    const backoff2 = new Date(now.getTime() + 20000); // 20 seconds (latest)
+    const backoff3 = new Date(now.getTime() + 15000); // 15 seconds
+
+    const error1 = new TxobError("error 1", { backoffUntil: backoff1 });
+    const error2 = new TxobError("error 2", { backoffUntil: backoff2 });
+    const error3 = new TxobError("error 3", { backoffUntil: backoff3 });
+
+    const handlerMap = {
+      evtType1: {
+        handler1: vi.fn(() => Promise.reject(error1)),
+        handler2: vi.fn(() => Promise.reject(error2)),
+        handler3: vi.fn(() => Promise.reject(error3)),
+      },
+    };
+
+    const evt1: TxOBEvent<keyof typeof handlerMap> = {
+      type: "evtType1",
+      id: "1",
+      timestamp: now,
+      data: {},
+      correlation_id: "abc123",
+      handler_results: {},
+      errors: 0,
+    };
+
+    const events = [evt1];
+    let callCount = 0;
+    mockClient.getEventsToProcess.mockImplementation(() => {
+      callCount++;
+      return Promise.resolve(callCount === 1 ? events : []);
+    });
+    mockTxClient.getEventByIdForUpdateSkipLocked.mockImplementation((id) => {
+      return Promise.resolve(events.find((e) => e.id === id) ?? null);
+    });
+    mockTxClient.updateEvent.mockImplementation(() => {
+      return Promise.resolve();
+    });
+
+    const processor = new EventProcessor({
+      client: mockClient,
+      handlerMap,
+      ...opts,
+    });
+    processor.start();
+    await sleep(50); // Wait for processing
+    await processor.stop();
+
+    expect(mockClient.getEventsToProcess).toHaveBeenCalled();
+    expect(mockClient.transaction).toHaveBeenCalledTimes(1);
+
+    // All handlers should have been called
+    expect(handlerMap.evtType1.handler1).toHaveBeenCalledOnce();
+    expect(handlerMap.evtType1.handler2).toHaveBeenCalledOnce();
+    expect(handlerMap.evtType1.handler3).toHaveBeenCalledOnce();
+
+    // Default backoff should also be called
+    expect(opts.backoff).toHaveBeenCalledWith(1);
+
+    // The latest backoff (backoff2 = 20 seconds) should be used
+    expect(mockTxClient.updateEvent).toHaveBeenCalledTimes(1);
+    const updateCall = mockTxClient.updateEvent.mock.calls[0][0];
+    expect(updateCall.backoff_until).toEqual(backoff2);
+    expect(updateCall.errors).toBe(1);
+  });
+
+  it("should use the latest backoff when TxobError backoff is later than default backoff", async () => {
+    const laterBackoff = new Date(now.getTime() + 30000); // 30 seconds
+    const defaultBackoffTime = new Date(now.getTime() + 5000); // 5 seconds
+
+    const opts = {
+      maxErrors: 5,
+      backoff: vi.fn(() => defaultBackoffTime),
+      pollingIntervalMs: 10,
+    };
+
+    const error = new TxobError("error with backoff", {
+      backoffUntil: laterBackoff,
+    });
+
+    const handlerMap = {
+      evtType1: {
+        handler1: vi.fn(() => Promise.reject(error)),
+      },
+    };
+
+    const evt1: TxOBEvent<keyof typeof handlerMap> = {
+      type: "evtType1",
+      id: "1",
+      timestamp: now,
+      data: {},
+      correlation_id: "abc123",
+      handler_results: {},
+      errors: 0,
+    };
+
+    const events = [evt1];
+    let callCount = 0;
+    mockClient.getEventsToProcess.mockImplementation(() => {
+      callCount++;
+      return Promise.resolve(callCount === 1 ? events : []);
+    });
+    mockTxClient.getEventByIdForUpdateSkipLocked.mockImplementation((id) => {
+      return Promise.resolve(events.find((e) => e.id === id) ?? null);
+    });
+    mockTxClient.updateEvent.mockImplementation(() => {
+      return Promise.resolve();
+    });
+
+    const processor = new EventProcessor({
+      client: mockClient,
+      handlerMap,
+      ...opts,
+    });
+    processor.start();
+    await sleep(50); // Wait for processing
+    await processor.stop();
+
+    expect(mockClient.transaction).toHaveBeenCalledTimes(1);
+    expect(opts.backoff).toHaveBeenCalledWith(1);
+
+    // The latest backoff (laterBackoff = 30 seconds) should be used, not the default (5 seconds)
+    const updateCall = mockTxClient.updateEvent.mock.calls[0][0];
+    expect(updateCall.backoff_until).toEqual(laterBackoff);
+    expect(updateCall.backoff_until?.getTime()).toBeGreaterThan(
+      defaultBackoffTime.getTime(),
+    );
+  });
+
+  it("should use default backoff when TxobError backoff is earlier than default backoff", async () => {
+    const earlierBackoff = new Date(now.getTime() + 2000); // 2 seconds
+    const defaultBackoffTime = new Date(now.getTime() + 5000); // 5 seconds
+
+    const opts = {
+      maxErrors: 5,
+      backoff: vi.fn(() => defaultBackoffTime),
+      pollingIntervalMs: 10,
+    };
+
+    const error = new TxobError("error with backoff", {
+      backoffUntil: earlierBackoff,
+    });
+
+    const handlerMap = {
+      evtType1: {
+        handler1: vi.fn(() => Promise.reject(error)),
+      },
+    };
+
+    const evt1: TxOBEvent<keyof typeof handlerMap> = {
+      type: "evtType1",
+      id: "1",
+      timestamp: now,
+      data: {},
+      correlation_id: "abc123",
+      handler_results: {},
+      errors: 0,
+    };
+
+    const events = [evt1];
+    let callCount = 0;
+    mockClient.getEventsToProcess.mockImplementation(() => {
+      callCount++;
+      return Promise.resolve(callCount === 1 ? events : []);
+    });
+    mockTxClient.getEventByIdForUpdateSkipLocked.mockImplementation((id) => {
+      return Promise.resolve(events.find((e) => e.id === id) ?? null);
+    });
+    mockTxClient.updateEvent.mockImplementation(() => {
+      return Promise.resolve();
+    });
+
+    const processor = new EventProcessor({
+      client: mockClient,
+      handlerMap,
+      ...opts,
+    });
+    processor.start();
+    await sleep(50); // Wait for processing
+    await processor.stop();
+
+    expect(mockClient.transaction).toHaveBeenCalledTimes(1);
+    expect(opts.backoff).toHaveBeenCalledWith(1);
+
+    // The latest backoff (defaultBackoffTime = 5 seconds) should be used, not the earlier one (2 seconds)
+    const updateCall = mockTxClient.updateEvent.mock.calls[0][0];
+    expect(updateCall.backoff_until).toEqual(defaultBackoffTime);
+    expect(updateCall.backoff_until?.getTime()).toBeGreaterThan(
+      earlierBackoff.getTime(),
+    );
+  });
 });
 
 describe("defaultBackoff", () => {

src/processor.ts

@@ -3,6 +3,7 @@ import { sleep } from "./sleep.js";
 import pLimit from "p-limit";
 import { deepClone } from "./clone.js";
 import PQueue from "p-queue";
+import { ErrorUnprocessableEventHandler, TxobError } from "./error.js";
 
 type TxOBEventHandlerResult = {
   processed_at?: Date;
@@ -199,6 +200,8 @@ const processEvent = async <TxOBEventType extends string>({
       `processing event`,
     );
 
+    let backoffs: Date[] = [];
+
     const handlerLimit = pLimit(maxHandlerConcurrency);
     await Promise.allSettled(
       Object.entries(eventHandlerMap).map(([handlerName, handler]) =>
@@ -263,6 +266,10 @@ const processEvent = async <TxOBEventType extends string>({
               });
               errored = true;
             } else {
+              if (error instanceof TxobError && error.backoffUntil) {
+                backoffs.push(error.backoffUntil);
+              }
+
               errored = true;
               handlerResults.errors?.push({
                 error: (error as Error)?.message ?? error,
@@ -299,7 +306,11 @@ const processEvent = async <TxOBEventType extends string>({
 
     if (errored) {
       lockedEvent.errors = Math.min(lockedEvent.errors + 1, maxErrors);
-      lockedEvent.backoff_until = backoff(lockedEvent.errors);
+      backoffs.push(backoff(lockedEvent.errors));
+      const latestBackoff = backoffs.sort(
+        (a, b) => b.getTime() - a.getTime(),
+      )[0];
+      lockedEvent.backoff_until = latestBackoff;
       if (lockedEvent.errors === maxErrors) {
         lockedEvent.backoff_until = null;
         lockedEvent.processed_at = getDate();
@@ -333,21 +344,6 @@ const processEvent = async <TxOBEventType extends string>({
   });
 };
 
-/**
- * ErrorUnprocessableEventHandler can be thrown by an event handler to indicate that the event handler is unprocessable.
- * It wraps the original error that caused the handler to be unprocessable.
- * This error will signal the processor to stop processing the event handler and mark the event handler as unprocessable.
- */
-export class ErrorUnprocessableEventHandler extends Error {
-  error: Error;
-
-  constructor(error: Error) {
-    const message = `unprocessable event handler: ${error.message}`;
-    super(message);
-    this.error = error;
-  }
-}
-
 export interface Logger {
   debug(message?: unknown, ...optionalParams: unknown[]): void;
   info(message?: unknown, ...optionalParams: unknown[]): void;