chronos

package module
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 12, 2026 License: MIT Imports: 17 Imported by: 0

README

chronos-go

CI Go Reference

A Redis-backed distributed task queue and scheduler for Go, with a type-safe generic API.

chronos-go is a modern alternative to asynq (now in maintenance mode). It keeps what people loved about asynq — a simple enqueue → handle model, at-least-once delivery, crash recovery — and fixes its biggest gaps: a distributed scheduler that runs each job once across many instances, unbounded stream/dead-letter growth, and unique-lock expiry during long processing.

Status: v0.x — usable and covered by tests against real Redis, but the API may still evolve before v1.0.0.

Highlights

  • Type-safe, generic API — no interface{} payloads, no manual json.Unmarshal. Define a task type, register a Handler[T].
  • Redis Streams + ZSETs — immediate work rides a Streams consumer group; delayed/retry/archived tasks live in sorted sets. Cluster-safe (hash-tagged keys).
  • Reliable — automatic retries with exponential backoff + jitter, crash recovery (XAUTOCLAIM), dead-letter with an OnDeadLetter hook.
  • Distributed scheduler — interval & cron jobs. A Redis leader election ensures only one instance enqueues each trigger; a deterministic task ID fences duplicates during leader hand-off.
  • Delayed execution & de-duplication — WithProcessIn / WithProcessAt, and WithUnique to collapse duplicate work.
  • Weighted priority queues — queue weights are honored via smooth weighted round-robin (no starvation), or set StrictPriority to always drain higher-weight queues first.
  • Chains & groups — run tasks in sequence (NewChain) or fan out in parallel and fire a callback when every member succeeds (NewGroup); a failure stops the flow, and re-running its dead-letter resumes it.
  • Heartbeat — refreshes the lease and unique lock of in-flight tasks, so a long-running task is neither reclaimed nor loses its lock mid-processing.
  • Self-cleaning — a janitor trims dead-lettered and retained-completed tasks by age and count, so Redis memory stays bounded.
  • Observable — an Inspector API + a chronos CLI, a runnable tour, and a Prometheus + Grafana stack in contrib/prometheus.

Requirements

  • Go 1.26+
  • Redis 6.2+ (uses XAUTOCLAIM)

Install

go get github.com/kenshin579/chronos-go

Quick start

Define a task, enqueue it with a client, and process it with a server.

package main

import (
	"context"
	"log"

	"github.com/redis/go-redis/v9"
	"github.com/kenshin579/chronos-go"
)

// A task type: any struct with a stable Kind() (value receiver).
type EmailArgs struct {
	UserID string `json:"user_id"`
	Body   string `json:"body"`
}

func (EmailArgs) Kind() string { return "email:send" }

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	ctx := context.Background()

	// --- worker side ---
	mux := chronos.NewMux()
	chronos.AddHandler(mux, func(ctx context.Context, t *chronos.Task[EmailArgs]) error {
		// t.Args is strongly typed — no casting, no json.Unmarshal.
		log.Printf("sending to %s: %s", t.Args.UserID, t.Args.Body)
		return nil
	})

	srv := chronos.NewServer(rdb, chronos.ServerConfig{
		Queues:      map[string]int{"default": 1},
		Concurrency: 10,
	})
	if err := srv.Start(ctx, mux); err != nil {
		log.Fatal(err)
	}
	defer srv.Shutdown(context.Background())

	// --- producer side ---
	client := chronos.NewClient(rdb)
	defer client.Close()

	if _, err := chronos.Enqueue(ctx, client, EmailArgs{UserID: "u1", Body: "hi"}); err != nil {
		log.Fatal(err)
	}

	select {} // keep the worker running
}
Enqueue options
chronos.Enqueue(ctx, client, EmailArgs{...},
	chronos.WithQueue("critical"),          // route to a queue
	chronos.WithMaxRetry(5),                 // retry budget (default 25)
	chronos.WithProcessIn(30*time.Minute),   // run later (delayed)
	chronos.WithUnique(10*time.Minute),      // dedup identical (kind+payload) tasks
	chronos.WithDeadLetterDiscard(),         // drop instead of archive on exhaustion
	chronos.WithRetention(24*time.Hour),     // keep the completed task for inspection
)
Handler outcomes
  • return nil → success (acked and removed — or kept for WithRetention for later inspection).
  • return an error → retried with backoff until MaxRetry is exhausted, then dead-lettered.
  • return chronos.SkipRetry(err) → dead-lettered immediately (permanent error).
  • panic → recovered, treated as a retryable error.

