Result of tx processing returned as QueuedTxResult

Currently it is quite easy to introduce concurrency issues while working with transaction object. For example, race issue will exist every time while transaction is processed in a separate goroutine and caller will try to check for an error before event to Done channel is sent. This change removes all the data that is updated on transaction and leaves it with ID, Args and Context (which is not used at the moment). Signed-off-by: Dmitry Shulyak <[email protected]>
status-im · Jan 31, 2018 · f086496 · f086496
1 parent 5383182
commit f086496
Show file tree

Hide file tree

Showing 13 changed files with 172 additions and 159 deletions.
diff --git a/e2e/transactions/transactions_test.go b/e2e/transactions/transactions_test.go
@@ -495,7 +495,7 @@ func (s *TransactionsTestSuite) TestDiscardQueuedTransaction() {
 			log.Info("transaction return event received", "id", event["id"].(string))
 
 			receivedErrMessage := event["error_message"].(string)
-			expectedErrMessage := queue.ErrQueuedTxDiscarded.Error()
+			expectedErrMessage := transactions.ErrQueuedTxDiscarded.Error()
 			s.Equal(receivedErrMessage, expectedErrMessage)
 
 			receivedErrCode := event["error_code"].(string)
@@ -511,7 +511,7 @@ func (s *TransactionsTestSuite) TestDiscardQueuedTransaction() {
 		To:    common.ToAddress(TestConfig.Account2.Address),
 		Value: (*hexutil.Big)(big.NewInt(1000000000000)),
 	})
-	s.EqualError(err, queue.ErrQueuedTxDiscarded.Error(), "transaction is expected to be discarded")
+	s.EqualError(err, transactions.ErrQueuedTxDiscarded.Error(), "transaction is expected to be discarded")
 
 	select {
 	case <-completeQueuedTransaction:
@@ -659,7 +659,7 @@ func (s *TransactionsTestSuite) TestDiscardMultipleQueuedTransactions() {
 			log.Info("transaction return event received", "id", event["id"].(string))
 
 			receivedErrMessage := event["error_message"].(string)
-			expectedErrMessage := queue.ErrQueuedTxDiscarded.Error()
+			expectedErrMessage := transactions.ErrQueuedTxDiscarded.Error()
 			s.Equal(receivedErrMessage, expectedErrMessage)
 
 			receivedErrCode := event["error_code"].(string)
@@ -681,7 +681,7 @@ func (s *TransactionsTestSuite) TestDiscardMultipleQueuedTransactions() {
 			To:    common.ToAddress(TestConfig.Account2.Address),
 			Value: (*hexutil.Big)(big.NewInt(1000000000000)),
 		})
-		require.EqualError(err, queue.ErrQueuedTxDiscarded.Error())
+		require.EqualError(err, transactions.ErrQueuedTxDiscarded.Error())
 		require.Equal(gethcommon.Hash{}, txHashCheck, "transaction returned hash, while it shouldn't")
 	}
 

diff --git a/geth/api/api.go b/geth/api/api.go
@@ -172,7 +172,7 @@ func (api *StatusAPI) CompleteTransaction(id common.QueuedTxID, password string)
 }
 
 // CompleteTransactions instructs backend to complete sending of multiple transactions
