Merge pull request #7915 from multiversx/contended-blocks-finality

contended blocks finality

Author: AdoAdoAdo

common/common.go

@@ -648,5 +648,11 @@ func IsContendedHeader(header data.HeaderHandler, parentHeader data.HeaderHandle
 		return false
 	}
 
-	return header.GetRound() > parentHeader.GetRound()+1
+	return IsContendedRound(header.GetRound(), parentHeader.GetRound())
+}
+
+// IsContendedRound returns true if rounds were skipped between a parent at parentRound and its
+// child at round
+func IsContendedRound(round uint64, parentRound uint64) bool {
+	return round > parentRound+1
 }

common/common_test.go

@@ -1388,3 +1388,11 @@ func TestIsContendedHeader(t *testing.T) {
 	require.True(t, common.IsContendedHeader(&testscommon.HeaderHandlerStub{RoundField: 3}, parent))
 	require.True(t, common.IsContendedHeader(&testscommon.HeaderHandlerStub{RoundField: 10}, parent))
 }
+
+func TestIsContendedRound(t *testing.T) {
+	t.Parallel()
+
+	require.False(t, common.IsContendedRound(2, 1))
+	require.True(t, common.IsContendedRound(3, 1))
+	require.True(t, common.IsContendedRound(10, 1))
+}

process/block/baseProcess.go

@@ -140,6 +140,7 @@ type baseProcessor struct {
 
 	processDataTriesOnCommitEpoch bool
 	lastRestartNonce              uint64
+	lastSignaledFinalNonce        uint64
 	pruningDelay                  uint32
 	processedMiniBlocksTracker    process.ProcessedMiniBlocksTracker
 	receiptsRepository            receiptsRepository

process/block/metablock.go

@@ -1697,6 +1697,11 @@ func (mp *metaProcessor) updateState(metaBlock data.MetaHeaderHandler, metaBlock
 
 	// for header v3, trie prnning is triggered in async mode from headers executor
 
+	if mp.enableEpochsHandler.IsFlagEnabledInEpoch(common.SupernovaFlag, metaBlock.GetEpoch()) {
+		mp.signalNewlyFinalBlocks(metaBlock, metaBlockHash, rootHash, prevMetaBlock)
+		return
+	}
+
 	outportFinalizedHeaderHash := metaBlockHash
 	if !common.IsFlagEnabledAfterEpochsStartBlock(metaBlock, mp.enableEpochsHandler, common.AndromedaFlag) {
 		outportFinalizedHeaderHash = metaBlock.GetPrevHash()
@@ -1706,6 +1711,46 @@ func (mp *metaProcessor) updateState(metaBlock data.MetaHeaderHandler, metaBlock
 	mp.blockChain.SetFinalBlockInfo(metaBlock.GetNonce(), metaBlockHash, rootHash)
 }
 
+// signalNewlyFinalBlocks emits the external finality signals for the blocks finalized since the
+// previous commit; with deferred finality the committed header itself may not be final yet
+func (mp *metaProcessor) signalNewlyFinalBlocks(
+	metaBlock data.MetaHeaderHandler,
+	metaBlockHash []byte,
+	rootHash []byte,
+	prevMetaBlock data.MetaHeaderHandler,
+) {
+	finalNonce := mp.forkDetector.GetHighestFinalBlockNonce()
+	if finalNonce <= mp.lastSignaledFinalNonce {
+		return
+	}
+
+	// finality is deferred by at most one block on meta (settle-on-child), so the previous block
+	// is the only one that can become final together with the committed one
+	isPrevNewlyFinal := finalNonce >= prevMetaBlock.GetNonce() &&
+		prevMetaBlock.GetNonce() > mp.lastSignaledFinalNonce
+	if isPrevNewlyFinal {
+		mp.setFinalizedHeaderHashInIndexer(metaBlock.GetPrevHash())
+	}
+
+	if finalNonce >= metaBlock.GetNonce() {
+		mp.setFinalizedHeaderHashInIndexer(metaBlockHash)
+		mp.blockChain.SetFinalBlockInfo(metaBlock.GetNonce(), metaBlockHash, rootHash)
+		mp.lastSignaledFinalNonce = metaBlock.GetNonce()
+		return
+	}
+
+	if finalNonce == prevMetaBlock.GetNonce() {
+		prevRootHash, _, err := mp.getRootHashAndValidatorRootHash(prevMetaBlock)
+		if err != nil {
+			log.Warn("signalNewlyFinalBlocks: cannot get root hash of the newly final block", "error", err.Error())
+		} else {
+			mp.blockChain.SetFinalBlockInfo(prevMetaBlock.GetNonce(), metaBlock.GetPrevHash(), prevRootHash)
+		}
+	}
+
+	mp.lastSignaledFinalNonce = finalNonce
+}
+
 func (mp *metaProcessor) pruneTrieHeaderV3(
 	metaBlock data.HeaderHandler,
 ) {

process/block/metablock_test.go

@@ -5870,3 +5870,94 @@ func TestMetaProcessor_CheckShardHeadersValidityContendedGate(t *testing.T) {
 		assert.Nil(t, err)
 	})
 }
+
+func TestMetaProcessor_UpdateStateSignalsNewlyFinalBlocksUnderSupernova(t *testing.T) {
+	t.Parallel()
+
+	hash4, hash5, hash6, hash7, hash8 := []byte("hash4"), []byte("hash5"), []byte("hash6"), []byte("hash7"), []byte("hash8")
+	newMetaBlock := func(nonce uint64, round uint64, prevHash []byte) *block.MetaBlock {
+		return &block.MetaBlock{
+			Nonce:                  nonce,
+			Round:                  round,
+			PrevHash:               prevHash,
+			RootHash:               append([]byte("rootHash"), byte(nonce)),
+			ValidatorStatsRootHash: append([]byte("validatorRootHash"), byte(nonce)),
+		}
+	}
+	hdr4 := newMetaBlock(4, 4, []byte("hash3"))
+	hdr5 := newMetaBlock(5, 5, hash4)
+	contendedHdr6 := newMetaBlock(6, 8, hash5)
+	contendedHdr7 := newMetaBlock(7, 11, hash6)
+	cleanHdr8 := newMetaBlock(8, 12, hash7)
+	headersByHash := map[string]data.HeaderHandler{
+		string(hash4): hdr4,
+		string(hash5): hdr5,
+		string(hash6): contendedHdr6,
+		string(hash7): contendedHdr7,
+	}
+
+	coreComponents, dataComponents, bootstrapComponents, statusComponents := createMockComponentHolders()
+	coreComponents.EnableEpochsHandlerField = &enableEpochsHandlerMock.EnableEpochsHandlerStub{
+		IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool {
+			return flag == common.AndromedaFlag || flag == common.SupernovaFlag
+		},
+	}
+
+	dataPool := initDataPool()
+	dataPool.HeadersCalled = func() dataRetriever.HeadersPool {
+		return &mock.HeadersCacherStub{
+			GetHeaderByHashCalled: func(hash []byte) (data.HeaderHandler, error) {
+				header, ok := headersByHash[string(hash)]
+				if !ok {
+					return nil, errors.New("header not found")
+				}
+				return header, nil
+			},
+		}
+	}
+	dataComponents.DataPool = dataPool
+
+	setFinalBlockInfos := make([]uint64, 0)
+	dataComponents.BlockChain = &testscommon.ChainHandlerStub{
+		GetGenesisHeaderCalled: func() data.HeaderHandler {
+			return &block.Header{Nonce: 0}
+		},
+		SetFinalBlockInfoCalled: func(nonce uint64, headerHash []byte, rootHash []byte) {
+			setFinalBlockInfos = append(setFinalBlockInfos, nonce)
+		},
+	}
+
+	outportCapture := &finalizedBlocksCapture{OutportStub: &outport.OutportStub{}}
+	statusComponents.Outport = outportCapture
+
+	finalNonce := uint64(5)
+	arguments := createMockMetaArguments(coreComponents, dataComponents, bootstrapComponents, statusComponents)
+	arguments.ForkDetector = &mock.ForkDetectorMock{
+		GetHighestFinalBlockNonceCalled: func() uint64 {
+			return finalNonce
+		},
+	}
+	mp, _ := processBlock.NewMetaProcessor(arguments)
+
+	// clean committed block is final and signaled at once
+	mp.UpdateState(hdr5, hash5)
+	require.Equal(t, [][]byte{hash4, hash5}, outportCapture.finalizedHashes)
+	require.Equal(t, []uint64{5}, setFinalBlockInfos)
+
+	// contended committed block is not signaled while unsettled
+	mp.UpdateState(contendedHdr6, hash6)
+	require.Equal(t, [][]byte{hash4, hash5}, outportCapture.finalizedHashes)
+	require.Equal(t, []uint64{5}, setFinalBlockInfos)
+
+	// the committed child settles its parent: the parent is signaled, the contended child is not
+	finalNonce = 6
+	mp.UpdateState(contendedHdr7, hash7)
+	require.Equal(t, [][]byte{hash4, hash5, hash6}, outportCapture.finalizedHashes)
+	require.Equal(t, []uint64{5, 6}, setFinalBlockInfos)
+
+	// clean child settles its parent and is instantly final: both signaled exactly once
+	finalNonce = 8
+	mp.UpdateState(cleanHdr8, hash8)
+	require.Equal(t, [][]byte{hash4, hash5, hash6, hash7, hash8}, outportCapture.finalizedHashes)
+	require.Equal(t, []uint64{5, 6, 8}, setFinalBlockInfos)
+}

process/block/shardblock.go

@@ -1302,12 +1302,81 @@ func (sp *shardProcessor) updateState(headers []data.HeaderHandler, currentHeade
 		return
 	}
 
+	if sp.enableEpochsHandler.IsFlagEnabledInEpoch(common.SupernovaFlag, currentHeader.GetEpoch()) {
+		sp.signalNewlyFinalBlocks(currentHeader, currentHeaderHash)
+		return
+	}
+
 	sp.setFinalizedHeaderHashInIndexer(currentHeaderHash)
 
 	scheduledHeaderRootHash, _ := sp.scheduledTxsExecutionHandler.GetScheduledRootHashForHeader(currentHeaderHash)
 	sp.setFinalBlockInfo(currentHeader, currentHeaderHash, scheduledHeaderRootHash)
 }
 
+// signalNewlyFinalBlocks emits the external finality signals for the blocks finalized since the
+// previous commit; with deferred finality the committed header itself may not be final yet
+func (sp *shardProcessor) signalNewlyFinalBlocks(currentHeader data.HeaderHandler, currentHeaderHash []byte) {
+	finalNonce := sp.forkDetector.GetHighestFinalBlockNonce()
+	finalHash := sp.forkDetector.GetHighestFinalBlockHash()
+	if len(finalHash) == 0 || finalNonce <= sp.lastSignaledFinalNonce {
+		return
+	}
+
+	newlyFinalHashes := sp.getNewlyFinalHashes(finalNonce, finalHash, currentHeader, currentHeaderHash)
+	for i := len(newlyFinalHashes) - 1; i >= 0; i-- {
+		sp.setFinalizedHeaderHashInIndexer(newlyFinalHashes[i])
+	}
+	sp.lastSignaledFinalNonce = finalNonce
+
+	if finalNonce != currentHeader.GetNonce() {
+		return
+	}
+
+	scheduledHeaderRootHash, _ := sp.scheduledTxsExecutionHandler.GetScheduledRootHashForHeader(currentHeaderHash)
+	sp.setFinalBlockInfo(currentHeader, currentHeaderHash, scheduledHeaderRootHash)
+}
+
+// getNewlyFinalHashes walks the prev-hash chain from the final block back to the last signaled
+// nonce and returns the hashes in descending nonce order
+func (sp *shardProcessor) getNewlyFinalHashes(
+	finalNonce uint64,
+	finalHash []byte,
+	currentHeader data.HeaderHandler,
+	currentHeaderHash []byte,
+) [][]byte {
+	hashes := [][]byte{finalHash}
+	isFirstSignal := sp.lastSignaledFinalNonce == 0
+	if isFirstSignal || finalNonce == sp.lastSignaledFinalNonce+1 {
+		return hashes
+	}
+
+	header := currentHeader
+	if !bytes.Equal(finalHash, currentHeaderHash) {
+		var err error
+		header, err = process.GetShardHeader(finalHash, sp.dataPool.Headers(), sp.marshalizer, sp.store)
+		if err != nil {
+			log.Warn("signalNewlyFinalBlocks: cannot load final header, signaling only its hash",
+				"final nonce", finalNonce, "error", err.Error())
+			return hashes
+		}
+	}
+
+	for nonce := finalNonce - 1; nonce > sp.lastSignaledFinalNonce; nonce-- {
+		hash := header.GetPrevHash()
+		var err error
+		header, err = process.GetShardHeader(hash, sp.dataPool.Headers(), sp.marshalizer, sp.store)
+		if err != nil {
+			log.Warn("signalNewlyFinalBlocks: cannot load newly final header, skipping older signals",
+				"nonce", nonce, "error", err.Error())
+			break
+		}
+
+		hashes = append(hashes, hash)
+	}
+
+	return hashes
+}
+
 func (sp *shardProcessor) pruneTrieHeaderV3(
 	header data.HeaderHandler,
 ) {

process/block/shardblock_test.go

@@ -8234,3 +8234,70 @@ func TestShardProcessor_CheckMetaHeadersValidityAndFinalityContendedGate(t *test
 		assert.Nil(t, err)
 	})
 }
