adaptivepool

README

go-adaptive-pool

go-adaptive-pool is a bounded worker pool for Go with an adaptive worker lifecycle and explicit backpressure, designed to keep systems stable under bursty load.

The goal is not to maximize throughput at all costs, but to prevent unbounded goroutine growth, avoid OOMs, and force overload to be handled explicitly instead of crashing later.

What this library does

This pool focuses on controlling concurrency and memory usage when job submission can outpace processing.

It provides:

• Bounded concurrency via a fixed-size queue
• Adaptive worker lifecycle. Workers scale up and down based on queue pressure
• Explicit backpressure. When the queue is full, submissions block or fail fast
• Observability via vendor-agnostic metrics (Prometheus and OpenTelemetry recorders included), structured slog logs and an OnTaskDone hook
• Inline retries with pluggable backoff, jitter and time budgets
• Safe shutdown with graceful draining and no goroutine leaks

The adaptive behavior here is worker lifecycle adaptation, not request-level concurrency control.

Key features

• Bounded queue
  • Fixed queue size to prevent unbounded memory growth.
• Adaptive worker lifecycle
  • Workers scale up and down based on queue utilization, within configured limits.
• Explicit backpressure
  • When overloaded, submissions block or are rejected. The caller must handle it.
• Observability
  • Metrics per pool and task kind: workers (active/busy), queue depth and wait, throughput, execution time, failures by reason, retries
  • No metrics dependency in the core: plug in the bundled Prometheus or OpenTelemetry recorder, or your own
  • Structured log/slog logging with task ID, kind, status and attempts
  • OnTaskDone hook to react to each task's outcome (dead-letter queues, tracing)
  • Ready-made Grafana dashboard in examples/observability
• Retries
  • Run inline on the worker (no re-queueing, no extra goroutines), zero allocations on the hot path
  • Constant, linear and exponential backoff, full jitter, per-wait cap and total time budget
  • Permanent errors, RetryAfter hints and per-error attempt limits
• Safe shutdown
  • Graceful draining of queued jobs and clean worker shutdown with no leaks.

Installation

go get github.com/iyashjayesh/go-adaptive-pool

The core has no dependency on Prometheus or OpenTelemetry. The metrics adapters are separate modules; add the one you use:

go get github.com/iyashjayesh/go-adaptive-pool/promrecorder   # Prometheus
go get github.com/iyashjayesh/go-adaptive-pool/otelrecorder   # OpenTelemetry

Quick Start

package main

import (
    "context"
    "log"
    "time"

    "github.com/iyashjayesh/go-adaptive-pool"
)

