scheduler

scheduler is a Go library for reliable distributed delayed-job execution with Redis as the first production backend.

It is designed for systems that need:

• delayed jobs with run_at
• distributed workers across many processes or nodes
• bounded parallelism and backpressure
• at-least-once delivery with retries
• lease-based recovery when workers crash
• a storage abstraction that can support more backends over time

✨ At A Glance

• ⚙️ Library-first Go API with producer and worker runtimes
• 🐹 Go-native design with simple package-first integration
• 📦 Redis backend with batched hot paths for better coordination efficiency
• 🔁 Retry policies, lease renewals, dead-letter handling, and recovery loops
• 🧭 Versioned queue topology for safe repartitioning over time
• 🧪 In-memory backend, integration tests, and benchmark coverage
• 🛠 Runnable examples for Redis queue migration and Redis worker demos

🎯 Why scheduler

Many task queues are easy to start with, but become harder to reason about once you need:

• predictable retry behavior
• safe recovery after consumer failure
• bounded worker concurrency
• operational tuning for throughput and latency
• a clean separation between queue semantics and datastore details

scheduler focuses on those primitives first.

🚧 Current Status

This repository contains a strong v1 foundation:

• producer API for delayed jobs
• worker runtime with configurable concurrency
• explicit job lifecycle states
• batched Redis promotion, leasing, renewals, completions, and recovery
• internal queue partitioning for Redis
• in-memory backend for tests and local development
• integration tests and benchmark scaffolding

Not in scope for v1:

• cron scheduling
• workflow / DAG orchestration
• exactly-once processing
• dashboard / admin UI

📬 Delivery Model

The delivery contract is at-least-once.

That means:

• a job is expected to run one or more times
• duplicate execution is possible during failures or lease recovery
• handlers should be idempotent

This is the tradeoff that makes reliable distributed processing practical at scale.

🔄 How It Works

Each job moves through explicit states:

scheduled -> ready -> leased -> succeeded

or, on failure:

scheduled -> ready -> leased -> retry_wait -> ready

or, after retry exhaustion:

scheduled -> ready -> leased -> dead

Runtime flow

1. A producer enqueues a job with a future run_at.
2. Workers poll the store and promote due jobs from scheduled or retry_wait to ready.
3. Workers lease ready jobs in batches.
4. Leased jobs are processed in parallel up to configured limits.
5. Active leases are renewed in batches.
6. Completed jobs are acknowledged in batches.
7. Failed jobs are retried with backoff or moved to dead.
8. Expired leases are recovered automatically and returned to ready.

Distribution model

Multiple workers can safely share the same Redis backend because ownership is temporary and explicit:

• a worker does not permanently remove a job
• it leases the job for a bounded time
• if the worker crashes, another worker can recover the job after lease expiry

🧠 Redis Design

Redis is the first production backend and the current hot path is heavily batched:

• due-job promotion is batched
• ready-job leasing is batched
• lease renewal is batched
• completion flushes are batched
• expired lease recovery is batched

To reduce hot-key pressure, each logical queue is internally partitioned into multiple Redis key groups:

• scheduled
• ready
• leased
• retry

Jobs are assigned to a partition by hashing the job ID, while the public queue name stays stable for application code.

Queue routing is versioned. Each job stores:

• queue_version
• partition

That makes repartitioning safer because a running system can keep draining old queue layouts while new jobs are written to a newer version.

🗂 Project Layout

• producer.go Producer API
• runtime.go Worker runtime, batching, lease tracking
• store.go Storage abstraction
• memory_store.go In-memory backend
• redis_store.go Redis backend with batching and partitions
• bench_test.go Store and worker benchmarks
• examples/redis_demo/main.go End-to-end Redis demo with producer and worker
• examples/redis_migration/main.go Queue topology migration utility
• assets/logo.svg Primary project logo
• assets/scheduler-hero.svg README hero artwork
• assets/scheduler-flow.svg Lifecycle diagram artwork

📥 Installation

go get github.com/yuseferi/scheduler

The project targets Go 1.26+.

If you want the repo task runner used by many Go projects, this repository now includes a Taskfile.yml for Task. You can install Task from the official project, for example:

brew install go-task

or:

go install github.com/go-task/task/v3/cmd/task@latest

Official docs:

• Task installation
• Task getting started

🚀 Quick Start

package main

