Send and subscribe transaction statuses endpoint

access/grpc/client.go

@@ -204,6 +204,13 @@ func (c *Client) GetTransactionResultsByBlockID(ctx context.Context, blockID flo
 	return c.grpc.GetTransactionResultsByBlockID(ctx, blockID)
 }
 
+func (c *Client) SendAndSubscribeTransactionStatuses(
+	ctx context.Context,
+	tx flow.Transaction,
+) (<-chan flow.TransactionStatus, <-chan error, error) {
+	return c.grpc.SendAndSubscribeTransactionStatuses(ctx, tx)
+}
+
 func (c *Client) GetAccount(ctx context.Context, address flow.Address) (*flow.Account, error) {
 	return c.grpc.GetAccount(ctx, address)
 }

access/grpc/grpc.go

@@ -947,7 +947,6 @@ func (c *BaseClient) GetExecutionDataByBlockID(
 	}
 
 	return convert.MessageToBlockExecutionData(ed.GetBlockExecutionData())
-
 }
 
 func (c *BaseClient) SubscribeExecutionDataByBlockID(
@@ -1129,3 +1128,62 @@ func (c *BaseClient) subscribeEvents(
 
 	return sub, errChan, nil
 }
+
+func (c *BaseClient) SendAndSubscribeTransactionStatuses(
+	ctx context.Context,
+	tx flow.Transaction,
+	opts ...grpc.CallOption,
+) (<-chan flow.TransactionStatus, <-chan error, error) {
+	txMsg, err := convert.TransactionToMessage(tx)
+	if err != nil {
+		return nil, nil, newEntityToMessageError(entityTransaction, err)
+	}
+
+	req := &access.SendAndSubscribeTransactionStatusesRequest{
+		Transaction:          txMsg,
+		EventEncodingVersion: c.eventEncoding,
+	}
+
+	subscribeClient, err := c.rpcClient.SendAndSubscribeTransactionStatuses(ctx, req, opts...)
+	if err != nil {
+		return nil, nil, newRPCError(err)
+	}
+
+	txStatusChan := make(chan flow.TransactionStatus)
+	errChan := make(chan error)
+
+	sendErr := func(err error) {
+		select {
+		case <-ctx.Done():
+		case errChan <- err:
+		}
+	}
+
+	go func() {
+		defer close(txStatusChan)
+		defer close(errChan)
+
+		for {
+			// Receive the next txStatus response
+			txStatusResponse, err := subscribeClient.Recv()
+			if err != nil {
+				if err == io.EOF {
+					// End of stream, return gracefully
+					return
+				}
+				sendErr(fmt.Errorf("error receiving blockHeader: %w", err))
+				return
+			}
+
+			txStatus := flow.TransactionStatus(txStatusResponse.GetTransactionResults().Status)
+
+			select {
+			case <-ctx.Done():
+				return
+			case txStatusChan <- txStatus:
+			}
+		}
+	}()
+
+	return txStatusChan, errChan, nil
+}

access/grpc/grpc_test.go

@@ -2149,3 +2149,145 @@ func (m *mockExecutionDataStream) Recv() (*executiondata.SubscribeExecutionDataR
 
 	return m.responses[m.offset], nil
 }
+
+func TestClient_SendAndSubscribeTransactionStatuses(t *testing.T) {
+	transactions := test.TransactionGenerator()
+
+	generateTransactionStatusResponses := func(count uint64, encodingVersion flow.EventEncodingVersion) []*access.SendAndSubscribeTransactionStatusesResponse {
+		var resTransactionStatuses []*access.SendAndSubscribeTransactionStatusesResponse
+		results := test.TransactionResultGenerator(encodingVersion)
+
+		for i := uint64(0); i < count; i++ {
+			expectedResult := results.New()
+			transactionResult, _ := convert.TransactionResultToMessage(expectedResult, encodingVersion)
+
+			response := &access.SendAndSubscribeTransactionStatusesResponse{
+				TransactionResults: transactionResult,
+			}
+
+			resTransactionStatuses = append(resTransactionStatuses, response)
+		}
+
+		return resTransactionStatuses
+	}
+
+	t.Run("Happy Path - CCF", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		responseCount := uint64(100)
+		tx := transactions.New()
+
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx:       ctx,
+			responses: generateTransactionStatusResponses(responseCount, flow.EventEncodingVersionCCF),
+		}
+
+		rpc.On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).Return(stream, nil)
+
+		txStatusesCh, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, *tx)
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoErrors(t, errCh, wg.Done)
+
+		for i := uint64(0); i < responseCount; i++ {
+			actualTxStatus := <-txStatusesCh
+			expectedTxStatus := flow.TransactionStatus(stream.responses[i].GetTransactionResults().Status)
+			require.Equal(t, expectedTxStatus, actualTxStatus)
+		}
+		cancel()
+
+		wg.Wait()
+	}))
+
+	t.Run("Happy Path - JSON-CDC", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		responseCount := uint64(100)
+		tx := transactions.New()
+
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx:       ctx,
+			responses: generateTransactionStatusResponses(responseCount, flow.EventEncodingVersionJSONCDC),
+		}
+
+		rpc.On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).Return(stream, nil)
+
+		txStatusesCh, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, *tx)
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoErrors(t, errCh, wg.Done)
+
+		for i := uint64(0); i < responseCount; i++ {
+			actualTxStatus := <-txStatusesCh
+			expectedTxStatus := flow.TransactionStatus(stream.responses[i].GetTransactionResults().Status)
+			require.Equal(t, expectedTxStatus, actualTxStatus)
+		}
+		cancel()
+
+		wg.Wait()
+	}))
+
+	t.Run("Stream returns error", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx: ctx,
+			err: status.Error(codes.Internal, "internal error"),
+		}
+
+		rpc.
+			On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).
+			Return(stream, nil)
+
+		txStatusChan, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, flow.Transaction{})
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoTxStatuses(t, txStatusChan, wg.Done)
+
+		errorCount := 0
+		for e := range errCh {
+			require.Error(t, e)
+			require.ErrorIs(t, e, stream.err)
+			errorCount += 1
+		}
+		cancel()
+
+		require.Equalf(t, 1, errorCount, "only 1 error is expected")
+
+		wg.Wait()
+	}))
+
+}
+
+type mockTransactionStatusesClientStream struct {
+	grpc.ClientStream
+
+	ctx       context.Context
+	err       error
+	offset    int
+	responses []*access.SendAndSubscribeTransactionStatusesResponse
+}
+
+func (m *mockTransactionStatusesClientStream) Recv() (*access.SendAndSubscribeTransactionStatusesResponse, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+
+	if m.offset >= len(m.responses) {
+		<-m.ctx.Done()
+		return nil, io.EOF
+	}
+	defer func() { m.offset++ }()
+
+	return m.responses[m.offset], nil
+}
+
+func assertNoTxStatuses[TxStatus any](t *testing.T, txStatusChan <-chan TxStatus, done func()) {
+	defer done()
+	for range txStatusChan {
+		require.FailNow(t, "should not receive txStatus")
+	}
+}