Support client-side caching (RESP3 tracking) for Redis Cluster

docs/user-guide/client-side-caching.md

@@ -0,0 +1,77 @@
+# Client-Side Caching
+
+Lettuce supports server-assisted client-side caching as described in the
+[Redis client-side caching documentation](https://redis.io/docs/latest/develop/reference/client-side-caching/).
+The `ClientSideCaching` utility creates a `CacheFrontend` that represents a
+two-level cache: values are first looked up in a local, application-provided
+cache and only fetched from Redis on a cache miss. Redis notifies the client
+through `CLIENT TRACKING` invalidation messages when a cached key is modified,
+and the corresponding local entry is evicted.
+
+## Standalone connections
+
+```java
+Map<String, String> clientCache = new ConcurrentHashMap<>();
+
+StatefulRedisConnection<String, String> connection = redisClient.connect();
+
+CacheFrontend<String, String> frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection,
+        TrackingArgs.Builder.enabled());
+
+String value = frontend.get(key);
+```
+
+The `CacheFrontend` is associated with the Redis connection. Close the
+frontend through `CacheFrontend.close()` to release the connection after use.
+
+The example above requires RESP3: invalidation messages are delivered as push
+messages on the connection itself, and the frontend evicts local entries
+automatically. With RESP2, Redis delivers invalidations only to a redirected
+client (`TrackingArgs.Builder.enabled().redirect(clientId)`), so the
+frontend's built-in invalidation listener never receives them. In that case
+subscribe to the `__redis__:invalidate` Pub/Sub channel on the redirect target
+connection and evict entries from the client-side cache
+(`CacheAccessor.evict(...)`) in the Pub/Sub listener yourself.
+
+## Redis Cluster connections
+
+Since version 7.7, client-side caching is also supported for Redis Cluster
+connections:
+
+```java
+Map<String, String> clientCache = new ConcurrentHashMap<>();
+
+StatefulRedisClusterConnection<String, String> connection = clusterClient.connect();
+
+CacheFrontend<String, String> frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection,
+        TrackingArgs.Builder.enabled());
+
+String value = frontend.get(key);
+```
+
+`CLIENT TRACKING` is enabled on each cluster node connection because
+invalidation messages for a key are emitted by the node that serves the key's
+slot. Upstream (master) nodes are tracked through their write-intent
+connections. When a `ReadFrom` setting routes reads to replicas, the replicas
+that the read policy can select are tracked through their read-intent
+connections as well.
+
+The following constraints apply to Redis Cluster:
+
+- **RESP3 is required.** Enabling the cache fails with an
+  `IllegalStateException` if a node connection did not negotiate RESP3.
+- **`REDIRECT` is not supported** as invalidation messages can originate from
+  any node. Redirected `TrackingArgs` are rejected with an
+  `IllegalArgumentException`.
+- **`OPTIN` is not supported** as the cache frontend does not issue
+  `CLIENT CACHING yes` before reads. Opt-in `TrackingArgs` are rejected with an
+  `IllegalArgumentException`.
+- **Topology and read-policy changes are not tracked.** Tracking is configured
+  for the topology and `ReadFrom` setting present when the cache is enabled.
+  Nodes added to the cluster afterwards do not have tracking enabled, and
+  changing the read policy through `setReadFrom(...)` does not track newly
+  selectable replicas. Applications that must observe such changes should
+  re-enable tracking afterwards.
+
+`FLUSHALL` and `FLUSHDB` emit a full invalidation, which clears the entire
+client-side cache.

mkdocs.yml

@@ -57,6 +57,7 @@ nav:
       - Kotlin API: user-guide/kotlin-api.md
       - Publish/Subscribe: user-guide/pubsub.md
       - Transactions/Multi: user-guide/transactions-multi.md
+      - Client-Side Caching: user-guide/client-side-caching.md
       - Redis Query Engine: user-guide/redis-search.md
       - Redis JSON: user-guide/redis-json.md
       - Redis Vector Sets: user-guide/vector-sets.md

src/main/java/io/lettuce/core/TrackingArgs.java

@@ -93,6 +93,16 @@ public TrackingArgs redirect(long clientId) {
         return this;
     }
 
+    /**
+     * Return whether invalidation messages are redirected to another client connection through {@code REDIRECT}.
+     *
+     * @return {@code true} if {@link #redirect(long)} was configured.
+     * @since 7.7
+     */
+    public boolean isRedirect() {
+        return redirect != null;
+    }
+
     /**
      * Enable tracking in broadcasting mode. In this mode invalidation messages are reported for all the prefixes specified,
      * regardless of the keys requested by the connection. Instead when the broadcasting mode is not enabled, Redis will track
@@ -145,6 +155,16 @@ public TrackingArgs optin() {
         return this;
     }
 
+    /**
+     * Return whether tracking requires a preceding {@code CLIENT CACHING yes} opt-in per read.
+     *
+     * @return {@code true} if {@link #optin()} was configured.
+     * @since 7.7
+     */
+    public boolean isOptin() {
+        return optin;
+    }
+
     /**
      * When broadcasting is NOT active, normally track keys in read only commands, unless they are called immediately after a
      * CLIENT CACHING no command.

src/main/java/io/lettuce/core/support/caching/CacheAccessor.java

@@ -59,4 +59,16 @@ static <K, V> CacheAccessor<K, V> forMap(Map<K, V> map) {
      */
     void evict(K key);
 
+    /**
+     * Evict all mappings from this cache. Invoked when Redis reports a full invalidation, for example after {@code FLUSHALL} or
+     * {@code FLUSHDB}.
+     * <p>
+     * The default implementation is a no-op; implementations should override this method to avoid serving stale entries after a
+     * flush.
+     *
+     * @since 7.7
+     */
+    default void clear() {
+    }
+
 }