import (
	"context"
	"log/slog"
	"time"

	"github.com/yuseferi/scheduler"
)

func main() {
	store := scheduler.NewMemoryStore()

	producer, err := scheduler.NewProducer(store, scheduler.DefaultProducerConfig())
	if err != nil {
		panic(err)
	}

	worker, err := scheduler.NewWorker(
		store,
		scheduler.DefaultWorkerConfig("worker-1"),
		slog.Default(),
		scheduler.NopMetrics{},
	)
	if err != nil {
		panic(err)
	}

	worker.Handle("email.send", func(ctx context.Context, job *scheduler.Job) error {
		slog.Info("processing job", "id", job.ID, "type", job.Type)
		return nil
	})

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go func() {
		if err := worker.Start(ctx); err != nil {
			panic(err)
		}
	}()

	_, err = producer.Enqueue(context.Background(), scheduler.EnqueueOptions{
		Queue:      "default",
		Type:       "email.send",
		Payload:    []byte(`{"to":"team@example.com"}`),
		RunAt:      time.Now().Add(2 * time.Second),
		MaxRetries: 3,
		Timeout:    15 * time.Second,
	})
	if err != nil {
		panic(err)
	}

	time.Sleep(3 * time.Second)
}

🏃 Common Commands

Once Task is installed, the usual project workflow becomes:

task
task check
task ci
task demo
task bench-short
task cover
task test-integration

Useful tasks included in this repository:

• task fmt Format all Go source files with gofmt
• task fmt-check Fail if any Go files are not formatted
• task vet Run go vet ./...
• task lint Run the basic lint suite (fmt-check + vet)
• task test Run the default unit-focused test suite
• task test-integration Run the Redis-backed integration suite with -tags=integration
• task test-race Run unit-focused tests with the race detector
• task test-race-integration Run unit + Redis integration tests with the race detector
• task cover Run unit-focused library coverage output
• task cover-integration Run library coverage with Redis integration tests enabled
• task bench Run the benchmark suite
• task bench-short Run a short benchmark smoke test
• task tidy Run go mod tidy
• task check Run the standard local verification flow
• task ci Run the CI-style verification flow
• task demo Run the Redis demo app
• task migrate -- ... Forward commands to the Redis queue migration example

🧩 Using In Other Projects

scheduler is designed to be embedded into another Go service rather than forcing you into a separate platform from day one.

Typical ways to use it in another project:

1. Inside a monolith or API service

Use the same application to:

• accept requests
• enqueue background jobs
• run one or more worker loops

Good fit for:

• internal tools
• SaaS backends
• early-stage products
• systems where the API and workers can live together

2. Producer / worker split

Run producers in your API service and run workers in separate processes or deployments.

Good fit for:

• larger systems
• higher throughput workloads
• teams that want independent scaling for API nodes and job consumers

Example:

• api-service enqueues jobs into Redis
• worker-service runs scheduler workers against the same Redis backend

3. Shared infrastructure package

Wrap scheduler inside your own internal package so every service uses the same conventions.

Good fit for:

• platform teams
• multi-service Go environments
• organizations that want standardized retries, queue names, and metrics

Example:

package background

import (
	"log/slog"

	"github.com/redis/go-redis/v9"
	"github.com/yuseferi/scheduler"
)

func NewRedisScheduler(workerID string) (*scheduler.Producer, *scheduler.Worker, error) {
	client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	store := scheduler.NewRedisStoreWithConfig(client, scheduler.RedisConfig{
		Prefix:        "my-company-jobs",
		Partitions:    8,
		ActiveVersion: 1,
	})

	producer, err := scheduler.NewProducer(store, scheduler.DefaultProducerConfig())
	if err != nil {
		return nil, nil, err
	}

	worker, err := scheduler.NewWorker(
		store,
		scheduler.DefaultWorkerConfig(workerID),
		slog.Default(),
		scheduler.NopMetrics{},
	)
	if err != nil {
		return nil, nil, err
	}

	return producer, worker, nil
}

4. Feature-specific queues

Another project can use scheduler for different background domains without changing the core runtime.

Examples:

• email.send
• webhook.dispatch
• invoice.generate
• report.build
• image.process
• payment.sync

Each service can register handlers by job type and keep its own queue conventions.

Minimal integration pattern

In a real application, the common shape looks like this:

