fix(grouper-metrics): (docs, log context, pushgateway options, duplicate-retry test

Kuchizu · Kuchizu · commit 650c4dae2710 · 2026-03-04T05:30:09.000+03:00
diff --git a/.env.sample b/.env.sample
@@ -32,9 +32,13 @@ PROMETHEUS_PUSHGATEWAY_URL=
 PROMETHEUS_PUSHGATEWAY_INTERVAL=10000
 
 # Grouper memory log controls
+# Number of handled tasks between memory checkpoint logs
 GROUPER_MEMORY_LOG_EVERY_TASKS=50
+# Number of handled tasks in one sustained-growth evaluation window
 GROUPER_MEMORY_GROWTH_WINDOW_TASKS=200
+# Sustained-growth warning threshold in megabytes (MB)
 GROUPER_MEMORY_GROWTH_WARN_MB=64
+# Single-handle growth warning threshold in megabytes (MB)
 GROUPER_MEMORY_HANDLE_GROWTH_WARN_MB=16
 
 # project token for error catching
diff --git a/lib/metrics.ts b/lib/metrics.ts
@@ -72,7 +72,7 @@ export function startMetricsPushing(workerName: string): () => void {
   const interval = getPushIntervalMs();
   const hostname = os.hostname();
   const id = nanoid(ID_SIZE);
-  const gateway = new client.Pushgateway(url, [], register);
+  const gateway = new client.Pushgateway(url, undefined, register);
 
   currentWorkerName = workerName;
 
diff --git a/workers/grouper/src/index.ts b/workers/grouper/src/index.ts
@@ -163,7 +163,7 @@ export default class GrouperWorker extends Worker {
       });
     } catch (error) {
       this.grouperMetrics.incrementErrorsTotal();
-      this.memoryMonitor.logHandleError(this.handledTasksCount, task.payload?.title);
+      this.memoryMonitor.logHandleError(this.handledTasksCount, task.payload?.title, task.projectId);
       throw error;
     }
   }
@@ -179,7 +179,7 @@ export default class GrouperWorker extends Worker {
     const memoryBeforeHandle = process.memoryUsage();
 
     this.grouperMetrics.observePayloadSize(taskPayloadSize);
-    this.memoryMonitor.logBeforeHandle(memoryBeforeHandle, handledTasksCount, taskPayloadSize);
+    this.memoryMonitor.logBeforeHandle(memoryBeforeHandle, handledTasksCount, taskPayloadSize, task.projectId);
 
     this.logger.info(`[handle] project=${task.projectId} catcher=${task.catcherType} title="${task.payload.title}" payloadSize=${taskPayloadSize}b backtraceFrames=${task.payload.backtrace?.length ?? 0}`);
 
@@ -208,6 +208,10 @@ export default class GrouperWorker extends Worker {
      */
     this.dataFilter.processEvent(task.payload);
 
+    /**
+     * Retry loop for duplicate-key races:
+     * another worker may save the first occurrence between `getEvent()` and `saveEvent()`.
+     */
     while (true) {
       uniqueEventHash = await this.getUniqueEventHash(task);
 
@@ -217,7 +221,7 @@ export default class GrouperWorker extends Worker {
       const similarEvent = await this.findSimilarEvent(task.projectId, task.payload.title);
 
       if (similarEvent) {
-        this.logger.info(`[handle] similar event found, groupHash=${similarEvent.groupHash} totalCount=${similarEvent.totalCount}`);
+        this.logger.info(`[handle] project=${task.projectId} title="${task.payload.title}" similar event found, groupHash=${similarEvent.groupHash} totalCount=${similarEvent.totalCount}`);
 
         /**
          * Override group hash with found event's group hash
@@ -227,7 +231,7 @@ export default class GrouperWorker extends Worker {
         existedEvent = similarEvent;
       } else {
         /**
-         * If we couldn't group by grouping pattern вЂ” try grouping by hash (title)
+         * If we couldn't group by grouping pattern - try grouping by hash (title).
          */
         /**
          * Find event by group hash.
@@ -244,7 +248,7 @@ export default class GrouperWorker extends Worker {
         try {
           const incrementAffectedUsers = !!task.payload.user;
 
-          this.logger.info(`[saveEvent] new event, payloadSize=${taskPayloadSize}b`);
+          this.logger.info(`[saveEvent] project=${task.projectId} title="${task.payload.title}" new event, payloadSize=${taskPayloadSize}b`);
 
           /**
            * Insert new event
@@ -277,7 +281,7 @@ export default class GrouperWorker extends Worker {
            */
           if (e.code?.toString() === DB_DUPLICATE_KEY_ERROR) {
             this.grouperMetrics.incrementDuplicateRetriesTotal();
-            this.logger.info(`[saveEvent] duplicate key, retrying as repetition`);
+            this.logger.info(`[saveEvent] project=${task.projectId} title="${task.payload.title}" duplicate key, retrying as repetition`);
 
             continue;
           }
@@ -306,7 +310,7 @@ export default class GrouperWorker extends Worker {
 
       const existedPayloadSize = Buffer.byteLength(JSON.stringify(existedEvent.payload));
 
-      this.logger.info(`[computeDelta] existedPayloadSize=${existedPayloadSize}b taskPayloadSize=${taskPayloadSize}b`);
+      this.logger.info(`[computeDelta] project=${task.projectId} title="${task.payload.title}" existedPayloadSize=${existedPayloadSize}b taskPayloadSize=${taskPayloadSize}b`);
 
       try {
         /**
@@ -323,7 +327,7 @@ export default class GrouperWorker extends Worker {
 
       this.grouperMetrics.observeDeltaSize(deltaSize);
 
-      this.logger.info(`[computeDelta] deltaSize=${deltaSize}b`);
+      this.logger.info(`[computeDelta] project=${task.projectId} title="${task.payload.title}" deltaSize=${deltaSize}b`);
 
       const newRepetition = {
         groupHash: uniqueEventHash,
diff --git a/workers/grouper/src/metrics/config.ts b/workers/grouper/src/metrics/config.ts
@@ -2,15 +2,45 @@
  * Parsed config for grouper memory monitoring.
  */
 export interface GrouperMemoryConfig {
+  /**
+   * Write periodic memory checkpoint every N handled tasks.
+   */
   logEveryTasks: number;
+
+  /**
+   * Number of handled tasks in one sustained-growth evaluation window.
+   */
   growthWindowTasks: number;
+
+  /**
+   * Warn when heap growth in the evaluation window is greater than this amount in MB.
+   */
   growthWarnMb: number;
+
+  /**
+   * Warn when a single handle() call grows heap by more than this amount in MB.
+   */
   handleGrowthWarnMb: number;
 }
 
+/**
+ * Default memory checkpoint interval in handled tasks.
+ */
 const DEFAULT_MEMORY_LOG_EVERY_TASKS = 50;
+
+/**
+ * Default sustained-growth window size in handled tasks.
+ */
 const DEFAULT_MEMORY_GROWTH_WINDOW_TASKS = 200;
+
+/**
+ * Default sustained-growth warning threshold in MB.
+ */
 const DEFAULT_MEMORY_GROWTH_WARN_MB = 64;
+
+/**
+ * Default single-handle growth warning threshold in MB.
+ */
 const DEFAULT_MEMORY_HANDLE_GROWTH_WARN_MB = 16;
 
 /**
diff --git a/workers/grouper/src/metrics/memoryMonitor.ts b/workers/grouper/src/metrics/memoryMonitor.ts
@@ -6,6 +6,10 @@ interface LoggerLike {
 }
 
 const ROUND_PRECISION = 100;
+
+/**
+ * Number of bytes in one mebibyte.
+ */
 // eslint-disable-next-line @typescript-eslint/no-magic-numbers
 const BYTES_IN_MEBIBYTE = 1024 * 1024;
 
@@ -15,7 +19,15 @@ const BYTES_IN_MEBIBYTE = 1024 * 1024;
 export default class GrouperMemoryMonitor {
   private readonly logger: LoggerLike;
   private readonly config: GrouperMemoryConfig;
+
+  /**
+   * Task number of the last sustained-growth checkpoint.
+   */
   private memoryCheckpointTask = 0;
+
+  /**
+   * Heap usage (bytes) saved at the last sustained-growth checkpoint.
+   */
   private memoryCheckpointHeapUsed = 0;
 
   /**
@@ -54,11 +66,12 @@ export default class GrouperMemoryMonitor {
    *
    * @param handledTasksCount - currently handled tasks count.
    * @param title - event title if available.
+   * @param projectId - project identifier.
    */
-  public logHandleError(handledTasksCount: number, title: string | undefined): void {
+  public logHandleError(handledTasksCount: number, title: string | undefined, projectId: string): void {
     const suffix = title ? `title="${title}"` : '';
 
-    this.logCheckpoint('handle-error', process.memoryUsage(), handledTasksCount, suffix);
+    this.logCheckpoint('handle-error', process.memoryUsage(), handledTasksCount, projectId, suffix);
   }
 
   /**
@@ -67,17 +80,18 @@ export default class GrouperMemoryMonitor {
    * @param memoryUsage - process memory usage.
    * @param handledTasksCount - currently handled tasks count.
    * @param payloadSizeBytes - task payload size.
+   * @param projectId - project identifier.
    */
-  public logBeforeHandle(memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number, payloadSizeBytes: number): void {
+  public logBeforeHandle(memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number, payloadSizeBytes: number, projectId: string): void {
     if (handledTasksCount !== 1 && handledTasksCount % this.config.logEveryTasks !== 0) {
       return;
     }
 
-    this.logCheckpoint('before-handle', memoryUsage, handledTasksCount, `payloadSize=${payloadSizeBytes}b`);
+    this.logCheckpoint('before-handle', memoryUsage, handledTasksCount, projectId, `payloadSize=${payloadSizeBytes}b`);
   }
 
   /**
-   * Log handle() completion memory details and growth checks.
+   * Log post-handle memory stats and emit warnings on suspicious growth.
    *
    * @param memoryBeforeHandle - memory usage before handling.
    * @param handledTasksCount - currently handled tasks count.
@@ -97,25 +111,26 @@ export default class GrouperMemoryMonitor {
     const heapDeltaMb = Math.round((heapDeltaBytes / BYTES_IN_MEBIBYTE) * ROUND_PRECISION) / ROUND_PRECISION;
 
     this.logger.info(
-      `[handle] done, ${this.formatMemoryUsage(memoryAfterHandle)} heapDelta=${heapDeltaMb}MB handled=${handledTasksCount}`
+      `[memory][project=${projectId}] [handle] done, ${this.formatMemoryUsage(memoryAfterHandle)} heapDelta=${heapDeltaMb}MB handled=${handledTasksCount}`
     );
 
     if (heapDeltaBytes > this.config.handleGrowthWarnMb * BYTES_IN_MEBIBYTE) {
       this.logger.warn(
-        `[memory] high heap growth in single handle: heapDelta=${heapDeltaMb}MB payloadSize=${payloadSizeBytes}b title="${title}" project=${projectId}`
+        `[memory][project=${projectId}] high heap growth in single handle: heapDelta=${heapDeltaMb}MB payloadSize=${payloadSizeBytes}b title="${title}"`
       );
     }
 
-    this.checkMemoryGrowthWindow(memoryAfterHandle, handledTasksCount);
+    this.checkMemoryGrowthWindow(memoryAfterHandle, handledTasksCount, projectId);
   }
 
   /**
    * Logs sustained heap growth over a configurable number of handled tasks.
    *
    * @param memoryUsage - current process memory usage.
    * @param handledTasksCount - currently handled tasks count.
+   * @param projectId - project identifier.
    */
-  private checkMemoryGrowthWindow(memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number): void {
+  private checkMemoryGrowthWindow(memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number, projectId: string): void {
     const tasksInWindow = handledTasksCount - this.memoryCheckpointTask;
 
     if (tasksInWindow < this.config.growthWindowTasks) {
@@ -127,12 +142,12 @@ export default class GrouperMemoryMonitor {
     const heapUsedNowMb = Math.round((memoryUsage.heapUsed / BYTES_IN_MEBIBYTE) * ROUND_PRECISION) / ROUND_PRECISION;
 
     this.logger.info(
-      `[memory] growth window tasks=${tasksInWindow} handled=${this.memoryCheckpointTask + 1}-${handledTasksCount} heapGrowth=${heapGrowthMb}MB heapUsedNow=${heapUsedNowMb}MB`
+      `[memory][project=${projectId}] growth window tasks=${tasksInWindow} handled=${this.memoryCheckpointTask + 1}-${handledTasksCount} heapGrowth=${heapGrowthMb}MB heapUsedNow=${heapUsedNowMb}MB`
     );
 
     if (heapGrowthBytes > this.config.growthWarnMb * BYTES_IN_MEBIBYTE) {
       this.logger.warn(
-        `[memory] possible leak detected: heap grew by ${heapGrowthMb}MB in ${tasksInWindow} handled tasks`
+        `[memory][project=${projectId}] possible leak detected: heap grew by ${heapGrowthMb}MB in ${tasksInWindow} handled tasks`
       );
     }
 
@@ -161,11 +176,13 @@ export default class GrouperMemoryMonitor {
    * @param stage - lifecycle stage.
    * @param memoryUsage - current process memory usage.
    * @param handledTasksCount - currently handled tasks count.
+   * @param projectId - optional project identifier.
    * @param suffix - optional extra suffix.
    */
-  private logCheckpoint(stage: string, memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number, suffix = ''): void {
+  private logCheckpoint(stage: string, memoryUsage: NodeJS.MemoryUsage, handledTasksCount: number, projectId?: string, suffix = ''): void {
     const extra = suffix ? ` ${suffix}` : '';
+    const prefix = projectId ? `[memory][project=${projectId}]` : '[memory]';
 
-    this.logger.info(`[memory] stage=${stage} handled=${handledTasksCount} ${this.formatMemoryUsage(memoryUsage)}${extra}`);
+    this.logger.info(`${prefix} stage=${stage} handled=${handledTasksCount} ${this.formatMemoryUsage(memoryUsage)}${extra}`);
   }
 }
diff --git a/workers/grouper/tests/index.test.ts b/workers/grouper/tests/index.test.ts
@@ -5,7 +5,7 @@ import type { RedisClientType } from 'redis';
 import { createClient } from 'redis';
 import type { Collection } from 'mongodb';
 import { MongoClient } from 'mongodb';
-import type { ErrorsCatcherType, EventAddons, EventData } from '@hawk.so/types';
+import type { ErrorsCatcherType, EventAddons, EventData, GroupedEventDBScheme } from '@hawk.so/types';
 import { MS_IN_SEC } from '../../../lib/utils/consts';
 import * as mongodb from 'mongodb';
 import { patch } from '@n1ru4l/json-patch-plus';
@@ -173,6 +173,47 @@ describe('GrouperWorker', () => {
       expect(await eventsCollection.find().count()).toBe(1);
     });
 
+    test('Should process event as repetition after duplicate key race', async () => {
+      const duplicateKeyErrorCode = 11000;
+      const task = generateTask();
+      const workerWithSaveEvent = worker as unknown as {
+        saveEvent: (
+          projectId: string,
+          groupedEventData: GroupedEventDBScheme
+        ) => Promise<mongodb.ObjectID>;
+      };
+      const originalSaveEvent = workerWithSaveEvent.saveEvent.bind(workerWithSaveEvent) as (
+        projectId: string,
+        groupedEventData: GroupedEventDBScheme
+      ) => Promise<mongodb.ObjectID>;
+      let shouldThrowDuplicate = true;
+      const saveEventSpy = jest.spyOn(workerWithSaveEvent, 'saveEvent');
+
+      saveEventSpy.mockImplementation(async (projectId: string, groupedEventData: GroupedEventDBScheme): Promise<mongodb.ObjectID> => {
+        if (shouldThrowDuplicate) {
+          shouldThrowDuplicate = false;
+
+          await originalSaveEvent(projectId, groupedEventData);
+
+          const duplicateError = new Error('duplicate key') as Error & { code: number };
+
+          duplicateError.code = duplicateKeyErrorCode;
+          throw duplicateError;
+        }
+
+        return originalSaveEvent(projectId, groupedEventData);
+      });
+
+      await worker.handle(task);
+
+      expect(saveEventSpy).toHaveBeenCalledTimes(1);
+      expect(await eventsCollection.find().count()).toBe(1);
+      expect(await repetitionsCollection.find().count()).toBe(1);
+      expect((await eventsCollection.findOne({})).totalCount).toBe(2);
+
+      saveEventSpy.mockRestore();
+    });
+
     test('Should increment total events count on each processing', async () => {
       await worker.handle(generateTask());
       await worker.handle(generateTask());
