worker

Feb 2, 2026

Changes in this version

type DurableTask

+ RunAt time.Time

type Task

+ RunAt time.Time

type TaskManager

+ func (tm *TaskManager) RegisterDurableTaskAfter(ctx context.Context, task DurableTask, delay time.Duration) error

+ func (tm *TaskManager) RegisterDurableTaskAt(ctx context.Context, task DurableTask, runAt time.Time) error

+ func (tm *TaskManager) RegisterTaskAfter(ctx context.Context, task *Task, delay time.Duration) error

+ func (tm *TaskManager) RegisterTaskAt(ctx context.Context, task *Task, runAt time.Time) error

Jan 31, 2026

Changes in this version

+ const DefaultQueueName

+ const DefaultQueueWeight

+ const DefaultTaskWeight

+ const MetadataQueueKey

+ const MetadataWeightKey

+ var ErrHandlerAlreadyRegistered = ewrap.New("handler already registered")

+ var ErrHandlerNameRequired = ewrap.New("handler name is required")

+ var ErrHandlerPayloadTypeMismatch = ewrap.New("handler payload type mismatch")

+ var ErrHandlerRegistryNil = ewrap.New("handler registry is nil")

+ var ErrHandlerSpecInvalid = ewrap.New("handler spec is invalid")

+ func AddTypedDurableHandler[T proto.Message](r *TypedDurableRegistry, name string, spec TypedDurableHandlerSpec[T]) error

+ func AddTypedHandler[T protoreflect.ProtoMessage](r *TypedHandlerRegistry, name string, spec TypedHandlerSpec[T]) error

type DurableHandlerSpec

+ func TypedDurableHandler[T proto.Message](spec TypedDurableHandlerSpec[T]) DurableHandlerSpec

type DurableTask

+ Queue string

+ Weight int

type HandlerSpec

+ func TypedHandler[T protoreflect.ProtoMessage](spec TypedHandlerSpec[T]) HandlerSpec

+ type QueueConfigurableBackend interface

+ ConfigureQueues func(defaultQueue string, weights map[string]int)

type RedisDurableBackend

+ func (b *RedisDurableBackend) ConfigureQueues(defaultQueue string, weights map[string]int)

type RedisDurableOption

+ func WithRedisDurableDefaultQueue(name string) RedisDurableOption

+ func WithRedisDurableQueueWeights(weights map[string]int) RedisDurableOption

type Task

+ Queue string

+ Weight int

type TaskManagerOption

+ func WithDefaultQueue(name string) TaskManagerOption

+ func WithQueueWeights(weights map[string]int) TaskManagerOption

+ type TypedDurableHandlerSpec struct

+ Fn func(ctx context.Context, payload T) (any, error)

+ Make func() T

+ type TypedDurableRegistry struct

+ func NewTypedDurableRegistry() *TypedDurableRegistry

+ func (r *TypedDurableRegistry) Add(name string, spec DurableHandlerSpec) error

+ func (r *TypedDurableRegistry) Handlers() map[string]DurableHandlerSpec

+ type TypedHandlerRegistry struct

+ func NewTypedHandlerRegistry() *TypedHandlerRegistry

+ func (r *TypedHandlerRegistry) Add(name string, spec HandlerSpec) error

+ func (r *TypedHandlerRegistry) Handlers() map[string]HandlerSpec

+ type TypedHandlerSpec struct

+ Fn func(ctx context.Context, payload T) (any, error)

+ Make func() T

Jan 31, 2026

Changes in this version

+ var ErrDurableLeaseNotFound = ewrap.New("durable lease not found")

+ type DurableBackend interface

+ Ack func(ctx context.Context, lease DurableTaskLease) error

+ Dequeue func(ctx context.Context, limit int, lease time.Duration) ([]DurableTaskLease, error)

+ Enqueue func(ctx context.Context, task DurableTask) error

+ Extend func(ctx context.Context, lease DurableTaskLease, leaseDuration time.Duration) error

+ Fail func(ctx context.Context, lease DurableTaskLease, err error) error

+ Nack func(ctx context.Context, lease DurableTaskLease, delay time.Duration) error

