scheduler

package module
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 4, 2026 License: MIT Imports: 16 Imported by: 0

README

scheduler

scheduler logo

scheduler hero

Go Reference Go Report Card Codecov CI MIT License

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

🧠 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

📥 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:

🚀 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
})
  • 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.

  1. Start with an active queue version

Example:

  • queue: default
  • active version: v1
  • partitions: 4
  1. 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
  1. Wait for the draining version to empty

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

  1. 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.

Documentation

Overview

Package scheduler provides delayed job scheduling with distributed workers, retries, lease-based processing, and a pluggable storage abstraction.

Delivery semantics are at-least-once. Handlers should be idempotent.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrJobNotFound indicates that a referenced job record does not exist.
	ErrJobNotFound = errors.New("job not found")
	// ErrLeaseConflict indicates that the caller does not own the current lease.
	ErrLeaseConflict = errors.New("lease conflict")
	// ErrHandlerNotFound indicates that no worker handler was registered for a job type.
	ErrHandlerNotFound = errors.New("handler not found")
	// ErrDuplicateJobID indicates that a job with the same ID already exists.
	ErrDuplicateJobID = errors.New("duplicate job id")
	// ErrInvalidJobState indicates that a job transition was attempted from an invalid state.
	ErrInvalidJobState = errors.New("invalid job state")
	// ErrDuplicateKey indicates that an idempotency key is already associated with another job.
	ErrDuplicateKey = errors.New("duplicate idempotency key")
	// ErrStoreRequired indicates that a producer or worker was created without a store.
	ErrStoreRequired = errors.New("store is required")
)

Functions

This section is empty.

Types

type EnqueueOptions 

type EnqueueOptions struct {
	ID             string
	Queue          string
	Type           string
	Payload        []byte
	Metadata       map[string]string
	IdempotencyKey string
	RunAt          time.Time
	Timeout        time.Duration
	MaxRetries     int
}

EnqueueOptions describes a job to be created by a Producer.

func (EnqueueOptions) Validate 

func (o EnqueueOptions) Validate(now time.Time) error

Validate verifies enqueue options before a job is created.

type Handler 

type Handler func(context.Context, *Job) error

Handler processes a leased job.

type Job 

type Job struct {
	ID             string            `json:"id"`
	Queue          string            `json:"queue"`
	QueueVersion   int               `json:"queue_version"`
	Partition      int               `json:"partition"`
	Type           string            `json:"type"`
	Payload        json.RawMessage   `json:"payload"`
	Metadata       map[string]string `json:"metadata,omitempty"`
	IdempotencyKey string            `json:"idempotency_key,omitempty"`
	State          JobState          `json:"state"`
	RunAt          time.Time         `json:"run_at"`
	CreatedAt      time.Time         `json:"created_at"`
	UpdatedAt      time.Time         `json:"updated_at"`
	Timeout        time.Duration     `json:"timeout"`
	Attempts       int               `json:"attempts"`
	MaxRetries     int               `json:"max_retries"`
	LastError      string            `json:"last_error,omitempty"`
	LeaseOwner     string            `json:"lease_owner,omitempty"`
	LeaseExpiresAt time.Time         `json:"lease_expires_at"`
	LastFinishedAt time.Time         `json:"last_finished_at"`
}

Job is the persisted unit of work handled by the scheduler.

type JobState 

type JobState string

JobState describes the lifecycle state of a job. 

const (
	// JobStateScheduled indicates a job is waiting for its run_at time.
	JobStateScheduled JobState = "scheduled"
	// JobStateReady indicates a job is available to be leased by a worker.
	JobStateReady JobState = "ready"
	// JobStateLeased indicates a worker currently owns the job lease.
	JobStateLeased JobState = "leased"
	// JobStateRetryWait indicates a job is waiting for its retry backoff window.
	JobStateRetryWait JobState = "retry_wait"
	// JobStateSucceeded indicates a job completed successfully.
	JobStateSucceeded JobState = "succeeded"
	// JobStateDead indicates a job exhausted retries and will not be retried automatically.
	JobStateDead JobState = "dead"
)

type Lease 

type Lease struct {
	Job            *Job
	LeaseID        string
	LeaseOwner     string
	LeaseExpiresAt time.Time
}

Lease describes temporary ownership of a job by a worker.

type LeaseCompletion 

type LeaseCompletion struct {
	JobID      string
	LeaseID    string
	FinishedAt time.Time
	RetryAt    time.Time
	Failure    string
	Dead       bool
}

