From 8a32e72fe5ff6be91d074ffe2b853e204ad74171 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Mon, 16 Mar 2026 16:27:51 +0200 Subject: [PATCH 1/2] improve delay messages on multikeys --- consensus/spos/bls/v2/export_test.go | 5 + consensus/spos/bls/v2/subroundEndRound.go | 12 +- consensus/spos/bls/v2/subroundSignature.go | 62 ++-- .../subroundSignatureCompetingBlock_test.go | 286 +++++++++++++++++- 4 files changed, 335 insertions(+), 30 deletions(-) diff --git a/consensus/spos/bls/v2/export_test.go b/consensus/spos/bls/v2/export_test.go index 106775e7bcf..58988e0d82f 100644 --- a/consensus/spos/bls/v2/export_test.go +++ b/consensus/spos/bls/v2/export_test.go @@ -351,6 +351,11 @@ func (sr *subroundSignature) WaitIfCompetingBlock(ctx context.Context, pkBytes [ return sr.waitIfCompetingBlock(ctx, pkBytes, nonce, currentHash) } +// WaitIfCompetingBlockForNode calls the unexported waitIfCompetingBlockForNode function +func (sr *subroundSignature) WaitIfCompetingBlockForNode(ctx context.Context, nonce uint64, currentHash []byte) bool { + return sr.waitIfCompetingBlockForNode(ctx, nonce, currentHash) +} + // ShouldSendProof calls the unexported shouldSendProof function func (sr *subroundEndRound) ShouldSendProof() bool { return sr.shouldSendProof() diff --git a/consensus/spos/bls/v2/subroundEndRound.go b/consensus/spos/bls/v2/subroundEndRound.go index 8772b904791..261b9123668 100644 --- a/consensus/spos/bls/v2/subroundEndRound.go +++ b/consensus/spos/bls/v2/subroundEndRound.go @@ -272,10 +272,8 @@ func (sr *subroundEndRound) doEndRoundJobByNode() bool { } proofSent, err := sr.sendProof() - shouldWaitForMoreSignatures := errors.Is(err, spos.ErrInvalidNumSigShares) - // if not enough valid signatures were detected, wait a bit more - // either more signatures will be received, either proof from another participant - if shouldWaitForMoreSignatures { + // Not enough valid signatures: wait for more or for a proof from another participant + if errors.Is(err, spos.ErrInvalidNumSigShares) { continue } @@ -389,6 +387,12 @@ func (sr *subroundEndRound) sendProof() (bool, error) { return false, err } + // Re-check grace period after aggregation which may have been slow under CPU contention + if !sr.shouldSendProof() { + log.Debug("sendProof: grace period expired during aggregation, not broadcasting") + return false, nil + } + // broadcast header proof err = sr.createAndBroadcastProof(sig, bitmap, currentSender) if err != nil && !errors.Is(err, ErrProofAlreadyPropagated) { diff --git a/consensus/spos/bls/v2/subroundSignature.go b/consensus/spos/bls/v2/subroundSignature.go index e7597812d53..4f7b864d687 100644 --- a/consensus/spos/bls/v2/subroundSignature.go +++ b/consensus/spos/bls/v2/subroundSignature.go @@ -108,6 +108,14 @@ func (sr *subroundSignature) doSignatureJob(ctx context.Context) bool { return false } + // Wait once for the entire node if competing block detected + nonce := sr.GetHeader().GetNonce() + currentHash := sr.GetData() + shouldAbort := sr.waitIfCompetingBlockForNode(ctx, nonce, currentHash) + if shouldAbort { + return false + } + isSelfSingleKeyInConsensusGroup := sr.IsNodeInConsensusGroup(sr.SelfPubKey()) && commonConsensus.ShouldConsiderSelfKeyInConsensus(sr.NodeRedundancyHandler()) if isSelfSingleKeyInConsensusGroup { if !sr.doSignatureJobForSingleKey(ctx) { @@ -251,16 +259,11 @@ func (sr *subroundSignature) doSignatureJobForManagedKeys(ctx context.Context) b return sentSigForAllKeys.IsSet() } -func (sr *subroundSignature) sendSignatureForManagedKey(ctx context.Context, idx int, pk string) bool { +func (sr *subroundSignature) sendSignatureForManagedKey(_ context.Context, idx int, pk string) bool { pkBytes := []byte(pk) nonce := sr.GetHeader().GetNonce() currentHash := sr.GetData() - shouldAbort := sr.waitIfCompetingBlock(ctx, pkBytes, nonce, currentHash) - if shouldAbort { - return false - } - signatureShare, err := sr.SigningHandler().CreateSignatureShareForPublicKey( currentHash, uint16(idx), @@ -301,16 +304,11 @@ func (sr *subroundSignature) checkGoRoutinesThrottler(ctx context.Context) error return nil } -func (sr *subroundSignature) doSignatureJobForSingleKey(ctx context.Context) bool { +func (sr *subroundSignature) doSignatureJobForSingleKey(_ context.Context) bool { pkBytes := []byte(sr.SelfPubKey()) nonce := sr.GetHeader().GetNonce() currentHash := sr.GetData() - shouldAbort := sr.waitIfCompetingBlock(ctx, pkBytes, nonce, currentHash) - if shouldAbort { - return false - } - selfIndex, err := sr.SelfConsensusGroupIndex() if err != nil { log.Debug("doSignatureJobForSingleKey.SelfConsensusGroupIndex: not in consensus group") @@ -341,9 +339,33 @@ func (sr *subroundSignature) doSignatureJobForSingleKey(ctx context.Context) boo return sr.completeSignatureSubRound(sr.SelfPubKey()) } -// waitIfCompetingBlock checks if this node already signed a different block for the same nonce. -// If so, it waits for a fraction of the round time to give the previous block's proof a chance to arrive. -// Returns true if the signing should be aborted (proof for previous block arrived or context cancelled). +// waitIfCompetingBlockForNode checks if any key managed by this node previously signed a different +// hash for the given nonce. If found, waits once for the entire node instead of per-key. +func (sr *subroundSignature) waitIfCompetingBlockForNode(ctx context.Context, nonce uint64, currentHash []byte) bool { + // Check self key first + selfPk := []byte(sr.SelfPubKey()) + previousHash, exists := sr.sentSignatureTracker.GetSignedHash(selfPk, nonce) + if exists && !bytes.Equal(previousHash, currentHash) { + return sr.waitIfCompetingBlock(ctx, selfPk, nonce, currentHash) + } + + // Check managed keys + for _, pk := range sr.ConsensusGroup() { + pkBytes := []byte(pk) + if !sr.IsKeyManagedBySelf(pkBytes) { + continue + } + previousHash, exists = sr.sentSignatureTracker.GetSignedHash(pkBytes, nonce) + if exists && !bytes.Equal(previousHash, currentHash) { + return sr.waitIfCompetingBlock(ctx, pkBytes, nonce, currentHash) + } + } + + return false +} + +// waitIfCompetingBlock waits if this node already signed a different block for the same nonce. +// The delay is measured from round start. Returns true if signing should be aborted. func (sr *subroundSignature) waitIfCompetingBlock(ctx context.Context, pkBytes []byte, nonce uint64, currentHash []byte) bool { previousHash, exists := sr.sentSignatureTracker.GetSignedHash(pkBytes, nonce) if !exists { @@ -354,10 +376,16 @@ func (sr *subroundSignature) waitIfCompetingBlock(ctx context.Context, pkBytes [ return false } - delay := time.Duration(float64(sr.RoundHandler().TimeDuration()) * competingBlockSignDelay) + // Delay is measured from round start, not from when this function is called + roundStart := sr.GetRoundTimeStamp() + targetTime := time.Duration(float64(sr.RoundHandler().TimeDuration()) * competingBlockSignDelay) + delay := sr.RoundHandler().RemainingTime(roundStart, targetTime) + if delay <= 0 { + log.Debug("waitIfCompetingBlock: already past competing block delay deadline, proceeding to sign") + return false + } // Cap the delay so signing still happens within the signature subround window. - roundStart := sr.GetRoundTimeStamp() sigEndDuration := time.Duration(sr.EndTime()) remaining := sr.RoundHandler().RemainingTime(roundStart, sigEndDuration) safetyMargin := 10 * time.Millisecond diff --git a/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go b/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go index ee1bb5bb02a..a82a54ffa8f 100644 --- a/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go +++ b/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go @@ -108,7 +108,7 @@ func TestWaitIfCompetingBlock_PreviousHashEqualsCurrent(t *testing.T) { assert.False(t, result, "should return false when previous hash equals current hash") } -func TestWaitIfCompetingBlock_NoTimeRemaining(t *testing.T) { +func TestWaitIfCompetingBlock_AlreadyPastDelayDeadline(t *testing.T) { t.Parallel() sr := createSubroundSignatureForCompetingBlockTests( @@ -123,7 +123,37 @@ func TestWaitIfCompetingBlock_NoTimeRemaining(t *testing.T) { return 100 * time.Millisecond }, RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { - // No time remaining in the signature subround + // Already past the competing block delay deadline (and subround end) + return 0 + }, + }, + ) + + result := sr.WaitIfCompetingBlock(context.Background(), []byte("pk"), 100, []byte("current_hash")) + assert.False(t, result, "should return false (proceed to sign) when already past delay deadline") +} + +func TestWaitIfCompetingBlock_NoTimeRemainingInSubround(t *testing.T) { + t.Parallel() + + sr := createSubroundSignatureForCompetingBlockTests( + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + return []byte("previous_hash"), true + }, + }, + nil, + &testscommon.RoundHandlerMock{ + TimeDurationCalled: func() time.Duration { + return 600 * time.Millisecond + }, + RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { + // targetTime = 300ms: still has time to target + // sigEndDuration (85ms): no time left + if maxTime > 200*time.Millisecond { + return 200 * time.Millisecond + } + // No time remaining in signature subround return 0 }, }, @@ -227,8 +257,8 @@ func TestWaitIfCompetingBlock_DeadlineExpiresNoProof(t *testing.T) { return 100 * time.Millisecond }, RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { - // Enough remaining time so delay is not capped to 0 - return 200 * time.Millisecond + // Simulate round just started: remaining = maxTime + return maxTime }, }, ) @@ -238,8 +268,9 @@ func TestWaitIfCompetingBlock_DeadlineExpiresNoProof(t *testing.T) { elapsed := time.Since(start) assert.False(t, result, "should return false (proceed to sign) when deadline expires") - // competingBlockSignDelay = 0.5, roundDuration = 100ms, delay = 50ms - // This should be capped to min(50ms, 200ms - 10ms) = 50ms + // competingBlockSignDelay = 0.5, roundDuration = 100ms + // targetTime = 50ms, sigEndDuration = 85ms (0.85 * 100ms) + // delay = min(50ms, 85ms - 10ms) = 50ms assert.GreaterOrEqual(t, elapsed, 40*time.Millisecond, "should have waited at least ~50ms") } @@ -259,10 +290,13 @@ func TestWaitIfCompetingBlock_DelayCappedBySubroundRemaining(t *testing.T) { }, &testscommon.RoundHandlerMock{ TimeDurationCalled: func() time.Duration { - return 600 * time.Millisecond // delay would be 300ms + return 600 * time.Millisecond // targetTime = 300ms }, RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { - return 60 * time.Millisecond // only 60ms left, maxDelay = 50ms + // Simulate round just started: remaining = maxTime + // targetTime = 300ms, sigEndDuration = 85ms (0.85 * roundTimeDuration=100ms) + // delay = min(300ms, 85ms - 10ms) = 75ms + return maxTime }, }, ) @@ -272,7 +306,7 @@ func TestWaitIfCompetingBlock_DelayCappedBySubroundRemaining(t *testing.T) { elapsed := time.Since(start) assert.False(t, result, "should return false (proceed to sign) after capped delay expires") - // maxDelay = 60ms - 10ms safety = 50ms + // delay should be capped to 75ms (sigEndDuration 85ms - 10ms safety), not full 300ms assert.Less(t, elapsed, 150*time.Millisecond, "delay should be capped, not full 300ms") } @@ -336,6 +370,240 @@ func TestWaitIfCompetingBlock_RecordSignedNonceCalledBeforeBroadcast(t *testing. assert.True(t, recordCalled, "RecordSignedNonce should be called before broadcast") } +func TestWaitIfCompetingBlockForNode_NoCompetingBlockForAnyKey(t *testing.T) { + t.Parallel() + + sr := createSubroundSignatureForCompetingBlockTests( + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + return nil, false // no key has previously signed + }, + }, + nil, + nil, + ) + + result := sr.WaitIfCompetingBlockForNode(context.Background(), 100, []byte("current_hash")) + assert.False(t, result, "should return false when no key has a competing block") +} + +func TestWaitIfCompetingBlockForNode_SameHashForAllKeys(t *testing.T) { + t.Parallel() + + currentHash := []byte("current_hash") + sr := createSubroundSignatureForCompetingBlockTests( + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + return currentHash, true // all keys signed the same hash + }, + }, + nil, + nil, + ) + + result := sr.WaitIfCompetingBlockForNode(context.Background(), 100, currentHash) + assert.False(t, result, "should return false when all keys signed the same hash") +} + +func TestWaitIfCompetingBlockForNode_SelfKeyHasCompetingBlock(t *testing.T) { + t.Parallel() + + container := consensusMocks.InitConsensusCore() + container.SetRoundHandler(&testscommon.RoundHandlerMock{ + TimeDurationCalled: func() time.Duration { + return 100 * time.Millisecond + }, + RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { + return maxTime + }, + }) + + consensusState := initializers.InitConsensusState() + ch := make(chan bool, 1) + + sr, _ := spos.NewSubround( + bls.SrBlock, + bls.SrSignature, + bls.SrEndRound, + roundTimeDuration, + 0.25, + 0.85, + "(SIGNATURE)", + consensusState, + ch, + executeStoredMessages, + container, + chainID, + currentPid, + &statusHandler.AppStatusHandlerStub{}, + ) + + selfPk := sr.SelfPubKey() + + srSignature, _ := v2.NewSubroundSignature( + sr, + &statusHandler.AppStatusHandlerStub{}, + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + if string(pkBytes) == selfPk { + return []byte("different_hash"), true + } + return nil, false + }, + }, + &consensusMocks.SposWorkerMock{}, + &dataRetrieverMock.ThrottlerStub{}, + ) + + srSignature.SetHeader(&block.Header{Nonce: 100}) + srSignature.SetData([]byte("current_hash")) + + start := time.Now() + result := srSignature.WaitIfCompetingBlockForNode(context.Background(), 100, []byte("current_hash")) + elapsed := time.Since(start) + + // Should have waited (delay from round start) and returned false (no proof arrived) + assert.False(t, result, "should return false after delay expires") + assert.GreaterOrEqual(t, elapsed, 40*time.Millisecond, "should have waited for competing block delay") +} + +func TestWaitIfCompetingBlockForNode_ManagedKeyHasCompetingBlock(t *testing.T) { + t.Parallel() + + container := consensusMocks.InitConsensusCore() + container.SetRoundHandler(&testscommon.RoundHandlerMock{ + TimeDurationCalled: func() time.Duration { + return 100 * time.Millisecond + }, + RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { + return maxTime + }, + }) + + // Self key has no competing block, but a managed key does + consensusState := initializers.InitConsensusStateWithKeysHandler( + &testscommon.KeysHandlerStub{ + IsKeyManagedByCurrentNodeCalled: func(pkBytes []byte) bool { + // Mark the first consensus group member as managed + return string(pkBytes) == "A" + }, + }, + ) + ch := make(chan bool, 1) + + sr, _ := spos.NewSubround( + bls.SrBlock, + bls.SrSignature, + bls.SrEndRound, + roundTimeDuration, + 0.25, + 0.85, + "(SIGNATURE)", + consensusState, + ch, + executeStoredMessages, + container, + chainID, + currentPid, + &statusHandler.AppStatusHandlerStub{}, + ) + + selfPk := sr.SelfPubKey() + + srSignature, _ := v2.NewSubroundSignature( + sr, + &statusHandler.AppStatusHandlerStub{}, + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + if string(pkBytes) == selfPk { + // Self key: no competing block + return nil, false + } + if string(pkBytes) == "A" { + // Managed key "A": has competing block + return []byte("old_hash"), true + } + return nil, false + }, + }, + &consensusMocks.SposWorkerMock{}, + &dataRetrieverMock.ThrottlerStub{}, + ) + + srSignature.SetHeader(&block.Header{Nonce: 100}) + srSignature.SetData([]byte("current_hash")) + + start := time.Now() + result := srSignature.WaitIfCompetingBlockForNode(context.Background(), 100, []byte("current_hash")) + elapsed := time.Since(start) + + // Managed key "A" has a competing block, so the node should wait + assert.False(t, result, "should return false after delay expires (no proof arrived)") + assert.GreaterOrEqual(t, elapsed, 40*time.Millisecond, "should have waited for competing block delay") +} + +func TestWaitIfCompetingBlockForNode_WaitsOnceNotPerKey(t *testing.T) { + t.Parallel() + + // This test verifies that waitIfCompetingBlockForNode returns after a single wait + // even when multiple keys have competing blocks - it should not wait per-key. + container := consensusMocks.InitConsensusCore() + container.SetRoundHandler(&testscommon.RoundHandlerMock{ + TimeDurationCalled: func() time.Duration { + return 100 * time.Millisecond + }, + RemainingTimeCalled: func(startTime time.Time, maxTime time.Duration) time.Duration { + return maxTime + }, + }) + + consensusState := initializers.InitConsensusState() + ch := make(chan bool, 1) + + sr, _ := spos.NewSubround( + bls.SrBlock, + bls.SrSignature, + bls.SrEndRound, + roundTimeDuration, + 0.25, + 0.85, + "(SIGNATURE)", + consensusState, + ch, + executeStoredMessages, + container, + chainID, + currentPid, + &statusHandler.AppStatusHandlerStub{}, + ) + + srSignature, _ := v2.NewSubroundSignature( + sr, + &statusHandler.AppStatusHandlerStub{}, + &testscommon.SentSignatureTrackerStub{ + GetSignedHashCalled: func(pkBytes []byte, nonce uint64) ([]byte, bool) { + // ALL keys have signed a different hash + return []byte("old_hash"), true + }, + }, + &consensusMocks.SposWorkerMock{}, + &dataRetrieverMock.ThrottlerStub{}, + ) + + srSignature.SetHeader(&block.Header{Nonce: 100}) + srSignature.SetData([]byte("current_hash")) + + start := time.Now() + result := srSignature.WaitIfCompetingBlockForNode(context.Background(), 100, []byte("current_hash")) + elapsed := time.Since(start) + + // Should return after ONE wait, not multiple + assert.False(t, result) + // targetTime = 50ms, sigEndDuration = 85ms, delay = min(50ms, 75ms) = 50ms + // Should only wait once (~50ms), not per-key + assert.Less(t, elapsed, 120*time.Millisecond, "should have waited only once, not per-key") +} + func TestShouldSendProof_GracePeriodNotExpired(t *testing.T) { t.Parallel() From 918129f337629c78d084fff29a677e0a6bbcbc72 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Mon, 16 Mar 2026 17:32:06 +0200 Subject: [PATCH 2/2] improve timing tests remove redundant log --- consensus/spos/bls/v2/subroundEndRound.go | 1 - consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go | 4 ---- 2 files changed, 5 deletions(-) diff --git a/consensus/spos/bls/v2/subroundEndRound.go b/consensus/spos/bls/v2/subroundEndRound.go index 261b9123668..2384bc15944 100644 --- a/consensus/spos/bls/v2/subroundEndRound.go +++ b/consensus/spos/bls/v2/subroundEndRound.go @@ -389,7 +389,6 @@ func (sr *subroundEndRound) sendProof() (bool, error) { // Re-check grace period after aggregation which may have been slow under CPU contention if !sr.shouldSendProof() { - log.Debug("sendProof: grace period expired during aggregation, not broadcasting") return false, nil } diff --git a/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go b/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go index a82a54ffa8f..96ad2f61953 100644 --- a/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go +++ b/consensus/spos/bls/v2/subroundSignatureCompetingBlock_test.go @@ -406,8 +406,6 @@ func TestWaitIfCompetingBlockForNode_SameHashForAllKeys(t *testing.T) { } func TestWaitIfCompetingBlockForNode_SelfKeyHasCompetingBlock(t *testing.T) { - t.Parallel() - container := consensusMocks.InitConsensusCore() container.SetRoundHandler(&testscommon.RoundHandlerMock{ TimeDurationCalled: func() time.Duration { @@ -468,8 +466,6 @@ func TestWaitIfCompetingBlockForNode_SelfKeyHasCompetingBlock(t *testing.T) { } func TestWaitIfCompetingBlockForNode_ManagedKeyHasCompetingBlock(t *testing.T) { - t.Parallel() - container := consensusMocks.InitConsensusCore() container.SetRoundHandler(&testscommon.RoundHandlerMock{ TimeDurationCalled: func() time.Duration {