Parallel Steps and Sleep

Imagine checking conditions at five ski resorts by calling each one on the phone. You call Mammoth. Wait. Call Palisades. Wait. By the time you get to Mt. Bachelor, Mammoth's conditions have changed. You wouldn't do this in real life. You'd call all five at once and deal with the answers as they come in.

The workflow from lesson 3.1 has one step that handles everything sequentially. If Mammoth's weather API takes 500ms and you have 5 resorts, that's 2.5 seconds of serial waiting. Let's split the work: one step per resort, all running in parallel. And after evaluation finishes, we'll use sleep() to schedule a re-check without cron jobs or external schedulers.

Outcome

Refactor the workflow to run one step per resort in parallel and use sleep() to schedule automatic re-evaluation.

Fast Track

1. Extract a evaluateResort step function that handles a single resort
2. Use Promise.all in the workflow to run all resort steps concurrently
3. Add sleep() to pause the workflow and re-check later if no alerts triggered

One Step per Resort

In 3.1, one big step did all the work. The problem: if the weather API fails for Mammoth, the entire step retries, including the successful fetches for the other 4 resorts.

Lesson 3.1 (one step):
  [evaluateAllAlerts] → mammoth → palisades → ... → mt-bachelor
  If palisades fails → entire step retries from mammoth

Lesson 3.2 (one step per resort):
  [evaluateResort: mammoth]    ┐
  [evaluateResort: palisades]  ├→ parallel, independent retries
  [evaluateResort: steamboat]  │
  [evaluateResort: targhee]    │
  [evaluateResort: bachelor]   ┘
  If palisades fails → only palisades retries

Each step is independently retryable. If Palisades times out, only Palisades retries. The other 4 results are already recorded.

Hands-on exercise 3.2

Refactor the workflow to use parallel steps and sleep:

Requirements:

1. Extract evaluateResort(resortId, alerts) as its own "use step" function
2. In the workflow function, group alerts by resort and dispatch parallel steps with Promise.all
3. After evaluation, if no alerts triggered and we haven't rechecked 3 times, sleep('30m') and re-evaluate
4. Return the final results with the number of rounds completed

Implementation hints:

• Promise.all in a workflow function dispatches steps concurrently. The platform runs them in parallel
• sleep('30m') suspends the workflow for 30 minutes without consuming resources. No server running, no memory used. After 30 minutes, it resumes
• For local testing, use sleep('10s') instead of sleep('30m') so you don't wait half an hour
• The workflow function can call itself recursively for re-checks by returning the result of another evaluateAlerts call with an incremented counter
• Data between workflow and step functions is serialized (passed by value). Return modified data from steps rather than mutating shared state

Try It

1. Trigger the workflow with alerts for multiple resorts:

   $ curl -X POST http://localhost:5173/api/workflow \
     -H "Content-Type: application/json" \
     -d '{"alerts": [{"id": "a1", "resortId": "mammoth", "condition": {"type": "conditions", "match": "powder"}, "originalQuery": "test", "createdAt": "2025-01-01", "triggered": false}, {"id": "a2", "resortId": "grand-targhee", "condition": {"type": "temperature", "operator": "lt", "value": 20, "unit": "fahrenheit"}, "originalQuery": "test", "createdAt": "2025-01-01", "triggered": false}, {"id": "a3", "resortId": "steamboat", "condition": {"type": "snowfall", "operator": "gt", "value": 6, "unit": "inches"}, "originalQuery": "test", "createdAt": "2025-01-01", "triggered": false}]}'

2. Open the Workflow dashboard:

   npx workflow web

   You should see three parallel evaluateResort steps, one for each resort. They start at roughly the same time instead of sequentially.

3. Check server logs:

   [Workflow] Round complete { round: 1, evaluated: 3, triggered: 0 }
   

4. Observe the sleep state:

   If no alerts triggered, the workflow enters a sleep state. In npx workflow web, you'll see the workflow paused, waiting to resume. For local testing with sleep('10s'), it resumes after 10 seconds and runs round 2.

Commit

git add -A
git commit -m "feat(workflow): parallel resort steps with sleep re-check"
git push

Done-When

• Each resort is processed by its own "use step" function
• Steps run in parallel via Promise.all in the workflow
• sleep() pauses the workflow between evaluation rounds
• Workflow rechecks up to 3 times if no alerts trigger
• npx workflow web shows parallel steps and sleep states

Solution

import { sleep } from 'workflow';
import type { Alert } from '$lib/schemas/alert';
 
interface EvaluateInput {
  alerts: Alert[];
  recheckCount?: number;
}
 
interface AlertResult {
  alertId: string;
  resortId: string;
  triggered: boolean;
}
 
export default async function evaluateAlerts(
  { alerts, recheckCount = 0 }: EvaluateInput
) {
  "use workflow";
 
  const alertsByResort = Object.groupBy(alerts, (a) => a.resortId);
  const resortIds = Object.keys(alertsByResort);
 
  // One step per resort, all in parallel
  const results = await Promise.all(
    resortIds.map((resortId) =>
      evaluateResort(resortId, alertsByResort[resortId]!)
    )
  );
 
  const allResults = results.flat();
  const triggered = allResults.filter((r) => r.triggered);
 
  console.log('[Workflow] Round complete', {
    round: recheckCount + 1,
    evaluated: allResults.length,
    triggered: triggered.length
  });
 
  // If nothing triggered and we haven't hit the recheck limit, sleep and try again
  if (triggered.length === 0 && recheckCount < 3) {
    await sleep('30m');
    return evaluateAlerts({ alerts, recheckCount: recheckCount + 1 });
  }
 
  return {
    results: allResults,
    rounds: recheckCount + 1,
    triggered: triggered.length
  };
}
 
async function evaluateResort(
  resortId: string,
  alerts: Alert[]
): Promise<AlertResult[]> {
  "use step";
 
  const { getResort } = await import('$lib/data/resorts');
  const { fetchWeather } = await import('$lib/services/weather');
  const { evaluateCondition } = await import('$lib/services/alerts');
 
  const resort = getResort(resortId);
  if (!resort) return [];
 
  const weather = await fetchWeather(resort);
 
  return alerts.map((alert) => ({
    alertId: alert.id,
    resortId,
    triggered: evaluateCondition(alert.condition, weather)
  }));
}

Two big changes from 3.1:

Parallel steps. evaluateResort is its own "use step" function. The workflow dispatches one per resort via Promise.all. The Workflow DevKit runs them concurrently, and each has its own retry budget. If Mammoth's weather API times out, only Mammoth retries. Steamboat's result is already saved.

Sleep. sleep('30m') suspends the workflow without consuming any resources. No server running, no function billed. After 30 minutes, the platform wakes the workflow and it continues from where it left off. The recursive call to evaluateAlerts with an incremented recheckCount runs a fresh round of evaluation. Three re-checks max, then it returns whatever it has.

Troubleshooting

Advanced: Racing Steps Against a Timeout

Promise.race lets you set a deadline on a group of steps:

import { sleep } from 'workflow';
 
const results = await Promise.race([
  Promise.all(
    resortIds.map((id) => evaluateResort(id, alertsByResort[id]!))
  ),
  sleep('30s').then(() => 'timeout' as const)
]);
 
if (results === 'timeout') {
  console.warn('[Workflow] Evaluation timed out after 30s');
  return { results: [], timedOut: true };
}

The workflow returns whatever finishes first: the actual results or the timeout. Useful when you'd rather return partial data than wait indefinitely.