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).

Author: dshulyak

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
+		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

geth/common/types.go

@@ -155,11 +155,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 RawCompleteTransactionResult
 }
 
 // SendTxArgs represents the arguments to submit a new transaction into the transaction pool.
@@ -203,7 +201,7 @@ type TxQueueManager interface {
 	QueueTransaction(tx *QueuedTx) error
 
 	// WaitForTransactions blocks until transaction is completed, discarded or timed out.
-	WaitForTransaction(tx *QueuedTx) error
+	WaitForTransaction(tx *QueuedTx) RawCompleteTransactionResult
 
 	SendTransactionRPCHandler(ctx context.Context, args ...interface{}) (interface{}, error)
 

geth/common/types_mock.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{}, 1),
+		Result:  make(chan RawCompleteTransactionResult, 1),
 	}
 }

geth/common/utils.go

@@ -58,8 +58,8 @@ type ReturnSendTransactionEvent struct {
 }
 
 // NotifyOnReturn returns handler that processes responses from internal tx manager
-func NotifyOnReturn(queuedTx *common.QueuedTx) {
-	if queuedTx.Err == nil {
+func NotifyOnReturn(queuedTx *common.QueuedTx, err error) {
+	if err == nil {
 		return
 	}
 
@@ -75,8 +75,8 @@ func NotifyOnReturn(queuedTx *common.QueuedTx) {
 			ID:           string(queuedTx.ID),
 			Args:         queuedTx.Args,
 			MessageID:    common.MessageIDFromContext(queuedTx.Context),
-			ErrorMessage: queuedTx.Err.Error(),
-			ErrorCode:    sendTransactionErrorCode(queuedTx.Err),
+			ErrorMessage: err.Error(),
+			ErrorCode:    sendTransactionErrorCode(err),
 		},
 	})
 }

geth/transactions/notifications.go

@@ -133,9 +133,6 @@ 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
-	}
 
 	log.Info("before enqueueTicker")
 	q.enqueueTicker <- struct{}{} // notify eviction loop that we are trying to insert new item
@@ -205,18 +202,16 @@ 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.RawCompleteTransactionResult{Hash: hash, Error: err}
 		q.remove(tx.ID)
-		tx.Hash = hash
-		tx.Done <- struct{}{}
 		return
 	}
 	_, transient := transientErrs[err.Error()]
 	if !transient {
+		q.transactions[tx.ID].Result <- common.RawCompleteTransactionResult{Error: err}
 		q.remove(tx.ID)
-		tx.Done <- struct{}{}
 	}
 }
 

geth/transactions/queue/queue.go

@@ -52,13 +52,6 @@ func (s *QueueTestSuite) TestGetTransaction() {
 	}
 }
 