1. Initialize Redis and create a scheduler.Store
2. Create one Producer for enqueueing jobs
3. Create one or more Worker instances for processing jobs
4. Register handlers by job type
5. Start workers during application startup
6. Enqueue jobs from request handlers, domain services, or event consumers

Example:

client := redis.NewClient(&redis.Options{
	Addr: "127.0.0.1:6379",
})

store := scheduler.NewRedisStoreWithConfig(client, scheduler.RedisConfig{
	Prefix:        "my-app",
	Partitions:    8,
	ActiveVersion: 1,
})

producer, _ := scheduler.NewProducer(store, scheduler.DefaultProducerConfig())

workerCfg := scheduler.DefaultWorkerConfig("orders-worker-1")
workerCfg.Queues = []string{"default"}
workerCfg.Concurrency = 16
workerCfg.BatchSize = 64
workerCfg.MaxInFlight = 128

worker, _ := scheduler.NewWorker(store, workerCfg, slog.Default(), scheduler.NopMetrics{})

worker.Handle("order.confirmation.send", func(ctx context.Context, job *scheduler.Job) error {
	// decode payload
	// send email / push / webhook
	return nil
})

Recommended integration guidance

• Keep handlers idempotent because delivery is at-least-once
• Start with a small number of queues and partition counts, then benchmark
• Scale workers horizontally before over-tuning a single process
• Use your own package around scheduler if multiple services need the same setup
• Treat Redis prefix and queue topology as production infrastructure choices

🧪 Try The Redis Demo

There is a runnable demo at examples/redis_demo/main.go.

Start Redis locally, then run:

go run ./examples/redis_demo

What it does:

• connects to Redis on 127.0.0.1:6379
• configures a Redis-backed scheduler store
• starts a worker
• enqueues a handful of delayed email jobs
• logs processing activity, queue version, and partition placement

🧰 Using Redis

client := redis.NewClient(&redis.Options{
	Addr: "127.0.0.1:6379",
})

store := scheduler.NewRedisStoreWithConfig(client, scheduler.RedisConfig{
	Prefix:     "scheduler",
	Partitions: 8,
})

Recommended starting point:

• Partitions: 1 for simple development or small workloads
• Partitions: 8 or higher when benchmarking higher throughput

If you want to manage queue topology explicitly:

store := scheduler.NewRedisStoreWithConfig(client, scheduler.RedisConfig{
	Prefix:        "scheduler",
	Partitions:    4,
	ActiveVersion: 1,
})

_, err := store.PromoteQueueVersion(ctx, "default", scheduler.QueueVersionConfig{
	Version:    2,
	Partitions: 8,
})
if err != nil {
	panic(err)
}

There is also a small CLI example for topology changes:

go run ./examples/redis_migration topology -addr 127.0.0.1:6379 -prefix scheduler -queue default
go run ./examples/redis_migration promote -addr 127.0.0.1:6379 -prefix scheduler -queue default -version 2 -partitions 8
go run ./examples/redis_migration complete -addr 127.0.0.1:6379 -prefix scheduler -queue default -version 1

⚙️ Configuration

ProducerConfig

DefaultProducerConfig() gives you:

• DefaultQueue: "default"
• DefaultMaxRetries: 3
• DefaultTimeout: 30s
• exponential retry policy with: 1s -> 2s -> 4s ... up to 1m

WorkerConfig

Important knobs:

• WorkerID Unique worker identity
• Queues Logical queues this worker should consume
• Concurrency Number of handlers that may execute in parallel
• BatchSize Number of jobs to claim or flush per batch
• QueueBuffer Extra local buffer multiplier when leasing
• PollInterval How often workers promote, recover, and attempt to lease
• LeaseDuration Visibility / ownership window for a leased job
• HeartbeatInterval How often active leases are renewed
• MaxInFlight Maximum number of active jobs in one worker
• RetryPolicy Backoff and retry controls

Example:

cfg := scheduler.DefaultWorkerConfig("email-worker-1")
cfg.Concurrency = 32
cfg.BatchSize = 64
cfg.QueueBuffer = 2
cfg.MaxInFlight = 128
cfg.PollInterval = 5 * time.Millisecond
cfg.HeartbeatInterval = 50 * time.Millisecond
cfg.LeaseDuration = 500 * time.Millisecond

🛡 Reliability Notes

The worker runtime is designed around failure recovery:

