feat: multi blob

common/batch/batch_cache_genesis_header_test.go

@@ -0,0 +1,142 @@
+package batch
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/morph-l2/go-ethereum/common/hexutil"
+	"github.com/morph-l2/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	// Fill this with hex-encoded batch header bytes, e.g. "0x00....".
+	// This test will use it as the genesis parent header to initialize cache.
+	globalGenesisBatchHeaderHex  = "0x00000000000000000000000000000000000000000000000000d81a073a4abd227068a2a334f4a41b3abba26144dc866a78ed28e2ae90f86f5a010657f37554c781402a22917dee2f75def7ab966d7b770905398eba3c4440140000000000000000000000000000000000000000000000000000000000000000290233e7a85533655c301d3e1043f03acd5427c73d1bbcbf8784db3f3974327f000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+	globalGenesisBatchHeader     *BatchHeaderBytes
+	globalGenesisBatchHeaderErr  error
+	globalGenesisBatchHeaderOnce sync.Once
+
+	// Global overrides for cache batch config in tests (instead of updateBatchConfigFromGov).
+	globalBatchTimeoutForTest  uint64 = 10000000
+	globalBlockIntervalForTest uint64 = 10000
+)
+
+func ensureGlobalGenesisBatchHeader() error {
+	globalGenesisBatchHeaderOnce.Do(func() {
+		if globalGenesisBatchHeaderHex == "" {
+			globalGenesisBatchHeaderErr = fmt.Errorf("globalGenesisBatchHeaderHex is empty")
+			return
+		}
+		raw, err := hexutil.Decode(globalGenesisBatchHeaderHex)
+		if err != nil {
+			globalGenesisBatchHeaderErr = fmt.Errorf("decode globalGenesisBatchHeaderHex failed: %w", err)
+			return
+		}
+		header := BatchHeaderBytes(raw)
+		if err := header.validate(); err != nil {
+			globalGenesisBatchHeaderErr = fmt.Errorf("invalid global genesis batch header: %w", err)
+			return
+		}
+		globalGenesisBatchHeader = &header
+	})
+	return globalGenesisBatchHeaderErr
+}
+
+// initCacheWithGlobalGenesisHeader initializes cache base fields from the
+// globally cached genesis batch header, instead of loading through Init().
+func initCacheWithGlobalGenesisHeader(cache *BatchCache) error {
+	if err := ensureGlobalGenesisBatchHeader(); err != nil {
+		return err
+	}
+	if globalGenesisBatchHeader == nil {
+		return ErrKeyNotFound
+	}
+	// Use global test knobs instead of querying gov config from chain.
+	cache.batchTimeOut = globalBatchTimeoutForTest
+	cache.blockInterval = globalBlockIntervalForTest
+	headerCopy := make(BatchHeaderBytes, len(*globalGenesisBatchHeader))
+	copy(headerCopy, *globalGenesisBatchHeader)
+	cache.parentBatchHeader = &headerCopy
+
+	prevStateRoot, err := cache.parentBatchHeader.PostStateRoot()
+	if err != nil {
+		return err
+	}
+	cache.prevStateRoot = prevStateRoot
+
+	totalL1MessagePopped, err := cache.parentBatchHeader.TotalL1MessagePopped()
+	if err != nil {
+		return err
+	}
+	cache.totalL1MessagePopped = totalL1MessagePopped
+
+	lastPackedBlockHeight, err := cache.parentBatchHeader.LastBlockNumber()
+	if err != nil {
+		lastPackedBlockHeight = 0
+	}
+	cache.lastPackedBlockHeight = lastPackedBlockHeight
+	cache.currentBlockNumber = lastPackedBlockHeight
+	cache.initDone = true
+
+	return nil
+}
+
+func TestBatchCacheInitWithGlobalGenesisHeader(t *testing.T) {
+	testDB := openTestKV(t)
+	a := func(uint64) bool { return true }
+	cache := NewBatchCache(nil, a, 3, l1Client, &SingleL2Client{C: l2Client}, rollupContract, l2Gov, testDB)
+
+	var batchCacheSyncMu sync.Mutex
+	done := make(chan error, 1)
+	go func() {
+		batchCacheSyncMu.Lock()
+		defer batchCacheSyncMu.Unlock()
+		for {
+			if err := initCacheWithGlobalGenesisHeader(cache); err != nil {
+				log.Error("init with global genesis header failed, wait for next try", "error", err)
+				time.Sleep(3 * time.Second)
+				continue
+			}
+			done <- nil
+			return
+		}
+	}()
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(20 * time.Second):
+		t.Fatal("timeout waiting for cache init with global genesis header")
+	}
+
+	require.True(t, cache.initDone)
+	require.NotNil(t, cache.parentBatchHeader)
+	version, err := cache.parentBatchHeader.Version()
+	require.NoError(t, err)
+	require.Equal(t, uint8(BatchHeaderVersion0), version)
+	require.Equal(t, cache.lastPackedBlockHeight, cache.currentBlockNumber)
+	_, err = cache.l2Clients.BlockNumber(context.Background())
+	require.NoError(t, err)
+
+	go testLoop(cache.ctx, 5*time.Second, func() {
+		batchCacheSyncMu.Lock()
+		defer batchCacheSyncMu.Unlock()
+		err := cache.AssembleCurrentBatchHeader()
+		if err != nil {
+			log.Error("Assemble current batch failed, wait for the next try", "error", err)
+		}
+	})
+
+	// Catch CTRL-C to ensure a graceful shutdown.
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, os.Interrupt)
+
+	// Wait until the interrupt signal is received from an OS signal.
+	<-interrupt
+}

