fix(testcontainers): add active rw probe to vshard storages readiness check

The readiness check from 3ad515a relied on vshard.router.info() reporting
bucket.available_rw == vshard.router.bucket_count(). This is a passive
view of the router-side bucket accounting and does not always match the
actual rw-handshaked state of each storage in the early moments of a
fresh cluster: the third CI run (run 27611921481) still failed with
'Timeout exceeded while waiting for vshard storages to complete
handshake' on spring-data32.RepositoryViaJavaConfigTest, even though
the previous data27/data31 classes had started successfully within
the same Maven reactor.

Add a second, active tier to the check: after the passive
vshard.router.info() pre-check passes, perform a real read-write call
(vshard.router.callrw(1, 'box.info', {})) wrapped in pcall. This is the
most direct signal that the storage-side handshake has actually
finished for at least one storage. A successful rw call proves that
the very first CRUD call (e.g. truncate issued by @beforeeach) will
not fail with VHANDSHAKE_NOT_COMPLETE.

The probe is intentionally kept on a single Lua line because the
testcontainers command pipe flattens newlines, which would otherwise
confuse the Tarantool Lua parser around 'end' (observed in local
verification: the multi-line form produced 'end expected near local').

Verified locally on 3.5.0:
  - spring-data31 TarantoolTemplateViaJavaConfigTest: 35/35 PASS
  - spring-data27 TarantoolTemplateViaJavaConfigTest: 35/35 PASS

Author: opencode

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

@@ -87,6 +87,19 @@ public class VshardClusterContainer extends GenericContainer<VshardClusterContai
           + " local total = vshard.router.bucket_count();"
           + " return { total, b.available_rw or 0, b.available_ro or 0,"
           + " b.unreachable or 0, b.unknown or 0 }";
+  // Actively exercises a read-write call (box.info) on a real storage through the router.
+  // This is the most reliable signal that vshard storage-side handshakes have completed:
+  // a successful callrw proves that at least one storage can serve a rw request, which is
+  // exactly the precondition for CRUD operations such as truncate. Returns the call result
+  // (or the error message) so the caller can distinguish VHANDSHAKE_NOT_COMPLETE from a
+  // successful probe. Kept on a single Lua line because the testcontainers command pipe
+  // flattens newlines, which would otherwise confuse the Tarantool Lua parser around `end`.
+  protected static final String VSHARD_RW_PROBE_COMMAND =
+      "local vshard = require('vshard'); "
+          + "local function probe() return vshard.router.callrw(1, 'box.info', {}) end; "
+          + "local ok, res = pcall(probe); "
+          + "if ok then return { true, '' } end; "
+          + "return { false, tostring(res) }";
   private static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory());
 
   public static final String ENV_TARANTOOL_VERSION = "TARANTOOL_VERSION";
@@ -670,25 +683,53 @@ protected boolean vshardIsBootstrapped() {
   }
 
   /**
-   * Returns {@code true} when {@code vshard.router.info()} reports every replica as {@code
-   * available}, there are no unreachable or unknown buckets, and every bucket is ready for
-   * read-write (i.e. the vshard handshake with each storage has fully completed on the storage
-   * side, not just the network). See {@link #waitUntilVshardStoragesAreReady(int)} for rationale.
+   * Returns {@code true} once the vshard cluster is fully ready to serve read-write requests. The
+   * check is intentionally two-tiered:
+   *
+   * <ol>
+   *   <li><b>Pre-check</b> via {@code vshard.router.info()}: every replicaset has a master, every
+   *       replica is {@code available}, there are no unreachable/unknown buckets. Fails fast on
+   *       topology problems.
+   *   <li><b>Active probe</b> via {@code vshard.router.callrw(1, 'box.info', {})}: a real
+   *       read-write call routed to a storage. This is the most reliable signal that the vshard
+   *       handshake on each storage has completed; without it, the very first CRUD call (e.g. a
+   *       {@code truncate} issued by {@code @BeforeEach}) can fail with {@code
+   *       VHANDSHAKE_NOT_COMPLETE} (vshard code 40) even when the topology check passes.
+   * </ol>
+   *
+   * See {@link #waitUntilVshardStoragesAreReady(int)} for rationale.
    */
   protected boolean vshardStoragesAreReady() {
     try {
-      List<?> result =
+      List<?> preCheck =
           TarantoolContainerClientHelper.executeCommandDecoded(
               this, VSHARD_STORAGES_READY_COMMAND, null);
-      if (result.isEmpty()) {
-        logger().warn("Vshard storages readiness probe returned an empty response");
+      boolean preOk = !preCheck.isEmpty() && Boolean.TRUE.equals(preCheck.get(0));
+      if (!preOk) {
+        logVshardStoragesStatus();
+        return false;
+      }
+      // Active rw probe: succeeds only when at least one storage is fully handshaked for rw.
+      // The Lua command returns a list {ok, msg}; the YAML decoder wraps it once more, so we
+      // unwrap one level before inspecting the boolean.
+      List<?> wrapped =
+          TarantoolContainerClientHelper.executeCommandDecoded(this, VSHARD_RW_PROBE_COMMAND, null);
+      if (wrapped.size() < 1 || !(wrapped.get(0) instanceof List)) {
+        logger().warn("Vshard rw probe returned an unexpected response: {}", wrapped);
+        return false;
+      }
+      List<?> probe = (List<?>) wrapped.get(0);
+      if (probe.size() < 2) {
+        logger().warn("Vshard rw probe returned a malformed payload: {}", probe);
         return false;
       }
-      boolean ready = Boolean.TRUE.equals(result.get(0));
-      if (!ready) {
+      boolean rwOk = Boolean.TRUE.equals(probe.get(0));
+      if (!rwOk) {
+        String err = String.valueOf(probe.get(1));
+        logger().warn("Vshard rw probe failed: {}", err);
         logVshardStoragesStatus();
       }
-      return ready;
+      return rwOk;
     } catch (Exception e) {
       logger().warn("Vshard storages readiness probe failed: {}", e.getMessage());
       return false;