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

docs(GWC-26): record alarm feed repairs as at-least-once; share the channel worker fake

Browse Source 
Code-review follow-up on fix/gwc-26-27-alarm-attach.

ApplyReconcile's snapshot-derived feed repairs are at-least-once, not
exactly-once: a reconcile reads the worker's current state while the matching
live transition may still be buffered in the monitor's lease, so both broadcast
and the duplicates are indistinguishable on the alarm feed. This pre-dates the
acked-state delta — the Raise/Clear presence repair has always had it, since
nothing serializes a reconcile pass against the in-flight live stream — so
closing it (serialization or timestamp dedup) stays out of scope for a P2 fix.
Documented instead, with the consumer contract stated explicitly (apply
transitions idempotently, never as an increment or toggle):

- ApplyReconcile gains a "Delivery semantics" comment.
- gateway.md softens the "defense in depth" prose to state the semantics.
- docs/Sessions.md carries the same caveat on the alarm-feed description.
- Tracker change-log records it as a known pre-existing characteristic and a
  candidate finding for the next review cycle.

Also hoists the ChannelWorkerClient fake — duplicated across the three alarm
test files — into TestSupport/, dropping the usings it took with it.
This commit is contained in:
Joseph Doherty
2026-08-07 06:00:16 -04:00
parent 1a75f61ebe
commit 09ccd9561f
8 changed files with 93 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs
+8
View File
@@ -530,6 +530,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
// from the worker's own authoritative snapshot to repair what the live feed missed. They are
// not MxEvents and never reach the gRPC StreamEvents path, so this feed-level repair does not
// breach the "never synthesize events" rule, which governs MxEvent emission.
//
// Delivery semantics: feed repair transitions are AT-LEAST-ONCE, not exactly-once. A reconcile
// reads the worker's current state while the corresponding live transition may still be
// buffered in the alarm lease's channel; both then broadcast, and the two are indistinguishable
// on the feed. This is inherent to the reconcile design and pre-dates the acked-state delta
// (the Raise/Clear repair has always had it), since nothing serializes a reconcile against the
// in-flight live stream. Consumers must therefore treat alarm state idempotently — apply a
// transition as "set the alarm to this state", never as an increment or a toggle.
private void ApplyReconcile(IEnumerable<ActiveAlarmSnapshot> snapshots)
{
Dictionary<string, ActiveAlarmSnapshot> next = new(StringComparer.Ordinal);
src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs
+1 -48
View File
@@ -1,5 +1,4 @@
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
@@ -8,7 +7,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
@@ -495,50 +494,4 @@ public sealed class AlarmFailoverEndToEndTests
/// <inheritdoc />
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}
/// <summary>Ready worker client whose event stream is the fake session manager's channel.</summary>
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
{
/// <inheritdoc />
public string SessionId { get; } = sessionId;
/// <inheritdoc />
public int? ProcessId { get; } = 4321;
/// <inheritdoc />
public WorkerClientState State { get; } = WorkerClientState.Ready;
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
/// <inheritdoc />
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(
WorkerCommand command,
TimeSpan timeout,
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
yield return workerEvent;
}
}
/// <inheritdoc />
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public void Kill(string reason)
{
}
/// <inheritdoc />
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
}
src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs
-48
View File
@@ -1,5 +1,4 @@
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
@@ -9,7 +8,6 @@ using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
@@ -467,50 +465,4 @@ public sealed class GatewayAlarmMonitorAttachOrderTests
return gate;
}
}
/// <summary>Worker client whose event stream is a test-driven channel.</summary>
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
{
/// <inheritdoc />
public string SessionId { get; } = sessionId;
/// <inheritdoc />
public int? ProcessId { get; } = 4321;
/// <inheritdoc />
public WorkerClientState State { get; } = WorkerClientState.Ready;
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
/// <inheritdoc />
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(
WorkerCommand command,
TimeSpan timeout,
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
yield return workerEvent;
}
}
/// <inheritdoc />
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public void Kill(string reason)
{
}
/// <inheritdoc />
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
}
src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs
+1 -48
View File
@@ -1,6 +1,5 @@
using System.Diagnostics.CodeAnalysis;
using System.Diagnostics.Metrics;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
@@ -10,7 +9,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
@@ -808,50 +807,4 @@ public sealed class GatewayAlarmMonitorProviderModeTests
/// <inheritdoc />
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}
/// <summary>Ready worker client whose event stream is the fake session manager's channel.</summary>
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
{
/// <inheritdoc />
public string SessionId { get; } = sessionId;
/// <inheritdoc />
public int? ProcessId { get; } = 4321;
/// <inheritdoc />
public WorkerClientState State { get; } = WorkerClientState.Ready;
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
/// <inheritdoc />
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(
WorkerCommand command,
TimeSpan timeout,
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
yield return workerEvent;
}
}
/// <inheritdoc />
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public void Kill(string reason)
{
}
/// <inheritdoc />
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
}
src/ZB.MOM.WW.MxGateway.Tests/TestSupport/ChannelWorkerClient.cs
+66
View File
@@ -0,0 +1,66 @@
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Workers;
namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
/// <summary>
/// Always-<see cref="WorkerClientState.Ready"/> <see cref="IWorkerClient"/> whose event
/// stream is a channel the test writes to. Lets a test attach a real
/// <c>GatewaySession</c> to a worker it drives by hand — the session can be marked Ready,
/// its <c>SessionEventDistributor</c> pump then drains this channel, and the test controls
/// exactly when each event is fanned.
/// </summary>
/// <remarks>
/// Commands are not scripted here: <see cref="InvokeAsync"/> returns an empty reply, because
/// the consumers of this fake route commands through their own <c>ISessionManager</c> double
/// rather than through the worker client. Use a purpose-built worker client instead when a
/// test needs command behavior.
/// </remarks>
/// <param name="sessionId">Session identifier the client reports.</param>
/// <param name="events">Channel whose events <see cref="ReadEventsAsync"/> yields, in order.</param>
public sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
{
/// <inheritdoc />
public string SessionId { get; } = sessionId;
/// <inheritdoc />
public int? ProcessId { get; } = 4321;
/// <inheritdoc />
public WorkerClientState State { get; } = WorkerClientState.Ready;
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
/// <inheritdoc />
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(
WorkerCommand command,
TimeSpan timeout,
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
yield return workerEvent;
}
}
/// <inheritdoc />
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public void Kill(string reason)
{
}
/// <inheritdoc />
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}