LeaseCompletion describes the final outcome for a leased job.

type LeaseRenewal 

type LeaseRenewal struct {
	JobID   string
	LeaseID string
}

LeaseRenewal identifies a leased job that should have its lease extended.

type MemoryStore 

type MemoryStore struct {
	// contains filtered or unexported fields
}

MemoryStore is an in-memory Store implementation for tests and local development.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates an empty in-memory scheduler store.

func (*MemoryStore) Ack 

func (s *MemoryStore) Ack(_ context.Context, jobID, leaseID string, now time.Time) error

Ack marks a single leased job as successful by delegating to CompleteLeases.

func (*MemoryStore) CompleteLeases 

func (s *MemoryStore) CompleteLeases(_ context.Context, completions []LeaseCompletion) error

CompleteLeases applies the final outcomes for multiple leased jobs.

func (*MemoryStore) Enqueue 

func (s *MemoryStore) Enqueue(_ context.Context, job *Job) error

Enqueue stores a new job in memory.

func (*MemoryStore) Fail 

func (s *MemoryStore) Fail(_ context.Context, jobID, leaseID string, now, retryAt time.Time, failure string, dead bool) error

Fail marks a single leased job as failed by delegating to CompleteLeases.

func (*MemoryStore) LeaseJobs 

func (s *MemoryStore) LeaseJobs(_ context.Context, queues []string, workerID string, now time.Time, limit int, leaseDuration time.Duration) ([]*Lease, error)

LeaseJobs leases ready jobs to the given worker.

func (*MemoryStore) PromoteDueJobs 

func (s *MemoryStore) PromoteDueJobs(_ context.Context, queues []string, now time.Time, limit int) (int, error)

PromoteDueJobs moves due jobs from scheduled or retry_wait to ready.

func (*MemoryStore) RecoverExpiredLeases 

func (s *MemoryStore) RecoverExpiredLeases(_ context.Context, queues []string, now time.Time, limit int) (int, error)

RecoverExpiredLeases returns expired leases back to the ready state.

func (*MemoryStore) RenewLease 

func (s *MemoryStore) RenewLease(_ context.Context, jobID, leaseID string, now time.Time, leaseDuration time.Duration) error

RenewLease renews a single lease by delegating to RenewLeases.

func (*MemoryStore) RenewLeases 

func (s *MemoryStore) RenewLeases(_ context.Context, renewals []LeaseRenewal, now time.Time, leaseDuration time.Duration) error

RenewLeases renews multiple active leases in memory.

func (*MemoryStore) Stats 

func (s *MemoryStore) Stats(_ context.Context, queues []string) ([]Stats, error)

Stats returns queue-level job counts grouped by state.

type Metrics 

type Metrics interface {
	// RecordEnqueue reports that a job was created.
	RecordEnqueue(context.Context, *Job)
	// RecordLease reports that a worker leased a job for execution.
	RecordLease(context.Context, *Job)
	// RecordSuccess reports a successful job completion and execution latency.
	RecordSuccess(context.Context, *Job, time.Duration)
	// RecordFailure reports a failed job completion and execution latency.
	RecordFailure(context.Context, *Job, error, time.Duration)
	// RecordRetry reports that a failed job was scheduled for another attempt.
	RecordRetry(context.Context, *Job, time.Duration)
}

Metrics receives worker lifecycle events for observability integrations.

type NopMetrics 

type NopMetrics struct{}

NopMetrics discards all metric events.

func (NopMetrics) RecordEnqueue 

func (NopMetrics) RecordEnqueue(context.Context, *Job)

RecordEnqueue implements Metrics and discards the event.

func (NopMetrics) RecordFailure 

func (NopMetrics) RecordFailure(context.Context, *Job, error, time.Duration)

RecordFailure implements Metrics and discards the event.

func (NopMetrics) RecordLease 

func (NopMetrics) RecordLease(context.Context, *Job)

RecordLease implements Metrics and discards the event.

func (NopMetrics) RecordRetry 

func (NopMetrics) RecordRetry(context.Context, *Job, time.Duration)

RecordRetry implements Metrics and discards the event.

func (NopMetrics) RecordSuccess 

func (NopMetrics) RecordSuccess(context.Context, *Job, time.Duration)

RecordSuccess implements Metrics and discards the event.

type Producer 

type Producer struct {
	// contains filtered or unexported fields
}

Producer creates jobs in the configured scheduler store.

func NewProducer 

