[Access] Implement gRPC streaming endpoint SubscribeBlocks

access/api.go

@@ -6,6 +6,7 @@ import (
 	"github.com/onflow/flow/protobuf/go/flow/access"
 	"github.com/onflow/flow/protobuf/go/flow/entities"
 
+	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 )
@@ -52,6 +53,10 @@ type API interface {
 
 	GetExecutionResultForBlockID(ctx context.Context, blockID flow.Identifier) (*flow.ExecutionResult, error)
 	GetExecutionResultByID(ctx context.Context, id flow.Identifier) (*flow.ExecutionResult, error)
+
+	// SubscribeBlocks subscribes to the blocks starting from a specific block ID and block height, with an optional block status.
+	// By default, returns sealed blocks.
+	SubscribeBlocks(ctx context.Context, startBlockID flow.Identifier, startHeight uint64, blockStatus flow.BlockStatus) subscription.Subscription
 }
 
 // TODO: Combine this with flow.TransactionResult?

access/handler.go

@@ -2,12 +2,16 @@ package access
 
 import (
 	"context"
+	"fmt"
+	"sync/atomic"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"github.com/onflow/flow-go/consensus/hotstuff"
 	"github.com/onflow/flow-go/consensus/hotstuff/signature"
+	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
@@ -17,6 +21,7 @@ import (
 )
 
 type Handler struct {
+	subscription.StreamingData
 	api                  API
 	chain                flow.Chain
 	signerIndicesDecoder hotstuff.BlockSignerDecoder
@@ -29,8 +34,17 @@ type HandlerOption func(*Handler)
 
 var _ access.AccessAPIServer = (*Handler)(nil)
 
-func NewHandler(api API, chain flow.Chain, finalizedHeader module.FinalizedHeaderCache, me module.Local, options ...HandlerOption) *Handler {
+func NewHandler(api API,
+	chain flow.Chain,
+	finalizedHeader module.FinalizedHeaderCache,
+	me module.Local,
+	maxStreams uint32,
+	options ...HandlerOption) *Handler {
 	h := &Handler{
+		StreamingData: subscription.StreamingData{
+			MaxStreams:  int32(maxStreams),
+			StreamCount: atomic.Int32{},
+		},
 		api:                  api,
 		chain:                chain,
 		finalizedHeaderCache: finalizedHeader,
@@ -701,6 +715,66 @@ func (h *Handler) GetExecutionResultByID(ctx context.Context, req *access.GetExe
 	}, nil
 }
 
+// SubscribeBlocks handles subscription requests for blocks.
+// It takes a SubscribeBlocksRequest and an AccessAPI_SubscribeBlocksServer stream as input.
+// The handler manages the subscription to block updates and sends the subscribed block information
+// to the client via the provided stream.
+func (h *Handler) SubscribeBlocks(request *access.SubscribeBlocksRequest, stream access.AccessAPI_SubscribeBlocksServer) error {
+	// check if the maximum number of streams is reached
+	if h.StreamCount.Load() >= h.MaxStreams {
+		return status.Errorf(codes.ResourceExhausted, "maximum number of streams reached")
+	}
+	h.StreamCount.Add(1)
+	defer h.StreamCount.Add(-1)
+
+	startBlockID := flow.ZeroID
+	if request.GetStartBlockId() != nil {
+		blockID, err := convert.BlockID(request.GetStartBlockId())
+		if err != nil {
+			return status.Errorf(codes.InvalidArgument, "could not convert start block ID: %v", err)
+		}
+		startBlockID = blockID
+	}
+
+	blockStatus := convert.MessageToBlockStatus(request.BlockStatus)
+	err := checkBlockStatus(blockStatus)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "invalid block status argument: %v", err)
+	}
+
+	sub := h.api.SubscribeBlocks(stream.Context(), startBlockID, request.GetStartBlockHeight(), blockStatus)
+	for {
+		v, ok := <-sub.Channel()
+		if !ok {
+			if sub.Err() != nil {
+				return rpc.ConvertError(sub.Err(), "stream encountered an error", codes.Internal)
+			}
+			return nil
+		}
+
+		blockResp, ok := v.(*flow.Block)
+		if !ok {
+			return status.Errorf(codes.Internal, "unexpected response type: %T", v)
+		}
+
+		resp, err := h.blockResponse(blockResp, request.GetFullBlockResponse(), blockStatus)
+		if err != nil {
+			return rpc.ConvertError(err, "could not convert block to message", codes.Internal)
+		}
+
+		err = stream.Send(&access.SubscribeBlocksResponse{
+			Block: resp.Block,
+		})
+		if err != nil {
+			return rpc.ConvertError(err, "could not send response", codes.Internal)
+		}
+	}
+}
+
+func (h *Handler) SendAndSubscribeTransactionStatuses(request *access.SendAndSubscribeTransactionStatusesRequest, stream access.AccessAPI_SendAndSubscribeTransactionStatusesServer) error {
+	panic("not implemented!")
+}
+
 func (h *Handler) blockResponse(block *flow.Block, fullResponse bool, status flow.BlockStatus) (*access.BlockResponse, error) {
 	metadata := h.buildMetadataResponse()
 
@@ -777,3 +851,10 @@ func WithBlockSignerDecoder(signerIndicesDecoder hotstuff.BlockSignerDecoder) fu
 		handler.signerIndicesDecoder = signerIndicesDecoder
 	}
 }
+
+func checkBlockStatus(blockStatus flow.BlockStatus) error {
+	if blockStatus == flow.BlockStatusUnknown {
+		return fmt.Errorf("block status is unlnown. Possible variants: BLOCK_FINALIZED, BLOCK_SEALED")
+	}
+	return nil
+}

cmd/access/node_builder/access_node_builder.go

@@ -45,6 +45,7 @@ import (
 	rpcConnection "github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	statestreambackend "github.com/onflow/flow-go/engine/access/state_stream/backend"
+	"github.com/onflow/flow-go/engine/access/subscription"
 	followereng "github.com/onflow/flow-go/engine/common/follower"
 	"github.com/onflow/flow-go/engine/common/requester"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
@@ -84,7 +85,7 @@ import (
 	"github.com/onflow/flow-go/network/p2p/p2pbuilder"
 	p2pconfig "github.com/onflow/flow-go/network/p2p/p2pbuilder/config"
 	"github.com/onflow/flow-go/network/p2p/p2pnet"
-	"github.com/onflow/flow-go/network/p2p/subscription"
+	networkingsubscription "github.com/onflow/flow-go/network/p2p/subscription"
 	"github.com/onflow/flow-go/network/p2p/translator"
 	"github.com/onflow/flow-go/network/p2p/unicast/protocols"
 	relaynet "github.com/onflow/flow-go/network/relay"
@@ -147,6 +148,7 @@ type AccessNodeConfig struct {
 	registersDBPath              string
 	checkpointFile               string
 	scriptExecutorConfig         query.QueryConfig
+	broadcaster                  *engine.Broadcaster
 }
 
 type PublicNetworkConfig struct {
@@ -195,14 +197,14 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		},
 		stateStreamConf: statestreambackend.Config{
 			MaxExecutionDataMsgSize: grpcutils.DefaultMaxMsgSize,
-			ExecutionDataCacheSize:  state_stream.DefaultCacheSize,
-			ClientSendTimeout:       state_stream.DefaultSendTimeout,
-			ClientSendBufferSize:    state_stream.DefaultSendBufferSize,
-			MaxGlobalStreams:        state_stream.DefaultMaxGlobalStreams,
+			ExecutionDataCacheSize:  subscription.DefaultCacheSize,
+			ClientSendTimeout:       subscription.DefaultSendTimeout,
+			ClientSendBufferSize:    subscription.DefaultSendBufferSize,
+			MaxGlobalStreams:        subscription.DefaultMaxGlobalStreams,
 			EventFilterConfig:       state_stream.DefaultEventFilterConfig,
-			ResponseLimit:           state_stream.DefaultResponseLimit,
-			HeartbeatInterval:       state_stream.DefaultHeartbeatInterval,
 			RegisterIDsRequestLimit: state_stream.DefaultRegisterIDsRequestLimit,
+			ResponseLimit:           subscription.DefaultResponseLimit,
+			HeartbeatInterval:       subscription.DefaultHeartbeatInterval,
 		},
 		stateStreamFilterConf:        nil,
 		ExecutionNodeAddress:         "localhost:9000",
@@ -236,6 +238,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		registersDBPath:              filepath.Join(homedir, ".flow", "execution_state"),
 		checkpointFile:               cmd.NotSet,
 		scriptExecutorConfig:         query.NewDefaultConfig(),
+		broadcaster:                  nil,
 	}
 }
 
@@ -811,7 +814,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 			if err != nil {
 				return nil, fmt.Errorf("could not get highest consecutive height: %w", err)
 			}
-			broadcaster := engine.NewBroadcaster()
+			builder.broadcaster = engine.NewBroadcaster()
 
 			builder.stateStreamBackend, err = statestreambackend.New(
 				node.Logger,
@@ -822,7 +825,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				node.Storage.Results,
 				builder.ExecutionDataStore,
 				executionDataStoreCache,
-				broadcaster,
+				builder.broadcaster,
 				builder.executionDataConfig.InitialBlockHeight,
 				highestAvailableHeight,
 				builder.RegistersAsyncStore)
@@ -838,7 +841,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				node.RootChainID,
 				builder.stateStreamGrpcServer,
 				builder.stateStreamBackend,
-				broadcaster,
+				builder.broadcaster,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream engine: %w", err)
@@ -1439,6 +1442,11 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				return nil, fmt.Errorf("could not parse script execution mode: %w", err)
 			}
 
