[Access] SendAndSubscribeTransactionStatuses endpoint implementation for Access Streaming API

access/api.go

@@ -194,6 +194,10 @@ type API interface {
 	//
 	// If invalid parameters will be supplied SubscribeBlockDigestsFromLatest will return a failed subscription.
 	SubscribeBlockDigestsFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription
+	// SubscribeTransactionStatuses streams transaction statuses starting from the reference block saved in the
+	// transaction itself until the block containing the transaction becomes sealed or expired. When the transaction
+	// status becomes TransactionStatusSealed or TransactionStatusExpired, the subscription will automatically shut down.
+	SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody) subscription.Subscription
 }
 
 // TODO: Combine this with flow.TransactionResult?

access/handler.go

@@ -1086,9 +1086,43 @@ func (h *Handler) getSubscriptionDataFromStartBlockID(msgBlockId []byte, msgBloc
 	return startBlockID, blockStatus, nil
 }
 
-func (h *Handler) SendAndSubscribeTransactionStatuses(_ *access.SendAndSubscribeTransactionStatusesRequest, _ access.AccessAPI_SendAndSubscribeTransactionStatusesServer) error {
-	// not implemented
-	return nil
+// SendAndSubscribeTransactionStatuses streams transaction statuses starting from the reference block saved in the
+// transaction itself until the block containing the transaction becomes sealed or expired. When the transaction
+// status becomes TransactionStatusSealed or TransactionStatusExpired, the subscription will automatically shut down.
+func (h *Handler) SendAndSubscribeTransactionStatuses(
+	request *access.SendAndSubscribeTransactionStatusesRequest,
+	stream access.AccessAPI_SendAndSubscribeTransactionStatusesServer,
+) error {
+	subCtx, cancel := context.WithCancel(stream.Context())
+	defer cancel()
+
+	// check if the maximum number of streams is reached
+	if h.StreamCount.Load() >= h.MaxStreams {
+		return status.Errorf(codes.ResourceExhausted, "maximum number of streams reached")
+	}
+	h.StreamCount.Add(1)
+	defer h.StreamCount.Add(-1)
+
+	tx, err := convert.MessageToTransaction(request.GetTransaction(), h.chain)
+	if err != nil {
+		return status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	err = h.api.SendTransaction(subCtx, &tx)
+	if err != nil {
+		return err
+	}
+
+	sub := h.api.SubscribeTransactionStatuses(subCtx, &tx)
+
+	return subscription.HandleSubscription(sub, func(txSubInfo *convert.TransactionSubscribeInfo) error {
+		err = stream.Send(convert.TransactionSubscribeInfoToMessage(txSubInfo))
+		if err != nil {
+			return rpc.ConvertError(err, "could not send response", codes.Internal)
+		}
+
+		return nil
+	})
 }
 
 func (h *Handler) blockResponse(block *flow.Block, fullResponse bool, status flow.BlockStatus) (*access.BlockResponse, error) {

engine/access/rpc/backend/backend.go

@@ -73,6 +73,7 @@ type Backend struct {
 	backendExecutionResults
 	backendNetwork
 	backendSubscribeBlocks
+	backendSubscribeTransactions
 
 	state             protocol.State
 	chainID           flow.ChainID
@@ -169,6 +170,15 @@ func New(params Params) (*Backend, error) {
 	// initialize node version info
 	nodeInfo := getNodeVersionInfo(params.State.Params())
 
+	transactionsLocalDataProvider := &TransactionsLocalDataProvider{
+		state:          params.State,
+		collections:    params.Collections,
+		blocks:         params.Blocks,
+		eventsIndex:    params.EventsIndex,
+		txResultsIndex: params.TxResultsIndex,
+		systemTxID:     systemTxID,
+	}
+
 	b := &Backend{
 		state:        params.State,
 		BlockTracker: params.BlockTracker,
@@ -186,30 +196,23 @@ func New(params Params) (*Backend, error) {
 			scriptExecMode:    params.ScriptExecutionMode,
 		},
 		backendTransactions: backendTransactions{
-			TransactionsLocalDataProvider: TransactionsLocalDataProvider{
-				state:          params.State,
-				collections:    params.Collections,
-				blocks:         params.Blocks,
-				eventsIndex:    params.EventsIndex,
-				txResultsIndex: params.TxResultsIndex,
-				systemTxID:     systemTxID,
-			},
-			log:                  params.Log,
-			staticCollectionRPC:  params.CollectionRPC,
-			chainID:              params.ChainID,
-			transactions:         params.Transactions,
-			executionReceipts:    params.ExecutionReceipts,
-			transactionValidator: configureTransactionValidator(params.State, params.ChainID),
-			transactionMetrics:   params.AccessMetrics,
-			retry:                retry,
-			connFactory:          params.ConnFactory,
-			previousAccessNodes:  params.HistoricalAccessNodes,
-			nodeCommunicator:     params.Communicator,
-			txResultCache:        txResCache,
-			txErrorMessagesCache: txErrorMessagesCache,
-			txResultQueryMode:    params.TxResultQueryMode,
-			systemTx:             systemTx,
-			systemTxID:           systemTxID,
+			TransactionsLocalDataProvider: transactionsLocalDataProvider,
+			log:                           params.Log,
+			staticCollectionRPC:           params.CollectionRPC,
+			chainID:                       params.ChainID,
+			transactions:                  params.Transactions,
+			executionReceipts:             params.ExecutionReceipts,
+			transactionValidator:          configureTransactionValidator(params.State, params.ChainID),
+			transactionMetrics:            params.AccessMetrics,
+			retry:                         retry,
+			connFactory:                   params.ConnFactory,
+			previousAccessNodes:           params.HistoricalAccessNodes,
+			nodeCommunicator:              params.Communicator,
+			txResultCache:                 txResCache,
+			txErrorMessagesCache:          txErrorMessagesCache,
+			txResultQueryMode:             params.TxResultQueryMode,
+			systemTx:                      systemTx,
+			systemTxID:                    systemTxID,
 		},
 		backendEvents: backendEvents{
 			log:               params.Log,
@@ -261,6 +264,16 @@ func New(params Params) (*Backend, error) {
 			sendBufferSize: params.SubscriptionParams.SendBufferSize,
 			blockTracker:   params.BlockTracker,
 		},
+		backendSubscribeTransactions: backendSubscribeTransactions{
+			txLocalDataProvider: transactionsLocalDataProvider,
+			log:                 params.Log,
+			executionResults:    params.ExecutionResults,
+			broadcaster:         params.SubscriptionParams.Broadcaster,
+			sendTimeout:         params.SubscriptionParams.SendTimeout,
+			responseLimit:       params.SubscriptionParams.ResponseLimit,
+			sendBufferSize:      params.SubscriptionParams.SendBufferSize,
+			blockTracker:        params.BlockTracker,
+		},
 		collections:       params.Collections,
 		executionReceipts: params.ExecutionReceipts,
 		connFactory:       params.ConnFactory,

engine/access/rpc/backend/backend_stream_transactions.go

@@ -0,0 +1,195 @@
+package backend
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/onflow/flow-go/module/irrecoverable"
+	"github.com/onflow/flow-go/state"
+
+	"github.com/onflow/flow-go/engine/common/rpc/convert"
+	"github.com/onflow/flow-go/module/counters"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/engine"
+	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/engine/common/rpc"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/storage"
+)
+
+// backendSubscribeTransactions handles transaction subscriptions.
+type backendSubscribeTransactions struct {
+	txLocalDataProvider *TransactionsLocalDataProvider
+	executionResults    storage.ExecutionResults
+	log                 zerolog.Logger
+	broadcaster         *engine.Broadcaster
+	sendTimeout         time.Duration
+	responseLimit       float64
+	sendBufferSize      int
+
+	blockTracker subscription.BlockTracker
+}
+
+// TransactionSubscriptionMetadata holds data representing the status state for each transaction subscription.
+type TransactionSubscriptionMetadata struct {
+	txID               flow.Identifier
+	txReferenceBlockID flow.Identifier
+	messageIndex       counters.StrictMonotonousCounter
+	blockWithTx        *flow.Header
+	blockID            flow.Identifier
+	txExecuted         bool
+	lastTxStatus       flow.TransactionStatus
+}
+
+// SubscribeTransactionStatuses subscribes to transaction status changes starting from the transaction reference block ID.
+// If invalid tx parameters will be supplied SubscribeTransactionStatuses will return a failed subscription.
+func (b *backendSubscribeTransactions) SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody) subscription.Subscription {
+	nextHeight, err := b.blockTracker.GetStartHeightFromBlockID(tx.ReferenceBlockID)
+	if err != nil {
+		return subscription.NewFailedSubscription(err, "could not get start height")
+	}
+
+	txInfo := TransactionSubscriptionMetadata{
+		txID:               tx.ID(),
+		txReferenceBlockID: tx.ReferenceBlockID,
+		messageIndex:       counters.NewMonotonousCounter(0),
+		blockWithTx:        nil,
+		blockID:            flow.ZeroID,
+		lastTxStatus:       flow.TransactionStatusUnknown,
+	}
+
+	sub := subscription.NewHeightBasedSubscription(
+		b.sendBufferSize,
+		nextHeight,
+		b.getTransactionStatusResponse(&txInfo),
+	)
+
+	go subscription.NewStreamer(b.log, b.broadcaster, b.sendTimeout, b.responseLimit, sub).Stream(ctx)
+
+	return sub
+}
+
+// getTransactionStatusResponse returns a callback function that produces transaction status
+// subscription responses based on new blocks.
+func (b *backendSubscribeTransactions) getTransactionStatusResponse(txInfo *TransactionSubscriptionMetadata) func(context.Context, uint64) (interface{}, error) {
+	return func(ctx context.Context, height uint64) (interface{}, error) {
+		highestHeight, err := b.blockTracker.GetHighestHeight(flow.BlockStatusFinalized)
+		if err != nil {
+			return nil, fmt.Errorf("could not get highest height for block %d: %w", height, err)
+		}
+
+		// Fail early if no block finalized notification has been received for the given height.
+		// Note: It's possible that the block is locally finalized before the notification is
+		// received. This ensures a consistent view is available to all streams.
+		if height > highestHeight {
+			return nil, fmt.Errorf("block %d is not available yet: %w", height, subscription.ErrBlockNotReady)
+		}
+
+		if txInfo.lastTxStatus == flow.TransactionStatusSealed || txInfo.lastTxStatus == flow.TransactionStatusExpired {
+			return nil, fmt.Errorf("transaction final status %s was already reported: %w", txInfo.lastTxStatus.String(), subscription.ErrEndOfData)
+		}
+
+		if txInfo.blockWithTx == nil {
+			// Check if block contains transaction.
+			txInfo.blockWithTx, txInfo.blockID, err = b.searchForTransactionBlock(height, txInfo)
+			if err != nil {
+				if errors.Is(err, storage.ErrNotFound) {
+					return nil, fmt.Errorf("could not find block %d in storage: %w", height, subscription.ErrBlockNotReady)
+				}
+
+				if !errors.Is(err, ErrTransactionNotInBlock) {
+					return nil, status.Errorf(codes.Internal, "could not get block %d: %v", height, err)
+				}
+			}
+		}
+
+		// Find the transaction status.
+		var txStatus flow.TransactionStatus
+		if txInfo.blockWithTx == nil {
+			txStatus, err = b.txLocalDataProvider.DeriveUnknownTransactionStatus(txInfo.txReferenceBlockID)
+		} else {
+			if !txInfo.txExecuted {
+				// Check if transaction was executed.
+				txInfo.txExecuted, err = b.searchForExecutionResult(txInfo.blockID)
+				if err != nil {
+					return nil, status.Errorf(codes.Internal, "failed to get execution result for block %s: %v", txInfo.blockID, err)
+				}
+			}
+
+			txStatus, err = b.txLocalDataProvider.DeriveTransactionStatus(txInfo.blockID, txInfo.blockWithTx.Height, txInfo.txExecuted)
+		}
+		if err != nil {
+			if !errors.Is(err, state.ErrUnknownSnapshotReference) {
+				irrecoverable.Throw(ctx, err)
+			}
+			return nil, rpc.ConvertStorageError(err)
+		}
+
+		// The same transaction status should not be reported, so return here with no response
+		if txInfo.lastTxStatus == txStatus {
+			return nil, nil
+		}
+		txInfo.lastTxStatus = txStatus
+
+		messageIndex := txInfo.messageIndex.Value()
+		if ok := txInfo.messageIndex.Set(messageIndex + 1); !ok {
+			return nil, status.Errorf(codes.Internal, "the message index has already been incremented to %d", txInfo.messageIndex.Value())
+		}
+
+		return &convert.TransactionSubscribeInfo{
+			ID:           txInfo.txID,
+			Status:       txInfo.lastTxStatus,
+			MessageIndex: messageIndex,
+		}, nil
+	}
+}
+
+// searchForTransactionBlock searches for the block containing the specified transaction.
+// It retrieves the block at the given height and checks if the transaction is included in that block.
+// Expected errors:
+// - subscription.ErrBlockNotReady when unable to retrieve the block or collection ID
+// - codes.Internal when other errors occur during block or collection lookup
+func (b *backendSubscribeTransactions) searchForTransactionBlock(
+	height uint64,
+	txInfo *TransactionSubscriptionMetadata,
+) (*flow.Header, flow.Identifier, error) {
+	block, err := b.txLocalDataProvider.blocks.ByHeight(height)
+	if err != nil {
+		return nil, flow.ZeroID, fmt.Errorf("error looking up block: %w", err)
+	}
+
+	collectionID, err := b.txLocalDataProvider.LookupCollectionIDInBlock(block, txInfo.txID)
+	if err != nil {
+		return nil, flow.ZeroID, fmt.Errorf("error looking up transaction in block: %w", err)
+	}
+
+	if collectionID != flow.ZeroID {
+		return block.Header, block.ID(), nil
+	}
+
+	return nil, flow.ZeroID, nil
+}
+
+// searchForExecutionResult searches for the execution result of a block. It retrieves the execution result for the specified block ID.
+// Expected errors:
+// - codes.Internal if an internal error occurs while retrieving execution result.
+func (b *backendSubscribeTransactions) searchForExecutionResult(
+	blockID flow.Identifier,
+) (bool, error) {
+	_, err := b.executionResults.ByBlockID(blockID)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			return false, nil
+		}
+		return false, fmt.Errorf("failed to get execution result for block %s: %w", blockID, err)
+	}
+
+	return true, nil
+}