e2e: tx load to use broadcast sync instead of commit (backport #6347) (#6352)

test/e2e/runner/load.go

@@ -13,9 +13,10 @@ import (
 	"github.com/tendermint/tendermint/types"
 )
 
-// Load generates transactions against the network until the given
-// context is cancelled.
-func Load(ctx context.Context, testnet *e2e.Testnet) error {
+// Load generates transactions against the network until the given context is
+// canceled. A multiplier of greater than one can be supplied if load needs to
+// be generated beyond a minimum amount.
+func Load(ctx context.Context, testnet *e2e.Testnet, multiplier int) error {
 	// Since transactions are executed across all nodes in the network, we need
 	// to reduce transaction load for larger networks to avoid using too much
 	// CPU. This gives high-throughput small networks and low-throughput large ones.
@@ -37,7 +38,7 @@ func Load(ctx context.Context, testnet *e2e.Testnet) error {
 	logger.Info(fmt.Sprintf("Starting transaction load (%v workers)...", concurrency))
 	started := time.Now()
 
-	go loadGenerate(ctx, chTx)
+	go loadGenerate(ctx, chTx, multiplier)
 
 	for w := 0; w < concurrency; w++ {
 		go loadProcess(ctx, testnet, chTx, chSuccess)
@@ -64,14 +65,14 @@ func Load(ctx context.Context, testnet *e2e.Testnet) error {
 	}
 }
 
-// loadGenerate generates jobs until the context is cancelled
-func loadGenerate(ctx context.Context, chTx chan<- types.Tx) {
+// loadGenerate generates jobs until the context is canceled
+func loadGenerate(ctx context.Context, chTx chan<- types.Tx, multiplier int) {
 	for i := 0; i < math.MaxInt64; i++ {
 		// We keep generating the same 1000 keys over and over, with different values.
 		// This gives a reasonable load without putting too much data in the app.
 		id := i % 1000
 
-		bz := make([]byte, 2048) // 4kb hex-encoded
+		bz := make([]byte, 1024) // 1kb hex-encoded
 		_, err := rand.Read(bz)
 		if err != nil {
 			panic(fmt.Sprintf("Failed to read random bytes: %v", err))
@@ -80,7 +81,8 @@ func loadGenerate(ctx context.Context, chTx chan<- types.Tx) {
 
 		select {
 		case chTx <- tx:
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(time.Duration(100/multiplier) * time.Millisecond)
+
 		case <-ctx.Done():
 			close(chTx)
 			return
@@ -103,10 +105,17 @@ func loadProcess(ctx context.Context, testnet *e2e.Testnet, chTx <-chan types.Tx
 			if err != nil {
 				continue
 			}
+
+			// check that the node is up
+			_, err = client.Health(ctx)
+			if err != nil {
+				continue
+			}
+
 			clients[node.Name] = client
 		}
-		_, err = client.BroadcastTxCommit(ctx, tx)
-		if err != nil {
+
+		if _, err = client.BroadcastTxSync(ctx, tx); err != nil {
 			continue
 		}
 		chSuccess <- tx

test/e2e/runner/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strconv"
 
 	"github.com/spf13/cobra"
 
@@ -59,7 +60,7 @@ func NewCLI() *CLI {
 			ctx, loadCancel := context.WithCancel(context.Background())
 			defer loadCancel()
 			go func() {
-				err := Load(ctx, cli.testnet)
+				err := Load(ctx, cli.testnet, 1)
 				if err != nil {
 					logger.Error(fmt.Sprintf("Transaction load failed: %v", err.Error()))
 				}
@@ -165,10 +166,20 @@ func NewCLI() *CLI {
 	})
 
 	cli.root.AddCommand(&cobra.Command{
-		Use:   "load",
-		Short: "Generates transaction load until the command is cancelled",
-		RunE: func(cmd *cobra.Command, args []string) error {
-			return Load(context.Background(), cli.testnet)
+		Use:   "load [multiplier]",
+		Args:  cobra.MaximumNArgs(1),
+		Short: "Generates transaction load until the command is canceled",
+		RunE: func(cmd *cobra.Command, args []string) (err error) {
+			m := 1
+
+			if len(args) == 1 {
+				m, err = strconv.Atoi(args[0])
+				if err != nil {
+					return err
+				}
+			}
+
+			return Load(context.Background(), cli.testnet, m)
 		},
 	})
 
@@ -228,7 +239,7 @@ Does not run any perbutations.
 			ctx, loadCancel := context.WithCancel(context.Background())
 			defer loadCancel()
 			go func() {
-				err := Load(ctx, cli.testnet)
+				err := Load(ctx, cli.testnet, 1)
 				if err != nil {
 					logger.Error(fmt.Sprintf("Transaction load failed: %v", err.Error()))
 				}

test/e2e/tests/app_test.go

@@ -1,6 +1,7 @@
 package e2e_test
 
 import (
+	"bytes"
 	"fmt"
 	"math/rand"
 	"testing"
@@ -81,17 +82,17 @@ func TestApp_Tx(t *testing.T) {
 		value := fmt.Sprintf("%x", bz)
 		tx := types.Tx(fmt.Sprintf("%v=%v", key, value))
 
-		resp, err := client.BroadcastTxCommit(ctx, tx)
+		_, err = client.BroadcastTxSync(ctx, tx)
 		require.NoError(t, err)
 
-		// wait for the tx to be persisted in the tx indexer
-		time.Sleep(500 * time.Millisecond)
-
 		hash := tx.Hash()
-		txResp, err := client.Tx(ctx, hash, false)
-		require.NoError(t, err)
-		assert.Equal(t, txResp.Tx, tx)
-		assert.Equal(t, txResp.Height, resp.Height)
+		waitTime := 20 * time.Second
+		require.Eventuallyf(t, func() bool {
+			txResp, err := client.Tx(ctx, hash, false)
+			return err == nil && bytes.Equal(txResp.Tx, tx)
+		}, waitTime, time.Second,
+			"submitted tx wasn't committed after %v", waitTime,
+		)
 
 		// NOTE: we don't test abci query of the light client
 		if node.Mode == e2e.ModeLight {