Repository navigation
feat!: job retries with backoff, task identity, and pluggable metrics - #7
Merged
Merged
Conversation
Task carries an optional ID (unique, for logs) and Kind (bounded, for metric labels). Submit delegates to SubmitTask. Nil jobs are rejected with ErrNilJob instead of panicking inside a worker. BREAKING CHANGE: the Pool interface gains SubmitTask; external implementations of Pool, such as mocks, must add it.
Replace log.Printf with slog. Failure and panic records carry worker, task_id, kind, duration and err (plus the stack for panics) and are logged with the job context so handlers can extract trace IDs. WithName binds a pool attribute to every record.
…time gauges Prometheus metrics were created but never registered, so they could not be scraped. WithPrometheusRegisterer registers them (rolled back on failure, unregistered on Shutdown) and WithName adds a constant pool label so pools can share a registry. Gauges are now GaugeFuncs evaluated at scrape time, which removes the 5s metrics ticker from the dispatcher. Adds queue_capacity and max_workers gauges.
…y workers
Job execution moves to execute.go. Each job finishes with a Status
(success, error, panic) shared by metrics and logs; panics become a
*PanicError carrying the stack.
Prometheus:
- jobs_submitted_total{kind}, jobs_rejected_total{kind,reason}
- jobs_completed_total{kind,status}
- job_duration_seconds{kind}, job_queue_wait_seconds{kind}
- busy_workers gauge, scale_up_events_total
Metrics() gains BusyWorkers, JobsFailed and AvgQueueWait.
BREAKING CHANGE: the Metrics interface gains BusyWorkers, JobsFailed
and AvgQueueWait; external implementations of Metrics must add them.
WithOnTaskDone registers a callback invoked once per executed task with a TaskResult (ID, kind, queue wait, duration, attempts, status, error). Submit returns before execution, so this is the only way for application code to react to outcomes, e.g. dead-letter queues or status updates. The hook runs on the worker goroutine; panics in it are recovered and logged so a faulty hook cannot take a worker down.
… injectable clock
WithRetry(RetryPolicy{...}) enables retries for every job in the pool.
Retries run inline on the worker that executed the job, so there is no
re-queueing and no extra goroutines; when disabled the execution path
is unchanged.
- MaxAttempts is required and finite (no infinite retries)
- Backoff is a stateless interface (Next(retry)); ConstantBackoff is
the first built-in
- Clock is injectable for deterministic tests
- Backoff waits abort as soon as the job context is done, and context
errors are never retried
- Panics are never retried
- The final error is a *RetryError carrying the stop reason, attempt
count and last error, all reachable via errors.Is/As
- New statuses: retries_exhausted, canceled
LinearBackoff waits Base*retry; ExponentialBackoff waits
Base*Multiplier^(retry-1) with a default multiplier of 1.5 (the
cenkalti/backoff convention). Both clamp to the maximum Duration
instead of overflowing.
The default policy backoff becomes ExponentialBackoff{Base: 100ms}.
Built-in backoffs are validated at New (no negative durations,
multiplier >= 1).
MaxDelay clamps any single backoff wait, so exponential growth cannot stall a worker for an unbounded time. Jitter enables full jitter (uniform in [0, min(MaxDelay, backoff))), the AWS-recommended scheme for avoiding thundering herds. Randomness comes from math/rand/v2's lock-free per-thread generator.
MaxElapsedTime caps the total wall-clock time a job may spend across attempts and waits, in addition to MaxAttempts. Retrying stops as soon as the next wait would exceed the budget, rather than sleeping first and then giving up, which frees the worker earlier. The clock is only read when a budget is set. Adds ErrRetryBudgetExceeded and the budget_exceeded status.
Errors are retried by default (opt-out polarity). Wrapping an error with Permanent(err) stops retrying immediately; the wrapper is transparent to Error() and errors.Is/As and works through further wrapping. Adds ErrPermanentFailure and the permanent status.
RetryAfter(err, d) lets a job propagate a downstream system's own backoff hint (e.g. an HTTP Retry-After header) in a transport-agnostic way. The hint replaces the computed backoff for that retry, is not jittered, and is still capped by MaxDelay to protect pool capacity.
ErrorLimits caps how many attempts may fail with a given error (matched with errors.Is, first match wins), e.g. retry rate limiting 5 times but timeouts only twice. MaxAttempts still bounds the job overall. Counters live in a stack buffer for up to 8 limits, so the common case adds no allocations per job.
- Prometheus: jobs_retried_total{kind} and retry_attempts_total{kind};
jobs_completed_total carries the retry stop status
- Metrics(): JobsRetried and RetryAttempts
- Debug-level "retrying task" log per retry with task_id, kind,
attempt, delay and err; skipped entirely when debug is disabled
- Final failure logs and TaskResult include the attempt count
Retry counters are derived from the attempt count when a job finishes,
so the retry loop itself performs no metric updates.
BREAKING CHANGE: the Metrics interface gains JobsRetried and
RetryAttempts; external implementations of Metrics must add them.
Graceful shutdown lets in-flight retries finish (they are bounded by MaxAttempts/MaxElapsedTime). When Shutdown's context expires, a forceStop channel is closed: pending backoff waits return immediately with ErrPoolShutdown (status "shutdown") and idle workers exit without draining the rest of the queue. Previously the non-blocking workerShutdown signal was missed by any worker busy at that moment.
errors.As forces its pointer target onto the heap, costing two small allocations per failed job (permanent/RetryAfter markers in the retry loop, PanicError/RetryError in status classification). - Retry markers are found with a generic, allocation-free walk of the error tree (findMarker) - statusOf uses a type switch: its input always comes straight from runJob or the retrier, and switching on the RetryError reason also avoids walking RetryError's []error Unwrap, which allocated Failed jobs now execute with zero allocations; an exhausted retry allocates only the returned *RetryError.
BenchmarkExecuteJob measures executeJob in isolation for metrics off/on, retry success, one retry, OnTaskDone hook, plain error and exhausted retries. TestExecuteJobAllocations fails if any of these paths starts allocating (skipped under -race, which adds its own allocations). Results (Apple M-series): ~50ns metrics off, ~125ns with Prometheus, 0 allocs on every path except the final *RetryError.
- Forced shutdown relied on non-blocking sends to an unbuffered workerShutdown channel, which any busy worker missed. forceStop (a closed channel every worker selects on) already covers this, so the per-worker signalling loop and channel are removed. - jobWrapper.id is no longer read since logs identify tasks by task ID. Generating it cost a contended atomic increment on every Submit.
A NaN multiplier passed the >= 1 check (NaN comparisons are false) and produced an undefined Duration conversion; +Inf turned every retry after the first into a maximum-length wait. Both are now rejected at New. Also restructures the RetryPolicy doc comment.
examples/observability runs a named pool under bursty synthetic load with retries, Permanent errors, RetryAfter hints and per-error limits, exposing metrics via promhttp on :2112. docker-compose brings up Prometheus and Grafana with a provisioned dashboard covering workers, queue, throughput, failures by status, p95 execution time and queue wait by kind, retries and rejections. Adds a run-observability Makefile target.
docs/RETRY_AND_OBSERVABILITY.md documents every feature with its use case and code samples: task identity, slog logging, Prometheus metrics and queries, in-process metrics, the OnTaskDone hook, each retry option, RetryError reasons, statuses, shutdown behaviour, Clock-based testing and benchmark results. It also calls out that the submit context is the job's execution context. README gains Retries and Observability sections, the new options and example; runnable godoc examples cover WithRetry, Permanent and WithOnTaskDone.
Verified against a running Prometheus + Grafana: the target is scraped, the dashboard is provisioned, and every panel query returns data. - Stat panels fall back to vector(0) so quiet metrics such as rejections show 0 instead of "No data" - Scale cooldown lowered to 1s so the pool visibly reaches max workers during the 30s bursts (with the 5s default it plateaued at ~22/32)
…etries With retries enabled, a job that fails once its context is done finishes with status canceled; without retries the same failure was reported as a generic error. A live dashboard run showed thousands of request-scoped jobs whose deadline expired in the queue counted as errors. The status is now canceled whenever the job failed and its context is done, with or without a retry policy. Errors returned while the job context is still active remain error.
…to show rejections The orders pool never rejects work (it submits with a background context), so the rejection panels stayed empty. A small previews pool (4 workers, queue of 8, no retries) now receives: - render_preview: request-scoped work with a 250ms deadline whose job should die with the request; disconnecting clients produce rejections with reason=canceled, and deadlines expiring in the queue produce completed status=canceled - track_event: best-effort telemetry submitted with a 5ms deadline, producing reason=timeout rejections during bursts. The job ignores its context, since the submit context is also the execution context. Producers stop before the pools shut down. A second named pool also exercises the dashboard's pool selector.
Shutdown closed the job queue while Submit calls could still send to it. Any Submit blocked on a full queue (backpressure) at that moment, or one that passed the shutdown check just before it, panicked with "send on closed channel", crashing the process during shutdown under load. - A closing channel, closed when Shutdown starts, wakes blocked Submits; they return ErrPoolShutdown (rejected, reason=shutdown) - An atomic count of in-flight Submits is incremented before the state check; Shutdown sets the state and waits for the count to reach zero before closing the queue, so no send can follow the close - Jobs accepted before the close are still drained as before Regression tests reproduce the blocked-Submit panic and race 16 submitters against Shutdown, asserting every accepted job runs. Both fail on the previous code. Throughput benchmarks are unchanged within noise.
The guide said retried jobs run with zero heap allocations. That holds for the retry logic and for zero-length waits, which the benchmarks used, but each non-zero backoff wait allocates a timer through time.After: 3 allocations, 248 bytes, measured identical to a bare time.After call. The guide, the allocation test comment and the Clock doc now say so, and explain why a per-worker reusable timer isn't worth the extra interface and plumbing for a wait that lasts milliseconds.
Submit selected between sending to the queue and ctx.Done(). When the context was already done, as with the README's "return immediately if full" pattern (context.WithTimeout(ctx, 0)), both cases were ready and Go picked one at random: about half of submissions were rejected with ErrTimeout even though the queue had space. Submit now tries a non-blocking send first and only waits with backpressure when the queue is full, so the context bounds waiting and never causes a rejection while there is room. The regression test fails on the previous code at the first submit.
…ntext The context passed to Submit both bounded the wait for queue space and was the context the job ran with. A caller could not wait at most N ms for a slot without that deadline also cancelling the job: the README's own "timeout after 5 seconds" example cancelled every job via its deferred cancel, and context.WithoutCancel removes the wait bound too. Task.Context, when set, is the context Run executes with (and that retries and backoff waits observe); the SubmitTask context then only bounds waiting. When nil, behaviour is unchanged. - Tests: job outlives a cancelled submit context and keeps its values; the submit context still bounds the wait (ErrTimeout); retries are cancelled through the execution context - Guide: new "Submit context vs execution context" section - README: backpressure examples rewritten with Task.Context - ExamplePool_SubmitTask
…l jobs Both the HTTP server and batch processor examples bounded Submit with a context they cancelled right after submitting, which was also the job's context: - http_server: the handler's deferred cancel (and the end of the request) cancelled every accepted job; a 200ms job was cancelled after 0.18ms. It now runs on context.WithoutCancel(r.Context()). - batch_processor: 0 of 100,000 tasks completed, all cancelled when their 100ms submit context was released. Jobs now run on the program context; 1,330 completed in the first 8 seconds. - observability: track_event no longer ignores its context to work around the coupling.
The core keeps its always-on atomic counters behind Metrics() and forwards pool events (submitted, rejected, completed, scaled up) to an internal recorder interface. Prometheus becomes one implementation with an Attach/Detach lifecycle for registering and unregistering collectors, labelled from the pool name. Kinds are normalized once in the core. No public API changes; the Prometheus tests compare the same metric output unchanged. The recorder is only created when a registerer is configured. Before, Prometheus vectors were built and updated on every job even when they were never registered, costing about 75ns per job for nothing: jobs without a registry now execute in ~50-60ns instead of ~125ns.
The core package no longer imports Prometheus. Pool events go to a pluggable Recorder, so any metrics backend can be used: - Recorder: JobSubmitted, JobRejected, JobCompleted, ScaledUp - Attacher: optional Attach(PoolInfo, Metrics) / Detach lifecycle for recorders that register collectors or export gauges - PoolInfo, RejectReason (RejectShutdown, RejectTimeout, RejectCanceled) - WithRecorder option promrecorder.New(reg) implements both interfaces with the same metric names and labels as before; the ported tests compare identical output. A Recorder serves one pool at a time (ErrAlreadyAttached) and can be reused once that pool shuts down. Recording a completed job does not allocate. Core tests use an in-memory recorder, and the observability example uses promrecorder. BREAKING CHANGE: WithPrometheusRegisterer and WithMetricsEnabled are removed. Replace WithPrometheusRegisterer(reg) with WithRecorder(promrecorder.New(reg)) and delete WithMetricsEnabled calls: without a recorder no metrics backend runs, and the counters behind Metrics() are always kept.
Resolving a labelled series with WithLabelValues hashes the labels and takes an RWMutex inside client_golang on every event. Each kind's series are now resolved once and cached in a sync.Map, and completion counters are cached per status, so the per-job path is a lock-free map read. Measured (ns per completed job): - 1 core: 67 -> 29 - 10 cores, eight kinds: 157 -> 14 - 10 cores, one kind, counter only: 126 -> 48 - 10 cores, one kind, histograms: 188 -> 250 (worse) The one regression is client_golang's histogram degrading when many cores observe the same series at once; the old per-call lock throttled access to it. It only shows with sub-microsecond jobs on a single kind. Once a kind is seen, its retry counters are exported at zero instead of appearing on the first retry, as Prometheus recommends for known series. The cache is cleared on Detach so a reused recorder reports to the new pool's collectors; a regression test covers this.
otelrecorder.New(meterProvider) reports pool events through the OpenTelemetry metrics API, so they reach any backend with an OpenTelemetry exporter or the Collector. A nil provider uses the global one. Names, units and attributes follow the OpenTelemetry semantic conventions, modelled on the database connection pool metrics: worker.count (by idle/busy state), worker.limit, worker.min, queue.usage, queue.limit, job.submissions, job.rejections, job.duration (by status; its count gives completions), job.wait_time, job.retries and worker.scale_ups, all under the adaptivepool namespace with an adaptivepool.pool.name attribute. Duration histograms use the bucket boundaries the conventions recommend for seconds. Gauges come from one registered callback that is unregistered on shutdown. Measurement options are precomputed per kind, status and reason, so recording does not allocate (about 200ns per completed job, dominated by the SDK). OpenTelemetry is pinned to v1.46.0, the newest release that supports Go 1.25; v1.47.0 would have raised the module's minimum Go version. Minimum version selection also raises testify to v1.12.1 and golang.org/x/sys to v0.47.0, which fixes GO-2026-5024.
The guide's Prometheus section becomes a Metrics section: choosing a recorder, promrecorder (unchanged metric table and queries), otelrecorder (semantic-convention names, units and attributes), writing a custom Recorder and Attacher, and serving Metrics() over HTTP without a backend. The performance table is re-measured: jobs without a recorder no longer pay for unused Prometheus vectors, and each recorder's own cost is listed. README replaces WithPrometheusRegisterer and WithMetricsEnabled with WithRecorder.
…rable The example hard-coded :2112, the conventional Prometheus client port, which other local services often use: in a live run it failed to bind while Prometheus silently scraped an unrelated service on that port. A -metrics-addr flag (default :2112) sets the address, `make run-observability ARGS=...` passes it through, and the example's README explains keeping prometheus.yml's target in sync.
Add a table of submission errors with the rejection reason each one is counted under, list them in the Submit and SubmitTask doc comments, and note that ErrQueueFull is not currently returned.
Shorten wordy doc comments, fix RetryAfter claiming an exact wait while also being capped, and document ErrPoolShutdown's role as a retry stop reason.
TrySubmit enqueues a task only if the queue has space and otherwise returns ErrQueueFull, which was exported but never returned. Rejections are recorded with the new queue_full reason. BREAKING CHANGE: the Pool interface gains TrySubmit; external implementations of Pool must add it.
Store the submit time as int64 nanoseconds since the pool was created, read from the monotonic clock, instead of a 24-byte time.Time. This cuts the queue buffer by 16 bytes per slot (16 KB at the default queue size) with no change to queue wait measurement.
Closed
iyashjayesh
reviewed
Oct 16, 2026
iyashjayesh
left a comment
iyashjayesh
left a comment
Owner
There was a problem hiding this comment.
Thanks for this, nice feature set. Tests and go vet pass. A few issues, mostly around forced shutdown; comments inline.
Shutdown waited for Submit calls to drain by spinning on an atomic counter with runtime.Gosched and no way to give up, which burns a CPU while any Submit is in flight. Submit now holds a read lock while it checks the state and sends, and Shutdown takes the write lock to close the queue. A waiting writer blocks new readers, so new Submits queue behind it instead of starving it, and blocked Submits still return via the closing channel. Parallel Submit throughput is unchanged (benchstat, n=8).
Three gaps let a forced shutdown (Shutdown's context expiring) leave workers running: - select picks at random when forceStop and the queue are both ready, so a worker kept starting queued jobs. It now re-checks forceStop after receiving a job and drops it like the rest of the queue. - A zero-length backoff returned before looking at the stop channel, so retries with no delay ran on at full speed until MaxAttempts. The wait now checks it too. - The recorder was detached when the 1s grace period ended even if workers were still running and reporting to it. It is now detached once the last worker exits. Each regression test fails without its fix.
Attach rewrote the recorder's collectors in place while events from a previous pool could still read them, a data race that could also count the old pool's events against the new pool's series. Both recorders now build their instruments into an immutable state that Attach publishes and Detach clears through an atomic pointer, so an event reads one consistent snapshot and is dropped once the recorder is detached. The pool also stops reporting after it detaches the recorder, so a late Submit on a shut down pool can't reach a recorder that has been attached to another pool. In-process counters keep counting. promrecorder now caches its rejection counters per kind and reason, in the per-kind series, instead of calling WithLabelValues on every rejection, which took a lock exactly when backpressure peaks: 24ns to 16ns on one core and 129ns to 51ns on ten (benchstat, n=8). A kind that was only ever rejected now also exposes its other series at zero.
When Task.Context is nil the submit context is also the job's context, but a done one was still queued if there was room (the fast path), so the job started with a cancelled context and could even report success if it ignored it. Such a call is now rejected up front with ErrTimeout or the context's error, counted under the timeout or canceled reason. With Task.Context set, the submit context only bounds the wait, so the fast path is kept and a done one is still queued when there is room. TrySubmit covers the "don't wait" case that a zero timeout used to.
…ckoff Permanent errors were only recognised inside a *RetryError, so without WithRetry they were reported as plain errors and the status depended on whether retries were configured. statusOf now also looks for the permanent marker, which does not allocate. TaskResult.Err stays the job's own error in that case. ExponentialBackoff with Base 0 returned 0 * +Inf = NaN once the power overflowed (about 1,024 retries), which converts to a huge negative Duration on amd64. A zero base now returns zero directly.
Removing the option broke existing callers at compile time. It never registered metrics anywhere before this change, so a no-op preserves behaviour: in-process metrics are always on, and exporting is opt-in through WithRecorder. The doc comment points callers there.
OnTaskDone panics were recovered, but Recorder calls were not: one that panicked on a worker or the dispatcher goroutine crashed the process, on a caller's goroutine it surfaced from Submit, and in Detach it left Shutdown waiting forever for shutdownDone. Every recorder call now recovers. Only the first panic is logged, with its stack, so a recorder that always panics doesn't flood the log, and the pool keeps running. The cost is about 2.5ns per event when a recorder is set (50.8ns to 53.2ns per job in BenchmarkExecuteJob) and nothing without one.
The core module no longer depends on Prometheus or OpenTelemetry. promrecorder, otelrecorder and examples/observability get their own go.mod and keep their import paths, so users add only the backend they use. The core's go.mod drops from 6 direct and 18 indirect requirements to 2 and 1. Each of these modules uses a replace directive to build against the core in this checkout, because no core release has the Recorder API yet; CONTRIBUTING.md describes the order to tag and drop them. The example is never imported, so it keeps its replace directives. ./... does not cross module boundaries, so the Makefile targets and the CI dependency, test, lint and benchmark steps now run in every module. Job names are unchanged. The Codecov step now passes each module's profile through the files input, the name codecov-action v4 uses. README, the guide and CONTRIBUTING gain the install commands, per-module test instructions and release steps, and CONTRIBUTING's stale test path is fixed.
Shutdown kept its result in a local variable and ran the work inside sync.Once, so only the call that ran it could return the outcome: a later call returned nil even when the first had timed out. sync.Once also holds every other caller until the first finishes, so a later call ignored its own context. The first call is now chosen with a compare-and-swap and stores its result on the pool before closing shutdownDone. A later call returns that result, or ctx's error if its own context ends first. A shutdown that has already finished takes precedence over an expired context, so the answer doesn't depend on a random select.
A submission whose deadline passed returned ErrTimeout, a plain error, so errors.Is(err, context.DeadlineExceeded) was false while a cancelled submission returned the context's own error and matched context.Canceled. Code that handles deadlines generically, such as HTTP middleware mapping them to 504, missed it. ErrTimeout now wraps context.DeadlineExceeded. It is still the same value, so err == ErrTimeout and errors.Is(err, ErrTimeout) keep working; only its message gains ": context deadline exceeded". It also exposes Timeout() to callers that look for it.
Contributor
Author
|
I also fixed these issues
|
Contributor
Author
|
While reviewing the code I thought of two things worth discussing: 1. What should happen to queued jobs on a forced shutdown? Options:
2. The queue-wait metric includes the time a caller blocks for space |
Owner
|
Thanks for the thorough fixes. I went through them and all modules pass On your two questions: 1. Queued jobs on forced shutdown: silently dropping them while 2. Queue-wait metric: keep the current meaning and document that it's measured from the |
A forced shutdown discarded queued jobs silently and Shutdown could return nil. Each dropped job is now reported to OnTaskDone with StatusShutdown, Attempts 0 and ErrPoolShutdown, and to a new Recorder.JobDropped method (jobs_dropped_total / adaptivepool.job.dropped). Shutdown returns a *ShutdownError with the dropped count, wrapping the context's error, and Metrics gains JobsDropped.
Owner
|
I pushed 1086793 to this branch to close out the forced-shutdown question, so you don't need to make that change yourself:
All modules pass |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #2
Adds configurable job retries with backoff, plus the observability needed to
run them in production: task identity, structured logging, a completion hook,
and metrics that work with Prometheus, OpenTelemetry, or any other backend.
What's new
Retries:
WithRetry(RetryPolicy{...})MaxAttempts, with pluggableBackoff:ConstantBackoff,LinearBackoff,ExponentialBackoffMaxDelaycap and fullJitter, as requested in Job Retries with Backoff #2MaxElapsedTime: a total time budget across all attemptsPermanent(err)marks failures that retrying can't fix (bad input, 4xx) so they fail immediatelyRetryAfter(err, d)makes the next retry wait a server-provided delay (e.g. aRetry-Afterheader) instead of the backoff, capped byMaxDelayErrorLimits: per-error attempt limits (for example, give up onerrUpstreamafter 2 attempts)*RetryErrorwith the reason (ErrRetriesExhausted,ErrRetryBudgetExceeded,ErrPermanentFailure)Submission, task identity and completion
SubmitTask(ctx, Task{ID, Kind, Run, Context});Submitstill works and delegates to itTask.Contextseparates the job's execution context from the submit wait,so a short submit deadline no longer cancels accepted jobs
TrySubmit(ctx, task)never waits: it returnsErrQueueFullwhen the queueis full, for best-effort work that should be dropped rather than queued
WithOnTaskDone(func(TaskResult))fires once per task with status, attempts,queue wait, duration and error, for dead-letter queues, DB updates or tracing
Logging:
WithLogger(*slog.Logger), defaulting toslog.Default(), replaceslog.PrintfMetrics: vendor-agnostic
Recorderinterface viaWithRecorderpromrecorder: Prometheus collectors with per-kind duration, queue wait,completion status, retries and rejections, and scrape-time gauges
otelrecorder: OpenTelemetry instruments named per the semantic conventionsgo get github.com/iyashjayesh/go-adaptive-pool/promrecorder),so the core's
go.modneeds only testify and goleakrecorder is recovered and logged once instead of crashing a worker
WithNamelabels each pool, so several pools can share a registryFixes
Submitcould panic with "send on closed channel" when racing withShutdowndeterministic: rejected up front when it is also the job's context, queued when
Task.Contextis set and there is roomcanceledBehaviour changes
Shutdown's context expires, workers stoptaking jobs, backoff waits (including zero-length ones) abort, and the recorder detaches once the
last worker exits. On
main, queued jobs kept running in the background. Dropped jobs currently getno per-job signal; this is open for discussion in a comment below.
Shutdowncalls return the first call's result, or their own context's error if it ends first.ErrTimeoutnow wrapscontext.DeadlineExceeded;err == ErrTimeoutanderrors.Iskeep working.WithMetricsEnabledis deprecated and does nothing; exporting is opt-in viaWithRecorder.log.Printftolog/slog; useslog.New(slog.DiscardHandler)to silence it.Breaking changes
Checked with
apidiffagainstmain:PoolinterfaceSubmitTaskandTrySubmit; breaks custom implementations and mocksMetricsinterfaceBusyWorkers,JobsFailed,JobsRetried,RetryAttempts,AvgQueueWait; breaks custom implementations and mocksConfig==, since it now holds the logger, hook and retry policyPerformance
main(~53ns vs ~57ns, benchstat n=10), and submit throughput is unchanged.sleeps costs one timer.
AllocsPerRuntests guard this.Docs and examples
docs/RETRY_AND_OBSERVABILITY.md: each feature with use cases and codesamples, plus reference tables for submit errors, retry stop reasons and statuses
Example*functions for pkg.go.devexamples/observability: two pools under synthetic load, with a Prometheus andGrafana docker-compose stack and a provisioned dashboard
CONTRIBUTING.mddescribes the module layout and the release orderTesting
gofmt,go vet,golangci-lint(0 issues), andgo test -racepass in every module on every commitprometheus/testutiland the OTelManualReader, and logs via a capturingsloghandlerSubmit/Shutdownrace, forced shutdown, recorder re-attach under-race, and done-context submissionTasks (module split)
v0.1.0promrecorder/go.modandotelrecorder/go.mod:go get github.com/iyashjayesh/go-adaptive-pool@v0.1.0, delete thereplacedirective and its comment, rungo mod tidy(thev0.0.0requirement is a placeholder until then)promrecorder/v0.1.0,otelrecorder/v0.1.0filesinput (it may need a token)examples/observabilitykeeps itsreplacedirectives, since nothing imports it.Notes for reviewers