workerpool

Jul 7, 2026

Changes in this version

+ var ErrDispatcherStopped = errors.New("workerpool: dispatcher stopped before task could run")

+ var ErrEmptyExecutorKey = errors.New("workerpool: executor key cannot be empty")

+ var ErrExecutorAlreadyRegistered = errors.New("workerpool: executor already registered")

+ var ErrExecutorNotFound = errors.New("workerpool: executor not found")

+ var ErrNilExecutor = errors.New("workerpool: executor cannot be nil")

+ var ErrNoExecutorRegistry = errors.New("workerpool: task sets executor key but no executor registry is configured")

+ var ErrPoolStopping = errors.New("workerpool: pool is stopping")

+ var ErrQueueFull = errors.New("workerpool: pool queue full")

+ var ErrTaskNilContext = errors.New("workerpool: task context is nil")

+ var ErrTaskNoExecutor = errors.New("workerpool: task must set either executor or executor key")

+ var ErrTenantNotFound = errors.New("workerpool: tenant not found")

+ var ErrTenantQueueFull = errors.New("workerpool: tenant queue full")

+ var ErrTenantShuttingDown = errors.New("workerpool: tenant is shutting down")

+ type AttemptsConfig struct

+ Count int

+ MaxDelay time.Duration

+ MinDelay time.Duration

+ type Config struct

+ GracefulTimeout time.Duration

+ RetryPolicy RetryPolicy

+ TaskQueueSize int

+ TaskTimeout time.Duration

+ TenantQueueSize int

+ TenantRefreshInterval time.Duration

+ WorkerCount int

+ func (c Config) Validate() error

+ type ExecutorRegistry struct

+ func NewExecutorRegistry() *ExecutorRegistry

+ func (r *ExecutorRegistry) Get(key string) (TaskExecutor, error)

+ func (r *ExecutorRegistry) Keys() []string

+ func (r *ExecutorRegistry) MustRegister(key string, exec TaskExecutor)

+ func (r *ExecutorRegistry) Register(key string, exec TaskExecutor) error

+ type HealthStatus struct

+ Healthy bool

+ PoolQueueCapacity int

+ PoolQueueDepth int

+ PoolWorkerCount int

+ Stopping bool

+ TenantCount int

+ Tenants []TenantHealth

+ type JitterConfig struct

+ MaxDelay time.Duration

+ MinDelay time.Duration

+ type RetryPolicy struct

+ Attempts AttemptsConfig

+ Jitter JitterConfig

+ type Task struct

+ Complete func(err error)

+ Ctx context.Context

+ Executor TaskExecutor

+ ExecutorKey string

+ TaskID uuid.UUID

+ TenantID uuid.UUID

+ type TaskExecutor interface

+ Execute func(ctx context.Context, tenantID uuid.UUID, workerID int) error

+ type Tenant interface

+ ID func() uuid.UUID

+ WorkerLimit func() int

+ type TenantHealth struct

+ QueueCapacity int

+ QueueDepth int

+ TenantID uuid.UUID

+ WorkerLimit int

+ type TenantProvider interface

+ List func(ctx context.Context) ([]Tenant, error)

+ type WorkerManager struct

+ func NewWorkerManager(p WorkerManagerParams) (*WorkerManager, error)

+ func (w *WorkerManager) GetTenantIDs() []uuid.UUID

+ func (w *WorkerManager) Health() HealthStatus

+ func (w *WorkerManager) Start() error

+ func (w *WorkerManager) Stop()

+ func (w *WorkerManager) SubmitTask(tenantID uuid.UUID, task Task) error

+ type WorkerManagerParams struct

+ Config Config

+ ExecutorRegistry *ExecutorRegistry

+ Logger *slog.Logger

+ TenantProvider TenantProvider