chronos

README

chronos-go

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.

v1.0.0 — the core package API is stable under semantic versioning. See Stability & compatibility.

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.

Pausing a queue

Inspector.PauseQueue (or chronos queue pause <q>, or the web console's ⏸ toggle) stops servers from consuming a queue within about one second — enqueueing, forwarding and recovery continue, so work simply accumulates as pending until you resume. In-flight tasks finish normally.

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.
• A group can also sit inside a chain as a parallel stage — see Parallel stages.

Passing results between steps

Register a handler with AddHandlerR and its success return value flows through the workflow: the next chain link reads it with chronos.PrevResult[R](task), and a group's callback receives every member's result in Add order via chronos.GroupResults[R](task) (or raw bytes with task.RawGroupResults()). Results are carried inside task messages — no extra keys, no TTL to manage — and survive retries, redeliveries and dead-letter re-runs. A result's JSON form is capped at 1 MiB (MaxResultSize); larger results dead-letter the task without retry, so pass a reference (object-store path, row ID) for big artifacts.

Each chain link receives only its immediate predecessor's result: an intermediate link registered with plain AddHandler (no result) breaks the relay, so the link after it gets ErrNoResult from PrevResult.

Parallel stages (fan-out → fan-in)

ThenGroup puts a group in the middle (or at the end) of a chain:

chronos.NewChain().
	Then(Validate{}).
	ThenGroup(chronos.NewGroup().
		Add(Encode{Res: "720p"}).
		Add(Encode{Res: "4k"}).
		OnComplete(BuildManifest{})).   // fan-in: receives GroupResults
	Then(Deploy{}).                     // receives the callback's result
	Enqueue(ctx, client)

Every member receives the previous stage's result (PrevResult), the callback fans the member results in, and its own result flows to the next stage. Failure semantics are unchanged: a dead-lettered member stalls the stage until you re-run it (RunTask), and a completed stage is fenced against predecessor redeliveries for as long as its callback hash lives — set WithRetention on the OnComplete callback to keep that window closed (the knob is the callback's retention, not the members'). A group cannot be the first stage — start with Then, or use NewGroup directly.

Chains as group members

A group member can be a chain (AddChain): each member runs its links in sequence, and the chain's final link reports the member's completion to the group (its last result becomes that member's GroupResults entry). This expresses fan-out-of-pipelines — e.g. migrate N tenants, each a dump→transform→load chain, in parallel, then a verify callback:

g := chronos.NewGroup()
for _, t := range tenants {
	g.AddChain(chronos.NewChain().Then(Dump{t}).Then(Transform{t}).Then(Load{t}))
}
g.OnComplete(Verify{}).Enqueue(ctx, client)

A dead-lettered member link stalls that member until you re-run it (RunTask); the chain then resumes to its final link and reports. Nesting is one level deep: a member chain may not contain a ThenGroup stage, and a group used as a ThenGroup stage may not have chain members.

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.

Registered schedules are published to a registry, so the Inspector, CLI and web console can list them (with last-fired and liveness) even before they first fire; entries no scheduler has refreshed for a minute show as stale.

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 queue pause default        # stop consuming (resume to undo)
  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")
  

• Web console — a browser task-management UI (card dashboard, chain/group visualization, bulk re-run, queue pause, schedule registry, cluster-aware) in contrib/webui:

  cd contrib/webui && go run ./cmd/webui                 # standalone
  go run ./cmd/webui --cluster --redis n1:7000           # Redis Cluster
  

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

Stability & compatibility

From v1.0.0, the core package (github.com/kenshin579/chronos-go) follows semantic versioning: breaking changes to its public API land only in a new major version. APIs slated for removal are marked with a // Deprecated: godoc comment for at least one minor release first.

Not covered by this guarantee: the contrib/ modules (contrib/webui, contrib/prometheus) — experimental, versioned separately — and any internal/ package. The core Metrics hook (zero dependencies) is part of the core API.

Supported Go versions: the two most recent stable releases.

Known limitations / roadmap

Delivery is at-least-once — a handler may run more than once (crash redelivery, or a recoverer reclaiming a task idle longer than RecoverMinIdle), so handlers must be idempotent.

• Scheduler fencing relies on a dedup TTL (10 × LeaderTTL), not a fencing token: a leader paused longer than that window and then resumed could double- fire a trigger. Instances must share one clock and Location.
• Unique locks cover in-flight tasks via heartbeat; a task waiting in the retry/scheduled set is covered only by its lock TTL, so set the TTL to exceed the task's total lifetime.
• Queue pause takes effect within ~1s (server-side cache). PauseQueue accepts any queue name; an unknown name lingers in the paused set until ResumeQueue clears it (no effect on active queues).
• Workflow nesting is one level: a group member may be a chain (AddChain), and a chain may contain a parallel stage (ThenGroup), but not deeper (a member chain cannot contain a ThenGroup, and a group cannot nest a group).
• Results are capped at 1 MiB (MaxResultSize); pass a reference for larger artifacts. Results are relay-only (no out-of-workflow result store).

Roadmap (not yet built): asynq→chronos migration guide, official Sentinel support, deeper workflow nesting.

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).

Before major releases, run make soak — an hour-long leak soak against a local Redis (heap / goroutine / keyspace trend check; -duration 4h for v1.0.0-grade ones). See benchmarks/README.md.

License

MIT

Documentation

Overview

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

A Client enqueues tasks; a Server consumes them by routing each to a handler registered on a Mux. Tasks are strongly typed: define an args type implementing TaskArgs, register a handler with AddHandler (or AddHandlerR to return a result), and enqueue with Enqueue.

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

mux := chronos.NewMux()
chronos.AddHandler(mux, func(ctx context.Context, t *chronos.Task[EmailArgs]) error {
	return send(t.Args.To)
})
// At least one queue is required: Start errors if Queues is empty.
srv := chronos.NewServer(rdb, chronos.ServerConfig{
	Queues:      map[string]int{"default": 1},
	Concurrency: 10,
})
srv.Start(ctx, mux)
chronos.Enqueue(ctx, client, EmailArgs{To: "a@b.c"})

Workflows

NewChain runs tasks in sequence (each link starts when its predecessor succeeds); NewGroup fans out tasks in parallel and fires an OnComplete callback when all finish. Results flow between steps: a handler registered with AddHandlerR exposes its result to the next chain link via PrevResult and to a group callback via GroupResults. Chain.ThenGroup embeds a parallel stage inside a chain, and Group.AddChain makes a group member a chain — together they express fan-out/fan-in pipelines.

Scheduling

NewScheduler runs periodic jobs (interval or cron) with leader election, so many instances may run a scheduler and only one fires each trigger.

Inspection

NewInspector reads queue stats, lists and re-runs tasks, pauses queues, and reports scheduler state — the read/operate surface behind the CLI and web console.

Delivery semantics

Delivery is at-least-once: a handler may run more than once (redelivery after a crash, or a recoverer reclaiming a stalled task), so handlers must be idempotent. See the module README for the full list of guarantees and caveats.

Stability

From v1.0.0 the core package follows semantic versioning. The contrib modules (contrib/webui, contrib/prometheus) and internal/ are experimental and not covered by that guarantee.

Example

Example shows the enqueue → handler round trip: define a typed args, register a handler on a Mux, start a Server, and enqueue.

package main

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"

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

type EmailArgs struct {
	To string `json:"to"`
}

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

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

	mux := chronos.NewMux()
	chronos.AddHandler(mux, func(ctx context.Context, t *chronos.Task[EmailArgs]) error {
		fmt.Println("sending to", t.Args.To)
		return nil
	})

	srv := chronos.NewServer(rdb, chronos.ServerConfig{
		Queues:      map[string]int{"default": 1},
		Concurrency: 10,
	})
	_ = srv.Start(ctx, mux)
	defer srv.Shutdown(context.Background())

	client := chronos.NewClient(rdb)
	_, _ = chronos.Enqueue(ctx, client, EmailArgs{To: "a@b.c"}, chronos.WithQueue("default"))
	// Output omitted: a real Redis round trip with async processing is not
	// deterministic, so go test compiles this example without running it.
}

