diff --git a/actpool/actpool_test.go b/actpool/actpool_test.go index 3ec07e4086..dbef4e06e4 100644 --- a/actpool/actpool_test.go +++ b/actpool/actpool_test.go @@ -393,7 +393,7 @@ func TestActPool_removeConfirmedActs(t *testing.T) { require.NoError(acct.AddBalance(big.NewInt(100000000000000000))) return 0, nil - }).Times(8) + }).Times(5) sf.EXPECT().Height().Return(uint64(1), nil).AnyTimes() ctx := genesis.WithGenesisContext(context.Background(), genesis.Default) require.NoError(ap.Add(ctx, tsf1)) @@ -1010,7 +1010,7 @@ func TestActPool_GetSize(t *testing.T) { require.NoError(acct.AddBalance(big.NewInt(100000000000000000))) return 0, nil - }).Times(8) + }).Times(5) sf.EXPECT().Height().Return(uint64(1), nil).AnyTimes() ctx := genesis.WithGenesisContext(context.Background(), genesis.Default) require.NoError(ap.Add(ctx, tsf1)) diff --git a/actpool/actqueue.go b/actpool/actqueue.go index 9f36d578a3..c97aac6e11 100644 --- a/actpool/actqueue.go +++ b/actpool/actqueue.go @@ -61,12 +61,14 @@ func (h *noncePriorityQueue) Pop() interface{} { // ActQueue is the interface of actQueue type ActQueue interface { Put(action.SealedEnvelope) error - FilterNonce(uint64) []action.SealedEnvelope - UpdateQueue(uint64) []action.SealedEnvelope + CleanConfirmedAct() []action.SealedEnvelope + UpdateQueue() []action.SealedEnvelope SetPendingNonce(uint64) PendingNonce() uint64 - SetPendingBalance(*big.Int) + AccountNonce() uint64 + SetAccountBalance(*big.Int) PendingBalance() *big.Int + AccountBalance() *big.Int Len() int Empty() bool PendingActs(context.Context) []action.SealedEnvelope @@ -84,22 +86,28 @@ type actQueue struct { index noncePriorityQueue // Current pending nonce tracking previous actions that can be committed to the next block for the account pendingNonce uint64 - // Current pending balance for the account - pendingBalance *big.Int + // Pending balance map + pendingBalance map[uint64]*big.Int + // Current account nonce + accountNonce uint64 + // Current account balance + accountBalance *big.Int clock clock.Clock ttl time.Duration mu sync.RWMutex } // NewActQueue create a new action queue -func NewActQueue(ap *actPool, address string, ops ...ActQueueOption) ActQueue { +func NewActQueue(ap *actPool, address string, pendingNonce uint64, balance *big.Int, ops ...ActQueueOption) ActQueue { aq := &actQueue{ ap: ap, address: address, items: make(map[uint64]action.SealedEnvelope), index: noncePriorityQueue{}, - pendingNonce: uint64(1), // Taking coinbase Action into account, pendingNonce should start with 1 - pendingBalance: big.NewInt(0), + pendingBalance: make(map[uint64]*big.Int), + pendingNonce: pendingNonce, + accountNonce: pendingNonce, + accountBalance: new(big.Int).Set(balance), clock: clock.New(), ttl: 0, } @@ -114,36 +122,79 @@ func (q *actQueue) Put(act action.SealedEnvelope) error { q.mu.Lock() defer q.mu.Unlock() nonce := act.Nonce() + + if cost, _ := act.Cost(); q.getPendingBalanceAtNonce(nonce).Cmp(cost) < 0 { + return action.ErrInsufficientFunds + } + if actInPool, exist := q.items[nonce]; exist { - // act of higher gas price cut in line - if act.GasPrice().Cmp(actInPool.GasPrice()) != 1 { + // act of higher gas price can cut in line + if nonce < q.pendingNonce && act.GasPrice().Cmp(actInPool.GasPrice()) != 1 { return action.ErrReplaceUnderpriced } // update action in q.items and q.index q.items[nonce] = act - for i, x := range q.index { - if x.nonce == nonce { + for i := range q.index { + if q.index[i].nonce == nonce { q.index[i].deadline = q.clock.Now().Add(q.ttl) break } } + q.updateFromNonce(nonce) return nil } heap.Push(&q.index, &nonceWithTTL{nonce: nonce, deadline: q.clock.Now().Add(q.ttl)}) q.items[nonce] = act + if nonce == q.pendingNonce { + q.updateFromNonce(q.pendingNonce) + } return nil } -// FilterNonce removes all actions from the map with a nonce lower than the given threshold -func (q *actQueue) FilterNonce(threshold uint64) []action.SealedEnvelope { +func (q *actQueue) getPendingBalanceAtNonce(nonce uint64) *big.Int { + if nonce > q.pendingNonce { + return q.getPendingBalanceAtNonce(q.pendingNonce) + } + if _, exist := q.pendingBalance[nonce]; !exist { + return new(big.Int).Set(q.accountBalance) + } + return new(big.Int).Set(q.pendingBalance[nonce]) +} + +func (q *actQueue) updateFromNonce(start uint64) { + if start > q.pendingNonce { + return + } + + for balance := q.getPendingBalanceAtNonce(start); ; start++ { + act, exist := q.items[start] + if !exist { + break + } + + cost, _ := act.Cost() + if balance.Cmp(cost) < 0 { + break + } + + balance = new(big.Int).Sub(balance, cost) + q.pendingBalance[start+1] = new(big.Int).Set(balance) + } + + q.pendingNonce = start +} + +// CleanConfirmedAct removes all actions from the map with a nonce lower than account's nonce +func (q *actQueue) CleanConfirmedAct() []action.SealedEnvelope { q.mu.Lock() defer q.mu.Unlock() var removed []action.SealedEnvelope // Pop off priority queue and delete corresponding entries from map until the threshold is reached - for q.index.Len() > 0 && (q.index)[0].nonce < threshold { + for q.index.Len() > 0 && (q.index)[0].nonce < q.accountNonce { nonce := heap.Pop(&q.index).(*nonceWithTTL).nonce removed = append(removed, q.items[nonce]) delete(q.items, nonce) + delete(q.pendingBalance, nonce) } return removed } @@ -159,8 +210,13 @@ func (q *actQueue) cleanTimeout() []action.SealedEnvelope { ) for i := 0; i < size; { if timeNow.After(q.index[i].deadline) { - removedFromQueue = append(removedFromQueue, q.items[q.index[i].nonce]) - delete(q.items, q.index[i].nonce) + nonce := q.index[i].nonce + if nonce < q.pendingNonce { + q.pendingNonce = nonce + } + removedFromQueue = append(removedFromQueue, q.items[nonce]) + delete(q.items, nonce) + delete(q.pendingBalance, nonce) q.index[i] = q.index[size-1] size-- continue @@ -174,52 +230,13 @@ func (q *actQueue) cleanTimeout() []action.SealedEnvelope { } // UpdateQueue updates the pending nonce and balance of the queue -func (q *actQueue) UpdateQueue(nonce uint64) []action.SealedEnvelope { +func (q *actQueue) UpdateQueue() []action.SealedEnvelope { q.mu.Lock() defer q.mu.Unlock() // First remove all timed out actions removedFromQueue := q.cleanTimeout() - // Now, starting from the current pending nonce, incrementally find the next pending nonce - // while updating pending balance if actions are payable - for ; ; nonce++ { - _, exist := q.items[nonce] - if !exist { - break - } - if !q.enoughBalance(q.items[nonce], true) { - break - } - } - q.pendingNonce = nonce - - // Find the index of new pending nonce within the queue - sort.Sort(q.index) - i := 0 - for ; i < q.index.Len(); i++ { - if q.index[i].nonce >= nonce { - break - } - } - // Case I: An unpayable action has been found while updating pending nonce/balance - // Remove all the subsequent actions in the queue starting from the index of new pending nonce - if _, exist := q.items[nonce]; exist { - removedFromQueue = append(removedFromQueue, q.removeActs(i)...) - return removedFromQueue - } - - // Case II: All actions are payable while updating pending nonce/balance - // Check all the subsequent actions in the queue starting from the index of new pending nonce - // Find the nonce index of the first unpayable action - // Remove all the subsequent actions in the queue starting from that index - for ; i < q.index.Len(); i++ { - nonce = q.index[i].nonce - act := q.items[nonce] - if !q.enoughBalance(act, false) { - break - } - } - removedFromQueue = append(removedFromQueue, q.removeActs(i)...) + q.updateFromNonce(q.pendingNonce) return removedFromQueue } @@ -228,6 +245,7 @@ func (q *actQueue) SetPendingNonce(nonce uint64) { q.mu.Lock() defer q.mu.Unlock() q.pendingNonce = nonce + q.accountNonce = nonce } // PendingNonce returns the current pending nonce of the queue @@ -237,18 +255,33 @@ func (q *actQueue) PendingNonce() uint64 { return q.pendingNonce } -// SetPendingBalance sets pending balance for the queue -func (q *actQueue) SetPendingBalance(balance *big.Int) { +// AccountNonce returns the current account nonce +func (q *actQueue) AccountNonce() uint64 { + q.mu.RLock() + defer q.mu.RUnlock() + return q.accountNonce +} + +// SetAccountBalance sets account balance for the queue +func (q *actQueue) SetAccountBalance(balance *big.Int) { q.mu.Lock() defer q.mu.Unlock() - q.pendingBalance = balance + q.accountBalance.Set(balance) + q.pendingBalance = make(map[uint64]*big.Int) } // PendingBalance returns the current pending balance of the queue func (q *actQueue) PendingBalance() *big.Int { q.mu.RLock() defer q.mu.RUnlock() - return q.pendingBalance + return q.getPendingBalanceAtNonce(q.pendingNonce) +} + +// AccountBalance returns the current account balance +func (q *actQueue) AccountBalance() *big.Int { + q.mu.RLock() + defer q.mu.RUnlock() + return new(big.Int).Set(q.accountBalance) } // Len returns the length of the action map @@ -262,7 +295,17 @@ func (q *actQueue) Len() int { func (q *actQueue) Empty() bool { q.mu.RLock() defer q.mu.RUnlock() - return q.Len() == 0 + return len(q.items) == 0 +} + +// Reset makes the queue into a dummy queue +func (q *actQueue) Reset() { + q.mu.Lock() + defer q.mu.Unlock() + q.items = make(map[uint64]action.SealedEnvelope) + q.index = noncePriorityQueue{} + q.pendingNonce = 0 + q.accountBalance = big.NewInt(0) } // PendingActs creates a consecutive nonce-sorted slice of actions @@ -270,7 +313,6 @@ func (q *actQueue) PendingActs(ctx context.Context) []action.SealedEnvelope { if q.Len() == 0 { return nil } - acts := make([]action.SealedEnvelope, 0, len(q.items)) addr, err := address.FromString(q.address) if err != nil { log.L().Error("Error when getting the address", zap.String("address", q.address), zap.Error(err)) @@ -281,14 +323,27 @@ func (q *actQueue) PendingActs(ctx context.Context) []action.SealedEnvelope { log.L().Error("Error when getting the nonce", zap.String("address", q.address), zap.Error(err)) return nil } - nonce := confirmedState.PendingNonce() + + var ( + nonce = confirmedState.PendingNonce() + balance = new(big.Int).Set(confirmedState.Balance) + acts = make([]action.SealedEnvelope, 0, len(q.items)) + ) q.mu.RLock() defer q.mu.RUnlock() for ; ; nonce++ { - if _, exist := q.items[nonce]; !exist { + act, exist := q.items[nonce] + if !exist { break } - acts = append(acts, q.items[nonce]) + + cost, _ := act.Cost() + if balance.Cmp(cost) < 0 { + break + } + + balance = new(big.Int).Sub(balance, cost) + acts = append(acts, act) } return acts } @@ -307,39 +362,3 @@ func (q *actQueue) AllActs() []action.SealedEnvelope { } return acts } - -// removeActs removes all the actions starting at idx from queue -func (q *actQueue) removeActs(idx int) []action.SealedEnvelope { - removedFromQueue := make([]action.SealedEnvelope, 0) - for i := idx; i < q.index.Len(); i++ { - removedFromQueue = append(removedFromQueue, q.items[q.index[i].nonce]) - delete(q.items, q.index[i].nonce) - } - q.index = q.index[:idx] - heap.Init(&q.index) - return removedFromQueue -} - -// enoughBalance helps check whether queue's pending balance is sufficient for the given action -func (q *actQueue) enoughBalance(act action.SealedEnvelope, updateBalance bool) bool { - cost, _ := act.Cost() - if q.pendingBalance.Cmp(cost) < 0 { - return false - } - - if updateBalance { - q.pendingBalance.Sub(q.pendingBalance, cost) - } - - return true -} - -// Reset resets the queue -func (q *actQueue) Reset() { - q.mu.Lock() - defer q.mu.Unlock() - q.items = make(map[uint64]action.SealedEnvelope) - q.index = noncePriorityQueue{} - q.pendingNonce = 1 - q.pendingBalance = big.NewInt(0) -} diff --git a/actpool/actqueue_test.go b/actpool/actqueue_test.go index 5074931edf..5934743f05 100644 --- a/actpool/actqueue_test.go +++ b/actpool/actqueue_test.go @@ -26,6 +26,10 @@ import ( "github.com/iotexproject/iotex-core/test/mock/mock_chainmanager" ) +const ( + maxBalance = 1e7 +) + func TestNoncePriorityQueue(t *testing.T) { require := require.New(t) pq := noncePriorityQueue{} @@ -58,7 +62,8 @@ func TestNoncePriorityQueue(t *testing.T) { func TestActQueuePut(t *testing.T) { require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) + q := NewActQueue(nil, "", 1, big.NewInt(0)).(*actQueue) + q.SetAccountBalance(big.NewInt(maxBalance)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(100), nil, uint64(0), big.NewInt(1)) require.NoError(err) require.NoError(q.Put(tsf1)) @@ -83,7 +88,8 @@ func TestActQueuePut(t *testing.T) { func TestActQueueFilterNonce(t *testing.T) { require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) + q := NewActQueue(nil, "", 1, big.NewInt(0)).(*actQueue) + q.SetAccountBalance(big.NewInt(maxBalance)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 1, big.NewInt(1), nil, uint64(0), big.NewInt(0)) require.NoError(err) tsf2, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(1), nil, uint64(0), big.NewInt(0)) @@ -93,7 +99,8 @@ func TestActQueueFilterNonce(t *testing.T) { require.NoError(q.Put(tsf1)) require.NoError(q.Put(tsf2)) require.NoError(q.Put(tsf3)) - q.FilterNonce(uint64(3)) + q.SetPendingNonce(3) + q.CleanConfirmedAct() require.Equal(1, len(q.items)) require.Equal(uint64(3), q.index[0].nonce) require.Equal(tsf3, q.items[q.index[0].nonce]) @@ -101,26 +108,26 @@ func TestActQueueFilterNonce(t *testing.T) { func TestActQueueUpdateNonce(t *testing.T) { require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) + q := NewActQueue(nil, "", 1, big.NewInt(0)).(*actQueue) + q.SetAccountBalance(big.NewInt(1010)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 1, big.NewInt(1), nil, uint64(0), big.NewInt(0)) require.NoError(err) tsf2, err := action.SignedTransfer(_addr2, _priKey1, 3, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) require.NoError(err) - tsf3, err := action.SignedTransfer(_addr2, _priKey1, 4, big.NewInt(10000), nil, uint64(0), big.NewInt(0)) + tsf3, err := action.SignedTransfer(_addr2, _priKey1, 4, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) require.NoError(err) - tsf4, err := action.SignedTransfer(_addr2, _priKey1, 6, big.NewInt(100000), nil, uint64(0), big.NewInt(0)) + tsf4, err := action.SignedTransfer(_addr2, _priKey1, 6, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) require.NoError(err) - tsf5, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(100000), nil, uint64(0), big.NewInt(0)) + tsf5, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) require.NoError(err) require.NoError(q.Put(tsf1)) require.NoError(q.Put(tsf2)) require.NoError(q.Put(tsf3)) require.NoError(q.Put(tsf4)) - q.pendingBalance = big.NewInt(1000) require.NoError(q.Put(tsf5)) - removed := q.UpdateQueue(uint64(2)) - require.Equal(uint64(2), q.pendingNonce) - require.Equal([]action.SealedEnvelope{tsf5, tsf2, tsf3, tsf4}, removed) + removed := q.UpdateQueue() + require.Equal(uint64(3), q.pendingNonce) + require.Equal(0, len(removed)) } func TestActQueuePendingActs(t *testing.T) { @@ -130,12 +137,14 @@ func TestActQueuePendingActs(t *testing.T) { sf := mock_chainmanager.NewMockStateReader(ctrl) sf.EXPECT().State(gomock.Any(), gomock.Any()).Do(func(accountState *state.Account, _ protocol.StateOption) { require.NoError(accountState.SetPendingNonce(accountState.PendingNonce() + 1)) + accountState.Balance = big.NewInt(maxBalance) }).Return(uint64(0), nil).Times(1) sf.EXPECT().Height().Return(uint64(1), nil).AnyTimes() ctx := genesis.WithGenesisContext(context.Background(), genesis.Default) ap, err := NewActPool(genesis.Default, sf, DefaultConfig, EnableExperimentalActions()) require.NoError(err) - q := NewActQueue(ap.(*actPool), identityset.Address(0).String()).(*actQueue) + q := NewActQueue(ap.(*actPool), identityset.Address(0).String(), 1, big.NewInt(0)).(*actQueue) + q.SetAccountBalance(big.NewInt(maxBalance)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(100), nil, uint64(0), big.NewInt(0)) require.NoError(err) tsf2, err := action.SignedTransfer(_addr2, _priKey1, 3, big.NewInt(100), nil, uint64(0), big.NewInt(0)) @@ -158,7 +167,8 @@ func TestActQueuePendingActs(t *testing.T) { func TestActQueueAllActs(t *testing.T) { require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) + q := NewActQueue(nil, "", 1, big.NewInt(0)).(*actQueue) + q.SetAccountBalance(big.NewInt(maxBalance)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 1, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) require.NoError(err) tsf3, err := action.SignedTransfer(_addr2, _priKey1, 3, big.NewInt(1000), nil, uint64(0), big.NewInt(0)) @@ -169,41 +179,10 @@ func TestActQueueAllActs(t *testing.T) { require.Equal([]action.SealedEnvelope{tsf1, tsf3}, actions) } -func TestActQueueRemoveActs(t *testing.T) { - require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) - tsf1, err := action.SignedTransfer(_addr2, _priKey1, 1, big.NewInt(100), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - tsf2, err := action.SignedTransfer(_addr2, _priKey1, 2, big.NewInt(100), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - tsf3, err := action.SignedTransfer(_addr2, _priKey1, 3, big.NewInt(100), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - require.NoError(q.Put(tsf1)) - require.NoError(q.Put(tsf2)) - require.NoError(q.Put(tsf3)) - removed := q.removeActs(0) - require.Equal(0, len(q.index)) - require.Equal(0, len(q.items)) - require.Equal([]action.SealedEnvelope{tsf1, tsf2, tsf3}, removed) - - tsf4, err := action.SignedTransfer(_addr2, _priKey1, 4, big.NewInt(10000), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - tsf5, err := action.SignedTransfer(_addr2, _priKey1, 5, big.NewInt(100000), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - tsf6, err := action.SignedTransfer(_addr2, _priKey1, 6, big.NewInt(100000), nil, uint64(0), big.NewInt(0)) - require.NoError(err) - require.NoError(q.Put(tsf4)) - require.NoError(q.Put(tsf5)) - require.NoError(q.Put(tsf6)) - removed = q.removeActs(1) - require.Equal(1, len(q.index)) - require.Equal(1, len(q.items)) - require.Equal([]action.SealedEnvelope{tsf5, tsf6}, removed) -} - func TestActQueueTimeOutAction(t *testing.T) { c := clock.NewMock() - q := NewActQueue(nil, "", WithClock(c), WithTimeOut(3*time.Minute)) + q := NewActQueue(nil, "", 1, big.NewInt(0), WithClock(c), WithTimeOut(3*time.Minute)) + q.SetAccountBalance(big.NewInt(maxBalance)) tsf1, err := action.SignedTransfer(_addr2, _priKey1, 1, big.NewInt(100), nil, uint64(0), big.NewInt(0)) require.NoError(t, err) tsf2, err := action.SignedTransfer(_addr2, _priKey1, 3, big.NewInt(100), nil, uint64(0), big.NewInt(0)) @@ -222,7 +201,7 @@ func TestActQueueTimeOutAction(t *testing.T) { func TestActQueueCleanTimeout(t *testing.T) { require := require.New(t) - q := NewActQueue(nil, "").(*actQueue) + q := NewActQueue(nil, "", 1, big.NewInt(0)).(*actQueue) q.ttl = 1 invalidTime := time.Now() validTime := time.Now().Add(10 * time.Minute) diff --git a/actpool/queueworker.go b/actpool/queueworker.go index a517f73c65..9b872fe37f 100644 --- a/actpool/queueworker.go +++ b/actpool/queueworker.go @@ -115,19 +115,18 @@ func (worker *queueWorker) Handle(job workerJob) error { } func (worker *queueWorker) getConfirmedState(ctx context.Context, sender address.Address) (uint64, *big.Int, error) { - // TODO: account Balance(confirmedBalance) will be returned in PR#3377 - confirmedState, err := accountutil.AccountState(ctx, worker.ap.sf, sender) - if err != nil { - return 0, nil, err - } worker.mu.RLock() queue := worker.accountActs[sender.String()] worker.mu.RUnlock() // account state isn't cached in the actpool if queue == nil { + confirmedState, err := accountutil.AccountState(ctx, worker.ap.sf, sender) + if err != nil { + return 0, nil, err + } return confirmedState.PendingNonce(), confirmedState.Balance, nil } - return confirmedState.PendingNonce(), queue.PendingBalance(), nil + return queue.AccountNonce(), queue.AccountBalance(), nil } func (worker *queueWorker) checkSelpWithState(act *action.SealedEnvelope, pendingNonce uint64, balance *big.Int) error { @@ -168,9 +167,12 @@ func (worker *queueWorker) putAction(sender string, act action.SealedEnvelope, p worker.mu.RUnlock() if queue == nil { - queue = NewActQueue(worker.ap, sender, WithTimeOut(worker.ap.cfg.ActionExpiry)) - queue.SetPendingNonce(pendingNonce) - queue.SetPendingBalance(confirmedBalance) + queue = NewActQueue(worker.ap, + sender, + pendingNonce, + confirmedBalance, + WithTimeOut(worker.ap.cfg.ActionExpiry), + ) worker.mu.Lock() worker.accountActs[sender] = queue worker.mu.Unlock() @@ -185,8 +187,6 @@ func (worker *queueWorker) putAction(sender string, act action.SealedEnvelope, p return err } - queue.UpdateQueue(queue.PendingNonce()) // TODO: to be removed - return nil } @@ -234,10 +234,10 @@ func (worker *queueWorker) Reset(ctx context.Context) { continue } queue.SetPendingNonce(confirmedState.PendingNonce()) - queue.SetPendingBalance(confirmedState.Balance) + queue.SetAccountBalance(confirmedState.Balance) // Remove all actions that are committed to new block - acts := queue.FilterNonce(queue.PendingNonce()) - acts2 := queue.UpdateQueue(queue.PendingNonce()) + acts := queue.CleanConfirmedAct() + acts2 := queue.UpdateQueue() worker.ap.removeInvalidActs(append(acts, acts2...)) // Delete the queue entry if it becomes empty if queue.Empty() { @@ -257,7 +257,7 @@ func (worker *queueWorker) PendingActions(ctx context.Context) []*pendingActions continue } // Remove the actions that are already timeout - acts := queue.UpdateQueue(queue.PendingNonce()) + acts := queue.UpdateQueue() worker.ap.removeInvalidActs(acts) actionArr = append(actionArr, &pendingActions{ sender: from,