First Standalone Job

This tutorial enqueues one typed step and reads its result. One enqueue creates one run with one node. Use it for account syncs, webhook delivery, file imports, and other work that does not need a workflow graph yet.

Start Postgres

docker run -d --name cyclone-pg \
  -e POSTGRES_HOST_AUTH_METHOD=trust \
  -p 5432:5432 \
  postgres:17-alpine

export DATABASE_URL='postgres://postgres@127.0.0.1:5432/postgres?sslmode=disable'

Create a Module

mkdir cyclone-job-demo
cd cyclone-job-demo
go mod init cyclone-job-demo
go get github.com/wafer-labs/cyclone
go get github.com/jackc/pgx/v5

Define a Step

Create main.go:

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "os/signal"
    "strings"
    "syscall"
    "time"

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

type CountInput struct {
    Text string `json:"text"`
}

type CountOutput struct {
    Words int `json:"words"`
}

var CountWords = cyclone.NewStep("count_words",
    func(ctx cyclone.Context, in CountInput) (CountOutput, error) {
        return CountOutput{Words: len(strings.Fields(in.Text))}, nil
    },
    cyclone.WithMaxAttempts(5),
)

A step is a Go function with a typed input and a typed output. Cyclone stores both as JSON.

Run the Program

Add main:

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()

    cfg, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatalf("parse config: %v", err)
    }
    // Two workers use two completion slots. Cyclone reserves six more connections.
    cfg.MaxConns = 8
    pool, err := pgxpool.NewWithConfig(ctx, cfg)
    if err != nil {
        log.Fatalf("connect: %v", err)
    }
    defer pool.Close()

    if err := cyclone.Migrate(ctx, pool); err != nil {
        log.Fatalf("migrate: %v", err)
    }

    client, err := cyclone.New(pool,
        cyclone.WithHandlers(CountWords),
        cyclone.WithWorkers(2),
    )
    if err != nil {
        log.Fatalf("client: %v", err)
    }

    // A failed Start cancels workers, thus the wait below stops too.
    workers, stopWorkers := context.WithCancel(ctx)
    defer stopWorkers()
    done := make(chan error, 1)
    go func() {
        done <- client.Start(workers)
        stopWorkers()
    }()

    run, err := CountWords.Enqueue(ctx, client, CountInput{Text: "durable work in Postgres"})
    if err != nil {
        log.Fatalf("enqueue: %v", err)
    }
    out, err := run.Result(workers)

    stopWorkers()
    if startErr := <-done; startErr != nil && !errors.Is(startErr, context.Canceled) {
        log.Fatalf("workers: %v", startErr)
    }
    if err != nil {
        log.Fatalf("result: %v", err)
    }
    fmt.Printf("run %s counted %d words\n", run.ID(), out.Words)

    // Shut down before the pool closes.
    shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    if err := client.Shutdown(shutdownCtx); err != nil {
        log.Printf("shutdown: %v", err)
    }
}

Start the program:

go run .

The program prints the run ID and counted 4 words.

The program does these steps:

1. It sets the pool size. A client needs MaxConns of at least the completion slots plus six. By default, the completion slots are equal to the worker count.
2. It applies the migrations. New does not apply them.
3. It gives the step to cyclone.New with cyclone.WithHandlers. The client accepts only the kinds that it has registered.
4. It starts the workers in a goroutine. Start blocks until its context ends.
5. It enqueues the step. Enqueue returns a run handle.
6. Result waits until the run is terminal. Then Result returns the typed output.
7. It stops the workers. It calls Shutdown before the pool closes.

Cyclone needs the READ COMMITTED isolation level as the connection default. That is the Postgres default. New reads the isolation level. New returns an error for a different level.

Result, Wait, and Output

A run handle has three ways to get the outcome of a run:

Method	Waits	Returns
run.Result(ctx)	Yes	The output. A failed or cancelled run returns an error that matches cyclone.ErrRunNotCompleted.
run.Wait(ctx)	Yes	The terminal RunStatus. A failed or cancelled run is not an error. Examine State.
run.Output(ctx)	No	The output of a completed run. A run that is not complete returns an error.

If the context of Result or Wait ends, the call returns the context error. The run continues.

Open a Run From a Stored ID

Store run.ID(). Open the run again in a different process. The process does not need the handler:

out, err := CountWords.Open(client, savedID).Result(ctx)

Each operation of this handle checks that the run is a standalone count_words run. A different run returns an error that matches cyclone.ErrRunMismatch. To examine any run without its definition, use client.OpenRun(savedID).

Jobs With No Output

An action returns only error. Its handle has Wait, but no Result:

var SendEmail = cyclone.NewAction("send_email",
    func(ctx cyclone.Context, in EmailInput) error {
        return deliver(ctx, in)
    },
)

run, err := SendEmail.Enqueue(ctx, client, EmailInput{To: "user@example.com"})
if err != nil {
    return err
}
status, err := run.Wait(ctx)

Give the action to cyclone.WithHandlers with the step.

Add a Delay

Delay the job with WithProcessIn:

run, err := SendEmail.Enqueue(ctx, client, in, cyclone.WithProcessIn(10*time.Minute))

Use WithProcessAt to start the standalone job at a specific time:

run, err := SendEmail.Enqueue(ctx, client, in, cyclone.WithProcessAt(sendAt))

Deduplicate Active Work

Use WithUnique when a second copy of the job is not necessary. Use WithUnique when a second copy of the job causes problems:

_, err := SendEmail.Enqueue(ctx, client, in,
    cyclone.WithUnique("welcome-email:user:42"),
)
var dup *cyclone.DuplicateError
if errors.As(err, &dup) {
    fmt.Println("already queued as", dup.RunID)
}

The key is active-only. After the first run completes, a new enqueue with the same key can succeed.

Check Status

status, err := run.Status(ctx)
if err != nil {
    return err
}
fmt.Println("run state:", status.State)

steps, err := run.Steps(ctx)
if err != nil {
    return err
}
for _, step := range steps {
    fmt.Println(step.Node, step.State, step.Attempts, step.Error)
}

Shut Down in a Real Service

A service continues until SIGTERM. Cancel the context of Start. Then call Shutdown with a new timeout context. Close the pool last.

See Graceful Shutdown for the full pattern.

Next

• First Workflow
• Unique Jobs
• Retries