diff --git a/design/client-side-cache.md b/design/client-side-cache.md index 53c7c0ae9..8789d90f7 100644 --- a/design/client-side-cache.md +++ b/design/client-side-cache.md @@ -1,4 +1,4 @@ -# Client-side caching in v4: the design as built +# Client-side caching in v4: the design as built Status: current as of 2026-10-12 (`a22ba168`; shipped in `4.0.127-alpha`). This is the record of **why** the cache is the way it is - each mechanism with what was decided, what was rejected, and the evidence - gathered from the @@ -186,6 +186,16 @@ invalidation, validating against a key that changed); the cross-index. The rule is that the cache is never staler than the member it answers for. Server invalidations and flushes stay per member. Rejected: flushing on every switch - with equal weights a group picks by latency, so the active member can change on any health pass, and a flush would throw away a healthy member's cache each time. +- **In a group the cache is configured once, on `MultiGroupOptions.ClientCache` (2026-10-27), and a member's own + `ConfigurationOptions.ClientCache` is refused.** Members hold separate caches and the group answers from the + active one, so members that cached differently - or one that did not cache at all - would change what callers + get on every failover; inheriting one definition is how they agree, and allowing a per-member override would be + allowing them to disagree. Plumbed like the group's circuit breaker: alongside the configuration, never written + into it (a sentinel member gets a clone, since its cache is built with the connection). A sentinel connection + itself never builds a cache. `CacheOptions.Prefixes` is now `IReadOnlyList` rather than strings - + keys are bytes, and a prefix that is not valid UTF-8 had no spelling - rendered to a private byte snapshot + once (a key's whole rendering, context prefix included), which both the matcher and the handshake's `PREFIX` + arguments read; nothing of the caller's list or arrays is kept, and the getter hands out copies. - **An inactive member is kept honest exactly as an active one is.** Group members always run `HeartbeatConsistencyChecks`, so each endpoint is pinged every heartbeat whether or not it is active; a half-open tracking connection trips the dead-socket rule (commands timing out, nothing received for 4x the timeout), and diff --git a/docs/ClientSideCaching.md b/docs/ClientSideCaching.md index 54df5941d..50415ab63 100644 --- a/docs/ClientSideCaching.md +++ b/docs/ClientSideCaching.md @@ -37,7 +37,7 @@ Requirements: The cache is configured in code only - it is not part of the connection string, because its settings are correctness choices that deserve to be read before they are set. -There is **one cache per multiplexer**, shared by every database, every `IDatabase`, and every context made from it. +There is **one cache per multiplexer**, shared by every database, every `IDatabase`, and every context made from it. On a [connection group](Failover) each member has a cache of its own and the options are set once, on `MultiGroupOptions.ClientCache` rather than on the members - see [Connection groups](#connection-groups). What it covers --- @@ -106,7 +106,7 @@ Controlling it |---|---|---| | `Enabled` | `true` | Whether the cache exists. Setting `ClientCache` at all is what turns caching on; this lets you keep the options and switch it off. | | `TrackingMode` | `Default` (= `Tracking` today, or `TrackingByPrefix` when `Prefixes` are set) | How the server decides which changes to announce, and whether reads are cached by default (`Tracking`, `Broadcast`), only inside `Prefixes` (`TrackingByPrefix`), or only when a call asks (`TrackingOptIn`) - see below. | -| `Prefixes` | none (all keys) | Which keys to cache. Setting any selects `TrackingByPrefix`, where the list decides for calls that pass no flag; under `Broadcast` it is also sent to the server to filter what is announced. | +| `Prefixes` | none (all keys) | Which keys to cache, as `RedisKey` values: strings, raw bytes, or a key made with `WithKeyPrefix` (taken whole). Copied when set; changing your list afterwards changes nothing. Setting any selects `TrackingByPrefix`, where the list decides for calls that pass no flag; under `Broadcast` it is also sent to the server to filter what is announced. | | `MaxBytes` | unbounded | A memory budget for cached replies. Entries that have not been read since they were stored are evicted, oldest first, to get back under it - see [Eviction](#eviction). | | `MaxEntries` | unbounded | A limit on the number of entries - use it alongside `MaxBytes` when replies are small and numerous. | | `MaxPayloadBytes` | 1 MiB | The largest single reply that may be cached. | @@ -193,7 +193,7 @@ Under `MaxBytes`, replies are stored in large slabs in the order they arrive and ### Tracking modes and prefixes - **`Tracking`** (the default without `Prefixes`): the server remembers which keys this client read and announces only those. Precise - invalidation traffic scales with what this client holds, not with how busy the server is - but the server holds a table per client, and evicts from it (announcing what it drops) under pressure, which costs a refill here. `Prefixes` cannot be combined with it. Negotiated as `CLIENT TRACKING ON OPTOUT`: every read is tracked and cacheable unless a call says otherwise with `CommandFlags.CacheOptOut`, which also tells the server not to track that read (a `CLIENT CACHING NO` goes out as one unit with it). -- **`TrackingByPrefix`** (the default when `Prefixes` are set): `Prefixes` decide for the calls that pass no flag. A read whose keys all fall under a prefix is treated as if it had said `CommandFlags.CacheOptIn`: it goes out behind `CLIENT CACHING YES`, the server tracks it, and it is stored on its first miss. Any other read goes to the server as it would without a cache. The flags still win: `CacheOptIn` on a read outside the prefixes caches it, `CacheOptOut` on a read inside them does not. Negotiated as `CLIENT TRACKING ON OPTIN`; the list is **not** sent to the server (the server only accepts `PREFIX` with `BCAST`), so the server tracks exactly the keys this client holds. The price is an extra frame and reply per miss inside the prefixes, which is the traffic the cache exists to shrink. Prefixes must not overlap one another (`"app:"` and `"app:user:"` together are rejected), and they are compared with the key as sent - so if you use `WithKeyPrefix`, include that prefix. +- **`TrackingByPrefix`** (the default when `Prefixes` are set): `Prefixes` decide for the calls that pass no flag. A read whose keys all fall under a prefix is treated as if it had said `CommandFlags.CacheOptIn`: it goes out behind `CLIENT CACHING YES`, the server tracks it, and it is stored on its first miss. Any other read goes to the server as it would without a cache. The flags still win: `CacheOptIn` on a read outside the prefixes caches it, `CacheOptOut` on a read inside them does not. Negotiated as `CLIENT TRACKING ON OPTIN`; the list is **not** sent to the server (the server only accepts `PREFIX` with `BCAST`), so the server tracks exactly the keys this client holds. The price is an extra frame and reply per miss inside the prefixes, which is the traffic the cache exists to shrink. Prefixes are `RedisKey` values, so a byte prefix is as good as a string one, and a key made with `WithKeyPrefix` can be given as it is. They must not overlap one another (`"app:"` and `"app:user:"` together are rejected), and they are compared with the key as sent - so if you build keys with `WithKeyPrefix`, the prefix must include that. - **`TrackingOptIn`**: opt-in by flag alone. Negotiated as `CLIENT TRACKING ON OPTIN`: the server tracks nothing, and nothing is cached, unless a call says so with `CommandFlags.CacheOptIn` - that read goes out behind `CLIENT CACHING YES`, is tracked, and is stored on its first miss. Every other read goes to the server as it would without a cache. For "cache these few hot keys and leave everything else alone" when the keys are not a prefix; the price is an extra frame and an extra reply per opted-in miss. `Prefixes` cannot be combined with it - that is `TrackingByPrefix`. - **`Broadcast`**: the server announces every change to every key matching your `Prefixes`, whoever made it, and keeps no record of what you read. Cheap for the server; the cost is on the client, which hears about keys it never cached, and that cost is set by how busy the server is rather than by anything this client does - which is why prefixes no longer select this mode by default. **Set `Prefixes`** if you choose it: with none, this connection hears about every write to every key in the instance. Keys outside the prefixes are simply not cached, and here the list is a hard boundary: the server was never asked about those keys, so no flag can opt one in. The per-call flags keep only their local meaning: the server has no per-client table to opt in to or out of. @@ -223,6 +223,26 @@ These behave as a write from this process does: the entries are dropped at once, The same object carries the cache's counters - `Count`, `Bytes`, `Stored`, `Evicted`, `Promoted`, `Expired`, `Refreshes`, `ServedStale` - cumulative since the connection was made. There is no hit counter, because a hit deliberately records nothing; see [Is it working?](#is-it-working) for the measurement that cannot be fooled. +### Connection groups + +With a [connection group](Failover) (geo-failover, Active-Active), the cache is configured once, on `MultiGroupOptions.ClientCache`, and every member inherits it; a member whose own `ConfigurationOptions.ClientCache` is set is refused when the group connects. Each member still needs `Protocol = RedisProtocol.Resp3`. + +```csharp +MultiGroupOptions options = new MultiGroupOptions.Builder +{ + ClientCache = new CacheOptions { Prefixes = ["app:"], MaxBytes = 64 * 1024 * 1024 }, +}; +await using var conn = await ConnectionMultiplexer.ConnectGroupAsync(members, options); +``` + +What that gives you: + +- **One cache per member, not one for the group.** A cached entry is only as good as the invalidation stream behind it, and that stream belongs to one member's connections. So each member keeps its own cache, and a read is served from, and stored into, the cache of whichever member is active at that moment. Nothing is copied or flushed when the active member changes: the member you left keeps its cache, still invalidated by its own server, ready for when you come back. +- **Your own writes reach every member.** A write made while member B is active reaches A's server only by replication, so without help A's cache could, after a failback, hand you the value you had just replaced. To prevent that, a write from this process invalidates the key in every member's cache. A then misses and asks its server, which may still be catching up; that is ordinary replication lag, and an uncached read would see the same. Server-sent invalidations, and the flush that follows a lost connection, stay with their own member. +- **The members must agree.** That is why the setting lives on the group: a member that cached differently, or not at all, would change what callers get on every failover. + +`GetClientCache()` on a group returns the active member's cache at the time of the call. `Invalidate` and `InvalidateAll` through it reach every member, as a write does; its counters describe that member only. + ### Per call, and per context - **`CommandFlags.CacheOptOut`** - this call neither reads from nor writes to the cache, and in `Tracking` mode the server is told not to track it: diff --git a/docs/Configuration.md b/docs/Configuration.md index 63f3e2f6f..bf09744e1 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -161,7 +161,7 @@ Additional code-only options: - HealthCheck (`HealthCheck`) - Default: `null` - **Client-side geographic failover.** An *active* health check used when the connection is a member of a connection group; when `null`, the group-level `MultiGroupOptions.HealthCheck` is used. See [Client-side geographic failover](Failover) - ClientCache (`CacheOptions`) - Default: `null` - - Enables client-side caching of reads, kept current by server-assisted invalidation; requires RESP3. `null` means no cache. See [Client-side caching](ClientSideCaching) + - Enables client-side caching of reads, kept current by server-assisted invalidation; requires RESP3. `null` means no cache. Not for a connection group member: there the cache is set once, on `MultiGroupOptions.ClientCache`, and a member that sets its own is refused. See [Client-side caching](ClientSideCaching) Tokens in the configuration string are comma-separated; any without an `=` sign are assumed to be redis server endpoints. Endpoints without an explicit port will use 6379 if ssl is not enabled, and 6380 if ssl is enabled. Tokens starting with `$` are taken to represent command maps, for example: `$config=cfg`. diff --git a/docs/Failover.md b/docs/Failover.md index c91fb29af..cf4f4cf97 100644 --- a/docs/Failover.md +++ b/docs/Failover.md @@ -86,10 +86,22 @@ Because the policies are immutable, there is no question of whether a change "ta | `FailbackHealthCheck` | `MultiGroupOptions.FailbackHealthCheck` | `ConnectionGroupMember.FailbackHealthCheck` | | `Weight` | *(none)* | `ConnectionGroupMember.Weight` | | `SkipInitialHealthCheck` | *(none)* | `ConnectionGroupMember.SkipInitialHealthCheck` | +| `ClientCache` | `MultiGroupOptions.ClientCache` | *(none; a member's `ConfigurationOptions.ClientCache` must be left unset, or the group refuses it)* | Resolution is always "member override, else group default". Resolving a group default never writes back into your `ConfigurationOptions` - you can safely reuse one `ConfigurationOptions` for a group member and for an unrelated `Connect` without the group's policies leaking across. `ConfigurationOptions` carries only the two availability settings that are meaningful for a **single** connection - `CircuitBreaker` and `RetryPolicy`. `HealthCheck` is a group-only concept and lives on `ConnectionGroupMember`. + +[Client-side caching](ClientSideCaching) is the one setting that goes the other way: on a group it is set **only** at the group level, and every member inherits it. Each member keeps a cache of its own (the invalidations that keep it honest arrive on that member's connections), and the group serves reads from whichever member is active, so members must cache alike or a failover would change what callers get. A member whose own `ConfigurationOptions.ClientCache` is set is refused. Every member needs `Protocol = RedisProtocol.Resp3`, as any cached connection does. + +```csharp +MultiGroupOptions options = new MultiGroupOptions.Builder +{ + ClientCache = new CacheOptions { Prefixes = ["app:"], MaxBytes = 64 * 1024 * 1024 }, +}; +``` + +See [Connection groups](ClientSideCaching#connection-groups) in the caching documentation for what happens to the caches across a failover. The library automatically selects the best available endpoint based on: 1. **Availability** - Connected endpoints are always preferred over disconnected ones diff --git a/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs b/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs index 8a4156822..8729b00cd 100644 --- a/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs +++ b/src/StackExchange.Redis/Availability/MultiGroupMultiplexer.cs @@ -550,7 +550,7 @@ internal static async Task ConnectAsync( // note the resolved circuit-breaker is passed *alongside* the configuration rather than // written into it; see ConnectionMultiplexer.GroupCircuitBreaker - pending[i] = ConnectionMultiplexer.ConnectGroupMemberAsync(config, log, members[i].ResolveCircuitBreaker(options)); + pending[i] = ConnectionMultiplexer.ConnectGroupMemberAsync(config, log, members[i].ResolveCircuitBreaker(options), options.ClientCache); } for (int i = 0; i < pending.Length; i++) @@ -593,6 +593,8 @@ internal static async Task ConnectAsync( /// private static void ValidateConsistentConfiguration(ConnectionGroupMember[] members) { + for (var i = 0; i < members.Length; i++) DemandNoMemberCache(members[i], i); + if (members.Length < 2) return; var first = members[0].Configuration; @@ -621,6 +623,27 @@ private static void ValidateConsistentConfiguration(ConnectionGroupMember[] memb static string Describe(int? database) => database is { } value ? value.ToString() : "unset"; } + /// + /// A member's cache comes from the group, never from the member: refuse a member that brings its own. + /// + /// + /// Each member holds a cache of its own, and the group answers reads from the active member's, so a + /// member that cached differently from the others - or not at all - would change what callers get on + /// every failover. Inheriting one is how members agree; + /// allowing a per-member override would be allowing them to disagree. + /// + private static void DemandNoMemberCache(ConnectionGroupMember member, int index) + { + if (member.Configuration.ClientCache is not null) + { + throw new ArgumentException( + $"ConfigurationOptions.ClientCache must not be set on a connection group member (member {index}, " + + $"'{member.Name}'); set MultiGroupOptions.ClientCache instead, which every member inherits. " + + "The group serves reads from whichever member is active, so members must cache alike.", + nameof(member)); + } + } + private readonly MultiGroupOptions _options; public MultiGroupOptions Options => _options; @@ -1550,11 +1573,12 @@ private void OnConnectionChanged( public async Task AddAsync(ConnectionGroupMember member, TextWriter? log = null) { // connect + DemandNoMemberCache(member, _members.Length); member.Init(_members.Length); member.Configuration.AbortOnConnectFail = false; // members are gated by health-checks, not connect-fail member.Configuration.HeartbeatConsistencyChecks = true; var muxer = await ConnectionMultiplexer.ConnectGroupMemberAsync( - member.Configuration, log, member.ResolveCircuitBreaker(_options)).ConfigureAwait(false); + member.Configuration, log, member.ResolveCircuitBreaker(_options), _options.ClientCache).ConfigureAwait(false); member.SetMultiplexer(muxer); // unless told otherwise, run an initial health-check so a healthy member can be selected immediately; diff --git a/src/StackExchange.Redis/Availability/MultiGroupOptions.cs b/src/StackExchange.Redis/Availability/MultiGroupOptions.cs index 2cf4d1ccd..2faa32b52 100644 --- a/src/StackExchange.Redis/Availability/MultiGroupOptions.cs +++ b/src/StackExchange.Redis/Availability/MultiGroupOptions.cs @@ -1,6 +1,7 @@ using System; using System.Diagnostics.CodeAnalysis; using RESPite; +using StackExchange.Redis.Caching; namespace StackExchange.Redis.Availability; @@ -18,7 +19,7 @@ public sealed class MultiGroupOptions internal static readonly TimeSpan DefaultFailbackDelay = TimeSpan.Zero; private static readonly MultiGroupOptions DefaultInstance = new( - HealthCheck.Default, CircuitBreaker.Default, RetryPolicy.Default, DefaultHealthCheckInterval, DefaultFailbackDelay, null); + HealthCheck.Default, CircuitBreaker.Default, RetryPolicy.Default, DefaultHealthCheckInterval, DefaultFailbackDelay, null, null); /// /// Default shared options. @@ -31,9 +32,11 @@ private MultiGroupOptions( RetryPolicy retryPolicy, TimeSpan healthCheckInterval, TimeSpan failbackDelay, - HealthCheck? failbackHealthCheck) + HealthCheck? failbackHealthCheck, + CacheOptions? clientCache) { HealthCheck = healthCheck; + ClientCache = clientCache; CircuitBreaker = circuitBreaker; RetryPolicy = retryPolicy; HealthCheckInterval = healthCheckInterval; @@ -82,6 +85,25 @@ private MultiGroupOptions( /// public HealthCheck? FailbackHealthCheck { get; } + /// + /// The client-side cache every member runs, or for no cache. + /// + /// + /// + /// Set here and nowhere else: a member whose own is set is + /// refused when the group connects. Each member keeps a cache of its own - the invalidations that keep it + /// honest arrive on that member's connections - and the group serves reads from the active member's, so + /// members must agree on what may be cached and how, or a failover would silently change what callers get. + /// One definition, inherited by every member, is how that agreement is kept. See the caching documentation + /// for how a group's caches behave across a failover. + /// + /// + /// Requires to be on every + /// member, as a cache does on any connection. + /// + /// + public CacheOptions? ClientCache { get; } + /// /// Allows configuration of . /// @@ -105,6 +127,7 @@ public Builder(MultiGroupOptions options) HealthCheckInterval = options.HealthCheckInterval; FailbackDelay = options.FailbackDelay; FailbackHealthCheck = options.FailbackHealthCheck; + ClientCache = options.ClientCache; } /// @@ -141,6 +164,9 @@ public Builder(MultiGroupOptions options) /// public HealthCheck? FailbackHealthCheck { get; set; } + /// + public CacheOptions? ClientCache { get; set; } + /// /// Create a new options instance. /// @@ -158,12 +184,13 @@ public MultiGroupOptions Create() && ReferenceEquals(RetryPolicy, RetryPolicy.Default) && HealthCheckInterval == DefaultHealthCheckInterval && FailbackDelay == DefaultFailbackDelay - && FailbackHealthCheck is null) + && FailbackHealthCheck is null + && ClientCache is null) { return DefaultInstance; } - return new MultiGroupOptions(HealthCheck, CircuitBreaker, RetryPolicy, HealthCheckInterval, FailbackDelay, FailbackHealthCheck); + return new MultiGroupOptions(HealthCheck, CircuitBreaker, RetryPolicy, HealthCheckInterval, FailbackDelay, FailbackHealthCheck, ClientCache); } /// diff --git a/src/StackExchange.Redis/Caching/CacheOptions.cs b/src/StackExchange.Redis/Caching/CacheOptions.cs index f585a6800..5d20bb5cd 100644 --- a/src/StackExchange.Redis/Caching/CacheOptions.cs +++ b/src/StackExchange.Redis/Caching/CacheOptions.cs @@ -282,23 +282,38 @@ public int EvictionSampleSize /// Either mistake is raised when the cache is built rather than at the handshake. /// /// - public IReadOnlyList Prefixes + public IReadOnlyList Prefixes { get { - return _prefixes; + // fresh keys over fresh copies, every time: a RedisKey over a byte[] hands that array back through + // its explicit conversion, and the bytes this cache matches on are nobody else's to change + var bytes = _prefixBytes; + if (bytes.Length == 0) return Array.Empty(); + var keys = new RedisKey[bytes.Length]; + for (var i = 0; i < keys.Length; i++) keys[i] = (byte[])bytes[i].Clone(); + return keys; } init { - _prefixes = value ?? throw new ArgumentNullException(nameof(value)); - _prefixBytes = Encode(_prefixes); + if (value is null) throw new ArgumentNullException(nameof(value)); + + // Rendered to bytes ONCE, here, and nothing of the caller's is kept: not the list, and not the + // array behind a byte[] key. Matching and the handshake's PREFIX arguments both read this + // snapshot, so a list or an array the caller changes afterwards changes neither. RedisKey rather + // than string because keys are bytes: a prefix that is not valid UTF-8 has no string spelling, + // and a key made with WithKeyPrefix already carries its context prefix, which is what the wire + // sees and what is matched - so it can be given as it is, rather than concatenated by hand. + _prefixBytes = Encode(value); } } - private readonly IReadOnlyList _prefixes = Array.Empty(); private readonly byte[][] _prefixBytes = []; + /// The prefixes as the bytes that are matched and sent; the snapshot itself, not to be changed. + internal byte[][] PrefixBytes => _prefixBytes; + /// Whether restricts what may be cached. /// /// Only ever true in and @@ -364,22 +379,26 @@ internal bool IsTracked(scoped ReadOnlySpan key) /// refuses a prefix list where one entry is a prefix of another. Catching it here means the failure /// arrives where the mistake was made, rather than as a handshake error much later. /// - private static byte[][] Encode(IReadOnlyList prefixes) + private static byte[][] Encode(IReadOnlyList prefixes) { if (prefixes.Count == 0) return []; var result = new byte[prefixes.Count][]; for (var i = 0; i < prefixes.Count; i++) { + // the key's whole rendering - a context prefix included - exactly as the writer would send it var prefix = prefixes[i]; - if (string.IsNullOrEmpty(prefix)) + var length = prefix.TotalLength(); + if (length == 0) { throw new ArgumentException( "An empty cache prefix matches every key; use an empty prefix list to track everything.", nameof(Prefixes)); } - result[i] = Encoding.UTF8.GetBytes(prefix); + var bytes = new byte[length]; + prefix.CopyTo(bytes); + result[i] = bytes; } for (var i = 0; i < result.Length; i++) @@ -389,13 +408,16 @@ private static byte[][] Encode(IReadOnlyList prefixes) if (i != j && result[i].AsSpan().StartsWith(result[j])) { throw new ArgumentException( - $"Cache prefixes must not overlap, but '{prefixes[i]}' starts with '{prefixes[j]}'.", + $"Cache prefixes must not overlap, but '{Describe(result[i])}' starts with '{Describe(result[j])}'.", nameof(Prefixes)); } } } return result; + + // for the message only; a prefix that is not text shows as whatever the decoder makes of it + static string Describe(byte[] prefix) => Encoding.UTF8.GetString(prefix); } } } diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 16c194ede..73fea6af5 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -195,7 +195,7 @@ static ConnectionMultiplexer() SetAutodetectFeatureFlags(); } - private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? serverType = null, EndPointCollection? endpoints = null, Availability.CircuitBreaker? groupCircuitBreaker = null) + private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? serverType = null, EndPointCollection? endpoints = null, Availability.CircuitBreaker? groupCircuitBreaker = null, Caching.CacheOptions? groupClientCache = null) { Interlocked.Increment(ref s_MuxerCreateCount); @@ -225,7 +225,12 @@ private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? se // honest arrive on a connection, and a connection belongs to the multiplexer. A cache per // database would have to be found from here anyway when a push lands, and a cache per context // would be handed the pushes of a connection it does not own. - if (RawConfig.ClientCache is { Enabled: true } cacheOptions) + // + // A group member's cache comes from the group (MultiGroupOptions.ClientCache), passed alongside the + // configuration rather than written into it, as the group's circuit breaker is; a member's own + // ConfigurationOptions.ClientCache is refused by the group. Never for a sentinel connection: it holds + // no data to cache, and CLIENT TRACKING is not something to negotiate with a sentinel. + if (!_isSentinel && (groupClientCache ?? RawConfig.ClientCache) is { Enabled: true } cacheOptions) { ClientCache = new RespClientCache(cacheOptions, commandMap: map); } @@ -238,9 +243,9 @@ private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? se lastHeartbeatTicks = Environment.TickCount; } - private static ConnectionMultiplexer CreateMultiplexer(ConfigurationOptions configuration, ILogger? log, ServerType? serverType, out EventHandler? connectHandler, EndPointCollection? endpoints = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false) + private static ConnectionMultiplexer CreateMultiplexer(ConfigurationOptions configuration, ILogger? log, ServerType? serverType, out EventHandler? connectHandler, EndPointCollection? endpoints = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false, Caching.CacheOptions? groupClientCache = null) { - var muxer = new ConnectionMultiplexer(configuration, serverType, endpoints, groupCircuitBreaker) { IsGroupMember = isGroupMember }; + var muxer = new ConnectionMultiplexer(configuration, serverType, endpoints, groupCircuitBreaker, groupClientCache) { IsGroupMember = isGroupMember }; connectHandler = null; if (log is not null) { @@ -633,10 +638,10 @@ public static Task ConnectAsync(ConfigurationOptions conf /// /// Connect a multiplexer that is a member of a connection group, applying the group's resolved - /// circuit-breaker without writing it back into the caller's - /// (which the caller may legitimately reuse for other connections). + /// circuit-breaker and client cache without writing them back into the caller's + /// (which the caller may legitimately reuse for other connections). /// - internal static Task ConnectGroupMemberAsync(ConfigurationOptions configuration, TextWriter? log, Availability.CircuitBreaker? groupCircuitBreaker) + internal static Task ConnectGroupMemberAsync(ConfigurationOptions configuration, TextWriter? log, Availability.CircuitBreaker? groupCircuitBreaker, Caching.CacheOptions? groupClientCache = null) { Dependencies.Assert(); Validate(configuration); @@ -645,11 +650,19 @@ internal static Task ConnectGroupMemberAsync(Configuratio { // the sentinel path builds the primary connection internally, so we cannot pass the breaker // down into construction; apply it afterwards - it is picked up by subsequent physical - // connections, and an explicit ConfigurationOptions.CircuitBreaker still applies throughout + // connections, and an explicit ConfigurationOptions.CircuitBreaker still applies throughout. + // The cache cannot be applied afterwards (it is built with the connection), so it rides on a + // clone of the configuration; the sentinel connection itself never builds one (see the ctor) + if (groupClientCache is not null) + { + configuration = configuration.Clone(); + configuration.ClientCache = groupClientCache; + } + return ApplyAfterConnectAsync(SentinelPrimaryConnectAsync(configuration, log), groupCircuitBreaker); } - return ConnectImplAsync(configuration, log, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: true); + return ConnectImplAsync(configuration, log, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: true, groupClientCache: groupClientCache); static async Task ApplyAfterConnectAsync(Task pending, Availability.CircuitBreaker? groupCircuitBreaker) { @@ -660,7 +673,7 @@ static async Task ApplyAfterConnectAsync(Task ConnectImplAsync(ConfigurationOptions configuration, TextWriter? writer = null, ServerType? serverType = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false) + private static async Task ConnectImplAsync(ConfigurationOptions configuration, TextWriter? writer = null, ServerType? serverType = null, Availability.CircuitBreaker? groupCircuitBreaker = null, bool isGroupMember = false, Caching.CacheOptions? groupClientCache = null) { IDisposable? killMe = null; EventHandler? connectHandler = null; @@ -672,7 +685,7 @@ private static async Task ConnectImplAsync(ConfigurationO var sw = ValueStopwatch.StartNew(); log?.LogInformationConnectingAsync(RuntimeInformation.FrameworkDescription, Utils.GetLibVersion()); - muxer = CreateMultiplexer(configuration, log, serverType, out connectHandler, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: isGroupMember); + muxer = CreateMultiplexer(configuration, log, serverType, out connectHandler, groupCircuitBreaker: groupCircuitBreaker, isGroupMember: isGroupMember, groupClientCache: groupClientCache); killMe = muxer; Interlocked.Increment(ref muxer._connectAttemptCount); bool configured = await muxer.ReconfigureAsync(first: true, reconfigureAll: false, log, null, "connect").ObserveErrors().ForAwait(); diff --git a/src/StackExchange.Redis/Contexts/RespConnectionExtensions.cs b/src/StackExchange.Redis/Contexts/RespConnectionExtensions.cs index 95531f98a..68cb5485b 100644 --- a/src/StackExchange.Redis/Contexts/RespConnectionExtensions.cs +++ b/src/StackExchange.Redis/Contexts/RespConnectionExtensions.cs @@ -63,7 +63,10 @@ public static RespDatabaseContext GetDatabaseContext(this IConnectionMultiplexer /// write from this process does, and its counters are the member currently answering reads. /// /// - /// The same instance every time, for the life of the connection; there is nothing to dispose. + /// On a plain connection, the same instance every time, for the life of the connection. On a group it is + /// the active member's cache at the time of the call: a view held across a failover still names + /// the member that was active when it was obtained - its invalidations still reach every member, but its + /// counters describe that member, not the one now answering reads. There is nothing to dispose. /// /// public static ClientCache? GetClientCache(this IConnectionMultiplexer multiplexer) diff --git a/src/StackExchange.Redis/Contexts/RespHandshake.cs b/src/StackExchange.Redis/Contexts/RespHandshake.cs index f0b3e35c9..b74e26cd0 100644 --- a/src/StackExchange.Redis/Contexts/RespHandshake.cs +++ b/src/StackExchange.Redis/Contexts/RespHandshake.cs @@ -1343,13 +1343,13 @@ private static async Task EnableClientTrackingAsync( } var options = cache.Options; - var prefixes = options.Prefixes; + var prefixes = options.PrefixBytes; // the snapshot the cache matches on: the same bytes go to the server var command = RedisCommand.CLIENT; // CLIENT TRACKING ON [BCAST] [PREFIX p]... if (options.ResolvedTrackingMode == Caching.CacheTrackingMode.Broadcast) { - if (prefixes.Count == 0) + if (prefixes.Length == 0) { await context.SendAsync($"{command}{RedisLiterals.TRACKING}{RedisLiterals.ON}{RedisLiterals.BCAST}") .ConfigureAwait(false); @@ -1358,7 +1358,7 @@ await context.SendAsync($"{command}{RedisLiterals.TRACKING}{RedisLiterals.ON}{Re { // built as values rather than composed in the interpolation, because the number of // PREFIX pairs is not known until here - var args = new RedisValue[3 + (prefixes.Count * 2)]; + var args = new RedisValue[3 + (prefixes.Length * 2)]; var index = 0; args[index++] = RedisLiterals.TRACKING; args[index++] = RedisLiterals.ON; @@ -1366,7 +1366,7 @@ await context.SendAsync($"{command}{RedisLiterals.TRACKING}{RedisLiterals.ON}{Re foreach (var prefix in prefixes) { args[index++] = RedisLiterals.PREFIX; - args[index++] = prefix.AsRedisValue(); + args[index++] = prefix; // bytes, verbatim: no re-encoding on the way out } await context.SendAsync($"{command}{args}").ConfigureAwait(false); diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 775b20796..ef0fcf4be 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -54,7 +54,7 @@ StackExchange.Redis.Caching.CacheOptions.MaxEntries.get -> int? StackExchange.Redis.Caching.CacheOptions.MaxEntries.init -> void StackExchange.Redis.Caching.CacheOptions.MaxPayloadBytes.get -> int? StackExchange.Redis.Caching.CacheOptions.MaxPayloadBytes.init -> void -StackExchange.Redis.Caching.CacheOptions.Prefixes.get -> System.Collections.Generic.IReadOnlyList! +StackExchange.Redis.Caching.CacheOptions.Prefixes.get -> System.Collections.Generic.IReadOnlyList! StackExchange.Redis.Caching.CacheOptions.Prefixes.init -> void StackExchange.Redis.Caching.CacheOptions.SweepInterval.get -> System.TimeSpan StackExchange.Redis.Caching.CacheOptions.SweepInterval.init -> void @@ -815,6 +815,9 @@ StackExchange.Redis.Availability.HealthCheckProbeScope.Member = 1 -> StackExchan StackExchange.Redis.Availability.MultiGroupOptions.Builder.FailbackHealthCheck.get -> StackExchange.Redis.Availability.HealthCheck? StackExchange.Redis.Availability.MultiGroupOptions.Builder.FailbackHealthCheck.set -> void StackExchange.Redis.Availability.MultiGroupOptions.FailbackHealthCheck.get -> StackExchange.Redis.Availability.HealthCheck? +StackExchange.Redis.Availability.MultiGroupOptions.ClientCache.get -> StackExchange.Redis.Caching.CacheOptions? +StackExchange.Redis.Availability.MultiGroupOptions.Builder.ClientCache.get -> StackExchange.Redis.Caching.CacheOptions? +StackExchange.Redis.Availability.MultiGroupOptions.Builder.ClientCache.set -> void virtual StackExchange.Redis.Availability.HealthCheckProbe.Scope.get -> StackExchange.Redis.Availability.HealthCheckProbeScope StackExchange.Redis.Availability.LagAwareOptions StackExchange.Redis.Availability.LagAwareOptions.CertificateValidation.get -> System.Net.Security.RemoteCertificateValidationCallback? diff --git a/tests/StackExchange.Redis.Tests/MultiGroupTests/GroupCacheTests.cs b/tests/StackExchange.Redis.Tests/MultiGroupTests/GroupCacheTests.cs index e3a6bfb99..a3ddafabc 100644 --- a/tests/StackExchange.Redis.Tests/MultiGroupTests/GroupCacheTests.cs +++ b/tests/StackExchange.Redis.Tests/MultiGroupTests/GroupCacheTests.cs @@ -35,10 +35,15 @@ private ConfigurationOptions Config(InProcessTestServer server) { var config = server.GetClientConfig(); config.Protocol = RedisProtocol.Resp3; - config.ClientCache = new CacheOptions { Admission = CacheAdmission.OnFirstMiss }; return config; } + // the cache is a group-level setting (MultiGroupOptions.ClientCache), inherited by every member + private static MultiGroupOptions.Builder Options() => new() + { + ClientCache = new CacheOptions { Admission = CacheAdmission.OnFirstMiss }, + }; + public enum WritePath { Database, @@ -54,7 +59,8 @@ public async Task AWriteWhileFailedOverIsNotServedFromTheOldMembersCache(WritePa using var serverB = new InProcessTestServer(log); ConnectionGroupMember a = new(Config(serverA), "a") { Weight = 9 }, b = new(Config(serverB), "b") { Weight = 2 }; - MultiGroupOptions options = new MultiGroupOptions.Builder { HealthCheck = new HealthCheck.Builder { Probe = HealthCheckProbe.Ping } }; + var options = Options(); + options.HealthCheck = new HealthCheck.Builder { Probe = HealthCheckProbe.Ping }; await using var conn = await ConnectionMultiplexer.ConnectGroupAsync([a, b], options); await GroupWait.AssertConnectedAsync(conn); await WaitForAsync(() => Task.FromResult(a.IsConnected && b.IsConnected), "both members connected"); @@ -105,7 +111,7 @@ public async Task ARemovedMembersCacheIsNoLongerStamped() using var serverB = new InProcessTestServer(log); ConnectionGroupMember a = new(Config(serverA), "a") { Weight = 9 }, b = new(Config(serverB), "b") { Weight = 2 }; - await using var conn = await ConnectionMultiplexer.ConnectGroupAsync([a, b]); + await using var conn = await ConnectionMultiplexer.ConnectGroupAsync([a, b], Options()); await GroupWait.AssertConnectedAsync(conn); var typed = Assert.IsType(conn); @@ -119,4 +125,59 @@ public async Task ARemovedMembersCacheIsNoLongerStamped() Assert.Empty(cacheA.Peers); Assert.Empty(cacheB.Peers); } + + [Fact] + public async Task TheGroupCacheReachesEveryMemberWithoutTouchingTheirOptions() + { + using var serverA = new InProcessTestServer(log); + using var serverB = new InProcessTestServer(log); + using var serverC = new InProcessTestServer(log); + ConnectionGroupMember a = new(Config(serverA), "a") { Weight = 9 }, b = new(Config(serverB), "b") { Weight = 2 }; + await using var conn = await ConnectionMultiplexer.ConnectGroupAsync([a, b], Options()); + await GroupWait.AssertConnectedAsync(conn); + var typed = Assert.IsType(conn); + + // every member runs the group's cache, and the callers' options were not written into + var cacheA = typed.GetMembers()[0].Multiplexer.ClientCache; + var cacheB = typed.GetMembers()[1].Multiplexer.ClientCache; + Assert.NotNull(cacheA); + Assert.NotNull(cacheB); + Assert.Same(typed.Options.ClientCache, cacheA.Options); + Assert.Same(typed.Options.ClientCache, cacheB.Options); + Assert.Null(a.Configuration.ClientCache); + Assert.Null(b.Configuration.ClientCache); + Assert.NotNull(conn.GetClientCache()); + + // ...including one that joins later, which also becomes a peer + ConnectionGroupMember c = new(Config(serverC), "c") { Weight = 1 }; + await typed.AddAsync(c); + var cacheC = c.Multiplexer.ClientCache; + Assert.NotNull(cacheC); + Assert.Contains(cacheC, cacheA.Peers); + Assert.Contains(cacheA, cacheC.Peers); + } + + [Fact] + public async Task AMemberThatBringsItsOwnCacheIsRefused() + { + // the cache is the group's, so that members cannot disagree about what may be cached - a failover + // would otherwise change what callers get; refused up front, before anything connects + using var serverA = new InProcessTestServer(log); + using var serverB = new InProcessTestServer(log); + var own = Config(serverB); + own.ClientCache = new CacheOptions(); + ConnectionGroupMember a = new(Config(serverA), "a") { Weight = 9 }, b = new(own, "b") { Weight = 2 }; + + var ex = await Assert.ThrowsAsync(() => ConnectionMultiplexer.ConnectGroupAsync([a, b], Options())); + Assert.Contains("MultiGroupOptions.ClientCache", ex.Message); + Assert.Contains("'b'", ex.Message); + + // and the same for a member added later + ConnectionGroupMember solo = new(Config(serverA), "a") { Weight = 9 }; + await using var conn = await ConnectionMultiplexer.ConnectGroupAsync([solo], Options()); + var typed = Assert.IsType(conn); + var late = new ConnectionGroupMember(own, "late"); + ex = await Assert.ThrowsAsync(() => typed.AddAsync(late)); + Assert.Contains("MultiGroupOptions.ClientCache", ex.Message); + } } diff --git a/tests/StackExchange.Redis.Tests/RespClientCacheTests.cs b/tests/StackExchange.Redis.Tests/RespClientCacheTests.cs index 9b2ab6b9a..6b59703a9 100644 --- a/tests/StackExchange.Redis.Tests/RespClientCacheTests.cs +++ b/tests/StackExchange.Redis.Tests/RespClientCacheTests.cs @@ -924,6 +924,57 @@ public void OneUntrackedKeySpoilsAMultiKeyCommand() Assert.True(Complete(cache, fill, "*3\r\n$1\r\n1\r\n$1\r\n2\r\n$1\r\n3\r\n")); } + /// The prefix list is copied when set: a caller changing their list afterwards changes nothing here. + [Fact] + public void PrefixesAreSnapshottedWhenSet() + { + var list = new List { "app:" }; + var raw = new byte[] { (byte)'r', (byte)'a', (byte)'w', (byte)':' }; + list.Add(raw); + var options = new CacheOptions { Prefixes = list }; + + // neither the list nor the array behind a byte key is ours any more + list[0] = "changed:"; + list.Add("other:"); + raw[0] = (byte)'X'; + + Assert.Equal(["app:", "raw:"u8.ToArray()], options.Prefixes); + Assert.True(options.IsTracked("app:1"u8)); + Assert.True(options.IsTracked("raw:1"u8)); + Assert.False(options.IsTracked("Xaw:1"u8)); + Assert.False(options.IsTracked("changed:1"u8)); + Assert.False(options.IsTracked("other:1"u8)); + + // ...and what the getter hands out is a copy too + ((byte[])options.Prefixes[0]!)[0] = (byte)'Z'; + Assert.True(options.IsTracked("app:1"u8)); + } + + /// A prefix is bytes: one that is not valid UTF-8 is matched and sent as given. + [Fact] + public void PrefixesMayBeArbitraryBytes() + { + byte[] prefix = [0xFF, 0x00, 0xC3, 0x28]; // not a valid UTF-8 sequence + var options = new CacheOptions { Prefixes = [prefix] }; + + Assert.Equal(prefix, options.PrefixBytes[0]); + Assert.NotSame(prefix, options.PrefixBytes[0]); + Assert.True(options.IsTracked([0xFF, 0x00, 0xC3, 0x28, (byte)'k'])); + Assert.False(options.IsTracked([0xFF, 0x00, 0xC3, (byte)'k'])); + } + + /// A key carrying a context prefix is taken whole - prefix plus value - as the wire sees it. + [Fact] + public void APrefixedKeyIsTakenWithItsPrefix() + { + var key = new RedisKey("tenant1:"u8.ToArray(), "user:"); + var options = new CacheOptions { Prefixes = [key] }; + + Assert.Equal("tenant1:user:"u8.ToArray(), options.PrefixBytes[0]); + Assert.True(options.IsTracked("tenant1:user:42"u8)); + Assert.False(options.IsTracked("user:42"u8)); + } + /// An empty prefix list means "track everything", so nothing is refused for being outside it. [Fact] public void NoPrefixesMeansEverythingIsCacheable()