Expose root execution info (#1662)

Author: THardy98

.github/workflows/ci.yml

@@ -323,3 +323,5 @@ jobs:
     secrets:
       ALGOLIA_API_KEY: ${{ secrets.ALGOLIA_API_KEY }}
       VERCEL_TOKEN: ${{ secrets.VERCEL_TOKEN }}
+      VERCEL_ORG_ID: ${{ secrets.VERCEL_ORG_ID }}
+      VERCEL_PROJECT_ID: ${{ secrets.VERCEL_PROJECT_ID }}

.github/workflows/docs.yml

@@ -15,6 +15,12 @@ on:
       VERCEL_TOKEN:
         required: false
         description: The Vercel token. Required if 'publish_target' is set.
+      VERCEL_ORG_ID:
+        required: false
+        description: The Vercel token. Required if 'publish_target' is set.
+      VERCEL_PROJECT_ID:
+        required: false
+        description: The Vercel token. Required if 'publish_target' is set.
 
 env:
   GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
@@ -62,10 +68,11 @@ jobs:
 
       - name: Publish docs
         if: ${{ inputs.publish_target }}
+        env:
+          VERCEL_ORG_ID: ${{ secrets.VERCEL_ORG_ID }}
+          VERCEL_PROJECT_ID: ${{ secrets.VERCEL_PROJECT_ID }}
         run: |
           npx vercel deploy packages/docs/build \
             -t '${{ secrets.VERCEL_TOKEN }}' \
-            --name typescript \
-            --scope temporal \
             --yes \
             ${{ inputs.publish_target == 'prod' && '--prod' || '' }}

packages/client/src/helpers.ts

@@ -78,6 +78,12 @@ export async function executionInfoFromRaw<T>(
           runId: raw.parentExecution.runId!,
         }
       : undefined,
+    rootExecution: raw.rootExecution
+      ? {
+          workflowId: raw.rootExecution.workflowId!,
+          runId: raw.rootExecution.runId!,
+        }
+      : undefined,
     raw: rawDataToEmbed,
     priority: decodePriority(raw.priority),
   };

packages/client/src/types.ts

@@ -51,6 +51,7 @@ export interface WorkflowExecutionInfo {
   searchAttributes: SearchAttributes; // eslint-disable-line deprecation/deprecation
   typedSearchAttributes: TypedSearchAttributes;
   parentExecution?: Required<proto.temporal.api.common.v1.IWorkflowExecution>;
+  rootExecution?: Required<proto.temporal.api.common.v1.IWorkflowExecution>;
   raw: RawWorkflowExecutionInfo;
   priority?: Priority;
 }

packages/test/src/test-integration-workflows.ts

@@ -24,7 +24,7 @@ import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
 import { Context, createLocalTestEnvironment, helpers, makeTestFunction } from './helpers-integration';
 import { overrideSdkInternalFlag } from './mock-internal-flags';
-import { asSdkLoggerSink, loadHistory, RUN_TIME_SKIPPING_TESTS } from './helpers';
+import { asSdkLoggerSink, loadHistory, RUN_TIME_SKIPPING_TESTS, waitUntil } from './helpers';
 
 const test = makeTestFunction({
   workflowsPath: __filename,
@@ -1337,3 +1337,78 @@ test('can register search attributes to dev server', async (t) => {
   t.deepEqual(desc.searchAttributes, { 'new-search-attr': [12] }); // eslint-disable-line deprecation/deprecation
   await env.teardown();
 });
+
+export async function ChildWorkflowInfo(): Promise<workflow.RootWorkflowInfo | undefined> {
+  let blocked = true;
+  workflow.setHandler(unblockSignal, () => {
+    blocked = false;
+  });
+  await workflow.condition(() => !blocked);
+  return workflow.workflowInfo().root;
+}
+
+export async function WithChildWorkflow(childWfId: string): Promise<workflow.RootWorkflowInfo | undefined> {
+  return await workflow.executeChild(ChildWorkflowInfo, {
+    workflowId: childWfId,
+  });
+}
+
+test('root execution is exposed', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+  const worker = await createWorker();
+
+  await worker.runUntil(async () => {
+    const childWfId = 'child-wf-id';
+    const handle = await startWorkflow(WithChildWorkflow, {
+      args: [childWfId],
+    });
+
+    const childHandle = t.context.env.client.workflow.getHandle(childWfId);
+    const childStarted = async (): Promise<boolean> => {
+      try {
+        await childHandle.describe();
+        return true;
+      } catch (e) {
+        if (e instanceof workflow.WorkflowNotFoundError) {
+          return false;
+        } else {
+          throw e;
+        }
+      }
+    };
+    await waitUntil(childStarted, 5000);
+    const childDesc = await childHandle.describe();
+    const parentDesc = await handle.describe();
+
+    t.true(childDesc.rootExecution?.workflowId === parentDesc.workflowId);
+    t.true(childDesc.rootExecution?.runId === parentDesc.runId);
+
+    await childHandle.signal(unblockSignal);
+    const childWfInfoRoot = await handle.result();
+    t.true(childWfInfoRoot?.workflowId === parentDesc.workflowId);
+    t.true(childWfInfoRoot?.runId === parentDesc.runId);
+  });
+});
+
+export async function rootWorkflow(): Promise<string> {
+  let result = '';
+  if (!workflow.workflowInfo().root) {
+    result += 'empty';
+  } else {
+    result += workflow.workflowInfo().root!.workflowId;
+  }
+  if (!workflow.workflowInfo().parent) {
+    result += ' ';
+    result += await workflow.executeChild(rootWorkflow);
+  }
+  return result;
+}
+
+test('Workflow can return root workflow', async (t) => {
+  const { createWorker, executeWorkflow } = helpers(t);
+  const worker = await createWorker();
+  await worker.runUntil(async () => {
+    const result = await executeWorkflow(rootWorkflow, { workflowId: 'test-root-workflow-length' });
+    t.deepEqual(result, 'empty test-root-workflow-length');
+  });
+});

packages/test/src/test-schedules.ts

@@ -773,7 +773,7 @@ if (RUN_INTEGRATION_TESTS) {
         const exists =
           desc.typedSearchAttributes.getAll().find((pair) => pair.key.name === attributeName) !== undefined;
         return exists === shouldExist;
-      }, 300);
+      }, 5000);
       return await handle.describe();
     };
 

