Make state writes conditional, so a setting change is not lost to a check

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
 

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.

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);
+
+    /// <summary>How many times a setting change is reapplied before giving up.</summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    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)
         }
     }
 
+    /// <summary>
+    /// Applies a change to this checker's stored state, and reapplies it if something else wrote
+    /// first.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// <para>
+    /// The change runs once per attempt, against freshly read state each time, so it must not
+    /// depend on having seen the previous value.
+    /// </para>
+    /// <para>
+    /// <c>StateChanged</c> 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.
+    /// </para>
+    /// </remarks>
+    /// <returns><c>true</c> if anything was written; <c>false</c> if the change was a no-op.</returns>
+    private async Task<bool> UpdateStateAsync(Action<PulseCheckerState> 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();
+        }
+    }
+
+    /// <summary>
+    /// The read-modify-write loop itself, without the lock and without the event.
+    /// </summary>
+    /// <remarks>
+    /// Separate from <see cref="UpdateStateAsync"/> because <see cref="TriggerAsync"/> 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.
+    /// </remarks>
+    /// <returns>The state before and after, and whether anything was written.</returns>
+    private async Task<(PulseCheckerState OldState, PulseCheckerState NewState, bool Changed)> ApplyAsync(
+        Action<PulseCheckerState> apply,
+        CancellationToken cancellationToken)
+    {
+        for (var attempt = 1; ; attempt++)
+        {
+            var entry = await _stateProvider
+                .GetStateEntryAsync<PulseCheckerState>(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.");
+            }
+        }
+    }
+
     /// <inheritdoc />
     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);
     }
 
     /// <inheritdoc />
     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);
     }
 
     /// <inheritdoc />
@@ -270,56 +370,33 @@ public async Task SetTagsAsync(IReadOnlyList<string> 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);
     }
 
     /// <inheritdoc />
     public async Task SetGroupAsync(string? group, CancellationToken cancellationToken = default)
     {
         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);
     }
 
     /// <inheritdoc />
     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);
     }
 
     /// <inheritdoc />
     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);
     }
 
     /// <inheritdoc />
@@ -342,11 +419,7 @@ public async Task ClearHistoryAsync(CancellationToken cancellationToken = defaul
     /// <inheritdoc />
     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);
     }
 
     /// <inheritdoc />
@@ -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<PulseCheckerState>(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
     /// <inheritdoc />
     public async Task<bool> 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);
     }
 
     /// <inheritdoc />
     public async Task<bool> 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);
     }
 
     /// <summary>