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

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

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

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

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
 )