[Access] Allow streaming from the spork root block

Author: peterargue

engine/access/state_stream/backend/backend.go

@@ -78,6 +78,7 @@ type StateStreamBackend struct {
 	execDataCache        *cache.ExecutionDataCache
 	registers            *execution.RegistersAsyncStore
 	registerRequestLimit int
+	sporkRootBlockHeight uint64
 }
 
 func New(
@@ -108,6 +109,7 @@ func New(
 		execDataCache:        execDataCache,
 		registers:            registers,
 		registerRequestLimit: registerIDsRequestLimit,
+		sporkRootBlockHeight: state.Params().SporkRootBlockHeight(),
 	}
 
 	b.ExecutionDataBackend = ExecutionDataBackend{
@@ -155,6 +157,26 @@ func (b *StateStreamBackend) getExecutionData(ctx context.Context, height uint64
 		return nil, fmt.Errorf("execution data for block %d is not available yet: %w", height, subscription.ErrBlockNotReady)
 	}
 
+	// the spork root block will never have execution data available. If requested, return an empty result.
+	if height == b.sporkRootBlockHeight {
+		sporkRootBlockID, err := b.headers.BlockIDByHeight(b.sporkRootBlockHeight)
+		if err != nil {
+			if !errors.Is(err, storage.ErrNotFound) {
+				return nil, fmt.Errorf("could not get block ID for root block height %d: %w", b.sporkRootBlockHeight, err)
+			}
+			// the spork root block will not exist locally if the node was bootstrapped with a newer
+			// root block. In this case, just use a placeholder. This blockID is only used when serving
+			// data for the root block, which will never happen if it does not exist locally.
+			sporkRootBlockID = flow.ZeroID
+		}
+
+		return &execution_data.BlockExecutionDataEntity{
+			BlockExecutionData: &execution_data.BlockExecutionData{
+				BlockID: sporkRootBlockID,
+			},
+		}, nil
+	}
+
 	execData, err := b.execDataCache.ByHeight(ctx, height)
 	if err != nil {
 		if errors.Is(err, storage.ErrNotFound) ||

engine/access/state_stream/backend/backend_account_statuses_test.go

@@ -128,7 +128,7 @@ func (s *BackendAccountStatusesSuite) subscribeFromStartBlockIdTestCases() []tes
 		{
 			name:            "happy path - all new blocks",
 			highestBackfill: -1, // no backfill
-			startValue:      s.rootBlock.ID(),
+			startValue:      s.blocks[0].ID(),
 		},
 		{
 			name:            "happy path - partial backfill",
@@ -140,11 +140,6 @@ func (s *BackendAccountStatusesSuite) subscribeFromStartBlockIdTestCases() []tes
 			highestBackfill: len(s.blocks) - 1, // backfill all blocks
 			startValue:      s.blocks[0].ID(),
 		},
-		{
-			name:            "happy path - start from root block by id",
-			highestBackfill: len(s.blocks) - 1, // backfill all blocks
-			startValue:      s.rootBlock.ID(),  // start from root block
-		},
 	}
 
 	return s.generateFiltersForTestCases(baseTests)
@@ -156,7 +151,7 @@ func (s *BackendAccountStatusesSuite) subscribeFromStartHeightTestCases() []test
 		{
 			name:            "happy path - all new blocks",
 			highestBackfill: -1, // no backfill
-			startValue:      s.rootBlock.Height,
+			startValue:      s.blocks[0].Height,
 		},
 		{
 			name:            "happy path - partial backfill",
@@ -168,11 +163,6 @@ func (s *BackendAccountStatusesSuite) subscribeFromStartHeightTestCases() []test
 			highestBackfill: len(s.blocks) - 1, // backfill all blocks
 			startValue:      s.blocks[0].Height,
 		},
-		{
-			name:            "happy path - start from root block by id",
-			highestBackfill: len(s.blocks) - 1,  // backfill all blocks
-			startValue:      s.rootBlock.Height, // start from root block
-		},
 	}
 
 	return s.generateFiltersForTestCases(baseTests)
@@ -311,33 +301,20 @@ func (s *BackendAccountStatusesSuite) subscribeToAccountStatuses(
 					s.broadcaster.Publish()
 				}
 
-				expectedEvents := map[string]flow.EventsList{}
-				for _, event := range s.blockEvents[b.ID()] {
-					if test.filters.Match(event) {
-						var address string
-						switch event.Type {
-						case state_stream.CoreEventAccountCreated:
-							address = s.accountCreatedAddress.HexWithPrefix()
-						case state_stream.CoreEventAccountContractAdded:
-							address = s.accountContractAdded.HexWithPrefix()
-						case state_stream.CoreEventAccountContractUpdated:
-							address = s.accountContractUpdated.HexWithPrefix()
-						}
-						expectedEvents[address] = append(expectedEvents[address], event)
-					}
-				}
+				expectedEvents := s.expectedAccountStatuses(b.ID(), test.filters)
 
 				// Consume execution data from subscription
 				unittest.RequireReturnsBefore(s.T(), func() {
 					v, ok := <-sub.Channel()
 					require.True(s.T(), ok, "channel closed while waiting for exec data for block %d %v: err: %v", b.Height, b.ID(), sub.Err())
 
-					resp, ok := v.(*AccountStatusesResponse)
-					require.True(s.T(), ok, "unexpected response type: %T", v)
+					expected := &AccountStatusesResponse{
+						BlockID:       b.ID(),
+						Height:        b.Height,
+						AccountEvents: expectedEvents,
+					}
+					s.requireEventsResponse(v, expected)
 
-					assert.Equal(s.T(), b.ID(), resp.BlockID)
-					assert.Equal(s.T(), b.Height, resp.Height)
-					assert.Equal(s.T(), expectedEvents, resp.AccountEvents)
 				}, 60*time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
 			}
 
@@ -408,6 +385,102 @@ func (s *BackendAccountStatusesSuite) TestSubscribeAccountStatusesFromLatestBloc
 	s.subscribeToAccountStatuses(call, s.subscribeFromLatestTestCases())
 }
 
