Merge stream M (cross-stream review fixes: terminal-worker attach, fault double-count, cancelled/stopping opens, docs)
This commit is contained in:
Joseph Doherty
@@ -641,7 +641,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
/// </summary>
|
||||
/// <param name="serverHandle">The ServerHandle from register.</param>
|
||||
/// <param name="tagAddresses">Tag addresses to read (one per result).</param>
|
||||
/// <param name="timeout">Per-call timeout for the snapshot lifecycle path; <see cref="TimeSpan.Zero"/> uses the gateway default.</param>
|
||||
/// <param name="timeout">One shared snapshot deadline for the whole command, not a per-tag budget: the worker advises every uncached tag, waits once for all of them, then tears them down. <see cref="TimeSpan.Zero"/> uses the worker default (1000 ms).</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>One <see cref="BulkReadResult"/> per requested tag, in request order.</returns>
|
||||
public async Task<IReadOnlyList<BulkReadResult>> ReadBulkAsync(
|
||||
|
||||
@@ -571,7 +571,7 @@ func runReadBulk(ctx context.Context, args []string, stdout, stderr io.Writer) e
|
||||
sessionID := flags.String("session-id", "", "gateway session id")
|
||||
serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
|
||||
items := flags.String("items", "", "comma-separated tag addresses")
|
||||
timeoutMs := flags.Int("timeout-ms", 0, "per-tag snapshot timeout in milliseconds (0 = worker default)")
|
||||
timeoutMs := flags.Int("timeout-ms", 0, "snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms)")
|
||||
|
||||
if err := flags.Parse(args); err != nil {
|
||||
return err
|
||||
@@ -754,7 +754,7 @@ func runBenchReadBulk(ctx context.Context, args []string, stdout, stderr io.Writ
|
||||
tagStart := flags.Int("tag-start", 1, "first machine number")
|
||||
tagPrefix := flags.String("tag-prefix", "TestMachine_", "tag prefix (machine number appended as %03d)")
|
||||
tagAttribute := flags.String("tag-attribute", "TestChangingInt", "attribute appended to each tag prefix")
|
||||
timeoutMs := flags.Int("timeout-ms", 1500, "per-tag snapshot timeout in milliseconds")
|
||||
timeoutMs := flags.Int("timeout-ms", 1500, "snapshot deadline in milliseconds, shared by all tags in each ReadBulk command")
|
||||
|
||||
if err := flags.Parse(args); err != nil {
|
||||
return err
|
||||
|
||||
@@ -549,8 +549,9 @@ func (s *Session) WriteSecured2Bulk(ctx context.Context, serverHandle int32, ent
|
||||
// MXAccess COM has no synchronous Read; the worker satisfies this by returning the
|
||||
// most recent cached OnDataChange value when the tag is already advised (WasCached=true),
|
||||
// or by taking a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle
|
||||
// otherwise. timeout bounds the wait per tag in the snapshot case; pass zero to use the
|
||||
// worker default. Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries
|
||||
// otherwise. timeout is one shared deadline for the whole command, not a per-tag budget:
|
||||
// the worker advises every uncached tag, waits once for all of them, then tears them down.
|
||||
// Pass zero to use the worker default (1000 ms). Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries
|
||||
// with WasSuccessful=false; the call never returns an error for per-tag MXAccess failures.
|
||||
//
|
||||
// A non-nil but empty tagAddresses slice is treated as a no-op and returns an empty
|
||||
|
||||
@@ -49125,11 +49125,15 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
|
||||
* reply returns the cached value WITHOUT modifying the existing
|
||||
* subscription (was_cached = true).
|
||||
* - Otherwise the worker takes the snapshot lifecycle itself: AddItem +
|
||||
* Advise, wait up to `timeout_ms` for the first OnDataChange, then
|
||||
* UnAdvise + RemoveItem before returning. The session is left exactly
|
||||
* as it was before the call (was_cached = false).
|
||||
* Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem
|
||||
* before returning. The session is left exactly as it was before the
|
||||
* call (was_cached = false).
|
||||
*
|
||||
* `timeout_ms == 0` uses the gateway-configured default (1000 ms).
|
||||
* `timeout_ms` is one deadline shared by the whole command, not a per-tag
|
||||
* budget: the worker adds and advises every uncached tag, waits once for all
|
||||
* of them, then tears them all down. A tag that has not delivered by the
|
||||
* deadline gets its own was_successful = false result.
|
||||
* `timeout_ms == 0` uses the worker's default (1000 ms).
|
||||
* </pre>
|
||||
*
|
||||
* Protobuf type {@code mxaccess_gateway.v1.ReadBulkCommand}
|
||||
@@ -49424,11 +49428,15 @@ public final class MxaccessGateway extends com.google.protobuf.GeneratedFile {
|
||||
* reply returns the cached value WITHOUT modifying the existing
|
||||
* subscription (was_cached = true).
|
||||
* - Otherwise the worker takes the snapshot lifecycle itself: AddItem +
|
||||
* Advise, wait up to `timeout_ms` for the first OnDataChange, then
|
||||
* UnAdvise + RemoveItem before returning. The session is left exactly
|
||||
* as it was before the call (was_cached = false).
|
||||
* Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem
|
||||
* before returning. The session is left exactly as it was before the
|
||||
* call (was_cached = false).
|
||||
*
|
||||
* `timeout_ms == 0` uses the gateway-configured default (1000 ms).
|
||||
* `timeout_ms` is one deadline shared by the whole command, not a per-tag
|
||||
* budget: the worker adds and advises every uncached tag, waits once for all
|
||||
* of them, then tears them all down. A tag that has not delivered by the
|
||||
* deadline gets its own was_successful = false result.
|
||||
* `timeout_ms == 0` uses the worker's default (1000 ms).
|
||||
* </pre>
|
||||
*
|
||||
* Protobuf type {@code mxaccess_gateway.v1.ReadBulkCommand}
|
||||
|
||||
+2
-2
@@ -1262,7 +1262,7 @@ public final class MxGatewayCli implements Callable<Integer> {
|
||||
@Option(
|
||||
names = "--timeout-ms",
|
||||
defaultValue = "0",
|
||||
description = "Per-tag snapshot timeout in milliseconds (0 = worker default).")
|
||||
description = "Snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms).")
|
||||
int timeoutMs;
|
||||
|
||||
ReadBulkCommand(MxGatewayCliClientFactory clientFactory) {
|
||||
@@ -1510,7 +1510,7 @@ public final class MxGatewayCli implements Callable<Integer> {
|
||||
@Option(
|
||||
names = "--timeout-ms",
|
||||
defaultValue = "0",
|
||||
description = "Per-tag snapshot timeout in milliseconds (0 = worker default).")
|
||||
description = "Snapshot deadline in milliseconds, shared by all tags in the command (0 = worker default, 1000 ms).")
|
||||
int timeoutMs;
|
||||
|
||||
@Option(names = "--iterations", defaultValue = "10", description = "Number of ReadBulk calls to perform.")
|
||||
|
||||
+6
-4
@@ -555,15 +555,17 @@ public final class MxGatewaySession implements AutoCloseable {
|
||||
* {@code OnDataChange} value for any tag that is already advised
|
||||
* ({@code wasCached == true}) without modifying the existing subscription,
|
||||
* and falls back to a full AddItem + Advise + wait + UnAdvise + RemoveItem
|
||||
* snapshot lifecycle otherwise. The supplied {@code timeout} bounds the
|
||||
* per-tag wait in the snapshot case; pass {@link Duration#ZERO} (or
|
||||
* {@code null}) to use the worker default (1000 ms). Per-tag failures
|
||||
* snapshot lifecycle otherwise. The supplied {@code timeout} is one shared
|
||||
* deadline for the whole command, not a per-tag budget: the worker advises
|
||||
* every uncached tag, waits once for all of them, then tears them down.
|
||||
* Pass {@link Duration#ZERO} (or {@code null}) to use the worker default
|
||||
* (1000 ms). Per-tag failures
|
||||
* appear as {@link BulkReadResult} entries with {@code wasSuccessful == false};
|
||||
* this method does not throw for per-tag MXAccess failures.
|
||||
*
|
||||
* @param serverHandle the {@code ServerHandle} owning the items
|
||||
* @param tagAddresses the tag addresses to read
|
||||
* @param timeout per-tag snapshot timeout (zero or null = worker default)
|
||||
* @param timeout snapshot deadline shared by all tags in the command (zero or null = worker default)
|
||||
* @return a per-tag {@link BulkReadResult} list
|
||||
* @throws MxGatewayException on transport or protocol failure
|
||||
* @throws NullPointerException if {@code tagAddresses} is {@code null}
|
||||
|
||||
@@ -448,8 +448,10 @@ class Session:
|
||||
``OnDataChange`` value for any tag that is already advised (``was_cached =
|
||||
True``) without modifying the existing subscription, and falls back to
|
||||
a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle
|
||||
otherwise. ``timeout_ms`` bounds the per-tag wait in the snapshot case;
|
||||
pass ``0`` to use the worker default (1000 ms).
|
||||
otherwise. ``timeout_ms`` is one shared deadline for the whole command,
|
||||
not a per-tag budget: the worker advises every uncached tag, waits once
|
||||
for all of them, then tears them down. Pass ``0`` to use the worker
|
||||
default (1000 ms).
|
||||
"""
|
||||
if tag_addresses is None:
|
||||
raise TypeError("tag_addresses is required")
|
||||
|
||||
@@ -392,7 +392,7 @@ def unsubscribe_bulk(**kwargs: Any) -> None:
|
||||
@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
|
||||
@click.option("--items", required=True, help="Comma-separated MXAccess tag addresses.")
|
||||
@click.option("--timeout-ms", default=0, type=int, show_default=True,
|
||||
help="Per-tag snapshot timeout in milliseconds. 0 = worker default.")
|
||||
help="Snapshot deadline in milliseconds, shared by all tags in the command. 0 = worker default (1000 ms).")
|
||||
@click.option("--correlation-id", default="", help="Client correlation id.")
|
||||
@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
|
||||
def read_bulk(**kwargs: Any) -> None:
|
||||
|
||||
@@ -215,7 +215,7 @@ enum Command {
|
||||
server_handle: i32,
|
||||
#[arg(long, value_delimiter = ',')]
|
||||
items: Vec<String>,
|
||||
/// Per-tag snapshot timeout in milliseconds. `0` uses the worker default (1000 ms).
|
||||
/// Snapshot deadline in milliseconds, shared by all tags in the command. `0` uses the worker default (1000 ms).
|
||||
#[arg(long, default_value_t = 0)]
|
||||
timeout_ms: u32,
|
||||
#[arg(long)]
|
||||
|
||||
@@ -533,7 +533,10 @@ impl Session {
|
||||
/// returning the most recent cached `OnDataChange` value when the tag is
|
||||
/// already advised (`was_cached = true`), or by taking a full AddItem +
|
||||
/// Advise + wait + UnAdvise + RemoveItem snapshot lifecycle otherwise.
|
||||
/// `timeout_ms == 0` lets the worker pick its default (1000 ms).
|
||||
/// `timeout_ms` is one shared deadline for the whole command, not a
|
||||
/// per-tag budget: the worker advises every uncached tag, waits once for
|
||||
/// all of them, then tears them down. `timeout_ms == 0` lets the worker
|
||||
/// pick its default (1000 ms).
|
||||
/// Per-tag failures appear as `BulkReadResult` entries with
|
||||
/// `was_successful = false`; the call never errors on per-tag failure.
|
||||
///
|
||||
|
||||
@@ -154,8 +154,8 @@ validation fails fast when either half of the headroom rule is violated.
|
||||
| `MxGateway:Sessions:LeaseSweepIntervalSeconds` | `30` | Hosted monitor interval for closing expired leases. Active event-stream subscribers keep a session from expiring while the stream remains attached. |
|
||||
| `MxGateway:Sessions:DetachGraceSeconds` | `30` | Detach-grace retention window. When positive, a session whose last external (gRPC) event-stream subscriber drops is retained in `Ready` for this many seconds so a client can reconnect; if no external subscriber re-attaches within the window, the lease monitor closes it with `detach-grace-expired`. The internal dashboard mirror does not count as an external subscriber, so a dashboard-only session still enters detach-grace. `0` disables retention and reverts to closing only on normal lease expiry. Must be zero or greater. Reconnect/replay itself is implemented separately (Task 12); this option controls retention and expiry only. The effective close happens within the next sweep cycle after the window elapses — up to `LeaseSweepIntervalSeconds` after expiry. Operators wanting a firm minimum retention bound should set `DetachGraceSeconds` greater than `LeaseSweepIntervalSeconds`. |
|
||||
| `MxGateway:Sessions:FaultedGraceSeconds` | `0` | Grace window before the lease monitor reaps a faulted session (killing its worker and freeing the slot). A faulted session is permanently unusable yet otherwise pins a session slot and a live x86 worker until its normal lease expires. `0` (the default) reaps it on the next sweep cycle, bounding blast radius; a positive value keeps the faulted session observable via `GetSessionStatus` for that window before it is reclaimed, with the close reason `faulted-reaped`. Must be zero or greater. |
|
||||
| `MxGateway:Sessions:UnattachedSessionGraceSeconds` | `0` | Idle window after which the lease monitor reaps a `Ready` session that has **never** attached a `StreamEvents` subscriber, with the close reason `unattached-grace-expired`. Exists because a client that opens a session and vanishes before it ever streams — OtOpcUa's `GalaxyDriver` failing part-way through start-up, for example — otherwise pins a session slot and a live x86 worker for the full `DefaultLeaseSeconds` (30 minutes). The window is measured from the later of the session becoming `Ready` and its last unary command, so a slow worker start does not eat it and a unary-only client that keeps calling is never reaped by this rule; it stops applying once the session's first subscriber attaches (`DetachGraceSeconds` and the normal lease govern it from then on), and the gateway's own alarm-monitor session, whose subscriber is internal, is exempt the same way. `0` (the default) disables it. Must be zero or greater. Set it comfortably above the time a well-behaved client needs between `OpenSession` and its first `StreamEvents`; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. |
|
||||
| `MxGateway:Sessions:AllowMultipleEventSubscribers` | `false` | Controls whether multiple `StreamEvents` subscribers may attach to one session. When `false` the session refuses a second subscriber with `AlreadyExists`. Set to `true` to enable fan-out via the `SessionEventDistributor`. |
|
||||
| `MxGateway:Sessions:UnattachedSessionGraceSeconds` | `0` | Idle window after which the lease monitor reaps a `Ready` session that has **never** attached a `StreamEvents` subscriber, with the close reason `unattached-grace-expired`. Exists because a client that opens a session and vanishes before it ever streams — OtOpcUa's `GalaxyDriver` failing part-way through start-up, for example — otherwise pins a session slot and a live x86 worker for the full `DefaultLeaseSeconds` (30 minutes). The window is measured from the later of the session becoming `Ready` and its last unary command, so a slow worker start does not eat it and a unary-only client that keeps calling is never reaped by this rule; it stops applying once the session's first subscriber attaches (`DetachGraceSeconds` and the normal lease govern it from then on), and the gateway's own alarm-monitor session, whose subscriber is internal, is exempt the same way. The dashboard Browse live-data session (`DashboardLiveDataService`) is not exempt: it is unary-only, so it is reaped once idle for the window, and it opens a new session through `EnsureReadyAsync` on its next use. `0` (the default) disables it. Must be zero or greater. Set it comfortably above the time a well-behaved client needs between `OpenSession` and its first `StreamEvents`; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. |
|
||||
| `MxGateway:Sessions:AllowMultipleEventSubscribers` | `false` | Controls whether multiple `StreamEvents` subscribers may attach to one session. When `false` the session refuses a second subscriber with `ResourceExhausted` (`EventSubscriberAlreadyActive`). Set to `true` to enable fan-out via the `SessionEventDistributor`. |
|
||||
| `MxGateway:Sessions:MaxEventSubscribersPerSession` | `8` | Maximum number of concurrent `StreamEvents` subscribers per session when `AllowMultipleEventSubscribers` is `true`. Effectively 1 when `AllowMultipleEventSubscribers` is `false`. Must be greater than zero. |
|
||||
| `MxGateway:Sessions:WorkerReadyWaitTimeoutMs` | `0` | Bounded time, in milliseconds, the gateway will wait for a worker to reach `Ready` when the session is already `Ready` but the worker state has transiently diverged (e.g. `Handshaking` after a heartbeat blip). Applies only to transient worker states; terminal states (`Faulted`/`Closing`/`Closed`/no worker) fail fast immediately regardless of this setting. `0` (the default) disables the wait and preserves the original fail-fast behavior. Must be greater than or equal to zero. |
|
||||
|
||||
|
||||
+12
-45
@@ -311,6 +311,7 @@ StatusCode statusCode = exception.ErrorCode switch
|
||||
SessionManagerErrorCode.SessionNotFound => StatusCode.NotFound,
|
||||
SessionManagerErrorCode.SessionNotReady => StatusCode.FailedPrecondition,
|
||||
SessionManagerErrorCode.EventSubscriberAlreadyActive => StatusCode.ResourceExhausted,
|
||||
SessionManagerErrorCode.EventSubscriberLimitReached => StatusCode.ResourceExhausted,
|
||||
SessionManagerErrorCode.EventQueueOverflow => StatusCode.ResourceExhausted,
|
||||
SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
|
||||
SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
|
||||
@@ -323,64 +324,30 @@ StatusCode statusCode = exception.ErrorCode switch
|
||||
|
||||
`WorkerClientException` follows the same pattern: `CommandTimeout` becomes `DeadlineExceeded`, `GatewayShutdown` and `CallerCancelled` both become `Cancelled`, `InvalidState` becomes `FailedPrecondition`, `ProtocolViolation` becomes `Internal`, `CommandTooLarge` becomes `ResourceExhausted`, and unmapped codes fall through to `Unavailable`. `CallerCancelled` is the caller's own token firing — a per-call deadline shorter than the gateway command timeout, or a disconnect — and is separate from `GatewayShutdown` only so metrics and logs stop attributing those calls to a shutdown that did not happen; the status the client observes is the same.
|
||||
|
||||
While the gateway is stopping, clients see three different outcomes. An in-flight `Invoke` gets `Cancelled` (`GatewayShutdown`). An active `StreamEvents` ends cleanly with `OK`, because the drain closes its session and completes the subscriber channel. A new `OpenSession` gets `Unavailable` (`GatewayStopping`), including an open whose session the drain closed while its worker was still starting. An `OpenSession` the caller cancels while its worker is starting gets `Cancelled`, not `Unavailable`, and is not counted as an open fault.
|
||||
|
||||
## Event Streaming Model
|
||||
|
||||
`EventStreamService` implements the `StreamEvents` pipeline. It exists as a separate service (rather than living inside `MxAccessGatewayService`) so that the channel, backpressure policy, and metric bookkeeping can be unit-tested without spinning up a Kestrel host.
|
||||
`EventStreamService` implements the `StreamEvents` pipeline. It exists as a separate service (rather than living inside `MxAccessGatewayService`) so that the attach, replay, filtering and metric bookkeeping can be unit-tested without spinning up a Kestrel host. It does not read the worker itself: each session owns one `SessionEventDistributor` whose single pump drains the worker event channel, maps each `WorkerEvent` to the public `MxEvent`, appends it to the session's replay ring, and fans it to every registered subscriber. A `StreamEvents` call is one of those subscribers. [Sessions](./Sessions.md) describes the distributor, the replay ring and detach-grace in full; this section covers the per-RPC boundary.
|
||||
|
||||
The pipeline has three stages:
|
||||
The pipeline has four stages:
|
||||
|
||||
1. The session is resolved via `sessionManager.TryGetSession`. A miss raises `SessionManagerException(SessionNotFound)` so the handler reports `StatusCode.NotFound`.
|
||||
2. `session.AttachEventSubscriber(...)` enforces the single-subscriber-per-session rule (or allows multiple subscribers if `Sessions:AllowMultipleEventSubscribers` is enabled). The returned `IDisposable` is released in the `finally` block, ensuring the subscriber slot is freed even when the client cancels mid-stream.
|
||||
3. A `Channel<MxEvent>` decouples the worker-side producer from the gRPC writer. The channel is bounded by `Events:QueueCapacity` and configured for a single reader and writer:
|
||||
2. The attach is owner-scoped: the caller's API key id must match the key that opened the session, or the call fails with `PermissionDenied` before anything is attached or delivered.
|
||||
3. The call registers a subscriber on the distributor. A fresh stream (`after_worker_sequence == 0`) uses `session.AttachEventSubscriber(...)`. A resume uses `session.AttachEventSubscriberWithReplay(...)`, which snapshots the replay ring for events newer than the cursor and registers the live subscriber under one lock, so the replay-to-live handoff has no gap and no duplicate. In single-subscriber mode (the default) a second attach fails with `EventSubscriberAlreadyActive`; with `Sessions:AllowMultipleEventSubscribers` enabled, an attach beyond `MaxEventSubscribersPerSession` fails with `EventSubscriberLimitReached`. Both map to `ResourceExhausted`. The lease is released in the `finally` block, so the subscriber slot is freed even when the client cancels mid-stream.
|
||||
4. The call yields events from the lease's bounded channel (capacity `Events:QueueCapacity`). On a resume it first yields a `ReplayGap` sentinel when the cursor predates the oldest retained event, then the replayed batch, then live events filtered to sequences above the last replayed one. Per-session ordering is preserved end-to-end, because the pump is the single writer to each subscriber channel and writes in `WorkerSequence` order.
|
||||
|
||||
```csharp
|
||||
Channel<MxEvent> eventQueue = Channel.CreateBounded<MxEvent>(
|
||||
new BoundedChannelOptions(options.Value.Events.QueueCapacity)
|
||||
{
|
||||
SingleReader = true,
|
||||
SingleWriter = true,
|
||||
FullMode = BoundedChannelFullMode.Wait,
|
||||
AllowSynchronousContinuations = false,
|
||||
});
|
||||
```
|
||||
|
||||
The producer reads `WorkerEvent`s from the session, maps them, and writes to the channel. Per-session ordering is preserved end-to-end: events arrive from the worker in `WorkerSequence` order, the producer is the single writer, and the consumer drains FIFO. The producer also honours the request's `AfterWorkerSequence` cursor by skipping any event whose `WorkerSequence` is at or before the requested cutoff, which lets clients resume after a disconnect without server-side replay state.
|
||||
The replay ring is server-side state that outlives a single RPC: with the default `DetachGraceSeconds` and `ReplayRetentionSeconds`, a client that drops can reconnect with its last `worker_sequence` and receive what it missed, as long as the events are still retained.
|
||||
|
||||
### Backpressure
|
||||
|
||||
When `TryWrite` fails the queue is full. The handling depends on `Events:BackpressurePolicy`:
|
||||
|
||||
```csharp
|
||||
if (!writer.TryWrite(publicEvent))
|
||||
{
|
||||
string message = $"Session {session.SessionId} event stream queue overflowed.";
|
||||
metrics.QueueOverflow("grpc-event-stream");
|
||||
if (options.Value.Events.BackpressurePolicy == EventBackpressurePolicy.FailFast)
|
||||
{
|
||||
session.MarkFaulted(message);
|
||||
metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString());
|
||||
}
|
||||
else
|
||||
{
|
||||
logger.LogDebug(
|
||||
"Disconnecting event stream for session {SessionId} after queue overflow.",
|
||||
session.SessionId);
|
||||
}
|
||||
|
||||
writer.TryComplete(new SessionManagerException(
|
||||
SessionManagerErrorCode.EventQueueOverflow,
|
||||
message));
|
||||
return;
|
||||
}
|
||||
```
|
||||
|
||||
Under `FailFast` the session is faulted so subsequent commands return `FailedPrecondition`; the client must reopen. Under the default policy only the stream is dropped and the session continues to accept commands, leaving recovery to the client (typically a fresh `StreamEvents` call with an updated `AfterWorkerSequence`). Either way, the consumer side observes `StatusCode.ResourceExhausted` via the `EventQueueOverflow` mapping above.
|
||||
The pump never blocks on a slow subscriber: it writes to each subscriber channel without waiting. When one subscriber's channel is full, the distributor's overflow handler records `QueueOverflow("grpc-event-stream")` and completes that subscriber's channel with `SessionManagerException(EventQueueOverflow)`, which the client observes as `ResourceExhausted`. Under `Events:BackpressurePolicy = FailFast` (the default) in single-subscriber mode the session is also faulted (and `Fault(EventQueueOverflow)` recorded), so subsequent commands return `FailedPrecondition` and the client must reopen. In multi-subscriber mode, or under `DisconnectSubscriber`, only that stream is dropped and the session keeps serving commands and other subscribers; the client recovers with a fresh `StreamEvents` that resumes from its last `worker_sequence`. The internal dashboard mirror is registered as an internal subscriber, so a slow dashboard only disconnects itself and never faults the session.
|
||||
|
||||
### Cancellation and cleanup
|
||||
|
||||
The handler creates a linked cancellation token (`streamCts`) so that completing the consumer (client disconnect, error, or graceful end-of-stream) also cancels the producer. The `finally` block cancels the source, disposes the subscriber slot, awaits the producer (swallowing the expected cancellation), and emits `StreamDisconnected(reason)` so dashboards see every disconnection. The reason is the bounded constant (`EventStreamDisconnectReasons`) for how the stream ended — `SessionClosed`, `Overflow`, `WorkerFault`, `ClientCancelled`, `Error`, or `Detached` — rather than one constant for every cause, so an operator can tell slow consumers from worker faults from routine client disconnects.
|
||||
The `finally` block removes the subscriber's backlog-gauge contribution, disposes the lease, and emits `StreamDisconnected(reason)` so dashboards see every disconnection. The reason is the bounded constant (`EventStreamDisconnectReasons`) for how the stream ended — `SessionClosed`, `Overflow`, `WorkerFault`, `ClientCancelled`, `Error`, or `Detached` — rather than one constant for every cause, so an operator can tell slow consumers from worker faults from routine client disconnects.
|
||||
|
||||
`WorkerClientException` thrown by the producer marks the session as faulted before completing the channel — the worker is presumed gone, and any subsequent command on that session must observe the fault rather than silently retry.
|
||||
When the worker event stream ends with a `WorkerClientException`, the pump completes every subscriber channel with it. Each subscriber marks the session faulted, so any subsequent command observes the fault rather than silently retrying, and records the `WorkerFault` disconnect reason. The fault itself is counted once in `mxgateway.faults`, by `WorkerClient` when it faults, however many subscribers observe it.
|
||||
|
||||
## Authorization Interceptor Integration
|
||||
|
||||
|
||||
+3
-3
@@ -124,7 +124,7 @@ The recording call sites describe the code paths that write into each instrument
|
||||
```csharp
|
||||
_metrics.SessionOpened();
|
||||
...
|
||||
_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString());
|
||||
_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString()); // not for a cancelled open or GatewayStopping
|
||||
...
|
||||
_metrics.SessionClosed();
|
||||
...
|
||||
@@ -185,10 +185,10 @@ metrics.StreamDisconnected(disconnectReason); // an EventStreamDisconnectReasons
|
||||
...
|
||||
metrics.QueueOverflow("grpc-event-stream");
|
||||
metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString());
|
||||
...
|
||||
metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
|
||||
```
|
||||
|
||||
A worker fault seen by a subscriber records only the `WorkerFault` disconnect reason, not a fault. `WorkerClient` records the fault once when it faults, and every subscriber of the session observes that same fault, so counting it per subscriber made one worker fault read as 1+K faults for K subscribers.
|
||||
|
||||
Rather than pushing a running total on every streamed event (which took the subscriber channel's `Count` lock and the metric lock per event), each subscriber registers its channel as a backlog source for the duration of the stream. The gauge callback and `GetSnapshot` sum those sources' `Count` on demand — only when the metric is scraped or projected — so the streaming hot path does no per-event gauge bookkeeping (GWC-15). Disposing the registration in the `finally` removes the subscriber's contribution, so the gauge reflects only the remaining subscribers and returns to zero once the last stream detaches. Negative source returns are clamped to zero when summed.
|
||||
|
||||
`Grpc/MxAccessGatewayService.cs` records gRPC event send latency around each response-stream write using the allocation-free timestamp API (no per-event `Stopwatch` object):
|
||||
|
||||
@@ -996,7 +996,7 @@ has been stale for longer than `HeartbeatStuckCeiling`, the watchdog fires
|
||||
healthy long-running commands: `StaRuntime.PumpPendingMessages()` refreshes
|
||||
`LastStaActivityUtc` (via `MarkActivity()`) every time it runs, and long-hold
|
||||
STA commands invoke it on every wait iteration (`ReadBulk` routes its
|
||||
per-tag wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`).
|
||||
shared snapshot wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`).
|
||||
A command that keeps pumping therefore keeps its activity timestamp fresh and
|
||||
never reaches the ceiling, while a genuinely stuck STA — one that has stopped
|
||||
pumping — accrues staleness and faults correctly. The ceiling is thus the
|
||||
|
||||
+14
-7
@@ -163,7 +163,7 @@ public sealed record SessionCloseResult(
|
||||
| `EventSubscriberLimitReached` | In multi-subscriber mode, an attach exceeded `MaxEventSubscribersPerSession` concurrent external subscribers. |
|
||||
| `EventQueueOverflow` | Reserved for the worker event channel overflow path. |
|
||||
| `SessionLimitExceeded` | `MaxSessions` is in use. |
|
||||
| `OpenFailed` | `OpenSessionAsync` failed; the inner exception carries the cause. |
|
||||
| `OpenFailed` | `OpenSessionAsync` failed; the inner exception carries the cause. A caller-cancelled open and an open refused because the gateway is stopping do not use this code. |
|
||||
| `CloseFailed` | A close started but did not complete cleanly — the worker's graceful shutdown and the kill fall-back both failed (or a kill threw); the session is removed and faulted. A shutdown failure whose kill succeeded is a successful close, not this. |
|
||||
| `PermissionDenied` | The caller may not perform the operation on this session (for example, attaching `StreamEvents` to a session opened by a different API key). |
|
||||
| `GatewayStopping` | The gateway is stopping and refuses new sessions (see [Shutdown Coordination](#shutdown-coordination)). Maps to gRPC `Unavailable`; not recorded as a fault. |
|
||||
@@ -172,9 +172,14 @@ public sealed record SessionCloseResult(
|
||||
|
||||
### Open
|
||||
|
||||
`SessionManager.OpenSessionAsync` allocates a session slot, builds the `GatewaySession`, registers it, and asks the factory to bring up the worker. Failures roll back every preceding step:
|
||||
`SessionManager.OpenSessionAsync` allocates a session slot, builds the `GatewaySession`, registers it, and asks the factory to bring up the worker. Failures roll back every preceding step. Two outcomes are not open failures and skip the fault metric and the warning: `GatewayStopping` (see [Stopping gate](#stopping-gate)) and the caller's own cancellation, which is rethrown as the `OperationCanceledException` so the gRPC layer maps it to `Cancelled` rather than `Unavailable`:
|
||||
|
||||
```csharp
|
||||
catch (OperationCanceledException exception) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
|
||||
@@ -197,9 +202,11 @@ catch (Exception exception)
|
||||
|
||||
The session is registered — and therefore visible to the dashboard's Kill button, the lease sweep and the host drain — for the whole worker startup, which can last up to `MxGateway:Worker:StartupTimeoutSeconds`. A kill or drain that lands in that window finds no worker client to stop, flips the session to `Closed`, removes it and disposes it. The worker client that `SessionWorkerClientFactory.CreateAsync` then returns must not be stored on that dead session and handed back to the caller as a successful open: nothing would ever shut the worker down (the close already ran) and the client would hold a `Closed` session.
|
||||
|
||||
`GatewaySession.AttachWorkerClient` therefore returns `false`, under `_syncRoot`, when the session is already `Closing`, `Closed` or `Faulted` or has been disposed. The manager runs the end of the open sequence — attach, `MarkReady`, `SessionOpened` — through `TryAttachWorkerClientAndMarkReadyAsync`, which holds the per-session close gate (`_closeLock`) so a concurrent `CloseAsync` / `KillWorkerWithCloseGateAsync` runs either entirely before it (the attach is refused) or entirely after it (an ordinary close of a Ready session). On a refused attach the manager kills and disposes the late worker client and the open fails with `OpenFailed`; the registry is already empty and the slot was released by whichever teardown closed the session.
|
||||
`GatewaySession.AttachWorkerClient` therefore returns `false`, under `_syncRoot`, when the session is already `Closing`, `Closed` or `Faulted` or has been disposed. The manager runs the end of the open sequence — attach, `MarkReady`, `SessionOpened` — through `TryAttachWorkerClientAndMarkReadyAsync`, which holds the per-session close gate (`_closeLock`) so a concurrent `CloseAsync` / `KillWorkerWithCloseGateAsync` runs either entirely before it (the attach is refused) or entirely after it (an ordinary close of a Ready session). `TryAttachWorkerClientAndMarkReadyAsync` returns a `SessionAttachOutcome` rather than a bool, because the open reports each refusal differently. On a refused attach the manager kills and disposes the late worker client; the registry is already empty and the slot was released by whichever teardown closed the session. A refusal caused by the host drain fails the open with `GatewayStopping`, the same code as an open refused at the door, so the client retries against the restarted gateway. A refusal caused by an admin kill fails it with `OpenFailed`.
|
||||
|
||||
Once `CreateAsync` has returned, a live worker exists that only the open path knows about, so the gate wait is **not cancellable** (it waits on `CancellationToken.None`; the gate is only ever held for the duration of one close or kill) and the manager discards the client on *any* exception out of the attach step — the caller's own cancellation (checked before the attach, so an abandoned open never yields a session), a disposed gate, anything thrown while becoming Ready. The failure then surfaces as `OpenFailed` with the cause as inner exception. An exception that escaped before the discard would be an orphaned x86 worker, which a restarted gateway terminates rather than adopts.
|
||||
The attach also refuses a worker client that is already `Faulted` or `Closed` (`SessionAttachOutcome.WorkerTerminal`). A worker can die right after its handshake, for example from an STA loop fault or exit code 7. Attaching it would hand the caller a `Ready` session over a dead worker, and nothing would notice until the lease expired. The client is discarded and the open fails with `OpenFailed`. A worker that dies just after this check is caught by the event pump instead (see [Worker exit outside a gateway close](#worker-exit-outside-a-gateway-close)).
|
||||
|
||||
Once `CreateAsync` has returned, a live worker exists that only the open path knows about, so the gate wait is **not cancellable** (it waits on `CancellationToken.None`; the gate is only ever held for the duration of one close or kill) and the manager discards the client on *any* exception out of the attach step — the caller's own cancellation (checked before the attach, so an abandoned open never yields a session), a disposed gate, anything thrown while becoming Ready. A cancellation surfaces as the `OperationCanceledException` itself; any other failure surfaces as `OpenFailed` with the cause as inner exception. An exception that escaped before the discard would be an orphaned x86 worker, which a restarted gateway terminates rather than adopts.
|
||||
|
||||
Because such a session was never counted as opened, it must not be counted as closed either. `GatewaySession.OpenRecorded` is set inside the same gate right after `SessionOpened` is recorded, and every teardown path (`KillWorkerAsync`, `CloseSessionCoreAsync`, the close-failed branch) records `SessionClosed` / `SessionRemoved` only when it is set — so `sessions.opened`, `sessions.closed` and the `sessions.open` gauge stay consistent across the startup window.
|
||||
|
||||
@@ -249,7 +256,7 @@ Mechanically: when the last external subscriber detaches and `DetachGraceSeconds
|
||||
|
||||
#### Worker exit outside a gateway close
|
||||
|
||||
`IWorkerClient` has no state-change callback, so a worker that reached `Closed` or `Faulted` *without the gateway closing it* — the client sent the `ShutdownWorker` control command and the worker exited cleanly, the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session reporting `Ready` until its lease expired (a live finding: `WorkerClient` was `Closed` on the shutdown ack, the session still `Ready`, even with an event stream attached). The session now observes it from its own worker-event pump source (`MapWorkerEventsAsync`): the single distributor pump is the session's one always-on view of the worker (it runs from `MarkReady` via the dashboard mirror), and `WorkerClient` enters its terminal state *before* it completes the event channel, so when the enumeration ends the terminal state is visible. `ObserveWorkerTerminalState` then faults a `Ready` session whose worker is `Closed`/`Faulted`.
|
||||
`IWorkerClient` has no state-change callback, so a worker that reached `Closed` or `Faulted` *without the gateway closing it* — the client sent the `ShutdownWorker` control command and the worker exited cleanly, the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session reporting `Ready` until its lease expired (a live finding: `WorkerClient` was `Closed` on the shutdown ack, the session still `Ready`, even with an event stream attached). The session now observes it from its own worker-event pump source (`MapWorkerEventsAsync`): the single distributor pump is the session's one always-on view of the worker (it runs from `MarkReady` via the dashboard mirror), and `WorkerClient` enters its terminal state *before* it completes the event channel, so when the enumeration ends the terminal state is visible. `ObserveWorkerTerminalState` then faults a `Ready` session whose worker is `Closed`/`Faulted`. The pump also calls it when its first readiness check (`GetReadyWorkerClientAsync`) fails. A worker that dies between the open's attach and the pump's first read never reaches the event enumeration, so its `finally` never runs. Without that second call the session would stay `Ready` over the dead worker until its lease expired.
|
||||
|
||||
The choice is **`Faulted`, not `Closed`**: `Closed` means the gateway closed the session, and a `Closed` session left in the registry would linger until its lease expired with nothing reaping it. A faulted session is permanently unusable (every command fails the readiness check, as the worker is gone), stays observable via `GetSessionStatus` with `FinalFault` naming the worker state, and is reaped by the faulted-session sweep below after `FaultedGraceSeconds` — which is where `SessionClosed` is accounted, exactly once, through the ordinary `CloseSessionCoreAsync` path (`WorkerClient.ShutdownAsync` on a `Closed` client returns immediately, so no kill fallback or warning). Only a `Ready` session follows its worker: during a gateway-initiated close the worker closing is the expected outcome, not a fault.
|
||||
|
||||
@@ -261,7 +268,7 @@ A session that faults (for example, a slow single-subscriber client that overflo
|
||||
|
||||
Detach-grace only helps a session that *had* a subscriber. A client that opens a session and then disappears before it ever calls `StreamEvents` — OtOpcUa's `GalaxyDriver` failing part-way through its start-up is the motivating case — leaves a `Ready` session with no subscriber, no detach timestamp and a lease 30 minutes out, pinning one of `MaxSessions` and a live x86 worker for all of `DefaultLeaseSeconds`. `MxGateway:Sessions:UnattachedSessionGraceSeconds` (default `0`, off) closes that gap: when positive, `CloseExpiredLeasesAsync` also checks `IsUnattachedGraceExpired(now)` and reaps, with the distinct `UnattachedGraceExpiredReason` (`"unattached-grace-expired"`), any `Ready` session that has never had a subscriber of either kind and has been idle for the window.
|
||||
|
||||
Four deliberate limits keep the rule from catching legitimate sessions. It applies only to `Ready` sessions, and its idle clock starts at `Ready` (`MarkReady` stamps `_readyAtUtc`), so a slow worker start neither reaps the session mid-startup nor eats the window before the client even has the session id. Idle is measured from the later of that Ready stamp and the session's *last client activity* (its most recent unary command), so a unary-only client (reads and writes, never streams) is not reaped while it keeps calling; the rule is an idle timeout for sessions that never streamed, not a deadline to stream. It stops applying the moment the first subscriber attaches, after which detach-grace and the normal lease govern the session. And an *internal* in-process subscriber counts as attached for this rule (`AttachInternalEventSubscriber` sets a separate `_everHadInternalEventSubscriber` latch): the gateway's own `GatewayAlarmMonitor` session only ever attaches that way, and it must not be reaped by its own gateway every cycle. The latch is separate from `_everHadEventSubscriber` because that one drives detach-grace, which is keyed on external subscribers only. The dashboard mirror, which every session registers at `MarkReady`, is deliberately *not* a subscriber for this rule. Like the other sweep conditions it is re-verified inside `TryBeginCloseIfExpired`, so a subscriber that attaches between selection and the close wins the race. Operators should set the window comfortably above a well-behaved client's open-to-first-`StreamEvents` time; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later.
|
||||
Four deliberate limits keep the rule from catching legitimate sessions. It applies only to `Ready` sessions, and its idle clock starts at `Ready` (`MarkReady` stamps `_readyAtUtc`), so a slow worker start neither reaps the session mid-startup nor eats the window before the client even has the session id. Idle is measured from the later of that Ready stamp and the session's *last client activity* (its most recent unary command), so a unary-only client (reads and writes, never streams) is not reaped while it keeps calling; the rule is an idle timeout for sessions that never streamed, not a deadline to stream. It stops applying the moment the first subscriber attaches, after which detach-grace and the normal lease govern the session. And an *internal* in-process subscriber counts as attached for this rule (`AttachInternalEventSubscriber` sets a separate `_everHadInternalEventSubscriber` latch): the gateway's own `GatewayAlarmMonitor` session only ever attaches that way, and it must not be reaped by its own gateway every cycle. The latch is separate from `_everHadEventSubscriber` because that one drives detach-grace, which is keyed on external subscribers only. The dashboard mirror, which every session registers at `MarkReady`, is deliberately *not* a subscriber for this rule. The alarm monitor is the only gateway-owned session exempted. The dashboard Browse page's live-data session (`DashboardLiveDataService`) is unary-only: it registers and reads but never attaches a subscriber, so once it has been idle for the window the rule reaps it. That is expected. The service checks its cached session on every call (`EnsureReadyAsync`) and opens and registers a new one when the old one is gone. Like the other sweep conditions it is re-verified inside `TryBeginCloseIfExpired`, so a subscriber that attaches between selection and the close wins the race. Operators should set the window comfortably above a well-behaved client's open-to-first-`StreamEvents` time; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later.
|
||||
|
||||
#### Reconnect and replay
|
||||
|
||||
@@ -376,7 +383,7 @@ Iterating over `Snapshot` rather than the live dictionary lets `RemoveSessionAsy
|
||||
|
||||
### Stopping gate
|
||||
|
||||
A drain that only ever looked at one snapshot could be outrun by `OpenSession`: the host raises `ApplicationStopping` *before* hosted services stop, and Kestrel keeps accepting gRPC calls until its own `StopAsync` runs, so a session opened behind the snapshot would have nobody left to close it and its x86 worker would survive the gateway as an orphan. `SessionManager` therefore keeps a stopping flag that is set by the host's `IHostApplicationLifetime.ApplicationStopping` (the manager registers for it in its constructor when the lifetime is injected) and, as a fallback for hosts without one, at the start of `ShutdownAsync`. Once set it is never cleared. `OpenSessionAsync` checks it twice: on entry, before a slot is taken, and again right after the session is registered. The second check is what closes the race — the flag is written before the drain takes its registry snapshot, so an open that does not observe it there registered its session before that snapshot and is drained, while one that does observe it rolls its own session back. Either way the caller receives `SessionManagerException` with `GatewayStopping`, which the gRPC layer maps to `Unavailable` so the client retries against the restarted gateway; it is not an open failure and records no fault metric.
|
||||
A drain that only ever looked at one snapshot could be outrun by `OpenSession`: the host raises `ApplicationStopping` *before* hosted services stop, and Kestrel keeps accepting gRPC calls until its own `StopAsync` runs, so a session opened behind the snapshot would have nobody left to close it and its x86 worker would survive the gateway as an orphan. `SessionManager` therefore keeps a stopping flag that is set by the host's `IHostApplicationLifetime.ApplicationStopping` (the manager registers for it in its constructor when the lifetime is injected) and, as a fallback for hosts without one, at the start of `ShutdownAsync`. Once set it is never cleared. `OpenSessionAsync` checks it twice: on entry, before a slot is taken, and again right after the session is registered. The second check is what closes the race — the flag is written before the drain takes its registry snapshot, so an open that does not observe it there registered its session before that snapshot and is drained, while one that does observe it rolls its own session back. Either way the caller receives `SessionManagerException` with `GatewayStopping`, which the gRPC layer maps to `Unavailable` so the client retries against the restarted gateway; it is not an open failure and records no fault metric. An open that passed both checks but whose session the drain closed while its worker was still starting gets the same code: the attach is refused, the late worker is discarded, and the manager rechecks the flag before it reports the refusal.
|
||||
|
||||
The drain loop above is the backstop to that gate: it re-snapshots until the registry is empty, bounded at `MaxShutdownDrainPasses` (`3`) so a session that somehow cannot be removed cannot hold the host stop forever. Anything still registered after the last pass is logged and left to the orphan terminator on the next start.
|
||||
|
||||
|
||||
@@ -16205,11 +16205,15 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
|
||||
/// reply returns the cached value WITHOUT modifying the existing
|
||||
/// subscription (was_cached = true).
|
||||
/// - Otherwise the worker takes the snapshot lifecycle itself: AddItem +
|
||||
/// Advise, wait up to `timeout_ms` for the first OnDataChange, then
|
||||
/// UnAdvise + RemoveItem before returning. The session is left exactly
|
||||
/// as it was before the call (was_cached = false).
|
||||
/// Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem
|
||||
/// before returning. The session is left exactly as it was before the
|
||||
/// call (was_cached = false).
|
||||
///
|
||||
/// `timeout_ms == 0` uses the gateway-configured default (1000 ms).
|
||||
/// `timeout_ms` is one deadline shared by the whole command, not a per-tag
|
||||
/// budget: the worker adds and advises every uncached tag, waits once for all
|
||||
/// of them, then tears them all down. A tag that has not delivered by the
|
||||
/// deadline gets its own was_successful = false result.
|
||||
/// `timeout_ms == 0` uses the worker's default (1000 ms).
|
||||
/// </summary>
|
||||
[global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
|
||||
public sealed partial class ReadBulkCommand : pb::IMessage<ReadBulkCommand>
|
||||
|
||||
@@ -524,11 +524,15 @@ message WriteSecured2BulkEntry {
|
||||
// reply returns the cached value WITHOUT modifying the existing
|
||||
// subscription (was_cached = true).
|
||||
// - Otherwise the worker takes the snapshot lifecycle itself: AddItem +
|
||||
// Advise, wait up to `timeout_ms` for the first OnDataChange, then
|
||||
// UnAdvise + RemoveItem before returning. The session is left exactly
|
||||
// as it was before the call (was_cached = false).
|
||||
// Advise, wait for the first OnDataChange, then UnAdvise + RemoveItem
|
||||
// before returning. The session is left exactly as it was before the
|
||||
// call (was_cached = false).
|
||||
//
|
||||
// `timeout_ms == 0` uses the gateway-configured default (1000 ms).
|
||||
// `timeout_ms` is one deadline shared by the whole command, not a per-tag
|
||||
// budget: the worker adds and advises every uncached tag, waits once for all
|
||||
// of them, then tears them all down. A tag that has not delivered by the
|
||||
// deadline gets its own was_successful = false result.
|
||||
// `timeout_ms == 0` uses the worker's default (1000 ms).
|
||||
message ReadBulkCommand {
|
||||
int32 server_handle = 1;
|
||||
repeated string tag_addresses = 2;
|
||||
|
||||
@@ -174,12 +174,13 @@ public sealed class EventStreamService(
|
||||
{
|
||||
// The distributor pump completes every subscriber channel with the source
|
||||
// fault when the worker event stream terminates abnormally; that surfaces
|
||||
// here. Mirror the original ProduceEventsAsync behavior: fault the
|
||||
// session and record the metric, then propagate the terminal fault to the
|
||||
// gRPC client.
|
||||
// here. Fault the session and propagate the terminal fault to the gRPC
|
||||
// client. The fault itself is NOT counted here: WorkerClient.SetFaulted
|
||||
// already recorded it once, and every subscriber of the session sees the
|
||||
// same fault, so counting it per subscriber inflated mxgateway.faults to
|
||||
// 1+K. This subscriber's share is the WorkerFault disconnect reason.
|
||||
disconnectReason = EventStreamDisconnectReasons.WorkerFault;
|
||||
session.MarkFaulted(workerException.Message);
|
||||
metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
|
||||
throw;
|
||||
}
|
||||
catch (SessionManagerException sessionException)
|
||||
|
||||
@@ -475,17 +475,27 @@ public sealed class GatewaySession
|
||||
/// accounting here, and <see cref="OpenRecorded"/> is set right after it returns.
|
||||
/// </param>
|
||||
/// <returns>
|
||||
/// <see langword="true"/> when the session became Ready; <see langword="false"/> when the
|
||||
/// attach was refused (see <see cref="AttachWorkerClient"/>), including when the session was
|
||||
/// disposed before the gate could be taken.
|
||||
/// <see cref="SessionAttachOutcome.Attached"/> when the session became Ready;
|
||||
/// <see cref="SessionAttachOutcome.SessionClosed"/> when the session refused the attach
|
||||
/// (see <see cref="AttachWorkerClient"/>), including when it was disposed before the gate
|
||||
/// could be taken; <see cref="SessionAttachOutcome.WorkerTerminal"/> when the worker client
|
||||
/// had already reached <see cref="WorkerClientState.Faulted"/> or
|
||||
/// <see cref="WorkerClientState.Closed"/>.
|
||||
/// </returns>
|
||||
/// <remarks>
|
||||
/// Not cancellable: the worker client already exists, so the only acceptable outcomes are
|
||||
/// "attached" or "refused, caller discards it". The gate is held only for the duration of
|
||||
/// a close or kill, so the wait is bounded by the worker shutdown timeout. The caller
|
||||
/// still decides what to do about a cancelled open before calling this.
|
||||
/// <para>
|
||||
/// A worker that dies right after its handshake (an STA loop fault, exit code 7) can
|
||||
/// be terminal before it is attached. Attaching it would hand the caller a Ready
|
||||
/// session over a dead worker that nothing notices until the lease expires, so it is
|
||||
/// refused here. A worker that dies just after this check is caught by the event
|
||||
/// pump's readiness failure path instead (see <c>MapWorkerEventsAsync</c>).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal async ValueTask<bool> TryAttachWorkerClientAndMarkReadyAsync(
|
||||
internal async ValueTask<SessionAttachOutcome> TryAttachWorkerClientAndMarkReadyAsync(
|
||||
IWorkerClient workerClient,
|
||||
Action onReady)
|
||||
{
|
||||
@@ -499,14 +509,27 @@ public sealed class GatewaySession
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// Disposed while the worker was starting: the session has been torn down.
|
||||
return false;
|
||||
return SessionAttachOutcome.SessionClosed;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (workerClient.State is WorkerClientState.Faulted or WorkerClientState.Closed)
|
||||
{
|
||||
// A session closed meanwhile is the stronger reason: it decides whether the open
|
||||
// reports a stopping gateway rather than a failed worker.
|
||||
lock (_syncRoot)
|
||||
{
|
||||
return _state is SessionState.Closing or SessionState.Closed or SessionState.Faulted
|
||||
|| _eventDistributorDisposed
|
||||
? SessionAttachOutcome.SessionClosed
|
||||
: SessionAttachOutcome.WorkerTerminal;
|
||||
}
|
||||
}
|
||||
|
||||
if (!AttachWorkerClient(workerClient))
|
||||
{
|
||||
return false;
|
||||
return SessionAttachOutcome.SessionClosed;
|
||||
}
|
||||
|
||||
MarkReady();
|
||||
@@ -516,7 +539,7 @@ public sealed class GatewaySession
|
||||
_openRecorded = true;
|
||||
}
|
||||
|
||||
return true;
|
||||
return SessionAttachOutcome.Attached;
|
||||
}
|
||||
finally
|
||||
{
|
||||
@@ -1001,7 +1024,22 @@ public sealed class GatewaySession
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
MxAccessGrpcMapper mapper = _eventStreaming.Mapper;
|
||||
IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
|
||||
IWorkerClient workerClient;
|
||||
try
|
||||
{
|
||||
workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (SessionManagerException)
|
||||
{
|
||||
// The worker can die between the open's attach and this first read (an STA loop
|
||||
// fault or exit code 7 right after the handshake). The readiness check then fails
|
||||
// here and the worker event stream below is never reached, so its finally never
|
||||
// runs; without this the session would stay Ready over a dead worker until its
|
||||
// lease expired.
|
||||
ObserveWorkerTerminalState();
|
||||
throw;
|
||||
}
|
||||
|
||||
TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow());
|
||||
|
||||
try
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Sessions;
|
||||
|
||||
/// <summary>
|
||||
/// Result of <see cref="GatewaySession.TryAttachWorkerClientAndMarkReadyAsync"/>: whether the
|
||||
/// open's worker client was attached, and if not, which side refused it. The open maps each
|
||||
/// refusal to a different failure, so the reason has to survive the call.
|
||||
/// </summary>
|
||||
internal enum SessionAttachOutcome
|
||||
{
|
||||
/// <summary>The worker client was attached and the session is <see cref="SessionState.Ready"/>.</summary>
|
||||
Attached,
|
||||
|
||||
/// <summary>
|
||||
/// The session was closed, faulted or disposed while its worker was starting (an admin
|
||||
/// kill or the host drain); the caller discards the client.
|
||||
/// </summary>
|
||||
SessionClosed,
|
||||
|
||||
/// <summary>
|
||||
/// The worker client was already <see cref="Workers.WorkerClientState.Faulted"/> or
|
||||
/// <see cref="Workers.WorkerClientState.Closed"/> — the worker died right after its
|
||||
/// handshake. Attaching it would leave a Ready session over a dead worker; the caller
|
||||
/// discards the client and fails the open.
|
||||
/// </summary>
|
||||
WorkerTerminal,
|
||||
}
|
||||
@@ -164,11 +164,11 @@ public sealed class SessionManager : ISessionManager
|
||||
// if a kill or the host drain closed it meanwhile — it must not be returned on a dead
|
||||
// session; and a caller that abandoned the open while the worker was starting gets
|
||||
// no session either.
|
||||
bool ready;
|
||||
SessionAttachOutcome outcome;
|
||||
try
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
ready = await session
|
||||
outcome = await session
|
||||
.TryAttachWorkerClientAndMarkReadyAsync(workerClient, _metrics.SessionOpened)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
@@ -178,14 +178,25 @@ public sealed class SessionManager : ISessionManager
|
||||
throw;
|
||||
}
|
||||
|
||||
if (!ready)
|
||||
if (outcome is SessionAttachOutcome.Attached)
|
||||
{
|
||||
await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false);
|
||||
throw new InvalidOperationException(
|
||||
$"Session {session.SessionId} was closed while its worker was starting.");
|
||||
return session;
|
||||
}
|
||||
|
||||
return session;
|
||||
WorkerClientState refusedWorkerState = workerClient.State;
|
||||
await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false);
|
||||
if (outcome is SessionAttachOutcome.WorkerTerminal)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"The worker for session {session.SessionId} reached {refusedWorkerState} before it could be attached.");
|
||||
}
|
||||
|
||||
// Refused because the session was closed while its worker was starting. When the host
|
||||
// drain did that, the caller should retry against the restarted gateway, exactly like
|
||||
// an open refused at the door, rather than see an open failure.
|
||||
ThrowIfStopping();
|
||||
throw new InvalidOperationException(
|
||||
$"Session {session.SessionId} was closed while its worker was starting.");
|
||||
}
|
||||
catch (SessionManagerException exception) when (exception.ErrorCode == SessionManagerErrorCode.GatewayStopping)
|
||||
{
|
||||
@@ -194,6 +205,15 @@ public sealed class SessionManager : ISessionManager
|
||||
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
catch (OperationCanceledException exception) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// The caller abandoned the open (deadline, disconnect). Not an open failure either:
|
||||
// roll back (a worker that arrived late was already discarded above), record no
|
||||
// fault, log nothing, and let the cancellation surface so the gRPC layer maps it to
|
||||
// Cancelled rather than Unavailable.
|
||||
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
|
||||
|
||||
@@ -358,7 +358,63 @@ public sealed class EventStreamServiceTests
|
||||
|
||||
Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
|
||||
Assert.Equal(SessionState.Faulted, session.State);
|
||||
|
||||
// The worker fault is counted once, by the worker client that faulted (this fake does
|
||||
// not count it); a subscriber observing it records a disconnect reason, not a fault.
|
||||
Assert.Equal(0, metrics.GetSnapshot().Faults);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One worker fault is one <c>mxgateway.faults</c> increment however many
|
||||
/// <c>StreamEvents</c> subscribers observe it: <see cref="WorkerClient"/> records it when it
|
||||
/// faults, and each subscriber records only its <c>WorkerFault</c> disconnect reason.
|
||||
/// Counting it per subscriber as well inflated the metric to 1+K for K subscribers.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task StreamEventsAsync_WhenWorkerFaultsUnderTwoSubscribers_CountsTheFaultOnce()
|
||||
{
|
||||
await using Workers.Fakes.FakeWorkerHarness fakeWorker = await Workers.Fakes.FakeWorkerHarness.CreateConnectedPairAsync();
|
||||
using GatewayMetrics metrics = new();
|
||||
using DisconnectReasonRecorder reasons = new(metrics);
|
||||
WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
|
||||
Task startTask = client.StartAsync(CancellationToken.None);
|
||||
await fakeWorker.CompleteStartupAsync();
|
||||
await startTask.WaitAsync(TestTimeout);
|
||||
GatewayOptions options = new()
|
||||
{
|
||||
Sessions = new SessionOptions { AllowMultipleEventSubscribers = true },
|
||||
};
|
||||
SessionManager manager = new(
|
||||
new SessionRegistry(),
|
||||
new SingleClientFactory(client),
|
||||
Options.Create(options),
|
||||
metrics);
|
||||
GatewaySession session = await manager.OpenSessionAsync(
|
||||
new SessionOpenRequest(RequestedBackend: null, ClientSessionName: "fault-count", ClientCorrelationId: null, CommandTimeout: null),
|
||||
"client-1",
|
||||
ownerKeyId: null,
|
||||
CancellationToken.None);
|
||||
EventStreamService service = new(manager, Options.Create(options), metrics);
|
||||
await using IAsyncEnumerator<MxEvent> first = service
|
||||
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
|
||||
.GetAsyncEnumerator();
|
||||
await using IAsyncEnumerator<MxEvent> second = service
|
||||
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
|
||||
.GetAsyncEnumerator();
|
||||
Task<bool> firstMove = first.MoveNextAsync().AsTask();
|
||||
Task<bool> secondMove = second.MoveNextAsync().AsTask();
|
||||
await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 2);
|
||||
|
||||
await fakeWorker.DisposeWorkerSideAsync();
|
||||
|
||||
await Assert.ThrowsAsync<WorkerClientException>(async () => await firstMove.WaitAsync(TestTimeout));
|
||||
await Assert.ThrowsAsync<WorkerClientException>(async () => await secondMove.WaitAsync(TestTimeout));
|
||||
Assert.Equal(WorkerClientState.Faulted, client.State);
|
||||
Assert.Equal(1, metrics.GetSnapshot().Faults);
|
||||
Assert.Equal(
|
||||
[EventStreamDisconnectReasons.WorkerFault, EventStreamDisconnectReasons.WorkerFault],
|
||||
reasons.Reasons);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -1038,6 +1094,15 @@ public sealed class EventStreamServiceTests
|
||||
/// measurement recorded on one <see cref="GatewayMetrics"/> instance (filtered by meter
|
||||
/// reference, so parallel tests sharing the meter name cannot cross-talk).
|
||||
/// </summary>
|
||||
private sealed class SingleClientFactory(IWorkerClient client) : ISessionWorkerClientFactory
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<IWorkerClient> CreateAsync(GatewaySession session, CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.FromResult(client);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class DisconnectReasonRecorder : IDisposable
|
||||
{
|
||||
private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
|
||||
|
||||
@@ -88,6 +88,78 @@ public sealed class GatewaySessionWorkerExitTests
|
||||
Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A worker that dies right after its handshake (STA loop fault, exit code 7) can reach
|
||||
/// <c>Faulted</c> before the open attaches it. The attach must refuse such a client: the
|
||||
/// open fails with <c>OpenFailed</c>, the client is discarded, and no session is left
|
||||
/// registered reporting <c>Ready</c> over a dead worker until its lease expires.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task OpenSession_WorkerAlreadyFaultedAtAttach_FailsOpenAndDiscardsClient()
|
||||
{
|
||||
await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
|
||||
using GatewayMetrics metrics = new();
|
||||
WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
|
||||
await StartClientAsync(fakeWorker, client);
|
||||
await fakeWorker.DisposeWorkerSideAsync();
|
||||
await WaitUntilAsync(() => client.State == WorkerClientState.Faulted);
|
||||
SessionRegistry registry = new();
|
||||
SessionManager manager = new(
|
||||
registry,
|
||||
new SingleClientFactory(client),
|
||||
Options.Create(new GatewayOptions()),
|
||||
metrics);
|
||||
|
||||
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
|
||||
async () => await OpenSessionAsync(manager));
|
||||
|
||||
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
|
||||
Assert.Contains("Faulted", exception.InnerException?.Message, StringComparison.Ordinal);
|
||||
Assert.Equal(0, registry.Count);
|
||||
GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
|
||||
Assert.Equal(0, snapshot.OpenSessions);
|
||||
Assert.Equal(0, snapshot.SessionsOpened);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The worker is still <c>Ready</c> when the open attaches it but faults before the event
|
||||
/// pump's first read (which the dashboard mirror starts at <c>MarkReady</c>). The pump's
|
||||
/// readiness check then fails without ever reaching the worker event stream, so it is that
|
||||
/// failure path that must notice the dead worker: the session follows it to
|
||||
/// <c>Faulted</c> and the faulted-session sweep reaps it, accounting the close once.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task WorkerFaultsBetweenAttachAndFirstPumpRead_SessionFaultsAndSweepReapsItOnce()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
ReadyOnceWorkerClient client = new();
|
||||
SessionManager manager = new(
|
||||
new SessionRegistry(),
|
||||
new SingleClientFactory(client),
|
||||
Options.Create(new GatewayOptions()),
|
||||
metrics,
|
||||
dashboardEventBroadcaster: TestSupport.NullDashboardEventBroadcaster.Instance);
|
||||
|
||||
GatewaySession session = await OpenSessionAsync(manager);
|
||||
|
||||
await WaitUntilAsync(() => session.State == SessionState.Faulted);
|
||||
Assert.Contains("worker", session.FinalFault, StringComparison.OrdinalIgnoreCase);
|
||||
Assert.Equal(1, metrics.GetSnapshot().OpenSessions);
|
||||
|
||||
int closedCount = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
|
||||
int secondPass = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
|
||||
|
||||
Assert.Equal(1, closedCount);
|
||||
Assert.Equal(0, secondPass);
|
||||
Assert.Equal(SessionState.Closed, session.State);
|
||||
Assert.False(manager.TryGetSession(session.SessionId, out _));
|
||||
GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
|
||||
Assert.Equal(1, snapshot.SessionsClosed);
|
||||
Assert.Equal(0, snapshot.OpenSessions);
|
||||
}
|
||||
|
||||
private static async Task StartClientAsync(FakeWorkerHarness fakeWorker, WorkerClient client)
|
||||
{
|
||||
Task startTask = client.StartAsync(CancellationToken.None);
|
||||
@@ -148,4 +220,56 @@ public sealed class GatewaySessionWorkerExitTests
|
||||
return Task.FromResult(client);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Worker client that reports <c>Ready</c> for its first state read (the open's attach
|
||||
/// check) and <c>Faulted</c> from then on, reproducing a worker that dies in the window
|
||||
/// between the attach and the event pump's first readiness check.
|
||||
/// </summary>
|
||||
private sealed class ReadyOnceWorkerClient : IWorkerClient
|
||||
{
|
||||
private int _stateReads;
|
||||
|
||||
/// <inheritdoc />
|
||||
public string SessionId => "session-ready-once";
|
||||
|
||||
/// <inheritdoc />
|
||||
public int? ProcessId => 4321;
|
||||
|
||||
/// <inheritdoc />
|
||||
public WorkerClientState State
|
||||
=> Interlocked.Increment(ref _stateReads) == 1 ? WorkerClientState.Ready : WorkerClientState.Faulted;
|
||||
|
||||
/// <inheritdoc />
|
||||
public DateTimeOffset LastHeartbeatAt => DateTimeOffset.UtcNow;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<WorkerCommandReply> InvokeAsync(WorkerCommand command, TimeSpan timeout, CancellationToken cancellationToken)
|
||||
=> throw new InvalidOperationException("The worker is faulted.");
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await Task.CompletedTask;
|
||||
throw new InvalidOperationException("The pump must not reach the event stream of a faulted worker.");
|
||||
#pragma warning disable CS0162 // Unreachable code: the yield makes this an async iterator.
|
||||
yield break;
|
||||
#pragma warning restore CS0162
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Kill(string reason)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -533,13 +533,17 @@ public sealed class SessionManagerTests
|
||||
// which flaked under CI load) so a regression that waited out the timeout is caught
|
||||
// while ordinary scheduling jitter never trips it.
|
||||
const int readyWaitTimeoutMs = 5000;
|
||||
FakeWorkerClient workerClient = new() { State = WorkerClientState.Faulted };
|
||||
FakeWorkerClient workerClient = new();
|
||||
SessionManager manager = CreateManager(
|
||||
new FakeSessionWorkerClientFactory(workerClient),
|
||||
options: CreateOptions(workerReadyWaitTimeoutMs: readyWaitTimeoutMs));
|
||||
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
|
||||
Assert.Equal(SessionState.Ready, session.State);
|
||||
|
||||
// The worker faults after the open attached it (an open refuses an already-faulted
|
||||
// worker). No event pump runs here, so nothing has moved the session off Ready yet.
|
||||
workerClient.State = WorkerClientState.Faulted;
|
||||
|
||||
Stopwatch stopwatch = Stopwatch.StartNew();
|
||||
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
|
||||
async () => await manager.InvokeAsync(
|
||||
@@ -1145,7 +1149,9 @@ public sealed class SessionManagerTests
|
||||
Assert.Equal(0, snapshot.SessionsClosed);
|
||||
|
||||
// The slot the killed open held must have been released exactly once: a replacement open
|
||||
// succeeds against MaxSessions = 1 (the released gate hands the same fake straight back).
|
||||
// succeeds against MaxSessions = 1 (the released gate hands the same fake straight back,
|
||||
// so undo the discard's kill first — an open refuses an already-faulted worker).
|
||||
workerClient.State = WorkerClientState.Ready;
|
||||
GatewaySession reopened = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
|
||||
Assert.Equal(SessionState.Ready, reopened.State);
|
||||
Assert.Equal(1, registry.Count);
|
||||
@@ -1155,11 +1161,13 @@ public sealed class SessionManagerTests
|
||||
/// A caller that abandons its open (deadline, disconnect) while the worker is starting
|
||||
/// must not leave the worker running: whatever fails between the worker arriving and the
|
||||
/// session becoming Ready — here the caller's own cancellation — the late worker client is
|
||||
/// killed and disposed before the failure surfaces.
|
||||
/// killed and disposed before the failure surfaces. The cancellation is not an open
|
||||
/// failure: it surfaces as the <see cref="OperationCanceledException"/> itself (which the
|
||||
/// gRPC layer maps to <c>Cancelled</c>) and records no fault.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task OpenSessionAsync_WhenCallerCancelledDuringWorkerStartup_DiscardsWorkerClient()
|
||||
public async Task OpenSessionAsync_WhenCallerCancelledDuringWorkerStartup_DiscardsWorkerClientAndRethrowsCancellation()
|
||||
{
|
||||
FakeWorkerClient workerClient = new();
|
||||
GatedSessionWorkerClientFactory factory = new(workerClient);
|
||||
@@ -1173,10 +1181,9 @@ public sealed class SessionManagerTests
|
||||
|
||||
await cancellation.CancelAsync();
|
||||
factory.ReleaseCreate();
|
||||
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () => await openTask);
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await openTask);
|
||||
|
||||
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
|
||||
Assert.IsAssignableFrom<OperationCanceledException>(exception.InnerException);
|
||||
Assert.Equal(0, metrics.GetSnapshot().Faults);
|
||||
Assert.Equal(1, workerClient.KillCount);
|
||||
Assert.Equal(1, workerClient.DisposeCount);
|
||||
Assert.NotEqual(SessionState.Ready, starting.State);
|
||||
@@ -1188,11 +1195,13 @@ public sealed class SessionManagerTests
|
||||
/// <summary>
|
||||
/// Same window as the kill case, driven by the host drain: a session still starting its
|
||||
/// worker when <c>ShutdownAsync</c> runs is closed by the drain, and the worker client that
|
||||
/// arrives afterwards is discarded rather than left running as an orphan.
|
||||
/// arrives afterwards is discarded rather than left running as an orphan. The refused open
|
||||
/// reports <c>GatewayStopping</c> (gRPC <c>Unavailable</c>, retry against the restarted
|
||||
/// gateway), like an open refused at the door, and is not recorded as a fault.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task OpenSessionAsync_WhenShutdownDuringWorkerStartup_FailsOpenAndDiscardsLateWorkerClient()
|
||||
public async Task OpenSessionAsync_WhenShutdownDuringWorkerStartup_ThrowsGatewayStoppingAndDiscardsLateWorkerClient()
|
||||
{
|
||||
FakeWorkerClient workerClient = new();
|
||||
GatedSessionWorkerClientFactory factory = new(workerClient);
|
||||
@@ -1207,7 +1216,8 @@ public sealed class SessionManagerTests
|
||||
factory.ReleaseCreate();
|
||||
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () => await openTask);
|
||||
|
||||
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
|
||||
Assert.Equal(SessionManagerErrorCode.GatewayStopping, exception.ErrorCode);
|
||||
Assert.Equal(0, metrics.GetSnapshot().Faults);
|
||||
Assert.Equal(SessionState.Closed, starting.State);
|
||||
Assert.Equal(1, workerClient.KillCount);
|
||||
Assert.Equal(1, workerClient.DisposeCount);
|
||||
|
||||
Reference in New Issue
Block a user