Add tests

Author: ofekshenawa

internal/routing/aggregator.go

@@ -218,6 +218,9 @@ func (a *AggMinAggregator) Finish() (interface{}, error) {
 	if a.firstErr != nil {
 		return nil, a.firstErr
 	}
+	if !a.hasResult {
+		return nil, fmt.Errorf("redis: no valid results to aggregate for min operation")
+	}
 	return a.min, nil
 }
 
@@ -264,6 +267,9 @@ func (a *AggMaxAggregator) Finish() (interface{}, error) {
 	if a.firstErr != nil {
 		return nil, a.firstErr
 	}
+	if !a.hasResult {
+		return nil, fmt.Errorf("redis: no valid results to aggregate for max operation")
+	}
 	return a.max, nil
 }
 
@@ -312,6 +318,9 @@ func (a *AggLogicalAndAggregator) Finish() (interface{}, error) {
 	if a.firstErr != nil {
 		return nil, a.firstErr
 	}
+	if !a.hasResult {
+		return nil, fmt.Errorf("redis: no valid results to aggregate for logical AND operation")
+	}
 	return a.result, nil
 }
 
@@ -360,6 +369,9 @@ func (a *AggLogicalOrAggregator) Finish() (interface{}, error) {
 	if a.firstErr != nil {
 		return nil, a.firstErr
 	}
+	if !a.hasResult {
+		return nil, fmt.Errorf("redis: no valid results to aggregate for logical OR operation")
+	}
 	return a.result, nil
 }
 

osscluster_router.go

@@ -312,12 +312,23 @@ func (c *ClusterClient) executeParallel(ctx context.Context, cmd Cmder, nodes []
 		close(results)
 	}()
 
-	// Collect results
+	// Collect results and check for errors
 	cmds := make([]Cmder, 0, len(nodes))
+	var firstErr error
+
 	for result := range results {
+		if result.err != nil && firstErr == nil {
+			firstErr = result.err
+		}
 		cmds = append(cmds, result.cmd)
 	}
 
+	// If there was an error and no policy specified, fail fast
+	if firstErr != nil && (policy == nil || policy.Response == routing.RespDefaultKeyless) {
+		cmd.SetErr(firstErr)
+		return firstErr
+	}
+
 	return c.aggregateResponses(cmd, cmds, policy)
 }
 
@@ -342,11 +353,11 @@ func (c *ClusterClient) aggregateMultiSlotResults(ctx context.Context, cmd Cmder
 		return firstErr
 	}
 
-	return c.aggregateKeyedResponses(ctx, cmd, keyedResults, keyOrder, policy)
+	return c.aggregateKeyedResponses(cmd, keyedResults, keyOrder, policy)
 }
 
 // aggregateKeyedResponses aggregates responses while preserving key order
-func (c *ClusterClient) aggregateKeyedResponses(ctx context.Context, cmd Cmder, keyedResults map[string]Cmder, keyOrder []string, policy *routing.CommandPolicy) error {
+func (c *ClusterClient) aggregateKeyedResponses(cmd Cmder, keyedResults map[string]Cmder, keyOrder []string, policy *routing.CommandPolicy) error {
 	if len(keyedResults) == 0 {
 		return fmt.Errorf("redis: no results to aggregate")
 	}