[Access] Implement integration test for new websockets [WIP]

access/api.go

@@ -203,41 +203,19 @@ type API interface {
 	//
 	// If invalid parameters will be supplied SubscribeBlockDigestsFromLatest will return a failed subscription.
 	SubscribeBlockDigestsFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription
-	// SubscribeTransactionStatusesFromStartBlockID subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the specified block ID. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
-	//
-	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
-	//   - txID: The identifier of the transaction to monitor.
-	//   - startBlockID: The block ID from which to start monitoring.
-	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromStartBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SubscribeTransactionStatusesFromStartHeight subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the specified block height. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
+	// SubscribeTransactionStatuses subscribes to transaction status updates for a given transaction ID. The subscription
+	// streams status updates until the transaction reaches the final state ([flow.TransactionStatusSealed] or
+	// [flow.TransactionStatusExpired]). When the transaction reaches one of these final states, the subscription will
+	// automatically terminate.
 	//
 	// Parameters:
 	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
 	//   - txID: The unique identifier of the transaction to monitor.
-	//   - startHeight: The block height from which to start monitoring.
 	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromStartHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SubscribeTransactionStatusesFromLatest subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the latest block. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
-	//
-	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
-	//   - txID: The unique identifier of the transaction to monitor.
-	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromLatest(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SendAndSubscribeTransactionStatuses sends a transaction to the network and subscribes to its status updates.
+	SubscribeTransactionStatuses(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SendAndSubscribeTransactionStatuses sends a transaction to the execution node and subscribes to its status updates.
 	// Monitoring begins from the reference block saved in the transaction itself and streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). Once a final status is reached, the subscription
+	// reaches the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). Once the final status has been reached, the subscription
 	// automatically terminates.
 	//
 	// Parameters:
@@ -261,6 +239,14 @@ type TransactionResult struct {
 	BlockHeight   uint64
 }
 
+func (r *TransactionResult) IsExecuted() bool {
+	return r.Status == flow.TransactionStatusExecuted || r.Status == flow.TransactionStatusSealed
+}
+
+func (r *TransactionResult) IsFinal() bool {
+	return r.Status == flow.TransactionStatusSealed || r.Status == flow.TransactionStatusExpired
+}
+
 func TransactionResultToMessage(result *TransactionResult) *access.TransactionResultResponse {
 	return &access.TransactionResultResponse{
 		Status:        entities.TransactionStatus(result.Status),

cmd/access/node_builder/access_node_builder.go

@@ -1454,10 +1454,19 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
+		// websockets config
 		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
+		flags.Uint64Var(&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection")
+		flags.Uint64Var(&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"the maximum number of responses that can be sent to a single client per second")
 	}).ValidateFlags(func() error {
 		if builder.supportsObserver && (builder.PublicNetworkConfig.BindAddress == cmd.NotSet || builder.PublicNetworkConfig.BindAddress == "") {
 			return errors.New("public-network-address must be set if supports-observer is true")

cmd/observer/node_builder/observer_builder.go

@@ -815,10 +815,19 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
+		// websockets config
 		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
+		flags.Uint64Var(&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection")
+		flags.Uint64Var(&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"the maximum number of responses that can be sent to a single client per second")
 	}).ValidateFlags(func() error {
 		if builder.executionDataSyncEnabled {
 			if builder.executionDataConfig.FetchTimeout <= 0 {

cmd/util/cmd/run-script/cmd.go

@@ -27,6 +27,8 @@ import (
 	"github.com/onflow/flow-go/module/metrics"
 )
 
+var ErrNotImplemented = errors.New("not implemented")
+
 var (
 	flagPayloads        string
 	flagState           string
@@ -532,35 +534,18 @@ func (*api) SubscribeBlockDigestsFromLatest(
 	return nil
 }
 
-func (a *api) SubscribeTransactionStatusesFromStartBlockID(
-	_ context.Context,
-	_ flow.Identifier,
-	_ flow.Identifier,
-	_ entities.EventEncodingVersion,
-) subscription.Subscription {
-	return nil
-}
-
-func (a *api) SubscribeTransactionStatusesFromStartHeight(
+func (a *api) SubscribeTransactionStatuses(
 	_ context.Context,
 	_ flow.Identifier,
-	_ uint64,
 	_ entities.EventEncodingVersion,
 ) subscription.Subscription {
-	return nil
+	return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SubscribeTransactionStatuses")
 }
 
-func (a *api) SubscribeTransactionStatusesFromLatest(
-	_ context.Context,
-	_ flow.Identifier,
-	_ entities.EventEncodingVersion,
-) subscription.Subscription {
-	return nil
-}
 func (a *api) SendAndSubscribeTransactionStatuses(
 	_ context.Context,
 	_ *flow.TransactionBody,
 	_ entities.EventEncodingVersion,
 ) subscription.Subscription {
-	return nil
+	return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SendAndSubscribeTransactionStatuses")
 }

engine/access/rest/http/request/address.go → engine/access/rest/common/parser/address.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/address_test.go → engine/access/rest/common/parser/address_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"strings"

engine/access/rest/http/request/arguments.go → engine/access/rest/common/parser/arguments.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -7,6 +7,7 @@ import (
 )
 
 const maxArgumentsLength = 100
+const MaxAllowedScriptArguments = 100
 
 type Arguments [][]byte
 
@@ -25,7 +26,7 @@ func (a *Arguments) Parse(raw []string) error {
 	}
 
 	if len(args) > maxArgumentsLength {
-		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments)
+		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments)
 	}
 
 	*a = args

engine/access/rest/http/request/arguments_test.go → engine/access/rest/common/parser/arguments_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -21,13 +21,13 @@ func TestArguments_InvalidParse(t *testing.T) {
 		assert.EqualError(t, err, "invalid argument encoding: illegal base64 data at input byte 0", a)
 	}
 
-	tooLong := make([]string, maxAllowedScriptArguments+1)
+	tooLong := make([]string, MaxAllowedScriptArguments+1)
 	for i := range tooLong {
 		tooLong[i] = "dGVzdA=="
 	}
 
 	err := arguments.Parse(tooLong)
-	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments))
+	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments))
 }
 
 func TestArguments_ValidParse(t *testing.T) {

engine/access/rest/http/request/proposal_key.go → engine/access/rest/common/parser/proposal_key.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/signature_test.go → engine/access/rest/common/parser/signature_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"encoding/hex"

engine/access/rest/http/request/signatures.go → engine/access/rest/common/parser/signatures.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/transaction.go → engine/access/rest/common/parser/transaction.go

@@ -1,24 +1,23 @@
-package request
+package parser
 
 import (
 	"fmt"
 	"io"
 
-	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+	"github.com/onflow/flow-go/engine/access/rest/common"
 	"github.com/onflow/flow-go/engine/access/rest/http/models"
 	"github.com/onflow/flow-go/engine/access/rest/util"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 )
 
 const maxAuthorizers = 100
-const maxAllowedScriptArguments = 100
 
 type Transaction flow.TransactionBody
 
 func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 	var tx models.TransactionsBody
-	err := parseBody(raw, &tx)
+	err := common.ParseBody(raw, &tx)
 	if err != nil {
 		return err
 	}
@@ -35,8 +34,8 @@ func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 	if len(tx.Authorizers) > maxAuthorizers {
 		return fmt.Errorf("too many authorizers. Maximum authorizers allowed: %d", maxAuthorizers)
 	}
-	if len(tx.Arguments) > maxAllowedScriptArguments {
-		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments)
+	if len(tx.Arguments) > MaxAllowedScriptArguments {
+		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments)
 	}
 	if tx.ReferenceBlockId == "" {
 		return fmt.Errorf("reference block not provided")
@@ -90,7 +89,7 @@ func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 		return fmt.Errorf("invalid transaction script encoding")
 	}
 
-	var blockID parser.ID
+	var blockID ID
 	err = blockID.Parse(tx.ReferenceBlockId)
 	if err != nil {
 		return fmt.Errorf("invalid reference block ID: %w", err)

engine/access/rest/http/request/transaction_test.go → engine/access/rest/common/parser/transaction_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"bytes"