Fix truncateValue []byte marker, bounded cleanup wait, mock Close race

- Add "...(truncated)" marker to []byte path in truncateValue
- Add bounded timeout (5s) to cleanup() wait on forwardMessages to
  prevent hanging on stuck client sockets
- Guard mockDetachedConn.Close() with mutex to fix data race under -race

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bootjp

proxy/pubsub.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 	"github.com/tidwall/redcon"
@@ -28,6 +29,10 @@ const (
 	cmdDiscard      = "DISCARD"
 	cmdPing         = "PING"
 	cmdQuit         = "QUIT"
+
+	// cleanupFwdTimeout bounds the wait for forwardMessages to exit during cleanup.
+	// If the client socket is stuck, we don't want to block indefinitely.
+	cleanupFwdTimeout = 5 * time.Second
 )
 
 // pubsubSession manages a single client's detached connection.
@@ -77,7 +82,13 @@ func (s *pubsubSession) cleanup() {
 	}
 	s.mu.Unlock()
 	if s.fwdDone != nil {
-		<-s.fwdDone
+		// Bounded wait: if forwardMessages is stuck on a slow/dead client socket,
+		// don't block cleanup indefinitely.
+		select {
+		case <-s.fwdDone:
+		case <-time.After(cleanupFwdTimeout):
+			s.logger.Warn("forwardMessages did not exit within timeout, proceeding with cleanup")
+		}
 	}
 	s.dconn.Close()
 }

proxy/pubsub_test.go

@@ -52,7 +52,12 @@ func (m *mockDetachedConn) ReadCommand() (redcon.Command, error) {
 }
 
 func (m *mockDetachedConn) Flush() error { return nil }
-func (m *mockDetachedConn) Close() error { m.closed = true; return nil }
+func (m *mockDetachedConn) Close() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.closed = true
+	return nil
+}
 
 func (m *mockDetachedConn) RemoteAddr() string { return "127.0.0.1:9999" }
 

proxy/sentry.go

@@ -149,9 +149,9 @@ func truncateValue(v any) string {
 		s = tv
 	case []byte:
 		if len(tv) > maxSentryValueLen {
-			tv = tv[:maxSentryValueLen]
+			return string(tv[:maxSentryValueLen]) + "...(truncated)"
 		}
-		s = string(tv)
+		return string(tv)
 	default:
 		s = fmt.Sprintf("%v", v)
 	}