kronosq

Redis-backed task queue library for Go

kronosq is a reliable, Redis-backed distributed task queue for Go. It provides a simple, familiar API for enqueueing background jobs, processing them with a scalable worker pool, and observing them through an embedded web dashboard — all without running a separate process.

Features

Simple, familiar API — ergonomics inspired by asynq; easy to adopt
Redis-backed — persistent, reliable task storage using Redis lists and sorted sets
At-least-once delivery — tasks survive worker crashes and are re-queued on restart; handlers should be idempotent
Automatic retries with exponential backoff — failed tasks are retried up to a configurable limit
Scheduled tasks — delay individual tasks or run them on a cron schedule
Dynamic worker auto-scaling — worker pool scales up under load and idles back down automatically
Task lifecycle event hooks — Slack, Discord, and Telegram notifiers built-in; bring your own with a simple function signature
Embedded dashboard — React SPA served directly from your binary; no separate process needed
Priority queues with weighted processing — route tasks to named queues with relative weights
Strict priority scheduling — StrictPriority mode guarantees higher-weight queues are always drained before lower-priority ones
Task processing deadlines — hard-cancel a task's context at an absolute wall-clock time with DeadlineAt
Pause & resume queues — temporarily halt processing for any queue without stopping workers
Middleware support — attach logging, recovery, or custom interceptors to ServeMux
Redis Sentinel & Cluster — connect to highly-available Redis topologies with no code changes
CLI tool (kronq) — inspect queues, browse tasks, re-run failures, pause/resume queues from the terminal

Architecture

Requirements

Go 1.21+
Redis 6.2+

Installation

go get github.com/Azzurriii/kronosq

Quick Start

The following example shows the two sides of kronosq: a producer that enqueues a task, and a worker server that processes it.

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"

    "github.com/Azzurriii/kronosq"
)

// --- Producer ---

func main() {
    redisOpt := kronosq.RedisClientOpt{Addr: "localhost:6379"}

    client := kronosq.NewClient(redisOpt)
    defer client.Close()

    payload, _ := json.Marshal(map[string]any{"user_id": 42})
    task := kronosq.NewTask("email:welcome", payload)

    if err := client.Enqueue(task, kronosq.Queue("default"), kronosq.MaxRetry(3)); err != nil {
        log.Fatal(err)
    }
    fmt.Println("task enqueued")
}

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/Azzurriii/kronosq"
)

// --- Worker server ---

func main() {
    redisOpt := kronosq.RedisClientOpt{Addr: "localhost:6379"}

    srv := kronosq.NewServer(kronosq.ServerConfig{
        Redis:  redisOpt,
        Queues: map[string]int{"default": 1},
    })

    mux := kronosq.NewServeMux()
    mux.HandleFunc("email:welcome", func(ctx context.Context, t *kronosq.Task) error {
        fmt.Printf("sending welcome email — payload: %s\n", t.Payload)
        return nil
    })

    if err := srv.Run(mux); err != nil {
        log.Fatal(err)
    }
}

Start Redis, then run both programs. The server blocks until SIGINT or SIGTERM and shuts down gracefully.

make up          # start Redis via Docker Compose
go run ./example # run the bundled example app

Task Options

Options are passed as variadic arguments to client.Enqueue.

Option	Type	Default	Description
`Queue(name string)`	`string`	`"default"`	Route the task to a named queue
`MaxRetry(n int)`	`int`	`25`	Maximum retry attempts before archiving
`Delay(d time.Duration)`	`duration`	`0`	Schedule the task to run after a delay
`Timeout(d time.Duration)`	`duration`	`30m`	Per-task processing timeout
`DeadlineAt(t time.Time)`	`time.Time`	zero	Hard cancel at an absolute wall-clock time; takes precedence over `Timeout` if sooner
`Unique(ttl time.Duration)`	`duration`	`0`	Deduplicate tasks by type within a TTL window

client.Enqueue(task,
    kronosq.Queue("critical"),
    kronosq.MaxRetry(5),
    kronosq.Delay(10*time.Minute),
    kronosq.Timeout(2*time.Minute),
    kronosq.DeadlineAt(time.Now().Add(1*time.Hour)),
    kronosq.Unique(1*time.Hour),
)

Priority Queues

Assign integer weights to queues. The processor polls higher-weight queues more frequently in proportion to their weight — a critical task is processed roughly 6 times as often as a low task.

