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

gpbft/gpbft.go

@@ -211,7 +211,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(
@@ -221,7 +222,8 @@ func newInstance(
 	data *SupplementalData,
 	powerTable *PowerTable,
 	aggregateVerifier Aggregate,
-	beacon []byte) (*instance, error) {
+	beacon []byte,
+	progress ProgressObserver) (*instance, error) {
 	if input.IsZero() {
 		return nil, fmt.Errorf("input is empty")
 	}
@@ -251,6 +253,7 @@ func newInstance(
 		},
 		decision: newQuorumState(powerTable),
 		tracer:   participant.tracer,
+		progress: progress,
 	}, nil
 }
 
@@ -486,6 +489,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
@@ -546,6 +550,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))
 }
@@ -603,6 +608,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))
 }
@@ -656,6 +662,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))
 }
@@ -716,6 +723,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 {
@@ -731,6 +739,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))
 }
@@ -744,6 +753,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))
@@ -779,6 +789,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)
@@ -806,6 +817,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))
 
@@ -847,6 +859,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))

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