diff --git a/docs/user-guide/client-side-caching.md b/docs/user-guide/client-side-caching.md new file mode 100644 index 0000000000..f3a74a6603 --- /dev/null +++ b/docs/user-guide/client-side-caching.md @@ -0,0 +1,84 @@ +# 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 clientCache = new ConcurrentHashMap<>(); + +StatefulRedisConnection connection = redisClient.connect(); + +CacheFrontend 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. Full +invalidations such as `FLUSHALL`/`FLUSHDB` deliver a `null` message without +keys — clear the entire client-side cache (`CacheAccessor.clear()`) when the +invalidation message is `null`. + +## Redis Cluster connections + +Since version 7.7, client-side caching is also supported for Redis Cluster +connections: + +```java +Map clientCache = new ConcurrentHashMap<>(); + +StatefulRedisClusterConnection connection = clusterClient.connect(); + +CacheFrontend 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`. +- **Prefix-limited `BCAST` is not supported** as the cache frontend caches all + keys regardless of prefix, so keys outside the configured prefixes would + never be invalidated. Prefixed `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. diff --git a/mkdocs.yml b/mkdocs.yml index f202dd2b75..b90689d070 100644 --- a/mkdocs.yml +++ b/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 diff --git a/src/main/java/io/lettuce/core/TrackingArgs.java b/src/main/java/io/lettuce/core/TrackingArgs.java index d028cf777d..0fba741356 100644 --- a/src/main/java/io/lettuce/core/TrackingArgs.java +++ b/src/main/java/io/lettuce/core/TrackingArgs.java @@ -93,6 +93,56 @@ public TrackingArgs redirect(long clientId) { return this; } + /** + * Return whether key tracking is enabled. + * + * @return {@code true} if {@link #enabled(boolean)} was configured with {@code true}. + * @since 7.7 + */ + public boolean isEnabled() { + return enabled; + } + + /** + * 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; + } + + /** + * Return whether broadcasting invalidation messages are limited to key prefixes through {@code PREFIX}. + * + * @return {@code true} if {@link #prefixes(String...)} was configured with at least one prefix. + * @since 7.7 + */ + public boolean hasPrefixes() { + return prefixes != null && prefixes.length > 0; + } + + /** + * Create a copy of {@code this} {@link TrackingArgs} to guard against later modifications of a shared, mutable instance. + * + * @return a new {@link TrackingArgs} with the same configuration. + * @since 7.7 + */ + public TrackingArgs copy() { + + TrackingArgs copy = new TrackingArgs(); + copy.enabled = this.enabled; + copy.redirect = this.redirect; + copy.bcast = this.bcast; + copy.prefixes = this.prefixes == null ? null : this.prefixes.clone(); + copy.prefixCharset = this.prefixCharset; + copy.optin = this.optin; + copy.optout = this.optout; + copy.noloop = this.noloop; + return copy; + } + /** * 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 +195,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. diff --git a/src/main/java/io/lettuce/core/support/caching/CacheAccessor.java b/src/main/java/io/lettuce/core/support/caching/CacheAccessor.java index 0e9f789c51..1a2c8ff66b 100644 --- a/src/main/java/io/lettuce/core/support/caching/CacheAccessor.java +++ b/src/main/java/io/lettuce/core/support/caching/CacheAccessor.java @@ -59,4 +59,16 @@ static CacheAccessor forMap(Map 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}. + *