srv := kronosq.NewServer(kronosq.ServerConfig{
    Redis: redisOpt,
    Queues: map[string]int{
        "critical": 6,
        "default":  3,
        "low":      1,
    },
})

Route a task to a specific queue at enqueue time:

client.Enqueue(task, kronosq.Queue("critical"))

Strict priority

Enable StrictPriority to guarantee that queues are always checked in weight order. Workers will not poll default until critical is empty.

srv := kronosq.NewServer(kronosq.ServerConfig{
    Redis:          redisOpt,
    StrictPriority: true,
    Queues: map[string]int{
        "critical": 3,
        "default":  2,
        "low":      1,
    },
})

Scheduled Tasks (Cron)

Static schedule with Scheduler

Register tasks at startup time using standard cron expressions:

scheduler := kronosq.NewScheduler(redisOpt)

task := kronosq.NewTask("report:generate", nil)
scheduler.Add("*/5 * * * *", task, kronosq.Queue("default"))

if err := scheduler.Run(); err != nil {
    log.Fatal(err)
}

Dynamic schedule with PeriodicTaskManager

For cron schedules that change at runtime (e.g., stored in a database), implement the PeriodicTaskConfigProvider interface:

type DBProvider struct{ db *sql.DB }

func (p *DBProvider) GetConfigs() ([]*kronosq.PeriodicTaskConfig, error) {
    // Load cron configs from your database or any external source.
    return []*kronosq.PeriodicTaskConfig{
        {
            Cronspec: "0 9 * * 1-5",
            Task:     kronosq.NewTask("report:daily", nil),
            Opts:     []kronosq.Option{kronosq.Queue("default")},
        },
    }, nil
}

mgr := kronosq.NewPeriodicTaskManager(redisOpt, &DBProvider{db: db})
if err := mgr.Run(); err != nil {
    log.Fatal(err)
}

The manager re-reads the provider on each tick, so schedule changes take effect without a restart.

Middleware

Attach middleware to ServeMux to wrap every handler with cross-cutting concerns such as logging, recovery, or tracing. Middleware is applied in registration order — first added is outermost.

mux := kronosq.NewServeMux()

// Built-in middleware
mux.Use(kronosq.LoggingMiddleware)
mux.Use(kronosq.RecoveryMiddleware)

// Custom middleware
mux.Use(func(next kronosq.Handler) kronosq.Handler {
    return kronosq.HandlerFunc(func(ctx context.Context, t *kronosq.Task) error {
        start := time.Now()
        err := next.ProcessTask(ctx, t)
        log.Printf("task %s took %s", t.Type, time.Since(start))
        return err
    })
})

mux.HandleFunc("email:welcome", handleWelcome)

Pause & Resume Queues

Pause a queue to temporarily stop workers from picking up new tasks, without shutting down the server. Tasks already in-flight complete normally.

inspector := kronosq.NewInspector(redisOpt)

inspector.PauseQueue("low")   // workers skip this queue
inspector.ResumeQueue("low")  // processing resumes

Pause and resume are also available in the dashboard UI and via kronq:

kronq queues pause low
kronq queues resume low

Event Hooks

kronosq fires events at each stage of a task's lifecycle. Register handlers on the server to react to these events — for example, to send an alert when a task fails.

import (
    "github.com/Azzurriii/kronosq"
    "github.com/Azzurriii/kronosq/pkg/notify"
)

srv := kronosq.NewServer(cfg)

// Notify Slack on failure.
srv.OnEvent(kronosq.EventFailed, notify.Slack("https://hooks.slack.com/services/..."))

// Notify Discord on success.
srv.OnEvent(kronosq.EventSucceeded, notify.Discord("https://discord.com/api/webhooks/..."))

// Notify Telegram on retry.
srv.OnEvent(kronosq.EventRetried, notify.Telegram("<bot-token>", "<chat-id>"))

Event types

Constant	Fires when
`kronosq.EventStarted`	A worker picks up the task and begins processing
`kronosq.EventSucceeded`	The handler returns `nil`
`kronosq.EventFailed`	The handler returns an error and retries are exhausted
`kronosq.EventRetried`	The handler returns an error but retries remain
`kronosq.EventArchived`	The task is moved to the archive (permanent failure)

Custom notifiers

Any function matching func(ctx context.Context, event kronosq.TaskEvent) error is a valid EventHandler:

