Cleans up some structure and comments based on review feedback.

Author: slackpad

fsm.go

@@ -109,6 +109,7 @@ func (r *Raft) runFSM() {
 				commitEntry.future.response = resp
 				commitEntry.future.respond(nil)
 			}
+
 		case <-r.shutdownCh:
 			return
 		}

future.go

@@ -53,7 +53,8 @@ type SnapshotFuture interface {
 	Future
 
 	// Open is a function you can call to access the underlying snapshot and
-	// its metadata.
+	// its metadata. This must not be called until after the Error method
+	// has returned.
 	Open() (*SnapshotMeta, io.ReadCloser, error)
 }
 
@@ -177,6 +178,11 @@ func (u *userSnapshotFuture) Open() (*SnapshotMeta, io.ReadCloser, error) {
 	if u.opener == nil {
 		return nil, nil, fmt.Errorf("no snapshot available")
 	} else {
+		// Invalidate the opener so it can't get called multiple times,
+		// which isn't generally safe.
+		defer func() {
+			u.opener = nil
+		}()
 		return u.opener()
 	}
 }

raft.go

@@ -559,7 +559,8 @@ func (r *Raft) leaderLoop() {
 			}
 
 		case future := <-r.userRestoreCh:
-			r.restoreUserSnapshot(future)
+			err := r.restoreUserSnapshot(future.meta, future.reader)
+			future.respond(err)
 
 		case c := <-r.configurationsCh:
 			c.configurations = r.configurations.Clone()
