[MEL] - Implement recording of preimages related to sequencer batches (DA providers) (#4133)

* [MEL] - Implement delayed message accumulation in native mode

* address PR comments

* add documentation for checkAgainstAccumulator and a minor fix

* undo changes to addressed review comments from other PRs

* dont make L2msg rlp optional

* make meldb take a KeyValueStore

* handle reorg in start step- reducing code diff

* Message extraction function works with logs instead of receipts

* only keep delayed message accumulation changes

* cleanup non related code

* Implement preimage recorder for DelayedMessageDatabase interface

* Update cmd/mel-replay/delayed_message_db_test.go

* fix test name

* code refactor

* fix lint

* Implement recording of preimages related to sequencer batches (DA providers)

* address PR comments

* add RecoverPayloadAndPreimages method on DACertificatePreimageReader

* implement RecoverPayloadAndPreimages method on EvilDAProvider

* fix schema.go

* bring in merkle partials calculation step in ExtractMessages function

* code refactor

* add documentation

* add changelog

* move recording related code to new package, melrecording

* reduce conflicts

* address PR comments

---------

Co-authored-by: Raul Jordan <[email protected]>
Co-authored-by: Pepper Lebeck-Jobe <[email protected]>

Author: 3 people

arbnode/mel/extraction/message_extraction_function.go

@@ -57,7 +57,7 @@ func ExtractMessages(
 	ctx context.Context,
 	inputState *mel.State,
 	parentChainHeader *types.Header,
-	dataProviders *daprovider.DAProviderRegistry,
+	dapReaders arbstate.DapReaderSource,
 	delayedMsgDatabase DelayedMessageDatabase,
 	txFetcher TransactionFetcher,
 	logsFetcher LogsFetcher,
@@ -68,7 +68,7 @@ func ExtractMessages(
 		inputState,
 		parentChainHeader,
 		chainConfig,
-		dataProviders,
+		dapReaders,
 		delayedMsgDatabase,
 		txFetcher,
 		logsFetcher,
@@ -90,7 +90,7 @@ func extractMessagesImpl(
 	inputState *mel.State,
 	parentChainHeader *types.Header,
 	chainConfig *params.ChainConfig,
-	dataProviders *daprovider.DAProviderRegistry,
+	dapReaders arbstate.DapReaderSource,
 	delayedMsgDatabase DelayedMessageDatabase,
 	txFetcher TransactionFetcher,
 	logsFetcher LogsFetcher,
@@ -154,6 +154,12 @@ func extractMessagesImpl(
 		}
 		state.DelayedMessagesSeen += 1
 	}
+	if len(delayedMessages) > 0 {
+		// Only need to calculate partials once, after all the delayed messages are `seen`
+		if err := state.GenerateDelayedMessagesSeenMerklePartialsAndRoot(); err != nil {
+			return nil, nil, nil, nil, err
+		}
+	}
 
 	// Batch posting reports are included in the same transaction as a batch, so there should
 	// always be the same number of reports as there are batches.
@@ -207,7 +213,7 @@ func extractMessagesImpl(
 			batch.SequenceNumber,
 			batch.BlockHash,
 			serialized,
-			dataProviders,
+			dapReaders,
 			daprovider.KeysetValidate,
 			chainConfig,
 		)

arbnode/mel/recording/dap_reader_source.go

@@ -0,0 +1,67 @@
+package melrecording
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/offchainlabs/nitro/arbstate"
+	"github.com/offchainlabs/nitro/daprovider"
+	"github.com/offchainlabs/nitro/util/containers"
+	"github.com/offchainlabs/nitro/validator"
+)
+
+// RecordingDAPReader implements recording of preimages when melextraction.ExtractMessages function is called by MEL validator for creation
+// of validation entry. Since ExtractMessages function would use daprovider.Reader interface to fetch the sequencer batch via RecoverPayload
+// we implement collecting of preimages as well in the same method and record it
+type RecordingDAPReader struct {
+	validatorCtx context.Context
+	reader       daprovider.Reader
+	preimages    daprovider.PreimagesMap
+}
+
+func (r *RecordingDAPReader) RecoverPayload(batchNum uint64, batchBlockHash common.Hash, sequencerMsg []byte) containers.PromiseInterface[daprovider.PayloadResult] {
+	promise := r.reader.RecoverPayloadAndPreimages(batchNum, batchBlockHash, sequencerMsg)
+	result, err := promise.Await(r.validatorCtx)
+	if err != nil {
+		return containers.NewReadyPromise(daprovider.PayloadResult{}, err)
+	}
+	validator.CopyPreimagesInto(r.preimages, result.Preimages)
+	return containers.NewReadyPromise(daprovider.PayloadResult{Payload: result.Payload}, nil)
+}
+
+func (r *RecordingDAPReader) CollectPreimages(batchNum uint64, batchBlockHash common.Hash, sequencerMsg []byte) containers.PromiseInterface[daprovider.PreimagesResult] {
+	return r.reader.CollectPreimages(batchNum, batchBlockHash, sequencerMsg)
+}
+
+func (r *RecordingDAPReader) RecoverPayloadAndPreimages(batchNum uint64, batchBlockHash common.Hash, sequencerMsg []byte) containers.PromiseInterface[daprovider.PayloadAndPreimagesResult] {
+	return r.reader.RecoverPayloadAndPreimages(batchNum, batchBlockHash, sequencerMsg)
+}
+
+// RecordingDAPReaderSource is used for recording preimages related to sequencer batches stored by da providers, given a
+// DapReaderSource it implements GetReader method to return a daprovider.Reader interface that records preimgaes. It takes
+// in a context variable (corresponding to creation of validation entry) from the MEL validator
+type RecordingDAPReaderSource struct {
+	validatorCtx context.Context
+	dapReaders   arbstate.DapReaderSource
+	preimages    daprovider.PreimagesMap
+}
+
+func NewRecordingDAPReaderSource(validatorCtx context.Context, dapReaders arbstate.DapReaderSource) *RecordingDAPReaderSource {
+	return &RecordingDAPReaderSource{
+		validatorCtx: validatorCtx,
+		dapReaders:   dapReaders,
+		preimages:    make(daprovider.PreimagesMap),
+	}
+}
+
+func (s *RecordingDAPReaderSource) GetReader(headerByte byte) daprovider.Reader {
+	reader := s.dapReaders.GetReader(headerByte)
+	return &RecordingDAPReader{
+		validatorCtx: s.validatorCtx,
+		reader:       reader,
+		preimages:    s.preimages,
+	}
+}
+
+func (s *RecordingDAPReaderSource) Preimages() daprovider.PreimagesMap { return s.preimages }

arbnode/mel/runner/recording_database.go …de/mel/recording/delayed_msg_database.goarbnode/mel/runner/recording_database.go renamed to arbnode/mel/recording/delayed_msg_database.go arbnode/mel/runner/recording_database.go renamed to arbnode/mel/recording/delayed_msg_database.go

@@ -1,30 +1,34 @@
-package melrunner
+package melrecording
 
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 
+	"github.com/offchainlabs/nitro/arbnode/db/read"
+	"github.com/offchainlabs/nitro/arbnode/db/schema"
 	"github.com/offchainlabs/nitro/arbnode/mel"
 	"github.com/offchainlabs/nitro/arbos/merkleAccumulator"
 )
 
-// RecordingDatabase holds an ethdb.KeyValueStore that contains delayed messages stored by native MEL and implements DelayedMessageDatabase
+// DelayedMsgDatabase holds an ethdb.KeyValueStore that contains delayed messages stored by native MEL and implements DelayedMessageDatabase
 // interface defined in 'mel'. It is solely used for recording of preimages relating to delayed messages needed for MEL validation
-type RecordingDatabase struct {
-	db        ethdb.KeyValueStore
-	preimages map[common.Hash][]byte
+type DelayedMsgDatabase struct {
+	db          ethdb.KeyValueStore
+	preimages   map[common.Hash][]byte
+	initialized bool
 }
 
-func NewRecordingDatabase(db ethdb.KeyValueStore) *RecordingDatabase {
-	return &RecordingDatabase{db, make(map[common.Hash][]byte)}
+func NewDelayedMsgDatabase(db ethdb.KeyValueStore) *DelayedMsgDatabase {
+	return &DelayedMsgDatabase{db, make(map[common.Hash][]byte), false}
 }
 
-func (r *RecordingDatabase) Initialize(ctx context.Context, state *mel.State) error {
+func (r *DelayedMsgDatabase) initialize(ctx context.Context, state *mel.State) error {
 	var acc *merkleAccumulator.MerkleAccumulator
 	for i := state.ParentChainBlockNumber; i > 0; i-- {
 		seenState, err := getState(ctx, r.db, i)
@@ -79,13 +83,19 @@ func (r *RecordingDatabase) Initialize(ctx context.Context, state *mel.State) er
 	return nil
 }
 
-func (r *RecordingDatabase) Preimages() map[common.Hash][]byte { return r.preimages }
+func (r *DelayedMsgDatabase) Preimages() map[common.Hash][]byte { return r.preimages }
 
-func (r *RecordingDatabase) ReadDelayedMessage(ctx context.Context, state *mel.State, index uint64) (*mel.DelayedInboxMessage, 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(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
@@ -98,3 +108,19 @@ func (r *RecordingDatabase) ReadDelayedMessage(ctx context.Context, state *mel.S
 	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(ctx context.Context, 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/runner/backlog.go

@@ -8,14 +8,14 @@ import (
 )
 
 // InitializeDelayedMessageBacklog is to be only called by the Start fsm step of MEL. This function fills the backlog based on the seen and read count from the given mel state
-func InitializeDelayedMessageBacklog(ctx context.Context, d *mel.DelayedMessageBacklog, db *Database, state *mel.State, finalizedAndReadIndexFetcher func(context.Context) (uint64, error)) error {
+func InitializeDelayedMessageBacklog(ctx context.Context, d *mel.DelayedMessageBacklog, db *Database, state *mel.State, finalizedAndReadIndexFetcher func() (uint64, error)) error {
 	if state.DelayedMessagesSeen == 0 && state.DelayedMessagesRead == 0 { // this is the first mel state so no need to initialize backlog even if the state isn't finalized yet
 		return nil
 	}
 	finalizedDelayedMessagesRead := state.DelayedMessagesRead // Assume to be finalized, then update if needed
 	var err error
 	if finalizedAndReadIndexFetcher != nil {
-		finalizedDelayedMessagesRead, err = finalizedAndReadIndexFetcher(ctx)
+		finalizedDelayedMessagesRead, err = finalizedAndReadIndexFetcher()
 		if err != nil {
 			return err
 		}

arbnode/mel/runner/backlog_test.go

@@ -110,7 +110,7 @@ func TestDelayedMessageBacklogInitialization(t *testing.T) {
 	require.NoError(t, err)
 	newDelayedMessageBacklog, err := mel.NewDelayedMessageBacklog(100, func() (uint64, error) { return 0, nil })
 	require.NoError(t, err)
-	require.NoError(t, InitializeDelayedMessageBacklog(ctx, newDelayedMessageBacklog, melDB, newState, func(context.Context) (uint64, error) { return 7, nil }))
+	require.NoError(t, InitializeDelayedMessageBacklog(ctx, newDelayedMessageBacklog, melDB, newState, func() (uint64, error) { return 7, nil }))
 	// Notice that instead of having seenUnread list from delayed index 13 to 25 inclusive we will have it from 7 to 25 as only till block=7 the chain has finalized and that block has DelayedMessagesRead=7
 	require.True(t, newDelayedMessageBacklog.Len() == 19)
 	newState.SetDelayedMessageBacklog(newDelayedMessageBacklog)

arbnode/mel/runner/database.go

@@ -70,24 +70,11 @@ func (d *Database) setHeadMelStateBlockNum(batch ethdb.KeyValueWriter, parentCha
 }
 
 func (d *Database) GetHeadMelStateBlockNum() (uint64, error) {
-	parentChainBlockNumberBytes, err := d.db.Get(schema.HeadMelStateBlockNumKey)
-	if err != nil {
-		return 0, err
-	}
-	var parentChainBlockNumber uint64
-	err = rlp.DecodeBytes(parentChainBlockNumberBytes, &parentChainBlockNumber)
-	if err != nil {
-		return 0, err
-	}
-	return parentChainBlockNumber, nil
+	return read.Value[uint64](d.db, schema.HeadMelStateBlockNumKey)
 }
 
 func (d *Database) State(ctx context.Context, parentChainBlockNumber uint64) (*mel.State, error) {
-	return getState(ctx, d.db, parentChainBlockNumber)
-}
-
-func getState(ctx context.Context, db ethdb.KeyValueStore, parentChainBlockNumber uint64) (*mel.State, error) {
-	state, err := read.Value[mel.State](db, read.Key(schema.MelStatePrefix, parentChainBlockNumber))
+	state, err := read.Value[mel.State](d.db, read.Key(schema.MelStatePrefix, parentChainBlockNumber))
 	if err != nil {
 		return nil, err
 	}
@@ -116,15 +103,10 @@ func (d *Database) SaveBatchMetas(ctx context.Context, state *mel.State, batchMe
 }
 
 func (d *Database) fetchBatchMetadata(seqNum uint64) (*mel.BatchMetadata, error) {
-	key := read.Key(schema.MelSequencerBatchMetaPrefix, seqNum)
-	batchMetadataBytes, err := d.db.Get(key)
+	batchMetadata, err := read.Value[mel.BatchMetadata](d.db, read.Key(schema.MelSequencerBatchMetaPrefix, seqNum))
 	if err != nil {
 		return nil, err
 	}
-	var batchMetadata mel.BatchMetadata
-	if err = rlp.DecodeBytes(batchMetadataBytes, &batchMetadata); err != nil {
-		return nil, err
-	}
 	return &batchMetadata, nil
 }
 
@@ -167,11 +149,7 @@ func (d *Database) ReadDelayedMessage(ctx context.Context, state *mel.State, ind
 }
 
 func (d *Database) fetchDelayedMessage(index uint64) (*mel.DelayedInboxMessage, error) {
-	return fetchDelayedMessage(d.db, index)
-}
-
-func fetchDelayedMessage(db ethdb.KeyValueStore, index uint64) (*mel.DelayedInboxMessage, error) {
-	delayed, err := read.Value[mel.DelayedInboxMessage](db, read.Key(schema.MelDelayedMessagePrefix, index))
+	delayed, err := read.Value[mel.DelayedInboxMessage](d.db, read.Key(schema.MelDelayedMessagePrefix, index))
 	if err != nil {
 		return nil, err
 	}

changelog/ganeshvanahalli-nit-4194.md

@@ -0,0 +1,2 @@
+### Added
+ - Introduces recording of preimages related to posted sequencer batches to DA providers for MEL validation

cmd/mel-replay/delayed_message_db_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 
 	"github.com/offchainlabs/nitro/arbnode/mel"
+	melrecording "github.com/offchainlabs/nitro/arbnode/mel/recording"
 	"github.com/offchainlabs/nitro/arbnode/mel/runner"
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbutil"
@@ -69,8 +70,7 @@ func TestRecordingPreimagesForReadDelayedMessage(t *testing.T) {
 	require.NoError(t, state.GenerateDelayedMessagesSeenMerklePartialsAndRoot())
 	require.NoError(t, melDB.SaveState(ctx, state))
 
-	recordingDB := melrunner.NewRecordingDatabase(db)
-	require.NoError(t, recordingDB.Initialize(ctx, state))
+	recordingDB := melrecording.NewDelayedMsgDatabase(db)
 	for i := startBlockNum; i < numMsgs; i++ {
 		require.NoError(t, state.AccumulateDelayedMessage(delayedMessages[i]))
 		state.DelayedMessagesSeen++

cmd/replay/main.go

@@ -243,6 +243,18 @@ func (r *DACertificatePreimageReader) CollectPreimages(
 	})
 }
 
+func (r *DACertificatePreimageReader) RecoverPayloadAndPreimages(
+	batchNum uint64,
+	batchBlockHash common.Hash,
+	sequencerMsg []byte,
+) containers.PromiseInterface[daprovider.PayloadAndPreimagesResult] {
+	return containers.DoPromise(context.Background(), func(ctx context.Context) (daprovider.PayloadAndPreimagesResult, error) {
+		// Stub implementation: RecoverPayloadAndPreimages is only called
+		// by the MEL validator to gather preimages before validation
+		return daprovider.PayloadAndPreimagesResult{Preimages: make(daprovider.PreimagesMap), Payload: nil}, nil
+	})
+}
+
 // To generate:
 // key, _ := crypto.HexToECDSA("0000000000000000000000000000000000000000000000000000000000000001")
 // sig, _ := crypto.Sign(make([]byte, 32), key)

daprovider/anytrust/util/util.go

@@ -92,6 +92,21 @@ func (d *reader) CollectPreimages(
 	})
 }
 
+// RecoverPayloadAndPreimages fetches the underlying payload and collects preimages from the DA provider given the batch header information
+func (d *reader) RecoverPayloadAndPreimages(
+	batchNum uint64,
+	batchBlockHash common.Hash,
+	sequencerMsg []byte,
+) containers.PromiseInterface[daprovider.PayloadAndPreimagesResult] {
+	return containers.DoPromise(context.Background(), func(ctx context.Context) (daprovider.PayloadAndPreimagesResult, error) {
+		payload, preimages, err := d.recoverInternal(ctx, batchNum, sequencerMsg, true, true)
+		return daprovider.PayloadAndPreimagesResult{
+			Payload:   payload,
+			Preimages: preimages,
+		}, err
+	})
+}
+
 // NewWriter is generally meant to be only used by nitro.
 // DA Providers should implement methods in the DAProviderWriter interface independently
 func NewWriter(anyTrustWriter Writer, maxMessageSize int) *writer {