Joseph Doherty
e541339c07
Resolve audit findings: correct WorkerEnvelope proto/route/metric/session facts; rewrite auth (ZB.MOM.WW.Auth migration), dashboard (ZB.MOM.WW.Theme), and StyleGuide (foreign-project copy-paste); document alarm subsystem, Ldap options, and gateway alarm broker; fix client CLI flags and package paths.
320 lines
22 KiB
Markdown
320 lines
22 KiB
Markdown
# Gateway Sessions
|
|
|
|
The sessions subsystem owns the in-memory representation of an active gateway-to-worker pairing and coordinates its lifecycle from open through close. Each `GatewaySession` corresponds to exactly one MXAccess worker process connected over a dedicated named pipe.
|
|
|
|
## Overview
|
|
|
|
A session is the gateway-side handle that callers use to invoke worker commands, stream worker events, and tear the worker down. The subsystem is split between the per-session state machine (`GatewaySession`), an in-memory directory (`SessionRegistry`), the orchestrator that opens and closes sessions (`SessionManager`), the worker construction step (`SessionWorkerClientFactory`), a hosted service that sweeps expired leases (`SessionLeaseMonitorHostedService`), and a hosted service that drains sessions during host shutdown (`SessionShutdownHostedService`).
|
|
|
|
The three interfaces (`ISessionManager`, `ISessionRegistry`, `ISessionWorkerClientFactory`) are wired as singletons, and both hosted services (`SessionLeaseMonitorHostedService`, `SessionShutdownHostedService`) are registered, by `SessionServiceCollectionExtensions.AddGatewaySessions`. The startup orphan-worker cleanup that runs before any session opens lives in the worker subsystem (`OrphanWorkerCleanupHostedService`); see [Gateway Restart and Orphan Cleanup](#gateway-restart-and-orphan-cleanup).
|
|
|
|
## Key Types
|
|
|
|
### GatewaySession
|
|
|
|
`GatewaySession` is a sealed class that holds the identity, configured timeouts, worker client reference, and current `SessionState` for one session. State is protected by a private `_syncRoot` lock so that property reads and transitions are observed atomically by concurrent gRPC calls and the lease sweeper.
|
|
|
|
The session id is an opaque string in the form `session-{guid:N}` and the per-session pipe name is `mxaccess-gateway-{ProcessId}-{SessionId}`. Encoding the gateway PID into the pipe name avoids collisions when an old gateway process leaks pipes that the OS has not yet reclaimed.
|
|
|
|
`SessionState` itself is the protobuf-generated enum from `ZB.MOM.WW.MxGateway.Contracts.Proto`, so it is shared between the gateway and clients on the wire.
|
|
|
|
`GatewaySession` also keeps an `_items` dictionary keyed by `(ServerHandle, ItemHandle)` mapping each subscribed item to its `SessionItemRegistration` (server handle, item handle, tag address). It is the gateway-side shadow of the items the worker has added, populated as `AddItem`-style commands succeed and pruned on `RemoveItem`. The shadow exists so the gateway can answer item lookups and clean up subscriptions without round-tripping the worker; the worker remains authoritative for the handles themselves (see [gateway.md](../gateway.md)).
|
|
|
|
```csharp
|
|
public void TransitionTo(SessionState nextState)
|
|
{
|
|
lock (_syncRoot)
|
|
{
|
|
if (_state is SessionState.Closed)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (_state is SessionState.Faulted && nextState is not SessionState.Closed)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (_state is SessionState.Closing
|
|
&& nextState is not SessionState.Closed
|
|
&& nextState is not SessionState.Faulted)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_state = nextState;
|
|
}
|
|
}
|
|
```
|
|
|
|
`Closed` is terminal, `Faulted` only allows a transition to `Closed`, and `Closing` only allows a transition to `Closed` or `Faulted`. This guards against late callbacks (worker exit, heartbeat timeout) re-animating a session that is already tearing down or torn down — once `CloseAsync` has set `Closing` under `_syncRoot`, no `TransitionTo(Ready)` from another thread can walk the session back to `Ready`. Both close-related writes (`Closing` and `Closed`) go through `_syncRoot` exactly like every other state write; `_closeLock` only serializes concurrent close attempts.
|
|
|
|
### SessionManager (ISessionManager)
|
|
|
|
`SessionManager` is the orchestrator. It exposes `OpenSessionAsync`, `TryGetSession`, `InvokeAsync`, `ReadEventsAsync`, `CloseSessionAsync`, `KillWorkerAsync`, `CloseExpiredLeasesAsync`, and `ShutdownAsync`. It composes `ISessionRegistry`, `ISessionWorkerClientFactory`, `GatewayMetrics`, and `GatewayOptions`.
|
|
|
|
`CloseSessionAsync` and `KillWorkerAsync` are both end-of-life paths but differ in what they offer the worker:
|
|
|
|
- `CloseSessionAsync` is the graceful path: it calls `GatewaySession.CloseAsync`, which asks the worker to shut down via `IWorkerClient.ShutdownAsync` and only kills the process as a fallback if shutdown fails.
|
|
- `KillWorkerAsync` is the forceful path used by the dashboard's admin Kill button: it calls `GatewaySession.KillWorkerWithCloseGateAsync`, which kills the worker process immediately with no graceful-shutdown attempt and transitions the session to `Closed`. Routing through `KillWorkerWithCloseGateAsync` (rather than the bare `GatewaySession.KillWorker`) acquires the per-session `_closeLock` so a kill and an in-flight graceful close serialize on the same "was the session already closed" observation that drives metric accounting; the method returns that observation so `KillWorkerAsync` increments `mxgateway.sessions.closed` at most once across concurrent callers.
|
|
|
|
Both paths converge on the same registry/metrics cleanup, so the open-session slot is released and `mxgateway.sessions.closed` is incremented either way.
|
|
|
|
Concurrency is bounded by a `SemaphoreSlim` initialized to `GatewayOptions.Sessions.MaxSessions`. Open requests that exceed the bound throw `SessionManagerException` with `SessionLimitExceeded` rather than queuing; the caller is expected to retry.
|
|
|
|
```csharp
|
|
private void EnsureSessionCapacity()
|
|
{
|
|
if (!_sessionSlots.Wait(0))
|
|
{
|
|
throw new SessionManagerException(
|
|
SessionManagerErrorCode.SessionLimitExceeded,
|
|
$"Gateway session limit {_options.Sessions.MaxSessions} has been reached.");
|
|
}
|
|
}
|
|
```
|
|
|
|
`SessionManager` also defines three close-reason constants — `DefaultCloseReason` (`"client-close"`), `GatewayShutdownReason` (`"gateway-shutdown"`), and `LeaseExpiredReason` (`"lease-expired"`) — so that the metrics and worker shutdown paths agree on a fixed vocabulary.
|
|
|
|
### SessionRegistry (ISessionRegistry)
|
|
|
|
`SessionRegistry` is a thin wrapper over a `ConcurrentDictionary<string, GatewaySession>` keyed by session id with `StringComparer.Ordinal`. `Snapshot` materializes the values into an array so iteration callers (lease sweeper, shutdown) do not race with concurrent `TryAdd` and `TryRemove` calls.
|
|
|
|
`ActiveCount` filters out sessions whose state is `Closed`; this is consumed by metrics and the dashboard, where `Count` would otherwise momentarily over-report during teardown.
|
|
|
|
### SessionWorkerClientFactory (ISessionWorkerClientFactory)
|
|
|
|
`SessionWorkerClientFactory.CreateAsync` is the only path that builds an `IWorkerClient`. It drives the session through the protobuf `SessionState` substates in order: `StartingWorker`, `WaitingForPipe`, `Handshaking`, `InitializingWorker`. The substates are wire-visible so the dashboard and clients can render startup progress.
|
|
|
|
A linked `CancellationTokenSource` enforces `session.StartupTimeout`. If startup fails or times out, the factory either kills the partially-constructed `WorkerClient` or, if the client was never built, kills the launched process and disposes the named pipe before rethrowing. A pure timeout is rewritten as `TimeoutException` so callers can distinguish it from caller-driven cancellation:
|
|
|
|
```csharp
|
|
if (exception is OperationCanceledException
|
|
&& startupCancellation.IsCancellationRequested
|
|
&& !cancellationToken.IsCancellationRequested)
|
|
{
|
|
throw new TimeoutException(
|
|
$"Worker session {session.SessionId} did not complete startup within {session.StartupTimeout}.",
|
|
exception);
|
|
}
|
|
```
|
|
|
|
The named pipe is created with `maxNumberOfServerInstances: 1` so a second worker cannot connect to the same pipe name even if the first launch is still pending. Combined with the per-session nonce passed to the worker, this is the gateway's defense against a foreign process answering a pipe.
|
|
|
|
The factory also seeds the worker client's `MaxPendingCommands` from `MxGateway:Sessions:MaxPendingCommandsPerSession` (default 128, validated `> 0` at startup). This caps how many commands can be in flight to a single worker at once; the `WorkerClient` rejects an enqueue past the cap and records `mxgateway.queues.overflows` tagged `worker-pending-commands`. The bound exists because the worker executes commands serially on one STA — an unbounded backlog would only grow memory and latency, not throughput.
|
|
|
|
### SessionShutdownHostedService
|
|
|
|
`SessionShutdownHostedService` is an `IHostedService` whose only job is to call `ISessionManager.ShutdownAsync` from `StopAsync`. It catches `OperationCanceledException` triggered by the host shutdown timeout and logs a warning so that an over-running shutdown does not surface as an unhandled exception.
|
|
|
|
### SessionOpenRequest
|
|
|
|
`SessionOpenRequest` is the gateway-internal record passed to `OpenSessionAsync`. It is constructed from the wire-level `OpenSessionRequest` via `SessionOpenRequest.FromContract`. Keeping a separate internal record means the gRPC layer can normalize input (defaulting backend, sanitizing strings) without leaking generated proto types into `SessionManager`.
|
|
|
|
```csharp
|
|
public sealed record SessionOpenRequest(
|
|
string? RequestedBackend,
|
|
string? ClientSessionName,
|
|
string? ClientCorrelationId,
|
|
Duration? CommandTimeout)
|
|
{
|
|
public static SessionOpenRequest FromContract(OpenSessionRequest request)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(request);
|
|
|
|
return new SessionOpenRequest(
|
|
request.RequestedBackend,
|
|
request.ClientSessionName,
|
|
request.ClientCorrelationId,
|
|
request.CommandTimeout);
|
|
}
|
|
}
|
|
```
|
|
|
|
### SessionCloseResult
|
|
|
|
`SessionCloseResult` is the record returned from a successful close. `AlreadyClosed` distinguishes "this call closed the session" from "the session was already closed when we acquired the close lock", which the metrics layer uses to avoid double-counting.
|
|
|
|
```csharp
|
|
public sealed record SessionCloseResult(
|
|
string SessionId,
|
|
SessionState FinalState,
|
|
bool AlreadyClosed);
|
|
```
|
|
|
|
### SessionCloseStartedException
|
|
|
|
`SessionCloseStartedException` is `internal` and is only thrown from inside `GatewaySession.CloseAsync` when the close path has already begun mutating worker state and a subsequent step fails. `SessionManager.CloseSessionCoreAsync` catches it, marks the session faulted, increments the close-failed metric, removes the session from the registry, and rethrows it wrapped as `SessionManagerException` with `CloseFailed`. The intermediate type exists so the public API surface only ever exposes `SessionManagerException`.
|
|
|
|
### SessionManagerException and SessionManagerErrorCode
|
|
|
|
`SessionManagerException` is the single public error type emitted from this subsystem; the code is carried in the `ErrorCode` property and is also surfaced to metrics tags via `SessionManagerErrorCode.ToString()`.
|
|
|
|
| Code | Meaning |
|
|
|------|---------|
|
|
| `SessionNotFound` | The session id is not in the registry. |
|
|
| `SessionNotReady` | The session or its `IWorkerClient` is not in `Ready` state. |
|
|
| `EventSubscriberAlreadyActive` | A second event subscriber attached when only one is allowed. |
|
|
| `EventQueueOverflow` | Reserved for the worker event channel overflow path. |
|
|
| `SessionLimitExceeded` | `MaxSessions` is in use. |
|
|
| `OpenFailed` | `OpenSessionAsync` failed; the inner exception carries the cause. |
|
|
| `CloseFailed` | A close started but did not complete cleanly; the session is removed and faulted. |
|
|
|
|
## Lifecycle
|
|
|
|
### 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:
|
|
|
|
```csharp
|
|
catch (Exception exception)
|
|
{
|
|
session?.MarkFaulted(exception.Message);
|
|
if (session is not null)
|
|
{
|
|
_registry.TryRemove(session.SessionId, out _);
|
|
await session.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
// If SessionOpened() already incremented the open-session gauge,
|
|
// a failure after that point (e.g. auto-subscribe rejection) must
|
|
// decrement it again so mxgateway.sessions.open does not leak.
|
|
if (sessionOpenedRecorded)
|
|
{
|
|
_metrics.SessionRemoved();
|
|
}
|
|
|
|
ReleaseSessionSlot();
|
|
_metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString());
|
|
_logger.LogWarning(
|
|
exception,
|
|
"Failed to open gateway session {SessionId}.",
|
|
session?.SessionId ?? "<not-created>");
|
|
|
|
throw new SessionManagerException(
|
|
SessionManagerErrorCode.OpenFailed,
|
|
session is null ? "Failed to create session." : $"Failed to open session {session.SessionId}.",
|
|
exception);
|
|
}
|
|
```
|
|
|
|
The order — fault, deregister, dispose, conditionally decrement the open-session gauge, release slot, record fault metric, log, rethrow — matters because releasing the semaphore before disposal would let the next open race the worker process tear-down on the same machine. The `SessionRemoved()` call is conditional on `sessionOpenedRecorded` (Server-006): a failure *after* `SessionOpened()` already incremented `mxgateway.sessions.open` (for example, an auto-subscribe rejection) must decrement the gauge so it does not leak, but a failure before that point must not.
|
|
|
|
### Run
|
|
|
|
While `Ready`, callers reach the worker through `SessionManager.InvokeAsync` or `ReadEventsAsync`. Both delegate to `GatewaySession`, which checks the state under lock and updates `LastClientActivityAt` on every invocation. `GatewaySession` also exposes typed bulk helpers (`AddItemBulkAsync`, `SubscribeBulkAsync`, etc.) that wrap `WorkerCommand` round-trips and translate non-`Ok` `ProtocolStatus` replies into `SessionManagerException` with `SessionNotReady`.
|
|
|
|
Event streaming uses `AttachEventSubscriber` which returns a disposable lease. When `allowMultipleSubscribers` is false the second attach throws `EventSubscriberAlreadyActive`; this prevents two gRPC streams from racing on the same worker event channel. Active event subscribers keep the session lease from expiring until the stream is disposed.
|
|
|
|
The single-subscriber rule is enforced at startup, not just at runtime: setting `MxGateway:Sessions:AllowMultipleEventSubscribers` to `true` is refused by `GatewayOptionsValidator` with "AllowMultipleEventSubscribers is not supported until event fan-out is implemented," so the gateway fails fast rather than booting in a configuration the event path cannot honor. Multi-subscriber fan-out is explicitly out of scope for v1 (see [Design Decisions](./DesignDecisions.md)).
|
|
|
|
Sessions open with `MxGateway:Sessions:DefaultLeaseSeconds` (default 1800) added to the open timestamp. Unary client activity refreshes the lease by the same duration. `ExtendLease` and `IsLeaseExpired` cooperate with `SessionManager.CloseExpiredLeasesAsync`, which iterates a registry snapshot and closes any session whose lease has expired with `LeaseExpiredReason`. `SessionLeaseMonitorHostedService` runs that sweep every `MxGateway:Sessions:LeaseSweepIntervalSeconds` seconds (default 30).
|
|
|
|
### Close
|
|
|
|
`GatewaySession.CloseAsync` is serialized by a per-session `SemaphoreSlim` (`_closeLock`) so only one close runs at a time, but every read/write of `_state` still passes through `_syncRoot` (via `TryBeginClose` and `MarkClosed`). The close path therefore obeys the same lock discipline as `TransitionTo` / `MarkFaulted`: it transitions to `Closing`, asks the worker client to shut down within `ShutdownTimeout`, and on success transitions to `Closed`. `DisposeAsync` waits on `_closeLock` once before disposing the semaphore so an in-flight close's `Release()` cannot race against the dispose. If `WorkerClient.ShutdownAsync` throws, the session falls back to `IWorkerClient.Kill` (forced close):
|
|
|
|
```csharp
|
|
if (_workerClient is not null)
|
|
{
|
|
try
|
|
{
|
|
await _workerClient.ShutdownAsync(ShutdownTimeout, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
try
|
|
{
|
|
_workerClient.Kill(reason);
|
|
}
|
|
catch (Exception killException)
|
|
{
|
|
throw new SessionCloseStartedException(
|
|
$"Session {SessionId} close failed after worker shutdown started.",
|
|
new AggregateException(exception, killException));
|
|
}
|
|
|
|
throw;
|
|
}
|
|
}
|
|
```
|
|
|
|
If both graceful shutdown and the kill fall-back fail, the original and kill exceptions are bundled into an `AggregateException` and surfaced as `SessionCloseStartedException`. `SessionManager.CloseSessionCoreAsync` then translates that into a `SessionManagerException` with `CloseFailed` and removes the session.
|
|
|
|
`GatewaySession.KillWorker` is the unconditional forced-close path. `SessionManager.KillWorkerAsync` — the explicit kill path that the dashboard's admin Kill button invokes — no longer calls it directly; it routes through `GatewaySession.KillWorkerWithCloseGateAsync` so the kill takes the per-session `_closeLock`. That method skips `WorkerClient.ShutdownAsync` entirely and forces the worker process down via `IWorkerClient.Kill`, which records the `mxgateway.workers.killed` counter through `GatewayMetrics.WorkerKilled(reason)`. The session is then removed from the registry and the open-session slot is released, identical to the cleanup that follows a successful `CloseSessionAsync` (which increments `mxgateway.sessions.closed`). There is no separate `KillCount` / `ShutdownCount`: worker terminations are counted by `mxgateway.workers.killed` (tagged with the kill reason), and session closes by `mxgateway.sessions.closed`.
|
|
|
|
## Shutdown Coordination
|
|
|
|
`SessionShutdownHostedService.StopAsync` calls `SessionManager.ShutdownAsync`, which closes every registered session with `GatewayShutdownReason`. The shutdown loop catches per-session exceptions and falls back to a forced kill so that one stuck worker cannot block the rest of the host. The fallback routes through `KillWorkerAsync` (not a bare `session.KillWorker`) so the kill takes the same close-gate and metric bookkeeping as the dashboard kill path (Server-046):
|
|
|
|
```csharp
|
|
public async Task ShutdownAsync(CancellationToken cancellationToken)
|
|
{
|
|
foreach (GatewaySession session in _registry.Snapshot())
|
|
{
|
|
try
|
|
{
|
|
await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
_logger.LogWarning(
|
|
exception,
|
|
"Graceful shutdown failed for session {SessionId}; killing worker.",
|
|
session.SessionId);
|
|
|
|
// CloseSessionCoreAsync's inner SessionCloseStartedException catch normally
|
|
// removes and accounts the session; this fallback only fires for sessions
|
|
// still in the registry, and reuses KillWorkerAsync for identical bookkeeping.
|
|
if (_registry.TryGet(session.SessionId, out _))
|
|
{
|
|
try
|
|
{
|
|
await KillWorkerAsync(session.SessionId, GatewayShutdownReason, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (SessionManagerException killException)
|
|
{
|
|
_logger.LogWarning(
|
|
killException,
|
|
"Worker kill fallback failed for session {SessionId}.",
|
|
session.SessionId);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
Iterating over `Snapshot` rather than the live dictionary lets the registry mutate inside the loop without throwing.
|
|
|
|
## Gateway Restart and Orphan Cleanup
|
|
|
|
A graceful shutdown drains sessions through `ShutdownAsync`, but a gateway crash or `Kill` leaves no chance to tear workers down. Those orphaned worker processes outlive the gateway that launched them, still holding their MXAccess COM instance and their named pipe. Because the pipe name encodes the *old* gateway PID, a fresh gateway will never reconnect to them — v1 deliberately does not reattach orphan workers (see [Design Decisions](./DesignDecisions.md)).
|
|
|
|
Instead, `OrphanWorkerCleanupHostedService` runs once on startup, before any session opens, and calls `OrphanWorkerTerminator.TerminateOrphans`. The terminator enumerates running processes matching the configured worker executable name, skips the current process, and kills any that it identifies as a leftover worker (matched against the configured executable path). Each kill records `mxgateway.workers.killed` tagged `OrphanStartupCleanup` and logs a warning. The sweep is best-effort: a failure to kill any one orphan (it may have already exited, or be inaccessible) is logged and swallowed so it cannot block gateway startup. This service lives in the worker subsystem, not the session subsystem, because it operates on OS processes rather than `GatewaySession` state.
|
|
|
|
## Dependency Injection
|
|
|
|
`SessionServiceCollectionExtensions.AddGatewaySessions` registers the three singletons and the two hosted services:
|
|
|
|
```csharp
|
|
public static IServiceCollection AddGatewaySessions(this IServiceCollection services)
|
|
{
|
|
services.AddSingleton<ISessionRegistry, SessionRegistry>();
|
|
services.AddSingleton<ISessionWorkerClientFactory, SessionWorkerClientFactory>();
|
|
services.AddSingleton<ISessionManager, SessionManager>();
|
|
services.AddHostedService<SessionLeaseMonitorHostedService>();
|
|
services.AddHostedService<SessionShutdownHostedService>();
|
|
|
|
return services;
|
|
}
|
|
```
|
|
|
|
The registry must be a singleton because its `ConcurrentDictionary` is the source of truth for session state across the gRPC service, the lease sweeper, the dashboard, and the shutdown hosted service. `SessionLeaseMonitorHostedService` runs the periodic expired-lease sweep; `SessionShutdownHostedService` drains sessions during host stop. Both are registered after `ISessionManager` so they resolve the same singleton manager when the host starts; `SessionShutdownHostedService` is registered last so it is the latter of the two to be constructed and is available to drain sessions on stop.
|
|
|
|
## Related Documentation
|
|
|
|
- [Gateway Process Design](./GatewayProcessDesign.md)
|
|
- [Gateway Configuration](./GatewayConfiguration.md)
|
|
- [Worker Process Launcher](./WorkerProcessLauncher.md)
|