srv.OnEvent(kronosq.EventFailed, func(ctx context.Context, e kronosq.TaskEvent) error {
    log.Printf("task %s failed after %s: %v", e.Task.ID, e.Latency, e.Error)
    return nil
})

Events fire asynchronously and never block task processing.

Auto-Scaling

kronosq can automatically grow and shrink the worker pool in response to queue depth. Configure it via AutoScaleConfig in ServerConfig:

srv := kronosq.NewServer(kronosq.ServerConfig{
    Redis:  redisOpt,
    Queues: map[string]int{"default": 3, "critical": 6},
    AutoScale: &kronosq.AutoScaleConfig{
        Min:              2,
        Max:              20,
        ScaleUpThreshold: 2.0,
        IdleTimeout:      30 * time.Second,
        CheckInterval:    5 * time.Second,
    },
})

Scale-up: when pendingTasks > activeWorkers × ScaleUpThreshold, workers are added up to Max.

Scale-down: when a worker has been idle longer than IdleTimeout, it is removed down to Min.

The auto-scaler is a pure goroutine/semaphore mechanism with no external dependencies. When AutoScale is omitted, concurrency defaults to the sum of all queue weights.

Redis High Availability

kronosq supports Redis Sentinel and Redis Cluster through the RedisConnOpt interface. Switch topologies without changing any other code.

Redis Sentinel

srv := kronosq.NewServer(kronosq.ServerConfig{
    Redis: kronosq.RedisSentinelOpt{
        MasterName:    "mymaster",
        SentinelAddrs: []string{"sentinel1:26379", "sentinel2:26379"},
    },
    Queues: map[string]int{"default": 1},
})

Redis Cluster

srv := kronosq.NewServer(kronosq.ServerConfig{
    Redis: kronosq.RedisClusterOpt{
        Addrs: []string{"node1:7000", "node2:7001", "node3:7002"},
    },
    Queues: map[string]int{"default": 1},
})

NewClient, NewInspector, NewScheduler, and NewPeriodicTaskManager all accept the same RedisConnOpt interface.

Dashboard

Mount the embedded dashboard on your existing HTTP server. It serves a React SPA and a REST API, both compiled into your binary — no separate deployment needed.

import (
    "net/http"

    "github.com/Azzurriii/kronosq"
    "github.com/Azzurriii/kronosq/pkg/dashboard"
)

inspector := kronosq.NewInspector(redisOpt)
http.Handle("/dashboard/", dashboard.Handler(inspector))

http.ListenAndServe(":8080", nil)
// Visit: http://localhost:8080/dashboard/

Dashboard REST API

Method	Route	Description
`GET`	`/dashboard/api/queues`	List all queues with stats and pause state
`GET`	`/dashboard/api/queues/:name/tasks`	Browse tasks with status filter and pagination
`POST`	`/dashboard/api/queues/:name/pause`	Pause a queue
`POST`	`/dashboard/api/queues/:name/resume`	Resume a paused queue
`GET`	`/dashboard/api/servers`	List active worker server instances
`POST`	`/dashboard/api/tasks/:id/run`	Re-enqueue an archived or failed task
`DELETE`	`/dashboard/api/tasks/:id`	Remove a task from Redis

The SPA is built with Vite + React. To rebuild it from source:

make ui   # outputs to pkg/dashboard/ui/dist/, embedded on next go build

CLI (kronq)

kronq is the command-line interface for managing queues and tasks against a live Redis instance.

Installation

# With Go
go install github.com/Azzurriii/kronosq/cmd/kronq@latest

# Or with curl
curl -fsSL https://raw.githubusercontent.com/Azzurriii/kronosq/main/scripts/install.sh | sh

Commands

Command	Description
`kronq queues`	List all queues and their stats (pending, active, failed counts)
`kronq queues pause <name>`	Pause a queue
`kronq queues resume <name>`	Resume a paused queue
`kronq tasks <queue>`	List tasks in a queue; filter by `--status pending\|active\|failed`
`kronq task inspect <id> -q <queue>`	Show full task details including payload
`kronq task run <id> -q <queue>`	Re-enqueue a failed task
`kronq task delete <id> -q <queue>`	Remove a task from Redis
`kronq servers`	List active worker server instances
`kronq version`	Print the CLI version
`kronq help [command]`	Show help for a command

Global flags