@@ -699,84 +700,82 @@ func (r *Raft) quorumSize() int {
 // so that the snapshot will be sent to followers and used for any new joiners.
 // This can only be run on the leader, and returns a future that can be used to
 // block until complete.
-func (r *Raft) restoreUserSnapshot(future *userRestoreFuture) {
+func (r *Raft) restoreUserSnapshot(meta *SnapshotMeta, reader io.ReadCloser) error {
 	defer metrics.MeasureSince([]string{"raft", "restoreUserSnapshot"}, time.Now())
-	err := func() error {
-		// Sanity check the version.
-		version := future.meta.Version
-		if version < SnapshotVersionMin || version > SnapshotVersionMax {
-			return fmt.Errorf("unsupported snapshot version %d", version)
-		}
 
-		// We don't support snapshots while there's a config change
-		// outstanding since the snapshot doesn't have a means to
-		// represent this state.
-		committedIndex := r.configurations.committedIndex
-		latestIndex := r.configurations.latestIndex
-		if committedIndex != latestIndex {
-			return fmt.Errorf("cannot restore snapshot now, wait until the configuration entry at %v has been applied (have applied %v)",
-				latestIndex, committedIndex)
-		}
+	// Sanity check the version.
+	version := meta.Version
+	if version < SnapshotVersionMin || version > SnapshotVersionMax {
+		return fmt.Errorf("unsupported snapshot version %d", version)
+	}
 
-		// We will overwrite the snapshot metadata with the current term,
-		// an index that's greater than the current index, or the last
-		// index in the snapshot. It's important that we leave a hole in
-		// the index so we know there's nothing in the Raft log there and
-		// replication will fault and send the snapshot.
-		term := r.getCurrentTerm()
-		lastIndex := r.getLastIndex()
-		if future.meta.Index > lastIndex {
-			lastIndex = future.meta.Index
-		}
-		lastIndex++
+	// We don't support snapshots while there's a config change
+	// outstanding since the snapshot doesn't have a means to
+	// represent this state.
+	committedIndex := r.configurations.committedIndex
+	latestIndex := r.configurations.latestIndex
+	if committedIndex != latestIndex {
+		return fmt.Errorf("cannot restore snapshot now, wait until the configuration entry at %v has been applied (have applied %v)",
+			latestIndex, committedIndex)
+	}
 
-		// Dump the snapshot. Note that we use the latest configuration,
-		// not the one that came with the snapshot.
-		sink, err := r.snapshots.Create(version, lastIndex, term,
-			r.configurations.latest, r.configurations.latestIndex, r.trans)
-		if err != nil {
-			return fmt.Errorf("failed to create snapshot: %v", err)
-		}
-		n, err := io.Copy(sink, future.reader)
-		if err != nil {
-			sink.Cancel()
-			return fmt.Errorf("failed to write snapshot: %v", err)
-		}
-		if n != future.meta.Size {
-			sink.Cancel()
-			return fmt.Errorf("failed to write snapshot, size didn't match (%d != %d)", n, future.meta.Size)
-		}
-		if err := sink.Close(); err != nil {
-			return fmt.Errorf("failed to close snapshot: %v", err)
-		}
-		r.logger.Printf("[INFO] raft: Copied %d bytes to local snapshot", n)
+	// We will overwrite the snapshot metadata with the current term,
+	// an index that's greater than the current index, or the last
+	// index in the snapshot. It's important that we leave a hole in
+	// the index so we know there's nothing in the Raft log there and
+	// replication will fault and send the snapshot.
+	term := r.getCurrentTerm()
+	lastIndex := r.getLastIndex()
+	if meta.Index > lastIndex {
+		lastIndex = meta.Index
+	}
+	lastIndex++
 
-		// Restore the snapshot into the FSM. If this fails we are in a
-		// bad state so we panic to take ourselves out.
-		fsm := &restoreFuture{ID: sink.ID()}
-		fsm.init()
-		select {
-		case r.fsmRestoreCh <- fsm:
-		case <-r.shutdownCh:
-			return ErrRaftShutdown
-		}
-		if err := fsm.Error(); err != nil {
-			panic(fmt.Errorf("failed to restore snapshot: %v", err))
-		}
+	// Dump the snapshot. Note that we use the latest configuration,
+	// not the one that came with the snapshot.
+	sink, err := r.snapshots.Create(version, lastIndex, term,
+		r.configurations.latest, r.configurations.latestIndex, r.trans)
+	if err != nil {
+		return fmt.Errorf("failed to create snapshot: %v", err)
+	}
+	n, err := io.Copy(sink, reader)
+	if err != nil {
+		sink.Cancel()
+		return fmt.Errorf("failed to write snapshot: %v", err)
+	}
+	if n != meta.Size {
+		sink.Cancel()
+		return fmt.Errorf("failed to write snapshot, size didn't match (%d != %d)", n, meta.Size)
+	}
+	if err := sink.Close(); err != nil {
+		return fmt.Errorf("failed to close snapshot: %v", err)
+	}
+	r.logger.Printf("[INFO] raft: Copied %d bytes to local snapshot", n)
 
-		// We set the last log so it looks like we've stored the empty
-		// index we burned. The last applied is set because we made the
-		// FSM take the snapshot state, and we store the last snapshot
-		// in the stable store since we created a snapshot as part of
-		// this process.
-		r.setLastLog(lastIndex, term)
-		r.setLastApplied(lastIndex)
-		r.setLastSnapshot(lastIndex, term)
-
-		r.logger.Printf("[INFO] raft: Restored user snapshot (index %d)", lastIndex)
-		return nil
-	}()
-	future.respond(err)
+	// Restore the snapshot into the FSM. If this fails we are in a
+	// bad state so we panic to take ourselves out.
+	fsm := &restoreFuture{ID: sink.ID()}
+	fsm.init()
+	select {
+	case r.fsmRestoreCh <- fsm:
+	case <-r.shutdownCh:
+		return ErrRaftShutdown
+	}
+	if err := fsm.Error(); err != nil {
+		panic(fmt.Errorf("failed to restore snapshot: %v", err))
+	}
+
+	// We set the last log so it looks like we've stored the empty
+	// index we burned. The last applied is set because we made the
+	// FSM take the snapshot state, and we store the last snapshot
+	// in the stable store since we created a snapshot as part of
+	// this process.
+	r.setLastLog(lastIndex, term)
+	r.setLastApplied(lastIndex)
+	r.setLastSnapshot(lastIndex, term)
+
+	r.logger.Printf("[INFO] raft: Restored user snapshot (index %d)", lastIndex)
+	return nil
 }
 
 // appendConfigurationEntry changes the configuration and adds a new