From d82201ec46f154bf7f0f2f55cbc086174268e0c7 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:32:03 +0200 Subject: [PATCH 1/4] Preserve the already fetched value for comparing between the rounds --- core/qbft/qbft.go | 24 +++++++++++++++++++++++- core/qbft/qbft_internal_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/core/qbft/qbft.go b/core/qbft/qbft.go index 34ae1f3cf..c4e15d7b9 100644 --- a/core/qbft/qbft.go +++ b/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 } } } diff --git a/core/qbft/qbft_internal_test.go b/core/qbft/qbft_internal_test.go index 33cf92af7..3184f0804 100644 --- a/core/qbft/qbft_internal_test.go +++ b/core/qbft/qbft_internal_test.go @@ -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{ From 2e9261a320ee5ed935149bffcb299ed7356f05ea Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Thu, 6 Aug 2026 18:06:10 +0200 Subject: [PATCH 2/4] Fix clock in tests --- core/qbft/clock_internal_test.go | 16 ++++++++++++++ core/qbft/qbft_internal_test.go | 37 ++++++++++++++++++++++---------- 2 files changed, 42 insertions(+), 11 deletions(-) diff --git a/core/qbft/clock_internal_test.go b/core/qbft/clock_internal_test.go index 281d8f53f..b0a860112 100644 --- a/core/qbft/clock_internal_test.go +++ b/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() diff --git a/core/qbft/qbft_internal_test.go b/core/qbft/qbft_internal_test.go index 3184f0804..101a742c6 100644 --- a/core/qbft/qbft_internal_test.go +++ b/core/qbft/qbft_internal_test.go @@ -1864,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) } @@ -1941,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) } } From 03e77e5b56f24df8f4c5a35ee34c1b062a8f62b2 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Fri, 7 Aug 2026 11:25:57 +0200 Subject: [PATCH 3/4] Fix after rebase --- core/qbft/qbft_internal_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/qbft/qbft_internal_test.go b/core/qbft/qbft_internal_test.go index 101a742c6..e045eaca1 100644 --- a/core/qbft/qbft_internal_test.go +++ b/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 { From ef34194864af3186fad5f78bc4e24ae26fb63639 Mon Sep 17 00:00:00 2001 From: kalo <24719519+KaloyanTanev@users.noreply.github.com> Date: Fri, 7 Aug 2026 11:43:33 +0200 Subject: [PATCH 4/4] app/eth2wrap: fix flaky TestMulti/cl2_before_cl1 --- app/eth2wrap/eth2wrap_test.go | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/app/eth2wrap/eth2wrap_test.go b/app/eth2wrap/eth2wrap_test.go index faa03b5cb..900b13af6 100644 --- a/app/eth2wrap/eth2wrap_test.go +++ b/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 <- ð2v1.PeerCount{Connected: 99} - - time.Sleep(time.Millisecond) - - cl1 <- ð2v1.PeerCount{Connected: 98} // This might flap? }, expRes: ð2api.Response[*eth2v1.PeerCount]{Data: ð2v1.PeerCount{Connected: 99}}, },