Fixed cleanup and stopping processing right after error was caught

Author: oskardudycz

src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.handling.int.spec.ts

@@ -3,7 +3,6 @@ import {
   asyncAwaiter,
   delay,
   getInMemoryDatabase,
-  JSONParser,
   type Event,
   type InMemoryReactorOptions,
   type RecordedMessage,
@@ -24,7 +23,7 @@ import {
   type EventStoreDBEventStoreConsumerType,
 } from './eventStoreDBEventStoreConsumer';
 
-const withDeadline = { timeout: 10000 };
+const withDeadline = { timeout: 1000000 };
 
 void describe('EventStoreDB event store started consumer', () => {
   let eventStoreDB: StartedEventStoreDBContainer;
@@ -191,7 +190,7 @@ void describe('EventStoreDB event store started consumer', () => {
           event.metadata.globalPosition ===
           appendResult.lastEventGlobalPosition,
         eachMessage: async (event) => {
-          await delay(Math.floor(Math.random() * 150));
+          await delay(Math.floor(Math.random() * 200));
 
           result.push(event);
         },
@@ -208,6 +207,69 @@ void describe('EventStoreDB event store started consumer', () => {
       }
     });
 
+    void it(
+      `stops processing on unhandled error in handler`,
+      { timeout: 1500000 },
+      async () => {
+        // Given
+        const guestId = uuid();
+        const otherGuestId = uuid();
+        const streamName = `guestStay-${otherGuestId}`;
+        const otherStreamName = `guestStay-${guestId}`;
+        const events: NumberRecorded[] = [
+          { type: 'NumberRecorded', data: { number: 1 } },
+          { type: 'NumberRecorded', data: { number: 2 } },
+          { type: 'NumberRecorded', data: { number: 3 } },
+          { type: 'NumberRecorded', data: { number: 4 } },
+          { type: 'NumberRecorded', data: { number: 5 } },
+          { type: 'NumberRecorded', data: { number: 6 } },
+          { type: 'NumberRecorded', data: { number: 7 } },
+          { type: 'NumberRecorded', data: { number: 8 } },
+          { type: 'NumberRecorded', data: { number: 9 } },
+          { type: 'NumberRecorded', data: { number: 10 } },
+        ];
+        const appendResult = await eventStore.appendToStream(
+          streamName,
+          events,
+        );
+        await eventStore.appendToStream(otherStreamName, events);
+
+        const result: NumberRecorded[] = [];
+
+        let shouldThrowRandomError = false;
+
+        // When
+        const consumer = eventStoreDBEventStoreConsumer({
+          connectionString,
+          from: { stream: streamName },
+        });
+        consumer.reactor<NumberRecorded>({
+          processorId: uuid(),
+          stopAfter: (event) =>
+            event.metadata.globalPosition ===
+            appendResult.lastEventGlobalPosition,
+          eachMessage: (event) => {
+            if (shouldThrowRandomError) {
+              return Promise.reject(new Error('Random error'));
+            }
+
+            result.push(event);
+
+            shouldThrowRandomError = !shouldThrowRandomError;
+            return Promise.resolve();
+          },
+        });
+
+        try {
+          await consumer.start();
+
+          assertThatArray(result.map((e) => e.data.number)).containsExactly(1);
+        } finally {
+          await consumer.close();
+        }
+      },
+    );
+
     void it(
       `handles all events from $all streams for subscription to stream`,
       withDeadline,
@@ -578,7 +640,6 @@ void describe('EventStoreDB event store started consumer', () => {
             eachMessage: async (event) => {
               await waitForStart.wait;
               result.push(event);
-              console.log(`Received event: ${JSONParser.stringify(event)}`);
             },
           });
 

src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.ts

@@ -123,12 +123,17 @@ export const eventStoreDBEventStoreConsumer = <
       }),
     );
 
+    const error = result.find((r) => r.status === 'rejected')?.reason as
+      | Error
+      | undefined;
+
     return result.some(
       (r) => r.status === 'fulfilled' && r.value?.type !== 'STOP',
     )
       ? undefined
       : {
           type: 'STOP',
+          error: error ? EmmettError.mapFrom(error) : undefined,
         };
   };
 

src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts

@@ -123,11 +123,13 @@ class SubscriptionSequentialHandler<
 > extends Transform {
   private options: SubscriptionSequentialHandlerOptions<MessageType>;
   private from: EventStoreDBEventStoreConsumerType | undefined;
+  public isRunning: boolean;
 
   constructor(options: SubscriptionSequentialHandlerOptions<MessageType>) {
     super({ objectMode: true, ...options });
     this.options = options;
     this.from = options.from;
+    this.isRunning = true;
   }
 
   async _transform(
@@ -136,23 +138,27 @@ class SubscriptionSequentialHandler<
     callback: (error?: Error | null) => void,
   ): Promise<void> {
     try {
-      if (!resolvedEvent.event) return;
+      if (!this.isRunning || !resolvedEvent.event) {
+        callback();
+        return;
+      }
 
       const message = mapFromESDBEvent(resolvedEvent, this.from);
       const messageCheckpoint = getCheckpoint(message);
 
       const result = await this.options.eachBatch([message]);
 
       if (result && result.type === 'STOP') {
+        this.isRunning = false;
         if (!result.error) this.push(messageCheckpoint);
 
         this.push(result);
+        this.push(null);
         callback();
         return;
       }
 
       this.push(messageCheckpoint);
-
       callback();
     } catch (error) {
       callback(error as Error);
@@ -172,6 +178,7 @@ export const eventStoreDBSubscription = <
   let isRunning = false;
 
   let start: Promise<void>;
+  let processor: SubscriptionSequentialHandler<MessageType>;
 
   let subscription: StreamSubscription<MessageType>;
 
@@ -186,8 +193,12 @@ export const eventStoreDBSubscription = <
 
   const stopSubscription = (callback?: () => void): Promise<void> => {
     isRunning = false;
+    if (processor) processor.isRunning = false;
     return subscription
       .unsubscribe()
+      .then(() => {
+        subscription.destroy();
+      })
       .catch((err) => console.error('Error during unsubscribe.%s', err))
       .finally(callback ?? (() => {}));
   };
@@ -204,32 +215,47 @@ export const eventStoreDBSubscription = <
           );
           subscription = subscribe(client, from, options);
 
-          pipeline(
-            subscription,
-            new SubscriptionSequentialHandler({
-              client,
-              from,
-              batchSize,
-              eachBatch,
-              resilience,
-            }),
-            new (class extends Writable {
-              async _write(
-                chunk: bigint | MessageHandlerResult,
-                _encoding: string,
-                done: () => void,
-              ) {
-                if (isBigint(chunk)) {
-                  options.startFrom = {
-                    lastCheckpoint: chunk,
-                  };
-                  done();
-                  return;
-                }
+          processor = new SubscriptionSequentialHandler({
+            client,
+            from,
+            batchSize,
+            eachBatch,
+            resilience,
+          });
+
+          const handler = new (class extends Writable {
+            async _write(
+              result: bigint | MessageHandlerResult,
+              _encoding: string,
+              done: () => void,
+            ) {
+              if (!isRunning) return;
+
+              if (isBigint(result)) {
+                options.startFrom = {
+                  lastCheckpoint: result,
+                };
+                done();
+                return;
+              }
 
-                await stopSubscription(done);
+              if (result && result.type === 'STOP' && result.error) {
+                console.error(
+                  `Subscription stopped with error code: ${result.error.errorCode}, message: ${
+                    result.error.message
+                  }.`,
+                );
               }
-            })({ objectMode: true }),
+
+              await stopSubscription();
+              done();
+            }
+          })({ objectMode: true });
+
+          pipeline(
+            subscription,
+            processor,
+            handler,
             async (error: Error | null) => {
               console.info(`Stopping subscription.`);
               await stopSubscription(() => {

src/packages/emmett/src/errors/index.ts

@@ -43,6 +43,24 @@ export class EmmettError extends Error {
     // 👇️ because we are extending a built-in class
     Object.setPrototypeOf(this, EmmettError.prototype);
   }
+
+  public static mapFrom(
+    error: Error | { message?: string; errorCode?: number },
+  ): EmmettError {
+    if (error instanceof EmmettError) {
+      return error;
+    }
+
+    return new EmmettError({
+      errorCode:
+        'errorCode' in error &&
+        error.errorCode !== undefined &&
+        error.errorCode !== null
+          ? error.errorCode
+          : 500,
+      message: error.message ?? 'An unknown error occurred',
+    });
+  }
 }
 
 export class ConcurrencyError extends EmmettError {