Set an OnDeadLetter hook on ServerConfig to alert on / inspect exhausted tasks.

Queue priority

ServerConfig.Queues maps queue name → weight. While every queue has work, a queue with weight 6 is dequeued about 6× as often as a queue with weight 1 (smooth weighted round-robin — lower-weight queues never starve). When the queue chosen for a round is empty, that round falls through to the highest-weight queue that does have work, so an idle high-priority queue never holds up lower ones. Weights <= 0 are treated as 1.

srv := chronos.NewServer(rdb, chronos.ServerConfig{
	Queues: map[string]int{
		"critical": 6,
		"default":  3,
		"low":      1,
	},
	// StrictPriority: true, // always drain critical first, then default, then low
})

With StrictPriority: true, higher-weight queues are always drained first; a lower-weight queue runs only while every higher one is empty.

Chains

Run tasks strictly in sequence — each link is enqueued only when the previous one succeeds:

info, err := chronos.NewChain().
	Then(EncodeArgs{VideoID: "v1"}).
	Then(ThumbnailArgs{VideoID: "v1"}, chronos.WithQueue("low")).
	Then(NotifyArgs{UserID: "u1"}, chronos.WithRetention(time.Hour)).
	Enqueue(ctx, client)
  • Per-link options: queue, retries, retention, delay (WithProcessIn on a link delays it relative to its predecessor's completion). WithTaskID and WithUnique are rejected inside chains.
  • Failure stops the chain. When a link exhausts its retries and is dead-lettered, its successors wait inside the dead-letter (ChainPending in the Inspector shows how many). Re-run it (chronos task run ...) after fixing the cause and the chain resumes from that point.
  • Handlers must stay idempotent (at-least-once). Successors are enqueued at most once while their record exists; a predecessor redelivered after its successor already finished (and was not retained) can recreate it — the standard at-least-once caveat. Per-link WithRetention closes that window for its duration.
  • Each link carries its remaining tail in its message, so very long chains grow the message size — keep chains reasonably short.

Groups

Fan out members in parallel and run a callback once all of them succeed:

info, err := chronos.NewGroup().
	Add(ResizeArgs{File: "a.jpg"}).
	Add(ResizeArgs{File: "b.jpg"}, chronos.WithQueue("low")).
	OnComplete(ReportArgs{Batch: "b1"}, chronos.WithRetention(time.Hour)).
	Enqueue(ctx, client)
  • Members run on any queues with per-member options; the callback fires exactly once while its record exists (idempotent tracking — an at-least-once redelivery cannot double-fire it).
  • A failed member parks the group. Its dead-letter shows the group (GroupID, remaining members via GroupPending in the Inspector); re-run it and, once it succeeds, the callback fires if it was the last one.
  • Group state lives for 7 days and every member completion renews it, so only a truly abandoned group (a member deleted, or dead-lettered and never re-run) expires — the callback then never fires. Members cannot be scheduled beyond that window, and WithDeadLetterDiscard is rejected for members (both would strand the group).
  • Enqueueing members is not atomic: if it fails midway, already-enqueued members still run, but the callback can never fire early.
  • Not yet composable with chains (no group-as-chain-link); callback payloads are fixed at build time (no result passing).

Scheduling (interval & cron)

Register periodic jobs on a Scheduler. Every instance may call Start; only the elected leader enqueues, so a job fires once cluster-wide.

sched := chronos.NewScheduler(rdb, chronos.SchedulerConfig{})

// every 30s (interval must be >= 1s)
chronos.RegisterInterval(sched, 30*time.Second, HealthCheckArgs{})

// standard 5-field cron
chronos.RegisterCron(sched, "0 0 * * *", DailyReportArgs{})

sched.Start(ctx)          // safe on every instance
defer sched.Shutdown(ctx)

Missed triggers (after a leader-election gap) are handled by a per-job WithMisfirePolicy — MisfireSkip (default) or MisfireFireOnce.

Observability

chronos-go is a headless library, so it ships tools to see what it is doing:

  • Runnable tour — every feature end-to-end, printed as it happens: 
    go run ./examples/tour
  • CLI — inspect and administer queues/tasks: 
    go run ./cmd/chronos queue ls
go run ./cmd/chronos task ls default archived
go run ./cmd/chronos task ls default completed  # inspect retained successes
go run ./cmd/chronos task run default <id>   # re-run a dead-letter
go run ./cmd/chronos task rm  default <id>
  • Inspector API — the same data programmatically (chronos.NewInspector).
  • Prometheus + Grafana — a Metrics hook (core, dependency-free) plus a ready stack in contrib/prometheus: 
    cd contrib/prometheus/deploy && docker compose up --build
# Grafana: http://localhost:3000  (dashboard "chronos-go")

See docs/OBSERVING.md for Redis-level inspection.

How it works

  • Immediate queue: a Redis Stream per queue with one consumer group. Workers XREADGROUP (blocking), run the handler, then XACK + XDEL. Queues are selected by smooth weighted round-robin (or strictly by weight with StrictPriority).
  • Delayed / retry / archived: sorted sets scored by run-at / retry-at / died-at; a forwarder promotes due entries back into the stream.
  • Crash recovery: a recoverer XAUTOCLAIMs entries whose worker went silent and re-queues or dead-letters them (attempts tracked in the task hash).
  • Scheduler: a leader (Redis SET NX PX lock + pub/sub resignation) runs the tick loop; each trigger is enqueued under a deterministic dedup key (<schedule>:<trigger-unix>) so a split-brain hand-off cannot double-enqueue.
  • Heartbeat: refreshes in-flight tasks' PEL idle (XCLAIM ... JUSTID) and unique-lock TTL so long-running work is safe.
  • Keys are wrapped in a {queue} hash tag so every multi-key Lua script is Redis Cluster-safe.

Performance

On an M4 Pro with local Redis (100-byte payloads, defaults on both sides, median of 3), chronos-go processes ~15k tasks/s at concurrency 16 and ~20k at 64 end to end — about 2.5–4x asynq at those settings, with enqueue throughput on par (~26k/s vs ~27k/s). Fetches are batched (XREADGROUP COUNT + pipelining), so throughput scales with worker concurrency; at low concurrency (C=1) asynq is ~1.5x faster. Full methodology, tables, and caveats: docs/BENCHMARKS.md — reproduce with make bench.

Redis Cluster

chronos-go works on Redis Cluster out of the box. Every key of a queue is wrapped in a {queue} hash tag, so a queue's keys share one slot (multi-key Lua stays atomic) while different queues spread across the cluster.

rdb := redis.NewClusterClient(&redis.ClusterOptions{
	Addrs: []string{"node1:6379", "node2:6379", "node3:6379"},
})
srv := chronos.NewServer(rdb, chronos.ServerConfig{ /* ... */ })

The CLI connects with --cluster (seed nodes, comma-separated — one is enough):

chronos --cluster --redis node1:6379,node2:6379 queue ls

Notes:

  • Redis Cluster has only logical database 0 (--db is standalone-only).
  • The global keys (chronos:queues, the scheduler leader lock) are accessed with single-key commands or single-key Lua scripts only, so they are cluster-safe without a hash tag.
  • Sentinel: inject a redis.NewFailoverClient — it satisfies the same redis.UniversalClient interface — but Sentinel is not part of our tested matrix yet.
Verifying against a real cluster

The repo ships a disposable 6-node cluster and a script-complete integration suite (every Lua script and command pattern runs on cluster at least once):

cd deploy/redis-cluster && docker compose up -d && cd ../..
make test-cluster

Delivery semantics

chronos-go is at-least-once: a task can run more than once (e.g. a worker crashes after finishing but before acking). Make handlers idempotent.

vs. asynq

asynq chronos-go
Payload API []byte + manual unmarshal generic Task[T] (type-safe)
Scheduler across instances app must ensure a single scheduler built-in leader election + deterministic dedup
Unique lock during long processing can expire (TTL only) heartbeat renews it
Stream / dead-letter growth — trimmed (XDEL) + janitor retention
Backend Redis Redis

Known limitations / roadmap

  • Scheduler fencing relies on a dedup-key TTL (no monotonic fencing token); a leader paused longer than the TTL could theoretically re-enqueue a trigger. All instances must share the same Location and reasonably synced clocks.
  • The unique lock is heartbeat-renewed only while a task is actively processing; while it waits in the scheduled/retry set, it is covered by its TTL — set the TTL comfortably above expected waiting time.
  • Not yet built: a web UI, chain×group composition (groups and chains cannot nest yet), result passing between workflow steps.

Development

Tests run against a real Redis (skipped if none is reachable at $REDIS_ADDR, default 127.0.0.1:6379). They share a logical DB, so run packages serially:

make check        # gofmt + vet + go test ./... -race -p 1 + contrib tests

Cluster integration tests are opt-in: make test-cluster (see deploy/redis-cluster).

License

MIT

Documentation

Overview

Package chronos is a Redis-backed distributed scheduler and task queue.

Index

Constants

View Source 
const DefaultMaxRetry = 25

DefaultMaxRetry is the retry budget used when WithMaxRetry is not given.

View Source 
const DefaultQueue = "default"

DefaultQueue is the queue used when none is specified.

Variables

View Source 
var (
	ErrTaskNotFound = errors.New("chronos: task not found")
	ErrInvalidState = errors.New("chronos: invalid task state")
)

Errors returned by the Inspector, exposed so callers (e.g. the web console) can map them to the right response.

View Source 
var ErrDuplicateTask = rdb.ErrDuplicateTask

ErrDuplicateTask is returned by Enqueue when WithUnique is used and an identical task already holds the unique lock.

Functions

func AddHandler 

func AddHandler[T TaskArgs](mux *Mux, fn func(ctx context.Context, task *Task[T]) error)

AddHandler registers a strongly-typed handler for tasks of type T. The Kind is read from the zero value of T, so T's Kind method must use a value receiver. Registering two handlers for the same Kind panics.

func DefaultRetryDelay 

func DefaultRetryDelay(retried int, _ error) time.Duration

DefaultRetryDelay is the default backoff: an exponential cap (base * 2^retried, clamped to retryMaxDelay) with full jitter — the actual delay is uniformly random in [0, cap]. Full jitter spreads retries to avoid thundering herds.

func RegisterCron 

func RegisterCron[T TaskArgs](s *Scheduler, spec string, args T, opts ...Option) error

RegisterCron registers args to be enqueued on a standard 5-field cron spec.

func RegisterInterval 

func RegisterInterval[T TaskArgs](s *Scheduler, interval time.Duration, args T, opts ...Option) error

RegisterInterval registers args to be enqueued every interval (>= 1s). Any sub-second component is truncated to whole seconds (e.g. 1500ms behaves as 1s). Register all schedules before calling Start (registration is not concurrency- safe with the running tick loop).

func SkipRetry 

func SkipRetry(err error) error

SkipRetry wraps err so that returning it from a handler dead-letters the task immediately, bypassing the remaining retry budget.

Types

type Chain added in v0.6.0 

type Chain struct {
	// contains filtered or unexported fields
}

Chain builds a sequence of tasks in which each link is enqueued only after the previous one succeeds. A link that exhausts its retries stops the chain; re-running the dead-lettered link (Inspector/CLI RunTask) resumes it, because every link carries its remaining tail inside its own message.

Handlers must be idempotent, as everywhere in chronos-go: a redelivered link may run its handler more than once. Its successor is enqueued at most once while that successor still exists; if a predecessor is redelivered after its successor already finished and was not retained, the successor can be recreated — the usual at-least-once caveat. Per-link WithRetention keeps the successor's record around and closes that window for its duration.

func NewChain added in v0.6.0 

func NewChain() *Chain

NewChain returns an empty chain builder.

func (*Chain) Enqueue added in v0.6.0 

func (ch *Chain) Enqueue(ctx context.Context, c *Client) (*TaskInfo, error)

Enqueue makes the first link available for processing and returns its TaskInfo. Later links run only as their predecessors succeed.

func (*Chain) Then added in v0.6.0 

func (ch *Chain) Then(args TaskArgs, opts ...Option) *Chain

Then appends a link. opts accepts the usual per-task options (WithQueue, WithMaxRetry, WithRetention, WithProcessIn, ...); WithTaskID and WithUnique are rejected at Enqueue time (the chain owns task IDs, and unique dedup inside chains is not supported).

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client enqueues tasks.

func NewClient 

func NewClient(r redis.UniversalClient) *Client

NewClient returns a Client backed by the given Redis client.

func (*Client) Close 

func (c *Client) Close() error

Close releases the client's resources. The underlying Redis client is owned by the caller and is not closed here.

type Group added in v0.7.0 

type Group struct {
	// contains filtered or unexported fields
}

Group builds a parallel fan-out: every member is enqueued at once, and when ALL members have succeeded, the callback task is enqueued exactly once while its record exists. A member that exhausts its retries parks the group: re-run its dead-letter (Inspector/CLI RunTask) and, once it succeeds, the group resumes — the same stop-and-resume rule chains follow. Abandoned groups (a member deleted and never re-run) expire after rdb.GroupTTL.

Handlers must be idempotent, as everywhere in chronos-go.

func NewGroup added in v0.7.0 

func NewGroup() *Group

NewGroup returns an empty group builder.

func (*Group) Add added in v0.7.0 

func (g *Group) Add(args TaskArgs, opts ...Option) *Group

Add appends a member task. Members run in parallel on their own queues with their own options; WithTaskID and WithUnique are rejected at Enqueue time.

func (*Group) Enqueue added in v0.7.0 

func (g *Group) Enqueue(ctx context.Context, c *Client) (*GroupInfo, error)

Enqueue validates every member up front, creates the group's pending-member record (so a partially failed enqueue can never fire the callback early), then enqueues every member. Enqueueing is sequential and non-atomic only against infrastructure failures: a network error midway leaves already enqueued members running and the group incomplete (its record expires after rdb.GroupTTL).

func (*Group) OnComplete added in v0.7.0 

func (g *Group) OnComplete(args TaskArgs, opts ...Option) *Group

OnComplete sets the callback task, enqueued once every member has succeeded. WithProcessIn delays it relative to the group's completion; WithProcessAt, WithTaskID and WithUnique are rejected.

type GroupInfo added in v0.7.0 

type GroupInfo struct {
	GroupID    string   // the group's identity
	MemberIDs  []string // deterministic member task IDs ("<groupID>:m<i>")
	CallbackID string   // the callback's task ID ("<groupID>:cb")
}

GroupInfo describes an enqueued group.

type Inspector 

type Inspector struct {
	// contains filtered or unexported fields
}

Inspector provides read and administrative access to queues and tasks. It is the foundation the CLI (and any future UI) is built on.

func NewInspector 

func NewInspector(r redis.UniversalClient) *Inspector

NewInspector returns an Inspector backed by the given Redis client.

func (*Inspector) DeleteTask 

func (i *Inspector) DeleteTask(ctx context.Context, qname, taskID string) error

DeleteTask removes a scheduled/retry/archived task and releases its unique lock.

func (*Inspector) GetTask added in v0.3.0 

func (i *Inspector) GetTask(ctx context.Context, qname, taskID string) (*TaskInfo, error)

GetTask returns full detail for a single stored task (scheduled/retry/archived/completed).

func (*Inspector) ListTasks 

func (i *Inspector) ListTasks(ctx context.Context, qname, state string, limit int) ([]*TaskInfo, error)

ListTasks returns up to limit tasks in the given state (scheduled|retry|archived).

func (*Inspector) Queues 

func (i *Inspector) Queues(ctx context.Context) ([]*QueueInfo, error)

Queues lists all known queues with their per-state counts.

func (*Inspector) RunTask 

func (i *Inspector) RunTask(ctx context.Context, qname, taskID string) error

RunTask promotes a scheduled/retry/archived task so it runs immediately.

type Metrics 

type Metrics interface {
	ObserveTask(queue, kind string, outcome TaskOutcome, dur time.Duration)
}

Metrics receives one observation per processed task. Implementations MUST be safe for concurrent use (the server calls it from worker goroutines). The zero/nil Metrics disables observation. The Prometheus implementation lives in the contrib/prometheus module so the core stays dependency-free.

type MisfirePolicy 

type MisfirePolicy int

MisfirePolicy decides what happens when a schedule's triggers were missed (e.g. during a leader-election gap or downtime). 

const (
	// MisfireSkip (default) discards missed triggers and resumes on schedule.
	MisfireSkip MisfirePolicy = iota
	// MisfireFireOnce fires exactly one catch-up trigger if any were missed.
	MisfireFireOnce
)

type Mux 

type Mux struct {
	// contains filtered or unexported fields
}

Mux routes tasks to handlers by their Kind.

func NewMux 

func NewMux() *Mux

NewMux returns an empty Mux.

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option customizes a single Enqueue call.

func WithDeadLetterDiscard 

func WithDeadLetterDiscard() Option

WithDeadLetterDiscard discards the task on retry exhaustion instead of storing it in the archived ZSET. The OnDeadLetter hook still fires.

func WithMaxRetry 

func WithMaxRetry(n int) Option

WithMaxRetry sets the maximum number of retries before the task is dead-lettered. Defaults to DefaultMaxRetry.

func WithMisfirePolicy 

func WithMisfirePolicy(p MisfirePolicy) Option

WithMisfirePolicy sets how a scheduled job handles missed triggers (after a leader-election gap or downtime). Only meaningful for RegisterInterval / RegisterCron; ignored by a plain Enqueue. Defaults to MisfireSkip.

func WithProcessAt 

func WithProcessAt(t time.Time) Option

WithProcessAt schedules the task to first become available at t. A non-future time enqueues immediately.

func WithProcessIn 

func WithProcessIn(d time.Duration) Option

WithProcessIn schedules the task to first become available after d. A non-positive d enqueues immediately.

func WithQueue 

func WithQueue(name string) Option

WithQueue routes the task to a specific queue.

func WithRetention added in v0.5.0 

func WithRetention(d time.Duration) Option

WithRetention keeps the task in the "completed" set for d after it succeeds, so it can be inspected (Inspector/CLI state "completed") before the janitor removes it. The default (no option) deletes the task immediately on success. Durations under one second are rounded up to one second; d <= 0 is ignored. Note: on high-throughput queues a long retention grows Redis memory — the per-queue MaxCompleted cap (ServerConfig) bounds the worst case.

func WithTaskID 

func WithTaskID(id string) Option

WithTaskID sets an explicit task ID. Enforced deduplication is provided by the unique lock introduced in a later milestone; in M1, re-enqueueing with the same ID is not guaranteed to prevent duplicates. When omitted a random UUID is generated.

func WithUnique 

func WithUnique(ttl time.Duration) Option

WithUnique deduplicates tasks by (kind + payload): while a matching task is anywhere in the pipeline (pending, retrying, scheduled), enqueueing another returns ErrDuplicateTask. The lock is released when the task reaches a terminal state (completed / archived / discarded).

While a task is actively being processed, the server's heartbeat renews the lock's TTL, so a single attempt that runs longer than ttl still holds the lock. ttl mainly bounds the lock for the time a task spends waiting (pending / scheduled / retry backoff) where no worker is renewing it; set it comfortably above the expected total waiting time. For a delayed task the lock TTL is automatically extended to cover the delay.

type QueueInfo 

type QueueInfo struct {
	Queue     string
	Pending   int64
	Active    int64
	Scheduled int64
	Retry     int64
	Archived  int64
	Completed int64
}

QueueInfo is a queue's per-state task counts.

type RetryDelayFunc 

type RetryDelayFunc func(retried int, err error) time.Duration

RetryDelayFunc computes how long to wait before the next retry, given the number of retries already performed and the error that caused the failure.

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler registers periodic jobs and, on whichever instance is elected leader, enqueues their due triggers. Every instance may call Start; only the leader enqueues, and deterministic dedup keys prevent double-enqueue during leader handover.

func NewScheduler 

func NewScheduler(r redis.UniversalClient, cfg SchedulerConfig) *Scheduler

NewScheduler returns a Scheduler backed by the given Redis client.

func (*Scheduler) Shutdown 

func (s *Scheduler) Shutdown(ctx context.Context) error

Shutdown stops the scheduler; if this instance is the leader it resigns so a follower can take over immediately.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context) error