Index

• Constants
• Variables
• func AddHandler[T TaskArgs](mux *Mux, fn func(ctx context.Context, task *Task[T]) error)
• func AddHandlerR[T TaskArgs, R any](mux *Mux, fn func(ctx context.Context, task *Task[T]) (R, error))
• func DefaultRetryDelay(retried int, _ error) time.Duration
• func GroupResults[R any, T TaskArgs](t *Task[T]) ([]R, error)
• func PrevResult[R any, T TaskArgs](t *Task[T]) (R, error)
• func RegisterCron[T TaskArgs](s *Scheduler, spec string, args T, opts ...Option) error
• func RegisterInterval[T TaskArgs](s *Scheduler, interval time.Duration, args T, opts ...Option) error
• func SkipRetry(err error) error
• type Chain
  • func NewChain() *Chain
  • func (ch *Chain) Enqueue(ctx context.Context, c *Client) (*TaskInfo, error)
  • func (ch *Chain) Then(args TaskArgs, opts ...Option) *Chain
  • func (ch *Chain) ThenGroup(g *Group) *Chain
• type Client
  • func NewClient(r redis.UniversalClient) *Client
  • func (c *Client) Close() error
• type Group
  • func NewGroup() *Group
  • func (g *Group) Add(args TaskArgs, opts ...Option) *Group
  • func (g *Group) AddChain(ch *Chain) *Group
  • func (g *Group) Enqueue(ctx context.Context, c *Client) (*GroupInfo, error)
  • func (g *Group) OnComplete(args TaskArgs, opts ...Option) *Group
