fix(grpc): typed alarm-feed overflow, memoized alarm snapshot, cancellable stream writes, uniform reply correlation id

- StreamAlarms: slow subscribers are completed with
  AlarmFeedSubscriberOverflowException and surface as ResourceExhausted
  (like a StreamEvents overflow) instead of warning-with-stack + Unavailable.
- GatewayAlarmMonitor.StreamAsync reuses the memoized alarm-set projection
  instead of cloning the whole set under the broadcast lock per attach; the
  prefix filter runs outside the lock.
- StreamEvents/StreamAlarms pass the call token to response writes; a write
  failure after client cancel maps to Cancelled without a warning.
- Every MxCommandReply.correlation_id now echoes the request's
  client_correlation_id (previously only constraint-denied replies did;
  normal replies leaked the internal envelope id).

src/ZB.MOM.WW.MxGateway.Server/Alarms/AlarmFeedSubscriberOverflowException.cs

@@ -0,0 +1,39 @@
+namespace ZB.MOM.WW.MxGateway.Server.Alarms;
+
+/// <summary>
+///     Completes an alarm-feed subscriber that fell behind its bounded queue and was dropped.
+/// </summary>
+/// <remarks>
+///     A typed exception rather than a bare <see cref="InvalidOperationException"/> so the gRPC
+///     layer can map it to <c>ResourceExhausted</c> — the status a <c>StreamEvents</c> overflow
+///     carries — without treating a slow client as a gateway failure (an unrecognized exception is
+///     logged as a warning with its stack and surfaced as <c>Unavailable</c>). The remedy is the
+///     same as for an event-stream overflow: reconnect, which re-snapshots the alarm set.
+/// </remarks>
+public sealed class AlarmFeedSubscriberOverflowException : Exception
+{
+    /// <summary>The message every dropped subscriber is completed with.</summary>
+    public const string DefaultMessage =
+        "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot.";
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    public AlarmFeedSubscriberOverflowException()
+        : base(DefaultMessage)
+    {
+    }
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    /// <param name="message">The error message.</param>
+    public AlarmFeedSubscriberOverflowException(string message)
+        : base(message)
+    {
+    }
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    /// <param name="message">The error message.</param>
+    /// <param name="innerException">The exception that caused this one.</param>
+    public AlarmFeedSubscriberOverflowException(string message, Exception innerException)
+        : base(message, innerException)
+    {
+    }
+}

src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs

@@ -967,7 +967,9 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
     }
 
     // Caller holds _sync. Pushes a feed message to every matching subscriber;
-    // a subscriber that has fallen behind is completed with an error and dropped.
+    // a subscriber that has fallen behind is completed with an AlarmFeedSubscriberOverflowException
+    // and dropped. The typed fault is what lets StreamAlarms answer ResourceExhausted instead of
+    // treating a slow client as a gateway failure.
     private void Broadcast(AlarmFeedMessage message, string reference)
     {
         for (int index = _subscribers.Count - 1; index >= 0; index--)
@@ -980,8 +982,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
 
             if (!subscriber.Channel.Writer.TryWrite(message))
             {
-                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
-                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
+                subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException());
                 _subscribers.RemoveAt(index);
             }
         }
@@ -997,8 +998,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
             Subscriber subscriber = _subscribers[index];
             if (!subscriber.Channel.Writer.TryWrite(message))
             {
-                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
-                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
+                subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException());
                 _subscribers.RemoveAt(index);
             }
         }
@@ -1046,9 +1046,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
             _subscribers.Add(subscriber);
             providerStatus = BuildProviderStatus();
             snapshotTruncated = _snapshotTruncated;
-            snapshot = _alarms.Values
-                .Where(alarm => prefix.Length == 0
-                    || alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal))
+
+            // Take the memoized projection rather than cloning the whole set per attach: the
+            // clone runs at most once per alarm-set generation (shared with CurrentAlarms), so a
+            // burst of reconnecting subscribers no longer holds the broadcast lock for one full
+            // copy each. The array is never mutated once built — a write to _alarms replaces it
+            // with null — so capturing the reference here pins exactly the generation this
+            // subscriber registered against, which is what keeps snapshot-then-live ordering.
+            snapshot = _currentAlarmsProjection ??= _alarms.Values
                 .Select(alarm => alarm.Clone())
                 .ToArray();
         }