+ type DurableCodec interface

+ Marshal func(msg proto.Message) ([]byte, error)

+ Unmarshal func(data []byte, msg proto.Message) error

+ type DurableHandlerSpec struct

+ Fn func(ctx context.Context, payload proto.Message) (any, error)

+ Make func() proto.Message

+ type DurableTask struct

+ Handler string

+ ID uuid.UUID

+ Message proto.Message

+ Metadata map[string]string

+ Payload []byte

+ Priority int

+ Retries int

+ RetryDelay time.Duration

+ type DurableTaskLease struct

+ Attempts int

+ LeaseID string

+ MaxRetries int

+ Task DurableTask

type GRPCServer

+ func (s *GRPCServer) RegisterDurableTasks(ctx context.Context, req *workerpb.RegisterDurableTasksRequest) (*workerpb.RegisterDurableTasksResponse, error)

+ type ProtoDurableCodec struct

+ func (ProtoDurableCodec) Marshal(msg proto.Message) ([]byte, error)

+ func (ProtoDurableCodec) Unmarshal(data []byte, msg proto.Message) error

+ type RedisDurableBackend struct

+ func NewRedisDurableBackend(client rueidis.Client, opts ...RedisDurableOption) (*RedisDurableBackend, error)

+ func (b *RedisDurableBackend) Ack(ctx context.Context, lease DurableTaskLease) error

+ func (b *RedisDurableBackend) Dequeue(ctx context.Context, limit int, lease time.Duration) ([]DurableTaskLease, error)

+ func (b *RedisDurableBackend) Enqueue(ctx context.Context, task DurableTask) error

+ func (b *RedisDurableBackend) Extend(ctx context.Context, lease DurableTaskLease, leaseDuration time.Duration) error

+ func (b *RedisDurableBackend) Fail(ctx context.Context, lease DurableTaskLease, err error) error

+ func (b *RedisDurableBackend) Nack(ctx context.Context, lease DurableTaskLease, delay time.Duration) error

+ type RedisDurableOption func(*RedisDurableBackend)

+ func WithRedisDurableBatchSize(size int) RedisDurableOption

+ func WithRedisDurablePrefix(prefix string) RedisDurableOption

type TaskManager

+ func NewTaskManagerWithOptions(ctx context.Context, opts ...TaskManagerOption) *TaskManager

+ func (tm *TaskManager) RegisterDurableTask(ctx context.Context, task DurableTask) error

+ func (tm *TaskManager) RegisterDurableTasks(ctx context.Context, tasks ...DurableTask) error

+ type TaskManagerOption func(*taskManagerConfig)

+ func WithDurableBackend(backend DurableBackend) TaskManagerOption

+ func WithDurableBatchSize(size int) TaskManagerOption

+ func WithDurableCodec(codec DurableCodec) TaskManagerOption

+ func WithDurableHandlers(handlers map[string]DurableHandlerSpec) TaskManagerOption

+ func WithDurableLease(lease time.Duration) TaskManagerOption

+ func WithDurableLeaseRenewalInterval(interval time.Duration) TaskManagerOption

+ func WithDurablePollInterval(interval time.Duration) TaskManagerOption

+ func WithMaxRetries(n int) TaskManagerOption

+ func WithMaxTasks(n int) TaskManagerOption

+ func WithMaxWorkers(n int) TaskManagerOption

+ func WithRetryDelay(delay time.Duration) TaskManagerOption

+ func WithTasksPerSecond(n float64) TaskManagerOption

+ func WithTimeout(timeout time.Duration) TaskManagerOption

Jan 28, 2026

Changes in this version

+ type GRPCAuthFunc func(ctx context.Context, method string, req any) error

+ type GRPCServerOption func(*GRPCServer)

+ func WithGRPCAuth(auth GRPCAuthFunc) GRPCServerOption

+ type OTelMetricsOption func(*otelMetricsConfig)

+ func WithOTelMeterName(name string) OTelMetricsOption

+ func WithOTelMeterVersion(version string) OTelMetricsOption

+ type ResultDropPolicy uint8

+ const DropNewest

+ const DropOldest

