Compare commits

..

2 Commits

Author SHA1 Message Date
Joseph Doherty 7b86bab705 Merge remote-tracking branch 'origin/main' into agent-1/issue-16-implement-blazor-server-dashboard 2026-04-26 18:38:19 -04:00
Joseph Doherty 56886c3b4e Implement Blazor Server dashboard 2026-04-26 18:37:16 -04:00
40 changed files with 901 additions and 1124 deletions
-3
View File
@@ -18,7 +18,6 @@ starting `MxGateway.Worker.exe` or loading MXAccess COM. The harness scripts:
- `WorkerHello` and `WorkerReady` startup, - `WorkerHello` and `WorkerReady` startup,
- command replies with matching correlation ids, - command replies with matching correlation ids,
- ordered `WorkerEvent` frames, - ordered `WorkerEvent` frames,
- `WorkerHeartbeat` frames,
- `WorkerFault` frames, - `WorkerFault` frames,
- shutdown acknowledgements, - shutdown acknowledgements,
- malformed protobuf payloads and oversized frame headers, - malformed protobuf payloads and oversized frame headers,
@@ -44,8 +43,6 @@ event streaming behavior:
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~FakeWorkerHarnessTests dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~FakeWorkerHarnessTests
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~SessionWorkerClientFactoryFakeWorkerTests dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~SessionWorkerClientFactoryFakeWorkerTests
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~GatewayEndToEndFakeWorkerSmokeTests dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~GatewayEndToEndFakeWorkerSmokeTests
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~WorkerClientTests
dotnet test src/MxGateway.Worker.Tests/MxGateway.Worker.Tests.csproj -p:Platform=x86 --filter FullyQualifiedName~WorkerPipeSessionTests
``` ```
Run the gateway test project after shared gateway test infrastructure changes: Run the gateway test project after shared gateway test infrastructure changes:
+33 -30
View File
@@ -34,9 +34,13 @@ SignalR circuit. Bootstrap is sufficient for a basic dashboard.
## Hosting Model ## Hosting Model
The dashboard is hosted by `MxGateway.Server` alongside the gRPC API. The dashboard is hosted by `MxGateway.Server` alongside the gRPC API. When
`MxGateway:Dashboard:Enabled` is `true`, `MapGatewayDashboard()` maps the
configured `Dashboard:PathBase` to the Blazor Server app and maps the login,
logout, and access-denied HTTP endpoints beside it. When dashboard hosting is
disabled, those routes are not mapped.
Suggested endpoint layout: Endpoint layout:
```text ```text
/dashboard /dashboard
@@ -45,7 +49,7 @@ Suggested endpoint layout:
/dashboard/workers /dashboard/workers
/dashboard/events /dashboard/events
/dashboard/settings /dashboard/settings
/_blazor /dashboard/_blazor
``` ```
The app should redirect `/` to `/dashboard` only if the deployment wants the The app should redirect `/` to `/dashboard` only if the deployment wants the
@@ -59,9 +63,10 @@ MxGateway.Server
Components/ Components/
App.razor App.razor
Routes.razor Routes.razor
DashboardPageBase.cs
DashboardDisplay.cs
Layout/ Layout/
DashboardLayout.razor DashboardLayout.razor
NavMenu.razor
Pages/ Pages/
DashboardHome.razor DashboardHome.razor
SessionsPage.razor SessionsPage.razor
@@ -69,26 +74,21 @@ MxGateway.Server
WorkersPage.razor WorkersPage.razor
EventsPage.razor EventsPage.razor
SettingsPage.razor SettingsPage.razor
Components/ Shared/
MetricCard.razor MetricCard.razor
SessionTable.razor StatusBadge.razor
WorkerTable.razor
EventRatePanel.razor
FaultList.razor FaultList.razor
Services/ DashboardSnapshotService.cs
DashboardSnapshotService.cs DashboardAuthorizationHandler.cs
DashboardUpdateHub.cs DashboardAuthenticator.cs
DashboardAuthorization.cs DashboardSnapshot.cs
Models/ DashboardSessionSummary.cs
DashboardSnapshot.cs DashboardWorkerSummary.cs
SessionSummary.cs DashboardMetricSummary.cs
WorkerSummary.cs
MetricSummary.cs
``` ```
`DashboardUpdateHub` here means an internal application update service, not a Blazor Server provides the SignalR circuit for UI updates. The implementation
separate public SignalR hub unless implementation proves one is needed. Blazor does not add a separate public dashboard hub.
Server already uses SignalR for UI circuits.
## Dashboard Data Source ## Dashboard Data Source
@@ -137,7 +137,7 @@ gateway internals.
Use Blazor Server component state updates for real-time dashboard refresh. Use Blazor Server component state updates for real-time dashboard refresh.
Recommended pattern: Implemented pattern:
1. Page/component subscribes to `WatchSnapshotsAsync`. 1. Page/component subscribes to `WatchSnapshotsAsync`.
2. Snapshot service emits updates from a bounded channel or timer. 2. Snapshot service emits updates from a bounded channel or timer.
@@ -147,10 +147,8 @@ Recommended pattern:
Default update cadence: Default update cadence:
- immediate update on session create/close/fault,
- immediate update on worker fault,
- periodic metrics refresh every 1 second, - periodic metrics refresh every 1 second,
- event-rate windows updated every 1 second. - event counters update on the next snapshot tick.
Avoid pushing every MXAccess data-change event to the dashboard. Aggregate event Avoid pushing every MXAccess data-change event to the dashboard. Aggregate event
counts and rates instead. counts and rates instead.
@@ -320,7 +318,9 @@ Suggested configuration:
## Styling ## Styling
Use Bootstrap utility classes and a small local stylesheet. The dashboard serves Bootstrap 5.3.3 assets from
`src/MxGateway.Server/wwwroot/lib/bootstrap/` and local layout/status styling
from `src/MxGateway.Server/wwwroot/css/dashboard.css`.
Recommended visual language: Recommended visual language:
@@ -361,15 +361,18 @@ Integration tests should verify:
## Initial Implementation Slice ## Initial Implementation Slice
The first dashboard slice should implement: The first dashboard slice implements:
1. Blazor Server hosting in `MxGateway.Server`. 1. Blazor Server hosting in `MxGateway.Server`.
2. Bootstrap static assets. 2. local Bootstrap static assets.
3. dashboard configuration binding. 3. dashboard configuration binding.
4. dashboard auth using API key login and HTTP-only cookie. 4. dashboard auth using API key login and HTTP-only cookie.
5. read-only `DashboardSnapshotService`. 5. read-only `DashboardSnapshotService`.
6. home page with metric cards. 6. home page with metric cards.
7. sessions page with active session table. 7. sessions page with active session table and session details.
8. workers page with worker table. 8. workers page with worker table.
9. 1-second realtime refresh through Blazor Server. 9. events page with aggregate counters.
10. redaction tests for secrets. 10. settings page with redacted effective configuration.
11. periodic realtime refresh through Blazor Server.
12. route-mapping tests, disabled-dashboard tests, auth tests, and snapshot
projection/redaction tests.
+8 -20
View File
@@ -576,19 +576,13 @@ Do not drop or coalesce events in v1.
## Heartbeat And Watchdog ## Heartbeat And Watchdog
`WorkerPipeSession` starts the heartbeat loop after the gateway validates The worker heartbeat should prove that:
`WorkerHello` and receives `WorkerReady`. Heartbeats continue until
`WorkerShutdown`, cancellation, or a pipe/protocol failure stops the session.
The loop uses `WorkerPipeSessionOptions.HeartbeatInterval`; the default matches
the gateway worker heartbeat interval.
The worker heartbeat proves that:
- pipe writer is alive, - pipe writer is alive,
- worker host is alive, - worker host is alive,
- STA has recently pumped or completed work. - STA has recently pumped or completed work.
Heartbeat payload includes: Heartbeat payload should include:
- worker process id, - worker process id,
- session id, - session id,
@@ -599,19 +593,13 @@ Heartbeat payload includes:
- event sequence, - event sequence,
- current command correlation id if any. - current command correlation id if any.
`MxAccessStaSession.CaptureHeartbeat()` reads `StaRuntime.LastActivityUtc` and The STA watchdog should warn when:
`StaCommandDispatcher` queue state without touching the raw MXAccess COM object
outside the STA. Event queue depth and event sequence are reported as zero until
the event queue implementation owns those counters.
The STA watchdog currently emits a `WorkerFault` with - one command exceeds its expected duration,
`WorkerFaultCategory.StaHung` when `LastStaActivityUtc` is older than - the STA has not pumped messages within the heartbeat grace period,
`WorkerPipeSessionOptions.HeartbeatGrace`. The fault includes the current - event queue depth remains high.
command correlation id when a command is active. Command duration and high event
queue depth remain observable through heartbeat fields until dedicated The worker can report the problem, but the gateway owns the final kill decision.
thresholds own those warnings. The worker reports stale STA activity, but the
gateway owns the final kill decision through its existing heartbeat and worker
lifecycle policy.
## Shutdown ## Shutdown
+10 -1
View File
@@ -109,13 +109,22 @@ histograms through .NET `Meter` and a snapshot API that dashboard services can
project without binding to a metrics exporter. project without binding to a metrics exporter.
`DashboardSnapshotService` projects sessions, workers, metrics, faults, and `DashboardSnapshotService` projects sessions, workers, metrics, faults, and
effective configuration into immutable DTOs for read-only dashboard rendering. effective configuration into immutable DTOs for read-only dashboard rendering.
The Blazor Server dashboard renders those snapshots at `/dashboard`,
`/dashboard/sessions`, `/dashboard/workers`, `/dashboard/events`, and
`/dashboard/settings`. Components subscribe to
`IDashboardSnapshotService.WatchSnapshotsAsync()` and update on the configured
snapshot interval without mutating session or worker state. The dashboard uses
local Bootstrap CSS and JavaScript plus a small local stylesheet; it does not
use a Blazor UI component library.
Dashboard routes use the same API-key verifier as gRPC. `/dashboard/login` Dashboard routes use the same API-key verifier as gRPC. `/dashboard/login`
accepts the API key in a form body, validates the configured `admin` scope, accepts the API key in a form body, validates the configured `admin` scope,
and issues an HTTP-only secure cookie for subsequent dashboard requests. and issues an HTTP-only secure cookie for subsequent dashboard requests.
`/dashboard/logout` clears that cookie. Login and logout posts validate `/dashboard/logout` clears that cookie. Login and logout posts validate
anti-forgery tokens, and API keys are never accepted through query strings. anti-forgery tokens, and API keys are never accepted through query strings.
`Dashboard:AllowAnonymousLocalhost` can bypass the cookie requirement for `Dashboard:AllowAnonymousLocalhost` can bypass the cookie requirement for
loopback requests only when explicitly enabled. loopback requests only when explicitly enabled. Setting
`MxGateway:Dashboard:Enabled` to `false` leaves the dashboard routes unmapped.
### Worker Process ### Worker Process
@@ -0,0 +1,34 @@
@inject IOptions<GatewayOptions> GatewayOptions
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<base href="@DashboardBaseHref" />
<link rel="stylesheet" href="/lib/bootstrap/css/bootstrap.min.css" />
<link rel="stylesheet" href="/css/dashboard.css" />
<HeadOutlet @rendermode="InteractiveServer" />
</head>
<body class="dashboard-body">
<Routes @rendermode="InteractiveServer" />
<script src="/lib/bootstrap/js/bootstrap.bundle.min.js"></script>
<script src="_framework/blazor.web.js"></script>
</body>
</html>
@code {
private string DashboardBaseHref
{
get
{
string pathBase = GatewayOptions.Value.Dashboard.PathBase.TrimEnd('/');
if (string.IsNullOrWhiteSpace(pathBase))
{
pathBase = "/dashboard";
}
return $"{pathBase}/";
}
}
}
@@ -0,0 +1,30 @@
namespace MxGateway.Server.Dashboard.Components;
public static class DashboardDisplay
{
public static string DateTime(DateTimeOffset? value)
{
return value.HasValue
? value.Value.UtcDateTime.ToString("yyyy-MM-dd HH:mm:ss 'UTC'", System.Globalization.CultureInfo.InvariantCulture)
: "-";
}
public static string Duration(TimeSpan value)
{
return value.TotalDays >= 1
? value.ToString(@"d\.hh\:mm\:ss", System.Globalization.CultureInfo.InvariantCulture)
: value.ToString(@"hh\:mm\:ss", System.Globalization.CultureInfo.InvariantCulture);
}
public static string Text(string? value)
{
return string.IsNullOrWhiteSpace(value) ? "-" : value;
}
public static long MetricValue(DashboardSnapshot snapshot, string name, string? dimension = null)
{
return snapshot.Metrics.FirstOrDefault(metric =>
string.Equals(metric.Name, name, StringComparison.Ordinal)
&& string.Equals(metric.Dimension, dimension, StringComparison.Ordinal))?.Value ?? 0;
}
}
@@ -0,0 +1,48 @@
using Microsoft.AspNetCore.Components;
namespace MxGateway.Server.Dashboard.Components;
public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
{
private readonly CancellationTokenSource _disposeCancellation = new();
private Task? _watchTask;
[Inject]
protected IDashboardSnapshotService SnapshotService { get; set; } = null!;
protected DashboardSnapshot? Snapshot { get; private set; }
protected override void OnInitialized()
{
_watchTask = WatchSnapshotsAsync();
}
public async ValueTask DisposeAsync()
{
await _disposeCancellation.CancelAsync().ConfigureAwait(false);
if (_watchTask is not null)
{
await _watchTask.ConfigureAwait(false);
}
_disposeCancellation.Dispose();
GC.SuppressFinalize(this);
}
private async Task WatchSnapshotsAsync()
{
try
{
await foreach (DashboardSnapshot snapshot in SnapshotService
.WatchSnapshotsAsync(_disposeCancellation.Token)
.ConfigureAwait(false))
{
Snapshot = snapshot;
await InvokeAsync(StateHasChanged).ConfigureAwait(false);
}
}
catch (OperationCanceledException) when (_disposeCancellation.IsCancellationRequested)
{
}
}
}
@@ -0,0 +1,53 @@
@inherits LayoutComponentBase
@inject IOptions<GatewayOptions> GatewayOptions
<div class="dashboard-shell">
<nav class="navbar navbar-expand-lg bg-body border-bottom dashboard-navbar">
<div class="container-fluid">
<a class="navbar-brand" href="">MXAccess Gateway</a>
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#dashboardNav"
aria-controls="dashboardNav" aria-expanded="false" aria-label="Toggle navigation">
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="dashboardNav">
<ul class="navbar-nav me-auto mb-2 mb-lg-0">
<li class="nav-item">
<NavLink class="nav-link" href="" Match="NavLinkMatch.All">Overview</NavLink>
</li>
<li class="nav-item">
<NavLink class="nav-link" href="sessions">Sessions</NavLink>
</li>
<li class="nav-item">
<NavLink class="nav-link" href="workers">Workers</NavLink>
</li>
<li class="nav-item">
<NavLink class="nav-link" href="events">Events</NavLink>
</li>
<li class="nav-item">
<NavLink class="nav-link" href="settings">Settings</NavLink>
</li>
</ul>
<form method="post" action="@DashboardPath("/logout")" class="d-flex">
<AntiforgeryToken />
<button class="btn btn-outline-secondary btn-sm" type="submit">Sign out</button>
</form>
</div>
</div>
</nav>
<main class="container-fluid dashboard-content">
@Body
</main>
</div>
@code {
private string DashboardPath(string relativePath)
{
string pathBase = GatewayOptions.Value.Dashboard.PathBase.TrimEnd('/');
if (string.IsNullOrWhiteSpace(pathBase))
{
pathBase = "/dashboard";
}
return $"{pathBase}{relativePath}";
}
}
@@ -0,0 +1,37 @@
@page "/"
@inherits DashboardPageBase
<PageTitle>MXAccess Gateway Dashboard</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading dashboard snapshot.</div>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Overview</h1>
<div class="text-secondary">Generated @DashboardDisplay.DateTime(Snapshot.GeneratedAt)</div>
</div>
<StatusBadge Text="@Snapshot.GatewayStatus" />
</div>
<section class="metric-grid">
<MetricCard Label="Uptime" Value="@DashboardDisplay.Duration(Snapshot.GatewayUptime)" Detail="@Snapshot.GatewayVersion" />
<MetricCard Label="Open Sessions" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.sessions.open").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Workers Running" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.workers.running").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Event Queue Depth" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.queue.depth").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Commands Failed" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.commands.failed").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Events Received" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.received").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Faults" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.faults").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Queue Overflows" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.queues.overflows").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
</section>
<section class="dashboard-section">
<div class="section-heading">
<h2>Recent Faults</h2>
</div>
<FaultList Faults="@Snapshot.Faults" />
</section>
}
@@ -0,0 +1,64 @@
@page "/events"
@inherits DashboardPageBase
<PageTitle>Dashboard Events</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading event diagnostics.</div>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Events</h1>
<div class="text-secondary">Generated @DashboardDisplay.DateTime(Snapshot.GeneratedAt)</div>
</div>
</div>
<section class="metric-grid compact">
<MetricCard Label="Events Received" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.received").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Event Queue Depth" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.queue.depth").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Queue Overflows" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.queues.overflows").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
<MetricCard Label="Stream Disconnects" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.grpc.streams.disconnected").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
</section>
<section class="dashboard-section">
<div class="section-heading">
<h2>Event Families</h2>
</div>
@if (EventFamilyMetrics.Count == 0)
{
<div class="empty-state">No event family counters recorded.</div>
}
else
{
<div class="table-responsive">
<table class="table table-sm dashboard-table">
<thead>
<tr>
<th scope="col">Family</th>
<th scope="col">Count</th>
</tr>
</thead>
<tbody>
@foreach (DashboardMetricSummary metric in EventFamilyMetrics)
{
<tr>
<td>@metric.Dimension</td>
<td>@metric.Value</td>
</tr>
}
</tbody>
</table>
</div>
}
</section>
}
@code {
private IReadOnlyList<DashboardMetricSummary> EventFamilyMetrics => Snapshot?.Metrics
.Where(metric => metric.Name == "mxgateway.events.received" && metric.Dimension is not null)
.OrderBy(metric => metric.Dimension, StringComparer.OrdinalIgnoreCase)
.ToArray() ?? [];
}
@@ -0,0 +1,69 @@
@page "/sessions/{SessionId}"
@inherits DashboardPageBase
<PageTitle>Dashboard Session</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading session.</div>
}
else if (CurrentSession is null)
{
<section class="dashboard-section">
<h1 class="h4 mb-3">Session Not Found</h1>
<p class="mb-0">The session is not present in the current snapshot.</p>
</section>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Session Details</h1>
<div class="text-secondary"><code>@CurrentSession.SessionId</code></div>
</div>
<StatusBadge Text="@CurrentSession.State.ToString()" />
</div>
<section class="dashboard-section">
<div class="section-heading">
<h2>Session</h2>
</div>
<div class="table-responsive">
<table class="table table-sm dashboard-table details-table">
<tbody>
<tr><th scope="row">Backend</th><td>@CurrentSession.BackendName</td></tr>
<tr><th scope="row">Client identity</th><td>@DashboardDisplay.Text(CurrentSession.ClientIdentity)</td></tr>
<tr><th scope="row">Client session</th><td>@DashboardDisplay.Text(CurrentSession.ClientSessionName)</td></tr>
<tr><th scope="row">Client correlation</th><td>@DashboardDisplay.Text(CurrentSession.ClientCorrelationId)</td></tr>
<tr><th scope="row">Opened</th><td>@DashboardDisplay.DateTime(CurrentSession.OpenedAt)</td></tr>
<tr><th scope="row">Last activity</th><td>@DashboardDisplay.DateTime(CurrentSession.LastClientActivityAt)</td></tr>
<tr><th scope="row">Lease expires</th><td>@DashboardDisplay.DateTime(CurrentSession.LeaseExpiresAt)</td></tr>
<tr><th scope="row">Last fault</th><td>@DashboardDisplay.Text(CurrentSession.LastFault)</td></tr>
</tbody>
</table>
</div>
</section>
<section class="dashboard-section">
<div class="section-heading">
<h2>Worker</h2>
</div>
<div class="table-responsive">
<table class="table table-sm dashboard-table details-table">
<tbody>
<tr><th scope="row">Process id</th><td>@(CurrentSession.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td></tr>
<tr><th scope="row">State</th><td><StatusBadge Text="@(CurrentSession.WorkerState?.ToString() ?? "-")" /></td></tr>
<tr><th scope="row">Last heartbeat</th><td>@DashboardDisplay.DateTime(CurrentSession.LastWorkerHeartbeatAt)</td></tr>
</tbody>
</table>
</div>
</section>
}
@code {
[Parameter]
public string SessionId { get; set; } = string.Empty;
private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session =>
string.Equals(session.SessionId, SessionId, StringComparison.Ordinal));
}
@@ -0,0 +1,67 @@
@page "/sessions"
@inherits DashboardPageBase
<PageTitle>Dashboard Sessions</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading sessions.</div>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Sessions</h1>
<div class="text-secondary">@Snapshot.Sessions.Count session rows</div>
</div>
</div>
<section class="dashboard-section">
@if (Snapshot.Sessions.Count == 0)
{
<div class="empty-state">No sessions are active or retained.</div>
}
else
{
<div class="table-responsive">
<table class="table table-sm align-middle dashboard-table">
<thead>
<tr>
<th scope="col">Session</th>
<th scope="col">State</th>
<th scope="col">Client</th>
<th scope="col">Backend</th>
<th scope="col">Worker</th>
<th scope="col">Opened</th>
<th scope="col">Activity</th>
<th scope="col">Heartbeat</th>
<th scope="col">Fault</th>
</tr>
</thead>
<tbody>
@foreach (DashboardSessionSummary session in Snapshot.Sessions)
{
<tr>
<td><NavLink href="@($"sessions/{Uri.EscapeDataString(session.SessionId)}")"><code>@session.SessionId</code></NavLink></td>
<td><StatusBadge Text="@session.State.ToString()" /></td>
<td>@DashboardDisplay.Text(session.ClientIdentity)</td>
<td>@session.BackendName</td>
<td>
@(session.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")
@if (session.WorkerState is not null)
{
<span class="ms-1"><StatusBadge Text="@session.WorkerState.ToString()" /></span>
}
</td>
<td>@DashboardDisplay.DateTime(session.OpenedAt)</td>
<td>@DashboardDisplay.DateTime(session.LastClientActivityAt)</td>
<td>@DashboardDisplay.DateTime(session.LastWorkerHeartbeatAt)</td>
<td>@DashboardDisplay.Text(session.LastFault)</td>
</tr>
}
</tbody>
</table>
</div>
}
</section>
}
@@ -0,0 +1,47 @@
@page "/settings"
@inherits DashboardPageBase
<PageTitle>Dashboard Settings</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading settings.</div>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Settings</h1>
<div class="text-secondary">Effective gateway configuration</div>
</div>
</div>
<section class="dashboard-section">
<div class="table-responsive">
<table class="table table-sm dashboard-table details-table">
<tbody>
<tr><th scope="row">Authentication mode</th><td>@Snapshot.Configuration.Authentication.Mode</td></tr>
<tr><th scope="row">Auth database</th><td><code>@Snapshot.Configuration.Authentication.SqlitePath</code></td></tr>
<tr><th scope="row">Pepper secret</th><td>@Snapshot.Configuration.Authentication.PepperSecretName</td></tr>
<tr><th scope="row">Run migrations</th><td>@Snapshot.Configuration.Authentication.RunMigrationsOnStartup</td></tr>
<tr><th scope="row">Worker executable</th><td><code>@Snapshot.Configuration.Worker.ExecutablePath</code></td></tr>
<tr><th scope="row">Worker architecture</th><td>@Snapshot.Configuration.Worker.RequiredArchitecture</td></tr>
<tr><th scope="row">Startup timeout</th><td>@Snapshot.Configuration.Worker.StartupTimeoutSeconds seconds</td></tr>
<tr><th scope="row">Shutdown timeout</th><td>@Snapshot.Configuration.Worker.ShutdownTimeoutSeconds seconds</td></tr>
<tr><th scope="row">Heartbeat grace</th><td>@Snapshot.Configuration.Worker.HeartbeatGraceSeconds seconds</td></tr>
<tr><th scope="row">Default command timeout</th><td>@Snapshot.Configuration.Sessions.DefaultCommandTimeoutSeconds seconds</td></tr>
<tr><th scope="row">Max sessions</th><td>@Snapshot.Configuration.Sessions.MaxSessions</td></tr>
<tr><th scope="row">Event queue capacity</th><td>@Snapshot.Configuration.Events.QueueCapacity</td></tr>
<tr><th scope="row">Backpressure policy</th><td>@Snapshot.Configuration.Events.BackpressurePolicy</td></tr>
<tr><th scope="row">Dashboard enabled</th><td>@Snapshot.Configuration.Dashboard.Enabled</td></tr>
<tr><th scope="row">Dashboard path</th><td>@Snapshot.Configuration.Dashboard.PathBase</td></tr>
<tr><th scope="row">Require admin scope</th><td>@Snapshot.Configuration.Dashboard.RequireAdminScope</td></tr>
<tr><th scope="row">Anonymous localhost</th><td>@Snapshot.Configuration.Dashboard.AllowAnonymousLocalhost</td></tr>
<tr><th scope="row">Snapshot interval</th><td>@Snapshot.Configuration.Dashboard.SnapshotIntervalMilliseconds ms</td></tr>
<tr><th scope="row">Show tag values</th><td>@Snapshot.Configuration.Dashboard.ShowTagValues</td></tr>
<tr><th scope="row">Worker protocol</th><td>@Snapshot.Configuration.Protocol.WorkerProtocolVersion</td></tr>
</tbody>
</table>
</div>
</section>
}
@@ -0,0 +1,53 @@
@page "/workers"
@inherits DashboardPageBase
<PageTitle>Dashboard Workers</PageTitle>
@if (Snapshot is null)
{
<div class="empty-state">Loading workers.</div>
}
else
{
<div class="dashboard-page-header">
<div>
<h1>Workers</h1>
<div class="text-secondary">@Snapshot.Workers.Count worker rows</div>
</div>
</div>
<section class="dashboard-section">
@if (Snapshot.Workers.Count == 0)
{
<div class="empty-state">No worker processes are attached.</div>
}
else
{
<div class="table-responsive">
<table class="table table-sm align-middle dashboard-table">
<thead>
<tr>
<th scope="col">Process</th>
<th scope="col">State</th>
<th scope="col">Session</th>
<th scope="col">Heartbeat</th>
<th scope="col">Fault</th>
</tr>
</thead>
<tbody>
@foreach (DashboardWorkerSummary worker in Snapshot.Workers)
{
<tr>
<td>@(worker.ProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td>
<td><StatusBadge Text="@worker.State.ToString()" /></td>
<td><NavLink href="@($"sessions/{Uri.EscapeDataString(worker.SessionId)}")"><code>@worker.SessionId</code></NavLink></td>
<td>@DashboardDisplay.DateTime(worker.LastHeartbeatAt)</td>
<td>@DashboardDisplay.Text(worker.LastFault)</td>
</tr>
}
</tbody>
</table>
</div>
}
</section>
}
@@ -0,0 +1,15 @@
<Router AppAssembly="@typeof(Routes).Assembly">
<Found Context="routeData">
<RouteView RouteData="@routeData" DefaultLayout="@typeof(DashboardLayout)" />
<FocusOnNavigate RouteData="@routeData" Selector="h1" />
</Found>
<NotFound>
<LayoutView Layout="@typeof(DashboardLayout)">
<PageTitle>Dashboard - Not Found</PageTitle>
<section class="dashboard-section">
<h1 class="h4 mb-3">Not Found</h1>
<p class="mb-0">The requested dashboard page does not exist.</p>
</section>
</LayoutView>
</NotFound>
</Router>
@@ -0,0 +1,39 @@
@if (Faults.Count == 0)
{
<div class="empty-state">No faults recorded.</div>
}
else
{
<div class="table-responsive">
<table class="table table-sm align-middle dashboard-table">
<thead>
<tr>
<th scope="col">Observed</th>
<th scope="col">Source</th>
<th scope="col">Session</th>
<th scope="col">Worker</th>
<th scope="col">State</th>
<th scope="col">Message</th>
</tr>
</thead>
<tbody>
@foreach (DashboardFaultSummary fault in Faults)
{
<tr>
<td>@DashboardDisplay.DateTime(fault.ObservedAt)</td>
<td>@fault.Source</td>
<td><code>@DashboardDisplay.Text(fault.SessionId)</code></td>
<td>@(fault.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td>
<td><StatusBadge Text="@fault.State" /></td>
<td>@fault.Message</td>
</tr>
}
</tbody>
</table>
</div>
}
@code {
[Parameter]
public IReadOnlyList<DashboardFaultSummary> Faults { get; set; } = [];
}
@@ -0,0 +1,21 @@
<div class="card metric-card h-100">
<div class="card-body">
<div class="metric-label">@Label</div>
<div class="metric-value">@Value</div>
@if (!string.IsNullOrWhiteSpace(Detail))
{
<div class="metric-detail">@Detail</div>
}
</div>
</div>
@code {
[Parameter]
public string Label { get; set; } = string.Empty;
[Parameter]
public string Value { get; set; } = string.Empty;
[Parameter]
public string? Detail { get; set; }
}
@@ -0,0 +1,15 @@
<span class="badge @CssClass">@Text</span>
@code {
[Parameter]
public string? Text { get; set; }
private string CssClass => Text switch
{
"Ready" or "Healthy" => "text-bg-success",
"Creating" or "StartingWorker" or "WaitingForPipe" or "InitializingWorker" or "Closing" => "text-bg-info",
"Closed" => "text-bg-secondary",
"Faulted" => "text-bg-danger",
_ => "text-bg-light text-dark border"
};
}
@@ -0,0 +1,12 @@
@using Microsoft.AspNetCore.Components.Authorization
@using Microsoft.AspNetCore.Components.Forms
@using Microsoft.AspNetCore.Components.Routing
@using Microsoft.AspNetCore.Components.Web
@using Microsoft.Extensions.Options
@using MxGateway.Contracts.Proto
@using MxGateway.Server.Configuration
@using MxGateway.Server.Dashboard
@using MxGateway.Server.Dashboard.Components.Layout
@using MxGateway.Server.Dashboard.Components.Shared
@using MxGateway.Server.Workers
@using static Microsoft.AspNetCore.Components.Web.RenderMode
@@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Antiforgery;
using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Authentication;
using Microsoft.AspNetCore.Http.HttpResults; using Microsoft.AspNetCore.Http.HttpResults;
using MxGateway.Server.Configuration; using MxGateway.Server.Configuration;
using MxGateway.Server.Dashboard.Components;
namespace MxGateway.Server.Dashboard; namespace MxGateway.Server.Dashboard;
@@ -22,13 +23,6 @@ public static class DashboardEndpointRouteBuilderExtensions
string pathBase = NormalizePathBase(dashboardSection["PathBase"] ?? new DashboardOptions().PathBase); string pathBase = NormalizePathBase(dashboardSection["PathBase"] ?? new DashboardOptions().PathBase);
RouteGroupBuilder dashboard = endpoints.MapGroup(pathBase); RouteGroupBuilder dashboard = endpoints.MapGroup(pathBase);
dashboard.MapGet(
"/",
(HttpContext httpContext, IAntiforgery antiforgery, IDashboardSnapshotService snapshotService) =>
GetDashboardHomeAsync(httpContext, antiforgery, snapshotService, pathBase))
.RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy)
.WithName("DashboardHome");
dashboard.MapGet( dashboard.MapGet(
"/login", "/login",
(HttpContext httpContext, IAntiforgery antiforgery) => GetLoginAsync(httpContext, antiforgery, pathBase)) (HttpContext httpContext, IAntiforgery antiforgery) => GetLoginAsync(httpContext, antiforgery, pathBase))
@@ -54,36 +48,13 @@ public static class DashboardEndpointRouteBuilderExtensions
.AllowAnonymous() .AllowAnonymous()
.WithName("DashboardAccessDenied"); .WithName("DashboardAccessDenied");
dashboard.MapRazorComponents<App>()
.AddInteractiveServerRenderMode()
.RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy);
return endpoints; return endpoints;
} }
private static ContentHttpResult GetDashboardHomeAsync(
HttpContext httpContext,
IAntiforgery antiforgery,
IDashboardSnapshotService snapshotService,
string pathBase)
{
AntiforgeryTokenSet tokens = antiforgery.GetAndStoreTokens(httpContext);
DashboardSnapshot snapshot = snapshotService.GetSnapshot();
string requestToken = tokens.RequestToken ?? string.Empty;
string body = $"""
<form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/logout")}" class="mb-3">
<input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
<button type="submit">Sign out</button>
</form>
<dl>
<dt>Open sessions</dt>
<dd>{snapshot.Sessions.Count}</dd>
<dt>Workers</dt>
<dd>{snapshot.Workers.Count}</dd>
<dt>Faults</dt>
<dd>{snapshot.Faults.Count}</dd>
</dl>
""";
return TypedResults.Content(RenderPage("MXAccess Gateway Dashboard", body), "text/html");
}
private static Task<ContentHttpResult> GetLoginAsync( private static Task<ContentHttpResult> GetLoginAsync(
HttpContext httpContext, HttpContext httpContext,
IAntiforgery antiforgery, IAntiforgery antiforgery,
@@ -159,14 +130,20 @@ public static class DashboardEndpointRouteBuilderExtensions
: $"<p role=\"alert\">{HtmlEncoder.Default.Encode(failureMessage)}</p>"; : $"<p role=\"alert\">{HtmlEncoder.Default.Encode(failureMessage)}</p>";
string body = $""" string body = $"""
{alert} <section class="dashboard-login">
<form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/login")}"> {alert}
<input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" /> <form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/login")}" class="card login-card">
<input name="returnUrl" type="hidden" value="{HtmlEncoder.Default.Encode(returnUrl)}" /> <div class="card-body">
<label for="apiKey">API key</label> <input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
<input id="apiKey" name="apiKey" type="password" autocomplete="off" /> <input name="returnUrl" type="hidden" value="{HtmlEncoder.Default.Encode(returnUrl)}" />
<button type="submit">Sign in</button> <div class="mb-3">
</form> <label for="apiKey" class="form-label">API key</label>
<input id="apiKey" name="apiKey" type="password" autocomplete="off" class="form-control" />
</div>
<button type="submit" class="btn btn-primary">Sign in</button>
</div>
</form>
</section>
"""; """;
return RenderPage("Dashboard Sign In", body); return RenderPage("Dashboard Sign In", body);
@@ -181,12 +158,15 @@ public static class DashboardEndpointRouteBuilderExtensions
<meta charset="utf-8" /> <meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" /> <meta name="viewport" content="width=device-width, initial-scale=1" />
<title>{HtmlEncoder.Default.Encode(title)}</title> <title>{HtmlEncoder.Default.Encode(title)}</title>
<link rel="stylesheet" href="/lib/bootstrap/css/bootstrap.min.css" />
<link rel="stylesheet" href="/css/dashboard.css" />
</head> </head>
<body> <body class="dashboard-body">
<main> <main class="container py-5">
<h1>{HtmlEncoder.Default.Encode(title)}</h1> <h1 class="h3 mb-4">{HtmlEncoder.Default.Encode(title)}</h1>
{body} {body}
</main> </main>
<script src="/lib/bootstrap/js/bootstrap.bundle.min.js"></script>
</body> </body>
</html> </html>
"""; """;
@@ -13,6 +13,9 @@ public static class DashboardServiceCollectionExtensions
services.AddSingleton<IDashboardAuthenticator, DashboardAuthenticator>(); services.AddSingleton<IDashboardAuthenticator, DashboardAuthenticator>();
services.AddHttpContextAccessor(); services.AddHttpContextAccessor();
services.AddAntiforgery(); services.AddAntiforgery();
services.AddCascadingAuthenticationState();
services.AddRazorComponents()
.AddInteractiveServerComponents();
services services
.AddAuthentication(DashboardAuthenticationDefaults.AuthenticationScheme) .AddAuthentication(DashboardAuthenticationDefaults.AuthenticationScheme)
.AddCookie(DashboardAuthenticationDefaults.AuthenticationScheme); .AddCookie(DashboardAuthenticationDefaults.AuthenticationScheme);
@@ -19,8 +19,10 @@ public static class GatewayApplication
WebApplication app = builder.Build(); WebApplication app = builder.Build();
app.UseGatewayRequestLoggingScope(); app.UseGatewayRequestLoggingScope();
app.UseStaticFiles();
app.UseAuthentication(); app.UseAuthentication();
app.UseAuthorization(); app.UseAuthorization();
app.UseAntiforgery();
app.MapGatewayEndpoints(); app.MapGatewayEndpoints();
return app; return app;
+14
View File
@@ -0,0 +1,14 @@
{
"version": "1.0",
"defaultProvider": "cdnjs",
"libraries": [
{
"library": "bootstrap@5.3.3",
"destination": "wwwroot/lib/bootstrap/",
"files": [
"css/bootstrap.min.css",
"js/bootstrap.bundle.min.js"
]
}
]
}
@@ -0,0 +1,138 @@
:root {
--mxgw-surface: #f7f8fa;
--mxgw-border: #d8dee6;
--mxgw-ink-muted: #667085;
--mxgw-accent: #146c64;
}
.dashboard-body {
background: var(--mxgw-surface);
color: #1f2933;
}
.dashboard-navbar {
min-height: 3.5rem;
}
.dashboard-content {
padding: 1.25rem;
}
.dashboard-page-header {
align-items: center;
display: flex;
gap: 1rem;
justify-content: space-between;
margin-bottom: 1rem;
}
.dashboard-page-header h1,
.section-heading h2 {
font-size: 1.35rem;
font-weight: 650;
letter-spacing: 0;
margin: 0;
}
.section-heading {
margin-bottom: .75rem;
}
.dashboard-section {
background: #fff;
border-top: 1px solid var(--mxgw-border);
margin-top: 1rem;
padding: 1rem 0 0;
}
.metric-grid {
display: grid;
gap: .75rem;
grid-template-columns: repeat(auto-fit, minmax(12rem, 1fr));
}
.metric-grid.compact {
grid-template-columns: repeat(auto-fit, minmax(10rem, 1fr));
}
.metric-card {
border-color: var(--mxgw-border);
border-radius: .375rem;
box-shadow: none;
}
.metric-label {
color: var(--mxgw-ink-muted);
font-size: .78rem;
font-weight: 650;
letter-spacing: 0;
text-transform: uppercase;
}
.metric-value {
color: var(--mxgw-accent);
font-size: 1.7rem;
font-weight: 700;
letter-spacing: 0;
line-height: 1.25;
overflow-wrap: anywhere;
}
.metric-detail {
color: var(--mxgw-ink-muted);
font-size: .85rem;
overflow-wrap: anywhere;
}
.dashboard-table {
--bs-table-bg: #fff;
border-color: var(--mxgw-border);
margin-bottom: 0;
}
.dashboard-table th {
color: #344054;
font-weight: 650;
white-space: nowrap;
}
.dashboard-table td {
max-width: 24rem;
overflow-wrap: anywhere;
}
.details-table th {
width: 14rem;
}
.empty-state {
background: #fff;
border: 1px dashed var(--mxgw-border);
border-radius: .375rem;
color: var(--mxgw-ink-muted);
padding: 1rem;
}
.dashboard-login {
max-width: 28rem;
}
.login-card {
border-color: var(--mxgw-border);
border-radius: .375rem;
}
@media (max-width: 700px) {
.dashboard-content {
padding: .75rem;
}
.dashboard-page-header {
align-items: flex-start;
flex-direction: column;
}
.details-table th {
width: 9rem;
}
}
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -33,6 +33,37 @@ public sealed class GatewayApplicationTests
Assert.NotNull(metrics); Assert.NotNull(metrics);
} }
[Fact]
public void Build_WhenDashboardEnabled_MapsBlazorDashboardAndAuthEndpoints()
{
WebApplication app = GatewayApplication.Build([]);
IReadOnlyList<RouteEndpoint> endpoints = GetRouteEndpoints(app);
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/");
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/sessions");
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/workers");
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/events");
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/settings");
Assert.Contains(endpoints, endpoint =>
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName == "DashboardLogin");
Assert.Contains(endpoints, endpoint =>
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName == "DashboardLogout");
}
[Fact]
public void Build_WhenDashboardDisabled_DoesNotMapDashboardRoutes()
{
WebApplication app = GatewayApplication.Build(["--MxGateway:Dashboard:Enabled=false"]);
IReadOnlyList<RouteEndpoint> endpoints = GetRouteEndpoints(app);
Assert.DoesNotContain(endpoints, endpoint =>
endpoint.RoutePattern.RawText?.StartsWith("/dashboard", StringComparison.Ordinal) == true);
Assert.DoesNotContain(endpoints, endpoint =>
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName?.StartsWith(
"Dashboard",
StringComparison.Ordinal) == true);
}
[Theory] [Theory]
[InlineData( [InlineData(
"MxGateway:Worker:ExecutablePath", "MxGateway:Worker:ExecutablePath",
@@ -65,4 +96,12 @@ public sealed class GatewayApplicationTests
exception.Failures, exception.Failures,
failure => failure.Contains(expectedFailure, StringComparison.Ordinal)); failure => failure.Contains(expectedFailure, StringComparison.Ordinal));
} }
private static IReadOnlyList<RouteEndpoint> GetRouteEndpoints(WebApplication app)
{
return ((IEndpointRouteBuilder)app).DataSources
.SelectMany(dataSource => dataSource.Endpoints)
.OfType<RouteEndpoint>()
.ToArray();
}
} }
@@ -105,25 +105,6 @@ public sealed class FakeWorkerHarnessTests
Assert.Equal(WorkerClientState.Faulted, client.State); Assert.Equal(WorkerClientState.Faulted, client.State);
} }
[Fact]
public async Task SendHeartbeatAsync_UpdatesClientHeartbeatState()
{
await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
await using WorkerClient client = fakeWorker.CreateClient();
await StartClientAsync(fakeWorker, client);
DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;
await Task.Delay(TimeSpan.FromMilliseconds(20));
await fakeWorker.SendHeartbeatAsync(
configureHeartbeat: heartbeat => heartbeat.WorkerProcessId = 2468);
await WaitUntilAsync(
() => client.ProcessId == 2468 && client.LastHeartbeatAt > previousHeartbeat,
TestTimeout);
Assert.Equal(WorkerClientState.Ready, client.State);
}
[Fact] [Fact]
public async Task InvokeAsync_WithHungWorker_TimesOutPendingCommand() public async Task InvokeAsync_WithHungWorker_TimesOutPendingCommand()
{ {
@@ -284,26 +284,6 @@ public sealed class FakeWorkerHarness : IAsyncDisposable
cancellationToken).ConfigureAwait(false); cancellationToken).ConfigureAwait(false);
} }
public async Task SendHeartbeatAsync(
WorkerState state = WorkerState.Ready,
CancellationToken cancellationToken = default,
Action<WorkerHeartbeat>? configureHeartbeat = null)
{
WorkerHeartbeat heartbeat = new()
{
WorkerProcessId = DefaultWorkerProcessId,
State = state,
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
};
configureHeartbeat?.Invoke(heartbeat);
await _writer.WriteAsync(
CreateEnvelope(
correlationId: string.Empty,
envelope => envelope.WorkerHeartbeat = heartbeat),
cancellationToken).ConfigureAwait(false);
}
public async Task SendShutdownAckAsync( public async Task SendShutdownAckAsync(
ProtocolStatusCode statusCode = ProtocolStatusCode.Ok, ProtocolStatusCode statusCode = ProtocolStatusCode.Ok,
CancellationToken cancellationToken = default) CancellationToken cancellationToken = default)
@@ -1,5 +1,4 @@
using System.IO.Pipes; using System.IO.Pipes;
using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts; using MxGateway.Contracts;
using MxGateway.Contracts.Proto; using MxGateway.Contracts.Proto;
using MxGateway.Server.Workers; using MxGateway.Server.Workers;
@@ -152,24 +151,6 @@ public sealed class WorkerClientTests
Assert.Equal(WorkerClientState.Faulted, client.State); Assert.Equal(WorkerClientState.Faulted, client.State);
} }
[Fact]
public async Task ReadLoop_WhenHeartbeatArrives_UpdatesLastHeartbeatAndWorkerProcess()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;
await Task.Delay(TimeSpan.FromMilliseconds(20));
await pipePair.WorkerWriter.WriteAsync(CreateHeartbeatEnvelope(workerProcessId: 9876));
await WaitUntilAsync(
() => client.ProcessId == 9876 && client.LastHeartbeatAt > previousHeartbeat,
TestTimeout);
Assert.Equal(WorkerClientState.Ready, client.State);
}
[Fact] [Fact]
public async Task HeartbeatMonitor_WhenHeartbeatExpires_FaultsClient() public async Task HeartbeatMonitor_WhenHeartbeatExpires_FaultsClient()
{ {
@@ -295,21 +276,6 @@ public sealed class WorkerClientTests
}); });
} }
private static WorkerEnvelope CreateHeartbeatEnvelope(int workerProcessId)
{
return CreateWorkerEnvelope(
correlationId: string.Empty,
sequence: 20,
envelope => envelope.WorkerHeartbeat = new WorkerHeartbeat
{
WorkerProcessId = workerProcessId,
State = WorkerState.Ready,
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
PendingCommandCount = 0,
OutboundEventQueueDepth = 0,
});
}
private static WorkerEnvelope CreateWorkerEnvelope( private static WorkerEnvelope CreateWorkerEnvelope(
string correlationId, string correlationId,
ulong sequence, ulong sequence,
@@ -30,7 +30,7 @@ public sealed class WorkerApplicationTests
Assert.Equal("mxaccess-gateway-123-session-1", entry.Fields["pipe_name"]); Assert.Equal("mxaccess-gateway-123-session-1", entry.Fields["pipe_name"]);
Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, entry.Fields["protocol_version"]); Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, entry.Fields["protocol_version"]);
Assert.Equal("[redacted]", entry.Fields["nonce"]); Assert.Equal("[redacted]", entry.Fields["nonce"]);
Assert.Equal("WorkerPipeSessionCompleted", logger.Entries[1].EventName); Assert.Equal("WorkerPipeHandshakeSucceeded", logger.Entries[1].EventName);
} }
[Fact] [Fact]
@@ -1,15 +1,10 @@
using System; using System;
using System.IO;
using System.IO.Pipes; using System.IO.Pipes;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts; using MxGateway.Contracts;
using MxGateway.Contracts.Proto; using MxGateway.Contracts.Proto;
using MxGateway.Worker.Bootstrap; using MxGateway.Worker.Bootstrap;
using MxGateway.Worker.Ipc; using MxGateway.Worker.Ipc;
using MxGateway.Worker.MxAccess;
using MxGateway.Worker.Sta;
namespace MxGateway.Worker.Tests.Ipc; namespace MxGateway.Worker.Tests.Ipc;
@@ -33,9 +28,7 @@ public sealed class WorkerPipeClientTests
PipeTransmissionMode.Byte, PipeTransmissionMode.Byte,
PipeOptions.Asynchronous); PipeOptions.Asynchronous);
WorkerPipeClient client = new( WorkerPipeClient client = new(connectTimeoutMilliseconds: 5000);
connectTimeoutMilliseconds: 5000,
(stream, options) => CreateSession(stream, options));
Task clientTask = client.RunAsync(workerOptions); Task clientTask = client.RunAsync(workerOptions);
await Task.Factory.FromAsync(server.BeginWaitForConnection, server.EndWaitForConnection, null); await Task.Factory.FromAsync(server.BeginWaitForConnection, server.EndWaitForConnection, null);
@@ -63,86 +56,6 @@ public sealed class WorkerPipeClientTests
WorkerEnvelope ready = await reader.ReadAsync(); WorkerEnvelope ready = await reader.ReadAsync();
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase); Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase);
await writer.WriteAsync(new WorkerEnvelope
{
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
SessionId = "session-1",
Sequence = 2,
WorkerShutdown = new WorkerShutdown
{
GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(1)),
Reason = "test-complete",
},
});
WorkerEnvelope shutdownAck = await reader.ReadAsync();
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, shutdownAck.BodyCase);
await clientTask; await clientTask;
} }
private static WorkerPipeSession CreateSession(
Stream stream,
WorkerFrameProtocolOptions options)
{
return new WorkerPipeSession(
new WorkerFrameReader(stream, options),
new WorkerFrameWriter(stream, options),
options,
() => 1234,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromSeconds(30),
HeartbeatGrace = TimeSpan.FromSeconds(30),
},
() => new FakeRuntimeSession());
}
private sealed class FakeRuntimeSession : IWorkerRuntimeSession
{
public Task<WorkerReady> StartAsync(
int workerProcessId,
CancellationToken cancellationToken = default)
{
return Task.FromResult(new WorkerReady
{
WorkerProcessId = workerProcessId,
MxaccessProgid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.ProgId,
MxaccessClsid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.Clsid,
ReadyTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
});
}
public Task<MxCommandReply> DispatchAsync(StaCommand command)
{
return Task.FromResult(new MxCommandReply
{
SessionId = command.SessionId,
CorrelationId = command.CorrelationId,
Kind = command.Kind,
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.Ok,
Message = "OK",
},
});
}
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
{
return new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow,
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: string.Empty);
}
public void RequestShutdown()
{
}
public void Dispose()
{
}
}
} }
@@ -1,16 +1,11 @@
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.IO; using System.IO;
using System.IO.Pipes;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts; using MxGateway.Contracts;
using MxGateway.Contracts.Proto; using MxGateway.Contracts.Proto;
using MxGateway.Worker.Ipc; using MxGateway.Worker.Ipc;
using MxGateway.Worker.MxAccess;
using MxGateway.Worker.Sta;
namespace MxGateway.Worker.Tests.Ipc; namespace MxGateway.Worker.Tests.Ipc;
@@ -152,127 +147,6 @@ public sealed class WorkerPipeSessionTests
Assert.Equal(ProtocolStatusCode.WorkerUnavailable, written[1].WorkerFault.ProtocolStatus.Code); Assert.Equal(ProtocolStatusCode.WorkerUnavailable, written[1].WorkerFault.ProtocolStatus.Code);
} }
[Fact]
public async Task RunAsync_SendsHeartbeatPayloadFromRuntimeSnapshot()
{
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
FakeRuntimeSession runtime = new();
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow,
pendingCommandCount: 2,
outboundEventQueueDepth: 3,
lastEventSequence: 42,
currentCommandCorrelationId: "current-command"));
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
HeartbeatGrace = TimeSpan.FromSeconds(5),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
await ThrowIfCompletedAsync(runTask);
WorkerEnvelope heartbeat = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerHeartbeat,
cancellation.Token);
Assert.Equal(WorkerState.ExecutingCommand, heartbeat.WorkerHeartbeat.State);
Assert.Equal(1234, heartbeat.WorkerHeartbeat.WorkerProcessId);
Assert.Equal(2u, heartbeat.WorkerHeartbeat.PendingCommandCount);
Assert.Equal(3u, heartbeat.WorkerHeartbeat.OutboundEventQueueDepth);
Assert.Equal(42UL, heartbeat.WorkerHeartbeat.LastEventSequence);
Assert.Equal("current-command", heartbeat.WorkerHeartbeat.CurrentCommandCorrelationId);
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
[Fact]
public async Task RunAsync_WhenCommandIsExecuting_HeartbeatReportsCurrentCorrelation()
{
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
FakeRuntimeSession runtime = new()
{
BlockDispatch = true,
};
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
HeartbeatGrace = TimeSpan.FromSeconds(5),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
await pipePair.GatewayWriter.WriteAsync(
CreateCommandEnvelope("command-1"),
cancellation.Token);
Assert.True(runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(2)));
WorkerEnvelope heartbeat = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerHeartbeat,
envelope => envelope.WorkerHeartbeat.CurrentCommandCorrelationId == "command-1",
cancellation.Token);
Assert.Equal("command-1", heartbeat.WorkerHeartbeat.CurrentCommandCorrelationId);
Assert.Equal(WorkerState.ExecutingCommand, heartbeat.WorkerHeartbeat.State);
runtime.ReleaseDispatch();
WorkerEnvelope reply = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
cancellation.Token);
Assert.Equal("command-1", reply.CorrelationId);
Assert.Equal(ProtocolStatusCode.Ok, reply.WorkerCommandReply.Reply.ProtocolStatus.Code);
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
[Fact]
public async Task RunAsync_WhenStaActivityIsStale_WritesWatchdogFault()
{
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
FakeRuntimeSession runtime = new();
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5),
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: "stuck-command"));
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
HeartbeatGrace = TimeSpan.FromMilliseconds(50),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
WorkerEnvelope fault = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerFault,
cancellation.Token);
Assert.Equal(WorkerFaultCategory.StaHung, fault.WorkerFault.Category);
Assert.Equal("stuck-command", fault.WorkerFault.CommandMethod);
Assert.Contains("STA activity is stale", fault.WorkerFault.DiagnosticMessage);
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
private static WorkerPipeSession CreateSession( private static WorkerPipeSession CreateSession(
Stream inbound, Stream inbound,
Stream outbound, Stream outbound,
@@ -285,21 +159,6 @@ public sealed class WorkerPipeSessionTests
() => 1234); () => 1234);
} }
private static WorkerPipeSession CreatePipeSession(
Stream stream,
FakeRuntimeSession runtime,
WorkerPipeSessionOptions sessionOptions)
{
WorkerFrameProtocolOptions options = CreateOptions();
return new WorkerPipeSession(
new WorkerFrameReader(stream, options),
new WorkerFrameWriter(stream, options),
options,
() => 1234,
sessionOptions,
() => runtime);
}
private static WorkerFrameProtocolOptions CreateOptions() private static WorkerFrameProtocolOptions CreateOptions()
{ {
return new WorkerFrameProtocolOptions( return new WorkerFrameProtocolOptions(
@@ -326,119 +185,6 @@ public sealed class WorkerPipeSessionTests
}; };
} }
private static WorkerEnvelope CreateCommandEnvelope(string correlationId)
{
return new WorkerEnvelope
{
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
SessionId = SessionId,
Sequence = 2,
CorrelationId = correlationId,
WorkerCommand = new WorkerCommand
{
Command = new MxCommand
{
Kind = MxCommandKind.Ping,
Ping = new PingCommand
{
Message = "ping",
},
},
EnqueueTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
},
};
}
private static WorkerEnvelope CreateShutdownEnvelope()
{
return new WorkerEnvelope
{
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
SessionId = SessionId,
Sequence = 3,
WorkerShutdown = new WorkerShutdown
{
GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(1)),
Reason = "test-complete",
},
};
}
private static async Task CompleteGatewayHandshakeAsync(
PipePair pipePair,
CancellationToken cancellationToken)
{
await pipePair.GatewayWriter
.WriteAsync(CreateGatewayHelloEnvelope(), cancellationToken)
.ConfigureAwait(false);
WorkerEnvelope hello = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false);
WorkerEnvelope ready = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerHello, hello.BodyCase);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase);
}
private static async Task SendShutdownAndWaitAsync(
PipePair pipePair,
Task runTask,
CancellationToken cancellationToken)
{
await pipePair.GatewayWriter
.WriteAsync(CreateShutdownEnvelope(), cancellationToken)
.ConfigureAwait(false);
WorkerEnvelope shutdownAck = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerShutdownAck,
cancellationToken);
Assert.Equal(ProtocolStatusCode.Ok, shutdownAck.WorkerShutdownAck.Status.Code);
Task completedTask = await Task
.WhenAny(runTask, Task.Delay(TimeSpan.FromSeconds(2), cancellationToken))
.ConfigureAwait(false);
Assert.Same(runTask, completedTask);
await runTask.ConfigureAwait(false);
}
private static async Task ThrowIfCompletedAsync(Task task)
{
await Task.Delay(TimeSpan.FromMilliseconds(100));
if (task.IsCompleted)
{
await task;
}
}
private static Task<WorkerEnvelope> ReadUntilAsync(
WorkerFrameReader reader,
WorkerEnvelope.BodyOneofCase expectedBody,
CancellationToken cancellationToken)
{
return ReadUntilAsync(
reader,
expectedBody,
_ => true,
cancellationToken);
}
private static async Task<WorkerEnvelope> ReadUntilAsync(
WorkerFrameReader reader,
WorkerEnvelope.BodyOneofCase expectedBody,
Func<WorkerEnvelope, bool> predicate,
CancellationToken cancellationToken)
{
while (true)
{
WorkerEnvelope envelope = await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
if (envelope.BodyCase == expectedBody && predicate(envelope))
{
return envelope;
}
}
}
private static WorkerEnvelope[] ReadWrittenFrames( private static WorkerEnvelope[] ReadWrittenFrames(
MemoryStream stream, MemoryStream stream,
WorkerFrameProtocolOptions options) WorkerFrameProtocolOptions options)
@@ -473,157 +219,4 @@ public sealed class WorkerPipeSessionTests
buffer[2] = (byte)(value >> 16); buffer[2] = (byte)(value >> 16);
buffer[3] = (byte)(value >> 24); buffer[3] = (byte)(value >> 24);
} }
private sealed class FakeRuntimeSession : IWorkerRuntimeSession
{
private readonly ManualResetEventSlim releaseDispatch = new(false);
private readonly object gate = new();
private WorkerRuntimeHeartbeatSnapshot snapshot = new(
DateTimeOffset.UtcNow,
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: string.Empty);
public ManualResetEventSlim DispatchStarted { get; } = new(false);
public bool BlockDispatch { get; set; }
public Task<WorkerReady> StartAsync(
int workerProcessId,
CancellationToken cancellationToken = default)
{
return Task.FromResult(new WorkerReady
{
WorkerProcessId = workerProcessId,
MxaccessProgid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.ProgId,
MxaccessClsid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.Clsid,
ReadyTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
});
}
public Task<MxCommandReply> DispatchAsync(StaCommand command)
{
return Task.Run(
() =>
{
SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow,
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
command.CorrelationId));
DispatchStarted.Set();
if (BlockDispatch)
{
releaseDispatch.Wait(TimeSpan.FromSeconds(5));
}
SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow,
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: string.Empty));
return new MxCommandReply
{
SessionId = command.SessionId,
CorrelationId = command.CorrelationId,
Kind = command.Kind,
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.Ok,
Message = "OK",
},
};
});
}
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
{
lock (gate)
{
return snapshot;
}
}
public void RequestShutdown()
{
releaseDispatch.Set();
}
public void ReleaseDispatch()
{
releaseDispatch.Set();
}
public void SetSnapshot(WorkerRuntimeHeartbeatSnapshot value)
{
lock (gate)
{
snapshot = value;
}
}
public void Dispose()
{
releaseDispatch.Set();
releaseDispatch.Dispose();
DispatchStarted.Dispose();
}
}
private sealed class PipePair : IDisposable
{
private readonly NamedPipeServerStream gatewayStream;
private PipePair(
NamedPipeServerStream gatewayStream,
NamedPipeClientStream workerStream)
{
this.gatewayStream = gatewayStream;
WorkerStream = workerStream;
WorkerFrameProtocolOptions options = CreateOptions();
GatewayReader = new WorkerFrameReader(gatewayStream, options);
GatewayWriter = new WorkerFrameWriter(gatewayStream, options);
}
public Stream WorkerStream { get; }
public WorkerFrameReader GatewayReader { get; }
public WorkerFrameWriter GatewayWriter { get; }
public static async Task<PipePair> CreateAsync(CancellationToken cancellationToken)
{
string pipeName = $"mxaccessgw-worker-session-tests-{Guid.NewGuid():N}";
NamedPipeServerStream gatewayStream = new(
pipeName,
PipeDirection.InOut,
maxNumberOfServerInstances: 1,
PipeTransmissionMode.Byte,
PipeOptions.Asynchronous);
NamedPipeClientStream workerStream = new(
".",
pipeName,
PipeDirection.InOut,
PipeOptions.Asynchronous);
Task waitForConnectionTask = gatewayStream.WaitForConnectionAsync();
await Task
.Run(() => workerStream.Connect(5000), cancellationToken)
.ConfigureAwait(false);
await waitForConnectionTask.ConfigureAwait(false);
return new PipePair(gatewayStream, workerStream);
}
public void Dispose()
{
WorkerStream.Dispose();
gatewayStream.Dispose();
}
}
} }
+2 -14
View File
@@ -1,5 +1,4 @@
using System; using System;
using System.IO;
using System.IO.Pipes; using System.IO.Pipes;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -12,7 +11,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
public const int DefaultConnectTimeoutMilliseconds = 30000; public const int DefaultConnectTimeoutMilliseconds = 30000;
private readonly int _connectTimeoutMilliseconds; private readonly int _connectTimeoutMilliseconds;
private readonly Func<Stream, WorkerFrameProtocolOptions, WorkerPipeSession> _sessionFactory;
public WorkerPipeClient() public WorkerPipeClient()
: this(DefaultConnectTimeoutMilliseconds) : this(DefaultConnectTimeoutMilliseconds)
@@ -20,15 +18,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
} }
public WorkerPipeClient(int connectTimeoutMilliseconds) public WorkerPipeClient(int connectTimeoutMilliseconds)
: this(
connectTimeoutMilliseconds,
(stream, frameOptions) => new WorkerPipeSession(stream, frameOptions))
{
}
public WorkerPipeClient(
int connectTimeoutMilliseconds,
Func<Stream, WorkerFrameProtocolOptions, WorkerPipeSession> sessionFactory)
{ {
if (connectTimeoutMilliseconds <= 0) if (connectTimeoutMilliseconds <= 0)
{ {
@@ -37,7 +26,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
"Worker pipe connect timeout must be greater than zero."); "Worker pipe connect timeout must be greater than zero.");
} }
_sessionFactory = sessionFactory ?? throw new ArgumentNullException(nameof(sessionFactory));
_connectTimeoutMilliseconds = connectTimeoutMilliseconds; _connectTimeoutMilliseconds = connectTimeoutMilliseconds;
} }
@@ -60,8 +48,8 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false); await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false);
WorkerPipeSession session = _sessionFactory(pipe, frameOptions); WorkerPipeSession session = new(pipe, frameOptions);
await session.RunAsync(cancellationToken).ConfigureAwait(false); await session.CompleteStartupHandshakeAsync(cancellationToken).ConfigureAwait(false);
} }
private Task ConnectAsync( private Task ConnectAsync(
+5 -329
View File
@@ -6,7 +6,6 @@ using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes; using Google.Protobuf.WellKnownTypes;
using MxGateway.Contracts.Proto; using MxGateway.Contracts.Proto;
using MxGateway.Worker.MxAccess; using MxGateway.Worker.MxAccess;
using MxGateway.Worker.Sta;
namespace MxGateway.Worker.Ipc; namespace MxGateway.Worker.Ipc;
@@ -14,14 +13,10 @@ public sealed class WorkerPipeSession
{ {
private readonly WorkerFrameProtocolOptions _options; private readonly WorkerFrameProtocolOptions _options;
private readonly Func<int> _processIdProvider; private readonly Func<int> _processIdProvider;
private readonly Func<IWorkerRuntimeSession> _runtimeSessionFactory;
private readonly WorkerPipeSessionOptions _sessionOptions;
private readonly WorkerFrameReader _reader; private readonly WorkerFrameReader _reader;
private readonly WorkerFrameWriter _writer; private readonly WorkerFrameWriter _writer;
private IWorkerRuntimeSession? _runtimeSession; private MxAccessStaSession? _mxAccessStaSession;
private long _nextSequence; private long _nextSequence;
private WorkerState _state = WorkerState.Starting;
private bool _watchdogFaultSent;
public WorkerPipeSession( public WorkerPipeSession(
Stream stream, Stream stream,
@@ -39,49 +34,11 @@ public sealed class WorkerPipeSession
WorkerFrameWriter writer, WorkerFrameWriter writer,
WorkerFrameProtocolOptions options, WorkerFrameProtocolOptions options,
Func<int> processIdProvider) Func<int> processIdProvider)
: this(
reader,
writer,
options,
processIdProvider,
new WorkerPipeSessionOptions(),
() => new MxAccessStaSession())
{
}
public WorkerPipeSession(
WorkerFrameReader reader,
WorkerFrameWriter writer,
WorkerFrameProtocolOptions options,
Func<int> processIdProvider,
WorkerPipeSessionOptions sessionOptions,
Func<IWorkerRuntimeSession> runtimeSessionFactory)
{ {
_reader = reader ?? throw new ArgumentNullException(nameof(reader)); _reader = reader ?? throw new ArgumentNullException(nameof(reader));
_writer = writer ?? throw new ArgumentNullException(nameof(writer)); _writer = writer ?? throw new ArgumentNullException(nameof(writer));
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
_processIdProvider = processIdProvider ?? throw new ArgumentNullException(nameof(processIdProvider)); _processIdProvider = processIdProvider ?? throw new ArgumentNullException(nameof(processIdProvider));
_sessionOptions = sessionOptions ?? throw new ArgumentNullException(nameof(sessionOptions));
_runtimeSessionFactory = runtimeSessionFactory ?? throw new ArgumentNullException(nameof(runtimeSessionFactory));
_sessionOptions.Validate();
}
public async Task RunAsync(CancellationToken cancellationToken = default)
{
_runtimeSession = _runtimeSessionFactory();
try
{
await CompleteStartupHandshakeAsync(
token => _runtimeSession.StartAsync(_processIdProvider(), token),
cancellationToken).ConfigureAwait(false);
await RunMessageLoopAsync(cancellationToken).ConfigureAwait(false);
}
finally
{
_runtimeSession?.Dispose();
_runtimeSession = null;
_state = WorkerState.Stopped;
}
} }
public Task CompleteStartupHandshakeAsync(CancellationToken cancellationToken = default) public Task CompleteStartupHandshakeAsync(CancellationToken cancellationToken = default)
@@ -119,14 +76,11 @@ public sealed class WorkerPipeSession
try try
{ {
WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false); WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
_state = WorkerState.Handshaking;
ValidateGatewayHello(envelope); ValidateGatewayHello(envelope);
await WriteWorkerHelloAsync(cancellationToken).ConfigureAwait(false); await WriteWorkerHelloAsync(cancellationToken).ConfigureAwait(false);
_state = WorkerState.InitializingSta;
WorkerReady ready = await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false); WorkerReady ready = await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false);
await WriteWorkerReadyAsync(ready, cancellationToken).ConfigureAwait(false); await WriteWorkerReadyAsync(ready, cancellationToken).ConfigureAwait(false);
_state = WorkerState.Ready;
} }
catch (WorkerFrameProtocolException exception) catch (WorkerFrameProtocolException exception)
{ {
@@ -186,174 +140,6 @@ public sealed class WorkerPipeSession
return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken); return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken);
} }
private async Task RunMessageLoopAsync(CancellationToken cancellationToken)
{
using CancellationTokenSource heartbeatCancellation = CancellationTokenSource
.CreateLinkedTokenSource(cancellationToken);
Task heartbeatTask = RunHeartbeatLoopAsync(heartbeatCancellation.Token);
try
{
while (!cancellationToken.IsCancellationRequested)
{
Task<WorkerEnvelope> readTask = _reader.ReadAsync(cancellationToken);
Task completedTask = await Task.WhenAny(readTask, heartbeatTask).ConfigureAwait(false);
if (completedTask == heartbeatTask)
{
await heartbeatTask.ConfigureAwait(false);
}
WorkerEnvelope envelope = await readTask.ConfigureAwait(false);
bool keepReading = await DispatchGatewayEnvelopeAsync(envelope, cancellationToken).ConfigureAwait(false);
if (!keepReading)
{
return;
}
}
}
finally
{
heartbeatCancellation.Cancel();
try
{
await heartbeatTask.ConfigureAwait(false);
}
catch (OperationCanceledException)
{
}
}
}
private async Task<bool> DispatchGatewayEnvelopeAsync(
WorkerEnvelope envelope,
CancellationToken cancellationToken)
{
switch (envelope.BodyCase)
{
case WorkerEnvelope.BodyOneofCase.WorkerCommand:
_ = ProcessCommandAsync(envelope, cancellationToken);
return true;
case WorkerEnvelope.BodyOneofCase.WorkerShutdown:
await ShutdownAsync(envelope.WorkerShutdown, cancellationToken).ConfigureAwait(false);
return false;
case WorkerEnvelope.BodyOneofCase.WorkerCancel:
return true;
default:
throw new WorkerFrameProtocolException(
WorkerFrameProtocolErrorCode.UnexpectedEnvelopeBody,
$"Worker received unexpected gateway envelope body {envelope.BodyCase}.");
}
}
private async Task ProcessCommandAsync(
WorkerEnvelope envelope,
CancellationToken cancellationToken)
{
IWorkerRuntimeSession runtimeSession = _runtimeSession
?? throw new InvalidOperationException("Worker runtime session has not been initialized.");
WorkerCommand workerCommand = envelope.WorkerCommand;
MxCommand command = workerCommand.Command;
StaCommand staCommand = new(
_options.SessionId,
envelope.CorrelationId,
command,
workerCommand.EnqueueTimestamp,
cancellationToken);
try
{
MxCommandReply reply = await runtimeSession.DispatchAsync(staCommand).ConfigureAwait(false);
await _writer
.WriteAsync(
CreateEnvelope(new WorkerCommandReply
{
Reply = reply,
CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
}),
cancellationToken)
.ConfigureAwait(false);
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
_state = WorkerState.Faulted;
await TryWriteFaultAsync(
CreateFault(
WorkerFaultCategory.MxaccessCommandFailed,
staCommand.MethodName,
exception),
cancellationToken).ConfigureAwait(false);
}
}
private async Task ShutdownAsync(
WorkerShutdown shutdown,
CancellationToken cancellationToken)
{
_state = WorkerState.ShuttingDown;
_runtimeSession?.RequestShutdown();
await _writer
.WriteAsync(
CreateEnvelope(
new WorkerShutdownAck
{
Status = new ProtocolStatus
{
Code = ProtocolStatusCode.Ok,
Message = string.IsNullOrWhiteSpace(shutdown.Reason)
? "Worker shutdown accepted."
: $"Worker shutdown accepted: {shutdown.Reason}",
},
}),
cancellationToken)
.ConfigureAwait(false);
}
private async Task RunHeartbeatLoopAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
await Task.Delay(_sessionOptions.HeartbeatInterval, cancellationToken).ConfigureAwait(false);
IWorkerRuntimeSession? runtimeSession = _runtimeSession;
if (runtimeSession is null)
{
continue;
}
WorkerRuntimeHeartbeatSnapshot snapshot = runtimeSession.CaptureHeartbeat();
await _writer
.WriteAsync(CreateEnvelope(CreateHeartbeat(snapshot)), cancellationToken)
.ConfigureAwait(false);
await ReportWatchdogFaultIfNeededAsync(snapshot, cancellationToken).ConfigureAwait(false);
}
}
private async Task ReportWatchdogFaultIfNeededAsync(
WorkerRuntimeHeartbeatSnapshot snapshot,
CancellationToken cancellationToken)
{
TimeSpan staleFor = DateTimeOffset.UtcNow - snapshot.LastStaActivityUtc;
if (staleFor <= _sessionOptions.HeartbeatGrace)
{
_watchdogFaultSent = false;
return;
}
if (_watchdogFaultSent)
{
return;
}
_watchdogFaultSent = true;
await TryWriteFaultAsync(
CreateFault(
WorkerFaultCategory.StaHung,
snapshot.CurrentCommandCorrelationId,
$"STA activity is stale by {staleFor}."),
cancellationToken).ConfigureAwait(false);
}
private async Task TryWriteFaultAsync( private async Task TryWriteFaultAsync(
WorkerFrameProtocolException exception, WorkerFrameProtocolException exception,
CancellationToken cancellationToken) CancellationToken cancellationToken)
@@ -392,25 +178,6 @@ public sealed class WorkerPipeSession
} }
} }
private async Task TryWriteFaultAsync(
WorkerFault fault,
CancellationToken cancellationToken)
{
try
{
await _writer
.WriteAsync(CreateEnvelope(fault), cancellationToken)
.ConfigureAwait(false);
}
catch (Exception faultWriteException) when (
faultWriteException is IOException
|| faultWriteException is ObjectDisposedException
|| faultWriteException is WorkerFrameProtocolException)
{
// The runtime fault remains observable through worker exit or pipe closure.
}
}
private WorkerEnvelope CreateEnvelope(WorkerHello hello) private WorkerEnvelope CreateEnvelope(WorkerHello hello)
{ {
return CreateBaseEnvelope(hello); return CreateBaseEnvelope(hello);
@@ -426,21 +193,6 @@ public sealed class WorkerPipeSession
return CreateBaseEnvelope(fault); return CreateBaseEnvelope(fault);
} }
private WorkerEnvelope CreateEnvelope(WorkerCommandReply reply)
{
return CreateBaseEnvelope(reply);
}
private WorkerEnvelope CreateEnvelope(WorkerShutdownAck shutdownAck)
{
return CreateBaseEnvelope(shutdownAck);
}
private WorkerEnvelope CreateEnvelope(WorkerHeartbeat heartbeat)
{
return CreateBaseEnvelope(heartbeat);
}
private WorkerEnvelope CreateBaseEnvelope(WorkerHello body) private WorkerEnvelope CreateBaseEnvelope(WorkerHello body)
{ {
WorkerEnvelope envelope = CreateBaseEnvelope(); WorkerEnvelope envelope = CreateBaseEnvelope();
@@ -462,28 +214,6 @@ public sealed class WorkerPipeSession
return envelope; return envelope;
} }
private WorkerEnvelope CreateBaseEnvelope(WorkerCommandReply body)
{
WorkerEnvelope envelope = CreateBaseEnvelope();
envelope.CorrelationId = body.Reply?.CorrelationId ?? string.Empty;
envelope.WorkerCommandReply = body;
return envelope;
}
private WorkerEnvelope CreateBaseEnvelope(WorkerShutdownAck body)
{
WorkerEnvelope envelope = CreateBaseEnvelope();
envelope.WorkerShutdownAck = body;
return envelope;
}
private WorkerEnvelope CreateBaseEnvelope(WorkerHeartbeat body)
{
WorkerEnvelope envelope = CreateBaseEnvelope();
envelope.WorkerHeartbeat = body;
return envelope;
}
private WorkerEnvelope CreateBaseEnvelope() private WorkerEnvelope CreateBaseEnvelope()
{ {
return new WorkerEnvelope return new WorkerEnvelope
@@ -501,39 +231,21 @@ public sealed class WorkerPipeSession
private async Task<WorkerReady> InitializeMxAccessAsync(CancellationToken cancellationToken) private async Task<WorkerReady> InitializeMxAccessAsync(CancellationToken cancellationToken)
{ {
_runtimeSession = new MxAccessStaSession(); _mxAccessStaSession = new MxAccessStaSession();
try try
{ {
return await _runtimeSession return await _mxAccessStaSession
.StartAsync(_processIdProvider(), cancellationToken) .StartAsync(_processIdProvider(), cancellationToken)
.ConfigureAwait(false); .ConfigureAwait(false);
} }
catch catch
{ {
_runtimeSession.Dispose(); _mxAccessStaSession.Dispose();
_runtimeSession = null; _mxAccessStaSession = null;
throw; throw;
} }
} }
private WorkerHeartbeat CreateHeartbeat(WorkerRuntimeHeartbeatSnapshot snapshot)
{
WorkerState state = string.IsNullOrWhiteSpace(snapshot.CurrentCommandCorrelationId)
? _state
: WorkerState.ExecutingCommand;
return new WorkerHeartbeat
{
WorkerProcessId = _processIdProvider(),
State = state,
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(snapshot.LastStaActivityUtc),
PendingCommandCount = snapshot.PendingCommandCount,
OutboundEventQueueDepth = snapshot.OutboundEventQueueDepth,
LastEventSequence = snapshot.LastEventSequence,
CurrentCommandCorrelationId = snapshot.CurrentCommandCorrelationId,
};
}
private WorkerReady CreateWorkerReady() private WorkerReady CreateWorkerReady()
{ {
return new WorkerReady return new WorkerReady
@@ -583,42 +295,6 @@ public sealed class WorkerPipeSession
return fault; return fault;
} }
private static WorkerFault CreateFault(
WorkerFaultCategory category,
string commandMethod,
Exception exception)
{
WorkerFault fault = CreateFault(
category,
commandMethod,
exception.Message);
fault.ExceptionType = exception.GetType().FullName ?? string.Empty;
fault.ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.WorkerUnavailable,
Message = exception.Message,
};
return fault;
}
private static WorkerFault CreateFault(
WorkerFaultCategory category,
string commandMethod,
string diagnosticMessage)
{
return new WorkerFault
{
Category = category,
CommandMethod = commandMethod ?? string.Empty,
DiagnosticMessage = diagnosticMessage,
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.WorkerUnavailable,
Message = diagnosticMessage,
},
};
}
private static WorkerFaultCategory MapFaultCategory(WorkerFrameProtocolErrorCode errorCode) private static WorkerFaultCategory MapFaultCategory(WorkerFrameProtocolErrorCode errorCode)
{ {
return errorCode switch return errorCode switch
@@ -1,36 +0,0 @@
using System;
namespace MxGateway.Worker.Ipc;
public sealed class WorkerPipeSessionOptions
{
public static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(5);
public static readonly TimeSpan DefaultHeartbeatGrace = TimeSpan.FromSeconds(15);
public WorkerPipeSessionOptions()
{
HeartbeatInterval = DefaultHeartbeatInterval;
HeartbeatGrace = DefaultHeartbeatGrace;
}
public TimeSpan HeartbeatInterval { get; set; }
public TimeSpan HeartbeatGrace { get; set; }
public void Validate()
{
if (HeartbeatInterval <= TimeSpan.Zero)
{
throw new ArgumentOutOfRangeException(
nameof(HeartbeatInterval),
"Worker heartbeat interval must be greater than zero.");
}
if (HeartbeatGrace <= TimeSpan.Zero)
{
throw new ArgumentOutOfRangeException(
nameof(HeartbeatGrace),
"Worker heartbeat grace must be greater than zero.");
}
}
}
@@ -1,20 +0,0 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MxGateway.Contracts.Proto;
using MxGateway.Worker.Sta;
namespace MxGateway.Worker.MxAccess;
public interface IWorkerRuntimeSession : IDisposable
{
Task<WorkerReady> StartAsync(
int workerProcessId,
CancellationToken cancellationToken = default);
Task<MxCommandReply> DispatchAsync(StaCommand command);
WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat();
void RequestShutdown();
}
@@ -7,7 +7,7 @@ using MxGateway.Worker.Sta;
namespace MxGateway.Worker.MxAccess; namespace MxGateway.Worker.MxAccess;
public sealed class MxAccessStaSession : IWorkerRuntimeSession public sealed class MxAccessStaSession : IDisposable
{ {
private readonly IMxAccessComObjectFactory factory; private readonly IMxAccessComObjectFactory factory;
private readonly IMxAccessEventSink eventSink; private readonly IMxAccessEventSink eventSink;
@@ -68,30 +68,6 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
return commandDispatcher.DispatchAsync(command); return commandDispatcher.DispatchAsync(command);
} }
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
{
uint pendingCommandCount = 0;
string currentCommandCorrelationId = string.Empty;
if (commandDispatcher is not null)
{
pendingCommandCount = (uint)commandDispatcher.PendingCommandCount;
currentCommandCorrelationId = commandDispatcher.CurrentCommandCorrelationId;
}
return new WorkerRuntimeHeartbeatSnapshot(
staRuntime.LastActivityUtc,
pendingCommandCount,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId);
}
public void RequestShutdown()
{
commandDispatcher?.RequestShutdown();
}
public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync( public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync(
CancellationToken cancellationToken = default) CancellationToken cancellationToken = default)
{ {
@@ -125,7 +101,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
return; return;
} }
RequestShutdown(); commandDispatcher?.RequestShutdown();
if (session is not null) if (session is not null)
{ {
@@ -1,30 +0,0 @@
using System;
namespace MxGateway.Worker.MxAccess;
public sealed class WorkerRuntimeHeartbeatSnapshot
{
public WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset lastStaActivityUtc,
uint pendingCommandCount,
uint outboundEventQueueDepth,
ulong lastEventSequence,
string currentCommandCorrelationId)
{
LastStaActivityUtc = lastStaActivityUtc;
PendingCommandCount = pendingCommandCount;
OutboundEventQueueDepth = outboundEventQueueDepth;
LastEventSequence = lastEventSequence;
CurrentCommandCorrelationId = currentCommandCorrelationId ?? string.Empty;
}
public DateTimeOffset LastStaActivityUtc { get; }
public uint PendingCommandCount { get; }
public uint OutboundEventQueueDepth { get; }
public ulong LastEventSequence { get; }
public string CurrentCommandCorrelationId { get; }
}
+1 -1
View File
@@ -84,7 +84,7 @@ public static class WorkerApplication
pipeClient.RunAsync(options).GetAwaiter().GetResult(); pipeClient.RunAsync(options).GetAwaiter().GetResult();
logger.Information("WorkerPipeSessionCompleted", new Dictionary<string, object?> logger.Information("WorkerPipeHandshakeSucceeded", new Dictionary<string, object?>
{ {
["session_id"] = options.SessionId, ["session_id"] = options.SessionId,
["pipe_name"] = options.PipeName, ["pipe_name"] = options.PipeName,