--redis   Redis address (default: localhost:6379, env: KRONQ_REDIS)
-o        Output format: table, json (default: table)

Examples

# List all queues
kronq queues

# Pause the low-priority queue temporarily
kronq queues pause low

# Show pending tasks in the critical queue
kronq tasks critical --status pending

# Inspect a specific task
kronq task inspect 3f2a1b4c -q critical

# Re-run a failed task
kronq task run 3f2a1b4c -q critical

# Retry all failed tasks in a queue
kronq tasks default --status failed --retry-all

# Check queue health — exits 1 if any queue has failures (useful in CI)
kronq queues --check-healthy

# Output as JSON
kronq queues -o json

Configuration Reference

ServerConfig

Field	Type	Default	Description
`Redis`	`RedisConnOpt`	required	Redis connection (`RedisClientOpt`, `RedisSentinelOpt`, or `RedisClusterOpt`)
`Queues`	`map[string]int`	`{"default": 1}`	Queue names and their relative processing weights
`StrictPriority`	`bool`	`false`	When true, higher-weight queues are always drained before lower-weight ones
`AutoScale`	`*AutoScaleConfig`	`nil` (disabled)	Worker pool auto-scaling; if nil, concurrency equals the sum of queue weights

AutoScaleConfig

Field	Type	Default	Description
`Min`	`int`	`2`	Minimum number of active workers
`Max`	`int`	required	Maximum number of active workers
`ScaleUpThreshold`	`float64`	`2.0`	Ratio of pending tasks to active workers that triggers a scale-up
`IdleTimeout`	`time.Duration`	`30s`	Duration a worker must be idle before it is removed
`CheckInterval`	`time.Duration`	`5s`	How often the auto-scaler evaluates queue depth

RedisClientOpt

Field	Type	Default	Description
`Addr`	`string`	`"localhost:6379"`	Redis server address
`Password`	`string`	`""`	Redis password (AUTH)
`DB`	`int`	`0`	Redis database index

RedisSentinelOpt

Field	Type	Description
`MasterName`	`string`	Sentinel master name
`SentinelAddrs`	`[]string`	Addresses of sentinel nodes
`Password`	`string`	Redis password
`DB`	`int`	Redis database index

RedisClusterOpt

Field	Type	Description
`Addrs`	`[]string`	Cluster node addresses
`Password`	`string`	Redis password

How it works

Client.Enqueue writes tasks to Redis lists and sorted sets. The Processor pool pulls tasks via BLMOVE, skipping any queue in the paused set. ServeMux routes each task by type through the middleware chain to the registered handler. On completion, EventBus fires registered notifiers asynchronously so they never block processing. The AutoScaler polls queue depth on an independent ticker and adjusts the goroutine pool size between Min and Max. The dashboard and kronq CLI read and write Redis state through the Inspector.

License

MIT — see LICENSE.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
cmd/kronq		cmd/kronq
example		example
internal		internal
pkg		pkg
scripts		scripts
ui		ui
.gitignore		.gitignore
.goreleaser.yaml		.goreleaser.yaml
CLAUDE.md		CLAUDE.md
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
architecture.png		architecture.png
client.go		client.go
client_test.go		client_test.go
doc.go		doc.go
docker-compose.yml		docker-compose.yml
event.go		event.go
go.mod		go.mod
go.sum		go.sum
inspector.go		inspector.go
inspector_test.go		inspector_test.go
options.go		options.go
redis_opt.go		redis_opt.go
scheduler.go		scheduler.go
scheduler_test.go		scheduler_test.go
servemux.go		servemux.go
servemux_test.go		servemux_test.go
server.go		server.go
server_test.go		server_test.go
task.go		task.go

Latest commit

History

Folders and files

Repository files navigation

kronosq

Table of Contents

Features

Architecture

Requirements

Installation

Quick Start

Task Options

Priority Queues

Strict priority

Scheduled Tasks (Cron)

Static schedule with Scheduler

Dynamic schedule with PeriodicTaskManager

Middleware

Pause & Resume Queues

Event Hooks

Event types

Custom notifiers

Auto-Scaling

Redis High Availability

Redis Sentinel

Redis Cluster

Dashboard

Dashboard REST API

CLI (kronq)

Installation

Commands

Global flags

Examples

Configuration Reference

ServerConfig

AutoScaleConfig

RedisClientOpt

RedisSentinelOpt

RedisClusterOpt

How it works

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages