[DI] Refactor mutex code guarding race condition (#5728)

When adding, removing or modifying a breakpoint there can be a race
condtion if two Remove Config configs are received in quick succession.
We already have a mutex to guard against this, but the code was in need
of a refactor because:

1. We had duplicate mutex code in two different files, where only one
   was needed
2. We couldn't easily test the main mutex code in a unit test as it was
   added in the calling file and therefore not isolated to the function
   containing the race condition.

Author: watson

packages/dd-trace/src/debugger/devtools_client/breakpoints.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const lock = require('mutexify/promise')()
+const mutex = require('mutexify/promise')()
 const { getGeneratedPosition } = require('./source-maps')
 const session = require('./session')
 const { compile: compileCondition, compileSegments, templateRequiresEvaluation } = require('./condition')
@@ -36,8 +36,9 @@ session.on('scriptLoadingStabilized', () => {
 })
 
 module.exports = {
-  addBreakpoint,
-  removeBreakpoint
+  addBreakpoint: lock(addBreakpoint),
+  removeBreakpoint: lock(removeBreakpoint),
+  modifyBreakpoint: lock(modifyBreakpoint)
 }
 
 async function addBreakpoint (probe) {
@@ -89,42 +90,36 @@ async function addBreakpoint (probe) {
     throw new Error(`Cannot compile expression: ${probe.when.dsl}`, { cause: err })
   }
 
-  const release = await lock()
-
-  try {
-    const locationKey = generateLocationKey(scriptId, lineNumber, columnNumber)
-    const breakpoint = locationToBreakpoint.get(locationKey)
-
-    log.debug(
-      '[debugger:devtools_client] %s breakpoint at %s:%d:%d (probe: %s, version: %d)',
-      breakpoint ? 'Updating' : 'Adding', url, lineNumber, columnNumber, probe.id, probe.version
-    )
-
-    if (breakpoint) {
-      // A breakpoint already exists at this location, so we need to add the probe to the existing breakpoint
-      await updateBreakpoint(breakpoint, probe)
-    } else {
-      // No breakpoint exists at this location, so we need to create a new one
-      const location = {
-        scriptId,
-        lineNumber: lineNumber - 1, // Beware! lineNumber is zero-indexed
-        columnNumber
-      }
-      let result
-      try {
-        result = await session.post('Debugger.setBreakpoint', {
-          location,
-          condition: probe.condition
-        })
-      } catch (err) {
-        throw new Error(`Error setting breakpoint for probe ${probe.id}`, { cause: err })
-      }
-      probeToLocation.set(probe.id, locationKey)
-      locationToBreakpoint.set(locationKey, { id: result.breakpointId, location, locationKey })
-      breakpointToProbes.set(result.breakpointId, new Map([[probe.id, probe]]))
+  const locationKey = generateLocationKey(scriptId, lineNumber, columnNumber)
+  const breakpoint = locationToBreakpoint.get(locationKey)
+
+  log.debug(
+    '[debugger:devtools_client] %s breakpoint at %s:%d:%d (probe: %s, version: %d)',
+    breakpoint ? 'Updating' : 'Adding', url, lineNumber, columnNumber, probe.id, probe.version
+  )
+
+  if (breakpoint) {
+    // A breakpoint already exists at this location, so we need to add the probe to the existing breakpoint
+    await updateBreakpointInternal(breakpoint, probe)
+  } else {
+    // No breakpoint exists at this location, so we need to create a new one
+    const location = {
+      scriptId,
+      lineNumber: lineNumber - 1, // Beware! lineNumber is zero-indexed
+      columnNumber
+    }
+    let result
+    try {
+      result = await session.post('Debugger.setBreakpoint', {
+        location,
+        condition: probe.condition
+      })
+    } catch (err) {
+      throw new Error(`Error setting breakpoint for probe ${probe.id}`, { cause: err })
     }
-  } finally {
-    release()
+    probeToLocation.set(probe.id, locationKey)
+    locationToBreakpoint.set(locationKey, { id: result.breakpointId, location, locationKey })
+    breakpointToProbes.set(result.breakpointId, new Map([[probe.id, probe]]))
   }
 }
 
@@ -139,37 +134,37 @@ async function removeBreakpoint ({ id }) {
 
   probes.delete(id)
 
-  const release = await lock()
+  const locationKey = probeToLocation.get(id)
+  const breakpoint = locationToBreakpoint.get(locationKey)
+  const probesAtLocation = breakpointToProbes.get(breakpoint.id)
 
-  try {
-    const locationKey = probeToLocation.get(id)
-    const breakpoint = locationToBreakpoint.get(locationKey)
-    const probesAtLocation = breakpointToProbes.get(breakpoint.id)
-
-    probesAtLocation.delete(id)
-    probeToLocation.delete(id)
-
-    if (probesAtLocation.size === 0) {
-      locationToBreakpoint.delete(locationKey)
-      breakpointToProbes.delete(breakpoint.id)
-      if (breakpointToProbes.size === 0) {
-        await stop() // TODO: Will this actually delete the breakpoint?
-      } else {
-        try {
-          await session.post('Debugger.removeBreakpoint', { breakpointId: breakpoint.id })
-        } catch (err) {
-          throw new Error(`Error removing breakpoint for probe ${id}`, { cause: err })
-        }
-      }
+  probesAtLocation.delete(id)
+  probeToLocation.delete(id)
+
+  if (probesAtLocation.size === 0) {
+    locationToBreakpoint.delete(locationKey)
+    breakpointToProbes.delete(breakpoint.id)
+    if (breakpointToProbes.size === 0) {
+      await stop() // This will also remove the breakpoint
     } else {
-      await updateBreakpoint(breakpoint)
+      try {
+        await session.post('Debugger.removeBreakpoint', { breakpointId: breakpoint.id })
+      } catch (err) {
+        throw new Error(`Error removing breakpoint for probe ${id}`, { cause: err })
+      }
     }
-  } finally {
-    release()
+  } else {
+    await updateBreakpointInternal(breakpoint)
   }
 }
 
-async function updateBreakpoint (breakpoint, probe) {
+// TODO: Modify existing probe instead of removing it (DEBUG-2817)
+async function modifyBreakpoint (probe) {
+  await removeBreakpoint(probe)
+  await addBreakpoint(probe)
+}
+
+async function updateBreakpointInternal (breakpoint, probe) {
   const probesAtLocation = breakpointToProbes.get(breakpoint.id)
   const conditionBeforeNewProbe = compileCompoundCondition(Array.from(probesAtLocation.values()))
 
@@ -235,6 +230,17 @@ function stop () {
   return session.post('Debugger.disable')
 }
 
+function lock (fn) {
+  return async function (...args) {
+    const release = await mutex()
+    try {
+      return await fn(...args)
+    } finally {
+      release()
+    }
+  }
+}
+
 // Only if all probes have a condition can we use a compound condition.
 // Otherwise, we need to evaluate each probe individually once the breakpoint is hit.
 function compileCompoundCondition (probes) {

packages/dd-trace/src/debugger/devtools_client/remote_config.js

@@ -1,8 +1,7 @@
 'use strict'
 
 const { workerData: { rcPort } } = require('node:worker_threads')
-const lock = require('mutexify/promise')()
-const { addBreakpoint, removeBreakpoint } = require('./breakpoints')
+const { addBreakpoint, removeBreakpoint, modifyBreakpoint } = require('./breakpoints')
 const { ackReceived, ackInstalled, ackError } = require('./status')
 const log = require('../../log')
 
@@ -62,40 +61,21 @@ async function processMsg (action, probe) {
     )
   }
 
-  // This lock is to ensure that we don't get the following race condition:
-  //
-  // When a breakpoint is being removed and there are no other breakpoints, we disable the debugger by calling
-  // `Debugger.disable` to free resources. However, if a new breakpoint is being added around the same time, we might
-  // have a race condition where the new breakpoint thinks that the debugger is already enabled because the removal of
-  // the other breakpoint hasn't had a chance to call `Debugger.disable` yet. Then once the code that's adding the new
-  // breakpoints tries to call `Debugger.setBreakpoint` it fails because in the meantime `Debugger.disable` was called.
-  //
-  // If the code is ever refactored to not tear down the debugger if there's no active breakpoints, we can safely remove
-  // this lock.
-  const release = await lock()
-
-  try {
-    switch (action) {
-      case 'unapply':
-        await removeBreakpoint(probe)
-        break
-      case 'apply':
-        await addBreakpoint(probe)
-        ackInstalled(probe)
-        break
-      case 'modify':
-        // TODO: Modify existing probe instead of removing it (DEBUG-2817)
-        await removeBreakpoint(probe)
-        await addBreakpoint(probe)
-        ackInstalled(probe) // TODO: Should we also send ackInstalled when modifying a probe?
-        break
-      default:
-        throw new Error(
-          // eslint-disable-next-line @stylistic/js/max-len
-          `Cannot process probe ${probe.id} (version: ${probe.version}) - unknown remote configuration action: ${action}`
-        )
-    }
-  } finally {
-    release()
+  switch (action) {
+    case 'unapply':
+      await removeBreakpoint(probe)
+      break
+    case 'apply':
+      await addBreakpoint(probe)
+      ackInstalled(probe)
+      break
+    case 'modify':
+      await modifyBreakpoint(probe)
+      ackInstalled(probe)
+      break
+    default:
+      throw new Error(
+        `Cannot process probe ${probe.id} (version: ${probe.version}) - unknown remote configuration action: ${action}`
+      )
   }
 }