Fix execTxn to prioritize pipeline-level errors over results

Pipeline-level errors (connection/transport failures) were only checked
when results were empty. Now checked first to avoid silently ignoring
transport errors when partial results are returned.

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

Author: bootjp

proxy/proxy.go

@@ -348,13 +348,14 @@ func (p *ProxyServer) execTxn(conn redcon.Conn, state *proxyConnState) {
 	cmds = append(cmds, []any{"EXEC"})
 
 	results, err := p.dual.Primary().Pipeline(ctx, cmds)
-	if len(results) > 0 {
+	if err != nil {
+		// Pipeline-level error (connection/transport failure) takes precedence.
+		writeRedisError(conn, err)
+	} else if len(results) > 0 {
 		// Write the EXEC result (last command in the pipeline).
 		lastResult := results[len(results)-1]
 		resp, rErr := lastResult.Result()
 		writeResponse(conn, resp, rErr)
-	} else if err != nil {
-		writeRedisError(conn, err)
 	}
 
 	// Async replay to secondary (bounded)

proxy/pubsub.go

@@ -369,12 +369,13 @@ func (s *pubsubSession) execTxn() {
 	results, err := s.proxy.dual.Primary().Pipeline(ctx, cmds)
 
 	s.mu.Lock()
-	if len(results) > 0 {
+	if err != nil {
+		// Pipeline-level error (connection/transport failure) takes precedence.
+		writeRedisError(s.dconn, err)
+	} else if len(results) > 0 {
 		lastResult := results[len(results)-1]
 		resp, rErr := lastResult.Result()
 		writeResponse(s.dconn, resp, rErr)
-	} else if err != nil {
-		writeRedisError(s.dconn, err)
 	}
 	_ = s.dconn.Flush()
 	s.mu.Unlock()