Transactional Enqueueing

Use transactional enqueueing when a standalone job must exist only if your business write commits. An order is the usual case. If you save the order row, fulfillment must start. If the order rolls back, fulfillment must not exist.

The Bug This Avoids

This code looks harmless. It has a crash window:

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

_, err := cyclone.Enqueue(ctx, client, ProcessOrder, ProcessOrderInput{OrderID: orderID})

If the process dies after Commit and before Enqueue, the order exists. No standalone job starts.

An enqueue before the commit has the opposite bug. A rollback leaves a standalone job that points at data that never committed.

Submit a Workflow in the Transaction

workflow.SubmitTx writes the run, the steps, and the dependency edges in your transaction. The package function cyclone.SubmitTx does the same.

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

var orderID int64
err = tx.QueryRow(ctx,
    `INSERT INTO orders (user_id, total_cents)
     VALUES ($1, $2)
     RETURNING id`,
    userID, totalCents,
).Scan(&orderID)
if err != nil {
    return err
}

run, err := FulfillOrder.SubmitTx(ctx, client, tx,
    FulfillInput{OrderID: orderID},
)
if err != nil {
    return err
}

return tx.Commit(ctx)

The order and the workflow commit together. If Commit fails, or the transaction rolls back, the Cyclone rows roll back with the order.

The handle from SubmitTx exists before the commit. The run does not exist yet. A handle operation uses the client pool. It returns cyclone.ErrRunNotFound until the commit. Commit first. Then read the run, or wait:

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

Enqueue One Job in the Transaction

For one standalone job, use EnqueueTx on the step or the action.

_, err = SendWelcomeEmail.EnqueueTx(ctx, client, tx,
    WelcomeEmailInput{UserID: userID},
)

_, err = SendReceipt.EnqueueTx(ctx, client, tx,
    ReceiptInput{OrderID: orderID},
)

For a batch, use EnqueueManyTx. The handles have the order of the inputs:

runs, err := SendNotification.EnqueueManyTx(ctx, client, tx, inputs)

Every inserted run shares your transaction with the application data.

Who Owns the Transaction

You own the transaction. Cyclone does not commit it. Cyclone does not roll it back. Cyclone does not begin a nested transaction for SubmitTx or EnqueueTx. TrackWakeups does not change that. The transaction that it returns commits as yours does. It rolls back as yours does.

This is the normal shape:

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

// Write app rows.
// Submit or enqueue Cyclone work.

return tx.Commit(ctx)

Do not pass a transaction to Cyclone and then commit it inside a step handler. Use ctx.CompletionTx() for a handler write. That call returns the completion transaction for that step.

Notifications

Cyclone never issues a Postgres notification inside your transaction. A NOTIFY in that transaction can make the commit fail. That happens when the Postgres global notification queue is exhausted. Cyclone never observes your commit. By default, workers claim the new work within one poll interval of your commit. The default poll interval is at most one second.

To wake workers at the commit, track the transaction. Publish after the commit:

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

if _, err := SendWelcomeEmail.EnqueueTx(ctx, client, ttx,
    WelcomeEmailInput{UserID: userID},
); err != nil {
    return err
}

if err := ttx.Commit(ctx); err != nil {
    return err
}
client.PublishWakeups(ctx, wakeups)

TrackWakeups returns a transaction. Every caller-transaction form accepts that transaction. The methods are EnqueueTx, EnqueueManyTx, and SubmitTx. The package functions are EnqueueTx, EnqueueActionTx, EnqueueManyTx, EnqueueActionManyTx, and SubmitTx. Each form records one wakeup for the queue of work that workers can claim at commit. A standalone job with WithProcessAt or WithProcessIn in the future records nothing. A wait node of a workflow records nothing. SubmitTx records the queue of every root node that is available at commit. PublishWakeups sends one wakeup for each distinct queue.

