kronosq

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.

Table of Contents

• Features
• Architecture
• Installation
• Quick Start
• Task Options
• Priority Queues
• Scheduled Tasks
• Middleware
• Pause & Resume Queues
• Event Hooks
• Auto-Scaling
• Redis High Availability
• Dashboard
• CLI (kronq)
• Configuration Reference
• How it works
• License

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.

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

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

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

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

AutoScaleConfig

RedisClientOpt

RedisSentinelOpt

RedisClusterOpt

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.