fix(testcontainers): wait for vshard storages to complete handshake before declaring cluster ready

VshardClusterConfigurator#configure() previously declared the cluster
ready after three checks: router is up, vshard.router.bootstrap() returns
cleanly, and crud._VERSION is reachable on the router. None of these
verifies that individual storages have completed the vshard handshake
during the initial rebalance; the router can answer 'bootstrap ok' while
some storages are still in the VHANDSHAKE_NOT_COMPLETE (code 40) state,
and a CRUD request that targets such a storage fails immediately.

Observed in tests-crud-integration (3.5.0):
  TarantoolTemplateViaJavaConfigTest.testKVTemplateDeleteEntities
    -> CrudException: Failed to truncate for storage-002
       VHANDSHAKE_NOT_COMPLETE, 'Handshake with storage-002-a have not
       been completed yet'

Add a fourth readiness step: waitUntilVshardStoragesAreReady polls
vshard.router.info() until every replica in every replicaset is in
status='available' and there are no unreachable buckets, with a 120s
budget. This guarantees that any subsequent CRUD request hits a fully
handshaked storage.

Verified locally on 3.5.0:
  - TarantoolCrudClientTest: 347/347 PASS (was failing with
    VHANDSHAKE_NOT_COMPLETE in CI)
  - TarantoolTemplateViaJavaConfigTest (spring-data27, the originally
    failing test): 35/35 PASS

This is a testcontainers/infra change; it is added to the same PR as
the pooling fixups because both were investigated together after the
box-integration deadlocks were resolved.

Author: opencode

testcontainers/src/main/java/org/testcontainers/containers/cluster/vshard/VshardClusterConfigurator.java

@@ -47,6 +47,8 @@ public void configure() {
     this.container.waitUntilVshardIsBootstrapped(
         VshardClusterContainer.TIMEOUT_VSHARD_BOOTSTRAP_IN_SECONDS);
     this.container.waitUntilCrudIsUp(VshardClusterContainer.TIMEOUT_CRUD_HEALTH_IN_SECONDS);
+    this.container.waitUntilVshardStoragesAreReady(
+        VshardClusterContainer.TIMEOUT_VSHARD_STORAGES_READY_IN_SECONDS);
     this.configured.set(true);
   }
 

testcontainers/src/main/java/org/testcontainers/containers/cluster/vshard/VshardClusterContainer.java

@@ -65,6 +65,16 @@ public class VshardClusterContainer extends GenericContainer<VshardClusterContai
   protected static final String VSHARD_BOOTSTRAP_COMMAND =
       "return require('vshard').router.bootstrap({if_not_bootstrapped = true})";
   protected static final String ROUTER_HEALTH_COMMAND = "return box.info.status";
+  protected static final String VSHARD_STORAGES_READY_COMMAND =
+      "local info = require('vshard').router.info();"
+          + " for _, rs in pairs(info.replicasets or {}) do"
+          + "   if rs.master == nil or rs.master == box.NULL then return false end;"
+          + "   for _, r in pairs(rs.replicas or {}) do"
+          + "     if r.status == nil or r.status ~= 'available' then return false end;"
+          + "   end;"
+          + " end;"
+          + " if info.bucket and (info.bucket.unreachable or 0) > 0 then return false end;"
+          + " return true";
   private static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory());
 
   public static final String ENV_TARANTOOL_VERSION = "TARANTOOL_VERSION";
@@ -82,6 +92,7 @@ public class VshardClusterContainer extends GenericContainer<VshardClusterContai
   protected static final int TIMEOUT_CRUD_HEALTH_IN_SECONDS = 60;
   protected static final int TIMEOUT_ROUTER_HEALTH_IN_SECONDS = 90;
   protected static final int TIMEOUT_VSHARD_BOOTSTRAP_IN_SECONDS = 90;
+  protected static final int TIMEOUT_VSHARD_STORAGES_READY_IN_SECONDS = 120;
   protected static final int TIMEOUT_CONTAINER_START_IN_SECONDS = 600;
 
   protected final String TARANTOOL_RUN_DIR;
@@ -557,6 +568,23 @@ protected void waitUntilVshardIsBootstrapped(int secondsToWait) {
     }
   }
 
+  /**
+   * Waits until every vshard replica is in the {@code available} state (i.e. the vshard handshake
+   * with each storage has completed) and there are no unreachable buckets. {@link
+   * #vshardIsBootstrapped()} only verifies that {@code vshard.router.bootstrap} returned cleanly,
+   * which does not preclude individual storages from still being in {@code VHANDSHAKE_NOT_COMPLETE}
+   * (vshard code 40) during the initial rebalance. Issuing a CRUD request against such a cluster
+   * fails immediately, so the readiness check must additionally inspect the per-replica status
+   * surfaced by {@code vshard.router.info()}.
+   */
+  protected void waitUntilVshardStoragesAreReady(int secondsToWait) {
+    if (!waitUntilTrue(secondsToWait, this::vshardStoragesAreReady)) {
+      throw new RuntimeException(
+          "Timeout exceeded while waiting for vshard storages to complete handshake."
+              + " See the specific error in logs.");
+    }
+  }
+
   protected boolean waitUntilTrue(int secondsToWait, Supplier<Boolean> waitFunc) {
     int secondsPassed = 0;
     boolean result = waitFunc.get();
@@ -629,6 +657,31 @@ protected boolean vshardIsBootstrapped() {
     }
   }
 
+  /**
+   * Returns {@code true} when {@code vshard.router.info()} reports every replica as {@code
+   * available} and {@code info.bucket.unreachable == 0}, i.e. the vshard handshake has completed
+   * for all storages. See {@link #waitUntilVshardStoragesAreReady(int)} for rationale.
+   */
+  protected boolean vshardStoragesAreReady() {
+    try {
+      List<?> result =
+          TarantoolContainerClientHelper.executeCommandDecoded(
+              this, VSHARD_STORAGES_READY_COMMAND, null);
+      if (result.isEmpty()) {
+        logger().warn("Vshard storages readiness probe returned an empty response");
+        return false;
+      }
+      boolean ready = Boolean.TRUE.equals(result.get(0));
+      if (!ready) {
+        logger().warn("Vshard storages are not handshaked yet");
+      }
+      return ready;
+    } catch (Exception e) {
+      logger().warn("Vshard storages readiness probe failed: {}", e.getMessage());
+      return false;
+    }
+  }
+
   protected String getFileName(String filePath) {
     if (filePath == null || filePath.isBlank()) {
       throw new IllegalArgumentException("File path must not be null or empty");