Merge pull request #236 from streamr-dev/memleak2

NET-229: Fix Memleak(s)

Author: timoxley

.github/workflows/test-build.yml

@@ -31,6 +31,13 @@ on:
 # Be sure to update both workflow files if you edit any env or trigger config
 
 jobs:
+  init:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Cancel Previous Runs
+        uses: styfle/cancel-workflow-action@0.8.0
+        with:
+          access_token: ${{ github.token }}
   build:
     name: Run build using Node 14.x
     runs-on: ubuntu-latest
@@ -188,6 +195,6 @@ jobs:
         name: run-client-testing
         with:
           max_attempts: 2
-          timeout_minutes: 8
+          timeout_minutes: 15
           retry_on: error
-          command: cd streamr-client-testing && java -jar build/libs/client_testing-1.0-SNAPSHOT.jar -s $TEST_NAME -c config/$CONFIG_NAME.conf -n $NUM_MESSAGES
+          command: cd streamr-client-testing && DEBUG='' java -jar build/libs/client_testing-1.0-SNAPSHOT.jar -s $TEST_NAME -c config/$CONFIG_NAME.conf -n $NUM_MESSAGES

.github/workflows/test-code.yml

@@ -31,6 +31,13 @@ on:
 # Be sure to update both workflow files if you edit any env or trigger config
 
 jobs:
+  init:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Cancel Previous Runs
+        uses: styfle/cancel-workflow-action@0.8.0
+        with:
+          access_token: ${{ github.token }}
   lint:
     name: Run linter using Node 14.x
     runs-on: ubuntu-latest
@@ -133,3 +140,25 @@ jobs:
           retry_on: error
           command: npm run test-flakey || echo "::warning::Flakey Tests Failed"
 
+  memory:
+    name:  Memory Tests using Node ${{ matrix.node-version }}
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        node-version: [12.x, 14.x]
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-node@v2
+        with:
+          node-version: ${{ matrix.node-version }}
+      - name: npm ci
+        run: npm ci
+      - uses: nick-invision/retry@v2
+        name: Run Test
+        with:
+          max_attempts: 2
+          timeout_minutes: 20
+          retry_on: error
+          command: npm run test-memory

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "streamr-client",
-  "version": "5.1.0",
+  "version": "5.2.0-beta",
   "description": "JavaScript client library for Streamr",
   "private": true,
   "repository": {
@@ -48,6 +48,7 @@
     "test-integration-no-resend": "jest --forceExit --testTimeout=10000 --testPathIgnorePatterns='resend|Resend' --testNamePattern='^((?!(resend|Resend|resent|Resent|gap|Gap)).)*$' test/integration/*.test.*",
     "test-integration-resend": "jest --forceExit --testTimeout=15000 --testNamePattern='(resend|Resend|resent|Resent)' test/integration/*.test.*",
     "test-integration-dataunions": "jest --forceExit --testTimeout=15000 --runInBand test/integration/dataunion",
+    "test-memory": "node --gc-global --predictable-gc-schedule node_modules/.bin/jest test/memory",
     "test-flakey": "jest --forceExit test/flakey/*",
     "test-browser": "node ./test/browser/server.js & node node_modules/nightwatch/bin/nightwatch ./test/browser/browser.js && pkill -f server.js",
     "install-example": "cd examples/webpack && npm ci",
@@ -113,6 +114,7 @@
     "jest-circus": "^26.6.3",
     "lodash-webpack-plugin": "^0.11.6",
     "nightwatch": "^1.5.1",
+    "pretty-bytes": "^5.6.0",
     "process": "^0.11.10",
     "sinon": "^9.2.4",
     "streamr-test-utils": "^1.3.1",

package.json

@@ -41,7 +41,10 @@ export default function Validator(client, opts) {
         },
         verify: CacheAsyncFn(SigningUtil.verify.bind(SigningUtil), {
             ...client.options.cache,
-            cachePromiseRejection: false,
+            // forcibly use small cache otherwise keeps n serialized messages in memory
+            maxSize: 100,
+            maxAge: 10000,
+            cachePromiseRejection: true,
             cacheKey: (args) => args.join('|'),
         })
     })

src/subscribe/Validator.js

