From 93a9497d754a089721956ef885fa2913c4e9b303 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Garamv=C3=B6lgyi?= Date: Mon, 18 Sep 2023 10:34:51 +0200 Subject: [PATCH] refactor(worker): order L1 messages by queue index (#512) * refactor(worker): order L1 messages by queue index * bump version * bump version --- core/types/transaction.go | 57 ++++++++++++++++++++++++++++++++++ core/types/transaction_test.go | 57 ++++++++++++++++++++++++++++++++++ miner/worker.go | 31 +++++++----------- params/version.go | 2 +- 4 files changed, 126 insertions(+), 21 deletions(-) diff --git a/core/types/transaction.go b/core/types/transaction.go index 84a960eb6d..b38bc38a86 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -20,14 +20,17 @@ import ( "bytes" "container/heap" "errors" + "fmt" "io" "math/big" + "sort" "sync/atomic" "time" "github.com/scroll-tech/go-ethereum/common" "github.com/scroll-tech/go-ethereum/common/math" "github.com/scroll-tech/go-ethereum/crypto" + "github.com/scroll-tech/go-ethereum/log" "github.com/scroll-tech/go-ethereum/rlp" ) @@ -522,6 +525,18 @@ func (s *TxByPriceAndTime) Pop() interface{} { return x } +// OrderedTransactionSet represents a set of transactions and some ordering on top of this set. +type OrderedTransactionSet interface { + // Peek returns the next transaction. + Peek() *Transaction + + // Shift removes the next transaction. + Shift() + + // Pop removes all transactions from the current account. + Pop() +} + // TransactionsByPriceAndNonce represents a set of transactions that can return // transactions in a profit-maximizing sorted order, while supporting removing // entire batches of transactions for non-executable accounts. @@ -590,6 +605,48 @@ func (t *TransactionsByPriceAndNonce) Pop() { heap.Pop(&t.heads) } +// L1MessagesByQueueIndex represents a set of L1 messages ordered by their queue indices. +type L1MessagesByQueueIndex struct { + msgs []L1MessageTx +} + +func NewL1MessagesByQueueIndex(msgs []L1MessageTx) (*L1MessagesByQueueIndex, error) { + // sort by queue index + sort.Slice(msgs, func(i, j int) bool { + return msgs[i].QueueIndex < msgs[j].QueueIndex + }) + + // check for duplicates/gaps + for ii := 0; ii < len(msgs)-1; ii++ { + current := msgs[ii].QueueIndex + next := msgs[ii+1].QueueIndex + if next != current+1 { + return nil, fmt.Errorf("invalid L1 message set, current index: %d, next index: %d", current, next) + } + } + + return &L1MessagesByQueueIndex{msgs: msgs}, nil +} + +func (t *L1MessagesByQueueIndex) Peek() *Transaction { + if len(t.msgs) == 0 { + return nil + } + return NewTx(&t.msgs[0]) +} + +func (t *L1MessagesByQueueIndex) Shift() { + t.msgs = t.msgs[1:] +} + +func (t *L1MessagesByQueueIndex) Pop() { + log.Error("Pop() is called on L1MessagesByQueueIndex") + + // this is a logic error, the intention should be "Shift()", + // so we will follow the same behavior in Pop + t.Shift() +} + // Message is a fully derived transaction and implements core.Message // // NOTE: In a future PR this will be removed. diff --git a/core/types/transaction_test.go b/core/types/transaction_test.go index e6cf5545bd..417bc546f7 100644 --- a/core/types/transaction_test.go +++ b/core/types/transaction_test.go @@ -27,6 +27,8 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" + "github.com/scroll-tech/go-ethereum/common" "github.com/scroll-tech/go-ethereum/crypto" "github.com/scroll-tech/go-ethereum/rlp" @@ -405,6 +407,61 @@ func TestTransactionTimeSort(t *testing.T) { } } +func TestL1MessageQueueIndexSort(t *testing.T) { + assert := assert.New(t) + + msgs := []L1MessageTx{ + {QueueIndex: 3, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 6, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 2, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{3}}, + {QueueIndex: 5, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{3}}, + {QueueIndex: 4, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{3}}, + } + + txset, err := NewL1MessagesByQueueIndex(msgs) + assert.NoError(err) + + nextIndex := uint64(1) + + for { + tx := txset.Peek() + if tx == nil { + break + } + + assert.True(tx.IsL1MessageTx()) + assert.Equal(nextIndex, tx.AsL1MessageTx().QueueIndex) + + txset.Shift() + nextIndex++ + } + + assert.Equal(uint64(7), nextIndex) +} + +func TestL1MessageQueueIndexSortInvalid(t *testing.T) { + assert := assert.New(t) + + msgs := []L1MessageTx{ + {QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 2, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + } + + _, err := NewL1MessagesByQueueIndex(msgs) + assert.Error(err) + + msgs = []L1MessageTx{ + {QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 3, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + {QueueIndex: 4, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}, + } + + _, err = NewL1MessagesByQueueIndex(msgs) + assert.Error(err) +} + // TestTransactionCoding tests serializing/de-serializing to/from rlp and JSON. func TestTransactionCoding(t *testing.T) { key, err := crypto.GenerateKey() diff --git a/miner/worker.go b/miner/worker.go index 91855b8930..1c27899638 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -962,7 +962,7 @@ func (w *worker) commitTransaction(tx *types.Transaction, coinbase common.Addres return receipt.Logs, traces, nil } -func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coinbase common.Address, interrupt *int32) (bool, bool) { +func (w *worker) commitTransactions(txs types.OrderedTransactionSet, coinbase common.Address, interrupt *int32) (bool, bool) { var circuitCapacityReached bool // Short circuit if current is nil @@ -1348,29 +1348,16 @@ func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64) w.commit(uncles, nil, false, tstart) } // fetch l1Txs - l1Txs := make(map[common.Address]types.Transactions) - pendingL1Txs := 0 + var l1Messages []types.L1MessageTx if w.chainConfig.Scroll.ShouldIncludeL1Messages() { - l1Messages := w.collectPendingL1Messages(env.nextL1MsgIndex) - pendingL1Txs = len(l1Messages) - for _, l1msg := range l1Messages { - tx := types.NewTx(&l1msg) - sender := l1msg.Sender - senderTxs, ok := l1Txs[sender] - if ok { - senderTxs = append(senderTxs, tx) - l1Txs[sender] = senderTxs - } else { - l1Txs[sender] = types.Transactions{tx} - } - } + l1Messages = w.collectPendingL1Messages(env.nextL1MsgIndex) } // Fill the block with all available pending transactions. pending := w.eth.TxPool().Pending(true) // Short circuit if there is no available pending transactions. // But if we disable empty precommit already, ignore it. Since // empty block is necessary to keep the liveness of the network. - if len(pending) == 0 && pendingL1Txs == 0 && atomic.LoadUint32(&w.noempty) == 0 { + if len(pending) == 0 && len(l1Messages) == 0 && atomic.LoadUint32(&w.noempty) == 0 { w.updateSnapshot() return } @@ -1383,9 +1370,13 @@ func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64) } } var skipCommit, circuitCapacityReached bool - if w.chainConfig.Scroll.ShouldIncludeL1Messages() && len(l1Txs) > 0 { - log.Trace("Processing L1 messages for inclusion", "count", pendingL1Txs) - txs := types.NewTransactionsByPriceAndNonce(w.current.signer, l1Txs, header.BaseFee) + if w.chainConfig.Scroll.ShouldIncludeL1Messages() && len(l1Messages) > 0 { + log.Trace("Processing L1 messages for inclusion", "count", len(l1Messages)) + txs, err := types.NewL1MessagesByQueueIndex(l1Messages) + if err != nil { + log.Error("Failed to create L1 message set", "l1Messages", l1Messages, "err", err) + return + } skipCommit, circuitCapacityReached = w.commitTransactions(txs, w.coinbase, interrupt) if skipCommit { return diff --git a/params/version.go b/params/version.go index c22c426b71..a49a7b4401 100644 --- a/params/version.go +++ b/params/version.go @@ -24,7 +24,7 @@ import ( const ( VersionMajor = 4 // Major version component of the current release VersionMinor = 4 // Minor version component of the current release - VersionPatch = 9 // Patch version component of the current release + VersionPatch = 10 // Patch version component of the current release VersionMeta = "sepolia" // Version metadata to append to the version string )