diff --git a/CHANGELOG.md b/CHANGELOG.md index 735b10b3..4ae3cd1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,63 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) `AddDistributedCache` enables it on the private provider it builds, because a silently dropped refresh shortens a session and fire-and-forget cannot report one. Honored by both `RedisCache` and `RedisHashCache`. +- **Conditional add: `ICache.TryAddAsync` / `ICache.TryAddAsync` (+ blocking `ICache.TryAdd`).** + Writes a key only if it does not already exist, returning `true` only to the caller that created it. + On Redis-backed caches this is StackExchange.Redis `When.NotExists` — `SET key value EX … NX` — a + single atomic round-trip with the TTL applied by the same command, so exactly one caller across all + nodes wins a given key and a won key is never briefly immortal. Intended for at-most-once semantics + keyed by something: idempotency keys, dedup markers whose TTL is the dedup window, electing which + replica runs a job. `false` deliberately conflates "the key already existed" with "the write could + not be completed" (disconnected, threw, or a `null`/`default` value the cache cannot represent) — + fail-closed, so a caller treating `true` as "I own this key" is never wrongly told it won; the same + conflation `IDistributedLock.TryAcquireAsync` already documents. Unlike `SetAsync`, it never + deletes: where `SetAsync` removes the key when handed a `null` with `CacheNullValues` off, + `TryAddAsync` reports `false` and leaves the key untouched. Per tier: `RedisCache` issues the `NX` + write; `MultilayerCache` runs one sequence for every provider — take the local lock, probe the + local tier, ask the L2, populate L1 (plus an invalidation broadcast) on a win, best-effort. A + broadcast or L1 failure never downgrades a real win to `false`, since that would strand the entry + with no owner; with the L2 disconnected the call returns `false` rather than granting a local-only + claim every node would also be granted (`SetAsync` degrades to a local write there). The probe is + what bounds exclusion at one winner per process, and the only thing doing so on the `InMemory` + provider, whose `NullCache` L2 tells every caller it added the key; it also reports a local hit as + a loss without asking the L2, which is fail-closed but costs a win the L2 would have granted when + the local copy outlived the shared one. The local lock is what makes that probe-then-write atomic, + so a conditional add takes it regardless of `Lock.LocalLockEnabled` (which trades single-flight for + throughput on `GetOrAddAsync` and must not be able to hand one key to two callers), and a caller + that cannot acquire it within `Lock.LocalLockTimeout` is told it lost rather than proceeding + unserialized. The L2 itself is only ever asked, never classified by type or provider name, so a + provider a consumer registers participates on the same terms — with the corollary that an L2 + arbitrating in-process only is trusted as though it arbitrated for every sharer. On the `InMemory` + provider no invalidation broadcast is published: `ChangeTokenFactory` accepts only `CacheRemoved` + and `CacheRefreshed` there, so peers ignore `CacheSet` — deliberately, since each node's memory is + the store rather than a copy of a shared one. + An expiration that is not in the future reports `false` on every tier: the entry would be evicted on + arrival, so a `true` would be handed to every later caller as well. `NullCache` returns `false` — + the one member where it does not degrade to "caching is off, carry on", because it cannot complete + the write (which is exactly what a fail-closed `false` means) and because `true` there would hand + every caller a claim of exclusive ownership; it is reached by accident, being what + `ICacheFactory.CreateCache` resolves to when the requested provider is absent or has + `Enabled=false`. Added to both interfaces as **default interface methods**, so + **required** members rather than default interface methods: a probe followed by a write is not + atomic, and no fallback body could stand in for one without either voiding the guarantee or hiding + which stores can arbitrate, so an implementation states what its own store can do. **This is + source-breaking for hand-written `ICache` / `ICache` implementations,** which must add the three + overloads. `NullCache` returns `true`, as its `SetAsync` does — the null store retains nothing, so + no key pre-exists and no caller loses; note that it provides no exclusion at all and is what + `ICacheFactory.CreateCache` resolves to for an absent or disabled provider, so assert the provider + you expect at startup when the `true` branch runs a side effect that must not repeat. `ICache.Compat.cs` carries the token-positional forwarders + (`TryAddAsync(key, value, token)` and the `TimeSpan?`/`DateTimeOffset?` pairs), matching `SetAsync`. + No multi-key overload: Redis has no atomic multi-key `NX`, and all-or-nothing versus per-key + semantics would be a guess. This is a cache primitive, not a lock — no ownership token, no early + release, and a later `SetAsync`/`RemoveAsync` ignores the claim; for a fencing token and explicit + release use `IDistributedLock`. No conditional-add member was added to the hash surface, where `NX` + is per-field (`HSETNX`) and a different shape. +- A cancellation raised while the `SET … NX` write is in flight now propagates instead of being + reported as `false`, which would have claimed the key belongs to someone else — a fact the cancelled + call never established. The write itself stays on the shared `Write` resilience pipeline: retries + fire on exceptions only, and re-issuing `NX` is harmless, since an attempt whose reply was lost is + refused by the key it just wrote and reports the same `false` the exception would have, while an + attempt that never reached Redis is recovered as the `true` it should have been. ### Changed @@ -77,6 +134,27 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) onto the builder object instead. Every `CacheOptions` value supplied through that entry point was silently ignored; `Enabled`, `AppShortName`, `KeyCasing` and the rest now take effect, which changes behavior for anyone who was unknowingly running on the defaults. +### Documentation + +- New recipe [`conditional-add.md`](docs/recipes/conditional-add.md) — claiming a key exactly once with + `TryAddAsync`, why check-then-set is not equivalent, and when to reach for `IDistributedLock` + instead. `interfaces.md` documents the member on both cache surfaces with a per-provider table of + who arbitrates and how far exclusion reaches; `concepts.md` places it next to the two lock + abstractions, whose goal is reducing redundant work rather than a decision the caller can branch on. + The recipe's worked example is a daily digest rather than a payment capture, and says why: a claim + marker records that someone *started*, never that anyone finished, so an operation that must + eventually happen needs a recorded outcome instead. Both documents also state plainly that the + ambiguity in `false` is not recoverable — a serialization or command failure returns `false` with + the connection snapshot still healthy — rather than pointing at `IConnectionState` as an escape + hatch, and both explain why the `NX` write is safe to retry where `SPOP` is not. +- New `IConnectionState` section in `interfaces.md`, a public type that was documented nowhere (and + that the `TryAddAsync` guidance linked to through an anchor resolving to `IDistributedLock`). It is + described as what it is — a cache-health snapshot for probes, metrics and backoff — not as a way to + explain a negative result. +- Corrected `interfaces.md`, which described the `IHashCache.SetAsync(…, HashCacheEntryOptions, …)` + overload as offering "conditional set, individual field TTL". `HashCacheEntryOptions` carries + neither: `HashCacheSetOption` selects write *scope* (`HashReplace` merges fields, `KeyReplace` drops + the key first), and there is no per-field TTL. ## [1.3.0] - 2026-08-19 diff --git a/docs/concepts.md b/docs/concepts.md index eda0d8b7..13fb468a 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -245,6 +245,20 @@ register `RedisDistributedLock` (the in-memory provider passes custom wiring that skips `AddInMemoryRedis()` should add `.AddRedisDistributedLock()` explicitly on the builder. +`ICache.TryAddAsync` sits alongside these but answers a different question. The +locks exist to reduce redundant work — the degradation above is deliberate, and +neither lock guarantees exactly-once generator execution. `TryAddAsync` instead +returns a decision the caller can branch on: it writes the key only if it is +absent (Redis `SET … NX`) and returns `true` only to the caller that created it, +so it is the right primitive for at-most-once semantics keyed by something (a +dedup marker, an idempotency key, electing which node runs a job). It is not a +lock: the claim expires on its own TTL, carries no ownership token, and any later +`SetAsync`/`RemoveAsync` on the key ignores it. When you need a fencing token and +an explicit release, use `IDistributedLock`; when you need an atomic, expiring +"first one here wins" marker, use `TryAddAsync`. See +[the interfaces reference](reference/interfaces.md#icache) for the per-provider +behavior, including why it fails closed when the L2 is disconnected. + When both locks are enabled, `GetOrAddAsync` takes the local lock first (cheap, in-process) and then — under a double-checked read to avoid redundant generator runs after waiting — attempts the distributed lock. A failure to acquire the diff --git a/docs/index.md b/docs/index.md index eac1d4e6..6219ba16 100644 --- a/docs/index.md +++ b/docs/index.md @@ -32,6 +32,7 @@ UiPath's internal multilayer caching library. L1 in-memory + L2 Redis, cross-nod - [provider-fallback.md](recipes/provider-fallback.md) — InMemoryRedis when Redis is on, InMemory otherwise. - [factory-extension-methods.md](recipes/factory-extension-methods.md) — typed `ICache` via `ICacheFactory` extension methods. - [batch-get-or-add.md](recipes/batch-get-or-add.md) — one source round trip for N cache misses via multi-key `GetOrAddAsync`. +- [conditional-add.md](recipes/conditional-add.md) — claim a key exactly once with `TryAddAsync` (Redis `SET … NX`). - [app-version-prefix.md](recipes/app-version-prefix.md) — auto-invalidate on deploy via assembly-version key prefix. - [mediatr-pipeline-behavior.md](recipes/mediatr-pipeline-behavior.md) — generic per-request cache as a MediatR behavior. - [hash-cache-with-metadata.md](recipes/hash-cache-with-metadata.md) — payload + freshness metadata side-by-side. diff --git a/docs/recipes/conditional-add.md b/docs/recipes/conditional-add.md new file mode 100644 index 00000000..89484c25 --- /dev/null +++ b/docs/recipes/conditional-add.md @@ -0,0 +1,157 @@ +# Claim a key exactly once with `TryAddAsync` + +**What:** Use `TryAddAsync` when you need "only one caller may proceed", keyed by something. It writes +the key only if it is absent and returns `true` only to the caller that created it. On a Redis-backed +cache that is StackExchange.Redis `When.NotExists` — `SET key value EX … NX` — one atomic round-trip, +so exactly one caller across every node wins. + +**When to use:** +- **At-most-once side effects,** where skipping is the safe failure. "Send this alert / this welcome + email once per user per day." +- **Electing a worker.** Which of N replicas runs a periodic job this tick. +- **Dedup markers** where the marker's TTL *is* the dedup window. +- **Suppressing duplicate work,** as an optimization ahead of an operation that is idempotent anyway. + +**When not to use:** as the sole guard on an operation that must eventually happen. The claim marker +records that someone started, not that anyone finished, so a winner that dies mid-flight leaves the +work undone until the TTL lapses — and `false` cannot tell you which of "already claimed" or "write +failed" you are looking at. Those cases need a recorded outcome, not a claim. + +**When not to use:** if you need a lock you can release early, or a fencing token that proves you +still hold it, use [`IDistributedLock`](../reference/interfaces.md#idistributedlock) instead. See +[Notes](#notes). + +## Code + +```csharp +using UiPath.Caching; + +public class DailyDigest(ICache cache, IMailer mailer) +{ + // The key is the dedup unit: one send per user per UTC calendar day. The TTL only + // cleans the marker up afterwards — it is not a rolling 24-hour window, so two + // triggers minutes apart either side of midnight use different keys and both send. + private static readonly TimeSpan MarkerTtl = TimeSpan.FromHours(24); + + public async Task SendOnceAsync(string userId, CancellationToken token) + { + var claimed = await cache.TryAddAsync( + (CacheKey)$"digest:{userId}:{DateTime.UtcNow:yyyyMMdd}", + DateTimeOffset.UtcNow, + MarkerTtl, + token: token); + + if (!claimed) + { + // Someone else claimed it, or the write could not be completed. Both mean + // "do not send" — which is the whole reason this side effect fits the + // primitive: skipping is the safe failure, so the ambiguity costs nothing. + return; + } + + await mailer.SendDigestAsync(userId, token); + } +} +``` + +`TryAddAsync` fits here because **not** sending is the safe failure. Note what it is *not* doing: +nothing recovers the missed digest if the winner crashes between the claim and the send, and nothing +tells a lost race apart from a failed write. If your side effect must eventually happen — capturing a +payment, say — a claim marker is the wrong shape and no amount of branching on `false` fixes it: the +marker says "someone is handling this", never "this was handled". Record the *outcome* instead +(`Pending` → `Captured` on a durable store, or an idempotency key the downstream provider itself +honors) and let redelivery retry until the outcome is written. + +The typed surface is the same shape without the `policy` parameter: + +```csharp +public class JobElection(ICache cache) +{ + public Task TryClaimTickAsync(string jobName, DateTimeOffset tick, CancellationToken token) => + cache.TryAddAsync( + (CacheKey)$"{jobName}:{tick:yyyyMMddHHmm}", + Environment.MachineName, + TimeSpan.FromMinutes(5), + token) + .AsTask(); +} +``` + +## Why not check-then-set + +The obvious hand-rolled version is not equivalent: + +```csharp +// BROKEN: two callers can both observe "absent" before either writes. +if (!await cache.ContainsAsync(key, token)) +{ + await cache.SetAsync(key, DateTimeOffset.UtcNow, MarkerTtl, token: token); + await payments.CaptureAsync(eventId, token); // runs twice under concurrency +} +``` + +The gap between the probe and the write is the whole problem, and no amount of narrowing closes it — +that is exactly the gap `NX` removes by making the decision part of the write. This is also why +`TryAddAsync` is a **required** member of `ICache` and `ICache` rather than a default interface +method, precisely so nothing can inherit the code above: a non-atomic emulation would void the only +guarantee the method makes, and a fallback that quietly reported one answer for every store would +hide which stores can actually arbitrate. An implementation says what its own store can do. + +## Notes + +- **`false` does not mean "the key existed".** It means "you did not create it" — the key already + existed, *or* the write could not be completed (store disconnected, write threw, or the value was a + `null`/`default` the cache cannot represent). This is fail-closed on purpose: nobody is ever wrongly + told they won. Design the `false` branch so that "skip the side effect" is the safe outcome — + nothing recovers the distinction. `IConnectionState.IsConnected` does not: a serialization or + command failure returns `false` with the connection snapshot still healthy, and the snapshot can + change either side of the call anyway. `IDistributedLock.TryAcquireAsync` does not either; it + documents backend-unavailable and already-held as the same `null`. If the two readings need + different handling, you need a primitive with a richer result than a `bool`. +- **A non-positive TTL claims nothing.** An expiration that is not in the future returns `false` on + every tier rather than a win, because the entry would be evicted on arrival and the next caller + would be told it won too. +- **Caching switched off means everyone wins.** `NullCache.TryAddAsync` returns `true` — it retains + nothing, so no key pre-exists and nobody loses — and it is what `ICacheFactory.CreateCache` falls + back to when the requested provider is missing or has `Enabled=false`. A mistyped provider name + therefore turns at-most-once into at-least-once with no error, so assert the provider you expect at + startup: `if (cacheFactory.CreateCache(KnownCacheProviderNames.Redis) is NullCache) throw …`. +- **The `NX` write is retryable,** so it stays on the shared `Write` resilience pipeline. Retries + fire on exceptions only, and the ambiguous case costs nothing: if the write lands but its reply is + lost, the retry is refused by the key it just wrote and reports `false` — the same answer the + un-retried exception would have produced — while a first attempt that never reached Redis is + recovered as the `true` it should have been. (`SPOP` in `UiPath.Caching.Queue` is the opposite: a + retry there pops a second item and loses the first, which is why it has its own opt-in pipeline.) +- **It never deletes.** `SetAsync` handed a `null` with `CacheNullValues` off *removes* the key; + `TryAddAsync` returns `false` and leaves it alone. With `CacheNullValues` on, a `null` claims the key + through the cached-null sentinel — so prefer a meaningful value (a timestamp, the machine name) that + makes the claim diagnosable in `redis-cli`. +- **The TTL is applied by the same command,** so a won key is never briefly immortal between the write + and a follow-up `EXPIRE`. Give every claim a TTL you are willing to wait out: there is no release, + so a crashed winner blocks the key until it expires. That is the main reason to prefer + `IDistributedLock` for long critical sections. +- **It is not a lock.** No ownership token, no early release, and a later `SetAsync`/`RemoveAsync` on + the key silently overwrites the claim. If a bug elsewhere writes that key, exclusion is gone with no + error. On the `InMemory` provider the reverse also holds: the local lock serializes conditional adds + against each other, but `SetAsync` takes no lock, so a set landing between the probe and the write + is overwritten by the claim, which still reports `true`. Redis has no such gap — `NX` is atomic + against a concurrent `SET`. +- **In-memory-only caches exclude in-process only.** With the `InMemory` provider there is no shared + store to arbitrate, so the local tier does — serialized by the local lock, which a conditional add + takes whatever `Lock.LocalLockEnabled` says, because here it *is* the guarantee rather than a + single-flight optimization. A caller that cannot acquire it within `Lock.LocalLockTimeout` is told + it lost. Two processes still both win. `InMemoryRedis` and `Redis` are cross-node correct. +- **Both tiers contribute, in one sequence:** the local tier is probed under the local lock, then + the L2 decides, then L1 is populated on a win. The probe bounds exclusion at one winner per + process — the only thing doing so when the L2 retains nothing — and reports a local hit as a loss + without asking the L2, which is fail-closed but costs a win the L2 would have granted if the local + copy outlived the shared one. When the L2 is disconnected the call returns `false` rather than + granting a local claim every node would also be granted, unlike `SetAsync`, which degrades to a + local-only write there. +- **There is no multi-key overload.** Redis has no atomic multi-key `NX`, and choosing all-or-nothing + versus per-key semantics on your behalf would be a guess. Claim keys one at a time, or take a single + claim on a key that stands for the whole batch. + +**See also:** [`ICache`](../reference/interfaces.md#icache), +[`IDistributedLock`](../reference/interfaces.md#idistributedlock), +[Concepts — locking](../concepts.md) diff --git a/docs/reference/interfaces.md b/docs/reference/interfaces.md index c26379fa..4aea64d4 100644 --- a/docs/reference/interfaces.md +++ b/docs/reference/interfaces.md @@ -53,6 +53,12 @@ public partial interface ICache ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CancellationToken token = default); + ValueTask TryAddAsync(CacheKey cacheKey, T? value, CancellationToken token = default); + + ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration, CancellationToken token = default); + + ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default); + ValueTask RefreshAsync(CacheKey cacheKey, CancellationToken token = default); ValueTask RefreshAsync(CacheKey cacheKey, TimeSpan? expiration, CancellationToken token = default); @@ -71,6 +77,8 @@ public partial interface ICache The multi-key `GetOrAddAsync` overloads pair each key with an opaque caller state (`TState`) — a database id, a request object, whatever identifies the entry in the caller's own vocabulary. The generator is invoked at most once, with only the states of the entries that missed, and results come back keyed by state. These three overloads are **abstract** on `ICache` — unlike the equivalent members on `ICache`, there is no default body, because `ICache` has no `GetCacheEntriesAsync` and so cannot distinguish a genuine miss from a cached `null` inside a default implementation. A hand-written `ICache` implementation (a test fake, typically) must add all three. There is no key-only convenience overload on the async surface: a caller whose keys are their own identity pairs each key with itself (`TState = CacheKey`). The blocking `GetOrAdd` facade above is the one place that accepts `CacheKey[]` directly and does that pairing for you. +`TryAddAsync` is the conditional-add (create-if-absent) member: it writes only when the key does not already exist, and returns `true` only to the caller that created it. On a Redis-backed cache it maps to StackExchange.Redis `When.NotExists` (`SET key value EX … NX`) — a single atomic round-trip, so exactly one caller across all nodes wins a given key. It is the primitive to reach for when you need at-most-once semantics keyed by something: a dedup marker, an idempotency key, a "who runs this job" election. See [`ICache.TryAddAsync`](#icache) for the full contract, including what a `false` return does and does not tell you. + > **Typical vs. power-user surface:** This is the standard typed surface. If you need to vary the value type or key per call rather than per cache instance, use [`ICache`](#icache) instead. `ICache` and `ICache` are different shapes for different problems — neither is strictly more capable. **Use this when:** @@ -140,6 +148,12 @@ public partial interface ICache : IDisposable ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default); + + ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + + ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + ValueTask RefreshAsync(CacheKey cacheKey, CachePolicy? policy = null, CancellationToken token = default); ValueTask RefreshAsync(CacheKey cacheKey, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default); @@ -158,6 +172,43 @@ public partial interface ICache : IDisposable The multi-key `GetOrAddAsync` overloads shown above pair each key with an opaque caller state (`TState`) and are **default interface methods** — each forwards to the shared `BatchGetOrAdd.RunAsync` machinery, so existing `ICache` implementations keep compiling without adding them. The generator is invoked at most once, only with the states of the entries that missed every cache layer, never with keys; results come back keyed by state, one entry per distinct requested state in first-occurrence order. Cache operations de-duplicate by `CacheKey`; results de-duplicate by state — when two states share a key the generator is asked once and the value is reported under both. Their parameter shape matches the single-key `GetOrAddAsync` exactly — `expiration`, `policy` and `token` all optional. `ICache.Compat.cs` carries no token-positional forwarder for them: that file is pre-`CachePolicy` back-compat sugar, and this API predates nothing. There is no key-only convenience overload; a caller whose keys are their own identity pairs each key with itself (`TState = CacheKey`). +`TryAddAsync` writes only if the key is absent — StackExchange.Redis `When.NotExists` (`SET … NX`) on Redis-backed caches, in one atomic command with the TTL applied by the same write. Four points decide whether it fits your problem: + +- **`false` is deliberately ambiguous.** It means "you did not create this key" — either it already existed, or the write could not be completed (backing store disconnected, write threw, or the value was a `null`/`default` that the cache has no way to represent). This is fail-closed by design: a caller treating `true` as "I own this key" is never wrongly told it won. The ambiguity is not recoverable: a serialization or command failure also returns `false`, with [`IConnectionState.IsConnected`](#iconnectionstate) still reporting healthy, and [`IDistributedLock.TryAcquireAsync`](#idistributedlock) conflates backend-unavailable with already-held in the same way. Design the `false` branch so that not proceeding is safe; if the two readings must be handled differently, the caller needs a primitive with a richer result than a `bool`. +- **It never deletes.** Where `SetAsync` removes the key when handed a `null` and `CacheNullValues` is off, `TryAddAsync` returns `false` and leaves the key untouched. With `CacheNullValues` on, a `null` claims the key via the cached-null sentinel. +- **It is a cache primitive, not a lock.** The entry expires on its own TTL, there is no ownership token, and any later `SetAsync`/`RemoveAsync` on the key ignores the claim. For mutual exclusion with a fencing token and explicit release, use [`IDistributedLock`](#idistributedlock). +- **An expiration that is not in the future claims nothing** and reports `false` on every tier. The entry would be evicted on arrival, so answering `true` would hand the same key to every later caller as well. + +Every provider runs the same sequence: take the local lock, probe the local tier, ask the L2, then populate the local tier on a win. Two tiers therefore contribute — the probe bounds exclusion at one winner per process, and the L2 decides how much further than that it reaches: + +| Provider | Who decides | Scope of exclusion | +| --- | --- | --- | +| `Redis` | Redis (`SET … NX`) | Cross-node | +| `InMemoryRedis` | L2 Redis, after a local probe that reports a hit as a loss; L1 populated only after a win, best-effort, plus an invalidation broadcast | Cross-node | +| `InMemoryRedis`, L2 disconnected | Nobody — returns `false` | None (fails closed rather than granting a claim every node would also get) | +| `InMemory` | The local probe, since `NullCache` as the L2 tells every caller it added the key. Serialized by the local lock — taken regardless of `Lock.LocalLockEnabled`, since here it *is* the guarantee; a caller that cannot acquire it within `Lock.LocalLockTimeout` is told it lost | In-process only, and against other conditional adds only: `SetAsync` takes no lock, so a set interleaved between the probe and the write is overwritten by the claim | +| `NullCache` | Nobody — returns `true` for every caller | None | +| Any other provider whose L2 resolved to `NullCache` | Whatever that L2 answers, so `true` for every caller. Which tier arbitrates is stated by the provider composing the cache — only `InMemory` arbitrates locally; every other provider asks its L2 and takes the answer as given | None | + +The L2 is asked through `ICache`, never classified by type or provider name, so a provider you register yourself participates on the same terms. Three consequences worth knowing: + +- **A local hit is reported as a loss without asking the L2.** Fail-closed and a round-trip saved, but a local copy that outlived the shared one costs a win the L2 would have granted. +- **An L2 that arbitrates in-process only is trusted as though it arbitrated for every sharer,** so pointing a provider's distributed tier at something in-process narrows that provider's exclusion with it. +- **A local write the L1 declines does not deny the win.** A size-limited `IMemoryCache` drops an entry it cannot fit without throwing; denying the claim there would strand a key the L2 already granted. On `InMemory`, where the L1 copy is the only copy, a `SizeLimit` too small to hold the value therefore means every caller wins. + +`NullCache.TryAddAsync` returns `true`, as its `SetAsync` does: the null store accepts every write and retains none, so no key can pre-exist and no caller loses the race — the same "caching is off, carry on" degradation the rest of that type applies. It provides no exclusion whatsoever, and it is what `ICacheFactory.CreateCache` falls back to when the requested provider is absent or has `Enabled=false`, so a mistyped or switched-off provider turns at-most-once into at-least-once. **Assert the provider you expect at startup** whenever the `true` branch runs a side effect that must not repeat: + +```csharp +if (cacheFactory.CreateCache(KnownCacheProviderNames.Redis) is NullCache) +{ + throw new InvalidOperationException("Webhook dedup needs the Redis provider registered and Enabled."); +} +``` + +**Retries are safe here,** unlike the destructive reads in `UiPath.Caching.Queue`, so the `NX` write stays on the shared `Write` pipeline. That pipeline retries on exceptions only — a `false` reply is never retried — and in the one ambiguous case the retry changes nothing: if attempt 1 creates the key and its reply is lost, the retried attempt is refused by that key and reports `false`, which is exactly what the un-retried exception would have reported. Where the first attempt never reached Redis, the retry recovers the correct `true` instead. Contrast `SPOP` behind `ISetCache.PopAsync`, where a retry pops a *second* item and loses the first — which is why `RedisSetCacheOptions.ResilienceKeyName` exists and defaults to no pipeline. + +The three overloads are **required** members of `ICache` and `ICache`, not default interface methods: a probe followed by a write is not atomic, and no fallback body could stand in for one without either voiding the guarantee or hiding which stores can arbitrate. A hand-written implementation therefore states what its own store can do — and existing implementations must add the member, which is a source-breaking change for them. There is no multi-key overload — Redis has no atomic multi-key `NX`, and "all-or-nothing" versus "per-key" would be a coin flip on the caller's behalf. + > **Typical vs. power-user surface:** `ICache` is the power-user surface. For most application code where the value type is fixed and you want automatic policy resolution, prefer [`ICache`](#icachet) — it is simpler and less error-prone. **Use this when:** @@ -228,7 +279,7 @@ public partial interface IHashCache } ``` -`IHashCache` is the typed hash-cache surface. Each cache key maps to a dictionary of named fields rather than a single value — the backing store is a Redis hash (or an in-memory equivalent). `GetItemAsync` retrieves a single field by name; `GetAsync` retrieves all fields or a subset. `SetAsync` accepts a `HashCacheEntryOptions` overload for fine-grained per-write control (e.g. conditional set, individual field TTL). Metadata (`GetMetadataAsync` / `SetMetadataAsync`) provides a side-channel string dictionary attached to the same key, useful for audit or versioning data. Sync overloads (`Get`, `GetItem`, `GetOrAdd`, `Set`, `Refresh`, `Remove`, `Contains`, etc.) are provided as blocking default interface methods. +`IHashCache` is the typed hash-cache surface. Each cache key maps to a dictionary of named fields rather than a single value — the backing store is a Redis hash (or an in-memory equivalent). `GetItemAsync` retrieves a single field by name; `GetAsync` retrieves all fields or a subset. `SetAsync` accepts a `HashCacheEntryOptions` overload for per-write control of expiration, metadata, and write scope — `HashCacheSetOption.HashReplace` merges the given fields into the existing hash, `KeyReplace` drops the key first so the written fields are the whole hash. Note that this is write *scope*, not a precondition: the hash surface has no conditional-add member, and `TryAddAsync` exists only on [`ICache`](#icache) / [`ICache`](#icachet). Metadata (`GetMetadataAsync` / `SetMetadataAsync`) provides a side-channel string dictionary attached to the same key, useful for audit or versioning data. Sync overloads (`Get`, `GetItem`, `GetOrAdd`, `Set`, `Refresh`, `Remove`, `Contains`, etc.) are provided as blocking default interface methods. > **Typical vs. power-user surface:** `IHashCache` is the standard typed hash surface. If you need to vary the value type per call, use [`IHashCache`](#ihashcache) instead. The two surfaces are different shapes for different problems. @@ -600,6 +651,29 @@ public interface IDistributedLock --- +## Connection state + +### `IConnectionState` + +```csharp +public interface IConnectionState +{ + event EventHandler? OnConnectionFailed; + + event EventHandler? OnConnectionRestored; + + event EventHandler? OnReconnected; + + bool IsConnected { get; } +} +``` + +A non-blocking snapshot of whether the backing store is reachable, plus the transitions as events. `IsConnected` never blocks and never throws. Implemented by `RedisCacheBase`, so `Redis`-provider caches expose it; the multilayer caches and `NullCache` do not, and neither does `Cache`, which holds its underlying `ICache` privately. + +What it is *not*: a way to explain a negative result. It is a cached snapshot refreshed on connection events and a timer, it says nothing about whether any particular command succeeded, and it is `true` both where there is nothing to disconnect from and where `ConnectionMonitorEnabled` is off. A `false` from `SetAsync` or [`TryAddAsync`](#icache) can perfectly well coincide with `IsConnected == true` — a serialization failure or a rejected command does that — so reading it afterwards does not recover why the call failed. + +**Use this when:** you are reporting or reacting to cache *health* — a readiness probe, a metric, a log line, or backing off writes while a tier is known down. Subscribe to `OnConnectionFailed` / `OnConnectionRestored` for the transitions rather than polling. + ## Telemetry seam ### `ICachingTelemetryProvider` diff --git a/src/UiPath.Caching.Abstractions/CacheOfT.cs b/src/UiPath.Caching.Abstractions/CacheOfT.cs index 6c920b66..bbb79b1f 100644 --- a/src/UiPath.Caching.Abstractions/CacheOfT.cs +++ b/src/UiPath.Caching.Abstractions/CacheOfT.cs @@ -95,6 +95,16 @@ public ValueTask SetAsync(CacheKey cacheKey, T? value, TimeSpan? expiratio public ValueTask SetAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default) => _cache.SetAsync(GetCacheKey(cacheKey), value, expiration, Policy, token); + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, CancellationToken token = default) => + _cache.TryAddAsync(GetCacheKey(cacheKey), value, policy: Policy, token: token); + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration, CancellationToken token = default) => + _cache.TryAddAsync(GetCacheKey(cacheKey), value, expiration, Policy, token); + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default) => + _cache.TryAddAsync(GetCacheKey(cacheKey), value, expiration, Policy, token); + + public ValueTask SetAsync(KeyValuePair[] keyValues, CancellationToken token = default) => _cache.SetAsync(GetKeyValuePairs(keyValues), policy: Policy, token: token); diff --git a/src/UiPath.Caching.Abstractions/CacheOptions.cs b/src/UiPath.Caching.Abstractions/CacheOptions.cs index da1fb64d..e397a160 100644 --- a/src/UiPath.Caching.Abstractions/CacheOptions.cs +++ b/src/UiPath.Caching.Abstractions/CacheOptions.cs @@ -35,6 +35,7 @@ public class CacheOptions public bool ConnectionMonitorEnabled { get; set; } + /// /// Size of the reusable semaphore pool inside the default local lock implementation. /// This is an allocation hint, not a hard concurrency cap — when the pool is exhausted the diff --git a/src/UiPath.Caching.Abstractions/ICache.Compat.cs b/src/UiPath.Caching.Abstractions/ICache.Compat.cs index d8f1a5e9..588b5f38 100644 --- a/src/UiPath.Caching.Abstractions/ICache.Compat.cs +++ b/src/UiPath.Caching.Abstractions/ICache.Compat.cs @@ -57,6 +57,18 @@ ValueTask SetAsync(KeyValuePair[] keyValues, TimeSpan? ex ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration, CancellationToken token = default) => SetAsync(keyValues, expiration, null, token); + [ExcludeFromCodeCoverage] + ValueTask TryAddAsync(CacheKey cacheKey, T? value, CancellationToken token = default) + => TryAddAsync(cacheKey, value, (CachePolicy?)null, token); + + [ExcludeFromCodeCoverage] + ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration, CancellationToken token = default) + => TryAddAsync(cacheKey, value, expiration, null, token); + + [ExcludeFromCodeCoverage] + ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default) + => TryAddAsync(cacheKey, value, expiration, null, token); + [ExcludeFromCodeCoverage] ValueTask RefreshAsync(CacheKey cacheKey, CancellationToken token = default) => RefreshAsync(cacheKey, (CachePolicy?)null, token); diff --git a/src/UiPath.Caching.Abstractions/ICache.cs b/src/UiPath.Caching.Abstractions/ICache.cs index 09f61d69..36f4aaf2 100644 --- a/src/UiPath.Caching.Abstractions/ICache.cs +++ b/src/UiPath.Caching.Abstractions/ICache.cs @@ -46,6 +46,26 @@ public partial interface ICache : IDisposable ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + /// + /// Writes only if is absent. Redis + /// SET … NX, one atomic round-trip. + /// + /// + /// true only if this call created the key. false conflates "it existed" with "the + /// write could not be completed", deliberately and fail-closed. Never deletes; not a lock. + /// + ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default); + + /// + /// + /// Lifetime if the entry is created, applied by the same command. Falls back to + /// CachePolicy.DistributedExpiration then the cache default. Not in the future: no-op. + /// + ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + + /// + ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default); + ValueTask RefreshAsync(CacheKey cacheKey, CachePolicy? policy = null, CancellationToken token = default); ValueTask RefreshAsync(CacheKey cacheKey, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default); diff --git a/src/UiPath.Caching.Abstractions/ICacheOfT.Sync.cs b/src/UiPath.Caching.Abstractions/ICacheOfT.Sync.cs index 1063bfbe..24c1aa18 100644 --- a/src/UiPath.Caching.Abstractions/ICacheOfT.Sync.cs +++ b/src/UiPath.Caching.Abstractions/ICacheOfT.Sync.cs @@ -84,6 +84,18 @@ bool Set(KeyValuePair[] keyValues, TimeSpan? expiration = null, Ca bool Set(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CancellationToken token = default) => SetAsync(keyValues, expiration, token).AsTask().GetAwaiter().GetResult(); + [ExcludeFromCodeCoverage] + bool TryAdd(CacheKey cacheKey, T? value, CancellationToken token = default) + => TryAddAsync(cacheKey, value, token).AsTask().GetAwaiter().GetResult(); + + [ExcludeFromCodeCoverage] + bool TryAdd(CacheKey cacheKey, T? value, TimeSpan? expiration, CancellationToken token = default) + => TryAddAsync(cacheKey, value, expiration, token).AsTask().GetAwaiter().GetResult(); + + [ExcludeFromCodeCoverage] + bool TryAdd(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default) + => TryAddAsync(cacheKey, value, expiration, token).AsTask().GetAwaiter().GetResult(); + [ExcludeFromCodeCoverage] bool Refresh(CacheKey cacheKey, CancellationToken token = default) => RefreshAsync(cacheKey, token).AsTask().GetAwaiter().GetResult(); diff --git a/src/UiPath.Caching.Abstractions/ICacheOfT.cs b/src/UiPath.Caching.Abstractions/ICacheOfT.cs index 47d49390..617d804b 100644 --- a/src/UiPath.Caching.Abstractions/ICacheOfT.cs +++ b/src/UiPath.Caching.Abstractions/ICacheOfT.cs @@ -35,6 +35,20 @@ public partial interface ICache ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CancellationToken token = default); + /// + /// Typed façade over + /// ; + /// see that member for the contract. + /// + ValueTask TryAddAsync(CacheKey cacheKey, T? value, CancellationToken token = default); + + /// + /// Lifetime of the entry if it is created. + ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration, CancellationToken token = default); + + /// + ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration, CancellationToken token = default); + ValueTask RefreshAsync(CacheKey cacheKey, CancellationToken token = default); ValueTask RefreshAsync(CacheKey cacheKey, TimeSpan? expiration, CancellationToken token = default); diff --git a/src/UiPath.Caching.Abstractions/NullCache.cs b/src/UiPath.Caching.Abstractions/NullCache.cs index 5da0af1d..d4d71b71 100644 --- a/src/UiPath.Caching.Abstractions/NullCache.cs +++ b/src/UiPath.Caching.Abstractions/NullCache.cs @@ -76,6 +76,20 @@ public ValueTask ContainsAsync(CacheKey cacheKey, CancellationToken tok public ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => ReturnTrueAsync(); + /// + /// Always true, as SetAsync is here: nothing is retained, so no key pre-exists and + /// nobody loses. No exclusion at all, then — and this type is what + /// ICacheFactory.CreateCache resolves to for an absent or disabled provider, so assert the + /// provider you expect at startup. + /// + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default) => ReturnTrueAsync(); + + /// + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => ReturnTrueAsync(); + + /// + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => ReturnTrueAsync(); + public ValueTask TimeToLiveAsync(CacheKey cacheKey, CancellationToken token = default) { NotCacheableException.ThrowIfNotCacheable(); diff --git a/src/UiPath.Caching.Abstractions/Policies/ResiliencePipelineNames.cs b/src/UiPath.Caching.Abstractions/Policies/ResiliencePipelineNames.cs index 8e4ffd38..8742275d 100644 --- a/src/UiPath.Caching.Abstractions/Policies/ResiliencePipelineNames.cs +++ b/src/UiPath.Caching.Abstractions/Policies/ResiliencePipelineNames.cs @@ -5,4 +5,5 @@ public static class ResiliencePipelineNames public const string Read = "read"; public const string Write = "write"; + } diff --git a/src/UiPath.Caching.Abstractions/PublicAPI.Unshipped.txt b/src/UiPath.Caching.Abstractions/PublicAPI.Unshipped.txt index e4688ba1..12654adc 100644 --- a/src/UiPath.Caching.Abstractions/PublicAPI.Unshipped.txt +++ b/src/UiPath.Caching.Abstractions/PublicAPI.Unshipped.txt @@ -1,21 +1,39 @@ #nullable enable +UiPath.Caching.Cache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.Cache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.Cache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask UiPath.Caching.CacheKey.CacheKey(string? name, UiPath.Caching.CacheKeyCasing casing) -> void UiPath.Caching.CacheKey.Casing.get -> UiPath.Caching.CacheKeyCasing UiPath.Caching.CacheKey.WithName(string? name) -> UiPath.Caching.CacheKey UiPath.Caching.CacheKeyCasing UiPath.Caching.CacheKeyCasing.Insensitive = 0 -> UiPath.Caching.CacheKeyCasing +UiPath.Caching.CacheKeyCasing.Sensitive = 1 -> UiPath.Caching.CacheKeyCasing UiPath.Caching.CacheKeyComparer UiPath.Caching.CacheKeyComparer.CacheKeyComparer() -> void -static UiPath.Caching.CacheKeyComparer.Insensitive.get -> UiPath.Caching.CacheKeyComparer! -static UiPath.Caching.CacheKeyComparer.Sensitive.get -> UiPath.Caching.CacheKeyComparer! -UiPath.Caching.CacheKeyCasing.Sensitive = 1 -> UiPath.Caching.CacheKeyCasing UiPath.Caching.CacheOptions.KeyCasing.get -> UiPath.Caching.CacheKeyCasing UiPath.Caching.CacheOptions.KeyCasing.set -> void +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration = null, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration = null, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAdd(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> bool +UiPath.Caching.ICache.TryAdd(UiPath.Caching.CacheKey cacheKey, T? value, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> bool +UiPath.Caching.ICache.TryAdd(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> bool +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.ICache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.NullCache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.DateTimeOffset? expiration = null, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.NullCache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, System.TimeSpan? expiration = null, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask +UiPath.Caching.NullCache.TryAddAsync(UiPath.Caching.CacheKey cacheKey, T? value, UiPath.Caching.CachePolicy? policy = null, System.Threading.CancellationToken token = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask UiPath.Caching.SystemJsonByteSerializerProxy -UiPath.Caching.SystemJsonByteSerializerProxy.SystemJsonByteSerializerProxy(System.Text.Json.JsonSerializerOptions? options = null) -> void -UiPath.Caching.SystemJsonByteSerializerProxy.Serialize(object? value) -> byte[]? UiPath.Caching.SystemJsonByteSerializerProxy.Deserialize(byte[]? value) -> T? -UiPath.Caching.SystemJsonByteSerializerProxy.TryDeserialize(string? value, out T? result) -> bool +UiPath.Caching.SystemJsonByteSerializerProxy.Serialize(object? value) -> byte[]? +UiPath.Caching.SystemJsonByteSerializerProxy.SystemJsonByteSerializerProxy(System.Text.Json.JsonSerializerOptions? options = null) -> void UiPath.Caching.SystemJsonByteSerializerProxy.TryDeserialize(object? value, out T? result) -> bool +UiPath.Caching.SystemJsonByteSerializerProxy.TryDeserialize(string? value, out T? result) -> bool static UiPath.Caching.CacheKey.DefaultCasing.get -> UiPath.Caching.CacheKeyCasing static UiPath.Caching.CacheKey.DefaultCasing.set -> void +static UiPath.Caching.CacheKeyComparer.Insensitive.get -> UiPath.Caching.CacheKeyComparer! +static UiPath.Caching.CacheKeyComparer.Sensitive.get -> UiPath.Caching.CacheKeyComparer! diff --git a/src/UiPath.Caching/MultilayerCache.cs b/src/UiPath.Caching/MultilayerCache.cs index 07a737df..93b9edef 100644 --- a/src/UiPath.Caching/MultilayerCache.cs +++ b/src/UiPath.Caching/MultilayerCache.cs @@ -656,6 +656,124 @@ public async ValueTask SetAsync(CacheKey cacheKey, T? value, DateTimeOf } } + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default) + { + policy ??= _defaultPolicy; + return TryAddAsync(cacheKey, value, ResolveWriteDuration(policy), policy, token); + } + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default) + { + policy ??= _defaultPolicy; + return TryAddAsync(cacheKey, value, _clock.ToDateTimeOffset(ResolveWriteDuration(policy, expiration)), policy, token); + } + + /// + /// One path for every provider: take the local lock, probe the local tier, let the L2 decide, + /// then populate the local tier — the reverse of SetAsync, which writes both tiers + /// unconditionally. The probe is what narrows an L2 that retains nothing, and so grants every + /// caller a win, back to one winner per process; where the L2 does arbitrate, a local hit means + /// the key was already claimed or read here, so the loss is reported without a round-trip. That + /// can cost a win the L2 would have granted, when the local copy outlived the shared one — the + /// fail-closed direction the ambiguous false already covers. A disconnected L2 answers + /// for itself (RedisCache checks its connection first) rather than being gated on + /// GetInnerCacheDisconnected, whose state also covers the broadcast transport: a dead + /// topic must not stop a healthy Redis. + /// + public async ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) + { + NotCacheableException.ThrowIfNotCacheable(); + policy ??= _defaultPolicy; + expiration ??= _clock.ToDateTimeOffset(ResolveWriteDuration(policy)); + var options = _entryBuilder.BuildEntryOptions(cacheKey, expiration, token); + + if (value is null && !_multiLayerCacheOptions.CacheNullValues) + { + LogTryAddSkippedUnrepresentableValue(options.CacheKey); + return false; + } + + if (options.Expiration <= _clock.UtcNow) + { + LogTryAddSkippedExpiredEntry(options.CacheKey, options.Expiration); + return false; + } + + var localMaxExpiration = policy.LocalExpiration ?? _multiLayerCacheOptions.LocalMaxExpiration; + if (localMaxExpiration is { } max && max <= TimeSpan.Zero) + { + LogTryAddSkippedNonPositiveLocalRetention(options.CacheKey, max); + return false; + } + + var localLock = await AcquireLocalLockAsync(options.CacheKey, policy.Lock, options.Token).ConfigureAwait(false); + if (localLock is null) + { + LogTryAddLocalLockUnavailable(options.CacheKey); + return false; + } + + using (localLock) + { + return await TryAddUnderLocalLockAsync(options, value, policy, localMaxExpiration).ConfigureAwait(false); + } + } + + /// + /// Probe the local tier, then let the L2 decide. The probe is what narrows a store that retains + /// nothing — and therefore grants every caller a win — back to one winner per process; where the + /// L2 does arbitrate, a local hit means the key was already claimed or read here, so reporting + /// the loss early is both correct and a round-trip saved. It can cost a win the L2 would have + /// granted, when the local copy outlived the shared one; that is the fail-closed direction the + /// ambiguous false already covers. + /// + private async ValueTask TryAddUnderLocalLockAsync(CacheEntryOptions options, T? value, CachePolicy policy, TimeSpan? localMaxExpiration) + { + if (_memoryCache.TryGetValue(options.CacheKey, out _)) + { + return false; + } + + bool added; + try + { + added = await _innerCache.TryAddAsync(options.CacheKey, value, options.Expiration, policy, options.Token).ConfigureAwait(false); + } + catch (Exception ex) when (!(ex is OperationCanceledException && options.Token.IsCancellationRequested)) + { + LogInnerCacheTryAddError(ex, options.CacheKey); + return false; + } + + if (!added) + { + return false; + } + + // Best-effort after the win: a loss reported here would strand the entry with no owner. + try + { + if (!await _eventPublisher.CacheSetAsync(options).ConfigureAwait(false)) + { + LogTryAddBroadcastNotPublished(options.CacheKey); + } + } + catch (Exception ex) + { + LogTryAddLocalPropagationFailed(ex, options.CacheKey); + } + + try + { + MemorySet(options, value, localMaxExpiration); + } + catch (Exception ex) + { + LogTryAddLocalPropagationFailed(ex, options.CacheKey); + } + + return true; + } public ValueTask SetAsync(KeyValuePair[] keyValues, CachePolicy? policy = null, CancellationToken token = default) { @@ -1124,6 +1242,27 @@ private readonly struct CacheEntryValue(CacheEntryOptions cacheEntry, T? valu [LoggerMessage(Level = LogLevel.Debug, Message = "Batch cache missed. generating {Count} keys for {CacheKey}")] private partial void LogBatchCacheMissed(CacheKey cacheKey, int count); + [LoggerMessage(Level = LogLevel.Debug, Message = "TryAdd skipped for {CacheKey}: a null value cannot be represented unless CacheNullValues is on.")] + private partial void LogTryAddSkippedUnrepresentableValue(CacheKey cacheKey); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Inner cache TryAdd for cacheKey {CacheKey}")] + private partial void LogInnerCacheTryAddError(Exception ex, CacheKey cacheKey); + + [LoggerMessage(Level = LogLevel.Warning, Message = "TryAdd won cacheKey {CacheKey} in the inner cache but could not broadcast or populate the local tier. The add still stands.")] + private partial void LogTryAddLocalPropagationFailed(Exception ex, CacheKey cacheKey); + + [LoggerMessage(Level = LogLevel.Warning, Message = "TryAdd for {CacheKey} reported not-added: the local lock could not be acquired within Lock.LocalLockTimeout, and without it two in-process callers could both be told they added the key. Raise Lock.LocalLockTimeout if this key is contended.")] + private partial void LogTryAddLocalLockUnavailable(CacheKey cacheKey); + + [LoggerMessage(Level = LogLevel.Debug, Message = "TryAdd skipped for {CacheKey}: the requested expiration {Expiration} is not in the future, so the entry would retain nothing.")] + private partial void LogTryAddSkippedExpiredEntry(CacheKey cacheKey, DateTimeOffset expiration); + + [LoggerMessage(Level = LogLevel.Debug, Message = "TryAdd skipped for {CacheKey}: the effective local retention {LocalMaxExpiration} is not positive, and on this provider it is the only retention.")] + private partial void LogTryAddSkippedNonPositiveLocalRetention(CacheKey cacheKey, TimeSpan localMaxExpiration); + + [LoggerMessage(Level = LogLevel.Warning, Message = "TryAdd won cacheKey {CacheKey} but the invalidation broadcast reported not-published. The add still stands; peers may serve a stale copy until it expires.")] + private partial void LogTryAddBroadcastNotPublished(CacheKey cacheKey); + [LoggerMessage(Level = LogLevel.Debug, Message = "Replacing cached key {CacheKey}")] private partial void LogReplacingCachedKey(CacheKey cacheKey); diff --git a/src/UiPath.Caching/MultilayerCacheBase.cs b/src/UiPath.Caching/MultilayerCacheBase.cs index d5bc0775..619377b5 100644 --- a/src/UiPath.Caching/MultilayerCacheBase.cs +++ b/src/UiPath.Caching/MultilayerCacheBase.cs @@ -100,10 +100,9 @@ protected async ValueTask RunUnderLocksAsync( CancellationToken token, LockProfile? policyLock = null) { - var localLockEnabled = policyLock?.LocalLockEnabled ?? _localLockEnabled; + var (localLockEnabled, localLockTimeout) = ResolveLocalLock(policyLock); var distributedLockEnabled = policyLock?.DistributedLockEnabled ?? _distributedLockEnabled; // Per-call LockProfile bypasses options validators; mirror LockSettingsValidator's accepted ranges and fall back when out-of-range. - var localLockTimeout = PositiveOrFallback(policyLock?.LocalLockTimeout, _localLockTimeout); var distributedLockTimeout = NonNegativeOrFallback(policyLock?.DistributedLockTimeout, _distributedLockTimeout); var distributedLockExpiry = PositiveOrFallback(policyLock?.DistributedLockExpiry, _distributedLockExpiry); @@ -193,6 +192,23 @@ private static TimeSpan NonNegativeOrFallback(TimeSpan? value, TimeSpan fallback return ApplyJitter(resolved, policy.JitterMaxDuration, _clock.UtcNow); } + /// + /// The local lock alone, for callers that need it for correctness rather than de-duplication. + /// Taken regardless of Lock.LocalLockEnabled, which only trades single-flight for + /// throughput on GetOrAddAsync; null means the acquire timed out, and the caller + /// must fail closed. + /// + private protected ValueTask AcquireLocalLockAsync(CacheKey cacheKey, LockProfile? policyLock, CancellationToken token) => + TryAcquireLocalLockAsync(cacheKey, ResolveLocalLock(policyLock).Timeout, token); + + /// + /// One place for the local-lock policy: a per-call wins over the + /// options. It bypasses the options validators, so the timeout falls back when out of range. + /// + private (bool Enabled, TimeSpan Timeout) ResolveLocalLock(LockProfile? policyLock) => + (policyLock?.LocalLockEnabled ?? _localLockEnabled, + PositiveOrFallback(policyLock?.LocalLockTimeout, _localLockTimeout)); + private async ValueTask TryAcquireLocalLockAsync(CacheKey cacheKey, TimeSpan localLockTimeout, CancellationToken token) { var lockKey = _localLockKeyPrefix + cacheKey.Name; diff --git a/src/UiPath.Caching/PublicAPI.Unshipped.txt b/src/UiPath.Caching/PublicAPI.Unshipped.txt index 1c5428d2..00628a24 100644 --- a/src/UiPath.Caching/PublicAPI.Unshipped.txt +++ b/src/UiPath.Caching/PublicAPI.Unshipped.txt @@ -1,20 +1,20 @@ #nullable enable -const UiPath.Caching.Config.DistributedCacheCollectionExtensions.DistributedCacheServiceKey = "UiPath.Caching.Distributed" -> string! -static UiPath.Caching.Config.DistributedCacheCollectionExtensions.AddDistributedCache(this UiPath.Caching.Config.ICachingBuilder! builder, string! providerName, System.Action? configure = null) -> UiPath.Caching.Config.ICachingBuilder! UiPath.Caching.Config.DistributedCacheCollectionExtensions UiPath.Caching.Distributed.UiPathDistributedCacheOptions -UiPath.Caching.Distributed.UiPathDistributedCacheOptions.PolicyName.get -> string? -UiPath.Caching.Distributed.UiPathDistributedCacheOptions.PolicyName.set -> void -UiPath.Caching.Distributed.UiPathDistributedCacheOptions.UiPathDistributedCacheOptions() -> void -UiPath.Caching.Distributed.UiPathDistributedCacheOptions.DefaultEntryExpiration.get -> System.TimeSpan? -UiPath.Caching.Distributed.UiPathDistributedCacheOptions.DefaultEntryExpiration.set -> void UiPath.Caching.Distributed.UiPathDistributedCacheOptions.AllowUnboundedEntries.get -> bool UiPath.Caching.Distributed.UiPathDistributedCacheOptions.AllowUnboundedEntries.set -> void -UiPath.Caching.Redis.RedisCacheOptions.AwaitRefresh.get -> bool -UiPath.Caching.Redis.RedisCacheOptions.AwaitRefresh.set -> void UiPath.Caching.Distributed.UiPathDistributedCacheOptions.CacheKeyStrategy.get -> UiPath.Caching.ICacheKeyStrategy? UiPath.Caching.Distributed.UiPathDistributedCacheOptions.CacheKeyStrategy.set -> void +UiPath.Caching.Distributed.UiPathDistributedCacheOptions.DefaultEntryExpiration.get -> System.TimeSpan? +UiPath.Caching.Distributed.UiPathDistributedCacheOptions.DefaultEntryExpiration.set -> void +UiPath.Caching.Distributed.UiPathDistributedCacheOptions.PolicyName.get -> string? +UiPath.Caching.Distributed.UiPathDistributedCacheOptions.PolicyName.set -> void UiPath.Caching.Distributed.UiPathDistributedCacheOptions.RedisKeyDifferentiator.get -> string? UiPath.Caching.Distributed.UiPathDistributedCacheOptions.RedisKeyDifferentiator.set -> void UiPath.Caching.Distributed.UiPathDistributedCacheOptions.RedisKeyStrategyFactory.get -> UiPath.Caching.Redis.IRedisKeyStrategyFactory? UiPath.Caching.Distributed.UiPathDistributedCacheOptions.RedisKeyStrategyFactory.set -> void +UiPath.Caching.Distributed.UiPathDistributedCacheOptions.UiPathDistributedCacheOptions() -> void +UiPath.Caching.Redis.RedisCacheOptions.AwaitRefresh.get -> bool +UiPath.Caching.Redis.RedisCacheOptions.AwaitRefresh.set -> void +const UiPath.Caching.Config.DistributedCacheCollectionExtensions.DistributedCacheServiceKey = "UiPath.Caching.Distributed" -> string! +static UiPath.Caching.Config.DistributedCacheCollectionExtensions.AddDistributedCache(this UiPath.Caching.Config.ICachingBuilder! builder, string! providerName, System.Action? configure = null) -> UiPath.Caching.Config.ICachingBuilder! diff --git a/src/UiPath.Caching/Redis/RedisCache.cs b/src/UiPath.Caching/Redis/RedisCache.cs index f42bbcbe..bc7ebceb 100644 --- a/src/UiPath.Caching/Redis/RedisCache.cs +++ b/src/UiPath.Caching/Redis/RedisCache.cs @@ -237,6 +237,23 @@ public ValueTask SetAsync(KeyValuePair[] keyValues, DateT return SetInternalAsync(keyValues, Clock.ToTimeSpan(effective), token); } + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default) => + TryAddAsync(cacheKey, value, expiration: (TimeSpan?)null, policy, token); + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default) + { + NotCacheableException.ThrowIfNotCacheable(); + var effective = ResolveExpiration(expiration, policy); + return TryAddInternalAsync(ToRedisKey(cacheKey, token), value, Clock.ToTimeSpan(effective), token); + } + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) + { + NotCacheableException.ThrowIfNotCacheable(); + var effective = ResolveExpiration(expiration, policy); + return TryAddInternalAsync(ToRedisKey(cacheKey, token), value, Clock.ToTimeSpan(effective), token); + } + public async ValueTask ContainsAsync(CacheKey cacheKey, CancellationToken token = default) { NotCacheableException.ThrowIfNotCacheable(); @@ -459,6 +476,64 @@ private async ValueTask SetInternalAsync(RedisKey redisKey, T? value, T return ret; } + /// + /// SET key value EX … NX in one round-trip: Redis decides, so no probe precedes the write. + /// Safe to retry — an attempt whose reply was lost is refused by the key it just wrote, and + /// reports the same false the exception would have. + /// + private async ValueTask TryAddInternalAsync(RedisKey redisKey, T? value, TimeSpan expiration, CancellationToken token) + { + bool ret = default; + token.ThrowIfCancellationRequested(); + + if (!IsConnected) + { + return false; + } + + var operation = StartOperation(nameof(TryAddAsync)); + try + { + var isNull = IsDefault(value); + if (expiration <= TimeSpan.Zero) + { + LogTryAddSkippedExpiredEntry(redisKey, expiration); + } + else if (isNull && !_cacheNullValues) + { + LogTryAddSkippedUnrepresentableValue(redisKey); + } + else + { + var serialized = isNull ? RedisValue.EmptyString : _serializer.Serialize(value); + + ret = await _write.ExecuteAsync(async token => + { + token.ThrowIfCancellationRequested(); + return await Database.StringSetAsync(redisKey, serialized, expiration, When.NotExists, CommandFlags.DemandMaster).ConfigureAwait(false); + }, default, token).ConfigureAwait(false); + } + operation.Stop(); + } + // false would claim someone else owns the key, which a cancelled call never established. + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + operation.Stop(); + throw; + } + catch (Exception ex) + { + operation.Stop(); + LogRedisCacheException(ex); + } + finally + { + operation.Track(ret); + } + + return ret; + } + private async ValueTask SetInternalAsync(KeyValuePair[] keyValues, TimeSpan expiration, CancellationToken token) { bool ret = default; @@ -881,6 +956,12 @@ private void AuditKeySize(RedisKey key, RedisValue value) [LoggerMessage(Level = LogLevel.Trace, Message = "Refreshing key {RedisKey} at expiration {Expiration}")] private partial void LogRefreshingKey(RedisKey redisKey, DateTimeOffset? expiration); + [LoggerMessage(Level = LogLevel.Debug, Message = "TryAdd skipped for {RedisKey}: a default value cannot be represented unless CacheNullValues is on.")] + private partial void LogTryAddSkippedUnrepresentableValue(RedisKey redisKey); + + [LoggerMessage(Level = LogLevel.Debug, Message = "TryAdd skipped for {RedisKey}: the requested expiration {Expiration} is not positive, so the entry would retain nothing.")] + private partial void LogTryAddSkippedExpiredEntry(RedisKey redisKey, TimeSpan expiration); + [LoggerMessage(Level = LogLevel.Warning, Message = "RedisCache exception.")] private partial void LogRedisCacheException(Exception ex); diff --git a/tests/UiPath.Caching.Tests/CacheOfTTryAddTests.cs b/tests/UiPath.Caching.Tests/CacheOfTTryAddTests.cs new file mode 100644 index 00000000..e98f84bc --- /dev/null +++ b/tests/UiPath.Caching.Tests/CacheOfTTryAddTests.cs @@ -0,0 +1,91 @@ +using UiPath.Caching.Tests.Fakes; + +namespace UiPath.Caching.Tests; + +/// +/// The typed façade. owns two things the untyped surface does not: the key +/// strategy that namespaces keys per type, and the policy snapshot taken at construction. Both must +/// apply to a conditional add exactly as they do to SetAsync, or a caller electing a winner +/// under Cache<A> would collide with one under Cache<B>. +/// +public class CacheOfTTryAddTests(ITestContextAccessor testContextAccessor) +{ + private CancellationToken Ct => testContextAccessor.Current.CancellationToken; + + [Fact] + public async Task TryAdd_forwards_to_the_inner_cache_and_reports_the_win() + { + var inner = new DictionaryCache(); + var sut = new Cache(inner); + + (await sut.TryAddAsync("k", "first", Ct)).Should().BeTrue(); + (await sut.TryAddAsync("k", "second", Ct)).Should().BeFalse(); + inner.TryAddCalls.Should().Be(2); + } + + [Fact] + public async Task TryAdd_does_not_overwrite_the_winner() + { + var inner = new DictionaryCache(); + var sut = new Cache(inner); + + await sut.TryAddAsync("k", "first", Ct); + await sut.TryAddAsync("k", "second", Ct); + + (await sut.GetAsync("k", Ct)).Should().Be("first"); + } + + [Fact] + public async Task TryAdd_applies_the_type_key_strategy() + { + var inner = new DictionaryCache(); + var keyStrategy = Substitute.For(); + keyStrategy.GetCacheKey(Arg.Any()).Returns((CacheKey)"namespaced:k"); + var sut = new Cache(inner, keyStrategy); + + await sut.TryAddAsync("k", "v", Ct); + + inner.Contains("namespaced:k").Should().BeTrue(); + inner.Contains("k").Should().BeFalse("an unnamespaced claim could collide with another type's key"); + } + + [Fact] + public async Task TryAdd_passes_the_constructed_policy_through() + { + var policy = new CachePolicy { DistributedExpiration = TimeSpan.FromMinutes(11) }; + var inner = Substitute.For(); + inner.TryAddAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + var sut = new Cache(inner, cacheKeyStrategy: null, policy); + + await sut.TryAddAsync("k", "v", Ct); + + await inner.Received(1).TryAddAsync(Arg.Any(), "v", policy, Arg.Any()); + } + + [Fact] + public async Task TryAdd_forwards_both_expiration_shapes_with_the_policy() + { + var policy = new CachePolicy { DistributedExpiration = TimeSpan.FromMinutes(11) }; + var inner = Substitute.For(); + var sut = new Cache(inner, cacheKeyStrategy: null, policy); + var ttl = TimeSpan.FromMinutes(3); + var absolute = DateTimeOffset.UtcNow.AddMinutes(3); + + await sut.TryAddAsync("k", "v", ttl, Ct); + await sut.TryAddAsync("k", "v", absolute, Ct); + + await inner.Received(1).TryAddAsync(Arg.Any(), "v", ttl, policy, Arg.Any()); + await inner.Received(1).TryAddAsync(Arg.Any(), "v", absolute, policy, Arg.Any()); + } + + [Fact] + public void TryAdd_blocking_forwarder_matches_the_async_result() + { + var inner = new DictionaryCache(); + ICache sut = new Cache(inner); + + sut.TryAdd("k", "first", Ct).Should().BeTrue(); + sut.TryAdd("k", "second", Ct).Should().BeFalse(); + } +} diff --git a/tests/UiPath.Caching.Tests/Fakes/DictionaryCache.cs b/tests/UiPath.Caching.Tests/Fakes/DictionaryCache.cs index 24c37b7a..26f1529d 100644 --- a/tests/UiPath.Caching.Tests/Fakes/DictionaryCache.cs +++ b/tests/UiPath.Caching.Tests/Fakes/DictionaryCache.cs @@ -13,6 +13,8 @@ internal sealed class DictionaryCache : ICache public int SetCalls { get; private set; } + public int TryAddCalls { get; private set; } + public List SetKeySets { get; } = []; public void Seed(CacheKey key, T? value) => _store[key] = value; @@ -86,6 +88,22 @@ public ValueTask SetAsync(KeyValuePair[] keyValues, TimeS public ValueTask SetAsync(KeyValuePair[] keyValues, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => SetAsync(keyValues, policy, token); + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, CachePolicy? policy = null, CancellationToken token = default) + { + TryAddCalls++; + if (value is null && !CacheNullValues) + { + return ValueTask.FromResult(false); + } + return ValueTask.FromResult(_store.TryAdd(cacheKey, value)); + } + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, TimeSpan? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => + TryAddAsync(cacheKey, value, policy, token); + + public ValueTask TryAddAsync(CacheKey cacheKey, T? value, DateTimeOffset? expiration = null, CachePolicy? policy = null, CancellationToken token = default) => + TryAddAsync(cacheKey, value, policy, token); + public ValueTask RemoveAsync(CacheKey cacheKey, CancellationToken token = default) => ValueTask.FromResult(_store.Remove(cacheKey)); diff --git a/tests/UiPath.Caching.Tests/MultilayerCacheTryAddTests.cs b/tests/UiPath.Caching.Tests/MultilayerCacheTryAddTests.cs new file mode 100644 index 00000000..e0c3483e --- /dev/null +++ b/tests/UiPath.Caching.Tests/MultilayerCacheTryAddTests.cs @@ -0,0 +1,539 @@ +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.Internal; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute.ExceptionExtensions; +using UiPath.Caching; +using UiPath.Caching.Config; +using UiPath.Caching.Locking; +using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Broadcast; + +namespace UiPath.Caching.Tests; + +/// +/// Conditional add across the two tiers. The invariant: L1 never arbitrates while an L2 exists — a +/// key missing locally may still exist in the shared store, so a local probe would hand the same win +/// to every node. +/// +public class MultilayerCacheTryAddTests(ITestContextAccessor testContextAccessor) : IAsyncLifetime +{ + private readonly IFixture _fixture = AutoFixtureCreator.NSubstitute(); + + private ICache _innerCache = default!; + private IChangeTokenFactory _changeTokenFactory = default!; + private ITopicFactory _topicFactory = default!; + private ITopicProviderWithConnectionState _topicProvider = default!; + private ITopic _topic = default!; + private IMemoryCache _memoryCache = default!; + private IMemoryCacheFactory _memoryCacheFactory = default!; + private InMemoryRedisCacheOptions _options = default!; + private TopicKey _topicKey = default!; + private CacheKey _cacheKey = default!; + private ILogger _logger = default!; + + private MultilayerCache? _sut; + + private MultilayerCache Sut => _sut ??= _fixture.Create(); + + private CancellationToken Ct => testContextAccessor.Current.CancellationToken; + + [Fact] + public async Task TryAdd_delegates_the_decision_to_the_inner_cache() + { + var value = _fixture.Create(); + _innerCache.TryAddAsync(_cacheKey, value, Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()).Returns(_ => true); + + var added = await Sut.TryAddAsync(_cacheKey, value, policy: null, token: Ct); + + added.Should().BeTrue(); + await _innerCache.Received(1).TryAddAsync(_cacheKey, value, Arg.Any(), Arg.Any(), Arg.Any()); + _memoryCache.Received(1).CreateEntry(_cacheKey); + } + + [Fact] + public async Task A_local_hit_reports_the_loss_without_asking_the_L2() + { + _memoryCache.TryGetValue(_cacheKey, out Arg.Any()) + .Returns(x => + { + x[1] = new TestCacheEntry { Value = _fixture.Create() }; + return true; + }); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeFalse(); + await _innerCache.DidNotReceive().TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_leaves_both_tiers_untouched_when_the_inner_cache_reports_a_loss() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(false); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeFalse(); + _memoryCache.DidNotReceive().CreateEntry(_cacheKey); + await _topic.DidNotReceive().PublishAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_broadcasts_after_a_win_so_peers_drop_stale_local_copies() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()).Returns(_ => true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + await _topic.Received(1).PublishAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_still_reports_the_win_when_the_broadcast_fails() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("broadcast down")); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeTrue("the key is claimed in the shared store; denying the win would strand it with no owner"); + } + + [Fact] + public async Task TryAdd_fails_closed_when_the_inner_cache_throws() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("redis down")); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeFalse(); + _memoryCache.DidNotReceive().CreateEntry(_cacheKey); + } + + [Fact] + public async Task TryAdd_claims_nothing_for_an_expiration_that_has_already_passed() + { + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), DateTimeOffset.UtcNow.AddMinutes(-5), token: Ct); + + added.Should().BeFalse(); + await _innerCache.DidNotReceive().TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + _memoryCache.DidNotReceive().CreateEntry(_cacheKey); + } + + [Fact] + public async Task TryAdd_claims_nothing_for_a_zero_expiration() + { + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), TimeSpan.Zero, token: Ct); + + added.Should().BeFalse(); + await _innerCache.DidNotReceive().TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task A_broadcast_that_reports_not_published_still_stands_and_is_logged() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()).Returns(_ => false); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeTrue(); + _logger.Received().Log( + LogLevel.Warning, + Arg.Any(), + Arg.Any(), + Arg.Any(), + Arg.Any>()); + } + + [Fact] + public async Task A_failed_broadcast_still_populates_the_local_tier() + { + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("broadcast down")); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeTrue(); + _memoryCache.Received(1).CreateEntry(_cacheKey); + } + + [Fact] + public async Task TryAdd_fails_closed_when_the_inner_cache_cannot_claim_the_key() + { + _options.UseLocalOnlyWhenDisconnected = true; + _options.ConnectionMonitorEnabled = true; + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(false); + _sut = null; + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeFalse("a local-only claim would be granted to every node independently"); + _memoryCache.DidNotReceive().CreateEntry(_cacheKey); + } + + [Fact] + public async Task A_disconnected_broadcast_transport_does_not_stop_a_healthy_L2_from_arbitrating() + { + _options.UseLocalOnlyWhenDisconnected = true; + _options.ConnectionMonitorEnabled = true; + _topicProvider.IsConnected.Returns(false); + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _sut = null; + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeTrue("the aggregate connection state covers the topic too, and broadcast is best-effort after a win"); + await _innerCache.Received(1).TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_surfaces_a_cancellation_raised_by_the_inner_cache() + { + using var cts = new CancellationTokenSource(); +#pragma warning disable CA2012 + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns>(_ => + { + cts.Cancel(); + throw new OperationCanceledException(cts.Token); + }); +#pragma warning restore CA2012 + + var act = async () => await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: cts.Token); + + await act.Should().ThrowAsync( + "reporting false would say someone else owns the key, which InMemoryRedis must not claim any more than Redis does"); + await _innerCache.Received(1).TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_of_a_null_value_never_deletes_and_never_reaches_the_inner_cache() + { + _options.CacheNullValues = false; + _sut = null; + + var added = await Sut.TryAddAsync(_cacheKey, default(string), policy: null, token: Ct); + + added.Should().BeFalse(); + _memoryCache.DidNotReceive().Remove(_cacheKey); + await _innerCache.DidNotReceive().RemoveAsync(_cacheKey, Arg.Any()); + await _innerCache.DidNotReceive().TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_of_a_null_value_reaches_the_inner_cache_when_CacheNullValues_is_on() + { + _options.CacheNullValues = true; + _sut = null; + _innerCache.TryAddAsync(_cacheKey, default, Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()).Returns(_ => true); + + var added = await Sut.TryAddAsync(_cacheKey, default(string), policy: null, token: Ct); + + added.Should().BeTrue(); + await _innerCache.Received(1).TryAddAsync(_cacheKey, default, Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_forwards_the_caller_expiration_to_the_inner_cache() + { + var ttl = TimeSpan.FromMinutes(7); + _innerCache.TryAddAsync(_cacheKey, Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + _topic.PublishAsync(Arg.Any(), Arg.Any()).Returns(_ => true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), ttl, token: Ct); + + await _innerCache.Received(1).TryAddAsync( + _cacheKey, + Arg.Any(), + Arg.Is(e => e.HasValue && e.Value > DateTimeOffset.UtcNow), + Arg.Any(), + Arg.Any()); + } + + [Fact] + public async Task The_L2_answer_is_taken_as_given_whatever_the_L2_is() + { + var inner = CreateInMemorySut(); + _fixture.Inject(inner); + using var sut = _fixture.Create(); + + var added = await sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: Ct); + + added.Should().BeTrue("the L2 granted the claim, and the outer cache does not second-guess it"); + } + + private static MultilayerCache CreateInMemorySut() + { + var options = new InMemoryCacheOptions(); + var cacheOptions = new CacheOptions { AppShortName = "test" }; + return new MultilayerCache( + KnownCacheProviderNames.InMemory, + NullCache.Instance, + new MemoryCacheFactory(null, NullLoggerFactory.Instance), + NullChangeTokenFactory.Instance, + NullTopicFactory.Instance, + NullCacheEventFactory.Instance, + NullTelemetryProvider.Instance, + options, + options, + cacheOptions, + localLock: new AsyncKeyedLocalLock(Options.Create(cacheOptions)), + distributedLock: NullDistributedLock.Instance, + policyFactory: NullCachePolicyFactory.Instance, + logger: NullLogger.Instance); + } + + [Fact] + public async Task TryAdd_rejects_a_null_key() + { + string? nullKey = null; + var act = async () => await Sut.TryAddAsync(nullKey!, _fixture.Create(), policy: null, token: Ct); + + await act.Should().ThrowAsync(); + } + + public ValueTask InitializeAsync() + { + _cacheKey = _fixture.Create(); + _topicKey = _fixture.Create(); + + _changeTokenFactory = _fixture.Freeze(); + _memoryCache = _fixture.Freeze(); + _innerCache = _fixture.Freeze(); + _logger = _fixture.Freeze(); + _logger.IsEnabled(Arg.Any()).Returns(true); + _options = new() + { + DefaultExpiration = TimeSpan.FromMinutes(10), + EntryFactory = new TestCacheEntryFactory(), + }; + + var cacheKeyStrategy = _fixture.Create(); + var topicKeyStrategy = _fixture.Create(); + cacheKeyStrategy.GetCacheKey(_cacheKey).Returns(_cacheKey); + topicKeyStrategy.GetTopicKey().Returns(_topicKey); + _topicFactory = _fixture.Freeze(); + _topicProvider = _fixture.Freeze(); + _topic = _fixture.Freeze>(); + _topicFactory.Get(Arg.Any()).Returns(_topicProvider); + _topicProvider.Create(_topicKey).Returns(_topic); + _topicProvider.Create(Arg.Any()).Returns(_topic); + _memoryCacheFactory = _fixture.Freeze(); + _memoryCacheFactory.Get(Arg.Any()).Returns(_ => _memoryCache); + _fixture.Inject(_options); + _fixture.Inject(new AsyncKeyedLocalLock(Options.Create(new CacheOptions { AppShortName = "test" }))); + _memoryCache.TryGetValue(Arg.Any(), out Arg.Any()).Returns(false); + _fixture.Inject(_options); + _fixture.Inject>(new CacheClearEventFormatterProxy()); + var cacheEventFactory = _fixture.Freeze(); + cacheEventFactory.Create(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(c => new TestCacheEvent + { + Id = c.ArgAt(3), + Data = c.Arg(), + Type = c.ArgAt(1), + }); + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() + { + _sut?.Dispose(); + GC.SuppressFinalize(this); + return ValueTask.CompletedTask; + } + + public interface ITopicProviderWithConnectionState : ITopicProvider, IConnectionState + { + } +} + +/// +/// The memory-only provider: a real over , so +/// the local tier is the storage and the arbiter. Exclusion here is in-process only, which +/// is the honest ceiling for a cache with no shared store — these tests pin that it is at least +/// correct within the process. +/// +public class InMemoryCacheTryAddTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static MultilayerCache CreateSut( + InMemoryCacheOptions? options = null, + ILocalLock? localLock = null) + { + options ??= new InMemoryCacheOptions(); + var cacheOptions = new CacheOptions { AppShortName = "test" }; + return new MultilayerCache( + KnownCacheProviderNames.InMemory, + NullCache.Instance, + new MemoryCacheFactory(null, NullLoggerFactory.Instance), + NullChangeTokenFactory.Instance, + NullTopicFactory.Instance, + NullCacheEventFactory.Instance, + NullTelemetryProvider.Instance, + options, + options, + cacheOptions, + localLock: localLock ?? new AsyncKeyedLocalLock(Options.Create(cacheOptions)), + distributedLock: NullDistributedLock.Instance, + policyFactory: NullCachePolicyFactory.Instance, + logger: NullLogger.Instance); + } + + [Fact] + public async Task First_caller_adds_and_the_second_loses() + { + using var sut = CreateSut(); + + (await sut.TryAddAsync("k", "first", policy: null, token: Ct)).Should().BeTrue(); + (await sut.TryAddAsync("k", "second", policy: null, token: Ct)).Should().BeFalse(); + } + + [Fact] + public async Task A_lost_add_does_not_overwrite_the_winner_value() + { + using var sut = CreateSut(); + + await sut.TryAddAsync("k", "first", policy: null, token: Ct); + await sut.TryAddAsync("k", "second", policy: null, token: Ct); + + (await sut.GetAsync("k", policy: null, token: Ct)).Should().Be("first"); + } + + [Fact] + public async Task The_key_is_claimable_again_once_removed() + { + using var sut = CreateSut(); + + await sut.TryAddAsync("k", "first", policy: null, token: Ct); + await sut.RemoveAsync("k", Ct); + + (await sut.TryAddAsync("k", "second", policy: null, token: Ct)).Should().BeTrue(); + (await sut.GetAsync("k", policy: null, token: Ct)).Should().Be("second"); + } + + [Fact] + public async Task TryAdd_does_not_claim_a_key_an_unconditional_set_already_wrote() + { + using var sut = CreateSut(); + + await sut.SetAsync("k", "written", policy: null, token: Ct); + + (await sut.TryAddAsync("k", "claimed", policy: null, token: Ct)).Should().BeFalse(); + } + + [Fact] + public async Task Exactly_one_of_many_concurrent_callers_wins() + { + using var sut = CreateSut(); + const int callers = 32; + + var results = await Task.WhenAll(Enumerable.Range(0, callers).Select(i => + Task.Run(async () => await sut.TryAddAsync("k", $"caller-{i}", policy: null, token: Ct), Ct))); + + results.Count(won => won).Should().Be(1, "the local lock is what makes probe-then-write atomic"); + } + + [Fact] + public async Task The_local_probe_narrows_an_L2_that_grants_everyone_a_win() + { + using var sut = CreateSut(); + + (await sut.TryAddAsync("k", "first", policy: null, token: Ct)).Should().BeTrue(); + (await sut.TryAddAsync("k", "second", policy: null, token: Ct)).Should().BeFalse(); + } + + [Fact] + public async Task Exactly_one_caller_still_wins_with_the_local_lock_disabled() + { + using var sut = CreateSut(new InMemoryCacheOptions { LocalLockEnabled = false }); + const int callers = 32; + + var results = await Task.WhenAll(Enumerable.Range(0, callers).Select(i => + Task.Run(async () => await sut.TryAddAsync("k", $"caller-{i}", policy: null, token: Ct), Ct))); + + results.Count(won => won).Should().Be(1, "an unserialized probe-then-write would hand the same win to several callers"); + } + + [Fact] + public async Task A_caller_that_cannot_take_the_local_lock_is_told_it_lost() + { + var options = new InMemoryCacheOptions { LocalLockTimeout = TimeSpan.FromMilliseconds(50) }; + using var sut = CreateSut(options, localLock: new NeverGrantingLocalLock()); + + var added = await sut.TryAddAsync("k", "first", policy: null, token: Ct); + + added.Should().BeFalse(); + (await sut.GetAsync("k", policy: null, token: Ct)).Should().BeNull("a loss must not write anything either"); + } + + [Fact] + public async Task A_size_limited_memory_cache_that_drops_the_entry_still_reports_the_win() + { + using var sut = CreateSut(new InMemoryCacheOptions { SizeLimit = 1, SizeProvider = new OversizedEntryProvider() }); + + (await sut.TryAddAsync("k", "first", policy: null, token: Ct)).Should().BeTrue(); + (await sut.TryAddAsync("k", "second", policy: null, token: Ct)).Should().BeTrue(); + } + + private sealed class OversizedEntryProvider : ICacheEntrySizeProvider + { + public long GetSize(ICacheEntry entry) => long.MaxValue; + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public async Task A_non_positive_local_retention_claims_nothing(int minutes) + { + using var sut = CreateSut(); + var policy = new CachePolicy { LocalExpiration = TimeSpan.FromMinutes(minutes) }; + + (await sut.TryAddAsync("k", "first", policy, token: Ct)).Should().BeFalse(); + (await sut.TryAddAsync("k", "second", policy, token: Ct)).Should().BeFalse(); + (await sut.GetAsync("k", policy: null, token: Ct)).Should().BeNull(); + } + + [Fact] + public async Task An_expiration_that_has_already_passed_claims_nothing() + { + using var sut = CreateSut(); + var past = DateTimeOffset.UtcNow.AddMinutes(-5); + + (await sut.TryAddAsync("k", "first", past, token: Ct)).Should().BeFalse(); + (await sut.TryAddAsync("k", "second", past, token: Ct)).Should().BeFalse(); + (await sut.GetAsync("k", policy: null, token: Ct)).Should().BeNull(); + } + + /// + /// Stands in for a local lock held by someone else for longer than the acquire budget: the wait is + /// abandoned by the linked timeout, which is the only way AcquireLocalLockAsync answers null. + /// + private sealed class NeverGrantingLocalLock : ILocalLock + { + public async ValueTask AcquireAsync(string key, CancellationToken token) + { + await Task.Delay(Timeout.Infinite, token).ConfigureAwait(false); + throw new InvalidOperationException("unreachable: the delay above only ever cancels"); + } + } +} diff --git a/tests/UiPath.Caching.Tests/NullCacheConditionalAddTests.cs b/tests/UiPath.Caching.Tests/NullCacheConditionalAddTests.cs new file mode 100644 index 00000000..a54aaba5 --- /dev/null +++ b/tests/UiPath.Caching.Tests/NullCacheConditionalAddTests.cs @@ -0,0 +1,42 @@ +using UiPath.Caching; + +namespace UiPath.Caching.Tests; + +public class NullCacheConditionalAddTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + [Fact] + public async Task TryAdd_reports_added_for_every_caller() + { + var sut = NullCache.Instance; + + (await sut.TryAddAsync("k", "first", policy: null, token: Ct)).Should().BeTrue(); + (await sut.TryAddAsync("k", "second", policy: null, token: Ct)).Should().BeTrue(); + } + + [Theory] + [InlineData(null)] + [InlineData(5)] + public async Task TryAdd_reports_added_whatever_the_expiration(int? minutes) + { + TimeSpan? ttl = minutes is { } m ? TimeSpan.FromMinutes(m) : null; + + (await NullCache.Instance.TryAddAsync("k", "v", ttl, token: Ct)).Should().BeTrue(); + (await NullCache.Instance.TryAddAsync("k", "v", ttl.HasValue ? DateTimeOffset.UtcNow.Add(ttl.Value) : null, token: Ct)).Should().BeTrue(); + } + + [Fact] + public async Task SetAsync_still_reports_success() + { + (await NullCache.Instance.SetAsync("k", "v", policy: null, token: Ct)).Should().BeTrue(); + } + + [Fact] + public void An_uncacheable_type_still_throws() + { + var act = () => NullCache.Instance.TryAddAsync("k", 1, policy: null, token: Ct); + + act.Should().Throw(); + } +} diff --git a/tests/UiPath.Caching.Tests/Redis/RedisCacheTryAddTests.cs b/tests/UiPath.Caching.Tests/Redis/RedisCacheTryAddTests.cs new file mode 100644 index 00000000..e7777c51 --- /dev/null +++ b/tests/UiPath.Caching.Tests/Redis/RedisCacheTryAddTests.cs @@ -0,0 +1,315 @@ +using Microsoft.Extensions.Internal; +using NSubstitute.ExceptionExtensions; +using StackExchange.Redis; +using UiPath.Caching; +using UiPath.Caching.Policies; +using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Telemetry; + +namespace UiPath.Caching.Tests.Redis; + +/// +/// Conditional add (TryAddAsync) on the Redis tier. The contract under test is narrow: one +/// SET … NX command, the Redis reply is the answer, and every non-win — lost race, +/// disconnected, thrown, unrepresentable value — reports false without a second round-trip. +/// +public class RedisCacheTryAddTests(ITestContextAccessor testContextAccessor) : IAsyncLifetime +{ + private readonly IFixture _fixture = AutoFixtureCreator.NSubstitute(); + private ISystemClock _clock = default!; + private RedisCacheOptions _cacheOptions = default!; + private IDatabase _database = default!; + private SystemJsonSerializerProxy _serializer = default!; + private readonly DateTimeOffset _now = DateTimeOffset.UtcNow; + private CacheKey _cacheKey = default!; + private RedisKey _redisKey = default!; + private IRedisConnector _connector = default!; + private IResiliencePipelineProvider _pipelineProvider = default!; + private bool _isConnected = true; + private readonly RecordingTelemetryProvider _telemetry = new(); + private RedisCache? _sut; + + private RedisCache Sut => _sut ??= _fixture.Create(); + + [Fact] + public async Task TryAdd_issues_a_single_NX_write_and_reports_the_win() + { + var value = _fixture.Create(); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + var added = await Sut.TryAddAsync(_cacheKey, value, policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeTrue(); + await _database.Received(1).StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster); + await _database.DidNotReceive().StringGetAsync(Arg.Any(), Arg.Any()); + await _database.DidNotReceive().KeyExistsAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_never_writes_unconditionally() + { + _database.StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), When.Always, Arg.Any()); + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), When.Exists, Arg.Any()); + } + + [Fact] + public async Task TryAdd_reports_not_added_when_the_key_already_exists() + { + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(false); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse(); + } + + [Fact] + public async Task TryAdd_writes_a_payload_a_reader_can_deserialize() + { + var value = _fixture.Create(); + RedisValue captured = default; + _database.StringSetAsync(_redisKey, Arg.Do(v => captured = v), Arg.Any(), When.NotExists, Arg.Any()) + .Returns(true); + + await Sut.TryAddAsync(_cacheKey, value, policy: null, token: testContextAccessor.Current.CancellationToken); + + _serializer.Deserialize(captured).Should().Be(value); + } + + [Fact] + public async Task TryAdd_fails_closed_when_disconnected() + { + _isConnected = false; + _database.StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse("a win that was never written would let a second caller win the same key"); + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_fails_closed_when_the_write_throws() + { + _database.StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new RedisException("test")); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse(); + } + + [Fact] + public async Task TryAdd_applies_the_caller_expiration_in_the_same_command() + { + var ttl = TimeSpan.FromMinutes(3); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), ttl, token: testContextAccessor.Current.CancellationToken); + + await _database.Received(1).StringSetAsync(_redisKey, Arg.Any(), ttl, When.NotExists, CommandFlags.DemandMaster); + await _database.DidNotReceive().KeyExpireAsync(Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_falls_back_to_the_policy_expiration() + { + var policyTtl = TimeSpan.FromMinutes(7); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: new CachePolicy { DistributedExpiration = policyTtl }, token: testContextAccessor.Current.CancellationToken); + + await _database.Received(1).StringSetAsync(_redisKey, Arg.Any(), policyTtl, When.NotExists, CommandFlags.DemandMaster); + } + + [Fact] + public async Task TryAdd_accepts_an_absolute_expiration() + { + var absolute = _now.AddMinutes(4); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), absolute, token: testContextAccessor.Current.CancellationToken); + + await _database.Received(1).StringSetAsync(_redisKey, Arg.Any(), TimeSpan.FromMinutes(4), When.NotExists, CommandFlags.DemandMaster); + } + + [Fact] + public async Task TryAdd_of_a_default_value_never_deletes_the_key() + { + _cacheOptions.CacheNullValues = false; + + var added = await Sut.TryAddAsync(_cacheKey, null, TimeSpan.FromMinutes(1), token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse("with no cached-null sentinel available there is nothing to claim the key with"); + await _database.DidNotReceive().KeyDeleteAsync(Arg.Any(), Arg.Any()); + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_of_a_default_value_claims_the_key_with_the_sentinel_when_CacheNullValues_is_on() + { + _cacheOptions.CacheNullValues = true; + _sut = null; + RedisValue captured = _fixture.Create(); + _database.StringSetAsync(_redisKey, Arg.Do(v => captured = v), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + var added = await Sut.TryAddAsync(_cacheKey, null, TimeSpan.FromMinutes(1), token: testContextAccessor.Current.CancellationToken); + + added.Should().BeTrue(); + captured.Length().Should().Be(0, "the empty string is the cached-null sentinel on the wire"); + } + + [Theory] + [InlineData(true, "Caching.Stats.Hits.Redis.TryAddAsync.String")] + [InlineData(false, "Caching.Stats.Misses.Redis.TryAddAsync.String")] + public async Task TryAdd_reports_the_outcome_under_its_own_metric_scope(bool redisAdded, string expectedMetric) + { + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(redisAdded); + + await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + _telemetry.Metrics.Should().ContainSingle(m => m.Name == expectedMetric); + } + + [Fact] + public async Task TryAdd_honors_a_cancelled_token_before_touching_redis() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var act = async () => await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: cts.Token); + + await act.Should().ThrowAsync(); + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task TryAdd_runs_the_NX_write_through_the_shared_write_pipeline() + { + var write = new CountingResiliencePipeline(); + _pipelineProvider.Get(ResiliencePipelineNames.Write).Returns(write); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns(true); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeTrue(); + write.Executions.Should().Be(1); + } + + [Fact] + public async Task TryAdd_surfaces_a_cancellation_raised_while_the_write_is_in_flight() + { + using var cts = new CancellationTokenSource(); + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .Returns>(_ => + { + cts.Cancel(); + throw new OperationCanceledException(cts.Token); + }); + + var act = async () => await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: cts.Token); + + await act.Should().ThrowAsync(); + } + + [Fact] + public async Task A_redis_failure_that_is_not_a_cancellation_still_reports_not_added() + { + _database.StringSetAsync(_redisKey, Arg.Any(), Arg.Any(), When.NotExists, CommandFlags.DemandMaster) + .ThrowsAsync(new InvalidOperationException("redis down")); + + var added = await Sut.TryAddAsync(_cacheKey, _fixture.Create(), policy: null, token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse(); + } + + [Theory] + [InlineData(0)] + [InlineData(-5)] + public async Task TryAdd_claims_nothing_for_a_non_positive_expiration(int minutes) + { + var added = await Sut.TryAddAsync( + _cacheKey, + _fixture.Create(), + TimeSpan.FromMinutes(minutes), + token: testContextAccessor.Current.CancellationToken); + + added.Should().BeFalse(); + await _database.DidNotReceive().StringSetAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); + } + + public ValueTask InitializeAsync() + { + const string prefix = "test"; + _cacheKey = _fixture.Create(); + _redisKey = string.Join(':', prefix, RedisTypePrefixes.String, _cacheKey).ToLowerInvariant(); + _clock = _fixture.Freeze(); + _clock.UtcNow.Returns(_ => _now); + _pipelineProvider = _fixture.Freeze(); + var noOpExecutor = new EmptyResiliencePipeline(); + _pipelineProvider.Get(ResiliencePipelineNames.Read).Returns(noOpExecutor); + _pipelineProvider.Get(ResiliencePipelineNames.Write).Returns(noOpExecutor); + var cacheKeyStrategy = _fixture.Create(); + cacheKeyStrategy.GetCacheKey(_cacheKey).Returns(_cacheKey); + var redisKeyStrategyFactory = _fixture.Create(); + var redisKeyStrategy = _fixture.Create(); + redisKeyStrategy.GetRedisKey(_cacheKey).Returns(_redisKey); + redisKeyStrategyFactory.Create(Arg.Any(), Arg.Any()).Returns(redisKeyStrategy); + _cacheOptions = new RedisCacheOptions + { + Clock = _clock, + CacheKeyStrategy = cacheKeyStrategy, + RedisKeyStrategyFactory = redisKeyStrategyFactory, + }; + + _database = _fixture.Freeze(); + _serializer = new SystemJsonSerializerProxy(); + _fixture.Inject>(_serializer); + _fixture.Inject(Options.Create(_cacheOptions)); + _fixture.Inject(_cacheOptions); + _fixture.Inject(_telemetry); + _fixture.Inject(new CacheOptions { AppShortName = "test", ConnectionMonitorEnabled = true }); + _connector = _fixture.Freeze(); + _connector.Database.Returns(_ => _database); + _connector.Version.Returns(_ => new Version(6, 0)); + _connector.IsConnected.Returns(_ => _isConnected); + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() + { + _sut?.Dispose(); + GC.SuppressFinalize(this); + return ValueTask.CompletedTask; + } + + /// + /// Executes the callback once and counts it, so a test can say which named pipeline a command was + /// routed through without reaching into Polly. + /// + private sealed class CountingResiliencePipeline : IResiliencePipeline + { + private int _executions; + + public int Executions => Volatile.Read(ref _executions); + + public ValueTask ExecuteAsync(Func> callback, TResult defaultValue, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _executions); + return callback(cancellationToken); + } + } +}