• Publish only after a successful commit. A Rollback on the tracked transaction discards the recorded wakeups. A failed Commit on it discards them too. A savepoint that you open with its Begin, and then roll back, discards them too. A publish after those calls sends nothing. The wrapper does not see a rollback of the original tx. A publish after that rollback sends only a spurious wakeup. Workers treat that wakeup as an early poll.
• PublishWakeups returns nothing. Cyclone logs a failed publication. Polling repairs that failure. That failure is not a failed commit.
• It clears the Wakeups. A second call publishes nothing.
• The call uses the client's WithNotifyCoalesce window. Shut a coalescing client down before you close its pool.
• The call does not depend on WithNotify. WithNotify controls only listening. A producer-only client with WithNotify(false) still wakes workers in other processes.

A notification is only a wakeup. Polling is the correctness floor. A missing notification delays work by a short time. It does not lose the work.

Enqueue From a Handler

A step handler uses the completion variants for follow-up work. Use them also for a kind that its worker does not serve. Cyclone inserts the run in the completion transaction of the handler. The run exists when that step completion commits:

var ReduceSession = cyclone.RegisterKind[ReduceInput, struct{}]("session.reduce",
    cyclone.WithQueue("reduce"),
)

// producer declares ReduceSession with RegisterKind and never starts.
var CallModel = cyclone.NewStep("model.call", func(ctx cyclone.Context, in CallInput) (CallOutput, error) {
    out, err := callModel(ctx, in)
    if err != nil {
        return CallOutput{}, err
    }
    _, err = cyclone.EnqueueCompletion(ctx, producer, ReduceSession,
        ReduceInput{SessionID: in.SessionID, Result: out.ID},
        cyclone.WithSequence(in.SessionID),
    )
    return out, err
}, cyclone.WithQueue("model"))

EnqueueCompletion, EnqueueActionCompletion, and SubmitCompletion take the ctx of the handler and a client:

• The first call opens the completion transaction in the same way as ctx.CompletionTx(). A business write on ctx.CompletionTx() commits with the enqueued runs. They commit with the completed transition of the step.
• Cyclone rolls those runs back with the rest of the completion in these cases. The cases are a handler error, a Snooze, an oversized output, a timeout, and a cancelled run. A completion that Cyclone rejects because a reclaim took the lease is also a case. A worker that dies before the commit leaves nothing behind.
• The client validates the kind, the payload limits, and the options. The client can be a producer-only client that declares the kind with RegisterKind. Then the worker that does the handler does not register that kind. The worker does not serve that kind. The run always goes to the database of the worker that does the handler.
• Options work as they do for Enqueue. A WithUnique duplicate returns a *DuplicateError through a savepoint. The completion transaction stays usable.
• The worker that does the handler publishes the queue wakeup after the commit. It publishes that wakeup with the completion wakeup. Both use its WithNotifyCoalesce window. A standalone job for a later time publishes nothing.
• Call these functions from the handler goroutine before it returns. Like the completion transaction, they are not safe for concurrent use.

In a handler unit test, give cyclonetest.NewContext a cyclonetest.TxRecorder as its completion transaction. The recorder records these enqueues. It does not write them. See Testing with cyclonetest.

Testing the Rollback Path

Test the rollback path. Most transactional bugs appear only when you test that path.

tx, err := env.Pool().Begin(ctx)
require.NoError(t, err)

run, err := SendWelcomeEmail.EnqueueTx(ctx, env.Client(), tx,
    WelcomeEmailInput{UserID: 42},
)
require.NoError(t, err)

require.NoError(t, tx.Rollback(ctx))

_, err = run.Status(ctx)
require.ErrorIs(t, err, cyclone.ErrRunNotFound)

runs, err := env.Client().ListRuns(ctx, cyclone.RunQuery{})
require.NoError(t, err)
require.Empty(t, runs.Runs)

Next

• Exactly-Once Effects
• Enqueue Standalone Jobs
• Submit DAG Workflows