Merge pull request hashicorp#384 from hashicorp/f-stricter-leaderch-semantics

raft.LeaderCh() always deliver latest transition

Author: Mahmood Ali

api.go

@@ -503,7 +503,7 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 		fsm:                   fsm,
 		fsmMutateCh:           make(chan interface{}, 128),
 		fsmSnapshotCh:         make(chan *reqSnapshotFuture),
-		leaderCh:              make(chan bool),
+		leaderCh:              make(chan bool, 1),
 		localID:               localID,
 		localAddr:             localAddr,
 		logger:                logger,
@@ -952,10 +952,17 @@ func (r *Raft) State() RaftState {
 	return r.getState()
 }
 
-// LeaderCh is used to get a channel which delivers signals on
-// acquiring or losing leadership. It sends true if we become
-// the leader, and false if we lose it. The channel is not buffered,
-// and does not block on writes.
+// LeaderCh is used to get a channel which delivers signals on acquiring or
+// losing leadership. It sends true if we become the leader, and false if we
+// lose it.
+//
+// Receivers can expect to receive a notification only if leadership
+// transition has occured.
+//
+// If receivers aren't ready for the signal, signals may drop and only the
+// latest leadership transition. For example, if a receiver receives subsequent
+// `true` values, they may deduce that leadership was lost and regained while
+// the the receiver was processing first leadership transition.
 func (r *Raft) LeaderCh() <-chan bool {
 	return r.leaderCh
 }

raft.go

@@ -364,7 +364,7 @@ func (r *Raft) runLeader() {
 	metrics.IncrCounter([]string{"raft", "state", "leader"}, 1)
 
 	// Notify that we are the leader
-	asyncNotifyBool(r.leaderCh, true)
+	overrideNotifyBool(r.leaderCh, true)
 
 	// Push to the notify channel if given
 	if notify := r.conf.NotifyCh; notify != nil {
@@ -420,7 +420,7 @@ func (r *Raft) runLeader() {
 		r.leaderLock.Unlock()
 
 		// Notify that we are not the leader
-		asyncNotifyBool(r.leaderCh, false)
+		overrideNotifyBool(r.leaderCh, false)
 
 		// Push to the notify channel if given
 		if notify := r.conf.NotifyCh; notify != nil {

util.go

@@ -96,6 +96,25 @@ func asyncNotifyBool(ch chan bool, v bool) {
 	}
 }
 
+// overrideNotifyBool is used to notify on a bool channel
+// but override existing value if value is present.
+// ch must be 1-item buffered channel.
+//
+// This method does not support multiple concurrent calls.
+func overrideNotifyBool(ch chan bool, v bool) {
+	select {
+	case ch <- v:
+		// value sent, all done
+	case <-ch:
+		// channel had an old value
+		select {
+		case ch <- v:
+		default:
+			panic("race: channel was sent concurrently")
+		}
+	}
+}
+
 // Decode reverses the encode operation on a byte slice input.
 func decodeMsgPack(buf []byte, out interface{}) error {
 	r := bytes.NewBuffer(buf)

util_test.go

@@ -100,3 +100,48 @@ func TestBackoff(t *testing.T) {
 		t.Fatalf("bad: %v", b)
 	}
 }
+
+func TestOverrideNotifyBool(t *testing.T) {
+	ch := make(chan bool, 1)
+
+	// sanity check - buffered channel don't have any values
+	select {
+	case v := <-ch:
+		t.Fatalf("unexpected receive: %v", v)
+	default:
+	}
+
+	// simple case of a single push
+	overrideNotifyBool(ch, false)
+	select {
+	case v := <-ch:
+		if v != false {
+			t.Fatalf("expected false but got %v", v)
+		}
+	default:
+		t.Fatalf("expected a value but is not ready")
+	}
+
+	// assert that function never blocks and only last item is received
+	overrideNotifyBool(ch, false)
+	overrideNotifyBool(ch, false)
+	overrideNotifyBool(ch, false)
+	overrideNotifyBool(ch, false)
+	overrideNotifyBool(ch, true)
+
+	select {
+	case v := <-ch:
+		if v != true {
+			t.Fatalf("expected true but got %v", v)
+		}
+	default:
+		t.Fatalf("expected a value but is not ready")
+	}
+
+	// no further value is available
+	select {
+	case v := <-ch:
+		t.Fatalf("unexpected receive: %v", v)
+	default:
+	}
+}