Start launches leader election and the tick loop. Safe to call on every instance; only the leader enqueues. Register all schedules before Start.

For correct cross-instance behavior, every instance must register the same schedules with the same SchedulerConfig.Location and have reasonably synchronized clocks: the deterministic dedup key is derived from the computed trigger instant, so divergent timezones or badly skewed clocks could let two instances treat the same logical trigger as different ones.

type SchedulerConfig 

type SchedulerConfig struct {
	// Location is the timezone for cron schedules. Defaults to time.Local.
	Location *time.Location
	// Logger receives operational logs. Defaults to slog.Default().
	Logger *slog.Logger
	// LeaderTTL is how long a leadership term lasts before it must be renewed.
	// Defaults to 5s. Failover happens within ~LeaderTTL of a leader dying.
	LeaderTTL time.Duration
}

SchedulerConfig configures a Scheduler.

type Server 

type Server struct {
	// contains filtered or unexported fields
}

Server fetches tasks from Redis and dispatches them to handlers.

func NewServer 

func NewServer(r redis.UniversalClient, cfg ServerConfig) *Server

NewServer returns a Server backed by the given Redis client.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown stops fetching and waits for in-flight tasks to finish, bounded by ctx. In-flight tasks that do not finish before ctx expires are left unacked and will be recovered by another instance in a later milestone.