@@ -1068,8 +1073,15 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
                 SnapshotStatus = new AlarmSnapshotStatus { Truncated = snapshotTruncated },
             };
 
+            // The prefix filter runs here, outside _sync, over the immutable projection.
             foreach (ActiveAlarmSnapshot alarm in snapshot)
             {
+                if (prefix.Length != 0
+                    && !alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal))
+                {
+                    continue;
+                }
+
                 yield return new AlarmFeedMessage { ActiveAlarm = alarm };
             }
 

src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs

@@ -144,6 +144,12 @@ public sealed class MxAccessGatewayService(
                 publicReply = bulkConstraintPlan.MergeDeniedInto(publicReply);
             }
 
+            // MxCommandReply.correlation_id always echoes the caller's client_correlation_id —
+            // the same value the constraint-denied replies carry — so a client can match every
+            // reply to its request. The worker reply carries the gateway's internal envelope id,
+            // which is a gateway↔worker routing detail and is overwritten here, empty included.
+            publicReply.CorrelationId = request.ClientCorrelationId;
+
             session.TrackCommandReply(commandToInvoke, publicReply);
             return publicReply;
         }
@@ -172,7 +178,10 @@ public sealed class MxAccessGatewayService(
                 // volume gateway path. Stopwatch.GetTimestamp/GetElapsedTime measure the exact
                 // same wall-clock span the former Stopwatch.StartNew()/.Elapsed did.
                 long sendStartTimestamp = Stopwatch.GetTimestamp();
-                await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
+
+                // The call token releases a write parked on a slow or vanished client when the
+                // call is cancelled, instead of leaving it to the transport's own timeout.
+                await responseStream.WriteAsync(publicEvent, context.CancellationToken).ConfigureAwait(false);
                 metrics.RecordEventStreamSend(
                     publicEvent.Family.ToString(),
                     Stopwatch.GetElapsedTime(sendStartTimestamp));
@@ -180,6 +189,11 @@ public sealed class MxAccessGatewayService(
         }
         catch (Exception exception) when (exception is not RpcException)
         {
+            if (IsClientAbort(exception, context))
+            {
+                throw ClientCancelled();
+            }
+
             throw MapException(exception);
         }
     }
@@ -225,11 +239,24 @@ public sealed class MxAccessGatewayService(
                 .WithCancellation(context.CancellationToken)
                 .ConfigureAwait(false))
             {
-                await responseStream.WriteAsync(message).ConfigureAwait(false);
+                await responseStream.WriteAsync(message, context.CancellationToken).ConfigureAwait(false);
             }
         }
+        catch (AlarmFeedSubscriberOverflowException overflow)
+        {
+            // The monitor dropped this subscriber for falling behind its bounded queue. That is a
+            // client-side slowness condition with a client-side remedy (reconnect and
+            // re-snapshot), so it maps to ResourceExhausted like a StreamEvents overflow rather
+            // than reaching MapException's warning-with-stack + Unavailable fallback.
+            throw new RpcException(new Status(StatusCode.ResourceExhausted, overflow.Message));
+        }
         catch (Exception exception) when (exception is not RpcException)
         {
+            if (IsClientAbort(exception, context))
+            {
+                throw ClientCancelled();
+            }
+
             throw MapException(exception);
         }
     }
@@ -264,6 +291,17 @@ public sealed class MxAccessGatewayService(
         }
     }
 
+    // A response write fails with an IOException (or an operation-cancelled exception) when the
+    // client disconnects mid-stream. With the call token already cancelled that is a routine
+    // client abort, not a gateway fault: it maps to Cancelled with no warning. The same exception
+    // on a live call token is a genuine transport failure and keeps the MapException path.
+    private static bool IsClientAbort(Exception exception, ServerCallContext context) =>
+        context.CancellationToken.IsCancellationRequested
+        && exception is IOException or OperationCanceledException;
+
+    private static RpcException ClientCancelled() =>
+        new(new Status(StatusCode.Cancelled, "gRPC request was canceled."));
+
     private string? ResolveClientIdentity()
     {
         return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId;