tasker

package module

Go to main page

Versions in this module

v0

v0.3.3

Nov 2, 2026

v0.3.2

Nov 2, 2026

v0.3.1

Nov 2, 2026

Changes in this version

+ const NodeLivenessWindow

v0.3.0

Nov 2, 2026

v0.2.1

Nov 1, 2026

Changes in this version

+ func ContextWithRow(ctx context.Context, row *JobRow) context.Context

type JobRow

+ func RowFromContext(ctx context.Context) (*JobRow, bool)

v0.2.0

Nov 1, 2026

Changes in this version

+ func AddDispatchMiddleware(mw DispatchMiddleware)

+ func SetDispatchMiddlewares(mws []DispatchMiddleware)

+ type DispatchMiddleware func(ctx context.Context, rows []*JobRow, next func(ctx context.Context) error) error

+ func GetDispatchMiddlewares() []DispatchMiddleware

v0.1.6

Oct 15, 2026

v0.1.5

Oct 15, 2026

v0.1.4

Sep 26, 2026

v0.1.3

Sep 24, 2026

Changes in this version

type ManagerOption

+ func WithDefaultTimeout(timeout time.Duration) ManagerOption

v0.1.2

Aug 31, 2026

Changes in this version

+ var ErrNodeNotFound = errors.New("node not found")

type Driver

+ ScheduleRetry func(ctx context.Context, id JobID, err error, scheduledAt time.Time) error

v0.1.1

Jul 9, 2026

v0.1.0

Jul 9, 2026

Changes in this version

+ var ErrInvalidPayload = errors.New("invalid job payload")

+ var ErrInvalidTransition = errors.New("invalid state transition")

+ var ErrJobAlreadyExists = errors.New("job already exists")

+ var ErrJobNotClaimed = errors.New("job not claimed by this worker")

+ var ErrJobNotFound = errors.New("job not found")

+ var ErrJobPanicked = errors.New("job execution panicked")

+ var ErrJobTimedOut = errors.New("job execution timed out")

+ var ErrMaxRetriesExceeded = errors.New("max retries exceeded")

+ var ErrQueueNotFound = errors.New("queue not found")

+ var ErrWorkerPoolStopped = errors.New("worker pool is stopped")

+ func AddGlobalMiddleware(mw JobMiddleware)

+ func ListRegisteredJobs() []string

+ func RegisterJob(name string, factory JobFactory)

+ func SetGlobal(m *Manager)

+ func SetGlobalMiddlewares(mws []JobMiddleware)

+ func ShouldRequeue(nodeTimeout time.Duration) bool

+ func ValidateTransition(from, to State) error

+ type AfterHandleHook interface

+ AfterHandle func(ctx context.Context) error

+ type AttemptError struct

+ Attempt int

+ Error string

+ Stack string

+ Timestamp time.Time

+ type BackoffConfig struct

+ Base time.Duration

+ MaxDelay time.Duration

+ Strategy BackoffStrategy

+ func DefaultBackoff() BackoffConfig

+ func FibonacciBackoff() BackoffConfig

+ type BackoffStrategy string

+ const BackoffExponential

+ const BackoffFibonacci

+ const BackoffFixed

+ type BeforeHandleHook interface

+ BeforeHandle func(ctx context.Context) error

+ type BeforeRetryHook interface

+ BeforeRetry func(ctx context.Context, attempt int, err error) error

+ type Config struct

+ DefaultMaxAttempts int

+ DefaultQueue QueueName

+ Driver Driver

+ NodeID NodeID

+ func DefaultAppConfig() Config

+ type DispatchOpt func(*dispatchOptions)

+ func OnQueue(name QueueName) DispatchOpt

+ func WithBatchID(id string) DispatchOpt

+ func WithDelay(d time.Duration) DispatchOpt

+ func WithMetadata(k, v string) DispatchOpt

+ func WithPriority(p int) DispatchOpt

+ func WithTags(tags ...string) DispatchOpt

+ type Driver interface

+ AcquireLock func(ctx context.Context, key string, nodeID NodeID, ttl time.Duration) (bool, error)

+ Cancel func(ctx context.Context, id JobID) (*JobRow, error)

+ CancelBatch func(ctx context.Context, ids []JobID) error

