Transactional Workflow

This tutorial submits a workflow in the same Postgres transaction as your business data.

The Two-Writes Problem

A typical background job enqueue is two writes:

1. Write your business data (INSERT into orders).
2. Enqueue the job (Enqueue(processOrderJob, ...)).

Those writes use separate transactions. A crash between them leaves one write committed. The crash loses the other write.

Why Enqueue-After-Commit Can Lose Work

// Scenario: crash between commit and enqueue
tx.Commit(ctx)        // ← order row committed
// CRASH HERE         // ← process dies
client.Enqueue(...)   // ← never runs

The order row exists in the database. The enqueue did not create the processing job. The order is stuck. No one knows to process it.

Why Enqueue-Before-Commit Can Create Phantom Work

// Scenario: rollback after enqueue
client.Enqueue(...)   // ← job enqueued
// ... something fails
tx.Rollback(ctx)      // ← order row rolled back

The job exists. The job will start. The order row does not exist. The handler processes a non-existent order.

Solution: SubmitTx

SubmitTx inserts the workflow run in the same transaction as your business data:

package main

import (
    "context"
    "log"

    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/wafer-labs/cyclone"
    "your-app/workflows"
)

func CreateOrder(ctx context.Context, pool *pgxpool.Pool, client *cyclone.Client, req OrderRequest) (int, cyclone.RunID, error) {
    tx, err := pool.Begin(ctx)
    if err != nil {
        return 0, "", err
    }
    defer tx.Rollback(ctx)

    // Business write:
    var orderID int
    if err := tx.QueryRow(ctx,
        `INSERT INTO orders (user_id, total, status)
         VALUES ($1, $2, 'pending')
         RETURNING id`,
        req.UserID, req.Total,
    ).Scan(&orderID); err != nil {
        return 0, "", err
    }

    // Workflow submitted in the same transaction:
    run, err := workflows.FulfillOrderWorkflow.SubmitTx(ctx, client, tx,
        workflows.FulfillInput{OrderID: orderID},
    )
    if err != nil {
        return 0, "", err
    }

    // Both commit, or neither commits.
    if err := tx.Commit(ctx); err != nil {
        return 0, "", err
    }

    return orderID, run.ID(), nil
}

If tx.Commit fails, the database keeps neither the order row nor the workflow run. If you call tx.Rollback, the database keeps neither the order row nor the workflow run.

Commit Before You Read

The handle from SubmitTx exists before the commit. The run does not exist before the commit. Handle operations use the client pool. The operations return cyclone.ErrRunNotFound until the commit. Do not wait inside the transaction. Commit first. Then read:

if err := tx.Commit(ctx); err != nil {
    return err
}
status, err := run.Wait(ctx)

Cyclone sends no notification inside your transaction. Workers find the committed run within one poll interval. See Transactional Enqueueing to wake them at once.

Solution: EnqueueTx for Standalone Jobs

The same pattern with EnqueueTx:

tx, err := pool.Begin(ctx)
if err != nil { return err }
defer tx.Rollback(ctx)

// Create a payment record:
var paymentID int
if err := tx.QueryRow(ctx,
    `INSERT INTO payments (user_id, amount) VALUES ($1, $2) RETURNING id`,
    userID, amount,
).Scan(&paymentID); err != nil {
    return err
}

// Enqueue charge processing in the same tx:
charge, err := ChargeStep.EnqueueTx(ctx, client, tx, ChargeInput{PaymentID: paymentID})
if err != nil { return err }

if err := tx.Commit(ctx); err != nil { return err }
// The run exists now. Result waits for the charge.
receipt, err := charge.Result(ctx)

ctx.CompletionTx() Inside a Handler

Inside a handler, ctx.CompletionTx() is the completion transaction. This transaction commits the state change of the step. Use it to write business mutations atomically with the step's completion:

var ProcessOrder = cyclone.NewAction("process_order",
    func(ctx cyclone.Context, in ProcessInput) error {
        _, err := ctx.CompletionTx().Exec(ctx, `
            UPDATE orders
            SET status = 'processing', processed_at = now(), cyclone_run_id = $1
            WHERE id = $2 AND status = 'pending'`,
            ctx.RunID(), in.OrderID)
        return err
    },
)

This UPDATE commits with the step's completion. If the worker crashes before commit, both the UPDATE and the step completion roll back. The WHERE status = 'pending' clause makes a retry a no-op after a successful write.

The Idempotency Key: (runID, node)

When you write through ctx.CompletionTx(), use (ctx.RunID(), ctx.Node()) as the deduplication key:

_, err := ctx.CompletionTx().Exec(ctx, `
    INSERT INTO order_events (run_id, node, order_id, event_type, created_at)
    VALUES ($1, $2, $3, 'processed', now())
    ON CONFLICT (run_id, node) DO NOTHING`,
    ctx.RunID(), ctx.Node(), in.OrderID)

The (run_id, node) pair is stable across all retries and lease reclaims for the same step. If you include attempt, every retry looks like a new event.

Simulate a Crash with cyclonetest

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

    var ProcessAction = cyclone.NewAction("process",
        func(ctx cyclone.Context, in ProcessInput) error {
            _, err := ctx.CompletionTx().Exec(ctx, `
                INSERT INTO events (run_id, node, data)
                VALUES ($1, $2, $3)
                ON CONFLICT (run_id, node) DO NOTHING`,
                ctx.RunID(), ctx.Node(), in.Data)
            return err
        },
    )
    env.Mux().Register(ProcessAction)

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

    run, err := ProcessAction.Enqueue(context.Background(), env.Client(), ProcessInput{Data: "test"})
    require.NoError(t, err)
    <-crashed
    status, err := run.Wait(context.Background())
    require.NoError(t, err)
    require.Equal(t, cyclone.RunCompleted, status.State)

    // Verify the event was written exactly once (ON CONFLICT deduplicated the second run).
    var count int
    env.Pool().QueryRow(context.Background(),
        `SELECT count(*) FROM events WHERE run_id = $1`, string(run.ID()),
    ).Scan(&count)
    require.Equal(t, 1, count)
}

Verify Atomicity

Either both the business write and the step completion commit, or neither does.

After a crash (KillAfter), the database does not have the business effect. After recovery, the database has the business effect exactly once:

1. Submit the job.
2. Arm KillAfter to crash after the ctx.CompletionTx() write but before commit.
3. Wait for the crash.
4. Query the business table. The row must not be there (rolled back).
5. Wait for recovery (run.Wait).
6. Query the business table. The row must be there exactly once.

Next

• Transactional Enqueueing: full guide
• Exactly-Once Effects: ctx.CompletionTx() in depth
• Idempotency: the at-least-once contract