Skip to content

Long-Running Tasks

For tasks that process many items or run for extended periods, use steps to cache side effects and checkpoints to allow graceful interruption.

Steps for Side Effects

Use steps to save the results of operations with side effects - a retry skips steps whose results were saved:

const processLargeDataset = conductor.createTask(
  { name: "process-dataset" },
  { invocable: true },
  async (event, ctx) => {
    const items = await ctx.step("fetch-items", async () => {
      return await fetchThousandsOfItems();
    });

    // Each item gets its own step - skipped on retry once saved
    for (const item of items) {
      await ctx.step(`process-${item.id}`, async () => {
        await processItem(item); // API call, database write, etc.
      });
    }

    return { processed: items.length };
  }
);

What steps do:

  • Save results of operations with side effects
  • Skip work whose result was saved when the task retries or resumes
  • Provide automatic release points for shutdown and time windows

A step interrupted before its result was saved runs again, so external side effects such as charges and emails still need an idempotency key. See What Steps Guarantee.

Use steps when:

  • Operations have side effects (API calls, payments, emails, database writes)
  • You want to skip completed work on retry
  • You can create meaningful unique names for each operation

Checkpoints for Graceful Interruption

Use checkpoints to allow graceful shutdown in loops where steps don't make sense:

const deleteOldData = conductor.createTask(
  { name: "delete-old-data" },
  { invocable: true },
  async (event, ctx) => {
    let hasMore = true;

    while (hasMore) {
      // Delete batch of 1000 oldest records
      hasMore = await deleteBatch(1000);

      // Allow graceful shutdown or window boundary
      await ctx.checkpoint();
    }

    return { completed: true };
  }
);

What checkpoints do:

  1. Check if worker wants to shutdown (SIGTERM/SIGINT)
  2. Check if task is outside its time window
  3. If either is true: release task to resume later
  4. If both are false: continue execution

Use checkpoints when:

  • You can't enumerate items upfront (batch deletion of "oldest N records")
  • Continuous polling until condition is met
  • Long computations without meaningful intermediate step names
  • You want more frequent release opportunities than step boundaries provide

⚠️ Important: Checkpoints save nothing. A released task runs its handler again from the start, so the work since the last saved step repeats. It must be idempotent or track its own progress.

Graceful Shutdown

Checkpoints enable zero-downtime deployments. When a worker receives a shutdown signal:

  1. Stops accepting new work - No new tasks fetched from database
  2. Waits for safe release point - Current task runs until it hits a checkpoint, step boundary or task completion
  3. Releases task - Task returned to the queue; it runs again from the start of its handler, with its saved steps
  4. Shuts down cleanly - Worker process exits

Saved steps are kept, so completed steps do not run again after a deployment or scale-down.

ctx.signal is aborted when shutdown starts. A task that throws because it saw the signal is released, not failed, so it does not use up an attempt.

Time Windows

Time windows restrict tasks to specific hours. Tasks outside their window pause at checkpoint or step boundaries:

const nightlyProcessing = conductor.createTask(
  {
    name: "heavy-processing",
    window: ["22:00", "06:00"],  // 10 PM to 6 AM only
  },
  { invocable: true },
  async (event, ctx) => {
    for (const item of event.payload.items) {
      await ctx.step(`process-${item.id}`, async () => {
        await processItem(item);
      });

      // If window closes (6 AM arrives), task pauses here
      await ctx.checkpoint();
    }
  }
);

Window format: 

window: [start, end]  // ["HH:MM", "HH:MM"] in 24-hour format

Behavior:

Time Task State
Inside window Runs normally
Outside window Pauses at next checkpoint/step, resumes when window opens
Window opens Paused tasks automatically resume

Use cases:

1. Database maintenance during off-hours:

const dbMaintenance = conductor.createTask(
  {
    name: "vacuum-database",
    window: ["02:00", "04:00"],  // 2-4 AM only
  },
  { cron: "0 2 * * *" },
  async (event, ctx) => {
    await ctx.step("vacuum", async () => {
      await vacuumDatabase();
    });

    await ctx.step("analyze", async () => {
      await analyzeDatabase();
    });
  }
);

2. Business hours processing:

const customerSupport = conductor.createTask(
  {
    name: "process-support-tickets",
    window: ["09:00", "17:00"],  // 9 AM - 5 PM
  },
  { invocable: true },
  async (event, ctx) => {
    // Only processes during business hours
    await notifyAgent(event.payload.ticketId);
  }
);

3. Heavy processing during off-peak hours:

const bulkExport = conductor.createTask(
  {
    name: "bulk-export",
    window: ["20:00", "06:00"],  // 8 PM - 6 AM
  },
  { invocable: true },
  async (event, ctx) => {
    const records = await ctx.step("fetch-records", async () => {
      return await fetchAllRecords();
    });

    for (let i = 0; i < records.length; i += 1000) {
      const batch = records.slice(i, i + 1000);

      await ctx.step(`export-batch-${i}`, async () => {
        await exportBatch(batch);
      });

      // If 6 AM arrives, pause here and resume at 8 PM
      await ctx.checkpoint();
    }
  }
);

What's Next?