type TaskManager

+ func (tm *TaskManager) SetMeterProvider(provider metric.MeterProvider, opts ...OTelMetricsOption) error

+ func (tm *TaskManager) SetResultsDropPolicy(policy ResultDropPolicy)

+ func (tm *TaskManager) SetTracer(tracer TaskTracer)

+ type TaskSpan interface

+ End func(err error)

+ type TaskTracer interface

+ Start func(ctx context.Context, task *Task) (context.Context, TaskSpan)

Jan 24, 2026

Changes in this version

+ type TaskHooks struct

+ OnFinish func(task *Task, status TaskStatus, result any, err error)

+ OnQueued func(task *Task)

+ OnRetry func(task *Task, delay time.Duration, attempt int)

+ OnStart func(task *Task)

type TaskManager

+ func (tm *TaskManager) SetHooks(hooks TaskHooks)

Jan 23, 2026

Changes in this version

+ const ErrMsgContextDone

type MetricsSnapshot

+ QueueDepth int

+ ResultsDropped int64

+ Retried int64

+ Running int64

+ TaskLatencyCount int64

+ TaskLatencyMax time.Duration

+ TaskLatencyTotal time.Duration

+ type RetentionPolicy struct

+ CleanupInterval time.Duration

+ MaxEntries int

+ TTL time.Duration

type Task

+ func (task *Task) CancelledAt() time.Time

+ func (task *Task) CompletedAt() time.Time

+ func (task *Task) StartedAt() time.Time

type TaskManager

+ func (tm *TaskManager) SetRetentionPolicy(policy RetentionPolicy)

+ func (tm *TaskManager) StopGraceful(ctx context.Context) error

+ func (tm *TaskManager) StopNow()

+ func (tm *TaskManager) SubscribeResults(buffer int) (<-chan Result, func())

Aug 18, 2025

Changes in this version

+ type GRPCServer struct

+ func NewGRPCServer(svc Service, handlers map[string]HandlerSpec) *GRPCServer

+ func (s *GRPCServer) CancelTask(ctx context.Context, req *workerpb.CancelTaskRequest) (*workerpb.CancelTaskResponse, error)

+ func (s *GRPCServer) GetTask(ctx context.Context, req *workerpb.GetTaskRequest) (*workerpb.GetTaskResponse, error)

+ func (s *GRPCServer) RegisterTasks(ctx context.Context, req *workerpb.RegisterTasksRequest) (*workerpb.RegisterTasksResponse, error)

+ func (s *GRPCServer) StreamResults(req *workerpb.StreamResultsRequest, ...) error

+ type HandlerSpec struct

+ Fn func(ctx context.Context, payload protoreflect.ProtoMessage) (any, error)

+ Make func() protoreflect.ProtoMessage

Aug 16, 2025

Changes in this version

+ type MetricsSnapshot struct

+ Cancelled int64

+ Completed int64

+ Failed int64

+ Scheduled int64

type TaskManager

+ ExecuteTaskErrors *ewrap.ErrorGroup

+ RegisterTaskErrors *ewrap.ErrorGroup

+ func (tm *TaskManager) GetMetrics() MetricsSnapshot

+ func (tm *TaskManager) SetMaxWorkers(maxWorkers int)

May 19, 2023

Changes in this version

type Task

+ Result atomic.Value

May 19, 2023

Changes in this version

+ const Completed

+ const Invalid

+ var ErrInvalidTaskContext = errors.New("invalid task context")

type Service

+ GetCancelledTasks func() <-chan Task

type Task

+ Execute TaskFunc

+ func NewTask(fn TaskFunc, ctx context.Context) (*Task, error)

type TaskManager

+ func (tm *TaskManager) GetCancelledTasks() <-chan Task

+ func (tm *TaskManager) IsEmpty() bool

May 17, 2023

Changes in this version

type Service

+ StreamResults func() <-chan Result

type Task

+ func (task *Task) ShouldSchedule() error

type TaskManager

+ func NewTaskManagerWithDefaults() *TaskManager

+ func (tm *TaskManager) StreamResults() <-chan Result

Mar 6, 2023

