Skip to main content

MAKE WORKFLOWS IDEMPOTENT

This guide shows you how to add guard expressions to workflow steps so that re-running a workflow skips steps whose work is already done.

Prerequisites

  • Swamp installed and on your PATH
  • An initialised repo (swamp repo init)
  • A workflow with at least one step that produces data
  • Familiarity with workflow definitions and CEL expressions

Guard a step with data truthiness

Add a guard field to any step. The value is a CEL expression wrapped in ${{ }}. When the expression evaluates to a truthy value, the step is skipped.

If your step produces data via a model method, check whether that data already exists:

steps:
  - name: provision
    guard: "${{ data.latest('provisioner', 'result') }}"
    task:
      type: model_method
      modelIdOrName: provisioner
      methodName: execute
      inputs:
        run: "echo 'provisioned'"

On the first run, data.latest('provisioner', 'result') returns a falsy value (no data exists yet), so the step executes:

 system │ Starting workflow provision-stack

   main │ start
   main │ step provision · provisioner · execute · start
   main │ provisioned
   main │ done provision

 system │ Completed workflow provision-stack succeeded

On the second run, the data exists and the guard evaluates truthy. The step is skipped:

   main │ skipped (guarded) · guard: data.latest('provisioner', 'result')
 system │ Starting workflow provision-stack

   main │ start
   main │ completed

 system │ Completed workflow provision-stack succeeded

Guard with a value comparison

When data existence alone is not sufficient, compare a specific attribute against an expected value:

steps:
  - name: configure-vpc
    guard: "${{ data.latest('vpc', 'result').attributes.cidr == '10.0.0.0/16' }}"
    task:
      type: model_method
      modelIdOrName: vpc
      methodName: create
      inputs:
        cidr: "10.0.0.0/16"

The step re-runs if the VPC's CIDR has drifted from the expected value.

Guard with a method call

If you need to check live state rather than stored data, call a model method inside the guard expression:

steps:
  - name: provision-dns
    guard: "${{ model.method('dns-checker', 'verify', {'domain': 'app.example.com'}).stdout }}"
    task:
      type: model_method
      modelIdOrName: dns-provisioner
      methodName: create
      inputs:
        domain: app.example.com

Guard forEach iterations independently

When a step uses forEach, the guard evaluates once per iteration. Each iteration can be skipped or executed independently based on its own guard result. Use self.{item} to vary the guard per iteration:

steps:
  - name: provision-${{ self.region }}
    guard: "${{ data.latest('provisioner-' + self.region, 'result') }}"
    forEach:
      item: region
      in: "${{ inputs.regions }}"
    task:
      type: model_method
      modelType: command/shell
      modelName: provisioner-${{ self.region }}
      methodName: execute
      inputs:
        run: "echo 'provisioned ${{ self.region }}'"

On the first run, both iterations execute:

   main │ start
   main │ step provision-us-east-1 · provisioner-us-east-1 · execute · start
   main │ step provision-eu-west-1 · provisioner-eu-west-1 · execute · start
   main │ provisioned eu-west-1
   main │ provisioned us-east-1
   main │ done provision-eu-west-1
   main │ done provision-us-east-1
   main │ completed

 system │ Completed workflow provision-stack succeeded

On the second run, both iterations are skipped:

   main │ skipped (guarded) · guard: data.latest('provisioner-us-east-1', 'result')
   main │ skipped (guarded) · guard: data.latest('provisioner-eu-west-1', 'result')
 system │ Starting workflow provision-stack

   main │ start
   main │ completed

 system │ Completed workflow provision-stack succeeded

If you add a new region to the input list, only the new region's iteration executes — the existing regions remain guarded.

Inspect guarded skips in JSON output

Run with --json to see structured skip events:

swamp workflow run provision-stack --json

Each guarded step emits a line:

{ "step": "provision", "job": "main", "status": "skipped", "reason": "guarded" }

The reason field distinguishes guarded skips ("guarded") from dependency skips ("dependency").

Recover a failed run

When a workflow fails mid-execution, use resume --from to re-enter the DAG at the failed step. Guards on earlier steps prevent them from re-running.

Given a three-step workflow where each step guards on its output data:

jobs:
  - name: main
    steps:
      - name: fetch
        guard: "${{ data.latest('fetcher', 'result') }}"
        task:
          type: model_method
          modelIdOrName: fetcher
          methodName: run
      - name: transform
        guard: "${{ data.latest('transformer', 'result') }}"
        task:
          type: model_method
          modelIdOrName: transformer
          methodName: run
        dependsOn:
          - step: fetch
            condition:
              type: succeeded
      - name: load
        guard: "${{ data.latest('loader', 'result') }}"
        task:
          type: model_method
          modelIdOrName: loader
          methodName: run
        dependsOn:
          - step: transform
            condition:
              type: succeeded

If transform fails on the first run, fetch has already completed and written its data. Fix the cause of the failure, then resume from the failed step:

swamp workflow resume etl-pipeline --from transform

The fetch step is not re-run — it completed before the failure. The transform step re-runs because its guard sees no completion data from the failed attempt. The load step proceeds after transform succeeds.

Recover forEach iterations

For forEach steps, --from targets the template name (the unexpanded step name from the YAML). The platform re-expands all iterations and each iteration's guard decides whether it runs:

steps:
  - name: deploy-${{ self.env }}
    guard: "${{ data.latest('deployer-' + self.env, 'result') }}"
    forEach:
      item: env
      in: "${{ inputs.environments }}"
    task:
      type: model_method
      modelType: command/shell
      modelName: deployer-${{ self.env }}
      methodName: execute
      inputs:
        run: "deploy --target ${{ self.env }}"

If deploy-staging succeeds but deploy-production fails:

swamp workflow resume deploy-pipeline --from deploy-$\{\{ self.env \}\}

The staging iteration is skipped (its guard sees existing data). The production iteration re-runs.

Reference

← PreviousVerify Infrastructure with Assert StepsNext →CEL Expressions