Joseph Doherty
614 lines
26 KiB
C#
614 lines
26 KiB
C#
using System.Collections.Concurrent;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests;
|
|
|
|
[Trait("Category", "Unit")]
|
|
public sealed class PollGroupEngineTests
|
|
{
|
|
private sealed class FakeSource
|
|
{
|
|
/// <summary>Gets the dictionary of test values keyed by reference.</summary>
|
|
public ConcurrentDictionary<string, object?> Values { get; } = new();
|
|
|
|
/// <summary>Gets or sets the number of read operations performed.</summary>
|
|
public int ReadCount;
|
|
|
|
/// <summary>Simulates reading values from the source.</summary>
|
|
/// <param name="refs">The references to read.</param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task returning the data value snapshots.</returns>
|
|
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
|
IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref ReadCount);
|
|
var now = DateTime.UtcNow;
|
|
IReadOnlyList<DataValueSnapshot> snapshots = refs
|
|
.Select(r => Values.TryGetValue(r, out var v)
|
|
? new DataValueSnapshot(v, 0u, now, now)
|
|
: new DataValueSnapshot(null, 0x80340000u, null, now))
|
|
.ToList();
|
|
return Task.FromResult(snapshots);
|
|
}
|
|
}
|
|
|
|
/// <summary>Verifies that the initial poll forces an event for every subscribed tag.</summary>
|
|
[Fact]
|
|
public async Task Initial_poll_force_raises_every_subscribed_tag()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["A"] = 1;
|
|
src.Values["B"] = "hello";
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle h, string r, DataValueSnapshot s)>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue((h, r, s)));
|
|
|
|
var handle = engine.Subscribe(["A", "B"], TimeSpan.FromMilliseconds(200));
|
|
await WaitForAsync(() => events.Count >= 2, TimeSpan.FromSeconds(2));
|
|
|
|
events.Select(e => e.r).ShouldBe(["A", "B"], ignoreOrder: true);
|
|
engine.Unsubscribe(handle).ShouldBeTrue();
|
|
}
|
|
|
|
/// <summary>Verifies that unchanged values are only raised once.</summary>
|
|
[Fact]
|
|
public async Task Unchanged_value_raises_only_once()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 42;
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue((h, r, s)));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
await Task.Delay(500, TestContext.Current.CancellationToken);
|
|
engine.Unsubscribe(handle);
|
|
|
|
events.Count.ShouldBe(1);
|
|
}
|
|
|
|
/// <summary>Verifies that value changes raise new events.</summary>
|
|
[Fact]
|
|
public async Task Value_change_raises_new_event()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 1;
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue((h, r, s)));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
|
|
src.Values["X"] = 2;
|
|
await WaitForAsync(() => events.Count >= 2, TimeSpan.FromSeconds(2));
|
|
|
|
engine.Unsubscribe(handle);
|
|
events.Last().Item3.Value.ShouldBe(2);
|
|
}
|
|
|
|
/// <summary>Verifies that unsubscribe halts the polling loop.</summary>
|
|
[Fact]
|
|
public async Task Unsubscribe_halts_the_loop()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 1;
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue((h, r, s)));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
|
|
engine.Unsubscribe(handle).ShouldBeTrue();
|
|
var afterUnsub = events.Count;
|
|
|
|
src.Values["X"] = 999;
|
|
await Task.Delay(400, TestContext.Current.CancellationToken);
|
|
events.Count.ShouldBe(afterUnsub);
|
|
}
|
|
|
|
/// <summary>Verifies that intervals below the configured floor are clamped.</summary>
|
|
[Fact]
|
|
public async Task Interval_below_floor_is_clamped()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 1;
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue((h, r, s)),
|
|
minInterval: TimeSpan.FromMilliseconds(200));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(5));
|
|
await Task.Delay(300, TestContext.Current.CancellationToken);
|
|
engine.Unsubscribe(handle);
|
|
|
|
// 300 ms window, 200 ms floor, stable value → initial push + at most 1 extra poll.
|
|
// With zero changes only the initial-data push fires.
|
|
events.Count.ShouldBe(1);
|
|
}
|
|
|
|
/// <summary>Verifies that multiple subscriptions operate independently.</summary>
|
|
[Fact]
|
|
public async Task Multiple_subscriptions_are_independent()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["A"] = 1;
|
|
src.Values["B"] = 2;
|
|
|
|
var a = new ConcurrentQueue<string>();
|
|
var b = new ConcurrentQueue<string>();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) =>
|
|
{
|
|
if (r == "A") a.Enqueue(r);
|
|
else if (r == "B") b.Enqueue(r);
|
|
});
|
|
|
|
var ha = engine.Subscribe(["A"], TimeSpan.FromMilliseconds(100));
|
|
var hb = engine.Subscribe(["B"], TimeSpan.FromMilliseconds(100));
|
|
|
|
await WaitForAsync(() => a.Count >= 1 && b.Count >= 1, TimeSpan.FromSeconds(2));
|
|
engine.Unsubscribe(ha);
|
|
var aCount = a.Count;
|
|
src.Values["B"] = 77;
|
|
await WaitForAsync(() => b.Count >= 2, TimeSpan.FromSeconds(2));
|
|
|
|
a.Count.ShouldBe(aCount);
|
|
b.Count.ShouldBeGreaterThanOrEqualTo(2);
|
|
engine.Unsubscribe(hb);
|
|
}
|
|
|
|
/// <summary>Verifies that reader exceptions do not crash the polling loop.</summary>
|
|
[Fact]
|
|
public async Task Reader_exception_does_not_crash_loop()
|
|
{
|
|
var throwCount = 0;
|
|
var readCount = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
if (Interlocked.Increment(ref readCount) <= 2)
|
|
{
|
|
Interlocked.Increment(ref throwCount);
|
|
throw new InvalidOperationException("boom");
|
|
}
|
|
var now = DateTime.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
|
|
refs.Select(r => new DataValueSnapshot(1, 0u, now, now)).ToList());
|
|
}
|
|
|
|
var events = new ConcurrentQueue<string>();
|
|
await using var engine = new PollGroupEngine(Reader,
|
|
(h, r, s) => events.Enqueue(r));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(2));
|
|
engine.Unsubscribe(handle);
|
|
|
|
throwCount.ShouldBe(2);
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
}
|
|
|
|
/// <summary>Verifies that unsubscribing an unknown handle returns false.</summary>
|
|
[Fact]
|
|
public async Task Unsubscribe_unknown_handle_returns_false()
|
|
{
|
|
var src = new FakeSource();
|
|
await using var engine = new PollGroupEngine(src.ReadAsync, (_, _, _) => { });
|
|
|
|
var foreign = new DummyHandle();
|
|
engine.Unsubscribe(foreign).ShouldBeFalse();
|
|
}
|
|
|
|
/// <summary>Verifies that the active subscription count tracks lifecycle changes.</summary>
|
|
[Fact]
|
|
public async Task ActiveSubscriptionCount_tracks_lifecycle()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 1;
|
|
await using var engine = new PollGroupEngine(src.ReadAsync, (_, _, _) => { });
|
|
|
|
engine.ActiveSubscriptionCount.ShouldBe(0);
|
|
var h1 = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(200));
|
|
var h2 = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(200));
|
|
engine.ActiveSubscriptionCount.ShouldBe(2);
|
|
|
|
engine.Unsubscribe(h1);
|
|
engine.ActiveSubscriptionCount.ShouldBe(1);
|
|
engine.Unsubscribe(h2);
|
|
engine.ActiveSubscriptionCount.ShouldBe(0);
|
|
}
|
|
|
|
/// <summary>Verifies that disposing the engine cancels all active subscriptions.</summary>
|
|
[Fact]
|
|
public async Task DisposeAsync_cancels_all_subscriptions()
|
|
{
|
|
var src = new FakeSource();
|
|
src.Values["X"] = 1;
|
|
|
|
var events = new ConcurrentQueue<string>();
|
|
var engine = new PollGroupEngine(src.ReadAsync,
|
|
(h, r, s) => events.Enqueue(r));
|
|
|
|
_ = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
_ = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100));
|
|
await WaitForAsync(() => events.Count >= 2, TimeSpan.FromSeconds(2));
|
|
|
|
await engine.DisposeAsync();
|
|
engine.ActiveSubscriptionCount.ShouldBe(0);
|
|
|
|
var afterDispose = events.Count;
|
|
await Task.Delay(300, TestContext.Current.CancellationToken);
|
|
// After dispose no more events — everything is cancelled.
|
|
events.Count.ShouldBe(afterDispose);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-001: an array-valued tag whose contents are unchanged across polls
|
|
/// must fire only the initial change event, not a spurious event on every tick, even
|
|
/// when the driver produces a fresh array instance on each read.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Array_valued_tag_unchanged_contents_raises_only_once()
|
|
{
|
|
// Each read produces a new int[] instance with the same contents — reference equality
|
|
// would consider these different, structural equality must not.
|
|
var callCount = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref callCount);
|
|
var now = DateTime.UtcNow;
|
|
// Fresh array instance every call — same logical value.
|
|
IReadOnlyList<DataValueSnapshot> snaps = refs
|
|
.Select(_ => new DataValueSnapshot(new int[] { 1, 2, 3 }, 0u, now, now))
|
|
.ToList();
|
|
return Task.FromResult(snaps);
|
|
}
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(Reader,
|
|
(h, r, s) => events.Enqueue((h, r, s)),
|
|
minInterval: TimeSpan.FromMilliseconds(50));
|
|
|
|
var handle = engine.Subscribe(["A"], TimeSpan.FromMilliseconds(50));
|
|
// Allow several poll cycles so a broken implementation would accumulate extra events.
|
|
await Task.Delay(400, TestContext.Current.CancellationToken);
|
|
engine.Unsubscribe(handle);
|
|
|
|
// Only the initial-data push should have fired; subsequent polls with identical
|
|
// array contents must not produce additional events.
|
|
events.Count.ShouldBe(1);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-001: an array-valued tag whose contents change between polls
|
|
/// must fire a change event for each distinct set of contents.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Array_valued_tag_changed_contents_raises_event()
|
|
{
|
|
var generation = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
var gen = Interlocked.Increment(ref generation);
|
|
var now = DateTime.UtcNow;
|
|
IReadOnlyList<DataValueSnapshot> snaps = refs
|
|
.Select(_ => new DataValueSnapshot(new int[] { gen, gen + 1 }, 0u, now, now))
|
|
.ToList();
|
|
return Task.FromResult(snaps);
|
|
}
|
|
|
|
var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>();
|
|
await using var engine = new PollGroupEngine(Reader,
|
|
(h, r, s) => events.Enqueue((h, r, s)),
|
|
minInterval: TimeSpan.FromMilliseconds(50));
|
|
|
|
var handle = engine.Subscribe(["A"], TimeSpan.FromMilliseconds(50));
|
|
await WaitForAsync(() => events.Count >= 3, TimeSpan.FromSeconds(2));
|
|
engine.Unsubscribe(handle);
|
|
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(3);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-002: a reader that returns fewer snapshots than input references
|
|
/// violates the documented contract. The engine must throw a descriptive exception
|
|
/// rather than silently stalling.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Reader_short_result_list_raises_descriptive_exception_and_loop_continues()
|
|
{
|
|
var shortReadCount = 0;
|
|
var normalReadCount = 0;
|
|
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
if (Interlocked.Increment(ref shortReadCount) <= 2)
|
|
{
|
|
// Return fewer snapshots than refs — contract violation.
|
|
IReadOnlyList<DataValueSnapshot> bad = new List<DataValueSnapshot>();
|
|
return Task.FromResult(bad);
|
|
}
|
|
Interlocked.Increment(ref normalReadCount);
|
|
IReadOnlyList<DataValueSnapshot> good = refs
|
|
.Select(r => new DataValueSnapshot(42, 0u, now, now))
|
|
.ToList();
|
|
return Task.FromResult(good);
|
|
}
|
|
|
|
var events = new ConcurrentQueue<string>();
|
|
await using var engine = new PollGroupEngine(Reader,
|
|
(h, r, s) => events.Enqueue(r),
|
|
minInterval: TimeSpan.FromMilliseconds(50));
|
|
|
|
// Even though the first reads violate the contract the loop must survive and eventually
|
|
// deliver changes once the reader returns correct results.
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(50));
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(2));
|
|
engine.Unsubscribe(handle);
|
|
|
|
// At least one event must have arrived from the well-formed reads.
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
// The short-read counter confirms the contract-violating reads were attempted.
|
|
shortReadCount.ShouldBeGreaterThanOrEqualTo(2);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-005: the engine documents that "transient poll errors are logged on
|
|
/// the driver health surface", but until an error callback exists the driver has no way
|
|
/// to observe a caught reader exception. Subscribing without supplying an error callback
|
|
/// must continue to swallow exceptions (backward compatible). When an error callback IS
|
|
/// supplied, every exception caught during a poll cycle must be routed to it.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Reader_exception_is_reported_to_onError_callback()
|
|
{
|
|
var observed = new ConcurrentQueue<Exception>();
|
|
var readCount = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
if (Interlocked.Increment(ref readCount) <= 3)
|
|
throw new InvalidOperationException($"boom-{readCount}");
|
|
var now = DateTime.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
|
|
refs.Select(_ => new DataValueSnapshot(1, 0u, now, now)).ToList());
|
|
}
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, _, _) => { },
|
|
minInterval: TimeSpan.FromMilliseconds(50),
|
|
onError: ex => observed.Enqueue(ex));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(50));
|
|
await WaitForAsync(() => observed.Count >= 3, TimeSpan.FromSeconds(3));
|
|
engine.Unsubscribe(handle);
|
|
|
|
observed.Count.ShouldBeGreaterThanOrEqualTo(3);
|
|
observed.All(e => e is InvalidOperationException).ShouldBeTrue();
|
|
observed.All(e => e.Message.StartsWith("boom-")).ShouldBeTrue();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-005: a contract-violating reader (Core.Abstractions-002 path) that
|
|
/// throws the descriptive <see cref="InvalidOperationException"/> from inside the engine
|
|
/// must also be routed to the error callback so the driver health surface can observe
|
|
/// repeated contract violations.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Reader_contract_violation_routes_to_onError_callback()
|
|
{
|
|
var observed = new ConcurrentQueue<Exception>();
|
|
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
// Always return zero snapshots — short-result-list contract violation.
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(new List<DataValueSnapshot>());
|
|
}
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, _, _) => { },
|
|
minInterval: TimeSpan.FromMilliseconds(50),
|
|
onError: ex => observed.Enqueue(ex));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(50));
|
|
await WaitForAsync(() => observed.Count >= 2, TimeSpan.FromSeconds(2));
|
|
engine.Unsubscribe(handle);
|
|
|
|
observed.Count.ShouldBeGreaterThanOrEqualTo(2);
|
|
observed.All(e => e is InvalidOperationException
|
|
&& e.Message.Contains("Reader contract violation"))
|
|
.ShouldBeTrue();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Core.Abstractions-005: the engine must defend itself against an <c>onError</c> handler
|
|
/// that itself throws — otherwise a buggy health-surface forwarder would crash the poll
|
|
/// loop and silently stall the subscription, defeating the whole point of the callback.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task OnError_handler_that_throws_does_not_crash_loop()
|
|
{
|
|
var readCount = 0;
|
|
var events = new ConcurrentQueue<string>();
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
if (Interlocked.Increment(ref readCount) <= 2)
|
|
throw new InvalidOperationException("boom");
|
|
var now = DateTime.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
|
|
refs.Select(_ => new DataValueSnapshot(1, 0u, now, now)).ToList());
|
|
}
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, r, _) => events.Enqueue(r),
|
|
minInterval: TimeSpan.FromMilliseconds(50),
|
|
onError: _ => throw new ApplicationException("error-handler-bug"));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(50));
|
|
// Wait long enough for the reader to recover and for the engine to deliver a change.
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(3));
|
|
engine.Unsubscribe(handle);
|
|
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
}
|
|
|
|
// ---- PollGroupEngine v2: failure backoff + caller-token-filtered OCE (CONV-1 / STAB-8 / STAB-14) ----
|
|
|
|
/// <summary>
|
|
/// Sustained reader failures must stretch the poll cadence (capped exponential backoff)
|
|
/// instead of hammering a dead device at the base interval every tick.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Backoff_SustainedReaderFailures_StretchesCadenceToCap()
|
|
{
|
|
var readCount = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref readCount);
|
|
throw new InvalidOperationException("dead device");
|
|
}
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, _, _) => { },
|
|
minInterval: TimeSpan.FromMilliseconds(20),
|
|
onError: _ => { },
|
|
backoffCap: TimeSpan.FromMilliseconds(120));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(20));
|
|
await Task.Delay(700, TestContext.Current.CancellationToken);
|
|
engine.Unsubscribe(handle);
|
|
|
|
// Backoff schedule from a 20 ms base: 20,40,80,120,120,... → ≈8-9 reads in 700 ms.
|
|
// A no-backoff 20 ms loop would fire ≈35. Assert the cadence is clearly stretched.
|
|
readCount.ShouldBeGreaterThanOrEqualTo(2);
|
|
readCount.ShouldBeLessThan(18);
|
|
}
|
|
|
|
/// <summary>A successful poll resets the failure count so the cadence snaps back to the base interval.</summary>
|
|
[Fact]
|
|
public async Task Backoff_SuccessResetsToInterval()
|
|
{
|
|
var readCount = 0;
|
|
var generation = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
// First few reads fail (winding the backoff up toward the cap); thereafter every read
|
|
// returns a fresh changing value so each successful poll raises a change event.
|
|
if (Interlocked.Increment(ref readCount) <= 3)
|
|
throw new InvalidOperationException("warming up failures");
|
|
var gen = Interlocked.Increment(ref generation);
|
|
var now = DateTime.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
|
|
refs.Select(_ => new DataValueSnapshot(gen, 0u, now, now)).ToList());
|
|
}
|
|
|
|
var events = new ConcurrentQueue<string>();
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, r, _) => events.Enqueue(r),
|
|
minInterval: TimeSpan.FromMilliseconds(20),
|
|
onError: _ => { },
|
|
backoffCap: TimeSpan.FromMilliseconds(300));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(20));
|
|
// After recovery the cadence must be the fast 20 ms base, not the wound-up cap — so many
|
|
// change events accumulate quickly. A non-resetting impl stays near the 300 ms cap (≈3).
|
|
await WaitForAsync(() => events.Count >= 8, TimeSpan.FromSeconds(3));
|
|
engine.Unsubscribe(handle);
|
|
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(8);
|
|
}
|
|
|
|
/// <summary>
|
|
/// STAB-14-class engine guard: a reader that throws <see cref="OperationCanceledException"/>
|
|
/// WITHOUT the caller/loop token being cancelled (a driver-internal timeout CTS) must be
|
|
/// treated as a normal failure — reported + retried — NOT as a loop teardown. Before the
|
|
/// fix the bare <c>catch (OCE) { return; }</c> killed the whole subscription on the first
|
|
/// such OCE.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task ReaderOperationCanceled_WithoutCallerCancellation_TreatedAsFailureNotTeardown()
|
|
{
|
|
var observed = new ConcurrentQueue<Exception>();
|
|
var events = new ConcurrentQueue<string>();
|
|
var readCount = 0;
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
{
|
|
// A driver-internal timeout OCE — the loop token (ct) is NOT cancelled.
|
|
if (Interlocked.Increment(ref readCount) <= 2)
|
|
throw new OperationCanceledException("driver-internal read timeout");
|
|
var now = DateTime.UtcNow;
|
|
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
|
|
refs.Select(_ => new DataValueSnapshot(1, 0u, now, now)).ToList());
|
|
}
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, r, _) => events.Enqueue(r),
|
|
minInterval: TimeSpan.FromMilliseconds(30),
|
|
onError: ex => observed.Enqueue(ex),
|
|
backoffCap: TimeSpan.FromMilliseconds(120));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(30));
|
|
// The loop must SURVIVE the reader OCEs and eventually deliver a change once the reader
|
|
// recovers — proving the OCE was not treated as teardown.
|
|
await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(3));
|
|
engine.Unsubscribe(handle);
|
|
|
|
events.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
observed.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
observed.ShouldAllBe(e => e is OperationCanceledException);
|
|
}
|
|
|
|
/// <summary>Caller (loop-token) cancellation still exits the loop promptly — well under the drain timeout.</summary>
|
|
[Fact]
|
|
public async Task CallerCancellation_StillExitsPromptly()
|
|
{
|
|
// Reader blocks until the loop token is cancelled (a well-behaved cancellable read).
|
|
Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
|
|
=> Task.Delay(Timeout.Infinite, ct).ContinueWith(
|
|
_ => (IReadOnlyList<DataValueSnapshot>)new List<DataValueSnapshot>(), ct);
|
|
|
|
await using var engine = new PollGroupEngine(
|
|
Reader,
|
|
(_, _, _) => { },
|
|
minInterval: TimeSpan.FromMilliseconds(30),
|
|
onError: _ => { },
|
|
backoffCap: TimeSpan.FromMilliseconds(120));
|
|
|
|
var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(30));
|
|
await Task.Delay(60, TestContext.Current.CancellationToken); // let the reader begin blocking
|
|
|
|
var sw = System.Diagnostics.Stopwatch.StartNew();
|
|
engine.Unsubscribe(handle).ShouldBeTrue();
|
|
sw.Stop();
|
|
|
|
// Prompt exit: the caller-token OCE unwinds the loop immediately — nowhere near the 5 s
|
|
// StopState drain ceiling.
|
|
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(2));
|
|
}
|
|
|
|
private sealed record DummyHandle : ISubscriptionHandle
|
|
{
|
|
/// <summary>Gets a diagnostic identifier for this handle.</summary>
|
|
public string DiagnosticId => "dummy";
|
|
}
|
|
|
|
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
|
|
{
|
|
var deadline = DateTime.UtcNow + timeout;
|
|
while (!condition() && DateTime.UtcNow < deadline)
|
|
await Task.Delay(20);
|
|
}
|
|
}
|