• type GroupInfo
• type Inspector
  • func NewInspector(r redis.UniversalClient) *Inspector
  • func (i *Inspector) DeleteTask(ctx context.Context, qname, taskID string) error
  • func (i *Inspector) GetTask(ctx context.Context, qname, taskID string) (*TaskInfo, error)
  • func (i *Inspector) GroupMembers(ctx context.Context, cbQueue, groupID string) ([]string, error)
  • func (i *Inspector) ListTasks(ctx context.Context, qname, state string, limit int) ([]*TaskInfo, error)
  • func (i *Inspector) PauseQueue(ctx context.Context, qname string) error
  • func (i *Inspector) PausedQueues(ctx context.Context) ([]string, error)
  • func (i *Inspector) Queues(ctx context.Context) ([]*QueueInfo, error)
  • func (i *Inspector) ResumeQueue(ctx context.Context, qname string) error
  • func (i *Inspector) RunTask(ctx context.Context, qname, taskID string) error
  • func (i *Inspector) SchedulerStatus(ctx context.Context) (*SchedulerStatus, error)
• type Metrics
• type MisfirePolicy
• type Mux
  • func NewMux() *Mux
• type Option
  • func WithDeadLetterDiscard() Option
  • func WithMaxRetry(n int) Option
  • func WithMisfirePolicy(p MisfirePolicy) Option
  • func WithProcessAt(t time.Time) Option
  • func WithProcessIn(d time.Duration) Option
  • func WithQueue(name string) Option
  • func WithRetention(d time.Duration) Option
  • func WithTaskID(id string) Option
  • func WithUnique(ttl time.Duration) Option
• type QueueInfo
• type RetryDelayFunc
• type ScheduleInfo
• type Scheduler
  • func NewScheduler(r redis.UniversalClient, cfg SchedulerConfig) *Scheduler
  • func (s *Scheduler) Shutdown(ctx context.Context) error
  • func (s *Scheduler) Start(ctx context.Context) error
• type SchedulerConfig
• type SchedulerStatus
• type Server
  • func NewServer(r redis.UniversalClient, cfg ServerConfig) *Server
  • func (s *Server) Shutdown(ctx context.Context) error
  • func (s *Server) Start(ctx context.Context, mux *Mux) error
• type ServerConfig
• type Task
  • func (t *Task[T]) ID() string
  • func (t *Task[T]) Queue() string
  • func (t *Task[T]) RawGroupResults() [][]byte
• type TaskArgs
• type TaskInfo
  • func Enqueue[T TaskArgs](ctx context.Context, c *Client, args T, opts ...Option) (*TaskInfo, error)
• type TaskOutcome

Examples

• Package
• AddHandlerR
• NewChain
• NewGroup
• NewInspector
• NewScheduler

Constants

const DefaultMaxRetry = 25

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

const DefaultQueue = "default"

DefaultQueue is the queue used when none is specified.

const MaxResultSize = 1 << 20

MaxResultSize bounds a handler result's JSON encoding. Larger results dead-letter the task without retry (the same value would be produced again) — pass a reference (object-store path, row ID) instead.

Variables

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.

var ErrDuplicateTask = rdb.ErrDuplicateTask

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

