worker

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Oct 19, 2026 License: MIT Imports: 13 Imported by: 0

README

go-worker

go-worker

A high-performance, zero-dependency worker pool for Go with retries, panic recovery, graceful shutdown, typed futures, error groups, batching, and a cron scheduler.

Go Reference Latest release Go version CI Coverage

Documentation · API Reference · Changelog

Why go-worker?

A go statement and a sync.WaitGroup get you concurrency. Running background work safely in production — bounded, retried, cancellable, observable, and shut down cleanly — is everything past that:

Capability go-worker Goroutines + WaitGroup
Bounded concurrency and backpressure ✓ You build it
Retries with backoff, jitter, and permanent errors ✓ You build it
Panic recovery with stack traces ✓ You build it
Per-attempt timeouts and cancellation ✓ You build it
Graceful drain and forced shutdown ✓ You build it
Cron, interval, and one-shot scheduling ✓ You build it
Debounce, throttle, and rate limiting ✓ You build it
Metrics and lifecycle hooks ✓ You build it

All of it in the standard library only — nothing to audit, nothing to update.

Performance

Submit            0.9 µs/op    5 allocs   (sustained, full queue)
SubmitWait        1.5 µs/op    8 allocs   (submit, run, and hand back the result)
Throughput       ~1.1M jobs/s             (no-op jobs, 8 workers)

An 8-worker pool with 12 goroutines submitting no-op jobs — so these are the pool's own overhead, not your jobs'. Reproduce with go test -bench=. -benchmem ./...; full numbers in Benchmarks below.

Features

  • Worker Pool: A fixed set of goroutines and a bounded queue, with blocking, non-blocking, and context-aware submits
  • Retries: Exponential backoff, a delay cap, jitter, per-attempt timeouts, and Permanent errors that skip retrying
  • Panic Recovery: A panicking job fails with a *PanicError carrying its stack trace; the worker keeps running
  • Graceful Shutdown: Close drains the queue; CloseWithTimeout force-cancels after a deadline
  • Pause/Resume: Stop picking up new work without shutting down
  • Middleware: Wrap every attempt for logging, tracing, or metrics
  • Lifecycle Hooks: Callbacks for job start/complete/fail and worker start/stop
  • Typed Futures: Generic async results with SubmitTyped and Future[T]
  • ErrorGroup & Group: Fan-out/fan-in with cancel-on-first-error or full error collection
  • Batch: Generic chunked processing for bulk operations
  • Scheduler: Cron expressions, intervals, daily times, and one-shot tasks with overlap policies and jitter
  • Ticker / Debouncer / Throttler / RateLimiter: Common flow-control patterns, built on the pool
  • Metrics: Atomic counters, success rate, and throughput
  • Zero Dependencies: Standard library only

Installation

go get github.com/KARTIKrocks/go-worker

Quick Start

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/KARTIKrocks/go-worker"
)

func main() {
    pool, err := worker.NewPool(
        worker.WithWorkers(8),
        worker.WithQueueSize(500),
        worker.WithMaxRetries(3),
        worker.WithJobTimeout(time.Minute), // per attempt
    )
    if err != nil {
        log.Fatal(err)
    }
    defer pool.Close() // drains queued and running jobs

    // Fire and forget
    if err := pool.Submit(func(ctx context.Context) error {
        fmt.Println("hello from a worker")
        return nil
    }); err != nil {
        log.Printf("submit: %v", err) // ErrPoolClosed
    }

    // Wait for the result, after retries
    if err := pool.SubmitWait(context.Background(), sendEmail); err != nil {
        log.Printf("email failed: %v", err)
    }

    // Typed async result
    future := worker.SubmitTyped(pool, countUsers)
    n, err := future.Await(context.Background())
    if err != nil {
        log.Printf("count failed: %v", err)
    }
    fmt.Println("users:", n)
}

Recurring work:

sched := worker.NewScheduler(pool)
sched.EveryFunc("cleanup", 5*time.Minute, cleanup)
if err := sched.CronFunc("report", "0 9 * * MON-FRI", sendReport); err != nil {
    log.Fatal(err) // invalid cron expression
}
sched.Start()
defer sched.Stop()

Note: go-worker is pre-1.0, and v0.1.0 changed several defaults — most notably, JobTimeout now defaults to no timeout and applies to each attempt. See the changelog when upgrading.

Documentation

Full guides live at kartikrocks.github.io/go-worker:

