Add automatic retries to data-sink-worker

backend/src/database/migrations/V1699626027__add-retries-to-results.sql

@@ -0,0 +1,3 @@
+ALTER TABLE integration.results
+ADD COLUMN retries INT,
+ADD COLUMN "delayedUntil" TIMESTAMP with time zone NULL;

backend/src/serverless/integrations/services/integrationTickProcessor.ts

@@ -1,7 +1,11 @@
 import { processPaginated, singleOrDefault } from '@crowd/common'
 import { INTEGRATION_SERVICES } from '@crowd/integrations'
 import { LoggerBase, getChildLogger } from '@crowd/logging'
-import { IntegrationRunWorkerEmitter, IntegrationStreamWorkerEmitter } from '@crowd/sqs'
+import {
+  IntegrationRunWorkerEmitter,
+  IntegrationStreamWorkerEmitter,
+  DataSinkWorkerEmitter,
+} from '@crowd/sqs'
 import { IntegrationRunState, IntegrationType } from '@crowd/types'
 import SequelizeRepository from '@/database/repositories/sequelizeRepository'
 import MicroserviceRepository from '@/database/repositories/microserviceRepository'
@@ -14,6 +18,7 @@ import { sendNodeWorkerMessage } from '../../utils/nodeWorkerSQS'
 import {
   getIntegrationRunWorkerEmitter,
   getIntegrationStreamWorkerEmitter,
+  getDataSinkWorkerEmitter,
 } from '../../utils/serviceSQS'
 import { IntegrationServiceBase } from './integrationServiceBase'
 
