Dublicate retry (#78)

Author: if0s

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 2.8.1 (December 28, 2023)
+* Fixed duplicate retries and added exponential backoff in `Subscribe to PubSub` trigger
+
 ## 2.8.0 (December 21, 2023)
 * Added new `Subscribe to PubSub` trigger
 

README.md

@@ -142,9 +142,9 @@ None.
 * **payload** - (object, required): Dynamically generated content of the event
 
 #### Limitations:
+* The component starts tracking changes after the first execution (it means you have to "run-now" flow with this trigger or wait for the first execution by the scheduler to establish a connection and only after this new event will be listened)
 * If you use **"Ordinary"** flow: 
   * Make sure that you execute it at least once per 3 days - according to the [documentation](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#replaying-an-event-stream) Salesforce stores events for up to 3 days.
-  * The component starts tracking changes after the first execution
 * To `Retrieve new sample from Salesforce v2` you need to trigger an event on Salesforce side or provide a sample manually
 
 ### Deprecated triggers

component.json

@@ -3,7 +3,7 @@
   "description": "Customer relationship management (CRM) software & cloud computing from the leader in CRM solutions for businesses large & small.",
   "docsUrl": "https://github.com/elasticio/salesforce-component-v2",
   "url": "http://www.salesforce.com/",
-  "version": "2.8.0",
+  "version": "2.8.1",
   "authClientTypes": [
     "oauth2"
   ],

lib/PubSubClient.js

@@ -140,35 +140,47 @@ class PubSubClient {
     this.subscription.write(writeOptions);
 
     this.subscription.on('data', (data) => {
-      const latestReplayId = Number(data.latestReplayId.readBigUInt64BE());
-      if (data.events) {
-        for (const event of data.events) {
-          const replayId = Number(event.replayId.readBigUInt64BE());
-          const payload = this.schema.type.fromBuffer(event.event.payload);
-          this.flattenSinglePropertyObjects(payload);
-          eventEmitter.emit('data', { event: { replayId }, payload });
-        }
-        if (!data.pendingNumRequested) {
-          this.logger.info(`requesting ${numRequested} more records`);
-          this.subscription.write({ topicName: this.topicName, numRequested });
+      try {
+        const latestReplayId = Number(data.latestReplayId.readBigUInt64BE());
+        if (data.events) {
+          for (const event of data.events) {
+            const replayId = Number(event.replayId.readBigUInt64BE());
+            const payload = this.schema.type.fromBuffer(event.event.payload);
+            this.flattenSinglePropertyObjects(payload);
+            eventEmitter.emit('data', { event: { replayId }, payload });
+          }
+          if (!data.pendingNumRequested) {
+            this.logger.info(`requesting ${numRequested} more records`);
+            this.subscription.write({ topicName: this.topicName, numRequested });
+          }
+        } else {
+          this.logger.debug(`Received keepalive message. Latest replay ID: ${latestReplayId}`);
+          data.latestReplayId = latestReplayId;
+          this.lastKeepAlive = new Date().getTime();
+          eventEmitter.emit('keepalive', data);
         }
-      } else {
-        this.logger.debug(`Received keepalive message. Latest replay ID: ${latestReplayId}`);
-        data.latestReplayId = latestReplayId;
-        this.lastKeepAlive = new Date().getTime();
-        eventEmitter.emit('keepalive', data);
+      } catch (err) {
+        this.logger.warn('Failed to decode incoming data, skipping');
       }
     });
     this.subscription.on('end', () => {
       this.logger.warn('gRPC stream ended');
       eventEmitter.emit('end');
     });
+    const tryStringify = (obj) => {
+      try {
+        return JSON.stringify(obj);
+      } catch (err) {
+        if (typeof obj === 'string') return obj;
+        return ('Can\'t parse message');
+      }
+    };
     this.subscription.on('error', (error) => {
-      this.logger.error(`gRPC stream error: ${JSON.stringify(error)}`);
+      this.logger.error(`gRPC stream error: ${tryStringify(error)}`);
       eventEmitter.emit('error', error);
     });
     this.subscription.on('status', (status) => {
-      this.logger.warn(`gRPC stream status: ${JSON.stringify(status)}`);
+      this.logger.warn(`gRPC stream status: ${tryStringify(status)}`);
       eventEmitter.emit('status', status);
     });
     this.eventEmitter = eventEmitter;
@@ -188,8 +200,8 @@ class PubSubClient {
       this.subscription = null;
       this.client.close();
       this.client = null;
-    // eslint-disable-next-line no-empty
-    } catch (_e) {}
+      // eslint-disable-next-line no-empty
+    } catch (_e) { }
   }
 }
 

lib/triggers/streamPlatformEventsPubSub.js

@@ -1,4 +1,4 @@
-/* eslint-disable no-use-before-define, no-unused-vars */
+/* eslint-disable no-use-before-define, no-unused-vars, no-empty */
 const { messages } = require('elasticio-node');
 const commons = require('@elastic.io/component-commons-library');
 const { callJSForceMethod } = require('../helpers/wrapper');
@@ -10,14 +10,16 @@ let context;
 let refreshTokenNeeded = false;
 let authFailureRetry = 1;
 let connectFailureRetry = 1;
+let keepaliveFailureRetry = 1;
 let nextStartReplayId;
 const MAX_AUTH_RETRIES = 2;
 const MAX_CONNECT_RETRIES = 5;
 let eventEmitter;
 let creationInProgress = false;
 
 const reconnectHandler = async (reconnectSleepTime, cfg) => {
-  if (client) client.close();
+  context.logger.warn(`Restarting in ${Math.round(reconnectSleepTime / 1000)} seconds`);
+  if (client) try { client.close(); } catch (_e) { }
   creationInProgress = false;
   client = null;
   await commons.sleep(reconnectSleepTime);
@@ -49,7 +51,7 @@ const isNumberNaN = (num) => Number(num).toString() === 'NaN';
 
 async function processTrigger(msg, cfg, snapshot) {
   context = this;
-  if (client) {
+  if (client && client.getLastKeepAlive()) {
     this.logger.info(`Subscription is still running, last keepalive from Salesforce ${new Date(client.getLastKeepAlive()).toISOString()}, waiting for new messages`);
     client.setLogger(context.logger);
     return;
@@ -92,20 +94,16 @@ async function processTrigger(msg, cfg, snapshot) {
   const instanceUrl = credentials.undefined_params.instance_url;
   nextStartReplayId = snapshot?.nextStartReplayId || initialReplayId;
 
-  let emitted = false;
+  let emittedOrAlive = false;
   const timeOut = async (t) => {
     await commons.sleep(t);
-    if (emitted) {
-      await commons.sleep(3000000);
-    } else {
-      throw new Error(`TimeOut ${t}ms limit reached`);
-    }
+    if (!emittedOrAlive) throw new Error(`TimeOut ${t}ms limit reached`);
   };
 
   context.logger.info('Preparing SalesForce connection...');
   try {
     await Promise.race([
-      await client.connect(accessToken, instanceUrl),
+      client.connect(accessToken, instanceUrl),
       timeOut(30000),
     ]);
     authFailureRetry = 1;
@@ -119,7 +117,7 @@ async function processTrigger(msg, cfg, snapshot) {
     } else if (connectFailureRetry <= MAX_CONNECT_RETRIES) {
       context.logger.warn(`Got error during connection to Salesforce: ${err.message}, lets call processTrigger one more (retry ${connectFailureRetry} of ${MAX_CONNECT_RETRIES})`);
       connectFailureRetry++;
-      await reconnectHandler(60000, cfg);
+      await reconnectHandler(Math.round(2 ** (connectFailureRetry + 5)) * 1000, cfg);
     } else {
       context.logger.error(`Failed to connect to Salesforce: ${err.message}`);
       await context.emit('error', err);
@@ -135,16 +133,19 @@ async function processTrigger(msg, cfg, snapshot) {
     context.logger.info('Got new message, emitting');
     await context.emit('data', messages.newMessageWithBody(event));
     nextStartReplayId = event.replayId;
-    emitted = true;
+    emittedOrAlive = true;
     await this.emit('snapshot', { nextStartReplayId });
   });
 
+  eventEmitter.once('keepalive', (data) => { context.logger.info('Got keepalive message'); });
+
   try {
     await Promise.race([
       new Promise((resolve, reject) => {
         context.logger.info('Processing existing events if present and waiting for keepalive message to finish');
         eventEmitter.on('keepalive', async (data) => {
-          connectFailureRetry = 1;
+          keepaliveFailureRetry = 1;
+          emittedOrAlive = true;
           nextStartReplayId = data.latestReplayId;
           await this.emit('snapshot', { nextStartReplayId });
           resolve();
@@ -156,22 +157,21 @@ async function processTrigger(msg, cfg, snapshot) {
       timeOut(300000),
     ]);
   } catch (err) {
-    if (!err.message.includes('INVALID_ARGUMENT') && connectFailureRetry <= MAX_CONNECT_RETRIES) {
-      context.logger.warn(`Failed to get keepalive message to finish: ${err.message}, lets call processTrigger one more time (retry ${connectFailureRetry} of ${MAX_CONNECT_RETRIES})`);
-      connectFailureRetry++;
-      await reconnectHandler(2000, cfg);
+    if (!err.message?.includes('INVALID_ARGUMENT') && keepaliveFailureRetry <= MAX_CONNECT_RETRIES) {
+      context.logger.warn(`Failed to get keepalive message to finish: ${err.message}, lets call processTrigger one more time (retry ${keepaliveFailureRetry} of ${MAX_CONNECT_RETRIES})`);
+      keepaliveFailureRetry++;
+      await reconnectHandler(Math.round(2 ** (keepaliveFailureRetry + 5)) * 1000, cfg);
     } else {
       context.logger.error(`Failed to get keepalive message to finish: ${err.message}`);
       await context.emit('error', err);
       throw err;
     }
-    await reconnectHandler(2000, cfg);
   }
 
   creationInProgress = false;
 
   if (process.env.ELASTICIO_FLOW_TYPE === 'ordinary') {
-    client.close();
+    try { client.close(); } catch (_e) { }
     client = null;
     context.logger.info('"Subscribe to PubSub" trigger finished successfully');
     return;