avoid triggering twice in parallel

Author: AdoAdoAdo

consensus/interface.go

@@ -228,6 +228,7 @@ type KeysHandler interface {
 // EquivalentProofsPool defines the behaviour of a proofs pool components
 type EquivalentProofsPool interface {
 	AddProof(headerProof data.HeaderProofHandler) bool
+	AddProofIfNoneAtNonce(headerProof data.HeaderProofHandler) (bool, data.HeaderProofHandler)
 	GetProof(shardID uint32, headerHash []byte) (data.HeaderProofHandler, error)
 	GetProofByNonce(headerNonce uint64, shardID uint32) (data.HeaderProofHandler, error)
 	HasProof(shardID uint32, headerHash []byte) bool

consensus/spos/bls/v2/proofSelfAssembly.go

@@ -15,13 +15,20 @@ import (
 // trySelfAssembleProof aggregates the snapshotted signature shares into a proof for the previous
 // round's block; it never creates a second proof: any existing proof at the nonce aborts it
 func trySelfAssembleProof(sr *spos.Subround, store signatureEvidenceHandler, ev *roundSignatureEvidence) {
-	if !ev.assemblyStarted.CompareAndSwap(false, true) {
+	currentRound := sr.RoundHandler().Index()
+	if ev.nextAssemblyRound.Load() > currentRound {
+		return // already attempted this round
+	}
+	if !ev.assemblyRunning.CompareAndSwap(false, true) {
 		return
 	}
-	if !shouldNodeSendProofForGroup(sr, ev.consensusGroup) {
+	defer ev.assemblyRunning.Store(false)
+	ev.nextAssemblyRound.Store(currentRound + 1)
+
+	if abortOnExistingProof(sr, store, ev) {
 		return
 	}
-	if abortOnExistingProof(sr, store, ev, "at entry") {
+	if !shouldNodeSendProofForGroup(sr, ev.consensusGroup) {
 		return
 	}
 
@@ -54,11 +61,14 @@ func trySelfAssembleProof(sr *spos.Subround, store signatureEvidenceHandler, ev
 		IsStartOfEpoch:      ev.isStartOfEpoch,
 	}
 
-	// re-check right before publish, aggregation took time
-	if abortOnExistingProof(sr, store, ev, "before publish") {
-		return
-	}
-	if !sr.EquivalentProofsPool().AddProof(proof) {
+	// atomic add: never overwrites a proof that arrived at the nonce during aggregation
+	added, existing := sr.EquivalentProofsPool().AddProofIfNoneAtNonce(proof)
+	if !added {
+		if !check.IfNil(existing) && !bytes.Equal(existing.GetHeaderHash(), ev.headerHash) {
+			log.Warn("trySelfAssembleProof: competing proof arrived during assembly, aborting broadcast",
+				"nonce", ev.nonce, "ownHash", ev.headerHash, "existingHash", existing.GetHeaderHash())
+		}
+		store.DropRetained(ev.nonce) // nonce settled either way
 		return
 	}
 	store.DropRetained(ev.nonce)
@@ -74,14 +84,14 @@ func trySelfAssembleProof(sr *spos.Subround, store signatureEvidenceHandler, ev
 		"nonce", ev.nonce, "headerHash", ev.headerHash)
 }
 
-func abortOnExistingProof(sr *spos.Subround, store signatureEvidenceHandler, ev *roundSignatureEvidence, stage string) bool {
+func abortOnExistingProof(sr *spos.Subround, store signatureEvidenceHandler, ev *roundSignatureEvidence) bool {
 	existing, err := sr.EquivalentProofsPool().GetProofByNonce(ev.nonce, ev.shardID)
 	if err != nil || check.IfNil(existing) {
 		return false
 	}
 
 	if !bytes.Equal(existing.GetHeaderHash(), ev.headerHash) {
-		log.Warn("trySelfAssembleProof: competing proof already exists at nonce, aborting "+stage,
+		log.Warn("trySelfAssembleProof: competing proof already exists at nonce, aborting assembly",
 			"nonce", ev.nonce, "ownHash", ev.headerHash, "existingHash", existing.GetHeaderHash())
 	}
 	store.DropRetained(ev.nonce) // nonce settled either way
@@ -115,10 +125,12 @@ func stripInvalidShares(sr *spos.Subround, ev *roundSignatureEvidence, bitmap []
 	sem := make(chan struct{}, maxParallelVerifications)
 	var wg sync.WaitGroup
 
+	numVerified := 0
 	for i := range ev.consensusGroup {
 		if !isIndexSetInBitmap(i, bitmap) || shares[i] == nil {
 			continue
 		}
+		numVerified++
 
 		wg.Add(1)
 		sem <- struct{}{}
@@ -140,6 +152,14 @@ func stripInvalidShares(sr *spos.Subround, ev *roundSignatureEvidence, bitmap []
 	}
 	wg.Wait()
 
+	// quorum evidence always holds at least one honest, valid share, so all shares failing
+	// points to an infrastructure error, not adversarial shares: keep the evidence intact
+	if numVerified > 0 && len(invalidIndices) == numVerified {
+		log.Warn("stripInvalidShares: all shares failed verification, keeping evidence for retry",
+			"nonce", ev.nonce, "numShares", numVerified)
+		return
+	}
+
 	ev.dropShares(invalidIndices)
 }
 

consensus/spos/bls/v2/proofSelfAssembly_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-core-go/data/block"
@@ -99,9 +100,9 @@ func createSelfAssemblySetup(t *testing.T) *selfAssemblyTestSetup {
 		GetProofByNonceCalled: func(headerNonce uint64, shardID uint32) (data.HeaderProofHandler, error) {
 			return nil, errors.New("proof not found")
 		},
-		AddProofCalled: func(headerProof data.HeaderProofHandler) bool {
+		AddProofIfNoneAtNonceCalled: func(headerProof data.HeaderProofHandler) (bool, data.HeaderProofHandler) {
 			numAdded.Add(1)
-			return true
+			return true, nil
 		},
 	}
 	container.SetEquivalentProofsPool(proofsPool)
@@ -211,7 +212,7 @@ func TestTrySelfAssembleProof_HappyPath(t *testing.T) {
 	assert.Nil(t, err, "the self-assembled aggregated signature must verify")
 }
 
-func TestTrySelfAssembleProof_AssemblyStartedOnlyOnce(t *testing.T) {
+func TestTrySelfAssembleProof_SingleAttemptPerRound(t *testing.T) {
 	t.Parallel()
 
 	setup := createSelfAssemblySetup(t)
@@ -220,7 +221,7 @@ func TestTrySelfAssembleProof_AssemblyStartedOnlyOnce(t *testing.T) {
 	trySelfAssembleProof(setup.subround, setup.store, ev)
 	trySelfAssembleProof(setup.subround, setup.store, ev)
 
-	assert.Equal(t, int32(1), setup.numAdded.Load(), "the CAS guard must prevent a second assembly")
+	assert.Equal(t, int32(1), setup.numAdded.Load(), "a second attempt in the same round must be skipped")
 }
 
 func TestTrySelfAssembleProof_StripsInvalidSharesAndRetries(t *testing.T) {
@@ -297,18 +298,20 @@ func TestTrySelfAssembleProof_AbortsOnCompetingProof(t *testing.T) {
 		setup := createSelfAssemblySetup(t)
 		ev := setup.createRealEvidence(t, []byte("header hash"), 7)
 
-		var numChecks atomic.Int32
-		setup.proofsPool.GetProofByNonceCalled = func(headerNonce uint64, shardID uint32) (data.HeaderProofHandler, error) {
-			if numChecks.Add(1) == 1 {
-				return nil, errors.New("proof not found")
-			}
-			return &block.HeaderProof{HeaderHash: []byte("competing hash"), HeaderNonce: headerNonce}, nil
+		// entry check finds no proof, the atomic add is beaten by a competing proof
+		setup.proofsPool.AddProofIfNoneAtNonceCalled = func(headerProof data.HeaderProofHandler) (bool, data.HeaderProofHandler) {
+			return false, &block.HeaderProof{HeaderHash: []byte("competing hash"), HeaderNonce: headerProof.GetHeaderNonce()}
 		}
 
+		setup.store.Capture(ev)
+		setup.store.Capture(nil)
+
 		trySelfAssembleProof(setup.subround, setup.store, ev)
 
-		assert.Nil(t, setup.broadcast.Load(), "broadcast must be aborted by the pre-publish re-check")
-		assert.Equal(t, int32(0), setup.numAdded.Load())
+		assert.Nil(t, setup.broadcast.Load(), "broadcast must be aborted by the atomic add")
+
+		_, ok := setup.store.GetRetainedQuorumEvidence(ev.nonce)
+		assert.False(t, ok, "settled nonce must drop the retained slot")
 	})
 
 	t.Run("proof for the own hash means already done", func(t *testing.T) {
@@ -348,6 +351,131 @@ func TestTrySelfAssembleProof_EpochBoundaryUsesSnapshotGroup(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestTrySelfAssembleProof_InProgressAttemptBlocksUntilDone(t *testing.T) {
+	t.Parallel()
+
+	var currentRound atomic.Int64
+	currentRound.Store(10)
+
+	proceed := make(chan struct{})
+	started := make(chan struct{}, 4)
+	var numAggregations atomic.Int32
+
+	container := consensusMocks.InitConsensusCore()
+	container.SetRoundHandler(&testscommon.RoundHandlerMock{
+		IndexCalled: func() int64 {
+			return currentRound.Load()
+		},
+	})
+	container.SetSigningHandler(&consensusMocks.SigningHandlerStub{
+		AggregateSigsWithKeysCalled: func(pubKeys []string, bitmap []byte, sigShares [][]byte, epoch uint32) ([]byte, error) {
+			numAggregations.Add(1)
+			started <- struct{}{}
+			<-proceed
+			return []byte("agg"), nil
+		},
+	})
+	container.SetEquivalentProofsPool(createUnsettledProofsPool())
+
+	sr := createFlowSubround(t, container, bls.SrSignature, "(SIGNATURE)")
+	store, _ := newSignatureEvidenceStore(createUnsettledProofsPool())
+	ev := createTestEvidence(9, 5, 7, 8)
+
+	// first attempt starts in round 10 and blocks inside the aggregation
+	go trySelfAssembleProof(sr, store, ev)
+	<-started
+
+	// next round while the first attempt is still executing: no second attempt
+	currentRound.Store(11)
+	trySelfAssembleProof(sr, store, ev)
+	assert.Equal(t, int32(1), numAggregations.Load())
+
+	close(proceed)
+	require.Eventually(t, func() bool {
+		return !ev.assemblyRunning.Load()
+	}, time.Second, time.Millisecond)
+
+	// round 11 had no attempt started, so a retry is allowed after completion
+	trySelfAssembleProof(sr, store, ev)
+	assert.Equal(t, int32(2), numAggregations.Load())
+
+	// a second attempt in the same round is skipped
+	trySelfAssembleProof(sr, store, ev)
+	assert.Equal(t, int32(2), numAggregations.Load())
+
+	// next round: retried again
+	currentRound.Store(12)
+	trySelfAssembleProof(sr, store, ev)
+	assert.Equal(t, int32(3), numAggregations.Load())
+}
+
+func TestTrySelfAssembleProof_SettledNonceDropsRetainedEvenWhenIneligible(t *testing.T) {
+	t.Parallel()
+
+	var numAggregations atomic.Int32
+	container := consensusMocks.InitConsensusCore()
+	container.SetSigningHandler(&consensusMocks.SigningHandlerStub{
+		AggregateSigsWithKeysCalled: func(pubKeys []string, bitmap []byte, sigShares [][]byte, epoch uint32) ([]byte, error) {
+			numAggregations.Add(1)
+			return []byte("agg"), nil
+		},
+	})
+	container.SetEquivalentProofsPool(createSettledProofsPool([]byte("competing hash")))
+
+	sr := createFlowSubround(t, container, bls.SrSignature, "(SIGNATURE)")
+	store, _ := newSignatureEvidenceStore(createUnsettledProofsPool())
+
+	ev := createTestEvidence(9, 5, 7, 8)
+	for i := range ev.consensusGroup {
+		ev.consensusGroup[i] = "other_" + ev.consensusGroup[i]
+	}
+	store.Capture(ev)
+	store.Capture(nil)
+	_, ok := store.GetRetainedQuorumEvidence(ev.nonce)
+	require.True(t, ok)
+
+	trySelfAssembleProof(sr, store, ev)
+
+	assert.Equal(t, int32(0), numAggregations.Load())
+	_, ok = store.GetRetainedQuorumEvidence(ev.nonce)
+	assert.False(t, ok, "the settled nonce must drop the retained slot before the eligibility check")
+}
+
+func TestTrySelfAssembleProof_KeepsEvidenceWhenAllSharesFailVerification(t *testing.T) {
+	t.Parallel()
+
+	var numAdded atomic.Int32
+	container := consensusMocks.InitConsensusCore()
+	container.SetSigningHandler(&consensusMocks.SigningHandlerStub{
+		AggregateSigsWithKeysCalled: func(pubKeys []string, bitmap []byte, sigShares [][]byte, epoch uint32) ([]byte, error) {
+			return nil, errors.New("infrastructure error")
+		},
+		VerifySigShareWithKeyCalled: func(pubKey []byte, sigShare []byte, message []byte, epoch uint32) error {
+			return errors.New("infrastructure error")
+		},
+	})
+	proofsPool := createUnsettledProofsPool()
+	proofsPool.AddProofIfNoneAtNonceCalled = func(headerProof data.HeaderProofHandler) (bool, data.HeaderProofHandler) {
+		numAdded.Add(1)
+		return true, nil
+	}
+	container.SetEquivalentProofsPool(proofsPool)
+
+	sr := createFlowSubround(t, container, bls.SrSignature, "(SIGNATURE)")
+	store, _ := newSignatureEvidenceStore(createUnsettledProofsPool())
+
+	ev := createTestEvidence(9, 5, 7, 8)
+	store.Capture(ev)
+	store.Capture(nil)
+
+	trySelfAssembleProof(sr, store, ev)
+
+	assert.Equal(t, 8, ev.getCount(), "evidence must stay intact on an all-shares failure")
+	assert.Equal(t, int32(0), numAdded.Load())
+	_, ok := store.GetRetainedQuorumEvidence(ev.nonce)
+	assert.True(t, ok, "no demotion on an all-shares failure")
+}
+
 func TestTrySelfAssembleProof_NotEligibleWithoutGroupKeys(t *testing.T) {
 	t.Parallel()
 

consensus/spos/bls/v2/signatureEvidence.go

@@ -13,16 +13,19 @@ import (
 // roundSignatureEvidence is a snapshot of the signature shares observed for one block,
 // taken at the start of the following round, before the consensus state is reset.
 type roundSignatureEvidence struct {
-	roundIndex      int64
-	nonce           uint64
-	headerHash      []byte
-	epoch           uint32
-	headerRound     uint64
-	shardID         uint32
-	isStartOfEpoch  bool
-	threshold       int
-	consensusGroup  []string
-	assemblyStarted atomic.Bool
+	roundIndex     int64
+	nonce          uint64
+	headerHash     []byte
+	epoch          uint32
+	headerRound    uint64
+	shardID        uint32
+	isStartOfEpoch bool
+	threshold      int
+	consensusGroup []string
+	// assemblyRunning is true strictly while an assembly attempt executes;
+	// nextAssemblyRound is the first round allowed to start the next attempt
+	assemblyRunning   atomic.Bool
+	nextAssemblyRound atomic.Int64
 
 	mut    sync.RWMutex
 	bitmap []byte

consensus/spos/bls/v2/signatureEvidenceFlow_test.go

@@ -179,7 +179,15 @@ func TestSubroundStartRound_CaptureSignatureEvidence(t *testing.T) {
 func TestSubroundStartRound_CaptureTriggersSelfAssemblyOnQuorum(t *testing.T) {
 	t.Parallel()
 
+	var chronologyRound atomic.Int64
+	chronologyRound.Store(11)
+
 	container := consensusMocks.InitConsensusCore()
+	container.SetRoundHandler(&testscommon.RoundHandlerMock{
+		IndexCalled: func() int64 {
+			return chronologyRound.Load()
+		},
+	})
 	container.SetSigningHandler(&consensusMocks.SigningHandlerStub{})
 	container.SetEquivalentProofsPool(createUnsettledProofsPool())
 
@@ -213,7 +221,14 @@ func TestSubroundStartRound_CaptureTriggersSelfAssemblyOnQuorum(t *testing.T) {
 		require.Fail(t, "self-assembly was not triggered at round start")
 	}
 
+	ev, found := store.GetPreviousRoundEvidence(5, 11)
+	require.True(t, found)
+	require.Eventually(t, func() bool {
+		return !ev.assemblyRunning.Load()
+	}, time.Second, time.Millisecond)
+
 	// retained evidence re-fires on the next round while the nonce stays unsettled
+	chronologyRound.Store(12)
 	srStartRound.SetHeader(nil)
 	srStartRound.SetData(nil)
 	srStartRound.captureSignatureEvidence()

consensus/spos/bls/v2/subroundStartRound.go

@@ -128,8 +128,6 @@ func (sr *subroundStartRound) captureSignatureEvidence() {
 		return
 	}
 
-	// allow one assembly attempt per round while the nonce stays unsettled
-	ev.assemblyStarted.Store(false)
 	go trySelfAssembleProof(sr.Subround, sr.signatureEvidence, ev)
 }
 
@@ -330,7 +328,7 @@ func (sr *subroundStartRound) computeNumManagedKeysInConsensusGroup(pubKeys []st
 	return numMultiKeysInConsensusGroup
 }
 
-func (sr *subroundStartRound) indexRoundIfNeeded(pubKeys []string) {
+func (sr *subroundStartRound) indexRoundIfNeeded(_ []string) {
 	sr.outportMutex.RLock()
 	defer sr.outportMutex.RUnlock()