@@ -26,6 +31,8 @@ export class IntegrationTickProcessor extends LoggerBase {
 
   private intStreamWorkerEmitter: IntegrationStreamWorkerEmitter
 
+  private dataSinkWorkerEmitter: DataSinkWorkerEmitter
+
   constructor(
     options: IServiceOptions,
     private readonly integrationServices: IntegrationServiceBase[],
@@ -46,6 +53,7 @@ export class IntegrationTickProcessor extends LoggerBase {
     if (!this.emittersInitialized) {
       this.intRunWorkerEmitter = await getIntegrationRunWorkerEmitter()
       this.intStreamWorkerEmitter = await getIntegrationStreamWorkerEmitter()
+      this.dataSinkWorkerEmitter = await getDataSinkWorkerEmitter()
 
       this.emittersInitialized = true
     }
@@ -220,6 +228,7 @@ export class IntegrationTickProcessor extends LoggerBase {
     await this.initEmitters()
     await this.intRunWorkerEmitter.checkRuns()
     await this.intStreamWorkerEmitter.checkStreams()
+    await this.dataSinkWorkerEmitter.checkResults()
 
     // TODO check streams as well
     this.log.trace('Checking for delayed integration runs!')

backend/src/serverless/utils/serviceSQS.ts

@@ -3,6 +3,7 @@ import {
   IntegrationStreamWorkerEmitter,
   IntegrationSyncWorkerEmitter,
   SearchSyncWorkerEmitter,
+  DataSinkWorkerEmitter,
   SqsClient,
   getSqsClient,
 } from '@crowd/sqs'
@@ -64,3 +65,12 @@ export const getIntegrationSyncWorkerEmitter = async (): Promise<IntegrationSync
   await integrationSyncWorkerEmitter.init()
   return integrationSyncWorkerEmitter
 }
+
+let dataSinkWorkerEmitter: DataSinkWorkerEmitter
+export const getDataSinkWorkerEmitter = async (): Promise<DataSinkWorkerEmitter> => {
+  if (dataSinkWorkerEmitter) return dataSinkWorkerEmitter
+
+  dataSinkWorkerEmitter = new DataSinkWorkerEmitter(getClient(), tracer, log)
+  await dataSinkWorkerEmitter.init()
+  return dataSinkWorkerEmitter
+}

services/apps/data_sink_worker/config/default.json

@@ -5,5 +5,8 @@
   "unleash": {},
   "temporal": {
     "automationsTaskQueue": "automations"
+  },
+  "worker": {
+    "maxStreamRetries": 5
   }
 }

services/apps/data_sink_worker/src/bin/process-results.ts

@@ -12,7 +12,12 @@ import { DbStore, getDbConnection } from '@crowd/database'
 import { getServiceTracer } from '@crowd/tracing'
 import { getServiceLogger } from '@crowd/logging'
 import { getRedisClient } from '@crowd/redis'
-import { NodejsWorkerEmitter, SearchSyncWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import {
+  NodejsWorkerEmitter,
+  SearchSyncWorkerEmitter,
+  DataSinkWorkerEmitter,
+  getSqsClient,
+} from '@crowd/sqs'
 import { initializeSentimentAnalysis } from '@crowd/sentiment'
 import { getUnleashClient } from '@crowd/feature-flags'
 import { Client as TemporalClient, getTemporalClient } from '@crowd/temporal'
@@ -49,13 +54,17 @@ setImmediate(async () => {
   const searchSyncWorkerEmitter = new SearchSyncWorkerEmitter(sqsClient, tracer, log)
   await searchSyncWorkerEmitter.init()
 
+  const dataSinkWorkerEmitter = new DataSinkWorkerEmitter(sqsClient, tracer, log)
+  await dataSinkWorkerEmitter.init()
+
   const dbConnection = await getDbConnection(DB_CONFIG())
   const store = new DbStore(log, dbConnection)
 
   const service = new DataSinkService(
     store,
     nodejsWorkerEmitter,
     searchSyncWorkerEmitter,
+    dataSinkWorkerEmitter,
     redisClient,
     unleash,
     temporal,

services/apps/data_sink_worker/src/conf/index.ts

@@ -11,6 +11,18 @@ export interface ISlackAlertingConfig {
   url: string
 }
 
+export interface IWorkerConfig {
+  maxStreamRetries: number
+}
+
+let workerSettings: IWorkerConfig
+export const WORKER_SETTINGS = (): IWorkerConfig => {
+  if (workerSettings) return workerSettings
+
+  workerSettings = config.get<IWorkerConfig>('worker')
+  return workerSettings
+}
+
 let redisConfig: IRedisConfiguration
 export const REDIS_CONFIG = (): IRedisConfiguration => {
   if (redisConfig) return redisConfig

services/apps/data_sink_worker/src/jobs/processOldResults.ts

@@ -3,7 +3,7 @@ import { DbConnection, DbStore } from '@crowd/database'
 import { Unleash } from '@crowd/feature-flags'
 import { Logger } from '@crowd/logging'
 import { RedisClient } from '@crowd/redis'
-import { NodejsWorkerEmitter, SearchSyncWorkerEmitter } from '@crowd/sqs'
+import { NodejsWorkerEmitter, SearchSyncWorkerEmitter, DataSinkWorkerEmitter } from '@crowd/sqs'
 import { Client as TemporalClient } from '@crowd/temporal'
 import DataSinkRepository from '../repo/dataSink.repo'
 import DataSinkService from '../service/dataSink.service'
@@ -16,6 +16,7 @@ export const processOldResultsJob = async (
   redis: RedisClient,
   nodejsWorkerEmitter: NodejsWorkerEmitter,
   searchSyncWorkerEmitter: SearchSyncWorkerEmitter,
+  dataSinkWorkerEmitter: DataSinkWorkerEmitter,
   unleash: Unleash | undefined,
   temporal: TemporalClient,
   log: Logger,
@@ -26,6 +27,7 @@ export const processOldResultsJob = async (
     store,
     nodejsWorkerEmitter,
     searchSyncWorkerEmitter,
+    dataSinkWorkerEmitter,
     redis,
     unleash,
     temporal,

services/apps/data_sink_worker/src/main.ts

@@ -1,7 +1,12 @@
 import { getDbConnection } from '@crowd/database'
 import { getServiceTracer } from '@crowd/tracing'
 import { getServiceLogger } from '@crowd/logging'
-import { NodejsWorkerEmitter, SearchSyncWorkerEmitter, getSqsClient } from '@crowd/sqs'
+import {
+  NodejsWorkerEmitter,
+  SearchSyncWorkerEmitter,
+  DataSinkWorkerEmitter,
+  getSqsClient,
+} from '@crowd/sqs'
 import {
   DB_CONFIG,
   SENTIMENT_CONFIG,
@@ -48,11 +53,14 @@ setImmediate(async () => {
 
   const searchSyncWorkerEmitter = new SearchSyncWorkerEmitter(sqsClient, tracer, log)
 
+  const dataWorkerEmitter = new DataSinkWorkerEmitter(sqsClient, tracer, log)
+
   const queue = new WorkerQueueReceiver(
     sqsClient,
     dbConnection,
     nodejsWorkerEmitter,
     searchSyncWorkerEmitter,
+    dataWorkerEmitter,
     redisClient,
     unleash,
     temporal,
@@ -64,6 +72,7 @@ setImmediate(async () => {
   try {
     await nodejsWorkerEmitter.init()
     await searchSyncWorkerEmitter.init()
+    await dataWorkerEmitter.init()
 
     let processing = false
     setInterval(async () => {
@@ -77,6 +86,7 @@ setImmediate(async () => {
             redisClient,
             nodejsWorkerEmitter,
             searchSyncWorkerEmitter,
+            dataWorkerEmitter,
             unleash,
             temporal,
             log,

services/apps/data_sink_worker/src/queue/index.ts

@@ -5,6 +5,7 @@ import {
   DATA_SINK_WORKER_QUEUE_SETTINGS,
   NodejsWorkerEmitter,
   SearchSyncWorkerEmitter,
+  DataSinkWorkerEmitter,
   SqsClient,
   SqsQueueReceiver,
 } from '@crowd/sqs'
@@ -25,6 +26,7 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
     private readonly dbConn: DbConnection,
     private readonly nodejsWorkerEmitter: NodejsWorkerEmitter,
     private readonly searchSyncWorkerEmitter: SearchSyncWorkerEmitter,
+    private readonly dataSinkWorkerEmitter: DataSinkWorkerEmitter,
     private readonly redisClient: RedisClient,
     private readonly unleash: Unleash | undefined,
     private readonly temporal: TemporalClient,
@@ -44,6 +46,7 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
           new DbStore(this.log, this.dbConn, undefined, false),
           this.nodejsWorkerEmitter,
           this.searchSyncWorkerEmitter,
+          this.dataSinkWorkerEmitter,
           this.redisClient,
           this.unleash,
           this.temporal,
@@ -65,6 +68,10 @@ export class WorkerQueueReceiver extends SqsQueueReceiver {
             )
             break
           }
+          case DataSinkWorkerQueueMessageType.CHECK_RESULTS: {
+            await service.checkResults()
+            break
+          }
 
           default:
             throw new Error(`Unknown message type: ${message.type}`)

services/apps/data_sink_worker/src/repo/dataSink.data.ts

@@ -17,6 +17,9 @@ export interface IResultData {
   plan: string
   isTrialPlan: boolean
   name: string
+
+  retries: number
+  delayedUntil: string | null
 }
 
 export interface IFailedResultData {

services/apps/data_sink_worker/src/repo/dataSink.repo.ts

@@ -1,6 +1,6 @@
 import { DbStore, RepositoryBase } from '@crowd/database'
 import { Logger } from '@crowd/logging'
-import { IIntegrationResult, IntegrationResultState, TenantPlans } from '@crowd/types'
+import { IIntegrationResult, IntegrationResultState, PlatformType, TenantPlans } from '@crowd/types'
 import { IFailedResultData, IResultData } from './dataSink.data'
 
 export default class DataSinkRepository extends RepositoryBase<DataSinkRepository> {
@@ -18,6 +18,8 @@ export default class DataSinkRepository extends RepositoryBase<DataSinkRepositor
            r."streamId",
            r."apiDataId",
            r."integrationId",
+           r.retries,
+           r."delayedUntil",
            i.platform,
            t."hasSampleData", 
            t."plan",
@@ -206,7 +208,8 @@ export default class DataSinkRepository extends RepositoryBase<DataSinkRepositor
       `update integration.results
         set state = $(newState),
             error = null,
-            "updatedAt" = now()
+            "updatedAt" = now(),
+            "delayedUntil" = null
         where id in ($(resultIds:csv))`,
       {
         resultIds,
@@ -224,4 +227,50 @@ export default class DataSinkRepository extends RepositoryBase<DataSinkRepositor
 
     return result.map((r) => r.id)
   }
+
+  public async delayResult(resultId: string, until: Date): Promise<void> {
+    const result = await this.db().result(
+      `update integration.results
+       set  state = $(state),
+            "delayedUntil" = $(until),
+            retries = coalesce(retries, 0) + 1,
+            "updatedAt" = now()
+       where id = $(resultId)`,
+      {
+        resultId,
+        until,
+        state: IntegrationResultState.DELAYED,
+      },
+    )
+
+    this.checkUpdateRowCount(result.rowCount, 1)
+  }
+
+  public async getDelayedResults(
+    limit: number,
+  ): Promise<{ id: string; tenantId: string; platform: PlatformType }[]> {
+    this.ensureTransactional()
+
+    try {
+      const results = await this.db().any(
+        `
+        select r.id, r."tenantId", i.platform
+        from integration.results r
+        join integrations i on r."integrationId" = i.id
+        where r.state = $(delayedState)
+          and r."delayedUntil" < now()
+        limit ${limit}
+        for update skip locked;
+        `,
+        {
+          delayedState: IntegrationResultState.DELAYED,
+        },
+      )
+
+      return results.map((s) => ({ id: s.id, tenantId: s.tenantId, platform: s.platform }))
+    } catch (err) {
+      this.log.error(err, 'Failed to get delayed results!')
+      throw err
+    }
+  }
 }