Merge branch 'master' into leo/db-ops-save-execution-results

Author: zhangchiqing

Diff for: admin/commands/storage/backfill_tx_error_messages.go

@@ -7,7 +7,6 @@ import (
 	"github.com/onflow/flow-go/admin"
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/engine/access/ingestion/tx_error_messages"
-	commonrpc "github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/state/protocol"
@@ -177,7 +176,7 @@ func (b *BackfillTxErrorMessagesCommand) parseExecutionNodeIds(executionNodeIdsI
 		if len(executionNodeIds) == 0 {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", "must be a non empty list of strings", executionNodeIdsIn)
 		}
-		requestedENIdentifiers, err := commonrpc.IdentifierList(executionNodeIds)
+		requestedENIdentifiers, err := flow.IdentifierListFromHex(executionNodeIds)
 		if err != nil {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", err.Error(), executionNodeIdsIn)
 		}

Diff for: cmd/access/node_builder/access_node_builder.go

@@ -232,9 +232,10 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 				IdleTimeout:    rest.DefaultIdleTimeout,
 				MaxRequestSize: commonrest.DefaultMaxRequestSize,
 			},
-			MaxMsgSize:      grpcutils.DefaultMaxMsgSize,
-			CompressorName:  grpcutils.NoCompressor,
-			WebSocketConfig: websockets.NewDefaultWebsocketConfig(),
+			MaxMsgSize:                grpcutils.DefaultMaxMsgSize,
+			CompressorName:            grpcutils.NoCompressor,
+			WebSocketConfig:           websockets.NewDefaultWebsocketConfig(),
+			EnableWebSocketsStreamAPI: false,
 		},
 		stateStreamConf: statestreambackend.Config{
 			MaxExecutionDataMsgSize: grpcutils.DefaultMaxMsgSize,
@@ -1475,6 +1476,13 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
 			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
+
+		flags.BoolVar(
+			&builder.rpcConf.EnableWebSocketsStreamAPI,
+			"experimental-enable-websockets-stream-api",
+			defaultConfig.rpcConf.EnableWebSocketsStreamAPI,
+			"[experimental] enables WebSockets Stream API that operates under /ws endpoint. this flag may change in a future release.",
+		)
 	}).ValidateFlags(func() error {
 		if builder.supportsObserver && (builder.PublicNetworkConfig.BindAddress == cmd.NotSet || builder.PublicNetworkConfig.BindAddress == "") {
 			return errors.New("public-network-address must be set if supports-observer is true")
@@ -2007,12 +2015,12 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			}
 
-			preferredENIdentifiers, err := commonrpc.IdentifierList(backendConfig.PreferredExecutionNodeIDs)
+			preferredENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.PreferredExecutionNodeIDs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for preferred EN map: %w", err)
 			}
 
-			fixedENIdentifiers, err := commonrpc.IdentifierList(backendConfig.FixedExecutionNodeIDs)
+			fixedENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.FixedExecutionNodeIDs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for fixed EN map: %w", err)
 			}

Diff for: cmd/bootstrap/cmd/block.go