var ErrNoResult = errors.New("chronos: no result")

ErrNoResult is returned when the previous step (or a group member) produced no result — the handler was registered with AddHandler, or this is the first chain link.

var ErrResultTooLarge = errors.New("chronos: result exceeds MaxResultSize")

ErrResultTooLarge marks a handler result that exceeds MaxResultSize.

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 AddHandlerR

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

AddHandlerR registers a handler whose success return value becomes the task's result: it is relayed to the next chain link (read with PrevResult) and collected for the group callback (read with GroupResults). The result is marshalled as JSON. Kind rules and duplicate-registration panics match AddHandler.

Example

ExampleAddHandlerR shows a handler that returns a result, which the next workflow step reads with PrevResult / GroupResults.

package main

import (
	"context"

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

type ResizeArgs struct {
	Path string `json:"path"`
}

func (ResizeArgs) Kind() string { return "img:resize" }

type ResizeResult struct {
	Bytes int `json:"bytes"`
}

func main() {
	mux := chronos.NewMux()
	chronos.AddHandlerR(mux, func(ctx context.Context, t *chronos.Task[ResizeArgs]) (ResizeResult, error) {
		return ResizeResult{Bytes: 1024}, nil
	})
	_ = mux
}

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 GroupResults

func GroupResults[R any, T TaskArgs](t *Task[T]) ([]R, error)

GroupResults decodes every member result in Add order. It assumes a homogeneous group (every member returned R); a member without a result fails with ErrNoResult. For heterogeneous or partial results use RawGroupResults.

func PrevResult

func PrevResult[R any, T TaskArgs](t *Task[T]) (R, error)

PrevResult decodes the previous chain link's result:

out, err := chronos.PrevResult[EncodeResult](task)

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

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

func NewChain() *Chain

NewChain returns an empty chain builder.

Example

ExampleNewChain runs tasks in sequence; each link starts after the previous one succeeds.

package main

import (
	"context"

	"github.com/redis/go-redis/v9"

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

type ResizeArgs struct {
	Path string `json:"path"`
}

func (ResizeArgs) Kind() string { return "img:resize" }

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	client := chronos.NewClient(rdb)
	_, _ = chronos.NewChain().
		Then(ResizeArgs{Path: "a.png"}).
		Then(ResizeArgs{Path: "b.png"}).
		Enqueue(context.Background(), client)
}

func (*Chain) Enqueue

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

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).

func (*Chain) ThenGroup

func (ch *Chain) ThenGroup(g *Group) *Chain

ThenGroup appends a parallel stage: every member of g runs concurrently (each receiving the previous stage's result via PrevResult) and g's OnComplete callback fans the member results in (GroupResults) before the chain continues with the callback's own result. g must not be reused or mutated afterwards. A group cannot be the chain's first stage — use NewGroup directly when no preceding step exists.

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

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

func NewGroup() *Group

NewGroup returns an empty group builder.

Example

ExampleNewGroup fans out members in parallel, then fires OnComplete once all succeed. ThenGroup embeds this as a chain stage; AddChain makes a member a chain.

package main

import (
	"context"

	"github.com/redis/go-redis/v9"

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

type ResizeArgs struct {
	Path string `json:"path"`
}

func (ResizeArgs) Kind() string { return "img:resize" }

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	client := chronos.NewClient(rdb)
	_, _ = chronos.NewGroup().
		Add(ResizeArgs{Path: "a.png"}).
		AddChain(chronos.NewChain().Then(ResizeArgs{Path: "b.png"}).Then(ResizeArgs{Path: "b2.png"})).
		OnComplete(ResizeArgs{Path: "manifest"}).
		Enqueue(context.Background(), client)
}

func (*Group) Add

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) AddChain

func (g *Group) AddChain(ch *Chain) *Group

AddChain appends a chain member: its links run in sequence, and the chain's FINAL link reports the member's completion to this group (its result becomes this member's GroupResults entry). The chain may not contain a ThenGroup stage (one-level nesting only) and its links may not use WithUnique/ WithTaskID or WithDeadLetterDiscard.

func (*Group) Enqueue

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

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

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.

Example

ExampleNewInspector reads operational state: queue stats, task listing, pause/resume, scheduler status.

package main

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"

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

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	insp := chronos.NewInspector(rdb)
	queues, _ := insp.Queues(context.Background())
	for _, q := range queues {
		fmt.Printf("%s: pending=%d paused=%v\n", q.Queue, q.Pending, q.Paused)
	}
}

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.

