scheduler

About scheduler

This large README is shown as a plain-text preview. Read the full README at the source

# scheduler

<p align="center">
  <img src="assets/logo.svg" alt="scheduler logo" width="160" />
</p>

<p align="center">
  <img src="assets/scheduler-hero.svg" alt="scheduler hero" width="100%" />
</p>

<p align="center">
  <a href="https://pkg.go.dev/github.com/yuseferi/scheduler"><img src="https://pkg.go.dev/badge/github.com/yuseferi/scheduler.svg" alt="Go Reference" /></a>
  <a href="https://goreportcard.com/report/github.com/yuseferi/scheduler"><img src="https://goreportcard.com/badge/github.com/yuseferi/scheduler" alt="Go Report Card" /></a>
  <a href="https://codecov.io/gh/yuseferi/scheduler"><img src="https://codecov.io/gh/yuseferi/scheduler/branch/main/graph/badge.svg" alt="Codecov" /></a>
  <a href="https://github.com/yuseferi/scheduler/actions/workflows/ci.yml"><img src="https://github.com/yuseferi/scheduler/actions/workflows/ci.yml/badge.svg" alt="CI" /></a>
  <a href="LICENSE"><img src="https://img.shields.io/badge/license-MIT-green.svg" alt="MIT License" /></a>
</p>

`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

![scheduler lifecycle](assets/scheduler-flow.svg)

## 🧠 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.go)
  Producer API
- [runtime.go](runtime.go)
  Worker runtime, batching, lease tracking
- [store.go](store.go)
  Storage abstraction
- [memory_store.go](memory_store.go)
  In-memory backend
- [redis_store.go](redis_store.go)
  Redis backend with batching and partitions
- [bench_test.go](bench_test.go)
  Store and worker benchmarks
- [examples/redis_demo/main.go](examples/redis_demo/main.go)
  End-to-end Redis demo with producer and worker
- [examples/redis_migration/main.go](examples/redis_migration/main.go)
  Queue topology migration utility
- [assets/logo.svg](assets/logo.svg)
  Primary project logo
- [assets/scheduler-hero.svg](assets/scheduler-hero.svg)
  README hero artwork
- [assets/scheduler-flow.svg](assets/scheduler-flow.svg)
  Lifecycle diagram artwork

## 📥 Installation

```bash
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`](Taskfile.yml) for [Task](https://taskfile.dev/). You can install Task from the official project, for example:

```bash
brew install go-task
```

or:

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

Official docs:

- [Task installation](https://taskfile.dev/docs/installation)
- [Task getting started](https://taskfile.dev/docs/getting-started)

## 🚀 Quick Start

```go
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":"[email protected]"}`),
		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:

```bash
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:

```go
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:

```go
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](examples/redis_demo/main.go).

Start Redis locally, then run:

```bash
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

```go
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:

```go
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:

```bash
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:

```go
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

```go
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

```go
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](examples/redis_migration/main.go).

Inspect current topology:

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

Promote a new active version:

```bash
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:

```bash
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](bench_test.go):

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

Run them with:

```bash
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:

```bash
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:

```bash
go test ./...
```

Redis integration test run:

```bash
go test -tags=integration ./...
```

Combined coverage run used in CI:

```bash
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:

```bash
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

Frequently Asked Questions