txDAG transfer (bnb-chain#28)

* txDAG transfer

* set flag of txDAG transaction to 'no dependency'

* encode/decode txDAG data with ABI

* set enable flag for txdag

* set txDAG receiver to a special address

* remove invalid flags

---------

Co-authored-by: andyzhang2023 <[email protected]>

cmd/geth/main.go

@@ -174,6 +174,7 @@ var (
 		utils.ParallelTxNumFlag,
 		utils.ParallelTxDAGFlag,
 		utils.ParallelTxDAGFileFlag,
+		utils.ParallelTxDAGSenderPrivFlag,
 		configFileFlag,
 		utils.LogDebugFlag,
 		utils.LogBacktraceAtFlag,

cmd/utils/flags.go

@@ -1130,6 +1130,13 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Usage:    "enable opcode optimization",
 		Category: flags.VMCategory,
 	}
+
+	ParallelTxDAGSenderPrivFlag = &cli.StringFlag{
+		Name:     "parallel.txdagsenderpriv",
+		Usage:    "private key of the sender who sends the TxDAG transactions",
+		Value:    "",
+		Category: flags.VMCategory,
+	}
 )
 
 var (
@@ -2044,6 +2051,13 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 		cfg.ParallelTxDAGFile = ctx.String(ParallelTxDAGFileFlag.Name)
 	}
 
