refactor to get batch submission/derivation working

arbitrator/jit/src/wavmio.rs

@@ -167,8 +167,8 @@ pub fn resolve_preimage_impl(
         .and_then(|m| m.get(&hash))
     else {
         let hash_hex = hex::encode(hash);
-        error!("Missing requested preimage for hash {hash_hex} in {name}")
-    };
+        error!("Missing requested preimage for hash {hash_hex} in {name} for type {:?}", preimage_type)
+        };
 
     // Check if preimage rehashes to the provided hash. Exclude blob preimages
     let calculated_hash: [u8; 32] = match preimage_type {
@@ -301,4 +301,4 @@ fn ready_hostio(env: &mut WasmEnv) -> MaybeEscape {
     env.process.socket = Some((writer, reader));
     env.process.forks = false;
     Ok(())
-}
+}

arbitrator/prover/src/machine.rs

@@ -3047,11 +3047,9 @@ impl Machine {
                                     .expect("Failed to generate KZG preimage proof");
                             }
                             PreimageType::EigenDAHash => {
-                                // TODO - Add eigenDA kzg preimage verification here
                                 println!("Generating proof for EigenDA preimage");
                                 prove_kzg_preimage_bn254(hash, &preimage, offset, &mut data)
                                     .expect("Failed to generate eigenDA KZG preimage proof");
-                                //data.extend(preimage);
                             }
                         }
                     } else if next_inst.opcode == Opcode::ReadInboxMessage {

arbitrator/prover/src/utils.rs

@@ -229,6 +229,8 @@ pub fn hash_preimage(preimage: &[u8], ty: PreimageType) -> Result<[u8; 32]> {
             let mut commitment_hash: [u8; 32] = Sha256::digest(&commitment_bytes).into();
             commitment_hash[0] = 1;
 
+            println!("commitment_hash: {:?}", commitment_hash);
+
             Ok(commitment_hash)
         }
     }

arbnode/batch_poster.go

@@ -42,6 +42,7 @@ import (
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/cmd/chaininfo"
 	"github.com/offchainlabs/nitro/cmd/genericconf"
+	"github.com/offchainlabs/nitro/eigenda"
 	"github.com/offchainlabs/nitro/execution"
 	"github.com/offchainlabs/nitro/solgen/go/bridgegen"
 	"github.com/offchainlabs/nitro/util"
@@ -106,6 +107,9 @@ type BatchPoster struct {
 	gasRefunderAddr    common.Address
 	building           *buildingBatch
 	dapWriter          daprovider.Writer
+	// This deviates from the DA spec but is necessary for the batch poster to work efficiently
+	// since we need to an extended method on the SequencerInbox contract
+	eigenDAWriter      eigenda.EigenDAWriter
 	dataPoster         *dataposter.DataPoster
 	redisLock          *redislock.Simple
 	messagesPerBatch   *arbmath.MovingAverage[uint64]
@@ -366,6 +370,7 @@ func NewBatchPoster(ctx context.Context, opts *BatchPosterOpts) (*BatchPoster, e
 		gasRefunderAddr:    opts.Config().gasRefunder,
 		bridgeAddr:         opts.DeployInfo.Bridge,
 		dapWriter:          opts.DAPWriter,
+		eigenDAWriter:      opts.EigenDAWriter,
 		redisLock:          redisLock,
 	}
 	b.messagesPerBatch, err = arbmath.NewMovingAverage[uint64](20)
@@ -699,7 +704,7 @@ type buildingBatch struct {
 	msgCount          arbutil.MessageIndex
 	haveUsefulMessage bool
 	use4844           bool
-	useEigenDA        bool
+	useEigenDA 	  bool
 }
 
 func newBatchSegments(firstDelayed uint64, config *BatchPosterConfig, backlog uint64, use4844 bool) *batchSegments {
@@ -1226,8 +1231,6 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context) (bool, error)
 			useEigenDA = true
 		}
 
-		println("use4844", use4844, "useEigenDA", useEigenDA)
-
 		b.building = &buildingBatch{
 			segments:      newBatchSegments(batchPosition.DelayedMessageCount, b.config(), b.GetBacklogEstimate(), use4844),
 			msgCount:      batchPosition.MessageCount,
@@ -1408,6 +1411,31 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context) (bool, error)
 		batchPosterDALastSuccessfulActionGauge.Update(time.Now().Unix())
 	}
 
