Fix: memory leak with when using Http(s)Agent keepAlive=true (#391)

* create spec to reproduce memory leak

* fix memory leak using keep-alive in nodejs

* fix lint on spec support file

* refactor test spec to improve readability

* use explicit callback assignments instead of .bind

* add spec for onRequestSocketAssigned calls

* add changeset

.changeset/friendly-bottles-unite.md

@@ -0,0 +1,5 @@
+---
+'mappersmith': minor
+---
+
+Fixes memory leak when using http(s) agent with keep-alive

spec/integration/node/integration.spec.js

@@ -1,3 +1,5 @@
+import { Agent as HttpAgent } from 'http'
+
 import 'core-js/stable'
 import 'regenerator-runtime/runtime'
 import md5 from 'js-md5'
@@ -8,10 +10,13 @@ import forge, { configs } from 'src/index'
 import createManifest from 'spec/integration/support/manifest'
 import { errorMessage, INVALID_ADDRESS } from 'spec/integration/support'
 
+import keepAlive from './support/keep-alive'
+
 describe('integration', () => {
   describe('HTTP', () => {
     const gateway = HTTP
     const params = { host: 'http://localhost:9090' }
+    const keepAliveHelper = keepAlive(params.host, gateway)
 
     integrationTestsForGateway(gateway, params)
 
@@ -51,11 +56,11 @@ describe('integration', () => {
       beforeEach(() => {
         gatewayConfigs = {
           onRequestWillStart: jasmine.createSpy('onRequestWillStart'),
-          onRequestSocketAssigned: jasmine.createSpy('onRequestWillStart'),
-          onSocketLookup: jasmine.createSpy('onRequestWillStart'),
-          onSocketConnect: jasmine.createSpy('onRequestWillStart'),
-          onResponseReadable: jasmine.createSpy('onRequestWillStart'),
-          onResponseEnd: jasmine.createSpy('onRequestWillStart'),
+          onRequestSocketAssigned: jasmine.createSpy('onRequestSocketAssigned'),
+          onSocketLookup: jasmine.createSpy('onSocketLookup'),
+          onSocketConnect: jasmine.createSpy('onSocketConnect'),
+          onResponseReadable: jasmine.createSpy('onResponseReadable'),
+          onResponseEnd: jasmine.createSpy('onResponseEnd'),
         }
 
         configs.gatewayConfigs.HTTP = gatewayConfigs
@@ -73,6 +78,62 @@ describe('integration', () => {
           done()
         })
       })
+
+      describe('without keep alive', () => {
+        let httpAgent
+
+        beforeEach(() => {
+          httpAgent = new HttpAgent({ keepAlive: false })
+          configs.gatewayConfigs.HTTP.configure = () => ({ agent: httpAgent })
+        })
+
+        it('does not reuse the socket and only attaches listeners once to the http agent sockets', (done) => {
+          keepAliveHelper.callApiTwice().then(() => {
+            keepAliveHelper.verifySockets(1, httpAgent.sockets)
+            keepAliveHelper.verifySockets(0, httpAgent.freeSockets)
+            done()
+          }).catch((response) => {
+            done.fail(`test failed with promise error: ${errorMessage(response)}`)
+          })
+        })
+
+        it('calls the `onRequestSocketAssigned` callback on every request', (done) => {
+          keepAliveHelper.callApiTwice().then(() => {
+            expect(gatewayConfigs.onRequestSocketAssigned).toHaveBeenCalledTimes(2)
+            done()
+          }).catch((response) => {
+            done.fail(`test failed with promise error: ${errorMessage(response)}`)
+          })
+        })
+      })
+
+      describe('with keep alive', () => {
+        let httpAgent
+
+        beforeEach(() => {
+          httpAgent = new HttpAgent({ keepAlive: true })
+          configs.gatewayConfigs.HTTP.configure = () => ({ agent: httpAgent })
+        })
+
+        it('reuses the socket, and only attaches listeners once to the reused socket', (done) => {
+          keepAliveHelper.callApiTwice().then(() => {
+            keepAliveHelper.verifySockets(0, httpAgent.sockets)
+            keepAliveHelper.verifySockets(1, httpAgent.freeSockets)
+            done()
+          }).catch((response) => {
+            done.fail(`test failed with promise error: ${errorMessage(response)}`)
+          })
+        })
+
+        it('calls the `onRequestSocketAssigned` callback on every request', (done) => {
+          keepAliveHelper.callApiTwice().then(() => {
+            expect(gatewayConfigs.onRequestSocketAssigned).toHaveBeenCalledTimes(2)
+            done()
+          }).catch((response) => {
+            done.fail(`test failed with promise error: ${errorMessage(response)}`)
+          })
+        })
+      })
     })
   })
 })

spec/integration/node/support/keep-alive.js

@@ -0,0 +1,23 @@
+import createManifest from 'spec/integration/support/manifest'
+
+import forge from 'src/index'
+
+export default function keepAlive(host, gateway) {
+  return {
+    verifySockets: (count, socketsReference) => {
+      const socketOrigins = Object.keys(socketsReference)
+      expect(socketOrigins.length).toBe(count)
+      if (count === 0) return
+
+      const sockets = socketsReference[socketOrigins[0]]
+      expect(sockets.length).toBe(count)
+      const socket = sockets[0]
+      expect(socket.listenerCount('lookup')).toBe(count)
+      expect(socket.listenerCount('connect')).toBe(count)
+    },
+    callApiTwice: () => {
+      const Client = forge(createManifest(host), gateway)
+      return Client.Book.all().then(() => Client.Book.all())
+    }
+  }
+}

src/gateway/http.ts

@@ -96,21 +96,25 @@ export class HTTP extends Gateway {
         httpOptions.onRequestSocketAssigned(requestParams)
       }
 
-      socket.on('lookup', () => {
-        if (httpOptions.onSocketLookup) {
-          httpOptions.onSocketLookup(requestParams)
-        }
-      })
-      socket.on('connect', () => {
-        if (httpOptions.onSocketConnect) {
-          httpOptions.onSocketConnect(requestParams)
-        }
-      })
-      socket.on('secureConnect', () => {
-        if (httpOptions.onSocketSecureConnect) {
-          httpOptions.onSocketSecureConnect(requestParams)
-        }
-      })
+      if (httpRequest.reusedSocket) {
+        return
+      }
+
+      if (httpOptions.onSocketLookup) {
+        socket.on('lookup', () => {
+          httpOptions.onSocketLookup?.(requestParams)
+        })
+      }
+      if (httpOptions.onSocketConnect) {
+        socket.on('connect', () => {
+          httpOptions.onSocketConnect?.(requestParams)
+        })
+      }
+      if (httpOptions.onSocketSecureConnect) {
+        socket.on('secureConnect', () => {
+          httpOptions.onSocketSecureConnect?.(requestParams)
+        })
+      }
     })
 
     httpRequest.on('error', (e) => this.onError(e))