Rebroadcast immediately once a round threshold is met (#1003)

* Rebroadcast immediately once a round threshold is met

The original GPBFT protocol dictates that the rebroadcast only gets
triggered after the phase timeout has expired. This is reasonable for
low number of rounds, where the vast majority of instances finalize
within a single round. However, there is an edge-case where when an
instance enters more than a few rounds the exponential increase of phase
timeout results in extended periods of total radio silence.

As a result, the state propagation is hindered greatly, which in turn
increases the chances of the instance taking even longer to finalize.

The changes here introduce a threshold at which rebroadcast is scheduled
without waiting for phase timeout to expire first (defaulting to 3). In
this design, the rebroadcast and successive rebroadcasts work exactly
the same way as dictated in GPBFT design, except the trigger is adjsuted
to consider rounds larger than the threshold as "insufficient progress".

The implementation reverts back to the vanilla GPBFT phase timeout as
the phases progress and resets rebroadcast parameters accordingly

A test is added to assert that when an instance skips to future rounds,
rebroadcast is triggered without wait, and continues to do so after
phase change.

* Refine logging

Author: masih

emulator/driver.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/filecoin-project/go-f3/gpbft"
 	"github.com/stretchr/testify/require"
@@ -105,3 +106,12 @@ func (d *Driver) deliverMessage(msg *gpbft.GMessage) error {
 		return d.subject.ReceiveMessage(ctx, validated)
 	}
 }
+
+// AdvanceTimeBy advances the current time of the driver by the given amount.
+// This allows the driver to simulate the passage of time in the emulated
+// gpbft.Participant. This is useful for testing timeouts and other time-based
+// behavior in the gpbft.Participant in high number of rounds, which
+// exponentially increases the phase timeout.
+func (d *Driver) AdvanceTimeBy(t time.Duration) {
+	d.host.now = d.host.now.Add(t)
+}

emulator/host.go

@@ -39,7 +39,9 @@ func (h *driverHost) maybeReceiveAlarm() bool {
 	if h.pendingAlarm == nil {
 		return false
 	}
-	h.now = *h.pendingAlarm
+	if h.now.Before(*h.pendingAlarm) {
+		h.now = *h.pendingAlarm
+	}
 	h.pendingAlarm = nil
 	return true
 }

gpbft/gpbft.go

