Merge branch 'master' into gregor/evm/direct-call

admin/README.md

@@ -96,6 +96,11 @@ curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"
 curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "stop-at-height", "data": { "height": 1111, "crash": false }}'
 ```
 
+### Trigger checkpoint creation on execution
+```
+curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "trigger-checkpoint"}'
+```
+
 ### Add/Remove/Get address to rate limit a payer from adding transactions to collection nodes' mempool
 ```
 curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "ingest-tx-rate-limit", "data": { "command": "add", "addresses": "a08d349e8037d6e5,e6765c6113547fb7" }}'

cmd/access/node_builder/access_node_builder.go

@@ -1505,12 +1505,12 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			cacheSize := int(backendConfig.ConnectionPoolSize)
 
 			var connBackendCache *rpcConnection.Cache
+			var err error
 			if cacheSize > 0 {
-				backendCache, err := backend.NewCache(node.Logger, accessMetrics, cacheSize)
+				connBackendCache, err = rpcConnection.NewCache(node.Logger, accessMetrics, cacheSize)
 				if err != nil {
-					return nil, fmt.Errorf("could not initialize backend cache: %w", err)
+					return nil, fmt.Errorf("could not initialize connection cache: %w", err)
 				}
-				connBackendCache = rpcConnection.NewCache(backendCache, cacheSize)
 			}
 
 			connFactory := &rpcConnection.ConnectionFactoryImpl{
@@ -1521,9 +1521,9 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				AccessMetrics:             accessMetrics,
 				Log:                       node.Logger,
 				Manager: rpcConnection.NewManager(
-					connBackendCache,
 					node.Logger,
 					accessMetrics,
+					connBackendCache,
 					config.MaxMsgSize,
 					backendConfig.CircuitBreakerConfig,
 					config.CompressorName,

cmd/bootstrap/run/execution_state.go

@@ -43,7 +43,7 @@ func GenerateExecutionState(
 		return flow.DummyStateCommitment, err
 	}
 
-	compactor, err := complete.NewCompactor(ledgerStorage, diskWal, zerolog.Nop(), capacity, checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+	compactor, err := complete.NewCompactor(ledgerStorage, diskWal, zerolog.Nop(), capacity, checkpointDistance, checkpointsToKeep, atomic.NewBool(false), metricsCollector)
 	if err != nil {
 		return flow.DummyStateCommitment, err
 	}

cmd/execution_builder.go

@@ -881,6 +881,7 @@ func (exeNode *ExecutionNode) LoadExecutionStateLedgerWALCompactor(
 		exeNode.exeConf.checkpointDistance,
 		exeNode.exeConf.checkpointsToKeep,
 		exeNode.toTriggerCheckpoint, // compactor will listen to the signal from admin tool for force triggering checkpointing
+		exeNode.collector,
 	)
 }
 

cmd/observer/node_builder/observer_builder.go

@@ -1210,12 +1210,12 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 		cacheSize := int(backendConfig.ConnectionPoolSize)
 
 		var connBackendCache *rpcConnection.Cache
+		var err error
 		if cacheSize > 0 {
-			backendCache, err := backend.NewCache(node.Logger, accessMetrics, cacheSize)
+			connBackendCache, err = rpcConnection.NewCache(node.Logger, accessMetrics, cacheSize)
 			if err != nil {
-				return nil, fmt.Errorf("could not initialize backend cache: %w", err)
+				return nil, fmt.Errorf("could not initialize connection cache: %w", err)
 			}
-			connBackendCache = rpcConnection.NewCache(backendCache, cacheSize)
 		}
 
 		connFactory := &rpcConnection.ConnectionFactoryImpl{
@@ -1226,9 +1226,9 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			AccessMetrics:             accessMetrics,
 			Log:                       node.Logger,
 			Manager: rpcConnection.NewManager(
-				connBackendCache,
 				node.Logger,
 				accessMetrics,
+				connBackendCache,
 				config.MaxMsgSize,
 				backendConfig.CircuitBreakerConfig,
 				config.CompressorName,

cmd/util/cmd/checkpoint-collect-stats/cmd.go

@@ -93,7 +93,7 @@ func run(*cobra.Command, []string) {
 	if err != nil {
 		log.Fatal().Err(err).Msg("cannot create ledger from write-a-head logs and checkpoints")
 	}
-	compactor, err := complete.NewCompactor(led, diskWal, zerolog.Nop(), complete.DefaultCacheSize, math.MaxInt, 1, atomic.NewBool(false))
+	compactor, err := complete.NewCompactor(led, diskWal, zerolog.Nop(), complete.DefaultCacheSize, math.MaxInt, 1, atomic.NewBool(false), &metrics.NoopCollector{})
 	if err != nil {
 		log.Fatal().Err(err).Msg("cannot create compactor")
 	}

cmd/util/cmd/exec-data-json-export/ledger_exporter.go

@@ -35,7 +35,7 @@ func ExportLedger(ledgerPath string, targetstate string, outputPath string) erro
 		return fmt.Errorf("cannot create ledger from write-a-head logs and checkpoints: %w", err)
 	}
 
-	compactor, err := complete.NewCompactor(led, diskWal, zerolog.Nop(), complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+	compactor, err := complete.NewCompactor(led, diskWal, zerolog.Nop(), complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false), &metrics.NoopCollector{})
 	if err != nil {
 		return fmt.Errorf("cannot create compactor: %w", err)
 	}

cmd/util/cmd/execution-state-extract/execution_state_extract.go

@@ -70,7 +70,7 @@ func extractExecutionState(
 
 	log.Info().Msg("init compactor")
 
-	compactor, err := complete.NewCompactor(led, diskWal, log, complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+	compactor, err := complete.NewCompactor(led, diskWal, log, complete.DefaultCacheSize, checkpointDistance, checkpointsToKeep, atomic.NewBool(false), &metrics.NoopCollector{})
 	if err != nil {
 		return fmt.Errorf("cannot create compactor: %w", err)
 	}

cmd/util/cmd/execution-state-extract/execution_state_extract_test.go

@@ -90,7 +90,7 @@ func TestExtractExecutionState(t *testing.T) {
 			require.NoError(t, err)
 			f, err := complete.NewLedger(diskWal, size*10, metr, zerolog.Nop(), complete.DefaultPathFinderVersion)
 			require.NoError(t, err)
-			compactor, err := complete.NewCompactor(f, diskWal, zerolog.Nop(), uint(size), checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+			compactor, err := complete.NewCompactor(f, diskWal, zerolog.Nop(), uint(size), checkpointDistance, checkpointsToKeep, atomic.NewBool(false), &metrics.NoopCollector{})
 			require.NoError(t, err)
 			<-compactor.Ready()
 
@@ -166,7 +166,7 @@ func TestExtractExecutionState(t *testing.T) {
 						checkpointDistance = math.MaxInt // A large number to prevent checkpoint creation.
 						checkpointsToKeep  = 1
 					)
-					compactor, err := complete.NewCompactor(storage, diskWal, zerolog.Nop(), uint(size), checkpointDistance, checkpointsToKeep, atomic.NewBool(false))
+					compactor, err := complete.NewCompactor(storage, diskWal, zerolog.Nop(), uint(size), checkpointDistance, checkpointsToKeep, atomic.NewBool(false), &metrics.NoopCollector{})
 					require.NoError(t, err)
 
 					<-compactor.Ready()

consensus/hotstuff/votecollector/combined_vote_processor_v2_test.go

@@ -118,7 +118,7 @@ func (s *CombinedVoteProcessorV2TestSuite) TestProcess_InvalidSignatureFormat()
 	rapid.Check(s.T(), func(t *rapid.T) {
 		// create a signature with invalid length
 		vote := unittest.VoteForBlockFixture(s.proposal.Block, func(vote *model.Vote) {
-			vote.SigData = unittest.RandomBytes(generator.Draw(t, "sig-size").(int))
+			vote.SigData = unittest.RandomBytes(generator.Draw(t, "sig-size"))
 		})
 		err := s.processor.Process(vote)
 		require.Error(s.T(), err)
@@ -434,8 +434,8 @@ func TestCombinedVoteProcessorV2_PropertyCreatingQCCorrectness(testifyT *testing
 
 	rapid.Check(testifyT, func(t *rapid.T) {
 		// draw participants in range 1 <= participants <= maxParticipants
-		participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants").(uint64)
-		beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners").(uint64)
+		participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants")
+		beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners")
 		stakingSignersCount := participants - beaconSignersCount
 		require.Equal(t, participants, stakingSignersCount+beaconSignersCount)
 
@@ -638,20 +638,20 @@ func TestCombinedVoteProcessorV2_PropertyCreatingQCCorrectness(testifyT *testing
 func TestCombinedVoteProcessorV2_PropertyCreatingQCLiveness(testifyT *testing.T) {
 	rapid.Check(testifyT, func(t *rapid.T) {
 		// draw beacon signers in range 1 <= beaconSignersCount <= 53
-		beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners").(uint64)
+		beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners")
 		// draw staking signers in range 0 <= stakingSignersCount <= 10
-		stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners").(uint64)
+		stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners")
 
 		stakingWeightRange, beaconWeightRange := rapid.Uint64Range(1, 10), rapid.Uint64Range(1, 10)
 
 		minRequiredWeight := uint64(0)
 		// draw weight for each signer randomly
 		stakingSigners := unittest.IdentityListFixture(int(stakingSignersCount), func(identity *flow.Identity) {
-			identity.InitialWeight = stakingWeightRange.Draw(t, identity.String()).(uint64)
+			identity.InitialWeight = stakingWeightRange.Draw(t, identity.String())
 			minRequiredWeight += identity.InitialWeight
 		})
 		beaconSigners := unittest.IdentityListFixture(int(beaconSignersCount), func(identity *flow.Identity) {
-			identity.InitialWeight = beaconWeightRange.Draw(t, identity.String()).(uint64)
+			identity.InitialWeight = beaconWeightRange.Draw(t, identity.String())
 			minRequiredWeight += identity.InitialWeight
 		})
 

consensus/hotstuff/votecollector/combined_vote_processor_v3_test.go

@@ -434,8 +434,8 @@ func TestCombinedVoteProcessorV3_PropertyCreatingQCCorrectness(testifyT *testing
 
 	rapid.Check(testifyT, func(t *rapid.T) {
 		// draw participants in range 1 <= participants <= maxParticipants
-		participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants").(uint64)
-		beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners").(uint64)
+		participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants")
+		beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners")
 		stakingSignersCount := participants - beaconSignersCount
 		require.Equal(t, participants, stakingSignersCount+beaconSignersCount)
 
@@ -749,20 +749,20 @@ func TestCombinedVoteProcessorV3_OnlyRandomBeaconSigners(testifyT *testing.T) {
 func TestCombinedVoteProcessorV3_PropertyCreatingQCLiveness(testifyT *testing.T) {
 	rapid.Check(testifyT, func(t *rapid.T) {
 		// draw beacon signers in range 1 <= beaconSignersCount <= 53
-		beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners").(uint64)
+		beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners")
 		// draw staking signers in range 0 <= stakingSignersCount <= 10
-		stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners").(uint64)
+		stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners")
 
 		stakingWeightRange, beaconWeightRange := rapid.Uint64Range(1, 10), rapid.Uint64Range(1, 10)
 
 		minRequiredWeight := uint64(0)
 		// draw weight for each signer randomly
 		stakingSigners := unittest.IdentityListFixture(int(stakingSignersCount), func(identity *flow.Identity) {
-			identity.InitialWeight = stakingWeightRange.Draw(t, identity.String()).(uint64)
+			identity.InitialWeight = stakingWeightRange.Draw(t, identity.String())
 			minRequiredWeight += identity.InitialWeight
 		})
 		beaconSigners := unittest.IdentityListFixture(int(beaconSignersCount), func(identity *flow.Identity) {
-			identity.InitialWeight = beaconWeightRange.Draw(t, identity.String()).(uint64)
+			identity.InitialWeight = beaconWeightRange.Draw(t, identity.String())
 			minRequiredWeight += identity.InitialWeight
 		})
 

engine/access/apiproxy/access_api_proxy_test.go

@@ -152,9 +152,9 @@ func TestNewFlowCachedAccessAPIProxy(t *testing.T) {
 		AccessMetrics:             metrics,
 		CollectionNodeGRPCTimeout: time.Second,
 		Manager: connection.NewManager(
-			nil,
 			unittest.Logger(),
 			metrics,
+			nil,
 			grpcutils.DefaultMaxMsgSize,
 			connection.CircuitBreakerConfig{},
 			grpcutils.NoCompressor,

engine/access/rpc/backend/backend.go

@@ -236,27 +236,6 @@ func New(params Params) (*Backend, error) {
 	return b, nil
 }
 
-// NewCache constructs cache for storing connections to other nodes.
-// No errors are expected during normal operations.
-func NewCache(
-	log zerolog.Logger,
-	metrics module.AccessMetrics,
-	connectionPoolSize int,
-) (*lru.Cache[string, *connection.CachedClient], error) {
-	cache, err := lru.NewWithEvict(connectionPoolSize, func(_ string, client *connection.CachedClient) {
-		go client.Close() // close is blocking, so run in a goroutine
-
-		log.Debug().Str("grpc_conn_evicted", client.Address).Msg("closing grpc connection evicted from pool")
-		metrics.ConnectionFromPoolEvicted()
-	})
-
-	if err != nil {
-		return nil, fmt.Errorf("could not initialize connection pool cache: %w", err)
-	}
-
-	return cache, nil
-}
-
 func identifierList(ids []string) (flow.IdentifierList, error) {
 	idList := make(flow.IdentifierList, len(ids))
 	for i, idStr := range ids {