fix(pkg/finality-grandpa): release round inputs and timers as rounds advance

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.

Author: timwu20

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]) {

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) })
 }

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]]

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)
+}