func (c *Consumer) executeJob(ctx context.Context, job *Job, handler HandlerFunc) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("Job %s panicked during execution: %v", job.ID, r)
		}
	}()
	return handler(ctx, *job)
}
Helpers
executeJob runs a job with the user provided handler function.
It returns an error if the handler returns an error or panics.
package curlyq
import (
	"context"
	"fmt"
	"math"
	"math/rand"
	"os"
	"os/signal"
	"sync"
	"time"
	"github.com/go-redis/redis/v7"
	"github.com/gofrs/uuid"
)
type Consumer struct {
	opts *ConsumerOpts
	client      *redis.Client
	errors      chan error
	id          string
	inflightSet string
	queue       *queue
	onAbort     *sync.Once
	processes   *sync.WaitGroup
	ackJobScript                *redis.Script
	enqueueScheduledJobsScript  *redis.Script
	getJobsScript               *redis.Script
	killJobScript               *redis.Script
	reenqueueActiveJobsScript   *redis.Script
	reenqueueOrphanedJobsScript *redis.Script
	registerConsumerScript      *redis.Script
	retryJobScript              *redis.Script
}
type ConsumerOpts struct {
	Address string
	Client *redis.Client
	Queue string
	Logger Logger
	JobMaxAttempts int
	JobMaxBackoff time.Duration
	ShutdownGracePeriod time.Duration
	CustodianConsumerTimeout time.Duration
	CustodianMaxAttempts int
	CustodianMaxBackoff time.Duration
	CustodianMaxJobs int
	CustodianPollInterval time.Duration
	HeartbeatMaxAttempts int
	HeartbeatMaxBackoff time.Duration
	HeartbeatPollInterval time.Duration
	PollerBufferSize int
	PollerMaxAttempts int
	PollerMaxBackoff time.Duration
	PollerPollDuration time.Duration
	ProcessorConcurrency int
	SchedulerMaxAttempts int
	SchedulerMaxBackoff time.Duration
	SchedulerMaxJobs int
	SchedulerPollInterval time.Duration
}
func (o *ConsumerOpts) withDefaults() *ConsumerOpts {
	opts := *o
	if opts.Logger == nil {
		opts.Logger = &DefaultLogger{}
	}
	if opts.JobMaxAttempts <= 0 {
		opts.JobMaxAttempts = 20
	}
	if opts.JobMaxBackoff <= 0 {
		opts.JobMaxBackoff = 168 * time.Hour
	}
	if opts.CustodianConsumerTimeout <= 0 {
		opts.CustodianConsumerTimeout = 1 * time.Minute
	} else if opts.CustodianConsumerTimeout < 5*time.Second {
		opts.CustodianConsumerTimeout = 5 * time.Second
	}
	if opts.CustodianMaxBackoff <= 0 {
		opts.CustodianMaxBackoff = 30 * time.Second
	}
	if opts.CustodianMaxJobs <= 0 {
		opts.CustodianMaxJobs = 50
	}
	if opts.CustodianPollInterval <= 0 {
		opts.CustodianPollInterval = 1 * time.Minute
	}
	if opts.HeartbeatMaxBackoff <= 0 {
		opts.HeartbeatMaxBackoff = 30 * time.Second
	}
	if opts.HeartbeatPollInterval <= 0 {
		opts.HeartbeatPollInterval = 1 * time.Minute
	} else if opts.HeartbeatPollInterval < 15*time.Second {
		opts.HeartbeatPollInterval = 15 * time.Second
	}
	if opts.PollerBufferSize <= 0 {
		opts.PollerBufferSize = 10
	}
	if opts.PollerMaxBackoff <= 0 {
		opts.PollerMaxBackoff = 30 * time.Second
	}
	if opts.PollerPollDuration <= 0 {
		opts.PollerPollDuration = 5 * time.Second
	} else if opts.PollerPollDuration < 1*time.Second {
		opts.PollerPollDuration = 1 * time.Second
	}
	if opts.ProcessorConcurrency <= 0 {
		opts.ProcessorConcurrency = 5
	}
	if opts.SchedulerPollInterval <= 0 {
		opts.SchedulerPollInterval = 5 * time.Second
	}
	if opts.SchedulerMaxBackoff <= 0 {
		opts.SchedulerMaxBackoff = 30 * time.Second
	}
	if opts.SchedulerMaxJobs <= 0 {
		opts.SchedulerMaxJobs = 50
	}
	return &opts
}
func NewConsumer(opts *ConsumerOpts) *Consumer {
	if opts.Address == "" && opts.Client == nil {
		panic("A redis client must be provided.")
	}
	if opts.Queue == "" {
		panic("A queue must be provided.")
	}
	var client *redis.Client
	if opts.Client != nil {
		client = opts.Client
	} else {
		client = redis.NewClient(&redis.Options{
			Addr: opts.Address,
		})
	}
	queue := newQueue(&queueOpts{
		Name: opts.Queue,
	})
	errors := make(chan error)
	id := uuid.Must(uuid.NewV4()).String()
	inflightSet := fmt.Sprintf("%s:%s", queue.inflightJobsPrefix, id)
	onAbort := &sync.Once{}
	processes := &sync.WaitGroup{}
	prepScripts()
	ackJobScript := loadLua("/lua/ack_job.lua")
	enqueueScheduledJobsScript := loadLua("/lua/enqueue_scheduled_jobs.lua")
	getJobsScript := loadLua("/lua/get_jobs.lua")
	killJobScript := loadLua("/lua/kill_job.lua")
	reenqueueActiveJobsScript := loadLua("/lua/reenqueue_active_jobs.lua")
	reenqueueOrphanedJobsScript := loadLua("/lua/reenqueue_orphaned_jobs.lua")
	registerConsumerScript := loadLua("/lua/register_consumer.lua")
	retryJobScript := loadLua("/lua/retry_job.lua")
	return &Consumer{
		opts: opts.withDefaults(),
		client: client,
		queue:  queue,
		errors:      errors,
		id:          id,
		inflightSet: inflightSet,
		onAbort:     onAbort,
		processes:   processes,
		ackJobScript:                ackJobScript,
		enqueueScheduledJobsScript:  enqueueScheduledJobsScript,
		getJobsScript:               getJobsScript,
		killJobScript:               killJobScript,
		reenqueueActiveJobsScript:   reenqueueActiveJobsScript,
		reenqueueOrphanedJobsScript: reenqueueOrphanedJobsScript,
		registerConsumerScript:      registerConsumerScript,
		retryJobScript:              retryJobScript,
	}
}
type HandlerFunc func(context.Context, Job) error
func (c *Consumer) ConsumeCtx(ctx context.Context, handler HandlerFunc) (err error) {
	c.opts.Logger.Info("Consumer started polling", "id", c.id, "queue", c.queue.name)
	defer c.opts.Logger.Info("Consumer finished polling", "id", c.id, "queue", c.queue.name)
	ctx, cancelCtx := context.WithCancel(ctx)
	defer cancelCtx()
	reqCtx, cancelReqCtx := context.WithCancel(context.Background())
	defer cancelReqCtx()
	c.client = c.client.WithContext(reqCtx)
	buffer := make(chan *Job, c.opts.PollerBufferSize)
	bufferFree := sync.NewCond(&sync.Mutex{})
	go func() {
		<-ctx.Done()
		bufferFree.Broadcast()
	}()
	err = c.registerConsumer()
	if err != nil {
		return err
	}
	c.processes.Add(5)
	go c.runHeartbeat(ctx)
	go c.runScheduler(ctx)
	go c.runCustodian(ctx)
	go c.runProcessor(ctx, buffer, bufferFree, handler)
	go c.runPoller(ctx, buffer, bufferFree)
	select {
	case <-ctx.Done():
	case err = <-c.errors:
		cancelCtx()
	}
	c.opts.Logger.Info("Consumer shutting down", "id", c.id, "queue", c.queue.name)
	done := make(chan struct{})
	go func() {
		defer close(done)
		c.processes.Wait()
	}()
	if c.opts.ShutdownGracePeriod <= 0 {
		<-done
	} else {
		select {
		case <-done:
			break
		case <-time.After(c.opts.ShutdownGracePeriod):
			c.onAbort.Do(func() {
				cancelReqCtx()
			})
			return fmt.Errorf("failed to shut down within ShutdownGracePeriod")
		}
	}
	return err
}
func (c *Consumer) Consume(handler HandlerFunc, signals ...os.Signal) error {
	if len(signals) == 0 {
		signals = []os.Signal{
			os.Interrupt,
		}
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	errChan := make(chan error)
	go func() {
		errChan <- c.ConsumeCtx(ctx, handler)
	}()
	termChan := make(chan os.Signal, 1)
	signal.Notify(termChan, signals...)
	select {
	case err := <-errChan:
		return err
	case <-termChan:
		cancel()
	}
	return <-errChan
}
func (c *Consumer) runCustodian(ctx context.Context) {
	defer c.processes.Done()
	c.opts.Logger.Debug("Custodian: process starting")
	defer c.opts.Logger.Debug("Custodian: process finished")
	ticker := time.NewTicker(c.opts.CustodianPollInterval)
	defer ticker.Stop()
	attempts := 0
	for {
		c.opts.Logger.Debug("Custodian: re-enqueueing orphaned jobs...")
		if ctx.Err() != nil {
			return
		}
		count, err := c.reenqueueOrphanedJobs()
		if err != nil {
			attempts++
			c.opts.Logger.Warn("Custodian: failed to re-enqueue jobs", "attempt", attempts, "error", err)
			c.backoff(ctx, "Custodian", attempts, c.opts.CustodianMaxBackoff, c.opts.CustodianMaxAttempts)
			continue
		} else {
			attempts = 0
			c.opts.Logger.Debug("Custodian: successfully re-enqueued jobs", "job_count", count)
			if count == c.opts.CustodianMaxJobs {
				continue
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			continue
		}
	}
}
func (c *Consumer) runHeartbeat(ctx context.Context) {
	defer c.processes.Done()
	c.opts.Logger.Debug("Heartbeat: process starting")
	defer c.opts.Logger.Debug("Heartbeat: process finished")
	ticker := time.NewTicker(c.opts.HeartbeatPollInterval)
	defer ticker.Stop()
	attempts := 0
	for {
		if ctx.Err() != nil {
			return
		}
		c.opts.Logger.Debug("Heartbeat: updating consumer...")
		err := c.registerConsumer()
		if err != nil {
			attempts++
			c.opts.Logger.Warn("Heartbeat: failed to update", "attempt", attempts, "error", err)
			c.backoff(ctx, "Heartbeat", attempts, c.opts.HeartbeatMaxBackoff, c.opts.HeartbeatMaxAttempts)
			continue
		} else {
			attempts = 0
			c.opts.Logger.Debug("Heartbeat: update successful")
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			continue
		}
	}
}
func (c *Consumer) runPoller(ctx context.Context, buffer chan *Job, bufferFree *sync.Cond) {
	defer c.processes.Done()
	c.opts.Logger.Debug("Poller: process started")
	defer c.opts.Logger.Debug("Poller: process finished")
	defer func() {
		close(buffer)
		jobs := []*Job{}
		for job := range buffer {
			jobs = append(jobs, job)
		}
		if len(jobs) > 0 {
			c.reenqueueActiveJobs(jobs)
		}
	}()
	pollAttempts := 0
	retrieveAttempts := 0
	for {
		bufferFree.L.Lock()
		for len(buffer) >= cap(buffer) && ctx.Err() == nil {
			bufferFree.Wait()
		}
		bufferFree.L.Unlock()
		if ctx.Err() != nil {
			return
		}
		c.opts.Logger.Debug("Poller: polling for new jobs...")
		err := c.pollActiveJobs()
		if err == redis.Nil {
			pollAttempts = 0
			c.opts.Logger.Debug("Poller: no new jobs detected")
			continue
		} else if err != nil {
			pollAttempts++
			c.opts.Logger.Warn("Poller: error polling jobs", "attempt", pollAttempts, "error", err)
			c.backoff(ctx, "Poller", pollAttempts, c.opts.PollerMaxBackoff, c.opts.PollerMaxAttempts)
			continue
		} else {
			pollAttempts = 0
			c.opts.Logger.Debug("Poller: detected new jobs")
		}
		count := cap(buffer) - len(buffer)
		c.opts.Logger.Debug("Poller: retrieving jobs...", "job_count", count)
		jobs, err := c.getJobs(count)
		if err != nil {
			retrieveAttempts++
			c.opts.Logger.Warn("Poller: error retrieving jobs", "attempt", retrieveAttempts, "error", err)
			c.backoff(ctx, "Poller", retrieveAttempts, c.opts.PollerMaxBackoff, c.opts.PollerMaxAttempts)
		} else {
			retrieveAttempts = 0
			c.opts.Logger.Debug("Poller: successfully retrieved jobs", "job_count", len(jobs))
			for _, job := range jobs {
				buffer <- job
			}
		}
	}
}
func (c *Consumer) runProcessor(ctx context.Context, buffer chan *Job, bufferFree *sync.Cond, handler HandlerFunc) {
	defer c.processes.Done()
	c.opts.Logger.Debug("Processor: process started")
	defer c.opts.Logger.Debug("Processor: process finished")
	tokens := make(chan struct{}, c.opts.ProcessorConcurrency)
	for i := 0; i < c.opts.ProcessorConcurrency; i++ {
		tokens <- struct{}{}
	}
	defer func() {
		for i := 0; i < c.opts.ProcessorConcurrency; i++ {
			<-tokens
		}
		close(tokens)
	}()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tokens:
		}
		select {
		case <-ctx.Done():
			tokens <- struct{}{}
			return
		case job, open := <-buffer:
			if !open {
				tokens <- struct{}{}
				return
			}
			bufferFree.L.Lock()
			bufferFree.Broadcast()
			bufferFree.L.Unlock()
			go func() {
				c.opts.Logger.Debug("Processing job", "id", job.ID)
				err := c.executeJob(ctx, job, handler)
				if err != nil {
					c.opts.Logger.Debug("Job failed", "id", job.ID)
					if job.Attempt < c.opts.JobMaxAttempts {
						c.opts.Logger.Debug("Retrying job", "id", job.ID, "retries", job.Attempt)
						_, err := c.retryJob(job)
						if err != nil {
							c.abort(ErrFailedToRetryJob{
								Job: *job,
								Err: err,
							})
						}
					} else {
						c.opts.Logger.Debug("Killing job", "id", job.ID, "retries", job.Attempt)
						_, err := c.killJob(job)
						if err != nil {
							c.abort(ErrFailedToKillJob{
								Job: *job,
								Err: err,
							})
						}
					}
				} else {
					c.opts.Logger.Debug("Job successful", "id", job.ID)
					_, err := c.ackJob(job)
					if err != nil {
						c.abort(ErrFailedToAckJob{
							Job: *job,
							Err: err,
						})
					}
				}
				tokens <- struct{}{}
			}()
		}
	}
}
func (c *Consumer) runScheduler(ctx context.Context) {
	defer c.processes.Done()
	c.opts.Logger.Debug("Scheduler: process starting")
	defer c.opts.Logger.Debug("Scheduler: process finished")
	ticker := time.NewTicker(c.opts.SchedulerPollInterval)
	defer ticker.Stop()
	attempts := 0
	for {
		c.opts.Logger.Debug("Scheduler: enqueueing jobs...")
		if ctx.Err() != nil {
			return
		}
		count, err := c.enqueueScheduledJobs()
		if err != nil {
			attempts++
			c.opts.Logger.Warn("Scheduler: failed to enqueue jobs", "attempt", attempts, "error", err)
			c.backoff(ctx, "Scheduler", attempts, c.opts.SchedulerMaxBackoff, c.opts.SchedulerMaxAttempts)
			continue
		} else {
			c.opts.Logger.Debug("Scheduler: jobs enqueued successfully", "job_count", count)
			attempts = 0
			if count == c.opts.SchedulerMaxJobs {
				continue
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			continue
		}
	}
}
func (c *Consumer) ackJob(job *Job) (bool, error) {
	keys := []string{
		c.inflightSet,
		c.queue.jobDataHash,
	}
	args := []interface{}{
		job.ID,
	}
	return c.ackJobScript.Run(c.client, keys, args...).Bool()
}
func (c *Consumer) enqueueScheduledJobs() (int, error) {
	keys := []string{
		c.queue.scheduledJobsSet,
		c.queue.activeJobsList,
		c.queue.signalList,
	}
	args := []interface{}{
		float64(time.Now().Unix()),
		c.opts.SchedulerMaxJobs,
	}
	return c.enqueueScheduledJobsScript.Run(c.client, keys, args...).Int()
}
func (c *Consumer) getJobs(count int) ([]*Job, error) {
	keys := []string{
		c.queue.consumersSet,
		c.queue.activeJobsList,
		c.inflightSet,
		c.queue.jobDataHash,
		c.queue.signalList,
	}
	args := []interface{}{
		count,
		c.inflightSet,
	}
	result, err := c.getJobsScript.Run(c.client, keys, args...).Result()
	if err != nil {
		return nil, err
	}
	messages := result.([]interface{})
	jobs := make([]*Job, len(messages))
	for idx, message := range messages {
		job := &Job{}
		err := job.fromMessage([]byte(message.(string)))
		if err != nil {
			return nil, err
		}
		jobs[idx] = job
	}
	return jobs, nil
}
func (c *Consumer) killJob(job *Job) (bool, error) {
	diedAt := float64(time.Now().Unix())
	job.Attempt = job.Attempt + 1
	msg, err := job.message()
	if err != nil {
		return false, err
	}
	keys := []string{
		c.inflightSet,
		c.queue.deadJobsSet,
		c.queue.jobDataHash,
	}
	args := []interface{}{
		job.ID,
		diedAt,
		msg,
	}
	return c.killJobScript.Run(c.client, keys, args...).Bool()
}
func (c *Consumer) pollActiveJobs() error {
	return c.client.BRPopLPush(
		c.queue.signalList,
		c.queue.signalList,
		c.opts.PollerPollDuration,
	).Err()
}
func (c *Consumer) reenqueueActiveJobs(jobs []*Job) error {
	keys := []string{
		c.inflightSet,
		c.queue.activeJobsList,
		c.queue.signalList,
	}
	args := make([]interface{}, len(jobs))
	for idx, job := range jobs {
		args[idx] = job.ID
	}
	return c.reenqueueActiveJobsScript.Run(c.client, keys, args...).Err()
}
func (c *Consumer) reenqueueOrphanedJobs() (int, error) {
	keys := []string{
		c.queue.consumersSet,
		c.queue.activeJobsList,
		c.queue.signalList,
	}
	expiredBefore := time.Now().Add(-c.opts.HeartbeatPollInterval).Add(-c.opts.CustodianConsumerTimeout)
	args := []interface{}{
		float64(expiredBefore.Unix()),
		c.opts.CustodianMaxJobs,
	}
	return c.reenqueueOrphanedJobsScript.Run(c.client, keys, args...).Int()
}
func (c *Consumer) registerConsumer() error {
	keys := []string{
		c.queue.consumersSet,
	}
	args := []interface{}{
		float64(time.Now().Unix()),
		c.inflightSet,
	}
	return c.registerConsumerScript.Run(c.client, keys, args...).Err()
}
func (c *Consumer) retryJob(job *Job) (bool, error) {
	backoff := expBackoff(job.Attempt, c.opts.JobMaxBackoff)
	retryAt := float64(time.Now().Add(time.Duration(backoff)).Unix())
	job.Attempt = job.Attempt + 1
	msg, err := job.message()
	if err != nil {
		return false, err
	}
	keys := []string{
		c.inflightSet,
		c.queue.scheduledJobsSet,
		c.queue.jobDataHash,
	}
	args := []interface{}{
		job.ID,
		retryAt,
		msg,
	}
	return c.retryJobScript.Run(c.client, keys, args...).Bool()
}