[Access] Connection pool evictions cause connection failures

Makefile

@@ -193,7 +193,7 @@ generate-mocks: install-mock-generators
 	mockery --name 'API' --dir="./access" --case=underscore --output="./access/mock" --outpkg="mock"
 	mockery --name 'API' --dir="./engine/protocol" --case=underscore --output="./engine/protocol/mock" --outpkg="mock"
 	mockery --name '.*' --dir="./engine/access/state_stream" --case=underscore --output="./engine/access/state_stream/mock" --outpkg="mock"
-	mockery --name 'ConnectionFactory' --dir="./engine/access/rpc/backend" --case=underscore --output="./engine/access/rpc/backend/mock" --outpkg="mock"
+	mockery --name 'ConnectionFactory' --dir="./engine/access/rpc/connection" --case=underscore --output="./engine/access/rpc/connection/mock" --outpkg="mock"
 	mockery --name '.*' --dir=model/fingerprint --case=underscore --output="./model/fingerprint/mock" --outpkg="mock"
 	mockery --name 'ExecForkActor' --structname 'ExecForkActorMock' --dir=module/mempool/consensus/mock/ --case=underscore --output="./module/mempool/consensus/mock/" --outpkg="mock"
 	mockery --name '.*' --dir=engine/verification/fetcher/ --case=underscore --output="./engine/verification/fetcher/mock" --outpkg="mockfetcher"

cmd/access/node_builder/access_node_builder.go

@@ -40,6 +40,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/rest/routes"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
+	rpcConnection "github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	followereng "github.com/onflow/flow-go/engine/common/follower"
 	"github.com/onflow/flow-go/engine/common/requester"
@@ -917,7 +918,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.rpcConf.CollectionAddr,
 				grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(int(builder.rpcConf.MaxMsgSize))),
 				grpc.WithTransportCredentials(insecure.NewCredentials()),
-				backend.WithClientUnaryInterceptor(builder.rpcConf.BackendConfig.CollectionClientTimeout))
+				rpcConnection.WithClientTimeoutOption(builder.rpcConf.BackendConfig.CollectionClientTimeout))
 			if err != nil {
 				return err
 			}
@@ -1047,16 +1048,24 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				return nil, fmt.Errorf("could not initialize backend cache: %w", err)
 			}
 
-			connFactory := &backend.ConnectionFactoryImpl{
+			var connBackendCache *rpcConnection.Cache
+			if backendCache != nil {
+				connBackendCache = rpcConnection.NewCache(backendCache, int(cacheSize))
+			}
+
+			connFactory := &rpcConnection.ConnectionFactoryImpl{
 				CollectionGRPCPort:        builder.collectionGRPCPort,
 				ExecutionGRPCPort:         builder.executionGRPCPort,
 				CollectionNodeGRPCTimeout: backendConfig.CollectionClientTimeout,
 				ExecutionNodeGRPCTimeout:  backendConfig.ExecutionClientTimeout,
-				ConnectionsCache:          backendCache,
-				CacheSize:                 cacheSize,
-				MaxMsgSize:                config.MaxMsgSize,
 				AccessMetrics:             accessMetrics,
 				Log:                       node.Logger,
+				Manager: rpcConnection.NewManager(
+					connBackendCache,
+					node.Logger,
+					accessMetrics,
+					config.MaxMsgSize,
+				),
 			}
 
 			backend := backend.New(

cmd/observer/node_builder/observer_builder.go

@@ -32,6 +32,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/rest/routes"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
+	rpcConnection "github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/follower"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/protocol"
@@ -904,16 +905,24 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			return nil, fmt.Errorf("could not initialize backend cache: %w", err)
 		}
 
-		connFactory := &backend.ConnectionFactoryImpl{
+		var connBackendCache *rpcConnection.Cache
+		if backendCache != nil {
+			connBackendCache = rpcConnection.NewCache(backendCache, int(cacheSize))
+		}
+
+		connFactory := &rpcConnection.ConnectionFactoryImpl{
 			CollectionGRPCPort:        0,
 			ExecutionGRPCPort:         0,
 			CollectionNodeGRPCTimeout: backendConfig.CollectionClientTimeout,
 			ExecutionNodeGRPCTimeout:  backendConfig.ExecutionClientTimeout,
-			ConnectionsCache:          backendCache,
-			CacheSize:                 cacheSize,
-			MaxMsgSize:                config.MaxMsgSize,
 			AccessMetrics:             accessMetrics,
 			Log:                       node.Logger,
+			Manager: rpcConnection.NewManager(
+				connBackendCache,
+				node.Logger,
+				accessMetrics,
+				config.MaxMsgSize,
+			),
 		}
 
 		accessBackend := backend.New(

cmd/util/cmd/execution-state-extract/export_report.json

@@ -1,6 +1,6 @@
 {
- "EpochCounter": 0,
- "PreviousStateCommitment": "1c9f9d343cb8d4610e0b2c1eb74d6ea2f2f8aef2d666281dc22870e3efaa607b",
- "CurrentStateCommitment": "1c9f9d343cb8d4610e0b2c1eb74d6ea2f2f8aef2d666281dc22870e3efaa607b",
- "ReportSucceeded": true
-}
+  "EpochCounter": 0,
+  "PreviousStateCommitment": "1c9f9d343cb8d4610e0b2c1eb74d6ea2f2f8aef2d666281dc22870e3efaa607b",
+  "CurrentStateCommitment": "1c9f9d343cb8d4610e0b2c1eb74d6ea2f2f8aef2d666281dc22870e3efaa607b",
+  "ReportSucceeded": true
+}

engine/access/rpc/backend/backend.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/cmd/build"
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
@@ -78,7 +79,7 @@ type Backend struct {
 	chainID           flow.ChainID
 	collections       storage.Collections
 	executionReceipts storage.ExecutionReceipts
-	connFactory       ConnectionFactory
+	connFactory       connection.ConnectionFactory
 }
 
 // Config defines the configurable options for creating Backend
@@ -104,7 +105,7 @@ func New(
 	executionResults storage.ExecutionResults,
 	chainID flow.ChainID,
 	accessMetrics module.AccessMetrics,
-	connFactory ConnectionFactory,
+	connFactory connection.ConnectionFactory,
 	retryEnabled bool,
 	maxHeightRange uint,
 	preferredExecutionNodeIDs []string,
@@ -224,16 +225,9 @@ func NewCache(
 	var cache *lru.Cache
 	cacheSize := connectionPoolSize
 	if cacheSize > 0 {
-		// TODO: remove this fallback after fixing issues with evictions
-		// It was observed that evictions cause connection errors for in flight requests. This works around
-		// the issue by forcing hte pool size to be greater than the number of ENs + LNs
-		if cacheSize < DefaultConnectionPoolSize {
-			log.Warn().Msg("connection pool size below threshold, setting pool size to default value ")
-			cacheSize = DefaultConnectionPoolSize
-		}
 		var err error
 		cache, err = lru.NewWithEvict(int(cacheSize), func(_, evictedValue interface{}) {
-			store := evictedValue.(*CachedClient)
+			store := evictedValue.(*connection.CachedClient)
 			store.Close()
 			log.Debug().Str("grpc_conn_evicted", store.Address).Msg("closing grpc connection evicted from pool")
 			if accessMetrics != nil {

engine/access/rpc/backend/backend_accounts.go

@@ -5,11 +5,13 @@ import (
 	"time"
 
 	"github.com/hashicorp/go-multierror"
-	execproto "github.com/onflow/flow/protobuf/go/flow/execution"
 	"github.com/rs/zerolog"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	execproto "github.com/onflow/flow/protobuf/go/flow/execution"
+
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
@@ -21,7 +23,7 @@ type backendAccounts struct {
 	state             protocol.State
 	headers           storage.Headers
 	executionReceipts storage.ExecutionReceipts
-	connFactory       ConnectionFactory
+	connFactory       connection.ConnectionFactory
 	log               zerolog.Logger
 }
 

engine/access/rpc/backend/backend_events.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
@@ -24,7 +25,7 @@ type backendEvents struct {
 	headers           storage.Headers
 	executionReceipts storage.ExecutionReceipts
 	state             protocol.State
-	connFactory       ConnectionFactory
+	connFactory       connection.ConnectionFactory
 	log               zerolog.Logger
 	maxHeightRange    uint
 }

engine/access/rpc/backend/backend_scripts.go

@@ -15,6 +15,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
@@ -29,7 +30,7 @@ type backendScripts struct {
 	headers            storage.Headers
 	executionReceipts  storage.ExecutionReceipts
 	state              protocol.State
-	connFactory        ConnectionFactory
+	connFactory        connection.ConnectionFactory
 	log                zerolog.Logger
 	metrics            module.BackendScriptsMetrics
 	loggedScripts      *lru.Cache

engine/access/rpc/backend/backend_test.go

@@ -6,6 +6,8 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
+
 	"github.com/dgraph-io/badger/v2"
 	accessproto "github.com/onflow/flow/protobuf/go/flow/access"
 	entitiesproto "github.com/onflow/flow/protobuf/go/flow/entities"
@@ -2399,7 +2401,7 @@ func (suite *Suite) setupReceipts(block *flow.Block) ([]*flow.ExecutionReceipt,
 	return receipts, ids
 }
 
-func (suite *Suite) setupConnectionFactory() ConnectionFactory {
+func (suite *Suite) setupConnectionFactory() connection.ConnectionFactory {
 	// create a mock connection factory
 	connFactory := new(backendmock.ConnectionFactory)
 	connFactory.On("GetExecutionAPIClient", mock.Anything).Return(suite.execClient, &mockCloser{}, nil)

engine/access/rpc/backend/backend_transactions.go

@@ -15,6 +15,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/onflow/flow-go/access"
+	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/fvm/blueprints"
@@ -37,7 +38,7 @@ type backendTransactions struct {
 	transactionMetrics   module.TransactionMetrics
 	transactionValidator *access.TransactionValidator
 	retry                *Retry
-	connFactory          ConnectionFactory
+	connFactory          connection.ConnectionFactory
 
 	previousAccessNodes []accessproto.AccessAPIClient
 	log                 zerolog.Logger