OpenTelemetry redundancy metrics + RoleChanged SignalR push. Closes instrumentation + live-push slices of task #198; the exporter wiring (OTLP vs Prometheus package decision) is split to new task #201 because the collector/scrape-endpoint choice is a fleet-ops decision that deserves its own PR rather than hardcoded here. New RedundancyMetrics class (Singleton-registered in DI) owning a System.Diagnostics.Metrics.Meter("ZB.MOM.WW.OtOpcUa.Redundancy", "1.0.0"). Three ObservableGauge instruments — otopcua.redundancy.primary_count / secondary_count / stale_count — all tagged by cluster.id, populated by SetClusterCounts(clusterId, primary, secondary, stale) which the poller calls at the tail of every tick; ObservableGauge callbacks snapshot the last value set under a lock so the reader (OTel collector, dotnet-counters) sees consistent tuples. One Counter — otopcua.redundancy.role_transition — tagged cluster.id, node.id, from_role, to_role; ideal for tracking "how often does Cluster-X failover" + "which node transitions most" aggregate queries. In-box Metrics API means zero NuGet dep here — the exporter PR adds OpenTelemetry.Extensions.Hosting + OpenTelemetry.Exporter.OpenTelemetryProtocol or OpenTelemetry.Exporter.Prometheus.AspNetCore to actually ship the data somewhere. FleetStatusPoller extended with role-change detection. Its PollOnceAsync now pulls ClusterNode rows alongside the existing ClusterNodeGenerationState scan, and a new PollRolesAsync walks every node comparing RedundancyRole to the _lastRole cache. On change: records the transition to RedundancyMetrics + emits a RoleChanged SignalR message to both FleetStatusHub.GroupName(cluster) + FleetStatusHub.FleetGroup so cluster-scoped + fleet-wide subscribers both see it. First observation per node is a bootstrap (cache fill) + NOT a transition — avoids spurious churn on service startup or pod restart. UpdateClusterGauges groups nodes by cluster + sets the three gauge values, using ClusterNodeService.StaleThreshold (shared 30s convention) for staleness so the /hosts page + the gauge agree. RoleChangedMessage record lives alongside NodeStateChangedMessage in FleetStatusPoller.cs. RedundancyTab.razor subscribes to the fleet-status hub on first parameters-set, filters RoleChanged events to the current cluster, reloads the node list + paints a blue info banner ("Role changed on node-a: Primary → Secondary at HH:mm:ss UTC") so operators see the transition without needing to poll-refresh the page. IAsyncDisposable closes the connection on tab swap-away. Two new RedundancyMetricsTests covering RecordRoleTransition tag emission (cluster.id + node.id + from_role + to_role all flow through the MeterListener callback) + ObservableGauge snapshot for two clusters (assert primary_count=1 for c1, stale_count=1 for c2). Existing FleetStatusPollerTests ctor-line updated to pass a RedundancyMetrics instance; all tests still pass. Full Admin.Tests suite 87/87 passing (was 85, +2). Admin project builds 0 errors. Task #201 captures the exporter-wiring follow-up — OpenTelemetry.Extensions.Hosting + OTLP vs Prometheus + /metrics endpoint decision, driven by fleet-ops infra direction.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/ZB.MOM.WW.OtOpcUa.Admin/Components/Pages/Clusters/RedundancyTab.razor

@@ -1,9 +1,17 @@
+@using Microsoft.AspNetCore.SignalR.Client
+@using ZB.MOM.WW.OtOpcUa.Admin.Hubs
 @using ZB.MOM.WW.OtOpcUa.Admin.Services
 @using ZB.MOM.WW.OtOpcUa.Configuration.Entities
 @using ZB.MOM.WW.OtOpcUa.Configuration.Enums
 @inject ClusterNodeService NodeSvc
+@inject NavigationManager Nav
+@implements IAsyncDisposable
 
 <h4>Redundancy topology</h4>
+@if (_roleChangedBanner is not null)
+{
+    <div class="alert alert-info small mb-2">@_roleChangedBanner</div>
+}
 <p class="text-muted small">
     One row per <code>ClusterNode</code> in this cluster. Role, <code>ApplicationUri</code>,
     and <code>ServiceLevelBase</code> are authored separately; the Admin UI shows them read-only
@@ -107,10 +115,41 @@ else
     [Parameter] public string ClusterId { get; set; } = string.Empty;
 
     private List<ClusterNode>? _nodes;
+    private HubConnection? _hub;
+    private string? _roleChangedBanner;
 
     protected override async Task OnParametersSetAsync()
     {
         _nodes = await NodeSvc.ListByClusterAsync(ClusterId, CancellationToken.None);
+        if (_hub is null) await ConnectHubAsync();
+    }
+
+    private async Task ConnectHubAsync()
+    {
+        _hub = new HubConnectionBuilder()
+            .WithUrl(Nav.ToAbsoluteUri("/hubs/fleet-status"))
+            .WithAutomaticReconnect()
+            .Build();
+
+        _hub.On<RoleChangedMessage>("RoleChanged", async msg =>
+        {
+            if (msg.ClusterId != ClusterId) return;
+            _roleChangedBanner = $"Role changed on {msg.NodeId}: {msg.FromRole} → {msg.ToRole} at {msg.ObservedAtUtc:HH:mm:ss 'UTC'}";
+            _nodes = await NodeSvc.ListByClusterAsync(ClusterId, CancellationToken.None);
+            await InvokeAsync(StateHasChanged);
+        });
+
+        await _hub.StartAsync();
+        await _hub.SendAsync("SubscribeCluster", ClusterId);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        if (_hub is not null)
+        {
+            await _hub.DisposeAsync();
+            _hub = null;
+        }
     }
 
     private static string RowClass(ClusterNode n) =>

src/ZB.MOM.WW.OtOpcUa.Admin/Hubs/FleetStatusPoller.cs

@@ -1,7 +1,9 @@
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.OtOpcUa.Admin.Services;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
 
 namespace ZB.MOM.WW.OtOpcUa.Admin.Hubs;
 
@@ -14,11 +16,13 @@ public sealed class FleetStatusPoller(
     IServiceScopeFactory scopeFactory,
     IHubContext<FleetStatusHub> fleetHub,
     IHubContext<AlertHub> alertHub,
-    ILogger<FleetStatusPoller> logger) : BackgroundService
+    ILogger<FleetStatusPoller> logger,
+    RedundancyMetrics redundancyMetrics) : BackgroundService
 {
     public TimeSpan PollInterval { get; init; } = TimeSpan.FromSeconds(5);
 
     private readonly Dictionary<string, NodeStateSnapshot> _last = new();
+    private readonly Dictionary<string, RedundancyRole> _lastRole = new(StringComparer.Ordinal);
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
@@ -42,6 +46,10 @@ public sealed class FleetStatusPoller(
         using var scope = scopeFactory.CreateScope();
         var db = scope.ServiceProvider.GetRequiredService<OtOpcUaConfigDbContext>();
 
+        var nodes = await db.ClusterNodes.AsNoTracking().ToListAsync(ct);
+        await PollRolesAsync(nodes, ct);
+        UpdateClusterGauges(nodes);
+
         var rows = await db.ClusterNodeGenerationStates.AsNoTracking()
             .Join(db.ClusterNodes.AsNoTracking(), s => s.NodeId, n => n.NodeId, (s, n) => new { s, n.ClusterId })
             .ToListAsync(ct);
@@ -85,9 +93,63 @@ public sealed class FleetStatusPoller(
     }
 
     /// <summary>Exposed for tests — forces a snapshot reset so stub data re-seeds.</summary>
-    internal void ResetCache() => _last.Clear();
+    internal void ResetCache()
+    {
+        _last.Clear();
+        _lastRole.Clear();
+    }
+
+    private async Task PollRolesAsync(IReadOnlyList<ClusterNode> nodes, CancellationToken ct)
+    {
+        foreach (var n in nodes)
+        {
+            var hadPrior = _lastRole.TryGetValue(n.NodeId, out var priorRole);
+            if (hadPrior && priorRole == n.RedundancyRole) continue;
+
+            _lastRole[n.NodeId] = n.RedundancyRole;
+            if (!hadPrior) continue; // first-observation bootstrap — not a transition
+
+            redundancyMetrics.RecordRoleTransition(
+                clusterId: n.ClusterId, nodeId: n.NodeId,
+                fromRole: priorRole.ToString(), toRole: n.RedundancyRole.ToString());
+
+            var msg = new RoleChangedMessage(
+                ClusterId: n.ClusterId, NodeId: n.NodeId,
+                FromRole: priorRole.ToString(), ToRole: n.RedundancyRole.ToString(),
+                ObservedAtUtc: DateTime.UtcNow);
+
+            await fleetHub.Clients.Group(FleetStatusHub.GroupName(n.ClusterId))
+                .SendAsync("RoleChanged", msg, ct);
+            await fleetHub.Clients.Group(FleetStatusHub.FleetGroup)
+                .SendAsync("RoleChanged", msg, ct);
+        }
+    }
+
+    private void UpdateClusterGauges(IReadOnlyList<ClusterNode> nodes)
+    {
+        var staleCutoff = DateTime.UtcNow - Services.ClusterNodeService.StaleThreshold;
+        foreach (var group in nodes.GroupBy(n => n.ClusterId))
+        {
+            var primary = group.Count(n => n.RedundancyRole == RedundancyRole.Primary);
+            var secondary = group.Count(n => n.RedundancyRole == RedundancyRole.Secondary);
+            var stale = group.Count(n => n.LastSeenAt is null || n.LastSeenAt.Value < staleCutoff);
+            redundancyMetrics.SetClusterCounts(group.Key, primary, secondary, stale);
+        }
+    }
 
     private readonly record struct NodeStateSnapshot(
         string NodeId, string ClusterId, long? GenerationId,
         string? Status, string? Error, DateTime? AppliedAt, DateTime? SeenAt);
 }
+
+/// <summary>
+///     Pushed by <see cref="FleetStatusPoller"/> when it observes a change in
+///     <see cref="ClusterNode.RedundancyRole"/>. Consumed by the Admin RedundancyTab to trigger
+///     an instant reload instead of waiting for the next on-parameter-set poll.
+/// </summary>
+public sealed record RoleChangedMessage(
+    string ClusterId,
+    string NodeId,
+    string FromRole,
+    string ToRole,
+    DateTime ObservedAtUtc);

src/ZB.MOM.WW.OtOpcUa.Admin/Program.cs

@@ -49,6 +49,7 @@ builder.Services.AddScoped<DraftValidationService>();
 builder.Services.AddScoped<AuditLogService>();
 builder.Services.AddScoped<HostStatusService>();
 builder.Services.AddScoped<ClusterNodeService>();
+builder.Services.AddSingleton<RedundancyMetrics>();
 builder.Services.AddScoped<EquipmentImportBatchService>();
 builder.Services.AddScoped<ZB.MOM.WW.OtOpcUa.Configuration.Services.ILdapGroupRoleMappingService,
     ZB.MOM.WW.OtOpcUa.Configuration.Services.LdapGroupRoleMappingService>();

src/ZB.MOM.WW.OtOpcUa.Admin/Services/EquipmentImportBatchService.cs

@@ -2,6 +2,7 @@ using Microsoft.EntityFrameworkCore;
 using ZB.MOM.WW.OtOpcUa.Admin.Services;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
 
 namespace ZB.MOM.WW.OtOpcUa.Admin.Services;
 
@@ -152,14 +153,37 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
 
         try
         {
-            foreach (var row in batch.Rows.Where(r => r.IsAccepted))
+            // Snapshot active reservations that overlap this batch's ZTag + SAPID set — one
+            // round-trip instead of N. Released rows (ReleasedAt IS NOT NULL) are ignored so
+            // an explicitly-released value can be reused.
+            var accepted = batch.Rows.Where(r => r.IsAccepted).ToList();
+            var zTags = accepted.Where(r => !string.IsNullOrWhiteSpace(r.ZTag))
+                .Select(r => r.ZTag).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
+            var sapIds = accepted.Where(r => !string.IsNullOrWhiteSpace(r.SAPID))
+                .Select(r => r.SAPID).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
+
+            var existingReservations = await db.ExternalIdReservations
+                .Where(r => r.ReleasedAt == null &&
+                    ((r.Kind == ReservationKind.ZTag && zTags.Contains(r.Value)) ||
+                     (r.Kind == ReservationKind.SAPID && sapIds.Contains(r.Value))))
+                .ToListAsync(ct).ConfigureAwait(false);
+            var resByKey = existingReservations.ToDictionary(
+                r => (r.Kind, r.Value.ToLowerInvariant()),
+                r => r);
+
+            var nowUtc = DateTime.UtcNow;
+            var firstPublishedBy = batch.CreatedBy;
+
+            foreach (var row in accepted)
             {
+                var equipmentUuid = Guid.TryParse(row.EquipmentUuid, out var u) ? u : Guid.NewGuid();
+
                 db.Equipment.Add(new Equipment
                 {
                     EquipmentRowId = Guid.NewGuid(),
                     GenerationId = generationId,
                     EquipmentId = row.EquipmentId,
-                    EquipmentUuid = Guid.TryParse(row.EquipmentUuid, out var u) ? u : Guid.NewGuid(),
+                    EquipmentUuid = equipmentUuid,
                     DriverInstanceId = driverInstanceIdForRows,
                     UnsLineId = unsLineIdForRows,
                     Name = row.Name,
@@ -176,10 +200,25 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
                     ManufacturerUri = row.ManufacturerUri,
                     DeviceManualUri = row.DeviceManualUri,
                 });
+
+                MergeReservation(row.ZTag, ReservationKind.ZTag, equipmentUuid, batch.ClusterId,
+                    firstPublishedBy, nowUtc, resByKey);
+                MergeReservation(row.SAPID, ReservationKind.SAPID, equipmentUuid, batch.ClusterId,
+                    firstPublishedBy, nowUtc, resByKey);
             }
 
-            batch.FinalisedAtUtc = DateTime.UtcNow;
-            await db.SaveChangesAsync(ct).ConfigureAwait(false);
+            batch.FinalisedAtUtc = nowUtc;
+            try
+            {
+                await db.SaveChangesAsync(ct).ConfigureAwait(false);
+            }
+            catch (DbUpdateException ex) when (IsReservationUniquenessViolation(ex))
+            {
+                throw new ExternalIdReservationConflictException(
+                    "Finalise rejected: one or more ZTag/SAPID values were reserved by another operator " +
+                    "between batch preview and commit. Inspect active reservations + retry after resolving the conflict.",
+                    ex);
+            }
             if (tx is not null) await tx.CommitAsync(ct).ConfigureAwait(false);
         }
         catch
@@ -193,6 +232,71 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
         }
     }
 