func NewProducer(store Store, config ProducerConfig) (*Producer, error)

NewProducer creates a Producer using the provided store and configuration.

func (*Producer) Enqueue 

func (p *Producer) Enqueue(ctx context.Context, opts EnqueueOptions) (*Job, error)

Enqueue creates and persists a new delayed job using the provided options.

type ProducerConfig 

type ProducerConfig struct {
	DefaultQueue       string
	DefaultMaxRetries  int
	DefaultTimeout     time.Duration
	DefaultRetryPolicy RetryPolicy
}

ProducerConfig configures default values applied by a Producer when enqueueing jobs.

func DefaultProducerConfig 

func DefaultProducerConfig() ProducerConfig

DefaultProducerConfig returns a production-friendly default producer configuration.

func (ProducerConfig) Validate 

func (c ProducerConfig) Validate() error

Validate verifies that the producer configuration is internally consistent.

type QueueTopology 

type QueueTopology struct {
	Active   QueueVersionConfig
	Draining []QueueVersionConfig
}

QueueTopology describes the active and draining versions for a logical queue.

func (QueueTopology) DrainingVersionNumbers 

func (q QueueTopology) DrainingVersionNumbers() []int

DrainingVersionNumbers returns the draining queue versions in ascending order.

type QueueVersionConfig 

type QueueVersionConfig struct {
	Version    int
	Partitions int
	Status     string
}

QueueVersionConfig describes a versioned queue layout and its partition count.

type RedisConfig 

type RedisConfig struct {
	Prefix          string
	Partitions      int
	ActiveVersion   int
	DrainingVersion []QueueVersionConfig
}

RedisConfig configures the Redis-backed store.

type RedisStore 

type RedisStore struct {
	// contains filtered or unexported fields
}

RedisStore is a Redis-backed Store implementation with batched coordination operations.

func NewRedisStore 

func NewRedisStore(client *redis.Client, prefix string) *RedisStore

NewRedisStore creates a Redis store with a default single-version topology.

func NewRedisStoreWithConfig 

func NewRedisStoreWithConfig(client *redis.Client, cfg RedisConfig) *RedisStore

NewRedisStoreWithConfig creates a Redis store with explicit topology defaults.

func (*RedisStore) Ack 

func (s *RedisStore) Ack(ctx context.Context, jobID, leaseID string, now time.Time) error

Ack marks a single leased job as successful by delegating to CompleteLeases.

func (*RedisStore) CompleteDrainingVersion 

func (s *RedisStore) CompleteDrainingVersion(ctx context.Context, queue string, version int) (QueueTopology, error)

CompleteDrainingVersion removes a drained queue version from queue topology metadata.

func (*RedisStore) CompleteLeases 

func (s *RedisStore) CompleteLeases(ctx context.Context, completions []LeaseCompletion) error

CompleteLeases applies the final outcomes for multiple leased jobs in Redis.

func (*RedisStore) ConfigureQueue 

func (s *RedisStore) ConfigureQueue(ctx context.Context, queue string, active QueueVersionConfig, draining []QueueVersionConfig) error

ConfigureQueue writes queue topology metadata for a logical queue.

func (*RedisStore) Enqueue 

func (s *RedisStore) Enqueue(ctx context.Context, job *Job) error

Enqueue stores a new job in Redis using the active queue topology.

func (*RedisStore) Fail 

func (s *RedisStore) Fail(ctx context.Context, jobID, leaseID string, now, retryAt time.Time, failure string, dead bool) error

Fail marks a single leased job as failed by delegating to CompleteLeases.

func (*RedisStore) LeaseJobs 

func (s *RedisStore) LeaseJobs(ctx context.Context, queues []string, workerID string, now time.Time, limit int, leaseDuration time.Duration) ([]*Lease, error)

LeaseJobs leases ready jobs from Redis across active and draining queue versions.

func (*RedisStore) PromoteDueJobs 

func (s *RedisStore) PromoteDueJobs(ctx context.Context, queues []string, now time.Time, limit int) (int, error)

PromoteDueJobs moves due jobs from scheduled or retry_wait to ready across queue versions.

func (*RedisStore) PromoteQueueVersion 

func (s *RedisStore) PromoteQueueVersion(ctx context.Context, queue string, nextVersion QueueVersionConfig) (QueueTopology, error)

PromoteQueueVersion activates a new queue version and marks the previous active version as draining.

func (*RedisStore) QueueTopology 

func (s *RedisStore) QueueTopology(ctx context.Context, queue string) (QueueTopology, error)

