Is your feature request related to a problem? Please describe.

• I want to add tracing to the underlying redis client instance using dd-trace-go.
• Is it possible to get access to the underlying redis instance so I can wrap the client with a function like the one below?

import (
	"github.com/go-redis/redis/v7"
	ddRedis "gopkg.in/DataDog/dd-trace-go.v1/contrib/go-redis/redis.v7"
)

func TraceClient(rClient redis.UniversalClient) {
	ddRedis.WrapClient(
		rClient,
		ddRedis.WithServiceName(config.ServiceName()+"-redis"),
		ddRedis.WithAnalytics(config.Enabled()),
	)
}

Describe alternatives you've considered

• I am also looking if the way to go about this is creating a RedisConnOpt interface that returns a wrapped redis instance with datadog tracing. What do you think?

https://pkg.go.dev/gopkg.in/DataDog/[email protected]/contrib/go-redis/redis.v7#WrapClient

Describe the bug When the task is active, the worker shutdown, then the task will always be active.

To Reproduce Steps to reproduce the behavior:

1. Set task sleep times.

func HandleWelcomeEmailTask(ctx context.Context, t *asynq.Task) error {
	id, err := t.Payload.GetInt("user_id")
	if err != nil {
		return err
	}
	fmt.Printf("[%s] - [Start] Send Welcome Email to User %d\n", time.Now().String(), id)
	time.Sleep(10*time.Second)
	fmt.Printf("[%s] - [ End ] Send Welcome Email to User %d\n", time.Now().String(), id)
	return nil
}

2. Start the worker.
3. Start the scheduler.
4. When the task is active, shutdown the worker.
5. The task will always be active.

Expected behavior I think the task should change the state to failed based on Timeout, MaxRetry, etc.

Screenshots If applicable, add screenshots to help explain your problem.

Environment (please complete the following information):

• OS: Windows
• Version of asynq package: v0.17.2

https://github.com/hibiken/asynq/blob/master/background.go#L253

normally, packages has two methods, Run and Stop, so i can cleanup other resources before or after Stop(), another scene is that user will run multipe backgroud server in main(), like kafka consumer

Is your feature request related to a problem? Please describe. Asynq currently removes the successfully processed jobs from Redis. That means the information about the processed jobs also vanishes like;

• When did the client pushed the job into the queue
• When is the job completed.
• Is a given job id processed successfully.

Describe the solution you'd like It would be nice to keep processed jobs until a given TTL and return a basic struct that contains something like id, queued_at, completed_at. I think to achieve that, the first thing is to generate a unique id for each task and keep them like asynq:{default}:processed:{job_id}

Describe alternatives you've considered Python RQ or Laravel Horizon might be good examples to take a look at.

Asking because i cannot find this in the wiki and others might find it useful.

Specifically, How long do archived tasks stay in Redis so we can keep track or retry manually?

We could offer a cmd for serving a comprehensive Web UI for management and monitoring

Describe the solution you'd like maybe we can add a sub cmd in asynq. eg. asynq webui.

eg. https://raw.githubusercontent.com/mperham/sidekiq/master/examples/web-ui.png

Is your feature request related to a problem? Please describe. I'm always frustrated when I need to call GetX method for all the values in Payload to get all the data (also need to make sure that I don't make a typo in payload key string). It'd be nice if asynq supported en/decoding protobuf or gob message directly so that I can load that data to an object with one method call.

Describe the solution you'd like initial proposal:

type EmailTaskPayload struct {
    UserID int
}

// client side.
func main() {
     c := asynq.NewClient(asynq.RedisClientOpt{Addr: ":6379"})

    t := asynq.NewTask("email:welcome", EmailTaskPayload{UserID: 42})
    err := c.Enqueue(t)
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
}

// background workers side
func main() {
    bg := asynq.NewBackground(r, &asynq.Config{
        Concurrency: 10,
    })

    mux := asynq.NewMux()
    mux.HandleFunc("email:welcome", emailHandler)

    bg.Run(mux)
}

func emailHandler(ctx context.Context, t *asynq.Task) error {
    var p EmailTaskPayload
    err := t.ReadPayload(&p)
    if err != nil {
        return err
    }
    fmt.Printf("Send Email to User %d\n", p.UserID)
    return nil
}

Describe alternatives you've considered None for now.

Additional context None for now.

Describe the bug The high priority queue is processed very slow compare to the lower queue

To Reproduce Steps to reproduce the behavior (Code snippets if applicable):

• Run a long-running task on the default queue. Many small but critical tasks will issue from here and added to the critical queue
• The default queue is hold all the time of worker pools
• The critical queue is stuck, and no task is process

Expected behavior

• The critical queue's task need to be selected to process first and clean out quickly

Screenshots

• Setup server with 3 queue

• Enqueue critical task like that

• The critical queue stuck

• The default queue is running on all other workers

Environment (please complete the following information):

• OS: Ubuntu 18.4
• Version of asynq package 0.18.3

Hi @hibiken,

First of all, thank you for writing lightest message queue in Go.

I'm writing a messaging app. When the QR is read, I start to process messages. I will use asynq for processing history messages and normal message. I guess I have to create an asynq server for each number (instance)?