+	var eigenDaBlobInfo *eigenda.EigenDABlobInfo
+	if b.eigenDAWriter != nil {
+		if !b.redisLock.AttemptLock(ctx) {
+			return false, errAttemptLockFailed
+		}
+
+		gotNonce, gotMeta, err := b.dataPoster.GetNextNonceAndMeta(ctx)
+		if err != nil {
+			batchPosterDAFailureCounter.Inc(1)
+			return false, err
+		}
+		if nonce != gotNonce || !bytes.Equal(batchPositionBytes, gotMeta) {
+			batchPosterDAFailureCounter.Inc(1)
+			return false, fmt.Errorf("%w: nonce changed from %d to %d while creating batch", storage.ErrStorageRace, nonce, gotNonce)
+		}
+		eigenDaBlobInfo, err = b.eigenDAWriter.Store(ctx, sequencerMsg)
+		if err != nil {
+			batchPosterDAFailureCounter.Inc(1)
+			return false, err
+		}
+
+		batchPosterDASuccessCounter.Inc(1)
+		batchPosterDALastSuccessfulActionGauge.Update(time.Now().Unix())
+	}
+
 	prevMessageCount := batchPosition.MessageCount
 	if b.config().Dangerous.AllowPostingFirstBatchWhenSequencerMessageCountMismatch && !b.postedFirstBatch {
 		// AllowPostingFirstBatchWhenSequencerMessageCountMismatch can be used when the
@@ -1438,7 +1466,7 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context) (bool, error)
 		prevMessageCount = 0
 	}
 
-	data, kzgBlobs, err := b.encodeAddBatch(new(big.Int).SetUint64(batchPosition.NextSeqNum), prevMessageCount, b.building.msgCount, sequencerMsg, b.building.segments.delayedMsg, b.building.use4844)
+	data, kzgBlobs, err := b.encodeAddBatch(new(big.Int).SetUint64(batchPosition.NextSeqNum), prevMessageCount, b.building.msgCount, sequencerMsg, b.building.segments.delayedMsg, b.building.use4844, b.building.useEigenDA, eigenDaBlobInfo)
 	if err != nil {
 		return false, err
 	}
@@ -1453,7 +1481,7 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context) (bool, error)
 	// In theory, this might reduce gas usage, but only by a factor that's already
 	// accounted for in `config.ExtraBatchGas`, as that same factor can appear if a user
 	// posts a new delayed message that we didn't see while gas estimating.
-	gasLimit, err := b.estimateGas(ctx, sequencerMsg, lastPotentialMsg.DelayedMessagesRead, data, kzgBlobs, nonce, accessList, blobInfo)
+	gasLimit, err := b.estimateGas(ctx, sequencerMsg, lastPotentialMsg.DelayedMessagesRead, data, kzgBlobs, nonce, accessList, eigenDaBlobInfo)
 	if err != nil {
 		return false, err
 	}

arbnode/node.go

@@ -522,11 +522,11 @@ func createNodeImpl(
 		return nil, err
 	}
 
+	var eigenDAReader eigenda.EigenDAReader
+	var eigenDAWriter eigenda.EigenDAWriter
 	var daWriter das.DataAvailabilityServiceWriter
 	var daReader das.DataAvailabilityServiceReader
 	var dasLifecycleManager *das.LifecycleManager