func main() {
    // creating pool
    pool, err := adaptivepool.New(
        adaptivepool.WithMinWorkers(4),
        adaptivepool.WithMaxWorkers(32),
        adaptivepool.WithQueueSize(1000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer pool.Shutdown(context.Background())

    // Submit job
    job := func(ctx context.Context) error {
        // Your work here
        time.Sleep(100 * time.Millisecond)
        return nil
    }

    if err := pool.Submit(context.Background(), job); err != nil {
        log.Printf("Failed to Submit: %v", err)
    }
}

Configuration Options

pool, err := adaptivepool.New(
    // Minimum workers (default: 1)
    adaptivepool.WithMinWorkers(4),
    
    // Maximum workers (default: runtime.NumCPU())
    adaptivepool.WithMaxWorkers(32),
    
    // Queue capacity (default: 1000)
    adaptivepool.WithQueueSize(5000),
    
    // Queue % to trigger scale-up (default: 0.7)
    adaptivepool.WithScaleUpThreshold(0.6),
    
    // Idle time before scale-down (default: 30s)
    adaptivepool.WithScaleDownIdleDuration(20*time.Second),
    
    // Cooldown between scaling operations (default: 5s)
    adaptivepool.WithScaleCooldown(3*time.Second),
    
    // Name attached to logs and metrics (default: "")
    adaptivepool.WithName("orders"),

    // Export metrics: promrecorder, otelrecorder or your own (default: none)
    adaptivepool.WithRecorder(promrecorder.New(prometheus.DefaultRegisterer)),

    // Structured logger (default: slog.Default())
    adaptivepool.WithLogger(slog.New(slog.NewJSONHandler(os.Stdout, nil))),

    // Called once per finished task (default: none)
    adaptivepool.WithOnTaskDone(func(r adaptivepool.TaskResult) { /* ... */ }),

    // Retry failed jobs (default: disabled)
    adaptivepool.WithRetry(adaptivepool.RetryPolicy{MaxAttempts: 3}),
)

Backpressure Handling

The pool enforces backpressure when the queue is full:

// Block until capacity is available
err := pool.Submit(context.Background(), job)

// Wait at most 5 seconds for capacity
waitCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err := pool.SubmitTask(waitCtx, adaptivepool.Task{
    Context: context.Background(), // the job's own context, unaffected by waitCtx
    Run:     job,
})
if errors.Is(err, adaptivepool.ErrTimeout) {
    // Handle overload
}

// Don't wait at all if the queue is full
err := pool.TrySubmit(context.Background(), adaptivepool.Task{Run: job})
if errors.Is(err, adaptivepool.ErrQueueFull) {
    // Drop or handle overload
}

The context passed to Submit is also the context the job runs with, so a wait deadline on it would also cancel the job. Task.Context separates the two; see Submit context vs execution context.

Retries

pool, _ := adaptivepool.New(
    adaptivepool.WithRetry(adaptivepool.RetryPolicy{
        MaxAttempts:    5,
        Backoff:        adaptivepool.ExponentialBackoff{Base: 100 * time.Millisecond},
        MaxDelay:       2 * time.Second,
        MaxElapsedTime: 10 * time.Second,
        Jitter:         true,
    }),
)

job := func(ctx context.Context) error {
    if err := validate(req); err != nil {
        return adaptivepool.Permanent(err) // never retried
    }
    return callAPI(ctx) // retried with backoff
}

See the Retries and Observability guide for every option and its use case.

Observability

Identify tasks so logs and metrics can tell them apart:

pool.SubmitTask(ctx, adaptivepool.Task{
    ID:   "order-8812",  // unique: logs and OnTaskDone only
    Kind: "send_email",  // bounded: metric label
    Run:  sendEmail,
})

Export metrics through a recorder. With Prometheus, plot them with the bundled Grafana dashboard:

reg := prometheus.NewRegistry()
pool, _ := adaptivepool.New(
    adaptivepool.WithName("orders"),
    adaptivepool.WithRecorder(promrecorder.New(reg)),
)
http.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))

With OpenTelemetry, use otelrecorder.New(meterProvider) instead; any other backend needs a four-method Recorder. See Metrics.

In-process metrics are always available:

metrics := pool.Metrics()

fmt.Printf("Queue Depth: %d\n", metrics.QueueDepth())
fmt.Printf("Active / Busy Workers: %d / %d\n", metrics.ActiveWorkers(), metrics.BusyWorkers())
fmt.Printf("Jobs Processed / Failed: %d / %d\n", metrics.JobsProcessed(), metrics.JobsFailed())
fmt.Printf("Jobs Rejected: %d\n", metrics.JobsRejected())
fmt.Printf("Retries: %d\n", metrics.RetryAttempts())
fmt.Printf("Avg Latency / Queue Wait: %v / %v\n", metrics.AvgJobLatency(), metrics.AvgQueueWait())

Graceful Shutdown

// Shutdown with timeout
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

if err := pool.Shutdown(ctx); err != nil {
    log.Printf("Shutdown error: %v", err)
}

Shutdown behavior:

1. Stops accepting new jobs (returns ErrPoolShutdown)
2. Drains in-flight jobs within timeout
3. Terminates all workers deterministically
4. If the timeout expires, drops queued jobs (each reported to OnTaskDone with status shutdown) and returns a *ShutdownError with the count, which wraps the context error

Examples

HTTP Server with Backpressure

See examples/http_server for a complete HTTP server that:

• Processes background jobs via the pool
• Returns 503 when overloaded
• Exposes metrics endpoint
• Handles graceful shutdown

cd examples/http_server
go run main.go

Batch Processing

