fix(pubsub): avoid panics on Channel conflict and Ring empty subscribe

.gitattributes

@@ -0,0 +1,6 @@
+*.go text eol=lf
+*.mod text eol=lf
+*.sum text eol=lf
+*.yml text eol=lf
+*.yaml text eol=lf
+*.md text eol=lf

pubsub.go

@@ -37,6 +37,11 @@ type PubSub struct {
 	closed bool
 	exit   chan struct{}
 
+	// stickyErr is set when PubSub is constructed in a failed state (e.g. Ring
+	// shard lookup failure). Subsequent operations return this error instead of
+	// panicking at construction time.
+	stickyErr error
+
 	cmd *Cmd
 
 	chOnce sync.Once
@@ -75,6 +80,9 @@ func (c *PubSub) conn(ctx context.Context, newChannels []string) (*pool.Conn, er
 	if c.closed {
 		return nil, pool.ErrClosed
 	}
+	if c.stickyErr != nil {
+		return nil, c.stickyErr
+	}
 	if c.cn != nil {
 		return c.cn, nil
 	}
@@ -217,6 +225,10 @@ func (c *PubSub) Close() error {
 		return pool.ErrClosed
 	}
 	c.closed = true
+	// Override construction-time stickyErr (e.g. Ring failedPubSub) so
+	// Channel receivers observe pool.ErrClosed and exit cleanly.
+	// Matches SingleConnPool.Close which sets stickyErr = ErrClosed.
+	c.stickyErr = pool.ErrClosed
 	close(c.exit)
 
 	// Call cleanup callback if set
@@ -575,8 +587,15 @@ func (c *PubSub) Channel(opts ...ChannelOption) <-chan *Message {
 		c.msgCh.initMsgChan()
 	})
 	if c.msgCh == nil {
-		err := fmt.Errorf("redis: Channel can't be called after ChannelWithSubscriptions")
-		panic(err)
+		// Already using ChannelWithSubscriptions — return a closed channel
+		// instead of panicking so callers can recover (issue #3761).
+		// Use Background: getContext reads c.cmd which the other channel's
+		// Receive loop may write concurrently (data race under -race).
+		internal.Logger.Printf(context.Background(),
+			"redis: Channel can't be called after ChannelWithSubscriptions")
+		ch := make(chan *Message)
+		close(ch)
+		return ch
 	}
 	return c.msgCh.msgCh
 }
@@ -600,8 +619,15 @@ func (c *PubSub) ChannelWithSubscriptions(opts ...ChannelOption) <-chan interfac
 		c.allCh.initAllChan()
 	})
 	if c.allCh == nil {
-		err := fmt.Errorf("redis: ChannelWithSubscriptions can't be called after Channel")
-		panic(err)
+		// Already using Channel — return a closed channel instead of panicking
+		// so callers can recover (issue #3761).
+		// Use Background: getContext reads c.cmd which the other channel's
+		// Receive loop may write concurrently (data race under -race).
+		internal.Logger.Printf(context.Background(),
+			"redis: ChannelWithSubscriptions can't be called after Channel")
+		ch := make(chan interface{})
+		close(ch)
+		return ch
 	}
 	return c.allCh.allCh
 }

pubsub_sticky_error_test.go