+ * 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() { + } + } diff --git a/src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java b/src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java index e512c88f7f..870bf3daeb 100644 --- a/src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java +++ b/src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java @@ -1,14 +1,35 @@ package io.lettuce.core.support.caching; +import java.net.SocketAddress; +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.RedisChannelHandler; +import io.lettuce.core.RedisConnectionException; +import io.lettuce.core.RedisConnectionStateListener; +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; +import io.netty.util.internal.logging.InternalLogger; +import io.netty.util.internal.logging.InternalLoggerFactory; /** * Utility to provide server-side assistance for client-side caches. This is a {@link CacheFrontend} that represents a two-level @@ -31,10 +52,13 @@ * @param Key type. * @param Value type. * @author Mark Paluch + * @author Julien Ruaux * @since 6.0 */ public class ClientSideCaching implements CacheFrontend { + private static final InternalLogger LOG = InternalLoggerFactory.getInstance(ClientSideCaching.class); + private final CacheAccessor cacheAccessor; private final RedisCache redisCache; @@ -92,11 +116,250 @@ public static CacheFrontend create(CacheAccessor cacheAccesso return create(cacheAccessor, redisCache); } + /** + * Enable server-assisted Client side caching for the given {@link CacheAccessor} and + * {@link StatefulRedisClusterConnection}. + *

+ * {@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. + *

+ * 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. + *

+ * 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. + *

+ * 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 Key type. + * @param Value type. + * @return the {@link CacheFrontend} for value retrieval. + * @throws IllegalArgumentException if {@code tracking} is {@code null}, disabled, redirected, opt-in or prefix-limited. + * @throws IllegalStateException if a node connection did not negotiate RESP3. + * @since 7.7 + */ + public static CacheFrontend enable(CacheAccessor cacheAccessor, + StatefulRedisClusterConnection connection, TrackingArgs tracking) { + + LettuceAssert.notNull(tracking, "TrackingArgs must not be null"); + LettuceAssert.isTrue(tracking.isEnabled(), "TrackingArgs must be enabled for Redis Cluster client-side caching"); + 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"); + LettuceAssert.isTrue(!tracking.hasPrefixes(), + "TrackingArgs PREFIX is not supported for Redis Cluster client-side caching as the cache frontend caches " + + "all keys regardless of prefix"); + + // snapshot the mutable args so reconnect replay does not observe later modifications + TrackingArgs trackingSnapshot = tracking.copy(); + + // clear cached entries when a tracked node connection reconnects: Redis dropped the tracking + // state with the old connection, so existing entries would no longer receive invalidations + Runnable clearAction = cacheAccessor::clear; + + List rollbackActions = new ArrayList<>(); + + try { + enableClusterTracking(connection, trackingSnapshot, clearAction, rollbackActions); + } catch (RuntimeException e) { + // do not leave earlier nodes half-configured with tracking and reconnect listeners + rollback(rollbackActions); + throw e; + } + + return create(cacheAccessor, connection); + } + + private static void enableClusterTracking(StatefulRedisClusterConnection connection, TrackingArgs tracking, + Runnable clearAction, List rollbackActions) { + + 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)) { + rollbackActions.add(enableTracking(connection, node.getUri(), ConnectionIntent.WRITE, tracking, clearAction)); + } + } + + for (RedisNodeDescription node : readCandidates(connection)) { + // reads from upstream nodes are served by the write-intent connections tracked above + if (node.getRole().isReplica()) { + try { + rollbackActions + .add(enableTracking(connection, node.getUri(), ConnectionIntent.READ, tracking, clearAction)); + } catch (RedisConnectionException e) { + // the read path tolerates unavailable replicas as long as another candidate connects, + // so an unreachable replica must not fail cache setup + LOG.warn("Cannot enable key tracking on replica {}, reads from this replica are not tracked", node.getUri(), + e); + } + } + } + } + + private static void rollback(List rollbackActions) { + + for (Runnable rollbackAction : rollbackActions) { + try { + rollbackAction.run(); + } catch (RuntimeException e) { + LOG.warn("Cannot roll back key tracking", e); + } + } + } + + private static Collection 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 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 boolean isReadCandidate(RedisClusterNode upstream, RedisClusterNode node) { + + if (upstream.getNodeId().equals(node.getNodeId())) { + return true; + } + + // consider only replicas that contain data from replication, mirroring the connection provider + return upstream.getNodeId().equals(node.getSlaveOf()) && node.getReplOffset() != 0; + } + + private static ReadFrom.Nodes slotGroup(StatefulRedisClusterConnection connection, RedisClusterNode upstream) { + + // preserve the partition order, mirroring PooledClusterConnectionProvider#getReadCandidates + List nodes = new ArrayList<>(); + + for (RedisClusterNode node : connection.getPartitions()) { + if (isReadCandidate(upstream, node)) { + nodes.add(node); + } + } + + return new ReadFrom.Nodes() { + + @Override + public List getNodes() { + return nodes; + } + + @Override + public Iterator iterator() { + return nodes.iterator(); + } + + }; + } + + private static Runnable enableTracking(StatefulRedisClusterConnection connection, RedisURI uri, + ConnectionIntent intent, TrackingArgs tracking, Runnable clearAction) { + + StatefulRedisConnection nodeConnection = connection.getConnection(uri.getHost(), uri.getPort(), intent); + StatefulRedisConnectionImpl nodeConnectionImpl = (StatefulRedisConnectionImpl) nodeConnection; + + ProtocolVersion protocolVersion = nodeConnectionImpl.getConnectionState().getNegotiatedProtocolVersion(); + LettuceAssert.assertState(protocolVersion == ProtocolVersion.RESP3, + "Client-side caching for Redis Cluster requires RESP3"); + + nodeConnection.sync().clientTracking(tracking); + + // CLIENT TRACKING is connection state that the reconnect handshake does not restore, re-apply it + RedisConnectionStateListener trackingRestorer = new RedisConnectionStateListener() { + + @Override + public void onRedisConnected(RedisChannelHandler connectionHandler, SocketAddress socketAddress) { + + // Redis dropped the tracking state with the old connection: entries cached so far no + // longer receive invalidations and must be evicted + clearAction.run(); + + nodeConnection.async().clientTracking(tracking).whenComplete((result, e) -> { + if (e != null) { + LOG.warn("Cannot re-enable key tracking on {} after reconnect, reads from this node are not tracked", + uri, e); + } else { + // commands buffered during the outage replay before this listener runs, so reads may + // have populated the cache untracked; evict them now that tracking is restored + clearAction.run(); + } + }); + } + + }; + nodeConnectionImpl.addListener(trackingRestorer); + + return () -> { + nodeConnectionImpl.removeListener(trackingRestorer); + nodeConnection.async().clientTracking(TrackingArgs.Builder.enabled(false)); + }; + } + + /** + * 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. + *

+ * 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 Key type. + * @param Value type. + * @return the {@link CacheFrontend} for value retrieval. + * @since 7.7 + */ + public static CacheFrontend create(CacheAccessor cacheAccessor, + StatefulRedisClusterConnection connection) { + + StatefulRedisClusterConnectionImpl connectionImpl = (StatefulRedisClusterConnectionImpl) connection; + RedisCodec codec = connectionImpl.getCodec(); + RedisCache redisCache = new ClusterRedisCache<>(connection, codec); + + return create(cacheAccessor, redisCache); + } + private static CacheFrontend create(CacheAccessor cacheAccessor, RedisCache redisCache) { ClientSideCaching caching = new ClientSideCaching<>(cacheAccessor, redisCache); redisCache.addInvalidationListener(caching::notifyInvalidate); + redisCache.addClearListener(cacheAccessor::clear); caching.addInvalidationListener(cacheAccessor::evict); return caching; diff --git a/src/main/java/io/lettuce/core/support/caching/ClusterRedisCache.java b/src/main/java/io/lettuce/core/support/caching/ClusterRedisCache.java new file mode 100644 index 0000000000..4e57dbbd12 --- /dev/null +++ b/src/main/java/io/lettuce/core/support/caching/ClusterRedisCache.java @@ -0,0 +1,80 @@ +package io.lettuce.core.support.caching; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; + +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.codec.RedisCodec; + +/** + * {@link RedisCache} implementation for Redis Cluster using {@code GET} and {@code SET} operations to map cache values to + * top-level keys. Invalidation messages are consumed from all cluster node connections through the cluster push listener. + * + * @param Key type. + * @param Value type. + * @author Julien Ruaux + * @since 7.7 + */ +class ClusterRedisCache implements RedisCache { + + private final StatefulRedisClusterConnection connection; + + private final RedisCodec codec; + + private final List clearListeners = new CopyOnWriteArrayList<>(); + + public ClusterRedisCache(StatefulRedisClusterConnection connection, RedisCodec codec) { + this.connection = connection; + this.codec = codec; + } + + @Override + public V get(K key) { + return connection.sync().get(key); + } + + @Override + public void put(K key, V value) { + connection.sync().set(key, value); + } + + @Override + public void addInvalidationListener(Consumer listener) { + + connection.addListener((node, message) -> { + if (message.getType().equals("invalidate")) { + + // decode only the key payload, the frame type element is not a key + List content = message.getContent(); + List keys = (List) content.get(1); + + if (keys == null) { + // null payload indicates a full invalidation, e.g. after FLUSHALL/FLUSHDB + clearListeners.forEach(Runnable::run); + } else { + for (Object key : keys) { + if (key == null) { + clearListeners.forEach(Runnable::run); + } else { + // decode from a duplicate so shared buffers are not consumed for other listeners + listener.accept(codec.decodeKey(((ByteBuffer) key).duplicate())); + } + } + } + } + }); + } + + @Override + public void addClearListener(Runnable listener) { + clearListeners.add(listener); + } + + @Override + public void close() { + connection.close(); + } + +} diff --git a/src/main/java/io/lettuce/core/support/caching/DefaultRedisCache.java b/src/main/java/io/lettuce/core/support/caching/DefaultRedisCache.java index 474a0df9d5..8e8d578415 100644 --- a/src/main/java/io/lettuce/core/support/caching/DefaultRedisCache.java +++ b/src/main/java/io/lettuce/core/support/caching/DefaultRedisCache.java @@ -1,6 +1,9 @@ package io.lettuce.core.support.caching; +import java.nio.ByteBuffer; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.codec.RedisCodec; @@ -17,6 +20,8 @@ class DefaultRedisCache implements RedisCache { private final RedisCodec codec; + private final List clearListeners = new CopyOnWriteArrayList<>(); + public DefaultRedisCache(StatefulRedisConnection connection, RedisCodec codec) { this.connection = connection; this.codec = codec; @@ -33,18 +38,37 @@ public void put(K key, V value) { } @Override - public void addInvalidationListener(java.util.function.Consumer listener) { + public void addInvalidationListener(Consumer listener) { connection.addListener(message -> { if (message.getType().equals("invalidate")) { - List content = message.getContent(codec::decodeKey); - List keys = (List) content.get(1); - keys.forEach(listener); + // decode only the key payload, the frame type element is not a key + List content = message.getContent(); + List keys = (List) content.get(1); + + if (keys == null) { + // null payload indicates a full invalidation, e.g. after FLUSHALL/FLUSHDB + clearListeners.forEach(Runnable::run); + } else { + for (Object key : keys) { + if (key == null) { + clearListeners.forEach(Runnable::run); + } else { + // decode from a duplicate so shared buffers are not consumed for other listeners + listener.accept(codec.decodeKey(((ByteBuffer) key).duplicate())); + } + } + } } }); } + @Override + public void addClearListener(Runnable listener) { + clearListeners.add(listener); + } + @Override public void close() { connection.close(); diff --git a/src/main/java/io/lettuce/core/support/caching/MapCacheAccessor.java b/src/main/java/io/lettuce/core/support/caching/MapCacheAccessor.java index ac9f21f109..5da91ee024 100644 --- a/src/main/java/io/lettuce/core/support/caching/MapCacheAccessor.java +++ b/src/main/java/io/lettuce/core/support/caching/MapCacheAccessor.java @@ -33,4 +33,9 @@ public void evict(K key) { map.remove(key); } + @Override + public void clear() { + map.clear(); + } + } diff --git a/src/main/java/io/lettuce/core/support/caching/RedisCache.java b/src/main/java/io/lettuce/core/support/caching/RedisCache.java index ded00d04f8..7386d6c620 100644 --- a/src/main/java/io/lettuce/core/support/caching/RedisCache.java +++ b/src/main/java/io/lettuce/core/support/caching/RedisCache.java @@ -34,6 +34,16 @@ public interface RedisCache { */ void addInvalidationListener(java.util.function.Consumer listener); + /** + * Register a {@code listener} that is notified if Redis reports a full invalidation of the cache, for example after + * {@code FLUSHALL} or {@code FLUSHDB}. The default implementation does not notify the listener. + * + * @param listener the listener to notify. + * @since 7.7 + */ + default void addClearListener(Runnable listener) { + } + /** * Closes this Redis cache and releases any connections associated with it. If the cache is already closed then invoking * this method has no effect. diff --git a/src/test/java/io/lettuce/core/support/caching/ClientsideCachingIntegrationTests.java b/src/test/java/io/lettuce/core/support/caching/ClientsideCachingIntegrationTests.java index 8ccd81fa4b..b6d3d5f697 100644 --- a/src/test/java/io/lettuce/core/support/caching/ClientsideCachingIntegrationTests.java +++ b/src/test/java/io/lettuce/core/support/caching/ClientsideCachingIntegrationTests.java @@ -164,6 +164,32 @@ void serverAssistedCachingShouldFetchValueFromRedis() { frontend.close(); } + @Test + void serverAssistedCachingShouldClearOnFlush() { + + Map clientCache = new ConcurrentHashMap<>(); + + StatefulRedisConnection otherParty = redisClient.connect(); + RedisCommands commands = otherParty.sync(); + + commands.set(key, value); + + StatefulRedisConnection connection = redisClient.connect(); + CacheFrontend frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection, + TrackingArgs.Builder.enabled()); + + assertThat(frontend.get(key)).isEqualTo(value); + assertThat(clientCache).hasSize(1); + + commands.flushall(); + + Wait.untilTrue(clientCache::isEmpty).waitOrTimeout(); + assertThat(frontend.get(key)).isNull(); + + otherParty.close(); + frontend.close(); + } + @Test void serverAssistedCachingShouldExpireValueFromRedis() throws InterruptedException { diff --git a/src/test/java/io/lettuce/core/support/caching/ClusterClientsideCachingIntegrationTests.java b/src/test/java/io/lettuce/core/support/caching/ClusterClientsideCachingIntegrationTests.java new file mode 100644 index 0000000000..5144b04c91 --- /dev/null +++ b/src/test/java/io/lettuce/core/support/caching/ClusterClientsideCachingIntegrationTests.java @@ -0,0 +1,140 @@ +package io.lettuce.core.support.caching; + +import static io.lettuce.TestTags.INTEGRATION_TEST; +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.IntStream; + +import javax.inject.Inject; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import io.lettuce.core.TrackingArgs; +import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.SlotHash; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.cluster.models.partitions.RedisClusterNode; +import io.lettuce.test.LettuceExtension; +import io.lettuce.test.Wait; +import io.lettuce.test.condition.EnabledOnCommand; + +/** + * Integration tests for server-side assisted cache invalidation using Redis Cluster. + * + * @author Julien Ruaux + */ +@Tag(INTEGRATION_TEST) +@ExtendWith(LettuceExtension.class) +@EnabledOnCommand("ACL") +public class ClusterClientsideCachingIntegrationTests { + + private final RedisClusterClient clusterClient; + + @Inject + public ClusterClientsideCachingIntegrationTests(RedisClusterClient clusterClient) { + this.clusterClient = clusterClient; + } + + @BeforeEach + void setUp() { + + try (StatefulRedisClusterConnection connection = clusterClient.connect()) { + connection.sync().flushall(); + } + } + + @Test + void serverAssistedCachingShouldUseClientCache() { + + Map clientCache = new ConcurrentHashMap<>(); + + StatefulRedisClusterConnection otherParty = clusterClient.connect(); + StatefulRedisClusterConnection connection = clusterClient.connect(); + + CacheFrontend frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection, + TrackingArgs.Builder.enabled()); + + String key = "key"; + otherParty.sync().set(key, "value"); + + assertThat(frontend.get(key)).isEqualTo("value"); + assertThat(clientCache).hasSize(1); + + otherParty.close(); + frontend.close(); + } + + @Test + void serverAssistedCachingShouldInvalidateAcrossAllNodes() { + + Map clientCache = new ConcurrentHashMap<>(); + + StatefulRedisClusterConnection otherParty = clusterClient.connect(); + StatefulRedisClusterConnection connection = clusterClient.connect(); + + CacheFrontend frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection, + TrackingArgs.Builder.enabled()); + + // Invalidation messages are emitted by the node serving a key's slot, so exercise a key on + // each upstream node. + for (RedisClusterNode node : connection.getPartitions()) { + + if (!node.is(RedisClusterNode.NodeFlag.UPSTREAM)) { + continue; + } + + String key = keyOwnedBy(connection, node); + + otherParty.sync().set(key, "value"); + assertThat(frontend.get(key)).isEqualTo("value"); + assertThat(clientCache).containsKey(key); + + otherParty.sync().set(key, "changed"); + + Wait.untilTrue(() -> !clientCache.containsKey(key)).waitOrTimeout(); + assertThat(frontend.get(key)).isEqualTo("changed"); + } + + otherParty.close(); + frontend.close(); + } + + @Test + void serverAssistedCachingShouldClearOnFlush() { + + Map clientCache = new ConcurrentHashMap<>(); + + StatefulRedisClusterConnection otherParty = clusterClient.connect(); + StatefulRedisClusterConnection connection = clusterClient.connect(); + + CacheFrontend frontend = ClientSideCaching.enable(CacheAccessor.forMap(clientCache), connection, + TrackingArgs.Builder.enabled()); + + String key = "key"; + otherParty.sync().set(key, "value"); + + assertThat(frontend.get(key)).isEqualTo("value"); + assertThat(clientCache).hasSize(1); + + otherParty.sync().flushall(); + + Wait.untilTrue(clientCache::isEmpty).waitOrTimeout(); + assertThat(frontend.get(key)).isNull(); + + otherParty.close(); + frontend.close(); + } + + private static String keyOwnedBy(StatefulRedisClusterConnection connection, RedisClusterNode node) { + + return IntStream.range(0, SlotHash.SLOT_COUNT).mapToObj(i -> "key-" + i) + .filter(key -> node.hasSlot(SlotHash.getSlot(key))).findFirst() + .orElseThrow(() -> new IllegalStateException("No key found for node " + node.getNodeId())); + } + +}