core/qbft: preserve local value on comparison failure

app/eth2wrap/eth2wrap_test.go

@@ -87,12 +87,8 @@ func TestMulti(t *testing.T) {
 		},
 		{
 			name: "cl2 before cl1",
-			handle: func(cl1, cl2 chan *eth2v1.PeerCount, cancel context.CancelFunc) {
+			handle: func(_, cl2 chan *eth2v1.PeerCount, _ context.CancelFunc) {
 				cl2 <- &eth2v1.PeerCount{Connected: 99}
-
-				time.Sleep(time.Millisecond)
-
-				cl1 <- &eth2v1.PeerCount{Connected: 98} // This might flap?
 			},
 			expRes: &eth2api.Response[*eth2v1.PeerCount]{Data: &eth2v1.PeerCount{Connected: 99}},
 		},

core/qbft/clock_internal_test.go

@@ -34,6 +34,22 @@ func (c *fakeClock) NewTimer(d time.Duration) (<-chan time.Time, func()) {
 	}
 }
 
+// NumActive returns the number of pending timers that have neither fired nor been stopped.
+func (c *fakeClock) NumActive() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	var count int
+
+	for _, ch := range c.chans {
+		if ch != nil {
+			count++
+		}
+	}
+
+	return count
+}
+
 // NowStr returns the current time as a debug string.
 func (c *fakeClock) NowStr() string {
 	c.mu.Lock()

core/qbft/qbft.go

@@ -490,9 +490,31 @@ func compare[I any, V comparable, C any](ctx context.Context, d Definition[I, V,
 	// If comparison or any other unexpected error occurs, the error is returned on compareErr channel.
 	go d.Compare(ctxCompare, msg, inputValueSourceCh, inputValueSource, compareErr, compareValue)
 
+	return awaitCompare(ctx, compareErr, compareValue, timerChan, inputValueSource)
+}
+
+// awaitCompare waits for the comparator's verdict and returns the latest local value.
+func awaitCompare[C any](ctx context.Context, compareErr <-chan error, compareValue <-chan C, timerChan <-chan time.Time, inputValueSource C) (C, error) {
+	// drainValue returns the value read by the comparator if it sent one, otherwise
+	// the current value. Both channels are buffered, so when the comparator reads
+	// the local value and then errors (e.g. a comparison mismatch), both sends may
+	// complete before this loop polls the select, which then picks a ready case at
+	// random. Without draining, the local value would be lost and subsequent rounds
+	// would block waiting for the already-consumed input value source channel.
+	drainValue := func() C {
+		select {
+		case v := <-compareValue:
+			return v
+		default:
+			return inputValueSource
+		}
+	}
+
 	for {
 		select {
 		case err := <-compareErr:
+			inputValueSource = drainValue()
+
 			if err != nil {
 				log.Warn(ctx, errCompare.Error(), err)
 				return inputValueSource, errCompare
@@ -502,7 +524,7 @@ func compare[I any, V comparable, C any](ctx context.Context, d Definition[I, V,
 		case inputValueSource = <-compareValue:
 		case <-timerChan:
 			log.Warn(ctx, "", errors.New("timeout waiting for local data, used for comparing with leader's proposed data"))
-			return inputValueSource, errTimeout
+			return drainValue(), errTimeout
 		}
 	}
 }

core/qbft/qbft_internal_test.go

@@ -1563,7 +1563,7 @@ func TestEquivocatingLeaderDoublePrepare(t *testing.T) {
 	def.Compare = func(_ context.Context, _ Msg[int64, int64, int64], _ <-chan int64, _ int64, returnErr chan error, _ chan int64) {
 		returnErr <- nil
 	}
-	def.Decide = func(context.Context, int64, int64, []Msg[int64, int64, int64]) {}
+	def.Decide = func(context.Context, int64, int64, int64, []Msg[int64, int64, int64]) {}
 
 	trans := Transport[int64, int64, int64]{
 		Broadcast: func(_ context.Context, typ MsgType, _ int64, _ int64, _ int64, value int64, _ int64, _ int64, _ []Msg[int64, int64, int64]) error {
@@ -1638,7 +1638,7 @@ func TestZeroValuePrePrepareRejected(t *testing.T) {
 	def.Compare = func(_ context.Context, _ Msg[int64, int64, int64], _ <-chan int64, _ int64, returnErr chan error, _ chan int64) {
 		returnErr <- nil
 	}
-	def.Decide = func(context.Context, int64, int64, []Msg[int64, int64, int64]) {}
+	def.Decide = func(context.Context, int64, int64, int64, []Msg[int64, int64, int64]) {}
 
 	trans := Transport[int64, int64, int64]{
 		Broadcast: func(_ context.Context, typ MsgType, _ int64, _ int64, _ int64, value int64, _ int64, _ int64, _ []Msg[int64, int64, int64]) error {
@@ -1696,6 +1696,32 @@ type testChainSplit struct {
 
 var errChainSplitHalt = errors.New("chain split halt")
 
+// TestCompareRetainsValueOnError verifies that the compare flow does not lose the
+// local value read by the comparator when the comparison also fails. The comparator
+// sends the value and the error back-to-back on buffered channels, so both select
+// cases can be ready simultaneously and the select picks one at random; the read
+// value must be returned regardless of which case wins, otherwise subsequent rounds
+// block forever on the already-consumed input value source channel.
+func TestCompareRetainsValueOnError(t *testing.T) {
+	const localValue = 42
+
+	// Repeat since the select picks randomly among the two ready cases.
+	for range 1000 {
+		// Model the racy state directly: the comparator already completed both
+		// sends before the await loop polled the select.
+		compareErr := make(chan error, 1)
+
+		compareValue := make(chan int64, 1)
+		compareValue <- localValue
+
+		compareErr <- errors.New("mismatch")
+
+		vs, err := awaitCompare(context.Background(), compareErr, compareValue, nil, 0)
+		require.ErrorIs(t, err, errCompare)
+		require.EqualValues(t, localValue, vs, "local value must not be lost when comparison fails")
+	}
+}
+
 func TestChainSplit(t *testing.T) {
 	t.Run("same value", func(t *testing.T) {
 		testQBFTChainSplit(t, testChainSplit{
@@ -1838,20 +1864,16 @@ func testQBFTChainSplit(t *testing.T, test testChainSplit) {
 			Receive: receive,
 		}
 
-		go func(i int64) {
-			// Only enqueue input values for instances that:
-			// - have a value delay
-			// - or expect multiple rounds
-			// - or otherwise only the leader of round 1.
-			vChan := make(chan int64, 1)
-			vsChan := make(chan int64, 1)
+		// Enqueue input values synchronously (channels are buffered), so nodes
+		// never wait on a feeder goroutine while virtual time advances.
+		vChan := make(chan int64, 1)
+		vsChan := make(chan int64, 1)
 
-			go func() {
-				vChan <- test.ValueSource[i]
+		vChan <- test.ValueSource[i]
 
-				vsChan <- test.ValueSource[i]
-			}()
+		vsChan <- test.ValueSource[i]
 
+		go func(i int64) {
 			runChan <- Run(ctx, defs, transport, instance, i, vChan, vsChan)
 		}(i)
 	}
@@ -1915,7 +1937,26 @@ func testQBFTChainSplit(t *testing.T, test testChainSplit) {
 				return
 			}
 		default:
+			// Only advance virtual time when the system is quiescent: every node
+			// still deciding is parked on its round timer, and no messages remain
+			// queued in the receive buffers. Otherwise real-world goroutine
+			// scheduling lag skews nodes multiple virtual seconds apart, pushing
+			// consensus to a later round than the scenario expects.
+			quiescent := clock.NumActive() >= n-done-count
+
+			for _, out := range receiveChannelsPerNode {
+				if len(out) > 0 {
+					quiescent = false
+					break
+				}
+			}
+
 			time.Sleep(time.Microsecond)
+
+			if !quiescent {
+				continue
+			}
+
 			clock.Advance(time.Millisecond * 1)
 		}
 	}