Resolve Server-007..014 code-review findings
Server-007: GalaxyHierarchyProjector re-filtered the whole hierarchy per page (O(total) paging). It now memoizes the filtered list per cache-entry + filter signature so subsequent pages are an O(pageSize) slice. Server-008: WatchDeployEvents re-resolved browse subtrees and rebuilt globs per streamed event. ResolveBrowseSubtrees is hoisted out of the loop and GalaxyGlobMatcher caches compiled Regex instances per pattern. Server-009: auth-store connections used no busy timeout or WAL. A new OpenConnectionAsync applies journal_mode=WAL and a busy_timeout; all auth call sites use it. docs/Authentication.md updated. Server-010: the dashboard rendered Rotate/Revoke for revoked keys, where Rotate silently reactivates them. ApiKeysPage now shows actions only for Active keys. docs/Authentication.md updated. Server-011: WorkerAlarmRpcDispatcher converted to a primary constructor and brought in line with module conventions. Server-012: CLAUDE.md corrected to the canonical *:* scope strings. Server-013 (partly re-triaged): three named coverage gaps were already closed; the genuine gap (WorkerExecutableValidator) is now covered. Server-014: rewrote stale "alarm path not yet wired" comments in MxAccessGatewayService to describe the production WorkerAlarmRpcDispatcher. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -165,19 +165,26 @@ else
|
||||
{
|
||||
<td>
|
||||
<div class="btn-group btn-group-sm" role="group" aria-label="API key actions">
|
||||
<button type="button" class="btn btn-outline-secondary"
|
||||
disabled="@IsBusy"
|
||||
@onclick="() => RotateApiKeyAsync(key.KeyId)">
|
||||
Rotate
|
||||
</button>
|
||||
@if (key.RevokedUtc is null)
|
||||
{
|
||||
@* Rotate clears revoked_utc, which would silently reactivate a
|
||||
deliberately revoked key. Only offer it for active keys so a
|
||||
revoked key is not un-revoked as a side effect of rotation. *@
|
||||
<button type="button" class="btn btn-outline-secondary"
|
||||
disabled="@IsBusy"
|
||||
@onclick="() => RotateApiKeyAsync(key.KeyId)">
|
||||
Rotate
|
||||
</button>
|
||||
<button type="button" class="btn btn-outline-danger"
|
||||
disabled="@IsBusy"
|
||||
@onclick="() => RevokeApiKeyAsync(key.KeyId)">
|
||||
Revoke
|
||||
</button>
|
||||
}
|
||||
else
|
||||
{
|
||||
<span class="text-muted small">No actions</span>
|
||||
}
|
||||
</div>
|
||||
</td>
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Text;
|
||||
using System.Text.RegularExpressions;
|
||||
|
||||
@@ -5,6 +6,14 @@ namespace MxGateway.Server.Galaxy;
|
||||
|
||||
public static class GalaxyGlobMatcher
|
||||
{
|
||||
/// <summary>
|
||||
/// Compiled-regex cache keyed by glob pattern. <c>IsMatch</c> is called once per
|
||||
/// object per <c>DiscoverHierarchy</c>/<c>WatchDeployEvents</c> evaluation, so the
|
||||
/// same handful of glob patterns are translated repeatedly; caching avoids
|
||||
/// rebuilding and recompiling the regex on every call.
|
||||
/// </summary>
|
||||
private static readonly ConcurrentDictionary<string, Regex> RegexCache = new(StringComparer.Ordinal);
|
||||
|
||||
public static bool IsMatch(string value, string glob)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(glob))
|
||||
@@ -12,11 +21,15 @@ public static class GalaxyGlobMatcher
|
||||
return true;
|
||||
}
|
||||
|
||||
return Regex.IsMatch(
|
||||
value ?? string.Empty,
|
||||
BuildRegex(glob),
|
||||
RegexOptions.CultureInvariant | RegexOptions.IgnoreCase,
|
||||
TimeSpan.FromMilliseconds(100));
|
||||
return GetOrCreateRegex(glob).IsMatch(value ?? string.Empty);
|
||||
}
|
||||
|
||||
private static Regex GetOrCreateRegex(string glob)
|
||||
{
|
||||
return RegexCache.GetOrAdd(glob, static pattern => new Regex(
|
||||
BuildRegex(pattern),
|
||||
RegexOptions.CultureInvariant | RegexOptions.IgnoreCase | RegexOptions.Compiled,
|
||||
TimeSpan.FromMilliseconds(100)));
|
||||
}
|
||||
|
||||
private static string BuildRegex(string glob)
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using Grpc.Core;
|
||||
@@ -7,6 +9,18 @@ namespace MxGateway.Server.Galaxy;
|
||||
|
||||
public static class GalaxyHierarchyProjector
|
||||
{
|
||||
/// <summary>
|
||||
/// Per-cache-entry memo of filtered, ordered <see cref="GalaxyObjectView"/> lists
|
||||
/// keyed by filter signature. Without it, paging through a large hierarchy
|
||||
/// re-applies every filter and re-scans the full <see cref="GalaxyHierarchyIndex.ObjectViews"/>
|
||||
/// collection on every page — O(total) per page, O(total²/pageSize) end-to-end.
|
||||
/// With it, the first page builds the filtered list and each subsequent page is an
|
||||
/// O(pageSize) slice. The table is keyed on the immutable cache-entry instance, so
|
||||
/// when the cache publishes a new entry the stale memo becomes unreachable and is
|
||||
/// reclaimed with it — no explicit invalidation needed.
|
||||
/// </summary>
|
||||
private static readonly ConditionalWeakTable<GalaxyHierarchyCacheEntry, ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>>> FilteredViewCache = new();
|
||||
|
||||
public static GalaxyHierarchyQueryResult Project(
|
||||
GalaxyHierarchyCacheEntry entry,
|
||||
DiscoverHierarchyRequest request,
|
||||
@@ -39,8 +53,6 @@ public static class GalaxyHierarchyProjector
|
||||
throw new ArgumentOutOfRangeException(nameof(pageSize), pageSize, "Page size must be greater than zero.");
|
||||
}
|
||||
|
||||
IReadOnlyList<GalaxyObjectView> views = entry.Index.ObjectViews;
|
||||
GalaxyObjectView? root = ResolveRoot(request, views);
|
||||
int? maxDepth = request.MaxDepth;
|
||||
if (maxDepth < 0)
|
||||
{
|
||||
@@ -49,30 +61,61 @@ public static class GalaxyHierarchyProjector
|
||||
"DiscoverHierarchy max_depth must be greater than or equal to zero when provided."));
|
||||
}
|
||||
|
||||
List<GalaxyObject> page = [];
|
||||
int matchedCount = 0;
|
||||
string filterSignature = ComputeFilterSignature(request, browseSubtreeGlobs);
|
||||
IReadOnlyList<GalaxyObjectView> matchedViews = GetFilteredViews(
|
||||
entry,
|
||||
request,
|
||||
browseSubtreeGlobs,
|
||||
maxDepth,
|
||||
filterSignature);
|
||||
|
||||
bool includeAttributes = IncludeAttributes(request);
|
||||
foreach (GalaxyObjectView view in views)
|
||||
List<GalaxyObject> page = new(Math.Min(pageSize, Math.Max(0, matchedViews.Count - offset)));
|
||||
int end = (int)Math.Min((long)offset + pageSize, matchedViews.Count);
|
||||
for (int index = offset; index < end; index++)
|
||||
{
|
||||
if (!MatchesRoot(view, root, maxDepth)
|
||||
|| !MatchesBrowseSubtrees(view, browseSubtreeGlobs)
|
||||
|| !MatchesFilters(view.Object, request))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (matchedCount >= offset && page.Count < pageSize)
|
||||
{
|
||||
page.Add(CloneObject(view.Object, includeAttributes));
|
||||
}
|
||||
|
||||
matchedCount++;
|
||||
page.Add(CloneObject(matchedViews[index].Object, includeAttributes));
|
||||
}
|
||||
|
||||
return new GalaxyHierarchyQueryResult(
|
||||
page,
|
||||
matchedCount,
|
||||
ComputeFilterSignature(request, browseSubtreeGlobs));
|
||||
matchedViews.Count,
|
||||
filterSignature);
|
||||
}
|
||||
|
||||
private static IReadOnlyList<GalaxyObjectView> GetFilteredViews(
|
||||
GalaxyHierarchyCacheEntry entry,
|
||||
DiscoverHierarchyRequest request,
|
||||
IReadOnlyList<string>? browseSubtreeGlobs,
|
||||
int? maxDepth,
|
||||
string filterSignature)
|
||||
{
|
||||
// ResolveRoot can throw RpcException(NotFound); run it before consulting the
|
||||
// memo so a bad root surfaces consistently regardless of cache state.
|
||||
IReadOnlyList<GalaxyObjectView> views = entry.Index.ObjectViews;
|
||||
GalaxyObjectView? root = ResolveRoot(request, views);
|
||||
|
||||
ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>> memo =
|
||||
FilteredViewCache.GetValue(entry, static _ => new ConcurrentDictionary<string, IReadOnlyList<GalaxyObjectView>>(StringComparer.Ordinal));
|
||||
|
||||
return memo.GetOrAdd(
|
||||
filterSignature,
|
||||
static (_, state) =>
|
||||
{
|
||||
List<GalaxyObjectView> matched = [];
|
||||
foreach (GalaxyObjectView view in state.Views)
|
||||
{
|
||||
if (MatchesRoot(view, state.Root, state.MaxDepth)
|
||||
&& MatchesBrowseSubtrees(view, state.BrowseSubtreeGlobs)
|
||||
&& MatchesFilters(view.Object, state.Request))
|
||||
{
|
||||
matched.Add(view);
|
||||
}
|
||||
}
|
||||
|
||||
return matched;
|
||||
},
|
||||
(Views: views, Root: root, MaxDepth: maxDepth, BrowseSubtreeGlobs: browseSubtreeGlobs, Request: request));
|
||||
}
|
||||
|
||||
public static GalaxyObject? FindObjectForTag(
|
||||
|
||||
@@ -115,6 +115,11 @@ public sealed class GalaxyRepositoryGrpcService(
|
||||
{
|
||||
DateTimeOffset? lastSeen = request.LastSeenDeployTime?.ToDateTimeOffset();
|
||||
|
||||
// The caller's identity (and therefore its browse-subtree constraints) is fixed
|
||||
// for the lifetime of the stream, so resolve the subtrees once rather than per
|
||||
// streamed event.
|
||||
IReadOnlyList<string> browseSubtrees = ResolveBrowseSubtrees();
|
||||
|
||||
await foreach (GalaxyDb.GalaxyDeployEventInfo info in notifier
|
||||
.SubscribeAsync(context.CancellationToken)
|
||||
.ConfigureAwait(false))
|
||||
@@ -129,7 +134,7 @@ public sealed class GalaxyRepositoryGrpcService(
|
||||
}
|
||||
lastSeen = null;
|
||||
|
||||
await responseStream.WriteAsync(MapDeployEvent(info, ResolveBrowseSubtrees()), context.CancellationToken).ConfigureAwait(false);
|
||||
await responseStream.WriteAsync(MapDeployEvent(info, browseSubtrees), context.CancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -161,13 +161,14 @@ public sealed class MxAccessGatewayService(
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// PR A.3 — surfaces the public AcknowledgeAlarm RPC. The gateway resolves the
|
||||
/// session and returns a successful reply; the actual worker-side ack call ships
|
||||
/// in <c>PR A.2</c> which adds the MxAccess alarm subscription + worker command
|
||||
/// handler. Clients calling this method today receive an OK reply with a
|
||||
/// "worker alarm path not yet wired" diagnostic — no PERMISSION_DENIED, no
|
||||
/// UNIMPLEMENTED, so the .NET / Python / Go / Java / Rust SDK call sites land
|
||||
/// on a stable surface.
|
||||
/// Surfaces the public AcknowledgeAlarm RPC. The gateway validates the request,
|
||||
/// resolves the session, and delegates to the registered
|
||||
/// <see cref="IAlarmRpcDispatcher"/>. DI binds the production
|
||||
/// <see cref="MxGateway.Server.Sessions.WorkerAlarmRpcDispatcher"/>, which routes
|
||||
/// the ack through the worker pipe IPC: an <c>alarm_full_reference</c> that parses
|
||||
/// as a canonical GUID forwards to <c>AcknowledgeAlarmCommand</c>; a
|
||||
/// <c>Provider!Group.Tag</c> reference forwards to <c>AcknowledgeAlarmByNameCommand</c>;
|
||||
/// anything else returns an <c>InvalidRequest</c> diagnostic.
|
||||
/// </remarks>
|
||||
public override async Task<AcknowledgeAlarmReply> AcknowledgeAlarm(
|
||||
AcknowledgeAlarmRequest request,
|
||||
@@ -189,11 +190,11 @@ public sealed class MxAccessGatewayService(
|
||||
// gRPC NotFound by the caller's MapException.
|
||||
_ = ResolveSession(request.SessionId);
|
||||
|
||||
// PR A.6 — delegate to the alarm dispatcher. NotWiredAlarmRpcDispatcher
|
||||
// (default) returns OK + a worker-pending diagnostic. Production
|
||||
// WorkerAlarmRpcDispatcher (dev-rig follow-up) routes through the
|
||||
// worker IPC to AlarmClient.AlarmAckByGUID with full operator-identity
|
||||
// fidelity.
|
||||
// Delegate to the registered alarm dispatcher. DI binds the production
|
||||
// WorkerAlarmRpcDispatcher, which routes the ack over the worker IPC by
|
||||
// GUID (AcknowledgeAlarmCommand) or by Provider!Group.Tag reference
|
||||
// (AcknowledgeAlarmByNameCommand). NotWiredAlarmRpcDispatcher is only the
|
||||
// null fallback used when no dispatcher is registered.
|
||||
return await alarmRpcDispatcher.AcknowledgeAsync(request, context.CancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
@@ -205,12 +206,12 @@ public sealed class MxAccessGatewayService(
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// PR A.3 — surfaces the public QueryActiveAlarms RPC as an empty stream until
|
||||
/// PR A.2 adds the worker-side QueryActiveAlarmsCommand that walks the
|
||||
/// MxAccess active-alarm collection. Clients can call the RPC and iterate the
|
||||
/// stream; today the stream completes immediately. Once A.2 ships, this
|
||||
/// handler will translate the request into a WorkerCommand and stream the
|
||||
/// resulting snapshots.
|
||||
/// Surfaces the public QueryActiveAlarms RPC. The gateway validates the request,
|
||||
/// resolves the session, and delegates to the registered
|
||||
/// <see cref="IAlarmRpcDispatcher"/>. DI binds the production
|
||||
/// <see cref="MxGateway.Server.Sessions.WorkerAlarmRpcDispatcher"/>, which issues a
|
||||
/// <c>QueryActiveAlarmsCommand</c> over the worker pipe IPC and streams each
|
||||
/// <c>ActiveAlarmSnapshot</c> from the worker reply.
|
||||
/// </remarks>
|
||||
public override async Task QueryActiveAlarms(
|
||||
QueryActiveAlarmsRequest request,
|
||||
@@ -226,11 +227,11 @@ public sealed class MxAccessGatewayService(
|
||||
}
|
||||
_ = ResolveSession(request.SessionId);
|
||||
|
||||
// PR A.7 — delegate to the alarm dispatcher. NotWiredAlarmRpcDispatcher
|
||||
// (default) yields an empty stream. Production WorkerAlarmRpcDispatcher
|
||||
// (dev-rig follow-up) walks the worker's IMxAccessAlarmConsumer
|
||||
// SnapshotActiveAlarms output and translates each AlarmRecord into an
|
||||
// ActiveAlarmSnapshot.
|
||||
// Delegate to the registered alarm dispatcher. DI binds the production
|
||||
// WorkerAlarmRpcDispatcher, which issues a QueryActiveAlarmsCommand over the
|
||||
// worker IPC and streams each ActiveAlarmSnapshot from the worker reply.
|
||||
// NotWiredAlarmRpcDispatcher is only the null fallback used when no
|
||||
// dispatcher is registered.
|
||||
await foreach (ActiveAlarmSnapshot snapshot in alarmRpcDispatcher
|
||||
.QueryActiveAlarmsAsync(request, context.CancellationToken)
|
||||
.WithCancellation(context.CancellationToken)
|
||||
|
||||
@@ -10,7 +10,17 @@ namespace MxGateway.Server.Security.Authentication;
|
||||
public sealed class AuthSqliteConnectionFactory(IOptions<GatewayOptions> options)
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates and configures a SQLite connection to the auth database.
|
||||
/// Busy timeout applied to every auth-store connection. SQLite retries a busy
|
||||
/// database for this long before surfacing <c>SQLITE_BUSY</c>, so the concurrent
|
||||
/// <c>MarkKeyUsedAsync</c> / audit-append writers degrade gracefully under load
|
||||
/// instead of failing the request path.
|
||||
/// </summary>
|
||||
private static readonly TimeSpan BusyTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
/// <summary>
|
||||
/// Creates an unopened SQLite connection to the auth database. Prefer
|
||||
/// <see cref="OpenConnectionAsync"/>, which also applies WAL journaling and the
|
||||
/// busy timeout.
|
||||
/// </summary>
|
||||
public SqliteConnection CreateConnection()
|
||||
{
|
||||
@@ -25,9 +35,44 @@ public sealed class AuthSqliteConnectionFactory(IOptions<GatewayOptions> options
|
||||
SqliteConnectionStringBuilder builder = new()
|
||||
{
|
||||
DataSource = sqlitePath,
|
||||
Mode = SqliteOpenMode.ReadWriteCreate
|
||||
Mode = SqliteOpenMode.ReadWriteCreate,
|
||||
Pooling = true,
|
||||
DefaultTimeout = (int)BusyTimeout.TotalSeconds,
|
||||
};
|
||||
|
||||
return new SqliteConnection(builder.ToString());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a SQLite connection, opens it, and configures WAL journaling and a
|
||||
/// non-zero busy timeout so concurrent readers and writers degrade gracefully
|
||||
/// rather than surfacing <c>SQLITE_BUSY</c> as a hard failure.
|
||||
/// </summary>
|
||||
public async Task<SqliteConnection> OpenConnectionAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
SqliteConnection connection = CreateConnection();
|
||||
try
|
||||
{
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await ConfigureConnectionAsync(connection, cancellationToken).ConfigureAwait(false);
|
||||
return connection;
|
||||
}
|
||||
catch
|
||||
{
|
||||
await connection.DisposeAsync().ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task ConfigureConnectionAsync(
|
||||
SqliteConnection connection,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// WAL is a persistent, database-level setting; re-applying it per connection
|
||||
// is cheap and a no-op once set. busy_timeout is per-connection state.
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText =
|
||||
$"PRAGMA journal_mode=WAL; PRAGMA busy_timeout={(int)BusyTimeout.TotalMilliseconds};";
|
||||
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,8 +10,7 @@ public sealed class SqliteApiKeyAdminStore(AuthSqliteConnectionFactory connectio
|
||||
/// <inheritdoc />
|
||||
public async Task CreateAsync(ApiKeyCreateRequest request, CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
@@ -44,8 +43,7 @@ public sealed class SqliteApiKeyAdminStore(AuthSqliteConnectionFactory connectio
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyList<ApiKeyRecord>> ListAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
@@ -70,8 +68,7 @@ public sealed class SqliteApiKeyAdminStore(AuthSqliteConnectionFactory connectio
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> RevokeAsync(string keyId, DateTimeOffset revokedUtc, CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
@@ -94,8 +91,7 @@ public sealed class SqliteApiKeyAdminStore(AuthSqliteConnectionFactory connectio
|
||||
DateTimeOffset rotatedUtc,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
|
||||
@@ -7,8 +7,7 @@ public sealed class SqliteApiKeyAuditStore(AuthSqliteConnectionFactory connectio
|
||||
/// <inheritdoc />
|
||||
public async Task AppendAsync(ApiKeyAuditEntry entry, CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
@@ -32,8 +31,7 @@ public sealed class SqliteApiKeyAuditStore(AuthSqliteConnectionFactory connectio
|
||||
return [];
|
||||
}
|
||||
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
|
||||
@@ -20,8 +20,7 @@ public sealed class SqliteApiKeyStore(AuthSqliteConnectionFactory connectionFact
|
||||
/// <inheritdoc />
|
||||
public async Task MarkKeyUsedAsync(string keyId, DateTimeOffset usedUtc, CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = """
|
||||
@@ -40,8 +39,7 @@ public sealed class SqliteApiKeyStore(AuthSqliteConnectionFactory connectionFact
|
||||
bool requireActive,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteCommand command = connection.CreateCommand();
|
||||
command.CommandText = requireActive
|
||||
|
||||
@@ -8,8 +8,7 @@ public sealed class SqliteAuthStoreMigrator(AuthSqliteConnectionFactory connecti
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
public async Task MigrateAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await using SqliteConnection connection = connectionFactory.CreateConnection();
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
await using SqliteConnection connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await using SqliteTransaction transaction =
|
||||
(SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using MxGateway.Server.Grpc;
|
||||
@@ -11,39 +8,33 @@ namespace MxGateway.Server.Sessions;
|
||||
/// <summary>
|
||||
/// Production <see cref="IAlarmRpcDispatcher"/> that routes the public
|
||||
/// <c>AcknowledgeAlarm</c> + <c>QueryActiveAlarms</c> RPCs through the
|
||||
/// worker pipe IPC. Replaces <see cref="NotWiredAlarmRpcDispatcher"/>
|
||||
/// once the worker AlarmCommandHandler is wired in.
|
||||
/// worker pipe IPC. DI binds this dispatcher; <see cref="NotWiredAlarmRpcDispatcher"/>
|
||||
/// is only the null fallback used when no dispatcher is registered.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <c>QueryActiveAlarms</c> is fully wired: issues a
|
||||
/// <c>QueryActiveAlarms</c> issues a
|
||||
/// <see cref="QueryActiveAlarmsCommand"/> over the pipe and yields
|
||||
/// each <see cref="ActiveAlarmSnapshot"/> from the
|
||||
/// <see cref="QueryActiveAlarmsReplyPayload"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <c>AcknowledgeAlarm</c> is partially wired: the public RPC's
|
||||
/// <see cref="AcknowledgeAlarmRequest.AlarmFullReference"/> is a
|
||||
/// <c>Provider!Group.Tag</c> string, but the worker's wnwrap consumer
|
||||
/// acks by GUID. When the supplied reference parses as a GUID
|
||||
/// directly, the dispatcher forwards it as-is. Otherwise it
|
||||
/// returns an <c>Unimplemented</c> diagnostic. Resolving
|
||||
/// reference→GUID requires an additional worker IPC command
|
||||
/// (e.g. <c>AlarmAckByName</c> wrapping
|
||||
/// <c>wwAlarmConsumerClass.AlarmAckByName</c>) and is tracked as
|
||||
/// a follow-up.
|
||||
/// <c>AcknowledgeAlarm</c> accepts either form of
|
||||
/// <see cref="AcknowledgeAlarmRequest.AlarmFullReference"/>: a canonical
|
||||
/// GUID forwards as an <see cref="AcknowledgeAlarmCommand"/>; a
|
||||
/// <c>Provider!Group.Tag</c> reference is parsed by
|
||||
/// <see cref="TryParseAlarmReference"/> and forwarded as an
|
||||
/// <see cref="AcknowledgeAlarmByNameCommand"/>. Any other reference
|
||||
/// returns an <c>InvalidRequest</c> diagnostic.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class WorkerAlarmRpcDispatcher : IAlarmRpcDispatcher
|
||||
public sealed class WorkerAlarmRpcDispatcher(
|
||||
ISessionRegistry sessionRegistry,
|
||||
TimeProvider? timeProvider = null) : IAlarmRpcDispatcher
|
||||
{
|
||||
private readonly ISessionRegistry sessionRegistry;
|
||||
private readonly TimeProvider timeProvider;
|
||||
|
||||
public WorkerAlarmRpcDispatcher(ISessionRegistry sessionRegistry, TimeProvider? timeProvider = null)
|
||||
{
|
||||
this.sessionRegistry = sessionRegistry ?? throw new System.ArgumentNullException(nameof(sessionRegistry));
|
||||
this.timeProvider = timeProvider ?? TimeProvider.System;
|
||||
}
|
||||
private readonly ISessionRegistry sessionRegistry = sessionRegistry
|
||||
?? throw new ArgumentNullException(nameof(sessionRegistry));
|
||||
private readonly TimeProvider timeProvider = timeProvider ?? TimeProvider.System;
|
||||
|
||||
/// <summary>
|
||||
/// Parse a full alarm reference of the form <c>Provider!Group.Tag</c>
|
||||
@@ -83,7 +74,7 @@ public sealed class WorkerAlarmRpcDispatcher : IAlarmRpcDispatcher
|
||||
AcknowledgeAlarmRequest request,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (request is null) throw new System.ArgumentNullException(nameof(request));
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
|
||||
if (!sessionRegistry.TryGet(request.SessionId, out GatewaySession session))
|
||||
{
|
||||
@@ -98,7 +89,7 @@ public sealed class WorkerAlarmRpcDispatcher : IAlarmRpcDispatcher
|
||||
}
|
||||
|
||||
WorkerCommand workerCommand;
|
||||
if (System.Guid.TryParse(request.AlarmFullReference, out System.Guid guid))
|
||||
if (Guid.TryParse(request.AlarmFullReference, out Guid guid))
|
||||
{
|
||||
workerCommand = new WorkerCommand
|
||||
{
|
||||
@@ -193,7 +184,7 @@ public sealed class WorkerAlarmRpcDispatcher : IAlarmRpcDispatcher
|
||||
QueryActiveAlarmsRequest request,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
if (request is null) throw new System.ArgumentNullException(nameof(request));
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
|
||||
if (!sessionRegistry.TryGet(request.SessionId, out GatewaySession session))
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user