Revert "Bump golang.org/x/time from 0.3.0 to 0.7.0"

Bump golang.org/x/time from 0.3.0 to 0.7.0

.github/dependabot.yaml

@@ -18,4 +18,7 @@ updates:
       interval: "weekly"
     labels:
       - "pr-deps"
-      
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"

.github/workflows/benchstat.yml

@@ -18,9 +18,9 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: 1.21.x
+          go-version: 1.23.x
       - name: Benchmark
         run: go test -run=^$ -bench=. -count=5 -timeout=60m ./... | tee -a new.txt
       - name: Upload Benchmark
@@ -42,9 +42,9 @@ jobs:
         with:
           ref: master
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: 1.21.x
+          go-version: 1.23.x
       - name: Benchmark
         run: go test -run=^$ -bench=. -count=5 -timeout=60m ./... | tee -a old.txt
       - name: Upload Benchmark
@@ -60,9 +60,9 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
-          go-version: 1.21.x
+          go-version: 1.23.x
       - name: Install benchstat
         run: go get -u golang.org/x/perf/cmd/benchstat
       - name: Download Incoming

.github/workflows/build.yml

@@ -7,7 +7,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest]
-        go-version: [1.20.x, 1.21.x]
+        go-version: [1.22.x, 1.23.x]
     runs-on: ${{ matrix.os }}
     services:
       redis:
@@ -18,7 +18,7 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go-version }}
           cache: false
@@ -45,7 +45,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest]
-        go-version: [1.20.x, 1.21.x]
+        go-version: [1.22.x, 1.23.x]
     runs-on: ${{ matrix.os }}
     services:
       redis:
@@ -56,7 +56,7 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go-version }}
           cache: false
@@ -67,3 +67,17 @@ jobs:
       - name: Test tools module
         run: cd tools && go test -race -v ./... && cd ..
 
+  golangci:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v6
+        with:
+          version: v1.61

Makefile

@@ -4,4 +4,8 @@ proto: internal/proto/asynq.proto
 	protoc -I=$(ROOT_DIR)/internal/proto \
 				 --go_out=$(ROOT_DIR)/internal/proto \
 				 --go_opt=module=github.com/hibiken/asynq/internal/proto \
-				 $(ROOT_DIR)/internal/proto/asynq.proto
+				 $(ROOT_DIR)/internal/proto/asynq.proto
+
+.PHONY: lint
+lint:
+	golangci-lint run

asynq.go

