diff --git a/access/api.go b/access/api.go index adeb7284c10..acd4d6138b8 100644 --- a/access/api.go +++ b/access/api.go @@ -203,10 +203,50 @@ 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, requiredEventEncodingVersion entities.EventEncodingVersion) 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. + // + // 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 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 + // automatically terminates. + // + // Parameters: + // - ctx: The context to manage the transaction sending and subscription lifecycle, including cancellation. + // - tx: The transaction body to be sent and monitored. + // - requiredEventEncodingVersion: The version of event encoding required for the subscription. + // + // If the transaction cannot be sent, the subscription will fail and return a failed subscription. + SendAndSubscribeTransactionStatuses(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 index b974e7034fc..bcf401a2884 100644 --- a/access/handler.go +++ b/access/handler.go @@ -1425,12 +1425,7 @@ func (h *Handler) SendAndSubscribeTransactionStatuses( return status.Error(codes.InvalidArgument, err.Error()) } - err = h.api.SendTransaction(ctx, &tx) - if err != nil { - return err - } - - sub := h.api.SubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion()) + sub := h.api.SendAndSubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion()) messageIndex := counters.NewMonotonousCounter(0) return subscription.HandleRPCSubscription(sub, func(txResults []*TransactionResult) error { diff --git a/access/mock/api.go b/access/mock/api.go index eaaf6c428f2..13c35b293d3 100644 --- a/access/mock/api.go +++ b/access/mock/api.go @@ -1145,6 +1145,26 @@ func (_m *API) Ping(ctx context.Context) error { return r0 } +// SendAndSubscribeTransactionStatuses provides a mock function with given fields: ctx, tx, requiredEventEncodingVersion +func (_m *API) SendAndSubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription { + ret := _m.Called(ctx, tx, requiredEventEncodingVersion) + + if len(ret) == 0 { + panic("no return value specified for SendAndSubscribeTransactionStatuses") + } + + var r0 subscription.Subscription + if rf, ok := ret.Get(0).(func(context.Context, *flow.TransactionBody, entities.EventEncodingVersion) subscription.Subscription); ok { + r0 = rf(ctx, tx, requiredEventEncodingVersion) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(subscription.Subscription) + } + } + + return r0 +} + // SendTransaction provides a mock function with given fields: ctx, tx func (_m *API) SendTransaction(ctx context.Context, tx *flow.TransactionBody) error { ret := _m.Called(ctx, tx) @@ -1343,17 +1363,57 @@ func (_m *API) SubscribeBlocksFromStartHeight(ctx context.Context, startHeight u return r0 } -// SubscribeTransactionStatuses provides a mock function with given fields: ctx, tx, requiredEventEncodingVersion -func (_m *API) SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription { - ret := _m.Called(ctx, tx, requiredEventEncodingVersion) +// SubscribeTransactionStatusesFromLatest provides a mock function with given fields: ctx, txID, requiredEventEncodingVersion +func (_m *API) SubscribeTransactionStatusesFromLatest(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription { + ret := _m.Called(ctx, txID, requiredEventEncodingVersion) if len(ret) == 0 { - panic("no return value specified for SubscribeTransactionStatuses") + panic("no return value specified for SubscribeTransactionStatusesFromLatest") } var r0 subscription.Subscription - if rf, ok := ret.Get(0).(func(context.Context, *flow.TransactionBody, entities.EventEncodingVersion) subscription.Subscription); ok { - r0 = rf(ctx, tx, requiredEventEncodingVersion) + if rf, ok := ret.Get(0).(func(context.Context, flow.Identifier, entities.EventEncodingVersion) subscription.Subscription); ok { + r0 = rf(ctx, txID, requiredEventEncodingVersion) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(subscription.Subscription) + } + } + + return r0 +} + +// SubscribeTransactionStatusesFromStartBlockID provides a mock function with given fields: ctx, txID, startBlockID, requiredEventEncodingVersion +func (_m *API) SubscribeTransactionStatusesFromStartBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription { + ret := _m.Called(ctx, txID, startBlockID, requiredEventEncodingVersion) + + if len(ret) == 0 { + panic("no return value specified for SubscribeTransactionStatusesFromStartBlockID") + } + + var r0 subscription.Subscription + if rf, ok := ret.Get(0).(func(context.Context, flow.Identifier, flow.Identifier, entities.EventEncodingVersion) subscription.Subscription); ok { + r0 = rf(ctx, txID, startBlockID, requiredEventEncodingVersion) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(subscription.Subscription) + } + } + + return r0 +} + +// SubscribeTransactionStatusesFromStartHeight provides a mock function with given fields: ctx, txID, startHeight, requiredEventEncodingVersion +func (_m *API) SubscribeTransactionStatusesFromStartHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription { + ret := _m.Called(ctx, txID, startHeight, requiredEventEncodingVersion) + + if len(ret) == 0 { + panic("no return value specified for SubscribeTransactionStatusesFromStartHeight") + } + + var r0 subscription.Subscription + if rf, ok := ret.Get(0).(func(context.Context, flow.Identifier, uint64, entities.EventEncodingVersion) subscription.Subscription); ok { + r0 = rf(ctx, txID, startHeight, requiredEventEncodingVersion) } else { if ret.Get(0) != nil { r0 = ret.Get(0).(subscription.Subscription) diff --git a/cmd/util/cmd/run-script/cmd.go b/cmd/util/cmd/run-script/cmd.go index 171f97e76b7..6814cd0583d 100644 --- a/cmd/util/cmd/run-script/cmd.go +++ b/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,10 +534,36 @@ func (*api) SubscribeBlockDigestsFromLatest( return nil } -func (*api) SubscribeTransactionStatuses( +func (a *api) SubscribeTransactionStatusesFromStartBlockID( + _ context.Context, + _ flow.Identifier, + _ flow.Identifier, + _ entities.EventEncodingVersion, +) subscription.Subscription { + return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SubscribeTransactionStatusesFromStartBlockID") +} + +func (a *api) SubscribeTransactionStatusesFromStartHeight( + _ context.Context, + _ flow.Identifier, + _ uint64, + _ entities.EventEncodingVersion, +) subscription.Subscription { + return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SubscribeTransactionStatusesFromStartHeight") +} + +func (a *api) SubscribeTransactionStatusesFromLatest( + _ context.Context, + _ flow.Identifier, + _ entities.EventEncodingVersion, +) subscription.Subscription { + return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SubscribeTransactionStatusesFromLatest") +} + +func (a *api) SendAndSubscribeTransactionStatuses( _ context.Context, _ *flow.TransactionBody, _ entities.EventEncodingVersion, ) subscription.Subscription { - return nil + return subscription.NewFailedSubscription(ErrNotImplemented, "failed to call SendAndSubscribeTransactionStatuses") } diff --git a/engine/access/rpc/backend/backend.go b/engine/access/rpc/backend/backend.go index d4666af9529..5ed2232b87b 100644 --- a/engine/access/rpc/backend/backend.go +++ b/engine/access/rpc/backend/backend.go @@ -260,6 +260,7 @@ func New(params Params) (*Backend, error) { executionResults: params.ExecutionResults, subscriptionHandler: params.SubscriptionHandler, blockTracker: params.BlockTracker, + sendTransaction: b.SendTransaction, } retry.SetBackend(b) diff --git a/engine/access/rpc/backend/backend_stream_transactions.go b/engine/access/rpc/backend/backend_stream_transactions.go index a82b365240e..358b3b9c80f 100644 --- a/engine/access/rpc/backend/backend_stream_transactions.go +++ b/engine/access/rpc/backend/backend_stream_transactions.go @@ -21,6 +21,9 @@ import ( "github.com/onflow/flow/protobuf/go/flow/entities" ) +// sendTransaction defines a function type for sending a transaction. +type sendTransaction func(ctx context.Context, tx *flow.TransactionBody) error + // backendSubscribeTransactions handles transaction subscriptions. type backendSubscribeTransactions struct { txLocalDataProvider *TransactionsLocalDataProvider @@ -30,38 +33,112 @@ type backendSubscribeTransactions struct { subscriptionHandler *subscription.SubscriptionHandler blockTracker subscription.BlockTracker + sendTransaction sendTransaction } -// TransactionSubscriptionMetadata holds data representing the status state for each transaction subscription. -type TransactionSubscriptionMetadata struct { +// transactionSubscriptionMetadata holds data representing the status state for each transaction subscription. +type transactionSubscriptionMetadata struct { *access.TransactionResult txReferenceBlockID flow.Identifier blockWithTx *flow.Header txExecuted bool eventEncodingVersion entities.EventEncodingVersion + shouldTriggerPending bool } -// 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( +// SendAndSubscribeTransactionStatuses sends a transaction and subscribes to its status updates. +// It starts monitoring the status from the transaction's reference block ID. +// If the transaction cannot be sent or an error occurs during subscription creation, a failed subscription is returned. +func (b *backendSubscribeTransactions) SendAndSubscribeTransactionStatuses( ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion, ) subscription.Subscription { - nextHeight, err := b.blockTracker.GetStartHeightFromBlockID(tx.ReferenceBlockID) + if err := b.sendTransaction(ctx, tx); err != nil { + b.log.Debug().Err(err).Str("tx_id", tx.ID().String()).Msg("failed to send transaction") + return subscription.NewFailedSubscription(err, "failed to send transaction") + } + + return b.createSubscription(ctx, tx.ID(), tx.ReferenceBlockID, 0, tx.ReferenceBlockID, requiredEventEncodingVersion, true) +} + +// SubscribeTransactionStatusesFromStartHeight subscribes to the status updates of a transaction. +// Monitoring starts from the specified block height. +// If the block height cannot be determined or an error occurs during subscription creation, a failed subscription is returned. +func (b *backendSubscribeTransactions) SubscribeTransactionStatusesFromStartHeight( + ctx context.Context, + txID flow.Identifier, + startHeight uint64, + requiredEventEncodingVersion entities.EventEncodingVersion, +) subscription.Subscription { + return b.createSubscription(ctx, txID, flow.ZeroID, startHeight, flow.ZeroID, requiredEventEncodingVersion, false) +} + +// SubscribeTransactionStatusesFromStartBlockID subscribes to the status updates of a transaction. +// Monitoring starts from the specified block ID. +// If the block ID cannot be determined or an error occurs during subscription creation, a failed subscription is returned. +func (b *backendSubscribeTransactions) SubscribeTransactionStatusesFromStartBlockID( + ctx context.Context, + txID flow.Identifier, + startBlockID flow.Identifier, + requiredEventEncodingVersion entities.EventEncodingVersion, +) subscription.Subscription { + return b.createSubscription(ctx, txID, startBlockID, 0, flow.ZeroID, requiredEventEncodingVersion, false) +} + +// SubscribeTransactionStatusesFromLatest subscribes to the status updates of a transaction. +// Monitoring starts from the latest block. +// If the block cannot be retrieved or an error occurs during subscription creation, a failed subscription is returned. +func (b *backendSubscribeTransactions) SubscribeTransactionStatusesFromLatest( + ctx context.Context, + txID flow.Identifier, + requiredEventEncodingVersion entities.EventEncodingVersion, +) subscription.Subscription { + header, err := b.txLocalDataProvider.state.Sealed().Head() if err != nil { - return subscription.NewFailedSubscription(err, "could not get start height") + b.log.Error().Err(err).Msg("failed to retrieve latest block") + return subscription.NewFailedSubscription(err, "failed to retrieve latest block") } - txInfo := TransactionSubscriptionMetadata{ + return b.createSubscription(ctx, txID, header.ID(), 0, flow.ZeroID, requiredEventEncodingVersion, false) +} + +// createSubscription initializes a subscription for monitoring a transaction's status. +// If the start height cannot be determined, a failed subscription is returned. +func (b *backendSubscribeTransactions) createSubscription( + ctx context.Context, + txID flow.Identifier, + startBlockID flow.Identifier, + startBlockHeight uint64, + referenceBlockID flow.Identifier, + requiredEventEncodingVersion entities.EventEncodingVersion, + shouldTriggerPending bool, +) subscription.Subscription { + var nextHeight uint64 + var err error + + // Get height to start subscription from + if startBlockID == flow.ZeroID { + if nextHeight, err = b.blockTracker.GetStartHeightFromHeight(startBlockHeight); err != nil { + b.log.Debug().Err(err).Uint64("block_height", startBlockHeight).Msg("failed to get start height") + return subscription.NewFailedSubscription(err, "failed to get start height") + } + } else { + if nextHeight, err = b.blockTracker.GetStartHeightFromBlockID(startBlockID); err != nil { + b.log.Debug().Err(err).Str("block_id", startBlockID.String()).Msg("failed to get start height") + return subscription.NewFailedSubscription(err, "failed to get start height") + } + } + + txInfo := transactionSubscriptionMetadata{ TransactionResult: &access.TransactionResult{ - TransactionID: tx.ID(), + TransactionID: txID, BlockID: flow.ZeroID, - Status: flow.TransactionStatusUnknown, }, - txReferenceBlockID: tx.ReferenceBlockID, + txReferenceBlockID: referenceBlockID, blockWithTx: nil, eventEncodingVersion: requiredEventEncodingVersion, + shouldTriggerPending: shouldTriggerPending, } return b.subscriptionHandler.Subscribe(ctx, nextHeight, b.getTransactionStatusResponse(&txInfo)) @@ -69,16 +146,19 @@ func (b *backendSubscribeTransactions) SubscribeTransactionStatuses( // 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) { +func (b *backendSubscribeTransactions) getTransactionStatusResponse(txInfo *transactionSubscriptionMetadata) func(context.Context, uint64) (interface{}, error) { return func(ctx context.Context, height uint64) (interface{}, error) { err := b.checkBlockReady(height) if err != nil { return nil, err } - // If the transaction status already reported the final status, return with no data available - if txInfo.Status == flow.TransactionStatusSealed || txInfo.Status == flow.TransactionStatusExpired { - return nil, fmt.Errorf("transaction final status %s was already reported: %w", txInfo.Status.String(), subscription.ErrEndOfData) + if txInfo.shouldTriggerPending { + return b.handlePendingStatus(txInfo) + } + + if b.isTransactionFinalStatus(txInfo) { + return nil, fmt.Errorf("transaction final status %s already reported: %w", txInfo.Status.String(), subscription.ErrEndOfData) } // If on this step transaction block not available, search for it. @@ -120,19 +200,8 @@ func (b *backendSubscribeTransactions) getTransactionStatusResponse(txInfo *Tran } // If block with transaction was not found, get transaction status to check if it different from last status - if txInfo.blockWithTx == nil { - txInfo.Status, err = b.txLocalDataProvider.DeriveUnknownTransactionStatus(txInfo.txReferenceBlockID) - } else if txInfo.Status == prevTxStatus { - // When a block with the transaction is available, it is possible to receive a new transaction status while - // searching for the transaction result. Otherwise, it remains unchanged. So, if the old and new transaction - // statuses are the same, the current transaction status should be retrieved. - txInfo.Status, err = b.txLocalDataProvider.DeriveTransactionStatus(txInfo.blockWithTx.Height, txInfo.txExecuted) - } - if err != nil { - if !errors.Is(err, state.ErrUnknownSnapshotReference) { - irrecoverable.Throw(ctx, err) - } - return nil, rpc.ConvertStorageError(err) + if txInfo.Status, err = b.getTransactionStatus(ctx, txInfo, prevTxStatus); err != nil { + return nil, err } // If the old and new transaction statuses are still the same, the status change should not be reported, so @@ -145,6 +214,48 @@ func (b *backendSubscribeTransactions) getTransactionStatusResponse(txInfo *Tran } } +// handlePendingStatus handles the initial pending status for a transaction. +func (b *backendSubscribeTransactions) handlePendingStatus(txInfo *transactionSubscriptionMetadata) (interface{}, error) { + txInfo.shouldTriggerPending = false + // The status of the first pending transaction should be returned immediately, as the transaction has already been sent. + // This should occur only once for each subscription. + txInfo.Status = flow.TransactionStatusPending + return b.generateResultsWithMissingStatuses(txInfo, flow.TransactionStatusUnknown) +} + +// isTransactionFinalStatus checks if a transaction has reached a final state (Sealed or Expired). +func (b *backendSubscribeTransactions) isTransactionFinalStatus(txInfo *transactionSubscriptionMetadata) bool { + return txInfo.Status == flow.TransactionStatusSealed || txInfo.Status == flow.TransactionStatusExpired +} + +// getTransactionStatus determines the current status of a transaction based on its metadata +// and previous status. It derives the transaction status by analyzing the transaction's +// execution block, if available, or its reference block. +// +// No errors expected during normal operations. +func (b *backendSubscribeTransactions) getTransactionStatus(ctx context.Context, txInfo *transactionSubscriptionMetadata, prevTxStatus flow.TransactionStatus) (flow.TransactionStatus, error) { + txStatus := txInfo.Status + var err error + + if txInfo.blockWithTx == nil { + txStatus, err = b.txLocalDataProvider.DeriveUnknownTransactionStatus(txInfo.txReferenceBlockID) + } else if txStatus == prevTxStatus { + // When a block with the transaction is available, it is possible to receive a new transaction status while + // searching for the transaction result. Otherwise, it remains unchanged. So, if the old and new transaction + // statuses are the same, the current transaction status should be retrieved. + txStatus, err = b.txLocalDataProvider.DeriveTransactionStatus(txInfo.blockWithTx.Height, txInfo.txExecuted) + } + + if err != nil { + if !errors.Is(err, state.ErrUnknownSnapshotReference) { + irrecoverable.Throw(ctx, err) + } + return flow.TransactionStatusUnknown, rpc.ConvertStorageError(err) + } + + return txStatus, nil +} + // generateResultsWithMissingStatuses checks if the current result differs from the previous result by more than one step. // If yes, it generates results for the missing transaction statuses. This is done because the subscription should send // responses for each of the statuses in the transaction lifecycle, and the message should be sent in the order of transaction statuses. @@ -153,7 +264,7 @@ func (b *backendSubscribeTransactions) getTransactionStatusResponse(txInfo *Tran // 2. pending(1) -> expired(5) // No errors expected during normal operations. func (b *backendSubscribeTransactions) generateResultsWithMissingStatuses( - txInfo *TransactionSubscriptionMetadata, + txInfo *transactionSubscriptionMetadata, prevTxStatus flow.TransactionStatus, ) ([]*access.TransactionResult, error) { // If the previous status is pending and the new status is expired, which is the last status, return its result. @@ -228,7 +339,7 @@ func (b *backendSubscribeTransactions) checkBlockReady(height uint64) error { // - codes.Internal when other errors occur during block or collection lookup func (b *backendSubscribeTransactions) searchForTransactionBlockInfo( height uint64, - txInfo *TransactionSubscriptionMetadata, + txInfo *transactionSubscriptionMetadata, ) (*flow.Header, flow.Identifier, uint64, flow.Identifier, error) { block, err := b.txLocalDataProvider.blocks.ByHeight(height) if err != nil { @@ -252,7 +363,7 @@ func (b *backendSubscribeTransactions) searchForTransactionBlockInfo( // - codes.Internal if an internal error occurs while retrieving execution result. func (b *backendSubscribeTransactions) searchForTransactionResult( ctx context.Context, - txInfo *TransactionSubscriptionMetadata, + txInfo *transactionSubscriptionMetadata, ) (*access.TransactionResult, error) { _, err := b.executionResults.ByBlockID(txInfo.BlockID) if err != nil { diff --git a/engine/access/rpc/backend/backend_stream_transactions_test.go b/engine/access/rpc/backend/backend_stream_transactions_test.go index 24cdf601f17..52049a4b0ef 100644 --- a/engine/access/rpc/backend/backend_stream_transactions_test.go +++ b/engine/access/rpc/backend/backend_stream_transactions_test.go @@ -14,6 +14,8 @@ import ( "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" + accessproto "github.com/onflow/flow/protobuf/go/flow/access" + accessapi "github.com/onflow/flow-go/access" "github.com/onflow/flow-go/engine" "github.com/onflow/flow-go/engine/access/index" @@ -121,6 +123,14 @@ func (s *TransactionStatusSuite) SetupTest() { s.blockTracker = subscriptionmock.NewBlockTracker(s.T()) s.resultsMap = map[flow.Identifier]*flow.ExecutionResult{} + s.colClient.On( + "SendTransaction", + mock.Anything, + mock.Anything, + ).Return(&accessproto.SendTransactionResponse{}, nil).Maybe() + + s.transactions.On("Store", mock.Anything).Return(nil).Maybe() + // generate blockCount consecutive blocks with associated seal, result and execution data s.rootBlock = unittest.BlockFixture() rootResult := unittest.ExecutionResultFixture(unittest.WithBlock(&s.rootBlock)) @@ -148,7 +158,7 @@ func (s *TransactionStatusSuite) SetupTest() { require.NoError(s.T(), err) s.blocks.On("ByHeight", mock.AnythingOfType("uint64")).Return(mocks.StorageMapGetter(s.blockMap)) - s.state.On("Final").Return(s.finalSnapshot, nil) + s.state.On("Final").Return(s.finalSnapshot, nil).Maybe() s.state.On("AtBlockID", mock.AnythingOfType("flow.Identifier")).Return(func(blockID flow.Identifier) protocolint.Snapshot { s.tempSnapshot.On("Head").Unset() s.tempSnapshot.On("Head").Return(func() *flow.Header { @@ -162,12 +172,12 @@ func (s *TransactionStatusSuite) SetupTest() { }, nil) return s.tempSnapshot - }, nil) + }, nil).Maybe() s.finalSnapshot.On("Head").Return(func() *flow.Header { finalizedHeader := s.finalizedBlock.Header return finalizedHeader - }, nil) + }, nil).Maybe() s.blockTracker.On("GetStartHeightFromBlockID", mock.Anything).Return(func(_ flow.Identifier) (uint64, error) { finalizedHeader := s.finalizedBlock.Header @@ -235,7 +245,7 @@ func (s *TransactionStatusSuite) addNewFinalizedBlock(parent *flow.Header, notif } } -// TestSubscribeTransactionStatusHappyCase tests the functionality of the SubscribeTransactionStatuses method in the Backend. +// TestSubscribeTransactionStatusHappyCase tests the functionality of the SubscribeTransactionStatusesFromStartBlockID method in the Backend. // It covers the emulation of transaction stages from pending to sealed, and receiving status updates. func (s *TransactionStatusSuite) TestSubscribeTransactionStatusHappyCase() { ctx, cancel := context.WithCancel(context.Background()) @@ -314,7 +324,7 @@ func (s *TransactionStatusSuite) TestSubscribeTransactionStatusHappyCase() { } // 1. Subscribe to transaction status and receive the first message with pending status - sub := s.backend.SubscribeTransactionStatuses(ctx, &transaction.TransactionBody, entities.EventEncodingVersion_CCF_V0) + sub := s.backend.SendAndSubscribeTransactionStatuses(ctx, &transaction.TransactionBody, entities.EventEncodingVersion_CCF_V0) checkNewSubscriptionMessage(sub, flow.TransactionStatusPending) // 2. Make transaction reference block sealed, and add a new finalized block that includes the transaction @@ -349,7 +359,7 @@ func (s *TransactionStatusSuite) TestSubscribeTransactionStatusHappyCase() { }, 100*time.Millisecond, "timed out waiting for subscription to shutdown") } -// TestSubscribeTransactionStatusExpired tests the functionality of the SubscribeTransactionStatuses method in the Backend +// TestSubscribeTransactionStatusExpired tests the functionality of the SubscribeTransactionStatusesFromStartBlockID method in the Backend // when transaction become expired func (s *TransactionStatusSuite) TestSubscribeTransactionStatusExpired() { ctx, cancel := context.WithCancel(context.Background()) @@ -380,7 +390,7 @@ func (s *TransactionStatusSuite) TestSubscribeTransactionStatusExpired() { } // Subscribe to transaction status and receive the first message with pending status - sub := s.backend.SubscribeTransactionStatuses(ctx, &transaction.TransactionBody, entities.EventEncodingVersion_CCF_V0) + sub := s.backend.SendAndSubscribeTransactionStatuses(ctx, &transaction.TransactionBody, entities.EventEncodingVersion_CCF_V0) checkNewSubscriptionMessage(sub, flow.TransactionStatusPending) // Generate 600 blocks without transaction included and check, that transaction still pending