feat: Add apis for cell staging

beacon/engine/errors.go

@@ -76,6 +76,7 @@ var (
 
 	GenericServerError       = &EngineAPIError{code: -32000, msg: "Server error"}
 	UnknownPayload           = &EngineAPIError{code: -38001, msg: "Unknown payload"}
+	UnknownPrediction        = &EngineAPIError{code: -38001, msg: "Unknown prediction"}
 	InvalidForkChoiceState   = &EngineAPIError{code: -38002, msg: "Invalid forkchoice state"}
 	InvalidPayloadAttributes = &EngineAPIError{code: -38003, msg: "Invalid payload attributes"}
 	TooLargeRequest          = &EngineAPIError{code: -38004, msg: "Too large request"}

beacon/engine/types.go

@@ -128,6 +128,14 @@ type BlobAndProofV2 struct {
 	CellProofs []hexutil.Bytes `json:"proofs"`
 }
 
+type BlobPredictionToStage struct {
+	TxHash        common.Hash     `json:"txHash"`    //32
+	BlobIndex     uint            `json:"blobIndex"` //4
+	Blob          hexutil.Bytes   `json:"blob"`      //
+	KzgCommitment hexutil.Bytes   `json:"kzgCommitment"`
+	CellProofs    []hexutil.Bytes `json:"cellProofs"`
+}
+
 // JSON type overrides for ExecutionPayloadEnvelope.
 type executionPayloadEnvelopeMarshaling struct {
 	BlockValue *hexutil.Big
@@ -176,11 +184,34 @@ func (b *PayloadID) UnmarshalText(input []byte) error {
 	return nil
 }
 
+// PayloadID is an identifier of the blob prediction process
+type PredictionID [8]byte
+
+func (b PredictionID) String() string {
+	return hexutil.Encode(b[:])
+}
+
+func (b PredictionID) MarshalText() ([]byte, error) {
+	return hexutil.Bytes(b[:]).MarshalText()
+}
+
+func (b *PredictionID) UnmarshalText(input []byte) error {
+	err := hexutil.UnmarshalFixedText("PredictionID", input, b[:])
+	if err != nil {
+		return fmt.Errorf("invalid prediction id %q: %w", input, err)
+	}
+	return nil
+}
+
 type ForkChoiceResponse struct {
 	PayloadStatus PayloadStatusV1 `json:"payloadStatus"`
 	PayloadID     *PayloadID      `json:"payloadId"`
 }
 
+type PredictionResponse struct {
+	PredictionID *PredictionID `json:"predictionId"`
+}
+
 type ForkchoiceStateV1 struct {
 	HeadBlockHash      common.Hash `json:"headBlockHash"`
 	SafeBlockHash      common.Hash `json:"safeBlockHash"`

consensus/misc/eip4844/eip4844.go

@@ -103,6 +103,15 @@ func CalcBlobFee(config *params.ChainConfig, header *types.Header) *big.Int {
 	return fakeExponential(minBlobGasPrice, new(big.Int).SetUint64(*header.ExcessBlobGas), new(big.Int).SetUint64(blobConfig.UpdateFraction))
 }
 
+// CalcBlobFeeWithoutHeader calculates the blobfee from given excess blob gas.
+func CalcBlobFeeWithoutHeader(config *params.ChainConfig, excessBlobGas uint64, timestamp uint64) *big.Int {
+	blobConfig := latestBlobConfig(config, timestamp)
+	if blobConfig == nil {
+		panic("calculating blob fee on unsupported fork")
+	}
+	return fakeExponential(minBlobGasPrice, new(big.Int).SetUint64(excessBlobGas), new(big.Int).SetUint64(blobConfig.UpdateFraction))
+}
+
 // MaxBlobsPerBlock returns the max blobs per block for a block at the given timestamp.
 func MaxBlobsPerBlock(cfg *params.ChainConfig, time uint64) int {
 	blobConfig := latestBlobConfig(cfg, time)

eth/catalyst/api.go

@@ -19,8 +19,10 @@ package catalyst
 
 import (
 	"crypto/sha256"
+	"encoding/binary"
 	"errors"
 	"fmt"
+	"math/rand"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -98,6 +100,8 @@ var caps = []string{
 	"engine_getPayloadV5",
 	"engine_getBlobsV1",
 	"engine_getBlobsV2",
+	"engine_getBlobsToStage",
+	"engine_notifyPrediction",
 	"engine_newPayloadV1",
 	"engine_newPayloadV2",
 	"engine_newPayloadV3",
@@ -133,6 +137,7 @@ type ConsensusAPI struct {
 
 	remoteBlocks *headerQueue  // Cache of remote payloads received
 	localBlocks  *payloadQueue // Cache of local payloads generated
+	predictions  *predictionQueue
 
 	// The forkchoice update and new payload method require us to return the
 	// latest valid hash in an invalid chain. To support that return, we need
@@ -185,6 +190,7 @@ func newConsensusAPIWithoutHeartbeat(eth *eth.Ethereum) *ConsensusAPI {
 		eth:               eth,
 		remoteBlocks:      newHeaderQueue(),
 		localBlocks:       newPayloadQueue(),
+		predictions:       newPredictionQueue(),
 		invalidBlocksHits: make(map[common.Hash]int),
 		invalidTipsets:    make(map[common.Hash]*types.Header),
 	}
@@ -402,6 +408,34 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
 	return valid(nil), nil
 }
 
+// todo(healthykim) window / max value configuration
+// todo(healthykim) blob ID calculation logic
+func (api *ConsensusAPI) NotifyPrediction(headBlockRoot common.Hash) (engine.PredictionResponse, error) {
+	max := uint(10)
+	window := uint(2)
+	timestamp := uint64(time.Now().Unix())
+	random := rand.Int()
+	buf := make([]byte, 8)
+	binary.LittleEndian.PutUint64(buf, uint64(random))
+
+	var predictionID engine.PredictionID
+	hasher := sha256.New()
+	hasher.Write(buf)
+	hasher.Write(headBlockRoot[:])
+	binary.Write(hasher, binary.BigEndian, timestamp)
+	copy(predictionID[:], hasher.Sum(nil)[:8])
+
+	// Timestamp setting logic - refer to prepareWork
+	prediction, _ := api.eth.Miner().PredictBlobTxs(predictionID, max, window, timestamp)
+	api.predictions.put(predictionID, prediction)
+
+	log.Debug("Prediction started for ", "predictionId", predictionID, "slot", headBlockRoot.String())
+
+	return engine.PredictionResponse{
+		PredictionID: &predictionID,
+	}, nil
+}
+
 // ExchangeTransitionConfigurationV1 checks the given configuration against
 // the configuration of the node.
 func (api *ConsensusAPI) ExchangeTransitionConfigurationV1(config engine.TransitionConfigurationV1) (*engine.TransitionConfigurationV1, error) {
@@ -575,6 +609,19 @@ func (api *ConsensusAPI) GetBlobsV2(hashes []common.Hash) ([]*engine.BlobAndProo
 	return res, nil
 }
 
+func (api *ConsensusAPI) GetBlobsToStage(id engine.PredictionID) ([]*engine.BlobPredictionToStage, error) {
+
+	prediction := api.predictions.get(id)
+	log.Info("Finding prediction result for ", "id=", id)
+
+	if prediction == nil {
+		log.Error("There is no prediction with given id", "prediction id", id)
+		return nil, engine.UnknownPrediction
+	}
+
+	return prediction.Convert(), nil
+}
+
 // Helper for NewPayload* methods.
 var invalidStatus = engine.PayloadStatusV1{Status: engine.INVALID}
 

eth/catalyst/queue.go

@@ -156,3 +156,66 @@ func (q *headerQueue) get(hash common.Hash) *types.Header {
 	}
 	return nil
 }
+
+const maxTrackedPredictions = 10
+
+// This is basically same with payloadQueue but to track prediction
+type predictionQueueItem struct {
+	id         engine.PredictionID
+	prediction *miner.BlobPrediction
+}
+
+type predictionQueue struct {
+	predictions []*predictionQueueItem
+	lock        sync.RWMutex
+}
+
+func newPredictionQueue() *predictionQueue {
+	return &predictionQueue{
+		predictions: make([]*predictionQueueItem, maxTrackedPredictions),
+	}
+}
+
+// put inserts a new payload into the queue at the given id.
+func (q *predictionQueue) put(id engine.PredictionID, prediction *miner.BlobPrediction) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	copy(q.predictions[1:], q.predictions)
+	q.predictions[0] = &predictionQueueItem{
+		id:         id,
+		prediction: prediction,
+	}
+}
+
+// get retrieves a previously stored payload item or nil if it does not exist.
+func (q *predictionQueue) get(id engine.PredictionID) *miner.BlobPrediction {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+
+	for _, item := range q.predictions {
+		if item == nil {
+			return nil // no more items
+		}
+		if item.id == id {
+			return item.prediction
+		}
+	}
+	return nil
+}
+
+// has checks if a particular payload is already tracked.
+func (q *predictionQueue) has(id engine.PredictionID) bool {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+
+	for _, item := range q.predictions {
+		if item == nil {
+			return false
+		}
+		if item.id == id {
+			return true
+		}
+	}
+	return false
+}

miner/miner.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/beacon/engine"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/consensus"
@@ -138,6 +139,11 @@ func (miner *Miner) BuildPayload(args *BuildPayloadArgs, witness bool) (*Payload
 	return miner.buildPayload(args, witness)
 }
 
+// BuildPayload builds the payload according to the provided parameters.
+func (miner *Miner) PredictBlobTxs(blobId engine.PredictionID, max uint, window uint, timestamp uint64) (*BlobPrediction, error) {
+	return miner.predictBlobs(blobId, max, window, timestamp)
+}
+
 // getPending retrieves the pending block based on the current head block.
 // The result might be nil if pending generation is failed.
 func (miner *Miner) getPending() *newPayloadResult {

miner/prediction.go

@@ -0,0 +1,182 @@
+// Copyright 2025 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package miner
+
+import (
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/beacon/engine"
+	"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
+	"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
+	"github.com/ethereum/go-ethereum/core/txpool"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/holiman/uint256"
+)
+
+type BlobPrediction struct {
+	id          engine.PredictionID // prediction is identified by its id
+	transaction []*types.Transaction
+	lock        sync.Mutex
+	stop        chan struct{} // closed before resolving
+}
+
+func (prediction *BlobPrediction) Convert() []*engine.BlobPredictionToStage {
+	prediction.lock.Lock()
+	defer prediction.lock.Unlock()
+
+	var res []*engine.BlobPredictionToStage
+	for _, tx := range prediction.transaction {
+		for blobIdx, blob := range tx.BlobTxSidecar().Blobs {
+			r := engine.BlobPredictionToStage{
+				TxHash: tx.Hash(),
+			}
+			r.Blob = blob[:]
+			r.BlobIndex = uint(blobIdx)
+			r.KzgCommitment = tx.BlobTxSidecar().Commitments[blobIdx][:]
+			proofs := tx.BlobTxSidecar().CellProofsAt(blobIdx)
+			for _, proof := range proofs {
+				r.CellProofs = append(r.CellProofs, proof[:])
+			}
+			res = append(res, &r)
+		}
+	}
+
+	return res
+}
+
+// Refer to fillTransaction
+// parent:
+// len: maximum # of blobs
+// header: For gas fee calculation
+// [Main Difference from fillTransactions]
+// - No prio/normal transaction
+func (miner *Miner) fillBlobs(blobId engine.PredictionID, max uint, W uint, timestamp uint64) ([]*types.Transaction, error) {
+	miner.confMu.RLock()
+	tip := miner.config.GasPrice
+	miner.confMu.RUnlock()
+
+	// Refer to `prepareWork`
+	// We don't need to reuse prepareWork itself as we need only gas-related fields
+	parent := miner.chain.CurrentBlock()
+
+	// Assume that other miners also have similar tip(minimum gas price) value
+	filter := txpool.PendingFilter{
+		MinTip: uint256.MustFromBig(tip),
+	}
+	// Predict base fee under the assumption that market will go down as much as possible
+	predictedBaseFee := eip1559.CalcBaseFee(miner.chainConfig, parent)
+	multiplier := new(big.Int).Exp(big.NewInt(875), new(big.Int).SetUint64(uint64(W)), nil)
+	divisor := new(big.Int).Exp(big.NewInt(1000), new(big.Int).SetUint64(uint64(W)), nil)
+
+	predictedBaseFee.Mul(predictedBaseFee, multiplier)
+	predictedBaseFee.Div(predictedBaseFee, divisor)
+	if predictedBaseFee.Cmp(new(big.Int).SetUint64(1)) <= 0 {
+		predictedBaseFee = new(big.Int).SetUint64(1)
+	}
+	// Predict blob fee under the assumption that market will go down as much as possible
+	// todo(healthykim): EIP-7918
+	excessBlobGas := eip4844.CalcExcessBlobGas(miner.chainConfig, parent, timestamp)
+	// CalcBlobFeeWithoutHeader is new function to calculate blob fee without loading all header info
+	predictedBlobFee := eip4844.CalcBlobFeeWithoutHeader(miner.chainConfig, excessBlobGas, timestamp)
+
+	exp := new(big.Int).SetUint64(uint64(W))
+	multiplier = new(big.Int).Exp(big.NewInt(1000), exp, nil)
+	divisor = new(big.Int).Exp(big.NewInt(1125), exp, nil)
+
+	predictedBlobFee = new(big.Int).Mul(predictedBlobFee, multiplier)
+	predictedBlobFee.Div(predictedBlobFee, divisor)
+	if predictedBlobFee.Cmp(new(big.Int).SetUint64(1)) <= 0 {
+		predictedBlobFee = new(big.Int).SetUint64(1)
+	}
+
+	// Refer to fillTransaction
+	filter.BaseFee = uint256.MustFromBig(predictedBaseFee)
+	filter.BlobFee = uint256.MustFromBig(predictedBlobFee)
+	filter.OnlyPlainTxs, filter.OnlyBlobTxs = false, true
+	pendingBlobTxs := miner.txpool.Pending(filter)
+
+	// Fill the block with all available pending transactions.
+	res := make([]*types.Transaction, 0, max)
+
+	if len(pendingBlobTxs) > 0 {
+		// Refer to newTransactionsByPriceAndNonce
+		// signer는 사용하지 않음 / 원래는 commitTransaction 넘겨주는용이지만 여기서는 무시
+		signer := types.MakeSigner(miner.chainConfig, parent.Number, parent.Time)
+		blobTxs := newTransactionsByPriceAndNonce(signer, pendingBlobTxs, predictedBaseFee)
+		for {
+			bltx, _ := blobTxs.Peek()
+			// Resolve to check if we have full blob data
+			// Can be changed
+			btx := bltx.Resolve()
+
+			if btx != nil {
+				res = append(res, btx)
+			}
+
+			blobTxs.Pop()
+
+			if len(res) >= int(max) || len(blobTxs.heads) == 0 {
+				break
+			}
+		}
+	}
+	return res, nil
+}
+
+func (miner *Miner) predictBlobs(blobId engine.PredictionID, max uint, W uint, timestamp uint64) (*BlobPrediction, error) {
+	prediction := &BlobPrediction{
+		id:          blobId,
+		transaction: make([]*types.Transaction, 0, max),
+		stop:        make(chan struct{}),
+	}
+
+	go func() {
+		timer := time.NewTimer(0)
+		defer timer.Stop()
+
+		// Prediction environment changes for every new slot
+		endTimer := time.NewTimer(time.Second * 12)
+
+		for {
+			select {
+			case <-timer.C:
+				res, err := miner.fillBlobs(blobId, max, W, timestamp)
+				if err == nil {
+					// update prediction
+					prediction.lock.Lock()
+					prediction.transaction = res
+					prediction.lock.Unlock()
+					log.Info("Prediction result", "res", prediction)
+				} else {
+					log.Info("Error while generating prediction", "id", prediction.id, "err", err)
+				}
+				timer.Reset(miner.config.Recommit)
+			case <-prediction.stop:
+				log.Info("Stopping work on prediction", "id", prediction.id, "reason", "delivery")
+				return
+			case <-endTimer.C:
+				log.Info("Stopping work on prediction", "id", prediction.id, "reason", "timeout")
+				return
+			}
+		}
+	}()
+
+	return prediction, nil
+}