diff --git a/CHANGELOG.md b/CHANGELOG.md index 69b72b9..13b03dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,23 @@ its code behaves exactly as it did. ### Added +- **Optimistic concurrency on `IStateProvider`.** A state write can now be made conditional on the + state not having changed since it was read, closing the lost update that has been documented on + `CosmosDbStateProvider` since the beginning: a setting changed from the dashboard could be + overwritten by a check that had read the state first. `GetStateEntryAsync` returns the state with + the version it was read at, and `TrySetStateAsync` writes only if that version is still current, + reporting a refused write rather than throwing -- under contention a conflict is expected, not + exceptional. `UpdateStateAsync` is the read-modify-write retry loop over the pair, and every + setting on `PulseChecker` goes through it -- as does a check storing its own result, which reads + and writes the whole state and so used to put every other field back to what it was when the + check started. Within one process a semaphore hid that; across two replicas sharing one store + nothing did, and that is the direction the dashboard actually loses to. + + This was expected to need a major release and did not. The new members are defaulted interface + methods: a provider written against the old two-method interface still compiles, still works, and + reports `SupportsOptimisticConcurrency == false` rather than pretending. CosmosDB uses ETags, + the relational providers a version column added to existing tables on startup, and the in-memory + provider a compare-and-swap. - **Schedules.** `PulseSchedule` says either "every this long" or "on this cron expression", and sits alongside `PulseInterval` rather than replacing it. The enum stopped at five minutes, which is short of what a certificate-expiry or disk-space check wants. Cron is standard Unix syntax and @@ -77,8 +94,6 @@ its code behaves exactly as it did. - `StateChanged` is raised on **every** check rather than only when state changes, because `PulseCheckerState` equality includes the last execution time. Anything reacting to it should compare the health itself, which the alerting and uptime packages do. -- `IStateProvider` still has no concurrency token, so two writers to one checker's state remain - last-write-wins, as documented on `CosmosDbStateProvider`. Adding one is a breaking change. ## [3.1.4] - 2026-07-19 diff --git a/README.md b/README.md index 534c350..368d396 100644 --- a/README.md +++ b/README.md @@ -818,11 +818,9 @@ Upgrading from v1.x? See the [v1 to v2 migration guide](https://github.com/ivanv Shipped since 3.1.4: alerting on transitions, OpenTelemetry metrics and traces, arbitrary intervals and cron, PostgreSQL / SQL Server / SQLite state providers, Hangfire / Coravel / Temporal -scheduling, ready-made checkers, uptime reporting, and leader election. What is left: +scheduling, ready-made checkers, uptime reporting, leader election, and optimistic concurrency on +`IStateProvider`. What is left: -- **Concurrency tokens on `IStateProvider`** -- state writes are last-write-wins, so a setting - changed from the dashboard can be overwritten by a check that read the state first. Resolving it - needs the interface to carry a version, which is a breaking change and a major release. - **`StateChanged` fires on every check** rather than only when state changes, because state equality includes the last execution time. Anything reacting to it should compare the health itself, which the alerting and uptime packages do. diff --git a/src/Healthie.Abstractions/PulseChecker.cs b/src/Healthie.Abstractions/PulseChecker.cs index d6288c4..1eb8826 100644 --- a/src/Healthie.Abstractions/PulseChecker.cs +++ b/src/Healthie.Abstractions/PulseChecker.cs @@ -24,6 +24,13 @@ namespace Healthie.Abstractions; public abstract class PulseChecker : IPulseChecker, IDisposable { private static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10); + + /// How many times a setting change is reapplied before giving up. + /// + /// Contention is one check loop against one person editing, so a conflict is rare and a second + /// one rarer. A larger number would only make a genuine livelock take longer to report. + /// + private const int MaxUpdateAttempts = 5; private readonly IStateProvider _stateProvider; private readonly ILogger? _logger; private readonly SemaphoreSlim _semaphore = new(1, 1); @@ -241,26 +248,119 @@ private async Task AcquireAsync(CancellationToken cancellationToken) } } + /// + /// Applies a change to this checker's stored state, and reapplies it if something else wrote + /// first. + /// + /// + /// + /// This closes the gap every setting change used to have. Reading the state, changing it and + /// writing it back is three steps, and a check finishing in between wrote its result over the + /// change -- or had its own result written over. Against a provider that versions its writes, + /// the write now only lands if nothing moved, and the change is reapplied to the newer state if + /// something did. + /// + /// + /// The change runs once per attempt, against freshly read state each time, so it must not + /// depend on having seen the previous value. + /// + /// + /// StateChanged is raised once, after the write that landed, comparing against the state + /// that write was made from rather than whatever was read on the first attempt. + /// + /// + /// true if anything was written; false if the change was a no-op. + private async Task UpdateStateAsync(Action apply, CancellationToken cancellationToken) + { + await AcquireAsync(cancellationToken).ConfigureAwait(false); + + try + { + var (oldState, newState, changed) = await ApplyAsync(apply, cancellationToken).ConfigureAwait(false); + + if (changed) + { + StateChanged?.Invoke(this, new PulseCheckerStateChangedEventArgs(oldState, newState)); + } + + return changed; + } + finally + { + _semaphore.Release(); + } + } + + /// + /// The read-modify-write loop itself, without the lock and without the event. + /// + /// + /// Separate from because needs the + /// same conditional write but already holds the semaphore and has its own telemetry to record + /// between the write and the event. A checker's own result is state like any other: read, + /// changed and written back over three steps, and a setting change landing in that gap used to + /// be reverted by it -- the direction the semaphore hides within one process and cannot touch + /// across two. + /// + /// The state before and after, and whether anything was written. + private async Task<(PulseCheckerState OldState, PulseCheckerState NewState, bool Changed)> ApplyAsync( + Action apply, + CancellationToken cancellationToken) + { + for (var attempt = 1; ; attempt++) + { + var entry = await _stateProvider + .GetStateEntryAsync(Name, cancellationToken) + .ConfigureAwait(false); + + var state = entry?.Value ?? CreateInitialState(); + + // History is a mutable list, so `with` alone would hand out a snapshot sharing it. + var oldState = state with { History = [.. state.History] }; + + apply(state); + + // Nothing to write, and nothing to tell anyone about. + if (Equals(oldState, state)) + { + return (oldState, state, false); + } + + // Three cases, and collapsing any two of them is a bug. Nothing stored -> ask for a + // create that loses to whoever creates first. Stored and versioned -> the version. + // Stored but unversioned (a row written before the provider could version) -> null, an + // unconditional write, because there is nothing to compare and demanding a version that + // does not exist would refuse every write for ever. + var version = _stateProvider.SupportsOptimisticConcurrency + ? entry is null ? IStateProvider.AbsentVersion : entry.Version + : null; + + if (await _stateProvider + .TrySetStateAsync(Name, state, version, cancellationToken) + .ConfigureAwait(false)) + { + return (oldState, state, true); + } + + if (attempt >= MaxUpdateAttempts) + { + throw new InvalidOperationException( + $"Could not update the state of pulse checker '{Name}' after {MaxUpdateAttempts} " + + "attempts: another writer changed it each time."); + } + } + } + /// public async Task SetIntervalAsync(PulseInterval interval, CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.Interval == interval) - return; - state.Interval = interval; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.Interval = interval, cancellationToken).ConfigureAwait(false); } /// public async Task SetUnhealthyThresholdAsync(uint threshold, CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.UnhealthyThreshold == threshold) - { - return; - } - state.UnhealthyThreshold = threshold; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.UnhealthyThreshold = threshold, cancellationToken).ConfigureAwait(false); } /// @@ -270,14 +370,7 @@ public async Task SetTagsAsync(IReadOnlyList tags, CancellationToken can var normalized = NormalizeTags(tags); - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.Tags.SequenceEqual(normalized)) - { - return; - } - - state.Tags = normalized; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.Tags = [.. normalized], cancellationToken).ConfigureAwait(false); } /// @@ -285,41 +378,25 @@ public async Task SetGroupAsync(string? group, CancellationToken cancellationTok { var normalized = NormalizeGroup(group); - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.Group == normalized) - { - return; - } - - state.Group = normalized; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.Group = normalized, cancellationToken).ConfigureAwait(false); } /// public async Task SetPinnedAsync(bool pinned, CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.IsPinned == pinned) - { - return; - } - - state.IsPinned = pinned; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.IsPinned = pinned, cancellationToken).ConfigureAwait(false); } /// public async Task ResetAsync(CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - - state.ConsecutiveFailureCount = 0; - - state.LastResult = new PulseCheckerResult( - PulseCheckerHealth.Healthy, - string.Empty); - - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync( + state => + { + state.ConsecutiveFailureCount = 0; + state.LastResult = new PulseCheckerResult(PulseCheckerHealth.Healthy, string.Empty); + }, + cancellationToken).ConfigureAwait(false); } /// @@ -342,11 +419,7 @@ public async Task ClearHistoryAsync(CancellationToken cancellationToken = defaul /// public async Task SetHistoryEnabledAsync(bool enabled, CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.IsHistoryEnabled == enabled) - return; - state.IsHistoryEnabled = enabled; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); + await UpdateStateAsync(state => state.IsHistoryEnabled = enabled, cancellationToken).ConfigureAwait(false); } /// @@ -398,18 +471,9 @@ public async Task TriggerAsync(CancellationToken cancellationToken = default) var result = await RunCheckAsync(cancellationToken).ConfigureAwait(false); var elapsed = Stopwatch.GetElapsedTime(startedAt); - PulseCheckerState state = await _stateProvider.GetStateAsync(Name, cancellationToken).ConfigureAwait(false) - ?? CreateInitialState(); - - // History is a mutable list, so `with` alone would hand out a snapshot that still - // shares it and would appear to change as this trigger appends to it. - var oldState = state with { History = [.. state.History] }; - - RecordResult(state, result, executedAt); - - await _stateProvider.SetStateAsync(Name, state, cancellationToken).ConfigureAwait(false); - - var changed = !Equals(oldState, state); + var (oldState, state, changed) = await ApplyAsync( + current => RecordResult(current, result, executedAt), + cancellationToken).ConfigureAwait(false); // Recorded after the write, so a check whose state could not be stored is not counted // as one that ran -- the same reason a storage failure is not a health result. @@ -561,23 +625,13 @@ private static PulseCheckerResult ApplyThreshold(PulseCheckerResult result, Puls /// public async Task StopAsync(CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (!state.IsActive) - return false; - state.IsActive = false; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); - return true; + return await UpdateStateAsync(state => state.IsActive = false, cancellationToken).ConfigureAwait(false); } /// public async Task StartAsync(CancellationToken cancellationToken = default) { - PulseCheckerState state = await GetStateAsync(cancellationToken).ConfigureAwait(false); - if (state.IsActive) - return false; - state.IsActive = true; - await SetStateAsync(state, cancellationToken).ConfigureAwait(false); - return true; + return await UpdateStateAsync(state => state.IsActive = true, cancellationToken).ConfigureAwait(false); } /// diff --git a/src/Healthie.Abstractions/StateProviding/IStateProvider.cs b/src/Healthie.Abstractions/StateProviding/IStateProvider.cs index 12e0b33..9cd5451 100644 --- a/src/Healthie.Abstractions/StateProviding/IStateProvider.cs +++ b/src/Healthie.Abstractions/StateProviding/IStateProvider.cs @@ -5,6 +5,105 @@ namespace Healthie.Abstractions.StateProviding; /// public interface IStateProvider { + /// + /// The version to pass when the write should only land if nothing is stored yet. + /// + /// + /// + /// Without it, the very first write for a checker has no version to compare against and would + /// go through unconditionally -- so two writers both finding nothing, both creating, and both + /// writing would lose one of the two changes. The same lost update the version exists to + /// prevent, at the one moment there is nothing to compare. + /// + /// + /// * is the value HTTP gives this meaning in an If-None-Match header, and the one + /// CosmosDB takes for the same purpose. No provider here generates a version that could collide + /// with it. + /// + /// + const string AbsentVersion = "*"; + + /// + /// Whether this provider can make a write conditional on the state not having changed. + /// + /// + /// Feature detection, so a caller can choose the conditional path rather than discovering by + /// exception that it is unavailable. A provider that returns true must honour the version + /// passed to ; one that returns false must refuse a + /// versioned write rather than ignore the version. + /// + bool SupportsOptimisticConcurrency => false; + + /// + /// Gets a pulse checker's state together with the version it was read at. + /// + /// The type of state to retrieve. + /// The unique name of the pulse checker. + /// A token to monitor for cancellation requests. + /// The stored state and its version, or null if nothing is stored. + /// + /// Defaulted to read without a version, so a provider written against the older interface keeps + /// working and simply reports that its reads cannot be used for a conditional write. + /// + async Task?> GetStateEntryAsync( + string name, + CancellationToken cancellationToken = default) + { + var state = await GetStateAsync(name, cancellationToken).ConfigureAwait(false); + + return state is null ? null : new StateEntry(state, Version: null); + } + + /// + /// Saves a pulse checker's state only if it has not changed since it was read. + /// + /// The type of state to save. + /// The unique name of the pulse checker. + /// The state to save. + /// + /// The version the state was read at, from . Pass null to + /// write unconditionally, which is what does. + /// + /// A token to monitor for cancellation requests. + /// + /// true if the state was written; false if something else had changed it since it + /// was read, and this write was refused. + /// + /// + /// A version was supplied and the provider cannot honour it. + /// + /// + /// + /// Returns a result rather than throwing, unlike Orleans and EF Core, which raise + /// InconsistentStateException and DbUpdateConcurrencyException. Under contention a + /// conflict is the expected outcome, not an exceptional one, and the caller's answer is always + /// the same: read again, reapply, retry. An exception on that path is noise in every log and a + /// cost on every retry. is that loop. + /// + /// + /// The default refuses a versioned write rather than performing an unconditional one. Ignoring + /// the version would lose exactly the update the version was passed to protect, and would do it + /// silently -- which is worse than not offering the operation. + /// + /// + async Task TrySetStateAsync( + string name, + TState state, + string? expectedVersion, + CancellationToken cancellationToken = default) + { + if (expectedVersion is not null) + { + throw new NotSupportedException( + $"{GetType().Name} cannot make a write conditional, so it cannot honour the version " + + $"'{expectedVersion}'. Check {nameof(SupportsOptimisticConcurrency)} before passing one."); + } + + await SetStateAsync(name, state, cancellationToken).ConfigureAwait(false); + + return true; + } + /// /// Gets the state of a specific pulse checker asynchronously. /// diff --git a/src/Healthie.Abstractions/StateProviding/StateEntry.cs b/src/Healthie.Abstractions/StateProviding/StateEntry.cs new file mode 100644 index 0000000..999d408 --- /dev/null +++ b/src/Healthie.Abstractions/StateProviding/StateEntry.cs @@ -0,0 +1,34 @@ +namespace Healthie.Abstractions.StateProviding; + +/// +/// A stored state together with the version it was read at. +/// +/// +/// +/// The version is what makes a later write conditional: hand it back to +/// and the write only lands if nothing else has +/// changed the state since. It is the same shape Orleans gives grain state, an ETag alongside the +/// value, and the same thing an If-Match header carries in the Azure SDKs. +/// +/// +/// Opaque on purpose. One provider's version is a row's transaction id, another's is a document +/// ETag, another's is a counter -- nothing outside the provider that produced it should read +/// meaning into it, or compare two of them for anything but equality. +/// +/// +/// The type of the stored state. +/// The stored state. +/// +/// The version it was read at, or null from a provider that does not version its writes. +/// +public sealed record StateEntry(TState Value, string? Version) +{ + /// + /// Whether this entry can be used for a conditional write. + /// + /// + /// False when the provider does not version. Writing such an entry back conditionally would be + /// asking for a guarantee nothing can give, so it is worth being able to tell. + /// + public bool IsVersioned => Version is not null; +} diff --git a/src/Healthie.Abstractions/StateProviding/StateProviderExtensions.cs b/src/Healthie.Abstractions/StateProviding/StateProviderExtensions.cs new file mode 100644 index 0000000..7a47d22 --- /dev/null +++ b/src/Healthie.Abstractions/StateProviding/StateProviderExtensions.cs @@ -0,0 +1,92 @@ +namespace Healthie.Abstractions.StateProviding; + +/// +/// Helpers over . +/// +public static class StateProviderExtensions +{ + /// How many times an update is reapplied before giving up. + /// + /// Contention here is two writers on one checker -- a scheduled check and someone editing it -- + /// so a conflict is rare and a second conflict on the retry rarer still. A handful of attempts + /// is generous; a number much larger would only turn a genuine livelock into a slow one. + /// + public const int DefaultMaxAttempts = 5; + + /// + /// Reads a state, applies a change to it, and writes it back only if nothing else changed it in + /// between -- reapplying the change against the newer state if something did. + /// + /// The type of state to update. + /// The provider holding the state. + /// The unique name of the pulse checker. + /// + /// Applies the change. Called with the current state, and again with a freshly read state on + /// each retry, so it must be safe to run more than once and must not depend on what it saw + /// before. + /// + /// Builds the state to store when nothing is stored yet. + /// How many times to try. Defaults to . + /// A token to monitor for cancellation requests. + /// The state as written. + /// + /// The update kept losing to another writer, times. + /// + /// + /// + /// This is the read-modify-write loop the Azure SDK guidance describes for conditional requests, + /// in the one place that has to get it right rather than in every caller. + /// + /// + /// Against a provider that does not version, this degrades to exactly what the library did + /// before -- read, change, write, last writer wins. That is not a silent downgrade: it is what + /// an unversioned store can offer, and + /// says which one you have. + /// + /// + public static async Task UpdateStateAsync( + this IStateProvider provider, + string name, + Action update, + Func create, + int maxAttempts = DefaultMaxAttempts, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(provider); + ArgumentException.ThrowIfNullOrWhiteSpace(name); + ArgumentNullException.ThrowIfNull(update); + ArgumentNullException.ThrowIfNull(create); + ArgumentOutOfRangeException.ThrowIfLessThan(maxAttempts, 1); + + for (var attempt = 1; ; attempt++) + { + var entry = await provider.GetStateEntryAsync(name, cancellationToken).ConfigureAwait(false); + var state = entry is null ? create() : entry.Value; + + update(state); + + // A provider that cannot version reports no version, and passing null asks for the + // unconditional write it is able to do rather than one it would have to refuse. + // Three cases, and collapsing any two of them is a bug. Nothing stored -> ask for a + // create that loses to whoever creates first. Stored and versioned -> the version. + // Stored but unversioned (a row written before the provider could version) -> null, an + // unconditional write, because there is nothing to compare and demanding a version that + // does not exist would refuse every write for ever. + var version = provider.SupportsOptimisticConcurrency + ? entry is null ? IStateProvider.AbsentVersion : entry.Version + : null; + + if (await provider.TrySetStateAsync(name, state, version, cancellationToken).ConfigureAwait(false)) + { + return state; + } + + if (attempt >= maxAttempts) + { + throw new InvalidOperationException( + $"Could not update the state of pulse checker '{name}' after {maxAttempts} attempts: " + + "another writer changed it each time."); + } + } + } +} diff --git a/src/Healthie.DependencyInjection/InMemoryStateProvider.cs b/src/Healthie.DependencyInjection/InMemoryStateProvider.cs index 6d758d5..d0bcb3f 100644 --- a/src/Healthie.DependencyInjection/InMemoryStateProvider.cs +++ b/src/Healthie.DependencyInjection/InMemoryStateProvider.cs @@ -24,7 +24,7 @@ public sealed class InMemoryStateProvider : IStateProvider { private static readonly JsonSerializerOptions SerializerOptions = new(JsonSerializerDefaults.General); - private readonly ConcurrentDictionary _states = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _states = new(StringComparer.Ordinal); /// public Task GetStateAsync(string name, CancellationToken cancellationToken = default) @@ -32,8 +32,8 @@ public sealed class InMemoryStateProvider : IStateProvider ArgumentException.ThrowIfNullOrWhiteSpace(name); cancellationToken.ThrowIfCancellationRequested(); - return _states.TryGetValue(name, out var json) - ? Task.FromResult(JsonSerializer.Deserialize(json, SerializerOptions)) + return _states.TryGetValue(name, out var entry) + ? Task.FromResult(JsonSerializer.Deserialize(entry.Json, SerializerOptions)) : Task.FromResult(default); } @@ -43,11 +43,72 @@ public Task SetStateAsync(string name, TState state, CancellationToken c ArgumentException.ThrowIfNullOrWhiteSpace(name); cancellationToken.ThrowIfCancellationRequested(); - _states[name] = JsonSerializer.Serialize(state, SerializerOptions); + _states[name] = (JsonSerializer.Serialize(state, SerializerOptions), NewVersion()); return Task.CompletedTask; } + /// + /// + /// Supported so that the provider every application starts on behaves like the durable ones. A + /// process with several replicas does not share this store, but one process still has a check + /// loop and a dashboard writing to the same state. + /// + public bool SupportsOptimisticConcurrency => true; + + private static string NewVersion() => Guid.NewGuid().ToString("N"); + + /// + public Task?> GetStateEntryAsync( + string name, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + cancellationToken.ThrowIfCancellationRequested(); + + if (!_states.TryGetValue(name, out var entry) + || JsonSerializer.Deserialize(entry.Json, SerializerOptions) is not { } state) + { + return Task.FromResult?>(null); + } + + return Task.FromResult?>(new StateEntry(state, entry.Version)); + } + + /// + public Task TrySetStateAsync( + string name, + TState state, + string? expectedVersion, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + cancellationToken.ThrowIfCancellationRequested(); + + var written = (JsonSerializer.Serialize(state, SerializerOptions), NewVersion()); + + if (expectedVersion is null) + { + _states[name] = written; + return Task.FromResult(true); + } + + // TryAdd succeeds only while nothing is stored, which is the create half of the guarantee. + if (expectedVersion == IStateProvider.AbsentVersion) + { + return Task.FromResult(_states.TryAdd(name, written)); + } + + // TryUpdate compares and swaps in one step, which is what makes this safe without a lock: + // a read-then-write here would have exactly the race it is meant to close. + if (_states.TryGetValue(name, out var current) && current.Version == expectedVersion) + { + return Task.FromResult(_states.TryUpdate(name, written, current)); + } + + return Task.FromResult(false); + } + /// public Task DeleteStateAsync(string name, CancellationToken cancellationToken = default) { @@ -74,8 +135,8 @@ public Task> GetStatesAsync( foreach (var name in names) { - if (_states.TryGetValue(name, out var json) - && JsonSerializer.Deserialize(json, SerializerOptions) is { } state) + if (_states.TryGetValue(name, out var entry) + && JsonSerializer.Deserialize(entry.Json, SerializerOptions) is { } state) { states[name] = state; } diff --git a/src/Healthie.LeaderElection/README.md b/src/Healthie.LeaderElection/README.md index da1d003..7dfe32e 100644 --- a/src/Healthie.LeaderElection/README.md +++ b/src/Healthie.LeaderElection/README.md @@ -13,7 +13,7 @@ Runs pulse checks on one replica at a time. Without it, **every replica runs every check**. Three replicas mean: - a database asked three times whether it is healthy, on every interval -- three sets of results racing to write the same state document, last write winning +- three sets of results racing to write the same state document, two of them wasted - one outage paging somebody three times None of that is visible from a dashboard, which is what makes it worth fixing before it matters. diff --git a/src/Healthie.StateProviding.CosmosDb/CosmosDbStateProvider.cs b/src/Healthie.StateProviding.CosmosDb/CosmosDbStateProvider.cs index 6f8a81a..a8a7275 100644 --- a/src/Healthie.StateProviding.CosmosDb/CosmosDbStateProvider.cs +++ b/src/Healthie.StateProviding.CosmosDb/CosmosDbStateProvider.cs @@ -10,16 +10,14 @@ namespace Healthie.StateProviding.CosmosDb; /// /// /// -/// Writes are last-write-wins. hands this provider a complete state -/// snapshot and gives it no way to report a conflict back, so when two writers read the same state -/// and write it back concurrently -- a scheduled check and a dashboard-initiated setting change, -/// say -- whichever writes last is kept and the other's change is lost. +/// is last-write-wins, which is what a check result wants: the most +/// recent result is the interesting one, and refusing it because a setting changed in between would +/// throw away the newer truth. /// /// -/// For check results that is the wanted behavior, since the most recent result is the interesting -/// one. Resolving it for setting changes needs a concurrency token on -/// itself: guarding the write with an ETag underneath the current interface can only turn a lost -/// update into a failed write, and a failed write is recorded as a failed health check. +/// A setting change wants the opposite, and gets it from , which +/// passes the version through as CosmosDB's own _etag on an If-Match. A write that +/// loses is refused rather than silently overwriting, and the caller reads again and reapplies. /// /// /// The CosmosDB container to store state documents in. @@ -75,6 +73,88 @@ await _container.UpsertItemAsync( .ConfigureAwait(false); } + /// + /// CosmosDB stamps every document with an _etag, so versioning costs nothing extra. + public bool SupportsOptimisticConcurrency => true; + + /// + public async Task?> GetStateEntryAsync( + string name, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + + try + { + ItemResponse> response = + await _container.ReadItemAsync>( + name, + new PartitionKey(name), + cancellationToken: cancellationToken) + .ConfigureAwait(false); + + EnsureStoredTypeMatches(name, response.Resource.StateType); + + return response.Resource.Value is { } value + ? new StateEntry(value, response.ETag) + : null; + } + catch (CosmosException ex) when (ex.StatusCode is HttpStatusCode.NotFound) + { + return null; + } + } + + /// + public async Task TrySetStateAsync( + string name, + TState state, + string? expectedVersion, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + + if (expectedVersion == IStateProvider.AbsentVersion) + { + try + { + // Create rather than upsert: CosmosDB refuses a second create for the same id, which + // is the guarantee wanted and needs no ETag to express. + await _container.CreateItemAsync( + new StateDocument(name, state), + new PartitionKey(name), + cancellationToken: cancellationToken) + .ConfigureAwait(false); + + return true; + } + catch (CosmosException ex) when (ex.StatusCode is HttpStatusCode.Conflict) + { + return false; + } + } + + var options = expectedVersion is null ? null : new ItemRequestOptions { IfMatchEtag = expectedVersion }; + + try + { + await _container.UpsertItemAsync( + new StateDocument(name, state), + new PartitionKey(name), + options, + cancellationToken) + .ConfigureAwait(false); + + return true; + } + catch (CosmosException ex) when (ex.StatusCode is HttpStatusCode.PreconditionFailed) + { + // 412 is CosmosDB reporting that the document moved on. Expected under contention, so + // it is a result rather than an exception by the time it reaches the caller. + return false; + } + } + /// public async Task DeleteStateAsync(string name, CancellationToken cancellationToken = default) { diff --git a/src/Healthie.StateProviding.CosmosDb/README.md b/src/Healthie.StateProviding.CosmosDb/README.md index cf46ca3..3f8ae02 100644 --- a/src/Healthie.StateProviding.CosmosDb/README.md +++ b/src/Healthie.StateProviding.CosmosDb/README.md @@ -58,9 +58,11 @@ builder.Services ## Concurrency -Writes are last-write-wins. When two writers read the same state and write it back concurrently — a scheduled check and a dashboard-initiated setting change, say — whichever writes last is kept, and the other's change is lost. +Writes can be made conditional. `GetStateEntryAsync` returns the state together with the document's ETag, and `TrySetStateAsync` sends it as `If-Match`, so a write made from a state that has since changed is refused (CosmosDB answers `412 PreconditionFailed`) rather than silently overwriting whoever changed it. When nothing is stored yet there is no ETag to match, so the write becomes a create, which CosmosDB refuses a second time with `409 Conflict` — the same guarantee at the one moment there is nothing to compare. -For check results that is the wanted behavior: the most recent result is the interesting one. For setting changes it is a real limitation. Resolving it needs a concurrency token on `IStateProvider` itself, which is planned for the next major version; guarding the write with an ETag underneath the current interface can only turn a lost update into a failed write, and a failed write is recorded as a failed health check. +A refused write is reported as `false`, not thrown. Under contention losing is the expected outcome and the answer is always the same: read again, reapply, write again. `StateProviderExtensions.UpdateStateAsync` is that loop. Both writers go through it — the setting change *and* the check storing its result — which is what makes the guarantee hold in both directions: a check that read the state before a setting changed no longer writes the old setting back over it. + +`SetStateAsync` still upserts unconditionally. It replaces a whole document with one the caller supplies, so there is no version to carry and nothing to compare — it is the escape hatch, not the path the library takes. ## See Also diff --git a/src/Healthie.StateProviding.Relational/RelationalDialect.cs b/src/Healthie.StateProviding.Relational/RelationalDialect.cs index f401e75..fc51118 100644 --- a/src/Healthie.StateProviding.Relational/RelationalDialect.cs +++ b/src/Healthie.StateProviding.Relational/RelationalDialect.cs @@ -27,12 +27,58 @@ namespace Healthie.StateProviding.Relational; /// Statement inserting or replacing one row, with {0} for the table name and the parameters /// @name, @state_type and @value. /// -public sealed record RelationalDialect(string Name, string CreateTableFormat, string UpsertFormat) +/// +/// Statement adding the version column to a table that predates it, with {0} for the table +/// name. Run only when the column is missing. +/// +/// +/// Statement inserting one row only if the name is not taken, reporting the outcome through rows +/// affected: one means it was written, zero that somebody else got there first. Optional, and +/// is used when it is not given. +/// +public sealed record RelationalDialect( + string Name, + string CreateTableFormat, + string UpsertFormat, + string AddVersionColumnFormat, + string? InsertIfAbsentFormat = null) { + /// + /// The fallback for a dialect that does not supply its own, which is every hand-built one. + /// + /// + /// It works everywhere and is not atomic: under READ COMMITTED two writers can both find + /// no row, both insert, and the loser take a primary key violation instead of being told it + /// lost. That is a thrown exception rather than a lost update, so it is safe in the sense that + /// matters and wrong in the sense that is visible. The three dialects below each override it + /// with a form their engine performs in one step. + /// + internal const string PortableInsertIfAbsentFormat = + "INSERT INTO {0} (name, state_type, value, version) " + + "SELECT @name, @state_type, @value, @version " + + "WHERE NOT EXISTS (SELECT 1 FROM {0} WHERE name = @name)"; + /// Reads one row. Identical on every engine, so it is not part of the dialect. internal const string SelectFormat = "SELECT state_type, value FROM {0} WHERE name = @name"; + /// Reads one row with the version to write back against. + internal const string SelectWithVersionFormat = + "SELECT state_type, value, version FROM {0} WHERE name = @name"; + + /// + /// Writes one row only if its version is still what the caller read. + /// + /// + /// The version in the WHERE clause is what makes this conditional, and rows-affected is how the + /// engine reports the outcome: zero means somebody else wrote first. It is the same shape EF + /// Core generates for a concurrency token, and it works on every engine without a stored + /// procedure or a lock. + /// + internal const string ConditionalUpdateFormat = + "UPDATE {0} SET state_type = @state_type, value = @value, version = @version " + + "WHERE name = @name AND version = @expected_version"; + /// Reads many rows at once. The parameter list is built per call, from its length. /// /// The names go in as parameters rather than as an interpolated list, so a checker name can @@ -55,9 +101,15 @@ public sealed record RelationalDialect(string Name, string CreateTableFormat, st "CREATE TABLE IF NOT EXISTS {0} (" + "name TEXT NOT NULL PRIMARY KEY, " + "state_type TEXT NULL, " + - "value TEXT NOT NULL)", - "INSERT INTO {0} (name, state_type, value) VALUES (@name, @state_type, @value) " + - "ON CONFLICT (name) DO UPDATE SET state_type = EXCLUDED.state_type, value = EXCLUDED.value"); + "value TEXT NOT NULL, " + + "version TEXT NULL)", + "INSERT INTO {0} (name, state_type, value, version) VALUES (@name, @state_type, @value, @version) " + + "ON CONFLICT (name) DO UPDATE SET state_type = EXCLUDED.state_type, value = EXCLUDED.value, version = EXCLUDED.version", + "ALTER TABLE {0} ADD COLUMN version TEXT NULL", + // The engine resolves the conflict itself, so there is no window between deciding to insert + // and inserting. A row that was already there reports zero rows affected, which is a refusal. + "INSERT INTO {0} (name, state_type, value, version) VALUES (@name, @state_type, @value, @version) " + + "ON CONFLICT (name) DO NOTHING"); /// /// name is capped at 450 characters because that is the longest a SQL Server primary key @@ -74,11 +126,18 @@ public sealed record RelationalDialect(string Name, string CreateTableFormat, st "IF OBJECT_ID(N'{0}', N'U') IS NULL CREATE TABLE {0} (" + "name NVARCHAR(450) NOT NULL PRIMARY KEY, " + "state_type NVARCHAR(4000) NULL, " + - "value NVARCHAR(MAX) NOT NULL)", - "UPDATE {0} WITH (UPDLOCK, SERIALIZABLE) SET state_type = @state_type, value = @value " + + "value NVARCHAR(MAX) NOT NULL, " + + "version NVARCHAR(64) NULL)", + "UPDATE {0} WITH (UPDLOCK, SERIALIZABLE) SET state_type = @state_type, value = @value, version = @version " + "WHERE name = @name; " + "IF @@ROWCOUNT = 0 " + - "INSERT INTO {0} (name, state_type, value) VALUES (@name, @state_type, @value);"); + "INSERT INTO {0} (name, state_type, value, version) VALUES (@name, @state_type, @value, @version);", + "ALTER TABLE {0} ADD version NVARCHAR(64) NULL", + // SQL Server has no ON CONFLICT, so the existence check takes the same UPDLOCK, HOLDLOCK the + // upsert above takes -- which is what stops a second writer reaching the same conclusion. + "INSERT INTO {0} (name, state_type, value, version) " + + "SELECT @name, @state_type, @value, @version " + + "WHERE NOT EXISTS (SELECT 1 FROM {0} WITH (UPDLOCK, HOLDLOCK) WHERE name = @name);"); /// SQLite, which needs no server and so suits a single node or a sample. public static RelationalDialect Sqlite { get; } = new( @@ -86,9 +145,13 @@ public sealed record RelationalDialect(string Name, string CreateTableFormat, st "CREATE TABLE IF NOT EXISTS {0} (" + "name TEXT NOT NULL PRIMARY KEY, " + "state_type TEXT NULL, " + - "value TEXT NOT NULL)", - "INSERT INTO {0} (name, state_type, value) VALUES (@name, @state_type, @value) " + - "ON CONFLICT(name) DO UPDATE SET state_type = excluded.state_type, value = excluded.value"); + "value TEXT NOT NULL, " + + "version TEXT NULL)", + "INSERT INTO {0} (name, state_type, value, version) VALUES (@name, @state_type, @value, @version) " + + "ON CONFLICT(name) DO UPDATE SET state_type = excluded.state_type, value = excluded.value, version = excluded.version", + "ALTER TABLE {0} ADD COLUMN version TEXT NULL", + "INSERT INTO {0} (name, state_type, value, version) VALUES (@name, @state_type, @value, @version) " + + "ON CONFLICT(name) DO NOTHING"); /// /// Checks a table name before it is put into a statement. @@ -122,8 +185,25 @@ internal static void ValidateTableName(string tableName) /// Removes one row. Identical on every engine, so it is not part of the dialect. internal const string DeleteFormat = "DELETE FROM {0} WHERE name = @name"; + /// + /// Adds the version column to a table created before it existed. + /// + /// + /// A plain ALTER, run only when the column is genuinely missing -- the initializer checks first + /// rather than relying on an IF NOT EXISTS that SQLite does not have for ADD COLUMN. + /// + internal string AddVersionColumn(string tableName) => + Format(AddVersionColumnFormat, tableName); + internal static string Select(string tableName) => Format(SelectFormat, tableName); + internal static string SelectWithVersion(string tableName) => Format(SelectWithVersionFormat, tableName); + + internal static string ConditionalUpdate(string tableName) => Format(ConditionalUpdateFormat, tableName); + + internal string InsertIfAbsent(string tableName) => + Format(InsertIfAbsentFormat ?? PortableInsertIfAbsentFormat, tableName); + internal static string Delete(string tableName) => Format(DeleteFormat, tableName); internal static string SelectMany(string tableName, int count) => diff --git a/src/Healthie.StateProviding.Relational/RelationalStateProvider.cs b/src/Healthie.StateProviding.Relational/RelationalStateProvider.cs index b4ea200..e5e0a4a 100644 --- a/src/Healthie.StateProviding.Relational/RelationalStateProvider.cs +++ b/src/Healthie.StateProviding.Relational/RelationalStateProvider.cs @@ -96,6 +96,7 @@ public async Task SetStateAsync( AddParameter(command, "@name", name); AddParameter(command, "@state_type", typeof(TState).FullName); AddParameter(command, "@value", JsonSerializer.Serialize(state)); + AddParameter(command, "@version", NewVersion()); await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); } @@ -171,6 +172,102 @@ public async Task DeleteStateAsync(string name, CancellationToken cancella return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0; } + /// + public bool SupportsOptimisticConcurrency => true; + + /// + /// A fresh version for a write. + /// + /// + /// A new value each time, generated here rather than by the database, so the same statement + /// works on every engine -- PostgreSQL has no auto-updating column and SQL Server's rowversion + /// is not portable. It is opaque to callers, so only its uniqueness matters. + /// + private static string NewVersion() => Guid.NewGuid().ToString("N"); + + /// + public async Task?> GetStateEntryAsync( + string name, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + + await using var connection = _connectionFactory(); + await connection.OpenAsync(cancellationToken).ConfigureAwait(false); + + await using var command = connection.CreateCommand(); + command.CommandText = RelationalDialect.SelectWithVersion(_tableName); + AddParameter(command, "@name", name); + + await using var reader = await command + .ExecuteReaderAsync(CommandBehavior.SingleRow, cancellationToken) + .ConfigureAwait(false); + + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return null; + } + + var storedStateType = await reader.IsDBNullAsync(0, cancellationToken).ConfigureAwait(false) + ? null + : reader.GetString(0); + + EnsureStoredTypeMatches(name, storedStateType); + + var value = JsonSerializer.Deserialize(reader.GetString(1)); + + if (value is null) + { + return null; + } + + // Null for a row written before the column existed. Reported as unversioned rather than + // invented: a caller then writes it unconditionally, exactly as it did before the upgrade, + // and the row carries a version from that write on. + var version = await reader.IsDBNullAsync(2, cancellationToken).ConfigureAwait(false) + ? null + : reader.GetString(2); + + return new StateEntry(value, version); + } + + /// + public async Task TrySetStateAsync( + string name, + TState state, + string? expectedVersion, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + + if (expectedVersion is null) + { + await SetStateAsync(name, state, cancellationToken).ConfigureAwait(false); + return true; + } + + await using var connection = _connectionFactory(); + await connection.OpenAsync(cancellationToken).ConfigureAwait(false); + + await using var command = connection.CreateCommand(); + command.CommandText = expectedVersion == IStateProvider.AbsentVersion + ? _dialect.InsertIfAbsent(_tableName) + : RelationalDialect.ConditionalUpdate(_tableName); + AddParameter(command, "@name", name); + AddParameter(command, "@state_type", typeof(TState).FullName); + AddParameter(command, "@value", JsonSerializer.Serialize(state)); + AddParameter(command, "@version", NewVersion()); + + if (expectedVersion != IStateProvider.AbsentVersion) + { + AddParameter(command, "@expected_version", expectedVersion); + } + + // Nothing updated means the version moved on, or the row is gone. Either way this write + // lost, which is exactly what the caller asked to be told. + return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0; + } + private static void AddParameter(DbCommand command, string name, string? value) { var parameter = command.CreateParameter(); diff --git a/src/Healthie.StateProviding.Relational/RelationalStateProviderInitializer.cs b/src/Healthie.StateProviding.Relational/RelationalStateProviderInitializer.cs index af73e79..3396501 100644 --- a/src/Healthie.StateProviding.Relational/RelationalStateProviderInitializer.cs +++ b/src/Healthie.StateProviding.Relational/RelationalStateProviderInitializer.cs @@ -1,4 +1,5 @@ using Healthie.Abstractions.StateProviding; +using System.Data; using System.Data.Common; namespace Healthie.StateProviding.Relational; @@ -25,6 +26,9 @@ public sealed class RelationalStateProviderInitializer( private readonly string _createTableSql = (dialect ?? throw new ArgumentNullException(nameof(dialect))) .CreateTable(Validated(tableName)); + private readonly string _addVersionColumnSql = dialect.AddVersionColumn(Validated(tableName)); + private readonly string _tableName = Validated(tableName); + private static string Validated(string tableName) { RelationalDialect.ValidateTableName(tableName); @@ -37,9 +41,77 @@ public async Task InitializeAsync(CancellationToken cancellationToken = default) await using var connection = _connectionFactory(); await connection.OpenAsync(cancellationToken).ConfigureAwait(false); - await using var command = connection.CreateCommand(); - command.CommandText = _createTableSql; + await using (var command = connection.CreateCommand()) + { + command.CommandText = _createTableSql; + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + await AddVersionColumnIfMissingAsync(connection, cancellationToken).ConfigureAwait(false); + } + + /// + /// Brings a table created before versioning existed up to date. + /// + /// + /// + /// The column is checked for rather than added blindly, because no engine here spells "add it + /// if it is missing" in a way the others also understand. So the check and the change are two + /// steps, and two instances starting together can both pass the check. The loser's ALTER + /// then fails, which would be a startup crash on a database that is in fact correct. + /// + /// + /// So a failure is not taken at face value: the columns are read again, and a failure that left + /// the column present was the other instance winning the race and is not an error. Anything else + /// is rethrown untouched. That avoids matching on error text, which differs per engine and would + /// swallow real failures along with this one. + /// + /// + /// Asking a query for no rows is how the columns are read, because every ADO.NET provider + /// answers it the same way and none of them needs the rows to describe the shape. + /// + /// + private async Task AddVersionColumnIfMissingAsync(DbConnection connection, CancellationToken cancellationToken) + { + if (await HasVersionColumnAsync(connection, cancellationToken).ConfigureAwait(false)) + { + return; + } + + try + { + await using var alter = connection.CreateCommand(); + alter.CommandText = _addVersionColumnSql; + + await alter.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + catch (DbException) when (!cancellationToken.IsCancellationRequested) + { + if (!await HasVersionColumnAsync(connection, cancellationToken).ConfigureAwait(false)) + { + throw; + } + } + } + + /// Whether the table already carries the version column. + private async Task HasVersionColumnAsync(DbConnection connection, CancellationToken cancellationToken) + { + await using var probe = connection.CreateCommand(); + probe.CommandText = $"SELECT * FROM {_tableName} WHERE 1 = 0"; + + await using var reader = await probe + .ExecuteReaderAsync(CommandBehavior.SchemaOnly, cancellationToken) + .ConfigureAwait(false); + + for (var i = 0; i < reader.FieldCount; i++) + { + if (string.Equals(reader.GetName(i), "version", StringComparison.OrdinalIgnoreCase)) + { + return true; + } + } - await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + return false; } } diff --git a/tests/Healthie.Tests.Unit/ConcurrencyTests.cs b/tests/Healthie.Tests.Unit/ConcurrencyTests.cs new file mode 100644 index 0000000..405914a --- /dev/null +++ b/tests/Healthie.Tests.Unit/ConcurrencyTests.cs @@ -0,0 +1,673 @@ +using Healthie.Abstractions.Enums; +using Healthie.Abstractions.Models; +using Healthie.Abstractions.StateProviding; +using Healthie.DependencyInjection; +using Healthie.StateProviding.Relational; +using Microsoft.Data.Sqlite; +using System.Data.Common; + +namespace Healthie.Tests.Unit; + +/// +/// The bug this closes was documented on CosmosDbStateProvider from the start: reading a state, +/// changing it and writing it back is three steps, so a check finishing in between wrote its result +/// over a setting change -- or had its result written over. A version on the write makes the loser +/// find out instead of guessing. +/// +public class OptimisticConcurrencyTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// A provider written against the original two-method interface. + private sealed class UnversionedProvider : IStateProvider + { + private readonly Dictionary _states = new(StringComparer.Ordinal); + + public Task GetStateAsync(string name, CancellationToken cancellationToken = default) + => Task.FromResult(_states.TryGetValue(name, out var s) ? (TState?)s : default); + + public Task SetStateAsync(string name, TState state, CancellationToken cancellationToken = default) + { + _states[name] = state!; + return Task.CompletedTask; + } + } + + [Fact] + public async Task AnUnversionedProvider_SaysSoRatherThanClaimingProtection() + { + IStateProvider provider = new UnversionedProvider(); + + Assert.False(provider.SupportsOptimisticConcurrency); + + var entry = await provider.GetStateEntryAsync("absent", Ct); + Assert.Null(entry); + } + + /// + /// The honest half: a provider that cannot honour a version refuses rather than writing + /// unconditionally, because ignoring it would lose exactly the update it was passed to protect. + /// + [Fact] + public async Task AnUnversionedProvider_RefusesAVersionedWrite() + { + IStateProvider provider = new UnversionedProvider(); + + var ex = await Assert.ThrowsAsync( + () => provider.TrySetStateAsync("x", new PulseCheckerState(), "some-version", Ct)); + + Assert.Contains(nameof(UnversionedProvider), ex.Message, StringComparison.Ordinal); + } + + [Fact] + public async Task AnUnversionedProvider_StillAcceptsAnUnconditionalWrite() + { + IStateProvider provider = new UnversionedProvider(); + + Assert.True(await provider.TrySetStateAsync("x", new PulseCheckerState(), expectedVersion: null, Ct)); + } + + [Fact] + public async Task AVersionedProvider_ReturnsAVersionItCanWriteBackAgainst() + { + IStateProvider provider = new InMemoryStateProvider(); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.EverySecond), Ct); + + var entry = await provider.GetStateEntryAsync("x", Ct); + + Assert.True(provider.SupportsOptimisticConcurrency); + Assert.True(entry!.IsVersioned); + Assert.True(await provider.TrySetStateAsync("x", entry.Value, entry.Version, Ct)); + } + + /// + /// The heart of it: a write made from a state that has since moved on is refused rather than + /// silently overwriting whatever moved it. + /// + [Fact] + public async Task AWriteFromAStaleRead_IsRefused() + { + IStateProvider provider = new InMemoryStateProvider(); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.EverySecond), Ct); + + var stale = await provider.GetStateEntryAsync("x", Ct); + + // Somebody else writes in between. + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.Every5Minutes), Ct); + + Assert.False(await provider.TrySetStateAsync("x", stale!.Value, stale.Version, Ct)); + + // And the other writer's value is still there, unclobbered. + var current = await provider.GetStateAsync("x", Ct); + Assert.Equal(PulseInterval.Every5Minutes, current!.Interval); + } + + [Fact] + public async Task EveryWrite_MovesTheVersionOn() + { + IStateProvider provider = new InMemoryStateProvider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + var first = (await provider.GetStateEntryAsync("x", Ct))!.Version; + + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.Every2Seconds), Ct); + var second = (await provider.GetStateEntryAsync("x", Ct))!.Version; + + Assert.NotEqual(first, second); + } + + /// + /// The retry loop from the Azure SDK's conditional-request guidance, in one place rather than + /// in every caller: read, reapply, write, and go round if somebody got in first. + /// + [Fact] + public async Task UpdateStateAsync_ReappliesAgainstWhoeverWonAndDoesNotLoseTheirChange() + { + IStateProvider provider = new InMemoryStateProvider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + var interfered = false; + + var result = await provider.UpdateStateAsync( + "x", + state => + { + // Interfere once, after the read but before the write, exactly as a check finishing + // mid-edit would. The first attempt must lose, and the retry must build on the + // interfering write rather than discard it. + if (!interfered) + { + interfered = true; + provider.SetStateAsync("x", new PulseCheckerState { Group = "written-by-someone-else" }, Ct) + .GetAwaiter().GetResult(); + } + + state.IsPinned = true; + }, + () => new PulseCheckerState(), + cancellationToken: Ct); + + Assert.True(result.IsPinned); + Assert.Equal("written-by-someone-else", result.Group); + } + + [Fact] + public async Task UpdateStateAsync_GivesUpRatherThanSpinningForEver() + { + IStateProvider provider = new InMemoryStateProvider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + var ex = await Assert.ThrowsAsync( + () => provider.UpdateStateAsync( + "x", + state => + { + // Interferes on every attempt, so no write can ever land. + provider.SetStateAsync("x", new PulseCheckerState { Group = Guid.NewGuid().ToString() }, Ct) + .GetAwaiter().GetResult(); + state.IsPinned = true; + }, + () => new PulseCheckerState(), + maxAttempts: 3, + cancellationToken: Ct)); + + Assert.Contains("3 attempts", ex.Message, StringComparison.Ordinal); + } + + [Fact] + public async Task UpdateStateAsync_CreatesTheStateWhenThereIsNone() + { + IStateProvider provider = new InMemoryStateProvider(); + + var result = await provider.UpdateStateAsync( + "brand-new", + (PulseCheckerState state) => state.IsPinned = true, + () => new PulseCheckerState(PulseInterval.Every3Seconds), + cancellationToken: Ct); + + Assert.True(result.IsPinned); + Assert.Equal(PulseInterval.Every3Seconds, result.Interval); + } + + /// + /// Against a provider that cannot version, the loop degrades to what the library did before -- + /// read, change, write. That is what an unversioned store can offer, not a silent downgrade. + /// + /// + /// The create half. Until there is a row there is no version to compare, so the write would go + /// through unconditionally -- and two writers both finding nothing would lose one of the two + /// changes. is the "only if it is still missing" + /// precondition HTTP spells If-None-Match: *. + /// + [Fact] + public async Task ACreateThatLosesToAnotherCreate_IsRefused() + { + IStateProvider provider = new InMemoryStateProvider(); + + Assert.True(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "first" }, IStateProvider.AbsentVersion, Ct)); + + Assert.False(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "second" }, IStateProvider.AbsentVersion, Ct)); + + Assert.Equal("first", (await provider.GetStateAsync("x", Ct))!.Group); + } + + [Fact] + public async Task UpdateStateAsync_StillWorksAgainstAnUnversionedProvider() + { + IStateProvider provider = new UnversionedProvider(); + + var result = await provider.UpdateStateAsync( + "x", + (PulseCheckerState state) => state.IsPinned = true, + () => new PulseCheckerState(), + cancellationToken: Ct); + + Assert.True(result.IsPinned); + } +} + +/// +/// A setting change made from the dashboard while checks are running is the case the whole feature +/// exists for, so it is driven through a real checker rather than the provider alone. +/// +public class PulseCheckerConcurrencyTests +{ + /// + /// A checker with a name of its own. + /// + /// + /// Takes only an and carries the name on a settable property, + /// because AddHealthie scans this assembly and registers every non-abstract PulseChecker + /// it finds. A constructor parameter the container cannot resolve breaks every other test that + /// scans -- which is exactly what the first version of this did, for the second time. + /// + private sealed class NamedTestChecker(IStateProvider states) : Healthie.Abstractions.PulseChecker(states) + { + public string CheckerName { get; init; } = "named-test-checker"; + + public override string Name => CheckerName; + + public override Task CheckAsync(CancellationToken cancellationToken = default) + => Task.FromResult(new PulseCheckerResult(PulseCheckerHealth.Healthy, "ok")); + } + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// + /// Wraps a provider and lets one competing write slip in between a read and the write that + /// follows it -- which is what a second replica, or the REST API on another instance, does. + /// + /// + /// A checker's semaphore already serialises its own check loop against its own setting changes, + /// so a single instance cannot show this. Modelling the other writer explicitly makes the race + /// deterministic instead of hoping two threads interleave the wrong way. + /// + private sealed class InterferingProvider(IStateProvider inner, Action interfere) : IStateProvider + { + private bool _done; + + public bool SupportsOptimisticConcurrency => inner.SupportsOptimisticConcurrency; + + public async Task GetStateAsync(string name, CancellationToken cancellationToken = default) + { + var state = await inner.GetStateAsync(name, cancellationToken); + InterfereOnce(); + return state; + } + + public Task SetStateAsync(string name, TState state, CancellationToken cancellationToken = default) + => inner.SetStateAsync(name, state, cancellationToken); + + public async Task?> GetStateEntryAsync(string name, CancellationToken cancellationToken = default) + { + var entry = await inner.GetStateEntryAsync(name, cancellationToken); + InterfereOnce(); + return entry; + } + + /// + /// Both read shapes interfere, so a test means the same thing whichever one the code under + /// test happens to call -- otherwise migrating a caller from one to the other would quietly + /// disarm the test rather than break it. + /// + private void InterfereOnce() + { + if (_done) + { + return; + } + + _done = true; + interfere(inner); + } + + public Task TrySetStateAsync(string name, TState state, string? expectedVersion, CancellationToken cancellationToken = default) + => inner.TrySetStateAsync(name, state, expectedVersion, cancellationToken); + } + + /// + /// The bug, exactly as CosmosDbStateProvider described it: another writer changes the state + /// between this one's read and its write. Without a version the later write wins and the other + /// change is gone; with one it is refused, reapplied, and both survive. + /// + [Fact] + public async Task AnotherWriterBetweenTheReadAndTheWrite_DoesNotLoseItsChange() + { + var store = new InMemoryStateProvider(); + var provider = new InterferingProvider( + store, + inner => inner.SetStateAsync( + "racy", + new PulseCheckerState { UnhealthyThreshold = 7 }, + CancellationToken.None).GetAwaiter().GetResult()); + + using var checker = new NamedTestChecker(provider) { CheckerName = "racy" }; + + await checker.SetGroupAsync("set-by-the-dashboard", Ct); + + var state = await store.GetStateAsync("racy", Ct); + + Assert.Equal("set-by-the-dashboard", state!.Group); + Assert.Equal(7u, state.UnhealthyThreshold); + } + + /// + /// The other direction, and the one the CHANGELOG actually describes: a setting is changed + /// while a check is running, and the check's own write must not revert it. + /// + /// + /// A checker writes its result by reading the whole state, changing the result fields and + /// writing the whole state back, so an unconditional write puts every other field back to what + /// it was when the check started. Within one process the semaphore hides this -- a check holds + /// it across both steps. Across two instances sharing one store, which is what a second replica + /// or the REST API on another node is, nothing does. + /// + [Fact] + public async Task ACheckWritingItsResult_DoesNotRevertASettingChangedWhileItRan() + { + var store = new InMemoryStateProvider(); + + var provider = new InterferingProvider( + store, + inner => inner.SetStateAsync( + "racy-check", + new PulseCheckerState { Group = "set-while-the-check-ran" }, + CancellationToken.None).GetAwaiter().GetResult()); + + using var checker = new NamedTestChecker(provider) { CheckerName = "racy-check" }; + + await checker.TriggerAsync(Ct); + + var state = await store.GetStateAsync("racy-check", Ct); + + // The check's own result landed... + Assert.NotNull(state!.LastResult); + Assert.Equal(PulseCheckerHealth.Healthy, state.LastResult!.Health); + + // ...without putting the group back to what it was before the setting change. + Assert.Equal("set-while-the-check-ran", state.Group); + } + + [Fact] + public async Task SettingSomethingToWhatItAlreadyIs_WritesNothing() + { + var states = new InMemoryStateProvider(); + using var checker = new AlwaysHealthyPulseChecker(states); + + await checker.SetPinnedAsync(true, Ct); + var afterFirst = (await states.GetStateEntryAsync(checker.Name, Ct))!.Version; + + await checker.SetPinnedAsync(true, Ct); + var afterSecond = (await states.GetStateEntryAsync(checker.Name, Ct))!.Version; + + Assert.Equal(afterFirst, afterSecond); + } + + [Fact] + public async Task StartAndStop_StillReportWhetherTheyChangedAnything() + { + using var checker = new AlwaysHealthyPulseChecker(new InMemoryStateProvider()); + + Assert.False(await checker.StartAsync(Ct)); // already active + Assert.True(await checker.StopAsync(Ct)); + Assert.False(await checker.StopAsync(Ct)); // already stopped + Assert.True(await checker.StartAsync(Ct)); + } +} + +/// +/// Driven against a real SQLite file, so the conditional UPDATE and the column migration are +/// executed rather than inspected. This is the provider PostgreSQL and SQL Server share. +/// +public sealed class RelationalConcurrencyTests : IAsyncLifetime +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private const string Table = "healthie_pulse_state"; + + private string _databasePath = string.Empty; + private string _connectionString = string.Empty; + + private DbConnection Connect() => new SqliteConnection(_connectionString); + + private RelationalStateProvider Provider() => new(Connect, RelationalDialect.Sqlite, Table); + + public ValueTask InitializeAsync() + { + _databasePath = Path.Combine(Path.GetTempPath(), $"healthie-cc-{Guid.NewGuid():N}.db"); + _connectionString = $"Data Source={_databasePath};Pooling=False"; + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() + { + SqliteConnection.ClearAllPools(); + File.Delete(_databasePath); + return ValueTask.CompletedTask; + } + + private Task InitializeSchemaAsync() => + new RelationalStateProviderInitializer(Connect, RelationalDialect.Sqlite, Table).InitializeAsync(Ct); + + [Fact] + public async Task AWriteFromAStaleRead_IsRefused() + { + await InitializeSchemaAsync(); + var provider = Provider(); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.EverySecond), Ct); + + var stale = await provider.GetStateEntryAsync("x", Ct); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.Every5Minutes), Ct); + + Assert.False(await provider.TrySetStateAsync("x", stale!.Value, stale.Version, Ct)); + Assert.Equal(PulseInterval.Every5Minutes, (await provider.GetStateAsync("x", Ct))!.Interval); + } + + [Fact] + public async Task AWriteFromACurrentRead_Lands() + { + await InitializeSchemaAsync(); + var provider = Provider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + var entry = await provider.GetStateEntryAsync("x", Ct); + entry!.Value.IsPinned = true; + + Assert.True(await provider.TrySetStateAsync("x", entry.Value, entry.Version, Ct)); + Assert.True((await provider.GetStateAsync("x", Ct))!.IsPinned); + } + + [Fact] + public async Task AConditionalWriteAgainstAMissingRow_IsRefusedRatherThanInserting() + { + await InitializeSchemaAsync(); + var provider = Provider(); + + Assert.False(await provider.TrySetStateAsync("never-stored", new PulseCheckerState(), "made-up-version", Ct)); + Assert.Null(await provider.GetStateAsync("never-stored", Ct)); + } + + [Fact] + public async Task ACreateThatLosesToAnotherCreate_IsRefused() + { + await InitializeSchemaAsync(); + var provider = Provider(); + + Assert.True(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "first" }, IStateProvider.AbsentVersion, Ct)); + + Assert.False(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "second" }, IStateProvider.AbsentVersion, Ct)); + + Assert.Equal("first", (await provider.GetStateAsync("x", Ct))!.Group); + } + + /// + /// A table created before versioning existed has no version column. The initializer has to add + /// it without losing the rows already there, and without an IF NOT EXISTS that SQLite lacks for + /// ADD COLUMN. + /// + [Fact] + public async Task ATablePredatingVersioning_IsMigratedWithoutLosingItsRows() + { + await using (var connection = Connect()) + { + await connection.OpenAsync(Ct); + await using var create = connection.CreateCommand(); + create.CommandText = + $"CREATE TABLE {Table} (name TEXT NOT NULL PRIMARY KEY, state_type TEXT NULL, value TEXT NOT NULL);" + + $"INSERT INTO {Table} (name, state_type, value) VALUES " + + $"('legacy', 'Healthie.Abstractions.Models.PulseCheckerState', '{{\"Interval\":\"Every30Seconds\"}}');"; + await create.ExecuteNonQueryAsync(Ct); + } + + await InitializeSchemaAsync(); + + var provider = Provider(); + var entry = await provider.GetStateEntryAsync("legacy", Ct); + + Assert.Equal(PulseInterval.Every30Seconds, entry!.Value.Interval); + + // The pre-existing row has no version yet, and says so rather than inventing one. + Assert.Null(entry.Version); + Assert.False(entry.IsVersioned); + } + + /// + /// The upgrade path. Every row already in the table has no version until it is next written, and + /// treating "stored but unversioned" as "not stored" refuses a write that can never succeed -- + /// so the very first setting change after an upgrade would fail, on every existing checker. + /// + [Fact] + public async Task AnUnversionedRow_CanStillBeUpdated() + { + await using (var connection = Connect()) + { + await connection.OpenAsync(Ct); + await using var create = connection.CreateCommand(); + create.CommandText = + $"CREATE TABLE {Table} (name TEXT NOT NULL PRIMARY KEY, state_type TEXT NULL, value TEXT NOT NULL);" + + $"INSERT INTO {Table} (name, state_type, value) VALUES " + + $"('legacy', 'Healthie.Abstractions.Models.PulseCheckerState', '{{\"Interval\":\"EveryMinute\"}}');"; + await create.ExecuteNonQueryAsync(Ct); + } + + await InitializeSchemaAsync(); + + var provider = Provider(); + + var updated = await provider.UpdateStateAsync( + "legacy", + (PulseCheckerState state) => state.Group = "set-after-upgrading", + () => new PulseCheckerState(), + cancellationToken: Ct); + + // The change landed, on top of what was already stored rather than over it. + Assert.Equal("set-after-upgrading", updated.Group); + Assert.Equal(PulseInterval.EveryMinute, updated.Interval); + + // And the row is versioned from here on, so the next write is protected. + Assert.True((await provider.GetStateEntryAsync("legacy", Ct))!.IsVersioned); + } + + /// + /// Every shipped dialect must resolve the create in one statement. The portable fallback is + /// correct but not atomic, so a dialect that quietly fell back to it would reintroduce the race + /// with nothing to show for it. + /// + [Fact] + public void EveryShippedDialect_ResolvesTheCreateItself() + { + RelationalDialect[] shipped = + [ + RelationalDialect.PostgreSql, + RelationalDialect.SqlServer, + RelationalDialect.Sqlite, + ]; + + foreach (var dialect in shipped) + { + Assert.NotNull(dialect.InsertIfAbsentFormat); + + // Either the engine resolves the conflict, or the existence check is taken under a lock. + var sql = dialect.InsertIfAbsentFormat!; + var resolvesItself = + sql.Contains("DO NOTHING", StringComparison.OrdinalIgnoreCase) || + sql.Contains("UPDLOCK", StringComparison.OrdinalIgnoreCase); + + Assert.True(resolvesItself, $"{dialect.Name} falls back to a check-then-insert race."); + } + } + + /// + /// Two instances starting together can both find the column missing. The loser's ALTER fails + /// against a database that is now correct, and a startup crash there is a crash for nothing. + /// + /// + /// The race is made deterministic by a statement that adds the column and then fails, which is + /// what the losing instance sees: the column is there, and its own command reported an error. + /// + [Fact] + public async Task LosingTheMigrationRace_IsNotAnError() + { + await CreateUnversionedTableAsync(); + + var addsThenFails = RelationalDialect.Sqlite with + { + AddVersionColumnFormat = + "ALTER TABLE {0} ADD COLUMN version TEXT NULL; SELECT this_is_not_a_column", + }; + + await new RelationalStateProviderInitializer(Connect, addsThenFails, Table).InitializeAsync(Ct); + + // Swallowed, because the column it was adding is there. + var provider = Provider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + Assert.True((await provider.GetStateEntryAsync("x", Ct))!.IsVersioned); + } + + /// + /// The other half: a failure that left the column missing is a real failure and is not hidden. + /// + [Fact] + public async Task AMigrationThatFailsWithoutAddingTheColumn_StillThrows() + { + await CreateUnversionedTableAsync(); + + var justFails = RelationalDialect.Sqlite with + { + AddVersionColumnFormat = "SELECT this_is_not_a_column FROM {0}", + }; + + await Assert.ThrowsAnyAsync( + () => new RelationalStateProviderInitializer(Connect, justFails, Table).InitializeAsync(Ct)); + } + + /// A table as it stood before the version column existed. + private async Task CreateUnversionedTableAsync() + { + await using var connection = Connect(); + await connection.OpenAsync(Ct); + + await using var create = connection.CreateCommand(); + create.CommandText = + $"CREATE TABLE {Table} (name TEXT NOT NULL PRIMARY KEY, state_type TEXT NULL, value TEXT NOT NULL)"; + + await create.ExecuteNonQueryAsync(Ct); + } + + [Fact] + public async Task RunningTheInitializerTwice_DoesNotAddTheColumnTwice() + { + await InitializeSchemaAsync(); + await InitializeSchemaAsync(); + + var provider = Provider(); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + Assert.True((await provider.GetStateEntryAsync("x", Ct))!.IsVersioned); + } + + [Fact] + public async Task AMigratedRow_BecomesVersionedOnItsNextWrite() + { + await using (var connection = Connect()) + { + await connection.OpenAsync(Ct); + await using var create = connection.CreateCommand(); + create.CommandText = + $"CREATE TABLE {Table} (name TEXT NOT NULL PRIMARY KEY, state_type TEXT NULL, value TEXT NOT NULL);" + + $"INSERT INTO {Table} (name, state_type, value) VALUES " + + $"('legacy', 'Healthie.Abstractions.Models.PulseCheckerState', '{{\"Interval\":\"EveryMinute\"}}');"; + await create.ExecuteNonQueryAsync(Ct); + } + + await InitializeSchemaAsync(); + + var provider = Provider(); + await provider.SetStateAsync("legacy", new PulseCheckerState(PulseInterval.Every2Minutes), Ct); + + Assert.True((await provider.GetStateEntryAsync("legacy", Ct))!.IsVersioned); + } +} diff --git a/tests/Healthie.Tests.Unit/CosmosDbConcurrencyTests.cs b/tests/Healthie.Tests.Unit/CosmosDbConcurrencyTests.cs new file mode 100644 index 0000000..aef30b6 --- /dev/null +++ b/tests/Healthie.Tests.Unit/CosmosDbConcurrencyTests.cs @@ -0,0 +1,237 @@ +using Healthie.Abstractions.Enums; +using Healthie.Abstractions.Models; +using Healthie.Abstractions.StateProviding; +using Healthie.StateProviding.CosmosDb; +using Microsoft.Azure.Cosmos; +using System.Net; +using System.Reflection; + +namespace Healthie.Tests.Unit; + +/// +/// The CosmosDB provider's conditional write, driven through a container that models the ETag rules +/// the service applies: every write mints a new ETag, an If-Match against a stale one is +/// refused with 412, and a second create for the same id is refused with 409. +/// +/// +/// A fake rather than a real container, because the guarantee under test is what the provider asks +/// for -- whether it sends the ETag it read, and whether it reads a refusal as a refusal rather +/// than letting it escape as an exception. That is decided entirely by the request the provider +/// builds, so a container that applies the documented rules answers it. It does not prove the +/// service behaves as documented; only a run against real CosmosDB does that, and none is wired up. +/// +public class CosmosDbConcurrencyTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// One stored document and the ETag it currently carries. + private sealed record Stored(object Document, string ETag); + + private sealed class FakeResponse(T resource, string etag) : ItemResponse + { + public override T Resource => resource; + + public override string ETag => etag; + + public override HttpStatusCode StatusCode => HttpStatusCode.OK; + + public override Headers Headers => new(); + + public override double RequestCharge => 0; + + public override string ActivityId => string.Empty; + + public override CosmosDiagnostics Diagnostics => null!; + } + + /// + /// The parts of a container this provider uses, with CosmosDB's concurrency rules and nothing + /// else. + /// + private sealed class FakeContainer : StubContainer + { + private readonly Dictionary _items = new(StringComparer.Ordinal); + private int _etags; + + /// How many writes were sent without a condition attached. + public int UnconditionalWrites { get; private set; } + + private string NextETag() => $"\"etag-{++_etags}\""; + + /// + /// The document's id, which is what CosmosDB keys on. + /// + /// + /// Read by reflection because the document type is internal to the provider. Case-insensitive + /// because the property is spelled id, in lower case, as the CosmosDB SDK requires. + /// + private static string IdOf(object document) => + document.GetType() + .GetProperty("id", BindingFlags.Public | BindingFlags.Instance | BindingFlags.IgnoreCase) + ?.GetValue(document) as string + ?? throw new InvalidOperationException($"{document.GetType().Name} has no id to store it under."); + + public override Task> ReadItemAsync( + string id, + PartitionKey partitionKey, + ItemRequestOptions? requestOptions = null, + CancellationToken cancellationToken = default) + { + if (!_items.TryGetValue(id, out var stored)) + { + throw new CosmosException("Not found", HttpStatusCode.NotFound, 0, string.Empty, 0); + } + + return Task.FromResult>(new FakeResponse((T)stored.Document, stored.ETag)); + } + + public override Task> UpsertItemAsync( + T item, + PartitionKey? partitionKey = null, + ItemRequestOptions? requestOptions = null, + CancellationToken cancellationToken = default) + { + var id = IdOf(item!); + var condition = requestOptions?.IfMatchEtag; + + if (condition is null) + { + UnconditionalWrites++; + } + else if (!_items.TryGetValue(id, out var current) || current.ETag != condition) + { + // Exactly what the service answers when the document moved on: the write is refused. + throw new CosmosException( + "Precondition failed", HttpStatusCode.PreconditionFailed, 0, string.Empty, 0); + } + + var etag = NextETag(); + _items[id] = new Stored(item!, etag); + + return Task.FromResult>(new FakeResponse(item, etag)); + } + + public override Task> CreateItemAsync( + T item, + PartitionKey? partitionKey = null, + ItemRequestOptions? requestOptions = null, + CancellationToken cancellationToken = default) + { + var id = IdOf(item!); + + if (_items.ContainsKey(id)) + { + throw new CosmosException("Conflict", HttpStatusCode.Conflict, 0, string.Empty, 0); + } + + var etag = NextETag(); + _items[id] = new Stored(item!, etag); + + return Task.FromResult>(new FakeResponse(item, etag)); + } + + } + + private static CosmosDbStateProvider Provider(out FakeContainer container) + { + container = new FakeContainer(); + return new CosmosDbStateProvider(container); + } + + [Fact] + public void TheProvider_SaysItCanVersionAWrite() + { + Assert.True(Provider(out _).SupportsOptimisticConcurrency); + } + + [Fact] + public async Task AReadEntry_CarriesTheDocumentsETag() + { + var provider = Provider(out _); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.EverySecond), Ct); + + var entry = await provider.GetStateEntryAsync("x", Ct); + + Assert.True(entry!.IsVersioned); + Assert.Equal(PulseInterval.EverySecond, entry.Value.Interval); + } + + [Fact] + public async Task AWriteFromACurrentRead_Lands() + { + var provider = Provider(out _); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + var entry = await provider.GetStateEntryAsync("x", Ct); + entry!.Value.IsPinned = true; + + Assert.True(await provider.TrySetStateAsync("x", entry.Value, entry.Version, Ct)); + Assert.True((await provider.GetStateAsync("x", Ct))!.IsPinned); + } + + /// + /// The point of the whole feature: a 412 is a refusal to be reported, not an exception to + /// escape. A checker turns a throw into a failed health check, which would report a healthy + /// component as down for the sake of this library's own bookkeeping. + /// + [Fact] + public async Task AWriteFromAStaleRead_IsRefusedRatherThanThrowing() + { + var provider = Provider(out _); + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.EverySecond), Ct); + + var stale = await provider.GetStateEntryAsync("x", Ct); + + // Somebody else writes, moving the ETag on. + await provider.SetStateAsync("x", new PulseCheckerState(PulseInterval.Every5Minutes), Ct); + + Assert.False(await provider.TrySetStateAsync("x", stale!.Value, stale.Version, Ct)); + + // And their write is still there. + Assert.Equal( + PulseInterval.Every5Minutes, + (await provider.GetStateAsync("x", Ct))!.Interval); + } + + [Fact] + public async Task ACreateThatLosesToAnotherCreate_IsRefused() + { + var provider = Provider(out _); + + Assert.True(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "first" }, IStateProvider.AbsentVersion, Ct)); + + Assert.False(await provider.TrySetStateAsync( + "x", new PulseCheckerState { Group = "second" }, IStateProvider.AbsentVersion, Ct)); + + Assert.Equal("first", (await provider.GetStateAsync("x", Ct))!.Group); + } + + /// + /// A conditional write must actually carry its condition. A provider that dropped the ETag + /// would pass every test above by writing unconditionally and never being refused. + /// + [Fact] + public async Task AVersionedWrite_IsSentAsAConditionalRequest() + { + var provider = Provider(out var container); + await provider.SetStateAsync("x", new PulseCheckerState(), Ct); + + var unconditionalSoFar = container.UnconditionalWrites; + + var entry = await provider.GetStateEntryAsync("x", Ct); + await provider.TrySetStateAsync("x", entry!.Value, entry.Version, Ct); + + Assert.Equal(unconditionalSoFar, container.UnconditionalWrites); + } + + [Fact] + public async Task AnUnversionedWrite_IsStillUnconditional() + { + var provider = Provider(out var container); + + await provider.TrySetStateAsync("x", new PulseCheckerState(), expectedVersion: null, Ct); + + Assert.Equal(1, container.UnconditionalWrites); + } +} diff --git a/tests/Healthie.Tests.Unit/StubContainer.cs b/tests/Healthie.Tests.Unit/StubContainer.cs new file mode 100644 index 0000000..30a1f62 --- /dev/null +++ b/tests/Healthie.Tests.Unit/StubContainer.cs @@ -0,0 +1,112 @@ +using Microsoft.Azure.Cosmos; +using Microsoft.Azure.Cosmos.Scripts; + +namespace Healthie.Tests.Unit; + +/// +/// A whose every member refuses, so a test double can override the few it +/// actually needs. +/// +/// +/// +/// is abstract with around forty members, of which this library calls four. +/// Left in the test that uses them, the other thirty-six would bury it. Refusing rather than +/// returning a default is deliberate: a test that reaches a member nobody stubbed has left the path +/// it meant to exercise, and should say so rather than quietly carry on with a null. +/// +/// +/// Hand-written rather than generated by a mocking library, because the repository has no mocking +/// dependency and one type is not a reason to take one. +/// +/// +internal abstract class StubContainer : Container +{ + private static NotSupportedException NotStubbed(string member) => + new($"{member} is not stubbed. Override it if the test under way is meant to reach it."); + + public override string Id => throw NotStubbed(nameof(Id)); + + public override Database Database => throw NotStubbed(nameof(Database)); + + public override Conflicts Conflicts => throw NotStubbed(nameof(Conflicts)); + + public override Scripts Scripts => throw NotStubbed(nameof(Scripts)); + + public override Task CreateItemStreamAsync(Stream streamPayload, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(CreateItemStreamAsync)); + + public override TransactionalBatch CreateTransactionalBatch(PartitionKey partitionKey) => throw NotStubbed(nameof(CreateTransactionalBatch)); + + public override Task DeleteContainerAsync(ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(DeleteContainerAsync)); + + public override Task DeleteContainerStreamAsync(ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(DeleteContainerStreamAsync)); + + public override Task> DeleteItemAsync(string id, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(DeleteItemAsync)); + + public override Task DeleteItemStreamAsync(string id, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(DeleteItemStreamAsync)); + + public override ChangeFeedEstimator GetChangeFeedEstimator(string processorName, Container leaseContainer) => throw NotStubbed(nameof(GetChangeFeedEstimator)); + + public override ChangeFeedProcessorBuilder GetChangeFeedEstimatorBuilder(string processorName, ChangesEstimationHandler estimationDelegate, TimeSpan? estimationPeriod) => throw NotStubbed(nameof(GetChangeFeedEstimatorBuilder)); + + public override FeedIterator GetChangeFeedIterator(ChangeFeedStartFrom changeFeedStartFrom, ChangeFeedMode changeFeedMode, ChangeFeedRequestOptions? changeFeedRequestOptions) => throw NotStubbed(nameof(GetChangeFeedIterator)); + + public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder(string processorName, ChangeFeedStreamHandler onChangesDelegate) => throw NotStubbed(nameof(GetChangeFeedProcessorBuilder)); + + public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder(string processorName, ChangeFeedHandler onChangesDelegate) => throw NotStubbed(nameof(GetChangeFeedProcessorBuilder)); + + public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilder(string processorName, ChangesHandler onChangesDelegate) => throw NotStubbed(nameof(GetChangeFeedProcessorBuilder)); + + public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderWithManualCheckpoint(string processorName, ChangeFeedStreamHandlerWithManualCheckpoint onChangesDelegate) => throw NotStubbed(nameof(GetChangeFeedProcessorBuilderWithManualCheckpoint)); + + public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderWithManualCheckpoint(string processorName, ChangeFeedHandlerWithManualCheckpoint onChangesDelegate) => throw NotStubbed(nameof(GetChangeFeedProcessorBuilderWithManualCheckpoint)); + + public override FeedIterator GetChangeFeedStreamIterator(ChangeFeedStartFrom changeFeedStartFrom, ChangeFeedMode changeFeedMode, ChangeFeedRequestOptions? changeFeedRequestOptions) => throw NotStubbed(nameof(GetChangeFeedStreamIterator)); + + public override Task> GetFeedRangesAsync(CancellationToken cancellationToken) => throw NotStubbed(nameof(GetFeedRangesAsync)); + + public override IOrderedQueryable GetItemLinqQueryable(bool allowSynchronousQueryExecution, string? continuationToken, QueryRequestOptions? requestOptions, CosmosLinqSerializerOptions? linqSerializerOptions) => throw NotStubbed(nameof(GetItemLinqQueryable)); + + public override FeedIterator GetItemQueryIterator(FeedRange feedRange, QueryDefinition queryDefinition, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryIterator)); + + public override FeedIterator GetItemQueryIterator(QueryDefinition queryDefinition, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryIterator)); + + public override FeedIterator GetItemQueryIterator(string? queryText, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryIterator)); + + public override FeedIterator GetItemQueryStreamIterator(FeedRange feedRange, QueryDefinition queryDefinition, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryStreamIterator)); + + public override FeedIterator GetItemQueryStreamIterator(QueryDefinition queryDefinition, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryStreamIterator)); + + public override FeedIterator GetItemQueryStreamIterator(string? queryText, string? continuationToken, QueryRequestOptions? requestOptions) => throw NotStubbed(nameof(GetItemQueryStreamIterator)); + + public override Task> PatchItemAsync(string id, PartitionKey partitionKey, IReadOnlyList patchOperations, PatchItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(PatchItemAsync)); + + public override Task PatchItemStreamAsync(string id, PartitionKey partitionKey, IReadOnlyList patchOperations, PatchItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(PatchItemStreamAsync)); + + public override Task ReadContainerAsync(ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadContainerAsync)); + + public override Task ReadContainerStreamAsync(ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadContainerStreamAsync)); + + public override Task ReadItemStreamAsync(string id, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadItemStreamAsync)); + + public override Task> ReadManyItemsAsync(IReadOnlyList<(string id, PartitionKey partitionKey)> items, ReadManyRequestOptions? readManyRequestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadManyItemsAsync)); + + public override Task ReadManyItemsStreamAsync(IReadOnlyList<(string id, PartitionKey partitionKey)> items, ReadManyRequestOptions? readManyRequestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadManyItemsStreamAsync)); + + public override Task ReadThroughputAsync(CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadThroughputAsync)); + + public override Task ReadThroughputAsync(RequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReadThroughputAsync)); + + public override Task ReplaceContainerAsync(ContainerProperties containerProperties, ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceContainerAsync)); + + public override Task ReplaceContainerStreamAsync(ContainerProperties containerProperties, ContainerRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceContainerStreamAsync)); + + public override Task> ReplaceItemAsync(T item, string id, PartitionKey? partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceItemAsync)); + + public override Task ReplaceItemStreamAsync(Stream streamPayload, string id, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceItemStreamAsync)); + + public override Task ReplaceThroughputAsync(ThroughputProperties throughputProperties, RequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceThroughputAsync)); + + public override Task ReplaceThroughputAsync(int throughput, RequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(ReplaceThroughputAsync)); + + public override Task UpsertItemStreamAsync(Stream streamPayload, PartitionKey partitionKey, ItemRequestOptions? requestOptions, CancellationToken cancellationToken) => throw NotStubbed(nameof(UpsertItemStreamAsync)); +}