+
+type finalizedBlocksCapture struct {
+	*outport.OutportStub
+	finalizedHashes [][]byte
+}
+
+func (capture *finalizedBlocksCapture) FinalizedBlock(finalizedBlock *outportcore.FinalizedBlock) {
+	capture.finalizedHashes = append(capture.finalizedHashes, finalizedBlock.HeaderHash)
+}
+
+func TestShardProcessor_UpdateStateSignalsNewlyFinalBlocksUnderSupernova(t *testing.T) {
+	t.Parallel()
+
+	coreComponents, dataComponents, bootstrapComponents, statusComponents := createComponentHolderMocks()
+	coreComponents.EnableEpochsHandlerField = &enableEpochsHandlerMock.EnableEpochsHandlerStub{
+		IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool {
+			return flag == common.AndromedaFlag || flag == common.SupernovaFlag
+		},
+	}
+	poolsHolder := dataRetrieverMock.NewPoolsHolderMock()
+	dataComponents.DataPool = poolsHolder
+	outportCapture := &finalizedBlocksCapture{OutportStub: &outport.OutportStub{}}
+	statusComponents.Outport = outportCapture
+
+	hash5, hash6, hash7, hash8 := []byte("hash5"), []byte("hash6"), []byte("hash7"), []byte("hash8")
+	hdr5 := &block.Header{Nonce: 5, Round: 5}
+	contendedHdr6 := &block.Header{Nonce: 6, Round: 8, PrevHash: hash5}
+	cleanHdr7 := &block.Header{Nonce: 7, Round: 9, PrevHash: hash6}
+	cleanHdr8 := &block.Header{Nonce: 8, Round: 10, PrevHash: hash7}
+	poolsHolder.Headers().AddHeader(hash6, contendedHdr6)
+	poolsHolder.Headers().AddHeader(hash7, cleanHdr7)
+
+	finalNonce, finalHash := uint64(5), hash5
+
+	arguments := CreateMockArguments(coreComponents, dataComponents, bootstrapComponents, statusComponents)
+	arguments.ForkDetector = &mock.ForkDetectorMock{
+		GetHighestFinalBlockNonceCalled: func() uint64 {
+			return finalNonce
+		},
+		GetHighestFinalBlockHashCalled: func() []byte {
+			return finalHash
+		},
+	}
+	sp, _ := blproc.NewShardProcessor(arguments)
+
+	// clean committed block is final and signaled at once
+	sp.UpdateStateStorage(nil, hdr5, hash5)
+	require.Equal(t, [][]byte{hash5}, outportCapture.finalizedHashes)
+	finalInfoNonce, finalInfoHash, _ := dataComponents.BlockChain.GetFinalBlockInfo()
+	require.Equal(t, uint64(5), finalInfoNonce)
+	require.Equal(t, hash5, finalInfoHash)
+
+	// contended committed block and its descendant are not signaled while unsettled
+	sp.UpdateStateStorage(nil, contendedHdr6, hash6)
+	sp.UpdateStateStorage(nil, cleanHdr7, hash7)
+	require.Equal(t, [][]byte{hash5}, outportCapture.finalizedHashes)
+	finalInfoNonce, _, _ = dataComponents.BlockChain.GetFinalBlockInfo()
+	require.Equal(t, uint64(5), finalInfoNonce)
+
+	// settlement advances the final checkpoint: each newly final block is signaled exactly once
+	finalNonce, finalHash = 8, hash8
+	sp.UpdateStateStorage(nil, cleanHdr8, hash8)
+	require.Equal(t, [][]byte{hash5, hash6, hash7, hash8}, outportCapture.finalizedHashes)
+	finalInfoNonce, finalInfoHash, _ = dataComponents.BlockChain.GetFinalBlockInfo()
+	require.Equal(t, uint64(8), finalInfoNonce)
+	require.Equal(t, hash8, finalInfoHash)
+}