+			highestAvailableHeight, err := builder.ExecutionDataRequester.HighestConsecutiveHeight()
+			if err != nil {
+				return nil, fmt.Errorf("could not get highest consecutive height: %w", err)
+			}
+
 			nodeBackend, err := backend.New(backend.Params{
 				State:                     node.State,
 				CollectionRPC:             builder.CollectionRPC,
@@ -1463,6 +1471,15 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				TxErrorMessagesCacheSize:  builder.TxErrorMessagesCacheSize,
 				ScriptExecutor:            builder.ScriptExecutor,
 				ScriptExecutionMode:       scriptExecMode,
+				SubscriptionParams: backend.SubscriptionParams{
+					Broadcaster:            builder.broadcaster,
+					SendTimeout:            builder.stateStreamConf.ClientSendTimeout,
+					ResponseLimit:          builder.stateStreamConf.ResponseLimit,
+					SendBufferSize:         int(builder.stateStreamConf.ClientSendBufferSize),
+					RootHeight:             builder.executionDataConfig.InitialBlockHeight,
+					HighestAvailableHeight: highestAvailableHeight,
+					Seals:                  node.Storage.Seals,
+				},
 			})
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize backend: %w", err)
@@ -1725,7 +1742,7 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 			Unicast: builder.FlowConfig.NetworkConfig.Unicast,
 		}).
 		SetBasicResolver(builder.Resolver).