QueueTopology returns the active and draining topology for a logical queue.

func (*RedisStore) RecoverExpiredLeases 

func (s *RedisStore) RecoverExpiredLeases(ctx context.Context, queues []string, now time.Time, limit int) (int, error)

RecoverExpiredLeases returns expired leases back to the ready state across queue versions.

func (*RedisStore) RenewLease 

func (s *RedisStore) RenewLease(ctx context.Context, jobID, leaseID string, now time.Time, leaseDuration time.Duration) error

RenewLease renews a single lease by delegating to RenewLeases.

func (*RedisStore) RenewLeases 

func (s *RedisStore) RenewLeases(ctx context.Context, renewals []LeaseRenewal, now time.Time, leaseDuration time.Duration) error

RenewLeases renews multiple active leases in Redis.

func (*RedisStore) Stats 

func (s *RedisStore) Stats(ctx context.Context, queues []string) ([]Stats, error)

Stats returns queue-level job counts grouped by state across queue versions.

type RetryPolicy 

type RetryPolicy struct {
	MaxRetries     int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Multiplier     float64
}

RetryPolicy controls retry backoff timing and attempt limits.

func (RetryPolicy) Backoff 

func (p RetryPolicy) Backoff(attempt int) time.Duration

Backoff returns the retry delay for the given attempt count.

func (RetryPolicy) Validate 

func (p RetryPolicy) Validate() error

Validate verifies that the retry policy is internally consistent.

type Stats 

type Stats struct {
	Queue     string
	Scheduled int64
	Ready     int64
	Leased    int64
	RetryWait int64
	Succeeded int64
	Dead      int64
}

Stats contains queue-level job counts grouped by lifecycle state.

type Store 

type Store interface {
	// Enqueue persists a new job record.
	Enqueue(context.Context, *Job) error
	// PromoteDueJobs moves due scheduled or retrying jobs into the ready state.
	PromoteDueJobs(context.Context, []string, time.Time, int) (int, error)
	// LeaseJobs claims up to limit ready jobs for a worker for the given lease duration.
	LeaseJobs(context.Context, []string, string, time.Time, int, time.Duration) ([]*Lease, error)
	// RenewLeases extends one or more active leases owned by a worker.
	RenewLeases(context.Context, []LeaseRenewal, time.Time, time.Duration) error
	// CompleteLeases records the final outcome for one or more leased jobs.
	CompleteLeases(context.Context, []LeaseCompletion) error
	// RenewLease extends a single active lease.
	RenewLease(context.Context, string, string, time.Time, time.Duration) error
	// Ack marks a leased job as successful.
	Ack(context.Context, string, string, time.Time) error
	// Fail marks a leased job as failed and optionally schedules a retry.
	Fail(context.Context, string, string, time.Time, time.Time, string, bool) error
	// RecoverExpiredLeases returns expired leases back to the ready state.
	RecoverExpiredLeases(context.Context, []string, time.Time, int) (int, error)
	// Stats returns queue-level lifecycle counts for the requested queues.
	Stats(context.Context, []string) ([]Stats, error)
}

Store defines the persistence contract used by producers and workers.

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

Worker executes leased jobs from a scheduler store using registered handlers.

func NewWorker 

func NewWorker(store Store, config WorkerConfig, logger *slog.Logger, metrics Metrics) (*Worker, error)

NewWorker creates a Worker using the provided store, configuration, logger, and metrics sink.

func (*Worker) Handle 

func (w *Worker) Handle(jobType string, handler Handler)

Handle registers a handler for a job type.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) error

Start runs the worker loop until the context is canceled.

type WorkerConfig 

type WorkerConfig struct {
	WorkerID          string
	Queues            []string
	Concurrency       int
	BatchSize         int
	QueueBuffer       int
	PollInterval      time.Duration
	LeaseDuration     time.Duration
	HeartbeatInterval time.Duration
	MaxInFlight       int
	RetryPolicy       RetryPolicy
}

WorkerConfig controls worker batching, concurrency, polling, leasing, and retries.

func DefaultWorkerConfig 

func DefaultWorkerConfig(workerID string) WorkerConfig

DefaultWorkerConfig returns a worker configuration with balanced defaults for general use.

func (WorkerConfig) Validate 

func (c WorkerConfig) Validate() error

Validate verifies that the worker configuration is internally consistent.

Source Files

View all Source files

Directories

Path Synopsis
examples
redis_demo command
redis_migration command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.