See examples/batch_processor for processing large batches with:

• Adaptive worker scaling
• Real-time progress tracking
• Throughput metrics

cd examples/batch_processor
go run main.go

Observability with Prometheus and Grafana

See examples/observability for a pool under bursty load with retries, plus a Prometheus and Grafana stack with a provisioned dashboard:

make run-observability
cd examples/observability && docker compose up   # Grafana on http://localhost:3000

Scope and non-goals

This library is intentionally narrow in scope.

It does NOT:

• Perform latency- or error-driven adaptive concurrency control
• Implement AIMD-style or feedback-loop–based limiters
• Replace adaptive limiters like failsafe-go or go-adaptive-limiter

It DOES:

• Enforce hard limits on concurrency and memory usage
• Prevent goroutine explosions under burst load
• Apply backpressure when the system is overloaded
• Make overload visible and explicit instead of failing implicitly

If you already have a well-tuned adaptive limiter controlling request concurrency, a fixed-size worker pool may be sufficient.

Why this exists

In many real systems, goroutines are cheap individually but unbounded submission is not.

Under traffic spikes, naive patterns often lead to:

• Runaway goroutine creation
• Unbounded queues
• Memory pressure and OOMs
• Failure modes that only appear under load

This pool enforces limits and predictable behavior by design.

1 Million RPS Stress Test

We performed an extreme pressure test (1M RPS target for 30s with 500KB tasks) to compare the adaptive pool against naive goroutine spawning.

Metric	Naive (No Pool)	Adaptive Pool
Peak RAM Usage	50.86 GB	1.41 GB
Average Latency	2,063 ms	161 ms
Peak Goroutines	104,473	5,002
Reliability	Fails Under Load	Rock Solid

Why the Pool Wins: Under extreme load, the Naive approach causes a "Goroutine Explosion" and "Memory Bomb" that forces the Go runtime into constant Garbage Collection, leading to unusable 2-second latencies. The go-adaptive-pool shields your system by enforcing backpressure and resource caps.

For a detailed deep-dive into this test and the mechanics of the pool, check out the blog post: Scaling to 1 Million RPS.

Run the comparison yourself:

make run-comparison

Benchmarks

Run standard micro-benchmarks:

go test -bench=. -benchmem -benchtime=10s

Sample micro-benchmark results:

BenchmarkPoolThroughput/workers=8-10     500000    2341 ns/op    128 B/op    3 allocs/op
BenchmarkAdaptivePool-10                 300000    3892 ns/op    256 B/op    5 allocs/op
BenchmarkNaiveGoroutines-10              200000    8234 ns/op    512 B/op   12 allocs/op

Design Principles

• Bounded > Unbounded: Fixed limits prevent cascading failures
• Explicit > Implicit: Backpressure forces correct overload handling
• Simple APIs > Clever Internals: Easy to use, hard to misuse
• Correct Shutdown > Fast Shutdown: No goroutine leaks, ever
• Metrics are Mandatory: Observability is not optional

Architecture

Producer → Submit(ctx) → Bounded Queue → Dispatcher → Adaptive Workers → Job Execution
                              ↓
                         Backpressure

Key components:

• Bounded Queue: Buffered channel with fixed capacity
• Dispatcher: Routes jobs to workers and monitors scaling
• Workers: Execute jobs with panic recovery, inline retries and metrics tracking
• Scaling Logic: Adjusts worker count based on queue utilization

Comparison with Other Libraries

Feature	go-adaptive-pool	ants	pond	errgroup
Adaptive Scaling	Yes	No	Yes	No
Explicit Backpressure	Yes	Partial	Partial	No
Prometheus / OpenTelemetry Metrics	Yes	No	No	No
Graceful Shutdown	Yes	Yes	Yes	Yes
Context Support	Yes	Partial	Yes	Yes
Zero Global State	Yes	No	Yes	Yes

Testing

The repository has four Go modules (the core, promrecorder, otelrecorder and examples/observability), and ./... does not cross module boundaries. make test runs every module with the race detector:

make test

To test one module, run go test inside its directory:

go test -race -v ./...
cd promrecorder && go test -race -coverprofile=coverage.out ./...
go tool cover -html=coverage.out

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.

Acknowledgments

Inspired by:

• ants - High-performance goroutine pool
• pond - Minimalistic worker pool
• asynq - Distributed task queue

Related Documentation

• Retries and Observability - Retry policy, logging, metrics and hooks
• Examples - Complete working examples
• Benchmarks - Performance comparisons

Star History

Documentation

Index

• Variables
• func Permanent(err error) error
• func RetryAfter(err error, d time.Duration) error
• type Attacher
• type Backoff
• type Clock
• type Config
• type ConstantBackoff
  • func (b ConstantBackoff) Next(int) time.Duration
• type ErrorLimit
• type ExponentialBackoff
  • func (b ExponentialBackoff) Next(retry int) time.Duration
• type Job
• type LinearBackoff
  • func (b LinearBackoff) Next(retry int) time.Duration
• type Metrics
• type Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMaxWorkers(n int) Option
  • func WithMetricsEnabled(bool) Option
  • func WithMinWorkers(n int) Option
  • func WithName(name string) Option
  • func WithOnTaskDone(fn func(TaskResult)) Option
  • func WithQueueSize(n int) Option
  • func WithRecorder(r Recorder) Option
  • func WithRetry(policy RetryPolicy) Option
  • func WithScaleCooldown(d time.Duration) Option
  • func WithScaleDownIdleDuration(d time.Duration) Option
  • func WithScaleUpThreshold(pct float64) Option
• type PanicError
  • func (e *PanicError) Error() string
• type Pool
  • func New(options ...Option) (Pool, error)
• type PoolInfo
• type Recorder
• type RejectReason
• type RetryError
  • func (e *RetryError) Error() string
  • func (e *RetryError) Unwrap() []error
• type RetryPolicy
• type ShutdownError
  • func (e *ShutdownError) Error() string
  • func (e *ShutdownError) Unwrap() error
• type Status
• type Task
• type TaskResult

Examples

• Permanent
• WithOnTaskDone
• WithRetry

Variables

var (
	// ErrPoolShutdown is returned by Submit once Shutdown has started, and is the
	// RetryError reason when a forced shutdown aborts a backoff wait
	ErrPoolShutdown = errors.New("pool is shutdown")

	// ErrQueueFull is returned by TrySubmit when the queue has no space
	ErrQueueFull = errors.New("queue is full")

	// ErrTimeout is returned when a job submission times out. It wraps
	// context.DeadlineExceeded, so errors.Is matches both.
	ErrTimeout = fmt.Errorf("submission timeout: %w", context.DeadlineExceeded)

	// ErrNilJob is returned when submitting a nil job
	ErrNilJob = errors.New("job is nil")

	// ErrRetriesExhausted is the RetryError reason when a job used all its attempts
	ErrRetriesExhausted = errors.New("retries exhausted")

	// ErrRetryBudgetExceeded is the RetryError reason when the next retry would exceed MaxElapsedTime
	ErrRetryBudgetExceeded = errors.New("retry time budget exceeded")

	// ErrPermanentFailure is the RetryError reason when a job returned an error wrapped with Permanent
	ErrPermanentFailure = errors.New("permanent failure")

	// ErrInvalidConfig is returned when pool configuration is invalid
	ErrInvalidConfig = errors.New("invalid pool configuration")
)

Functions

func Permanent

func Permanent(err error) error

Permanent marks err as not retryable: the retry engine stops immediately. Use it for failures that cannot succeed on retry, such as invalid input. Permanent(nil) returns nil.

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"

	adaptivepool "github.com/iyashjayesh/go-adaptive-pool"
)

func main() {
	done := make(chan adaptivepool.TaskResult, 1)
	pool, err := adaptivepool.New(
		adaptivepool.WithLogger(slog.New(slog.DiscardHandler)),
		adaptivepool.WithRetry(adaptivepool.RetryPolicy{MaxAttempts: 5}),
		adaptivepool.WithOnTaskDone(func(r adaptivepool.TaskResult) { done <- r }),
	)
	if err != nil {
		panic(err)
	}
	defer func() { _ = pool.Shutdown(context.Background()) }()

	errInvalid := errors.New("invalid payload")
	_ = pool.Submit(context.Background(), func(context.Context) error {
		return adaptivepool.Permanent(errInvalid)
	})

	r := <-done
	fmt.Println(r.Status, r.Attempts)
	fmt.Println(errors.Is(r.Err, adaptivepool.ErrPermanentFailure), errors.Is(r.Err, errInvalid))
}

