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

Resolve Tests-003, -004, -005, -006 code-review findings

Browse Source 
Tests-003: temp auth-DB directories leaked under %TEMP%. Added the
TempDatabaseDirectory IDisposable helper (clears the Sqlite connection pool,
then recursively deletes); SqliteAuthStoreTests and ApiKeyAdminCliRunnerTests
now dispose every directory they create.

Tests-004: added end-to-end coverage composing the real authorization
interceptor in front of the real MxAccessGatewayService, plus scope-resolver
tests confirming an unmapped request type fails closed to the admin scope.

Tests-005: added coverage for a worker faulting mid-command — a pipe
disconnect and a worker fault while an InvokeAsync is in flight both fail the
pending invoke. No product change needed.

Tests-006 (re-triaged): the flaky ReadLoop_WhenClientFaults_KillsOwnedWorkerProcess
is a test race, not a product bug — the kill runs synchronously inside
SetFaulted. Rewrote it to await FakeWorkerProcess exit deterministically, and
replaced fixed Task.Delay timing in the late-reply and heartbeat tests with
FIFO ordering and an injected ManualTimeProvider.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-05-18 21:44:55 -04:00
parent 98f9b7792b
commit 5ade3f4f48
8 changed files with 539 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs
+119 -9
View File
@@ -71,9 +71,11 @@ public sealed class WorkerClientTests
async () => await timedOutInvokeTask);
Assert.Equal(WorkerClientErrorCode.CommandTimeout, exception.ErrorCode);
// Send the stale reply for the already-timed-out command, then the second
// command's reply. The pipe is FIFO, so the read loop processes (and discards)
// the stale reply before the second reply — no fixed Task.Delay needed.
await pipePair.WorkerWriter.WriteAsync(
CreateCommandReplyEnvelope(timedOutCommand.CorrelationId, MxCommandKind.Ping));
await Task.Delay(TimeSpan.FromMilliseconds(50));
Task<WorkerCommandReply> secondInvokeTask = client.InvokeAsync(
CreateCommand(MxCommandKind.GetWorkerInfo),
@@ -142,7 +144,14 @@ public sealed class WorkerClientTests
Assert.Equal(WorkerClientState.Faulted, client.State);
}
/// <summary>Verifies that the read loop faults the client when the pipe disconnects.</summary>
/// <summary>
/// Verifies that when the client faults it kills the owned worker process.
/// The assertion waits on <see cref="FakeWorkerProcess.WaitForExitAsync"/>, which
/// completes exactly when <c>Kill</c> runs, instead of polling <c>client.State</c>.
/// Polling state is racy: <see cref="WorkerClient.SetFaulted"/> publishes the
/// <c>Faulted</c> state before it calls <c>KillOwnedProcess</c>, so a state-based
/// wait can observe <c>Faulted</c> while <c>KillCount</c> is still 0.
/// </summary>
[Fact]
public async Task ReadLoop_WhenClientFaults_KillsOwnedWorkerProcess()
{
@@ -164,15 +173,77 @@ public sealed class WorkerClientTests
await pipePair.WorkerWriter.WriteAsync(
CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));
await WaitUntilAsync(
() => client.State == WorkerClientState.Faulted,
TestTimeout);
// Deterministic: this completes the instant Kill() runs, with no timing window.
using CancellationTokenSource exitTimeout = new(TestTimeout);
await process.WaitForExitAsync(exitTimeout.Token);
Assert.Equal(WorkerClientState.Faulted, client.State);
Assert.Equal(1, process.KillCount);
Assert.True(process.KillEntireProcessTree);
Assert.True(process.HasExited);
}
/// <summary>
/// Verifies that a worker faulting mid-command — the pipe dropping while an
/// <see cref="WorkerClient.InvokeAsync"/> is still pending — completes the pending
/// invoke task with a <see cref="WorkerClientException"/> carrying the
/// pipe-disconnected error code rather than hanging until the command timeout.
/// </summary>
[Fact]
public async Task InvokeAsync_WhenPipeDisconnectsMidCommand_FailsPendingInvokeWithPipeDisconnected()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
CreateCommand(MxCommandKind.Ping),
TestTimeout,
CancellationToken.None);
// The worker received the command but disconnects before replying.
WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
await pipePair.DisposeWorkerSideAsync();
WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
async () => await invokeTask.WaitAsync(TestTimeout));
Assert.Equal(WorkerClientErrorCode.PipeDisconnected, exception.ErrorCode);
await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout);
Assert.Equal(WorkerClientState.Faulted, client.State);
}
/// <summary>
/// Verifies that a worker emitting a <c>WorkerFault</c> envelope while an
/// <see cref="WorkerClient.InvokeAsync"/> is pending completes the pending invoke
/// task with a <see cref="WorkerClientException"/> carrying the worker-faulted
/// error code.
/// </summary>
[Fact]
public async Task InvokeAsync_WhenWorkerFaultsMidCommand_FailsPendingInvokeWithWorkerFaulted()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
CreateCommand(MxCommandKind.Ping),
TestTimeout,
CancellationToken.None);
WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
await pipePair.WorkerWriter.WriteAsync(CreateWorkerFaultEnvelope("scripted mid-command fault"));
WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
async () => await invokeTask.WaitAsync(TestTimeout));
Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout);
Assert.Equal(WorkerClientState.Faulted, client.State);
}
[Fact]
public async Task ReadLoop_WhenPipeDisconnects_FaultsClient()
{
@@ -244,15 +315,22 @@ public sealed class WorkerClientTests
Assert.True(process.Disposed);
}
/// <summary>
/// Verifies that a heartbeat envelope updates the last-heartbeat timestamp and worker
/// process id. Uses a <see cref="ManualTimeProvider"/> so the timestamp advance is
/// deterministic instead of relying on a wall-clock <c>Task.Delay</c> exceeding
/// <see cref="DateTimeOffset.UtcNow"/> resolution.
/// </summary>
[Fact]
public async Task ReadLoop_WhenHeartbeatArrives_UpdatesLastHeartbeatAndWorkerProcess()
{
ManualTimeProvider clock = new(DateTimeOffset.Parse("2026-05-18T12:00:00Z", System.Globalization.CultureInfo.InvariantCulture));
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await using WorkerClient client = CreateClient(pipePair, timeProvider: clock);
await CompleteHandshakeAsync(client, pipePair);
DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;
await Task.Delay(TimeSpan.FromMilliseconds(20));
clock.Advance(TimeSpan.FromSeconds(1));
await pipePair.WorkerWriter.WriteAsync(CreateHeartbeatEnvelope(workerProcessId: 9876));
await WaitUntilAsync(
@@ -260,6 +338,7 @@ public sealed class WorkerClientTests
TestTimeout);
Assert.Equal(WorkerClientState.Ready, client.State);
Assert.Equal(previousHeartbeat + TimeSpan.FromSeconds(1), client.LastHeartbeatAt);
}
/// <summary>Verifies that the heartbeat monitor faults the client when the heartbeat expires.</summary>
@@ -288,7 +367,8 @@ public sealed class WorkerClientTests
PipePair pipePair,
WorkerClientOptions? options = null,
GatewayMetrics? metrics = null,
WorkerProcessHandle? processHandle = null)
WorkerProcessHandle? processHandle = null,
TimeProvider? timeProvider = null)
{
WorkerFrameProtocolOptions frameOptions = new(SessionId);
WorkerClientConnection connection = new(
@@ -298,7 +378,7 @@ public sealed class WorkerClientTests
frameOptions,
processHandle);
return new WorkerClient(connection, options, metrics);
return new WorkerClient(connection, options, metrics, timeProvider);
}
private static WorkerProcessHandle CreateProcessHandle(FakeWorkerProcess process)
@@ -399,6 +479,23 @@ public sealed class WorkerClientTests
});
}
private static WorkerEnvelope CreateWorkerFaultEnvelope(string diagnosticMessage)
{
return CreateWorkerEnvelope(
correlationId: string.Empty,
sequence: 30,
envelope => envelope.WorkerFault = new WorkerFault
{
Category = WorkerFaultCategory.MxaccessCommandFailed,
DiagnosticMessage = diagnosticMessage,
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.WorkerUnavailable,
Message = diagnosticMessage,
},
});
}
private static WorkerEnvelope CreateHeartbeatEnvelope(int workerProcessId)
{
return CreateWorkerEnvelope(
@@ -509,6 +606,19 @@ public sealed class WorkerClientTests
}
}
/// <summary>Time provider with a manually advanced clock for deterministic timestamp tests.</summary>
private sealed class ManualTimeProvider(DateTimeOffset start) : TimeProvider
{
private DateTimeOffset _now = start;
/// <inheritdoc />
public override DateTimeOffset GetUtcNow() => _now;
/// <summary>Advances the manual clock by the given amount.</summary>
/// <param name="delta">Amount of time to add to the current clock value.</param>
public void Advance(TimeSpan delta) => _now += delta;
}
private sealed class FakeWorkerProcess : IWorkerProcess
{
private readonly TaskCompletionSource _exited = new(TaskCreationOptions.RunContinuationsAsynchronously);