[Access] Pending status returns immediately from send and subscribe transaction statuses

[Guitarheroua]

Closes #6573, part of #6767

This PR implements functionality to return a Pending status immediately upon the start of a subscription.

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 32.77311% with 80 lines in your changes missing coverage. Please review.

Project coverage is 41.13%. Comparing base (0f4013e) to head (a162ba2).

Files with missing lines	Patch %	Lines
access/mock/api.go	0.00%	44 Missing ⚠️
.../access/rpc/backend/backend_stream_transactions.go	57.57%	23 Missing and 5 partials ⚠️
cmd/util/cmd/run-script/cmd.go	0.00%	7 Missing ⚠️
access/handler.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #6737   +/-   ##
=======================================
  Coverage   41.12%   41.13%           
=======================================
  Files        2107     2107           
  Lines      185335   185422   +87     
=======================================
+ Hits        76228    76278   +50     
- Misses     102706   102739   +33     
- Partials     6401     6405    +4     

Flag	Coverage Δ
unittests	41.13% <32.77%> (+<0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[peterargue]

added a few small comments, but otherwise looks good

[peterargue]

added a couple more comments, but otherwise looks good.

[peterargue]

looks good other than missing irrecoverable error

[peterargue]

please wrap this with some context so it's obvious why the node crashed

[durkmurder]

Suggested change

	SubscribeTransactionStatusesFromStartBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
	SubscribeTransactionStatusesFromBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription

To me it seems that "Start" can be omitted but that is just a preference for shorter function names.

[durkmurder]

Suggested change

	SubscribeTransactionStatusesFromStartHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
	SubscribeTransactionStatusesFromHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription

[durkmurder]

Suggested change

	// 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.
	// 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 the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). When the transaction reaches one of
	// these final states, the subscription will automatically terminate.

[durkmurder]

Suggested change

	// 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.
	// 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 the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). When the transaction reaches one of
	// these final states, the subscription will automatically terminate.

[durkmurder]

Suggested change

	// 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.
	// 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 the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). When the transaction reaches one of
	// these final states, the subscription will automatically terminate.

[durkmurder]

Suggested change

	// 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.
	// 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 the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). Once the final status has been reached, the subscription
	// automatically terminates.

[durkmurder]

🤯 I find this logic very hard to follow if you are not familiar with it. It includes a lot of state changes and mutating fields. I am wondering if we can make it easier to follow by restructuring the code path a bit.

A proposal to this:

// 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")
		}
	}

	return b.subscriptionHandler.Subscribe(ctx, nextHeight, b.getTransactionStatusResponse(txID, referenceBlockID, requiredEventEncodingVersion, shouldTriggerPending))
}

I have moved creation of metadata closer to the usage, it's easier to understand that way why it's valid. I had hard time understanding how are we mutating the txInfo field.

// getTransactionStatusResponse returns a callback function that produces transaction status
// subscription responses based on new blocks.
func (b *backendSubscribeTransactions) getTransactionStatusResponse(
	txID, referenceBlockID flow.Identifier,
	requiredEventEncodingVersion entities.EventEncodingVersion,
	shouldTriggerPending bool,
) func(context.Context, uint64) (interface{}, error) {
	txInfo := &transactionSubscriptionMetadata{
		TransactionResult: &access.TransactionResult{
			TransactionID: txID,
			BlockID:       flow.ZeroID,
		},
		txReferenceBlockID:   referenceBlockID,
		blockWithTx:          nil,
		eventEncodingVersion: requiredEventEncodingVersion,
	}
	triggerPendingOnce := atomic.NewBool(false)
	return func(ctx context.Context, height uint64) (interface{}, error) {
		err := b.checkBlockReady(height)
		if err != nil {
			return nil, err
		}

		if shouldTriggerPending && triggerPendingOnce.CompareAndSwap(false, true) {
			// 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)
		}

		if txInfo.IsFinal() {
			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.
		if txInfo.blockWithTx == nil {
			// Search for transaction`s block information.
			txInfo.blockWithTx,
				txInfo.BlockID,
				txInfo.BlockHeight,
				txInfo.CollectionID,
				err = b.searchForTransactionBlockInfo(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)
				}
			}
		}

		// Get old status here, as it could be replaced by status from founded tx result
		prevTxStatus := txInfo.Status

		// Check, if transaction executed and transaction result already available
		if txInfo.blockWithTx == nil {
			txInfo.Status, err = b.txLocalDataProvider.DeriveUnknownTransactionStatus(txInfo.txReferenceBlockID)
			if err != nil {
				if !errors.Is(err, state.ErrUnknownSnapshotReference) {
					irrecoverable.Throw(ctx, err)
				}
				return nil, rpc.ConvertStorageError(err)
			}
		} else if !txInfo.IsExecuted() {
			txResult, err := b.searchForTransactionResult(ctx, txInfo)
			if err != nil {
				return nil, status.Errorf(codes.Internal, "failed to get execution result for block %s: %v", txInfo.BlockID, err)
			}

			// If transaction result was found, fully replace it in metadata. New transaction status already included in result.
			if txResult != nil {
				txInfo.TransactionResult = txResult
			}
		}

		if prevTxStatus == txInfo.Status {
			// 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.IsExecuted())
			if err != nil {
				if !errors.Is(err, state.ErrUnknownSnapshotReference) {
					irrecoverable.Throw(ctx, err)
				}
				return nil, rpc.ConvertStorageError(err)
			}
		}

		// If the old and new transaction statuses are still the same, the status change should not be reported, so
		// return here with no response.
		if prevTxStatus == txInfo.Status {
			return nil, nil
		}

		return b.generateResultsWithMissingStatuses(txInfo, prevTxStatus)
	}
}

In this section I got rid from inner functions to avoid extra levels of search, they are quite simple and I believe it's easier to understand the logic when they are inlined. Additionally, I have removed extra fields and refactored other functions to pass only bare minimum data.

This lead to the simplifications to the structure:

// transactionSubscriptionMetadata holds data representing the status state for each transaction subscription.
type transactionSubscriptionMetadata struct {
	*access.TransactionResult
	txReferenceBlockID   flow.Identifier
	blockWithTx          *flow.Header
	eventEncodingVersion entities.EventEncodingVersion
}

To support new code path I have added helper functions:

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
}

Hopefully this proposal can make the underlying implementation simpler.
@peterargue would like to hear your thoughts on this as well.