From fe8232a40ffb3853700dcf2fcb4cd69d0ff6d08c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Garamv=C3=B6lgyi?= Date: Fri, 29 Sep 2023 21:16:37 +0200 Subject: [PATCH] fix(worker): short-circuit l1 message iteration (#525) * fix(worker): short-circuit l1 message iteration * Update core/rawdb/accessors_l1_message_test.go Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com> --------- Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com> --- core/rawdb/accessors_l1_message.go | 52 ++++++++++++++++++++++--- core/rawdb/accessors_l1_message_test.go | 32 +++++++++++++++ core/rawdb/schema.go | 1 + params/version.go | 2 +- 4 files changed, 80 insertions(+), 7 deletions(-) diff --git a/core/rawdb/accessors_l1_message.go b/core/rawdb/accessors_l1_message.go index c2c19f9056..19ee8f2244 100644 --- a/core/rawdb/accessors_l1_message.go +++ b/core/rawdb/accessors_l1_message.go @@ -43,7 +43,38 @@ func ReadSyncedL1BlockNumber(db ethdb.Reader) *uint64 { return &value } +// WriteHighestSyncedQueueIndex writes the highest synced L1 message queue index to the database. +func WriteHighestSyncedQueueIndex(db ethdb.KeyValueWriter, queueIndex uint64) { + value := big.NewInt(0).SetUint64(queueIndex).Bytes() + + if err := db.Put(highestSyncedQueueIndexKey, value); err != nil { + log.Crit("Failed to update highest synced L1 message queue index", "err", err) + } +} + +// ReadHighestSyncedQueueIndex retrieves the highest synced L1 message queue index. +func ReadHighestSyncedQueueIndex(db ethdb.Reader) uint64 { + data, err := db.Get(highestSyncedQueueIndexKey) + if err != nil && isNotFoundErr(err) { + return 0 + } + if err != nil { + log.Crit("Failed to read highest synced L1 message queue index from database", "err", err) + } + if len(data) == 0 { + return 0 + } + + number := new(big.Int).SetBytes(data) + if !number.IsUint64() { + log.Crit("Unexpected highest synced L1 block number in database", "number", number) + } + + return number.Uint64() +} + // WriteL1Message writes an L1 message to the database. +// We assume that L1 messages are written to DB following their queue index order. func WriteL1Message(db ethdb.KeyValueWriter, l1Msg types.L1MessageTx) { bytes, err := rlp.EncodeToBytes(l1Msg) if err != nil { @@ -52,6 +83,8 @@ func WriteL1Message(db ethdb.KeyValueWriter, l1Msg types.L1MessageTx) { if err := db.Put(L1MessageKey(l1Msg.QueueIndex), bytes); err != nil { log.Crit("Failed to store L1 message", "err", err) } + + WriteHighestSyncedQueueIndex(db, l1Msg.QueueIndex) } // WriteL1Messages writes an array of L1 messages to the database. @@ -91,20 +124,23 @@ func ReadL1Message(db ethdb.Reader, queueIndex uint64) *types.L1MessageTx { // allows us to iterate over L1 messages in the database. It // implements an interface similar to ethdb.Iterator. type L1MessageIterator struct { - inner ethdb.Iterator - keyLength int + inner ethdb.Iterator + keyLength int + maxQueueIndex uint64 } // IterateL1MessagesFrom creates an L1MessageIterator that iterates over // all L1 message in the database starting at the provided enqueue index. -func IterateL1MessagesFrom(db ethdb.Iteratee, fromQueueIndex uint64) L1MessageIterator { +func IterateL1MessagesFrom(db ethdb.Database, fromQueueIndex uint64) L1MessageIterator { start := encodeBigEndian(fromQueueIndex) it := db.NewIterator(l1MessagePrefix, start) keyLength := len(l1MessagePrefix) + 8 + maxQueueIndex := ReadHighestSyncedQueueIndex(db) return L1MessageIterator{ - inner: it, - keyLength: keyLength, + inner: it, + keyLength: keyLength, + maxQueueIndex: maxQueueIndex, } } @@ -145,7 +181,7 @@ func (it *L1MessageIterator) Release() { } // ReadL1MessagesFrom retrieves up to `maxCount` L1 messages starting at `startIndex`. -func ReadL1MessagesFrom(db ethdb.Iteratee, startIndex, maxCount uint64) []types.L1MessageTx { +func ReadL1MessagesFrom(db ethdb.Database, startIndex, maxCount uint64) []types.L1MessageTx { msgs := make([]types.L1MessageTx, 0, maxCount) it := IterateL1MessagesFrom(db, startIndex) defer it.Release() @@ -170,6 +206,10 @@ func ReadL1MessagesFrom(db ethdb.Iteratee, startIndex, maxCount uint64) []types. msgs = append(msgs, msg) index += 1 count -= 1 + + if msg.QueueIndex == it.maxQueueIndex { + break + } } return msgs diff --git a/core/rawdb/accessors_l1_message_test.go b/core/rawdb/accessors_l1_message_test.go index fe203d755a..314d6a604a 100644 --- a/core/rawdb/accessors_l1_message_test.go +++ b/core/rawdb/accessors_l1_message_test.go @@ -48,6 +48,11 @@ func TestReadWriteL1Message(t *testing.T) { if got == nil || got.QueueIndex != queueIndex { t.Fatal("L1 message mismatch", "expected", queueIndex, "got", got) } + + max := ReadHighestSyncedQueueIndex(db) + if max != 123 { + t.Fatal("max index mismatch", "expected", 123, "got", max) + } } func TestIterateL1Message(t *testing.T) { @@ -62,6 +67,11 @@ func TestIterateL1Message(t *testing.T) { db := NewMemoryDatabase() WriteL1Messages(db, msgs) + max := ReadHighestSyncedQueueIndex(db) + if max != 1000 { + t.Fatal("max index mismatch", "expected", 1000, "got", max) + } + it := IterateL1MessagesFrom(db, 103) defer it.Release() @@ -125,3 +135,25 @@ func TestReadWriteLastL1MessageInL2Block(t *testing.T) { } } } + +func TestIterationStopsAtMaxQueueIndex(t *testing.T) { + msgs := []types.L1MessageTx{ + newL1MessageTx(100), + newL1MessageTx(101), + newL1MessageTx(102), + newL1MessageTx(103), + } + + db := NewMemoryDatabase() + WriteL1Messages(db, msgs) + + // artificially change max index from 103 to 102 + WriteHighestSyncedQueueIndex(db, 102) + + // iteration should terminate at 102 and not read 103 + got := ReadL1MessagesFrom(db, 100, 10) + + if len(got) != 3 { + t.Fatal("Invalid length", "expected", 3, "got", len(got)) + } +} diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 960e8cbcf6..bd637abaef 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -107,6 +107,7 @@ var ( syncedL1BlockNumberKey = []byte("LastSyncedL1BlockNumber") l1MessagePrefix = []byte("l1") // l1MessagePrefix + queueIndex (uint64 big endian) -> L1MessageTx firstQueueIndexNotInL2BlockPrefix = []byte("q") // firstQueueIndexNotInL2BlockPrefix + L2 block hash -> enqueue index + highestSyncedQueueIndexKey = []byte("HighestSyncedQueueIndex") // Row consumption rowConsumptionPrefix = []byte("rc") // rowConsumptionPrefix + hash -> row consumption by block diff --git a/params/version.go b/params/version.go index 0602ba04a0..cac46c3109 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 = 13 // Patch version component of the current release + VersionPatch = 14 // Patch version component of the current release VersionMeta = "sepolia" // Version metadata to append to the version string )