[Access] Add integration tests covering the 3 streaming blocks endpoints

cmd/observer/node_builder/observer_builder.go

@@ -1660,6 +1660,19 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			),
 		}
 
+		broadcaster := engine.NewBroadcaster()
+		// create BlockTracker that will track for new blocks (finalized and sealed) and
+		// handles block-related operations.
+		blockTracker, err := subscription.NewBlockTracker(
+			node.State,
+			builder.FinalizedRootBlock.Header.Height,
+			node.Storage.Headers,
+			broadcaster,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize block tracker: %w", err)
+		}
+
 		backendParams := backend.Params{
 			State:                     node.State,
 			Blocks:                    node.Storage.Blocks,
@@ -1678,6 +1691,14 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			Log:                       node.Logger,
 			SnapshotHistoryLimit:      backend.DefaultSnapshotHistoryLimit,
 			Communicator:              backend.NewNodeCommunicator(backendConfig.CircuitBreakerConfig.Enabled),
+			BlockTracker:              blockTracker,
+			SubscriptionHandler: subscription.NewSubscriptionHandler(
+				builder.Logger,
+				broadcaster,
+				builder.stateStreamConf.ClientSendTimeout,
+				builder.stateStreamConf.ResponseLimit,
+				builder.stateStreamConf.ClientSendBufferSize,
+			),
 		}
 
 		if builder.localServiceAPIEnabled {

engine/access/apiproxy/access_api_proxy.go

@@ -347,48 +347,57 @@ func (h *FlowAccessAPIRouter) GetExecutionResultByID(context context.Context, re
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlocksFromStartBlockID(req *access.SubscribeBlocksFromStartBlockIDRequest, server access.AccessAPI_SubscribeBlocksFromStartBlockIDServer) error {
-	// SubscribeBlocksFromStartBlockID is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlocksFromStartBlockID not implemented")
+	err := h.local.SubscribeBlocksFromStartBlockID(req, server)
+	h.log(LocalApiService, "SubscribeBlocksFromStartBlockID", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlocksFromStartHeight(req *access.SubscribeBlocksFromStartHeightRequest, server access.AccessAPI_SubscribeBlocksFromStartHeightServer) error {
-	// SubscribeBlocksFromStartHeight is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlocksFromStartHeight not implemented")
+	err := h.local.SubscribeBlocksFromStartHeight(req, server)
+	h.log(LocalApiService, "SubscribeBlocksFromStartHeight", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlocksFromLatest(req *access.SubscribeBlocksFromLatestRequest, server access.AccessAPI_SubscribeBlocksFromLatestServer) error {
-	// SubscribeBlocksFromLatest is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlocksFromLatest not implemented")
+	err := h.local.SubscribeBlocksFromLatest(req, server)
+	h.log(LocalApiService, "SubscribeBlocksFromLatest", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockHeadersFromStartBlockID(req *access.SubscribeBlockHeadersFromStartBlockIDRequest, server access.AccessAPI_SubscribeBlockHeadersFromStartBlockIDServer) error {
-	// SubscribeBlockHeadersFromStartBlockID is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockHeadersFromStartBlockID not implemented")
+	err := h.local.SubscribeBlockHeadersFromStartBlockID(req, server)
+	h.log(LocalApiService, "SubscribeBlockHeadersFromStartBlockID", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockHeadersFromStartHeight(req *access.SubscribeBlockHeadersFromStartHeightRequest, server access.AccessAPI_SubscribeBlockHeadersFromStartHeightServer) error {
-	// SubscribeBlockHeadersFromStartHeight is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockHeadersFromStartHeight not implemented")
+	err := h.local.SubscribeBlockHeadersFromStartHeight(req, server)
+	h.log(LocalApiService, "SubscribeBlockHeadersFromStartHeight", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockHeadersFromLatest(req *access.SubscribeBlockHeadersFromLatestRequest, server access.AccessAPI_SubscribeBlockHeadersFromLatestServer) error {
-	// SubscribeBlockHeadersFromLatest is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockHeadersFromLatest not implemented")
+	err := h.local.SubscribeBlockHeadersFromLatest(req, server)
+	h.log(LocalApiService, "SubscribeBlockHeadersFromLatest", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockDigestsFromStartBlockID(req *access.SubscribeBlockDigestsFromStartBlockIDRequest, server access.AccessAPI_SubscribeBlockDigestsFromStartBlockIDServer) error {
-	// SubscribeBlockDigestsFromStartBlockID is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockDigestsFromStartBlockID not implemented")
+	err := h.local.SubscribeBlockDigestsFromStartBlockID(req, server)
+	h.log(LocalApiService, "SubscribeBlockDigestsFromStartBlockID", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockDigestsFromStartHeight(req *access.SubscribeBlockDigestsFromStartHeightRequest, server access.AccessAPI_SubscribeBlockDigestsFromStartHeightServer) error {
-	// SubscribeBlockDigestsFromStartHeight is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockDigestsFromStartHeight not implemented")
+	err := h.local.SubscribeBlockDigestsFromStartHeight(req, server)
+	h.log(LocalApiService, "SubscribeBlockDigestsFromStartHeight", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SubscribeBlockDigestsFromLatest(req *access.SubscribeBlockDigestsFromLatestRequest, server access.AccessAPI_SubscribeBlockDigestsFromLatestServer) error {
-	// SubscribeBlockDigestsFromLatest is not implemented for observer yet
-	return status.Errorf(codes.Unimplemented, "method SubscribeBlockDigestsFromLatest not implemented")
+	err := h.local.SubscribeBlockDigestsFromLatest(req, server)
+	h.log(LocalApiService, "SubscribeBlockDigestsFromLatest", err)
+	return err
 }
 
 func (h *FlowAccessAPIRouter) SendAndSubscribeTransactionStatuses(req *access.SendAndSubscribeTransactionStatusesRequest, server access.AccessAPI_SendAndSubscribeTransactionStatusesServer) error {

integration/tests/access/cohort3/grpc_state_stream_test.go

@@ -58,7 +58,7 @@ type GrpcStateStreamSuite struct {
 	net *testnet.FlowNetwork
 
 	// RPC methods to test
-	testedRPCs func() []RPCTest
+	testedRPCs func() []subscribeEventsRPCTest
 }
 
 func (s *GrpcStateStreamSuite) TearDownTest() {
@@ -178,17 +178,17 @@ func (s *GrpcStateStreamSuite) TestHappyPath() {
 
 			testANStream, err := rpc.call(s.ctx, sdkClientTestAN, startValue, &executiondata.EventFilter{})
 			s.Require().NoError(err)
-			testANEvents, testANErrs, err := SubscribeEventsHandler(s.ctx, testANStream)
+			testANEvents, testANErrs, err := SubscribeHandler(s.ctx, testANStream.Recv, eventsResponseHandler)
 			s.Require().NoError(err)
 
 			controlANStream, err := rpc.call(s.ctx, sdkClientControlAN, startValue, &executiondata.EventFilter{})
 			s.Require().NoError(err)
-			controlANEvents, controlANErrs, err := SubscribeEventsHandler(s.ctx, controlANStream)
+			controlANEvents, controlANErrs, err := SubscribeHandler(s.ctx, controlANStream.Recv, eventsResponseHandler)
 			s.Require().NoError(err)
 
 			testONStream, err := rpc.call(s.ctx, sdkClientTestON, startValue, &executiondata.EventFilter{})
 			s.Require().NoError(err)
-			testONEvents, testONErrs, err := SubscribeEventsHandler(s.ctx, testONStream)
+			testONEvents, testONErrs, err := SubscribeHandler(s.ctx, testONStream.Recv, eventsResponseHandler)
 			s.Require().NoError(err)
 
 			if rpc.generateEvents {
@@ -213,7 +213,7 @@ func (s *GrpcStateStreamSuite) TestHappyPath() {
 			foundONTxCount := 0
 			messageIndex := counters.NewMonotonousCounter(0)
 
-			r := newResponseTracker()
+			r := NewResponseTracker(compareEventsResponse)
 
 			for {
 				select {
@@ -226,7 +226,7 @@ func (s *GrpcStateStreamSuite) TestHappyPath() {
 				case event := <-testANEvents:
 					if has(event.Events, targetEvent) {
 						s.T().Logf("adding access test events: %d %d %v", event.Height, len(event.Events), event.Events)
-						r.Add(s.T(), event.Height, "access_test", &event)
+						r.Add(s.T(), event.Height, "access_test", event)
 						foundANTxCount++
 					}
 				case event := <-controlANEvents:
@@ -236,12 +236,12 @@ func (s *GrpcStateStreamSuite) TestHappyPath() {
 						}
 
 						s.T().Logf("adding control events: %d %d %v", event.Height, len(event.Events), event.Events)
-						r.Add(s.T(), event.Height, "access_control", &event)
+						r.Add(s.T(), event.Height, "access_control", event)
 					}
 				case event := <-testONEvents:
 					if has(event.Events, targetEvent) {
 						s.T().Logf("adding observer test events: %d %d %v", event.Height, len(event.Events), event.Events)
-						r.Add(s.T(), event.Height, "observer_test", &event)
+						r.Add(s.T(), event.Height, "observer_test", event)
 						foundONTxCount++
 					}
 				}
@@ -270,14 +270,14 @@ func (s *GrpcStateStreamSuite) generateEvents(client *testnet.Client, txCount in
 	}
 }
 
-type RPCTest struct {
+type subscribeEventsRPCTest struct {
 	name           string
 	call           func(ctx context.Context, client executiondata.ExecutionDataAPIClient, startValue interface{}, filter *executiondata.EventFilter) (executiondata.ExecutionDataAPI_SubscribeEventsClient, error)
 	generateEvents bool // add ability to integration test generate new events or use old events to decrease running test time
 }
 
-func (s *GrpcStateStreamSuite) getRPCs() []RPCTest {
-	return []RPCTest{
+func (s *GrpcStateStreamSuite) getRPCs() []subscribeEventsRPCTest {
+	return []subscribeEventsRPCTest{
 		{
 			name: "SubscribeEventsFromLatest",
 			call: func(ctx context.Context, client executiondata.ExecutionDataAPIClient, _ interface{}, filter *executiondata.EventFilter) (executiondata.ExecutionDataAPI_SubscribeEventsClient, error) {
@@ -330,44 +330,78 @@ func (s *GrpcStateStreamSuite) getRPCs() []RPCTest {
 	}
 }
 
-type ResponseTracker struct {
-	r  map[uint64]map[string]SubscribeEventsResponse
-	mu sync.RWMutex
+// ResponseTracker is a generic tracker for responses.
+type ResponseTracker[T any] struct {
+	r       map[uint64]map[string]T
+	mu      sync.RWMutex
+	compare func(t *testing.T, responses map[uint64]map[string]T, blockHeight uint64) error
 }
 
-func newResponseTracker() *ResponseTracker {
-	return &ResponseTracker{
-		r: make(map[uint64]map[string]SubscribeEventsResponse),
+// NewResponseTracker creates a new ResponseTracker.
+func NewResponseTracker[T any](
+	compare func(t *testing.T, responses map[uint64]map[string]T, blockHeight uint64) error,
+) *ResponseTracker[T] {
+	return &ResponseTracker[T]{
+		r:       make(map[uint64]map[string]T),
+		compare: compare,
 	}
 }
 
-func (r *ResponseTracker) Add(t *testing.T, blockHeight uint64, name string, events *SubscribeEventsResponse) {
+func (r *ResponseTracker[T]) Add(t *testing.T, blockHeight uint64, name string, response T) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
 	if _, ok := r.r[blockHeight]; !ok {
-		r.r[blockHeight] = make(map[string]SubscribeEventsResponse)
+		r.r[blockHeight] = make(map[string]T)
 	}
-	r.r[blockHeight][name] = *events
+	r.r[blockHeight][name] = response
 
-	if len(r.r[blockHeight]) != 3 {
-		return
+	err := r.compare(t, r.r, blockHeight)
+	if err != nil {
+		log.Fatalf("comparison error at block height %d: %v", blockHeight, err)
 	}
 
-	err := r.compare(t, r.r[blockHeight]["access_control"], r.r[blockHeight]["access_test"])
+	delete(r.r, blockHeight)
+}
+
+func eventsResponseHandler(msg *executiondata.SubscribeEventsResponse) (*SubscribeEventsResponse, error) {
+	events := convert.MessagesToEvents(msg.GetEvents())
+
+	return &SubscribeEventsResponse{
+		EventsResponse: backend.EventsResponse{
+			Height:         msg.GetBlockHeight(),
+			BlockID:        convert.MessageToIdentifier(msg.GetBlockId()),
+			Events:         events,
+			BlockTimestamp: msg.GetBlockTimestamp().AsTime(),
+		},
+		MessageIndex: msg.MessageIndex,
+	}, nil
+}
+
+func compareEventsResponse(t *testing.T, responses map[uint64]map[string]*SubscribeEventsResponse, blockHeight uint64) error {
+	if len(responses[blockHeight]) != 3 {
+		return nil
+	}
+	accessControlData := responses[blockHeight]["access_control"]
+	accessTestData := responses[blockHeight]["access_test"]
+	observerTestData := responses[blockHeight]["observer_test"]
+
+	// Compare access_control with access_test
+	err := compareEvents(t, accessControlData, accessTestData)
 	if err != nil {
-		log.Fatalf("failure comparing access and access data %d: %v", blockHeight, err)
+		return fmt.Errorf("failure comparing access and access data: %d: %v", blockHeight, err)
 	}
 
-	err = r.compare(t, r.r[blockHeight]["access_control"], r.r[blockHeight]["observer_test"])
+	// Compare access_control with observer_test
+	err = compareEvents(t, accessControlData, observerTestData)
 	if err != nil {
-		log.Fatalf("failure comparing access and observer data %d: %v", blockHeight, err)
+		return fmt.Errorf("failure comparing access and observer data: %d: %v", blockHeight, err)
 	}
 
-	delete(r.r, blockHeight)
+	return nil
 }
 
-func (r *ResponseTracker) compare(t *testing.T, controlData SubscribeEventsResponse, testData SubscribeEventsResponse) error {
+func compareEvents(t *testing.T, controlData, testData *SubscribeEventsResponse) error {
 	require.Equal(t, controlData.BlockID, testData.BlockID)
 	require.Equal(t, controlData.Height, testData.Height)
 	require.Equal(t, controlData.BlockTimestamp, testData.BlockTimestamp)
@@ -396,11 +430,12 @@ func getClient(address string) (executiondata.ExecutionDataAPIClient, error) {
 	return executiondata.NewExecutionDataAPIClient(conn), nil
 }
 
-func SubscribeEventsHandler(
+func SubscribeHandler[T any, V any](
 	ctx context.Context,
-	stream executiondata.ExecutionDataAPI_SubscribeEventsClient,
-) (<-chan SubscribeEventsResponse, <-chan error, error) {
-	sub := make(chan SubscribeEventsResponse)
+	recv func() (T, error),
+	responseHandler func(T) (V, error),
+) (<-chan V, <-chan error, error) {
+	sub := make(chan V)
 	errChan := make(chan error)
 
 	sendErr := func(err error) {
@@ -415,26 +450,20 @@ func SubscribeEventsHandler(
 		defer close(errChan)
 
 		for {
-			resp, err := stream.Recv()
+			resp, err := recv()
 			if err != nil {
 				if err == io.EOF {
 					return
 				}
 
-				sendErr(fmt.Errorf("error receiving event: %w", err))
+				sendErr(fmt.Errorf("error receiving response: %w", err))
 				return
 			}
 
-			events := convert.MessagesToEvents(resp.GetEvents())
-
-			response := SubscribeEventsResponse{
-				EventsResponse: backend.EventsResponse{
-					Height:         resp.GetBlockHeight(),
-					BlockID:        convert.MessageToIdentifier(resp.GetBlockId()),
-					Events:         events,
-					BlockTimestamp: resp.GetBlockTimestamp().AsTime(),
-				},
-				MessageIndex: resp.MessageIndex,
+			response, err := responseHandler(resp)
+			if err != nil {
+				sendErr(fmt.Errorf("error converting response: %w", err))
+				return
 			}
 
 			select {