Guide Covers
Getting Started Create a pool and submit work
Worker Pool Every way to submit, and what each waits for
Configuration Every option and its default
Retries and Panics What is retried, backoff, jitter, PanicError
Lifecycle Pause, resume, graceful and forced shutdown
Middleware Wrapping every attempt
Hooks Job and worker lifecycle callbacks
Future Typed async results
Groups Group and ErrorGroup fan-out
Batch Chunked bulk processing
Scheduler Tasks, overlap policies, runtime control
Schedules and Cron Cron syntax, time zones, custom schedules
Ticker, Debouncer, Throttler, Rate Limiter Flow-control helpers
Metrics Snapshot, derived rates, Prometheus export
Errors Sentinel errors and matching

Exact type signatures are generated from source on pkg.go.dev.

Runnable programs are in examples/ — basic, retries, middleware, future, errorgroup, batch, scheduler, rate-limiter, and graceful-shutdown.

Benchmarks

Measured on an Intel i5-11400H @ 2.70GHz (12 cores), Go 1.27, Linux; median of three runs. Each benchmark uses an 8-worker pool with 12 goroutines submitting no-op jobs in parallel.

go test -bench=. -benchmem ./...
Benchmark ns/op B/op Allocs
Submit 909 248 5
SubmitWait 1,473 392 8
SubmitWait with one middleware 1,466 440 11
TrySubmit 60 7 0

Submit runs with the queue full, so its figure is the pool's steady-state throughput. TrySubmit mostly measures the rejection path: with the queue full, most calls return ErrPoolFull immediately.

Thread Safety

Every Pool, Scheduler, Future, group, and flow-control helper method is safe for concurrent use. Batch is the exception: do not call Add while Process runs.

Security

Every push and pull request to main is scanned by CodeQL, and govulncheck gates every merge on advisories reachable from this code's call graph. go-worker has no third-party dependencies, so that scan covers the standard library it uses; Dependabot tracks the GitHub Actions and the docs site.

Please report vulnerabilities privately through GitHub Security Advisories, not public issues.

License

MIT

Contributing

Contributions welcome — open an issue to discuss a change, or send a pull request. Run make ci before opening one.

Documentation

Overview

Package worker provides a high-performance, zero-dependency worker pool for concurrent job processing in Go.

Features

  • Configurable worker count and bounded job queue
  • Graceful shutdown with optional timeout
  • Automatic retries with exponential backoff
  • Panic recovery (workers survive panics)
  • Middleware chain for cross-cutting concerns
  • Lifecycle hooks for observability
  • Atomic metrics (submitted, completed, failed, panicked, active, latency)
  • Pause/Resume for flow control
  • ErrorGroup and Group for fan-out/fan-in patterns
  • Generic Batch processor for bulk operations
  • Generic Future for typed async results
  • Scheduler with cron, interval, daily, and one-shot schedules
  • Ticker, Debouncer, Throttler, and RateLimiter helpers
  • Zero external dependencies

Quick Start 

pool, err := worker.NewPool(
    worker.WithWorkers(8),
    worker.WithQueueSize(500),
    worker.WithMaxRetries(3),
)
if err != nil {
    log.Fatal(err)
}
defer pool.Close()

// Fire and forget
pool.Submit(func(ctx context.Context) error {
    return doWork(ctx)
})

// Wait for result
err = pool.SubmitWait(ctx, func(ctx context.Context) error {
    return doWork(ctx)
})

// Typed async result
future := worker.SubmitTyped(pool, func(ctx context.Context) (int, error) {
    return compute(ctx)
})
val, err := future.Await(ctx)

Package worker provides a high-performance worker pool for concurrent job processing with retries, panic recovery, middleware, and observability.

Basic usage: 

pool, err := worker.NewPool(
    worker.WithWorkers(8),
    worker.WithQueueSize(500),
)
if err != nil {
    log.Fatal(err)
}
defer pool.Close()

pool.Submit(func(ctx context.Context) error {
    return doWork(ctx)
})

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrPoolClosed    = errors.New("worker: pool is closed")
	ErrPoolFull      = errors.New("worker: queue is full")
	ErrJobPanic      = errors.New("worker: job panicked")
	ErrInvalidConfig = errors.New("worker: invalid configuration")
)

Sentinel errors returned by Pool operations.

View Source 
var (
	ErrRateLimited    = errors.New("worker: rate limit exceeded")
	ErrLimiterStopped = errors.New("worker: rate limiter stopped")
)

Sentinel errors returned by RateLimiter.

Functions

func Permanent added in v0.1.0 

func Permanent(err error) error

Permanent wraps err so the pool does not retry it. The returned error unwraps to err, so errors.Is and errors.As see through it. Permanent(nil) returns nil. 

