diff --git a/beacon/engine/errors.go b/beacon/engine/errors.go index 62773a0ea9..032a0724f0 100644 --- a/beacon/engine/errors.go +++ b/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"} diff --git a/beacon/engine/types.go b/beacon/engine/types.go index 76bfd22a23..17c793541b 100644 --- a/beacon/engine/types.go +++ b/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"` diff --git a/consensus/misc/eip4844/eip4844.go b/consensus/misc/eip4844/eip4844.go index a800f4786d..2706b4aac1 100644 --- a/consensus/misc/eip4844/eip4844.go +++ b/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) diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go index cbbdbbb361..d681c5e798 100644 --- a/eth/catalyst/api.go +++ b/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} diff --git a/eth/catalyst/queue.go b/eth/catalyst/queue.go index 634dc1b2e6..ba2c1315fd 100644 --- a/eth/catalyst/queue.go +++ b/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 +} diff --git a/miner/miner.go b/miner/miner.go index 3cec054d19..b852563fa8 100644 --- a/miner/miner.go +++ b/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 { diff --git a/miner/prediction.go b/miner/prediction.go new file mode 100644 index 0000000000..c6e50af7c3 --- /dev/null +++ b/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 . + +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 +}