Merge remote-tracking branch 'origin/main' into wb/issue-9790

test/e2e/Makefile

@@ -1,7 +1,7 @@
 all: docker generator runner
 
 docker:
-	docker build --tag tendermint/e2e-node -f docker/Dockerfile ../..
+	docker build --tag tendermint/e2e-node --tag tendermint/e2e-node:local-version -f docker/Dockerfile ../..
 
 # We need to build support for database backends into the app in
 # order to build a binary with a Tendermint node in it (for built-in

test/e2e/pkg/infra/docker/docker.go

@@ -72,7 +72,7 @@ services:
     labels:
       e2e: true
     container_name: {{ .Name }}
-    image: tendermint/e2e-node
+    image: tendermint/e2e-node:{{ .Version }}
 {{- if eq .ABCIProtocol "builtin" }}
     entrypoint: /usr/bin/entrypoint-builtin
 {{- end }}

test/e2e/pkg/manifest.go

@@ -81,6 +81,13 @@ type ManifestNode struct {
 	// is generated), and seed nodes run in seed mode with the PEX reactor enabled.
 	Mode string `toml:"mode"`
 
+	// Version specifies which version of Tendermint this node is. Specifying different
+	// versions for different nodes allows for testing the interaction of different
+	// node's compatibility. Note that in order to use a node at a particular version,
+	// there must be a docker image of the test app tagged with this version present
+	// on the machine where the test is being run.
+	Version string `toml:"version"`
+
 	// SyncApp specifies whether this node should use a synchronized application
 	// with an unsynchronized local client. By default this is `false`, meaning
 	// that the node will run an unsynchronized application with a synchronized

test/e2e/pkg/testnet.go

@@ -79,6 +79,7 @@ type Testnet struct {
 // Node represents a Tendermint node in a testnet.
 type Node struct {
 	Name             string
+	Version          string
 	Testnet          *Testnet
 	Mode             Mode
 	SyncApp          bool // Should we use a synchronized app with an unsynchronized local client?
@@ -174,8 +175,13 @@ func LoadTestnet(manifest Manifest, fname string, ifd InfrastructureData) (*Test
 		if ifd.Provider != "docker" {
 			extIP = ind.IPAddress
 		}
+		v := nodeManifest.Version
+		if v == "" {
+			v = "local-version"
+		}
 		node := &Node{
 			Name:             name,
+			Version:          v,
 			Testnet:          testnet,
 			PrivvalKey:       keyGen.Generate(manifest.KeyType),
 			NodeKey:          keyGen.Generate("ed25519"),

test/e2e/runner/load.go

@@ -26,6 +26,7 @@ func Load(ctx context.Context, testnet *e2e.Testnet) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	logger.Info("load", "msg", log.NewLazySprintf("Starting transaction load (%v workers)...", workerPoolSize))
 	started := time.Now()
 	u := [16]byte(uuid.New()) // generate run ID on startup
 
@@ -38,11 +39,7 @@ func Load(ctx context.Context, testnet *e2e.Testnet) error {
 		}
 
 		for w := 0; w < testnet.LoadTxConnections; w++ {
-			cli, err := n.Client()
-			if err != nil {
-				return err
-			}
-			go loadProcess(ctx, txCh, chSuccess, cli)
+			go loadProcess(ctx, txCh, chSuccess, n)
 		}
 	}
 
@@ -95,10 +92,12 @@ func loadGenerate(ctx context.Context, txCh chan<- types.Tx, testnet *e2e.Testne
 // is canceled.
 func createTxBatch(ctx context.Context, txCh chan<- types.Tx, testnet *e2e.Testnet, id []byte) {
 	wg := &sync.WaitGroup{}
+	genCh := make(chan struct{})
 	for i := 0; i < workerPoolSize; i++ {
 		wg.Add(1)
 		go func() {
-			for i := 0; i < testnet.LoadTxBatchSize; i++ {
+			defer wg.Done()
+			for range genCh {
 				tx, err := payload.NewBytes(&payload.Payload{
 					Id:          id,
 					Size:        uint64(testnet.LoadTxSizeBytes),
@@ -115,18 +114,33 @@ func createTxBatch(ctx context.Context, txCh chan<- types.Tx, testnet *e2e.Testn
 					return
 				}
 			}
-			wg.Done()
 		}()
 	}
+	for i := 0; i < testnet.LoadTxBatchSize; i++ {
+		select {
+		case genCh <- struct{}{}:
+		case <-ctx.Done():
+			break
+		}
+	}
+	close(genCh)
 	wg.Wait()
 }
 
 // loadProcess processes transactions by sending transactions received on the txCh
 // to the client.
-func loadProcess(ctx context.Context, txCh <-chan types.Tx, chSuccess chan<- struct{}, client *rpchttp.HTTP) {
+func loadProcess(ctx context.Context, txCh <-chan types.Tx, chSuccess chan<- struct{}, n *e2e.Node) {
+	var client *rpchttp.HTTP
 	var err error
 	s := struct{}{}
 	for tx := range txCh {
+		if client == nil {
+			client, err = n.Client()
+			if err != nil {
+				logger.Info("non-fatal error creating node client", "error", err)
+				continue
+			}
+		}
 		if _, err = client.BroadcastTxSync(ctx, tx); err != nil {
 			continue
 		}