fix(workflow)!: Make breakpoints work inside workflow isolate context (#819)

Author: mjameswh

packages/test/src/test-bundler.ts

@@ -37,14 +37,13 @@ if (RUN_INTEGRATION_TESTS) {
 
   test('Worker can be created from bundle path', async (t) => {
     const taskQueue = `${t.title}-${uuid4()}`;
-    const { code, sourceMap } = await bundleWorkflowCode({
+    const { code } = await bundleWorkflowCode({
       workflowsPath: require.resolve('./workflows'),
     });
     const uid = uuid4();
     const codePath = pathJoin(os.tmpdir(), `workflow-bundle-${uid}.js`);
-    const sourceMapPath = pathJoin(os.tmpdir(), `workflow-bundle-${uid}.map.js`);
-    await Promise.all([writeFile(codePath, code), writeFile(sourceMapPath, sourceMap)]);
-    const workflowBundle = { codePath, sourceMapPath };
+    await writeFile(codePath, code);
+    const workflowBundle = { codePath };
     const worker = await Worker.create({
       taskQueue,
       workflowBundle,

packages/test/src/test-local-activities.ts

@@ -1,6 +1,6 @@
 import { ApplicationFailure, defaultPayloadConverter, WorkflowClient, WorkflowFailedError } from '@temporalio/client';
 import { temporal } from '@temporalio/proto';
-import { bundleWorkflowCode, Worker, WorkflowBundleWithSourceMap } from '@temporalio/worker';
+import { bundleWorkflowCode, Worker, WorkflowBundle } from '@temporalio/worker';
 import { isCancellation } from '@temporalio/workflow';
 import anyTest, { TestInterface } from 'ava';
 import { firstValueFrom, Subject } from 'rxjs';
@@ -10,7 +10,7 @@ import { RUN_INTEGRATION_TESTS } from './helpers';
 import * as workflows from './workflows/local-activity-testers';
 
 interface Context {
-  workflowBundle: WorkflowBundleWithSourceMap;
+  workflowBundle: WorkflowBundle;
   taskQueue: string;
   client: WorkflowClient;
   getWorker: () => Promise<Worker>;

packages/test/src/test-workflows.ts

@@ -10,6 +10,7 @@ import { msToTs } from '@temporalio/internal-workflow-common';
 import { coresdk } from '@temporalio/proto';
 import { WorkflowCodeBundler } from '@temporalio/worker/lib/workflow/bundler';
 import { VMWorkflow, VMWorkflowCreator } from '@temporalio/worker/lib/workflow/vm';
+import { parseWorkflowCode } from '@temporalio/worker/lib/worker';
 import { WorkflowInfo } from '@temporalio/workflow';
 import anyTest, { ExecutionContext, TestInterface } from 'ava';
 import dedent from 'dedent';
@@ -46,8 +47,8 @@ const test = anyTest as TestInterface<Context>;
 test.before(async (t) => {
   const workflowsPath = path.join(__dirname, 'workflows');
   const bundler = new WorkflowCodeBundler({ workflowsPath });
-  const { code, sourceMap } = await bundler.createBundle();
-  t.context.workflowCreator = await TestVMWorkflowCreator.create(code, sourceMap, 100);
+  const workflowBundle = parseWorkflowCode((await bundler.createBundle()).code);
+  t.context.workflowCreator = await TestVMWorkflowCreator.create(workflowBundle, 100);
 });
 
 test.after.always(async (t) => {

packages/worker/src/index.ts

@@ -34,7 +34,7 @@ export {
   ReplayWorkerOptions,
   WorkerOptions,
   WorkflowBundleOption,
-  WorkflowBundlePathWithSourceMap,
+  WorkflowBundlePath,
 } from './worker-options';
 export { WorkflowInboundLogInterceptor, workflowLogAttributes } from './workflow-log-interceptor';
-export { BundleOptions, bundleWorkflowCode, WorkflowBundleWithSourceMap } from './workflow/bundler';
+export { BundleOptions, bundleWorkflowCode, WorkflowBundle } from './workflow/bundler';

packages/worker/src/worker-options.ts

@@ -9,26 +9,25 @@ import { Runtime } from './runtime';
 import { InjectedSinks } from './sinks';
 import { GiB } from './utils';
 import { LoggerSinks } from './workflow-log-interceptor';
-import { WorkflowBundleWithSourceMap } from './workflow/bundler';
+import { WorkflowBundle } from './workflow/bundler';
 import * as v8 from 'v8';
 import * as os from 'os';
 
 export type { WebpackConfiguration };
 
-export interface WorkflowBundlePathWithSourceMap {
+export interface WorkflowBundlePath {
   codePath: string;
-  sourceMapPath: string;
 }
-export type WorkflowBundleOption = WorkflowBundleWithSourceMap | WorkflowBundlePathWithSourceMap;
+export type WorkflowBundleOption = WorkflowBundle | WorkflowBundlePath;
 
-export function isCodeBundleOption(bundleOpt: WorkflowBundleOption): bundleOpt is WorkflowBundleWithSourceMap {
+export function isCodeBundleOption(bundleOpt: WorkflowBundleOption): bundleOpt is WorkflowBundle {
   const opt = bundleOpt as any; // Cast to access properties without TS complaining
-  return typeof opt.code === 'string' && typeof opt.sourceMap === 'string';
+  return typeof opt.code === 'string';
 }
 
-export function isPathBundleOption(bundleOpt: WorkflowBundleOption): bundleOpt is WorkflowBundlePathWithSourceMap {
+export function isPathBundleOption(bundleOpt: WorkflowBundleOption): bundleOpt is WorkflowBundlePath {
   const opt = bundleOpt as any; // Cast to access properties without TS complaining
-  return typeof opt.codePath === 'string' && opt.sourceMapPath;
+  return typeof opt.codePath === 'string';
 }
 
 /**

packages/worker/src/worker.ts

@@ -75,10 +75,13 @@ import {
   WorkerOptions,
 } from './worker-options';
 import { WorkflowCodecRunner } from './workflow-codec-runner';
-import { WorkflowBundleWithSourceMap, WorkflowCodeBundler } from './workflow/bundler';
+import { WorkflowBundle, WorkflowCodeBundler } from './workflow/bundler';
 import { Workflow, WorkflowCreator } from './workflow/interface';
 import { ThreadedVMWorkflowCreator } from './workflow/threaded-vm';
 import { VMWorkflowCreator } from './workflow/vm';
+import type { RawSourceMap } from 'source-map';
+import * as vm from 'node:vm';
+import * as path from 'node:path';
 
 import IWorkflowActivationJob = coresdk.workflow_activation.IWorkflowActivationJob;
 import EvictionReason = coresdk.workflow_activation.RemoveFromCache.EvictionReason;
@@ -145,13 +148,9 @@ export interface WorkerConstructor {
   create(
     connection: NativeConnection,
     options: CompiledWorkerOptions,
-    bundle?: WorkflowBundleWithSourceMap
-  ): Promise<NativeWorkerLike>;
-  createReplay(
-    options: CompiledWorkerOptions,
-    history: History,
-    bundle: WorkflowBundleWithSourceMap
+    bundle?: WorkflowBundle
   ): Promise<NativeWorkerLike>;
+  createReplay(options: CompiledWorkerOptions, history: History, bundle: WorkflowBundle): Promise<NativeWorkerLike>;
 }
 
 function isOptionsWithBuildId<T extends CompiledWorkerOptions>(options: T): options is T & { buildId: string } {
@@ -180,7 +179,7 @@ export class NativeWorker implements NativeWorkerLike {
   public static async create(
     connection: NativeConnection,
     options: CompiledWorkerOptions,
-    bundle?: WorkflowBundleWithSourceMap
+    bundle?: WorkflowBundle
   ): Promise<NativeWorkerLike> {
     const runtime = Runtime.instance();
     const nativeWorker = await runtime.registerWorker(
@@ -193,7 +192,7 @@ export class NativeWorker implements NativeWorkerLike {
   public static async createReplay(
     options: CompiledWorkerOptions,
     history: History,
-    bundle: WorkflowBundleWithSourceMap
+    bundle: WorkflowBundle
   ): Promise<NativeWorkerLike> {
     const runtime = Runtime.instance();
     const nativeWorker = await runtime.createReplayWorker(addBuildIdIfMissing(options, bundle.code), history);
@@ -410,9 +409,8 @@ export class Worker {
         ...(compiledOptions.workflowBundle && isCodeBundleOption(compiledOptions.workflowBundle)
           ? {
               // Avoid dumping workflow bundle code to the console
-              workflowBundle: <WorkflowBundleWithSourceMap>{
+              workflowBundle: <WorkflowBundle>{
                 code: `<string of length ${compiledOptions.workflowBundle.code.length}>`,
-                sourceMap: `<string of length ${compiledOptions.workflowBundle.sourceMap.length}>`,
               },
             }
           : {}),
@@ -441,14 +439,14 @@ export class Worker {
   }
 
   protected static async createWorkflowCreator(
-    bundle: WorkflowBundleWithSourceMap,
+    workflowBundle: WorkflowBundleWithSourceMap,
     compiledOptions: CompiledWorkerOptions
   ): Promise<WorkflowCreator> {
     if (compiledOptions.debugMode) {
-      return await VMWorkflowCreator.create(bundle.code, bundle.sourceMap, compiledOptions.isolateExecutionTimeoutMs);
+      return await VMWorkflowCreator.create(workflowBundle, compiledOptions.isolateExecutionTimeoutMs);
     } else {
       return await ThreadedVMWorkflowCreator.create({
-        ...bundle,
+        workflowBundle,
         threadPoolSize: compiledOptions.workflowThreadPoolSize,
         isolateExecutionTimeoutMs: compiledOptions.isolateExecutionTimeoutMs,
       });
@@ -560,20 +558,17 @@ export class Worker {
       });
       const bundle = await bundler.createBundle();
       logger.info('Workflow bundle created', { size: `${toMB(bundle.code.length)}MB` });
-      return bundle;
+      return parseWorkflowCode(bundle.code);
     } else if (compiledOptions.workflowBundle) {
       if (compiledOptions.bundlerOptions) {
         throw new ValueError(`You cannot set both WorkerOptions.workflowBundle and .bundlerOptions`);
       }
 
       if (isCodeBundleOption(compiledOptions.workflowBundle)) {
-        return compiledOptions.workflowBundle;
+        return parseWorkflowCode(compiledOptions.workflowBundle.code);
       } else if (isPathBundleOption(compiledOptions.workflowBundle)) {
-        const [code, sourceMap] = await Promise.all([
-          fs.readFile(compiledOptions.workflowBundle.codePath, 'utf8'),
-          fs.readFile(compiledOptions.workflowBundle.sourceMapPath, 'utf8'),
-        ]);
-        return { code, sourceMap };
+        const code = await fs.readFile(compiledOptions.workflowBundle.codePath, 'utf8');
+        return parseWorkflowCode(code, compiledOptions.workflowBundle.codePath);
       } else {
         throw new TypeError('Invalid WorkflowOptions.workflowBundle');
       }
@@ -1649,6 +1644,51 @@ export class Worker {
   }
 }
 
+export interface WorkflowBundleWithSourceMap {
+  code: string;
+  sourceMap: RawSourceMap;
+  filename: string;
+}
+
+export function parseWorkflowCode(code: string, codePath?: string): WorkflowBundleWithSourceMap {
+  const sourceMappingUrlDataRegex = /\s*\n[/][/][#]\s+sourceMappingURL=data:(?:[^,]*;)base64,([0-9A-Za-z+/=]+)\s*$/;
+  const sourceMapMatcher = code.match(sourceMappingUrlDataRegex);
+  if (!sourceMapMatcher) throw new Error("Can't extract inlined source map from the provided Workflow Bundle");
+
+  const sourceMapJson = Buffer.from(sourceMapMatcher[1], 'base64').toString();
+  const sourceMap: RawSourceMap = JSON.parse(sourceMapJson);
+
+  // JS debuggers (at least VSCode's) have a few requirements regarding the script and its source map, notably:
+  // - The script file name's must look like an absolute path (relative paths are treated as node internals scripts)
+  // - If the script contains a sourceMapURL directive, the executable 'file' indicated by the source map must match the
+  //   filename of the script itself. If the source map's file is a relative path, then it gets resolved relative to cwd
+  const filename = path.resolve(process.cwd(), codePath ?? sourceMap.file);
+  if (filename !== codePath) {
+    sourceMap.file = filename;
+    const patchedSourceMapJson = Buffer.from(JSON.stringify(sourceMap)).toString('base64');
+    const fixedSourceMappingUrl = `\n//# sourceMappingURL=data:application/json;base64,${patchedSourceMapJson}`;
+    code = code.slice(0, -sourceMapMatcher[1].length) + fixedSourceMappingUrl;
+  }
+
+  // Preloading the script makes breakpoints significantly more reliable and more responsive
+  let script: vm.Script | undefined = new vm.Script(code, { filename });
+  let context: any = vm.createContext({});
+  try {
+    script.runInContext(context);
+  } catch (e) {
+    // Context has not been properly configured, so eventual errors are possible. Just ignore at this point
+  }
+
+  // Keep these objects from GC long enough for debugger to complete parsing the source map and reporting locations
+  // to the node process. Otherwise, the debugger risks source mapping resolution errors, meaning breakpoints wont work.
+  setTimeout(() => {
+    script = undefined;
+    context = undefined;
+  }, 10000);
+
+  return { code, sourceMap, filename };
+}
+
 type NonNullableObject<T> = { [P in keyof T]-?: NonNullable<T[P]> };
 
 /**

packages/worker/src/workflow/bundler.ts

@@ -26,9 +26,8 @@ export function moduleMatches(userModule: string, modules: string[]): boolean {
   return modules.some((module) => userModule === module || userModule.startsWith(`${module}/`));
 }
 
-export interface WorkflowBundleWithSourceMap {
+export interface WorkflowBundle {
   code: string;
-  sourceMap: string;
 }
 
 /**
@@ -64,9 +63,9 @@ export class WorkflowCodeBundler {
   }
 
   /**
-   * @return a {@link WorkflowBundleWithSourceMap} containing bundled code and source map
+   * @return a {@link WorkflowBundle} containing bundled code, including inlined source map
    */
-  public async createBundle(): Promise<WorkflowBundleWithSourceMap> {
+  public async createBundle(): Promise<WorkflowBundle> {
     const vol = new memfs.Volume();
     const ufs = new unionfs.Union();
 
@@ -99,12 +98,11 @@ export class WorkflowCodeBundler {
     const entrypointPath = this.makeEntrypointPath(ufs, this.workflowsPath);
 
     this.genEntrypoint(vol, entrypointPath);
-    await this.bundle(ufs, memoryFs, entrypointPath, distDir);
+    const bundleFilePath = await this.bundle(ufs, memoryFs, entrypointPath, distDir);
 
     // Cast because the type definitions are inaccurate
     return {
-      code: memoryFs.readFileSync(path.join(distDir, 'main.js'), 'utf8') as string,
-      sourceMap: memoryFs.readFileSync(path.join(distDir, 'main.source.js'), 'utf8') as string,
+      code: memoryFs.readFileSync(bundleFilePath, 'utf8') as string,
     };
   }
 
@@ -165,7 +163,7 @@ export { api };
     outputFilesystem: memfs.IFs,
     entry: string,
     distDir: string
-  ): Promise<void> {
+  ): Promise<string> {
     const captureProblematicModules: webpack.Configuration['externals'] = async (
       data,
       _callback
@@ -221,11 +219,10 @@ export { api };
       },
       entry: [entry],
       mode: 'development',
-      devtool: 'source-map',
+      devtool: 'inline-source-map',
       output: {
         path: distDir,
-        filename: 'main.js',
-        sourceMapFilename: 'main.source.js',
+        filename: 'workflow-isolate-[fullhash].js',
         devtoolModuleFilenameTemplate: '[absolute-resource-path]',
         library: '__TEMPORAL__',
       },
@@ -241,7 +238,7 @@ export { api };
     compiler.outputFileSystem = outputFilesystem as any;
 
     try {
-      await new Promise<void>((resolve, reject) => {
+      return await new Promise<string>((resolve, reject) => {
         compiler.run((err, stats) => {
           if (stats !== undefined) {
             const hasError = stats.hasErrors();
@@ -274,12 +271,13 @@ export { api };
 
               reject(err);
             }
+
+            const outputFilename = Object.keys(stats.compilation.assets)[0];
+            if (!err) {
+              resolve(path.join(distDir, outputFilename));
+            }
           }
-          if (err) {
-            reject(err);
-          } else {
-            resolve();
-          }
+          reject(err);
         });
       });
     } finally {
@@ -335,7 +333,7 @@ export interface BundleOptions {
  * When using with {@link Worker.runReplayHistory}, make sure to pass the same interceptors and payload converter used
  * when the history was generated.
  */
-export async function bundleWorkflowCode(options: BundleOptions): Promise<WorkflowBundleWithSourceMap> {
+export async function bundleWorkflowCode(options: BundleOptions): Promise<WorkflowBundle> {
   const bundler = new WorkflowCodeBundler(options);
   return await bundler.createBundle();
 }

packages/worker/src/workflow/threaded-vm.ts

@@ -13,6 +13,7 @@ import { coresdk } from '@temporalio/proto';
 import { IllegalStateError, SinkCall } from '@temporalio/workflow';
 import { Worker as NodeWorker } from 'worker_threads';
 import { UnexpectedError } from '../errors';
+import { WorkflowBundleWithSourceMap } from '../worker';
 import { Workflow, WorkflowCreateOptions, WorkflowCreator } from './interface';
 import { WorkerThreadInput, WorkerThreadRequest } from './workflow-worker-thread/input';
 import { WorkerThreadOutput, WorkerThreadResponse } from './workflow-worker-thread/output';
@@ -118,8 +119,7 @@ export class WorkerThreadClient {
 }
 
 export interface ThreadedVMWorkflowCreatorOptions {
-  code: string;
-  sourceMap: string;
+  workflowBundle: WorkflowBundleWithSourceMap;
   threadPoolSize: number;
   isolateExecutionTimeoutMs: number;
 }
@@ -137,15 +137,14 @@ export class ThreadedVMWorkflowCreator implements WorkflowCreator {
    */
   static async create({
     threadPoolSize,
-    code,
-    sourceMap,
+    workflowBundle,
     isolateExecutionTimeoutMs,
   }: ThreadedVMWorkflowCreatorOptions): Promise<ThreadedVMWorkflowCreator> {
     const workerThreadClients = Array(threadPoolSize)
       .fill(0)
       .map(() => new WorkerThreadClient(new NodeWorker(require.resolve('./workflow-worker-thread'))));
     await Promise.all(
-      workerThreadClients.map((client) => client.send({ type: 'init', code, sourceMap, isolateExecutionTimeoutMs }))
+      workerThreadClients.map((client) => client.send({ type: 'init', workflowBundle, isolateExecutionTimeoutMs }))
     );
     return new this(workerThreadClients);
   }