feat(workflow): Add historySizeInBytes and continueAsNewSuggested to WorkflowInfo (#695)

packages/client/src/helpers.ts

@@ -59,6 +59,8 @@ export async function executionInfoFromRaw<T>(
     },
     // Safe to convert to number, max history length is 50k, which is much less than Number.MAX_SAFE_INTEGER
     historyLength: raw.historyLength!.toNumber(),
+    // Exact truncation for multi-petabyte histories
+    historySizeBytes: raw.historySizeBytes!.toNumber(),
     startTime: tsToDate(raw.startTime!),
     executionTime: optionalTsToDate(raw.executionTime),
     closeTime: optionalTsToDate(raw.closeTime),

packages/client/src/types.ts

@@ -36,6 +36,7 @@ export interface WorkflowExecutionInfo {
   taskQueue: string;
   status: { code: proto.temporal.api.enums.v1.WorkflowExecutionStatus; name: WorkflowExecutionStatusName };
   historyLength: number;
+  historySizeBytes: number;
   startTime: Date;
   executionTime?: Date;
   closeTime?: Date;

packages/test/src/integration-tests.ts

@@ -680,6 +680,7 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       },
     });
     const result = await workflow.result();
+    t.assert(typeof result.historySizeBytes === 'number' && result.historySizeBytes > 300);
     t.deepEqual(result, {
       memo: {
         nested: { object: true },
@@ -694,6 +695,9 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       workflowType: 'returnWorkflowInfo',
       workflowId,
       historyLength: 3,
+      // historySizeBytes changes in every run, e.g., due to variable encoding of process id
+      historySizeBytes: result.historySizeBytes,
+      continueAsNewSuggested: false,
       startTime: result.startTime,
       runStartTime: result.runStartTime,
       // unsafe.now is a function, so doesn't make it through serialization, but .now is required, so we need to cast

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

@@ -216,6 +216,49 @@ test('Condition 0 patch sets a timer', async (t) => {
   t.false(await worker.runUntil(executeWorkflow(conditionTimeout0)));
 });
 
+export async function historySizeBytesGrows(): Promise<[number, number]> {
+  const before = workflow.workflowInfo().historySizeBytes;
+  await workflow.sleep(1);
+  const after = workflow.workflowInfo().historySizeBytes;
+  return [before, after];
+}
+
+test('HistorySizeBytes grows with new WFT', async (t) => {
+  const { createWorker, executeWorkflow } = helpers(t);
+  const worker = await createWorker();
+  const [before, after] = await worker.runUntil(executeWorkflow(historySizeBytesGrows));
+  t.assert(after > before && before > 100);
+});
+
+test('HistorySizeBytes is visible in WorkflowExecutionInfo', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+  const worker = await createWorker();
+  const handle = await startWorkflow(historySizeBytesGrows);
+
+  await worker.runUntil(handle.result());
+  const historySizeBytes = (await handle.describe()).historySizeBytes;
+  t.assert(historySizeBytes > 100);
+});
+
+export async function suggestedCAN(): Promise<boolean> {
+  const MAX_EVENTS = 40_000;
+  const BATCH_SIZE = 100;
+  while (workflow.workflowInfo().historyLength < MAX_EVENTS) {
+    await Promise.all(new Array(BATCH_SIZE).fill(undefined).map((_) => workflow.sleep(1)));
+    if (workflow.workflowInfo().continueAsNewSuggested) {
+      return true;
+    }
+  }
+  return false;
+}
+
+test('ContinueAsNew is suggested', async (t) => {
+  const { createWorker, executeWorkflow } = helpers(t);
+  const worker = await createWorker();
+  const flaggedCAN = await worker.runUntil(executeWorkflow(suggestedCAN));
+  t.assert(flaggedCAN);
+});
+
 test('Activity initialInterval is not getting rounded', async (t) => {
   const { createWorker, startWorkflow } = helpers(t);
   const worker = await createWorker({

packages/test/src/test-sinks.ts

@@ -55,14 +55,9 @@ if (RUN_INTEGRATION_TESTS) {
       fn: string;
     }
 
-    const dummyDate = new Date(2000, 1, 0, 0, 0, 0);
     function fixWorkflowInfoDates(input: WorkflowInfo): WorkflowInfo {
       delete (input.unsafe as any).now;
-      return {
-        ...input,
-        startTime: dummyDate,
-        runStartTime: dummyDate,
-      };
+      return input;
     }
 
     const recordedCalls: RecordedCall[] = [];
@@ -112,6 +107,11 @@ if (RUN_INTEGRATION_TESTS) {
       await wf.result();
       return wf;
     });
+
+    // Capture volatile values that are hard to predict
+    const { historySizeBytes, startTime, runStartTime } = recordedCalls[0].info;
+    t.assert(typeof historySizeBytes === 'number' && historySizeBytes > 300);
+
     const info: WorkflowInfo = {
       namespace: 'default',
       firstExecutionRunId: wf.firstExecutionRunId,
@@ -134,8 +134,11 @@ if (RUN_INTEGRATION_TESTS) {
       parent: undefined,
       searchAttributes: {},
       historyLength: 3,
-      startTime: dummyDate,
-      runStartTime: dummyDate,
+      continueAsNewSuggested: false,
+      // values ignored for the purpose of comparison
+      historySizeBytes,
+      startTime,
+      runStartTime,
       // unsafe.now() doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: {
         isReplaying: false,

packages/test/src/test-workflows.ts

@@ -111,6 +111,8 @@ async function createWorkflow(
       taskQueue: 'test',
       searchAttributes: {},
       historyLength: 3,
+      historySizeBytes: 300,
+      continueAsNewSuggested: false,
       unsafe: { isReplaying: false, now: Date.now },
       startTime: new Date(),
       runStartTime: new Date(),
@@ -353,6 +355,14 @@ test('successString', async (t) => {
   compareCompletion(t, req, makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload('success'))]));
 });
 
+test('continueAsNewSuggested', async (t) => {
+  const { workflowType } = t.context;
+  const activation = makeStartWorkflow(workflowType);
+  activation.continueAsNewSuggested = true;
+  const req = await activate(t, activation);
+  compareCompletion(t, req, makeSuccess([makeCompleteWorkflowExecution(defaultPayloadConverter.toPayload(true))]));
+});
+
 function cleanWorkflowFailureStackTrace(
   req: coresdk.workflow_completion.IWorkflowActivationCompletion,
   commandIndex = 0

packages/test/src/workflows/continue-as-new-suggested.ts

@@ -0,0 +1,5 @@
+import { workflowInfo } from '@temporalio/workflow';
+
+export async function continueAsNewSuggested(): Promise<boolean> {
+  return workflowInfo().continueAsNewSuggested;
+}

packages/test/src/workflows/index.ts

@@ -28,6 +28,7 @@ export * from './condition';
 export * from './condition-completion-race';
 export * from './condition-timeout-0';
 export * from './continue-as-new-same-workflow';
+export * from './continue-as-new-suggested';
 export * from './continue-as-new-to-different-workflow';
 export * from './core-issue-589';
 export * from './date';

packages/worker/src/worker.ts

@@ -1193,6 +1193,9 @@ export class Worker {
                         // 0 is the default, and not a valid value, since crons are at least a minute apart
                         cronScheduleToScheduleInterval: optionalTsToMs(cronScheduleToScheduleInterval) || undefined,
                         historyLength: activation.historyLength,
+                        // Exact truncation for multi-petabyte histories
+                        historySizeBytes: activation.historySizeBytes.toNumber(),
+                        continueAsNewSuggested: activation.continueAsNewSuggested,
                         unsafe: {
                           now: () => Date.now(), // re-set in initRuntime
                           isReplaying: activation.isReplaying,

packages/workflow/src/interfaces.ts

@@ -70,6 +70,23 @@ export interface WorkflowInfo {
    */
   historyLength: number;
 
+  /**
+   * Size of Workflow history in bytes until the current Workflow Task.
+   *
+   * This value changes during the lifetime of an Execution.
+   *
+   * You may safely use this information to decide when to {@link continueAsNew}.
+   */
+  historySizeBytes: number;
+
+  /**
+   * A hint provided by the current WorkflowTaskStarted event recommending whether to
+   * {@link continueAsNew}.
+   *
+   * This value changes during the lifetime of an Execution.
+   */
+  continueAsNewSuggested: boolean;
+
   /**
    * Task queue this Workflow is executing on
    */

packages/workflow/src/worker-interface.ts

@@ -189,8 +189,18 @@ export function activate(activation: coresdk.workflow_activation.WorkflowActivat
       if (activation.historyLength == null) {
         throw new TypeError('Got activation with no historyLength');
       }
+      if (activation.historySizeBytes == null) {
+        throw new TypeError('Got activation with no historySizeBytes');
+      }
+      if (activation.continueAsNewSuggested == null) {
+        throw new TypeError('Got activation with no continueAsNewSuggested');
+      }
+
       activator.info.unsafe.isReplaying = activation.isReplaying ?? false;
       activator.info.historyLength = activation.historyLength;
+      // Exact truncation for multi-petabyte histories
+      activator.info.historySizeBytes = activation.historySizeBytes.toNumber();
+      activator.info.continueAsNewSuggested = activation.continueAsNewSuggested;
     }
 
     // Cast from the interface to the class which has the `variant` attribute.