ensure that holder opens before node is deemed ready

cluster.go

@@ -286,13 +286,13 @@ func (c *Cluster) setState(state string) {
 		// - ClusterStateStarting
 	}
 
+	c.logger().Printf("Change cluster state from %s to %s", c.State, state)
 	c.State = state
 }
 
-func (c *Cluster) setNodeState(state string) {
+func (c *Cluster) SetNodeState(state string) error {
 	if c.IsCoordinator() {
-		c.Topology.nodeStates[c.URI] = state
-		return
+		return c.ReceiveNodeState(c.URI, state)
 	}
 
 	// Send node state to coordinator.
@@ -305,8 +305,10 @@ func (c *Cluster) setNodeState(state string) {
 		URI: c.Coordinator,
 	}
 	if err := c.Broadcaster.SendTo(node, ns); err != nil {
-		c.logger().Printf("sending node state error: err=%s", err)
+		return fmt.Errorf("sending node state error: err=%s", err)
 	}
+
+	return nil
 }
 
 func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
@@ -691,18 +693,15 @@ func (c *Cluster) Open() error {
 
 	// Only the coordinator needs to consider the .topology file.
 	if c.IsCoordinator() {
-		state, err := c.considerTopology()
+		err := c.considerTopology()
 		if err != nil {
 			return fmt.Errorf("considerTopology: %v", err)
 		}
-		// Add the local node to the cluster and update state.
-		c.AddNode(c.URI)
-		c.setState(state)
-	} else {
-		// Add the local node to the cluster.
-		c.AddNode(c.URI)
 	}
 
+	// Add the local node to the cluster.
+	c.AddNode(c.URI)
+
 	// Start the EventReceiver.
 	if err := c.EventReceiver.Start(c); err != nil {
 		return fmt.Errorf("starting EventReceiver: %v", err)
@@ -921,6 +920,10 @@ func (c *Cluster) CompleteCurrentJob(state string) error {
 // FollowResizeInstruction is run by any node that receives a ResizeInstruction.
 func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error {
 	go func() {
+
+		// Make sure the holder has opened.
+		<-c.Holder.opened
+
 		// Prepare the return message.
 		complete := &internal.ResizeInstructionComplete{
 			JobID: instr.JobID,
@@ -1273,6 +1276,11 @@ func (c *Cluster) loadTopology() error {
 
 // saveTopology writes the current topology to disk.
 func (c *Cluster) saveTopology() error {
+
+	if err := os.MkdirAll(c.Path, 0777); err != nil {
+		return err
+	}
+
 	if buf, err := proto.Marshal(encodeTopology(c.Topology)); err != nil {
 		return err
 	} else if err := ioutil.WriteFile(filepath.Join(c.Path, ".topology"), buf, 0666); err != nil {
@@ -1301,25 +1309,25 @@ func decodeTopology(topology *internal.Topology) (*Topology, error) {
 	return t, nil
 }
 
-func (c *Cluster) considerTopology() (string, error) {
-	// If there is no .topology file, it's safe to go to state NORMAL.
+func (c *Cluster) considerTopology() error {
+	// If there is no .topology file, it's safe to proceed.
 	if len(c.Topology.NodeSet) == 0 {
-		return ClusterStateNormal, nil
+		return nil
 	}
 
 	// The local node (coordinator) must be in the .topology.
 	if !c.Topology.ContainsURI(c.Coordinator) {
-		return "", fmt.Errorf("coordinator %s is not in topology: %v", c.Coordinator, c.Topology.NodeSet)
+		return fmt.Errorf("coordinator %s is not in topology: %v", c.Coordinator, c.Topology.NodeSet)
 	}
 
-	// If local node is the only thing in .topology, continue to state NORMAL.
-	if len(c.Topology.NodeSet) == 1 {
-		return ClusterStateNormal, nil
-	}
+	// If local node is the only thing in .topology, continue.
+	//if len(c.Topology.NodeSet) == 1 {
+	//	return nil
+	//}
 
 	// Keep the cluster in state "STARTING" until hearing from all nodes.
 	// Topology contains 2+ hosts.
-	return ClusterStateStarting, nil
+	return nil
 }
 
 // ReceiveEvent represents an implementation of EventHandler.

holder.go

@@ -46,6 +46,9 @@ type Holder struct {
 	indexes map[string]*Index
 	hasData bool
 
+	// opened channel is closed once Open() completes.
+	opened chan struct{}
+
 	Broadcaster Broadcaster
 	// Close management
 	wg      sync.WaitGroup
@@ -69,6 +72,8 @@ func NewHolder() *Holder {
 		indexes: make(map[string]*Index),
 		closing: make(chan struct{}, 0),
 
+		opened: make(chan struct{}),
+
 		Broadcaster: NopBroadcaster,
 		Stats:       NopStatsClient,
 
@@ -156,6 +161,8 @@ func (h *Holder) Open() error {
 	go func() { defer h.wg.Done(); h.monitorCacheFlush() }()
 
 	h.Stats.Open()
+
+	close(h.opened)
 	return nil
 }
 

server.go

@@ -196,13 +196,15 @@ func (s *Server) Open() error {
 	if err := s.Holder.Open(); err != nil {
 		return fmt.Errorf("opening Holder: %v", err)
 	}
-	s.Cluster.setNodeState(NodeStateReady)
+	if err := s.Cluster.SetNodeState(NodeStateReady); err != nil {
+		return fmt.Errorf("setting nodeState: %v", err)
+	}
 
 	// Listen for joining nodes.
 	// This needs to start after the Holder has opened so that nodes can join
 	// the cluster without waiting for data to load on the coordinator. Before
-	// this starts, the joins are queued up in the Cluster.joiningURIs buffered
-	// channel.
+	// this starts, the joins are queued up in the Cluster.joiningLeavingNodes
+	// buffered channel.
 	s.Cluster.ListenForJoins()
 
 	// Start background monitoring.

test/cluster.go

@@ -250,8 +250,13 @@ func (t *TestCluster) SetState(state string) {
 // Open opens all clusters in the test cluster.
 func (t *TestCluster) Open() error {
 	for _, c := range t.Clusters {
-		err := c.Open()
-		if err != nil {
+		if err := c.Open(); err != nil {
+			return err
+		}
+		if err := c.Holder.Open(); err != nil {
+			return err
+		}
+		if err := c.SetNodeState(pilosa.NodeStateReady); err != nil {
 			return err
 		}
 	}