test: fix flaky test

Helper method that performs initial assigning of master/replica roles
and is widely used in ConnectionPool tests was adjusted to wait for
the roles to be applied successfully.
Prior to this patch it doesn't, so sometimes subsequent test code might
work unexpectedly (the problem was caught with
TestConnectionHandlerOpenUpdateClose)

Closes #452

Author: elhimov

test_helpers/pool_helper.go

@@ -2,8 +2,10 @@ package test_helpers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
+	"sync"
 	"time"
 
 	"github.com/tarantool/go-tarantool/v2"
@@ -206,6 +208,45 @@ func SetInstanceRO(ctx context.Context, dialer tarantool.Dialer, connOpts tarant
 		return err
 	}
 
+	checkRole := func(conn *tarantool.Connection, isReplica bool) string {
+		data, err := conn.Do(tarantool.NewCallRequest("box.info")).Get()
+		switch {
+		case err != nil:
+			return fmt.Sprintf("failed to get box.info: %s", err)
+		case len(data) < 1:
+			return "box.info is empty"
+		}
+
+		status, statusFound := data[0].(map[interface{}]interface{})["status"]
+		readonly, readonlyFound := data[0].(map[interface{}]interface{})["ro"]
+		switch {
+		case !statusFound:
+			return "box.info.status is missing"
+		case status != "running":
+			return fmt.Sprintf("box.info.status='%s' (waiting for 'running')", status)
+		case !readonlyFound:
+			return "box.info.ro is missing"
+		case readonly != isReplica:
+			return fmt.Sprintf("box.info.ro='%v' (waiting for '%v')", readonly, isReplica)
+		default:
+			return ""
+		}
+	}
+
+	problem := "not checked yet"
+
+	// Wait for the role to be applied.
+	for len(problem) != 0 {
+		select {
+		case <-time.After(10 * time.Millisecond):
+		case <-ctx.Done():
+			return fmt.Errorf("%w: failed to apply role, the last problem: %s",
+				ctx.Err(), problem)
+		}
+
+		problem = checkRole(conn, isReplica)
+	}
+
 	return nil
 }
 
@@ -215,16 +256,23 @@ func SetClusterRO(dialers []tarantool.Dialer, connOpts tarantool.Opts,
 		return fmt.Errorf("number of servers should be equal to number of roles")
 	}
 
+	ctx, cancel := GetConnectContext()
+	defer cancel()
+
+	// Apply roles in parallel.
+	errs := make([]error, len(dialers))
+	var wg sync.WaitGroup
+	wg.Add(len(dialers))
 	for i, dialer := range dialers {
-		ctx, cancel := GetConnectContext()
-		err := SetInstanceRO(ctx, dialer, connOpts, roles[i])
-		cancel()
-		if err != nil {
-			return err
-		}
+		// Pass loop variables to avoid its closure.
+		go func(i int, dialer tarantool.Dialer) {
+			defer wg.Done()
+			errs[i] = SetInstanceRO(ctx, dialer, connOpts, roles[i])
+		}(i, dialer)
 	}
+	wg.Wait()
 
-	return nil
+	return errors.Join(errs...)
 }
 
 func StartTarantoolInstances(instsOpts []StartOpts) ([]*TarantoolInstance, error) {