client: add failover broadcast listener and sentinel active probe

src/main/java/io/valkey/JedisClusterInfoCache.java

@@ -26,13 +26,36 @@
 import io.valkey.annots.Internal;
 import io.valkey.exceptions.JedisClusterOperationException;
 import io.valkey.exceptions.JedisException;
+import io.valkey.util.IOUtils;
 import io.valkey.util.SafeEncoder;
 
+/**
+ * JedisClusterInfoCache maintains the cluster topology (slot-to-node mapping) and provides
+ * two complementary mechanisms to keep it up-to-date:
+ *
+ * <ol>
+ *   <li><b>Periodic topology refresh</b> (existing): a background thread calls
+ *       {@code CLUSTER SLOTS} at a configurable interval.</li>
+ *   <li><b>Failover broadcast listener</b> (new, feature/cluster-failover-pubsub): a
+ *       per-node subscriber thread listens on the {@code +switch-master} Pub/Sub channel.
+ *       When the Valkey server publishes a failover notification (introduced by the
+ *       companion server-side patch in {@code cluster_legacy.c}), the listener immediately
+ *       triggers a full slot-cache refresh so that clients react in real time instead of
+ *       waiting for the next periodic cycle or an error-driven refresh.</li>
+ * </ol>
+ *
+ * The failover listener is started automatically when the cache is constructed with a
+ * non-null {@code startNodes} set.  It can be disabled by calling
+ * {@link #setFailoverListenerEnabled(boolean)} before the first use.
+ */
 @Internal
 public class JedisClusterInfoCache {
 
   private static final Logger logger = LoggerFactory.getLogger(JedisClusterInfoCache.class);
 
+  /** Pub/Sub channel published by the server on every cluster master failover. */
+  static final String CLUSTER_FAILOVER_CHANNEL = "+switch-master";
+
   private final Map<String, ConnectionPool> nodes = new HashMap<>();
   private final ConnectionPool[] slots = new ConnectionPool[Protocol.CLUSTER_HASHSLOTS];
   private final HostAndPort[] slotNodes = new HostAndPort[Protocol.CLUSTER_HASHSLOTS];
@@ -48,11 +71,25 @@ public class JedisClusterInfoCache {
 
   private static final int MASTER_NODE_INDEX = 2;
 
+  /** Whether the failover broadcast listener is enabled (default: true). */
+  private volatile boolean failoverListenerEnabled = true;
+
   /**
-   * The single thread executor for the topology refresh task.
+   * The single thread executor for the periodic topology refresh task.
    */
   private ScheduledExecutorService topologyRefreshExecutor = null;
 
+  /**
+   * Background threads that subscribe to {@value #CLUSTER_FAILOVER_CHANNEL} on every
+   * known cluster node.  Each thread triggers an immediate slot-cache refresh when a
+   * failover notification is received.
+   */
+  private final List<FailoverListener> failoverListeners = new ArrayList<>();
+
+  // -------------------------------------------------------------------------
+  // Inner classes
+  // -------------------------------------------------------------------------
+
   class TopologyRefreshTask implements Runnable {
     @Override
     public void run() {
@@ -62,6 +99,94 @@ public void run() {
     }
   }
 
+  /**
+   * Background thread that subscribes to the {@value #CLUSTER_FAILOVER_CHANNEL} channel on
+   * a single cluster node.  When a failover notification is received the thread immediately
+   * calls {@link #renewClusterSlots(Connection)} so that the slot cache is updated without
+   * waiting for the next periodic refresh cycle.
+   *
+   * <p>The thread reconnects automatically after connection failures, using an exponential
+   * back-off capped at {@value #MAX_RETRY_WAIT_MILLIS} ms.
+   */
+  class FailoverListener extends Thread {
+
+    private static final long INITIAL_RETRY_WAIT_MILLIS = 1_000L;
+    private static final long MAX_RETRY_WAIT_MILLIS = 30_000L;
+
+    private final HostAndPort node;
+    private volatile Jedis jedis;
+    private volatile boolean running = false;
+
+    FailoverListener(HostAndPort node) {
+      super("ClusterFailoverListener-[" + node + "]");
+      this.node = node;
+      setDaemon(true);
+    }
+
+    @Override
+    public void run() {
+      running = true;
+      long retryWaitMillis = INITIAL_RETRY_WAIT_MILLIS;
+
+      while (running) {
+        try {
+          if (!running) break;
+
+          jedis = new Jedis(node, clientConfig);
+
+          // Perform an immediate refresh when (re-)connecting so that any failover
+          // that happened while the listener was disconnected is not missed.
+          logger.debug("FailoverListener connected to {}, performing initial slot refresh.", node);
+          renewClusterSlots(jedis.getClient());
+
+          // Reset back-off on successful connection.
+          retryWaitMillis = INITIAL_RETRY_WAIT_MILLIS;
+
+          jedis.subscribe(new JedisPubSub() {
+            @Override
+            public void onMessage(String channel, String message) {
+              logger.info(
+                  "Cluster failover notification received from {} on channel '{}': {}",
+                  node, channel, message);
+              // Trigger an immediate topology refresh.  We pass null so that the
+              // implementation falls back to startNodes / shuffled pool for the
+              // CLUSTER SLOTS query, which is safer than reusing the subscriber
+              // connection that is currently blocked in the subscribe loop.
+              renewClusterSlots(null);
+            }
+          }, CLUSTER_FAILOVER_CHANNEL);
+
+        } catch (JedisException e) {
+          if (running) {
+            logger.warn(
+                "FailoverListener lost connection to {}. Retrying in {}ms.", node, retryWaitMillis, e);
+            try {
+              Thread.sleep(retryWaitMillis);
+            } catch (InterruptedException ie) {
+              Thread.currentThread().interrupt();
+              logger.debug("FailoverListener interrupted while sleeping.", ie);
+            }
+            // Exponential back-off.
+            retryWaitMillis = Math.min(retryWaitMillis * 2, MAX_RETRY_WAIT_MILLIS);
+          } else {
+            logger.debug("FailoverListener shutting down for {}.", node);
+          }
+        } finally {
+          IOUtils.closeQuietly(jedis);
+        }
+      }
+    }
+
+    void shutdown() {
+      running = false;
+      IOUtils.closeQuietly(jedis);
+    }
+  }
+
+  // -------------------------------------------------------------------------
+  // Constructors
+  // -------------------------------------------------------------------------
+
   public JedisClusterInfoCache(final JedisClientConfig clientConfig, final Set<HostAndPort> startNodes) {
     this(clientConfig, null, startNodes);
   }
@@ -85,6 +210,49 @@ public JedisClusterInfoCache(final JedisClientConfig clientConfig,
     }
   }
 
+  // -------------------------------------------------------------------------
+  // Failover listener management
+  // -------------------------------------------------------------------------
+
+  /**
+   * Enable or disable the failover broadcast listener.  Must be called before the cache is
+   * used (i.e. before {@link #discoverClusterNodesAndSlots(Connection)} is invoked).
+   *
+   * @param enabled {@code true} to enable (default), {@code false} to disable
+   */
+  public void setFailoverListenerEnabled(boolean enabled) {
+    this.failoverListenerEnabled = enabled;
+  }
+
+  /**
+   * Start a {@link FailoverListener} for every node currently known to the cache.
+   * Called automatically after the initial slot discovery.
+   */
+  private void startFailoverListeners() {
+    if (!failoverListenerEnabled || startNodes == null || startNodes.isEmpty()) {
+      return;
+    }
+    // Stop any existing listeners first (e.g. after a full reset).
+    stopFailoverListeners();
+    for (HostAndPort node : startNodes) {
+      FailoverListener listener = new FailoverListener(node);
+      failoverListeners.add(listener);
+      listener.start();
+      logger.info("Started FailoverListener for cluster node {}.", node);
+    }
+  }
+
+  private void stopFailoverListeners() {
+    for (FailoverListener listener : failoverListeners) {
+      listener.shutdown();
+    }
+    failoverListeners.clear();
+  }
+
+  // -------------------------------------------------------------------------
+  // Slot discovery
+  // -------------------------------------------------------------------------
+
   /**
    * Check whether the number and order of slots in the cluster topology are equal to CLUSTER_HASHSLOTS
    * @param slotsInfo the cluster topology
@@ -148,6 +316,8 @@ public void discoverClusterNodesAndSlots(Connection jedis) {
     } finally {
       w.unlock();
     }
+    // Start failover listeners after the initial topology is known.
+    startFailoverListeners();
   }
 
   public void renewClusterSlots(Connection jedis) {
@@ -381,6 +551,7 @@ public void reset() {
 
   public void close() {
     reset();
+    stopFailoverListeners();
     if (topologyRefreshExecutor != null) {
       logger.info("Cluster topology refresh shutdown, startNodes: {}", startNodes);
       topologyRefreshExecutor.shutdownNow();