From 99420b564e4f3c71aef9702d1f14f6d9132ba67b Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Mon, 20 Jul 2020 12:27:58 -0500
Subject: [PATCH] Handle cluster shutdown during a resize

The new logic to send resize instructions more makes it easier
to hit this, but it's probably always been a theoretically possible
bug to hit: If you are shutting a cluster down, then you stop accepting
connections, which means that if you have an existing resize job, you
can't get responses for it. Which means that the other nodes will
fail to notify you of the success or failure of resize instructions,
so the code waiting on the resize job's status waits forever.

When closing, we bail immediately on that; we don't need to wait for
those notifications. We still have a buffer, and a reasonable confidence
that we'll never write more than one result status, so if one of them
*does* somehow show up and cause the job to have a status,
writing the status won't block.
---
 cluster.go | 25 +++++++++++++++++++------
 1 file changed, 19 insertions(+), 6 deletions(-)

diff --git a/cluster.go b/cluster.go
index 429c1c970..479a5c6f2 100644
--- a/cluster.go
+++ b/cluster.go
@@ -1219,7 +1219,12 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
 
 	// Wait for the resizeJob to finish or be aborted.
 	c.logger.Printf("wait for jobResult")
-	jobResult := <-j.result
+	var jobResult string
+	select {
+	case <-c.closing:
+		return errors.New("cluster shut down during resize")
+	case jobResult = <-j.result:
+	}
 
 	// Make sure j.run() didn't return an error.
 	if eg.Wait() != nil {
@@ -1354,6 +1359,9 @@ func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJo
 // the resize instructions to other nodes in the cluster.
 func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) {
 	j := newResizeJob(c.nodes, nodeAction.node, nodeAction.action)
+	// A *new* node which is being added needs a schema update even if
+	// there's no data to send it.
+	var sendSchemaToNewNode string
 	j.Broadcaster = c.broadcaster
 
 	// toCluster is a clone of Cluster with the new node added/removed for comparison.
@@ -1366,6 +1374,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 		toCluster.removeNodeBasicSorted(nodeAction.node.ID)
 	} else if nodeAction.action == resizeJobActionAdd {
 		toCluster.addNodeBasicSorted(nodeAction.node)
+		sendSchemaToNewNode = nodeAction.node.ID
 	}
 
 	indexes := c.holder.Indexes()
@@ -1431,11 +1440,15 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 	}
 
 	for _, node := range toCluster.nodes {
-		// We may send a resize instruction that has no sources that
-		// the node needs to read from -- for instance, if there's no
-		// data in any fragments it would process. But it still needs
-		// to get the NodeStatus to pick up the schema so it knows
-		// about existing indexes.
+		dataToSend := len(fragmentSourcesByNode[node.ID]) != 0 || len(translationSourcesByNode[node.ID]) != 0
+		// If we're adding a new node, that node needs to get a resize
+		// instruction even if there's no data it needs to read.
+		// Existing nodes already got the schema and are assumed to be
+		// up to date on it.
+		if !dataToSend && node.ID != sendSchemaToNewNode {
+			j.IDs[node.ID] = true
+			continue
+		}
 		instr := &ResizeInstruction{
 			JobID:              j.ID,
 			Node:               toCluster.unprotectedNodeByID(node.ID),