process/sync/baseForkDetector.go

@@ -18,6 +18,7 @@ type headerInfo struct {
 	nonce    uint64
 	round    uint64
 	hash     []byte
+	prevHash []byte
 	state    process.BlockHeaderState
 	hasProof bool
 }
@@ -412,6 +413,43 @@ func (bfd *baseForkDetector) setFinalCheckpoint(finalCheckpoint *checkpointInfo)
 	bfd.mutFork.Unlock()
 }
 
+// advanceFinalCheckpoint sets the final checkpoint only forward, so concurrent computations
+// cannot regress an already finalized nonce
+func (bfd *baseForkDetector) advanceFinalCheckpoint(finalCheckpoint *checkpointInfo) {
+	bfd.mutFork.Lock()
+	if finalCheckpoint.nonce > bfd.fork.finalCheckpoint.nonce {
+		bfd.fork.finalCheckpoint = finalCheckpoint
+	}
+	bfd.mutFork.Unlock()
+}
+
+func (bfd *baseForkDetector) isSupernovaForHeader(header data.HeaderHandler) bool {
+	return bfd.enableEpochsHandler.IsFlagEnabledInEpoch(common.SupernovaFlag, header.GetEpoch())
+}
+
+func isParentCheckpoint(checkpoint *checkpointInfo, header data.HeaderHandler) bool {
+	if checkpoint.nonce+1 != header.GetNonce() {
+		return false
+	}
+
+	return len(checkpoint.hash) == 0 || bytes.Equal(checkpoint.hash, header.GetPrevHash())
+}
+
+// canInstantlyFinalize returns false for a contended header or one whose parent is not final yet;
+// such headers finalize later, on settlement (R-INTRA)
+func (bfd *baseForkDetector) canInstantlyFinalize(header data.HeaderHandler) bool {
+	if !bfd.isSupernovaForHeader(header) {
+		return true
+	}
+
+	finalCheckpoint := bfd.finalCheckpoint()
+	if !isParentCheckpoint(finalCheckpoint, header) {
+		return false
+	}
+
+	return !common.IsContendedRound(header.GetRound(), finalCheckpoint.round)
+}
+
 // RestoreToGenesis sets class variables to theirs initial values
 func (bfd *baseForkDetector) RestoreToGenesis() {
 	bfd.mutHeaders.Lock()
@@ -935,6 +973,7 @@ func (bfd *baseForkDetector) processReceivedBlock(
 		nonce:    header.GetNonce(),
 		round:    header.GetRound(),
 		hash:     headerHash,
+		prevHash: header.GetPrevHash(),
 		state:    state,
 		hasProof: hasProof,
 	}