fix(debugger): rebuild breakpoint condition when a probe stops capturing

integration-tests/debugger/capture-disabled.spec.js

@@ -0,0 +1,69 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+const { setTimeout: sleep } = require('node:timers/promises')
+const { inspect } = require('node:util')
+
+const { setup } = require('./utils')
+
+// The guardrail counters are converted into telemetry metrics every 10 seconds, which are then sent on the next
+// telemetry heartbeat
+const GUARDRAIL_METRICS_FLUSH_INTERVAL_MS = 10_000
+
+// The probe is limited to one event per second, so two hits in quick succession skip the second one
+const PER_PROBE_RATE_LIMIT_WINDOW_MS = 1_000
+
+describe('Dynamic Instrumentation', function () {
+  const t = setup({
+    testApp: 'target-app/capture-disabled.js',
+    dependencies: ['fastify'],
+    env: { DD_TELEMETRY_HEARTBEAT_INTERVAL: '1' },
+  })
+
+  describe('probe whose capture gets permanently disabled', function () {
+    this.timeout(GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 3)
+
+    it('should stop treating the probe as snapshot producing', async function () {
+      const results = []
+      t.agent.on('debugger-input', ({ payload }) => results.push(...payload))
+
+      const checkMetrics = t.agent.assertTelemetryReceived({
+        fn: ({ payload }) => {
+          const skipped = payload.payload.series.filter((entry) => {
+            return entry.metric === 'events.skipped' && entry.tags.includes('reason:rateLimitProbe')
+          })
+          const eventTypes = skipped.map(({ tags }) => tags.find((tag) => tag.startsWith('event_type:')))
+          assert.deepStrictEqual(eventTypes, ['event_type:log'], `Unexpected skipped events: ${inspect(skipped)}`)
+          assert.strictEqual(skipped[0].points[0][1], 1)
+        },
+        requestType: 'generate-metrics',
+        timeout: GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 2,
+        resolveAtFirstSuccess: true,
+        namespace: 'live_debugger',
+      })
+
+      const firstResult = new Promise((resolve) => t.agent.once('debugger-input', resolve))
+      t.agent.addRemoteConfig(t.generateRemoteConfig({
+        captureSnapshot: true,
+        sampling: { snapshotsPerSecond: 1_000 / PER_PROBE_RATE_LIMIT_WINDOW_MS },
+      }))
+      await t.triggerBreakpoint()
+
+      // The first hit trips the large object safety threshold, which disables capture for the probe
+      const { payload: [{ debugger: { snapshot } }] } = await firstResult
+      assert.strictEqual(snapshot.captures.lines[t.breakpoint.line].locals.huge.notCapturedReason, 'fieldCount')
+      assert.match(snapshot.evaluationErrors[0].message, /exceeds the maximum number of allowed properties/)
+
+      // Two hits within the rate limit window: the first emits a capture-less event, the second is skipped. The skip
+      // must be classified as a log event now that the probe no longer produces snapshots.
+      await sleep(PER_PROBE_RATE_LIMIT_WINDOW_MS + 100)
+      await Promise.all([t.request(t.breakpoint.url), t.request(t.breakpoint.url)])
+
+      await checkMetrics
+
+      assert.strictEqual(results.length, 2, `Expected two probe results, got ${inspect(results)}`)
+      assert.strictEqual(results[1].debugger.snapshot.captures, undefined, 'should not capture anything anymore')
+      assert.match(results[1].debugger.snapshot.evaluationErrors[0].message, /exceeds the maximum number/)
+    })
+  })
+})

integration-tests/debugger/target-app/capture-disabled.js

@@ -0,0 +1,26 @@
+'use strict'
+
+// @ts-expect-error This code is running in a sandbox where dd-trace is available
+require('dd-trace/init')
+// @ts-expect-error This code is running in a sandbox where fastify is available
+const Fastify = require('fastify')
+const {
+  LARGE_OBJECT_SKIP_THRESHOLD,
+  // @ts-expect-error This code is running in a sandbox where dd-trace is available
+} = require('dd-trace/packages/dd-trace/src/debugger/devtools_client/snapshot/constants')
+
+const fastify = Fastify({ logger: { level: 'error' } })
+
+fastify.get('/object-over-safety-threshold', function handler () {
+  // Just over the threshold at which the collector gives up on capturing snapshots at this location for good
+  const huge = Object.fromEntries(Array.from({ length: LARGE_OBJECT_SKIP_THRESHOLD + 1 }, (_, i) => [`p${i}`, i]))
+  return { size: Object.keys(huge).length } // BREAKPOINT: /object-over-safety-threshold
+})
+
+fastify.listen({ port: process.env.APP_PORT || 0 }, (err) => {
+  if (err) {
+    fastify.log.error(err)
+    process.exit(1)
+  }
+  process.send?.({ port: fastify.server.address().port })
+})

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

@@ -53,6 +53,7 @@ module.exports = {
   addBreakpoint: lock(addBreakpoint),
   removeBreakpoint: lock(removeBreakpoint),
   modifyBreakpoint: lock(modifyBreakpoint),
+  refreshBreakpoint: lock(refreshBreakpoint),
 }
 
 async function addBreakpoint (probe) {
@@ -234,23 +235,50 @@ async function modifyBreakpoint (probe) {
   await addBreakpoint(probe)
 }
 
-async function updateBreakpointInternal (breakpoint, probe) {
+/**
+ * Rebuild the breakpoint condition at a probe's location from the current state of the probes attached to it.
+ *
+ * The breakpoint condition bakes in whether each probe produces snapshots, which decides if a hit counts against the
+ * global snapshot rate limit and how a skipped hit is classified. That changes when the pause handler permanently
+ * disables capture for a probe after a fatal capture error, so the condition has to be recompiled.
+ *
+ * A probe that has been removed in the meantime is ignored: its location no longer needs the update.
+ *
+ * @param {{ id: string }} probe - A probe attached to the breakpoint to refresh.
+ * @returns {Promise<void>}
+ */
+async function refreshBreakpoint ({ id }) {
+  if (!sessionStarted) return
+  const locationKey = probeToLocation.get(id)
+  if (locationKey === undefined) return
+  await updateBreakpointInternal(locationToBreakpoint.get(locationKey), undefined, `while refreshing ${locationKey}`)
+}
+
+/**
+ * Replace the breakpoint at a location with one whose condition matches the probes currently attached to it.
+ *
+ * @param {{ id: string, location: object, locationKey: string }} breakpoint - The breakpoint to replace.
+ * @param {object} [probe] - A probe to attach to the breakpoint first, when one is being added.
+ * @param {string} [context] - What the update is part of, for error messages. Derived from `probe` when omitted.
+ * @returns {Promise<void>}
+ */
+async function updateBreakpointInternal (breakpoint, probe, context) {
   const probesAtLocation = breakpointToProbes.get(breakpoint.id)
 
-  // If a probe is provided, add it to the breakpoint. If not, it's because we're removing a probe. In both cases the
-  // breakpoint condition must be rebuilt to match the remaining probes at the location.
+  // If a probe is provided, add it to the breakpoint. If not, it's because we're removing a probe or the probes at the
+  // location changed. In all cases the breakpoint condition must be rebuilt to match the probes at the location.
   if (probe) {
     probesAtLocation.set(probe.id, probe)
     probeToLocation.set(probe.id, breakpoint.locationKey)
+    context ??= `while adding probe ${probe.id} (version: ${probe.version})`
+  } else {
+    context ??= `after removing probe from ${breakpoint.locationKey}`
   }
 
   try {
     await session.post('Debugger.removeBreakpoint', { breakpointId: breakpoint.id })
   } catch (err) {
-    const message = probe
-      ? `Error replacing breakpoint while adding probe ${probe.id} (version: ${probe.version})`
-      : `Error replacing breakpoint after removing probe from ${breakpoint.locationKey}`
-    throw new Error(message, { cause: err })
+    throw new Error(`Error replacing breakpoint ${context}`, { cause: err })
   }
   breakpointToProbes.delete(breakpoint.id)
   let result
@@ -260,10 +288,7 @@ async function updateBreakpointInternal (breakpoint, probe) {
       condition: compileBreakpointCondition([...probesAtLocation.values()]),
     }))
   } catch (err) {
-    const message = probe
-      ? `Error setting breakpoint while adding probe ${probe.id} (version: ${probe.version})`
-      : `Error setting breakpoint after removing probe from ${breakpoint.locationKey}`
-    throw new Error(message, { cause: err })
+    throw new Error(`Error setting breakpoint ${context}`, { cause: err })
   }
   breakpoint.id = result.breakpointId
   breakpointToProbes.set(result.breakpointId, probesAtLocation)

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

@@ -13,6 +13,7 @@ const {
   SAMPLED_PROBE_OVERFLOW_INDEX,
 } = require('../probe_sampler_constants')
 const { breakpointToProbes, samplingIndexToProbe } = require('./state')
+const { refreshBreakpoint } = require('./breakpoints')
 const session = require('./session')
 const { getLocalStateForCallFrame, evaluateCaptureExpressions } = require('./snapshot')
 const send = require('./send')
