Merge pull request #5620 from The-K-R-O-K/AndriiSlisarchuk/5566-tx-re…

…sult-in-sub-tx-statuses [Access] Added transaction result in SendAndSubscribeTransactionStatuses response
onflow · Apr 20, 2024 · 0983dbb · 0983dbb
2 parents 41926ad + 6a06688
commit 0983dbb
Show file tree

Hide file tree

Showing 14 changed files with 329 additions and 187 deletions.
diff --git a/access/api.go b/access/api.go
@@ -197,7 +197,7 @@ type API interface {
 	// 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
+	SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
 }
 
 // TODO: Combine this with flow.TransactionResult?

diff --git a/access/handler.go b/access/handler.go
@@ -14,6 +14,7 @@ import (
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/counters"
 
 	"github.com/onflow/flow/protobuf/go/flow/access"
 	"github.com/onflow/flow/protobuf/go/flow/entities"
@@ -1112,11 +1113,23 @@ func (h *Handler) SendAndSubscribeTransactionStatuses(
 		return err
 	}
 
-	sub := h.api.SubscribeTransactionStatuses(ctx, &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)
+	sub := h.api.SubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion())
+
+	messageIndex := counters.NewMonotonousCounter(0)
+	return subscription.HandleSubscription(sub, func(txResults []*TransactionResult) error {
+		for i := range txResults {
+			value := messageIndex.Value()
+			if ok := messageIndex.Set(value + 1); !ok {
+				return status.Errorf(codes.Internal, "the message index has already been incremented to %d", messageIndex.Value())
+			}
+
+			err = stream.Send(&access.SendAndSubscribeTransactionStatusesResponse{
+				TransactionResults: TransactionResultToMessage(txResults[i]),
+				MessageIndex:       value,
+			})
+			if err != nil {
+				return rpc.ConvertError(err, "could not send response", codes.Internal)
+			}
 		}
 
 		return nil

diff --git a/access/mock/api.go b/access/mock/api.go
diff --git a/engine/access/rpc/backend/backend.go b/engine/access/rpc/backend/backend.go
@@ -162,15 +162,6 @@ 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,
@@ -187,25 +178,6 @@ func New(params Params) (*Backend, error) {
 			scriptExecutor:    params.ScriptExecutor,
 			scriptExecMode:    params.ScriptExecutionMode,
 		},
-		backendTransactions: backendTransactions{
-			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,
 			chain:             params.ChainID.Chain(),
@@ -253,22 +225,55 @@ func New(params Params) (*Backend, error) {
 			subscriptionHandler: params.SubscriptionHandler,
 			blockTracker:        params.BlockTracker,
 		},
-		backendSubscribeTransactions: backendSubscribeTransactions{
-			txLocalDataProvider: transactionsLocalDataProvider,
-			log:                 params.Log,
-			executionResults:    params.ExecutionResults,
-			subscriptionHandler: params.SubscriptionHandler,
-			blockTracker:        params.BlockTracker,
-		},
+
 		collections:       params.Collections,
 		executionReceipts: params.ExecutionReceipts,
 		connFactory:       params.ConnFactory,
 		chainID:           params.ChainID,
 		nodeInfo:          nodeInfo,
 	}
 
+	transactionsLocalDataProvider := &TransactionsLocalDataProvider{
+		state:          params.State,
+		collections:    params.Collections,
+		blocks:         params.Blocks,
+		eventsIndex:    params.EventsIndex,
+		txResultsIndex: params.TxResultsIndex,
+		systemTxID:     systemTxID,
+	}
+
+	b.backendTransactions = backendTransactions{
+		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,
+	}
+
+	// TODO: The TransactionErrorMessage interface should be reorganized in future, as it is implemented in backendTransactions but used in TransactionsLocalDataProvider, and its initialization is somewhat quirky.
 	b.backendTransactions.txErrorMessages = b
 
+	b.backendSubscribeTransactions = backendSubscribeTransactions{
+		txLocalDataProvider: transactionsLocalDataProvider,
+		backendTransactions: &b.backendTransactions,
+		log:                 params.Log,
+		executionResults:    params.ExecutionResults,
+		subscriptionHandler: params.SubscriptionHandler,
+		blockTracker:        params.BlockTracker,
+	}
+
 	retry.SetBackend(b)
 
 	preferredENIdentifiers, err = identifierList(params.PreferredExecutionNodeIDs)