-func (s *QueueTestSuite) TestEnqueueProcessedTransaction() {
-	// enqueue will fail if transaction with hash will be enqueued
-	tx := common.CreateTransaction(context.Background(), common.SendTxArgs{})
-	tx.Hash = gethcommon.Hash{1}
-	s.Equal(ErrQueuedTxAlreadyProcessed, s.queue.Enqueue(tx))
-}
-
 func (s *QueueTestSuite) testDone(hash gethcommon.Hash, err error) *common.QueuedTx {
 	tx := common.CreateTransaction(context.Background(), common.SendTxArgs{})
 	s.NoError(s.queue.Enqueue(tx))
@@ -69,12 +62,12 @@ func (s *QueueTestSuite) testDone(hash gethcommon.Hash, err error) *common.Queue
 func (s *QueueTestSuite) TestDoneSuccess() {
 	hash := gethcommon.Hash{1}
 	tx := s.testDone(hash, nil)
-	s.NoError(tx.Err)
-	s.Equal(hash, tx.Hash)
-	s.False(s.queue.Has(tx.ID))
 	// event is sent only if transaction was removed from a queue
 	select {
-	case <-tx.Done:
+	case rst := <-tx.Result:
+		s.NoError(rst.Error)
+		s.Equal(hash, rst.Hash)
+		s.False(s.queue.Has(tx.ID))
 	default:
 		s.Fail("No event was sent to Done channel")
 	}
@@ -84,23 +77,22 @@ func (s *QueueTestSuite) TestDoneTransientError() {
 	hash := gethcommon.Hash{1}
 	err := keystore.ErrDecrypt
 	tx := s.testDone(hash, err)
-	s.Equal(keystore.ErrDecrypt, tx.Err)
-	s.NotEqual(hash, tx.Hash)
-	s.Equal(gethcommon.Hash{}, tx.Hash)
 	s.True(s.queue.Has(tx.ID))
+	_, inp := s.queue.inprogress[tx.ID]
+	s.False(inp)
 }
 
 func (s *QueueTestSuite) TestDoneError() {
 	hash := gethcommon.Hash{1}
 	err := errors.New("test")
 	tx := s.testDone(hash, err)
-	s.Equal(err, tx.Err)
-	s.NotEqual(hash, tx.Hash)
-	s.Equal(gethcommon.Hash{}, tx.Hash)
-	s.False(s.queue.Has(tx.ID))
 	// event is sent only if transaction was removed from a queue
 	select {
-	case <-tx.Done:
+	case rst := <-tx.Result:
+		s.Equal(err, rst.Error)
+		s.NotEqual(hash, rst.Hash)
+		s.Equal(gethcommon.Hash{}, rst.Hash)
+		s.False(s.queue.Has(tx.ID))
 	default:
 		s.Fail("No event was sent to Done channel")
 	}

geth/transactions/queue/queue_test.go

@@ -25,22 +25,24 @@ const (
 
 // Manager provides means to manage internal Status Backend (injected into LES)
 type Manager struct {
-	nodeManager    common.NodeManager
-	accountManager common.AccountManager
-	txQueue        *queue.TxQueue
-	ethTxClient    EthTransactor
-	addrLock       *AddrLocker
-	notify         bool
+	nodeManager           common.NodeManager
+	accountManager        common.AccountManager
+	txQueue               *queue.TxQueue
+	ethTxClient           EthTransactor
+	addrLock              *AddrLocker
+	notify                bool
+	sendCompletionTimeout time.Duration
 }
 
 // NewManager returns a new Manager.
 func NewManager(nodeManager common.NodeManager, accountManager common.AccountManager) *Manager {
 	return &Manager{
-		nodeManager:    nodeManager,
-		accountManager: accountManager,
-		txQueue:        queue.NewQueue(),
-		addrLock:       &AddrLocker{},
-		notify:         true,
+		nodeManager:           nodeManager,
+		accountManager:        accountManager,
+		txQueue:               queue.NewQueue(),
+		addrLock:              &AddrLocker{},
+		notify:                true,
+		sendCompletionTimeout: DefaultTxSendCompletionTimeout * time.Second,
 	}
 }
 
@@ -75,34 +77,38 @@ func (m *Manager) QueueTransaction(tx *common.QueuedTx) error {
 		to = tx.Args.To.Hex()
 	}
 	log.Info("queue a new transaction", "id", tx.ID, "from", tx.Args.From.Hex(), "to", to)
-	err := m.txQueue.Enqueue(tx)
+	if err := m.txQueue.Enqueue(tx); err != nil {
+		return err
+	}
 	if m.notify {
 		NotifyOnEnqueue(tx)
 	}
-	return err
+	return nil
 }
 
 func (m *Manager) txDone(tx *common.QueuedTx, hash gethcommon.Hash, err error) {
 	m.txQueue.Done(tx.ID, hash, err) //nolint: errcheck
 	if m.notify {
-		NotifyOnReturn(tx)
+		NotifyOnReturn(tx, err)
 	}
 }
 
 // WaitForTransaction adds a transaction to the queue and blocks
 // until it's completed, discarded or times out.
-func (m *Manager) WaitForTransaction(tx *common.QueuedTx) error {
+func (m *Manager) WaitForTransaction(tx *common.QueuedTx) common.RawCompleteTransactionResult {
 	log.Info("wait for transaction", "id", tx.ID)
 	// now wait up until transaction is:
 	// - completed (via CompleteQueuedTransaction),
 	// - discarded (via DiscardQueuedTransaction)
 	// - or times out
-	select {
-	case <-tx.Done:
-	case <-time.After(DefaultTxSendCompletionTimeout * time.Second):
-		m.txDone(tx, gethcommon.Hash{}, queue.ErrQueuedTxTimedOut)
+	for {
+		select {
+		case rst := <-tx.Result:
+			return rst
+		case <-time.After(m.sendCompletionTimeout):
+			m.txDone(tx, gethcommon.Hash{}, queue.ErrQueuedTxTimedOut)
+		}
 	}
-	return tx.Err
 }
 
 // CompleteTransaction instructs backend to complete sending of a given transaction.
@@ -242,7 +248,7 @@ func (m *Manager) DiscardTransaction(id common.QueuedTxID) error {
 	}
 	err = m.txQueue.Done(id, gethcommon.Hash{}, queue.ErrQueuedTxDiscarded)
 	if m.notify {
-		NotifyOnReturn(tx)
+		NotifyOnReturn(tx, queue.ErrQueuedTxDiscarded)
 	}
 	return err
 }
@@ -267,19 +273,17 @@ func (m *Manager) DiscardTransactions(ids []common.QueuedTxID) map[common.Queued
 // It accepts one param which is a slice with a map of transaction params.
 func (m *Manager) SendTransactionRPCHandler(ctx context.Context, args ...interface{}) (interface{}, error) {
 	log.Info("SendTransactionRPCHandler called")
-
 	// TODO(adam): it's a hack to parse arguments as common.RPCCall can do that.
 	// We should refactor parsing these params to a separate struct.
 	rpcCall := common.RPCCall{Params: args}
 	tx := common.CreateTransaction(ctx, rpcCall.ToSendTxArgs())
-
 	if err := m.QueueTransaction(tx); err != nil {
 		return nil, err
 	}
-
-	if err := m.WaitForTransaction(tx); err != nil {
-		return nil, err
+	rst := m.WaitForTransaction(tx)
+	if rst.Error != nil {
+		return nil, rst.Error
 	}
-
-	return tx.Hash.Hex(), nil
+	// handle empty hash
+	return rst.Hash.Hex(), nil
 }