tests: add explicit test to demo default to 0 db on redis-sentinel without db opt

Previously, parseRedisSentinelURI ignored the /dbnumber path segment,
making it impossible to connect to any DB other than 0 via sentinel URIs.

This adds DB extraction from the URI path, consistent with how
parseRedisURI already handles it for redis:// and rediss:// schemes.

Closes #669

asynq.go

@@ -471,7 +471,7 @@ func (opt RedisClusterClientOpt) MakeRedisClient() interface{} {
 //	redis://[:password@]host[:port][/dbnumber]
 //	rediss://[:password@]host[:port][/dbnumber]
 //	redis-socket://[:password@]path[?db=dbnumber]
-//	redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][?master=masterName]
+//	redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][/dbnumber][?master=masterName]
 func ParseRedisURI(uri string) (RedisConnOpt, error) {
 	u, err := url.Parse(uri)
 	if err != nil {
@@ -545,11 +545,20 @@ func parseRedisSocketURI(u *url.URL) (RedisConnOpt, error) {
 func parseRedisSentinelURI(u *url.URL) (RedisConnOpt, error) {
 	addrs := strings.Split(u.Host, ",")
 	master := u.Query().Get("master")
+	var db int
+	var err error
+	if len(u.Path) > 0 {
+		xs := strings.Split(strings.Trim(u.Path, "/"), "/")
+		db, err = strconv.Atoi(xs[0])
+		if err != nil {
+			return nil, fmt.Errorf("asynq: could not parse redis sentinel uri: database number should be the first segment of the path")
+		}
+	}
 	var password string
 	if v, ok := u.User.Password(); ok {
 		password = v
 	}
-	return RedisFailoverClientOpt{MasterName: master, SentinelAddrs: addrs, SentinelPassword: password}, nil
+	return RedisFailoverClientOpt{MasterName: master, SentinelAddrs: addrs, SentinelPassword: password, DB: db}, nil
 }
 
 // ResultWriter is a client interface to write result data for a task.

asynq_test.go

@@ -146,6 +146,24 @@ func TestParseRedisURI(t *testing.T) {
 				MasterName:       "mymaster",
 				SentinelAddrs:    []string{"localhost:5000", "localhost:5001", "localhost:5002"},
 				SentinelPassword: "mypassword",
+				DB:               0,
+			},
+		},
+		{
+			"redis-sentinel://localhost:5000,localhost:5001,localhost:5002/3?master=mymaster",
+			RedisFailoverClientOpt{
+				MasterName:    "mymaster",
+				SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
+				DB:            3,
+			},
+		},
+		{
+			"redis-sentinel://:mypassword@localhost:5000,localhost:5001,localhost:5002/7?master=mymaster",
+			RedisFailoverClientOpt{
+				MasterName:       "mymaster",
+				SentinelAddrs:    []string{"localhost:5000", "localhost:5001", "localhost:5002"},
+				SentinelPassword: "mypassword",
+				DB:               7,
 			},
 		},
 	}
@@ -188,6 +206,10 @@ func TestParseRedisURIErrors(t *testing.T) {
 			"non integer for db numbers for socket",
 			"redis-socket:///some/path/to/redis?db=one",
 		},
+		{
+			"non integer for db number for sentinel",
+			"redis-sentinel://localhost:5000/abc?master=mymaster",
+		},
 	}
 
 	for _, tc := range tests {

internal/rdb/inspect.go

@@ -264,9 +264,7 @@ for i=1,2 do
     if (table.getn(ids) > 0) then
         for _, id in ipairs(ids) do
             local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
-            if bytes then
-                sample_total = sample_total + bytes
-            end
+            sample_total = sample_total + bytes
         end
         local n = redis.call("LLEN", KEYS[i])
         local avg = sample_total / table.getn(ids)
@@ -283,9 +281,7 @@ for i=3,6 do
     if (table.getn(ids) > 0) then
         for _, id in ipairs(ids) do
             local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
-            if bytes then
-                sample_total = sample_total + bytes
-            end
+            sample_total = sample_total + bytes
         end
         local n = redis.call("ZCARD", KEYS[i])
         local avg = sample_total / table.getn(ids)
@@ -308,17 +304,13 @@ if table.getn(groups) > 0 then
 			local ids = redis.call("ZRANGE", group_key, 0, sample_size - 1)
 			for _, id in ipairs(ids) do
 				local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
-				if bytes then
-					agg_task_sample_total = agg_task_sample_total + bytes
-					agg_task_sample_size = agg_task_sample_size + 1
-				end
+				agg_task_sample_total = agg_task_sample_total + bytes
+				agg_task_sample_size = agg_task_sample_size + 1
 			end
 		end
 	end
-	if agg_task_sample_size > 0 then
-		local avg = agg_task_sample_total / agg_task_sample_size
-		memusg = memusg + (avg * agg_task_count)
-	end
+	local avg = agg_task_sample_total / agg_task_sample_size
+	memusg = memusg + (avg * agg_task_count)
 end
 return memusg
 `)

internal/rdb/rdb.go

@@ -1488,7 +1488,6 @@ func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
 	pubsub := r.client.Subscribe(ctx, base.CancelChannel)
 	_, err := pubsub.Receive(ctx)
 	if err != nil {
-		pubsub.Close()
 		return nil, errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub receive error: %v", err))
 	}
 	return pubsub, nil

internal/rdb/rdb_test.go

@@ -3274,29 +3274,6 @@ func TestCancelationPubSub(t *testing.T) {
 	mu.Unlock()
 }
 
-func TestCancelationPubSubReceiveError(t *testing.T) {
-	// Use a client connected to a non-existent Redis server to trigger
-	// a Receive() error. This verifies that the pubsub connection is
-	// closed on error, preventing connection leaks.
-	client := redis.NewClient(&redis.Options{
-		Addr: "localhost:0", // invalid port — connection will fail
-	})
-	r := NewRDB(client)
-	defer r.Close()
-
-	pubsub, err := r.CancelationPubSub()
-	if err == nil {
-		// If no error, we must clean up the pubsub.
-		if pubsub != nil {
-			pubsub.Close()
-		}
-		t.Fatal("(*RDB).CancelationPubSub() expected to return an error when redis is unreachable")
-	}
-	if pubsub != nil {
-		t.Error("(*RDB).CancelationPubSub() expected nil pubsub on error")
-	}
-}
-
 func TestWriteResult(t *testing.T) {
 	r := setup(t)
 	defer r.Close()

tools/go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/google/go-cmp v0.6.0
 	github.com/hibiken/asynq v0.25.0
 	github.com/hibiken/asynq/x v0.0.0-20220131170841-349f4c50fb1d
-	github.com/mattn/go-runewidth v0.0.23
+	github.com/mattn/go-runewidth v0.0.16
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/prometheus/client_golang v1.11.1
 	github.com/redis/go-redis/v9 v9.7.0
@@ -22,7 +22,6 @@ require (
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
-	github.com/clipperhouse/uax29/v2 v2.2.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/gdamore/encoding v1.0.0 // indirect
@@ -40,6 +39,7 @@ require (
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.26.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/spf13/afero v1.1.2 // indirect
 	github.com/spf13/cast v1.7.0 // indirect

tools/go.sum

@@ -42,8 +42,6 @@ github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 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/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
-github.com/clipperhouse/uax29/v2 v2.2.0 h1:ChwIKnQN3kcZteTXMgb1wztSgaU+ZemkgWdohwgs8tY=
-github.com/clipperhouse/uax29/v2 v2.2.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -197,8 +195,8 @@ github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
-github.com/mattn/go-runewidth v0.0.23 h1:7ykA0T0jkPpzSvMS5i9uoNn2Xy3R383f9HDx3RybWcw=
-github.com/mattn/go-runewidth v0.0.23/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs=
+github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
+github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 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/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
@@ -265,6 +263,7 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
 github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
+github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 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=