Improved error handling

Author: mike182uk

src/fedify/mq/gcloud-pubsub-mq.ts

@@ -1,11 +1,11 @@
-import type { Message, PubSub } from '@google-cloud/pubsub';
-
 import type {
     MessageQueue,
     MessageQueueEnqueueOptions,
     MessageQueueListenOptions,
 } from '@fedify/fedify';
+import type { Message, PubSub } from '@google-cloud/pubsub';
 import type { Logger } from '@logtape/logtape';
+import * as Sentry from '@sentry/node';
 
 export class GCloudPubSubMessageQueue implements MessageQueue {
     private pubSubClient: PubSub;
@@ -56,6 +56,10 @@ export class GCloudPubSubMessageQueue implements MessageQueue {
             this.logger.error(
                 `Failed to enqueue message [FedifyID: ${message.id}]: ${error}`,
             );
+
+            Sentry.captureException(error);
+
+            throw error;
         }
     }
 
@@ -67,31 +71,49 @@ export class GCloudPubSubMessageQueue implements MessageQueue {
             this.subscriptionIdentifier,
         );
 
-        subscription.on('message', async (message: Message) => {
-            const fedifyId = message.attributes.fedifyId ?? 'unknown';
+        subscription
+            .on('message', async (message: Message) => {
+                const fedifyId = message.attributes.fedifyId ?? 'unknown';
 
-            this.logger.info(
-                `Handling message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`,
-            );
+                this.logger.info(
+                    `Handling message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`,
+                );
 
-            try {
-                const json = JSON.parse(message.data.toString());
+                try {
+                    const json = JSON.parse(message.data.toString());
 
-                await handler(json);
+                    await handler(json);
 
-                message.ack();
+                    message.ack();
 
-                this.logger.info(
-                    `Acknowledged message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`,
-                );
-            } catch (error) {
-                message.nack();
+                    this.logger.info(
+                        `Acknowledged message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]`,
+                    );
+                } catch (error) {
+                    message.nack();
+
+                    this.logger.error(
+                        `Failed to handle message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]: ${error}`,
+                    );
 
+                    Sentry.captureException(error);
+                }
+            })
+            .on('error', (error) => {
                 this.logger.error(
-                    `Failed to handle message [FedifyID: ${fedifyId}, PubSubID: ${message.id}]: ${error}`,
+                    `Subscription [${this.subscriptionIdentifier}] error occurred: ${error}`,
                 );
-            }
-        });
+
+                Sentry.captureException(error);
+
+                // This is a fatal error, so we should throw to stop the listener / process
+                throw error;
+            })
+            .on('close', () => {
+                this.logger.info(
+                    `Subscription [${this.subscriptionIdentifier}] closed`,
+                );
+            });
 
         return await new Promise((resolve) => {
             options.signal?.addEventListener('abort', () => {

src/fedify/mq/gcloud-pubsub-mq.unit.test.ts

@@ -93,15 +93,15 @@ describe('enqueue', () => {
         );
     });
 
-    it('should log an error if the message fails to be published', async () => {
-        mockTopic.publishMessage = vi
-            .fn()
-            .mockRejectedValue(new Error('Failed to publish message'));
+    it('should throw an error and log if the message fails to be published', async () => {
+        const error = new Error('Failed to publish message');
+
+        mockTopic.publishMessage = vi.fn().mockRejectedValue(error);
 
         const enqueuePromise = messageQueue.enqueue(MESSAGE);
 
         vi.runAllTimers();
-        await enqueuePromise;
+        await expect(enqueuePromise).rejects.toThrow(error);
 
         expect(mockLogger.error).toHaveBeenCalledWith(
             `Failed to enqueue message [FedifyID: ${MESSAGE.id}]: Error: Failed to publish message`,
@@ -145,6 +145,7 @@ describe('listen', () => {
                         if (event === 'message') {
                             callback(MESSAGE);
                         }
+                        return mockSubscription;
                     },
                 ),
             removeAllListeners: vi.fn().mockReturnThis(),