middleware

Workers

A worker lifecycle library for Go — manage background goroutines with panic recovery, configurable restart, tracing, and structured shutdown.

Built on suture for Erlang-style supervisor trees. Part of the ColdBrew framework.

API Reference

middleware

import "github.com/go-coldbrew/workers/middleware"

Package middleware provides optional interceptors for go\-coldbrew/workers. None are applied by default. Use with [workers.Worker.Interceptors] or [workers.WithInterceptors].

Index

• func DefaultInterceptors() []workers.Middleware
• func DistributedLock(locker Locker, opts ...LockOption) workers.Middleware
• func Duration(observe func(name string, d time.Duration)) workers.Middleware
• func LogContext() workers.Middleware
• func Recover(onPanic func(name string, v any)) workers.Middleware
• func Slog() workers.Middleware
• func Timeout(d time.Duration) workers.Middleware
• func Tracing() workers.Middleware
• type LockOption
  • func WithKeyFunc(fn func(name string) string) LockOption
  • func WithOnNotAcquired(fn func(ctx context.Context, name string) error) LockOption
  • func WithSkipOnNotAcquired(logFn func(ctx context.Context, name string)) LockOption
  • func WithTTLFunc(fn func(name string) time.Duration) LockOption
• type Locker

func DefaultInterceptors

func DefaultInterceptors() []workers.Middleware

DefaultInterceptors returns the standard observability stack: Recover, LogContext, Tracing, Slog.

Usage:

workers.Run(ctx, myWorkers,
    workers.WithInterceptors(middleware.DefaultInterceptors()...),
)

func DistributedLock

func DistributedLock(locker Locker, opts ...LockOption) workers.Middleware

DistributedLock acquires a distributed lock before each cycle. If the lock is held by another instance, the cycle is skipped (or the onNotAcquired callback is invoked). Release uses context.WithoutCancel so that context cancellation does not prevent lock cleanup.

func Duration

func Duration(observe func(name string, d time.Duration)) workers.Middleware

Duration measures wall-clock time of each cycle and calls observe.

func LogContext

func LogContext() workers.Middleware

LogContext injects worker name and attempt into the log context so all log calls inside the worker automatically include them.

func Recover

func Recover(onPanic func(name string, v any)) workers.Middleware

Recover catches panics per-cycle, calls onPanic (if non-nil), and returns an error. The panic does not propagate to the supervisor — even if onPanic itself panics.

func Slog

func Slog() workers.Middleware

Slog emits structured log lines per cycle via go-coldbrew/log: "cycle start" (Info) before the handler, then "cycle end" (Info) on success or "cycle error" (Error) on failure. Pair with LogContext to include worker name and attempt in every log line automatically.

func Timeout

func Timeout(d time.Duration) workers.Middleware

Timeout enforces a per-cycle deadline. Distinct from [workers.Worker.WithTimeout] which controls graceful shutdown.

func Tracing

func Tracing() workers.Middleware

Tracing creates an OTEL span per cycle via go-coldbrew/tracing. The span is named "worker:<name>:cycle" and records errors. In typical use the span is a trace root because worker contexts carry no parent span; sampling is determined by the global TracerProvider's sampler.

The OTEL trace ID is injected into the log context as "trace" for correlation with the tracing backend.

type LockOption

LockOption configures DistributedLock behavior.

type LockOption func(*lockConfig)

func WithKeyFunc

func WithKeyFunc(fn func(name string) string) LockOption

WithKeyFunc sets a custom function to derive the lock key from the worker name. Default: "worker-lock:<name>".

func WithOnNotAcquired

func WithOnNotAcquired(fn func(ctx context.Context, name string) error) LockOption

WithOnNotAcquired sets a callback invoked when the lock is held by another instance. The cycle is skipped. Default: skip silently (return nil).

Caution: returning a non-nil error from this callback triggers the framework's normal error handling — for periodic workers, this means restart with backoff. If you want to log and skip, return nil from this callback or use WithSkipOnNotAcquired.

func WithSkipOnNotAcquired

func WithSkipOnNotAcquired(logFn func(ctx context.Context, name string)) LockOption

WithSkipOnNotAcquired is a convenience LockOption that calls logFn when the lock is held and skips the cycle (returns nil, no restart). If logFn is nil, the cycle is skipped silently (same as the default but explicit in intent).

func WithTTLFunc

func WithTTLFunc(fn func(name string) time.Duration) LockOption

WithTTLFunc sets a custom function to derive the lock TTL from the worker name. Default: 30s.

type Locker

Locker abstracts a distributed lock backend (e.g., Redis, etcd, Consul). If your lock implementation already has Acquire(ctx, key, ttl) (bool, error) and Release(ctx, key) error methods, it satisfies this interface directly — no adapter needed.

type Locker interface {
    // Acquire attempts to acquire a lock for the given key with a TTL.
    // Returns true if the lock was acquired, false if held by another instance.
    Acquire(ctx context.Context, key string, ttl time.Duration) (bool, error)
    // Release releases a previously acquired lock.
    Release(ctx context.Context, key string) error
}

Generated by gomarkdoc