services/horizon + horizonclient: Add new async transaction submission endpoint

.gitignore

@@ -5,6 +5,7 @@
 /services/horizon/captive-core
 /services/horizon/horizon
 /services/horizon/stellar-horizon
+/bucket-cache
 .vscode
 .idea
 debug

clients/horizonclient/client.go

@@ -440,6 +440,44 @@ func (c *Client) OperationDetail(id string) (ops operations.Operation, err error
 	return ops, nil
 }
 
+// validateFeeBumpTx checks if the inner transaction has a memo or not and converts the transaction object to
+// base64 string.
+func (c *Client) validateFeeBumpTx(transaction *txnbuild.FeeBumpTransaction, opts SubmitTxOpts) (string, error) {
+	var err error
+	if inner := transaction.InnerTransaction(); !opts.SkipMemoRequiredCheck && inner.Memo() == nil {
+		err = c.checkMemoRequired(inner)
+		if err != nil {
+			return "", err
+		}
+	}
+
+	txeBase64, err := transaction.Base64()
+	if err != nil {
+		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
+		return "", err
+	}
+	return txeBase64, nil
+}
+
+// validateTx checks if the transaction has a memo or not and converts the transaction object to
+// base64 string.
+func (c *Client) validateTx(transaction *txnbuild.Transaction, opts SubmitTxOpts) (string, error) {
+	var err error
+	if !opts.SkipMemoRequiredCheck && transaction.Memo() == nil {
+		err = c.checkMemoRequired(transaction)
+		if err != nil {
+			return "", err
+		}
+	}
+
+	txeBase64, err := transaction.Base64()
+	if err != nil {
+		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
+		return "", err
+	}
+	return txeBase64, nil
+}
+
 // SubmitTransactionXDR submits a transaction represented as a base64 XDR string to the network. err can be either error object or horizon.Error object.
 // See https://developers.stellar.org/api/resources/transactions/post/
 func (c *Client) SubmitTransactionXDR(transactionXdr string) (tx hProtocol.Transaction,
@@ -469,16 +507,8 @@ func (c *Client) SubmitFeeBumpTransaction(transaction *txnbuild.FeeBumpTransacti
 func (c *Client) SubmitFeeBumpTransactionWithOptions(transaction *txnbuild.FeeBumpTransaction, opts SubmitTxOpts) (tx hProtocol.Transaction, err error) {
 	// only check if memo is required if skip is false and the inner transaction
 	// doesn't have a memo.
-	if inner := transaction.InnerTransaction(); !opts.SkipMemoRequiredCheck && inner.Memo() == nil {
-		err = c.checkMemoRequired(inner)
-		if err != nil {
-			return
-		}
-	}
-
-	txeBase64, err := transaction.Base64()
+	txeBase64, err := c.validateFeeBumpTx(transaction, opts)
 	if err != nil {
-		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
 		return
 	}
 
@@ -505,20 +535,68 @@ func (c *Client) SubmitTransaction(transaction *txnbuild.Transaction) (tx hProto
 func (c *Client) SubmitTransactionWithOptions(transaction *txnbuild.Transaction, opts SubmitTxOpts) (tx hProtocol.Transaction, err error) {
 	// only check if memo is required if skip is false and the transaction
 	// doesn't have a memo.
-	if !opts.SkipMemoRequiredCheck && transaction.Memo() == nil {
-		err = c.checkMemoRequired(transaction)
-		if err != nil {
-			return
-		}
+	txeBase64, err := c.validateTx(transaction, opts)
+	if err != nil {
+		return
 	}
 
-	txeBase64, err := transaction.Base64()
+	return c.SubmitTransactionXDR(txeBase64)
+}
+
+// AsyncSubmitTransactionXDR submits a base64 XDR transaction using the transactions_async endpoint. err can be either error object or horizon.Error object.
+func (c *Client) AsyncSubmitTransactionXDR(transactionXdr string) (txResp hProtocol.AsyncTransactionSubmissionResponse,
+	err error) {
+	request := submitRequest{endpoint: "transactions_async", transactionXdr: transactionXdr}
+	err = c.sendRequest(request, &txResp)
+	return
+}
+
+// AsyncSubmitFeeBumpTransaction submits an async fee bump transaction to the network. err can be either an
+// error object or a horizon.Error object.
+//
+// This function will always check if the destination account requires a memo in the transaction as
+// defined in SEP0029: https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0029.md
+//
+// If you want to skip this check, use SubmitTransactionWithOptions.
+func (c *Client) AsyncSubmitFeeBumpTransaction(transaction *txnbuild.FeeBumpTransaction) (txResp hProtocol.AsyncTransactionSubmissionResponse, err error) {
+	return c.AsyncSubmitFeeBumpTransactionWithOptions(transaction, SubmitTxOpts{})
+}
+
+// AsyncSubmitFeeBumpTransactionWithOptions submits an async fee bump transaction to the network, allowing
+// you to pass SubmitTxOpts. err can be either an error object or a horizon.Error object.
+func (c *Client) AsyncSubmitFeeBumpTransactionWithOptions(transaction *txnbuild.FeeBumpTransaction, opts SubmitTxOpts) (txResp hProtocol.AsyncTransactionSubmissionResponse, err error) {
+	// only check if memo is required if skip is false and the inner transaction
+	// doesn't have a memo.
+	txeBase64, err := c.validateFeeBumpTx(transaction, opts)
 	if err != nil {
-		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
 		return
 	}
 
-	return c.SubmitTransactionXDR(txeBase64)
+	return c.AsyncSubmitTransactionXDR(txeBase64)
+}
+
+// AsyncSubmitTransaction submits an async transaction to the network. err can be either an
+// error object or a horizon.Error object.
+//
+// This function will always check if the destination account requires a memo in the transaction as
+// defined in SEP0029: https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0029.md
+//
+// If you want to skip this check, use SubmitTransactionWithOptions.
+func (c *Client) AsyncSubmitTransaction(transaction *txnbuild.Transaction) (txResp hProtocol.AsyncTransactionSubmissionResponse, err error) {
+	return c.AsyncSubmitTransactionWithOptions(transaction, SubmitTxOpts{})
+}
+
+// AsyncSubmitTransactionWithOptions submits an async transaction to the network, allowing
+// you to pass SubmitTxOpts. err can be either an error object or a horizon.Error object.
+func (c *Client) AsyncSubmitTransactionWithOptions(transaction *txnbuild.Transaction, opts SubmitTxOpts) (txResp hProtocol.AsyncTransactionSubmissionResponse, err error) {
+	// only check if memo is required if skip is false and the transaction
+	// doesn't have a memo.
+	txeBase64, err := c.validateTx(transaction, opts)
+	if err != nil {
+		return
+	}
+
+	return c.AsyncSubmitTransactionXDR(txeBase64)
 }
 
 // Transactions returns stellar transactions (https://developers.stellar.org/api/resources/transactions/list/)

clients/horizonclient/internal.go

@@ -27,7 +27,10 @@ func decodeResponse(resp *http.Response, object interface{}, horizonUrl string,
 	}
 	setCurrentServerTime(u.Hostname(), resp.Header["Date"], clock)
 
-	if !(resp.StatusCode >= 200 && resp.StatusCode < 300) {
+	// While this part of code assumes that any error < 200 or error >= 300 is a Horizon problem, it is not
+	// true for the response from /transactions_async endpoint which does give these codes for certain responses
+	// from core.
+	if !(resp.StatusCode >= 200 && resp.StatusCode < 300) && (resp.Request == nil || resp.Request.URL == nil || resp.Request.URL.Path != "/transactions_async") {
 		horizonError := &Error{
 			Response: resp,
 		}

clients/horizonclient/main.go

@@ -194,6 +194,11 @@ type ClientInterface interface {
 	SubmitTransactionWithOptions(transaction *txnbuild.Transaction, opts SubmitTxOpts) (hProtocol.Transaction, error)
 	SubmitFeeBumpTransaction(transaction *txnbuild.FeeBumpTransaction) (hProtocol.Transaction, error)
 	SubmitTransaction(transaction *txnbuild.Transaction) (hProtocol.Transaction, error)
+	AsyncSubmitTransactionXDR(transactionXdr string) (hProtocol.AsyncTransactionSubmissionResponse, error)
+	AsyncSubmitFeeBumpTransactionWithOptions(transaction *txnbuild.FeeBumpTransaction, opts SubmitTxOpts) (hProtocol.AsyncTransactionSubmissionResponse, error)
+	AsyncSubmitTransactionWithOptions(transaction *txnbuild.Transaction, opts SubmitTxOpts) (hProtocol.AsyncTransactionSubmissionResponse, error)
+	AsyncSubmitFeeBumpTransaction(transaction *txnbuild.FeeBumpTransaction) (hProtocol.AsyncTransactionSubmissionResponse, error)
+	AsyncSubmitTransaction(transaction *txnbuild.Transaction) (hProtocol.AsyncTransactionSubmissionResponse, error)
 	Transactions(request TransactionRequest) (hProtocol.TransactionsPage, error)
 	TransactionDetail(txHash string) (hProtocol.Transaction, error)
 	OrderBook(request OrderBookRequest) (hProtocol.OrderBookSummary, error)

clients/horizonclient/mocks.go

@@ -121,6 +121,36 @@
 	return a.Get(0).(hProtocol.Transaction), a.Error(1)
 }
 
+// AsyncSubmitTransactionXDR is a mocking method
+func (m *MockClient) AsyncSubmitTransactionXDR(transactionXdr string) (hProtocol.AsyncTransactionSubmissionResponse, error) {
+	a := m.Called(transactionXdr)
+	return a.Get(0).(hProtocol.AsyncTransactionSubmissionResponse), a.Error(1)
+}
+
+// AsyncSubmitFeeBumpTransaction is a mocking method
+func (m *MockClient) AsyncSubmitFeeBumpTransaction(transaction *txnbuild.FeeBumpTransaction) (hProtocol.AsyncTransactionSubmissionResponse, error) {
+	a := m.Called(transaction)
+	return a.Get(0).(hProtocol.AsyncTransactionSubmissionResponse), a.Error(1)
+}
+
+// AsyncSubmitTransaction is a mocking method
+func (m *MockClient) AsyncSubmitTransaction(transaction *txnbuild.Transaction) (hProtocol.AsyncTransactionSubmissionResponse, error) {
+	a := m.Called(transaction)
+	return a.Get(0).(hProtocol.AsyncTransactionSubmissionResponse), a.Error(1)
+}
+
+// AsyncSubmitFeeBumpTransactionWithOptions is a mocking method
+func (m *MockClient) AsyncSubmitFeeBumpTransactionWithOptions(transaction *txnbuild.FeeBumpTransaction, opts SubmitTxOpts) (hProtocol.AsyncTransactionSubmissionResponse, error) {
+	a := m.Called(transaction, opts)
+	return a.Get(0).(hProtocol.AsyncTransactionSubmissionResponse), a.Error(1)
+}
+
+// AsyncSubmitTransactionWithOptions is a mocking method
+func (m *MockClient) AsyncSubmitTransactionWithOptions(transaction *txnbuild.Transaction, opts SubmitTxOpts) (hProtocol.AsyncTransactionSubmissionResponse, error) {
+	a := m.Called(transaction, opts)
+	return a.Get(0).(hProtocol.AsyncTransactionSubmissionResponse), a.Error(1)
+}
+
 // Transactions is a mocking method
 func (m *MockClient) Transactions(request TransactionRequest) (hProtocol.TransactionsPage, error) {
 	a := m.Called(request)

clients/stellarcore/metrics_client.go

@@ -0,0 +1,78 @@
+package stellarcore
+
+import (
+	"context"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	proto "github.com/stellar/go/protocols/stellarcore"
+	"github.com/stellar/go/xdr"
+)
+
+type ClientWithMetrics interface {
+	SubmitTransaction(ctx context.Context, rawTx string, envelope xdr.TransactionEnvelope) (resp *proto.TXResponse, err error)
+}
+
+type clientWithMetrics struct {
+	CoreClient Client
+
+	TxSubMetrics struct {
+		// SubmissionDuration exposes timing metrics about the rate and latency of
+		// submissions to stellar-core
+		SubmissionDuration *prometheus.SummaryVec
+	}
+}
+
+func (c *clientWithMetrics) SubmitTransaction(ctx context.Context, rawTx string, envelope xdr.TransactionEnvelope) (*proto.TXResponse, error) {
+	startTime := time.Now()
+	response, err := c.CoreClient.SubmitTransaction(ctx, rawTx)
+	c.updateTxSubMetrics(time.Since(startTime).Seconds(), envelope, response, err)
+
+	return response, err
+}
+
+func (c *clientWithMetrics) updateTxSubMetrics(duration float64, envelope xdr.TransactionEnvelope, response *proto.TXResponse, err error) {
+	label := prometheus.Labels{}
+	if err != nil {
+		label["status"] = "request_error"
+	} else if response.IsException() {
+		label["status"] = "exception"
+	} else {
+		label["status"] = response.Status
+	}
+
+	switch envelope.Type {
+	case xdr.EnvelopeTypeEnvelopeTypeTxV0:
+		label["envelope_type"] = "v0"
+	case xdr.EnvelopeTypeEnvelopeTypeTx:
+		label["envelope_type"] = "v1"
+	case xdr.EnvelopeTypeEnvelopeTypeTxFeeBump:
+		label["envelope_type"] = "fee-bump"
+	}
+
+	c.TxSubMetrics.SubmissionDuration.With(label).Observe(duration)
+}
+
+func NewClientWithMetrics(client Client, registry *prometheus.Registry, prometheusSubsystem string) ClientWithMetrics {
+	submissionDuration := prometheus.NewSummaryVec(prometheus.SummaryOpts{
+		Namespace:  "horizon",
+		Subsystem:  prometheusSubsystem,
+		Name:       "submission_duration_seconds",
+		Help:       "submission durations to Stellar-Core, sliding window = 10m",
+		Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+	}, []string{"status", "envelope_type"})
+
+	registry.MustRegister(
+		submissionDuration,
+	)
+
+	return &clientWithMetrics{
+		CoreClient: client,
+		TxSubMetrics: struct {
+			SubmissionDuration *prometheus.SummaryVec
+		}{
+			SubmissionDuration: submissionDuration,
+		},
+	}
+}

clients/stellarcore/mocks.go

@@ -0,0 +1,22 @@
+package stellarcore
+
+import (
+	"context"
+	proto "github.com/stellar/go/protocols/stellarcore"
+	"github.com/stellar/go/xdr"
+	"github.com/stretchr/testify/mock"
+)
+
+type MockClientWithMetrics struct {
+	mock.Mock
+}
+
+// SubmitTransaction mocks the SubmitTransaction method
+func (m *MockClientWithMetrics) SubmitTransaction(ctx context.Context, rawTx string, envelope xdr.TransactionEnvelope) (*proto.TXResponse, error) {
+	args := m.Called(ctx, rawTx, envelope)
+	return args.Get(0).(*proto.TXResponse), args.Error(1)
+}
+
+func (m *MockClientWithMetrics) UpdateTxSubMetrics(duration float64, envelope xdr.TransactionEnvelope, response *proto.TXResponse, err error) {
+	m.Called(duration, envelope, response, err)
+}

protocols/horizon/main.go

@@ -567,6 +567,22 @@ type InnerTransaction struct {
 	MaxFee     int64    `json:"max_fee,string"`
 }
 
+// AsyncTransactionSubmissionResponse represents the response returned by Horizon
+// when using the transaction-async endpoint.
+type AsyncTransactionSubmissionResponse struct {
+	// ErrorResultXDR is present only if Status is equal to proto.TXStatusError.
+	// ErrorResultXDR is a TransactionResult xdr string which contains details on why
+	// the transaction could not be accepted by stellar-core.
+	ErrorResultXDR string `json:"errorResultXdr,omitempty"`
+	// TxStatus represents the status of the transaction submission returned by stellar-core.
+	// It can be one of: proto.TXStatusPending, proto.TXStatusDuplicate,
+	// proto.TXStatusTryAgainLater, or proto.TXStatusError.
+	TxStatus string `json:"tx_status"`
+	// Hash is a hash of the transaction which can be used to look up whether
+	// the transaction was included in the ledger.
+	Hash string `json:"hash"`
+}
+
 // MarshalJSON implements a custom marshaler for Transaction.
 // The memo field should be omitted if and only if the
 // memo_type is "none".

services/friendbot/internal/friendbot_handler.go

@@ -1,12 +1,12 @@
 package internal
 
 import (
+	"github.com/stellar/go/support/render/httpjson"
 	"net/http"
 	"net/url"
 
 	"github.com/stellar/go/protocols/horizon"
 	"github.com/stellar/go/strkey"
-	"github.com/stellar/go/support/render/hal"
 	"github.com/stellar/go/support/render/problem"
 )
 
@@ -23,7 +23,7 @@ func (handler *FriendbotHandler) Handle(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	hal.Render(w, *result)
+	httpjson.Render(w, *result, httpjson.HALJSON)
 }
 
 // doHandle is just a convenience method that returns the object to be rendered

services/horizon/internal/actions/claimable_balance.go

@@ -20,6 +20,10 @@ import (
 // GetClaimableBalanceByIDHandler is the action handler for all end-points returning a claimable balance.
 type GetClaimableBalanceByIDHandler struct{}
 
+func (handler GetClaimableBalanceByIDHandler) HttpStatus(resp interface{}) int {
+	return http.StatusOK
+}
+
 // ClaimableBalanceQuery query struct for claimables_balances/id end-point
 type ClaimableBalanceQuery struct {
 	ID string `schema:"id" valid:"claimableBalanceID,required"`

services/horizon/internal/actions/fee_stats.go

@@ -12,6 +12,10 @@ import (
 type FeeStatsHandler struct {
 }
 
+func (handler FeeStatsHandler) HttpStatus(resp interface{}) int {
+	return http.StatusOK
+}
+
 // GetResource fee stats resource
 func (handler FeeStatsHandler) GetResource(w HeaderWriter, r *http.Request) (interface{}, error) {
 	feeStats := horizon.FeeStats{}

services/horizon/internal/actions/ledger.go

@@ -59,6 +59,10 @@ type GetLedgerByIDHandler struct {
 	LedgerState *ledger.State
 }
 
+func (handler GetLedgerByIDHandler) HttpStatus(resp interface{}) int {
+	return http.StatusOK
+}
+
 func (handler GetLedgerByIDHandler) GetResource(w HeaderWriter, r *http.Request) (interface{}, error) {
 	qp := LedgerByIDQuery{}
 	err := getParams(&qp, r)