Question re: Continue As New Typescript Single Entity Pattern Example

Community Support
,
1

In the docs, here: Features - TypeScript SDK feature guide | Temporal Documentation

They describe the single-entity design pattern for counting the number of updates processed by a workflow, and then calling continue-as-new when that number exceeds a threshold.

interface Input {
  /* Define your Workflow input type here */
}
interface Update {
  /* Define your Workflow update type here */
}

const MAX_ITERATIONS = 1;

export async function entityWorkflow(
  input: Input,
  isNew = true,
): Promise<void> {
  try {
    const pendingUpdates = Array<Update>();
    setHandler(updateSignal, (updateCommand) => {
      pendingUpdates.push(updateCommand);
    });

    if (isNew) {
      await setup(input);
    }

    for (let iteration = 1; iteration <= MAX_ITERATIONS; ++iteration) {
      // Ensure that we don't block the Workflow Execution forever waiting
      // for updates, which means that it will eventually Continue-As-New
      // even if it does not receive updates.
      await condition(() => pendingUpdates.length > 0, '1 day');

      while (pendingUpdates.length) {
        const update = pendingUpdates.shift();
        await runAnActivityOrChildWorkflow(update);
      }
    }
  } catch (err) {
    if (isCancellation(err)) {
      await CancellationScope.nonCancellable(async () => {
        await cleanup();
      });
    }
    throw err;
  }
  await continueAsNew<typeof entityWorkflow>(input, false);
}

The block Iā€™m confused about is:


    for (let iteration = 1; iteration <= MAX_ITERATIONS; ++iteration) {
      // Ensure that we don't block the Workflow Execution forever waiting
      // for updates, which means that it will eventually Continue-As-New
      // even if it does not receive updates.
      await condition(() => pendingUpdates.length > 0, '1 day');

      while (pendingUpdates.length) {
        const update = pendingUpdates.shift();
        await runAnActivityOrChildWorkflow(update);
      }
    }

Why do you need the while loop? What is the difference between that, and writing:


    for (let iteration = 1; iteration <= MAX_ITERATIONS; ++iteration) {
      // Ensure that we don't block the Workflow Execution forever waiting
      // for updates, which means that it will eventually Continue-As-New
      // even if it does not receive updates.
      await condition(() => pendingUpdates.length > 0, '1 day');
      if(pendingUpdates.length > 0) {
        const update = pendingUpdates.shift();
        await runAnActivityOrChildWorkflow(update);
      }
}
2

This is needed to ensure that pending updates are not lost due to the following scenario:

  • The loop executes the last iteration
  • There is more than one pending update or there is 1 pending update, but another is received when runAnActivityOrChildWorkflow is executing.
  • The workflow continues as new, losing all the unprocessed pending updates.

The sample code processes all pending updates even if their number exceeds the number of iterations. There is also a guarantee that updates are not lost due to a race condition between the update and the workflow calling continue-as-new. This is achieved by ensuring that the signal handler is always invoked before the main workflow coroutine is unblocked.

1 Like