fix(adminui): evict unconfigured drivers from status stores; isolate throwing subscribers (R04 L1, L5)

The driver-status and resilience stores had no removal path, so a deleted
driver stayed on /hosts as a frozen row until restart. Both stores gain
RetainOnly, which /hosts calls after each successful config load (never on a
failed one).

The stores and InProcessBroadcaster now raise their events one subscriber at a
time, so a throwing handler cannot starve later subscribers or escape onto the
bridge actor. The alert/script-log bridges publish inside their try, and the
resilience bridge catches a store failure instead of restarting.

Refs #745
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor

@@ -17,6 +17,7 @@
 @using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers
 @inject ActorSystem ActorSystem
 @inject IDriverStatusSnapshotStore DriverStore
+@inject IDriverResilienceStatusStore ResilienceStore
 @inject IDbContextFactory<OtOpcUaConfigDbContext> DbFactory
 @inject Microsoft.Extensions.Logging.ILogger<Hosts> Logger
 @implements IAsyncDisposable
@@ -428,6 +429,16 @@ else
             _instances = await db.DriverInstances.AsNoTracking()
                 .Select(d => new HostsDriverInstanceInfo(d.DriverInstanceId, d.ClusterId, d.Name, d.DriverType, d.Enabled))
                 .ToListAsync(_cts.Token);