tx-submitter/batch/batch_cache_test.go → common/batch/batch_cache_test.go

@@ -3,16 +3,11 @@ package batch
 import (
 	"os"
 	"os/signal"
-	"path/filepath"
 	"sync"
 	"testing"
 	"time"
 
 	"morph-l2/bindings/bindings"
-	"morph-l2/tx-submitter/db"
-	"morph-l2/tx-submitter/iface"
-	"morph-l2/tx-submitter/types"
-	"morph-l2/tx-submitter/utils"
 
 	"github.com/morph-l2/go-ethereum/log"
 	"github.com/stretchr/testify/require"
@@ -24,28 +19,15 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-	l2Caller, err = types.NewL2Caller([]iface.L2Client{l2Client})
+	l2Gov, err = NewL2Gov(l2Client)
 	if err != nil {
 		panic(err)
 	}
 }
 
-// setupTestDB creates a temporary database for testing
-func setupTestDB(t *testing.T) *db.Db {
-	testDir := filepath.Join(t.TempDir(), "testleveldb")
-	os.RemoveAll(testDir)
-	t.Cleanup(func() {
-		os.RemoveAll(testDir)
-	})
-
-	testDB, err := db.New(testDir)
-	require.NoError(t, err)
-	return testDB
-}
-
 func TestBatchCacheInitServer(t *testing.T) {
-	testDB := setupTestDB(t)
-	cache := NewBatchCache(nil, l1Client, []iface.L2Client{l2Client}, rollupContract, l2Caller, testDB)
+	testDB := openTestKV(t)
+	cache := NewBatchCache(nil, nil, 2, l1Client, &SingleL2Client{C: l2Client}, rollupContract, l2Gov, testDB)
 
 	var batchCacheSyncMu sync.Mutex
 
@@ -62,7 +44,7 @@ func TestBatchCacheInitServer(t *testing.T) {
 		}
 	}()
 
-	go utils.Loop(cache.ctx, 5*time.Second, func() {
+	go testLoop(cache.ctx, 5*time.Second, func() {
 		batchCacheSyncMu.Lock()
 		defer batchCacheSyncMu.Unlock()
 		err := cache.AssembleCurrentBatchHeader()
@@ -71,24 +53,21 @@ func TestBatchCacheInitServer(t *testing.T) {
 		}
 	})
 
-	// Catch CTRL-C to ensure a graceful shutdown.
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)
-
-	// Wait until the interrupt signal is received from an OS signal.
 	<-interrupt
 }
 
 func TestBatchCacheInit(t *testing.T) {
-	testDB := setupTestDB(t)
-	cache := NewBatchCache(nil, l1Client, []iface.L2Client{l2Client}, rollupContract, l2Caller, testDB)
+	testDB := openTestKV(t)
+	cache := NewBatchCache(nil, nil, 2, l1Client, &SingleL2Client{C: l2Client}, rollupContract, l2Gov, testDB)
 	err := cache.InitAndSyncFromRollup()
 	require.NoError(t, err)
 }
 
 func TestBatchCacheInitByBlockRange(t *testing.T) {
-	testDB := setupTestDB(t)
-	cache := NewBatchCache(nil, l1Client, []iface.L2Client{l2Client}, rollupContract, l2Caller, testDB)
+	testDB := openTestKV(t)
+	cache := NewBatchCache(nil, nil, 2, l1Client, &SingleL2Client{C: l2Client}, rollupContract, l2Gov, testDB)
 	err := cache.InitFromRollupByRange()
 	require.NoError(t, err)
 }

tx-submitter/batch/batch_data.go → common/batch/batch_data.go

@@ -4,15 +4,10 @@ import (
 	"encoding/binary"
 	"fmt"
 
-	"morph-l2/node/zstd"
-	"morph-l2/tx-submitter/types"
-
 	"github.com/morph-l2/go-ethereum/common"
 	"github.com/morph-l2/go-ethereum/crypto"
-)
 
-var (
-	EmptyVersionedHash = common.HexToHash("0x010657f37554c781402a22917dee2f75def7ab966d7b770905398eba3c444014")
+	"morph-l2/common/blob"
 )
 
 type BatchData struct {
@@ -95,7 +90,7 @@ func (cks *BatchData) DataHashV2() (common.Hash, error) {
 	lastBlockContext := cks.blockContexts[len(cks.blockContexts)-60:]
 
 	// Parse block height
-	height, err := types.HeightFromBlockContextBytes(lastBlockContext)
+	height, err := HeightFromBlockContextBytes(lastBlockContext)
 	if err != nil {
 		return common.Hash{}, fmt.Errorf("failed to parse blockContext: context length=%d, lastBlockContext=%x, err=%w",
 			len(cks.blockContexts), lastBlockContext, err)
@@ -108,8 +103,8 @@ func (cks *BatchData) DataHashV2() (common.Hash, error) {
 func (cks *BatchData) calculateHash(height uint64) common.Hash {
 	// Preallocate memory for efficiency
 	hashData := make([]byte, 8+2+len(cks.l1TxHashes)) // 8 bytes for height, 2 bytes for l1TxNum
-	copy(hashData[:8], types.Uint64ToBigEndianBytes(height))
-	copy(hashData[8:10], types.Uint16ToBigEndianBytes(cks.l1TxNum))
+	copy(hashData[:8], Uint64ToBigEndianBytes(height))
+	copy(hashData[8:10], Uint16ToBigEndianBytes(cks.l1TxNum))
 	copy(hashData[10:], cks.l1TxHashes)
 
 	return crypto.Keccak256Hash(hashData)
@@ -126,16 +121,17 @@ func (cks *BatchData) TxsPayloadV2() []byte {
 
 func (cks *BatchData) BlockNum() uint16 { return cks.blockNum }
 
-func (cks *BatchData) EstimateCompressedSizeWithNewPayload(txPayload []byte) (bool, error) {
+func (cks *BatchData) EstimateCompressedSizeWithNewPayload(txPayload []byte, maxBlobCount int) (bool, error) {
+	limit := maxBlobCount * blob.MaxBlobBytesSize
 	blobBytes := append(cks.txsPayload, txPayload...)
-	if len(blobBytes) <= MaxBlobBytesSize {
+	if len(blobBytes) <= limit {
 		return false, nil
 	}
-	compressed, err := zstd.CompressBatchBytes(blobBytes)
+	compressed, err := blob.CompressBatchBytes(blobBytes)
 	if err != nil {
 		return false, err
 	}
-	return len(compressed) > MaxBlobBytesSize, nil
+	return len(compressed) > limit, nil
 }
 
 func (cks *BatchData) combinePayloads(newBlockContext, newTxPayload []byte) []byte {
@@ -150,15 +146,16 @@ func (cks *BatchData) combinePayloads(newBlockContext, newTxPayload []byte) []by
 
 // WillExceedCompressedSizeLimit checks if the size of the combined block contexts
 // and transaction payloads (after compression) exceeds the maximum allowed size.
-func (cks *BatchData) WillExceedCompressedSizeLimit(newBlockContext, newTxPayload []byte) (bool, error) {
+func (cks *BatchData) WillExceedCompressedSizeLimit(newBlockContext, newTxPayload []byte, maxBlobCount int) (bool, error) {
+	limit := maxBlobCount * blob.MaxBlobBytesSize
 	// Combine the existing and new block contexts and transaction payloads
 	combinedBytes := cks.combinePayloads(newBlockContext, newTxPayload)
-	if len(combinedBytes) <= MaxBlobBytesSize {
+	if len(combinedBytes) <= limit {
 		return false, nil
 	}
-	compressed, err := zstd.CompressBatchBytes(combinedBytes)
+	compressed, err := blob.CompressBatchBytes(combinedBytes)
 	if err != nil {
 		return false, fmt.Errorf("compression failed: %w", err)
 	}
-	return len(compressed) > MaxBlobBytesSize, nil
+	return len(compressed) > limit, nil
 }

tx-submitter/batch/batch_header.go → common/batch/batch_header.go

@@ -16,9 +16,15 @@ type (
 const (
 	expectedLengthV0 = 249
 	expectedLengthV1 = 257
+	// V2 reuses the V1 wire format (257 bytes). The only semantic
+	// difference is that the 32-byte field at offset 57 stores
+	// keccak256(blobhash(0) || ... || blobhash(N-1)) instead of a
+	// single blob versioned hash.
+	expectedLengthV2 = 257
 
 	BatchHeaderVersion0 = 0
 	BatchHeaderVersion1 = 1
+	BatchHeaderVersion2 = 2
 )
 
 var (
@@ -42,6 +48,10 @@ func (b BatchHeaderBytes) validate() error {
 		if len(b) != expectedLengthV1 {
 			return ErrInvalidBatchHeaderLength
 		}
+	case BatchHeaderVersion2:
+		if len(b) != expectedLengthV2 {
+			return ErrInvalidBatchHeaderLength
+		}
 	default:
 		return ErrInvalidBatchHeaderVersion
 	}
@@ -94,10 +104,32 @@ func (b BatchHeaderBytes) DataHash() (common.Hash, error) {
 	return common.BytesToHash(b[25:57]), nil
 }
 
+// BlobVersionedHash returns the EIP-4844 blob versioned hash recorded at
+// offset [57:89]. This is only meaningful for V0/V1 batches, where the field
+// holds the single blob's versioned hash. For V2 batches the same offset
+// holds an aggregated hash; callers must use BlobHashesHash instead.
 func (b BatchHeaderBytes) BlobVersionedHash() (common.Hash, error) {
 	if err := b.validate(); err != nil {
 		return common.Hash{}, err
 	}
+	version, _ := b.Version()
+	if version >= BatchHeaderVersion2 {
+		return common.Hash{}, errors.New("BlobVersionedHash is not available for V2+; use BlobHashesHash")
+	}
+	return common.BytesToHash(b[57:89]), nil
+}
+
+// BlobHashesHash returns the aggregated blob hash recorded at offset [57:89]
+// for V2+ batches, defined as keccak256(blobhash(0) || ... || blobhash(N-1)).
+// V0/V1 batches do not aggregate and will return an error.
+func (b BatchHeaderBytes) BlobHashesHash() (common.Hash, error) {
+	if err := b.validate(); err != nil {
+		return common.Hash{}, err
+	}
+	version, _ := b.Version()
+	if version < BatchHeaderVersion2 {
+		return common.Hash{}, errors.New("BlobHashesHash is only available for V2+; use BlobVersionedHash")
+	}
 	return common.BytesToHash(b[57:89]), nil
 }