Cyclone

Appearance

Enqueue Standalone Jobs ​

How to enqueue single background jobs without a workflow graph.

Prerequisites ​

Use a cyclone.Client with your handlers registered. See Installation and Migrations.

What Are Standalone Jobs? ​

A standalone job is a one-node run. It has no DAG and no upstream dependency. It uses the same execution model, lease recovery, retry machinery, and observability as a workflow step. Use a standalone job for independent background work. Send a notification, process a file, or sync a record.

Enqueue a Step ​

go
runID, err := cyclone.Enqueue(ctx, client, FetchStep, FetchInput{URL: "https://example.com"})
if err != nil {
    return fmt.Errorf("enqueue: %w", err)
}
fmt.Println("job started:", runID)

Enqueue opens its own transaction. Enqueue inserts the run and the step. Enqueue commits. It then publishes a NOTIFY to wake sleeping workers. The publish is best-effort. Polling is the fallback. The returned runID identifies the run.

Enqueue an Action ​

go
runID, err := cyclone.EnqueueAction(ctx, client, SendEmailAction, EmailInput{
    To:      "user@example.com",
    Subject: "Your report is ready",
})

Enqueue Inside a Business Transaction ​

Use EnqueueTx when the standalone job must exist only if your business write commits:

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

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

runID, err := cyclone.EnqueueTx(ctx, client, tx, ProcessOrderStep,
    ProcessOrderInput{OrderID: orderID},
)
if err != nil {
    return err
}

return tx.Commit(ctx)

See Transactional Enqueueing for the full treatment.

Enqueue with a Specific Queue ​

go
cyclone.Enqueue(ctx, client, ReportStep, in, cyclone.WithQueue("bulk"))

See Queues and Priority.

Enqueue with Priority ​

go
// High priority — claimed before default-priority jobs
cyclone.Enqueue(ctx, client, UrgentStep, in, cyclone.WithPriority(10))

Enqueue with a Delay ​

go
// Process in 10 minutes
cyclone.Enqueue(ctx, client, RetryWebhookStep, in,
    cyclone.WithProcessIn(10*time.Minute),
)

// Process at a specific time
midnight := time.Now().UTC().Truncate(24*time.Hour).Add(24*time.Hour)
cyclone.Enqueue(ctx, client, NightlyReportStep, in,
    cyclone.WithProcessAt(midnight),
)

Cyclone inserts the step in the available state with a future scheduled_at. Workers skip it until the time arrives.

Enqueue with a Per-Attempt Timeout ​

go
cyclone.Enqueue(ctx, client, ImportStep, in,
    cyclone.WithTimeout(5*time.Minute),
)

See Timeouts, Cancellation, and Leases.

Enqueue with Custom Retry Behavior ​

go
// Override max attempts for this specific enqueue:
cyclone.Enqueue(ctx, client, WebhookStep, in,
    cyclone.WithMaxAttempts(3),
)

Or set defaults at the step definition:

go
var WebhookStep = cyclone.NewStep("webhook", webhookFn,
    cyclone.WithMaxAttempts(5),
    cyclone.WithBackoff(func(attempt int) time.Duration {
        return time.Duration(attempt) * 30 * time.Second
    }),
)

Enqueue with Retention ​

go
cyclone.Enqueue(ctx, client, QuickStep, in,
    cyclone.WithRetention(24*time.Hour), // delete terminal run after 24h
)

Enqueue with Uniqueness ​

go
runID, err := cyclone.Enqueue(ctx, client, SyncUserStep, SyncInput{UserID: 99},
    cyclone.WithUnique("user:99"),
)
if err != nil {
    var dup *cyclone.DuplicateError
    if errors.As(err, &dup) {
        fmt.Printf("already syncing as %s\n", dup.RunID)
        return nil
    }
    return err
}

See Unique Jobs.

Inspect Run Status ​

go
run, err := client.GetRun(ctx, runID)
if err != nil {
    return err
}
switch run.State {
case cyclone.RunCompleted:
    fmt.Println("done")
case cyclone.RunFailed:
    steps, _ := client.ListSteps(ctx, runID)
    for _, s := range steps {
        if s.State == cyclone.StepExhausted {
            fmt.Printf("failed: %s\n", s.Error)
        }
    }
case cyclone.RunRunning:
    fmt.Println("still running, attempt", /* check steps */)
case cyclone.RunCancelled:
    fmt.Println("cancelled")
}

Read Output After Completion ​

The Enqueue method on a step returns a *cyclone.ResultRun[O]. The package function cyclone.Enqueue returns a run ID. Result waits for the run. Result returns the output of the step:

go
run, err := FetchStep.Enqueue(ctx, client, FetchInput{URL: "https://example.com"})
if err != nil {
    return err
}
out, err := run.Result(ctx) // out is a FetchOutput

For a standalone job, the node ID is the kind of the step. run.Result reads that output. cyclone.OutputNode[FetchOutput](ctx, client, run.ID(), "fetch") reads the same output. See Read Step Output.

Retry/Discard Behavior ​

The default configuration:

  • Cyclone allows 5 attempts with exponential backoff (1s to ~1h with jitter).
  • Cyclone retries every error except an error wrapped with cyclone.Permanent.
  • After exhaustion, the step moves to discarded. The run moves to failed.

Reopen a discarded step with client.Retry(ctx, runID, stepKind) or client.RetryAll(ctx, runID).

See Retries and Permanent Failures.

Cyclone stores each run in Postgres.