Testing Workflows

This tutorial tests workflows and jobs against real Postgres using cyclonetest.

Install and Import cyclonetest

go get github.com/wafer-labs/cyclone/cyclonetest

import (
    "context"
    "testing"

    "github.com/wafer-labs/cyclone"
    "github.com/wafer-labs/cyclone/cyclonetest"
    "github.com/stretchr/testify/require"
)

You do not set up a database yourself. cyclonetest.New starts a postgres:18-alpine container with Testcontainers. Set CYCLONETEST_DSN to reuse an existing instance.

Create an Isolated Test Environment

func TestMyWorkflow(t *testing.T) {
    env := cyclonetest.New(t)
}

Each call creates its own schema. Parallel tests are safe. cyclonetest removes the container and the schema when the test ends.

Register Handlers and Submit

Give the handlers to the test client with cyclone.WithHandlers. You can also register them on env.Mux() after New.

func TestFetchAndSave(t *testing.T) {
    env := cyclonetest.New(t,
        cyclonetest.WithClientOptions(cyclone.WithHandlers(FetchStep, SaveAction)))

    run, err := CrawlWorkflow.Submit(context.Background(), env.Client(),
        CrawlInput{URL: "https://httpbin.org/json"})
    require.NoError(t, err)
    require.NotEmpty(t, run.ID())
}

Wait Without Sleeps

env.RunUntilComplete(run.ID()) polls until the run reaches a terminal state: RunCompleted, RunFailed, or RunCancelled. You do not need time.Sleep:

status := env.RunUntilComplete(run.ID())

The default timeout is 30 seconds. Override it with cyclonetest.WithTimeout:

env := cyclonetest.New(t, cyclonetest.WithTimeout(5*time.Second))

Assert Run State

env.RunUntilComplete(run.ID())
env.RequireCompleted(run.ID()) // fails test with step snapshot if not RunCompleted

Assert Step State

env.RequireStepState(run.ID(), "save", cyclone.StepCompleted)

step := env.Step(run.ID(), "save")
require.Equal(t, cyclone.StepCompleted, step.State)
require.Equal(t, 0, step.Attempts) // Attempts counts failed executions, so a first-try success is 0

Assert the Result

func TestDoubleResult(t *testing.T) {
    env := cyclonetest.New(t,
        cyclonetest.WithClientOptions(cyclone.WithHandlers(DoubleStep)))

    run, err := DoubleStep.Enqueue(context.Background(), env.Client(), 21)
    require.NoError(t, err)

    cyclonetest.RequireResult(t, env, run, 42)
}

RequireResult waits for the run, up to the Env timeout. If the run does not complete, RequireResult fails the test. In that case, RequireResult prints the run and step states. If the result is not 42, RequireResult fails the test. In that case, RequireResult prints the two values.

RequireResult also checks the result of a workflow from cyclone.DefineResult:

var DoubleWorkflow = cyclone.DefineResult("double",
    func(b *cyclone.Builder, in int) (*cyclone.Future[int], error) {
        return DoubleStep.Do(b, cyclone.Lit(in)), nil
    })

run, err := DoubleWorkflow.Submit(context.Background(), env.Client(), 21)
require.NoError(t, err)
cyclonetest.RequireResult(t, env, run, 42)

To check a node that is not the result, give the node an ID with cyclone.ID. Read it with cyclone.OutputNode after the run completes:

var QuadrupleWorkflow = cyclone.DefineResult("quadruple",
    func(b *cyclone.Builder, in int) (*cyclone.Future[int], error) {
        first := DoubleStep.Do(b, cyclone.Lit(in), cyclone.ID("first"))
        return DoubleStep.Do(b, first), nil
    })

run, err := QuadrupleWorkflow.Submit(context.Background(), env.Client(), 21)
require.NoError(t, err)
cyclonetest.RequireResult(t, env, run, 84)

out, err := cyclone.OutputNode[int](context.Background(), env.Client(), run.ID(), "first")
require.NoError(t, err)
require.Equal(t, 42, out)