+// requireEventsResponse ensures that the received event information matches the expected data.
+func (s *BackendAccountStatusesSuite) requireEventsResponse(v interface{}, expected *AccountStatusesResponse) {
+	actual, ok := v.(*AccountStatusesResponse)
+	require.True(s.T(), ok, "unexpected response type: %T", v)
+
+	assert.Equal(s.T(), expected.BlockID, actual.BlockID)
+	assert.Equal(s.T(), expected.Height, actual.Height)
+	assert.Equal(s.T(), expected.AccountEvents, actual.AccountEvents)
+}
+
+// TestSubscribeAccountStatusesFromSporkRootBlock tests that events subscriptions starting from the spork
+// root block return an empty result for the root block.
+func (s *BackendAccountStatusesSuite) TestSubscribeAccountStatusesFromSporkRootBlock() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// setup the backend to have 1 available block
+	s.highestBlockHeader = s.blocks[0].ToHeader()
+
+	rootEventResponse := &AccountStatusesResponse{
+		BlockID:       s.rootBlock.ID(),
+		Height:        s.rootBlock.Height,
+		AccountEvents: map[string]flow.EventsList{},
+	}
+
+	filter, err := state_stream.NewAccountStatusFilter(state_stream.DefaultEventFilterConfig, chainID.Chain(), []string{}, []string{})
+	require.NoError(s.T(), err)
+
+	expectedEvents := s.expectedAccountStatuses(s.blocks[0].ID(), filter)
+	firstEventResponse := &AccountStatusesResponse{
+		BlockID:       s.blocks[0].ID(),
+		Height:        s.blocks[0].Height,
+		AccountEvents: expectedEvents,
+	}
+
+	assertSubscriptionResponses := func(sub subscription.Subscription, cancel context.CancelFunc) {
+		// the first response should have details from the root block and no events
+		resp := <-sub.Channel()
+		s.requireEventsResponse(resp, rootEventResponse)
+
+		// the second response should have details from the first block and its events
+		resp = <-sub.Channel()
+		s.requireEventsResponse(resp, firstEventResponse)
+
+		cancel()
+		resp, ok := <-sub.Channel()
+		assert.False(s.T(), ok)
+		assert.Nil(s.T(), resp)
+		assert.ErrorIs(s.T(), sub.Err(), context.Canceled)
+	}
+
+	s.Run("by height", func() {
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		s.executionDataTracker.On("GetStartHeightFromHeight", s.rootBlock.Height).
+			Return(func(startHeight uint64) (uint64, error) {
+				return s.executionDataTrackerReal.GetStartHeightFromHeight(startHeight)
+			})
+
+		sub := s.backend.SubscribeAccountStatusesFromStartHeight(subCtx, s.rootBlock.Height, filter)
+		assertSubscriptionResponses(sub, subCancel)
+	})
+
+	s.Run("by ID", func() {
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		s.executionDataTracker.On("GetStartHeightFromBlockID", s.rootBlock.ID()).
+			Return(func(startBlockID flow.Identifier) (uint64, error) {
+				return s.executionDataTrackerReal.GetStartHeightFromBlockID(startBlockID)
+			})
+
+		sub := s.backend.SubscribeAccountStatusesFromStartBlockID(subCtx, s.rootBlock.ID(), filter)
+		assertSubscriptionResponses(sub, subCancel)
+	})
+
+	s.Run("by latest", func() {
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		// simulate the case where the latest block is also the root block
+		s.snapshot.On("Head").Unset()
+		s.snapshot.On("Head").Return(s.rootBlock.ToHeader(), nil).Once()
+
+		s.executionDataTracker.On("GetStartHeightFromLatest", mock.Anything).
+			Return(func(ctx context.Context) (uint64, error) {
+				return s.executionDataTrackerReal.GetStartHeightFromLatest(ctx)
+			})
+
+		sub := s.backend.SubscribeAccountStatusesFromLatestBlock(subCtx, filter)
+		assertSubscriptionResponses(sub, subCancel)
+	})
+
+}
+
 // TestSubscribeAccountStatusesHandlesErrors tests handling of expected errors in the SubscribeAccountStatuses.
 func (s *BackendExecutionDataSuite) TestSubscribeAccountStatusesHandlesErrors() {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -455,3 +528,22 @@ func (s *BackendExecutionDataSuite) TestSubscribeAccountStatusesHandlesErrors()
 		assert.Equal(s.T(), codes.NotFound, status.Code(sub.Err()), "expected NotFound, got %v: %v", status.Code(sub.Err()).String(), sub.Err())
 	})
 }
+
+func (s *BackendAccountStatusesSuite) expectedAccountStatuses(blockID flow.Identifier, filter state_stream.AccountStatusFilter) map[string]flow.EventsList {
+	expectedEvents := map[string]flow.EventsList{}
+	for _, event := range s.blockEvents[blockID] {
+		if filter.Match(event) {
+			var address string
+			switch event.Type {
+			case state_stream.CoreEventAccountCreated:
+				address = s.accountCreatedAddress.HexWithPrefix()
+			case state_stream.CoreEventAccountContractAdded:
+				address = s.accountContractAdded.HexWithPrefix()
+			case state_stream.CoreEventAccountContractUpdated:
+				address = s.accountContractUpdated.HexWithPrefix()
+			}
+			expectedEvents[address] = append(expectedEvents[address], event)
+		}
+	}
+	return expectedEvents
+}