feat(x/async): implement PrepareProposal and ProcessProposal

PrepareProposal injects an AsyncInjectedTx in the block, right after the
slinky transaction, i.e. the tx #0 is slinky, the tx #1 is x/async, and
the rest are the normal transactions of the block.

ProcessProposal checks if the transactions bytes in position #1 in the
proposal are a valid AsyncInjectedTx.

warden/x/async/keeper/abci.go

@@ -5,6 +5,7 @@ import (
 
 	cometabci "github.com/cometbft/cometbft/abci/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/skip-mev/slinky/abci/ve"
 	"github.com/warden-protocol/wardenprotocol/prophet"
 	types "github.com/warden-protocol/wardenprotocol/warden/x/async/types/v1beta1"
 )
@@ -84,12 +85,51 @@ func (k Keeper) PrepareProposalHandler() sdk.PrepareProposalHandler {
 		resp := &cometabci.ResponsePrepareProposal{
 			Txs: req.Txs,
 		}
+
+		if !ve.VoteExtensionsEnabled(ctx) {
+			return resp, nil
+		}
+
+		log := ctx.Logger().With("module", "prophet")
+		asyncTx, err := k.buildAsyncTx(req.LocalLastCommit.Votes)
+		if err != nil {
+			log.Error("failed to build async tx", "err", err)
+			return resp, nil
+		}
+		resp.Txs = trimExcessBytes(resp.Txs, req.MaxTxBytes-int64(len(asyncTx)))
+		resp.Txs = injectTx(asyncTx, 1, resp.Txs)
+
 		return resp, nil
 	}
 }
 
 func (k Keeper) ProcessProposalHandler() sdk.ProcessProposalHandler {
 	return func(ctx sdk.Context, req *cometabci.RequestProcessProposal) (*cometabci.ResponseProcessProposal, error) {
+		resp := &cometabci.ResponseProcessProposal{
+			Status: cometabci.ResponseProcessProposal_ACCEPT,
+		}
+
+		if !ve.VoteExtensionsEnabled(ctx) || len(req.Txs) < 2 {
+			return resp, nil
+		}
+
+		log := ctx.Logger().With("module", "prophet")
+		asyncTx := req.Txs[1]
+		if len(asyncTx) == 0 {
+			return resp, nil
+		}
+
+		var tx types.AsyncInjectedTx
+		if err := tx.Unmarshal(asyncTx); err != nil {
+			log.Error("failed to unmarshal async tx", "err", err)
+			// probably not an async tx?
+			// but slinky in this case rejects their proposal so maybe we
+			// should do the same?
+			return &cometabci.ResponseProcessProposal{
+				Status: cometabci.ResponseProcessProposal_ACCEPT,
+			}, nil
+		}
+
 		return &cometabci.ResponseProcessProposal{
 			Status: cometabci.ResponseProcessProposal_ACCEPT,
 		}, nil
@@ -104,3 +144,47 @@ func (k Keeper) PreBlocker() sdk.PreBlocker {
 		return resp, nil
 	}
 }
+
+func (k Keeper) buildAsyncTx(votes []cometabci.ExtendedVoteInfo) ([]byte, error) {
+	tx := types.AsyncInjectedTx{
+		ExtendedVotesInfo: votes,
+	}
+
+	txBytes, err := tx.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	return txBytes, nil
+}
+
+func injectTx(newTx []byte, position int, appTxs [][]byte) [][]byte {
+	if position < 0 {
+		panic("position must be >= 0")
+	}
+
+	if position == 0 {
+		return append([][]byte{newTx}, appTxs...)
+	}
+
+	if position >= len(appTxs) {
+		return append(appTxs, newTx)
+	}
+
+	return append(appTxs[:position], append([][]byte{newTx}, appTxs[position:]...)...)
+}
+
+func trimExcessBytes(txs [][]byte, maxSizeBytes int64) [][]byte {
+	var (
+		returnedTxs   [][]byte
+		consumedBytes int64
+	)
+	for _, tx := range txs {
+		consumedBytes += int64(len(tx))
+		if consumedBytes > maxSizeBytes {
+			break
+		}
+		returnedTxs = append(returnedTxs, tx)
+	}
+	return returnedTxs
+}

warden/x/async/keeper/abci_test.go

@@ -0,0 +1,137 @@
+package keeper
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTrimExcessBytes(t *testing.T) {
+	type args struct {
+		txs          [][]byte
+		maxSizeBytes int64
+	}
+	tests := []struct {
+		name string
+		args args
+		want [][]byte
+	}{
+		{
+			name: "don't trim",
+			args: args{
+				txs:          [][]byte{[]byte("tx1"), []byte("tx2")},
+				maxSizeBytes: 10,
+			},
+			want: [][]byte{[]byte("tx1"), []byte("tx2")},
+		},
+		{
+			name: "trim one tx precise",
+			args: args{
+				txs:          [][]byte{[]byte("tx1"), []byte("tx2"), []byte("tx3")},
+				maxSizeBytes: 6,
+			},
+			want: [][]byte{[]byte("tx1"), []byte("tx2")},
+		},
+		{
+			name: "trim one tx with 1 byte excess",
+			args: args{
+				txs:          [][]byte{[]byte("tx1"), []byte("tx2"), []byte("tx3")},
+				maxSizeBytes: 7,
+			},
+			want: [][]byte{[]byte("tx1"), []byte("tx2")},
+		},
+		{
+			name: "trim one tx with 2 bytes excess",
+			args: args{
+				txs:          [][]byte{[]byte("tx1"), []byte("tx2"), []byte("tx3")},
+				maxSizeBytes: 8,
+			},
+			want: [][]byte{[]byte("tx1"), []byte("tx2")},
+		},
+		{
+			name: "empty list",
+			args: args{
+				txs:          [][]byte{},
+				maxSizeBytes: 8,
+			},
+			want: nil,
+		},
+		{
+			name: "nil list",
+			args: args{
+				txs:          nil,
+				maxSizeBytes: 8,
+			},
+			want: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, trimExcessBytes(tt.args.txs, tt.args.maxSizeBytes))
+		})
+	}
+}
+
+func TestInjectTx(t *testing.T) {
+	type args struct {
+		newTx    []byte
+		position int
+		appTxs   [][]byte
+	}
+	tests := []struct {
+		name string
+		args args
+		want [][]byte
+	}{
+		{
+			name: "inject at the beginning, empty list",
+			args: args{
+				newTx:    []byte("newTx"),
+				position: 0,
+				appTxs:   [][]byte{},
+			},
+			want: [][]byte{[]byte("newTx")},
+		},
+		{
+			name: "inject at the beginning, non-empty list",
+			args: args{
+				newTx:    []byte("newTx"),
+				position: 0,
+				appTxs:   [][]byte{[]byte("appTx1"), []byte("appTx2")},
+			},
+			want: [][]byte{[]byte("newTx"), []byte("appTx1"), []byte("appTx2")},
+		},
+		{
+			name: "position is over the length of the list, empty list",
+			args: args{
+				newTx:    []byte("newTx"),
+				position: 2,
+				appTxs:   [][]byte{},
+			},
+			want: [][]byte{[]byte("newTx")},
+		},
+		{
+			name: "position is over the length of the list, non-empty list",
+			args: args{
+				newTx:    []byte("newTx"),
+				position: 4,
+				appTxs:   [][]byte{[]byte("appTx1"), []byte("appTx2")},
+			},
+			want: [][]byte{[]byte("appTx1"), []byte("appTx2"), []byte("newTx")},
+		},
+		{
+			name: "position in the middle",
+			args: args{
+				newTx:    []byte("newTx"),
+				position: 1,
+				appTxs:   [][]byte{[]byte("appTx1"), []byte("appTx2")},
+			},
+			want: [][]byte{[]byte("appTx1"), []byte("newTx"), []byte("appTx2")},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, injectTx(tt.args.newTx, tt.args.position, tt.args.appTxs))
+		})
+	}
+}