fix(driver.sql): compare-and-swap the health field so a concurrent transition cannot be lost (Driver.Sql-010)
This commit is contained in:
@@ -686,7 +686,7 @@ public sealed class SqlDriver
|
|||||||
// ReinitializeAsync has recorded Faulted (the engine's ~5 s teardown wait is shorter than the 15 s
|
// ReinitializeAsync has recorded Faulted (the engine's ~5 s teardown wait is shorter than the 15 s
|
||||||
// default operationTimeout, so a stale poll can still complete afterwards) must not flip Faulted back
|
// default operationTimeout, so a stale poll can still complete afterwards) must not flip Faulted back
|
||||||
// to Healthy with no real recovery. Only a successful (re)initialize clears Faulted.
|
// to Healthy with no real recovery. Only a successful (re)initialize clears Faulted.
|
||||||
SetHealthUnlessFaulted(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
|
RecordHealthyRead();
|
||||||
TransitionHostTo(HostState.Running);
|
TransitionHostTo(HostState.Running);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -782,36 +782,70 @@ public sealed class SqlDriver
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="reason">The operator-facing reason, which must never carry the connection string.</param>
|
/// <param name="reason">The operator-facing reason, which must never carry the connection string.</param>
|
||||||
private void Degrade(string reason)
|
private void Degrade(string reason)
|
||||||
{
|
=> UpdateHealthUnlessFaulted(
|
||||||
var current = ReadHealth();
|
current => new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason));
|
||||||
SetHealthUnlessFaulted(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason));
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Publishes <paramref name="value"/> unless the driver is already <see cref="DriverState.Faulted"/>
|
/// Records a poll that returned at least one Good value: Healthy, with the evidence clock advanced.
|
||||||
/// — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize
|
/// <para>The timestamp is taken <b>inside</b> the transform, so a retried compare-and-swap publishes
|
||||||
/// health write routes through here: <see cref="Degrade"/> and the poll's Healthy verdict in
|
/// a fresh one rather than the one this call started with — which a concurrent advance could already
|
||||||
/// <see cref="ObservePollOutcome"/>, so a late in-flight poll cannot un-fault a driver a concurrent
|
/// have overtaken.</para>
|
||||||
/// <see cref="ReinitializeAsync"/> just faulted.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="value">The health snapshot to publish when the driver is not Faulted.</param>
|
private void RecordHealthyRead()
|
||||||
private void SetHealthUnlessFaulted(DriverHealth value)
|
=> UpdateHealthUnlessFaulted(
|
||||||
{
|
static _ => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
|
||||||
if (ReadHealth().State == DriverState.Faulted) return;
|
|
||||||
WriteHealth(value);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>Advances <c>LastSuccessfulRead</c> without changing the current state or error.</summary>
|
/// <summary>Advances <c>LastSuccessfulRead</c> without changing the current state or error.</summary>
|
||||||
private void TouchLastSuccessfulRead()
|
private void TouchLastSuccessfulRead()
|
||||||
|
=> UpdateHealth(static current => current with { LastSuccessfulRead = DateTime.UtcNow });
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Applies <paramref name="project"/> unless the driver is already <see cref="DriverState.Faulted"/>
|
||||||
|
/// — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize
|
||||||
|
/// state write routes through here: <see cref="Degrade"/> and the poll's Healthy verdict in
|
||||||
|
/// <see cref="ObservePollOutcome"/>, so a late in-flight poll cannot un-fault a driver a concurrent
|
||||||
|
/// <see cref="ReinitializeAsync"/> just faulted. The guard is evaluated against the <em>same</em>
|
||||||
|
/// snapshot the write then swaps out, so a Faulted verdict landing mid-update cannot be lost.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="project">Builds the snapshot to publish from the current one.</param>
|
||||||
|
private void UpdateHealthUnlessFaulted(Func<DriverHealth, DriverHealth> project)
|
||||||
|
=> UpdateHealth(current => current.State == DriverState.Faulted ? null : project(current));
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The one read-modify-write of the health field, as a compare-and-swap retry loop.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Driver.Sql-010. Health is transitioned from every poll loop (one per subscription) plus any
|
||||||
|
/// ad-hoc <see cref="ReadAsync"/>, and every transition needs the current snapshot to build the next
|
||||||
|
/// one — <see cref="Degrade"/> preserves <c>LastSuccessfulRead</c>, the Faulted guard reads the state.
|
||||||
|
/// A plain <c>Volatile</c> read followed by a <c>Volatile</c> write lets a concurrent update land in
|
||||||
|
/// between and be silently overwritten: an advance of the evidence clock the comms-loss watchdog
|
||||||
|
/// reads is lost (and the clock even moves backwards), or a Faulted verdict is downgraded despite the
|
||||||
|
/// guard. <paramref name="transform"/> is re-run on each attempt, so it must be side-effect-free and
|
||||||
|
/// must derive everything — timestamps included — from the snapshot it is handed or from the clock.
|
||||||
|
/// Returning null declines the update.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="transform">Builds the snapshot to publish, or returns null to leave health untouched.</param>
|
||||||
|
private void UpdateHealth(Func<DriverHealth, DriverHealth?> transform)
|
||||||
{
|
{
|
||||||
var current = ReadHealth();
|
while (true)
|
||||||
WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow });
|
{
|
||||||
|
var current = Volatile.Read(ref _health);
|
||||||
|
var next = transform(current);
|
||||||
|
if (next is null) return;
|
||||||
|
if (ReferenceEquals(Interlocked.CompareExchange(ref _health, next, current), current)) return;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>Barrier-protected read of the multi-threaded health field.</summary>
|
/// <summary>Barrier-protected read of the multi-threaded health field.</summary>
|
||||||
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
|
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
|
||||||
|
|
||||||
/// <summary>Barrier-protected publish of a new health snapshot.</summary>
|
/// <summary>
|
||||||
|
/// Unconditional publish of a new health snapshot, for the lifecycle transitions that are
|
||||||
|
/// last-writer-wins by definition (Initialize / Shutdown). Everything driven by a poll goes through
|
||||||
|
/// <see cref="UpdateHealth"/> instead, because it has to read the current snapshot first.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="value">The health snapshot to publish.</param>
|
||||||
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
|
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
|
||||||
|
|
||||||
/// <summary>Records a host-state change and raises <see cref="OnHostStatusChanged"/> — on transitions only.</summary>
|
/// <summary>Records a host-state change and raises <see cref="OnHostStatusChanged"/> — on transitions only.</summary>
|
||||||
|
|||||||
@@ -356,6 +356,55 @@ public sealed class SqlDriverTests
|
|||||||
driver.GetHealth().LastSuccessfulRead.ShouldBe(before);
|
driver.GetHealth().LastSuccessfulRead.ShouldBe(before);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Driver.Sql-010: the health field is read-modify-written from several poll loops ----
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task HealthTransitions_fromConcurrentPolls_neverRewindTheEvidenceClock()
|
||||||
|
{
|
||||||
|
// Driver.Sql-010. Every health transition is a read-modify-write of one field: Degrade reads the
|
||||||
|
// current snapshot to PRESERVE LastSuccessfulRead, then writes a new record. With a plain
|
||||||
|
// Volatile read + write, a Degrade that read the clock at t0 can be published after a concurrent
|
||||||
|
// advance to t1 — so the clock goes BACKWARDS, and the advance is lost. That is not cosmetic: the
|
||||||
|
// comms-loss watchdog measures how long the clock has been frozen, so a lost advance is a spurious
|
||||||
|
// 90 s reconnect. Monotonicity is the observable the lost update violates, and it is exactly what a
|
||||||
|
// compare-and-swap guarantees.
|
||||||
|
//
|
||||||
|
// A driver with several subscriptions runs one poll loop each, plus any ad-hoc ReadAsync — so two
|
||||||
|
// classifications landing at once is the ordinary case, not a contrived one.
|
||||||
|
using var fixture = new SqlitePollFixture();
|
||||||
|
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
|
||||||
|
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
|
||||||
|
|
||||||
|
const int iterations = 30_000;
|
||||||
|
var advancing = new DataValueSnapshot(1.0, SqlStatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow);
|
||||||
|
var degrading = new DataValueSnapshot(null, SqlStatusCodes.BadTimeout, DateTime.UtcNow, DateTime.UtcNow);
|
||||||
|
using var done = new CancellationTokenSource();
|
||||||
|
|
||||||
|
var rewinds = 0;
|
||||||
|
var observer = Task.Run(() =>
|
||||||
|
{
|
||||||
|
var highWater = driver.GetHealth().LastSuccessfulRead ?? DateTime.MinValue;
|
||||||
|
while (!done.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
var seen = driver.GetHealth().LastSuccessfulRead ?? DateTime.MinValue;
|
||||||
|
if (seen < highWater) Interlocked.Increment(ref rewinds);
|
||||||
|
else highWater = seen;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
var writers = new[]
|
||||||
|
{
|
||||||
|
Task.Run(() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([advancing]); }),
|
||||||
|
Task.Run(() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([degrading]); }),
|
||||||
|
};
|
||||||
|
|
||||||
|
await Task.WhenAll(writers);
|
||||||
|
await done.CancelAsync();
|
||||||
|
await observer;
|
||||||
|
|
||||||
|
Volatile.Read(ref rewinds).ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
// ---- ISubscribable (poll-engine overlay) ----
|
// ---- ISubscribable (poll-engine overlay) ----
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
Reference in New Issue
Block a user