+	if ctx.IsSet(ParallelTxDAGSenderPrivFlag.Name) {
+		priHex := ctx.String(ParallelTxDAGSenderPrivFlag.Name)
+		if cfg.Miner.ParallelTxDAGSenderPriv, err = crypto.HexToECDSA(priHex); err != nil {
+			Fatalf("Failed to parse txdag private key of %s, err: %v", ParallelTxDAGSenderPrivFlag.Name, err)
+		}
+	}
+
 	if ctx.IsSet(VMOpcodeOptimizeFlag.Name) {
 		cfg.EnableOpcodeOptimizing = ctx.Bool(VMOpcodeOptimizeFlag.Name)
 		if cfg.EnableOpcodeOptimizing {

core/blockchain.go

@@ -2835,6 +2835,10 @@ func (bc *BlockChain) TxDAGEnabled() bool {
 	return bc.enableTxDAG
 }
 
+func (bc *BlockChain) TxDAGFileOpened() bool {
+	return bc.txDAGWriteCh != nil
+}
+
 func (bc *BlockChain) SetupTxDAGGeneration(output string, readFile bool) {
 	log.Info("node enable TxDAG feature", "output", output)
 	bc.enableTxDAG = true

core/parallel_state_processor.go

@@ -736,10 +736,17 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat
 		txDAG types.TxDAG
 	)
 	if p.bc.enableTxDAG {
-		// TODO(galaio): load TxDAG from block
-		// or load cache txDAG from file
-		if txDAG == nil && p.bc.txDAGReader != nil {
+		var err error
+		if p.bc.txDAGReader != nil {
+			// load cache txDAG from file first
 			txDAG = p.bc.txDAGReader.TxDAG(block.NumberU64())
+
+		} else {
+			// load TxDAG from block
+			txDAG, err = types.GetTxDAG(block)
+			if err != nil {
+				log.Debug("pevm decode txdag failed", "block", block.NumberU64(), "err", err)
+			}
 		}
 		if err := types.ValidateTxDAG(txDAG, len(block.Transactions())); err != nil {
 			log.Warn("pevm cannot apply wrong txdag",

core/types/dag.go

@@ -7,11 +7,41 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/rlp"
 	"golang.org/x/exp/slices"
 )
 
+const TxDAGAbiJson = `
+[
+  {
+    "type": "function",
+    "name": "setTxDAG",
+    "inputs": [
+      {
+        "name": "data",
+        "type": "bytes",
+        "internalType": "bytes"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
+  }
+]
+`
+
+var TxDAGABI abi.ABI
+
+func init() {
+	var err error
+	// must be able to register the TxDAGABI
+	TxDAGABI, err = abi.JSON(strings.NewReader(TxDAGAbiJson))
+	if err != nil {
+		panic(err)
+	}
+}
+
 // TxDAGType Used to extend TxDAG and customize a new DAG structure
 const (
 	EmptyTxDAGType byte = iota
@@ -48,6 +78,37 @@ type TxDAG interface {
 	SetTxDep(int, TxDep) error
 }
 
+func DecodeTxDAGCalldata(data []byte) (TxDAG, error) {
+	// trim the method id before unpack
+	if len(data) < 4 {
+		return nil, fmt.Errorf("invalid txDAG calldata, len(data)=%d", len(data))
+	}
+	calldata, err := TxDAGABI.Methods["setTxDAG"].Inputs.Unpack(data[4:])
+	if err != nil {
+		return nil, fmt.Errorf("failed to call abi unpack, err: %v", err)
+	}
+	if len(calldata) <= 0 {
+		return nil, fmt.Errorf("invalid txDAG calldata, len(calldata)=%d", len(calldata))
+	}
+	data, ok := calldata[0].([]byte)
+	if !ok {
+		return nil, fmt.Errorf("invalid txDAG calldata parameter")
+	}
+	return DecodeTxDAG(data)
+}
+
+func EncodeTxDAGCalldata(dag TxDAG) ([]byte, error) {
+	data, err := EncodeTxDAG(dag)
+	if err != nil {
+		return nil, fmt.Errorf("failed to encode txDAG, err: %v", err)
+	}
+	data, err = TxDAGABI.Pack("setTxDAG", data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to call abi pack, err: %v", err)
+	}
+	return data, nil
+}
+
 func EncodeTxDAG(dag TxDAG) ([]byte, error) {
 	if dag == nil {
 		return nil, errors.New("input nil TxDAG")
@@ -118,6 +179,17 @@ func ValidatePlainTxDAG(d TxDAG, txCnt int) error {
 	return nil
 }
 
+// GetTxDAG return TxDAG bytes from block if there is any, or return nil if not exist
+// the txDAG is stored in the calldata of the last transaction of the block
+func GetTxDAG(block *Block) (TxDAG, error) {
+	txs := block.Transactions()
+	if txs.Len() <= 0 {
+		return nil, fmt.Errorf("no txdag found")
+	}
+	// get data from the last tx
+	return DecodeTxDAGCalldata(txs[txs.Len()-1].Data())
+}
+
 func TxDependency(d TxDAG, i int) []uint64 {
 	if d == nil || i < 0 || i >= d.TxCount() {
 		return []uint64{}

core/types/dag_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/cometbft/cometbft/libs/rand"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -16,6 +17,19 @@ var (
 	mockHash = common.HexToHash("0xdc13f8d7bdb8ec4de02cd4a50a1aa2ab73ec8814e0cdb550341623be3dd8ab7a")
 )
 
+func TestEncodeTxDAGCalldata(t *testing.T) {
+	tg := mockSimpleDAG()
+	data, err := EncodeTxDAGCalldata(tg)
+	assert.Equal(t, nil, err)
+	tg, err = DecodeTxDAGCalldata(data)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, tg.TxDep(6).TxIndexes[0], uint64(2))
+	assert.Equal(t, tg.TxDep(6).TxIndexes[1], uint64(5))
+
+	_, err = DecodeTxDAGCalldata(nil)
+	assert.NotEqual(t, nil, err)
+}
+
 func TestTxDAG_SetTxDep(t *testing.T) {
 	dag := mockSimpleDAG()
 	require.NoError(t, dag.SetTxDep(9, NewTxDep(nil, NonDependentRelFlag)))

miner/miner.go

@@ -19,6 +19,7 @@ package miner
 
 import (
 	"context"
+	"crypto/ecdsa"
 	"errors"
 	"fmt"
 	"math/big"
@@ -109,6 +110,8 @@ type Config struct {
 	EffectiveGasCeil          uint64 // if non-zero, a gas ceiling to apply independent of the header's gaslimit value
 
 	Mev MevConfig // Mev configuration
+
+	ParallelTxDAGSenderPriv *ecdsa.PrivateKey // The private key for the parallel tx DAG sender
 }
 
 // DefaultConfig contains default settings for miner.

miner/worker.go

@@ -18,6 +18,7 @@ package miner
 
 import (
 	"context"
+	"crypto/ecdsa"
 	"errors"
 	"fmt"
 	"math/big"
@@ -40,6 +41,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/txpool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth/tracers"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
@@ -102,6 +104,10 @@ var (
 	txErrReplayMeter           = metrics.NewRegisteredMeter("miner/tx/replay", nil)
 )
 
+var (
+	DefaultTxDAGAddress = common.HexToAddress("0xda90000000000000000000000000000000000000")
+)
+
 // environment is the worker's current environment and holds all
 // information of the sealing block generation.
 type environment struct {
@@ -916,10 +922,36 @@ func (w *worker) commitTransactions(env *environment, plainTxs, blobTxs *transac
 	}
 	var coalescedLogs []*types.Log
 
+	//append the tx DAG transaction to the block
+	appendTxDAG := func() {
+		// whether enable TxDAG
+		if !w.chain.TxDAGEnabled() {
+			return
+		}
+		// whether export to file
+		if w.chain.TxDAGFileOpened() {
+			return
+		}
+		// TODO this is a placeholder for the tx DAG data that will be generated by the stateDB
+		txForDAG, err := w.generateDAGTx(env.signer, env.tcount, env.coinbase)
+		if err != nil {
+			log.Warn("failed to generate DAG tx", "err", err)
+			return
+		}
+		logs, err := w.commitTransaction(env, txForDAG)
+		if err != nil {
+			log.Warn("failed to commit DAG tx", "err", err)
+			return
+		}
+		coalescedLogs = append(coalescedLogs, logs...)
+		env.tcount++
+	}
+
 	for {
 		// Check interruption signal and abort building if it's fired.
 		if interrupt != nil {
 			if signal := interrupt.Load(); signal != commitInterruptNone {
+				appendTxDAG()
 				return signalToErr(signal)
 			}
 		}
@@ -1018,6 +1050,7 @@ func (w *worker) commitTransactions(env *environment, plainTxs, blobTxs *transac
 			txErrUnknownMeter.Mark(1)
 		}
 	}
+	appendTxDAG()
 	if !w.isRunning() && len(coalescedLogs) > 0 {
 		// We don't push the pendingLogsEvent while we are sealing. The reason is that
 		// when we are sealing, the worker will regenerate a sealing block every 3 seconds.
@@ -1036,6 +1069,59 @@ func (w *worker) commitTransactions(env *environment, plainTxs, blobTxs *transac
 	return nil
 }
 
+// generateDAGTx generates a DAG transaction for the block
+func (w *worker) generateDAGTx(signer types.Signer, txIndex int, coinbase common.Address) (*types.Transaction, error) {
+	statedb, err := w.chain.State()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get state db, err: %v", err)
+	}
+
+	if signer == nil {
+		return nil, fmt.Errorf("current signer is nil")
+	}
+
+	//privateKey, err := crypto.HexToECDSA(privateKeyHex)
+	sender := w.config.ParallelTxDAGSenderPriv
+	receiver := DefaultTxDAGAddress
+	if sender == nil {
+		return nil, fmt.Errorf("missing sender private key")
+	}
+
+	// get txDAG data from the stateDB
+	txDAG, err := statedb.ResolveTxDAG(txIndex, []common.Address{coinbase, params.OptimismBaseFeeRecipient, params.OptimismL1FeeRecipient})
+	if txDAG == nil {
+		return nil, err
+	}
+	// txIndex is the index of this txDAG transaction
+	txDAG.SetTxDep(txIndex, types.TxDep{Flags: &types.NonDependentRelFlag})
+
+	publicKey := sender.Public()
+	publicKeyECDSA, ok := publicKey.(*ecdsa.PublicKey)
+	if !ok {
+		return nil, fmt.Errorf("error casting public key to ECDSA")
+	}
+	fromAddress := crypto.PubkeyToAddress(*publicKeyECDSA)
+
+	// get nonce from the
+	nonce := statedb.GetNonce(fromAddress)
+
+	data, err := types.EncodeTxDAGCalldata(txDAG)
+	if err != nil {
+		return nil, fmt.Errorf("failed to encode txDAG, err: %v", err)
+	}
+
+	// Create the transaction
+	tx := types.NewTransaction(nonce, receiver, big.NewInt(0), 21100, big.NewInt(0), data)
+
+	// Sign the transaction with the private key
+	signedTx, err := types.SignTx(tx, signer, sender)
+	if err != nil {
+		return nil, fmt.Errorf("failed to sign transaction, err: %v", err)
+	}
+
+	return signedTx, nil
+}
+
 // generateParams wraps various of settings for generating sealing task.
 type generateParams struct {
 	timestamp   uint64            // The timestamp for sealing task