Revert "feat(tx-submitter): optimize rollup transaction processing" (#738)

Author: WorldDogs

tx-submitter/.gitignore

@@ -1,56 +1,9 @@
-# IDE and Editor files
-.idea/
-.vscode/
-*.swp
-*.swo
-*~
-
-# OS generated files
-.DS_Store
-.DS_Store?
-._*
-.Spotlight-V100
-.Trashes
-ehthumbs.db
-Thumbs.db
-
-# Go specific
-*.exe
-*.exe~
-*.dll
-*.so
-*.dylib
-*.test
-*.out
-go.work
-
-# Binary files
-tx-submitter
-**/tx-submitter
-build/
-*debug_bin*
-
-# Config and Environment files
+.idea
 .env*
-
-# Test and Debug files
-test/
-testleveldb/
-test_data/
-debug/
-*.test
+build
+tx-submitter
+.DS_Store
+.vscode
 *.log
-*.pprof
-cpu.prof
-mem.prof
-trace.out
-
-# Coverage reports
-coverage.txt
-
-# Transaction files
-journal.rlp
-journal*.rlp
-
-# Temporary files
-tmp/
+*debug_bin*
+journal.rlp

tx-submitter/constants/methods.go

@@ -8,11 +8,10 @@ import (
 	"morph-l2/tx-submitter/utils"
 
 	"github.com/morph-l2/go-ethereum/ethdb/leveldb"
-	"github.com/syndtr/goleveldb/leveldb/errors"
 )
 
 var (
-	ErrKeyNotFound = errors.ErrNotFound
+	ErrKeyNotFound = fmt.Errorf("not found")
 )
 
 type Db struct {
@@ -56,9 +55,6 @@ func (d *Db) GetString(key string) (string, error) {
 	defer d.m.Unlock()
 	v, err := d.db.Get([]byte(key))
 	if err != nil {
-		if err == errors.ErrNotFound {
-			return "", ErrKeyNotFound
-		}
 		return "", fmt.Errorf("failed to get key from leveldb %w", err)
 	}
 	return string(v), nil

tx-submitter/db/db.go

@@ -16,10 +16,10 @@ type EventIndexer struct {
 	deployBlock *big.Int // Block number of contract deployment
 	filterQuery ethereum.FilterQuery
 	indexStep   uint64 // index step
-	storage     IEventStorage
+	storage     *EventInfoStorage
 }
 
-func NewEventIndexer(client *ethclient.Client, deployedBlock *big.Int, filter ethereum.FilterQuery, indexStep uint64, storage IEventStorage) *EventIndexer {
+func NewEventIndexer(client *ethclient.Client, deployedBlock *big.Int, filter ethereum.FilterQuery, indexStep uint64, storage *EventInfoStorage) *EventIndexer {
 	return &EventIndexer{
 		client:      client,
 		deployBlock: deployedBlock,
@@ -32,8 +32,8 @@ func NewEventIndexer(client *ethclient.Client, deployedBlock *big.Int, filter et
 func (l *EventIndexer) Index() {
 	log.Info("event indexer started")
 
-	if l.storage.BlockProcessed() == 0 {
-		l.storage.SetBlockProcessed(l.deployBlock.Uint64())
+	if l.storage.BlockProcessed == 0 {
+		l.storage.BlockProcessed = l.deployBlock.Uint64()
 		err := l.storage.Store()
 		if err != nil {
 			log.Error("failed to store initial block number", "error", err)
@@ -45,40 +45,43 @@ func (l *EventIndexer) Index() {
 	defer ticker.Stop()
 
 	for range ticker.C {
+
 		// Get the current block number
 		currentBlock, err := l.client.BlockNumber(context.Background())
 		if err != nil {
 			log.Error("failed to get current block number", "error", err)
 			continue
 		}
 
-		if currentBlock <= l.storage.BlockProcessed() {
-			log.Info("no new block to index", "current_block", currentBlock, "last_processed_block", l.storage.BlockProcessed())
+		if currentBlock <= l.storage.BlockProcessed {
+			log.Info("no new block to index", "current_block", currentBlock, "last_processed_block", l.storage.BlockProcessed)
 			continue
 		}
 
 		// Perform indexing operation
-		indexedEventInfo, err := l.index(l.client, big.NewInt(int64(l.storage.BlockProcessed())), big.NewInt(int64(currentBlock)))
+		indexedEventInfo, err := l.index(l.client, big.NewInt(int64(l.storage.BlockProcessed)), big.NewInt(int64(currentBlock)))
 		if err != nil {
 			log.Error("indexing operation failed", "error", err)
 			continue
 		}
 
 		if indexedEventInfo != nil {
-			l.storage.SetBlockProcessed(indexedEventInfo.BlockProcessed)
+			l.storage.EventInfo = *indexedEventInfo
 		} else {
-			l.storage.SetBlockProcessed(currentBlock)
+			l.storage.EventInfo = EventInfo{
+				BlockProcessed: currentBlock,
+			}
 		}
-
 		// Update storage
 		err = l.storage.Store()
 		if err != nil {
 			log.Error("event index complete, failed to update storage", "error", err)
 		} else {
-			info := l.storage.EventInfo()
-			log.Info("event index complete, storage updated", "processed_block", info.BlockProcessed, "block_time", info.BlockTime)
+			log.Info("event index complete, storage updated", "processed_block", l.storage.EventInfo.BlockProcessed, "block_time", l.storage.EventInfo.BlockTime)
 		}
+
 	}
+
 }
 
 // filter logs from from_block to to_block
@@ -140,6 +143,6 @@ func (ei *EventIndexer) index(client *ethclient.Client, fromBlock, toBlock *big.
 func (l *EventIndexer) GetFilter() ethereum.FilterQuery {
 	return l.filterQuery
 }
-func (l *EventIndexer) GetStorage() IEventStorage {
+func (l *EventIndexer) GetStorage() *EventInfoStorage {
 	return l.storage
 }

tx-submitter/db/interface.go

@@ -7,16 +7,12 @@ import (
 
 	"morph-l2/tx-submitter/db"
 	"morph-l2/tx-submitter/params"
+	"morph-l2/tx-submitter/utils"
 )
 
 type IEventStorage interface {
 	Store() error
 	Load() error
-	BlockProcessed() uint64
-	SetBlockProcessed(blockNum uint64)
-	BlockTime() uint64
-	SetBlockTime(blockTime uint64)
-	EventInfo() EventInfo
 }
 
 type EventInfo struct {
@@ -25,20 +21,21 @@ type EventInfo struct {
 }
 
 type EventInfoStorage struct {
-	eventInfo EventInfo
-	db        db.Database
-	mu        sync.RWMutex
+	EventInfo
+	db *db.Db
+	mu sync.Mutex
 }
 
-func NewEventInfoStorage(db db.Database) *EventInfoStorage {
+func NewEventInfoStorage(db *db.Db) *EventInfoStorage {
 	return &EventInfoStorage{
 		db: db,
 	}
 }
 
 func (e *EventInfoStorage) Store() error {
+
 	// Convert struct to JSON string
-	jsonData, err := json.Marshal(e.eventInfo)
+	jsonData, err := json.Marshal(e.EventInfo)
 	if err != nil {
 		return fmt.Errorf("failed to convert struct to JSON: %w", err)
 	}
@@ -52,57 +49,31 @@ func (e *EventInfoStorage) Store() error {
 	}
 	return nil
 }
-
 func (e *EventInfoStorage) Load() error {
 	e.mu.Lock()
 	defer e.mu.Unlock()
-
-	jsonStr, err := e.db.GetString(params.EventInfoKey)
+	evnetInfo, err := e.db.GetString(params.EventInfoKey)
 	if err != nil {
-		if err == db.ErrKeyNotFound {
-			// Initialize with default values if not found
-			e.eventInfo = EventInfo{}
+		if utils.ErrStringMatch(err, db.ErrKeyNotFound) {
+			e.EventInfo = EventInfo{}
+			jsonData, err := json.Marshal(e.EventInfo)
+			if err != nil {
+				return fmt.Errorf("failed to marshal json: %w", err)
+			}
+			err = e.db.PutString(params.EventInfoKey, string(jsonData))
+			if err != nil {
+				return fmt.Errorf("failed to init eventinfo to db: %w", err)
+			}
 			return nil
 		}
-		return fmt.Errorf("failed to read from db: %w", err)
+		return fmt.Errorf("failed to load eventinfo from db: %w", err)
 	}
 
-	err = json.Unmarshal([]byte(jsonStr), &e.eventInfo)
+	// parse json data to struct
+	err = json.Unmarshal([]byte(evnetInfo), &e.EventInfo)
 	if err != nil {
-		return fmt.Errorf("failed to parse JSON: %w", err)
+		return fmt.Errorf("failed to unmarshal JSON: %w", err)
 	}
 
 	return nil
 }
-
-func (e *EventInfoStorage) BlockProcessed() uint64 {
-	e.mu.RLock()
-	defer e.mu.RUnlock()
-	return e.eventInfo.BlockProcessed
-}
-
-func (e *EventInfoStorage) SetBlockProcessed(blockNum uint64) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	e.eventInfo.BlockProcessed = blockNum
-}
-
-func (e *EventInfoStorage) BlockTime() uint64 {
-	e.mu.RLock()
-	defer e.mu.RUnlock()
-	return e.eventInfo.BlockTime
-}
-
-func (e *EventInfoStorage) SetBlockTime(blockTime uint64) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	e.eventInfo.BlockTime = blockTime
-}
-
-// EventInfo returns a copy of the current event info.
-// This ensures thread safety without exposing the internal state.
-func (e *EventInfoStorage) EventInfo() EventInfo {
-	e.mu.RLock()
-	defer e.mu.RUnlock()
-	return e.eventInfo
-}

tx-submitter/event/indexer.go

@@ -2,39 +2,27 @@ package event
 
 import (
 	"morph-l2/tx-submitter/db"
-	"os"
-	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 )
 
 func TestEventInfoStorage(t *testing.T) {
-	// Create a unique test directory
-	testDir := filepath.Join(t.TempDir(), "testleveldb")
 
-	// Cleanup before test (in case it exists)
-	os.RemoveAll(testDir)
-
-	// Cleanup after test
-	t.Cleanup(func() {
-		os.RemoveAll(testDir)
-	})
-
-	db, err := db.New(testDir)
+	db, err := db.New("./testleveldb")
 	require.NoError(t, err)
 	storage := NewEventInfoStorage(db)
 	err = storage.Load()
 	require.NoError(t, err)
 
-	storage.SetBlockTime(100)
-	storage.SetBlockProcessed(100)
+	storage.BlockTime = 100
+	storage.BlockProcessed = 100
 	err = storage.Store()
 	require.NoError(t, err)
 
 	storage2 := NewEventInfoStorage(db)
 	err = storage2.Load()
 	require.NoError(t, err)
-	require.Equal(t, storage.BlockTime(), storage2.BlockTime())
-	require.Equal(t, storage.BlockProcessed(), storage2.BlockProcessed())
+	require.Equal(t, storage.BlockTime, storage2.BlockTime)
+	require.Equal(t, storage.BlockProcessed, storage2.BlockProcessed)
 }

tx-submitter/event/storage.go

@@ -82,7 +82,7 @@ func (j *Journal) ParseAllTxs() ([]*types.Transaction, error) {
 	content, err := readFileContent(j.path)
 
 	if err != nil {
-		return nil, fmt.Errorf("failed to read journal file: %w", err)
+		return nil, fmt.Errorf("failed to parse txs: %w", err)
 	}
 
 	if len(content) == 0 {