feat!: job retries with backoff, task identity, and pluggable metrics

Makefile

@@ -1,4 +1,4 @@
-.PHONY: all build test lint bench run-http run-batch help clean
+.PHONY: all build test lint bench run-http run-batch run-observability help clean
 
 # Go parameters
 GOCMD=go
@@ -21,6 +21,7 @@ help:
 	@echo "  bench         Run all benchmarks"
 	@echo "  run-http      Run the HTTP server example"
 	@echo "  run-batch     Run the batch processor example"
+	@echo "  run-observability Run the Prometheus/Grafana observability example"
 	@echo "  clean         Clean build artifacts and coverage files"
 	@echo "  help          Show this help message"
 
@@ -49,6 +50,10 @@ run-batch:
 	@echo "Running batch processor example..."
 	$(GOCMD) run examples/batch_processor/main.go
 
+run-observability:
+	@echo "Running observability example (metrics on :2112/metrics unless ARGS sets -metrics-addr)..."
+	$(GOCMD) run ./examples/observability $(ARGS)
+
 run-1m-with:
 	@echo "Running 1M RPS simulator WITH adaptivepool..."
 	$(GOCMD) run examples/one_million_simulator/with_pool/main.go

README.md

@@ -20,7 +20,8 @@ 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 built-in Prometheus metrics
+- 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.
@@ -34,10 +35,15 @@ The `adaptive` behavior here is worker lifecycle adaptation, not request-level c
 - *Explicit backpressure*
     - When overloaded, submissions block or are rejected. The caller must handle it.
 - *Observability*
-    - Built-in Prometheus metrics:
-        - queue depth
-        - throughput
-        - latency
+    - 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`](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.
 
@@ -107,8 +113,20 @@ pool, err := adaptivepool.New(
     // Cooldown between scaling operations (default: 5s)
     adaptivepool.WithScaleCooldown(3*time.Second),
 
-    // Enable/disable metrics (default: true)
-    adaptivepool.WithMetricsEnabled(true),
+    // 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}),
 )
 ```
 
@@ -118,35 +136,89 @@ The pool enforces backpressure when the queue is full:
 
 ```go
 // Block until capacity is available
-ctx := context.Background()
-err := pool.Submit(ctx, job)
+err := pool.Submit(context.Background(), job)
 
-// Timeout after 5 seconds
-ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+// Wait at most 5 seconds for capacity
+waitCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 defer cancel()
-err := pool.Submit(ctx, job)
-if err == adaptivepool.ErrTimeout {
+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
 }
 
-// Return immediately if full
-ctx, cancel := context.WithTimeout(context.Background(), 0)
-defer cancel()
-err := pool.Submit(ctx, job)
+// 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](docs/RETRY_AND_OBSERVABILITY.md#submit-context-vs-execution-context).
+
+## Retries
+
+```go
+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
+}
 ```
 
-## Metrics
+See the [Retries and Observability guide](docs/RETRY_AND_OBSERVABILITY.md) for every option and its use case.
+
+## Observability
 
-Access pool metrics for observability:
+Identify tasks so logs and metrics can tell them apart:
+
+```go
+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](examples/observability):
+
+```go
+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](docs/RETRY_AND_OBSERVABILITY.md#metrics).
+
+In-process metrics are always available:
 
 ```go
 metrics := pool.Metrics()
 
 fmt.Printf("Queue Depth: %d\n", metrics.QueueDepth())
-fmt.Printf("Active Workers: %d\n", metrics.ActiveWorkers())
-fmt.Printf("Jobs Processed: %d\n", metrics.JobsProcessed())
+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("Avg Latency: %v\n", metrics.AvgJobLatency())
+fmt.Printf("Retries: %d\n", metrics.RetryAttempts())
+fmt.Printf("Avg Latency / Queue Wait: %v / %v\n", metrics.AvgJobLatency(), metrics.AvgQueueWait())
 ```
 
 ## Graceful Shutdown
@@ -194,6 +266,15 @@ cd examples/batch_processor
 go run main.go
 ```
 
