Merge pull request #6737 from The-K-R-O-K/AndriiSlisarchuk/6573-pending-should-return-immediately

[Access] Implement subscribe transaction statuses by transaction ID

Author: Guitarheroua

access/api.go

@@ -203,41 +203,20 @@ 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.
+	// SubscribeTransactionStatuses subscribes to transaction status updates for a given transaction ID. Monitoring starts
+	// from the latest block to obtain the current transaction status. If the transaction is already in the final state
+	// ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]), all statuses will be prepared and sent to the client
+	// sequentially. If the transaction is not in the final state, the subscription will stream status updates until the transaction
+	// reaches the final state. Once a final state is reached, 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.
-	//
-	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
+	//   - ctx: 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 +240,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),

access/mock/api.go

@@ -27,6 +27,8 @@ import (
 	"github.com/onflow/flow-go/module/metrics"
 )
 
+var ErrNotImplemented = errors.New("not implemented")
+
 var (
 	flagPayloads        string
 	flagState           string
@@ -533,35 +535,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")
 }

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

@@ -134,9 +134,9 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 		{
 			name:      "transaction statuses topic",
 			topic:     TransactionStatusesTopic,
-			arguments: models.Arguments{},
+			arguments: models.Arguments{"tx_id": unittest.IdentifierFixture().String()},
 			setupSubscription: func() {
-				s.setupSubscription(s.accessApi.On("SubscribeTransactionStatusesFromLatest", mock.Anything, mock.Anything, mock.Anything))
+				s.setupSubscription(s.accessApi.On("SubscribeTransactionStatuses", mock.Anything, mock.Anything, mock.Anything))
 			},
 			assertExpectations: func() {
 				s.stateStreamApi.AssertExpectations(s.T())

engine/access/rest/websockets/data_providers/factory_test.go

@@ -11,7 +11,6 @@ import (
 	"github.com/onflow/flow-go/access"
 	commonmodels "github.com/onflow/flow-go/engine/access/rest/common/models"
 	"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/websockets/models"
 	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/model/flow"
@@ -20,13 +19,6 @@ import (
 	"github.com/onflow/flow/protobuf/go/flow/entities"
 )
 
-// transactionStatusesArguments contains the arguments required for subscribing to transaction statuses
-type transactionStatusesArguments struct {
-	TxID             flow.Identifier // ID of the transaction to monitor.
-	StartBlockID     flow.Identifier // ID of the block to start subscription from
-	StartBlockHeight uint64          // Height of the block to start subscription from
-}
-
 // TransactionStatusesDataProvider is responsible for providing tx statuses
 type TransactionStatusesDataProvider struct {
 	*baseDataProvider
@@ -54,8 +46,8 @@ func NewTransactionStatusesDataProvider(
 		linkGenerator: linkGenerator,
 	}
 
-	// Initialize arguments passed to the provider.
-	txStatusesArgs, err := parseTransactionStatusesArguments(arguments)
+	// Initialize txID passed to the provider.
+	txID, err := parseTransactionID(arguments)
 	if err != nil {
 		return nil, fmt.Errorf("invalid arguments for tx statuses data provider: %w", err)
 	}
@@ -68,7 +60,7 @@ func NewTransactionStatusesDataProvider(
 		arguments,
 		cancel,
 		send,
-		p.createSubscription(subCtx, txStatusesArgs), // Set up a subscription to tx statuses based on arguments.
+		p.createSubscription(subCtx, txID), // Set up a subscription to tx statuses based on arguments.
 	)
 
 	return p, nil
@@ -84,17 +76,9 @@ func (p *TransactionStatusesDataProvider) Run() error {
 // createSubscription creates a new subscription using the specified input arguments.
 func (p *TransactionStatusesDataProvider) createSubscription(
 	ctx context.Context,
-	args transactionStatusesArguments,
+	txID flow.Identifier,
 ) subscription.Subscription {
-	if args.StartBlockID != flow.ZeroID {
-		return p.api.SubscribeTransactionStatusesFromStartBlockID(ctx, args.TxID, args.StartBlockID, entities.EventEncodingVersion_JSON_CDC_V0)
-	}
-
-	if args.StartBlockHeight != request.EmptyHeight {
-		return p.api.SubscribeTransactionStatusesFromStartHeight(ctx, args.TxID, args.StartBlockHeight, entities.EventEncodingVersion_JSON_CDC_V0)
-	}
-
-	return p.api.SubscribeTransactionStatusesFromLatest(ctx, args.TxID, entities.EventEncodingVersion_JSON_CDC_V0)
+	return p.api.SubscribeTransactionStatuses(ctx, txID, entities.EventEncodingVersion_JSON_CDC_V0)
 }
 
 // handleResponse processes a tx statuses and sends the formatted response.
@@ -123,42 +107,30 @@ func (p *TransactionStatusesDataProvider) handleResponse() func(txResults []*acc
 	}
 }
 
-// parseAccountStatusesArguments validates and initializes the account statuses arguments.
-func parseTransactionStatusesArguments(
+// parseTransactionID validates and initializes the transaction ID argument.
+func parseTransactionID(
 	arguments models.Arguments,
-) (transactionStatusesArguments, error) {
+) (flow.Identifier, error) {
 	allowedFields := []string{
-		"start_block_id",
-		"start_block_height",
 		"tx_id",
 	}
 	err := ensureAllowedFields(arguments, allowedFields)
 	if err != nil {
-		return transactionStatusesArguments{}, err
-	}
-
-	var args transactionStatusesArguments
-
-	// Parse block arguments
-	startBlockID, startBlockHeight, err := parseStartBlock(arguments)
-	if err != nil {
-		return transactionStatusesArguments{}, err
+		return flow.ZeroID, err
 	}
-	args.StartBlockID = startBlockID
-	args.StartBlockHeight = startBlockHeight
 
 	if txIDIn, ok := arguments["tx_id"]; ok && txIDIn != "" {
 		result, ok := txIDIn.(string)
 		if !ok {
-			return transactionStatusesArguments{}, fmt.Errorf("'tx_id' must be a string")
+			return flow.ZeroID, fmt.Errorf("'tx_id' must be a string")
 		}
-		var txID parser.ID
-		err := txID.Parse(result)
+		var txIDParsed parser.ID
+		err := txIDParsed.Parse(result)
 		if err != nil {
-			return transactionStatusesArguments{}, fmt.Errorf("invalid 'tx_id': %w", err)
+			return flow.ZeroID, fmt.Errorf("invalid 'tx_id': %w", err)
 		}
-		args.TxID = txID.Flow()
+		return txIDParsed.Flow(), nil
 	}
 
-	return args, nil
+	return flow.ZeroID, fmt.Errorf("arguments are invalid")
 }