package node import ( "errors" "fmt" "sync" "time" "github.com/ethereum/go-ethereum/accounts/keystore" "github.com/status-im/status-go/geth/common" "github.com/status-im/status-go/geth/log" ) const ( // DefaultTxQueueCap defines how many items can be queued. DefaultTxQueueCap = int(35) // DefaultTxSendQueueCap defines how many items can be passed to sendTransaction() w/o blocking. DefaultTxSendQueueCap = int(70) // DefaultTxSendCompletionTimeout defines how many seconds to wait before returning result in sentTransaction(). DefaultTxSendCompletionTimeout = 300 ) var ( ErrQueuedTxIDNotFound = errors.New("transaction hash not found") ErrQueuedTxTimedOut = errors.New("transaction sending timed out") ErrQueuedTxDiscarded = errors.New("transaction has been discarded") ErrInvalidCompleteTxSender = errors.New("transaction can only be completed by the same account which created it") ) // TxQueue is capped container that holds pending transactions type TxQueue struct { transactions map[common.QueuedTxID]*common.QueuedTx mu sync.RWMutex // to guard transactions map evictableIDs chan common.QueuedTxID enqueueTicker chan struct{} incomingPool chan *common.QueuedTx // when this channel is closed, all queue channels processing must cease (incoming queue, processing queued items etc) stopped chan struct{} stoppedGroup sync.WaitGroup // to make sure that all routines are stopped // when items are enqueued notify subscriber txEnqueueHandler common.EnqueuedTxHandler // when tx is returned (either successfully or with error) notify subscriber txReturnHandler common.EnqueuedTxReturnHandler } // NewTransactionQueue make new transaction queue func NewTransactionQueue() *TxQueue { log.Info("initializing transaction queue") return &TxQueue{ transactions: make(map[common.QueuedTxID]*common.QueuedTx), evictableIDs: make(chan common.QueuedTxID, DefaultTxQueueCap), // will be used to evict in FIFO enqueueTicker: make(chan struct{}), incomingPool: make(chan *common.QueuedTx, DefaultTxSendQueueCap), } } // Start starts enqueue and eviction loops func (q *TxQueue) Start() { log.Info("starting transaction queue") if q.stopped != nil { return } q.stopped = make(chan struct{}) q.stoppedGroup.Add(2) go q.evictionLoop() go q.enqueueLoop() } // Stop stops transaction enqueue and eviction loops func (q *TxQueue) Stop() { log.Info("stopping transaction queue") if q.stopped == nil { return } close(q.stopped) // stops all processing loops (enqueue, eviction etc) q.stoppedGroup.Wait() q.stopped = nil log.Info("finally stopped transaction queue") } // evictionLoop frees up queue to accommodate another transaction item func (q *TxQueue) evictionLoop() { defer HaltOnPanic() evict := func() { if len(q.transactions) >= DefaultTxQueueCap { // eviction is required to accommodate another/last item q.Remove(<-q.evictableIDs) } } for { select { case <-time.After(250 * time.Millisecond): // do not wait for manual ticks, check queue regularly evict() case <-q.enqueueTicker: // when manually requested evict() case <-q.stopped: log.Info("transaction queue's eviction loop stopped") q.stoppedGroup.Done() return } } } // enqueueLoop process incoming enqueue requests func (q *TxQueue) enqueueLoop() { defer HaltOnPanic() // enqueue incoming transactions for { select { case queuedTx := <-q.incomingPool: log.Info("transaction enqueued requested", "tx", queuedTx.ID) q.Enqueue(queuedTx) log.Info("transaction enqueued", "tx", queuedTx.ID) case <-q.stopped: log.Info("transaction queue's enqueue loop stopped") q.stoppedGroup.Done() return } } } // Reset is to be used in tests only, as it simply creates new transaction map, w/o any cleanup of the previous one func (q *TxQueue) Reset() { q.mu.Lock() defer q.mu.Unlock() q.transactions = make(map[common.QueuedTxID]*common.QueuedTx) q.evictableIDs = make(chan common.QueuedTxID, DefaultTxQueueCap) } // EnqueueAsync enqueues incoming transaction in async manner, returns as soon as possible func (q *TxQueue) EnqueueAsync(tx *common.QueuedTx) error { q.incomingPool <- tx return nil } // Enqueue enqueues incoming transaction func (q *TxQueue) Enqueue(tx *common.QueuedTx) error { log.Info(fmt.Sprintf("enqueue transaction: %s", tx.ID)) if q.txEnqueueHandler == nil { //discard, until handler is provided log.Info("there is no txEnqueueHandler") return nil } log.Info("before enqueueTicker") 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.mu.Lock() q.transactions[tx.ID] = tx q.mu.Unlock() // notify handler log.Info("calling txEnqueueHandler") q.txEnqueueHandler(tx) return nil } // Get returns transaction by transaction identifier func (q *TxQueue) Get(id common.QueuedTxID) (*common.QueuedTx, error) { q.mu.RLock() defer q.mu.RUnlock() if tx, ok := q.transactions[id]; ok { return tx, nil } return nil, ErrQueuedTxIDNotFound } // Remove removes transaction by transaction identifier func (q *TxQueue) Remove(id common.QueuedTxID) { q.mu.Lock() defer q.mu.Unlock() delete(q.transactions, id) } // Count returns number of currently queued transactions func (q *TxQueue) Count() int { q.mu.RLock() defer q.mu.RUnlock() return len(q.transactions) } // Has checks whether transaction with a given identifier exists in queue func (q *TxQueue) Has(id common.QueuedTxID) bool { q.mu.RLock() defer q.mu.RUnlock() _, ok := q.transactions[id] return ok } // SetEnqueueHandler sets callback handler, that is triggered on enqueue operation func (q *TxQueue) SetEnqueueHandler(fn common.EnqueuedTxHandler) { q.txEnqueueHandler = fn } // SetTxReturnHandler sets callback handler, that is triggered when transaction is finished executing func (q *TxQueue) SetTxReturnHandler(fn common.EnqueuedTxReturnHandler) { q.txReturnHandler = fn } // NotifyOnQueuedTxReturn is invoked when transaction is ready to return // Transaction can be in error state, or executed successfully at this point. func (q *TxQueue) NotifyOnQueuedTxReturn(queuedTx *common.QueuedTx, err error) { if q == nil { return } // discard, if transaction is not found if queuedTx == nil { return } // on success, remove item from the queue and stop propagating if err == nil { q.Remove(queuedTx.ID) return } // error occurred, send upward notification if q.txReturnHandler == nil { // discard, until handler is provided return } // remove from queue on any error (except for transient ones) and propagate transientErrs := map[error]bool{ keystore.ErrDecrypt: true, // wrong password ErrInvalidCompleteTxSender: true, // completing tx create from another account } if !transientErrs[err] { // remove only on unrecoverable errors q.Remove(queuedTx.ID) } // notify handler q.txReturnHandler(queuedTx, err) }