Reduce usage producer retry counts; fix usage producer fallback logic (#6755)

Author: jdolle

.changeset/tender-flowers-refuse.md

@@ -0,0 +1,5 @@
+---
+'hive': patch
+---
+
+Correctly set usage service state to Ready after processing all of the fallback queue.

packages/services/usage/src/fallback-queue.ts

@@ -1,5 +1,6 @@
 import pLimit from 'p-limit';
 import type { ServiceLogger } from '@hive/service-common';
+import * as Sentry from '@sentry/node';
 
 // Average message size is ~800kb
 // 1000 messages = 800mb
@@ -63,6 +64,11 @@ export function createFallbackQueue(config: {
         queue.map(msgValue =>
           limit(() =>
             config.send(msgValue[0], msgValue[1]).catch(error => {
+              Sentry.setTags({
+                message: 'Failed to flush message before stopping',
+                numOfOperations: msgValue[1],
+              });
+              Sentry.captureException(error);
               config.logger.error(
                 {
                   error: error instanceof Error ? error.message : String(error),

packages/services/usage/src/usage.ts

@@ -35,7 +35,7 @@ const retryOptions = {
   initialRetryTime: 500,
   factor: 0.2,
   multiplier: 2,
-  retries: 10,
+  retries: 5,
 } satisfies RetryOptions; // why satisfies? To be able to use `retryOptions.retries` and get `number` instead of `number | undefined`
 
 export function splitReport(report: RawReport, numOfChunks: number) {
@@ -165,7 +165,7 @@ export function createUsage(config: {
       return Object.keys(report.map).length;
     },
     split(report, numOfChunks) {
-      logger.info('Splitting into %s', numOfChunks);
+      logger.info('Splitting report into %s (id=%s)', numOfChunks, report.id);
       return splitReport(report, numOfChunks);
     },
     onRetry(reports) {
@@ -212,21 +212,15 @@ export function createUsage(config: {
           rawOperationWrites.inc(numOfOperations);
           logger.info(`Flushed (id=%s, operations=%s)`, batchId, numOfOperations);
         }
-
-        changeStatus(Status.Ready);
       } catch (error: any) {
         rawOperationFailures.inc(numOfOperations);
 
         changeStatus(Status.Unhealthy);
-        logger.error(`Failed to flush (id=%s, error=%s)`, batchId, error.message);
-        Sentry.setTags({
+        logger.error(
+          `Failed to flush. Adding to fallback queue (id=%s, error=%s)`,
           batchId,
-          message: error.message,
-          numOfOperations,
-        });
-        Sentry.captureException(error);
-
-        logger.info('Adding to fallback queue (id=%s)', batchId);
+          error.message,
+        );
         fallback.add(value, numOfOperations);
 
         throw error;
@@ -255,6 +249,11 @@ export function createUsage(config: {
       } finally {
         stopTimer();
       }
+
+      if (fallback.size() === 0) {
+        logger.info('Fallback queue flushed');
+        changeStatus(Status.Ready);
+      }
     },
     logger: logger.child({ component: 'fallback' }),
   });
@@ -306,7 +305,7 @@ export function createUsage(config: {
       },
     ),
     readiness() {
-      return status === Status.Ready && fallback.size() === 0;
+      return status === Status.Ready;
     },
     async start() {
       logger.info('Starting Kafka producer');