func (*Server) Start 

func (s *Server) Start(ctx context.Context, mux *Mux) error

Start ensures consumer groups exist and launches the fetch loop. It returns once startup is complete; processing continues in the background until Shutdown is called or ctx is cancelled.

type ServerConfig 

type ServerConfig struct {
	// Queues maps queue name to weight. While every queue has work, a queue with
	// weight 6 is dequeued about 6x as often as a queue with weight 1 (smooth
	// weighted round-robin — no queue starves). When the queue picked for a round
	// is empty, that round falls through to the highest-weight queue that does
	// have work, so an idle high-priority queue never blocks lower ones. Weights
	// <= 0 are treated as 1; very large weights are capped.
	Queues map[string]int
	// StrictPriority, if true, always drains higher-weight queues first: a
	// lower-weight queue is served only while every higher-weight queue is
	// empty. Ties are broken by queue name for determinism.
	StrictPriority bool
	// Concurrency is the max number of tasks processed simultaneously.
	Concurrency int
	// Logger receives operational logs. Defaults to slog.Default().
	Logger *slog.Logger

	// RetryDelayFunc computes the backoff before a retry. Defaults to
	// DefaultRetryDelay (exponential + full jitter).
	RetryDelayFunc RetryDelayFunc
	// OnDeadLetter is invoked when a task exhausts its retries (or returns a
	// SkipRetry error). It fires whether the task is archived or discarded.
	//
	// The heartbeat keeps an actively-processing task's lease fresh, so the
	// recoverer does not normally reclaim it. It may still fire more than once in
	// pathological cases (the server/heartbeat unavailable long enough for the
	// lease to lapse), so make the hook idempotent (the archived ZSET entry is
	// deduplicated by task ID).
	OnDeadLetter func(ctx context.Context, info *TaskInfo, err error)

	// Metrics, if set, receives one observation per processed task. Use the
	// contrib/prometheus implementation, or your own. Defaults to nil (disabled).
	Metrics Metrics

	// ForwardInterval is how often the retry ZSET is scanned for due tasks.
	// Defaults to 1s.
	ForwardInterval time.Duration
	// RecoverInterval is how often stuck PEL entries are reclaimed. Defaults to 15s.
	RecoverInterval time.Duration
	// RecoverMinIdle is how long a PEL entry must be idle before it is treated as
	// abandoned. Defaults to 30s when unset (<= 0). The heartbeat refreshes the
	// lease of in-flight tasks every HeartbeatInterval, so a task that runs longer
	// than RecoverMinIdle is safe as long as this server (its heartbeat) is alive;
	// RecoverMinIdle is the window after a worker actually dies before its tasks
	// are reclaimed.
	RecoverMinIdle time.Duration

	// HeartbeatInterval is how often the server refreshes the lease and unique
	// lock of in-flight tasks. Defaults to RecoverMinIdle/3. Must be shorter than
	// RecoverMinIdle so an actively-processing task is never reclaimed.
	HeartbeatInterval time.Duration

	// ArchivedRetention is how long a dead-lettered (archived) task is kept
	// before the janitor deletes it. Defaults to 7 days (168h).
	ArchivedRetention time.Duration
	// MaxArchived caps the number of archived tasks per queue; the janitor
	// deletes the oldest beyond this even within the retention window. Defaults
	// to 10000. Set negative to disable the size cap.
	MaxArchived int
	// MaxCompleted caps the number of retained completed tasks per queue; the
	// janitor deletes the oldest beyond this even before their retention
	// expires. Defaults to 10000. Set negative to disable the size cap.
	MaxCompleted int
	// JanitorInterval is how often the janitor runs. Defaults to 1 minute.
	JanitorInterval time.Duration
}

