FEATURE: Serializable Snapshots

.github/workflows/ci.yml

@@ -4,9 +4,11 @@ on:
   push:
     branches: 
       - master
+      - feature/serializable-snapshots
   pull_request:
     branches:
       - master
+      - feature/serializable-snapshots
 
 jobs:
   golangci:

access/api.go

@@ -39,6 +39,8 @@ type API interface {
 
 	GetEventsForHeightRange(ctx context.Context, eventType string, startHeight, endHeight uint64) ([]flow.BlockEvents, error)
 	GetEventsForBlockIDs(ctx context.Context, eventType string, blockIDs []flow.Identifier) ([]flow.BlockEvents, error)
+
+	GetLatestProtocolStateSnapshot(ctx context.Context) ([]byte, error)
 }
 
 // TODO: Combine this with flow.TransactionResult?

access/handler.go

@@ -2,7 +2,6 @@ package access
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/golang/protobuf/ptypes"
 	"github.com/onflow/flow/protobuf/go/flow/access"
@@ -36,11 +35,6 @@ func (h *Handler) Ping(ctx context.Context, _ *access.PingRequest) (*access.Ping
 	return &access.PingResponse{}, nil
 }
 
-func (h *Handler) GetLatestProtocolStateSnapshot(_ context.Context,
-	_ *access.GetLatestProtocolStateSnapshotRequest) (*access.ProtocolStateSnapshotResponse, error) {
-	return nil, fmt.Errorf("unimplemented method")
-}
-
 func (h *Handler) GetNetworkParameters(
 	ctx context.Context,
 	_ *access.GetNetworkParametersRequest,
@@ -411,6 +405,18 @@ func (h *Handler) GetEventsForBlockIDs(
 	}, nil
 }
 
