Cyclone

Appearance

Scheduled Job ​

This tutorial registers a durable recurring job that fires on a cron expression.

Define the Step ​

go
package jobs

import (
    "fmt"
    "time"

    "github.com/wafer-labs/cyclone"
)

type DailyReportInput struct {
    Recipient string `json:"recipient"`
}

type DailyReportOutput struct {
    GeneratedAt time.Time `json:"generated_at"`
    RowCount    int       `json:"row_count"`
}

var DailyReportStep = cyclone.NewStep("daily_report",
    func(ctx cyclone.Context, in DailyReportInput) (DailyReportOutput, error) {
        // Generate and send the report.
        count, err := generateReport(ctx, in.Recipient)
        if err != nil {
            return DailyReportOutput{}, err
        }
        return DailyReportOutput{
            GeneratedAt: time.Now(),
            RowCount:    count,
        }, nil
    },
    cyclone.WithMaxAttempts(3),
    cyclone.WithTimeout(10*time.Minute),
    cyclone.WithQueue("reports"),
)

Register the Schedule ​

Call Schedule on the step at startup. WithScheduleID makes the call idempotent. Call Schedule each time the process starts:

go
package main

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

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

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

    pool, _ := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
    defer pool.Close()

    cyclone.Migrate(ctx, pool)

    client, _ := cyclone.New(pool,
        cyclone.WithHandlers(jobs.DailyReportStep),
        cyclone.WithWorkerQueues(map[string]int{
            "default": 4,
            "reports": 1,
        }),
    )

    schedID, err := jobs.DailyReportStep.Schedule(ctx, client,
        "0 8 * * 1-5", // 08:00 UTC, Monday to Friday
        jobs.DailyReportInput{Recipient: "team@example.com"},
        cyclone.WithScheduleID("weekday-daily-report"),
    )
    if err != nil {
        log.Fatalf("schedule: %v", err)
    }
    log.Printf("schedule registered: %s\n", schedID)

    go client.Start(ctx)

    <-ctx.Done()
    shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    client.Shutdown(shutdownCtx)
}

Cron Syntax ​

The spec argument uses standard 5-field cron syntax:

Examples:

text
"0 8 * * 1-5"    // weekdays at 08:00 UTC
"*/15 * * * *"   // every 15 minutes
"0 0 1 * *"      // first of every month at midnight
"@daily"         // once per day at midnight
"@hourly"        // once per hour
"@every 30m"     // every 30 minutes

Timezones ​

Use WithScheduleLocation for non-UTC timezones. Cyclone handles DST correctly:

go
loc, err := time.LoadLocation("America/New_York")
if err != nil {
    log.Fatal(err)
}

jobs.DailyReportStep.Schedule(ctx, client,
    "0 9 * * 1-5", // 09:00 Eastern (adjusts for DST)
    jobs.DailyReportInput{Recipient: "nyc-team@example.com"},
    cyclone.WithScheduleID("weekday-report-nyc"),
    cyclone.WithScheduleLocation(loc),
)

Idempotent Startup Registration ​

If you call Schedule again with the same WithScheduleID, the call updates the schedule row. The update is an upsert. The call does not create a duplicate. One example is a call after a deploy. You can register the schedule on every startup.

go
// To change the cron spec, change the call. The upsert updates the row.
jobs.DailyReportStep.Schedule(ctx, client,
    "0 7 * * 1-5", // changed from 08:00 to 07:00
    jobs.DailyReportInput{Recipient: "team@example.com"},
    cyclone.WithScheduleID("weekday-daily-report"), // same ID → upsert
)

Force Due in Tests ​

Use env.ForceDue(schedID) to move next_run_at to now(). The scanner then fires the schedule immediately in tests:

go
func TestDailyReportFires(t *testing.T) {
    env := cyclonetest.New(t,
        cyclonetest.WithClientOptions(cyclone.WithHandlers(jobs.DailyReportStep)))

    schedID, err := jobs.DailyReportStep.Schedule(context.Background(), env.Client(),
        "@daily",
        jobs.DailyReportInput{Recipient: "test@example.com"},
        cyclone.WithScheduleID("test-daily-report"),
    )
    require.NoError(t, err)

    // Immediately trigger the schedule:
    env.ForceDue(schedID)

    // Wait for the fired run to complete:
    env.RunUntilIdle()
    // All runs are now terminal; check the most recent run.
}

Inspect Schedules ​

sql
-- All schedules and their next fire time:
SELECT id, kind, cron_spec, location, next_run_at, consecutive_failures, enabled
FROM cyclone_schedules
ORDER BY next_run_at;

-- Schedules with failures:
SELECT id, kind, consecutive_failures
FROM cyclone_schedules
WHERE consecutive_failures > 0;

-- Recent fires:
SELECT sf.schedule_id, sf.scheduled_for, r.state, r.finished_at
FROM cyclone_schedule_fires sf
JOIN cyclone_runs r ON r.id = sf.run_id
ORDER BY sf.scheduled_for DESC
LIMIT 10;

Next ​

Cyclone stores each run in Postgres.