-func (api *StatusAPI) CompleteTransactions(ids []common.QueuedTxID, password string) map[common.QueuedTxID]common.RawCompleteTransactionResult {
+func (api *StatusAPI) CompleteTransactions(ids []common.QueuedTxID, password string) map[common.QueuedTxID]common.TransactionResult {
 	return api.b.txQueueManager.CompleteTransactions(ids, password)
 }
 

diff --git a/geth/api/backend.go b/geth/api/backend.go
@@ -200,22 +200,19 @@ func (m *StatusBackend) CallRPC(inputJSON string) string {
 }
 
 // SendTransaction creates a new transaction and waits until it's complete.
-func (m *StatusBackend) SendTransaction(ctx context.Context, args common.SendTxArgs) (gethcommon.Hash, error) {
+func (m *StatusBackend) SendTransaction(ctx context.Context, args common.SendTxArgs) (hash gethcommon.Hash, err error) {
 	if ctx == nil {
 		ctx = context.Background()
 	}
-
 	tx := common.CreateTransaction(ctx, args)
-
-	if err := m.txQueueManager.QueueTransaction(tx); err != nil {
-		return gethcommon.Hash{}, err
+	if err = m.txQueueManager.QueueTransaction(tx); err != nil {
+		return hash, err
 	}
-
-	if err := m.txQueueManager.WaitForTransaction(tx); err != nil {
-		return gethcommon.Hash{}, err
+	rst := m.txQueueManager.WaitForTransaction(tx)
+	if rst.Error != nil {
+		return hash, rst.Error
 	}
-
-	return tx.Hash, nil
+	return rst.Hash, nil
 }
 
 // CompleteTransaction instructs backend to complete sending of a given transaction
@@ -224,7 +221,7 @@ func (m *StatusBackend) CompleteTransaction(id common.QueuedTxID, password strin
 }
 
 // CompleteTransactions instructs backend to complete sending of multiple transactions
-func (m *StatusBackend) CompleteTransactions(ids []common.QueuedTxID, password string) map[common.QueuedTxID]common.RawCompleteTransactionResult {
+func (m *StatusBackend) CompleteTransactions(ids []common.QueuedTxID, password string) map[common.QueuedTxID]common.TransactionResult {
 	return m.txQueueManager.CompleteTransactions(ids, password)
 }
 

diff --git a/geth/common/types.go b/geth/common/types.go
@@ -141,8 +141,8 @@ type AccountManager interface {
 	AddressToDecryptedAccount(address, password string) (accounts.Account, *keystore.Key, error)
 }
 
-// RawCompleteTransactionResult is a JSON returned from transaction complete function (used internally)
-type RawCompleteTransactionResult struct {
+// TransactionResult is a JSON returned from transaction complete function (used internally)
+type TransactionResult struct {
 	Hash  common.Hash
 	Error error
 }
@@ -158,11 +158,9 @@ type QueuedTxID string
 // QueuedTx holds enough information to complete the queued transaction.
 type QueuedTx struct {
 	ID      QueuedTxID
-	Hash    common.Hash
 	Context context.Context
 	Args    SendTxArgs
-	Done    chan struct{}
-	Err     error
+	Result  chan TransactionResult
 }
 
 // SendTxArgs represents the arguments to submit a new transaction into the transaction pool.

diff --git a/geth/common/utils.go b/geth/common/utils.go
@@ -157,9 +157,8 @@ func Fatalf(reason interface{}, args ...interface{}) {
 func CreateTransaction(ctx context.Context, args SendTxArgs) *QueuedTx {
 	return &QueuedTx{
 		ID:      QueuedTxID(uuid.New()),
-		Hash:    common.Hash{},
 		Context: ctx,
 		Args:    args,
-		Done:    make(chan struct{}),
+		Result:  make(chan TransactionResult, 1),
 	}
 }
diff --git a/geth/transactions/errors.go b/geth/transactions/errors.go
@@ -0,0 +1,10 @@
+package transactions
+
+import "errors"
+
+var (
+	//ErrQueuedTxTimedOut - error transaction sending timed out
+	ErrQueuedTxTimedOut = errors.New("transaction sending timed out")
+	//ErrQueuedTxDiscarded - error transaction discarded
+	ErrQueuedTxDiscarded = errors.New("transaction has been discarded")
+)
diff --git a/geth/transactions/fake/mock.go b/geth/transactions/fake/mock.go
diff --git a/geth/transactions/fake/txservice.go b/geth/transactions/fake/txservice.go
@@ -11,9 +11,9 @@ import (
 )
 
 // NewTestServer returns a mocked test server
-func NewTestServer(ctrl *gomock.Controller) (*rpc.Server, *MockFakePublicTransactionPoolAPI) {
+func NewTestServer(ctrl *gomock.Controller) (*rpc.Server, *MockPublicTransactionPoolAPI) {
 	srv := rpc.NewServer()
-	svc := NewMockFakePublicTransactionPoolAPI(ctrl)
+	svc := NewMockPublicTransactionPoolAPI(ctrl)
 	if err := srv.RegisterName("eth", svc); err != nil {
 		panic(err)
 	}

diff --git a/geth/transactions/notifications.go b/geth/transactions/notifications.go
@@ -1,12 +1,9 @@
 package transactions
 
 import (
-	"strconv"
-
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 	"github.com/status-im/status-go/geth/common"
 	"github.com/status-im/status-go/geth/signal"
-	"github.com/status-im/status-go/geth/transactions/queue"
 )
 
 const (
@@ -30,10 +27,10 @@ const (
 )
 
 var txReturnCodes = map[error]int{
-	nil:                        SendTransactionNoErrorCode,
-	keystore.ErrDecrypt:        SendTransactionPasswordErrorCode,
-	queue.ErrQueuedTxTimedOut:  SendTransactionTimeoutErrorCode,
-	queue.ErrQueuedTxDiscarded: SendTransactionDiscardedErrorCode,
+	nil:                  SendTransactionNoErrorCode,
+	keystore.ErrDecrypt:  SendTransactionPasswordErrorCode,
+	ErrQueuedTxTimedOut:  SendTransactionTimeoutErrorCode,
+	ErrQueuedTxDiscarded: SendTransactionDiscardedErrorCode,
 }
 
 // SendTransactionEvent is a signal sent on a send transaction request
@@ -61,17 +58,17 @@ type ReturnSendTransactionEvent struct {
 	Args         common.SendTxArgs `json:"args"`
 	MessageID    string            `json:"message_id"`
 	ErrorMessage string            `json:"error_message"`
-	ErrorCode    string            `json:"error_code"`
+	ErrorCode    int               `json:"error_code,string"`
 }
 
 // NotifyOnReturn returns handler that processes responses from internal tx manager
-func NotifyOnReturn(queuedTx *common.QueuedTx) {
-	// discard notifications with empty tx
-	if queuedTx == nil {
+func NotifyOnReturn(queuedTx *common.QueuedTx, err error) {
+	// we don't want to notify a user if tx was sent successfully
+	if err == nil {
 		return
 	}
-	// we don't want to notify a user if tx sent successfully
-	if queuedTx.Err == nil {
+	// discard notifications with empty tx
+	if queuedTx == nil {
 		return
 	}
 	signal.Send(signal.Envelope{
@@ -80,8 +77,8 @@ func NotifyOnReturn(queuedTx *common.QueuedTx) {
 			ID:           string(queuedTx.ID),
 			Args:         queuedTx.Args,
 			MessageID:    common.MessageIDFromContext(queuedTx.Context),
-			ErrorMessage: queuedTx.Err.Error(),
-			ErrorCode:    strconv.Itoa(sendTransactionErrorCode(queuedTx.Err)),
+			ErrorMessage: err.Error(),
+			ErrorCode:    sendTransactionErrorCode(err),
 		},
 	})
 }

diff --git a/geth/transactions/queue/queue.go b/geth/transactions/queue/queue.go
@@ -19,16 +19,12 @@ const (
 )
 
 var (
+	// ErrQueuedTxExist - transaction was already enqueued
+	ErrQueuedTxExist = errors.New("transaction already exist in queue")
 	//ErrQueuedTxIDNotFound - error transaction hash not found
 	ErrQueuedTxIDNotFound = errors.New("transaction hash not found")
-	//ErrQueuedTxTimedOut - error transaction sending timed out
-	ErrQueuedTxTimedOut = errors.New("transaction sending timed out")
-	//ErrQueuedTxDiscarded - error transaction discarded
-	ErrQueuedTxDiscarded = errors.New("transaction has been discarded")
-	//ErrQueuedTxInProgress - error transaction in progress
+	//ErrQueuedTxInProgress - error transaction is in progress
 	ErrQueuedTxInProgress = errors.New("transaction is in progress")
-	//ErrQueuedTxAlreadyProcessed - error transaction has already processed
-	ErrQueuedTxAlreadyProcessed = errors.New("transaction has been already processed")
 	//ErrInvalidCompleteTxSender - error transaction with invalid sender
 	ErrInvalidCompleteTxSender = errors.New("transaction can only be completed by the same account which created it")
 )
@@ -133,15 +129,17 @@ func (q *TxQueue) Reset() {
 // Enqueue enqueues incoming transaction
 func (q *TxQueue) Enqueue(tx *common.QueuedTx) error {
 	log.Info(fmt.Sprintf("enqueue transaction: %s", tx.ID))
-	if (tx.Hash != gethcommon.Hash{} || tx.Err != nil) {
-		return ErrQueuedTxAlreadyProcessed
+	q.mu.RLock()
+	if _, ok := q.transactions[tx.ID]; ok {
+		return ErrQueuedTxExist
 	}
+	q.mu.RUnlock()
 
-	log.Info("before enqueueTicker")
+	// we can't hold a lock in this part
+	log.Debug("notifying eviction loop")
 	q.enqueueTicker <- struct{}{} // notify eviction loop that we are trying to insert new item
-	log.Info("before evictableIDs")
-	q.evictableIDs <- tx.ID // this will block when we hit DefaultTxQueueCap
-	log.Info("after evictableIDs")
+	q.evictableIDs <- tx.ID       // this will block when we hit DefaultTxQueueCap
+	log.Debug("notified eviction loop")
 
 	q.mu.Lock()
 	q.transactions[tx.ID] = tx
@@ -204,17 +202,15 @@ func (q *TxQueue) Done(id common.QueuedTxID, hash gethcommon.Hash, err error) er
 
 func (q *TxQueue) done(tx *common.QueuedTx, hash gethcommon.Hash, err error) {
 	delete(q.inprogress, tx.ID)
-	tx.Err = err
 	// hash is updated only if err is nil, but transaction is not removed from a queue
 	if err == nil {
+		q.transactions[tx.ID].Result <- common.TransactionResult{Hash: hash, Error: err}
 		q.remove(tx.ID)
-		tx.Hash = hash
-		close(tx.Done)
 		return
 	}
 	if _, transient := transientErrs[err.Error()]; !transient {
+		q.transactions[tx.ID].Result <- common.TransactionResult{Error: err}
 		q.remove(tx.ID)
-		close(tx.Done)
 	}
 }