Refactor validation logic into a mutex-free pluggable struct

Refactor the validation logic out of `Participant` and into its own
dedicated struct that is mutex-free and listens to the progress made by
the participant to infer the correct validation path.

The change above significantly reduces the need for mutex control over
current instance, which makes it easier to plug in extra conditional
behaviour, e.g. #583.

Fixes #561

Author: masih

gpbft/gpbft.go

@@ -209,7 +209,8 @@ type instance struct {
 	// independently of protocol phases/rounds.
 	decision *quorumState
 	// tracer traces logic logs for debugging and simulation purposes.
-	tracer Tracer
+	tracer   Tracer
+	progress ProgressObserver
 }
 
 func newInstance(
@@ -218,7 +219,8 @@ func newInstance(
 	input ECChain,
 	data *SupplementalData,
 	powerTable PowerTable,
-	beacon []byte) (*instance, error) {
+	beacon []byte,
+	progress ProgressObserver) (*instance, error) {
 	if input.IsZero() {
 		return nil, fmt.Errorf("input is empty")
 	}
@@ -247,6 +249,7 @@ func newInstance(
 		},
 		decision: newQuorumState(powerTable),
 		tracer:   participant.tracer,
+		progress: progress,
 	}, nil
 }
 
@@ -482,6 +485,7 @@ func (i *instance) beginQuality() error {
 	i.phaseTimeout = i.alarmAfterSynchrony()
 	i.resetRebroadcastParams()
 	i.broadcast(i.round, QUALITY_PHASE, i.proposal, false, nil)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrQualityPhase))
 	metrics.currentPhase.Record(context.TODO(), int64(QUALITY_PHASE))
 	return nil
@@ -542,6 +546,7 @@ func (i *instance) beginConverge(justification *Justification) {
 	i.getRound(i.round).converged.SetSelfValue(i.proposal, justification)
 
 	i.broadcast(i.round, CONVERGE_PHASE, i.proposal, true, justification)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrConvergePhase))
 	metrics.currentPhase.Record(context.TODO(), int64(CONVERGE_PHASE))
 }
@@ -599,6 +604,7 @@ func (i *instance) beginPrepare(justification *Justification) {
 	i.resetRebroadcastParams()
 
 	i.broadcast(i.round, PREPARE_PHASE, i.value, false, justification)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrPreparePhase))
 	metrics.currentPhase.Record(context.TODO(), int64(PREPARE_PHASE))
 }
@@ -652,6 +658,7 @@ func (i *instance) beginCommit() {
 	}
 
 	i.broadcast(i.round, COMMIT_PHASE, i.value, false, justification)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrCommitPhase))
 	metrics.currentPhase.Record(context.TODO(), int64(COMMIT_PHASE))
 }
@@ -712,6 +719,7 @@ func (i *instance) tryCommit(round uint64) error {
 func (i *instance) beginDecide(round uint64) {
 	i.phase = DECIDE_PHASE
 	i.resetRebroadcastParams()
+
 	var justification *Justification
 	// Value cannot be empty here.
 	if quorum, ok := i.getRound(round).committed.FindStrongQuorumFor(i.value.Key()); ok {
@@ -727,6 +735,7 @@ func (i *instance) beginDecide(round uint64) {
 	// Since each node sends only one DECIDE message, they must share the same vote
 	// in order to be aggregated.
 	i.broadcast(0, DECIDE_PHASE, i.value, false, justification)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrDecidePhase))
 	metrics.currentPhase.Record(context.TODO(), int64(DECIDE_PHASE))
 }
@@ -740,6 +749,7 @@ func (i *instance) skipToDecide(value ECChain, justification *Justification) {
 	i.value = i.proposal
 	i.resetRebroadcastParams()
 	i.broadcast(0, DECIDE_PHASE, i.value, false, justification)
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrDecidePhase))
 	metrics.currentPhase.Record(context.TODO(), int64(DECIDE_PHASE))
 	metrics.skipCounter.Add(context.TODO(), 1, metric.WithAttributes(attrSkipToDecide))
