Quickstart

This page builds one program. The program applies the schema, starts two workers, enqueues one step, and prints the result.

Requirements

• Go 1.26.3 or later
• Postgres 14 or later
• A database that this process can reach

The connection default must be READ COMMITTED. That is the Postgres default. New returns an error for a different default.

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-quickstart
cd cyclone-quickstart
go mod init cyclone-quickstart
go get github.com/wafer-labs/cyclone
go get github.com/jackc/pgx/v5

Write the program

Create main.go.

package main

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

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

var double = cyclone.NewStep("double", func(_ cyclone.Context, n int) (int, error) {
    return n * 2, nil
})

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(double), cyclone.WithWorkers(2))
    if err != nil {
        log.Fatalf("client: %v", err)
    }

    workers, stopWorkers := context.WithCancel(ctx)
    defer stopWorkers()
    done := make(chan error, 1)
    go func() {
        done <- client.Start(workers)
        stopWorkers()
    }()

    run, err := double.Enqueue(ctx, client, 21)
    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 doubled 21 to %d\n", run.ID(), out)

    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 doubled 21 to 42.

What the program does

1. MaxConns is 8. A client needs at least the completion slots plus six connections. The default slot count is the worker count.
2. Migrate applies the schema. New does not apply it.
3. WithHandlers registers double. A worker accepts a step only when the worker has that handler.
4. Start blocks until you cancel its context.
5. Enqueue inserts one run and returns a handle.
6. Result waits until the run is terminal. Then Result returns the output.
7. Shutdown runs before pool.Close.

Rules that prevent the first failures

Call Migrate before New.

Set MaxConns to at least the completion-slot count plus 6. With two workers and the default slot count, the minimum is 8.

Keep the pool default at READ COMMITTED.

Cancel the context of Start before you call Shutdown. Close the pool last.

A cancel of the context that you give to Result stops the wait. The cancel does not cancel the durable run.

New(pool) with no handlers returns a producer. Start on a producer returns ErrProducerOnly. A producer can enqueue work. A producer cannot call a handler.

Next

• Concepts defines the words these pages use.
• First standalone job adds delay, a unique key, and an action.
• First workflow connects two steps and stores a row with the step completion.