Output:
permanent 1
true true

func RetryAfter

func RetryAfter(err error, d time.Duration) error

RetryAfter makes the next retry wait d instead of the backoff delay, e.g. for a Retry-After header. The wait is capped by MaxDelay and not jittered. RetryAfter(nil, d) returns nil.

Types

type Attacher

type Attacher interface {
	Attach(info PoolInfo, stats Metrics) error
	Detach()
}

Attacher is implemented by recorders that export gauges or need registration. New calls Attach before starting the pool, and an error fails New. Shutdown calls Detach once every worker has exited, after which the pool reports no more events. stats can be read at any time to report current values such as queue depth.

type Backoff

type Backoff interface {
	Next(retry int) time.Duration
}

Backoff computes the delay before a retry. retry starts at 1 for the first retry. Implementations must be safe for concurrent use; the built-in ones are stateless values.

type Clock

type Clock interface {
	Now() time.Time
	After(d time.Duration) <-chan time.Time
}

Clock abstracts time so retry timing can be tested deterministically.

type Config

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

Config holds the configuration for the pool

type ConstantBackoff

type ConstantBackoff struct {
	Delay time.Duration
}

ConstantBackoff waits the same Delay before every retry.

func (ConstantBackoff) Next

func (b ConstantBackoff) Next(int) time.Duration

Next implements Backoff

type ErrorLimit

type ErrorLimit struct {
	Err         error
	MaxAttempts int
}

ErrorLimit limits how many attempts may fail with errors matching Err.

type ExponentialBackoff

type ExponentialBackoff struct {
	Base       time.Duration
	Multiplier float64
}

ExponentialBackoff waits Base * Multiplier^(retry-1). A zero Multiplier defaults to 1.5.

func (ExponentialBackoff) Next

func (b ExponentialBackoff) Next(retry int) time.Duration

Next implements Backoff

type Job

type Job func(ctx context.Context) error

Job represents a unit of work to be executed by the pool

type LinearBackoff

type LinearBackoff struct {
	Base time.Duration
}

LinearBackoff waits Base * retry: Base, 2*Base, 3*Base, ...

func (LinearBackoff) Next

func (b LinearBackoff) Next(retry int) time.Duration

Next implements Backoff

type Metrics

type Metrics interface {
	// QueueDepth returns the current number of jobs in the queue
	QueueDepth() int

	// ActiveWorkers returns the current number of active workers
	ActiveWorkers() int

	// BusyWorkers returns the number of workers currently executing a job
	BusyWorkers() int

	// JobsProcessed returns the total number of jobs processed
	JobsProcessed() int64

	// JobsFailed returns the number of processed jobs that ended with an error or panic
	JobsFailed() int64

	// JobsRejected returns the total number of jobs rejected
	JobsRejected() int64

	// JobsDropped returns the number of queued jobs a forced shutdown discarded without running
	JobsDropped() int64

	// JobsRetried returns the number of processed jobs that needed at least one retry
	JobsRetried() int64

	// RetryAttempts returns the total number of retries across all jobs
	RetryAttempts() int64

	// AvgJobLatency returns the average job execution time
	AvgJobLatency() time.Duration

	// AvgQueueWait returns the average time jobs spent queued before execution
	AvgQueueWait() time.Duration
}

Metrics provides observability into pool behavior

type Option

type Option func(*Config)

Option is a functional option for configuring the pool

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the structured logger used by the pool. Default: slog.Default()

func WithMaxWorkers

func WithMaxWorkers(n int) Option

WithMaxWorkers sets the maximum number of workers Default: runtime.NumCPU()

func WithMetricsEnabled

func WithMetricsEnabled(bool) Option