@@ -316,6 +316,9 @@ type RedisFailoverClientOpt struct {
 	// https://redis.io/topics/sentinel.
 	SentinelAddrs []string
 
+	// Redis sentinel username.
+	SentinelUsername string
+
 	// Redis sentinel password.
 	SentinelPassword string
 
@@ -364,6 +367,7 @@ func (opt RedisFailoverClientOpt) MakeRedisClient() interface{} {
 	return redis.NewFailoverClient(&redis.FailoverOptions{
 		MasterName:       opt.MasterName,
 		SentinelAddrs:    opt.SentinelAddrs,
+		SentinelUsername: opt.SentinelUsername,
 		SentinelPassword: opt.SentinelPassword,
 		Username:         opt.Username,
 		Password:         opt.Password,

benchmark_test.go

@@ -55,7 +55,7 @@ func BenchmarkEndToEndSimple(b *testing.B) {
 		}
 		b.StartTimer() // end setup
 
-		srv.Start(HandlerFunc(handler))
+		_ = srv.Start(HandlerFunc(handler))
 		wg.Wait()
 
 		b.StopTimer() // begin teardown
@@ -117,7 +117,7 @@ func BenchmarkEndToEnd(b *testing.B) {
 		}
 		b.StartTimer() // end setup
 
-		srv.Start(HandlerFunc(handler))
+		_ = srv.Start(HandlerFunc(handler))
 		wg.Wait()
 
 		b.StopTimer() // begin teardown
@@ -174,7 +174,7 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 		}
 		b.StartTimer() // end setup
 
-		srv.Start(HandlerFunc(handler))
+		_ = srv.Start(HandlerFunc(handler))
 		wg.Wait()
 
 		b.StopTimer() // begin teardown
@@ -215,7 +215,7 @@ func BenchmarkClientWhileServerRunning(b *testing.B) {
 		handler := func(ctx context.Context, t *Task) error {
 			return nil
 		}
-		srv.Start(HandlerFunc(handler))
+		_ = srv.Start(HandlerFunc(handler))
 
 		b.StartTimer() // end setup
 

client.go

@@ -419,6 +419,11 @@ func (c *Client) EnqueueContext(ctx context.Context, task *Task, opts ...Option)
 	return newTaskInfo(msg, state, opt.processAt, nil), nil
 }
 
+// Ping performs a ping against the redis connection.
+func (c *Client) Ping() error {
+	return c.broker.Ping()
+}
+
 func (c *Client) enqueue(ctx context.Context, msg *base.TaskMessage, uniqueTTL time.Duration) error {
 	if uniqueTTL > 0 {
 		return c.broker.EnqueueUnique(ctx, msg, uniqueTTL)
@@ -428,7 +433,7 @@ func (c *Client) enqueue(ctx context.Context, msg *base.TaskMessage, uniqueTTL t
 
 func (c *Client) schedule(ctx context.Context, msg *base.TaskMessage, t time.Time, uniqueTTL time.Duration) error {
 	if uniqueTTL > 0 {
-		ttl := t.Add(uniqueTTL).Sub(time.Now())
+		ttl := time.Until(t.Add(uniqueTTL))
 		return c.broker.ScheduleUnique(ctx, msg, t, ttl)
 	}
 	return c.broker.Schedule(ctx, msg, t)

client_test.go

@@ -556,11 +556,11 @@ func TestClientEnqueueWithGroupOption(t *testing.T) {
 			},
 		},
 		{
-			desc: "With Group and ProcessIn options",
+			desc: "With Group and ProcessAt options",
 			task: task,
 			opts: []Option{
 				Group("mygroup"),
-				ProcessIn(30 * time.Minute),
+				ProcessAt(now.Add(30 * time.Minute)),
 			},
 			wantInfo: &TaskInfo{
 				Queue:         "default",
@@ -1173,7 +1173,7 @@ func TestClientEnqueueUniqueWithProcessAtOption(t *testing.T) {
 		}
 
 		gotTTL := r.TTL(context.Background(), base.UniqueKey(base.DefaultQueueName, tc.task.Type(), tc.task.Payload())).Val()
-		wantTTL := tc.at.Add(tc.ttl).Sub(time.Now())
+		wantTTL := time.Until(tc.at.Add(tc.ttl))
 		if !cmp.Equal(wantTTL.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
 			t.Errorf("TTL = %v, want %v", gotTTL, wantTTL)
 			continue

go.mod

@@ -1,6 +1,6 @@
 module github.com/hibiken/asynq
 
-go 1.20
+go 1.22
 
 require (
 	github.com/golang/protobuf v1.5.3

go.sum

@@ -1,11 +1,15 @@
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0XL9UY=
+github.com/frankban/quicktest v1.14.4/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
@@ -15,16 +19,21 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/redis/go-redis/v9 v9.4.0 h1:Yzoz33UZw9I/mFhx4MNrB6Fk+XHO1VukNcCa1+lwyKk=
 github.com/redis/go-redis/v9 v9.4.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/spf13/cast v1.5.1 h1:R+kOtfhWQE6TVQzY+4D7wJLBgkdVasCEFxSUBYBYIlA=
 github.com/spf13/cast v1.5.1/go.mod h1:b9PdjNptOpzXr7Rq1q9gJML/2cdGQAo69NKzQ10KN48=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
@@ -37,3 +46,4 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
 google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
 google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

heartbeat.go

@@ -120,7 +120,9 @@ func (h *heartbeater) start(wg *sync.WaitGroup) {
 		for {
 			select {
 			case <-h.done:
-				h.broker.ClearServerState(h.host, h.pid, h.serverID)
+				if err := h.broker.ClearServerState(h.host, h.pid, h.serverID); err != nil {
+					h.logger.Errorf("Failed to clear server state: %v", err)
+				}
 				h.logger.Debug("Heartbeater done")
 				timer.Stop()
 				return

internal/base/base.go

@@ -14,7 +14,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
+	"github.com/golang/protobuf/ptypes" //nolint: staticcheck
 	"github.com/hibiken/asynq/internal/errors"
 	pb "github.com/hibiken/asynq/internal/proto"
 	"github.com/hibiken/asynq/internal/timeutil"
@@ -379,7 +379,7 @@ func EncodeServerInfo(info *ServerInfo) ([]byte, error) {
 	for q, p := range info.Queues {
 		queues[q] = int32(p)
 	}
-	started, err := ptypes.TimestampProto(info.Started)
+	started, err := ptypes.TimestampProto(info.Started) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -406,7 +406,7 @@ func DecodeServerInfo(b []byte) (*ServerInfo, error) {
 	for q, p := range pbmsg.GetQueues() {
 		queues[q] = int(p)
 	}
-	startTime, err := ptypes.Timestamp(pbmsg.GetStartTime())
+	startTime, err := ptypes.Timestamp(pbmsg.GetStartTime()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -441,11 +441,11 @@ func EncodeWorkerInfo(info *WorkerInfo) ([]byte, error) {
 	if info == nil {
 		return nil, fmt.Errorf("cannot encode nil worker info")
 	}
-	startTime, err := ptypes.TimestampProto(info.Started)
+	startTime, err := ptypes.TimestampProto(info.Started) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
-	deadline, err := ptypes.TimestampProto(info.Deadline)
+	deadline, err := ptypes.TimestampProto(info.Deadline) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -468,11 +468,11 @@ func DecodeWorkerInfo(b []byte) (*WorkerInfo, error) {
 	if err := proto.Unmarshal(b, &pbmsg); err != nil {
 		return nil, err
 	}
-	startTime, err := ptypes.Timestamp(pbmsg.GetStartTime())
+	startTime, err := ptypes.Timestamp(pbmsg.GetStartTime()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
-	deadline, err := ptypes.Timestamp(pbmsg.GetDeadline())
+	deadline, err := ptypes.Timestamp(pbmsg.GetDeadline()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -519,11 +519,11 @@ func EncodeSchedulerEntry(entry *SchedulerEntry) ([]byte, error) {
 	if entry == nil {
 		return nil, fmt.Errorf("cannot encode nil scheduler entry")
 	}
-	next, err := ptypes.TimestampProto(entry.Next)
+	next, err := ptypes.TimestampProto(entry.Next) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
-	prev, err := ptypes.TimestampProto(entry.Prev)
+	prev, err := ptypes.TimestampProto(entry.Prev) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -544,11 +544,11 @@ func DecodeSchedulerEntry(b []byte) (*SchedulerEntry, error) {
 	if err := proto.Unmarshal(b, &pbmsg); err != nil {
 		return nil, err
 	}
-	next, err := ptypes.Timestamp(pbmsg.GetNextEnqueueTime())
+	next, err := ptypes.Timestamp(pbmsg.GetNextEnqueueTime()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
-	prev, err := ptypes.Timestamp(pbmsg.GetPrevEnqueueTime())
+	prev, err := ptypes.Timestamp(pbmsg.GetPrevEnqueueTime()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -578,7 +578,7 @@ func EncodeSchedulerEnqueueEvent(event *SchedulerEnqueueEvent) ([]byte, error) {
 	if event == nil {
 		return nil, fmt.Errorf("cannot encode nil enqueue event")
 	}
-	enqueuedAt, err := ptypes.TimestampProto(event.EnqueuedAt)
+	enqueuedAt, err := ptypes.TimestampProto(event.EnqueuedAt) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}
@@ -595,7 +595,7 @@ func DecodeSchedulerEnqueueEvent(b []byte) (*SchedulerEnqueueEvent, error) {
 	if err := proto.Unmarshal(b, &pbmsg); err != nil {
 		return nil, err
 	}
-	enqueuedAt, err := ptypes.Timestamp(pbmsg.GetEnqueueTime())
+	enqueuedAt, err := ptypes.Timestamp(pbmsg.GetEnqueueTime()) //nolint: staticcheck
 	if err != nil {
 		return nil, err
 	}

internal/rdb/rdb.go

@@ -829,6 +829,7 @@ const (
 // KEYS[6] -> asynq:{<qname>}:failed:<yyyy-mm-dd>
 // KEYS[7] -> asynq:{<qname>}:processed
 // KEYS[8] -> asynq:{<qname>}:failed
+// KEYS[9] -> asynq:{<qname>}:t:
 // -------
 // ARGV[1] -> task ID
 // ARGV[2] -> updated base.TaskMessage value
@@ -845,8 +846,22 @@ if redis.call("ZREM", KEYS[3], ARGV[1]) == 0 then
   return redis.error_reply("NOT FOUND")
 end
 redis.call("ZADD", KEYS[4], ARGV[3], ARGV[1])
-redis.call("ZREMRANGEBYSCORE", KEYS[4], "-inf", ARGV[4])
-redis.call("ZREMRANGEBYRANK", KEYS[4], 0, -ARGV[5])
+local old = redis.call("ZRANGE", KEYS[4], "-inf", ARGV[4], "BYSCORE")
+if #old > 0 then
+	for _, id in ipairs(old) do
+		redis.call("DEL", KEYS[9] .. id)
+	end
+	redis.call("ZREM", KEYS[4], unpack(old))
+end
+
+local extra = redis.call("ZRANGE", KEYS[4], 0, -ARGV[5])
+if #extra > 0 then
+	for _, id in ipairs(extra) do
+		redis.call("DEL", KEYS[9] .. id)
+	end
+	redis.call("ZREM", KEYS[4], unpack(extra))
+end
+
 redis.call("HSET", KEYS[1], "msg", ARGV[2], "state", "archived")
 local n = redis.call("INCR", KEYS[5])
 if tonumber(n) == 1 then
@@ -889,6 +904,7 @@ func (r *RDB) Archive(ctx context.Context, msg *base.TaskMessage, errMsg string)
 		base.FailedKey(msg.Queue, now),
 		base.ProcessedTotalKey(msg.Queue),
 		base.FailedTotalKey(msg.Queue),
+		base.TaskKeyPrefix(msg.Queue),
 	}
 	argv := []interface{}{
 		msg.ID,

internal/rdb/rdb_test.go

@@ -2002,7 +2002,6 @@ func TestArchive(t *testing.T) {
 	}
 	errMsg := "SMTP server not responding"
 
-	// TODO(hibiken): add test cases for trimming
 	tests := []struct {
 		active       map[string][]*base.TaskMessage
 		lease        map[string][]base.Z
@@ -2171,6 +2170,163 @@ func TestArchive(t *testing.T) {
 	}
 }
 
+func TestArchiveTrim(t *testing.T) {
+	r := setup(t)
+	defer r.Close()
+	now := time.Now()
+	r.SetClock(timeutil.NewSimulatedClock(now))
+
+	t1 := &base.TaskMessage{
+		ID:      uuid.NewString(),
+		Type:    "send_email",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 25,
+		Timeout: 1800,
+	}
+	t2 := &base.TaskMessage{
+		ID:      uuid.NewString(),
+		Type:    "reindex",
+		Payload: nil,
+		Queue:   "default",
+		Retry:   25,
+		Retried: 0,
+		Timeout: 3000,
+	}
+	errMsg := "SMTP server not responding"
+
+	maxArchiveSet := make([]base.Z, 0)
+	for i := 0; i < maxArchiveSize-1; i++ {
+		maxArchiveSet = append(maxArchiveSet, base.Z{Message: &base.TaskMessage{
+			ID:      uuid.NewString(),
+			Type:    "generate_csv",
+			Payload: nil,
+			Queue:   "default",
+			Retry:   25,
+			Retried: 0,
+			Timeout: 60,
+		}, Score: now.Add(-time.Hour + -time.Second*time.Duration(i)).Unix()})
+	}
+
+	wantMaxArchiveSet := make([]base.Z, 0)
+	// newly archived task should be at the front
+	wantMaxArchiveSet = append(wantMaxArchiveSet, base.Z{Message: h.TaskMessageWithError(*t1, errMsg, now), Score: now.Unix()})
+	// oldest task should be dropped from the set
+	wantMaxArchiveSet = append(wantMaxArchiveSet, maxArchiveSet[:len(maxArchiveSet)-1]...)
+
+	tests := []struct {
+		toArchive    map[string][]*base.TaskMessage
+		lease        map[string][]base.Z
+		archived     map[string][]base.Z
+		wantArchived map[string][]base.Z
+	}{
+		{ // simple, 1 to be archived, 1 already archived, both are in the archive set
+			toArchive: map[string][]*base.TaskMessage{
+				"default": {t1},
+			},
+			lease: map[string][]base.Z{
+				"default": {
+					{Message: t1, Score: now.Add(10 * time.Second).Unix()},
+				},
+			},
+			archived: map[string][]base.Z{
+				"default": {
+					{Message: t2, Score: now.Add(-time.Hour).Unix()},
+				},
+			},
+			wantArchived: map[string][]base.Z{
+				"default": {
+					{Message: h.TaskMessageWithError(*t1, errMsg, now), Score: now.Unix()},
+					{Message: t2, Score: now.Add(-time.Hour).Unix()},
+				},
+			},
+		},
+		{ // 1 to be archived, 1 already archived but past expiry, only the newly archived task should be left
+			toArchive: map[string][]*base.TaskMessage{
+				"default": {t1},
+			},
+			lease: map[string][]base.Z{
+				"default": {
+					{Message: t1, Score: now.Add(10 * time.Second).Unix()},
+				},
+			},
+			archived: map[string][]base.Z{
+				"default": {
+					{Message: t2, Score: now.Add(-time.Hour * 24 * (archivedExpirationInDays + 1)).Unix()},
+				},
+			},
+			wantArchived: map[string][]base.Z{
+				"default": {
+					{Message: h.TaskMessageWithError(*t1, errMsg, now), Score: now.Unix()},
+				},
+			},
+		},
+		{ // 1 to be archived, maxArchiveSize in archive set, archive set should be trimmed back to maxArchiveSize and newly archived task should be in the set
+			toArchive: map[string][]*base.TaskMessage{
+				"default": {t1},
+			},
+			lease: map[string][]base.Z{
+				"default": {
+					{Message: t1, Score: now.Add(10 * time.Second).Unix()},
+				},
+			},
+			archived: map[string][]base.Z{
+				"default": maxArchiveSet,
+			},
+			wantArchived: map[string][]base.Z{
+				"default": wantMaxArchiveSet,
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case
+		h.SeedAllActiveQueues(t, r.client, tc.toArchive)
+		h.SeedAllLease(t, r.client, tc.lease)
+		h.SeedAllArchivedQueues(t, r.client, tc.archived)
+
+		for _, tasks := range tc.toArchive {
+			for _, target := range tasks {
+				err := r.Archive(context.Background(), target, errMsg)
+				if err != nil {
+					t.Errorf("(*RDB).Archive(%v, %v) = %v, want nil", target, errMsg, err)
+					continue
+				}
+			}
+		}
+
+		for queue, want := range tc.wantArchived {
+			gotArchived := h.GetArchivedEntries(t, r.client, queue)
+
+			if diff := cmp.Diff(want, gotArchived, h.SortZSetEntryOpt, zScoreCmpOpt, timeCmpOpt); diff != "" {
+				t.Errorf("mismatch found in %q after calling (*RDB).Archive: (-want, +got):\n%s", base.ArchivedKey(queue), diff)
+			}
+
+			// check that only keys present in the archived set are in rdb
+			vals := r.client.Keys(context.Background(), base.TaskKeyPrefix(queue)+"*").Val()
+			if len(vals) != len(gotArchived) {
+				t.Errorf("len of keys = %v, want %v", len(vals), len(gotArchived))
+				return
+			}
+
+			for _, val := range vals {
+				found := false
+				for _, entry := range gotArchived {
+					if strings.Contains(val, entry.Message.ID) {
+						found = true
+						break
+					}
+				}
+
+				if !found {
+					t.Errorf("key %v not found in archived set (it was orphaned by the archive trim)", val)
+				}
+			}
+		}
+	}
+}
+
 func TestForwardIfReadyWithGroup(t *testing.T) {
 	r := setup(t)
 	defer r.Close()
@@ -3050,7 +3206,7 @@ func TestCancelationPubSub(t *testing.T) {
 	publish := []string{"one", "two", "three"}
 
 	for _, msg := range publish {
-		r.PublishCancelation(msg)
+		_ = r.PublishCancelation(msg)
 	}
 
 	// allow for message to reach subscribers.

periodic_task_manager.go

@@ -7,7 +7,6 @@ package asynq
 import (
 	"crypto/sha256"
 	"fmt"
-	"io"
 	"sort"
 	"sync"
 	"time"
@@ -79,13 +78,13 @@ type PeriodicTaskConfig struct {
 
 func (c *PeriodicTaskConfig) hash() string {
 	h := sha256.New()
-	io.WriteString(h, c.Cronspec)
-	io.WriteString(h, c.Task.Type())
+	_, _ = h.Write([]byte(c.Cronspec))
+	_, _ = h.Write([]byte(c.Task.Type()))
 	h.Write(c.Task.Payload())
 	opts := stringifyOptions(c.Opts)
 	sort.Strings(opts)
 	for _, opt := range opts {
-		io.WriteString(h, opt)
+		_, _ = h.Write([]byte(opt))
 	}
 	return fmt.Sprintf("%x", h.Sum(nil))
 }

periodic_task_manager_test.go

@@ -32,6 +32,7 @@ func (p *FakeConfigProvider) GetConfigs() ([]*PeriodicTaskConfig, error) {
 }
 
 func TestNewPeriodicTaskManager(t *testing.T) {
+	redisConnOpt := getRedisConnOpt(t)
 	cfgs := []*PeriodicTaskConfig{
 		{Cronspec: "* * * * *", Task: NewTask("foo", nil)},
 		{Cronspec: "* * * * *", Task: NewTask("bar", nil)},
@@ -43,14 +44,14 @@ func TestNewPeriodicTaskManager(t *testing.T) {
 		{
 			desc: "with provider and redisConnOpt",
 			opts: PeriodicTaskManagerOpts{
-				RedisConnOpt:               RedisClientOpt{Addr: ":6379"},
+				RedisConnOpt:               redisConnOpt,
 				PeriodicTaskConfigProvider: &FakeConfigProvider{cfgs: cfgs},
 			},
 		},
 		{
 			desc: "with sync option",
 			opts: PeriodicTaskManagerOpts{
-				RedisConnOpt:               RedisClientOpt{Addr: ":6379"},
+				RedisConnOpt:               redisConnOpt,
 				PeriodicTaskConfigProvider: &FakeConfigProvider{cfgs: cfgs},
 				SyncInterval:               5 * time.Minute,
 			},
@@ -58,7 +59,7 @@ func TestNewPeriodicTaskManager(t *testing.T) {
 		{
 			desc: "with scheduler option",
 			opts: PeriodicTaskManagerOpts{
-				RedisConnOpt:               RedisClientOpt{Addr: ":6379"},
+				RedisConnOpt:               redisConnOpt,
 				PeriodicTaskConfigProvider: &FakeConfigProvider{cfgs: cfgs},
 				SyncInterval:               5 * time.Minute,
 				SchedulerOpts: &SchedulerOpts{
@@ -74,37 +75,33 @@ func TestNewPeriodicTaskManager(t *testing.T) {
 			t.Errorf("%s; NewPeriodicTaskManager returned error: %v", tc.desc, err)
 		}
 	}
-}
 
-func TestNewPeriodicTaskManagerError(t *testing.T) {
-	cfgs := []*PeriodicTaskConfig{
-		{Cronspec: "* * * * *", Task: NewTask("foo", nil)},
-		{Cronspec: "* * * * *", Task: NewTask("bar", nil)},
-	}
-	tests := []struct {
-		desc string
-		opts PeriodicTaskManagerOpts
-	}{
-		{
-			desc: "without provider",
-			opts: PeriodicTaskManagerOpts{
-				RedisConnOpt: RedisClientOpt{Addr: ":6379"},
+	t.Run("error", func(t *testing.T) {
+		tests := []struct {
+			desc string
+			opts PeriodicTaskManagerOpts
+		}{
+			{
+				desc: "without provider",
+				opts: PeriodicTaskManagerOpts{
+					RedisConnOpt: redisConnOpt,
+				},
 			},
-		},
-		{
-			desc: "without redisConOpt",
-			opts: PeriodicTaskManagerOpts{
-				PeriodicTaskConfigProvider: &FakeConfigProvider{cfgs: cfgs},
+			{
+				desc: "without redisConOpt",
+				opts: PeriodicTaskManagerOpts{
+					PeriodicTaskConfigProvider: &FakeConfigProvider{cfgs: cfgs},
+				},
 			},
-		},
-	}
-
-	for _, tc := range tests {
-		_, err := NewPeriodicTaskManager(tc.opts)
-		if err == nil {
-			t.Errorf("%s; NewPeriodicTaskManager did not return error", tc.desc)
 		}
-	}
+
+		for _, tc := range tests {
+			_, err := NewPeriodicTaskManager(tc.opts)
+			if err == nil {
+				t.Errorf("%s; NewPeriodicTaskManager did not return error", tc.desc)
+			}
+		}
+	})
 }
 
 func TestPeriodicTaskConfigHash(t *testing.T) {

processor.go

@@ -8,7 +8,7 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"math/rand"
+	"math/rand/v2"
 	"runtime"
 	"runtime/debug"
 	"sort"
@@ -181,7 +181,8 @@ func (p *processor) exec() {
 			// Sleep to avoid slamming redis and let scheduler move tasks into queues.
 			// Note: We are not using blocking pop operation and polling queues instead.
 			// This adds significant load to redis.
-			time.Sleep(p.taskCheckInterval)
+			jitter := rand.N(p.taskCheckInterval)
+			time.Sleep(p.taskCheckInterval/2 + jitter)
 			<-p.sema // release token
 			return
 		case err != nil:
@@ -261,7 +262,8 @@ func (p *processor) requeue(l *base.Lease, msg *base.TaskMessage) {
 		// If lease is not valid, do not write to redis; Let recoverer take care of it.
 		return
 	}
-	ctx, _ := context.WithDeadline(context.Background(), l.Deadline())
+	ctx, cancel := context.WithDeadline(context.Background(), l.Deadline())
+	defer cancel()
 	err := p.broker.Requeue(ctx, msg)
 	if err != nil {
 		p.logger.Errorf("Could not push task id=%s back to queue: %v", msg.ID, err)
@@ -283,7 +285,8 @@ func (p *processor) markAsComplete(l *base.Lease, msg *base.TaskMessage) {
 		// If lease is not valid, do not write to redis; Let recoverer take care of it.
 		return
 	}
-	ctx, _ := context.WithDeadline(context.Background(), l.Deadline())
+	ctx, cancel := context.WithDeadline(context.Background(), l.Deadline())
+	defer cancel()
 	err := p.broker.MarkAsComplete(ctx, msg)
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s type=%q from %q to %q:  %+v",
@@ -304,7 +307,8 @@ func (p *processor) markAsDone(l *base.Lease, msg *base.TaskMessage) {
 		// If lease is not valid, do not write to redis; Let recoverer take care of it.
 		return
 	}
-	ctx, _ := context.WithDeadline(context.Background(), l.Deadline())
+	ctx, cancel := context.WithDeadline(context.Background(), l.Deadline())
+	defer cancel()
 	err := p.broker.Done(ctx, msg)
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not remove task id=%s type=%q from %q err: %+v", msg.ID, msg.Type, base.ActiveKey(msg.Queue), err)
@@ -323,20 +327,23 @@ func (p *processor) markAsDone(l *base.Lease, msg *base.TaskMessage) {
 // the task should not be retried and should be archived instead.
 var SkipRetry = errors.New("skip retry for the task")
 
+// RevokeTask is used as a return value from Handler.ProcessTask to indicate that
+// the task should not be retried or archived.
+var RevokeTask = errors.New("revoke task")
+
 func (p *processor) handleFailedMessage(ctx context.Context, l *base.Lease, msg *base.TaskMessage, err error) {
 	if p.errHandler != nil {
 		p.errHandler.HandleError(ctx, NewTask(msg.Type, msg.Payload), err)
 	}
-	if !p.isFailureFunc(err) {
-		// retry the task without marking it as failed
-		p.retry(l, msg, err, false /*isFailure*/)
-		return
-	}
-	if msg.Retried >= msg.Retry || errors.Is(err, SkipRetry) {
+	switch {
+	case errors.Is(err, RevokeTask):
+		p.logger.Warnf("revoke task id=%s", msg.ID)
+		p.markAsDone(l, msg)
+	case msg.Retried >= msg.Retry || errors.Is(err, SkipRetry):
 		p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
 		p.archive(l, msg, err)
-	} else {
-		p.retry(l, msg, err, true /*isFailure*/)
+	default:
+		p.retry(l, msg, err, p.isFailureFunc(err))
 	}
 }
 
@@ -345,7 +352,8 @@ func (p *processor) retry(l *base.Lease, msg *base.TaskMessage, e error, isFailu
 		// If lease is not valid, do not write to redis; Let recoverer take care of it.
 		return
 	}
-	ctx, _ := context.WithDeadline(context.Background(), l.Deadline())
+	ctx, cancel := context.WithDeadline(context.Background(), l.Deadline())
+	defer cancel()
 	d := p.retryDelayFunc(msg.Retried, e, NewTask(msg.Type, msg.Payload))
 	retryAt := time.Now().Add(d)
 	err := p.broker.Retry(ctx, msg, retryAt, e.Error(), isFailure)
@@ -367,7 +375,8 @@ func (p *processor) archive(l *base.Lease, msg *base.TaskMessage, e error) {
 		// If lease is not valid, do not write to redis; Let recoverer take care of it.
 		return
 	}
-	ctx, _ := context.WithDeadline(context.Background(), l.Deadline())
+	ctx, cancel := context.WithDeadline(context.Background(), l.Deadline())
+	defer cancel()
 	err := p.broker.Archive(ctx, msg, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.ActiveKey(msg.Queue), base.ArchivedKey(msg.Queue))
@@ -404,8 +413,7 @@ func (p *processor) queues() []string {
 			names = append(names, qname)
 		}
 	}
-	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	r.Shuffle(len(names), func(i, j int) { names[i], names[j] = names[j], names[i] })
+	rand.Shuffle(len(names), func(i, j int) { names[i], names[j] = names[j], names[i] })
 	return uniq(names, len(p.queueConfig))
 }
 

processor_test.go

@@ -295,6 +295,7 @@ func TestProcessorRetry(t *testing.T) {
 
 	errMsg := "something went wrong"
 	wrappedSkipRetry := fmt.Errorf("%s:%w", errMsg, SkipRetry)
+	wrappedRevokeTask := fmt.Errorf("%s:%w", errMsg, RevokeTask)
 
 	tests := []struct {
 		desc         string              // test description
@@ -312,7 +313,7 @@ func TestProcessorRetry(t *testing.T) {
 			pending: []*base.TaskMessage{m1, m2, m3, m4},
 			delay:   time.Minute,
 			handler: HandlerFunc(func(ctx context.Context, task *Task) error {
-				return fmt.Errorf(errMsg)
+				return errors.New(errMsg)
 			}),
 			wait:         2 * time.Second,
 			wantErrMsg:   errMsg,
@@ -346,6 +347,32 @@ func TestProcessorRetry(t *testing.T) {
 			wantArchived: []*base.TaskMessage{m1, m2},
 			wantErrCount: 2, // ErrorHandler should still be called with SkipRetry error
 		},
+		{
+			desc:    "Should revoke task",
+			pending: []*base.TaskMessage{m1, m2},
+			delay:   time.Minute,
+			handler: HandlerFunc(func(ctx context.Context, task *Task) error {
+				return RevokeTask // return RevokeTask without wrapping
+			}),
+			wait:         2 * time.Second,
+			wantErrMsg:   RevokeTask.Error(),
+			wantRetry:    []*base.TaskMessage{},
+			wantArchived: []*base.TaskMessage{},
+			wantErrCount: 2, // ErrorHandler should still be called with RevokeTask error
+		},
+		{
+			desc:    "Should revoke task (with error wrapping)",
+			pending: []*base.TaskMessage{m1, m2},
+			delay:   time.Minute,
+			handler: HandlerFunc(func(ctx context.Context, task *Task) error {
+				return wrappedRevokeTask
+			}),
+			wait:         2 * time.Second,
+			wantErrMsg:   wrappedRevokeTask.Error(),
+			wantRetry:    []*base.TaskMessage{},
+			wantArchived: []*base.TaskMessage{},
+			wantErrCount: 2, // ErrorHandler should still be called with RevokeTask error
+		},
 	}
 
 	for _, tc := range tests {

scheduler.go

@@ -289,7 +289,9 @@ func (s *Scheduler) runHeartbeater() {
 		select {
 		case <-s.done:
 			s.logger.Debugf("Scheduler heatbeater shutting down")
-			s.rdb.ClearSchedulerEntries(s.id)
+			if err := s.rdb.ClearSchedulerEntries(s.id); err != nil {
+				s.logger.Errorf("Failed to clear the scheduler entries: %v", err)
+			}
 			ticker.Stop()
 			return
 		case <-ticker.C:
@@ -336,3 +338,14 @@ func (s *Scheduler) clearHistory() {
 		}
 	}
 }
+
+// Ping performs a ping against the redis connection.
+func (s *Scheduler) Ping() error {
+	s.state.mu.Lock()
+	defer s.state.mu.Unlock()
+	if s.state.value == srvStateClosed {
+		return nil
+	}
+
+	return s.rdb.Ping()
+}

scheduler_test.go

@@ -5,12 +5,12 @@
 package asynq
 
 import (
-	"github.com/redis/go-redis/v9"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/redis/go-redis/v9"
 
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/testutil"
@@ -114,7 +114,7 @@ func TestSchedulerWhenRedisDown(t *testing.T) {
 
 	// Connect to non-existent redis instance to simulate a redis server being down.
 	scheduler := NewScheduler(
-		RedisClientOpt{Addr: ":9876"},
+		RedisClientOpt{Addr: ":9876"}, // no Redis listening to this port.
 		&SchedulerOpts{EnqueueErrorHandler: errorHandler},
 	)
 

server.go

@@ -9,7 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
-	"math/rand"
+	"math/rand/v2"
 	"runtime"
 	"strings"
 	"sync"
@@ -400,9 +400,8 @@ func toInternalLogLevel(l LogLevel) log.Level {
 // DefaultRetryDelayFunc is the default RetryDelayFunc used if one is not specified in Config.
 // It uses exponential back-off strategy to calculate the retry delay.
 func DefaultRetryDelayFunc(n int, e error, t *Task) time.Duration {
-	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	// Formula taken from https://github.com/mperham/sidekiq.
-	s := int(math.Pow(float64(n), 4)) + 15 + (r.Intn(30) * (n + 1))
+	s := int(math.Pow(float64(n), 4)) + 15 + (rand.IntN(30) * (n + 1))
 	return time.Duration(s) * time.Second
 }
 
@@ -633,6 +632,10 @@ func NewServerFromRedisClient(c redis.UniversalClient, cfg Config) *Server {
 // One exception to this rule is when ProcessTask returns a SkipRetry error.
 // If the returned error is SkipRetry or an error wraps SkipRetry, retry is
 // skipped and the task will be immediately archived instead.
+//
+// Another exception to this rule is when ProcessTask returns a RevokeTask error.
+// If the returned error is RevokeTask or an error wraps RevokeTask, the task
+// will not be retried or archived.
 type Handler interface {
 	ProcessTask(context.Context, *Task) error
 }
@@ -759,7 +762,7 @@ func (srv *Server) Shutdown() {
 func (srv *Server) Stop() {
 	srv.state.mu.Lock()
 	if srv.state.value != srvStateActive {
-		// Invalid calll to Stop, server can only go from Active state to Stopped state.
+		// Invalid call to Stop, server can only go from Active state to Stopped state.
 		srv.state.mu.Unlock()
 		return
 	}
@@ -770,3 +773,16 @@ func (srv *Server) Stop() {
 	srv.processor.stop()
 	srv.logger.Info("Processor stopped")
 }
+
+// Ping performs a ping against the redis connection.
+//
+// This is an alternative to the HealthCheckFunc available in the Config object.
+func (srv *Server) Ping() error {
+	srv.state.mu.Lock()
+	defer srv.state.mu.Unlock()
+	if srv.state.value == srvStateClosed {
+		return nil
+	}
+
+	return srv.broker.Ping()
+}

server_test.go

@@ -85,13 +85,13 @@ func TestServerRun(t *testing.T) {
 	ignoreOpt := goleak.IgnoreTopFunction("github.com/redis/go-redis/v9/internal/pool.(*ConnPool).reaper")
 	defer goleak.VerifyNone(t, ignoreOpt)
 
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv := NewServer(getRedisConnOpt(t), Config{LogLevel: testLogLevel})
 
 	done := make(chan struct{})
 	// Make sure server exits when receiving TERM signal.
 	go func() {
 		time.Sleep(2 * time.Second)
-		syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+		_ = syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
 		done <- struct{}{}
 	}()
 
@@ -110,7 +110,7 @@ func TestServerRun(t *testing.T) {
 }
 
 func TestServerErrServerClosed(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv := NewServer(getRedisConnOpt(t), Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -123,7 +123,7 @@ func TestServerErrServerClosed(t *testing.T) {
 }
 
 func TestServerErrNilHandler(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv := NewServer(getRedisConnOpt(t), Config{LogLevel: testLogLevel})
 	err := srv.Start(nil)
 	if err == nil {
 		t.Error("Starting server with nil handler: (*Server).Start(nil) did not return error")
@@ -132,7 +132,7 @@ func TestServerErrNilHandler(t *testing.T) {
 }
 
 func TestServerErrServerRunning(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv := NewServer(getRedisConnOpt(t), Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -153,7 +153,7 @@ func TestServerWithRedisDown(t *testing.T) {
 	}()
 	r := rdb.NewRDB(setup(t))
 	testBroker := testbroker.NewTestBroker(r)
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv := NewServer(getRedisConnOpt(t), Config{LogLevel: testLogLevel})
 	srv.broker = testBroker
 	srv.forwarder.broker = testBroker
 	srv.heartbeater.broker = testBroker

tools/asynq/cmd/root.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"strings"
 	"text/tabwriter"
+	"time"
 	"unicode"
 	"unicode/utf8"
 
@@ -369,6 +370,11 @@ func createRDB() *rdb.RDB {
 	return rdb.NewRDB(c)
 }
 
+// createClient creates a Client instance using flag values and returns it.
+func createClient() *asynq.Client {
+	return asynq.NewClient(getRedisConnOpt())
+}
+
 // createInspector creates a Inspector instance using flag values and returns it.
 func createInspector() *asynq.Inspector {
 	return asynq.NewInspector(getRedisConnOpt())
@@ -456,3 +462,37 @@ func isPrintable(data []byte) bool {
 	}
 	return !isAllSpace
 }
+
+// Helper to turn a command line flag into a duration
+func getDuration(cmd *cobra.Command, arg string) time.Duration {
+	durationStr, err := cmd.Flags().GetString(arg)
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	duration, err := time.ParseDuration(durationStr)
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	return duration
+}
+
+// Helper to turn a command line flag into a time
+func getTime(cmd *cobra.Command, arg string) time.Time {
+	timeStr, err := cmd.Flags().GetString(arg)
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	timeVal, err := time.Parse(time.RFC3339, timeStr)
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	return timeVal
+}

tools/asynq/cmd/task.go

@@ -53,6 +53,24 @@ func init() {
 	taskRunCmd.MarkFlagRequired("queue")
 	taskRunCmd.MarkFlagRequired("id")
 
+	taskCmd.AddCommand(taskEnqueueCmd)
+	taskEnqueueCmd.Flags().StringP("type_name", "t", "", "type name to enqueue the task as (required)")
+	taskEnqueueCmd.Flags().StringP("payload", "l", "", "payload to enqueue (required)")
+	// The following are the various OptionTypes; if not specified we won't pass them so that composeOptions()
+	// can apply its own defaults
+	taskEnqueueCmd.Flags().Int("retry", 0, "maximum retries")
+	taskEnqueueCmd.Flags().String("queue", "", "queue to enqueue the task to")
+	taskEnqueueCmd.Flags().String("id", "", "id to enqueue the task as")
+	taskEnqueueCmd.Flags().String("timeout", "", "timeout for the task (how long it can run); must be parseable as a time.Duration")
+	taskEnqueueCmd.Flags().String("deadline", "", "deadline for the task; must be in RFC3339 format")
+	taskEnqueueCmd.Flags().String("unique", "", "unique period for the task (duration within which it is guaranteed to be unique); must be parseable as a time.Duration")
+	taskEnqueueCmd.Flags().String("process_at", "", "process at time for the task; must be in RFC3339 format")
+	taskEnqueueCmd.Flags().String("process_in", "", "process in window for the task; must be parseable as a time.Duration")
+	taskEnqueueCmd.Flags().String("retention", "", "retention window for the task; must be parseable as a time.Duration")
+	taskEnqueueCmd.Flags().String("group", "", "group for the task")
+	taskEnqueueCmd.MarkFlagRequired("type_name")
+	taskEnqueueCmd.MarkFlagRequired("payload")
+
 	taskCmd.AddCommand(taskArchiveAllCmd)
 	taskArchiveAllCmd.Flags().StringP("queue", "q", "", "queue to which the tasks belong (required)")
 	taskArchiveAllCmd.Flags().StringP("state", "s", "", "state of the tasks; one of { pending | aggregating | scheduled | retry } (required)")
@@ -151,6 +169,16 @@ var taskRunCmd = &cobra.Command{
 		$ asynq task run --queue=myqueue --id=f1720682-f5a6-4db1-8953-4f48ae541d0f`),
 }
 
+var taskEnqueueCmd = &cobra.Command{
+	Use:   "enqueue --type_name=footype --payload=barpayload",
+	Short: "Enqueue a task",
+	Args:  cobra.NoArgs,
+	Run:   taskEnqueue,
+	Example: heredoc.Doc(`
+		$ asynq task enqueue -t footype -l barpayload
+		$ asynq task enqueue -t footask -l barpayload --retry 3 --id f1720682-f5a6-4db1-8953-4f48ae541d0f --queue bazqueue --timeout 100s --deadline 2024-12-14T01:23:45Z --unique 100s --process_at 2024-12-14T01:22:05Z --process_in 100s --retention 5h --group baygroup`),
+}
+
 var taskArchiveAllCmd = &cobra.Command{
 	Use:   "archiveall --queue=<queue> --state=<state>",
 	Short: "Archive all tasks in the given state",
@@ -521,6 +549,95 @@ func taskRun(cmd *cobra.Command, args []string) {
 	fmt.Println("task is now pending")
 }
 
+func taskEnqueue(cmd *cobra.Command, args []string) {
+	typeName, err := cmd.Flags().GetString("type_name")
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	payload, err := cmd.Flags().GetString("payload")
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	// For all of the optional flags, we need to explicitly check whether they were set or
+	// not; for consistency we want to use the defaults set in composeOptions() rather than
+	// the ones in the flag definitions.
+	opts := []asynq.Option{}
+	if cmd.Flags().Changed("retry") {
+		retry, err := cmd.Flags().GetInt("retry")
+		if err != nil {
+			fmt.Printf("error: %v\n", err)
+			os.Exit(1)
+		}
+		opts = append(opts, asynq.MaxRetry(retry))
+	}
+
+	if cmd.Flags().Changed("queue") {
+		queue, err := cmd.Flags().GetString("queue")
+		if err != nil {
+			fmt.Printf("error: %v\n", err)
+			os.Exit(1)
+		}
+		opts = append(opts, asynq.Queue(queue))
+	}
+
+	if cmd.Flags().Changed("id") {
+		id, err := cmd.Flags().GetString("id")
+		if err != nil {
+			fmt.Printf("error: %v\n", err)
+			os.Exit(1)
+		}
+		opts = append(opts, asynq.TaskID(id))
+	}
+
+	if cmd.Flags().Changed("timeout") {
+		opts = append(opts, asynq.Timeout(getDuration(cmd, "timeout")))
+	}
+
+	if cmd.Flags().Changed("deadline") {
+		opts = append(opts, asynq.Deadline(getTime(cmd, "deadline")))
+	}
+
+	if cmd.Flags().Changed("unique") {
+		opts = append(opts, asynq.Unique(getDuration(cmd, "unique")))
+	}
+
+	if cmd.Flags().Changed("process_at") {
+		opts = append(opts, asynq.ProcessAt(getTime(cmd, "process_at")))
+	}
+
+	if cmd.Flags().Changed("process_in") {
+		opts = append(opts, asynq.ProcessIn(getDuration(cmd, "process_in")))
+	}
+
+	if cmd.Flags().Changed("retention") {
+		opts = append(opts, asynq.Retention(getDuration(cmd, "retention")))
+	}
+
+	if cmd.Flags().Changed("group") {
+		group, err := cmd.Flags().GetString("group")
+		if err != nil {
+			fmt.Printf("error: %v\n", err)
+			os.Exit(1)
+		}
+		opts = append(opts, asynq.Group(group))
+	}
+
+	c := createClient()
+	task := asynq.NewTask(typeName, []byte(payload), opts...)
+
+	taskInfo, err := c.Enqueue(task)
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("Enqueued task %s to queue %s\n", taskInfo.ID, taskInfo.Queue)
+}
+
 func taskArchiveAll(cmd *cobra.Command, args []string) {
 	qname, err := cmd.Flags().GetString("queue")
 	if err != nil {
@@ -653,3 +770,4 @@ func taskRunAll(cmd *cobra.Command, args []string) {
 	}
 	fmt.Printf("%d tasks are now pending\n", n)
 }
+

tools/go.mod

@@ -1,6 +1,6 @@
 module github.com/hibiken/asynq/tools
 
-go 1.20
+go 1.22
 
 require (
 	github.com/MakeNowJust/heredoc/v2 v2.0.1

tools/go.sum

@@ -61,6 +61,7 @@ github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5Kwzbycv
 github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0XL9UY=
+github.com/frankban/quicktest v1.14.4/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -176,9 +177,11 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxv
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lucasb-eyer/go-colorful v1.2.0 h1:1nnpGOrhyZZuNyfu1QjKiUICQ74+3FNCN69Aj6K7nkY=
 github.com/lucasb-eyer/go-colorful v1.2.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
 github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzRKO2BQ4=
@@ -266,6 +269,7 @@ github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzG
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=

x/go.mod

@@ -1,6 +1,6 @@
 module github.com/hibiken/asynq/x
 
-go 1.20
+go 1.22
 
 require (
 	github.com/google/uuid v1.4.0

x/go.sum

@@ -10,8 +10,10 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -20,6 +22,7 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
+github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
@@ -49,6 +52,7 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
@@ -66,9 +70,11 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxv
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -105,6 +111,7 @@ github.com/redis/go-redis/v9 v9.3.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=