address PR comments and use reader interface in StatelessBlockValidator

arbnode/node.go

@@ -552,14 +552,20 @@ func createNodeImpl(
 
 	var statelessBlockValidator *staker.StatelessBlockValidator
 	if config.BlockValidator.ValidationServerConfigs[0].URL != "" {
+		var dapReaders []daprovider.Reader
+		if daReader != nil {
+			dapReaders = append(dapReaders, daprovider.NewReaderForDAS(daReader))
+		}
+		if blobReader != nil {
+			dapReaders = append(dapReaders, daprovider.NewReaderForBlobReader(blobReader))
+		}
 		statelessBlockValidator, err = staker.NewStatelessBlockValidator(
 			inboxReader,
 			inboxTracker,
 			txStreamer,
 			exec,
 			rawdb.NewTable(arbDb, storage.BlockValidatorPrefix),
-			daReader,
-			blobReader,
+			dapReaders,
 			func() *staker.BlockValidatorConfig { return &configFetcher.Get().BlockValidator },
 			stack,
 		)

arbstate/daprovider/reader.go

@@ -4,16 +4,12 @@
 package daprovider
 
 import (
-	"bytes"
 	"context"
-	"encoding/binary"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/offchainlabs/nitro/arbutil"
-	"github.com/offchainlabs/nitro/das/dastree"
 	"github.com/offchainlabs/nitro/util/blobs"
 )
 
@@ -54,89 +50,7 @@ func (d *readerForDAS) RecoverPayloadFromBatch(
 	preimageRecorder PreimageRecorder,
 	validateSeqMsg bool,
 ) ([]byte, error) {
-	cert, err := DeserializeDASCertFrom(bytes.NewReader(sequencerMsg[40:]))
-	if err != nil {
-		log.Error("Failed to deserialize DAS message", "err", err)
-		return nil, nil
-	}
-	version := cert.Version
-
-	if version >= 2 {
-		log.Error("Your node software is probably out of date", "certificateVersion", version)
-		return nil, nil
-	}
-
-	getByHash := func(ctx context.Context, hash common.Hash) ([]byte, error) {
-		newHash := hash
-		if version == 0 {
-			newHash = dastree.FlatHashToTreeHash(hash)
-		}
-
-		preimage, err := d.dasReader.GetByHash(ctx, newHash)
-		if err != nil && hash != newHash {
-			log.Debug("error fetching new style hash, trying old", "new", newHash, "old", hash, "err", err)
-			preimage, err = d.dasReader.GetByHash(ctx, hash)
-		}
-		if err != nil {
-			return nil, err
-		}
-
-		switch {
-		case version == 0 && crypto.Keccak256Hash(preimage) != hash:
-			fallthrough
-		case version == 1 && dastree.Hash(preimage) != hash:
-			log.Error(
-				"preimage mismatch for hash",
-				"hash", hash, "err", ErrHashMismatch, "version", version,
-			)
-			return nil, ErrHashMismatch
-		}
-		return preimage, nil
-	}
-
-	keysetPreimage, err := getByHash(ctx, cert.KeysetHash)
-	if err != nil {
-		log.Error("Couldn't get keyset", "err", err)
-		return nil, err
-	}
-	if preimageRecorder != nil {
-		dastree.RecordHash(preimageRecorder, keysetPreimage)
-	}
-
-	keyset, err := DeserializeKeyset(bytes.NewReader(keysetPreimage), !validateSeqMsg)
-	if err != nil {
-		return nil, fmt.Errorf("%w. Couldn't deserialize keyset, err: %w, keyset hash: %x batch num: %d", ErrSeqMsgValidation, err, cert.KeysetHash, batchNum)
-	}
-	err = keyset.VerifySignature(cert.SignersMask, cert.SerializeSignableFields(), cert.Sig)
-	if err != nil {
-		log.Error("Bad signature on DAS batch", "err", err)
-		return nil, nil
-	}
-
-	maxTimestamp := binary.BigEndian.Uint64(sequencerMsg[8:16])
-	if cert.Timeout < maxTimestamp+MinLifetimeSecondsForDataAvailabilityCert {
-		log.Error("Data availability cert expires too soon", "err", "")
-		return nil, nil
-	}
-
-	dataHash := cert.DataHash
-	payload, err := getByHash(ctx, dataHash)
-	if err != nil {
-		log.Error("Couldn't fetch DAS batch contents", "err", err)
-		return nil, err
-	}
-
-	if preimageRecorder != nil {
-		if version == 0 {
-			treeLeaf := dastree.FlatHashToTreeLeaf(dataHash)
-			preimageRecorder(dataHash, payload, arbutil.Keccak256PreimageType)
-			preimageRecorder(crypto.Keccak256Hash(treeLeaf), treeLeaf, arbutil.Keccak256PreimageType)
-		} else {
-			dastree.RecordHash(preimageRecorder, payload)
-		}
-	}
-
-	return payload, nil
+	return RecoverPayloadFromDasBatch(ctx, batchNum, sequencerMsg, d.dasReader, preimageRecorder, validateSeqMsg)
 }
 
 // NewReaderForBlobReader is generally meant to be only used by nitro.

arbstate/daprovider/util.go

@@ -136,25 +136,15 @@ func RecoverPayloadFromDasBatch(
 	batchNum uint64,
 	sequencerMsg []byte,
 	dasReader DASReader,
-	preimages map[arbutil.PreimageType]map[common.Hash][]byte,
-	keysetValidationMode KeysetValidationMode,
+	preimageRecorder PreimageRecorder,
+	validateSeqMsg bool,
 ) ([]byte, error) {
-	var keccakPreimages map[common.Hash][]byte
-	if preimages != nil {
-		if preimages[arbutil.Keccak256PreimageType] == nil {
-			preimages[arbutil.Keccak256PreimageType] = make(map[common.Hash][]byte)
-		}
-		keccakPreimages = preimages[arbutil.Keccak256PreimageType]
-	}
 	cert, err := DeserializeDASCertFrom(bytes.NewReader(sequencerMsg[40:]))
 	if err != nil {
 		log.Error("Failed to deserialize DAS message", "err", err)
 		return nil, nil
 	}
 	version := cert.Version
-	recordPreimage := func(key common.Hash, value []byte, ty arbutil.PreimageType) {
-		keccakPreimages[key] = value
-	}
 
 	if version >= 2 {
 		log.Error("Your node software is probably out of date", "certificateVersion", version)
@@ -194,18 +184,13 @@ func RecoverPayloadFromDasBatch(
 		log.Error("Couldn't get keyset", "err", err)
 		return nil, err
 	}
-	if keccakPreimages != nil {
-		dastree.RecordHash(recordPreimage, keysetPreimage)
+	if preimageRecorder != nil {
+		dastree.RecordHash(preimageRecorder, keysetPreimage)
 	}
 
-	keyset, err := DeserializeKeyset(bytes.NewReader(keysetPreimage), keysetValidationMode == KeysetDontValidate)
+	keyset, err := DeserializeKeyset(bytes.NewReader(keysetPreimage), !validateSeqMsg)
 	if err != nil {
-		logLevel := log.Error
-		if keysetValidationMode == KeysetPanicIfInvalid {
-			logLevel = log.Crit
-		}
-		logLevel("Couldn't deserialize keyset", "err", err, "keysetHash", cert.KeysetHash, "batchNum", batchNum)
-		return nil, nil
+		return nil, fmt.Errorf("%w. Couldn't deserialize keyset, err: %w, keyset hash: %x batch num: %d", ErrSeqMsgValidation, err, cert.KeysetHash, batchNum)
 	}
 	err = keyset.VerifySignature(cert.SignersMask, cert.SerializeSignableFields(), cert.Sig)
 	if err != nil {
@@ -226,13 +211,13 @@ func RecoverPayloadFromDasBatch(
 		return nil, err
 	}
 
-	if keccakPreimages != nil {
+	if preimageRecorder != nil {
 		if version == 0 {
 			treeLeaf := dastree.FlatHashToTreeLeaf(dataHash)
-			keccakPreimages[dataHash] = payload
-			keccakPreimages[crypto.Keccak256Hash(treeLeaf)] = treeLeaf
+			preimageRecorder(dataHash, payload, arbutil.Keccak256PreimageType)
+			preimageRecorder(crypto.Keccak256Hash(treeLeaf), treeLeaf, arbutil.Keccak256PreimageType)
 		} else {
-			dastree.RecordHash(recordPreimage, payload)
+			dastree.RecordHash(preimageRecorder, payload)
 		}
 	}
 

arbstate/inbox.go

@@ -49,7 +49,7 @@ const MaxDecompressedLen int = 1024 * 1024 * 16 // 16 MiB
 const maxZeroheavyDecompressedLen = 101*MaxDecompressedLen/100 + 64
 const MaxSegmentsPerSequencerMessage = 100 * 1024
 
-func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash common.Hash, data []byte, daProviders []daprovider.Reader, keysetValidationMode daprovider.KeysetValidationMode) (*sequencerMessage, error) {
+func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash common.Hash, data []byte, dapReaders []daprovider.Reader, keysetValidationMode daprovider.KeysetValidationMode) (*sequencerMessage, error) {
 	if len(data) < 40 {
 		return nil, errors.New("sequencer message missing L1 header")
 	}
@@ -74,13 +74,13 @@ func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash
 	// Stage 1: Extract the payload from any data availability header.
 	// It's important that multiple DAS strategies can't both be invoked in the same batch,
 	// as these headers are validated by the sequencer inbox and not other DASs.
-	// We try to extract payload from the first occuring valid DA provider in the daProviders list
+	// We try to extract payload from the first occuring valid DA reader in the dapReaders list
 	if len(payload) > 0 {
 		foundDA := false
 		var err error
-		for _, provider := range daProviders {
-			if provider != nil && provider.IsValidHeaderByte(payload[0]) {
-				payload, err = provider.RecoverPayloadFromBatch(ctx, batchNum, batchBlockHash, data, nil, keysetValidationMode != daprovider.KeysetDontValidate)
+		for _, dapReader := range dapReaders {
+			if dapReader != nil && dapReader.IsValidHeaderByte(payload[0]) {
+				payload, err = dapReader.RecoverPayloadFromBatch(ctx, batchNum, batchBlockHash, data, nil, keysetValidationMode != daprovider.KeysetDontValidate)
 				if err != nil {
 					// Matches the way keyset validation was done inside DAS readers i.e logging the error
 					//  But other daproviders might just want to return the error
@@ -164,7 +164,7 @@ func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash
 type inboxMultiplexer struct {
 	backend                   InboxBackend
 	delayedMessagesRead       uint64
-	daProviders               []daprovider.Reader
+	dapReaders                []daprovider.Reader
 	cachedSequencerMessage    *sequencerMessage
 	cachedSequencerMessageNum uint64
 	cachedSegmentNum          uint64
@@ -174,11 +174,11 @@ type inboxMultiplexer struct {
 	keysetValidationMode      daprovider.KeysetValidationMode
 }
 
-func NewInboxMultiplexer(backend InboxBackend, delayedMessagesRead uint64, daProviders []daprovider.Reader, keysetValidationMode daprovider.KeysetValidationMode) arbostypes.InboxMultiplexer {
+func NewInboxMultiplexer(backend InboxBackend, delayedMessagesRead uint64, dapReaders []daprovider.Reader, keysetValidationMode daprovider.KeysetValidationMode) arbostypes.InboxMultiplexer {
 	return &inboxMultiplexer{
 		backend:              backend,
 		delayedMessagesRead:  delayedMessagesRead,
-		daProviders:          daProviders,
+		dapReaders:           dapReaders,
 		keysetValidationMode: keysetValidationMode,
 	}
 }
@@ -200,7 +200,7 @@ func (r *inboxMultiplexer) Pop(ctx context.Context) (*arbostypes.MessageWithMeta
 		}
 		r.cachedSequencerMessageNum = r.backend.GetSequencerInboxPosition()
 		var err error
-		r.cachedSequencerMessage, err = parseSequencerMessage(ctx, r.cachedSequencerMessageNum, batchBlockHash, bytes, r.daProviders, r.keysetValidationMode)
+		r.cachedSequencerMessage, err = parseSequencerMessage(ctx, r.cachedSequencerMessageNum, batchBlockHash, bytes, r.dapReaders, r.keysetValidationMode)
 		if err != nil {
 			return nil, err
 		}

das/syncing_fallback_storage.go

@@ -213,9 +213,14 @@ func (s *l1SyncService) processBatchDelivered(ctx context.Context, batchDelivere
 
 	data = append(header, data...)
 	preimages := make(map[arbutil.PreimageType]map[common.Hash][]byte)
-	if _, err = daprovider.RecoverPayloadFromDasBatch(ctx, deliveredEvent.BatchSequenceNumber.Uint64(), data, s.dataSource, preimages, daprovider.KeysetValidate); err != nil {
-		log.Error("recover payload failed", "txhash", batchDeliveredLog.TxHash, "data", data)
-		return err
+	preimageRecorder := daprovider.RecordPreimagesTo(preimages)
+	if _, err = daprovider.RecoverPayloadFromDasBatch(ctx, deliveredEvent.BatchSequenceNumber.Uint64(), data, s.dataSource, preimageRecorder, true); err != nil {
+		if errors.Is(err, daprovider.ErrSeqMsgValidation) {
+			log.Error(err.Error())
+		} else {
+			log.Error("recover payload failed", "txhash", batchDeliveredLog.TxHash, "data", data)
+			return err
+		}
 	}
 	for _, preimages := range preimages {
 		for hash, contents := range preimages {

staker/l1_validator.go

@@ -10,7 +10,6 @@ import (
 	"math/big"
 	"time"
 
-	"github.com/offchainlabs/nitro/arbstate/daprovider"
 	"github.com/offchainlabs/nitro/staker/txbuilder"
 	"github.com/offchainlabs/nitro/util/arbmath"
 	"github.com/offchainlabs/nitro/validator"
@@ -50,7 +49,6 @@ type L1Validator struct {
 	wallet         ValidatorWalletInterface
 	callOpts       bind.CallOpts
 
-	das                daprovider.DASReader
 	inboxTracker       InboxTrackerInterface
 	txStreamer         TransactionStreamerInterface
 	blockValidator     *BlockValidator
@@ -62,7 +60,6 @@ func NewL1Validator(
 	wallet ValidatorWalletInterface,
 	validatorUtilsAddress common.Address,
 	callOpts bind.CallOpts,
-	das daprovider.DASReader,
 	inboxTracker InboxTrackerInterface,
 	txStreamer TransactionStreamerInterface,
 	blockValidator *BlockValidator,
@@ -90,7 +87,6 @@ func NewL1Validator(
 		builder:        builder,
 		wallet:         wallet,
 		callOpts:       callOpts,
-		das:            das,
 		inboxTracker:   inboxTracker,
 		txStreamer:     txStreamer,
 		blockValidator: blockValidator,

staker/staker.go

@@ -291,7 +291,7 @@ func NewStaker(
 	}
 	client := l1Reader.Client()
 	val, err := NewL1Validator(client, wallet, validatorUtilsAddress, callOpts,
-		statelessBlockValidator.daService, statelessBlockValidator.inboxTracker, statelessBlockValidator.streamer, blockValidator)
+		statelessBlockValidator.inboxTracker, statelessBlockValidator.streamer, blockValidator)
 	if err != nil {
 		return nil, err
 	}