+    /// <summary>
+    ///     Merge one external-ID reservation for an equipment row. Three outcomes:
+    ///     (1) value is empty → skip; (2) reservation exists for same <paramref name="equipmentUuid"/>
+    ///     → bump <c>LastPublishedAt</c>; (3) reservation exists for a different EquipmentUuid
+    ///     → throw <see cref="ExternalIdReservationConflictException"/> with the conflicting UUID
+    ///     so the caller sees which equipment already owns the value; (4) no reservation → create new.
+    /// </summary>
+    private void MergeReservation(
+        string? value,
+        ReservationKind kind,
+        Guid equipmentUuid,
+        string clusterId,
+        string firstPublishedBy,
+        DateTime nowUtc,
+        Dictionary<(ReservationKind, string), ExternalIdReservation> cache)
+    {
+        if (string.IsNullOrWhiteSpace(value)) return;
+
+        var key = (kind, value.ToLowerInvariant());
+        if (cache.TryGetValue(key, out var existing))
+        {
+            if (existing.EquipmentUuid != equipmentUuid)
+                throw new ExternalIdReservationConflictException(
+                    $"{kind} '{value}' is already reserved by EquipmentUuid {existing.EquipmentUuid} " +
+                    $"(first published {existing.FirstPublishedAt:u} on cluster '{existing.ClusterId}'). " +
+                    $"Refusing to re-assign to {equipmentUuid}.");
+
+            existing.LastPublishedAt = nowUtc;
+            return;
+        }
+
+        var fresh = new ExternalIdReservation
+        {
+            ReservationId = Guid.NewGuid(),
+            Kind = kind,
+            Value = value,
+            EquipmentUuid = equipmentUuid,
+            ClusterId = clusterId,
+            FirstPublishedAt = nowUtc,
+            FirstPublishedBy = firstPublishedBy,
+            LastPublishedAt = nowUtc,
+        };
+        db.ExternalIdReservations.Add(fresh);
+        cache[key] = fresh;
+    }
+
+    /// <summary>
+    ///     True when the <see cref="DbUpdateException"/> root-cause was the filtered-unique
+    ///     index <c>UX_ExternalIdReservation_KindValue_Active</c> — i.e. another transaction
+    ///     won the race between our cache-load + commit. SQL Server surfaces this as 2601 / 2627.
+    /// </summary>
+    private static bool IsReservationUniquenessViolation(DbUpdateException ex)
+    {
+        for (Exception? inner = ex; inner is not null; inner = inner.InnerException)
+        {
+            if (inner is Microsoft.Data.SqlClient.SqlException sql &&
+                (sql.Number == 2601 || sql.Number == 2627) &&
+                sql.Message.Contains("UX_ExternalIdReservation_KindValue_Active", StringComparison.OrdinalIgnoreCase))
+            {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /// <summary>List batches created by the given user. Finalised batches are archived; include them on demand.</summary>
     public async Task<IReadOnlyList<EquipmentImportBatch>> ListByUserAsync(string createdBy, bool includeFinalised, CancellationToken ct)
     {
@@ -205,3 +309,16 @@ public sealed class EquipmentImportBatchService(OtOpcUaConfigDbContext db)
 
 public sealed class ImportBatchNotFoundException(string message) : Exception(message);
 public sealed class ImportBatchAlreadyFinalisedException(string message) : Exception(message);
+
+/// <summary>
+///     Thrown when a <c>FinaliseBatchAsync</c> call detects that one of its ZTag/SAPID values is
+///     already reserved by a different EquipmentUuid — either from a prior published generation
+///     or a concurrent finalise that won the race. The operator sees the message + the conflicting
+///     equipment ownership so they can resolve the conflict (pick a new ZTag, release the existing
+///     reservation via <c>sp_ReleaseExternalIdReservation</c>, etc.) and retry the finalise.
+/// </summary>
+public sealed class ExternalIdReservationConflictException : Exception
+{
+    public ExternalIdReservationConflictException(string message) : base(message) { }
+    public ExternalIdReservationConflictException(string message, Exception inner) : base(message, inner) { }
+}

src/ZB.MOM.WW.OtOpcUa.Admin/Services/RedundancyMetrics.cs

@@ -0,0 +1,102 @@
+using System.Diagnostics.Metrics;
+
+namespace ZB.MOM.WW.OtOpcUa.Admin.Services;
+
+/// <summary>
+///     OpenTelemetry-compatible instrumentation for the redundancy surface. Uses in-box
+///     <see cref="System.Diagnostics.Metrics"/> so no NuGet dependency is required to emit —
+///     any MeterListener (dotnet-counters, OpenTelemetry.Extensions.Hosting OTLP exporter,
+///     Prometheus exporter, etc.) picks up the instruments by the <see cref="MeterName"/>.
+/// </summary>
+/// <remarks>
+///     Exporter configuration (OTLP, Prometheus, etc.) is intentionally NOT wired here —
+///     that's a deployment-ops decision that belongs in <c>Program.cs</c> behind an
+///     <c>appsettings</c> toggle. This class owns only the Meter + instruments so the
+///     production data stream exists regardless of exporter availability.
+///
+///     Counter + gauge names follow the otel-semantic-conventions pattern:
+///     <c>otopcua.redundancy.*</c> with tags for ClusterId + (for transitions) FromRole/ToRole/NodeId.
+/// </remarks>
+public sealed class RedundancyMetrics : IDisposable
+{
+    public const string MeterName = "ZB.MOM.WW.OtOpcUa.Redundancy";
+
+    private readonly Meter _meter;
+    private readonly Counter<long> _roleTransitions;
+    private readonly object _gaugeLock = new();
+    private readonly Dictionary<string, ClusterGaugeState> _gaugeState = new();
+
+    public RedundancyMetrics()
+    {
+        _meter = new Meter(MeterName, version: "1.0.0");
+        _roleTransitions = _meter.CreateCounter<long>(
+            "otopcua.redundancy.role_transition",
+            unit: "{transition}",
+            description: "Observed RedundancyRole changes per node — tagged FromRole, ToRole, NodeId, ClusterId.");
+
+        // Observable gauges — the callback reports whatever the last Observe*Count call stashed.
+        _meter.CreateObservableGauge(
+            "otopcua.redundancy.primary_count",
+            ObservePrimaryCounts,
+            unit: "{node}",
+            description: "Count of Primary-role nodes per cluster (should be 1 for N+1 redundant clusters, 0 during failover).");
+        _meter.CreateObservableGauge(
+            "otopcua.redundancy.secondary_count",
+            ObserveSecondaryCounts,
+            unit: "{node}",
+            description: "Count of Secondary-role nodes per cluster.");
+        _meter.CreateObservableGauge(
+            "otopcua.redundancy.stale_count",
+            ObserveStaleCounts,
+            unit: "{node}",
+            description: "Count of cluster nodes whose LastSeenAt is older than StaleThreshold.");
+    }
+
+    /// <summary>
+    ///     Update the per-cluster snapshot consumed by the ObservableGauges. Poller calls this
+    ///     at the end of every tick so the collectors see fresh numbers on the next observation
+    ///     window (by default 1s for dotnet-counters, configurable per exporter).
+    /// </summary>
+    public void SetClusterCounts(string clusterId, int primary, int secondary, int stale)
+    {
+        lock (_gaugeLock)
+        {
+            _gaugeState[clusterId] = new ClusterGaugeState(primary, secondary, stale);
+        }
+    }
+
+    /// <summary>
+    ///     Increment the role_transition counter when a node's RedundancyRole changes. Tags
+    ///     allow breakdowns by from/to roles (e.g. Primary → Secondary for planned failover vs
+    ///     Primary → Standalone for emergency recovery) + by cluster for multi-site fleets.
+    /// </summary>
+    public void RecordRoleTransition(string clusterId, string nodeId, string fromRole, string toRole)
+    {
+        _roleTransitions.Add(1,
+            new KeyValuePair<string, object?>("cluster.id", clusterId),
+            new KeyValuePair<string, object?>("node.id", nodeId),
+            new KeyValuePair<string, object?>("from_role", fromRole),
+            new KeyValuePair<string, object?>("to_role", toRole));
+    }
+
+    public void Dispose() => _meter.Dispose();
+
+    private IEnumerable<Measurement<long>> ObservePrimaryCounts() => SnapshotGauge(s => s.Primary);
+    private IEnumerable<Measurement<long>> ObserveSecondaryCounts() => SnapshotGauge(s => s.Secondary);
+    private IEnumerable<Measurement<long>> ObserveStaleCounts() => SnapshotGauge(s => s.Stale);
+
+    private IEnumerable<Measurement<long>> SnapshotGauge(Func<ClusterGaugeState, int> selector)
+    {
+        List<Measurement<long>> results;
+        lock (_gaugeLock)
+        {
+            results = new List<Measurement<long>>(_gaugeState.Count);
+            foreach (var (cluster, state) in _gaugeState)
+                results.Add(new Measurement<long>(selector(state),
+                    new KeyValuePair<string, object?>("cluster.id", cluster)));
+        }
+        return results;
+    }
+
+    private readonly record struct ClusterGaugeState(int Primary, int Secondary, int Stale);
+}

src/ZB.MOM.WW.OtOpcUa.Core/Resilience/AlarmSurfaceInvoker.cs

@@ -0,0 +1,129 @@
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
+
+/// <summary>
+///     Wraps the three mutating surfaces of <see cref="IAlarmSource"/>
+///     (<see cref="IAlarmSource.SubscribeAlarmsAsync"/>, <see cref="IAlarmSource.UnsubscribeAlarmsAsync"/>,
+///     <see cref="IAlarmSource.AcknowledgeAsync"/>) through <see cref="CapabilityInvoker"/> so the
+///     Phase 6.1 resilience pipeline runs — retry semantics match
+///     <see cref="DriverCapability.AlarmSubscribe"/> (retries by default) and
+///     <see cref="DriverCapability.AlarmAcknowledge"/> (does NOT retry per decision #143).
+/// </summary>
+/// <remarks>
+///     <para>Multi-host dispatch: when the driver implements <see cref="IPerCallHostResolver"/>,
+///     each source-node-id is resolved individually + grouped by host so a dead PLC inside a
+///     multi-device driver doesn't poison the sibling hosts' breakers. Drivers with a single
+///     host fall back to <see cref="IDriver.DriverInstanceId"/> as the single-host key.</para>
+///
+///     <para>Why this lives here + not on <see cref="CapabilityInvoker"/>: alarm surfaces have a
+///     handle-returning shape (SubscribeAlarmsAsync returns <see cref="IAlarmSubscriptionHandle"/>)
+///     + a per-call fan-out (AcknowledgeAsync gets a batch of
+///     <see cref="AlarmAcknowledgeRequest"/>s that may span multiple hosts). Keeping the fan-out
+///     logic here keeps the invoker's execute-overloads narrow.</para>
+/// </remarks>
+public sealed class AlarmSurfaceInvoker
+{
+    private readonly CapabilityInvoker _invoker;
+    private readonly IAlarmSource _alarmSource;
+    private readonly IPerCallHostResolver? _hostResolver;
+    private readonly string _defaultHost;
+
+    public AlarmSurfaceInvoker(
+        CapabilityInvoker invoker,
+        IAlarmSource alarmSource,
+        string defaultHost,
+        IPerCallHostResolver? hostResolver = null)
+    {
+        ArgumentNullException.ThrowIfNull(invoker);
+        ArgumentNullException.ThrowIfNull(alarmSource);
+        ArgumentException.ThrowIfNullOrWhiteSpace(defaultHost);
+
+        _invoker = invoker;
+        _alarmSource = alarmSource;
+        _defaultHost = defaultHost;
+        _hostResolver = hostResolver;
+    }
+
+    /// <summary>
+    ///     Subscribe to alarm events for a set of source node ids, fanning out by resolved host
+    ///     so per-host breakers / bulkheads apply. Returns one handle per host — callers that
+    ///     don't care about per-host separation may concatenate them.
+    /// </summary>
+    public async Task<IReadOnlyList<IAlarmSubscriptionHandle>> SubscribeAsync(
+        IReadOnlyList<string> sourceNodeIds,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(sourceNodeIds);
+        if (sourceNodeIds.Count == 0) return [];
+
+        var byHost = GroupByHost(sourceNodeIds);
+        var handles = new List<IAlarmSubscriptionHandle>(byHost.Count);
+        foreach (var (host, ids) in byHost)
+        {
+            var handle = await _invoker.ExecuteAsync(
+                DriverCapability.AlarmSubscribe,
+                host,
+                async ct => await _alarmSource.SubscribeAlarmsAsync(ids, ct).ConfigureAwait(false),
+                cancellationToken).ConfigureAwait(false);
+            handles.Add(handle);
+        }
+        return handles;
+    }
+
+    /// <summary>Cancel an alarm subscription. Routes through the AlarmSubscribe pipeline for parity.</summary>
+    public ValueTask UnsubscribeAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(handle);
+        return _invoker.ExecuteAsync(
+            DriverCapability.AlarmSubscribe,
+            _defaultHost,
+            async ct => await _alarmSource.UnsubscribeAlarmsAsync(handle, ct).ConfigureAwait(false),
+            cancellationToken);
+    }
+
+    /// <summary>
+    ///     Acknowledge alarms. Fans out by resolved host; each host's batch runs through the
+    ///     AlarmAcknowledge pipeline (no-retry per decision #143 — an alarm-ack is not idempotent
+    ///     at the plant-floor acknowledgement level even if the OPC UA spec permits re-issue).
+    /// </summary>
+    public async Task AcknowledgeAsync(
+        IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(acknowledgements);
+        if (acknowledgements.Count == 0) return;
+
+        var byHost = _hostResolver is null
+            ? new Dictionary<string, List<AlarmAcknowledgeRequest>> { [_defaultHost] = acknowledgements.ToList() }
+            : acknowledgements
+                .GroupBy(a => _hostResolver.ResolveHost(a.SourceNodeId))
+                .ToDictionary(g => g.Key, g => g.ToList());
+
+        foreach (var (host, batch) in byHost)
+        {
+            var batchSnapshot = batch; // capture for the lambda
+            await _invoker.ExecuteAsync(
+                DriverCapability.AlarmAcknowledge,
+                host,
+                async ct => await _alarmSource.AcknowledgeAsync(batchSnapshot, ct).ConfigureAwait(false),
+                cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    private Dictionary<string, List<string>> GroupByHost(IReadOnlyList<string> sourceNodeIds)
+    {
+        if (_hostResolver is null)
+            return new Dictionary<string, List<string>> { [_defaultHost] = sourceNodeIds.ToList() };
+
+        var result = new Dictionary<string, List<string>>(StringComparer.Ordinal);
+        foreach (var id in sourceNodeIds)
+        {
+            var host = _hostResolver.ResolveHost(id);
+            if (!result.TryGetValue(host, out var list))
+                result[host] = list = new List<string>();
+            list.Add(id);
+        }
+        return result;
+    }
+}

tests/ZB.MOM.WW.OtOpcUa.Admin.Tests/EquipmentImportBatchServiceTests.cs

@@ -23,11 +23,13 @@ public sealed class EquipmentImportBatchServiceTests : IDisposable
 
     public void Dispose() => _db.Dispose();
 
+    // Unique SAPID per row — FinaliseBatch reserves ZTag + SAPID via filtered-unique index, so
+    // two rows sharing a SAPID under different EquipmentUuids collide as intended.
     private static EquipmentCsvRow Row(string zTag, string name = "eq-1") => new()
     {
         ZTag = zTag,
         MachineCode = "mc",
-        SAPID = "sap",
+        SAPID = $"sap-{zTag}",
         EquipmentId = "eq-id",
         EquipmentUuid = Guid.NewGuid().ToString(),
         Name = name,
@@ -162,4 +164,93 @@ public sealed class EquipmentImportBatchServiceTests : IDisposable
         await _svc.DropBatchAsync(Guid.NewGuid(), CancellationToken.None);
         // no throw
     }
+
+    [Fact]
+    public async Task FinaliseBatch_Creates_ExternalIdReservations_ForZTagAndSAPID()
+    {
+        var batch = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
+        await _svc.StageRowsAsync(batch.Id, [Row("z-new-1")], [], CancellationToken.None);
+
+        await _svc.FinaliseBatchAsync(batch.Id, 1, "drv", "line", CancellationToken.None);
+
+        var active = await _db.ExternalIdReservations.AsNoTracking()
+            .Where(r => r.ReleasedAt == null)
+            .ToListAsync();
+        active.Count.ShouldBe(2);
+        active.ShouldContain(r => r.Kind == ZB.MOM.WW.OtOpcUa.Configuration.Enums.ReservationKind.ZTag && r.Value == "z-new-1");
+        active.ShouldContain(r => r.Kind == ZB.MOM.WW.OtOpcUa.Configuration.Enums.ReservationKind.SAPID && r.Value == "sap-z-new-1");
+    }
+
+    [Fact]
+    public async Task FinaliseBatch_SameEquipmentUuid_ReusesExistingReservation()
+    {
+        var batch1 = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
+        var sharedUuid = Guid.NewGuid();
+        var row = new EquipmentCsvRow
+        {
+            ZTag = "z-shared", MachineCode = "mc", SAPID = "sap-shared",
+            EquipmentId = "eq-1", EquipmentUuid = sharedUuid.ToString(),
+            Name = "eq-1", UnsAreaName = "a", UnsLineName = "l",
+        };
+        await _svc.StageRowsAsync(batch1.Id, [row], [], CancellationToken.None);
+        await _svc.FinaliseBatchAsync(batch1.Id, 1, "drv", "line", CancellationToken.None);
+
+        var countAfterFirst = _db.ExternalIdReservations.Count(r => r.ReleasedAt == null);
+
+        // Second finalise with same EquipmentUuid + same ZTag — should NOT create a duplicate.
+        var batch2 = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
+        await _svc.StageRowsAsync(batch2.Id, [row], [], CancellationToken.None);
+        await _svc.FinaliseBatchAsync(batch2.Id, 2, "drv", "line", CancellationToken.None);
+
+        _db.ExternalIdReservations.Count(r => r.ReleasedAt == null).ShouldBe(countAfterFirst);
+    }
+
+    [Fact]
+    public async Task FinaliseBatch_DifferentEquipmentUuid_SameZTag_Throws_Conflict()
+    {
+        var batchA = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
+        var rowA = new EquipmentCsvRow
+        {
+            ZTag = "z-collide", MachineCode = "mc-a", SAPID = "sap-a",
+            EquipmentId = "eq-a", EquipmentUuid = Guid.NewGuid().ToString(),
+            Name = "a", UnsAreaName = "ar", UnsLineName = "ln",
+        };
+        await _svc.StageRowsAsync(batchA.Id, [rowA], [], CancellationToken.None);
+        await _svc.FinaliseBatchAsync(batchA.Id, 1, "drv", "line", CancellationToken.None);
+
+        var batchB = await _svc.CreateBatchAsync("c1", "bob", CancellationToken.None);
+        var rowB = new EquipmentCsvRow
+        {
+            ZTag = "z-collide", MachineCode = "mc-b", SAPID = "sap-b", // same ZTag, different EquipmentUuid
+            EquipmentId = "eq-b", EquipmentUuid = Guid.NewGuid().ToString(),
+            Name = "b", UnsAreaName = "ar", UnsLineName = "ln",
+        };
+        await _svc.StageRowsAsync(batchB.Id, [rowB], [], CancellationToken.None);
+
+        var ex = await Should.ThrowAsync<ExternalIdReservationConflictException>(() =>
+            _svc.FinaliseBatchAsync(batchB.Id, 2, "drv", "line", CancellationToken.None));
+        ex.Message.ShouldContain("z-collide");
+
+        // Second finalise must have rolled back — no partial Equipment row for batch B.
+        var equipmentB = await _db.Equipment.AsNoTracking()
+            .Where(e => e.EquipmentId == "eq-b")
+            .ToListAsync();
+        equipmentB.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public async Task FinaliseBatch_EmptyZTagAndSAPID_SkipsReservation()
+    {
+        var batch = await _svc.CreateBatchAsync("c1", "alice", CancellationToken.None);
+        var row = new EquipmentCsvRow
+        {
+            ZTag = "", MachineCode = "mc", SAPID = "",
+            EquipmentId = "eq-nil", EquipmentUuid = Guid.NewGuid().ToString(),
+            Name = "nil", UnsAreaName = "ar", UnsLineName = "ln",
+        };
+        await _svc.StageRowsAsync(batch.Id, [row], [], CancellationToken.None);
+        await _svc.FinaliseBatchAsync(batch.Id, 1, "drv", "line", CancellationToken.None);
+
+        _db.ExternalIdReservations.Count().ShouldBe(0);
+    }
 }

tests/ZB.MOM.WW.OtOpcUa.Admin.Tests/FleetStatusPollerTests.cs

@@ -5,6 +5,7 @@ using Microsoft.Extensions.Logging.Abstractions;
 using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Admin.Hubs;
+using ZB.MOM.WW.OtOpcUa.Admin.Services;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
 using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
@@ -97,7 +98,7 @@ END";
 
         var poller = new FleetStatusPoller(
             _sp.GetRequiredService<IServiceScopeFactory>(),
-            fleetHub, alertHub, NullLogger<FleetStatusPoller>.Instance);
+            fleetHub, alertHub, NullLogger<FleetStatusPoller>.Instance, new RedundancyMetrics());
 
         await poller.PollOnceAsync(CancellationToken.None);
 
@@ -142,7 +143,7 @@ END";
 
         var poller = new FleetStatusPoller(
             _sp.GetRequiredService<IServiceScopeFactory>(),
-            fleetHub, alertHub, NullLogger<FleetStatusPoller>.Instance);
+            fleetHub, alertHub, NullLogger<FleetStatusPoller>.Instance, new RedundancyMetrics());
 
         await poller.PollOnceAsync(CancellationToken.None);
 

tests/ZB.MOM.WW.OtOpcUa.Admin.Tests/RedundancyMetricsTests.cs

@@ -0,0 +1,70 @@
+using System.Diagnostics.Metrics;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Admin.Services;
+
+namespace ZB.MOM.WW.OtOpcUa.Admin.Tests;
+
+[Trait("Category", "Unit")]
+public sealed class RedundancyMetricsTests
+{
+    [Fact]
+    public void RecordRoleTransition_Increments_Counter_WithExpectedTags()
+    {
+        using var metrics = new RedundancyMetrics();
+        using var listener = new MeterListener();
+        var observed = new List<(long Value, Dictionary<string, object?> Tags)>();
+        listener.InstrumentPublished = (instrument, l) =>
+        {
+            if (instrument.Meter.Name == RedundancyMetrics.MeterName &&
+                instrument.Name == "otopcua.redundancy.role_transition")
+            {
+                l.EnableMeasurementEvents(instrument);
+            }
+        };
+        listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
+        {
+            var dict = new Dictionary<string, object?>();
+            foreach (var tag in tags) dict[tag.Key] = tag.Value;
+            observed.Add((value, dict));
+        });
+        listener.Start();
+
+        metrics.RecordRoleTransition("c1", "node-a", "Primary", "Secondary");
+
+        observed.Count.ShouldBe(1);
+        observed[0].Value.ShouldBe(1);
+        observed[0].Tags["cluster.id"].ShouldBe("c1");
+        observed[0].Tags["node.id"].ShouldBe("node-a");
+        observed[0].Tags["from_role"].ShouldBe("Primary");
+        observed[0].Tags["to_role"].ShouldBe("Secondary");
+    }
+
+    [Fact]
+    public void SetClusterCounts_Observed_Via_ObservableGauges()
+    {
+        using var metrics = new RedundancyMetrics();
+        metrics.SetClusterCounts("c1", primary: 1, secondary: 2, stale: 0);
+        metrics.SetClusterCounts("c2", primary: 0, secondary: 1, stale: 1);
+
+        var observations = new List<(string Name, long Value, string Cluster)>();
+        using var listener = new MeterListener();
+        listener.InstrumentPublished = (instrument, l) =>
+        {
+            if (instrument.Meter.Name == RedundancyMetrics.MeterName)
+                l.EnableMeasurementEvents(instrument);
+        };
+        listener.SetMeasurementEventCallback<long>((instrument, value, tags, _) =>
+        {
+            string? cluster = null;
+            foreach (var t in tags) if (t.Key == "cluster.id") cluster = t.Value as string;
+            observations.Add((instrument.Name, value, cluster ?? "?"));
+        });
+        listener.Start();
+        listener.RecordObservableInstruments();
+
+        observations.ShouldContain(o => o.Name == "otopcua.redundancy.primary_count" && o.Cluster == "c1" && o.Value == 1);
+        observations.ShouldContain(o => o.Name == "otopcua.redundancy.secondary_count" && o.Cluster == "c1" && o.Value == 2);
+        observations.ShouldContain(o => o.Name == "otopcua.redundancy.stale_count" && o.Cluster == "c2" && o.Value == 1);
+    }
+}

tests/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/AlarmSurfaceInvokerTests.cs

@@ -0,0 +1,127 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Resilience;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Tests.Resilience;
+
+[Trait("Category", "Unit")]
+public sealed class AlarmSurfaceInvokerTests
+{
+    private static readonly DriverResilienceOptions TierAOptions = new() { Tier = DriverTier.A };
+
+    [Fact]
+    public async Task SubscribeAsync_EmptyList_ReturnsEmpty_WithoutDriverCall()
+    {
+        var driver = new FakeAlarmSource();
+        var surface = NewSurface(driver, defaultHost: "h");
+
+        var handles = await surface.SubscribeAsync([], CancellationToken.None);
+
+        handles.Count.ShouldBe(0);
+        driver.SubscribeCallCount.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_SingleHost_RoutesThroughDefaultHost()
+    {
+        var driver = new FakeAlarmSource();
+        var surface = NewSurface(driver, defaultHost: "h1");
+
+        var handles = await surface.SubscribeAsync(["src-1", "src-2"], CancellationToken.None);
+
+        handles.Count.ShouldBe(1);
+        driver.SubscribeCallCount.ShouldBe(1);
+        driver.LastSubscribedIds.ShouldBe(["src-1", "src-2"]);
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_MultiHost_FansOutByResolvedHost()
+    {
+        var driver = new FakeAlarmSource();
+        var resolver = new StubResolver(new Dictionary<string, string>
+        {
+            ["src-1"] = "plc-a",
+            ["src-2"] = "plc-b",
+            ["src-3"] = "plc-a",
+        });
+        var surface = NewSurface(driver, defaultHost: "default-ignored", resolver: resolver);
+
+        var handles = await surface.SubscribeAsync(["src-1", "src-2", "src-3"], CancellationToken.None);
+
+        handles.Count.ShouldBe(2);                 // one per distinct host
+        driver.SubscribeCallCount.ShouldBe(2);     // one driver call per host
+    }
+
+    [Fact]
+    public async Task AcknowledgeAsync_DoesNotRetry_OnFailure()
+    {
+        var driver = new FakeAlarmSource { AcknowledgeShouldThrow = true };
+        var surface = NewSurface(driver, defaultHost: "h1");
+
+        await Should.ThrowAsync<InvalidOperationException>(() =>
+            surface.AcknowledgeAsync([new AlarmAcknowledgeRequest("s", "c", null)], CancellationToken.None));
+
+        driver.AcknowledgeCallCount.ShouldBe(1, "AlarmAcknowledge must not retry — decision #143");
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_Retries_Transient_Failures()
+    {
+        var driver = new FakeAlarmSource { SubscribeFailuresBeforeSuccess = 2 };
+        var surface = NewSurface(driver, defaultHost: "h1");
+
+        await surface.SubscribeAsync(["src"], CancellationToken.None);
+
+        driver.SubscribeCallCount.ShouldBe(3, "AlarmSubscribe retries by default — decision #143");
+    }
+
+    private static AlarmSurfaceInvoker NewSurface(
+        IAlarmSource driver,
+        string defaultHost,
+        IPerCallHostResolver? resolver = null)
+    {
+        var builder = new DriverResiliencePipelineBuilder();
+        var invoker = new CapabilityInvoker(builder, "drv-1", () => TierAOptions);
+        return new AlarmSurfaceInvoker(invoker, driver, defaultHost, resolver);
+    }
+
+    private sealed class FakeAlarmSource : IAlarmSource
+    {
+        public int SubscribeCallCount { get; private set; }
+        public int AcknowledgeCallCount { get; private set; }
+        public int SubscribeFailuresBeforeSuccess { get; set; }
+        public bool AcknowledgeShouldThrow { get; set; }
+        public IReadOnlyList<string> LastSubscribedIds { get; private set; } = [];
+
+        public Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
+            IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
+        {
+            SubscribeCallCount++;
+            LastSubscribedIds = sourceNodeIds;
+            if (SubscribeCallCount <= SubscribeFailuresBeforeSuccess)
+                throw new InvalidOperationException("transient");
+            return Task.FromResult<IAlarmSubscriptionHandle>(new StubHandle($"h-{SubscribeCallCount}"));
+        }
+
+        public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
+            => Task.CompletedTask;
+
+        public Task AcknowledgeAsync(
+            IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken)
+        {
+            AcknowledgeCallCount++;
+            if (AcknowledgeShouldThrow) throw new InvalidOperationException("ack boom");
+            return Task.CompletedTask;
+        }
+
+        public event EventHandler<AlarmEventArgs>? OnAlarmEvent { add { } remove { } }
+    }
+
+    private sealed record StubHandle(string DiagnosticId) : IAlarmSubscriptionHandle;
+
+    private sealed class StubResolver(Dictionary<string, string> map) : IPerCallHostResolver
+    {
+        public string ResolveHost(string fullReference) => map[fullReference];
+    }
+}