Loops and parallel steps

Run steps in loops, side by side, and behind conditions, and split big jobs across several runs.

Steps are ordinary function calls, so they work in loops, in conditions, and in Promise.all. On each call, the SDK matches them with their saved results in the order your code reaches them. This page shows how to write each case, and how to split jobs too big for a single run.

Loops

Each step in a loop is saved on its own. An import cut off halfway resumes at the first batch with no saved result:

nelkap/workflows/import-contacts.ts

import { importRows, splitIntoBatches } from '../../contacts';
import { nelkap } from '../client';

export async function importContacts(fileId: string) {
  const batches = await nelkap.step.run('batches', () =>
    splitIntoBatches(fileId, 500),
  );

  for (const [index, batchId] of batches.entries()) {
    await nelkap.step.run(`batch ${index}`, () =>
      importRows(batchId),
    );
  }
}

The list of batches comes from a step too, so every call of the run loops over the same list.

Name each step after its item, like batch 3. A loop of step.run('send') works too, since steps with the same name are matched in order, but the run’s page then can’t tell you which step handled which item.

Parallel steps

Steps can run at the same time. Start them together and wait for all of them:

const [profile, orders] = await Promise.all([
  nelkap.step.run('profile', () => getProfile(userId)),
  nelkap.step.run('orders', () => getOrders(userId)),
]);

The two functions run in parallel. Their results are saved in the order your code called the steps, so a run that resumes later matches them the same way.

The same works for a list, with one step per item:

await Promise.all(
  userIds.map((userId) =>
    nelkap.step.run(`digest ${userId}`, () =>
      sendDigest(userId),
    ),
  ),
);

Each item’s step counts toward the run’s limit of 100 steps. For longer lists, see Fan-out.

Conditions

A step can sit behind an if, as long as the condition takes the same path on every call. Base it on the payload or on the result of an earlier step:

const user = await nelkap.step.run('user', () =>
  getUser(userId),
);

if (user.plan === 'pro') {
  await nelkap.step.run('pro welcome', () =>
    sendProWelcome(userId),
  );
}

Never base a condition on something read outside a step, like the time or a database. A run that resumes could take the other path, and its steps would no longer match the saved ones. See Determinism.

Step limit

A run holds up to 100 steps, of every type, and 32 MB of saved steps in all. A run that goes over fails with stepLimitReached. Two ways keep big jobs under the limit:

• Put several items in each step. A step that imports 500 rows counts once.
• Split the job across runs, as below.

Fan-out

For a job of any size, give each part its own run. A first workflow lists the parts and starts a run for each one, then each run does its part:

nelkap/workflows/import-file.ts

import { importRows, splitIntoBatches } from '../../contacts';
import { nelkap } from '../client';

export async function importFile(fileId: string) {
  const batches = await nelkap.step.run('batches', () =>
    splitIntoBatches(fileId, 500),
  );

  for (const [index, batchId] of batches.entries()) {
    await nelkap.step.run(`start batch ${index}`, () =>
      nelkap.run(importBatch, batchId),
    );
  }
}

export async function importBatch(batchId: string) {
  await nelkap.step.run('import', () => importRows(batchId));
}

Starting each run in a step means a run that resumes later doesn’t start the same batch twice. Give importBatch a concurrency limit to set how many batches run at once.

The first workflow still counts one step per batch. For more batches than one run can start, add a level: the first workflow starts a run per chunk of batches, and each of those starts a run per batch.

The first run ends once it has started every batch, without waiting for them. To act when the last batch is done, count the batches done in your database, as the large import example does.