Merge pull request #5767 from onflow/bastian/sync-stable-cadence-16

access/api.go

@@ -197,7 +197,7 @@ type API interface {
 	// SubscribeTransactionStatuses streams transaction statuses starting from the reference block saved in the
 	// transaction itself until the block containing the transaction becomes sealed or expired. When the transaction
 	// status becomes TransactionStatusSealed or TransactionStatusExpired, the subscription will automatically shut down.
-	SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody) subscription.Subscription
+	SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
 }
 
 // TODO: Combine this with flow.TransactionResult?

access/handler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/counters"
 
 	"github.com/onflow/flow/protobuf/go/flow/access"
 	"github.com/onflow/flow/protobuf/go/flow/entities"
@@ -1112,11 +1113,23 @@ func (h *Handler) SendAndSubscribeTransactionStatuses(
 		return err
 	}
 
-	sub := h.api.SubscribeTransactionStatuses(ctx, &tx)
-	return subscription.HandleSubscription(sub, func(txSubInfo *convert.TransactionSubscribeInfo) error {
-		err = stream.Send(convert.TransactionSubscribeInfoToMessage(txSubInfo))
-		if err != nil {
-			return rpc.ConvertError(err, "could not send response", codes.Internal)
+	sub := h.api.SubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion())
+
+	messageIndex := counters.NewMonotonousCounter(0)
+	return subscription.HandleSubscription(sub, func(txResults []*TransactionResult) error {
+		for i := range txResults {
+			value := messageIndex.Value()
+			if ok := messageIndex.Set(value + 1); !ok {
+				return status.Errorf(codes.Internal, "the message index has already been incremented to %d", messageIndex.Value())
+			}
+
+			err = stream.Send(&access.SendAndSubscribeTransactionStatusesResponse{
+				TransactionResults: TransactionResultToMessage(txResults[i]),
+				MessageIndex:       value,
+			})
+			if err != nil {
+				return rpc.ConvertError(err, "could not send response", codes.Internal)
+			}
 		}
 
 		return nil

admin/commands/storage/read_protocol_snapshot.go

@@ -57,7 +57,7 @@ func (s *ProtocolSnapshotCommand) Handler(_ context.Context, req *admin.CommandR
 
 	s.logger.Info().Uint("blocksToSkip", blocksToSkip).Msgf("admintool: generating protocol snapshot")
 
-	snapshot, sealedHeight, commit, err := common.GenerateProtocolSnapshotForCheckpoint(
+	snapshot, sealedHeight, commit, checkpointFile, err := common.GenerateProtocolSnapshotForCheckpoint(
 		s.logger, s.state, s.headers, s.seals, s.checkpointDir, blocksToSkip)
 	if err != nil {
 		return nil, fmt.Errorf("could not generate protocol snapshot for checkpoint, checkpointDir %v: %w",
@@ -79,10 +79,19 @@ func (s *ProtocolSnapshotCommand) Handler(_ context.Context, req *admin.CommandR
 		Hex("finalized_block_id", logging.Entity(header)).
 		Uint64("sealed_height", sealedHeight).
 		Hex("sealed_commit", commit[:]). // not the commit for the finalized height, but for the sealed height
+		Str("checkpoint_file", checkpointFile).
 		Uint("blocks_to_skip", blocksToSkip).
 		Msgf("admintool: protocol snapshot generated successfully")
 
-	return commands.ConvertToMap(serializable.Encodable())
+	return commands.ConvertToMap(protocolSnapshotResponse{
+		Snapshot:   serializable.Encodable(),
+		Checkpoint: checkpointFile,
+	})
+}
+
+type protocolSnapshotResponse struct {
+	Snapshot   inmem.EncodableSnapshot `json:"snapshot"`
+	Checkpoint string                  `json:"checkpoint"`
 }
 
 func (s *ProtocolSnapshotCommand) Validator(req *admin.CommandRequest) error {

cmd/access/node_builder/access_node_builder.go

@@ -1315,7 +1315,7 @@ func (builder *FlowAccessNodeBuilder) InitIDProviders() {
 				filter.And(
 					filter.HasRole[flow.Identity](flow.RoleConsensus),
 					filter.Not(filter.HasNodeID[flow.Identity](node.Me.NodeID())),
-					underlay.NotEjectedFilter,
+					filter.NotEjectedFilter,
 				),
 				builder.IdentityProvider,
 			)

cmd/bootstrap/cmd/check_machine_account.go

@@ -13,6 +13,7 @@ import (
 	sdk "github.com/onflow/flow-go-sdk"
 	client "github.com/onflow/flow-go-sdk/access/grpc"
 	"github.com/onflow/flow-go/cmd"
+	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	model "github.com/onflow/flow-go/model/bootstrap"
 	"github.com/onflow/flow-go/module/epochs"
 )
@@ -44,7 +45,10 @@ func checkMachineAccountRun(_ *cobra.Command, _ []string) {
 
 	// read the private node information - used to get the role
 	var nodeInfoPriv model.NodeInfoPriv
-	readJSON(filepath.Join(flagOutdir, fmt.Sprintf(model.PathNodeInfoPriv, nodeID)), &nodeInfoPriv)
+	err = common.ReadJSON(filepath.Join(flagOutdir, fmt.Sprintf(model.PathNodeInfoPriv, nodeID)), &nodeInfoPriv)
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to read json")
+	}
 
 	// read the machine account info file
 	machineAccountInfo := readMachineAccountInfo(nodeID)
@@ -97,7 +101,10 @@ func readMachineAccountInfo(nodeID string) model.NodeMachineAccountInfo {
 	var machineAccountInfo model.NodeMachineAccountInfo
 
 	path := filepath.Join(flagOutdir, fmt.Sprintf(model.PathNodeMachineAccountInfoPriv, nodeID))
-	readJSON(path, &machineAccountInfo)
+	err := common.ReadJSON(path, &machineAccountInfo)
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to read json")
+	}
 
 	return machineAccountInfo
 }

cmd/bootstrap/cmd/db_encryption_key.go

@@ -7,6 +7,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/onflow/flow-go/cmd/bootstrap/utils"
+	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	model "github.com/onflow/flow-go/model/bootstrap"
 )
 
@@ -35,7 +36,7 @@ func dbEncryptionKeyRun(_ *cobra.Command, _ []string) {
 	log = log.With().Str("path", dbEncryptionKeyPath).Logger()
 
 	// check if the key already exists
-	exists, err := pathExists(path.Join(flagOutdir, dbEncryptionKeyPath))
+	exists, err := common.PathExists(path.Join(flagOutdir, dbEncryptionKeyPath))
 	if err != nil {
 		log.Fatal().Err(err).Msg("could not check if db encryption key already exists")
 	}
@@ -50,5 +51,10 @@ func dbEncryptionKeyRun(_ *cobra.Command, _ []string) {
 	}
 	log.Info().Msg("generated db encryption key")
 
-	writeText(dbEncryptionKeyPath, dbEncryptionKey)
+	err = common.WriteText(dbEncryptionKeyPath, flagOutdir, dbEncryptionKey)
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to write file")
+	}
+
+	log.Info().Msgf("wrote file %s/%s", flagOutdir, dbEncryptionKeyPath)
 }

cmd/bootstrap/cmd/dkg.go

@@ -6,6 +6,7 @@ import (
 	"github.com/onflow/crypto"
 
 	bootstrapDKG "github.com/onflow/flow-go/cmd/bootstrap/dkg"
+	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	model "github.com/onflow/flow-go/model/bootstrap"
 	"github.com/onflow/flow-go/model/dkg"
 	"github.com/onflow/flow-go/model/encodable"
@@ -38,17 +39,25 @@ func runBeaconKG(nodes []model.NodeInfo) dkg.DKGData {
 		encKey := encodable.RandomBeaconPrivKey{PrivateKey: privKey}
 		privKeyShares = append(privKeyShares, encKey)
 
-		writeJSON(fmt.Sprintf(model.PathRandomBeaconPriv, nodeID), encKey)
+		err = common.WriteJSON(fmt.Sprintf(model.PathRandomBeaconPriv, nodeID), flagOutdir, encKey)
+		if err != nil {
+			log.Fatal().Err(err).Msg("failed to write json")
+		}
+		log.Info().Msgf("wrote file %s/%s", flagOutdir, fmt.Sprintf(model.PathRandomBeaconPriv, nodeID))
 	}
 
 	// write full DKG info that will be used to construct QC
-	writeJSON(model.PathRootDKGData, inmem.EncodableFullDKG{
+	err = common.WriteJSON(model.PathRootDKGData, flagOutdir, inmem.EncodableFullDKG{
 		GroupKey: encodable.RandomBeaconPubKey{
 			PublicKey: dkgData.PubGroupKey,
 		},
 		PubKeyShares:  pubKeyShares,
 		PrivKeyShares: privKeyShares,
 	})
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to write json")
+	}
+	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathRootDKGData)
 
 	return dkgData
 }