+// GetLatestProtocolStateSnapshot returns the latest serializable Snapshot
+func (h *Handler) GetLatestProtocolStateSnapshot(ctx context.Context, req *access.GetLatestProtocolStateSnapshotRequest) (*access.ProtocolStateSnapshotResponse, error) {
+	snapshot, err := h.api.GetLatestProtocolStateSnapshot(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &access.ProtocolStateSnapshotResponse{
+		SerializedSnapshot: snapshot,
+	}, nil
+}
+
 func blockResponse(block *flow.Block) (*access.BlockResponse, error) {
 	msg, err := convert.BlockToMessage(block)
 	if err != nil {

cmd/bootstrap/cmd/seal.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"crypto/rand"
 	"encoding/hex"
 
 	"github.com/onflow/flow-go/cmd/bootstrap/run"
@@ -24,14 +25,20 @@ func constructRootResultAndSeal(
 	}
 
 	participants := model.ToIdentityList(participantNodes)
-	blockID := block.ID()
+
+	randomSource := make([]byte, flow.EpochSetupRandomSourceLength)
+	_, err = rand.Read(randomSource)
+	if err != nil {
+		log.Fatal().Err(err).Msg("could not generate random source for epoch setup event")
+	}
 
 	epochSetup := &flow.EpochSetup{
 		Counter:      flagEpochCounter,
+		FirstView:    block.Header.View,
 		FinalView:    block.Header.View + leader.EstimatedSixMonthOfViews,
 		Participants: participants,
 		Assignments:  assignments,
-		RandomSource: blockID[:],
+		RandomSource: randomSource,
 	}
 
 	dkgLookup := model.ToDKGLookup(dkgData, participants)
@@ -43,8 +50,8 @@ func constructRootResultAndSeal(
 		DKGParticipants: dkgLookup,
 	}
 
-	result := run.GenerateRootResult(block, stateCommit)
-	seal := run.GenerateRootSeal(result, epochSetup, epochCommit)
+	result := run.GenerateRootResult(block, stateCommit, epochSetup, epochCommit)
+	seal := run.GenerateRootSeal(result)
 
 	writeJSON(model.PathRootResult, result)
 	writeJSON(model.PathRootSeal, seal)

cmd/bootstrap/run/result.go

@@ -5,11 +5,18 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 )
 
-func GenerateRootResult(block *flow.Block, commit flow.StateCommitment) *flow.ExecutionResult {
+func GenerateRootResult(
+	block *flow.Block,
+	commit flow.StateCommitment,
+	epochSetup *flow.EpochSetup,
+	epochCommit *flow.EpochCommit,
+) *flow.ExecutionResult {
+
 	result := &flow.ExecutionResult{
 		PreviousResultID: flow.ZeroID,
 		BlockID:          block.ID(),
 		Chunks:           chunks.ChunkListFromCommit(commit),
+		ServiceEvents:    []flow.ServiceEvent{epochSetup.ServiceEvent(), epochCommit.ServiceEvent()},
 	}
 	return result
 }

cmd/bootstrap/run/seal.go

@@ -4,13 +4,12 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 )
 
-func GenerateRootSeal(result *flow.ExecutionResult, setup *flow.EpochSetup, commit *flow.EpochCommit) *flow.Seal {
+func GenerateRootSeal(result *flow.ExecutionResult) *flow.Seal {
 	finalState, _ := result.FinalStateCommitment()
 	seal := &flow.Seal{
-		BlockID:       result.BlockID,
-		ResultID:      result.ID(),
-		FinalState:    finalState,
-		ServiceEvents: []flow.ServiceEvent{setup.ServiceEvent(), commit.ServiceEvent()},
+		BlockID:    result.BlockID,
+		ResultID:   result.ID(),
+		FinalState: finalState,
 	}
 	return seal
 }

cmd/bootstrap/transit/main.go

@@ -17,8 +17,11 @@ import (
 	"golang.org/x/crypto/nacl/box"
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
+	"google.golang.org/grpc"
 
+	"github.com/onflow/flow-go-sdk/client"
 	"github.com/onflow/flow-go/cmd/bootstrap/build"
+	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/ledger/complete/wal"
 	"github.com/onflow/flow-go/model/bootstrap"
 	"github.com/onflow/flow-go/model/flow"
@@ -55,18 +58,21 @@ var (
 
 func main() {
 
-	var bootDir, keyDir, wrapID, role string
-	var version, pull, push, prepare bool
+	var bootDir, keyDir, wrapID, role, accessAddress string
+	var version, pull, push, prepare, downloadSnapshot bool
 
 	flag.BoolVar(&version, "v", false, "View version and commit information")
 	flag.StringVar(&bootDir, "d", "~/bootstrap", "The bootstrap directory containing your node-info files")
 	flag.StringVar(&keyDir, "t", "", "Token provided by the Flow team to access the transit server")
 	flag.BoolVar(&pull, "pull", false, "Fetch keys and metadata from the transit server")
+	flag.BoolVar(&downloadSnapshot, "download-snapshot", false, "Download the latest protocol state snapshot from an access node and write to disk")
 	flag.BoolVar(&push, "push", false, "Upload public keys to the transit server")
 	flag.BoolVar(&prepare, "prepare", false, "Generate transit keys for push step")
 	flag.StringVar(&role, "role", "", `node role (can be "collection", "consensus", "execution", "verification" or "access")`)
 	flag.StringVar(&wrapID, "x-server-wrap", "", "(Flow Team Use), wrap response keys for consensus node")
+	flag.StringVar(&accessAddress, "access-address", "", "The address of an access node")
 	flag.StringVar(&flowBucket, "flow-bucket", "flow-genesis-bootstrap", "Storage for the transit server")
+
 	flag.Parse()
 
 	// always print version information
@@ -96,16 +102,21 @@ func main() {
 		return
 	}
 
-	if optionsSelected(pull, push, prepare) != 1 {
+	if optionsSelected(pull, push, prepare, downloadSnapshot) != 1 {
 		flag.Usage()
-		log.Fatal("Exactly one of -pull, -push, or -prepare must be specified\n")
+		log.Fatal("Exactly one of -pull, -push, -download-snapshot, or -prepare must be specified\n")
 	}
 
-	if !prepare && keyDir == "" {
+	if (push || pull) && keyDir == "" {
 		flag.Usage()
 		log.Fatal("Access key, '-t', required for push and pull commands")
 	}
 
+	if downloadSnapshot && accessAddress == "" {
+		flag.Usage()
+		log.Fatal("Access address, '-access-address', required to download latest protocl snapshot")
+	}
+
 	nodeID, err := fetchNodeID(bootDir)
 	if err != nil {
 		log.Fatalf("Could not determine node ID: %s\n", err)
@@ -130,6 +141,11 @@ func main() {
 		runPrepare(bootDir, nodeID, flowRole)
 		return
 	}
+
+	if prepare {
+		runDownloadSnapshot(ctx, bootDir, nodeID, "")
+		return
+	}
 }
 
 // Read the NodeID file to build other paths from
@@ -161,11 +177,11 @@ func printVersion() {
 // Run the push process
 // - create transit keypair (if the role type is Consensus)
 // - upload files to GCS bucket
-func runPush(ctx context.Context, bootDir, token, nodeId string, role flow.Role) {
+func runPush(ctx context.Context, bootDir, token, nodeID string, role flow.Role) {
 	log.Println("Running push")
 
 	if role == flow.RoleConsensus {
-		err := generateKeys(bootDir, nodeId)
+		err := generateKeys(bootDir, nodeID)
 		if err != nil {
 			log.Fatalf("Failed to push: %s", err)
 		}
@@ -174,18 +190,18 @@ func runPush(ctx context.Context, bootDir, token, nodeId string, role flow.Role)
 	files := getFilesToUpload(role)
 
 	for _, file := range files {
-		err := bucketUpload(ctx, bootDir, fmt.Sprintf(file, nodeId), token)
+		err := bucketUpload(ctx, bootDir, fmt.Sprintf(file, nodeID), token)
 		if err != nil {
 			log.Fatalf("Failed to push: %s", err)
 		}
 	}
 }
 
-func runPull(ctx context.Context, bootDir, token, nodeId string, role flow.Role) {
+func runPull(ctx context.Context, bootDir, token, nodeID string, role flow.Role) {
 
 	log.Println("Running pull")
 
-	extraFiles := getAdditionalFilesToDownload(role, nodeId)
+	extraFiles := getAdditionalFilesToDownload(role, nodeID)
 
 	var err error
 
@@ -213,26 +229,26 @@ func runPull(ctx context.Context, bootDir, token, nodeId string, role flow.Role)
 	}
 
 	if role == flow.RoleConsensus {
-		err = unwrapFile(bootDir, nodeId)
+		err = unwrapFile(bootDir, nodeID)
 		if err != nil {
 			log.Fatalf("Failed to pull: %s", err)
 		}
 	}
 
-	rootFile := filepath.Join(bootDir, bootstrap.PathRootBlock)
+	rootFile := filepath.Join(bootDir, bootstrap.PathRootProtocolStateSnapshot)
 	rootMD5, err := getFileMD5(rootFile)
 	if err != nil {
 		log.Fatalf("Failed to calculate md5 of %s: %v", rootFile, err)
 	}
-	log.Printf("MD5 of the root block is: %s\n", rootMD5)
+	log.Printf("MD5 of the root protocol snapshot is: %s\n", rootMD5)
 }
 
 // Run the prepare process
 // - create transit keypair (if the role type is Consensus)
-func runPrepare(bootdir, nodeId string, role flow.Role) {
+func runPrepare(bootdir, nodeID string, role flow.Role) {
 	if role == flow.RoleConsensus {
 		log.Println("creating transit-keys")
-		err := generateKeys(bootdir, nodeId)
+		err := generateKeys(bootdir, nodeID)
 		if err != nil {
 			log.Fatalf("Failed to prepare: %s", err)
 		}
@@ -241,11 +257,67 @@ func runPrepare(bootdir, nodeId string, role flow.Role) {
 	log.Printf("no preparation needed for role: %s", role.String())
 }
 
+// runDownloadSnapshot fetches the latest protocol snapshot from an access node and
+// writes it to the `root-protocol-snapshot.json` file
+func runDownloadSnapshot(ctx context.Context, bootDir, nodeIDHex, accessAddress string) {
+
+	// convert the nodeID string to `flow.Identifier`
+	nodeID, err := flow.HexStringToIdentifier(nodeIDHex)
+	if err != nil {
+		log.Fatal("could not convert node id to identifier")
+	}
+
+	// create a flow client with given access address
+	flowClient, err := client.New(accessAddress, grpc.WithInsecure())
+	if err != nil {
+		log.Fatalf("could not create flow client: %v", err)
+	}
+
+	// get latest snapshot bytes encoded as JSON
+	bytes, err := flowClient.GetLatestProtocolStateSnapshot(ctx)
+	if err != nil {
+		log.Fatal("could not get latest protocol snapshot from access node")
+	}
+
+	// unmarshal bytes to snapshot
+	snapshot, err := convert.BytesToInmemSnapshot(bytes)
+	if err != nil {
+		log.Fatalf("could not convert array of bytes to snapshot: %v", err)
+	}
+
+	// check if given NodeID is part of the current or next epoch
+	currentIdentities, err := snapshot.Epochs().Current().InitialIdentities()
+	if err != nil {
+		log.Fatal("could not get initial identities from current epoch")
+	}
+	if _, exists := currentIdentities.ByNodeID(nodeID); exists {
+		err := utilsio.WriteFile(filepath.Join(bootDir, bootstrap.PathRootProtocolStateSnapshot), bytes)
+		if err != nil {
+			log.Fatalf("could not write snapshot to disk: %v", err)
+		}
+		return
+	}
+
+	nextIdentities, err := snapshot.Epochs().Next().InitialIdentities()
+	if err != nil {
+		log.Fatal("could not get initial identities from next epoch")
+	}
+	if _, exists := nextIdentities.ByNodeID(nodeID); exists {
+		err := utilsio.WriteFile(filepath.Join(bootDir, bootstrap.PathRootProtocolStateSnapshot), bytes)
+		if err != nil {
+			log.Fatalf("could not write snapshot to disk: %v", err)
+		}
+		return
+	}
+
+	log.Fatalf("could not write snapshot, given node ID does not belong to current or next epoch: %v", nodeID)
+}
+
 // generateKeys creates the transit keypair and writes them to disk for later
-func generateKeys(bootDir, nodeId string) error {
+func generateKeys(bootDir, nodeID string) error {
 
-	privPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPriv, nodeId))
-	pubPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeId))
+	privPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPriv, nodeID))
+	pubPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeID))
 
 	if utilsio.FileExists(privPath) && utilsio.FileExists(pubPath) {
 		log.Print("transit-key-path priv & pub both exist, exiting")
@@ -275,14 +347,14 @@ func generateKeys(bootDir, nodeId string) error {
 	return nil
 }
 
-func unwrapFile(bootDir, nodeId string) error {
+func unwrapFile(bootDir, nodeID string) error {
 
 	log.Print("Decrypting Random Beacon key")
 
-	pubKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeId))
-	privKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPriv, nodeId))
-	ciphertextPath := filepath.Join(bootDir, fmt.Sprintf(FilenameRandomBeaconCipher, nodeId))
-	plaintextPath := filepath.Join(bootDir, fmt.Sprintf(bootstrap.PathRandomBeaconPriv, nodeId))
+	pubKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeID))
+	privKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPriv, nodeID))
+	ciphertextPath := filepath.Join(bootDir, fmt.Sprintf(FilenameRandomBeaconCipher, nodeID))
+	plaintextPath := filepath.Join(bootDir, fmt.Sprintf(bootstrap.PathRandomBeaconPriv, nodeID))
 
 	ciphertext, err := utilsio.ReadFile(ciphertextPath)
 	if err != nil {
@@ -318,10 +390,10 @@ func unwrapFile(bootDir, nodeId string) error {
 	return nil
 }
 
-func wrapFile(bootDir, nodeId string) error {
-	pubKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeId))
-	plaintextPath := filepath.Join(bootDir, fmt.Sprintf(bootstrap.PathRandomBeaconPriv, nodeId))
-	ciphertextPath := filepath.Join(bootDir, fmt.Sprintf(FilenameRandomBeaconCipher, nodeId))
+func wrapFile(bootDir, nodeID string) error {
+	pubKeyPath := filepath.Join(bootDir, fmt.Sprintf(FilenameTransitKeyPub, nodeID))
+	plaintextPath := filepath.Join(bootDir, fmt.Sprintf(bootstrap.PathRandomBeaconPriv, nodeID))
+	ciphertextPath := filepath.Join(bootDir, fmt.Sprintf(FilenameRandomBeaconCipher, nodeID))
 
 	plaintext, err := utilsio.ReadFile(plaintextPath)
 	if err != nil {
@@ -461,10 +533,10 @@ func getFilesToUpload(role flow.Role) []string {
 	}
 }
 
-func getAdditionalFilesToDownload(role flow.Role, nodeId string) []string {
+func getAdditionalFilesToDownload(role flow.Role, nodeID string) []string {
 	switch role {
 	case flow.RoleConsensus:
-		return []string{fmt.Sprintf(filesToDownloadConsensus, nodeId)}
+		return []string{fmt.Sprintf(filesToDownloadConsensus, nodeID)}
 	}
 	return make([]string, 0)
 }