Add memory logs and destroy stream on successful case (#78)

## Description

<!--
    A brief description of what the PR does/changes.
    Use active voice and present tense, e.g., This PR fixes ...
-->
This PR adds few more safety http stream destroys and adds memory
logging every 10 seconds in main thread.

## Connected Issues
- https://app.devrev.ai/devrev/works/ISS-214557
- https://app.devrev.ai/devrev/works/ISS-214092


## Checklist

- [x] Tests added/updated and ran with `npm run test` OR no tests
needed.
- [x] Ran backwards compatibility tests with `npm run
test:backwards-compatibility`.
- [x] Tested airdrop-template linked to this PR.
- [x] Documentation updated and provided a link to PR / new docs OR
`no-docs` written:
<!-- Add this once we have eslint prepared:
- [ ] Code formatted and checked with `npm run lint`. -->

Author: radovanjorgic

src/common/constants.ts

@@ -69,5 +69,6 @@ export const LIBRARY_VERSION = getLibraryVersion();
 
 export const DEFAULT_LAMBDA_TIMEOUT = 10 * 60 * 1000; // 10 minutes
 export const HARD_TIMEOUT_MULTIPLIER = 1.3;
+export const MEMORY_LOG_INTERVAL = 10 * 1000; // 10 seconds
 
 export const DEFAULT_SLEEP_DELAY_MS = 3 * 60 * 1000; // 3 minutes

src/common/helpers.ts

@@ -1,3 +1,7 @@
+import * as path from 'path';
+import { readFileSync } from 'fs';
+import * as v8 from 'v8';
+
 import {
   AirdropEvent,
   EventType,
@@ -10,9 +14,10 @@ import {
   LoaderReport,
   StatsFileObject,
 } from '../types/loading';
-import { readFileSync } from 'fs';
-import * as path from 'path';
-import { MAX_DEVREV_FILENAME_EXTENSION_LENGTH, MAX_DEVREV_FILENAME_LENGTH } from './constants';
+import {
+  MAX_DEVREV_FILENAME_EXTENSION_LENGTH,
+  MAX_DEVREV_FILENAME_LENGTH,
+} from './constants';
 
 export function getTimeoutErrorEventType(eventType: EventType): {
   eventType: ExtractorEventType | LoaderEventType;
@@ -210,13 +215,66 @@ export function truncateFilename(filename: string): string {
   console.warn(
     `Filename length exceeds the maximum limit of ${MAX_DEVREV_FILENAME_LENGTH} characters. Truncating filename.`
   );
-  
+
   let extension = filename.slice(-MAX_DEVREV_FILENAME_EXTENSION_LENGTH);
   // Calculate how many characters are available for the name part after accounting for the extension and "..."
-  const availableNameLength = MAX_DEVREV_FILENAME_LENGTH - MAX_DEVREV_FILENAME_EXTENSION_LENGTH - 3; // -3 for "..."
+  const availableNameLength =
+    MAX_DEVREV_FILENAME_LENGTH - MAX_DEVREV_FILENAME_EXTENSION_LENGTH - 3; // -3 for "..."
 
   // Truncate the name part and add an ellipsis
   const truncatedFilename = filename.slice(0, availableNameLength);
 
   return `${truncatedFilename}...${extension}`;
 }
+
+export interface MemoryInfo {
+  rssUsedMB: string;
+  rssUsedPercent: string; // Critical for OOM detection
+  heapUsedPercent: string; // GC pressure indicator
+  externalMB: string; // C++ objects and buffers (HTTP streams, etc.)
+  arrayBuffersMB: string; // Buffer data (unclosed streams show here)
+  formattedMessage: string;
+}
+
+export function getMemoryUsage(): MemoryInfo | null {
+  try {
+    const memUsage = process.memoryUsage();
+    const heapStats = v8.getHeapStatistics();
+
+    const rssUsedMB = memUsage.rss / 1024 / 1024;
+    const heapLimitMB = heapStats.heap_size_limit / 1024 / 1024;
+
+    const effectiveMemoryLimitMB = heapLimitMB;
+
+    // Calculate heap values for consistent format
+    const heapUsedMB = heapStats.used_heap_size / 1024 / 1024;
+    const heapTotalMB = heapStats.heap_size_limit / 1024 / 1024;
+
+    // Calculate external and buffer values (critical for detecting stream leaks)
+    const externalMB = memUsage.external / 1024 / 1024;
+    const arrayBuffersMB = memUsage.arrayBuffers / 1024 / 1024;
+
+    // Critical percentages for OOM detection
+    const rssUsedPercent =
+      ((rssUsedMB / effectiveMemoryLimitMB) * 100).toFixed(2) + '%';
+    const heapUsedPercent =
+      ((heapStats.used_heap_size / heapStats.heap_size_limit) * 100).toFixed(
+        2
+      ) + '%';
+
+    // Detailed message showing RSS breakdown for leak detection
+    const formattedMessage = `Memory: RSS ${rssUsedMB.toFixed(2)}/${effectiveMemoryLimitMB.toFixed(2)}MB (${rssUsedPercent}) [Heap ${heapUsedMB.toFixed(2)}/${heapTotalMB.toFixed(2)}MB (${heapUsedPercent}) + External ${externalMB.toFixed(2)}MB + Buffers ${arrayBuffersMB.toFixed(2)}MB].`;
+
+    return {
+      rssUsedMB: rssUsedMB.toFixed(2),
+      rssUsedPercent,
+      heapUsedPercent,
+      externalMB: externalMB.toFixed(2),
+      arrayBuffersMB: arrayBuffersMB.toFixed(2),
+      formattedMessage,
+    };
+  } catch (err) {
+    console.error('Error retrieving memory usage:', (err as Error).message);
+    return null;
+  }
+}

src/uploader/uploader.ts

@@ -184,6 +184,7 @@ export class Uploader {
         maxRedirects: 0, // Prevents buffering
         validateStatus: () => true, // Prevents errors on redirects
       });
+      this.destroyStream(fileStream);
       return response;
     } catch (error) {
       console.error('Error while streaming artifact.', serializeError(error));

src/workers/spawn.ts

@@ -8,7 +8,7 @@ import {
   ExtractorEventType,
 } from '../types/extraction';
 import { emit } from '../common/control-protocol';
-import { getTimeoutErrorEventType } from '../common/helpers';
+import { getTimeoutErrorEventType, getMemoryUsage } from '../common/helpers';
 import { Logger, serializeError } from '../logger/logger';
 import {
   GetWorkerPathInterface,
@@ -23,6 +23,7 @@ import { LogLevel } from '../logger/logger.interfaces';
 import {
   DEFAULT_LAMBDA_TIMEOUT,
   HARD_TIMEOUT_MULTIPLIER,
+  MEMORY_LOG_INTERVAL,
 } from '../common/constants';
 
 function getWorkerPath({
@@ -166,6 +167,7 @@ export class Spawn {
   private lambdaTimeout: number;
   private softTimeoutTimer: ReturnType<typeof setTimeout> | undefined;
   private hardTimeoutTimer: ReturnType<typeof setTimeout> | undefined;
+  private memoryMonitoringInterval: ReturnType<typeof setInterval> | undefined;
   private logger: Logger;
   private resolve: (value: void | PromiseLike<void>) => void;
 
@@ -228,6 +230,14 @@ export class Spawn {
         this.alreadyEmitted = true;
       }
     });
+
+    // Log memory usage every 10 seconds
+    this.memoryMonitoringInterval = setInterval(() => {
+      const memoryInfo = getMemoryUsage();
+      if (memoryInfo) {
+        this.logger.info(memoryInfo.formattedMessage);
+      }
+    }, MEMORY_LOG_INTERVAL);
   }
 
   private clearTimeouts(): void {
@@ -237,6 +247,9 @@ export class Spawn {
     if (this.hardTimeoutTimer) {
       clearTimeout(this.hardTimeoutTimer);
     }
+    if (this.memoryMonitoringInterval) {
+      clearInterval(this.memoryMonitoringInterval);
+    }
   }
 
   private async exitFromMainThread(): Promise<void> {

src/workers/worker-adapter.ts

@@ -727,6 +727,7 @@ export class WorkerAdapter<ConnectorState> {
         console.warn(
           `Error while streaming to artifact for attachment ID ${attachment.id}. Skipping attachment.`
         );
+        this.destroyHttpStream(httpStream);
         return;
       }