Versions in this module Expand all Collapse all v1 v1.1.0 Apr 4, 2026 Changes in this version + var ErrDuplicateJobID = errors.New("duplicate job id") + var ErrDuplicateKey = errors.New("duplicate idempotency key") + var ErrHandlerNotFound = errors.New("handler not found") + var ErrInvalidJobState = errors.New("invalid job state") + var ErrJobNotFound = errors.New("job not found") + var ErrLeaseConflict = errors.New("lease conflict") + var ErrStoreRequired = errors.New("store is required") + type EnqueueOptions struct + ID string + IdempotencyKey string + MaxRetries int + Metadata map[string]string + Payload []byte + Queue string + RunAt time.Time + Timeout time.Duration + Type string + func (o EnqueueOptions) Validate(now time.Time) error + type Handler func(context.Context, *Job) error + type Job struct + Attempts int + CreatedAt time.Time + ID string + IdempotencyKey string + LastError string + LastFinishedAt time.Time + LeaseExpiresAt time.Time + LeaseOwner string + MaxRetries int + Metadata map[string]string + Partition int + Payload json.RawMessage + Queue string + QueueVersion int + RunAt time.Time + State JobState + Timeout time.Duration + Type string + UpdatedAt time.Time + type JobState string + const JobStateDead + const JobStateLeased + const JobStateReady + const JobStateRetryWait + const JobStateScheduled + const JobStateSucceeded + type Lease struct + Job *Job + LeaseExpiresAt time.Time + LeaseID string + LeaseOwner string + type LeaseCompletion struct + Dead bool + Failure string + FinishedAt time.Time + JobID string + LeaseID string + RetryAt time.Time + type LeaseRenewal struct + JobID string + LeaseID string + type MemoryStore struct + func NewMemoryStore() *MemoryStore + func (s *MemoryStore) Ack(_ context.Context, jobID, leaseID string, now time.Time) error + func (s *MemoryStore) CompleteLeases(_ context.Context, completions []LeaseCompletion) error + func (s *MemoryStore) Enqueue(_ context.Context, job *Job) error + func (s *MemoryStore) Fail(_ context.Context, jobID, leaseID string, now, retryAt time.Time, ...) error + func (s *MemoryStore) LeaseJobs(_ context.Context, queues []string, workerID string, now time.Time, limit int, ...) ([]*Lease, error) + func (s *MemoryStore) PromoteDueJobs(_ context.Context, queues []string, now time.Time, limit int) (int, error) + func (s *MemoryStore) RecoverExpiredLeases(_ context.Context, queues []string, now time.Time, limit int) (int, error) + func (s *MemoryStore) RenewLease(_ context.Context, jobID, leaseID string, now time.Time, ...) error + func (s *MemoryStore) RenewLeases(_ context.Context, renewals []LeaseRenewal, now time.Time, ...) error + func (s *MemoryStore) Stats(_ context.Context, queues []string) ([]Stats, error) + type Metrics interface + RecordEnqueue func(context.Context, *Job) + RecordFailure func(context.Context, *Job, error, time.Duration) + RecordLease func(context.Context, *Job) + RecordRetry func(context.Context, *Job, time.Duration) + RecordSuccess func(context.Context, *Job, time.Duration) + type NopMetrics struct + func (NopMetrics) RecordEnqueue(context.Context, *Job) + func (NopMetrics) RecordFailure(context.Context, *Job, error, time.Duration) + func (NopMetrics) RecordLease(context.Context, *Job) + func (NopMetrics) RecordRetry(context.Context, *Job, time.Duration) + func (NopMetrics) RecordSuccess(context.Context, *Job, time.Duration) + type Producer struct + func NewProducer(store Store, config ProducerConfig) (*Producer, error) + func (p *Producer) Enqueue(ctx context.Context, opts EnqueueOptions) (*Job, error) + type ProducerConfig struct + DefaultMaxRetries int + DefaultQueue string + DefaultRetryPolicy RetryPolicy + DefaultTimeout time.Duration + func DefaultProducerConfig() ProducerConfig + func (c ProducerConfig) Validate() error + type QueueTopology struct + Active QueueVersionConfig + Draining []QueueVersionConfig + func (q QueueTopology) DrainingVersionNumbers() []int + type QueueVersionConfig struct + Partitions int + Status string + Version int + type RedisConfig struct + ActiveVersion int + DrainingVersion []QueueVersionConfig + Partitions int + Prefix string + type RedisStore struct + func NewRedisStore(client *redis.Client, prefix string) *RedisStore + func NewRedisStoreWithConfig(client *redis.Client, cfg RedisConfig) *RedisStore + func (s *RedisStore) Ack(ctx context.Context, jobID, leaseID string, now time.Time) error + func (s *RedisStore) CompleteDrainingVersion(ctx context.Context, queue string, version int) (QueueTopology, error) + func (s *RedisStore) CompleteLeases(ctx context.Context, completions []LeaseCompletion) error + func (s *RedisStore) ConfigureQueue(ctx context.Context, queue string, active QueueVersionConfig, ...) error + func (s *RedisStore) Enqueue(ctx context.Context, job *Job) error + func (s *RedisStore) Fail(ctx context.Context, jobID, leaseID string, now, retryAt time.Time, ...) error + func (s *RedisStore) LeaseJobs(ctx context.Context, queues []string, workerID string, now time.Time, ...) ([]*Lease, error) + func (s *RedisStore) PromoteDueJobs(ctx context.Context, queues []string, now time.Time, limit int) (int, error) + func (s *RedisStore) PromoteQueueVersion(ctx context.Context, queue string, nextVersion QueueVersionConfig) (QueueTopology, error) + func (s *RedisStore) QueueTopology(ctx context.Context, queue string) (QueueTopology, error) + func (s *RedisStore) RecoverExpiredLeases(ctx context.Context, queues []string, now time.Time, limit int) (int, error) + func (s *RedisStore) RenewLease(ctx context.Context, jobID, leaseID string, now time.Time, ...) error + func (s *RedisStore) RenewLeases(ctx context.Context, renewals []LeaseRenewal, now time.Time, ...) error + func (s *RedisStore) Stats(ctx context.Context, queues []string) ([]Stats, error) + type RetryPolicy struct + InitialBackoff time.Duration + MaxBackoff time.Duration + MaxRetries int + Multiplier float64 + func (p RetryPolicy) Backoff(attempt int) time.Duration + func (p RetryPolicy) Validate() error + type Stats struct + Dead int64 + Leased int64 + Queue string + Ready int64 + RetryWait int64 + Scheduled int64 + Succeeded int64 + type Store interface + Ack func(context.Context, string, string, time.Time) error + CompleteLeases func(context.Context, []LeaseCompletion) error + Enqueue func(context.Context, *Job) error + Fail func(context.Context, string, string, time.Time, time.Time, string, bool) error + LeaseJobs func(context.Context, []string, string, time.Time, int, time.Duration) ([]*Lease, error) + PromoteDueJobs func(context.Context, []string, time.Time, int) (int, error) + RecoverExpiredLeases func(context.Context, []string, time.Time, int) (int, error) + RenewLease func(context.Context, string, string, time.Time, time.Duration) error + RenewLeases func(context.Context, []LeaseRenewal, time.Time, time.Duration) error + Stats func(context.Context, []string) ([]Stats, error) + type Worker struct + func NewWorker(store Store, config WorkerConfig, logger *slog.Logger, metrics Metrics) (*Worker, error) + func (w *Worker) Handle(jobType string, handler Handler) + func (w *Worker) Start(ctx context.Context) error + type WorkerConfig struct + BatchSize int + Concurrency int + HeartbeatInterval time.Duration + LeaseDuration time.Duration + MaxInFlight int + PollInterval time.Duration + QueueBuffer int + Queues []string + RetryPolicy RetryPolicy + WorkerID string + func DefaultWorkerConfig(workerID string) WorkerConfig + func (c WorkerConfig) Validate() error v1.0.0 Apr 3, 2026
Go to main page