From 24c1b1d6926b922179f70ccefd98b7b8bcf8ecba Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 18:04:51 -0400 Subject: [PATCH 1/7] refactor(pkg/finality-grandpa): make closing globalIn the voter's shutdown signal Voter.Stop both signalled shutdown and tore the voter down, while the caller separately owned the globalIn channel. Neither alone was sufficient, and that split is where the forwarder leak lived. Closing globalIn is now the only shutdown signal, and Stop becomes Wait: it blocks until the Start loop has returned, then releases what the voter itself owns. - processIncoming returns a sentinel when globalIn closes, and Start reports it as a nil error, since an orderly shutdown is not a failure. - stopChan is gone. Removing it is what fixes the leak: the poll loop no longer has an early exit, so it can only leave by observing the close, which means first draining whatever the forwarder is holding on the unbuffered out. - wakerChan.start wakes its consumer on exit. Without it the poll loop would sit on the waker until some unrelated source happened to fire. - Wait takes the WaitGroup token out in NewVoter and arbitrates ownership with runState. Start runs in a goroutine, so Wait can otherwise reach wg.Wait before Start reaches wg.Add, and proceed to tear down a voter coming up. - Wait is idempotent, so an owner that both supervises the voter and shuts the node down can reach it twice. Handle closed channels at the other wakerChan consumers too, where an unchecked receive stayed permanently ready and handed out zero values: pruneBackgroundRounds spun while holding v.inner, which also blocked VoterState; votingRound.processIncoming dereferenced a nil Message interface; and finishPrevoting read a closed channel as "no best chain", parking the round in prevoting forever. Fix a pre-existing race in waker.wake, which read w.wakeCh inside the spawned goroutine after releasing the lock, racing register's write. Give the test network per-node teardown. route delivered outside the lock, so closing a single node's channel could race a send. --- pkg/finality-grandpa/bridge_state.go | 10 +- pkg/finality-grandpa/environment_test.go | 32 ++- pkg/finality-grandpa/voter.go | 115 ++++++++-- pkg/finality-grandpa/voter_lifecycle_test.go | 226 +++++++++++++++++++ pkg/finality-grandpa/voter_test.go | 35 ++- pkg/finality-grandpa/voting_round.go | 20 +- 6 files changed, 399 insertions(+), 39 deletions(-) create mode 100644 pkg/finality-grandpa/voter_lifecycle_test.go diff --git a/pkg/finality-grandpa/bridge_state.go b/pkg/finality-grandpa/bridge_state.go index 1e18608bb1..81fa892b40 100644 --- a/pkg/finality-grandpa/bridge_state.go +++ b/pkg/finality-grandpa/bridge_state.go @@ -17,13 +17,17 @@ func newWaker() *waker { } func (w *waker) wake() { + // Read the channel under the lock and hand the value to the goroutine. Reading + // w.wakeCh inside the goroutine would be unsynchronised: it runs after wake has + // returned and released the lock, so it can race register's write. w.RLock() - defer w.RUnlock() - if w.wakeCh == nil { + ch := w.wakeCh + w.RUnlock() + if ch == nil { return } go func() { - w.wakeCh <- struct{}{} + ch <- struct{}{} }() } diff --git a/pkg/finality-grandpa/environment_test.go b/pkg/finality-grandpa/environment_test.go index e6286b4f99..3983c08f8e 100644 --- a/pkg/finality-grandpa/environment_test.go +++ b/pkg/finality-grandpa/environment_test.go @@ -259,13 +259,30 @@ func (bm *BroadcastNetwork[M, N]) AddNode(f func(N) M, out chan N) (in chan M) { func (bm *BroadcastNetwork[M, N]) route() { defer bm.routeWG.Done() for msg := range bm.receiver { + // Deliver under the lock. RemoveNode closes a node's channel, and closing a + // channel a producer is about to send on panics, so the two have to be + // serialised. Senders are buffered, so this does not block in practice. bm.mu.Lock() bm.history = append(bm.history, msg) - senders := append([]chan M(nil), bm.senders...) - bm.mu.Unlock() - for _, sender := range senders { + for _, sender := range bm.senders { sender <- msg } + bm.mu.Unlock() + } +} + +// RemoveNode deregisters a node's inbound channel and closes it. Closing that +// channel is how the voter reading it is asked to shut down; it happens under +// bm.mu so it cannot race a delivery in route. +func (bm *BroadcastNetwork[M, N]) RemoveNode(in chan M) { + bm.mu.Lock() + defer bm.mu.Unlock() + for i, sender := range bm.senders { + if sender == in { + bm.senders = append(bm.senders[:i], bm.senders[i+1:]...) + close(in) + return + } } } @@ -402,6 +419,15 @@ func (n *Network) MakeGlobalComms( }, out) } +// StopGlobalComms closes the inbound channel handed to a voter by +// MakeGlobalComms, which is how that voter is shut down. +func (n *Network) StopGlobalComms(in chan GlobalInItem[string, uint32, Signature, ID]) { + n.mtx.Lock() + defer n.mtx.Unlock() + + n.globalMessages.RemoveNode(in) +} + func (n *Network) SendMessage(message CommunicationIn[string, uint32, Signature, ID]) { n.globalMessages.SendMessage(GlobalInItem[string, uint32, Signature, ID]{message, nil}) } diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index 2b0d48afac..54efc56020 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -4,6 +4,7 @@ package grandpa import ( + "errors" "fmt" "sync" "sync/atomic" @@ -29,7 +30,15 @@ func newWakerChan[Item any](in chan Item) *wakerChan[Item] { } func (wc *wakerChan[Item]) start() { - defer close(wc.out) + defer func() { + close(wc.out) + // Wake the consumer so it observes the close on its next poll. Closing the + // input is how a caller shuts the voter down, so without this the poll loop + // would sit on the waker until some unrelated source happened to fire. + if w := wc.waker.Load(); w != nil { + w.wake() + } + }() if wc.in == nil { return } @@ -535,11 +544,31 @@ type Voter[Hash constraints.Ordered, Number constraints.Unsigned, Signature comp // assumptions from round-to-round. lastFinalizedInRounds HashNumber[Hash, Number] - stopTimeout time.Duration - stopChan chan struct{} - wg sync.WaitGroup + // waitTimeout bounds how long Wait will block for the Start loop to return. + waitTimeout time.Duration + // wg holds a single token, taken out in NewVoter. Start releases it when its + // loop returns; Wait releases it instead when Start never ran. Taking it in + // Start would race Wait, which reaches wg.Wait as soon as it is called. + wg sync.WaitGroup + waitOnce sync.Once + waitErr error + // runState decides which of Start and Wait owns releasing the wg token. + runState atomic.Int32 } +// Voter lifecycle states for Voter.runState. +const ( + voterFresh int32 = iota + voterRunning + voterDone +) + +// errVoterShutdown travels back through the poll path when the globalIn channel +// given to NewVoter is closed, which is how a caller asks the voter to stop. It +// is a shutdown signal rather than a failure, so Start reports it as a nil +// error; it never reaches the caller. +var errVoterShutdown = errors.New("voter shutdown: global incoming stream closed") + // NewVoter creates a new `Voter` tracker with given round number and base block. // // Provide data about the last completed round. If there is no @@ -597,7 +626,7 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c bestRound: bestRound, pastRounds: *pastRounds, } - return &Voter[Hash, Number, Signature, ID]{ + v := &Voter[Hash, Number, Signature, ID]{ env: env, voters: voters, inner: inner, @@ -606,9 +635,12 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c lastFinalizedInRounds: lastFinalized, globalIn: newWakerChan(globalIn), globalOut: newBuffered(globalOutPresend), - stopChan: make(chan struct{}), - stopTimeout: 30 * time.Second, + waitTimeout: 30 * time.Second, } + // Held until Start's loop returns, or until Wait claims it because Start + // never ran. + v.wg.Add(1) + return v } func (v *Voter[Hash, Number, Signature, ID]) pruneBackgroundRounds(waker *waker) error { @@ -644,7 +676,18 @@ pastRounds: finalizedNotifications: for { select { - case notif := <-v.finalizedNotifications.channel(): + case notif, ok := <-v.finalizedNotifications.channel(): + // This channel is the voter's own, and Stop closes it only after Start + // has returned, so a live poll loop cannot legitimately see it closed: + // that means the voter was torn down and is being polled anyway. The + // stream carrying finalization is gone and cannot be reopened. + // (Unchecked, the receive would also stay ready forever, yielding + // zero-value notifications that change nothing, spinning here while + // holding v.inner — which blocks VoterState too.) + if !ok { + v.inner.Unlock() + return fmt.Errorf("finalization notification stream closed") + } fNum := notif.Number v.inner.pastRounds.UpdateFinalized(fNum) if v.setLastFinalizedNumber(fNum) { @@ -681,7 +724,15 @@ func (v *Voter[Hash, Number, Signature, ID]) processIncoming(waker *waker) error loop: for { select { - case item := <-v.globalIn.channel(): + case item, ok := <-v.globalIn.channel(): + // The forwarder closes this channel when globalIn ends, which is how a + // caller shuts the voter down. Unwinding through the poll path is what + // ends the Start loop. (Unchecked, the receive would also stay ready + // forever, handing out zero-value items that match no case below and + // spinning the loop.) + if !ok { + return errVoterShutdown + } if item.Error != nil { return item.Error } @@ -894,30 +945,56 @@ func (v *Voter[Hash, Number, Signature, ID]) setLastFinalizedNumber(finalizedNum return false } +// Start runs the voter until the globalIn channel given to NewVoter is closed, +// which is how a caller asks it to shut down. It blocks, so callers run it in a +// goroutine and join it with Wait. It returns nil on an orderly shutdown, and an +// error if the voter failed. +// +// A voter runs once: a second Start, or a Start after Wait, declines. func (v *Voter[Hash, Number, Signature, ID]) Start() error { //skipcq: RVV-B0001 - v.wg.Add(1) + if !v.runState.CompareAndSwap(voterFresh, voterRunning) { + return fmt.Errorf("voter has already been started") + } defer v.wg.Done() waker := newWaker() for { ready, err := v.poll(waker) if err != nil { + if errors.Is(err, errVoterShutdown) { + return nil + } return err } if ready { return nil } - select { - case <-waker.channel(): - case <-v.stopChan: - return fmt.Errorf("early voter stop") - } + <-waker.channel() } } -func (v *Voter[Hash, Number, Signature, ID]) Stop() error { - close(v.stopChan) +// Wait blocks until the Start loop has returned, then releases what the voter +// itself owns: its finalization channel and every round timer. +// +// Wait does not signal shutdown — close the globalIn channel passed to NewVoter +// to do that. Calling Wait without closing it blocks until waitTimeout and +// reports an error, leaving the voter running. +// +// Idempotent: later calls block until the first has finished and return the same +// result. +func (v *Voter[Hash, Number, Signature, ID]) Wait() error { + v.waitOnce.Do(func() { v.waitErr = v.wait() }) + return v.waitErr +} + +func (v *Voter[Hash, Number, Signature, ID]) wait() error { v.globalOut.Close() - timeout := time.NewTimer(v.stopTimeout) + // Start never ran and now never will, so release its token or the wait below + // would block for the whole timeout. + if v.runState.CompareAndSwap(voterFresh, voterDone) { + v.wg.Done() + } + timeout := time.NewTimer(v.waitTimeout) + defer timeout.Stop() wgDone := make(chan struct{}) go func() { defer close(wgDone) @@ -925,7 +1002,7 @@ func (v *Voter[Hash, Number, Signature, ID]) Stop() error { }() select { case <-timeout.C: - return fmt.Errorf("timeout for Voter.Stop()") + return fmt.Errorf("timeout waiting for the voter to stop: was globalIn closed?") case <-wgDone: } diff --git a/pkg/finality-grandpa/voter_lifecycle_test.go b/pkg/finality-grandpa/voter_lifecycle_test.go new file mode 100644 index 0000000000..134e533cfc --- /dev/null +++ b/pkg/finality-grandpa/voter_lifecycle_test.go @@ -0,0 +1,226 @@ +// Copyright 2023 ChainSafe Systems (ON) +// SPDX-License-Identifier: LGPL-3.0-only + +package grandpa + +import ( + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type lifecycleItem = GlobalInItem[string, uint32, Signature, ID] + +// newLifecycleVoter builds a voter over its own inbound channel, for tests about +// starting and stopping rather than about voting. +func newLifecycleVoter(t *testing.T, network *Network, globalIn chan lifecycleItem, +) *Voter[string, uint32, Signature, ID] { + t.Helper() + var localID ID = 5 + voters := NewVoterSet([]IDWeight[ID]{{localID, 100}}) + + env := newEnvironment(network, localID) + var lastFinalized HashNumber[string, uint32] + env.WithChain(func(chain *dummyChain) { + chain.PushBlocks(GenesisHash, []string{"A", "B", "C"}) + lastFinalized.Hash, lastFinalized.Number = chain.LastFinalized() + }) + + return NewVoter[string, uint32, Signature, ID]( + &env, *voters, globalIn, + func(CommunicationOut[string, uint32, Signature, ID]) error { return nil }, + 0, nil, lastFinalized, lastFinalized, + ) +} + +// forwardersInState counts wakerChan.start goroutines blocked in the given +// runtime state: "chan send" is one parked on the unbuffered `out` holding an +// item, "chan receive" one waiting on `in`. +func forwardersInState(state string) int { + buf := make([]byte, 1<<22) + n := runtime.Stack(buf, true) + count := 0 + for _, blk := range strings.Split(string(buf[:n]), "\ngoroutine ") { + nl := strings.Index(blk, "\n") + if nl < 0 { + continue + } + if strings.Contains(blk[:nl], state) && + strings.Contains(blk[nl:], "wakerChan") && strings.Contains(blk[nl:], ".start(") { + count++ + } + } + return count +} + +// Closing globalIn is the shutdown signal: Start returns of its own accord, and +// reports nil because an orderly shutdown is not a failure. +func TestVoter_CloseGlobalInShutsDown(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + + errCh := make(chan error, 1) + go func() { errCh <- v.Start() }() + time.Sleep(50 * time.Millisecond) + + close(globalIn) + select { + case err := <-errCh: + assert.NoError(t, err, "an orderly shutdown is not an error") + case <-time.After(5 * time.Second): + t.Fatal("Start did not return after globalIn was closed") + } + assert.NoError(t, v.Wait()) +} + +// The retired voter's forwarder must be gone once Wait returns. There is no +// stopChan any more, so the poll loop cannot exit before it has drained what the +// forwarder is holding: it only leaves by observing the close. +func TestVoter_RotationLeavesNoForwarder(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + baseSend := forwardersInState("chan send") + baseRecv := forwardersInState("chan receive") + + const rotations = 10 + for i := 0; i < rotations; i++ { + globalIn := make(chan lifecycleItem, 100) + v := newLifecycleVoter(t, network, globalIn) + + done := make(chan struct{}) + go func() { defer close(done); _ = v.Start() }() + time.Sleep(30 * time.Millisecond) + + // Inbound traffic in flight when the set rotates. + for j := 0; j < 100; j++ { + select { + case globalIn <- lifecycleItem{}: + default: + } + } + + close(globalIn) + <-done + require.NoError(t, v.Wait()) + } + + time.Sleep(300 * time.Millisecond) + send := forwardersInState("chan send") - baseSend + recv := forwardersInState("chan receive") - baseRecv + t.Logf("after %d rotations: send-parked=%+d recv-parked=%+d", rotations, send, recv) + assert.Zero(t, send, "a retired voter's globalIn forwarder is still holding an item") +} + +// Rebuilding the voter across an authority-set rotation: close, wait, build the +// next one, start it. Start is launched from a goroutine, so a rotation can reach +// the next Wait before that Start has begun. +func TestVoter_RebuildAcrossRotations(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + var starts sync.WaitGroup + globalIn := make(chan lifecycleItem, 100) + first := newLifecycleVoter(t, network, globalIn) + voter := first + starts.Add(1) + go func() { defer starts.Done(); _ = first.Start() }() + + const rotations = 100 + for i := 0; i < rotations; i++ { + close(globalIn) + require.NoError(t, voter.Wait()) + + globalIn = make(chan lifecycleItem, 100) + v := newLifecycleVoter(t, network, globalIn) + voter = v + starts.Add(1) + go func() { defer starts.Done(); _ = v.Start() }() + } + close(globalIn) + require.NoError(t, voter.Wait()) + + done := make(chan struct{}) + go func() { starts.Wait(); close(done) }() + select { + case <-done: + case <-time.After(30 * time.Second): + t.Fatal("a Start goroutine never returned") + } +} + +// Wait does not signal shutdown. Calling it without closing globalIn must report +// a timeout rather than block forever or claim success. +func TestVoter_WaitWithoutCloseTimesOut(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + v.waitTimeout = 200 * time.Millisecond + + done := make(chan struct{}) + go func() { defer close(done); _ = v.Start() }() + time.Sleep(50 * time.Millisecond) + + assert.ErrorContains(t, v.Wait(), "timeout waiting for the voter to stop") + + close(globalIn) + <-done +} + +// Wait is idempotent, and concurrent callers all get the same answer. +func TestVoter_WaitIsIdempotent(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + + done := make(chan struct{}) + go func() { defer close(done); _ = v.Start() }() + time.Sleep(50 * time.Millisecond) + close(globalIn) + <-done + + const callers = 4 + errs := make([]error, callers) + var wg sync.WaitGroup + for i := 0; i < callers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + errs[i] = v.Wait() + }(i) + } + wg.Wait() + for i, err := range errs { + assert.Equal(t, errs[0], err, "Wait caller %d saw a different result", i) + } + assert.NoError(t, v.Wait()) +} + +// A voter runs once. Waiting on one that never started must not sit out the whole +// timeout, and a later Start must decline rather than come up into a torn-down +// voter. +func TestVoter_WaitBeforeStart(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + v.waitTimeout = 10 * time.Second + + began := time.Now() + assert.NoError(t, v.Wait()) + assert.Less(t, time.Since(began), time.Second, "Wait sat out its timeout for a Start that never ran") + assert.Error(t, v.Start(), "a voter that has been waited on must not start") +} diff --git a/pkg/finality-grandpa/voter_test.go b/pkg/finality-grandpa/voter_test.go index da0a4cef72..1f7dc96506 100644 --- a/pkg/finality-grandpa/voter_test.go +++ b/pkg/finality-grandpa/voter_test.go @@ -50,12 +50,13 @@ func TestVoter_TalkingToMyself(t *testing.T) { go func() { defer close(done) err := voter.Start() - // stops early, so this should return an error - assert.Error(t, err) + // closing globalIn is an orderly shutdown, not a failure + assert.NoError(t, err) }() <-finalized - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) <-done } @@ -104,7 +105,8 @@ func TestVoter_FinalizingAtFaultThreshold(t *testing.T) { go func() { defer wg.Done() <-finalized - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) }() } @@ -127,6 +129,7 @@ func TestVoter_ExposingVoterState(t *testing.T) { var wg sync.WaitGroup voters := make([]*Voter[string, uint32, Signature, ID], votersOnline) voterStates := make([]VoterState[ID], votersOnline) + globalIns := make([]chan GlobalInItem[string, uint32, Signature, ID], votersOnline) // some voters offline for i := 0; i < votersOnline; i++ { localID := ID(i) @@ -157,6 +160,7 @@ func TestVoter_ExposingVoterState(t *testing.T) { voters[i] = voter voterStates[i] = voter.VoterState() + globalIns[i] = globalIn wg.Add(1) go func() { @@ -203,8 +207,9 @@ func TestVoter_ExposingVoterState(t *testing.T) { }{2, expectedRoundState}, ) - for _, v := range voters { - err := v.Stop() + for i, v := range voters { + network.StopGlobalComms(globalIns[i]) + err := v.Wait() assert.NoError(t, err) } } @@ -246,7 +251,8 @@ func TestVoter_BroadcastCommit(t *testing.T) { go voter.Start() <-commitsIn - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) } @@ -351,7 +357,8 @@ waitForCommits: } assert.Equal(t, 1, commitCount) - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) } @@ -417,7 +424,8 @@ func TestVoter_ImportCommitForAnyRound(t *testing.T) { finalized := <-env.FinalizedStream() assert.Equal(t, finalized.Commit, commit) - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) } @@ -542,7 +550,8 @@ func TestVoter_SkipsToLatestRoundAfterCatchUp(t *testing.T) { }, voterState.Get().BackgroundRounds[5]) - err := unsyncedVoter.Stop() + network.StopGlobalComms(globalIn) + err := unsyncedVoter.Wait() assert.NoError(t, err) } @@ -584,7 +593,8 @@ func TestVoter_PickUpFromPriorWithoutGrandparentState(t *testing.T) { } } - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) } @@ -690,7 +700,8 @@ waitForPrevote: <-env.concludedCalled assert.Equal(t, [2]uint64{2, 1}, env.LastCompletedAndConcluded()) - err := voter.Stop() + network.StopGlobalComms(globalIn) + err := voter.Wait() assert.NoError(t, err) } diff --git a/pkg/finality-grandpa/voting_round.go b/pkg/finality-grandpa/voting_round.go index 70808b3558..426d79069f 100644 --- a/pkg/finality-grandpa/voting_round.go +++ b/pkg/finality-grandpa/voting_round.go @@ -4,6 +4,7 @@ package grandpa import ( + "fmt" "time" "golang.org/x/exp/constraints" @@ -459,7 +460,15 @@ func (vr *votingRound[Hash, Number, Signature, ID, E]) processIncoming(waker *wa while: for { select { - case incoming := <-vr.incoming.channel(): + case incoming, ok := <-vr.incoming.channel(): + // roundData.Incoming belongs to the environment. A round that is still + // voting cannot recover from losing it, and the zero value would carry a + // nil Message interface, which handleVote dereferences. Unchecked, the + // receive also stays permanently ready, so the default arm never runs + // and the 1ms timerChan escape below is never armed. + if !ok { + return fmt.Errorf("round %d: incoming message stream closed", vr.roundNumber()) + } log.Tracef("Round %d: Got incoming message", vr.roundNumber()) if timer != nil { timer.Stop() @@ -590,8 +599,15 @@ func (vr *votingRound[Hash, Number, Signature, ID, E]) prevote(w *waker, lastRou wakerChan := newWakerChan(bestChain) wakerChan.setWaker(waker) var best *HashNumber[Hash, Number] - res := <-wakerChan.channel() + res, ok := <-wakerChan.channel() switch { + case !ok: + // The environment owns bestChain and always sends one value before + // closing, so an empty closed channel means the stream went away. Kept + // distinct from the default arm below: a real {nil, nil} value means + // "no best chain yet" and legitimately parks the round in prevoting, + // whereas a closed channel would park it there forever. + return fmt.Errorf("round %d: best chain stream closed", vr.roundNumber()) case res.Error != nil: return res.Error case res.Value != nil: From af6438b7747e31a6362cca9f26130a2d44db48c9 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 18:19:03 -0400 Subject: [PATCH 2/7] refactor(pkg/finality-grandpa): start the voter in NewVoter and drop Start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Start existed only so callers could schedule the poll loop themselves, and every one of them did the same thing with it: run it in a goroutine straight after construction. The window it opened between building a voter and running one is where the lifecycle bugs lived — Wait could reach wg.Wait before Start reached wg.Add, and a voter could be torn down while coming up. NewVoter now runs the voter, so there is no such window and no half-live state to observe. That removes the machinery that existed to police it: the runState atomic and its CAS in both Start and Wait, the lifecycle constants, the WaitGroup token arbitrated between two owners, and the "already started" error path. What replaces it is a done channel the run loop closes, which also publishes runErr. Add Done, returning that channel. Owners that supervise a voter alongside other work can select on it instead of committing to a blocking Wait, which is what voterWork was hand-rolling with an error channel and a goroutine. The public surface is now NewVoter, Done, Wait and VoterState, with close(globalIn) as the shutdown signal. Tests no longer wire globalIn after construction. Passing a placeholder and then overwriting voter.globalIn was always a smell; against a running voter it would be a data race, so globalIn is now built before NewVoter and passed in. --- internal/client/consensus/grandpa/grandpa.go | 16 +-- pkg/finality-grandpa/voter.go | 108 ++++++++--------- pkg/finality-grandpa/voter_lifecycle_test.go | 115 ++++++++----------- pkg/finality-grandpa/voter_test.go | 78 +++---------- 4 files changed, 128 insertions(+), 189 deletions(-) diff --git a/internal/client/consensus/grandpa/grandpa.go b/internal/client/consensus/grandpa/grandpa.go index 4c592b3546..ec45c641c9 100644 --- a/internal/client/consensus/grandpa/grandpa.go +++ b/internal/client/consensus/grandpa/grandpa.go @@ -436,7 +436,7 @@ type voterWork[ E runtime.Extrinsic, ] struct { voter *grandpa.Voter[H, N, primitives.AuthoritySignature, primitives.AuthorityID] - voterErrChan <-chan error + voterDone <-chan struct{} sharedVoterState *SharedVoterState[primitives.AuthorityID] env *environment[H, N, Hasher, Header, E] voterCommandsRx <-chan voterCommand @@ -554,14 +554,9 @@ func (vw *voterWork[H, N, Hasher, Header, E]) rebuildVoter() { // Repoint shared_voter_state so that the RPC endpoint can query the state vw.sharedVoterState.reset(voter.VoterState()) + // NewVoter runs the voter; Done fires when it stops, and Wait yields why. vw.voter = voter - errChan := make(chan error) - go func() { - err := voter.Start() - errChan <- err - close(errChan) - }() - vw.voterErrChan = errChan + vw.voterDone = voter.Done() case voterSetStatePaused[H, N]: default: panic("unreachable") @@ -651,9 +646,10 @@ func (vw *voterWork[H, N, Hasher, Header, E]) handleVoterCommand(command voterCo func (vw *voterWork[H, N, Hasher, Header, E]) poll() error { select { - case err := <-vw.voterErrChan: + case <-vw.voterDone: + err := vw.voter.Wait() if err == nil { - // voters don't conclude naturally + // nothing here closes globalIn, so the voter has no orderly way to stop return fmt.Errorf("consensus-grandpa inner voter has concluded: %w", ErrSafety) } vc, isVoterCommand := err.(voterCommand) diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index 54efc56020..3be29f2706 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -544,32 +544,35 @@ type Voter[Hash constraints.Ordered, Number constraints.Unsigned, Signature comp // assumptions from round-to-round. lastFinalizedInRounds HashNumber[Hash, Number] - // waitTimeout bounds how long Wait will block for the Start loop to return. + // waitTimeout bounds how long Wait blocks for the run loop to finish. waitTimeout time.Duration - // wg holds a single token, taken out in NewVoter. Start releases it when its - // loop returns; Wait releases it instead when Start never ran. Taking it in - // Start would race Wait, which reaches wg.Wait as soon as it is called. - wg sync.WaitGroup - waitOnce sync.Once - waitErr error - // runState decides which of Start and Wait owns releasing the wg token. - runState atomic.Int32 + // done is closed by the run loop as it returns, publishing runErr with it. + done chan struct{} + // runErr is written by the run loop before done is closed, so a read after + // receiving from done is ordered. + runErr error + // teardownOnce keeps the teardown to a single run: it closes channels, so a + // second pass would panic, and an owner that both supervises the voter and + // shuts the node down can reach Wait twice. + teardownOnce sync.Once + teardownErr error } -// Voter lifecycle states for Voter.runState. -const ( - voterFresh int32 = iota - voterRunning - voterDone -) - // errVoterShutdown travels back through the poll path when the globalIn channel // given to NewVoter is closed, which is how a caller asks the voter to stop. It // is a shutdown signal rather than a failure, so Start reports it as a nil // error; it never reaches the caller. var errVoterShutdown = errors.New("voter shutdown: global incoming stream closed") -// NewVoter creates a new `Voter` tracker with given round number and base block. +// NewVoter creates a new `Voter` tracker with given round number and base block +// and starts it running. There is no separate start step, so a voter is never +// observable in a constructed-but-idle state. +// +// The voter runs until globalIn is closed, which is how a caller asks it to shut +// down. Closing it belongs to the caller, who must therefore be the only writer +// to it by that point, or must serialise its writers against the close. Wait +// then blocks for the voter to finish and releases what it owns; Done offers the +// same signal without blocking. // // Provide data about the last completed round. If there is no // known last completed round, the genesis state (round number 0, no votes, genesis base), @@ -636,10 +639,12 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c globalIn: newWakerChan(globalIn), globalOut: newBuffered(globalOutPresend), waitTimeout: 30 * time.Second, + done: make(chan struct{}), } - // Held until Start's loop returns, or until Wait claims it because Start - // never ran. - v.wg.Add(1) + go func() { + defer close(v.done) + v.runErr = v.run() + }() return v } @@ -945,17 +950,9 @@ func (v *Voter[Hash, Number, Signature, ID]) setLastFinalizedNumber(finalizedNum return false } -// Start runs the voter until the globalIn channel given to NewVoter is closed, -// which is how a caller asks it to shut down. It blocks, so callers run it in a -// goroutine and join it with Wait. It returns nil on an orderly shutdown, and an -// error if the voter failed. -// -// A voter runs once: a second Start, or a Start after Wait, declines. -func (v *Voter[Hash, Number, Signature, ID]) Start() error { //skipcq: RVV-B0001 - if !v.runState.CompareAndSwap(voterFresh, voterRunning) { - return fmt.Errorf("voter has already been started") - } - defer v.wg.Done() +// run is the voter's poll loop. NewVoter starts it; it ends when the globalIn +// channel closes, or on the first error. +func (v *Voter[Hash, Number, Signature, ID]) run() error { waker := newWaker() for { ready, err := v.poll(waker) @@ -972,40 +969,43 @@ func (v *Voter[Hash, Number, Signature, ID]) Start() error { //skipcq: RVV-B0001 } } -// Wait blocks until the Start loop has returned, then releases what the voter -// itself owns: its finalization channel and every round timer. +// Wait blocks until the voter has finished, then releases what it owns: its +// finalization channel and every round timer. It returns nil on an orderly +// shutdown, and the voter's error if it failed. // // Wait does not signal shutdown — close the globalIn channel passed to NewVoter -// to do that. Calling Wait without closing it blocks until waitTimeout and -// reports an error, leaving the voter running. +// to do that. Calling Wait without closing it reports a timeout and leaves the +// voter running. // -// Idempotent: later calls block until the first has finished and return the same -// result. +// Idempotent, and safe to call from several goroutines: the teardown runs once +// and every caller gets the same result. func (v *Voter[Hash, Number, Signature, ID]) Wait() error { - v.waitOnce.Do(func() { v.waitErr = v.wait() }) - return v.waitErr -} - -func (v *Voter[Hash, Number, Signature, ID]) wait() error { - v.globalOut.Close() - // Start never ran and now never will, so release its token or the wait below - // would block for the whole timeout. - if v.runState.CompareAndSwap(voterFresh, voterDone) { - v.wg.Done() - } timeout := time.NewTimer(v.waitTimeout) defer timeout.Stop() - wgDone := make(chan struct{}) - go func() { - defer close(wgDone) - v.wg.Wait() - }() select { + case <-v.done: case <-timeout.C: + // Deliberately does not touch runErr: the loop is still running and still + // owns it. return fmt.Errorf("timeout waiting for the voter to stop: was globalIn closed?") - case <-wgDone: } + v.teardownOnce.Do(func() { v.teardownErr = v.teardown() }) + if v.teardownErr != nil { + return v.teardownErr + } + // Ordered by the receive from done above. + return v.runErr +} + +// Done is closed when the run loop has finished. It lets an owner select on the +// voter's termination alongside its own work; call Wait afterwards for the +// result and to release what the voter owns. +func (v *Voter[Hash, Number, Signature, ID]) Done() <-chan struct{} { + return v.done +} +func (v *Voter[Hash, Number, Signature, ID]) teardown() error { + v.globalOut.Close() close(v.finalizedNotifications.in) switch state := v.inner.bestRound.state.(type) { case statePrecommitted: diff --git a/pkg/finality-grandpa/voter_lifecycle_test.go b/pkg/finality-grandpa/voter_lifecycle_test.go index 134e533cfc..83ec394e9c 100644 --- a/pkg/finality-grandpa/voter_lifecycle_test.go +++ b/pkg/finality-grandpa/voter_lifecycle_test.go @@ -17,7 +17,7 @@ import ( type lifecycleItem = GlobalInItem[string, uint32, Signature, ID] // newLifecycleVoter builds a voter over its own inbound channel, for tests about -// starting and stopping rather than about voting. +// shutting down rather than about voting. NewVoter starts it. func newLifecycleVoter(t *testing.T, network *Network, globalIn chan lifecycleItem, ) *Voter[string, uint32, Signature, ID] { t.Helper() @@ -58,46 +58,64 @@ func forwardersInState(state string) int { return count } -// Closing globalIn is the shutdown signal: Start returns of its own accord, and -// reports nil because an orderly shutdown is not a failure. +// Closing globalIn is the shutdown signal: the voter stops of its own accord and +// Wait reports nil, because an orderly shutdown is not a failure. func TestVoter_CloseGlobalInShutsDown(t *testing.T) { network := NewNetwork() defer network.Stop() globalIn := make(chan lifecycleItem, 10) v := newLifecycleVoter(t, network, globalIn) - - errCh := make(chan error, 1) - go func() { errCh <- v.Start() }() time.Sleep(50 * time.Millisecond) close(globalIn) select { - case err := <-errCh: - assert.NoError(t, err, "an orderly shutdown is not an error") + case <-v.Done(): case <-time.After(5 * time.Second): - t.Fatal("Start did not return after globalIn was closed") + t.Fatal("the voter did not stop after globalIn was closed") } assert.NoError(t, v.Wait()) } +// Done lets an owner observe termination without committing to a blocking Wait, +// and is still closed for a voter that has already been waited on. +func TestVoter_DoneSignalsTermination(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + time.Sleep(50 * time.Millisecond) + + select { + case <-v.Done(): + t.Fatal("Done fired while the voter was still running") + default: + } + + close(globalIn) + require.NoError(t, v.Wait()) + + select { + case <-v.Done(): + default: + t.Fatal("Done was not closed after the voter stopped") + } +} + // The retired voter's forwarder must be gone once Wait returns. There is no -// stopChan any more, so the poll loop cannot exit before it has drained what the -// forwarder is holding: it only leaves by observing the close. +// early exit from the poll loop, so it can only leave by observing the close, +// which means first draining whatever the forwarder is holding. func TestVoter_RotationLeavesNoForwarder(t *testing.T) { network := NewNetwork() defer network.Stop() baseSend := forwardersInState("chan send") - baseRecv := forwardersInState("chan receive") const rotations = 10 for i := 0; i < rotations; i++ { globalIn := make(chan lifecycleItem, 100) v := newLifecycleVoter(t, network, globalIn) - - done := make(chan struct{}) - go func() { defer close(done); _ = v.Start() }() time.Sleep(30 * time.Millisecond) // Inbound traffic in flight when the set rotates. @@ -109,30 +127,22 @@ func TestVoter_RotationLeavesNoForwarder(t *testing.T) { } close(globalIn) - <-done require.NoError(t, v.Wait()) } time.Sleep(300 * time.Millisecond) - send := forwardersInState("chan send") - baseSend - recv := forwardersInState("chan receive") - baseRecv - t.Logf("after %d rotations: send-parked=%+d recv-parked=%+d", rotations, send, recv) - assert.Zero(t, send, "a retired voter's globalIn forwarder is still holding an item") + assert.Equal(t, baseSend, forwardersInState("chan send"), + "a retired voter's globalIn forwarder is still holding an item") } -// Rebuilding the voter across an authority-set rotation: close, wait, build the -// next one, start it. Start is launched from a goroutine, so a rotation can reach -// the next Wait before that Start has begun. +// Rebuilding the voter across an authority-set rotation. With no separate start +// step there is no window in which a half-live voter can be torn down. func TestVoter_RebuildAcrossRotations(t *testing.T) { network := NewNetwork() defer network.Stop() - var starts sync.WaitGroup globalIn := make(chan lifecycleItem, 100) - first := newLifecycleVoter(t, network, globalIn) - voter := first - starts.Add(1) - go func() { defer starts.Done(); _ = first.Start() }() + voter := newLifecycleVoter(t, network, globalIn) const rotations = 100 for i := 0; i < rotations; i++ { @@ -140,25 +150,15 @@ func TestVoter_RebuildAcrossRotations(t *testing.T) { require.NoError(t, voter.Wait()) globalIn = make(chan lifecycleItem, 100) - v := newLifecycleVoter(t, network, globalIn) - voter = v - starts.Add(1) - go func() { defer starts.Done(); _ = v.Start() }() + voter = newLifecycleVoter(t, network, globalIn) } close(globalIn) require.NoError(t, voter.Wait()) - - done := make(chan struct{}) - go func() { starts.Wait(); close(done) }() - select { - case <-done: - case <-time.After(30 * time.Second): - t.Fatal("a Start goroutine never returned") - } } // Wait does not signal shutdown. Calling it without closing globalIn must report -// a timeout rather than block forever or claim success. +// a timeout rather than block forever or claim success, and must leave the voter +// running. func TestVoter_WaitWithoutCloseTimesOut(t *testing.T) { network := NewNetwork() defer network.Stop() @@ -166,15 +166,19 @@ func TestVoter_WaitWithoutCloseTimesOut(t *testing.T) { globalIn := make(chan lifecycleItem, 10) v := newLifecycleVoter(t, network, globalIn) v.waitTimeout = 200 * time.Millisecond - - done := make(chan struct{}) - go func() { defer close(done); _ = v.Start() }() time.Sleep(50 * time.Millisecond) assert.ErrorContains(t, v.Wait(), "timeout waiting for the voter to stop") + select { + case <-v.Done(): + t.Fatal("a timed-out Wait stopped the voter") + default: + } + // The voter is still live, so the real shutdown still works. + v.waitTimeout = 5 * time.Second close(globalIn) - <-done + assert.NoError(t, v.Wait()) } // Wait is idempotent, and concurrent callers all get the same answer. @@ -184,12 +188,8 @@ func TestVoter_WaitIsIdempotent(t *testing.T) { globalIn := make(chan lifecycleItem, 10) v := newLifecycleVoter(t, network, globalIn) - - done := make(chan struct{}) - go func() { defer close(done); _ = v.Start() }() time.Sleep(50 * time.Millisecond) close(globalIn) - <-done const callers = 4 errs := make([]error, callers) @@ -207,20 +207,3 @@ func TestVoter_WaitIsIdempotent(t *testing.T) { } assert.NoError(t, v.Wait()) } - -// A voter runs once. Waiting on one that never started must not sit out the whole -// timeout, and a later Start must decline rather than come up into a torn-down -// voter. -func TestVoter_WaitBeforeStart(t *testing.T) { - network := NewNetwork() - defer network.Stop() - - globalIn := make(chan lifecycleItem, 10) - v := newLifecycleVoter(t, network, globalIn) - v.waitTimeout = 10 * time.Second - - began := time.Now() - assert.NoError(t, v.Wait()) - assert.Less(t, time.Since(began), time.Second, "Wait sat out its timeout for a Start that never ran") - assert.Error(t, v.Start(), "a voter that has been waited on must not start") -} diff --git a/pkg/finality-grandpa/voter_test.go b/pkg/finality-grandpa/voter_test.go index 1f7dc96506..a6eb467918 100644 --- a/pkg/finality-grandpa/voter_test.go +++ b/pkg/finality-grandpa/voter_test.go @@ -31,11 +31,12 @@ func TestVoter_TalkingToMyself(t *testing.T) { }) globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) finalized := env.FinalizedStream() voter := NewVoter[string, uint32, Signature, ID]( &env, *voters, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, @@ -43,22 +44,11 @@ func TestVoter_TalkingToMyself(t *testing.T) { lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - - done := make(chan struct{}) - go func() { - defer close(done) - err := voter.Start() - // closing globalIn is an orderly shutdown, not a failure - assert.NoError(t, err) - }() - <-finalized network.StopGlobalComms(globalIn) + // closing globalIn is an orderly shutdown, not a failure err := voter.Wait() assert.NoError(t, err) - <-done } func TestVoter_FinalizingAtFaultThreshold(t *testing.T) { @@ -86,10 +76,11 @@ func TestVoter_FinalizingAtFaultThreshold(t *testing.T) { // run voter in background. scheduling it to shut down at the end. finalized := env.FinalizedStream() globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voters, - make(chan GlobalInItem[string, uint32, Signature, ID]), + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, @@ -97,11 +88,7 @@ func TestVoter_FinalizingAtFaultThreshold(t *testing.T) { lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - wg.Add(1) - go voter.Start() go func() { defer wg.Done() <-finalized @@ -144,10 +131,11 @@ func TestVoter_ExposingVoterState(t *testing.T) { // run voter in background. scheduling it to shut down at the end. finalized := env.FinalizedStream() globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - make(chan GlobalInItem[string, uint32, Signature, ID]), + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, @@ -155,9 +143,6 @@ func TestVoter_ExposingVoterState(t *testing.T) { lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - voters[i] = voter voterStates[i] = voter.VoterState() globalIns[i] = globalIn @@ -194,9 +179,6 @@ func TestVoter_ExposingVoterState(t *testing.T) { voterState.Get(), ) - for _, v := range voters { - go v.Start() - } wg.Wait() assert.Equal(t, @@ -232,10 +214,11 @@ func TestVoter_BroadcastCommit(t *testing.T) { // run voter in background. scheduling it to shut down at the end. globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - make(chan GlobalInItem[string, uint32, Signature, ID]), + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, @@ -245,10 +228,6 @@ func TestVoter_BroadcastCommit(t *testing.T) { commitsIn := network.MakeGlobalComms(globalOut) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - - go voter.Start() <-commitsIn network.StopGlobalComms(globalIn) @@ -298,22 +277,17 @@ func TestVoter_BroadcastCommitOnlyIfNewer(t *testing.T) { // run voter in background. scheduling it to shut down at the end. globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, lastFinalized, lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - - go func() { - voter.Start() - }() item := <-roundIn // wait for a prevote @@ -396,10 +370,11 @@ func TestVoter_ImportCommitForAnyRound(t *testing.T) { // run voter in background. scheduling it to shut down at the end. globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, @@ -407,13 +382,6 @@ func TestVoter_ImportCommitForAnyRound(t *testing.T) { lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - - go func() { - voter.Start() - }() - // Send the commit message co := CommunicationOutCommit[string, uint32, Signature, ID]{ Number: 0, @@ -455,18 +423,17 @@ func TestVoter_SkipsToLatestRoundAfterCatchUp(t *testing.T) { }) globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) unsyncedVoter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 0, nil, lastFinalized, lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - unsyncedVoter.globalIn = newWakerChan(globalIn) prevote := func(id uint32) SignedPrevote[string, uint32, Signature, ID] { return SignedPrevote[string, uint32, Signature, ID]{ @@ -500,9 +467,6 @@ func TestVoter_SkipsToLatestRoundAfterCatchUp(t *testing.T) { _, ok := voterState.Get().BackgroundRounds[5] assert.False(t, ok) - // spawn the voter in the background - go unsyncedVoter.Start() - finalized := env.FinalizedStream() // wait until it's caught up, it should skip to round 6 and send a @@ -573,20 +537,18 @@ func TestVoter_PickUpFromPriorWithoutGrandparentState(t *testing.T) { // run voter in background. scheduling it to shut down at the end. globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 10, nil, lastFinalized, lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - go voter.Start() for finalized := range env.FinalizedStream() { if finalized.Number >= 6 { break @@ -665,19 +627,17 @@ func TestVoter_PickUpFromPriorWithGrandparentStatus(t *testing.T) { // run voter in background. scheduling it to shut down at the end. globalOut := make(chan CommunicationOut[string, uint32, Signature, ID]) + globalIn := network.MakeGlobalComms(globalOut) voter := NewVoter[string, uint32, Signature, ID]( &env, *voterSet, - nil, + globalIn, func(co CommunicationOut[string, uint32, Signature, ID]) error { globalOut <- co; return nil }, 1, lastRoundVotes, lastFinalized, lastFinalized, ) - globalIn := network.MakeGlobalComms(globalOut) - voter.globalIn = newWakerChan(globalIn) - go voter.Start() // wait until we see a prevote on round 3 from our local ID, // indicating that the round 3 has started. From be255b062a24375220619f4a8603fb8ffb4ffa19 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 18:36:27 -0400 Subject: [PATCH 3/7] refactor(pkg/finality-grandpa): deliver the voter's error on Done and drop Wait A voter has one owner, so the outcome does not need broadcast semantics. Done becomes a chan error carrying the terminal error directly, which collapses the two-step "select on Done, then call Wait" into one receive. Wait existed to make the teardown unskippable by binding it to the only way of learning why the voter stopped. Tearing down on the voter's own goroutine before publishing gives that for free and more strongly: receiving from Done means the voter has not merely stopped but finished releasing its finalization channel and every round timer. Callers have no teardown obligation at all. That removes the rest of the lifecycle machinery: Wait, the teardown sync.Once and its stored error, the separate runErr field, and waitTimeout along with the timeout error that was the one non-nil return meaning "still running". A caller wanting a deadline composes it against Done. The public surface is now NewVoter, Done and VoterState, with close(globalIn) as the shutdown signal. The trade is that the error goes to a single receiver and the channel closes behind it, so a later receive yields nil. That is documented on Done and pinned by a test. --- internal/client/consensus/grandpa/grandpa.go | 7 +- pkg/finality-grandpa/voter.go | 80 +++++------- pkg/finality-grandpa/voter_lifecycle_test.go | 123 ++++++++----------- pkg/finality-grandpa/voter_test.go | 18 +-- 4 files changed, 91 insertions(+), 137 deletions(-) diff --git a/internal/client/consensus/grandpa/grandpa.go b/internal/client/consensus/grandpa/grandpa.go index ec45c641c9..94ac233f10 100644 --- a/internal/client/consensus/grandpa/grandpa.go +++ b/internal/client/consensus/grandpa/grandpa.go @@ -436,7 +436,7 @@ type voterWork[ E runtime.Extrinsic, ] struct { voter *grandpa.Voter[H, N, primitives.AuthoritySignature, primitives.AuthorityID] - voterDone <-chan struct{} + voterDone <-chan error sharedVoterState *SharedVoterState[primitives.AuthorityID] env *environment[H, N, Hasher, Header, E] voterCommandsRx <-chan voterCommand @@ -554,7 +554,7 @@ func (vw *voterWork[H, N, Hasher, Header, E]) rebuildVoter() { // Repoint shared_voter_state so that the RPC endpoint can query the state vw.sharedVoterState.reset(voter.VoterState()) - // NewVoter runs the voter; Done fires when it stops, and Wait yields why. + // NewVoter runs the voter; Done yields why it stopped, once it has. vw.voter = voter vw.voterDone = voter.Done() case voterSetStatePaused[H, N]: @@ -646,8 +646,7 @@ func (vw *voterWork[H, N, Hasher, Header, E]) handleVoterCommand(command voterCo func (vw *voterWork[H, N, Hasher, Header, E]) poll() error { select { - case <-vw.voterDone: - err := vw.voter.Wait() + case err := <-vw.voterDone: if err == nil { // nothing here closes globalIn, so the voter has no orderly way to stop return fmt.Errorf("consensus-grandpa inner voter has concluded: %w", ErrSafety) diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index 3be29f2706..5dafb04c1b 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -544,18 +544,10 @@ type Voter[Hash constraints.Ordered, Number constraints.Unsigned, Signature comp // assumptions from round-to-round. lastFinalizedInRounds HashNumber[Hash, Number] - // waitTimeout bounds how long Wait blocks for the run loop to finish. - waitTimeout time.Duration - // done is closed by the run loop as it returns, publishing runErr with it. - done chan struct{} - // runErr is written by the run loop before done is closed, so a read after - // receiving from done is ordered. - runErr error - // teardownOnce keeps the teardown to a single run: it closes channels, so a - // second pass would panic, and an owner that both supervises the voter and - // shuts the node down can reach Wait twice. - teardownOnce sync.Once - teardownErr error + // done carries the voter's terminal error, published once the voter has torn + // itself down, and is closed straight after. Buffered so the voter never waits + // on an owner that has stopped listening. + done chan error } // errVoterShutdown travels back through the poll path when the globalIn channel @@ -638,12 +630,15 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c lastFinalizedInRounds: lastFinalized, globalIn: newWakerChan(globalIn), globalOut: newBuffered(globalOutPresend), - waitTimeout: 30 * time.Second, - done: make(chan struct{}), + done: make(chan error, 1), } go func() { - defer close(v.done) - v.runErr = v.run() + err := v.run() + // Tear down before publishing, so that receiving from done means the voter + // has not merely stopped but finished releasing what it owned. + v.teardown() + v.done <- err + close(v.done) }() return v } @@ -969,42 +964,31 @@ func (v *Voter[Hash, Number, Signature, ID]) run() error { } } -// Wait blocks until the voter has finished, then releases what it owns: its -// finalization channel and every round timer. It returns nil on an orderly -// shutdown, and the voter's error if it failed. +// Done yields the voter's terminal error once it has stopped: nil if it was shut +// down by closing the globalIn channel given to NewVoter, otherwise the error it +// failed with. Every error is terminal — the voter does not recover from one and +// carry on — so receiving here means the voter has finished and has already +// released what it owned. // -// Wait does not signal shutdown — close the globalIn channel passed to NewVoter -// to do that. Calling Wait without closing it reports a timeout and leaves the -// voter running. +// Select on it to supervise the voter alongside other work: // -// Idempotent, and safe to call from several goroutines: the teardown runs once -// and every caller gets the same result. -func (v *Voter[Hash, Number, Signature, ID]) Wait() error { - timeout := time.NewTimer(v.waitTimeout) - defer timeout.Stop() - select { - case <-v.done: - case <-timeout.C: - // Deliberately does not touch runErr: the loop is still running and still - // owns it. - return fmt.Errorf("timeout waiting for the voter to stop: was globalIn closed?") - } - v.teardownOnce.Do(func() { v.teardownErr = v.teardown() }) - if v.teardownErr != nil { - return v.teardownErr - } - // Ordered by the receive from done above. - return v.runErr -} - -// Done is closed when the run loop has finished. It lets an owner select on the -// voter's termination alongside its own work; call Wait afterwards for the -// result and to release what the voter owns. -func (v *Voter[Hash, Number, Signature, ID]) Done() <-chan struct{} { +// select { +// case err := <-voter.Done(): +// // the voter has stopped; err says why +// case cmd := <-commands: +// } +// +// The error is delivered to one receiver, and the channel is closed immediately +// after, so later receives yield nil. A voter has a single owner; if more than +// one party needs the outcome, that owner must fan it out. +func (v *Voter[Hash, Number, Signature, ID]) Done() <-chan error { return v.done } -func (v *Voter[Hash, Number, Signature, ID]) teardown() error { +// teardown releases what the voter owns: its finalization channel and every +// round timer. It runs on the voter's own goroutine once the run loop is done, +// so callers have no teardown obligation. +func (v *Voter[Hash, Number, Signature, ID]) teardown() { v.globalOut.Close() close(v.finalizedNotifications.in) switch state := v.inner.bestRound.state.(type) { @@ -1041,8 +1025,6 @@ func (v *Voter[Hash, Number, Signature, ID]) teardown() error { close(round.roundCommitter.importCommits.in) } } - - return nil } func (v *Voter[Hash, Number, Signature, ID]) poll(waker *waker) (bool, error) { //skipcq: RVV-B0001 diff --git a/pkg/finality-grandpa/voter_lifecycle_test.go b/pkg/finality-grandpa/voter_lifecycle_test.go index 83ec394e9c..0a4b075420 100644 --- a/pkg/finality-grandpa/voter_lifecycle_test.go +++ b/pkg/finality-grandpa/voter_lifecycle_test.go @@ -6,7 +6,6 @@ package grandpa import ( "runtime" "strings" - "sync" "testing" "time" @@ -58,8 +57,8 @@ func forwardersInState(state string) int { return count } -// Closing globalIn is the shutdown signal: the voter stops of its own accord and -// Wait reports nil, because an orderly shutdown is not a failure. +// Closing globalIn is the shutdown signal, and Done reports nil for it: an +// orderly shutdown is not a failure. func TestVoter_CloseGlobalInShutsDown(t *testing.T) { network := NewNetwork() defer network.Stop() @@ -68,18 +67,24 @@ func TestVoter_CloseGlobalInShutsDown(t *testing.T) { v := newLifecycleVoter(t, network, globalIn) time.Sleep(50 * time.Millisecond) - close(globalIn) select { case <-v.Done(): + t.Fatal("Done fired while the voter was still running") + default: + } + + close(globalIn) + select { + case err := <-v.Done(): + assert.NoError(t, err, "an orderly shutdown is not an error") case <-time.After(5 * time.Second): t.Fatal("the voter did not stop after globalIn was closed") } - assert.NoError(t, v.Wait()) } -// Done lets an owner observe termination without committing to a blocking Wait, -// and is still closed for a voter that has already been waited on. -func TestVoter_DoneSignalsTermination(t *testing.T) { +// Receiving from Done means the voter has finished releasing what it owned, not +// merely that its loop stopped. +func TestVoter_DoneImpliesTeardown(t *testing.T) { network := NewNetwork() defer network.Stop() @@ -87,25 +92,45 @@ func TestVoter_DoneSignalsTermination(t *testing.T) { v := newLifecycleVoter(t, network, globalIn) time.Sleep(50 * time.Millisecond) - select { - case <-v.Done(): - t.Fatal("Done fired while the voter was still running") - default: - } + close(globalIn) + require.NoError(t, <-v.Done()) + + // Teardown closes the voter's own finalization channel, which ends its + // forwarder and closes the channel the voter was reading. + require.Eventually(t, func() bool { + select { + case _, open := <-v.finalizedNotifications.channel(): + return !open + default: + return false + } + }, 2*time.Second, 10*time.Millisecond, "teardown did not release the finalization channel") +} + +// The error goes to one receiver and the channel closes behind it. A voter has a +// single owner; this pins the contract so it is not discovered by accident. +func TestVoter_DoneDeliversToOneReceiver(t *testing.T) { + network := NewNetwork() + defer network.Stop() + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + time.Sleep(50 * time.Millisecond) close(globalIn) - require.NoError(t, v.Wait()) + require.NoError(t, <-v.Done()) select { - case <-v.Done(): - default: - t.Fatal("Done was not closed after the voter stopped") + case err, open := <-v.Done(): + assert.False(t, open, "Done should be closed after delivering") + assert.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Done blocked after delivering; it should be closed") } } -// The retired voter's forwarder must be gone once Wait returns. There is no -// early exit from the poll loop, so it can only leave by observing the close, -// which means first draining whatever the forwarder is holding. +// The retired voter's forwarder must be gone once Done fires. There is no early +// exit from the poll loop, so it can only leave by observing the close, which +// means first draining whatever the forwarder is holding. func TestVoter_RotationLeavesNoForwarder(t *testing.T) { network := NewNetwork() defer network.Stop() @@ -127,7 +152,7 @@ func TestVoter_RotationLeavesNoForwarder(t *testing.T) { } close(globalIn) - require.NoError(t, v.Wait()) + require.NoError(t, <-v.Done()) } time.Sleep(300 * time.Millisecond) @@ -147,63 +172,11 @@ func TestVoter_RebuildAcrossRotations(t *testing.T) { const rotations = 100 for i := 0; i < rotations; i++ { close(globalIn) - require.NoError(t, voter.Wait()) + require.NoError(t, <-voter.Done()) globalIn = make(chan lifecycleItem, 100) voter = newLifecycleVoter(t, network, globalIn) } close(globalIn) - require.NoError(t, voter.Wait()) -} - -// Wait does not signal shutdown. Calling it without closing globalIn must report -// a timeout rather than block forever or claim success, and must leave the voter -// running. -func TestVoter_WaitWithoutCloseTimesOut(t *testing.T) { - network := NewNetwork() - defer network.Stop() - - globalIn := make(chan lifecycleItem, 10) - v := newLifecycleVoter(t, network, globalIn) - v.waitTimeout = 200 * time.Millisecond - time.Sleep(50 * time.Millisecond) - - assert.ErrorContains(t, v.Wait(), "timeout waiting for the voter to stop") - select { - case <-v.Done(): - t.Fatal("a timed-out Wait stopped the voter") - default: - } - - // The voter is still live, so the real shutdown still works. - v.waitTimeout = 5 * time.Second - close(globalIn) - assert.NoError(t, v.Wait()) -} - -// Wait is idempotent, and concurrent callers all get the same answer. -func TestVoter_WaitIsIdempotent(t *testing.T) { - network := NewNetwork() - defer network.Stop() - - globalIn := make(chan lifecycleItem, 10) - v := newLifecycleVoter(t, network, globalIn) - time.Sleep(50 * time.Millisecond) - close(globalIn) - - const callers = 4 - errs := make([]error, callers) - var wg sync.WaitGroup - for i := 0; i < callers; i++ { - wg.Add(1) - go func(i int) { - defer wg.Done() - errs[i] = v.Wait() - }(i) - } - wg.Wait() - for i, err := range errs { - assert.Equal(t, errs[0], err, "Wait caller %d saw a different result", i) - } - assert.NoError(t, v.Wait()) + require.NoError(t, <-voter.Done()) } diff --git a/pkg/finality-grandpa/voter_test.go b/pkg/finality-grandpa/voter_test.go index a6eb467918..188a8b61d1 100644 --- a/pkg/finality-grandpa/voter_test.go +++ b/pkg/finality-grandpa/voter_test.go @@ -47,7 +47,7 @@ func TestVoter_TalkingToMyself(t *testing.T) { <-finalized network.StopGlobalComms(globalIn) // closing globalIn is an orderly shutdown, not a failure - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } @@ -93,7 +93,7 @@ func TestVoter_FinalizingAtFaultThreshold(t *testing.T) { defer wg.Done() <-finalized network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) }() } @@ -191,7 +191,7 @@ func TestVoter_ExposingVoterState(t *testing.T) { for i, v := range voters { network.StopGlobalComms(globalIns[i]) - err := v.Wait() + err := <-v.Done() assert.NoError(t, err) } } @@ -231,7 +231,7 @@ func TestVoter_BroadcastCommit(t *testing.T) { <-commitsIn network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } @@ -332,7 +332,7 @@ waitForCommits: assert.Equal(t, 1, commitCount) network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } @@ -393,7 +393,7 @@ func TestVoter_ImportCommitForAnyRound(t *testing.T) { assert.Equal(t, finalized.Commit, commit) network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } @@ -515,7 +515,7 @@ func TestVoter_SkipsToLatestRoundAfterCatchUp(t *testing.T) { voterState.Get().BackgroundRounds[5]) network.StopGlobalComms(globalIn) - err := unsyncedVoter.Wait() + err := <-unsyncedVoter.Done() assert.NoError(t, err) } @@ -556,7 +556,7 @@ func TestVoter_PickUpFromPriorWithoutGrandparentState(t *testing.T) { } network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } @@ -661,7 +661,7 @@ waitForPrevote: assert.Equal(t, [2]uint64{2, 1}, env.LastCompletedAndConcluded()) network.StopGlobalComms(globalIn) - err := voter.Wait() + err := <-voter.Done() assert.NoError(t, err) } From 1c36d0692362b05ecb2d93be0a7ec9298f9dd863 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 18:39:45 -0400 Subject: [PATCH 4/7] docs(pkg/finality-grandpa): drop references to the removed Start and Wait Comments still described the lifecycle in terms of methods that no longer exist, including one predating this branch that referred to stalling Stop(). --- pkg/finality-grandpa/voter.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index 5dafb04c1b..87ba301edc 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -552,8 +552,8 @@ type Voter[Hash constraints.Ordered, Number constraints.Unsigned, Signature comp // errVoterShutdown travels back through the poll path when the globalIn channel // given to NewVoter is closed, which is how a caller asks the voter to stop. It -// is a shutdown signal rather than a failure, so Start reports it as a nil -// error; it never reaches the caller. +// is a shutdown signal rather than a failure, so the run loop reports it as a +// nil error; it never reaches the caller. var errVoterShutdown = errors.New("voter shutdown: global incoming stream closed") // NewVoter creates a new `Voter` tracker with given round number and base block @@ -562,9 +562,8 @@ var errVoterShutdown = errors.New("voter shutdown: global incoming stream closed // // The voter runs until globalIn is closed, which is how a caller asks it to shut // down. Closing it belongs to the caller, who must therefore be the only writer -// to it by that point, or must serialise its writers against the close. Wait -// then blocks for the voter to finish and releases what it owns; Done offers the -// same signal without blocking. +// to it by that point, or must serialise its writers against the close. Done +// then yields why the voter stopped, once it has released what it owned. // // Provide data about the last completed round. If there is no // known last completed round, the genesis state (round number 0, no votes, genesis base), @@ -647,7 +646,7 @@ func (v *Voter[Hash, Number, Signature, ID]) pruneBackgroundRounds(waker *waker) // Collect finalize notifications under the lock, then invoke // env.FinalizeBlock outside it. Holding inner.Mutex across user-supplied // callbacks is a deadlock hazard: a slow environment can block readers - // of the voter state and stall Stop(). + // of the voter state and stall the voter's teardown. v.inner.Lock() pastRounds: @@ -677,10 +676,11 @@ finalizedNotifications: for { select { case notif, ok := <-v.finalizedNotifications.channel(): - // This channel is the voter's own, and Stop closes it only after Start - // has returned, so a live poll loop cannot legitimately see it closed: - // that means the voter was torn down and is being polled anyway. The - // stream carrying finalization is gone and cannot be reopened. + // This channel is the voter's own, and the teardown closes it only once + // the run loop has returned, so a live poll loop cannot legitimately see + // it closed: that means the voter was torn down and is being polled + // anyway. The stream carrying finalization is gone and cannot be + // reopened. // (Unchecked, the receive would also stay ready forever, yielding // zero-value notifications that change nothing, spinning here while // holding v.inner — which blocks VoterState too.) @@ -727,7 +727,7 @@ loop: case item, ok := <-v.globalIn.channel(): // The forwarder closes this channel when globalIn ends, which is how a // caller shuts the voter down. Unwinding through the poll path is what - // ends the Start loop. (Unchecked, the receive would also stay ready + // ends the run loop. (Unchecked, the receive would also stay ready // forever, handing out zero-value items that match no case below and // spinning the loop.) if !ok { From 6042451cb5ffcdad7bd97b1f9af6cb198a818f28 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 18:43:36 -0400 Subject: [PATCH 5/7] docs(pkg/finality-grandpa): state the voter's contract rather than its rationale --- pkg/finality-grandpa/bridge_state.go | 5 +- pkg/finality-grandpa/environment_test.go | 11 ++-- pkg/finality-grandpa/voter.go | 70 ++++++++++-------------- pkg/finality-grandpa/voting_round.go | 16 ++---- 4 files changed, 41 insertions(+), 61 deletions(-) diff --git a/pkg/finality-grandpa/bridge_state.go b/pkg/finality-grandpa/bridge_state.go index 81fa892b40..4fba41a423 100644 --- a/pkg/finality-grandpa/bridge_state.go +++ b/pkg/finality-grandpa/bridge_state.go @@ -17,9 +17,8 @@ func newWaker() *waker { } func (w *waker) wake() { - // Read the channel under the lock and hand the value to the goroutine. Reading - // w.wakeCh inside the goroutine would be unsynchronised: it runs after wake has - // returned and released the lock, so it can race register's write. + // Read under the lock and hand the value to the goroutine, which outlives the + // lock and would otherwise race register's write. w.RLock() ch := w.wakeCh w.RUnlock() diff --git a/pkg/finality-grandpa/environment_test.go b/pkg/finality-grandpa/environment_test.go index 3983c08f8e..5c5e95ca2a 100644 --- a/pkg/finality-grandpa/environment_test.go +++ b/pkg/finality-grandpa/environment_test.go @@ -259,9 +259,9 @@ func (bm *BroadcastNetwork[M, N]) AddNode(f func(N) M, out chan N) (in chan M) { func (bm *BroadcastNetwork[M, N]) route() { defer bm.routeWG.Done() for msg := range bm.receiver { - // Deliver under the lock. RemoveNode closes a node's channel, and closing a - // channel a producer is about to send on panics, so the two have to be - // serialised. Senders are buffered, so this does not block in practice. + // Under the lock: RemoveNode closes a node's channel, and closing one a + // producer is about to send on panics. Senders are buffered, so holding it + // across the delivery does not block. bm.mu.Lock() bm.history = append(bm.history, msg) for _, sender := range bm.senders { @@ -271,9 +271,8 @@ func (bm *BroadcastNetwork[M, N]) route() { } } -// RemoveNode deregisters a node's inbound channel and closes it. Closing that -// channel is how the voter reading it is asked to shut down; it happens under -// bm.mu so it cannot race a delivery in route. +// RemoveNode deregisters a node's inbound channel and closes it, shutting down +// the voter reading it. Held under bm.mu so it cannot race a delivery in route. func (bm *BroadcastNetwork[M, N]) RemoveNode(in chan M) { bm.mu.Lock() defer bm.mu.Unlock() diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index 87ba301edc..e7ca454e33 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -32,9 +32,8 @@ func newWakerChan[Item any](in chan Item) *wakerChan[Item] { func (wc *wakerChan[Item]) start() { defer func() { close(wc.out) - // Wake the consumer so it observes the close on its next poll. Closing the - // input is how a caller shuts the voter down, so without this the poll loop - // would sit on the waker until some unrelated source happened to fire. + // Wake the consumer so it polls once more and sees the close, which is how + // a shutdown reaches it. if w := wc.waker.Load(); w != nil { w.wake() } @@ -544,26 +543,22 @@ type Voter[Hash constraints.Ordered, Number constraints.Unsigned, Signature comp // assumptions from round-to-round. lastFinalizedInRounds HashNumber[Hash, Number] - // done carries the voter's terminal error, published once the voter has torn - // itself down, and is closed straight after. Buffered so the voter never waits - // on an owner that has stopped listening. + // done carries the terminal error, published after teardown and closed behind + // it. Buffered, so the voter never waits on an owner that has stopped + // listening. done chan error } -// errVoterShutdown travels back through the poll path when the globalIn channel -// given to NewVoter is closed, which is how a caller asks the voter to stop. It -// is a shutdown signal rather than a failure, so the run loop reports it as a -// nil error; it never reaches the caller. +// errVoterShutdown marks an orderly shutdown as it unwinds through the poll +// path. The run loop translates it to a nil error, so it never reaches a caller. var errVoterShutdown = errors.New("voter shutdown: global incoming stream closed") -// NewVoter creates a new `Voter` tracker with given round number and base block -// and starts it running. There is no separate start step, so a voter is never -// observable in a constructed-but-idle state. +// NewVoter creates a new `Voter` tracker with given round number and base block, +// and starts it running. // -// The voter runs until globalIn is closed, which is how a caller asks it to shut -// down. Closing it belongs to the caller, who must therefore be the only writer -// to it by that point, or must serialise its writers against the close. Done -// then yields why the voter stopped, once it has released what it owned. +// Close globalIn to shut the voter down. It belongs to the caller, who must be +// its only writer by that point or must serialise its writers against the close. +// Done then yields why the voter stopped. // // Provide data about the last completed round. If there is no // known last completed round, the genesis state (round number 0, no votes, genesis base), @@ -633,8 +628,7 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c } go func() { err := v.run() - // Tear down before publishing, so that receiving from done means the voter - // has not merely stopped but finished releasing what it owned. + // Before publishing, so a receive on done means teardown is complete. v.teardown() v.done <- err close(v.done) @@ -676,14 +670,10 @@ finalizedNotifications: for { select { case notif, ok := <-v.finalizedNotifications.channel(): - // This channel is the voter's own, and the teardown closes it only once - // the run loop has returned, so a live poll loop cannot legitimately see - // it closed: that means the voter was torn down and is being polled - // anyway. The stream carrying finalization is gone and cannot be - // reopened. - // (Unchecked, the receive would also stay ready forever, yielding - // zero-value notifications that change nothing, spinning here while - // holding v.inner — which blocks VoterState too.) + // The voter's own channel, closed only by teardown, so a live poll loop + // cannot legitimately see it closed: finalization is gone and cannot be + // reopened. A closed channel is permanently ready, so this check is what + // ends the loop rather than the default arm. if !ok { v.inner.Unlock() return fmt.Errorf("finalization notification stream closed") @@ -725,11 +715,9 @@ loop: for { select { case item, ok := <-v.globalIn.channel(): - // The forwarder closes this channel when globalIn ends, which is how a - // caller shuts the voter down. Unwinding through the poll path is what - // ends the run loop. (Unchecked, the receive would also stay ready - // forever, handing out zero-value items that match no case below and - // spinning the loop.) + // Closed once globalIn is: the caller's shutdown signal. A closed channel + // is permanently ready, so this check is what ends the loop rather than + // the default arm. if !ok { return errVoterShutdown } @@ -964,11 +952,9 @@ func (v *Voter[Hash, Number, Signature, ID]) run() error { } } -// Done yields the voter's terminal error once it has stopped: nil if it was shut -// down by closing the globalIn channel given to NewVoter, otherwise the error it -// failed with. Every error is terminal — the voter does not recover from one and -// carry on — so receiving here means the voter has finished and has already -// released what it owned. +// Done yields why the voter stopped: nil if globalIn was closed, otherwise the +// error it failed with. Every error is terminal. Receiving here means the voter +// has finished and released what it owned, so there is nothing further to call. // // Select on it to supervise the voter alongside other work: // @@ -978,16 +964,16 @@ func (v *Voter[Hash, Number, Signature, ID]) run() error { // case cmd := <-commands: // } // -// The error is delivered to one receiver, and the channel is closed immediately -// after, so later receives yield nil. A voter has a single owner; if more than -// one party needs the outcome, that owner must fan it out. +// The error goes to one receiver and the channel closes behind it, so later +// receives yield nil. A voter has a single owner; if others need the outcome, +// that owner fans it out. func (v *Voter[Hash, Number, Signature, ID]) Done() <-chan error { return v.done } // teardown releases what the voter owns: its finalization channel and every -// round timer. It runs on the voter's own goroutine once the run loop is done, -// so callers have no teardown obligation. +// round timer. It runs on the voter's own goroutine, so callers have no teardown +// obligation. func (v *Voter[Hash, Number, Signature, ID]) teardown() { v.globalOut.Close() close(v.finalizedNotifications.in) diff --git a/pkg/finality-grandpa/voting_round.go b/pkg/finality-grandpa/voting_round.go index 426d79069f..9c251fc826 100644 --- a/pkg/finality-grandpa/voting_round.go +++ b/pkg/finality-grandpa/voting_round.go @@ -461,11 +461,9 @@ while: for { select { case incoming, ok := <-vr.incoming.channel(): - // roundData.Incoming belongs to the environment. A round that is still - // voting cannot recover from losing it, and the zero value would carry a - // nil Message interface, which handleVote dereferences. Unchecked, the - // receive also stays permanently ready, so the default arm never runs - // and the 1ms timerChan escape below is never armed. + // roundData.Incoming belongs to the environment. A round still voting + // cannot recover from losing it, and the zero value carries a nil Message + // that handleVote would dereference. if !ok { return fmt.Errorf("round %d: incoming message stream closed", vr.roundNumber()) } @@ -602,11 +600,9 @@ func (vr *votingRound[Hash, Number, Signature, ID, E]) prevote(w *waker, lastRou res, ok := <-wakerChan.channel() switch { case !ok: - // The environment owns bestChain and always sends one value before - // closing, so an empty closed channel means the stream went away. Kept - // distinct from the default arm below: a real {nil, nil} value means - // "no best chain yet" and legitimately parks the round in prevoting, - // whereas a closed channel would park it there forever. + // An empty closed channel means the stream went away. Distinct from the + // default arm below, where a real {nil, nil} value means "no best chain + // yet" and legitimately parks the round in prevoting. return fmt.Errorf("round %d: best chain stream closed", vr.roundNumber()) case res.Error != nil: return res.Error From b0d0ec0a7aa7dfb11f636d65205ec9b37a0f20cb Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 5 Aug 2026 19:07:59 -0400 Subject: [PATCH 6/7] fix(pkg/finality-grandpa): release round inputs and timers as rounds advance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A running voter accumulated four goroutines per round, none of them released until the process ended. Over 8 rounds the forwarder count grew by 32. Timers accounted for three of the four. A timer wrapped an unbuffered channel in a wakerChan purely to deliver a wake, but nothing anywhere reads a timer's out: consumers use SetWaker and Elapsed, and Elapsed reads an atomic. So every timer that fired left its forwarder parked on a send with no receiver that would ever exist, and Close could not reach it — closing the input does not release a goroutine blocked on a send. The wakerChan is gone; the timer holds its waker directly and wakes it from the goroutine that already watches the deadline. Close now ends that goroutine when a round finishes before its timer fires. Setting expired before waking also fixes a race: waking first let a poller observe the timer as pending and go back to sleep. The fourth is roundData.Incoming, which belongs to the environment. Its forwarder ends only when that channel closes, so a round input left open outlives its round. Environment.RoundData now documents that implementations close it once the round concludes and release any still open at shutdown, and the test environment does so from Concluded. RoundData is called from two sites and may be reached more than once for a round number, so each call's channel is tracked rather than one per round. --- pkg/finality-grandpa/environment_test.go | 39 +++++++++++++++++++- pkg/finality-grandpa/timer.go | 33 ++++++++++------- pkg/finality-grandpa/voter.go | 6 +++ pkg/finality-grandpa/voter_lifecycle_test.go | 39 ++++++++++++++++++++ 4 files changed, 102 insertions(+), 15 deletions(-) diff --git a/pkg/finality-grandpa/environment_test.go b/pkg/finality-grandpa/environment_test.go index 5c5e95ca2a..c7e7236f18 100644 --- a/pkg/finality-grandpa/environment_test.go +++ b/pkg/finality-grandpa/environment_test.go @@ -26,7 +26,11 @@ type environment struct { network *Network listeners []chan listenerItem lastCompleteAndConcluded [2]uint64 - mtx sync.Mutex + // roundIn holds the inbound channels handed to the voter, per round, so they + // can be closed once the round concludes. RoundData is called more than once + // for a round number, hence a slice. + roundIn map[uint64][]chan SignedMessageError[string, uint32, Signature, ID] + mtx sync.Mutex concludedCalled chan struct{} } @@ -36,6 +40,7 @@ func newEnvironment(network *Network, localID ID) environment { chain: newDummyChain(), localID: localID, network: network, + roundIn: make(map[uint64][]chan SignedMessageError[string, uint32, Signature, ID]), concludedCalled: make(chan struct{}), } } @@ -84,6 +89,12 @@ func (e *environment) RoundData( outgoing := make(Output[string, uint32]) incoming := e.network.MakeRoundComms(round, e.localID, outgoing) + // Remember it so Concluded can close it: the voter reads this channel through + // a forwarding goroutine that ends only when the channel does. + e.mtx.Lock() + e.roundIn[round] = append(e.roundIn[round], incoming) + e.mtx.Unlock() + var outgoingFunc = func(m Message[string, uint32]) error { outgoing <- m return nil @@ -123,8 +134,16 @@ func (e *environment) Concluded( _ HistoricalVotes[string, uint32, Signature, ID], ) error { e.mtx.Lock() - defer e.mtx.Unlock() e.lastCompleteAndConcluded[1] = round + incoming := e.roundIn[round] + delete(e.roundIn, round) + e.mtx.Unlock() + + // The round is over, so release the inbound channels handed out for it. + for _, in := range incoming { + e.network.StopRoundComms(round, in) + } + go func() { e.concludedCalled <- struct{}{} }() @@ -418,6 +437,22 @@ func (n *Network) MakeGlobalComms( }, out) } +// StopRoundComms closes one inbound channel handed out by MakeRoundComms. Only +// that node's channel: the round network is shared, and other voters may still +// be in this round. +func (n *Network) StopRoundComms( + roundNumber uint64, + in chan SignedMessageError[string, uint32, Signature, ID], +) { + n.mtx.Lock() + round, ok := n.rounds[roundNumber] + n.mtx.Unlock() + + if ok { + round.RemoveNode(in) + } +} + // StopGlobalComms closes the inbound channel handed to a voter by // MakeGlobalComms, which is how that voter is shut down. func (n *Network) StopGlobalComms(in chan GlobalInItem[string, uint32, Signature, ID]) { diff --git a/pkg/finality-grandpa/timer.go b/pkg/finality-grandpa/timer.go index 97c2755f7a..d798ec6798 100644 --- a/pkg/finality-grandpa/timer.go +++ b/pkg/finality-grandpa/timer.go @@ -9,39 +9,46 @@ import ( "time" ) +// timer reports whether a deadline has passed and wakes whoever is polling it +// when that changes. Rounds create several and discard them as they advance, so +// Close releases one whose round finished before it fired. type timer struct { - wakerChan *wakerChan[error] + waker atomic.Pointer[waker] + stop chan struct{} closeOnce sync.Once expired atomic.Bool } func newTimer(in <-chan time.Time) *timer { - inErr := make(chan error) - wc := newWakerChan(inErr) - t := timer{wakerChan: wc} + t := timer{stop: make(chan struct{})} go t.poll(in) return &t } func (t *timer) poll(in <-chan time.Time) { - <-in - t.closeOnce.Do(func() { - t.wakerChan.in <- nil - close(t.wakerChan.in) - }) + select { + case <-in: + case <-t.stop: + return + } + // Ordered: waking before expired is set would send the poller back to sleep + // having seen the timer as still pending. t.expired.Store(true) + if w := t.waker.Load(); w != nil { + w.wake() + } } func (t *timer) SetWaker(waker *waker) { - t.wakerChan.setWaker(waker) + t.waker.Store(waker) } func (t *timer) Elapsed() (bool, error) { return t.expired.Load(), nil } +// Close releases a timer that has not fired. Idempotent, and a no-op once the +// timer has elapsed. func (t *timer) Close() { - t.closeOnce.Do(func() { - close(t.wakerChan.in) - }) + t.closeOnce.Do(func() { close(t.stop) }) } diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index e7ca454e33..facdddc2d4 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -127,6 +127,12 @@ type Environment[Hash comparable, Number constraints.Unsigned, Signature compara // // Furthermore, this means that actual logic of creating and verifying // signatures is flexible and can be maintained outside this crate. + // + // The Incoming channel belongs to the implementation, which must close it once + // the round has concluded, and release any still open when it shuts down. The + // voter reads it through a forwarding goroutine that ends only when the + // channel does, so a round input left open outlives its round. RoundData may + // be called more than once for a round number; each call owns its channel. RoundData( round uint64, ) RoundData[Hash, Number, Signature, ID, Message[Hash, Number]] diff --git a/pkg/finality-grandpa/voter_lifecycle_test.go b/pkg/finality-grandpa/voter_lifecycle_test.go index 0a4b075420..f1aa4e1fb4 100644 --- a/pkg/finality-grandpa/voter_lifecycle_test.go +++ b/pkg/finality-grandpa/voter_lifecycle_test.go @@ -180,3 +180,42 @@ func TestVoter_RebuildAcrossRotations(t *testing.T) { close(globalIn) require.NoError(t, <-voter.Done()) } + +// A voter that keeps running must not accumulate goroutines. Every round wraps +// its inbound stream and its timers, and both have to be released as rounds +// advance rather than only at shutdown. +func TestVoter_RoundsDoNotAccumulateForwarders(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + globalIn := make(chan lifecycleItem, 10) + v := newLifecycleVoter(t, network, globalIn) + defer func() { + close(globalIn) + <-v.Done() + }() + + live := func() int { + return forwardersInState("chan receive") + forwardersInState("chan send") + } + + // Let the voter settle into a steady state before taking the baseline, so + // start-up rounds are not counted as growth. + time.Sleep(2 * time.Second) + v.inner.Lock() + firstRound := v.inner.bestRound.roundNumber() + v.inner.Unlock() + base := live() + + time.Sleep(8 * time.Second) + v.inner.Lock() + lastRound := v.inner.bestRound.roundNumber() + v.inner.Unlock() + grew := live() - base + + rounds := lastRound - firstRound + require.Greater(t, rounds, uint64(2), "test needs several rounds to have elapsed") + t.Logf("%d rounds elapsed, forwarders grew by %+d", rounds, grew) + assert.LessOrEqual(t, grew, 2, + "forwarders grow with rounds: %d over %d rounds", grew, rounds) +} From 8a36b17d5f2a1aa806fb4596bca1b79bce945a09 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Thu, 6 Aug 2026 11:20:33 -0400 Subject: [PATCH 7/7] fix(pkg/finality-grandpa): reject a nil globalIn in NewVoter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closing globalIn is the shutdown signal, and a nil channel reads as already closed, so a voter built on one stopped immediately and reported a clean shutdown through Done — indistinguishable from a successful one. --- pkg/finality-grandpa/voter.go | 7 +++++++ pkg/finality-grandpa/voter_lifecycle_test.go | 12 ++++++++++++ 2 files changed, 19 insertions(+) diff --git a/pkg/finality-grandpa/voter.go b/pkg/finality-grandpa/voter.go index facdddc2d4..f0dfb8c236 100644 --- a/pkg/finality-grandpa/voter.go +++ b/pkg/finality-grandpa/voter.go @@ -585,6 +585,13 @@ func NewVoter[Hash constraints.Ordered, Number constraints.Unsigned, Signature c lastRoundBase HashNumber[Hash, Number], lastFinalized HashNumber[Hash, Number], ) *Voter[Hash, Number, Signature, ID] { + // A nil globalIn would otherwise yield a voter that stops immediately and + // reports a clean shutdown, since closing that channel is the shutdown signal + // and a nil one reads as already closed. + if globalIn == nil { + panic("grandpa: NewVoter requires a non-nil globalIn; closing it is how the voter is shut down") + } + finalizedSender := make(chan finalizedNotification[Hash, Number, Signature, ID], 1) finalizedNotifications := finalizedSender lastFinalizedNumber := lastFinalized.Number diff --git a/pkg/finality-grandpa/voter_lifecycle_test.go b/pkg/finality-grandpa/voter_lifecycle_test.go index f1aa4e1fb4..81203e92e3 100644 --- a/pkg/finality-grandpa/voter_lifecycle_test.go +++ b/pkg/finality-grandpa/voter_lifecycle_test.go @@ -219,3 +219,15 @@ func TestVoter_RoundsDoNotAccumulateForwarders(t *testing.T) { assert.LessOrEqual(t, grew, 2, "forwarders grow with rounds: %d over %d rounds", grew, rounds) } + +// globalIn is mandatory: closing it is the shutdown signal, and a nil channel +// reads as already closed, so a voter built on one would stop immediately and +// report a clean shutdown. +func TestVoter_NilGlobalInPanics(t *testing.T) { + network := NewNetwork() + defer network.Stop() + + assert.PanicsWithValue(t, + "grandpa: NewVoter requires a non-nil globalIn; closing it is how the voter is shut down", + func() { newLifecycleVoter(t, network, nil) }) +}