[MEL] - Implement L2 messages accumulation and introduce MessageReader to extract messages from preimages

arbnode/mel/extraction/message_extraction_function.go

@@ -245,5 +245,11 @@ func extractMessagesImpl(
 			ParentChainBlock:    state.ParentChainBlockNumber,
 		})
 	}
+	if len(messages) > 0 {
+		// Only need to calculate partials once, after all the messages are extracted
+		if err := state.GenerateMessageMerklePartialsAndRoot(); err != nil {
+			return nil, nil, nil, nil, err
+		}
+	}
 	return state, messages, delayedMessages, batchMetas, nil
 }

arbnode/mel/messages.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/rlp"
 
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbutil"
@@ -57,18 +58,23 @@ func (m *DelayedInboxMessage) AfterInboxAcc() common.Hash {
 	return crypto.Keccak256Hash(m.BeforeInboxAcc[:], hash)
 }
 
-// Hash will replace AfterInboxAcc
 func (m *DelayedInboxMessage) Hash() common.Hash {
-	hash := crypto.Keccak256(
-		[]byte{m.Message.Header.Kind},
-		m.Message.Header.Poster.Bytes(),
-		arbmath.UintToBytes(m.Message.Header.BlockNumber),
-		arbmath.UintToBytes(m.Message.Header.Timestamp),
-		m.Message.Header.RequestId.Bytes(),
-		arbmath.U256Bytes(m.Message.Header.L1BaseFee),
-		crypto.Keccak256(m.Message.L2msg),
-	)
-	return crypto.Keccak256Hash(hash)
+	encoded, err := rlp.EncodeToBytes(m.WithMELRelevantFields())
+	if err != nil {
+		panic(err)
+	}
+	return crypto.Keccak256Hash(encoded)
+}
+
+func (m *DelayedInboxMessage) WithMELRelevantFields() *DelayedInboxMessage {
+	return &DelayedInboxMessage{
+		BlockHash: m.BlockHash,
+		Message: &arbostypes.L1IncomingMessage{
+			Header: m.Message.Header,
+			L2msg:  m.Message.L2msg,
+		},
+		ParentChainBlockNumber: m.ParentChainBlockNumber,
+	}
 }
 
 type BatchMetadata struct {

arbnode/mel/recording/delayed_msg_database.go

@@ -44,7 +44,51 @@ func NewDelayedMsgDatabase(db ethdb.KeyValueStore, preimages daprovider.Preimage
 	}, nil
 }
 
-func (r *DelayedMsgDatabase) initialize(ctx context.Context, state *mel.State) error {
+func (r *DelayedMsgDatabase) ReadDelayedMessage(ctx context.Context, state *mel.State, index uint64) (*mel.DelayedInboxMessage, error) {
+	if index == 0 { // Init message
+		// This message cannot be found in the database as it is supposed to be seen and read in the same block, so we persist that in DelayedMessageBacklog
+		return state.GetDelayedMessageBacklog().GetInitMsg(), nil
+	}
+	if !r.initialized {
+		if err := r.initialize(state); err != nil {
+			return nil, fmt.Errorf("error initializing recording database for MEL validation: %w", err)
+		}
+		r.initialized = true
+	}
+	// Lightweight operation that is needed as state.Clone() clears the seenDelayedMsgsAcc
+	if err := r.initSeenDelayedMsgsAccForRecording(state); err != nil {
+		return nil, fmt.Errorf("error initializing seenDelayedMsgsAcc for recording: %w", err)
+	}
+	delayed, err := fetchDelayedMessage(r.db, index)
+	if err != nil {
+		return nil, err
+	}
+	delayedMsgBytes, err := rlp.EncodeToBytes(delayed.WithMELRelevantFields())
+	if err != nil {
+		return nil, err
+	}
+	hashDelayedHash := crypto.Keccak256(delayed.Hash().Bytes())
+	r.preimages[common.BytesToHash(hashDelayedHash)] = delayedMsgBytes
+	return delayed, nil
+}
+
+func fetchDelayedMessage(db ethdb.KeyValueStore, index uint64) (*mel.DelayedInboxMessage, error) {
+	delayed, err := read.Value[mel.DelayedInboxMessage](db, read.Key(schema.MelDelayedMessagePrefix, index))
+	if err != nil {
+		return nil, err
+	}
+	return &delayed, nil
+}
+
+func getState(db ethdb.KeyValueStore, parentChainBlockNumber uint64) (*mel.State, error) {
+	state, err := read.Value[mel.State](db, read.Key(schema.MelStatePrefix, parentChainBlockNumber))
+	if err != nil {
+		return nil, err
+	}
+	return &state, nil
+}
+
+func (r *DelayedMsgDatabase) initialize(state *mel.State) error {
 	var acc *merkleAccumulator.MerkleAccumulator
 	for i := state.ParentChainBlockNumber; i > 0; i-- {
 		seenState, err := getState(r.db, i)
@@ -87,6 +131,11 @@ func (r *DelayedMsgDatabase) initialize(ctx context.Context, state *mel.State) e
 	if err != nil {
 		return err
 	}
+	return nil
+}
+
+func (r *DelayedMsgDatabase) initSeenDelayedMsgsAccForRecording(state *mel.State) error {
+	var err error
 	seenAcc := state.GetSeenDelayedMsgsAcc()
 	if seenAcc == nil {
 		seenAcc, err = merkleAccumulator.NewNonpersistentMerkleAccumulatorFromPartials(mel.ToPtrSlice(state.DelayedMessageMerklePartials))
@@ -98,43 +147,3 @@ func (r *DelayedMsgDatabase) initialize(ctx context.Context, state *mel.State) e
 	state.SetSeenDelayedMsgsAcc(seenAcc)
 	return nil
 }
-
-func (r *DelayedMsgDatabase) ReadDelayedMessage(ctx context.Context, state *mel.State, index uint64) (*mel.DelayedInboxMessage, error) {
-	if index == 0 { // Init message
-		// This message cannot be found in the database as it is supposed to be seen and read in the same block, so we persist that in DelayedMessageBacklog
-		return state.GetDelayedMessageBacklog().GetInitMsg(), nil
-	}
-	if !r.initialized {
-		if err := r.initialize(ctx, state); err != nil {
-			return nil, fmt.Errorf("error initializing recording database for MEL validation: %w", err)
-		}
-		r.initialized = true
-	}
-	delayed, err := fetchDelayedMessage(r.db, index)
-	if err != nil {
-		return nil, err
-	}
-	delayedMsgBytes, err := rlp.EncodeToBytes(delayed)
-	if err != nil {
-		return nil, err
-	}
-	hashDelayedHash := crypto.Keccak256(delayed.Hash().Bytes())
-	r.preimages[common.BytesToHash(hashDelayedHash)] = delayedMsgBytes
-	return delayed, nil
-}
-
-func fetchDelayedMessage(db ethdb.KeyValueStore, index uint64) (*mel.DelayedInboxMessage, error) {
-	delayed, err := read.Value[mel.DelayedInboxMessage](db, read.Key(schema.MelDelayedMessagePrefix, index))
-	if err != nil {
-		return nil, err
-	}
-	return &delayed, nil
-}
-
-func getState(db ethdb.KeyValueStore, parentChainBlockNumber uint64) (*mel.State, error) {
-	state, err := read.Value[mel.State](db, read.Key(schema.MelStatePrefix, parentChainBlockNumber))
-	if err != nil {
-		return nil, err
-	}
-	return &state, nil
-}

arbnode/mel/state.go

@@ -4,12 +4,17 @@ package mel
 
 import (
 	"context"
+	"errors"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rlp"
 
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbos/merkleAccumulator"
+	"github.com/offchainlabs/nitro/arbutil"
+	"github.com/offchainlabs/nitro/daprovider"
 )
 
 // State defines the main struct describing the results of processing a single parent
@@ -24,12 +29,13 @@ type State struct {
 	DelayedMessagePostingTargetAddress common.Address
 	ParentChainBlockHash               common.Hash
 	ParentChainPreviousBlockHash       common.Hash
-	MessageAccumulator                 common.Hash
-	DelayedMessagesSeenRoot            common.Hash
-	MsgCount                           uint64
 	BatchCount                         uint64
+	MsgCount                           uint64
+	MsgRoot                            common.Hash
 	DelayedMessagesRead                uint64
 	DelayedMessagesSeen                uint64
+	DelayedMessagesSeenRoot            common.Hash
+	MessageMerklePartials              []common.Hash `rlp:"optional"`
 	DelayedMessageMerklePartials       []common.Hash `rlp:"optional"`
 
 	delayedMessageBacklog *DelayedMessageBacklog // delayedMessageBacklog is initialized once in the Start fsm step of mel runner and is persisted across all future states
@@ -39,6 +45,11 @@ type State struct {
 	// from parent chain. It resets after the current melstate is finished generating
 	// and is reinitialized using appropriate DelayedMessageMerklePartials of the state
 	seenDelayedMsgsAcc *merkleAccumulator.MerkleAccumulator
+	// msgsAcc is MerkleAccumulator that accumulates all the L2 messages extracted. It
+	// resets after the current melstate is finished generating and is reinitialized using
+	// appropriate MessageMerklePartials of the state
+	msgsAcc          *merkleAccumulator.MerkleAccumulator
+	msgPreimagesDest daprovider.PreimagesMap
 }
 
 // DelayedMessageDatabase can read delayed messages by their global index.
@@ -79,7 +90,11 @@ type MessageConsumer interface {
 }
 
 func (s *State) Hash() common.Hash {
-	return common.Hash{}
+	encoded, err := rlp.EncodeToBytes(s)
+	if err != nil {
+		panic(err)
+	}
+	return crypto.Keccak256Hash(encoded)
 }
 
 // Performs a deep clone of the state struct to prevent any unintended
@@ -89,18 +104,24 @@ func (s *State) Clone() *State {
 	delayedMessageTarget := common.Address{}
 	parentChainHash := common.Hash{}
 	parentChainPrevHash := common.Hash{}
-	msgAcc := common.Hash{}
+	msgAccRoot := common.Hash{}
 	delayedMsgSeenRoot := common.Hash{}
 	copy(batchPostingTarget[:], s.BatchPostingTargetAddress[:])
 	copy(delayedMessageTarget[:], s.DelayedMessagePostingTargetAddress[:])
 	copy(parentChainHash[:], s.ParentChainBlockHash[:])
 	copy(parentChainPrevHash[:], s.ParentChainPreviousBlockHash[:])
-	copy(msgAcc[:], s.MessageAccumulator[:])
+	copy(msgAccRoot[:], s.MsgRoot[:])
 	copy(delayedMsgSeenRoot[:], s.DelayedMessagesSeenRoot[:])
+	var messageMerklePartials []common.Hash
+	for _, msgPartial := range s.MessageMerklePartials {
+		clone := common.Hash{}
+		copy(clone[:], msgPartial[:])
+		messageMerklePartials = append(messageMerklePartials, clone)
+	}
 	var delayedMessageMerklePartials []common.Hash
-	for _, partial := range s.DelayedMessageMerklePartials {
+	for _, delayedPartial := range s.DelayedMessageMerklePartials {
 		clone := common.Hash{}
-		copy(clone[:], partial[:])
+		copy(clone[:], delayedPartial[:])
 		delayedMessageMerklePartials = append(delayedMessageMerklePartials, clone)
 	}
 	var delayedMessageBacklog *DelayedMessageBacklog
@@ -115,20 +136,46 @@ func (s *State) Clone() *State {
 		DelayedMessagePostingTargetAddress: delayedMessageTarget,
 		ParentChainBlockHash:               parentChainHash,
 		ParentChainPreviousBlockHash:       parentChainPrevHash,
-		MessageAccumulator:                 msgAcc,
+		MsgRoot:                            msgAccRoot,
 		DelayedMessagesSeenRoot:            delayedMsgSeenRoot,
 		MsgCount:                           s.MsgCount,
 		BatchCount:                         s.BatchCount,
 		DelayedMessagesRead:                s.DelayedMessagesRead,
 		DelayedMessagesSeen:                s.DelayedMessagesSeen,
+		MessageMerklePartials:              messageMerklePartials,
 		DelayedMessageMerklePartials:       delayedMessageMerklePartials,
 		delayedMessageBacklog:              delayedMessageBacklog,
 		readCountFromBacklog:               s.readCountFromBacklog,
+		// we copy msgPreimagesDest as is to continue recordng of msg preimages
+		msgPreimagesDest: s.msgPreimagesDest,
 	}
 }
 
 func (s *State) AccumulateMessage(msg *arbostypes.MessageWithMetadata) error {
-	// TODO: Unimplemented.
+	if s.msgsAcc == nil {
+		log.Debug("Initializing MelState's msgsAcc")
+		// This is very low cost hence better to reconstruct msgsAcc from fresh partals instead of risking using a dirty acc
+		acc, err := merkleAccumulator.NewNonpersistentMerkleAccumulatorFromPartials(ToPtrSlice(s.MessageMerklePartials))
+		if err != nil {
+			return err
+		}
+		s.msgsAcc = acc
+		if s.msgPreimagesDest != nil {
+			s.msgsAcc.RecordPreimagesTo(s.msgPreimagesDest[arbutil.Keccak256PreimageType])
+			_, err := s.msgsAcc.Root()
+			if err != nil {
+				return err
+			}
+		}
+	}
+	msgBytes, err := rlp.EncodeToBytes(msg.WithMELRelevantFields())
+	if err != nil {
+		return err
+	}
+	// In recording mode this would also record the message preimages needed for MEL validation
+	if _, err := s.msgsAcc.Append(msg.Hash(), msgBytes...); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -163,18 +210,33 @@ func (s *State) AccumulateDelayedMessage(msg *DelayedInboxMessage) error {
 	return nil
 }
 
+func (s *State) GenerateMessageMerklePartialsAndRoot() error {
+	var err error
+	s.MessageMerklePartials, s.MsgRoot, err = getPartialsAndRoot(s.msgsAcc)
+	return err
+}
+
 func (s *State) GenerateDelayedMessagesSeenMerklePartialsAndRoot() error {
-	partialsPtrs, err := s.seenDelayedMsgsAcc.GetPartials()
+	var err error
+	s.DelayedMessageMerklePartials, s.DelayedMessagesSeenRoot, err = getPartialsAndRoot(s.seenDelayedMsgsAcc)
+	return err
+}
+
+func getPartialsAndRoot(acc *merkleAccumulator.MerkleAccumulator) ([]common.Hash, common.Hash, error) {
+	partialsPtrs, err := acc.GetPartials()
 	if err != nil {
-		return err
+		return nil, common.Hash{}, err
 	}
-	s.DelayedMessageMerklePartials = FromPtrSlice(partialsPtrs)
-	root, err := s.seenDelayedMsgsAcc.Root()
+	partials := FromPtrSlice(partialsPtrs)
+	root, err := acc.Root()
 	if err != nil {
-		return err
+		return nil, common.Hash{}, err
 	}
-	s.DelayedMessagesSeenRoot = root
-	return nil
+	return partials, root, err
+}
+
+func (s *State) SetMsgsAcc(acc *merkleAccumulator.MerkleAccumulator) {
+	s.msgsAcc = acc
 }
 
 func (s *State) GetSeenDelayedMsgsAcc() *merkleAccumulator.MerkleAccumulator {
@@ -207,6 +269,20 @@ func (s *State) ReorgTo(newState *State) error {
 	return nil
 }
 
+// RecordMsgPreimagesTo initializes the state's msgPreimagesDest to record preimages
+// related to the extracted messages needed for MEL validation into the given preimages map,
+// this will be used to initialize msgsAcc when accumulating messages
+func (s *State) RecordMsgPreimagesTo(preimagesMap daprovider.PreimagesMap) error {
+	if preimagesMap == nil {
+		return errors.New("msg preimages recording destination cannot be nil")
+	}
+	if _, ok := preimagesMap[arbutil.Keccak256PreimageType]; !ok {
+		preimagesMap[arbutil.Keccak256PreimageType] = make(map[common.Hash][]byte)
+	}
+	s.msgPreimagesDest = preimagesMap
+	return nil
+}
+
 func ToPtrSlice[T any](list []T) []*T {
 	var ptrs []*T
 	for _, item := range list {

arbos/arbostypes/incomingmessage.go

@@ -68,6 +68,9 @@ type L1IncomingMessage struct {
 	// Only used for `L1MessageType_BatchPostingReport`
 	// note: the legacy field is used in json to support older clients
 	// in rlp it's used to distinguish old from new (old will load into first arg)
+	//
+	// NOTE: These fields are not included when storing a L1MessageType_BatchPostingReport
+	// type delayed message or L2 message into the preimages map for MEL validation
 	LegacyBatchGasCost *uint64         `json:"batchGasCost,omitempty" rlp:"optional"`
 	BatchDataStats     *BatchDataStats `json:"batchDataTokens,omitempty" rlp:"optional"`
 }