Add removing old ballots and broadcast EXP ballots

lib/consensus/isaac.go

@@ -225,6 +225,52 @@ func (is *ISAAC) Vote(b ballot.Ballot) (isNew bool, err error) {
 	return
 }
 
+// RemoveOldBallots checkes that `blt` has valid confirmed and proposed time.
+// If it is invalid, it is removed.
+// And if the invalid ballot is make by itself,
+// return `needRenewal` = true for rebroadcasting EXP ballot
+func (is *ISAAC) RemoveOldBallots(blt ballot.Ballot) (needRenewal bool) {
+	runningRound, found := is.RunningRounds[blt.VotingBasis().Index()]
+	if !found {
+		// if RunningRound is not found, this ballot will be stopped.
+		return false
+	}
+	var roundVote *RoundVote
+	var err error
+	roundVote, err = runningRound.RoundVote(blt.Proposer())
+	if err != nil {
+		return false
+	}
+
+	switch blt.State() {
+	case ballot.StateSIGN:
+		for nodeAddr, blt := range roundVote.SIGN {
+			if err = ballot.CheckHasCorrectTime(blt.ProposerConfirmed()); err == nil {
+				if err = ballot.CheckHasCorrectTime(blt.Confirmed()); err == nil {
+					continue
+				}
+			}
+			delete(roundVote.SIGN, nodeAddr)
+			if nodeAddr == is.Node.Address() {
+				needRenewal = true
+			}
+		}
+	case ballot.StateACCEPT:
+		for nodeAddr, blt := range roundVote.ACCEPT {
+			if err = ballot.CheckHasCorrectTime(blt.ProposerConfirmed()); err == nil {
+				if err = ballot.CheckHasCorrectTime(blt.Confirmed()); err == nil {
+					continue
+				}
+			}
+			delete(roundVote.ACCEPT, nodeAddr)
+			if nodeAddr == is.Node.Address() {
+				needRenewal = true
+			}
+		}
+	}
+	return needRenewal
+}
+
 func (is *ISAAC) CanGetVotingResult(blt ballot.Ballot) (rv RoundVoteResult, vh voting.Hole, finished bool) {
 	is.RLock()
 	defer is.RUnlock()

lib/node/runner/checker.go

@@ -380,6 +380,35 @@ func BallotIsSameProposer(c common.Checker, args ...interface{}) (err error) {
 	return
 }
 
+// BallotRenewal checks that the ballot confirmed and proposed time is valid and
+// if the time is invalid, then it broadcasts expired ballot
+func BallotRenewal(c common.Checker, args ...interface{}) (err error) {
+	checker := c.(*BallotChecker)
+	nr := checker.NodeRunner
+	is := nr.Consensus()
+	blt := checker.Ballot
+
+	if !blt.State().IsValidForVote() {
+		return
+	}
+
+	needRenewal := is.RemoveOldBallots(blt)
+	if needRenewal {
+		expiredBlt := is.GenerateExpiredBallot(blt.VotingBasis(), blt.State())
+		nr.SetSendRecord(
+			consensus.ISAACState{
+				Height:      blt.VotingBasis().Height,
+				Round:       blt.VotingBasis().Round,
+				BallotState: blt.State(),
+			},
+			false,
+		)
+		nr.BroadcastBallot(expiredBlt)
+	}
+
+	return
+}
+
 // BallotCheckResult checks the voting result.
 func BallotCheckResult(c common.Checker, args ...interface{}) (err error) {
 	checker := c.(*BallotChecker)

lib/node/runner/node_runner.go

@@ -54,6 +54,7 @@ var DefaultHandleSIGNBallotCheckerFuncs = []common.CheckerFunc{
 	BallotAlreadyVoted,
 	BallotVote,
 	BallotIsSameProposer,
+	BallotRenewal,
 	BallotCheckResult,
 	ExpiredInSIGN,
 	ACCEPTBallotBroadcast,
@@ -64,6 +65,7 @@ var DefaultHandleACCEPTBallotCheckerFuncs = []common.CheckerFunc{
 	BallotAlreadyVoted,
 	BallotVote,
 	BallotIsSameProposer,
+	BallotRenewal,
 	BallotCheckResult,
 	FinishedBallotStore,
 }

lib/node/runner/remove_old_ballots_test.go

@@ -0,0 +1,75 @@
+package runner
+
+import (
+	"testing"
+	"time"
+
+	"boscoin.io/sebak/lib/ballot"
+	"boscoin.io/sebak/lib/common"
+	"boscoin.io/sebak/lib/consensus"
+	"boscoin.io/sebak/lib/voting"
+	"github.com/stretchr/testify/require"
+)
+
+func insertOldRunningRound(t *testing.T, is *consensus.ISAAC, blt ballot.Ballot) {
+	if runningRound, found := is.RunningRounds[blt.VotingBasis().Index()]; !found {
+		rr, err := consensus.NewRunningRound(blt.Proposer(), blt)
+		require.NoError(t, err)
+		is.RunningRounds[blt.VotingBasis().Index()] = rr
+	} else {
+		runningRound.Vote(blt)
+	}
+}
+
+func TestRemoveOldBallots(t *testing.T) {
+	nr, nodes, _ := createNodeRunnerForTesting(3, common.NewTestConfig(), nil)
+
+	basis := voting.Basis{
+		Height: 10,
+		Round:  0,
+	}
+
+	is := nr.consensus
+
+	invalidBlt := *ballot.NewBallot(nodes[0].Address(), nodes[0].Address(), basis, []string{})
+	invalidBlt.SetVote(ballot.StateSIGN, voting.YES)
+	invalidBlt.B.Proposed.Confirmed = common.FormatISO8601(time.Now().Add(-2 * time.Minute))
+	invalidBlt.B.Confirmed = common.FormatISO8601(time.Now().Add(-3 * time.Minute))
+
+	proposedInvalidBlt := *ballot.NewBallot(nodes[1].Address(), nodes[1].Address(), basis, []string{})
+	proposedInvalidBlt.SetVote(ballot.StateSIGN, voting.YES)
+	proposedInvalidBlt.B.Proposed.Confirmed = common.FormatISO8601(time.Now().Add(-2 * time.Minute))
+	proposedInvalidBlt.B.Confirmed = common.FormatISO8601(time.Now())
+
+	validBlt := *ballot.NewBallot(nodes[2].Address(), nodes[2].Address(), basis, []string{})
+	validBlt.SetVote(ballot.StateSIGN, voting.YES)
+	validBlt.B.Proposed.Confirmed = common.FormatISO8601(time.Now())
+	validBlt.B.Confirmed = common.FormatISO8601(time.Now())
+
+	insertOldRunningRound(t, is, invalidBlt)
+	insertOldRunningRound(t, is, proposedInvalidBlt)
+	insertOldRunningRound(t, is, validBlt)
+
+	rr := is.RunningRounds[basis.Index()]
+	require.True(t, rr.IsVoted(invalidBlt))
+	require.True(t, rr.IsVoted(proposedInvalidBlt))
+	require.True(t, rr.IsVoted(validBlt))
+
+	needRenewal := is.RemoveOldBallots(invalidBlt)
+	require.True(t, needRenewal)
+
+	require.False(t, rr.IsVoted(invalidBlt))
+	require.True(t, rr.IsVoted(proposedInvalidBlt))
+	require.True(t, rr.IsVoted(validBlt))
+
+	needRenewal = is.RemoveOldBallots(proposedInvalidBlt)
+	require.False(t, needRenewal) // because it's not made by itself
+
+	require.False(t, rr.IsVoted(proposedInvalidBlt))
+	require.True(t, rr.IsVoted(validBlt))
+
+	needRenewal = is.RemoveOldBallots(validBlt)
+	require.False(t, needRenewal)
+
+	require.True(t, rr.IsVoted(validBlt))
+}

tests/retry/run.sh

@@ -60,11 +60,20 @@ sleep 1
 # Check sync
 docker run --rm --network host ${CLIENT_IMAGE} sync.sh
 
+# Network problem for a short time
 docker stop ${NODE3}
 sleep 10
 docker start ${NODE3}
 
-# Check sync after starting NODE3
+# Ensure consensus is in progress
+docker run --rm --network host ${CLIENT_IMAGE} consensus_alive.sh
+
+# Network problem for a long time
+docker stop ${NODE3}
+sleep 61
+docker start ${NODE3}
+
+# Ensure consensus is in progress
 docker run --rm --network host ${CLIENT_IMAGE} consensus_alive.sh
 
 # Shut down the containers - we need to do so for integration reports to be written