Use errorMonitor symbol instead of listening to the error event

This is important to guarantee user code behaves as it did without
the instrumentation. So user crashes should happen as before.

Author: BridgeAR

packages/datadog-instrumentations/src/child_process.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('node:events')
 const util = require('util')
 
 const {
@@ -227,7 +228,7 @@ function wrapChildProcessAsyncMethod (ChildProcess, shell = false) {
         if (childProcess) {
           let errorExecuted = false
 
-          childProcess.on('error', (e) => {
+          childProcess.on(errorMonitor, (e) => {
             errorExecuted = true
             childProcessChannel.error.publish(e)
           })

packages/datadog-instrumentations/src/couchbase.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('events')
 const {
   channel,
   addHook,
@@ -186,7 +187,7 @@ addHook({ name: 'couchbase', file: 'lib/bucket.js', versions: ['^2.6.12'] }, Buc
         finishCh.publish(undefined)
       }))
 
-      emitter.once('error', asyncResource.bind((error) => {
+      emitter.once(errorMonitor, asyncResource.bind((error) => {
         errorCh.publish(error)
         finishCh.publish(undefined)
       }))

packages/datadog-instrumentations/src/fs.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const { errorMonitor } = require('events')
 const { channel, addHook } = require('./helpers/instrument')
 const shimmer = require('../../datadog-shimmer')
 
@@ -199,16 +200,16 @@ function wrapCreateStream (original) {
         }
         const onFinish = () => {
           finishChannel.runStores(ctx, () => {})
-          stream.off('close', onFinish)
-          stream.off('end', onFinish)
-          stream.off('finish', onFinish)
-          stream.off('error', onError)
+          stream.removeListener('close', onFinish)
+          stream.removeListener('end', onFinish)
+          stream.removeListener('finish', onFinish)
+          stream.removeListener(errorMonitor, onError)
         }
 
         stream.once('close', onFinish)
         stream.once('end', onFinish)
         stream.once('finish', onFinish)
-        stream.once('error', onError)
+        stream.once(errorMonitor, onError)
 
         return stream
       } catch (error) {

packages/datadog-instrumentations/src/http/client.js

@@ -3,6 +3,7 @@
 /* eslint-disable no-fallthrough */
 
 const url = require('url')
+const { errorMonitor } = require('node:events')
 const { channel, addHook } = require('../helpers/instrument')
 const shimmer = require('../../../datadog-shimmer')
 
@@ -88,7 +89,7 @@ function patch (http, methodName) {
                 const res = arg
                 ctx.res = res
                 res.on('end', finish)
-                res.on('error', finish)
+                res.on(errorMonitor, finish)
                 break
               }
               case 'connect':

packages/datadog-instrumentations/src/mysql2.js

@@ -1,5 +1,7 @@
 'use strict'
 
+const { errorMonitor } = require('node:events')
+
 const {
   channel,
   addHook,
@@ -138,7 +140,7 @@ function wrapConnection (Connection, version) {
           onResult.apply(this, arguments)
         }, 'bound-anonymous-fn', this))
       } else {
-        this.on('error', asyncResource.bind(error => errorCh.publish(error)))
+        this.on(errorMonitor, asyncResource.bind(error => errorCh.publish(error)))
         this.on('end', asyncResource.bind(() => finishCh.publish(undefined)))
       }
 

packages/datadog-instrumentations/src/net.js

@@ -1,5 +1,7 @@
 'use strict'
 
+const { errorMonitor } = require('events')
+
 const { channel, addHook } = require('./helpers/instrument')
 const shimmer = require('../../datadog-shimmer')
 
@@ -99,7 +101,7 @@ function getOptions (args) {
 }
 
 function setupListeners (socket, protocol, ctx, finishCh, errorCh) {
-  const events = ['connect', 'error', 'close', 'timeout']
+  const events = ['connect', errorMonitor, 'close', 'timeout']
 
   const wrapListener = function (error) {
     if (error) {

packages/datadog-instrumentations/src/pg.js

@@ -129,21 +129,12 @@ function wrapQuery (query) {
         }
       } else if (newQuery.once) {
         newQuery
-          .once('error', finish)
+          .once(errorMonitor, finish)
           .once('end', (res) => finish(null, res))
       } else {
         newQuery.then((res) => finish(null, res), finish)
       }
 
-      if (stream) {
-        newQuery.on('end', () => {
-          finish(null, [])
-        })
-        newQuery.on(errorMonitor, (err) => {
-          finish(err)
-        })
-      }
-
       try {
         return retval
       } catch (err) {

packages/datadog-instrumentations/test/child_process.spec.js

@@ -102,6 +102,10 @@ describe('child process', () => {
               it('should execute error callback', (done) => {
                 const childEmitter = childProcess[methodName]('invalid_command_test')
 
+                expect(childEmitter.listenerCount('error')).to.equal(methodName.includes('spawn') ? 0 : 1)
+
+                childEmitter.once('error', () => {})
+
                 childEmitter.once('close', () => {
                   expect(start).to.have.been.calledOnceWith({
                     command: 'invalid_command_test',

packages/datadog-instrumentations/test/mysql2.spec.js

@@ -4,6 +4,7 @@ const { channel } = require('../src/helpers/instrument')
 const agent = require('../../dd-trace/test/plugins/agent')
 const { assert } = require('chai')
 const semver = require('semver')
+const { once } = require('events')
 
 describe('mysql2 instrumentation', () => {
   withVersions('mysql2', 'mysql2', version => {
@@ -435,16 +436,18 @@ describe('mysql2 instrumentation', () => {
               query.on('end', () => done())
             })
 
-            it('should work without abortController.abort()', (done) => {
+            it('should work without abortController.abort()', async () => {
               startCh.subscribe(noop)
               const query = pool.query(sql)
 
-              query.on('error', err => done(err))
-              query.on('end', () => {
-                sinon.assert.called(apmQueryStart)
+              let error
+              query.on('error', err => { error = err })
+              expect(query.listenerCount('error')).to.equal(1)
 
-                done()
-              })
+              await once(query, 'end')
+              expect(query.listenerCount('error')).to.equal(1)
+              sinon.assert.called(apmQueryStart)
+              expect(error).to.be.undefined
             })
 
             it('should work without subscriptions', (done) => {

packages/datadog-plugin-fs/test/index.spec.js

@@ -675,7 +675,10 @@ describe('Plugin', () => {
               'file.flag': 'r'
             }
           })
-          fs.createReadStream(__filename).on('error', done).resume()
+          const stream = fs.createReadStream(__filename)
+          expect(stream.listenerCount('error')).to.equal(0)
+
+          stream.on('error', done).resume()
         })
 
         it('should be instrumented when closed', (done) => {