Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

worker

package module

Go to main page

Versions in this module

v0

v0.2.0

Oct 19, 2026

v0.1.0

Sep 29, 2026

Changes in this version

+ var ErrLimiterStopped = errors.New("worker: rate limiter stopped")

+ var ErrRateLimited = errors.New("worker: rate limit exceeded")

+ func Permanent(err error) error

type Option

+ func WithRetryJitter(fraction float64) Option

+ type PanicError struct

+ Stack []byte

+ Value any

+ func (e *PanicError) Error() string

+ func (e *PanicError) Is(target error) bool

+ func (e *PanicError) Unwrap() error

v0.0.1

Feb 17, 2026

Changes in this version

+ var ErrInvalidConfig = errors.New("worker: invalid configuration")

+ var ErrJobPanic = errors.New("worker: job panicked")

+ var ErrPoolClosed = errors.New("worker: pool is closed")

+ var ErrPoolFull = errors.New("worker: queue is full")

+ type Batch struct

+ func NewBatch[T any](pool *Pool, batchSize int, processor func(context.Context, []T) error) *Batch[T]

+ func (b *Batch[T]) Add(items ...T)

+ func (b *Batch[T]) Clear()

+ func (b *Batch[T]) Len() int

+ func (b *Batch[T]) Process(ctx context.Context) error

+ func (b *Batch[T]) ProcessAll(ctx context.Context) []error

+ type Config struct

+ JobTimeout time.Duration

+ MaxRetries int

+ MaxRetryDelay time.Duration

+ PanicHandler func(job Job, recovered any)

+ QueueSize int

+ RetryBackoff bool

+ RetryDelay time.Duration

+ Workers int

+ type CronSchedule struct

+ func ParseCron(expr string) (*CronSchedule, error)

+ func (s *CronSchedule) Next(after time.Time) time.Time

+ func (s *CronSchedule) String() string

+ type DailySchedule struct

+ func NewDailySchedule(times ...string) (*DailySchedule, error)

+ func (s *DailySchedule) Next(after time.Time) time.Time

+ type Debouncer struct

+ func NewDebouncer(pool *Pool, delay time.Duration) *Debouncer

+ func NewDebouncerLeading(pool *Pool, delay time.Duration) *Debouncer

+ func (d *Debouncer) Cancel()

+ func (d *Debouncer) Flush()

+ func (d *Debouncer) Submit(job Job)

+ func (d *Debouncer) SubmitFunc(fn func(context.Context) error)

+ type ErrorGroup struct

+ func NewErrorGroup(pool *Pool) *ErrorGroup

+ func NewErrorGroupContext(ctx context.Context, pool *Pool) *ErrorGroup

+ func (g *ErrorGroup) Context() context.Context

+ func (g *ErrorGroup) Go(fn func(ctx context.Context) error)

+ func (g *ErrorGroup) Wait() error

+ type FixedTimeSchedule struct

+ Interval time.Duration

+ Start time.Time

+ func (s *FixedTimeSchedule) Next(after time.Time) time.Time

+ type Future struct

+ func SubmitTyped[T any](pool *Pool, fn func(ctx context.Context) (T, error)) *Future[T]

+ func (f *Future[T]) Await(ctx context.Context) (T, error)

+ func (f *Future[T]) Done() bool

+ type Group struct

+ func NewGroup(pool *Pool) *Group

+ func NewGroupContext(ctx context.Context, pool *Pool) *Group

+ func (g *Group) Go(fn func(ctx context.Context) error)

+ func (g *Group) Wait() error

+ func (g *Group) WaitAll() []error

+ type Hooks struct

+ OnJobComplete func(job Job, duration time.Duration)

+ OnJobFailed func(job Job, err error, duration time.Duration)

+ OnJobStart func(job Job)

+ OnWorkerStart func(workerID int)

+ OnWorkerStop func(workerID int)

+ type IntervalSchedule struct

+ Interval time.Duration

+ func (s *IntervalSchedule) Next(after time.Time) time.Time

+ type Job interface

+ Process func(ctx context.Context) error

+ type JobFunc func(ctx context.Context) error

+ func (f JobFunc) Process(ctx context.Context) error

+ type Logger interface

+ Debug func(msg string, args ...any)

+ Error func(msg string, args ...any)

+ Info func(msg string, args ...any)

+ Warn func(msg string, args ...any)

+ type Metrics struct

+ ActiveWorkers atomic.Int32

+ JobsCompleted atomic.Int64

+ JobsFailed atomic.Int64

+ JobsPanicked atomic.Int64

+ JobsRetried atomic.Int64

+ JobsSubmitted atomic.Int64

+ QueueLength atomic.Int32

+ TotalDuration atomic.Int64

+ func (m *Metrics) Snapshot() MetricsSnapshot

+ type MetricsSnapshot struct

+ ActiveWorkers int32

+ Elapsed time.Duration

+ JobsCompleted int64

+ JobsFailed int64

+ JobsPanicked int64

+ JobsRetried int64

+ JobsSubmitted int64

+ QueueLength int32

+ TotalDuration time.Duration

+ func (s MetricsSnapshot) AverageJobDuration() time.Duration

+ func (s MetricsSnapshot) SuccessRate() float64

+ func (s MetricsSnapshot) Throughput() float64

+ type Middleware func(next func(ctx context.Context) error) func(ctx context.Context) error

+ type OnceSchedule struct

+ At time.Time

+ func (s *OnceSchedule) Next(after time.Time) time.Time

+ type Option func(p *Pool, c *Config)

+ func WithHooks(h Hooks) Option

+ func WithJobTimeout(d time.Duration) Option

+ func WithLogger(l Logger) Option

+ func WithMaxRetries(n int) Option

+ func WithMaxRetryDelay(d time.Duration) Option

+ func WithMiddleware(mw ...Middleware) Option

+ func WithPanicHandler(fn func(job Job, recovered any)) Option

+ func WithQueueSize(n int) Option

+ func WithRetryBackoff(enabled bool) Option

+ func WithRetryDelay(d time.Duration) Option

+ func WithWorkers(n int) Option

+ type OverlapPolicy int

+ const OverlapAllow

+ const OverlapQueue

+ const OverlapSkip

+ type Pool struct

+ func NewPool(opts ...Option) (*Pool, error)

+ func (p *Pool) ActiveWorkers() int

+ func (p *Pool) Close() error

+ func (p *Pool) CloseWithTimeout(timeout time.Duration) error

+ func (p *Pool) IsClosed() bool

+ func (p *Pool) IsPaused() bool

+ func (p *Pool) Pause()

+ func (p *Pool) QueueLength() int

+ func (p *Pool) Resume()

+ func (p *Pool) Snapshot() MetricsSnapshot

+ func (p *Pool) Submit(fn func(ctx context.Context) error) error

+ func (p *Pool) SubmitContext(ctx context.Context, job Job) error

+ func (p *Pool) SubmitJob(job Job) error

+ func (p *Pool) SubmitJobWait(ctx context.Context, job Job) error

+ func (p *Pool) SubmitWait(ctx context.Context, fn func(ctx context.Context) error) error

+ func (p *Pool) TrySubmit(fn func(ctx context.Context) error) error

+ func (p *Pool) TrySubmitJob(job Job) error

+ type RateLimiter struct

+ func NewRateLimiter(pool *Pool, n int, interval time.Duration) *RateLimiter

+ func (rl *RateLimiter) Stop()

+ func (rl *RateLimiter) Submit(ctx context.Context, job Job) error

+ func (rl *RateLimiter) TrySubmit(job Job) error

+ type Schedule interface

+ Next func(after time.Time) time.Time

+ type Scheduler struct

+ func NewScheduler(pool *Pool, opts ...SchedulerOption) *Scheduler

+ func (s *Scheduler) At(name string, start time.Time, interval time.Duration, job Job, ...) *Scheduler

+ func (s *Scheduler) Cron(name, expr string, job Job, opts ...TaskOption) error

+ func (s *Scheduler) CronFunc(name, expr string, fn func(context.Context) error, opts ...TaskOption) error

+ func (s *Scheduler) Daily(name string, times []string, job Job, opts ...TaskOption) error

+ func (s *Scheduler) DailyFunc(name string, times []string, fn func(context.Context) error, ...) error

+ func (s *Scheduler) Every(name string, interval time.Duration, job Job, opts ...TaskOption) *Scheduler

+ func (s *Scheduler) EveryFunc(name string, interval time.Duration, fn func(context.Context) error, ...) *Scheduler

+ func (s *Scheduler) IsRunning() bool

+ func (s *Scheduler) Once(name string, at time.Time, job Job, opts ...TaskOption) *Scheduler

+ func (s *Scheduler) OnceFunc(name string, at time.Time, fn func(context.Context) error, opts ...TaskOption) *Scheduler

+ func (s *Scheduler) Pause(name string) bool

+ func (s *Scheduler) Remove(name string)

+ func (s *Scheduler) Resume(name string) bool

+ func (s *Scheduler) Schedule(name string, sched Schedule, job Job, opts ...TaskOption) *Scheduler

+ func (s *Scheduler) ScheduleFunc(name string, sched Schedule, fn func(context.Context) error, ...) *Scheduler

+ func (s *Scheduler) Start()

+ func (s *Scheduler) Stop()

+ func (s *Scheduler) TaskInfo(name string) *TaskInfo

+ func (s *Scheduler) Tasks() []string

+ func (s *Scheduler) Trigger(name string) bool

+ type SchedulerOption func(*Scheduler)

+ func WithOnTaskEnd(fn func(name string, err error, d time.Duration)) SchedulerOption

+ func WithOnTaskStart(fn func(name string)) SchedulerOption

+ func WithSchedulerLogger(l Logger) SchedulerOption

+ func WithTickInterval(d time.Duration) SchedulerOption

+ type TaskInfo struct

+ LastDuration time.Duration

+ LastError error

+ LastRun time.Time

+ MaxRuns int

+ Name string

+ Next time.Time

+ Paused bool

+ RunCount int64

+ Running bool

+ TotalErrors int64

+ TotalRuns int64

+ func (i *TaskInfo) SuccessRate() float64

+ type TaskOption func(*scheduledTask)

+ func WithJitter(d time.Duration) TaskOption

+ func WithMaxRuns(n int) TaskOption

+ func WithOverlapPolicy(p OverlapPolicy) TaskOption

+ func WithRunImmediate() TaskOption

+ type Throttler struct

+ func NewThrottler(pool *Pool, interval time.Duration) *Throttler

+ func (t *Throttler) Force(job Job)

+ func (t *Throttler) Reset()

+ func (t *Throttler) Submit(job Job) bool

+ func (t *Throttler) SubmitTrailing(job Job)

+ type Ticker struct

+ func NewTicker(pool *Pool, interval time.Duration, job Job) *Ticker

+ func NewTickerFunc(pool *Pool, interval time.Duration, fn func(context.Context) error) *Ticker

+ func NewTickerImmediate(pool *Pool, interval time.Duration, job Job) *Ticker

+ func (t *Ticker) Start()

+ func (t *Ticker) Stop()