src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java

@@ -1,14 +1,29 @@
 package io.lettuce.core.support.caching;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
 
+import io.lettuce.core.ReadFrom;
+import io.lettuce.core.RedisURI;
 import io.lettuce.core.StatefulRedisConnectionImpl;
 import io.lettuce.core.TrackingArgs;
 import io.lettuce.core.api.StatefulRedisConnection;
+import io.lettuce.core.cluster.StatefulRedisClusterConnectionImpl;
+import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.internal.LettuceAssert;
+import io.lettuce.core.models.role.RedisNodeDescription;
+import io.lettuce.core.protocol.ConnectionIntent;
+import io.lettuce.core.protocol.ProtocolVersion;
 
 /**
  * Utility to provide server-side assistance for client-side caches. This is a {@link CacheFrontend} that represents a two-level
@@ -31,6 +46,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Julien Ruaux
  * @since 6.0
  */
 public class ClientSideCaching<K, V> implements CacheFrontend<K, V> {
@@ -92,11 +108,165 @@ public static <K, V> CacheFrontend<K, V> create(CacheAccessor<K, V> cacheAccesso
         return create(cacheAccessor, redisCache);
     }
 
+    /**
+     * Enable server-assisted Client side caching for the given {@link CacheAccessor} and
+     * {@link StatefulRedisClusterConnection}.
+     * <p>
+     * {@code CLIENT TRACKING} is enabled on each cluster node connection: invalidation messages for a key are emitted by the
+     * node that serves the key's slot (or, for replica reads, by the replica the value was read from), so tracking must be
+     * active there. Upstream (master) nodes are tracked through their write-intent connections. Replicas are tracked through
+     * their read-intent connections only when the connection's {@link ReadFrom} setting can select them for reads, so no
+     * replica connections are opened when reads are routed to upstream nodes only (the default). Note that a keyless
+     * {@code CLIENT TRACKING} command issued through the cluster command API would be routed to the default connection only,
+     * whose push messages are not associated with cluster node connections.
+     * <p>
+     * Client-side caching for Redis Cluster requires RESP3. {@code TrackingArgs} redirection is not supported as invalidation
+     * messages can originate from any node, and {@code OPTIN} tracking is not supported as the cache frontend does not issue
+     * {@code CLIENT CACHING yes} before reads; this method throws {@link IllegalArgumentException} for redirected or opt-in
+     * tracking parameters and {@link IllegalStateException} if a node connection did not negotiate RESP3.
+     * <p>
+     * Tracking is configured for the topology and {@link ReadFrom} setting present at the time of this call. Nodes added to the
+     * cluster afterwards do not have tracking enabled, and changing the read policy through
+     * {@link StatefulRedisClusterConnection#setReadFrom} does not track newly selectable replicas. Applications that must
+     * observe such changes should re-enable tracking afterwards.
+     * <p>
+     * Note that the {@link CacheFrontend} is associated with a Redis connection. Make sure to {@link CacheFrontend#close()
+     * close} the frontend object to release the Redis connection after use.
+     *
+     * @param cacheAccessor the accessor used to interact with the client-side cache.
+     * @param connection the Redis Cluster connection to use. The connection will be associated with {@link CacheFrontend} and
+     *        must be closed through {@link CacheFrontend#close()}.
+     * @param tracking the tracking parameters.
+     * @param <K> Key type.
+     * @param <V> Value type.
+     * @return the {@link CacheFrontend} for value retrieval.
+     * @since 7.7
+     */
+    public static <K, V> CacheFrontend<K, V> enable(CacheAccessor<K, V> cacheAccessor,
+            StatefulRedisClusterConnection<K, V> connection, TrackingArgs tracking) {
+
+        LettuceAssert.isTrue(!tracking.isRedirect(),
+                "TrackingArgs REDIRECT is not supported for Redis Cluster client-side caching");
+        LettuceAssert.isTrue(!tracking.isOptin(), "TrackingArgs OPTIN is not supported for Redis Cluster client-side caching");
+
+        for (RedisClusterNode node : connection.getPartitions()) {
+            // use host/port connections: slot-routed commands are served by connections keyed
+            // by intent, host and port, not by the nodeId-keyed connections
+            if (isServingUpstream(node)) {
+                enableTracking(connection, node.getUri(), ConnectionIntent.WRITE, tracking);
+            }
+        }
+
+        for (RedisNodeDescription node : readCandidates(connection)) {
+            // reads from upstream nodes are served by the write-intent connections tracked above
+            if (node.getRole().isReplica()) {
+                enableTracking(connection, node.getUri(), ConnectionIntent.READ, tracking);
+            }
+        }
+
+        return create(cacheAccessor, connection);
+    }
+
+    private static Collection<RedisNodeDescription> readCandidates(StatefulRedisClusterConnection<?, ?> connection) {
+
+        ReadFrom readFrom = connection.getReadFrom();
+
+        if (readFrom == null) {
+            // without a ReadFrom setting, all reads are routed to upstream nodes
+            return Collections.emptyList();
+        }
+
+        // mirror the per-slot selection of the cluster connection provider: ReadFrom sees an
+        // upstream node and its replicas, not the whole cluster
+        Set<RedisNodeDescription> candidates = new LinkedHashSet<>();
+
+        for (RedisClusterNode upstream : connection.getPartitions()) {
+            if (isServingUpstream(upstream)) {
+                candidates.addAll(readFrom.select(slotGroup(connection, upstream)));
+            }
+        }
+
+        return candidates;
+    }
+
+    private static boolean isServingUpstream(RedisClusterNode node) {
+        // upstream nodes without slots cannot serve slot-routed reads or writes
+        return node.is(RedisClusterNode.NodeFlag.UPSTREAM) && !node.hasNoSlots();
+    }
+
+    private static ReadFrom.Nodes slotGroup(StatefulRedisClusterConnection<?, ?> connection, RedisClusterNode upstream) {
+
+        List<RedisNodeDescription> nodes = new ArrayList<>();
+        nodes.add(upstream);
+
+        for (RedisClusterNode node : connection.getPartitions()) {
+            // consider only replicas that contain data from replication, mirroring the connection provider
+            if (node.is(RedisClusterNode.NodeFlag.REPLICA) && upstream.getNodeId().equals(node.getSlaveOf())
+                    && node.getReplOffset() != 0) {
+                nodes.add(node);
+            }
+        }
+
+        return new ReadFrom.Nodes() {
+
+            @Override
+            public List<RedisNodeDescription> getNodes() {
+                return nodes;
+            }
+
+            @Override
+            public Iterator<RedisNodeDescription> iterator() {
+                return nodes.iterator();
+            }
+
+        };
+    }
+
+    private static <K, V> void enableTracking(StatefulRedisClusterConnection<K, V> connection, RedisURI uri,
+            ConnectionIntent intent, TrackingArgs tracking) {
+
+        StatefulRedisConnection<K, V> nodeConnection = connection.getConnection(uri.getHost(), uri.getPort(), intent);
+
+        ProtocolVersion protocolVersion = ((StatefulRedisConnectionImpl<K, V>) nodeConnection).getConnectionState()
+                .getNegotiatedProtocolVersion();
+        LettuceAssert.assertState(protocolVersion == ProtocolVersion.RESP3,
+                "Client-side caching for Redis Cluster requires RESP3");
+
+        nodeConnection.sync().clientTracking(tracking);
+    }
+
+    /**
+     * Create a server-assisted Client side caching for the given {@link CacheAccessor} and
+     * {@link StatefulRedisClusterConnection}. This method expects that client key tracking is already configured on the cluster
+     * node connections.
+     * <p>
+     * Note that the {@link CacheFrontend} is associated with a Redis connection. Make sure to {@link CacheFrontend#close()
+     * close} the frontend object to release the Redis connection after use.
+     *
+     * @param cacheAccessor the accessor used to interact with the client-side cache.
+     * @param connection the Redis Cluster connection to use. The connection will be associated with {@link CacheFrontend} and
+     *        must be closed through {@link CacheFrontend#close()}.
+     * @param <K> Key type.
+     * @param <V> Value type.
+     * @return the {@link CacheFrontend} for value retrieval.
+     * @since 7.7
+     */
+    public static <K, V> CacheFrontend<K, V> create(CacheAccessor<K, V> cacheAccessor,
+            StatefulRedisClusterConnection<K, V> connection) {
+
+        StatefulRedisClusterConnectionImpl<K, V> connectionImpl = (StatefulRedisClusterConnectionImpl) connection;
+        RedisCodec<K, V> codec = connectionImpl.getCodec();
+        RedisCache<K, V> redisCache = new ClusterRedisCache<>(connection, codec);
+
+        return create(cacheAccessor, redisCache);
+    }
+
     private static <K, V> CacheFrontend<K, V> create(CacheAccessor<K, V> cacheAccessor, RedisCache<K, V> redisCache) {
 
         ClientSideCaching<K, V> caching = new ClientSideCaching<>(cacheAccessor, redisCache);
 
         redisCache.addInvalidationListener(caching::notifyInvalidate);
+        redisCache.addClearListener(cacheAccessor::clear);
         caching.addInvalidationListener(cacheAccessor::evict);
 
         return caching;