Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

bugfix: adding deletion of invalid txs from pool db in the finalizer. #1696

Merged
merged 1 commit into from
Feb 27, 2023
Merged

bugfix: adding deletion of invalid txs from pool db in the finalizer. #1696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 6 additions & 7 deletions sequencer/addrqueue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,16 +68,16 @@ func (a *addrQueue) deleteTx(txHash common.Hash) (deletedReadyTx *TxTracker) {
for _, txTracker := range a.notReadyTxs {
if txTracker.HashStr == txHashStr {
delete(a.notReadyTxs, txTracker.Nonce)
break
}
}
return nil
}
}

// updateCurrentNonceBalance updates the nonce and balance of the addrQueue and updates the ready and notReady txs
func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (newReadyTx, prevReadyTx *TxTracker) {
func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (newReadyTx, prevReadyTx *TxTracker, toDelete []*TxTracker) {
var oldReadyTx *TxTracker = nil
txsToDelete := make([]*TxTracker, 0)

if balance != nil {
a.currentBalance = balance
Expand All @@ -88,14 +88,13 @@ func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (
a.currentNonce = *nonce

//TODO: we need to update in the DB the deleted txs?
txToDelete := []uint64{}
for _, txTracker := range a.notReadyTxs {
if txTracker.Nonce < a.currentNonce {
txToDelete = append(txToDelete, txTracker.Nonce)
txsToDelete = append(txsToDelete, txTracker)
}
}
for _, delTxNonce := range txToDelete {
delete(a.notReadyTxs, delTxNonce)
for _, txTracker := range txsToDelete {
delete(a.notReadyTxs, txTracker.Nonce)
}
}
}
Expand Down Expand Up @@ -126,7 +125,7 @@ func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (
a.notReadyTxs[oldReadyTx.Nonce] = oldReadyTx
}

return a.readyTx, oldReadyTx
return a.readyTx, oldReadyTx, txsToDelete
}

// UpdateTxZKCounters updates the ZKCounters for the given tx (txHash)
Expand Down
24 changes: 18 additions & 6 deletions sequencer/finalizer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,15 +381,15 @@ func (f *finalizer) handleSuccessfulTxProcessResp(ctx context.Context, tx *TxTra

previousL2BlockStateRoot := f.batch.stateRoot
// Store the processed transaction, add it to the batch and update status in the pool atomically
f.storeProcessedTx(previousL2BlockStateRoot, tx, result)
f.storeProcessedTx(ctx, previousL2BlockStateRoot, tx, result)
f.processRequest.OldStateRoot = result.NewStateRoot
f.batch.stateRoot = result.NewStateRoot
f.batch.localExitRoot = result.NewLocalExitRoot

return nil
}

func (f *finalizer) storeProcessedTx(previousL2BlockStateRoot common.Hash, tx *TxTracker, result *state.ProcessBatchResponse) {
func (f *finalizer) storeProcessedTx(ctx context.Context, previousL2BlockStateRoot common.Hash, tx *TxTracker, result *state.ProcessBatchResponse) {
if tx == nil || len(result.Responses) == 0 {
return
}
Expand All @@ -406,7 +406,13 @@ func (f *finalizer) storeProcessedTx(previousL2BlockStateRoot common.Hash, tx *T
}

start := time.Now()
f.worker.UpdateAfterSingleSuccessfulTxExecution(tx.From, result.ReadWriteAddresses)
txsToDelete := f.worker.UpdateAfterSingleSuccessfulTxExecution(tx.From, result.ReadWriteAddresses)
for _, txToDelete := range txsToDelete {
err := f.dbManager.UpdateTxStatus(ctx, txToDelete.Hash, pool.TxStatusFailed)
if err != nil {
log.Errorf("failed to update status to failed in the pool for tx: %s, err: %s", txToDelete.Hash.String(), err)
}
}
metrics.WorkerProcessingTime(time.Since(start))
f.batch.countOfTxs += 1
}
Expand All @@ -423,9 +429,9 @@ func (f *finalizer) handleTransactionError(ctx context.Context, result *state.Pr
f.worker.DeleteTx(tx.Hash, tx.From)
metrics.WorkerProcessingTime(time.Since(start))
go func() {
err := f.dbManager.UpdateTxStatus(ctx, tx.Hash, pool.TxStatusInvalid)
err := f.dbManager.UpdateTxStatus(ctx, tx.Hash, pool.TxStatusFailed)
if err != nil {
log.Errorf("failed to update tx status, err: %s", err)
log.Errorf("failed to update status to failed in the pool for tx: %s, err: %s", tx.Hash.String(), err)
}
}()
} else if executor.IsIntrinsicError(errorCode) {
Expand All @@ -439,7 +445,13 @@ func (f *finalizer) handleTransactionError(ctx context.Context, result *state.Pr
balance = addressInfo.Balance
}
start := time.Now()
f.worker.MoveTxToNotReady(tx.Hash, tx.From, nonce, balance)
txsToDelete := f.worker.MoveTxToNotReady(tx.Hash, tx.From, nonce, balance)
for _, txToDelete := range txsToDelete {
err := f.dbManager.UpdateTxStatus(ctx, txToDelete.Hash, pool.TxStatusFailed)
if err != nil {
log.Errorf("failed to update status to failed in the pool for tx: %s, err: %s", txToDelete.Hash.String(), err)
}
}
metrics.WorkerProcessingTime(time.Since(start))
}
}
Expand Down
3 changes: 2 additions & 1 deletion sequencer/finalizer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,7 +247,8 @@ func TestFinalizer_handleTransactionError(t *testing.T) {
// arrange
if tc.expectedDeleteCall {
workerMock.On("DeleteTx", oldHash, sender).Return().Once()
dbManagerMock.On("UpdateTxStatus", ctx, oldHash, pool.TxStatusInvalid).Return(nil).Once()
dbManagerMock.On("UpdateTxStatus", ctx, oldHash, pool.TxStatusFailed).Return(nil).Once()
dbManagerMock.On("DeleteTransactionFromPool", ctx, tx.Hash).Return(nil).Once()
}
if tc.expectedMoveCall {
workerMock.On("MoveTxToNotReady", oldHash, sender, &nonce, big.NewInt(0)).Return().Once()
Expand Down
4 changes: 2 additions & 2 deletions sequencer/interfaces.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,10 +75,10 @@ type stateInterface interface {

type workerInterface interface {
GetBestFittingTx(resources batchResources) *TxTracker
UpdateAfterSingleSuccessfulTxExecution(from common.Address, touchedAddresses map[common.Address]*state.InfoReadWrite)
UpdateAfterSingleSuccessfulTxExecution(from common.Address, touchedAddresses map[common.Address]*state.InfoReadWrite) []*TxTracker
UpdateTx(txHash common.Hash, from common.Address, ZKCounters state.ZKCounters)
AddTx(ctx context.Context, txTracker *TxTracker)
MoveTxToNotReady(txHash common.Hash, from common.Address, actualNonce *uint64, actualBalance *big.Int)
MoveTxToNotReady(txHash common.Hash, from common.Address, actualNonce *uint64, actualBalance *big.Int) []*TxTracker
DeleteTx(txHash common.Hash, from common.Address)
HandleL2Reorg(txHashes []common.Hash)
NewTxTracker(tx types.Transaction, isClaim bool, counters state.ZKCounters) (*TxTracker, error)
Expand Down
6 changes: 4 additions & 2 deletions sequencer/mock_worker.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

23 changes: 13 additions & 10 deletions sequencer/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,12 +90,12 @@ func (w *Worker) AddTx(ctx context.Context, tx *TxTracker) {
}
}

func (w *Worker) applyAddressUpdate(from common.Address, fromNonce *uint64, fromBalance *big.Int) (*TxTracker, *TxTracker) {
func (w *Worker) applyAddressUpdate(from common.Address, fromNonce *uint64, fromBalance *big.Int) (*TxTracker, *TxTracker, []*TxTracker) {
addrQueue, found := w.pool[from.String()]

// TODO: What happens if addr no found. Could it be possible if addrQueue has not been yet created for this from addr (touchedAddresses)
if found {
newReadyTx, prevReadyTx := addrQueue.updateCurrentNonceBalance(fromNonce, fromBalance)
newReadyTx, prevReadyTx, txsToDelete := addrQueue.updateCurrentNonceBalance(fromNonce, fromBalance)

// Update the EfficiencyList (if needed)
if prevReadyTx != nil {
Expand All @@ -105,37 +105,39 @@ func (w *Worker) applyAddressUpdate(from common.Address, fromNonce *uint64, from
w.efficiencyList.add(newReadyTx)
}

return newReadyTx, prevReadyTx
return newReadyTx, prevReadyTx, txsToDelete
}

return nil, nil
return nil, nil, nil
}

// UpdateAfterSingleSuccessfulTxExecution updates the touched addresses after execute on Executor a successfully tx
func (w *Worker) UpdateAfterSingleSuccessfulTxExecution(from common.Address, touchedAddresses map[common.Address]*state.InfoReadWrite) {
func (w *Worker) UpdateAfterSingleSuccessfulTxExecution(from common.Address, touchedAddresses map[common.Address]*state.InfoReadWrite) []*TxTracker {
w.workerMutex.Lock()
defer w.workerMutex.Unlock()
if len(touchedAddresses) == 0 {
log.Errorf("UpdateAfterSingleSuccessfulTxExecution touchedAddresses is nil or empty")
}

txsToDelete := make([]*TxTracker, 0)
touchedFrom, found := touchedAddresses[from]
if found {
fromNonce, fromBalance := touchedFrom.Nonce, touchedFrom.Balance
w.applyAddressUpdate(from, fromNonce, fromBalance)
_, _, txsToDelete = w.applyAddressUpdate(from, fromNonce, fromBalance)
} else {
log.Errorf("UpdateAfterSingleSuccessfulTxExecution from(%s) not found in touchedAddresses", from.String())
}

for addr, addressInfo := range touchedAddresses {
if addr != from {
w.applyAddressUpdate(addr, nil, addressInfo.Balance)
_, _, txsToDeleteTemp := w.applyAddressUpdate(addr, nil, addressInfo.Balance)
txsToDelete = append(txsToDelete, txsToDeleteTemp...)
}
}
return txsToDelete
}

// MoveTxToNotReady move a tx to not ready after it fails to execute
func (w *Worker) MoveTxToNotReady(txHash common.Hash, from common.Address, actualNonce *uint64, actualBalance *big.Int) {
func (w *Worker) MoveTxToNotReady(txHash common.Hash, from common.Address, actualNonce *uint64, actualBalance *big.Int) []*TxTracker {
w.workerMutex.Lock()
defer w.workerMutex.Unlock()

Expand All @@ -152,8 +154,9 @@ func (w *Worker) MoveTxToNotReady(txHash common.Hash, from common.Address, actua
// TODO: how to manage this?
}
}
_, _, txsToDelete := w.applyAddressUpdate(from, actualNonce, actualBalance)

w.applyAddressUpdate(from, actualNonce, actualBalance)
return txsToDelete
}

// DeleteTx delete the tx after it fails to execute
Expand Down