worker

package module
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Feb 17, 2026 License: MIT Imports: 10 Imported by: 0

README

go-worker

Go Reference Go Report Card

A high-performance, zero-dependency worker pool for Go with retries, scheduling, and observability.

Features

  • Worker Pool — configurable goroutine pool with bounded queue
  • Retries — automatic retries with exponential backoff
  • Panic Recovery — workers survive panics and keep processing
  • Middleware — composable chain for logging, tracing, metrics
  • Lifecycle Hooks — callbacks for job start/complete/fail and worker start/stop
  • Metrics — atomic counters for submitted/completed/failed/panicked/active/latency
  • Pause/Resume — flow control without shutdown
  • ErrorGroup & Group — fan-out/fan-in with error collection or cancel-on-first-error
  • Batch — generic batch processor for bulk operations
  • Future — generic typed async results
  • Scheduler — cron expressions, fixed intervals, daily times, one-shot
  • Ticker / Debouncer / Throttler / RateLimiter — common patterns built-in
  • Zero Dependencies — stdlib only

Install

go get github.com/KARTIKrocks/go-worker

Requires Go 1.23+.

Quick Start

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

// Fire and forget
pool.Submit(func(ctx context.Context) error {
    fmt.Println("Hello from worker!")
    return nil
})

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

Configuration

All configuration uses functional options:

pool, err := worker.NewPool(
    worker.WithWorkers(8),           // goroutine count (default: 4)
    worker.WithQueueSize(500),       // buffered channel size (default: 100)
    worker.WithJobTimeout(time.Minute), // per-job timeout (default: 30s, 0=none)
    worker.WithMaxRetries(3),        // retry failed jobs (default: 0)
    worker.WithRetryDelay(time.Second), // base retry delay (default: 1s)
    worker.WithRetryBackoff(true),      // exponential backoff (default: true)
    worker.WithMaxRetryDelay(time.Minute), // cap retry delay (default: 0=no cap)
    worker.WithPanicHandler(func(job worker.Job, r any) {
        log.Printf("panic: %v", r)
    }),
    worker.WithLogger(slogLogger),   // any Logger interface
    worker.WithMiddleware(mw1, mw2), // middleware chain
    worker.WithHooks(worker.Hooks{
        OnJobStart:    func(j worker.Job) { /* ... */ },
        OnJobComplete: func(j worker.Job, d time.Duration) { /* ... */ },
        OnJobFailed:   func(j worker.Job, err error, d time.Duration) { /* ... */ },
        OnWorkerStart: func(id int) { /* ... */ },
        OnWorkerStop:  func(id int) { /* ... */ },
    }),
)

Submitting Jobs

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

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

// Non-blocking (returns ErrPoolFull if queue is full)
err := pool.TrySubmit(func(ctx context.Context) error {
    return doWork(ctx)
})

// With context (blocks until enqueued or ctx expires)
err := pool.SubmitContext(ctx, myJob)

// Job interface
type MyJob struct{ ID string }
func (j *MyJob) Process(ctx context.Context) error { return nil }
pool.SubmitJob(&MyJob{ID: "1"})

Typed Results (Future)

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

val, err := future.Await(ctx)

if future.Done() {
    // result is ready
}

Error Group

Cancel all jobs on first error (like errgroup.Group):

g := worker.NewErrorGroup(pool)

g.Go(func(ctx context.Context) error { return fetchA(ctx) })
g.Go(func(ctx context.Context) error { return fetchB(ctx) })
g.Go(func(ctx context.Context) error { return fetchC(ctx) })

err := g.Wait() // first error, or nil

Group (collect all errors)

g := worker.NewGroup(pool)

g.Go(func(ctx context.Context) error { return task1(ctx) })
g.Go(func(ctx context.Context) error { return task2(ctx) })

errs := g.WaitAll() // []error

Batch Processing

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

batch.Add(users...)
err := batch.Process(ctx) // cancels remaining on first error

// Or collect all errors:
errs := batch.ProcessAll(ctx)

Scheduler

sched := worker.NewScheduler(pool)

// Fixed interval
sched.EveryFunc("cleanup", 5*time.Minute, cleanup)

// Cron expression
sched.CronFunc("report", "0 9 * * 1-5", generateReport)

// Daily at specific times
sched.DailyFunc("backup", []string{"02:00", "14:00"}, backup)

// One-shot
sched.OnceFunc("migrate", targetTime, migrate)

// With options
sched.EveryFunc("sync", time.Hour, syncData,
    worker.WithRunImmediate(),
    worker.WithJitter(30*time.Second),
    worker.WithOverlapPolicy(worker.OverlapSkip),
    worker.WithMaxRuns(10),
)

sched.Start()
defer sched.Stop()

// Runtime control
sched.Pause("sync")
sched.Resume("sync")
sched.Trigger("sync")  // manual fire
sched.Remove("sync")
info := sched.TaskInfo("cleanup")

Middleware

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)
        log.Printf("job took %v", time.Since(start))
        return err
    }
}

pool, _ := worker.NewPool(worker.WithMiddleware(timingMiddleware))

Metrics

snap := pool.Snapshot()

