feat: prioritize protocol txs

app/app.go

@@ -343,6 +343,10 @@ func New(
 
 	app.App = appBuilder.Build(db, traceStore, baseAppOptions...)
 
+	// Set the priority proposal handler
+	proposalHandler := NewPriorityProposalHandler(app.Logger(), app.txConfig.TxDecoder())
+	app.App.BaseApp.SetPrepareProposal(proposalHandler.PrepareProposal())
+
 	// Register legacy modules
 	app.registerIBCModules()
 

app/proposal_handler.go

@@ -0,0 +1,78 @@
+package app
+
+import (
+	"cosmossdk.io/log"
+	bundlestypes "github.com/KYVENetwork/chain/x/bundles/types"
+	abci "github.com/cometbft/cometbft/abci/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"reflect"
+	"slices"
+)
+
+type PriorityProposalHandler struct {
+	logger    log.Logger
+	txDecoder sdk.TxDecoder
+}
+
+func NewPriorityProposalHandler(logger log.Logger, decoder sdk.TxDecoder) *PriorityProposalHandler {
+	return &PriorityProposalHandler{
+		logger:    logger,
+		txDecoder: decoder,
+	}
+}
+
+var priorityTypes = []string{
+	reflect.TypeOf(bundlestypes.MsgSubmitBundleProposal{}).Name(),
+	reflect.TypeOf(bundlestypes.MsgVoteBundleProposal{}).Name(),
+	reflect.TypeOf(bundlestypes.MsgClaimUploaderRole{}).Name(),
+	reflect.TypeOf(bundlestypes.MsgSkipUploaderRole{}).Name(),
+}
+
+// PrepareProposal returns a PrepareProposalHandler that separates transactions into different queues
+// This function is only called by the block proposer and therefore does NOT need to be deterministic
+func (h *PriorityProposalHandler) PrepareProposal() sdk.PrepareProposalHandler {
+	return func(ctx sdk.Context, req *abci.RequestPrepareProposal) (*abci.ResponsePrepareProposal, error) {
+		// Separate the transactions into different queues
+		// priorityQueue: transactions that should be executed before the default transactions
+		// defaultQueue: transactions that should be executed last
+		// The order is priorityQueue -> defaultQueue
+
+		var priorityQueue [][]byte
+		var defaultQueue [][]byte
+
+		// Iterate through the transactions and separate them into different queues
+		for _, rawTx := range req.Txs {
+			tx, err := h.txDecoder(rawTx)
+			if err != nil {
+				h.logger.Error("failed to decode transaction", "error", err)
+				continue
+			}
+			msgs, err := tx.GetMsgsV2()
+			if err != nil {
+				h.logger.Error("failed to get messages from transaction", "error", err)
+				continue
+			}
+
+			// We only care about transactions with a single message
+			if len(msgs) == 1 {
+				msg := msgs[0]
+				msgType := string(msg.ProtoReflect().Type().Descriptor().Name())
+
+				if slices.Contains(priorityTypes, msgType) {
+					priorityQueue = append(priorityQueue, rawTx)
+					continue
+				}
+			}
+
+			// Otherwise, add the tx to the default queue
+			defaultQueue = append(defaultQueue, rawTx)
+		}
+
+		// Append the default queue to the priority queue
+		priorityQueue = append(priorityQueue, defaultQueue...)
+
+		return &abci.ResponsePrepareProposal{
+			Txs: priorityQueue,
+		}, nil
+	}
+}

go.work.sum

@@ -311,6 +311,10 @@ github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKz
 github.com/Joker/jade v1.0.1-0.20190614124447-d475f43051e7/go.mod h1:6E6s8o2AE4KhCrqr6GRJjdC/gNfTdxkIXvuGZZda2VM=
 github.com/Joker/jade v1.1.3 h1:Qbeh12Vq6BxURXT1qZBRHsDxeURB8ztcL6f3EXSGeHk=
 github.com/Joker/jade v1.1.3/go.mod h1:T+2WLyt7VH6Lp0TRxQrUYEs64nRc83wkMQrfeIQKduM=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520123918-cc448f6c1b65 h1:vxgj8yZdJuIJGrVaHd54DSN2gl+ERQtWdYAZJDKYWvM=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520123918-cc448f6c1b65/go.mod h1:pupV0YN3A56/u9kHj9U1F8MdDUEolBIn05F0W1q/0oI=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520124515-fd4cc797e6fd h1:lKJ7X9Q+KbQviDxpY4OaalC8/oZ64rXfTNTny4jfuE0=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520124515-fd4cc797e6fd/go.mod h1:pupV0YN3A56/u9kHj9U1F8MdDUEolBIn05F0W1q/0oI=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible h1:1G1pk05UrOh0NlF1oeaaix1x8XzrfjIDK47TY0Zehcw=
 github.com/Masterminds/semver/v3 v3.2.0 h1:3MEsd0SM6jqZojhjLWWeBY+Kcjy9i6MQAeY7YgDP83g=
 github.com/Masterminds/semver/v3 v3.2.0/go.mod h1:qvl/7zhW3nngYb5+80sSMF+FG2BjYrf8m9wsX0PNOMQ=

interchaintest/go.mod

@@ -29,6 +29,7 @@ require (
 	cosmossdk.io/errors v1.0.1 // indirect
 	cosmossdk.io/log v1.3.1 // indirect
 	cosmossdk.io/store v1.1.0 // indirect
+	cosmossdk.io/x/evidence v0.1.0 // indirect
 	cosmossdk.io/x/feegrant v0.1.0 // indirect
 	cosmossdk.io/x/tx v0.13.1 // indirect
 	cosmossdk.io/x/upgrade v0.1.1 // indirect
@@ -56,6 +57,7 @@ require (
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/chzyer/readline v1.5.1 // indirect
+	github.com/cockroachdb/apd/v2 v2.0.2 // indirect
 	github.com/cockroachdb/errors v1.11.1 // indirect
 	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
 	github.com/cockroachdb/pebble v1.1.0 // indirect
@@ -71,7 +73,7 @@ require (
 	github.com/cosmos/gogoproto v1.4.12 // indirect
 	github.com/cosmos/iavl v1.1.1 // indirect
 	github.com/cosmos/ibc-go/modules/capability v1.0.0 // indirect
-	github.com/cosmos/ibc-go/v8 v8.1.1 // indirect
+	github.com/cosmos/ibc-go/v8 v8.2.0 // indirect
 	github.com/cosmos/ics23/go v0.10.0 // indirect
 	github.com/cosmos/ledger-cosmos-go v0.13.3 // indirect
 	github.com/danieljoos/wincred v1.1.2 // indirect
@@ -242,6 +244,7 @@ require (
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gotest.tools/v3 v3.5.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect
@@ -275,7 +278,7 @@ replace (
 	// needed for strangelove's interchaintest
 	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
 	// use kyve flavored strangelove-ventures/interchaintest
-	github.com/strangelove-ventures/interchaintest/v8 => github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240422073541-22e23c056376
+	github.com/strangelove-ventures/interchaintest/v8 => github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520124515-fd4cc797e6fd
 	// replace broken goleveldb
 	github.com/syndtr/goleveldb => github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
 )

interchaintest/go.sum

@@ -239,8 +239,8 @@ github.com/FactomProject/btcutilecc v0.0.0-20130527213604-d3a63a5752ec h1:1Qb69m
 github.com/FactomProject/btcutilecc v0.0.0-20130527213604-d3a63a5752ec/go.mod h1:CD8UlnlLDiqb36L110uqiP2iSflVjx9g/3U9hCI4q2U=
 github.com/KYVENetwork/cosmos-sdk v0.50.5-kyve-rc2 h1:SUX6bCgG72BuWAkr7vE2VKFGU4SqBaE0uKUd3qeIk2o=
 github.com/KYVENetwork/cosmos-sdk v0.50.5-kyve-rc2/go.mod h1:sM3HLOjUE6rwAiuwEOEtPd2DUcXG+uCktW+CdID+ZMM=
-github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240422073541-22e23c056376 h1:I/yGiXZzQwWfVUdT9+jvGHiRkWAft3U3s1eOFQfel8I=
-github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240422073541-22e23c056376/go.mod h1:pupV0YN3A56/u9kHj9U1F8MdDUEolBIn05F0W1q/0oI=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520124515-fd4cc797e6fd h1:lKJ7X9Q+KbQviDxpY4OaalC8/oZ64rXfTNTny4jfuE0=
+github.com/KYVENetwork/interchaintest/v8 v8.0.0-20240520124515-fd4cc797e6fd/go.mod h1:pupV0YN3A56/u9kHj9U1F8MdDUEolBIn05F0W1q/0oI=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow=
 github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
@@ -393,8 +393,8 @@ github.com/cosmos/iavl v1.1.1 h1:64nTi8s3gEoGqhA8TyAWFWfz7/pg0anKzHNSc1ETc7Q=
 github.com/cosmos/iavl v1.1.1/go.mod h1:jLeUvm6bGT1YutCaL2fIar/8vGUE8cPZvh/gXEWDaDM=
 github.com/cosmos/ibc-go/modules/capability v1.0.0 h1:r/l++byFtn7jHYa09zlAdSeevo8ci1mVZNO9+V0xsLE=
 github.com/cosmos/ibc-go/modules/capability v1.0.0/go.mod h1:D81ZxzjZAe0ZO5ambnvn1qedsFQ8lOwtqicG6liLBco=
-github.com/cosmos/ibc-go/v8 v8.1.1 h1:N2+GA86yACcXnKWCKtqdbCwP0/Eo8pH79+6e7TicULU=
-github.com/cosmos/ibc-go/v8 v8.1.1/go.mod h1:o1ipS95xpdjqNcB8Drq0eI3Sn4FRLigjll42ec1ECuU=
+github.com/cosmos/ibc-go/v8 v8.2.0 h1:7oCzyy1sZCcgpeQLnHxC56brsSz3KWwQGKXalXwXFzE=
+github.com/cosmos/ibc-go/v8 v8.2.0/go.mod h1:wj3qx75iC/XNnsMqbPDCIGs0G6Y3E/lo3bdqCyoCy+8=
 github.com/cosmos/ics23/go v0.10.0 h1:iXqLLgp2Lp+EdpIuwXTYIQU+AiHj9mOC2X9ab++bZDM=
 github.com/cosmos/ics23/go v0.10.0/go.mod h1:ZfJSmng/TBNTBkFemHHHj5YY7VAU/MBU980F4VU1NG0=
 github.com/cosmos/keyring v1.2.0 h1:8C1lBP9xhImmIabyXW4c3vFjjLiBdGCmfLUfeZlV1Yo=

interchaintest/proposal_handler/proposal_handler_test.go

@@ -0,0 +1,193 @@
+package proposal_handler_test
+
+import (
+	"context"
+	"cosmossdk.io/math"
+	bundlestypes "github.com/KYVENetwork/chain/x/bundles/types"
+	stakerstypes "github.com/KYVENetwork/chain/x/stakers/types"
+	sdkclient "github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/client/flags"
+	"github.com/cosmos/cosmos-sdk/client/tx"
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	"github.com/strangelove-ventures/interchaintest/v8"
+	"github.com/strangelove-ventures/interchaintest/v8/testutil"
+	"reflect"
+	"testing"
+
+	"github.com/strangelove-ventures/interchaintest/v8/chain/cosmos"
+	"go.uber.org/zap/zaptest"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+/*
+
+TEST CASES - proposal_handler.go
+
+* Execute multiple transactions and check their order
+* Execute transactions that exceed max tx bytes
+
+*/
+
+func TestProposalHandler(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "interchaintest/ProposalHandler Test Suite")
+}
+
+var _ = Describe("proposal_handler.go", Ordered, func() {
+	var chain *cosmos.CosmosChain
+
+	var ctx context.Context
+	var interchain *interchaintest.Interchain
+
+	var broadcaster *cosmos.Broadcaster
+	var wallets []*cosmos.CosmosWallet
+
+	BeforeAll(func() {
+		numFullNodes := 0
+		numValidators := 2
+		factory := interchaintest.NewBuiltinChainFactory(
+			zaptest.NewLogger(GinkgoT()),
+			[]*interchaintest.ChainSpec{mainnetChainSpec(numValidators, numFullNodes)},
+		)
+
+		chains, err := factory.Chains(GinkgoT().Name())
+		Expect(err).To(BeNil())
+		chain = chains[0].(*cosmos.CosmosChain)
+
+		interchain = interchaintest.NewInterchain().
+			AddChain(chain)
+
+		broadcaster = cosmos.NewBroadcaster(GinkgoT(), chain)
+		broadcaster.ConfigureClientContextOptions(func(clientContext sdkclient.Context) sdkclient.Context {
+			return clientContext.
+				WithBroadcastMode(flags.BroadcastAsync)
+		})
+		broadcaster.ConfigureFactoryOptions(func(factory tx.Factory) tx.Factory {
+			return factory.
+				WithGas(flags.DefaultGasLimit * 10)
+		})
+
+		ctx = context.Background()
+		client, network := interchaintest.DockerSetup(GinkgoT())
+
+		err = interchain.Build(ctx, nil, interchaintest.InterchainBuildOptions{
+			TestName:         GinkgoT().Name(),
+			Client:           client,
+			NetworkID:        network,
+			SkipPathCreation: true,
+		})
+		Expect(err).To(BeNil())
+
+		for i := 0; i < 10; i++ {
+			wallets = append(wallets, interchaintest.GetAndFundTestUsers(
+				GinkgoT(), ctx, GinkgoT().Name(), math.NewInt(10_000_000_000), chain,
+			)[0].(*cosmos.CosmosWallet))
+		}
+	})
+
+	AfterAll(func() {
+		_ = chain.StopAllNodes(ctx)
+		_ = interchain.Close()
+	})
+
+	It("Execute multiple transactions and check their order", func() {
+		// ARRANGE
+		err := testutil.WaitForBlocks(ctx, 1, chain)
+		Expect(err).To(BeNil())
+
+		height, err := chain.Height(ctx)
+		Expect(err).To(BeNil())
+
+		// ACT
+
+		// Execute different transactions
+		// We don't care about the results, they only have to be included in a block
+		broadcastMsg(ctx, broadcaster, wallets[0], &banktypes.MsgSend{FromAddress: wallets[0].FormattedAddress()})
+		broadcastMsg(ctx, broadcaster, wallets[1], &stakerstypes.MsgCreateStaker{Creator: wallets[1].FormattedAddress()})
+		broadcastMsg(ctx, broadcaster, wallets[2], &bundlestypes.MsgClaimUploaderRole{Creator: wallets[2].FormattedAddress()}) // priority msg
+		broadcastMsg(ctx, broadcaster, wallets[3], &stakerstypes.MsgJoinPool{Creator: wallets[3].FormattedAddress(), Valaddress: wallets[0].FormattedAddress(), PoolId: 0})
+		broadcastMsg(ctx, broadcaster, wallets[4], &banktypes.MsgSend{FromAddress: wallets[4].FormattedAddress()})
+		broadcastMsg(ctx, broadcaster, wallets[5], &bundlestypes.MsgVoteBundleProposal{Creator: wallets[5].FormattedAddress()})   // priority msg
+		broadcastMsg(ctx, broadcaster, wallets[6], &bundlestypes.MsgSkipUploaderRole{Creator: wallets[6].FormattedAddress()})     // priority msg
+		broadcastMsg(ctx, broadcaster, wallets[7], &bundlestypes.MsgSubmitBundleProposal{Creator: wallets[7].FormattedAddress()}) // priority msg
+
+		expectedOrder := []string{
+			// priority msgs
+			reflect.TypeOf(bundlestypes.MsgClaimUploaderRole{}).Name(),
+			reflect.TypeOf(bundlestypes.MsgVoteBundleProposal{}).Name(),
+			reflect.TypeOf(bundlestypes.MsgSkipUploaderRole{}).Name(),
+			reflect.TypeOf(bundlestypes.MsgSubmitBundleProposal{}).Name(),
+			// default msgs
+			reflect.TypeOf(banktypes.MsgSend{}).Name(),
+			reflect.TypeOf(stakerstypes.MsgCreateStaker{}).Name(),
+			reflect.TypeOf(stakerstypes.MsgJoinPool{}).Name(),
+			reflect.TypeOf(banktypes.MsgSend{}).Name(),
+		}
+
+		afterHeight, err := chain.Height(ctx)
+		Expect(err).To(BeNil())
+		Expect(afterHeight).To(Equal(height))
+
+		// Wait for the transactions to be included in a block
+		err = testutil.WaitForBlocks(ctx, 2, chain)
+		Expect(err).To(BeNil())
+
+		// ASSERT
+
+		// Check the order of the transactions
+		checkTxsOrder(ctx, chain, height+1, expectedOrder)
+	})
+
+	It("Execute transactions that exceed max tx bytes", func() {
+		// ARRANGE
+		err := testutil.WaitForBlocks(ctx, 1, chain)
+		Expect(err).To(BeNil())
+
+		height, err := chain.Height(ctx)
+		Expect(err).To(BeNil())
+
+		// ACT
+		const duplications = 40
+		broadcastMsg(ctx, broadcaster, wallets[0], &stakerstypes.MsgCreateStaker{Creator: wallets[0].FormattedAddress()})
+		broadcastMsgs(ctx, broadcaster, wallets[1], duplicateMsg(&banktypes.MsgSend{FromAddress: wallets[1].FormattedAddress()}, duplications)...)
+		broadcastMsg(ctx, broadcaster, wallets[2], &bundlestypes.MsgSkipUploaderRole{Creator: wallets[2].FormattedAddress()}) // priority msg
+
+		// this will not make it into the actual block, so it goes into the next one with all following msgs
+		broadcastMsgs(ctx, broadcaster, wallets[3], duplicateMsg(&banktypes.MsgSend{FromAddress: wallets[4].FormattedAddress()}, duplications)...)
+		broadcastMsg(ctx, broadcaster, wallets[4], &stakerstypes.MsgJoinPool{Creator: wallets[5].FormattedAddress(), Valaddress: wallets[0].FormattedAddress(), PoolId: 0})
+		broadcastMsg(ctx, broadcaster, wallets[5], &bundlestypes.MsgVoteBundleProposal{Creator: wallets[6].FormattedAddress()}) // priority msg
+
+		afterHeight, err := chain.Height(ctx)
+		Expect(err).To(BeNil())
+		Expect(afterHeight).To(Equal(height))
+
+		// Wait for the transactions to be included in a block
+		err = testutil.WaitForBlocks(ctx, 2, chain)
+		Expect(err).To(BeNil())
+
+		// ASSERT
+		var msgTypes []string
+		for i := 0; i < duplications; i++ {
+			msgTypes = append(msgTypes, reflect.TypeOf(banktypes.MsgSend{}).Name())
+		}
+
+		// Check that only the first block contains the first transactions
+		checkTxsOrder(ctx, chain, height+1, append(
+			[]string{
+				reflect.TypeOf(bundlestypes.MsgSkipUploaderRole{}).Name(), // priority msg
+				reflect.TypeOf(stakerstypes.MsgCreateStaker{}).Name(),
+			},
+			msgTypes...,
+		))
+		// The second block should contain the rest of the transactions
+		checkTxsOrder(ctx, chain, height+2, append(
+			msgTypes,
+			[]string{
+				reflect.TypeOf(bundlestypes.MsgVoteBundleProposal{}).Name(), // priority msg
+				reflect.TypeOf(stakerstypes.MsgJoinPool{}).Name(),
+			}...,
+		))
+	})
+})