kyu

kyu

A distributed job queue library for Go, backed by PostgreSQL and Redis.

kyu jobs visualized on Grafana

kyu is a Go-native distributed job queue whose primary design concern is that it be operable in production: dead-letter management, a live dashboard, and a CLI ship with the queue rather than as external tooling. PostgreSQL is the durable source of truth for every job, attempt, and error, while Redis exists only as the low-latency priority index that decides what runs next — a deliberate split that uses each backend for what it does best instead of either one doing everything.

PostgreSQL is the source of truth - every job, its full history, retry count, and error message are persisted there. Redis acts as the priority queue - workers pop job IDs from a sorted set and fetch the full record from Postgres to process. Jobs survive a Redis restart because nothing is lost if the sorted set is cleared.

---

Contents

• How it works
• Scope
• Installation
• Quick start
• Enqueuing jobs
• Config
• Registering handlers
• Middleware
• Job lifecycle
• Inspecting jobs
• Dead letter queue
• Operations
• Completion callbacks
• RunOnce
• Metrics
• CLI
  • Dashboard
• Benchmarks
• Docker Compose
• Running tests

How it works

Kyu runs five concurrent subsystems once you call Start.

The worker pool pops job IDs from a Redis sorted set, fetches the full job record from Postgres, runs the registered handler, then updates the job status. Failed jobs with retries remaining are re-queued with exponential backoff. Failed jobs with no retries left are marked dead (see Dead letter queue).

Jobs are claimed with optimistic locking. When a worker picks up a job it stamps the row running with its locked_by identity, and every transition out of that state - completed, failed, or dead - is an update guarded by the same locked_by value. If a worker's claim was lost because another worker reaped or re-claimed the job, its write is ignored rather than clobbering the current owner's state.

The scheduler ticks every SchedulerInterval and queries Postgres for scheduled or failed jobs whose time has arrived, pushing their IDs back into Redis.

The stale reaper ticks every ReaperInterval and resets any job stuck in the running state longer than StaleJobTimeout. This handles workers that crashed mid-job.

The orphan reaper ticks every OrphanCheckInterval and re-queues pending jobs that are missing from Redis. This covers jobs popped off the queue by a worker that crashed before it could mark them running (and jobs left behind if the Redis sorted set was ever cleared).

The metrics server exposes a Prometheus /metrics endpoint on MetricsPort.

---

Scope

kyu's boundaries are decisions, not gaps.

Handlers are Go functions compiled into the consumer's binary. There is no handler DSL, no sidecar runtime, and no sandbox layer. This is deliberate: handlers share the same toolchain, type system, and deployment pipeline as the application that enqueues jobs, and kyu takes on no protocol or security surface for executing untrusted code.

kyu is a job queue, not a workflow orchestrator. There is no job-dependency graph and no DAG engine. Jobs are independent units of work that may run concurrently; retries and backoff apply per job. A sequence of dependent steps is expressed in application code, with each handler enqueuing the next job in the chain when it completes.

---

Installation

The kyu library is a Go module; the kyu serve CLI ships with it.

As a library (embed the queue in your Go program):

go get github.com/codetesla51/kyu

As a CLI (installs the kyu binary on your GOBIN):

go install github.com/codetesla51/kyu/cmd/kyu@latest

As a binary (from the GitHub Releases page, built by CI from the version tag):

curl -sL https://github.com/codetesla51/kyu/releases/latest/download/kyu-linux-amd64 -o kyu
chmod +x kyu && ./kyu serve

Replace kyu-linux-amd64 with kyu-linux-arm64, kyu-darwin-amd64, kyu-darwin-arm64, or kyu-windows-amd64.exe for other platforms. A multi-arch Linux image is also published on GHCR: ghcr.io/codetesla51/kyu:v1.0.0.

Requires PostgreSQL and Redis. The jobs table and indexes are created by the embedded goose migrations, which run automatically on the first Connect call. If you manage the schema yourself, set DisableAutoMigrate: true (see Config). Queries are generated by sqlc from queries/.

---

Quick start

package main

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

    "github.com/codetesla51/kyu"
)

