dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

docs: dashboard live-data session reaping, subscriber-refusal code, distributor-based StreamEvents model

Browse Source 
- Sessions.md / GatewayConfiguration.md: the dashboard Browse live-data session is
  unary-only, is reaped by UnattachedSessionGraceSeconds when idle, and reopens via
  EnsureReadyAsync; only the alarm monitor is exempt.
- GatewayConfiguration.md: a refused second subscriber gets ResourceExhausted, not AlreadyExists.
- Grpc.md: rewrite the event streaming model around the SessionEventDistributor pump and
  replay ring; add EventSubscriberLimitReached to the mapping snippet; describe what
  clients see while the gateway is stopping.
This commit is contained in:
Joseph Doherty
2026-10-22 08:43:41 -04:00
parent a0b5e91227
commit 3ffba83723
3 changed files with 15 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/Grpc.md
+12 -45
View File
@@ -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