change implementation

eth/api.go

@@ -836,40 +836,28 @@ func (api *ScrollAPI) CalculateRowConsumptionByBlockNumber(ctx context.Context,
 
 // SetRollupEventSyncedL1Height sets the synced L1 height for rollup event synchronization
 func (api *ScrollAPI) SetRollupEventSyncedL1Height(height uint64) error {
-	prevHeightPtr := rawdb.ReadRollupEventSyncedL1BlockNumber(api.eth.ChainDb())
-
-	if prevHeightPtr == nil {
-		log.Warn("No previous rollup event synced L1 height found in database")
-		return fmt.Errorf("no previous rollup event synced L1 height found in database")
-	}
-
-	prevHeight := *prevHeightPtr
-	if height >= prevHeight {
-		log.Warn("New rollup event synced L1 height is not lower than previous height", "newHeight", height, "prevHeight", prevHeight)
-		return fmt.Errorf("new rollup event synced L1 height (%d) is not lower than previous height (%d)", height, prevHeight)
+	rollupSyncService := api.eth.GetRollupSyncService()
+	if rollupSyncService == nil {
+		return errors.New("RollupSyncService is not available")
 	}
 
 	log.Info("Setting rollup event synced L1 height", "height", height)
-	rawdb.WriteRollupEventSyncedL1BlockNumber(api.eth.ChainDb(), height)
+
+	rollupSyncService.ResetToHeight(height)
+
 	return nil
 }
 
 // SetL1MessageSyncedL1Height sets the synced L1 height for L1 message synchronization
 func (api *ScrollAPI) SetL1MessageSyncedL1Height(height uint64) error {
-	prevHeightPtr := rawdb.ReadSyncedL1BlockNumber(api.eth.ChainDb())
-
-	if prevHeightPtr == nil {
-		log.Warn("No previous L1 message synced L1 height found in database")
-		return fmt.Errorf("no previous L1 message synced L1 height found in database")
-	}
-
-	prevHeight := *prevHeightPtr
-	if height >= prevHeight {
-		log.Warn("New L1 message synced L1 height is not lower than previous height", "newHeight", height, "prevHeight", prevHeight)
-		return fmt.Errorf("new L1 message synced L1 height (%d) is not lower than previous height (%d)", height, prevHeight)
+	syncService := api.eth.GetSyncService()
+	if syncService == nil {
+		return errors.New("SyncService is not available")
 	}
 
 	log.Info("Setting L1 message synced L1 height", "height", height)
-	rawdb.WriteSyncedL1BlockNumber(api.eth.ChainDb(), height)
+
+	syncService.ResetToHeight(height)
+
 	return nil
 }

eth/backend.go

@@ -609,3 +609,15 @@ func (s *Ethereum) Stop() error {
 
 	return nil
 }
+
+// GetRollupSyncService returns the RollupSyncService of the Ethereum instance.
+// It returns nil if the service is not initialized.
+func (e *Ethereum) GetRollupSyncService() *rollup_sync_service.RollupSyncService {
+	return e.rollupSyncService
+}
+
+// GetSyncService returns the SyncService of the Ethereum instance.
+// It returns nil if the service is not initialized.
+func (e *Ethereum) GetSyncService() *sync_service.SyncService {
+	return e.syncService
+}

rollup/rollup_sync_service/rollup_sync_service.go

@@ -157,6 +157,19 @@ func (s *RollupSyncService) Stop() {
 	}
 }
 
+// ResetToHeight resets the RollupSyncService to a specific L1 block height
+func (s *RollupSyncService) ResetToHeight(height uint64) {
+	s.Stop()
+
+	s.latestProcessedBlock = height
+
+	rawdb.WriteRollupEventSyncedL1BlockNumber(s.db, height)
+
+	log.Info("Reset rollup sync service", "height", height)
+
+	go s.Start()
+}
+
 func (s *RollupSyncService) fetchRollupEvents() {
 	latestConfirmed, err := s.client.getLatestFinalizedBlockNumber()
 	if err != nil {

rollup/sync_service/sync_service.go

@@ -139,6 +139,19 @@ func (s *SyncService) Stop() {
 	}
 }
 
+// ResetToHeight resets the SyncService to a specific L1 block height
+func (s *SyncService) ResetToHeight(height uint64) {
+	s.Stop()
+
+	s.latestProcessedBlock = height
+
+	rawdb.WriteSyncedL1BlockNumber(s.db, height)
+
+	log.Info("Reset sync service", "height", height)
+
+	go s.Start()
+}
+
 // SubscribeNewL1MsgsEvent registers a subscription of NewL1MsgsEvent and
 // starts sending event to the given channel.
 func (s *SyncService) SubscribeNewL1MsgsEvent(ch chan<- core.NewL1MsgsEvent) event.Subscription {