Refactor how messages are processed

app/abci.go

@@ -2,19 +2,16 @@ package app
 
 import (
 	"bytes"
-	"errors"
-	"fmt"
 	"sort"
 
 	"github.com/celestiaorg/celestia-app/x/payment/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/cosmos-sdk/x/auth/signing"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/pkg/consts"
 	core "github.com/tendermint/tendermint/proto/tendermint/types"
 )
 
-// This file should contain all of the altered ABCI methods
-
 // PreprocessTxs fullfills the celestia-core version of the ACBI interface, by
 // performing basic validation for the incoming txs, and by cleanly separating
 // share messages from transactions
@@ -30,32 +27,49 @@ func (app *App) PreprocessTxs(txs abci.RequestPreprocessTxs) abci.ResponsePrepro
 			continue
 		}
 
+		authTx, ok := tx.(signing.Tx)
+		if !ok {
+			continue
+		}
+
 		// don't process the tx if the transaction doesn't contain a
-		// PayForMessage sdk.Msg
-		if !hasWirePayForMessage(tx) {
+		//  MsgPayForMessage sdk.Msg
+		if !hasWirePayForMessage(authTx) {
 			processedTxs = append(processedTxs, rawTx)
 			continue
 		}
 
 		// only support transactions that contain a single sdk.Msg
-		if len(tx.GetMsgs()) != 1 {
+		if len(authTx.GetMsgs()) != 1 {
 			continue
 		}
 
-		msg := tx.GetMsgs()[0]
+		msg := authTx.GetMsgs()[0]
+		wireMsg, ok := msg.(*types.MsgWirePayForMessage)
+		if !ok {
+			continue
+		}
 
 		// run basic validation on the transaction
-		err = tx.ValidateBasic()
+		err = authTx.ValidateBasic()
 		if err != nil {
 			continue
 		}
 
-		// process the message
-		coreMsg, signedTx, err := app.processMsg(msg)
+		// parse wire message and create a single message
+		coreMsg, unsignedPFM, sig, err := types.ProcessWirePayForMessage(wireMsg, app.SquareSize())
 		if err != nil {
 			continue
 		}
 
+		// create the signed PayForMessage using the fees, gas limit, and sequence from
+		// the original transaction, along with the appropriate signature.
+		signedTx, err := types.BuildPayForMessageTxFromWireTx(authTx, app.txConfig.NewTxBuilder(), sig, unsignedPFM)
+		if err != nil {
+			app.Logger().Error("failure to create signed PayForMessage", err)
+			continue
+		}
+
 		// increment the share counter by the number of shares taken by the message
 		sharesTaken := uint64(len(coreMsg.Data) / types.ShareSize)
 		shareCounter += sharesTaken
@@ -65,14 +79,12 @@ func (app *App) PreprocessTxs(txs abci.RequestPreprocessTxs) abci.ResponsePrepro
 			break
 		}
 
-		// encode the processed tx
-		rawProcessedTx, err := app.appCodec.Marshal(signedTx)
+		rawProcessedTx, err := app.txConfig.TxEncoder()(signedTx)
 		if err != nil {
 			continue
 		}
 
-		// add the message and tx to the output
-		shareMsgs = append(shareMsgs, &coreMsg)
+		shareMsgs = append(shareMsgs, coreMsg)
 		processedTxs = append(processedTxs, rawProcessedTx)
 	}
 
@@ -87,75 +99,16 @@ func (app *App) PreprocessTxs(txs abci.RequestPreprocessTxs) abci.ResponsePrepro
 	}
 }
 
