Testing with cyclonetest

cyclonetest starts Cyclone against real Postgres. Most queue bugs live in transactions, leases, and state transitions, not in mocks.

Create an Environment

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

    env.Mux().Register(SendWelcomeEmail)

    runID, err := cyclone.EnqueueAction(context.Background(), env.Client(),
        SendWelcomeEmail,
        WelcomeEmailInput{UserID: 42},
    )
    require.NoError(t, err)

    env.RunUntilComplete(runID)
    env.RequireCompleted(runID)
}

By default, cyclonetest.New starts a Postgres container with Testcontainers. Set CYCLONETEST_DSN to reuse an existing database instead:

export CYCLONETEST_DSN='postgres://postgres@127.0.0.1:5432/postgres?sslmode=disable'
go test ./...

Each test gets its own schema. t.Parallel() is safe.

Register Before Workers Start

Register handlers on env.Mux() before the client starts. Register first. Then call env.Client().

env := cyclonetest.New(t)

env.Mux().Register(FetchPage)
env.Mux().Register(SavePage)

runID, err := cyclone.Submit(ctx, env.Client(), CrawlWorkflow, CrawlInput{
    URL: "https://example.com",
})

You can also give the handlers to the client. They survive a restart:

env := cyclonetest.New(t,
    cyclonetest.WithClientOptions(cyclone.WithHandlers(FetchPage, SavePage)))

Use WithAutoStart(false) when the test needs queued work to sit idle before workers claim it.

env := cyclonetest.New(t, cyclonetest.WithAutoStart(false))
env.Mux().Register(SyncAccount)

runID, _ := cyclone.EnqueueAction(ctx, env.Client(), SyncAccount, in)

env.Start()
env.RunUntilComplete(runID)

Stop and restart

env.Stop() cancels the worker context. It waits until the goroutines of the client drain. You can then assert on the final state. Those asserts do not race the goroutines. The harness reports a failure from Client.Start on t. It does not drop that failure.

env.Start() after a Stop() builds a fresh cyclone.Client. The new client uses the same pool, registry, and options. From then on, env.Client() returns that client. cyclone.Client is single-use. Its Start refuses a second call. The Env replaces the client. The harness then still has workers.

Calling Start twice with no Stop between the calls fails the test. That call is not a restart. It would leave the workers of the first client active. Two fleets would then use the same schema.

The restart keeps the same registry. Handlers therefore survive the restart. Workflow schedules do not. ScheduleWorkflow stores its builder on the Client. After a restart, you must call it again on the new env.Client(). A restarted worker process does the same call at boot:

env.Stop()
env.Start()
// Handlers registered on env.Mux() still resolve. This does not:
_, _ = cyclone.ScheduleWorkflow(ctx, env.Client(), "0 * * * *", MyWorkflow, in,
    cyclone.WithScheduleID("nightly"))

Step schedules and action schedules stay valid after a restart. They fire from the schedule row. They do not use a registered builder.

env.Start()
// ... first execution ...
env.Stop()          // workers drained; the step's result is already recorded

env.Start()         // fresh client, same pool and handlers
// ... the work runs again ...

Use this sequence to model a worker restart. Stop the workers in the middle of a run. Assert the rows that the database holds. Start a new process. Watch recovery take control.

Wait Without Sleeps

Do not use time.Sleep in tests. Let the harness drive the run.

env.RunUntilComplete(runID)
env.RequireCompleted(runID)

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

RunUntilIdle waits until no work is claimable or in progress. It is useful for schedules and for parked waits.

Assert a Result

cyclonetest.RequireResult waits for the result of a step handle. It compares that result with the value that you expect:

env := cyclonetest.New(t, cyclonetest.WithClientOptions(cyclone.WithHandlers(Double)))

run, err := Double.Enqueue(ctx, env.Client(), 21)
require.NoError(t, err)
cyclonetest.RequireResult(t, env, run, 42)

It waits for the Env timeout at most. If the run does not complete, the check fails the test. It prints the run and its steps. If the result is different, the check fails the test. It prints the two values.

Assert Output

Select the workflow result with cyclone.DefineResult. Assert that result with RequireResult:

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

run, err := ComputeWorkflow.Submit(ctx, env.Client(), 21)
require.NoError(t, err)
cyclonetest.RequireResult(t, env, run, 42)

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