+### Observability with Prometheus and Grafana
+
+See [examples/observability](examples/observability) for a pool under bursty load with retries, plus a Prometheus and Grafana stack with a provisioned dashboard:
+
+```bash
+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.
@@ -280,7 +361,7 @@ Producer → Submit(ctx) → Bounded Queue → Dispatcher → Adaptive Workers 
 Key components:
 - **Bounded Queue**: Buffered channel with fixed capacity
 - **Dispatcher**: Routes jobs to workers and monitors scaling
-- **Workers**: Execute jobs with panic recovery and metrics tracking
+- **Workers**: Execute jobs with panic recovery, inline retries and metrics tracking
 - **Scaling Logic**: Adjusts worker count based on queue utilization
 
 ## Comparison with Other Libraries
@@ -289,7 +370,7 @@ Key components:
 |---------|-----------------|------|------|----------|
 | Adaptive Scaling | Yes | No | Yes | No |
 | Explicit Backpressure | Yes | Partial | Partial | No |
-| Prometheus Metrics | Yes | No | No | 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 |
@@ -326,6 +407,7 @@ Inspired by:
 
 ## Related Documentation
 
+- [Retries and Observability](docs/RETRY_AND_OBSERVABILITY.md) - Retry policy, logging, metrics and hooks
 - [Examples](examples/) - Complete working examples
 - [Benchmarks](.) - Performance comparisons
 

backoff_test.go

@@ -0,0 +1,52 @@
+package adaptivepool
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBackoffSequences(t *testing.T) {
+	ms := time.Millisecond
+	tests := []struct {
+		name    string
+		backoff Backoff
+		want    []time.Duration
+	}{
+		{"constant", ConstantBackoff{Delay: 50 * ms}, []time.Duration{50 * ms, 50 * ms, 50 * ms}},
+		{"linear", LinearBackoff{Base: 100 * ms}, []time.Duration{100 * ms, 200 * ms, 300 * ms}},
+		{"exponential default multiplier", ExponentialBackoff{Base: 100 * ms}, []time.Duration{100 * ms, 150 * ms, 225 * ms}},
+		{"exponential x2", ExponentialBackoff{Base: 100 * ms, Multiplier: 2}, []time.Duration{100 * ms, 200 * ms, 400 * ms}},
+		{"pointer receiver", &LinearBackoff{Base: ms}, []time.Duration{ms, 2 * ms, 3 * ms}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for i, want := range tt.want {
+				assert.Equal(t, want, tt.backoff.Next(i+1), "retry %d", i+1)
+			}
+		})
+	}
+}
+
+func TestBackoffOverflowClamps(t *testing.T) {
+	assert.Equal(t, time.Duration(math.MaxInt64), ExponentialBackoff{Base: time.Second, Multiplier: 10}.Next(100))
+	assert.Equal(t, time.Duration(math.MaxInt64), LinearBackoff{Base: time.Hour}.Next(math.MaxInt32))
+}
+
+func TestBackoffValidation(t *testing.T) {
+	invalid := []Backoff{
+		ConstantBackoff{Delay: -1},
+		LinearBackoff{Base: -1},
+		ExponentialBackoff{Base: -1},
+		ExponentialBackoff{Base: time.Second, Multiplier: 0.5},
+		&ExponentialBackoff{Base: time.Second, Multiplier: 0.5},
+		ExponentialBackoff{Base: time.Second, Multiplier: math.NaN()},
+		ExponentialBackoff{Base: time.Second, Multiplier: math.Inf(1)},
+	}
+	for _, b := range invalid {
+		_, err := New(WithRetry(RetryPolicy{MaxAttempts: 2, Backoff: b}))
+		assert.ErrorIs(t, err, ErrInvalidConfig, "%#v", b)
+	}
+}

backpressure_test.go

@@ -225,3 +225,85 @@ func TestConcurrentSubmissions(t *testing.T) {
 
 	assert.Equal(t, int64(expectedJobs), pool.Metrics().JobsProcessed())
 }
+
+// Regression: with an already-expired context, Submit raced the queue send
+// against ctx.Done() and rejected about half of submissions at random even
+// though the queue had space.
+func TestSubmitPrefersQueueSpaceOverDoneContext(t *testing.T) {
+	p := newTestPool(t, WithMinWorkers(1), WithMaxWorkers(1), WithQueueSize(1000), WithLogger(discardLogger()))
+
+	release := make(chan struct{})
+	defer close(release)
+	job := func(context.Context) error { //nolint:unparam
+		<-release
+		return nil
+	}
+
+	for i := range 1000 {
+		ctx, cancel := context.WithTimeout(context.Background(), 0)
+		err := p.Submit(ctx, job)
+		cancel()
+		require.NoError(t, err, "submit %d rejected with queue space available", i)
+	}
+
+	// The worker holds one job; refill its slot so the queue is full
+	require.Eventually(t, func() bool { return p.Metrics().BusyWorkers() == 1 }, time.Second, time.Millisecond)
+	require.NoError(t, p.Submit(context.Background(), job))
+
+	// Once the queue is full, a done context rejects immediately
+	ctx, cancel := context.WithTimeout(context.Background(), 0)
+	defer cancel()
+	assert.ErrorIs(t, p.Submit(ctx, job), ErrTimeout)
+}
+
+func TestTrySubmitQueueFull(t *testing.T) {
+	rec := &fakeRecorder{}
+	p := newTestPool(t, WithMinWorkers(1), WithMaxWorkers(1), WithQueueSize(1), WithLogger(discardLogger()), WithRecorder(rec))
+
+	release := make(chan struct{})
+	defer close(release)
+	block := func(context.Context) error { //nolint:unparam
+		<-release
+		return nil
+	}
+
+	ctx := context.Background()
+	require.NoError(t, p.TrySubmit(ctx, Task{Run: block}))
+	require.Eventually(t, func() bool { return p.Metrics().BusyWorkers() == 1 }, time.Second, time.Millisecond)
+	require.NoError(t, p.TrySubmit(ctx, Task{Run: block}))
+
+	assert.ErrorIs(t, p.TrySubmit(ctx, Task{Kind: "email", Run: block}), ErrQueueFull)
+	assert.Equal(t, int64(1), p.Metrics().JobsRejected())
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	assert.Equal(t, []string{"email/queue_full"}, rec.rejected)
+}
+
+func TestTrySubmitContext(t *testing.T) {
+	p := newTestPool(t, WithLogger(discardLogger()))
+
+	type key struct{}
+	got := make(chan any, 2)
+	run := func(ctx context.Context) error {
+		got <- ctx.Value(key{})
+		return nil
+	}
+
+	ctx := context.WithValue(context.Background(), key{}, "submit")
+	require.NoError(t, p.TrySubmit(ctx, Task{Run: run}))
+	assert.Equal(t, "submit", <-got)
+
+	taskCtx := context.WithValue(context.Background(), key{}, "task")
+	require.NoError(t, p.TrySubmit(ctx, Task{Run: run, Context: taskCtx}))
+	assert.Equal(t, "task", <-got)
+}
+
+func TestTrySubmitErrors(t *testing.T) {
+	p, err := New(WithLogger(discardLogger()))
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	assert.ErrorIs(t, p.TrySubmit(ctx, Task{}), ErrNilJob)
+	require.NoError(t, p.Shutdown(ctx))
+	assert.ErrorIs(t, p.TrySubmit(ctx, Task{Run: func(context.Context) error { return nil }}), ErrPoolShutdown)
+}