feat(alarms): feed-level snapshot_status truncation frame on StreamAlarms
This commit is contained in:
Joseph Doherty
@@ -16,7 +16,8 @@ namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
|
||||
/// <summary>
|
||||
/// Carries the worker's truncated-fetch verdict across the gateway: worker
|
||||
/// reply payload → <see cref="GatewayAlarmMonitor"/> → the public
|
||||
/// <c>QueryActiveAlarms</c> stream.
|
||||
/// <c>QueryActiveAlarms</c> stream (per record) and the live
|
||||
/// <c>StreamAlarms</c> feed (the <c>snapshot_status</c> frame).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -145,6 +146,302 @@ public sealed class AlarmTruncationSignalTests
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The feed-level edge. A reconcile that flips the verdict to truncated
|
||||
/// pushes exactly one <c>snapshot_status</c> frame to every attached
|
||||
/// subscriber; a following reconcile carrying the same verdict pushes
|
||||
/// none. Without the edge test the frame could be emitted per reconcile —
|
||||
/// a status frame every few seconds forever, which consumers would learn
|
||||
/// to ignore.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ReconcileFlippingToTruncated_EmitsOneSnapshotStatusFrame()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
StubSessionManager sessions = new();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForReconcileAsync(WaitTimeout);
|
||||
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource attached = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, attached, streamCts.Token);
|
||||
await attached.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// The worker's next fetch comes back capped. A provider-mode event forces the reconcile
|
||||
// immediately, so the test never waits on the periodic timer; each probe also broadcasts a
|
||||
// provider_status frame, which is the barrier the assertions below count on.
|
||||
sessions.SnapshotTruncated = true;
|
||||
sessions.EmitEvent(ProviderModeProbe(1));
|
||||
|
||||
// A second, identical reconcile: same verdict, so no second frame. The third probe is only
|
||||
// a barrier — the monitor handles provider-mode events one at a time and awaits the
|
||||
// reconcile inside each, so its provider_status proves probe 2's reconcile has completed.
|
||||
sessions.EmitEvent(ProviderModeProbe(2));
|
||||
sessions.EmitEvent(ProviderModeProbe(3));
|
||||
await WaitUntilAsync(() => CountOf(received, AlarmFeedMessage.PayloadOneofCase.ProviderStatus) >= 4, WaitTimeout);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
bool[] verdicts = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotStatus)
|
||||
.Select(m => m.SnapshotStatus.Truncated)
|
||||
.ToArray();
|
||||
|
||||
// The first is this subscriber's open-time baseline; the second is the edge.
|
||||
Assert.Equal([false, true], verdicts);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The clearing edge. A fetch that comes back under the cap restores
|
||||
/// absence authority, and the feed says so — otherwise a consumer that
|
||||
/// saw the truncated frame would caveat its alarm set forever.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ReconcileClearingTruncation_EmitsTheClearingFrame()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
StubSessionManager sessions = new() { SnapshotTruncated = true };
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForReconcileAsync(WaitTimeout);
|
||||
await WaitUntilAsync(() => monitor.SnapshotTruncated, WaitTimeout);
|
||||
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource attached = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, attached, streamCts.Token);
|
||||
await attached.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
sessions.SnapshotTruncated = false;
|
||||
sessions.EmitEvent(ProviderModeProbe(1));
|
||||
|
||||
await WaitUntilAsync(
|
||||
() => CountOf(received, AlarmFeedMessage.PayloadOneofCase.SnapshotStatus) >= 2,
|
||||
WaitTimeout);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
bool[] verdicts = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotStatus)
|
||||
.Select(m => m.SnapshotStatus.Truncated)
|
||||
.ToArray();
|
||||
Assert.Equal([true, false], verdicts);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Late-joiner priming while the verdict is truncated. The frame must
|
||||
/// arrive after <c>provider_status</c> and before the cached
|
||||
/// <c>active_alarm</c> frames: a consumer that applies the snapshot as it
|
||||
/// streams needs the completeness caveat in hand before the records it
|
||||
/// qualifies, not after.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task SubscriberAttachingWhileTruncated_IsPrimedBeforeTheCachedSnapshot()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
StubSessionManager sessions = new()
|
||||
{
|
||||
SnapshotTruncated = true,
|
||||
Snapshots = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: true)],
|
||||
};
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForReconcileAsync(WaitTimeout);
|
||||
await WaitUntilAsync(() => monitor.SnapshotTruncated && monitor.CurrentAlarms.Count == 1, WaitTimeout);
|
||||
|
||||
AlarmFeedMessage[] preamble = await ReadPreambleAsync(monitor);
|
||||
|
||||
Assert.Equal(
|
||||
[
|
||||
AlarmFeedMessage.PayloadOneofCase.ProviderStatus,
|
||||
AlarmFeedMessage.PayloadOneofCase.SnapshotStatus,
|
||||
AlarmFeedMessage.PayloadOneofCase.ActiveAlarm,
|
||||
AlarmFeedMessage.PayloadOneofCase.SnapshotComplete,
|
||||
],
|
||||
preamble.Select(m => m.PayloadCase).ToArray());
|
||||
Assert.True(preamble[1].SnapshotStatus.Truncated);
|
||||
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The control for the priming frame: it is unconditional, so a
|
||||
/// subscriber attaching to a complete feed is told so explicitly rather
|
||||
/// than having to infer "no frame means not truncated" — an inference
|
||||
/// that is indistinguishable from a gateway that never sends the frame.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task SubscriberAttachingWhileComplete_IsPrimedWithTheFalseVerdict()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
StubSessionManager sessions = new();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForReconcileAsync(WaitTimeout);
|
||||
|
||||
AlarmFeedMessage[] preamble = await ReadPreambleAsync(monitor);
|
||||
|
||||
Assert.Equal(
|
||||
[
|
||||
AlarmFeedMessage.PayloadOneofCase.ProviderStatus,
|
||||
AlarmFeedMessage.PayloadOneofCase.SnapshotStatus,
|
||||
AlarmFeedMessage.PayloadOneofCase.SnapshotComplete,
|
||||
],
|
||||
preamble.Select(m => m.PayloadCase).ToArray());
|
||||
Assert.False(preamble[1].SnapshotStatus.Truncated);
|
||||
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A monitor restart drops the truncation verdict with the cache
|
||||
/// generation it describes. That drop is a completeness change like any
|
||||
/// other, so subscribers — which outlive the monitor's worker session —
|
||||
/// must see the clearing frame; otherwise a feed that silently re-seeds
|
||||
/// leaves them caveating a set that is no longer truncated.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task MonitorRestartClearingTruncation_EmitsTheClearingFrame()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
StubSessionManager sessions = new() { SnapshotTruncated = true };
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForReconcileAsync(WaitTimeout);
|
||||
await WaitUntilAsync(() => monitor.SnapshotTruncated, WaitTimeout);
|
||||
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource attached = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, attached, streamCts.Token);
|
||||
await attached.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// Ending the worker event stream faults the monitor lifecycle, which clears the cache.
|
||||
sessions.EndWorkerEventStream();
|
||||
|
||||
await WaitUntilAsync(
|
||||
() => CountOf(received, AlarmFeedMessage.PayloadOneofCase.SnapshotStatus) >= 2,
|
||||
WaitTimeout);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
// Only the first two: the supervisor re-opens the session after its backoff, and that
|
||||
// second lifecycle legitimately re-reports the truncated verdict.
|
||||
bool[] verdicts = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotStatus)
|
||||
.Select(m => m.SnapshotStatus.Truncated)
|
||||
.Take(2)
|
||||
.ToArray();
|
||||
Assert.Equal([true, false], verdicts);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
private static int CountOf(List<AlarmFeedMessage> received, AlarmFeedMessage.PayloadOneofCase payloadCase)
|
||||
{
|
||||
lock (received)
|
||||
{
|
||||
return received.Count(m => m.PayloadCase == payloadCase);
|
||||
}
|
||||
}
|
||||
|
||||
// Drains the monitor's feed into received (every frame, preamble included), signalling gate
|
||||
// once the subscriber is registered — that is, on its first frame.
|
||||
private static Task ReadFeedAsync(
|
||||
GatewayAlarmMonitor monitor,
|
||||
List<AlarmFeedMessage> received,
|
||||
TaskCompletionSource gate,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.Run(
|
||||
async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, cancellationToken))
|
||||
{
|
||||
lock (received)
|
||||
{
|
||||
received.Add(message);
|
||||
}
|
||||
|
||||
gate.TrySetResult();
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected when the test cancels the stream.
|
||||
}
|
||||
},
|
||||
CancellationToken.None);
|
||||
}
|
||||
|
||||
// Reads one subscriber's open-time preamble: everything up to and including snapshot_complete.
|
||||
private static async Task<AlarmFeedMessage[]> ReadPreambleAsync(GatewayAlarmMonitor monitor)
|
||||
{
|
||||
List<AlarmFeedMessage> preamble = [];
|
||||
using CancellationTokenSource streamCts = new(WaitTimeout);
|
||||
await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, streamCts.Token))
|
||||
{
|
||||
preamble.Add(message);
|
||||
if (message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return [.. preamble];
|
||||
}
|
||||
|
||||
// A no-op provider-mode event. The monitor forces an immediate reconcile after every one,
|
||||
// which is how these tests drive a reconcile pass without waiting on the periodic timer.
|
||||
private static MxEvent ProviderModeProbe(ulong sequence) => new()
|
||||
{
|
||||
Family = MxEventFamily.OnAlarmProviderModeChanged,
|
||||
WorkerSequence = sequence,
|
||||
OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent
|
||||
{
|
||||
Mode = AlarmProviderMode.Alarmmgr,
|
||||
Reason = "probe",
|
||||
At = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
||||
},
|
||||
};
|
||||
|
||||
private static ActiveAlarmSnapshot NewSnapshot(string reference, bool fromTruncatedSnapshot)
|
||||
{
|
||||
return new ActiveAlarmSnapshot
|
||||
@@ -226,18 +523,43 @@ public sealed class AlarmTruncationSignalTests
|
||||
private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
|
||||
private readonly TaskCompletionSource _reconciled =
|
||||
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly object _sync = new();
|
||||
private bool _snapshotTruncated;
|
||||
private IReadOnlyList<ActiveAlarmSnapshot> _snapshots = [];
|
||||
|
||||
/// <summary>Gets or sets the truncation verdict the scripted reply carries.</summary>
|
||||
public bool SnapshotTruncated { get; init; }
|
||||
/// <summary>
|
||||
/// Gets or sets the truncation verdict the scripted reply carries. Settable mid-test so
|
||||
/// a suite can drive the verdict across reconciles and observe the feed-level edge.
|
||||
/// </summary>
|
||||
public bool SnapshotTruncated
|
||||
{
|
||||
get { lock (_sync) { return _snapshotTruncated; } }
|
||||
set { lock (_sync) { _snapshotTruncated = value; } }
|
||||
}
|
||||
|
||||
/// <summary>Gets or sets the snapshots the scripted reply carries.</summary>
|
||||
public IReadOnlyList<ActiveAlarmSnapshot> Snapshots { get; init; } = [];
|
||||
public IReadOnlyList<ActiveAlarmSnapshot> Snapshots
|
||||
{
|
||||
get { lock (_sync) { return _snapshots; } }
|
||||
set { lock (_sync) { _snapshots = value; } }
|
||||
}
|
||||
|
||||
/// <summary>Completes once the monitor has issued its first QueryActiveAlarms.</summary>
|
||||
/// <param name="timeout">The maximum time to wait.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public Task WaitForReconcileAsync(TimeSpan timeout) => _reconciled.Task.WaitAsync(timeout);
|
||||
|
||||
/// <summary>Pushes a worker event into the session's distributor pump.</summary>
|
||||
/// <param name="mxEvent">The event to push.</param>
|
||||
public void EmitEvent(MxEvent mxEvent) =>
|
||||
_events.Writer.TryWrite(new WorkerEvent { Event = mxEvent });
|
||||
|
||||
/// <summary>
|
||||
/// Ends the worker event stream, which faults the monitor's lifecycle exactly as a
|
||||
/// worker exit would and takes it through its cache-clearing teardown.
|
||||
/// </summary>
|
||||
public void EndWorkerEventStream() => _events.Writer.TryComplete();
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<GatewaySession> OpenSessionAsync(
|
||||
SessionOpenRequest request,
|
||||
@@ -275,8 +597,13 @@ public sealed class AlarmTruncationSignalTests
|
||||
|
||||
if (command.Command?.Kind == MxCommandKind.QueryActiveAlarms)
|
||||
{
|
||||
QueryActiveAlarmsReplyPayload payload = new() { SnapshotTruncated = SnapshotTruncated };
|
||||
payload.Snapshots.AddRange(Snapshots.Select(snapshot => snapshot.Clone()));
|
||||
QueryActiveAlarmsReplyPayload payload = new();
|
||||
lock (_sync)
|
||||
{
|
||||
payload.SnapshotTruncated = _snapshotTruncated;
|
||||
payload.Snapshots.AddRange(_snapshots.Select(snapshot => snapshot.Clone()));
|
||||
}
|
||||
|
||||
reply.QueryActiveAlarms = payload;
|
||||
_reconciled.TrySetResult();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user