From c0a27746bc6497924222c5d1f1008bfdc8dc2e4e Mon Sep 17 00:00:00 2001 From: Yoshihito Aso Date: Thu, 24 Sep 2026 14:03:41 +0900 Subject: [PATCH] Refactor timer management to event-driven model --- consensus/istanbul/qbft/core/core.go | 41 +- consensus/istanbul/qbft/core/events.go | 2 - consensus/istanbul/qbft/core/handler.go | 28 +- consensus/istanbul/qbft/core/preprepare.go | 13 +- consensus/istanbul/qbft/core/request.go | 18 +- consensus/istanbul/qbft/core/timer.go | 122 ++++++ consensus/istanbul/qbft/core/timer_test.go | 477 +++++++++++++++++++++ 7 files changed, 637 insertions(+), 64 deletions(-) create mode 100644 consensus/istanbul/qbft/core/timer.go create mode 100644 consensus/istanbul/qbft/core/timer_test.go diff --git a/consensus/istanbul/qbft/core/core.go b/consensus/istanbul/qbft/core/core.go index 39e6043ce..953990695 100644 --- a/consensus/istanbul/qbft/core/core.go +++ b/consensus/istanbul/qbft/core/core.go @@ -52,6 +52,7 @@ func New(backend istanbul.Backend, config *istanbul.Config) istanbul.Core { pendingRequests: prque.New(), pendingRequestsMu: new(sync.Mutex), consensusTimestamp: time.Time{}, + afterFunc: func(d time.Duration, f func()) timer { return time.AfterFunc(d, f) }, } c.validateFn = c.checkValidatorSignature @@ -66,11 +67,9 @@ type core struct { state State logger log.Logger - backend istanbul.Backend - events *event.TypeMuxSubscription - finalCommittedSub *event.TypeMuxSubscription - timeoutSub *event.TypeMuxSubscription - futurePreprepareTimer *time.Timer + backend istanbul.Backend + events *event.TypeMuxSubscription + finalCommittedSub *event.TypeMuxSubscription valSet istanbul.ValidatorSet validateFn func([]byte, []byte) (common.Address, error) @@ -82,8 +81,7 @@ type core struct { currentMutex sync.Mutex handlerWg *sync.WaitGroup - roundChangeSet *roundChangeSet - roundChangeTimer *time.Timer + roundChangeSet *roundChangeSet QBFTPreparedPrepares []*qbfttypes.Prepare @@ -92,8 +90,12 @@ type core struct { consensusTimestamp time.Time - newRoundMutex sync.Mutex - newRoundTimer *time.Timer + // Timer state belongs to the event loop. Callbacks only send immutable events. + timers [timerKindCount]scheduledTimer + afterFunc func(time.Duration, func()) timer + timerEvents chan timerEvent + timerDone chan struct{} + stopOnce sync.Once } func (c *core) currentView() *istanbul.View { @@ -186,6 +188,8 @@ func (c *core) startNewRound(round *big.Int) { c.valSet = c.backend.Validators(lastProposal) } + c.stopTimer() + // New snapshot for new round c.updateRoundState(newView, c.valSet, roundChange) @@ -243,24 +247,17 @@ func (c *core) Address() common.Address { } func (c *core) stopFuturePreprepareTimer() { - if c.futurePreprepareTimer != nil { - c.futurePreprepareTimer.Stop() - } + c.cancelTimer(futurePreprepareTimer) } func (c *core) stopTimer() { - c.stopFuturePreprepareTimer() - if c.roundChangeTimer != nil { - c.roundChangeTimer.Stop() + for kind := timerKind(0); kind < timerKindCount; kind++ { + c.cancelTimer(kind) } } func (c *core) newRoundChangeTimer() { - c.stopTimer() - - for c.current == nil { // wait because it is asynchronous in handleRequest - time.Sleep(10 * time.Millisecond) - } + c.stopFuturePreprepareTimer() // set timeout based on the round number baseTimeout := time.Duration(c.config.GetConfig(c.current.Sequence()).RequestTimeout) * time.Millisecond @@ -294,9 +291,7 @@ func (c *core) newRoundChangeTimer() { } c.currentLogger(true, nil).Trace("QBFT: start new ROUND-CHANGE timer", "timeout", timeout.Seconds()) - c.roundChangeTimer = time.AfterFunc(timeout, func() { - c.sendEvent(timeoutEvent{}) - }) + c.scheduleTimer(roundChangeTimer, timeout, timerEvent{}) } func (c *core) checkValidatorSignature(data []byte, sig []byte) (common.Address, error) { diff --git a/consensus/istanbul/qbft/core/events.go b/consensus/istanbul/qbft/core/events.go index fdf8fb33b..c4195ec8e 100644 --- a/consensus/istanbul/qbft/core/events.go +++ b/consensus/istanbul/qbft/core/events.go @@ -25,5 +25,3 @@ type backlogEvent struct { src istanbul.Validator msg qbfttypes.QBFTMessage } - -type timeoutEvent struct{} diff --git a/consensus/istanbul/qbft/core/handler.go b/consensus/istanbul/qbft/core/handler.go index e77f7087a..1841d549a 100644 --- a/consensus/istanbul/qbft/core/handler.go +++ b/consensus/istanbul/qbft/core/handler.go @@ -19,6 +19,7 @@ package core import ( "fmt" "math/big" + "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus/istanbul" @@ -32,12 +33,15 @@ func (c *core) Start() error { c.logger.Info("QBFT: start") // Tests will handle events itself, so we have to make subscribeEvents() // be able to call in test. + c.timerEvents = make(chan timerEvent) + c.timerDone = make(chan struct{}) + c.stopOnce = sync.Once{} c.subscribeEvents() - c.handlerWg.Add(1) - go c.handleEvents() - // Start a new round from last sequence + 1 + // Initialize consensus before the event loop can receive messages. c.startNewRound(common.Big0) + c.handlerWg.Add(1) + go c.handleEvents() return nil } @@ -45,7 +49,7 @@ func (c *core) Start() error { // Stop implements core.Engine.Stop func (c *core) Stop() error { c.logger.Info("QBFT: stopping...") - c.stopTimer() + c.stopOnce.Do(func() { close(c.timerDone) }) c.unsubscribeEvents() // Make sure the handler goroutine exits @@ -65,9 +69,6 @@ func (c *core) subscribeEvents() { // internal events backlogEvent{}, ) - c.timeoutSub = c.backend.EventMux().Subscribe( - timeoutEvent{}, - ) c.finalCommittedSub = c.backend.EventMux().Subscribe( istanbul.FinalCommittedEvent{}, ) @@ -76,7 +77,6 @@ func (c *core) subscribeEvents() { // Unsubscribe all events func (c *core) unsubscribeEvents() { c.events.Unsubscribe() - c.timeoutSub.Unsubscribe() c.finalCommittedSub.Unsubscribe() } @@ -94,12 +94,18 @@ func (c *core) unsubscribeEvents() { func (c *core) handleEvents() { // Clear state defer func() { + c.stopOnce.Do(func() { close(c.timerDone) }) + c.stopTimer() c.current = nil c.handlerWg.Done() }() for { select { + case <-c.timerDone: + return + case event := <-c.timerEvents: + c.handleTimerEvent(event) case event, ok := <-c.events.Chan(): if !ok { return @@ -141,12 +147,6 @@ func (c *core) handleEvents() { // if successfully processed, we gossip message to other validators c.backend.Gossip(c.valSet, ev.msg.Code(), data) } - case _, ok := <-c.timeoutSub.Chan(): - // we received a round change timeout - if !ok { - return - } - c.handleTimeoutMsg() case event, ok := <-c.finalCommittedSub.Chan(): // our block proposal got committed if !ok { diff --git a/consensus/istanbul/qbft/core/preprepare.go b/consensus/istanbul/qbft/core/preprepare.go index 098a37c41..0ea28e890 100644 --- a/consensus/istanbul/qbft/core/preprepare.go +++ b/consensus/istanbul/qbft/core/preprepare.go @@ -132,15 +132,8 @@ func (c *core) handlePreprepareMsg(preprepare *qbfttypes.Preprepare) error { if err == consensus.ErrFutureBlock { logger.Info("QBFT: PRE-PREPARE block proposal is in the future (will be treated again later)", "duration", duration) - // start a timer to re-input PRE-PREPARE message as a backlog event - c.stopFuturePreprepareTimer() - c.futurePreprepareTimer = time.AfterFunc(duration, func() { - _, validator := c.valSet.GetByAddress(preprepare.Source()) - c.sendEvent(backlogEvent{ - src: validator, - msg: preprepare, - }) - }) + // Retry through the event loop only while this view and timer remain valid. + c.scheduleTimer(futurePreprepareTimer, duration, timerEvent{preprepare: preprepare}) } else { logger.Warn("QBFT: invalid PRE-PREPARE block proposal", "err", err) } @@ -152,6 +145,8 @@ func (c *core) handlePreprepareMsg(preprepare *qbfttypes.Preprepare) error { if c.state == StateAcceptRequest { c.logger.Info("QBFT: accepted PRE-PREPARE message") + c.cancelTimer(emptyBlockTimer) + // Re-initialize ROUND-CHANGE timer c.newRoundChangeTimer() c.consensusTimestamp = time.Now() diff --git a/consensus/istanbul/qbft/core/request.go b/consensus/istanbul/qbft/core/request.go index 782ec73b4..68d33f041 100644 --- a/consensus/istanbul/qbft/core/request.go +++ b/consensus/istanbul/qbft/core/request.go @@ -44,6 +44,7 @@ func (c *core) handleRequest(request *Request) error { return err } + c.cancelTimer(emptyBlockTimer) c.current.pendingRequest = request if c.state == StateAcceptRequest { config := c.config.GetConfig(c.current.Sequence()) @@ -54,14 +55,6 @@ func (c *core) handleRequest(request *Request) error { // Send PRE-PREPARE message to other validators c.sendPreprepareMsg(request) } else { // emptyBlockPeriod is set - c.newRoundMutex.Lock() - defer c.newRoundMutex.Unlock() - - if c.newRoundTimer != nil { - c.newRoundTimer.Stop() - c.newRoundTimer = nil - } - delay := time.Duration(0) block, ok := request.Proposal.(*types.Block) @@ -77,14 +70,7 @@ func (c *core) handleRequest(request *Request) error { } } if delay > 0 { - c.newRoundTimer = time.AfterFunc(delay, func() { - c.newRoundTimer = nil - // Start ROUND-CHANGE timer - c.newRoundChangeTimer() - - // Send PRE-PREPARE message to other validators - c.sendPreprepareMsg(request) - }) + c.scheduleTimer(emptyBlockTimer, delay, timerEvent{proposalHash: request.Proposal.Hash()}) } else { // Start ROUND-CHANGE timer c.newRoundChangeTimer() diff --git a/consensus/istanbul/qbft/core/timer.go b/consensus/istanbul/qbft/core/timer.go new file mode 100644 index 000000000..3d3b22dbb --- /dev/null +++ b/consensus/istanbul/qbft/core/timer.go @@ -0,0 +1,122 @@ +package core + +import ( + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus/istanbul" + qbfttypes "github.com/ethereum/go-ethereum/consensus/istanbul/qbft/types" + "github.com/ethereum/go-ethereum/rlp" +) + +type timer interface { + Stop() bool +} + +type timerKind uint8 + +const ( + roundChangeTimer timerKind = iota + emptyBlockTimer + futurePreprepareTimer + timerKindCount +) + +func (kind timerKind) String() string { + switch kind { + case roundChangeTimer: + return "round-change" + case emptyBlockTimer: + return "empty-block" + case futurePreprepareTimer: + return "future-preprepare" + default: + return "unknown" + } +} + +type scheduledTimer struct { + timer timer + generation uint64 +} + +type timerEvent struct { + kind timerKind + view istanbul.View + generation uint64 + proposalHash common.Hash + preprepare *qbfttypes.Preprepare +} + +// cancelTimer also invalidates callbacks which have already fired. All timer +// registration, cancellation and event handling runs on the consensus loop. +func (c *core) cancelTimer(kind timerKind) { + slot := &c.timers[kind] + if slot.timer != nil { + slot.timer.Stop() + slot.timer = nil + c.currentLogger(true, nil).Trace("QBFT: stop timer", "kind", kind, "generation", slot.generation) + } + slot.generation++ +} + +func (c *core) scheduleTimer(kind timerKind, delay time.Duration, ev timerEvent) { + c.cancelTimer(kind) + slot := &c.timers[kind] + ev.kind = kind + ev.view = *c.currentView() // Copy big.Int values; do not capture mutable round state. + ev.generation = slot.generation + // Capture this execution's channels, so an old callback cannot enter a + // restarted core. Callbacks never read or mutate consensus or timer state. + events, done := c.timerEvents, c.timerDone + slot.timer = c.afterFunc(delay, func() { + select { + case events <- ev: + case <-done: + } + }) + c.currentLogger(true, nil).Trace("QBFT: start timer", "kind", kind, "generation", ev.generation, "delay", delay) +} + +func (c *core) handleTimerEvent(ev timerEvent) { + select { + case <-c.timerDone: + return + default: + } + if ev.kind >= timerKindCount { + return + } + slot := &c.timers[ev.kind] + logger := c.currentLogger(true, nil).New("kind", ev.kind, "generation", ev.generation, + "timer.sequence", ev.view.Sequence, "timer.round", ev.view.Round) + if slot.timer == nil || ev.generation != slot.generation || c.current == nil || ev.view.Cmp(c.currentView()) != 0 { + logger.Trace("QBFT: discard timer event", "reason", "inactive timer or stale generation/view") + return + } + c.cancelTimer(ev.kind) // Consume once, even if the callback is delivered twice. + logger.Trace("QBFT: handle timer event") + switch ev.kind { + case roundChangeTimer: + c.handleTimeoutMsg() + case emptyBlockTimer: + request := c.current.pendingRequest + if c.state != StateAcceptRequest || request == nil || request.Proposal.Hash() != ev.proposalHash { + logger.Trace("QBFT: discard timer event", "reason", "proposal or state changed") + return + } + c.newRoundChangeTimer() + c.sendPreprepareMsg(request) + case futurePreprepareTimer: + // Apply the normal view/state checks again before retrying verification. + if err := c.handleDecodedMessage(ev.preprepare); err != nil { + return + } + data, err := rlp.EncodeToBytes(ev.preprepare) + if err != nil { + logger.Error("QBFT: can not encode deferred PRE-PREPARE", "err", err) + return + } + c.backend.Gossip(c.valSet, ev.preprepare.Code(), data) + } +} diff --git a/consensus/istanbul/qbft/core/timer_test.go b/consensus/istanbul/qbft/core/timer_test.go new file mode 100644 index 000000000..da702b946 --- /dev/null +++ b/consensus/istanbul/qbft/core/timer_test.go @@ -0,0 +1,477 @@ +package core + +import ( + "math/big" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/consensus/istanbul" + qbfttypes "github.com/ethereum/go-ethereum/consensus/istanbul/qbft/types" + "github.com/ethereum/go-ethereum/consensus/istanbul/validator" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/event" +) + +// manualTimer deliberately allows firing after Stop: a real AfterFunc callback +// may already be running when it is cancelled. +type manualTimer struct { + mu sync.Mutex + stopped bool + delay time.Duration + fire func() +} + +func (m *manualTimer) Stop() bool { + m.mu.Lock() + defer m.mu.Unlock() + wasActive := !m.stopped + m.stopped = true + return wasActive +} + +type timerTestClock struct { + created chan *manualTimer +} + +func (clock *timerTestClock) afterFunc(d time.Duration, f func()) timer { + m := &manualTimer{delay: d, fire: f} + clock.created <- m + return m +} + +// Only methods used by these consensus scenarios are implemented. An unexpected +// backend interaction fails the test rather than silently succeeding. +type timerTestBackend struct { + istanbul.Backend + address common.Address + validators istanbul.ValidatorSet + mux event.TypeMux + last *types.Block + verify func(istanbul.Proposal) (time.Duration, error) + broadcasts chan uint64 +} + +func (b *timerTestBackend) Address() common.Address { return b.address } +func (b *timerTestBackend) EventMux() *event.TypeMux { return &b.mux } +func (b *timerTestBackend) Validators(istanbul.Proposal) istanbul.ValidatorSet { + return b.validators +} +func (b *timerTestBackend) LastProposal() (istanbul.Proposal, common.Address) { + return b.last, common.Address{} +} +func (b *timerTestBackend) HasBadProposal(common.Hash) bool { return false } +func (b *timerTestBackend) Sign([]byte) ([]byte, error) { return make([]byte, 65), nil } +func (b *timerTestBackend) Broadcast(_ istanbul.ValidatorSet, code uint64, _ []byte) error { + b.broadcasts <- code + return nil +} +func (b *timerTestBackend) Gossip(istanbul.ValidatorSet, uint64, []byte) error { return nil } +func (b *timerTestBackend) Verify(p istanbul.Proposal) (time.Duration, error) { + if b.verify != nil { + return b.verify(p) + } + return 0, nil +} + +func timerBlock(number int64, withTx bool) *types.Block { + b := types.NewBlockWithHeader(&types.Header{Number: big.NewInt(number), Time: uint64(number), Difficulty: big.NewInt(1)}) + if withTx { + b = b.WithBody([]*types.Transaction{types.NewTransaction(0, common.Address{}, big.NewInt(1), 21000, big.NewInt(0), nil)}, nil) + } + return b +} + +func newTimerTestCore(t *testing.T) (*core, *timerTestBackend, *timerTestClock) { + t.Helper() + cfg := *istanbul.DefaultConfig + cfg.EmptyBlockPeriod = 10 + cfg.ProposerPolicy = istanbul.NewRoundRobinProposerPolicy() + addresses := []common.Address{{1}, {2}, {3}, {4}} + b := &timerTestBackend{ + address: addresses[0], validators: validator.NewSet(addresses, cfg.ProposerPolicy), + last: timerBlock(588, false), broadcasts: make(chan uint64, 100), + } + c := New(b, &cfg).(*core) + clock := &timerTestClock{created: make(chan *manualTimer, 100)} + c.afterFunc = clock.afterFunc + c.timerEvents = make(chan timerEvent) + c.timerDone = make(chan struct{}) + c.startNewRound(common.Big0) + t.Cleanup(func() { + c.stopOnce.Do(func() { close(c.timerDone) }) + c.stopTimer() + b.mux.Stop() + }) + return c, b, clock +} + +func receiveTimer(t *testing.T, clock *timerTestClock) *manualTimer { + t.Helper() + select { + case m := <-clock.created: + return m + case <-time.After(5 * time.Second): + t.Fatal("timer was not registered") + return nil + } +} + +func captureTimerEvent(t *testing.T, c *core, m *manualTimer) timerEvent { + t.Helper() + done := make(chan struct{}) + go func() { m.fire(); close(done) }() + select { + case ev := <-c.timerEvents: + <-done + return ev + case <-time.After(5 * time.Second): + t.Fatal("timer callback did not deliver an event") + return timerEvent{} + } +} + +func acceptTimerProposal(t *testing.T, c *core, p istanbul.Proposal) { + t.Helper() + msg := qbfttypes.NewPreprepare(c.current.Sequence(), c.current.Round(), p) + msg.SetSource(c.valSet.GetProposer().Address()) + if err := c.handlePreprepareMsg(msg); err != nil { + t.Fatal(err) + } +} + +func TestTimerEmptyDelayAndPreprepare(t *testing.T) { + for _, tc := range []struct { + name string + delayFirst bool + proposer bool + }{ + {"proposer/preprepare-first", false, true}, + {"proposer/delay-first", true, true}, + {"validator/preprepare-first", false, false}, + {"validator/delay-first", true, false}, + } { + t.Run(tc.name, func(t *testing.T) { + c, b, clock := newTimerTestCore(t) + if !tc.proposer { + b.address = common.Address{2} + c.address = b.address + } + r := &Request{Proposal: timerBlock(589, false)} + if err := c.handleRequest(r); err != nil { + t.Fatal(err) + } + delay := receiveTimer(t, clock) + if delay.delay != 9*time.Second { + t.Fatalf("delay = %v", delay.delay) + } + ev := captureTimerEvent(t, c, delay) + var obsolete *manualTimer + if tc.delayFirst { + c.handleTimerEvent(ev) + obsolete = receiveTimer(t, clock) + } + acceptTimerProposal(t, c, r.Proposal) + current := receiveTimer(t, clock) + if !tc.delayFirst { + c.handleTimerEvent(ev) + } + if c.timers[emptyBlockTimer].timer != nil || c.timers[roundChangeTimer].timer != current { + t.Fatal("PRE-PREPARE did not leave exactly the current round timer") + } + if obsolete != nil { + c.handleTimerEvent(captureTimerEvent(t, c, obsolete)) + if c.current.Round().Sign() != 0 { + t.Fatal("obsolete timer changed round") + } + } + valid := captureTimerEvent(t, c, current) + c.handleTimerEvent(valid) + c.handleTimerEvent(valid) + if c.current.Round().Uint64() != 1 { + t.Fatal("valid timeout must change round exactly once") + } + }) + } +} + +func TestTimerStaleTimeoutAcrossBlocks(t *testing.T) { + c, b, clock := newTimerTestCore(t) + r := &Request{Proposal: timerBlock(589, false)} + if err := c.handleRequest(r); err != nil { + t.Fatal(err) + } + c.handleTimerEvent(captureTimerEvent(t, c, receiveTimer(t, clock))) + old := receiveTimer(t, clock) + acceptTimerProposal(t, c, r.Proposal) + receiveTimer(t, clock) + b.last = r.Proposal.(*types.Block) + if err := c.handleFinalCommitted(); err != nil { + t.Fatal(err) + } + r = &Request{Proposal: timerBlock(590, true)} + if err := c.handleRequest(r); err != nil { + t.Fatal(err) + } + receiveTimer(t, clock) + acceptTimerProposal(t, c, r.Proposal) + receiveTimer(t, clock) + b.last = r.Proposal.(*types.Block) + if err := c.handleFinalCommitted(); err != nil { + t.Fatal(err) + } + if err := c.handleRequest(&Request{Proposal: timerBlock(591, false)}); err != nil { + t.Fatal(err) + } + delay := receiveTimer(t, clock) + c.handleTimerEvent(captureTimerEvent(t, c, old)) + if c.current.Sequence().Int64() != 591 || c.current.Round().Sign() != 0 || c.timers[emptyBlockTimer].timer != delay { + t.Fatal("old block's timeout affected the current empty block") + } +} + +func TestTimerDelayedProposalInvalidation(t *testing.T) { + for _, action := range []string{"replacement", "preprepare", "committed-state", "final-committed", "round-change"} { + t.Run(action, func(t *testing.T) { + c, b, clock := newTimerTestCore(t) + r := &Request{Proposal: timerBlock(589, false)} + if err := c.handleRequest(r); err != nil { + t.Fatal(err) + } + old := captureTimerEvent(t, c, receiveTimer(t, clock)) + switch action { + case "replacement": + if err := c.handleRequest(&Request{Proposal: timerBlock(589, true)}); err != nil { + t.Fatal(err) + } + case "preprepare": + acceptTimerProposal(t, c, r.Proposal) + case "committed-state": + c.setState(StateCommitted) + case "final-committed": + b.last = r.Proposal.(*types.Block) + if err := c.handleFinalCommitted(); err != nil { + t.Fatal(err) + } + case "round-change": + c.startNewRound(common.Big1) + } + broadcasts, timers := len(b.broadcasts), len(clock.created) + c.handleTimerEvent(old) + if len(b.broadcasts) != broadcasts || len(clock.created) != timers { + t.Fatal("invalidated delay broadcast a proposal or reset a timer") + } + }) + } +} + +func TestTimerRequestPeriods(t *testing.T) { + for _, tc := range []struct { + name string + empty uint64 + block uint64 + tx bool + delay time.Duration + kind timerKind + }{ + {"disabled", 0, 1, false, 10 * time.Second, roundChangeTimer}, + {"equal", 1, 1, false, 10 * time.Second, roundChangeTimer}, + {"shorter", 1, 2, false, 10 * time.Second, roundChangeTimer}, + {"empty", 10, 1, false, 9 * time.Second, emptyBlockTimer}, + {"transaction", 10, 1, true, 10 * time.Second, roundChangeTimer}, + } { + t.Run(tc.name, func(t *testing.T) { + c, b, clock := newTimerTestCore(t) + c.config.EmptyBlockPeriod = tc.empty + c.config.BlockPeriod = tc.block + r := &Request{Proposal: timerBlock(589, tc.tx)} + originalTime := r.Proposal.(*types.Block).Time() + if err := c.handleRequest(r); err != nil { + t.Fatal(err) + } + m := receiveTimer(t, clock) + if m.delay != tc.delay || c.timers[tc.kind].timer != m { + t.Fatal("unexpected request timer") + } + if tc.kind == emptyBlockTimer { + if len(b.broadcasts) != 0 || r.Proposal.(*types.Block).Time() != originalTime+9 { + t.Fatal("incorrect empty block timing") + } + c.handleTimerEvent(captureTimerEvent(t, c, m)) + } + if len(b.broadcasts) != 1 || <-b.broadcasts != qbfttypes.PreprepareCode { + t.Fatal("expected a proposal broadcast") + } + }) + } +} + +func TestTimerRoundTimeoutBackoff(t *testing.T) { + for _, tc := range []struct { + round int64 + cap uint64 + want time.Duration + }{ + {0, 0, 10 * time.Second}, {1, 0, 20 * time.Second}, {2, 0, 40 * time.Second}, + {2, 30, 30 * time.Second}, {100, 30, 30 * time.Second}, + } { + c, _, clock := newTimerTestCore(t) + c.current.SetRound(big.NewInt(tc.round)) + c.config.MaxRequestTimeoutSeconds = tc.cap + c.newRoundChangeTimer() + if got := receiveTimer(t, clock).delay; got != tc.want { + t.Fatalf("round %d: %v, want %v", tc.round, got, tc.want) + } + } +} + +func TestTimerFuturePreprepare(t *testing.T) { + for _, stale := range []bool{false, true} { + c, b, clock := newTimerTestCore(t) + calls := 0 + b.verify = func(istanbul.Proposal) (time.Duration, error) { + calls++ + if calls == 1 { + return time.Second, consensus.ErrFutureBlock + } + return 0, nil + } + p := qbfttypes.NewPreprepare(c.current.Sequence(), c.current.Round(), timerBlock(589, false)) + p.SetSource(c.valSet.GetProposer().Address()) + if err := c.handlePreprepareMsg(p); err != consensus.ErrFutureBlock { + t.Fatalf("error = %v", err) + } + m := receiveTimer(t, clock) + if stale { + c.startNewRound(common.Big1) + } + c.handleTimerEvent(captureTimerEvent(t, c, m)) + if stale { + if calls != 1 { + t.Fatal("retried stale PRE-PREPARE") + } + } else if calls != 2 || c.state != StatePreprepared { + t.Fatal("valid future PRE-PREPARE was not accepted") + } + } +} + +func TestTimerStopRestart(t *testing.T) { + c, b, clock := newTimerTestCore(t) + if err := c.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Stop() }) + if err := b.mux.Post(istanbul.RequestEvent{Proposal: timerBlock(589, false)}); err != nil { + t.Fatal(err) + } + old := receiveTimer(t, clock) + // Race delivery against Stop. Both a blocked sender and an already consumed + // event must allow shutdown to finish. + delivered := make(chan struct{}) + go func() { old.fire(); close(delivered) }() + stopped := make(chan struct{}) + go func() { c.Stop(); close(stopped) }() + for _, done := range []chan struct{}{delivered, stopped} { + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("shutdown blocked") + } + } + for len(clock.created) > 0 { + <-clock.created + } + for len(b.broadcasts) > 0 { + <-b.broadcasts + } + if err := c.Start(); err != nil { + t.Fatal(err) + } + // This callback still targets the old channels and must return immediately. + delivered = make(chan struct{}) + go func() { old.fire(); close(delivered) }() + select { + case <-delivered: + case <-time.After(5 * time.Second): + t.Fatal("old callback entered restarted core") + } + if err := b.mux.Post(istanbul.RequestEvent{Proposal: timerBlock(589, false)}); err != nil { + t.Fatal(err) + } + current := receiveTimer(t, clock) + current.fire() + select { + case code := <-b.broadcasts: + if code != qbfttypes.PreprepareCode { + t.Fatalf("unexpected message: %d", code) + } + case <-time.After(5 * time.Second): + t.Fatal("restarted core failed to propose") + } + if err := c.Stop(); err != nil { + t.Fatal(err) + } +} + +func TestTimerFuturePreprepareReplacement(t *testing.T) { + c, b, clock := newTimerTestCore(t) + calls := 0 + b.verify = func(istanbul.Proposal) (time.Duration, error) { + calls++ + return time.Second, consensus.ErrFutureBlock + } + p := qbfttypes.NewPreprepare(c.current.Sequence(), c.current.Round(), timerBlock(589, false)) + p.SetSource(c.valSet.GetProposer().Address()) + for i := 0; i < 2; i++ { + if err := c.handlePreprepareMsg(p); err != consensus.ErrFutureBlock { + t.Fatalf("error = %v", err) + } + } + old, current := receiveTimer(t, clock), receiveTimer(t, clock) + c.handleTimerEvent(captureTimerEvent(t, c, old)) + if calls != 2 || c.timers[futurePreprepareTimer].timer != current { + t.Fatal("superseded retry affected the current timer") + } + b.verify = nil + c.handleTimerEvent(captureTimerEvent(t, c, current)) + if c.state != StatePreprepared { + t.Fatal("replacement retry was not accepted") + } +} + +// Exercise the production AfterFunc adapter and the real event loop as well as +// the deterministic, manually delivered timer scenarios above. +func TestTimerRealTimeout(t *testing.T) { + c, b, _ := newTimerTestCore(t) + c.afterFunc = func(d time.Duration, f func()) timer { return time.AfterFunc(d, f) } + c.config.EmptyBlockPeriod = 0 + c.config.RequestTimeout = 1 + if err := c.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Stop() }) + if err := b.mux.Post(istanbul.RequestEvent{Proposal: timerBlock(589, true)}); err != nil { + t.Fatal(err) + } + for _, want := range []uint64{qbfttypes.PreprepareCode, qbfttypes.RoundChangeCode} { + select { + case got := <-b.broadcasts: + if got != want { + t.Fatalf("message = %d, want %d", got, want) + } + case <-time.After(5 * time.Second): + t.Fatal("real timer failed to trigger a round change") + } + } + if err := c.Stop(); err != nil { + t.Fatal(err) + } + for _, slot := range c.timers { + if slot.timer != nil { + t.Fatal("timer remained active after shutdown") + } + } +}