Joseph Doherty
0d367590db
GalaxyTelemetryTests (ActivityListener) and GatewayGalaxyDataWriterFailClosedTests (MeterListener, filtered only by meter and instrument name) observe every Galaxy span or measurement in the process. They now run in a collection with parallelization disabled (F-FU6), because a parallel driver test, or its background probe seed, polluted their counts. The stream-cancel assertion in GalaxyDriverShutdownTests goes back to a bounded wait: the instantaneous IsCompleted check introduced in round 1 was racy. Refs #735 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
387 lines
20 KiB
C#
387 lines
20 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.MxGateway.Contracts.Proto.Galaxy;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
|
|
|
|
/// <summary>
|
|
/// <see cref="GalaxyDriver.ShutdownAsync"/> used to be a no-op (review 2026-10-21, G-1/G-2/G-9):
|
|
/// the gRPC runtime outlived the driver on every respawn, removal and <c>/raw</c> browse, and
|
|
/// the host's comms-loss recovery (<c>ShutdownAsync</c> → <c>InitializeAsync</c> on the SAME
|
|
/// object) neither tore down nor rebuilt anything. These tests pin the teardown at every
|
|
/// seam a fake can reach — the StreamEvents stream, the deploy-watch stream, the alarm feed,
|
|
/// the owned channels + session — and the rebuild on the next <c>InitializeAsync</c>.
|
|
/// </summary>
|
|
public sealed class GalaxyDriverShutdownTests
|
|
{
|
|
private const int WaitMs = 10_000;
|
|
|
|
private static GalaxyDriverOptions Opts(bool watchDeploy = true) => new(
|
|
new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
|
|
new GalaxyMxAccessOptions("ShutdownTest"),
|
|
new GalaxyRepositoryOptions(WatchDeployEvents: watchDeploy),
|
|
new GalaxyReconnectOptions());
|
|
|
|
// ───────────────────────────── fakes ─────────────────────────────
|
|
|
|
/// <summary>
|
|
/// Subscriber whose event stream records the token it was opened with and completes a
|
|
/// signal when that token fires — the observable for "the StreamEvents stream was cancelled".
|
|
/// </summary>
|
|
private sealed class StreamingSubscriber : IGalaxySubscriber
|
|
{
|
|
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
|
private int _nextHandle;
|
|
|
|
public Dictionary<string, int> HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase);
|
|
public List<int> Unsubscribed { get; } = [];
|
|
public int StreamsOpened;
|
|
public TaskCompletionSource StreamCancelled { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
|
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
|
{
|
|
var results = new List<SubscribeResult>();
|
|
foreach (var r in fullReferences)
|
|
{
|
|
var h = Interlocked.Increment(ref _nextHandle);
|
|
HandleByRef[r] = h;
|
|
results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true });
|
|
}
|
|
return Task.FromResult<IReadOnlyList<SubscribeResult>>(results);
|
|
}
|
|
|
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
|
|
{
|
|
lock (Unsubscribed) Unsubscribed.AddRange(itemHandles);
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
Interlocked.Increment(ref StreamsOpened);
|
|
using var _ = cancellationToken.Register(() => StreamCancelled.TrySetResult());
|
|
await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
yield return ev;
|
|
}
|
|
}
|
|
|
|
public ValueTask EmitAsync(int handle, double value) => _events.Writer.WriteAsync(new MxEvent
|
|
{
|
|
Family = MxEventFamily.OnDataChange,
|
|
ItemHandle = handle,
|
|
Value = new MxValue { DoubleValue = value },
|
|
Quality = 192,
|
|
SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
|
});
|
|
}
|
|
|
|
/// <summary>
|
|
/// Deploy-watch source that counts OPEN enumerations (incremented on open, decremented when
|
|
/// the enumerator is cancelled or completes) — the observable for "exactly one watcher".
|
|
/// </summary>
|
|
private sealed class CountingDeployWatchSource : IGalaxyDeployWatchSource
|
|
{
|
|
private int _open;
|
|
private int _opened;
|
|
public Channel<DeployEvent> Current { get; private set; } = Channel.CreateUnbounded<DeployEvent>();
|
|
|
|
public int Open => Volatile.Read(ref _open);
|
|
public int Opened => Volatile.Read(ref _opened);
|
|
|
|
public async IAsyncEnumerable<DeployEvent> WatchAsync(
|
|
DateTimeOffset? lastSeenDeployTime, [EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
Interlocked.Increment(ref _open);
|
|
Interlocked.Increment(ref _opened);
|
|
var channel = Current = Channel.CreateUnbounded<DeployEvent>();
|
|
try
|
|
{
|
|
await foreach (var ev in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
yield return ev;
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
Interlocked.Decrement(ref _open);
|
|
}
|
|
}
|
|
|
|
public ValueTask PushAsync(ulong sequence, DateTimeOffset deployTime) => Current.Writer.WriteAsync(new DeployEvent
|
|
{
|
|
Sequence = sequence,
|
|
TimeOfLastDeployPresent = true,
|
|
TimeOfLastDeploy = Timestamp.FromDateTimeOffset(deployTime),
|
|
});
|
|
}
|
|
|
|
/// <summary>Canned platform set so a production-path init never dials the (fake) gateway for one.</summary>
|
|
private sealed class NoPlatforms : IGalaxyPlatformSource
|
|
{
|
|
public Task<IReadOnlyList<string>> GetPlatformTagNamesAsync(CancellationToken cancellationToken)
|
|
=> Task.FromResult<IReadOnlyList<string>>([]);
|
|
}
|
|
|
|
private sealed class RecordingAlarmFeed : IGalaxyAlarmFeed
|
|
{
|
|
public int Starts;
|
|
public int Disposes;
|
|
public event EventHandler<GalaxyAlarmTransition>? OnAlarmTransition;
|
|
public void Start() => Starts++;
|
|
public void Emit() => OnAlarmTransition?.Invoke(this, new GalaxyAlarmTransition(
|
|
"Tank01.Level.HiHi", "Tank01", "AnalogLimitAlarm.HiHi", GalaxyAlarmTransitionKind.Raise,
|
|
AlarmSeverity.High, 800, 750, null, DateTime.UtcNow, "", "", "Process", "hi-hi"));
|
|
public ValueTask DisposeAsync() { Disposes++; return ValueTask.CompletedTask; }
|
|
}
|
|
|
|
private static async Task WaitUntilAsync(Func<bool> condition, string because)
|
|
{
|
|
var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs);
|
|
while (!condition())
|
|
{
|
|
if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because);
|
|
await Task.Delay(10);
|
|
}
|
|
}
|
|
|
|
// ───────────────────────────── G-1: streams + pump ─────────────────────────────
|
|
|
|
/// <summary>
|
|
/// After ShutdownAsync the StreamEvents stream is cancelled, the pump is gone and the
|
|
/// registry is empty. Positive control: an event delivered through that same stream
|
|
/// reached OnDataChange before the shutdown.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task ShutdownAsync_cancels_the_event_stream_and_releases_the_pump()
|
|
{
|
|
var sub = new StreamingSubscriber();
|
|
await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, subscriber: sub);
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
|
|
var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
driver.OnDataChange += (_, e) => delivered.TrySetResult(e);
|
|
await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
|
|
driver.HasEventPumpForTests.ShouldBeTrue();
|
|
|
|
await sub.EmitAsync(sub.HandleByRef["Tank01.Level"], 42.0);
|
|
(await Task.WhenAny(delivered.Task, Task.Delay(WaitMs))).ShouldBe(delivered.Task,
|
|
"positive control: the stream must deliver before the shutdown");
|
|
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
|
|
// Bounded wait, not an instantaneous IsCompleted check: the pump's dispose does not promise the
|
|
// enumerator's cancellation registration has run by the time it returns (observed flaky).
|
|
(await Task.WhenAny(sub.StreamCancelled.Task, Task.Delay(WaitMs))).ShouldBe(sub.StreamCancelled.Task,
|
|
"ShutdownAsync must cancel the StreamEvents stream");
|
|
driver.HasEventPumpForTests.ShouldBeFalse();
|
|
driver.TrackedSubscriptionCountForTests.ShouldBe(0, "handles died with the session; the host re-subscribes after re-init");
|
|
driver.GetHealth().State.ShouldBe(DriverState.Unknown);
|
|
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull("the liveness evidence survives a shutdown");
|
|
}
|
|
|
|
/// <summary>
|
|
/// An injected alarm feed is detached on shutdown (the injector owns it, so it is NOT
|
|
/// disposed) and re-wired by the next SubscribeAlarmsAsync after a re-init. Positive
|
|
/// control on both sides of the shutdown.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task ShutdownAsync_detaches_the_alarm_feed_and_reinit_rewires_it()
|
|
{
|
|
var feed = new RecordingAlarmFeed();
|
|
var sub = new StreamingSubscriber();
|
|
await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null,
|
|
subscriber: sub, alarmFeed: feed);
|
|
var received = 0;
|
|
driver.OnAlarmEvent += (_, _) => Interlocked.Increment(ref received);
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
await driver.SubscribeAlarmsAsync(["Tank01.Level.HiHi"], CancellationToken.None);
|
|
feed.Emit();
|
|
received.ShouldBe(1, "positive control: the feed is wired before the shutdown");
|
|
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
feed.Emit();
|
|
received.ShouldBe(1, "a detached feed must not reach OnAlarmEvent");
|
|
feed.Disposes.ShouldBe(0, "an injected feed belongs to the injector");
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
await driver.SubscribeAlarmsAsync(["Tank01.Level.HiHi"], CancellationToken.None);
|
|
feed.Emit();
|
|
received.ShouldBe(2, "re-init + re-subscribe must re-wire the feed");
|
|
feed.Starts.ShouldBe(2);
|
|
}
|
|
|
|
// ───────────────────────────── G-2: one DeployWatcher ─────────────────────────────
|
|
|
|
/// <summary>
|
|
/// Five watchdog cycles (ShutdownAsync → InitializeAsync) leave exactly one open watch
|
|
/// stream. Positive control: a real deploy change then raises OnRediscoveryNeeded exactly
|
|
/// once — five stacked watchers would raise five times.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Five_watchdog_cycles_leave_exactly_one_DeployWatcher()
|
|
{
|
|
var source = new CountingDeployWatchSource();
|
|
var sub = new StreamingSubscriber();
|
|
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
|
|
deployWatchSource: source);
|
|
var raised = 0;
|
|
driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised);
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
for (var cycle = 0; cycle < 5; cycle++)
|
|
{
|
|
// Let this cycle's watch stream actually open before the next shutdown, so every cycle
|
|
// holds a live stream the teardown has to cancel (a loop cancelled before its first
|
|
// open would make the count vacuous).
|
|
await WaitUntilAsync(() => source.Opened == cycle + 1, $"cycle {cycle}: the watch stream must open");
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
}
|
|
|
|
await WaitUntilAsync(() => source.Opened == 6 && source.Open == 1,
|
|
$"expected six opens and exactly one open watch stream, saw opened={source.Opened} open={source.Open}");
|
|
driver.DeployWatcherForTests.ShouldNotBeNull();
|
|
|
|
// Bootstrap (suppressed) then a genuine change.
|
|
await source.PushAsync(1, new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
|
|
await source.PushAsync(2, new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero));
|
|
await WaitUntilAsync(() => Volatile.Read(ref raised) >= 1, "positive control: a deploy change must raise rediscovery");
|
|
await Task.Delay(200);
|
|
Volatile.Read(ref raised).ShouldBe(1, "stacked watchers would each raise for the same deploy");
|
|
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
await WaitUntilAsync(() => source.Open == 0, "ShutdownAsync must cancel the deploy-watch stream");
|
|
driver.DeployWatcherForTests.ShouldBeNull();
|
|
}
|
|
|
|
/// <summary>A second InitializeAsync without a shutdown in between replaces the watcher rather than stacking one.</summary>
|
|
[Fact]
|
|
public async Task InitializeAsync_twice_replaces_the_previous_DeployWatcher()
|
|
{
|
|
var source = new CountingDeployWatchSource();
|
|
var sub = new StreamingSubscriber();
|
|
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
|
|
deployWatchSource: source);
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
var first = driver.DeployWatcherForTests;
|
|
await WaitUntilAsync(() => source.Opened == 1, "the first watch stream must open");
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
|
|
await WaitUntilAsync(() => source.Opened == 2 && source.Open == 1,
|
|
$"expected two opens and one open watch stream, saw opened={source.Opened} open={source.Open}");
|
|
driver.DeployWatcherForTests.ShouldNotBeSameAs(first);
|
|
}
|
|
|
|
// ───────────────────────────── G-1/G-9: owned runtime, channels + session ─────────────────────────────
|
|
|
|
/// <summary>
|
|
/// The production path (no injected seams): InitializeAsync builds the owned client, session
|
|
/// and supervisor; ShutdownAsync closes the session, disposes both channels and nulls the
|
|
/// seams; the next InitializeAsync opens a NEW session on NEW channels — the watchdog's
|
|
/// recovery now genuinely rebuilds. The session open is replaced by a seam; the gRPC
|
|
/// channels are real but lazy (nothing dials until a call is made).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task ShutdownAsync_closes_the_session_disposes_both_channels_and_reinit_rebuilds()
|
|
{
|
|
var opens = 0;
|
|
var source = new CountingDeployWatchSource();
|
|
await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, deployWatchSource: source,
|
|
platformSource: new NoPlatforms())
|
|
{
|
|
SessionOpenOverrideForTests = _ => { opens++; return Task.CompletedTask; },
|
|
};
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
opens.ShouldBe(1);
|
|
var client = driver.OwnedMxClientForTests.ShouldNotBeNull();
|
|
var session = driver.OwnedMxSessionForTests.ShouldNotBeNull();
|
|
session.IsConnected.ShouldBeTrue();
|
|
driver.SupervisorForTests.ShouldNotBeNull();
|
|
driver.ProbeWatcherForTests.ShouldNotBeNull();
|
|
driver.DeployWatcherForTests.ShouldNotBeNull();
|
|
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
|
var stamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
|
|
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
|
|
session.IsConnected.ShouldBeFalse("ShutdownAsync must close the MX session");
|
|
await Should.ThrowAsync<ObjectDisposedException>(() => client.OpenSessionAsync(cancellationToken: CancellationToken.None),
|
|
"the owned gateway channel must be disposed");
|
|
driver.OwnedMxClientForTests.ShouldBeNull();
|
|
driver.OwnedMxSessionForTests.ShouldBeNull();
|
|
driver.SupervisorForTests.ShouldBeNull();
|
|
driver.ProbeWatcherForTests.ShouldBeNull();
|
|
driver.DeployWatcherForTests.ShouldBeNull();
|
|
await WaitUntilAsync(() => source.Open == 0, "the deploy-watch stream must be cancelled");
|
|
driver.GetHealth().State.ShouldBe(DriverState.Unknown);
|
|
driver.GetHealth().LastSuccessfulRead.ShouldBe(stamp);
|
|
|
|
// The watchdog's second half: a fresh runtime on the same object.
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
opens.ShouldBe(2, "re-init must open a NEW session, not reuse the dead one");
|
|
driver.OwnedMxClientForTests.ShouldNotBeNull().ShouldNotBeSameAs(client);
|
|
driver.OwnedMxSessionForTests.ShouldNotBeNull().ShouldNotBeSameAs(session);
|
|
driver.OwnedMxSessionForTests!.IsConnected.ShouldBeTrue();
|
|
driver.SupervisorForTests.ShouldNotBeNull();
|
|
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
|
await WaitUntilAsync(() => source.Open == 1, "re-init must start one deploy watcher");
|
|
}
|
|
|
|
/// <summary>
|
|
/// A connect that fails mid-build releases the partial runtime (no orphaned client per
|
|
/// watchdog retry), lands Faulted, and the next InitializeAsync starts from nothing.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Failed_InitializeAsync_releases_the_partial_runtime_and_the_retry_rebuilds()
|
|
{
|
|
var attempts = 0;
|
|
await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null,
|
|
platformSource: new NoPlatforms())
|
|
{
|
|
SessionOpenOverrideForTests = _ => ++attempts == 1
|
|
? throw new InvalidOperationException("gateway refused")
|
|
: Task.CompletedTask,
|
|
};
|
|
|
|
var ex = await Should.ThrowAsync<InvalidOperationException>(() => driver.InitializeAsync("{}", CancellationToken.None));
|
|
ex.Message.ShouldBe("gateway refused");
|
|
driver.OwnedMxClientForTests.ShouldBeNull("a failed connect must not orphan the client");
|
|
driver.OwnedMxSessionForTests.ShouldBeNull();
|
|
driver.GetHealth().State.ShouldBe(DriverState.Faulted);
|
|
driver.GetHealth().LastError.ShouldBe("gateway refused");
|
|
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
attempts.ShouldBe(2);
|
|
driver.OwnedMxSessionForTests.ShouldNotBeNull().IsConnected.ShouldBeTrue();
|
|
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
|
|
}
|
|
|
|
/// <summary>DisposeAsync after ShutdownAsync is harmless, and ShutdownAsync after DisposeAsync is a no-op.</summary>
|
|
[Fact]
|
|
public async Task Shutdown_then_dispose_and_dispose_then_shutdown_are_both_harmless()
|
|
{
|
|
var sub = new StreamingSubscriber();
|
|
var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, subscriber: sub);
|
|
await driver.InitializeAsync("{}", CancellationToken.None);
|
|
await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
|
|
|
|
await driver.ShutdownAsync(CancellationToken.None);
|
|
await Should.NotThrowAsync(async () => await driver.DisposeAsync());
|
|
await Should.NotThrowAsync(() => driver.ShutdownAsync(CancellationToken.None));
|
|
await Should.ThrowAsync<ObjectDisposedException>(() => driver.InitializeAsync("{}", CancellationToken.None));
|
|
}
|
|
}
|