First Workflow

This tutorial connects two steps in a workflow. The workflow fetches a URL. Then the workflow saves a row in Postgres. Read First Standalone Job first.

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

Add the Handlers

Create main.go:

package main

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

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

type CrawlInput struct {
    URL string `json:"url"`
}

type FetchOutput struct {
    URL        string `json:"url"`
    StatusCode int    `json:"status_code"`
    Bytes      int    `json:"bytes"`
}

var FetchURL = cyclone.NewStep("fetch_url",
    func(ctx cyclone.Context, in CrawlInput) (FetchOutput, error) {
        req, err := http.NewRequestWithContext(ctx, http.MethodGet, in.URL, nil)
        if err != nil {
            return FetchOutput{}, err
        }

        resp, err := http.DefaultClient.Do(req)
        if err != nil {
            return FetchOutput{}, err
        }
        defer resp.Body.Close()

        body, err := io.ReadAll(resp.Body)
        if err != nil {
            return FetchOutput{}, err
        }

        return FetchOutput{
            URL:        in.URL,
            StatusCode: resp.StatusCode,
            Bytes:      len(body),
        }, nil
    })

type SaveReceipt struct {
    URL   string `json:"url"`
    Bytes int    `json:"bytes"`
}

var SavePage = cyclone.NewStep("save_page",
    func(ctx cyclone.Context, page FetchOutput) (SaveReceipt, error) {
        _, err := ctx.CompletionTx().Exec(ctx,
            `INSERT INTO crawl_results (run_id, node, url, status_code, bytes)
             VALUES ($1, $2, $3, $4, $5)
             ON CONFLICT (run_id, node) DO NOTHING`,
            ctx.RunID(), ctx.Node(), page.URL, page.StatusCode, page.Bytes)
        if err != nil {
            return SaveReceipt{}, err
        }
        return SaveReceipt{URL: page.URL, Bytes: page.Bytes}, nil
    })

FetchURL returns typed output. SavePage writes that output through ctx.CompletionTx(). The row then commits with the step completion.

Define the Workflow

Add this below the handlers:

var CrawlWorkflow = cyclone.DefineResult("crawl",
    func(b *cyclone.Builder, in CrawlInput) (*cyclone.Future[SaveReceipt], error) {
        page := FetchURL.Do(b, cyclone.Lit(in))
        return SavePage.Do(b, page), nil
    })

FetchURL.Do returns a *cyclone.Future[FetchOutput]. The future is the input of SavePage. This is the data edge. If SavePage took a different input type, the code would not compile.

The closure returns the future of SavePage. This selects the workflow result. The output of SavePage is the result of the run. The future must come from the same call of the closure. A nil future fails the submit before Cyclone inserts anything.

The builder declares work. The builder does not call the handlers. The workers start each node when its dependencies are complete. The workers start nodes with no edge between them in parallel.

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)
    }
    _, err = pool.Exec(ctx, `
        CREATE TABLE IF NOT EXISTS crawl_results (
            run_id text NOT NULL,
            node text NOT NULL,
            url text NOT NULL,
            status_code int NOT NULL,
            bytes int NOT NULL,
            created_at timestamptz NOT NULL DEFAULT now(),
            PRIMARY KEY (run_id, node)
        )`)
    if err != nil {
        log.Fatalf("create table: %v", err)
    }

    client, err := cyclone.New(pool,
        cyclone.WithHandlers(FetchURL, SavePage),
        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 := CrawlWorkflow.Submit(ctx, client, CrawlInput{URL: "https://example.com"})
    if err != nil {
        log.Fatalf("submit: %v", err)
    }
    receipt, err := run.Result(workers)

    stopWorkers()
    if startErr := <-done; startErr != nil && !errors.Is(startErr, context.Canceled) {
        log.Printf("workers: %v", startErr)
    }
    if err != nil {
        log.Printf("result: %v", err)
    } else {
        fmt.Printf("run %s saved %s: bytes=%d\n", run.ID(), receipt.URL, receipt.Bytes)
    }

    // 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 the receipt. Postgres has one row in crawl_results.

Submit returns a *cyclone.ResultRun[SaveReceipt]. run.Result waits until the whole run is terminal. Then run.Result returns the output of the result node. It waits for every branch, not only for the result node. A failed or cancelled run returns an error that matches cyclone.ErrRunNotCompleted.

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

The workflow handle has the same three ways to get the outcome as the step handle in First Standalone Job:

Method	Waits	Returns
run.Result(ctx)	Yes	The output of the result node. 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 the result node of a completed run. While a branch continues, the error matches cyclone.ErrNotCompleted.

If a branch with cyclone.OnDepFailure(cyclone.CancelBranch) cancels the result node, the run can complete with no result. Then Result and Output return an error that matches cyclone.ErrNotCompleted.

Open a Run From a Stored ID

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

receipt, err := CrawlWorkflow.Open(client, savedID).Result(ctx)

The run stored its result node when Cyclone inserted it. Open reads that node. Open does not call the closure. Each operation checks that the run belongs to the workflow crawl. A run of a different workflow returns an error that matches cyclone.ErrRunMismatch.

Read Another Node

Give the node an ID with cyclone.ID to read a node that is not the result. Then read the node with cyclone.OutputNode:

page := FetchURL.Do(b, cyclone.Lit(in), cyclone.ID("fetch"))

out, err := cyclone.OutputNode[FetchOutput](ctx, client, run.ID(), "fetch")

OutputNode reads one completed node in any run state.

A Workflow With No Result

cyclone.Define makes a workflow with no result. Its closure returns only an error. Submit returns a *cyclone.RunHandle. The handle has Wait, but no Result or Output.

What Happened

Cyclone inserted one run, two step rows, and one dependency edge in one transaction. The run row also holds the ID of the result node, save_page. The worker claimed fetch_url. The worker stored its JSON output. Then the worker made save_page available. save_page wrote to your table through ctx.CompletionTx(). Cyclone committed that write with the step completion.

That is the core loop.

Next

• Transactional Workflow
• Submit DAG Workflows
• Signals and Waits
• Testing Workflows