feat: batch processing non-finalised blocks

finality-provider/service/fp_instance.go

@@ -194,52 +194,81 @@ func (fp *FinalityProviderInstance) IsRunning() bool {
 func (fp *FinalityProviderInstance) finalitySigSubmissionLoop() {
 	defer fp.wg.Done()
 
+	var targetHeight uint64
 	for {
 		select {
 		case b := <-fp.poller.GetBlockInfoChan():
-			fp.logger.Debug(
-				"the finality-provider received a new block, start processing",
+			channelSize := len(fp.poller.blockInfoChan)
+			fp.logger.Debug("the finality-provider received a new block",
+				zap.Int("poller_channel_size", channelSize),
 				zap.String("pk", fp.GetBtcPkHex()),
 				zap.Uint64("height", b.Height),
 				zap.String("block_hash", hex.EncodeToString(b.Hash)),
 			)
 
-			// check whether the block has been processed before
-			if fp.hasProcessed(b.Height) {
-				continue
-			}
-			// check whether the finality provider has voting power
-			hasVp, err := fp.hasVotingPower(b.Height)
-			if err != nil {
-				fp.reportCriticalErr(err)
-				continue
-			}
-			if !hasVp {
-				// the finality provider does not have voting power
-				// and it will never will at this block
-				fp.MustSetLastProcessedHeight(b.Height)
-				fp.metrics.IncrementFpTotalBlocksWithoutVotingPower(fp.GetBtcPkHex())
-				continue
-			}
-			// check whether the randomness has been committed
-			// the retry will end if max retry times is reached
-			// or the target block is finalized
-			isFinalized, err := fp.retryCheckRandomnessUntilBlockFinalized(b)
-			if err != nil {
-				if !errors.Is(err, ErrFinalityProviderShutDown) {
-					fp.reportCriticalErr(err)
+			// Fetch all available blocks
+			// Note: not all the blocks in the range will have votes cast
+			// due to lack of voting power or public randomness, so we may
+			// have gaps during processing
+			pollerBlocks := []*types.BlockInfo{b}
+			for {
+				select {
+				case b := <-fp.poller.GetBlockInfoChan():
+					fp.logger.Debug(
+						"the finality-provider received a new block",
+						zap.String("pk", fp.GetBtcPkHex()),
+						zap.Uint64("height", b.Height),
+						zap.String("block_hash", hex.EncodeToString(b.Hash)),
+					)
+
+					// check whether the block has been processed before
+					if fp.hasProcessed(b.Height) {
+						continue
+					}
+					// check whether the finality provider has voting power
+					hasVp, err := fp.hasVotingPower(b.Height)
+					if err != nil {
+						fp.reportCriticalErr(err)
+						continue
+					}
+					if !hasVp {
+						// the finality provider does not have voting power
+						// and it will never will at this block
+						fp.MustSetLastProcessedHeight(b.Height)
+						fp.metrics.IncrementFpTotalBlocksWithoutVotingPower(fp.GetBtcPkHex())
+						continue
+					}
+					// check whether the randomness has been committed
+					// the retry will end if max retry times is reached
+					// or the target block is finalized
+					isFinalized, err := fp.retryCheckRandomnessUntilBlockFinalized(b)
+					if err != nil {
+						if !errors.Is(err, ErrFinalityProviderShutDown) {
+							fp.reportCriticalErr(err)
+						}
+						break
+					}
+					// the block is finalized, no need to submit finality signature
+					if isFinalized {
+						fp.MustSetLastProcessedHeight(b.Height)
+						continue
+					}
+
+					pollerBlocks = append(pollerBlocks, b)
+				default:
+					goto processBlocks
 				}
-				break
 			}
-			// the block is finalized, no need to submit finality signature
-			if isFinalized {
-				fp.MustSetLastProcessedHeight(b.Height)
+		processBlocks:
+			if len(pollerBlocks) == 0 {
 				continue
 			}
-
-			// use the copy of the block to avoid the impact to other receivers
-			nextBlock := *b
-			res, err := fp.retrySubmitFinalitySignatureUntilBlockFinalized(&nextBlock)
+			fp.logger.Debug(
+				"the finality-provider received new block(s), start processing",
+				zap.Int("block_count", len(pollerBlocks)),
+			)
+			targetHeight = pollerBlocks[len(pollerBlocks)-1].Height
+			res, err := fp.retrySubmitFinalitySignatureUntilBlocksFinalized(pollerBlocks)
 			if err != nil {
 				fp.metrics.IncrementFpTotalFailedVotes(fp.GetBtcPkHex())
 				if !errors.Is(err, ErrFinalityProviderShutDown) {
@@ -254,13 +283,13 @@ func (fp *FinalityProviderInstance) finalitySigSubmissionLoop() {
 				continue
 			}
 			fp.logger.Info(
-				"successfully submitted a finality signature to the consumer chain",
+				"successfully submitted the finality signature to the consumer chain",
 				zap.String("consumer_id", string(fp.GetChainID())),
 				zap.String("pk", fp.GetBtcPkHex()),
-				zap.Uint64("height", b.Height),
+				zap.Uint64("start_height", pollerBlocks[0].Height),
+				zap.Uint64("end_height", targetHeight),
 				zap.String("tx_hash", res.TxHash),
 			)
-
 		case targetBlock := <-fp.laggingTargetChan:
 			res, err := fp.tryFastSync(targetBlock)
 			fp.isLagging.Store(false)
@@ -601,23 +630,23 @@ func (fp *FinalityProviderInstance) retryCheckRandomnessUntilBlockFinalized(targ
 	}
 }
 
-// retrySubmitFinalitySignatureUntilBlockFinalized periodically tries to submit finality signature until success or the block is finalized
+// retrySubmitFinalitySignatureUntilBlocksFinalized periodically tries to submit finality signature until success or the block is finalized
 // error will be returned if maximum retries have been reached or the query to the consumer chain fails
-func (fp *FinalityProviderInstance) retrySubmitFinalitySignatureUntilBlockFinalized(targetBlock *types.BlockInfo) (*types.TxResponse, error) {
+func (fp *FinalityProviderInstance) retrySubmitFinalitySignatureUntilBlocksFinalized(targetBlocks []*types.BlockInfo) (*types.TxResponse, error) {
 	var failedCycles uint32
-
+	targetHeight := targetBlocks[len(targetBlocks)-1].Height
 	// we break the for loop if the block is finalized or the signature is successfully submitted
 	// error will be returned if maximum retries have been reached or the query to the consumer chain fails
 	for {
 		// error will be returned if max retries have been reached
-		res, err := fp.SubmitFinalitySignature(targetBlock)
+		res, err := fp.SubmitBatchFinalitySignatures(targetBlocks)
 		if err != nil {
-
 			fp.logger.Debug(
 				"failed to submit finality signature to the consumer chain",
 				zap.String("pk", fp.GetBtcPkHex()),
 				zap.Uint32("current_failures", failedCycles),
-				zap.Uint64("target_block_height", targetBlock.Height),
+				zap.Uint64("target_start_height", targetBlocks[0].Height),
+				zap.Uint64("target_end_height", targetHeight),
 				zap.Error(err),
 			)
 
@@ -640,15 +669,15 @@ func (fp *FinalityProviderInstance) retrySubmitFinalitySignatureUntilBlockFinali
 		select {
 		case <-time.After(fp.cfg.SubmissionRetryInterval):
 			// periodically query the index block to be later checked whether it is Finalized
-			finalized, err := fp.consumerCon.QueryIsBlockFinalized(targetBlock.Height)
+			finalized, err := fp.consumerCon.QueryIsBlockFinalized(targetHeight)
 			if err != nil {
-				return nil, fmt.Errorf("failed to query block finalization at height %v: %w", targetBlock.Height, err)
+				return nil, fmt.Errorf("failed to query block finalization at height %v: %w", targetHeight, err)
 			}
 			if finalized {
 				fp.logger.Debug(
 					"the block is already finalized, skip submission",
 					zap.String("pk", fp.GetBtcPkHex()),
-					zap.Uint64("target_height", targetBlock.Height),
+					zap.Uint64("target_height", targetHeight),
 				)
 				// TODO: returning nil here is to safely break the loop
 				//  the error still exists

itest/opstackl2/op_test_manager.go

@@ -361,6 +361,8 @@ func createBaseFpConfig(fpHomeDir string, index int, logger *zap.Logger) *fpcfg.
 	cfg.RandomnessCommitInterval = 2 * time.Second
 	cfg.NumPubRand = 64
 	cfg.MinRandHeightGap = 1000
+	cfg.FastSyncGap = 60
+	cfg.FastSyncLimit = 100
 	return cfg
 }