WithMetricsEnabled has no effect.

Deprecated: the pool always keeps in-process metrics, available through Metrics(). To export them, use WithRecorder, e.g. WithRecorder(promrecorder.New(reg)).

func WithMinWorkers

func WithMinWorkers(n int) Option

WithMinWorkers sets the minimum number of workers Default: 1

func WithName

func WithName(name string) Option

WithName names the pool. The name is attached to every log record and passed to recorders in PoolInfo, so multiple pools in one process can be told apart. Default: "" (unnamed)

func WithOnTaskDone

func WithOnTaskDone(fn func(TaskResult)) Option

WithOnTaskDone sets a hook called exactly once per executed task, whatever the outcome, and once per queued task dropped by a forced shutdown. It lets application code react to results (dead-letter queues, status updates, tracing) since Submit returns before the task runs.

The hook runs synchronously on the worker goroutine, so keep it fast and hand off heavy work. A panic in the hook is recovered and logged.

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"

	adaptivepool "github.com/iyashjayesh/go-adaptive-pool"
)

func main() {
	done := make(chan adaptivepool.TaskResult, 1)
	pool, err := adaptivepool.New(
		adaptivepool.WithLogger(slog.New(slog.DiscardHandler)),
		adaptivepool.WithOnTaskDone(func(r adaptivepool.TaskResult) { done <- r }),
	)
	if err != nil {
		panic(err)
	}
	defer func() { _ = pool.Shutdown(context.Background()) }()

	_ = pool.SubmitTask(context.Background(), adaptivepool.Task{
		ID:   "order-8812",
		Kind: "send_email",
		Run:  func(context.Context) error { return errors.New("smtp: connection refused") },
	})

	r := <-done
	fmt.Printf("%s %s %s: %v\n", r.Kind, r.ID, r.Status, r.Err)
}

Output:
send_email order-8812 error: smtp: connection refused

func WithQueueSize

func WithQueueSize(n int) Option

WithQueueSize sets the bounded queue capacity Default: 1000

func WithRecorder

func WithRecorder(r Recorder) Option

WithRecorder sends pool events to r, e.g. a Prometheus or OpenTelemetry recorder. The pool's own counters behind Metrics() are always kept. Default: none

func WithRetry

func WithRetry(policy RetryPolicy) Option

WithRetry enables retries for every job submitted to the pool. Default: disabled

Example

package main

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

	adaptivepool "github.com/iyashjayesh/go-adaptive-pool"
)

func main() {
	done := make(chan adaptivepool.TaskResult, 1)
	pool, err := adaptivepool.New(
		adaptivepool.WithLogger(slog.New(slog.DiscardHandler)),
		adaptivepool.WithRetry(adaptivepool.RetryPolicy{
			MaxAttempts: 5,
			Backoff:     adaptivepool.ExponentialBackoff{Base: time.Millisecond},
			MaxDelay:    10 * time.Millisecond,
			Jitter:      true,
		}),
		adaptivepool.WithOnTaskDone(func(r adaptivepool.TaskResult) { done <- r }),
	)
	if err != nil {
		panic(err)
	}
	defer func() { _ = pool.Shutdown(context.Background()) }()

	calls := 0
	_ = pool.Submit(context.Background(), func(context.Context) error {
		calls++
		if calls < 3 {
			return errors.New("temporarily unavailable")
		}
		return nil
	})

	r := <-done
	fmt.Println(r.Status, "after", r.Attempts, "attempts")
}

Output:
success after 3 attempts

func WithScaleCooldown

func WithScaleCooldown(d time.Duration) Option

WithScaleCooldown sets the minimum duration between scaling operations Default: 5s

func WithScaleDownIdleDuration

func WithScaleDownIdleDuration(d time.Duration) Option

WithScaleDownIdleDuration sets the idle duration before scaling down workers Default: 30s

func WithScaleUpThreshold

func WithScaleUpThreshold(pct float64) Option

WithScaleUpThreshold sets the queue utilization percentage (0.0 to 1.0) that triggers scale-up Default: 0.7 (70%)

type PanicError

type PanicError struct {
	Value any
	Stack []byte
}

