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

@@ -521,6 +521,78 @@ func (c *Client) SubmitTransactionWithOptions(transaction *txnbuild.Transaction,
 	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.
+	if inner := transaction.InnerTransaction(); !opts.SkipMemoRequiredCheck && inner.Memo() == nil {
+		err = c.checkMemoRequired(inner)
+		if err != nil {
+			return
+		}
+	}
+
+	txeBase64, err := transaction.Base64()
+	if err != nil {
+		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
+		return
+	}
+
+	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.
+	if !opts.SkipMemoRequiredCheck && transaction.Memo() == nil {
+		err = c.checkMemoRequired(transaction)
+		if err != nil {
+			return
+		}
+	}
+
+	txeBase64, err := transaction.Base64()
+	if err != nil {
+		err = errors.Wrap(err, "Unable to convert transaction object to base64 string")
+		return
+	}
+
+	return c.AsyncSubmitTransactionXDR(txeBase64)
+}
+
 // Transactions returns stellar transactions (https://developers.stellar.org/api/resources/transactions/list/)
 // It can be used to return transactions for an account, a ledger,and all transactions on the network.
 func (c *Client) Transactions(request TransactionRequest) (txs hProtocol.TransactionsPage, err error) {

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/client.go

@@ -29,6 +29,16 @@ type Client struct {
 	URL string
 }
 
+type ClientInterface interface {
+	Upgrade(ctx context.Context, version int) (err error)
+	GetLedgerEntry(ctx context.Context, ledgerKey xdr.LedgerKey) (proto.GetLedgerEntryResponse, error)
+	Info(ctx context.Context) (resp *proto.InfoResponse, err error)
+	SetCursor(ctx context.Context, id string, cursor int32) (err error)
+	SubmitTransaction(ctx context.Context, envelope string) (resp *proto.TXResponse, err error)
+	WaitForNetworkSync(ctx context.Context) error
+	ManualClose(ctx context.Context) (err error)
+}
+
 // drainReponse is a helper method for draining the body stream off the http
 // response object and optionally close the stream. It would also update the
 // error but only as long as there wasn't an error before - this would allow

clients/stellarcore/metrics_client.go

@@ -0,0 +1,129 @@
+package stellarcore
+
+import (
+	"context"
+	"github.com/prometheus/client_golang/prometheus"
+	proto "github.com/stellar/go/protocols/stellarcore"
+	"github.com/stellar/go/xdr"
+	"time"
+)
+
+type ClientWithMetricsInterface interface {
+	SubmitTransaction(ctx context.Context, rawTx string, envelope xdr.TransactionEnvelope) (resp *proto.TXResponse, err error)
+}
+
+type ClientWithMetrics struct {
+	CoreClient ClientInterface
+
+	TxSubMetrics struct {
+		// SubmissionDuration exposes timing metrics about the rate and latency of
+		// submissions to stellar-core
+		SubmissionDuration *prometheus.SummaryVec
+
+		// SubmissionsCounter tracks the rate of transactions that have
+		// been submitted to this process
+		SubmissionsCounter *prometheus.CounterVec
+
+		// V0TransactionsCounter tracks the rate of v0 transaction envelopes that
+		// have been submitted to this process
+		V0TransactionsCounter *prometheus.CounterVec
+
+		// V1TransactionsCounter tracks the rate of v1 transaction envelopes that
+		// have been submitted to this process
+		V1TransactionsCounter *prometheus.CounterVec
+
+		// FeeBumpTransactionsCounter tracks the rate of fee bump transaction envelopes that
+		// have been submitted to this process
+		FeeBumpTransactionsCounter *prometheus.CounterVec
+	}
+}
+
+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) {
+	var label prometheus.Labels
+	if err != nil {
+		label = prometheus.Labels{"status": "request_error"}
+	} else if response.IsException() {
+		label = prometheus.Labels{"status": "exception"}
+	} else {
+		label = prometheus.Labels{"status": response.Status}
+	}
+
+	c.TxSubMetrics.SubmissionDuration.With(label).Observe(duration)
+	c.TxSubMetrics.SubmissionsCounter.With(label).Inc()
+
+	switch envelope.Type {
+	case xdr.EnvelopeTypeEnvelopeTypeTxV0:
+		c.TxSubMetrics.V0TransactionsCounter.With(label).Inc()
+	case xdr.EnvelopeTypeEnvelopeTypeTx:
+		c.TxSubMetrics.V1TransactionsCounter.With(label).Inc()
+	case xdr.EnvelopeTypeEnvelopeTypeTxFeeBump:
+		c.TxSubMetrics.FeeBumpTransactionsCounter.With(label).Inc()
+	}
+}
+
+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"})
+	submissionsCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "horizon",
+		Subsystem: prometheusSubsystem,
+		Name:      "submissions_count",
+		Help:      "number of submissions, sliding window = 10m",
+	}, []string{"status"})
+	v0TransactionsCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "horizon",
+		Subsystem: prometheusSubsystem,
+		Name:      "v0_count",
+		Help:      "number of v0 transaction envelopes submitted, sliding window = 10m",
+	}, []string{"status"})
+	v1TransactionsCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "horizon",
+		Subsystem: prometheusSubsystem,
+		Name:      "v1_count",
+		Help:      "number of v1 transaction envelopes submitted, sliding window = 10m",
+	}, []string{"status"})
+	feeBumpTransactionsCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "horizon",
+		Subsystem: prometheusSubsystem,
+		Name:      "feebump_count",
+		Help:      "number of fee bump transaction envelopes submitted, sliding window = 10m",
+	}, []string{"status"})
+
+	registry.MustRegister(
+		submissionDuration,
+		submissionsCounter,
+		v0TransactionsCounter,
+		v1TransactionsCounter,
+		feeBumpTransactionsCounter,
+	)
+
+	return &ClientWithMetrics{
+		CoreClient: &client,
+		TxSubMetrics: struct {
+			SubmissionDuration         *prometheus.SummaryVec
+			SubmissionsCounter         *prometheus.CounterVec
+			V0TransactionsCounter      *prometheus.CounterVec
+			V1TransactionsCounter      *prometheus.CounterVec
+			FeeBumpTransactionsCounter *prometheus.CounterVec
+		}{
+			SubmissionDuration:         submissionDuration,
+			SubmissionsCounter:         submissionsCounter,
+			V0TransactionsCounter:      v0TransactionsCounter,
+			V1TransactionsCounter:      v1TransactionsCounter,
+			FeeBumpTransactionsCounter: feeBumpTransactionsCounter,
+		},
+	}
+}

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,27 @@ 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"`
+	// DiagnosticEventsXDR is present only if Status is equal to proto.TXStatusError.
+	// DiagnosticEventsXDR is a base64-encoded slice of xdr.DiagnosticEvent
+	DiagnosticEventsXDR string `json:"diagnosticEventsXdr,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"`
+	// HttpStatus represents the corresponding http status code.
+	HttpStatus int `json:"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/horizon/internal/actions/submit_transaction.go

@@ -3,16 +3,16 @@ package actions
 import (
 	"context"
 	"encoding/hex"
+	"github.com/stellar/go/network"
+	"github.com/stellar/go/support/errors"
 	"mime"
 	"net/http"
 
-	"github.com/stellar/go/network"
 	"github.com/stellar/go/protocols/horizon"
 	"github.com/stellar/go/protocols/stellarcore"
 	hProblem "github.com/stellar/go/services/horizon/internal/render/problem"
 	"github.com/stellar/go/services/horizon/internal/resourceadapter"
 	"github.com/stellar/go/services/horizon/internal/txsub"
-	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/support/render/hal"
 	"github.com/stellar/go/support/render/problem"
 	"github.com/stellar/go/xdr"
@@ -36,7 +36,7 @@ type envelopeInfo struct {
 	parsed    xdr.TransactionEnvelope
 }
 
-func (handler SubmitTransactionHandler) extractEnvelopeInfo(raw string, passphrase string) (envelopeInfo, error) {
+func extractEnvelopeInfo(raw string, passphrase string) (envelopeInfo, error) {
 	result := envelopeInfo{raw: raw}
 	err := xdr.SafeUnmarshalBase64(raw, &result.parsed)
 	if err != nil {
@@ -59,7 +59,7 @@ func (handler SubmitTransactionHandler) extractEnvelopeInfo(raw string, passphra
 	return result, nil
 }
 
-func (handler SubmitTransactionHandler) validateBodyType(r *http.Request) error {
+func validateBodyType(r *http.Request) error {
 	c := r.Header.Get("Content-Type")
 	if c == "" {
 		return nil
@@ -137,7 +137,7 @@ func (handler SubmitTransactionHandler) response(r *http.Request, info envelopeI
 }
 
 func (handler SubmitTransactionHandler) GetResource(w HeaderWriter, r *http.Request) (interface{}, error) {
-	if err := handler.validateBodyType(r); err != nil {
+	if err := validateBodyType(r); err != nil {
 		return nil, err
 	}
 
@@ -157,7 +157,7 @@ func (handler SubmitTransactionHandler) GetResource(w HeaderWriter, r *http.Requ
 		return nil, err
 	}
 
-	info, err := handler.extractEnvelopeInfo(raw, handler.NetworkPassphrase)
+	info, err := extractEnvelopeInfo(raw, handler.NetworkPassphrase)
 	if err != nil {
 		return nil, &problem.P{
 			Type:   "transaction_malformed",