From 6954c73459382d47329135705afeb3f733bad9c2 Mon Sep 17 00:00:00 2001
From: Jonas Theis <4181434+jonastheis@users.noreply.github.com>
Date: Wed, 5 Mar 2025 18:00:31 +0800
Subject: [PATCH] fix(sync service): skip messages with lower index (#1134)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* skip messages with lower index

* address review comments

* chore: auto version bump [bot]

---------

Co-authored-by: Thegaram <Thegaram@users.noreply.github.com>
---
 params/version.go                   |  2 +-
 rollup/sync_service/sync_service.go | 32 ++++++++++++++++++++++++++---
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/params/version.go b/params/version.go
index 1e641d4b57..6badac58f8 100644
--- a/params/version.go
+++ b/params/version.go
@@ -24,7 +24,7 @@ import (
 const (
 	VersionMajor = 5         // Major version component of the current release
 	VersionMinor = 8         // Minor version component of the current release
-	VersionPatch = 20        // Patch version component of the current release
+	VersionPatch = 21        // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )
 
diff --git a/rollup/sync_service/sync_service.go b/rollup/sync_service/sync_service.go
index 65743b5e7e..a8edba793b 100644
--- a/rollup/sync_service/sync_service.go
+++ b/rollup/sync_service/sync_service.go
@@ -1,6 +1,7 @@
 package sync_service
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"reflect"
@@ -15,6 +16,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/metrics"
 	"github.com/scroll-tech/go-ethereum/node"
 	"github.com/scroll-tech/go-ethereum/params"
+	"github.com/scroll-tech/go-ethereum/rlp"
 )
 
 const (
@@ -272,19 +274,43 @@ func (s *SyncService) fetchMessages() {
 
 		if len(msgs) > 0 {
 			log.Debug("Received new L1 events", "fromBlock", from, "toBlock", to, "count", len(msgs))
-			rawdb.WriteL1Messages(batchWriter, msgs) // collect messages in memory
-			numMsgsCollected += len(msgs)
 		}
 
 		for _, msg := range msgs {
 			if msg.QueueIndex > 0 {
 				queueIndex++
 			}
+
 			// check if received queue index matches expected queue index
-			if msg.QueueIndex != queueIndex {
+			if msg.QueueIndex > queueIndex {
 				log.Error("Unexpected queue index in SyncService", "expected", queueIndex, "got", msg.QueueIndex, "msg", msg)
 				return // do not flush inconsistent data to disk
 			}
+
+			// compare with stored message in database, abort if not equal, ignore if already exists
+			if msg.QueueIndex < queueIndex {
+				log.Warn("Duplicate queue index in SyncService", "expected", queueIndex, "got", msg.QueueIndex)
+
+				receivedMsgBytes, err := rlp.EncodeToBytes(msg)
+				if err != nil {
+					log.Error("Failed to encode message", "err", err)
+					return
+				}
+				storedMsgBytes := rawdb.ReadL1MessageRLP(s.db, msg.QueueIndex)
+				if !bytes.Equal(storedMsgBytes, receivedMsgBytes) {
+					storedL1Message := rawdb.ReadL1Message(s.db, msg.QueueIndex)
+					log.Error("Stored message at same queue index does not match received message", "queueIndex", msg.QueueIndex, "expected", storedL1Message, "got", msg)
+					return
+				}
+
+				// already exists, ignore
+				queueIndex--
+				continue
+			}
+
+			// store message to database (collected in memory and flushed periodically)
+			rawdb.WriteL1Message(batchWriter, msg)
+			numMsgsCollected++
 		}
 
 		numBlocksPendingDbWrite += to - from + 1