PanicError is the error recorded when a job panics. Stack holds the goroutine stack captured at the time of the panic.

func (*PanicError) Error

func (e *PanicError) Error() string

type Pool

type Pool interface {
	// Submit submits a job to the pool for execution. If the queue is full
	// it blocks until there is space, ctx is done or the pool shuts down.
	// ctx is also the context the job runs with; use SubmitTask with
	// Task.Context to give the job a different one.
	//
	// It returns ErrNilJob for a nil job, ErrPoolShutdown once Shutdown has
	// started, ErrTimeout if ctx's deadline passes while waiting, and
	// context.Canceled if ctx is cancelled while waiting. A ctx that is
	// already done is rejected the same way, without queueing, unless
	// Task.Context is set.
	Submit(ctx context.Context, job Job) error

	// SubmitTask is like Submit but carries a task ID and kind for logs and
	// metrics, and an optional execution context separate from ctx. It
	// returns the same errors as Submit.
	SubmitTask(ctx context.Context, task Task) error

	// TrySubmit is like SubmitTask but never waits: if the queue is full it
	// returns ErrQueueFull. ctx is only the job's context, used when
	// task.Context is nil.
	TrySubmit(ctx context.Context, task Task) error

	// Shutdown gracefully shuts down the pool
	// Waits for in-flight jobs to complete within the context timeout
	// If ctx ends first, jobs still queued are dropped without running, each
	// reported with StatusShutdown, and it returns a *ShutdownError.
	// It may be called more than once: later calls return the first call's
	// result, or ctx's error if ctx ends before the shutdown does
	Shutdown(ctx context.Context) error

	// Metrics returns the pool metrics
	Metrics() Metrics
}

Pool represents a worker pool that can execute jobs concurrently

func New

func New(options ...Option) (Pool, error)

New Creates a new adaptive worker pool

type PoolInfo

type PoolInfo struct {
	Name       string
	MinWorkers int
	MaxWorkers int
	QueueSize  int
}

PoolInfo describes the pool a recorder is attached to

type Recorder

type Recorder interface {
	// JobSubmitted is called when a job is accepted into the queue
	JobSubmitted(kind string)
	// JobRejected is called when a submission is rejected
	JobRejected(kind string, reason RejectReason)
	// JobCompleted is called once per executed job with its final status,
	// number of attempts, time spent queued and execution time
	JobCompleted(kind string, status Status, attempts int, queueWait, duration time.Duration)
	// JobDropped is called once per queued job that a forced shutdown
	// discarded without running it
	JobDropped(kind string)
	// ScaledUp is called when the pool adds workers
	ScaledUp(added int)
}

Recorder receives pool events, decoupling the pool from any metrics backend. See the promrecorder and otelrecorder packages for Prometheus and OpenTelemetry implementations.

Kinds are normalized before being passed in: an empty kind is reported as "default". Methods run on submitting and worker goroutines, so implementations must be safe for concurrent use and fast. A panic in a method is recovered and logged once, so a faulty recorder can't crash the pool.

type RejectReason

type RejectReason string

RejectReason explains why a submission was rejected

const (
	// RejectShutdown means the pool was shutting down
	RejectShutdown RejectReason = "shutdown"
	// RejectTimeout means the submit context's deadline passed while waiting for queue space
	RejectTimeout RejectReason = "timeout"
	// RejectCanceled means the submit context was canceled while waiting for queue space
	RejectCanceled RejectReason = "canceled"
	// RejectQueueFull means TrySubmit found the queue full
	RejectQueueFull RejectReason = "queue_full"
)

type RetryError

type RetryError struct {
	// Reason explains why retrying stopped
	Reason error
	// Attempts is the number of times the job ran
	Attempts int
	// Err is the error returned by the last attempt
	Err error
}

RetryError is the final error of a job that was retried and gave up. errors.Is matches both the stop Reason (e.g. ErrRetriesExhausted or context.Canceled) and the job's last error.

func (*RetryError) Error

func (e *RetryError) Error() string

func (*RetryError) Unwrap

func (e *RetryError) Unwrap() []error

