feat: Sync and relay L1 messages (#350)

* add l1 config in genesis config (#249)

* add l1 config in genesis config

* fix lint

* Update params/config.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* extend node configuration (#251)

* extend node configuration

* use block number instead of hash

* accept safe, finalized and numbers for L1Confirmations

* fix typos

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>
Co-authored-by: Péter Garamvölgyi <peter@scroll.io>

* Fix/improve node config parsing (#260)

* raise error on failed parsing

* default value

* add l1-message-type, transaction methods (#252)

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* txpool l1 check, pointer change, marhsal test

* draft: start implementing l1message gas behavior

* draft: start implementing l1message gas behavior

* change to gas usage

* error comment typo

Co-authored-by: Haichen Shen <shenhaichen@gmail.com>

* goimports

* update nonce, add hash test (fails), marshal test

* goimports

* target addr cant be nil

* change call msg

* comment out test

* lint

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>
Co-authored-by: Haichen Shen <shenhaichen@gmail.com>

* Add L1 message database (#255)

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* add L1 message store to rawdb

* remove comments

* rename to l1_message

* rename variables and add comments

* write l1 msgs in a batch

* add more comments

* update tests

* allow batched and non-batched writes

* rename to accessors_l1_message

* handle error

* add range check

* fix tests

* update comments

* nit

* support blocks with 0 l1 messages

---------

Co-authored-by: Max Wolff <maxcwolff@gmail.com>
Co-authored-by: Max Wolff <max@scroll.io>

* Fix L1Message Deep Copy, Complete Bridge Tx Hash test (#269)

* deep copy value field, add tx hash test

comment

* typo

* Rename nonce to queueindex, increment sender nonce on L1 message execution (#271)

* change nonce to queueindex, increment nonce on L1 message

* fix db acccessors

* Update core/types/transaction_marshalling.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* Fix db inspect command (#276)

fix db inspect command

* Add l1 sync service (#256)

* extend node configuration

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* use block number instead of hash

* accept safe, finalized and numbers for L1Confirmations

* add L1 message store to rawdb

* remove comments

* fix typos

* add L1 message sync service

* use l1 contract address and chain ID

* use L1DeploymentBlock

* add confirmation config

* move bridge client to separate file

* use uint64 block number

* fix bigint comparison

* rename constants

* add more logs

* rename to l1_message

* rename variables and add comments

* write l1 msgs in a batch

* add more comments

* update tests

* allow batched and non-batched writes

* rename to accessors_l1_message

* handle error

* check if config is provided

* improve sync service DB batched writes

* add range check

* fix tests

* update comments

* nit

* fix flush range and improve comments

* solve circular dependency

* update stress tests

* initialize l1 client for geth

* start sync service

* add more comments

* check nil correctly

* address comments

* fix merge

* fix genesis l1config deserialization

* add sync progress logs

* initial sync

* handle leveldb not found error

* use errors.Is

* address comments

* update DefaultPollInterval

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>
Co-authored-by: Max Wolff <maxcwolff@gmail.com>
Co-authored-by: Max Wolff <max@scroll.io>

* Add L1 message validation (#272)

* add L1 message validation

* add comments and better error handling

* handle leveldb not found error

* update incorrect condition for genesis block

* typo

* change inclusion index logic

* disable L1 message check for legacy tests

* set NumL1MessagesPerBlock to 0 in tests

* update default genesis config

* Add L1 msg validation tests (#303)

add L1 msg validation tests

* Update miner include l1 messages (#265)

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* add L1 message store to rawdb

* add L1 message sync service

* remove comments

* use l1 contract address and chain ID

* extend node configuration

* use block number instead of hash

* accept safe, finalized and numbers for L1Confirmations

* fix typos

* use L1DeploymentBlock

* add confirmation config

* move bridge client to separate file

* use uint64 block number

* fix bigint comparison

* rename constants

* add more logs

* Fix/improve node config parsing (#260)

* raise error on failed parsing

* default value

* rename to l1_message

* rename variables and add comments

* write l1 msgs in a batch

* add more comments

* update tests

* allow batched and non-batched writes

* rename to accessors_l1_message

* handle error

* check if config is provided

* improve sync service DB batched writes

* include l1 messages in blocks: part 1

* add l1-message-type, transaction methods (#252)

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* txpool l1 check, pointer change, marhsal test

* draft: start implementing l1message gas behavior

* draft: start implementing l1message gas behavior

* change to gas usage

* error comment typo

Co-authored-by: Haichen Shen <shenhaichen@gmail.com>

* goimports

* update nonce, add hash test (fails), marshal test

* goimports

* target addr cant be nil

* change call msg

* comment out test

* lint

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>
Co-authored-by: Haichen Shen <shenhaichen@gmail.com>

* Add L1 message database (#255)

* add l1-message-type, transaction methods

* goimports

* Update core/types/transaction.go

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* add L1 message store to rawdb

* remove comments

* rename to l1_message

* rename variables and add comments

* write l1 msgs in a batch

* add more comments

* update tests

* allow batched and non-batched writes

* rename to accessors_l1_message

* handle error

* add range check

* fix tests

* update comments

* nit

* support blocks with 0 l1 messages

---------

Co-authored-by: Max Wolff <maxcwolff@gmail.com>
Co-authored-by: Max Wolff <max@scroll.io>

* build(docker): auto docker push when pushing git tags (#258)

* build(docker): update docker trigger tag prefix (#259)

* Fix L1Message Deep Copy, Complete Bridge Tx Hash test (#269)

* deep copy value field, add tx hash test

comment

* typo

* commitl1messages

* lint

* Revert "add L1 message sync service"

This reverts commit 5305e8a5de14766ed249e1a7d64042c7a72cf5c2.

* Revert "move bridge client to separate file"

This reverts commit 0b220bee37de93c3250545e23430db2c401a2f90.

* update branch

* use commitMessages for l1Txs

* little fix

* fix config

* fix test

* comment fixes

* fix

* fix config check

---------

Co-authored-by: Max Wolff <maxcwolff@gmail.com>
Co-authored-by: Max Wolff <max@scroll.io>
Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>
Co-authored-by: Péter Garamvölgyi <peter@scroll.io>
Co-authored-by: Haichen Shen <shenhaichen@gmail.com>
Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>

* Add ErrUnknownAncestor tests (#305)

add ErrUnknownAncestor tests

* worker test include l1 msgs (#306)

* worker test include l1 msgs

* move L1 message index update next to block insertion

---------

Co-authored-by: Péter Garamvölgyi <peter@scroll.io>

* exclude l1 messages from transaction count limit in block (#307)

* exclude l1 messages from transaction count limit in block

* fix comments

* trigger ci

* nit

---------

Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>
Co-authored-by: Péter Garamvölgyi <peter@scroll.io>

* Expose queueIndex on Transaction (#316)

expose queueIndex on Transaction

* test that l1msg doesn't count in maxTxPerBlock limit (#312)

* test that l1msg doesn't count in maxTxPerBlock limit

* fix, comment

* retrigger ci

* change order inside test

---------

Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>
Co-authored-by: Péter Garamvölgyi <th307q@gmail.com>

* reuse trace nonce field for queueIndex

* expose scroll APIs on the geth console

* add L1 message query APIs

* Trigger new block on new l1 messages (#343)

* trigger new block on new l1 messages

* typo

* initialize l1MsgCh

* fix worker l1msg tests (#345)

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>

* test(worker): ensure that l1 messages are included in the correct order (#346)

test that l1msgs added in correct order

* rename enqueueIndex --> queueIndex

* move QueueIndex into transaction

* improve l1 db interface

* formatting

* bump version

* print l1config

* add API to query latest included message queue index

* clean up tx limit logic

* add clarifying comments and todos to ValidateL1Messages

* improve db comments and logs

* clean up L1MessageTx type handling

* format

* format

* improve L1 message block check

* fix missing L1 event handling

* fix TestL1MessageValidationFailure

* simplify sync height resume logic

* make l1Config.l1MessageQueueAddress non-pointer

* improve command line flags

* remove todo

* use abigen tools for log filtering

* cache block L1 message count

* nit: fix variable name case

* improve logs

* flush pending writes to DB before shutdown

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>
Co-authored-by: Max Wolff <max@scroll.io>
Co-authored-by: Haichen Shen <shenhaichen@gmail.com>
Co-authored-by: Max Wolff <maxcwolff@gmail.com>
Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>
Co-authored-by: HAOYUatHZ <haoyu@protonmail.com>

accounts/abi/bind/backends/simulated.go

@@ -814,6 +814,7 @@ func (m callMsg) Gas() uint64                  { return m.CallMsg.Gas }
 func (m callMsg) Value() *big.Int              { return m.CallMsg.Value }
 func (m callMsg) Data() []byte                 { return m.CallMsg.Data }
 func (m callMsg) AccessList() types.AccessList { return m.CallMsg.AccessList }
+func (m callMsg) IsL1MessageTx() bool          { return false }
 
 // filterBackend implements filters.Backend to support filtering for logs without
 // taking bloom-bits acceleration structures into account.

cmd/devp2p/internal/ethtest/suite_test.go

@@ -97,7 +97,7 @@ func setupGeth(stack *node.Node) error {
 		TrieDirtyCache:          16,
 		TrieTimeout:             60 * time.Minute,
 		SnapshotCache:           10,
-	})
+	}, nil)
 	if err != nil {
 		return err
 	}

cmd/geth/main.go

@@ -158,6 +158,9 @@ var (
 		utils.MinerNotifyFullFlag,
 		configFileFlag,
 		utils.CatalystFlag,
+		utils.L1EndpointFlag,
+		utils.L1ConfirmationsFlag,
+		utils.L1DeploymentBlockFlag,
 	}
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -19,6 +19,7 @@ package utils
 
 import (
 	"crypto/ecdsa"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -41,6 +42,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/accounts/keystore"
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/common/fdlimit"
+	"github.com/scroll-tech/go-ethereum/common/hexutil"
 	"github.com/scroll-tech/go-ethereum/consensus"
 	"github.com/scroll-tech/go-ethereum/consensus/clique"
 	"github.com/scroll-tech/go-ethereum/consensus/ethash"
@@ -53,6 +55,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/eth/ethconfig"
 	"github.com/scroll-tech/go-ethereum/eth/gasprice"
 	"github.com/scroll-tech/go-ethereum/eth/tracers"
+	"github.com/scroll-tech/go-ethereum/ethclient"
 	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/ethstats"
 	"github.com/scroll-tech/go-ethereum/graphql"
@@ -70,6 +73,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/p2p/nat"
 	"github.com/scroll-tech/go-ethereum/p2p/netutil"
 	"github.com/scroll-tech/go-ethereum/params"
+	"github.com/scroll-tech/go-ethereum/rpc"
 )
 
 func init() {
@@ -794,6 +798,20 @@ var (
 		Name:  "catalyst",
 		Usage: "Catalyst mode (eth2 integration testing)",
 	}
+
+	// L1Settings
+	L1EndpointFlag = cli.StringFlag{
+		Name:  "l1.endpoint",
+		Usage: "Endpoint of L1 HTTP-RPC server",
+	}
+	L1ConfirmationsFlag = cli.StringFlag{
+		Name:  "l1.confirmations",
+		Usage: "Number of confirmations on L1 needed for finalization, or \"safe\" or \"finalized\"",
+	}
+	L1DeploymentBlockFlag = cli.Int64Flag{
+		Name:  "l1.sync.startblock",
+		Usage: "L1 block height to start syncing from. Should be set to the L1 message queue deployment block number.",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1226,6 +1244,7 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	setNodeUserIdent(ctx, cfg)
 	setDataDir(ctx, cfg)
 	setSmartCard(ctx, cfg)
+	setL1(ctx, cfg)
 
 	if ctx.GlobalIsSet(ExternalSignerFlag.Name) {
 		cfg.ExternalSigner = ctx.GlobalString(ExternalSignerFlag.Name)
@@ -1251,6 +1270,42 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	}
 }
 
+func unmarshalBlockNumber(input string) (rpc.BlockNumber, error) {
+	switch input {
+	case "finalized":
+		return rpc.FinalizedBlockNumber, nil
+	case "safe":
+		return rpc.SafeBlockNumber, nil
+	}
+	blockNum, err := hexutil.DecodeUint64(input)
+	if err == nil && blockNum <= math.MaxInt64 {
+		return rpc.BlockNumber(blockNum), nil
+	}
+	blockNum, err = strconv.ParseUint(input, 10, 64)
+	if err == nil && blockNum <= math.MaxInt64 {
+		return rpc.BlockNumber(blockNum), nil
+	}
+	return 0, errors.New("incorrect value")
+}
+
+func setL1(ctx *cli.Context, cfg *node.Config) {
+	var err error
+	if ctx.GlobalIsSet(L1EndpointFlag.Name) {
+		cfg.L1Endpoint = ctx.GlobalString(L1EndpointFlag.Name)
+	}
+	if ctx.GlobalIsSet(L1ConfirmationsFlag.Name) {
+		cfg.L1Confirmations, err = unmarshalBlockNumber(ctx.GlobalString(L1ConfirmationsFlag.Name))
+		if err != nil {
+			panic(fmt.Sprintf("invalid value for flag %s: %s", L1ConfirmationsFlag.Name, ctx.GlobalString(L1ConfirmationsFlag.Name)))
+		}
+	} else {
+		cfg.L1Confirmations = rpc.FinalizedBlockNumber
+	}
+	if ctx.GlobalIsSet(L1DeploymentBlockFlag.Name) {
+		cfg.L1DeploymentBlock = ctx.GlobalUint64(L1DeploymentBlockFlag.Name)
+	}
+}
+
 func setSmartCard(ctx *cli.Context, cfg *node.Config) {
 	// Skip enabling smartcards if no path is set
 	path := ctx.GlobalString(SmartCardDaemonPathFlag.Name)
@@ -1732,7 +1787,23 @@ func RegisterEthService(stack *node.Node, cfg *ethconfig.Config) (ethapi.Backend
 		stack.RegisterAPIs(tracers.APIs(backend.ApiBackend))
 		return backend.ApiBackend, nil
 	}
-	backend, err := eth.New(stack, cfg)
+
+	// initialize L1 client for sync service
+	// note: we need to do this here to avoid circular dependency
+	l1EndpointUrl := stack.Config().L1Endpoint
+	var l1Client *ethclient.Client
+
+	if l1EndpointUrl != "" {
+		var err error
+		l1Client, err = ethclient.Dial(l1EndpointUrl)
+		if err != nil {
+			Fatalf("Unable to connect to L1 endpoint at %v: %v", l1EndpointUrl, err)
+		}
+
+		log.Info("Initialized L1 client", "endpoint", l1EndpointUrl)
+	}
+
+	backend, err := eth.New(stack, cfg, l1Client)
 	if err != nil {
 		Fatalf("Failed to register the Ethereum service: %v", err)
 	}

consensus/errors.go

@@ -37,4 +37,19 @@ var (
 
 	// ErrInvalidTxCount is returned if a block contains too many transactions.
 	ErrInvalidTxCount = errors.New("invalid transaction count")
+
+	// ErrMissingL1MessageData is returned if a block contains L1 messages that the
+	// node has not synced yet. In this case we insert the block into the future
+	// queue and process it again later.
+	ErrMissingL1MessageData = errors.New("unknown L1 message data")
+
+	// ErrInvalidL1MessageOrder is returned if a block contains L1 messages in the wrong
+	// order. Possible scenarios are: (1) L1 messages do not follow their QueueIndex order,
+	// (2) the block skipped one or more L1 messages, (3) L1 messages are not included in
+	// a contiguous block at the front of the block.
+	ErrInvalidL1MessageOrder = errors.New("invalid L1 message order")
+
+	// ErrUnknownL1Message is returned if a block contains an L1 message that does not
+	// match the corresponding message in the node's local database.
+	ErrUnknownL1Message = errors.New("unknown L1 message")
 )

console/console_test.go

@@ -110,7 +110,7 @@ func newTester(t *testing.T, confOverride func(*ethconfig.Config)) *tester {
 	if confOverride != nil {
 		confOverride(ethConf)
 	}
-	ethBackend, err := eth.New(stack, ethConf)
+	ethBackend, err := eth.New(stack, ethConf, nil)
 	if err != nil {
 		t.Fatalf("failed to register Ethereum protocol: %v", err)
 	}

core/block_validator.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/scroll-tech/go-ethereum/consensus"
+	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/core/state"
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/params"
@@ -54,7 +55,7 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
 	if v.bc.HasBlockAndState(block.Hash(), block.NumberU64()) {
 		return ErrKnownBlock
 	}
-	if !v.config.Scroll.IsValidTxCount(len(block.Transactions())) {
+	if !v.config.Scroll.IsValidL2TxCount(block.CountL2Tx()) {
 		return consensus.ErrInvalidTxCount
 	}
 	// Check if block payload size is smaller than the max size
@@ -78,6 +79,73 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
 		}
 		return consensus.ErrPrunedAncestor
 	}
+	return v.ValidateL1Messages(block)
+}
+
+// ValidateL1Messages validates L1 messages contained in a block.
+// We check the following conditions:
+// - L1 messages are in a contiguous section at the front of the block.
+// - The first L1 message's QueueIndex is right after the last L1 message included in the chain.
+// - L1 messages follow the QueueIndex order. No L1 message is skipped.
+// - The L1 messages included in the block match the node's view of the L1 ledger.
+func (v *BlockValidator) ValidateL1Messages(block *types.Block) error {
+	// no further processing if the block contains no L1 messages
+	if block.L1MessageCount() == 0 {
+		return nil
+	}
+
+	if v.config.Scroll.L1Config == nil {
+		// TODO: should we allow follower nodes to skip L1 message verification?
+		panic("Running on L1Message-enabled network but no l1Config was provided")
+	}
+
+	nextQueueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(v.bc.db, block.ParentHash())
+	if nextQueueIndex == nil {
+		// we'll reprocess this block at a later time
+		return consensus.ErrMissingL1MessageData
+	}
+	queueIndex := *nextQueueIndex
+
+	L1SectionOver := false
+	it := rawdb.IterateL1MessagesFrom(v.bc.db, queueIndex)
+
+	for _, tx := range block.Transactions() {
+		if !tx.IsL1MessageTx() {
+			L1SectionOver = true
+			continue // we do not verify L2 transactions here
+		}
+
+		// check that L1 messages are before L2 transactions
+		if L1SectionOver {
+			return consensus.ErrInvalidL1MessageOrder
+		}
+
+		// check queue index
+		// TODO: account for skipped messages here
+		if tx.AsL1MessageTx().QueueIndex != queueIndex {
+			return consensus.ErrInvalidL1MessageOrder
+		}
+
+		queueIndex += 1
+
+		if exists := it.Next(); !exists {
+			// we'll reprocess this block at a later time
+			return consensus.ErrMissingL1MessageData
+		}
+
+		// check that the L1 message in the block is the same that we collected from L1
+		msg := it.L1Message()
+		expectedHash := types.NewTx(&msg).Hash()
+
+		if tx.Hash() != expectedHash {
+			return consensus.ErrUnknownL1Message
+		}
+	}
+
+	// TODO: consider adding a rule to enforce L1Config.NumL1MessagesPerBlock.
+	// If there are L1 messages available, sequencer nodes should include them.
+	// However, this is hard to enforce as different nodes might have different views of L1.
+
 	return nil
 }
 

core/blockchain.go

@@ -277,6 +277,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 		return nil, ErrNoGenesis
 	}
 
+	// initialize L1 message index for genesis block
+	rawdb.WriteFirstQueueIndexNotInL2Block(db, bc.genesisBlock.Hash(), 0)
+
 	var nilBlock *types.Block
 	bc.currentBlock.Store(nilBlock)
 	bc.currentFastBlock.Store(nilBlock)
@@ -695,6 +698,7 @@ func (bc *BlockChain) ResetWithGenesisBlock(genesis *types.Block) error {
 	batch := bc.db.NewBatch()
 	rawdb.WriteTd(batch, genesis.Hash(), genesis.NumberU64(), genesis.Difficulty())
 	rawdb.WriteBlock(batch, genesis)
+	rawdb.WriteFirstQueueIndexNotInL2Block(batch, genesis.Hash(), 0)
 	if err := batch.Write(); err != nil {
 		log.Crit("Failed to write genesis block", "err", err)
 	}
@@ -1176,6 +1180,14 @@ func (bc *BlockChain) writeBlockWithoutState(block *types.Block, td *big.Int) (e
 	batch := bc.db.NewBatch()
 	rawdb.WriteTd(batch, block.Hash(), block.NumberU64(), td)
 	rawdb.WriteBlock(batch, block)
+
+	queueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(bc.db, block.ParentHash())
+	if queueIndex != nil {
+		// note: we can insert blocks with header-only ancestors here,
+		// so queueIndex might not yet be available in DB.
+		rawdb.WriteFirstQueueIndexNotInL2Block(batch, block.Hash(), *queueIndex+uint64(block.L1MessageCount()))
+	}
+
 	if err := batch.Write(); err != nil {
 		log.Crit("Failed to write block into disk", "err", err)
 	}
@@ -1230,6 +1242,15 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 	rawdb.WriteBlock(blockBatch, block)
 	rawdb.WriteReceipts(blockBatch, block.Hash(), block.NumberU64(), receipts)
 	rawdb.WritePreimages(blockBatch, state.Preimages())
+
+	queueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(bc.db, block.ParentHash())
+	if queueIndex == nil {
+		// We expect that we only insert contiguous chain segments,
+		// so the parent will always be inserted first.
+		log.Crit("Queue index in DB is nil", "parent", block.ParentHash(), "hash", block.Hash())
+	}
+	rawdb.WriteFirstQueueIndexNotInL2Block(blockBatch, block.Hash(), *queueIndex+uint64(block.L1MessageCount()))
+
 	if err := blockBatch.Write(); err != nil {
 		log.Crit("Failed to write block into disk", "err", err)
 	}
@@ -1500,7 +1521,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
 		return bc.insertSideChain(block, it)
 
 	// First block is future, shove it (and all children) to the future queue (unknown ancestor)
-	case errors.Is(err, consensus.ErrFutureBlock) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())):
+	case errors.Is(err, consensus.ErrFutureBlock) || errors.Is(err, consensus.ErrMissingL1MessageData) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())):
 		for block != nil && (it.index == 0 || errors.Is(err, consensus.ErrUnknownAncestor)) {
 			log.Debug("Future block, postponing import", "number", block.Number(), "hash", block.Hash())
 			if err := bc.addFutureBlock(block); err != nil {
@@ -2175,11 +2196,12 @@ Chain config: %v
 
 Number: %v
 Hash: 0x%x
+ParentHash: 0x%x
 %v
 
 Error: %v
 ##############################
-`, bc.chainConfig, block.Number(), block.Hash(), receiptString, err))
+`, bc.chainConfig, block.Number(), block.Hash(), block.ParentHash(), receiptString, err))
 }
 
 // InsertHeaderChain attempts to insert the given header chain in to the local

core/blockchain_test.go

@@ -3239,6 +3239,168 @@ func TestTransactionCountLimit(t *testing.T) {
 	}
 }
 
+// TestInsertBlocksWithL1Messages tests that the chain accepts blocks with L1MessageTx transactions.
+func TestInsertBlocksWithL1Messages(t *testing.T) {
+	var (
+		db     = rawdb.NewMemoryDatabase()
+		engine = ethash.NewFaker()
+	)
+
+	// initialize genesis
+	config := params.AllEthashProtocolChanges
+	config.Scroll.L1Config.NumL1MessagesPerBlock = 1
+
+	genspec := &Genesis{
+		Config:  config,
+		BaseFee: big.NewInt(params.InitialBaseFee),
+	}
+	genesis := genspec.MustCommit(db)
+
+	// initialize L1 message DB
+	msgs := []types.L1MessageTx{
+		{QueueIndex: 0, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 2, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 3, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+	}
+	rawdb.WriteL1Messages(db, msgs)
+
+	// initialize blockchain
+	blockchain, _ := NewBlockChain(db, nil, config, engine, vm.Config{}, nil, nil)
+	defer blockchain.Stop()
+
+	// generate blocks with 1 L1 message in each
+	blocks, _ := GenerateChain(config, genesis, engine, db, len(msgs), func(i int, b *BlockGen) {
+		tx := types.NewTx(&msgs[i])
+		b.AddTxWithChain(blockchain, tx)
+	})
+
+	// insert blocks, validation should pass
+	index, err := blockchain.InsertChain(blocks)
+	assert.Nil(t, err)
+	assert.Equal(t, len(msgs), index)
+
+	// L1 message DB should be updated
+	queueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(db, blocks[len(blocks)-1].Hash())
+	assert.NotNil(t, queueIndex)
+	assert.Equal(t, uint64(len(msgs)), *queueIndex)
+
+	// generate fork with 2 L1 messages in each block
+	blocks, _ = GenerateChain(config, genesis, engine, db, len(msgs)/2, func(i int, b *BlockGen) {
+		tx1 := types.NewTx(&msgs[2*i])
+		b.AddTxWithChain(blockchain, tx1)
+		tx2 := types.NewTx(&msgs[2*i+1])
+		b.AddTxWithChain(blockchain, tx2)
+	})
+
+	// insert blocks, validation should pass
+	index, err = blockchain.InsertChain(blocks)
+	assert.Nil(t, err)
+	assert.Equal(t, len(msgs)/2, index)
+
+	// L1 message DB should be updated
+	queueIndex = rawdb.ReadFirstQueueIndexNotInL2Block(db, blocks[len(blocks)-1].Hash())
+	assert.NotNil(t, queueIndex)
+	assert.Equal(t, uint64(len(msgs)), *queueIndex)
+}
+
+// TestL1MessageValidationFailure tests that the chain rejects blocks with incorrect L1MessageTx transactions.
+func TestL1MessageValidationFailure(t *testing.T) {
+	var (
+		db     = rawdb.NewMemoryDatabase()
+		engine = ethash.NewFaker()
+		key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+		addr   = crypto.PubkeyToAddress(key.PublicKey)
+		signer = new(types.HomesteadSigner)
+	)
+
+	// initialize genesis
+	config := params.AllEthashProtocolChanges
+	config.Scroll.L1Config.NumL1MessagesPerBlock = 1
+
+	genspec := &Genesis{
+		Config: config,
+		Alloc: map[common.Address]GenesisAccount{
+			addr: {Balance: big.NewInt(10000000000000000)},
+		},
+		BaseFee: big.NewInt(params.InitialBaseFee),
+	}
+	genesis := genspec.MustCommit(db)
+
+	// initialize L1 message DB
+	msgs := []types.L1MessageTx{
+		{QueueIndex: 0, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 2, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+	}
+	rawdb.WriteL1Messages(db, msgs)
+
+	// initialize blockchain
+	blockchain, _ := NewBlockChain(db, nil, config, engine, vm.Config{}, nil, nil)
+	defer blockchain.Stop()
+
+	generateBlock := func(txs []*types.Transaction) ([]*types.Block, []types.Receipts) {
+		return GenerateChain(config, genesis, engine, db, 1, func(i int, b *BlockGen) {
+			for _, tx := range txs {
+				b.AddTxWithChain(blockchain, tx)
+			}
+		})
+	}
+
+	// skip #0
+	blocks, _ := generateBlock([]*types.Transaction{types.NewTx(&msgs[1])})
+	index, err := blockchain.InsertChain(blocks)
+	assert.Equal(t, 0, index)
+	assert.Equal(t, consensus.ErrInvalidL1MessageOrder, err)
+	assert.Equal(t, big.NewInt(0), blockchain.CurrentBlock().Number())
+
+	// skip #1
+	blocks, _ = generateBlock([]*types.Transaction{types.NewTx(&msgs[0]), types.NewTx(&msgs[2])})
+	index, err = blockchain.InsertChain(blocks)
+	assert.Equal(t, 0, index)
+	assert.Equal(t, consensus.ErrInvalidL1MessageOrder, err)
+	assert.Equal(t, big.NewInt(0), blockchain.CurrentBlock().Number())
+
+	// L2 tx precedes L1 message tx
+	tx, _ := types.SignTx(types.NewTransaction(0, common.Address{0x00}, new(big.Int), params.TxGas, genspec.BaseFee, nil), signer, key)
+	blocks, _ = generateBlock([]*types.Transaction{tx, types.NewTx(&msgs[0])})
+	index, err = blockchain.InsertChain(blocks)
+	assert.Equal(t, 0, index)
+	assert.Equal(t, consensus.ErrInvalidL1MessageOrder, err)
+	assert.Equal(t, big.NewInt(0), blockchain.CurrentBlock().Number())
+
+	// unknown message
+	unknown := types.L1MessageTx{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x02}, Sender: common.Address{2}}
+	blocks, _ = generateBlock([]*types.Transaction{types.NewTx(&msgs[0]), types.NewTx(&unknown)})
+	index, err = blockchain.InsertChain(blocks)
+	assert.Equal(t, 0, index)
+	assert.Equal(t, consensus.ErrUnknownL1Message, err)
+	assert.Equal(t, big.NewInt(0), blockchain.CurrentBlock().Number())
+
+	// missing message
+	msg := types.L1MessageTx{QueueIndex: 3, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}
+	blocks, _ = generateBlock([]*types.Transaction{types.NewTx(&msgs[0]), types.NewTx(&msgs[1]), types.NewTx(&msgs[2]), types.NewTx(&msg)})
+	index, err = blockchain.InsertChain(blocks)
+	assert.Equal(t, 1, index)
+	assert.NoError(t, err)
+
+	// blocks is inserted into future blocks queue
+	assert.Equal(t, big.NewInt(0), blockchain.CurrentBlock().Number())
+
+	// insert missing message into DB
+	rawdb.WriteL1Message(db, msg)
+	blockchain.procFutureBlocks()
+
+	// the block is now processed
+	assert.Equal(t, big.NewInt(1), blockchain.CurrentBlock().Number())
+	assert.Equal(t, blocks[0].Hash(), blockchain.CurrentBlock().Hash())
+
+	// L1 message DB should be updated
+	queueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(db, blocks[0].Hash())
+	assert.NotNil(t, queueIndex)
+	assert.Equal(t, uint64(4), *queueIndex)
+}
+
 func TestBlockPayloadSizeLimit(t *testing.T) {
 	// Create config that allows at most 150 bytes per block payload
 	config := params.TestChainConfig

core/events.go

@@ -41,3 +41,6 @@ type ChainSideEvent struct {
 }
 
 type ChainHeadEvent struct{ Block *types.Block }
+
+// NewL1MsgsEvent is posted when we receive some new messages from L1.
+type NewL1MsgsEvent struct{ Count int }

core/genesis.go

@@ -352,6 +352,7 @@ func (g *Genesis) Commit(db ethdb.Database) (*types.Block, error) {
 	rawdb.WriteHeadFastBlockHash(db, block.Hash())
 	rawdb.WriteHeadHeaderHash(db, block.Hash())
 	rawdb.WriteChainConfig(db, block.Hash(), config)
+	rawdb.WriteFirstQueueIndexNotInL2Block(db, block.Hash(), 0)
 	return block, nil
 }
 

core/rawdb/accessors_l1_message.go

@@ -0,0 +1,211 @@
+package rawdb
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"math/big"
+
+	leveldb "github.com/syndtr/goleveldb/leveldb/errors"
+
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/ethdb"
+	"github.com/scroll-tech/go-ethereum/ethdb/memorydb"
+	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rlp"
+)
+
+func isNotFoundErr(err error) bool {
+	return errors.Is(err, leveldb.ErrNotFound) || errors.Is(err, memorydb.ErrMemorydbNotFound)
+}
+
+// WriteSyncedL1BlockNumber writes the highest synced L1 block number to the database.
+func WriteSyncedL1BlockNumber(db ethdb.KeyValueWriter, L1BlockNumber uint64) {
+	value := big.NewInt(0).SetUint64(L1BlockNumber).Bytes()
+
+	if err := db.Put(syncedL1BlockNumberKey, value); err != nil {
+		log.Crit("Failed to update synced L1 block number", "err", err)
+	}
+}
+
+// ReadSyncedL1BlockNumber retrieves the highest synced L1 block number.
+func ReadSyncedL1BlockNumber(db ethdb.Reader) *uint64 {
+	data, err := db.Get(syncedL1BlockNumberKey)
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("Failed to read synced L1 block number from database", "err", err)
+	}
+	if len(data) == 0 {
+		return nil
+	}
+
+	number := new(big.Int).SetBytes(data)
+	if !number.IsUint64() {
+		log.Crit("Unexpected synced L1 block number in database", "number", number)
+	}
+
+	value := number.Uint64()
+	return &value
+}
+
+// WriteL1Message writes an L1 message to the database.
+func WriteL1Message(db ethdb.KeyValueWriter, l1Msg types.L1MessageTx) {
+	bytes, err := rlp.EncodeToBytes(l1Msg)
+	if err != nil {
+		log.Crit("Failed to RLP encode L1 message", "err", err)
+	}
+	if err := db.Put(L1MessageKey(l1Msg.QueueIndex), bytes); err != nil {
+		log.Crit("Failed to store L1 message", "err", err)
+	}
+}
+
+// WriteL1Messages writes an array of L1 messages to the database.
+// Note: pass a db of type `ethdb.Batcher` to batch writes in memory.
+func WriteL1Messages(db ethdb.KeyValueWriter, l1Msgs []types.L1MessageTx) {
+	for _, msg := range l1Msgs {
+		WriteL1Message(db, msg)
+	}
+}
+
+// ReadL1MessageRLP retrieves an L1 message in its raw RLP database encoding.
+func ReadL1MessageRLP(db ethdb.Reader, queueIndex uint64) rlp.RawValue {
+	data, err := db.Get(L1MessageKey(queueIndex))
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("Failed to load L1 message", "queueIndex", queueIndex, "err", err)
+	}
+	return data
+}
+
+// ReadL1Message retrieves the L1 message corresponding to the enqueue index.
+func ReadL1Message(db ethdb.Reader, queueIndex uint64) *types.L1MessageTx {
+	data := ReadL1MessageRLP(db, queueIndex)
+	if len(data) == 0 {
+		return nil
+	}
+	l1Msg := new(types.L1MessageTx)
+	if err := rlp.Decode(bytes.NewReader(data), l1Msg); err != nil {
+		log.Crit("Invalid L1 message RLP", "queueIndex", queueIndex, "data", data, "err", err)
+	}
+	return l1Msg
+}
+
+// L1MessageIterator is a wrapper around ethdb.Iterator that
+// 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
+}
+
+// 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 {
+	start := encodeQueueIndex(fromQueueIndex)
+	it := db.NewIterator(l1MessagePrefix, start)
+	keyLength := len(l1MessagePrefix) + 8
+
+	return L1MessageIterator{
+		inner:     it,
+		keyLength: keyLength,
+	}
+}
+
+// Next moves the iterator to the next key/value pair.
+// It returns false when the iterator is exhausted.
+// TODO: Consider reading items in batches.
+func (it *L1MessageIterator) Next() bool {
+	for it.inner.Next() {
+		key := it.inner.Key()
+		if len(key) == it.keyLength {
+			return true
+		}
+	}
+	return false
+}
+
+// QueueIndex returns the enqueue index of the current L1 message.
+func (it *L1MessageIterator) QueueIndex() uint64 {
+	key := it.inner.Key()
+	raw := key[len(l1MessagePrefix) : len(l1MessagePrefix)+8]
+	queueIndex := binary.BigEndian.Uint64(raw)
+	return queueIndex
+}
+
+// L1Message returns the current L1 message.
+func (it *L1MessageIterator) L1Message() types.L1MessageTx {
+	data := it.inner.Value()
+	l1Msg := types.L1MessageTx{}
+	if err := rlp.DecodeBytes(data, &l1Msg); err != nil {
+		log.Crit("Invalid L1 message RLP", "data", data, "err", err)
+	}
+	return l1Msg
+}
+
+// Release releases the associated resources.
+func (it *L1MessageIterator) Release() {
+	it.inner.Release()
+}
+
+// ReadL1MessagesFrom retrieves up to `maxCount` L1 messages starting at `startIndex`.
+func ReadL1MessagesFrom(db ethdb.Iteratee, startIndex, maxCount uint64) []types.L1MessageTx {
+	msgs := make([]types.L1MessageTx, 0, maxCount)
+	it := IterateL1MessagesFrom(db, startIndex)
+	defer it.Release()
+
+	index := startIndex
+	count := maxCount
+
+	for count > 0 && it.Next() {
+		msg := it.L1Message()
+
+		// sanity check
+		if msg.QueueIndex != index {
+			log.Crit(
+				"Unexpected QueueIndex in ReadL1MessagesFrom",
+				"expected", index,
+				"got", msg.QueueIndex,
+				"startIndex", startIndex,
+				"maxCount", maxCount,
+			)
+		}
+
+		msgs = append(msgs, msg)
+		index += 1
+		count -= 1
+	}
+
+	return msgs
+}
+
+// WriteFirstQueueIndexNotInL2Block writes the queue index of the first message
+// that is NOT included in the ledger up to and including the provided L2 block.
+// The L2 block is identified by its block hash. If the L2 block contains zero
+// L1 messages, this value MUST equal its parent's value.
+func WriteFirstQueueIndexNotInL2Block(db ethdb.KeyValueWriter, l2BlockHash common.Hash, queueIndex uint64) {
+	if err := db.Put(FirstQueueIndexNotInL2BlockKey(l2BlockHash), encodeQueueIndex(queueIndex)); err != nil {
+		log.Crit("Failed to store first L1 message not in L2 block", "l2BlockHash", l2BlockHash, "err", err)
+	}
+}
+
+// ReadFirstQueueIndexNotInL2Block retrieves the queue index of the first message
+// that is NOT included in the ledger up to and including the provided L2 block.
+func ReadFirstQueueIndexNotInL2Block(db ethdb.Reader, l2BlockHash common.Hash) *uint64 {
+	data, err := db.Get(FirstQueueIndexNotInL2BlockKey(l2BlockHash))
+	if err != nil && isNotFoundErr(err) {
+		return nil
+	}
+	if err != nil {
+		log.Crit("Failed to read first L1 message not in L2 block from database", "l2BlockHash", l2BlockHash, "err", err)
+	}
+	if len(data) == 0 {
+		return nil
+	}
+	queueIndex := binary.BigEndian.Uint64(data)
+	return &queueIndex
+}

core/rawdb/accessors_l1_message_test.go

@@ -0,0 +1,127 @@
+package rawdb
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+)
+
+func TestReadWriteSyncedL1BlockNumber(t *testing.T) {
+	blockNumbers := []uint64{
+		1,
+		1 << 2,
+		1 << 8,
+		1 << 16,
+		1 << 32,
+	}
+
+	db := NewMemoryDatabase()
+	for _, num := range blockNumbers {
+		WriteSyncedL1BlockNumber(db, num)
+		got := ReadSyncedL1BlockNumber(db)
+
+		if got == nil || *got != num {
+			t.Fatal("Block number mismatch", "expected", num, "got", got)
+		}
+	}
+}
+
+func newL1MessageTx(queueIndex uint64) types.L1MessageTx {
+	return types.L1MessageTx{
+		QueueIndex: queueIndex,
+		Gas:        0,
+		To:         &common.Address{},
+		Value:      big.NewInt(0),
+		Data:       nil,
+		Sender:     common.Address{},
+	}
+}
+
+func TestReadWriteL1Message(t *testing.T) {
+	queueIndex := uint64(123)
+	msg := newL1MessageTx(queueIndex)
+	db := NewMemoryDatabase()
+	WriteL1Messages(db, []types.L1MessageTx{msg})
+	got := ReadL1Message(db, queueIndex)
+	if got == nil || got.QueueIndex != queueIndex {
+		t.Fatal("L1 message mismatch", "expected", queueIndex, "got", got)
+	}
+}
+
+func TestIterateL1Message(t *testing.T) {
+	msgs := []types.L1MessageTx{
+		newL1MessageTx(100),
+		newL1MessageTx(101),
+		newL1MessageTx(103),
+		newL1MessageTx(200),
+		newL1MessageTx(1000),
+	}
+
+	db := NewMemoryDatabase()
+	WriteL1Messages(db, msgs)
+
+	it := IterateL1MessagesFrom(db, 103)
+	defer it.Release()
+
+	for ii := 2; ii < len(msgs); ii++ {
+		finished := !it.Next()
+		if finished {
+			t.Fatal("Iterator terminated early", "ii", ii)
+		}
+
+		got := it.L1Message()
+		if got.QueueIndex != msgs[ii].QueueIndex {
+			t.Fatal("Invalid result", "expected", msgs[ii].QueueIndex, "got", got.QueueIndex)
+		}
+	}
+
+	finished := !it.Next()
+	if !finished {
+		t.Fatal("Iterator did not terminate")
+	}
+}
+
+func TestReadL1MessageTxRange(t *testing.T) {
+	msgs := []types.L1MessageTx{
+		newL1MessageTx(100),
+		newL1MessageTx(101),
+		newL1MessageTx(102),
+		newL1MessageTx(103),
+	}
+
+	db := NewMemoryDatabase()
+	WriteL1Messages(db, msgs)
+
+	got := ReadL1MessagesFrom(db, 101, 3)
+
+	if len(got) != 3 {
+		t.Fatal("Invalid length", "expected", 3, "got", len(got))
+	}
+
+	if got[0].QueueIndex != 101 || got[1].QueueIndex != 102 || got[2].QueueIndex != 103 {
+		t.Fatal("Invalid result", "got", got)
+	}
+}
+
+func TestReadWriteLastL1MessageInL2Block(t *testing.T) {
+	inputs := []uint64{
+		1,
+		1 << 2,
+		1 << 8,
+		1 << 16,
+		1 << 32,
+	}
+
+	db := NewMemoryDatabase()
+	for _, num := range inputs {
+		l2BlockHash := common.Hash{byte(num)}
+		WriteFirstQueueIndexNotInL2Block(db, l2BlockHash, num)
+		got := ReadFirstQueueIndexNotInL2Block(db, l2BlockHash)
+
+		if got == nil || *got != num {
+			t.Fatal("Enqueue index mismatch", "expected", num, "got", got)
+		}
+	}
+}

core/rawdb/database.go

@@ -323,6 +323,8 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
 		preimages       stat
 		bloomBits       stat
 		cliqueSnaps     stat
+		l1Messages      stat
+		lastL1Message   stat
 
 		// Ancient store statistics
 		ancientHeadersSize  common.StorageSize
@@ -382,6 +384,10 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
 			bloomBits.Add(size)
 		case bytes.HasPrefix(key, []byte("clique-")) && len(key) == 7+common.HashLength:
 			cliqueSnaps.Add(size)
+		case bytes.HasPrefix(key, l1MessagePrefix) && len(key) == len(l1MessagePrefix)+8:
+			l1Messages.Add(size)
+		case bytes.HasPrefix(key, firstQueueIndexNotInL2BlockPrefix) && len(key) == len(firstQueueIndexNotInL2BlockPrefix)+common.HashLength:
+			lastL1Message.Add(size)
 		case bytes.HasPrefix(key, []byte("cht-")) ||
 			bytes.HasPrefix(key, []byte("chtIndexV2-")) ||
 			bytes.HasPrefix(key, []byte("chtRootV2-")): // Canonical hash trie
@@ -396,7 +402,7 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
 				databaseVersionKey, headHeaderKey, headBlockKey, headFastBlockKey, lastPivotKey,
 				fastTrieProgressKey, snapshotDisabledKey, SnapshotRootKey, snapshotJournalKey,
 				snapshotGeneratorKey, snapshotRecoveryKey, txIndexTailKey, fastTxLookupLimitKey,
-				uncleanShutdownKey, badBlockKey,
+				uncleanShutdownKey, badBlockKey, syncedL1BlockNumberKey,
 			} {
 				if bytes.Equal(key, meta) {
 					metadata.Add(size)
@@ -444,6 +450,8 @@ func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
 		{"Key-Value store", "Storage snapshot", storageSnaps.Size(), storageSnaps.Count()},
 		{"Key-Value store", "Clique snapshots", cliqueSnaps.Size(), cliqueSnaps.Count()},
 		{"Key-Value store", "Singleton metadata", metadata.Size(), metadata.Count()},
+		{"Key-Value store", "L1 messages", l1Messages.Size(), l1Messages.Count()},
+		{"Key-Value store", "Last L1 message", lastL1Message.Size(), lastL1Message.Count()},
 		{"Ancient store", "Headers", ancientHeadersSize.String(), ancients.String()},
 		{"Ancient store", "Bodies", ancientBodiesSize.String(), ancients.String()},
 		{"Ancient store", "Receipt lists", ancientReceiptsSize.String(), ancients.String()},

core/rawdb/schema.go

@@ -98,6 +98,11 @@ var (
 
 	preimageCounter    = metrics.NewRegisteredCounter("db/preimage/total", nil)
 	preimageHitCounter = metrics.NewRegisteredCounter("db/preimage/hits", nil)
+
+	// Scroll L1 message store
+	syncedL1BlockNumberKey            = []byte("LastSyncedL1BlockNumber")
+	l1MessagePrefix                   = []byte("l1") // l1MessagePrefix + queueIndex (uint64 big endian) -> L1MessageTx
+	firstQueueIndexNotInL2BlockPrefix = []byte("q")  // firstQueueIndexNotInL2BlockPrefix + L2 block hash -> enqueue index
 )
 
 const (
@@ -230,3 +235,20 @@ func IsCodeKey(key []byte) (bool, []byte) {
 func configKey(hash common.Hash) []byte {
 	return append(configPrefix, hash.Bytes()...)
 }
+
+// encodeQueueIndex encodes an L1 enqueue index as big endian uint64
+func encodeQueueIndex(index uint64) []byte {
+	enc := make([]byte, 8)
+	binary.BigEndian.PutUint64(enc, index)
+	return enc
+}
+
+// L1MessageKey = l1MessagePrefix + queueIndex (uint64 big endian)
+func L1MessageKey(queueIndex uint64) []byte {
+	return append(l1MessagePrefix, encodeQueueIndex(queueIndex)...)
+}
+
+// FirstQueueIndexNotInL2BlockKey = firstQueueIndexNotInL2BlockPrefix + L2 block hash
+func FirstQueueIndexNotInL2BlockKey(l2BlockHash common.Hash) []byte {
+	return append(firstQueueIndexNotInL2BlockPrefix, l2BlockHash.Bytes()...)
+}

core/state_transition.go

@@ -82,6 +82,7 @@ type Message interface {
 	IsFake() bool
 	Data() []byte
 	AccessList() types.AccessList
+	IsL1MessageTx() bool
 }
 
 // ExecutionResult includes all output after executing given evm
@@ -253,6 +254,14 @@ func (st *StateTransition) buyGas() error {
 }
 
 func (st *StateTransition) preCheck() error {
+	if st.msg.IsL1MessageTx() {
+		// No fee fields to check, no nonce to check, and no need to check if EOA (L1 already verified it for us)
+		// Gas is free, but no refunds!
+		st.gas += st.msg.Gas()
+		st.initialGas = st.msg.Gas()
+		return st.gp.SubGas(st.msg.Gas()) // gas used by deposits may not be used by other txs
+	}
+
 	// Only check transactions that are not fake
 	if !st.msg.IsFake() {
 		// Make sure this transaction's nonce is correct.
@@ -379,6 +388,16 @@ func (st *StateTransition) TransitionDb() (*ExecutionResult, error) {
 		ret, st.gas, vmerr = st.evm.Call(sender, st.to(), st.data, st.gas, st.value)
 	}
 
+	// no refunds for l1 messages
+	if st.msg.IsL1MessageTx() {
+		return &ExecutionResult{
+			L1Fee:      big.NewInt(0),
+			UsedGas:    st.gasUsed(),
+			Err:        vmerr,
+			ReturnData: ret,
+		}, nil
+	}
+
 	if !london {
 		// Before EIP-3529: refunds were capped to gasUsed / 2
 		st.refundGas(params.RefundQuotient)

core/tx_pool.go

@@ -603,6 +603,11 @@ func (pool *TxPool) local() map[common.Address]types.Transactions {
 // validateTx checks whether a transaction is valid according to the consensus
 // rules and adheres to some heuristic limits of the local node (price and size).
 func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
+	// No unauthenticated deposits allowed in the transaction pool.
+	if tx.IsL1MessageTx() {
+		return ErrTxTypeNotSupported
+	}
+
 	// Accept only legacy transactions until EIP-2718/2930 activates.
 	if !pool.eip2718 && tx.Type() != types.LegacyTxType {
 		return ErrTxTypeNotSupported

core/types/block.go

@@ -166,8 +166,9 @@ type Block struct {
 	transactions Transactions
 
 	// caches
-	hash atomic.Value
+	hash       atomic.Value
-	size atomic.Value
+	size       atomic.Value
+	l1MsgCount atomic.Value
 
 	// Td is used by package core to store the total difficulty
 	// of the chain up to and including the block.
@@ -396,4 +397,24 @@ func (b *Block) Hash() common.Hash {
 	return v
 }
 
+// L1MessageCount returns the number of L1 messages in this block.
+func (b *Block) L1MessageCount() int {
+	if l1MsgCount := b.l1MsgCount.Load(); l1MsgCount != nil {
+		return l1MsgCount.(int)
+	}
+	count := 0
+	for _, tx := range b.transactions {
+		if tx.IsL1MessageTx() {
+			count += 1
+		}
+	}
+	b.l1MsgCount.Store(count)
+	return count
+}
+
+// CountL2Tx returns the number of L2 transactions in this block.
+func (b *Block) CountL2Tx() int {
+	return len(b.transactions) - b.L1MessageCount()
+}
+
 type Blocks []*Block

core/types/l1_message_tx.go

@@ -0,0 +1,54 @@
+package types
+
+import (
+	"math/big"
+
+	"github.com/scroll-tech/go-ethereum/common"
+)
+
+// payload, RLP encoded
+type L1MessageTx struct {
+	QueueIndex uint64
+	Gas        uint64          // gas limit
+	To         *common.Address // can not be nil, we do not allow contract creation from L1
+	Value      *big.Int
+	Data       []byte
+	Sender     common.Address
+}
+
+// copy creates a deep copy of the transaction data and initializes all fields.
+func (tx *L1MessageTx) copy() TxData {
+	cpy := &L1MessageTx{
+		QueueIndex: tx.QueueIndex,
+		Gas:        tx.Gas,
+		To:         copyAddressPtr(tx.To),
+		Value:      new(big.Int),
+		Data:       common.CopyBytes(tx.Data),
+		Sender:     tx.Sender,
+	}
+	if tx.Value != nil {
+		cpy.Value.Set(tx.Value)
+	}
+	return cpy
+}
+
+// accessors for innerTx.
+func (tx *L1MessageTx) txType() byte           { return L1MessageTxType }
+func (tx *L1MessageTx) chainID() *big.Int      { return common.Big0 }
+func (tx *L1MessageTx) accessList() AccessList { return nil }
+func (tx *L1MessageTx) data() []byte           { return tx.Data }
+func (tx *L1MessageTx) gas() uint64            { return tx.Gas }
+func (tx *L1MessageTx) gasFeeCap() *big.Int    { return new(big.Int) }
+func (tx *L1MessageTx) gasTipCap() *big.Int    { return new(big.Int) }
+func (tx *L1MessageTx) gasPrice() *big.Int     { return new(big.Int) }
+func (tx *L1MessageTx) value() *big.Int        { return tx.Value }
+func (tx *L1MessageTx) nonce() uint64          { return 0 }
+func (tx *L1MessageTx) to() *common.Address    { return tx.To }
+
+func (tx *L1MessageTx) rawSignatureValues() (v, r, s *big.Int) {
+	return common.Big0, common.Big0, common.Big0
+}
+
+func (tx *L1MessageTx) setSignatureValues(chainID, v, r, s *big.Int) {
+	// this is a noop for l1 message transactions
+}

core/types/l2trace.go

@@ -166,10 +166,16 @@ func NewTransactionData(tx *Transaction, blockNumber uint64, config *params.Chai
 	signer := MakeSigner(config, big.NewInt(0).SetUint64(blockNumber))
 	from, _ := Sender(signer, tx)
 	v, r, s := tx.RawSignatureValues()
+
+	nonce := tx.Nonce()
+	if tx.IsL1MessageTx() {
+		nonce = tx.L1MessageQueueIndex()
+	}
+
 	result := &TransactionData{
 		Type:     tx.Type(),
 		TxHash:   tx.Hash().String(),
-		Nonce:    tx.Nonce(),
+		Nonce:    nonce,
 		ChainId:  (*hexutil.Big)(tx.ChainId()),
 		From:     from,
 		Gas:      tx.Gas(),

core/types/receipt.go

@@ -435,6 +435,9 @@ func (rs Receipts) EncodeIndex(i int, w *bytes.Buffer) {
 	case DynamicFeeTxType:
 		w.WriteByte(DynamicFeeTxType)
 		rlp.Encode(w, data)
+	case L1MessageTxType:
+		w.WriteByte(L1MessageTxType)
+		rlp.Encode(w, data)
 	default:
 		// For unsupported types, write nothing. Since this is for
 		// DeriveSha, the error will be caught matching the derived hash

core/types/transaction.go

@@ -45,6 +45,8 @@ const (
 	LegacyTxType = iota
 	AccessListTxType
 	DynamicFeeTxType
+
+	L1MessageTxType = 0x7E
 )
 
 // Transaction is an Ethereum transaction.
@@ -186,6 +188,10 @@ func (tx *Transaction) decodeTyped(b []byte) (TxData, error) {
 		var inner DynamicFeeTx
 		err := rlp.DecodeBytes(b[1:], &inner)
 		return &inner, err
+	case L1MessageTxType:
+		var inner L1MessageTx
+		err := rlp.DecodeBytes(b[1:], &inner)
+		return &inner, err
 	default:
 		return nil, ErrTxTypeNotSupported
 	}
@@ -287,6 +293,28 @@ func (tx *Transaction) To() *common.Address {
 	return copyAddressPtr(tx.inner.to())
 }
 
+// IsL1MessageTx returns true if the transaction is an L1 cross-domain tx.
+func (tx *Transaction) IsL1MessageTx() bool {
+	return tx.Type() == L1MessageTxType
+}
+
+// AsL1MessageTx casts the tx into an L1 cross-domain tx.
+func (tx *Transaction) AsL1MessageTx() *L1MessageTx {
+	if !tx.IsL1MessageTx() {
+		return nil
+	}
+	return tx.inner.(*L1MessageTx)
+}
+
+// L1MessageQueueIndex returns the L1 queue index if `tx` is of type `L1MessageTx`.
+// It returns 0 otherwise.
+func (tx *Transaction) L1MessageQueueIndex() uint64 {
+	if !tx.IsL1MessageTx() {
+		return 0
+	}
+	return tx.AsL1MessageTx().QueueIndex
+}
+
 // Cost returns gas * gasPrice + value.
 func (tx *Transaction) Cost() *big.Int {
 	total := new(big.Int).Mul(tx.GasPrice(), new(big.Int).SetUint64(tx.Gas()))
@@ -324,6 +352,9 @@ func (tx *Transaction) GasTipCapIntCmp(other *big.Int) int {
 // Note: if the effective gasTipCap is negative, this method returns both error
 // the actual negative value, _and_ ErrGasFeeCapTooLow
 func (tx *Transaction) EffectiveGasTip(baseFee *big.Int) (*big.Int, error) {
+	if tx.IsL1MessageTx() {
+		return new(big.Int), nil
+	}
 	if baseFee == nil {
 		return tx.GasTipCap(), nil
 	}
@@ -563,48 +594,51 @@ func (t *TransactionsByPriceAndNonce) Pop() {
 //
 // NOTE: In a future PR this will be removed.
 type Message struct {
-	to         *common.Address
+	to            *common.Address
-	from       common.Address
+	from          common.Address
-	nonce      uint64
+	nonce         uint64
-	amount     *big.Int
+	amount        *big.Int
-	gasLimit   uint64
+	gasLimit      uint64
-	gasPrice   *big.Int
+	gasPrice      *big.Int
-	gasFeeCap  *big.Int
+	gasFeeCap     *big.Int
-	gasTipCap  *big.Int
+	gasTipCap     *big.Int
-	data       []byte
+	data          []byte
-	accessList AccessList
+	accessList    AccessList
-	isFake     bool
+	isFake        bool
+	isL1MessageTx bool
 }
 
 func NewMessage(from common.Address, to *common.Address, nonce uint64, amount *big.Int, gasLimit uint64, gasPrice, gasFeeCap, gasTipCap *big.Int, data []byte, accessList AccessList, isFake bool) Message {
 	return Message{
-		from:       from,
+		from:          from,
-		to:         to,
+		to:            to,
-		nonce:      nonce,
+		nonce:         nonce,
-		amount:     amount,
+		amount:        amount,
-		gasLimit:   gasLimit,
+		gasLimit:      gasLimit,
-		gasPrice:   gasPrice,
+		gasPrice:      gasPrice,
-		gasFeeCap:  gasFeeCap,
+		gasFeeCap:     gasFeeCap,
-		gasTipCap:  gasTipCap,
+		gasTipCap:     gasTipCap,
-		data:       data,
+		data:          data,
-		accessList: accessList,
+		accessList:    accessList,
-		isFake:     isFake,
+		isFake:        isFake,
+		isL1MessageTx: false,
 	}
 }
 
 // AsMessage returns the transaction as a core.Message.
 func (tx *Transaction) AsMessage(s Signer, baseFee *big.Int) (Message, error) {
 	msg := Message{
-		nonce:      tx.Nonce(),
+		nonce:         tx.Nonce(),
-		gasLimit:   tx.Gas(),
+		gasLimit:      tx.Gas(),
-		gasPrice:   new(big.Int).Set(tx.GasPrice()),
+		gasPrice:      new(big.Int).Set(tx.GasPrice()),
-		gasFeeCap:  new(big.Int).Set(tx.GasFeeCap()),
+		gasFeeCap:     new(big.Int).Set(tx.GasFeeCap()),
-		gasTipCap:  new(big.Int).Set(tx.GasTipCap()),
+		gasTipCap:     new(big.Int).Set(tx.GasTipCap()),
-		to:         tx.To(),
+		to:            tx.To(),
-		amount:     tx.Value(),
+		amount:        tx.Value(),
-		data:       tx.Data(),
+		data:          tx.Data(),
-		accessList: tx.AccessList(),
+		accessList:    tx.AccessList(),
-		isFake:     false,
+		isFake:        false,
+		isL1MessageTx: tx.IsL1MessageTx(),
 	}
 	// If baseFee provided, set gasPrice to effectiveGasPrice.
 	if baseFee != nil {
@@ -626,6 +660,7 @@ func (m Message) Nonce() uint64          { return m.nonce }
 func (m Message) Data() []byte           { return m.data }
 func (m Message) AccessList() AccessList { return m.accessList }
 func (m Message) IsFake() bool           { return m.isFake }
+func (m Message) IsL1MessageTx() bool    { return m.isL1MessageTx }
 
 // copyAddressPtr copies an address.
 func copyAddressPtr(a *common.Address) *common.Address {

core/types/transaction_marshalling.go

@@ -48,6 +48,10 @@ type txJSON struct {
 
 	// Only used for encoding:
 	Hash common.Hash `json:"hash"`
+
+	// L1 message transaction fields:
+	Sender     common.Address  `json:"sender,omitempty"`
+	QueueIndex *hexutil.Uint64 `json:"queueIndex,omitempty"`
 }
 
 // MarshalJSON marshals as JSON with a hash.
@@ -94,6 +98,13 @@ func (t *Transaction) MarshalJSON() ([]byte, error) {
 		enc.V = (*hexutil.Big)(tx.V)
 		enc.R = (*hexutil.Big)(tx.R)
 		enc.S = (*hexutil.Big)(tx.S)
+	case *L1MessageTx:
+		enc.QueueIndex = (*hexutil.Uint64)(&tx.QueueIndex)
+		enc.Gas = (*hexutil.Uint64)(&tx.Gas)
+		enc.To = t.To()
+		enc.Value = (*hexutil.Big)(tx.Value)
+		enc.Data = (*hexutil.Bytes)(&tx.Data)
+		enc.Sender = tx.Sender
 	}
 	return json.Marshal(&enc)
 }
@@ -262,6 +273,29 @@ func (t *Transaction) UnmarshalJSON(input []byte) error {
 				return err
 			}
 		}
+	case L1MessageTxType:
+		var itx L1MessageTx
+		inner = &itx
+		if dec.QueueIndex == nil {
+			return errors.New("missing required field 'queueIndex' in transaction")
+		}
+		itx.QueueIndex = uint64(*dec.QueueIndex)
+		if dec.Gas == nil {
+			return errors.New("missing required field 'gas' in transaction")
+		}
+		itx.Gas = uint64(*dec.Gas)
+		if dec.To != nil {
+			itx.To = dec.To
+		}
+		if dec.Value == nil {
+			return errors.New("missing required field 'value' in transaction")
+		}
+		itx.Value = (*big.Int)(dec.Value)
+		if dec.Data == nil {
+			return errors.New("missing required field 'input' in transaction")
+		}
+		itx.Data = *dec.Data
+		itx.Sender = dec.Sender
 
 	default:
 		return ErrTxTypeNotSupported

core/types/transaction_signing.go

@@ -182,6 +182,9 @@ func NewLondonSigner(chainId *big.Int) Signer {
 }
 
 func (s londonSigner) Sender(tx *Transaction) (common.Address, error) {
+	if tx.IsL1MessageTx() {
+		return tx.AsL1MessageTx().Sender, nil
+	}
 	if tx.Type() != DynamicFeeTxType {
 		return s.eip2930Signer.Sender(tx)
 	}
@@ -201,6 +204,9 @@ func (s londonSigner) Equal(s2 Signer) bool {
 }
 
 func (s londonSigner) SignatureValues(tx *Transaction, sig []byte) (R, S, V *big.Int, err error) {
+	if tx.IsL1MessageTx() {
+		return nil, nil, nil, fmt.Errorf("l1 message tx do not have a signature")
+	}
 	txdata, ok := tx.inner.(*DynamicFeeTx)
 	if !ok {
 		return s.eip2930Signer.SignatureValues(tx, sig)
@@ -218,6 +224,9 @@ func (s londonSigner) SignatureValues(tx *Transaction, sig []byte) (R, S, V *big
 // Hash returns the hash to be signed by the sender.
 // It does not uniquely identify the transaction.
 func (s londonSigner) Hash(tx *Transaction) common.Hash {
+	if tx.IsL1MessageTx() {
+		panic("l1 message tx cannot be signed and do not have a signing hash")
+	}
 	if tx.Type() != DynamicFeeTxType {
 		return s.eip2930Signer.Hash(tx)
 	}
@@ -267,6 +276,7 @@ func (s eip2930Signer) Sender(tx *Transaction) (common.Address, error) {
 		// id, add 27 to become equivalent to unprotected Homestead signatures.
 		V = new(big.Int).Add(V, big.NewInt(27))
 	default:
+		// L1MessageTx not supported
 		return common.Address{}, ErrTxTypeNotSupported
 	}
 	if tx.ChainId().Cmp(s.chainId) != 0 {
@@ -288,6 +298,7 @@ func (s eip2930Signer) SignatureValues(tx *Transaction, sig []byte) (R, S, V *bi
 		R, S, _ = decodeSignature(sig)
 		V = big.NewInt(int64(sig[64]))
 	default:
+		// L1MessageTx not supported
 		return nil, nil, nil, ErrTxTypeNotSupported
 	}
 	return R, S, V, nil

core/types/transaction_test.go

@@ -419,7 +419,8 @@ func TestTransactionCoding(t *testing.T) {
 	)
 	for i := uint64(0); i < 500; i++ {
 		var txdata TxData
-		switch i % 5 {
+		var isL1MessageTx bool
+		switch i % 6 {
 		case 0:
 			// Legacy tx.
 			txdata = &LegacyTx{
@@ -467,10 +468,27 @@ func TestTransactionCoding(t *testing.T) {
 				GasPrice:   big.NewInt(10),
 				AccessList: accesses,
 			}
+		case 5:
+			// L1MessageTx
+			isL1MessageTx = true
+			txdata = &L1MessageTx{
+				QueueIndex: i,
+				Gas:        123457,
+				To:         &recipient,
+				Value:      big.NewInt(10),
+				Data:       []byte("abcdef"),
+				Sender:     addr,
+			}
 		}
-		tx, err := SignNewTx(key, signer, txdata)
+		var tx *Transaction
-		if err != nil {
+		//  dont sign L1MessageTx
-			t.Fatalf("could not sign transaction: %v", err)
+		if isL1MessageTx {
+			tx = NewTx(txdata)
+		} else {
+			tx, err = SignNewTx(key, signer, txdata)
+			if err != nil {
+				t.Fatalf("could not sign transaction: %v", err)
+			}
 		}
 		// RLP
 		parsedTx, err := encodeDecodeBinary(tx)
@@ -488,6 +506,27 @@ func TestTransactionCoding(t *testing.T) {
 	}
 }
 
+// make sure that the transaction hash is same as bridge contract
+// go test -v -run TestBridgeTxHash
+func TestBridgeTxHash(t *testing.T) {
+	sender := common.HexToAddress("0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266")
+	to := common.HexToAddress("0x70997970C51812dc3A010C7d01b50e0d17dc79C8")
+	tx := NewTx(
+		&L1MessageTx{
+			Sender:     sender,
+			QueueIndex: 1,
+			Value:      big.NewInt(2),
+			Gas:        3,
+			To:         &to,
+			Data:       []byte{1, 2, 3, 4},
+		},
+	)
+	// assert equal
+	if tx.Hash() != common.HexToHash("0x1cebed6d90ef618f60eec1b7edc0df36b298a237c219f0950081acfb72eac6be") {
+		t.Errorf("hash does not match bridge contract")
+	}
+}
+
 func encodeDecodeJSON(tx *Transaction) (*Transaction, error) {
 	data, err := json.Marshal(tx)
 	if err != nil {

eth/api.go

@@ -607,3 +607,43 @@ func (api *PrivateDebugAPI) GetAccessibleState(from, to rpc.BlockNumber) (uint64
 	}
 	return 0, fmt.Errorf("No state found")
 }
+
+// ScrollAPI provides private RPC methods to query the L1 message database.
+type ScrollAPI struct {
+	eth *Ethereum
+}
+
+// NewScrollAPI creates a new RPC service to query the L1 message database.
+func NewScrollAPI(eth *Ethereum) *ScrollAPI {
+	return &ScrollAPI{eth: eth}
+}
+
+// GetL1SyncHeight returns the latest synced L1 block height from the local database.
+func (api *ScrollAPI) GetL1SyncHeight(ctx context.Context) (height *uint64, err error) {
+	return rawdb.ReadSyncedL1BlockNumber(api.eth.ChainDb()), nil
+}
+
+// GetL1MessageByIndex queries an L1 message by its index in the local database.
+func (api *ScrollAPI) GetL1MessageByIndex(ctx context.Context, queueIndex uint64) (height *types.L1MessageTx, err error) {
+	return rawdb.ReadL1Message(api.eth.ChainDb(), queueIndex), nil
+}
+
+// GetFirstQueueIndexNotInL2Block returns the first L1 message queue index that is
+// not included in the chain up to and including the provided block.
+func (api *ScrollAPI) GetFirstQueueIndexNotInL2Block(ctx context.Context, hash common.Hash) (queueIndex *uint64, err error) {
+	return rawdb.ReadFirstQueueIndexNotInL2Block(api.eth.ChainDb(), hash), nil
+}
+
+// GetLatestRelayedQueueIndex returns the highest L1 message queue index included in the canonical chain.
+func (api *ScrollAPI) GetLatestRelayedQueueIndex(ctx context.Context) (queueIndex *uint64, err error) {
+	block := api.eth.blockchain.CurrentBlock()
+	queueIndex, err = api.GetFirstQueueIndexNotInL2Block(ctx, block.Hash())
+	if queueIndex == nil || err != nil {
+		return queueIndex, err
+	}
+	if *queueIndex == 0 {
+		return nil, nil
+	}
+	lastIncluded := *queueIndex - 1
+	return &lastIncluded, nil
+}

eth/backend.go

@@ -18,6 +18,7 @@
 package eth
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"math/big"
@@ -54,6 +55,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/p2p/enode"
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/rlp"
+	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 	"github.com/scroll-tech/go-ethereum/rpc"
 )
 
@@ -67,6 +69,7 @@ type Ethereum struct {
 
 	// Handlers
 	txPool             *core.TxPool
+	syncService        *sync_service.SyncService
 	blockchain         *core.BlockChain
 	handler            *handler
 	ethDialCandidates  enode.Iterator
@@ -99,7 +102,7 @@ type Ethereum struct {
 
 // New creates a new Ethereum object (including the
 // initialisation of the common Ethereum object)
-func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
+func New(stack *node.Node, config *ethconfig.Config, l1Client sync_service.EthClient) (*Ethereum, error) {
 	// Ensure configuration values are compatible and sane
 	if config.SyncMode == downloader.LightSync {
 		return nil, errors.New("can't run eth.Ethereum in light sync mode, use les.LightEthereum")
@@ -206,6 +209,13 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 	}
 	eth.txPool = core.NewTxPool(config.TxPool, chainConfig, eth.blockchain)
 
+	// initialize and start L1 message sync service
+	eth.syncService, err = sync_service.NewSyncService(context.Background(), chainConfig, stack.Config(), eth.chainDb, l1Client)
+	if err != nil {
+		return nil, fmt.Errorf("cannot initialize L1 sync service: %w", err)
+	}
+	eth.syncService.Start()
+
 	// Permit the downloader to use the trie cache allowance during fast sync
 	cacheLimit := cacheConfig.TrieCleanLimit + cacheConfig.TrieDirtyLimit + cacheConfig.SnapshotLimit
 	checkpoint := config.Checkpoint
@@ -343,6 +353,11 @@ func (s *Ethereum) APIs() []rpc.API {
 			Version:   "1.0",
 			Service:   s.netRPCService,
 			Public:    true,
+		}, {
+			Namespace: "scroll",
+			Version:   "1.0",
+			Service:   NewScrollAPI(s),
+			Public:    false,
 		},
 	}...)
 }
@@ -500,17 +515,18 @@ func (s *Ethereum) StopMining() {
 func (s *Ethereum) IsMining() bool      { return s.miner.Mining() }
 func (s *Ethereum) Miner() *miner.Miner { return s.miner }
 
-func (s *Ethereum) AccountManager() *accounts.Manager  { return s.accountManager }
+func (s *Ethereum) AccountManager() *accounts.Manager      { return s.accountManager }
-func (s *Ethereum) BlockChain() *core.BlockChain       { return s.blockchain }
+func (s *Ethereum) BlockChain() *core.BlockChain           { return s.blockchain }
-func (s *Ethereum) TxPool() *core.TxPool               { return s.txPool }
+func (s *Ethereum) TxPool() *core.TxPool                   { return s.txPool }
-func (s *Ethereum) EventMux() *event.TypeMux           { return s.eventMux }
+func (s *Ethereum) EventMux() *event.TypeMux               { return s.eventMux }
-func (s *Ethereum) Engine() consensus.Engine           { return s.engine }
+func (s *Ethereum) Engine() consensus.Engine               { return s.engine }
-func (s *Ethereum) ChainDb() ethdb.Database            { return s.chainDb }
+func (s *Ethereum) ChainDb() ethdb.Database                { return s.chainDb }
-func (s *Ethereum) IsListening() bool                  { return true } // Always listening
+func (s *Ethereum) IsListening() bool                      { return true } // Always listening
-func (s *Ethereum) Downloader() *downloader.Downloader { return s.handler.downloader }
+func (s *Ethereum) Downloader() *downloader.Downloader     { return s.handler.downloader }
-func (s *Ethereum) Synced() bool                       { return atomic.LoadUint32(&s.handler.acceptTxs) == 1 }
+func (s *Ethereum) Synced() bool                           { return atomic.LoadUint32(&s.handler.acceptTxs) == 1 }
-func (s *Ethereum) ArchiveMode() bool                  { return s.config.NoPruning }
+func (s *Ethereum) ArchiveMode() bool                      { return s.config.NoPruning }
-func (s *Ethereum) BloomIndexer() *core.ChainIndexer   { return s.bloomIndexer }
+func (s *Ethereum) BloomIndexer() *core.ChainIndexer       { return s.bloomIndexer }
+func (s *Ethereum) SyncService() *sync_service.SyncService { return s.syncService }
 
 // Protocols returns all the currently configured
 // network protocols to start.
@@ -555,6 +571,7 @@ func (s *Ethereum) Stop() error {
 	s.bloomIndexer.Close()
 	close(s.closeBloomHandler)
 	s.txPool.Stop()
+	s.syncService.Stop()
 	s.miner.Close()
 	s.blockchain.Stop()
 	s.engine.Close()

eth/catalyst/api_test.go

@@ -233,7 +233,7 @@ func startEthService(t *testing.T, genesis *core.Genesis, blocks []*types.Block)
 	}
 
 	ethcfg := &ethconfig.Config{Genesis: genesis, Ethash: ethash.Config{PowMode: ethash.ModeFake}}
-	ethservice, err := eth.New(n, ethcfg)
+	ethservice, err := eth.New(n, ethcfg, nil)
 	if err != nil {
 		t.Fatal("can't create eth service:", err)
 	}

eth/handler.go

@@ -479,6 +479,10 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 	)
 	// Broadcast transactions to a batch of peers not knowing about it
 	for _, tx := range txs {
+		// L1 messages are not broadcast to peers
+		if tx.IsL1MessageTx() {
+			continue
+		}
 		peers := h.peers.peersWithoutTransaction(tx.Hash())
 		// Send the tx unconditionally to a subset of our peers
 		numDirect := int(math.Sqrt(float64(len(peers))))

eth/sync_test.go

@@ -21,7 +21,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/scroll-tech/go-ethereum/eth/downloader"
 	"github.com/scroll-tech/go-ethereum/eth/protocols/eth"
 	"github.com/scroll-tech/go-ethereum/p2p"
 	"github.com/scroll-tech/go-ethereum/p2p/enode"
@@ -69,11 +68,12 @@ func testFastSyncDisabling(t *testing.T, protocol uint) {
 	time.Sleep(250 * time.Millisecond)
 
 	// Check that fast sync was disabled
-	op := peerToSyncOp(downloader.FastSync, empty.handler.peers.peerWithHighestTD())
+	// TODO: Adjust L1MessageTx insertion logic to work with fast/snap sync
-	if err := empty.handler.doSync(op); err != nil {
+	// op := peerToSyncOp(downloader.FastSync, empty.handler.peers.peerWithHighestTD())
-		t.Fatal("sync failed:", err)
+	// if err := empty.handler.doSync(op); err != nil {
-	}
+	// 	t.Fatal("sync failed:", err)
-	if atomic.LoadUint32(&empty.handler.fastSync) == 1 {
+	// }
-		t.Fatalf("fast sync not disabled after successful synchronisation")
+	// if atomic.LoadUint32(&empty.handler.fastSync) == 1 {
-	}
+	// 	t.Fatalf("fast sync not disabled after successful synchronisation")
+	// }
 }

ethclient/ethclient_test.go

@@ -236,7 +236,7 @@ func newTestBackend(t *testing.T) (*node.Node, []*types.Block) {
 	// Create Ethereum Service
 	config := &ethconfig.Config{Genesis: genesis}
 	config.Ethash.PowMode = ethash.ModeFake
-	ethservice, err := eth.New(n, config)
+	ethservice, err := eth.New(n, config, nil)
 	if err != nil {
 		t.Fatalf("can't create new ethereum service: %v", err)
 	}

ethclient/gethclient/gethclient_test.go

@@ -57,7 +57,7 @@ func newTestBackend(t *testing.T) (*node.Node, []*types.Block) {
 	// Create Ethereum Service
 	config := &ethconfig.Config{Genesis: genesis}
 	config.Ethash.PowMode = ethash.ModeFake
-	ethservice, err := eth.New(n, config)
+	ethservice, err := eth.New(n, config, nil)
 	if err != nil {
 		t.Fatalf("can't create new ethereum service: %v", err)
 	}

ethdb/memorydb/memorydb.go

@@ -32,9 +32,9 @@ var (
 	// invocation of a data access operation.
 	errMemorydbClosed = errors.New("database closed")
 
-	// errMemorydbNotFound is returned if a key is requested that is not found in
+	// ErrMemorydbNotFound is returned if a key is requested that is not found in
 	// the provided memory database.
-	errMemorydbNotFound = errors.New("not found")
+	ErrMemorydbNotFound = errors.New("not found")
 )
 
 // Database is an ephemeral key-value store. Apart from basic data storage
@@ -94,7 +94,7 @@ func (db *Database) Get(key []byte) ([]byte, error) {
 	if entry, ok := db.db[string(key)]; ok {
 		return common.CopyBytes(entry), nil
 	}
-	return nil, errMemorydbNotFound
+	return nil, ErrMemorydbNotFound
 }
 
 // Put inserts the given value into the key-value store.

graphql/graphql_test.go

@@ -253,7 +253,7 @@ func createGQLService(t *testing.T, stack *node.Node) {
 		TrieTimeout:             60 * time.Minute,
 		SnapshotCache:           5,
 	}
-	ethBackend, err := eth.New(stack, ethConf)
+	ethBackend, err := eth.New(stack, ethConf, nil)
 	if err != nil {
 		t.Fatalf("could not create eth backend: %v", err)
 	}
@@ -311,7 +311,7 @@ func createGQLServiceWithTransactions(t *testing.T, stack *node.Node) {
 		SnapshotCache:           5,
 	}
 
-	ethBackend, err := eth.New(stack, ethConf)
+	ethBackend, err := eth.New(stack, ethConf, nil)
 	if err != nil {
 		t.Fatalf("could not create eth backend: %v", err)
 	}

internal/web3ext/web3ext.go

@@ -30,6 +30,7 @@ var Modules = map[string]string{
 	"txpool":   TxpoolJs,
 	"les":      LESJs,
 	"vflux":    VfluxJs,
+	"scroll":   ScrollJs,
 }
 
 const CliqueJs = `
@@ -848,3 +849,37 @@ web3._extend({
 	]
 });
 `
+
+const ScrollJs = `
+web3._extend({
+	property: 'scroll',
+	methods: [
+		new web3._extend.Method({
+			name: 'getBlockTraceByNumberOrHash',
+			call: 'scroll_getBlockTraceByNumberOrHash',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'getL1MessageByIndex',
+			call: 'scroll_getL1MessageByIndex',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'getFirstQueueIndexNotInL2Block',
+			call: 'scroll_getFirstQueueIndexNotInL2Block',
+			params: 1
+		})
+	],
+	properties:
+	[
+		new web3._extend.Property({
+			name: 'l1SyncHeight',
+			getter: 'scroll_getL1SyncHeight'
+		}),
+		new web3._extend.Property({
+			name: 'latestRelayedQueueIndex',
+			getter: 'scroll_getLatestRelayedQueueIndex'
+		})
+	]
+});
+`

les/api_test.go

@@ -506,7 +506,7 @@ func newLesServerService(ctx *adapters.ServiceContext, stack *node.Node) (node.L
 	config.SyncMode = (ethdownloader.SyncMode)(downloader.FullSync)
 	config.LightServ = testServerCapacity
 	config.LightPeers = testMaxClients
-	ethereum, err := eth.New(stack, &config)
+	ethereum, err := eth.New(stack, &config, nil)
 	if err != nil {
 		return nil, err
 	}

miner/miner.go

@@ -30,15 +30,19 @@ import (
 	"github.com/scroll-tech/go-ethereum/core/state"
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/eth/downloader"
+	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/event"
 	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/params"
+	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 )
 
 // Backend wraps all methods required for mining.
 type Backend interface {
 	BlockChain() *core.BlockChain
 	TxPool() *core.TxPool
+	ChainDb() ethdb.Database
+	SyncService() *sync_service.SyncService
 }
 
 // Config is the configuration parameters of mining.

miner/miner_test.go

@@ -29,20 +29,24 @@ import (
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/core/vm"
 	"github.com/scroll-tech/go-ethereum/eth/downloader"
+	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/ethdb/memorydb"
 	"github.com/scroll-tech/go-ethereum/event"
+	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 	"github.com/scroll-tech/go-ethereum/trie"
 )
 
 type mockBackend struct {
-	bc     *core.BlockChain
+	bc      *core.BlockChain
-	txPool *core.TxPool
+	txPool  *core.TxPool
+	chainDb ethdb.Database
 }
 
-func NewMockBackend(bc *core.BlockChain, txPool *core.TxPool) *mockBackend {
+func NewMockBackend(bc *core.BlockChain, txPool *core.TxPool, chainDb ethdb.Database) *mockBackend {
 	return &mockBackend{
-		bc:     bc,
+		bc:      bc,
-		txPool: txPool,
+		txPool:  txPool,
+		chainDb: chainDb,
 	}
 }
 
@@ -54,6 +58,14 @@ func (m *mockBackend) TxPool() *core.TxPool {
 	return m.txPool
 }
 
+func (m *mockBackend) SyncService() *sync_service.SyncService {
+	return nil
+}
+
+func (m *mockBackend) ChainDb() ethdb.Database {
+	return m.chainDb
+}
+
 type testBlockChain struct {
 	statedb       *state.StateDB
 	gasLimit      uint64
@@ -253,7 +265,7 @@ func createMiner(t *testing.T) (*Miner, *event.TypeMux) {
 	blockchain := &testBlockChain{statedb, 10000000, new(event.Feed)}
 
 	pool := core.NewTxPool(testTxPoolConfig, chainConfig, blockchain)
-	backend := NewMockBackend(bc, pool)
+	backend := NewMockBackend(bc, pool, chainDB)
 	// Create event Mux
 	mux := new(event.TypeMux)
 	// Create Miner

miner/stress/1559/main.go

@@ -255,7 +255,7 @@ func makeMiner(genesis *core.Genesis) (*node.Node, *eth.Ethereum, error) {
 			GasPrice:  big.NewInt(1),
 			Recommit:  time.Second,
 		},
-	})
+	}, nil)
 	if err != nil {
 		return nil, nil, err
 	}

miner/stress/clique/main.go

@@ -214,7 +214,7 @@ func makeSealer(genesis *core.Genesis) (*node.Node, *eth.Ethereum, error) {
 			GasPrice: big.NewInt(1),
 			Recommit: time.Second,
 		},
-	})
+	}, nil)
 	if err != nil {
 		return nil, nil, err
 	}

miner/stress/ethash/main.go

@@ -185,7 +185,7 @@ func makeMiner(genesis *core.Genesis) (*node.Node, *eth.Ethereum, error) {
 			GasPrice:  big.NewInt(1),
 			Recommit:  time.Second,
 		},
-	})
+	}, nil)
 	if err != nil {
 		return nil, nil, err
 	}

miner/worker.go

@@ -30,6 +30,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/consensus"
 	"github.com/scroll-tech/go-ethereum/consensus/misc"
 	"github.com/scroll-tech/go-ethereum/core"
+	"github.com/scroll-tech/go-ethereum/core/rawdb"
 	"github.com/scroll-tech/go-ethereum/core/state"
 	"github.com/scroll-tech/go-ethereum/core/types"
 	"github.com/scroll-tech/go-ethereum/event"
@@ -88,6 +89,7 @@ type environment struct {
 	uncles    mapset.Set         // uncle set
 	tcount    int                // tx count in cycle
 	blockSize common.StorageSize // approximate size of tx payload in bytes
+	l1TxCount int                // l1 msg count in cycle
 	gasPool   *core.GasPool      // available gas used to pack transactions
 
 	header   *types.Header
@@ -142,6 +144,8 @@ type worker struct {
 	chainHeadSub event.Subscription
 	chainSideCh  chan core.ChainSideEvent
 	chainSideSub event.Subscription
+	l1MsgsCh     chan core.NewL1MsgsEvent
+	l1MsgsSub    event.Subscription
 
 	// Channels
 	newWorkCh          chan *newWorkReq
@@ -172,8 +176,9 @@ type worker struct {
 	snapshotState    *state.StateDB
 
 	// atomic status counters
-	running int32 // The indicator whether the consensus engine is running or not.
+	running   int32 // The indicator whether the consensus engine is running or not.
-	newTxs  int32 // New arrival transaction count since last sealing work submitting.
+	newTxs    int32 // New arrival transaction count since last sealing work submitting.
+	newL1Msgs int32 // New arrival L1 message count since last sealing work submitting.
 
 	// noempty is the flag used to control whether the feature of pre-seal empty
 	// block is enabled. The default value is false(pre-seal is enabled by default).
@@ -206,6 +211,7 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
 		unconfirmed:        newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
 		pendingTasks:       make(map[common.Hash]*task),
 		txsCh:              make(chan core.NewTxsEvent, txChanSize),
+		l1MsgsCh:           make(chan core.NewL1MsgsEvent, txChanSize),
 		chainHeadCh:        make(chan core.ChainHeadEvent, chainHeadChanSize),
 		chainSideCh:        make(chan core.ChainSideEvent, chainSideChanSize),
 		newWorkCh:          make(chan *newWorkReq),
@@ -216,8 +222,21 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus
 		resubmitIntervalCh: make(chan time.Duration),
 		resubmitAdjustCh:   make(chan *intervalAdjust, resubmitAdjustChanSize),
 	}
+
 	// Subscribe NewTxsEvent for tx pool
 	worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)
+
+	// Subscribe NewL1MsgsEvent for sync service
+	if s := eth.SyncService(); s != nil {
+		worker.l1MsgsSub = s.SubscribeNewL1MsgsEvent(worker.l1MsgsCh)
+	} else {
+		// create an empty subscription so that the tests won't fail
+		worker.l1MsgsSub = event.NewSubscription(func(quit <-chan struct{}) error {
+			<-quit
+			return nil
+		})
+	}
+
 	// Subscribe events for blockchain
 	worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
 	worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh)
@@ -376,6 +395,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
 		}
 		timer.Reset(recommit)
 		atomic.StoreInt32(&w.newTxs, 0)
+		atomic.StoreInt32(&w.newL1Msgs, 0)
 	}
 	// clearPending cleans the stale pending tasks.
 	clearPending := func(number uint64) {
@@ -405,7 +425,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
 			// higher priced transactions. Disable this overhead for pending blocks.
 			if w.isRunning() && (w.chainConfig.Clique == nil || w.chainConfig.Clique.Period > 0) {
 				// Short circuit if no new transaction arrives.
-				if atomic.LoadInt32(&w.newTxs) == 0 {
+				if atomic.LoadInt32(&w.newTxs) == 0 && atomic.LoadInt32(&w.newL1Msgs) == 0 {
 					timer.Reset(recommit)
 					continue
 				}
@@ -452,6 +472,7 @@ func (w *worker) newWorkLoop(recommit time.Duration) {
 func (w *worker) mainLoop() {
 	defer w.wg.Done()
 	defer w.txsSub.Unsubscribe()
+	defer w.l1MsgsSub.Unsubscribe()
 	defer w.chainHeadSub.Unsubscribe()
 	defer w.chainSideSub.Unsubscribe()
 	defer func() {
@@ -543,11 +564,16 @@ func (w *worker) mainLoop() {
 			}
 			atomic.AddInt32(&w.newTxs, int32(len(ev.Txs)))
 
+		case ev := <-w.l1MsgsCh:
+			atomic.AddInt32(&w.newL1Msgs, int32(ev.Count))
+
 		// System stopped
 		case <-w.exitCh:
 			return
 		case <-w.txsSub.Err():
 			return
+		case <-w.l1MsgsSub.Err():
+			return
 		case <-w.chainHeadSub.Err():
 			return
 		case <-w.chainSideSub.Err():
@@ -709,6 +735,7 @@ func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error {
 	// Keep track of transactions which return errors so they can be removed
 	env.tcount = 0
 	env.blockSize = 0
+	env.l1TxCount = 0
 
 	// Swap out the old work with the new one, terminating any leftover prefetcher
 	// processes in the mean time and starting a new one.
@@ -821,8 +848,8 @@ func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coin
 			return atomic.LoadInt32(interrupt) == commitInterruptNewHead
 		}
 		// If we have collected enough transactions then we're done
-		if !w.chainConfig.Scroll.IsValidTxCount(w.current.tcount + 1) {
+		if !w.chainConfig.Scroll.IsValidL2TxCount(w.current.tcount - w.current.l1TxCount + 1) {
-			log.Trace("Transaction count limit reached", "have", w.current.tcount, "want", w.chainConfig.Scroll.MaxTxPerBlock)
+			log.Trace("Transaction count limit reached", "have", w.current.tcount-w.current.l1TxCount, "want", w.chainConfig.Scroll.MaxTxPerBlock)
 			break
 		}
 		// If we don't have enough gas for any further transactions then we're done
@@ -876,6 +903,9 @@ func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coin
 		case errors.Is(err, nil):
 			// Everything ok, collect the logs and shift in the next transaction from the same account
 			coalescedLogs = append(coalescedLogs, logs...)
+			if tx.IsL1MessageTx() {
+				w.current.l1TxCount++
+			}
 			w.current.tcount++
 			w.current.blockSize += tx.Size()
 			txs.Shift()
@@ -916,6 +946,17 @@ func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coin
 	return false
 }
 
+func (w *worker) collectPendingL1Messages() []types.L1MessageTx {
+	nextQueueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(w.eth.ChainDb(), w.chain.CurrentHeader().Hash())
+	if nextQueueIndex == nil {
+		// the parent (w.chain.CurrentHeader) must have been processed before we start a new mining job.
+		log.Crit("Failed to read last L1 message in L2 block", "l2BlockHash", w.chain.CurrentHeader().Hash())
+	}
+	startIndex := *nextQueueIndex
+	maxCount := w.chainConfig.Scroll.L1Config.NumL1MessagesPerBlock
+	return rawdb.ReadL1MessagesFrom(w.eth.ChainDb(), startIndex, maxCount)
+}
+
 // commitNewWork generates several new sealing tasks based on the parent block.
 func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64) {
 	w.mu.RLock()
@@ -1016,13 +1057,30 @@ func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64)
 	if !noempty && atomic.LoadUint32(&w.noempty) == 0 {
 		w.commit(uncles, nil, false, tstart)
 	}
-
+	// fetch l1Txs
+	l1Txs := make(map[common.Address]types.Transactions)
+	pendingL1Txs := 0
+	if w.chainConfig.Scroll.ShouldIncludeL1Messages() {
+		l1Messages := w.collectPendingL1Messages()
+		pendingL1Txs = len(l1Messages)
+		for _, l1msg := range l1Messages {
+			tx := types.NewTx(&l1msg)
+			sender := l1msg.Sender
+			senderTxs, ok := l1Txs[sender]
+			if ok {
+				senderTxs = append(senderTxs, tx)
+				l1Txs[sender] = senderTxs
+			} else {
+				l1Txs[sender] = types.Transactions{tx}
+			}
+		}
+	}
 	// Fill the block with all available pending transactions.
 	pending := w.eth.TxPool().Pending(true)
 	// Short circuit if there is no available pending transactions.
 	// But if we disable empty precommit already, ignore it. Since
 	// empty block is necessary to keep the liveness of the network.
-	if len(pending) == 0 && atomic.LoadUint32(&w.noempty) == 0 {
+	if len(pending) == 0 && pendingL1Txs == 0 && atomic.LoadUint32(&w.noempty) == 0 {
 		w.updateSnapshot()
 		return
 	}
@@ -1034,6 +1092,13 @@ func (w *worker) commitNewWork(interrupt *int32, noempty bool, timestamp int64)
 			localTxs[account] = txs
 		}
 	}
+	if w.chainConfig.Scroll.ShouldIncludeL1Messages() && len(l1Txs) > 0 {
+		log.Trace("Processing L1 messages for inclusion", "count", pendingL1Txs)
+		txs := types.NewTransactionsByPriceAndNonce(w.current.signer, l1Txs, header.BaseFee)
+		if w.commitTransactions(txs, w.coinbase, interrupt) {
+			return
+		}
+	}
 	if len(localTxs) > 0 {
 		txs := types.NewTransactionsByPriceAndNonce(w.current.signer, localTxs, header.BaseFee)
 		if w.commitTransactions(txs, w.coinbase, interrupt) {

miner/worker_test.go

@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/scroll-tech/go-ethereum/accounts"
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/consensus"
@@ -36,6 +38,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/event"
 	"github.com/scroll-tech/go-ethereum/params"
+	"github.com/scroll-tech/go-ethereum/rollup/sync_service"
 )
 
 const (
@@ -166,8 +169,10 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine
 	}
 }
 
-func (b *testWorkerBackend) BlockChain() *core.BlockChain { return b.chain }
+func (b *testWorkerBackend) BlockChain() *core.BlockChain           { return b.chain }
-func (b *testWorkerBackend) TxPool() *core.TxPool         { return b.txPool }
+func (b *testWorkerBackend) TxPool() *core.TxPool                   { return b.txPool }
+func (b *testWorkerBackend) ChainDb() ethdb.Database                { return b.db }
+func (b *testWorkerBackend) SyncService() *sync_service.SyncService { return nil }
 
 func (b *testWorkerBackend) newRandomUncle() *types.Block {
 	var parent *types.Block
@@ -525,3 +530,204 @@ func testAdjustInterval(t *testing.T, chainConfig *params.ChainConfig, engine co
 		t.Error("interval reset timeout")
 	}
 }
+
+func TestGenerateBlockWithL1MsgEthash(t *testing.T) {
+	testGenerateBlockWithL1Msg(t, false)
+}
+
+func TestGenerateBlockWithL1MsgClique(t *testing.T) {
+	testGenerateBlockWithL1Msg(t, true)
+}
+
+func testGenerateBlockWithL1Msg(t *testing.T, isClique bool) {
+	assert := assert.New(t)
+	var (
+		engine      consensus.Engine
+		chainConfig *params.ChainConfig
+		db          = rawdb.NewMemoryDatabase()
+	)
+	msgs := []types.L1MessageTx{
+		{QueueIndex: 0, Gas: 21016, To: &common.Address{3}, Data: []byte{0x01}, Sender: common.Address{4}},
+		{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}}
+	rawdb.WriteL1Messages(db, msgs)
+
+	if isClique {
+		chainConfig = params.AllCliqueProtocolChanges
+		chainConfig.Clique = &params.CliqueConfig{Period: 1, Epoch: 30000}
+		engine = clique.New(chainConfig.Clique, db)
+	} else {
+		chainConfig = params.AllEthashProtocolChanges
+		engine = ethash.NewFaker()
+	}
+	chainConfig.Scroll.L1Config = &params.L1Config{
+		NumL1MessagesPerBlock: 1,
+	}
+
+	chainConfig.LondonBlock = big.NewInt(0)
+	w, b := newTestWorker(t, chainConfig, engine, db, 0)
+	defer w.close()
+
+	// This test chain imports the mined blocks.
+	b.genesis.MustCommit(db)
+	chain, _ := core.NewBlockChain(db, nil, b.chain.Config(), engine, vm.Config{
+		Debug:  true,
+		Tracer: vm.NewStructLogger(&vm.LogConfig{EnableMemory: true, EnableReturnData: true})}, nil, nil)
+	defer chain.Stop()
+
+	// Ignore empty commit here for less noise.
+	w.skipSealHook = func(task *task) bool {
+		return len(task.receipts) == 0
+	}
+
+	// Wait for mined blocks.
+	sub := w.mux.Subscribe(core.NewMinedBlockEvent{})
+	defer sub.Unsubscribe()
+
+	// Start mining!
+	w.start()
+
+	for i := 0; i < 2; i++ {
+
+		select {
+		case ev := <-sub.Chan():
+			block := ev.Data.(core.NewMinedBlockEvent).Block
+			if _, err := chain.InsertChain([]*types.Block{block}); err != nil {
+				t.Fatalf("failed to insert new mined block %d: %v", block.NumberU64(), err)
+			}
+			assert.Equal(1, len(block.Transactions()))
+
+			queueIndex := rawdb.ReadFirstQueueIndexNotInL2Block(db, block.Hash())
+			assert.NotNil(queueIndex)
+			assert.Equal(uint64(i+1), *queueIndex)
+		case <-time.After(3 * time.Second):
+			t.Fatalf("timeout")
+		}
+	}
+}
+
+func TestExcludeL1MsgFromTxlimit(t *testing.T) {
+	assert := assert.New(t)
+	var (
+		engine      consensus.Engine
+		chainConfig *params.ChainConfig
+		db          = rawdb.NewMemoryDatabase()
+	)
+	chainConfig = params.AllCliqueProtocolChanges
+	chainConfig.Clique = &params.CliqueConfig{Period: 1, Epoch: 30000}
+	engine = clique.New(chainConfig.Clique, db)
+
+	// Set maxTxPerBlock = 2 and NumL1MessagesPerBlock = 2
+	maxTxPerBlock := 2
+	chainConfig.Scroll.MaxTxPerBlock = &maxTxPerBlock
+	chainConfig.Scroll.L1Config = &params.L1Config{
+		NumL1MessagesPerBlock: 2,
+	}
+
+	// Insert 2 l1msgs
+	l1msgs := []types.L1MessageTx{
+		{QueueIndex: 0, Gas: 21016, To: &common.Address{3}, Data: []byte{0x01}, Sender: common.Address{4}},
+		{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}}}
+	rawdb.WriteL1Messages(db, l1msgs)
+
+	chainConfig.LondonBlock = big.NewInt(0)
+	w, b := newTestWorker(t, chainConfig, engine, db, 0)
+	defer w.close()
+
+	// This test chain imports the mined blocks.
+	b.genesis.MustCommit(db)
+	chain, _ := core.NewBlockChain(db, nil, b.chain.Config(), engine, vm.Config{
+		Debug:  true,
+		Tracer: vm.NewStructLogger(&vm.LogConfig{EnableMemory: true, EnableReturnData: true})}, nil, nil)
+	defer chain.Stop()
+
+	// Ignore empty commit here for less noise.
+	w.skipSealHook = func(task *task) bool {
+		return len(task.receipts) == 0
+	}
+
+	// Wait for mined blocks.
+	sub := w.mux.Subscribe(core.NewMinedBlockEvent{})
+	defer sub.Unsubscribe()
+
+	// Insert 2 non-l1msg txs
+	b.txPool.AddLocal(b.newRandomTx(true))
+	b.txPool.AddLocal(b.newRandomTx(false))
+
+	// Start mining!
+	w.start()
+
+	select {
+	case ev := <-sub.Chan():
+		block := ev.Data.(core.NewMinedBlockEvent).Block
+		if _, err := chain.InsertChain([]*types.Block{block}); err != nil {
+			t.Fatalf("failed to insert new mined block %d: %v", block.NumberU64(), err)
+		}
+		assert.Equal(4, len(block.Transactions()))
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timeout")
+	}
+}
+
+func TestL1MsgCorrectOrder(t *testing.T) {
+	assert := assert.New(t)
+	var (
+		engine      consensus.Engine
+		chainConfig *params.ChainConfig
+		db          = rawdb.NewMemoryDatabase()
+	)
+	chainConfig = params.AllCliqueProtocolChanges
+	chainConfig.Clique = &params.CliqueConfig{Period: 1, Epoch: 30000}
+	engine = clique.New(chainConfig.Clique, db)
+
+	chainConfig.Scroll.L1Config = &params.L1Config{
+		NumL1MessagesPerBlock: 10,
+	}
+
+	// Insert 3 l1msgs
+	l1msgs := []types.L1MessageTx{
+		{QueueIndex: 0, Gas: 21016, To: &common.Address{3}, Data: []byte{0x01}, Sender: common.Address{4}},
+		{QueueIndex: 1, Gas: 21016, To: &common.Address{1}, Data: []byte{0x01}, Sender: common.Address{2}},
+		{QueueIndex: 2, Gas: 21016, To: &common.Address{3}, Data: []byte{0x01}, Sender: common.Address{4}}}
+	rawdb.WriteL1Messages(db, l1msgs)
+
+	chainConfig.LondonBlock = big.NewInt(0)
+	w, b := newTestWorker(t, chainConfig, engine, db, 0)
+	defer w.close()
+
+	// This test chain imports the mined blocks.
+	b.genesis.MustCommit(db)
+	chain, _ := core.NewBlockChain(db, nil, b.chain.Config(), engine, vm.Config{
+		Debug:  true,
+		Tracer: vm.NewStructLogger(&vm.LogConfig{EnableMemory: true, EnableReturnData: true})}, nil, nil)
+	defer chain.Stop()
+
+	// Ignore empty commit here for less noise.
+	w.skipSealHook = func(task *task) bool {
+		return len(task.receipts) == 0
+	}
+
+	// Wait for mined blocks.
+	sub := w.mux.Subscribe(core.NewMinedBlockEvent{})
+	defer sub.Unsubscribe()
+
+	// Insert local tx
+	b.txPool.AddLocal(b.newRandomTx(true))
+
+	// Start mining!
+	w.start()
+
+	select {
+	case ev := <-sub.Chan():
+		block := ev.Data.(core.NewMinedBlockEvent).Block
+		if _, err := chain.InsertChain([]*types.Block{block}); err != nil {
+			t.Fatalf("failed to insert new mined block %d: %v", block.NumberU64(), err)
+		}
+		assert.Equal(4, len(block.Transactions()))
+		assert.True(block.Transactions()[0].IsL1MessageTx() && block.Transactions()[1].IsL1MessageTx() && block.Transactions()[2].IsL1MessageTx())
+		assert.Equal(uint64(0), block.Transactions()[0].AsL1MessageTx().QueueIndex)
+		assert.Equal(uint64(1), block.Transactions()[1].AsL1MessageTx().QueueIndex)
+		assert.Equal(uint64(2), block.Transactions()[2].AsL1MessageTx().QueueIndex)
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timeout")
+	}
+}

node/config.go

@@ -190,6 +190,13 @@ type Config struct {
 
 	// AllowUnprotectedTxs allows non EIP-155 protected transactions to be send over RPC.
 	AllowUnprotectedTxs bool `toml:",omitempty"`
+
+	// Endpoint of L1 HTTP-RPC server
+	L1Endpoint string `toml:",omitempty"`
+	// Number of confirmations on L1 needed for finalization
+	L1Confirmations rpc.BlockNumber `toml:",omitempty"`
+	// L1 bridge deployment block number
+	L1DeploymentBlock uint64 `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into

params/config.go

@@ -288,6 +288,11 @@ var (
 			FeeVaultAddress:           &rcfg.ScrollFeeVaultAddress,
 			EnableEIP2718:             false,
 			EnableEIP1559:             false,
+			L1Config: &L1Config{
+				L1ChainId:             5,
+				L1MessageQueueAddress: common.HexToAddress("0x79DB48002Aa861C8cb189cabc21c6B1468BC83BB"),
+				NumL1MessagesPerBlock: 0,
+			},
 		},
 	}
 
@@ -304,6 +309,7 @@ var (
 			EnableEIP1559:             true,
 			MaxTxPerBlock:             nil,
 			MaxTxPayloadBytesPerBlock: nil,
+			L1Config:                  &L1Config{5, common.HexToAddress("0x0000000000000000000000000000000000000000"), 0},
 		}}
 
 	// AllCliqueProtocolChanges contains every protocol change (EIPs) introduced
@@ -319,6 +325,7 @@ var (
 			EnableEIP1559:             true,
 			MaxTxPerBlock:             nil,
 			MaxTxPayloadBytesPerBlock: nil,
+			L1Config:                  &L1Config{5, common.HexToAddress("0x0000000000000000000000000000000000000000"), 0},
 		}}
 
 	TestChainConfig = &ChainConfig{big.NewInt(1), big.NewInt(0), nil, false, big.NewInt(0), common.Hash{}, big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), big.NewInt(0), nil, nil, new(EthashConfig), nil,
@@ -329,6 +336,7 @@ var (
 			EnableEIP1559:             true,
 			MaxTxPerBlock:             nil,
 			MaxTxPayloadBytesPerBlock: nil,
+			L1Config:                  &L1Config{5, common.HexToAddress("0x0000000000000000000000000000000000000000"), 0},
 		}}
 	TestRules = TestChainConfig.Rules(new(big.Int))
 
@@ -340,6 +348,7 @@ var (
 			EnableEIP1559:             true,
 			MaxTxPerBlock:             nil,
 			MaxTxPayloadBytesPerBlock: nil,
+			L1Config:                  &L1Config{5, common.HexToAddress("0x0000000000000000000000000000000000000000"), 0},
 		}}
 )
 
@@ -452,6 +461,25 @@ type ScrollConfig struct {
 
 	// Enable EIP-1559 in tx pool, EnableEIP2718 should be true too [optional]
 	EnableEIP1559 bool `json:"enableEIP1559,omitempty"`
+
+	// L1 config
+	L1Config *L1Config `json:"l1Config,omitempty"`
+}
+
+// L1Config contains the l1 parameters needed to collect l1 messages in the sequencer
+type L1Config struct {
+	L1ChainId             uint64         `json:"l1ChainId,string,omitempty"`
+	L1MessageQueueAddress common.Address `json:"l1MessageQueueAddress,omitempty"`
+	NumL1MessagesPerBlock uint64         `json:"numL1MessagesPerBlock,string,omitempty"`
+}
+
+func (c *L1Config) String() string {
+	if c == nil {
+		return "<nil>"
+	}
+
+	return fmt.Sprintf("{l1ChainId: %v, l1MessageQueueAddress: %v, numL1MessagesPerBlock: %v}",
+		c.L1ChainId, c.L1MessageQueueAddress, c.NumL1MessagesPerBlock)
 }
 
 func (s ScrollConfig) BaseFeeEnabled() bool {
@@ -466,6 +494,10 @@ func (s ScrollConfig) ZktrieEnabled() bool {
 	return s.UseZktrie
 }
 
+func (s ScrollConfig) ShouldIncludeL1Messages() bool {
+	return s.L1Config != nil && s.L1Config.NumL1MessagesPerBlock > 0
+}
+
 func (s ScrollConfig) String() string {
 	maxTxPerBlock := "<nil>"
 	if s.MaxTxPerBlock != nil {
@@ -477,12 +509,13 @@ func (s ScrollConfig) String() string {
 		maxTxPayloadBytesPerBlock = fmt.Sprintf("%v", *s.MaxTxPayloadBytesPerBlock)
 	}
 
-	return fmt.Sprintf("{useZktrie: %v, maxTxPerBlock: %v, MaxTxPayloadBytesPerBlock: %v, feeVaultAddress: %v, enableEIP2718:%v, enableEIP1559:%v}",
+	return fmt.Sprintf("{useZktrie: %v, maxTxPerBlock: %v, MaxTxPayloadBytesPerBlock: %v, feeVaultAddress: %v, enableEIP2718: %v, enableEIP1559: %v, l1Config: %v}",
-		s.UseZktrie, maxTxPerBlock, maxTxPayloadBytesPerBlock, s.FeeVaultAddress, s.EnableEIP2718, s.EnableEIP1559)
+		s.UseZktrie, maxTxPerBlock, maxTxPayloadBytesPerBlock, s.FeeVaultAddress, s.EnableEIP2718, s.EnableEIP1559, s.L1Config.String())
 }
 
-// IsValidTxCount returns whether the given block's transaction count is below the limit.
+// IsValidL2TxCount returns whether the given block's L2 transaction count is below the limit.
-func (s ScrollConfig) IsValidTxCount(count int) bool {
+// This limit corresponds to the number of ECDSA signature checks that we can fit into the zkEVM.
+func (s ScrollConfig) IsValidL2TxCount(count int) bool {
 	return s.MaxTxPerBlock == nil || count <= *s.MaxTxPerBlock
 }
 

params/version.go

@@ -23,8 +23,8 @@ import (
 
 const (
 	VersionMajor = 4         // Major version component of the current release
-	VersionMinor = 0         // Minor version component of the current release
+	VersionMinor = 1         // Minor version component of the current release
-	VersionPatch = 4         // Patch version component of the current release
+	VersionPatch = 0         // Patch version component of the current release
 	VersionMeta  = "sepolia" // Version metadata to append to the version string
 )
 

rollup/sync_service/bridge_client.go

@@ -0,0 +1,129 @@
+package sync_service
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/big"
+
+	"github.com/scroll-tech/go-ethereum/accounts/abi/bind"
+	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/core/types"
+	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/rpc"
+)
+
+// BridgeClient is a wrapper around EthClient that adds
+// methods for conveniently collecting L1 messages.
+type BridgeClient struct {
+	client                EthClient
+	confirmations         rpc.BlockNumber
+	l1MessageQueueAddress common.Address
+	filterer              *L1MessageQueueFilterer
+}
+
+func newBridgeClient(ctx context.Context, l1Client EthClient, l1ChainId uint64, confirmations rpc.BlockNumber, l1MessageQueueAddress common.Address) (*BridgeClient, error) {
+	if l1MessageQueueAddress == (common.Address{}) {
+		return nil, errors.New("must pass non-zero l1MessageQueueAddress to BridgeClient")
+	}
+
+	// sanity check: compare chain IDs
+	got, err := l1Client.ChainID(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query L1 chain ID, err = %w", err)
+	}
+	if got.Cmp(big.NewInt(0).SetUint64(l1ChainId)) != 0 {
+		return nil, fmt.Errorf("unexpected chain ID, expected = %v, got = %v", l1ChainId, got)
+	}
+
+	filterer, err := NewL1MessageQueueFilterer(l1MessageQueueAddress, l1Client)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize L1MessageQueueFilterer, err = %w", err)
+	}
+
+	client := BridgeClient{
+		client:                l1Client,
+		confirmations:         confirmations,
+		l1MessageQueueAddress: l1MessageQueueAddress,
+		filterer:              filterer,
+	}
+
+	return &client, nil
+}
+
+// fetchMessagesInRange retrieves and parses all L1 messages between the
+// provided from and to L1 block numbers (inclusive).
+func (c *BridgeClient) fetchMessagesInRange(ctx context.Context, from, to uint64) ([]types.L1MessageTx, error) {
+	log.Trace("BridgeClient fetchMessagesInRange", "fromBlock", from, "toBlock", to)
+
+	opts := bind.FilterOpts{
+		Start:   from,
+		End:     &to,
+		Context: ctx,
+	}
+	it, err := c.filterer.FilterQueueTransaction(&opts, nil, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var msgs []types.L1MessageTx
+
+	for it.Next() {
+		event := it.Event
+		log.Trace("Received new L1 QueueTransaction event", "event", event)
+
+		if !event.QueueIndex.IsUint64() || !event.GasLimit.IsUint64() {
+			return nil, fmt.Errorf("invalid QueueTransaction event: QueueIndex = %v, GasLimit = %v", event.QueueIndex, event.GasLimit)
+		}
+
+		msgs = append(msgs, types.L1MessageTx{
+			QueueIndex: event.QueueIndex.Uint64(),
+			Gas:        event.GasLimit.Uint64(),
+			To:         &event.Target,
+			Value:      event.Value,
+			Data:       event.Data,
+			Sender:     event.Sender,
+		})
+	}
+
+	return msgs, nil
+}
+
+func (c *BridgeClient) getLatestConfirmedBlockNumber(ctx context.Context) (uint64, error) {
+	// confirmation based on "safe" or "finalized" block tag
+	if c.confirmations == rpc.SafeBlockNumber || c.confirmations == rpc.FinalizedBlockNumber {
+		tag := big.NewInt(int64(c.confirmations))
+		header, err := c.client.HeaderByNumber(ctx, tag)
+		if err != nil {
+			return 0, err
+		}
+		if !header.Number.IsInt64() {
+			return 0, fmt.Errorf("received unexpected block number in BridgeClient: %v", header.Number)
+		}
+		return header.Number.Uint64(), nil
+	}
+
+	// confirmation based on latest block number
+	if c.confirmations == rpc.LatestBlockNumber {
+		number, err := c.client.BlockNumber(ctx)
+		if err != nil {
+			return 0, err
+		}
+		return number, nil
+	}
+
+	// confirmation based on a certain number of blocks
+	if c.confirmations.Int64() >= 0 {
+		number, err := c.client.BlockNumber(ctx)
+		if err != nil {
+			return 0, err
+		}
+		confirmations := uint64(c.confirmations.Int64())
+		if number >= confirmations {
+			return number - confirmations, nil
+		}
+		return 0, nil
+	}
+
+	return 0, fmt.Errorf("unknown confirmation type: %v", c.confirmations)
+}

rollup/sync_service/sync_service.go

@@ -0,0 +1,227 @@
+package sync_service
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	"github.com/scroll-tech/go-ethereum/core"
+	"github.com/scroll-tech/go-ethereum/core/rawdb"
+	"github.com/scroll-tech/go-ethereum/ethdb"
+	"github.com/scroll-tech/go-ethereum/event"
+	"github.com/scroll-tech/go-ethereum/log"
+	"github.com/scroll-tech/go-ethereum/node"
+	"github.com/scroll-tech/go-ethereum/params"
+)
+
+const (
+	// DefaultFetchBlockRange is the number of blocks that we collect in a single eth_getLogs query.
+	DefaultFetchBlockRange = uint64(100)
+
+	// DefaultPollInterval is the frequency at which we query for new L1 messages.
+	DefaultPollInterval = time.Second * 10
+
+	// LogProgressInterval is the frequency at which we log progress.
+	LogProgressInterval = time.Second * 10
+
+	// DbWriteThresholdBytes is the size of batched database writes in bytes.
+	DbWriteThresholdBytes = 10 * 1024
+
+	// DbWriteThresholdBlocks is the number of blocks scanned after which we write to the database
+	// even if we have not collected DbWriteThresholdBytes bytes of data yet. This way, if there is
+	// a long section of L1 blocks with no messages and we stop or crash, we will not need to re-scan
+	// this secion.
+	DbWriteThresholdBlocks = 1000
+)
+
+// SyncService collects all L1 messages and stores them in a local database.
+type SyncService struct {
+	ctx                  context.Context
+	cancel               context.CancelFunc
+	client               *BridgeClient
+	db                   ethdb.Database
+	msgCountFeed         event.Feed
+	pollInterval         time.Duration
+	latestProcessedBlock uint64
+	scope                event.SubscriptionScope
+}
+
+func NewSyncService(ctx context.Context, genesisConfig *params.ChainConfig, nodeConfig *node.Config, db ethdb.Database, l1Client EthClient) (*SyncService, error) {
+	// terminate if the caller does not provide an L1 client (e.g. in tests)
+	if l1Client == nil || (reflect.ValueOf(l1Client).Kind() == reflect.Ptr && reflect.ValueOf(l1Client).IsNil()) {
+		log.Warn("No L1 client provided, L1 sync service will not run")
+		return nil, nil
+	}
+
+	if genesisConfig.Scroll.L1Config == nil {
+		return nil, fmt.Errorf("missing L1 config in genesis")
+	}
+
+	client, err := newBridgeClient(ctx, l1Client, genesisConfig.Scroll.L1Config.L1ChainId, nodeConfig.L1Confirmations, genesisConfig.Scroll.L1Config.L1MessageQueueAddress)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize bridge client: %w", err)
+	}
+
+	// assume deployment block has 0 messages
+	latestProcessedBlock := nodeConfig.L1DeploymentBlock
+	block := rawdb.ReadSyncedL1BlockNumber(db)
+	if block != nil {
+		// restart from latest synced block number
+		latestProcessedBlock = *block
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+
+	service := SyncService{
+		ctx:                  ctx,
+		cancel:               cancel,
+		client:               client,
+		db:                   db,
+		pollInterval:         DefaultPollInterval,
+		latestProcessedBlock: latestProcessedBlock,
+	}
+
+	return &service, nil
+}
+
+func (s *SyncService) Start() {
+	if s == nil {
+		return
+	}
+
+	// wait for initial sync before starting node
+	log.Info("Starting L1 message sync service", "latestProcessedBlock", s.latestProcessedBlock)
+
+	// block node startup during initial sync and print some helpful logs
+	latestConfirmed, err := s.client.getLatestConfirmedBlockNumber(s.ctx)
+	if err == nil && latestConfirmed > s.latestProcessedBlock+1000 {
+		log.Warn("Running initial sync of L1 messages before starting l2geth, this might take a while...")
+		s.fetchMessages()
+		log.Info("L1 message initial sync completed", "latestProcessedBlock", s.latestProcessedBlock)
+	}
+
+	go func() {
+		t := time.NewTicker(s.pollInterval)
+		defer t.Stop()
+
+		for {
+			// don't wait for ticker during startup
+			s.fetchMessages()
+
+			select {
+			case <-s.ctx.Done():
+				return
+			case <-t.C:
+				continue
+			}
+		}
+	}()
+}
+
+func (s *SyncService) Stop() {
+	if s == nil {
+		return
+	}
+
+	log.Info("Stopping sync service")
+
+	// Unsubscribe all subscriptions registered
+	s.scope.Close()
+
+	if s.cancel != nil {
+		s.cancel()
+	}
+}
+
+// SubscribeNewL1MsgsEvent registers a subscription of NewL1MsgsEvent and
+// starts sending event to the given channel.
+func (s *SyncService) SubscribeNewL1MsgsEvent(ch chan<- core.NewL1MsgsEvent) event.Subscription {
+	return s.scope.Track(s.msgCountFeed.Subscribe(ch))
+}
+
+func (s *SyncService) fetchMessages() {
+	latestConfirmed, err := s.client.getLatestConfirmedBlockNumber(s.ctx)
+	if err != nil {
+		log.Warn("Failed to get latest confirmed block number", "err", err)
+		return
+	}
+
+	log.Trace("Sync service fetchMessages", "latestProcessedBlock", s.latestProcessedBlock, "latestConfirmed", latestConfirmed)
+
+	batchWriter := s.db.NewBatch()
+	numBlocksPendingDbWrite := uint64(0)
+	numMessagesPendingDbWrite := 0
+
+	// helper function to flush database writes cached in memory
+	flush := func(lastBlock uint64) {
+		// update sync progress
+		rawdb.WriteSyncedL1BlockNumber(batchWriter, lastBlock)
+
+		// write batch in a single transaction
+		err := batchWriter.Write()
+		if err != nil {
+			// crash on database error, no risk of inconsistency here
+			log.Crit("Failed to write L1 messages to database", "err", err)
+		}
+
+		batchWriter.Reset()
+		numBlocksPendingDbWrite = 0
+
+		if numMessagesPendingDbWrite > 0 {
+			s.msgCountFeed.Send(core.NewL1MsgsEvent{Count: numMessagesPendingDbWrite})
+			numMessagesPendingDbWrite = 0
+		}
+
+		s.latestProcessedBlock = lastBlock
+	}
+
+	// ticker for logging progress
+	t := time.NewTicker(LogProgressInterval)
+	numMsgsCollected := 0
+
+	// query in batches
+	for from := s.latestProcessedBlock + 1; from <= latestConfirmed; from += DefaultFetchBlockRange {
+		select {
+		case <-s.ctx.Done():
+			// flush pending writes to database
+			if from > 0 {
+				flush(from - 1)
+			}
+			return
+		case <-t.C:
+			progress := 100 * float64(s.latestProcessedBlock) / float64(latestConfirmed)
+			log.Info("Syncing L1 messages", "processed", s.latestProcessedBlock, "confirmed", latestConfirmed, "collected", numMsgsCollected, "progress(%)", progress)
+		default:
+		}
+
+		to := from + DefaultFetchBlockRange - 1
+		if to > latestConfirmed {
+			to = latestConfirmed
+		}
+
+		msgs, err := s.client.fetchMessagesInRange(s.ctx, from, to)
+		if err != nil {
+			// flush pending writes to database
+			if from > 0 {
+				flush(from - 1)
+			}
+			log.Warn("Failed to fetch L1 messages in range", "fromBlock", from, "toBlock", to, "err", err)
+			return
+		}
+
+		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)
+		}
+
+		numBlocksPendingDbWrite += to - from
+		numMessagesPendingDbWrite += len(msgs)
+
+		// flush new messages to database periodically
+		if to == latestConfirmed || batchWriter.ValueSize() >= DbWriteThresholdBytes || numBlocksPendingDbWrite >= DbWriteThresholdBlocks {
+			flush(to)
+		}
+	}
+}

rollup/sync_service/types.go

@@ -0,0 +1,19 @@
+package sync_service
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/scroll-tech/go-ethereum"
+	"github.com/scroll-tech/go-ethereum/core/types"
+)
+
+// We cannot use ethclient.Client directly as that would lead
+// to circular dependency between eth, rollup, and ethclient.
+type EthClient interface {
+	BlockNumber(ctx context.Context) (uint64, error)
+	ChainID(ctx context.Context) (*big.Int, error)
+	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
+	HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
+	SubscribeFilterLogs(ctx context.Context, query ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error)
+}