For a task that does not exist or has already been processed it is an idempotent no-op and returns nil (unlike GetTask, it does not return ErrTaskNotFound).

func (*Inspector) GetTask

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) GroupMembers

func (i *Inspector) GroupMembers(ctx context.Context, cbQueue, groupID string) ([]string, error)

GroupMembers returns the IDs of a group's not-yet-succeeded members. cbQueue is the callback queue (a member TaskInfo carries it as GroupQueue).

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) PauseQueue

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

PauseQueue stops servers from consuming the queue (within about one second). Enqueueing, forwarding and recovery continue — work accumulates as pending.

The queue name is not validated: a name that does not exist is still accepted and stays in the paused list until ResumeQueue clears it, with no effect on any active queue.

func (*Inspector) PausedQueues

func (i *Inspector) PausedQueues(ctx context.Context) ([]string, error)

PausedQueues lists currently paused queue names.

func (*Inspector) Queues

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

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

func (*Inspector) ResumeQueue

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

ResumeQueue lifts a pause; consumption restarts within about one second.

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.

For a task that does not exist or has already been processed it is an idempotent no-op and returns nil (unlike GetTask, it does not return ErrTaskNotFound).

func (*Inspector) SchedulerStatus

func (i *Inspector) SchedulerStatus(ctx context.Context) (*SchedulerStatus, error)

SchedulerStatus returns the scheduler's leader and every known schedule: the registry (registered schedules, fired or not) merged by ID with the fire-history keys (which may also hold pre-registry leftovers).

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

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 (used for lookup and correlation). When omitted a random UUID is generated.

This is not deduplication: enqueueing twice with the same ID does not reject the second call — it may overwrite the existing task's body and enqueue it again. For content-based duplicate suppression use WithUnique.

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
	Paused    bool
}

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 ScheduleInfo

type ScheduleInfo struct {
	ID        string
	Kind      string // "" for fire-history-only entries (pre-registry leftovers)
	Queue     string
	Spec      string    // "@every 30s" or a 5-field cron expression
	LastFired time.Time // zero if the schedule never fired
	LastSeen  time.Time // zero for fire-history-only entries
	Stale     bool      // registry entry not refreshed within staleAfter
}

ScheduleInfo is one schedule's registry entry merged with its fire history.

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.

Example

ExampleNewScheduler registers a periodic job. Every instance may call Start; leader election ensures only one fires each trigger.

package main

import (
	"context"
	"time"

	"github.com/redis/go-redis/v9"

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

type EmailArgs struct {
	To string `json:"to"`
}

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

func main() {
	rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	s := chronos.NewScheduler(rdb, chronos.SchedulerConfig{})
	_ = chronos.RegisterInterval(s, time.Hour, EmailArgs{To: "digest@b.c"})
	_ = s.Start(context.Background())
	defer s.Shutdown(context.Background())
}

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 in which cron schedules are evaluated. Defaults to
	// time.Local. A CRON_TZ=/TZ= prefix on an individual spec takes precedence over
	// this for that schedule.
	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 SchedulerStatus

type SchedulerStatus struct {
	LeaderID  string
	Schedules []ScheduleInfo
}

SchedulerStatus reports the current scheduler leader and every schedule that has fired at least once (registered-but-never-fired schedules are invisible; a future schedule registry will list them).

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.
	//
	// At least one queue is required: there is no implicit default queue, and
	// Start returns an error if Queues is empty.
	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.

func (*Task[T]) RawGroupResults

func (t *Task[T]) RawGroupResults() [][]byte

RawGroupResults returns raw member results in Add order (nil = no result). Nil when this task is not a group callback or no member produced a result.

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)
	ChainIndex   int      // this task's 0-based position in its chain
	ChainNext    []string // kinds of the links waiting behind this task, in order

	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.
	GroupQueue   string // queue holding the group's state (= callback queue)

	// HasResult reports whether the handler produced a result (AddHandlerR).
	// The result itself travels the workflow (PrevResult/GroupResults); the
	// Inspector only reports its presence and size.
	HasResult bool
	// ResultSize is the result's JSON size in bytes (0 when HasResult=false).
	ResultSize int
}

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"
)