Unwrap returns the stop reason and the last attempt's error

type RetryPolicy

type RetryPolicy struct {
	// MaxAttempts is the maximum number of times a job runs, including the
	// first attempt. Required and must be >= 1.
	MaxAttempts int

	// Backoff computes the wait before each retry.
	// Default: ExponentialBackoff{Base: 100ms}
	Backoff Backoff

	// MaxDelay caps any single wait, including RetryAfter hints, so a job
	// cannot hold a worker for an unbounded time. Zero means no cap.
	MaxDelay time.Duration

	// Jitter draws each wait uniformly from [0, delay), after MaxDelay, to
	// spread out retries of jobs that failed together.
	Jitter bool

	// MaxElapsedTime bounds a job's total time across attempts and waits.
	// Retrying stops if the next wait would exceed it; a running attempt is
	// never interrupted. Zero means no budget.
	MaxElapsedTime time.Duration

	// ErrorLimits caps attempts for specific errors. Failures are matched
	// with errors.Is against each limit in order; the first match counts.
	// MaxAttempts still applies to the job as a whole.
	ErrorLimits []ErrorLimit

	// Clock is the time source used for backoff waits. Override it in tests.
	// Default: the system clock
	Clock Clock
}

RetryPolicy configures how failed jobs are retried. Retries run inline on the worker; errors are retried unless wrapped with Permanent or the job's context is done, and panics are never retried. A job that stops retrying fails with a *RetryError.

A graceful Shutdown lets retries finish; a forced one aborts backoff waits.

type ShutdownError

type ShutdownError struct {
	// Dropped is the number of queued jobs that were discarded without running
	Dropped int
	// Err is the error of Shutdown's context
	Err error
}

ShutdownError is returned by Shutdown when its context ended before the queue drained. errors.Is matches the context's error.

func (*ShutdownError) Error

func (e *ShutdownError) Error() string

func (*ShutdownError) Unwrap

func (e *ShutdownError) Unwrap() error

Unwrap returns the context's error

type Status

type Status string

Status describes how a task finished. The same values are used for the "status" metric label and log attribute.

const (
	// StatusSuccess means the job returned nil
	StatusSuccess Status = "success"
	// StatusError means the job returned an error while its context was still active
	StatusError Status = "error"
	// StatusPanic means the job panicked; the error is a *PanicError
	StatusPanic Status = "panic"
	// StatusRetriesExhausted means the job failed on every allowed attempt
	StatusRetriesExhausted Status = "retries_exhausted"
	// StatusBudgetExceeded means retrying stopped because of MaxElapsedTime
	StatusBudgetExceeded Status = "budget_exceeded"
	// StatusPermanent means the job returned an error wrapped with Permanent,
	// with or without retries
	StatusPermanent Status = "permanent"
	// StatusShutdown means a forced pool shutdown aborted retrying, or dropped
	// the task from the queue before it ran (Attempts is 0)
	StatusShutdown Status = "shutdown"
	// StatusCanceled means the job failed after its context was done
	StatusCanceled Status = "canceled"
)

type Task

type Task struct {
	// ID identifies a single task (e.g. an order or request ID). It appears in
	// logs and TaskResult only, never in metric labels, so it may be unique.
	ID string

	// Kind groups tasks of the same type (e.g. "send_email") and is used as a
	// metric label. Keep the set of kinds small and bounded.
	Kind string

	// Run is the work to execute.
	Run Job

	// Context is the context Run executes with. When nil, the ctx passed to
	// SubmitTask is used for both waiting for queue space and execution. Set it
	// so a submit deadline doesn't cancel the job; see ExamplePool_SubmitTask.
	Context context.Context
}

Task is a Job annotated with identity used for observability.

type TaskResult

type TaskResult struct {
	ID   string
	Kind string

	// QueueWait is the time between submission and the start of execution
	QueueWait time.Duration
	// Duration is the total execution time, including any retries
	Duration time.Duration
	// Attempts is the number of times the job was run
	Attempts int

	Status Status
	// Err is the final error, nil on success
	Err error
}

TaskResult describes how a task finished. It is passed to the OnTaskDone hook.