@@ -0,0 +1,67 @@
+package redis_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func TestRingSubscribeEmptyChannelsNoPanic(t *testing.T) {
+	ring := redis.NewRing(&redis.RingOptions{
+		Addrs: map[string]string{"shard1": "localhost:6379"},
+	})
+	defer ring.Close()
+
+	pubsub := ring.Subscribe(context.Background())
+	if pubsub == nil {
+		t.Fatal("expected non-nil PubSub")
+	}
+	// Receive should surface sticky error, not panic.
+	_, err := pubsub.Receive(context.Background())
+	if err == nil {
+		t.Fatal("expected sticky error from empty Subscribe")
+	}
+	if err := pubsub.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	// After Close, sticky construction error is overridden by ErrClosed so
+	// Channel/initMsgChan can terminate (pool.ErrClosed is the exit signal).
+	_, err = pubsub.Receive(context.Background())
+	if err != redis.ErrClosed {
+		t.Fatalf("after Close want redis.ErrClosed, got %v", err)
+	}
+}
+
+func TestPubSubChannelMutualExclusionNoPanic(t *testing.T) {
+	// Construct via a client that may not be reachable — Subscribe without
+	// channels just builds a PubSub handle.
+	t.Run("ChannelAfterChannelWithSubscriptions", func(t *testing.T) {
+		client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1"})
+		defer client.Close()
+		pubsub := client.Subscribe(context.Background())
+		defer pubsub.Close()
+
+		_ = pubsub.ChannelWithSubscriptions()
+		ch := pubsub.Channel()
+		// Conflicting call returns an already-closed channel (no panic).
+		msg, ok := <-ch
+		if ok || msg != nil {
+			t.Fatalf("expected closed channel from conflicting Channel() call, got msg=%v ok=%v", msg, ok)
+		}
+	})
+
+	t.Run("ChannelWithSubscriptionsAfterChannel", func(t *testing.T) {
+		client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1"})
+		defer client.Close()
+		pubsub := client.Subscribe(context.Background())
+		defer pubsub.Close()
+
+		_ = pubsub.Channel()
+		ch := pubsub.ChannelWithSubscriptions()
+		msg, ok := <-ch
+		if ok || msg != nil {
+			t.Fatalf("expected closed channel from conflicting ChannelWithSubscriptions() call, got msg=%v ok=%v", msg, ok)
+		}
+	})
+}

pubsub_test.go

@@ -674,4 +674,17 @@ var _ = Describe("PubSub", func() {
 		Expect(msg.Channel).To(Equal("mychannel"))
 		Expect(msg.Payload).To(Equal(text))
 	})
+
+	It("closes Channel when PubSub with sticky error is closed", func() {
+		ring := redis.NewRing(&redis.RingOptions{})
+		defer ring.Close()
+
+		// Ring.Subscribe with no channels returns a failedPubSub with stickyErr
+		pubsub := ring.Subscribe(ctx)
+
+		ch := pubsub.Channel()
+		Expect(pubsub.Close()).To(Succeed())
+
+		Eventually(ch, 5*time.Second).Should(BeClosed())
+	})
 })

ring.go

@@ -705,43 +705,54 @@ func (c *Ring) Len() int {
 	return c.sharding.Len()
 }
 
+// failedPubSub returns a PubSub that surfaces err on every operation instead of
+// panicking at construction (issue #3761).
+func (c *Ring) failedPubSub(err error) *PubSub {
+	if err == nil {
+		err = fmt.Errorf("redis: pubsub failed")
+	}
+	pubsub := &PubSub{
+		opt:       c.opt.clientOptions(),
+		stickyErr: err,
+	}
+	pubsub.init()
+	return pubsub
+}
+
 // Subscribe subscribes the client to the specified channels.
 func (c *Ring) Subscribe(ctx context.Context, channels ...string) *PubSub {
 	if len(channels) == 0 {
-		panic("at least one channel is required")
+		return c.failedPubSub(fmt.Errorf("redis: at least one channel is required"))
 	}
 
 	shard, err := c.sharding.GetByKey(channels[0])
 	if err != nil {
-		// TODO: return PubSub with sticky error
-		panic(err)
+		return c.failedPubSub(err)
 	}
 	return shard.Client.Subscribe(ctx, channels...)
 }
 
 // PSubscribe subscribes the client to the given patterns.
 func (c *Ring) PSubscribe(ctx context.Context, channels ...string) *PubSub {
 	if len(channels) == 0 {
-		panic("at least one channel is required")
+		return c.failedPubSub(fmt.Errorf("redis: at least one channel is required"))
 	}
 
 	shard, err := c.sharding.GetByKey(channels[0])
 	if err != nil {
-		// TODO: return PubSub with sticky error
-		panic(err)
+		return c.failedPubSub(err)
 	}
 	return shard.Client.PSubscribe(ctx, channels...)
 }
 
 // SSubscribe Subscribes the client to the specified shard channels.
 func (c *Ring) SSubscribe(ctx context.Context, channels ...string) *PubSub {
 	if len(channels) == 0 {
-		panic("at least one channel is required")
+		return c.failedPubSub(fmt.Errorf("redis: at least one channel is required"))
 	}
 	shard, err := c.sharding.GetByKey(channels[0])
 	if err != nil {
-		// TODO: return PubSub with sticky error
-		panic(err)
+		return c.failedPubSub(err)
 	}
 	return shard.Client.SSubscribe(ctx, channels...)
 }