@@ -1,4 +1,4 @@
-import { pOrderedResolve, Defer, pTimeout } from './index'
+import { Defer, pTimeout } from './index'
 
 async function endGenerator(gtr: AsyncGenerator, error?: Error) {
     return error
@@ -127,20 +127,11 @@ export default class PushQueue<T> {
     }
 
     static transform<TT, U>(src: AnyIterable<TT>, fn: (value: TT) => U, opts = {}) {
-        const buffer = new PushQueue<TT>([], opts)
-        const orderedFn = pOrderedResolve(fn) // push must be run in sequence
+        const buffer = new PushQueue<U>([], opts)
         ;(async () => { // eslint-disable-line semi-style
-            const tasks = []
             for await (const value of src) {
-                // run in parallel
-                const task = orderedFn(value).then(() => (
-                    buffer.push(value)
-                )).catch((err) => {
-                    buffer.throw(err)
-                })
-                tasks.push(task)
+                buffer.push(fn(value))
             }
-            await Promise.all(tasks)
             if (buffer.autoEnd) {
                 buffer.end()
             }
@@ -169,7 +160,7 @@ export default class PushQueue<T> {
         }
 
         if (end) {
-            await this.end()
+            this.end()
         }
 
         return Promise.resolve()
@@ -235,8 +226,6 @@ export default class PushQueue<T> {
             // for next()
             this.error = err
         }
-
-        await this._cleanup()
     }
 
     get length() {
@@ -245,18 +234,30 @@ export default class PushQueue<T> {
     }
 
     async _cleanup() {
+        // capture error and pending next promises
+        const { error, nextQueue } = this
         this.finished = true
-        const { error } = this
-        for (const p of this.nextQueue) {
+        this.error = undefined
+        this.pending = 0
+        // empty buffer then reassign
+        this.buffer.length = 0
+        this.buffer = []
+        // reassign nextQueue, emptying would mutate value we captured
+        this.nextQueue = []
+        const doneValue = { value: undefined, done: true }
+        // resolve all pending next promises
+        while (nextQueue.length) {
+            const p = nextQueue.shift()
+            if (!p) { continue }
+
             if (error) {
                 p.reject(error)
             } else {
-                p.resolve(undefined)
+                p.resolve(doneValue)
             }
         }
-        this.pending = 0
-        this.buffer.length = 0
-        return this.onEnd(this.error)
+
+        return this.onEnd(error)
     }
 
     push(...values: (T | null)[]) {
@@ -340,7 +341,6 @@ export default class PushQueue<T> {
 
                 const deferred = Defer<T>()
                 this.nextQueue.push(deferred)
-
                 deferred.catch(() => {}) // prevent unhandledrejection
                 const value = await deferred
 
@@ -389,6 +389,7 @@ export default class PushQueue<T> {
         try {
             yield* this.iterator
         } finally {
+            this._cleanup()
             this.finished = true
             if (this.signal) {
                 this.signal.removeEventListener('abort', this.onAbort)

src/utils/PushQueue.ts

@@ -220,13 +220,32 @@ export function CacheFn(fn: Parameters<typeof mem>[0], {
 type PromiseResolve = L.Compulsory<Parameters<Promise<any>['then']>>[0]
 type PromiseReject = L.Compulsory<Parameters<Promise<any>['then']>>[1]
 
-export function Defer<T>(executor: (...args: Parameters<Promise<T>['then']>) => void = () => {}) {
-    let resolve: PromiseResolve = () => {}
-    let reject: PromiseReject = () => {}
+const noop = () => {}
+
+export function Defer<T>(executor: (...args: Parameters<Promise<T>['then']>) => void = noop) {
+    let resolveFn: PromiseResolve | undefined
+    let rejectFn: PromiseResolve | undefined
+    const resolve: PromiseReject = (value) => {
+        if (resolveFn) {
+            const r = resolveFn
+            resolveFn = undefined
+            rejectFn = undefined
+            r(value)
+        }
+    }
+    const reject: PromiseReject = (error) => {
+        if (rejectFn) {
+            const r = rejectFn
+            resolveFn = undefined
+            rejectFn = undefined
+            r(error)
+        }
+    }
+
     // eslint-disable-next-line promise/param-names
     const p = new Promise((_resolve, _reject) => {
-        resolve = _resolve
-        reject = _reject
+        resolveFn = _resolve
+        rejectFn = _reject
         executor(resolve, reject)
     })
     p.catch(() => {}) // prevent unhandledrejection
@@ -295,21 +314,26 @@ export function LimitAsyncFnByKey<KeyType>(limit = 1) {
                     // clean up if no more active entries (if not cleared)
                     pending.delete(id)
                 }
-                queueOnEmptyTasks.delete(id)
+
+                if (queueOnEmptyTasks.has(id)) {
+                    queueOnEmptyTasks.get(id).resolve(undefined)
+                    queueOnEmptyTasks.delete(id)
+                }
+
                 onQueueEmpty.resolve()
             }
         }
     }
 
     f.getOnQueueEmpty = async (id: KeyType) => {
-        return queueOnEmptyTasks.get(id) || pending.set(id, Defer()).get(id)
+        return queueOnEmptyTasks.get(id) || queueOnEmptyTasks.set(id, Defer()).get(id)
     }
 
     f.clear = () => {
         // note: does not cancel promises
         pending.forEach((p) => p.clearQueue())
         pending.clear()
-        queueOnEmptyTasks.forEach((p) => p.resolve())
+        queueOnEmptyTasks.forEach((p) => p.resolve(undefined))
         queueOnEmptyTasks.clear()
     }
     return f