func main() {
    q := kyu.New(kyu.Config{
        DSN:         "postgres://user:pass@localhost:5432/mydb?sslmode=disable",
        RedisAddr:   "localhost:6380",
        Workers:     5,
        MetricsPort: 9090,
    })

    q.Register("send_email", func(ctx context.Context, payload string) error {
        log.Printf("sending email: %s", payload)
        return nil
    })

    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    if err := q.Connect(ctx); err != nil {
        log.Fatal(err)
    }

    if err := q.Start(ctx); err != nil {
        log.Fatal(err)
    }
}

Start blocks. On SIGINT or SIGTERM the context is cancelled, workers finish their current jobs, the scheduler and metrics server shut down cleanly, and Start returns.

The required call order is: New -> Register -> Connect -> Enqueue / Start. Connect must be called before Enqueue or Start - both require an open database and Redis connection.

---

Enqueuing jobs

Jobs are enqueued independently of Start - you can enqueue from a separate service, an HTTP handler, or anywhere you have access to the Queue.

// run immediately
jobID, err := q.Enqueue(ctx, "send_email", `{"to":"user@example.com"}`, kyu.EnqueueOptions{
    MaxRetries: 3,
    Priority:   1,
})

// run after 1 minute
at := time.Now().Add(1 * time.Minute)
jobID, err := q.Enqueue(ctx, "send_email", `{"to":"user@example.com"}`, kyu.EnqueueOptions{
    MaxRetries:  3,
    Priority:    0,
    ScheduledAt: &at,   // pointer so nil means "no schedule, run now"
    TimeOut:     10 * time.Second,
})

// high priority - processed before lower priority jobs
jobID, err := q.Enqueue(ctx, "process_payment", `{"order_id":"123"}`, kyu.EnqueueOptions{
    MaxRetries: 5,
    Priority:   10, // higher score = picked up first
})

ScheduledAt is a pointer because nil means "run immediately" and a real value means "run at this time". A plain time.Time cannot represent the absence of a value.

Priority maps directly to the Redis sorted set score. Workers always pop the highest score first, so higher numbers are processed before lower ones.

Enqueue multiple jobs atomically - a single COPY insert into Postgres and a single ZADD into Redis:

ids, err := q.EnqueueMany(ctx, []kyu.EnqueueRequest{
    {JobType: "send_email", Payload: `{"to":"a@example.com"}`, Options: kyu.EnqueueOptions{MaxRetries: 3, Priority: 1}},
    {JobType: "send_email", Payload: `{"to":"b@example.com"}`, Options: kyu.EnqueueOptions{MaxRetries: 3, Priority: 2}},
    {JobType: "process_payment", Payload: `{"order_id":"123"}`, Options: kyu.EnqueueOptions{MaxRetries: 5, Priority: 10}},
})

IDs are returned in the same order as the input requests.

---

Config

kyu.Config{
    // Required
    DSN:       "postgres://user:pass@localhost:5432/db?sslmode=disable",
    RedisAddr: "localhost:6380",

    // Worker pool
    Workers: 5, // number of concurrent goroutines processing jobs

    // Queue
    QueueName: "kyu:default", // Redis sorted set key - use different names to isolate queues

    // Metrics
    MetricsPort: 9090, // set to 0 to disable

    // Stale job reaper
    // A job stuck in "running" beyond this duration is reset to "pending"
    // and re-queued. This handles crashed workers.
    StaleJobTimeout: 10 * time.Minute,

    // Loop intervals. All default to sane values if left zero.
    SchedulerInterval:    5 * time.Second, // promotes scheduled/failed jobs whose time arrived
    ReaperInterval:       1 * time.Minute, // scans for stale running jobs
    OrphanCheckInterval: 1 * time.Minute,  // re-queues pending jobs missing from Redis

    // Completion callbacks (optional)
    // POSTs a JSON body {job_id, status, payload, error} to this URL whenever
    // a job completes. Empty disables callbacks.
    CallbackURL: "https://hooks.example.com/job-done",

    // Postgres connection pool
    MaxOpenConns:    25,
    MaxIdleConns:    25,
    ConnMaxLifetime: 5 * time.Minute,

    // Migrations (optional)
    // kyu applies its embedded goose migrations on Connect by default. Set to
    // true if you manage the kyu schema yourself (e.g. with your own migration
    // tool against the same database) to skip that step.
    DisableAutoMigrate: false,

    Logger: log.Default(),
}