@@ -775,6 +785,7 @@ func (i *instance) getRound(r uint64) *roundState {
 func (i *instance) beginNextRound() {
 	i.log("moving to round %d with %s", i.round+1, i.proposal.String())
 	i.round += 1
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.currentRound.Record(context.TODO(), int64(i.round))
 
 	prevRound := i.getRound(i.round - 1)
@@ -802,6 +813,7 @@ func (i *instance) beginNextRound() {
 func (i *instance) skipToRound(round uint64, chain ECChain, justification *Justification) {
 	i.log("skipping from round %d to round %d with %s", i.round, round, i.proposal.String())
 	i.round = round
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.currentRound.Record(context.TODO(), int64(i.round))
 	metrics.skipCounter.Add(context.TODO(), 1, metric.WithAttributes(attrSkipToRound))
 
@@ -843,6 +855,7 @@ func (i *instance) terminate(decision *Justification) {
 	i.value = decision.Vote.Value
 	i.terminationValue = decision
 	i.resetRebroadcastParams()
+	i.progress.NotifyProgress(i.instanceID, i.round, i.phase)
 	metrics.phaseCounter.Add(context.TODO(), 1, metric.WithAttributes(attrTerminatedPhase))
 	metrics.roundHistogram.Record(context.TODO(), int64(i.round))
 	metrics.currentPhase.Record(context.TODO(), int64(TERMINATED_PHASE))
@@ -940,7 +953,7 @@ func (i *instance) rebroadcastTimeoutElapsed() bool {
 
 func (i *instance) rebroadcast() error {
 	// Rebroadcast quality and all messages from the current and previous rounds, unless the
-	// instance has progressed to DECIDE phase. In which case, only DECIDE message is
+	// instance has progress to DECIDE phase. In which case, only DECIDE message is
 	// rebroadcasted.
 	//
 	// Note that the implementation here rebroadcasts more messages than FIP-0086

gpbft/gpbft_test.go

@@ -281,7 +281,7 @@ func TestGPBFT_WithEvenPowerDistribution(t *testing.T) {
 	t.Run("Queues future instance messages during current instance", func(t *testing.T) {
 		instance, driver := newInstanceAndDriver(t)
 		futureInstance := emulator.NewInstance(t,
-			42,
+			8,
 			gpbft.PowerEntries{
 				gpbft.PowerEntry{
 					ID:    0,

gpbft/options.go

@@ -8,11 +8,12 @@ import (
 	"time"
 )
 
-var (
+const (
 	defaultDelta                        = 3 * time.Second
 	defaultDeltaBackOffExponent         = 2.0
 	defaultMaxCachedInstances           = 10
 	defaultMaxCachedMessagesPerInstance = 25_000
+	defaultCommitteeLookback            = 10
 )
 
 // Option represents a configurable parameter.
@@ -22,6 +23,7 @@ type options struct {
 	delta                time.Duration
 	deltaBackOffExponent float64
 
+	committeeLookback  uint64
 	maxLookaheadRounds uint64
 	rebroadcastAfter   func(int) time.Duration
 
@@ -36,6 +38,7 @@ func newOptions(o ...Option) (*options, error) {
 	opts := &options{
 		delta:                        defaultDelta,
 		deltaBackOffExponent:         defaultDeltaBackOffExponent,
+		committeeLookback:            defaultCommitteeLookback,
 		rebroadcastAfter:             defaultRebroadcastAfter,
 		maxCachedInstances:           defaultMaxCachedInstances,
 		maxCachedMessagesPerInstance: defaultMaxCachedMessagesPerInstance,
@@ -118,6 +121,15 @@ func WithMaxCachedMessagesPerInstance(v int) Option {
 	}
 }
 
+// WithCommitteeLookback sets the number of instances in the past from which the
+// committee for the latest instance is derived. Defaults to 10 if unset.
+func WithCommitteeLookback(lookback uint64) Option {
+	return func(o *options) error {
+		o.committeeLookback = lookback
+		return nil
+	}
+}
+
 var defaultRebroadcastAfter = exponentialBackoffer(1.3, 0.1, 3*time.Second, 30*time.Second)
 
 // WithRebroadcastBackoff sets the duration after the gPBFT timeout has elapsed, at