if errors.Is(err, sql.ErrNoRows) {
    return worker.Permanent(err) // retrying will not help
}

Types

type Batch 

type Batch[T any] struct {
	// contains filtered or unexported fields
}

Batch processes items in configurable-sized chunks using a worker pool. It is useful for bulk database inserts, API calls, or similar fan-out patterns.

Batch is NOT safe for concurrent use. Do not call Add concurrently with Process or ProcessAll. 

batch := worker.NewBatch(pool, 100, func(ctx context.Context, users []User) error {
    return db.BulkInsert(ctx, users)
})
batch.Add(users...)
err := batch.Process(ctx)
Example 
package main

import (
	"context"
	"fmt"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, _ := worker.NewPool(worker.WithWorkers(4))
	defer pool.Close()

	batch := worker.NewBatch(pool, 3, func(ctx context.Context, items []int) error {
		fmt.Printf("processing batch of %d items\n", len(items))
		return nil
	})

	for i := range 7 {
		batch.Add(i)
	}
	_ = batch.Process(context.Background())
}

func NewBatch 

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

NewBatch creates a batch processor. The processor function is called once per chunk of up to batchSize items.

func (*Batch[T]) Add 

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

Add appends items to be processed.

func (*Batch[T]) Clear 

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

Clear removes all pending items.

func (*Batch[T]) Len 

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

Len returns the number of pending items.

func (*Batch[T]) Process 

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

Process processes all items in batches using an ErrorGroup. It cancels remaining batches on the first error.

func (*Batch[T]) ProcessAll 

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

ProcessAll processes all items and collects all errors (does not cancel on first error).

type CronSchedule 

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

CronSchedule fires based on cron-like expressions.

Format: "minute hour day-of-month month day-of-week"

Each field accepts *, single values, ranges (1-5), lists (1,3,5) and steps (*/15, 0-30/10, 5/20). Months accept JAN-DEC and weekdays SUN-SAT (case insensitive); weekday 7 is also Sunday. The macros @yearly (@annually), @monthly, @weekly, @daily (@midnight) and @hourly are supported.

Times that do not exist because of a daylight-saving jump are skipped that day.

As in standard cron, when both day-of-month and day-of-week are restricted a day matches if either does; a field starting with "*" (such as "*/2") counts as unrestricted for this rule.

func ParseCron 

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

ParseCron parses a standard 5-field cron expression or a macro.

Examples: 

"*/15 * * * *"    — every 15 minutes
"0 9 * * MON-FRI" — 9 AM on weekdays
"0 0 1 * *"       — midnight on the 1st of each month
"@hourly"         — at the start of every hour

func (*CronSchedule) Next 

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

Next implements Schedule.

func (*CronSchedule) String 

func (s *CronSchedule) String() string

String returns the original cron expression.

type DailySchedule 

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

DailySchedule fires at specific HH:MM wall-clock times each day, in the time zone of the time passed to Next. A time skipped by a daylight-saving change (02:30 when clocks jump from 02:00 to 03:00) fires after the jump (03:30).

func NewDailySchedule 

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

NewDailySchedule creates a schedule from "HH:MM" strings.

func (*DailySchedule) Next 

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

Next implements Schedule.

type Debouncer 

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

Debouncer collapses rapid submissions, executing only the last job after a quiet period of the configured delay.

Jobs submitted when a timer fires are submitted with Pool.SubmitJob; if that fails (for example, the pool is closed) the job is dropped.

func NewDebouncer 

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

NewDebouncer creates a trailing-edge debouncer.

func NewDebouncerLeading 

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

NewDebouncerLeading creates a leading-edge debouncer that fires immediately on the first call, then ignores subsequent calls until delay has passed without a call.

func (*Debouncer) Cancel 

func (d *Debouncer) Cancel()

Cancel cancels the pending debounced job. Like time.Timer.Stop, it cannot recall a job whose quiet period has already ended: once the timer fires, the job is no longer pending and is submitted even if Cancel runs meanwhile.

func (*Debouncer) Flush 

func (d *Debouncer) Flush() error

Flush immediately submits the pending job, if any, and returns the submit error. In leading mode there is never a pending job; Flush only ends the current window.

func (*Debouncer) Submit 

func (d *Debouncer) Submit(job Job)

Submit submits a job with debouncing.

func (*Debouncer) SubmitFunc 

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

SubmitFunc submits a function with debouncing.

type ErrorGroup 

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