packages/test/src/test-sinks.ts

@@ -117,6 +117,7 @@ if (RUN_INTEGRATION_TESTS) {
       lastResult: undefined,
       memo: {},
       parent: undefined,
+      root: undefined,
       searchAttributes: {},
       // FIXME: consider rehydrating the class before passing to sink functions or
       // create a variant of WorkflowInfo that corresponds to what we actually get in sinks.

packages/test/src/test-typed-search-attributes.ts

@@ -135,7 +135,7 @@ if (test?.serial?.before) {
         Object.keys(untypedKeys).every((key) => key in resp.customAttributes) &&
         Object.keys(typedKeys).every((key) => key in resp.customAttributes)
       );
-    }, 300);
+    }, 5000);
   });
 }
 

packages/worker/src/utils.ts

@@ -1,5 +1,5 @@
-import type { coresdk } from '@temporalio/proto';
-import { IllegalStateError, ParentWorkflowInfo } from '@temporalio/workflow';
+import type { coresdk, temporal } from '@temporalio/proto';
+import { IllegalStateError, ParentWorkflowInfo, RootWorkflowInfo } from '@temporalio/workflow';
 
 export const MiB = 1024 ** 2;
 
@@ -28,3 +28,19 @@ export function convertToParentWorkflowType(
     namespace: parent.namespace,
   };
 }
+
+export function convertToRootWorkflowType(
+  root: temporal.api.common.v1.IWorkflowExecution | null | undefined
+): RootWorkflowInfo | undefined {
+  if (root == null) {
+    return undefined;
+  }
+  if (!root.workflowId || !root.runId) {
+    throw new IllegalStateError('Root workflow execution is missing a field that should be defined');
+  }
+
+  return {
+    workflowId: root.workflowId,
+    runId: root.runId,
+  };
+}

packages/worker/src/worker.ts

@@ -71,7 +71,7 @@ import {
 } from './replay';
 import { History, Runtime } from './runtime';
 import { CloseableGroupedObservable, closeableGroupBy, mapWithState, mergeMapWithState } from './rxutils';
-import { byteArrayToBuffer, convertToParentWorkflowType } from './utils';
+import { byteArrayToBuffer, convertToParentWorkflowType, convertToRootWorkflowType } from './utils';
 import {
   CompiledWorkerOptions,
   CompiledWorkerOptionsWithBuildId,
@@ -1259,6 +1259,7 @@ export class Worker {
       randomnessSeed,
       workflowType,
       parentWorkflowInfo,
+      rootWorkflow,
       workflowExecutionTimeout,
       workflowRunTimeout,
       workflowTaskTimeout,
@@ -1281,6 +1282,7 @@ export class Worker {
       searchAttributes: {},
       typedSearchAttributes: new TypedSearchAttributes(),
       parent: convertToParentWorkflowType(parentWorkflowInfo),
+      root: convertToRootWorkflowType(rootWorkflow),
       taskQueue: this.options.taskQueue,
       namespace: this.options.namespace,
       firstExecutionRunId,

packages/workflow/src/index.ts

@@ -96,6 +96,7 @@ export {
   StackTraceFileSlice,
   ParentClosePolicy,
   ParentWorkflowInfo,
+  RootWorkflowInfo,
   StackTraceSDKInfo,
   StackTrace,
   UnsafeWorkflowInfo,

packages/workflow/src/interfaces.ts

@@ -62,6 +62,22 @@ export interface WorkflowInfo {
    */
   readonly parent?: ParentWorkflowInfo;
 
+  /**
+   * The root workflow execution, defined as follows:
+   * 1. A workflow without a parent workflow is its own root workflow.
+   * 2. A workflow with a parent workflow has the same root workflow as
+   * its parent.
+   *
+   * When there is no parent workflow, i.e., the workflow is its own root workflow,
+   * this field is `undefined`.
+   *
+   * Note that Continue-as-New (or reset) propagates the workflow parentage relationship,
+   * and therefore, whether the new workflow has the same root workflow as the original one
+   * depends on whether it had a parent.
+   *
+   */
+  readonly root?: RootWorkflowInfo;
+
   /**
    * Result from the previous Run (present if this is a Cron Workflow or was Continued As New).
    *
@@ -228,6 +244,11 @@ export interface ParentWorkflowInfo {
   namespace: string;
 }
 
+export interface RootWorkflowInfo {
+  workflowId: string;
+  runId: string;
+}
+
 /**
  * Not an actual error, used by the Workflow runtime to abort execution when {@link continueAsNew} is called
  */