-		SetSubscriptionFilter(subscription.NewRoleBasedFilter(flow.RoleAccess, builder.IdentityProvider)).
+		SetSubscriptionFilter(networkingsubscription.NewRoleBasedFilter(flow.RoleAccess, builder.IdentityProvider)).
 		SetConnectionManager(connManager).
 		SetRoutingSystem(func(ctx context.Context, h host.Host) (routing.Routing, error) {
 			return dht.NewDHT(ctx, h, protocols.FlowPublicDHTProtocolID(builder.SporkID), builder.Logger, networkMetrics, dht.AsServer())

engine/access/access_test.go

@@ -27,6 +27,7 @@ import (
 	accessmock "github.com/onflow/flow-go/engine/access/mock"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	connectionmock "github.com/onflow/flow-go/engine/access/rpc/connection/mock"
+	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/factory"
@@ -169,6 +170,7 @@ func (suite *Suite) RunTest(
 			suite.chainID.Chain(),
 			suite.finalizedHeaderCache,
 			suite.me,
+			subscription.DefaultMaxGlobalStreams,
 			access.WithBlockSignerDecoder(suite.signerIndicesDecoder),
 		)
 		f(handler, db, all)
@@ -336,7 +338,7 @@ func (suite *Suite) TestSendTransactionToRandomCollectionNode() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me)
+		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// Send transaction 1
 		resp, err := handler.SendTransaction(context.Background(), sendReq1)
@@ -658,7 +660,7 @@ func (suite *Suite) TestGetSealedTransaction() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me)
+		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// create the ingest engine
 		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections,
@@ -796,7 +798,7 @@ func (suite *Suite) TestGetTransactionResult() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me)
+		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// create the ingest engine
 		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections,
