Fix shard stuck

consensus/spos/bls/v2/subroundEndRound.go

@@ -247,26 +247,40 @@ func (sr *subroundEndRound) doEndRoundJobByNode() bool {
 		if !sr.waitForSignalSync() {
 			return false
 		}
+		_, _ = sr.sendProof()
 	}
 
-	proof, ok := sr.sendProof()
-	if !ok {
-		return false
+	return sr.finalizeConfirmedBlock()
+}
+
+func (sr *subroundEndRound) waitForProof(shardID uint32, headerHash []byte) bool {
+	if sr.EquivalentProofsPool().HasProof(shardID, headerHash) {
+		return true
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), sr.RoundHandler().TimeDuration())
+	defer cancel()
+
+	for {
+		select {
+		case <-time.After(time.Millisecond):
+			if sr.EquivalentProofsPool().HasProof(shardID, headerHash) {
+				return true
+			}
+		case <-ctx.Done():
+			return false
+		}
 	}
+}
+
+func (sr *subroundEndRound) finalizeConfirmedBlock() bool {
+	sr.waitForProof(sr.ShardCoordinator().SelfId(), sr.GetData())
 
 	err := sr.commitBlock()
 	if err != nil {
 		return false
 	}
 
-	// if proof not nil, it was created and broadcasted so it has to be added to the pool
-	if proof != nil {
-		ok := sr.EquivalentProofsPool().AddProof(proof)
-		if !ok {
-			log.Trace("doEndRoundJobByNode.AddProof", "added", ok)
-		}
-	}
-
 	sr.SetStatus(sr.Current(), spos.SsFinished)
 
 	sr.worker.DisplayStatistics()

consensus/spos/bls/v2/subroundEndRound_test.go

@@ -1227,8 +1227,6 @@ func TestSubroundEndRound_DoEndRoundJobByNode(t *testing.T) {
 	t.Run("should work with equivalent messages flag active", func(t *testing.T) {
 		t.Parallel()
 
-		providedPrevSig := []byte("prev sig")
-		providedPrevBitmap := []byte{1, 1, 1, 1, 1, 1, 1, 1, 1}
 		container := consensusMocks.InitConsensusCore()
 		container.SetBlockchain(&testscommon.ChainHandlerStub{
 			GetGenesisHeaderCalled: func() data.HeaderHandler {
@@ -1241,16 +1239,7 @@ func TestSubroundEndRound_DoEndRoundJobByNode(t *testing.T) {
 			},
 		}
 		container.SetEnableEpochsHandler(enableEpochsHandler)
-
-		wasSetCurrentHeaderProofCalled := false
-		container.SetEquivalentProofsPool(&dataRetriever.ProofsPoolMock{
-			AddProofCalled: func(headerProof data.HeaderProofHandler) bool {
-				wasSetCurrentHeaderProofCalled = true
-				require.NotEqual(t, providedPrevSig, headerProof.GetAggregatedSignature())
-				require.NotEqual(t, providedPrevBitmap, headerProof.GetPubKeysBitmap())
-				return true
-			},
-		})
+		container.SetEquivalentProofsPool(&dataRetriever.ProofsPoolMock{})
 
 		ch := make(chan bool, 1)
 		consensusState := initializers.InitConsensusState()
@@ -1295,7 +1284,6 @@ func TestSubroundEndRound_DoEndRoundJobByNode(t *testing.T) {
 
 		r := srEndRound.DoEndRoundJobByNode()
 		require.True(t, r)
-		require.True(t, wasSetCurrentHeaderProofCalled)
 	})
 }
 

process/sync/baseForkDetector.go

@@ -760,11 +760,13 @@ func (bfd *baseForkDetector) processReceivedBlock(
 	bfd.setHighestNonceReceived(header.GetNonce())
 
 	if state == process.BHProposed || !hasProof {
+		log.Trace("forkDetector.processReceivedBlock: block is proposed or has no proof", "state", state, "has proof", hasProof)
 		return
 	}
 
 	isHeaderReceivedTooLate := bfd.isHeaderReceivedTooLate(header, state, process.BlockFinality)
 	if isHeaderReceivedTooLate {
+		log.Trace("forkDetector.processReceivedBlock: block is received too late", "initial state", state)
 		state = process.BHReceivedTooLate
 	}
 
@@ -778,6 +780,7 @@ func (bfd *baseForkDetector) processReceivedBlock(
 	}
 
 	if !bfd.append(hInfo) {
+		log.Trace("forkDetector.processReceivedBlock: header not appended", "nonce", hInfo.nonce, "hash", hInfo.hash)
 		return
 	}
 

process/sync/metaForkDetector.go

@@ -108,7 +108,11 @@ func (mfd *metaForkDetector) doJobOnBHProcessed(
 	_ [][]byte,
 ) {
 	mfd.setFinalCheckpoint(mfd.lastCheckpoint())
-	mfd.addCheckpoint(&checkpointInfo{nonce: header.GetNonce(), round: header.GetRound(), hash: headerHash})
+	newCheckpoint := &checkpointInfo{nonce: header.GetNonce(), round: header.GetRound(), hash: headerHash}
+	mfd.addCheckpoint(newCheckpoint)
+	if mfd.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, header.GetEpoch()) {
+		mfd.setFinalCheckpoint(newCheckpoint)
+	}
 	mfd.removePastOrInvalidRecords()
 }
 

process/sync/shardForkDetector.go

@@ -112,7 +112,11 @@ func (sfd *shardForkDetector) doJobOnBHProcessed(
 ) {
 	_ = sfd.appendSelfNotarizedHeaders(selfNotarizedHeaders, selfNotarizedHeadersHashes, core.MetachainShardId)
 	sfd.computeFinalCheckpoint()
-	sfd.addCheckpoint(&checkpointInfo{nonce: header.GetNonce(), round: header.GetRound(), hash: headerHash})
+	newCheckpoint := &checkpointInfo{nonce: header.GetNonce(), round: header.GetRound(), hash: headerHash}
+	sfd.addCheckpoint(newCheckpoint)
+	if sfd.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, header.GetEpoch()) {
+		sfd.setFinalCheckpoint(newCheckpoint)
+	}
 	sfd.removePastOrInvalidRecords()
 }