feat(taiko-client): introduce payloadQueue for P2P gossip messages (#19195)

Co-authored-by: maskpp <[email protected]>
Co-authored-by: Gavin Yu <[email protected]>
Co-authored-by: Jeffery Walsh <[email protected]>
Co-authored-by: jeff <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 5 people

packages/taiko-client/bindings/encoding/struct.go

@@ -150,7 +150,8 @@ func TransitionContestedEventToV2(
 	}
 }
 
-// BlockVerifiedEventToV2 converts a *ontakeBindings.OntakeClients.TaikoL1ClientBlockVerified to *ontakeBindings.OntakeClients.TaikoL1ClientBlockVerifiedV2.
+// BlockVerifiedEventToV2 converts a *ontakeBindings.OntakeClients.TaikoL1ClientBlockVerified
+// to *ontakeBindings.OntakeClients.TaikoL1ClientBlockVerifiedV2.
 func BlockVerifiedEventToV2(e *ontakeBindings.TaikoL1ClientBlockVerified) *ontakeBindings.TaikoL1ClientBlockVerifiedV2 {
 	return &ontakeBindings.TaikoL1ClientBlockVerifiedV2{
 		BlockId:   e.BlockId,

packages/taiko-client/driver/chain_syncer/blob/blocks_inserter/common.go

@@ -449,7 +449,6 @@ func assembleCreateExecutionPayloadMetaPacaya(
 	}, anchorTx, nil
 }
 
-// updateL1OriginForBatch updates the each block's L1 origin in the given batch.
 func updateL1OriginForBatch(
 	ctx context.Context,
 	rpc *rpc.Client,
@@ -466,26 +465,32 @@ func updateL1OriginForBatch(
 
 	for i := 0; i < len(meta.GetBlocks()); i++ {
 		g.Go(func() error {
-			blockID := new(big.Int).SetUint64(meta.GetLastBlockID() - uint64(i))
+			blockID := new(big.Int).SetUint64(meta.GetLastBlockID() - uint64(len(meta.GetBlocks())-1-i))
+
 			header, err := rpc.L2.HeaderByNumber(ctx, blockID)
 			if err != nil {
 				return fmt.Errorf("failed to get block by number %d: %w", blockID, err)
 			}
+
 			l1Origin := &rawdb.L1Origin{
 				BlockID:       blockID,
 				L2BlockHash:   header.Hash(),
 				L1BlockHeight: meta.GetRawBlockHeight(),
 				L1BlockHash:   meta.GetRawBlockHash(),
 			}
-			// Update the l1Origin and headL1Origin cursor for that preconfirmed block.
+
 			if _, err := rpc.L2Engine.UpdateL1Origin(ctx, l1Origin); err != nil {
 				return fmt.Errorf("failed to update L1 origin: %w", err)
 			}
+
+			// If this is the most recent block, update the HeadL1Origin.
 			if i == len(meta.GetBlocks())-1 {
+				log.Info("Update head L1 origin", "blockID", blockID, "l1Origin", l1Origin)
 				if _, err := rpc.L2Engine.SetHeadL1Origin(ctx, l1Origin.BlockID); err != nil {
 					return fmt.Errorf("failed to write head L1 origin: %w", err)
 				}
 			}
+
 			return nil
 		})
 	}

packages/taiko-client/driver/chain_syncer/blob/blocks_inserter/pacaya.go

@@ -195,15 +195,24 @@ func (i *BlocksInserterPacaya) InsertBlocks(
 
 		log.Debug("Payload data", "hash", lastPayloadData.BlockHash, "txs", len(lastPayloadData.Transactions))
 
+		// Wait till the corresponding L2 header to be existed in the L2 EE.
+		if _, err := i.rpc.WaitL2Header(ctx, new(big.Int).SetUint64(lastPayloadData.Number)); err != nil {
+			return fmt.Errorf("failed to wait for L2 header (%d): %w", lastPayloadData.Number, err)
+		}
+
 		log.Info(
 			"🔗 New L2 block inserted",
 			"blockID", lastPayloadData.Number,
 			"hash", lastPayloadData.BlockHash,
+			"coinbase", lastPayloadData.FeeRecipient.Hex(),
 			"transactions", len(lastPayloadData.Transactions),
 			"timestamp", lastPayloadData.Timestamp,
 			"baseFee", utils.WeiToGWei(lastPayloadData.BaseFeePerGas),
 			"withdrawals", len(lastPayloadData.Withdrawals),
 			"batchID", meta.GetBatchID(),
+			"gasLimit", lastPayloadData.GasLimit,
+			"gasUsed", lastPayloadData.GasUsed,
+			"parentHash", lastPayloadData.ParentHash,
 			"indexInBatch", j,
 		)
 
@@ -213,14 +222,31 @@ func (i *BlocksInserterPacaya) InsertBlocks(
 	return nil
 }
 
-// InsertPreconfBlockFromExecutionPayload inserts a preconf block from the given execution payload.
-func (i *BlocksInserterPacaya) InsertPreconfBlockFromExecutionPayload(
+// InsertPreconfBlocksFromExecutionPayloads inserts preconf blocks from the given execution payloads.
+func (i *BlocksInserterPacaya) InsertPreconfBlocksFromExecutionPayloads(
 	ctx context.Context,
-	executableData *eth.ExecutionPayload,
-) (*types.Header, error) {
+	executionPayloads []*eth.ExecutionPayload,
+) ([]*types.Header, error) {
 	i.mutex.Lock()
 	defer i.mutex.Unlock()
 
+	headers := make([]*types.Header, len(executionPayloads))
+	for j, executableData := range executionPayloads {
+		header, err := i.insertPreconfBlockFromExecutionPayload(ctx, executableData)
+		if err != nil {
+			return nil, fmt.Errorf("failed to insert preconf block: %w", err)
+		}
+		headers[j] = header
+	}
+
+	return headers, nil
+}
+
+// insertPreconfBlockFromExecutionPayload the inner method to insert a preconf block from the given execution payload.
+func (i *BlocksInserterPacaya) insertPreconfBlockFromExecutionPayload(
+	ctx context.Context,
+	executableData *eth.ExecutionPayload,
+) (*types.Header, error) {
 	// Ensure the preconfirmation block number is greater than the current head L1 origin block ID.
 	headL1Origin, err := i.rpc.L2.HeadL1Origin(ctx)
 	if err != nil && err.Error() != ethereum.NotFound.Error() {
@@ -231,7 +257,7 @@ func (i *BlocksInserterPacaya) InsertPreconfBlockFromExecutionPayload(
 	if headL1Origin != nil {
 		if uint64(executableData.BlockNumber) <= headL1Origin.BlockID.Uint64() {
 			return nil, fmt.Errorf(
-				"preconfirmation block number (%d) is less than or equal to the current head L1 origin block ID (%d)",
+				"preconfirmation block ID (%d) is less than or equal to the current head L1 origin block ID (%d)",
 				executableData.BlockNumber,
 				headL1Origin.BlockID,
 			)
@@ -242,6 +268,11 @@ func (i *BlocksInserterPacaya) InsertPreconfBlockFromExecutionPayload(
 		return nil, fmt.Errorf("no transactions data in the payload")
 	}
 
+	// Decompress the transactions list.
+	if executableData.Transactions[0], err = utils.DecompressPacaya(executableData.Transactions[0]); err != nil {
+		return nil, fmt.Errorf("failed to decompress transactions list bytes: %w", err)
+	}
+
 	var u256BaseFee = uint256.Int(executableData.BaseFeePerGas)
 	payload, err := createExecutionPayloadsAndSetHead(
 		ctx,

packages/taiko-client/driver/chain_syncer/chain_syncer.go

@@ -47,13 +47,7 @@ func New(
 	go tracker.Track(ctx)
 
 	beaconSyncer := beaconsync.NewSyncer(ctx, rpc, state, tracker)
-	blobSyncer, err := blob.NewSyncer(
-		ctx,
-		rpc,
-		state,
-		tracker,
-		blobServerEndpoint,
-	)
+	blobSyncer, err := blob.NewSyncer(ctx, rpc, state, tracker, blobServerEndpoint)
 	if err != nil {
 		return nil, err
 	}

packages/taiko-client/driver/driver.go

@@ -117,6 +117,7 @@ func (d *Driver) InitFromConfig(ctx context.Context, cfg *Config) (err error) {
 		if d.preconfBlockServer, err = preconfBlocks.New(
 			d.PreconfBlockServerCORSOrigins,
 			d.PreconfBlockServerJWTSecret,
+			d.TaikoL2Address,
 			d.l2ChainSyncer.BlobSyncer().BlocksInserterPacaya(),
 			d.rpc,
 		); err != nil {