fix: bcast messages

app/app.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"maps"
@@ -78,6 +79,7 @@ import (
 	servertypes "github.com/cosmos/cosmos-sdk/server/types"
 	testdata_pulsar "github.com/cosmos/cosmos-sdk/testutil/testdata/testpb"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	"github.com/cosmos/cosmos-sdk/types/msgservice"
 	signingtypes "github.com/cosmos/cosmos-sdk/types/tx/signing"
@@ -1046,29 +1048,60 @@ func (app *TacChainApp) configureEVMMempool(appOpts servertypes.AppOptions, logg
 }
 
 func (app *TacChainApp) broadcastEVMTransactions(ethTxs []*ethtypes.Transaction) error {
+	signer := ethtypes.LatestSigner(evmvmtypes.GetEthChainConfig())
+	baseDenom := evmvmtypes.GetEVMCoinDenom()
+
+	// One bad transaction must not cost the rest of the batch its broadcast, so
+	// failures are collected and reported once at the end.
+	var errs []error
+
 	for _, ethTx := range ethTxs {
+		// The sender has to be recovered and set here: MsgEthereumTx.ValidateBasic
+		// rejects a message without it, so a broadcast built with FromEthereumTx
+		// alone is refused by the receiving mempool with "sender address is
+		// missing". Locally that goes unnoticed, since block proposal reads this
+		// node's mempool directly, but the transaction never reaches its peers.
 		msg := &evmvmtypes.MsgEthereumTx{}
-		msg.FromEthereumTx(ethTx)
+		if err := msg.FromSignedEthereumTx(ethTx, signer); err != nil {
+			errs = append(errs, fmt.Errorf("failed to recover sender of transaction %s: %w", ethTx.Hash().Hex(), err))
+			continue
+		}
 
-		txBuilder := app.txConfig.NewTxBuilder()
-		if err := txBuilder.SetMsgs(msg); err != nil {
-			return fmt.Errorf("failed to set msg in tx builder: %w", err)
+		// Build through the message itself rather than filling a builder by hand:
+		// an EVM message only passes the ante handler inside a tx carrying the
+		// ExtensionOptionsEthereumTx option, and the fee and gas limit have to be
+		// taken off the transaction. Assembling this here by hand is what left
+		// them out and got the broadcast refused with "MsgEthereumTx needs to be
+		// contained within a tx with 'ExtensionOptionsEthereumTx' option".
+		cosmosTx, err := msg.BuildTx(app.txConfig.NewTxBuilder(), baseDenom)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("failed to build cosmos tx for %s: %w", ethTx.Hash().Hex(), err))
+			continue
 		}
 
-		txBytes, err := app.txConfig.TxEncoder()(txBuilder.GetTx())
+		txBytes, err := app.txConfig.TxEncoder()(cosmosTx)
 		if err != nil {
-			return fmt.Errorf("failed to encode transaction: %w", err)
+			errs = append(errs, fmt.Errorf("failed to encode transaction %s: %w", ethTx.Hash().Hex(), err))
+			continue
 		}
 
 		res, err := app.clientCtx.BroadcastTxSync(txBytes)
 		if err != nil {
-			return fmt.Errorf("failed to broadcast transaction %s: %w", ethTx.Hash().Hex(), err)
+			errs = append(errs, fmt.Errorf("failed to broadcast transaction %s: %w", ethTx.Hash().Hex(), err))
+			continue
+		}
+		// The submitting path has already handed this transaction to Comet, so
+		// this node's own cache answers the peer broadcast with "tx already in
+		// mempool". That is the expected outcome here, not a failure.
+		if res.Code == sdkerrors.ErrTxInMempoolCache.ABCICode() {
+			continue
 		}
 		if res.Code != 0 {
-			return fmt.Errorf("transaction %s rejected by mempool: code=%d, log=%s", ethTx.Hash().Hex(), res.Code, res.RawLog)
+			errs = append(errs, fmt.Errorf("transaction %s rejected by mempool: code=%d, log=%s", ethTx.Hash().Hex(), res.Code, res.RawLog))
 		}
 	}
-	return nil
+
+	return errors.Join(errs...)
 }
 
 func (app *TacChainApp) setPostHandler() {

app/evm_mempool_broadcast_test.go

@@ -17,15 +17,44 @@ import (
 	cmttypes "github.com/cometbft/cometbft/types"
 	"github.com/cosmos/cosmos-sdk/client"
 	simtestutil "github.com/cosmos/cosmos-sdk/testutil/sims"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+	authante "github.com/cosmos/cosmos-sdk/x/auth/ante"
 	"github.com/cosmos/evm/mempool/txpool/legacypool"
 	evmtypes "github.com/cosmos/evm/x/vm/types"
 	ethcmn "github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	ethcrypto "github.com/ethereum/go-ethereum/crypto"
 )
 
+// signedEthTx builds a transaction the broadcast path can recover a sender from.
+// An unsigned one is not usable here: MsgEthereumTx.ValidateBasic rejects a
+// message without a sender, so the broadcast has to derive it from the signature.
+func signedEthTx(t *testing.T) (*ethtypes.Transaction, ethcmn.Address) {
+	t.Helper()
+
+	key, err := ethcrypto.GenerateKey()
+	require.NoError(t, err)
+
+	to := ethcmn.Address{}
+	tx, err := ethtypes.SignNewTx(key, ethtypes.LatestSigner(evmtypes.GetEthChainConfig()), &ethtypes.LegacyTx{
+		Nonce:    1,
+		To:       &to,
+		Value:    big.NewInt(0),
+		Gas:      21_000,
+		GasPrice: big.NewInt(1),
+	})
+	require.NoError(t, err)
+
+	return tx, ethcrypto.PubkeyToAddress(key.PublicKey)
+}
+
 type broadcastRecorder struct {
 	rpcmock.Client
 
+	// code the node answers the broadcast with; zero means accepted
+	code uint32
+
 	mu    sync.Mutex
 	calls int
 	tx    cmttypes.Tx
@@ -37,7 +66,14 @@ func (r *broadcastRecorder) BroadcastTxSync(_ context.Context, tx cmttypes.Tx) (
 
 	r.calls++
 	r.tx = append(r.tx[:0], tx...)
-	return &coretypes.ResultBroadcastTx{}, nil
+	return &coretypes.ResultBroadcastTx{Code: r.code}, nil
+}
+
+func (r *broadcastRecorder) setCode(code uint32) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.code = code
+	r.calls = 0
 }
 
 func (r *broadcastRecorder) callCount() int {
@@ -87,14 +123,7 @@ func TestEVMMempoolBroadcastTxFnUsesUpdatedClientCtx(t *testing.T) {
 		WithClient(rpcClient),
 	)
 
-	to := ethcmn.Address{}
-	ethTx := ethtypes.NewTx(&ethtypes.LegacyTx{
-		Nonce:    1,
-		To:       &to,
-		Value:    big.NewInt(0),
-		Gas:      21_000,
-		GasPrice: big.NewInt(1),
-	})
+	ethTx, sender := signedEthTx(t)
 
 	// Before the explicit BroadCastTxFn override, this callback captured the
 	// empty client.Context from app construction and returned "no RPC client is
@@ -117,6 +146,27 @@ func TestEVMMempoolBroadcastTxFnUsesUpdatedClientCtx(t *testing.T) {
 	msg, ok := msgs[0].(*evmtypes.MsgEthereumTx)
 	require.True(t, ok)
 	require.Equal(t, ethTx.Hash(), msg.Hash())
+
+	// Without the sender the receiving mempool refuses the message with
+	// "sender address is missing", so the transaction never reaches its peers.
+	require.Equal(t, sender.Bytes(), []byte(msg.From),
+		"broadcast message must carry the recovered sender")
+	require.NoError(t, msg.ValidateBasic())
+
+	// The ante handler routes a transaction to the EVM path by this extension
+	// option and nothing else, so a broadcast without it is refused before the
+	// message is even looked at.
+	extTx, ok := decodedTx.(authante.HasExtensionOptionsTx)
+	require.True(t, ok)
+	opts := extTx.GetExtensionOptions()
+	require.Len(t, opts, 1, "broadcast tx must carry the ethereum extension option")
+	require.Equal(t, "/cosmos.evm.vm.v1.ExtensionOptionsEthereumTx", opts[0].GetTypeUrl())
+
+	// Fee and gas live on the ethereum transaction and have to be carried over.
+	feeTx, ok := decodedTx.(sdk.FeeTx)
+	require.True(t, ok)
+	require.Equal(t, ethTx.Gas(), feeTx.GetGas())
+	require.False(t, feeTx.GetFee().IsZero(), "broadcast tx must carry a fee")
 }
 
 func TestEVMMempoolBroadcastTxFnDoesNotBlockOnBroadcast(t *testing.T) {
@@ -144,14 +194,7 @@ func TestEVMMempoolBroadcastTxFnDoesNotBlockOnBroadcast(t *testing.T) {
 		WithClient(rpcClient),
 	)
 
-	to := ethcmn.Address{}
-	ethTx := ethtypes.NewTx(&ethtypes.LegacyTx{
-		Nonce:    1,
-		To:       &to,
-		Value:    big.NewInt(0),
-		Gas:      21_000,
-		GasPrice: big.NewInt(1),
-	})
+	ethTx, _ := signedEthTx(t)
 
 	done := make(chan error, 1)
 	go func() {
@@ -178,3 +221,46 @@ func TestEVMMempoolBroadcastTxFnDoesNotBlockOnBroadcast(t *testing.T) {
 		t.Fatal("background broadcast goroutine did not exit")
 	}
 }
+
+func TestBroadcastEVMTransactionsMempoolCodes(t *testing.T) {
+	tacApp := NewTacChainAppWithCustomOptions(t, true, SetupOptions{
+		Logger:  log.NewTestLogger(t),
+		DB:      dbm.NewMemDB(),
+		AppOpts: simtestutil.NewAppOptionsWithFlagHome(t.TempDir()),
+	})
+
+	ethTx, _ := signedEthTx(t)
+
+	// RegisterTxService may only run once per app, so the same client is reused
+	// and its answer changed per case.
+	rpcClient := &broadcastRecorder{}
+	tacApp.RegisterTxService(client.Context{}.
+		WithTxConfig(tacApp.txConfig).
+		WithClient(rpcClient),
+	)
+
+	for _, tc := range []struct {
+		name    string
+		code    uint32
+		wantErr bool
+	}{
+		// The submitting path has already given this transaction to Comet, so
+		// the node's own cache answers the peer broadcast as a duplicate. That
+		// is the normal outcome, not something to report.
+		{"duplicate in mempool cache", sdkerrors.ErrTxInMempoolCache.ABCICode(), false},
+		{"accepted", 0, false},
+		{"rejected", sdkerrors.ErrInvalidRequest.ABCICode(), true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rpcClient.setCode(tc.code)
+
+			err := tacApp.broadcastEVMTransactions([]*ethtypes.Transaction{ethTx})
+			if tc.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, 1, rpcClient.callCount())
+		})
+	}
+}

app/upgrades.go

@@ -27,7 +27,7 @@ var Upgrades = []upgrades.Upgrade{
 	v104.Upgrade, // ed25519 precompile
 	v160.Upgrade, // upgrade to cosmos/evm v0.6.0
 	v160spbhotfix.Upgrade,
-	v161.Upgrade, // vesting schedule shift (+6mo) & vesting account migration
+	v161.Upgrade, // no-op handler; binary-level fixes only
 }
 
 // RegisterUpgradeHandlers registers the chain upgrade handlers