@@ -190,6 +191,9 @@ session.on('Debugger.paused', async ({ params }) => {
   const dd = processDD(evalResults[0]) // the first result is the dd tags, the rest are the probe template results
   let messageIndex = 1
 
+  // A probe whose capture got permanently disabled during this pause, if any
+  let captureDisabledProbe
+
   // TODO: Send multiple probes in one HTTP request as an array (DEBUG-2848)
   for (const probe of probes) {
     const snapshot = {
@@ -221,6 +225,7 @@ session.on('Debugger.paused', async ({ params }) => {
           expr: '',
           message: error.message,
         }))
+        captureDisabledProbe ??= probe
       }
       snapshot.captures = {
         lines: { [probe.location.lines[0]]: { locals: processLocalState() } },
@@ -237,6 +242,7 @@ session.on('Debugger.paused', async ({ params }) => {
             expr: '',
             message: error.message,
           }))
+          captureDisabledProbe ??= probe
         }
 
         snapshot.captures = {
@@ -297,6 +303,18 @@ session.on('Debugger.paused', async ({ params }) => {
       config.propagateProcessTags.enabled ? processTags.serialized : undefined,
       eventType, incompleteReasons)
   }
+
+  if (captureDisabledProbe !== undefined) {
+    // The breakpoint condition bakes in whether each probe produces snapshots, which decides if a hit counts against
+    // the global snapshot rate limit and how a skipped hit is classified. Rebuild it now that this changed. All probes
+    // at the location share the breakpoint, so one refresh covers every probe disabled during this pause.
+    refreshBreakpoint(captureDisabledProbe).catch((err) => {
+      log.error(
+        '[debugger:devtools_client] Error refreshing breakpoint after disabling capture for probe %s (version: %s)',
+        captureDisabledProbe.id, captureDisabledProbe.version, err
+      )
+    })
+  }
 })
 
 function processDD (result) {

packages/dd-trace/test/debugger/devtools_client/breakpoints.spec.js

@@ -1144,6 +1144,80 @@ describe('breakpoints', function () {
     })
   })
 
+  describe('refreshBreakpoint', function () {
+    it('should rebuild the breakpoint condition from the current state of the probes at the location',
+      async function () {
+        await addProbe({ captureSnapshot: true })
+        await addProbe({ id: 'probe-2', where: { sourceFile: 'test2.js', lines: ['20'] } })
+        const probe = stateMock.breakpointToProbes.get(breakpointId)?.get('probe-1')
+        const otherProbe = stateMock.breakpointToProbes.get('bp-script-1:19:0')?.get('probe-2')
+        assert(probe !== undefined && otherProbe !== undefined)
+        sessionMock.post.resetHistory()
+
+        // What the pause handler does when it permanently disables capture for the probe
+        probe.captureSnapshot = false
+
+        await breakpoints.refreshBreakpoint(probe)
+
+        sinon.assert.calledWith(sessionMock.post.firstCall, 'Debugger.removeBreakpoint', { breakpointId })
+        sinon.assert.calledWith(sessionMock.post.secondCall, 'Debugger.setBreakpoint', {
+          location: {
+            scriptId: 'script-1',
+            lineNumber: 9,
+            columnNumber: 0,
+          },
+          condition: compileBreakpointCondition([
+            { id: 'probe-1', samplingIndex: 0, nsBetweenSampling: 1000000000n, captureSnapshot: false },
+          ]),
+        })
+        sinon.assert.calledTwice(sessionMock.post)
+
+        assert.strictEqual(stateMock.probeToLocation.get('probe-1'), 'script-1:10:0')
+        assert.strictEqual(stateMock.breakpointToProbes.get(breakpointId)?.get('probe-1'), probe)
+        assert.strictEqual(stateMock.breakpointToProbes.get('bp-script-1:19:0')?.get('probe-2'), otherProbe,
+          'should leave the other locations alone')
+      })
+
+    it('should ignore a probe that has been removed in the meantime', async function () {
+      await addProbe()
+      await breakpoints.removeBreakpoint({ id: 'probe-1' })
+      sessionMock.post.resetHistory()
+
+      await breakpoints.refreshBreakpoint({ id: 'probe-1' })
+
+      sinon.assert.notCalled(sessionMock.post)
+    })
+
+    it('should ignore a probe when the debugger is not started', async function () {
+      await breakpoints.refreshBreakpoint({ id: 'probe-1' })
+
+      sinon.assert.notCalled(sessionMock.post)
+    })
+
+    it('should wrap errors when setting the replacement breakpoint fails', async function () {
+      await addProbe()
+      sessionMock.post.resetHistory()
+
+      const cause = new Error('inspector failure')
+      sessionMock.post.callsFake((method) => {
+        if (method === 'Debugger.setBreakpoint') {
+          return Promise.reject(cause)
+        }
+        return Promise.resolve({})
+      })
+
+      await assert.rejects(
+        breakpoints.refreshBreakpoint({ id: 'probe-1' }),
+        (err) => {
+          assert(err instanceof Error)
+          assert.strictEqual(err.message, 'Error setting breakpoint while refreshing script-1:10:0')
+          assert.strictEqual(err.cause, cause)
+          return true
+        }
+      )
+    })
+  })
+
   describe('re-evaluation', function () {
     it('should log errors from async probe re-evaluation', async function () {
       await addProbe()