Stub a Handler

Use StubHandler when the test cares about workflow wiring and not about the real side effect.

var sent []string

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

The stub must match the registered handler signature. Actions return error. Steps return (O, error).

Force Failures

FailNext makes the next n executions of a kind return an error.

env.FailNext("send_email", 2, errors.New("SMTP unavailable"))

runID, _ := cyclone.EnqueueAction(ctx, env.Client(), SendEmail, in)
env.RunUntilComplete(runID)

step := env.Step(runID, "send_email")
require.Equal(t, 2, step.Attempts) // the two failures; the success that followed adds none

Use this for retry behavior. This is clearer than a custom handler with counters in every test.

Simulate a Crash

KillAfter kills the worker path after the handler returns. The completion transaction has not committed yet. The lease then lapses. Cyclone executes the step again.

env := cyclonetest.New(t, cyclonetest.WithLeaseTTL(200*time.Millisecond))
env.Mux().Register(ChargeCard)

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

runID, _ := cyclone.EnqueueAction(ctx, env.Client(), ChargeCard, in)
<-crashed

env.RunUntilComplete(runID)
env.RequireCompleted(runID)

This test checks whether a ctx.CompletionTx() write commits exactly once.

Test Signals and Timers

For WaitSignal, park the run. Deliver the signal. Then finish the run.

env.RunUntilIdle()
env.RequireStepState(runID, "wait_signal:paid", cyclone.StepWaitingSignal)

err := env.Client().Signal(ctx, runID, "paid", Receipt{PaymentID: "pi_123"})
require.NoError(t, err)

env.RunUntilComplete(runID)
env.RequireCompleted(runID)

For WaitTimer or a bounded WaitSignal, move the stored deadline into the past.

env.RunUntilIdle()
env.RequireStepState(runID, "wait_timer", cyclone.StepWaitingTimer)

env.ElapseWait(runID, "wait_timer")
env.RunUntilComplete(runID)

To assert the SignalResult payload, read the wait node with cyclone.OutputNode. The ID of a WaitSignal node is wait_signal: plus the signal name:

got, err := cyclone.OutputNode[cyclone.SignalResult[Receipt]](ctx, env.Client(), runID, "wait_signal:paid")
require.NoError(t, err)
require.Equal(t, Receipt{PaymentID: "pi_123"}, got.Value)

Test Child Runs

A child run executes on the workers of the Env. It executes like any other run. Register the child workflow and its steps. env.ChildRun(runID, node) waits until the child node has a child run. It then returns that run ID:

env := cyclonetest.New(t, cyclonetest.WithClientOptions(
    cyclone.WithHandlers(price, pack, label, email, Fulfillment)))

run, err := Checkout.Submit(ctx, env.Client(), "order-1")
require.NoError(t, err)
childID := env.ChildRun(run.ID(), "fulfillment")
env.RequireCompleted(env.RunUntilComplete(childID).ID)
cyclonetest.RequireResult(t, env, run, "sent")

AdvanceTime and ElapseWait also move the WithMaxWait limit of a waiting child node. A failed assertion prints the parent of a child run. It also prints the child run of each child node.

RunUntilIdle also waits in two child cases. It waits while a child node waits for its child run to start. It waits for the engine after that child run ends. If no worker of the Env registered that workflow, RunUntilIdle waits until the timeout.

Assert Transactional Submits

RequireSubmitted checks that SubmitTx recorded a workflow. You give it the workflow and the input. It decodes that input from the TxRecorder:

rec := &cyclonetest.TxRecorder{}

// In your code that calls MyWorkflow.SubmitTx(ctx, client, tx, input):
run, err := MyWorkflow.SubmitTx(ctx, client, rec, MyInput{UserID: 42})

// In the test:
sub := cyclonetest.RequireSubmitted(t, rec, MyWorkflow, MyInput{UserID: 42})
// sub.RunID and sub.Input are available for further assertions

RequireSubmitted fails t with a clear message when the recorder has no matching submit. The returned Submission carries the run ID and the raw input bytes.

RequireSubmitted takes a workflow from cyclone.Define. For a workflow from cyclone.DefineResult, call rec.Submitted(wf.Name()). Decode sub.Input.

