fix: timeout error despite topic publishing messages

Author: eetrip

src/pubsub-publisher/pubsub-publisher.module.spec.ts

@@ -38,7 +38,6 @@ describe('PubSubPublisherModule', () => {
     const publishJsonMock = jest
       .spyOn(topic, 'publishMessage')
       .mockResolvedValue(mockedRes as never);
-    const clearTimeoutMock = jest.spyOn(global, 'clearTimeout');
 
     const data = {
       a: 1,
@@ -51,7 +50,6 @@ describe('PubSubPublisherModule', () => {
 
     expect(res).toBe(mockedRes[0]);
     expect(publishJsonMock).toBeCalledWith({ data: Buffer.from(JSON.stringify(data)), attributes });
-    expect(clearTimeoutMock).toBeCalled();
   });
 
   it('publishes to topic raw buffer', async () => {
@@ -63,7 +61,6 @@ describe('PubSubPublisherModule', () => {
     const publishJsonMock = jest
       .spyOn(topic, 'publishMessage')
       .mockResolvedValue(mockedRes as never);
-    const clearTimeoutMock = jest.spyOn(global, 'clearTimeout');
 
     const data = Buffer.from('hello world');
     const attributes = {
@@ -74,7 +71,6 @@ describe('PubSubPublisherModule', () => {
 
     expect(res).toBe(mockedRes[0]);
     expect(publishJsonMock).toBeCalledWith({ data, attributes });
-    expect(clearTimeoutMock).toBeCalled();
   });
 
   it('rejects with PubSubTimeoutError if publishing hangs', async () => {

src/pubsub-publisher/pubsub-publisher.service.ts

@@ -37,36 +37,11 @@ export class PubSubPublisherService {
     data: Buffer,
     attributes: Record<string, string>
   ): Promise<string> {
-    const timeout = setTimeoutAsync(this.settings.requestTimeoutMillis);
-
-    const [result] = await Promise.race([
-      topic.publishMessage({ data, attributes }),
-      timeout.then(() => {
-        throw new PubSubTimeoutError();
-      }),
-    ]);
-
-    timeout.clear();
-
+    const [result] = await topic.publishMessage({ data, attributes });
     return result;
   }
 }
 
-function setTimeoutAsync(millis: number) {
-  let timeoutResolve: () => void;
-  let timeout: NodeJS.Timeout;
-  const promise = new Promise<void>((resolve) => {
-    timeoutResolve = resolve;
-    timeout = setTimeout(() => resolve(), millis);
-  });
-  return Object.assign(promise, {
-    clear: () => {
-      clearTimeout(timeout);
-      timeoutResolve();
-    },
-  });
-}
-
 export class PubSubTimeoutError extends InternalServerErrorException {
   constructor() {
     super('Request timeout');

src/pubsub/pubsub.module.ts

@@ -6,6 +6,7 @@ import { PubSubService } from './pubsub.service';
 import { Token } from '../utils/token';
 import { AsyncSettings, SettingsProvider } from '../utils/settings-provider';
 import { credentials } from '@grpc/grpc-js';
+import { PubSubPublisherSettings } from '../pubsub-publisher/pubsub-publisher.module';
 
 export const PubSubSettings = Symbol('PubSubSettings');
 export type PubSubSettings = PubSubInstanceOrSettings & {
@@ -78,6 +79,10 @@ export class PubSubModule {
       module: PubSubModule,
       global: true,
       providers: [
+        <SettingsProvider<PubSubPublisherSettings>>{
+          provide: PubSubPublisherSettings,
+          ...settings,
+        },
         <SettingsProvider<PubSubSettings>>{
           provide: PubSubSettings,
           ...settings,
@@ -89,9 +94,12 @@ export class PubSubModule {
         },
         {
           provide: PubSubService,
-          inject: [PubSub, PubSubSettings],
-          useFactory: (pubSub: PubSub, settings: PubSubSettings) =>
-            new PubSubService(pubSub, settings),
+          inject: [PubSub, PubSubSettings, PubSubPublisherSettings],
+          useFactory: (
+            pubSub: PubSub,
+            settings: PubSubSettings,
+            timeoutMillis: PubSubPublisherSettings
+          ) => new PubSubService(pubSub, settings, timeoutMillis),
         },
       ],
       exports: [PubSubService],

src/pubsub/pubsub.service.ts

@@ -2,6 +2,9 @@ import { Topic, Subscription, PubSub } from '@google-cloud/pubsub';
 import { Injectable, OnModuleDestroy } from '@nestjs/common';
 import { PubSubSettings, SubscriptionSettings, TopicSettings } from './pubsub.module';
 import { Token } from '../utils/token';
+import { defaultRetryOptions } from '../utils/default-retry-options';
+import { PubSubPublisherSettings } from '../pubsub-publisher/pubsub-publisher.module';
+import { mergeObjects } from '../utils/deep-merge-object';
 
 @Injectable()
 export class PubSubService implements OnModuleDestroy {
@@ -11,7 +14,11 @@ export class PubSubService implements OnModuleDestroy {
   private openTopics: Record<Token, Topic> = {};
   private openSubscriptions: Record<Token, Subscription> = {};
 
-  constructor(public readonly pubSub: PubSub, private settings: PubSubSettings) {
+  constructor(
+    public readonly pubSub: PubSub,
+    private settings: PubSubSettings,
+    private timeoutMillis: PubSubPublisherSettings
+  ) {
     this.topicsSettings = settings.topics;
     this.subscriptionsSettings = createSubscriptionsStore(settings.topics);
   }
@@ -23,7 +30,11 @@ export class PubSubService implements OnModuleDestroy {
         throw new Error(`Cannot find topic by alias: ${String(token)}`);
       }
 
-      const { name, options } = topicSettings;
+      const { name, options: userOptions } = topicSettings;
+      const options = mergeObjects(
+        defaultRetryOptions(this.timeoutMillis.requestTimeoutMillis),
+        userOptions as Record<string, unknown>
+      );
       return this.pubSub.topic(name, options);
     });
   }

src/utils/deep-merge-object.ts

@@ -0,0 +1,41 @@
+export const mergeObjects = (
+  obj1: Record<string, unknown>,
+  obj2: Record<string, unknown> = {}
+): Record<string, unknown> => {
+  // create a new object that will be the merged version of obj1 and obj2
+  const mergedObj: Record<string, unknown> = {};
+
+  // loop through all the keys in obj1
+  for (const key of Object.keys(obj1)) {
+    // if the key is not present in obj2, or if the value at that key is not an object,
+    // add the key-value pair to the merged object
+    if (!obj2.hasOwnProperty(key) || typeof obj1[key] !== 'object') {
+      mergedObj[key] = obj1[key];
+    } else {
+      // if the value at the key in obj1 is an object and the same key exists in obj2,
+      // merge the objects and add the merged object to the mergedObj
+      mergedObj[key] = mergeObjects(
+        obj1[key] as Record<string, unknown>,
+        obj2[key] as Record<string, unknown>
+      );
+    }
+  }
+
+  // loop through all the keys in obj2
+  for (const key of Object.keys(obj2)) {
+    // if the key is not present in obj1 or if the value at that key is not an object,
+    // add the key-value pair to the merged object
+    if (!obj1.hasOwnProperty(key) || typeof obj2[key] !== 'object') {
+      mergedObj[key] = obj2[key];
+    } else {
+      // if the value at the key in obj2 is an object and the same key exists in obj1,
+      // merge the objects and add the merged object to the mergedObj
+      mergedObj[key] = mergeObjects(
+        obj1[key] as Record<string, unknown>,
+        obj2[key] as Record<string, unknown>
+      );
+    }
+  }
+
+  return mergedObj;
+};

src/utils/deep-merge-objects.spec.ts

@@ -0,0 +1,35 @@
+import { mergeObjects } from './deep-merge-object';
+
+describe('deep merge two objects', () => {
+  it('merges two objects with nested values', () => {
+    const retryDefaults = {
+      prop1: 'prop should stay',
+      retry: {
+        backoffSettings: {
+          prop2: 'prop should stay',
+          timeoutInMillis: 1,
+        },
+      },
+    };
+    const options = {
+      prop3: 'prop should stay',
+      retry: {
+        backoffSettings: {
+          timeoutInMillis: 100,
+          prop4: 'prop should stay',
+        },
+      },
+    };
+    expect(mergeObjects(retryDefaults, options)).toEqual({
+      prop1: 'prop should stay',
+      prop3: 'prop should stay',
+      retry: {
+        backoffSettings: {
+          timeoutInMillis: 100,
+          prop2: 'prop should stay',
+          prop4: 'prop should stay',
+        },
+      },
+    });
+  });
+});

src/utils/default-retry-options.spec.ts

@@ -0,0 +1,19 @@
+import { defaultRetryOptions } from './default-retry-options';
+
+describe('get default retry options', () => {
+  it('returns default retry options with passed timeout in milliseconds', () => {
+    const timeoutMillis = 100;
+    expect(defaultRetryOptions(timeoutMillis)).toEqual({
+      gaxOpts: {
+        retry: {
+          backoffSettings: {
+            initialRetryDelayMillis: 1,
+            retryDelayMultiplier: 1.3,
+            maxRetryDelayMillis: 100,
+            totalTimeoutMillis: timeoutMillis,
+          },
+        },
+      },
+    });
+  });
+});

src/utils/default-retry-options.ts

@@ -0,0 +1,14 @@
+export type DefaultRetryOptions = Record<string, unknown>;
+
+export const defaultRetryOptions = (timeoutMillis: number): DefaultRetryOptions => ({
+  gaxOpts: {
+    retry: {
+      backoffSettings: {
+        initialRetryDelayMillis: 1,
+        retryDelayMultiplier: 1.3,
+        maxRetryDelayMillis: 100,
+        totalTimeoutMillis: timeoutMillis,
+      },
+    },
+  },
+});