+
+            // Evict snapshots of driver instances that are no longer configured (deleted, or re-created
+            // under a new id). The stores have no other removal path, so without this a removed driver
+            // stays on this page as a frozen row until the admin node restarts. Only a SUCCESSFUL load
+            // reaches here; the failure path below blanks _instances and must never sweep the stores.
+            var configured = _instances
+                .Select(i => i.DriverInstanceId)
+                .ToHashSet(StringComparer.OrdinalIgnoreCase);
+            DriverStore.RetainOnly(configured);
+            ResilienceStore.RetainOnly(configured);
         }
         catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException)
         {

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/AlertSignalRBridge.cs

@@ -57,15 +57,16 @@ public sealed class AlertSignalRBridge : ReceiveActor
     private async Task ForwardAsync(AlarmTransitionEvent msg)
     {
         // In-process fan-out first — this is what the Blazor Server Alerts page reads. The hub push
-        // is kept for any out-of-process (e.g. WASM) SignalR client.
-        _broadcaster.Publish(msg);
+        // is kept for any out-of-process (e.g. WASM) SignalR client. Both sit inside the try so a
+        // broadcast failure cannot restart the bridge (and drop its DPS subscription).
         try
         {
+            _broadcaster.Publish(msg);
             await _hub.Clients.All.SendAsync(AlertHub.MethodName, msg);
         }
         catch (Exception ex)
         {
-            _log.Warning(ex, "AlertSignalRBridge: SignalR push failed for {AlarmId}", msg.AlarmId);
+            _log.Warning(ex, "AlertSignalRBridge: forward failed for {AlarmId}", msg.AlarmId);
         }
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/DriverResilienceStatusBridge.cs

@@ -1,5 +1,6 @@
 using Akka.Actor;
 using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.Event;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
 
 namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
@@ -17,6 +18,7 @@ public sealed class DriverResilienceStatusBridge : ReceiveActor
     public const string TopicName = DriverResilienceStatusChanged.TopicName;
 
     private readonly IDriverResilienceStatusStore _store;
+    private readonly ILoggingAdapter _log = Context.GetLogger();
 
     /// <summary>Creates actor props for a <see cref="DriverResilienceStatusBridge"/>.</summary>
     /// <param name="store">Resilience status store updated on every forwarded snapshot.</param>
@@ -29,7 +31,12 @@ public sealed class DriverResilienceStatusBridge : ReceiveActor
     public DriverResilienceStatusBridge(IDriverResilienceStatusStore store)
     {
         _store = store;
-        Receive<DriverResilienceStatusChanged>(msg => _store.Upsert(msg));
+        Receive<DriverResilienceStatusChanged>(msg =>
+        {
+            // A store failure must not restart the bridge (a restart re-subscribes DPS and drops messages).
+            try { _store.Upsert(msg); }
+            catch (Exception ex) { _log.Warning(ex, "DriverResilienceStatusBridge: upsert failed for {0}", msg.DriverInstanceId); }
+        });
         Receive<SubscribeAck>(_ => { /* DPS confirmation */ });
     }
 

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverResilienceStatusStore.cs

@@ -27,6 +27,17 @@ public interface IDriverResilienceStatusStore
     /// <returns>A read-only collection of the last-known resilience snapshot for each pair.</returns>
     IReadOnlyCollection<DriverResilienceStatusChanged> GetAll();
 
+    /// <summary>
+    ///     Evicts every snapshot whose driver instance is not in <paramref name="configuredInstanceIds"/>.
+    ///     The stores have no other removal path, so a deleted (or re-created under a new id) driver would
+    ///     otherwise keep its last snapshot until the process restarts. The <c>/hosts</c> page calls this
+    ///     after each successful config load; a failed load must never call it with an empty set.
+    /// </summary>
+    /// <param name="configuredInstanceIds">The configured driver-instance ids to keep. The set's own
+    ///     comparer decides matching.</param>
+    /// <returns>The number of entries removed.</returns>
+    int RetainOnly(IReadOnlySet<string> configuredInstanceIds);
+
     /// <summary>
     ///     Raised after every <see cref="Upsert"/> with the just-stored snapshot. Lets the Blazor Server
     ///     driver status panel receive live updates by reading this singleton directly. Handlers run on

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs

@@ -25,6 +25,17 @@ public interface IDriverStatusSnapshotStore
     /// <returns>A read-only collection of the last-known health snapshot for each driver instance.</returns>
     IReadOnlyCollection<DriverHealthChanged> GetAll();
 
+    /// <summary>
+    ///     Evicts every snapshot whose driver instance is not in <paramref name="configuredInstanceIds"/>.
+    ///     The stores have no other removal path, so a deleted (or re-created under a new id) driver would
+    ///     otherwise keep its last snapshot until the process restarts. The <c>/hosts</c> page calls this
+    ///     after each successful config load; a failed load must never call it with an empty set.
+    /// </summary>
+    /// <param name="configuredInstanceIds">The configured driver-instance ids to keep. The set's own
+    ///     comparer decides matching.</param>
+    /// <returns>The number of entries removed.</returns>
+    int RetainOnly(IReadOnlySet<string> configuredInstanceIds);
+
     /// <summary>
     ///     Raised after every <see cref="Upsert"/> with the just-stored snapshot. Lets in-process
     ///     consumers (the Blazor Server <c>DriverStatusPanel</c>) receive live updates by reading

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IInProcessBroadcaster.cs

@@ -1,3 +1,5 @@
+using Microsoft.Extensions.Logging;
+
 namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
 
 /// <summary>
@@ -59,6 +61,11 @@ public sealed class InProcessBroadcaster<T> : IInProcessBroadcaster<T>
     // render thread, so access must be serialised.
     private readonly object _connectionLock = new();
     private bool _isConnected;
+    private readonly ILogger<InProcessBroadcaster<T>>? _logger;
+
+    /// <summary>Creates the broadcaster.</summary>
+    /// <param name="logger">Optional logger for a subscriber that throws during a raise.</param>
+    public InProcessBroadcaster(ILogger<InProcessBroadcaster<T>>? logger = null) => _logger = logger;
 
     /// <summary>Raised once per <see cref="Publish"/> with the published item, on the caller's thread.</summary>
     public event Action<T>? Received;
@@ -67,8 +74,10 @@ public sealed class InProcessBroadcaster<T> : IInProcessBroadcaster<T>
     public event Action<bool>? ConnectionStateChanged;
 
     /// <inheritdoc />
-    // Capture-then-invoke (via ?.) so a concurrent unsubscribe can't null the delegate mid-raise.
-    public void Publish(T item) => Received?.Invoke(item);
+    // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise; one subscriber
+    // at a time so a throwing one cannot starve the rest or escape onto the bridge actor.
+    public void Publish(T item) =>
+        SafeMulticast.Invoke(Received, item, ex => _logger?.LogDebug(ex, "In-process broadcast subscriber threw"));
 
     /// <inheritdoc />
     public bool IsConnected
@@ -99,6 +108,7 @@ public sealed class InProcessBroadcaster<T> : IInProcessBroadcaster<T>
             handler = ConnectionStateChanged;
         }
 
-        handler?.Invoke(connected);
+        SafeMulticast.Invoke(handler, connected,
+            ex => _logger?.LogDebug(ex, "In-process connection-state subscriber threw"));
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverResilienceStatusStore.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
 
 namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
@@ -10,6 +11,12 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
 public sealed class InMemoryDriverResilienceStatusStore : IDriverResilienceStatusStore
 {
     private readonly ConcurrentDictionary<(string Instance, string Host), DriverResilienceStatusChanged> _byPair = new();
+    private readonly ILogger<InMemoryDriverResilienceStatusStore>? _logger;
+
+    /// <summary>Creates the store.</summary>
+    /// <param name="logger">Optional logger for a subscriber that throws during a raise.</param>
+    public InMemoryDriverResilienceStatusStore(ILogger<InMemoryDriverResilienceStatusStore>? logger = null)
+        => _logger = logger;
 
     /// <summary>Raised after every <see cref="Upsert"/> with the just-stored snapshot.</summary>
     public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
@@ -18,8 +25,22 @@ public sealed class InMemoryDriverResilienceStatusStore : IDriverResilienceStatu
     public void Upsert(DriverResilienceStatusChanged snapshot)
     {
         _byPair[(snapshot.DriverInstanceId, snapshot.HostName)] = snapshot;
-        // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise.
-        SnapshotChanged?.Invoke(snapshot);
+        // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise; one
+        // subscriber at a time so a throwing one cannot starve the rest (or escape onto the bridge actor).
+        SafeMulticast.Invoke(SnapshotChanged, snapshot,
+            ex => _logger?.LogDebug(ex, "Driver resilience subscriber threw for {DriverInstanceId}", snapshot.DriverInstanceId));
+    }
+
+    /// <inheritdoc />
+    public int RetainOnly(IReadOnlySet<string> configuredInstanceIds)
+    {
+        var removed = 0;
+        foreach (var key in _byPair.Keys)
+        {
+            if (!configuredInstanceIds.Contains(key.Instance) && _byPair.TryRemove(key, out _)) removed++;
+        }
+
+        return removed;
     }
 
     /// <inheritdoc />

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
 
 namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
@@ -10,6 +11,12 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
 public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotStore
 {
     private readonly ConcurrentDictionary<string, DriverHealthChanged> _byInstance = new();
+    private readonly ILogger<InMemoryDriverStatusSnapshotStore>? _logger;
+
+    /// <summary>Creates the store.</summary>
+    /// <param name="logger">Optional logger for a subscriber that throws during a raise.</param>
+    public InMemoryDriverStatusSnapshotStore(ILogger<InMemoryDriverStatusSnapshotStore>? logger = null)
+        => _logger = logger;
 
     /// <summary>Raised whenever a driver status snapshot is stored or updated.</summary>
     public event Action<DriverHealthChanged>? SnapshotChanged;
@@ -18,8 +25,22 @@ public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotSto
     public void Upsert(DriverHealthChanged snapshot)
     {
         _byInstance[snapshot.DriverInstanceId] = snapshot;
-        // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise.
-        SnapshotChanged?.Invoke(snapshot);
+        // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise; one
+        // subscriber at a time so a throwing one cannot starve the rest (or escape onto the bridge actor).
+        SafeMulticast.Invoke(SnapshotChanged, snapshot,
+            ex => _logger?.LogDebug(ex, "Driver status subscriber threw for {DriverInstanceId}", snapshot.DriverInstanceId));
+    }
+
+    /// <inheritdoc />
+    public int RetainOnly(IReadOnlySet<string> configuredInstanceIds)
+    {
+        var removed = 0;
+        foreach (var id in _byInstance.Keys)
+        {
+            if (!configuredInstanceIds.Contains(id) && _byInstance.TryRemove(id, out _)) removed++;
+        }
+
+        return removed;
     }
 
     /// <inheritdoc />

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/SafeMulticast.cs

@@ -0,0 +1,39 @@
+namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
+
+/// <summary>
+///     Raises a multicast delegate one subscriber at a time, so a subscriber that throws cannot stop
+///     delivery to the subscribers after it. The in-process stores and broadcasters raise their events on
+///     the bridge actor's thread; a plain <c>handler?.Invoke(x)</c> stops at the first throwing handler (for
+///     example a Blazor circuit whose renderer is mid-teardown), so the remaining panels miss the update and
+///     the exception escapes into the bridge actor, which restarts it.
+/// </summary>
+internal static class SafeMulticast
+{
+    /// <summary>Invokes every subscriber of <paramref name="handler"/> with <paramref name="arg"/>, isolating failures.</summary>
+    /// <typeparam name="T">The event payload type.</typeparam>
+    /// <param name="handler">The captured multicast delegate; null means no subscribers.</param>
+    /// <param name="arg">The payload passed to each subscriber.</param>
+    /// <param name="onError">Optional sink for a subscriber's exception; it must not throw.</param>
+    /// <returns>The number of subscribers that threw.</returns>
+    public static int Invoke<T>(Action<T>? handler, T arg, Action<Exception>? onError = null)
+    {
+        if (handler is null) return 0;
+
+        var failures = 0;
+        foreach (var subscriber in handler.GetInvocationList())
+        {
+            try
+            {
+                ((Action<T>)subscriber)(arg);
+            }
+            catch (Exception ex)
+            {
+                failures++;
+                try { onError?.Invoke(ex); }
+                catch { /* the error sink must never take delivery down with it */ }
+            }
+        }
+
+        return failures;
+    }
+}

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/ScriptLogSignalRBridge.cs

@@ -52,10 +52,11 @@ public sealed class ScriptLogSignalRBridge : ReceiveActor
     private async Task ForwardAsync(ScriptLogEntry msg)
     {
         // In-process fan-out first — this is what the Blazor Server Script log page reads. The hub
-        // push is kept for any out-of-process (e.g. WASM) SignalR client.
-        _broadcaster.Publish(msg);
+        // push is kept for any out-of-process (e.g. WASM) SignalR client. Both sit inside the try so a
+        // broadcast failure cannot restart the bridge (and drop its DPS subscription).
         try
         {
+            _broadcaster.Publish(msg);
             await _hub.Clients.All.SendAsync(ScriptLogHub.MethodName, msg);
         }
         catch (Exception ex)