bft: enqueue a decision when its view is aborted

internal/bft/controller.go

@@ -890,18 +890,26 @@ func (c *Controller) Decide(proposal types.Proposal, signatures []types.Signatur
 	c.currViewLock.RUnlock()
 
 	delivered := make(chan struct{})
-	select {
-	case c.decisionChan <- decision{
+	d := decision{
 		proposal:   proposal,
 		requests:   requests,
 		signatures: signatures,
 		view:       view,
 		delivered:  delivered,
-	}:
+	}
+	select {
+	case c.decisionChan <- d:
 	case <-c.stopChan:
 		// In case we are in the middle of shutting down,
 		// abort deciding.
 		return
+	case <-view.AbortChan():
+		// The controller loop is aborting this view and waits for it to stop. Do
+		// not block the view, or the abort and the decision would wait for each
+		// other, but do not drop the decision either: hand it to the controller
+		// in the background.
+		go c.enqueueDecision(d)
+		return
 	}
 
 	select {
@@ -911,6 +919,19 @@ func (c *Controller) Decide(proposal types.Proposal, signatures []types.Signatur
 	}
 }
 
+// enqueueDecision hands a decision to the controller when the view that decided it is
+// already stopping. The decision was accepted by a quorum, so it must not be lost:
+// the controller delivers it once it is no longer busy. If the view changer has
+// already delivered the same proposal, MutuallyExclusiveDeliver skips delivering it
+// again by comparing the checkpoint sequence, so the application does not receive it
+// twice.
+func (c *Controller) enqueueDecision(d decision) {
+	select {
+	case c.decisionChan <- d:
+	case <-c.stopChan:
+	}
+}
+
 func (c *Controller) removeDeliveredFromPool(d decision) {
 	for _, reqInfo := range d.requests {
 		if err := c.RequestPool.RemoveRequest(reqInfo); err != nil {

internal/bft/completion_signal_test.go → internal/bft/controller_decide_test.go

@@ -7,6 +7,7 @@ package bft
 
 import (
 	"testing"
+	"time"
 
 	"github.com/hyperledger/SmartBFT/pkg/types"
 	protos "github.com/hyperledger/SmartBFT/smartbftprotos"
@@ -48,6 +49,49 @@ func TestControllerDecideViewBookkeeping(t *testing.T) {
 	})
 }
 
+// TestDecideDoesNotDropDecisionOnAbortedView covers the hand-off between the view
+// and the controller: a view that is being aborted must not stay blocked on a full
+// decision channel, but the committed decision must still reach the controller.
+func TestDecideDoesNotDropDecisionOnAbortedView(t *testing.T) {
+	view := &View{abortChan: make(chan struct{})}
+	controller := &Controller{
+		currView:     view,
+		decisionChan: make(chan decision, 1),
+		stopChan:     make(chan struct{}),
+	}
+	controller.decisionChan <- decision{proposal: types.Proposal{Payload: []byte("queued")}}
+
+	decided := make(chan struct{})
+	go func() {
+		controller.Decide(types.Proposal{Payload: []byte("committed")}, nil, nil)
+		close(decided)
+	}()
+
+	select {
+	case <-decided:
+		t.Fatal("Decide returned before the view was aborted")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	view.stop()
+
+	select {
+	case <-decided:
+	case <-time.After(time.Second):
+		t.Fatal("Decide did not return after the view was aborted")
+	}
+
+	queued := <-controller.decisionChan
+	require.Equal(t, []byte("queued"), queued.proposal.Payload)
+
+	select {
+	case committed := <-controller.decisionChan:
+		require.Equal(t, []byte("committed"), committed.proposal.Payload)
+	case <-time.After(time.Second):
+		t.Fatal("the committed decision was dropped")
+	}
+}
+
 // newDecidingController returns a controller whose current view is the given view,
 // and whose own ID is the leader of the current view, so that a delivered decision
 // would normally acquire the leader token.