+ Claim func(ctx context.Context, queue QueueName, nodeID NodeID, max int) ([]*JobRow, error)

+ Close func() error

+ Complete func(ctx context.Context, id JobID, output []byte) error

+ DeregisterNode func(ctx context.Context, nodeID NodeID) error

+ Enqueue func(ctx context.Context, job *JobRow) error

+ EnqueueBatch func(ctx context.Context, jobs []*JobRow) error

+ Fail func(ctx context.Context, id JobID, err error) error

+ GetByID func(ctx context.Context, id JobID) (*JobRow, error)

+ GlobalStats func(ctx context.Context) (*GlobalStats, error)

+ Heartbeat func(ctx context.Context, nodeID NodeID, ttl time.Duration) error

+ IsLeader func(ctx context.Context, nodeID NodeID) (bool, error)

+ JobStats func(ctx context.Context, kind string) (*JobTypeStats, error)

+ LeaderElection func(ctx context.Context, nodeID NodeID, ttl time.Duration) (bool, error)

+ ListNodes func(ctx context.Context) ([]NodeInfo, error)

+ Ping func(ctx context.Context) error

+ Prune func(ctx context.Context, before time.Time, states []State) (int64, error)

+ QueryJobs func(ctx context.Context, filter JobFilter) ([]*JobRow, int64, error)

+ QueueStats func(ctx context.Context, queue QueueName) (*QueueStats, error)

+ RegisterNode func(ctx context.Context, node NodeInfo, ttl time.Duration) error

+ ReleaseLock func(ctx context.Context, key string, nodeID NodeID) error

+ RequeueStale func(ctx context.Context, timeout time.Duration) (int64, error)

+ ResignLeadership func(ctx context.Context, nodeID NodeID) error

+ Retry func(ctx context.Context, id JobID) (*JobRow, error)

+ RetryBatch func(ctx context.Context, ids []JobID) error

+ type GlobalStats struct

+ FailedJobs int64

+ JobsPerMinute int64

+ PausedMasters int

+ Processes int

+ RecentJobs int64

+ Status string

+ type Job interface

+ Handle func(ctx context.Context) error

+ type JobFactory func() Job

+ func GetRegisteredJob(name string) (JobFactory, bool)

+ type JobFilter struct

+ Kinds []string

+ Limit int

+ Offset int

+ Order string

+ OrderBy string

+ Queues []QueueName

+ Search string

+ States []State

+ Tags []string

+ type JobID uint64

+ type JobMiddleware func(ctx context.Context, job Job, payload []byte, ...) error

+ func GetGlobalMiddlewares() []JobMiddleware

+ func LoggingMiddleware() JobMiddleware

+ func RecoveryMiddleware() JobMiddleware

+ type JobRow struct

+ Attempt int

+ AttemptedAt *time.Time

+ AttemptedBy []NodeID

+ BatchID string

+ CompletedAt *time.Time

+ CreatedAt time.Time

+ Errors []AttemptError

+ FinalizedAt *time.Time

+ ID JobID

+ Kind string

+ MaxAttempts int

+ Metadata map[string]string

+ NodeID NodeID

+ Output []byte

+ Payload []byte

+ Priority int

+ Queue QueueName

+ ScheduledAt time.Time

+ StartedAt *time.Time

+ State State

+ Tags []string

+ Timeout time.Duration

+ UUID string

+ UniqueKey string

+ type JobTypeStats struct

+ AvgRuntimeMs float64

+ FailedCount int64

+ Kind string

+ Samples int64

+ Throughput float64

+ TotalCount int64

+ type Manager struct

+ func Global() *Manager

+ func NewConfiguredManager(cfg Config) *Manager

+ func NewManager(opts ...ManagerOption) *Manager

+ func (m *Manager) Cancel(ctx context.Context, id JobID) (*JobRow, error)

+ func (m *Manager) CancelBatch(ctx context.Context, ids []JobID) error

+ func (m *Manager) Chain(ctx context.Context, jobs []Job, opts ...DispatchOpt) ([]*JobRow, error)

+ func (m *Manager) Config() ManagerConfig

+ func (m *Manager) Dispatch(ctx context.Context, job Job, opts ...DispatchOpt) (*JobRow, error)

