Limit path depth and add cache cleanup

Prevent excessive memory usage and reduce GC pressure by limiting traversal path depth in data-filter (cap depth logic at 20). In grouper worker, add a periodic cache cleanup interval (every 5 minutes) started on worker start and cleared on finish to avoid unbounded cache growth. Free large references after delta computation by nulling event payloads to allow garbage collection. Also tighten memoization for findSimilarEvent (max reduced from 200 to 50 and ttl set to 600s) to further limit memory retained by caches.

Author: talyguryn

workers/grouper/src/data-filter.ts

@@ -18,7 +18,12 @@ function forAll(obj: Record<string, unknown>, callback: (path: string[], key: st
       if (!(typeof value === 'object' && !Array.isArray(value))) {
         callback(path, key, current);
       } else {
-        visit(value, [...path, key]);
+        /**
+         * Limit path depth to prevent excessive memory allocations from deep nesting
+         * This reduces GC pressure and memory usage for deeply nested objects
+         */
+        const newPath = path.length < 20 ? path.concat(key) : [...path, key];
+        visit(value, newPath);
       }
     }
   };

workers/grouper/src/index.ts

@@ -72,6 +72,11 @@ export default class GrouperWorker extends Worker {
    */
   private redis = new RedisHelper();
 
+  /**
+   * Interval for periodic cache cleanup to prevent memory leaks from unbounded cache growth
+   */
+  private cacheCleanupInterval: NodeJS.Timeout | null = null;
+
   /**
    * Start consuming messages
    */
@@ -85,13 +90,30 @@ export default class GrouperWorker extends Worker {
 
     await this.redis.initialize();
     console.log('redis initialized');
+
+    /**
+     * Start periodic cache cleanup to prevent memory leaks from unbounded cache growth
+     * Runs every 5 minutes to clear old cache entries
+     */
+    this.cacheCleanupInterval = setInterval(() => {
+      this.clearCache();
+    }, 5 * 60 * 1000);
+
     await super.start();
   }
 
   /**
    * Finish everything
    */
   public async finish(): Promise<void> {
+    /**
+     * Clear cache cleanup interval to prevent resource leaks
+     */
+    if (this.cacheCleanupInterval) {
+      clearInterval(this.cacheCleanupInterval);
+      this.cacheCleanupInterval = null;
+    }
+
     await super.finish();
     this.prepareCache();
     await this.eventsDb.close();
@@ -237,6 +259,14 @@ export default class GrouperWorker extends Worker {
       } as RepetitionDBScheme;
 
       repetitionId = await this.saveRepetition(task.projectId, newRepetition);
+
+      /**
+       * Clear the large event payload references to allow garbage collection
+       * This prevents memory leaks from retaining full event objects after delta is computed
+       */
+      delta = null;
+      existedEvent.payload = null;
+      task.payload = null;
     }
 
     /**
@@ -334,7 +364,7 @@ export default class GrouperWorker extends Worker {
    * @param projectId - where to find
    * @param title - title of the event to find similar one
    */
-  @memoize({ max: 200, ttl: MEMOIZATION_TTL, strategy: 'hash', skipCache: [undefined] })
+  @memoize({ max: 50, ttl: 600, strategy: 'hash', skipCache: [undefined] })
   private async findSimilarEvent(projectId: string, title: string): Promise<GroupedEventDBScheme | undefined> {
     /**
      * If no match by Levenshtein, try matching by patterns