Merge branch 'master' into leo/refactor-execution-fork-evidence

Author: zhangchiqing

admin/commands/storage/backfill_tx_error_messages.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/rs/zerolog"
+
 	"github.com/onflow/flow-go/admin"
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/engine/access/ingestion/tx_error_messages"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
+	"github.com/onflow/flow-go/module/util"
 	"github.com/onflow/flow-go/state/protocol"
 )
 
@@ -25,16 +28,19 @@ type backfillTxErrorMessagesRequest struct {
 // BackfillTxErrorMessagesCommand executes a command to backfill
 // transaction error messages by fetching them from execution nodes.
 type BackfillTxErrorMessagesCommand struct {
+	log                 zerolog.Logger
 	state               protocol.State
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore
 }
 
 // NewBackfillTxErrorMessagesCommand creates a new instance of BackfillTxErrorMessagesCommand
 func NewBackfillTxErrorMessagesCommand(
+	log zerolog.Logger,
 	state protocol.State,
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore,
 ) commands.AdminCommand {
 	return &BackfillTxErrorMessagesCommand{
+		log:                 log.With().Str("command", "backfill-tx-error-messages").Logger(),
 		state:               state,
 		txErrorMessagesCore: txErrorMessagesCore,
 	}
@@ -147,6 +153,17 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 
 	data := request.ValidatorData.(*backfillTxErrorMessagesRequest)
 
+	total := data.endHeight - data.startHeight + 1
+	progress := util.LogProgress(b.log,
+		util.DefaultLogProgressConfig("backfilling", int(total)),
+	)
+
+	b.log.Info().
+		Uint64("start_height", data.startHeight).
+		Uint64("end_height", data.endHeight).
+		Uint64("blocks", total).
+		Msgf("starting backfill")
+
 	for height := data.startHeight; height <= data.endHeight; height++ {
 		header, err := b.state.AtHeight(height).Head()
 		if err != nil {
@@ -158,6 +175,8 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 		if err != nil {
 			return nil, fmt.Errorf("error encountered while processing transaction result error message for block: %d, %w", height, err)
 		}
+
+		progress(1)
 	}
 
 	return nil, nil
@@ -170,13 +189,18 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 // - admin.InvalidAdminReqParameterError - if execution-node-ids is empty or has an invalid format.
 func (b *BackfillTxErrorMessagesCommand) parseExecutionNodeIds(executionNodeIdsIn interface{}, allIdentities flow.IdentityList) (flow.IdentitySkeletonList, error) {
 	var ids flow.IdentityList
-
 	switch executionNodeIds := executionNodeIdsIn.(type) {
-	case []string:
+	case []any:
 		if len(executionNodeIds) == 0 {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", "must be a non empty list of strings", executionNodeIdsIn)
 		}
-		requestedENIdentifiers, err := flow.IdentifierListFromHex(executionNodeIds)
+
+		idStrings := make([]string, len(executionNodeIds))
+		for i, id := range executionNodeIds {
+			idStrings[i] = id.(string)
+		}
+
+		requestedENIdentifiers, err := flow.IdentifierListFromHex(idStrings)
 		if err != nil {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", err.Error(), executionNodeIdsIn)
 		}

admin/commands/storage/backfill_tx_error_messages_test.go

@@ -163,6 +163,7 @@ func (suite *BackfillTxErrorMessagesSuite) SetupTest() {
 	)
 
 	suite.command = NewBackfillTxErrorMessagesCommand(
+		suite.log,
 		suite.state,
 		suite.txResultErrorMessagesCore,
 	)
@@ -237,7 +238,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateInvalidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1),         // raw json parses to float64
 				"end-height":         float64(endHeight), // raw json parses to float64
-				"execution-node-ids": []string{suite.allENIDs[0].NodeID.String()},
+				"execution-node-ids": []any{suite.allENIDs[0].NodeID.String()},
 			},
 		})
 		suite.Error(err)
@@ -290,7 +291,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateInvalidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1), // raw json parses to float64
 				"end-height":         float64(4), // raw json parses to float64
-				"execution-node-ids": []string{invalidENID.String()},
+				"execution-node-ids": []any{invalidENID.String()},
 			},
 		})
 		suite.Error(err)
@@ -321,7 +322,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateValidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1), // raw json parses to float64
 				"end-height":         float64(3), // raw json parses to float64
-				"execution-node-ids": []string{suite.allENIDs[0].NodeID.String()},
+				"execution-node-ids": []any{suite.allENIDs[0].NodeID.String()},
 			},
 		})
 		suite.NoError(err)
