Temporal Patterns

Appearance

Sliding Window Sliding Window

TLDR

Keep exactly windowSize child Workflows running at all times — each completion signal triggers the next record to start immediately. Use this when your record set is arbitrarily large, you need bounded concurrency to protect downstream systems, and you want higher throughput than a sequential Batch Iterator provides.

Overview ​

The Sliding Window pattern maintains a fixed-size pool of concurrently running child Workflows. As each child completes it signals the parent, which immediately starts a replacement — keeping the concurrency level constant and progressing at the rate of the fastest processor. Continue-as-New prevents the parent's history from growing without bound.

Problem ​

The Batch Iterator processes records sequentially — the overall throughput is limited by the slowest record in each page. The Fan-Out pattern starts all children at once, which can overwhelm downstream systems when the record set is large.

You need a way to process an arbitrarily large record set with bounded concurrency, maximum throughput within that bound, and protection against history bloat.

Solution ​

The parent Workflow starts exactly windowSize child Workflows simultaneously. Each child processes one record and, when finished, signals the parent with a completion notification. The parent maintains a count of completed children and starts a new child for the next record as soon as a slot becomes free.

Continue-as-New is called after the parent has started windowSize children. Because child Workflows have stable Workflow IDs and Continue-as-New preserves the parent's Workflow ID, children started by a previous run can still signal the current run.

The following describes each step in the diagram:

  1. The parent Workflow starts with a list of record IDs and a configured windowSize.
  2. It starts the first windowSize children concurrently, one per record. Each child reads the parent's Workflow ID from its own context, so it knows where to signal without being passed the ID explicitly.
  3. As each child completes, it sends a completion signal to the parent.
  4. The parent receives the signal, increments its completion counter, and starts the next child (the next record in the list).
  5. After starting windowSize children in total, the parent calls continueAsNew with the updated start index. The window slides forward without gaps because the parent's Workflow ID is preserved across runs.
  6. Children from previous runs that have not yet signalled will find the new run when they send the signal, because the parent Workflow ID remains the same.

Implementation ​

The following examples show how each SDK implements the Sliding Window pattern.

typescript
// workflows.ts
import {
  ApplicationFailure,
  ParentClosePolicy,
  condition,
  continueAsNew,
  defineSignal,
  getExternalWorkflowHandle,
  proxyActivities,
  setHandler,
  startChild,
  workflowInfo,
} from "@temporalio/workflow";
import type * as activities from "./activities";
import { TASK_QUEUE, WINDOW_SIZE } from "./shared";

const { processRecord } = proxyActivities<typeof activities>({
  startToCloseTimeout: "30 seconds",
});

export const completionSignal = defineSignal<[string]>("recordCompleted");

// Child workflow: processes one record and signals the parent on completion.
// The parent's workflow ID is read from context and is stable across the
// parent's continueAsNew runs.
export async function recordProcessorWorkflow(recordId: string): Promise<void> {
  await processRecord(recordId);
  // Ignore NOT_FOUND — the parent's final run may have already completed.
  try {
    const parent = getExternalWorkflowHandle(workflowInfo().parent!.workflowId);
    await parent.signal(completionSignal, recordId);
  } catch (err) {
    if (!(err instanceof ApplicationFailure && err.type === 'NOT_FOUND')) throw err;
  }
}

export async function slidingWindowWorkflow(input: SlidingWindowInput): Promise<number> {
  const {
    recordIds,
    windowSize = WINDOW_SIZE,
    startIndex = 0,
  } = input;
  const parentId = workflowInfo().workflowId;
  // Total records completed across all runs, carried over via continue-as-new.
  let totalProcessed = input.totalProcessed ?? 0;
  // Children started in this run; triggers continue-as-new once it hits windowSize.
  let dispatched = 0;
  // Live in-flight count: +1 per start, -1 per completion signal, carried across runs.
  let active = input.active ?? 0;

  setHandler(completionSignal, () => {
    active--;
    totalProcessed++;
  });

  // Slide the window: keep the window full, starting one child per free slot.
  // The first (windowSize - active) slots are already free, so those children
  // start without waiting; after that, each start waits for an in-flight child
  // to signal that its slot has freed.
  let nextIndex = startIndex;

  while (nextIndex < recordIds.length) {
    // Backpressure: block until the window has a free slot. When one is already
    // free (active < windowSize) this returns immediately; when full it waits
    // for a child's completion signal to decrement active via the handler.
    await condition(() => active < windowSize);
    await startChild(recordProcessorWorkflow, {
      args: [recordIds[nextIndex]],
      workflowId: `${parentId}/record-${recordIds[nextIndex]}`,
      taskQueue: TASK_QUEUE,
      parentClosePolicy: ParentClosePolicy.ABANDON,
    });
    nextIndex++;
    dispatched++;
    active++;

    // Once this run has filled the window with fresh children, continue-as-new
    // so history stays bounded. Carry active (the live in-flight count) so the
    // next run knows exactly how many children will still signal it.
    if (dispatched >= windowSize) {
      await continueAsNew<typeof slidingWindowWorkflow>({ recordIds, windowSize, startIndex: nextIndex, totalProcessed, active });
    }
  }

  // Wait for all remaining in-flight children to complete.
  await condition(() => active === 0);
  return totalProcessed;
}
python
# workflows.py
import asyncio
from datetime import timedelta
from temporalio import workflow
from temporalio.exceptions import ApplicationError
from temporalio.workflow import ParentClosePolicy, continue_as_new
from activities import process_record
from shared import TASK_QUEUE, WINDOW_SIZE

COMPLETION_SIGNAL = "recordCompleted"


@workflow.defn
class RecordProcessorWorkflow:
    # Child workflow: processes one record and signals the parent on completion.
    # The parent's workflow ID is read from context and is stable across the
    # parent's continue_as_new runs.
    @workflow.run
    async def run(self, record_id: str) -> None:
        await workflow.execute_activity(
            process_record,
            record_id,
            start_to_close_timeout=timedelta(seconds=30),
        )
        # Ignore NOT_FOUND — the parent's final run may have already completed.
        parent = workflow.get_external_workflow_handle(workflow.info().parent.workflow_id)
        try:
            await parent.signal(COMPLETION_SIGNAL, record_id)
        except ApplicationError as e:
            if "not found" not in str(e).lower():
                raise


@workflow.defn
class SlidingWindowWorkflow:
    def __init__(self) -> None:
        # Live in-flight count: +1 per start, -1 per completion signal, carried across runs.
        # Instance field (not a run() local) because the signal handler is a separate
        # method and completions can signal before run() starts.
        self._active = 0
        # Total records completed across all runs, carried over via continue-as-new.
        self._total_processed = 0

    @workflow.signal(name=COMPLETION_SIGNAL)
    def record_completed(self, record_id: str) -> None:
        self._active -= 1
        self._total_processed += 1

    @workflow.run
    async def run(self, input: SlidingWindowInput) -> int:
        # Use += so any completions that signal before run() starts are preserved.
        self._total_processed += input.total_processed
        self._active += input.active
        record_ids = input.record_ids
        window_size = input.window_size
        start_index = input.start_index
        parent_id = workflow.info().workflow_id
        next_index = start_index
        # Children started in this run; triggers continue-as-new once it hits window_size.
        dispatched = 0

        # Slide the window: keep the window full, starting one child per free slot.
        # The first (window_size - active) slots are already free, so those
        # children start without waiting; after that, each start waits for an
        # in-flight child to signal that its slot has freed.
        while next_index < len(record_ids):
            # Backpressure: block until the window has a free slot. Returns
            # immediately when one is free; when full it waits for a child's
            # completion signal to decrement _active via the handler.
            await workflow.wait_condition(lambda: self._active < window_size)
            await workflow.start_child_workflow(
                RecordProcessorWorkflow.run,
                record_ids[next_index],
                id=f"{parent_id}/record-{record_ids[next_index]}",
                task_queue=TASK_QUEUE,
                parent_close_policy=ParentClosePolicy.ABANDON,
            )
            next_index += 1
            dispatched += 1
            self._active += 1

            # Once this run has filled the window with fresh children, continue-as-new
            # so history stays bounded. Carry _active (the live in-flight count) so the
            # next run knows exactly how many children will still signal it.
            if dispatched >= window_size:
                continue_as_new(args=[SlidingWindowInput(
                    record_ids=record_ids,
                    window_size=window_size,
                    start_index=next_index,
                    total_processed=self._total_processed,
                    active=self._active,
                )])

        # Wait for all remaining in-flight children to complete.
        await workflow.wait_condition(lambda: self._active == 0)
        return self._total_processed
go
// workflows.go
package main

import (
	"strings"
	"time"

	enums "go.temporal.io/api/enums/v1"
	"go.temporal.io/sdk/workflow"
)

const CompletionSignal = "recordCompleted"

// RecordProcessorWorkflow processes one record and signals the parent on completion.
// The parent's workflow ID is read from context and is stable across the parent's
// ContinueAsNew runs.
func RecordProcessorWorkflow(ctx workflow.Context, recordID string) error {
	ao := workflow.ActivityOptions{StartToCloseTimeout: 30 * time.Second}
	ctx = workflow.WithActivityOptions(ctx, ao)

	if err := workflow.ExecuteActivity(ctx, ProcessRecord, recordID).Get(ctx, nil); err != nil {
		return err
	}

	// Ignore not-found — the parent's final run may have already completed.
	parentWorkflowID := workflow.GetInfo(ctx).ParentWorkflowExecution.ID
	err := workflow.SignalExternalWorkflow(ctx, parentWorkflowID, "", CompletionSignal, recordID).Get(ctx, nil)
	if err != nil && strings.Contains(err.Error(), "not found") {
		return nil
	}
	return err
}

func SlidingWindowWorkflow(ctx workflow.Context, input SlidingWindowInput) (int, error) {
	windowSize := input.WindowSize
	if windowSize <= 0 {
		windowSize = WindowSize
	}
	recordIDs := input.RecordIDs
	parentID := workflow.GetInfo(ctx).WorkflowExecution.ID

	completedCh := workflow.GetSignalChannel(ctx, CompletionSignal)
	nextIndex := input.StartIndex
	// Total records completed across all runs, carried over via ContinueAsNew.
	totalProcessed := input.TotalProcessed
	// Children started in this run; triggers ContinueAsNew once it hits windowSize.
	dispatched := 0
	// Live in-flight count: +1 per start, -1 per completion signal, carried across runs.
	active := input.Active

	startChild := func(recordID string) error {
		cwo := workflow.ChildWorkflowOptions{
			WorkflowID:        parentID + "/record-" + recordID,
			TaskQueue:         TaskQueue,
			ParentClosePolicy: enums.PARENT_CLOSE_POLICY_ABANDON,
		}
		future := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, cwo), RecordProcessorWorkflow, recordID)
		// Wait for the child to be started so the command is committed before any ContinueAsNew.
		return future.GetChildWorkflowExecution().Get(ctx, nil)
	}

	// Slide the window: keep the window full, starting one child per free slot.
	// The first (windowSize - active) slots are already free, so those children
	// start without waiting; after that, each start waits for an in-flight child
	// to signal that its slot has freed.
	for nextIndex < len(recordIDs) {
		// Backpressure: if the window is full, block on the completion channel
		// until an in-flight child signals, freeing a slot (active--). When a slot
		// is already free (active < windowSize), start without waiting.
		if active >= windowSize {
			completedCh.Receive(ctx, nil)
			totalProcessed++
			active--
		}
		if err := startChild(recordIDs[nextIndex]); err != nil {
			return totalProcessed, err
		}
		nextIndex++
		dispatched++
		active++

		// Once this run has filled the window with fresh children, continue-as-new
		// so history stays bounded. Carry active (the live in-flight count) so the
		// next run knows exactly how many children will still signal it.
		if dispatched >= windowSize {
			return 0, workflow.NewContinueAsNewError(ctx, SlidingWindowWorkflow, SlidingWindowInput{
				RecordIDs:      recordIDs,
				WindowSize:     windowSize,
				StartIndex:     nextIndex,
				TotalProcessed: totalProcessed,
				Active:         active,
			})
		}
	}

	// Drain all remaining in-flight children.
	for active > 0 {
		completedCh.Receive(ctx, nil)
		totalProcessed++
		active--
	}
	return totalProcessed, nil
}
java
// SlidingWindowWorkflow.java
import io.temporal.workflow.*;
import java.util.List;

@WorkflowInterface
public interface SlidingWindowWorkflow {
    @WorkflowMethod
    int run(Shared.SlidingWindowInput input);

    @SignalMethod
    void recordCompleted(String recordId);
}

// SlidingWindowWorkflowImpl.java
public class SlidingWindowWorkflowImpl implements SlidingWindowWorkflow {
    // Live in-flight count: +1 per start, -1 per completion signal, carried across runs.
    // Instance field (not a run() local) because the signal handler is a separate
    // method and completions can signal before run() starts.
    private int active = 0;
    // Total records completed across all runs, carried over via Continue-as-New.
    private int totalProcessed = 0;

    @Override
    public void recordCompleted(String recordId) {
        active--;
        totalProcessed++;
    }

    @Override
    public int run(Shared.SlidingWindowInput input) {
        // Use += so completions that signal before run() starts are preserved.
        this.totalProcessed += input.totalProcessed;
        this.active += input.active;
        int windowSize = input.windowSize > 0 ? input.windowSize : Shared.WINDOW_SIZE;
        List<String> recordIds = input.recordIds;
        String parentId = Workflow.getInfo().getWorkflowId();
        int nextIndex = input.startIndex;
        // Children started in this run; triggers Continue-as-New once it hits windowSize.
        int dispatched = 0;

        // Slide the window: keep the window full, starting one child per free slot.
        // The first (windowSize - active) slots are already free, so those children
        // start without waiting; after that, each start waits for an in-flight child
        // to signal that its slot has freed.
        while (nextIndex < recordIds.size()) {
            // Backpressure: block until the window has a free slot. Returns immediately
            // when one is free; when full it waits for a child's completion signal to
            // decrement active via the handler.
            Workflow.await(() -> active < windowSize);
            startChild(recordIds.get(nextIndex), parentId);
            nextIndex++;
            dispatched++;
            active++;

            // Once this run has filled the window with fresh children, continue-as-new
            // so history stays bounded. Carry active (the live in-flight count) so the
            // next run knows exactly how many children will still signal it.
            if (dispatched >= windowSize) {
                Workflow.newContinueAsNewStub(SlidingWindowWorkflow.class)
                    .run(new Shared.SlidingWindowInput(recordIds, windowSize, nextIndex, this.totalProcessed, active));
            }
        }

        // Wait for all remaining in-flight children to complete.
        Workflow.await(() -> active == 0);
        return this.totalProcessed;
    }

    private void startChild(String recordId, String parentId) {
        ChildWorkflowOptions opts = ChildWorkflowOptions.newBuilder()
            .setWorkflowId(parentId + "/record-" + recordId)
            .setTaskQueue(Shared.TASK_QUEUE)
            .setParentClosePolicy(ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON)
            .build();
        RecordProcessorWorkflow child = Workflow.newChildWorkflowStub(RecordProcessorWorkflow.class, opts);
        Async.procedure(child::run, recordId);
        // Wait until the child has actually started before the loop continues (and
        // before any Continue-as-New, which would otherwise race child startup).
        Workflow.getWorkflowExecution(child).get();
    }
}

Best Practices ​

  • Preserve the parent Workflow ID across Continue-as-New. The parent's Workflow ID is stable across continueAsNew runs — do not generate a new one. Children use signalExternalWorkflow with that ID, so they always reach the current run.
  • Use PARENT_CLOSE_POLICY_ABANDON on child Workflows. This lets children that were started by a previous run complete normally even after the parent has continued as new.
  • Size the window conservatively at first. Each in-flight child counts toward the 2,000 unfinished-actions limit for the parent. A window of 50–200 is a reasonable starting point depending on child duration and downstream capacity.
  • Pass only IDs (not full records) to child Workflows. Workflow inputs are stored in event history. Keep them small.
  • Carry minimal state into continueAsNew. Only pass windowSize, startIndex, and the record ID list (or a reference to it). Do not accumulate results in the parent — collect them out-of-band if needed.

Common Pitfalls ​

  • Losing signals across Continue-as-New. If a child signals before the parent's new run has registered the signal handler, the signal can be buffered and delivered correctly — Temporal buffers signals for existing Workflow IDs. However, ensure the signal handler is registered before any await, not conditionally.
  • Race between CAN and remaining signal draining. After continueAsNew, the new run must handle signals from children started by the previous run. Pass nextIndex (the next unstarted record) and active (the live in-flight count) to the new run so it knows how many carried-over children to expect signals from, without re-starting them.
  • Thundering herd on startup. Starting hundreds of children simultaneously causes a burst of Activity polls. Ramp up the window gradually or use the Batch Iterator if rate limiting is more important than throughput.

Temporal Design Patterns Catalog