Joseph Doherty
292 lines
14 KiB
C#
292 lines
14 KiB
C#
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using NSubstitute;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
|
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
|
|
|
/// <summary>
|
|
/// WP2.1b — the DataConnectionActor against a batch-capable adapter: chunked subscribe,
|
|
/// chunked/bounded re-subscribe, the seed deadline, batched tag-resolution probes, the
|
|
/// partial-failure surface and the coalesced quality push. The existing suite covers the
|
|
/// same actor against a NON batch-capable adapter, which keeps the per-tag fallback pinned.
|
|
/// </summary>
|
|
public class DataConnectionActorBatchTests : TestKit
|
|
{
|
|
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
|
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
|
|
|
|
private static DataConnectionOptions Options() => new()
|
|
{
|
|
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
|
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150),
|
|
TagResolutionRetryMaxInterval = TimeSpan.FromMilliseconds(600),
|
|
WriteTimeout = TimeSpan.FromSeconds(5),
|
|
SeedReadMaxAttempts = 1,
|
|
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
|
|
SubscribeBatchSize = 5,
|
|
SubscribeBatchDelay = TimeSpan.FromMilliseconds(50),
|
|
SeedReadBatchSize = 4,
|
|
SeedReadMaxParallelism = 2,
|
|
SeedOverallTimeout = TimeSpan.FromSeconds(30),
|
|
QualityFlushInterval = TimeSpan.FromMilliseconds(200)
|
|
};
|
|
|
|
private IActorRef CreateActor(FakeBatchDataConnection adapter, DataConnectionOptions options, string name) =>
|
|
Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
name, adapter, options, _health, _factory, "OpcUa")), name);
|
|
|
|
private static string[] Tags(int count) => Enumerable.Range(1, count).Select(i => $"tag{i}").ToArray();
|
|
|
|
[Fact]
|
|
public void Subscribe_IssuesOneAdapterRoundTripPerChunk_NotPerTag()
|
|
{
|
|
// 12 tags at SubscribeBatchSize 5 → 3 batch calls (5/5/2) and ZERO per-tag
|
|
// subscribes. This is finding #3: the old path was one adapter call (and one OPC UA
|
|
// ApplyChanges) per tag.
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, Options(), "batch-chunking");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-chunking", Tags(12), DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
|
|
var batches = adapter.SubscribeBatches.ToList();
|
|
Assert.Equal(3, batches.Count);
|
|
Assert.Equal([5, 5, 2], batches.Select(b => b.Count));
|
|
Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
|
|
}
|
|
|
|
[Fact]
|
|
public void Seed_UsesChunkedBulkReads_NotPerTagReads()
|
|
{
|
|
// Seeding 6 tags at SeedReadBatchSize 4 → 2 bulk reads, no single-tag reads.
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, Options(), "batch-seed");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed", Tags(6), DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
|
|
var reads = adapter.ReadBatches.ToList();
|
|
Assert.Equal(2, reads.Count);
|
|
Assert.Equal(6, reads.Sum(r => r.Count));
|
|
Assert.Equal(0, Volatile.Read(ref adapter.SingleReadCalls));
|
|
// The seeded value still reaches the instance actor after registration.
|
|
ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Good, TimeSpan.FromSeconds(5));
|
|
}
|
|
|
|
[Fact]
|
|
public void Seed_HangingReads_ReturnAtTheOverallDeadline_AndTheSubscribeIsStillAcked()
|
|
{
|
|
// A device whose reads never answer must not hold SubscribeCompleted (and therefore
|
|
// the instance's ack) open: the whole seed is bounded by SeedOverallTimeout.
|
|
var options = Options();
|
|
options.SeedOverallTimeout = TimeSpan.FromMilliseconds(600);
|
|
options.SeedReadTimeout = TimeSpan.FromSeconds(30); // deliberately longer than the deadline
|
|
options.SeedReadMaxAttempts = 3;
|
|
|
|
var adapter = new FakeBatchDataConnection { HangReads = true };
|
|
var actor = CreateActor(adapter, options, "batch-seed-deadline");
|
|
|
|
var started = DateTimeOffset.UtcNow;
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed-deadline", Tags(3), DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
|
|
var elapsed = DateTimeOffset.UtcNow - started;
|
|
|
|
// Bounded by the overall deadline, not by SeedReadTimeout (30s) or the retry budget.
|
|
Assert.True(elapsed < TimeSpan.FromSeconds(5), $"seed took {elapsed.TotalSeconds:F1}s");
|
|
// No value was seeded — the tags stay Uncertain until a change notification.
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
}
|
|
|
|
[Fact]
|
|
public void PartialFailure_PerTagRow_MarksTagBad_ButTheRequestStillSucceeds()
|
|
{
|
|
// A per-tag failure row is a tag-resolution problem: Bad quality is pushed for that
|
|
// tag and the subscribe is still acked Success (the surviving tags are live).
|
|
var adapter = new FakeBatchDataConnection();
|
|
adapter.FailingTags.Add("tag2");
|
|
var actor = CreateActor(adapter, Options(), "batch-partial");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-partial", Tags(3), DateTimeOffset.UtcNow));
|
|
|
|
ExpectMsg<TagValueUpdate>(u => u.TagPath == "tag2" && u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
}
|
|
|
|
[Fact]
|
|
public void ThrownBatch_AtConnectionLevel_FailsTheRequestAndDrivesReconnect()
|
|
{
|
|
// A THROWN batch means the whole chunk failed at connection level: the response must
|
|
// say so, and the actor must enter Reconnecting (which pushes bad quality for the
|
|
// connection).
|
|
var adapter = new FakeBatchDataConnection
|
|
{
|
|
BatchSubscribeThrows = () => new InvalidOperationException("client is not connected")
|
|
};
|
|
var actor = CreateActor(adapter, Options(), "batch-connection-fault");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-connection-fault", Tags(3), DateTimeOffset.UtcNow));
|
|
|
|
ExpectMsg<SubscribeTagsResponse>(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5));
|
|
ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
|
|
}
|
|
|
|
[Fact]
|
|
public void ResolutionProbes_AreBatched_NotOnePerTag()
|
|
{
|
|
// Every unresolved tag is probed in ONE batch call per chunk. Pre-fix the retry tick
|
|
// fired one SubscribeAsync task per unresolved tag, every tick, forever.
|
|
var options = Options();
|
|
options.SubscribeBatchSize = 10;
|
|
var adapter = new FakeBatchDataConnection();
|
|
foreach (var tag in Tags(6))
|
|
adapter.FailingTags.Add(tag);
|
|
|
|
var actor = CreateActor(adapter, options, "batch-probe");
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-probe", Tags(6), DateTimeOffset.UtcNow));
|
|
|
|
for (var i = 0; i < 6; i++)
|
|
ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
|
|
ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
|
|
|
|
var initialCalls = adapter.SubscribeBatches.Count;
|
|
AwaitCondition(() => adapter.SubscribeBatches.Count > initialCalls, TimeSpan.FromSeconds(5));
|
|
|
|
// The probe round carried all six unresolved tags in a single call.
|
|
var probe = adapter.SubscribeBatches.Skip(initialCalls).First();
|
|
Assert.Equal(6, probe.Count);
|
|
Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
|
|
}
|
|
|
|
[Fact]
|
|
public void ResolutionProbes_BackOff_SoADeadDeviceIsNotProbedAtFullRate()
|
|
{
|
|
// Floor 150ms, ceiling 600ms: a fixed-interval retry would fire ~13 rounds in 2s;
|
|
// the backoff sequence (150, 300, 600, 600 …) fires far fewer.
|
|
var options = Options();
|
|
options.SubscribeBatchSize = 10;
|
|
var adapter = new FakeBatchDataConnection();
|
|
adapter.FailingTags.Add("tag1");
|
|
|
|
var actor = CreateActor(adapter, options, "batch-backoff");
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-backoff", ["tag1"], DateTimeOffset.UtcNow));
|
|
ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
|
|
ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
|
|
|
|
var initialCalls = adapter.SubscribeBatches.Count;
|
|
Thread.Sleep(2000);
|
|
var rounds = adapter.SubscribeBatches.Count - initialCalls;
|
|
|
|
Assert.InRange(rounds, 2, 8);
|
|
}
|
|
|
|
[Fact]
|
|
public void QualityCounters_ArePushedOnTransitionOnly_AndCoalesced()
|
|
{
|
|
// Repeated values at UNCHANGED quality move no counter and must produce no collector
|
|
// push at all; a genuine transition produces exactly one push per coalescing window.
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, Options(), "batch-quality");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality", ["tag1"], DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
AwaitCondition(() => adapter.ValueCallback != null, TimeSpan.FromSeconds(5));
|
|
|
|
// Let the seed's own transition flush, then start counting.
|
|
Thread.Sleep(400);
|
|
_health.ClearReceivedCalls();
|
|
|
|
for (var i = 0; i < 20; i++)
|
|
adapter.ValueCallback!("tag1", new TagValue(i, QualityCode.Good, DateTimeOffset.UtcNow));
|
|
|
|
Thread.Sleep(400);
|
|
Assert.DoesNotContain(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");
|
|
|
|
// One genuine transition → exactly one coalesced push.
|
|
adapter.ValueCallback!("tag1", new TagValue(1, QualityCode.Bad, DateTimeOffset.UtcNow));
|
|
AwaitCondition(
|
|
() => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagQuality") == 1,
|
|
TimeSpan.FromSeconds(3));
|
|
Thread.Sleep(300);
|
|
Assert.Single(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");
|
|
}
|
|
|
|
[Fact]
|
|
public void QualityCounters_FlushImmediatelyOnDisconnect()
|
|
{
|
|
// "Immediate bad quality on disconnect" must never wait for the coalescing timer.
|
|
var options = Options();
|
|
options.QualityFlushInterval = TimeSpan.FromSeconds(30); // would mask a deferred push
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, options, "batch-quality-disconnect");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality-disconnect", ["tag1"], DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
ExpectMsg<TagValueUpdate>(TimeSpan.FromSeconds(5)); // the seed
|
|
_health.ClearReceivedCalls();
|
|
|
|
adapter.RaiseDisconnected();
|
|
|
|
ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
|
|
AwaitCondition(
|
|
() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateTagQuality"),
|
|
TimeSpan.FromSeconds(3));
|
|
}
|
|
|
|
[Fact]
|
|
public void Unsubscribe_ReleasesEveryHandleInOneRoundTrip()
|
|
{
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, Options(), "batch-unsubscribe");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-unsubscribe", Tags(6), DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
|
|
actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-unsubscribe", DateTimeOffset.UtcNow));
|
|
|
|
AwaitCondition(() => adapter.UnsubscribeBatches.Count == 1, TimeSpan.FromSeconds(5));
|
|
Assert.True(adapter.UnsubscribeBatches.TryDequeue(out var released));
|
|
Assert.Equal(6, released!.Count);
|
|
}
|
|
|
|
[Fact]
|
|
public void Reconnect_ReSubscribesInPacedChunks_AndRestoresTheResolvedCount()
|
|
{
|
|
// The reconnect re-subscribe is chunked and paced (SubscribeBatchDelay between
|
|
// chunks) rather than firing one task per tag, and the per-tag results repopulate
|
|
// _subscriptionIds so a later unsubscribe still releases every adapter handle.
|
|
var options = Options();
|
|
options.SubscribeBatchSize = 4;
|
|
options.SubscribeBatchDelay = TimeSpan.FromMilliseconds(120);
|
|
|
|
var adapter = new FakeBatchDataConnection();
|
|
var actor = CreateActor(adapter, options, "batch-resubscribe");
|
|
|
|
actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-resubscribe", Tags(8), DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
|
|
|
|
var beforeReconnect = adapter.SubscribeBatches.Count;
|
|
adapter.RaiseDisconnected();
|
|
|
|
// Two re-subscribe chunks of 4.
|
|
AwaitCondition(() => adapter.SubscribeBatches.Count >= beforeReconnect + 2, TimeSpan.FromSeconds(10));
|
|
var reconnectChunks = adapter.SubscribeBatches.Skip(beforeReconnect).Take(2).ToList();
|
|
Assert.Equal([4, 4], reconnectChunks.Select(c => c.Count));
|
|
|
|
// Paced: the second chunk is not issued back-to-back with the first.
|
|
var times = adapter.SubscribeBatchTimes.Skip(beforeReconnect).Take(2).ToList();
|
|
Assert.True(times[1] - times[0] >= TimeSpan.FromMilliseconds(100),
|
|
$"chunks were {(times[1] - times[0]).TotalMilliseconds:F0}ms apart");
|
|
|
|
// Every tag is registered again, so unsubscribe releases all 8 handles.
|
|
actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-resubscribe", DateTimeOffset.UtcNow));
|
|
AwaitCondition(
|
|
() => adapter.UnsubscribeBatches.Any(b => b.Count == 8),
|
|
TimeSpan.FromSeconds(10));
|
|
}
|
|
}
|