@@ -989,7 +991,7 @@ func (suite *Suite) TestExecuteScript() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(suite.backend, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me)
+		handler := access.NewHandler(suite.backend, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// initialize metrics related storage
 		metrics := metrics.NewNoopCollector()

engine/access/apiproxy/access_api_proxy.go

@@ -230,6 +230,14 @@ func (h *FlowAccessAPIRouter) GetExecutionResultByID(context context.Context, re
 	return res, err
 }
 
+func (h *FlowAccessAPIRouter) SubscribeBlocks(req *access.SubscribeBlocksRequest, server access.AccessAPI_SubscribeBlocksServer) error {
+	panic("not implemented")
+}
+
+func (h *FlowAccessAPIRouter) SendAndSubscribeTransactionStatuses(req *access.SendAndSubscribeTransactionStatusesRequest, server access.AccessAPI_SendAndSubscribeTransactionStatusesServer) error {
+	panic("not implemented")
+}
+
 // FlowAccessAPIForwarder forwards all requests to a set of upstream access nodes or observers
 type FlowAccessAPIForwarder struct {
 	*forwarder.Forwarder

engine/access/integration_unsecure_grpc_server_test.go

@@ -22,8 +22,8 @@ import (
 	accessmock "github.com/onflow/flow-go/engine/access/mock"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
-	"github.com/onflow/flow-go/engine/access/state_stream"
 	statestreambackend "github.com/onflow/flow-go/engine/access/state_stream/backend"
+	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/blobs"
 	"github.com/onflow/flow-go/module/execution"
@@ -118,7 +118,7 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 
 	suite.broadcaster = engine.NewBroadcaster()
 
-	suite.execDataHeroCache = herocache.NewBlockExecutionData(state_stream.DefaultCacheSize, suite.log, metrics.NewNoopCollector())
+	suite.execDataHeroCache = herocache.NewBlockExecutionData(subscription.DefaultCacheSize, suite.log, metrics.NewNoopCollector())
 	suite.execDataCache = cache.NewExecutionDataCache(suite.eds, suite.headers, suite.seals, suite.results, suite.execDataHeroCache)
 
 	accessIdentity := unittest.IdentityFixture(unittest.WithRole(flow.RoleAccess))
@@ -232,8 +232,8 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 	).Maybe()
 
 	conf := statestreambackend.Config{
-		ClientSendTimeout:    state_stream.DefaultSendTimeout,
-		ClientSendBufferSize: state_stream.DefaultSendBufferSize,
+		ClientSendTimeout:    subscription.DefaultSendTimeout,
+		ClientSendBufferSize: subscription.DefaultSendBufferSize,
 	}
 
 	stateStreamBackend, err := statestreambackend.New(