From d27afc42f1258d2fbf9a7b8718f9b34b8b9970da Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Tue, 16 Feb 2021 15:10:24 -0600
Subject: [PATCH] don't cancel the context if replicas should be attempted

---
 executor.go | 21 ++++++++++++++++-----
 1 file changed, 16 insertions(+), 5 deletions(-)

diff --git a/executor.go b/executor.go
index 93619c432..bde668f7e 100644
--- a/executor.go
+++ b/executor.go
@@ -46,6 +46,8 @@ const (
 
 	columnLabel = "col"
 	rowLabel    = "row"
+
+	errConnectionRefused = "connect: connection refused"
 )
 
 // executor recursively executes calls in a PQL query across all shards.
@@ -3501,7 +3503,6 @@ func (e *executor) executeGroupByShard(ctx context.Context, qcx *Qcx, index stri
 }
 
 func (e *executor) executeRows(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) {
-
 	// Fetch field name from argument.
 	// Check "field" first for backwards compatibility.
 	// TODO: remove at Pilosa 2.0
@@ -5579,7 +5580,12 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 			// On error retry against remaining nodes. If an error returns then
 			// the context will cancel and cause all open goroutines to return.
 
-			if resp.err != nil {
+			// We distinguish here between an error which indicates that the
+			// node is not available (and therefore we need to failover to a
+			// replica) and a valid error from a healthy node. In the case of
+			// the latter, there's no need to retry a replica, we should trust
+			// the error from the healthy node and return that immediately.
+			if resp.err != nil && strings.Contains(resp.err.Error(), errConnectionRefused) {
 				// Filter out unavailable nodes.
 				nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
 
@@ -5587,15 +5593,16 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 				if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
 					return nil, resp.err
 				} else if err != nil {
-					return nil, errors.Wrap(err, "calling mapper")
+					return nil, errors.Wrap(err, "mapping on secondary node")
 				}
 				continue
+			} else if resp.err != nil {
+				return nil, errors.Wrap(resp.err, "mapping on primary node")
 			}
 
 			// Reduce value.
 			result = reduceFn(ctx, result, resp.result)
 			if err, ok := result.(error); ok {
-				cancel()
 				return nil, err
 			}
 
@@ -5689,7 +5696,11 @@ func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch cha
 				// The cancel coming after the above send is intentional.
 				// We want to report the actual error that happened
 				// before we cause anything to return "context canceled".
-				if resp.err != nil {
+				// Also, we only want to call cancel if the error is from a
+				// healthy node. If the error is "connection refused" because
+				// the node is unavailable, we don't call cancel, and instead
+				// let the mapper continue trying replica nodes.
+				if resp.err != nil && !strings.Contains(resp.err.Error(), errConnectionRefused) {
 					cancel()
 				}
 			}