Message Handler Case:

• Number Instance -> PhoneNumber (Start asynq server when read qr)

• QR Read -> Handle All History Messages(avarage 2-3 million message) -> Start worker queue (instance_id_history).

• Normal Message (Schedule Message) -> Start worker queue (instance_id_normal)

Should I always listen to 2 queues? Or I can stop listening to this queue when the HistoryMessages is finished, right?

I'll be listening to 2000 queue when there are 1000 number(instances). Do you think a redis cluster(sentinel) is necessary to handle this traffic?

What kind of path do you think we should follow with asynq?

Sincerely :)

Instead of (queue, type, payload) use any custom key. This allows to enforce unique tasks not limited to same payload and/or type.

Tested with --redis_cluster flag.

Is your feature request related to a problem? Please describe. N/A

Describe the solution you'd like Is it possible to set concurrency individually for every queue like it is supported here.

Describe alternatives you've considered Currently I tried to wire up the server in a way where I calculate total concurrency and set max priority and assuming that the combination will provide some sort of similar effects as in gocraft/work. Storing the maxConcurrency and priority info per queue and using it for initialisation of the server when starting the adapter (a simple wrapper over asynq).

type Adapter struct {
	taskConfigMap map[string]taskConfig
	tcMutex       sync.RWMutex
}

type taskConfig struct {
	priority       int
	maxConcurrency int
}

func (a *Adapter) server() *asynq.Server {
	return asynq.NewServer(a.redisConnOpt, asynq.Config{
		Concurrency:    a.concurrency(),
		Queues:         a.queuePriorityMap(),
	})
}

func (a *Adapter) queuePriorityMap() map[string]int {
	a.tcMutex.RLock()
	defer a.tcMutex.RUnlock()

	m := make(map[string]int)

	for tn, cfg := range f.taskConfigMap {
		if cfg.priority == 0 {
			m[tn] = 1
			continue
		}

		m[tn] = cfg.priority
	}

	return m
}

func (a *Adapter) concurrency() int {
	a.tcMutex.RLock()
	defer a.tcMutex.RUnlock()

	concurrency := 0
	for _, cfg := range f.taskConfigMap {
		if cfg.maxConcurrency == 0 {
			concurrency++
			continue
		}

		concurrency += cfg.maxConcurrency
	}

	if rtcpu := runtime.NumCPU(); concurrency < rtcpu {
		return rtcpu
	}

	return concurrency
}

Additional context Ref: Priority Sampler from gocraft/work

When the network is bad, there will be tasks without msg fields (only one state field), and the task ID cannot be obtained. If the ID is empty, the default ID will be set

Is your feature request related to a problem? Please describe. Our app has many tenants and sometimes tasks fail for one reason or another, it would be helpful to allow tenants to see jobs queued for them (emails, etc) and show the status.

Describe the solution you'd like Method(s) to query tasks from the Redis server.

Describe alternatives you've considered I looked at just re-implementing the query logic from the cli myself, but this seems like it should be in the main library.

Hi! I was trying to monitor asynq jobs with asynqmon and then I realise the parseRedisURI method doesn't parse username from HTTP standard authentication inside uri. https://github.com/hibiken/asynq/blob/0b8cfad7034159dadccf95e56b96305c0d2a4f7a/asynq.go#L462 I wanna know it's intentional or just missing or whatever. If it wasn't intentional I can add it myself.

Is your feature request related to a problem? Please describe. I have worked on a distributed semaphore in the past. However, one question that I have is that the lock count is checked after the task is dequeued from the queue, and then if the lock can't be acquired, it is again re-queued with a delay set by retryBackoff.

This should be okay in normal circumstances, but in case of downtime on dependent upstreams, I am worried that this can put unnecessary pressure on Redis and de-queueing/re-queueing may also interfere with the order of tasks in the queue.

Is there a way, the dequeue routine can take into account of this lock count before the dequeue even happens?

Describe the solution you'd like Instead of checking for the lock after the de-queue, I would like to hook the lock checker before the de-queue begins.

Describe alternatives you've considered I tried to see if I can use the Pause queue feature to help with this, however, I kinda feel like it is just a workaround, and not sure if it is okay to do so.

Describe the bug Errors are not bubbled up to the caller when starting server. For example:

ERROR: cannot subscribe to cancelation channel: UNKNOWN: redis pubsub receive error: dial tcp: address redis.dev.xyz: missing port in address

To Reproduce

srv := asynq.NewServer(
	asynq.RedisClientOpt{Addr: redisURL},
	asynq.Config{
		Queues: map[string]int{
			"high":    6,
			"default": 3,
			"low":     1,
		},
		ErrorHandler: asynq.ErrorHandlerFunc(HandleErrorFunc),
	},
)

mux := asynq.NewServeMux()
mux.HandleFunc(TypeFirebaseSyncUser, HandleFirebaseSyncUserTask)

if err := srv.Run(mux); err != nil {
  // No error is ever returned here
  log.Fatalf("could not run server: %v", err)
}

Expected behavior If an error happens such as not being able to connect to redis, the caller should be able to do what they need to do with that error. This is a problem because when I run this server in a container, the container continues to report as running with no issues, even though there are definitely issues.