fix(sitestream): deathwatch resets live-cache liveness on aggregator termination (plan R2-07 T10)

This commit is contained in:
Joseph Doherty
2026-07-13 11:01:54 -04:00
parent 76ef97f729
commit 5b16429635
4 changed files with 97 additions and 2 deletions
+1 -1
View File
@@ -191,7 +191,7 @@ Per-leaf alarm rendering (leaf nodes are individual conditions for native alarms
- **Live updates** — the page is driven by a **transient, per-site central live alarm cache** (`ISiteAlarmLiveCache`, owned by the Communication component; see [Component-Communication](Component-Communication.md)). On site select the page subscribes to the cache; the cache runs one shared, reference-counted per-site aggregator that **seeds** from the snapshot fan-out and then stays warm on a single **site-wide, alarm-only** `SubscribeSite` gRPC stream (seed-then-stream, dedup by `(InstanceUniqueName, AlarmName, SourceReference)`). Applied deltas raise an in-process change event (mirroring `IDeploymentStatusNotifier`) that the Blazor circuit pushes to the browser via `StateHasChanged()` — no new SignalR hub. `AlarmSummaryService.BuildFromLiveAlarms` rebuilds the roll-up + rows from the cache's current alarm set. The cache is **purely in-memory on the active central node** — there is still **no persisted central alarm store**; on a NodeA↔NodeB failover the new active node re-seeds from scratch. - **Live updates** — the page is driven by a **transient, per-site central live alarm cache** (`ISiteAlarmLiveCache`, owned by the Communication component; see [Component-Communication](Component-Communication.md)). On site select the page subscribes to the cache; the cache runs one shared, reference-counted per-site aggregator that **seeds** from the snapshot fan-out and then stays warm on a single **site-wide, alarm-only** `SubscribeSite` gRPC stream (seed-then-stream, dedup by `(InstanceUniqueName, AlarmName, SourceReference)`). Applied deltas raise an in-process change event (mirroring `IDeploymentStatusNotifier`) that the Blazor circuit pushes to the browser via `StateHasChanged()` — no new SignalR hub. `AlarmSummaryService.BuildFromLiveAlarms` rebuilds the roll-up + rows from the cache's current alarm set. The cache is **purely in-memory on the active central node** — there is still **no persisted central alarm store**; on a NodeA↔NodeB failover the new active node re-seeds from scratch.
- **View** — roll-up tiles (total active, worst severity, unacked count, per-`AlarmKind` counts) plus a flat, sortable, filterable table. Filters cover instance, `AlarmKind` (Computed / NativeOpcUa / NativeMxAccess), state, acked/unacked, severity threshold, and name search. - **View** — roll-up tiles (total active, worst severity, unacked count, per-`AlarmKind` counts) plus a flat, sortable, filterable table. Filters cover instance, `AlarmKind` (Computed / NativeOpcUa / NativeMxAccess), state, acked/unacked, severity threshold, and name search.
- **Read-only** — there are no ack / shelve / suppress controls (native alarms remain read-only by design). - **Read-only** — there are no ack / shelve / suppress controls (native alarms remain read-only by design).
- **Refresh** — manual refresh button plus the 15s poll timer (mirroring the Health dashboard), now retained as a **fallback + `NotReporting` authority** behind the live cache: when the cache reports `IsLive`, the page renders live-cache state; when a stream is unhealthy or a site has not yet seeded, the poll keeps the page fresh so a stream failure never blanks it. When live, the poll updates only the `NotReporting` list and leaves the row set to the delta path, so a slow fan-out can never momentarily revert a fresher live delta (R2 N5). (Aggregated live stream **delivered 2026-07-10** — see `docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`.) - **Refresh** — manual refresh button plus the 15s poll timer (mirroring the Health dashboard), now retained as a **fallback + `NotReporting` authority** behind the live cache: when the cache reports `IsLive`, the page renders live-cache state; when a stream is unhealthy, **the aggregator has died (deathwatch resets `IsLive`)**, or a site has not yet seeded, the poll keeps the page fresh so a stream failure never blanks it. When live, the poll updates only the `NotReporting` list and leaves the row set to the delta path, so a slow fan-out can never momentarily revert a fresher live delta (R2 N5). (Aggregated live stream **delivered 2026-07-10** — see `docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`.)
- **Reuse** — the alarm badge/formatter markup is factored out of Debug View into a shared `AlarmStateBadges` component consumed by both Debug View and this page. - **Reuse** — the alarm badge/formatter markup is factored out of Debug View into a shared `AlarmStateBadges` component consumed by both Debug View and this page.
### Parked Message Management (Deployment Role) ### Parked Message Management (Deployment Role)
@@ -60,6 +60,6 @@ public interface ISiteAlarmLiveCache
/// cache is authoritative. /// cache is authoritative.
/// </summary> /// </summary>
/// <param name="siteId">The numeric site id.</param> /// <param name="siteId">The numeric site id.</param>
/// <returns><c>true</c> once the site's aggregator has seeded and published at least once.</returns> /// <returns><c>true</c> while a <b>living</b> aggregator has seeded and published; reverts to false if the aggregator terminates (R2 N6) — a frozen snapshot is never reported as live.</returns>
bool IsLive(int siteId); bool IsLive(int siteId);
} }
@@ -239,6 +239,14 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
TimeSpan.FromSeconds(60))); // stability window — former StabilityWindow static default TimeSpan.FromSeconds(60))); // stability window — former StabilityWindow static default
entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}");
var aggregator = entry.Actor;
// Deathwatch (arch-review R2 N6): if the aggregator terminates for any reason,
// liveness must reset — a dead feed must never keep IsLive == true, or the page
// grafts a frozen snapshot over fresh poll data for up to 15 s.
system.ActorOf(Props.Create(() => new AggregatorTerminationWatcher(
aggregator!, () => OnAggregatorTerminated(entry.SiteId, aggregator!))));
entry.StartRetryTimer?.Dispose(); entry.StartRetryTimer?.Dispose();
entry.StartRetryTimer = null; entry.StartRetryTimer = null;
_logger.LogInformation("Started live alarm aggregator for site {SiteId} ({SiteIdentifier})", _logger.LogInformation("Started live alarm aggregator for site {SiteId} ({SiteIdentifier})",
@@ -421,6 +429,57 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
} }
} }
// ── Deathwatch (R2 N6) ──────────────────────────────────────────────────────
/// <summary>Watches one aggregator and fires a callback exactly once on Terminated.</summary>
private sealed class AggregatorTerminationWatcher : ReceiveActor
{
public AggregatorTerminationWatcher(IActorRef watched, Action onTerminated)
{
Context.Watch(watched);
Receive<Terminated>(_ =>
{
onTerminated();
Context.Stop(Self);
});
}
}
/// <summary>
/// Terminated hook (R2 N6). A DELIBERATE stop (linger fired, entry already removed
/// from _sites) is a no-op — the ReferenceEquals guard also ignores a stale watcher
/// firing after a newer aggregator replaced the dead one. A crash-death with live
/// viewers resets liveness (IsLive → false, snapshot cleared, so the page's poll is
/// authoritative again) and arms the existing bounded start-retry so the feature
/// self-heals, mirroring StartAggregatorAsync's transient-failure path.
/// </summary>
private void OnAggregatorTerminated(int siteId, IActorRef deadActor)
{
lock (_lock)
{
if (!_sites.TryGetValue(siteId, out var entry) || !ReferenceEquals(entry.Actor, deadActor))
return;
entry.Actor = null;
entry.HasPublished = false;
entry.Current = Empty;
if (entry.Subscribers.Count > 0 && !entry.Starting)
{
entry.Starting = true;
entry.StartRetryTimer?.Dispose();
entry.StartRetryTimer = new Timer(
_ => { _ = StartAggregatorAsync(entry); },
state: null,
dueTime: _options.LiveAlarmCacheReconcileInterval,
period: Timeout.InfiniteTimeSpan);
}
}
_logger.LogWarning(
"Live alarm aggregator for site {SiteId} terminated unexpectedly; liveness reset " +
"(page falls back to polling) and a restart is armed while viewers remain", siteId);
}
// ── Nested types ──────────────────────────────────────────────────────────── // ── Nested types ────────────────────────────────────────────────────────────
/// <summary>Per-site bookkeeping. All mutable fields are guarded by the service's <c>_lock</c>.</summary> /// <summary>Per-site bookkeeping. All mutable fields are guarded by the service's <c>_lock</c>.</summary>
@@ -302,4 +302,40 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
AwaitCondition(() => service.IsLive(3), TimeSpan.FromSeconds(5)); AwaitCondition(() => service.IsLive(3), TimeSpan.FromSeconds(5));
Assert.True(factory.GetOrCreateCount >= 1); // a client was created for the single endpoint Assert.True(factory.GetOrCreateCount >= 1); // a client was created for the single endpoint
} }
// ── R2 T10: deathwatch un-sticks IsLive on aggregator death (N6) ──
[Fact]
public async Task Aggregator_Death_Resets_IsLive_And_Clears_The_Snapshot()
{
var service = CreateService(TimeSpan.FromMinutes(5), out _);
using var sub = service.Subscribe(SiteId, () => { });
AwaitAssert(() => Assert.True(service.IsLive(SiteId)), TimeSpan.FromSeconds(10));
// Kill the aggregator out from under the service (crash-death, NOT a linger stop).
var aggregator = await Sys.ActorSelection($"/user/site-alarm-aggregator-{SiteId}-*")
.ResolveOne(TimeSpan.FromSeconds(5));
Sys.Stop(aggregator);
AwaitAssert(() =>
{
Assert.False(service.IsLive(SiteId)); // sticky no more (R2 N6)
Assert.Empty(service.GetCurrentAlarms(SiteId)); // frozen snapshot never grafted
}, TimeSpan.FromSeconds(10));
}
[Fact]
public void Linger_Stop_Does_Not_Resurrect_The_Aggregator()
{
// Last viewer leaves, linger fires, entry removed: the deathwatch on the
// deliberately-stopped actor must be a no-op (entry gone), not a restart.
var service = CreateService(TimeSpan.FromMilliseconds(50), out var factory);
var sub = service.Subscribe(SiteId, () => { });
AwaitAssert(() => Assert.True(service.IsLive(SiteId)), TimeSpan.FromSeconds(10));
sub.Dispose();
AwaitAssert(() => Assert.False(service.IsLive(SiteId)), TimeSpan.FromSeconds(10));
var startsAfterStop = factory.GetOrCreateCount;
Thread.Sleep(300);
Assert.Equal(startsAfterStop, factory.GetOrCreateCount); // no self-heal restart fired
}
} }