@@ -550,7 +550,7 @@ func (i *instance) tryQuality() error {
 	// Wait either for a strong quorum that agree on our proposal, or for the timeout
 	// to expire.
 	foundQuorum := i.quality.HasStrongQuorumFor(i.proposal.Key())
-	timeoutExpired := atOrAfter(i.participant.host.Time(), i.phaseTimeout)
+	timeoutExpired := i.phaseTimeoutElapsed()
 
 	if foundQuorum || timeoutExpired {
 		// If strong quorum of input is found the proposal will remain unchanged.
@@ -606,8 +606,11 @@ func (i *instance) tryConverge() error {
 		return fmt.Errorf("unexpected phase %s, expected %s", i.current.Phase, CONVERGE_PHASE)
 	}
 	// The CONVERGE phase timeout doesn't wait to hear from >⅔ of power.
-	timeoutExpired := atOrAfter(i.participant.host.Time(), i.phaseTimeout)
+	timeoutExpired := i.phaseTimeoutElapsed()
 	if !timeoutExpired {
+		if i.shouldRebroadcast() {
+			i.tryRebroadcast()
+		}
 		return nil
 	}
 	commitRoundState := i.getRound(i.current.Round - 1).committed
@@ -666,9 +669,8 @@ func (i *instance) tryPrepare() error {
 	prepared := i.getRound(i.current.Round).prepared
 	proposalKey := i.proposal.Key()
 	foundQuorum := prepared.HasStrongQuorumFor(proposalKey)
-	timedOut := atOrAfter(i.participant.host.Time(), i.phaseTimeout)
 	quorumNotPossible := !prepared.CouldReachStrongQuorumFor(proposalKey, false)
-	phaseComplete := timedOut && prepared.ReceivedFromStrongQuorum()
+	phaseComplete := i.phaseTimeoutElapsed() && prepared.ReceivedFromStrongQuorum()
 
 	if foundQuorum {
 		i.value = i.proposal
@@ -678,7 +680,7 @@ func (i *instance) tryPrepare() error {
 
 	if foundQuorum || quorumNotPossible || phaseComplete {
 		i.beginCommit()
-	} else if timedOut {
+	} else if i.shouldRebroadcast() {
 		i.tryRebroadcast()
 	}
 	return nil
@@ -714,8 +716,7 @@ func (i *instance) tryCommit(round uint64) error {
 	// no check on the current phase.
 	committed := i.getRound(round).committed
 	quorumValue, foundStrongQuorum := committed.FindStrongQuorumValue()
-	timedOut := atOrAfter(i.participant.host.Time(), i.phaseTimeout)
-	phaseComplete := timedOut && committed.ReceivedFromStrongQuorum()
+	phaseComplete := i.phaseTimeoutElapsed() && committed.ReceivedFromStrongQuorum()
 
 	switch {
 	case foundStrongQuorum && !quorumValue.IsZero():
@@ -751,7 +752,7 @@ func (i *instance) tryCommit(round uint64) error {
 			}
 		}
 		i.beginNextRound()
-	case timedOut:
+	case i.shouldRebroadcast():
 		// The phase has timed out. Attempt to re-broadcast messages.
 		i.tryRebroadcast()
 	}
@@ -940,16 +941,39 @@ func (i *instance) tryRebroadcast() {
 		// instance phase and schedule the first alarm:
 		//  * If in DECIDE phase, use current time as offset. Because, DECIDE phase does
 		//    not have any phase timeout and may be too far in the past.
+		//  * If the current phase is beyond the immediate rebroadcast threshold, use
+		//    the current time as offset to avoid extended periods of radio silence
+		//    when phase timeout grows exponentially large.
 		//  * Otherwise, use the phase timeout.
 		var rebroadcastTimeoutOffset time.Time
-		if i.current.Phase == DECIDE_PHASE {
+		if i.current.Phase == DECIDE_PHASE || i.current.Round > i.participant.rebroadcastImmediatelyAfterRound {
 			rebroadcastTimeoutOffset = i.participant.host.Time()
 		} else {
 			rebroadcastTimeoutOffset = i.phaseTimeout
 		}
 		i.rebroadcastTimeout = rebroadcastTimeoutOffset.Add(i.participant.rebroadcastAfter(0))
-		i.participant.host.SetAlarm(i.rebroadcastTimeout)
-		i.log("scheduled initial rebroadcast at %v", i.rebroadcastTimeout)
+		if i.phaseTimeoutElapsed() {
+			// The phase timeout has already elapsed; therefore, there's no risk of
+			// overriding any existing alarm. Simply set the alarm for rebroadcast.
+			i.participant.host.SetAlarm(i.rebroadcastTimeout)
+			i.log("scheduled initial rebroadcast at %v", i.rebroadcastTimeout)
+		} else if i.rebroadcastTimeout.Before(i.phaseTimeout) {
+			// The rebroadcast timeout is set before the phase timeout; therefore, it should
+			// trigger before the phase timeout. Override the alarm with rebroadcast timeout
+			// and check for phase timeout in the next cycle of rebroadcast.
+			i.participant.host.SetAlarm(i.rebroadcastTimeout)
+			i.log("scheduled initial rebroadcast at %v before phase timeout at %v", i.rebroadcastTimeout, i.phaseTimeout)
+		} else {
+			// The phase timeout is set before the rebroadcast timeout. Therefore, there must
+			// have been an alarm set already for the phase. Do nothing, because the GPBFT
+			// process loop will trigger the phase alarm, which in turn tries the current
+			// phase and eventually will try rebroadcast.
+			//
+			// Therefore, reset the rebroadcast parameters to re-attempt setting the initial
+			// rebroadcast timeout once the phase expires.
+			i.log("Resetting rebroadcast as rebroadcast timeout at %v is after phase timeout at %v and the current phase has not timed out yet.", i.rebroadcastTimeout, i.phaseTimeout)
+			i.resetRebroadcastParams()
+		}
 	case i.rebroadcastTimeoutElapsed():
 		// Rebroadcast now that the corresponding timeout has elapsed, and schedule the
 		// successive rebroadcast.
@@ -958,13 +982,27 @@ func (i *instance) tryRebroadcast() {
 
 		// Use current host time as the offset for the next alarm to assure that rate of
 		// broadcasted messages grows relative to the actual time at which an alarm is
-		// triggered , not the absolute alarm time. This would avoid a "runaway
+		// triggered, not the absolute alarm time. This would avoid a "runaway
 		// rebroadcast" scenario where rebroadcast timeout consistently remains behind
 		// current time due to the discrepancy between set alarm time and the actual time
 		// at which the alarm is triggered.
 		i.rebroadcastTimeout = i.participant.host.Time().Add(i.participant.rebroadcastAfter(i.rebroadcastAttempts))
-		i.participant.host.SetAlarm(i.rebroadcastTimeout)
-		i.log("scheduled next rebroadcast at %v", i.rebroadcastTimeout)
+		if i.phaseTimeoutElapsed() {
+			// The phase timeout has already elapsed; therefore, there's no risk of
+			// overriding any existing alarm. Simply set the alarm for rebroadcast.
+			i.participant.host.SetAlarm(i.rebroadcastTimeout)
+			i.log("scheduled next rebroadcast at %v", i.rebroadcastTimeout)
+		} else if i.rebroadcastTimeout.Before(i.phaseTimeout) {
+			// The rebroadcast timeout is set before the phase timeout; therefore, it should
+			// trigger before the phase timeout. Override the alarm with rebroadcast timeout
+			// and check for phase timeout in the next cycle of rebroadcast.
+			i.participant.host.SetAlarm(i.rebroadcastTimeout)
+			i.log("scheduled next rebroadcast at %v before phase timeout at %v", i.rebroadcastTimeout, i.phaseTimeout)
+		} else {
+			// The rebroadcast timeout is set after the phase timeout. Set the alarm for phase timeout instead.
+			i.log("Reverted to phase timeout at %v as it is before the next rebroadcast timeout at %v", i.phaseTimeout, i.rebroadcastTimeout)
+			i.participant.host.SetAlarm(i.phaseTimeout)
+		}
 	default:
 		// Rebroadcast timeout is set but has not elapsed yet; nothing to do.
 	}
@@ -980,6 +1018,14 @@ func (i *instance) rebroadcastTimeoutElapsed() bool {
 	return atOrAfter(now, i.rebroadcastTimeout)
 }
 
+func (i *instance) shouldRebroadcast() bool {
+	return i.phaseTimeoutElapsed() || i.current.Round > i.participant.rebroadcastImmediatelyAfterRound
+}
+
+func (i *instance) phaseTimeoutElapsed() bool {
+	return atOrAfter(i.participant.host.Time(), i.phaseTimeout)
+}
+
 func (i *instance) rebroadcast() {
 	// 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

gpbft/gpbft_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"io"
 	"testing"
+	"time"
 
 	"github.com/filecoin-project/go-f3/emulator"
 	"github.com/filecoin-project/go-f3/gpbft"
@@ -522,6 +523,51 @@ func TestGPBFT_WithEvenPowerDistribution(t *testing.T) {
 		})
 		driver.RequireDecision(instance.ID(), futureRoundProposal)
 	})
+
+	t.Run("Rebroadcasts independent of phase timeout after 3 rounds", func(t *testing.T) {
+		instance, driver := newInstanceAndDriver(t)
+		driver.RequireStartInstance(instance.ID())
+		driver.RequireQuality()
+		justification := instance.NewJustification(12, gpbft.PREPARE_PHASE, instance.Proposal(), 0, 1)
+		driver.RequireDeliverMessage(&gpbft.GMessage{
+			Sender:        0,
+			Vote:          instance.NewConverge(13, instance.Proposal()),
+			Ticket:        emulator.ValidTicket,
+			Justification: justification,
+		})
+		driver.RequireDeliverMessage(&gpbft.GMessage{
+			Sender:        0,
+			Vote:          instance.NewPrepare(13, instance.Proposal()),
+			Ticket:        emulator.ValidTicket,
+			Justification: justification,
+		})
+
+		requireRebroadcast := func() {
+			driver.RequireQuality()
+			driver.RequireConverge(13, instance.Proposal(), justification)
+			driver.RequireNoBroadcast()
+		}
+
+		driver.RequireConverge(13, instance.Proposal(), justification)       // Should schedule rebroadcast.
+		driver.RequireNoBroadcast()                                          // Assert no broadcast until timeout expires.
+		driver.RequireDeliverAlarm()                                         // Expire the timeout.
+		requireRebroadcast()                                                 // Assert rebroadcast, which should schedule the next rebroadcast.
+		driver.RequireDeliverAlarm()                                         // Trigger alarm
+		requireRebroadcast()                                                 // Expect rebroadcast again, because phase timeout should be enough in the future since we are in round 13.
+		driver.AdvanceTimeBy(24 * time.Hour)                                 // Now, advance the clock beyond the rebroadcast timeout.
+		driver.RequireDeliverAlarm()                                         // Trigger alarm, which should trigger the phase timeout instead of rebroadcast timeout.
+		driver.RequirePrepareAtRound(13, instance.Proposal(), justification) // Expect progress to PREPARE phase; that is no rebroadcast.
+		driver.RequireNoBroadcast()                                          // Expect no further broadcast because PREPARE phase is not timed out yet.
+
+		// Now, because we are beyond round 3, we should expect rebroadcast even though the phase timeout
+		// hasn't expired yet. This is because the rebroadcast is set to trigger immediately beyond round 3.
+		// Therefore, assert that rebroadcast is triggered, and this time it includes a PREPARE message.
+		driver.RequireDeliverAlarm()
+		driver.RequireQuality()
+		driver.RequirePrepareAtRound(13, instance.Proposal(), justification)
+		driver.RequireConverge(13, instance.Proposal(), justification)
+		driver.RequireNoBroadcast() // Nothing else should be broadcast until the next alarm.
+	})
 }
 
 func TestGPBFT_WithExactOneThirdToTwoThirdPowerDistribution(t *testing.T) {

gpbft/options.go

@@ -25,9 +25,10 @@ type options struct {
 
 	qualityDeltaMulti float64
 
-	committeeLookback  uint64
-	maxLookaheadRounds uint64
-	rebroadcastAfter   func(int) time.Duration
+	committeeLookback                uint64
+	maxLookaheadRounds               uint64
+	rebroadcastAfter                 func(int) time.Duration
+	rebroadcastImmediatelyAfterRound uint64
 
 	maxCachedInstances           int
 	maxCachedMessagesPerInstance int
@@ -38,13 +39,14 @@ type options struct {
 
 func newOptions(o ...Option) (*options, error) {
 	opts := &options{
-		delta:                        defaultDelta,
-		deltaBackOffExponent:         defaultDeltaBackOffExponent,
-		qualityDeltaMulti:            1.0,
-		committeeLookback:            defaultCommitteeLookback,
-		rebroadcastAfter:             defaultRebroadcastAfter,
-		maxCachedInstances:           defaultMaxCachedInstances,
-		maxCachedMessagesPerInstance: defaultMaxCachedMessagesPerInstance,
+		delta:                            defaultDelta,
+		deltaBackOffExponent:             defaultDeltaBackOffExponent,
+		qualityDeltaMulti:                1.0,
+		committeeLookback:                defaultCommitteeLookback,
+		rebroadcastAfter:                 defaultRebroadcastAfter,
+		rebroadcastImmediatelyAfterRound: 3,
+		maxCachedInstances:               defaultMaxCachedInstances,
+		maxCachedMessagesPerInstance:     defaultMaxCachedMessagesPerInstance,
 	}
 	for _, apply := range o {
 		if err := apply(opts); err != nil {
@@ -143,6 +145,17 @@ func WithCommitteeLookback(lookback uint64) Option {
 	}
 }
 
+// WithRebroadcastImmediatelyAfterRound sets the round after which rebroadcast is
+// scheduled without waiting for phase timeout to expire first.
+//
+// Defaults to 3 if unset.
+func WithRebroadcastImmediatelyAfterRound(round uint64) Option {
+	return func(o *options) error {
+		o.rebroadcastImmediatelyAfterRound = round
+		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