ErrorGroup runs multiple jobs concurrently and cancels the group context on the first error. Modeled after golang.org/x/sync/errgroup.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, _ := worker.NewPool(worker.WithWorkers(4))
	defer pool.Close()

	g := worker.NewErrorGroup(pool)

	g.Go(func(ctx context.Context) error {
		// Task 1
		return nil
	})
	g.Go(func(ctx context.Context) error {
		// Task 2
		return nil
	})

	if err := g.Wait(); err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println("all tasks completed")
}

Output:
all tasks completed

func NewErrorGroup 

func NewErrorGroup(pool *Pool) *ErrorGroup

NewErrorGroup creates an error group using context.Background.

func NewErrorGroupContext 

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

NewErrorGroupContext creates an error group bound to a parent context. The derived context is cancelled when the first function returns an error or when ErrorGroup.Wait returns.

func (*ErrorGroup) Context 

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

Context returns the group's derived context. It is cancelled on the first error or when Wait returns.

func (*ErrorGroup) Go 

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

Go submits a function to the error group. If fn returns an error, the group context is cancelled, signaling other functions to stop.

func (*ErrorGroup) Wait 

func (g *ErrorGroup) Wait() error

Wait blocks until all functions complete and returns the first error, if any.

type FixedTimeSchedule 

type FixedTimeSchedule struct {
	Start    time.Time
	Interval time.Duration
}

FixedTimeSchedule fires at a start time, then repeats at a fixed interval. A non-positive Interval fires once at Start.

func (*FixedTimeSchedule) Next 

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

Next implements Schedule.

type Future 

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future represents an asynchronous computation that will produce a value of type T. 

future := worker.SubmitTyped[int](pool, func(ctx context.Context) (int, error) {
    return computeExpensiveThing(ctx)
})
result, err := future.Await(ctx)

func SubmitTyped 

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

SubmitTyped submits a typed function and returns a Future for the result. With pool retries enabled, the Future resolves with the outcome of the final attempt. If fn panics, the Future resolves with an error wrapping ErrJobPanic.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, _ := worker.NewPool()
	defer pool.Close()

	future := worker.SubmitTyped(pool, func(ctx context.Context) (int, error) {
		return 42, nil
	})

	val, err := future.Await(context.Background())
	fmt.Printf("val=%d err=%v\n", val, err)
}

Output:
val=42 err=<nil>

func (*Future[T]) Await 

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

Await blocks until the result is available or the context expires.

func (*Future[T]) Done 

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

Done reports whether the result is already available (non-blocking).

type Group 

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

Group runs multiple jobs concurrently on a pool and collects results. Unlike ErrorGroup, it does not cancel on the first error.

func NewGroup 

func NewGroup(pool *Pool) *Group

NewGroup creates a job group using context.Background.

func NewGroupContext 

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

NewGroupContext creates a job group bound to a parent context.

func (*Group) Go 

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

Go submits a function to the group for concurrent execution. The function's context is cancelled when the group's context is, when the pool's job timeout expires, or when the pool is force-closed.

func (*Group) Wait 

func (g *Group) Wait() error

Wait blocks until all submitted functions complete and returns the first error.

func (*Group) WaitAll 

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

WaitAll blocks until all submitted functions complete and returns all errors.

type Hooks 

type Hooks struct {
	OnJobStart    func(job Job)
	OnJobComplete func(job Job, duration time.Duration)
	OnJobFailed   func(job Job, err error, duration time.Duration)
	OnWorkerStart func(workerID int)
	OnWorkerStop  func(workerID int)
}

Hooks contains lifecycle callbacks. All fields are optional. A panic in a hook is recovered (and logged if a Logger is set) so it cannot crash the worker.

type IntervalSchedule 

type IntervalSchedule struct {
	Interval time.Duration
}

IntervalSchedule fires at fixed intervals. A non-positive Interval never fires.

func (*IntervalSchedule) Next 

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

Next implements Schedule.

type Job 

type Job interface {
	Process(ctx context.Context) error
}

Job represents a unit of work. Implement this interface for complex jobs that need to carry state. For simple functions, use Pool.Submit directly.

type JobFunc 

type JobFunc func(ctx context.Context) error

JobFunc adapts a function to the Job interface.

func (JobFunc) Process 

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

Process implements Job.

type Logger 

type Logger interface {
	Debug(msg string, args ...any)
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
}

Logger is a structured logging interface compatible with log/slog.

type MetricsSnapshot 

type MetricsSnapshot struct {
	JobsSubmitted int64
	JobsCompleted int64
	JobsFailed    int64
	JobsRetried   int64
	JobsPanicked  int64
	ActiveWorkers int32
	QueueLength   int32
	TotalDuration time.Duration // sum of all job durations
	Elapsed       time.Duration // wall-clock time since pool creation
}

MetricsSnapshot is an immutable point-in-time copy of pool metrics.

func (MetricsSnapshot) AverageJobDuration 

func (s MetricsSnapshot) AverageJobDuration() time.Duration

AverageJobDuration returns the mean duration of completed jobs. Returns 0 if no jobs have completed.

func (MetricsSnapshot) SuccessRate 

func (s MetricsSnapshot) SuccessRate() float64

SuccessRate returns the fraction of jobs that completed without error (0.0–1.0). Returns 1.0 if no jobs have been processed.

func (MetricsSnapshot) Throughput 

func (s MetricsSnapshot) Throughput() float64

Throughput returns the average jobs completed per second of wall-clock time. Returns 0 if no jobs have completed or if no time has elapsed.

type Middleware 

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

Middleware wraps job processing. Use for cross-cutting concerns like logging, tracing, or metrics collection. 

func timingMiddleware(next func(ctx context.Context) error) func(ctx context.Context) error {
    return func(ctx context.Context) error {
        start := time.Now()
        err := next(ctx)
        fmt.Println("took", time.Since(start))
        return err
    }
}

type OnceSchedule 

type OnceSchedule struct {
	At time.Time
	// contains filtered or unexported fields
}

OnceSchedule fires exactly once at a given time.

func (*OnceSchedule) Next 

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

Next implements Schedule.

type Option 

type Option func(p *Pool)

Option configures a Pool. Pass to NewPool; create one with the With... functions.

func WithHooks 

func WithHooks(h Hooks) Option

WithHooks sets lifecycle hooks for the pool.

func WithJobTimeout 

func WithJobTimeout(d time.Duration) Option

WithJobTimeout sets the maximum duration of each attempt of a job; with retries, every attempt gets its own timeout, and a timed-out attempt is retried. Zero means no timeout. Default: 0.

func WithLogger 

func WithLogger(l Logger) Option

WithLogger sets the structured logger for the pool.

func WithMaxRetries 

func WithMaxRetries(n int) Option

WithMaxRetries sets how many times a failed job is retried. Default: 0 (no retries). Errors wrapped with Permanent and panics are never retried.

func WithMaxRetryDelay 

func WithMaxRetryDelay(d time.Duration) Option

WithMaxRetryDelay caps the delay between retries, with or without exponential backoff. Zero means no cap. Default: 0.

func WithMiddleware 

func WithMiddleware(mw ...Middleware) Option

WithMiddleware appends middleware to the processing chain. Middleware is applied in the order given (first added = outermost wrapper).

func WithPanicHandler 

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

WithPanicHandler sets a handler called when a job panics. The handler receives the job and the recovered value. The job's error is a *PanicError, which also carries the stack trace.

func WithQueueSize 

func WithQueueSize(n int) Option

WithQueueSize sets the buffered channel capacity for the job queue. Default: 100.

func WithRetryBackoff 

func WithRetryBackoff(enabled bool) Option

WithRetryBackoff enables exponential backoff on retries. Default: true.

func WithRetryDelay 

func WithRetryDelay(d time.Duration) Option

WithRetryDelay sets the base delay between retries. Default: 1s.

func WithRetryJitter added in v0.1.0 

func WithRetryJitter(fraction float64) Option

WithRetryJitter randomizes each retry delay by subtracting up to fraction of it (0 to 1), so retries of many failing jobs do not happen in lockstep. Default: 0 (no jitter).

func WithWorkers 

func WithWorkers(n int) Option

WithWorkers sets the number of worker goroutines. Default: 4.

type OverlapPolicy 

type OverlapPolicy int

OverlapPolicy defines behavior when a task is still running at its next scheduled time. 

const (
	// OverlapSkip skips the new execution if the previous is still running (default).
	OverlapSkip OverlapPolicy = iota
	// OverlapAllow allows concurrent executions.
	OverlapAllow
	// OverlapQueue queues the new execution to run after the current completes.
	OverlapQueue
)

type PanicError added in v0.1.0 

type PanicError struct {
	Value any    // the value passed to panic
	Stack []byte // stack trace of the panicking goroutine
}

PanicError is the error a job returns when it panics. It matches ErrJobPanic with errors.Is, and unwraps to the panic value if that value is itself an error.

func (*PanicError) Error added in v0.1.0 

func (e *PanicError) Error() string

func (*PanicError) Is added in v0.1.0 

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

Is reports whether target is ErrJobPanic.

func (*PanicError) Unwrap added in v0.1.0 

func (e *PanicError) Unwrap() error

Unwrap returns the panic value if it is an error, or nil.

type Pool 

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

Pool manages a set of worker goroutines that process jobs concurrently. It is safe for concurrent use by multiple goroutines.

func NewPool 

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

NewPool creates and starts a new worker pool. Workers begin processing immediately. Use Pool.Close for graceful shutdown.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, err := worker.NewPool(
		worker.WithWorkers(4),
		worker.WithQueueSize(100),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()

	_ = pool.Submit(func(ctx context.Context) error {
		fmt.Println("Hello from worker!")
		return nil
	})

	pool.Close()
}

Output:
Hello from worker!
Example (WithMiddleware) 
package main

import (
	"context"
	"log"
	"time"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	logging := func(next func(ctx context.Context) error) func(ctx context.Context) error {
		return func(ctx context.Context) error {
			start := time.Now()
			err := next(ctx)
			log.Printf("job took %v, err=%v", time.Since(start), err)
			return err
		}
	}

	pool, _ := worker.NewPool(
		worker.WithMiddleware(logging),
	)
	defer pool.Close()
}
Example (WithRetries) 
package main

import (
	"context"
	"log"
	"time"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, _ := worker.NewPool(
		worker.WithWorkers(2),
		worker.WithMaxRetries(3),
		worker.WithRetryDelay(100*time.Millisecond),
		worker.WithRetryBackoff(true),
		worker.WithPanicHandler(func(job worker.Job, r any) {
			log.Printf("job panicked: %v", r)
		}),
	)
	defer pool.Close()

	_ = pool.Submit(func(ctx context.Context) error {
		// This will be retried up to 3 times on failure
		return nil
	})
}

func (*Pool) ActiveWorkers 

func (p *Pool) ActiveWorkers() int

ActiveWorkers returns the number of workers currently processing a job.

func (*Pool) Close 

func (p *Pool) Close() error

Close gracefully shuts down the pool. It closes the job queue and waits for all in-flight jobs to complete. Close is idempotent.

func (*Pool) CloseWithTimeout 

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

CloseWithTimeout gracefully shuts down the pool. If timeout > 0 and workers don't finish in time, in-flight jobs are cancelled via context and jobs still in the queue are dropped; waiters on dropped jobs receive ErrPoolClosed.

func (*Pool) IsClosed 

func (p *Pool) IsClosed() bool

IsClosed reports whether the pool has been closed.

func (*Pool) IsPaused 

func (p *Pool) IsPaused() bool

IsPaused reports whether the pool is paused.

func (*Pool) Pause 

func (p *Pool) Pause()

Pause pauses all workers. Workers finish their current job but do not pick up new ones until Pool.Resume is called.

func (*Pool) QueueLength 

func (p *Pool) QueueLength() int

QueueLength returns the current number of jobs in the queue. Submitters blocked waiting for queue space are not counted.

func (*Pool) Resume 

func (p *Pool) Resume()

Resume resumes a paused pool.

func (*Pool) Snapshot 

func (p *Pool) Snapshot() MetricsSnapshot

Snapshot returns a point-in-time copy of pool metrics.

func (*Pool) Submit 

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

Submit submits a function for processing. It blocks if the queue is full until space is available or the pool is closed.

func (*Pool) SubmitContext 

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

SubmitContext submits a Job with a context. If the context expires before the job can be enqueued, the context error is returned. If the queue is full, it blocks until space is available, the context expires, or the pool closes.

func (*Pool) SubmitJob 

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

SubmitJob submits a Job for processing. It blocks if the queue is full.

func (*Pool) SubmitJobWait 

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

SubmitJobWait submits a Job and blocks until it completes. The job runs with a context derived from ctx, so cancelling ctx both stops the wait (returning ctx.Err()) and cancels the job.

func (*Pool) SubmitWait 

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

SubmitWait submits a function and blocks until it completes, returning its error.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/KARTIKrocks/go-worker"
)

func main() {
	pool, _ := worker.NewPool()
	defer pool.Close()

	err := pool.SubmitWait(context.Background(), func(ctx context.Context) error {
		// Do work...
		return nil
	})
	fmt.Println("error:", err)
}

Output:
error: <nil>

func (*Pool) TrySubmit 

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

TrySubmit attempts to submit a function without blocking. Returns ErrPoolFull if the queue is full.

func (*Pool) TrySubmitJob 

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

TrySubmitJob attempts to submit a Job without blocking.

type RateLimiter 

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

RateLimiter is a token-bucket rate limiter that wraps a worker pool.

func NewRateLimiter 

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

NewRateLimiter creates a rate limiter that allows n jobs per interval. It panics if n < 1 or interval <= 0.

func (*RateLimiter) Stop 

func (rl *RateLimiter) Stop()

Stop stops the token refiller. Subsequent submits return ErrLimiterStopped.

func (*RateLimiter) Submit 

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

Submit waits for a token and then submits the job. Respects the provided context. Returns ErrLimiterStopped once RateLimiter.Stop has been called. If the pool rejects the job, the token is returned.

func (*RateLimiter) TrySubmit 

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

TrySubmit attempts to submit without waiting for a token. It returns ErrRateLimited if no token is available, or the pool's error (such as ErrPoolFull) if the pool rejects the job, in which case the token is returned.

type Schedule 

type Schedule interface {
	// Next returns the next execution time after the given time.
	// A zero time means the schedule has no more executions.
	Next(after time.Time) time.Time
}

Schedule determines when a task should next run.

type Scheduler 

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

Scheduler manages periodic task execution on a worker pool.

func NewScheduler 

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

NewScheduler creates a new scheduler that dispatches work to pool.

func (*Scheduler) At 

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

At schedules a job starting at a specific time, then repeating at interval.

func (*Scheduler) Cron 

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

Cron schedules a job using a cron expression.

func (*Scheduler) CronFunc 

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

CronFunc is a convenience wrapper around Scheduler.Cron.

func (*Scheduler) Daily 

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

Daily schedules a job at specific "HH:MM" times each day.

func (*Scheduler) DailyFunc 

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

DailyFunc is a convenience wrapper around Scheduler.Daily.

func (*Scheduler) Every 

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

Every schedules a job at a fixed interval.

func (*Scheduler) EveryFunc 

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

EveryFunc is a convenience wrapper around Scheduler.Every.

func (*Scheduler) IsRunning 

func (s *Scheduler) IsRunning() bool

IsRunning reports whether the scheduler loop is active.

func (*Scheduler) Once 

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

Once schedules a job to run exactly once at the given time.

func (*Scheduler) OnceFunc 

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

OnceFunc is a convenience wrapper around Scheduler.Once.

func (*Scheduler) Pause 

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

Pause pauses a task by name. Returns false if the task does not exist.

func (*Scheduler) Remove 

func (s *Scheduler) Remove(name string)

Remove cancels and removes a scheduled task.

func (*Scheduler) Resume 

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

Resume resumes a paused task. Returns false if the task does not exist.

func (*Scheduler) Schedule 

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

Schedule adds a job with a custom Schedule. Scheduling a name that is already registered replaces that task and cancels its in-flight runs.

func (*Scheduler) ScheduleFunc 

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

ScheduleFunc is a convenience wrapper around Scheduler.Schedule.

func (*Scheduler) Start 

func (s *Scheduler) Start()

Start begins the scheduler loop. It is idempotent. A stopped scheduler cannot be restarted: Start after Scheduler.Stop does nothing.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop stops the scheduler and waits for the loop goroutine to exit. It is permanent: the scheduler cannot be started again. The contexts of in-flight task runs are cancelled, but Stop does not wait for those runs to return.

func (*Scheduler) TaskInfo 

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

TaskInfo returns information about a named task, or nil if not found.

func (*Scheduler) Tasks 

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

Tasks returns the names of all registered tasks.

func (*Scheduler) Trigger 

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

Trigger runs a task now, outside its normal schedule. Like a scheduled run, it respects the task's pause state, WithMaxRuns limit and overlap policy. It reports whether a run was started or queued; false also means the task does not exist.

type SchedulerOption 

type SchedulerOption func(*Scheduler)

SchedulerOption configures a Scheduler.

func WithOnTaskEnd 

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

WithOnTaskEnd sets a callback invoked when a task finishes.

func WithOnTaskStart 

func WithOnTaskStart(fn func(name string)) SchedulerOption

WithOnTaskStart sets a callback invoked when a task begins execution.

func WithSchedulerLogger 

func WithSchedulerLogger(l Logger) SchedulerOption

WithSchedulerLogger sets the scheduler's logger.

func WithTickInterval 

func WithTickInterval(d time.Duration) SchedulerOption

WithTickInterval sets the scheduler's tick resolution. Default: 100ms. Lower values give more precise scheduling but use more CPU.

type TaskInfo 

type TaskInfo struct {
	Name         string
	Next         time.Time
	Running      bool
	Paused       bool
	RunCount     int64
	MaxRuns      int
	LastRun      time.Time
	LastError    error
	LastDuration time.Duration
	TotalRuns    int64
	TotalErrors  int64
}

TaskInfo is a snapshot of a scheduled task's state.

func (*TaskInfo) SuccessRate 

func (i *TaskInfo) SuccessRate() float64

SuccessRate returns the fraction of successful runs (0.0–1.0).

type TaskOption 

type TaskOption func(*scheduledTask)

TaskOption configures an individual scheduled task.

func WithJitter 

func WithJitter(d time.Duration) TaskOption

WithJitter adds random jitter to the schedule to prevent thundering herd.

func WithMaxRuns 

func WithMaxRuns(n int) TaskOption

WithMaxRuns limits the number of executions (0 = unlimited).

func WithOverlapPolicy 

func WithOverlapPolicy(p OverlapPolicy) TaskOption

WithOverlapPolicy sets the behavior when a task is still running at its next scheduled time. Default: OverlapSkip.

func WithRunImmediate 

func WithRunImmediate() TaskOption

WithRunImmediate causes the task to fire immediately when the scheduler starts. For a OnceSchedule task, the immediate run replaces the scheduled one.

type Throttler 

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

Throttler limits job execution to at most once per interval.

func NewThrottler 

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

NewThrottler creates a throttler.

func (*Throttler) Force 

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

Force submits a job unconditionally, starting a new throttle window and dropping any pending trailing job. It returns the submit error.

func (*Throttler) Reset 

func (t *Throttler) Reset()

Reset clears the throttle state and drops any pending trailing job.

func (*Throttler) Submit 

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

Submit submits a job if enough time has passed since the last execution. It returns true if the job was submitted successfully.

func (*Throttler) SubmitTrailing 

func (t *Throttler) SubmitTrailing(job Job)

SubmitTrailing submits a job that will fire at the end of the throttle window. Repeated calls within the window replace the pending job. If the window is already open, the job is submitted immediately. Submit errors are dropped.

type Ticker 

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

Ticker submits a job at a fixed interval.

func NewTicker 

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

NewTicker creates a ticker that submits job every interval. It panics if interval <= 0.

func NewTickerFunc 

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

NewTickerFunc creates a ticker from a function.

func NewTickerImmediate 

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

NewTickerImmediate creates a ticker that fires immediately, then at interval.

func (*Ticker) Start 

func (t *Ticker) Start()

Start begins the ticker loop. It is idempotent.

func (*Ticker) Stop 

func (t *Ticker) Stop()

Stop stops the ticker and waits for the goroutine to exit. Runs already submitted to the pool are not cancelled.

Source Files

View all Source files

Directories

Path Synopsis
examples
basic command
Basic demonstrates the core worker pool functionality: fire-and-forget, wait-for-result, non-blocking, and context-aware submission.
 Basic demonstrates the core worker pool functionality: fire-and-forget, wait-for-result, non-blocking, and context-aware submission.
batch command
Batch demonstrates bulk processing of items in configurable-sized chunks.
 Batch demonstrates bulk processing of items in configurable-sized chunks.
errorgroup command
ErrorGroup demonstrates fan-out/fan-in patterns using ErrorGroup (cancel on first error) and Group (collect all errors).
 ErrorGroup demonstrates fan-out/fan-in patterns using ErrorGroup (cancel on first error) and Group (collect all errors).
future command
Future demonstrates generic typed async results using SubmitTyped and Future[T].
 Future demonstrates generic typed async results using SubmitTyped and Future[T].
graceful-shutdown command
Graceful-shutdown demonstrates Close, CloseWithTimeout, and context cancellation behavior during pool shutdown.
 Graceful-shutdown demonstrates Close, CloseWithTimeout, and context cancellation behavior during pool shutdown.
middleware command
Middleware demonstrates composable middleware chains for cross-cutting concerns like logging, timing, and tracing.
 Middleware demonstrates composable middleware chains for cross-cutting concerns like logging, timing, and tracing.
rate-limiter command
Rate-limiter demonstrates the Ticker, Debouncer, Throttler, and RateLimiter helpers.
 Rate-limiter demonstrates the Ticker, Debouncer, Throttler, and RateLimiter helpers.
retries command
Retries demonstrates automatic retries with exponential backoff, max retry delay cap, permanent errors, and panic recovery.
 Retries demonstrates automatic retries with exponential backoff, max retry delay cap, permanent errors, and panic recovery.
scheduler command
Scheduler demonstrates periodic task execution using cron expressions, fixed intervals, and runtime control (pause, resume, trigger, remove).
 Scheduler demonstrates periodic task execution using cron expressions, fixed intervals, and runtime control (pause, resume, trigger, remove).

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.