Sync feature/stable-cadence branch with master

.github/workflows/ci.yml

@@ -14,6 +14,9 @@ on:
       - 'auto-cadence-upgrade/**'
       - 'feature/**'
       - 'v[0-9]+.[0-9]+'
+  merge_group:
+    branches:
+      - master
 
 env:
   GO_VERSION: "1.20"
@@ -189,10 +192,13 @@ jobs:
       matrix:
         make:
           - make -C integration access-tests
-          - make -C integration bft-tests
+          - make -C integration bft-framework-tests
+          - make -C integration bft-protocol-tests
+          - make -C integration bft-gossipsub-tests
           - make -C integration collection-tests
           - make -C integration consensus-tests
-          - make -C integration epochs-tests
+          - make -C integration epochs-cohort1-tests
+          - make -C integration epochs-cohort2-tests
           - make -C integration execution-tests
           - make -C integration ghost-tests
           - make -C integration mvp-tests

.github/workflows/flaky-test-monitor.yml

@@ -8,6 +8,7 @@ on:
   push:
     paths:
       - '.github/workflows/flaky-test-monitor.yml'
+      - '.github/workflows/ci.yml'
 
 env:
   BIGQUERY_DATASET: production_src_flow_test_metrics
@@ -128,13 +129,19 @@ jobs:
         include:
           - target: access-tests
             test_category: integration-access
-          - target: bft-tests
-            test_category: integration-bft
+          - target: bft-protocol-tests
+            test_category: integration-bft-protocol
+          - target: bft-framework-tests
+            test_category: integration-bft-framework
+          - target: bft-gossipsub-tests
+            test_category: integration-bft-gossipsub
           - target: collection-tests
             test_category: integration-collection
           - target: consensus-tests
             test_category: integration-consensus
-          - target: epochs-tests
+          - target: epochs-cohort1-tests
+            test_category: integration-epochs
+          - target: epochs-cohort2-tests
             test_category: integration-epochs
           - target: execution-tests
             test_category: integration-execution

.github/workflows/tools.yml

@@ -55,6 +55,9 @@ jobs:
         if [[ "${{ inputs.promote }}" = true ]]; then
           echo "promoting boot-tools.tar"
           gsutil cp boot-tools.tar gs://flow-genesis-bootstrap/boot-tools.tar
+          SUMMARY=$'# Tool Build and Upload Summary \n Your tools were uploaded to the following GCS objects \n * Boot Tools gs://flow-genesis-bootstrap/boot-tools.tar \n * Util util.tar gs://flow-genesis-bootstrap/tools/${{ inputs.tag }}/util.tar'
         else
           echo "not promoting boot-tools.tar"
+          SUMMARY=$'# Tool Build and Upload Summary \n Your tools were uploaded to the following GCS objects \n * Boot Tools gs://flow-genesis-bootstrap/tools/${{ inputs.tag }}/boot-tools.tar \n * Util util.tar gs://flow-genesis-bootstrap/tools/${{ inputs.tag }}/util.tar'
         fi
+        echo "$SUMMARY" >> $GITHUB_STEP_SUMMARY

bors.toml