-	var eigenDAReader eigenda.EigenDAReader
-	var eigenDAWriter eigenda.EigenDAWriter
 	if config.DataAvailability.Enable {
 		if config.BatchPoster.Enable {
 			daWriter, daReader, dasLifecycleManager, err = das.CreateBatchPosterDAS(ctx, &config.DataAvailability, dataSigner, l1client, deployInfo.SequencerInbox)
@@ -564,6 +564,9 @@ func createNodeImpl(
 		return nil, errors.New("data availability service required but unconfigured")
 	}
 	var dapReaders []daprovider.Reader
+	if eigenDAReader != nil {
+		dapReaders = append(dapReaders, eigenda.NewReaderForEigenDA(eigenDAReader))
+	}
 	if daReader != nil {
 		dapReaders = append(dapReaders, daprovider.NewReaderForDAS(daReader))
 	}
@@ -697,6 +700,7 @@ func createNodeImpl(
 		if daWriter != nil {
 			dapWriter = daprovider.NewWriterForDAS(daWriter)
 		}
+
 		batchPoster, err = NewBatchPoster(ctx, &BatchPosterOpts{
 			DataPosterDB:  rawdb.NewTable(arbDb, storage.BatchPosterPrefix),
 			L1Reader:      l1Reader,
@@ -708,6 +712,7 @@ func createNodeImpl(
 			DeployInfo:    deployInfo,
 			TransactOpts:  txOptsBatchPoster,
 			DAPWriter:     dapWriter,
+			EigenDAWriter: eigenDAWriter,
 			ParentChainID: parentChainID,
 		})
 		if err != nil {

arbnode/sequencer_inbox.go

@@ -18,7 +18,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/offchainlabs/nitro/arbstate/daprovider"
 	"github.com/offchainlabs/nitro/arbutil"
-	"github.com/offchainlabs/nitro/eigenda"
 
 	"github.com/offchainlabs/nitro/solgen/go/bridgegen"
 )
@@ -180,14 +179,11 @@ func (m *SequencerInboxBatch) getSequencerData(ctx context.Context, client arbut
 		calldata := tx.Data()
 		println("appending EigenDA message header flag to calldata")
 		// append the eigenDA header flag to the front
-		data := []byte{eigenda.EigenDAMessageHeaderFlag}
+		data := []byte{daprovider.EigenDAMessageHeaderFlag}
 		data = append(data, calldata[:]...)
 
 		println(fmt.Sprintf("Returning the following calldata: %s", hexutil.Encode(data)))
-
-		// format of eigenDA data is
-		// [0 - 1] header flag
-		// [1 - len(data)] calldata
+
 		return data, nil
 	default:
 		return nil, fmt.Errorf("batch has invalid data location %v", m.dataLocation)

arbstate/daprovider/util.go

@@ -78,11 +78,13 @@ const ZeroheavyMessageHeaderFlag byte = 0x20
 // BlobHashesHeaderFlag indicates that this message contains EIP 4844 versioned hashes of the committments calculated over the blob data for the batch data.
 const BlobHashesHeaderFlag byte = L1AuthenticatedMessageHeaderFlag | 0x10 // 0x50
 
+const EigenDAMessageHeaderFlag byte = 0xed
+
 // BrotliMessageHeaderByte indicates that the message is brotli-compressed.
 const BrotliMessageHeaderByte byte = 0
 
 // KnownHeaderBits is all header bits with known meaning to this nitro version
-const KnownHeaderBits byte = DASMessageHeaderFlag | TreeDASMessageHeaderFlag | L1AuthenticatedMessageHeaderFlag | ZeroheavyMessageHeaderFlag | BlobHashesHeaderFlag | BrotliMessageHeaderByte
+const KnownHeaderBits byte = DASMessageHeaderFlag | TreeDASMessageHeaderFlag | L1AuthenticatedMessageHeaderFlag | ZeroheavyMessageHeaderFlag | BlobHashesHeaderFlag | BrotliMessageHeaderByte | EigenDAMessageHeaderFlag
 
 // hasBits returns true if `checking` has all `bits`
 func hasBits(checking byte, bits byte) bool {

arbstate/daprovider/writer.go

@@ -11,6 +11,7 @@ import (
 )
 
 type Writer interface {
+	Type() string
 	// Store posts the batch data to the invoking DA provider
 	// And returns sequencerMsg which is later used to retrieve the batch data
 	Store(
@@ -45,3 +46,7 @@ func (d *writerForDAS) Store(ctx context.Context, message []byte, timeout uint64
 		return Serialize(cert), nil
 	}
 }
+
+func (d *writerForDAS) Type() string {
+	return "DAS"
+}

arbstate/inbox.go

@@ -7,13 +7,13 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
-	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
 
@@ -63,11 +63,7 @@ func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash
 		segments:             [][]byte{},
 	}
 
-	log.Info("Reading calldata payload from sequencer inbox", "calldata", hexutil.Encode(data))
 	payload := data[40:]
-	log.Info("Inbox parse sequencer message: ", "payload", hex.EncodeToString(payload))
-	log.Info("Inbox parse header message: ", "header", hex.EncodeToString(data[:40]))
-	log.Info("Parsed header", "struct", fmt.Sprintf("%+v", parsedMsg))
 
 	// Stage 0: Check if our node is out of date and we don't understand this batch type
 	// If the parent chain sequencer inbox smart contract authenticated this batch,
@@ -87,6 +83,7 @@ func parseSequencerMessage(ctx context.Context, batchNum uint64, batchBlockHash
 		var err error
 		for _, dapReader := range dapReaders {
 			if dapReader != nil && dapReader.IsValidHeaderByte(payload[0]) {
+				println(fmt.Sprintf("Data being sent for recovery: %s", hexutil.Encode(payload)))
 				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

cmd/replay/db.go

@@ -35,6 +35,8 @@ func (db PreimageDb) Get(key []byte) ([]byte, error) {
 	} else {
 		return nil, fmt.Errorf("preimage DB attempted to access non-hash key %v", hex.EncodeToString(key))
 	}
+
+	println("Reading pre-image", hex.EncodeToString(hash[:]))
 	return wavmio.ResolveTypedPreimage(arbutil.Keccak256PreimageType, hash)
 }
 

cmd/replay/main.go

@@ -124,9 +124,6 @@ func (dasReader *PreimageDASReader) ExpirationPolicy(ctx context.Context) (dapro
 }
 
 type BlobPreimageReader struct{}
-
-type PreimageEigenDAReader struct{}
-
 func (r *BlobPreimageReader) GetBlobs(
 	ctx context.Context,
 	batchBlockHash common.Hash,
@@ -152,18 +149,26 @@ func (r *BlobPreimageReader) Initialize(ctx context.Context) error {
 	return nil
 }
 
+
+type EigenDAPreimageReader struct{}
 // QueryBlob returns the blob for the given cert from the preimage oracle using the hash of the
 // certificate kzg commitment for identifying the preimage.
-func (dasReader *PreimageEigenDAReader) QueryBlob(ctx context.Context, cert *eigenda.EigenDABlobInfo, domain string) ([]byte, error) {
+func (dasReader *EigenDAPreimageReader) QueryBlob(ctx context.Context, cert *eigenda.EigenDABlobInfo, domain string) ([]byte, error) {
 	kzgCommit, err := cert.SerializeCommitment()
 	if err != nil {
 		return nil, err
 	}
+
+	println("kzgCommit: ", hex.EncodeToString(kzgCommit))
 	shaDataHash := sha256.New()
 	shaDataHash.Write(kzgCommit)
 	dataHash := shaDataHash.Sum([]byte{})
 	dataHash[0] = 1
-	preimage, err := wavmio.ResolveTypedPreimage(arbutil.EigenDaPreimageType, common.BytesToHash(dataHash))
+
+	hash := common.BytesToHash(dataHash)
+	println("Querying blob for hash: ", hash.String())
+
+	preimage, err := wavmio.ResolveTypedPreimage(arbutil.EigenDaPreimageType, hash)
 	if err != nil {
 		return nil, err
 	}
@@ -178,6 +183,8 @@ func (dasReader *PreimageEigenDAReader) QueryBlob(ctx context.Context, cert *eig
 	return decodedBlob, nil
 }
 
+
+
 // To generate:
 // key, _ := crypto.HexToECDSA("0000000000000000000000000000000000000000000000000000000000000001")
 // sig, _ := crypto.Sign(make([]byte, 32), key)
@@ -234,10 +241,11 @@ func main() {
 			delayedMessagesRead = lastBlockHeader.Nonce.Uint64()
 		}
 		var dasReader daprovider.DASReader
+		var eigenDAReader *EigenDAPreimageReader
 		if dasEnabled {
 			dasReader = &PreimageDASReader{}
 		} else if eigenDAEnabled {
-			eigenDAReader = &PreimageEigenDAReader{}
+			eigenDAReader = &EigenDAPreimageReader{}
 		}
 		backend := WavmInbox{}
 		var keysetValidationMode = daprovider.KeysetPanicIfInvalid
@@ -248,6 +256,10 @@ func main() {
 		if dasReader != nil {
 			dapReaders = append(dapReaders, daprovider.NewReaderForDAS(dasReader))
 		}
+		if eigenDAReader != nil {
+			dapReaders = append(dapReaders, eigenda.NewReaderForEigenDA(eigenDAReader))
+		}
+
 		dapReaders = append(dapReaders, daprovider.NewReaderForBlobReader(&BlobPreimageReader{}))
 		inboxMultiplexer := arbstate.NewInboxMultiplexer(backend, delayedMessagesRead, dapReaders, keysetValidationMode)
 		ctx := context.Background()
@@ -301,7 +313,7 @@ func main() {
 			}
 		}
 
-		message := readMessage(chainConfig.ArbitrumChainParams.DataAvailabilityCommittee, chainConfig.ArbitrumChainParams.EigenDA)
+		message := readMessage(false, true)
 
 		chainContext := WavmChainContext{}
 		batchFetcher := func(batchNum uint64) ([]byte, error) {