@@ -389,7 +390,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestHandleBackfillTxErrorMessages() {
 			Data: map[string]interface{}{
 				"start-height":       float64(startHeight), // raw json parses to float64
 				"end-height":         float64(endHeight),   // raw json parses to float64
-				"execution-node-ids": []string{executorID.String()},
+				"execution-node-ids": []any{executorID.String()},
 			},
 		}
 		suite.Require().NoError(suite.command.Validator(req))
@@ -432,6 +433,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestHandleBackfillTxErrorMessagesErro
 	suite.Run("error when txErrorMessagesCore is nil", func() {
 		req := &admin.CommandRequest{Data: map[string]interface{}{}}
 		command := NewBackfillTxErrorMessagesCommand(
+			suite.log,
 			suite.state,
 			nil,
 		)

cmd/access/node_builder/access_node_builder.go

@@ -1601,6 +1601,11 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			return errors.New("rpc-max-execution-response-message-size must be greater than 0")
 		}
 
+		// indexing tx error messages is only supported when tx results are also indexed
+		if builder.storeTxResultErrorMessages && !builder.executionDataIndexingEnabled {
+			return errors.New("execution-data-indexing-enabled must be set if store-tx-result-error-messages is enabled")
+		}
+
 		return nil
 	})
 }
@@ -1956,13 +1961,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			return nil
 		}).
-		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
-			if builder.storeTxResultErrorMessages {
-				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
-			}
-
-			return nil
-		}).
 		Component("version control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			if !builder.versionControlEnabled {
 				noop := &module.NoopReadyDoneAware{}
@@ -2262,34 +2260,45 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		}).
 		AdminCommand("backfill-tx-error-messages", func(config *cmd.NodeConfig) commands.AdminCommand {
 			return storageCommands.NewBackfillTxErrorMessagesCommand(
+				builder.Logger,
 				builder.State,
 				builder.TxResultErrorMessagesCore,
 			)
 		})
 
 	if builder.storeTxResultErrorMessages {
-		builder.Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
-			processedTxErrorMessagesBlockHeight = store.NewConsumerProgress(
-				builder.ProtocolDB,
-				module.ConsumeProgressEngineTxErrorMessagesBlockHeight,
-			)
-			return nil
-		})
-		builder.Component("transaction result error messages engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			engine, err := tx_error_messages.New(
-				node.Logger,
-				node.State,
-				node.Storage.Headers,
-				processedTxErrorMessagesBlockHeight,
-				builder.TxResultErrorMessagesCore,
-			)
-			if err != nil {
-				return nil, err
-			}
-			builder.FollowerDistributor.AddOnBlockFinalizedConsumer(engine.OnFinalizedBlock)
+		builder.
+			Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
+				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(
+					node.Metrics.Cache,
+					node.ProtocolDB,
+					bstorage.DefaultCacheSize,
+				)
+				return nil
+			}).
+			Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
+				processedTxErrorMessagesBlockHeight = store.NewConsumerProgress(
+					builder.ProtocolDB,
+					module.ConsumeProgressEngineTxErrorMessagesBlockHeight,
+				)
+				return nil
+			}).
+			Component("transaction result error messages engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+				engine, err := tx_error_messages.New(
+					node.Logger,
+					metrics.NewTransactionErrorMessagesCollector(),
+					node.State,
+					node.Storage.Headers,
+					processedTxErrorMessagesBlockHeight,
+					builder.TxResultErrorMessagesCore,
+				)
+				if err != nil {
+					return nil, err
+				}
+				builder.FollowerDistributor.AddOnBlockFinalizedConsumer(engine.OnFinalizedBlock)
 
-			return engine, nil
-		})
+				return engine, nil
+			})
 	}
 
 	if builder.supportsObserver {

cmd/collection/main.go

@@ -636,6 +636,7 @@ func main() {
 				node.Logger,
 				node.Me,
 				node.ProtocolDB,
+				node.StorageLockMgr,
 				node.State,
 				node.Metrics.Engine,
 				node.Metrics.Mempool,

cmd/consensus/main.go

@@ -622,7 +622,7 @@ func main() {
 			notifier.AddFollowerConsumer(followerDistributor)
 
 			// initialize the persister
-			persist, err := persister.New(node.ProtocolDB, node.RootChainID)
+			persist, err := persister.New(node.ProtocolDB, node.RootChainID, node.StorageLockMgr)
 			if err != nil {
 				return nil, err
 			}

cmd/scaffold.go

@@ -303,7 +303,7 @@ func (fnb *FlowNodeBuilder) EnqueuePingService() {
 		var hotstuffViewFunc func() (uint64, error)
 		// Setup consensus nodes to report their HotStuff view
 		if fnb.BaseConfig.NodeRole == flow.RoleConsensus.String() {
-			hotstuffReader, err := persister.NewReader(node.ProtocolDB, node.RootChainID)
+			hotstuffReader, err := persister.NewReader(node.ProtocolDB, node.RootChainID, node.StorageLockMgr)
 			if err != nil {
 				return nil, err
 			}

cmd/util/cmd/read-hotstuff/cmd/get_liveness.go

@@ -22,8 +22,10 @@ func init() {
 
 func runGetLivenessData(*cobra.Command, []string) {
 	err := common.WithStorage(flagDatadir, func(db storage.DB) error {
+		lockManager := storage.NewTestingLockManager()
+
 		chainID := flow.ChainID(flagChain)
-		reader, err := persister.NewReader(db, chainID)
+		reader, err := persister.NewReader(db, chainID, lockManager)
 		if err != nil {
 			log.Fatal().Err(err).Msg("could not create reader from db")
 		}

cmd/util/cmd/read-hotstuff/cmd/get_safety.go

@@ -22,9 +22,10 @@ func init() {
 
 func runGetSafetyData(*cobra.Command, []string) {
 	err := common.WithStorage(flagDatadir, func(db storage.DB) error {
+		lockManager := storage.NewTestingLockManager()
 
 		chainID := flow.ChainID(flagChain)
-		reader, err := persister.NewReader(db, chainID)
+		reader, err := persister.NewReader(db, chainID, lockManager)
 		if err != nil {
 			log.Fatal().Err(err).Msg("could not create reader from db")
 		}

cmd/util/cmd/read-light-block/read_light_block_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/operation"
 	"github.com/onflow/flow-go/storage/operation/dbtest"
-	"github.com/onflow/flow-go/storage/procedure"
 	"github.com/onflow/flow-go/storage/store"
 	"github.com/onflow/flow-go/utils/unittest"
 )
@@ -41,14 +40,14 @@ func TestReadClusterRange(t *testing.T) {
 		for _, block := range blocks {
 			err = unittest.WithLock(t, lockManager, storage.LockInsertOrFinalizeClusterBlock, func(lctx lockctx.Context) error {
 				err := db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-					return procedure.InsertClusterBlock(lctx, rw, unittest.ClusterProposalFromBlock(block))
+					return operation.InsertClusterBlock(lctx, rw, unittest.ClusterProposalFromBlock(block))
 				})
 				if err != nil {
 					return err
 				}
 
 				return db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-					return procedure.FinalizeClusterBlock(lctx, rw, block.ID())
+					return operation.FinalizeClusterBlock(lctx, rw, block.ID())
 				})
 			})
 			require.NoError(t, err)

cmd/util/cmd/verify_execution_result/cmd.go

@@ -82,6 +82,12 @@ func run(*cobra.Command, []string) {
 		Bool("stop_on_mismatch", flagStopOnMismatch).
 		Logger()
 
+	// Log configuration before starting verification so users can cancel and restart with different values if needed
+	if !flagStopOnMismatch {
+		lg.Info().Msgf("note flag --stop_on_mismatch is false, so mismatches (if any) are logged but do not stop the verification")
+		lg.Info().Msgf("look for 'could not verify' in the log for any mismatch, or try again with --stop_on_mismatch true to stop on first mismatch")
+	}
+
 	if flagFromTo != "" {
 		from, to, err := parseFromTo(flagFromTo)
 		if err != nil {
@@ -93,16 +99,15 @@ func run(*cobra.Command, []string) {
 		if err != nil {
 			lg.Fatal().Err(err).Msgf("could not verify range from %d to %d", from, to)
 		}
-		lg.Info().Msgf("successfully verified range from %d to %d", from, to)
-
+		lg.Info().Msgf("finished verified range from %d to %d", from, to)
 	} else {
 		lg.Info().Msgf("verifying last %d sealed blocks", flagLastK)
 		err := verifier.VerifyLastKHeight(lockManager, flagLastK, chainID, flagDatadir, flagChunkDataPackDir, flagWorkerCount, flagStopOnMismatch, flagtransactionFeesDisabled, flagScheduledCallbacksEnabled)
 		if err != nil {
 			lg.Fatal().Err(err).Msg("could not verify last k height")
 		}
 
-		lg.Info().Msgf("successfully verified last %d sealed blocks", flagLastK)
+		lg.Info().Msgf("finished verified last %d sealed blocks", flagLastK)
 	}
 }