+ func (m *Manager) DispatchBatch(ctx context.Context, jobs []Job, opts ...DispatchOpt) ([]*JobRow, error)

+ func (m *Manager) Driver() Driver

+ func (m *Manager) GetJob(ctx context.Context, id JobID) (*JobRow, error)

+ func (m *Manager) GlobalStats(ctx context.Context) (*GlobalStats, error)

+ func (m *Manager) JobStats(ctx context.Context, kind string) (*JobTypeStats, error)

+ func (m *Manager) QueryJobs(ctx context.Context, filter JobFilter) ([]*JobRow, int64, error)

+ func (m *Manager) QueueStats(ctx context.Context, queue QueueName) (*QueueStats, error)

+ func (m *Manager) Retry(ctx context.Context, id JobID) (*JobRow, error)

+ func (m *Manager) RetryBatch(ctx context.Context, ids []JobID) error

+ func (m *Manager) Stop(ctx context.Context) error

+ func (m *Manager) Work(ctx context.Context, queue QueueName, concurrency int) error

+ type ManagerConfig struct

+ DefaultMaxAttempts int

+ DefaultQueue QueueName

+ DefaultTimeout time.Duration

+ NodeID NodeID

+ func DefaultConfig() ManagerConfig

+ type ManagerOption func(*Manager)

+ func WithDefaultQueue(name QueueName) ManagerOption

+ func WithDriver(d Driver) ManagerOption

+ func WithMaxAttempts(n int) ManagerOption

+ func WithNodeID(id NodeID) ManagerOption

+ type NodeID string

+ type NodeInfo struct

+ Host string

+ ID NodeID

+ LastHeartbeat time.Time

+ Port int

+ Queues []QueueName

+ StartedAt time.Time

+ Status string

+ Version string

+ Workers int

+ type Pool struct

+ func NewPool(mgr *Manager, queue QueueName, maxWorkers int) *Pool

+ func (p *Pool) Queue() QueueName

+ func (p *Pool) Scale(target int) error

+ func (p *Pool) Start(ctx context.Context) error

+ func (p *Pool) Stop(ctx context.Context) error

+ func (p *Pool) WorkerCount() int

+ type PoolManager struct

+ func NewPoolManager(mgr *Manager) *PoolManager

+ func (pm *PoolManager) Pools() map[QueueName]*Pool

+ func (pm *PoolManager) ScalePool(queue QueueName, target int) error

+ func (pm *PoolManager) StartPool(ctx context.Context, queue QueueName, concurrency int) (*Pool, error)

+ func (pm *PoolManager) StopAll(ctx context.Context) error

+ func (pm *PoolManager) StopPool(ctx context.Context, queue QueueName) error

+ func (pm *PoolManager) WorkerCount() int

+ type QueueName string

+ type QueueStats struct

+ Available int64

+ AvgRuntimeMs float64

+ Completed int64

+ Failed int64

+ Queue QueueName

+ Retryable int64

+ Running int64

+ Scheduled int64

+ ThroughputPerMin float64

+ WaitTimeMs float64

+ type ShouldBatch interface

+ BatchID func() string

+ type ShouldDelay interface

+ Delay func() time.Duration

+ type ShouldFail interface

+ Failed func(ctx context.Context, payload []byte, err error) error

+ type ShouldLock interface

+ LockKey func() string

+ LockTimeout func() time.Duration

+ type ShouldMiddleware interface

+ Middleware func() []JobMiddleware

+ type ShouldQueue interface

+ Queue func() QueueName

+ type ShouldRetry interface

+ MaxAttempts func() int

+ type ShouldRetryUntil interface

+ RetryUntil func() time.Time

+ type ShouldRetryWithBackoff interface

+ RetryBackoff func() BackoffConfig

+ type ShouldTag interface

+ Tags func() []string

+ type ShouldTimeout interface

+ Timeout func() time.Duration

+ type State string

+ const StateAvailable

+ const StateCancelled

+ const StateCompleted

+ const StateFailed

+ const StatePending

+ const StateRetryable

+ const StateRunning

+ const StateScheduled

+ func ComputeNextState(job *JobRow, err error) (State, time.Duration)

+ func (s State) Terminal() bool

+ func (s State) Valid() bool

+ type Transition struct

+ From State

+ To State

+ func (t Transition) Valid() bool

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL