fix: make the code transform idempotent per channel

lib/transforms.js

@@ -124,6 +124,106 @@ function traceAny (state, node, _parent, ancestry) {
   }
 }
 
+/**
+ * Returns `true` if `stmt` is a `return <channelVariable>.start.runStores(...)`
+ * statement, the shape every wrapper template
+ * (`wrapSync`/`wrapPromise`/`wrapCallback`/`wrapAuto`) ends with.
+ *
+ * @param {import('estree').Node} stmt
+ * @param {string} channelVariable
+ * @returns {boolean}
+ */
+function isChannelRunStoresReturn (stmt, channelVariable) {
+  if (stmt.type !== 'ReturnStatement') return false
+  const callee = stmt.argument?.type === 'CallExpression' ? stmt.argument.callee : null
+  return callee?.type === 'MemberExpression' &&
+    callee.property?.name === 'runStores' &&
+    callee.object?.type === 'MemberExpression' &&
+    callee.object.property?.name === 'start' &&
+    callee.object.object?.type === 'Identifier' &&
+    callee.object.object.name === channelVariable
+}
+
+/**
+ * Given the body statements of a wrapper function, returns the function that was
+ * moved into its `__apm$traced` closure as `const __apm$wrapped = <fn>` (i.e. the
+ * next layer down: either an inner wrapper or the original function). Returns
+ * `null` when there is no such closure, which marks the bottom of the stack (the
+ * original, un-wrapped body).
+ *
+ * @param {import('estree').Statement[]} body
+ * @returns {import('estree').Function|null}
+ */
+function nextWrappedFunction (body) {
+  const declInit = (stmts, name) => {
+    const decl = stmts?.find(stmt =>
+      stmt.type === 'VariableDeclaration' &&
+      stmt.declarations.some(d => d.id?.name === name)
+    )
+    return decl?.declarations.find(d => d.id?.name === name)?.init ?? null
+  }
+
+  const traced = declInit(body, '__apm$traced')
+  const wrapped = declInit(traced?.body?.body, '__apm$wrapped')
+  const type = wrapped?.type
+  return type === 'FunctionExpression' || type === 'ArrowFunctionExpression' || type === 'FunctionDeclaration'
+    ? wrapped
+    : null
+}
+
+/**
+ * Returns `true` if `node` is already wrapped for `channelVariable` at any layer
+ * of the wrapper stack this transform builds.
+ *
+ * The check is scoped to the specific channel (not just "is this instrumented at
+ * all") on purpose: re-wrapping the same function on the *same* channel
+ * double-publishes to one channel and is the double-instrumentation bug we skip,
+ * whereas wrapping it on a *different* channel is a second, independent APM whose
+ * instrumentation must not be silently dropped.
+ *
+ * It walks down the `__apm$wrapped` closure chain rather than only inspecting the
+ * outermost layer, so an already-wrapped channel is still detected after a
+ * different channel has been wrapped on top of it. The walk only follows this
+ * transform's own scaffolding, so unrelated nested functions in the original body
+ * are never mistaken for a wrapper.
+ *
+ * @param {import('estree').Node} node - A function-like node.
+ * @param {string} channelVariable - The channel variable name for this config.
+ * @returns {boolean}
+ */
+function isWrappedForChannel (node, channelVariable) {
+  const visited = new Set()
+  let fn = node
+  while (fn && !visited.has(fn)) {
+    visited.add(fn)
+    const body = fn.body?.body
+    if (!Array.isArray(body)) return false
+    if (body.some(stmt => isChannelRunStoresReturn(stmt, channelVariable))) return true
+    fn = nextWrappedFunction(body)
+  }
+  return false
+}
+
+/**
+ * Returns `true` if `ctor` already declares `bindingName`, the per-channel
+ * `const <channelVar>$<method> = this["<method>"]` binding injected by
+ * {@link traceInstanceMethod}. Used to keep re-transforming already instrumented
+ * output idempotent for instance methods wrapped at runtime inside the
+ * constructor, without dropping a different channel's patch.
+ *
+ * @param {import('estree').MethodDefinition} ctor - The class constructor node.
+ * @param {string} bindingName - The channel-namespaced saved-original binding.
+ * @returns {boolean}
+ */
+function constructorPatchesMethod (ctor, bindingName) {
+  const body = ctor?.value?.body?.body
+  if (!Array.isArray(body)) return false
+  return body.some(stmt =>
+    stmt.type === 'VariableDeclaration' &&
+    stmt.declarations.some(decl => decl.id?.name === bindingName)
+  )
+}
+
 /**
  * Wraps a function node's body with diagnostics_channel tracing.
  *
@@ -135,6 +235,13 @@ function traceAny (state, node, _parent, ancestry) {
  * @param {import('estree').Program} program
  */
 function traceFunction (state, node, program) {
+  // Idempotency guard: the name-based esquery selectors re-match the outer
+  // wrapper function on a second pass (the `[async]` token is structural, and
+  // the wrapper keeps the target's name), so bail out if this node is already
+  // wrapped for this channel instead of double-publishing to it. A different
+  // channel (a second, independent APM) is intentionally allowed to wrap it.
+  if (isWrappedForChannel(node, formatChannelVariable(state.channelName))) return
+
   state.transforms.tracingChannelDeclaration(state, program)
 
   const { functionQuery: { methodName, privateMethodName, functionName, expressionName, propertyName } } = state
@@ -177,7 +284,7 @@ function traceFunction (state, node, program) {
  * @param {import('estree').Program} program
  */
 function traceInstanceMethod (state, node, program) {
-  const { functionQuery, operator } = state
+  const { functionQuery, operator, channelName } = state
   const { methodName } = functionQuery
 
   // No methodName means a constructor-only config — the constructor FunctionExpression
@@ -196,6 +303,17 @@ function traceInstanceMethod (state, node, program) {
   // wrap it in the constructor instead.
   let ctor = classBody.body.find(({ kind }) => kind === 'constructor')
 
+  // The binding that captures the previous `this[method]` is namespaced by
+  // channel so that patches for different channels stack in the constructor
+  // (each wraps the previous one) rather than colliding on one `const`.
+  const savedBinding = `${formatChannelVariable(channelName)}$${methodName}`
+
+  // Idempotency guard: skip if this channel has already patched this method into
+  // the constructor. Scoped per channel (not per method) so a re-run of the same
+  // config is a no-op while a different channel is still allowed to add its own
+  // patch, keeping this path consistent with `traceFunction`.
+  if (ctor && constructorPatchesMethod(ctor, savedBinding)) return
+
   state.transforms.tracingChannelDeclaration(state, program)
 
   if (!ctor) {
@@ -209,11 +327,11 @@ function traceInstanceMethod (state, node, program) {
   }
 
   const ctorBody = parse(`
-    const __apm$${methodName} = this["${methodName}"]
+    const ${savedBinding} = this["${methodName}"]
     this["${methodName}"] = function () {}
-    if (typeof __apm$${methodName} === 'function') {
+    if (typeof ${savedBinding} === 'function') {
       Object.defineProperty(this["${methodName}"], 'length', {
-        value: __apm$${methodName}.length,
+        value: ${savedBinding}.length,
         configurable: true
       })
     }
@@ -224,7 +342,7 @@ function traceInstanceMethod (state, node, program) {
   fn.params = [{ type: 'RestElement', argument: { type: 'Identifier', name: '__apm$args' } }]
 
   fn.async = operator === 'tracePromise'
-  fn.body = wrap(state, { type: 'Identifier', name: `__apm$${methodName}` }, program)
+  fn.body = wrap(state, { type: 'Identifier', name: savedBinding }, program)
   wrapSuper(state, fn)
 
   ctor.value.body.body.push(...ctorBody)

tests/tests.test.mjs

@@ -849,3 +849,91 @@ describe('async_iterator_cjs', () => {
     ])
   })
 })
+
+describe('idempotency', () => {
+  const M = { name: TEST_MODULE_NAME, versionRange: '>=0.0.1', filePath: TEST_MODULE_PATH }
+
+  const transform = (code, configs) =>
+    create(configs)
+      .getTransformer(TEST_MODULE_NAME, TEST_MODULE_VERSION, TEST_MODULE_PATH)
+      .transform(code, 'cjs').code
+
+  const cases = [
+    ['class method (Async)', 'class Undici { async fetch (url) { return 42 } }\nmodule.exports = { Undici }\n',
+      [{ channelName: 'Undici:fetch', module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Async' } }]],
+    ['class method (Sync)', 'class Undici { fetch (url) { return 42 } }\nmodule.exports = { Undici }\n',
+      [{ channelName: 'Undici:fetch', module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Sync' } }]],
+    ['class method (Callback)', 'class Undici { fetch (url, cb) { cb(null, 42) } }\nmodule.exports = { Undici }\n',
+      [{ channelName: 'Undici:fetch', module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Callback', callbackIndex: 1 } }]],
+    ['function declaration (Async)', 'async function fetch (url) { return 42 }\nmodule.exports = { fetch }\n',
+      [{ channelName: 'fetch', module: M, functionQuery: { functionName: 'fetch', kind: 'Async' } }]],
+    ['function declaration (Sync)', 'function fetch (url) { return 42 }\nmodule.exports = { fetch }\n',
+      [{ channelName: 'fetch', module: M, functionQuery: { functionName: 'fetch', kind: 'Sync' } }]],
+    ['function declaration (Callback)', 'function fetch (url, cb) { cb(null, 42) }\nmodule.exports = { fetch }\n',
+      [{ channelName: 'fetch', module: M, functionQuery: { functionName: 'fetch', kind: 'Callback', callbackIndex: 1 } }]],
+    ['function expression (Async)', 'const fetch = async function fetch (url) { return 42 }\nmodule.exports = { fetch }\n',
+      [{ channelName: 'fetch', module: M, functionQuery: { expressionName: 'fetch', kind: 'Async' } }]],
+    ['instance method via constructor (Async)',
+      'class Base {}\nBase.prototype.fetch = async function (url) { return 42 }\nclass Sub extends Base {}\nmodule.exports = { Sub }\n',
+      [{ channelName: 'Base:fetch', module: M, functionQuery: { className: 'Base', methodName: 'fetch', kind: 'Async' } }]],
+  ]
+
+  const runStoresCount = (code) => code.split('.start.runStores(').length - 1
+
+  for (const [label, code, configs] of cases) {
+    test(`a second pass is a no-op: ${label}`, () => {
+      const once = transform(code, configs)
+      const twice = transform(once, configs)
+      assert.equal(twice, once, 'second pass should not change already-instrumented output')
+      assert.equal(runStoresCount(twice), runStoresCount(once))
+    })
+  }
+
+  test('a second config with a different channel wraps independently without double-publishing to either channel', () => {
+    const code = 'class Undici { async fetch (url) { return 42 } }\nmodule.exports = { Undici }\n'
+    const once = transform(code, [{ channelName: 'chanA', module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Async' } }])
+    const twice = transform(once, [{ channelName: 'chanB', module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Async' } }])
+
+    // Both APMs coexist (both channels are wrapped) ...
+    assert.equal(runStoresCount(twice), 2)
+    // ... but neither channel is published to more than once, so no channel
+    // double-fires (the actual duplicate-span bug is same-channel only).
+    assert.equal(twice.split('tr_ch_apm$chanA.start.runStores(').length - 1, 1)
+    assert.equal(twice.split('tr_ch_apm$chanB.start.runStores(').length - 1, 1)
+  })
+
+  test('does not re-wrap a channel that is nested under a different channel', () => {
+    const code = 'class Undici { async fetch (url) { return 42 } }\nmodule.exports = { Undici }\n'
+    const chan = (name) => [{ channelName: name, module: M, functionQuery: { className: 'Undici', methodName: 'fetch', kind: 'Async' } }]
+
+    // chanA (innermost) -> chanB wraps over it -> chanA again: the second chanA
+    // must be skipped even though chanA is no longer the outermost wrapper.
+    const a = transform(code, chan('chanA'))
+    const ab = transform(a, chan('chanB'))
+    const aba = transform(ab, chan('chanA'))
+
+    assert.equal(aba.split('tr_ch_apm$chanA.start.runStores(').length - 1, 1)
+    assert.equal(aba.split('tr_ch_apm$chanB.start.runStores(').length - 1, 1)
+    // Re-applying chanA to the already-chanA-wrapped output is a no-op.
+    assert.equal(aba, ab)
+  })
+
+  test('runtime-patched instance method: same channel is idempotent, different channel coexists', () => {
+    // Inherited/prototype method (not on the class body) is wrapped at runtime
+    // inside the constructor.
+    const code = 'class Base {}\nBase.prototype.fetch = async function (url) { return 42 }\nclass Sub extends Base {}\nmodule.exports = { Sub }\n'
+    const chan = (name) => [{ channelName: name, module: M, functionQuery: { className: 'Base', methodName: 'fetch', kind: 'Async' } }]
+
+    // Same channel twice is a no-op.
+    const a = transform(code, chan('chanA'))
+    assert.equal(transform(a, chan('chanA')), a)
+
+    // A different channel adds its own patch (both APMs coexist) ...
+    const ab = transform(a, chan('chanB'))
+    assert.ok(ab.includes('tracingChannel("orchestrion:undici:chanA")'))
+    assert.ok(ab.includes('tracingChannel("orchestrion:undici:chanB")'))
+    // ... and each channel is still patched exactly once.
+    assert.equal(ab.split('tr_ch_apm$chanA.start.runStores(').length - 1, 1)
+    assert.equal(ab.split('tr_ch_apm$chanB.start.runStores(').length - 1, 1)
+  })
+})