memory leak fix (#104)

Author: if0s

.circleci/config.yml

@@ -95,7 +95,7 @@ jobs:
       - node/install:
           node-version: << pipeline.parameters.node-version >>
       - setup_remote_docker:
-          version: 19.03.13
+          version: default
           docker_layer_caching: true
       # build and push Docker image
       - run:

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 1.5.1 (October 16, 2024)
+* Fixed memory leak in `Re-assembled message` action, when you have more than 1 000 000 messages
+
 ## 1.5.0 (May 29, 2024)
 * Revitalized `Re-assembled message` action - now all messages are stored in memory within the component and saved to the external storage (Maester) only when needed
 * Added new checkbox `Emit result as array` to `Re-assembled message` action

component.json

@@ -1,7 +1,7 @@
 {
   "title": "Splitter",
   "description": "Splits a message into multiple messages.",
-  "version": "1.5.0",
+  "version": "1.5.1",
   "actions": {
     "split": {
       "deprecated": true,

lib/utils.js

@@ -19,6 +19,16 @@ const getUserAgent = () => {
   return `${compName}/${compVersion} maester-client/${maesterClientVersion}`;
 };
 
+const logMemoryUsage = (logger) => {
+  const format = (num) => `${Math.round(num / 1024 / 1024)} MB`;
+  const {
+    rss,
+    heapTotal,
+    heapUsed,
+  } = process.memoryUsage();
+  logger.debug(`Memory used - rss ${format(rss)}, heapTotal ${format(heapTotal)}, heapUsed ${format(heapUsed)}`);
+};
+
 class GroupsProcessor {
   constructor(context, cfg) {
     this.cfg = cfg;
@@ -69,6 +79,8 @@ class GroupsProcessor {
     this.groupsStorage[groupUniqueId] ||= {};
     const group = this.groupsStorage[groupUniqueId];
     group.messages ||= [];
+    const msgSize = JSON.stringify(messageData || {}).length;
+    group.groupDataSize = group.groupDataSize ? group.groupDataSize + msgSize : msgSize;
     group.messages.push({ [messageId]: messageData || {} });
     group.readyAfter = timestamp(new Date()) + timersec;
     group.groupSize = groupSize;
@@ -86,11 +98,11 @@ class GroupsProcessor {
     const group = this.groupsStorage[groupUniqueId];
     const readyBySize = this.groupReadyBySize(group);
     const readyByTime = this.cfg.mode === 'groupSize' ? false : this.groupReadyByTime(group);
-    const readyByOverload = JSON.stringify(this.groupsStorage).length > MAX_LOCAL_STORAGE_SIZE;
+    const readyByOverload = group.groupDataSize > MAX_LOCAL_STORAGE_SIZE;
     if (readyBySize || readyByTime || readyByOverload) {
       const groupOriginalId = getOriginalGroupId(groupUniqueId);
       if (readyByOverload) {
-        this.logger.warn(`Local storage is overloaded group "${groupOriginalId}" will be emitted to prevent data loss`);
+        this.logger.warn(`Local storage is overloaded group "${groupOriginalId}" will be emitted to prevent data loss (${Math.round(group.groupDataSize / 1024 / 1024)} MB used)`);
       } else {
         this.logger.info(`Group "${groupOriginalId}" is ready by ${readyBySize ? 'Size' : 'Time'}`);
       }
@@ -99,7 +111,8 @@ class GroupsProcessor {
       delete group.messages;
       delete group.readyAfter;
       delete group.maesterId;
-
+      delete group.groupDataSize;
+      logMemoryUsage(this.logger);
       await this.emitGroup(fixedGroup, groupOriginalId);
       this.logger.info(`Message with groupId: ${groupOriginalId} was emitted`);
       if (fixedGroup.maesterId) {
@@ -161,7 +174,7 @@ class GroupsProcessor {
       this.logger.info('All groups saved to maester, stopping status checker');
       this.statusCheckerRunning = false;
     } else {
-      await sleep(10);
+      await sleep(100);
       await this.statusChecker();
     }
   }