interop: experimental sequencer executing message check support

cmd/geth/main.go

@@ -159,6 +159,7 @@ var (
 		utils.RollupSequencerTxConditionalRateLimitFlag,
 		utils.RollupHistoricalRPCFlag,
 		utils.RollupHistoricalRPCTimeoutFlag,
+		utils.RollupInteropRPCFlag,
 		utils.RollupDisableTxPoolGossipFlag,
 		utils.RollupComputePendingBlock,
 		utils.RollupHaltOnIncompatibleProtocolVersionFlag,

cmd/utils/flags.go

@@ -949,6 +949,12 @@ var (
 		Category: flags.RollupCategory,
 	}
 
+	RollupInteropRPCFlag = &cli.StringFlag{
+		Name:     "rollup.interoprpc",
+		Usage:    "RPC endpoint for interop message verification (experimental).",
+		Category: flags.RollupCategory,
+	}
+
 	RollupDisableTxPoolGossipFlag = &cli.BoolFlag{
 		Name:     "rollup.disabletxpoolgossip",
 		Usage:    "Disable transaction pool gossip.",
@@ -1966,6 +1972,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	if ctx.IsSet(RollupHistoricalRPCTimeoutFlag.Name) {
 		cfg.RollupHistoricalRPCTimeout = ctx.Duration(RollupHistoricalRPCTimeoutFlag.Name)
 	}
+	if ctx.IsSet(RollupInteropRPCFlag.Name) {
+		cfg.InteropMessageRPC = ctx.String(RollupInteropRPCFlag.Name)
+	}
 	cfg.RollupDisableTxPoolGossip = ctx.Bool(RollupDisableTxPoolGossipFlag.Name)
 	cfg.RollupDisableTxPoolAdmission = cfg.RollupSequencerHTTP != "" && !ctx.Bool(RollupEnableTxPoolAdmissionFlag.Name)
 	cfg.RollupHaltOnIncompatibleProtocolVersion = ctx.String(RollupHaltOnIncompatibleProtocolVersionFlag.Name)

core/types/interoptypes/interop.go

@@ -0,0 +1,158 @@
+package interoptypes
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/holiman/uint256"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+var ExecutingMessageEventTopic = crypto.Keccak256Hash([]byte("ExecutingMessage(bytes32,(address,uint256,uint256,uint256,uint256))"))
+
+type Message struct {
+	Identifier  Identifier  `json:"identifier"`
+	PayloadHash common.Hash `json:"payloadHash"`
+}
+
+func (m *Message) DecodeEvent(topics []common.Hash, data []byte) error {
+	if len(topics) != 2 { // event hash, indexed payloadHash
+		return fmt.Errorf("unexpected number of event topics: %d", len(topics))
+	}
+	if topics[0] != ExecutingMessageEventTopic {
+		return fmt.Errorf("unexpected event topic %q", topics[0])
+	}
+	if len(data) != 32*5 {
+		return fmt.Errorf("unexpected identifier data length: %d", len(data))
+	}
+	take := func(length uint) []byte {
+		taken := data[:length]
+		data = data[length:]
+		return taken
+	}
+	takeZeroes := func(length uint) error {
+		for _, v := range take(length) {
+			if v != 0 {
+				return errors.New("expected zero")
+			}
+		}
+		return nil
+	}
+	if err := takeZeroes(12); err != nil {
+		return fmt.Errorf("invalid address padding: %w", err)
+	}
+	m.Identifier.Origin = common.Address(take(20))
+	if err := takeZeroes(32 - 8); err != nil {
+		return fmt.Errorf("invalid block number padding: %w", err)
+	}
+	m.Identifier.BlockNumber = binary.BigEndian.Uint64(take(8))
+	if err := takeZeroes(32 - 8); err != nil {
+		return fmt.Errorf("invalid log index padding: %w", err)
+	}
+	m.Identifier.LogIndex = binary.BigEndian.Uint64(take(8))
+	if err := takeZeroes(32 - 8); err != nil {
+		return fmt.Errorf("invalid timestamp padding: %w", err)
+	}
+	m.Identifier.Timestamp = binary.BigEndian.Uint64(take(8))
+	m.Identifier.ChainID.SetBytes32(take(32))
+	m.PayloadHash = topics[1]
+	return nil
+}
+
+func ExecutingMessagesFromLogs(logs []*types.Log) ([]Message, error) {
+	var executingMessages []Message
+	for i, l := range logs {
+		if l.Address == params.InteropCrossL2InboxAddress {
+			var msg Message
+			if err := msg.DecodeEvent(l.Topics, l.Data); err != nil {
+				return nil, fmt.Errorf("invalid executing message %d, tx-log %d: %w", len(executingMessages), i, err)
+			}
+			executingMessages = append(executingMessages, msg)
+		}
+	}
+	return executingMessages, nil
+}
+
+type Identifier struct {
+	Origin      common.Address
+	BlockNumber uint64
+	LogIndex    uint64
+	Timestamp   uint64
+	ChainID     uint256.Int // flat, not a pointer, to make Identifier safe as map key
+}
+
+type identifierMarshaling struct {
+	Origin      common.Address `json:"origin"`
+	BlockNumber hexutil.Uint64 `json:"blockNumber"`
+	LogIndex    hexutil.Uint64 `json:"logIndex"`
+	Timestamp   hexutil.Uint64 `json:"timestamp"`
+	ChainID     hexutil.U256   `json:"chainID"`
+}
+
+func (id Identifier) MarshalJSON() ([]byte, error) {
+	var enc identifierMarshaling
+	enc.Origin = id.Origin
+	enc.BlockNumber = hexutil.Uint64(id.BlockNumber)
+	enc.LogIndex = hexutil.Uint64(id.LogIndex)
+	enc.Timestamp = hexutil.Uint64(id.Timestamp)
+	enc.ChainID = (hexutil.U256)(id.ChainID)
+	return json.Marshal(&enc)
+}
+
+func (id *Identifier) UnmarshalJSON(input []byte) error {
+	var dec identifierMarshaling
+	if err := json.Unmarshal(input, &dec); err != nil {
+		return err
+	}
+	id.Origin = dec.Origin
+	id.BlockNumber = uint64(dec.BlockNumber)
+	id.LogIndex = uint64(dec.LogIndex)
+	id.Timestamp = uint64(dec.Timestamp)
+	id.ChainID = (uint256.Int)(dec.ChainID)
+	return nil
+}
+
+type SafetyLevel string
+
+func (lvl SafetyLevel) String() string {
+	return string(lvl)
+}
+
+func (lvl SafetyLevel) Valid() bool {
+	switch lvl {
+	case Finalized, Safe, CrossUnsafe, Unsafe:
+		return true
+	default:
+		return false
+	}
+}
+
+func (lvl SafetyLevel) MarshalText() ([]byte, error) {
+	return []byte(lvl), nil
+}
+
+func (lvl *SafetyLevel) UnmarshalText(text []byte) error {
+	if lvl == nil {
+		return errors.New("cannot unmarshal into nil SafetyLevel")
+	}
+	x := SafetyLevel(text)
+	if !x.Valid() {
+		return fmt.Errorf("unrecognized safety level: %q", text)
+	}
+	*lvl = x
+	return nil
+}
+
+const (
+	Finalized   SafetyLevel = "finalized"
+	Safe        SafetyLevel = "safe"
+	CrossUnsafe SafetyLevel = "cross-unsafe"
+	Unsafe      SafetyLevel = "unsafe"
+)

eth/backend.go

@@ -20,6 +20,7 @@ package eth
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/big"
 	"runtime"
@@ -38,10 +39,12 @@ import (
 	"github.com/ethereum/go-ethereum/core/txpool/blobpool"
 	"github.com/ethereum/go-ethereum/core/txpool/legacypool"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/types/interoptypes"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
+	"github.com/ethereum/go-ethereum/eth/interop"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/eth/tracers"
@@ -80,6 +83,8 @@ type Ethereum struct {
 	seqRPCService        *rpc.Client
 	historicalRPCService *rpc.Client
 
+	interopRPC *interop.InteropClient
+
 	// DB interfaces
 	chainDb ethdb.Database // Block chain database
 
@@ -336,6 +341,10 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		eth.historicalRPCService = client
 	}
 
+	if config.InteropMessageRPC != "" {
+		eth.interopRPC = interop.NewInteropClient(config.InteropMessageRPC)
+	}
+
 	// Start the RPC service
 	eth.netRPCService = ethapi.NewNetAPI(eth.p2pServer, networkID)
 
@@ -475,6 +484,9 @@ func (s *Ethereum) Stop() error {
 	if s.historicalRPCService != nil {
 		s.historicalRPCService.Close()
 	}
+	if s.interopRPC != nil {
+		s.interopRPC.Close()
+	}
 
 	// Clean shutdown marker as the last thing before closing db
 	s.shutdownTracker.Stop()
@@ -540,3 +552,10 @@ func (s *Ethereum) HandleRequiredProtocolVersion(required params.ProtocolVersion
 	}
 	return nil
 }
+
+func (s *Ethereum) CheckMessages(ctx context.Context, messages []interoptypes.Message, minSafety interoptypes.SafetyLevel) error {
+	if s.interopRPC == nil {
+		return errors.New("cannot check interop messages, no RPC available")
+	}
+	return s.interopRPC.CheckMessages(ctx, messages, minSafety)
+}

eth/ethconfig/config.go

@@ -190,6 +190,8 @@ type Config struct {
 	RollupDisableTxPoolGossip               bool
 	RollupDisableTxPoolAdmission            bool
 	RollupHaltOnIncompatibleProtocolVersion string
+
+	InteropMessageRPC string
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain config.

eth/interop/interop.go

@@ -0,0 +1,53 @@
+package interop
+
+import (
+	"context"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/core/types/interoptypes"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type InteropClient struct {
+	mu       sync.Mutex
+	client   *rpc.Client
+	endpoint string
+	closed   bool // don't allow lazy-dials after Close
+}
+
+// maybeDial dials the endpoint if it was not already.
+func (cl *InteropClient) maybeDial(ctx context.Context) error {
+	cl.mu.Lock()
+	defer cl.mu.Unlock()
+	if cl.client != nil || cl.closed {
+		return nil
+	}
+	rpcClient, err := rpc.DialContext(ctx, cl.endpoint)
+	if err != nil {
+		return err
+	}
+	cl.client = rpcClient
+	return nil
+}
+
+func (cl *InteropClient) Close() {
+	cl.mu.Lock()
+	defer cl.mu.Unlock()
+	if cl.client != nil {
+		cl.Close()
+	}
+	cl.closed = true
+}
+
+func NewInteropClient(rpcEndpoint string) *InteropClient {
+	return &InteropClient{endpoint: rpcEndpoint}
+}
+
+// CheckMessages checks if the given messages meet the given minimum safety level.
+func (cl *InteropClient) CheckMessages(ctx context.Context, messages []interoptypes.Message, minSafety interoptypes.SafetyLevel) error {
+	// we lazy-dial the endpoint, so we can start geth, and build blocks, without supervisor endpoint availability.
+	if err := cl.maybeDial(ctx); err != nil { // a single dial attempt is made, the next call may retry.
+		return err
+	}
+	return cl.client.CallContext(ctx, nil, "supervisor_checkMessages", messages, minSafety)
+}

fork.yaml

@@ -114,6 +114,7 @@ def:
           description: |
             The block-building code (in the "miner" package because of Proof-Of-Work legacy of ethereum) implements the
             changes to support the transaction-inclusion, tx-pool toggle and gaslimit parameters of the Engine API.
+            This also includes experimental support for interop executing-messages to be verified through an RPC.
           globs:
             - "miner/*"
         - title: "Tx-pool tx cost updates"