Joseph Doherty
503 lines
23 KiB
C#
503 lines
23 KiB
C#
using System.Collections.Concurrent;
|
|
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
|
|
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
|
|
|
|
/// <summary>
|
|
/// Tests for <see cref="SiteAlarmAggregatorActor"/> (plan #10, Task 4) — the transient,
|
|
/// in-memory, per-site live alarm aggregator. Covers the tricky bits: seed-then-stream
|
|
/// ordering (a delta arriving before the seed completes is neither lost nor double-applied),
|
|
/// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect
|
|
/// re-seed, and periodic reconcile authoritative-replace (drift correction).
|
|
/// </summary>
|
|
public class SiteAlarmAggregatorActorTests : TestKit
|
|
{
|
|
private const string SiteId = "site-alpha";
|
|
private const string GrpcNodeA = "http://localhost:5100";
|
|
private const string GrpcNodeB = "http://localhost:5200";
|
|
private const string Instance = "Site1.Pump01";
|
|
|
|
public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING")
|
|
{
|
|
}
|
|
|
|
// Former process-global static test seams, now ctor-injected per actor (R2 T12).
|
|
private static readonly TimeSpan TestReconnectDelay = TimeSpan.FromMilliseconds(50);
|
|
private static readonly TimeSpan TestStabilityWindow = TimeSpan.FromSeconds(30);
|
|
|
|
// ── Test doubles ────────────────────────────────────────────────────────────
|
|
|
|
/// <summary>
|
|
/// Controllable seed fan-out: each invocation returns a fresh Task the test completes
|
|
/// on demand (so seed/reconcile timing relative to buffered deltas is deterministic).
|
|
/// </summary>
|
|
private sealed class SeedStub
|
|
{
|
|
private readonly ConcurrentQueue<TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>> _pending = new();
|
|
private int _callCount;
|
|
|
|
public int CallCount => Volatile.Read(ref _callCount);
|
|
|
|
public Task<IReadOnlyList<AlarmStateChanged>> Seed(CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref _callCount);
|
|
var tcs = new TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
ct.Register(() => tcs.TrySetCanceled());
|
|
_pending.Enqueue(tcs);
|
|
return tcs.Task;
|
|
}
|
|
|
|
/// <summary>Completes the oldest not-yet-completed seed call with the given rows.</summary>
|
|
public void CompleteNext(params AlarmStateChanged[] rows)
|
|
{
|
|
SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
|
|
if (_pending.TryDequeue(out var tcs))
|
|
tcs.TrySetResult(rows);
|
|
else
|
|
throw new InvalidOperationException("No pending seed to complete.");
|
|
}
|
|
|
|
/// <summary>Faults the oldest not-yet-completed seed call (simulates a whole-fan-out failure).</summary>
|
|
public void FaultNext()
|
|
{
|
|
SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
|
|
if (_pending.TryDequeue(out var tcs))
|
|
tcs.TrySetException(new InvalidOperationException("seed boom"));
|
|
else
|
|
throw new InvalidOperationException("No pending seed to fault.");
|
|
}
|
|
}
|
|
|
|
private sealed class PublishSink
|
|
{
|
|
private readonly object _lock = new();
|
|
public List<IReadOnlyList<AlarmStateChanged>> Snapshots { get; } = new();
|
|
|
|
public void Publish(IReadOnlyList<AlarmStateChanged> snapshot)
|
|
{
|
|
lock (_lock) { Snapshots.Add(snapshot); }
|
|
}
|
|
|
|
public IReadOnlyList<AlarmStateChanged>? Latest
|
|
{
|
|
get { lock (_lock) { return Snapshots.Count == 0 ? null : Snapshots[^1]; } }
|
|
}
|
|
|
|
public int Count { get { lock (_lock) { return Snapshots.Count; } } }
|
|
}
|
|
|
|
private sealed record SiteSub(
|
|
string CorrelationId, Action<AlarmStateChanged> OnAlarm, Action<Exception> OnError, CancellationToken Ct);
|
|
|
|
private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
|
|
{
|
|
private readonly object _lock = new();
|
|
private readonly List<SiteSub> _subs = new();
|
|
private readonly List<string> _unsubscribed = new();
|
|
|
|
public List<SiteSub> Subs { get { lock (_lock) { return _subs.ToList(); } } }
|
|
public List<string> Unsubscribed { get { lock (_lock) { return _unsubscribed.ToList(); } } }
|
|
|
|
public MockSiteAlarmStreamClient() : base() { }
|
|
|
|
public override Task SubscribeSiteAsync(
|
|
string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError, CancellationToken ct)
|
|
{
|
|
lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, ct)); }
|
|
var tcs = new TaskCompletionSource();
|
|
ct.Register(() => tcs.TrySetResult());
|
|
return tcs.Task; // never completes until cancelled (simulates a live stream)
|
|
}
|
|
|
|
public override void Unsubscribe(string correlationId)
|
|
{
|
|
lock (_lock) { _unsubscribed.Add(correlationId); }
|
|
}
|
|
}
|
|
|
|
private sealed class MockSiteAlarmStreamClientFactory : SiteStreamGrpcClientFactory
|
|
{
|
|
private readonly ConcurrentDictionary<string, MockSiteAlarmStreamClient> _byEndpoint = new();
|
|
|
|
public MockSiteAlarmStreamClientFactory()
|
|
: base(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance) { }
|
|
|
|
public MockSiteAlarmStreamClient ClientFor(string endpoint) =>
|
|
_byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient());
|
|
|
|
public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
|
|
=> ClientFor(grpcEndpoint);
|
|
|
|
public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint)
|
|
=> _byEndpoint.TryGetValue(grpcEndpoint, out var c) ? c : null;
|
|
}
|
|
|
|
private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor(
|
|
TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null)
|
|
{
|
|
var seed = new SeedStub();
|
|
var sink = new PublishSink();
|
|
var factory = new MockSiteAlarmStreamClientFactory();
|
|
|
|
var props = Props.Create(() => new SiteAlarmAggregatorActor(
|
|
SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
|
|
reconcileInterval ?? TimeSpan.FromMinutes(10),
|
|
publishCoalesce ?? TimeSpan.Zero,
|
|
TestReconnectDelay,
|
|
TestStabilityWindow));
|
|
|
|
var actor = Sys.ActorOf(props);
|
|
return (actor, seed, sink, factory);
|
|
}
|
|
|
|
private static AlarmStateChanged Alarm(
|
|
string alarmName, string sourceRef, int priority, DateTimeOffset ts,
|
|
bool placeholder = false, string instance = Instance) =>
|
|
new(instance, alarmName, AlarmState.Active, priority, ts)
|
|
{
|
|
SourceReference = sourceRef,
|
|
IsConfiguredPlaceholder = placeholder,
|
|
Kind = placeholder ? AlarmKind.NativeOpcUa : AlarmKind.Computed
|
|
};
|
|
|
|
private static AlarmStateChanged? Find(IReadOnlyList<AlarmStateChanged>? rows, string alarmName, string sourceRef) =>
|
|
rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef);
|
|
|
|
// ── Tests ───────────────────────────────────────────────────────────────────
|
|
|
|
[Fact]
|
|
public void Delta_Arriving_Before_Seed_Completes_Is_Buffered_And_Applied_Exactly_Once()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
// Live delta arrives DURING the seed window (new key, not in the seed).
|
|
var t1 = DateTimeOffset.UtcNow;
|
|
actor.Tell(Alarm("Overheat", "", 900, t1));
|
|
|
|
// Still buffering — nothing published yet.
|
|
Assert.Equal(0, sink.Count);
|
|
|
|
// Seed completes with a DIFFERENT alarm.
|
|
seed.CompleteNext(Alarm("PumpFault", "", 500, t1.AddSeconds(-1)));
|
|
|
|
AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
|
|
var latest = sink.Latest!;
|
|
Assert.NotNull(Find(latest, "PumpFault", "")); // seeded
|
|
var overheat = Find(latest, "Overheat", ""); // buffered gap-window delta survived
|
|
Assert.NotNull(overheat);
|
|
Assert.Single(latest, r => r.AlarmName == "Overheat"); // exactly once
|
|
}
|
|
|
|
[Fact]
|
|
public void Buffered_Delta_Older_Than_Seed_Entry_Is_Dropped()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
var tSeed = DateTimeOffset.UtcNow;
|
|
// Buffered delta for the SAME key but OLDER than the seed entry.
|
|
actor.Tell(Alarm("PumpFault", "", 100, tSeed.AddSeconds(-1)));
|
|
seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));
|
|
|
|
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
|
|
// Give a beat to ensure no late (dropped) delta sneaks in.
|
|
Thread.Sleep(150);
|
|
var latest = sink.Latest!;
|
|
var pumpFault = Find(latest, "PumpFault", "");
|
|
Assert.NotNull(pumpFault);
|
|
Assert.Equal(500, pumpFault!.Priority); // seed value kept; older buffered delta dropped
|
|
}
|
|
|
|
[Fact]
|
|
public void Buffered_Delta_Newer_Than_Seed_Entry_Is_Applied()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
var tSeed = DateTimeOffset.UtcNow;
|
|
actor.Tell(Alarm("PumpFault", "", 900, tSeed.AddSeconds(1))); // newer than seed
|
|
seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));
|
|
|
|
AwaitCondition(() => sink.Latest is { } l && Find(l, "PumpFault", "")?.Priority == 900,
|
|
TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
[Fact]
|
|
public void Dedup_By_AlarmKey_Distinguishes_SourceReference()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(); // empty seed
|
|
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
|
|
|
|
var t = DateTimeOffset.UtcNow;
|
|
// Same instance + alarm name, different source reference → two distinct rows.
|
|
actor.Tell(Alarm("LevelAlarm", "Tank01.Level.HiHi", 700, t));
|
|
actor.Tell(Alarm("LevelAlarm", "Tank01.Level.LoLo", 300, t));
|
|
|
|
AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
|
|
var latest = sink.Latest!;
|
|
Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.HiHi"));
|
|
Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.LoLo"));
|
|
}
|
|
|
|
[Fact]
|
|
public void Same_Key_Live_Update_Replaces_In_Place()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext();
|
|
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
|
|
|
|
var t = DateTimeOffset.UtcNow;
|
|
actor.Tell(Alarm("PumpFault", "", 500, t));
|
|
AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 500, TimeSpan.FromSeconds(3));
|
|
|
|
// Newer update for the same key replaces (not duplicates).
|
|
actor.Tell(Alarm("PumpFault", "", 800, t.AddSeconds(1)));
|
|
AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 800, TimeSpan.FromSeconds(3));
|
|
Assert.Single(sink.Latest!); // still exactly one row for the key
|
|
}
|
|
|
|
[Fact]
|
|
public void Seeded_Native_Alarm_And_Its_Live_Delta_Collapse_Onto_One_Key()
|
|
{
|
|
// Identity parity across the two paths (snapshot seed vs live stream): a native
|
|
// alarm seeded with a non-empty SourceReference and a later live delta for the
|
|
// SAME (instance, alarmName, sourceRef) must collapse to ONE in-place-updated row,
|
|
// not a ghost duplicate. Both paths key on SourceReference, so they must agree.
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
var t = DateTimeOffset.UtcNow;
|
|
seed.CompleteNext(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 300, t));
|
|
AwaitCondition(() => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi") is not null,
|
|
TimeSpan.FromSeconds(3));
|
|
|
|
// Live delta for the exact same identity, newer → updates in place.
|
|
actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));
|
|
|
|
AwaitCondition(
|
|
() => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi")?.Priority == 800,
|
|
TimeSpan.FromSeconds(3));
|
|
Assert.Single(sink.Latest!); // exactly one row — no seed/live ghost duplicate
|
|
}
|
|
|
|
[Fact]
|
|
public void Placeholder_Seeded_Coexists_With_Live_Real_Alarm_And_Is_Not_Wiped()
|
|
{
|
|
var (actor, seed, sink, _) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
var t = DateTimeOffset.UtcNow;
|
|
// Seed carries a configured-placeholder row (binding with no active conditions).
|
|
var placeholder = Alarm("Motor1.MotorAlarms", "", 0, t, placeholder: true);
|
|
seed.CompleteNext(placeholder);
|
|
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
|
|
|
|
// A live REAL condition for the same instance arrives (distinct AlarmKey).
|
|
actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));
|
|
|
|
AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
|
|
var latest = sink.Latest!;
|
|
// Placeholder still present (not wiped by the real alarm under a different key).
|
|
var ph = Find(latest, "Motor1.MotorAlarms", "");
|
|
Assert.NotNull(ph);
|
|
Assert.True(ph!.IsConfiguredPlaceholder);
|
|
Assert.NotNull(Find(latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi"));
|
|
}
|
|
|
|
[Fact]
|
|
public void GrpcError_Flips_Node_Reconnects_And_ReSeeds()
|
|
{
|
|
var (_, seed, sink, factory) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(); // initial seed done
|
|
// Wait for the seed to actually be APPLIED (published) so the re-seed on error is
|
|
// not skipped as "fan-out already in flight".
|
|
AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
|
|
|
|
// Fail the NodeA stream.
|
|
factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("NodeA down"));
|
|
|
|
// Reconnect must reach NodeB, AND a re-seed fan-out must have been triggered.
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
|
|
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
|
|
// The failed NodeA stream was unsubscribed (relay released, not left zombie).
|
|
Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed);
|
|
}
|
|
|
|
[Fact]
|
|
public void Periodic_Reconcile_Authoritatively_Replaces_Stale_Rows()
|
|
{
|
|
var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
var t = DateTimeOffset.UtcNow;
|
|
seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // seed has a row
|
|
AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));
|
|
|
|
// Reconcile tick fires → seed #2 (e.g. instance was disabled) returns EMPTY.
|
|
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(); // empty
|
|
|
|
// The stale PumpFault row must disappear (authoritative replace, not merge).
|
|
AwaitCondition(() => sink.Latest is { } l && l.Count == 0, TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
[Fact]
|
|
public void Failed_Initial_Seed_Does_Not_Publish_But_A_Later_Reconcile_Recovers()
|
|
{
|
|
var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
|
|
// The initial seed fan-out throws as a whole.
|
|
seed.FaultNext();
|
|
|
|
// A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet).
|
|
Thread.Sleep(200);
|
|
Assert.Equal(0, sink.Count);
|
|
|
|
// The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared
|
|
// on failure), completes, and the actor publishes → now live.
|
|
AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(Alarm("PumpFault", "", 500, DateTimeOffset.UtcNow));
|
|
AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
[Fact]
|
|
public void Stream_GivenUp_After_MaxRetries_Reopens_On_Reconcile_Tick()
|
|
{
|
|
// After the retry budget is exhausted the live stream is left down, but the
|
|
// aggregator is NOT stopped — the next reconcile tick self-heals it (resets the
|
|
// budget + reopens) so a sustained site outage never permanently kills the feed.
|
|
var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300));
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
|
|
// Drive initial + 3 retries (each flips node) to exhaust the budget.
|
|
factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
|
|
factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
|
|
factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));
|
|
|
|
int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
|
|
var before = TotalSubs(); // 4
|
|
|
|
// Fourth error exceeds MaxRetries → stream given up (no immediate reopen).
|
|
factory.ClientFor(GrpcNodeB).Subs[1].OnError(new Exception("4"));
|
|
|
|
// The reconcile tick reopens the stream (self-heal).
|
|
AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
// ── R2 T11: queued re-seed after reconnect + stream-generation stamp (N7.1/N7.2) ──
|
|
|
|
[Fact]
|
|
public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped()
|
|
{
|
|
var (actor, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(); // initial seed done (CallCount 1)
|
|
AwaitAssert(() => Assert.Equal(1, sink.Count));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
|
|
actor.Tell(new RunReconcile()); // reconcile fan-out now in flight (CallCount 2)
|
|
AwaitAssert(() => Assert.Equal(2, seed.CallCount));
|
|
|
|
// Stream error while the reconcile is in flight → the failover re-seed must not be
|
|
// silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until the next
|
|
// 60s reconcile tick.
|
|
factory.ClientFor(GrpcNodeA).Subs.Last().OnError(new Exception("stream fault"));
|
|
|
|
seed.CompleteNext(); // finish the in-flight reconcile
|
|
AwaitAssert(() => Assert.Equal(3, seed.CallCount)); // queued re-seed ran immediately after
|
|
}
|
|
|
|
[Fact]
|
|
public void LateErrorFromAPreviousStreamGeneration_IsIgnored()
|
|
{
|
|
var (_, seed, _, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext();
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
|
|
int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
|
|
|
|
var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single();
|
|
firstSub.OnError(new Exception("real fault")); // gen 1 error → flip, reconnect
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); // gen 2 stream open
|
|
AwaitAssert(() => Assert.Equal(2, TotalSubs()));
|
|
|
|
firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in
|
|
// Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored.
|
|
Thread.Sleep(400);
|
|
Assert.Equal(2, TotalSubs());
|
|
}
|
|
|
|
// ── R2 T10: live-delta publish coalescing (N6) ──
|
|
|
|
[Fact]
|
|
public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow()
|
|
{
|
|
var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.FromMilliseconds(150));
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext(); // empty initial seed
|
|
AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate
|
|
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
|
|
var now = DateTimeOffset.UtcNow;
|
|
for (var i = 0; i < 50; i++)
|
|
sub.OnAlarm(Alarm($"A{i}", "", 900, now)); // 50 distinct keys, one burst
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
Assert.Equal(50, sink.Latest!.Count); // nothing lost
|
|
Assert.InRange(sink.Count, 2, 4); // pre-fix: 51 publishes (1 seed + 1 per delta)
|
|
}, TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
[Fact]
|
|
public void ZeroCoalesce_PreservesLegacyPerDeltaPublish()
|
|
{
|
|
var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext();
|
|
AwaitAssert(() => Assert.Equal(1, sink.Count));
|
|
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
|
|
var now = DateTimeOffset.UtcNow;
|
|
sub.OnAlarm(Alarm("A1", "", 900, now));
|
|
sub.OnAlarm(Alarm("A2", "", 900, now));
|
|
AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta
|
|
}
|
|
|
|
[Fact]
|
|
public void Stop_Message_TearsDown_Grpc_And_Stops_Actor()
|
|
{
|
|
var (actor, seed, _, factory) = CreateActor();
|
|
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
|
|
seed.CompleteNext();
|
|
|
|
Watch(actor);
|
|
actor.Tell(new StopSiteAlarmAggregator());
|
|
|
|
ExpectTerminated(actor, TimeSpan.FromSeconds(3));
|
|
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Unsubscribed.Contains("corr-1"),
|
|
TimeSpan.FromSeconds(3));
|
|
}
|
|
}
|