Joseph Doherty
1b3169273d
Review 2026-10-21 G-4, G-7, G-11, G-10. - G-4: PerPlatformProbeWatcher publishes its probe handles into the driver's SubscriptionRegistry under the reserved ProbeSubscriptionId (-1), so the EventPump's fan-out finally reaches the watcher — every ScanState event used to be dropped as "stale after unsubscribe" and per-platform host status never updated. ResubscribeAllAsync delegates the reserved entry to the watcher's new ResubscribeAsync, so the probe survives a reconnect; StopAsync removes the entry. Probe traffic is routed by the reserved handle and never reaches the public OnDataChange. The watcher is also built on the seam path (subscriber only), which is what makes the wiring testable without a gateway. - G-7: GalaxyDiscoverer.DiscoverAsync returns the hierarchy it fetched and the driver selects its $WinPlatform/$AppEngine probe targets from that same sweep instead of asking the gateway for the whole galaxy twice. - G-11: OnPumpDataChange stamps LastSuccessfulRead from inbound stream data, throttled (1 s default), so the comms-loss watchdog has honest liveness evidence under a Degraded supervisor. - G-10: an EventPump dispatch-loop exit that is not a cancellation now reports a stream fault so the supervisor recreates the pump, instead of leaving a deaf pump reporting Healthy. Refs #735 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
97 lines
4.1 KiB
C#
97 lines
4.1 KiB
C#
using System.Runtime.CompilerServices;
|
|
using System.Threading.Channels;
|
|
using Google.Protobuf.WellKnownTypes;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
|
|
|
|
/// <summary>
|
|
/// Review 2026-10-21 G-10: a dispatch-loop exit left the read loop draining into a channel
|
|
/// nobody read, with health Healthy and nothing telling the supervisor — a clean termination
|
|
/// evading the watchdog. The exit now reports a stream fault. The fault is injected through
|
|
/// the pump's handle factory, the one dispatch-side seam that runs outside the per-handler
|
|
/// isolation.
|
|
/// </summary>
|
|
public sealed class EventPumpDispatchFaultTests
|
|
{
|
|
private const int WaitMs = 10_000;
|
|
|
|
private sealed class ChannelSubscriber : IGalaxySubscriber
|
|
{
|
|
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
|
|
|
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
|
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
|
=> Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
|
|
|
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
yield return ev;
|
|
}
|
|
}
|
|
|
|
public ValueTask EmitAsync(int handle) => _events.Writer.WriteAsync(new MxEvent
|
|
{
|
|
Family = MxEventFamily.OnDataChange,
|
|
ItemHandle = handle,
|
|
Value = new MxValue { DoubleValue = 1 },
|
|
Quality = 192,
|
|
SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
|
});
|
|
}
|
|
|
|
/// <summary>A dispatch-loop exception reports a stream fault carrying that exception.</summary>
|
|
[Fact]
|
|
public async Task Dispatch_loop_exit_reports_a_stream_fault()
|
|
{
|
|
var subscriber = new ChannelSubscriber();
|
|
var registry = new SubscriptionRegistry();
|
|
registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
|
|
var fault = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
await using var pump = new EventPump(
|
|
subscriber, registry,
|
|
handleFactory: _ => throw new InvalidDataException("synthetic dispatch failure"),
|
|
channelCapacity: 4, clientName: "DispatchFault",
|
|
onStreamFault: ex => fault.TrySetResult(ex));
|
|
pump.Start();
|
|
|
|
await subscriber.EmitAsync(7);
|
|
|
|
(await Task.WhenAny(fault.Task, Task.Delay(WaitMs))).ShouldBe(fault.Task,
|
|
"a dispatch-loop exit must be reported as a stream fault");
|
|
(await fault.Task).ShouldBeOfType<InvalidDataException>();
|
|
}
|
|
|
|
/// <summary>Positive control: the same event with a working handle factory dispatches and reports no fault.</summary>
|
|
[Fact]
|
|
public async Task Same_event_with_a_working_factory_dispatches_without_a_fault()
|
|
{
|
|
var subscriber = new ChannelSubscriber();
|
|
var registry = new SubscriptionRegistry();
|
|
registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
|
|
var faulted = false;
|
|
var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
await using var pump = new EventPump(
|
|
subscriber, registry, channelCapacity: 4, clientName: "DispatchFault",
|
|
onStreamFault: _ => faulted = true);
|
|
pump.OnDataChange += (_, e) => delivered.TrySetResult(e);
|
|
pump.Start();
|
|
|
|
await subscriber.EmitAsync(7);
|
|
|
|
(await Task.WhenAny(delivered.Task, Task.Delay(WaitMs))).ShouldBe(delivered.Task);
|
|
(await delivered.Task).FullReference.ShouldBe("Tank01.Level");
|
|
faulted.ShouldBeFalse();
|
|
}
|
|
}
|