set max queued events to prevent memory leaks

Author: dillonstreator

examples/pg/processor.ts

@@ -41,7 +41,7 @@ let processor: EventProcessor<EventType> | undefined = undefined;
           return;
         },
         thing3: async (event) => {
-          await sleep(20_000);
+          await sleep(Math.random() * 10_000);
           console.log(`${event.id} thing3 ${event.correlation_id}`);
           if (Math.random() > 0.75) throw new Error("some issue");
 

src/pg/client.ts

@@ -24,7 +24,7 @@ export const createProcessorClient = <EventType extends string>(
     const events = await querier.query<
       Pick<TxOBEvent<EventType>, "id" | "errors">
     >(
-      `SELECT id, errors FROM ${escapeIdentifier(table)} WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 LIMIT ${limit}`,
+      `SELECT id, errors FROM ${escapeIdentifier(table)} WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 ORDER BY timestamp ASC LIMIT ${limit}`,
       [opts.maxErrors],
     );
     return events.rows;

src/processor.ts

@@ -77,6 +77,7 @@ const defaultPollingIntervalMs = 5000;
 const defaultMaxErrors = 5;
 const defaultMaxEventConcurrency = 20;
 const defaultMaxHandlerConcurrency = 10;
+const defaultMaxQueuedEvents = 100;
 
 type TxOBProcessEventsOpts<TxOBEventType extends string> = {
   maxErrors: number;
@@ -85,6 +86,7 @@ type TxOBProcessEventsOpts<TxOBEventType extends string> = {
   logger?: Logger;
   maxEventConcurrency?: number;
   maxHandlerConcurrency?: number;
+  maxQueuedEvents?: number;
   onEventMaxErrorsReached?: (opts: {
     event: Readonly<TxOBEvent<TxOBEventType>>;
     txClient: TxOBTransactionProcessorClient<TxOBEventType>;
@@ -327,17 +329,6 @@ const processEvent = async <TxOBEventType extends string>({
       lockedEvent.processed_at = getDate();
     }
 
-    logger?.debug(
-      {
-        eventId: lockedEvent.id,
-        type: lockedEvent.type,
-        lockedEvent,
-        correlationId: lockedEvent.correlation_id,
-        errored,
-      },
-      "updating event",
-    );
-
     await txClient.updateEvent(lockedEvent);
   });
 };
@@ -369,6 +360,7 @@ export class EventProcessor<TxOBEventType extends string> {
   private handlerMap: TxOBEventHandlerMap<TxOBEventType>;
   private opts: Omit<TxOBProcessEventsOpts<TxOBEventType>, "signal"> & {
     pollingIntervalMs: number;
+    maxQueuedEvents: number;
   };
   private abortController: AbortController;
   private queue: PQueue;
@@ -390,6 +382,7 @@ export class EventProcessor<TxOBEventType extends string> {
       backoff: defaultBackoff,
       maxEventConcurrency: defaultMaxEventConcurrency,
       maxHandlerConcurrency: defaultMaxHandlerConcurrency,
+      maxQueuedEvents: defaultMaxQueuedEvents,
       ...opts,
     };
     this.client = client;
@@ -415,6 +408,19 @@ export class EventProcessor<TxOBEventType extends string> {
       try {
         do {
           try {
+            // Skip polling if we're at capacity to prevent memory leaks
+            if (queuedEventIds.size >= this.opts.maxQueuedEvents) {
+              this.opts.logger?.debug(
+                {
+                  queuedCount: queuedEventIds.size,
+                  maxQueuedEvents: this.opts.maxQueuedEvents,
+                },
+                "skipping poll - queue at capacity",
+              );
+              await sleep(this.opts.pollingIntervalMs);
+              continue;
+            }
+
             const events = await this.client.getEventsToProcess({
               ...this.opts,
               signal: this.abortController.signal,