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() || undefined,
     startTime: tsToDate(raw.startTime!),
     executionTime: optionalTsToDate(raw.executionTime),
     closeTime: optionalTsToDate(raw.closeTime),

packages/client/src/types.ts

@@ -36,6 +36,8 @@ export interface WorkflowExecutionInfo {
   taskQueue: string;
   status: { code: proto.temporal.api.enums.v1.WorkflowExecutionStatus; name: WorkflowExecutionStatusName };
   historyLength: number;
+  // Only available in server version 1.20 or later
+  historySizeBytes?: number;
   startTime: Date;
   executionTime?: Date;
   closeTime?: Date;

packages/common/src/time.ts

@@ -1,5 +1,4 @@
-// eslint-disable-next-line import/no-named-as-default
-import Long from 'long';
+import Long from 'long'; // eslint-disable-line import/no-named-as-default
 import ms, { StringValue } from 'ms';
 import type { google } from '@temporalio/proto';
 import { ValueError } from './errors';

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,
+      continueAsNewSuggested: false,
+      // values ignored for the purpose of comparison
+      historySizeBytes: result.historySizeBytes,
       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,52 @@ test('Condition 0 patch sets a timer', async (t) => {
   t.false(await worker.runUntil(executeWorkflow(conditionTimeout0)));
 });
 
+export async function historySizeBytesGrows(): Promise<[number | undefined, number | undefined]> {
+  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.true(after && before && 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.true(historySizeBytes && historySizeBytes > 100);
+});
+
+export async function suggestedCAN(): Promise<boolean> {
+  const maxEvents = 40_000;
+  const batchSize = 100;
+  if (workflow.workflowInfo().continueAsNewSuggested) {
+    return false;
+  }
+  while (workflow.workflowInfo().historyLength < maxEvents) {
+    await Promise.all(new Array(batchSize).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.true(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

@@ -1157,6 +1157,8 @@ export class Worker {
                         searchAttributes,
                       } = startWorkflow;
 
+                      // Exact truncation for multi-petabyte histories
+                      const historySizeBytes = activation.historySizeBytes.toNumber();
                       const workflowInfo: WorkflowInfo = {
                         workflowId,
                         runId: activation.runId,
@@ -1193,6 +1195,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,
+                        // A zero value means that it was not set by the server
+                        historySizeBytes: historySizeBytes ? historySizeBytes : undefined,
+                        continueAsNewSuggested: activation.continueAsNewSuggested,
                         unsafe: {
                           now: () => Date.now(), // re-set in initRuntime
                           isReplaying: activation.isReplaying,

packages/workflow/src/interfaces.ts

@@ -70,6 +70,27 @@ 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.
+   *
+   * This value is only available in server versions > 1.20
+   *
+   * 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.
+   *
+   * Supported only on Temporal Server 1.20+, always `false` on older servers.
+   */
+  continueAsNewSuggested: boolean;
+
   /**
    * Task queue this Workflow is executing on
    */

packages/workflow/src/worker-interface.ts

@@ -186,11 +186,14 @@ export function activate(activation: coresdk.workflow_activation.WorkflowActivat
         // timestamp will not be updated for activation that contain only queries
         activator.now = tsToMs(activation.timestamp);
       }
-      if (activation.historyLength == null) {
-        throw new TypeError('Got activation with no historyLength');
-      }
+
+      // The Rust Core ensures that these activation fields are not null
       activator.info.unsafe.isReplaying = activation.isReplaying ?? false;
-      activator.info.historyLength = activation.historyLength;
+      activator.info.historyLength = activation.historyLength as number;
+      // Exact truncation for multi-petabyte histories
+      // historySizeBytes === 0 means WFT was generated by pre-1.20.0 server, and the history size is unknown
+      activator.info.historySizeBytes = activation.historySizeBytes?.toNumber() || undefined;
+      activator.info.continueAsNewSuggested = activation.continueAsNewSuggested ?? false;
     }
 
     // Cast from the interface to the class which has the `variant` attribute.