Changes in this version

+ const DefaultMaxRetries

+ const DefaultMaxTasks

+ const DefaultRetryDelay

+ const DefaultTasksPerSecond

+ const DefaultTimeout

+ const Failed

+ const Queued

+ const Running

+ var ErrTaskAlreadyStarted = errors.New("task already started")

+ var ErrTaskCancelled = errors.New("task cancelled")

+ var ErrTaskCompleted = errors.New("task completed")

+ var ErrTaskTimeout = errors.New("task timeout")

+ type Result struct

+ Error error

+ Result interface{}

+ Task *Task

+ func (r *Result) String() string

type Service

+ CancelAllAndWait func()

+ Close func()

+ GetActiveTasks func() int

+ GetCancelled func() <-chan Task

+ RegisterTasks func(ctx context.Context, tasks ...Task)

+ StartWorkers func()

+ Wait func(timeout time.Duration)

type Task

+ CancelFunc context.CancelFunc

+ Description string

+ Name string

+ Retries int

+ RetryDelay time.Duration

+ Status TaskStatus

+ func (task *Task) CancelledChan() <-chan struct{}

+ func (task *Task) ShouldExecute() error

+ func (task *Task) WaitCancelled()

type TaskManager

+ Cancelled chan Task

+ MaxRetries int

+ MaxTasks int

+ MaxWorkers int

+ RetryDelay time.Duration

+ Tasks chan Task

+ Timeout time.Duration

+ func (tm *TaskManager) CancelAllAndWait()

+ func (tm *TaskManager) Close()

+ func (tm *TaskManager) GetActiveTasks() int

+ func (tm *TaskManager) GetCancelled() <-chan Task

+ func (tm *TaskManager) RegisterTasks(ctx context.Context, tasks ...Task)

+ func (tm *TaskManager) StartWorkers()

+ func (tm *TaskManager) Wait(timeout time.Duration)

+ type TaskStatus uint8

+ func (ts TaskStatus) String() string

Jan 29, 2023

Changes in this version

+ const Cancelled

+ const ContextDeadlineReached

+ const RateLimited

+ var ErrInvalidTaskFunc = errors.New("invalid task function")

+ var ErrInvalidTaskID = errors.New("invalid task id")

+ var ErrTaskNotFound = errors.New("task not found")

+ type CancelReason uint8

type Service

+ ExecuteTask func(id uuid.UUID) (interface{}, error)

type Task

+ CancelReason CancelReason

+ Error atomic.Value

+ func (t *Task) IsValid() (err error)

+ type TaskFunc func() interface

type TaskManager

+ func (tm *TaskManager) ExecuteTask(id uuid.UUID) (interface{}, error)

Jan 28, 2023

Changes in this version

+ type Middleware func(Service) Service

+ type Service interface

+ CancelAll func()

+ CancelTask func(id uuid.UUID)

+ GetResults func() <-chan interface{}

+ GetTask func(id uuid.UUID) (task Task, ok bool)

+ GetTasks func() []Task

+ RegisterTask func(tasks ...Task)

+ Start func(numWorkers int)

+ Stop func()

+ func NewTaskManager(maxTasks int, limit float64) Service

+ func RegisterMiddleware(svc Service, mw ...Middleware) Service

+ type Task struct

+ Cancel context.CancelFunc

+ Cancelled atomic.Int64

+ Completed atomic.Int64

+ Ctx context.Context

+ Fn func() interface{}

+ ID uuid.UUID

+ Priority int

+ Started atomic.Int64

+ type TaskManager struct

+ Registry sync.Map

+ Results chan interface{}

+ func (tm *TaskManager) CancelAll()

+ func (tm *TaskManager) CancelTask(id uuid.UUID)

+ func (tm *TaskManager) GetResults() <-chan interface{}

+ func (tm *TaskManager) GetTask(id uuid.UUID) (task Task, ok bool)

+ func (tm *TaskManager) GetTasks() []Task

+ func (tm *TaskManager) RegisterTask(tasks ...Task)

+ func (tm *TaskManager) Start(numWorkers int)

+ func (tm *TaskManager) Stop()