ServerConfig configures a Server.

type Task 

type Task[T TaskArgs] struct {
	Args T
	// contains filtered or unexported fields
}

Task is a strongly-typed task delivered to a handler.

func (*Task[T]) ID 

func (t *Task[T]) ID() string

ID returns the task's unique identifier.

func (*Task[T]) Queue 

func (t *Task[T]) Queue() string

Queue returns the queue the task was enqueued to.

type TaskArgs 

type TaskArgs interface {
	Kind() string
}

TaskArgs is implemented by every task payload type. Kind returns a stable identifier used to route the task to its handler; it MUST be defined on a value receiver so it can be called on the zero value during registration.

type TaskInfo 

type TaskInfo struct {
	ID    string
	Kind  string
	Queue string

	// The following are populated by Inspector.ListTasks / GetTask for tasks
	// stored in a state ZSET (scheduled / retry / archived).
	State         string    // "scheduled" | "retry" | "archived" | ...
	Payload       []byte    // raw task payload
	Retried       int       // retries already attempted
	MaxRetry      int       // retry budget
	LastErr       string    // most recent failure message ("" if none)
	NextProcessAt time.Time // ZSET score as a time: scheduled-for / retry-at / died-at / expires-at (completed)
	CompletedAt   time.Time // when the task finished successfully (zero unless retained)

	ChainPending int // number of chain links still queued behind this task (0 = none/last)

	GroupID      string // group this task belongs to ("" = none)
	GroupPending int    // members of that group not yet succeeded; populated by GetTask only. 0 also when the group finished or its record expired (see rdb.GroupTTL) or the lookup failed — it is a hint, not an authority.
}

TaskInfo describes an enqueued or stored task. Enqueue returns one with only ID/Kind/Queue set; the Inspector fills the rest for stored tasks.

func Enqueue 

func Enqueue[T TaskArgs](ctx context.Context, c *Client, args T, opts ...Option) (*TaskInfo, error)

Enqueue serializes args and makes the task available for immediate processing. It is a package-level function rather than a method because Go methods cannot have type parameters.

type TaskOutcome 

type TaskOutcome string

TaskOutcome is the terminal result of processing one task, reported to Metrics. 

const (
	// OutcomeSuccess: the handler returned nil; the task was acked and removed.
	OutcomeSuccess TaskOutcome = "success"
	// OutcomeRetry: the handler failed and the task was scheduled for retry.
	OutcomeRetry TaskOutcome = "retry"
	// OutcomeDeadLetter: the task exhausted retries (or returned SkipRetry) and
	// was archived or discarded.
	OutcomeDeadLetter TaskOutcome = "dead_letter"
)

Source Files

View all Source files

Directories

Path Synopsis
cmd
chronos command
Command chronos is a CLI for inspecting and administering chronos-go queues.
 Command chronos is a CLI for inspecting and administering chronos-go queues.
examples
jobscheduler
Package jobscheduler is a verification adapter proving chronos-go can back operator-review's string+[]byte JobScheduler interface.
 Package jobscheduler is a verification adapter proving chronos-go can back operator-review's string+[]byte JobScheduler interface.
tour command
Command tour is a narrated, runnable walkthrough of everything chronos-go can do: the core queue, reliability (retry / dead-letter), delayed + unique tasks, the Inspector, the distributed scheduler with leader failover, the retention janitor, the heartbeat, weighted priority queues, completed-task retention, task chains, and task groups.
 Command tour is a narrated, runnable walkthrough of everything chronos-go can do: the core queue, reliability (retry / dead-letter), delayed + unique tasks, the Inspector, the distributed scheduler with leader failover, the retention janitor, the heartbeat, weighted priority queues, completed-task retention, task chains, and task groups.
internal
base
Package base defines the Redis key layout, task states, and message serialization shared across chronos-go internals.
 Package base defines the Redis key layout, task states, and message serialization shared across chronos-go internals.
rdb
Package rdb implements the Redis operations backing chronos-go: enqueueing tasks, dequeueing via a consumer group, and acking completion.
 Package rdb implements the Redis operations backing chronos-go: enqueueing tasks, dequeueing via a consumer group, and acking completion.
testutil
Package testutil provides shared test helpers for chronos-go.
 Package testutil provides shared test helpers for chronos-go.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.