feat(node): Adds domain implementation of AsyncContextStrategy (#7767)

Author: timfish

packages/core/src/hub.ts

@@ -55,7 +55,7 @@ const DEFAULT_BREADCRUMBS = 100;
  */
 export interface AsyncContextStrategy {
   getCurrentHub: () => Hub | undefined;
-  runWithAsyncContext<T, A>(callback: (hub: Hub, ...args: A[]) => T, ...args: A[]): T;
+  runWithAsyncContext<T>(callback: (hub: Hub) => T, ...args: unknown[]): T;
 }
 
 /**
@@ -536,25 +536,44 @@ export function getCurrentHub(): Hub {
     }
   }
 
+  // Prefer domains over global if they are there (applicable only to Node environment)
+  if (isNodeEnv()) {
+    return getHubFromActiveDomain(registry);
+  }
+
+  // Return hub that lives on a global object
+  return getGlobalHub(registry);
+}
+
+function getGlobalHub(registry: Carrier = getMainCarrier()): Hub {
   // If there's no hub, or its an old API, assign a new one
   if (!hasHubOnCarrier(registry) || getHubFromCarrier(registry).isOlderThan(API_VERSION)) {
     setHubOnCarrier(registry, new Hub());
   }
 
-  // Prefer domains over global if they are there (applicable only to Node environment)
-  if (isNodeEnv()) {
-    return getHubFromActiveDomain(registry);
-  }
   // Return hub that lives on a global object
   return getHubFromCarrier(registry);
 }
 
+/**
+ * @private Private API with no semver guarantees!
+ *
+ * If the carrier does not contain a hub, a new hub is created with the global hub client and scope.
+ */
+export function ensureHubOnCarrier(carrier: Carrier): void {
+  // If there's no hub on current domain, or it's an old API, assign a new one
+  if (!hasHubOnCarrier(carrier) || getHubFromCarrier(carrier).isOlderThan(API_VERSION)) {
+    const globalHubTopStack = getGlobalHub().getStackTop();
+    setHubOnCarrier(carrier, new Hub(globalHubTopStack.client, Scope.clone(globalHubTopStack.scope)));
+  }
+}
+
 /**
  * @private Private API with no semver guarantees!
  *
  * Sets the global async context strategy
  */
-export function setAsyncContextStrategy(strategy: AsyncContextStrategy): void {
+export function setAsyncContextStrategy(strategy: AsyncContextStrategy | undefined): void {
   // Get main carrier (global for every environment)
   const registry = getMainCarrier();
   registry.__SENTRY__ = registry.__SENTRY__ || {};
@@ -566,15 +585,15 @@ export function setAsyncContextStrategy(strategy: AsyncContextStrategy): void {
  *
  * Runs the given callback function with the global async context strategy
  */
-export function runWithAsyncContext<T, A>(callback: (hub: Hub, ...args: A[]) => T, ...args: A[]): T {
+export function runWithAsyncContext<T>(callback: (hub: Hub) => T, ...args: unknown[]): T {
   const registry = getMainCarrier();
 
   if (registry.__SENTRY__ && registry.__SENTRY__.acs) {
     return registry.__SENTRY__.acs.runWithAsyncContext(callback, ...args);
   }
 
   // if there was no strategy, fallback to just calling the callback
-  return callback(getCurrentHub(), ...args);
+  return callback(getCurrentHub());
 }
 
 /**

packages/core/src/index.ts

@@ -26,6 +26,7 @@ export {
   getMainCarrier,
   runWithAsyncContext,
   setHubOnCarrier,
+  ensureHubOnCarrier,
   setAsyncContextStrategy,
 } from './hub';
 export { makeSession, closeSession, updateSession } from './session';

packages/node/src/async/domain.ts

@@ -0,0 +1,45 @@
+import type { Carrier, Hub } from '@sentry/core';
+import {
+  ensureHubOnCarrier,
+  getCurrentHub as getCurrentHubCore,
+  getHubFromCarrier,
+  setAsyncContextStrategy,
+} from '@sentry/core';
+import * as domain from 'domain';
+import { EventEmitter } from 'events';
+
+function getCurrentHub(): Hub | undefined {
+  // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-explicit-any
+  const activeDomain = (domain as any).active as Carrier;
+
+  // If there's no active domain, just return undefined and the global hub will be used
+  if (!activeDomain) {
+    return undefined;
+  }
+
+  ensureHubOnCarrier(activeDomain);
+
+  return getHubFromCarrier(activeDomain);
+}
+
+function runWithAsyncContext<T, A>(callback: (hub: Hub) => T, ...args: A[]): T {
+  const local = domain.create();
+
+  for (const emitter of args) {
+    if (emitter instanceof EventEmitter) {
+      local.add(emitter);
+    }
+  }
+
+  return local.bind(() => {
+    const hub = getCurrentHubCore();
+    return callback(hub);
+  })();
+}
+
+/**
+ * Sets the async context strategy to use Node.js domains.
+ */
+export function setDomainAsyncContextStrategy(): void {
+  setAsyncContextStrategy({ getCurrentHub, runWithAsyncContext });
+}

packages/node/test/async/domain.test.ts

@@ -0,0 +1,69 @@
+import { getCurrentHub, Hub, runWithAsyncContext, setAsyncContextStrategy } from '@sentry/core';
+import * as domain from 'domain';
+
+import { setDomainAsyncContextStrategy } from '../../src/async/domain';
+
+describe('domains', () => {
+  afterAll(() => {
+    // clear the strategy
+    setAsyncContextStrategy(undefined);
+  });
+
+  test('without domain', () => {
+    // @ts-ignore property active does not exist on domain
+    expect(domain.active).toBeFalsy();
+    const hub = getCurrentHub();
+    expect(hub).toEqual(new Hub());
+  });
+
+  test('domain hub scope inheritance', () => {
+    const globalHub = getCurrentHub();
+    globalHub.configureScope(scope => {
+      scope.setExtra('a', 'b');
+      scope.setTag('a', 'b');
+      scope.addBreadcrumb({ message: 'a' });
+    });
+    runWithAsyncContext(hub => {
+      expect(globalHub).toEqual(hub);
+    });
+  });
+
+  test('domain hub single instance', () => {
+    setDomainAsyncContextStrategy();
+
+    runWithAsyncContext(hub => {
+      expect(hub).toBe(getCurrentHub());
+    });
+  });
+
+  test('concurrent domain hubs', done => {
+    setDomainAsyncContextStrategy();
+
+    let d1done = false;
+    let d2done = false;
+
+    runWithAsyncContext(hub => {
+      hub.getStack().push({ client: 'process' } as any);
+      expect(hub.getStack()[1]).toEqual({ client: 'process' });
+      // Just in case so we don't have to worry which one finishes first
+      // (although it always should be d2)
+      setTimeout(() => {
+        d1done = true;
+        if (d2done) {
+          done();
+        }
+      });
+    });
+
+    runWithAsyncContext(hub => {
+      hub.getStack().push({ client: 'local' } as any);
+      expect(hub.getStack()[1]).toEqual({ client: 'local' });
+      setTimeout(() => {
+        d2done = true;
+        if (d1done) {
+          done();
+        }
+      });
+    });
+  });
+});