• if a handler returns an error, the job is retried with backoff
• if retries are exhausted, the job moves to dead
• if a worker dies mid-flight, lease expiry allows another worker to recover the job
• if Redis is shared by many workers, lease ownership prevents simultaneous execution by healthy workers

Because the system is at-least-once, idempotency should be treated as part of the handler contract.

🔀 Repartitioning and Queue Migration

scheduler now supports versioned Redis queue topology.

That means partition changes should be done as a version transition, not by mutating one live layout in place.

Recommended migration flow

1. Start with an active queue version

Example:

• queue: default
• active version: v1
• partitions: 4

2. Promote a new active version

topology, err := store.PromoteQueueVersion(ctx, "default", scheduler.QueueVersionConfig{
	Version:    2,
	Partitions: 8,
})

After this:

• new jobs go to v2
• v1 becomes draining
• workers keep scanning both active and draining versions

3. Wait for the draining version to empty

Use queue stats and operational checks to confirm v1 is empty before completing migration.

4. Mark the draining version complete

topology, err = store.CompleteDrainingVersion(ctx, "default", 1)

After this:

• v2 remains active
• v1 metadata is removed
• workers stop scanning v1

Why this matters

Changing partition count in place can strand jobs because hash routing changes.

Versioned queue topology avoids that by:

• storing routing on each job
• keeping old and new layouts visible during migration
• allowing safe active + draining reads

CLI example

The repository includes a runnable example at examples/redis_migration/main.go.

Inspect current topology:

go run ./examples/redis_migration topology \
  -addr 127.0.0.1:6379 \
  -prefix scheduler \
  -queue default

Promote a new active version:

go run ./examples/redis_migration promote \
  -addr 127.0.0.1:6379 \
  -prefix scheduler \
  -queue default \
  -version 2 \
  -partitions 8

Remove a drained version after it is empty:

go run ./examples/redis_migration complete \
  -addr 127.0.0.1:6379 \
  -prefix scheduler \
  -queue default \
  -version 1

📊 Benchmarks

The project includes two main benchmark groups in bench_test.go:

• BenchmarkStoreLifecycle Measures enqueue -> promote -> lease -> renew -> complete
• BenchmarkWorkerThroughput Measures end-to-end runtime-driven processing

Run them with:

go test -run '^$' -bench 'Benchmark(StoreLifecycle|WorkerThroughput)' ./...

Short sample results from this machine:

Benchmark	Result
BenchmarkStoreLifecycle/memory/full-cycle	49 us/op
BenchmarkStoreLifecycle/redis/full-cycle/p1	2262 us/op
BenchmarkStoreLifecycle/redis/full-cycle/p8	2813 us/op
BenchmarkWorkerThroughput/memory/worker	91 us/op
BenchmarkWorkerThroughput/redis/worker/p1	1227 us/op
BenchmarkWorkerThroughput/redis/worker/p8	3525 us/op

These numbers were produced from a very short smoke run with -benchtime=1x, so they are useful for correctness and relative direction, not for capacity claims.

For meaningful sizing work, run longer benchmarks, for example:

go test -run '^$' -bench 'Benchmark(StoreLifecycle|WorkerThroughput)' -benchtime=3s ./...

✅ Testing

The repository now separates unit-style tests from Redis-backed integration tests.

Default local test run:

go test ./...

Redis integration test run:

go test -tags=integration ./...

Combined coverage run used in CI:

go test -tags=integration -coverprofile=coverage.out .

Integration tests do not connect to an already-running Redis instance. They use testcontainers-go to start an isolated ephemeral Redis just for the test run.

Local prerequisites for integration tests:

• Docker or another compatible container runtime available to testcontainers-go

If no compatible container runtime is available, the Redis integration tests are skipped. Coverage reporting intentionally targets the main library package and does not count the runnable examples/ binaries.

Run the Redis benchmarks only:

go test -run '^$' -bench BenchmarkStoreLifecycle/redis ./...

🌱 Roadmap Ideas

Strong next steps for the project:

• multi-worker contention benchmarks
• producer burst benchmarks
• Lua-based stronger atomicity for more Redis transitions
• benchmark presets and tuning guides
• more backends such as SQLite or Postgres
• queue-level rate limiting and priority support

🤝 Contributing

The project is still early and evolving quickly. Good contributions would include:

• correctness hardening
• higher-quality benchmarks
• Redis tuning and load-testing
• additional datastore implementations
• examples and production guidance

📄 License

This project is licensed under the MIT License.