workflow.SubmitOnceTx records the same way. This includes a workflow from cyclone.DefineResult. sub.SubmitKey holds its key. The recorder does not enforce the key. Each call records a submission. Each call returns created true.

RequireEnqueued is the counterpart for EnqueueTx. Its Submission carries every resolved option. That set includes scheduled enqueues:

rec := cyclonetest.NewTxRecorder()
_, _ = cyclone.EnqueueTx(ctx, client, rec, SendReminder, in,
    cyclone.WithQueue("bulk"),
    cyclone.WithPriority(7),
    cyclone.WithProcessIn(24*time.Hour),
    cyclone.WithSequence("acct:42"),
)

sub := cyclonetest.RequireEnqueued(t, rec, "send_reminder", in)
// sub.Queue == "bulk", sub.Priority == 7,
// sub.ProcessIn == 24h, sub.SequenceKey == "acct:42"

ProcessAt and ProcessIn are mutually exclusive. The last option wins. At most one of the two is set.

A handler records the same way when it calls EnqueueCompletion, EnqueueActionCompletion, or SubmitCompletion. Recording happens when the context comes from cyclonetest.NewContext. The completion transaction must be a recorder. The client still validates the kind. Pass a client that declares that kind:

rec := cyclonetest.NewTxRecorder()
ctx := cyclonetest.NewContext(cyclonetest.WithCompletionTx(rec))

// handleModelCall is the function passed to cyclone.NewStep. It calls
// cyclone.EnqueueCompletion(ctx, producer, ReduceSession, ...).
_, err := handleModelCall(ctx, CallInput{SessionID: "s-1"})
require.NoError(t, err)

sub := cyclonetest.RequireEnqueued(t, rec, "session.reduce", ReduceInput{SessionID: "s-1"})
// sub.SequenceKey == "s-1"

Outside a worker, Cyclone publishes nothing. There is no completion to publish after. A context whose CompletionTx is not a pgx.Tx makes the completion enqueue return an error.

Advance Time Without Sleeping

AdvanceTime shifts Postgres clocks forward for schedules, backoff windows, and wait timers. It does this with no real wall-clock delay:

env := cyclonetest.New(t)

// Register a schedule and let it fire.
env.RunUntilIdle()

// Fast-forward 30 days to the next scheduled window:
env.AdvanceTime(30 * 24 * time.Hour)
env.RunUntilIdle()

It rewinds scheduled_at on available, waiting_signal, and waiting_timer steps. It advances next_run_at on enabled schedules. Pair it with RunUntilComplete or RunUntilIdle. You can then test a multi-day retry and timeout policy in milliseconds. AdvanceTime does not accelerate the clock of the worker. The caller still drives the workers by hand.

It also rewinds the scheduled_at of waiting_child steps. The WithMaxWait limit of a child node then elapses.

For a paused run, AdvanceTime also rewinds three values. It rewinds the scheduled_at of its paused steps. It rewinds deadline_at. It rewinds paused_at. A timer or a deadline therefore keeps its remaining time across a pause. ElapseWait on a wait of a paused run removes the time that the wait has left. The wait lapses at Resume.

Force a schedule due now

ForceDue(scheduleID) sets the next_run_at of a schedule to now(). The next scanner tick fires that schedule. This is a one-time fake-time policy. Use it when the test needs the slot due immediately. You do not give a time span. It is the single-schedule form of AdvanceTime. When the test asserts a specific scheduled_for, capture the slot value before any fire. Keep the scanner off for that capture. Or set WithAutoStart(false).

env := cyclonetest.New(t, cyclonetest.WithAutoStart(false))
// ... register the schedule ...
env.ForceDue(schedID)
env.Start()                       // scanner races the test's triggers on the due slot
env.RunUntilIdle()

Pair it with RunUntilIdle to drain the firing run. See Schedules for pause, resume, and trigger flows that use ForceDue.

Useful Options

cyclonetest.New(t,
    cyclonetest.WithWorkers(4),
    cyclonetest.WithLeaseTTL(200*time.Millisecond),
    cyclonetest.WithTimeout(5*time.Second),
    cyclonetest.WithClientOptions(cyclone.WithMaxSignalWait(time.Hour)),
)

Use a short lease TTL only in a test that needs recovery to happen quickly.

Next

• Testing Workflows
• Transactional Enqueueing
• Signals and Waits
• Schedules
• Sequences
• Retries