All fields have defaults. kyu.New(kyu.Config{}) connects to local Postgres and Redis with 5 workers.

A note on auto-migrations: kyu shares the database's goose version table when it migrates. If the same database is managed by another goose-based project, the two migration histories can conflict. Either give kyu its own database or set DisableAutoMigrate: true and apply kyu's migrations yourself (they live in db/goose_migrations/).

If you are running multiple applications against the same Redis instance, set a unique QueueName per application. Workers compete for any job in their queue - two apps sharing the same queue name will process each other's jobs.

---

Registering handlers

Handlers receive the context and the payload string you passed at enqueue time. Return an error to trigger a retry (if retries remain) or mark the job dead (if none remain). Because handlers are plain Go functions compiled into the binary (see Scope), they share the application's toolchain and deployment lifecycle.

q.Register("send_email", func(ctx context.Context, payload string) error {
    var data struct {
        To      string `json:"to"`
        Subject string `json:"subject"`
    }
    if err := json.Unmarshal([]byte(payload), &data); err != nil {
        return err // will retry
    }
    return sendEmail(ctx, data.To, data.Subject)
})

q.Register("process_payment", func(ctx context.Context, payload string) error {
    // respect context cancellation for long-running work
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
    }
    return chargeCard(payload)
})

Handlers are safe to register concurrently. Registering the same job type twice overwrites the first handler.

---

Middleware

Middleware wraps every job execution regardless of type. Middlewares are applied in registration order - the first registered is the outermost wrapper. In the example below, the logging middleware runs first, then timing, so the log line appears before the duration line.

// order: logging wraps timing wraps handler
q.Use(loggingMiddleware)  // outermost
q.Use(timingMiddleware)   // inner

// logging
q.Use(func(ctx context.Context, jobType, payload string, next func() error) error {
    log.Printf("job started: %s", jobType)
    err := next()
    if err != nil {
        log.Printf("job failed: %s: %v", jobType, err)
    }
    return err
})

// timing
q.Use(func(ctx context.Context, jobType, payload string, next func() error) error {
    start := time.Now()
    err := next()
    log.Printf("job=%s duration=%s", jobType, time.Since(start))
    return err
})

// panic recovery
q.Use(func(ctx context.Context, jobType, payload string, next func() error) error {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("job panicked: %s: %v", jobType, r)
        }
    }()
    return next()
})

---

Job lifecycle