Stub Handlers

Replace a real handler with a test double:

var sent []string
env.StubHandler("send_email", func(ctx cyclone.Context, in EmailInput) error {
    sent = append(sent, in.To)
    return nil
})

run, _ := OrderWorkflow.Submit(context.Background(), env.Client(), orderInput)
env.RunUntilComplete(run.ID())

require.ElementsMatch(t, []string{"customer@example.com"}, sent)

The stub must match the handler's Go signature for the kind.

Inject Failures

Use env.FailNext to make the next N executions return an error:

func TestRetryBehavior(t *testing.T) {
    env := cyclonetest.New(t, cyclonetest.WithWorkers(2))
    env.Mux().Register(WebhookStep)

    // Fail the first 2 attempts.
    env.FailNext("webhook", 2, errors.New("connection refused"))

    run, _ := WebhookStep.Enqueue(context.Background(), env.Client(), in)
    env.RunUntilComplete(run.ID())
    env.RequireCompleted(run.ID())

    step := env.Step(run.ID(), "webhook")
    require.Equal(t, 2, step.Attempts) // the 2 failures; the success that followed adds none
}

Simulate Lease Expiry

env.ExpireLease moves the step's lease_expiry to the past. This triggers recovery:

func TestLeaseRecovery(t *testing.T) {
    env := cyclonetest.New(t, cyclonetest.WithLeaseTTL(100*time.Millisecond))
    env.Mux().Register(FetchStep)

    run, _ := FetchStep.Enqueue(context.Background(), env.Client(), in)

    // Wait for the step to start running, then expire the lease.
    time.Sleep(50 * time.Millisecond) // brief pause for claim
    env.ExpireLease(run.ID(), "fetch")

    env.RunUntilComplete(run.ID())
    env.RequireCompleted(run.ID())

    step := env.Step(run.ID(), "fetch")
    // A reclaim does not increment the attempt counter.
    require.Equal(t, 0, step.Attempts)
}

The recovery sweep reclaims the step. Then the step starts again.

Simulate Crash Before Commit

env.KillAfter kills the worker goroutine after the handler body completes, before the completion transaction commits:

func TestCrashIdempotency(t *testing.T) {
    env := cyclonetest.New(t, cyclonetest.WithLeaseTTL(200*time.Millisecond))

    var insertCount int
    var ChargeAction = cyclone.NewAction("charge",
        func(ctx cyclone.Context, in ChargeInput) error {
            // Write through ctx.CompletionTx() — rolled back on crash.
            _, err := ctx.CompletionTx().Exec(ctx, `
                INSERT INTO charges (run_id, node, amount)
                VALUES ($1, $2, $3)
                ON CONFLICT (run_id, node) DO NOTHING`,
                ctx.RunID(), ctx.Node(), in.Amount)
            return err
        },
    )
    env.Mux().Register(ChargeAction)

    crashed := make(chan struct{})
    env.KillAfter("charge", func() {
        close(crashed)
    })

    run, _ := ChargeAction.Enqueue(context.Background(), env.Client(), ChargeInput{Amount: 100})
    <-crashed
    env.RunUntilComplete(run.ID())
    env.RequireCompleted(run.ID())

    // The charge row should exist exactly once.
    env.Pool().QueryRow(context.Background(),
        `SELECT count(*) FROM charges WHERE run_id = $1`, string(run.ID()),
    ).Scan(&insertCount)
    require.Equal(t, 1, insertCount)
}

Parallel Tests

Tests that use cyclonetest.New are safe in parallel. Each test gets its own isolated schema:

func TestA(t *testing.T) { t.Parallel(); env := cyclonetest.New(t); /* ... */ }
func TestB(t *testing.T) { t.Parallel(); env := cyclonetest.New(t); /* ... */ }

Next

• Testing with cyclonetest: all cyclonetest features
• Idempotency: why you need crash-safe tests