Refactor error handling in pubsub session

Author: bootjp

proxy/proxy.go

@@ -231,7 +231,7 @@ func (p *ProxyServer) startPubSubSession(conn redcon.Conn, cmdName string, args
 	}
 	if err != nil {
 		upstream.Close()
-		conn.WriteError("ERR " + err.Error())
+		writeRedisError(conn, err)
 		return
 	}
 

proxy/pubsub.go

@@ -2,6 +2,7 @@ package proxy
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"strings"
@@ -326,7 +327,7 @@ func (s *pubsubSession) reenterPubSub(cmdName string, args [][]byte) {
 	}
 	if err != nil {
 		upstream.Close()
-		s.writeError("ERR " + err.Error())
+		s.writeRedisError(err)
 		return
 	}
 
@@ -401,6 +402,7 @@ func (s *pubsubSession) handleSubscribe(args [][]byte) {
 	channels := byteSlicesToStrings(args[1:])
 	if err := s.upstream.Subscribe(context.Background(), channels...); err != nil {
 		s.logger.Warn("upstream subscribe failed", "err", err)
+		s.writeRedisError(err)
 		return
 	}
 	s.mu.Lock()
@@ -424,6 +426,7 @@ func (s *pubsubSession) handlePSubscribe(args [][]byte) {
 	pats := byteSlicesToStrings(args[1:])
 	if err := s.upstream.PSubscribe(context.Background(), pats...); err != nil {
 		s.logger.Warn("upstream psubscribe failed", "err", err)
+		s.writeRedisError(err)
 		return
 	}
 	s.mu.Lock()
@@ -563,6 +566,17 @@ func (s *pubsubSession) writeError(msg string) {
 	_ = s.dconn.Flush()
 }
 
+// writeRedisError writes an upstream error, preserving redis.Error prefixes verbatim
+// and normalizing other errors to "ERR ..." (matching writeRedisError in proxy.go).
+func (s *pubsubSession) writeRedisError(err error) {
+	var redisErr redis.Error
+	if errors.As(err, &redisErr) {
+		s.writeError(redisErr.Error())
+		return
+	}
+	s.writeError("ERR " + err.Error())
+}
+
 func (s *pubsubSession) writeString(msg string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

proxy/pubsub_test.go

@@ -484,41 +484,6 @@ func TestPubSub_SelectAuthSilentlyAccepted(t *testing.T) {
 	}
 }
 
-// TestPubSub_DuplicateSubscribeDoesNotOvercount verifies that subscribing to the
-// same channel multiple times doesn't inflate the subscription count.
-func TestPubSub_DuplicateSubscribeDoesNotOvercount(t *testing.T) {
-	s := newTestSession(newMockDetachedConn())
-
-	// First subscribe
-	s.channelSet["ch1"] = struct{}{}
-	assert.Equal(t, 1, s.subCount())
-
-	// Duplicate subscribe (idempotent)
-	s.channelSet["ch1"] = struct{}{}
-	assert.Equal(t, 1, s.subCount(), "duplicate subscribe must not increase count")
-
-	// Add new channel
-	s.channelSet["ch2"] = struct{}{}
-	assert.Equal(t, 2, s.subCount())
-}
-
-// TestPubSub_UnsubscribeNonExistent verifies that unsubscribing from a channel
-// that was never subscribed does not affect the count.
-func TestPubSub_UnsubscribeNonExistent(t *testing.T) {
-	s := newTestSession(newMockDetachedConn())
-
-	s.channelSet["ch1"] = struct{}{}
-	s.channelSet["ch2"] = struct{}{}
-
-	// Delete non-existent: no panic, no effect
-	delete(s.channelSet, "never-subscribed")
-	assert.Equal(t, 2, s.subCount())
-
-	// Delete existing
-	delete(s.channelSet, "ch1")
-	assert.Equal(t, 1, s.subCount())
-}
-
 // TestPubSub_CleanupClosesUpstream verifies that cleanup closes upstream and dconn.
 func TestPubSub_CleanupClosesUpstream(t *testing.T) {
 	dconn := newMockDetachedConn()