-// pfmURL is the URL expected for pfm. NOTE: this will be deleted when we upgrade from
-// sdk v0.44.0
-var pfmURL = sdk.MsgTypeURL(&types.MsgWirePayForMessage{})
-
 func hasWirePayForMessage(tx sdk.Tx) bool {
 	for _, msg := range tx.GetMsgs() {
 		msgName := sdk.MsgTypeURL(msg)
-		if msgName == pfmURL {
+		if msgName == types.URLMsgWirePayforMessage {
 			return true
 		}
-		// note: this is what we will use in the future as proto.MessageName is
-		// deprecated
-		// svcMsg, ok := msg.(sdk.ServiceMsg) if !ok {
-		//  continue
-		// } if svcMsg.SerivceMethod == types.TypeMsgPayforMessage {
-		//  return true
-		// }
 	}
 	return false
 }
 
-// processMsgs will perform the processing required by PreProcessTxs for a set
-// of sdk.Msg's from a single sdk.Tx
-func (app *App) processMsg(msg sdk.Msg) (core.Message, *types.TxSignedTransactionDataPayForMessage, error) {
-	squareSize := app.SquareSize()
-	// reject all msgs in tx if a single included msg is not correct type
-	wireMsg, ok := msg.(*types.MsgWirePayForMessage)
-	if !ok {
-		return core.Message{},
-			nil,
-			errors.New("transaction contained a message type other than types.MsgWirePayForMessage")
-	}
-
-	// make sure that a ShareCommitAndSignature of the correct size is
-	// included in the message
-	var shareCommit types.ShareCommitAndSignature
-	for _, commit := range wireMsg.MessageShareCommitment {
-		if commit.K == squareSize {
-			shareCommit = commit
-		}
-	}
-	// K == 0 means there was no share commit with the desired current square size
-	if shareCommit.K == 0 {
-		return core.Message{},
-			nil,
-			fmt.Errorf("No share commit for correct square size. Current square size: %d", squareSize)
-	}
-
-	// add the message to the list of core message to be returned to ll-core
-	coreMsg := core.Message{
-		NamespaceId: wireMsg.GetMessageNameSpaceId(),
-		Data:        wireMsg.GetMessage(),
-	}
-
-	// wrap the signed transaction data
-	sTxData, err := wireMsg.SignedTransactionDataPayForMessage(squareSize)
-	if err != nil {
-		return core.Message{}, nil, err
-	}
-
-	signedData := &types.TxSignedTransactionDataPayForMessage{
-		Message:   sTxData,
-		Signature: shareCommit.Signature,
-		PublicKey: wireMsg.PublicKey,
-	}
-
-	return coreMsg, signedData, nil
-}
-
 // SquareSize returns the current square size. Currently, the square size is
 // hardcoded. todo(evan): don't hardcode the square size
 func (app *App) SquareSize() uint64 {

app/abci_test.go

@@ -18,8 +18,6 @@ import (
 	"github.com/cosmos/cosmos-sdk/server"
 	"github.com/cosmos/cosmos-sdk/simapp"
 	sdk "github.com/cosmos/cosmos-sdk/types"
-	"github.com/cosmos/cosmos-sdk/types/tx/signing"
-	authsigning "github.com/cosmos/cosmos-sdk/x/auth/signing"
 	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
 	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
 	"github.com/spf13/cast"
@@ -28,6 +26,7 @@ import (
 	"github.com/tendermint/spm/cosmoscmd"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/pkg/consts"
 	core "github.com/tendermint/tendermint/proto/tendermint/types"
 	dbm "github.com/tendermint/tm-db"
 )
@@ -39,40 +38,6 @@ func init() {
 	simapp.GetSimulatorFlags()
 }
 
-func TestProcessMsg(t *testing.T) {
-	kb := keyring.NewInMemory()
-	info, _, err := kb.NewMnemonic(testingKeyAcc, keyring.English, "", "", hd.Secp256k1)
-	if err != nil {
-		t.Error(err)
-	}
-	ns := []byte{1, 1, 1, 1, 1, 1, 1, 1}
-	message := bytes.Repeat([]byte{1}, 256)
-
-	// create a signed MsgWirePayFroMessage
-	msg := generateSignedWirePayForMessage(t, types.SquareSize, ns, message, kb)
-
-	testApp := setupApp(t, info.GetPubKey())
-
-	tests := []struct {
-		name string
-		args sdk.Msg
-		want core.Message
-	}{
-		{
-			name: "basic",
-			args: msg,
-			want: core.Message{NamespaceId: msg.MessageNameSpaceId, Data: msg.Message},
-		},
-	}
-	for _, tt := range tests {
-		result, _, err := testApp.processMsg(tt.args)
-		if err != nil {
-			t.Error(err)
-		}
-		assert.Equal(t, tt.want, result, tt.name)
-	}
-}
-
 func TestPreprocessTxs(t *testing.T) {
 	kb := keyring.NewInMemory()
 	info, _, err := kb.NewMnemonic(testingKeyAcc, keyring.English, "", "", hd.Secp256k1)
@@ -151,7 +116,7 @@ func setupApp(t *testing.T, pub cryptotypes.PubKey) *App {
 		anteOpt,
 	)
 
-	genesisState := NewDefaultGenesisState(encCfg.Marshaler)
+	genesisState := newDefaultGenesisState(encCfg.Marshaler)
 
 	genesisState, err := addGenesisAccount(sdk.AccAddress(pub.Address().Bytes()), genesisState, encCfg.Marshaler)
 	if err != nil {
@@ -244,19 +209,10 @@ func addGenesisAccount(addr sdk.AccAddress, appState map[string]json.RawMessage,
 
 func generateRawTx(t *testing.T, txConfig client.TxConfig, ns, message []byte, ring keyring.Keyring) (rawTx []byte) {
 	// create a msg
-	msg := generateSignedWirePayForMessage(t, types.SquareSize, ns, message, ring)
-
-	info, err := ring.Key(testingKeyAcc)
-	if err != nil {
-		t.Error(err)
-	}
+	msg := generateSignedWirePayForMessage(t, consts.MaxSquareSize, ns, message, ring)
 
-	// this is returning a tx.wrapper
-	builder := txConfig.NewTxBuilder()
-	err = builder.SetMsgs(msg)
-	if err != nil {
-		t.Error(err)
-	}
+	krs := generateKeyringSigner(t, "test")
+	builder := krs.NewTxBuilder()
 
 	coin := sdk.Coin{
 		Denom:  "token",
@@ -267,97 +223,64 @@ func generateRawTx(t *testing.T, txConfig client.TxConfig, ns, message []byte, r
 	builder.SetGasLimit(10000)
 	builder.SetTimeoutHeight(99)
 
-	signingData := authsigning.SignerData{
-		ChainID:       "test-chain",
-		AccountNumber: 0,
-		Sequence:      0,
-	}
+	tx, err := krs.BuildSignedTx(builder, msg)
+	require.NoError(t, err)
 
-	// Important set the Signature to nil BEFORE actually signing
-	sigData := signing.SingleSignatureData{
-		SignMode:  signing.SignMode_SIGN_MODE_DIRECT,
-		Signature: nil,
-	}
+	// encode the tx
+	rawTx, err = txConfig.TxEncoder()(tx)
+	require.NoError(t, err)
 
-	sig := signing.SignatureV2{
-		PubKey:   info.GetPubKey(),
-		Data:     &sigData,
-		Sequence: 0,
-	}
+	return rawTx
+}
 
-	// set the empty signature
-	err = builder.SetSignatures(sig)
-	if err != nil {
-		if err != nil {
-			t.Error(err)
-		}
-	}
+func generateSignedWirePayForMessage(t *testing.T, k uint64, ns, message []byte, ring keyring.Keyring) *types.MsgWirePayForMessage {
+	signer := generateKeyringSigner(t, "test")
 
-	// Generate the bytes to be signed.
-	bytesToSign, err := txConfig.
-		SignModeHandler().
-		GetSignBytes(
-			signing.SignMode_SIGN_MODE_DIRECT,
-			signingData,
-			builder.GetTx(),
-		)
+	msg, err := types.NewWirePayForMessage(ns, message, k)
 	if err != nil {
 		t.Error(err)
 	}
 
-	// Sign those bytes
-	sigBytes, _, err := ring.Sign(testingKeyAcc, bytesToSign)
+	err = msg.SignShareCommitments(signer, signer.NewTxBuilder())
 	if err != nil {
 		t.Error(err)
 	}
 
-	// Construct the SignatureV2 struct
-	sigData = signing.SingleSignatureData{
-		SignMode:  signing.SignMode_SIGN_MODE_DIRECT,
-		Signature: sigBytes,
-	}
+	return msg
+}
 
-	sigV2 := signing.SignatureV2{
-		PubKey:   info.GetPubKey(),
-		Data:     &sigData,
-		Sequence: 0,
-	}
+func generateKeyring(t *testing.T, accts ...string) keyring.Keyring {
+	t.Helper()
+	kb := keyring.NewInMemory()
 
-	// set the actual signature
-	err = builder.SetSignatures(sigV2)
-	if err != nil {
+	for _, acc := range accts {
+		_, _, err := kb.NewMnemonic(acc, keyring.English, "", "", hd.Secp256k1)
 		if err != nil {
 			t.Error(err)
 		}
 	}
 
-	// finish the tx
-	tx := builder.GetTx()
-
-	// encode the tx
-	rawTx, err = txConfig.TxEncoder()(tx)
+	_, err := kb.NewAccount(testAccName, testMnemo, "1234", "", hd.Secp256k1)
 	if err != nil {
-		t.Error(err)
+		panic(err)
 	}
 
-	return rawTx
+	return kb
 }
 
-func generateSignedWirePayForMessage(t *testing.T, k uint64, ns, message []byte, ring keyring.Keyring) *types.MsgWirePayForMessage {
-	info, err := ring.Key(testingKeyAcc)
-	if err != nil {
-		t.Error(err)
-	}
-
-	msg, err := types.NewMsgWirePayForMessage(ns, message, info.GetPubKey().Bytes(), &types.TransactionFee{}, k)
-	if err != nil {
-		t.Error(err)
-	}
+func generateKeyringSigner(t *testing.T, accts ...string) *types.KeyringSigner {
+	kr := generateKeyring(t, accts...)
+	return types.NewKeyringSigner(kr, testAccName, testChainID)
+}
 
-	err = msg.SignShareCommitments(testingKeyAcc, ring)
-	if err != nil {
-		t.Error(err)
-	}
+const (
+	// nolint:lll
+	testMnemo   = `ramp soldier connect gadget domain mutual staff unusual first midnight iron good deputy wage vehicle mutual spike unlock rocket delay hundred script tumble choose`
+	testAccName = "test-account"
+	testChainID = "test-chain-1"
+)
 
-	return msg
+// newDefaultGenesisState generates the default state for the application.
+func newDefaultGenesisState(cdc codec.JSONCodec) GenesisState {
+	return ModuleBasics.DefaultGenesis(cdc)
 }