[Access] Add implementation for transaction statuses data providers

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?

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 {

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

@@ -532,7 +532,32 @@ func (*api) SubscribeBlockDigestsFromLatest(
 	return nil
 }
 
-func (*api) SubscribeTransactionStatuses(
+func (a *api) SubscribeTransactionStatusesFromStartBlockID(
+	_ context.Context,
+	_ flow.Identifier,
+	_ flow.Identifier,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+
+func (a *api) SubscribeTransactionStatusesFromStartHeight(
+	_ context.Context,
+	_ flow.Identifier,
+	_ uint64,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+
+func (a *api) SubscribeTransactionStatusesFromLatest(
+	_ context.Context,
+	_ flow.Identifier,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+func (a *api) SendAndSubscribeTransactionStatuses(
 	_ context.Context,
 	_ *flow.TransactionBody,
 	_ entities.EventEncodingVersion,

cmd/util/cmd/verify_execution_result/cmd.go

@@ -53,27 +53,33 @@ func run(*cobra.Command, []string) {
 	chainID := flow.ChainID(flagChain)
 	_ = chainID.Chain()
 
+	lg := log.With().
+		Str("chain", string(chainID)).
+		Str("datadir", flagDatadir).
+		Str("chunk_data_pack_dir", flagChunkDataPackDir).
+		Logger()
+
 	if flagFromTo != "" {
 		from, to, err := parseFromTo(flagFromTo)
 		if err != nil {
-			log.Fatal().Err(err).Msg("could not parse from_to")
+			lg.Fatal().Err(err).Msg("could not parse from_to")
 		}
 
-		log.Info().Msgf("verifying range from %d to %d", from, to)
+		lg.Info().Msgf("verifying range from %d to %d", from, to)
 		err = verifier.VerifyRange(from, to, chainID, flagDatadir, flagChunkDataPackDir)
 		if err != nil {
-			log.Fatal().Err(err).Msgf("could not verify range from %d to %d", from, to)
+			lg.Fatal().Err(err).Msgf("could not verify range from %d to %d", from, to)
 		}
-		log.Info().Msgf("successfully verified range from %d to %d", from, to)
+		lg.Info().Msgf("successfully verified range from %d to %d", from, to)
 
 	} else {
-		log.Info().Msgf("verifying last %d sealed blocks", flagLastK)
+		lg.Info().Msgf("verifying last %d sealed blocks", flagLastK)
 		err := verifier.VerifyLastKHeight(flagLastK, chainID, flagDatadir, flagChunkDataPackDir)
 		if err != nil {
-			log.Fatal().Err(err).Msg("could not verify last k height")
+			lg.Fatal().Err(err).Msg("could not verify last k height")
 		}
 
-		log.Info().Msgf("successfully verified last %d sealed blocks", flagLastK)
+		lg.Info().Msgf("successfully verified last %d sealed blocks", flagLastK)
 	}
 }
 

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

@@ -135,7 +135,6 @@ func parseAccountStatusesArguments(
 	eventFilterConfig state_stream.EventFilterConfig,
 ) (accountStatusesArguments, error) {
 	var args accountStatusesArguments
-	//var err error
 
 	// Check for mutual exclusivity of start_block_id and start_block_height early
 	startBlockIDIn, hasStartBlockID := arguments["start_block_id"]

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

@@ -76,7 +76,6 @@ func (s *EventsProviderSuite) TestEventsDataProvider_HappyPath() {
 			Events:         expectedEvents,
 			BlockTimestamp: s.rootBlock.Header.Timestamp,
 		})
-
 	}
 
 	testHappyPath(

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

@@ -103,8 +103,7 @@ func (s *DataProviderFactoryImpl) NewDataProvider(
 	case AccountStatusesTopic:
 		return NewAccountStatusesDataProvider(ctx, s.logger, s.stateStreamApi, topic, arguments, ch, s.chain, s.eventFilterConfig, s.heartbeatInterval)
 	case TransactionStatusesTopic:
-		// TODO: Implemented handlers for each topic should be added in respective case
-		return nil, fmt.Errorf(`topic "%s" not implemented yet`, topic)
+		return NewTransactionStatusesDataProvider(ctx, s.logger, s.accessApi, topic, arguments, ch)
 	default:
 		return nil, fmt.Errorf("unsupported topic \"%s\"", topic)
 	}

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

@@ -132,6 +132,17 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 				s.stateStreamApi.AssertExpectations(s.T())
 			},
 		},
+		{
+			name:      "transaction statuses topic",
+			topic:     TransactionStatusesTopic,
+			arguments: models.Arguments{},
+			setupSubscription: func() {
+				s.setupSubscription(s.accessApi.On("SubscribeTransactionStatusesFromLatest", mock.Anything, mock.Anything, mock.Anything))
+			},
+			assertExpectations: func() {
+				s.stateStreamApi.AssertExpectations(s.T())
+			},
+		},
 	}
 
 	for _, test := range testCases {