Merge pull request #7082 from onflow/illia-malachyn/7080-add-heatbeat-interval-to-providers

[Access] Add heartbeat interval to providers

Author: peterargue

Diff for: engine/access/rest/websockets/data_providers/account_statuses_provider.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/engine/access/rest/http/request"
+	"github.com/onflow/flow-go/engine/access/rest/util"
 	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
@@ -20,9 +21,10 @@ import (
 
 // accountStatusesArguments contains the arguments required for subscribing to account statuses
 type accountStatusesArguments struct {
-	StartBlockID     flow.Identifier                  // ID of the block to start subscription from
-	StartBlockHeight uint64                           // Height of the block to start subscription from
-	Filter           state_stream.AccountStatusFilter // Filter applied to events for a given subscription
+	StartBlockID      flow.Identifier                  // ID of the block to start subscription from
+	StartBlockHeight  uint64                           // Height of the block to start subscription from
+	Filter            state_stream.AccountStatusFilter // Filter applied to events for a given subscription
+	HeartbeatInterval *uint64                          // Maximum number of blocks message won't be sent. Nil if not set
 }
 
 type AccountStatusesDataProvider struct {
@@ -64,6 +66,9 @@ func NewAccountStatusesDataProvider(
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments for account statuses data provider: %w", err)
 	}
+	if accountStatusesArgs.HeartbeatInterval != nil {
+		p.heartbeatInterval = *accountStatusesArgs.HeartbeatInterval
+	}
 
 	subCtx, cancel := context.WithCancel(ctx)
 
@@ -140,10 +145,22 @@ func parseAccountStatusesArguments(
 	chain flow.Chain,
 	eventFilterConfig state_stream.EventFilterConfig,
 ) (accountStatusesArguments, error) {
+	allowedFields := []string{
+		"start_block_id",
+		"start_block_height",
+		"event_types",
+		"account_addresses",
+		"heartbeat_interval",
+	}
+	err := ensureAllowedFields(arguments, allowedFields)
+	if err != nil {
+		return accountStatusesArguments{}, err
+	}
+
 	var args accountStatusesArguments
 
 	// Parse block arguments
-	startBlockID, startBlockHeight, err := ParseStartBlock(arguments)
+	startBlockID, startBlockHeight, err := parseStartBlock(arguments)
 	if err != nil {
 		return args, err
 	}
@@ -155,12 +172,12 @@ func parseAccountStatusesArguments(
 	if eventTypesIn, ok := arguments["event_types"]; ok && eventTypesIn != "" {
 		result, ok := eventTypesIn.([]string)
 		if !ok {
-			return args, fmt.Errorf("'event_types' must be an array of string")
+			return accountStatusesArguments{}, fmt.Errorf("'event_types' must be an array of string")
 		}
 
 		err := eventTypes.Parse(result)
 		if err != nil {
-			return args, fmt.Errorf("invalid 'event_types': %w", err)
+			return accountStatusesArguments{}, fmt.Errorf("invalid 'event_types': %w", err)
 		}
 	}
 
@@ -169,14 +186,29 @@ func parseAccountStatusesArguments(
 	if accountAddressesIn, ok := arguments["account_addresses"]; ok && accountAddressesIn != "" {
 		accountAddresses, ok = accountAddressesIn.([]string)
 		if !ok {
-			return args, fmt.Errorf("'account_addresses' must be an array of string")
+			return accountStatusesArguments{}, fmt.Errorf("'account_addresses' must be an array of string")
 		}
 	}
 
+	var heartbeatInterval uint64
+	if heartbeatIntervalIn, ok := arguments["heartbeat_interval"]; ok && heartbeatIntervalIn != "" {
+		result, ok := heartbeatIntervalIn.(string)
+		if !ok {
+			return accountStatusesArguments{}, fmt.Errorf("'heartbeat_interval' must be a string")
+		}
+
+		heartbeatInterval, err = util.ToUint64(result)
+		if err != nil {
+			return accountStatusesArguments{}, fmt.Errorf("invalid 'heartbeat_interval': %w", err)
+		}
+
+		args.HeartbeatInterval = &heartbeatInterval
+	}
+
 	// Initialize the event filter with the parsed arguments
 	args.Filter, err = state_stream.NewAccountStatusFilter(eventFilterConfig, chain, eventTypes.Flow(), accountAddresses)
 	if err != nil {
-		return args, fmt.Errorf("failed to create event filter: %w", err)
+		return accountStatusesArguments{}, fmt.Errorf("failed to create event filter: %w", err)
 	}
 
 	return args, nil

Diff for: engine/access/rest/websockets/data_providers/args_validation.go

@@ -0,0 +1,19 @@
+package data_providers
+
+import (
+	"fmt"
+)
+
+func ensureAllowedFields(data map[string]interface{}, allowedFields []string) error {
+	fieldsMap := make(map[string]bool, len(allowedFields))
+	for _, field := range allowedFields {
+		fieldsMap[field] = true
+	}
+
+	for key := range data {
+		if !fieldsMap[key] {
+			return fmt.Errorf("unexpected field: '%s'", key)
+		}
+	}
+	return nil
+}

Diff for: engine/access/rest/websockets/data_providers/block_digests_provider.go

@@ -39,7 +39,7 @@ func NewBlockDigestsDataProvider(
 	}
 
 	// Parse arguments passed to the provider.
-	blockArgs, err := ParseBlocksArguments(arguments)
+	blockArgs, err := parseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}

Diff for: engine/access/rest/websockets/data_providers/block_headers_provider.go

@@ -40,7 +40,7 @@ func NewBlockHeadersDataProvider(
 	}
 
 	// Parse arguments passed to the provider.
-	blockArgs, err := ParseBlocksArguments(arguments)
+	blockArgs, err := parseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}

Diff for: engine/access/rest/websockets/data_providers/blocks_provider.go

@@ -54,7 +54,7 @@ func NewBlocksDataProvider(
 
 	// Parse arguments passed to the provider.
 	var err error
-	p.arguments, err = ParseBlocksArguments(arguments)
+	p.arguments, err = parseBlocksArguments(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments: %w", err)
 	}
@@ -108,37 +108,47 @@ func (p *BlocksDataProvider) createSubscription(ctx context.Context, args blocks
 	return p.api.SubscribeBlocksFromLatest(ctx, args.BlockStatus)
 }
 
-// ParseBlocksArguments validates and initializes the blocks arguments.
-func ParseBlocksArguments(arguments models.Arguments) (blocksArguments, error) {
+// parseBlocksArguments validates and initializes the blocks arguments.
+func parseBlocksArguments(arguments models.Arguments) (blocksArguments, error) {
+	allowedFields := []string{
+		"start_block_id",
+		"start_block_height",
+		"block_status",
+	}
+	err := ensureAllowedFields(arguments, allowedFields)
+	if err != nil {
+		return blocksArguments{}, err
+	}
+
 	var args blocksArguments
 
 	// Parse 'block_status'
 	if blockStatusIn, ok := arguments["block_status"]; ok {
 		result, ok := blockStatusIn.(string)
 		if !ok {
-			return args, fmt.Errorf("'block_status' must be string")
+			return blocksArguments{}, fmt.Errorf("'block_status' must be string")
 		}
 		blockStatus, err := parser.ParseBlockStatus(result)
 		if err != nil {
-			return args, err
+			return blocksArguments{}, err
 		}
 		args.BlockStatus = blockStatus
 	} else {
-		return args, fmt.Errorf("'block_status' must be provided")
+		return blocksArguments{}, fmt.Errorf("'block_status' must be provided")
 	}
 
 	// Parse block arguments
-	startBlockID, startBlockHeight, err := ParseStartBlock(arguments)
+	startBlockID, startBlockHeight, err := parseStartBlock(arguments)
 	if err != nil {
-		return args, err
+		return blocksArguments{}, err
 	}
 	args.StartBlockID = startBlockID
 	args.StartBlockHeight = startBlockHeight
 
 	return args, nil
 }
 
-func ParseStartBlock(arguments models.Arguments) (flow.Identifier, uint64, error) {
+func parseStartBlock(arguments models.Arguments) (flow.Identifier, uint64, error) {
 	startBlockIDIn, hasStartBlockID := arguments["start_block_id"]
 	startBlockHeightIn, hasStartBlockHeight := arguments["start_block_height"]
 

Diff for: engine/access/rest/websockets/data_providers/blocks_provider_test.go

@@ -217,6 +217,7 @@ func (s *BlocksProviderSuite) expectedBlockResponses(
 // 1. Missing 'block_status' argument.
 // 2. Invalid 'block_status' argument.
 // 3. Providing both 'start_block_id' and 'start_block_height' simultaneously.
+// 4. Providing unexpected argument.
 func (s *BlocksProviderSuite) TestBlocksDataProvider_InvalidArguments() {
 	ctx := context.Background()
 	send := make(chan interface{})
@@ -239,6 +240,7 @@ func (s *BlocksProviderSuite) TestBlocksDataProvider_InvalidArguments() {
 // 1. Missing the required 'block_status' argument.
 // 2. Providing an unknown or invalid 'block_status' value.
 // 3. Supplying both 'start_block_id' and 'start_block_height' simultaneously, which is not allowed.
+// 4. Providing unexpected argument.
 func (s *BlocksProviderSuite) invalidArgumentsTestCases() []testErrType {
 	return []testErrType{
 		{
@@ -264,5 +266,14 @@ func (s *BlocksProviderSuite) invalidArgumentsTestCases() []testErrType {
 			},
 			expectedErrorMsg: "can only provide either 'start_block_id' or 'start_block_height'",
 		},
+		{
+			name: "unexpected argument",
+			arguments: map[string]interface{}{
+				"block_status":        parser.Finalized,
+				"start_block_id":      unittest.BlockFixture().ID().String(),
+				"unexpected_argument": "dummy",
+			},
+			expectedErrorMsg: "unexpected field: 'unexpected_argument'",
+		},
 	}
 }

Diff for: engine/access/rest/websockets/data_providers/events_provider.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/engine/access/rest/http/request"
+	"github.com/onflow/flow-go/engine/access/rest/util"
 	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
@@ -16,11 +17,12 @@ import (
 	"github.com/onflow/flow-go/module/counters"
 )
 
-// eventsArguments contains the arguments required for subscribing to events
+// eventsArguments contains the arguments a user passes to subscribe to events
 type eventsArguments struct {
-	StartBlockID     flow.Identifier          // ID of the block to start subscription from
-	StartBlockHeight uint64                   // Height of the block to start subscription from
-	Filter           state_stream.EventFilter // Filter applied to events for a given subscription
+	StartBlockID      flow.Identifier          // ID of the block to start subscription from
+	StartBlockHeight  uint64                   // Height of the block to start subscription from
+	Filter            state_stream.EventFilter // Filter applied to events for a given subscription
+	HeartbeatInterval *uint64                  // Maximum number of blocks message won't be sent. Nil if not set
 }
 
 // EventsDataProvider is responsible for providing events
@@ -63,6 +65,9 @@ func NewEventsDataProvider(
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments for events data provider: %w", err)
 	}
+	if eventArgs.HeartbeatInterval != nil {
+		p.heartbeatInterval = *eventArgs.HeartbeatInterval
+	}
 
 	subCtx, cancel := context.WithCancel(ctx)
 
@@ -139,10 +144,23 @@ func parseEventsArguments(
 	chain flow.Chain,
 	eventFilterConfig state_stream.EventFilterConfig,
 ) (eventsArguments, error) {
+	allowedFields := []string{
+		"start_block_id",
+		"start_block_height",
+		"event_types",
+		"addresses",
+		"contracts",
+		"heartbeat_interval",
+	}
+	err := ensureAllowedFields(arguments, allowedFields)
+	if err != nil {
+		return eventsArguments{}, err
+	}
+
 	var args eventsArguments
 
 	// Parse block arguments
-	startBlockID, startBlockHeight, err := ParseStartBlock(arguments)
+	startBlockID, startBlockHeight, err := parseStartBlock(arguments)
 	if err != nil {
 		return args, err
 	}
@@ -154,12 +172,12 @@ func parseEventsArguments(
 	if eventTypesIn, ok := arguments["event_types"]; ok && eventTypesIn != "" {
 		result, ok := eventTypesIn.([]string)
 		if !ok {
-			return args, fmt.Errorf("'event_types' must be an array of string")
+			return eventsArguments{}, fmt.Errorf("'event_types' must be an array of string")
 		}
 
 		err := eventTypes.Parse(result)
 		if err != nil {
-			return args, fmt.Errorf("invalid 'event_types': %w", err)
+			return eventsArguments{}, fmt.Errorf("invalid 'event_types': %w", err)
 		}
 	}
 
@@ -168,7 +186,7 @@ func parseEventsArguments(
 	if addressesIn, ok := arguments["addresses"]; ok && addressesIn != "" {
 		addresses, ok = addressesIn.([]string)
 		if !ok {
-			return args, fmt.Errorf("'addresses' must be an array of string")
+			return eventsArguments{}, fmt.Errorf("'addresses' must be an array of string")
 		}
 	}
 
@@ -177,14 +195,29 @@ func parseEventsArguments(
 	if contractsIn, ok := arguments["contracts"]; ok && contractsIn != "" {
 		contracts, ok = contractsIn.([]string)
 		if !ok {
-			return args, fmt.Errorf("'contracts' must be an array of string")
+			return eventsArguments{}, fmt.Errorf("'contracts' must be an array of string")
 		}
 	}
 
+	var heartbeatInterval uint64
+	if heartbeatIntervalIn, ok := arguments["heartbeat_interval"]; ok && heartbeatIntervalIn != "" {
+		result, ok := heartbeatIntervalIn.(string)
+		if !ok {
+			return eventsArguments{}, fmt.Errorf("'heartbeat_interval' must be a string")
+		}
+
+		heartbeatInterval, err = util.ToUint64(result)
+		if err != nil {
+			return eventsArguments{}, fmt.Errorf("invalid 'heartbeat_interval': %w", err)
+		}
+
+		args.HeartbeatInterval = &heartbeatInterval
+	}
+
 	// Initialize the event filter with the parsed arguments
 	args.Filter, err = state_stream.NewEventFilter(eventFilterConfig, chain, eventTypes.Flow(), addresses, contracts)
 	if err != nil {
-		return args, fmt.Errorf("failed to create event filter: %w", err)
+		return eventsArguments{}, fmt.Errorf("failed to create event filter: %w", err)
 	}
 
 	return args, nil

Diff for: engine/access/rest/websockets/data_providers/events_provider_test.go

@@ -330,6 +330,8 @@ func (s *EventsProviderSuite) TestEventsDataProvider_StateStreamNotConfigured()
 // 1. Supplying both 'start_block_id' and 'start_block_height' simultaneously, which is not allowed.
 // 2. Providing invalid 'start_block_id' value.
 // 3. Providing invalid 'start_block_height' value.
+// 4. Providing invalid 'heartbeat_interval' value.
+// 5. Providing unexpected argument.
 func invalidArgumentsTestCases() []testErrType {
 	return []testErrType{
 		{
@@ -354,5 +356,20 @@ func invalidArgumentsTestCases() []testErrType {
 			},
 			expectedErrorMsg: "value must be an unsigned 64 bit integer",
 		},
+		{
+			name: "invalid 'heartbeat_interval' argument",
+			arguments: map[string]interface{}{
+				"heartbeat_interval": "-1",
+			},
+			expectedErrorMsg: "value must be an unsigned 64 bit integer",
+		},
+		{
+			name: "unexpected argument",
+			arguments: map[string]interface{}{
+				"start_block_id":      unittest.BlockFixture().ID().String(),
+				"unexpected_argument": "dummy",
+			},
+			expectedErrorMsg: "unexpected field: 'unexpected_argument'",
+		},
 	}
 }