diff --git a/internal/bft/controller.go b/internal/bft/controller.go index d1067eb8..7499e195 100644 --- a/internal/bft/controller.go +++ b/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 { diff --git a/internal/bft/completion_signal_test.go b/internal/bft/controller_decide_test.go similarity index 74% rename from internal/bft/completion_signal_test.go rename to internal/bft/controller_decide_test.go index 8e4fde3c..d1989c3c 100644 --- a/internal/bft/completion_signal_test.go +++ b/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.