@@ -42,7 +42,10 @@ func constructRootEpochEvents(
 	participants flow.IdentityList,
 	assignments flow.AssignmentList,
 	clusterQCs []*flow.QuorumCertificate,
-	dkgData dkg.DKGData) (*flow.EpochSetup, *flow.EpochCommit) {
+	dkgData dkg.ThresholdKeySet,
+	dkgIndexMap flow.DKGIndexMap,
+) (*flow.EpochSetup, *flow.EpochCommit) {
+
 	epochSetup := &flow.EpochSetup{
 		Counter:            flagEpochCounter,
 		FirstView:          firstView,
@@ -77,6 +80,7 @@ func constructRootEpochEvents(
 		ClusterQCs:         flow.ClusterQCVoteDatasFromQCs(qcsWithSignerIDs),
 		DKGGroupKey:        dkgData.PubGroupKey,
 		DKGParticipantKeys: dkgData.PubKeyShares,
+		DKGIndexMap:        dkgIndexMap,
 	}
 	return epochSetup, epochCommit
 }

Diff for: cmd/bootstrap/cmd/dkg.go

@@ -10,34 +10,31 @@ import (
 	model "github.com/onflow/flow-go/model/bootstrap"
 	"github.com/onflow/flow-go/model/dkg"
 	"github.com/onflow/flow-go/model/encodable"
+	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/state/protocol/inmem"
 )
 
-func runBeaconKG(nodes []model.NodeInfo) dkg.DKGData {
+func runBeaconKG(nodes []model.NodeInfo) (dkg.ThresholdKeySet, flow.DKGIndexMap) {
 	n := len(nodes)
-
 	log.Info().Msgf("read %v node infos for DKG", n)
 
 	log.Debug().Msgf("will run DKG")
-	var dkgData dkg.DKGData
-	var err error
-	dkgData, err = bootstrapDKG.RandomBeaconKG(n, GenerateRandomSeed(crypto.KeyGenSeedMinLen))
+	randomBeaconData, err := bootstrapDKG.RandomBeaconKG(n, GenerateRandomSeed(crypto.KeyGenSeedMinLen))
 	if err != nil {
 		log.Fatal().Err(err).Msg("error running DKG")
 	}
 	log.Info().Msgf("finished running DKG")
 
-	pubKeyShares := make([]encodable.RandomBeaconPubKey, 0, len(dkgData.PubKeyShares))
-	for _, pubKey := range dkgData.PubKeyShares {
-		pubKeyShares = append(pubKeyShares, encodable.RandomBeaconPubKey{PublicKey: pubKey})
-	}
-
-	privKeyShares := make([]encodable.RandomBeaconPrivKey, 0, len(dkgData.PrivKeyShares))
-	for i, privKey := range dkgData.PrivKeyShares {
+	encodableParticipants := make([]inmem.ThresholdParticipant, 0, len(nodes))
+	for i, privKey := range randomBeaconData.PrivKeyShares {
 		nodeID := nodes[i].NodeID
 
 		encKey := encodable.RandomBeaconPrivKey{PrivateKey: privKey}
-		privKeyShares = append(privKeyShares, encKey)
+		encodableParticipants = append(encodableParticipants, inmem.ThresholdParticipant{
+			PrivKeyShare: encKey,
+			PubKeyShare:  encodable.RandomBeaconPubKey{PublicKey: randomBeaconData.PubKeyShares[i]},
+			NodeID:       nodeID,
+		})
 
 		err = common.WriteJSON(fmt.Sprintf(model.PathRandomBeaconPriv, nodeID), flagOutdir, encKey)
 		if err != nil {
@@ -46,18 +43,22 @@ func runBeaconKG(nodes []model.NodeInfo) dkg.DKGData {
 		log.Info().Msgf("wrote file %s/%s", flagOutdir, fmt.Sprintf(model.PathRandomBeaconPriv, nodeID))
 	}
 
+	indexMap := make(flow.DKGIndexMap, len(nodes))
+	for i, node := range nodes {
+		indexMap[node.NodeID] = i
+	}
+
 	// write full DKG info that will be used to construct QC
-	err = common.WriteJSON(model.PathRootDKGData, flagOutdir, inmem.EncodableFullDKG{
+	err = common.WriteJSON(model.PathRootDKGData, flagOutdir, inmem.ThresholdKeySet{
 		GroupKey: encodable.RandomBeaconPubKey{
-			PublicKey: dkgData.PubGroupKey,
+			PublicKey: randomBeaconData.PubGroupKey,
 		},
-		PubKeyShares:  pubKeyShares,
-		PrivKeyShares: privKeyShares,
+		Participants: encodableParticipants,
 	})
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to write json")
 	}
 	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathRootDKGData)
 
-	return dkgData
+	return randomBeaconData, indexMap
 }

Diff for: cmd/bootstrap/cmd/finalize.go

@@ -152,7 +152,7 @@ func finalize(cmd *cobra.Command, args []string) {
 	log.Info().Msgf("received votes total: %v", len(votes))
 
 	log.Info().Msg("reading dkg data")
-	dkgData := readDKGData()
+	dkgData, _ := readRandomBeaconKeys()
 	log.Info().Msg("")
 
 	log.Info().Msg("reading intermediary bootstrapping data")
@@ -192,7 +192,7 @@ func finalize(cmd *cobra.Command, args []string) {
 		intermediaryData.ProtocolVersion,
 		func(epochStateID flow.Identifier) (protocol_state.KVStoreAPI, error) {
 			return kvstore.NewDefaultKVStore(
-				intermediaryData.EpochCommitSafetyThreshold,
+				intermediaryData.FinalizationSafetyThreshold,
 				intermediaryData.EpochExtensionViewCount,
 				epochStateID)
 		},
@@ -352,29 +352,28 @@ func readRootBlock() *flow.Block {
 	return rootBlock
 }
 
-// readDKGData reads DKG data from disc, this file needs to be prepared with
-// rootblock command
-func readDKGData() dkg.DKGData {
-	encodableDKG, err := utils.ReadData[inmem.EncodableFullDKG](flagDKGDataPath)
+// readRandomBeaconKeys reads the threshold key data from disc.
+// This file needs to be prepared with rootblock command
+func readRandomBeaconKeys() (dkg.ThresholdKeySet, flow.DKGIndexMap) {
+	encodableDKG, err := utils.ReadData[inmem.ThresholdKeySet](flagDKGDataPath)
 	if err != nil {
-		log.Fatal().Err(err).Msg("could not read DKG data")
+		log.Fatal().Err(err).Msg("loading threshold key data for Random Beacon failed")
 	}
 
-	dkgData := dkg.DKGData{
+	dkgData := dkg.ThresholdKeySet{
 		PrivKeyShares: nil,
 		PubGroupKey:   encodableDKG.GroupKey.PublicKey,
 		PubKeyShares:  nil,
 	}
 
-	for _, pubKey := range encodableDKG.PubKeyShares {
-		dkgData.PubKeyShares = append(dkgData.PubKeyShares, pubKey.PublicKey)
-	}
-
-	for _, privKey := range encodableDKG.PrivKeyShares {
-		dkgData.PrivKeyShares = append(dkgData.PrivKeyShares, privKey.PrivateKey)
+	indexMap := make(flow.DKGIndexMap, len(encodableDKG.Participants))
+	for i, participant := range encodableDKG.Participants {
+		dkgData.PubKeyShares = append(dkgData.PubKeyShares, participant.PubKeyShare.PublicKey)
+		dkgData.PrivKeyShares = append(dkgData.PrivKeyShares, participant.PrivKeyShare.PrivateKey)
+		indexMap[participant.NodeID] = i
 	}
 
-	return dkgData
+	return dkgData, indexMap
 }
 
 // Validation utility methods ------------------------------------------------

Diff for: cmd/bootstrap/cmd/finalize_test.go

@@ -74,7 +74,7 @@ func TestFinalize_HappyPath(t *testing.T) {
 		flagNumViewsInEpoch = 100_000
 		flagNumViewsInStakingAuction = 50_000
 		flagNumViewsInDKGPhase = 2_000
-		flagEpochCommitSafetyThreshold = 1_000
+		flagFinalizationSafetyThreshold = 1_000
 		flagEpochExtensionViewCount = 100_000
 		flagUseDefaultEpochTargetEndTime = true
 		flagEpochTimingRefCounter = 0

Diff for: cmd/bootstrap/cmd/intermediary.go

@@ -18,9 +18,9 @@ type IntermediaryBootstrappingData struct {
 // like the root block).
 // This is used to pass data between the rootblock command and the finalize command.
 type IntermediaryParamsData struct {
-	ProtocolVersion            uint
-	EpochCommitSafetyThreshold uint64
-	EpochExtensionViewCount    uint64
+	ProtocolVersion             uint
+	FinalizationSafetyThreshold uint64
+	EpochExtensionViewCount     uint64
 }
 
 // IntermediaryEpochData stores the root epoch and the epoch config for the execution state

Diff for: cmd/bootstrap/cmd/qc.go

@@ -13,10 +13,10 @@ import (
 )
 
 // constructRootQC constructs root QC based on root block, votes and dkg info
-func constructRootQC(block *flow.Block, votes []*model.Vote, allNodes, internalNodes []bootstrap.NodeInfo, dkgData dkg.DKGData) *flow.QuorumCertificate {
+func constructRootQC(block *flow.Block, votes []*model.Vote, allNodes, internalNodes []bootstrap.NodeInfo, randomBeaconData dkg.ThresholdKeySet) *flow.QuorumCertificate {
 
 	identities := bootstrap.ToIdentityList(allNodes)
-	participantData, err := run.GenerateQCParticipantData(allNodes, internalNodes, dkgData)
+	participantData, err := run.GenerateQCParticipantData(allNodes, internalNodes, randomBeaconData)
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to generate QC participant data")
 	}
@@ -36,8 +36,8 @@ func constructRootQC(block *flow.Block, votes []*model.Vote, allNodes, internalN
 }
 
 // NOTE: allNodes must be in the same order as when generating the DKG
-func constructRootVotes(block *flow.Block, allNodes, internalNodes []bootstrap.NodeInfo, dkgData dkg.DKGData) {
-	participantData, err := run.GenerateQCParticipantData(allNodes, internalNodes, dkgData)
+func constructRootVotes(block *flow.Block, allNodes, internalNodes []bootstrap.NodeInfo, randomBeaconData dkg.ThresholdKeySet) {
+	participantData, err := run.GenerateQCParticipantData(allNodes, internalNodes, randomBeaconData)
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to generate QC participant data")
 	}

Diff for: cmd/bootstrap/cmd/rootblock.go

@@ -23,18 +23,18 @@ import (
 )
 
 var (
-	flagRootChain                  string
-	flagRootParent                 string
-	flagRootHeight                 uint64
-	flagRootTimestamp              string
-	flagProtocolVersion            uint
-	flagEpochCommitSafetyThreshold uint64
-	flagEpochExtensionViewCount    uint64
-	flagCollectionClusters         uint
-	flagEpochCounter               uint64
-	flagNumViewsInEpoch            uint64
-	flagNumViewsInStakingAuction   uint64
-	flagNumViewsInDKGPhase         uint64
+	flagRootChain                   string
+	flagRootParent                  string
+	flagRootHeight                  uint64
+	flagRootTimestamp               string
+	flagProtocolVersion             uint
+	flagFinalizationSafetyThreshold uint64
+	flagEpochExtensionViewCount     uint64
+	flagCollectionClusters          uint
+	flagEpochCounter                uint64
+	flagNumViewsInEpoch             uint64
+	flagNumViewsInStakingAuction    uint64
+	flagNumViewsInDKGPhase          uint64
 	// Epoch target end time config
 	flagUseDefaultEpochTargetEndTime bool
 	flagEpochTimingRefCounter        uint64
@@ -93,14 +93,14 @@ func addRootBlockCmdFlags() {
 	rootBlockCmd.Flags().Uint64Var(&flagRootHeight, "root-height", 0, "height of the root block")
 	rootBlockCmd.Flags().StringVar(&flagRootTimestamp, "root-timestamp", time.Now().UTC().Format(time.RFC3339), "timestamp of the root block (RFC3339)")
 	rootBlockCmd.Flags().UintVar(&flagProtocolVersion, "protocol-version", flow.DefaultProtocolVersion, "major software version used for the duration of this spork")
-	rootBlockCmd.Flags().Uint64Var(&flagEpochCommitSafetyThreshold, "epoch-commit-safety-threshold", 500, "defines epoch commitment deadline")
+	rootBlockCmd.Flags().Uint64Var(&flagFinalizationSafetyThreshold, "finalization-safety-threshold", 500, "defines finalization safety threshold")
 	rootBlockCmd.Flags().Uint64Var(&flagEpochExtensionViewCount, "epoch-extension-view-count", 100_000, "length of epoch extension in views, default is 100_000 which is approximately 1 day")
 
 	cmd.MarkFlagRequired(rootBlockCmd, "root-chain")
 	cmd.MarkFlagRequired(rootBlockCmd, "root-parent")
 	cmd.MarkFlagRequired(rootBlockCmd, "root-height")
 	cmd.MarkFlagRequired(rootBlockCmd, "protocol-version")
-	cmd.MarkFlagRequired(rootBlockCmd, "epoch-commit-safety-threshold")
+	cmd.MarkFlagRequired(rootBlockCmd, "finalization-safety-threshold")
 	cmd.MarkFlagRequired(rootBlockCmd, "epoch-extension-view-count")
 
 	// Epoch timing config - these values must be set identically to `EpochTimingConfig` in the FlowEpoch smart contract.
@@ -138,7 +138,7 @@ func rootBlock(cmd *cobra.Command, args []string) {
 	// validate epoch configs
 	err := validateEpochConfig()
 	if err != nil {
-		log.Fatal().Err(err).Msg("invalid or unsafe epoch commit threshold config")
+		log.Fatal().Err(err).Msg("invalid or unsafe config for finalization safety threshold")
 	}
 	err = validateOrPopulateEpochTimingConfig()
 	if err != nil {
@@ -190,7 +190,7 @@ func rootBlock(cmd *cobra.Command, args []string) {
 	log.Info().Msg("")
 
 	log.Info().Msg("running DKG for consensus nodes")
-	dkgData := runBeaconKG(model.FilterByRole(stakingNodes, flow.RoleConsensus))
+	randomBeaconData, dkgIndexMap := runBeaconKG(model.FilterByRole(stakingNodes, flow.RoleConsensus))
 	log.Info().Msg("")
 
 	// create flow.IdentityList representation of the participant set
@@ -216,17 +216,17 @@ func rootBlock(cmd *cobra.Command, args []string) {
 	log.Info().Msg("")
 
 	log.Info().Msg("constructing intermediary bootstrapping data")
-	epochSetup, epochCommit := constructRootEpochEvents(header.View, participants, assignments, clusterQCs, dkgData)
-	epochConfig := generateExecutionStateEpochConfig(epochSetup, clusterQCs, dkgData)
+	epochSetup, epochCommit := constructRootEpochEvents(header.View, participants, assignments, clusterQCs, randomBeaconData, dkgIndexMap)
+	epochConfig := generateExecutionStateEpochConfig(epochSetup, clusterQCs, randomBeaconData)
 	intermediaryEpochData := IntermediaryEpochData{
 		RootEpochSetup:       epochSetup,
 		RootEpochCommit:      epochCommit,
 		ExecutionStateConfig: epochConfig,
 	}
 	intermediaryParamsData := IntermediaryParamsData{
-		EpochCommitSafetyThreshold: flagEpochCommitSafetyThreshold,
-		EpochExtensionViewCount:    flagEpochExtensionViewCount,
-		ProtocolVersion:            flagProtocolVersion,
+		FinalizationSafetyThreshold: flagFinalizationSafetyThreshold,
+		EpochExtensionViewCount:     flagEpochExtensionViewCount,
+		ProtocolVersion:             flagProtocolVersion,
 	}
 	intermediaryData := IntermediaryBootstrappingData{
 		IntermediaryEpochData:  intermediaryEpochData,
@@ -241,7 +241,7 @@ func rootBlock(cmd *cobra.Command, args []string) {
 
 	log.Info().Msg("constructing root block")
 	rootProtocolState, err := kvstore.NewDefaultKVStore(
-		flagEpochCommitSafetyThreshold,
+		flagFinalizationSafetyThreshold,
 		flagEpochExtensionViewCount,
 		inmem.EpochProtocolStateFromServiceEvents(epochSetup, epochCommit).ID(),
 	)
@@ -261,7 +261,7 @@ func rootBlock(cmd *cobra.Command, args []string) {
 		block,
 		model.FilterByRole(stakingNodes, flow.RoleConsensus),
 		model.FilterByRole(internalNodes, flow.RoleConsensus),
-		dkgData,
+		randomBeaconData,
 	)
 	log.Info().Msg("")
 }
@@ -270,16 +270,16 @@ func rootBlock(cmd *cobra.Command, args []string) {
 func validateEpochConfig() error {
 	chainID := parseChainID(flagRootChain)
 	dkgFinalView := flagNumViewsInStakingAuction + flagNumViewsInDKGPhase*3 // 3 DKG phases
-	epochCommitDeadline := flagNumViewsInEpoch - flagEpochCommitSafetyThreshold
+	epochCommitDeadline := flagNumViewsInEpoch - flagFinalizationSafetyThreshold
 
 	defaultEpochSafetyParams, err := protocol.DefaultEpochSafetyParams(chainID)
 	if err != nil {
 		return fmt.Errorf("could not get default epoch commit safety threshold: %w", err)
 	}
 
 	// sanity check: the safety threshold is >= the default for the chain
-	if flagEpochCommitSafetyThreshold < defaultEpochSafetyParams.FinalizationSafetyThreshold {
-		return fmt.Errorf("potentially unsafe epoch config: epoch commit safety threshold smaller than expected (%d < %d)", flagEpochCommitSafetyThreshold, defaultEpochSafetyParams.FinalizationSafetyThreshold)
+	if flagFinalizationSafetyThreshold < defaultEpochSafetyParams.FinalizationSafetyThreshold {
+		return fmt.Errorf("potentially unsafe epoch config: epoch commit safety threshold smaller than expected (%d < %d)", flagFinalizationSafetyThreshold, defaultEpochSafetyParams.FinalizationSafetyThreshold)
 	}
 	if flagEpochExtensionViewCount < defaultEpochSafetyParams.EpochExtensionViewCount {
 		return fmt.Errorf("potentially unsafe epoch config: epoch extension view count smaller than expected (%d < %d)", flagEpochExtensionViewCount, defaultEpochSafetyParams.EpochExtensionViewCount)
@@ -302,7 +302,7 @@ func validateEpochConfig() error {
 func generateExecutionStateEpochConfig(
 	epochSetup *flow.EpochSetup,
 	clusterQCs []*flow.QuorumCertificate,
-	dkgData dkg.DKGData,
+	dkgData dkg.ThresholdKeySet,
 ) epochs.EpochConfig {
 
 	randomSource := make([]byte, flow.EpochSetupRandomSourceLength)