The Replay ID validation failed (#85)

Author: if0s

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 2.8.4 (July 11, 2024)
+* Attempt to fix error `The Replay ID validation failed` when `Subscribe to PubSub` trigger does't emit messages more than three days
+* Update Sailor version to 2.7.2
+* Update component-commons-library version to 3.2.0
+
 ## 2.8.3 (February 27, 2024)
 * The component interface has not changed. This is a technical enhancement! Introduced baseURL parameter in the `Raw Request` Action's configuration of the axios library. Refer to the [Readme](README.md#raw-request) for the details. 
 It will not affect any of the existing integration. Instead, it gives more flexibility allowing to call other REST endpoints than the standard `/services/data` [(#82)](https://github.com/elasticio/salesforce-component-v2/issues/82)

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": "https://www.salesforce.com/",
-  "version": "2.8.3",
+  "version": "2.8.4",
   "authClientTypes": [
     "oauth2"
   ],

lib/triggers/streamPlatformEventsPubSub.js

@@ -102,7 +102,13 @@ async function processTrigger(msg, cfg, snapshot) {
   const { credentials } = await getSecret(this, secretId, msg.id);
   const accessToken = credentials.access_token;
   const instanceUrl = credentials.undefined_params.instance_url;
-  nextStartReplayId = nextStartReplayId || snapshot?.nextStartReplayId || initialReplayId;
+  nextStartReplayId = nextStartReplayId || snapshot?.nextStartReplayId;
+
+  let replayIdFromConfig = false;
+  if (!nextStartReplayId && initialReplayId) {
+    replayIdFromConfig = true;
+    nextStartReplayId = initialReplayId;
+  }
 
   let emittedOrAlive = false;
   const timeOut = async (t) => {
@@ -171,6 +177,12 @@ async function processTrigger(msg, cfg, snapshot) {
       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 if (err.message.includes('INVALID_ARGUMENT') && !replayIdFromConfig) {
+      context.logger.error(`Failed to get keepalive message to finish, replayId in seems to be invalid, it will be ignored: ${err.message}, lets call processTrigger one more time (retry ${keepaliveFailureRetry} of ${MAX_CONNECT_RETRIES})`);
+      keepaliveFailureRetry++;
+      await context.emit('snapshot', { });
+      nextStartReplayId = null;
+      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);