add: email verification worker priority queueing, backpressure, heartbeat progress (#2577)

* refactor email verification handler

* add debug logs to rate limiter

* disable executeEvenly for rate-limiter

* add queue based email verification handler

* remove unused import & type any

* remove remainingTokens size from logs

Author: malek10xdev

backend/src/emailVerificationWorker.ts

@@ -23,10 +23,12 @@ const emailStatusVerifier = new EmailStatusVerifierFactory(ENV, logger);
 const emailStatusCache = new RedisEmailStatusCache(redisClient);
 const contacts = new PgContacts(pool, logger);
 
-const { processStreamData } = initializeEmailVerificationProcessor(
+const streamsHandler = initializeEmailVerificationProcessor(
   contacts,
   emailStatusCache,
-  emailStatusVerifier
+  emailStatusVerifier,
+  redisClient,
+  logger
 );
 
 const tasksManagementSubscriber = new RedisSubscriber<PubSubMessage>(
@@ -47,7 +49,7 @@ const emailsStreamConsumer = new EmailVerificationConsumer(
   tasksManagementSubscriber,
   emailStreamsConsumer,
   ENV.REDIS_EMAIL_VERIFICATION_CONSUMER_BATCH_SIZE,
-  processStreamData,
+  streamsHandler,
   redisClient,
   logger
 );

backend/src/services/email-status/EmailStatusVerifierFactory.ts

@@ -203,7 +203,7 @@ export default class EmailStatusVerifierFactory {
         }
       },
       new TokenBucketRateLimiter({
-        executeEvenly: true,
+        executeEvenly: false,
         uniqueKey: 'email-verification-reacher',
         distribution: Distribution.Memory,
         requests: config.REACHER_RATE_LIMITER_REQUESTS,
@@ -225,7 +225,7 @@ export default class EmailStatusVerifierFactory {
     const client = new MailerCheckClient(
       { apiToken: MAILERCHECK_API_KEY },
       new TokenBucketRateLimiter({
-        executeEvenly: true,
+        executeEvenly: false,
         uniqueKey: 'email-verification-mailercheck',
         distribution: Distribution.Memory,
         requests: 60,
@@ -247,14 +247,14 @@ export default class EmailStatusVerifierFactory {
     const client = new ZerobounceClient(
       { apiToken: ZEROBOUNCE_API_KEY },
       new TokenBucketRateLimiter({
-        executeEvenly: true,
+        executeEvenly: false,
         uniqueKey: 'email-verification-zerobounce-single',
         distribution: Distribution.Memory,
         requests: ZerobounceClient.SINGLE_VALIDATION_PER_10_SECONDS,
         intervalSeconds: 60
       }),
       new TokenBucketRateLimiter({
-        executeEvenly: true,
+        executeEvenly: false,
         uniqueKey: 'email-verification-zerobounce-bulk',
         distribution: Distribution.Memory,
         requests: ZerobounceClient.BATCH_VALIDATION_PER_MINUTE,

backend/src/services/rate-limiter/index.ts

@@ -6,6 +6,7 @@ import {
 } from 'rate-limiter-flexible';
 import redis from '../../utils/redis';
 import pool from '../../db/pg';
+import logger from '../../utils/logger';
 
 export interface IRateLimiter {
   throttleRequests<T>(callback: () => Promise<T>): Promise<T>;
@@ -30,13 +31,22 @@ export interface RateLimiterOptions {
 export class TokenBucketRateLimiter implements IRateLimiter {
   private readonly limiter: RateLimiterQueue;
 
+  private readonly options: RateLimiterOptions;
+
   constructor({
     intervalSeconds,
     requests,
     uniqueKey,
     executeEvenly,
     distribution
   }: RateLimiterOptions) {
+    this.options = {
+      intervalSeconds,
+      requests,
+      uniqueKey,
+      executeEvenly,
+      distribution
+    };
     const baseLimiter = TokenBucketRateLimiter.createLimiter({
       intervalSeconds,
       requests,
@@ -87,7 +97,20 @@ export class TokenBucketRateLimiter implements IRateLimiter {
    * Executes a request and consumes 1 tokens.
    */
   async throttleRequests<T>(callback: () => Promise<T>): Promise<T> {
+    logger.debug('Rate limiter: queueing request', {
+      instance: this.options.uniqueKey
+    });
+
+    const startTime = performance.now();
     await this.limiter.removeTokens(1);
+    const waitTimeMs = (performance.now() - startTime).toFixed(2);
+
+    logger.debug('Rate limiter: request granted', {
+      instance: this.options.uniqueKey,
+      metrics: {
+        waitTimeMs
+      }
+    });
     return callback();
   }
 

backend/src/workers/email-verification/EmailsVerificationConsumer.ts

@@ -2,7 +2,10 @@ import { Redis } from 'ioredis';
 import { Logger } from 'winston';
 import RedisSubscriber from '../../utils/pubsub/redis/RedisSubscriber';
 import MultipleStreamsConsumer from '../../utils/streams/MultipleStreamsConsumer';
-import { EmailVerificationData } from './emailVerificationHandlers';
+import {
+  EmailVerificationData,
+  EmailVerificationHandler
+} from './emailVerificationHandlers';
 
 export interface PubSubMessage {
   miningId: string;
@@ -20,10 +23,7 @@ export default class EmailVerificationConsumer {
     private readonly taskManagementSubscriber: RedisSubscriber<PubSubMessage>,
     private readonly emailStreamsConsumer: MultipleStreamsConsumer<EmailVerificationData>,
     private readonly batchSize: number,
-    private readonly emailProcessor: (
-      data: EmailVerificationData[],
-      progressCallback: (verified: number) => Promise<void>
-    ) => Promise<void>,
+    private readonly streamsHandler: EmailVerificationHandler,
     private readonly redisClient: Redis,
     private readonly logger: Logger
   ) {
@@ -34,8 +34,10 @@ export default class EmailVerificationConsumer {
         if (emailsStream) {
           if (command === 'REGISTER') {
             this.activeStreams.add(emailsStream);
+            this.streamsHandler.registerStream(emailsStream);
           } else {
             this.activeStreams.delete(emailsStream);
+            this.streamsHandler.unregisterStream(emailsStream);
           }
         }
 
@@ -68,17 +70,11 @@ export default class EmailVerificationConsumer {
             this.logger.info(
               `Consuming ${data.length} emails from stream name ${streamName}`
             );
-            const progressCallback = async (verified: number) => {
-              await this.redisClient.publish(
-                streamName.split('-')[1],
-                JSON.stringify({
-                  miningId: streamName.split('-')[1],
-                  progressType: 'verifiedContacts',
-                  count: verified
-                })
-              );
-            };
-            const processed = await this.emailProcessor(data, progressCallback);
+
+            const processed = await this.streamsHandler.handle(
+              streamName,
+              data
+            );
 
             if (!this.activeStreams.has(streamName)) {
               return null;
@@ -121,7 +117,7 @@ export default class EmailVerificationConsumer {
 
     setTimeout(() => {
       this.consume();
-    }, 3000);
+    }, 10000);
   }
 
   /**