fix(alarms): dedup reconcile/live duplicate broadcasts on the alarm feed (NEXT-03)
A periodic reconcile can synthesize a repair transition whose matching live transition is still buffered in the alarm lease; both then broadcast as indistinguishable duplicates on StreamAlarms and the dashboard hub. Nothing serializes the two paths, and a correct serialization needs a worker-side high-water mark on QueryActiveAlarms (proto + worker change + a stall path), so this closes the common case with a local best-effort dedup instead: both paths already carry the same worker-derived identity — the worker stamps record.TransitionTimestampUtc into both OnAlarmTransitionEvent's transition_timestamp and ActiveAlarmSnapshot.last_transition_timestamp — so ApplyTransition suppresses a live transition whose (timestamp, resulting state) the cache already carries from a repair. The Clear leg has no cache entry left to compare, so ApplyReconcile tombstones each synthesized Clear by the instance's original_raise_timestamp for one reconcile generation; a matching live Clear consumes the tombstone, while a new raise/clear cycle carries a newer raise timestamp and passes. Suppression fires only on a positive marker match — unset timestamps keep today's behavior — so the documented at-least-once consumer contract stands (gateway.md, Sessions.md updated in the same change). Tests: two new regressions drive the exact race through the GWC-26 harness (repair-then-buffered-live for Raise and for Clear, each with a genuine follow-up transition proving no over-suppression); GatewayAlarmMonitor suites 18/18.
This commit is contained in:
Joseph Doherty
@@ -34,6 +34,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
private readonly Dictionary<string, ActiveAlarmSnapshot> _alarms = new(StringComparer.Ordinal);
|
||||
private readonly List<Subscriber> _subscribers = [];
|
||||
|
||||
// NEXT-03 dedup tombstones, guarded by _sync: alarm instances whose Clear was synthesized by
|
||||
// the most recent reconcile pass, keyed by reference with the instance's original raise
|
||||
// timestamp as the identity marker. A buffered live Clear for the same instance is a duplicate
|
||||
// of the repair and is suppressed. One generation deep: each reconcile pass replaces the map,
|
||||
// so a tombstone lives at least one reconcile interval — far longer than the lease buffer the
|
||||
// duplicate would be sitting in — and the map stays bounded by the feed's churn per interval.
|
||||
private readonly Dictionary<string, Timestamp> _clearedByReconcile = new(StringComparer.Ordinal);
|
||||
|
||||
// Current provider status (mode + degraded + reason + since), guarded by _sync.
|
||||
// Initialized to the alarm-manager, not-degraded baseline so a late joiner sees
|
||||
// a sensible status even before any OnAlarmProviderModeChanged event arrives.
|
||||
@@ -413,17 +421,60 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
{
|
||||
if (transition.TransitionKind == AlarmTransitionKind.Clear)
|
||||
{
|
||||
_alarms.Remove(reference);
|
||||
bool wasKnown = _alarms.Remove(reference);
|
||||
if (!wasKnown && IsDuplicateOfReconcileClear(reference, transition))
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
_alarms[reference] = SnapshotFromTransition(transition);
|
||||
ActiveAlarmSnapshot snapshot = SnapshotFromTransition(transition);
|
||||
bool duplicate = _alarms.TryGetValue(reference, out ActiveAlarmSnapshot? existing)
|
||||
&& IsDuplicateOfCachedState(existing, snapshot);
|
||||
_alarms[reference] = snapshot;
|
||||
if (duplicate)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
Broadcast(new AlarmFeedMessage { Transition = transition }, reference);
|
||||
}
|
||||
}
|
||||
|
||||
// NEXT-03: best-effort dedup of the reconcile/live race. A reconcile that already synthesized
|
||||
// this transition as a feed repair left the cache carrying the worker's transition timestamp
|
||||
// and resulting state — both derived from the same worker-side value the live transition
|
||||
// carries — so an exact (timestamp, state) match means this live transition's outcome has
|
||||
// already been broadcast. Suppress only on a positive match: an unset timestamp on either
|
||||
// side keeps today's at-least-once behavior.
|
||||
private static bool IsDuplicateOfCachedState(ActiveAlarmSnapshot existing, ActiveAlarmSnapshot incoming)
|
||||
{
|
||||
return existing.LastTransitionTimestamp is not null
|
||||
&& incoming.LastTransitionTimestamp is not null
|
||||
&& existing.LastTransitionTimestamp.Equals(incoming.LastTransitionTimestamp)
|
||||
&& existing.CurrentState == incoming.CurrentState;
|
||||
}
|
||||
|
||||
// NEXT-03, the Clear leg. A reconcile Clear repair removes the cache entry before the buffered
|
||||
// live Clear drains, so there is no cached state to compare against; the tombstone recorded by
|
||||
// ApplyReconcile identifies the cleared instance by its original raise timestamp instead. The
|
||||
// match consumes the tombstone, so a genuinely new raise/clear cycle (which carries a newer
|
||||
// original raise timestamp) is never swallowed. Caller holds _sync.
|
||||
private bool IsDuplicateOfReconcileClear(string reference, OnAlarmTransitionEvent transition)
|
||||
{
|
||||
if (transition.OriginalRaiseTimestamp is not null
|
||||
&& _clearedByReconcile.TryGetValue(reference, out Timestamp? clearedInstance)
|
||||
&& clearedInstance.Equals(transition.OriginalRaiseTimestamp))
|
||||
{
|
||||
_clearedByReconcile.Remove(reference);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
// Handles the worker's provider-mode-change event: updates the stored provider
|
||||
// status, broadcasts it to every subscriber (provider status is global, not
|
||||
// alarm-scoped), records the switch metric, and forces a cache reconcile so the
|
||||
@@ -533,11 +584,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
//
|
||||
// Delivery semantics: feed repair transitions are AT-LEAST-ONCE, not exactly-once. A reconcile
|
||||
// reads the worker's current state while the corresponding live transition may still be
|
||||
// buffered in the alarm lease's channel; both then broadcast, and the two are indistinguishable
|
||||
// on the feed. This is inherent to the reconcile design and pre-dates the acked-state delta
|
||||
// (the Raise/Clear repair has always had it), since nothing serializes a reconcile against the
|
||||
// in-flight live stream. Consumers must therefore treat alarm state idempotently — apply a
|
||||
// transition as "set the alarm to this state", never as an increment or a toggle.
|
||||
// buffered in the alarm lease's channel; both would then broadcast, and the two are
|
||||
// indistinguishable on the feed, since nothing serializes a reconcile against the in-flight
|
||||
// live stream. ApplyTransition narrows that window with a best-effort dedup (NEXT-03): a live
|
||||
// transition whose worker timestamp and resulting state the cache already carries — or whose
|
||||
// Clear matches a tombstone recorded below — was already broadcast as a repair and is
|
||||
// suppressed. The dedup fires only on a positive marker match, so the contract stays
|
||||
// at-least-once: consumers must still treat alarm state idempotently — apply a transition as
|
||||
// "set the alarm to this state", never as an increment or a toggle.
|
||||
private void ApplyReconcile(IEnumerable<ActiveAlarmSnapshot> snapshots)
|
||||
{
|
||||
Dictionary<string, ActiveAlarmSnapshot> next = new(StringComparer.Ordinal);
|
||||
@@ -551,10 +605,19 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
|
||||
lock (_sync)
|
||||
{
|
||||
// Previous-generation tombstones have outlived the buffered live transitions they
|
||||
// guard against (one full reconcile interval); start this pass's generation fresh.
|
||||
_clearedByReconcile.Clear();
|
||||
|
||||
foreach (KeyValuePair<string, ActiveAlarmSnapshot> existing in _alarms)
|
||||
{
|
||||
if (!next.ContainsKey(existing.Key))
|
||||
{
|
||||
if (existing.Value.OriginalRaiseTimestamp is not null)
|
||||
{
|
||||
_clearedByReconcile[existing.Key] = existing.Value.OriginalRaiseTimestamp;
|
||||
}
|
||||
|
||||
Broadcast(
|
||||
new AlarmFeedMessage { Transition = TransitionFromSnapshot(existing.Value, AlarmTransitionKind.Clear) },
|
||||
existing.Key);
|
||||
|
||||
@@ -155,6 +155,133 @@ public sealed class GatewayAlarmMonitorAttachOrderTests
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// NEXT-03. A reconcile Raise repair applied while the matching live Raise is still
|
||||
/// buffered must not double-broadcast: the live transition carrying the same worker
|
||||
/// timestamp and resulting state the cache already holds is a duplicate and is suppressed.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task LiveTransitionMatchingReconcileRepair_IsSuppressed()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
await using FakeSessionManager sessions = new();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForSubscribeStartAsync(WaitTimeout);
|
||||
|
||||
// The reconcile sees the raised alarm first (its Raise repair broadcasts before this
|
||||
// reader attaches) and stamps the cache with the worker's transition timestamp.
|
||||
Timestamp raiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
|
||||
sessions.SetReconcileSnapshot(SnapshotAt(AlarmConditionState.Active, raiseTime, raiseTime));
|
||||
sessions.EmitEvent(ProviderModeProbe(1));
|
||||
await WaitUntilAsync(
|
||||
() => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference),
|
||||
WaitTimeout);
|
||||
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
|
||||
await snapshotComplete.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// The buffered live Raise drains with the same worker timestamp — a duplicate of the
|
||||
// repair. The follow-up Acknowledge with a newer timestamp is genuine and must pass.
|
||||
sessions.EmitEvent(TransitionAt(2, AlarmTransitionKind.Raise, raiseTime, raiseTime));
|
||||
Timestamp ackTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 12, 0, 5, TimeSpan.Zero));
|
||||
sessions.EmitEvent(TransitionAt(3, AlarmTransitionKind.Acknowledge, ackTime, raiseTime));
|
||||
|
||||
await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
|
||||
WaitTimeout);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
AlarmFeedMessage[] transitions = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
|
||||
.ToArray();
|
||||
AlarmFeedMessage single = Assert.Single(transitions);
|
||||
Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// NEXT-03, the Clear leg. A reconcile Clear repair removes the cache entry, so the
|
||||
/// buffered live Clear is deduped through the tombstone keyed on the instance's original
|
||||
/// raise timestamp — and a genuinely new raise/clear cycle is never swallowed.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task LiveClearMatchingReconcileClearRepair_IsSuppressed()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
await using FakeSessionManager sessions = new();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForSubscribeStartAsync(WaitTimeout);
|
||||
|
||||
Timestamp raiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 0, TimeSpan.Zero));
|
||||
sessions.SetReconcileSnapshot(SnapshotAt(AlarmConditionState.Active, raiseTime, raiseTime));
|
||||
sessions.EmitEvent(ProviderModeProbe(1));
|
||||
await WaitUntilAsync(
|
||||
() => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference),
|
||||
WaitTimeout);
|
||||
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
|
||||
await snapshotComplete.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// The worker no longer reports the alarm: the reconcile synthesizes the Clear repair and
|
||||
// tombstones the instance by its original raise timestamp.
|
||||
sessions.SetReconcileSnapshot();
|
||||
sessions.EmitEvent(ProviderModeProbe(2));
|
||||
await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Clear,
|
||||
WaitTimeout);
|
||||
|
||||
// The buffered live Clear for the SAME instance is a duplicate of the repair; the Raise
|
||||
// that follows starts a new instance and must pass.
|
||||
Timestamp clearTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 10, TimeSpan.Zero));
|
||||
sessions.EmitEvent(TransitionAt(3, AlarmTransitionKind.Clear, clearTime, raiseTime));
|
||||
Timestamp newRaiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 20, TimeSpan.Zero));
|
||||
sessions.EmitEvent(TransitionAt(4, AlarmTransitionKind.Raise, newRaiseTime, newRaiseTime));
|
||||
|
||||
await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Raise,
|
||||
WaitTimeout);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
AlarmTransitionKind[] kinds = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
|
||||
.Select(m => m.Transition.TransitionKind)
|
||||
.ToArray();
|
||||
Assert.Equal([AlarmTransitionKind.Clear, AlarmTransitionKind.Raise], kinds);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
private static GatewayAlarmMonitor CreateMonitor(FakeSessionManager sessions, GatewayMetrics metrics)
|
||||
{
|
||||
AlarmsOptions options = new()
|
||||
@@ -254,6 +381,31 @@ public sealed class GatewayAlarmMonitorAttachOrderTests
|
||||
SourceProvider = AlarmProviderMode.Alarmmgr,
|
||||
};
|
||||
|
||||
// Snapshot carrying the worker-side identity markers the NEXT-03 dedup compares on.
|
||||
private static ActiveAlarmSnapshot SnapshotAt(
|
||||
AlarmConditionState state,
|
||||
Timestamp lastTransition,
|
||||
Timestamp originalRaise)
|
||||
{
|
||||
ActiveAlarmSnapshot snapshot = Snapshot(state);
|
||||
snapshot.LastTransitionTimestamp = lastTransition;
|
||||
snapshot.OriginalRaiseTimestamp = originalRaise;
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
// Live transition with explicit worker timestamps, for driving the NEXT-03 dedup.
|
||||
private static MxEvent TransitionAt(
|
||||
ulong sequence,
|
||||
AlarmTransitionKind kind,
|
||||
Timestamp transitionTimestamp,
|
||||
Timestamp originalRaise)
|
||||
{
|
||||
MxEvent mxEvent = Transition(sequence, kind);
|
||||
mxEvent.OnAlarmTransition.TransitionTimestamp = transitionTimestamp;
|
||||
mxEvent.OnAlarmTransition.OriginalRaiseTimestamp = originalRaise;
|
||||
return mxEvent;
|
||||
}
|
||||
|
||||
private static async Task<AlarmFeedMessage> WaitForAsync(
|
||||
List<AlarmFeedMessage> received,
|
||||
Func<AlarmFeedMessage, bool> predicate,
|
||||
|
||||
Reference in New Issue
Block a user