@@ -17,10 +17,13 @@ status = [
   "Unit Tests (utils)",
   "Unit Tests (others)",
   "Integration Tests (make -C integration access-tests)",
-  "Integration Tests (make -C integration bft-tests)",
+  "Integration Tests (make -C integration bft-framework-tests)",
+  "Integration Tests (make -C integration bft-gossipsub-tests)",
+  "Integration Tests (make -C integration bft-protocol-tests)",
   "Integration Tests (make -C integration collection-tests)",
   "Integration Tests (make -C integration consensus-tests)",
-  "Integration Tests (make -C integration epochs-tests)",
+  "Integration Tests (make -C integration epochs-cohort1-tests)",
+  "Integration Tests (make -C integration epochs-cohort2-tests)",
   "Integration Tests (make -C integration execution-tests)",
   "Integration Tests (make -C integration ghost-tests)",
   "Integration Tests (make -C integration mvp-tests)",

cmd/access/node_builder/access_node_builder.go

@@ -802,6 +802,7 @@ func (builder *FlowAccessNodeBuilder) initNetwork(nodeID module.Local,
 		IdentityProvider:    builder.IdentityProvider,
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             builder.SporkID,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  builder.Logger,
 			SpamRecordCacheSize:     builder.FlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,
@@ -1404,7 +1405,7 @@ func (builder *FlowAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 		Libp2pNode:            libp2pNode,
 		FlowId:                nodeID,
 		BitSwapMetrics:        builder.Metrics.Bitswap,
-		RootBlockID:           builder.SporkID,
+		SporkId:               builder.SporkID,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		IdTranslator:          builder.IDTranslator,
 		Codec:                 builder.CodecFactory(),

cmd/execution_builder.go

@@ -13,6 +13,7 @@ import (
 
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	badgerDB "github.com/dgraph-io/badger/v2"
 	"github.com/ipfs/go-cid"
 	badger "github.com/ipfs/go-ds-badger2"
 	"github.com/onflow/flow-core-contracts/lib/go/templates"
@@ -82,6 +83,7 @@ import (
 	storageerr "github.com/onflow/flow-go/storage"
 	storage "github.com/onflow/flow-go/storage/badger"
 	"github.com/onflow/flow-go/storage/badger/procedure"
+	sutil "github.com/onflow/flow-go/storage/util"
 )
 
 const (
@@ -627,14 +629,48 @@ func (exeNode *ExecutionNode) LoadExecutionDataGetter(node *NodeConfig) error {
 	return nil
 }
 
+func openChunkDataPackDB(dbPath string, logger zerolog.Logger) (*badgerDB.DB, error) {
+	log := sutil.NewLogger(logger)
+
+	opts := badgerDB.
+		DefaultOptions(dbPath).
+		WithKeepL0InMemory(true).
+		WithLogger(log).
+
+		// the ValueLogFileSize option specifies how big the value of a
+		// key-value pair is allowed to be saved into badger.
+		// exceeding this limit, will fail with an error like this:
+		// could not store data: Value with size <xxxx> exceeded 1073741824 limit
+		// Maximum value size is 10G, needed by execution node
+		// TODO: finding a better max value for each node type
+		WithValueLogFileSize(256 << 23).
+		WithValueLogMaxEntries(100000) // Default is 1000000
+
+	db, err := badgerDB.Open(opts)
+	if err != nil {
+		return nil, fmt.Errorf("could not open chunk data pack badger db at path %v: %w", dbPath, err)
+	}
+	return db, nil
+}
+
 func (exeNode *ExecutionNode) LoadExecutionState(
 	node *NodeConfig,
 ) (
 	module.ReadyDoneAware,
 	error,
 ) {
 
-	chunkDataPacks := storage.NewChunkDataPacks(node.Metrics.Cache, node.DB, node.Storage.Collections, exeNode.exeConf.chunkDataPackCacheSize)
+	chunkDataPackDB, err := openChunkDataPackDB(exeNode.exeConf.chunkDataPackDir, node.Logger)
+	if err != nil {
+		return nil, err
+	}
+	exeNode.builder.ShutdownFunc(func() error {
+		if err := chunkDataPackDB.Close(); err != nil {
+			return fmt.Errorf("error closing chunk data pack database: %w", err)
+		}
+		return nil
+	})
+	chunkDataPacks := storage.NewChunkDataPacks(node.Metrics.Cache, chunkDataPackDB, node.Storage.Collections, exeNode.exeConf.chunkDataPackCacheSize)
 
 	// Needed for gRPC server, make sure to assign to main scoped vars
 	exeNode.events = storage.NewEvents(node.Metrics.Cache, node.DB)

cmd/execution_config.go

@@ -32,6 +32,7 @@ type ExecutionConfig struct {
 	transactionResultsCacheSize          uint
 	checkpointDistance                   uint
 	checkpointsToKeep                    uint
+	chunkDataPackDir                     string
 	chunkDataPackCacheSize               uint
 	chunkDataPackRequestsCacheSize       uint32
 	requestInterval                      time.Duration
@@ -80,6 +81,7 @@ func (exeConf *ExecutionConfig) SetupFlags(flags *pflag.FlagSet) {
 	flags.BoolVar(&exeConf.computationConfig.ExtensiveTracing, "extensive-tracing", false, "adds high-overhead tracing to execution")
 	flags.BoolVar(&exeConf.computationConfig.CadenceTracing, "cadence-tracing", false, "enables cadence runtime level tracing")
 	flags.IntVar(&exeConf.computationConfig.MaxConcurrency, "computer-max-concurrency", 1, "set to greater than 1 to enable concurrent transaction execution")
+	flags.StringVar(&exeConf.chunkDataPackDir, "chunk-data-pack-dir", filepath.Join(homedir, ".flow", "chunk_data_packs"), "directory to use for storing chunk data packs")
 	flags.UintVar(&exeConf.chunkDataPackCacheSize, "chdp-cache", storage.DefaultCacheSize, "cache size for chunk data packs")
 	flags.Uint32Var(&exeConf.chunkDataPackRequestsCacheSize, "chdp-request-queue", mempool.DefaultChunkDataPackRequestQueueSize, "queue size for chunk data pack requests")
 	flags.DurationVar(&exeConf.requestInterval, "request-interval", 60*time.Second, "the interval between requests for the requester engine")

cmd/observer/node_builder/observer_builder.go

@@ -493,6 +493,7 @@ func (builder *ObserverServiceBuilder) initNetwork(nodeID module.Local,
 		IdentityProvider:    builder.IdentityProvider,
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             builder.SporkID,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  builder.Logger,
 			SpamRecordCacheSize:     builder.FlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,
@@ -1031,7 +1032,7 @@ func (builder *ObserverServiceBuilder) initMiddleware(nodeID flow.Identifier,
 		Libp2pNode:            libp2pNode,
 		FlowId:                nodeID,
 		BitSwapMetrics:        builder.Metrics.Bitswap,
-		RootBlockID:           builder.SporkID,
+		SporkId:               builder.SporkID,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		IdTranslator:          builder.IDTranslator,
 		Codec:                 builder.CodecFactory(),

cmd/scaffold.go

@@ -440,7 +440,7 @@ func (fnb *FlowNodeBuilder) InitFlowNetworkWithConduitFactory(
 		Libp2pNode:            fnb.LibP2PNode,
 		FlowId:                fnb.Me.NodeID(),
 		BitSwapMetrics:        fnb.Metrics.Bitswap,
-		RootBlockID:           fnb.SporkID,
+		SporkId:               fnb.SporkID,
 		UnicastMessageTimeout: fnb.FlowConfig.NetworkConfig.UnicastMessageTimeout,
 		IdTranslator:          fnb.IDTranslator,
 		Codec:                 fnb.CodecFactory(),
@@ -465,6 +465,7 @@ func (fnb *FlowNodeBuilder) InitFlowNetworkWithConduitFactory(
 		Logger:              fnb.Logger,
 		Codec:               fnb.CodecFactory(),
 		Me:                  fnb.Me,
+		SporkId:             fnb.SporkID,
 		MiddlewareFactory:   func() (network.Middleware, error) { return fnb.Middleware, nil },
 		Topology:            topology.NewFullyConnectedTopology(),
 		SubscriptionManager: subscriptionManager,

cmd/util/cmd/truncate-database/cmd.go

@@ -9,7 +9,8 @@ import (
 )
 
 var (
-	flagDatadir string
+	flagDatadir          string
+	flagChunkDataPackDir string
 )
 
 var Cmd = &cobra.Command{
@@ -24,14 +25,22 @@ func init() {
 		"directory that stores the protocol state")
 	_ = Cmd.MarkFlagRequired("datadir")
 
+	Cmd.Flags().StringVar(&flagChunkDataPackDir, "chunk-data-pack-dir", "",
+		"directory that stores the chunk data pack")
+	_ = Cmd.MarkFlagRequired("chunk-data-pack-dir")
 }
 
 func run(*cobra.Command, []string) {
 
-	log.Info().Msg("Opening database with truncate")
+	log.Info().Msg("Opening protocol database with truncate")
 
 	db := common.InitStorageWithTruncate(flagDatadir, true)
 	defer db.Close()
 
+	log.Info().Msg("ProtocolDB Truncated")
+
+	chunkdb := common.InitStorageWithTruncate(flagChunkDataPackDir, true)
+	defer chunkdb.Close()
+
 	log.Info().Msg("Truncated")
 }

engine/execution/block_result.go

@@ -172,6 +172,7 @@ func (ar *BlockAttestationResult) ChunkAt(index int) *flow.Chunk {
 		len(execRes.TransactionResults()),
 		attestRes.eventCommit,
 		attestRes.endStateCommit,
+		execRes.executionSnapshot.TotalComputationUsed(),
 	)
 }
 

engine/execution/execution_test.go

@@ -566,6 +566,10 @@ func TestBroadcastToMultipleVerificationNodes(t *testing.T) {
 			receipt, _ = args[2].(*flow.ExecutionReceipt)
 
 			assert.Equal(t, block.ID(), receipt.ExecutionResult.BlockID)
+			for i, chunk := range receipt.ExecutionResult.Chunks {
+				assert.EqualValues(t, i, chunk.CollectionIndex)
+				assert.Greater(t, chunk.TotalComputationUsed, uint64(0))
+			}
 		}).
 		Return(nil)
 

engine/execution/ingestion/engine_test.go

@@ -418,7 +418,9 @@ func TestChunkIndexIsSet(t *testing.T) {
 		unittest.StateCommitmentFixture(),
 		21,
 		unittest.IdentifierFixture(),
-		unittest.StateCommitmentFixture())
+		unittest.StateCommitmentFixture(),
+		17995,
+	)
 
 	assert.Equal(t, i, int(chunk.Index))
 	assert.Equal(t, i, int(chunk.CollectionIndex))
@@ -433,11 +435,29 @@ func TestChunkNumberOfTxsIsSet(t *testing.T) {
 		unittest.StateCommitmentFixture(),
 		i,
 		unittest.IdentifierFixture(),
-		unittest.StateCommitmentFixture())
+		unittest.StateCommitmentFixture(),
+		17995,
+	)
 
 	assert.Equal(t, i, int(chunk.NumberOfTransactions))
 }
 
+func TestChunkTotalComputationUsedIsSet(t *testing.T) {
+
+	i := mathRand.Uint64()
+	chunk := flow.NewChunk(
+		unittest.IdentifierFixture(),
+		3,
+		unittest.StateCommitmentFixture(),
+		21,
+		unittest.IdentifierFixture(),
+		unittest.StateCommitmentFixture(),
+		i,
+	)
+
+	assert.Equal(t, i, chunk.TotalComputationUsed)
+}
+
 func TestExecuteOneBlock(t *testing.T) {
 	runWithEngine(t, func(ctx testingContext) {
 

engine/execution/state/state.go

@@ -283,24 +283,51 @@ func (s *state) SaveExecutionResults(
 		trace.EXEStateSaveExecutionResults)
 	defer span.End()
 
+	err := s.saveExecutionResults(ctx, result)
+	if err != nil {
+		return fmt.Errorf("could not save execution results: %w", err)
+	}
+
+	//outside batch because it requires read access
+	err = s.UpdateHighestExecutedBlockIfHigher(childCtx, result.ExecutableBlock.Block.Header)
+	if err != nil {
+		return fmt.Errorf("cannot update highest executed block: %w", err)
+	}
+	return nil
+}
+
+func (s *state) saveExecutionResults(
+	ctx context.Context,
+	result *execution.ComputationResult,
+) (err error) {
 	header := result.ExecutableBlock.Block.Header
 	blockID := header.ID()
 
+	err = s.chunkDataPacks.Store(result.AllChunkDataPacks())
+	if err != nil {
+		return fmt.Errorf("can not store multiple chunk data pack: %w", err)
+	}
+
 	// Write Batch is BadgerDB feature designed for handling lots of writes
-	// in efficient and automatic manner, hence pushing all the updates we can
+	// in efficient and atomic manner, hence pushing all the updates we can
 	// as tightly as possible to let Badger manage it.
 	// Note, that it does not guarantee atomicity as transactions has size limit,
 	// but it's the closest thing to atomicity we could have
 	batch := badgerstorage.NewBatch(s.db)
 
-	for _, chunkDataPack := range result.AllChunkDataPacks() {
-		err := s.chunkDataPacks.BatchStore(chunkDataPack, batch)
+	defer func() {
+		// Rollback if an error occurs during batch operations
 		if err != nil {
-			return fmt.Errorf("cannot store chunk data pack: %w", err)
+			chunks := result.AllChunkDataPacks()
+			chunkIDs := make([]flow.Identifier, 0, len(chunks))
+			for _, chunk := range chunks {
+				chunkIDs = append(chunkIDs, chunk.ID())
+			}
+			_ = s.chunkDataPacks.Remove(chunkIDs)
 		}
-	}
+	}()
 
-	err := s.events.BatchStore(blockID, []flow.EventsList{result.AllEvents()}, batch)
+	err = s.events.BatchStore(blockID, []flow.EventsList{result.AllEvents()}, batch)
 	if err != nil {
 		return fmt.Errorf("cannot store events: %w", err)
 	}
@@ -347,11 +374,6 @@ func (s *state) SaveExecutionResults(
 		return fmt.Errorf("batch flush error: %w", err)
 	}
 
-	//outside batch because it requires read access
-	err = s.UpdateHighestExecutedBlockIfHigher(childCtx, header)
-	if err != nil {
-		return fmt.Errorf("cannot update highest executed block: %w", err)
-	}
 	return nil
 }
 

engine/execution/state/unittest/fixtures.go

@@ -3,14 +3,17 @@ package unittest
 import (
 	"github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/engine/execution"
+	"github.com/onflow/flow-go/fvm/meter"
 	"github.com/onflow/flow-go/fvm/storage/snapshot"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/mempool/entity"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
 func StateInteractionsFixture() *snapshot.ExecutionSnapshot {
-	return &snapshot.ExecutionSnapshot{}
+	return &snapshot.ExecutionSnapshot{
+		Meter: meter.NewMeter(meter.DefaultParameters()),
+	}
 }
 
 func ComputationResultFixture(