pending
   |
   |-- (scheduler promotes to Redis when scheduled_at is reached)
   |
   `-- running
          |
          |-- completed       handler returned nil
          |
          |-- failed          handler returned error, retries remain
          |      `-- re-enqueued with exponential backoff (1s, 2s, 4s, ...)
          |
          |-- dead            handler returned error, no retries left
          |
          `-- cancelled       CancelJob was called before the job ran

Failed jobs use exponential backoff between retries - a job that has failed once waits 1 second, twice waits 2 seconds, three times waits 4 seconds, and so on. The scheduler picks them back up once their scheduled_at arrives.

Jobs table columns:

Column	Description
id	UUID, primary key
job_type	matches the name passed to Register
payload	arbitrary string passed to the handler
status	pending, running, failed, completed, dead, cancelled
priority	higher score = picked up first
scheduled_at	job will not run until this time
max_retries	maximum retry attempts
retry_count	number of attempts so far
error_message	last error returned by the handler
locked_by	which worker is running it, e.g. worker-3
locked_at	when the worker locked it
completed_at	when it finished successfully

---

Inspecting jobs

// get a single job by ID
job, err := q.Inspect(ctx, jobID)
log.Printf("status=%s retries=%d error=%s", job.Status, job.RetryCount, job.ErrorMessage)

// get all jobs that exhausted their retries
dead, err := q.DeadJobs(ctx)
for _, j := range dead {
    log.Printf("dead: id=%s type=%s attempts=%d error=%s",
        j.ID, j.JobType, j.RetryCount, j.ErrorMessage)
}

// cancel a job that hasn't started yet
// works on: pending, scheduled, failed
// has no effect once a job is running
err := q.CancelJob(ctx, jobID)

---

Dead letter queue

Jobs that exhaust all retries are marked dead and stay persisted so you can inspect, retry, or purge them.

// list every dead job
dead, err := q.ListDead(ctx)
for _, j := range dead {
    log.Printf("dead: id=%s type=%s error=%s", j.ID, j.JobType, j.ErrorMessage)
}

// inspect a single dead job by ID
job, err := q.InspectDead(ctx, jobID)

// retry one job - resets retry_count, clears the error, and re-enqueues
// it with a fresh set of retries so it can run again immediately
err := q.Retry(ctx, jobID)

// retry every dead job, re-enqueuing them all on the given Redis queue
n, err := q.RetryAllDead(ctx, "kyu:retry-backlog")

// purge a dead job (soft delete - sets deleted_at)
err := q.DeleteDead(ctx, jobID)

Retry and RetryAllDead restore jobs to pending and push them back onto a queue with their original priority. RetryAllDead returns the number of jobs retried.

RetryAllDead takes the target queue as an argument. Retrying onto a separate queue, such as kyu:retry-backlog, isolates replayed jobs from live traffic: they are only processed by workers listening on that queue, so the replay can run on a dedicated retry worker or during a controlled maintenance window.

---

Operations

Runtime controls, health checks, and introspection for a running queue.

// stop the worker pool from popping new jobs; in-flight jobs finish first
q.Pause()
// let the pool pop again
q.Resume()
if q.IsPaused() { /* ... */ }

// health check - verifies both Postgres and Redis are reachable
if err := q.Ping(ctx); err != nil {
    log.Fatal(err)
}

Pause is per-process state: it stops this Queue instance's workers, while other processes sharing the queue keep working.

// point-in-time job counts plus this queue's Redis depth
stats, err := q.Stats(ctx)
log.Printf("pending=%d running=%d dead=%d depth=%d",
    stats.Pending, stats.Running, stats.Dead, stats.QueueDepth)

// list jobs, most recent first, filtered and paginated
jobs, err := q.ListJobs(ctx, kyu.JobFilter{Status: "failed", Limit: 50})
jobs, err = q.ListJobs(ctx, kyu.JobFilter{JobType: "send_email", Limit: 100, Offset: 100})

// sorted names of every registered job type
types := q.JobTypes()

// read-only config summary
info := q.Info()

// worker pool state and size
workers := q.Workers()
n := q.WorkerCount()
name := q.QueueName()

JobFilter matches any combination of Status and JobType; a zero Limit returns the 100 most recent matching jobs, and Offset pages through longer lists. The status counts in QueueStats are global to the database (all queues share the jobs table), while QueueDepth is the number of IDs waiting in this queue's Redis sorted set.

// reset a failed or cancelled job back to pending and re-enqueue it
err := q.Reset(ctx, jobID)

// soft-delete any job by ID (also removes it from the pending queue)
err := q.Delete(ctx, jobID)

// soft-delete every job in one status; returns the number affected
n, err := q.Purge(ctx, "completed")

Delete and Purge set deleted_at (soft delete) so rows disappear from listings but stay in the table. Reset applies to failed and cancelled jobs; dead jobs use Retry instead (see Dead letter queue). Purge accepts one of: pending, running, completed, failed, scheduled, cancelled, dead.

---

Completion callbacks

Set Config.CallbackURL and kyu POSTs a JSON webhook whenever a job completes:

{"job_id":"...","status":"completed","payload":"...","error":""}

The request is fire-and-forget (sent in a goroutine, 10s timeout) and failures are only logged, so callbacks never slow down or break job processing. An empty CallbackURL disables callbacks entirely.

---

RunOnce (cron mode)

RunOnce drains the current queue and returns instead of running a persistent loop. Use it when you want an external scheduler (cron, Kubernetes CronJob) to control when work happens rather than running workers continuously.

if err := q.Connect(ctx); err != nil {
    log.Fatal(err)
}
// processes everything currently in Redis, then returns
if err := q.RunOnce(ctx); err != nil {
    log.Fatal(err)
}

---

Metrics

When MetricsPort is set, a Prometheus /metrics endpoint is available on that port. Each Queue instance uses its own private Prometheus registry so multiple instances in the same process do not conflict.

Metric	Type	Description
kyu_jobs_total	counter	total jobs ever submitted
kyu_jobs_processed_total	counter vec	completed jobs, labelled by status
kyu_job_failures_total	counter vec	failures, labelled by job_type
kyu_jobs_dead_total	counter	jobs that exhausted all retries
kyu_queue_depth	gauge	jobs currently waiting in Redis

Prometheus scrape config:

scrape_configs:
  - job_name: kyu
    static_configs:
      - targets: ["localhost:9090"]

---

CLI

cmd/kyu builds a small CLI that runs a self-contained instance and manages jobs against any kyu queue.

go run ./cmd/kyu --help

Command	Description
kyu serve	Run workers, scheduler, reapers, metrics, and the web dashboard. This is the default command, so bare kyu also works.
kyu enqueue <type> [payload]	Enqueue a job and print its ID. Flags: --priority, --retries, --schedule (RFC3339), --timeout, --count.
kyu inspect <id>	Print the details of a job. Add --dead to look it up in the dead letter queue.
kyu version	Print the kyu version and Go runtime.

Connection settings are persistent flags (--dsn, --redis-addr, --redis-password, --queue), each with a matching environment variable. For example:

# spin up a queue with the web dashboard
kyu serve --queue my:queue --workers 8

# enqueue a job for it from anywhere
kyu enqueue send_email '{"to":"user@example.com"}' --retries 3 --priority 5

# see what happened to it
kyu inspect <job_id>

kyu serve is intended for exploring the system and for simple deployments. Production workloads embed the library in their own binary and register real handlers with Register; the demo handlers that ship with serve are examples, not production handlers.

Dashboard

A web UI for monitoring and managing a running kyu queue. kyu serve starts it by default on the address given by --dashboard-addr, and it can also be embedded in any Go program via the dashboard package: http.Handle("/", dashboard.Handler(q)). The UI and its JSON API are embedded in the binary, so there is nothing to install beyond a Postgres and Redis reachable from the machine running it.

kyu serve --dsn "$DATABASE_URL" --redis-addr localhost:6380

The dashboard connects to the default local stack (postgres://localhost:5432/kyu, localhost:6380) and serves on :8080. Every setting can be a flag or an environment variable:

Variable	Default	Flag	Purpose
DASHBOARD_ADDR	:8080	--dashboard-addr	HTTP listen address for the UI
DATABASE_URL	postgres://localhost:5432/kyu?...	--dsn	Postgres DSN
REDIS_ADDR	localhost:6379	--redis-addr	Redis address
REDIS_PASSWORD	(empty)	--redis-password	Redis password
KYU_QUEUE	kyu:default	--queue	Redis queue key this instance owns
KYU_WORKERS	4	--workers	number of worker goroutines
KYU_METRICS_PORT	9090	--metrics-port	Prometheus metrics port
KYU_STALE_TIMEOUT	30s	--stale-timeout	stale running-job reset threshold
KYU_ORPHAN_INTERVAL	30s	--orphan-interval	orphan reaper tick interval

Features:

• Live overview - stats, per-status totals, queue depth, and a jobs stream pushed over SSE; pause/resume to inspect a point in time.
• Jobs - searchable and filterable by status and job type, with a canvas and a table view, paginated (25/50/100 per page, or "Load more" to page through more than the live snapshot).
• Dead letter queue - inspect dead jobs, retry individually or all at once, or purge.
• Workers - see each worker's busy state and current job.
• Create - enqueue jobs straight from the UI (type, JSON payload, priority, max retries, schedule).
• Tools - purge jobs by status.

kyu serve registers a small set of demo handlers so the dashboard has work to show: order_created, send_email, failing_job (fails every run to exercise retries and the dead letter queue), and flaky_job (fails twice then succeeds). Enqueuing any other job type will exhaust retries and land in the dead letter queue with unknown job type - register your own handlers in your own binary (see Registering handlers) or embed dashboard.Handler(q) in your server.

---

Benchmarks

Measured on an Intel Core i5-6300U (4 cores, 2.4GHz).

BenchmarkRegister               ~52 ns/op     0 B/op    0 allocs/op
BenchmarkExecute                ~950 ns/op  320 B/op    5 allocs/op
BenchmarkExecuteWithMiddleware  ~1.2 µs/op  480 B/op    7 allocs/op
BenchmarkExecuteParallel        ~600 ns/op  320 B/op    5 allocs/op

Job dispatch runs in under one microsecond. Zero allocations on Register. Each additional middleware layer costs one closure allocation (~160 bytes). Under parallel load the registry mutex shows no measurable contention. In practice throughput is bounded by Postgres write latency and Redis round-trip time, not by the dispatch path.

Load testing (Barrage)

Load tested with Barrage, a cross-layer HTTP/Postgres/Redis load tester. On a single untuned Postgres instance, Kyu sustains ~750 writes/sec with 100% success and zero job loss (DB mean ~166ms, p99 ~1s); Redis never became a bottleneck in any run (p99 <70ms even at Postgres's worst). Clean final result — 60s run, 20s ramp, concurrency 50, at 300 enqueues/s, 900 DB ops/s, 1500 Redis cmds/s:

Runner	Requests	Success	Rate	P50	P95	P99	Max
HTTP POST /api/jobs	15,000	100%	251.5/s	52ms	169ms	275ms	1.57s
Postgres	44,992	100%	749.9/s	96ms	547ms	966ms	3.37s
Redis	74,999	100%	1250.0/s	3.7ms	21.7ms	41.9ms	377ms

The journey surfaced two non-obvious findings worth knowing before you deploy at nontrivial concurrency:

• The connection ceiling bites before the write path. Stock Postgres (max_connections=100) against a large app pool + direct DB clients fails first with too many clients errors and looks like a throughput problem. Raise max_connections to match pool sizing (KYU_MAX_OPEN_CONNS).
• An index was costing writes. A partial (status) WHERE deleted_at IS NULL index showed no read benefit under this write-heavy test shape and was pure write tax on every INSERT/UPDATE — removing it (plus clearing test bloat) produced a measured ~3× latency improvement (DB p99 2999ms→966ms, mean 431ms→166ms; HTTP p99 579ms→275ms) at identical load.

The full investigation log — every run, the pool-ceiling diagnosis, the index EXPLAIN ANALYZE work, and the clean before/after isolation — is in benchmarks/README.md. Reproducible config: benchmarks/load_test.yaml. Full HTML reports: report.html at the repo root (Barrage's live output, the most recent run) and benchmarks/reports/run4-clean.html (archived copy of that final clean run).

Honesty note: this measures cross-layer latency/throughput correlation under simultaneous combined load — not a single-job causal trace from enqueue to completion.

---

Docker Compose

A docker-compose.yml is included with Postgres, Redis, Prometheus, Grafana, and the kyu serve CLI running the demo handlers with the web dashboard on port 8080. Grafana is provisioned automatically on startup (grafana/provisioning) with a Prometheus datasource and a pre-built dashboard covering queue depth, job throughput, failure rates by job type, goroutine count, and memory usage. Default Grafana login is admin/admin.

docker compose up --build

Then open the dashboard at localhost:8080 and enqueue a job from the Create form or with kyu enqueue send_email '{"to":"demo@example.com"}'.

Service	Address
Dashboard	localhost:8080
Prometheus	localhost:9090
Grafana	localhost:3000
Postgres	localhost:5432
Redis	localhost:6380

---

Running tests

Unit tests - no infrastructure required:

go test -short ./...

Integration tests - requires Postgres on 5432 and Redis on 6380.

Before running, update the DSN and Redis address in kyu_unit_test.go to match your local setup (both the unit and integration tests live in this single file). The default credentials in the test file are for local development only.

go test ./...

Benchmarks:

go test -bench=. -benchmem -count=3