fmt.Printf("Submitted:  %d\n", snap.JobsSubmitted)
fmt.Printf("Completed:  %d\n", snap.JobsCompleted)
fmt.Printf("Failed:     %d\n", snap.JobsFailed)
fmt.Printf("Panicked:   %d\n", snap.JobsPanicked)
fmt.Printf("Retried:    %d\n", snap.JobsRetried)
fmt.Printf("Active:     %d\n", snap.ActiveWorkers)
fmt.Printf("Queued:     %d\n", snap.QueueLength)
fmt.Printf("Avg Dur:    %v\n", snap.AverageJobDuration())
fmt.Printf("Success:    %.1f%%\n", snap.SuccessRate()*100)
fmt.Printf("Throughput: %.1f/s\n", snap.Throughput())

Pool Control

pool.Pause()   // workers finish current job, then wait
pool.Resume()  // workers continue

pool.Close()   // graceful shutdown (waits for all jobs)
pool.CloseWithTimeout(30 * time.Second) // force after timeout

Helpers

// Ticker — submit a job at fixed intervals
ticker := worker.NewTickerFunc(pool, time.Minute, heartbeat)
ticker.Start()
defer ticker.Stop()

// Debouncer — only execute the last call after a quiet period
debouncer := worker.NewDebouncer(pool, 500*time.Millisecond)
debouncer.Submit(job) // only last one within 500ms runs

// Throttler — at most one execution per interval
throttler := worker.NewThrottler(pool, time.Second)
throttler.Submit(job) // runs at most 1/sec

// RateLimiter — token bucket
rl := worker.NewRateLimiter(pool, 10, time.Second) // 10 jobs/sec
rl.Submit(ctx, job)
defer rl.Stop()

Errors

errors.Is(err, worker.ErrPoolClosed)    // pool was closed
errors.Is(err, worker.ErrPoolFull)      // queue is full (TrySubmit)
errors.Is(err, worker.ErrJobPanic)      // job panicked
errors.Is(err, worker.ErrInvalidConfig) // bad configuration

License

MIT

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.

Functions

This section is empty.

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 Config 

type Config struct {
	Workers       int
	QueueSize     int
	JobTimeout    time.Duration
	MaxRetries    int
	RetryDelay    time.Duration
	MaxRetryDelay time.Duration // 0 means no cap
	RetryBackoff  bool
	PanicHandler  func(job Job, recovered any)
}

Config holds worker pool configuration. Use Option functions to set values.

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"

Supports *, */n, n-m, and comma-separated lists.

func ParseCron 

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

ParseCron parses a standard 5-field cron expression.

Examples: 

"*/15 * * * *"  — every 15 minutes
"0 9 * * 1-5"  — 9 AM on weekdays
"0 0 1 * *"    — midnight on the 1st of each month

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 times each day.

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.

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 within the delay window.

func (*Debouncer) Cancel 

func (d *Debouncer) Cancel()

Cancel cancels the pending debounced job.

func (*Debouncer) Flush 

func (d *Debouncer) Flush()

Flush immediately executes the pending job, if any.

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.

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.

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 receives the group's context, so group-level cancellation is visible.

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.

type IntervalSchedule 

type IntervalSchedule struct {
	Interval time.Duration
}

IntervalSchedule fires at fixed intervals.

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 Metrics 

type Metrics struct {
	JobsSubmitted atomic.Int64
	JobsCompleted atomic.Int64
	JobsFailed    atomic.Int64
	JobsRetried   atomic.Int64
	JobsPanicked  atomic.Int64
	ActiveWorkers atomic.Int32
	QueueLength   atomic.Int32
	TotalDuration atomic.Int64 // nanoseconds
	// contains filtered or unexported fields
}

Metrics holds atomic counters for pool statistics. Use Metrics.Snapshot to get an approximate point-in-time view.

func (*Metrics) Snapshot 

func (m *Metrics) Snapshot() MetricsSnapshot

Snapshot returns an approximate point-in-time copy of all metrics. Individual counters are read atomically, but the snapshot as a whole is not taken under a single lock, so values may be slightly inconsistent under concurrent load.

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, c *Config)

Option configures a Pool. Pass to NewPool.

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 for a single job execution. Zero means no timeout. Default: 30s.

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).

func WithMaxRetryDelay 

func WithMaxRetryDelay(d time.Duration) Option

WithMaxRetryDelay caps the maximum delay between retries when using 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.

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 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 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, remaining jobs are cancelled via context.

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.

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.

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.

func (*RateLimiter) Stop 

func (rl *RateLimiter) Stop()

Stop stops the token refiller.

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.

func (*RateLimiter) TrySubmit 

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

TrySubmit attempts to submit without waiting for a token.

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.

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.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop stops the scheduler and waits for the loop goroutine to exit. In-flight tasks submitted to the pool continue independently.

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 manually fires a task immediately, outside its normal schedule.

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.

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)

Force submits a job unconditionally, resetting the throttle window.

func (*Throttler) Reset 

func (t *Throttler) Reset()

Reset clears the throttle state.

func (*Throttler) Submit 

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

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

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.

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.

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.

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, and panic recovery.
 Retries demonstrates automatic retries with exponential backoff, max retry delay cap, 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.